dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-10-15 03:59:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

ntf server: priority db pool for notifications and client commands, concurrent resubscriptions, CLI options (#1529)

Browse Source 
* 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
This commit is contained in:
Evgeny
2025-04-30 11:40:32 +01:00
committed by GitHub
parent 7d0115daec
commit 0a3d014f5d
26 changed files with 532 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/ntf-server/Main.hs
-1
View File
@@ -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
+1 -1
View File
@@ -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
+7 -7
View File
@@ -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 ()
src/Simplex/FileTransfer/Server.hs
+9 -9
View File
@@ -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
+2 -2
View File
@@ -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
+17 -12
View File
@@ -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
+31 -7
View File
@@ -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
src/Simplex/Messaging/Client/Agent.hs
+2 -2
View File
@@ -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 ()
src/Simplex/Messaging/Notifications/Server.hs
+37 -35
View File
@@ -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
src/Simplex/Messaging/Notifications/Server/Env.hs
+1 -1
View File
@@ -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"
src/Simplex/Messaging/Notifications/Server/Main.hs
+21 -10
View File
@@ -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 <-
src/Simplex/Messaging/Notifications/Server/Push/APNS.hs
+12 -1
View File
@@ -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 ()
src/Simplex/Messaging/Notifications/Server/Store/Postgres.hs
+62 -27
View File
@@ -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)
src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql
+178
View File
@@ -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;
src/Simplex/Messaging/Server.hs
+22 -22
View File
@@ -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
src/Simplex/Messaging/Server/CLI.hs
+20 -1
View File
@@ -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) <-
src/Simplex/Messaging/Server/Env/STM.hs
+3 -2
View File
@@ -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)
src/Simplex/Messaging/Server/Main.hs
+5 -3
View File
@@ -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
src/Simplex/Messaging/Server/StoreLog.hs
+3 -3
View File
@@ -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
src/Simplex/Messaging/Transport/Server.hs
+1 -1
View File
@@ -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
src/Simplex/Messaging/Util.hs
+1
View File
@@ -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 =
tests/NtfClient.hs
+2 -3
View File
@@ -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
tests/PostgresSchemaDump.hs
+70
View File
@@ -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
tests/SMPClient.hs
+5 -1
View File
@@ -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
tests/ServerTests/SchemaDump.hs
-98
View File
@@ -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
tests/Test.hs
+20 -5
View File
@@ -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