From a53323f729b8404fa175597cc283b35c5890dcd7 Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Mon, 18 May 2026 05:58:33 +0000 Subject: [PATCH] retry connecting client --- src/Simplex/Messaging/Notifications/Server.hs | 18 ++++++++-------- .../Messaging/Notifications/Server/Env.hs | 21 +++++++++++++------ .../Notifications/Server/Push/APNS.hs | 4 ++-- 3 files changed, 26 insertions(+), 17 deletions(-) diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index 54722601e..bb521623c 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -535,7 +535,7 @@ ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {msgQ, agentQ}} = stats <- asks serverStats forever $ do ((_, srv@(SMPServer (h :| _) _ _), _), _thVersion, sessionId, ts) <- atomically $ readTBQueue msgQ - forM ts $ \(ntfId, t) -> case t of + forM_ ts $ \(ntfId, t) -> case t of STUnexpectedError e -> logError $ "SMP client unexpected error: " <> tshow e -- uncorrelated response, should not happen STResponse {} -> pure () -- it was already reported as timeout error STEvent msgOrErr -> do @@ -639,14 +639,14 @@ showServer' :: SMPServer -> Text showServer' = decodeLatin1 . strEncode . host pushNotification :: NtfPushServer -> Maybe T.Text -> OwnServer -> NtfTknRec -> PushNotification -> M () -pushNotification s srvHost_ isOwn tkn ntf = do - q <- getOrCreatePushWorker s srvHost_ isOwn +pushNotification s srvHost_ isOwn tkn@NtfTknRec {token = DeviceToken pp _} ntf = do + q <- getOrCreatePushWorker s (srvHost_, pp) isOwn atomically $ writeTBQueue q (tkn, ntf) -getOrCreatePushWorker :: NtfPushServer -> Maybe T.Text -> OwnServer -> M (TBQueue (NtfTknRec, PushNotification)) -getOrCreatePushWorker s@NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize} srvHost_ isOwn = do +getOrCreatePushWorker :: NtfPushServer -> (Maybe T.Text, PushProvider) -> OwnServer -> M (TBQueue (NtfTknRec, PushNotification)) +getOrCreatePushWorker s@NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize} key@(srvHost_, _) isOwn = do ts <- liftIO getCurrentTime - atomically (getSessVar pushWorkerSeq srvHost_ pushWorkers ts) >>= \case + atomically (getSessVar pushWorkerSeq key pushWorkers ts) >>= \case Left v -> do q <- liftIO $ newTBQueueIO pushQSize tId <- mkWeakThreadId =<< forkIO (runPushWorker s srvHost_ isOwn q) @@ -716,7 +716,7 @@ runPushWorker s srvHost_ isOwn q = forever $ do _ -> err e err e = logError ("Push provider error (" <> tshow pp <> ", " <> tshow ntfTknId <> "): " <> tshow e) $> Left e -pushWorkersQLength :: TMap (Maybe T.Text) PushWorkerVar -> IO Natural +pushWorkersQLength :: TMap (Maybe T.Text, PushProvider) PushWorkerVar -> IO Natural pushWorkersQLength workers = do ws <- readTVarIO workers foldM addQLength 0 ws @@ -731,11 +731,11 @@ periodicNtfsThread s = do st <- asks store ntfsInterval <- asks $ periodicNtfsInterval . config let interval = 1000000 * ntfsInterval - q <- getOrCreatePushWorker s Nothing False + UnliftIO unlift <- askUnliftIO liftIO $ forever $ do threadDelay interval now <- systemSeconds <$> getSystemTime - cnt <- withPeriodicNtfTokens st now $ \tkn -> atomically $ writeTBQueue q (tkn, PNCheckMessages) + cnt <- withPeriodicNtfTokens st now $ \tkn -> unlift $ pushNotification s Nothing False tkn PNCheckMessages logNote $ "Scheduled periodic notifications: " <> tshow cnt runNtfClientTransport :: Transport c => THandleNTF c 'TServer -> M () diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 2bf413346..62e9deb67 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -33,6 +33,7 @@ import Control.Monad import Crypto.Random import Data.Functor (($>)) import Data.Int (Int64) +import Simplex.Messaging.Agent.RetryInterval import Data.List.NonEmpty (NonEmpty) import qualified Data.Text as T import Data.Time.Clock (getCurrentTime) @@ -167,7 +168,7 @@ data SMPSubscriber = SMPSubscriber } data NtfPushServer = NtfPushServer - { pushWorkers :: TMap (Maybe T.Text) PushWorkerVar, -- keyed by SMP server hostname, Nothing for non-server notifications + { pushWorkers :: TMap (Maybe T.Text, PushProvider) PushWorkerVar, pushWorkerSeq :: TVar Int, pushQSize :: Natural, pushClients :: TMap PushProvider PushClientVar, @@ -193,14 +194,22 @@ newNtfPushServer pushQSize apnsConfig = do pushClientSeq <- newTVarIO 0 pure NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize, pushClients, pushClientSeq, apnsConfig} - --- | Single-flight access to the per-provider push client. +-- | Single-flight access to the per-provider push client with bounded retry. -- The returned PushClientVar is the handle retryDeliver passes to removeSessVar to evict -- this specific instance before re-fetching. getPushClient :: NtfPushServer -> PushProvider -> IO (PushProviderClient, PushClientVar) -getPushClient s pp = do - ts <- getCurrentTime - atomically (getSessVar (pushClientSeq s) pp (pushClients s) ts) >>= either (newPushClient s pp) waitForPushClient +getPushClient s pp = + withRetryIntervalCount reconnectInterval $ \n _delay loop -> do + ts <- getCurrentTime + E.try (atomically (getSessVar (pushClientSeq s) pp (pushClients s) ts) >>= either (newPushClient s pp) waitForPushClient) >>= \case + Right result -> pure result + Left e + | n < 2 -> do + logError $ "getPushClient error (" <> tshow pp <> "): " <> tshow (e :: E.SomeException) + loop + | otherwise -> E.throwIO e + where + reconnectInterval = RetryInterval {initialInterval = 1_000_000, increaseAfter = 10_000_000, maxInterval = 10_000_000} newPushClient :: NtfPushServer -> PushProvider -> PushClientVar -> IO (PushProviderClient, PushClientVar) newPushClient NtfPushServer {pushClients, apnsConfig} pp v = do diff --git a/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs b/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs index e7e377f5c..48bc5e468 100644 --- a/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs +++ b/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs @@ -23,7 +23,7 @@ module Simplex.Messaging.Notifications.Server.Push.APNS apnsPushProviderClient, ) where -import Control.Exception (Exception) +import Control.Exception (Exception, throwIO) import Control.Logger.Simple import Control.Monad import Control.Monad.Except @@ -230,7 +230,7 @@ data APNSPushClient = APNSPushClient createAPNSPushClient :: HostName -> APNSPushClientConfig -> IO APNSPushClient createAPNSPushClient apnsHost apnsCfg@APNSPushClientConfig {authKeyFileEnv, authKeyAlg, authKeyIdEnv, appTeamId} = do https2Client <- newTVarIO Nothing - void $ connectHTTPS2 apnsHost apnsCfg https2Client + connectHTTPS2 apnsHost apnsCfg https2Client >>= either (throwIO . userError . show) (\_ -> pure ()) privateKey <- readECPrivateKey =<< getEnv authKeyFileEnv authKeyId <- T.pack <$> getEnv authKeyIdEnv let jwtHeader = JWTHeader {alg = authKeyAlg, kid = authKeyId}