diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index 2c5d69b4a..5bd6e2064 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -669,12 +669,12 @@ ntfPush s@NtfPushServer {pushQ, srvDeliveryLocks} = forever $ do | otherwise = liftIO $ logError "bad notification token status" deliverNotification :: NtfPostgresStore -> PushProvider -> NtfTknRec -> PushNotification -> IO (Either PushProviderError ()) deliverNotification st pp tkn@NtfTknRec {ntfTknId} ntf = do - deliver <- getPushClient s pp + (deliver, clientVar) <- getPushClient s pp runExceptT (deliver tkn ntf) >>= \case Right _ -> pure $ Right () Left e -> case e of - PPConnection ce -> retryDeliver $ "connection " <> tshow ce - PPRetryLater r -> retryDeliver r + PPConnection ce -> retryDeliver clientVar $ "connection " <> tshow ce + PPRetryLater r -> retryDeliver clientVar r PPCryptoError _ -> err e PPResponseError {} -> err e PPTokenInvalid r -> do @@ -682,10 +682,13 @@ ntfPush s@NtfPushServer {pushQ, srvDeliveryLocks} = forever $ do err e PPPermanentError -> err e where - retryDeliver :: Text -> IO (Either PushProviderError ()) - retryDeliver reason = do + -- removeSessVar checks identity, so concurrent retries collapse: the first + -- one removes the failing client, subsequent ones observe a fresh replacement + retryDeliver :: PushClientVar -> Text -> IO (Either PushProviderError ()) + retryDeliver oldVar reason = do logWarn $ "retrying push (" <> tshow pp <> ", " <> tshow ntfTknId <> "): " <> reason - deliver <- newPushClient s pp + atomically $ removeSessVar oldVar pp (pushClients s) + (deliver, _) <- getPushClient s pp runExceptT (deliver tkn ntf) >>= \case Right _ -> pure $ Right () Left e -> case e of diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 8c514a228..626c36a41 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -13,19 +13,20 @@ module Simplex.Messaging.Notifications.Server.Env SMPSubscriberVar, SMPSubscriber (..), NtfPushServer (..), + PushClientVar, NtfRequest (..), NtfServerClient (..), defaultInactiveClientExpiration, newNtfServerEnv, newNtfSubscriber, newNtfPushServer, - newPushClient, getPushClient, getDeliveryLock, newNtfServerClient, ) where import Control.Concurrent (ThreadId) +import qualified Control.Exception as E import Control.Logger.Simple import Control.Monad import Crypto.Random @@ -167,35 +168,53 @@ data SMPSubscriber = SMPSubscriber data NtfPushServer = NtfPushServer { pushQ :: TBQueue (Maybe T.Text, NtfTknRec, PushNotification), -- Maybe Text is a hostname of "own" server - pushClients :: TMap PushProvider PushProviderClient, + pushClients :: TMap PushProvider PushClientVar, + pushClientSeq :: TVar Int, -- one lock per srvHost_ serializes per-server delivery while different servers proceed in parallel srvDeliveryLocks :: TMap (Maybe T.Text) (MVar ()), apnsConfig :: APNSPushClientConfig } +-- The Either communicates client-creation failure from the winner to the waiters. +type PushClientVar = SessionVar (Either E.SomeException PushProviderClient) + newNtfPushServer :: Natural -> APNSPushClientConfig -> IO NtfPushServer newNtfPushServer qSize apnsConfig = do pushQ <- newTBQueueIO qSize pushClients <- TM.emptyIO + pushClientSeq <- newTVarIO 0 srvDeliveryLocks <- TM.emptyIO - pure NtfPushServer {pushQ, pushClients, srvDeliveryLocks, apnsConfig} + pure NtfPushServer {pushQ, pushClients, pushClientSeq, srvDeliveryLocks, apnsConfig} getDeliveryLock :: TMap (Maybe T.Text) (MVar ()) -> Maybe T.Text -> IO (MVar ()) getDeliveryLock locks k = do newLock <- newMVar () atomically $ TM.lookup k locks >>= maybe (TM.insert k newLock locks $> newLock) pure -newPushClient :: NtfPushServer -> PushProvider -> IO PushProviderClient -newPushClient NtfPushServer {apnsConfig, pushClients} pp = do - c <- case apnsProviderHost pp of +-- | Single-flight access to the per-provider push client. +-- take (getSessVar) → create (newPushClient) or wait (waitForPushClient). +-- 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 + +newPushClient :: NtfPushServer -> PushProvider -> PushClientVar -> IO (PushProviderClient, PushClientVar) +newPushClient NtfPushServer {pushClients, apnsConfig} pp v = do + r <- E.try $ case apnsProviderHost pp of Nothing -> pure $ \_ _ -> pure () Just host -> apnsPushProviderClient <$> createAPNSPushClient host apnsConfig - atomically $ TM.insert pp c pushClients - pure c + atomically $ do + putTMVar (sessionVar v) r + case r of + Left _ -> removeSessVar v pp pushClients + Right _ -> pure () + either E.throwIO (\c -> pure (c, v)) r -getPushClient :: NtfPushServer -> PushProvider -> IO PushProviderClient -getPushClient s@NtfPushServer {pushClients} pp = - TM.lookupIO pp pushClients >>= maybe (newPushClient s pp) pure +waitForPushClient :: PushClientVar -> IO (PushProviderClient, PushClientVar) +waitForPushClient v = + atomically (readTMVar $ sessionVar v) >>= either E.throwIO (\c -> pure (c, v)) data NtfRequest = NtfReqNew CorrId ANewNtfEntity