diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 0b64fabf6..3072b0849 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -118,6 +118,7 @@ data PClient msg = PClient transportSession :: TransportSession msg, transportHost :: TransportHost, tcpTimeout :: Int, + pingErrorCount :: TVar Int, clientCorrId :: TVar Natural, sentCommands :: TMap CorrId (Request msg), sndQ :: TBQueue (NonEmpty SentRawTransmission), @@ -165,8 +166,10 @@ data NetworkConfig = NetworkConfig tcpTimeout :: Int, -- | TCP keep-alive options, Nothing to skip enabling keep-alive tcpKeepAlive :: Maybe KeepAliveOpts, - -- | period for SMP ping commands (microseconds) + -- | period for SMP ping commands (microseconds, 0 to disable) smpPingInterval :: Int, + -- | the count of PING errors after which SMP client terminates (and will be reconnected), 0 to disable + smpPingCount :: Int, logTLSErrors :: Bool } deriving (Eq, Show, Generic, FromJSON) @@ -196,6 +199,7 @@ defaultNetworkConfig = tcpTimeout = 5_000_000, tcpKeepAlive = Just defaultKeepAliveOpts, smpPingInterval = 600_000_000, -- 10min + smpPingCount = 3, logTLSErrors = False } @@ -279,6 +283,7 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, mkProtocolClient :: TransportHost -> STM (PClient msg) mkProtocolClient transportHost = do connected <- newTVar False + pingErrorCount <- newTVar 0 clientCorrId <- newTVar 0 sentCommands <- TM.empty sndQ <- newTBQueue qSize @@ -289,6 +294,7 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, transportSession, transportHost, tcpTimeout, + pingErrorCount, clientCorrId, sentCommands, sndQ, @@ -340,9 +346,15 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ ping :: ProtocolClient msg -> IO () - ping c = forever $ do + ping c@ProtocolClient {client_ = PClient {pingErrorCount}} = do threadDelay smpPingInterval - runExceptT $ sendProtocolCommand c Nothing "" protocolPing + runExceptT (sendProtocolCommand c Nothing "" protocolPing) >>= \case + Left PCEResponseTimeout -> do + cnt <- atomically $ stateTVar pingErrorCount $ \cnt -> (cnt + 1, cnt + 1) + when (maxCnt == 0 || cnt < maxCnt) $ ping c + _ -> ping c -- sendProtocolCommand resets pingErrorCount + where + maxCnt = smpPingCount networkConfig process :: ProtocolClient msg -> IO () process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c) @@ -556,26 +568,28 @@ sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd) -- | Send multiple commands with batching and collect responses sendProtocolCommands :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Either ProtocolClientError msg)) -sendProtocolCommands c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout}} cs = do +sendProtocolCommands c@ProtocolClient {client_ = PClient {sndQ}} cs = do ts <- mapM (runExceptT . mkTransmission c) cs mapM_ (atomically . writeTBQueue sndQ . L.map fst) . L.nonEmpty . rights $ L.toList ts forConcurrently ts $ \case - Right (_, r) -> withTimeout . atomically $ takeTMVar r + Right (_, r) -> withTimeout c $ atomically $ takeTMVar r Left e -> pure $ Left e - where - withTimeout a = fromMaybe (Left PCEResponseTimeout) <$> timeout tcpTimeout a -- | Send Protocol command sendProtocolCommand :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> Maybe C.APrivateSignKey -> QueueId -> ProtoCommand msg -> ExceptT ProtocolClientError IO msg -sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout}} pKey qId cmd = do +sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}} pKey qId cmd = do (t, r) <- mkTransmission c (pKey, qId, cmd) ExceptT $ sendRecv t r where -- two separate "atomically" needed to avoid blocking sendRecv :: SentRawTransmission -> TMVar (Response msg) -> IO (Response msg) - sendRecv t r = atomically (writeTBQueue sndQ [t]) >> withTimeout (atomically $ takeTMVar r) - where - withTimeout a = fromMaybe (Left PCEResponseTimeout) <$> timeout tcpTimeout a + sendRecv t r = atomically (writeTBQueue sndQ [t]) >> withTimeout c (atomically $ takeTMVar r) + +withTimeout :: ProtocolClient msg -> IO (Either ProtocolClientError msg) -> IO (Either ProtocolClientError msg) +withTimeout ProtocolClient {client_ = PClient {tcpTimeout, pingErrorCount}} a = + timeout tcpTimeout a >>= \case + Just r -> atomically (writeTVar pingErrorCount 0) >> pure r + _ -> pure $ Left PCEResponseTimeout mkTransmission :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> ClientCommand msg -> ExceptT ProtocolClientError IO (SentRawTransmission, TMVar (Response msg)) mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCorrId, sentCommands}} (pKey, qId, cmd) = do diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index 02f07da00..ce3a0972b 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -372,6 +372,7 @@ verifyNtfTransmission (sig_, signed, (corrId, entId, _)) cmd = do t_ <- atomically $ getActiveNtfToken st subTknId verifyToken' t_ $ verifiedSubCmd s c else pure $ maybe False (dummyVerifyCmd signed) sig_ `seq` VRFailed + NtfCmd SSubscription PING -> pure $ VRVerified $ NtfReqPing corrId entId NtfCmd SSubscription c -> do s_ <- atomically $ getNtfSubscription st entId case s_ of @@ -529,6 +530,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu incNtfStat subDeleted pure NROk PING -> pure NRPong + NtfReqPing corrId entId -> pure (corrId, entId, NRPong) getId :: M NtfEntityId getId = getRandomBytes =<< asks (subIdBytes . config) getRegCode :: M NtfRegCode diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 068af65c9..f3002e1bc 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -148,6 +148,7 @@ getPushClient s@NtfPushServer {pushClients} pp = data NtfRequest = NtfReqNew CorrId ANewNtfEntity | forall e. NtfEntityI e => NtfReqCmd (SNtfEntity e) (NtfEntityRec e) (Transmission (NtfCommand e)) + | NtfReqPing CorrId NtfEntityId data NtfServerClient = NtfServerClient { rcvQ :: TBQueue NtfRequest, diff --git a/tests/SMPAgentClient.hs b/tests/SMPAgentClient.hs index 39f8c86d8..51782c848 100644 --- a/tests/SMPAgentClient.hs +++ b/tests/SMPAgentClient.hs @@ -190,17 +190,8 @@ agentCfg = { tcpPort = agentTestPort, tbqSize = 4, database = testDB, - smpCfg = - defaultClientConfig - { qSize = 1, - defaultTransport = (testPort, transport @TLS), - networkConfig = defaultNetworkConfig {tcpTimeout = 500_000} - }, - ntfCfg = - defaultClientConfig - { qSize = 1, - defaultTransport = (ntfTestPort, transport @TLS) - }, + smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS)}, + ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS)}, reconnectInterval = defaultReconnectInterval {initialInterval = 50_000}, ntfWorkerDelay = 1000, ntfSMPWorkerDelay = 1000,