transport: optionally bound the time a connection with unacknowledged data takes to fail

TCP keep-alive probes are only sent while there is no unacknowledged data, so
a connection that died with data in flight is only reported as failed when the
OS gives up retransmitting - ~15 minutes on Linux and Android. When no network
change is reported to the agent (a silent route change, a client without a
network observer) that is how long delivery keeps failing on the dead client.

KeepAliveOpts gets unackedDataTimeout: when it is set, the sockets that set TCP
keep-alive also set TCP_USER_TIMEOUT (TCP_RXT_CONNDROPTIME on Darwin, not set
on Windows) to the keep-alive detection time, so that such a connection fails
after the same time as an idle one. On Linux this timeout also bounds the
zero-window persist timer, so a server that stops reading for that long then
fails the connection.

The option is off unless the client sets it, so that it can be enabled by the
users who test it before it becomes the default. It is a Maybe to keep the
clients that do not send it working. As it is derived from the keep-alive
options where they are set, the two detection times cannot diverge and lowering
the keep-alive settings lowers both.

It is removed for the connections via SOCKS proxy - the socket is then
connected to the proxy, its acknowledgements only mean that the proxy buffered
the data, and a slow circuit should not fail the connection.

cabal.project

@@ -31,7 +31,7 @@ source-repository-package
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/direct-sqlcipher.git
-    tag: f814ee68b16a9447fbb467ccc8f29bdd3546bfd9
+    tag: 2330df3dc8c4b660674d3c02d76bb86f66c2abee
 
 source-repository-package
     type: git

plans/2026-10-08-fast-rotation-old-queue-drain.md

@@ -0,0 +1,58 @@
+# Fast rotation: draining the old receive queue
+
+Recipient only. The sender is unchanged.
+
+## Decisions
+
+- New `RcvSwitchStatus` value:
+
+  | constructor      | encoding        |
+  |------------------|-----------------|
+  | `RSReceivedQEND` | `received_qend` |
+
+  - Stored in `rcv_queues.switch_status`.
+  - `canAbortRcvSwitch` returns `False` for it.
+- `qEndMsg`, by the queue that delivered `QEND`:
+  - the queue named in `QEND`: the named queue is marked deleted, `ICDeleteRcvQueue` is enqueued.
+  - the new queue: the named queue is set to `RSReceivedQEND` and stays subscribed.
+  - `NSCCreate` and `SWITCH QDRcv SPCompleted`: sent when the named queue had no `RSReceivedQEND`.
+- Deletion of an `RSReceivedQEND` queue:
+  - trigger: `OK` response to an `ACK` on that queue, without a `GET` lock on it.
+  - action: the queue is marked deleted, `ICDeleteRcvQueue` is enqueued.
+- Bound, a `cleanupManager` step, for each connection with an `RSReceivedQEND` queue:
+  - condition: another receive queue of the connection has an active subscription.
+  - `delete_errors + 1 < deleteErrorCount`: `delete_errors` is incremented.
+  - otherwise: the queue is deleted as above; the first `DEL` error of `ICDeleteRcvQueue` removes it locally.
+- `ackSMPMessage` result, `type QueueDrained = Bool`:
+  - `OK`: `True`.
+  - `MSG`: `False`; the message is written to `msgQ`.
+- Apps: `received_qend` is added to `RcvSwitchStatus`.
+
+## Mechanics
+
+- `Agent/Protocol.hs`: `RSReceivedQEND`.
+- `Agent/Store.hs`: `canAbortRcvSwitch`.
+- `Client.hs`: `QueueDrained`; `ackSMPMessage` returns it.
+- `Agent/Client.hs`: `sendAck` returns `QueueDrained`.
+- `Agent/Store/AgentStore.hs`: `getEndedRcvQueueConnIds`:
+
+  ```sql
+  SELECT conn_id FROM rcv_queues WHERE switch_status = ? AND deleted = 0
+  ```
+
+- `Agent.hs`:
+  - `deleteRcvQueueAsync`: `setRcvQueueDeleted`, then `ICDeleteRcvQueue`.
+  - `qEndMsg`.
+  - `ackQueueMessage`.
+  - `cleanupManager`: `expireEndedRcvQueues`.
+- Apps:
+  - `apps/ios/SimpleXChat/APITypes.swift`
+  - `apps/multiplatform/common/src/commonMain/kotlin/chat/simplex/common/model/SimpleXAPI.kt`
+
+## Tests
+
+- `QEND` on the new queue while a message waits on the old queue:
+  - the message is delivered;
+  - the old queue is deleted after the `ACK` that returns `OK`.
+- The same with a recipient restart before the `ACK`.
+- Old server stopped: the old queue is deleted by the bound.

simplexmq.cabal

@@ -252,6 +252,7 @@ library
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260823_snd_files_entitlement
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260919_ratchet_verify_codes
           Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260929_ratchet_indexes
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20261005_commands_autoincrement
           Simplex.Messaging.Agent.Store.SQLite.Util
   if flag(client_postgres) || flag(server_postgres)
       exposed-modules:

src/Simplex/FileTransfer/Agent.hs

@@ -568,9 +568,10 @@ runXFTPSndWorker c srv Worker {doWork} = do
       notify c sndFileEntityId $ SFPROG uploaded total
       when complete $ do
         (sndDescr, rcvDescrs) <- sndFileToDescrs sf
+        flip catchAllErrors (\e -> logError $ "XFTP snd worker error: " <> tshow e) $ do
+          lift . forM_ prefixPath $ removePath <=< toFSFilePath
+          withStore' c $ \db -> updateSndFileComplete db sndFileId
         notify c sndFileEntityId $ SFDONE sndDescr rcvDescrs (sndFileExpiresAt chunks)
-        lift . forM_ prefixPath $ removePath <=< toFSFilePath
-        withStore' c $ \db -> updateSndFileComplete db sndFileId
       where
         addRecipients :: SndFileChunk -> SndFileChunkReplica -> AM SndFileChunkReplica
         addRecipients ch@SndFileChunk {numRecipients} cr@SndFileChunkReplica {sndChunkReplicaId, rcvIdsKeys}

src/Simplex/FileTransfer/Protocol.hs

@@ -80,7 +80,7 @@ import Simplex.Messaging.Protocol
     tParse,
   )
 import Simplex.Messaging.Transport (THandleParams (..), TransportError (..), TransportPeer (..))
-import Simplex.Messaging.Util ((<$?>))
+import Simplex.Messaging.Util (eitherToMaybe, (<$?>))
 
 xftpBlockSize :: Int
 xftpBlockSize = 16384
@@ -383,7 +383,7 @@ checkParty' c = case testEquality (sFileParty @p) (sFileParty @p') of
 xftpEncodeAuthTransmission :: ProtocolEncoding XFTPVersion XFTPErrorType c => THandleParams XFTPVersion 'TClient -> C.APrivateAuthKey -> Transmission c -> Either TransportError ByteString
 xftpEncodeAuthTransmission thParams@THandleParams {thAuth} pKey t@(corrId, _, _) = do
   let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams t
-  xftpEncodeBatch1 . (,tToSend) =<< authTransmission thAuth False (Just pKey) (C.cbNonce $ bs corrId) tForAuth
+  xftpEncodeBatch1 . (,tToSend) =<< authTransmission thAuth False (Just pKey) (eitherToMaybe $ C.cbNonce $ bs corrId) tForAuth
 
 xftpEncodeTransmission :: ProtocolEncoding XFTPVersion XFTPErrorType c => THandleParams XFTPVersion p -> Transmission c -> Either TransportError ByteString
 xftpEncodeTransmission thParams t = xftpEncodeBatch1 (Nothing, encodeTransmission thParams t)

src/Simplex/Messaging/Agent.hs

@@ -506,7 +506,7 @@ joinConnection c nm userId connId enableNtfs = withAgentEnv c .:: joinConn c nm
 
 -- | Allow connection to continue after CONF notification (LET command)
 allowConnection :: AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> AE ()
-allowConnection c = withAgentEnv c .:. allowConnection' c
+allowConnection c = withAgentEnv c .:. allowConnection' c Nothing
 {-# INLINE allowConnection #-}
 
 -- | Accept contact after REQ notification (ACPT command)
@@ -636,7 +636,7 @@ sendMessagesB c = withAgentEnv c . sendMessagesB' c
 {-# INLINE sendMessagesB #-}
 
 ackMessage :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AE ()
-ackMessage c = withAgentEnv c .:. ackMessage' c
+ackMessage c = withAgentEnv c .:. ackMessage' c Nothing
 {-# INLINE ackMessage #-}
 
 getConnectionQueueInfo :: AgentClient -> NetworkRequestMode -> ConnId -> AE ServerQueueInfo
@@ -1621,8 +1621,8 @@ delInvSL c connId srv lnkId =
   withStore' c (\db -> deleteInvShortLink db (protoServer srv) lnkId) `catchE` \e ->
     liftIO $ nonBlockingWriteTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "error deleting short link " <> show e))
 
-joinConnSrvAsync :: AgentClient -> ConnId -> (ConnData -> Maybe SndQueue -> AM (SndQueue, (Maybe (CR.SndE2ERatchetParams 'C.X448), Maybe SMP.LinkId))) -> ConnInfo -> SubscriptionMode -> SMPServerWithAuth -> AM SndQueueSecured
-joinConnSrvAsync c connId startJoin cInfo subMode srv = do
+joinConnSrvAsync :: AgentClient -> AsyncCmdId -> ConnId -> (ConnData -> Maybe SndQueue -> AM (SndQueue, (Maybe (CR.SndE2ERatchetParams 'C.X448), Maybe SMP.LinkId))) -> ConnInfo -> SubscriptionMode -> SMPServerWithAuth -> AM SndQueueSecured
+joinConnSrvAsync c cmdId connId startJoin cInfo subMode srv = do
   SomeConn cType conn <- withStore c (`getConn` connId)
   let cData = toConnData conn
   case conn of
@@ -1637,7 +1637,7 @@ joinConnSrvAsync c connId startJoin cInfo subMode srv = do
     doJoin :: ConnData -> Maybe RcvQueue -> Maybe SndQueue -> AM SndQueueSecured
     doJoin cData rq_ sq_ = do
       (sq, (e2eSndParams, lnkId_)) <- startJoin cData sq_
-      secureConfirmQueueAsync c cData rq_ sq srv cInfo e2eSndParams subMode
+      secureConfirmQueueAsync c cmdId cData rq_ sq srv cInfo e2eSndParams subMode
         >>= (mapM_ (delInvSL c connId srv) lnkId_ $>)
 
 createReplyQueue :: AgentClient -> NetworkRequestMode -> ConnData -> SndQueue -> SubscriptionMode -> SMPServerWithAuth -> AM SMPQueueInfo
@@ -1652,13 +1652,15 @@ createReplyQueue c nm ConnData {userId, connId, enableNtfs} SndQueue {smpClientV
   pure qInfo
 
 -- | Approve confirmation (LET command) in Reader monad
-allowConnection' :: AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> AM ()
-allowConnection' c connId confId ownConnInfo = withConnLock c connId "allowConnection" $ do
+allowConnection' :: AgentClient -> Maybe AsyncCmdId -> ConnId -> ConfirmationId -> ConnInfo -> AM ()
+allowConnection' c cmdId_ connId confId ownConnInfo = withConnLock c connId "allowConnection" $ do
   withStore c (`getConn` connId) >>= \case
     SomeConn _ (RcvConnection _ RcvQueue {server, rcvId}) -> do
-      AcceptedConfirmation {senderConf = SMPConfirmation {senderKey}} <-
-        withStore c $ \db -> acceptConfirmation db confId ownConnInfo
-      enqueueCommand c "" connId (Just server) . AInternalCommand $ ICAllowSecure rcvId senderKey
+      withStore c $ \db -> runExceptT $ do
+        AcceptedConfirmation {senderConf = SMPConfirmation {senderKey}} <- ExceptT $ acceptConfirmation db confId ownConnInfo
+        ExceptT $ createCommand db "" connId (Just server) $ AInternalCommand $ ICAllowSecure rcvId senderKey
+        liftIO $ mapM_ (deleteCommand db) cmdId_
+      lift . void $ getAsyncCmdWorker True c connId (Just server)
     _ -> throwE $ CMD PROHIBITED "allowConnection"
 
 -- | Accept contact (ACPT command) in Reader monad
@@ -1744,7 +1746,7 @@ prepareReply c userId Invitation {invitationId, connReq, serviceRequest, created
     CRInvitationDR dr -> do
       cData <- newConnToAcceptDR c userId "" dr False
       sq <- startJoinInvitationDR c userId cData dr
-      storeConfirmation c cData sq Nothing innerMsg
+      storeConfirmation c Nothing cData sq Nothing innerMsg
       pure (cData, sq)
 
 serviceReqBinding :: CR.Ratchet a -> ByteString
@@ -2160,7 +2162,7 @@ getAsyncCmdWorker :: Bool -> AgentClient -> ConnId -> Maybe SMPServer -> AM' Wor
 getAsyncCmdWorker hasWork c connId server =
   getAgentWorker "async_cmd" hasWork c (connId, server) (asyncCmdWorkers c) (runCommandProcessing c connId server)
 
-data CommandCompletion = CCMoved | CCCompleted
+data CommandCompletion = CCMoved | CCCompleted | CCDeleted
 
 runCommandProcessing :: AgentClient -> ConnId -> Maybe SMPServer -> Worker -> AM ()
 runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
@@ -2195,15 +2197,15 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
             notify $ LDATA fixedData linkData connReq
         JOIN (JRInvitationDR dr@DRInvitation {replyQueue}) subMode ownCInfo -> noServer $ do
           triedHosts <- newTVarIO S.empty
-          tryCommand . withNextSrv c userId storageSrvs triedHosts [qServer replyQueue] $ \srv -> do
+          tryAtomicCommand . withNextSrv c userId storageSrvs triedHosts [qServer replyQueue] $ \srv -> do
             let startJoin cData sq_ = (,(Nothing, Nothing)) <$> maybe (startJoinInvitationDR c userId cData dr) pure sq_
-            sqSecured <- joinConnSrvAsync c connId startJoin ownCInfo subMode srv
+            sqSecured <- joinConnSrvAsync c cmdId connId startJoin ownCInfo subMode srv
             notify $ JOINED sqSecured
         JOIN (JRConnReq enableNtfs (ACR _ cReq@(CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _)) pqEnc) subMode connInfo -> noServer $ do
           triedHosts <- newTVarIO S.empty
-          tryCommand . withNextSrv c userId storageSrvs triedHosts [qServer q] $ \srv -> do
+          tryAtomicCommand . withNextSrv c userId storageSrvs triedHosts [qServer q] $ \srv -> do
             let startJoin _ sq_ = first snd <$> startJoinInvitation c userId connId sq_ enableNtfs cReq pqEnc
-            sqSecured <- joinConnSrvAsync c connId startJoin connInfo subMode srv
+            sqSecured <- joinConnSrvAsync c cmdId connId startJoin connInfo subMode srv
             notify $ JOINED sqSecured
         -- TODO TBC using joinConnSrvAsync for contact URIs, with receive queue created asynchronously.
         -- Currently joinConnSrv is used because even joinConnSrvAsync for invitation URIs creates receive queue synchronously.
@@ -2225,35 +2227,37 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
                       if temporaryOrHostError e
                         then throwE e
                         else atomically $ void $ tryPutTMVar v $ Left e -- will not overwrite existing result
-        LET confId ownCInfo -> withServer' . tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK
-        ACK msgId rcptInfo_ -> withServer' . tryCommand $ ackMessage' c connId msgId rcptInfo_ >> notify OK
+        LET confId ownCInfo -> withServer' . tryAtomicCommand $ allowConnection' c (Just cmdId) connId confId ownCInfo >> notify OK
+        ACK msgId rcptInfo_ -> withServer' . tryAtomicCommand $ ackMessage' c (Just cmdId) connId msgId rcptInfo_ >> notify OK
         SWCH ->
-          noServer . tryWithLock "switchConnection" $
+          noServer . tryAtomicWithLock "switchConnection" $
             withStore c (`getConn` connId) >>= \case
               SomeConn _ conn@(DuplexConnection _ (replaced :| _rqs) _) ->
-                switchDuplexConnection c NRMBackground conn replaced >>= notify . SWITCH QDRcv SPStarted
+                switchDuplexConnection c NRMBackground (Just cmdId) conn replaced >>= notify . SWITCH QDRcv SPStarted
               _ -> throwE $ CMD PROHIBITED "SWCH: not duplex"
-        DEL -> withServer' . tryCommand $ deleteConnection' c NRMBackground connId >> notify OK
+        DEL -> withServer' . tryAtomicCommand $ deleteConnection' c NRMBackground connId >> notify OK
       AInternalCommand cmd -> case cmd of
         ICAckDel rId srvMsgId msgId -> withServer $ \srv ->
-          tryCommand $ withConnLockNotify c connId "ICAckDel" $ do
+          tryAtomicCommand $ withConnLockNotify c connId "ICAckDel" $ do
             t_ <- ack srv rId srvMsgId
-            withStore' c (\db -> deleteMsg db connId msgId)
+            withStore' c (\db -> deleteMsg db connId msgId >> deleteCommand db cmdId)
             pure t_
         ICAck rId srvMsgId -> withServer $ \srv ->
           tryCommand $ withConnLockNotify c connId "ICAck" $ ack srv rId srvMsgId
         ICAllowSecure _rId senderKey -> withServer' . tryMoveableWithLock "ICAllowSecure" $ do
           (SomeConn _ conn, AcceptedConfirmation {senderConf, ownConnInfo}) <-
             withStore c $ \db -> runExceptT $ (,) <$> ExceptT (getConn db connId) <*> ExceptT (getAcceptedConfirmation db connId)
+          let connectReplyQueues_ cData sq_ = case L.nonEmpty $ smpReplyQueues senderConf of
+                Just qs -> connectReplyQueues c cmdId cData ownConnInfo sq_ qs $> CCDeleted
+                Nothing -> pure CCCompleted
           case conn of
             RcvConnection cData rq -> do
               mapM_ (secure rq) senderKey
-              mapM_ (connectReplyQueues c cData ownConnInfo Nothing) (L.nonEmpty $ smpReplyQueues senderConf)
-              pure CCCompleted
+              connectReplyQueues_ cData Nothing
             -- duplex connection is matched to handle SKEY retries
             DuplexConnection cData _ (sq :| _) -> do
-              tryAllErrors (mapM_ (connectReplyQueues c cData ownConnInfo (Just sq)) (L.nonEmpty $ smpReplyQueues senderConf)) >>= \case
-                Right () -> pure CCCompleted
+              tryAllErrors (connectReplyQueues_ cData (Just sq)) >>= \case
+                Right cc -> pure cc
                 Left e
                   | temporaryOrHostError e && Just server /= server_ -> do
                       -- In case the server is different we update server to remove command from this (connId, srv) queue
@@ -2264,36 +2268,36 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
               where
                 server = qServer sq
             _ -> throwE $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd)
-        ICDuplexSecure _rId senderKey -> withServer' . tryWithLock "ICDuplexSecure" . withDuplexConn $ \(DuplexConnection cData (rq :| _) (sq :| _)) -> do
+        ICDuplexSecure _rId senderKey -> withServer' . tryAtomicWithLock "ICDuplexSecure" . withDuplexConn $ \(DuplexConnection cData (rq :| _) (sq :| _)) -> do
           secure rq senderKey
-          void $ enqueueMessage c cData sq SMP.MsgFlags {notification = True} HELLO
+          void $ enqueueMessage c (Just cmdId) cData sq SMP.MsgFlags {notification = True} HELLO
         ICReplyDel -> withServer' . tryWithLock "ICReplyDel" $
           withStore c (`getConn` connId) >>= \case
             SomeConn _ (SndConnection cData sq) -> sendReplySync c NRMBackground (cData, sq)
             _ -> throwE $ INTERNAL "ICReplyDel: incorrect connection type"
         -- ICDeleteConn is no longer used, but it can be present in old client databases
         ICDeleteConn -> withStore' c (`deleteCommand` cmdId)
-        ICDeleteRcvQueue rId -> withServer $ \srv -> tryWithLock "ICDeleteRcvQueue" $ do
+        ICDeleteRcvQueue rId -> withServer $ \srv -> tryAtomicWithLock "ICDeleteRcvQueue" $ do
           rq <- withStore c (\db -> getDeletedRcvQueue db connId srv rId)
           maxErrs <- asks $ deleteErrorCount . config
           tryAllErrors (deleteQueue c NRMBackground rq) >>= \case
             Left e | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> do
               withStore' c (`incRcvDeleteErrors` rq)
               throwE e
-            _ -> withStore' c (`deleteConnRcvQueue` rq)
+            _ -> withStore' c $ \db -> deleteConnRcvQueue db rq >> deleteCommand db cmdId
         ICQSecure rId senderKey ->
-          withServer $ \srv -> tryWithLock "ICQSecure" . withDuplexConn $ \(DuplexConnection cData rqs sqs) ->
+          withServer $ \srv -> tryAtomicWithLock "ICQSecure" . withDuplexConn $ \(DuplexConnection cData rqs sqs) ->
             case find (sameQueue (srv, rId)) rqs of
               Just rq'@RcvQueue {server, sndId, status, dbReplaceQueueId = Just replaceQId} ->
                 case find ((replaceQId ==) . dbQId) rqs of
-                  Just rq1 -> when (status == Confirmed) $ do
+                  Just rq1 -> if status /= Confirmed then withStore' c (`deleteCommand` cmdId) else do
                     secureQueue c NRMBackground rq' senderKey
                     -- we may add more statistics special to queue rotation later on,
                     -- not accounting secure during rotation for now:
                     -- atomically $ incSMPServerStat c userId server connSecured
                     withStore' c $ \db -> setRcvQueueStatus db rq' Secured
                     void . enqueueMessages c cData sqs SMP.noMsgFlags $ QUSE [((server, sndId), True)]
-                    rq1' <- withStore' c $ \db -> setRcvSwitchStatus db rq1 $ Just RSSendingQUSE
+                    rq1' <- withStore' c $ \db -> setRcvSwitchStatus db rq1 (Just RSSendingQUSE) <* deleteCommand db cmdId
                     let rqs' = updatedQs rq1' rqs
                         conn' = DuplexConnection cData rqs' sqs
                     cStats <- connectionStats c conn'
@@ -2326,7 +2330,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
                     forM_ (find (\q -> sndSwchStatus q == Just SSSendingQEND) sqs) $ \oldSq ->
                       void $ enqueueMessages c cData [oldSq, sq'] SMP.noMsgFlags $ QEND [qAddress oldSq]
         ICQDelete rId -> do
-          withServer $ \srv -> tryWithLock "ICQDelete" . withDuplexConn $ \(DuplexConnection cData@ConnData {enableNtfs} rqs sqs) -> do
+          withServer $ \srv -> tryAtomicWithLock "ICQDelete" . withDuplexConn $ \(DuplexConnection cData@ConnData {enableNtfs} rqs sqs) -> do
             case removeQ (srv, rId) rqs of
               Nothing -> internalErr "ICQDelete: queue address not found in connection"
               Just (rq'@RcvQueue {primary}, rq'' : rqs')
@@ -2337,10 +2341,10 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
                       Right () -> finalizeSwitch
                       Left e
                         | temporaryOrHostError e -> throwE e
-                        | otherwise -> finalizeSwitch >> throwE e
+                        | otherwise -> finalizeSwitch >> notify (ERR e)
                 where
                   finalizeSwitch = do
-                    withStore' c $ \db -> deleteConnRcvQueue db rq'
+                    withStore' c $ \db -> deleteConnRcvQueue db rq' >> deleteCommand db cmdId
                     when enableNtfs $ do
                       ns <- asks ntfSupervisor
                       liftIO $ sendNtfSubCommand ns (NSCCreate, [connId])
@@ -2373,6 +2377,7 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
             SomeConn _ conn@DuplexConnection {} -> a conn
             _ -> internalErr "command requires duplex connection"
         tryCommand action = tryMoveableCommand (action $> CCCompleted)
+        tryAtomicCommand action = tryMoveableCommand (action $> CCDeleted)
         tryMoveableCommand action = withRetryInterval2 ri $ \_ loop -> do
           liftIO $ waitWhileSuspended c
           liftIO $ waitForUserNetwork c
@@ -2388,7 +2393,9 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
               | otherwise -> cmdError e
             Right CCCompleted -> withStore' c (`deleteCommand` cmdId)
             Right CCMoved -> pure () -- command processing moved to another command queue
+            Right CCDeleted -> pure ()
         tryWithLock name = tryCommand . withConnLock c connId name
+        tryAtomicWithLock name = tryAtomicCommand . withConnLock c connId name
         tryMoveableWithLock name = tryMoveableCommand . withConnLock c connId name
         internalErr s = cmdError $ INTERNAL $ s <> ": " <> show (agentCommandTag command)
         cmdError e = notify (ERR e) >> withStore' c (`deleteCommand` cmdId)
@@ -2410,7 +2417,7 @@ enqueueMessages' c cData sqs msgFlags aMessage =
 
 enqueueMessagesB :: Traversable t => AgentClient -> t (Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage)) -> AM' (t (Either AgentErrorType (AgentMsgId, PQEncryption)))
 enqueueMessagesB c reqs = do
-  reqs' <- enqueueMessageB c reqs
+  reqs' <- enqueueMessageB c Nothing reqs
   enqueueSavedMessageB c $ mapMaybe snd $ rights $ toList reqs'
   pure $ fst <$$> reqs'
 
@@ -2420,17 +2427,19 @@ isActiveSndQ ConnData {connAgentVersion} sq@SndQueue {status, sndSwchStatus} =
     && (status == Secured || status == Active || (connAgentVersion >= rpcAddressSMPAgentVersion && securingSndQueue sq))
 {-# INLINE isActiveSndQ #-}
 
-enqueueMessage :: AgentClient -> ConnData -> SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, PQEncryption)
-enqueueMessage c cData sq msgFlags aMessage =
-  ExceptT $ fmap fst . runIdentity <$> enqueueMessageB c (Identity (Right (Right (cData, [sq]), Nothing, msgFlags, vrValue aMessage)))
+enqueueMessage :: AgentClient -> Maybe AsyncCmdId -> ConnData -> SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, PQEncryption)
+enqueueMessage c cmdId_ cData sq msgFlags aMessage =
+  ExceptT $ fmap fst . runIdentity <$> enqueueMessageB c cmdId_ (Identity (Right (Right (cData, [sq]), Nothing, msgFlags, vrValue aMessage)))
 {-# INLINE enqueueMessage #-}
 
 -- this function is used only for sending messages in batch, it returns the list of successes to enqueue additional deliveries
-enqueueMessageB :: forall t. Traversable t => AgentClient -> t (Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage)) -> AM' (t (Either AgentErrorType ((AgentMsgId, PQEncryption), Maybe ([SndQueue], AgentMsgId))))
-enqueueMessageB c reqs = do
+enqueueMessageB :: forall t. Traversable t => AgentClient -> Maybe AsyncCmdId -> t (Either AgentErrorType (Either AgentErrorType (ConnData, NonEmpty SndQueue), Maybe PQEncryption, MsgFlags, ValueOrRef AMessage)) -> AM' (t (Either AgentErrorType ((AgentMsgId, PQEncryption), Maybe ([SndQueue], AgentMsgId))))
+enqueueMessageB c cmdId_ reqs = do
   cfg <- asks config
   (_, reqMids) <- unsafeWithStore c $ \db -> do
-    mapAccumLM (\ids r -> storeSentMsg db cfg ids r `E.catchAny` \e -> (ids,) <$> handleInternal e) IM.empty reqs
+    res@(_, rs) <- mapAccumLM (\ids r -> storeSentMsg db cfg ids r `E.catchAny` \e -> (ids,) <$> handleInternal e) IM.empty reqs
+    when (all isRight rs) $ mapM_ (deleteCommand db) cmdId_
+    pure res
   forME reqMids $ submitSentMsg c
   where
     handleInternal :: E.SomeException -> IO (Either AgentErrorType b)
@@ -2756,8 +2765,8 @@ withConnLockNotify c connId name action = do
   t_ <- withConnLock c connId name action
   forM_ t_ $ atomically . writeTBQueue (subQ c)
 
-ackMessage' :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AM ()
-ackMessage' c connId msgId rcptInfo_ = withConnLockNotify c connId "ackMessage" $ do
+ackMessage' :: AgentClient -> Maybe AsyncCmdId -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AM ()
+ackMessage' c cmdId_ connId msgId rcptInfo_ = withConnLockNotify c connId "ackMessage" $ do
   SomeConn _ conn <- withStore c (`getConn` connId)
   case conn of
     DuplexConnection {} -> do
@@ -2779,7 +2788,7 @@ ackMessage' c connId msgId rcptInfo_ = withConnLockNotify c connId "ackMessage"
       (rq, srvMsgId) <- withStore c $ \db -> setMsgUserAck db connId $ InternalId msgId
       ackQueueMessage c rq srvMsgId
     del :: AM ()
-    del = withStore' c $ \db -> deleteMsg db connId $ InternalId msgId
+    del = withStore' c $ \db -> deleteMsg db connId (InternalId msgId) >> mapM_ (deleteCommand db) cmdId_
     sendRcpt :: Connection 'CDuplex -> AM ()
     sendRcpt (DuplexConnection cData _ sqs) = do
       msg@RcvMsg {msgType, msgReceipt} <- withStore c $ \db -> getRcvMsg db connId $ InternalId msgId
@@ -2814,11 +2823,11 @@ switchConnection' c nm connId =
         | otherwise -> do
             when (ratchetSyncSendProhibited cData) $ throwE $ CMD PROHIBITED "switchConnection: send prohibited"
             rq' <- withStore' c $ \db -> setRcvSwitchStatus db rq $ Just RSSwitchStarted
-            switchDuplexConnection c nm conn rq'
+            switchDuplexConnection c nm Nothing conn rq'
       _ -> throwE $ CMD PROHIBITED "switchConnection: not duplex"
 
-switchDuplexConnection :: AgentClient -> NetworkRequestMode -> Connection 'CDuplex -> RcvQueue -> AM ConnectionStats
-switchDuplexConnection c nm (DuplexConnection cData@ConnData {connId, userId} rqs sqs) rq@RcvQueue {server, dbQueueId = DBEntityId dbQueueId, sndId} = do
+switchDuplexConnection :: AgentClient -> NetworkRequestMode -> Maybe AsyncCmdId -> Connection 'CDuplex -> RcvQueue -> AM ConnectionStats
+switchDuplexConnection c nm cmdId_ (DuplexConnection cData@ConnData {connId, userId} rqs sqs) rq@RcvQueue {server, dbQueueId = DBEntityId dbQueueId, sndId} = do
   checkRQSwchStatus rq RSSwitchStarted
   clientVRange <- asks $ smpClientVRange . config
   -- try to get the server that is different from all queues, or at least from the primary rcv queue
@@ -2831,7 +2840,7 @@ switchDuplexConnection c nm (DuplexConnection cData@ConnData {connId, userId} rq
   rq'' <- withStore c $ \db -> addConnRcvQueue db connId rq' SMSubscribe
   lift $ addNewQueueSubscription c rq'' tSess sessId serviceId_
   void . enqueueMessages c cData sqs SMP.noMsgFlags $ QADD [(qUri, Just (server, sndId))]
-  rq1 <- withStore' c $ \db -> setRcvSwitchStatus db rq $ Just RSSendingQADD
+  rq1 <- withStore' c $ \db -> setRcvSwitchStatus db rq (Just RSSendingQADD) <* mapM_ (deleteCommand db) cmdId_
   let rqs' = updatedQs rq1 rqs <> [rq'']
   connectionStats c $ DuplexConnection cData rqs' sqs
 
@@ -2884,22 +2893,27 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni
     _ -> throwE $ CMD PROHIBITED "synchronizeRatchet: not duplex"
 
 ackQueueMessage :: AgentClient -> RcvQueue -> SMP.MsgId -> AM (Maybe ATransmission)
-ackQueueMessage c rq@RcvQueue {userId, connId, server} srvMsgId = do
+ackQueueMessage c rq@RcvQueue {userId, connId, server, rcvSwchStatus} srvMsgId = do
   atomically $ incSMPServerStat c userId server ackAttempts
   tryAllErrors (sendAck c rq srvMsgId) >>= \case
-    Right _ -> sendMsgNtf ackMsgs
-    Left (SMP _ SMP.NO_MSG) -> sendMsgNtf ackNoMsgErrs
+    Right drained -> sendMsgNtf ackMsgs $ drained && rcvSwchStatus == Just RSReceivedQEND
+    Left (SMP _ SMP.NO_MSG) -> sendMsgNtf ackNoMsgErrs False
     Left e -> do
       unless (temporaryOrHostError e) $ atomically $ incSMPServerStat c userId server ackOtherErrs
       throwE e
   where
-    sendMsgNtf stat = do
+    sendMsgNtf stat ended = do
       atomically $ incSMPServerStat c userId server stat
       ifM (liftIO $ hasGetLock c rq)
         (do atomically $ releaseGetLock c rq
             brokerTs_ <- eitherToMaybe <$> tryAllErrors (withStore c $ \db -> getRcvMsgBrokerTs db connId srvMsgId)
             pure $ Just ("", connId, AEvt SAEConn $ MSGNTF srvMsgId brokerTs_))
-        (pure Nothing)
+        (Nothing <$ when ended (deleteRcvQueueAsync c rq))
+
+deleteRcvQueueAsync :: AgentClient -> RcvQueue -> AM ()
+deleteRcvQueueAsync c rq@RcvQueue {connId, server, rcvId} = do
+  withStore' c (`setRcvQueueDeleted` rq)
+  enqueueCommand c "" connId (Just server) $ AInternalCommand $ ICDeleteRcvQueue rcvId
 
 -- | Suspend SMP agent connection (OFF command) in Reader monad
 suspendConnection' :: AgentClient -> NetworkRequestMode -> ConnId -> AM ()
@@ -3429,6 +3443,7 @@ cleanupManager c@AgentClient {subQ} = do
     run SFERR deleteSndFilesPrefixPaths
     run SFERR deleteExpiredReplicasForDeletion
     run ERR deleteExpiredServiceReqs
+    run ERR expireEndedRcvQueues
     liftIO $ threadDelay' int
   where
     run :: forall e. AEntityI e => (AgentErrorType -> AEvent e) -> AM () -> AM' ()
@@ -3486,6 +3501,20 @@ cleanupManager c@AgentClient {subQ} = do
         deleteExpiredServiceRequests db $ addUTCTime (negate serviceResponseTimeout) now
         getExpiredServiceConns db now
       deleteConnectionsAsync' c False expiredConns
+    expireEndedRcvQueues = do
+      maxErrs <- asks $ deleteErrorCount . config
+      connIds <- withStore' c getEndedRcvQueueConnIds
+      forM_ connIds $ \connId -> flip catchAllErrors (notify connId . ERR) $
+        withConnLock c connId "cleanupManager" $
+          withStore c (`getConn` connId) >>= \case
+            SomeConn _ (DuplexConnection _ rqs _) -> do
+              let (ended, current) = L.partition ((Just RSReceivedQEND ==) . rcvSwchStatus) rqs
+              whenM (anyM $ map (atomically . hasActiveSubscription c) current) $
+                forM_ ended $ \rq ->
+                  if deleteErrors rq + 1 < maxErrs
+                    then withStore' c (`incRcvDeleteErrors` rq)
+                    else deleteRcvQueueAsync c rq
+            _ -> pure ()
     notify :: forall e. AEntityI e => AEntityId -> AEvent e -> AM ()
     notify entId cmd = atomically $ writeTBQueue subQ ("", entId, AEvt (sAEntity @e) cmd)
 
@@ -3643,7 +3672,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                               HELLO -> helloMsg srvMsgId msgMeta conn'' >> ackDel msgId
                               -- note that there is no ACK sent for A_MSG, it is sent with agent's user ACK command
                               A_MSG body -> do
-                                logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret' srvMsgId
+                                logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret' (SMP.unMsgId srvMsgId)
                                 notify $ MSG msgMeta msgFlags body
                                 pure ACKPending
                               A_RCVD rcpts -> qDuplex conn'' "RCVD" $ messagesRcvd rcpts msgMeta
@@ -3653,7 +3682,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                               QUSE qs -> qDuplexAckDel conn'' "QUSE" $ qUseMsg srvMsgId qs
                               -- no action needed for QTEST
                               -- any message in the new queue will mark it active and trigger deletion of the old queue
-                              QTEST _ -> logServer "<--" c srv rId ("MSG <QTEST>:" <> logSecret' srvMsgId) >> ackDel msgId
+                              QTEST _ -> logServer "<--" c srv rId ("MSG <QTEST>:" <> logSecret' (SMP.unMsgId srvMsgId)) >> ackDel msgId
                               QEND addrs -> qDuplexAckDel conn'' "QEND" $ qEndMsg srvMsgId addrs
                               EREADY _ -> qDuplexAckDel conn'' "EREADY" $ ereadyMsg rcPrev
                             where
@@ -3688,7 +3717,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                                       else
                                         liftEither (parse smpP (AGENT $ A_MESSAGE "parse msg body 1") agentMsgBody) >>= \case
                                           AgentMessage _ (A_MSG body) -> do
-                                            logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret' srvMsgId
+                                            logServer "<--" c srv rId $ "MSG <MSG>:" <> logSecret' (SMP.unMsgId srvMsgId)
                                             notify $ MSG msgMeta msgFlags body
                                             pure ACKPending
                                           _ -> ack
@@ -3834,7 +3863,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
 
           smpConfirmation :: SMP.MsgId -> Connection c -> Maybe C.APublicAuthKey -> C.PublicKeyX25519 -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> ByteString -> VersionSMPC -> VersionSMPA -> AM ()
           smpConfirmation srvMsgId conn' senderKey e2ePubKey e2eEncryption encConnInfo phVer agentVersion = do
-            logServer "<--" c srv rId $ "MSG <CONF>:" <> logSecret' srvMsgId
+            logServer "<--" c srv rId $ "MSG <CONF>:" <> logSecret' (SMP.unMsgId srvMsgId)
             checkConfVersions agentVersion phVer
             let ConnData {pqSupport, serviceRequestExpiresAt} = toConnData conn'
             case status of
@@ -3943,7 +3972,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
 
           helloMsg :: SMP.MsgId -> MsgMeta -> Connection c -> AM ()
           helloMsg srvMsgId MsgMeta {pqEncryption} conn' = do
-            logServer "<--" c srv rId $ "MSG <HELLO>:" <> logSecret' srvMsgId
+            logServer "<--" c srv rId $ "MSG <HELLO>:" <> logSecret' (SMP.unMsgId srvMsgId)
             case status of
               Active -> prohibited "hello: active"
               _ ->
@@ -3961,13 +3990,13 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
               enqueueDuplexHello :: SndQueue -> AM ()
               enqueueDuplexHello sq = do
                 let cData' = toConnData conn'
-                void $ enqueueMessage c cData' sq SMP.MsgFlags {notification = True} HELLO
+                void $ enqueueMessage c Nothing cData' sq SMP.MsgFlags {notification = True} HELLO
 
           continueSending :: SMP.MsgId -> (SMPServer, SMP.SenderId) -> Connection 'CDuplex -> AM ()
           continueSending srvMsgId addr (DuplexConnection _ _ sqs) =
             case findQ addr sqs of
               Just sq -> do
-                logServer "<--" c srv rId $ "MSG <QCONT>:" <> logSecret' srvMsgId
+                logServer "<--" c srv rId $ "MSG <QCONT>:" <> logSecret' (SMP.unMsgId srvMsgId)
                 atomically $
                   TM.lookup (qAddress sq) (smpDeliveryWorkers c)
                     >>= mapM_ (\(_, retryLock) -> tryPutTMVar retryLock ())
@@ -3976,7 +4005,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
 
           messagesRcvd :: NonEmpty AMessageReceipt -> MsgMeta -> Connection 'CDuplex -> AM ACKd
           messagesRcvd rcpts msgMeta@MsgMeta {broker = (srvMsgId, _)} _ = do
-            logServer "<--" c srv rId $ "MSG <RCPT>:" <> logSecret' srvMsgId
+            logServer "<--" c srv rId $ "MSG <RCPT>:" <> logSecret' (SMP.unMsgId srvMsgId)
             rs <- forM rcpts $ \rcpt -> clientReceipt rcpt `catchAllErrors` \e -> notify (ERR e) $> Nothing
             case L.nonEmpty . catMaybes $ L.toList rs of
               Just rs' -> notify (RCVD msgMeta rs') $> ACKPending
@@ -4017,7 +4046,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                           lockConnForUpdate db connId
                           liftIO $ mapM_ (deleteConnSndQueue db connId) delSqs
                           addConnSndQueue db connId (sq_ :: NewSndQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
-                        logServer "<--" c srv rId $ "MSG <QADD>:" <> logSecret' srvMsgId <> " " <> logSecret (senderId queueAddress)
+                        logServer "<--" c srv rId $ "MSG <QADD>:" <> logSecret' (SMP.unMsgId srvMsgId) <> " " <> logSecret (senderId queueAddress)
                         swchStatus <-
                           if connAgentVersion >= rpcAddressSMPAgentVersion
                             then do
@@ -4047,7 +4076,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
               Just rq'@RcvQueue {rcvId, e2ePrivKey = dhPrivKey, smpClientVersion = cVer, status = status'}
                 | status' == New || status' == Confirmed -> do
                     checkRQSwchStatus rq RSSendingQADD
-                    logServer "<--" c srv rId $ "MSG <QKEY>:" <> logSecret' srvMsgId <> " " <> logSecret senderId
+                    logServer "<--" c srv rId $ "MSG <QKEY>:" <> logSecret' (SMP.unMsgId srvMsgId) <> " " <> logSecret senderId
                     let dhSecret = C.dh' dhPublicKey dhPrivKey
                     withStore' c $ \db -> setRcvQueueConfirmedE2E db rq' dhSecret $ min cVer cVer'
                     enqueueCommand c "" connId (Just smpServer) $ AInternalCommand $ ICQSecure rcvId senderKey
@@ -4069,7 +4098,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                 case find ((replaceQId ==) . dbQId) sqs of
                   Just sq1 -> do
                     checkSQSwchStatus sq1 SSSendingQKEY
-                    logServer "<--" c srv rId $ "MSG <QUSE>:" <> logSecret' srvMsgId <> " " <> logSecret (snd addr)
+                    logServer "<--" c srv rId $ "MSG <QUSE>:" <> logSecret' (SMP.unMsgId srvMsgId) <> " " <> logSecret (snd addr)
                     withStore' c $ \db -> setSndQueueStatus db sq' Secured
                     let sq'' = (sq' :: SndQueue) {status = Secured}
                     -- sending QTEST to the new queue only, the old one will be removed if sent successfully
@@ -4085,19 +4114,23 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
           -- processed by queue recipient
           qEndMsg :: SMP.MsgId -> NonEmpty SndQAddr -> Connection 'CDuplex -> AM ()
           qEndMsg srvMsgId addrs (DuplexConnection cData'@ConnData {enableNtfs} rqs sqs) =
-            case L.partition (\rq' -> any (`sameQAddress` sndAddress rq') addrs) rqs of
+            case L.partition isRemoved rqs of
               (removed@(_ : _), keptRq : keptRqs) -> do
-                logServer "<--" c srv rId $ "MSG <QEND>:" <> logSecret' srvMsgId
-                forM_ removed $ \rq'@RcvQueue {server = rmServer, rcvId} -> do
-                  withStore' c $ \db -> setRcvQueueDeleted db rq'
-                  enqueueCommand c "" connId (Just rmServer) $ AInternalCommand $ ICDeleteRcvQueue rcvId
-                when enableNtfs $ do
-                  ns <- asks ntfSupervisor
-                  liftIO $ sendNtfSubCommand ns (NSCCreate, [connId])
-                let conn' = DuplexConnection cData' (keptRq :| keptRqs) sqs
-                cStats <- connectionStats c conn'
-                notify $ SWITCH QDRcv SPCompleted cStats
+                logServer "<--" c srv rId $ "MSG <QEND>:" <> logSecret' (SMP.unMsgId srvMsgId)
+                let completed = any ((Just RSReceivedQEND ==) . rcvSwchStatus) removed
+                if isRemoved rq
+                  then mapM_ (deleteRcvQueueAsync c) removed
+                  else withStore' c $ \db -> forM_ removed $ \rq' -> setRcvSwitchStatus db rq' $ Just RSReceivedQEND
+                unless completed $ do
+                  when enableNtfs $ do
+                    ns <- asks ntfSupervisor
+                    liftIO $ sendNtfSubCommand ns (NSCCreate, [connId])
+                  let conn' = DuplexConnection cData' (keptRq :| keptRqs) sqs
+                  cStats <- connectionStats c conn'
+                  notify $ SWITCH QDRcv SPCompleted cStats
               _ -> pure ()
+            where
+              isRemoved rq' = any (`sameQAddress` sndAddress rq') addrs
 
           qError :: String -> AM a
           qError = throwE . AGENT . A_QUEUE
@@ -4111,7 +4144,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
 
           smpInvitation :: SMP.MsgId -> Connection c -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> AM ()
           smpInvitation srvMsgId conn' connReq@(CRInvitationUri crData (CR.E2ERatchetParamsUri _ k1 k2 kem_)) cInfo = do
-            logServer "<--" c srv rId $ "MSG <KEY>:" <> logSecret' srvMsgId
+            logServer "<--" c srv rId $ "MSG <KEY>:" <> logSecret' (SMP.unMsgId srvMsgId)
             case conn' of
               ContactConnection _ RcvQueue {sndId} -> do
                 -- show connection request even if invitaion via contact address is not compatible.
@@ -4128,7 +4161,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
 
           smpContactRequest :: SMP.MsgId -> Connection c -> VersionSMPA -> CR.SndE2ERatchetParams 'C.X448 -> RatchetKeyId -> ByteString -> VersionSMPC -> AM ()
           smpContactRequest srvMsgId conn' agentVersion e2eSndParams ratchetKeyId encConnInfo phVer = do
-            logServer "<--" c srv rId $ "MSG <KEY>:" <> logSecret' srvMsgId
+            logServer "<--" c srv rId $ "MSG <KEY>:" <> logSecret' (SMP.unMsgId srvMsgId)
             case conn' of
               ContactConnection {} -> do
                 checkConfVersions agentVersion phVer
@@ -4276,8 +4309,8 @@ switchStatusError q expected actual =
       <> (", actual=" <> show actual)
 
 -- used only in background
-connectReplyQueues :: AgentClient -> ConnData -> ConnInfo -> Maybe SndQueue -> NonEmpty SMPQueueInfo -> AM ()
-connectReplyQueues c cData@ConnData {userId, connId} ownConnInfo sq_ (qInfo :| _) = do
+connectReplyQueues :: AgentClient -> AsyncCmdId -> ConnData -> ConnInfo -> Maybe SndQueue -> NonEmpty SMPQueueInfo -> AM ()
+connectReplyQueues c cmdId cData@ConnData {userId, connId} ownConnInfo sq_ (qInfo :| _) = do
   clientVRange <- asks $ smpClientVRange . config
   case qInfo `proveCompatible` clientVRange of
     Nothing -> throwE $ AGENT A_VERSION
@@ -4285,17 +4318,17 @@ connectReplyQueues c cData@ConnData {userId, connId} ownConnInfo sq_ (qInfo :| _
       -- in case of SKEY retry the connection is already duplex
       sq' <- maybe upgradeConn pure sq_
       void $ agentSecureSndQueue c NRMBackground cData sq'
-      enqueueConfirmation c cData sq' ownConnInfo Nothing
+      enqueueConfirmation c cmdId cData sq' ownConnInfo Nothing
       where
         upgradeConn = do
           (sq, _) <- lift $ newSndQueue userId connId qInfo' Nothing
           withStore c $ \db -> upgradeRcvConnToDuplex db connId sq
 
-secureConfirmQueueAsync :: AgentClient -> ConnData -> Maybe RcvQueue -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> SubscriptionMode -> AM SndQueueSecured
-secureConfirmQueueAsync c cData rq_ sq srv connInfo e2eEncryption_ subMode = do
+secureConfirmQueueAsync :: AgentClient -> AsyncCmdId -> ConnData -> Maybe RcvQueue -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> SubscriptionMode -> AM SndQueueSecured
+secureConfirmQueueAsync c cmdId cData rq_ sq srv connInfo e2eEncryption_ subMode = do
   sqSecured <- agentSecureSndQueue c NRMBackground cData sq
   qInfo <- mkAgentConfirmation c NRMBackground cData rq_ sq srv connInfo subMode
-  storeConfirmation c cData sq e2eEncryption_ qInfo
+  storeConfirmation c (Just cmdId) cData sq e2eEncryption_ qInfo
   lift $ submitPendingMsg c sq
   pure sqSecured
 
@@ -4339,13 +4372,13 @@ mkAgentConfirmation c nm cData rq_ sq srv connInfo subMode = do
     Just rq@RcvQueue {smpClientVersion = v} -> pure $ SMPQueueInfo v $ rcvSMPQueueAddress rq
   pure $ AgentConnInfoReply (qInfo :| []) connInfo
 
-enqueueConfirmation :: AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> AM ()
-enqueueConfirmation c cData sq connInfo e2eEncryption_ = do
-  storeConfirmation c cData sq e2eEncryption_ $ AgentConnInfo connInfo
+enqueueConfirmation :: AgentClient -> AsyncCmdId -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> AM ()
+enqueueConfirmation c cmdId cData sq connInfo e2eEncryption_ = do
+  storeConfirmation c (Just cmdId) cData sq e2eEncryption_ $ AgentConnInfo connInfo
   lift $ submitPendingMsg c sq
 
-storeConfirmation :: AgentClient -> ConnData -> SndQueue -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> AgentMessage -> AM ()
-storeConfirmation c cData@ConnData {connId, pqSupport, connAgentVersion = v} sq e2eEncryption_ agentMsg = do
+storeConfirmation :: AgentClient -> Maybe AsyncCmdId -> ConnData -> SndQueue -> Maybe (CR.SndE2ERatchetParams 'C.X448) -> AgentMessage -> AM ()
+storeConfirmation c cmdId_ cData@ConnData {connId, pqSupport, connAgentVersion = v} sq e2eEncryption_ agentMsg = do
   currentE2EVersion <- asks $ maxVersion . e2eEncryptVRange . config
   withStore c $ \db -> runExceptT $ do
     internalTs <- liftIO getCurrentTime
@@ -4359,6 +4392,7 @@ storeConfirmation c cData@ConnData {connId, pqSupport, connAgentVersion = v} sq
         msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, pqEncryption, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, sndMsgPrepData_ = Nothing}
     liftIO $ createSndMsg db connId msgData
     liftIO $ createSndMsgDelivery db sq internalId
+    liftIO $ mapM_ (deleteCommand db) cmdId_
 
 enqueueRatchetKeyMsgs :: AgentClient -> ConnData -> NonEmpty SndQueue -> InternalId -> AM' ()
 enqueueRatchetKeyMsgs c cData (sq :| sqs) (InternalId msgId) = do

src/Simplex/Messaging/Agent/Client.hs

@@ -1982,7 +1982,7 @@ decryptSMPMessage :: RcvQueue -> SMP.RcvMessage -> AM SMP.ClientRcvMsgBody
 decryptSMPMessage rq SMP.RcvMessage {msgId, msgBody = SMP.EncRcvMsgBody body} =
   liftEither $ parse SMP.clientRcvMsgBodyP (AGENT $ A_MESSAGE "decrypt message") =<< decrypt body
   where
-    decrypt = agentCbDecrypt (rcvDhSecret rq) (C.cbNonce msgId)
+    decrypt = agentCbDecrypt (rcvDhSecret rq) msgId
 
 secureQueue :: AgentClient -> NetworkRequestMode -> RcvQueue -> SndPublicAuthKey -> AM ()
 secureQueue c nm rq@RcvQueue {rcvId, rcvPrivateKey} senderKey =
@@ -2080,9 +2080,9 @@ disableQueuesNtfs c = sendTSessionBatches "NDEL" (mkSMPTSession . snd) disableQu
     queueCreds :: DisableQueueNtfReq -> (SMP.RecipientId, SMP.RcvPrivateAuthKey)
     queueCreds (_, RcvQueue {rcvPrivateKey, rcvId}) = (rcvId, rcvPrivateKey)
 
-sendAck :: AgentClient -> RcvQueue -> MsgId -> AM ()
+sendAck :: AgentClient -> RcvQueue -> MsgId -> AM QueueDrained
 sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId =
-  withSMPClient c NRMBackground rq ("ACK:" <> logSecret' msgId) $ \smp ->
+  withSMPClient c NRMBackground rq ("ACK:" <> logSecret' (SMP.unMsgId msgId)) $ \smp ->
     ackSMPMessage smp rcvPrivateKey rcvId msgId
 
 hasGetLock :: SomeRcvQueue q => AgentClient -> q -> IO Bool
@@ -2277,7 +2277,7 @@ agentCbEncryptOnce clientVersion dhRcvPubKey msg = do
 
 -- | NaCl crypto-box decrypt - both for messages received from the server
 -- and per-queue E2E encrypted messages from the sender that were inside.
-agentCbDecrypt :: C.DhSecretX25519 -> C.CbNonce -> ByteString -> Either AgentErrorType ByteString
+agentCbDecrypt :: C.CbNonceI n => C.DhSecretX25519 -> n -> ByteString -> Either AgentErrorType ByteString
 agentCbDecrypt dhSecret nonce msg =
   first cryptoError $
     C.cbDecrypt dhSecret nonce msg

src/Simplex/Messaging/Agent/Protocol.hs

@@ -612,6 +612,7 @@ data RcvSwitchStatus
   | RSSendingQADD
   | RSSendingQUSE
   | RSReceivedMessage
+  | RSReceivedQEND
   deriving (Eq, Show)
 
 instance StrEncoding RcvSwitchStatus where
@@ -620,12 +621,14 @@ instance StrEncoding RcvSwitchStatus where
     RSSendingQADD -> "sending_qadd"
     RSSendingQUSE -> "sending_quse"
     RSReceivedMessage -> "received_message"
+    RSReceivedQEND -> "received_qend"
   strP =
     A.takeTill (== ' ') >>= \case
       "switch_started" -> pure RSSwitchStarted
       "sending_qadd" -> pure RSSendingQADD
       "sending_quse" -> pure RSSendingQUSE
       "received_message" -> pure RSReceivedMessage
+      "received_qend" -> pure RSReceivedQEND
       _ -> fail "bad RcvSwitchStatus"
 
 instance ToField RcvSwitchStatus where toField = toField . decodeLatin1 . strEncode

src/Simplex/Messaging/Agent/Store.hs

@@ -219,6 +219,7 @@ canAbortRcvSwitch ConnData {connAgentVersion} = maybe False canAbort . rcvSwchSt
       -- if switch is in RSReceivedMessage status, aborting switch (deleting new queue)
       -- will break the connection because the sender would have original queue deleted
       RSReceivedMessage -> False
+      RSReceivedQEND -> False
 
 data ClientNtfCreds = ClientNtfCreds
   { -- | key pair to be used by the notification server to authorize transmissions

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

@@ -100,6 +100,7 @@ module Simplex.Messaging.Agent.Store.AgentStore
     setSndQueuePrimary,
     deleteConnRcvQueue,
     incRcvDeleteErrors,
+    getEndedRcvQueueConnIds,
     deleteConnSndQueue,
     getPrimaryRcvQueue,
     getRcvQueue,
@@ -763,6 +764,10 @@ incRcvDeleteErrors :: DB.Connection -> RcvQueue -> IO ()
 incRcvDeleteErrors db RcvQueue {connId, dbQueueId} =
   DB.execute db "UPDATE rcv_queues SET delete_errors = delete_errors + 1 WHERE conn_id = ? AND rcv_queue_id = ?" (connId, dbQueueId)
 
+getEndedRcvQueueConnIds :: DB.Connection -> IO [ConnId]
+getEndedRcvQueueConnIds db =
+  map fromOnly <$> DB.query db "SELECT conn_id FROM rcv_queues WHERE switch_status = ? AND deleted = 0" (Only RSReceivedQEND)
+
 deleteConnRcvQueue :: DB.Connection -> RcvQueue -> IO ()
 deleteConnRcvQueue db RcvQueue {connId, dbQueueId} =
   DB.execute db "DELETE FROM rcv_queues WHERE conn_id = ? AND rcv_queue_id = ?" (connId, dbQueueId)
@@ -1255,7 +1260,7 @@ getLastMsg db connId msgId =
         LEFT JOIN snd_messages s ON s.conn_id = r.conn_id AND s.rcpt_internal_id = r.internal_id
         WHERE r.conn_id = ? AND r.broker_id = ?
       |]
-      (connId, Binary msgId)
+      (connId, msgId)
 
 toRcvMsg :: (Int64, InternalTs, BrokerId, BrokerTs) :. (AgentMsgId, MsgIntegrity, MsgHash, AgentMessageType, MsgBody, PQEncryption, Maybe AgentMsgId, Maybe MsgReceiptStatus, BoolInt) -> RcvMsg
 toRcvMsg ((agentMsgId, internalTs, brokerId, brokerTs) :. (sndMsgId, integrity, internalHash, msgType, msgBody, pqEncryption, rcptInternalId_, rcptStatus_, BI userAck)) =
@@ -1287,7 +1292,7 @@ checkRcvMsgHashExists db connId hash =
 getRcvMsgBrokerTs :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Either StoreError BrokerTs)
 getRcvMsgBrokerTs db connId msgId =
   firstRow fromOnly (SEMsgNotFound "getRcvMsgBrokerTs") $
-    DB.query db "SELECT broker_ts FROM rcv_messages WHERE conn_id = ? AND broker_id = ?" (connId, Binary msgId)
+    DB.query db "SELECT broker_ts FROM rcv_messages WHERE conn_id = ? AND broker_id = ?" (connId, msgId)
 
 deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
 deleteMsg db connId msgId =
@@ -3035,7 +3040,7 @@ insertRcvMsgDetails_ db connId RcvQueue {dbQueueId} RcvMsgData {msgMeta, interna
       VALUES
         (?,?,?,?,?,?,?,?,?,?)
     |]
-    (connId, dbQueueId, internalRcvId, fst recipient, sndMsgId, Binary (fst broker), snd broker, Binary internalHash, Binary externalPrevSndHash, integrity)
+    (connId, dbQueueId, internalRcvId, fst recipient, sndMsgId, fst broker, snd broker, Binary internalHash, Binary externalPrevSndHash, integrity)
   DB.execute db "INSERT INTO encrypted_rcv_message_hashes (conn_id, hash) VALUES (?,?)" (connId, Binary encryptedMsgHash)
 
 updateRcvMsgHash :: DB.Connection -> ConnId -> AgentMsgId -> InternalRcvId -> MsgHash -> IO ()
@@ -3183,7 +3188,8 @@ createRcvFile db gVar userId fd@FileDescription {chunks} prefixPath tmpPath file
 createRcvFileRedirect :: DB.Connection -> TVar ChaChaDRG -> UserId -> FileDescription 'FRecipient -> FilePath -> FilePath -> CryptoFile -> FilePath -> CryptoFile -> Bool -> IO (Either StoreError RcvFileId)
 createRcvFileRedirect _ _ _ FileDescription {redirect = Nothing} _ _ _ _ _ _ = pure $ Left $ SEInternal "createRcvFileRedirect called without redirect"
 createRcvFileRedirect db gVar userId redirectFd@FileDescription {chunks = redirectChunks, redirect = Just RedirectFileInfo {size, digest}} prefixPath redirectPath redirectFile dstPath dstFile approvedRelays = runExceptT $ do
-  (dstEntityId, dstId) <- ExceptT $ insertRcvFile db gVar userId dummyDst prefixPath dstPath dstFile Nothing Nothing approvedRelays
+  nonce <- atomically $ C.randomCbNonce gVar
+  (dstEntityId, dstId) <- ExceptT $ insertRcvFile db gVar userId (dummyDst nonce) prefixPath dstPath dstFile Nothing Nothing approvedRelays
   (_, redirectId) <- ExceptT $ insertRcvFile db gVar userId redirectFd prefixPath redirectPath redirectFile (Just dstId) (Just dstEntityId) approvedRelays
   liftIO $
     forM_ redirectChunks $ \fc@FileChunk {replicas} -> do
@@ -3191,7 +3197,7 @@ createRcvFileRedirect db gVar userId redirectFd@FileDescription {chunks = redire
       forM_ (zip [1 ..] replicas) $ \(rno, replica) -> insertRcvFileChunkReplica db rno replica chunkId
   pure dstEntityId
   where
-    dummyDst =
+    dummyDst nonce =
       FileDescription
         { party = SFRecipient,
           size,
@@ -3199,7 +3205,7 @@ createRcvFileRedirect db gVar userId redirectFd@FileDescription {chunks = redire
           redirect = Nothing,
           -- updated later with updateRcvFileRedirect
           key = C.unsafeSbKey $ B.replicate 32 '#',
-          nonce = C.cbNonce "",
+          nonce,
           chunkSize = FileSize 0,
           chunks = []
         }

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

@@ -100,7 +100,8 @@ withTransaction st = withTransactionPriority st False
 {-# INLINE withTransaction #-}
 
 withTransactionPriority :: DBStore -> Bool -> (DB.Connection -> IO a) -> IO a
-withTransactionPriority st priority action = withConnectionPriority st priority $ dbBusyLoop . transaction
+withTransactionPriority st@DBStore {dbClosed} priority action = withConnectionPriority st priority $ \db ->
+  ifM (readTVarIO dbClosed) (fail "withTransaction: database closed") (dbBusyLoop $ transaction db)
   where
     transaction db@DB.Connection {conn} = SQL.withImmediateTransaction conn $ action db
 

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

@@ -53,6 +53,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260712_address_dr_rpc
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260823_snd_files_entitlement
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260919_ratchet_verify_codes
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20260929_ratchet_indexes
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20261005_commands_autoincrement
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Query, Maybe Query)]
@@ -105,7 +106,8 @@ schemaMigrations =
     ("m20260712_address_dr_rpc", m20260712_address_dr_rpc, Just down_m20260712_address_dr_rpc),
     ("m20260823_snd_files_entitlement", m20260823_snd_files_entitlement, Just down_m20260823_snd_files_entitlement),
     ("m20260919_ratchet_verify_codes", m20260919_ratchet_verify_codes, Just down_m20260919_ratchet_verify_codes),
-    ("m20260929_ratchet_indexes", m20260929_ratchet_indexes, Just down_m20260929_ratchet_indexes)
+    ("m20260929_ratchet_indexes", m20260929_ratchet_indexes, Just down_m20260929_ratchet_indexes),
+    ("m20261005_commands_autoincrement", m20261005_commands_autoincrement, Just down_m20261005_commands_autoincrement)
   ]
 
 -- | The list of migrations in ascending order by date

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

@@ -0,0 +1,33 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20261005_commands_autoincrement where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20261005_commands_autoincrement :: Query
+m20261005_commands_autoincrement =
+  [sql|
+INSERT INTO sqlite_sequence (name, seq)
+SELECT 'commands', MAX(ROWID) FROM commands;
+
+PRAGMA writable_schema=1;
+
+UPDATE sqlite_master SET sql = replace(sql, 'command_id INTEGER PRIMARY KEY,', 'command_id INTEGER PRIMARY KEY AUTOINCREMENT,')
+WHERE name = 'commands' AND type = 'table';
+
+PRAGMA writable_schema=RESET;
+  |]
+
+down_m20261005_commands_autoincrement :: Query
+down_m20261005_commands_autoincrement =
+  [sql|
+DELETE FROM sqlite_sequence WHERE name = 'commands';
+
+PRAGMA writable_schema=1;
+
+UPDATE sqlite_master SET sql = replace(sql, 'command_id INTEGER PRIMARY KEY AUTOINCREMENT,', 'command_id INTEGER PRIMARY KEY,')
+WHERE name = 'commands' AND type = 'table';
+
+PRAGMA writable_schema=RESET;
+  |]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -247,7 +247,7 @@ CREATE TABLE ntf_subscriptions(
   ON DELETE RESTRICT ON UPDATE CASCADE
 ) WITHOUT ROWID, STRICT;
 CREATE TABLE commands(
-  command_id INTEGER PRIMARY KEY,
+  command_id INTEGER PRIMARY KEY AUTOINCREMENT,
   conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
   host TEXT,
   port TEXT,
@@ -261,6 +261,7 @@ CREATE TABLE commands(
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE
 ) STRICT;
+CREATE TABLE sqlite_sequence(name,seq);
 CREATE TABLE snd_message_deliveries(
   snd_message_delivery_id INTEGER PRIMARY KEY AUTOINCREMENT,
   conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
@@ -269,7 +270,6 @@ CREATE TABLE snd_message_deliveries(
   failed INTEGER DEFAULT 0,
   FOREIGN KEY(conn_id, internal_id) REFERENCES messages ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED
 ) STRICT;
-CREATE TABLE sqlite_sequence(name,seq);
 CREATE TABLE users(
   user_id INTEGER PRIMARY KEY AUTOINCREMENT
   ,

src/Simplex/Messaging/Client.hs

@@ -67,6 +67,7 @@ module Simplex.Messaging.Client
     enableSMPQueuesNtfs,
     disableSMPQueuesNtfs,
     sendSMPMessage,
+    QueueDrained,
     ackSMPMessage,
     suspendSMPQueue,
     deleteSMPQueue,
@@ -150,7 +151,7 @@ import Data.Int (Int64)
 import Data.List (find, isSuffixOf)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
-import Data.Maybe (catMaybes, fromMaybe)
+import Data.Maybe (catMaybes, fromMaybe, isJust)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock (UTCTime (..), diffUTCTime, getCurrentTime)
@@ -442,9 +443,12 @@ defaultNetworkConfig =
 
 transportClientConfig :: NetworkConfig -> NetworkRequestMode -> TransportHost -> Bool -> Maybe [ALPN] -> TransportClientConfig
 transportClientConfig NetworkConfig {socksProxy, socksMode, tcpConnectTimeout, tcpKeepAlive, logTLSErrors} nm host useSNI clientALPN =
-  TransportClientConfig {socksProxy = useSocksProxy socksMode, tcpConnectTimeout = tOut, tcpKeepAlive, logTLSErrors, clientCredentials = Nothing, clientALPN, useSNI}
+  TransportClientConfig {socksProxy = socks, tcpConnectTimeout = tOut, tcpKeepAlive = keepAlive, logTLSErrors, clientCredentials = Nothing, clientALPN, useSNI}
   where
     tOut = netTimeoutInt tcpConnectTimeout nm
+    socks = useSocksProxy socksMode
+    -- with SOCKS proxy the socket is connected to the proxy, and a slow circuit should not fail the connection
+    keepAlive = (\ka -> if isJust socks then ka {unackedDataTimeout = Nothing} else ka) <$> tcpKeepAlive
     socksProxy' = (\(SocksProxyWithAuth _ proxy) -> proxy) <$> socksProxy
     useSocksProxy SMAlways = socksProxy'
     useSocksProxy SMOnion = case host of
@@ -1087,15 +1091,17 @@ resolvedNameOrNotFound d NameResponse {registration} = case registration of
   NRRegistered {nameRecord} -> T.toLower (nrName nameRecord) == fullDomainName d
   _ -> True
 
+type QueueDrained = Bool
+
 -- | Acknowledge message delivery (server deletes the message).
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#acknowledge-message-delivery
 -- This command is always sent in background request mode
-ackSMPMessage :: SMPClient -> RcvPrivateAuthKey -> QueueId -> MsgId -> ExceptT SMPClientError IO ()
+ackSMPMessage :: SMPClient -> RcvPrivateAuthKey -> QueueId -> MsgId -> ExceptT SMPClientError IO QueueDrained
 ackSMPMessage c rpKey rId msgId =
   sendSMPCommand c NRMBackground (Just rpKey) rId (ACK msgId) >>= \case
-    OK -> return ()
-    cmd@MSG {} -> liftIO $ writeSMPMessage c rId cmd
+    OK -> pure True
+    cmd@MSG {} -> liftIO (writeSMPMessage c rId cmd) $> False
     r -> throwE $ unexpectedResponse r
 
 -- | Irreversibly suspend SMP queue.
@@ -1228,7 +1234,7 @@ proxySMPCommand c@ProtocolClient {thParams = proxyThParams, client_ = PClient {c
   -- encode
   let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth serverThParams (CorrId corrId, sId, Cmd (sParty @p) command)
   -- serviceAuth is False here – proxied commands are not used with service certificates
-  auth <- liftEitherWith PCETransportError $ authTransmission serverThAuth False spKey nonce tForAuth
+  auth <- liftEitherWith PCETransportError $ authTransmission serverThAuth False spKey (Just nonce) tForAuth
   b <- case batchTransmissions serverThParams [Right (auth, tToSend)] of
     [] -> throwE $ PCETransportError TELargeMsg
     TBError e _ : _ -> throwE $ PCETransportError e
@@ -1262,7 +1268,7 @@ proxySMPCommand c@ProtocolClient {thParams = proxyThParams, client_ = PClient {c
 -- receives RRES :: Maybe C.CbNonce -> EncFwdResponse -> BrokerMsg
 -- proxy should send PRES to the client with EncResponse
 -- Always uses background timeout mode
-forwardSMPTransmission :: SMPClient -> CorrId -> VersionSMP -> C.PublicKeyX25519 -> EncTransmission -> ExceptT SMPClientError IO (Maybe C.CbNonce, EncResponse)
+forwardSMPTransmission :: SMPClient -> C.CorrCbNonce -> VersionSMP -> C.PublicKeyX25519 -> EncTransmission -> ExceptT SMPClientError IO (Maybe C.CbNonce, EncResponse)
 forwardSMPTransmission c@ProtocolClient {thParams, client_ = PClient {clientCorrId = g}} fwdCorrId fwdVersion fwdKey fwdTransmission = do
   -- prepare params
   sessSecret <- case thAuth thParams of
@@ -1400,7 +1406,7 @@ mkTransmission_ :: forall v err msg. Protocol v err msg => ProtocolClient v err
 mkTransmission_ ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands}} nonce_ (entityId, pKey_, command) = do
   nonce@(C.CbNonce corrId) <- maybe (atomically $ C.randomCbNonce clientCorrId) pure nonce_
   let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (CorrId corrId, entityId, command)
-      auth = authTransmission (thAuth thParams) (useServiceAuth command) pKey_ nonce tForAuth
+      auth = authTransmission (thAuth thParams) (useServiceAuth command) pKey_ (Just nonce) tForAuth
   r <- mkRequest (CorrId corrId)
   pure ((,tToSend) <$> auth, r)
   where
@@ -1419,14 +1425,14 @@ mkTransmission_ ProtocolClient {thParams, client_ = PClient {clientCorrId, sentC
       atomically $ TM.insert corrId r sentCommands
       pure r
 
-authTransmission :: Maybe (THandleAuth 'TClient) -> Bool -> Maybe C.APrivateAuthKey -> C.CbNonce -> ByteString -> Either TransportError (Maybe TAuthorizations)
-authTransmission thAuth serviceAuth pKey_ nonce t = traverse authenticate pKey_
+authTransmission :: Maybe (THandleAuth 'TClient) -> Bool -> Maybe C.APrivateAuthKey -> Maybe C.CbNonce -> ByteString -> Either TransportError (Maybe TAuthorizations)
+authTransmission thAuth serviceAuth pKey_ nonce_ t = traverse authenticate pKey_
   where
     authenticate :: C.APrivateAuthKey -> Either TransportError TAuthorizations
     authenticate (C.APrivateAuthKey a pk) = (,serviceSig) <$> case a of
-      C.SX25519 -> case thAuth of
-        Just THAuthClient {peerServerPubKey = k} -> Right $ TAAuthenticator $ C.cbAuthenticate k pk nonce t'
-        Nothing -> Left TENoServerAuth
+      C.SX25519 -> case (thAuth, nonce_) of
+        (Just THAuthClient {peerServerPubKey = k}, Just nonce) -> Right $ TAAuthenticator $ C.cbAuthenticate k pk nonce t'
+        _ -> Left TENoServerAuth
       C.SEd25519 -> sign pk
       C.SEd448 -> sign pk
     -- When command is signed by both entity key and service key,

src/Simplex/Messaging/Crypto.hs

@@ -140,8 +140,10 @@ module Simplex.Messaging.Crypto
     gcmIV,
 
     -- * NaCl crypto_box
-    CbNonce (unCbNonce),
+    CbNonceI (..),
+    CbNonce,
     pattern CbNonce,
+    CorrCbNonce,
     cbEncrypt,
     cbEncryptNoPad,
     cbEncryptMaxLenBS,
@@ -152,8 +154,12 @@ module Simplex.Messaging.Crypto
     sbEncryptNoPad,
     sbDecryptNoPad,
     cbNonce,
+    unsafeCbNonce,
+    corrCbNonce,
     randomCbNonce,
     reverseNonce,
+    xorNonce,
+    toCbNonce,
 
     -- * NaCl crypto_secretbox
     SbKey (unSbKey),
@@ -232,6 +238,7 @@ import Data.ASN1.Types
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (bimap, first)
+import Data.Bits (xor)
 import Data.ByteArray (ByteArrayAccess)
 import qualified Data.ByteArray as BA
 import Data.ByteString.Base64 (decode)
@@ -254,7 +261,7 @@ import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (.
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll, parseString)
-import Simplex.Messaging.Util ((<$?>))
+import Simplex.Messaging.Util (packZipWith, (<$?>))
 
 -- | Cryptographic algorithms.
 data Algorithm = Ed25519 | Ed448 | X25519 | X448
@@ -1324,8 +1331,8 @@ sbEncryptNoPad (SbKey key) (CbNonce nonce) = cryptoBox key nonce
 {-# INLINE sbEncryptNoPad #-}
 
 -- | NaCl @crypto_box@ encrypt with a shared DH secret and 192-bit nonce.
-cbEncryptMaxLenBS :: KnownNat i => DhSecret X25519 -> CbNonce -> MaxLenBS i -> ByteString
-cbEncryptMaxLenBS (DhSecretX25519 secret) (CbNonce nonce) = cryptoBox secret nonce . unMaxLenBS . padMaxLenBS
+cbEncryptMaxLenBS :: (KnownNat i, CbNonceI n) => DhSecret X25519 -> n -> MaxLenBS i -> ByteString
+cbEncryptMaxLenBS (DhSecretX25519 secret) nonce = cryptoBox secret (unCbNonce nonce) . unMaxLenBS . padMaxLenBS
 {-# INLINE cbEncryptMaxLenBS #-}
 
 cryptoBox :: ByteArrayAccess key => key -> ByteString -> ByteString -> ByteString
@@ -1335,8 +1342,8 @@ cryptoBox secret nonce s = BA.convert tag <> c
     tag = Poly1305.auth rs c
 
 -- | NaCl @crypto_box@ decrypt with a shared DH secret and 192-bit nonce.
-cbDecrypt :: DhSecret X25519 -> CbNonce -> ByteString -> Either CryptoError ByteString
-cbDecrypt (DhSecretX25519 secret) = sbDecrypt_ secret
+cbDecrypt :: CbNonceI n => DhSecret X25519 -> n -> ByteString -> Either CryptoError ByteString
+cbDecrypt (DhSecretX25519 secret) = sbDecrypt_ secret . toCbNonce
 {-# INLINE cbDecrypt #-}
 
 -- | NaCl @crypto_box@ decrypt with a shared DH secret and 192-bit nonce (without unpadding).
@@ -1383,9 +1390,16 @@ cbAuthenticate k pk nonce msg = CbAuthenticator $ cbEncryptNoPad (dh' k pk) nonc
 cbVerify :: PublicKeyX25519 -> PrivateKeyX25519 -> CbNonce -> CbAuthenticator -> ByteString -> Bool
 cbVerify k pk nonce (CbAuthenticator s) authorized = cbDecryptNoPad (dh' k pk) nonce s == Right (sha512Hash authorized)
 
-newtype CbNonce = CryptoBoxNonce {unCbNonce :: ByteString}
+class CbNonceI n where
+  unCbNonce :: n -> ByteString
+
+newtype CbNonce = CryptoBoxNonce ByteString
   deriving (Eq, Show)
-  deriving newtype (FromField)
+
+instance CbNonceI CbNonce where
+  unCbNonce (CryptoBoxNonce s) = s
+
+instance FromField CbNonce where fromField = blobFieldDecoder cbNonce
 
 instance ToField CbNonce where toField (CryptoBoxNonce s) = toField $ Binary s
 
@@ -1396,7 +1410,7 @@ pattern CbNonce s <- CryptoBoxNonce s
 
 instance StrEncoding CbNonce where
   strEncode (CbNonce s) = strEncode s
-  strP = cbNonce <$> strP
+  strP = cbNonce <$?> strP
 
 instance ToJSON CbNonce where
   toJSON = strToJSON
@@ -1405,13 +1419,16 @@ instance ToJSON CbNonce where
 instance FromJSON CbNonce where
   parseJSON = strParseJSON "CbNonce"
 
-cbNonce :: ByteString -> CbNonce
+cbNonce :: ByteString -> Either String CbNonce
 cbNonce s
-  | len == 24 = CryptoBoxNonce s
-  | len > 24 = CryptoBoxNonce . fst $ B.splitAt 24 s
-  | otherwise = CryptoBoxNonce $ s <> B.replicate (24 - len) (toEnum 0)
-  where
-    len = B.length s
+  | B.length s == 24 = Right $ CryptoBoxNonce s
+  | otherwise = Left "CbNonce: invalid length"
+
+unsafeCbNonce :: ByteString -> CbNonce
+unsafeCbNonce s = either error id $ cbNonce s
+
+corrCbNonce :: ByteString -> Either String CorrCbNonce
+corrCbNonce s = CorrCbNonce . unCbNonce <$> cbNonce s
 
 randomCbNonce :: TVar ChaChaDRG -> STM CbNonce
 randomCbNonce = fmap CryptoBoxNonce . randomBytes 24
@@ -1419,13 +1436,30 @@ randomCbNonce = fmap CryptoBoxNonce . randomBytes 24
 randomBytes :: Int -> TVar ChaChaDRG -> STM ByteString
 randomBytes n gVar = stateTVar gVar $ randomBytesGenerate n
 
-reverseNonce :: CbNonce -> CbNonce
-reverseNonce (CryptoBoxNonce s) = CryptoBoxNonce (B.reverse s)
+reverseNonce :: CbNonceI n => n -> CbNonce
+reverseNonce = CryptoBoxNonce . B.reverse . unCbNonce
+
+xorNonce :: CbNonceI n => ByteString -> n -> CbNonce
+xorNonce s nonce = CryptoBoxNonce $ packZipWith xor s n <> B.drop (B.length s) n
+  where
+    n = unCbNonce nonce
+
+toCbNonce :: CbNonceI n => n -> CbNonce
+toCbNonce = CryptoBoxNonce . unCbNonce
 
 instance Encoding CbNonce where
   smpEncode = unCbNonce
   smpP = CryptoBoxNonce <$> A.take 24
 
+newtype CorrCbNonce = CorrCbNonce ByteString
+
+instance CbNonceI CorrCbNonce where
+  unCbNonce (CorrCbNonce s) = s
+
+instance Encoding CorrCbNonce where
+  smpEncode = smpEncode . unCbNonce
+  smpP = corrCbNonce <$?> smpP
+
 newtype SbKey = SecretBoxKey {unSbKey :: ByteString}
   deriving (Eq, Show)
   deriving newtype (FromField)

src/Simplex/Messaging/Protocol.hs

@@ -145,7 +145,10 @@ module Simplex.Messaging.Protocol
     RcvNtfDhSecret,
     Message (..),
     RcvMessage (..),
-    MsgId,
+    MsgId (unMsgId),
+    mkMsgId,
+    unsafeMsgId,
+    randomMsgId,
     MsgBody,
     IdsHash (..),
     ServiceSub (..),
@@ -233,8 +236,10 @@ module Simplex.Messaging.Protocol
 where
 
 import Control.Applicative (optional, (<|>))
+import Control.Concurrent.STM (STM, TVar)
 import Control.Exception (Exception, SomeException, displayException, fromException)
 import Control.Monad.Except
+import Crypto.Random (ChaChaDRG)
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Aeson as J
 import qualified Data.Aeson.TH as J
@@ -269,7 +274,7 @@ import qualified GHC.TypeLits as TE
 import qualified GHC.TypeLits as Type
 import Network.Socket (ServiceName)
 import qualified Network.TLS as TLS
-import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..))
+import Simplex.Messaging.Agent.Store.DB (Binary (..), FromField (..), ToField (..), blobFieldDecoder)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
@@ -281,7 +286,7 @@ import Simplex.Messaging.ServiceScheme
 import Simplex.Messaging.SimplexName (LabelHash, SimplexDomain (..), SimplexTLD (..), fullDomainName, labelHash)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (TransportHost, TransportHosts (..))
-import Simplex.Messaging.Util (bshow, eitherToMaybe, packZipWith, safeDecodeUtf8, (<$?>))
+import Simplex.Messaging.Util (bshow, eitherToMaybe, safeDecodeUtf8, (<$?>))
 import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
 
@@ -703,24 +708,24 @@ instance Encoding NewNtfCreds where
 newtype EncTransmission = EncTransmission ByteString
   deriving (Show)
 
-encTransmissionNonce :: VersionSMP -> C.CbNonce -> C.CbNonce
-encTransmissionNonce v nonce@(C.CbNonce s)
-  | v >= fwdNoncesSMPVersion = C.cbNonce $ packZipWith xor (smpEncode v) s <> BS.drop 2 s
-  | otherwise = nonce
+encTransmissionNonce :: C.CbNonceI n => VersionSMP -> n -> C.CbNonce
+encTransmissionNonce v nonce
+  | v >= fwdNoncesSMPVersion = C.xorNonce (smpEncode v) nonce
+  | otherwise = C.toCbNonce nonce
 
 data FwdTransmission = FwdTransmission
-  { fwdCorrId :: CorrId,
+  { fwdCorrId :: C.CorrCbNonce,
     fwdVersion :: VersionSMP,
     fwdKey :: C.PublicKeyX25519,
     fwdTransmission :: EncTransmission
   }
 
 instance Encoding FwdTransmission where
-  smpEncode FwdTransmission {fwdCorrId = CorrId corrId, fwdVersion, fwdKey, fwdTransmission = EncTransmission t} =
-    smpEncode (corrId, fwdVersion, fwdKey, Tail t)
+  smpEncode FwdTransmission {fwdCorrId, fwdVersion, fwdKey, fwdTransmission = EncTransmission t} =
+    smpEncode (fwdCorrId, fwdVersion, fwdKey, Tail t)
   smpP = do
-    (corrId, fwdVersion, fwdKey, Tail t) <- smpP
-    pure FwdTransmission {fwdCorrId = CorrId corrId, fwdVersion, fwdKey, fwdTransmission = EncTransmission t}
+    (fwdCorrId, fwdVersion, fwdKey, Tail t) <- smpP
+    pure FwdTransmission {fwdCorrId, fwdVersion, fwdKey, fwdTransmission = EncTransmission t}
 
 newtype EncFwdTransmission = EncFwdTransmission ByteString
   deriving (Show)
@@ -766,16 +771,16 @@ newtype EncFwdResponse = EncFwdResponse ByteString
   deriving (Eq, Show)
 
 data FwdResponse = FwdResponse
-  { fwdCorrId :: CorrId,
+  { fwdCorrId :: C.CorrCbNonce,
     fwdResponse :: EncResponse
   }
 
 instance Encoding FwdResponse where
-  smpEncode FwdResponse {fwdCorrId = CorrId corrId, fwdResponse = EncResponse t} =
-    smpEncode (corrId, Tail t)
+  smpEncode FwdResponse {fwdCorrId, fwdResponse = EncResponse t} =
+    smpEncode (fwdCorrId, Tail t)
   smpP = do
-    (corrId, Tail t) <- smpP
-    pure FwdResponse {fwdCorrId = CorrId corrId, fwdResponse = EncResponse t}
+    (fwdCorrId, Tail t) <- smpP
+    pure FwdResponse {fwdCorrId, fwdResponse = EncResponse t}
 
 newtype EncResponse = EncResponse ByteString
   deriving (Eq, Show)
@@ -798,7 +803,7 @@ toMsgInfo = \case
   Message {msgId, msgTs} -> msgInfo msgId msgTs MTMessage
   MessageQuota {msgId, msgTs} -> msgInfo msgId msgTs MTQuota
   where
-    msgInfo msgId msgTs msgType = MsgInfo {msgId = decodeLatin1 $ B64.encode msgId, msgTs = systemToUTCTime msgTs, msgType}
+    msgInfo msgId msgTs msgType = MsgInfo {msgId = decodeLatin1 $ B64.encode $ unMsgId msgId, msgTs = systemToUTCTime msgTs, msgType}
 
 messageId :: Message -> MsgId
 messageId = \case
@@ -1499,7 +1504,34 @@ type RcvNtfPublicDhKey = C.PublicKeyX25519
 type RcvNtfDhSecret = C.DhSecretX25519
 
 -- | SMP message server ID.
-type MsgId = ByteString
+newtype MsgId = MsgId {unMsgId :: ByteString}
+  deriving (Eq, Show)
+
+instance C.CbNonceI MsgId where
+  unCbNonce (MsgId s) = B.take 24 s
+
+instance Encoding MsgId where
+  smpEncode (MsgId s) = smpEncode s
+  smpP = mkMsgId <$?> smpP
+
+instance StrEncoding MsgId where
+  strEncode (MsgId s) = strEncode s
+  strP = mkMsgId <$?> strP
+
+instance FromField MsgId where fromField = blobFieldDecoder mkMsgId
+
+instance ToField MsgId where toField (MsgId s) = toField $ Binary s
+
+mkMsgId :: ByteString -> Either String MsgId
+mkMsgId s
+  | B.length s >= 24 = Right $ MsgId s
+  | otherwise = Left "MsgId: invalid length"
+
+unsafeMsgId :: ByteString -> MsgId
+unsafeMsgId s = either error id $ mkMsgId s
+
+randomMsgId :: Int -> TVar ChaChaDRG -> STM MsgId
+randomMsgId n = fmap MsgId . C.randomBytes (max 24 n)
 
 -- | SMP message body.
 type MsgBody = ByteString
@@ -2417,7 +2449,9 @@ tDecodeServer THandleParams {sessionId, thVersion = v, implySessId} = \case
           where
             cmdOrErr = parseProtocol @v @err @cmd v command >>= checkCredentials tAuth entityId
             t :: a -> (CorrId, EntityId, a)
-            t = (corrId,entityId,)
+            -- IDs are slices of the ~16 KB received block and are kept as subscription keys,
+            -- so without a copy one live key retains the whole block
+            t = (CorrId $ B.copy $ bs corrId,EntityId $ B.copy $ unEntityId entityId,)
         Left _ -> tError corrId PEBlock
     | otherwise -> tError corrId PESession
   Left _ -> tError "" PEBlock

src/Simplex/Messaging/Server.hs

@@ -1334,7 +1334,7 @@ verifyCmdAuthorization thAuth tAuth authorized corrId key = maybe False (verify
 
 verifyCmdAuth :: Maybe (THandleAuth 'TServer) -> C.PublicKeyX25519 -> C.CbAuthenticator -> ByteString -> CorrId -> Bool
 verifyCmdAuth thAuth k authenticator authorized (CorrId corrId) = case thAuth of
-  Just THAuthServer {serverPrivKey = pk} -> C.cbVerify k pk (C.cbNonce corrId) authenticator authorized
+  Just THAuthServer {serverPrivKey = pk} -> either (const False) (\nonce -> C.cbVerify k pk nonce authenticator authorized) $ C.cbNonce corrId
   Nothing -> False
 
 dummyVerifyCmd :: Maybe (THandleAuth 'TServer) -> Maybe TAuthorizations -> ByteString -> CorrId -> Bool
@@ -1453,14 +1453,15 @@ client
         ServerStats {pMsgFwds, pMsgFwdsOwn} <- asks serverStats
         let inc = mkIncProxyStats pMsgFwds pMsgFwdsOwn
         liftIO (lookupSMPServerClient a sessId) >>= \case
-          Just (own, smp) -> do
+          Just (own, smp) | Right fwdCorrId <- C.corrCbNonce (bs corrId) -> do
             inc own pRequests
             forkProxiedCmd $ do
-              liftIO (runExceptT (forwardSMPTransmission smp corrId fwdV pubKey encBlock) `E.catches` clientHandlers)  >>= \case
+              liftIO (runExceptT (forwardSMPTransmission smp fwdCorrId fwdV pubKey encBlock) `E.catches` clientHandlers)  >>= \case
                 Right (nonce_, r) -> PRES nonce_ r <$ inc own pSuccesses
                 Left e -> ERR (smpProxyError e) <$ case e of
                   PCEProtocolError {} -> inc own pSuccesses
                   _ -> inc own pErrorsOther
+          Just _ -> pure $ Just $ ERR $ CMD SYNTAX
           Nothing -> inc False pRequests >> inc False pErrorsConnect $> Just (ERR $ PROXY NO_SESSION)
       where
         forkProxiedCmd :: M s BrokerMsg -> M s (Maybe BrokerMsg)
@@ -1950,7 +1951,7 @@ client
             getDelivered :: Sub -> STM (Maybe (ServerSub, SystemSeconds))
             getDelivered Sub {delivered, subThread} = do
               readTVar delivered $>>= \(msgId', ts) ->
-                if msgId == msgId' || B.null msgId
+                if msgId == msgId'
                   then writeTVar delivered Nothing $> Just (subThread, ts)
                   else pure Nothing
             updateStats :: ServerStats -> Bool -> SystemSeconds -> Message -> IO ()
@@ -1996,7 +1997,7 @@ client
                       when (isJust (queueData qr) && isSecuredMsgQueue qr) $ void $ liftIO $
                         deleteQueueLinkData (queueStore ms) q
                       ServerConfig {messageExpiration, expireMessagesOnSend, msgIdBytes} <- asks config
-                      msgId <- randomId' msgIdBytes
+                      msgId <- atomically . randomMsgId msgIdBytes =<< asks random
                       msg_ <- liftIO $ runExceptT $ do
                         when expireMessagesOnSend $ mapM_ (expireMessages stats) messageExpiration
                         msg <- liftIO $ mkMessage msgId body
@@ -2109,7 +2110,7 @@ client
               liftIO $ storeNtf ns nId ntf
               incStat . ntfCount =<< asks serverStats
 
-            mkMessageNotification :: ByteString -> SystemTime -> RcvNtfDhSecret -> M s MsgNtf
+            mkMessageNotification :: MsgId -> SystemTime -> RcvNtfDhSecret -> M s MsgNtf
             mkMessageNotification msgId msgTs rcvNtfDhSecret = do
               ntfNonce <- atomically . C.randomCbNonce =<< asks random
               let msgMeta = NMsgMeta {msgId, msgTs}
@@ -2120,12 +2121,12 @@ client
         processForwardedCommand (EncFwdTransmission s) = fmap (either (Just . ERR) id) . runExceptT $ do
           THAuthServer {serverPrivKey, sessSecret'} <- maybe (throwE $ transportErr TENoServerAuth) pure (thAuth thParams')
           sessSecret <- maybe (throwE $ transportErr TENoServerAuth) pure sessSecret'
-          let proxyNonce = C.cbNonce $ bs corrId
+          proxyNonce <- liftEitherWith (const CRYPTO) $ C.cbNonce $ bs corrId
           s' <- liftEitherWith (const CRYPTO) $ C.cbDecryptNoPad sessSecret proxyNonce s
           FwdTransmission {fwdCorrId, fwdVersion, fwdKey, fwdTransmission = EncTransmission et} <- liftEitherWith (const $ CMD SYNTAX) $ smpDecode s'
           unless (fwdVersion `isCompatible` thServerVRange thParams') $ throwE $ transportErr TEVersion
           let clientSecret = C.dh' fwdKey serverPrivKey
-              clientNonce = C.cbNonce $ bs fwdCorrId
+              clientNonce = fwdCorrId
           b <- liftEitherWith (const CRYPTO) $ C.cbDecrypt clientSecret (encTransmissionNonce fwdVersion clientNonce) et
           let clntTHParams = smpTHParamsSetVersion fwdVersion thParams'
           -- only allowing single forwarded transactions
@@ -2187,7 +2188,7 @@ client
           MessageQuota {} -> RcvMsgQuota msgTs'
           where
             encrypt :: KnownNat i => C.MaxLenBS i -> RcvMessage
-            encrypt body = RcvMessage msgId' . EncRcvMsgBody $ C.cbEncryptMaxLenBS (rcvDhSecret qr) (C.cbNonce msgId') body
+            encrypt body = RcvMessage msgId' . EncRcvMsgBody $ C.cbEncryptMaxLenBS (rcvDhSecret qr) msgId' body
             msgId' = messageId msg
             msgTs' = messageTs msg
 
@@ -2236,7 +2237,7 @@ client
                     SubPending -> QSubPending
                     SubThread _ -> QSubThread
                 ProhibitSub -> pure QProhibitSub
-              qDelivered <- decodeLatin1 . encode . fst <$$> readTVarIO delivered
+              qDelivered <- decodeLatin1 . encode . unMsgId . fst <$$> readTVarIO delivered
               pure QSub {qSubThread, qDelivered}
 
         ok :: Transmission BrokerMsg

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

@@ -571,7 +571,10 @@ newProhibitedSub = do
   return Sub {subThread = ProhibitSub, delivered}
 
 newEnv :: ServerConfig s -> IO (Env s)
-newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines, namesConfig} = do
+newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines, namesConfig, msgIdBytes} = do
+  when (msgIdBytes < 24) $ do
+    logError "Error: msgIdBytes must be at least 24"
+    exitFailure
   serverActive <- newTVarIO True
   server <- newServer
   msgStore_ <- case serverStoreCfg of

src/Simplex/Messaging/Server/MsgStore/Postgres.hs

@@ -202,7 +202,7 @@ instance MsgStoreClass PostgresMsgStore where
           DB.query
             db
             "SELECT quota_written, was_empty FROM write_message(?,?,?,?,?,?,?)"
-            (recipientId' q, Binary (messageId msg), systemSeconds (messageTs msg), msgQuota, ntf, Binary body, quota)
+            (recipientId' q, messageId msg, systemSeconds (messageTs msg), msgQuota, ntf, Binary body, quota)
     where
       toResult = \case
         ((msgQuota, wasEmpty) : _) -> if msgQuota then Nothing else Just (msg, wasEmpty)
@@ -271,14 +271,14 @@ instance MsgStoreClass PostgresMsgStore where
     uninterruptibleMask_ $
       withDB' "tryDelMsg" (queueStore_ ms) $ \db ->
         maybeFirstRow toMessage $
-          DB.query db "SELECT r_msg_id, r_msg_ts, r_msg_quota, r_msg_ntf_flag, r_msg_body FROM try_del_msg(?, ?)" (recipientId' q, Binary msgId)
+          DB.query db "SELECT r_msg_id, r_msg_ts, r_msg_quota, r_msg_ntf_flag, r_msg_body FROM try_del_msg(?, ?)" (recipientId' q, msgId)
 
   tryDelPeekMsg :: PostgresMsgStore -> PostgresQueue -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
   tryDelPeekMsg ms q msgId =
     uninterruptibleMask_ $
       withDB' "tryDelPeekMsg" (queueStore_ ms) $ \db ->
         toResult . map toMessage
-          <$> DB.query db "SELECT r_msg_id, r_msg_ts, r_msg_quota, r_msg_ntf_flag, r_msg_body FROM try_del_peek_msg(?, ?)" (recipientId' q, Binary msgId)
+          <$> DB.query db "SELECT r_msg_id, r_msg_ts, r_msg_quota, r_msg_ntf_flag, r_msg_body FROM try_del_peek_msg(?, ?)" (recipientId' q, msgId)
     where
       toResult = \case
         [] -> (Nothing, Nothing)
@@ -297,13 +297,13 @@ uninterruptibleMask_ :: ExceptT ErrorType IO a -> ExceptT ErrorType IO a
 uninterruptibleMask_ = ExceptT . E.uninterruptibleMask_ . runExceptT
 {-# INLINE uninterruptibleMask_ #-}
 
-toMaybeMessage :: (Maybe (Binary MsgId), Maybe Int64, Maybe Bool, Maybe Bool, Maybe (Binary MsgBody)) -> Maybe Message
+toMaybeMessage :: (Maybe MsgId, Maybe Int64, Maybe Bool, Maybe Bool, Maybe (Binary MsgBody)) -> Maybe Message
 toMaybeMessage = \case
   (Just msgId, Just ts, Just msgQuota, Just ntf, Just body) -> Just $ toMessage (msgId, ts, msgQuota, ntf, body)
   _ -> Nothing
 
-toMessage :: (Binary MsgId, Int64, Bool, Bool, Binary MsgBody) -> Message
-toMessage (Binary msgId, ts, msgQuota, ntf, Binary body)
+toMessage :: (MsgId, Int64, Bool, Bool, Binary MsgBody) -> Message
+toMessage (msgId, ts, msgQuota, ntf, Binary body)
   | msgQuota = MessageQuota {msgId, msgTs}
   | otherwise = Message {msgId, msgTs, msgFlags = MsgFlags ntf, msgBody = C.unsafeMaxLenBS body} -- TODO [messages] unsafeMaxLenBS?
   where
@@ -429,7 +429,7 @@ messageRecToText rId msg =
     tabFields = BB.char7 ',' `intersperse` fields
     fields =
       [ renderField (toField rId),
-        renderField (toField $ Binary (messageId msg)),
+        renderField (toField $ messageId msg),
         renderField (toField $ systemSeconds (messageTs msg)),
         renderField (toField msgQuota),
         renderField (toField ntf),

src/Simplex/Messaging/Server/QueueStore/Postgres.hs

@@ -825,6 +825,10 @@ instance ToField EncDataBytes where toField (EncDataBytes s) = toField (Binary s
 
 deriving newtype instance FromField EncDataBytes
 
+instance ToField MsgId where toField = toField . Binary . unMsgId
+
+instance FromField MsgId where fromField = blobFieldDecoder mkMsgId
+
 deriving newtype instance ToField (RoundedSystemTime t)
 
 deriving newtype instance FromField (RoundedSystemTime t)

src/Simplex/Messaging/Transport/KeepAlive.hs

@@ -11,6 +11,7 @@ module Simplex.Messaging.Transport.KeepAlive
   ) where
 
 import qualified Data.Aeson.TH as J
+import Control.Monad (when)
 import Foreign.C (CInt (..))
 import Network.Socket
 import Simplex.Messaging.Parsers (defaultJSON)
@@ -18,16 +19,25 @@ import Simplex.Messaging.Parsers (defaultJSON)
 data KeepAliveOpts = KeepAliveOpts
   { keepIdle :: Int,
     keepIntvl :: Int,
-    keepCnt :: Int
+    keepCnt :: Int,
+    -- | A connection with unacknowledged data gets no keep-alive probes, so the OS fails it only when it
+    -- gives up retransmitting - ~15 minutes on Linux and Android. When this is set such a connection is
+    -- failed after the keep-alive detection time instead. Absent or False - the OS default.
+    unackedDataTimeout :: Maybe Bool
   }
   deriving (Eq, Show)
 
+-- seconds after which a connection with no response to the probes is reported as failed
+keepAliveDetectionTime :: KeepAliveOpts -> Int
+keepAliveDetectionTime KeepAliveOpts {keepIdle, keepIntvl, keepCnt} = keepIdle + keepIntvl * keepCnt
+
 defaultKeepAliveOpts :: KeepAliveOpts
 defaultKeepAliveOpts =
   KeepAliveOpts
     { keepIdle = 30,
       keepIntvl = 15,
-      keepCnt = 4
+      keepCnt = 4,
+      unackedDataTimeout = Nothing
     }
 
 _SOL_TCP :: CInt
@@ -53,8 +63,12 @@ _TCP_KEEPCNT = 16
 
 #if defined(darwin_HOST_OS)
 foreign import capi "netinet/tcp.h value TCP_KEEPALIVE" _TCP_KEEPIDLE :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_RXT_CONNDROPTIME" _TCP_RXT_CONNDROPTIME :: CInt
 #else
 foreign import capi "netinet/tcp.h value TCP_KEEPIDLE" _TCP_KEEPIDLE :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_USER_TIMEOUT" _TCP_USER_TIMEOUT :: CInt
 #endif
 
 foreign import capi "netinet/tcp.h value TCP_KEEPINTVL" _TCP_KEEPINTVL :: CInt
@@ -63,11 +77,22 @@ foreign import capi "netinet/tcp.h value TCP_KEEPCNT" _TCP_KEEPCNT :: CInt
 
 #endif
 
+-- | seconds
+setSocketUnackedDataTimeout :: Socket -> Int -> IO ()
+#if defined(mingw32_HOST_OS)
+setSocketUnackedDataTimeout _ _ = pure ()
+#elif defined(darwin_HOST_OS)
+setSocketUnackedDataTimeout sock = setSocketOption sock (SockOpt _SOL_TCP _TCP_RXT_CONNDROPTIME)
+#else
+setSocketUnackedDataTimeout sock t = setSocketOption sock (SockOpt _SOL_TCP _TCP_USER_TIMEOUT) (t * 1000) -- TCP_USER_TIMEOUT is in milliseconds
+#endif
+
 setSocketKeepAlive :: Socket -> KeepAliveOpts -> IO ()
-setSocketKeepAlive sock KeepAliveOpts {keepCnt, keepIdle, keepIntvl} = do
+setSocketKeepAlive sock ka@KeepAliveOpts {keepCnt, keepIdle, keepIntvl, unackedDataTimeout} = do
   setSocketOption sock KeepAlive 1
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPIDLE) keepIdle
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPINTVL) keepIntvl
   setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPCNT) keepCnt
+  when (unackedDataTimeout == Just True) $ setSocketUnackedDataTimeout sock $ keepAliveDetectionTime ka
 
 $(J.deriveJSON defaultJSON ''KeepAliveOpts)

tests/AgentTests/FunctionalAPITests.hs

@@ -550,6 +550,10 @@ functionalAPITests ps = do
       testServerMatrix2 ps testFastSwitchConnection
     it "should switch delivery to the new queue when the old server is down" $
       testFastSwitchDeadOldServer ps
+    it "should drain the old queue after QEND on the new queue" $
+      testFastSwitchDrainOldQueue ps
+    it "should drain the old queue after QEND on the new queue and restart" $
+      testFastSwitchDrainOldQueueRestart ps
     describe "should switch delivery to the new queue" $
       testServerMatrix2 ps testSwitchConnection
     describe "should switch to new queue asynchronously" $
@@ -3729,7 +3733,7 @@ testFastSwitchDeadOldServer :: HasCallStack => (ASrvTransport, AStoreType) -> IO
 testFastSwitchDeadOldServer ps@(t, ASType qsType _) = do
   let bServers = initAgentServers {smp = userServers [testSMPServer2]}
   withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \_ ->
-    withAgent 1 agentCfg initAgentServers testDB $ \a ->
+    withAgent 1 aCfg initAgentServers testDB $ \a ->
       withAgent 2 agentCfg bServers testDB2 $ \b -> do
         (aId, bId) <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
           (aId, bId) <- makeConnection a b
@@ -3745,7 +3749,10 @@ testFastSwitchDeadOldServer ps@(t, ASType qsType _) = do
           queuedReceived <- drainSwitchCompletedRcvMsg a bId "queued while down"
           liftIO $ queuedReceived `shouldBe` True
           drainSwitchCompleted b aId QDSnd
+          waitOneRcvQueue a bId
           exchangeGreetingsMsgId 7 a bId b aId
+  where
+    aCfg = agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3}
 
 -- drains switch and network events until the connection reports SPCompleted in the given direction,
 -- tolerating DOWN/UP and intermediate phases (the old server is stopped mid-rotation)
@@ -3765,6 +3772,69 @@ drainSwitchCompletedRcvMsg c connId body = go False
         (_, connId', AEvt SAEConn (Msg' mId _ body')) | connId' == connId && body' == body -> ackMessage c connId' mId Nothing >> go True
         _ -> go seen
 
+testFastSwitchDrainOldQueue :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testFastSwitchDrainOldQueue ps =
+  withSmpServers2 ps $ withAgentClients2 $ \a b -> runRight_ $ do
+    (aId, bId, aM1Id) <- endSwitchOnNewQueue a b
+    ackMessage a bId aM1Id Nothing
+    drainOldQueue a bId b aId
+
+testFastSwitchDrainOldQueueRestart :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testFastSwitchDrainOldQueueRestart ps =
+  withSmpServers2 ps $
+    withAgent 2 agentCfg initAgentServers testDB2 $ \b -> do
+      (aId, bId, _) <- withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight $ endSwitchOnNewQueue a b
+      withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight_ $ do
+        subscribeConnection a bId
+        drainOldQueue a bId b aId
+
+endSwitchOnNewQueue :: HasCallStack => AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId, AgentMsgId)
+endSwitchOnNewQueue a b = do
+  (aId, bId) <- makeConnection a b
+  exchangeGreetings a bId b aId
+  liftIO $ setProtocolServers a 1 [noAuthSrvCfg testSMPServer2]
+  m1Id <- sendMessage b aId SMP.noMsgFlags "m1"
+  get b ##> ("", aId, SENT m1Id)
+  ("", c, Msg' aM1Id _ "m1") <- get a
+  liftIO $ c `shouldBe` bId
+  m2Id <- sendMessage b aId SMP.noMsgFlags "m2"
+  get b ##> ("", aId, SENT m2Id)
+  _ <- switchConnectionAsync a "" bId
+  fastSwitchComplete a bId b aId
+  stats <- getConnectionServers a bId
+  liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Nothing, Just RSReceivedQEND]
+  pure (aId, bId, aM1Id)
+
+drainOldQueue :: HasCallStack => AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
+drainOldQueue a bId b aId = do
+  ("", c, MSG MsgMeta {recipient = (aM2Id, _)} _ "m2") <- get a
+  liftIO $ c `shouldBe` bId
+  ackMessage a bId aM2Id Nothing
+  waitOneRcvQueue a bId
+  bMsgId <- sendMessage b aId SMP.noMsgFlags "hello"
+  get b ##> ("", aId, SENT bMsgId)
+  ("", c', MSG MsgMeta {recipient = (aMsgId, _)} _ "hello") <- get a
+  liftIO $ c' `shouldBe` bId
+  ackMessage a bId aMsgId Nothing
+  aMsgId' <- sendMessage a bId SMP.noMsgFlags "hello too"
+  get a ##> ("", bId, SENT aMsgId')
+  ("", c'', MSG MsgMeta {recipient = (bMsgId', _)} _ "hello too") <- get b
+  liftIO $ c'' `shouldBe` aId
+  ackMessage b aId bMsgId' Nothing
+  liftIO $ noMessages a "nothing else should be delivered to alice"
+  liftIO $ noMessages b "nothing else should be delivered to bob"
+
+waitOneRcvQueue :: HasCallStack => AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
+waitOneRcvQueue c connId = go (300 :: Int)
+  where
+    go n = do
+      stats <- getConnectionServers c connId
+      case rcvSwchStatuses' stats of
+        [Nothing] -> pure ()
+        statuses
+          | n > 0 -> liftIO (threadDelay 100000) >> go (n - 1)
+          | otherwise -> liftIO $ expectationFailure $ "old receive queue was not deleted: " <> show statuses
+
 phaseRcv :: AgentClient -> ByteString -> SwitchPhase -> [Maybe RcvSwitchStatus] -> ExceptT AgentErrorType IO ()
 phaseRcv c connId p swchStatuses = phase c connId QDRcv p (\stats -> rcvSwchStatuses' stats `shouldMatchList` swchStatuses)
 
@@ -4861,7 +4931,7 @@ testServerQueueInfo = do
     Just srvMsgId <- checkMsgQ bob aliceId 1
     get bob =##> \case
       ("", c, MSG MsgMeta {integrity = MsgOk, broker = (smId, _), recipient = (mId, _), pqEncryption = PQEncOn} _ "hello") ->
-        c == aliceId && decodeLatin1 (B64.encode smId) == srvMsgId && mId == msgId
+        c == aliceId && decodeLatin1 (B64.encode $ SMP.unMsgId smId) == srvMsgId && mId == msgId
       _ -> False
     ackMessage bob aliceId msgId Nothing
     liftIO $ threadDelay 200000

tests/AgentTests/NotificationTests.hs

@@ -230,7 +230,7 @@ testNotificationToken apns = do
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
       getMockNotification apns tkn
     verification <- ntfData .-> "verification"
-    nonce <- C.cbNonce <$> ntfData .-> "nonce"
+    nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
     verifyNtfToken a tkn nonce verification
     NTActive <- checkNtfToken a tkn
     deleteNtfToken a tkn
@@ -254,7 +254,7 @@ testNtfTokenRepeatRegistration apns = do
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
       getMockNotification apns tkn
     verification <- ntfData .-> "verification"
-    nonce <- C.cbNonce <$> ntfData .-> "nonce"
+    nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}} <-
       getMockNotification apns tkn
@@ -273,14 +273,14 @@ testNtfTokenSecondRegistration apns =
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
       getMockNotification apns tkn
     verification <- ntfData .-> "verification"
-    nonce <- C.cbNonce <$> ntfData .-> "nonce"
+    nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
     verifyNtfToken a tkn nonce verification
 
     NTRegistered <- registerNtfToken a' tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}} <-
       getMockNotification apns tkn
     verification' <- ntfData' .-> "verification"
-    nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
+    nonce' <- C.unsafeCbNonce <$> ntfData' .-> "nonce"
 
     -- at this point the first token is still active
     NTActive <- checkNtfToken a tkn
@@ -311,7 +311,7 @@ testNtfTokenServerRestart t apns = do
     -- so that repeat verification happens without restarting the clients, when notification arrives
     withNtfServer t $ runRight_ $ do
       verification <- ntfData .-> "verification"
-      nonce <- C.cbNonce <$> ntfData .-> "nonce"
+      nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
       verifyNtfToken a' tkn nonce verification
       NTActive <- checkNtfToken a' tkn
       pure ()
@@ -327,7 +327,7 @@ testNtfTokenServerRestartReverify t apns = do
       pure ntfData
     runRight_ $ do
       verification <- ntfData .-> "verification"
-      nonce <- C.cbNonce <$> ntfData .-> "nonce"
+      nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
       Left (BROKER _ (NETWORK _)) <- tryE $ verifyNtfToken a tkn nonce verification
       pure ()
   threadDelay 1500000
@@ -348,7 +348,7 @@ testNtfTokenServerRestartReverifyTimeout t apns = do
       APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
         getMockNotification apns tkn
       verification <- ntfData .-> "verification"
-      nonce <- C.cbNonce <$> ntfData .-> "nonce"
+      nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
       verifyNtfToken a tkn nonce verification
       pure (nonce, verification)
     -- this emulates the situation when server verified token but the client did not receive the response
@@ -393,7 +393,7 @@ testNtfTokenServerRestartReregister t apns = do
       APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
         getMockNotification apns tkn
       verification <- ntfData .-> "verification"
-      nonce <- C.cbNonce <$> ntfData .-> "nonce"
+      nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
       verifyNtfToken a' tkn nonce verification
       NTActive <- checkNtfToken a' tkn
       pure ()
@@ -428,7 +428,7 @@ testNtfTokenServerRestartReregisterTimeout t apns = do
       APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
         getMockNotification apns tkn
       verification <- ntfData .-> "verification"
-      nonce <- C.cbNonce <$> ntfData .-> "nonce"
+      nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
       verifyNtfToken a' tkn nonce verification
       NTActive <- checkNtfToken a' tkn
       pure ()
@@ -450,7 +450,7 @@ testNtfTokenMultipleServers t apns = do
         APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
           getMockNotification apns tkn
         verification <- ntfData .-> "verification"
-        nonce <- C.cbNonce <$> ntfData .-> "nonce"
+        nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
         verifyNtfToken a tkn nonce verification
         NTActive <- checkNtfToken a tkn
         -- shut down the "other" server
@@ -566,7 +566,7 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}} <-
       getMockNotification apns tkn
     verification <- ntfData .-> "verification"
-    vNonce <- C.cbNonce <$> ntfData .-> "nonce"
+    vNonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
     verifyNtfToken alice tkn vNonce verification
     NTActive <- checkNtfToken alice tkn
     -- send message
@@ -657,7 +657,7 @@ registerTestToken a token mode apns = do
   Just APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}} <-
     timeout 1000000 $ getMockNotification apns tkn
   verification' <- ntfData' .-> "verification"
-  nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
+  nonce' <- C.unsafeCbNonce <$> ntfData' .-> "nonce"
   verifyNtfToken a tkn nonce' verification'
   NTActive <- checkNtfToken a tkn
   pure tkn
@@ -1035,7 +1035,7 @@ messageNotification apns tkn = do
   500000 `timeout` getMockNotification apns tkn >>= \case
     Nothing -> error "no notification"
     Just APNSMockRequest {notification = APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData}} -> do
-      nonce <- C.cbNonce <$> ntfData .-> "nonce"
+      nonce <- C.unsafeCbNonce <$> ntfData .-> "nonce"
       message <- ntfData .-> "message"
       pure (nonce, message)
     _ -> error "bad notification"

tests/AgentTests/SQLiteTests.hs

@@ -513,7 +513,7 @@ hw = encodeUtf8 "Hello world!"
 ts :: UTCTime
 ts = UTCTime (fromGregorian 2021 02 24) (secondsToDiffTime 0)
 
-mkRcvMsgData :: InternalId -> InternalRcvId -> ExternalSndId -> BrokerId -> MsgHash -> RcvMsgData
+mkRcvMsgData :: InternalId -> InternalRcvId -> ExternalSndId -> ByteString -> MsgHash -> RcvMsgData
 mkRcvMsgData internalId internalRcvId externalSndId brokerId internalHash =
   RcvMsgData
     { internalRcvId,
@@ -522,7 +522,7 @@ mkRcvMsgData internalId internalRcvId externalSndId brokerId internalHash =
           { integrity = MsgOk,
             recipient = (unId internalId, ts),
             sndMsgId = externalSndId,
-            broker = (brokerId, ts),
+            broker = (SMP.unsafeMsgId $ mconcat (replicate 24 brokerId), ts),
             pqEncryption = CR.PQEncOn
           },
       msgType = AM_A_MSG_,
@@ -688,7 +688,7 @@ hasMigrations :: DBStore -> Expectation
 hasMigrations st = getMigrations st `shouldReturn` True
 
 errorGettingMigrations :: DBStore -> Expectation
-errorGettingMigrations st = getMigrations st `shouldThrow` \(e :: SomeException) -> "ErrorMisuse" `isInfixOf` show e
+errorGettingMigrations st = getMigrations st `shouldThrow` \(e :: SomeException) -> "withTransaction: database closed" `isInfixOf` show e
 
 testGetPendingQueueMsg :: DBStore -> Expectation
 testGetPendingQueueMsg st = do

tests/AgentTests/SchemaDump.hs

@@ -136,7 +136,9 @@ skipComparisonForDownMigrations =
     -- snd_secure and last_broker_ts columns swap order on down migration
     "m20250322_short_links",
     -- on down migration idx_processed_ratchet_key_hashes_hash index moves down to the end of the file
-    "m20260929_ratchet_indexes"
+    "m20260929_ratchet_indexes",
+    -- sqlite_sequence moves after commands table after VACUUM
+    "m20261005_commands_autoincrement"
   ]
 
 getSchema :: FilePath -> FilePath -> IO String

tests/CoreTests/BatchingTests.hs

@@ -14,11 +14,13 @@ import Control.Monad
 import Crypto.Random (ChaChaDRG)
 import qualified Data.ByteString as B
 import Data.ByteString.Char8 (ByteString)
+import Data.ByteString.Unsafe (unsafeUseAsCString)
 import qualified Data.List.NonEmpty as L
 import Data.Time.Clock.System (SystemTime, getSystemTime)
 import qualified Data.X509 as X
 import qualified Data.X509.CertificateStore as XS
 import qualified Data.X509.File as XF
+import Foreign.Ptr (plusPtr)
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
@@ -40,6 +42,8 @@ batchingTests = do
     it "should batch subscription responses with message" testBatchSubResponses
     it "should break on message that does not fit" testClientBatchWithMessage
     it "should break on large message" testClientBatchWithLargeMessage
+  describe "tDecodeServer" $
+    it "should copy IDs out of received block" testDecodeServerCopiesIds
 
 testBatchSubscriptions :: IO ()
 testBatchSubscriptions = do
@@ -166,6 +170,26 @@ testClientBatchWithLargeMessage = do
   (length rs1', length rs2') `shouldBe` (75, 135)
   all lenOk [s1', s2'] `shouldBe` True
 
+testDecodeServerCopiesIds :: IO ()
+testDecodeServerCopiesIds = do
+  sessId <- atomically . C.randomBytes 32 =<< C.newRandom
+  subs <- replicateM 2 $ randomSUB sessId
+  let thParams = testTHandleParams sessId
+  [TBTransmissions s 2 _] <- pure $ batchTransmissions thParams $ L.fromList subs
+  forM_ (tParse thParams s) $ \t -> do
+    let (corrId, entId, _) = tDecodeClient @SMPVersion @ErrorType @Cmd thParams t
+    Right (_, _, (corrId', entId', _)) <- pure $ tDecodeServer @SMPVersion @ErrorType @Cmd (testTHandleParams sessId) t
+    (corrId', entId') `shouldBe` (corrId, entId)
+    sharesBuffer s (bs corrId) `shouldReturn` True
+    sharesBuffer s (unEntityId entId) `shouldReturn` True
+    sharesBuffer s (bs corrId') `shouldReturn` False
+    sharesBuffer s (unEntityId entId') `shouldReturn` False
+
+sharesBuffer :: ByteString -> ByteString -> IO Bool
+sharesBuffer block s =
+  unsafeUseAsCString block $ \blockPtr -> unsafeUseAsCString s $ \ptr ->
+    pure $ ptr >= blockPtr && ptr < blockPtr `plusPtr` B.length block
+
 testClientStub :: IO (ProtocolClient SMPVersion ErrorType BrokerMsg)
 testClientStub = do
   g <- C.newRandom
@@ -183,7 +207,7 @@ randomSUB sessId = do
   thAuth_ <- testTHandleAuth g rKey
   let thParams = testTHandleParams sessId
       TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (CorrId corrId, EntityId rId, Cmd SRecipient SUB)
-  pure $ (,tToSend) <$> authTransmission thAuth_ True (Just rpKey) nonce tForAuth
+  pure $ (,tToSend) <$> authTransmission thAuth_ True (Just rpKey) (Just nonce) tForAuth
 
 randomSUBCmd :: ProtocolClient SMPVersion ErrorType BrokerMsg -> IO (PCTransmission ErrorType BrokerMsg)
 randomSUBCmd c = do
@@ -202,7 +226,7 @@ randomNMSGCmd :: SystemTime -> IO (Transmission BrokerMsg)
 randomNMSGCmd ts = do
   g <- C.newRandom
   nId <- atomically $ C.randomBytes 24 g
-  msgId <- atomically $ C.randomBytes 24 g
+  msgId <- atomically $ randomMsgId 24 g
   (k, pk) <- atomically $ C.generateKeyPair g
   nonce <- atomically $ C.randomCbNonce g
   let msgMeta = NMsgMeta {msgId, msgTs = ts}
@@ -221,7 +245,7 @@ randomMSG = do
   g <- C.newRandom
   corrId <- atomically $ C.randomBytes 24 g
   rId <- atomically $ C.randomBytes 24 g
-  msgId <- atomically $ C.randomBytes 24 g
+  msgId <- atomically $ randomMsgId 24 g
   msg <- atomically $ C.randomBytes maxMessageLength g
   pure (CorrId corrId, EntityId rId, MSG RcvMessage {msgId, msgBody = EncRcvMsgBody msg})
 
@@ -235,9 +259,9 @@ randomSEND sessId len = do
   msg <- atomically $ C.randomBytes len g
   let thParams = testTHandleParams sessId
       TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (CorrId corrId, EntityId sId, Cmd SSender $ SEND noMsgFlags msg)
-  pure $ (,tToSend) <$> authTransmission thAuth_ False (Just spKey) nonce tForAuth
+  pure $ (,tToSend) <$> authTransmission thAuth_ False (Just spKey) (Just nonce) tForAuth
 
-testTHandleParams :: ByteString -> THandleParams SMPVersion 'TClient
+testTHandleParams :: ByteString -> THandleParams SMPVersion p
 testTHandleParams sessionId =
   THandleParams
     { sessionId,

tests/CoreTests/MsgStoreTests.hs

@@ -34,7 +34,7 @@ import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import SMPClient (testStoreLogFile, testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
 import Simplex.Messaging.Crypto (pattern MaxLenBS)
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Protocol (EncDataBytes (..), EntityId (..), ErrorType (..), LinkId, Message (..), QueueLinkData, RecipientId, SParty (..), noMsgFlags)
+import Simplex.Messaging.Protocol (EncDataBytes (..), EntityId (..), ErrorType (..), LinkId, Message (..), QueueLinkData, RecipientId, SParty (..), noMsgFlags, randomMsgId)
 import Simplex.Messaging.Server (exportMessages, importMessages, printMessageStats)
 import Simplex.Messaging.Server.Env.STM (MsgStore (..), journalMsgStoreDepth, readWriteQueueStore)
 import Simplex.Messaging.Server.Expiration (ExpirationConfig (..), expireBeforeEpoch)
@@ -147,7 +147,7 @@ mkMessage :: MonadIO m => ByteString -> m Message
 mkMessage body = liftIO $ do
   g <- C.newRandom
   msgTs <- getSystemTime
-  msgId <- atomically $ C.randomBytes 24 g
+  msgId <- atomically $ randomMsgId 24 g
   pure Message {msgId, msgTs, msgFlags = noMsgFlags, msgBody = C.unsafeMaxLenBS body}
 
 pattern Msg :: ByteString -> Maybe Message

tests/NtfServerTests.hs

@@ -136,7 +136,7 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           let dhSecret = C.dh' ntfDh dhPriv
               decryptCode nd =
                 let Right verification = nd .-> "verification"
-                    Right nonce = C.cbNonce <$> nd .-> "nonce"
+                    Right nonce = C.cbNonce =<< nd .-> "nonce"
                     Right pt = C.cbDecrypt dhSecret nonce verification
                  in NtfRegCode pt
           let code = decryptCode ntfData
@@ -160,7 +160,7 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           -- receive notification
           APNSMockRequest {notification} <- getMockNotification apns tkn
           let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData'} = notification
-              Right nonce' = C.cbNonce <$> ntfData' .-> "nonce"
+              Right nonce' = C.cbNonce =<< ntfData' .-> "nonce"
               Right message = ntfData' .-> "message"
               Right ntfDataDecrypted = C.cbDecrypt dhSecret nonce' message
               Right pnMsgs1 = parse pnMessagesP (AP.INTERNAL "error parsing PNMessageData") ntfDataDecrypted
@@ -171,7 +171,7 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           notifierId `shouldBe` nId
           -- receive message
           Resp "" _ (MSG RcvMessage {msgId = mId1, msgBody = EncRcvMsgBody body}) <- tGet1 rh
-          Right ClientRcvMsgBody {msgTs = mTs, msgBody} <- pure $ parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt rcvDhSecret (C.cbNonce mId1) body)
+          Right ClientRcvMsgBody {msgTs = mTs, msgBody} <- pure $ parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt rcvDhSecret mId1 body)
           mId1 `shouldBe` msgId
           mTs `shouldBe` msgTs
           (msgBody, "hello") #== "delivered from queue"
@@ -183,7 +183,7 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData2}} <-
             getMockNotification apns tkn'
           let Right verification2 = ntfData2 .-> "verification"
-              Right nonce2 = C.cbNonce <$> ntfData2 .-> "nonce"
+              Right nonce2 = C.cbNonce =<< ntfData2 .-> "nonce"
               Right code2 = NtfRegCode <$> C.cbDecrypt dhSecret nonce2 verification2
           RespNtf "8" _ NROk <- signSendRecvNtf nh tknKey ("8", tId, TVFY code2)
           RespNtf "8a" _ (NRTkn NTActive) <- signSendRecvNtf nh tknKey ("8a", tId, TCHK)
@@ -191,7 +191,7 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           Resp "9" _ OK <- signSendRecv sh sKey ("9", sId, _SEND' "hello 2")
           APNSMockRequest {notification = notification3} <- getMockNotification apns tkn'
           let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData3} = notification3
-              Right nonce3 = C.cbNonce <$> ntfData3 .-> "nonce"
+              Right nonce3 = C.cbNonce =<< ntfData3 .-> "nonce"
               Right message3 = ntfData3 .-> "message"
               Right ntfDataDecrypted3 = C.cbDecrypt dhSecret nonce3 message3
               Right pnMsgs2 = parse pnMessagesP (AP.INTERNAL "error parsing PNMessageData") ntfDataDecrypted3
@@ -199,7 +199,7 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           smpServer3 `shouldBe` srv
           notifierId3 `shouldBe` nId
           Resp "" _ (MSG RcvMessage {msgId = mId2, msgBody = EncRcvMsgBody body2}) <- tGet1 rh
-          Right ClientRcvMsgBody {msgBody = "hello 2"} <- pure $ parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt rcvDhSecret (C.cbNonce mId2) body2)
+          Right ClientRcvMsgBody {msgBody = "hello 2"} <- pure $ parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt rcvDhSecret mId2 body2)
           Resp "10" _ OK <- signSendRecv rh rKey ("10", rId, ACK mId2)
 
           q2 <- createQueue rh sPub nPub
@@ -218,7 +218,7 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           Resp "14" _ OK <- signSendRecv sh sKey ("14", sId, _SEND' "hello 3")
           APNSMockRequest {notification = notification4} <- getMockNotification apns tkn'
           let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData4} = notification4
-              Right nonce4 = C.cbNonce <$> ntfData4 .-> "nonce"
+              Right nonce4 = C.cbNonce =<< ntfData4 .-> "nonce"
               Right message4 = ntfData4 .-> "message"
               Right ntfDataDecrypted4 = C.cbDecrypt dhSecret nonce4 message4
               Right pnMsgs4 = parse pnMessagesP (AP.INTERNAL "error parsing PNMessageData") ntfDataDecrypted4
@@ -226,14 +226,14 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           smpServer4 `shouldBe` srv
           notifierId4 `shouldBe` nId
           Resp "" _ (MSG RcvMessage {msgId = mId3, msgBody = EncRcvMsgBody body3}) <- tGet1 rh
-          Right ClientRcvMsgBody {msgBody = "hello 3"} <- pure $ parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt rcvDhSecret (C.cbNonce mId3) body3)
+          Right ClientRcvMsgBody {msgBody = "hello 3"} <- pure $ parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt rcvDhSecret mId3 body3)
           Resp "15" _ OK <- signSendRecv rh rKey ("15", rId, ACK mId3)
 
           -- deliver to queue with ntf sub created while SMP was offline
           Resp "16" _ OK <- signSendRecv sh sKey ("16", sId', _SEND' "hello 4")
           APNSMockRequest {notification = notification5} <- getMockNotification apns tkn'
           let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData5} = notification5
-              Right nonce5 = C.cbNonce <$> ntfData5 .-> "nonce"
+              Right nonce5 = C.cbNonce =<< ntfData5 .-> "nonce"
               Right message5 = ntfData5 .-> "message"
               Right ntfDataDecrypted5 = C.cbDecrypt dhSecret nonce5 message5
               Right pnMsgs5 = parse pnMessagesP (AP.INTERNAL "error parsing PNMessageData") ntfDataDecrypted5
@@ -241,7 +241,7 @@ testNotificationSubscription (ATransport t, msType) createQueue =
           smpServer5 `shouldBe` srv
           notifierId5 `shouldBe` nId'
           Resp "" _ (MSG RcvMessage {msgId = mId4, msgBody = EncRcvMsgBody body4}) <- tGet1 rh
-          Right ClientRcvMsgBody {msgBody = "hello 4"} <- pure $ parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt rcvDhSecret' (C.cbNonce mId4) body4)
+          Right ClientRcvMsgBody {msgBody = "hello 4"} <- pure $ parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt rcvDhSecret' mId4 body4)
           Resp "17" _ OK <- signSendRecv rh rKey' ("17", rId', ACK mId4)
           pure ()
 
@@ -307,7 +307,7 @@ registerToken nh apns token = do
   let dhSecret = C.dh' ntfDh dhPriv
       decryptCode nd =
         let Right verification = nd .-> "verification"
-            Right nonce = C.cbNonce <$> nd .-> "nonce"
+            Right nonce = C.cbNonce =<< nd .-> "nonce"
             Right pt = C.cbDecrypt dhSecret nonce verification
          in NtfRegCode pt
   let code = decryptCode ntfData

tests/SMPProxyTests.hs

@@ -75,6 +75,8 @@ smpProxyTests = do
     xit "batching proxy requests" todo
     it "relay rejects forwarded command with changed version" $ \_ ->
       testChangedFwdVersion
+    it "proxy rejects forwarded correlation ID that is not 24 bytes" $ \_ ->
+      testFwdCorrIdSize
   describe "deliver message via SMP proxy" $ do
     let srv1 = SMPServer testHost testPort testKeyHash
         srv2 = SMPServer testHost2 testPort2 testKeyHash
@@ -472,10 +474,22 @@ testChangedFwdVersion =
     nonce@(C.CbNonce corrId) <- atomically $ C.randomCbNonce g
     let v = currentClientSMPRelayVersion
     et <- either (fail . show) (pure . SMP.EncTransmission) $ C.cbEncrypt (C.dh' peerServerPubKey cmdPrivKey) (SMP.encTransmissionNonce v nonce) "" SMP.paddedProxiedTLength
-    let forward fwdVersion = forwardSMPTransmission rc (SMP.CorrId corrId) fwdVersion cmdPubKey et
+    let forward fwdVersion = forwardSMPTransmission rc (either error id $ C.corrCbNonce corrId) fwdVersion cmdPubKey et
     _ <- runExceptT' $ forward v
     runExceptT (forward $ prevVersion v) `shouldReturn` Left (PCEProtocolError SMP.CRYPTO)
 
+testFwdCorrIdSize :: IO ()
+testFwdCorrIdSize =
+  withSmpServerConfigOn (transport @TLS) proxyCfg testPort $ \_ ->
+    withSmpServerConfigOn (transport @TLS) cfgJ2 testPort2 $ \_ ->
+      testSMPClient_ "localhost" testPort supportedServerSMPRelayVRange Nothing $ \(th :: THandleSMP TLS 'TClient) -> do
+        (_, _, Right (SMP.PKEY sessId _ _)) <- sendRecv th (Nothing, "1", NoEntity, SMP.PRXY testSMPServer2 Nothing)
+        g <- C.newRandom
+        (cmdPubKey, _) <- atomically $ C.generateKeyPair g
+        forM_ (["", "2", "3333333333333333333333333"] :: [ByteString]) $ \corrId -> do
+          (_, _, reply) <- sendRecv th (Nothing, corrId, SMP.EntityId sessId, SMP.PFWD currentClientSMPRelayVersion cmdPubKey (SMP.EncTransmission ""))
+          reply `shouldBe` Right (SMP.ERR $ SMP.CMD SMP.SYNTAX)
+
 -- Shared "phase 2" of the reconnection tests: start a healthy relay, confirm it is reachable
 -- directly (PING, not via the proxy) so a proxy failure can only mean the proxy didn't reconnect,
 -- let any stored connection error expire, then require the proxy to establish the session (PKEY).

tests/ServerTests.hs

@@ -166,7 +166,7 @@ signSend_ h@THandle {params} (C.APrivateAuthKey a pk) serviceKey_ (corrId, qId,
     authorize t = (,(`C.sign'` t) <$> serviceKey_) <$> case a of
       C.SEd25519 -> Just . TASignature . C.ASignature C.SEd25519 $ C.sign' pk t'
       C.SEd448 -> Just . TASignature . C.ASignature C.SEd448 $ C.sign' pk t'
-      C.SX25519 -> (\THAuthClient {peerServerPubKey = k} -> TAAuthenticator $ C.cbAuthenticate k pk (C.cbNonce corrId) t') <$> thAuth params
+      C.SX25519 -> (\THAuthClient {peerServerPubKey = k} -> TAAuthenticator $ C.cbAuthenticate k pk (C.unsafeCbNonce corrId) t') <$> thAuth params
 #if !MIN_VERSION_base(4,18,0)
       _sx448 -> undefined -- ghc8107 fails to the branch excluded by types
 #endif
@@ -194,12 +194,12 @@ _SEND = SEND noMsgFlags
 _SEND' :: MsgBody -> Command 'Sender
 _SEND' = SEND MsgFlags {notification = True}
 
-decryptMsgV2 :: C.DhSecret 'C.X25519 -> ByteString -> ByteString -> Either C.CryptoError ByteString
-decryptMsgV2 dhShared = C.cbDecrypt dhShared . C.cbNonce
+decryptMsgV2 :: C.DhSecret 'C.X25519 -> MsgId -> ByteString -> Either C.CryptoError ByteString
+decryptMsgV2 = C.cbDecrypt
 
-decryptMsgV3 :: C.DhSecret 'C.X25519 -> ByteString -> ByteString -> Either String MsgBody
+decryptMsgV3 :: C.DhSecret 'C.X25519 -> MsgId -> ByteString -> Either String MsgBody
 decryptMsgV3 dhShared nonce body =
-  case parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt dhShared (C.cbNonce nonce) body) of
+  case parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt dhShared nonce body) of
     Right ClientRcvMsgBody {msgBody} -> Right msgBody
     Right ClientRcvMsgQuota {} -> Left "ClientRcvMsgQuota"
     Left e -> Left e
@@ -425,7 +425,7 @@ testCreateDelete =
       Resp "dabc" _ err8 <- sendRecv sh ("", "dabc", sId, _SEND "hello")
       (err8, ERR AUTH) #== "rejects unsigned SEND too when deleted"
 
-      Resp "abcd" _ err11 <- signSendRecv rh rKey ("abcd", rId, ACK "")
+      Resp "abcd" _ err11 <- signSendRecv rh rKey ("abcd", rId, ACK mId2)
       (err11, ERR AUTH) #== "rejects ACK when conn deleted - the second message is deleted"
 
       Resp "bcda" _ err9 <- signSendRecv rh rKey ("bcda", rId, OFF)
@@ -609,7 +609,7 @@ testGetSubCommands =
       Resp "4" _ (Msg mId2 msg2) <- signSendRecv rh1 rKey ("4", rId, ACK mId1)
       (dec mId2 msg2, Right "hello 2") #== "received from queue via SUB"
       -- bad msgId returns error
-      Resp "5" _ (ERR NO_MSG) <- signSendRecv rh2 rKey ("5", rId, ACK "1234")
+      Resp "5" _ (ERR NO_MSG) <- signSendRecv rh2 rKey ("5", rId, ACK mId2)
       -- already ACK'd by subscriber, but still returns OK when msgId matches
       Resp "5a" _ OK <- signSendRecv rh2 rKey ("5a", rId, ACK mId1)
       -- msg2 is not lost - even if subscriber does not ACK it, it is delivered to getter
@@ -1302,9 +1302,9 @@ testTiming =
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair goodKeyAlg g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
-      Resp "abcd" NoEntity (Ids rId sId srvDh) <- signSendRecv rh rKey ("abcd", NoEntity, New rPub dhPub)
+      Resp "abcdabcdabcdabcdabcdabcd" NoEntity (Ids rId sId srvDh) <- signSendRecv rh rKey ("abcdabcdabcdabcdabcdabcd", NoEntity, New rPub dhPub)
       let dec = decryptMsgV3 $ C.dh' srvDh dhPriv
-      Resp "cdab" _ resp <- signSendRecv rh rKey ("cdab", rId, SUB)
+      Resp "cdabcdabcdabcdabcdabcdab" _ resp <- signSendRecv rh rKey ("cdabcdabcdabcdabcdabcdab", rId, SUB)
       case resp of
         OK -> pure ()
         SOK Nothing -> pure ()
@@ -1314,9 +1314,9 @@ testTiming =
       runTimingTest rh badKey rId SUB
 
       (sPub, sKey) <- atomically $ C.generateAuthKeyPair goodKeyAlg g
-      Resp "dabc" _ OK <- signSendRecv rh rKey ("dabc", rId, KEY sPub)
+      Resp "dabcdabcdabcdabcdabcdabc" _ OK <- signSendRecv rh rKey ("dabcdabcdabcdabcdabcdabc", rId, KEY sPub)
 
-      Resp "bcda" _ OK <- signSendRecv sh sKey ("bcda", sId, _SEND "hello")
+      Resp "bcdabcdabcdabcdabcdabcda" _ OK <- signSendRecv sh sKey ("bcdabcdabcdabcdabcdabcda", sId, _SEND "hello")
       Resp "" _ (Msg mId msg) <- tGet1 rh
       (dec mId msg, Right "hello") #== "delivered from queue"
 
@@ -1327,15 +1327,15 @@ testTiming =
           threadDelay 100000
           _ <- timeRepeat n $ do
             -- "warm up" the server
-            Resp "dabc" _ (ERR AUTH) <- signSendRecv h badKey ("dabc", EntityId "1234", cmd)
+            Resp "dabcdabcdabcdabcdabcdabc" _ (ERR AUTH) <- signSendRecv h badKey ("dabcdabcdabcdabcdabcdabc", EntityId "1234", cmd)
             return ()
           threadDelay 100000
           timeWrongKey <- timeRepeat n $ do
-            Resp "cdab" _ (ERR AUTH) <- signSendRecv h badKey ("cdab", qId, cmd)
+            Resp "cdabcdabcdabcdabcdabcdab" _ (ERR AUTH) <- signSendRecv h badKey ("cdabcdabcdabcdabcdabcdab", qId, cmd)
             return ()
           threadDelay 100000
           timeNoQueue <- timeRepeat n $ do
-            Resp "dabc" _ (ERR AUTH) <- signSendRecv h badKey ("dabc", EntityId "1234", cmd)
+            Resp "dabcdabcdabcdabcdabcdabc" _ (ERR AUTH) <- signSendRecv h badKey ("dabcdabcdabcdabcdabcdabc", EntityId "1234", cmd)
             return ()
           let ok = similarTime timeNoQueue timeWrongKey msType
           unless ok . putStrLn . unwords $
@@ -1814,10 +1814,10 @@ serverSyntaxTests (ATransport t) = do
     it "no parameters" $ (sampleSig, "abcd", "12345678", SEND_) >#> ("", "abcd", "12345678", ERR $ CMD SYNTAX)
     it "no queue ID" $ (sampleSig, "bcda", "", (SEND_, ' ', noMsgFlags, ' ', "hello" :: ByteString)) >#> ("", "bcda", "", ERR $ CMD NO_ENTITY)
   describe "ACK" $ do
-    it "valid syntax" $ (sampleSig, "cdab", "12345678", (ACK_, ' ', "1234" :: ByteString)) >#> ("", "cdab", "12345678", ERR AUTH)
+    it "valid syntax" $ (sampleSig, "cdab", "12345678", (ACK_, ' ', "123456789012345678901234" :: ByteString)) >#> ("", "cdab", "12345678", ERR AUTH)
     it "no parameters" $ (sampleSig, "abcd", "12345678", ACK_) >#> ("", "abcd", "12345678", ERR $ CMD SYNTAX)
-    it "no queue ID" $ (sampleSig, "bcda", "", (ACK_, ' ', "1234" :: ByteString)) >#> ("", "bcda", "", ERR $ CMD NO_AUTH)
-    it "no signature" $ ("", "cdab", "12345678", (ACK_, ' ', "1234" :: ByteString)) >#> ("", "cdab", "12345678", ERR $ CMD NO_AUTH)
+    it "no queue ID" $ (sampleSig, "bcda", "", (ACK_, ' ', "123456789012345678901234" :: ByteString)) >#> ("", "bcda", "", ERR $ CMD NO_AUTH)
+    it "no signature" $ ("", "cdab", "12345678", (ACK_, ' ', "123456789012345678901234" :: ByteString)) >#> ("", "cdab", "12345678", ERR $ CMD NO_AUTH)
   describe "PING" $ do
     it "valid syntax" $ ("", "abcd", "", PING_) >#> ("", "abcd", "", PONG)
   describe "broker response not allowed" $ do

tests/XFTPWebTests.hs

@@ -768,7 +768,7 @@ tsSecretboxTests :: Spec
 tsSecretboxTests = describe "crypto/secretbox" $ do
   let key32 = B.pack [1 .. 32]
       nonce24 = B.pack [1 .. 24]
-      cbNonceVal = C.cbNonce nonce24
+      cbNonceVal = C.unsafeCbNonce nonce24
       sbKeyVal = C.unsafeSbKey key32
 
   describe "NaCl secretbox (tag prepended)" $ do
@@ -979,7 +979,7 @@ tsFileCryptoTests :: Spec
 tsFileCryptoTests = describe "crypto/file" $ do
   let key32 = B.pack [1 .. 32]
       nonce24 = B.pack [1 .. 24]
-      cbNonceVal = C.cbNonce nonce24
+      cbNonceVal = C.unsafeCbNonce nonce24
       sbKeyVal = C.unsafeSbKey key32
 
   describe "FileHeader encoding" $ do
@@ -2329,7 +2329,7 @@ tsClientTests = describe "protocol/client" $ do
             C.cbAuthenticate
               (C.PublicKeyX25519 pubA)
               (C.PrivateKeyX25519 privB)
-              (C.cbNonce nonce24)
+              (C.unsafeCbNonce nonce24)
               msg
       tsResult <-
         callNode $
@@ -2352,7 +2352,7 @@ tsClientTests = describe "protocol/client" $ do
             C.cbAuthenticate
               (C.PublicKeyX25519 pubA)
               (C.PrivateKeyX25519 privB)
-              (C.cbNonce nonce24)
+              (C.unsafeCbNonce nonce24)
               msg
       B.length expected `shouldBe` 80
 
@@ -2363,7 +2363,7 @@ tsClientTests = describe "protocol/client" $ do
             C.cbAuthenticate
               (C.PublicKeyX25519 pubA)
               (C.PrivateKeyX25519 privB)
-              (C.cbNonce nonce24)
+              (C.unsafeCbNonce nonce24)
               msg
       tsResult <-
         callNode $
@@ -2389,7 +2389,7 @@ tsClientTests = describe "protocol/client" $ do
             C.cbAuthenticate
               (C.PublicKeyX25519 pubA)
               (C.PrivateKeyX25519 privB)
-              (C.cbNonce nonce24)
+              (C.unsafeCbNonce nonce24)
               msg
       tsResult <-
         callNode $
@@ -2427,7 +2427,7 @@ tsClientTests = describe "protocol/client" $ do
             C.cbVerify
               (C.PublicKeyX25519 pubB)
               (C.PrivateKeyX25519 privA)
-              (C.cbNonce nonce24)
+              (C.unsafeCbNonce nonce24)
               (C.CbAuthenticator tsAuth)
               msg
       hsValid `shouldBe` True
@@ -2438,7 +2438,7 @@ tsClientTests = describe "protocol/client" $ do
 
     it "encryptTransportChunk matches Haskell" $ do
       let plaintext = B.pack [100 .. 199]
-          state0 = either (error . show) id $ LC.cbInit dhSecret (C.cbNonce nonce24)
+          state0 = either (error . show) id $ LC.cbInit dhSecret (C.unsafeCbNonce nonce24)
           (cipher, state1) = LC.sbEncryptChunk state0 plaintext
           tag = BA.convert $ LC.sbAuth state1 :: B.ByteString
           expected = cipher <> tag
@@ -2457,7 +2457,7 @@ tsClientTests = describe "protocol/client" $ do
 
     it "decryptTransportChunk decrypts Haskell-encrypted data" $ do
       let plaintext = B.pack ([200 .. 255] <> [0 .. 99])
-          state0 = either (error . show) id $ LC.cbInit dhSecret (C.cbNonce nonce24)
+          state0 = either (error . show) id $ LC.cbInit dhSecret (C.unsafeCbNonce nonce24)
           (cipher, state1) = LC.sbEncryptChunk state0 plaintext
           tag = BA.convert $ LC.sbAuth state1 :: B.ByteString
           encData = cipher <> tag
@@ -2542,7 +2542,7 @@ tsDownloadTests = describe "download" $ do
       -- File-level key/nonce (different from transport)
       fileKey32 = B.pack [1 .. 32]
       fileNonce24 = B.pack [1 .. 24]
-      fileCbNonce = C.cbNonce fileNonce24
+      fileCbNonce = C.unsafeCbNonce fileNonce24
       fileSbKey = C.unsafeSbKey fileKey32
 
   describe "processFileResponse" $ do
@@ -2568,7 +2568,7 @@ tsDownloadTests = describe "download" $ do
           dhSecretBytes = case dhSecret of C.DhSecretX25519 k -> BA.convert k :: B.ByteString
           chunkData = B.pack [50 .. 149]
           chunkDigest = C.sha256Hash chunkData
-          state0 = either (error . show) id $ LC.cbInit dhSecret (C.cbNonce nonce24)
+          state0 = either (error . show) id $ LC.cbInit dhSecret (C.unsafeCbNonce nonce24)
           (cipher, state1) = LC.sbEncryptChunk state0 chunkData
           tag = BA.convert (LC.sbAuth state1) :: B.ByteString
           encData = cipher <> tag
@@ -2592,7 +2592,7 @@ tsDownloadTests = describe "download" $ do
           dhSecretBytes = case dhSecret of C.DhSecretX25519 k -> BA.convert k :: B.ByteString
           chunkData = B.pack [50 .. 149]
           wrongDigest = B.replicate 32 0xff
-          state0 = either (error . show) id $ LC.cbInit dhSecret (C.cbNonce nonce24)
+          state0 = either (error . show) id $ LC.cbInit dhSecret (C.unsafeCbNonce nonce24)
           (cipher, state1) = LC.sbEncryptChunk state0 chunkData
           tag = BA.convert (LC.sbAuth state1) :: B.ByteString
           encData = cipher <> tag
@@ -2615,7 +2615,7 @@ tsDownloadTests = describe "download" $ do
       let dhSecret = C.dh' (C.PublicKeyX25519 pubA) (C.PrivateKeyX25519 privB)
           dhSecretBytes = case dhSecret of C.DhSecretX25519 k -> BA.convert k :: B.ByteString
           chunkData = B.pack [10 .. 50]
-          state0 = either (error . show) id $ LC.cbInit dhSecret (C.cbNonce nonce24)
+          state0 = either (error . show) id $ LC.cbInit dhSecret (C.unsafeCbNonce nonce24)
           (cipher, state1) = LC.sbEncryptChunk state0 chunkData
           tag = BA.convert (LC.sbAuth state1) :: B.ByteString
           encData = cipher <> tag
@@ -2652,7 +2652,7 @@ tsDownloadTests = describe "download" $ do
       -- Step 2: transport-level encryption (simulates server sending chunk)
       let dhSecret = C.dh' (C.PublicKeyX25519 pubA) (C.PrivateKeyX25519 privB)
           dhSecretBytes = case dhSecret of C.DhSecretX25519 k -> BA.convert k :: B.ByteString
-          ts0 = either (error . show) id $ LC.cbInit dhSecret (C.cbNonce nonce24)
+          ts0 = either (error . show) id $ LC.cbInit dhSecret (C.unsafeCbNonce nonce24)
           (transportCipher, ts1) = LC.sbEncryptChunk ts0 fileEncrypted
           transportTag = BA.convert (LC.sbAuth ts1) :: B.ByteString
           transportEncData = transportCipher <> transportTag
@@ -2702,7 +2702,7 @@ tsDownloadTests = describe "download" $ do
           dhSecret1 = C.dh' (C.PublicKeyX25519 pubA) (C.PrivateKeyX25519 privB)
           dhSecret1Bytes = case dhSecret1 of C.DhSecretX25519 k -> BA.convert k :: B.ByteString
           nonce1 = nonce24
-          t1s0 = either (error . show) id $ LC.cbInit dhSecret1 (C.cbNonce nonce1)
+          t1s0 = either (error . show) id $ LC.cbInit dhSecret1 (C.unsafeCbNonce nonce1)
           (t1cipher, t1s1) = LC.sbEncryptChunk t1s0 fileChunk1
           t1tag = BA.convert (LC.sbAuth t1s1) :: B.ByteString
           transportEnc1 = t1cipher <> t1tag
@@ -2710,7 +2710,7 @@ tsDownloadTests = describe "download" $ do
           nonce2 = B.pack [24 .. 47]
           dhSecret2 = C.dh' (C.PublicKeyX25519 pubB) (C.PrivateKeyX25519 privA)
           dhSecret2Bytes = case dhSecret2 of C.DhSecretX25519 k -> BA.convert k :: B.ByteString
-          t2s0 = either (error . show) id $ LC.cbInit dhSecret2 (C.cbNonce nonce2)
+          t2s0 = either (error . show) id $ LC.cbInit dhSecret2 (C.unsafeCbNonce nonce2)
           (t2cipher, t2s1) = LC.sbEncryptChunk t2s0 fileChunk2
           t2tag = BA.convert (LC.sbAuth t2s1) :: B.ByteString
           transportEnc2 = t2cipher <> t2tag
@@ -2748,7 +2748,7 @@ tsDownloadTests = describe "download" $ do
       -- Client side: generate FGET command
       let dhSecret = C.dh' (C.PublicKeyX25519 pubA) (C.PrivateKeyX25519 privB)
           chunkData = "FGET round-trip test data" :: B.ByteString
-          state0 = either (error . show) id $ LC.cbInit dhSecret (C.cbNonce nonce24)
+          state0 = either (error . show) id $ LC.cbInit dhSecret (C.unsafeCbNonce nonce24)
           (cipher, state1) = LC.sbEncryptChunk state0 chunkData
           tag = BA.convert (LC.sbAuth state1) :: B.ByteString
           encData = cipher <> tag