From 0a3d014f5d861e0c61314e328de52f704e102f81 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Wed, 30 Apr 2025 11:40:32 +0100 Subject: [PATCH] ntf server: priority db pool for notifications and client commands, concurrent resubscriptions, CLI options (#1529) * ntf server: option to skip specific tokens when importing store log * logs * batch later * subscribe in parallel, bigger queues * ntf server: test schema and migrations * optimize subscriptions * log level * resubscribe before starting * better subscribing * pooled concurrent resubscriptions * 5 * async resubscribe * typo * priority pool * db batch size --- apps/ntf-server/Main.hs | 1 - simplexmq.cabal | 2 +- src/Simplex/FileTransfer/Client/Main.hs | 14 +- src/Simplex/FileTransfer/Server.hs | 18 +- src/Simplex/FileTransfer/Server/Env.hs | 4 +- src/Simplex/Messaging/Agent/Store/Postgres.hs | 29 +-- .../Messaging/Agent/Store/Postgres/Common.hs | 38 +++- src/Simplex/Messaging/Client/Agent.hs | 4 +- src/Simplex/Messaging/Notifications/Server.hs | 72 +++---- .../Messaging/Notifications/Server/Env.hs | 2 +- .../Messaging/Notifications/Server/Main.hs | 31 ++- .../Notifications/Server/Push/APNS.hs | 13 +- .../Notifications/Server/Store/Postgres.hs | 89 ++++++--- .../Server/Store/ntf_server_schema.sql | 178 ++++++++++++++++++ src/Simplex/Messaging/Server.hs | 44 ++--- src/Simplex/Messaging/Server/CLI.hs | 21 ++- src/Simplex/Messaging/Server/Env/STM.hs | 5 +- src/Simplex/Messaging/Server/Main.hs | 8 +- src/Simplex/Messaging/Server/StoreLog.hs | 6 +- src/Simplex/Messaging/Transport/Server.hs | 2 +- src/Simplex/Messaging/Util.hs | 1 + tests/NtfClient.hs | 5 +- tests/PostgresSchemaDump.hs | 70 +++++++ tests/SMPClient.hs | 6 +- tests/ServerTests/SchemaDump.hs | 98 ---------- tests/Test.hs | 25 ++- 26 files changed, 532 insertions(+), 254 deletions(-) create mode 100644 src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql create mode 100644 tests/PostgresSchemaDump.hs delete mode 100644 tests/ServerTests/SchemaDump.hs diff --git a/apps/ntf-server/Main.hs b/apps/ntf-server/Main.hs index ac93580b9..82da000a8 100644 --- a/apps/ntf-server/Main.hs +++ b/apps/ntf-server/Main.hs @@ -15,7 +15,6 @@ logCfg = LogConfig {lc_file = Nothing, lc_stderr = True} main :: IO () main = do - setLogLevel LogInfo cfgPath <- getEnvPath "NTF_SERVER_CFG_PATH" defaultCfgPath logPath <- getEnvPath "NTF_SERVER_LOG_PATH" defaultLogPath withGlobalLogging logCfg $ ntfServerCLI cfgPath logPath diff --git a/simplexmq.cabal b/simplexmq.cabal index e3e1e61e7..3c8c0ae24 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -494,7 +494,7 @@ test-suite simplexmq-test AgentTests.NotificationTests NtfClient NtfServerTests - ServerTests.SchemaDump + PostgresSchemaDump hs-source-dirs: tests apps/smp-server/web diff --git a/src/Simplex/FileTransfer/Client/Main.hs b/src/Simplex/FileTransfer/Client/Main.hs index a330e21a7..c73cac637 100644 --- a/src/Simplex/FileTransfer/Client/Main.hs +++ b/src/Simplex/FileTransfer/Client/Main.hs @@ -280,7 +280,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re let chunkSpecs = prepareChunkSpecs encPath chunkSizes fdRcv = FileDescription {party = SFRecipient, size = FileSize encSize, digest = FileDigest digest, key, nonce, chunkSize = FileSize defChunkSize, chunks = [], redirect = Nothing} fdSnd = FileDescription {party = SFSender, size = FileSize encSize, digest = FileDigest digest, key, nonce, chunkSize = FileSize defChunkSize, chunks = [], redirect = Nothing} - logInfo $ "encrypted file to " <> tshow encPath + logDebug $ "encrypted file to " <> tshow encPath pure (encPath, fdRcv, fdSnd, chunkSpecs, encSize) uploadFile :: TVar ChaChaDRG -> [XFTPChunkSpec] -> TVar [Int64] -> Int64 -> ExceptT CLIError IO [SentFileChunk] uploadFile g chunks uploadedChunks encSize = do @@ -293,14 +293,14 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re -- TODO shuffle/unshuffle chunks -- the reason we don't do pooled downloads here within one server is that http2 library doesn't handle cleint concurrency, even though -- upload doesn't allow other requests within the same client until complete (but download does allow). - logInfo $ "uploading " <> tshow (length chunks) <> " chunks..." + logDebug $ "uploading " <> tshow (length chunks) <> " chunks..." (errs, rs) <- partitionEithers . concat <$> liftIO (pooledForConcurrentlyN 16 chunks' . mapM $ runExceptT . uploadFileChunk a) mapM_ throwE errs pure $ map snd (sortOn fst rs) where uploadFileChunk :: XFTPClientAgent -> (Int, XFTPChunkSpec, XFTPServerWithAuth) -> ExceptT CLIError IO (Int, SentFileChunk) uploadFileChunk a (chunkNo, chunkSpec@XFTPChunkSpec {chunkSize}, ProtoServerWithAuth xftpServer auth) = do - logInfo $ "uploading chunk " <> tshow chunkNo <> " to " <> showServer xftpServer <> "..." + logDebug $ "uploading chunk " <> tshow chunkNo <> " to " <> showServer xftpServer <> "..." (sndKey, spKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g rKeys <- atomically $ L.fromList <$> replicateM numRecipients (C.generateAuthKeyPair C.SEd25519 g) digest <- liftIO $ getChunkDigest chunkSpec @@ -308,7 +308,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re c <- withRetry retryCount $ getXFTPServerClient a xftpServer (sndId, rIds) <- withRetry retryCount $ createXFTPChunk c spKey ch (L.map fst rKeys) auth withReconnect a xftpServer retryCount $ \c' -> uploadXFTPChunk c' spKey sndId chunkSpec - logInfo $ "uploaded chunk " <> tshow chunkNo + logDebug $ "uploaded chunk " <> tshow chunkNo uploaded <- atomically . stateTVar uploadedChunks $ \cs -> let cs' = fromIntegral chunkSize : cs in (sum cs', cs') liftIO $ do @@ -418,11 +418,11 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath, downloadFileChunk :: TVar ChaChaDRG -> XFTPClientAgent -> FilePath -> FileSize Int64 -> TVar [Int64] -> FileChunk -> ExceptT CLIError IO (Int, FilePath) downloadFileChunk g a encPath (FileSize encSize) downloadedChunks FileChunk {chunkNo, chunkSize, digest, replicas = replica : _} = do let FileChunkReplica {server, replicaId, replicaKey} = replica - logInfo $ "downloading chunk " <> tshow chunkNo <> " from " <> showServer server <> "..." + logDebug $ "downloading chunk " <> tshow chunkNo <> " from " <> showServer server <> "..." chunkPath <- uniqueCombine encPath $ show chunkNo let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest) withReconnect a server retryCount $ \c -> downloadXFTPChunk g c replicaKey (unChunkReplicaId replicaId) chunkSpec - logInfo $ "downloaded chunk " <> tshow chunkNo <> " to " <> T.pack chunkPath + logDebug $ "downloaded chunk " <> tshow chunkNo <> " to " <> T.pack chunkPath downloaded <- atomically . stateTVar downloadedChunks $ \cs -> let cs' = fromIntegral (unFileSize chunkSize) : cs in (sum cs', cs') liftIO $ do @@ -467,7 +467,7 @@ cliDeleteFile DeleteOptions {fileDescription, retryCount, yes} = do deleteFileChunk a FileChunk {chunkNo, replicas = replica : _} = do let FileChunkReplica {server, replicaId, replicaKey} = replica withReconnect a server retryCount $ \c -> deleteXFTPChunk c replicaKey (unChunkReplicaId replicaId) - logInfo $ "deleted chunk " <> tshow chunkNo <> " from " <> showServer server + logDebug $ "deleted chunk " <> tshow chunkNo <> " from " <> showServer server deleteFileChunk _ _ = throwE $ CLIError "chunk has no replicas" cliFileDescrInfo :: InfoOptions -> ExceptT CLIError IO () diff --git a/src/Simplex/FileTransfer/Server.hs b/src/Simplex/FileTransfer/Server.hs index 9bdb9b28e..945c74ac8 100644 --- a/src/Simplex/FileTransfer/Server.hs +++ b/src/Simplex/FileTransfer/Server.hs @@ -181,7 +181,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira stopServer = do withFileLog closeStoreLog saveServerStats - logInfo "Server stopped" + logNote "Server stopped" expireFilesThread_ :: XFTPServerConfig -> [M ()] expireFilesThread_ XFTPServerConfig {fileExpiration = Just fileExp} = [expireFiles fileExp] @@ -560,13 +560,13 @@ expireServerFiles itemDelay expCfg = do usedStart <- readTVarIO $ usedStorage st old <- liftIO $ expireBeforeEpoch expCfg files' <- readTVarIO (files st) - logInfo $ "Expiration check: " <> tshow (M.size files') <> " files" + logNote $ "Expiration check: " <> tshow (M.size files') <> " files" forM_ (M.keys files') $ \sId -> do mapM_ threadDelay itemDelay atomically (expiredFilePath st sId old) >>= mapM_ (maybeRemove $ delete st sId) usedEnd <- readTVarIO $ usedStorage st - logInfo $ "Used " <> mbs usedStart <> " -> " <> mbs usedEnd <> ", " <> mbs (usedStart - usedEnd) <> " reclaimed." + logNote $ "Used " <> mbs usedStart <> " -> " <> mbs usedEnd <> ", " <> mbs (usedStart - usedEnd) <> " reclaimed." where mbs bs = tshow (bs `div` 1048576) <> "mb" maybeRemove del = maybe del (remove del) @@ -600,15 +600,15 @@ saveServerStats = >>= mapM_ (\f -> asks serverStats >>= liftIO . getFileServerStatsData >>= liftIO . saveStats f) where saveStats f stats = do - logInfo $ "saving server stats to file " <> T.pack f + logNote $ "saving server stats to file " <> T.pack f B.writeFile f $ strEncode stats - logInfo "server stats saved" + logNote "server stats saved" restoreServerStats :: M () restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats where restoreStats f = whenM (doesFileExist f) $ do - logInfo $ "restoring server stats from file " <> T.pack f + logNote $ "restoring server stats from file " <> T.pack f liftIO (strDecode <$> B.readFile f) >>= \case Right d@FileServerStatsData {_filesCount = statsFilesCount, _filesSize = statsFilesSize} -> do s <- asks serverStats @@ -617,10 +617,10 @@ restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStat _filesSize <- readTVarIO usedStorage liftIO $ setFileServerStats s d {_filesCount, _filesSize} renameFile f $ f <> ".bak" - logInfo "server stats restored" + logNote "server stats restored" when (statsFilesCount /= _filesCount) $ logWarn $ "Files count differs: stats: " <> tshow statsFilesCount <> ", store: " <> tshow _filesCount when (statsFilesSize /= _filesSize) $ logWarn $ "Files size differs: stats: " <> tshow statsFilesSize <> ", store: " <> tshow _filesSize - logInfo $ "Restored " <> tshow (_filesSize `div` 1048576) <> " MBs in " <> tshow _filesCount <> " files" + logNote $ "Restored " <> tshow (_filesSize `div` 1048576) <> " MBs in " <> tshow _filesCount <> " files" Left e -> do - logInfo $ "error restoring server stats: " <> T.pack e + logNote $ "error restoring server stats: " <> T.pack e liftIO exitFailure diff --git a/src/Simplex/FileTransfer/Server/Env.hs b/src/Simplex/FileTransfer/Server/Env.hs index 4bccda0ff..21e6d06dd 100644 --- a/src/Simplex/FileTransfer/Server/Env.hs +++ b/src/Simplex/FileTransfer/Server/Env.hs @@ -103,8 +103,8 @@ newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, xftpCrede used <- countUsedStorage <$> readTVarIO (files store) atomically $ writeTVar (usedStorage store) used forM_ fileSizeQuota $ \quota -> do - logInfo $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used) - when (quota < used) $ logInfo "WARNING: storage quota is less than used storage, no files can be uploaded!" + logNote $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used) + when (quota < used) $ logWarn "WARNING: storage quota is less than used storage, no files can be uploaded!" tlsServerCreds <- loadServerCredential xftpCredentials Fingerprint fp <- loadFingerprint xftpCredentials serverStats <- newFileServerStats =<< getCurrentTime diff --git a/src/Simplex/Messaging/Agent/Store/Postgres.hs b/src/Simplex/Messaging/Agent/Store/Postgres.hs index 50494e781..075e4be48 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres.hs @@ -33,7 +33,6 @@ import qualified Simplex.Messaging.Agent.Store.Postgres.DB as DB import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..)) import Simplex.Messaging.Util (ifM, safeDecodeUtf8) import System.Exit (exitFailure) -import UnliftIO.MVar -- | Create a new Postgres DBStore with the given connection string, schema name and migrations. -- If passed schema does not exist in connectInfo database, it will be created. @@ -54,23 +53,26 @@ createDBStore opts migrations confirmMigrations = do connectPostgresStore :: DBOpts -> IO DBStore connectPostgresStore DBOpts {connstr, schema, poolSize, createSchema} = do - dbSem <- newMVar () - dbPool <- newTBQueueIO poolSize + dbPriorityPool <- newDBStorePool poolSize + dbPool <- newDBStorePool poolSize dbClosed <- newTVarIO True - let st = DBStore {dbConnstr = connstr, dbSchema = schema, dbPoolSize = fromIntegral poolSize, dbPool, dbSem, dbNew = False, dbClosed} - dbNew <- connectPool st createSchema + let st = DBStore {dbConnstr = connstr, dbSchema = schema, dbPoolSize = fromIntegral poolSize, dbPriorityPool, dbPool, dbNew = False, dbClosed} + dbNew <- connectStore st createSchema pure st {dbNew} -- uninterruptibleMask_ here and below is used here so that it is not interrupted half-way, -- it relies on the assumption that when dbClosed = True, the queue is empty, -- and when it is False, the queue is full (or will have connections returned to it by the threads that use them). -connectPool :: DBStore -> Bool -> IO Bool -connectPool DBStore {dbConnstr, dbSchema, dbPoolSize, dbPool, dbClosed} createSchema = uninterruptibleMask_ $ do +connectStore :: DBStore -> Bool -> IO Bool +connectStore DBStore {dbConnstr, dbSchema, dbPoolSize, dbPriorityPool, dbPool, dbClosed} createSchema = uninterruptibleMask_ $ do (conn, dbNew) <- connectDB dbConnstr dbSchema createSchema -- TODO [postgres] analogue for dbBusyLoop? - conns <- replicateM (dbPoolSize - 1) $ fst <$> connectDB dbConnstr dbSchema False - mapM_ (atomically . writeTBQueue dbPool) (conn : conns) + writeConns dbPriorityPool . (conn :) =<< mkConns (dbPoolSize - 1) + writeConns dbPool =<< mkConns dbPoolSize atomically $ writeTVar dbClosed False pure dbNew + where + writeConns pool conns = mapM_ (atomically . writeTBQueue (dbPoolConns pool)) conns + mkConns n = replicateM n $ fst <$> connectDB dbConnstr dbSchema False connectDB :: ByteString -> ByteString -> Bool -> IO (DB.Connection, Bool) connectDB connstr schema createSchema = do @@ -111,16 +113,19 @@ doesSchemaExist db schema = do pure schemaExists closeDBStore :: DBStore -> IO () -closeDBStore DBStore {dbPool, dbPoolSize, dbClosed} = +closeDBStore DBStore {dbPoolSize, dbPriorityPool, dbPool, dbClosed} = ifM (readTVarIO dbClosed) (putStrLn "closeDBStore: already closed") $ uninterruptibleMask_ $ do - replicateM_ dbPoolSize $ atomically (readTBQueue dbPool) >>= DB.close + closePool dbPriorityPool + closePool dbPool atomically $ writeTVar dbClosed True + where + closePool pool = replicateM_ dbPoolSize $ atomically (readTBQueue $ dbPoolConns pool) >>= DB.close reopenDBStore :: DBStore -> IO () reopenDBStore st = ifM (readTVarIO $ dbClosed st) - (void $ connectPool st False) + (void $ connectStore st False) (putStrLn "reopenDBStore: already opened") -- not used with postgres client (used for ExecAgentStoreSQL, ExecChatStoreSQL) diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Common.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Common.hs index a71376a20..3ca0a755e 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Common.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Common.hs @@ -6,7 +6,9 @@ module Simplex.Messaging.Agent.Store.Postgres.Common ( DBStore (..), + DBStorePool (..), DBOpts (..), + newDBStorePool, withConnection, withConnection', withTransaction, @@ -20,6 +22,7 @@ import Control.Concurrent.STM import Control.Exception (bracket) import Data.ByteString (ByteString) import qualified Database.PostgreSQL.Simple as PSQL +import Numeric.Natural (Natural) import Simplex.Messaging.Agent.Store.Postgres.Options -- TODO [postgres] use log_min_duration_statement instead of custom slow queries (SQLite's Connection type) @@ -27,19 +30,40 @@ data DBStore = DBStore { dbConnstr :: ByteString, dbSchema :: ByteString, dbPoolSize :: Int, - dbPool :: TBQueue PSQL.Connection, - -- MVar is needed for fair pool distribution, without STM retry contention. - -- Only one thread can be blocked on STM read. - dbSem :: MVar (), + dbPriorityPool :: DBStorePool, + dbPool :: DBStorePool, + -- dbPoolSize :: Int, + -- dbPool :: TBQueue PSQL.Connection, + -- -- MVar is needed for fair pool distribution, without STM retry contention. + -- -- Only one thread can be blocked on STM read. + -- dbSem :: MVar (), dbClosed :: TVar Bool, dbNew :: Bool } +newDBStorePool :: Natural -> IO DBStorePool +newDBStorePool poolSize = do + dbSem <- newMVar () + dbPoolConns <- newTBQueueIO poolSize + pure DBStorePool {dbSem, dbPoolConns} + +data DBStorePool = DBStorePool + { dbPoolConns :: TBQueue PSQL.Connection, + -- MVar is needed for fair pool distribution, without STM retry contention. + -- Only one thread can be blocked on STM read. + dbSem :: MVar () + } + withConnectionPriority :: DBStore -> Bool -> (PSQL.Connection -> IO a) -> IO a -withConnectionPriority DBStore {dbPool, dbSem} _priority = +withConnectionPriority DBStore {dbPriorityPool, dbPool} priority = + withConnectionPool $ if priority then dbPriorityPool else dbPool +{-# INLINE withConnectionPriority #-} + +withConnectionPool :: DBStorePool -> (PSQL.Connection -> IO a) -> IO a +withConnectionPool DBStorePool {dbPoolConns, dbSem} = bracket - (withMVar dbSem $ \_ -> atomically $ readTBQueue dbPool) - (atomically . writeTBQueue dbPool) + (withMVar dbSem $ \_ -> atomically $ readTBQueue dbPoolConns) + (atomically . writeTBQueue dbPoolConns) withConnection :: DBStore -> (PSQL.Connection -> IO a) -> IO a withConnection st = withConnectionPriority st False diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index dab0a4040..1e3d71b66 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/src/Simplex/Messaging/Client/Agent.hs @@ -42,7 +42,7 @@ import Simplex.Messaging.Session import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport -import Simplex.Messaging.Util (catchAll_, ifM, toChunks, whenM, ($>>=), (<$$>)) +import Simplex.Messaging.Util (catchAll_, ifM, safeDecodeUtf8, toChunks, tshow, whenM, ($>>=), (<$$>)) import System.Timeout (timeout) import UnliftIO (async) import qualified UnliftIO.Exception as E @@ -321,7 +321,7 @@ withSMP ca srv action = (getSMPServerClient' ca srv >>= action) `catchE` logSMPE where logSMPError :: SMPClientError -> ExceptT SMPClientError IO a logSMPError e = do - liftIO $ putStrLn $ "SMP error (" <> show srv <> "): " <> show e + logInfo $ "SMP error (" <> safeDecodeUtf8 (strEncode $ host srv) <> "): " <> tshow e throwE e subscribeQueuesSMP :: SMPClientAgent -> SMPServer -> NonEmpty (RecipientId, RcvPrivateAuthKey) -> IO () diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index 5407a3bc9..677f14758 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -17,6 +17,7 @@ module Simplex.Messaging.Notifications.Server where import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (mapConcurrently) import Control.Logger.Simple import Control.Monad import Control.Monad.Except @@ -104,7 +105,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions} liftIO $ putStrLn "Server started in 'maintenance' mode, exiting" stopServer liftIO $ exitSuccess - resubscribe s + void $ forkIO $ resubscribe s raceAny_ ( ntfSubscriber s : ntfPush ps @@ -135,12 +136,12 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions} stopServer :: M () stopServer = do - logInfo "Saving server state..." + logNote "Saving server state..." saveServer NtfSubscriber {smpSubscribers, smpAgent} <- asks subscriber liftIO $ readTVarIO smpSubscribers >>= mapM_ (\SMPSubscriber {subThreadId} -> readTVarIO subThreadId >>= mapM_ (deRefWeak >=> mapM_ killThread)) liftIO $ closeSMPClientAgent smpAgent - logInfo "Server stopped" + logNote "Server stopped" saveServer :: M () saveServer = asks store >>= liftIO . closeNtfDbStore >> saveServerStats @@ -153,7 +154,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions} logServerStats :: Int64 -> Int64 -> FilePath -> M () logServerStats startAt logInterval statsFilePath = do initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime - logInfo $ "server stats log enabled: " <> T.pack statsFilePath + logNote $ "server stats log enabled: " <> T.pack statsFilePath liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0) NtfServerStats {fromTime, tknCreated, tknVerified, tknDeleted, tknReplaced, subCreated, subDeleted, ntfReceived, ntfDelivered, ntfFailed, ntfCronDelivered, ntfCronFailed, ntfVrfQueued, ntfVrfDelivered, ntfVrfFailed, ntfVrfInvalidTkn, activeTokens, activeSubs} <- asks serverStats @@ -267,12 +268,12 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions} countSubs acc@(metrics, !otherSrvs) (srv@(SMPServer (h :| _) _ _), srvSubs) = result . M.size <$> readTVarIO srvSubs where - result 0 = acc result cnt | isOwnServer a srv = let !ownSrvSubs' = M.alter (Just . maybe cnt (+ cnt)) host ownSrvSubs metrics' = metrics {ownSrvSubs = ownSrvSubs'} :: NtfSMPSubMetrics in (metrics', otherSrvs) + | cnt == 0 = acc | otherwise = let metrics' = metrics {otherSrvSubCount = otherSrvSubCount + cnt} :: NtfSMPSubMetrics in (metrics', S.insert host otherSrvs) @@ -410,25 +411,33 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg, startOptions} hPutStrLn h "AUTH" resubscribe :: NtfSubscriber -> M () -resubscribe NtfSubscriber {newSubQ} = do - logInfo "Preparing SMP resubscriptions..." +resubscribe NtfSubscriber {smpAgent = ca} = do st <- asks store batchSize <- asks $ subsBatchSize . config liftIO $ do srvs <- getUsedSMPServers st - count <- foldM (subscribeSrvSubs st batchSize) (0 :: Int) srvs - logInfo $ "SMP resubscriptions queued (" <> tshow count <> " subscriptions)" + logNote $ "Starting SMP resubscriptions for " <> tshow (length srvs) <> " servers..." + counts <- mapConcurrently (subscribeSrvSubs st batchSize) srvs + logNote $ "Completed all SMP resubscriptions for " <> tshow (length srvs) <> " servers (" <> tshow (sum counts) <> " subscriptions)" where - subscribeSrvSubs st batchSize !count srv = do - (n, subs_) <- - foldNtfSubscriptions st srv batchSize (0, []) $ \(!i, subs) sub -> - if length subs == batchSize - then write (L.fromList subs) $> (i + 1, []) - else pure (i + 1, sub : subs) - mapM_ write $ L.nonEmpty subs_ - pure $ count + n + subscribeSrvSubs st batchSize srv = do + let srvStr = safeDecodeUtf8 (strEncode $ L.head $ host srv) + logNote $ "Starting SMP resubscriptions for " <> srvStr + n <- loop 0 Nothing + logNote $ "Completed SMP resubscriptions for " <> srvStr <> " (" <> tshow n <> " subscriptions)" + pure n where - write subs = atomically $ writeTBQueue newSubQ (srv, subs) + dbBatchSize = batchSize * 100 + loop n afterSubId_ = + getServerNtfSubscriptions st srv afterSubId_ dbBatchSize >>= \case + Left _ -> exitFailure + Right [] -> pure n + Right subs -> do + mapM_ (subscribeQueuesNtfs ca srv . L.map snd) $ toChunks batchSize subs + let len = length subs + n' = n + len + afterSubId_' = Just $ fst $ last subs + if len < dbBatchSize then pure n' else loop n' afterSubId_' ntfSubscriber :: NtfSubscriber -> M () ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAgent {msgQ, agentQ}} = do @@ -437,14 +446,10 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge subscribe :: M () subscribe = forever $ do (srv, subs) <- atomically $ readTBQueue newSubQ - -- TODO [ntfdb] as we now group by server before putting subs to queue, - -- maybe this "subscribe" thread can be removed completely, - -- and the caller would directly write to SMPSubscriber queues SMPSubscriber {subscriberSubQ} <- getSMPSubscriber srv atomically $ writeTQueue subscriberSubQ subs - -- TODO [ntfdb] this does not guarantee that only one subscriber per server is created - -- there should be TMVar in the map + -- TODO [ntfdb] this does not guarantee that only one subscriber per server is created (there should be TMVar in the map) -- This does not need changing if single newSubQ remains, but if it is removed, it need to change getSMPSubscriber :: SMPServer -> M SMPSubscriber getSMPSubscriber smpServer = @@ -466,10 +471,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge subs <- atomically $ readTQueue subscriberSubQ updated <- liftIO $ batchUpdateSubStatus st subs NSPending logSubStatus smpServer "subscribing" (L.length subs) updated - liftIO $ subscribeQueues smpServer subs - - subscribeQueues :: SMPServer -> NonEmpty ServerNtfSub -> IO () - subscribeQueues srv subs = subscribeQueuesNtfs ca srv (L.map snd subs) + liftIO $ subscribeQueuesNtfs ca smpServer $ L.map snd subs receiveSMP :: M () receiveSMP = forever $ do @@ -613,7 +615,7 @@ periodicNtfsThread NtfPushServer {pushQ} = do threadDelay interval now <- systemSeconds <$> getSystemTime cnt <- withPeriodicNtfTokens st now $ \tkn -> atomically $ writeTBQueue pushQ (tkn, PNCheckMessages) - logInfo $ "Scheduled periodic notifications: " <> tshow cnt + logNote $ "Scheduled periodic notifications: " <> tshow cnt runNtfClientTransport :: Transport c => THandleNTF c 'TServer -> M () runNtfClientTransport th@THandle {params} = do @@ -829,14 +831,14 @@ incNtfStat statSel = do restoreServerLastNtfs :: NtfSTMStore -> FilePath -> IO () restoreServerLastNtfs st f = whenM (doesFileExist f) $ do - logInfo $ "restoring last notifications from file " <> T.pack f + logNote $ "restoring last notifications from file " <> T.pack f runExceptT (liftIO (B.readFile f) >>= mapM restoreNtf . B.lines) >>= \case Left e -> do logError . T.pack $ "error restoring last notifications: " <> e exitFailure Right _ -> do renameFile f $ f <> ".bak" - logInfo "last notifications restored" + logNote "last notifications restored" where restoreNtf s = do TNMRv1 tknId ntf <- liftEither . first (ntfErr "parsing") $ strDecode s @@ -851,21 +853,21 @@ saveServerStats = >>= mapM_ (\f -> asks serverStats >>= liftIO . getNtfServerStatsData >>= liftIO . saveStats f) where saveStats f stats = do - logInfo $ "saving server stats to file " <> T.pack f + logNote $ "saving server stats to file " <> T.pack f B.writeFile f $ strEncode stats - logInfo "server stats saved" + logNote "server stats saved" restoreServerStats :: M () restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats where restoreStats f = whenM (doesFileExist f) $ do - logInfo $ "restoring server stats from file " <> T.pack f + logNote $ "restoring server stats from file " <> T.pack f liftIO (strDecode <$> B.readFile f) >>= \case Right d -> do s <- asks serverStats liftIO $ setNtfServerStats s d renameFile f $ f <> ".bak" - logInfo "server stats restored" + logNote "server stats restored" Left e -> do - logInfo $ "error restoring server stats: " <> T.pack e + logNote $ "error restoring server stats: " <> T.pack e liftIO exitFailure diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 415f341cb..2f55a1dd0 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -106,7 +106,7 @@ newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsCo where compactDbStoreLog = \case Just f -> do - logInfo $ "compacting store log " <> T.pack f + logNote $ "compacting store log " <> T.pack f newNtfSTMStore >>= readWriteNtfSTMStore False f >>= closeStoreLog Nothing -> do logError "Error: `--compact-log` used without `enable: on` option in STORE_LOG section of INI file" diff --git a/src/Simplex/Messaging/Notifications/Server/Main.hs b/src/Simplex/Messaging/Notifications/Server/Main.hs index f59c565bc..a073eee18 100644 --- a/src/Simplex/Messaging/Notifications/Server/Main.hs +++ b/src/Simplex/Messaging/Notifications/Server/Main.hs @@ -10,12 +10,15 @@ module Simplex.Messaging.Notifications.Server.Main where +import Control.Logger.Simple (setLogLevel) import Control.Monad ((<$!>)) import qualified Data.ByteString.Char8 as B import Data.Functor (($>)) import Data.Ini (lookupValue, readIniFile) import Data.Int (Int64) import Data.Maybe (fromMaybe) +import Data.Set (Set) +import qualified Data.Set as S import qualified Data.Text as T import Data.Text.Encoding (encodeUtf8) import qualified Data.Text.IO as T @@ -27,6 +30,7 @@ import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..)) import Simplex.Messaging.Client (HostMode (..), NetworkConfig (..), ProtocolClientConfig (..), SocksMode (..), defaultNetworkConfig, textToHostMode) import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..), defaultSMPClientAgentConfig) import qualified Simplex.Messaging.Crypto as C +import Simplex.Messaging.Notifications.Protocol (NtfTokenId) import Simplex.Messaging.Notifications.Server (runNtfServer, restoreServerLastNtfs) import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..), defaultInactiveClientExpiration) import Simplex.Messaging.Notifications.Server.Push.APNS (defaultAPNSPushClientConfig) @@ -36,8 +40,9 @@ import Simplex.Messaging.Notifications.Server.StoreLog (readWriteNtfSTMStore) import Simplex.Messaging.Notifications.Transport (supportedServerNTFVRange) import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), pattern NtfServer) import Simplex.Messaging.Server.CLI -import Simplex.Messaging.Server.Env.STM (StartOptions) +import Simplex.Messaging.Server.Env.STM (StartOptions (..)) import Simplex.Messaging.Server.Expiration +import Simplex.Messaging.Server.Main (strParse) import Simplex.Messaging.Server.Main.Init (iniDbOpts) import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..)) import Simplex.Messaging.Server.StoreLog (closeStoreLog) @@ -72,7 +77,7 @@ ntfServerCLI cfgPath logPath = storeLogExists <- doesFileExist storeLogFilePath lastNtfsExists <- doesFileExist defaultLastNtfsFile case cmd of - SCImport + SCImport skipTokens | schemaExists && (storeLogExists || lastNtfsExists) -> exitConfigureNtfStore connstr schema | schemaExists -> do putStrLn $ "Schema " <> B.unpack schema <> " already exists in PostrgreSQL database: " <> B.unpack connstr @@ -94,7 +99,7 @@ ntfServerCLI cfgPath logPath = restoreServerLastNtfs stmStore defaultLastNtfsFile let storeCfg = PostgresStoreCfg {dbOpts = dbOpts {createSchema = True}, dbStoreLogPath = Nothing, confirmMigrations = MCConsole, deletedTTL = iniDeletedTTL ini} ps <- newNtfDbStore storeCfg - (tCnt, sCnt, nCnt) <- importNtfSTMStore ps stmStore + (tCnt, sCnt, nCnt) <- importNtfSTMStore ps stmStore skipTokens renameFile storeLogFile $ storeLogFile <> ".bak" putStrLn $ "Import completed: " <> show tCnt <> " tokens, " <> show sCnt <> " subscriptions, " <> show nCnt <> " last token notifications." putStrLn "Configure database options in INI file." @@ -161,9 +166,6 @@ ntfServerCLI cfgPath logPath = <> iniDbOpts dbOptions defaultNtfDBOpts <> "Time to retain deleted entities in the database, days.\n" <> ("# db_deleted_ttl: " <> tshow defaultDeletedTTL <> "\n\n") - <> "# Last notifications are optionally saved and restored when the server restarts,\n\ - \# they are preserved in the .bak file until the next restart.\n" - <> ("restore_last_notifications: " <> onOff enableStoreLog <> "\n\n") <> "log_stats: off\n\n\ \[AUTH]\n\ \# control_port_admin_password:\n\ @@ -200,6 +202,7 @@ ntfServerCLI cfgPath logPath = <> ("# check_interval: " <> tshow (checkInterval defaultInactiveClientExpiration) <> "\n") enableStoreLog' = settingIsOn "STORE_LOG" "enable" runServer startOptions ini = do + setLogLevel $ logLevel startOptions hSetBuffering stdout LineBuffering hSetBuffering stderr LineBuffering fp <- checkSavedFingerprint cfgPath defaultX509Config @@ -230,8 +233,8 @@ ntfServerCLI cfgPath logPath = subIdBytes = 24, regCodeBytes = 32, clientQSize = 64, - subQSize = 512, - pushQSize = 16384, + subQSize = 2048, + pushQSize = 32768, smpAgentCfg = defaultSMPClientAgentConfig { smpCfg = @@ -296,7 +299,7 @@ data CliCommand | Delete | Database StoreCmd DBOpts -data StoreCmd = SCImport | SCExport +data StoreCmd = SCImport (Set NtfTokenId) | SCExport data InitOptions = InitOptions { enableStoreLog :: Bool, @@ -332,9 +335,17 @@ cliCommandP cfgPath logPath iniFile = where databaseCmdP = hsubparser - ( command "import" (info (pure SCImport) (progDesc $ "Import store logs into a new PostgreSQL database schema")) + ( command "import" (info (SCImport <$> skipTokensP) (progDesc $ "Import store logs into a new PostgreSQL database schema")) <> command "export" (info (pure SCExport) (progDesc $ "Export PostgreSQL database schema to store logs")) ) + skipTokensP :: Parser (Set NtfTokenId) + skipTokensP = + option + strParse + ( long "skip-tokens" + <> help "Skip tokens during import" + <> value S.empty + ) initP :: Parser InitOptions initP = do enableStoreLog <- diff --git a/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs b/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs index 7439e4fea..39aeb9329 100644 --- a/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs +++ b/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs @@ -33,15 +33,19 @@ import qualified Data.ByteString.Base64.URL as U import Data.ByteString.Builder (lazyByteString) import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Lazy.Char8 as LB +import qualified Data.CaseInsensitive as CI import Data.Int (Int64) +import Data.List (find) import Data.List.NonEmpty (NonEmpty (..)) import Data.Map.Strict (Map) import Data.Maybe (isNothing) import Data.Text (Text) import qualified Data.Text as T +import Data.Text.Encoding (encodeUtf8) import Data.Time.Clock.System import qualified Data.X509 as X import qualified Data.X509.CertificateStore as XS +import Network.HPACK.Token as HT import Network.HTTP.Types (Status) import qualified Network.HTTP.Types as N import Network.HTTP2.Client (Request) @@ -330,9 +334,16 @@ apnsPushProviderClient c@APNSPushClient {nonceDrg, apnsCfg} tkn@NtfTknRec {token HTTP2Response {response, respBody = HTTP2Body {bodyHead}} <- liftHTTPS2 $ sendRequest http2 req Nothing let status = H.responseStatus response reason' = maybe "" reason $ J.decodeStrict' bodyHead - logDebug $ "APNS response: " <> T.pack (show status) <> " " <> reason' + if status == Just N.ok200 + then logDebug $ "APNS response: ok" <> apnsIds response + else logWarn $ "APNS error: " <> T.pack (show status) <> " " <> reason' <> apnsIds response result status reason' where + apnsIds response = headerStr "apns-id" <> headerStr "apns-unique-id" + where + headerStr name = + maybe "" (\(_, v) -> ", " <> name <> ": " <> safeDecodeUtf8 v) $ + find (\(t, _) -> HT.tokenKey t == CI.mk (encodeUtf8 name)) (fst (H.responseHeaders response)) result :: Maybe Status -> Text -> ExceptT PushProviderError IO () result status reason' | status == Just N.ok200 = pure () diff --git a/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs b/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs index 031eda711..112d4cedd 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs +++ b/src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs @@ -113,7 +113,7 @@ closeNtfDbStore NtfPostgresStore {dbStore, dbStoreLog} = do addNtfToken :: NtfPostgresStore -> NtfTknRec -> IO (Either ErrorType ()) addNtfToken st tkn = - withDB "addNtfToken" st $ \db -> + withFastDB "addNtfToken" st $ \db -> E.try (DB.execute db insertNtfTknQuery $ ntfTknToRow tkn) >>= bimapM handleDuplicate (\_ -> withLog "addNtfToken" st (`logCreateToken` tkn)) @@ -127,7 +127,7 @@ insertNtfTknQuery = replaceNtfToken :: NtfPostgresStore -> NtfTknRec -> IO (Either ErrorType ()) replaceNtfToken st NtfTknRec {ntfTknId, token = token@(DeviceToken pp ppToken), tknStatus, tknRegCode = code@(NtfRegCode regCode)} = - withDB "replaceNtfToken" st $ \db -> runExceptT $ do + withFastDB "replaceNtfToken" st $ \db -> runExceptT $ do ExceptT $ assertUpdated <$> DB.execute db @@ -156,7 +156,7 @@ findNtfTokenRegistration st (NewNtfTkn (DeviceToken pp ppToken) tknVerifyKey _) getNtfToken_ :: ToRow q => NtfPostgresStore -> Query -> q -> IO (Either ErrorType (Maybe NtfTknRec)) getNtfToken_ st cond params = - withDB' "getNtfToken" st $ \db -> do + withFastDB' "getNtfToken" st $ \db -> do tkn_ <- maybeFirstRow rowToNtfTkn $ DB.query db (ntfTknQuery <> cond) params mapM_ (updateTokenDate st db) tkn_ pure tkn_ @@ -185,7 +185,7 @@ rowToNtfTkn (ntfTknId, pp, Binary ppToken, tknStatus, tknVerifyKey, tknDhPrivKey deleteNtfToken :: NtfPostgresStore -> NtfTokenId -> IO (Either ErrorType [(SMPServer, [NotifierId])]) deleteNtfToken st tknId = - withDB "deleteNtfToken" st $ \db -> runExceptT $ do + withFastDB "deleteNtfToken" st $ \db -> runExceptT $ do -- This SELECT obtains exclusive lock on token row and prevents any inserts -- into other tables for this token ID until the deletion completes. _ <- ExceptT $ firstRow (fromOnly @Int) AUTH $ @@ -230,7 +230,7 @@ rowToSMPQueue (host, port, kh, nId) = SMPQueueNtf (SMPServer host port kh) nId updateTknCronInterval :: NtfPostgresStore -> NtfTokenId -> Word16 -> IO (Either ErrorType ()) updateTknCronInterval st tknId cronInt = - withDB "updateTknCronInterval" st $ \db -> runExceptT $ do + withFastDB "updateTknCronInterval" st $ \db -> runExceptT $ do ExceptT $ assertUpdated <$> DB.execute db "UPDATE tokens SET cron_interval = ? WHERE token_id = ?" (cronInt, tknId) withLog "updateTknCronInterval" st $ \sl -> logTokenCron sl tknId 0 @@ -254,10 +254,27 @@ getUsedSMPServers st = |] (Only (In [NSNew, NSPending, NSActive, NSInactive])) -foldNtfSubscriptions :: NtfPostgresStore -> SMPServer -> Int -> s -> (s -> ServerNtfSub -> IO s) -> IO s -foldNtfSubscriptions st srv fetchCount state action = - withConnection (dbStore st) $ \db -> - DB.foldWithOptions opts db query params state $ \s -> action s . toServerNtfSub +getServerNtfSubscriptions :: NtfPostgresStore -> SMPServer -> Maybe NtfSubscriptionId -> Int -> IO (Either ErrorType [ServerNtfSub]) +getServerNtfSubscriptions st srv afterSubId_ count = + withDB' "getServerNtfSubscriptions" st $ \db -> do + subs <- + map toServerNtfSub <$> case afterSubId_ of + Nothing -> + DB.query db (query <> orderLimit) (srvToRow srv :. (statusIn, count)) + Just afterSubId -> + DB.query db (query <> " AND s.subscription_id > ?" <> orderLimit) (srvToRow srv :. (statusIn, afterSubId, count)) + void $ + DB.executeMany + db + [sql| + UPDATE subscriptions s + SET status = upd.status + FROM (VALUES(?, ?)) AS upd(status, subscription_id) + WHERE s.subscription_id = (upd.subscription_id :: BYTEA) + AND s.status != upd.status + |] + (map ((NSPending,) . fst) subs) + pure subs where query = [sql| @@ -267,15 +284,15 @@ foldNtfSubscriptions st srv fetchCount state action = WHERE p.smp_host = ? AND p.smp_port = ? AND p.smp_keyhash = ? AND s.status IN ? |] - params = srvToRow srv :. Only (In [NSNew, NSPending, NSActive, NSInactive]) - opts = DB.defaultFoldOptions {DB.fetchQuantity = DB.Fixed fetchCount} + orderLimit = " ORDER BY s.subscription_id LIMIT ?" + statusIn = In [NSNew, NSPending, NSActive, NSInactive] toServerNtfSub (ntfSubId, notifierId, notifierKey) = (ntfSubId, (notifierId, notifierKey)) -- Returns token and subscription. -- If subscription exists but belongs to another token, returns Left AUTH findNtfSubscription :: NtfPostgresStore -> NtfTokenId -> SMPQueueNtf -> IO (Either ErrorType (NtfTknRec, Maybe NtfSubRec)) findNtfSubscription st tknId q = - withDB "findNtfSubscription" st $ \db -> runExceptT $ do + withFastDB "findNtfSubscription" st $ \db -> runExceptT $ do tkn@NtfTknRec {ntfTknId, tknStatus} <- ExceptT $ getNtfToken st tknId unless (allowNtfSubCommands tknStatus) $ throwE AUTH liftIO $ updateTokenDate st db tkn @@ -296,7 +313,7 @@ findNtfSubscription st tknId q = getNtfSubscription :: NtfPostgresStore -> NtfSubscriptionId -> IO (Either ErrorType (NtfTknRec, NtfSubRec)) getNtfSubscription st subId = - withDB "getNtfSubscription" st $ \db -> runExceptT $ do + withFastDB "getNtfSubscription" st $ \db -> runExceptT $ do r@(tkn@NtfTknRec {tknStatus}, _) <- ExceptT $ firstRow rowToNtfTknSub AUTH $ DB.query @@ -333,7 +350,7 @@ mkNtfSubRec ntfSubId (NewNtfSub tokenId smpQueue notifierKey) = updateTknStatus :: NtfPostgresStore -> NtfTknRec -> NtfTknStatus -> IO (Either ErrorType ()) updateTknStatus st tkn status = - withDB' "updateTknStatus" st $ \db -> updateTknStatus_ st db tkn status + withFastDB' "updateTknStatus" st $ \db -> updateTknStatus_ st db tkn status updateTknStatus_ :: NtfPostgresStore -> DB.Connection -> NtfTknRec -> NtfTknStatus -> IO () updateTknStatus_ st db NtfTknRec {ntfTknId} status = do @@ -343,13 +360,13 @@ updateTknStatus_ st db NtfTknRec {ntfTknId} status = do -- unless it was already active setTknStatusConfirmed :: NtfPostgresStore -> NtfTknRec -> IO (Either ErrorType ()) setTknStatusConfirmed st NtfTknRec {ntfTknId} = - withDB' "updateTknStatus" st $ \db -> do + withFastDB' "updateTknStatus" st $ \db -> do updated <- DB.execute db "UPDATE tokens SET status = ? WHERE token_id = ? AND status != ? AND status != ?" (NTConfirmed, ntfTknId, NTConfirmed, NTActive) when (updated > 0) $ withLog "updateTknStatus" st $ \sl -> logTokenStatus sl ntfTknId NTConfirmed setTokenActive :: NtfPostgresStore -> NtfTknRec -> IO (Either ErrorType ()) setTokenActive st tkn@NtfTknRec {ntfTknId, token = DeviceToken pp ppToken} = - withDB' "setTokenActive" st $ \db -> do + withFastDB' "setTokenActive" st $ \db -> do updateTknStatus_ st db tkn NTActive -- this removes other instances of the same token, e.g. because of repeated token registration attempts tknIds <- @@ -377,7 +394,7 @@ updateTokenCronSentAt st tknId now = addNtfSubscription :: NtfPostgresStore -> NtfSubRec -> IO (Either ErrorType Bool) addNtfSubscription st sub = - withDB "addNtfSubscription" st $ \db -> runExceptT $ do + withFastDB "addNtfSubscription" st $ \db -> runExceptT $ do srvId :: Int64 <- ExceptT $ upsertServer db $ ntfSubServer' sub n <- liftIO $ DB.execute db insertNtfSubQuery $ ntfSubToRow srvId sub withLog "addNtfSubscription" st (`logCreateSubscription` sub) @@ -423,14 +440,14 @@ ntfSubToRow srvId NtfSubRec {ntfSubId, tokenId, smpQueue = SMPQueueNtf _ nId, no deleteNtfSubscription :: NtfPostgresStore -> NtfSubscriptionId -> IO (Either ErrorType ()) deleteNtfSubscription st subId = - withDB "deleteNtfSubscription" st $ \db -> runExceptT $ do + withFastDB "deleteNtfSubscription" st $ \db -> runExceptT $ do ExceptT $ assertUpdated <$> DB.execute db "DELETE FROM subscriptions WHERE subscription_id = ?" (Only subId) withLog "deleteNtfSubscription" st (`logDeleteSubscription` subId) updateSrvSubStatus :: NtfPostgresStore -> SMPQueueNtf -> NtfSubStatus -> IO (Either ErrorType ()) updateSrvSubStatus st q status = - withDB' "updateSrvSubStatus" st $ \db -> do + withFastDB' "updateSrvSubStatus" st $ \db -> do subId_ :: Maybe NtfSubscriptionId <- maybeFirstRow fromOnly $ DB.query @@ -507,7 +524,7 @@ batchUpdateStatus_ st srv mkParams = batchUpdateSubStatus :: NtfPostgresStore -> NonEmpty ServerNtfSub -> NtfSubStatus -> IO Int64 batchUpdateSubStatus st subs status = - fmap (fromRight (-1)) $ withDB' "batchUpdateSubStatus" st $ \db -> do + fmap (fromRight (-1)) $ withFastDB' "batchUpdateSubStatus" st $ \db -> do let params = L.toList $ L.map (\(subId, _) -> (status, subId)) subs subIds <- DB.returning @@ -533,7 +550,7 @@ batchUpdateSubStatus st subs status = addTokenLastNtf :: NtfPostgresStore -> PNMessageData -> IO (Either ErrorType (NtfTknRec, NonEmpty PNMessageData)) addTokenLastNtf st newNtf = - withDB "addTokenLastNtf" st $ \db -> runExceptT $ do + withFastDB "addTokenLastNtf" st $ \db -> runExceptT $ do (tkn@NtfTknRec {ntfTknId = tId, tknStatus}, sId) <- ExceptT $ firstRow toTokenSubId AUTH $ DB.query @@ -607,8 +624,8 @@ getEntityCounts st = count (Only n : _) = n count [] = 0 -importNtfSTMStore :: NtfPostgresStore -> NtfSTMStore -> IO (Int64, Int64, Int64) -importNtfSTMStore NtfPostgresStore {dbStore = s} stmStore = do +importNtfSTMStore :: NtfPostgresStore -> NtfSTMStore -> S.Set NtfTokenId -> IO (Int64, Int64, Int64) +importNtfSTMStore NtfPostgresStore {dbStore = s} stmStore skipTokens = do (tIds, tCnt) <- importTokens subLookup <- readTVarIO $ subscriptionLookup stmStore sCnt <- importSubscriptions tIds subLookup @@ -638,6 +655,7 @@ importNtfSTMStore NtfPostgresStore {dbStore = s} stmStore = do case M.lookup (tokenKey tkn) deviceTokens of Just ts | length ts < 2 -> pure True + | ntfTknId `S.member` skipTokens -> False <$ putStrLn ("Skipped token " <> enc ntfTknId <> " from --skip-tokens") | otherwise -> readTVarIO tknStatus >>= \case NTConfirmed -> do @@ -645,8 +663,8 @@ importNtfSTMStore NtfPostgresStore {dbStore = s} stmStore = do if hasSubs then pure True else do - anyActive <- anyM $ map (\NtfTknData {tknStatus = tknStatus'} -> (NTActive ==) <$> readTVarIO tknStatus') ts - if anyActive + anyBetterToken <- anyM $ map (\NtfTknData {tknStatus = tknStatus'} -> activeOrInvalid <$> readTVarIO tknStatus') ts + if anyBetterToken then False <$ putStrLn ("Skipped duplicate inactive token " <> enc ntfTknId) else case findIndex (\NtfTknData {ntfTknId = tId} -> tId == ntfTknId) ts of Just 0 -> pure True -- keeping the first token @@ -654,6 +672,10 @@ importNtfSTMStore NtfPostgresStore {dbStore = s} stmStore = do Nothing -> True <$ putStrLn "Error: no device token in the list" _ -> pure True Nothing -> True <$ putStrLn "Error: no device token in lookup map" + activeOrInvalid = \case + NTActive -> True + NTInvalid _ -> True + _ -> False -- importTkn db !n tkn@NtfTknData {ntfTknId} = do -- tknRow <- ntfTknToRow <$> mkTknRec tkn -- (DB.execute db insertNtfTknQuery tknRow >>= pure . (n + )) `E.catch` \(e :: E.SomeException) -> @@ -799,12 +821,25 @@ exportNtfDbStore NtfPostgresStore {dbStore = s, dbStoreLog = Just sl} lastNtfsFi |] encodeLastNtf tknId ntf = strEncode (TNMRv1 tknId ntf) `B.snoc` '\n' +withFastDB' :: String -> NtfPostgresStore -> (DB.Connection -> IO a) -> IO (Either ErrorType a) +withFastDB' op st action = withFastDB op st $ fmap Right . action +{-# INLINE withFastDB' #-} + withDB' :: String -> NtfPostgresStore -> (DB.Connection -> IO a) -> IO (Either ErrorType a) withDB' op st action = withDB op st $ fmap Right . action +{-# INLINE withDB' #-} + +withFastDB :: forall a. String -> NtfPostgresStore -> (DB.Connection -> IO (Either ErrorType a)) -> IO (Either ErrorType a) +withFastDB op st = withDB_ op st True +{-# INLINE withFastDB #-} withDB :: forall a. String -> NtfPostgresStore -> (DB.Connection -> IO (Either ErrorType a)) -> IO (Either ErrorType a) -withDB op st action = - E.uninterruptibleMask_ $ E.try (withTransaction (dbStore st) action) >>= either logErr pure +withDB op st = withDB_ op st False +{-# INLINE withDB #-} + +withDB_ :: forall a. String -> NtfPostgresStore -> Bool -> (DB.Connection -> IO (Either ErrorType a)) -> IO (Either ErrorType a) +withDB_ op st priority action = + E.uninterruptibleMask_ $ E.try (withTransactionPriority (dbStore st) priority action) >>= either logErr pure where logErr :: E.SomeException -> IO (Either ErrorType a) logErr e = logError ("STORE: " <> T.pack err) $> Left (STORE err) diff --git a/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql b/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql new file mode 100644 index 000000000..4c98a1161 --- /dev/null +++ b/src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql @@ -0,0 +1,178 @@ + + +SET statement_timeout = 0; +SET lock_timeout = 0; +SET idle_in_transaction_session_timeout = 0; +SET client_encoding = 'UTF8'; +SET standard_conforming_strings = on; +SELECT pg_catalog.set_config('search_path', '', false); +SET check_function_bodies = false; +SET xmloption = content; +SET client_min_messages = warning; +SET row_security = off; + + +CREATE SCHEMA ntf_server; + + +SET default_table_access_method = heap; + + +CREATE TABLE ntf_server.last_notifications ( + token_ntf_id bigint NOT NULL, + token_id bytea NOT NULL, + subscription_id bytea NOT NULL, + sent_at timestamp with time zone NOT NULL, + nmsg_nonce bytea NOT NULL, + nmsg_data bytea NOT NULL +); + + + +ALTER TABLE ntf_server.last_notifications ALTER COLUMN token_ntf_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME ntf_server.last_notifications_token_ntf_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE ntf_server.migrations ( + name text NOT NULL, + ts timestamp without time zone NOT NULL, + down text +); + + + +CREATE TABLE ntf_server.smp_servers ( + smp_server_id bigint NOT NULL, + smp_host text NOT NULL, + smp_port text NOT NULL, + smp_keyhash bytea NOT NULL +); + + + +ALTER TABLE ntf_server.smp_servers ALTER COLUMN smp_server_id ADD GENERATED ALWAYS AS IDENTITY ( + SEQUENCE NAME ntf_server.smp_servers_smp_server_id_seq + START WITH 1 + INCREMENT BY 1 + NO MINVALUE + NO MAXVALUE + CACHE 1 +); + + + +CREATE TABLE ntf_server.subscriptions ( + subscription_id bytea NOT NULL, + token_id bytea NOT NULL, + smp_server_id bigint, + smp_notifier_id bytea NOT NULL, + smp_notifier_key bytea NOT NULL, + status text NOT NULL +); + + + +CREATE TABLE ntf_server.tokens ( + token_id bytea NOT NULL, + push_provider text NOT NULL, + push_provider_token bytea NOT NULL, + status text NOT NULL, + verify_key bytea NOT NULL, + dh_priv_key bytea NOT NULL, + dh_secret bytea NOT NULL, + reg_code bytea NOT NULL, + cron_interval bigint NOT NULL, + cron_sent_at bigint, + updated_at bigint +); + + + +ALTER TABLE ONLY ntf_server.last_notifications + ADD CONSTRAINT last_notifications_pkey PRIMARY KEY (token_ntf_id); + + + +ALTER TABLE ONLY ntf_server.migrations + ADD CONSTRAINT migrations_pkey PRIMARY KEY (name); + + + +ALTER TABLE ONLY ntf_server.smp_servers + ADD CONSTRAINT smp_servers_pkey PRIMARY KEY (smp_server_id); + + + +ALTER TABLE ONLY ntf_server.subscriptions + ADD CONSTRAINT subscriptions_pkey PRIMARY KEY (subscription_id); + + + +ALTER TABLE ONLY ntf_server.tokens + ADD CONSTRAINT tokens_pkey PRIMARY KEY (token_id); + + + +CREATE INDEX idx_last_notifications_subscription_id ON ntf_server.last_notifications USING btree (subscription_id); + + + +CREATE INDEX idx_last_notifications_token_id_sent_at ON ntf_server.last_notifications USING btree (token_id, sent_at); + + + +CREATE UNIQUE INDEX idx_last_notifications_token_subscription ON ntf_server.last_notifications USING btree (token_id, subscription_id); + + + +CREATE UNIQUE INDEX idx_smp_servers ON ntf_server.smp_servers USING btree (smp_host, smp_port, smp_keyhash); + + + +CREATE UNIQUE INDEX idx_subscriptions_smp_server_id_notifier_id ON ntf_server.subscriptions USING btree (smp_server_id, smp_notifier_id); + + + +CREATE INDEX idx_subscriptions_smp_server_id_status ON ntf_server.subscriptions USING btree (smp_server_id, status); + + + +CREATE INDEX idx_subscriptions_token_id ON ntf_server.subscriptions USING btree (token_id); + + + +CREATE UNIQUE INDEX idx_tokens_push_provider_token ON ntf_server.tokens USING btree (push_provider, push_provider_token, verify_key); + + + +CREATE INDEX idx_tokens_status_cron_interval_sent_at ON ntf_server.tokens USING btree (status, cron_interval, ((cron_sent_at + (cron_interval * 60)))); + + + +ALTER TABLE ONLY ntf_server.last_notifications + ADD CONSTRAINT last_notifications_subscription_id_fkey FOREIGN KEY (subscription_id) REFERENCES ntf_server.subscriptions(subscription_id) ON UPDATE RESTRICT ON DELETE CASCADE; + + + +ALTER TABLE ONLY ntf_server.last_notifications + ADD CONSTRAINT last_notifications_token_id_fkey FOREIGN KEY (token_id) REFERENCES ntf_server.tokens(token_id) ON UPDATE RESTRICT ON DELETE CASCADE; + + + +ALTER TABLE ONLY ntf_server.subscriptions + ADD CONSTRAINT subscriptions_smp_server_id_fkey FOREIGN KEY (smp_server_id) REFERENCES ntf_server.smp_servers(smp_server_id) ON UPDATE RESTRICT ON DELETE RESTRICT; + + + +ALTER TABLE ONLY ntf_server.subscriptions + ADD CONSTRAINT subscriptions_token_id_fkey FOREIGN KEY (token_id) REFERENCES ntf_server.tokens(token_id) ON UPDATE RESTRICT ON DELETE CASCADE; + + + diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index 2f3dea57f..db041c4e7 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -208,14 +208,14 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt sigIntHandler = Just (sigIntAction, toDyn ()) void $ liftIO $ setHandler sigINT sigIntHandler atomically $ readTMVar flagINT - logInfo "Received SIGINT, stopping server..." + logNote "Received SIGINT, stopping server..." stopServer :: Server -> M () stopServer s = do asks serverActive >>= atomically . (`writeTVar` False) - logInfo "Saving server state..." + logNote "Saving server state..." withLock' (savingLock s) "final" $ saveServer True >> closeServer - logInfo "Server stopped" + logNote "Server stopped" saveServer :: Bool -> M () saveServer drainMsgs = do @@ -383,9 +383,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt expire :: forall s. MsgStoreClass s => s -> ServerStats -> Int64 -> IO () expire ms stats interval = do threadDelay' interval - logInfo "Started expiring messages..." + logNote "Started expiring messages..." n <- compactQueues @(StoreQueue s) $ queueStore ms - when (n > 0) $ logInfo $ "Removed " <> tshow n <> " old deleted queues from the database." + when (n > 0) $ logNote $ "Removed " <> tshow n <> " old deleted queues from the database." now <- systemSeconds <$> getSystemTime tryAny (expireOldMessages False ms now ttl) >>= \case Right msgStats@MessageStats {storedMsgsCount = stored, expiredMsgsCount = expired} -> do @@ -1827,15 +1827,15 @@ saveServerMessages :: Bool -> AMsgStore -> IO () saveServerMessages drainMsgs = \case AMS SQSMemory SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of Just f -> exportMessages False ms f drainMsgs - Nothing -> logInfo "undelivered messages are not saved" - AMS _ SMSJournal _ -> logInfo "closed journal message storage" + Nothing -> logNote "undelivered messages are not saved" + AMS _ SMSJournal _ -> logNote "closed journal message storage" exportMessages :: MsgStoreClass s => Bool -> s -> FilePath -> Bool -> IO () exportMessages tty ms f drainMsgs = do - logInfo $ "saving messages to file " <> T.pack f + logNote $ "saving messages to file " <> T.pack f liftIO $ withFile f WriteMode $ \h -> tryAny (unsafeWithAllMsgQueues tty True ms $ saveQueueMsgs h) >>= \case - Right (Sum total) -> logInfo $ "messages saved: " <> tshow total + Right (Sum total) -> logNote $ "messages saved: " <> tshow total Left e -> do logError $ "error exporting messages: " <> tshow e exitFailure @@ -1864,10 +1864,10 @@ processServerMessages StartOptions {skipWarnings} = do processJournalMessages old_ expire ms | expire = Just <$> case old_ of Just old -> do - logInfo "expiring journal store messages..." + logNote "expiring journal store messages..." run $ processExpireQueue old Nothing -> do - logInfo "validating journal store messages..." + logNote "validating journal store messages..." run processValidateQueue | otherwise = logWarn "skipping message expiration" $> Nothing where @@ -1885,7 +1885,7 @@ processServerMessages StartOptions {skipWarnings} = do importMessages :: forall s. MsgStoreClass s => Bool -> s -> FilePath -> Maybe Int64 -> Bool -> IO MessageStats importMessages tty ms f old_ skipWarnings = do - logInfo $ "restoring messages from file " <> T.pack f + logNote $ "restoring messages from file " <> T.pack f (_, (storedMsgsCount, expiredMsgsCount, overQuota)) <- foldLogLines tty f restoreMsg (Nothing, (0, 0, M.empty)) renameFile f $ f <> ".bak" @@ -1950,17 +1950,17 @@ importMessages tty ms f old_ skipWarnings = do printMessageStats :: T.Text -> MessageStats -> IO () printMessageStats name MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} = - logInfo $ name <> " stored: " <> tshow storedMsgsCount <> ", expired: " <> tshow expiredMsgsCount <> ", queues: " <> tshow storedQueues + logNote $ name <> " stored: " <> tshow storedMsgsCount <> ", expired: " <> tshow expiredMsgsCount <> ", queues: " <> tshow storedQueues saveServerNtfs :: M () saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs where saveNtfs f = do - logInfo $ "saving notifications to file " <> T.pack f + logNote $ "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" + logNote "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 @@ -1973,7 +1973,7 @@ restoreServerNtfs = Nothing -> pure newMessageStats where restoreNtfs f = do - logInfo $ "restoring notifications from file " <> T.pack f + logNote $ "restoring notifications from file " <> T.pack f ns <- asks ntfStore old <- asks (notificationExpiration . config) >>= liftIO . expireBeforeEpoch liftIO $ @@ -1985,7 +1985,7 @@ restoreServerNtfs = renameFile f $ f <> ".bak" let NtfStore ns' = ns storedQueues <- M.size <$> readTVarIO ns' - logInfo $ "notifications restored, " <> tshow lineCount <> " lines processed" + logNote $ "notifications restored, " <> tshow lineCount <> " lines processed" pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} where restoreNtf :: NtfStore -> Int64 -> (Int, Int, Int) -> LB.ByteString -> ExceptT String IO (Int, Int, Int) @@ -2006,15 +2006,15 @@ saveServerStats = >>= mapM_ (\f -> asks serverStats >>= liftIO . getServerStatsData >>= liftIO . saveStats f) where saveStats f stats = do - logInfo $ "saving server stats to file " <> T.pack f + logNote $ "saving server stats to file " <> T.pack f B.writeFile f $ strEncode stats - logInfo "server stats saved" + logNote "server stats saved" restoreServerStats :: Maybe MessageStats -> MessageStats -> M () restoreServerStats msgStats_ ntfStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats where restoreStats f = whenM (doesFileExist f) $ do - logInfo $ "restoring server stats from file " <> T.pack f + logNote $ "restoring server stats from file " <> T.pack f liftIO (strDecode <$> B.readFile f) >>= \case Right d@ServerStatsData {_qCount = statsQCount, _msgCount = statsMsgCount, _ntfCount = statsNtfCount} -> do s <- asks serverStats @@ -2026,12 +2026,12 @@ restoreServerStats msgStats_ ntfStats = asks (serverStatsBackupFile . config) >> _msgNtfExpired' = _msgNtfExpired d + expiredMsgsCount ntfStats liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired', _msgNtfExpired = _msgNtfExpired'} renameFile f $ f <> ".bak" - logInfo "server stats restored" + logNote "server stats restored" compareCounts "Queue" statsQCount _qCount compareCounts "Message" statsMsgCount _msgCount compareCounts "Notification" statsNtfCount _ntfCount Left e -> do - logInfo $ "error restoring server stats: " <> T.pack e + logNote $ "error restoring server stats: " <> T.pack e liftIO exitFailure compareCounts name statsCnt storeCnt = when (statsCnt /= storeCnt) $ logWarn $ name <> " count differs: stats: " <> tshow statsCnt <> ", store: " <> tshow storeCnt \ No newline at end of file diff --git a/src/Simplex/Messaging/Server/CLI.hs b/src/Simplex/Messaging/Server/CLI.hs index a678825af..ceb131b7f 100644 --- a/src/Simplex/Messaging/Server/CLI.hs +++ b/src/Simplex/Messaging/Server/CLI.hs @@ -12,6 +12,7 @@ module Simplex.Messaging.Server.CLI where +import Control.Logger.Simple (LogLevel (..)) import Control.Monad import Data.ASN1.Types (asn1CharacterToString) import Data.ByteString.Char8 (ByteString) @@ -218,6 +219,14 @@ startOptionsP = do ( long "compact-log" <> help "Compact store log (always enabled with `memory` storage for queues)" ) + logLevel <- + option + parseLogLevel + ( long "log-level" + <> metavar "LOG_LEVEL" + <> help "Logging level" + <> value LogInfo + ) skipWarnings <- switch ( long "skip-warnings" @@ -231,7 +240,7 @@ startOptionsP = do <> help "Confirm PostgreSQL database migration: up, down (default is manual confirmation)" <> value MCConsole ) - pure StartOptions {maintenance, compactLog, skipWarnings, confirmMigrations} + pure StartOptions {maintenance, compactLog, logLevel, skipWarnings, confirmMigrations} where parseConfirmMigrations :: ReadM MigrationConfirmation parseConfirmMigrations = eitherReader $ \case @@ -239,6 +248,16 @@ startOptionsP = do "down" -> Right MCYesUpDown _ -> Left "invalid migration confirmation, pass 'up' or 'down'" +parseLogLevel :: ReadM LogLevel +parseLogLevel = eitherReader $ \case + "trace" -> Right LogTrace + "debug" -> Right LogDebug + "info" -> Right LogInfo + "note" -> Right LogNote + "warn" -> Right LogWarn + "error" -> Right LogError + _ -> Left "Invalid log level" + genOnline :: FilePath -> CertOptions -> IO () genOnline cfgPath CertOptions {signAlgorithm_, commonName_} = do (signAlgorithm, commonName) <- diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index 5f3be4a98..8895ba8ed 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -136,6 +136,7 @@ data ServerConfig = ServerConfig data StartOptions = StartOptions { maintenance :: Bool, compactLog :: Bool, + logLevel :: LogLevel, skipWarnings :: Bool, confirmMigrations :: MigrationConfirmation } @@ -367,13 +368,13 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp where loadStoreLog :: StoreQueueClass q => (RecipientId -> QueueRec -> IO q) -> FilePath -> STMQueueStore q -> IO () loadStoreLog mkQ f st = do - logInfo $ "restoring queues from file " <> T.pack f + logNote $ "restoring queues from file " <> T.pack f sl <- readWriteQueueStore False mkQ f st setStoreLog st sl #if defined(dbServerPostgres) compactDbStoreLog = \case Just f -> do - logInfo $ "compacting queues in file " <> T.pack f + logNote $ "compacting queues in file " <> T.pack f st <- newMsgStore STMStoreConfig {storePath = Nothing, quota = msgQueueQuota} -- we don't need to have locks in the map sl <- readWriteQueueStore False (mkQueue st False) f (queueStore st) diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index 844d8d86d..081c91706 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -320,6 +320,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath = SPRandom -> BasicAuth <$> randomBase64 32 randomBase64 n = strEncode <$> (atomically . C.randomBytes n =<< C.newRandom) runServer startOptions ini = do + setLogLevel $ logLevel startOptions hSetBuffering stdout LineBuffering hSetBuffering stderr LineBuffering fp <- checkSavedFingerprint cfgPath defaultX509Config @@ -650,7 +651,7 @@ data CliCommand | Start StartOptions | Delete | Journal StoreCmd - | Database StoreCmd DBOpts + | Database StoreCmd DBOpts data StoreCmd = SCImport | SCExport | SCDelete @@ -835,5 +836,6 @@ cliCommandP cfgPath logPath iniFile = <> metavar (metavar' <> "_COUNTRY") <> help (help' <> " country") ) - strParse :: StrEncoding a => ReadM a - strParse = eitherReader $ parseAll strP . encodeUtf8 . T.pack + +strParse :: StrEncoding a => ReadM a +strParse = eitherReader $ parseAll strP . encodeUtf8 . T.pack diff --git a/src/Simplex/Messaging/Server/StoreLog.hs b/src/Simplex/Messaging/Server/StoreLog.hs index 80a2b75aa..dffc818e3 100644 --- a/src/Simplex/Messaging/Server/StoreLog.hs +++ b/src/Simplex/Messaging/Server/StoreLog.hs @@ -267,7 +267,7 @@ readWriteStoreLog readStore writeStore f st = logWarn $ "Server terminated abnormally on last start, restoring state from " <> T.pack tempBackup whenM (doesFileExist f) $ do renameFile f (f <> ".bak") - logInfo $ "preserved incomplete state " <> f' <> " as " <> (f' <> ".bak") + logNote $ "preserved incomplete state " <> f' <> " as " <> (f' <> ".bak") renameFile tempBackup f readWriteLog = do -- log backup is made in two steps to mitigate the crash during the compacting. @@ -280,14 +280,14 @@ readWriteStoreLog readStore writeStore f st = pure s writeLog msg = do s <- openWriteStoreLog False f - logInfo msg + logNote msg writeStore s st pure s renameBackup = do ts <- getCurrentTime let timedBackup = f <> "." <> iso8601Show ts renameFile tempBackup timedBackup - logInfo $ "original state preserved as " <> T.pack timedBackup + logNote $ "original state preserved as " <> T.pack timedBackup removeStoreLogBackups :: FilePath -> IO () removeStoreLogBackups f = do diff --git a/src/Simplex/Messaging/Transport/Server.hs b/src/Simplex/Messaging/Transport/Server.hs index 1f0d82195..0be54eb7b 100644 --- a/src/Simplex/Messaging/Transport/Server.hs +++ b/src/Simplex/Messaging/Transport/Server.hs @@ -202,7 +202,7 @@ startTCPServer started host port = withSocketsDo $ resolve >>= open >>= setStart sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr) setSocketOption sock ReuseAddr 1 withFdSocket sock setCloseOnExecIfNeeded - logInfo $ "binding to " <> tshow (addrAddress addr) + logNote $ "binding to " <> tshow (addrAddress addr) bind sock $ addrAddress addr listen sock 1024 pure sock diff --git a/src/Simplex/Messaging/Util.hs b/src/Simplex/Messaging/Util.hs index 2d92b4b5e..3d00257a2 100644 --- a/src/Simplex/Messaging/Util.hs +++ b/src/Simplex/Messaging/Util.hs @@ -224,6 +224,7 @@ groupOn = groupBy . eqOn groupAllOn :: Ord k => (a -> k) -> [a] -> [[a]] groupAllOn f = groupOn f . sortOn f +-- n must be > 0 toChunks :: Int -> [a] -> [NonEmpty a] toChunks _ [] = [] toChunks n xs = diff --git a/tests/NtfClient.hs b/tests/NtfClient.hs index 11d0400d6..f20264cb8 100644 --- a/tests/NtfClient.hs +++ b/tests/NtfClient.hs @@ -34,7 +34,7 @@ import Network.HTTP.Types (Status) import qualified Network.HTTP.Types as N import qualified Network.HTTP2.Server as H import Network.Socket -import SMPClient (ntfTestPort, prevRange, serverBracket) +import SMPClient (defaultStartOptions, ntfTestPort, prevRange, serverBracket) import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..)) import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..)) import Simplex.Messaging.Client (ProtocolClientConfig (..), chooseTransportHost, defaultNetworkConfig) @@ -48,7 +48,6 @@ import Simplex.Messaging.Notifications.Server.Push.APNS import Simplex.Messaging.Notifications.Server.Push.APNS.Internal import Simplex.Messaging.Notifications.Transport import Simplex.Messaging.Protocol -import Simplex.Messaging.Server.Env.STM (StartOptions (..)) import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..)) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport @@ -162,7 +161,7 @@ ntfServerCfg = prometheusMetricsFile = ntfTestPrometheusMetricsFile, ntfServerVRange = supportedServerNTFVRange, transportConfig = defaultTransportServerConfig, - startOptions = StartOptions {maintenance = False, compactLog = False, skipWarnings = False, confirmMigrations = MCYesUp} + startOptions = defaultStartOptions } ntfServerCfgVPrev :: NtfServerConfig diff --git a/tests/PostgresSchemaDump.hs b/tests/PostgresSchemaDump.hs new file mode 100644 index 000000000..de96e76ac --- /dev/null +++ b/tests/PostgresSchemaDump.hs @@ -0,0 +1,70 @@ +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} + +module PostgresSchemaDump (postgresSchemaDumpTest) where + +import Control.Concurrent (threadDelay) +import Control.DeepSeq +import Control.Monad (unless, void) +import qualified Data.ByteString.Char8 as B +import Data.List (dropWhileEnd) +import Data.Maybe (fromJust, isJust) +import Simplex.Messaging.Agent.Store.Postgres (closeDBStore, createDBStore) +import Simplex.Messaging.Agent.Store.Postgres.Common (DBOpts (..)) +import qualified Simplex.Messaging.Agent.Store.Postgres.Migrations as Migrations +import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationsToRun (..), toDownMigration) +import Simplex.Messaging.Util (ifM, whenM) +import System.Directory (doesFileExist, removeFile) +import System.Environment (lookupEnv) +import System.Process (readCreateProcess, shell) +import Test.Hspec + +testSchemaPath :: FilePath +testSchemaPath = "tests/tmp/test_schema.sql" + +postgresSchemaDumpTest :: [Migration] -> [String] -> DBOpts -> FilePath -> Spec +postgresSchemaDumpTest migrations skipComparisonForDownMigrations testDBOpts@DBOpts {connstr, schema = testDBSchema} srcSchemaPath = do + it "verify and overwrite schema dump" testVerifySchemaDump + it "verify schema down migrations" testSchemaMigrations + where + testVerifySchemaDump = do + savedSchema <- ifM (doesFileExist srcSchemaPath) (readFile srcSchemaPath) (pure "") + savedSchema `deepseq` pure () + void $ createDBStore testDBOpts migrations MCConsole + getSchema srcSchemaPath `shouldReturn` savedSchema + + testSchemaMigrations = do + let noDownMigrations = dropWhileEnd (\Migration {down} -> isJust down) migrations + Right st <- createDBStore testDBOpts noDownMigrations MCError + mapM_ (testDownMigration st) $ drop (length noDownMigrations) migrations + closeDBStore st + whenM (doesFileExist testSchemaPath) $ removeFile testSchemaPath + where + testDownMigration st m = do + putStrLn $ "down migration " <> name m + let downMigr = fromJust $ toDownMigration m + schema <- getSchema testSchemaPath + Migrations.run st $ MTRUp [m] + schema' <- getSchema testSchemaPath + schema' `shouldNotBe` schema + Migrations.run st $ MTRDown [downMigr] + unless (name m `elem` skipComparisonForDownMigrations) $ do + schema'' <- getSchema testSchemaPath + schema'' `shouldBe` schema + Migrations.run st $ MTRUp [m] + schema''' <- getSchema testSchemaPath + schema''' `shouldBe` schema' + + getSchema :: FilePath -> IO String + getSchema schemaPath = do + ci <- (Just "true" ==) <$> lookupEnv "CI" + let cmd = + ("pg_dump " <> B.unpack connstr <> " --schema " <> B.unpack testDBSchema) + <> " --schema-only --no-owner --no-privileges --no-acl --no-subscriptions --no-tablespaces > " + <> schemaPath + void $ readCreateProcess (shell cmd) "" + threadDelay 20000 + let sed = (if ci then "sed -i" else "sed -i ''") + void $ readCreateProcess (shell $ sed <> " '/^--/d' " <> schemaPath) "" + sch <- readFile schemaPath + sch `deepseq` pure sch diff --git a/tests/SMPClient.hs b/tests/SMPClient.hs index 801b7f42e..693049cca 100644 --- a/tests/SMPClient.hs +++ b/tests/SMPClient.hs @@ -14,6 +14,7 @@ module SMPClient where +import Control.Logger.Simple (LogLevel (..)) import Control.Monad.Except (runExceptT) import Data.ByteString.Char8 (ByteString) import Data.List.NonEmpty (NonEmpty) @@ -223,9 +224,12 @@ cfgMS msType = allowSMPProxy = False, serverClientConcurrency = 2, information = Nothing, - startOptions = StartOptions {maintenance = False, compactLog = False, skipWarnings = False, confirmMigrations = MCYesUp} + startOptions = defaultStartOptions } +defaultStartOptions :: StartOptions +defaultStartOptions = StartOptions {maintenance = False, compactLog = False, logLevel = LogError, skipWarnings = False, confirmMigrations = MCYesUp} + serverStoreConfig :: AStoreType -> AServerStoreCfg serverStoreConfig = serverStoreConfig_ False diff --git a/tests/ServerTests/SchemaDump.hs b/tests/ServerTests/SchemaDump.hs deleted file mode 100644 index e3ffdb5cb..000000000 --- a/tests/ServerTests/SchemaDump.hs +++ /dev/null @@ -1,98 +0,0 @@ -{-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE OverloadedStrings #-} - -module ServerTests.SchemaDump where - -import Control.Concurrent (threadDelay) -import Control.DeepSeq -import Control.Monad (unless, void) -import qualified Data.ByteString.Char8 as B -import Data.List (dropWhileEnd) -import Data.Maybe (fromJust, isJust) -import SMPClient -import Simplex.Messaging.Agent.Store.Postgres (closeDBStore, createDBStore) -import Simplex.Messaging.Agent.Store.Postgres.Common (DBOpts (..)) -import qualified Simplex.Messaging.Agent.Store.Postgres.Migrations as Migrations -import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationsToRun (..), toDownMigration) -import Simplex.Messaging.Server.QueueStore.Postgres.Migrations (serverMigrations) -import Simplex.Messaging.Util (ifM) -import System.Directory (doesFileExist, removeFile) -import System.Environment (lookupEnv) -import System.Process (readCreateProcess, readCreateProcessWithExitCode, shell) -import Test.Hspec - -testDBSchema :: B.ByteString -testDBSchema = "smp_server" - -serverSchemaPath :: FilePath -serverSchemaPath = "src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql" - -testSchemaPath :: FilePath -testSchemaPath = "tests/tmp/test_server_schema.sql" - -testServerDBOpts :: DBOpts -testServerDBOpts = - DBOpts - { connstr = testServerDBConnstr, - schema = testDBSchema, - poolSize = 3, - createSchema = True - } - -serverSchemaDumpTest :: Spec -serverSchemaDumpTest = do - it "verify and overwrite schema dump" testVerifySchemaDump - it "verify schema down migrations" testSchemaMigrations - -testVerifySchemaDump :: IO () -testVerifySchemaDump = do - savedSchema <- ifM (doesFileExist serverSchemaPath) (readFile serverSchemaPath) (pure "") - savedSchema `deepseq` pure () - void $ createDBStore testServerDBOpts serverMigrations MCConsole - getSchema serverSchemaPath `shouldReturn` savedSchema - -testSchemaMigrations :: IO () -testSchemaMigrations = do - let noDownMigrations = dropWhileEnd (\Migration {down} -> isJust down) serverMigrations - Right st <- createDBStore testServerDBOpts noDownMigrations MCError - mapM_ (testDownMigration st) $ drop (length noDownMigrations) serverMigrations - closeDBStore st - removeFile testSchemaPath - where - testDownMigration st m = do - putStrLn $ "down migration " <> name m - let downMigr = fromJust $ toDownMigration m - schema <- getSchema testSchemaPath - Migrations.run st $ MTRUp [m] - schema' <- getSchema testSchemaPath - schema' `shouldNotBe` schema - Migrations.run st $ MTRDown [downMigr] - unless (name m `elem` skipComparisonForDownMigrations) $ do - schema'' <- getSchema testSchemaPath - schema'' `shouldBe` schema - Migrations.run st $ MTRUp [m] - schema''' <- getSchema testSchemaPath - schema''' `shouldBe` schema' - -skipComparisonForDownMigrations :: [String] -skipComparisonForDownMigrations = - [ -- snd_secure moves to the bottom on down migration - "20250320_short_links" - ] - -getSchema :: FilePath -> IO String -getSchema schemaPath = do - ci <- (Just "true" ==) <$> lookupEnv "CI" - let cmd = - ("pg_dump " <> B.unpack testServerDBConnstr <> " --schema " <> B.unpack testDBSchema) - <> " --schema-only --no-owner --no-privileges --no-acl --no-subscriptions --no-tablespaces > " - <> schemaPath - (code, out, err) <- readCreateProcessWithExitCode (shell cmd) "" - print code - putStrLn $ "out: " <> out - putStrLn $ "err: " <> err - threadDelay 20000 - let sed = (if ci then "sed -i" else "sed -i ''") - void $ readCreateProcess (shell $ sed <> " '/^--/d' " <> schemaPath) "" - sch <- readFile schemaPath - sch `deepseq` pure sch diff --git a/tests/Test.hs b/tests/Test.hs index 653538faf..f0827f5fe 100644 --- a/tests/Test.hs +++ b/tests/Test.hs @@ -43,9 +43,11 @@ import AgentTests.SchemaDump (schemaDumpTest) #if defined(dbServerPostgres) import NtfServerTests (ntfServerTests) -import NtfClient (ntfTestServerDBConnectInfo) -import SMPClient (testServerDBConnectInfo) -import ServerTests.SchemaDump +import NtfClient (ntfTestServerDBConnectInfo, ntfTestStoreDBOpts) +import PostgresSchemaDump (postgresSchemaDumpTest) +import SMPClient (testServerDBConnectInfo, testStoreDBOpts) +import Simplex.Messaging.Notifications.Server.Store.Migrations (ntfServerMigrations) +import Simplex.Messaging.Server.QueueStore.Postgres.Migrations (serverMigrations) #endif #if defined(dbPostgres) || defined(dbServerPostgres) @@ -58,7 +60,7 @@ logCfg = LogConfig {lc_file = Nothing, lc_stderr = True} main :: IO () main = do -- TODO [ntfdb] running wiht LogWarn level shows potential issue "Queue count differs" - setLogLevel LogError -- LogInfo + setLogLevel LogError -- LogInfo -- also change in SMPClient.hs in defaultStartOptions withGlobalLogging logCfg $ do setEnv "APNS_KEY_ID" "H82WD9K9AQ" setEnv "APNS_KEY_FILE" "./tests/fixtures/AuthKey_H82WD9K9AQ.p8" @@ -93,7 +95,13 @@ main = do describe "Agent core tests" agentCoreTests #if defined(dbServerPostgres) around_ (postgressBracket testServerDBConnectInfo) $ - describe "Server schema dump" serverSchemaDumpTest + describe "SMP server schema dump" $ + postgresSchemaDumpTest + serverMigrations + [ "20250320_short_links" -- snd_secure moves to the bottom on down migration + ] -- skipComparisonForDownMigrations + testStoreDBOpts + "src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql" aroundAll_ (postgressBracket testServerDBConnectInfo) $ describe "SMP server via TLS, postgres+jornal message store" $ before (pure (transport @TLS, ASType SQSPostgres SMSJournal)) serverTests @@ -107,6 +115,13 @@ main = do -- describe "SMP syntax" $ serverSyntaxTests (transport @WS) -- before (pure (transport @WS, ASType SQSMemory SMSJournal)) serverTests #if defined(dbServerPostgres) + around_ (postgressBracket ntfTestServerDBConnectInfo) $ + describe "Ntf server schema dump" $ + postgresSchemaDumpTest + ntfServerMigrations + [] -- skipComparisonForDownMigrations + ntfTestStoreDBOpts + "src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql" aroundAll_ (postgressBracket ntfTestServerDBConnectInfo) $ do describe "Notifications server" $ ntfServerTests (transport @TLS) aroundAll_ (postgressBracket testServerDBConnectInfo) $ do