ntf-server: single-flight push client creation via SessionVar

Match the take/create/wait pattern in Agent/Client.hs
(newProtocolClient / waitForProtocolClient). pushClients now wraps
clients in SessionVar (Either SomeException PushProviderClient) so
concurrent first-time access and concurrent retries collapse to a
single mkClient call; waiters observe the winner's result via
readTMVar (or its error). retryDeliver evicts the failing client by
SessionVar identity before re-fetching.

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

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