add /debug subs

src/Simplex/Chat.hs

@@ -46,6 +46,7 @@ import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, listToMaybe, mapMaybe, maybeToList)
+import qualified Data.Set as S
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
@@ -232,6 +233,7 @@ newChatController
     outputQ <- newTBQueueIO tbqSize
     connNetworkStatuses <- atomically TM.empty
     agentDeliveryStatuses <- atomically TM.empty
+    agentSubscriptions <- newTVarIO S.empty
     subscriptionMode <- newTVarIO SMSubscribe
     chatLock <- newEmptyTMVarIO
     entityLocks <- atomically TM.empty
@@ -269,6 +271,7 @@ newChatController
           outputQ,
           connNetworkStatuses,
           agentDeliveryStatuses,
+          agentSubscriptions,
           subscriptionMode,
           chatLock,
           entityLocks,
@@ -408,6 +411,7 @@ startChatController mainApp = do
 subscribeUsers :: Bool -> [User] -> CM' ()
 subscribeUsers onlyNeeded users = do
   let (us, us') = partition activeUser users
+  asks agentSubscriptions >>= atomically . (`writeTVar` S.empty)
   vr <- chatVersionRange'
   subscribe vr us
   subscribe vr us'
@@ -2206,6 +2210,14 @@ processChatCommand' vr = \case
             queueStatus = tshow rqStatus,
             connection
           }
+  DebugSubs -> do
+    ads <- mapM readTVarIO =<< readTVarIO =<< asks agentDeliveryStatuses
+    conns <- readTVarIO =<< asks agentSubscriptions
+    pure . CRDebugSubs . S.toList $ foldl' (\cs (acId, ds) -> if agentDeliveryOk ds then S.delete acId cs else cs) conns (M.toList ads)
+  DebugSubsDetails -> do
+    CRDebugSubs broken <- processChatCommand' vr DebugSubs
+    crs <- mapM (processChatCommand' vr . DebugConnection . Right) broken
+    pure $ CRDebugSubsDetails [dcs | CRDebugConnection dcs <- crs]
   DebugLocks -> lift $ do
     chatLockName <- atomically . tryReadTMVar =<< asks chatLock
     chatEntityLocks <- getLocks =<< asks entityLocks
@@ -3301,6 +3313,7 @@ subscribeUserConnections vr onlyNeeded agentBatchSubscribe user = do
         let conns = concat [ctConns, ucConns, mConns, sftConns, rftConns, pcConns]
         pure (conns, cts, ucs, gs, ms, sfts, rfts, pcs)
   -- subscribe using batched commands
+  asks agentSubscriptions >>= \v -> atomically . modifyTVar' v $ \cs -> foldl' (\cs' acId -> S.insert (AgentConnId acId) cs') cs conns
   rs <- withAgent $ \a -> agentBatchSubscribe a conns
   -- send connection events to view
   contactSubsToView rs cts ce
@@ -6849,6 +6862,7 @@ deleteAgentConnectionAsync user acId = deleteAgentConnectionAsync' user acId Fal
 deleteAgentConnectionAsync' :: User -> ConnId -> Bool -> CM ()
 deleteAgentConnectionAsync' user acId waitDelivery = do
   withAgent (\a -> deleteConnectionAsync a waitDelivery acId) `catchChatError` (toView . CRChatError (Just user))
+  asks agentSubscriptions >>= \v -> atomically $ modifyTVar v $ S.delete (AgentConnId acId)
 
 deleteAgentConnectionsAsync :: User -> [ConnId] -> CM ()
 deleteAgentConnectionsAsync user acIds = deleteAgentConnectionsAsync' user acIds False
@@ -6857,6 +6871,7 @@ deleteAgentConnectionsAsync' :: User -> [ConnId] -> Bool -> CM ()
 deleteAgentConnectionsAsync' _ [] _ = pure ()
 deleteAgentConnectionsAsync' user acIds waitDelivery = do
   withAgent (\a -> deleteConnectionsAsync a waitDelivery acIds) `catchChatError` (toView . CRChatError (Just user))
+  asks agentSubscriptions >>= \v -> atomically $ modifyTVar v $ \as -> foldl' (\as' acId -> S.delete (AgentConnId acId) as') as acIds
 
 agentXFTPDeleteRcvFile :: RcvFileId -> FileTransferId -> CM ()
 agentXFTPDeleteRcvFile aFileId fileId = do
@@ -7389,6 +7404,8 @@ chatCommandP =
       ("/version" <|> "/v") $> ShowVersion,
       "/debug delivery" *> (DebugDelivery <$> (" all" $> True <|> pure False)),
       "/debug conn " *> (DebugConnection <$> ((Left <$> A.decimal) <|> (Right . AgentConnId <$> base64P))),
+      "/debug subs" $> DebugSubs,
+      "/debug subs details" $> DebugSubsDetails,
       "/debug locks" $> DebugLocks,
       "/debug event " *> (DebugEvent <$> jsonP),
       "/get stats" $> GetAgentStats,

src/Simplex/Chat/Controller.hs

@@ -39,6 +39,7 @@ import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
+import Data.Set (Set)
 import Data.String
 import Data.Text (Text)
 import Data.Text.Encoding (decodeLatin1)
@@ -208,6 +209,7 @@ data ChatController = ChatController
     outputQ :: TBQueue (Maybe CorrId, Maybe RemoteHostId, ChatResponse),
     connNetworkStatuses :: TMap AgentConnId NetworkStatus,
     agentDeliveryStatuses :: TMap AgentConnId (TVar AgentDeliveryStatus),
+    agentSubscriptions :: TVar (Set AgentConnId),
     subscriptionMode :: TVar SubscriptionMode,
     chatLock :: Lock,
     entityLocks :: TMap ChatLockEntity Lock,
@@ -508,6 +510,8 @@ data ChatCommand
   | ShowVersion
   | DebugDelivery Bool
   | DebugConnection (Either Int64 AgentConnId)
+  | DebugSubs -- Check that all `subscribeUserConnections` are in place and healthy
+  | DebugSubsDetails -- Run DebugConnection on each unhealthy sub
   | DebugLocks
   | DebugEvent ChatResponse
   | GetAgentStats
@@ -757,6 +761,8 @@ data ChatResponse
   | CRSlowSQLQueries {chatQueries :: [SlowSQLQuery], agentQueries :: [SlowSQLQuery]}
   | CRDebugDelivery {debugDelivery :: Map Text AgentDeliveryStatus}
   | CRDebugConnection {debugConnection :: DebugConnectionStatus}
+  | CRDebugSubs {brokenSubs :: [AgentConnId]}
+  | CRDebugSubsDetails {debugConnections :: [DebugConnectionStatus]}
   | CRDebugLocks {chatLockName :: Maybe String, chatEntityLocks :: Map String String, agentLocks :: AgentLocks}
   | CRAgentStats {agentStats :: [[String]]}
   | CRAgentWorkersDetails {agentWorkersDetails :: AgentWorkersDetails}

src/Simplex/Chat/View.hs

@@ -353,6 +353,8 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe
      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)]
+  CRDebugSubs cs -> [plain $ LB.unpack (J.encode cs)]
+  CRDebugSubsDetails 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),