delivery to non-current queues

rfcs/2022-08-14-queue-rotation.md

@@ -0,0 +1,57 @@
+# SMP queue rotation
+
+## Problem
+
+1. Long term usage of the queue allows the servers to analyse long term traffic patterns.
+
+2. If the user changes the configured SMP server(s), the previously created contacts do not migrate to the new server(s).
+
+## Solution
+
+Additional messages exchanged by SMP agents to negotiate migration of the queues to the new server. I considered combining this with queue redundancy, but it increases scope and adds a lot of complexity before it is needed.
+
+The proposed approach separates queue rotation from any queue redundancy that may be added in the future.
+
+### Messages
+
+Additional agent messages required for the protocol:
+
+`QNEW`: notify the sender that the queue has to be rotated to the new one, includes the address (server, sender ID and DH key) of the new queue. Encoded as `QN`.
+
+`QKEYS`: pass sender's server key and DH key via existing connection (SMP confirmation message will not be used, to avoid the same "race" of the initial key exchange that would create the risk of intercepting the queue for the attacker), encoded as `QK`.
+
+`QREADY`: instruct the sender that the new is ready to use with sender's queue address as parameter, encoded as `QR` - sender will send HELLO to the new queue.
+
+`QTEST`: sender sends to the new queue to confirm it's working, encoded as `QT`
+
+`QSWITCH`: instruct the sender to use the new queue with sender's queue ID as parameter, encoded as `QS` - sent after receiving `QTEST`.
+
+`QHELLO`: sender sends to the new queue to confirm switch was successful - all new messages after this message will be sent to the new queue. Encoded as `QH`
+
+### Protocol
+
+```
+participant A as Alice
+participant B as Bob
+participant R as Server that has A's receive queue
+participant S as Server that has A's send queue (B's receive queue)
+participant R' as Server that hosts the new A's receive queue
+
+A ->> R': create new queue
+A ->> S ->> B: QNEW (R'): address of the new queue
+B ->> R ->> A: QKEYS (R'): sender's key for the new queue (to avoid the race of SMP confirmation for the initial exchange)
+B ->> R ->> A: continue sending new messages to the old queue
+A ->> R': secure queue
+A ->> S ->> B: QREADY (R'): notify sender that the queue is secured
+B ->> R' ->> A: QTEST: to validate that the sender can send messages to the new queue before switching to it
+A ->> S ->> B: QSWITCH (R'): instruction to start using the new queue
+B ->> R' ->> A: QHELLO: to confirm that the delivery is now switched to the new queue
+B ->> R' ->> A: the first message received to the new queue before the old one is drained and deleted should not be processed, it should be stored in the agent memory (and not acknowledged) and only processed once the old queue is drained.
+A ->> R: suspend queue, receive all messages
+A ->> R: delete queue
+```
+
+It will also require extending SMP protocol:
+
+- add message flag / meta-data indicating that this is the last message and the server has no more messages available (so that the recipient knows when it's safe to delete the queue). Alternatively it can be NUL message that is sent after the last suspended message is received and in response to OFF command.
+- when queue is suspended the server should return the remaining message count. It should be ok to suspend the queue again, so that if the agent is restarted and the queue status is suspended it can be suspended again to check the number of remaining messages.

simplexmq.cabal

@@ -52,6 +52,7 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220625_v2_ntf_mode
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220811_onion_hosts
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220822_queue_rotation
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands
       Simplex.Messaging.Client
       Simplex.Messaging.Client.Agent

src/Simplex/Messaging/Agent.hs

@@ -55,6 +55,7 @@ module Simplex.Messaging.Agent
     resubscribeConnections,
     sendMessage,
     ackMessage,
+    switchConnection,
     suspendConnection,
     deleteConnection,
     getConnectionServers,
@@ -75,7 +76,7 @@ module Simplex.Messaging.Agent
   )
 where
 
-import Control.Concurrent.STM (flushTBQueue, stateTVar)
+import Control.Concurrent.STM (flushTBQueue, retry, stateTVar)
 import Control.Logger.Simple (logInfo, showText)
 import Control.Monad.Except
 import Control.Monad.IO.Unlift (MonadUnliftIO)
@@ -91,9 +92,11 @@ import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (isJust)
+import qualified Data.Set as S
 import qualified Data.Text as T
 import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
+import Data.Word (Word16)
 import qualified Database.SQLite.Simple as DB
 import Simplex.Messaging.Agent.Client
 import Simplex.Messaging.Agent.Env.SQLite
@@ -111,8 +114,9 @@ import Simplex.Messaging.Notifications.Protocol (DeviceToken, NtfRegCode (NtfReg
 import Simplex.Messaging.Notifications.Server.Push.APNS (PNMessageData (..))
 import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Parsers (parse)
-import Simplex.Messaging.Protocol (BrokerMsg, ErrorType (AUTH), MsgBody, MsgFlags, NtfServer, SMPMsgMeta)
+import Simplex.Messaging.Protocol (BrokerMsg, ErrorType (AUTH), MsgBody, MsgFlags, NtfServer, SMPMsgMeta, SndPublicVerifyKey)
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
@@ -210,8 +214,12 @@ sendMessage c = withAgentEnv c .:. sendMessage' c
 ackMessage :: AgentErrorMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessage c = withAgentEnv c .: ackMessage' c
 
+-- | Switch connection to the new receive queue
+switchConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()
+switchConnection c = withAgentEnv c . switchConnection' c
+
 -- | Suspend SMP agent connection (OFF command)
-suspendConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()
+suspendConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m Word16
 suspendConnection c = withAgentEnv c . suspendConnection' c
 
 -- | Delete SMP agent connection (DEL command)
@@ -349,7 +357,7 @@ allowConnectionAsync' c connId confId ownConnInfo =
 ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessageAsync' c connId msgId =
   withStore c (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection _ rq _) -> enqueueAck rq
+    SomeConn _ (DuplexConnection _ rq _ _ _) -> enqueueAck rq
     SomeConn _ (RcvConnection _ rq) -> enqueueAck rq
     SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
     SomeConn _ (ContactConnection _ _) -> throwError $ CMD PROHIBITED
@@ -366,9 +374,9 @@ newConn c connId asyncMode enableNtfs cMode =
 newConnSrv :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> SConnectionMode c -> SMPServer -> m (ConnId, ConnectionRequestUri c)
 newConnSrv c connId asyncMode enableNtfs cMode srv = do
   clientVRange <- asks $ smpClientVRange . config
-  (rq, qUri) <- newRcvQueue c srv clientVRange
+  (rq, qUri) <- newRcvQueue c srv clientVRange True
   connId' <- setUpConn asyncMode rq
-  addSubscription c rq connId'
+  addSubscription c rq connId' True
   when enableNtfs $ do
     ns <- asks ntfSupervisor
     atomically $ sendNtfSubCommand ns (connId', NSCCreate)
@@ -391,8 +399,12 @@ newConnSrv c connId asyncMode enableNtfs cMode srv = do
       withStore c $ \db -> createRcvConn db g cData rq cMode
 
 joinConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
-joinConn c connId asyncMode enableNtfs connReq cInfo =
-  getSMPServer c >>= joinConnSrv c connId asyncMode enableNtfs connReq cInfo
+joinConn c connId asyncMode enableNtfs cReq cInfo = do
+  srv <- case cReq of
+    CRInvitationUri ConnReqUriData {crSmpQueues = SMPQueueUri {queueAddress} :| _} _ ->
+      getNextSMPServer c [smpServer (queueAddress :: SMPQueueAddress)]
+    _ -> getSMPServer c
+  joinConnSrv c connId asyncMode enableNtfs cReq cInfo srv
 
 joinConnSrv :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> SMPServer -> m ConnId
 joinConnSrv c connId asyncMode enableNtfs (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2eRcvParamsUri) cInfo srv = do
@@ -406,7 +418,7 @@ joinConnSrv c connId asyncMode enableNtfs (CRInvitationUri (ConnReqUriData _ age
       (pk1, pk2, e2eSndParams) <- liftIO . CR.generateE2EParams $ version e2eRcvParams
       (_, rcDHRs) <- liftIO C.generateKeyPair'
       let rc = CR.initSndRatchet rcDHRr rcDHRs $ CR.x3dhSnd pk1 pk2 e2eRcvParams
-      sq <- newSndQueue qInfo
+      sq <- newSndQueue qInfo True
       let duplexHS = connAgentVersion /= 1
           cData = ConnData {connId, connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS}
       connId' <- setUpConn asyncMode cData sq rc
@@ -448,9 +460,9 @@ joinConnSrv _c _connId True _enableNtfs (CRContactUri _) _cInfo _srv = do
 
 createReplyQueue :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> SMPServer -> m SMPQueueInfo
 createReplyQueue c ConnData {connId, enableNtfs} SndQueue {smpClientVersion} srv = do
-  (rq, qUri) <- newRcvQueue c srv $ versionToRange smpClientVersion
+  (rq, qUri) <- newRcvQueue c srv (versionToRange smpClientVersion) True
   let qInfo = toVersionT qUri smpClientVersion
-  addSubscription c rq connId
+  addSubscription c rq connId True
   withStore c $ \db -> upgradeSndConnToDuplex db connId rq
   when enableNtfs $ do
     ns <- asks ntfSupervisor
@@ -490,7 +502,9 @@ rejectContact' c contactConnId invId =
 processConfirmation :: AgentMonad m => AgentClient -> RcvQueue -> SMPConfirmation -> m ()
 processConfirmation c rq@RcvQueue {e2ePrivKey, smpClientVersion = v} SMPConfirmation {senderKey, e2ePubKey, smpClientVersion = v'} = do
   let dhSecret = C.dh' e2ePubKey e2ePrivKey
-  withStore' c $ \db -> setRcvQueueConfirmedE2E db rq dhSecret $ min v v'
+  withStore' c $ \db -> setRcvQueueConfirmedE2E db rq senderKey dhSecret $ min v v'
+  -- TODO if this call to secureQueue fails the connection will not complete
+  -- add secure rcv queue on subscription
   secureQueue c rq senderKey
   withStore' c $ \db -> setRcvQueueStatus db rq Secured
 
@@ -500,68 +514,164 @@ subscribeConnection' c connId =
   withStore c (`getConn` connId) >>= \conn -> do
     resumeConnCmds c connId
     case conn of
-      SomeConn _ (DuplexConnection cData rq sq) -> do
+      SomeConn _ (DuplexConnection cData rq sq rq' sq') -> do
         resumeMsgDelivery c cData sq
+        mapM_ (resumeMsgDelivery c cData) sq'
+        void . forkIO $ doRcvQueueAction c cData rq sq
+        mapM_ subscribe rq' `catchError` \_ -> pure ()
         subscribe rq
       SomeConn _ (SndConnection cData sq) -> do
         resumeMsgDelivery c cData sq
         case status (sq :: SndQueue) of
-          Confirmed -> pure ()
+          Confirmed -> pure () -- TODO secure queue if this is a new server version
           Active -> throwError $ CONN SIMPLEX
           _ -> throwError $ INTERNAL "unexpected queue status"
       SomeConn _ (RcvConnection _ rq) -> subscribe rq
       SomeConn _ (ContactConnection _ rq) -> subscribe rq
       SomeConn _ (NewConnection _) -> pure ()
   where
+    -- TODO sndQueueAction?
     subscribe :: RcvQueue -> m ()
-    subscribe rq = do
-      subscribeQueue c rq connId
+    subscribe rq@RcvQueue {currRcvQueue} = do
+      subscribeQueue c rq connId currRcvQueue
       ns <- asks ntfSupervisor
       atomically $ sendNtfSubCommand ns (connId, NSCCreate)
 
+-- TODO expire actions
+doRcvQueueAction :: AgentMonad m => AgentClient -> ConnData -> RcvQueue -> SndQueue -> m ()
+doRcvQueueAction c cData rq@RcvQueue {rcvQueueAction} sq =
+  forM_ rcvQueueAction $ \(a, _ts) -> case a of
+    RQACreateNextQueue -> createNextRcvQueue c cData rq sq
+    RQASecureNextQueue -> withNextRcvQueue secureNextRcvQueue
+    RQASuspendCurrQueue -> withNextRcvQueue suspendCurrRcvQueue
+    RQADeleteCurrQueue -> withNextRcvQueue deleteCurrRcvQueue
+  where
+    withNextRcvQueue :: AgentMonad m => (AgentClient -> ConnData -> RcvQueue -> SndQueue -> RcvQueue -> m ()) -> m ()
+    withNextRcvQueue action = do
+      withStore' c (`getNextRcvQueue` rq) >>= \case
+        Just rq' -> action c cData rq sq rq'
+        _ -> do
+          -- notify agent internal error
+          pure ()
+
+createNextRcvQueue :: AgentMonad m => AgentClient -> ConnData -> RcvQueue -> SndQueue -> m ()
+createNextRcvQueue c cData@ConnData {connId} rq@RcvQueue {server, sndId} sq = do
+  clientVRange <- asks $ smpClientVRange . config
+  nextQueueUri <-
+    withStore' c (`getNextRcvQueue` rq) >>= \case
+      Just RcvQueue {server = smpServer, sndId = senderId, e2ePrivKey} -> do
+        let queueAddress = SMPQueueAddress {smpServer, senderId, dhPublicKey = C.publicKey e2ePrivKey}
+        pure SMPQueueUri {clientVRange, queueAddress}
+      _ -> do
+        srv <- getNextSMPServer c [server]
+        liftIO $ putStrLn $ "createNextRcvQueue " <> show server <> " -> " <> show srv
+        (rq', qUri) <- newRcvQueue c srv clientVRange False
+        withStore' c $ \db -> dbCreateNextRcvQueue db connId rq rq'
+        pure qUri
+  void $ enqueueMessage c cData sq SMP.noMsgFlags QNEW {currentAddress = (server, sndId), nextQueueUri}
+  withStore' c $ \db -> setRcvQueueAction db rq Nothing
+
+secureNextRcvQueue :: AgentMonad m => AgentClient -> ConnData -> RcvQueue -> SndQueue -> RcvQueue -> m ()
+secureNextRcvQueue c cData rq sq rq'@RcvQueue {server, sndId, status, sndPublicKey} = do
+  when (status == Confirmed) $ case sndPublicKey of
+    Just sKey -> do
+      secureQueue c rq' sKey
+      withStore' c $ \db -> setRcvQueueStatus db rq' Secured
+    _ -> do
+      -- TODO notify user: no sender key
+      pure ()
+  void . enqueueMessage c cData sq SMP.noMsgFlags $ QREADY (server, sndId)
+  withStore' c $ \db -> setRcvQueueAction db rq Nothing
+
+suspendCurrRcvQueue :: AgentMonad m => AgentClient -> ConnData -> RcvQueue -> SndQueue -> RcvQueue -> m ()
+suspendCurrRcvQueue c cData rq sq rq' = do
+  msgCount <- suspendQueue c rq
+  withStore' c $ \db -> setRcvQueueStatus db rq Disabled
+  when (msgCount == 0) $ currRcvQueueDrained c cData rq sq rq'
+
+currRcvQueueDrained :: AgentMonad m => AgentClient -> ConnData -> RcvQueue -> SndQueue -> RcvQueue -> m ()
+currRcvQueueDrained c cData rq sq rq' = do
+  withStore' c $ \db -> setRcvQueueAction db rq $ Just RQADeleteCurrQueue
+  deleteCurrRcvQueue c cData rq sq rq'
+
+deleteCurrRcvQueue :: AgentMonad m => AgentClient -> ConnData -> RcvQueue -> SndQueue -> RcvQueue -> m ()
+deleteCurrRcvQueue c cData@ConnData {connId} rq sq rq'@RcvQueue {server, rcvId} = do
+  deleteQueue c rq
+  withStore' c $ \db -> switchCurrRcvQueue db rq rq'
+  atomically $
+    TM.lookupDelete (server, rcvId) (nextRcvQueueMsgs c)
+      >>= mapM_ ((mapM_ . writeTBQueue $ msgQ c) . reverse)
+  sq' <- withStore' c (`getNextSndQueue` sq)
+  let sStats = connectionStats $ DuplexConnection cData rq' sq Nothing sq'
+  atomically $ writeTBQueue (subQ c) ("", connId, SWITCH SPCompleted sStats)
+
 subscribeConnections' :: forall m. AgentMonad m => AgentClient -> [ConnId] -> m (Map ConnId (Either AgentErrorType ()))
 subscribeConnections' _ [] = pure M.empty
 subscribeConnections' c connIds = do
+  -- load connections or errors if they are absent
   conns :: Map ConnId (Either StoreError SomeConn) <- M.fromList . zip connIds <$> withStore' c (forM connIds . getConn)
+  -- split errors to separate map
   let (errs, cs) = M.mapEither id conns
       errs' = M.map (Left . storeError) errs
+      -- split subscription results for connections without rcvQs from rcvQs
       (subRs, rcvQs) = M.mapEither rcvQueueOrResult cs
-      srvRcvQs :: Map SMPServer (Map ConnId (RcvQueue, ConnData)) = M.foldlWithKey' addRcvQueue M.empty rcvQs
-  mapM_ (mapM_ (uncurry $ resumeMsgDelivery c) . sndQueue) cs
+      -- prepare map for batch subscriptions
+      srvRcvQs :: Map SMPServer (Map (ConnId, Bool) RcvQueue) = M.foldlWithKey' addRcvQueues M.empty rcvQs
+  -- TODO start message delivery for non-current queues
+  mapM_ (mapM_ resumeDelivery . sndQueue) cs
   mapM_ (resumeConnCmds c) $ M.keys cs
+  -- send batch subscriptions concurrently to all servers
   rcvRs <- mapConcurrently subscribe (M.assocs srvRcvQs)
+  -- filter out results for secondary queues, leaving only results for the current queues
+  let rcvRs' = map (M.mapKeys fst . M.filterWithKey (\(_, current) _ -> current)) rcvRs
   ns <- asks ntfSupervisor
   tkn <- readTVarIO (ntfTkn ns)
-  when (instantNotifications tkn) . void . forkIO $ sendNtfCreate ns rcvRs
-  let rs = M.unions $ errs' : subRs : rcvRs
+  when (instantNotifications tkn) . void . forkIO $ sendNtfCreate ns rcvRs'
+  let rs = M.unions $ errs' : subRs : rcvRs'
+  -- send notification to the user in case results have a different size from expected
   notifyResultError rs
+  -- perform pending rcvQ actions
+  void . forkIO . forM_ cs $ \case
+    SomeConn _ (DuplexConnection cData rq sq _ _) -> doRcvQueueAction c cData rq sq
+    _ -> pure ()
+  -- TODO secure Confirmed queues if this is a new server version
   pure rs
   where
-    rcvQueueOrResult :: SomeConn -> Either (Either AgentErrorType ()) (RcvQueue, ConnData)
+    rcvQueueOrResult :: SomeConn -> Either (Either AgentErrorType ()) (RcvQueue, Maybe RcvQueue)
     rcvQueueOrResult = \case
-      SomeConn _ (DuplexConnection cData rq _) -> Right (rq, cData)
+      SomeConn _ (DuplexConnection _ rq _ rq' _) -> Right (rq, rq')
       SomeConn _ (SndConnection _ sq) -> Left $ sndSubResult sq
-      SomeConn _ (RcvConnection cData rq) -> Right (rq, cData)
-      SomeConn _ (ContactConnection cData rq) -> Right (rq, cData)
-      SomeConn _ (NewConnection _) -> Left (Right ())
+      SomeConn _ (RcvConnection _ rq) -> Right (rq, Nothing)
+      SomeConn _ (ContactConnection _ rq) -> Right (rq, Nothing)
+      SomeConn _ (NewConnection _) -> Left $ Right ()
     sndSubResult :: SndQueue -> Either AgentErrorType ()
     sndSubResult sq = case status (sq :: SndQueue) of
       Confirmed -> Right ()
       Active -> Left $ CONN SIMPLEX
       _ -> Left $ INTERNAL "unexpected queue status"
-    addRcvQueue :: Map SMPServer (Map ConnId (RcvQueue, ConnData)) -> ConnId -> (RcvQueue, ConnData) -> Map SMPServer (Map ConnId (RcvQueue, ConnData))
-    addRcvQueue m connId rq@(RcvQueue {server}, _) = M.alter (Just . maybe (M.singleton connId rq) (M.insert connId rq)) server m
-    subscribe :: (SMPServer, Map ConnId (RcvQueue, ConnData)) -> m (Map ConnId (Either AgentErrorType ()))
-    subscribe (srv, qs) = snd <$> subscribeQueues c srv (M.map fst qs)
+    addRcvQueues :: Map SMPServer (Map (ConnId, Bool) RcvQueue) -> ConnId -> (RcvQueue, Maybe RcvQueue) -> Map SMPServer (Map (ConnId, Bool) RcvQueue)
+    addRcvQueues m connId (rq, rq') =
+      let m' = addRcvQueue m connId rq
+       in maybe m' (addRcvQueue m' connId) rq'
+    addRcvQueue :: Map SMPServer (Map (ConnId, Bool) RcvQueue) -> ConnId -> RcvQueue -> Map SMPServer (Map (ConnId, Bool) RcvQueue)
+    addRcvQueue m connId rq@RcvQueue {server, currRcvQueue} =
+      let sub = (connId, currRcvQueue)
+       in M.alter (Just . maybe (M.singleton sub rq) (M.insert sub rq)) server m
+    subscribe :: (SMPServer, Map (ConnId, Bool) RcvQueue) -> m (Map (ConnId, Bool) (Either AgentErrorType ()))
+    subscribe (srv, qs) = snd <$> subscribeQueues c srv qs
     sendNtfCreate :: NtfSupervisor -> [Map ConnId (Either AgentErrorType ())] -> m ()
     sendNtfCreate ns rcvRs =
       forM_ (concatMap M.assocs rcvRs) $ \case
         (connId, Right _) -> atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCCreate)
         _ -> pure ()
-    sndQueue :: SomeConn -> Maybe (ConnData, SndQueue)
+    resumeDelivery :: (ConnData, SndQueue, Maybe SndQueue) -> m ()
+    resumeDelivery (cData, sq, sq') = do
+      resumeMsgDelivery c cData sq
+      mapM_ (resumeMsgDelivery c cData) sq'
+    sndQueue :: SomeConn -> Maybe (ConnData, SndQueue, Maybe SndQueue)
     sndQueue = \case
-      SomeConn _ (DuplexConnection cData _ sq) -> Just (cData, sq)
-      SomeConn _ (SndConnection cData sq) -> Just (cData, sq)
+      SomeConn _ (DuplexConnection cData _ sq _ sq') -> Just (cData, sq, sq')
+      SomeConn _ (SndConnection cData sq) -> Just (cData, sq, Nothing)
       _ -> Nothing
     notifyResultError :: Map ConnId (Either AgentErrorType ()) -> m ()
     notifyResultError rs = do
@@ -588,7 +698,7 @@ getConnectionMessage' :: AgentMonad m => AgentClient -> ConnId -> m (Maybe SMPMs
 getConnectionMessage' c connId = do
   whenM (atomically $ hasActiveSubscription c connId) . throwError $ CMD PROHIBITED
   withStore c (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection _ rq _) -> getQueueMessage c rq
+    SomeConn _ (DuplexConnection _ rq _ _ _) -> getQueueMessage c rq
     SomeConn _ (RcvConnection _ rq) -> getQueueMessage c rq
     SomeConn _ (ContactConnection _ rq) -> getQueueMessage c rq
     SomeConn _ SndConnection {} -> throwError $ CONN SIMPLEX
@@ -625,7 +735,7 @@ getNotificationMessage' c nonce encNtfInfo = do
 sendMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> MsgFlags -> MsgBody -> m AgentMsgId
 sendMessage' c connId msgFlags msg =
   withStore c (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection cData _ sq) -> enqueueMsg cData sq
+    SomeConn _ (DuplexConnection cData _ sq _ _) -> enqueueMsg cData sq
     SomeConn _ (SndConnection cData sq) -> enqueueMsg cData sq
     _ -> throwError $ CONN SIMPLEX
   where
@@ -694,6 +804,7 @@ runCommandProcessing c@AgentClient {subQ} server = do
               withNextSrv usedSrvs $ \srv -> do
                 (_, cReq) <- newConnSrv c connId True enableNtfs cMode srv
                 notify connId $ INV (ACR cMode cReq)
+            -- TODO exclude the server of the invitation, as in joinConn
             JOIN enableNtfs (ACR _ cReq) connInfo ->
               withNextSrv usedSrvs $ \srv ->
                 void $ joinConnSrv c connId True enableNtfs cReq connInfo srv
@@ -725,7 +836,7 @@ runCommandProcessing c@AgentClient {subQ} server = do
 -- ^ ^ ^ async command processing /
 
 enqueueMessage :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> MsgFlags -> AMessage -> m AgentMsgId
-enqueueMessage c cData@ConnData {connId, connAgentVersion} sq msgFlags aMessage = do
+enqueueMessage c cData@ConnData {connId, connAgentVersion} sq@SndQueue {currSndQueue} msgFlags aMessage = do
   resumeMsgDelivery c cData sq
   msgId <- storeSentMsg
   queuePendingMsgs c sq [msgId]
@@ -742,22 +853,24 @@ enqueueMessage c cData@ConnData {connId, connAgentVersion} sq msgFlags aMessage
       encAgentMessage <- agentRatchetEncrypt db connId agentMsgStr e2eEncUserMsgLength
       let msgBody = smpEncode $ AgentMsgEnvelope {agentVersion = connAgentVersion, encAgentMessage}
           msgType = agentMessageType agentMsg
-          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody, internalHash, prevMsgHash}
+          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody, internalHash, prevMsgHash, currSndQueue}
       liftIO $ createSndMsg db connId msgData
       pure internalId
 
 resumeMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
-resumeMsgDelivery c cData@ConnData {connId} sq@SndQueue {server, sndId} = do
+resumeMsgDelivery c cData@ConnData {connId} sq@SndQueue {server, sndId, currSndQueue = current} = do
   let qKey = (server, sndId)
   unlessM (queueDelivering qKey) $
     async (runSmpQueueMsgDelivery c cData sq)
       >>= \a -> atomically (TM.insert qKey a $ smpQueueMsgDeliveries c)
   unlessM connQueued $
-    withStore' c (`getPendingMsgs` connId)
+    withStore' c (\db -> getPendingMsgs db connId current)
       >>= queuePendingMsgs c sq
   where
     queueDelivering qKey = atomically $ TM.member qKey (smpQueueMsgDeliveries c)
-    connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connMsgsQueued c)
+    connQueued = atomically $
+      stateTVar (connMsgsQueued c) $ \s ->
+        let k = (connId, current) in (S.member k s, S.insert k s)
 
 queuePendingMsgs :: AgentMonad m => AgentClient -> SndQueue -> [InternalId] -> m ()
 queuePendingMsgs c sq msgIds = atomically $ do
@@ -791,7 +904,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
     E.try (withStore c $ \db -> getPendingMsgData db connId msgId) >>= \case
       Left (e :: E.SomeException) ->
         notify $ MERR mId (INTERNAL $ show e)
-      Right (rq_, PendingMsgData {msgType, msgBody, msgFlags, internalTs}) ->
+      Right (rq_, PendingMsgData {msgType, msgBody, msgFlags, internalTs}) -> do
         withRetryInterval ri $ \loop -> do
           resp <- tryError $ case msgType of
             AM_CONN_INFO -> sendConfirmation c sq msgBody
@@ -803,6 +916,12 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                 SMP SMP.QUOTA -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
+                  AM_QTEST_ -> do
+                    -- cancel switching, delete new send queue
+                    pure ()
+                  AM_QHELLO_ -> do
+                    -- cancel switching, delete new send queue
+                    pure ()
                   _ -> retrySending loop
                 SMP SMP.AUTH -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
@@ -821,6 +940,18 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                         _ -> connError msgId NOT_ACCEPTED
                   AM_REPLY_ -> notifyDel msgId $ ERR e
                   AM_A_MSG_ -> notifyDel msgId $ MERR mId e
+                  AM_QNEW_ -> pure ()
+                  AM_QKEYS_ -> do
+                    -- TODO new send queue status = Confirmed
+                    pure ()
+                  AM_QREADY_ -> pure ()
+                  AM_QTEST_ -> do
+                    -- cancel switching, delete new send queue
+                    pure ()
+                  AM_QSWITCH_ -> pure ()
+                  AM_QHELLO_ -> do
+                    -- cancel switching, delete new send queue
+                    pure ()
                 _
                   -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server),
                   -- the message sending would be retried
@@ -846,10 +977,6 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                   case rq_ of
                     -- party initiating connection (in v1)
                     Just RcvQueue {status} ->
-                      -- it is unclear why subscribeQueue was needed here,
-                      -- message delivery can only be enabled for queues that were created in the current session or subscribed
-                      -- subscribeQueue c rq connId
-                      --
                       -- If initiating party were to send CON to the user without waiting for reply HELLO (to reduce handshake time),
                       -- it would lead to the non-deterministic internal ID of the first sent message, at to some other race conditions,
                       -- because it can be sent before HELLO is received
@@ -864,6 +991,10 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                       qInfo <- createReplyQueue c cData sq srv
                       void . enqueueMessage c cData sq SMP.noMsgFlags $ REPLY [qInfo]
                 AM_A_MSG_ -> notify $ SENT mId
+                AM_QHELLO_ -> do
+                  -- withStore' c $ \db -> setSndQueueStatus db sq Active
+                  -- what else should happen here?
+                  pure ()
                 _ -> pure ()
               delMsg msgId
   where
@@ -883,7 +1014,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
 ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessage' c connId msgId = do
   withStore c (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection _ rq _) -> ack rq
+    SomeConn _ (DuplexConnection _ rq _ _ _) -> ack rq
     SomeConn _ (RcvConnection _ rq) -> ack rq
     SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
     SomeConn _ (ContactConnection _ _) -> throwError $ CMD PROHIBITED
@@ -898,11 +1029,22 @@ ackMessage' c connId msgId = do
         e -> throwError e
       withStore' c $ \db -> deleteMsg db connId mId
 
+-- | Switch connection to the new receive queue
+switchConnection' :: AgentMonad m => AgentClient -> ConnId -> m ()
+switchConnection' c connId =
+  withStore c (`getConn` connId) >>= \case
+    SomeConn _ (DuplexConnection cData rq sq _ _) -> do
+      -- TODO check that rotation is possible (whether the current server supports it)
+      withStore' c $ \db -> setRcvQueueAction db rq $ Just RQACreateNextQueue
+      createNextRcvQueue c cData rq sq
+    SomeConn _ SndConnection {} -> throwError $ CONN SIMPLEX
+    _ -> throwError $ CMD PROHIBITED
+
 -- | Suspend SMP agent connection (OFF command) in Reader monad
-suspendConnection' :: AgentMonad m => AgentClient -> ConnId -> m ()
+suspendConnection' :: AgentMonad m => AgentClient -> ConnId -> m Word16
 suspendConnection' c connId =
   withStore c (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection _ rq _) -> suspendQueue c rq
+    SomeConn _ (DuplexConnection _ rq _ _ _) -> suspendQueue c rq
     SomeConn _ (RcvConnection _ rq) -> suspendQueue c rq
     SomeConn _ (ContactConnection _ rq) -> suspendQueue c rq
     SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
@@ -912,7 +1054,7 @@ suspendConnection' c connId =
 deleteConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
 deleteConnection' c connId =
   withStore c (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection _ rq _) -> delete rq
+    SomeConn _ (DuplexConnection _ rq _ nextRq_ _) -> delete rq >> mapM_ (deleteQueue c) nextRq_
     SomeConn _ (RcvConnection _ rq) -> delete rq
     SomeConn _ (ContactConnection _ rq) -> delete rq
     SomeConn _ (SndConnection _ _) -> withStore' c (`deleteConn` connId)
@@ -927,15 +1069,20 @@ deleteConnection' c connId =
       atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCDelete)
 
 getConnectionServers' :: AgentMonad m => AgentClient -> ConnId -> m ConnectionStats
-getConnectionServers' c connId = connServers <$> withStore c (`getConn` connId)
+getConnectionServers' c connId = do
+  SomeConn _ conn <- withStore c (`getConn` connId)
+  pure $ connectionStats conn
+
+connectionStats :: Connection c -> ConnectionStats
+connectionStats conn = case conn of
+  RcvConnection _ rq -> ConnectionStats {rcvServers = rcvSrvs rq, sndServers = [], nextRcvServers = [], nextSndServers = []}
+  SndConnection _ sq -> ConnectionStats {rcvServers = [], sndServers = sndSrvs sq, nextRcvServers = [], nextSndServers = []}
+  DuplexConnection _ rq sq nextRq_ nextSq_ -> ConnectionStats {rcvServers = rcvSrvs rq, sndServers = sndSrvs sq, nextRcvServers = maybe [] rcvSrvs nextRq_, nextSndServers = maybe [] sndSrvs nextSq_}
+  ContactConnection _ rq -> ConnectionStats {rcvServers = rcvSrvs rq, sndServers = [], nextRcvServers = [], nextSndServers = []}
+  NewConnection _ -> ConnectionStats {rcvServers = [], sndServers = [], nextRcvServers = [], nextSndServers = []}
   where
-    connServers :: SomeConn -> ConnectionStats
-    connServers = \case
-      SomeConn _ (RcvConnection _ RcvQueue {server}) -> ConnectionStats {rcvServers = [server], sndServers = []}
-      SomeConn _ (SndConnection _ SndQueue {server}) -> ConnectionStats {rcvServers = [], sndServers = [server]}
-      SomeConn _ (DuplexConnection _ RcvQueue {server = s1} SndQueue {server = s2}) -> ConnectionStats {rcvServers = [s1], sndServers = [s2]}
-      SomeConn _ (ContactConnection _ RcvQueue {server}) -> ConnectionStats {rcvServers = [server], sndServers = []}
-      SomeConn _ (NewConnection _) -> ConnectionStats {rcvServers = [], sndServers = []}
+    rcvSrvs RcvQueue {server} = [server]
+    sndSrvs SndQueue {server} = [server]
 
 -- | Change servers to be used for creating new queues, in Reader monad
 setSMPServers' :: AgentMonad m => AgentClient -> NonEmpty SMPServer -> m ()
@@ -1068,7 +1215,7 @@ getNtfTokenData' c =
 toggleConnectionNtfs' :: forall m. AgentMonad m => AgentClient -> ConnId -> Bool -> m ()
 toggleConnectionNtfs' c connId enable = do
   withStore c (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection cData _ _) -> toggle cData
+    SomeConn _ (DuplexConnection cData _ _ _ _) -> toggle cData
     SomeConn _ (RcvConnection cData _) -> toggle cData
     SomeConn _ (ContactConnection cData _) -> toggle cData
     _ -> throwError $ CONN SIMPLEX
@@ -1184,11 +1331,11 @@ pickServer = \case
 getNextSMPServer :: AgentMonad m => AgentClient -> [SMPServer] -> m SMPServer
 getNextSMPServer c usedSrvs = do
   srvs <- readTVarIO $ smpServers c
-  case L.nonEmpty $ deleteFirstsBy different (L.toList srvs) usedSrvs of
+  case L.nonEmpty $ deleteFirstsBy sameAddress (L.toList srvs) usedSrvs of
     Just srvs' -> pickServer srvs'
     _ -> pickServer srvs
   where
-    different (SMPServer host port _) (SMPServer host' port' _) = host /= host' || port /= port'
+    sameAddress (SMPServer host port _) (SMPServer host' port' _) = host == host' && port == port'
 
 subscriber :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
 subscriber c@AgentClient {msgQ} = forever $ do
@@ -1199,15 +1346,12 @@ subscriber c@AgentClient {msgQ} = forever $ do
       Right _ -> return ()
 
 processSMPTransmission :: forall m. AgentMonad m => AgentClient -> ServerTransmission BrokerMsg -> m ()
-processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cmd) =
-  withStore c (\db -> getRcvConn db srv rId) >>= \case
-    SomeConn _ conn@(DuplexConnection cData rq _) -> processSMP conn cData rq
-    SomeConn _ conn@(RcvConnection cData rq) -> processSMP conn cData rq
-    SomeConn _ conn@(ContactConnection cData rq) -> processSMP conn cData rq
-    _ -> atomically $ writeTBQueue subQ ("", "", ERR $ CONN NOT_FOUND)
+processSMPTransmission c@AgentClient {smpClients, subQ} transmission@(srv, v, sessId, rId, cmd) = do
+  (rq, SomeConn _ conn) <- withStore c $ \db -> getRcvConn db srv rId
+  processSMP conn (connData conn) rq
   where
     processSMP :: Connection c -> ConnData -> RcvQueue -> m ()
-    processSMP conn cData@ConnData {connId, duplexHandshake} rq@RcvQueue {e2ePrivKey, e2eDhSecret, status} =
+    processSMP conn cData@ConnData {connId, duplexHandshake} rq@RcvQueue {e2ePrivKey, e2eDhSecret, status, currRcvQueue} =
       case cmd of
         SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} -> handleNotifyAck $ do
           SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} <- decryptSMPMessage v rq msg
@@ -1217,6 +1361,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
           unless (phVer `isCompatible` clientVRange) . throwError $ AGENT A_VERSION
           case (e2eDhSecret, e2ePubKey_) of
             (Nothing, Just e2ePubKey) -> do
+              unless (currRcvQueue) . throwError $ INTERNAL "can only be sent to the current queue"
               let e2eDh = C.dh' e2ePubKey e2ePrivKey
               decryptClientMessage e2eDh clientMsg >>= \case
                 (SMP.PHConfirmation senderKey, AgentConfirmation {e2eEncryption, encConnInfo, agentVersion}) ->
@@ -1229,12 +1374,22 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                 (SMP.PHEmpty, AgentMsgEnvelope _ encAgentMsg) ->
                   tryError agentClientMsg >>= \case
                     Right (Just (msgId, msgMeta, aMessage)) -> case aMessage of
-                      HELLO -> helloMsg >> ack >> withStore' c (\db -> deleteMsg db connId msgId)
-                      REPLY cReq -> replyMsg cReq >> ack >> withStore' c (\db -> deleteMsg db connId msgId)
+                      HELLO -> helloMsg >> ackDelete msgId
+                      REPLY cReq -> replyMsg cReq >> ackDelete 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>"
-                        notify $ MSG msgMeta msgFlags body
+                      A_MSG body
+                        | currRcvQueue -> do
+                          logServer "<--" c srv rId "MSG <MSG>"
+                          notify $ MSG msgMeta msgFlags body
+                        | otherwise -> atomically $ TM.alter addTransmission (srv, rId) (nextRcvQueueMsgs c)
+                        where
+                          addTransmission = Just . maybe [transmission] (transmission :)
+                      QNEW currAddr nextQUri -> rqNewMsg currAddr nextQUri >> ackDelete msgId
+                      QKEYS sKey nextQInfo -> rqKeys sKey nextQInfo $ ackDelete msgId
+                      QREADY addr -> rqReady addr >> ackDelete msgId
+                      QTEST -> rqTest >> ackDelete msgId
+                      QSWITCH addr -> rqSwitch addr >> ackDelete msgId
+                      QHELLO -> rqHello $ ackDelete msgId
                     Right _ -> prohibited >> ack
                     Left e@(AGENT A_DUPLICATE) -> do
                       withStore' c (\db -> getLastMsg db connId srvMsgId) >>= \case
@@ -1244,9 +1399,18 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                             withStore' c $ \db -> deleteMsg db connId internalId
                           | otherwise -> do
                             liftEither (parse smpP (AGENT A_MESSAGE) agentMsgBody) >>= \case
-                              AgentMessage _ (A_MSG body) -> do
-                                logServer "<--" c srv rId "MSG <MSG>"
-                                notify $ MSG msgMeta msgFlags body
+                              AgentMessage _ (A_MSG body)
+                                | currRcvQueue -> do
+                                  logServer "<--" c srv rId "MSG <MSG>"
+                                  notify $ MSG msgMeta msgFlags body
+                                | otherwise -> atomically $ TM.alter addTransmission (srv, rId) (nextRcvQueueMsgs c)
+                                where
+                                  addTransmission = Just . maybe [transmission] prependIfDifferent
+                                  prependIfDifferent = \case
+                                    [] -> [transmission]
+                                    ts@((_, _, _, _, cmd') : _)
+                                      | cmd == cmd' -> ts
+                                      | otherwise -> transmission : ts
                               _ -> pure ()
                         _ -> throwError e
                     Left e -> throwError e
@@ -1276,6 +1440,8 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
               sendAck c rq srvMsgId `catchError` \case
                 SMP SMP.NO_MSG -> pure ()
                 e -> throwError e
+            ackDelete :: InternalId -> m ()
+            ackDelete msgId = ack >> withStore' c (\db -> deleteMsg db connId msgId)
             handleNotifyAck :: m () -> m ()
             handleNotifyAck m = m `catchError` \e -> notify (ERR e) >> ack
         SMP.END ->
@@ -1291,6 +1457,10 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                 | otherwise -> ignored
               _ -> ignored
             ignored = pure "END from disconnected client - ignored"
+        SMP.LEN 0 -> do
+          -- load nextRq
+          -- currRcvQueueDrained c rq nextRq
+          pure ()
         _ -> do
           logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
           notify . ERR $ BROKER UNEXPECTED
@@ -1351,7 +1521,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                       queueServer (SMPQueueInfo _ SMPQueueAddress {smpServer}) = smpServer
                   _ -> prohibited
               -- party accepting connection
-              (DuplexConnection _ _ sq, Nothing) -> do
+              (DuplexConnection _ _ sq _ _, Nothing) -> do
                 withStore c (\db -> runExceptT $ agentRatchetDecrypt db connId encConnInfo) >>= parseMessage >>= \case
                   AgentConnInfo connInfo -> do
                     notify $ INFO connInfo
@@ -1363,13 +1533,14 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
 
         helloMsg :: m ()
         helloMsg = do
+          unless currRcvQueue . throwError $ INTERNAL "can only be sent to the current queue"
           logServer "<--" c srv rId "MSG <HELLO>"
           case status of
             Active -> prohibited
             _ -> do
               withStore' c $ \db -> setRcvQueueStatus db rq Active
               case conn of
-                DuplexConnection _ _ sq@SndQueue {status = sndStatus}
+                DuplexConnection _ _ sq@SndQueue {status = sndStatus} _ _
                   -- `sndStatus == Active` when HELLO was previously sent, and this is the reply HELLO
                   -- this branch is executed by the accepting party in duplexHandshake mode (v2)
                   -- and by the initiating party in v1
@@ -1384,6 +1555,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
 
         replyMsg :: L.NonEmpty SMPQueueInfo -> m ()
         replyMsg smpQueues = do
+          unless currRcvQueue . throwError $ INTERNAL "can only be sent to the current queue"
           logServer "<--" c srv rId "MSG <REPLY>"
           case duplexHandshake of
             Just True -> prohibited
@@ -1393,6 +1565,122 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                 connectReplyQueues c cData ownConnInfo smpQueues `catchError` (notify . ERR)
               _ -> prohibited
 
+        -- processed by queue sender
+        rqNewMsg :: (SMPServer, SMP.SenderId) -> SMPQueueUri -> m ()
+        rqNewMsg (smpServer, senderId) nextQUri = case conn of
+          DuplexConnection _ _ sq@SndQueue {server, sndId, currSndQueue = curr} nextRq_ _ -> do
+            liftIO $ print $ "rqNewMsg " <> show (SMP.port server) <> " " <> show curr
+            unless (smpServer == server && senderId == sndId) . throwError $ INTERNAL "incorrect queue address"
+            clientVRange <- asks $ smpClientVRange . config
+            case (nextQUri `compatibleVersion` clientVRange) of
+              Just qInfo@(Compatible nextQInfo) -> do
+                sq'@SndQueue {sndPublicKey, e2ePubKey, server = srv', currSndQueue = curr'} <- newSndQueue qInfo False
+                withStore' c $ \db -> dbCreateNextSndQueue db connId sq sq'
+                liftIO $ print $ "rqNewMsg: next SndQueue " <> show (SMP.port srv') <> " " <> show curr'
+                case (sndPublicKey, e2ePubKey) of
+                  (Just nextSenderKey, Just dhPublicKey) -> do
+                    let qAddr = (queueAddress (nextQInfo :: SMPQueueInfo)) {dhPublicKey}
+                        nextQueueInfo = (nextQInfo :: SMPQueueInfo) {queueAddress = qAddr}
+                    void $ enqueueMessage c cData sq SMP.noMsgFlags QKEYS {nextSenderKey, nextQueueInfo}
+                    let conn' = DuplexConnection cData rq sq nextRq_ (Just sq')
+                    notify . SWITCH SPStarted $ connectionStats conn'
+                  _ -> throwError $ INTERNAL "absent sender keys"
+              _ -> throwError $ AGENT A_VERSION
+          _ -> throwError $ INTERNAL "message can only be sent to duplex connection"
+
+        -- processed by queue recipient
+        rqKeys :: SndPublicVerifyKey -> SMPQueueInfo -> m () -> m ()
+        rqKeys senderKey qInfo@(SMPQueueInfo clntVer' SMPQueueAddress {smpServer, senderId, dhPublicKey}) ackDelete = do
+          unless currRcvQueue . throwError $ INTERNAL "message can only be sent to current queue"
+          liftIO $ print $ "rqKeys " <> show (SMP.port srv) <> " " <> show currRcvQueue
+          case conn of
+            DuplexConnection _ _ sq nextRq_ _ -> do
+              clientVRange <- asks $ smpClientVRange . config
+              unless (qInfo `isCompatible` clientVRange) . throwError $ AGENT A_VERSION
+              case nextRq_ of
+                Just rq'@RcvQueue {server, sndId, e2ePrivKey = dhPrivKey, smpClientVersion = clntVer, currRcvQueue = curr'} -> do
+                  liftIO $ print $ "rqKeys next RcvQueue " <> show (SMP.port server) <> " " <> show curr'
+                  unless (smpServer == server && senderId == sndId) . throwError $ INTERNAL "incorrect queue address"
+                  let dhSecret = C.dh' dhPublicKey dhPrivKey
+                  withStore' c $ \db -> do
+                    setRcvQueueConfirmedE2E db rq' senderKey dhSecret $ min clntVer clntVer'
+                    setRcvQueueAction db rq $ Just RQASecureNextQueue
+                  ackDelete
+                  secureNextRcvQueue c cData rq sq rq'
+                _ -> throwError $ INTERNAL "message can only be sent during rotation"
+            _ -> throwError $ INTERNAL "message can only be sent to duplex connection"
+
+        -- processed by queue sender
+        rqReady :: (SMPServer, SMP.SenderId) -> m ()
+        rqReady (smpServer, senderId) = case conn of
+          DuplexConnection _ _ SndQueue {server = srv1, currSndQueue = curr} _ nextSq_ -> do
+            liftIO $ print $ "rqReady " <> show (SMP.port srv1) <> " " <> show curr
+            case nextSq_ of
+              Just sq'@SndQueue {server, sndId, currSndQueue = curr'} -> do
+                liftIO $ print $ "rqReady next SndQueue " <> show (SMP.port server) <> " " <> show curr'
+                unless (smpServer == server && senderId == sndId) . throwError $ INTERNAL "incorrect queue address"
+                void $ enqueueMessage c cData sq' SMP.noMsgFlags QTEST
+              _ -> throwError $ INTERNAL "message can only be sent during rotation"
+          _ -> throwError $ INTERNAL "message can only be sent to duplex connection"
+
+        -- processed by queue recipient, received from the new queue
+        rqTest :: m ()
+        rqTest = do
+          liftIO $ print $ "rqTest " <> show (SMP.port srv) <> " " <> show currRcvQueue
+          when currRcvQueue . throwError $ INTERNAL "2: message can only be sent to the next queue"
+          case conn of
+            DuplexConnection _ _ sq _ _ -> do
+              let RcvQueue {server, sndId} = rq
+              void . enqueueMessage c cData sq SMP.noMsgFlags $ QSWITCH (server, sndId)
+            _ -> throwError $ INTERNAL "message can only be sent to duplex connection"
+
+        -- processed by queue sender
+        rqSwitch :: (SMPServer, SMP.SenderId) -> m ()
+        rqSwitch (smpServer, senderId) = case conn of
+          DuplexConnection _ _ sq@SndQueue {server, sndId} nextRq_ nextSq_ -> case nextSq_ of
+            Just sq'@SndQueue {server = server', sndId = sndId'} -> do
+              unless (smpServer == server' && senderId == sndId') . throwError $ INTERNAL "incorrect queue address"
+              let qKey = (server, sndId)
+                  qKey' = (server', sndId')
+              ok <-
+                switchQueues qKey qKey' `catchError` \e -> do
+                  atomically (switchDeliveries qKey' qKey)
+                  throwError e
+              unless ok $ throwError $ INTERNAL "switching snd queue failed in STM"
+              void $ enqueueMessage c cData sq' SMP.noMsgFlags QHELLO
+              let conn' = DuplexConnection cData rq sq' nextRq_ Nothing
+              notify . SWITCH SPCompleted $ connectionStats conn'
+              where
+                switchQueues :: MsgDeliveryKey -> MsgDeliveryKey -> m Bool
+                switchQueues k k' = withStore' c $ \db -> do
+                  ok <- atomically $ (switchDeliveries k k' $> True) `orElse` pure False
+                  when ok $ switchCurrSndQueue db sq
+                  pure ok
+                switchDeliveries :: MsgDeliveryKey -> MsgDeliveryKey -> STM ()
+                switchDeliveries k k' = do
+                  switchDelivery smpQueueMsgQueues k k'
+                  switchDelivery smpQueueMsgDeliveries k k'
+                switchDelivery :: (AgentClient -> TMap MsgDeliveryKey a) -> MsgDeliveryKey -> MsgDeliveryKey -> STM ()
+                switchDelivery sel k k' =
+                  TM.lookupDelete k (sel c) >>= \case
+                    Just d -> TM.insert k' d (sel c)
+                    _ -> retry
+            _ -> throwError $ INTERNAL "message can only be sent during rotation"
+          _ -> throwError $ INTERNAL "message can only be sent to duplex connection"
+
+        -- processed by queue recipient, received from the new queue
+        rqHello :: m () -> m ()
+        rqHello ackDelete = do
+          when currRcvQueue . throwError $ INTERNAL "1: message can only be sent to the next queue"
+          case conn of
+            DuplexConnection _ currRq sq _ _ -> do
+              withStore' c $ \db -> do
+                setRcvQueueStatus db rq Active
+                setRcvQueueAction db currRq $ Just RQASuspendCurrQueue
+              ackDelete
+              suspendCurrRcvQueue c cData currRq sq rq
+            _ -> throwError $ INTERNAL "message can only be sent to duplex connection"
+
         smpInvitation :: ConnectionRequestUri 'CMInvitation -> ConnInfo -> m ()
         smpInvitation connReq@(CRInvitationUri crData _) cInfo = do
           logServer "<--" c srv rId "MSG <KEY>"
@@ -1422,7 +1710,7 @@ connectReplyQueues c cData@ConnData {connId} ownConnInfo (qInfo :| _) = do
   case qInfo `proveCompatible` clientVRange of
     Nothing -> throwError $ AGENT A_VERSION
     Just qInfo' -> do
-      sq <- newSndQueue qInfo'
+      sq <- newSndQueue qInfo' True
       withStore c $ \db -> upgradeRcvConnToDuplex db connId sq
       enqueueConfirmation c cData sq ownConnInfo Nothing
 
@@ -1460,7 +1748,7 @@ enqueueConfirmation c cData@ConnData {connId, connAgentVersion} sq connInfo e2eE
       encConnInfo <- agentRatchetEncrypt db connId agentMsgStr e2eEncConnInfoLength
       let msgBody = smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption, encConnInfo}
           msgType = agentMessageType agentMsg
-          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash}
+          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, currSndQueue = True}
       liftIO $ createSndMsg db connId msgData
       pure internalId
 
@@ -1481,20 +1769,22 @@ agentRatchetDecrypt db connId encAgentMsg = do
   liftIO $ updateRatchet db connId rc' skippedDiff
   liftEither $ first (SEAgentError . cryptoError) agentMsgBody_
 
-newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> m SndQueue
-newSndQueue qInfo =
+newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> Bool -> m SndQueue
+newSndQueue qInfo current =
   asks (cmdSignAlg . config) >>= \case
-    C.SignAlg a -> newSndQueue_ a qInfo
+    C.SignAlg a -> newSndQueue_ a qInfo current
 
 newSndQueue_ ::
   (C.SignatureAlgorithm a, C.AlgorithmI a, MonadUnliftIO m) =>
   C.SAlgorithm a ->
   Compatible SMPQueueInfo ->
+  Bool ->
   m SndQueue
-newSndQueue_ a (Compatible (SMPQueueInfo smpClientVersion SMPQueueAddress {smpServer, senderId, dhPublicKey = rcvE2ePubDhKey})) = do
+newSndQueue_ a (Compatible (SMPQueueInfo smpClientVersion SMPQueueAddress {smpServer, senderId, dhPublicKey = rcvE2ePubDhKey})) current = do
   -- this function assumes clientVersion is compatible - it was tested before
   (sndPublicKey, sndPrivateKey) <- liftIO $ C.generateSignatureKeyPair a
   (e2ePubKey, e2ePrivKey) <- liftIO C.generateKeyPair'
+  createdAt <- liftIO getCurrentTime
   pure
     SndQueue
       { server = smpServer,
@@ -1504,5 +1794,10 @@ newSndQueue_ a (Compatible (SMPQueueInfo smpClientVersion SMPQueueAddress {smpSe
         e2eDhSecret = C.dh' rcvE2ePubDhKey e2ePrivKey,
         e2ePubKey = Just e2ePubKey,
         status = New,
-        smpClientVersion
+        currSndQueue = current,
+        dbNextSndQueueId = Nothing,
+        sndQueueAction = Nothing,
+        smpClientVersion,
+        createdAt,
+        updatedAt = createdAt
       }

src/Simplex/Messaging/Agent/Client.hs

@@ -16,6 +16,7 @@
 
 module Simplex.Messaging.Agent.Client
   ( AgentClient (..),
+    MsgDeliveryKey,
     newAgentClient,
     withAgentLock,
     closeAgentClient,
@@ -90,6 +91,7 @@ import Data.Maybe (listToMaybe)
 import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text.Encoding
+import Data.Time.Clock (getCurrentTime)
 import Data.Word (Word16)
 import qualified Database.SQLite.Simple as DB
 import Simplex.Messaging.Agent.Env.SQLite
@@ -146,6 +148,8 @@ type SMPClientVar = TMVar (Either AgentErrorType SMPClient)
 
 type NtfClientVar = TMVar (Either AgentErrorType NtfClient)
 
+type MsgDeliveryKey = (SMPServer, SMP.SenderId)
+
 data AgentClient = AgentClient
   { active :: TVar Bool,
     rcvQ :: TBQueue (ATransmission 'Client),
@@ -157,11 +161,13 @@ data AgentClient = AgentClient
     ntfClients :: TMap NtfServer NtfClientVar,
     useNetworkConfig :: TVar NetworkConfig,
     subscrConns :: TVar (Set ConnId),
-    activeSubs :: TMap2 SMPServer ConnId RcvQueue,
-    pendingSubs :: TMap2 SMPServer ConnId RcvQueue,
-    connMsgsQueued :: TMap ConnId Bool,
-    smpQueueMsgQueues :: TMap (SMPServer, SMP.SenderId) (TQueue InternalId),
-    smpQueueMsgDeliveries :: TMap (SMPServer, SMP.SenderId) (Async ()),
+    -- Bool in tuple keys shows whether the queue is the current one (True) or the one the connection switches to (False)
+    activeSubs :: TMap2 SMPServer (ConnId, Bool) RcvQueue,
+    pendingSubs :: TMap2 SMPServer (ConnId, Bool) RcvQueue,
+    connMsgsQueued :: TVar (Set (ConnId, Bool)),
+    smpQueueMsgQueues :: TMap MsgDeliveryKey (TQueue InternalId),
+    smpQueueMsgDeliveries :: TMap MsgDeliveryKey (Async ()),
+    nextRcvQueueMsgs :: TMap (SMPServer, SMP.RecipientId) [ServerTransmission BrokerMsg],
     connCmdsQueued :: TMap ConnId Bool,
     asyncCmdQueues :: TMap (Maybe SMPServer) (TQueue AsyncCmdId),
     asyncCmdProcesses :: TMap (Maybe SMPServer) (Async ()),
@@ -213,9 +219,10 @@ newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do
   subscrConns <- newTVar S.empty
   activeSubs <- TM2.empty
   pendingSubs <- TM2.empty
-  connMsgsQueued <- TM.empty
+  connMsgsQueued <- newTVar S.empty
   smpQueueMsgQueues <- TM.empty
   smpQueueMsgDeliveries <- TM.empty
+  nextRcvQueueMsgs <- TM.empty
   connCmdsQueued <- TM.empty
   asyncCmdQueues <- TM.empty
   asyncCmdProcesses <- TM.empty
@@ -230,7 +237,7 @@ newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do
   asyncClients <- newTVar []
   clientId <- stateTVar (clientCounter agentEnv) $ \i -> let i' = i + 1 in (i', i')
   lock <- newTMVar ()
-  return AgentClient {active, rcvQ, subQ, msgQ, smpServers, smpClients, ntfServers, ntfClients, useNetworkConfig, subscrConns, activeSubs, pendingSubs, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, connCmdsQueued, asyncCmdQueues, asyncCmdProcesses, ntfNetworkOp, rcvNetworkOp, msgDeliveryOp, sndNetworkOp, databaseOp, agentState, getMsgLocks, reconnections, asyncClients, clientId, agentEnv, lock}
+  return AgentClient {active, rcvQ, subQ, msgQ, smpServers, smpClients, ntfServers, ntfClients, useNetworkConfig, subscrConns, activeSubs, pendingSubs, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, nextRcvQueueMsgs, connCmdsQueued, asyncCmdQueues, asyncCmdProcesses, ntfNetworkOp, rcvNetworkOp, msgDeliveryOp, sndNetworkOp, databaseOp, agentState, getMsgLocks, reconnections, asyncClients, clientId, agentEnv, lock}
 
 agentDbPath :: AgentClient -> FilePath
 agentDbPath AgentClient {agentEnv = Env {store = SQLiteStore {dbFilePath}}} = dbFilePath
@@ -266,7 +273,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do
       removeClientAndSubs >>= (`forM_` serverDown)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
       where
-        removeClientAndSubs :: IO (Maybe (Map ConnId RcvQueue))
+        removeClientAndSubs :: IO (Maybe (Map (ConnId, Bool) RcvQueue))
         removeClientAndSubs = atomically $ do
           TM.delete srv smpClients
           TM2.lookupDelete1 srv (activeSubs c) >>= mapM updateSubs
@@ -275,12 +282,12 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do
               TM2.insert1 srv cVar $ pendingSubs c
               readTVar cVar
 
-        serverDown :: Map ConnId RcvQueue -> IO ()
+        serverDown :: Map (ConnId, Bool) RcvQueue -> IO ()
         serverDown cs = whenM (readTVarIO active) $ do
           notifySub "" $ hostEvent DISCONNECT client
-          let conns = M.keys cs
-          unless (null conns) $ do
-            notifySub "" $ DOWN srv conns
+          let conns = map fst . filter snd $ M.keys cs
+          unless (null conns) $ notifySub "" $ DOWN srv conns
+          unless (null cs) $ do
             atomically $ mapM_ (releaseGetLock c) cs
             unliftIO u reconnectServer
 
@@ -301,15 +308,15 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do
         atomically (TM2.lookup1 srv (pendingSubs c) >>= mapM readTVar)
           >>= mapM_ resubscribe
       where
-        resubscribe :: Map ConnId RcvQueue -> m ()
+        resubscribe :: Map (ConnId, Bool) RcvQueue -> m ()
         resubscribe qs = do
           (client_, (errs, oks)) <- second (M.mapEither id) <$> subscribeQueues c srv qs
           liftIO $ do
             mapM_ (notifySub "" . hostEvent CONNECT) client_
-            unless (M.null oks) $ do
-              notifySub "" . UP srv $ M.keys oks
+            let conns = map fst . filter snd $ M.keys oks
+            unless (null conns) $ notifySub "" $ UP srv conns
           let (tempErrs, finalErrs) = M.partition temporaryAgentError errs
-          liftIO . mapM_ (\(connId, e) -> notifySub connId $ ERR e) $ M.assocs finalErrs
+          liftIO . mapM_ (\((connId, current), e) -> when current . notifySub connId $ ERR e) $ M.assocs finalErrs
           mapM_ throwError . listToMaybe $ M.elems tempErrs
 
     notifySub :: ConnId -> ACommand 'Agent -> IO ()
@@ -470,10 +477,10 @@ protocolClientError protocolError_ = \case
   e@PCESignatureError {} -> INTERNAL $ show e
   e@PCEIOError {} -> INTERNAL $ show e
 
-newRcvQueue :: AgentMonad m => AgentClient -> SMPServer -> VersionRange -> m (RcvQueue, SMPQueueUri)
-newRcvQueue c srv vRange =
+newRcvQueue :: AgentMonad m => AgentClient -> SMPServer -> VersionRange -> Bool -> m (RcvQueue, SMPQueueUri)
+newRcvQueue c srv vRange current =
   asks (cmdSignAlg . config) >>= \case
-    C.SignAlg a -> newRcvQueue_ a c srv vRange
+    C.SignAlg a -> newRcvQueue_ a c srv vRange current
 
 newRcvQueue_ ::
   (C.SignatureAlgorithm a, C.AlgorithmI a, AgentMonad m) =>
@@ -481,14 +488,16 @@ newRcvQueue_ ::
   AgentClient ->
   SMPServer ->
   VersionRange ->
+  Bool ->
   m (RcvQueue, SMPQueueUri)
-newRcvQueue_ a c srv vRange = do
+newRcvQueue_ a c srv vRange current = do
   (recipientKey, rcvPrivateKey) <- liftIO $ C.generateSignatureKeyPair a
   (dhKey, privDhKey) <- liftIO C.generateKeyPair'
   (e2eDhKey, e2ePrivKey) <- liftIO C.generateKeyPair'
   logServer "-->" c srv "" "NEW"
   QIK {rcvId, sndId, rcvPublicDhKey} <-
     withClient c srv $ \smp -> createSMPQueue smp rcvPrivateKey recipientKey dhKey
+  createdAt <- liftIO getCurrentTime
   logServer "<--" c srv "" $ B.unwords ["IDS", logSecret rcvId, logSecret sndId]
   let rq =
         RcvQueue
@@ -498,30 +507,36 @@ newRcvQueue_ a c srv vRange = do
             rcvDhSecret = C.dh' rcvPublicDhKey privDhKey,
             e2ePrivKey,
             e2eDhSecret = Nothing,
-            sndId = Just sndId,
+            sndId,
+            sndPublicKey = Nothing,
             status = New,
+            rcvQueueAction = Nothing,
+            currRcvQueue = current,
+            dbNextRcvQueueId = Nothing,
+            clientNtfCreds = Nothing,
             smpClientVersion = maxVersion vRange,
-            clientNtfCreds = Nothing
+            createdAt,
+            updatedAt = createdAt
           }
   pure (rq, SMPQueueUri vRange $ SMPQueueAddress srv sndId e2eDhKey)
 
-subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
-subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
+subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> Bool -> m ()
+subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId current = do
   whenM (atomically . TM.member (server, rcvId) $ getMsgLocks c) . throwError $ CMD PROHIBITED
   atomically $ do
-    modifyTVar (subscrConns c) $ S.insert connId
-    TM2.insert server connId rq $ pendingSubs c
+    when current $ modifyTVar (subscrConns c) $ S.insert connId
+    TM2.insert server (connId, current) rq $ pendingSubs c
   withLogClient c server rcvId "SUB" $ \smp ->
-    liftIO (runExceptT (subscribeSMPQueue smp rcvPrivateKey rcvId) >>= processSubResult c rq connId)
+    liftIO (runExceptT (subscribeSMPQueue smp rcvPrivateKey rcvId) >>= processSubResult c rq connId current)
       >>= either throwError pure
 
-processSubResult :: AgentClient -> RcvQueue -> ConnId -> Either ProtocolClientError () -> IO (Either ProtocolClientError ())
-processSubResult c rq connId r = do
+processSubResult :: AgentClient -> RcvQueue -> ConnId -> Bool -> Either ProtocolClientError () -> IO (Either ProtocolClientError ())
+processSubResult c rq connId current r = do
   case r of
     Left e ->
       atomically . unless (temporaryClientError e) $
-        TM2.delete connId (pendingSubs c)
-    _ -> addSubscription c rq connId
+        TM2.delete (connId, current) (pendingSubs c)
+    _ -> addSubscription c rq connId current
   pure r
 
 temporaryClientError :: ProtocolClientError -> Bool
@@ -537,12 +552,12 @@ temporaryAgentError = \case
   _ -> False
 
 -- | subscribe multiple queues - all passed queues should be on the same server
-subscribeQueues :: AgentMonad m => AgentClient -> SMPServer -> Map ConnId RcvQueue -> m (Maybe SMPClient, Map ConnId (Either AgentErrorType ()))
+subscribeQueues :: AgentMonad m => AgentClient -> SMPServer -> Map (ConnId, Bool) RcvQueue -> m (Maybe SMPClient, Map (ConnId, Bool) (Either AgentErrorType ()))
 subscribeQueues c srv qs = do
   (errs, qs_) <- partitionEithers <$> mapM checkQueue (M.assocs qs)
-  forM_ qs_ $ \(connId, rq@RcvQueue {server}) -> atomically $ do
-    modifyTVar (subscrConns c) $ S.insert connId
-    TM2.insert server connId rq $ pendingSubs c
+  forM_ qs_ $ \(sub@(connId, current), rq@RcvQueue {server}) -> atomically $ do
+    when current . modifyTVar (subscrConns c) $ S.insert connId
+    TM2.insert server sub rq $ pendingSubs c
   case L.nonEmpty qs_ of
     Just qs' -> do
       smp_ <- tryError (getSMPServerClient c srv)
@@ -551,9 +566,9 @@ subscribeQueues c srv qs = do
         Right smp -> do
           logServer "-->" c srv (bshow (length qs_) <> " queues") "SUB"
           let qs2 = L.map (queueCreds . snd) qs'
-          rs' :: [((ConnId, RcvQueue), Either ProtocolClientError ())] <-
+          rs' :: [(((ConnId, Bool), RcvQueue), Either ProtocolClientError ())] <-
             liftIO $ zip qs_ . L.toList <$> subscribeSMPQueues smp qs2
-          forM_ rs' $ \((connId, rq), r) -> liftIO $ processSubResult c rq connId r
+          forM_ rs' $ \(((connId, current), rq), r) -> liftIO $ processSubResult c rq connId current r
           pure $ map (bimap fst (first $ protocolClientError SMP)) rs'
     _ -> pure $ (Nothing, M.fromList errs)
   where
@@ -562,20 +577,25 @@ subscribeQueues c srv qs = do
       pure $ if prohibited || srv /= server then Left (connId, Left $ CMD PROHIBITED) else Right rq
     queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)
 
-addSubscription :: MonadIO m => AgentClient -> RcvQueue -> ConnId -> m ()
-addSubscription c rq@RcvQueue {server} connId = atomically $ do
+addSubscription :: MonadIO m => AgentClient -> RcvQueue -> ConnId -> Bool -> m ()
+addSubscription c rq@RcvQueue {server} connId current = atomically $ do
   modifyTVar (subscrConns c) $ S.insert connId
-  TM2.insert server connId rq $ activeSubs c
-  TM2.delete connId $ pendingSubs c
+  let sub = (connId, current)
+  TM2.insert server sub rq $ activeSubs c
+  TM2.delete sub $ pendingSubs c
 
 hasActiveSubscription :: AgentClient -> ConnId -> STM Bool
-hasActiveSubscription c connId = TM2.member connId $ activeSubs c
+hasActiveSubscription c connId = TM2.member (connId, True) $ activeSubs c
 
 removeSubscription :: AgentClient -> ConnId -> STM ()
 removeSubscription c connId = do
   modifyTVar (subscrConns c) $ S.delete connId
-  TM2.delete connId $ activeSubs c
-  TM2.delete connId $ pendingSubs c
+  delete (connId, True)
+  delete (connId, False)
+  where
+    delete sub = do
+      TM2.delete sub $ activeSubs c
+      TM2.delete sub $ pendingSubs c
 
 getSubscriptions :: AgentClient -> STM (Set ConnId)
 getSubscriptions = readTVar . subscrConns
@@ -659,7 +679,7 @@ releaseGetLock :: AgentClient -> RcvQueue -> STM ()
 releaseGetLock c RcvQueue {server, rcvId} =
   TM.lookup (server, rcvId) (getMsgLocks c) >>= mapM_ (`tryPutTMVar` ())
 
-suspendQueue :: AgentMonad m => AgentClient -> RcvQueue -> m ()
+suspendQueue :: AgentMonad m => AgentClient -> RcvQueue -> m Word16
 suspendQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
   withLogClient c server rcvId "OFF" $ \smp ->
     suspendSMPQueue smp rcvPrivateKey rcvId

src/Simplex/Messaging/Agent/Protocol.hs

@@ -45,6 +45,7 @@ module Simplex.Messaging.Agent.Protocol
     SAParty (..),
     MsgHash,
     MsgMeta (..),
+    SwitchPhase (..),
     ConnectionStats (..),
     SMPConfirmation (..),
     AgentMsgEnvelope (..),
@@ -237,6 +238,7 @@ data ACommand (p :: AParty) where
   DISCONNECT :: AProtocolType -> TransportHost -> ACommand Agent
   DOWN :: SMPServer -> [ConnId] -> ACommand Agent
   UP :: SMPServer -> [ConnId] -> ACommand Agent
+  SWITCH :: SwitchPhase -> ConnectionStats -> ACommand Agent
   SEND :: MsgFlags -> MsgBody -> ACommand Client
   MID :: AgentMsgId -> ACommand Agent
   SENT :: AgentMsgId -> ACommand Agent
@@ -255,19 +257,41 @@ deriving instance Eq (ACommand p)
 
 deriving instance Show (ACommand p)
 
+data SwitchPhase = SPStarted | SPCompleted
+  deriving (Eq, Show)
+
+instance StrEncoding SwitchPhase where
+  strEncode = \case
+    SPStarted -> "started"
+    SPCompleted -> "completed"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "started" -> pure SPStarted
+      "completed" -> pure SPCompleted
+      _ -> fail "bad SwitchPhase"
+
 data ConnectionStats = ConnectionStats
   { rcvServers :: [SMPServer],
-    sndServers :: [SMPServer]
+    sndServers :: [SMPServer],
+    nextRcvServers :: [SMPServer],
+    nextSndServers :: [SMPServer]
   }
   deriving (Eq, Show, Generic)
 
 instance StrEncoding ConnectionStats where
-  strEncode ConnectionStats {rcvServers, sndServers} =
-    "rcv=" <> strEncodeList rcvServers <> " snd=" <> strEncodeList sndServers
+  strEncode s =
+    B.unwords
+      [ "rcv=" <> strEncodeList (rcvServers s),
+        "snd=" <> strEncodeList (sndServers s),
+        "next_rcv=" <> strEncodeList (nextRcvServers s),
+        "next_snd=" <> strEncodeList (nextSndServers s)
+      ]
   strP = do
     rcvServers <- "rcv=" *> strListP
     sndServers <- " snd=" *> strListP
-    pure ConnectionStats {rcvServers, sndServers}
+    nextRcvServers <- " next_rcv=" *> strListP
+    nextSndServers <- " next_snd=" *> strListP
+    pure ConnectionStats {rcvServers, sndServers, nextRcvServers, nextSndServers}
 
 instance ToJSON ConnectionStats where toEncoding = J.genericToEncoding J.defaultOptions
 
@@ -429,7 +453,18 @@ instance Encoding AgentMessage where
       'M' -> AgentMessage <$> smpP <*> smpP
       _ -> fail "bad AgentMessage"
 
-data AgentMessageType = AM_CONN_INFO | AM_CONN_INFO_REPLY | AM_HELLO_ | AM_REPLY_ | AM_A_MSG_
+data AgentMessageType
+  = AM_CONN_INFO
+  | AM_CONN_INFO_REPLY
+  | AM_HELLO_
+  | AM_REPLY_
+  | AM_A_MSG_
+  | AM_QNEW_
+  | AM_QKEYS_
+  | AM_QREADY_
+  | AM_QTEST_
+  | AM_QSWITCH_
+  | AM_QHELLO_
   deriving (Eq, Show)
 
 instance Encoding AgentMessageType where
@@ -439,6 +474,12 @@ instance Encoding AgentMessageType where
     AM_HELLO_ -> "H"
     AM_REPLY_ -> "R"
     AM_A_MSG_ -> "M"
+    AM_QNEW_ -> "QN"
+    AM_QKEYS_ -> "QK"
+    AM_QREADY_ -> "QR"
+    AM_QTEST_ -> "QT"
+    AM_QSWITCH_ -> "QS"
+    AM_QHELLO_ -> "QH"
   smpP =
     A.anyChar >>= \case
       'C' -> pure AM_CONN_INFO
@@ -446,6 +487,15 @@ instance Encoding AgentMessageType where
       'H' -> pure AM_HELLO_
       'R' -> pure AM_REPLY_
       'M' -> pure AM_A_MSG_
+      'Q' ->
+        A.anyChar >>= \case
+          'N' -> pure AM_QNEW_
+          'K' -> pure AM_QKEYS_
+          'R' -> pure AM_QREADY_
+          'T' -> pure AM_QTEST_
+          'S' -> pure AM_QSWITCH_
+          'H' -> pure AM_QHELLO_
+          _ -> fail "bad AgentMessageType"
       _ -> fail "bad AgentMessageType"
 
 agentMessageType :: AgentMessage -> AgentMessageType
@@ -461,6 +511,12 @@ agentMessageType = \case
     -- REPLY is only used in v1
     REPLY _ -> AM_REPLY_
     A_MSG _ -> AM_A_MSG_
+    QNEW {} -> AM_QNEW_
+    QKEYS {} -> AM_QKEYS_
+    QREADY {} -> AM_QREADY_
+    QTEST -> AM_QTEST_
+    QSWITCH {} -> AM_QSWITCH_
+    QHELLO -> AM_QHELLO_
 
 data APrivHeader = APrivHeader
   { -- | sequential ID assigned by the sending agent
@@ -475,7 +531,16 @@ instance Encoding APrivHeader where
     smpEncode (sndMsgId, prevMsgHash)
   smpP = APrivHeader <$> smpP <*> smpP
 
-data AMsgType = HELLO_ | REPLY_ | A_MSG_
+data AMsgType
+  = HELLO_
+  | REPLY_
+  | A_MSG_
+  | QNEW_
+  | QKEYS_
+  | QREADY_
+  | QTEST_
+  | QSWITCH_
+  | QHELLO_
   deriving (Eq)
 
 instance Encoding AMsgType where
@@ -483,11 +548,26 @@ instance Encoding AMsgType where
     HELLO_ -> "H"
     REPLY_ -> "R"
     A_MSG_ -> "M"
+    QNEW_ -> "QN"
+    QKEYS_ -> "QK"
+    QREADY_ -> "QR"
+    QTEST_ -> "QT"
+    QSWITCH_ -> "QS"
+    QHELLO_ -> "QH"
   smpP =
-    smpP >>= \case
+    A.anyChar >>= \case
       'H' -> pure HELLO_
       'R' -> pure REPLY_
       'M' -> pure A_MSG_
+      'Q' ->
+        A.anyChar >>= \case
+          'N' -> pure QNEW_
+          'K' -> pure QKEYS_
+          'R' -> pure QREADY_
+          'T' -> pure QTEST_
+          'S' -> pure QSWITCH_
+          'H' -> pure QHELLO_
+          _ -> fail "bad AMsgType"
       _ -> fail "bad AMsgType"
 
 -- | Messages sent between SMP agents once SMP queue is secured.
@@ -500,6 +580,18 @@ data AMessage
     REPLY (L.NonEmpty SMPQueueInfo)
   | -- | agent envelope for the client message
     A_MSG MsgBody
+  | -- | instruct sender to switch the queue to another
+    QNEW {currentAddress :: (SMPServer, SMP.SenderId), nextQueueUri :: SMPQueueUri}
+  | -- | send server key and queue e2e DH key to the recipient
+    QKEYS {nextSenderKey :: SndPublicVerifyKey, nextQueueInfo :: SMPQueueInfo}
+  | -- | inform the sender that the queue is ready to use - sender sends QHELLO to it
+    QREADY {nextAddress :: (SMPServer, SMP.SenderId)}
+  | -- | the message sent by the sender to the new queue to test delivery
+    QTEST
+  | -- | instruct the sender to start sending messages to the new queue - after recipient receives HELLO
+    QSWITCH {nextAddress :: (SMPServer, SMP.SenderId)}
+  | -- | confirm that the delivery is switched, all new messages will be sent to the new queue
+    QHELLO
   deriving (Show)
 
 instance Encoding AMessage where
@@ -507,12 +599,27 @@ instance Encoding AMessage where
     HELLO -> smpEncode HELLO_
     REPLY smpQueues -> smpEncode (REPLY_, smpQueues)
     A_MSG body -> smpEncode (A_MSG_, Tail body)
+    QNEW currAddr nextQUri -> smpEncode (QNEW_, currAddr, strEncode nextQUri)
+    QKEYS sKey nextQInfo -> smpEncode (QKEYS_, sKey, nextQInfo)
+    QREADY addr -> smpEncode (QREADY_, addr)
+    QTEST -> smpEncode QTEST_
+    QSWITCH addr -> smpEncode (QSWITCH_, addr)
+    QHELLO -> smpEncode QHELLO_
   smpP =
     smpP
       >>= \case
         HELLO_ -> pure HELLO
         REPLY_ -> REPLY <$> smpP
         A_MSG_ -> A_MSG . unTail <$> smpP
+        QNEW_ -> do
+          currentAddress <- smpP
+          nextQueueUri <- strDecode <$?> smpP
+          pure QNEW {currentAddress, nextQueueUri}
+        QKEYS_ -> QKEYS <$> smpP <*> smpP
+        QREADY_ -> QREADY <$> smpP
+        QTEST_ -> pure QTEST
+        QSWITCH_ -> QSWITCH <$> smpP
+        QHELLO_ -> pure QHELLO
 
 instance forall m. ConnectionModeI m => StrEncoding (ConnectionRequestUri m) where
   strEncode = \case
@@ -948,6 +1055,7 @@ commandP parseByteString =
     <|> "DISCONNECT " *> disconnectResp
     <|> "DOWN " *> downResp
     <|> "UP " *> upResp
+    <|> "SWITCH " *> switchResp
     <|> "SEND " *> sendCmd
     <|> "MID " *> msgIdResp
     <|> "SENT " *> sentResp
@@ -975,6 +1083,7 @@ commandP parseByteString =
     disconnectResp = ACmd SAgent .: DISCONNECT <$> strP_ <*> strP
     downResp = ACmd SAgent .: DOWN <$> strP_ <*> connections
     upResp = ACmd SAgent .: UP <$> strP_ <*> connections
+    switchResp = ACmd SAgent .: SWITCH <$> strP_ <*> strP
     sendCmd = ACmd SClient .: SEND <$> smpP <* A.space <*> parseByteString
     msgIdResp = ACmd SAgent . MID <$> A.decimal
     sentResp = ACmd SAgent . SENT <$> A.decimal
@@ -1013,6 +1122,7 @@ serializeCommand = \case
   DISCONNECT p h -> B.unwords ["DISCONNECT", strEncode p, strEncode h]
   DOWN srv conns -> B.unwords ["DOWN", strEncode srv, connections conns]
   UP srv conns -> B.unwords ["UP", strEncode srv, connections conns]
+  SWITCH phase srvs -> B.unwords ["SWITCH", strEncode phase, strEncode srvs]
   SEND msgFlags msgBody -> "SEND " <> smpEncode msgFlags <> " " <> serializeBinary msgBody
   MID mId -> "MID " <> bshow mId
   SENT mId -> "SENT " <> bshow mId

src/Simplex/Messaging/Agent/Store.hs

@@ -3,7 +3,9 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE StandaloneDeriving #-}
 {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
 
@@ -18,6 +20,7 @@ import Data.Type.Equality
 import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (RatchetX448)
+import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
   ( MsgBody,
     MsgFlags,
@@ -49,13 +52,25 @@ data RcvQueue = RcvQueue
     -- | public sender's DH key and agreed shared DH secret for simple per-queue e2e
     e2eDhSecret :: Maybe C.DhSecretX25519,
     -- | sender queue ID
-    sndId :: Maybe SMP.SenderId,
+    sndId :: SMP.SenderId,
+    -- | public key used by the server to verify sender's transmissions
+    -- it is Maybe as previously it was not saved - old queues may have NULL in it.
+    -- For all new queues it is never cleared.
+    sndPublicKey :: Maybe C.APublicVerifyKey,
     -- | queue status
     status :: QueueStatus,
+    -- | action to perform, to be done on connection subscription, if it fails and not reset
+    rcvQueueAction :: Maybe (RcvQueueAction, UTCTime),
+    -- | True for the current receive queue
+    currRcvQueue :: Bool,
+    -- | database ID of the new queue created for this queue to switch to (queue rotation)
+    dbNextRcvQueueId :: Maybe Int64,
+    -- | credentials used in context of notifications
+    clientNtfCreds :: Maybe ClientNtfCreds,
     -- | SMP client version
     smpClientVersion :: Version,
-    -- | credentials used in context of notifications
-    clientNtfCreds :: Maybe ClientNtfCreds
+    createdAt :: UTCTime,
+    updatedAt :: UTCTime
   }
   deriving (Eq, Show)
 
@@ -76,6 +91,8 @@ data SndQueue = SndQueue
     -- | sender queue ID
     sndId :: SMP.SenderId,
     -- | key pair used by the sender to sign transmissions
+    -- sndPublicKey is Maybe as previously it was not saved - old queues may have NULL in it.
+    -- For all new queues it is never cleared.
     sndPublicKey :: Maybe C.APublicVerifyKey,
     sndPrivateKey :: SndPrivateSignKey,
     -- | DH public key used to negotiate per-queue e2e encryption
@@ -84,8 +101,16 @@ data SndQueue = SndQueue
     e2eDhSecret :: C.DhSecretX25519,
     -- | queue status
     status :: QueueStatus,
+    -- | action to perform, to be done on connection subscription, if it fails and not reset
+    sndQueueAction :: Maybe (SndQueueAction, UTCTime),
+    -- | True for the current send queue
+    currSndQueue :: Bool,
+    -- | database ID of the new queue created for this queue to switch to (queue rotation)
+    dbNextSndQueueId :: Maybe Int64,
     -- | SMP client version
-    smpClientVersion :: Version
+    smpClientVersion :: Version,
+    createdAt :: UTCTime,
+    updatedAt :: UTCTime
   }
   deriving (Eq, Show)
 
@@ -108,13 +133,21 @@ data Connection (d :: ConnType) where
   NewConnection :: ConnData -> Connection CNew
   RcvConnection :: ConnData -> RcvQueue -> Connection CRcv
   SndConnection :: ConnData -> SndQueue -> Connection CSnd
-  DuplexConnection :: ConnData -> RcvQueue -> SndQueue -> Connection CDuplex
+  DuplexConnection :: ConnData -> RcvQueue -> SndQueue -> Maybe RcvQueue -> Maybe SndQueue -> Connection CDuplex
   ContactConnection :: ConnData -> RcvQueue -> Connection CContact
 
 deriving instance Eq (Connection d)
 
 deriving instance Show (Connection d)
 
+connData :: Connection d -> ConnData
+connData = \case
+  NewConnection cData -> cData
+  RcvConnection cData _ -> cData
+  SndConnection cData _ -> cData
+  DuplexConnection cData _ _ _ _ -> cData
+  ContactConnection cData _ -> cData
+
 data SConnType :: ConnType -> Type where
   SCNew :: SConnType CNew
   SCRcv :: SConnType CRcv
@@ -159,6 +192,36 @@ data ConnData = ConnData
   }
   deriving (Eq, Show)
 
+data RcvQueueAction
+  = RQACreateNextQueue
+  | RQASecureNextQueue
+  | RQASuspendCurrQueue
+  | RQADeleteCurrQueue
+  deriving (Eq, Show)
+
+instance TextEncoding RcvQueueAction where
+  textEncode = \case
+    RQACreateNextQueue -> "create"
+    RQASecureNextQueue -> "secure"
+    RQASuspendCurrQueue -> "suspend"
+    RQADeleteCurrQueue -> "delete"
+  textDecode = \case
+    "create" -> Just RQACreateNextQueue
+    "secure" -> Just RQASecureNextQueue
+    "suspend" -> Just RQASuspendCurrQueue
+    "delete" -> Just RQADeleteCurrQueue
+    _ -> Nothing
+
+data SndQueueAction = SQASwitchQueue
+  deriving (Eq, Show)
+
+instance TextEncoding SndQueueAction where
+  textEncode = \case
+    SQASwitchQueue -> "switch"
+  textDecode = \case
+    "switch" -> Just SQASwitchQueue
+    _ -> Nothing
+
 -- * Confirmation types
 
 data NewConfirmation = NewConfirmation
@@ -230,7 +293,8 @@ data SndMsgData = SndMsgData
     msgFlags :: MsgFlags,
     msgBody :: MsgBody,
     internalHash :: MsgHash,
-    prevMsgHash :: MsgHash
+    prevMsgHash :: MsgHash,
+    currSndQueue :: Bool
   }
 
 data PendingMsgData = PendingMsgData
@@ -238,7 +302,8 @@ data PendingMsgData = PendingMsgData
     msgType :: AgentMessageType,
     msgFlags :: MsgFlags,
     msgBody :: MsgBody,
-    internalTs :: InternalTs
+    internalTs :: InternalTs,
+    currSndQueue :: Bool
   }
   deriving (Show)
 

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

@@ -15,6 +15,7 @@
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
+{-# LANGUAGE TypeOperators #-}
 {-# LANGUAGE UndecidableInstances #-}
 {-# OPTIONS_GHC -fno-warn-orphans #-}
 
@@ -40,6 +41,13 @@ module Simplex.Messaging.Agent.Store.SQLite
     setSndQueueStatus,
     getRcvQueue,
     setRcvQueueNtfCreds,
+    getNextRcvQueue,
+    getNextSndQueue,
+    dbCreateNextRcvQueue,
+    dbCreateNextSndQueue,
+    setRcvQueueAction,
+    switchCurrRcvQueue,
+    switchCurrSndQueue,
     -- Confirmations
     createConfirmation,
     acceptConfirmation,
@@ -283,7 +291,7 @@ updateNewConnRcv db connId rq@RcvQueue {server} =
     updateConn :: IO (Either StoreError ())
     updateConn = do
       upsertServer_ db server
-      insertRcvQueue_ db connId rq
+      void $ insertRcvQueue_ db connId rq
       pure $ Right ()
 
 updateNewConnSnd :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError ())
@@ -296,7 +304,7 @@ updateNewConnSnd db connId sq@SndQueue {server} =
     updateConn :: IO (Either StoreError ())
     updateConn = do
       upsertServer_ db server
-      insertSndQueue_ db connId sq
+      void $ insertSndQueue_ db connId sq
       pure $ Right ()
 
 createRcvConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> IO (Either StoreError ConnId)
@@ -304,28 +312,36 @@ createRcvConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandsh
   createConn_ gVar cData $ \connId -> do
     upsertServer_ db server
     DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?, ?, ?, ?, ?)" (connId, cMode, connAgentVersion, enableNtfs, duplexHandshake)
-    insertRcvQueue_ db connId q
+    void $ insertRcvQueue_ db connId q
 
 createSndConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> SndQueue -> IO (Either StoreError ConnId)
-createSndConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandshake} q@SndQueue {server} =
+createSndConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandshake} q@SndQueue {server} = do
   createConn_ gVar cData $ \connId -> do
     upsertServer_ db server
     DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?, ?, ?, ?, ?)" (connId, SCMInvitation, connAgentVersion, enableNtfs, duplexHandshake)
-    insertSndQueue_ db connId q
+    -- TODO add queue ID in insertSndQueue_
+    void $ insertSndQueue_ db connId q
 
-getRcvConn :: DB.Connection -> SMPServer -> SMP.RecipientId -> IO (Either StoreError SomeConn)
-getRcvConn db ProtocolServer {host, port} rcvId =
-  DB.queryNamed
-    db
-    [sql|
-      SELECT q.conn_id
-      FROM rcv_queues q
-      WHERE q.host = :host AND q.port = :port AND q.rcv_id = :rcv_id;
-    |]
-    [":host" := host, ":port" := port, ":rcv_id" := rcvId]
-    >>= \case
-      [Only connId] -> getConn db connId
-      _ -> pure $ Left SEConnNotFound
+getRcvConn :: DB.Connection -> SMPServer -> SMP.RecipientId -> IO (Either StoreError (RcvQueue, SomeConn))
+getRcvConn db ProtocolServer {host, port} rcvId = runExceptT $ do
+  (rq, connId) <-
+    ExceptT . firstRow (\(qRow :. Only connId) -> (toRcvQueue qRow, connId)) SEConnNotFound $
+      DB.query
+        db
+        [sql|
+          SELECT q.host, q.port, s.key_hash,
+            q.rcv_id, q.rcv_private_key, q.rcv_dh_secret, q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.snd_key, q.status,
+            q.rcv_queue_action, q.rcv_queue_action_ts, q.curr_rcv_queue, q.next_rcv_queue_id,
+            q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
+            q.smp_client_version, q.created_at, q.updated_at,
+            q.conn_id
+          FROM rcv_queues q
+          INNER JOIN servers s ON q.host = s.host AND q.port = s.port
+          WHERE q.host = ? AND q.port = ? AND q.rcv_id = ?
+        |]
+        (host, port, rcvId)
+  conn <- ExceptT $ getConn db connId
+  pure (rq, conn)
 
 deleteConn :: DB.Connection -> ConnId -> IO ()
 deleteConn db connId =
@@ -339,19 +355,19 @@ upgradeRcvConnToDuplex db connId sq@SndQueue {server} =
   getConn db connId $>>= \case
     (SomeConn _ RcvConnection {}) -> do
       upsertServer_ db server
-      insertSndQueue_ db connId sq
+      -- TODO save with queue ID
+      void $ insertSndQueue_ db connId sq
       pure $ Right ()
     (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
 
 upgradeSndConnToDuplex :: DB.Connection -> ConnId -> RcvQueue -> IO (Either StoreError ())
 upgradeSndConnToDuplex db connId rq@RcvQueue {server} =
-  getConn db connId >>= \case
-    Right (SomeConn _ SndConnection {}) -> do
+  getConn db connId $>>= \case
+    SomeConn _ SndConnection {} -> do
       upsertServer_ db server
-      insertRcvQueue_ db connId rq
+      void $ insertRcvQueue_ db connId rq
       pure $ Right ()
-    Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
-    _ -> pure $ Left SEConnNotFound
+    SomeConn c _ -> pure . Left . SEBadConnType $ connType c
 
 setRcvQueueStatus :: DB.Connection -> RcvQueue -> QueueStatus -> IO ()
 setRcvQueueStatus db RcvQueue {rcvId, server = ProtocolServer {host, port}} status =
@@ -365,19 +381,21 @@ setRcvQueueStatus db RcvQueue {rcvId, server = ProtocolServer {host, port}} stat
     |]
     [":status" := status, ":host" := host, ":port" := port, ":rcv_id" := rcvId]
 
-setRcvQueueConfirmedE2E :: DB.Connection -> RcvQueue -> C.DhSecretX25519 -> Version -> IO ()
-setRcvQueueConfirmedE2E db RcvQueue {rcvId, server = ProtocolServer {host, port}} e2eDhSecret smpClientVersion =
+setRcvQueueConfirmedE2E :: DB.Connection -> RcvQueue -> C.APublicVerifyKey -> C.DhSecretX25519 -> Version -> IO ()
+setRcvQueueConfirmedE2E db RcvQueue {rcvId, server = ProtocolServer {host, port}} sndPublicKey e2eDhSecret smpClientVersion =
   DB.executeNamed
     db
     [sql|
       UPDATE rcv_queues
       SET e2e_dh_secret = :e2e_dh_secret,
+          snd_key = :snd_key,
           status = :status,
           smp_client_version = :smp_client_version
       WHERE host = :host AND port = :port AND rcv_id = :rcv_id
     |]
     [ ":status" := Confirmed,
       ":e2e_dh_secret" := e2eDhSecret,
+      ":snd_key" := sndPublicKey,
       ":smp_client_version" := smpClientVersion,
       ":host" := host,
       ":port" := port,
@@ -415,6 +433,89 @@ setRcvQueueNtfCreds db connId clientNtfCreds =
       Just ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret} -> (Just ntfPublicKey, Just ntfPrivateKey, Just notifierId, Just rcvNtfDhSecret)
       Nothing -> (Nothing, Nothing, Nothing, Nothing)
 
+getNextRcvQueue :: DB.Connection -> RcvQueue -> IO (Maybe RcvQueue)
+getNextRcvQueue db RcvQueue {dbNextRcvQueueId} = case dbNextRcvQueueId of
+  Just rqId ->
+    maybeFirstRow toRcvQueue $
+      DB.query
+        db
+        [sql|
+          SELECT q.host, q.port, s.key_hash,
+            q.rcv_id, q.rcv_private_key, q.rcv_dh_secret, q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.snd_key, q.status,
+            q.rcv_queue_action, q.rcv_queue_action_ts, q.curr_rcv_queue, q.next_rcv_queue_id,
+            q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
+            q.smp_client_version, q.created_at, q.updated_at
+          FROM rcv_queues q
+          INNER JOIN servers s ON q.host = s.host AND q.port = s.port
+          WHERE q.rcv_queue_id = ? AND q.curr_rcv_queue = ?
+        |]
+        (rqId, False)
+  _ -> pure Nothing
+
+getNextSndQueue :: DB.Connection -> SndQueue -> IO (Maybe SndQueue)
+getNextSndQueue db SndQueue {dbNextSndQueueId} = case dbNextSndQueueId of
+  Just sqId ->
+    maybeFirstRow toSndQueue $
+      DB.query
+        db
+        [sql|
+          SELECT q.host, q.port, s.key_hash,
+            q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status,
+            q.snd_queue_action, q.snd_queue_action_ts, q.curr_snd_queue, q.next_snd_queue_id,
+            q.smp_client_version, q.created_at, q.updated_at
+          FROM snd_queues q
+          INNER JOIN servers s ON q.host = s.host AND q.port = s.port
+          WHERE q.snd_queue_id = ? AND q.curr_snd_queue = ?
+        |]
+        (sqId, False)
+  _ -> pure Nothing
+
+dbCreateNextRcvQueue :: DB.Connection -> ConnId -> RcvQueue -> RcvQueue -> IO ()
+dbCreateNextRcvQueue db connId RcvQueue {server = (SMPServer host port _), rcvId} rq' = do
+  rqId <- insertRcvQueue_ db connId rq'
+  DB.execute
+    db
+    [sql|
+      UPDATE rcv_queues
+      SET next_rcv_queue_id = ?
+      WHERE host = ? AND port = ? AND rcv_id = ? AND curr_rcv_queue = ?
+    |]
+    (rqId, host, port, rcvId, True)
+
+dbCreateNextSndQueue :: DB.Connection -> ConnId -> SndQueue -> SndQueue -> IO ()
+dbCreateNextSndQueue db connId SndQueue {server = (SMPServer host port _), sndId} sq' = do
+  sqId <- insertSndQueue_ db connId sq'
+  DB.execute
+    db
+    [sql|
+      UPDATE snd_queues
+      SET next_snd_queue_id = ?
+      WHERE host = ? AND port = ? AND snd_id = ? AND curr_snd_queue = ?
+    |]
+    (sqId, host, port, sndId, True)
+
+setRcvQueueAction :: DB.Connection -> RcvQueue -> Maybe RcvQueueAction -> IO ()
+setRcvQueueAction db RcvQueue {server = (SMPServer host port _), rcvId} rqAction_ = do
+  ts <- getCurrentTime
+  DB.execute
+    db
+    [sql|
+      UPDATE rcv_queues
+      SET rcv_queue_action = ?, rcv_queue_action_ts = ?
+      WHERE host = ? AND port = ? AND rcv_id = ? AND curr_rcv_queue = ?
+    |]
+    (rqAction_, ts, host, port, rcvId, True)
+
+switchCurrRcvQueue :: DB.Connection -> RcvQueue -> RcvQueue -> IO ()
+switchCurrRcvQueue db RcvQueue {server = (SMPServer host port _), rcvId} RcvQueue {dbNextRcvQueueId} = do
+  DB.execute db "DELETE FROM rcv_queues WHERE host = ? AND port = ? AND rcv_id = ? AND curr_rcv_queue = ?" (host, port, rcvId, True)
+  DB.execute db "UPDATE rcv_queues SET curr_rcv_queue = ? WHERE rcv_queue_id = ? AND curr_rcv_queue = ?" (True, dbNextRcvQueueId, False)
+
+switchCurrSndQueue :: DB.Connection -> SndQueue -> IO ()
+switchCurrSndQueue db SndQueue {server = (SMPServer host port _), sndId, dbNextSndQueueId} = do
+  DB.execute db "DELETE FROM snd_queues WHERE host = ? AND port = ? AND snd_id = ? AND curr_snd_queue = ?" (host, port, sndId, True)
+  DB.execute db "UPDATE snd_queues SET curr_snd_queue = ? WHERE snd_queue_id = ? AND curr_snd_queue = ?" (True, dbNextSndQueueId, False)
+
 type SMPConfirmationRow = (SndPublicVerifyKey, C.PublicKeyX25519, ConnInfo, Maybe [SMPQueueInfo], Maybe Version)
 
 smpConfirmation :: SMPConfirmationRow -> SMPConfirmation
@@ -594,21 +695,21 @@ getPendingMsgData db connId msgId = do
       DB.query
         db
         [sql|
-          SELECT m.msg_type, m.msg_flags, m.msg_body, m.internal_ts
+          SELECT m.msg_type, m.msg_flags, m.msg_body, m.internal_ts, s.curr_snd_queue
           FROM messages m
           JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
           WHERE m.conn_id = ? AND m.internal_id = ?
         |]
         (connId, msgId)
-    pendingMsgData :: (AgentMessageType, Maybe MsgFlags, MsgBody, InternalTs) -> PendingMsgData
-    pendingMsgData (msgType, msgFlags_, msgBody, internalTs) =
+    pendingMsgData :: (AgentMessageType, Maybe MsgFlags, MsgBody, InternalTs, Bool) -> PendingMsgData
+    pendingMsgData (msgType, msgFlags_, msgBody, internalTs, currSndQueue) =
       let msgFlags = fromMaybe SMP.noMsgFlags msgFlags_
-       in PendingMsgData {msgId, msgType, msgFlags, msgBody, internalTs}
+       in PendingMsgData {msgId, msgType, msgFlags, msgBody, internalTs, currSndQueue}
 
-getPendingMsgs :: DB.Connection -> ConnId -> IO [InternalId]
-getPendingMsgs db connId =
+getPendingMsgs :: DB.Connection -> ConnId -> Bool -> IO [InternalId]
+getPendingMsgs db connId current =
   map fromOnly
-    <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ?" (Only connId)
+    <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ? AND curr_snd_queue = ?" (connId, current)
 
 setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError SMP.MsgId)
 setMsgUserAck db connId agentMsgId = do
@@ -1105,6 +1206,14 @@ instance ToField (NonEmpty TransportHost) where toField = toField . decodeLatin1
 
 instance FromField (NonEmpty TransportHost) where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
 
+instance ToField RcvQueueAction where toField = toField . textEncode
+
+instance FromField RcvQueueAction where fromField = fromTextField_ textDecode
+
+instance ToField SndQueueAction where toField = toField . textEncode
+
+instance FromField SndQueueAction where fromField = fromTextField_ textDecode
+
 instance ToField (ACommand p) where toField = toField . serializeCommand
 
 instance FromField ACmd where fromField = blobFieldParser dbCommandP
@@ -1180,93 +1289,132 @@ upsertNtfServer_ db ProtocolServer {host, port, keyHash} = do
 
 -- * createRcvConn helpers
 
-insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()
-insertRcvQueue_ dbConn connId RcvQueue {..} = do
+insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO Int64
+insertRcvQueue_ db connId RcvQueue {..} = do
+  qId <- newQueueId_ <$> DB.query_ db "SELECT rcv_queue_id FROM rcv_queues ORDER BY rcv_queue_id DESC LIMIT 1"
   DB.execute
-    dbConn
+    db
     [sql|
       INSERT INTO rcv_queues
-        (host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status, smp_client_version) VALUES (?,?,?,?,?,?,?,?,?,?,?);
+        (rcv_queue_id, host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status, curr_rcv_queue, smp_client_version, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?);
     |]
-    ((host server, port server, rcvId, connId) :. (rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, smpClientVersion))
+    ((qId, host server, port server, rcvId, connId) :. (rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (currRcvQueue, smpClientVersion, createdAt, updatedAt))
+  pure qId
 
 -- * createSndConn helpers
 
-insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()
-insertSndQueue_ dbConn connId SndQueue {..} = do
+insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO Int64
+insertSndQueue_ db connId SndQueue {..} = do
+  qId <- newQueueId_ <$> DB.query_ db "SELECT snd_queue_id FROM snd_queues ORDER BY snd_queue_id DESC LIMIT 1"
   DB.execute
-    dbConn
+    db
     [sql|
       INSERT INTO snd_queues
-        (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, smp_client_version) VALUES (?,?,?,?,?,?,?,?,?,?);
+        (snd_queue_id, host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, curr_snd_queue, smp_client_version, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?);
     |]
-    (host server, port server, sndId, connId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, smpClientVersion)
+    ((qId, host server, port server, sndId, connId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status) :. (currSndQueue, smpClientVersion, createdAt, updatedAt))
+  pure qId
+
+newQueueId_ :: [Only (Maybe Int64)] -> Int64
+newQueueId_ [] = 1
+newQueueId_ (Only maxId_ : _) = maybe 1 (+ 1) maxId_
 
 -- * getConn helpers
 
 getConn :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn)
-getConn dbConn connId =
-  getConnData dbConn connId >>= \case
+getConn db connId =
+  getConnData db connId >>= \case
     Nothing -> pure $ Left SEConnNotFound
-    Just (connData, cMode) -> do
-      rQ <- getRcvQueueByConnId_ dbConn connId
-      sQ <- getSndQueueByConnId_ dbConn connId
-      pure $ case (rQ, sQ, cMode) of
-        (Just rcvQ, Just sndQ, CMInvitation) -> Right $ SomeConn SCDuplex (DuplexConnection connData rcvQ sndQ)
-        (Just rcvQ, Nothing, CMInvitation) -> Right $ SomeConn SCRcv (RcvConnection connData rcvQ)
-        (Nothing, Just sndQ, CMInvitation) -> Right $ SomeConn SCSnd (SndConnection connData sndQ)
-        (Just rcvQ, Nothing, CMContact) -> Right $ SomeConn SCContact (ContactConnection connData rcvQ)
-        (Nothing, Nothing, _) -> Right $ SomeConn SCNew (NewConnection connData)
-        _ -> Left SEConnNotFound
+    Just (cData, cMode) -> do
+      rq_ <- getRcvQueueByConnId_ db connId
+      sq_ <- getSndQueueByConnId_ db connId
+      case (rq_, sq_, cMode) of
+        (Just rq, Just sq, CMInvitation) -> do
+          rq' <- getNextRcvQueue db rq
+          sq' <- getNextSndQueue db sq
+          pure . Right $ SomeConn SCDuplex (DuplexConnection cData rq sq rq' sq')
+        (Just rq, Nothing, CMInvitation) -> pure . Right $ SomeConn SCRcv (RcvConnection cData rq)
+        (Nothing, Just sq, CMInvitation) -> pure . Right $ SomeConn SCSnd (SndConnection cData sq)
+        (Just rq, Nothing, CMContact) -> pure . Right $ SomeConn SCContact (ContactConnection cData rq)
+        (Nothing, Nothing, _) -> pure . Right $ SomeConn SCNew (NewConnection cData)
+        _ -> pure $ Left SEConnNotFound
 
 getConnData :: DB.Connection -> ConnId -> IO (Maybe (ConnData, ConnectionMode))
 getConnData dbConn connId' =
-  connData
-    <$> DB.query dbConn "SELECT conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake FROM connections WHERE conn_id = ?;" (Only connId')
+  maybeFirstRow toConnData $
+    DB.query dbConn "SELECT conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake FROM connections WHERE conn_id = ?;" (Only connId')
   where
-    connData [(connId, cMode, connAgentVersion, enableNtfs_, duplexHandshake)] = Just (ConnData {connId, connAgentVersion, enableNtfs = fromMaybe True enableNtfs_, duplexHandshake}, cMode)
-    connData _ = Nothing
+    toConnData (connId, cMode, connAgentVersion, enableNtfs_, duplexHandshake) = (ConnData {connId, connAgentVersion, enableNtfs = fromMaybe True enableNtfs_, duplexHandshake}, cMode)
+
+type RcvQueueRow =
+  ServerRow
+    :. (SMP.RecipientId, SMP.RcvPrivateSignKey, SMP.RcvDhSecret, C.PrivateKeyX25519, Maybe C.DhSecretX25519, SMP.SenderId, Maybe C.APublicVerifyKey, QueueStatus)
+    :. (Maybe RcvQueueAction, Maybe UTCTime, Bool, Maybe Int64)
+    :. NtfCredsRow
+    :. (Maybe Version, UTCTime, UTCTime)
+
+type ServerRow = (NonEmpty TransportHost, String, C.KeyHash)
+
+type NtfCredsRow = (Maybe SMP.NtfPublicVerifyKey, Maybe SMP.NtfPrivateSignKey, Maybe SMP.NotifierId, Maybe RcvNtfDhSecret)
+
+toRcvQueue :: RcvQueueRow -> RcvQueue
+toRcvQueue (srvRow :. (rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, sndPublicKey, status) :. (rqAction_, rqActionTs_, currRcvQueue, dbNextRcvQueueId) :. ntfCredsRow :. (smpClientVersion_, createdAt, updatedAt)) =
+  let server = toSMPServer srvRow
+      smpClientVersion = fromMaybe 1 smpClientVersion_
+      rcvQueueAction = (,) <$> rqAction_ <*> rqActionTs_
+      clientNtfCreds = toNtfCreds ntfCredsRow
+   in RcvQueue {server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, sndPublicKey, status, rcvQueueAction, currRcvQueue, dbNextRcvQueueId, smpClientVersion, clientNtfCreds, createdAt, updatedAt}
+
+toSMPServer :: ServerRow -> SMPServer
+toSMPServer (host, port, keyHash) = SMPServer host port keyHash
+
+toNtfCreds :: NtfCredsRow -> Maybe ClientNtfCreds
+toNtfCreds (Just ntfPublicKey, Just ntfPrivateKey, Just notifierId, Just rcvNtfDhSecret) = Just $ ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
+toNtfCreds _ = Nothing
 
 getRcvQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe RcvQueue)
 getRcvQueueByConnId_ dbConn connId =
-  listToMaybe . map rcvQueue
-    <$> DB.query
+  maybeFirstRow toRcvQueue $
+    DB.query
       dbConn
       [sql|
-        SELECT s.key_hash, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
-          q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status, q.smp_client_version,
-          q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret
+        SELECT q.host, q.port, s.key_hash,
+          q.rcv_id, q.rcv_private_key, q.rcv_dh_secret, q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.snd_key, q.status,
+          q.rcv_queue_action, q.rcv_queue_action_ts, q.curr_rcv_queue, q.next_rcv_queue_id,
+          q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
+          q.smp_client_version, q.created_at, q.updated_at
         FROM rcv_queues q
         INNER JOIN servers s ON q.host = s.host AND q.port = s.port
-        WHERE q.conn_id = ?;
+        WHERE q.conn_id = ? AND q.curr_rcv_queue = ?
       |]
-      (Only connId)
-  where
-    rcvQueue ((keyHash, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, smpClientVersion_) :. (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_)) =
-      let server = SMPServer host port keyHash
-          smpClientVersion = fromMaybe 1 smpClientVersion_
-          clientNtfCreds = case (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_) of
-            (Just ntfPublicKey, Just ntfPrivateKey, Just notifierId, Just rcvNtfDhSecret) -> Just $ ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
-            _ -> Nothing
-       in RcvQueue {server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, smpClientVersion, clientNtfCreds}
+      (connId, True)
 
 getSndQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe SndQueue)
-getSndQueueByConnId_ dbConn connId =
-  sndQueue
-    <$> DB.query
-      dbConn
+getSndQueueByConnId_ db connId =
+  maybeFirstRow toSndQueue $
+    DB.query
+      db
       [sql|
-        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status, q.smp_client_version
+        SELECT q.host, q.port, s.key_hash,
+          q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status,
+          q.snd_queue_action, q.snd_queue_action_ts, q.curr_snd_queue, q.next_snd_queue_id,
+          q.smp_client_version, q.created_at, q.updated_at
         FROM snd_queues q
         INNER JOIN servers s ON q.host = s.host AND q.port = s.port
-        WHERE q.conn_id = ?;
+        WHERE q.conn_id = ? AND q.curr_snd_queue = ?
       |]
-      (Only connId)
-  where
-    sndQueue [(keyHash, host, port, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, smpClientVersion)] =
-      let server = SMPServer host port keyHash
-       in Just SndQueue {server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, smpClientVersion}
-    sndQueue _ = Nothing
+      (connId, True)
+
+type SndQueueRow =
+  ServerRow
+    :. (SMP.SenderId, Maybe C.APublicVerifyKey, SMP.SndPrivateSignKey, Maybe C.PublicKeyX25519, C.DhSecretX25519, QueueStatus, Maybe SndQueueAction, Maybe UTCTime, Bool, Maybe Int64)
+    :. (Version, UTCTime, UTCTime)
+
+toSndQueue :: SndQueueRow -> SndQueue
+toSndQueue (srvRow :. (sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, sqAction_, sqActionTs_, currSndQueue, dbNextSndQueueId) :. (smpClientVersion, createdAt, updatedAt)) =
+  let server = toSMPServer srvRow
+      sndQueueAction = (,) <$> sqAction_ <*> sqActionTs_
+   in SndQueue {server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, sndQueueAction, currSndQueue, dbNextSndQueueId, smpClientVersion, createdAt, updatedAt}
 
 -- * updateRcvIds helpers
 
@@ -1421,15 +1569,16 @@ insertSndMsgDetails_ dbConn connId SndMsgData {..} =
     dbConn
     [sql|
       INSERT INTO snd_messages
-        ( conn_id, internal_snd_id, internal_id, internal_hash, previous_msg_hash)
+        ( conn_id, internal_snd_id, internal_id, internal_hash, previous_msg_hash, curr_snd_queue)
       VALUES
-        (:conn_id,:internal_snd_id,:internal_id,:internal_hash,:previous_msg_hash);
+        (:conn_id,:internal_snd_id,:internal_id,:internal_hash,:previous_msg_hash,:curr_snd_queue);
     |]
     [ ":conn_id" := connId,
       ":internal_snd_id" := internalSndId,
       ":internal_id" := internalId,
       ":internal_hash" := internalHash,
-      ":previous_msg_hash" := prevMsgHash
+      ":previous_msg_hash" := prevMsgHash,
+      ":curr_snd_queue" := currSndQueue
     ]
 
 updateHashSnd_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()

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

@@ -36,6 +36,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220608_v2
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220625_v2_ntf_mode
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220811_onion_hosts
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220822_queue_rotation
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -52,6 +53,7 @@ schemaMigrations =
     ("m20220625_v2_ntf_mode", m20220625_v2_ntf_mode),
     ("m20220811_onion_hosts", m20220811_onion_hosts),
     ("m20220817_connection_ntfs", m20220817_connection_ntfs),
+    ("m20220822_queue_rotation", m20220822_queue_rotation),
     ("m20220905_commands", m20220905_commands)
   ]
 

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

@@ -0,0 +1,61 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220822_queue_rotation where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220822_queue_rotation :: Query
+m20220822_queue_rotation =
+  [sql|
+-- * rcv_queues
+
+ALTER TABLE rcv_queues ADD COLUMN rcv_queue_id INTEGER NULL;
+ALTER TABLE rcv_queues ADD COLUMN rcv_queue_action TEXT NULL;
+ALTER TABLE rcv_queues ADD COLUMN rcv_queue_action_ts TEXT NULL;
+
+ALTER TABLE rcv_queues ADD COLUMN curr_rcv_queue INTEGER CHECK (curr_rcv_queue NOT NULL);
+UPDATE rcv_queues SET curr_rcv_queue = 1;
+
+ALTER TABLE rcv_queues ADD COLUMN next_rcv_queue_id INTEGER NULL;
+  -- REFERENCES rcv_queues (rcv_queue_id) ON DELETE SET NULL;
+-- next_rcv_queue = 1: this is the new queue the connection is switching to
+-- next_rcv_queue_id: the ID of the next queue
+
+ALTER TABLE rcv_queues ADD COLUMN created_at TEXT CHECK (created_at NOT NULL);
+UPDATE rcv_queues SET created_at = '1970-01-01 00:00:00';
+
+ALTER TABLE rcv_queues ADD COLUMN updated_at TEXT CHECK (updated_at NOT NULL);
+UPDATE rcv_queues SET updated_at = '1970-01-01 00:00:00';
+
+CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(rcv_queue_id);
+CREATE UNIQUE INDEX idx_next_rcv_queue_id ON rcv_queues(next_rcv_queue_id);
+
+-- * snd_queues
+
+ALTER TABLE snd_queues ADD COLUMN snd_queue_id INTEGER NULL;
+ALTER TABLE snd_queues ADD COLUMN snd_queue_action TEXT NULL;
+ALTER TABLE snd_queues ADD COLUMN snd_queue_action_ts TEXT NULL;
+
+ALTER TABLE snd_queues ADD COLUMN curr_snd_queue INTEGER CHECK (curr_snd_queue NOT NULL);
+UPDATE snd_queues SET curr_snd_queue = 1;
+
+ALTER TABLE snd_queues ADD COLUMN next_snd_queue_id INTEGER NULL;
+  -- REFERENCES snd_queues (snd_queue_id) ON DELETE SET NULL;
+-- next_snd_queue = 1: this is the new queue the connection is switching to
+-- next_snd_queue_id: the ID of the next queue
+
+ALTER TABLE snd_queues ADD COLUMN created_at TEXT CHECK (created_at NOT NULL);
+UPDATE snd_queues SET created_at = '1970-01-01 00:00:00';
+
+ALTER TABLE snd_queues ADD COLUMN updated_at TEXT CHECK (updated_at NOT NULL);
+UPDATE snd_queues SET updated_at = '1970-01-01 00:00:00';
+
+CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(snd_queue_id);
+CREATE UNIQUE INDEX idx_next_snd_queue_id ON snd_queues(next_snd_queue_id);
+
+-- * 
+
+ALTER TABLE snd_messages ADD COLUMN curr_snd_queue INTEGER CHECK (curr_snd_queue NOT NULL);
+UPDATE snd_messages SET curr_snd_queue = 1;
+|]

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

@@ -41,6 +41,13 @@ CREATE TABLE rcv_queues(
   ntf_private_key BLOB,
   ntf_id BLOB,
   rcv_ntf_dh_secret BLOB,
+  rcv_queue_id INTEGER NULL,
+  rcv_queue_action TEXT NULL,
+  rcv_queue_action_ts TEXT NULL,
+  curr_rcv_queue INTEGER CHECK(curr_rcv_queue NOT NULL),
+  next_rcv_queue_id INTEGER NULL,
+  created_at TEXT CHECK(created_at NOT NULL),
+  updated_at TEXT CHECK(updated_at NOT NULL),
   PRIMARY KEY(host, port, rcv_id),
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE,
@@ -58,6 +65,13 @@ CREATE TABLE snd_queues(
   smp_client_version INTEGER NOT NULL DEFAULT 1,
   snd_public_key BLOB,
   e2e_pub_key BLOB,
+  snd_queue_id INTEGER NULL,
+  snd_queue_action TEXT NULL,
+  snd_queue_action_ts TEXT NULL,
+  curr_snd_queue INTEGER CHECK(curr_snd_queue NOT NULL),
+  next_snd_queue_id INTEGER NULL,
+  created_at TEXT CHECK(created_at NOT NULL),
+  updated_at TEXT CHECK(updated_at NOT NULL),
   PRIMARY KEY(host, port, snd_id),
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE
@@ -99,6 +113,7 @@ CREATE TABLE snd_messages(
   internal_id INTEGER NOT NULL,
   internal_hash BLOB NOT NULL,
   previous_msg_hash BLOB NOT NULL DEFAULT x'',
+  curr_snd_queue INTEGER CHECK(curr_snd_queue NOT NULL),
   PRIMARY KEY(conn_id, internal_snd_id),
   FOREIGN KEY(conn_id, internal_id) REFERENCES messages
   ON DELETE CASCADE
@@ -194,6 +209,10 @@ CREATE TABLE ntf_subscriptions(
   FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
   ON DELETE RESTRICT ON UPDATE CASCADE
 ) WITHOUT ROWID;
+CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(rcv_queue_id);
+CREATE UNIQUE INDEX idx_next_rcv_queue_id ON rcv_queues(next_rcv_queue_id);
+CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(snd_queue_id);
+CREATE UNIQUE INDEX idx_next_snd_queue_id ON snd_queues(next_snd_queue_id);
 CREATE TABLE commands(
   command_id INTEGER PRIMARY KEY,
   conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,

src/Simplex/Messaging/Client.hs

@@ -76,6 +76,7 @@ import Data.List (find)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Maybe (fromMaybe)
+import Data.Word (Word16)
 import GHC.Generics (Generic)
 import Network.Socket (ServiceName)
 import Numeric.Natural
@@ -374,8 +375,11 @@ subscribeSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT Pr
 subscribeSMPQueue c rpKey rId =
   sendSMPCommand c (Just rpKey) rId SUB >>= \case
     OK -> return ()
-    cmd@MSG {} -> liftIO $ writeSMPMessage c rId cmd
+    cmd@MSG {} -> deliver cmd
+    cmd@LEN {} -> deliver cmd
     r -> throwE . PCEUnexpectedResponse $ bshow r
+  where
+    deliver = liftIO . writeSMPMessage c rId
 
 -- | Subscribe to multiple SMP queues batching commands if supported.
 subscribeSMPQueues :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> IO (NonEmpty (Either ProtocolClientError ()))
@@ -384,9 +388,12 @@ subscribeSMPQueues c qs = sendProtocolCommands c cs >>= mapM response . L.zip qs
     cs = L.map (\(rpKey, rId) -> (Just rpKey, rId, Cmd SRecipient SUB)) qs
     response ((_, rId), r) = case r of
       Right OK -> pure $ Right ()
-      Right cmd@MSG {} -> writeSMPMessage c rId cmd $> Right ()
+      Right cmd@MSG {} -> deliver cmd
+      Right cmd@LEN {} -> deliver cmd
       Right r' -> pure . Left . PCEUnexpectedResponse $ bshow r'
       Left e -> pure $ Left e
+      where
+        deliver cmd = writeSMPMessage c rId cmd $> Right ()
 
 writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO ()
 writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ c)
@@ -402,8 +409,11 @@ getSMPMessage :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT Protoc
 getSMPMessage c rpKey rId =
   sendSMPCommand c (Just rpKey) rId GET >>= \case
     OK -> pure Nothing
-    cmd@(MSG msg) -> liftIO (writeSMPMessage c rId cmd) $> Just msg
+    cmd@(MSG msg) -> deliver cmd $> Just msg
+    cmd@LEN {} -> deliver cmd $> Nothing
     r -> throwE . PCEUnexpectedResponse $ bshow r
+  where
+    deliver = liftIO . writeSMPMessage c rId
 
 -- | Subscribe to the SMP queue notifications.
 --
@@ -468,15 +478,21 @@ ackSMPMessage :: SMPClient -> RcvPrivateSignKey -> QueueId -> MsgId -> ExceptT P
 ackSMPMessage c rpKey rId msgId =
   sendSMPCommand c (Just rpKey) rId (ACK msgId) >>= \case
     OK -> return ()
-    cmd@MSG {} -> liftIO $ writeSMPMessage c rId cmd
+    cmd@MSG {} -> deliver cmd
+    cmd@LEN {} -> deliver cmd
     r -> throwE . PCEUnexpectedResponse $ bshow r
+  where
+    deliver = liftIO . writeSMPMessage c rId
 
 -- | Irreversibly suspend SMP queue.
 -- The existing messages from the queue will still be delivered.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#suspend-queue
-suspendSMPQueue :: SMPClient -> RcvPrivateSignKey -> QueueId -> ExceptT ProtocolClientError IO ()
-suspendSMPQueue = okSMPCommand OFF
+suspendSMPQueue :: SMPClient -> RcvPrivateSignKey -> QueueId -> ExceptT ProtocolClientError IO Word16
+suspendSMPQueue c pKey qId =
+  sendSMPCommand c (Just pKey) qId OFF >>= \case
+    LEN len -> return len
+    r -> throwE . PCEUnexpectedResponse $ bshow r
 
 -- | Irreversibly delete SMP queue and all messages in it.
 --

src/Simplex/Messaging/Protocol.hs

@@ -141,6 +141,7 @@ import Data.Maybe (isJust, isNothing)
 import Data.String
 import Data.Time.Clock.System (SystemTime (..))
 import Data.Type.Equality
+import Data.Word (Word16)
 import GHC.Generics (Generic)
 import GHC.TypeLits (type (+))
 import Generic.Random (genericArbitraryU)
@@ -282,6 +283,7 @@ data BrokerMsg where
   NID :: NotifierId -> RcvNtfPublicDhKey -> BrokerMsg
   NMSG :: C.CbNonce -> EncNMsgMeta -> BrokerMsg
   END :: BrokerMsg
+  LEN :: Word16 -> BrokerMsg
   OK :: BrokerMsg
   ERR :: ErrorType -> BrokerMsg
   PONG :: BrokerMsg
@@ -436,6 +438,7 @@ data BrokerMsgTag
   | NID_
   | NMSG_
   | END_
+  | LEN_
   | OK_
   | ERR_
   | PONG_
@@ -495,6 +498,7 @@ instance Encoding BrokerMsgTag where
     NID_ -> "NID"
     NMSG_ -> "NMSG"
     END_ -> "END"
+    LEN_ -> "LEN"
     OK_ -> "OK"
     ERR_ -> "ERR"
     PONG_ -> "PONG"
@@ -507,6 +511,7 @@ instance ProtocolMsgTag BrokerMsgTag where
     "NID" -> Just NID_
     "NMSG" -> Just NMSG_
     "END" -> Just END_
+    "LEN" -> Just LEN_
     "OK" -> Just OK_
     "ERR" -> Just ERR_
     "PONG" -> Just PONG_
@@ -945,6 +950,7 @@ instance ProtocolEncoding BrokerMsg where
     NID nId srvNtfDh -> e (NID_, ' ', nId, srvNtfDh)
     NMSG nmsgNonce encNMsgMeta -> e (NMSG_, ' ', nmsgNonce, encNMsgMeta)
     END -> e END_
+    LEN len -> e (LEN_, ' ', len)
     OK -> e OK_
     ERR err -> e (ERR_, ' ', err)
     PONG -> e PONG_
@@ -965,6 +971,7 @@ instance ProtocolEncoding BrokerMsg where
     NID_ -> NID <$> _smpP <*> smpP
     NMSG_ -> NMSG <$> _smpP <*> smpP
     END_ -> pure END
+    LEN_ -> LEN <$> _smpP
     OK_ -> pure OK
     ERR_ -> ERR <$> _smpP
     PONG_ -> pure PONG

src/Simplex/Messaging/Server.hs

@@ -429,8 +429,12 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
 
         suspendQueue_ :: QueueStore -> m (Transmission BrokerMsg)
         suspendQueue_ st = do
-          withLog (`logDeleteQueue` queueId)
-          okResp <$> atomically (suspendQueue st queueId)
+          withLog (`logSuspendQueue` queueId)
+          ms <- asks msgStore
+          quota <- asks $ msgQueueQuota . config
+          let queueLen _ = fmap fromIntegral . queueLength =<< getMsgQueue ms queueId quota
+          len <- atomically (suspendQueue st queueId >>= mapM queueLen)
+          pure (corrId, queueId, either ERR LEN len)
 
         subscribeQueue :: QueueRec -> RecipientId -> m (Transmission BrokerMsg)
         subscribeQueue qr rId =
@@ -586,7 +590,9 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
                 Just msg ->
                   let encMsg = encryptMsg qr msg
                    in atomically (setDelivered s msg) $> (corrId, rId, MSG encMsg)
-                _ -> forkSub $> ok
+                _
+                  | status qr == QueueActive -> forkSub $> ok
+                  | otherwise -> pure (corrId, rId, LEN 0)
             _ -> pure ok
           where
             forkSub :: m ()
@@ -607,13 +613,13 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
 
         encryptMsg :: QueueRec -> Message -> RcvMessage
         encryptMsg qr Message {msgId, msgTs, msgFlags, msgBody}
-          | thVersion == 1 || thVersion == 2 = encrypt msgBody
-          | otherwise = encrypt $ encodeRcvMsgBody RcvMsgBody {msgTs, msgFlags, msgBody}
+          | thVersion == 1 || thVersion == 2 = encrypt msgTs msgFlags msgBody
+          | otherwise = encrypt (MkSystemTime 0 0) noMsgFlags $ encodeRcvMsgBody RcvMsgBody {msgTs, msgFlags, msgBody}
           where
-            encrypt :: KnownNat i => C.MaxLenBS i -> RcvMessage
-            encrypt body =
+            encrypt :: KnownNat i => SystemTime -> MsgFlags -> C.MaxLenBS i -> RcvMessage
+            encrypt msgTs' msgFlags' body =
               let encBody = EncRcvMsgBody $ C.cbEncryptMaxLenBS (rcvDhSecret qr) (C.cbNonce msgId) body
-               in RcvMessage msgId msgTs msgFlags encBody
+               in RcvMessage msgId msgTs' msgFlags' encBody
 
         setDelivered :: Sub -> Message -> STM Bool
         setDelivered s Message {msgId} = tryPutTMVar (delivered s) msgId

src/Simplex/Messaging/Server/MsgStore.hs

@@ -25,6 +25,7 @@ class MonadMsgStore s q m | s -> q where
 
 class MonadMsgQueue q m where
   isFull :: q -> m Bool
+  queueLength :: q -> m Natural
   writeMsg :: q -> Message -> m () -- non blocking
   tryPeekMsg :: q -> m (Maybe Message) -- non blocking
   peekMsg :: q -> m Message -- blocking

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

@@ -9,7 +9,7 @@
 
 module Simplex.Messaging.Server.MsgStore.STM where
 
-import Control.Concurrent.STM.TBQueue (flushTBQueue)
+import Control.Concurrent.STM.TBQueue (flushTBQueue, lengthTBQueue)
 import Control.Monad (when)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
@@ -48,6 +48,9 @@ instance MonadMsgQueue MsgQueue STM where
   isFull :: MsgQueue -> STM Bool
   isFull = isFullTBQueue . msgQueue
 
+  queueLength :: MsgQueue -> STM Natural
+  queueLength = lengthTBQueue . msgQueue
+
   writeMsg :: MsgQueue -> Message -> STM ()
   writeMsg = writeTBQueue . msgQueue
 

src/Simplex/Messaging/Server/QueueStore.hs

@@ -15,7 +15,7 @@ data QueueRec = QueueRec
     senderId :: SenderId,
     senderKey :: Maybe SndPublicVerifyKey,
     notifier :: Maybe NtfCreds,
-    status :: QueueStatus
+    status :: ServerQueueStatus
   }
   deriving (Eq, Show)
 
@@ -32,7 +32,7 @@ instance StrEncoding NtfCreds where
     (notifierId, notifierKey, rcvNtfDhSecret) <- strP
     pure NtfCreds {notifierId, notifierKey, rcvNtfDhSecret}
 
-data QueueStatus = QueueActive | QueueOff deriving (Eq, Show)
+data ServerQueueStatus = QueueActive | QueueOff deriving (Eq, Show)
 
 class MonadQueueStore s m where
   addQueue :: s -> QueueRec -> m (Either ErrorType ())

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

@@ -58,8 +58,10 @@ instance MonadQueueStore QueueStore STM where
   secureQueue QueueStore {queues} rId sKey =
     withQueue rId queues $ \qVar ->
       readTVar qVar >>= \q -> case senderKey q of
-        Just _ -> pure Nothing
-        _ -> writeTVar qVar q {senderKey = Just sKey} $> Just q
+        Just k -> pure $ if sKey == k then Just q else Nothing
+        _ ->
+          let q' = q {senderKey = Just sKey}
+           in writeTVar qVar q' $> Just q'
 
   addQueueNotifier :: QueueStore -> RecipientId -> NtfCreds -> STM (Either ErrorType QueueRec)
   addQueueNotifier QueueStore {queues, notifiers} rId ntfCreds@NtfCreds {notifierId = nId} = do

src/Simplex/Messaging/Server/StoreLog.hs

@@ -17,6 +17,7 @@ module Simplex.Messaging.Server.StoreLog
     logCreateQueue,
     logSecureQueue,
     logAddNotifier,
+    logSuspendQueue,
     logDeleteQueue,
     logDeleteNotifier,
     readWriteStoreLog,
@@ -36,7 +37,7 @@ import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), QueueStatus (..))
+import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), ServerQueueStatus (..))
 import Simplex.Messaging.Transport (trimCR)
 import System.Directory (doesFileExist)
 import System.IO
@@ -51,6 +52,7 @@ data StoreLogRecord
   = CreateQueue QueueRec
   | SecureQueue QueueId SndPublicVerifyKey
   | AddNotifier QueueId NtfCreds
+  | SuspendQueue QueueId
   | DeleteQueue QueueId
   | DeleteNotifier QueueId
 
@@ -81,6 +83,7 @@ instance StrEncoding StoreLogRecord where
     CreateQueue q -> strEncode (Str "CREATE", q)
     SecureQueue rId sKey -> strEncode (Str "SECURE", rId, sKey)
     AddNotifier rId ntfCreds -> strEncode (Str "NOTIFIER", rId, ntfCreds)
+    SuspendQueue rId -> strEncode (Str "SUSPEND", rId)
     DeleteQueue rId -> strEncode (Str "DELETE", rId)
     DeleteNotifier rId -> strEncode (Str "NDELETE", rId)
 
@@ -88,6 +91,7 @@ instance StrEncoding StoreLogRecord where
     "CREATE " *> (CreateQueue <$> strP)
       <|> "SECURE " *> (SecureQueue <$> strP_ <*> strP)
       <|> "NOTIFIER " *> (AddNotifier <$> strP_ <*> strP)
+      <|> "SUSPEND " *> (SuspendQueue <$> strP)
       <|> "DELETE " *> (DeleteQueue <$> strP)
       <|> "NDELETE " *> (DeleteNotifier <$> strP)
 
@@ -126,6 +130,9 @@ logSecureQueue s qId sKey = writeStoreLogRecord s $ SecureQueue qId sKey
 logAddNotifier :: StoreLog 'WriteMode -> QueueId -> NtfCreds -> IO ()
 logAddNotifier s qId ntfCreds = writeStoreLogRecord s $ AddNotifier qId ntfCreds
 
+logSuspendQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
+logSuspendQueue s = writeStoreLogRecord s . SuspendQueue
+
 logDeleteQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
 logDeleteQueue s = writeStoreLogRecord s . DeleteQueue
 
@@ -161,6 +168,7 @@ readQueues (ReadStoreLog _ h) = LB.hGetContents h >>= returnResult . procStoreLo
       CreateQueue q -> M.insert (recipientId q) q m
       SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m
       AddNotifier qId ntfCreds -> M.adjust (\q -> q {notifier = Just ntfCreds}) qId m
+      SuspendQueue qId -> M.adjust (\q -> q {status = QueueOff}) qId m
       DeleteQueue qId -> M.delete qId m
       DeleteNotifier qId -> M.adjust (\q -> q {notifier = Nothing}) qId m
     printError :: LogParsingError -> IO ()

tests/AgentTests.hs

@@ -331,7 +331,7 @@ testServerConnectionAfterError t _ = do
   withAgent1 $ \bob -> do
     withAgent2 $ \alice -> do
       bob #: ("1", "alice", "SUB") #> ("1", "alice", ERR (BROKER NETWORK))
-      alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER NETWORK))
+      alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER TIMEOUT))
       withServer $ do
         alice <#= \case ("", "bob", SENT 4) -> True; ("", "", UP s ["bob"]) -> s == server; _ -> False
         alice <#= \case ("", "bob", SENT 4) -> True; ("", "", UP s ["bob"]) -> s == server; _ -> False

tests/AgentTests/FunctionalAPITests.hs

@@ -26,7 +26,7 @@ import qualified Data.Map as M
 import qualified Data.Set as S
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import SMPAgentClient
-import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
+import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..))
 import Simplex.Messaging.Agent.Protocol
@@ -114,6 +114,9 @@ functionalAPITests t = do
   describe "Batching SMP commands" $ do
     it "should subscribe to multiple subscriptions with batching" $
       testBatchedSubscriptions t
+  describe "Switching receive queues" $
+    xit "should switch to a new queue" $
+      testSwitchRcvQueue t
   describe "Async agent commands" $ do
     it "should connect using async agent commands" $
       withSmpServer t testAsyncCommands
@@ -194,8 +197,8 @@ runAgentClientTest alice bob baseId = do
     get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
     ackMessage alice bobId $ baseId + 3
     get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
+    _ <- suspendConnection alice bobId
     ackMessage alice bobId $ baseId + 4
-    suspendConnection alice bobId
     5 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2"
     get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
     deleteConnection alice bobId
@@ -233,7 +236,7 @@ runAgentClientContactTest alice bob baseId = do
     ackMessage alice bobId $ baseId + 3
     get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
     ackMessage alice bobId $ baseId + 4
-    suspendConnection alice bobId
+    _ <- suspendConnection alice bobId
     5 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 2"
     get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
     deleteConnection alice bobId
@@ -565,6 +568,22 @@ testBatchedSubscriptions t = do
         killThread t1
         pure res
 
+testSwitchRcvQueue :: ATransport -> IO ()
+testSwitchRcvQueue t = do
+  a <- getSMPAgentClient agentCfg initAgentServers2
+  b <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers2
+  Right () <- withSmpServer t . withSmpServerOn t testPort2 . runExceptT $ do
+    (aId, bId) <- makeConnection a b
+    exchangeGreetings a bId b aId
+    switchConnection a bId
+    r1 <- get b
+    liftIO $ print r1
+    -- r2 <- get b
+    -- liftIO $ print r2
+    r3 <- get a
+    liftIO $ print r3
+  pure ()
+
 testAsyncCommands :: IO ()
 testAsyncCommands = do
   alice <- getSMPAgentClient agentCfg initAgentServers

tests/AgentTests/SQLiteTests.hs

@@ -27,6 +27,7 @@ import Simplex.Messaging.Agent.Store.SQLite
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Protocol as SMP
+import System.IO.Unsafe (unsafePerformIO)
 import System.Random
 import Test.Hspec
 import UnliftIO.Directory (removeFile)
@@ -150,6 +151,9 @@ testPrivDhKey = "MC4CAQAwBQYDK2VuBCIEINCzbVFaCiYHoYncxNY8tSIfn0pXcIAhLBfFc0m+gOp
 testDhSecret :: C.DhSecretX25519
 testDhSecret = "01234567890123456789012345678901"
 
+currentTime :: UTCTime
+currentTime = unsafePerformIO getCurrentTime
+
 rcvQueue1 :: RcvQueue
 rcvQueue1 =
   RcvQueue
@@ -159,10 +163,16 @@ rcvQueue1 =
       rcvDhSecret = testDhSecret,
       e2ePrivKey = testPrivDhKey,
       e2eDhSecret = Nothing,
-      sndId = Just "2345",
+      sndId = "2345",
+      sndPublicKey = Nothing,
       status = New,
+      currRcvQueue = True,
+      dbNextRcvQueueId = Nothing,
+      rcvQueueAction = Nothing,
+      clientNtfCreds = Nothing,
       smpClientVersion = 1,
-      clientNtfCreds = Nothing
+      createdAt = currentTime,
+      updatedAt = currentTime
     }
 
 sndQueue1 :: SndQueue
@@ -175,7 +185,12 @@ sndQueue1 =
       e2ePubKey = Nothing,
       e2eDhSecret = testDhSecret,
       status = New,
-      smpClientVersion = 1
+      currSndQueue = True,
+      dbNextSndQueueId = Nothing,
+      sndQueueAction = Nothing,
+      smpClientVersion = 1,
+      createdAt = currentTime,
+      updatedAt = currentTime
     }
 
 testCreateRcvConn :: SpecWith SQLiteStore
@@ -189,7 +204,7 @@ testCreateRcvConn =
     upgradeRcvConnToDuplex db "conn1" sndQueue1
       `shouldReturn` Right ()
     getConn db "conn1"
-      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1))
+      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1 Nothing Nothing))
 
 testCreateRcvConnRandomId :: SpecWith SQLiteStore
 testCreateRcvConnRandomId =
@@ -201,7 +216,7 @@ testCreateRcvConnRandomId =
     upgradeRcvConnToDuplex db connId sndQueue1
       `shouldReturn` Right ()
     getConn db connId
-      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1))
+      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1 Nothing Nothing))
 
 testCreateRcvConnDuplicate :: SpecWith SQLiteStore
 testCreateRcvConnDuplicate =
@@ -222,7 +237,7 @@ testCreateSndConn =
     upgradeSndConnToDuplex db "conn1" rcvQueue1
       `shouldReturn` Right ()
     getConn db "conn1"
-      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1))
+      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1 Nothing Nothing))
 
 testCreateSndConnRandomID :: SpecWith SQLiteStore
 testCreateSndConnRandomID =
@@ -234,7 +249,7 @@ testCreateSndConnRandomID =
     upgradeSndConnToDuplex db connId rcvQueue1
       `shouldReturn` Right ()
     getConn db connId
-      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1))
+      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1 Nothing Nothing))
 
 testCreateSndConnDuplicate :: SpecWith SQLiteStore
 testCreateSndConnDuplicate =
@@ -252,7 +267,7 @@ testGetRcvConn =
     g <- newTVarIO =<< drgNew
     _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
     getRcvConn db smpServer recipientId
-      `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1))
+      `shouldReturn` Right (rcvQueue1, SomeConn SCRcv (RcvConnection cData1 rcvQueue1))
 
 testDeleteRcvConn :: SpecWith SQLiteStore
 testDeleteRcvConn =
@@ -287,7 +302,7 @@ testDeleteDuplexConn =
     _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
     _ <- upgradeRcvConnToDuplex db "conn1" sndQueue1
     getConn db "conn1"
-      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1))
+      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1 Nothing Nothing))
     deleteConn db "conn1"
       `shouldReturn` ()
     -- TODO check queues are deleted as well
@@ -308,7 +323,12 @@ testUpgradeRcvConnToDuplex =
               e2ePubKey = Nothing,
               e2eDhSecret = testDhSecret,
               status = New,
-              smpClientVersion = 1
+              currSndQueue = True,
+              dbNextSndQueueId = Nothing,
+              sndQueueAction = Nothing,
+              smpClientVersion = 1,
+              createdAt = currentTime,
+              updatedAt = currentTime
             }
     upgradeRcvConnToDuplex db "conn1" anotherSndQueue
       `shouldReturn` Left (SEBadConnType CSnd)
@@ -329,10 +349,16 @@ testUpgradeSndConnToDuplex =
               rcvDhSecret = testDhSecret,
               e2ePrivKey = testPrivDhKey,
               e2eDhSecret = Nothing,
-              sndId = Just "4567",
+              sndId = "4567",
+              sndPublicKey = Nothing,
               status = New,
+              currRcvQueue = False,
+              dbNextRcvQueueId = Nothing,
+              rcvQueueAction = Nothing,
+              clientNtfCreds = Nothing,
               smpClientVersion = 1,
-              clientNtfCreds = Nothing
+              createdAt = currentTime,
+              updatedAt = currentTime
             }
     upgradeSndConnToDuplex db "conn1" anotherRcvQueue
       `shouldReturn` Left (SEBadConnType CRcv)
@@ -371,15 +397,15 @@ testSetQueueStatusDuplex =
     _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation
     _ <- upgradeRcvConnToDuplex db "conn1" sndQueue1
     getConn db "conn1"
-      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1))
+      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1 Nothing Nothing))
     setRcvQueueStatus db rcvQueue1 Secured
       `shouldReturn` ()
     getConn db "conn1"
-      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1))
+      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1 Nothing Nothing))
     setSndQueueStatus db sndQueue1 Confirmed
       `shouldReturn` ()
     getConn db "conn1"
-      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1 {status = Confirmed}))
+      `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1 {status = Confirmed} Nothing Nothing))
 
 hw :: ByteString
 hw = encodeUtf8 "Hello world!"
@@ -434,7 +460,8 @@ mkSndMsgData internalId internalSndId internalHash =
       msgFlags = SMP.noMsgFlags,
       msgBody = hw,
       internalHash,
-      prevMsgHash = internalHash
+      prevMsgHash = internalHash,
+      currSndQueue = True
     }
 
 testCreateSndMsg_ :: DB.Connection -> PrevSndMsgHash -> ConnId -> SndMsgData -> Expectation

tests/ServerTests.hs

@@ -146,8 +146,10 @@ testCreateSecureV2 _ =
       (ok2, OK) #== "secures queue"
       (rId2, rId) #== "same queue ID in response 3"
 
-      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub)
-      (err4, ERR AUTH) #== "rejects KEY if already secured"
+      Resp "abcd" _ OK <- signSendRecv h rKey ("abcd", rId, KEY sPub)
+      (sPub', _) <- C.generateSignatureKeyPair C.SEd448
+      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub')
+      (err4, ERR AUTH) #== "rejects if secured with different key"
 
       Resp "bcda" _ ok3 <- signSendRecv h sKey ("bcda", sId, _SEND "hello again")
       (ok3, OK) #== "accepts signed SEND"
@@ -208,8 +210,10 @@ testCreateSecure (ATransport t) =
       (ok2, OK) #== "secures queue"
       (rId2, rId) #== "same queue ID in response 3"
 
-      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub)
-      (err4, ERR AUTH) #== "rejects KEY if already secured"
+      Resp "abcd" _ OK <- signSendRecv h rKey ("abcd", rId, KEY sPub)
+      (sPub', _) <- C.generateSignatureKeyPair C.SEd448
+      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub')
+      (err4, ERR AUTH) #== "rejects if secured with different key"
 
       Resp "bcda" _ ok3 <- signSendRecv h sKey ("bcda", sId, _SEND "hello again")
       (ok3, OK) #== "accepts signed SEND"
@@ -261,8 +265,7 @@ testCreateDelete (ATransport t) =
       Resp "bcda" _ err2 <- signSendRecv rh rKey ("bcda", sId, OFF)
       (err2, ERR AUTH) #== "rejects OFF with sender's ID"
 
-      Resp "cdab" rId2 ok3 <- signSendRecv rh rKey ("cdab", rId, OFF)
-      (ok3, OK) #== "suspends queue"
+      Resp "cdab" rId2 (LEN 2) <- signSendRecv rh rKey ("cdab", rId, OFF)
       (rId2, rId) #== "same queue ID in response 2"
 
       Resp "dabc" _ err3 <- signSendRecv sh sKey ("dabc", sId, _SEND "hello")
@@ -271,12 +274,16 @@ testCreateDelete (ATransport t) =
       Resp "abcd" _ err4 <- sendRecv sh ("", "abcd", sId, _SEND "hello")
       (err4, ERR AUTH) #== "reject unsigned SEND too"
 
-      Resp "bcda" _ ok4 <- signSendRecv rh rKey ("bcda", rId, OFF)
-      (ok4, OK) #== "accepts OFF when suspended"
+      Resp "bcda" _ (LEN _) <- signSendRecv rh rKey ("bcda", rId, OFF)
 
       Resp "cdab" _ (Msg mId2 msg2) <- signSendRecv rh rKey ("cdab", rId, SUB)
       (dec mId2 msg2, Right "hello") #== "accepts SUB when suspended and delivers the message again (because was not ACKed)"
 
+      Resp "abcd" _ (Msg mId3 msg3) <- signSendRecv rh rKey ("abcd", rId, ACK mId2)
+      (dec mId3 msg3, Right "hello 2") #== "deliver the next message on ACK"
+
+      Resp "bcda" _ (LEN 0) <- signSendRecv rh rKey ("bcda", rId, ACK mId3)
+
       Resp "dabc" _ err5 <- sendRecv rh (sampleSig, "dabc", rId, DEL)
       (err5, ERR AUTH) #== "rejects DEL with wrong signature"