From e24d537eaaa32230eec1fefcd957d474d7b8d4d5 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Thu, 8 Oct 2026 16:03:06 +0100 Subject: [PATCH] agent: fix race when switching reception to the new queue (#1933) * agent: fix race when switching reception to the new queue * return bool --------- Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com> --- ...026-10-08-fast-rotation-old-queue-drain.md | 58 +++++++++++++++ src/Simplex/Messaging/Agent.hs | 54 ++++++++++---- src/Simplex/Messaging/Agent/Client.hs | 2 +- src/Simplex/Messaging/Agent/Protocol.hs | 3 + src/Simplex/Messaging/Agent/Store.hs | 1 + .../Messaging/Agent/Store/AgentStore.hs | 5 ++ src/Simplex/Messaging/Client.hs | 9 ++- tests/AgentTests/FunctionalAPITests.hs | 72 ++++++++++++++++++- 8 files changed, 184 insertions(+), 20 deletions(-) create mode 100644 plans/2026-10-08-fast-rotation-old-queue-drain.md diff --git a/plans/2026-10-08-fast-rotation-old-queue-drain.md b/plans/2026-10-08-fast-rotation-old-queue-drain.md new file mode 100644 index 000000000..8cf3413cf --- /dev/null +++ b/plans/2026-10-08-fast-rotation-old-queue-drain.md @@ -0,0 +1,58 @@ +# Fast rotation: draining the old receive queue + +Recipient only. The sender is unchanged. + +## Decisions + +- New `RcvSwitchStatus` value: + + | constructor | encoding | + |------------------|-----------------| + | `RSReceivedQEND` | `received_qend` | + + - Stored in `rcv_queues.switch_status`. + - `canAbortRcvSwitch` returns `False` for it. +- `qEndMsg`, by the queue that delivered `QEND`: + - the queue named in `QEND`: the named queue is marked deleted, `ICDeleteRcvQueue` is enqueued. + - the new queue: the named queue is set to `RSReceivedQEND` and stays subscribed. + - `NSCCreate` and `SWITCH QDRcv SPCompleted`: sent when the named queue had no `RSReceivedQEND`. +- Deletion of an `RSReceivedQEND` queue: + - trigger: `OK` response to an `ACK` on that queue, without a `GET` lock on it. + - action: the queue is marked deleted, `ICDeleteRcvQueue` is enqueued. +- Bound, a `cleanupManager` step, for each connection with an `RSReceivedQEND` queue: + - condition: another receive queue of the connection has an active subscription. + - `delete_errors + 1 < deleteErrorCount`: `delete_errors` is incremented. + - otherwise: the queue is deleted as above; the first `DEL` error of `ICDeleteRcvQueue` removes it locally. +- `ackSMPMessage` result, `type QueueDrained = Bool`: + - `OK`: `True`. + - `MSG`: `False`; the message is written to `msgQ`. +- Apps: `received_qend` is added to `RcvSwitchStatus`. + +## Mechanics + +- `Agent/Protocol.hs`: `RSReceivedQEND`. +- `Agent/Store.hs`: `canAbortRcvSwitch`. +- `Client.hs`: `QueueDrained`; `ackSMPMessage` returns it. +- `Agent/Client.hs`: `sendAck` returns `QueueDrained`. +- `Agent/Store/AgentStore.hs`: `getEndedRcvQueueConnIds`: + + ```sql + SELECT conn_id FROM rcv_queues WHERE switch_status = ? AND deleted = 0 + ``` + +- `Agent.hs`: + - `deleteRcvQueueAsync`: `setRcvQueueDeleted`, then `ICDeleteRcvQueue`. + - `qEndMsg`. + - `ackQueueMessage`. + - `cleanupManager`: `expireEndedRcvQueues`. +- Apps: + - `apps/ios/SimpleXChat/APITypes.swift` + - `apps/multiplatform/common/src/commonMain/kotlin/chat/simplex/common/model/SimpleXAPI.kt` + +## Tests + +- `QEND` on the new queue while a message waits on the old queue: + - the message is delivered; + - the old queue is deleted after the `ACK` that returns `OK`. +- The same with a recipient restart before the `ACK`. +- Old server stopped: the old queue is deleted by the bound. diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 833815f36..4d6913411 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -2893,22 +2893,27 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni _ -> throwE $ CMD PROHIBITED "synchronizeRatchet: not duplex" ackQueueMessage :: AgentClient -> RcvQueue -> SMP.MsgId -> AM (Maybe ATransmission) -ackQueueMessage c rq@RcvQueue {userId, connId, server} srvMsgId = do +ackQueueMessage c rq@RcvQueue {userId, connId, server, rcvSwchStatus} srvMsgId = do atomically $ incSMPServerStat c userId server ackAttempts tryAllErrors (sendAck c rq srvMsgId) >>= \case - Right _ -> sendMsgNtf ackMsgs - Left (SMP _ SMP.NO_MSG) -> sendMsgNtf ackNoMsgErrs + Right drained -> sendMsgNtf ackMsgs $ drained && rcvSwchStatus == Just RSReceivedQEND + Left (SMP _ SMP.NO_MSG) -> sendMsgNtf ackNoMsgErrs False Left e -> do unless (temporaryOrHostError e) $ atomically $ incSMPServerStat c userId server ackOtherErrs throwE e where - sendMsgNtf stat = do + sendMsgNtf stat ended = do atomically $ incSMPServerStat c userId server stat ifM (liftIO $ hasGetLock c rq) (do atomically $ releaseGetLock c rq brokerTs_ <- eitherToMaybe <$> tryAllErrors (withStore c $ \db -> getRcvMsgBrokerTs db connId srvMsgId) pure $ Just ("", connId, AEvt SAEConn $ MSGNTF srvMsgId brokerTs_)) - (pure Nothing) + (Nothing <$ when ended (deleteRcvQueueAsync c rq)) + +deleteRcvQueueAsync :: AgentClient -> RcvQueue -> AM () +deleteRcvQueueAsync c rq@RcvQueue {connId, server, rcvId} = do + withStore' c (`setRcvQueueDeleted` rq) + enqueueCommand c "" connId (Just server) $ AInternalCommand $ ICDeleteRcvQueue rcvId -- | Suspend SMP agent connection (OFF command) in Reader monad suspendConnection' :: AgentClient -> NetworkRequestMode -> ConnId -> AM () @@ -3438,6 +3443,7 @@ cleanupManager c@AgentClient {subQ} = do run SFERR deleteSndFilesPrefixPaths run SFERR deleteExpiredReplicasForDeletion run ERR deleteExpiredServiceReqs + run ERR expireEndedRcvQueues liftIO $ threadDelay' int where run :: forall e. AEntityI e => (AgentErrorType -> AEvent e) -> AM () -> AM' () @@ -3495,6 +3501,20 @@ cleanupManager c@AgentClient {subQ} = do deleteExpiredServiceRequests db $ addUTCTime (negate serviceResponseTimeout) now getExpiredServiceConns db now deleteConnectionsAsync' c False expiredConns + expireEndedRcvQueues = do + maxErrs <- asks $ deleteErrorCount . config + connIds <- withStore' c getEndedRcvQueueConnIds + forM_ connIds $ \connId -> flip catchAllErrors (notify connId . ERR) $ + withConnLock c connId "cleanupManager" $ + withStore c (`getConn` connId) >>= \case + SomeConn _ (DuplexConnection _ rqs _) -> do + let (ended, current) = L.partition ((Just RSReceivedQEND ==) . rcvSwchStatus) rqs + whenM (anyM $ map (atomically . hasActiveSubscription c) current) $ + forM_ ended $ \rq -> + if deleteErrors rq + 1 < maxErrs + then withStore' c (`incRcvDeleteErrors` rq) + else deleteRcvQueueAsync c rq + _ -> pure () notify :: forall e. AEntityI e => AEntityId -> AEvent e -> AM () notify entId cmd = atomically $ writeTBQueue subQ ("", entId, AEvt (sAEntity @e) cmd) @@ -4094,19 +4114,23 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), THandlePar -- processed by queue recipient qEndMsg :: SMP.MsgId -> NonEmpty SndQAddr -> Connection 'CDuplex -> AM () qEndMsg srvMsgId addrs (DuplexConnection cData'@ConnData {enableNtfs} rqs sqs) = - case L.partition (\rq' -> any (`sameQAddress` sndAddress rq') addrs) rqs of + case L.partition isRemoved rqs of (removed@(_ : _), keptRq : keptRqs) -> do logServer "<--" c srv rId $ "MSG :" <> logSecret' (SMP.unMsgId 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 + let completed = any ((Just RSReceivedQEND ==) . rcvSwchStatus) removed + if isRemoved rq + then mapM_ (deleteRcvQueueAsync c) removed + else withStore' c $ \db -> forM_ removed $ \rq' -> setRcvSwitchStatus db rq' $ Just RSReceivedQEND + unless completed $ do + when enableNtfs $ do + ns <- asks ntfSupervisor + liftIO $ sendNtfSubCommand ns (NSCCreate, [connId]) + let conn' = DuplexConnection cData' (keptRq :| keptRqs) sqs + cStats <- connectionStats c conn' + notify $ SWITCH QDRcv SPCompleted cStats _ -> pure () + where + isRemoved rq' = any (`sameQAddress` sndAddress rq') addrs qError :: String -> AM a qError = throwE . AGENT . A_QUEUE diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 766298948..08eb64ea9 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -2080,7 +2080,7 @@ disableQueuesNtfs c = sendTSessionBatches "NDEL" (mkSMPTSession . snd) disableQu queueCreds :: DisableQueueNtfReq -> (SMP.RecipientId, SMP.RcvPrivateAuthKey) queueCreds (_, RcvQueue {rcvPrivateKey, rcvId}) = (rcvId, rcvPrivateKey) -sendAck :: AgentClient -> RcvQueue -> MsgId -> AM () +sendAck :: AgentClient -> RcvQueue -> MsgId -> AM QueueDrained sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId = withSMPClient c NRMBackground rq ("ACK:" <> logSecret' (SMP.unMsgId msgId)) $ \smp -> ackSMPMessage smp rcvPrivateKey rcvId msgId diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index 5f7977e45..3ea9f3b1a 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -612,6 +612,7 @@ data RcvSwitchStatus | RSSendingQADD | RSSendingQUSE | RSReceivedMessage + | RSReceivedQEND deriving (Eq, Show) instance StrEncoding RcvSwitchStatus where @@ -620,12 +621,14 @@ instance StrEncoding RcvSwitchStatus where RSSendingQADD -> "sending_qadd" RSSendingQUSE -> "sending_quse" RSReceivedMessage -> "received_message" + RSReceivedQEND -> "received_qend" strP = A.takeTill (== ' ') >>= \case "switch_started" -> pure RSSwitchStarted "sending_qadd" -> pure RSSendingQADD "sending_quse" -> pure RSSendingQUSE "received_message" -> pure RSReceivedMessage + "received_qend" -> pure RSReceivedQEND _ -> fail "bad RcvSwitchStatus" instance ToField RcvSwitchStatus where toField = toField . decodeLatin1 . strEncode diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index e75cdb05e..67415c718 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/src/Simplex/Messaging/Agent/Store.hs @@ -219,6 +219,7 @@ canAbortRcvSwitch ConnData {connAgentVersion} = maybe False canAbort . rcvSwchSt -- if switch is in RSReceivedMessage status, aborting switch (deleting new queue) -- will break the connection because the sender would have original queue deleted RSReceivedMessage -> False + RSReceivedQEND -> False data ClientNtfCreds = ClientNtfCreds { -- | key pair to be used by the notification server to authorize transmissions diff --git a/src/Simplex/Messaging/Agent/Store/AgentStore.hs b/src/Simplex/Messaging/Agent/Store/AgentStore.hs index 2816d4615..2d3c7b3f2 100644 --- a/src/Simplex/Messaging/Agent/Store/AgentStore.hs +++ b/src/Simplex/Messaging/Agent/Store/AgentStore.hs @@ -100,6 +100,7 @@ module Simplex.Messaging.Agent.Store.AgentStore setSndQueuePrimary, deleteConnRcvQueue, incRcvDeleteErrors, + getEndedRcvQueueConnIds, deleteConnSndQueue, getPrimaryRcvQueue, getRcvQueue, @@ -763,6 +764,10 @@ incRcvDeleteErrors :: DB.Connection -> RcvQueue -> IO () incRcvDeleteErrors db RcvQueue {connId, dbQueueId} = DB.execute db "UPDATE rcv_queues SET delete_errors = delete_errors + 1 WHERE conn_id = ? AND rcv_queue_id = ?" (connId, dbQueueId) +getEndedRcvQueueConnIds :: DB.Connection -> IO [ConnId] +getEndedRcvQueueConnIds db = + map fromOnly <$> DB.query db "SELECT conn_id FROM rcv_queues WHERE switch_status = ? AND deleted = 0" (Only RSReceivedQEND) + deleteConnRcvQueue :: DB.Connection -> RcvQueue -> IO () deleteConnRcvQueue db RcvQueue {connId, dbQueueId} = DB.execute db "DELETE FROM rcv_queues WHERE conn_id = ? AND rcv_queue_id = ?" (connId, dbQueueId) diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 87b4319e7..9989a96eb 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -67,6 +67,7 @@ module Simplex.Messaging.Client enableSMPQueuesNtfs, disableSMPQueuesNtfs, sendSMPMessage, + QueueDrained, ackSMPMessage, suspendSMPQueue, deleteSMPQueue, @@ -1087,15 +1088,17 @@ resolvedNameOrNotFound d NameResponse {registration} = case registration of NRRegistered {nameRecord} -> T.toLower (nrName nameRecord) == fullDomainName d _ -> True +type QueueDrained = Bool + -- | Acknowledge message delivery (server deletes the message). -- -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#acknowledge-message-delivery -- This command is always sent in background request mode -ackSMPMessage :: SMPClient -> RcvPrivateAuthKey -> QueueId -> MsgId -> ExceptT SMPClientError IO () +ackSMPMessage :: SMPClient -> RcvPrivateAuthKey -> QueueId -> MsgId -> ExceptT SMPClientError IO QueueDrained ackSMPMessage c rpKey rId msgId = sendSMPCommand c NRMBackground (Just rpKey) rId (ACK msgId) >>= \case - OK -> return () - cmd@MSG {} -> liftIO $ writeSMPMessage c rId cmd + OK -> pure True + cmd@MSG {} -> liftIO (writeSMPMessage c rId cmd) $> False r -> throwE $ unexpectedResponse r -- | Irreversibly suspend SMP queue. diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 1d6f0c121..bdc37f6e2 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -550,6 +550,10 @@ functionalAPITests ps = do testServerMatrix2 ps testFastSwitchConnection it "should switch delivery to the new queue when the old server is down" $ testFastSwitchDeadOldServer ps + it "should drain the old queue after QEND on the new queue" $ + testFastSwitchDrainOldQueue ps + it "should drain the old queue after QEND on the new queue and restart" $ + testFastSwitchDrainOldQueueRestart ps describe "should switch delivery to the new queue" $ testServerMatrix2 ps testSwitchConnection describe "should switch to new queue asynchronously" $ @@ -3729,7 +3733,7 @@ testFastSwitchDeadOldServer :: HasCallStack => (ASrvTransport, AStoreType) -> IO testFastSwitchDeadOldServer ps@(t, ASType qsType _) = do let bServers = initAgentServers {smp = userServers [testSMPServer2]} withSmpServerConfigOn t (cfgJ2QS qsType) testPort2 $ \_ -> - withAgent 1 agentCfg initAgentServers testDB $ \a -> + withAgent 1 aCfg initAgentServers testDB $ \a -> withAgent 2 agentCfg bServers testDB2 $ \b -> do (aId, bId) <- withSmpServerStoreLogOn ps testPort $ \_ -> runRight $ do (aId, bId) <- makeConnection a b @@ -3745,7 +3749,10 @@ testFastSwitchDeadOldServer ps@(t, ASType qsType _) = do queuedReceived <- drainSwitchCompletedRcvMsg a bId "queued while down" liftIO $ queuedReceived `shouldBe` True drainSwitchCompleted b aId QDSnd + waitOneRcvQueue a bId exchangeGreetingsMsgId 7 a bId b aId + where + aCfg = agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} -- drains switch and network events until the connection reports SPCompleted in the given direction, -- tolerating DOWN/UP and intermediate phases (the old server is stopped mid-rotation) @@ -3765,6 +3772,69 @@ drainSwitchCompletedRcvMsg c connId body = go False (_, connId', AEvt SAEConn (Msg' mId _ body')) | connId' == connId && body' == body -> ackMessage c connId' mId Nothing >> go True _ -> go seen +testFastSwitchDrainOldQueue :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testFastSwitchDrainOldQueue ps = + withSmpServers2 ps $ withAgentClients2 $ \a b -> runRight_ $ do + (aId, bId, aM1Id) <- endSwitchOnNewQueue a b + ackMessage a bId aM1Id Nothing + drainOldQueue a bId b aId + +testFastSwitchDrainOldQueueRestart :: HasCallStack => (ASrvTransport, AStoreType) -> IO () +testFastSwitchDrainOldQueueRestart ps = + withSmpServers2 ps $ + withAgent 2 agentCfg initAgentServers testDB2 $ \b -> do + (aId, bId, _) <- withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight $ endSwitchOnNewQueue a b + withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight_ $ do + subscribeConnection a bId + drainOldQueue a bId b aId + +endSwitchOnNewQueue :: HasCallStack => AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId, AgentMsgId) +endSwitchOnNewQueue a b = do + (aId, bId) <- makeConnection a b + exchangeGreetings a bId b aId + liftIO $ setProtocolServers a 1 [noAuthSrvCfg testSMPServer2] + m1Id <- sendMessage b aId SMP.noMsgFlags "m1" + get b ##> ("", aId, SENT m1Id) + ("", c, Msg' aM1Id _ "m1") <- get a + liftIO $ c `shouldBe` bId + m2Id <- sendMessage b aId SMP.noMsgFlags "m2" + get b ##> ("", aId, SENT m2Id) + _ <- switchConnectionAsync a "" bId + fastSwitchComplete a bId b aId + stats <- getConnectionServers a bId + liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Nothing, Just RSReceivedQEND] + pure (aId, bId, aM1Id) + +drainOldQueue :: HasCallStack => AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO () +drainOldQueue a bId b aId = do + ("", c, MSG MsgMeta {recipient = (aM2Id, _)} _ "m2") <- get a + liftIO $ c `shouldBe` bId + ackMessage a bId aM2Id Nothing + waitOneRcvQueue a bId + bMsgId <- sendMessage b aId SMP.noMsgFlags "hello" + get b ##> ("", aId, SENT bMsgId) + ("", c', MSG MsgMeta {recipient = (aMsgId, _)} _ "hello") <- get a + liftIO $ c' `shouldBe` bId + ackMessage a bId aMsgId Nothing + aMsgId' <- sendMessage a bId SMP.noMsgFlags "hello too" + get a ##> ("", bId, SENT aMsgId') + ("", c'', MSG MsgMeta {recipient = (bMsgId', _)} _ "hello too") <- get b + liftIO $ c'' `shouldBe` aId + ackMessage b aId bMsgId' Nothing + liftIO $ noMessages a "nothing else should be delivered to alice" + liftIO $ noMessages b "nothing else should be delivered to bob" + +waitOneRcvQueue :: HasCallStack => AgentClient -> ConnId -> ExceptT AgentErrorType IO () +waitOneRcvQueue c connId = go (300 :: Int) + where + go n = do + stats <- getConnectionServers c connId + case rcvSwchStatuses' stats of + [Nothing] -> pure () + statuses + | n > 0 -> liftIO (threadDelay 100000) >> go (n - 1) + | otherwise -> liftIO $ expectationFailure $ "old receive queue was not deleted: " <> show statuses + phaseRcv :: AgentClient -> ByteString -> SwitchPhase -> [Maybe RcvSwitchStatus] -> ExceptT AgentErrorType IO () phaseRcv c connId p swchStatuses = phase c connId QDRcv p (\stats -> rcvSwchStatuses' stats `shouldMatchList` swchStatuses)