fix tests

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

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

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 ""

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)