bump mq

cabal.project

@@ -12,7 +12,7 @@ constraints: zip +disable-bzip2 +disable-zstd
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/simplexmq.git
-    tag: 93fd424f86086c6f378b50e343f32ec47f8b0c3f
+    tag: 757742f5ed4eacc6cd29021983b3b5961e40a6d8
 
 source-repository-package
     type: git

scripts/nix/sha256map.nix

@@ -1,5 +1,5 @@
 {
-  "https://github.com/simplex-chat/simplexmq.git"."93fd424f86086c6f378b50e343f32ec47f8b0c3f" = "1jijzj8k4pxv4ri76n1m1c576w0g78vl39z8x2mhyyfip7zcal4i";
+  "https://github.com/simplex-chat/simplexmq.git"."757742f5ed4eacc6cd29021983b3b5961e40a6d8" = "1wzihy98ahdmnp542bh7crs8sjbw2llgvhvy8n70rvcrlh00xsd4";
   "https://github.com/simplex-chat/hs-socks.git"."a30cc7a79a08d8108316094f8f2f82a0c5e1ac51" = "0yasvnr7g91k76mjkamvzab2kvlb1g5pspjyjn2fr6v83swjhj38";
   "https://github.com/simplex-chat/direct-sqlcipher.git"."f814ee68b16a9447fbb467ccc8f29bdd3546bfd9" = "1ql13f4kfwkbaq7nygkxgw84213i0zm7c1a8hwvramayxl38dq5d";
   "https://github.com/simplex-chat/sqlcipher-simple.git"."a46bd361a19376c5211f1058908fc0ae6bf42446" = "1z0r78d8f0812kxbgsm735qf6xx8lvaz27k1a0b4a2m0sshpd5gl";

src/Simplex/Chat.hs

@@ -91,14 +91,18 @@ import qualified Simplex.FileTransfer.Description as FD
 import Simplex.FileTransfer.Protocol (FileParty (..), FilePartyI)
 import Simplex.Messaging.Agent as Agent
 import Simplex.Messaging.Agent.Client (AgentStatsKey (..), SubInfo (..), agentClientStore, getAgentWorkersDetails, getAgentWorkersSummary, temporaryAgentError, withLockMap)
+import qualified Simplex.Messaging.Agent.Client as AC
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore, defaultAgentConfig)
 import Simplex.Messaging.Agent.Lock (withLock)
 import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Agent.Protocol as AP (AgentErrorType (..))
+import qualified Simplex.Messaging.Agent.Store as AS
 import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), MigrationError, SQLiteStore (dbNew), execSQL, upMigration, withConnection)
+import qualified Simplex.Messaging.Agent.Store.SQLite as ADB
 import Simplex.Messaging.Agent.Store.SQLite.DB (SlowQueryStats (..))
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import qualified Simplex.Messaging.Agent.TRcvQueues as RQ
 import Simplex.Messaging.Client (defaultNetworkConfig)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs (..))
@@ -227,6 +231,7 @@ newChatController
     inputQ <- newTBQueueIO tbqSize
     outputQ <- newTBQueueIO tbqSize
     connNetworkStatuses <- atomically TM.empty
+    agentDeliveryStatuses <- atomically TM.empty
     subscriptionMode <- newTVarIO SMSubscribe
     chatLock <- newEmptyTMVarIO
     entityLocks <- atomically TM.empty
@@ -263,6 +268,7 @@ newChatController
           inputQ,
           outputQ,
           connNetworkStatuses,
+          agentDeliveryStatuses,
           subscriptionMode,
           chatLock,
           entityLocks,
@@ -2149,6 +2155,59 @@ processChatCommand' vr = \case
     chatMigrations <- map upMigration <$> withStore' (Migrations.getCurrent . DB.conn)
     agentMigrations <- withAgent getAgentMigrations
     pure $ CRVersionInfo {versionInfo, chatMigrations, agentMigrations}
+  DebugDelivery showAll -> do
+    ads <- mapM readTVarIO =<< readTVarIO =<< asks agentDeliveryStatuses
+    let collect (acId, ds) = if not showAll && agentDeliveryOk ds then Nothing else Just (decodeLatin1 $ strEncode acId, ds)
+    pure $ CRDebugDelivery . M.fromList . mapMaybe collect $ M.toList ads
+  DebugConnection (Left connId) -> withUser $ \user -> do
+    Connection {agentConnId} <- withStore $ \db -> getConnectionById db vr user connId
+    processChatCommand $ DebugConnection (Right agentConnId)
+  DebugConnection (Right acId@(AgentConnId acId')) -> do
+    user@User {userId} <- withStore' (`getUserByAConnId` acId) >>= maybe (throwError . ChatErrorStore $ SEUserNotFoundByAConnId acId) pure
+    AS.RcvQueue {server, rcvId = rcvId', status = rqStatus} <- withAgent $ \ac ->
+      -- dive into agent internals
+      liftIOEither . (`runReaderT` agentEnv ac) . runExceptT $ AC.withStore ac $ \adb ->
+        ADB.getPrimaryRcvQueue adb acId'
+    let tSess = (userId, server, Just acId')
+        SMP.ProtocolServer {host, port} = server
+    AgentClient {activeSubs, pendingSubs, smpClients, smpSubWorkers} <- withAgent pure
+    inActive <- any (\AS.RcvQueue {rcvId} -> rcvId == rcvId') <$> atomically (RQ.getSessQueues tSess activeSubs)
+    inPending <- any (\AS.RcvQueue {rcvId} -> rcvId == rcvId') <$> atomically (RQ.getSessQueues tSess pendingSubs)
+    smpClient <- atomically (TM.lookup (tSess $> Nothing) smpClients) >>= mapM (\AC.SessionVar {sessionVar} -> atomically (tryReadTMVar sessionVar))
+    smpClientIsolated <- atomically (TM.lookup tSess smpClients) >>= mapM (\AC.SessionVar {sessionVar} -> atomically (tryReadTMVar sessionVar))
+    let smpClientStatus = case smpClient <|> smpClientIsolated of
+          Nothing -> "missing"
+          Just Nothing -> "connecting"
+          Just (Just (Left err)) -> tshow err
+          Just (Just (Right _client)) -> "connected"
+    subWorker <- atomically (TM.lookup (tSess $> Nothing) smpSubWorkers) >>= mapM (\AC.SessionVar {sessionVar} -> atomically (tryReadTMVar sessionVar))
+    subWorkerIsolated <- atomically (TM.lookup tSess smpSubWorkers) >>= mapM (\AC.SessionVar {sessionVar} -> atomically (tryReadTMVar sessionVar))
+    let subWorkerStatus = case subWorker <|> subWorkerIsolated of
+          Nothing -> "idle"
+          Just Nothing -> "waiting"
+          Just (Just _async) -> "working"
+    connection <-
+      withStore (\db -> getConnectionEntity db vr user acId) >>= \case
+        RcvDirectMsgConnection {entityConnection} -> pure entityConnection
+        RcvGroupMsgConnection {entityConnection} -> pure entityConnection
+        SndFileConnection {entityConnection} -> pure entityConnection
+        RcvFileConnection {entityConnection} -> pure entityConnection
+        UserContactConnection {entityConnection} -> pure entityConnection
+    deliveryStatus <- mapM readTVarIO . M.lookup acId =<< readTVarIO =<< asks agentDeliveryStatuses
+    networkStatus <- M.lookup acId <$> chatReadVar connNetworkStatuses
+    pure $
+      CRDebugConnection
+        DebugConnectionStatus
+          { deliveryStatus,
+            networkStatus,
+            inActive,
+            inPending,
+            server = (decodeLatin1 $ strEncode host, port),
+            smpClientStatus,
+            subWorkerStatus,
+            queueStatus = tshow rqStatus,
+            connection
+          }
   DebugLocks -> lift $ do
     chatLockName <- atomically . tryReadTMVar =<< asks chatLock
     chatEntityLocks <- getLocks =<< asks entityLocks
@@ -2185,6 +2244,7 @@ processChatCommand' vr = \case
             SubInfo {server, subError = Just e} -> M.alter (Just . maybe [e] (e :)) server m
             _ -> m
   GetAgentSubsDetails -> lift $ CRAgentSubsDetails <$> withAgent' getAgentSubscriptions
+  GetAgentSubsDiff -> lift $ CRAgentSubsDiff <$> withAgent' AC.diffSubscriptions
   -- CustomChatCommand is unsupported, it can be processed in preCmdHook
   -- in a modified CLI app or core - the hook should return Either ChatResponse ChatCommand
   CustomChatCommand _cmd -> withUser $ \user -> pure $ chatCmdError (Just user) "not supported"
@@ -2484,8 +2544,6 @@ processChatCommand' vr = \case
       toView $ CRNewChatItem user (AChatItem SCTDirect SMDSnd (DirectChat ct) ci)
       forM_ (timed_ >>= timedDeleteAt') $
         startProximateTimedItemThread user (ChatRef CTDirect contactId, chatItemId' ci)
-    drgRandomBytes :: Int -> CM ByteString
-    drgRandomBytes n = asks random >>= atomically . C.randomBytes n
     privateGetUser :: UserId -> CM User
     privateGetUser userId =
       tryChatError (withStore (`getUser` userId)) >>= \case
@@ -3004,7 +3062,7 @@ receiveFile' user ft rcvInline_ filePath_ = do
   where
     processError = \case
       -- TODO AChatItem in Cancelled events
-      ChatErrorAgent (SMP SMP.AUTH) _ -> pure $ CRRcvFileAcceptedSndCancelled user ft
+      ChatErrorAgent (SMP _ SMP.AUTH) _ -> pure $ CRRcvFileAcceptedSndCancelled user ft
       ChatErrorAgent (CONN DUPLICATE) _ -> pure $ CRRcvFileAcceptedSndCancelled user ft
       e -> throwError e
 
@@ -3341,7 +3399,7 @@ subscribeUserConnections vr onlyNeeded agentBatchSubscribe user = do
             errorNetworkStatus :: ChatError -> String
             errorNetworkStatus = \case
               ChatErrorAgent (BROKER _ NETWORK) _ -> "network"
-              ChatErrorAgent (SMP SMP.AUTH) _ -> "contact deleted"
+              ChatErrorAgent (SMP _ SMP.AUTH) _ -> "contact deleted"
               e -> show e
     -- TODO possibly below could be replaced with less noisy events for API
     contactLinkSubsToView :: Map ConnId (Either AgentErrorType ()) -> Map ConnId UserContact -> CM ()
@@ -3546,13 +3604,29 @@ processAgentMessage _ connId DEL_CONN =
 processAgentMessage _ "" (ERR e) =
   toView $ CRChatError Nothing $ ChatErrorAgent e Nothing
 processAgentMessage corrId connId msg = do
-  lockEntity <- critical (withStore (`getChatLockEntity` AgentConnId connId))
+  let acId = AgentConnId connId
+      acTag = aCommandTag msg
+  when (acTag == MSG_ || acTag == RCVD_) . lift $ trackNewDelivery acId msg
+  lockEntity <- critical (withStore (`getChatLockEntity` acId))
   withEntityLock "processAgentMessage" lockEntity $ do
     vr <- chatVersionRange
     -- getUserByAConnId never throws logical errors, only SEDBBusyError can be thrown here
-    critical (withStore' (`getUserByAConnId` AgentConnId connId)) >>= \case
+    critical (withStore' (`getUserByAConnId` acId)) >>= \case
       Just user -> processAgentMessageConn vr user corrId connId msg `catchChatError` (toView . CRChatError (Just user))
-      _ -> throwChatError $ CENoConnectionUser (AgentConnId connId)
+      _ -> throwChatError $ CENoConnectionUser acId
+
+-- TODO: clean up deliveries
+trackNewDelivery :: AgentConnId -> ACommand 'Agent 'AEConn -> CM' ()
+trackNewDelivery acId msg = do
+  now <- liftIO getCurrentTime
+  asks agentDeliveryStatuses >>= atomically . TM.alterF (updateConn now) acId
+  where
+    (isMSG, msgBodyPfx) = case msg of
+      MSG _ _ msgBody -> (True, T.take 1000 $ safeDecodeUtf8 msgBody)
+      _ -> (False, "")
+    updateConn lastCmd = \case
+      Nothing -> Just <$> newTVar AgentDeliveryStatus {lastCmd, tracking = "create", isMSG, connId = Nothing, msgBodyPfx, eventTag = Nothing, ackSent = Nothing, pendingAcks = M.empty}
+      Just v -> Just v <$ modifyTVar' v (\AgentDeliveryStatus {pendingAcks} -> AgentDeliveryStatus {lastCmd, tracking = "create", isMSG, connId = Nothing, msgBodyPfx, eventTag = Nothing, ackSent = Nothing, pendingAcks = M.filter not pendingAcks})
 
 -- CRITICAL error will be shown to the user as alert with restart button in Android/desktop apps.
 -- SEDBBusyError will only be thrown on IO exceptions or SQLError during DB queries,
@@ -3777,15 +3851,20 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
       _ -> toView $ CRSubscriptionEnd user entity
     MSGNTF smpMsgInfo -> toView $ CRNtfMessage user entity $ ntfMsgInfo smpMsgInfo
     _ -> case entity of
-      RcvDirectMsgConnection conn contact_ ->
+      RcvDirectMsgConnection conn contact_ -> do
+        storeDeliveryConn conn
         processDirectMessage agentMessage entity conn contact_
-      RcvGroupMsgConnection conn gInfo m ->
+      RcvGroupMsgConnection conn gInfo m -> do
+        storeDeliveryConn conn
         processGroupMessage agentMessage entity conn gInfo m
-      RcvFileConnection conn ft ->
+      RcvFileConnection conn ft -> do
+        storeDeliveryConn conn
         processRcvFileConn agentMessage entity conn ft
-      SndFileConnection conn ft ->
+      SndFileConnection conn ft -> do
+        storeDeliveryConn conn
         processSndFileConn agentMessage entity conn ft
-      UserContactConnection conn uc ->
+      UserContactConnection conn uc -> do
+        storeDeliveryConn conn
         processUserContactRequest agentMessage entity conn uc
   where
     updateConnStatus :: ConnectionEntity -> CM ConnectionEntity
@@ -3795,7 +3874,8 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
         withStore' $ \db -> updateConnectionStatus db conn connStatus
         pure $ updateEntityConnStatus acEntity connStatus
       Nothing -> pure acEntity
-
+    storeDeliveryConn :: Connection -> CM ()
+    storeDeliveryConn Connection {connId} = lift $ agentDeliveryStatus (AgentConnId agentConnId) "storeDeliveryConn" $ \ad -> ad {connId = Just connId}
     agentMsgConnStatus :: ACommand 'Agent e -> Maybe ConnStatus
     agentMsgConnStatus = \case
       CONF {} -> Just ConnRequested
@@ -3874,6 +3954,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
             (ct', conn') <- updateContactPQRcv user ct conn pqEncryption
             checkIntegrityCreateItem (CDDirectRcv ct') msgMeta `catchChatError` \_ -> pure ()
             (conn'', msg@RcvMessage {chatMsgEvent = ACME _ event}) <- saveDirectRcvMSG conn' msgMeta msgBody
+            lift $ agentDeliveryStatus (AgentConnId agentConnId) "eventTag" $ \ad -> ad {eventTag = Just $! tshow (toCMEventTag event)}
             let ct'' = ct' {activeConn = Just conn''} :: Contact
             assertDirectAllowed user MDRcv ct'' $ toCMEventTag event
             case event of
@@ -4277,11 +4358,17 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
                 _ -> messageWarning "sendXGrpMemCon: member category GCPreMember or GCPostMember is expected"
       MSG msgMeta _msgFlags msgBody -> do
         withAckMessage agentConnId msgMeta True $ do
+          lift $ agentDeliveryStatus (AgentConnId agentConnId) "group MSG" id
           checkIntegrityCreateItem (CDGroupRcv gInfo m) msgMeta `catchChatError` \_ -> pure ()
+          lift $ agentDeliveryStatus (AgentConnId agentConnId) "after checkIntegrityCreateItem" id
           forM_ aChatMsgs $ \case
-            Right (ACMsg _ chatMsg) ->
+            Right (ACMsg _ chatMsg) -> do
+              lift $ agentDeliveryStatus (AgentConnId agentConnId) "has event" id
               processEvent chatMsg `catchChatError` \e -> toView $ CRChatError (Just user) e
-            Left e -> toView $ CRChatError (Just user) (ChatError . CEException $ "error parsing chat message: " <> e)
+            Left e -> do
+              lift $ agentDeliveryStatus (AgentConnId agentConnId) "has error" id
+              toView $ CRChatError (Just user) (ChatError . CEException $ "error parsing chat message: " <> e)
+          lift $ agentDeliveryStatus (AgentConnId agentConnId) "after forM_" id
           forwardMsg_ `catchChatError` \_ -> pure ()
           checkSendRcpt $ rights aChatMsgs
         where
@@ -4289,7 +4376,9 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
           brokerTs = metaBrokerTs msgMeta
           processEvent :: MsgEncodingI e => ChatMessage e -> CM ()
           processEvent chatMsg = do
+            lift $ agentDeliveryStatus (AgentConnId agentConnId) "processEvent" id
             (m', conn', msg@RcvMessage {chatMsgEvent = ACME _ event}) <- saveGroupRcvMsg user groupId m conn msgMeta msgBody chatMsg
+            lift $ agentDeliveryStatus (AgentConnId agentConnId) "eventTag in processEvent" $ \ad -> ad {eventTag = Just $! tshow (toCMEventTag event)}
             case event of
               XMsgNew mc -> memberCanSend m' $ newGroupContentMessage gInfo m' mc msg brokerTs False
               XMsgFileDescr sharedMsgId fileDescr -> memberCanSend m' $ groupMessageFileDescription gInfo m' sharedMsgId fileDescr
@@ -4467,7 +4556,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
         MERR _ err -> do
           cancelSndFileTransfer user ft True >>= mapM_ (deleteAgentConnectionAsync user)
           case err of
-            SMP SMP.AUTH -> unless (fileStatus == FSCancelled) $ do
+            SMP _ SMP.AUTH -> unless (fileStatus == FSCancelled) $ do
               ci <- withStore $ \db -> do
                 liftIO (lookupChatRefByFileId db user fileId) >>= \case
                   Just (ChatRef CTDirect _) -> liftIO $ updateFileCancelled db user fileId CIFSSndCancelled
@@ -4622,7 +4711,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
     incAuthErrCounter :: ConnectionEntity -> Connection -> AgentErrorType -> CM ()
     incAuthErrCounter connEntity conn err = do
       case err of
-        SMP SMP.AUTH -> do
+        SMP _ SMP.AUTH -> do
           authErrCounter' <- withStore' $ \db -> incConnectionAuthErrCounter db user conn
           when (authErrCounter' >= authErrDisableCount) $ do
             toView $ CRConnectionDisabled connEntity
@@ -4633,6 +4722,10 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
     withCompletedCommand :: forall e. AEntityI e => Connection -> ACommand 'Agent e -> (CommandData -> CM ()) -> CM ()
     withCompletedCommand Connection {connId} agentMsg action = do
       let agentMsgTag = APCT (sAEntity @e) $ aCommandTag agentMsg
+      pending_ <- mapM readTVarIO =<< atomically . TM.lookup acId =<< asks agentDeliveryStatuses
+      if agentMsgTag == APCT SAEConn OK_ && corrId /= "" && maybe False (M.member ackKey . pendingAcks) pending_
+        then lift $ agentDeliveryStatus acId "withCompletedCommand" $ \ad@AgentDeliveryStatus {pendingAcks} -> ad {pendingAcks = M.adjust (const True) ackKey pendingAcks}
+        else do
           cmdData_ <- withStore' $ \db -> getCommandDataByCorrId db user corrId
           case cmdData_ of
             Just cmdData@CommandData {cmdId, cmdConnId = Just cmdConnId', cmdFunction}
@@ -4641,8 +4734,10 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
                   action cmdData
               | otherwise -> err cmdId $ "not matching connection id or unexpected response, corrId = " <> show corrId
             Just CommandData {cmdId, cmdConnId = Nothing} -> err cmdId $ "no command connection id, corrId = " <> show corrId
-        Nothing -> throwChatError . CEAgentCommandError $ "command not found, corrId = " <> show corrId
+            Nothing -> logWarn $ "command not found, corrId = " <> ackKey
       where
+        acId = AgentConnId agentConnId
+        ackKey = decodeLatin1 $ strEncode corrId
         err cmdId msg = do
           withStore' $ \db -> updateCommandStatus db user cmdId CSError
           throwChatError . CEAgentCommandError $ msg
@@ -4663,18 +4758,27 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
         Right withRcpt -> ackMsg msgMeta $ if withRcpt then Just "" else Nothing
         -- If showCritical is True, then these errors don't result in ACK and show user visible alert
         -- This prevents losing the message that failed to be processed.
-        Left (ChatErrorStore SEDBBusyError {message}) | showCritical -> throwError $ ChatErrorAgent (CRITICAL True message) Nothing
+        Left (ChatErrorStore SEDBBusyError {message}) | showCritical -> do
+          lift $ agentDeliveryStatus (AgentConnId agentConnId) "SEDBBusyError" id
+          throwError $ ChatErrorAgent (CRITICAL True message) Nothing
         Left e -> ackMsg msgMeta Nothing >> throwError e
       where
         ackMsg :: MsgMeta -> Maybe MsgReceiptInfo -> CM ()
-        ackMsg MsgMeta {recipient = (msgId, _)} rcpt = withAgent $ \a -> ackMessageAsync a "" cId msgId rcpt
+        ackMsg MsgMeta {recipient = (msgId, _)} rcpt = do
+          ackCorrId <- drgRandomBytes 24
+          lift $ agentDeliveryStatus (AgentConnId agentConnId) "ackMsg" id
+          withAgent $ \a -> ackMessageAsync a ackCorrId cId msgId rcpt
+          now <- liftIO getCurrentTime
+          let ackKey = decodeLatin1 $ strEncode ackCorrId
+          lift . agentDeliveryStatus (AgentConnId agentConnId) "after ackMessageAsync" $ \ad@AgentDeliveryStatus {pendingAcks} ->
+            ad {ackSent = Just (now, ackKey), pendingAcks = M.insert ackKey False pendingAcks}
 
     sentMsgDeliveryEvent :: Connection -> AgentMsgId -> CM ()
     sentMsgDeliveryEvent Connection {connId} msgId =
       withStore' $ \db -> updateSndMsgDeliveryStatus db connId msgId MDSSndSent
 
     agentErrToItemStatus :: AgentErrorType -> CIStatus 'MDSnd
-    agentErrToItemStatus (SMP AUTH) = CISSndErrorAuth
+    agentErrToItemStatus (SMP _ AUTH) = CISSndErrorAuth
     agentErrToItemStatus err = CISSndError . T.unpack . safeDecodeUtf8 $ strEncode err
 
     badRcvFileChunk :: RcvFileTransfer -> String -> CM ()
@@ -5966,6 +6070,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
         processForwardedMsg author chatMsg = do
           let body = LB.toStrict $ J.encode msg
           rcvMsg@RcvMessage {chatMsgEvent = ACME _ event} <- saveGroupFwdRcvMsg user groupId m author body chatMsg
+          lift $ agentDeliveryStatus (AgentConnId agentConnId) "eventTag in processForwardedMsg" $ \ad -> ad {eventTag = Just $! tshow (toCMEventTag event)}
           case event of
             XMsgNew mc -> memberCanSend author $ newGroupContentMessage gInfo author mc rcvMsg msgTs True
             XMsgFileDescr sharedMsgId fileDescr -> memberCanSend author $ groupMessageFileDescription gInfo author sharedMsgId fileDescr
@@ -6590,9 +6695,10 @@ sendPendingGroupMessages user GroupMember {groupMemberId, localDisplayName} conn
 
 -- TODO [batch send] refactor direct message processing same as groups (e.g. checkIntegrity before processing)
 saveDirectRcvMSG :: Connection -> MsgMeta -> MsgBody -> CM (Connection, RcvMessage)
-saveDirectRcvMSG conn@Connection {connId} agentMsgMeta msgBody =
+saveDirectRcvMSG conn@Connection {connId, agentConnId} agentMsgMeta msgBody =
   case parseChatMessages msgBody of
     [Right (ACMsg _ ChatMessage {chatVRange, msgId = sharedMsgId_, chatMsgEvent})] -> do
+      lift $ agentDeliveryStatus agentConnId "eventTag in saveDirectRcvMSG" $ \ad -> ad {eventTag = Just $! tshow (toCMEventTag chatMsgEvent)}
       conn' <- updatePeerChatVRange conn chatVRange
       let agentMsgId = fst $ recipient agentMsgMeta
           newMsg = NewRcvMessage {chatMsgEvent, msgBody}
@@ -7295,12 +7401,15 @@ chatCommandP =
       "/_download " *> (APIDownloadStandaloneFile <$> A.decimal <* A.space <*> strP_ <*> cryptoFileP),
       ("/quit" <|> "/q" <|> "/exit") $> QuitChat,
       ("/version" <|> "/v") $> ShowVersion,
+      "/debug delivery" *> (DebugDelivery <$> (" all" $> True <|> pure False)),
+      "/debug conn " *> (DebugConnection <$> ((Left <$> A.decimal) <|> (Right . AgentConnId <$> base64P))),
       "/debug locks" $> DebugLocks,
       "/debug event " *> (DebugEvent <$> jsonP),
       "/get stats" $> GetAgentStats,
       "/reset stats" $> ResetAgentStats,
       "/get subs" $> GetAgentSubs,
       "/get subs details" $> GetAgentSubsDetails,
+      "/get subs diff" $> GetAgentSubsDiff,
       "/get workers" $> GetAgentWorkers,
       "/get workers details" $> GetAgentWorkersDetails,
       "//" *> (CustomChatCommand <$> A.takeByteString)
@@ -7456,6 +7565,9 @@ timeItToView s action = do
   toView' $ CRTimedAction s diff
   pure a
 
+drgRandomBytes :: Int -> CM ByteString
+drgRandomBytes n = asks random >>= atomically . C.randomBytes n
+
 mkValidName :: String -> String
 mkValidName = reverse . dropWhile isSpace . fst3 . foldl' addChar ("", '\NUL', 0 :: Int)
   where
@@ -7500,3 +7612,8 @@ xftpSndFileRedirect user ftId vfd = do
 
 dummyFileDescr :: FileDescr
 dummyFileDescr = FileDescr {fileDescrText = "", fileDescrPartNo = 0, fileDescrComplete = False}
+
+agentDeliveryStatus :: AgentConnId -> Text -> (AgentDeliveryStatus -> AgentDeliveryStatus) -> CM' ()
+agentDeliveryStatus acId tracking f = do
+  ads <- asks agentDeliveryStatuses
+  atomically $ TM.lookup acId ads >>= mapM_ (`modifyTVar'` \ds -> (f ds) {tracking})

src/Simplex/Chat/Controller.hs

@@ -66,7 +66,7 @@ import Simplex.Chat.Types.Shared
 import Simplex.Chat.Util (liftIOEither)
 import Simplex.FileTransfer.Description (FileDescriptionURI)
 import Simplex.Messaging.Agent (AgentClient, SubscriptionsInfo)
-import Simplex.Messaging.Agent.Client (AgentLocks, AgentWorkersDetails (..), AgentWorkersSummary (..), ProtocolTestFailure, UserNetworkInfo)
+import Simplex.Messaging.Agent.Client (AgentLocks, AgentWorkersDetails (..), AgentWorkersSummary (..), ProtocolTestFailure, SubscriptionsDiff, UserNetworkInfo)
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, NetworkConfig)
 import Simplex.Messaging.Agent.Lock
 import Simplex.Messaging.Agent.Protocol
@@ -207,6 +207,7 @@ data ChatController = ChatController
     inputQ :: TBQueue String,
     outputQ :: TBQueue (Maybe CorrId, Maybe RemoteHostId, ChatResponse),
     connNetworkStatuses :: TMap AgentConnId NetworkStatus,
+    agentDeliveryStatuses :: TMap AgentConnId (TVar AgentDeliveryStatus),
     subscriptionMode :: TVar SubscriptionMode,
     chatLock :: Lock,
     entityLocks :: TMap ChatLockEntity Lock,
@@ -233,6 +234,23 @@ data ChatController = ChatController
     contactMergeEnabled :: TVar Bool
   }
 
+data AgentDeliveryStatus = AgentDeliveryStatus
+  { lastCmd :: UTCTime,
+    tracking :: Text,
+    isMSG :: Bool, -- False for RCVD
+    connId :: Maybe Int64, -- chat connection ID
+    msgBodyPfx :: Text,
+    eventTag :: Maybe Text, -- tshow of ACMEventTag (for JSON instances)
+    ackSent :: Maybe (UTCTime, Text), -- strEncode of random CorrId
+    pendingAcks :: Map Text Bool
+  }
+  deriving (Show) -- for ChatResponse
+
+agentDeliveryOk :: AgentDeliveryStatus -> Bool
+agentDeliveryOk AgentDeliveryStatus {ackSent, pendingAcks} = case ackSent of
+  Nothing -> False
+  Just (_, corrId) -> M.lookup corrId pendingAcks == Just True && and pendingAcks
+
 data HelpSection = HSMain | HSFiles | HSGroups | HSContacts | HSMyAddress | HSIncognito | HSMarkdown | HSMessages | HSRemote | HSSettings | HSDatabase
   deriving (Show)
 
@@ -488,12 +506,15 @@ data ChatCommand
   | APIStandaloneFileInfo FileDescriptionURI
   | QuitChat
   | ShowVersion
+  | DebugDelivery Bool
+  | DebugConnection (Either Int64 AgentConnId)
   | DebugLocks
   | DebugEvent ChatResponse
   | GetAgentStats
   | ResetAgentStats
   | GetAgentSubs
   | GetAgentSubsDetails
+  | GetAgentSubsDiff
   | GetAgentWorkers
   | GetAgentWorkersDetails
   | -- The parser will return this command for strings that start from "//".
@@ -735,12 +756,15 @@ data ChatResponse
   | CRContactPQEnabled {user :: User, contact :: Contact, pqEnabled :: PQEncryption}
   | CRSQLResult {rows :: [Text]}
   | CRSlowSQLQueries {chatQueries :: [SlowSQLQuery], agentQueries :: [SlowSQLQuery]}
+  | CRDebugDelivery {debugDelivery :: Map Text AgentDeliveryStatus}
+  | CRDebugConnection {debugConnection :: DebugConnectionStatus}
   | CRDebugLocks {chatLockName :: Maybe String, chatEntityLocks :: Map String String, agentLocks :: AgentLocks}
   | CRAgentStats {agentStats :: [[String]]}
   | CRAgentWorkersDetails {agentWorkersDetails :: AgentWorkersDetails}
   | CRAgentWorkersSummary {agentWorkersSummary :: AgentWorkersSummary}
   | CRAgentSubs {activeSubs :: Map Text Int, pendingSubs :: Map Text Int, removedSubs :: Map Text [String]}
   | CRAgentSubsDetails {agentSubs :: SubscriptionsInfo}
+  | CRAgentSubsDiff {agentSubsDiff :: SubscriptionsDiff}
   | CRConnectionDisabled {connectionEntity :: ConnectionEntity}
   | CRAgentRcvQueueDeleted {agentConnId :: AgentConnId, server :: SMPServer, agentQueueId :: AgentQueueId, agentError_ :: Maybe AgentErrorType}
   | CRAgentConnDeleted {agentConnId :: AgentConnId}
@@ -755,6 +779,21 @@ data ChatResponse
   | CRCustomChatResponse {user_ :: Maybe User, response :: Text}
   deriving (Show)
 
+data DebugConnectionStatus = DebugConnectionStatus
+  { deliveryStatus :: Maybe AgentDeliveryStatus,
+    networkStatus :: Maybe NetworkStatus,
+    -- from agent's TRecvQ via rcvId
+    inActive :: Bool, -- should the delivery work right now?
+    inPending :: Bool, -- is there a temporary error?
+    -- from receive queue
+    queueStatus :: Text,
+    server :: (Text, String), -- what's the server for this connection? -- XXX: reveals private servers and association
+    smpClientStatus :: Text, -- is there an active client for it?
+    subWorkerStatus :: Text, -- a session was recently restarted and tries to resubscribe
+    connection :: Connection
+  }
+  deriving (Show)
+
 -- some of these can only be used as command responses
 allowRemoteEvent :: ChatResponse -> Bool
 allowRemoteEvent = \case
@@ -1456,6 +1495,10 @@ $(JQ.deriveJSON (sumTypeJSON $ dropPrefix "RCSR") ''RemoteCtrlStopReason)
 
 $(JQ.deriveJSON (sumTypeJSON $ dropPrefix "RHSR") ''RemoteHostStopReason)
 
+$(JQ.deriveJSON J.defaultOptions ''AgentDeliveryStatus)
+
+$(JQ.deriveJSON J.defaultOptions ''DebugConnectionStatus)
+
 $(JQ.deriveJSON (sumTypeJSON $ dropPrefix "CR") ''ChatResponse)
 
 $(JQ.deriveFromJSON defaultJSON ''ArchiveConfig)

src/Simplex/Chat/Store/Shared.hs

@@ -59,6 +59,7 @@ data StoreError
   = SEDuplicateName
   | SEUserNotFound {userId :: UserId}
   | SEUserNotFoundByName {contactName :: ContactName}
+  | SEUserNotFoundByAConnId {agentConnId :: AgentConnId}
   | SEUserNotFoundByContactId {contactId :: ContactId}
   | SEUserNotFoundByGroupId {groupId :: GroupId}
   | SEUserNotFoundByFileId {fileId :: FileTransferId}

src/Simplex/Chat/View.hs

@@ -351,6 +351,8 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe
             <> (" :: avg: " <> sShow timeAvg <> " ms")
             <> (" :: " <> plain (T.unwords $ T.lines query))
      in ("Chat queries" : map viewQuery chatQueries) <> [""] <> ("Agent queries" : map viewQuery agentQueries)
+  CRDebugDelivery ads -> [plain $ LB.unpack (J.encode ads)]
+  CRDebugConnection cs -> [plain $ LB.unpack (J.encode cs)]
   CRDebugLocks {chatLockName, chatEntityLocks, agentLocks} ->
     [ maybe "no chat lock" (("chat lock: " <>) . plain) chatLockName,
       plain $ "chat entity locks: " <> LB.unpack (J.encode chatEntityLocks),
@@ -370,6 +372,10 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe
       <> ("pending subscriptions: " : map sShow pendingSubscriptions)
       <> ("removed subscriptions: " : map sShow removedSubscriptions)
   CRAgentWorkersSummary {agentWorkersSummary} -> ["agent workers summary: " <> plain (LB.unpack $ J.encode agentWorkersSummary)]
+  CRAgentSubsDiff asd ->
+    [ "agent subs diff:",
+      plain . LB.unpack $ J.encode asd
+    ]
   CRAgentWorkersDetails {agentWorkersDetails} ->
     [ "agent workers details:",
       plain . LB.unpack $ J.encode agentWorkersDetails -- this would be huge, but copypastable when has its own line
@@ -1178,8 +1184,8 @@ viewServerTestResult :: AProtoServerWithAuth -> Maybe ProtocolTestFailure -> [St
 viewServerTestResult (AProtoServerWithAuth p _) = \case
   Just ProtocolTestFailure {testStep, testError} ->
     result
-      <> [pName <> " server requires authorization to create queues, check password" | testStep == TSCreateQueue && testError == SMP SMP.AUTH]
+      <> [pName <> " server requires authorization to create queues, check password" | SMP _ SMP.AUTH <- [testError], testStep == TSCreateQueue]
-      <> [pName <> " server requires authorization to upload files, check password" | testStep == TSCreateFile && testError == XFTP XFTPTransport.AUTH]
+      <> [pName <> " server requires authorization to upload files, check password" | XFTP _ XFTPTransport.AUTH <- [testError], testStep == TSCreateFile]
       <> ["Possibly, certificate fingerprint in " <> pName <> " server address is incorrect" | testStep == TSConnect && brokerErr]
     where
       result = [pName <> " server test failed at " <> plain (drop 2 $ show testStep) <> ", error: " <> plain (strEncode testError)]
@@ -2019,7 +2025,7 @@ viewChatError logLevel testView = \case
     e -> ["chat database error: " <> sShow e]
   ChatErrorAgent err entity_ -> case err of
     CMD PROHIBITED -> [withConnEntity <> "error: command is prohibited"]
-    SMP SMP.AUTH ->
+    SMP _ SMP.AUTH ->
       [ withConnEntity
           <> "error: connection authorization failed - this could happen if connection was deleted,\
              \ secured with different credentials, or due to a bug - please re-create the connection"

tests/ChatTests/Direct.hs

@@ -1069,6 +1069,7 @@ testMaintenanceMode tmp = do
       bob <# "alice> hi again"
       bob #> "@alice hello"
       alice <# "bob> hello"
+      threadDelay 100000
       -- export / delete / import
       alice ##> "/_stop"
       alice <## "chat stopped"
@@ -1149,6 +1150,7 @@ testDatabaseEncryption tmp = do
       alice <## "error: chat not stopped"
       alice ##> "/db decrypt mykey"
       alice <## "error: chat not stopped"
+      threadDelay 100000
       alice ##> "/_stop"
       alice <## "chat stopped"
       alice ##> "/db decrypt mykey"