diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index a670dd3e2..669b39031 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/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) diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index e41d7a811..0d59f00b8 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/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 = [] @@ -138,7 +138,7 @@ data SMPClientAgent p = SMPClientAgent dbService :: Maybe DBService, active :: TVar Bool, startedAt :: UTCTime, - msgQ :: TBQueue (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg), + msgQ :: Maybe (TBQueue (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg)), agentQ :: TBQueue SMPClientAgentEvent, randomDrg :: TVar ChaChaDRG, smpClients :: TMap SMPServer SMPClientVar, @@ -162,7 +162,8 @@ newSMPClientAgent :: SParty p -> SMPClientAgentConfig -> Maybe DBService -> TVar newSMPClientAgent agentParty agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} dbService randomDrg = do active <- newTVarIO True startedAt <- getCurrentTime - msgQ <- 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 @@ -264,7 +265,7 @@ connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, m Nothing -> getClient cfg where cfg = smpCfg agentCfg - getClient cfg' = getProtocolClient randomDrg NRMBackground (1, srv, Nothing) cfg' [] (Just msgQ) startedAt clientDisconnected + getClient cfg' = getProtocolClient randomDrg NRMBackground (1, srv, Nothing) cfg' [] msgQ startedAt clientDisconnected clientDisconnected :: SMPClient -> IO () clientDisconnected smp = do diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index 0b3aa717b..f04ca4e35 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -526,10 +526,10 @@ subscribeNtfs NtfSubscriber {smpSubscribers, subscriberSeq, smpAgent = ca} st sm subscribeQueuesNtfs ca smpServer' [sub] ntfSubscriber :: NtfSubscriber -> M () -ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ, agentQ}} = +ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ = msgQ_, agentQ}} = race_ receiveSMP receiveAgent where - receiveSMP = do + receiveSMP = forM_ msgQ_ $ \msgQ -> do st <- asks store ps <- asks pushServer stats <- asks serverStats diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index 098c18517..747226eb5 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/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, diff --git a/tests/SMPClient.hs b/tests/SMPClient.hs index cc6862b13..1dcd350bc 100644 --- a/tests/SMPClient.hs +++ b/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,