agent: fast queue rotation not requiring the current server to be online (#1847)

* agent: fast queue rotation does not requiring the current server to be online

* update plan

* update plan and rfc

* corrections

* update plan

* implementation

* test

* refactor

* test switching notificaitons

* rename

* fix, refactor

* fix possible crash

* simplify

* refactor

* simplify

* refactor

* simplify

* order, avoid re-reading connection

---------

Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>

plans/2026-08-09-fast-queue-rotation-plan.md

@@ -0,0 +1,187 @@
+# Fast queue rotation — implementation plan
+
+Branch: ep/drop-agent-versions. RFC: ../rfcs/2026-08-09-fast-queue-rotation.md.
+
+Model: redundant delivery, no flip. `QADD` adds the new receive queue R'. From `QADD` until R' is
+secured the sender writes every message to both old and R' (double delivery, not a move); once R' is
+secured the sender writes new messages to R' only, while old delivers its already-scheduled tail and
+the `QEND` appended to it. `QEND` removes a named queue. The recipient drops duplicates (double
+ratchet), so the order and which queue delivers do not matter, as long as every message arrives on at
+least one queue. Rotation away from a dead server works because every message up to securing is
+scheduled on R' too.
+
+Roles: A initiates (its receive queue rotates; A receives on R'). B sends to A and secures R'.
+
+## Why redundant delivery removes the hard parts
+
+- No boundary, no drain, no last-message id. A never decides how much of old to read.
+- A dead old server loses nothing B still holds: every undelivered message is scheduled on R' as well.
+- A dead new server does not suspend delivery: old keeps delivering until R' is secured.
+- The double ratchet already drops duplicates (`AGENT A_DUPLICATE`) and tolerates bounded reordering,
+  and the delivery schema already writes one message to several send queues (`enqueueMessageB` +
+  `enqueueSavedMessageB`).
+
+## The one ordering constraint
+
+A must hold R''s secret before it reads any R' data message. A data message reaching R' before the
+confirmation is dropped as "no keys" (`processClientMsg`, `(Nothing, Nothing)` arm, line 3611), which
+loses it when old is dead. So the confirmation is the first message B sends on R'. R''s delivery
+worker does not start while R' is securing, so its accumulated rows cannot outrun the confirmation.
+`ICQSndSecure` sends the confirmation and only then starts the worker. This holds on restart too (see
+Worker gate).
+
+## New definitions
+
+Agent/Protocol.hs
+- Condition fast rotation on the existing `rpcAddressSMPAgentVersion` (v8, `Protocol.hs:322`).
+- New `AMessage` constructor `QEND SndQAddr` (tag `QE`), the address of the queue to remove. v8-only,
+  and only sent during fast rotation, so peers below v8 never parse it.
+- `SndSwitchStatus` constructors `SSSecuringQueue` (old, while R' secures) and `SSSendingQEND` (old,
+  after R' is secured — it drains its tail and `QEND` but takes no new messages).
+- `InternalCommand` constructor `ICQSndSecure SMP.SenderId`.
+
+No receive-side switch status, no boundary, no drain state.
+
+## Schema
+
+None. `SSSecuringQueue` uses `snd_queues.switch_status`; R' is secured into `rcv_queues.e2e_dh_secret`.
+No new columns, no migration.
+
+## Sender B
+
+### Dual scheduling from QADD
+
+`enqueueMessageB` writes a delivery row for the head send queue and for each `filter isActiveSndQ`
+tail queue (`Agent.hs:2345`). Adjust the selection two ways: additionally include a securing
+replacement queue on a v8 connection (`connAgentVersion cData >= rpcAddressSMPAgentVersion && status == New && isJust dbReplaceQueueId`),
+and exclude a terminating queue (`sndSwchStatus == Just SSSendingQEND`). The version guard keeps the
+slow path unchanged — there R' is also `New` with a replace reference during `QKEY`/`QUSE`, but it must
+not be dual-scheduled. `SSSendingQEND` is a fast-path-only status, so the exclusion never affects the
+slow path. The gate below is inert for the slow path anyway, since it never starts R''s worker while
+`New`.
+
+- `QADD` until R' secured: old is the head (active) and R' is the securing replacement, so every `SEND`
+  writes both rows. old delivers at once; R''s rows accumulate behind its gate.
+- R' secured: R' is the head (primary) and old is `SSSendingQEND` (excluded), so a `SEND` writes R'
+  only. old keeps its worker and delivers whatever was already scheduled on it, plus `QEND`.
+
+### Worker gate
+
+`submitPendingMsg` (`Agent.hs:2437`) and `resumeMsgDelivery` (`2421`) — the two `getDeliveryWorker`
+callers that start delivery — skip a queue with `status == New && isJust dbReplaceQueueId`, so neither
+a `SEND` nor startup starts R''s worker while it secures. Startup resumes delivery through
+`resumeMsgDelivery` (`resumeDelivery` line 1848, and `getAllSndQueuesForDelivery` line 1943), so R' is
+skipped there; `resumeAllCommands` (1883) resumes R''s `ICQSndSecure`, which secures R' and only then
+starts its worker.
+
+### Steps
+
+`qAddMsg` (fast branch, under the connection lock, `Agent.hs:3855`):
+- Add R' as the slow path does (line 3870): `addConnSndQueue (sq_) {primary = True, dbReplaceQueueId = Just old}`, `New`.
+- Duplicate **every** undelivered message on old to R': for each pending row on old
+  (`SELECT internal_id FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = old AND failed = 0`),
+  `createSndMsgDelivery db R' internalId`. This is the loss-prevention step: old's not-yet-sent
+  messages are duplicated onto R', so if old later fails they are already on R'. If old is already
+  down, nothing was sent and the whole backlog is duplicated.
+- `enqueueCommand (Just newSrv) (ICQSndSecure sndId)`; `setSndSwitchStatus SSSecuringQueue` on old
+  (where the slow path sets `SSSendingQKEY`, line 3874); notify `SWITCH QDSnd SPStarted`. old keeps
+  delivering; R''s worker is gated.
+
+`ICQSndSecure sId` (retryable, under `tryWithLock`):
+1. If old is already gone, a prior attempt finished; return. Otherwise find R' by `sId`.
+2. `secureSndQueue` (SKEY) R' — idempotent, since `sndPrivateKey` was persisted by `qAddMsg`
+   (`QueueStore/STM.hs:213`: same key → `Right ()`, different key → `AUTH`).
+3. Send the confirmation on R' (`sendConfirmation`; empty body, both peers already know each other;
+   `e2eEncryption_ = Nothing`, no ratchet step). It is the first message on R'.
+4. On success, in one transaction: `setSndQueueStatus R' Active` (the gate lifts), `setSndQueuePrimary R'`
+   (R' becomes the head; its replace reference is cleared), and `setSndSwitchStatus old (Just SSSendingQEND)`
+   (old takes no new messages but keeps its worker). Then `submitPendingMsg c R'` (the worker starts and
+   flushes the accumulated rows after the confirmation), `enqueueMessages [old, R'] (QEND oldAddr)`
+   (appended after old's tail, delivered on both), and notify `SWITCH QDSnd SPSecured`. From here a
+   `SEND` goes to R' only; old delivers its tail then `QEND` and is removed when `QEND` is sent.
+
+A temporary error retries from step 1; nothing is torn down. A permanent `AUTH` (should not occur, R'
+was secured with B's own key) leaves both queues and surfaces `A_QUEUE`.
+
+`QEND` sent — new `AM_QEND_` arm in `runSmpQueueMsgDelivery`, modelled on `AM_QTEST_` (`Agent.hs:2567`):
+on a successful send of `QEND addr`, remove the named send queue (`TM.delete` its worker,
+`deleteConnSndQueue addr`), make the remaining queue the sole primary (`setSndQueuePrimary`, which
+clears its `replace_snd_queue_id`), and notify `SWITCH QDSnd SPCompleted` (as `AM_QTEST_` does, line
+2591). This re-primary is a no-op on the send side — step 4 already made R' primary before `QEND` was
+enqueued. The handler is idempotent — a second `QEND` send finds the named queue already gone and does
+nothing. `QEND` is sent on both queues; removing old's send queue also drops any `QEND` still pending
+on old. The R' copy reliably removes old
+and reaches A even when old is dead; the old copy is best effort. Once old's send queue is gone, `SEND`
+schedules to R' only.
+
+## Recipient A
+
+A is subscribed to old (primary, `RSSendingQADD`) and R' (created at rotation start,
+`dbReplaceQueueId = old`).
+
+- **Confirmation on R'.** In `processClientMsg`, `(Nothing, Just e2ePubKey)` case, add an arm before the
+  `senderCanSecure` arm (`Agent.hs:3476`), guarded by `isJust (dbReplaceQueueId rq)`. In one
+  transaction: `setRcvQueueConfirmedE2E rq (C.dh' e2ePubKey e2ePrivKey) (min v phVer)` (secures R') and
+  `setRcvQueuePrimary R'` (clears R''s replace reference). Then `ack`, and notify `SWITCH QDRcv SPConfirmed`.
+  No conn-info processing, no ratchet step, no deferral. Redelivery is idempotent: R' now has `e2e_dh_secret`, so a re-sent
+  confirmation reaches the `(Just e2eDh, Just _)` arm (line 3608) and is acked — correct here, since
+  there is no backlog to hold.
+- **Data on R'.** With R''s replace reference cleared, a data message on R' takes the ordinary path
+  (`(_, dbReplaceQueueId=Nothing)`, line 3503) — no old-deletion, no `RSSendingQUSE` check. A copy
+  already read on old is dropped as `A_DUPLICATE`; a copy read first on R' advances the ratchet and
+  old's copy is then the duplicate.
+- **`QEND oldAddr` on either queue.** New `AMessage` handler (`qEndMsg`, a `qDuplex` handler like
+  `qAddMsg`): `findRQ oldAddr` the receive queue to remove. Mark it deleted (`setRcvQueueDeleted`, so
+  `getRcvQueuesByConnId_`'s `deleted = 0` filter excludes it at once and a restart does not resurrect
+  it) and `enqueueCommand (Just oldServer) (ICDeleteRcvQueue oldRcvId)` for the server `DEL` and record
+  removal — the async, crash-safe path `abortConnectionSwitch` uses, which resumes on restart and does
+  not block `QEND`, **not** the synchronous `deleteQueue` of `finalizeSwitch`, which would stall if old
+  is unreachable. `ICDeleteRcvQueue` (`Agent.hs:2224`) currently retries a temporary error forever;
+  bound it with the same persisted `rcv_queues.delete_errors`/`deleteErrorCount` mechanism `deleteQueueRec`
+  uses (2884): on a temporary error `incRcvDeleteErrors`, and at the limit `deleteConnRcvQueue` and
+  stop. The count is in the database, so the bound survives restarts, and its only other caller
+  (`abortConnectionSwitch'`, 2739) deletes an alive queue that succeeds well before the limit. `qEndMsg`
+  does **not** re-primary R' — the confirmation arm (above) owns R''s primary flag and replace
+  reference. `QEND` on old and the confirmation on R' travel on different queues with no order between
+  them, so `QEND` on old can be processed first (it sits only behind old's tail); re-primarying then
+  would clear R''s `dbReplaceQueueId` and the later confirmation would miss the rotation arm and never
+  secure R'. So `qEndMsg` only removes the named queue. Re-create the notification subscription
+  (`when enableNtfs $ sendNtfSubCommand ns (NSCCreate, [connId])`); notify
+  `SWITCH QDRcv SPCompleted`; `ackDel` the `QEND`. Received on both queues, the second finds it already
+  marked deleted and is a no-op.
+
+No drain, no boundary, no finalize command. Old is removed when `QEND` arrives, not by counting.
+
+## Abort / version
+
+- Fast rotation runs only when `connAgentVersion >= v8`; otherwise the `QKEY`/`QUSE` slow path runs
+  unchanged.
+- `canAbortRcvSwitch` (`Agent/Store.hs:210`) returns false for `RSSendingQADD` when `connAgentVersion >= v8`
+  (at v8 B always chooses fast, so A treats a sent `QADD` as committed). Its signature gains
+  `connAgentVersion`; both callers pass it from `cData` — `abortConnectionSwitch'` (2730) and
+  `rcvQueueInfo` in `connectionStats` (2976).
+
+## Losses and duplication
+
+- No boundary loss: the `QADD` step **duplicates** old's entire undelivered backlog onto R', and every
+  later message up to securing is scheduled on both queues, so if old fails it loses nothing B still
+  holds. The only messages old can strand are those its server already accepted but had not handed to
+  A — the ordinary store-and-forward risk, present whenever a server fails with unread messages, and
+  empty if old was already down (a down server accepted nothing).
+- Duplicates: the double ratchet drops them (`A_DUPLICATE`); `checkMsgIntegrity`'s `MsgDuplicate` is
+  only a flag, not the mechanism.
+- The 512 skip bound (`Crypto/Ratchet.hs:953`) does not bite on the rotation: each queue delivers in
+  order and every message B still holds is on R', so A reads a contiguous stream with only small
+  cross-queue reordering. A store-and-forward residual (above) is an ordinary loss, not introduced here.
+
+## Tests
+
+- new/new, old stopped right after `QADD`: rotation completes; all messages delivered on R'; old
+  removed by `QEND` on R'.
+- new/new, both alive: messages delivered on both, deduped; old removed by `QEND`.
+- new/old and old/new: fall back to the `QKEY`/`QUSE` path.
+- crash during securing: restart does not start R''s worker; `ICQSndSecure` resumes, secures R',
+  starts the worker, sends `QEND`.
+- `QEND` received on both queues: old removed once, the second receipt is a no-op.
+- `QEND` on old processed before the confirmation on R': R' still secures, because `qEndMsg` does not
+  clear R''s replace reference; rotation completes.

protocol/diagrams/duplex-messaging/queue-rotation-fast.mmd

@@ -5,13 +5,18 @@ sequenceDiagram
   participant S as Server<br>that has A's send queue<br>(B's receive queue)
   participant B as Bob
 
-  A ->> R': NEW: create new queue<br>(allow SKEY)
-  A ->> S: SEND: QADD (R'): send address<br>of the new queue(s)
+  A ->> R': NEW: create new queue (SKEY allowed)
+  A ->> S: SEND: QADD (R')
   S ->> B: MSG: QADD (R')
-  B ->> R': SKEY: secure new queue
-  B ->> R': SEND: QTEST
-  R' ->> A: MSG: QTEST
-  A ->> R: DEL: delete the old queue
-  B ->> R': SEND: send messages to the new queue
-  R' ->> A: MSG: receive messages from the new queue
-  
+  B ->> R: SEND: messages (also scheduled on R')
+  R ->> A: MSG: messages
+  B ->> R': SKEY: authorize B as sender
+  B ->> R': SEND: confirmation (establishes R' secret)
+  R' ->> A: MSG: confirmation (A secures R')
+  B ->> R': SEND: held copies (deduped), then new messages
+  R' ->> A: MSG: messages
+  B ->> R: SEND: remaining tail, then QEND
+  R ->> A: MSG: QEND
+  B ->> R': SEND: QEND
+  R' ->> A: MSG: QEND
+  A ->> R: DEL: delete the current queue

rfcs/2026-08-09-fast-queue-rotation.md

@@ -0,0 +1,97 @@
+---
+Proposed: 2026-08-09
+Protocol: agent-protocol v8
+Diagram: ../protocol/diagrams/duplex-messaging/queue-rotation-fast.svg
+---
+
+# Fast queue rotation
+
+## Problem
+
+In the current rotation the peer returns `QKEY` to the initiator over the initiator's current
+receiving queue. When that queue's server is unavailable the rotation cannot complete, so a client
+cannot move away from a failed server.
+
+## Solution
+
+Both the current rotation and v8 add a queue, deliver to both queues while the rotation is in
+progress, and remove the old queue; the recipient drops duplicates in both. The main difference is where
+the new queue's secret is established. In the current rotation it is established over the current
+queue, by `QKEY`, so it cannot complete when the current server is down. In v8 the peer establishes the
+new queue's secret over the new queue itself — a confirmation it sends on R' — so establishing the
+secret no longer depends on the current queue, and the rotation completes even when the current server
+is down.
+
+v8 also starts writing to both queues earlier: from the moment the queue is added, including the
+current queue's not-yet-delivered backlog. So the initiator adds the new queue with `QADD`; from that
+point the peer writes every message to both the current queue and R'. Once R' is secured the peer
+writes new messages to it alone, while the current queue delivers whatever was already scheduled on it
+and a final `QEND`, and is then removed. Because the recipient drops duplicates, neither the order of
+arrival nor which queue carries a message matters, provided each message arrives on at least one queue
+— with one exception, the confirmation, which is always the first message on R'. A dead new queue does
+not stop delivery either, because the current queue keeps delivering until R' is secured.
+
+Roles: A initiates (its receiving queue rotates; A receives on the new queue R'). B is the peer (B
+holds the sending queue to A, secures R', and delivers to both).
+
+Sequence:
+
+    A -> R'       : create new queue (messaging mode, SKEY allowed)
+    A -> S -> B   : QADD(R')          (over A's sending queue; A's current server untouched)
+    B             : from QADD, schedule every message and the current backlog on both queues
+    B -> current  : deliver the scheduled messages (R' holds its copies while securing)
+    B -> R'       : SKEY              (authorize B as sender)
+    B -> R'       : confirmation      (empty; establishes R' secret; first message on R')
+    B -> R'       : deliver R''s held copies (A dedups), then new messages to R' only
+    B -> current  : deliver the remaining tail, then QEND(current)
+    B -> R'       : QEND(current)
+    A             : on QEND, delete the current queue; keep receiving on R'
+
+## Confirmation
+
+The confirmation is the only message a recipient can read on a queue that is not yet secured, and it
+establishes the queue's shared secret without depending on the current queue. Because a data message
+that reached R' before the confirmation could not be read, the confirmation is the first message the
+peer sends on R', and the peer does not start ordinary delivery on R' until the confirmation has been
+sent.
+
+The confirmation body is empty: both parties already know each other, so no profile or reply queue is
+sent. It is sealed by the queue's box (keyed by the shared secret being established) and is not
+additionally encrypted with the double ratchet, so rotation does not advance the message ratchet.
+
+## Termination
+
+`QEND` names a queue to remove and is delivered on both queues. On receipt the recipient deletes the
+named queue; on send the peer removes its sending queue of that address. `QEND` is a general
+queue-removal message — the peer can remove either queue with it — so on the wire a rotation is the
+addition of a queue (`QADD`) and the later removal of the replaced one (`QEND`), each an ordinary
+operation on the queue set rather than a `QTEST`-style completion. Delivering `QEND` on the removed
+queue is best effort; the copy on the surviving queue removes it and reaches the recipient even when
+the removed server is dead.
+
+## Per-queue secret
+
+R' secret is a fresh Diffie-Hellman between the peer's queue key, sent in the confirmation header, and
+the initiator's R' key. It does not depend on any current queue, so redundancy of current queues is
+unaffected.
+
+## Compatibility
+
+Fast rotation runs only when the connection's agreed agent protocol version is 8 or higher; the peer
+chooses it. Otherwise the `QKEY`/`QUSE` exchange is used. `QEND` is defined at version 8 and is only
+sent during fast rotation, so peers below version 8 never receive it.
+
+    new A / new B  : fast (QADD, confirmation on R', QEND)
+    new A / old B  : slow (old B returns QKEY; new A keeps the QKEY/QUSE handling)
+    old A / new B  : slow (agreed version below 8; new B returns QKEY)
+    old A / old B  : slow
+
+The recipient does not choose by version; it reacts to whichever message arrives — `QKEY`, or a
+confirmation on R' followed later by `QEND`.
+
+## Dead current server
+
+The initiator keeps reading messages on the new queue and removes the current queue when `QEND`
+arrives there, without waiting for the current server. Nothing is lost, because every message is
+scheduled on the new queue; the only cleanup that a dead current server delays is the deletion
+of its queue, which is retried a bounded number of times and then abandoned.

src/Simplex/Messaging/Agent.hs

@@ -2224,8 +2224,12 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
         ICDeleteConn -> withStore' c (`deleteCommand` cmdId)
         ICDeleteRcvQueue rId -> withServer $ \srv -> tryWithLock "ICDeleteRcvQueue" $ do
           rq <- withStore c (\db -> getDeletedRcvQueue db connId srv rId)
-          deleteQueue c NRMBackground rq
-          withStore' c (`deleteConnRcvQueue` rq)
+          maxErrs <- asks $ deleteErrorCount . config
+          tryAllErrors (deleteQueue c NRMBackground rq) >>= \case
+            Left e | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> do
+              withStore' c (`incRcvDeleteErrors` rq)
+              throwE e
+            _ -> withStore' c (`deleteConnRcvQueue` rq)
         ICQSecure rId senderKey ->
           withServer $ \srv -> tryWithLock "ICQSecure" . withDuplexConn $ \(DuplexConnection cData rqs sqs) ->
             case find (sameQueue (srv, rId)) rqs of
@@ -2245,6 +2249,31 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
                     notify $ SWITCH QDRcv SPSecured cStats
                   _ -> internalErr "ICQSecure: no switching queue found"
               _ -> internalErr "ICQSecure: queue address not found in connection"
+        ICQSndSecure sId ->
+          withServer $ \srv -> tryWithLock "ICQSndSecure" . withDuplexConn $ \(DuplexConnection cData@ConnData {connAgentVersion} rqs sqs) ->
+            case findQ (srv, sId) sqs of
+              Nothing -> internalErr "ICQSndSecure: queue address not found in connection"
+              Just sq'@SndQueue {dbReplaceQueueId} ->
+                case dbReplaceQueueId >>= \replaceQId -> find ((replaceQId ==) . dbQId) sqs of
+                  Just oldSq -> do
+                    secureSndQueue c NRMBackground sq'
+                    let confMsg = smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption_ = Nothing, encConnInfo = ""}
+                    void $ sendConfirmation c NRMBackground sq' confMsg
+                    oldSq' <- withStore' c $ \db -> do
+                      setSndQueueStatus db sq' Active
+                      setSndQueuePrimary db connId sq'
+                      setSndSwitchStatus db oldSq $ Just SSSendingQEND
+                    let sq'' = (sq' :: SndQueue) {status = Active, primary = True, dbReplaceQueueId = Nothing}
+                    pending <- withStore' c $ \db -> countSndQueueDeliveries db sq''
+                    atomically $ modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + pending}
+                    lift $ resumeMsgDelivery c sq''
+                    void $ enqueueMessages c cData [oldSq, sq''] SMP.noMsgFlags $ QEND [qAddress oldSq]
+                    let conn' = DuplexConnection cData rqs (updatedQs oldSq' $ updatedQs sq'' sqs)
+                    cStats <- connectionStats c conn'
+                    notify $ SWITCH QDSnd SPSecured cStats
+                  Nothing ->
+                    forM_ (find (\q -> sndSwchStatus q == Just SSSendingQEND) sqs) $ \oldSq ->
+                      void $ enqueueMessages c cData [oldSq, sq'] SMP.noMsgFlags $ QEND [qAddress oldSq]
         ICQDelete rId -> do
           withServer $ \srv -> tryWithLock "ICQDelete" . withDuplexConn $ \(DuplexConnection cData@ConnData {enableNtfs} rqs sqs) -> do
             case removeQ (srv, rId) rqs of
@@ -2269,9 +2298,11 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
                     notify $ SWITCH QDRcv SPCompleted cStats
               _ -> internalErr "ICQDelete: cannot delete the only queue in connection"
         where
-          ack srv rId srvMsgId = do
-            rq <- withStore c $ \db -> getRcvQueue db connId srv rId
-            ackQueueMessage c rq srvMsgId
+          ack srv rId srvMsgId =
+            withStore' c (\db -> getRcvQueue db connId srv rId) >>= \case
+              Right rq -> ackQueueMessage c rq srvMsgId
+              Left SEConnNotFound -> pure Nothing
+              Left e -> throwE $ storeError e
           secure :: RcvQueue -> SMP.SndPublicAuthKey -> AM ()
           secure rq@RcvQueue {server} senderKey = do
             secureQueue c NRMBackground rq senderKey
@@ -2326,8 +2357,10 @@ enqueueMessagesB c reqs = do
   enqueueSavedMessageB c $ mapMaybe snd $ rights $ toList reqs'
   pure $ fst <$$> reqs'
 
-isActiveSndQ :: SndQueue -> Bool
-isActiveSndQ SndQueue {status} = status == Secured || status == Active
+isActiveSndQ :: ConnData -> SndQueue -> Bool
+isActiveSndQ ConnData {connAgentVersion} sq@SndQueue {status, sndSwchStatus} =
+  sndSwchStatus /= Just SSSendingQEND
+    && (status == Secured || status == Active || (connAgentVersion >= rpcAddressSMPAgentVersion && securingSndQueue sq))
 {-# INLINE isActiveSndQ #-}
 
 enqueueMessage :: AgentClient -> ConnData -> SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, PQEncryption)
@@ -2341,9 +2374,9 @@ enqueueMessageB c reqs = do
   cfg <- asks config
   (_, reqMids) <- unsafeWithStore c $ \db -> do
     mapAccumLM (\ids r -> storeSentMsg db cfg ids r `E.catchAny` \e -> (ids,) <$> handleInternal e) IM.empty reqs
-  forME reqMids $ \((csqs_, _, _, _), InternalId msgId, pqSecr) -> forM csqs_ $ \(_, sq :| sqs) -> do
+  forME reqMids $ \((csqs_, _, _, _), InternalId msgId, pqSecr) -> forM csqs_ $ \(cData, sq :| sqs) -> do
     submitPendingMsg c sq
-    let sqs' = filter isActiveSndQ sqs
+    let sqs' = filter (isActiveSndQ cData) sqs
     pure ((msgId, pqSecr), if null sqs' then Nothing else Just (sqs', msgId))
   where
     storeSentMsg ::
@@ -2423,9 +2456,13 @@ resumeMsgDelivery :: AgentClient -> SndQueue -> AM' ()
 -- hasWork is passed as False to avoid unnecessary write to TMVar:
 -- - new worker is always created by "some work to do".
 -- - if the worker already exists, there is no need to "push" it again.
-resumeMsgDelivery = void .: getDeliveryWorker False
+resumeMsgDelivery c sq = unless (securingSndQueue sq) $ void $ getDeliveryWorker False c sq
 {-# INLINE resumeMsgDelivery #-}
 
+securingSndQueue :: SndQueue -> Bool
+securingSndQueue SndQueue {status, dbReplaceQueueId} = status == New && isJust dbReplaceQueueId
+{-# INLINE securingSndQueue #-}
+
 getDeliveryWorker :: Bool -> AgentClient -> SndQueue -> AM' (Worker, TMVar ())
 getDeliveryWorker hasWork c sq =
   getAgentWorker' fst mkLock "msg_delivery" hasWork c (qAddress sq) (smpDeliveryWorkers c) (runSmpQueueMsgDelivery c sq)
@@ -2435,7 +2472,7 @@ getDeliveryWorker hasWork c sq =
       pure (w, retryLock)
 
 submitPendingMsg :: AgentClient -> SndQueue -> AM' ()
-submitPendingMsg c sq = do
+submitPendingMsg c sq = unless (securingSndQueue sq) $ do
   atomically $ modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + 1}
   void $ getDeliveryWorker True c sq
 
@@ -2506,6 +2543,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server,
                     AM_QKEY_ -> qError msgId "QKEY: AUTH"
                     AM_QUSE_ -> qError msgId "QUSE: AUTH"
                     AM_QTEST_ -> qError msgId "QTEST: AUTH"
+                    AM_QEND_ -> delMsg msgId
                     AM_EREADY_ -> notifyDel msgId err
                     AM_SRV_REQ -> logError "AM_SRV_REQ: unexpected stored message" >> delMsg msgId
                     AM_SRV_RESP -> notifyDel msgId err
@@ -2593,6 +2631,20 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} sq@SndQueue {userId, connId, server,
                             _ -> internalErr msgId "sent QTEST: there is only one queue in connection"
                         _ -> internalErr msgId "sent QTEST: queue not in connection or not replacing another queue"
                     _ -> internalErr msgId "QTEST sent not in duplex connection"
+                AM_QEND_ -> withConnLockNotify c connId "runSmpQueueMsgDelivery AM_QEND_" $ do
+                  SomeConn _ conn <- withStore c (`getConn` connId)
+                  case conn of
+                    DuplexConnection cData' rqs sqs ->
+                      forM (removeQP (\sq' -> sndSwchStatus sq' == Just SSSendingQEND) sqs) $ \case
+                        (oldSq, sq'' : sqs') -> do
+                          atomically $ TM.delete (qAddress oldSq) $ smpDeliveryWorkers c
+                          withStore' c $ \db -> do
+                            deletePendingMsgs db connId oldSq
+                            deleteConnSndQueue db connId oldSq
+                          cStats <- connectionStats c $ DuplexConnection cData' rqs (sq'' :| sqs')
+                          pure ("", connId, AEvt SAEConn $ SWITCH QDSnd SPCompleted cStats)
+                        (_, []) -> pure ("", connId, AEvt SAEConn $ ERR $ INTERNAL "sent QEND: no remaining queue in connection")
+                    _ -> internalErr msgId "QEND sent not in duplex connection"
                 AM_EREADY_ -> pure ()
                 AM_SRV_REQ -> logError "AM_SRV_REQ: unexpected stored message"
                 AM_SRV_RESP -> notify $ SSENT mId proxySrv_
@@ -2728,7 +2780,7 @@ abortConnectionSwitch' c connId =
     withStore c (`getConn` connId) >>= \case
       SomeConn _ (DuplexConnection cData rqs sqs) -> case switchingRQ rqs of
         Just rq
-          | canAbortRcvSwitch rq -> do
+          | canAbortRcvSwitch cData rq -> do
               when (ratchetSyncSendProhibited cData) $ throwE $ CMD PROHIBITED "abortConnectionSwitch: send prohibited"
               -- multiple queues to which the connections switches were possible when repeating switch was allowed
               let (delRqs, keepRqs) = L.partition ((Just (dbQId rq) ==) . dbReplaceQId) rqs
@@ -2757,7 +2809,7 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni
           AgentConfig {e2eEncryptVRange} <- asks config
           g <- asks random
           (pks, e2eParams) <- liftIO $ CR.generateRcvE2EParams g (maxVersion e2eEncryptVRange) pqSupport'
-          enqueueRatchetKeyMsgs c sqs e2eParams
+          enqueueRatchetKeyMsgs c cData' sqs e2eParams
           withStore' c $ \db -> do
             setConnRatchetSync db connId RSStarted
             setRatchetX3dhKeys db connId pks
@@ -2943,12 +2995,12 @@ getConnectionRatchetAdHash' c connId = do
 connectionStats :: AgentClient -> Connection c -> AM ConnectionStats
 connectionStats c = \case
   RcvConnection cData rq -> do
-    rcvQueuesInfo <- (: []) <$> rcvQueueInfo rq
+    rcvQueuesInfo <- (: []) <$> rcvQueueInfo cData rq
     pure (stats cData) {rcvQueuesInfo, subStatus = connSubStatus rcvQueuesInfo}
   SndConnection cData sq -> do
     pure (stats cData) {sndQueuesInfo = [sndQueueInfo sq]}
   DuplexConnection cData rqs sqs -> do
-    rcvQueuesInfo <- mapM rcvQueueInfo (L.toList rqs)
+    rcvQueuesInfo <- mapM (rcvQueueInfo cData) (L.toList rqs)
     pure
       (stats cData)
         { rcvQueuesInfo,
@@ -2956,7 +3008,7 @@ connectionStats c = \case
           subStatus = connSubStatus rcvQueuesInfo
         }
   ContactConnection cData rq -> do
-    rcvQueuesInfo <- (: []) <$> rcvQueueInfo rq
+    rcvQueuesInfo <- (: []) <$> rcvQueueInfo cData rq
     pure (stats cData) {rcvQueuesInfo, subStatus = connSubStatus rcvQueuesInfo}
   NewConnection cData ->
     pure $ stats cData
@@ -2971,10 +3023,10 @@ connectionStats c = \case
           ratchetSyncSupported = True,
           subStatus = Nothing
         }
-    rcvQueueInfo :: RcvQueue -> AM RcvQueueInfo
-    rcvQueueInfo rq@RcvQueue {server, status, rcvSwchStatus} = do
+    rcvQueueInfo :: ConnData -> RcvQueue -> AM RcvQueueInfo
+    rcvQueueInfo cData rq@RcvQueue {server, status, rcvSwchStatus} = do
       subStatus <- atomically checkQueueSubStatus
-      pure $ RcvQueueInfo {rcvServer = server, status, rcvSwitchStatus = rcvSwchStatus, canAbortSwitch = canAbortRcvSwitch rq, subStatus}
+      pure $ RcvQueueInfo {rcvServer = server, status, rcvSwitchStatus = rcvSwchStatus, canAbortSwitch = canAbortRcvSwitch cData rq, subStatus}
       where
         checkQueueSubStatus :: STM SubscriptionStatus
         checkQueueSubStatus =
@@ -3522,6 +3574,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                               -- no action needed for QTEST
                               -- any message in the new queue will mark it active and trigger deletion of the old queue
                               QTEST _ -> logServer "<--" c srv rId ("MSG <QTEST>:" <> logSecret' srvMsgId) >> ackDel msgId
+                              QEND addrs -> qDuplexAckDel conn'' "QEND" $ qEndMsg srvMsgId addrs
                               EREADY _ -> qDuplexAckDel conn'' "EREADY" $ ereadyMsg rcPrev
                             where
                               qDuplexAckDel :: Connection c -> String -> (Connection 'CDuplex -> AM ()) -> AM ACKd
@@ -3702,6 +3755,18 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
             checkConfVersions agentVersion phVer
             let ConnData {pqSupport, serviceRequestExpiresAt} = toConnData conn'
             case status of
+              New | isJust (dbReplaceQId rq) -> case conn' of
+                DuplexConnection cData' rqs sqs -> do
+                  let dhSecret = C.dh' e2ePubKey e2ePrivKey
+                      clientVersion = min agreedClientVerion phVer
+                  withStore' c $ \db -> do
+                    setRcvQueueConfirmedE2E db rq dhSecret clientVersion
+                    setRcvQueuePrimary db connId rq
+                  let rq' = (rq :: RcvQueue) {status = Confirmed, e2eDhSecret = Just dhSecret, smpClientVersion = clientVersion, primary = True, dbReplaceQueueId = Nothing}
+                      conn'' = DuplexConnection cData' (updatedQs rq' rqs) sqs
+                  cStats <- connectionStats c conn''
+                  notify $ SWITCH QDRcv SPConfirmed cStats
+                _ -> prohibited "conf: rotation not in duplex connection"
               New -> case conn' of
                 -- party initiating connection
                 RcvConnection {} -> do
@@ -3853,7 +3918,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
           -- processed by queue sender
           qAddMsg :: SMP.MsgId -> NonEmpty (SMPQueueUri, Maybe SndQAddr) -> Connection 'CDuplex -> AM ()
           qAddMsg _ ((_, Nothing) :| _) _ = qError "adding queue without switching is not supported"
-          qAddMsg srvMsgId ((qUri, Just addr) :| _) (DuplexConnection cData' rqs sqs) = do
+          qAddMsg srvMsgId ((qUri, Just addr) :| _) (DuplexConnection cData'@ConnData {connAgentVersion} rqs sqs) = do
             when (ratchetSyncSendProhibited cData') $ throwE $ AGENT (A_QUEUE "ratchet is not synchronized")
             clientVRange <- asks $ smpClientVRange . config
             case qUri `compatibleVersion` clientVRange of
@@ -3870,9 +3935,17 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                           liftIO $ mapM_ (deleteConnSndQueue db connId) delSqs
                           addConnSndQueue db connId (sq_ :: NewSndQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
                         logServer "<--" c srv rId $ "MSG <QADD>:" <> logSecret' srvMsgId <> " " <> logSecret (senderId queueAddress)
-                        let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}}
-                        void . enqueueMessages c cData' sqs SMP.noMsgFlags $ QKEY [(sqInfo', C.toPublic sndPrivateKey)]
-                        sq1 <- withStore' c $ \db -> setSndSwitchStatus db sq $ Just SSSendingQKEY
+                        swchStatus <-
+                          if connAgentVersion >= rpcAddressSMPAgentVersion
+                            then do
+                              withStore' c $ \db -> copyPendingSndDeliveries db sq sq2
+                              enqueueCommand c "" connId (Just $ qServer sq2) $ AInternalCommand $ ICQSndSecure (snd $ qAddress sq2)
+                              pure SSSecuringQueue
+                            else do
+                              let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}}
+                              void . enqueueMessages c cData' sqs SMP.noMsgFlags $ QKEY [(sqInfo', C.toPublic sndPrivateKey)]
+                              pure SSSendingQKEY
+                        sq1 <- withStore' c $ \db -> setSndSwitchStatus db sq $ Just swchStatus
                         let sqs'' = updatedQs sq1 sqs' <> [sq2]
                             conn' = DuplexConnection cData' rqs sqs''
                         cStats <- connectionStats c conn'
@@ -3926,6 +3999,23 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                   _ -> qError "QUSE: switching SndQueue not found in connection"
               _ -> qError "QUSE: switched queue address not found in connection"
 
+          -- processed by queue recipient
+          qEndMsg :: SMP.MsgId -> NonEmpty SndQAddr -> Connection 'CDuplex -> AM ()
+          qEndMsg srvMsgId addrs (DuplexConnection cData'@ConnData {enableNtfs} rqs sqs) =
+            case L.partition (\rq' -> any (`sameQAddress` sndAddress rq') addrs) rqs of
+              (removed@(_ : _), keptRq : keptRqs) -> do
+                logServer "<--" c srv rId $ "MSG <QEND>:" <> logSecret' srvMsgId
+                forM_ removed $ \rq'@RcvQueue {server = rmServer, rcvId} -> do
+                  withStore' c $ \db -> setRcvQueueDeleted db rq'
+                  enqueueCommand c "" connId (Just rmServer) $ AInternalCommand $ ICDeleteRcvQueue rcvId
+                when enableNtfs $ do
+                  ns <- asks ntfSupervisor
+                  liftIO $ sendNtfSubCommand ns (NSCCreate, [connId])
+                let conn' = DuplexConnection cData' (keptRq :| keptRqs) sqs
+                cStats <- connectionStats c conn'
+                notify $ SWITCH QDRcv SPCompleted cStats
+              _ -> pure ()
+
           qError :: String -> AM a
           qError = throwE . AGENT . A_QUEUE
 
@@ -4030,7 +4120,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar
                   sendReplyKey = do
                     g <- asks random
                     (pks, e2eParams) <- liftIO $ CR.generateRcvE2EParams g e2eVersion pqSupport
-                    enqueueRatchetKeyMsgs c sqs e2eParams
+                    enqueueRatchetKeyMsgs c cData' sqs e2eParams
                     pure pks
                   notifyRatchetSyncError = do
                     let cData'' = cData' {ratchetSyncState = RSRequired} :: ConnData
@@ -4172,10 +4262,10 @@ storeConfirmation c cData@ConnData {connId, pqSupport, connAgentVersion = v} sq
     liftIO $ createSndMsg db connId msgData
     liftIO $ createSndMsgDelivery db sq internalId
 
-enqueueRatchetKeyMsgs :: AgentClient -> NonEmpty SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM ()
-enqueueRatchetKeyMsgs c (sq :| sqs) e2eEncryption = do
+enqueueRatchetKeyMsgs :: AgentClient -> ConnData -> NonEmpty SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM ()
+enqueueRatchetKeyMsgs c cData (sq :| sqs) e2eEncryption = do
   msgId <- enqueueRatchetKey c sq e2eEncryption
-  mapM_ (lift . enqueueSavedMessage c msgId) $ filter isActiveSndQ sqs
+  mapM_ (lift . enqueueSavedMessage c msgId) $ filter (isActiveSndQ cData) sqs
 
 enqueueRatchetKey :: AgentClient -> SndQueue -> CR.RcvE2ERatchetParams 'C.X448 -> AM AgentMsgId
 enqueueRatchetKey c sq@SndQueue {connId} e2eEncryption = do

src/Simplex/Messaging/Agent/Protocol.hs

@@ -639,16 +639,22 @@ instance FromJSON RcvSwitchStatus where
 data SndSwitchStatus
   = SSSendingQKEY
   | SSSendingQTEST
+  | SSSecuringQueue
+  | SSSendingQEND
   deriving (Eq, Show)
 
 instance StrEncoding SndSwitchStatus where
   strEncode = \case
     SSSendingQKEY -> "sending_qkey"
     SSSendingQTEST -> "sending_qtest"
+    SSSecuringQueue -> "securing_queue"
+    SSSendingQEND -> "sending_qend"
   strP =
     A.takeTill (== ' ') >>= \case
       "sending_qkey" -> pure SSSendingQKEY
       "sending_qtest" -> pure SSSendingQTEST
+      "securing_queue" -> pure SSSecuringQueue
+      "sending_qend" -> pure SSSendingQEND
       _ -> fail "bad SndSwitchStatus"
 
 instance ToField SndSwitchStatus where toField = toField . decodeLatin1 . strEncode
@@ -960,6 +966,7 @@ data AgentMessageType
   | AM_QKEY_
   | AM_QUSE_
   | AM_QTEST_
+  | AM_QEND_
   | AM_EREADY_
   | AM_SRV_REQ
   | AM_SRV_RESP
@@ -979,6 +986,7 @@ instance Encoding AgentMessageType where
     AM_QKEY_ -> "QK"
     AM_QUSE_ -> "QU"
     AM_QTEST_ -> "QT"
+    AM_QEND_ -> "QE"
     AM_EREADY_ -> "E"
     AM_SRV_REQ -> "A"
     AM_SRV_RESP -> "P"
@@ -998,6 +1006,7 @@ instance Encoding AgentMessageType where
           'K' -> pure AM_QKEY_
           'U' -> pure AM_QUSE_
           'T' -> pure AM_QTEST_
+          'E' -> pure AM_QEND_
           _ -> fail "bad AgentMessageType"
       'E' -> pure AM_EREADY_
       'A' -> pure AM_SRV_REQ
@@ -1037,6 +1046,7 @@ data AMsgType
   | QKEY_
   | QUSE_
   | QTEST_
+  | QEND_
   | EREADY_
   deriving (Eq)
 
@@ -1050,6 +1060,7 @@ instance Encoding AMsgType where
     QKEY_ -> "QK"
     QUSE_ -> "QU"
     QTEST_ -> "QT"
+    QEND_ -> "QE"
     EREADY_ -> "E"
   smpP =
     A.anyChar >>= \case
@@ -1063,6 +1074,7 @@ instance Encoding AMsgType where
           'K' -> pure QKEY_
           'U' -> pure QUSE_
           'T' -> pure QTEST_
+          'E' -> pure QEND_
           _ -> fail "bad AMsgType"
       'E' -> pure EREADY_
       _ -> fail "bad AMsgType"
@@ -1087,6 +1099,8 @@ data AMessage
     QUSE (NonEmpty (SndQAddr, Bool))
   | -- sent by the sender to test new queues and to complete switching
     QTEST (NonEmpty SndQAddr)
+  | -- sent by the sender to remove queues from the connection (fast rotation, v8)
+    QEND (NonEmpty SndQAddr)
   | -- ratchet re-synchronization is complete, with last decrypted sender message id (recipient's `last_external_snd_msg_id`)
     EREADY AgentMsgId
   deriving (Show)
@@ -1105,6 +1119,7 @@ aMessageType = \case
   QKEY _ -> AM_QKEY_
   QUSE _ -> AM_QUSE_
   QTEST _ -> AM_QTEST_
+  QEND _ -> AM_QEND_
   EREADY _ -> AM_EREADY_
 
 -- | this type is used to send as part of the protocol between different clients
@@ -1157,6 +1172,7 @@ instance Encoding AMessage where
     QKEY qs -> smpEncode (QKEY_, qs)
     QUSE qs -> smpEncode (QUSE_, qs)
     QTEST qs -> smpEncode (QTEST_, qs)
+    QEND qs -> smpEncode (QEND_, qs)
     EREADY lastDecryptedMsgId -> smpEncode (EREADY_, lastDecryptedMsgId)
   smpP =
     smpP
@@ -1169,6 +1185,7 @@ instance Encoding AMessage where
         QKEY_ -> QKEY <$> smpP
         QUSE_ -> QUSE <$> smpP
         QTEST_ -> QTEST <$> smpP
+        QEND_ -> QEND <$> smpP
         EREADY_ -> EREADY <$> smpP
 
 instance ToField AMessage where toField = toField . Binary . smpEncode

src/Simplex/Messaging/Agent/Store.hs

@@ -207,12 +207,13 @@ rcvSMPQueueAddress :: RcvQueue -> SMPQueueAddress
 rcvSMPQueueAddress RcvQueue {server, sndId, e2ePrivKey, queueMode} =
   SMPQueueAddress server sndId (C.publicKey e2ePrivKey) queueMode
 
-canAbortRcvSwitch :: RcvQueue -> Bool
-canAbortRcvSwitch = maybe False canAbort . rcvSwchStatus
+canAbortRcvSwitch :: ConnData -> RcvQueue -> Bool
+canAbortRcvSwitch ConnData {connAgentVersion} = maybe False canAbort . rcvSwchStatus
   where
     canAbort = \case
       RSSwitchStarted -> True
-      RSSendingQADD -> True
+      -- at agent version 8 and above the peer always chooses fast rotation, so a sent QADD is committed
+      RSSendingQADD -> connAgentVersion < rpcAddressSMPAgentVersion
       -- if switch is in RSSendingQUSE, a race condition with sender deleting the original queue is possible
       RSSendingQUSE -> False
       -- if switch is in RSReceivedMessage status, aborting switch (deleting new queue)
@@ -538,6 +539,7 @@ data InternalCommand
   | ICDeleteConn
   | ICDeleteRcvQueue SMP.RecipientId
   | ICQSecure SMP.RecipientId SMP.SndPublicAuthKey
+  | ICQSndSecure SMP.SenderId
   | ICQDelete SMP.RecipientId
   | ICReplyDel
 
@@ -549,6 +551,7 @@ data InternalCommandTag
   | ICDeleteConn_
   | ICDeleteRcvQueue_
   | ICQSecure_
+  | ICQSndSecure_
   | ICQDelete_
   | ICReplyDel_
   deriving (Show)
@@ -562,6 +565,7 @@ instance StrEncoding InternalCommand where
     ICDeleteConn -> strEncode ICDeleteConn_
     ICDeleteRcvQueue rId -> strEncode (ICDeleteRcvQueue_, rId)
     ICQSecure rId senderKey -> strEncode (ICQSecure_, rId, senderKey)
+    ICQSndSecure sId -> strEncode (ICQSndSecure_, sId)
     ICQDelete rId -> strEncode (ICQDelete_, rId)
     ICReplyDel -> strEncode ICReplyDel_
   strP =
@@ -573,6 +577,7 @@ instance StrEncoding InternalCommand where
       ICDeleteConn_ -> pure ICDeleteConn
       ICDeleteRcvQueue_ -> ICDeleteRcvQueue <$> _strP
       ICQSecure_ -> ICQSecure <$> _strP <*> _strP
+      ICQSndSecure_ -> ICQSndSecure <$> _strP
       ICQDelete_ -> ICQDelete <$> _strP
       ICReplyDel_ -> pure ICReplyDel
 
@@ -585,6 +590,7 @@ instance StrEncoding InternalCommandTag where
     ICDeleteConn_ -> "DELETE_CONN"
     ICDeleteRcvQueue_ -> "DELETE_RCV_QUEUE"
     ICQSecure_ -> "QSECURE"
+    ICQSndSecure_ -> "QSND_SECURE"
     ICQDelete_ -> "QDELETE"
     ICReplyDel_ -> "REPLY_DEL"
   strP =
@@ -596,6 +602,7 @@ instance StrEncoding InternalCommandTag where
       "DELETE_CONN" -> pure ICDeleteConn_
       "DELETE_RCV_QUEUE" -> pure ICDeleteRcvQueue_
       "QSECURE" -> pure ICQSecure_
+      "QSND_SECURE" -> pure ICQSndSecure_
       "QDELETE" -> pure ICQDelete_
       "REPLY_DEL" -> pure ICReplyDel_
       _ -> fail "bad InternalCommandTag"
@@ -614,6 +621,7 @@ internalCmdTag = \case
   ICDeleteConn -> ICDeleteConn_
   ICDeleteRcvQueue {} -> ICDeleteRcvQueue_
   ICQSecure {} -> ICQSecure_
+  ICQSndSecure {} -> ICQSndSecure_
   ICQDelete _ -> ICQDelete_
   ICReplyDel -> ICReplyDel_
 

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

@@ -131,6 +131,8 @@ module Simplex.Messaging.Agent.Store.AgentStore
     createSndMsg,
     updateSndMsgHash,
     createSndMsgDelivery,
+    copyPendingSndDeliveries,
+    countSndQueueDeliveries,
     getSndMsgViaRcpt,
     updateSndMsgRcpt,
     getPendingQueueMsg,
@@ -1041,6 +1043,24 @@ createSndMsgDelivery :: DB.Connection -> SndQueue -> InternalId -> IO ()
 createSndMsgDelivery db SndQueue {connId, dbQueueId} msgId =
   DB.execute db "INSERT INTO snd_message_deliveries (conn_id, snd_queue_id, internal_id) VALUES (?, ?, ?)" (connId, dbQueueId, msgId)
 
+-- copies every undelivered (failed = 0) delivery from one snd queue to another, for redundant delivery during fast rotation
+copyPendingSndDeliveries :: DB.Connection -> SndQueue -> SndQueue -> IO ()
+copyPendingSndDeliveries db SndQueue {connId, dbQueueId = fromQueueId} SndQueue {dbQueueId = toQueueId} =
+  DB.execute
+    db
+    [sql|
+      INSERT INTO snd_message_deliveries (conn_id, snd_queue_id, internal_id)
+      SELECT conn_id, ?, internal_id
+      FROM snd_message_deliveries
+      WHERE conn_id = ? AND snd_queue_id = ? AND failed = 0
+    |]
+    (toQueueId, connId, fromQueueId)
+
+countSndQueueDeliveries :: DB.Connection -> SndQueue -> IO Int
+countSndQueueDeliveries db SndQueue {connId, dbQueueId} =
+  maybeFirstRow' 0 fromOnly $
+    DB.query db "SELECT count(1) FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ? AND failed = 0" (connId, dbQueueId)
+
 getSndMsgViaRcpt :: DB.Connection -> ConnId -> InternalSndId -> IO (Either StoreError SndMsg)
 getSndMsgViaRcpt db connId sndMsgId =
   firstRow toSndMsg (SEMsgNotFound "getSndMsgViaRcpt") $

tests/AgentTests/FunctionalAPITests.hs

@@ -30,6 +30,7 @@ module AgentTests.FunctionalAPITests
     makeConnection,
     exchangeGreetings,
     switchComplete,
+    fastSwitchComplete,
     createConnection,
     joinConnection,
     sendMessage,
@@ -52,6 +53,7 @@ module AgentTests.FunctionalAPITests
     pattern Msg',
     pattern SENT,
     agentCfgVPrevPQ,
+    agentCfgV7,
   )
 where
 
@@ -527,6 +529,10 @@ functionalAPITests ps = do
     it "should handle service unavailable on startup" $ testServiceUnavailableOnStartup ps
     it "migrate connections to and from service" $ testMigrateConnectionsToService ps
   describe "Connection switch" $ do
+    describe "should switch delivery to the new queue with fast rotation" $
+      testServerMatrix2 ps testFastSwitchConnection
+    it "should switch delivery to the new queue when the old server is down" $
+      testFastSwitchDeadOldServer ps
     describe "should switch delivery to the new queue" $
       testServerMatrix2 ps testSwitchConnection
     describe "should switch to new queue asynchronously" $
@@ -3517,9 +3523,13 @@ testUsersNoServer ps = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
   where
     aCfg = agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3}
 
+-- fast rotation runs at agent version 8+; these tests pin to v7 to exercise the QKEY/QUSE slow path and switch abort
+agentCfgV7 :: AgentConfig
+agentCfgV7 = agentCfg {smpAgentVRange = mkVersionRange 6 7}
+
 testSwitchConnection :: InitialAgentServers -> IO ()
 testSwitchConnection servers =
-  withAgentClientsCfgServers2 agentCfg agentCfg servers $ \a b -> runRight_ $ do
+  withAgentClientsCfgServers2 agentCfgV7 agentCfgV7 servers $ \a b -> runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetings a bId b aId
     testFullSwitch a bId b aId 8
@@ -3543,6 +3553,68 @@ switchComplete a bId b aId = do
   phaseSnd b aId SPCompleted [Nothing]
   phaseRcv a bId SPCompleted [Nothing]
 
+testFastSwitchConnection :: InitialAgentServers -> IO ()
+testFastSwitchConnection servers =
+  withAgentClientsCfgServers2 agentCfg agentCfg servers $ \a b -> runRight_ $ do
+    (aId, bId) <- makeConnection a b
+    exchangeGreetings a bId b aId
+    stats <- switchConnectionAsync a "" bId
+    liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Just RSSwitchStarted]
+    fastSwitchComplete a bId b aId
+    exchangeGreetingsMsgId 6 a bId b aId
+
+fastSwitchComplete :: AgentClient -> ByteString -> AgentClient -> ByteString -> ExceptT AgentErrorType IO ()
+fastSwitchComplete a bId b aId = do
+  phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing]
+  phaseSnd b aId SPStarted [Just SSSecuringQueue, Nothing]
+  phaseSnd b aId SPSecured [Just SSSendingQEND, Nothing]
+  phaseRcv a bId SPConfirmed [Just RSSendingQADD, Nothing]
+  phaseRcv a bId SPCompleted [Nothing]
+  phaseSnd b aId SPCompleted [Nothing]
+
+-- A's old receive queue is on server1 (stopped after the connection is set up); B's queue and the new queue are on server2.
+-- Fast rotation completes over the live server: B secures the new queue and sends the confirmation and QEND on it,
+-- so the recipient moves to it and removes the old queue without the old server.
+testFastSwitchDeadOldServer :: HasCallStack => (ASrvTransport, AStoreType) -> IO ()
+testFastSwitchDeadOldServer ps@(t, ASType qsType _) = do
+  let bServers = initAgentServers {smp = userServers [testSMPServer2]}
+  withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \_ ->
+    withAgent 1 agentCfg initAgentServers testDB $ \a ->
+      withAgent 2 agentCfg bServers testDB2 $ \b -> do
+        (aId, bId) <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do
+          (aId, bId) <- makeConnection a b
+          exchangeGreetings a bId b aId
+          -- create the rotated queue on the live server
+          liftIO $ setProtocolServers a 1 [noAuthSrvCfg testSMPServer2]
+          pure (aId, bId)
+        nGet a =##> \case ("", "", DOWN _ cs) -> bId `elem` cs; _ -> False
+        runRight_ $ do
+          -- a message queued while the old server is down must survive the rotation and arrive on the new queue
+          _ <- sendMessage b aId SMP.noMsgFlags "queued while down"
+          _ <- switchConnectionAsync a "" bId
+          queuedReceived <- drainSwitchCompletedRcvMsg a bId "queued while down"
+          liftIO $ queuedReceived `shouldBe` True
+          drainSwitchCompleted b aId QDSnd
+          exchangeGreetingsMsgId 7 a bId b aId
+
+-- drains switch and network events until the connection reports SPCompleted in the given direction,
+-- tolerating DOWN/UP and intermediate phases (the old server is stopped mid-rotation)
+drainSwitchCompleted :: AgentClient -> ByteString -> QueueDirection -> ExceptT AgentErrorType IO ()
+drainSwitchCompleted c connId d =
+  pGet c >>= \case
+    (_, connId', AEvt SAEConn (SWITCH d' SPCompleted _)) | connId' == connId && d' == d -> pure ()
+    _ -> drainSwitchCompleted c connId d
+
+-- like drainSwitchCompleted for QDRcv, additionally acking and reporting a message matching the body seen while draining
+drainSwitchCompletedRcvMsg :: AgentClient -> ByteString -> MsgBody -> ExceptT AgentErrorType IO Bool
+drainSwitchCompletedRcvMsg c connId body = go False
+  where
+    go seen =
+      pGet c >>= \case
+        (_, connId', AEvt SAEConn (SWITCH QDRcv SPCompleted _)) | connId' == connId -> pure seen
+        (_, connId', AEvt SAEConn (Msg' mId _ body')) | connId' == connId && body' == body -> ackMessage c connId' mId Nothing >> go True
+        _ -> go seen
+
 phaseRcv :: AgentClient -> ByteString -> SwitchPhase -> [Maybe RcvSwitchStatus] -> ExceptT AgentErrorType IO ()
 phaseRcv c connId p swchStatuses = phase c connId QDRcv p (\stats -> rcvSwchStatuses' stats `shouldMatchList` swchStatuses)
 
@@ -3599,9 +3671,9 @@ testSwitchAsync servers = do
     testFullSwitch a bId b aId 14
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent 1 agentCfg servers testDB
+    withA = withAgent 1 agentCfgV7 servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent 2 agentCfg servers testDB2
+    withB = withAgent 2 agentCfgV7 servers testDB2
 
 withAgent :: HasCallStack => Int -> AgentConfig -> InitialAgentServers -> String -> (HasCallStack => AgentClient -> IO a) -> IO a
 withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) (\a -> disposeAgentClient a >> threadDelay 100000)
@@ -3617,7 +3689,7 @@ sessionSubscribe withC connIds a =
 
 testSwitchDelete :: InitialAgentServers -> IO ()
 testSwitchDelete servers =
-  withAgentClientsCfgServers2 agentCfg agentCfg servers $ \a b -> runRight_ $ do
+  withAgentClientsCfgServers2 agentCfgV7 agentCfgV7 servers $ \a b -> runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetings a bId b aId
     liftIO $ disposeAgentClient b
@@ -3675,9 +3747,9 @@ testAbortSwitchStarted servers = do
     testFullSwitch a bId b aId 16
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent 1 agentCfg servers testDB
+    withA = withAgent 1 agentCfgV7 servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent 2 agentCfg servers testDB2
+    withB = withAgent 2 agentCfgV7 servers testDB2
 
 testAbortSwitchStartedReinitiate :: HasCallStack => InitialAgentServers -> IO ()
 testAbortSwitchStartedReinitiate servers = do
@@ -3726,9 +3798,9 @@ testAbortSwitchStartedReinitiate servers = do
     testFullSwitch a bId b aId 16
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent 1 agentCfg servers testDB
+    withA = withAgent 1 agentCfgV7 servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent 2 agentCfg servers testDB2
+    withB = withAgent 2 agentCfgV7 servers testDB2
 
 switchPhaseRcvP :: ConnId -> SwitchPhase -> [Maybe RcvSwitchStatus] -> ATransmission -> Bool
 switchPhaseRcvP cId sphase swchStatuses = switchPhaseP cId QDRcv sphase (\stats -> rcvSwchStatuses' stats == swchStatuses)
@@ -3780,9 +3852,9 @@ testCannotAbortSwitchSecured servers = do
     testFullSwitch a bId b aId 14
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent 1 agentCfg servers testDB
+    withA = withAgent 1 agentCfgV7 servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent 2 agentCfg servers testDB2
+    withB = withAgent 2 agentCfgV7 servers testDB2
 
 testSwitch2Connections :: HasCallStack => InitialAgentServers -> IO ()
 testSwitch2Connections servers = do
@@ -3838,9 +3910,9 @@ testSwitch2Connections servers = do
     testFullSwitch a bId2 b aId2 14
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent 1 agentCfg servers testDB
+    withA = withAgent 1 agentCfgV7 servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent 2 agentCfg servers testDB2
+    withB = withAgent 2 agentCfgV7 servers testDB2
 
 testSwitch2ConnectionsAbort1 :: HasCallStack => InitialAgentServers -> IO ()
 testSwitch2ConnectionsAbort1 servers = do
@@ -3891,9 +3963,9 @@ testSwitch2ConnectionsAbort1 servers = do
     testFullSwitch a bId2 b aId2 12
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent 1 agentCfg servers testDB
+    withA = withAgent 1 agentCfgV7 servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent 2 agentCfg servers testDB2
+    withB = withAgent 2 agentCfgV7 servers testDB2
 
 testCreateQueueAuth :: HasCallStack => (Maybe BasicAuth, VersionSMP) -> (Maybe BasicAuth, VersionSMP) -> SndQueueSecured -> AgentMsgId -> IO Int
 testCreateQueueAuth clnt1 clnt2 sqSecured baseId = do

tests/AgentTests/NotificationTests.hs

@@ -17,7 +17,8 @@ module AgentTests.NotificationTests where
 
 -- import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging)
 import AgentTests.FunctionalAPITests
-  ( agentCfgVPrevPQ,
+  ( agentCfgV7,
+    agentCfgVPrevPQ,
     createConnection,
     exchangeGreetings,
     get,
@@ -28,6 +29,7 @@ import AgentTests.FunctionalAPITests
     runRight_,
     sendMessage,
     switchComplete,
+    fastSwitchComplete,
     testServerMatrix2,
     withAgent,
     withAgentClients2,
@@ -164,10 +166,14 @@ notificationTests ps@(t, _) = do
     it "should resume batched subscriptions after SMP server is restarted" $
       withAPNSMockServer $ \apns ->
         withNtfServer t $ testNotificationsSMPRestartBatch 50 ps apns
-  describe "should switch notifications to the new queue" $
+  describe "should switch notifications to the new queue (slow rotation)" $
     testServerMatrix2 ps $ \servers ->
       withAPNSMockServer $ \apns ->
-        withNtfServer t $ testSwitchNotifications servers apns
+        withNtfServer t $ testSwitchNotifications agentCfgV7 switchComplete servers apns
+  describe "should switch notifications to the new queue (fast rotation)" $
+    testServerMatrix2 ps $ \servers ->
+      withAPNSMockServer $ \apns ->
+        withNtfServer t $ testSwitchNotifications agentCfg fastSwitchComplete servers apns
   it "should keep sending notifications for old token" $
     withSmpServer ps $
       withAPNSMockServer $ \apns ->
@@ -868,9 +874,9 @@ testNotificationsSMPRestartBatch n ps@(t, ASType qsType _) apns =
         killThread t1
         pure res
 
-testSwitchNotifications :: InitialAgentServers -> APNSMockServer -> IO ()
-testSwitchNotifications servers apns =
-  withAgentClientsCfgServers2 agentCfg agentCfg servers $ \a b -> runRight_ $ do
+testSwitchNotifications :: AgentConfig -> (AgentClient -> ByteString -> AgentClient -> ByteString -> ExceptT AgentErrorType IO ()) -> InitialAgentServers -> APNSMockServer -> IO ()
+testSwitchNotifications cfg completeSwitch servers apns =
+  withAgentClientsCfgServers2 cfg cfg servers $ \a b -> runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetings a bId b aId
     _ <- registerTestToken a "abcd" NMInstant apns
@@ -883,7 +889,7 @@ testSwitchNotifications servers apns =
           ackMessage a bId msgId Nothing
     testMessage "hello"
     _ <- switchConnectionAsync a "" bId
-    switchComplete a bId b aId
+    completeSwitch a bId b aId
     liftIO $ threadDelay 500000
     testMessage "hello again"