agent: use strict tables (#1686)

* agent: use strict tables

* migrate existing tables to strict

* test: verify that all tables are strict

* fix column types for device_token and ntf_mode

* fix encodings and column types for ntf_sub_action and ntf_sub_smp_action

* update schema

* remove debug.trace

* log

Author: epoberezkin

simplexmq.cabal

@@ -167,6 +167,7 @@ library
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250702_conn_invitations_remove_cascade_delete
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251009_queue_to_subscribe
           Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251010_client_notices
+          Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251230_strict_tables
   else
       exposed-modules:
           Simplex.Messaging.Agent.Store.SQLite
@@ -216,6 +217,7 @@ library
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20250702_conn_invitations_remove_cascade_delete
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20251009_queue_to_subscribe
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20251010_client_notices
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20251230_strict_tables
           Simplex.Messaging.Agent.Store.SQLite.Util
   if flag(client_postgres) || flag(server_postgres)
       exposed-modules:

src/Simplex/Messaging/Agent/Client.hs

@@ -240,6 +240,7 @@ import Simplex.Messaging.Agent.Stats
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.AgentStore (getClientNotices, updateClientNotices)
 import Simplex.Messaging.Agent.Store.Common (DBStore, withTransaction)
+import Simplex.Messaging.Agent.Store.DB (SQLError)
 import qualified Simplex.Messaging.Agent.Store.DB as DB
 import Simplex.Messaging.Agent.Store.Entity
 import Simplex.Messaging.Agent.TSessionSubs (TSessionSubs)
@@ -2124,7 +2125,9 @@ withWork_ c doWork getWork action =
       | otherwise -> notifyErr INTERNAL e
   where
     noWork = liftIO $ noWorkToDo doWork
-    notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
+    notifyErr err e = do
+      logError $ "withWork_ error: " <> tshow e
+      atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
 
 withWorkItems :: (AnyStoreError e', MonadIO m) => AgentClient -> TMVar () -> ExceptT e m (Either e' [Either e' a]) -> (NonEmpty a -> ExceptT e m ()) -> ExceptT e m ()
 withWorkItems c doWork getWork action = do
@@ -2145,7 +2148,9 @@ withWorkItems c doWork getWork action = do
       | otherwise -> notifyErr INTERNAL e
   where
     noWork = liftIO $ noWorkToDo doWork
-    notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
+    notifyErr err e = do
+      logError $ "withWorkItems error: " <> tshow e
+      atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
 
 noWorkToDo :: TMVar () -> IO ()
 noWorkToDo = void . atomically . tryTakeTMVar
@@ -2243,24 +2248,19 @@ withStore :: AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> AM a
 withStore c action = do
   st <- asks store
   withExceptT storeError . ExceptT . liftIO . agentOperationBracket c AODatabase (\_ -> pure ()) $
-    withTransaction st action `E.catches` handleDBErrors
+    withTransaction st action `E.catch` handleDBErrors
   where
+    handleDBErrors :: E.SomeException -> IO (Either StoreError a)
+    handleDBErrors e = pure $ Left $ case E.fromException e of
+      Just (e' :: SQLError) ->
 #if defined(dbPostgres)
-    -- TODO [postgres] postgres specific error handling
-    handleDBErrors :: [E.Handler IO (Either StoreError a)]
-    handleDBErrors =
-      [ E.Handler $ \(E.SomeException e) -> pure . Left $ SEInternal $ bshow e
-      ]
+        SEInternal $ bshow e'
 #else
-    handleDBErrors :: [E.Handler IO (Either StoreError a)]
-    handleDBErrors =
-      [ E.Handler $ \(e :: SQL.SQLError) ->
-          let se = SQL.sqlError e
-              busy = se == SQL.ErrorBusy || se == SQL.ErrorLocked
-           in pure . Left . (if busy then SEDatabaseBusy else SEInternal) $ bshow se,
-        E.Handler $ \(E.SomeException e) -> pure . Left $ SEInternal $ bshow e
-      ]
+        let se = SQL.sqlError e'
+            busy = se == SQL.ErrorBusy || se == SQL.ErrorLocked
+         in (if busy then SEDatabaseBusy else SEInternal) $ bshow e'
 #endif
+      Nothing -> SEInternal $ bshow e
 
 unsafeWithStore :: AgentClient -> (DB.Connection -> IO a) -> AM' a
 unsafeWithStore c action = do

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -36,6 +36,7 @@ import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
 import Data.Maybe (catMaybes)
 import qualified Data.Set as S
+import qualified Data.Text as T
 import Data.Time (UTCTime, addUTCTime, getCurrentTime)
 import Data.Time.Clock (diffUTCTime)
 import Simplex.Messaging.Agent.Client
@@ -46,13 +47,13 @@ import Simplex.Messaging.Agent.Stats
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.AgentStore
 import qualified Simplex.Messaging.Agent.Store.DB as DB
-import Simplex.Messaging.Client (NetworkRequestMode (..))
+import Simplex.Messaging.Client (NetworkRequestMode (..), nonBlockingWriteTBQueue)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Protocol (NtfServer, sameSrvAddr)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Util (catchAllErrors, diffToMicroseconds, threadDelay', tryAllErrors, tshow, whenM)
+import Simplex.Messaging.Util (catchAllErrors, catchAllErrors', diffToMicroseconds, threadDelay', tryAllErrors, tshow, whenM)
 import System.Random (randomR)
 import UnliftIO
 import UnliftIO.Concurrent (forkIO)
@@ -66,19 +67,15 @@ runNtfSupervisor c = do
     Right _ -> pure ()
   forever $ do
     cmd <- atomically . readTBQueue $ ntfSubQ ns
-    handleErr . agentOperationBracket c AONtfNetwork waitUntilActive $
-      runExceptT (processNtfCmd c cmd) >>= \case
-        Left e -> notifyErr e
-        Right _ -> return ()
+    handleErr $ agentOperationBracket c AONtfNetwork waitUntilActive $
+      processNtfCmd c cmd `catchAllErrors'` notifyErr
   where
     startTknDelete :: AM ()
     startTknDelete = do
       pendingDelServers <- withStore' c getPendingDelTknServers
       lift . forM_ pendingDelServers $ getNtfTknDelWorker True c
     handleErr :: AM' () -> AM' ()
-    handleErr = E.handle $ \(e :: E.SomeException) -> do
-      logError $ "runNtfSupervisor error " <> tshow e
-      notifyErr e
+    handleErr = E.handle $ \(e :: E.SomeException) -> notifyErr e
     notifyErr e = notifyInternalError' c $ "runNtfSupervisor error " <> show e
 
 partitionErrs :: (a -> ConnId) -> [a] -> [Either AgentErrorType b] -> ([(ConnId, AgentErrorType)], [b])
@@ -505,16 +502,18 @@ workerInternalError c connId internalErrStr = do
 
 -- TODO change error
 notifyInternalError :: MonadIO m => AgentClient -> ConnId -> String -> m ()
-notifyInternalError AgentClient {subQ} connId internalErrStr = atomically $ writeTBQueue subQ ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr)
-{-# INLINE notifyInternalError #-}
+notifyInternalError AgentClient {subQ} connId internalErrStr = do
+  logError $ T.pack internalErrStr
+  liftIO $ nonBlockingWriteTBQueue subQ ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr)
 
 notifyInternalError' :: MonadIO m => AgentClient -> String -> m ()
-notifyInternalError' AgentClient {subQ} internalErrStr = atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ INTERNAL internalErrStr)
+notifyInternalError' c = notifyInternalError c ""
 {-# INLINE notifyInternalError' #-}
 
 notifyErrs :: MonadIO m => AgentClient -> [(ConnId, AgentErrorType)] -> m ()
-notifyErrs c = mapM_ (notifySub c . ERRS) . L.nonEmpty
-{-# INLINE notifyErrs #-}
+notifyErrs c errs_ = forM_ (L.nonEmpty errs_) $ \errs -> do
+  logError $ "notifyErrs: " <> tshow errs
+  notifySub c $ ERRS errs
 
 getNtfToken :: AM' (Maybe NtfToken)
 getNtfToken = do

src/Simplex/Messaging/Agent/Protocol.hs

@@ -698,9 +698,9 @@ instance ToJSON NotificationsMode where
 instance FromJSON NotificationsMode where
   parseJSON = strParseJSON "NotificationsMode"
 
-instance ToField NotificationsMode where toField = toField . strEncode
+instance ToField NotificationsMode where toField = toField . decodeLatin1 . strEncode
 
-instance FromField NotificationsMode where fromField = blobFieldDecoder $ parseAll strP
+instance FromField NotificationsMode where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
 
 data NotificationInfo = NotificationInfo
   { ntfConnId :: ConnId,

src/Simplex/Messaging/Agent/Store.hs

@@ -16,7 +16,7 @@
 
 module Simplex.Messaging.Agent.Store where
 
-import Control.Exception (Exception)
+import Control.Exception (Exception (..))
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Char8 (ByteString)
 import Data.Int (Int64)
@@ -31,6 +31,7 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval (RI2State)
 import Simplex.Messaging.Agent.Store.Entity
 import Simplex.Messaging.Agent.Store.Common
+import Simplex.Messaging.Agent.Store.DB (SQLError)
 import Simplex.Messaging.Agent.Store.Interface (createDBStore)
 import Simplex.Messaging.Agent.Store.Migrations.App (appMigrations)
 import Simplex.Messaging.Agent.Store.Shared (MigrationConfig (..), MigrationError (..))
@@ -752,7 +753,9 @@ data StoreError
   deriving (Eq, Show, Exception)
 
 instance AnyError StoreError where
-  fromSomeException = SEInternal . bshow
+  fromSomeException e = SEInternal $ case fromException e of
+    Just (e' :: SQLError) -> bshow e'
+    Nothing -> bshow e
 
 class (Show e, AnyError e) => AnyStoreError e where
   isWorkItemError :: e -> Bool

src/Simplex/Messaging/Agent/Store/AgentStore.hs

@@ -285,7 +285,7 @@ import Simplex.Messaging.Agent.Stats
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.Common
 import qualified Simplex.Messaging.Agent.Store.DB as DB
-import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..), FromField (..), ToField (..), blobFieldDecoder, fromTextField_)
+import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..), FromField (..), ToField (..), SQLError, blobFieldDecoder, fromTextField_)
 import Simplex.Messaging.Agent.Store.Entity
 import Simplex.Messaging.Client (SMPTransportSession)
 import qualified Simplex.Messaging.Crypto as C
@@ -308,25 +308,24 @@ import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 #if defined(dbPostgres)
 import Data.List (sortOn)
-import Database.PostgreSQL.Simple (In (..), Only (..), Query, SqlError, (:.) (..))
+import Database.PostgreSQL.Simple (In (..), Only (..), Query, (:.) (..))
 import Database.PostgreSQL.Simple.Errors (constraintViolation)
 import Database.PostgreSQL.Simple.SqlQQ (sql)
 #else
-import Database.SQLite.Simple (FromRow (..), Only (..), Query (..), SQLError, ToRow (..), field, (:.) (..))
+import Database.SQLite.Simple (FromRow (..), Only (..), Query (..), ToRow (..), field, (:.) (..))
 import qualified Database.SQLite.Simple as SQL
 import Database.SQLite.Simple.QQ (sql)
 #endif
 
 checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)
 checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)
 
+handleSQLError :: StoreError -> SQLError -> StoreError
 #if defined(dbPostgres)
-handleSQLError :: StoreError -> SqlError -> StoreError
 handleSQLError err e = case constraintViolation e of
   Just _ -> err
   Nothing -> SEInternal $ bshow e
 #else
-handleSQLError :: StoreError -> SQLError -> StoreError
 handleSQLError err e
   | SQL.sqlError e == SQL.ErrorConstraint = err
   | otherwise = SEInternal $ bshow e
@@ -1428,7 +1427,7 @@ createNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer
       INSERT INTO ntf_tokens
         (provider, device_token, ntf_host, ntf_port, tkn_id, tkn_pub_key, tkn_priv_key, tkn_pub_dh_key, tkn_priv_dh_key, tkn_dh_secret, tkn_status, tkn_action, ntf_mode) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     |]
-    ((provider, token, host, port, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret) :. (ntfTknStatus, ntfTknAction, ntfMode))
+    ((provider, Binary token, host, port, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret) :. (ntfTknStatus, ntfTknAction, ntfMode))
 
 getSavedNtfToken :: DB.Connection -> IO (Maybe NtfToken)
 getSavedNtfToken db = do
@@ -1443,7 +1442,7 @@ getSavedNtfToken db = do
         JOIN ntf_servers s USING (ntf_host, ntf_port)
       |]
   where
-    ntfToken ((host, port, keyHash) :. (provider, dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret) :. (ntfTknStatus, ntfTknAction, ntfMode_)) =
+    ntfToken ((host, port, keyHash) :. (provider, Binary dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret) :. (ntfTknStatus, ntfTknAction, ntfMode_)) =
       let ntfServer = NtfServer host port keyHash
           ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey)
           ntfMode = fromMaybe NMPeriodic ntfMode_
@@ -1459,7 +1458,7 @@ updateNtfTokenRegistration db NtfToken {deviceToken = DeviceToken provider token
       SET tkn_id = ?, tkn_dh_secret = ?, tkn_status = ?, tkn_action = ?, updated_at = ?
       WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ?
     |]
-    (tknId, ntfDhSecret, NTRegistered, Nothing :: Maybe NtfTknAction, updatedAt, provider, token, host, port)
+    (tknId, ntfDhSecret, NTRegistered, Nothing :: Maybe NtfTknAction, updatedAt, provider, Binary token, host, port)
 
 updateDeviceToken :: DB.Connection -> NtfToken -> DeviceToken -> IO ()
 updateDeviceToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} (DeviceToken toProvider toToken) = do
@@ -1471,7 +1470,7 @@ updateDeviceToken db NtfToken {deviceToken = DeviceToken provider token, ntfServ
       SET provider = ?, device_token = ?, tkn_status = ?, tkn_action = ?, updated_at = ?
       WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ?
     |]
-    (toProvider, toToken, NTRegistered, Nothing :: Maybe NtfTknAction, updatedAt, provider, token, host, port)
+    (toProvider, Binary toToken, NTRegistered, Nothing :: Maybe NtfTknAction, updatedAt, provider, Binary token, host, port)
 
 updateNtfMode :: DB.Connection -> NtfToken -> NotificationsMode -> IO ()
 updateNtfMode db NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} ntfMode = do
@@ -1483,7 +1482,7 @@ updateNtfMode db NtfToken {deviceToken = DeviceToken provider token, ntfServer =
       SET ntf_mode = ?, updated_at = ?
       WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ?
     |]
-    (ntfMode, updatedAt, provider, token, host, port)
+    (ntfMode, updatedAt, provider, Binary token, host, port)
 
 updateNtfToken :: DB.Connection -> NtfToken -> NtfTknStatus -> Maybe NtfTknAction -> IO ()
 updateNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} tknStatus tknAction = do
@@ -1495,7 +1494,7 @@ updateNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer
       SET tkn_status = ?, tkn_action = ?, updated_at = ?
       WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ?
     |]
-    (tknStatus, tknAction, updatedAt, provider, token, host, port)
+    (tknStatus, tknAction, updatedAt, provider, Binary token, host, port)
 
 removeNtfToken :: DB.Connection -> NtfToken -> IO ()
 removeNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} =
@@ -1505,7 +1504,7 @@ removeNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer
       DELETE FROM ntf_tokens
       WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ?
     |]
-    (provider, token, host, port)
+    (provider, Binary token, host, port)
 
 addNtfTokenToDelete :: DB.Connection -> NtfServer -> C.APrivateAuthKey -> NtfTokenId -> IO ()
 addNtfTokenToDelete db ProtocolServer {host, port, keyHash} ntfPrivKey tknId =
@@ -1819,7 +1818,7 @@ getActiveNtfToken db =
       |]
       (Only NTActive)
   where
-    ntfToken ((host, port, keyHash) :. (provider, dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret) :. (ntfTknStatus, ntfTknAction, ntfMode_)) =
+    ntfToken ((host, port, keyHash) :. (provider, Binary dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret) :. (ntfTknStatus, ntfTknAction, ntfMode_)) =
       let ntfServer = NtfServer host port keyHash
           ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey)
           ntfMode = fromMaybe NMPeriodic ntfMode_

src/Simplex/Messaging/Agent/Store/Postgres/DB.hs

@@ -6,6 +6,7 @@ module Simplex.Messaging.Agent.Store.Postgres.DB
     PSQL.Connection,
     FromField (..),
     ToField (..),
+    SQLError,
     PSQL.connect,
     PSQL.close,
     execute,
@@ -33,6 +34,8 @@ import Database.PostgreSQL.Simple.TypeInfo.Static (textOid, varcharOid)
 
 newtype BoolInt = BI {unBI :: Bool}
 
+type SQLError = PSQL.SqlError
+
 instance FromField BoolInt where
   fromField field dat = BI . (/= (0 :: Int)) <$> fromField field dat
   {-# INLINE fromField #-}

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/App.hs

@@ -10,6 +10,7 @@ import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250322_short_links
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20250702_conn_invitations_remove_cascade_delete
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251009_queue_to_subscribe
 import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251010_client_notices
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20251230_strict_tables
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Text, Maybe Text)]
@@ -19,7 +20,8 @@ schemaMigrations =
     ("20250322_short_links", m20250322_short_links, Just down_m20250322_short_links),
     ("20250702_conn_invitations_remove_cascade_delete", m20250702_conn_invitations_remove_cascade_delete, Just down_m20250702_conn_invitations_remove_cascade_delete),
     ("20251009_queue_to_subscribe", m20251009_queue_to_subscribe, Just down_m20251009_queue_to_subscribe),
-    ("20251010_client_notices", m20251010_client_notices, Just down_m20251010_client_notices)
+    ("20251010_client_notices", m20251010_client_notices, Just down_m20251010_client_notices),
+    ("20251230_strict_tables", m20251230_strict_tables, Just down_m20251230_strict_tables)
   ]
 
 -- | The list of migrations in ascending order by date