From a2137ee841bb597cea06601fe8de261d0db48466 Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Sat, 27 Jun 2026 16:57:04 +0000 Subject: [PATCH] fix tests --- src/Simplex/Messaging/Agent.hs | 32 +++++++------------ src/Simplex/Messaging/Agent/Client.hs | 4 +-- .../Messaging/Agent/NtfSubSupervisor.hs | 2 +- tests/AgentTests/FunctionalAPITests.hs | 28 +++++++++++----- 4 files changed, 34 insertions(+), 32 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 8df5291aa..87e910634 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -1862,14 +1862,10 @@ runCommandProcessing c connId server_ Worker {doWork} = do lift $ waitForWork doWork liftIO $ throwWhenInactive c atomically $ beginAgentOperation c AOSndNetwork - withWork c doWork (\db -> getPendingServerCommand db connId server_) $ runProcessCmd (riFast ri) + withWork c doWork (\db -> getPendingServerCommand db connId server_) $ processCmd (riFast ri) where - runProcessCmd ri cmd = do - pending <- newTVarIO [] - processCmd ri cmd pending - mapM_ (liftIO . notifyEvent c) . reverse =<< readTVarIO pending - processCmd :: RetryInterval -> PendingCommand -> TVar [ATransmission] -> AM () - processCmd ri PendingCommand {cmdId, corrId, userId, command} pendingCmds = case command of + processCmd :: RetryInterval -> PendingCommand -> AM () + processCmd ri PendingCommand {cmdId, corrId, userId, command} = case command of AClientCommand cmd -> case cmd of NEW enableNtfs (ACM cMode) pqEnc subMode -> noServer $ do triedHosts <- newTVarIO S.empty @@ -2023,7 +2019,7 @@ runCommandProcessing c connId server_ Worker {doWork} = do internalErr s = cmdError $ INTERNAL $ s <> ": " <> show (agentCommandTag command) cmdError e = notify (ERR e) >> withStore' c (`deleteCommand` cmdId) notify :: forall e. AEntityI e => AEvent e -> AM () - notify cmd = atomically $ modifyTVar' pendingCmds ((corrId, connId, AEvt (sAEntity @e) cmd) :) + notify cmd = liftIO $ notifyEvent c (corrId, connId, AEvt (sAEntity @e) cmd) -- ^ ^ ^ async command processing / enqueueMessages :: AgentClient -> ConnData -> NonEmpty SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, PQEncryption) @@ -3068,7 +3064,7 @@ processSMPTransmissions :: AgentClient -> ServerTransmissionBatch SMPVersion Err processSMPTransmissions c (tSess@(userId, srv, _), THandleParams {thAuth, sessionId = sessId}, ts) = do upConnIds <- newTVarIO [] serviceRQs <- newTVarIO ([] :: [RcvQueue]) - forM_ ts $ \(entId, t) -> case t of + forM_ ts $ \(entId, t) -> E.uninterruptibleMask_ $ case t of STEvent msgOrErr | entId == SMP.NoEntity -> case msgOrErr of Right msg -> case msg of @@ -3077,7 +3073,7 @@ processSMPTransmissions c (tSess@(userId, srv, _), THandleParams {thAuth, sessio _ -> logError $ "unexpected event: " <> tshow msg Left e -> notifyErr "" e | otherwise -> withRcvConn entId $ \rq@RcvQueue {connId} conn -> case msgOrErr of - Right msg -> runProcessSMP rq conn (toConnData conn) msg + Right msg -> processSMP rq conn (toConnData conn) msg Left e -> lift $ do processClientNotice rq e notifyErr connId e @@ -3088,11 +3084,11 @@ processSMPTransmissions c (tSess@(userId, srv, _), THandleParams {thAuth, sessio Right (SMP.SOK serviceId_) -> liftIO $ processSubOk rq upConnIds serviceRQs serviceId_ Right msg@SMP.MSG {} -> do liftIO $ processSubOk rq upConnIds serviceRQs Nothing -- the connection is UP even when processing this particular message fails - runProcessSMP rq conn (toConnData conn) msg + processSMP rq conn (toConnData conn) msg Right r -> lift $ processSubErr rq $ unexpectedResponse r Left e -> lift $ unless (temporaryClientError e) $ processSubErr rq e -- timeout/network was already reported SMP.ACK _ -> case respOrErr of - Right msg@SMP.MSG {} -> runProcessSMP rq conn (toConnData conn) msg + Right msg@SMP.MSG {} -> processSMP rq conn (toConnData conn) msg _ -> pure () -- TODO process OK response to ACK _ -> pure () -- TODO process expired response to DEL STResponse {} -> pure () -- TODO process expired responses to sent messages @@ -3141,18 +3137,12 @@ processSMPTransmissions c (tSess@(userId, srv, _), THandleParams {thAuth, sessio notify' connId msg = liftIO $ notifyEvent c ("", connId, AEvt (sAEntity @e) msg) notifyErr :: ConnId -> SMPClientError -> AM' () notifyErr connId = notify' connId . ERR . protocolClientError SMP (B.unpack $ strEncode srv) - runProcessSMP :: RcvQueue -> Connection c -> ConnData -> BrokerMsg -> AM () - runProcessSMP rq conn cData msg = do - pending <- newTVarIO [] - processSMP rq conn cData msg pending - mapM_ (liftIO . notifyEvent c) . reverse =<< readTVarIO pending - processSMP :: forall c. RcvQueue -> Connection c -> ConnData -> BrokerMsg -> TVar [ATransmission] -> AM () + processSMP :: forall c. RcvQueue -> Connection c -> ConnData -> BrokerMsg -> AM () processSMP rq@RcvQueue {rcvId = rId, queueMode, e2ePrivKey, e2eDhSecret, status, smpClientVersion = agreedClientVerion} conn cData@ConnData {connId, connAgentVersion = agreedAgentVersion, ratchetSyncState = rss} - smpMsg - pendingMsgs = + smpMsg = withConnLock c connId "processSMP" $ case smpMsg of SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} -> do atomically $ incSMPServerStat c userId srv recvMsgs @@ -3352,7 +3342,7 @@ processSMPTransmissions c (tSess@(userId, srv, _), THandleParams {thAuth, sessio notify :: forall e m. (AEntityI e, MonadIO m) => AEvent e -> m () notify = notify_ connId notify_ :: forall e m. (AEntityI e, MonadIO m) => ConnId -> AEvent e -> m () - notify_ connId' msg = atomically $ modifyTVar' pendingMsgs (("", connId', AEvt (sAEntity @e) msg) :) + notify_ connId' msg = liftIO $ notifyEvent c ("", connId', AEvt (sAEntity @e) msg) prohibited :: Text -> AM () prohibited s = do diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 3ba921e99..e357d45cc 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -834,7 +834,7 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do handleNotify = E.handleAny $ notifySub' c "" . ERR . INTERNAL . show notifySub' :: forall e m. (AEntityI e, MonadIO m) => AgentClient -> ConnId -> AEvent e -> m () -notifySub' c connId cmd = liftIO $ notifyEvent c (B.empty, connId, AEvt (sAEntity @e) cmd) +notifySub' c connId cmd = liftIO $ nonBlockingNotifyEvent c (B.empty, connId, AEvt (sAEntity @e) cmd) {-# INLINE notifySub' #-} notifySub :: MonadIO m => AgentClient -> AEvent 'AENone -> m () @@ -925,7 +925,7 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v = Right client -> do logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")" atomically $ putTMVar (sessionVar v) (Right client) - liftIO $ notifyEvent c ("", "", AEvt SAENone $ hostEvent CONNECT client) + liftIO $ nonBlockingNotifyEvent c ("", "", AEvt SAENone $ hostEvent CONNECT client) pure client Left e -> do ei <- asks $ persistErrorInterval . config diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index b66285acd..01c452392 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -504,7 +504,7 @@ workerInternalError c connId internalErrStr = do notifyInternalError :: MonadIO m => AgentClient -> ConnId -> String -> m () notifyInternalError c connId internalErrStr = do logError $ T.pack internalErrStr - liftIO $ notifyEvent c ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr) + liftIO $ nonBlockingNotifyEvent c ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr) notifyInternalError' :: MonadIO m => AgentClient -> String -> m () notifyInternalError' c = notifyInternalError c "" diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 489a7e70b..4f685baad 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -147,7 +147,10 @@ import Simplex.Messaging.Server.QueueStore.Types (QueueStoreClass (..)) data AgentClient = AgentClient { client :: A.AgentClient, - subQ :: TBQueue ATransmission + subQ :: TBQueue ATransmission, + -- DOWN/UP events skipped by a typed get' (e.g. get @'AEConn), kept in arrival order + -- so that a later nGet can still consume them instead of them being silently dropped. + skippedQ :: TVar [ATransmission] } type AEntityTransmission e = (ACorrId, ConnId, AEvent e) @@ -185,12 +188,20 @@ nGet :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AENo nGet c = withFrozenCallStack $ get' @'AENone c get' :: forall e m. (MonadIO m, AEntityI e, HasCallStack) => AgentClient -> m (AEntityTransmission e) -get' c = withFrozenCallStack $ do - t@(corrId, connId, AEvt e cmd) <- pGet c - case testEquality e (sAEntity @e) of - Just Refl -> pure (corrId, connId, cmd) - _ | skipEvent t -> get' c - | otherwise -> error $ "unexpected command " <> show cmd +get' c@AgentClient {skippedQ} = withFrozenCallStack $ do + -- return a previously skipped event if it matches the requested entity (used by nGet for DOWN/UP); + -- a typed get' never matches the skipped DOWN/UP, so they stay buffered for nGet + skipped <- atomically $ stateTVar skippedQ $ \case + t@(_, _, AEvt e _) : ts | isJust (testEquality e (sAEntity @e)) -> (Just t, ts) + ts -> (Nothing, ts) + case skipped of + Just (corrId, connId, AEvt e cmd) | Just Refl <- testEquality e (sAEntity @e) -> pure (corrId, connId, cmd) + _ -> do + t@(corrId, connId, AEvt e cmd) <- pGet c + case testEquality e (sAEntity @e) of + Just Refl -> pure (corrId, connId, cmd) + _ | skipEvent t -> atomically (modifyTVar' skippedQ (<> [t])) >> get' c + | otherwise -> error $ "unexpected command " <> show cmd where skipEvent (_, _, AEvt _ A.DOWN {}) = True skipEvent (_, _, AEvt _ A.UP {}) = True @@ -4431,10 +4442,11 @@ getSMPAgentClient' :: Int -> AgentConfig -> InitialAgentServers -> String -> IO getSMPAgentClient' clientId cfg' initServers dbPath = do Right st <- liftIO $ createStore dbPath subQ <- newTBQueueIO 1024 + skippedQ <- newTVarIO [] Right client <- runExceptT $ getSMPAgentClient_ clientId cfg' initServers st (\_ -> atomically . writeTBQueue subQ) when (dbNew st) $ insertUser st startSMPAgentClient client False - pure AgentClient {client, subQ} + pure AgentClient {client, subQ, skippedQ} #if defined(dbPostgres) createStore :: String -> IO (Either MigrationError DBStore)