From 0b488ceb66750e234529c0d626c66b3361597dfa Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Sun, 17 Sep 2023 15:16:24 +0100 Subject: [PATCH 1/2] ntf server: faster restoring from store log (#846) * ntf server: faster restoring from store log * only log sub status when changed --- src/Simplex/Messaging/Notifications/Server.hs | 32 ++++++++++--------- .../Messaging/Notifications/Server/Env.hs | 4 +++ .../Notifications/Server/StoreLog.hs | 18 ++++++----- 3 files changed, 31 insertions(+), 23 deletions(-) diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index 1daeea8ab..bfef05f30 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -275,12 +275,11 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge updateSubStatus smpQueue status = do st <- asks store - atomically (findNtfSubscription st smpQueue) - >>= mapM_ - ( \NtfSubData {ntfSubId, subStatus} -> do - atomically $ writeTVar subStatus status - withNtfLog $ \sl -> logSubscriptionStatus sl ntfSubId status - ) + atomically (findNtfSubscription st smpQueue) >>= mapM_ update + where + update NtfSubData {ntfSubId, subStatus} = do + old <- atomically $ stateTVar subStatus (,status) + when (old /= status) $ withNtfLog $ \sl -> logSubscriptionStatus sl ntfSubId status ntfPush :: NtfPushServer -> M () ntfPush s@NtfPushServer {pushQ} = forever $ do @@ -292,7 +291,10 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do | status /= NTInvalid && status /= NTExpired -> deliverNotification pp tkn ntf >>= \case Right _ -> do - status_ <- atomically $ stateTVar tknStatus $ \status' -> if status' == NTActive then (Nothing, NTActive) else (Just NTConfirmed, NTConfirmed) + status_ <- atomically $ stateTVar tknStatus $ \case + NTActive -> (Nothing, NTActive) + NTConfirmed -> (Nothing, NTConfirmed) + _ -> (Just NTConfirmed, NTConfirmed) forM_ status_ $ \status' -> withNtfLog $ \sl -> logTokenStatus sl ntfTknId status' _ -> pure () | otherwise -> logError "bad notification token status" @@ -309,7 +311,7 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do | status == NTActive = action | otherwise = liftIO $ logError "bad notification token status" deliverNotification :: PushProvider -> NtfTknData -> PushNotification -> M (Either PushProviderError ()) - deliverNotification pp tkn@NtfTknData {ntfTknId, tknStatus} ntf = do + deliverNotification pp tkn ntf = do deliver <- liftIO $ getPushClient s pp liftIO (runExceptT $ deliver tkn ntf) >>= \case Right _ -> pure $ Right () @@ -318,19 +320,20 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do PPRetryLater -> retryDeliver PPCryptoError _ -> err e PPResponseError _ _ -> err e - PPTokenInvalid -> updateTknStatus NTInvalid >> err e + PPTokenInvalid -> updateTknStatus tkn NTInvalid >> err e PPPermanentError -> err e where retryDeliver :: M (Either PushProviderError ()) retryDeliver = do deliver <- liftIO $ newPushClient s pp liftIO (runExceptT $ deliver tkn ntf) >>= either err (pure . Right) - updateTknStatus :: NtfTknStatus -> M () - updateTknStatus status = do - atomically $ writeTVar tknStatus status - withNtfLog $ \sl -> logTokenStatus sl ntfTknId status err e = logError (T.pack $ "Push provider error (" <> show pp <> "): " <> show e) $> Left e +updateTknStatus :: NtfTknData -> NtfTknStatus -> M () +updateTknStatus NtfTknData {ntfTknId, tknStatus} status = do + old <- atomically $ stateTVar tknStatus (,status) + when (old /= status) $ withNtfLog $ \sl -> logTokenStatus sl ntfTknId status + runNtfClientTransport :: Transport c => THandle c -> M () runNtfClientTransport th@THandle {sessionId} = do qSize <- asks $ clientQSize . config @@ -462,10 +465,9 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu | (status == NTRegistered || status == NTConfirmed || status == NTActive) && tknRegCode == code -> do logDebug "TVFY - token verified" st <- asks store - atomically $ writeTVar tknStatus NTActive + updateTknStatus tkn NTActive tIds <- atomically $ removeInactiveTokenRegistrations st tkn forM_ tIds cancelInvervalNotifications - withNtfLog $ \s -> logTokenStatus s tknId NTActive incNtfStat tknVerified pure NROk | otherwise -> do diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 38604b62b..d6989c4d4 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -3,11 +3,13 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE KindSignatures #-} {-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} module Simplex.Messaging.Notifications.Server.Env where import Control.Concurrent (ThreadId) import Control.Concurrent.Async (Async) +import Control.Logger.Simple import Control.Monad.IO.Unlift import Crypto.Random import Data.ByteString.Char8 (ByteString) @@ -84,7 +86,9 @@ newNtfServerEnv :: (MonadUnliftIO m, MonadRandom m) => NtfServerConfig -> m NtfE newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsConfig, storeLogFile, caCertificateFile, certificateFile, privateKeyFile} = do idsDrg <- newTVarIO =<< drgNew store <- atomically newNtfStore + logInfo "restoring subscriptions..." storeLog <- liftIO $ mapM (`readWriteNtfStore` store) storeLogFile + logInfo "restored subscriptions" subscriber <- atomically $ newNtfSubscriber subQSize smpAgentCfg pushServer <- atomically $ newNtfPushServer pushQSize apnsConfig tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile diff --git a/src/Simplex/Messaging/Notifications/Server/StoreLog.hs b/src/Simplex/Messaging/Notifications/Server/StoreLog.hs index 7ad0b2fe6..fb7b78e21 100644 --- a/src/Simplex/Messaging/Notifications/Server/StoreLog.hs +++ b/src/Simplex/Messaging/Notifications/Server/StoreLog.hs @@ -4,6 +4,7 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE StrictData #-} module Simplex.Messaging.Notifications.Server.StoreLog ( StoreLog, @@ -22,7 +23,7 @@ module Simplex.Messaging.Notifications.Server.StoreLog where import Control.Concurrent.STM -import Control.Monad (void) +import Control.Monad import qualified Data.Attoparsec.ByteString.Char8 as A import qualified Data.ByteString.Char8 as B import Data.Word (Word16) @@ -151,6 +152,7 @@ instance StrEncoding NtfSubRec where logNtfStoreRecord :: StoreLog 'WriteMode -> NtfStoreLogRecord -> IO () logNtfStoreRecord = writeStoreLogRecord +{-# INLINE logNtfStoreRecord #-} logCreateToken :: StoreLog 'WriteMode -> NtfTknData -> IO () logCreateToken s tkn = logNtfStoreRecord s . CreateToken =<< atomically (mkTknRec tkn) @@ -194,9 +196,11 @@ readNtfStore f st = mapM_ addNtfLogRecord . B.lines =<< B.readFile f CreateToken r@NtfTknRec {ntfTknId} -> do tkn <- mkTknData r addNtfToken st ntfTknId tkn - TokenStatus tknId status -> - getNtfToken st tknId - >>= mapM_ (\NtfTknData {tknStatus} -> writeTVar tknStatus status) + TokenStatus tknId status -> do + tkn_ <- getNtfToken st tknId + forM_ tkn_ $ \tkn@NtfTknData {tknStatus} -> do + writeTVar tknStatus status + when (status == NTActive) $ void $ removeInactiveTokenRegistrations st tkn UpdateToken tknId token' tknRegCode -> getNtfToken st tknId >>= mapM_ @@ -221,7 +225,5 @@ readNtfStore f st = mapM_ addNtfLogRecord . B.lines =<< B.readFile f writeNtfStore :: StoreLog 'WriteMode -> NtfStore -> IO () writeNtfStore s NtfStore {tokens, subscriptions} = do - atomically (readTVar tokens >>= mapM mkTknRec) - >>= mapM_ (writeStoreLogRecord s . CreateToken) - atomically (readTVar subscriptions >>= mapM mkSubRec) - >>= mapM_ (writeStoreLogRecord s . CreateSubscription) + mapM_ (logCreateToken s) =<< readTVarIO tokens + mapM_ (logCreateSubscription s) =<< readTVarIO subscriptions From 00b48babf3a3d7aad9d73eed49b3f251556a3fa0 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Sun, 17 Sep 2023 22:23:00 +0100 Subject: [PATCH 2/2] 5.4.0.3: ntf 1.6.2 --- package.yaml | 2 +- simplexmq.cabal | 2 +- src/Simplex/Messaging/Notifications/Server/Main.hs | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/package.yaml b/package.yaml index 3ed5b82fa..c5c37b568 100644 --- a/package.yaml +++ b/package.yaml @@ -1,5 +1,5 @@ name: simplexmq -version: 5.4.0.2 +version: 5.4.0.3 synopsis: SimpleXMQ message broker description: | This package includes <./docs/Simplex-Messaging-Server.html server>, diff --git a/simplexmq.cabal b/simplexmq.cabal index dc9ab7e2e..aca882438 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -5,7 +5,7 @@ cabal-version: 1.12 -- see: https://github.com/sol/hpack name: simplexmq -version: 5.4.0.2 +version: 5.4.0.3 synopsis: SimpleXMQ message broker description: This package includes <./docs/Simplex-Messaging-Server.html server>, <./docs/Simplex-Messaging-Client.html client> and diff --git a/src/Simplex/Messaging/Notifications/Server/Main.hs b/src/Simplex/Messaging/Notifications/Server/Main.hs index c141966d8..216890be2 100644 --- a/src/Simplex/Messaging/Notifications/Server/Main.hs +++ b/src/Simplex/Messaging/Notifications/Server/Main.hs @@ -30,7 +30,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout) import Text.Read (readMaybe) ntfServerVersion :: String -ntfServerVersion = "1.6.1" +ntfServerVersion = "1.6.2" defaultSMPBatchDelay :: Int defaultSMPBatchDelay = 10000