Merge branch 'master' into rcv-services

Author: epoberezkin

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.5.0.11
+version:        6.5.0.12
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Agent.hs

@@ -365,9 +365,9 @@ setConnShortLinkAsync :: AgentClient -> ACorrId -> ConnId -> UserConnLinkData 'C
 setConnShortLinkAsync c = withAgentEnv c .:: setConnShortLinkAsync' c
 {-# INLINE setConnShortLinkAsync #-}
 
--- | Get and verify data from short link (LGET/LKEY command) asynchronously, synchronous response is new connection id
-getConnShortLinkAsync :: AgentClient -> UserId -> ACorrId -> ConnShortLink 'CMContact -> AE ConnId
-getConnShortLinkAsync c = withAgentEnv c .:. getConnShortLinkAsync' c
+-- | Get and verify data from short link (LGET/LKEY command) asynchronously, synchronous response is new/passed connection id
+getConnShortLinkAsync :: AgentClient -> UserId -> ACorrId -> Maybe ConnId -> ConnShortLink 'CMContact -> AE ConnId
+getConnShortLinkAsync c = withAgentEnv c .:: getConnShortLinkAsync' c
 {-# INLINE getConnShortLinkAsync #-}
 
 -- | Join SMP agent connection (JOIN command) asynchronously, synchronous response is new connection id.
@@ -1041,14 +1041,22 @@ setConnShortLinkAsync' c corrId connId userLinkData clientData =
       _ -> throwE $ CMD PROHIBITED "setConnShortLinkAsync: invalid connection or mode"
     enqueueCommand c corrId connId (Just srv) $ AClientCommand $ LSET userLinkData clientData
 
-getConnShortLinkAsync' :: AgentClient -> UserId -> ACorrId -> ConnShortLink 'CMContact -> AM ConnId
-getConnShortLinkAsync' c userId corrId shortLink@(CSLContact _ _ srv _) = do
-  g <- asks random
-  connId <- withStore c $ \db -> do
-    -- server is created so the command is processed in server queue,
-    -- not blocking other "no server" commands
-    void $ createServer db srv
-    prepareNewConn db g
+getConnShortLinkAsync' :: AgentClient -> UserId -> ACorrId -> Maybe ConnId -> ConnShortLink 'CMContact -> AM ConnId
+getConnShortLinkAsync' c userId corrId connId_ shortLink@(CSLContact _ _ srv _) = do
+  connId <- case connId_ of
+    Just existingConnId -> do
+      -- connId and srv can be unrelated: connId is used as "mailbox" for LDATA delivery,
+      -- while srv is the short link's server for the LGET request.
+      -- E.g., owner's relay connection (connId, on server A) fetches relay's group link data (srv = server B).
+      -- This works because enqueueCommand stores (connId, srv) independently in the commands table,
+      -- the network request targets srv, and event delivery uses connId via corrId correlation.
+      withStore' c $ \db -> void $ createServer db srv
+      pure existingConnId
+    Nothing -> do
+      g <- asks random
+      withStore c $ \db -> do
+        void $ createServer db srv
+        prepareNewConn db g
   enqueueCommand c corrId connId (Just srv) $ AClientCommand $ LGET shortLink
   pure connId
   where

src/Simplex/Messaging/Agent/Client.hs

@@ -808,13 +808,17 @@ 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
-        (pendingSubs, pendingSS) <- atomically $ SS.getPendingSubs tSess $ currentSubs c
-        unless (M.null pendingSubs && isNothing pendingSS) $ do
+        pending_ <- atomically $ do
+          pending@(pendingSubs, pendingSS) <- SS.getPendingSubs tSess $ currentSubs c
+          if M.null pendingSubs && isNothing pendingSS
+            then cleanup v $> Nothing
+            else pure $ Just pending
+        forM_ pending_ $ \(pendingSubs, pendingSS) -> do
           liftIO $ waitUntilForeground c
           liftIO $ waitForUserNetwork c
           mapM_ (handleNotify . void . runExceptT . resubscribeClientService c tSess) pendingSS

src/Simplex/Messaging/Client/Agent.hs

@@ -324,11 +324,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.try @E.SomeException runSubWorker) >> atomically (cleanup v)
+      a <- async $ void $ E.try @E.SomeException $ 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

tests/AgentTests/FunctionalAPITests.hs

@@ -2740,7 +2740,7 @@ testGetConnShortLinkAsync ps = withAgentClients2 $ \alice bob ->
         newLinkData = UserContactLinkData userCtData
     (_, CCLink qInfo (Just shortLink)) <- A.createConnection alice NRMInteractive 1 True True SCMContact (Just newLinkData) Nothing IKPQOn SMSubscribe
     -- get link data async - creates new connection for bob
-    newId <- getConnShortLinkAsync bob 1 "1" shortLink
+    newId <- getConnShortLinkAsync bob 1 "1" Nothing shortLink
     ("1", newId', LDATA FixedLinkData {linkConnReq = qInfo'} (ContactLinkData _ userCtData')) <- get bob
     liftIO $ newId' `shouldBe` newId
     liftIO $ qInfo' `shouldBe` qInfo