ntf: add APNS delivery metrics, rename NTFDIAG

src/Simplex/Messaging/Notifications/Server.hs

@@ -51,6 +51,7 @@ import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
+import GHC.Clock (getMonotonicTimeNSec)
 import GHC.IORef (atomicSwapIORef)
 import GHC.Conc (ThreadStatus (..), threadStatus)
 import GHC.Stats (GCDetails (..), RTSStats (..), getRTSStats)
@@ -122,7 +123,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions}
   raceAny_
     ( ntfSubscriber s
         : periodicNtfsThread ps
-        : leakDiagnosticsThread
+        : ntfDiagnosticsThread
         : map runServer transports
           <> serverStatsThread_ cfg
           <> prometheusMetricsThread_ cfg
@@ -689,23 +690,23 @@ runPushWorker s srvHost_ isOwn q = forever $ do
   (tkn@NtfTknRec {ntfTknId, token = t@(DeviceToken pp _), tknStatus}, ntf) <- atomically (readTBQueue q)
   liftIO $ logDebug $ "sending push notification to " <> T.pack (show pp)
   st <- asks store
+  stats <- asks serverStats
   case ntf of
     PNVerification _ ->
-      liftIO (deliverNotification st pp tkn ntf) >>= \case
+      liftIO (deliverNotification stats st pp tkn ntf) >>= \case
         Right _ -> do
           void $ liftIO $ setTknStatusConfirmed st tkn
           incNtfStatT t ntfVrfDelivered
         Left _ -> incNtfStatT t ntfVrfFailed
     PNCheckMessages ->
-      liftIO (deliverNotification st pp tkn ntf) >>= \case
+      liftIO (deliverNotification stats st pp tkn ntf) >>= \case
         Right _ -> do
           void $ liftIO $ updateTokenCronSentAt st ntfTknId . systemSeconds =<< getSystemTime
           incNtfStatT t ntfCronDelivered
         Left _ -> incNtfStatT t ntfCronFailed
     PNMessage {} -> checkActiveTkn tknStatus $ do
-      stats <- asks serverStats
       liftIO $ updatePeriodStats (activeTokens stats) ntfTknId
-      liftIO (deliverNotification st pp tkn ntf) >>= \case
+      liftIO (deliverNotification stats st pp tkn ntf) >>= \case
         Left _ -> do
           incNtfStatT t ntfFailed
           when isOwn $ liftIO $ mapM_ (`incServerStat` ntfFailedOwn stats) srvHost_
@@ -717,23 +718,38 @@ runPushWorker s srvHost_ isOwn q = forever $ do
     checkActiveTkn status action
       | status == NTActive = action
       | otherwise = liftIO $ logError "bad notification token status"
-    deliverNotification :: NtfPostgresStore -> PushProvider -> NtfTknRec -> PushNotification -> IO (Either PushProviderError ())
-    deliverNotification st pp tkn@NtfTknRec {ntfTknId} ntf' = do
-      (deliver, clientVar) <- getPushClient s pp
-      runExceptT (deliver tkn ntf') >>= \case
-        Right _ -> pure $ Right ()
-        Left e -> case e of
-          PPConnection ce -> retryDeliver clientVar $ "connection " <> tshow ce
-          PPRetryLater r -> retryDeliver clientVar r
-          PPCryptoError _ -> err e
-          PPResponseError {} -> err e
-          PPTokenInvalid r -> do
-            void $ updateTknStatus st tkn $ NTInvalid $ Just r
-            err e
-          PPPermanentError -> err e
+    deliverNotification :: NtfServerStats -> NtfPostgresStore -> PushProvider -> NtfTknRec -> PushNotification -> IO (Either PushProviderError ())
+    deliverNotification stats st pp tkn@NtfTknRec {ntfTknId} ntf' = do
+      t0 <- getMonotonicTimeNSec
+      res <- deliver1
+      t1 <- getMonotonicTimeNSec
+      incNtfStat_ stats apnsReqTotal
+      addNtfStat_ stats apnsLatencyMicros $ fromIntegral $ (t1 - t0) `div` 1000
+      either (incNtfStat_ stats . apnsErrSel) (const $ pure ()) res
+      pure res
       where
+        deliver1 = do
+          (deliver, clientVar) <- getPushClient s pp
+          runExceptT (deliver tkn ntf') >>= \case
+            Right _ -> pure $ Right ()
+            Left e -> case e of
+              PPConnection ce -> retryDeliver clientVar $ "connection " <> tshow ce
+              PPRetryLater r -> retryDeliver clientVar r
+              PPCryptoError _ -> err e
+              PPResponseError {} -> err e
+              PPTokenInvalid r -> do
+                void $ updateTknStatus st tkn $ NTInvalid $ Just r
+                err e
+              PPPermanentError -> err e
+        -- connection and retry-later failures stall the shared APNS connection and are the
+        -- likely push-queue-spike driver; everything else is a per-token/response reject
+        apnsErrSel = \case
+          PPConnection _ -> apnsErrConn
+          PPRetryLater _ -> apnsErrConn
+          _ -> apnsErrResponse
         retryDeliver :: PushClientVar -> Text -> IO (Either PushProviderError ())
         retryDeliver oldVar reason = do
+          incNtfStat_ stats apnsRetries
           logWarn $ "retrying push (" <> tshow pp <> ", " <> tshow ntfTknId <> "): " <> reason
           atomically $ removeSessVar oldVar pp (pushClients s)
           (deliver, _) <- getPushClient s pp
@@ -835,27 +851,27 @@ deadSubWorkers workers = do
         Nothing -> pure acc
         Just a -> maybe acc (const $ acc + 1) <$> poll a
 
--- Periodic leak diagnostics: a single greppable "LEAKDIAG" line censusing every growable
--- in-memory structure of the ntf server. Interval seconds via NTF_LEAKDIAG_SEC (default 60).
--- Whichever counter grows monotonically over time is the leak.
-leakDiagnosticsThread :: M ()
-leakDiagnosticsThread = do
-  secStr <- liftIO $ lookupEnv "NTF_LEAKDIAG_SEC"
+-- Periodic diagnostics: a single greppable "NTFDIAG" line censusing every growable
+-- in-memory structure of the ntf server. Interval seconds via NTF_DIAG_SEC (default 60).
+-- A counter that grows monotonically is a leak; one that spikes and recovers is backpressure.
+ntfDiagnosticsThread :: M ()
+ntfDiagnosticsThread = do
+  secStr <- liftIO $ lookupEnv "NTF_DIAG_SEC"
   -- clamped on both sides: a very small value would flood the log, and a very large one
   -- would overflow the microsecond conversion below and make threadDelay return at once
   let sec = min 86400 $ max 5 $ fromMaybe 60 (secStr >>= readMaybe)
   env <- ask
-  labelMyThread "leakDiagnosticsThread"
+  labelMyThread "ntfDiagnosticsThread"
   -- never let a diagnostics error crash the server (this thread is in raceAny_)
   liftIO $ forever $ do
     threadDelay $ sec * 1000000
-    tryAny (logNtfLeakStats env) >>= either (logError . ("LEAKDIAG error: " <>) . tshow) (const $ pure ())
+    tryAny (logNtfDiagStats env) >>= either (logError . ("NTFDIAG error: " <>) . tshow) (const $ pure ())
 
 -- Only structures that are NOT already exported as Prometheus metrics or via the control
 -- port are logged here - see Notifications/Server/Prometheus.hs for what is already tracked
 -- (push queue total, sub/service sub counts, SMP sessions, thread count, period stats).
-logNtfLeakStats :: NtfEnv -> IO ()
-logNtfLeakStats NtfEnv {subscriber, pushServer, store = NtfPostgresStore {dbStore}} = do
+logNtfDiagStats :: NtfEnv -> IO ()
+logNtfDiagStats NtfEnv {subscriber, pushServer, store = NtfPostgresStore {dbStore}} = do
   let NtfSubscriber {smpSubscribers, smpAgent = ca} = subscriber
       NtfPushServer {pushWorkers, pushQSize} = pushServer
       SMPClientAgent {msgQ, agentQ, agentCfg = SMPClientAgentConfig {msgQSize, agentQSize}} = ca
@@ -871,7 +887,7 @@ logNtfLeakStats NtfEnv {subscriber, pushServer, store = NtfPostgresStore {dbStor
   (liveBytes, memInUse) <- rtsBytes
   logNote $
     T.concat
-      [ "LEAKDIAG",
+      [ "NTFDIAG",
         -- push worker map only grows, and its key includes the client-supplied SMP host
         f "pushw_count" pwCount,
         -- workers whose thread has terminated: their queue is never drained again
@@ -1159,6 +1175,10 @@ incNtfStat_ :: NtfServerStats -> (NtfServerStats -> IORef Int) -> IO ()
 incNtfStat_ stats statSel = atomicModifyIORef'_ (statSel stats) (+ 1)
 {-# INLINE incNtfStat_ #-}
 
+addNtfStat_ :: NtfServerStats -> (NtfServerStats -> IORef Int) -> Int -> IO ()
+addNtfStat_ stats statSel n = atomicModifyIORef'_ (statSel stats) (+ n)
+{-# INLINE addNtfStat_ #-}
+
 restoreServerLastNtfs :: NtfSTMStore -> FilePath -> IO ()
 restoreServerLastNtfs st f =
   whenM (doesFileExist f) $ do

src/Simplex/Messaging/Notifications/Server/Prometheus.hs

@@ -95,7 +95,12 @@ ntfPrometheusMetrics sm rtm ts =
         _ntfVrfQueued,
         _ntfVrfDelivered,
         _ntfVrfFailed,
-        _ntfVrfInvalidTkn
+        _ntfVrfInvalidTkn,
+        _apnsReqTotal,
+        _apnsLatencyMicros,
+        _apnsRetries,
+        _apnsErrConn,
+        _apnsErrResponse
       } = statsData
     time =
       "# Recorded at: " <> T.pack (iso8601Show ts) <> "\n\
@@ -213,6 +218,26 @@ ntfPrometheusMetrics sm rtm ts =
       \# TYPE simplex_ntf_notifications_verification_invalid_tkn counter\n\
       \simplex_ntf_notifications_verification_invalid_tkn " <> mshow _ntfVrfInvalidTkn <> "\n# ntfVrfInvalidTkn\n\
       \\n\
+      \# HELP simplex_ntf_apns_requests_total APNS delivery attempts (including retried)\n\
+      \# TYPE simplex_ntf_apns_requests_total counter\n\
+      \simplex_ntf_apns_requests_total " <> mshow _apnsReqTotal <> "\n# apnsReqTotal\n\
+      \\n\
+      \# HELP simplex_ntf_apns_latency_micros_total Summed APNS delivery latency in microseconds\n\
+      \# TYPE simplex_ntf_apns_latency_micros_total counter\n\
+      \simplex_ntf_apns_latency_micros_total " <> mshow _apnsLatencyMicros <> "\n# apnsLatencyMicros\n\
+      \\n\
+      \# HELP simplex_ntf_apns_retries_total APNS deliveries retried after connection eviction\n\
+      \# TYPE simplex_ntf_apns_retries_total counter\n\
+      \simplex_ntf_apns_retries_total " <> mshow _apnsRetries <> "\n# apnsRetries\n\
+      \\n\
+      \# HELP simplex_ntf_apns_errors_connection_total APNS failures from connection or retry-later errors\n\
+      \# TYPE simplex_ntf_apns_errors_connection_total counter\n\
+      \simplex_ntf_apns_errors_connection_total " <> mshow _apnsErrConn <> "\n# apnsErrConn\n\
+      \\n\
+      \# HELP simplex_ntf_apns_errors_response_total APNS failures from response, token or permanent errors\n\
+      \# TYPE simplex_ntf_apns_errors_response_total counter\n\
+      \simplex_ntf_apns_errors_response_total " <> mshow _apnsErrResponse <> "\n# apnsErrResponse\n\
+      \\n\
       \# HELP simplex_ntf_notifications_total Total number of last notifications stored.\n\
       \# TYPE simplex_ntf_notifications_total gauge\n\
       \simplex_ntf_notifications_total " <> mshow lastNtfCount <> "\n# lastNtfCount\n\

src/Simplex/Messaging/Notifications/Server/Stats.hs

@@ -51,6 +51,11 @@ data NtfServerStats = NtfServerStats
     ntfVrfDelivered :: IORef Int,
     ntfVrfFailed :: IORef Int,
     ntfVrfInvalidTkn :: IORef Int,
+    apnsReqTotal :: IORef Int,
+    apnsLatencyMicros :: IORef Int,
+    apnsRetries :: IORef Int,
+    apnsErrConn :: IORef Int,
+    apnsErrResponse :: IORef Int,
     activeTokens :: PeriodStats,
     activeSubs :: PeriodStats
   }
@@ -77,6 +82,11 @@ data NtfServerStatsData = NtfServerStatsData
     _ntfVrfDelivered :: Int,
     _ntfVrfFailed :: Int,
     _ntfVrfInvalidTkn :: Int,
+    _apnsReqTotal :: Int,
+    _apnsLatencyMicros :: Int,
+    _apnsRetries :: Int,
+    _apnsErrConn :: Int,
+    _apnsErrResponse :: Int,
     _activeTokens :: PeriodStatsData,
     _activeSubs :: PeriodStatsData
   }
@@ -104,6 +114,11 @@ newNtfServerStats ts = do
   ntfVrfDelivered <- newIORef 0
   ntfVrfFailed <- newIORef 0
   ntfVrfInvalidTkn <- newIORef 0
+  apnsReqTotal <- newIORef 0
+  apnsLatencyMicros <- newIORef 0
+  apnsRetries <- newIORef 0
+  apnsErrConn <- newIORef 0
+  apnsErrResponse <- newIORef 0
   activeTokens <- newPeriodStats
   activeSubs <- newPeriodStats
   pure
@@ -129,6 +144,11 @@ newNtfServerStats ts = do
         ntfVrfDelivered,
         ntfVrfFailed,
         ntfVrfInvalidTkn,
+        apnsReqTotal,
+        apnsLatencyMicros,
+        apnsRetries,
+        apnsErrConn,
+        apnsErrResponse,
         activeTokens,
         activeSubs
       }
@@ -156,6 +176,11 @@ getNtfServerStatsData s@NtfServerStats {fromTime} = do
   _ntfVrfDelivered <- readIORef $ ntfVrfDelivered s
   _ntfVrfFailed <- readIORef $ ntfVrfFailed s
   _ntfVrfInvalidTkn <- readIORef $ ntfVrfInvalidTkn s
+  _apnsReqTotal <- readIORef $ apnsReqTotal s
+  _apnsLatencyMicros <- readIORef $ apnsLatencyMicros s
+  _apnsRetries <- readIORef $ apnsRetries s
+  _apnsErrConn <- readIORef $ apnsErrConn s
+  _apnsErrResponse <- readIORef $ apnsErrResponse s
   _activeTokens <- getPeriodStatsData $ activeTokens s
   _activeSubs <- getPeriodStatsData $ activeSubs s
   pure
@@ -181,6 +206,11 @@ getNtfServerStatsData s@NtfServerStats {fromTime} = do
         _ntfVrfDelivered,
         _ntfVrfFailed,
         _ntfVrfInvalidTkn,
+        _apnsReqTotal,
+        _apnsLatencyMicros,
+        _apnsRetries,
+        _apnsErrConn,
+        _apnsErrResponse,
         _activeTokens,
         _activeSubs
       }
@@ -209,6 +239,11 @@ setNtfServerStats s@NtfServerStats {fromTime} d@NtfServerStatsData {_fromTime} =
   writeIORef (ntfVrfDelivered s) $! _ntfVrfDelivered d
   writeIORef (ntfVrfFailed s) $! _ntfVrfFailed d
   writeIORef (ntfVrfInvalidTkn s) $! _ntfVrfInvalidTkn d
+  writeIORef (apnsReqTotal s) $! _apnsReqTotal d
+  writeIORef (apnsLatencyMicros s) $! _apnsLatencyMicros d
+  writeIORef (apnsRetries s) $! _apnsRetries d
+  writeIORef (apnsErrConn s) $! _apnsErrConn d
+  writeIORef (apnsErrResponse s) $! _apnsErrResponse d
   setPeriodStats (activeTokens s) (_activeTokens d)
   setPeriodStats (activeSubs s) (_activeSubs d)
 
@@ -236,6 +271,11 @@ instance StrEncoding NtfServerStatsData where
         _ntfVrfDelivered,
         _ntfVrfFailed,
         _ntfVrfInvalidTkn,
+        _apnsReqTotal,
+        _apnsLatencyMicros,
+        _apnsRetries,
+        _apnsErrConn,
+        _apnsErrResponse,
         _activeTokens,
         _activeSubs
       } =
@@ -261,6 +301,11 @@ instance StrEncoding NtfServerStatsData where
         "ntfVrfDelivered=" <> strEncode _ntfVrfDelivered,
         "ntfVrfFailed=" <> strEncode _ntfVrfFailed,
         "ntfVrfInvalidTkn=" <> strEncode _ntfVrfInvalidTkn,
+        "apnsReqTotal=" <> strEncode _apnsReqTotal,
+        "apnsLatencyMicros=" <> strEncode _apnsLatencyMicros,
+        "apnsRetries=" <> strEncode _apnsRetries,
+        "apnsErrConn=" <> strEncode _apnsErrConn,
+        "apnsErrResponse=" <> strEncode _apnsErrResponse,
         "activeTokens:",
         strEncode _activeTokens,
         "activeSubs:",
@@ -288,6 +333,11 @@ instance StrEncoding NtfServerStatsData where
     _ntfVrfDelivered <- opt "ntfVrfDelivered="
     _ntfVrfFailed <- opt "ntfVrfFailed="
     _ntfVrfInvalidTkn <- opt "ntfVrfInvalidTkn="
+    _apnsReqTotal <- opt "apnsReqTotal="
+    _apnsLatencyMicros <- opt "apnsLatencyMicros="
+    _apnsRetries <- opt "apnsRetries="
+    _apnsErrConn <- opt "apnsErrConn="
+    _apnsErrResponse <- opt "apnsErrResponse="
     _ <- "activeTokens:" <* A.endOfLine
     _activeTokens <- strP <* A.endOfLine
     _ <- "activeSubs:" <* A.endOfLine
@@ -315,6 +365,11 @@ instance StrEncoding NtfServerStatsData where
           _ntfVrfDelivered,
           _ntfVrfFailed,
           _ntfVrfInvalidTkn,
+          _apnsReqTotal,
+          _apnsLatencyMicros,
+          _apnsRetries,
+          _apnsErrConn,
+          _apnsErrResponse,
           _activeTokens,
           _activeSubs
         }