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

Author: epoberezkin

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

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

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,22 +293,22 @@ 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
           let ch = FileInfo {sndKey, size = chunkSize, digest}
           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 ()

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

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

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)

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,26 +22,48 @@ 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)
 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

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 ()