Revert "smp server: persist notifications to avoid losing them when ntf server is offline (#1336)"

This reverts commit 0739f7b.

Author: epoberezkin

simplexmq.cabal

@@ -174,7 +174,6 @@ library
       Simplex.Messaging.Server.Main
       Simplex.Messaging.Server.MsgStore
       Simplex.Messaging.Server.MsgStore.STM
-      Simplex.Messaging.Server.NtfStore
       Simplex.Messaging.Server.QueueStore
       Simplex.Messaging.Server.QueueStore.QueueInfo
       Simplex.Messaging.Server.QueueStore.STM

src/Simplex/Messaging/Notifications/Server.hs

@@ -220,8 +220,9 @@ 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 -> atomically (writeTBQueue pushQ (tkn, PNMessage (PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} :| []))))
+              atomically $
+                findNtfSubscriptionToken st smpQueue
+                  >>= mapM_ (\tkn -> 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

@@ -38,7 +38,6 @@ 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
@@ -62,7 +61,6 @@ 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
@@ -72,8 +70,6 @@ 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,
@@ -86,8 +82,6 @@ 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,
@@ -116,16 +110,6 @@ 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
@@ -143,7 +127,6 @@ data Env = Env
     serverIdentity :: KeyHash,
     queueStore :: QueueStore,
     msgStore :: STMMsgStore,
-    ntfStore :: NtfStore,
     random :: TVar ChaChaDRG,
     storeLog :: Maybe (StoreLog 'WriteMode),
     tlsServerCreds :: T.Credential,
@@ -246,7 +229,6 @@ 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
@@ -262,7 +244,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, ntfStore, random, storeLog, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
+  pure Env {config, serverInfo, server, serverIdentity, queueStore, msgStore, random, storeLog, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
   where
     getCredentials protocol creds = do
       files <- missingCreds

src/Simplex/Messaging/Server/Env/STM.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
+import Simplex.Messaging.Server.Env.STM (ServerConfig (..), defMsgExpirationDays, defaultInactiveClientExpiration, defaultMessageExpiration, defaultProxyClientConcurrency)
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
 import Simplex.Messaging.Transport (batchCmdsSMPVersion, sendingProxySMPVersion, simplexMQVersion, supportedServerSMPRelayVRange)
@@ -154,8 +154,7 @@ 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")
-                <> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
+                <> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n\n")
                 <> "# Log daily server statistics to CSV file\n"
                 <> ("log_stats: " <> onOff logStats <> "\n\n")
                 <> "[AUTH]\n\
@@ -269,11 +268,6 @@ 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 =
@@ -292,8 +286,13 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                   },
               httpCredentials = (\WebHttpsParams {key, cert} -> ServerCredentials {caCertificateFile = Nothing, privateKeyFile = key, certificateFile = cert}) <$> webHttpsParams',
               storeLogFile = enableStoreLog $> storeLogFilePath,
-              storeMsgsFile = restoreMessagesFile $ combine logPath "smp-server-messages.log",
-              storeNtfsFile = restoreMessagesFile $ combine logPath "smp-server-ntfs.log",
+              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,
               -- allow creating new queues by default
               allowNewQueues = fromMaybe True $ iniOnOff "AUTH" "new_queues" ini,
               newQueueBasicAuth = either error id <$!> strDecodeIni "AUTH" "create_password" ini,
@@ -304,10 +303,6 @@ 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
@@ -319,7 +314,6 @@ 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/Main.hs

@@ -67,10 +67,8 @@ 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,
@@ -119,10 +117,8 @@ data ServerStatsData = ServerStatsData
     _msgRecvNtf :: Int,
     _activeQueuesNtf :: PeriodStatsData,
     _msgNtfs :: Int,
-    _msgNtfsB :: Int,
     _msgNtfNoSub :: Int,
     _msgNtfLost :: Int,
-    _msgNtfExpired :: Int,
     _pRelays :: ProxyStatsData,
     _pRelaysOwn :: ProxyStatsData,
     _pMsgFwds :: ProxyStatsData,
@@ -173,10 +169,8 @@ 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
@@ -224,10 +218,8 @@ newServerStats ts = do
         msgRecvNtf,
         activeQueuesNtf,
         msgNtfs,
-        msgNtfsB,
         msgNtfNoSub,
         msgNtfLost,
-        msgNtfExpired,
         pRelays,
         pRelaysOwn,
         pMsgFwds,
@@ -277,10 +269,8 @@ 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
@@ -328,10 +318,8 @@ getServerStatsData s = do
         _msgRecvNtf,
         _activeQueuesNtf,
         _msgNtfs,
-        _msgNtfsB,
         _msgNtfNoSub,
         _msgNtfLost,
-        _msgNtfExpired,
         _pRelays,
         _pRelaysOwn,
         _pMsgFwds,
@@ -382,10 +370,8 @@ 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
@@ -434,10 +420,8 @@ 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:",
@@ -491,10 +475,8 @@ 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
@@ -554,10 +536,8 @@ instance StrEncoding ServerStatsData where
           _msgSentNtf,
           _msgRecvNtf,
           _msgNtfs,
-          _msgNtfsB,
           _msgNtfNoSub,
           _msgNtfLost,
-          _msgNtfExpired,
           _activeQueues,
           _activeQueuesNtf,
           _pRelays,

src/Simplex/Messaging/Server/NtfStore.hs

@@ -9,7 +9,6 @@ module Simplex.Messaging.TMap
     member,
     memberIO,
     insert,
-    insertM,
     delete,
     lookupInsert,
     lookupDelete,
@@ -63,10 +62,6 @@ 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 #-}