From fa794d7878c82c370f1547f01e76f9691d229b92 Mon Sep 17 00:00:00 2001
From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>
Date: Fri, 5 Jan 2024 17:36:57 +0000
Subject: [PATCH] agent: send errors on subscription timeouts (#937)

* agent: debug server reconnections

* update lock only if it has the same name already

* report subscription timeouts

* clean up
---
 src/Simplex/Messaging/Agent/Client.hs     | 36 ++++++++++++++++-------
 src/Simplex/Messaging/Agent/Env/SQLite.hs |  4 +++
 2 files changed, 29 insertions(+), 11 deletions(-)

diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs
index b9728f79c..42f69cc21 100644
--- a/src/Simplex/Messaging/Agent/Client.hs
+++ b/src/Simplex/Messaging/Agent/Client.hs
@@ -210,8 +210,7 @@ import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
 import System.Random (randomR)
-import System.Timeout (timeout)
-import UnliftIO (mapConcurrently)
+import UnliftIO (mapConcurrently, timeout)
 import UnliftIO.Async (async)
 import UnliftIO.Directory (getTemporaryDirectory)
 import UnliftIO.Exception (bracket)
@@ -498,10 +497,11 @@ getSMPServerClient :: forall m. AgentMonad m => AgentClient -> SMPTransportSessi
 getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
   atomically (getClientVar tSess smpClients)
-    >>= either
-      (newProtocolClient c tSess smpClients connectClient reconnectSMPClient)
-      (waitForProtocolClient c tSess)
+    >>= either newClient (waitForProtocolClient c tSess)
   where
+    newClient v = do
+      tc <- newTVarIO 0
+      newProtocolClient c tSess smpClients connectClient (reconnectSMPClient 0 tc) v
     connectClient :: m SMPClient
     connectClient = do
       cfg <- getClientConfig c smpCfg
@@ -539,14 +539,28 @@ reconnectServer c tSess = newAsyncAction tryReconnectSMPClient $ reconnections c
   where
     tryReconnectSMPClient aId = do
       ri <- asks $ reconnectInterval . config
-      withRetryInterval ri $ \_ loop ->
-        reconnectSMPClient c tSess `catchAgentError` const loop
+      timeoutCounts <- newTVarIO 0
+      withRetryIntervalCount ri $ \n _ loop ->
+        reconnectSMPClient n timeoutCounts c tSess `catchAgentError` const loop
       atomically . removeAsyncAction aId $ reconnections c
 
-reconnectSMPClient :: forall m. AgentMonad m => AgentClient -> SMPTransportSession -> m ()
-reconnectSMPClient c tSess@(_, srv, _) =
-  withLockMap_ (reconnectLocks c) tSess "reconnect" $
-    atomically (RQ.getSessQueues tSess $ pendingSubs c) >>= mapM_ resubscribe . L.nonEmpty
+reconnectSMPClient :: forall m. AgentMonad m => Int -> TVar Int -> AgentClient -> SMPTransportSession -> m ()
+reconnectSMPClient n tc c tSess@(_, srv, _) = do
+  ts <- liftIO getCurrentTime
+  let label = unwords ["reconnect", show n, show ts]
+  withLockMap_ (reconnectLocks c) tSess label $ do
+    qs <- atomically (RQ.getSessQueues tSess $ pendingSubs c)
+    NetworkConfig {tcpTimeout} <- readTVarIO $ useNetworkConfig c
+    -- this allows 3x of timeout per batch of subscription (90 queues per batch empirically)
+    let t = (length qs `div` 90 + 1) * tcpTimeout * 3
+    t `timeout` mapM_ resubscribe (L.nonEmpty qs) >>= \case
+      Just _ -> atomically $ writeTVar tc 0
+      Nothing -> do
+        tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
+        maxTC <- asks $ maxSubscriptionTimeouts . config
+        let err = if tc' >= maxTC then CRITICAL True else INTERNAL
+            msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
+        atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err msg)
   where
     resubscribe :: NonEmpty RcvQueue -> m ()
     resubscribe qs = do
diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs
index 9d6657627..a8cdb8180 100644
--- a/src/Simplex/Messaging/Agent/Env/SQLite.hs
+++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs
@@ -94,6 +94,7 @@ data AgentConfig = AgentConfig
     cleanupInterval :: Int64,
     cleanupStepInterval :: Int,
     maxWorkerRestartsPerMin :: Int,
+    maxSubscriptionTimeouts :: Int,
     storedMsgDataTTL :: NominalDiffTime,
     rcvFilesTTL :: NominalDiffTime,
     sndFilesTTL :: NominalDiffTime,
@@ -161,6 +162,9 @@ defaultAgentConfig =
       cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
       cleanupStepInterval = 200000, -- 200ms
       maxWorkerRestartsPerMin = 5,
+      -- 3 consecutive subscription timeouts will result in alert to the user
+      -- this is a fallback, as the timeout set to 3x of expected timeout, to avoid potential locking.
+      maxSubscriptionTimeouts = 3,
       storedMsgDataTTL = 21 * nominalDay,
       rcvFilesTTL = 2 * nominalDay,
       sndFilesTTL = nominalDay,