Re-commit: smp server: persist notifications to avoid losing them when ntf server is offline (#1336)

This reverts commit 0ba3e69872.

src/Simplex/Messaging/Notifications/Server.hs

@@ -220,9 +220,8 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
               NtfPushServer {pushQ} <- asks pushServer
               stats <- asks serverStats
               liftIO $ updatePeriodStats (activeSubs stats) ntfId
-              atomically $
-                findNtfSubscriptionToken st smpQueue
-                  >>= mapM_ (\tkn -> writeTBQueue pushQ (tkn, PNMessage (PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} :| [])))
+              atomically (findNtfSubscriptionToken st smpQueue)
+                >>= mapM_ (\tkn -> atomically (writeTBQueue pushQ (tkn, PNMessage (PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} :| []))))
               incNtfStat ntfReceived
             Right SMP.END ->
               whenM (atomically $ activeClientSession' ca sessionId srv) $

src/Simplex/Messaging/Server.hs

@@ -58,7 +58,7 @@ import Data.IORef
 import Data.Int (Int64)
 import qualified Data.IntMap.Strict as IM
 import qualified Data.IntSet as IS
-import Data.List (intercalate, mapAccumR)
+import Data.List (foldl', intercalate, mapAccumR)
 import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
@@ -88,6 +88,7 @@ import Simplex.Messaging.Server.Env.STM as Env
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore
 import Simplex.Messaging.Server.MsgStore.STM
+import Simplex.Messaging.Server.NtfStore
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Server.QueueStore.STM as QS
@@ -137,13 +138,16 @@ smpServer :: TMVar Bool -> ServerConfig -> Maybe AttachHTTP -> M ()
 smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHTTP_ = do
   s <- asks server
   pa <- asks proxyAgent
-  expired <- restoreServerMessages
-  restoreServerStats expired
+  expiredMsgs <- restoreServerMessages
+  expiredNtfs <- restoreServerNtfs
+  restoreServerStats expiredMsgs expiredNtfs
   raceAny_
     ( serverThread s "server subscribedQ" subscribedQ subscribers subClients pendingSubEvents subscriptions cancelSub
         : serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubClients pendingNtfSubEvents ntfSubscriptions (\_ -> pure ())
+        : deliverNtfsThread s
         : sendPendingEvtsThread s
         : receiveFromProxyAgent pa
+        : expireNtfsThread cfg
         : map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
     )
     `finally` withLock' (savingLock s) "final" (saveServer False >> closeServer)
@@ -171,7 +175,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
     fromTLSCredentials (_, pk) = C.x509ToPrivate (pk, []) >>= C.privKey
 
     saveServer :: Bool -> M ()
-    saveServer keepMsgs = withLog closeStoreLog >> saveServerMessages keepMsgs >> saveServerStats
+    saveServer keepMsgs = withLog closeStoreLog >> saveServerMessages keepMsgs >> saveServerNtfs >> saveServerStats
 
     closeServer :: M ()
     closeServer = asks (smpAgent . proxyAgent) >>= liftIO . closeSMPClientAgent
@@ -230,6 +234,31 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
         -- remove client from server's subscribed cients
         removeWhenNoSubs c = whenM (null <$> readTVar (clientSubs c)) $ modifyTVar' (subClnts s) $ IM.delete (clientId c)
 
+    deliverNtfsThread :: Server -> M ()
+    deliverNtfsThread Server {ntfSubClients} = do
+      ntfInt <- asks $ ntfDeliveryInterval . config
+      ns <- asks ntfStore
+      stats <- asks serverStats
+      liftIO $ forever $ do
+        threadDelay ntfInt
+        readTVarIO ntfSubClients >>= mapM_ (deliverNtfs ns stats)
+      where
+        deliverNtfs ns stats Client {clientId, ntfSubscriptions, sndQ, connected} = whenM currentClient $
+          readTVarIO ntfSubscriptions >>= \subs -> do
+            ts_ <- foldM addNtfs [] (M.keys subs)
+            mapM_ (atomically . writeTBQueue sndQ) $ L.nonEmpty ts_
+            updateNtfStats $ length ts_
+          where
+            currentClient = (&&) <$> readTVarIO connected <*> (IM.member clientId <$> readTVarIO ntfSubClients)
+            addNtfs :: [Transmission BrokerMsg] -> NotifierId -> IO [Transmission BrokerMsg]
+            addNtfs acc nId =
+              (foldl' (\acc' ntf -> nmsg nId ntf : acc') acc) -- reverses, to order by time
+                <$> flushNtfs ns nId
+            nmsg nId MsgNtf {ntfNonce, ntfEncMeta} = (CorrId "", nId, NMSG ntfNonce ntfEncMeta)
+            updateNtfStats len = when (len > 0) $ liftIO $ do
+              atomicModifyIORef'_ (msgNtfs stats) (+ len)
+              atomicModifyIORef'_ (msgNtfsB stats) (+ (len `div` 80 + 1)) -- up to 80 NMSG in the batch
+
     sendPendingEvtsThread :: Server -> M ()
     sendPendingEvtsThread s = do
       endInt <- asks $ pendingENDInterval . config
@@ -259,8 +288,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
             updateEndStats = do
               stats <- asks serverStats
               let len = L.length qEvts
-              liftIO $ atomicModifyIORef'_ (qSubEnd stats) (+ len)
-              liftIO $ atomicModifyIORef'_ (qSubEndB stats) (+ (len `div` 255 + 1)) -- up to 255 ENDs or DELDs in the batch
+              when (len > 0) $ liftIO $ do
+                atomicModifyIORef'_ (qSubEnd stats) (+ len)
+                atomicModifyIORef'_ (qSubEndB stats) (+ (len `div` 255 + 1)) -- up to 255 ENDs or DELDs in the batch
 
     receiveFromProxyAgent :: ProxyAgent -> M ()
     receiveFromProxyAgent ProxyAgent {smpAgent = SMPClientAgent {agentQ}} =
@@ -294,6 +324,18 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
           deleted <- liftIO $ deleteExpiredMsgs q old
           liftIO $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
 
+    expireNtfsThread :: ServerConfig -> M ()
+    expireNtfsThread ServerConfig {notificationExpiration = expCfg} = do
+      ns <- asks ntfStore
+      let interval = checkInterval expCfg * 1000000
+      stats <- asks serverStats
+      labelMyThread "expireNtfsThread"
+      liftIO $ forever $ do
+        threadDelay' interval
+        old <- expireBeforeEpoch expCfg
+        expired <- deleteExpiredNtfs ns old
+        when (expired > 0) $ atomicModifyIORef'_ (msgNtfExpired stats) (+ expired)
+
     serverStatsThread_ :: ServerConfig -> [M ()]
     serverStatsThread_ ServerConfig {logStatsInterval = Just interval, logStatsStartTime, serverStatsLogFile} =
       [logServerStats logStatsStartTime interval serverStatsLogFile]
@@ -351,8 +393,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
           msgRecvNtf' <- atomicSwapIORef msgRecvNtf 0
           psNtf <- liftIO $ periodStatCounts activeQueuesNtf ts
           msgNtfs' <- atomicSwapIORef (msgNtfs ss) 0
+          msgNtfsB' <- atomicSwapIORef (msgNtfsB ss) 0
           msgNtfNoSub' <- atomicSwapIORef (msgNtfNoSub ss) 0
           msgNtfLost' <- atomicSwapIORef (msgNtfLost ss) 0
+          msgNtfExpired' <- atomicSwapIORef (msgNtfExpired ss) 0
           pRelays' <- getResetProxyStatsData pRelays
           pRelaysOwn' <- getResetProxyStatsData pRelaysOwn
           pMsgFwds' <- getResetProxyStatsData pMsgFwds
@@ -424,7 +468,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                        show qSubEnd',
                        show qSubEndB',
                        show ntfDeletedB',
-                       show ntfSubB'
+                       show ntfSubB',
+                       show msgNtfsB',
+                       show msgNtfExpired'
                      ]
               )
         liftIO $ threadDelay' interval
@@ -534,6 +580,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                 hPutStrLn h $ "other GET events (auth, duplicate, prohibited): " <> show gets
                 putStat "msgSentNtf" msgSentNtf
                 putStat "msgRecvNtf" msgRecvNtf
+                putStat "msgNtfs" msgNtfs
+                putStat "msgNtfsB" msgNtfsB
+                putStat "msgNtfExpired" msgNtfExpired
                 putStat "qCount" qCount
                 putStat "msgCount" msgCount
                 putProxyStat "pRelays" pRelays
@@ -932,7 +981,7 @@ forkClient Client {endThreads, endThreadSeq} label action = do
   mkWeakThreadId t >>= atomically . modifyTVar' endThreads . IM.insert tId
 
 client :: THandleParams SMPVersion 'TServer -> Client -> Server -> M ()
-client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId, procThreads} Server {subscribedQ, ntfSubscribedQ, subscribers, notifiers} = do
+client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId, procThreads} Server {subscribedQ, ntfSubscribedQ, subscribers} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
   forever $
     atomically (readTBQueue rcvQ)
@@ -1135,6 +1184,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
           liftIO (deleteQueueNotifier st entId) >>= \case
             Right (Just nId) -> do
               -- Possibly, the same should be done if the queue is suspended, but currently we do not use it
+              asks ntfStore >>= liftIO . (`deleteNtfs` nId)
               atomically $ writeTQueue ntfSubscribedQ (nId, clientId, False)
               incStat . ntfDeleted =<< asks serverStats
               pure ok
@@ -1276,7 +1326,14 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
               Message {msgFlags} -> do
                 stats <- asks serverStats
                 incStat $ msgRecv stats
-                when isGet $ incStat $ msgRecvGet stats
+                if isGet
+                  then incStat $ msgRecvGet stats
+                  else pure () -- TODO skip notification delivery for delivered message  
+                  -- skipping delivery fails tests, it should be counted in msgNtfSkipped
+                  -- forM_ (notifierId <$> notifier qr) $ \nId -> do
+                  --   ns <- asks ntfStore
+                  --   atomically $ TM.lookup nId ns >>=
+                  --     mapM_ (\MsgNtf {ntfMsgId} -> when (msgId == msgId') $ TM.delete nId ns)
                 liftIO $ atomicModifyIORef'_ (msgCount stats) (subtract 1)
                 liftIO $ updatePeriodStats (activeQueues stats) entId
                 when (notification msgFlags) $ do
@@ -1310,7 +1367,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                         Just (msg, wasEmpty) -> time "SEND ok" $ do
                           when wasEmpty $ liftIO $ tryDeliverMessage msg
                           when (notification msgFlags) $ do
-                            mapM_ (`trySendNotification` msg) (notifier qr)
+                            mapM_ (`enqueueNotification` msg) (notifier qr)
                             incStat $ msgSentNtf stats
                             liftIO $ updatePeriodStats (activeQueuesNtf stats) (recipientId qr)
                           incStat $ msgSent stats
@@ -1395,39 +1452,20 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                             deliver q s
                             writeTVar st NoSub
 
-            trySendNotification :: NtfCreds -> Message -> M ()
-            trySendNotification NtfCreds {notifierId, rcvNtfDhSecret} msg = do
-              stats <- asks serverStats
-              liftIO (TM.lookupIO notifierId notifiers) >>= \case
-                Nothing -> do
-                  incStat $ msgNtfNoSub stats
-                  logWarn "No notification subscription"
-                Just ntfClnt -> do
-                  let updateStats True = incStat $ msgNtfs stats
-                      updateStats _ = do
-                        incStat $ msgNtfLost stats
-                        logWarn "Dropped message notification"
-                  writeNtf notifierId msg rcvNtfDhSecret ntfClnt >>= mapM_ updateStats
+            enqueueNotification :: NtfCreds -> Message -> M ()
+            enqueueNotification _ MessageQuota {} = pure ()
+            enqueueNotification NtfCreds {notifierId = nId, rcvNtfDhSecret} Message {msgId, msgTs} = do
+              -- stats <- asks serverStats
+              ns <- asks ntfStore
+              ntf <- mkMessageNotification msgId msgTs rcvNtfDhSecret
+              liftIO $ storeNtf ns nId ntf
 
-            writeNtf :: NotifierId -> Message -> RcvNtfDhSecret -> TVar Client -> M (Maybe Bool)
-            writeNtf nId msg rcvNtfDhSecret ntfClnt = case msg of
-              Message {msgId, msgTs} -> Just <$> do
-                (nmsgNonce, encNMsgMeta) <- mkMessageNotification msgId msgTs rcvNtfDhSecret
-                -- must be in one STM transaction to avoid the queue becoming full between the check and writing
-                atomically $ do
-                  Client {sndQ = q} <- readTVar ntfClnt
-                  ifM
-                    (isFullTBQueue q)
-                    (pure $ False)
-                    (True <$ writeTBQueue q [(CorrId "", nId, NMSG nmsgNonce encNMsgMeta)])
-              _ -> pure Nothing
-
-            mkMessageNotification :: ByteString -> SystemTime -> RcvNtfDhSecret -> M (C.CbNonce, EncNMsgMeta)
+            mkMessageNotification :: ByteString -> SystemTime -> RcvNtfDhSecret -> M MsgNtf
             mkMessageNotification msgId msgTs rcvNtfDhSecret = do
-              cbNonce <- atomically . C.randomCbNonce =<< asks random
+              ntfNonce <- atomically . C.randomCbNonce =<< asks random
               let msgMeta = NMsgMeta {msgId, msgTs}
-                  encNMsgMeta = C.cbEncrypt rcvNtfDhSecret cbNonce (smpEncode msgMeta) 128
-              pure . (cbNonce,) $ fromRight "" encNMsgMeta
+                  encNMsgMeta = C.cbEncrypt rcvNtfDhSecret ntfNonce (smpEncode msgMeta) 128
+              pure $ MsgNtf {ntfMsgId = msgId, ntfTs = msgTs, ntfNonce, ntfEncMeta = fromRight "" encNMsgMeta}
 
         processForwardedCommand :: EncFwdTransmission -> M BrokerMsg
         processForwardedCommand (EncFwdTransmission s) = fmap (either ERR id) . runExceptT $ do
@@ -1528,9 +1566,10 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                 -- queue is usually deleted by the same client that is currently subscribed,
                 -- we delete subscription here, so the client with no subscriptions can be disconnected.
                 TM.delete entId subscriptions
-              forM_ (notifierId <$> notifier q) $ \nId ->
+              forM_ (notifierId <$> notifier q) $ \nId -> do
                 -- queue is deleted by a different client from the one subscribed to notifications,
                 -- so we don't need to remove subscription from the current client.
+                asks ntfStore >>= liftIO . (`deleteNtfs` nId)
                 atomically $ writeTQueue ntfSubscribedQ (nId, clientId, False)
               updateDeletedStats q
               pure ok
@@ -1657,6 +1696,50 @@ restoreServerMessages =
             msgErr :: Show e => String -> e -> String
             msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
 
+saveServerNtfs :: M ()
+saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs
+  where
+    saveNtfs f = do
+      logInfo $ "saving notifications to file " <> T.pack f
+      NtfStore ns <- asks ntfStore
+      liftIO . withFile f WriteMode $ \h ->
+        readTVarIO ns >>= mapM_ (saveQueueNtfs h) . M.assocs
+      logInfo "notifications saved"
+      where
+        -- reverse on save, to save notifications in order, will become reversed again when restoring.
+        saveQueueNtfs h (nId, v) = BLD.hPutBuilder h . encodeNtfs nId . reverse =<< readTVarIO v
+        encodeNtfs nId = mconcat . map (\ntf -> BLD.byteString (strEncode $ NLRv1 nId ntf) <> BLD.char8 '\n')
+
+restoreServerNtfs :: M Int
+restoreServerNtfs =
+  asks (storeNtfsFile . config) >>= \case
+    Just f -> ifM (doesFileExist f) (restoreNtfs f) (pure 0)
+    Nothing -> pure 0
+  where
+    restoreNtfs f = do
+      logInfo $ "restoring notifications from file " <> T.pack f
+      ns <- asks ntfStore
+      old <- asks (notificationExpiration . config) >>= liftIO . expireBeforeEpoch
+      runExceptT (liftIO (LB.readFile f) >>= foldM (restoreNtf ns old) 0 . LB.lines) >>= \case
+        Left e -> do
+          logError . T.pack $ "error restoring notifications: " <> e
+          liftIO exitFailure
+        Right expired -> do
+          renameFile f $ f <> ".bak"
+          logInfo "notifications restored"
+          pure expired
+      where
+        restoreNtf ns old !expired s' = do
+          NLRv1 nId ntf <- liftEither . first (ntfErr "parsing") $ strDecode s
+          liftIO $ addToNtfs nId ntf
+          where
+            s = LB.toStrict s'
+            addToNtfs nId ntf@MsgNtf {ntfTs}
+              | systemSeconds ntfTs < old = pure (expired + 1)
+              | otherwise = storeNtf ns nId ntf $> expired
+            ntfErr :: Show e => String -> e -> String
+            ntfErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
+
 saveServerStats :: M ()
 saveServerStats =
   asks (serverStatsBackupFile . config)
@@ -1667,8 +1750,8 @@ saveServerStats =
       B.writeFile f $ strEncode stats
       logInfo "server stats saved"
 
-restoreServerStats :: Int -> M ()
-restoreServerStats expiredWhileRestoring = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
+restoreServerStats :: Int -> Int -> M ()
+restoreServerStats expiredMsgs expiredNtfs = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
   where
     restoreStats f = whenM (doesFileExist f) $ do
       logInfo $ "restoring server stats from file " <> T.pack f
@@ -1677,7 +1760,7 @@ restoreServerStats expiredWhileRestoring = asks (serverStatsBackupFile . config)
           s <- asks serverStats
           _qCount <- fmap M.size . readTVarIO . queues =<< asks queueStore
           _msgCount <- liftIO . foldM (\(!n) q -> (n +) <$> getQueueSize q) 0 =<< readTVarIO =<< asks msgStore
-          liftIO $ setServerStats s d {_qCount, _msgCount, _msgExpired = _msgExpired d + expiredWhileRestoring}
+          liftIO $ setServerStats s d {_qCount, _msgCount, _msgExpired = _msgExpired d + expiredMsgs, _msgNtfExpired = _msgNtfExpired d + expiredNtfs}
           renameFile f $ f <> ".bak"
           logInfo "server stats restored"
           when (_qCount /= statsQCount) $ logWarn $ "Queue count differs: stats: " <> tshow statsQCount <> ", store: " <> tshow _qCount

src/Simplex/Messaging/Server/Env/STM.hs

@@ -38,6 +38,7 @@ import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
 import Simplex.Messaging.Server.MsgStore.STM
+import Simplex.Messaging.Server.NtfStore
 import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..))
 import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.Stats
@@ -61,6 +62,7 @@ data ServerConfig = ServerConfig
     msgIdBytes :: Int,
     storeLogFile :: Maybe FilePath,
     storeMsgsFile :: Maybe FilePath,
+    storeNtfsFile :: Maybe FilePath,
     -- | set to False to prohibit creating new queues
     allowNewQueues :: Bool,
     -- | simple password that the clients need to pass in handshake to be able to create new queues
@@ -70,6 +72,8 @@ data ServerConfig = ServerConfig
     controlPortAdminAuth :: Maybe BasicAuth,
     -- | time after which the messages can be removed from the queues and check interval, seconds
     messageExpiration :: Maybe ExpirationConfig,
+    -- | notification expiration interval (seconds)
+    notificationExpiration :: ExpirationConfig,
     -- | time after which the socket with inactive client can be disconnected (without any messages or commands, incl. PING),
     -- and check interval, seconds
     inactiveClientExpiration :: Maybe ExpirationConfig,
@@ -82,6 +86,8 @@ data ServerConfig = ServerConfig
     serverStatsLogFile :: FilePath,
     -- | file to save and restore stats
     serverStatsBackupFile :: Maybe FilePath,
+    -- | notification delivery interval
+    ntfDeliveryInterval :: Int,
     -- | interval between sending pending END events to unsubscribed clients, seconds
     pendingENDInterval :: Int,
     smpCredentials :: ServerCredentials,
@@ -110,6 +116,16 @@ defaultMessageExpiration =
       checkInterval = 43200 -- seconds, 12 hours
     }
 
+defNtfExpirationHours :: Int64
+defNtfExpirationHours = 24
+
+defaultNtfExpiration :: ExpirationConfig
+defaultNtfExpiration =
+  ExpirationConfig
+    { ttl = defNtfExpirationHours * 3600, -- seconds
+      checkInterval = 3600 -- seconds, 1 hour
+    }
+
 defaultInactiveClientExpiration :: ExpirationConfig
 defaultInactiveClientExpiration =
   ExpirationConfig
@@ -127,6 +143,7 @@ data Env = Env
     serverIdentity :: KeyHash,
     queueStore :: QueueStore,
     msgStore :: STMMsgStore,
+    ntfStore :: NtfStore,
     random :: TVar ChaChaDRG,
     storeLog :: Maybe (StoreLog 'WriteMode),
     tlsServerCreds :: T.Credential,
@@ -229,6 +246,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAg
   server <- newServer
   queueStore <- newQueueStore
   msgStore <- newMsgStore
+  ntfStore <- NtfStore <$> TM.emptyIO
   random <- C.newRandom
   storeLog <-
     forM storeLogFile $ \f -> do
@@ -244,7 +262,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAg
   clientSeq <- newTVarIO 0
   clients <- newTVarIO mempty
   proxyAgent <- newSMPProxyAgent smpAgentCfg random
-  pure Env {config, serverInfo, server, serverIdentity, queueStore, msgStore, random, storeLog, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
+  pure Env {config, serverInfo, server, serverIdentity, queueStore, msgStore, ntfStore, random, storeLog, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
   where
     getCredentials protocol creds = do
       files <- missingCreds

src/Simplex/Messaging/Server/Main.hs

@@ -38,7 +38,7 @@ import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (BasicAuth (..), ProtoServerWithAuth (ProtoServerWithAuth), pattern SMPServer)
 import Simplex.Messaging.Server (AttachHTTP, runSMPServer)
 import Simplex.Messaging.Server.CLI
-import Simplex.Messaging.Server.Env.STM (ServerConfig (..), defMsgExpirationDays, defaultInactiveClientExpiration, defaultMessageExpiration, defaultProxyClientConcurrency)
+import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
 import Simplex.Messaging.Transport (batchCmdsSMPVersion, sendingProxySMPVersion, simplexMQVersion, supportedServerSMPRelayVRange)
@@ -154,7 +154,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 <> "# Undelivered messages are optionally saved and restored when the server restarts,\n\
                    \# they are preserved in the .bak file until the next restart.\n"
                 <> ("restore_messages: " <> onOff enableStoreLog <> "\n")
-                <> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n\n")
+                <> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n")
+                <> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
                 <> "# Log daily server statistics to CSV file\n"
                 <> ("log_stats: " <> onOff logStats <> "\n\n")
                 <> "[AUTH]\n\
@@ -268,6 +269,11 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
         enableStoreLog = settingIsOn "STORE_LOG" "enable" ini
         logStats = settingIsOn "STORE_LOG" "log_stats" ini
         c = combine cfgPath . ($ defaultX509Config)
+        restoreMessagesFile path = case iniOnOff "STORE_LOG" "restore_messages" ini of
+          Just True -> Just path
+          Just False -> Nothing
+          -- if the setting is not set, it is enabled when store log is enabled
+          _ -> enableStoreLog $> path
         transports = iniTransports ini
         sharedHTTP = any (\(_, _, addHTTP) -> addHTTP) transports
         serverConfig =
@@ -286,13 +292,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                   },
               httpCredentials = (\WebHttpsParams {key, cert} -> ServerCredentials {caCertificateFile = Nothing, privateKeyFile = key, certificateFile = cert}) <$> webHttpsParams',
               storeLogFile = enableStoreLog $> storeLogFilePath,
-              storeMsgsFile =
-                let messagesPath = combine logPath "smp-server-messages.log"
-                 in case iniOnOff "STORE_LOG" "restore_messages" ini of
-                      Just True -> Just messagesPath
-                      Just False -> Nothing
-                      -- if the setting is not set, it is enabled when store log is enabled
-                      _ -> enableStoreLog $> messagesPath,
+              storeMsgsFile = restoreMessagesFile $ combine logPath "smp-server-messages.log",
+              storeNtfsFile = restoreMessagesFile $ combine logPath "smp-server-ntfs.log",
               -- allow creating new queues by default
               allowNewQueues = fromMaybe True $ iniOnOff "AUTH" "new_queues" ini,
               newQueueBasicAuth = either error id <$!> strDecodeIni "AUTH" "create_password" ini,
@@ -303,6 +304,10 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                   defaultMessageExpiration
                     { ttl = 86400 * readIniDefault defMsgExpirationDays "STORE_LOG" "expire_messages_days" ini
                     },
+              notificationExpiration =
+                defaultNtfExpiration
+                  { ttl = 3600 * readIniDefault defNtfExpirationHours "STORE_LOG" "expire_ntfs_hours" ini
+                  },
               inactiveClientExpiration =
                 settingIsOn "INACTIVE_CLIENTS" "disconnect" ini
                   $> ExpirationConfig
@@ -314,6 +319,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
               serverStatsLogFile = combine logPath "smp-server-stats.daily.log",
               serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log",
               pendingENDInterval = 15000000, -- 15 seconds
+              ntfDeliveryInterval = 3000000, -- 3 seconds
               smpServerVRange = supportedServerSMPRelayVRange,
               transportConfig =
                 defaultTransportServerConfig

src/Simplex/Messaging/Server/NtfStore.hs

@@ -0,0 +1,77 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Server.NtfStore where
+
+import Control.Concurrent.STM
+import Control.Monad (foldM)
+import Data.Int (Int64)
+import qualified Data.Map.Strict as M
+import Data.Time.Clock.System (SystemTime (..))
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Encoding.String
+import Simplex.Messaging.Protocol (EncNMsgMeta, MsgId, NotifierId)
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+
+newtype NtfStore = NtfStore (TMap NotifierId (TVar [MsgNtf]))
+
+data MsgNtf = MsgNtf
+  { ntfMsgId :: MsgId,
+    ntfTs :: SystemTime,
+    ntfNonce :: C.CbNonce,
+    ntfEncMeta :: EncNMsgMeta
+  }
+
+storeNtf :: NtfStore -> NotifierId -> MsgNtf -> IO ()
+storeNtf (NtfStore ns) nId ntf = do
+  TM.lookupIO nId ns >>= atomically . maybe newNtfs (`modifyTVar'` (ntf :))
+  -- TODO coalesce messages here once the client is updated to process multiple messages
+  -- for single notification.
+  -- when (isJust prevNtf) $ incStat $ msgNtfReplaced stats
+  where
+    newNtfs = TM.lookup nId ns >>= maybe (TM.insertM nId (newTVar [ntf]) ns) (`modifyTVar'` (ntf :))
+
+deleteNtfs :: NtfStore -> NotifierId -> IO ()
+deleteNtfs (NtfStore ns) nId = atomically $ TM.delete nId ns
+
+flushNtfs :: NtfStore -> NotifierId -> IO [MsgNtf]
+flushNtfs (NtfStore ns) nId = do
+  TM.lookupIO nId ns >>= maybe (pure []) swapNtfs
+  where
+    swapNtfs v =
+      readTVarIO v >>= \case
+        [] -> pure []
+        -- if notifications available, atomically swap with empty array
+        _ -> atomically (swapTVar v [])
+
+deleteExpiredNtfs :: NtfStore -> Int64 -> IO Int
+deleteExpiredNtfs (NtfStore ns) old =
+  foldM (\expired -> fmap (expired +) . expireQueue) 0 . M.keys =<< readTVarIO ns
+  where
+    expireQueue nId = TM.lookupIO nId ns >>= maybe (pure 0) expire
+    expire v = readTVarIO v >>= \case
+      [] -> pure 0
+      _ ->
+        atomically $ readTVar v >>= \case
+          [] -> pure 0
+          -- check the last message first, it is the earliest
+          ntfs | systemSeconds (ntfTs $ last $ ntfs) < old -> do
+            let !ntfs' = filter (\MsgNtf {ntfTs = ts} -> systemSeconds ts >= old) ntfs
+            writeTVar v ntfs'
+            pure $! length ntfs - length ntfs'
+          _ -> pure 0
+
+data NtfLogRecord = NLRv1 NotifierId MsgNtf
+
+instance StrEncoding MsgNtf where
+  strEncode MsgNtf {ntfMsgId, ntfTs, ntfNonce, ntfEncMeta} = strEncode (ntfMsgId, ntfTs, ntfNonce, ntfEncMeta)
+  strP = do
+    (ntfMsgId, ntfTs, ntfNonce, ntfEncMeta) <- strP
+    pure MsgNtf {ntfMsgId, ntfTs, ntfNonce, ntfEncMeta}
+
+instance StrEncoding NtfLogRecord where
+  strEncode (NLRv1 nId ntf) = strEncode (Str "v1", nId, ntf)
+  strP = "v1 " *> (NLRv1 <$> strP_ <*> strP)

src/Simplex/Messaging/Server/Stats.hs

@@ -67,8 +67,10 @@ data ServerStats = ServerStats
     msgRecvNtf :: IORef Int, -- received messages with NTF flag
     activeQueuesNtf :: PeriodStats,
     msgNtfs :: IORef Int, -- messages notications delivered to NTF server (<= msgSentNtf)
+    msgNtfsB :: IORef Int, -- messages notication batches delivered to NTF server
     msgNtfNoSub :: IORef Int, -- no subscriber to notifications (e.g., NTF server not connected)
     msgNtfLost :: IORef Int, -- notification is lost because NTF delivery queue is full
+    msgNtfExpired :: IORef Int, -- expired
     pRelays :: ProxyStats,
     pRelaysOwn :: ProxyStats,
     pMsgFwds :: ProxyStats,
@@ -117,8 +119,10 @@ data ServerStatsData = ServerStatsData
     _msgRecvNtf :: Int,
     _activeQueuesNtf :: PeriodStatsData,
     _msgNtfs :: Int,
+    _msgNtfsB :: Int,
     _msgNtfNoSub :: Int,
     _msgNtfLost :: Int,
+    _msgNtfExpired :: Int,
     _pRelays :: ProxyStatsData,
     _pRelaysOwn :: ProxyStatsData,
     _pMsgFwds :: ProxyStatsData,
@@ -169,8 +173,10 @@ newServerStats ts = do
   msgRecvNtf <- newIORef 0
   activeQueuesNtf <- newPeriodStats
   msgNtfs <- newIORef 0
+  msgNtfsB <- newIORef 0
   msgNtfNoSub <- newIORef 0
   msgNtfLost <- newIORef 0
+  msgNtfExpired <- newIORef 0
   pRelays <- newProxyStats
   pRelaysOwn <- newProxyStats
   pMsgFwds <- newProxyStats
@@ -218,8 +224,10 @@ newServerStats ts = do
         msgRecvNtf,
         activeQueuesNtf,
         msgNtfs,
+        msgNtfsB,
         msgNtfNoSub,
         msgNtfLost,
+        msgNtfExpired,
         pRelays,
         pRelaysOwn,
         pMsgFwds,
@@ -269,8 +277,10 @@ getServerStatsData s = do
   _msgRecvNtf <- readIORef $ msgRecvNtf s
   _activeQueuesNtf <- getPeriodStatsData $ activeQueuesNtf s
   _msgNtfs <- readIORef $ msgNtfs s
+  _msgNtfsB <- readIORef $ msgNtfsB s
   _msgNtfNoSub <- readIORef $ msgNtfNoSub s
   _msgNtfLost <- readIORef $ msgNtfLost s
+  _msgNtfExpired <- readIORef $ msgNtfExpired s
   _pRelays <- getProxyStatsData $ pRelays s
   _pRelaysOwn <- getProxyStatsData $ pRelaysOwn s
   _pMsgFwds <- getProxyStatsData $ pMsgFwds s
@@ -318,8 +328,10 @@ getServerStatsData s = do
         _msgRecvNtf,
         _activeQueuesNtf,
         _msgNtfs,
+        _msgNtfsB,
         _msgNtfNoSub,
         _msgNtfLost,
+        _msgNtfExpired,
         _pRelays,
         _pRelaysOwn,
         _pMsgFwds,
@@ -370,8 +382,10 @@ setServerStats s d = do
   writeIORef (msgRecvNtf s) $! _msgRecvNtf d
   setPeriodStats (activeQueuesNtf s) (_activeQueuesNtf d)
   writeIORef (msgNtfs s) $! _msgNtfs d
+  writeIORef (msgNtfsB s) $! _msgNtfsB d
   writeIORef (msgNtfNoSub s) $! _msgNtfNoSub d
   writeIORef (msgNtfLost s) $! _msgNtfLost d
+  writeIORef (msgNtfExpired s) $! _msgNtfExpired d
   setProxyStats (pRelays s) $! _pRelays d
   setProxyStats (pRelaysOwn s) $! _pRelaysOwn d
   setProxyStats (pMsgFwds s) $! _pMsgFwds d
@@ -420,8 +434,10 @@ instance StrEncoding ServerStatsData where
         "msgSentNtf=" <> strEncode (_msgSentNtf d),
         "msgRecvNtf=" <> strEncode (_msgRecvNtf d),
         "msgNtfs=" <> strEncode (_msgNtfs d),
+        "msgNtfsB=" <> strEncode (_msgNtfsB d),
         "msgNtfNoSub=" <> strEncode (_msgNtfNoSub d),
         "msgNtfLost=" <> strEncode (_msgNtfLost d),
+        "msgNtfExpired=" <> strEncode (_msgNtfExpired d),
         "activeQueues:",
         strEncode (_activeQueues d),
         "activeQueuesNtf:",
@@ -475,8 +491,10 @@ instance StrEncoding ServerStatsData where
     _msgSentNtf <- opt "msgSentNtf="
     _msgRecvNtf <- opt "msgRecvNtf="
     _msgNtfs <- opt "msgNtfs="
+    _msgNtfsB <- opt "msgNtfsB="
     _msgNtfNoSub <- opt "msgNtfNoSub="
     _msgNtfLost <- opt "msgNtfLost="
+    _msgNtfExpired <- opt "msgNtfExpired="
     _activeQueues <-
       optional ("activeQueues:" <* A.endOfLine) >>= \case
         Just _ -> strP <* optional A.endOfLine
@@ -536,8 +554,10 @@ instance StrEncoding ServerStatsData where
           _msgSentNtf,
           _msgRecvNtf,
           _msgNtfs,
+          _msgNtfsB,
           _msgNtfNoSub,
           _msgNtfLost,
+          _msgNtfExpired,
           _activeQueues,
           _activeQueuesNtf,
           _pRelays,

src/Simplex/Messaging/TMap.hs

@@ -9,6 +9,7 @@ module Simplex.Messaging.TMap
     member,
     memberIO,
     insert,
+    insertM,
     delete,
     lookupInsert,
     lookupDelete,
@@ -62,6 +63,10 @@ insert :: Ord k => k -> a -> TMap k a -> STM ()
 insert k v m = modifyTVar' m $ M.insert k v
 {-# INLINE insert #-}
 
+insertM :: Ord k => k -> STM a -> TMap k a -> STM ()
+insertM k f m = modifyTVar' m . M.insert k =<< f
+{-# INLINE insertM #-}
+
 delete :: Ord k => k -> TMap k a -> STM ()
 delete k m = modifyTVar' m $ M.delete k
 {-# INLINE delete #-}