5.2.1 (ntf 1.5.1, xftp 1.0.1) (#799)

* resubscribe in batches in Client.Agent

* fix test

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.2.0
+version: 5.2.1
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.2.0
+version:        5.2.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -63,6 +63,7 @@ library
       Simplex.Messaging.Agent.Server
       Simplex.Messaging.Agent.Store
       Simplex.Messaging.Agent.Store.SQLite
+      Simplex.Messaging.Agent.Store.SQLite.Common
       Simplex.Messaging.Agent.Store.SQLite.Migrations
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
@@ -533,6 +534,7 @@ test-suite simplexmq-test
       CoreTests.EncodingTests
       CoreTests.ProtocolErrorTests
       CoreTests.RetryIntervalTests
+      CoreTests.UtilTests
       CoreTests.VersionRangeTests
       FileDescriptionTests
       NtfClient

src/Simplex/FileTransfer/Agent.hs

@@ -71,7 +71,6 @@ import System.FilePath (takeFileName, (</>))
 import UnliftIO
 import UnliftIO.Concurrent
 import UnliftIO.Directory
-import qualified UnliftIO.Exception as E
 
 startWorkers :: AgentMonad m => AgentClient -> Maybe FilePath -> m ()
 startWorkers c workDir = do
@@ -162,7 +161,7 @@ addWorker c wsSel runWorker runWorkerNoSrv srv_ = do
       let runWorker' = case srv_ of
             Just srv -> runWorker c srv doWork
             Nothing -> runWorkerNoSrv c doWork
-      worker <- async $ runWorker' `E.finally` atomically (TM.delete srv_ ws)
+      worker <- async $ runWorker' `agentFinally` atomically (TM.delete srv_ ws)
       atomically $ TM.insert srv_ (doWork, worker) ws
     Just (doWork, _) ->
       void . atomically $ tryPutTMVar doWork ()
@@ -187,10 +186,10 @@ runXFTPRcvWorker c srv doWork = do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryInterval ri' $ \delay' loop ->
             downloadFileChunk fc replica
-              `catchError` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e
+              `catchAgentError` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e
           where
             retryLoop loop e replicaDelay = do
-              flip catchError (\_ -> pure ()) $ do
+              flip catchAgentError (\_ -> pure ()) $ do
                 notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config)
                 when notifyOnRetry $ notify c rcvFileEntityId $ RFERR e
                 closeXFTPServerClient c userId server digest
@@ -249,7 +248,7 @@ runXFTPRcvLocalWorker c doWork = do
       case nextFile of
         Nothing -> noWorkToDo
         Just f@RcvFile {rcvFileId, rcvFileEntityId, tmpPath} ->
-          decryptFile f `catchError` (rcvWorkerInternalError c rcvFileId rcvFileEntityId tmpPath . show)
+          decryptFile f `catchAgentError` (rcvWorkerInternalError c rcvFileId rcvFileEntityId tmpPath . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     decryptFile :: RcvFile -> m ()
     decryptFile RcvFile {rcvFileId, rcvFileEntityId, key, nonce, tmpPath, savePath, status, chunks} = do
@@ -300,7 +299,7 @@ sendFileExperimental c@AgentClient {xftpServers} userId filePath numRecipients =
       createDirectory outputDir
       let tempPath = workPath </> "snd"
       createDirectoryIfMissing False tempPath
-      runSend fileName outputDir tempPath `catchError` \e -> do
+      runSend fileName outputDir tempPath `catchAgentError` \e -> do
         cleanup outputDir tempPath
         notify c sndFileId $ SFERR e
       where
@@ -370,7 +369,7 @@ runXFTPSndPrepareWorker c doWork = do
       case nextFile of
         Nothing -> noWorkToDo
         Just f@SndFile {sndFileId, sndFileEntityId, prefixPath} ->
-          prepareFile f `catchError` (sndWorkerInternalError c sndFileId sndFileEntityId prefixPath . show)
+          prepareFile f `catchAgentError` (sndWorkerInternalError c sndFileId sndFileEntityId prefixPath . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     prepareFile :: SndFile -> m ()
     prepareFile SndFile {prefixPath = Nothing} =
@@ -424,7 +423,7 @@ runXFTPSndPrepareWorker c doWork = do
               usedSrvs <- newTVarIO ([] :: [XFTPServer])
               withRetryInterval (riFast ri) $ \_ loop ->
                 createWithNextSrv usedSrvs
-                  `catchError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
+                  `catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
               where
                 retryLoop loop = atomically (assertAgentForeground c) >> loop
             createWithNextSrv usedSrvs = do
@@ -460,10 +459,10 @@ runXFTPSndWorker c srv doWork = do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryInterval ri' $ \delay' loop ->
             uploadFileChunk fc replica
-              `catchError` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e
+              `catchAgentError` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e
           where
             retryLoop loop e replicaDelay = do
-              flip catchError (\_ -> pure ()) $ do
+              flip catchAgentError (\_ -> pure ()) $ do
                 notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config)
                 when notifyOnRetry $ notify c sndFileEntityId $ SFERR e
                 closeXFTPServerClient c userId server digest
@@ -579,8 +578,8 @@ deleteSndFileInternal c sndFileEntityId = do
 
 deleteSndFileRemote :: forall m. AgentMonad m => AgentClient -> UserId -> SndFileId -> ValidFileDescription 'FSender -> m ()
 deleteSndFileRemote c userId sndFileEntityId (ValidFileDescription FileDescription {chunks}) = do
-  deleteSndFileInternal c sndFileEntityId `catchError` (notify c sndFileEntityId . SFERR)
-  forM_ chunks $ \ch -> deleteFileChunk ch `catchError` (notify c sndFileEntityId . SFERR)
+  deleteSndFileInternal c sndFileEntityId `catchAgentError` (notify c sndFileEntityId . SFERR)
+  forM_ chunks $ \ch -> deleteFileChunk ch `catchAgentError` (notify c sndFileEntityId . SFERR)
   where
     deleteFileChunk :: FileChunk -> m ()
     deleteFileChunk FileChunk {digest, replicas = replica@FileChunkReplica {server} : _} = do
@@ -594,7 +593,7 @@ addXFTPDelWorker c srv = do
   atomically (TM.lookup srv ws) >>= \case
     Nothing -> do
       doWork <- newTMVarIO ()
-      worker <- async $ runXFTPDelWorker c srv doWork `E.finally` atomically (TM.delete srv ws)
+      worker <- async $ runXFTPDelWorker c srv doWork `agentFinally` atomically (TM.delete srv ws)
       atomically $ TM.insert srv (doWork, worker) ws
     Just (doWork, _) ->
       void . atomically $ tryPutTMVar doWork ()
@@ -619,10 +618,10 @@ runXFTPDelWorker c srv doWork = do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryInterval ri' $ \delay' loop ->
             deleteChunkReplica replica
-              `catchError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e
+              `catchAgentError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e
           where
             retryLoop loop e replicaDelay = do
-              flip catchError (\_ -> pure ()) $ do
+              flip catchAgentError (\_ -> pure ()) $ do
                 notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config)
                 when notifyOnRetry $ notify c "" $ SFERR e
                 closeXFTPServerClient c userId server chunkDigest

src/Simplex/FileTransfer/Client/Main.hs

@@ -74,7 +74,7 @@ import UnliftIO
 import UnliftIO.Directory
 
 xftpClientVersion :: String
-xftpClientVersion = "0.1.0"
+xftpClientVersion = "1.0.1"
 
 chunkSize1 :: Word32
 chunkSize1 = kb 256

src/Simplex/FileTransfer/Server/Main.hs

@@ -32,7 +32,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 xftpServerVersion :: String
-xftpServerVersion = "1.0.0"
+xftpServerVersion = "1.0.1"
 
 xftpServerCLI :: FilePath -> FilePath -> IO ()
 xftpServerCLI cfgPath logPath = do

src/Simplex/Messaging/Agent.hs

@@ -75,6 +75,7 @@ module Simplex.Messaging.Agent
     setNtfServers,
     setNetworkConfig,
     getNetworkConfig,
+    reconnectAllServers,
     registerNtfToken,
     verifyNtfToken,
     checkNtfToken,
@@ -316,13 +317,16 @@ setNetworkConfig :: MonadUnliftIO m => AgentClient -> NetworkConfig -> m ()
 setNetworkConfig c cfg' = do
   cfg <- atomically $ do
     swapTVar (useNetworkConfig c) cfg'
-  liftIO . when (cfg /= cfg') $ do
-    closeProtocolServerClients c smpClients
-    closeProtocolServerClients c ntfClients
+  when (cfg /= cfg') $ reconnectAllServers c
 
 getNetworkConfig :: AgentErrorMonad m => AgentClient -> m NetworkConfig
 getNetworkConfig = readTVarIO . useNetworkConfig
 
+reconnectAllServers :: MonadUnliftIO m => AgentClient -> m ()
+reconnectAllServers c = liftIO $ do
+  closeProtocolServerClients c smpClients
+  closeProtocolServerClients c ntfClients
+
 -- | Register device notifications token
 registerNtfToken :: AgentErrorMonad m => AgentClient -> DeviceToken -> NotificationsMode -> m NtfTknStatus
 registerNtfToken c = withAgentEnv c .: registerNtfToken' c
@@ -498,7 +502,7 @@ acceptContactAsync' c corrId enableNtfs invId ownConnInfo = do
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) -> do
       withStore' c $ \db -> acceptInvitation db invId ownConnInfo
-      joinConnAsync c userId corrId enableNtfs connReq ownConnInfo `catchError` \err -> do
+      joinConnAsync c userId corrId enableNtfs connReq ownConnInfo `catchAgentError` \err -> do
         withStore' c (`unacceptInvitation` invId)
         throwError err
     _ -> throwError $ CMD PROHIBITED
@@ -561,7 +565,7 @@ newConnSrv c userId connId enableNtfs cMode clientData srv = do
 newRcvConnSrv :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> Maybe CRClientData -> SMPServerWithAuth -> m (ConnId, ConnectionRequestUri c)
 newRcvConnSrv c userId connId enableNtfs cMode clientData srv = do
   AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
-  (rq, qUri) <- newRcvQueue c userId connId srv smpClientVRange `catchError` \e -> liftIO (print e) >> throwError e
+  (rq, qUri) <- newRcvQueue c userId connId srv smpClientVRange `catchAgentError` \e -> liftIO (print e) >> throwError e
   void . withStore c $ \db -> updateNewConnRcv db connId rq
   addSubscription c rq
   when enableNtfs $ do
@@ -667,7 +671,7 @@ acceptContact' c connId enableNtfs invId ownConnInfo = withConnLock c connId "ac
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) -> do
       withStore' c $ \db -> acceptInvitation db invId ownConnInfo
-      joinConn c userId connId False enableNtfs connReq ownConnInfo `catchError` \err -> do
+      joinConn c userId connId False enableNtfs connReq ownConnInfo `catchAgentError` \err -> do
         withStore' c (`unacceptInvitation` invId)
         throwError err
     _ -> throwError $ CMD PROHIBITED
@@ -783,7 +787,7 @@ getNotificationMessage' c nonce encNtfInfo = do
       ntfData <- agentCbDecrypt dhSecret nonce encNtfInfo
       PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} <- liftEither (parse strP (INTERNAL "error parsing PNMessageData") ntfData)
       (ntfConnId, rcvNtfDhSecret) <- withStore c (`getNtfRcvQueue` smpQueue)
-      ntfMsgMeta <- (eitherToMaybe . smpDecode <$> agentCbDecrypt rcvNtfDhSecret nmsgNonce encNMsgMeta) `catchError` \_ -> pure Nothing
+      ntfMsgMeta <- (eitherToMaybe . smpDecode <$> agentCbDecrypt rcvNtfDhSecret nmsgNonce encNMsgMeta) `catchAgentError` \_ -> pure Nothing
       maxMsgs <- asks $ ntfMaxMessages . config
       (NotificationInfo {ntfConnId, ntfTs, ntfMsgMeta},) <$> getNtfMessages ntfConnId maxMsgs ntfMsgMeta []
     _ -> throwError $ CMD PROHIBITED
@@ -868,8 +872,8 @@ runCommandProcessing c@AgentClient {subQ} server_ = do
     atomically $ throwWhenInactive c
     cmdId <- atomically $ readTQueue cq
     atomically $ beginAgentOperation c AOSndNetwork
-    E.try (withStore c $ \db -> getPendingCommand db cmdId) >>= \case
-      Left (e :: E.SomeException) -> atomically $ writeTBQueue subQ ("", "", APC SAEConn $ ERR $ INTERNAL $ show e)
+    tryAgentError (withStore c $ \db -> getPendingCommand db cmdId) >>= \case
+      Left e -> atomically $ writeTBQueue subQ ("", "", APC SAEConn $ ERR e)
       Right cmd -> processCmd (riFast ri) cmdId cmd
   where
     processCmd :: RetryInterval -> AsyncCmdId -> PendingCommand -> m ()
@@ -1074,9 +1078,8 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
     atomically $ beginAgentOperation c AOSndNetwork
     atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in queuePendingMsgs
     let mId = unId msgId
-    E.try (withStore c $ \db -> getPendingMsgData db connId msgId) >>= \case
-      Left (e :: E.SomeException) ->
-        notify $ MERR mId (INTERNAL $ show e)
+    tryAgentError (withStore c $ \db -> getPendingMsgData db connId msgId) >>= \case
+      Left e -> notify $ MERR mId e
       Right (rq_, PendingMsgData {msgType, msgBody, msgFlags, msgRetryState, internalTs}) -> do
         let ri' = maybe id updateRetryInterval2 msgRetryState ri
         withRetryLock2 ri' qLock $ \riState loop -> do
@@ -1306,7 +1309,7 @@ synchronizeRatchet' c connId force = withConnLock c connId "synchronizeRatchet"
 
 ackQueueMessage :: AgentMonad m => AgentClient -> RcvQueue -> SMP.MsgId -> m ()
 ackQueueMessage c rq srvMsgId =
-  sendAck c rq srvMsgId `catchError` \case
+  sendAck c rq srvMsgId `catchAgentError` \case
     SMP SMP.NO_MSG -> pure ()
     e -> throwError e
 
@@ -1507,7 +1510,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
         replaceToken :: NtfTokenId -> m NtfTknStatus
         replaceToken tknId = do
           ns <- asks ntfSupervisor
-          tryReplace ns `catchError` \e ->
+          tryReplace ns `catchAgentError` \e ->
             if temporaryOrHostError e
               then throwError e
               else do
@@ -1614,7 +1617,7 @@ deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
     let ntfTknAction = Just NTADelete
     withStore' c $ \db -> updateNtfToken db tkn ntfTknStatus ntfTknAction
     atomically $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
-    agentNtfDeleteToken c tknId tkn `catchError` \case
+    agentNtfDeleteToken c tknId tkn `catchAgentError` \case
       NTF AUTH -> pure ()
       e -> throwError e
   withStore' c $ \db -> removeNtfToken db tkn
@@ -1724,16 +1727,16 @@ cleanupManager c@AgentClient {subQ} = do
   int <- asks (cleanupInterval . config)
   forever $ do
     void . runExceptT $ do
-      deleteConns `catchError` (notify "" . ERR)
-      deleteRcvMsgHashes `catchError` (notify "" . ERR)
-      deleteProcessedRatchetKeyHashes `catchError` (notify "" . ERR)
-      deleteRcvFilesExpired `catchError` (notify "" . RFERR)
-      deleteRcvFilesDeleted `catchError` (notify "" . RFERR)
-      deleteRcvFilesTmpPaths `catchError` (notify "" . RFERR)
-      deleteSndFilesExpired `catchError` (notify "" . SFERR)
-      deleteSndFilesDeleted `catchError` (notify "" . SFERR)
-      deleteSndFilesPrefixPaths `catchError` (notify "" . SFERR)
-      deleteExpiredReplicasForDeletion `catchError` (notify "" . SFERR)
+      deleteConns `catchAgentError` (notify "" . ERR)
+      deleteRcvMsgHashes `catchAgentError` (notify "" . ERR)
+      deleteProcessedRatchetKeyHashes `catchAgentError` (notify "" . ERR)
+      deleteRcvFilesExpired `catchAgentError` (notify "" . RFERR)
+      deleteRcvFilesDeleted `catchAgentError` (notify "" . RFERR)
+      deleteRcvFilesTmpPaths `catchAgentError` (notify "" . RFERR)
+      deleteSndFilesExpired `catchAgentError` (notify "" . SFERR)
+      deleteSndFilesDeleted `catchAgentError` (notify "" . SFERR)
+      deleteSndFilesPrefixPaths `catchAgentError` (notify "" . SFERR)
+      deleteExpiredReplicasForDeletion `catchAgentError` (notify "" . SFERR)
     liftIO $ threadDelay' int
   where
     deleteConns =
@@ -1749,33 +1752,33 @@ cleanupManager c@AgentClient {subQ} = do
     deleteRcvFilesExpired = do
       rcvFilesTTL <- asks $ rcvFilesTTL . config
       rcvExpired <- withStore' c (`getRcvFilesExpired` rcvFilesTTL)
-      forM_ rcvExpired $ \(dbId, entId, p) -> flip catchError (notify entId . RFERR) $ do
+      forM_ rcvExpired $ \(dbId, entId, p) -> flip catchAgentError (notify entId . RFERR) $ do
         removePath =<< toFSFilePath p
         withStore' c (`deleteRcvFile'` dbId)
     deleteRcvFilesDeleted = do
       rcvDeleted <- withStore' c getCleanupRcvFilesDeleted
-      forM_ rcvDeleted $ \(dbId, entId, p) -> flip catchError (notify entId . RFERR) $ do
+      forM_ rcvDeleted $ \(dbId, entId, p) -> flip catchAgentError (notify entId . RFERR) $ do
         removePath =<< toFSFilePath p
         withStore' c (`deleteRcvFile'` dbId)
     deleteRcvFilesTmpPaths = do
       rcvTmpPaths <- withStore' c getCleanupRcvFilesTmpPaths
-      forM_ rcvTmpPaths $ \(dbId, entId, p) -> flip catchError (notify entId . RFERR) $ do
+      forM_ rcvTmpPaths $ \(dbId, entId, p) -> flip catchAgentError (notify entId . RFERR) $ do
         removePath =<< toFSFilePath p
         withStore' c (`updateRcvFileNoTmpPath` dbId)
     deleteSndFilesExpired = do
       sndFilesTTL <- asks $ sndFilesTTL . config
       sndExpired <- withStore' c (`getSndFilesExpired` sndFilesTTL)
-      forM_ sndExpired $ \(dbId, entId, p) -> flip catchError (notify entId . SFERR) $ do
+      forM_ sndExpired $ \(dbId, entId, p) -> flip catchAgentError (notify entId . SFERR) $ do
         forM_ p $ removePath <=< toFSFilePath
         withStore' c (`deleteSndFile'` dbId)
     deleteSndFilesDeleted = do
       sndDeleted <- withStore' c getCleanupSndFilesDeleted
-      forM_ sndDeleted $ \(dbId, entId, p) -> flip catchError (notify entId . SFERR) $ do
+      forM_ sndDeleted $ \(dbId, entId, p) -> flip catchAgentError (notify entId . SFERR) $ do
         forM_ p $ removePath <=< toFSFilePath
         withStore' c (`deleteSndFile'` dbId)
     deleteSndFilesPrefixPaths = do
       sndPrefixPaths <- withStore' c getCleanupSndFilesPrefixPaths
-      forM_ sndPrefixPaths $ \(dbId, entId, p) -> flip catchError (notify entId . SFERR) $ do
+      forM_ sndPrefixPaths $ \(dbId, entId, p) -> flip catchAgentError (notify entId . SFERR) $ do
         removePath =<< toFSFilePath p
         withStore' c (`updateSndFileNoPrefixPath` dbId)
     deleteExpiredReplicasForDeletion = do
@@ -1868,7 +1871,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                                 | rss `notElem` ([RSOk, RSStarted] :: [RatchetSyncState]) = do
                                   let cData'' = (toConnData conn') {ratchetSyncState = RSOk} :: ConnData
                                       conn'' = updateConnection cData'' conn'
-                                  notify . RSYNC RSOk $ connectionStats conn''
+                                  notify . RSYNC RSOk Nothing $ connectionStats conn''
                                   withStore' c $ \db -> setConnRatchetSync db connId RSOk
                                   pure conn''
                                 | otherwise = pure conn'
@@ -1892,10 +1895,10 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                               notifySync :: m ()
                               notifySync = qDuplex conn' "AGENT A_CRYPTO error" $ \connDuplex -> do
                                 let rss' = cryptoErrToSyncState e
-                                when (rss == RSOk || (rss == RSAllowed && rss' == RSRequired)) $ do
+                                when (rss `elem` ([RSOk, RSAllowed, RSRequired] :: [RatchetSyncState])) $ do
                                   let cData'' = (toConnData conn') {ratchetSyncState = rss'} :: ConnData
                                       conn'' = updateConnection cData'' connDuplex
-                                  notify . RSYNC rss' $ connectionStats conn''
+                                  notify . RSYNC rss' (Just e) $ connectionStats conn''
                                   withStore' c $ \db -> setConnRatchetSync db connId rss'
                           Left e -> checkDuplicateHash e encryptedMsgHash >> ack
                         where
@@ -1940,7 +1943,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
               ackDel :: InternalId -> m ()
               ackDel = enqueueCmd . ICAckDel rId srvMsgId
               handleNotifyAck :: m () -> m ()
-              handleNotifyAck m = m `catchError` \e -> notify (ERR e) >> ack
+              handleNotifyAck m = m `catchAgentError` \e -> notify (ERR e) >> ack
           SMP.END ->
             atomically (TM.lookup tSess smpClients $>>= tryReadTMVar >>= processEND)
               >>= logServer "<--" c srv rId
@@ -2062,7 +2065,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                 RcvConnection {} -> do
                   AcceptedConfirmation {ownConnInfo} <- withStore c (`getAcceptedConfirmation` connId)
                   let cData' = toConnData conn'
-                  connectReplyQueues c cData' ownConnInfo smpQueues `catchError` (notify . ERR)
+                  connectReplyQueues c cData' ownConnInfo smpQueues `catchAgentError` (notify . ERR)
                 _ -> prohibited
 
           continueSending :: (SMPServer, SMP.SenderId) -> Connection 'CDuplex -> m ()
@@ -2206,7 +2209,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
               notifyAgreed = do
                 let cData'' = cData' {ratchetSyncState = RSAgreed} :: ConnData
                     conn'' = updateConnection cData'' conn'
-                notify . RSYNC RSAgreed $ connectionStats conn''
+                notify . RSYNC RSAgreed Nothing $ connectionStats conn''
               recreateRatchet :: CR.Ratchet 'C.X448 -> m ()
               recreateRatchet rc = withStore' c $ \db -> do
                 setConnRatchetSync db connId RSAgreed

src/Simplex/Messaging/Agent/Client.hs

@@ -443,7 +443,7 @@ reconnectServer c tSess = newAsyncAction tryReconnectSMPClient $ reconnections c
     tryReconnectSMPClient aId = do
       ri <- asks $ reconnectInterval . config
       withRetryInterval ri $ \_ loop ->
-        reconnectSMPClient c tSess `catchError` const loop
+        reconnectSMPClient c tSess `catchAgentError` const loop
       atomically . removeAsyncAction aId $ reconnections c
 
 reconnectSMPClient :: forall m. AgentMonad m => AgentClient -> SMPTransportSession -> m ()
@@ -640,7 +640,7 @@ withLockMap_ locks key = withGetLock $ TM.lookup key locks >>= maybe newLock pur
 withClient_ :: forall a m err msg. (AgentMonad m, ProtocolServerClient err msg) => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> m a) -> m a
 withClient_ c tSess@(userId, srv, _) statCmd action = do
   cl <- getProtocolServerClient c tSess
-  (action cl <* stat cl "OK") `catchError` logServerError cl
+  (action cl <* stat cl "OK") `catchAgentError` logServerError cl
   where
     stat cl = liftIO . incClientStat c userId cl statCmd
     logServerError :: Client msg -> AgentErrorType -> m a

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -6,6 +6,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
 
@@ -17,6 +18,9 @@ module Simplex.Messaging.Agent.Env.SQLite
     NetworkConfig (..),
     defaultAgentConfig,
     defaultReconnectInterval,
+    tryAgentError,
+    catchAgentError,
+    agentFinally,
     Env (..),
     newSMPAgentEnv,
     createAgentStore,
@@ -52,9 +56,10 @@ import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (TLS, Transport (..))
 import Simplex.Messaging.Transport.Client (defaultSMPPort)
+import Simplex.Messaging.Util (allFinally, catchAllErrors, tryAllErrors)
 import Simplex.Messaging.Version
 import System.Random (StdGen, newStdGen)
-import UnliftIO (Async)
+import UnliftIO (Async, SomeException)
 import UnliftIO.STM
 
 type AgentMonad' m = (MonadUnliftIO m, MonadReader Env m)
@@ -225,3 +230,19 @@ newXFTPAgent = do
   xftpSndWorkers <- TM.empty
   xftpDelWorkers <- TM.empty
   pure XFTPAgent {xftpWorkDir, xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers}
+
+tryAgentError :: AgentMonad m => m a -> m (Either AgentErrorType a)
+tryAgentError = tryAllErrors mkInternal
+{-# INLINE tryAgentError #-}
+
+catchAgentError :: AgentMonad m => m a -> (AgentErrorType -> m a) -> m a
+catchAgentError = catchAllErrors mkInternal
+{-# INLINE catchAgentError #-}
+
+agentFinally :: AgentMonad m => m a -> m b -> m a
+agentFinally = allFinally mkInternal
+{-# INLINE agentFinally #-}
+
+mkInternal :: SomeException -> AgentErrorType
+mkInternal = INTERNAL . show
+{-# INLINE mkInternal #-}

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -147,7 +147,7 @@ processNtfSub c (connId, cmd) = do
       atomically (TM.lookup srv ws) >>= \case
         Nothing -> do
           doWork <- newTMVarIO ()
-          worker <- async $ runWorker c srv doWork `E.finally` atomically (TM.delete srv ws)
+          worker <- async $ runWorker c srv doWork `agentFinally` atomically (TM.delete srv ws)
           atomically $ TM.insert srv (doWork, worker) ws
         Just (doWork, _) ->
           void . atomically $ tryPutTMVar doWork ()
@@ -173,7 +173,7 @@ runNtfWorker c srv doWork = do
           ri <- asks $ reconnectInterval . config
           withRetryInterval ri $ \_ loop ->
             processAction a
-              `catchError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show)
+              `catchAgentError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     processAction :: (NtfSubscription, NtfSubNTFAction, NtfActionTs) -> m ()
     processAction (sub@NtfSubscription {connId, smpServer, ntfSubId}, action, actionTs) = do
@@ -213,7 +213,7 @@ runNtfWorker c srv doWork = do
           NSADelete -> case ntfSubId of
             Just nSubId ->
               (getNtfToken >>= mapM_ (agentNtfDeleteSubscription c nSubId))
-                `E.finally` continueDeletion
+                `agentFinally` continueDeletion
             _ -> continueDeletion
             where
               continueDeletion = do
@@ -224,7 +224,7 @@ runNtfWorker c srv doWork = do
           NSARotate -> case ntfSubId of
             Just nSubId ->
               (getNtfToken >>= mapM_ (agentNtfDeleteSubscription c nSubId))
-                `E.finally` deleteCreate
+                `agentFinally` deleteCreate
             _ -> deleteCreate
             where
               deleteCreate = do
@@ -257,7 +257,7 @@ runNtfSMPWorker c srv doWork = do
           ri <- asks $ reconnectInterval . config
           withRetryInterval ri $ \_ loop ->
             processAction a
-              `catchError` retryOnError c "NtfSMPWorker" loop (workerInternalError c connId . show)
+              `catchAgentError` retryOnError c "NtfSMPWorker" loop (workerInternalError c connId . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     processAction :: (NtfSubscription, NtfSubSMPAction, NtfActionTs) -> m ()
     processAction (sub@NtfSubscription {connId, ntfServer}, smpAction, actionTs) = do

src/Simplex/Messaging/Agent/Protocol.hs

@@ -326,7 +326,7 @@ data ACommand (p :: AParty) (e :: AEntity) where
   DOWN :: SMPServer -> [ConnId] -> ACommand Agent AENone
   UP :: SMPServer -> [ConnId] -> ACommand Agent AENone
   SWITCH :: QueueDirection -> SwitchPhase -> ConnectionStats -> ACommand Agent AEConn
-  RSYNC :: RatchetSyncState -> ConnectionStats -> ACommand Agent AEConn
+  RSYNC :: RatchetSyncState -> Maybe AgentCryptoError -> ConnectionStats -> ACommand Agent AEConn
   SEND :: MsgFlags -> MsgBody -> ACommand Client AEConn
   MID :: AgentMsgId -> ACommand Agent AEConn
   SENT :: AgentMsgId -> ACommand Agent AEConn
@@ -1449,6 +1449,20 @@ instance ToJSON AgentCryptoError where
   toJSON = J.genericToJSON $ sumTypeJSON id
   toEncoding = J.genericToEncoding $ sumTypeJSON id
 
+instance StrEncoding AgentCryptoError where
+  strP =
+    "DECRYPT_AES" $> DECRYPT_AES
+      <|> "DECRYPT_CB" $> DECRYPT_CB
+      <|> "RATCHET_HEADER" $> RATCHET_HEADER
+      <|> "RATCHET_EARLIER " *> (RATCHET_EARLIER <$> strP)
+      <|> "RATCHET_SKIPPED " *> (RATCHET_SKIPPED <$> strP)
+  strEncode = \case
+    DECRYPT_AES -> "DECRYPT_AES"
+    DECRYPT_CB -> "DECRYPT_CB"
+    RATCHET_HEADER -> "RATCHET_HEADER"
+    RATCHET_EARLIER n -> "RATCHET_EARLIER " <> strEncode n
+    RATCHET_SKIPPED n -> "RATCHET_SKIPPED " <> strEncode n
+
 instance ToJSON SMPAgentError where
   toJSON = J.genericToJSON $ sumTypeJSON id
   toEncoding = J.genericToEncoding $ sumTypeJSON id
@@ -1467,7 +1481,7 @@ instance StrEncoding AgentErrorType where
       <|> "AGENT QUEUE " *> (AGENT . A_QUEUE <$> parseRead A.takeByteString)
       <|> "AGENT " *> (AGENT <$> parseRead1)
       <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)
-      <|> "INACTIVE" *> pure INACTIVE
+      <|> "INACTIVE" $> INACTIVE
     where
       textP = T.unpack . safeDecodeUtf8 <$> A.takeTill (== ' ')
   strEncode = \case
@@ -1658,7 +1672,7 @@ commandP binaryP =
           DOWN_ -> s (DOWN <$> strP_ <*> connections)
           UP_ -> s (UP <$> strP_ <*> connections)
           SWITCH_ -> s (SWITCH <$> strP_ <*> strP_ <*> strP)
-          RSYNC_ -> s (RSYNC <$> strP_ <*> strP)
+          RSYNC_ -> s (RSYNC <$> strP_ <*> strP <*> strP)
           MID_ -> s (MID <$> A.decimal)
           SENT_ -> s (SENT <$> A.decimal)
           MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
@@ -1717,7 +1731,7 @@ serializeCommand = \case
   DOWN srv conns -> B.unwords [s DOWN_, s srv, connections conns]
   UP srv conns -> B.unwords [s UP_, s srv, connections conns]
   SWITCH dir phase srvs -> s (SWITCH_, dir, phase, srvs)
-  RSYNC rrState cstats -> s (RSYNC_, rrState, cstats)
+  RSYNC rrState cryptoErr cstats -> s (RSYNC_, rrState, cryptoErr, cstats)
   SEND msgFlags msgBody -> B.unwords [s SEND_, smpEncode msgFlags, serializeBinary msgBody]
   MID mId -> s (MID_, Str $ bshow mId)
   SENT mId -> s (SENT_, Str $ bshow mId)

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

@@ -206,7 +206,6 @@ module Simplex.Messaging.Agent.Store.SQLite
   )
 where
 
-import Control.Concurrent (threadDelay)
 import Control.Concurrent.STM (stateTVar)
 import Control.Monad.Except
 import Crypto.Random (ChaChaDRG, randomBytesGenerate)
@@ -229,7 +228,7 @@ import Data.Ord (Down (..))
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
-import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, diffUTCTime, getCurrentTime)
+import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime)
 import Data.Word (Word32)
 import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..))
 import qualified Database.SQLite.Simple as DB
@@ -246,6 +245,7 @@ import Simplex.FileTransfer.Types
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval (RI2State (..))
 import Simplex.Messaging.Agent.Store
+import Simplex.Messaging.Agent.Store.SQLite.Common
 import Simplex.Messaging.Agent.Store.SQLite.Migrations (DownMigration (..), MTRError, Migration (..), MigrationsToRun (..), mtrErrorDescription)
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import qualified Simplex.Messaging.Crypto as C
@@ -258,25 +258,18 @@ import Simplex.Messaging.Parsers (blobFieldParser, dropPrefix, fromTextField_, s
 import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport.Client (TransportHost)
-import Simplex.Messaging.Util (bshow, diffToMilliseconds, eitherToMaybe, groupOn, ($>>=), (<$$>))
+import Simplex.Messaging.Util (bshow, eitherToMaybe, groupOn, ($>>=), (<$$>))
 import Simplex.Messaging.Version
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
 import System.FilePath (takeDirectory)
 import System.IO (hFlush, stdout)
-import UnliftIO.Exception (bracket, onException)
+import UnliftIO.Exception (onException)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
 -- * SQLite Store implementation
 
-data SQLiteStore = SQLiteStore
-  { dbFilePath :: FilePath,
-    dbEncrypted :: TVar Bool,
-    dbConnection :: TMVar DB.Connection,
-    dbNew :: Bool
-  }
-
 data MigrationError
   = MEUpgrade {upMigrations :: [UpMigration]}
   | MEDowngrade {downMigrations :: [String]}
@@ -333,35 +326,35 @@ createSQLiteStore dbFilePath dbKey migrations confirmMigrations = do
     Left e -> closeSQLiteStore st $> Left e
 
 migrateSchema :: SQLiteStore -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError ())
-migrateSchema st migrations confirmMigrations = withConnection st $ \db -> do
-  Migrations.initialize db
-  Migrations.get db migrations >>= \case
+migrateSchema st migrations confirmMigrations = do
+  Migrations.initialize st
+  Migrations.get st migrations >>= \case
     Left e -> do
       when (confirmMigrations == MCConsole) $ confirmOrExit ("Database state error: " <> mtrErrorDescription e)
       pure . Left $ MigrationError e
     Right MTRNone -> pure $ Right ()
     Right ms@(MTRUp ums)
-      | dbNew st -> Migrations.run db ms $> Right ()
+      | dbNew st -> Migrations.run st ms $> Right ()
       | otherwise -> case confirmMigrations of
-        MCYesUp -> run db ms
-        MCYesUpDown -> run db ms
-        MCConsole -> confirm err >> run db ms
+        MCYesUp -> run ms
+        MCYesUpDown -> run ms
+        MCConsole -> confirm err >> run ms
         MCError -> pure $ Left err
       where
         err = MEUpgrade $ map upMigration ums -- "The app has a newer version than the database.\nConfirm to back up and upgrade using these migrations: " <> intercalate ", " (map name ums)
     Right ms@(MTRDown dms) -> case confirmMigrations of
-      MCYesUpDown -> run db ms
-      MCConsole -> confirm err >> run db ms
+      MCYesUpDown -> run ms
+      MCConsole -> confirm err >> run ms
       MCYesUp -> pure $ Left err
       MCError -> pure $ Left err
       where
         err = MEDowngrade $ map downName dms
   where
     confirm err = confirmOrExit $ migrationErrorDescription err
-    run db ms = do
+    run ms = do
       let f = dbFilePath st
       copyFile f (f <> ".bak")
-      Migrations.run db ms
+      Migrations.run st ms
       pure $ Right ()
 
 confirmOrExit :: String -> IO ()
@@ -375,9 +368,10 @@ confirmOrExit s = do
 connectSQLiteStore :: FilePath -> String -> IO SQLiteStore
 connectSQLiteStore dbFilePath dbKey = do
   dbNew <- not <$> doesFileExist dbFilePath
-  dbConnection <- newTMVarIO =<< connectDB dbFilePath dbKey
+  dbConn <- dbBusyLoop $ connectDB dbFilePath dbKey
+  dbConnVar <- newTMVarIO dbConn
   dbEncrypted <- newTVarIO . not $ null dbKey
-  pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection, dbNew}
+  pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection = dbConnVar, dbNew}
 
 connectDB :: FilePath -> String -> IO DB.Connection
 connectDB path key = do
@@ -389,13 +383,11 @@ connectDB path key = do
     prepare db = do
       let exec = SQLite3.exec $ DB.connectionHandle db
       unless (null key) . exec $ "PRAGMA key = " <> sqlString key <> ";"
-      exec . fromQuery $
-        [sql|
-          PRAGMA foreign_keys = ON;
-          -- PRAGMA trusted_schema = OFF;
-          PRAGMA secure_delete = ON;
-          PRAGMA auto_vacuum = FULL;
-        |]
+      exec "PRAGMA busy_timeout = 100;"
+      exec "PRAGMA foreign_keys = ON;"
+      -- exec "PRAGMA trusted_schema = OFF;"
+      exec "PRAGMA secure_delete = ON;"
+      exec "PRAGMA auto_vacuum = FULL;"
 
 closeSQLiteStore :: SQLiteStore -> IO ()
 closeSQLiteStore st = atomically (takeTMVar $ dbConnection st) >>= DB.close
@@ -438,37 +430,6 @@ handleSQLError err e
   | DB.sqlError e == DB.ErrorConstraint = err
   | otherwise = SEInternal $ bshow e
 
-withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withConnection SQLiteStore {dbConnection} =
-  bracket
-    (atomically $ takeTMVar dbConnection)
-    (atomically . putTMVar dbConnection)
-
-withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withTransaction = withTransactionCtx Nothing
-
-withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withTransactionCtx ctx_ st action = withConnection st $ loop 500 3_000_000
-  where
-    loop :: Int -> Int -> DB.Connection -> IO a
-    loop t tLim db =
-      transactionWithCtx `E.catch` \(e :: SQLError) ->
-        if tLim > t && DB.sqlError e == DB.ErrorBusy
-          then do
-            threadDelay t
-            loop (t * 9 `div` 8) (tLim - t) db
-          else E.throwIO e
-      where
-        transactionWithCtx = case ctx_ of
-          Nothing -> DB.withImmediateTransaction db (action db)
-          Just ctx -> do
-            t1 <- getCurrentTime
-            r <- DB.withImmediateTransaction db (action db)
-            t2 <- getCurrentTime
-            putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
-            putStrLn $ "withTransactionCtx end   :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
-            pure r
-
 createUserRecord :: DB.Connection -> IO UserId
 createUserRecord db = do
   DB.execute_ db "INSERT INTO users DEFAULT VALUES"

src/Simplex/Messaging/Agent/Store/SQLite/Common.hs

@@ -0,0 +1,61 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Common
+  ( SQLiteStore (..),
+    withConnection,
+    withTransaction,
+    withTransactionCtx,
+    dbBusyLoop,
+  )
+where
+
+import Control.Concurrent (threadDelay)
+import Data.Time.Clock (diffUTCTime, getCurrentTime)
+import Database.SQLite.Simple (SQLError)
+import qualified Database.SQLite.Simple as DB
+import Simplex.Messaging.Util (diffToMilliseconds)
+import UnliftIO.Exception (bracket)
+import qualified UnliftIO.Exception as E
+import UnliftIO.STM
+
+data SQLiteStore = SQLiteStore
+  { dbFilePath :: FilePath,
+    dbEncrypted :: TVar Bool,
+    dbConnection :: TMVar DB.Connection,
+    dbNew :: Bool
+  }
+
+withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withConnection SQLiteStore {dbConnection} =
+  bracket
+    (atomically $ takeTMVar dbConnection)
+    (atomically . putTMVar dbConnection)
+
+withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransaction = withTransactionCtx Nothing
+
+withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransactionCtx ctx_ st action = withConnection st $ \db -> dbBusyLoop (transactionWithCtx db)
+  where
+    transactionWithCtx db = case ctx_ of
+      Nothing -> DB.withImmediateTransaction db (action db)
+      Just ctx -> do
+        t1 <- getCurrentTime
+        r <- DB.withImmediateTransaction db (action db)
+        t2 <- getCurrentTime
+        putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
+        putStrLn $ "withTransactionCtx end   :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
+        pure r
+
+dbBusyLoop :: forall a. IO a -> IO a
+dbBusyLoop action = loop 500 3000000
+  where
+    loop :: Int -> Int -> IO a
+    loop t tLim =
+      action `E.catch` \(e :: SQLError) ->
+        if tLim > t && DB.sqlError e == DB.ErrorBusy
+          then do
+            threadDelay t
+            loop (t * 9 `div` 8) (tLim - t)
+          else E.throwIO e

src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs

@@ -42,6 +42,7 @@ import Database.SQLite.Simple.QQ (sql)
 import qualified Database.SQLite3 as SQLite3
 import GHC.Generics (Generic)
 import Simplex.Messaging.Agent.Protocol (extraSMPServerHosts)
+import Simplex.Messaging.Agent.Store.SQLite.Common
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220322_notifications
@@ -99,44 +100,42 @@ app = sortOn name $ map migration schemaMigrations
   where
     migration (name, up, down) = Migration {name, up = fromQuery up, down = fromQuery <$> down}
 
-get :: Connection -> [Migration] -> IO (Either MTRError MigrationsToRun)
-get db migrations = migrationsToRun migrations <$> getCurrent db
+get :: SQLiteStore -> [Migration] -> IO (Either MTRError MigrationsToRun)
+get st migrations = migrationsToRun migrations <$> withTransaction st getCurrent
 
 getCurrent :: Connection -> IO [Migration]
 getCurrent db = map toMigration <$> DB.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;"
   where
     toMigration (name, down) = Migration {name, up = "", down}
 
-run :: Connection -> MigrationsToRun -> IO ()
-run db = \case
+run :: SQLiteStore -> MigrationsToRun -> IO ()
+run st = \case
   MTRUp [] -> pure ()
-  MTRUp ms -> mapM_ runUp ms >> execSQL "VACUUM;"
+  MTRUp ms -> mapM_ runUp ms >> withConnection st (`execSQL` "VACUUM;")
   MTRDown ms -> mapM_ runDown $ reverse ms
   MTRNone -> pure ()
   where
-    runUp Migration {name, up, down} = do
-      when (name == "m20220811_onion_hosts") updateServers
-      DB.withImmediateTransaction db $ insert >> execSQL up
+    runUp Migration {name, up, down} = withTransaction st $ \db -> do
+      when (name == "m20220811_onion_hosts") $ updateServers db
+      insert db >> execSQL db up
       where
-        insert = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
-        updateServers = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
-          DB.withImmediateTransaction db $
-            let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
-             in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
-    runDown DownMigration {downName, downQuery} = do
-      DB.withImmediateTransaction db $ do
-        execSQL downQuery
-        DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
-    execSQL = SQLite3.exec $ DB.connectionHandle db
+        insert db = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
+        updateServers db = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
+          let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
+           in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
+    runDown DownMigration {downName, downQuery} = withTransaction st $ \db -> do
+      execSQL db downQuery
+      DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
+    execSQL db = SQLite3.exec $ DB.connectionHandle db
 
-initialize :: Connection -> IO ()
-initialize db = do
+initialize :: SQLiteStore -> IO ()
+initialize st = withTransaction st $ \db -> do
   cs :: [Text] <- map fromOnly <$> DB.query_ db "SELECT name FROM pragma_table_info('migrations')"
   case cs of
-    [] -> createMigrations
+    [] -> createMigrations db
     _ -> when ("down" `notElem` cs) $ DB.execute_ db "ALTER TABLE migrations ADD COLUMN down TEXT"
   where
-    createMigrations =
+    createMigrations db =
       DB.execute_
         db
         [sql|

src/Simplex/Messaging/Client.hs

@@ -207,9 +207,9 @@ defaultNetworkConfig =
       hostMode = HMOnionViaSocks,
       requiredHostMode = False,
       sessionMode = TSMUser,
-      tcpConnectTimeout = 7_500_000,
-      tcpTimeout = 5_000_000,
-      tcpTimeoutPerKb = 10_000, -- 10ms, should be less than 130ms to avoid Int overflow on 32 bit systems
+      tcpConnectTimeout = 15_000_000,
+      tcpTimeout = 10_000_000,
+      tcpTimeoutPerKb = 20_000, -- 20ms, should be less than 130ms to avoid Int overflow on 32 bit systems
       tcpKeepAlive = Just defaultKeepAliveOpts,
       smpPingInterval = 600_000_000, -- 10min
       smpPingCount = 3,

src/Simplex/Messaging/Client/Agent.hs

@@ -39,7 +39,7 @@ import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPS
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, tryE, ($>>=))
+import Simplex.Messaging.Util (catchAll_, tryE, ($>>=), toChunks)
 import System.Timeout (timeout)
 import UnliftIO (async)
 import UnliftIO.Exception (Exception)
@@ -66,7 +66,8 @@ data SMPClientAgentConfig = SMPClientAgentConfig
   { smpCfg :: ProtocolClientConfig,
     reconnectInterval :: RetryInterval,
     msgQSize :: Natural,
-    agentQSize :: Natural
+    agentQSize :: Natural,
+    agentSubsBatchSize :: Int
   }
 
 defaultSMPClientAgentConfig :: SMPClientAgentConfig
@@ -79,8 +80,9 @@ defaultSMPClientAgentConfig =
             increaseAfter = 10 * second,
             maxInterval = 10 * second
           },
-      msgQSize = 64,
-      agentQSize = 64
+      msgQSize = 256,
+      agentQSize = 256,
+      agentSubsBatchSize = 900
     }
   where
     second = 1000000
@@ -222,9 +224,9 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
           SPRecipient -> False
 
         subscribe_ :: SMPClient -> SMPSubParty -> [(SMPSub, C.APrivateSignKey)] -> ExceptT SMPClientError IO ()
-        subscribe_ smp party subs =
-          case L.nonEmpty subs of
-            Just subs' -> do
+        subscribe_ smp party = mapM_ subscribeBatch . toChunks (agentSubsBatchSize agentCfg)
+          where
+            subscribeBatch subs' = do
               let subs'' :: (NonEmpty (QueueId, C.APrivateSignKey)) = L.map (first snd) subs'
               rs <- liftIO $ smpSubscribeQueues party ca smp srv subs''
               let rs' :: (NonEmpty ((SMPSub, C.APrivateSignKey), Either SMPClientError ())) =
@@ -238,7 +240,6 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
               mapM_ (atomically . removePendingSubscription ca srv . fst) finalErrs 
               mapM_ (liftIO . notify . CASubError srv) $ L.nonEmpty finalErrs
               mapM_ (throwE . snd) $ listToMaybe tempErrs
-            Nothing -> pure ()
 
     notify :: SMPClientAgentEvent -> IO ()
     notify evt = atomically $ writeTBQueue (agentQ ca) evt

src/Simplex/Messaging/Notifications/Server.hs

@@ -158,18 +158,13 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
     subscribe = forever $ do
       subs <- atomically (readTBQueue newSubQ)
       let ss = L.groupAllWith server subs
+      batchSize <- asks $ subsBatchSize . config
       forM_ ss $ \serverSubs -> do
         let srv = server $ L.head serverSubs
-            batches = toChunks 900 $ L.toList serverSubs
+            batches = toChunks batchSize $ L.toList serverSubs
         SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber srv
         mapM_ (atomically . writeTQueue subscriberSubQ) batches
 
-    toChunks :: Int -> [a] -> [NonEmpty a]
-    toChunks _ [] = []
-    toChunks n xs =
-      let (ys, xs') = splitAt n xs
-      in maybe id (:) (L.nonEmpty ys) (toChunks n xs')
-
     server :: NtfEntityRec 'Subscription -> SMPServer
     server (NtfSub sub) = ntfSubServer sub
 

src/Simplex/Messaging/Notifications/Server/Env.hs

@@ -46,6 +46,7 @@ data NtfServerConfig = NtfServerConfig
     pushQSize :: Natural,
     smpAgentCfg :: SMPClientAgentConfig,
     apnsConfig :: APNSPushClientConfig,
+    subsBatchSize :: Int,
     inactiveClientExpiration :: Maybe ExpirationConfig,
     storeLogFile :: Maybe FilePath,
     -- CA certificate private key is not needed for initialization

src/Simplex/Messaging/Notifications/Server/Main.hs

@@ -30,7 +30,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 ntfServerVersion :: String
-ntfServerVersion = "1.5.0"
+ntfServerVersion = "1.5.1"
 
 defaultSMPBatchDelay :: Int
 defaultSMPBatchDelay = 10000
@@ -115,6 +115,7 @@ ntfServerCLI cfgPath logPath =
               pushQSize = 1048,
               smpAgentCfg = defaultSMPClientAgentConfig {smpCfg = (smpCfg defaultSMPClientAgentConfig) {batchDelay}},
               apnsConfig = defaultAPNSPushClientConfig,
+              subsBatchSize = 900,
               inactiveClientExpiration = Nothing,
               storeLogFile = enableStoreLog $> storeLogFilePath,
               caCertificateFile = c caCrtFile,

src/Simplex/Messaging/Transport/Server.hs

@@ -53,8 +53,9 @@ defaultTransportServerConfig = TransportServerConfig
   }
 
 serverTransportConfig :: TransportServerConfig -> TransportConfig
-serverTransportConfig TransportServerConfig {logTLSErrors, transportTimeout} =
-  TransportConfig {logTLSErrors, transportTimeout = Just transportTimeout}
+serverTransportConfig TransportServerConfig {logTLSErrors} =
+  -- TransportConfig {logTLSErrors, transportTimeout = Just transportTimeout}
+  TransportConfig {logTLSErrors, transportTimeout = Nothing}
 
 -- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.
 --

src/Simplex/Messaging/Util.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
@@ -13,12 +12,15 @@ import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Int (Int64)
+import Data.List (groupBy, sortOn)
+import Data.List.NonEmpty (NonEmpty)
+import qualified Data.List.NonEmpty as L
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
 import Data.Time (NominalDiffTime)
 import UnliftIO.Async
-import Data.List (groupBy, sortOn)
+import qualified UnliftIO.Exception as UE
 
 raceAny_ :: MonadUnliftIO m => [m a] -> m ()
 raceAny_ = r []
@@ -99,21 +101,44 @@ catchAll_ :: IO a -> IO a -> IO a
 catchAll_ a = catchAll a . const
 {-# INLINE catchAll_ #-}
 
+tryAllErrors :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> m (Either e a)
+tryAllErrors err action = tryError action `UE.catch` (pure . Left . err)
+{-# INLINE tryAllErrors #-}
+
+catchAllErrors :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> (e -> m a) -> m a
+catchAllErrors err action handle = tryAllErrors err action >>= either handle pure
+{-# INLINE catchAllErrors #-}
+
+catchThrow :: (MonadUnliftIO m, MonadError e m) => m a -> (E.SomeException -> e) -> m a
+catchThrow action err  = catchAllErrors err action throwError
+{-# INLINE catchThrow #-}
+
+allFinally :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> m b -> m a
+allFinally err action final = tryAllErrors err action >>= \r -> final >> either throwError pure r
+{-# INLINE allFinally #-}
+
 eitherToMaybe :: Either a b -> Maybe b
 eitherToMaybe = either (const Nothing) Just
 {-# INLINE eitherToMaybe #-}
 
 groupOn :: Eq k => (a -> k) -> [a] -> [[a]]
 groupOn = groupBy . eqOn
-    -- it is equivalent to groupBy ((==) `on` f),
-    -- but it redefines `on` to avoid duplicate computation for most values.
-    -- source: https://hackage.haskell.org/package/extra-1.7.13/docs/src/Data.List.Extra.html#groupOn
-    -- the on2 in this package is specialized to only use `==` as the function, `eqOn f` is equivalent to `(==) `on` f`
-    where eqOn f = \x -> let fx = f x in \y -> fx == f y
+  -- it is equivalent to groupBy ((==) `on` f),
+  -- but it redefines `on` to avoid duplicate computation for most values.
+  -- source: https://hackage.haskell.org/package/extra-1.7.13/docs/src/Data.List.Extra.html#groupOn
+  -- the on2 in this package is specialized to only use `==` as the function, `eqOn f` is equivalent to `(==) `on` f`
+  where
+    eqOn f = \x -> let fx = f x in \y -> fx == f y
 
 groupAllOn :: Ord k => (a -> k) -> [a] -> [[a]]
 groupAllOn f = groupOn f . sortOn f
 
+toChunks :: Int -> [a] -> [NonEmpty a]
+toChunks _ [] = []
+toChunks n xs =
+  let (ys, xs') = splitAt n xs
+  in maybe id (:) (L.nonEmpty ys) (toChunks n xs')
+
 safeDecodeUtf8 :: ByteString -> Text
 safeDecodeUtf8 = decodeUtf8With onError
   where

tests/AgentTests/FunctionalAPITests.hs

@@ -849,13 +849,13 @@ setupDesynchronizedRatchet alice bob = do
 
 ratchetSyncP :: ConnId -> RatchetSyncState -> AEntityTransmission 'AEConn -> Bool
 ratchetSyncP cId rss = \case
-  (_, cId', RSYNC rss' ConnectionStats {ratchetSyncState}) ->
+  (_, cId', RSYNC rss' _ ConnectionStats {ratchetSyncState}) ->
     cId' == cId && rss' == rss && ratchetSyncState == rss
   _ -> False
 
 ratchetSyncP' :: ConnId -> RatchetSyncState -> ATransmission 'Agent -> Bool
 ratchetSyncP' cId rss = \case
-  (_, cId', APC SAEConn (RSYNC rss' ConnectionStats {ratchetSyncState})) ->
+  (_, cId', APC SAEConn (RSYNC rss' _ ConnectionStats {ratchetSyncState})) ->
     cId' == cId && rss' == rss && ratchetSyncState == rss
   _ -> False
 

tests/AgentTests/SchemaDump.hs

@@ -50,13 +50,13 @@ testSchemaMigrations = do
       putStrLn $ "down migration " <> name m
       let downMigr = fromJust $ toDownMigration m
       schema <- getSchema testDB testSchema
-      withConnection st (`Migrations.run` MTRUp [m])
+      Migrations.run st $ MTRUp [m]
       schema' <- getSchema testDB testSchema
       schema' `shouldNotBe` schema
-      withConnection st (`Migrations.run` MTRDown [downMigr])
+      Migrations.run st $ MTRDown [downMigr]
       schema'' <- getSchema testDB testSchema
       schema'' `shouldBe` schema
-      withConnection st (`Migrations.run` MTRUp [m])
+      Migrations.run st $ MTRUp [m]
       schema''' <- getSchema testDB testSchema
       schema''' `shouldBe` schema'
 

tests/CoreTests/UtilTests.hs

@@ -0,0 +1,119 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module CoreTests.UtilTests where
+
+import Control.Exception (Exception, SomeException, throwIO)
+import Control.Monad.Except
+import Data.IORef
+import Simplex.Messaging.Util
+import Simplex.Messaging.Client.Agent ()
+import Test.Hspec
+import qualified UnliftIO.Exception as UE
+
+utilTests :: Spec
+utilTests = do
+  describe "problems of lifted try, catch and finally (don't use them)" $ do
+    describe "lifted try" $ do
+      it "does not catch errors" $ do
+        runExceptT (UE.try throwTestError >>= either handleCatch pure) `shouldReturn` Left (TestError "error")
+        runExceptT (UE.try throwTestException >>= either handleCatch pure) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+      it "with SomeException catches all errors but wraps ExceptT errors" $ do
+        runExceptT (UE.try throwTestError >>= either handleException pure) `shouldReturn` Right "caught InternalException {unInternalException = TestError \"error\"}"
+        runExceptT (UE.try throwTestException >>= either handleException pure) `shouldReturn` Right "caught user error (error)"
+    describe "lifted catch" $ do
+      it "does not catch" $ do
+        runExceptT (throwTestError `UE.catch` handleCatch) `shouldReturn` Left (TestError "error")
+        runExceptT (throwTestException `UE.catch` handleCatch) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+      it "with SomeException catches all errors but wraps ExceptT errors" $ do
+        runExceptT (throwTestError `UE.catch` handleException) `shouldReturn` Right "caught InternalException {unInternalException = TestError \"error\"}"
+        runExceptT (throwTestException `UE.catch` handleException) `shouldReturn` Right "caught user error (error)"
+    describe "lifted finally" $ do
+      it "with ExceptT error executes final action and stays in ExceptT monad" $ withFinal $ \final ->
+        runExceptT (throwTestError `UE.finally` final) `shouldReturn` Left (TestError "error")
+      it "with exception executes final action (not always - race condition?) and throws exception" $ withFinal $ \final ->
+        runExceptT (throwTestException `UE.finally` final) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+  describe "problems of tryError and catchError (don't use them)" $ do
+    describe "tryError" $ do
+      it "catches ExceptT errors but not Exceptions" $ do
+        runExceptT (tryError throwTestError >>= either handleCatch pure) `shouldReturn` Right "caught TestError \"error\""
+        runExceptT (tryError throwTestException >>= either handleCatch pure) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+    describe "catchError" $ do
+      it "catches ExceptT errors but not Exceptions" $ do
+        runExceptT (throwTestError `catchError` handleCatch) `shouldReturn` Right "caught TestError \"error\""
+        runExceptT (throwTestException `catchError` handleCatch) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+  describe "tryAllErrors" $ do
+    it "should return ExceptT error as Left" $
+      runExceptT (tryAllErrors testErr throwTestError) `shouldReturn` Right (Left (TestError "error"))
+    it "should return SomeException as Left" $
+      runExceptT (tryAllErrors testErr throwTestException) `shouldReturn` Right (Left (TestException "user error (error)"))
+    it "should return no errors as Right" $
+      runExceptT (tryAllErrors testErr noErrors) `shouldReturn` Right (Right "no errors")
+  describe "tryAllErrors specialized as tryTestError" $ do
+    let tryTestError = tryAllErrors testErr
+    it "should return ExceptT error as Left" $
+      runExceptT (tryTestError throwTestError) `shouldReturn` Right (Left (TestError "error"))
+    it "should return SomeException as Left" $
+      runExceptT (tryTestError throwTestException) `shouldReturn` Right (Left (TestException "user error (error)"))
+    it "should return no errors as Right" $
+      runExceptT (tryTestError noErrors) `shouldReturn` Right (Right "no errors")
+  describe "catchAllErrors" $ do
+    it "should catch ExceptT error" $
+      runExceptT (catchAllErrors testErr throwTestError handleCatch) `shouldReturn` Right "caught TestError \"error\""
+    it "should catch SomeException" $
+      runExceptT (catchAllErrors testErr throwTestException handleCatch) `shouldReturn` Right "caught TestException \"user error (error)\""
+    it "should not throw if there are no errors" $
+      runExceptT (catchAllErrors testErr noErrors throwError) `shouldReturn` Right "no errors"
+  describe "catchAllErrors specialized as catchTestError" $ do
+    let catchTestError = catchAllErrors testErr
+    it "should catch ExceptT error" $
+      runExceptT (throwTestError `catchTestError` handleCatch) `shouldReturn` Right "caught TestError \"error\""
+    it "should catch SomeException" $
+      runExceptT (throwTestException `catchTestError` handleCatch) `shouldReturn` Right "caught TestException \"user error (error)\""
+    it "should not throw if there are no errors" $
+      runExceptT (noErrors `catchTestError` throwError) `shouldReturn` Right "no errors"
+  describe "catchThrow" $ do
+    it "should re-throw ExceptT error" $
+      runExceptT (throwTestError `catchThrow` testErr) `shouldReturn` Left (TestError "error")
+    it "should catch SomeException and throw as ExceptT error" $
+      runExceptT (throwTestException `catchThrow` testErr) `shouldReturn` Left (TestException "user error (error)")
+    it "should not throw if there are no exceptions" $
+      runExceptT (noErrors `catchThrow` testErr) `shouldReturn` Right "no errors"
+  describe "allFinally should run final action" $ do
+    it "then throw ExceptT error" $ withFinal $ \final ->
+      runExceptT (allFinally testErr throwTestError final) `shouldReturn` Left (TestError "error")
+    it "then throw SomeException as ExceptT error" $ withFinal $ \final ->
+      runExceptT (allFinally testErr throwTestException final) `shouldReturn` Left (TestException "user error (error)")
+    it "and should not throw if there are no exceptions" $ withFinal $ \final ->
+      runExceptT (allFinally testErr noErrors final) `shouldReturn` Right "no errors"
+  describe "allFinally specialized as testFinally should run final action" $ do
+    let testFinally = allFinally testErr
+    it "then throw ExceptT error" $ withFinal $ \final ->
+      runExceptT (throwTestError `testFinally` final) `shouldReturn` Left (TestError "error")
+    it "then throw SomeException as ExceptT error" $ withFinal $ \final ->
+      runExceptT (throwTestException `testFinally` final) `shouldReturn` Left (TestException "user error (error)")
+    it "and should not throw if there are no exceptions" $ withFinal $ \final ->
+      runExceptT (noErrors `testFinally` final) `shouldReturn` Right "no errors"
+  where
+    throwTestError :: ExceptT TestError IO String
+    throwTestError = throwError $ TestError "error"
+    throwTestException :: ExceptT TestError IO String
+    throwTestException = liftIO $ throwIO $ userError "error"
+    noErrors :: ExceptT TestError IO String
+    noErrors = pure "no errors"
+    testErr :: SomeException -> TestError
+    testErr = TestException . show
+    handleCatch :: TestError -> ExceptT TestError IO String
+    handleCatch e = pure $ "caught " <> show e
+    handleException :: SomeException -> ExceptT TestError IO String
+    handleException e = pure $ "caught " <> show e
+    withFinal :: (ExceptT TestError IO String -> IO ()) -> IO ()
+    withFinal test = do
+      r <- newIORef False
+      let final = liftIO $ writeIORef r True >> pure "final"
+      test final
+      readIORef r `shouldReturn` True
+
+data TestError = TestError String | TestException String
+  deriving (Eq, Show)
+
+instance Exception TestError

tests/NtfClient.hs

@@ -89,6 +89,7 @@ ntfServerCfg =
           { apnsPort = apnsTestPort,
             caStoreFile = "tests/fixtures/ca.crt"
           },
+      subsBatchSize = 900,
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       storeLogFile = Nothing,
       -- CA certificate private key is not needed for initialization

tests/Test.hs

@@ -7,6 +7,7 @@ import CoreTests.CryptoTests
 import CoreTests.EncodingTests
 import CoreTests.ProtocolErrorTests
 import CoreTests.RetryIntervalTests
+import CoreTests.UtilTests
 import CoreTests.VersionRangeTests
 import FileDescriptionTests (fileDescriptionTests)
 import NtfServerTests (ntfServerTests)
@@ -39,6 +40,7 @@ main = do
           describe "Version range" versionRangeTests
           describe "Encryption tests" cryptoTests
           describe "Retry interval tests" retryIntervalTests
+          describe "Util tests" utilTests
         describe "SMP server via TLS" $ serverTests (transport @TLS)
         describe "SMP server via WebSockets" $ serverTests (transport @WS)
         describe "Notifications server" $ ntfServerTests (transport @TLS)