retry connecting client

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 ()

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

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}