smp server: do not create messaging queues in SMP proxy to prevent deadlock in processing

Author: epoberezkin

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -50,7 +50,6 @@ import Data.Bits (xor)
 import Data.ByteArray (ScrubbedBytes)
 import qualified Data.ByteArray as BA
 import Data.ByteString (ByteString)
-import qualified Data.ByteString as B
 import Data.Functor (($>))
 import Data.IORef
 import Data.Maybe (fromMaybe)

src/Simplex/Messaging/Client/Agent.hs

@@ -107,7 +107,7 @@ data SMPClientAgentConfig = SMPClientAgentConfig
   { smpCfg :: ProtocolClientConfig SMPVersion,
     reconnectInterval :: RetryInterval,
     persistErrorInterval :: NominalDiffTime,
-    msgQSize :: Natural,
+    msgQSize :: Maybe Natural,
     agentQSize :: Natural,
     agentSubsBatchSize :: Int,
     ownServerDomains :: [ByteString]
@@ -124,7 +124,7 @@ defaultSMPClientAgentConfig =
             maxInterval = 10 * second
           },
       persistErrorInterval = 30, -- seconds
-      msgQSize = 2048,
+      msgQSize = Just 2048,
       agentQSize = 2048,
       agentSubsBatchSize = 1360,
       ownServerDomains = []
@@ -162,9 +162,8 @@ newSMPClientAgent :: SParty p -> SMPClientAgentConfig -> Maybe DBService -> TVar
 newSMPClientAgent agentParty agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} dbService randomDrg = do
   active <- newTVarIO True
   startedAt <- getCurrentTime
-  -- Only subscribing agents receive server transmissions. A queue nobody reads fills with late
-  -- responses (Client.hs, processMsg) until the client's `process` thread blocks writing to it.
-  msgQ <- forM (serviceParty agentParty) $ \_ -> newTBQueueIO msgQSize
+  -- Only subscribing agents receive server transmissions, should not be created until processed to prevent deadlock.
+  msgQ <- mapM newTBQueueIO msgQSize
   agentQ <- newTBQueueIO agentQSize
   smpClients <- TM.emptyIO
   smpSessions <- TM.emptyIO

src/Simplex/Messaging/Server/Main.hs

@@ -604,6 +604,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                               }
                         },
                     ownServerDomains = either (const []) textToOwnServers $ lookupValue "PROXY" "own_server_domains" ini,
+                    msgQSize = Nothing, -- to prevent accumulation of late responses, and deadlocks in SMP proxy
                     persistErrorInterval = 30 -- seconds
                   },
               allowSMPProxy = True,

tests/SMPClient.hs

@@ -275,7 +275,7 @@ cfgMS msType = withStoreCfg (testServerStoreConfig msType) $ \serverStoreCfg ->
       smpServerVRange = supportedServerSMPRelayVRange,
       transportConfig = mkTransportServerConfig True (Just alpnSupportedSMPHandshakes) True,
       controlPort = Nothing,
-      smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1}, -- seconds
+      smpAgentCfg = defaultSMPClientAgentConfig {persistErrorInterval = 1, msgQSize = Nothing}, -- seconds
       allowSMPProxy = False,
       serverClientConcurrency = 2,
       serverResolverConcurrency = defaultNameResolverConcurrency,