diff --git a/plans/2026-08-09-fast-queue-rotation-plan.md b/plans/2026-08-09-fast-queue-rotation-plan.md new file mode 100644 index 0000000000..2af69ba5ae --- /dev/null +++ b/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. diff --git a/protocol/diagrams/duplex-messaging/queue-rotation-fast.mmd b/protocol/diagrams/duplex-messaging/queue-rotation-fast.mmd index 75887dd0bd..952729f4ac 100644 --- a/protocol/diagrams/duplex-messaging/queue-rotation-fast.mmd +++ b/protocol/diagrams/duplex-messaging/queue-rotation-fast.mmd @@ -5,13 +5,18 @@ sequenceDiagram participant S as Server
that has A's send queue
(B's receive queue) participant B as Bob - A ->> R': NEW: create new queue
(allow SKEY) - A ->> S: SEND: QADD (R'): send address
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 - \ No newline at end of file + 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 diff --git a/protocol/diagrams/duplex-messaging/queue-rotation-fast.svg b/protocol/diagrams/duplex-messaging/queue-rotation-fast.svg index 8230748235..77b986d75f 100644 --- a/protocol/diagrams/duplex-messaging/queue-rotation-fast.svg +++ b/protocol/diagrams/duplex-messaging/queue-rotation-fast.svg @@ -1,3 +1,3 @@ -BobServerthat has A's send queue(B's receive queue)New serverthat has the new A'sreceive queueCurrent serverthat has A'sreceive queueAliceBobServerthat has A's send queue(B's receive queue)New serverthat has the new A'sreceive queueCurrent serverthat has A'sreceive queueAliceNEW: create new queue(allow SKEY)SEND: QADD (R'): send addressof the new queue(s)MSG: QADD (R')SKEY: secure new queueSEND: QTESTMSG: QTESTDEL: delete the old queueSEND: send messages to the new queueMSG: receive messages from the new queue \ No newline at end of file +BobServerthat has A's send queue(B's receive queue)New serverthat has the new A'sreceive queueCurrent serverthat has A'sreceive queueAliceBobServerthat has A's send queue(B's receive queue)New serverthat has the new A'sreceive queueCurrent serverthat has A'sreceive queueAliceNEW: create new queue (SKEY allowed)SEND: QADD (R')MSG: QADD (R')SEND: messages (also scheduled on R')MSG: messagesSKEY: authorize B as senderSEND: confirmation (establishes R' secret)MSG: confirmation (A secures R')SEND: held copies (deduped), then new messagesMSG: messagesSEND: remaining tail, then QENDMSG: QENDSEND: QENDMSG: QENDDEL: delete the current queue \ No newline at end of file diff --git a/rfcs/2026-08-09-fast-queue-rotation.md b/rfcs/2026-08-09-fast-queue-rotation.md new file mode 100644 index 0000000000..d5cb3cf555 --- /dev/null +++ b/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. diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 371e1fac78..979704fd39 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/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 :" <> 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 :" <> 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 :" <> 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 diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index cfdb94a6bf..78630b9cb7 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index 371fae7988..607e6d1da2 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/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_ diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index dacd3c602d..04fbcf7296 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/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") $ diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 8e6f1a08cc..d831cdea77 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/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 diff --git a/tests/AgentTests/NotificationTests.hs b/tests/AgentTests/NotificationTests.hs index ed6455deba..d81c08a413 100644 --- a/tests/AgentTests/NotificationTests.hs +++ b/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"