agent: fix race when pending subscriptions are never subscribed (#1756)

* agent: fix race when pending subscriptions are never subscribed

* small agent

---------

Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>

Author: epoberezkin

src/Simplex/Messaging/Agent/Client.hs

@@ -764,12 +764,15 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do
         (pure Nothing) -- prevent race with cleanup and adding pending queues in another call
         (Just <$> getSessVar workerSeq tSess smpSubWorkers ts)
     newSubWorker v = do
-      a <- async $ void (E.tryAny runSubWorker) >> atomically (cleanup v)
+      a <- async $ void $ E.tryAny $ runSubWorker v
       atomically $ putTMVar (sessionVar v) a
-    runSubWorker = do
+    runSubWorker v = do
       ri <- asks $ reconnectInterval . config
       withRetryForeground ri isForeground (isNetworkOnline c) $ \_ loop -> do
-        pending <- atomically $ SS.getPendingSubs tSess $ currentSubs c
+        pending <- atomically $ do
+          qs <- SS.getPendingSubs tSess $ currentSubs c
+          when (M.null qs) $ cleanup v
+          pure qs
         unless (M.null pending) $ do
           liftIO $ waitUntilForeground c
           liftIO $ waitForUserNetwork c

src/Simplex/Messaging/Client/Agent.hs

@@ -307,11 +307,14 @@ reconnectClient ca@SMPClientAgent {active, agentCfg, smpSubWorkers, workerSeq} s
         (Just <$> getSessVar workerSeq srv smpSubWorkers ts)
     newSubWorker :: SessionVar (Async ()) -> IO ()
     newSubWorker v = do
-      a <- async $ void (E.tryAny runSubWorker) >> atomically (cleanup v)
+      a <- async $ void $ E.tryAny $ runSubWorker v
       atomically $ putTMVar (sessionVar v) a
-    runSubWorker =
+    runSubWorker v =
       withRetryInterval (reconnectInterval agentCfg) $ \_ loop -> do
-        subs <- getPending TM.lookupIO readTVarIO
+        subs <- atomically $ do
+          s <- getPending TM.lookup readTVar
+          when (noPending s) $ cleanup v
+          pure s
         unless (noPending subs) $ whenM (readTVarIO active) $ do
           void $ netTimeoutInt tcpConnectTimeout NRMBackground `timeout` runExceptT (reconnectSMPClient ca srv subs)
           loop