Compare commits

..
Author SHA1 Message Date
Evgeny Poberezkin bd97cb0449 6.3.0.4 2025-02-10 15:12:54 +00:00
Evgeny a58d3540ad agent: refactor migrations (#1452)
* agent: refactor migrations

* fix postgres
2025-02-07 17:17:53 +00:00
Evgenyandspaced4ndy 5dbe6337ea ntf server: additional statistics, new invalid token reasons (#1451)
* agent: check ntf token status on registration

* remove check

* update on check

* refactor

* version

* fix

* ntf server: additional statistics

* swap

* version

* more stats

* test, verify invalid

* rename

* exclude test token from stats

* increase delay

* handle invalid token in retry, more reasons

* focus tests

* disable new tests in CI

* fix

---------

Co-authored-by: spaced4ndy <[email protected]>
2025-02-07 12:19:11 +00:00
spaced4ndyandEvgeny Poberezkin b633f89c1a agent: check ntf token status on registration (#1450)
* agent: check ntf token status on registration

* remove check

* update on check

* refactor

* version

* fix

* test, verify invalid

* rename

* increase delay

* disable new tests in CI

* fix

---------

Co-authored-by: Evgeny Poberezkin <[email protected]>
2025-02-07 11:36:29 +00:00
Evgeny 944a22a2fb ntf server: record token invalidation reason, add date of the last token activity (#1449)
* ntf server: record token invalidation reason, add date of the last token activity

* update time

* rename

* optional

* include token ID in delivery error

* version

* protocol version

* fix, log error
2025-02-06 17:11:35 +00:00
Evgeny ce24f83b64 refactor STM queues (#1447) 2025-02-05 12:04:27 +00:00
31 changed files with 853 additions and 554 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
cabal-version: 1.12
name: simplexmq
version: 6.3.0.3
version: 6.3.0.4
synopsis: SimpleXMQ message broker
description: This package includes <./docs/Simplex-Messaging-Server.html server>,
<./docs/Simplex-Messaging-Client.html client> and
+21 -14
View File
@@ -170,8 +170,7 @@ import Simplex.Messaging.Agent.Store
import Simplex.Messaging.Agent.Store.AgentStore
import Simplex.Messaging.Agent.Store.Common (DBStore)
import qualified Simplex.Messaging.Agent.Store.DB as DB
import Simplex.Messaging.Agent.Store.Interface (closeDBStore, execSQL)
import qualified Simplex.Messaging.Agent.Store.Migrations as Migrations
import Simplex.Messaging.Agent.Store.Interface (closeDBStore, execSQL, getCurrentMigrations)
import Simplex.Messaging.Agent.Store.Shared (UpMigration (..), upMigration)
import Simplex.Messaging.Client (SMPClientError, ServerTransmission (..), ServerTransmissionBatch, temporaryClientError, unexpectedResponse)
import qualified Simplex.Messaging.Crypto as C
@@ -1946,7 +1945,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
-- possible improvement: add minimal time before repeat registration
(Just tknId, Nothing)
| savedDeviceToken == suppliedDeviceToken ->
when (ntfTknStatus == NTRegistered) (registerToken tkn) $> NTRegistered
registerToken tkn $> NTRegistered
| otherwise -> replaceToken tknId
(Just tknId, Just (NTAVerify code))
| savedDeviceToken == suppliedDeviceToken ->
@@ -1955,14 +1954,16 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
(Just tknId, Just NTACheck)
| savedDeviceToken == suppliedDeviceToken -> do
ns <- asks ntfSupervisor
atomically $ nsUpdateToken ns tkn {ntfMode = suppliedNtfMode}
when (ntfTknStatus == NTActive) $ do
cron <- asks $ ntfCron . config
agentNtfEnableCron c tknId tkn cron
when (suppliedNtfMode == NMInstant) $ initializeNtfSubs c
when (suppliedNtfMode == NMPeriodic && savedNtfMode == NMInstant) $ deleteNtfSubs c NSCSmpDelete
-- possible improvement: get updated token status from the server, or maybe TCRON could return the current status
pure ntfTknStatus
let tkn' = tkn {ntfMode = suppliedNtfMode}
atomically $ nsUpdateToken ns tkn'
agentNtfCheckToken c tknId tkn' >>= \case
NTActive -> do
cron <- asks $ ntfCron . config
agentNtfEnableCron c tknId tkn cron
when (suppliedNtfMode == NMInstant) $ initializeNtfSubs c
when (suppliedNtfMode == NMPeriodic && savedNtfMode == NMInstant) $ deleteNtfSubs c NSCSmpDelete
t tkn' (NTActive, Just NTACheck) $ pure ()
status -> t tkn' (status, Nothing) $ pure ()
| otherwise -> replaceToken tknId
-- deprecated
(Just _tknId, Just NTADelete) -> deleteToken c tkn $> NTExpired
@@ -2029,9 +2030,15 @@ verifyNtfToken' c deviceToken nonce code =
checkNtfToken' :: AgentClient -> DeviceToken -> AM NtfTknStatus
checkNtfToken' c deviceToken =
withStore' c getSavedNtfToken >>= \case
Just tkn@NtfToken {deviceToken = savedDeviceToken, ntfTokenId = Just tknId} -> do
Just tkn@NtfToken {deviceToken = savedDeviceToken, ntfTokenId = Just tknId, ntfTknAction} -> do
when (deviceToken /= savedDeviceToken) . throwE $ CMD PROHIBITED "checkNtfToken: different token"
agentNtfCheckToken c tknId tkn
status <- agentNtfCheckToken c tknId tkn
let action = case status of
NTInvalid _ -> Nothing
NTExpired -> Nothing
_ -> ntfTknAction
withStore' c $ \db -> updateNtfToken db tkn status action
pure status
_ -> throwE $ CMD PROHIBITED "checkNtfToken: no token"
deleteNtfToken' :: AgentClient -> DeviceToken -> AM ()
@@ -2172,7 +2179,7 @@ execAgentStoreSQL :: AgentClient -> Text -> AE [Text]
execAgentStoreSQL c sql = withAgentEnv c $ withStore' c (`execSQL` sql)
getAgentMigrations :: AgentClient -> AE [UpMigration]
getAgentMigrations c = withAgentEnv c $ map upMigration <$> withStore' c Migrations.getCurrent
getAgentMigrations c = withAgentEnv c $ map upMigration <$> withStore' c getCurrentMigrations
debugAgentLocks :: AgentClient -> IO AgentLocks
debugAgentLocks AgentClient {connLocks = cs, invLocks = is, deleteLock = d} = do
+2 -3
View File
@@ -30,8 +30,7 @@ import Data.Type.Equality
import Simplex.Messaging.Agent.Protocol
import Simplex.Messaging.Agent.RetryInterval (RI2State)
import Simplex.Messaging.Agent.Store.Common
import Simplex.Messaging.Agent.Store.Interface (DBOpts, createDBStore)
import qualified Simplex.Messaging.Agent.Store.Migrations as Migrations
import Simplex.Messaging.Agent.Store.Interface (DBOpts, appMigrations, createDBStore)
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..), MigrationError (..))
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Crypto.Ratchet (PQEncryption, PQSupport, RatchetX448)
@@ -54,7 +53,7 @@ import Simplex.Messaging.Protocol
import qualified Simplex.Messaging.Protocol as SMP
createStore :: DBOpts -> MigrationConfirmation -> IO (Either MigrationError DBStore)
createStore dbOpts = createDBStore dbOpts Migrations.app
createStore dbOpts = createDBStore dbOpts appMigrations
-- * Queue types
+21 -38
View File
@@ -1,16 +1,11 @@
{-# LANGUAGE CPP #-}
{-# LANGUAGE LambdaCase #-}
module Simplex.Messaging.Agent.Store.Migrations
( Migration (..),
MigrationsToRun (..),
DownMigration (..),
Migrations.app,
Migrations.getCurrent,
get,
Migrations.initialize,
Migrations.run,
migrateSchema,
DBMigrate (..),
sharedMigrateSchema,
-- for tests
migrationsToRun,
toDownMigration,
@@ -21,19 +16,9 @@ import Control.Monad
import Data.Char (toLower)
import Data.Functor (($>))
import Data.Maybe (isNothing, mapMaybe)
import Simplex.Messaging.Agent.Store.Common
import Simplex.Messaging.Agent.Store.Shared
import System.Exit (exitFailure)
import System.IO (hFlush, stdout)
#if defined(dbPostgres)
import qualified Simplex.Messaging.Agent.Store.Postgres.Migrations as Migrations
#else
import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
import System.Directory (copyFile)
#endif
get :: DBStore -> [Migration] -> IO (Either MTRError MigrationsToRun)
get st migrations = migrationsToRun migrations <$> withTransaction st Migrations.getCurrent
migrationsToRun :: [Migration] -> [Migration] -> Either MTRError MigrationsToRun
migrationsToRun [] [] = Right MTRNone
@@ -48,44 +33,42 @@ migrationsToRun (a : as) (d : ds)
| name a == name d = migrationsToRun as ds
| otherwise = Left $ MTREDifferent (name a) (name d)
migrateSchema :: DBStore -> [Migration] -> MigrationConfirmation -> Bool -> IO (Either MigrationError ())
migrateSchema st migrations confirmMigrations vacuum = do
Migrations.initialize st
get st migrations >>= \case
data DBMigrate = DBMigrate
{ initialize :: IO (),
getCurrent :: IO [Migration],
run :: MigrationsToRun -> IO (),
backup :: IO ()
}
sharedMigrateSchema :: DBMigrate -> Bool -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError ())
sharedMigrateSchema dbm dbNew' migrations confirmMigrations = do
initialize dbm
currentMs <- getCurrent dbm
case migrationsToRun migrations currentMs of
Left e -> do
when (confirmMigrations == MCConsole) $ confirmOrExit ("Database state error: " <> mtrErrorDescription e)
pure . Left $ MigrationError e
Right MTRNone -> pure $ Right ()
Right ms@(MTRUp ums)
| dbNew st -> Migrations.run st vacuum ms $> Right ()
| dbNew' -> run dbm ms $> Right ()
| otherwise -> case confirmMigrations of
MCYesUp -> runWithBackup st vacuum ms
MCYesUpDown -> runWithBackup st vacuum ms
MCConsole -> confirm err >> runWithBackup st vacuum ms
MCYesUp -> runWithBackup ms
MCYesUpDown -> runWithBackup ms
MCConsole -> confirm err >> runWithBackup ms
MCError -> pure $ Left err
where
err = MEUpgrade $ map upMigration ums -- "The app has a newer version than the database.\nConfirm to back up and upgrade using these migrations: " <> intercalate ", " (map name ums)
Right ms@(MTRDown dms) -> case confirmMigrations of
MCYesUpDown -> runWithBackup st vacuum ms
MCConsole -> confirm err >> runWithBackup st vacuum ms
MCYesUpDown -> runWithBackup ms
MCConsole -> confirm err >> runWithBackup ms
MCYesUp -> pure $ Left err
MCError -> pure $ Left err
where
err = MEDowngrade $ map downName dms
where
runWithBackup ms = backup dbm >> run dbm ms $> Right ()
confirm err = confirmOrExit $ migrationErrorDescription err
runWithBackup :: DBStore -> Bool -> MigrationsToRun -> IO (Either a ())
#if defined(dbPostgres)
runWithBackup st vacuum ms = Migrations.run st vacuum ms $> Right ()
#else
runWithBackup st vacuum ms = do
let f = dbFilePath st
copyFile f (f <> ".bak")
Migrations.run st vacuum ms
pure $ Right ()
#endif
confirmOrExit :: String -> IO ()
confirmOrExit s = do
putStrLn s
+11 -2
View File
@@ -6,6 +6,8 @@
module Simplex.Messaging.Agent.Store.Postgres
( DBOpts (..),
Migrations.appMigrations,
Migrations.getCurrentMigrations,
createDBStore,
closeDBStore,
reopenDBStore,
@@ -22,7 +24,8 @@ import Data.Text (Text)
import Database.PostgreSQL.Simple (Only (..))
import qualified Database.PostgreSQL.Simple as PSQL
import Database.PostgreSQL.Simple.SqlQQ (sql)
import Simplex.Messaging.Agent.Store.Migrations (migrateSchema)
import Simplex.Messaging.Agent.Store.Migrations (DBMigrate (..), sharedMigrateSchema)
import qualified Simplex.Messaging.Agent.Store.Postgres.Migrations as Migrations
import Simplex.Messaging.Agent.Store.Postgres.Common
import qualified Simplex.Messaging.Agent.Store.Postgres.DB as DB
import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..))
@@ -43,10 +46,16 @@ data DBOpts = DBOpts
createDBStore :: DBOpts -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError DBStore)
createDBStore DBOpts {connstr, schema} migrations confirmMigrations = do
st <- connectPostgresStore connstr schema
r <- migrateSchema st migrations confirmMigrations True `onException` closeDBStore st
r <- migrateSchema st `onException` closeDBStore st
case r of
Right () -> pure $ Right st
Left e -> closeDBStore st $> Left e
where
migrateSchema st =
let initialize = Migrations.initialize st
getCurrent = withTransaction st Migrations.getCurrentMigrations
dbm = DBMigrate {initialize, getCurrent, run = Migrations.run st, backup = pure ()}
in sharedMigrateSchema dbm (dbNew st) migrations confirmMigrations
connectPostgresStore :: ByteString -> String -> IO DBStore
connectPostgresStore dbConnstr dbSchema = do
@@ -17,7 +17,7 @@ module Simplex.Messaging.Agent.Store.Postgres.DB
where
import Control.Monad (void)
import Data.Int (Int32, Int64)
import Data.Int (Int64)
import Data.Word (Word16, Word32)
import Database.PostgreSQL.Simple (ResultError (..))
import qualified Database.PostgreSQL.Simple as PSQL
@@ -5,10 +5,10 @@
{-# LANGUAGE TupleSections #-}
module Simplex.Messaging.Agent.Store.Postgres.Migrations
( app,
( appMigrations,
initialize,
run,
getCurrent,
getCurrentMigrations,
)
where
@@ -34,8 +34,8 @@ schemaMigrations =
]
-- | The list of migrations in ascending order by date
app :: [Migration]
app = sortOn name $ map migration schemaMigrations
appMigrations :: [Migration]
appMigrations = sortOn name $ map migration schemaMigrations
where
migration (name, up, down) = Migration {name, up, down = down}
@@ -53,8 +53,8 @@ initialize st = withTransaction' st $ \db ->
)
|]
run :: DBStore -> Bool -> MigrationsToRun -> IO ()
run st _vacuum = \case
run :: DBStore -> MigrationsToRun -> IO ()
run st = \case
MTRUp [] -> pure ()
MTRUp ms -> mapM_ runUp ms
MTRDown ms -> mapM_ runDown $ reverse ms
@@ -72,7 +72,7 @@ run st _vacuum = \case
withMVar (connectionHandle db) $ \pqConn ->
void $ LibPQ.exec pqConn (TE.encodeUtf8 query)
getCurrent :: PSQL.Connection -> IO [Migration]
getCurrent db = map toMigration <$> PSQL.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;"
getCurrentMigrations :: PSQL.Connection -> IO [Migration]
getCurrentMigrations db = map toMigration <$> PSQL.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;"
where
toMigration (name, down) = Migration {name, up = T.pack "", down}
+14 -3
View File
@@ -26,6 +26,8 @@
module Simplex.Messaging.Agent.Store.SQLite
( DBOpts (..),
Migrations.appMigrations,
Migrations.getCurrentMigrations,
createDBStore,
closeDBStore,
reopenDBStore,
@@ -52,12 +54,13 @@ import Database.SQLite.Simple (Query (..))
import qualified Database.SQLite.Simple as SQL
import Database.SQLite.Simple.QQ (sql)
import qualified Database.SQLite3 as SQLite3
import Simplex.Messaging.Agent.Store.Migrations (migrateSchema)
import Simplex.Messaging.Agent.Store.Migrations (DBMigrate (..), sharedMigrateSchema)
import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
import Simplex.Messaging.Agent.Store.SQLite.Common
import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..))
import Simplex.Messaging.Util (ifM, safeDecodeUtf8)
import System.Directory (createDirectoryIfMissing, doesFileExist)
import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
import System.FilePath (takeDirectory)
import UnliftIO.Exception (bracketOnError, onException)
import UnliftIO.MVar
@@ -78,10 +81,18 @@ createDBStore DBOpts {dbFilePath, dbKey, keepKey, track, vacuum} migrations conf
let dbDir = takeDirectory dbFilePath
createDirectoryIfMissing True dbDir
st <- connectSQLiteStore dbFilePath dbKey keepKey track
r <- migrateSchema st migrations confirmMigrations vacuum `onException` closeDBStore st
r <- migrateSchema st `onException` closeDBStore st
case r of
Right () -> pure $ Right st
Left e -> closeDBStore st $> Left e
where
migrateSchema st =
let initialize = Migrations.initialize st
getCurrent = withTransaction st Migrations.getCurrentMigrations
run = Migrations.run st vacuum
backup = copyFile dbFilePath (dbFilePath <> ".bak")
dbm = DBMigrate {initialize, getCurrent, run, backup}
in sharedMigrateSchema dbm (dbNew st) migrations confirmMigrations
connectSQLiteStore :: FilePath -> ScrubbedBytes -> Bool -> DB.TrackQueries -> IO DBStore
connectSQLiteStore dbFilePath key keepKey track = do
@@ -8,10 +8,10 @@
{-# LANGUAGE TupleSections #-}
module Simplex.Messaging.Agent.Store.SQLite.Migrations
( app,
( appMigrations,
initialize,
run,
getCurrent,
getCurrentMigrations,
)
where
@@ -112,13 +112,13 @@ schemaMigrations =
]
-- | The list of migrations in ascending order by date
app :: [Migration]
app = sortOn name $ map migration schemaMigrations
appMigrations :: [Migration]
appMigrations = sortOn name $ map migration schemaMigrations
where
migration (name, up, down) = Migration {name, up = fromQuery up, down = fromQuery <$> down}
getCurrent :: DB.Connection -> IO [Migration]
getCurrent DB.Connection {DB.conn} = map toMigration <$> SQL.query_ conn "SELECT name, down FROM migrations ORDER BY name ASC;"
getCurrentMigrations :: DB.Connection -> IO [Migration]
getCurrentMigrations DB.Connection {DB.conn} = map toMigration <$> SQL.query_ conn "SELECT name, down FROM migrations ORDER BY name ASC;"
where
toMigration (name, down) = Migration {name, up = "", down}
@@ -11,7 +11,7 @@
module Simplex.Messaging.Notifications.Protocol where
import Control.Applicative ((<|>))
import Control.Applicative (optional, (<|>))
import Data.Aeson (FromJSON (..), ToJSON (..), (.:), (.=))
import qualified Data.Aeson as J
import qualified Data.Aeson.Encoding as JE
@@ -32,7 +32,7 @@ import Simplex.Messaging.Agent.Store.DB (FromField (..), ToField (..))
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Notifications.Transport (NTFVersion, ntfClientHandshake)
import Simplex.Messaging.Notifications.Transport (NTFVersion, invalidReasonNTFVersion, ntfClientHandshake)
import Simplex.Messaging.Parsers (fromTextField_)
import Simplex.Messaging.Protocol hiding (Command (..), CommandTag (..))
import Simplex.Messaging.Util (eitherToMaybe, (<$?>))
@@ -296,12 +296,18 @@ data NtfResponse
instance ProtocolEncoding NTFVersion ErrorType NtfResponse where
type Tag NtfResponse = NtfResponseTag
encodeProtocol _v = \case
encodeProtocol v = \case
NRTknId entId dhKey -> e (NRTknId_, ' ', entId, dhKey)
NRSubId entId -> e (NRSubId_, ' ', entId)
NROk -> e NROk_
NRErr err -> e (NRErr_, ' ', err)
NRTkn stat -> e (NRTkn_, ' ', stat)
NRTkn stat -> e (NRTkn_, ' ', stat')
where
stat'
| v >= invalidReasonNTFVersion = stat
| otherwise = case stat of
NTInvalid _ -> NTInvalid Nothing
_ -> stat
NRSub stat -> e (NRSub_, ' ', stat)
NRPong -> e NRPong_
where
@@ -520,7 +526,7 @@ data NtfTknStatus
| -- | state after registration (TNEW)
NTRegistered
| -- | if initial notification failed (push provider error) or verification failed
NTInvalid
NTInvalid (Maybe NTInvalidReason)
| -- | Token confirmed via notification (accepted by push provider or verification code received by client)
NTConfirmed
| -- | after successful verification (TVFY)
@@ -533,7 +539,7 @@ instance Encoding NtfTknStatus where
smpEncode = \case
NTNew -> "NEW"
NTRegistered -> "REGISTERED"
NTInvalid -> "INVALID"
NTInvalid r_ -> "INVALID" <> maybe "" (\r -> ',' `B.cons` strEncode r) r_
NTConfirmed -> "CONFIRMED"
NTActive -> "ACTIVE"
NTExpired -> "EXPIRED"
@@ -541,12 +547,33 @@ instance Encoding NtfTknStatus where
A.takeTill (== ' ') >>= \case
"NEW" -> pure NTNew
"REGISTERED" -> pure NTRegistered
"INVALID" -> pure NTInvalid
"INVALID" -> NTInvalid <$> optional (A.char ',' *> strP)
"CONFIRMED" -> pure NTConfirmed
"ACTIVE" -> pure NTActive
"EXPIRED" -> pure NTExpired
_ -> fail "bad NtfTknStatus"
instance StrEncoding NTInvalidReason where
strEncode = smpEncode
strP = smpP
data NTInvalidReason = NTIRBadToken | NTIRTokenNotForTopic | NTIRExpiredToken | NTIRUnregistered
deriving (Eq, Show)
instance Encoding NTInvalidReason where
smpEncode = \case
NTIRBadToken -> "BAD"
NTIRTokenNotForTopic -> "TOPIC"
NTIRExpiredToken -> "EXPIRED"
NTIRUnregistered -> "UNREGISTERED"
smpP =
A.takeTill (== ' ') >>= \case
"BAD" -> pure NTIRBadToken
"TOPIC" -> pure NTIRTokenNotForTopic
"EXPIRED" -> pure NTIRExpiredToken
"UNREGISTERED" -> pure NTIRUnregistered
_ -> fail "bad NTInvalidReason"
instance StrEncoding NtfTknStatus where
strEncode = smpEncode
strP = smpP
+91 -43
View File
@@ -58,6 +58,7 @@ import Simplex.Messaging.Protocol (EntityId (..), ErrorType (..), ProtocolServer
import qualified Simplex.Messaging.Protocol as SMP
import Simplex.Messaging.Server
import Simplex.Messaging.Server.Control (CPClientRole (..))
import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, getSystemDate)
import Simplex.Messaging.Server.Stats (PeriodStats (..), PeriodStatCounts (..), periodStatCounts, updatePeriodStats)
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
@@ -135,7 +136,8 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
logInfo $ "server stats log enabled: " <> T.pack statsFilePath
liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
NtfServerStats {fromTime, tknCreated, tknVerified, tknDeleted, subCreated, subDeleted, ntfReceived, ntfDelivered, activeTokens, activeSubs} <- asks serverStats
NtfServerStats {fromTime, tknCreated, tknVerified, tknDeleted, tknReplaced, subCreated, subDeleted, ntfReceived, ntfDelivered, ntfFailed, ntfCronDelivered, ntfCronFailed, ntfVrfQueued, ntfVrfDelivered, ntfVrfFailed, ntfVrfInvalidTkn, activeTokens, activeSubs} <-
asks serverStats
let interval = 1000000 * logInterval
forever $ do
withFile statsFilePath AppendMode $ \h -> liftIO $ do
@@ -145,10 +147,18 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
tknCreated' <- atomicSwapIORef tknCreated 0
tknVerified' <- atomicSwapIORef tknVerified 0
tknDeleted' <- atomicSwapIORef tknDeleted 0
tknReplaced' <- atomicSwapIORef tknReplaced 0
subCreated' <- atomicSwapIORef subCreated 0
subDeleted' <- atomicSwapIORef subDeleted 0
ntfReceived' <- atomicSwapIORef ntfReceived 0
ntfDelivered' <- atomicSwapIORef ntfDelivered 0
ntfFailed' <- atomicSwapIORef ntfFailed 0
ntfCronDelivered' <- atomicSwapIORef ntfCronDelivered 0
ntfCronFailed' <- atomicSwapIORef ntfCronFailed 0
ntfVrfQueued' <- atomicSwapIORef ntfVrfQueued 0
ntfVrfDelivered' <- atomicSwapIORef ntfVrfDelivered 0
ntfVrfFailed' <- atomicSwapIORef ntfVrfFailed 0
ntfVrfInvalidTkn' <- atomicSwapIORef ntfVrfInvalidTkn 0
tkn <- liftIO $ periodStatCounts activeTokens ts
sub <- liftIO $ periodStatCounts activeSubs ts
hPutStrLn h $
@@ -167,7 +177,15 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
monthCount tkn,
dayCount sub,
weekCount sub,
monthCount sub
monthCount sub,
show tknReplaced',
show ntfFailed',
show ntfCronDelivered',
show ntfCronFailed',
show ntfVrfQueued',
show ntfVrfDelivered',
show ntfVrfFailed',
show ntfVrfInvalidTkn'
]
liftIO $ threadDelay' interval
@@ -224,9 +242,18 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
putStat "tknCreated" tknCreated
putStat "tknVerified" tknVerified
putStat "tknDeleted" tknDeleted
putStat "tknReplaced" tknReplaced
putStat "subCreated" subCreated
putStat "subDeleted" subDeleted
putStat "ntfReceived" ntfReceived
putStat "ntfDelivered" ntfDelivered
putStat "ntfFailed" ntfFailed
putStat "ntfCronDelivered" ntfCronDelivered
putStat "ntfCronFailed" ntfCronFailed
putStat "ntfVrfQueued" ntfVrfQueued
putStat "ntfVrfDelivered" ntfVrfDelivered
putStat "ntfVrfFailed" ntfVrfFailed
putStat "ntfVrfInvalidTkn" ntfVrfInvalidTkn
getStat (day . activeTokens) >>= \v -> hPutStrLn h $ "daily active tokens: " <> show (IS.size v)
getStat (day . activeSubs) >>= \v -> hPutStrLn h $ "daily active subscriptions: " <> show (IS.size v)
CPStatsRTS -> tryAny getRTSStats >>= either (hPrint h) (hPrint h)
@@ -241,15 +268,19 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
#else
hPutStrLn h "Threads: not available on GHC 8.10"
#endif
NtfSubscriber {smpSubscribers, smpAgent = a} <- unliftIO u $ asks subscriber
NtfEnv {subscriber, pushServer} <- unliftIO u ask
let NtfSubscriber {smpSubscribers, smpAgent = a} = subscriber
NtfPushServer {pushQ} = pushServer
SMPClientAgent {smpClients, smpSessions, srvSubs, pendingSrvSubs, smpSubWorkers} = a
putSMPWorkers a "SMP subcscribers" smpSubscribers
let SMPClientAgent {smpClients, smpSessions, srvSubs, pendingSrvSubs, smpSubWorkers} = a
putSMPWorkers a "SMP clients" smpClients
putSMPWorkers a "SMP subscription workers" smpSubWorkers
sessions <- readTVarIO smpSessions
hPutStrLn h $ "SMP sessions count: " <> show (M.size sessions)
putSMPSubs a "SMP subscriptions" srvSubs
putSMPSubs a "Pending SMP subscriptions" pendingSrvSubs
sz <- atomically $ lengthTBQueue pushQ
hPutStrLn h $ "Push notifications queue length: " <> show sz
where
putSMPSubs :: SMPClientAgent -> String -> TMap SMPServer (TMap SMPSub a) -> IO ()
putSMPSubs a name v = do
@@ -431,35 +462,35 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
ntfPush :: NtfPushServer -> M ()
ntfPush s@NtfPushServer {pushQ} = forever $ do
(tkn@NtfTknData {ntfTknId, token = DeviceToken pp _, tknStatus}, ntf) <- atomically (readTBQueue pushQ)
(tkn@NtfTknData {ntfTknId, token = t@(DeviceToken pp _), tknStatus}, ntf) <- atomically (readTBQueue pushQ)
liftIO $ logDebug $ "sending push notification to " <> T.pack (show pp)
status <- readTVarIO tknStatus
case ntf of
PNVerification _
| status /= NTInvalid && status /= NTExpired ->
deliverNotification pp tkn ntf >>= \case
Right _ -> do
status_ <- atomically $ stateTVar tknStatus $ \case
NTActive -> (Nothing, NTActive)
NTConfirmed -> (Nothing, NTConfirmed)
_ -> (Just NTConfirmed, NTConfirmed)
forM_ status_ $ \status' -> withNtfLog $ \sl -> logTokenStatus sl ntfTknId status'
_ -> pure ()
| otherwise -> logError "bad notification token status"
PNVerification _ ->
deliverNotification pp tkn ntf >>= \case
Right _ -> do
status_ <- atomically $ stateTVar tknStatus $ \case
NTActive -> (Nothing, NTActive)
NTConfirmed -> (Nothing, NTConfirmed)
_ -> (Just NTConfirmed, NTConfirmed)
forM_ status_ $ \status' -> withNtfLog $ \sl -> logTokenStatus sl ntfTknId status'
incNtfStatT t ntfVrfDelivered
Left _ -> incNtfStatT t ntfVrfFailed
PNCheckMessages -> checkActiveTkn status $ do
void $ deliverNotification pp tkn ntf
deliverNotification pp tkn ntf
>>= incNtfStatT t . (\case Left _ -> ntfCronFailed; Right () -> ntfCronDelivered)
PNMessage {} -> checkActiveTkn status $ do
stats <- asks serverStats
liftIO $ updatePeriodStats (activeTokens stats) ntfTknId
void $ deliverNotification pp tkn ntf
incNtfStat ntfDelivered
deliverNotification pp tkn ntf
>>= incNtfStatT t . (\case Left _ -> ntfFailed; Right () -> ntfDelivered)
where
checkActiveTkn :: NtfTknStatus -> M () -> M ()
checkActiveTkn status action
| status == NTActive = action
| otherwise = liftIO $ logError "bad notification token status"
deliverNotification :: PushProvider -> NtfTknData -> PushNotification -> M (Either PushProviderError ())
deliverNotification pp tkn ntf = do
deliverNotification pp tkn@NtfTknData {ntfTknId} ntf = do
deliver <- liftIO $ getPushClient s pp
liftIO (runExceptT $ deliver tkn ntf) >>= \case
Right _ -> pure $ Right ()
@@ -467,15 +498,19 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do
PPConnection _ -> retryDeliver
PPRetryLater -> retryDeliver
PPCryptoError _ -> err e
PPResponseError _ _ -> err e
PPTokenInvalid -> updateTknStatus tkn NTInvalid >> err e
PPResponseError {} -> err e
PPTokenInvalid r -> updateTknStatus tkn (NTInvalid $ Just r) >> err e
PPPermanentError -> err e
where
retryDeliver :: M (Either PushProviderError ())
retryDeliver = do
deliver <- liftIO $ newPushClient s pp
liftIO (runExceptT $ deliver tkn ntf) >>= either err (pure . Right)
err e = logError (T.pack $ "Push provider error (" <> show pp <> "): " <> show e) $> Left e
liftIO (runExceptT $ deliver tkn ntf) >>= \case
Right _ -> pure $ Right ()
Left e -> case e of
PPTokenInvalid r -> updateTknStatus tkn (NTInvalid $ Just r) >> err e
_ -> err e
err e = logError ("Push provider error (" <> tshow pp <> ", " <> tshow ntfTknId <> "): " <> tshow e) $> Left e
updateTknStatus :: NtfTknData -> NtfTknStatus -> M ()
updateTknStatus NtfTknData {ntfTknId, tknStatus} status = do
@@ -509,13 +544,17 @@ receive th@THandle {params = THandleParams {thAuth}} NtfServerClient {rcvQ, sndQ
where
cmdAction t@(_, _, (corrId, entId, cmdOrError)) =
case cmdOrError of
Left e -> pure $ Left (corrId, entId, NRErr e)
Left e -> do
logError $ "invalid client request: " <> tshow e
pure $ Left (corrId, entId, NRErr e)
Right cmd ->
verified <$> verifyNtfTransmission ((,C.cbNonce (SMP.bs corrId)) <$> thAuth) t cmd
verified =<< verifyNtfTransmission ((,C.cbNonce (SMP.bs corrId)) <$> thAuth) t cmd
where
verified = \case
VRVerified req -> Right req
VRFailed -> Left (corrId, entId, NRErr AUTH)
VRVerified req -> pure $ Right req
VRFailed -> do
logError "unauthorized client request"
pure $ Left (corrId, entId, NRErr AUTH)
write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
send :: Transport c => THandleNTF c 'TServer -> NtfServerClient -> IO ()
@@ -524,7 +563,7 @@ send h@THandle {params} NtfServerClient {sndQ, sndActiveAt} = forever $ do
void . liftIO $ tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
atomically . (writeTVar sndActiveAt $!) =<< liftIO getSystemTime
data VerificationResult = VRVerified NtfRequest | VRFailed
data VerificationResult = VRVerified (Maybe NtfTknData, NtfRequest) | VRFailed
verifyNtfTransmission :: Maybe (THandleAuth 'TServer, C.CbNonce) -> SignedTransmission ErrorType NtfCmd -> NtfCmd -> M VerificationResult
verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
@@ -538,34 +577,34 @@ verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
Just t@NtfTknData {tknVerifyKey}
| k == tknVerifyKey -> verifiedTknCmd t c
| otherwise -> VRFailed
_ -> VRVerified (NtfReqNew corrId (ANE SToken tkn))
Nothing -> VRVerified (Nothing, NtfReqNew corrId (ANE SToken tkn))
else VRFailed
NtfCmd SToken c -> do
t_ <- atomically $ getNtfToken st entId
t_ <- liftIO $ getNtfTokenIO st entId
verifyToken t_ (`verifiedTknCmd` c)
NtfCmd SSubscription c@(SNEW sub@(NewNtfSub tknId smpQueue _)) -> do
s_ <- atomically $ findNtfSubscription st smpQueue
case s_ of
Nothing -> do
t_ <- atomically $ getActiveNtfToken st tknId
verifyToken' t_ $ VRVerified (NtfReqNew corrId (ANE SSubscription sub))
verifyToken' t_ $ VRVerified (t_, NtfReqNew corrId (ANE SSubscription sub))
Just s@NtfSubData {tokenId = subTknId} ->
if subTknId == tknId
then do
t_ <- atomically $ getActiveNtfToken st subTknId
verifyToken' t_ $ verifiedSubCmd s c
verifyToken' t_ $ verifiedSubCmd t_ s c
else pure $ maybe False (dummyVerifyCmd auth_ authorized) tAuth `seq` VRFailed
NtfCmd SSubscription PING -> pure $ VRVerified $ NtfReqPing corrId entId
NtfCmd SSubscription PING -> pure $ VRVerified (Nothing, NtfReqPing corrId entId)
NtfCmd SSubscription c -> do
s_ <- atomically $ getNtfSubscription st entId
s_ <- liftIO $ getNtfSubscriptionIO st entId
case s_ of
Just s@NtfSubData {tokenId = subTknId} -> do
t_ <- atomically $ getActiveNtfToken st subTknId
verifyToken' t_ $ verifiedSubCmd s c
verifyToken' t_ $ verifiedSubCmd t_ s c
_ -> pure $ maybe False (dummyVerifyCmd auth_ authorized) tAuth `seq` VRFailed
where
verifiedTknCmd t c = VRVerified (NtfReqCmd SToken (NtfTkn t) (corrId, entId, c))
verifiedSubCmd s c = VRVerified (NtfReqCmd SSubscription (NtfSub s) (corrId, entId, c))
verifiedTknCmd t c = VRVerified (Just t, NtfReqCmd SToken (NtfTkn t) (corrId, entId, c))
verifiedSubCmd t_ s c = VRVerified (t_, NtfReqCmd SSubscription (NtfSub s) (corrId, entId, c))
verifyToken :: Maybe NtfTknData -> (NtfTknData -> VerificationResult) -> M VerificationResult
verifyToken t_ positiveVerificationResult =
pure $ case t_ of
@@ -579,11 +618,17 @@ verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
client :: NtfServerClient -> NtfSubscriber -> NtfPushServer -> M ()
client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPushServer {pushQ, intervalNotifiers} =
forever $
forever $ do
ts <- liftIO getSystemDate
atomically (readTBQueue rcvQ)
>>= mapM processCommand
>>= mapM (\(tkn_, req) -> updateTokenDate ts tkn_ >> processCommand req)
>>= atomically . writeTBQueue sndQ
where
updateTokenDate :: RoundedSystemTime -> Maybe NtfTknData -> M ()
updateTokenDate ts' = mapM_ $ \NtfTknData {ntfTknId, tknUpdatedAt} -> do
let t' = Just ts'
t <- atomically $ swapTVar tknUpdatedAt t'
unless (t' == t) $ withNtfLog $ \s -> logUpdateTokenTime s ntfTknId ts'
processCommand :: NtfRequest -> M (Transmission NtfResponse)
processCommand = \case
NtfReqNew corrId (ANE SToken newTkn@(NewNtfTkn token _ dhPubKey)) -> do
@@ -593,9 +638,11 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
let dhSecret = C.dh' dhPubKey srvDhPrivKey
tknId <- getId
regCode <- getRegCode
tkn <- atomically $ mkNtfTknData tknId newTkn ks dhSecret regCode
ts <- liftIO $ getSystemDate
tkn <- liftIO $ mkNtfTknData tknId newTkn ks dhSecret regCode ts
atomically $ addNtfToken st tknId tkn
atomically $ writeTBQueue pushQ (tkn, PNVerification regCode)
incNtfStatT token ntfVrfQueued
withNtfLog (`logCreateToken` tkn)
incNtfStatT token tknCreated
pure (corrId, NoEntity, NRTknId tknId srvDhPubKey)
@@ -609,6 +656,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
if tknDhSecret == dhSecret
then do
atomically $ writeTBQueue pushQ (tkn, PNVerification tknRegCode)
incNtfStatT token ntfVrfQueued
pure $ NRTknId ntfTknId srvDhPubKey
else pure $ NRErr AUTH
TVFY code -- this allows repeated verification for cases when client connection dropped before server response
@@ -636,9 +684,9 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
let tkn' = tkn {token = token', tknRegCode = regCode}
addNtfToken st tknId tkn'
writeTBQueue pushQ (tkn', PNVerification regCode)
incNtfStatT token ntfVrfQueued
withNtfLog $ \s -> logUpdateToken s tknId token' regCode
incNtfStatT token tknDeleted
incNtfStatT token tknCreated
incNtfStatT token tknReplaced
pure NROk
TDEL -> do
logDebug "TDEL"
@@ -159,7 +159,7 @@ data NtfRequest
| NtfReqPing CorrId NtfEntityId
data NtfServerClient = NtfServerClient
{ rcvQ :: TBQueue (NonEmpty NtfRequest),
{ rcvQ :: TBQueue (NonEmpty (Maybe NtfTknData, NtfRequest)),
sndQ :: TBQueue (NonEmpty (Transmission NtfResponse)),
ntfThParams :: THandleParams NTFVersion 'TServer,
connected :: TVar Bool,
@@ -154,7 +154,7 @@ ntfServerCLI cfgPath logPath =
regCodeBytes = 32,
clientQSize = 64,
subQSize = 512,
pushQSize = 1048,
pushQSize = 16384,
smpAgentCfg =
defaultSMPClientAgentConfig
{ smpCfg =
@@ -308,7 +308,7 @@ data PushProviderError
= PPConnection HTTP2ClientError
| PPCryptoError C.CryptoError
| PPResponseError (Maybe Status) Text
| PPTokenInvalid
| PPTokenInvalid NTInvalidReason
| PPRetryLater
| PPPermanentError
deriving (Show, Exception)
@@ -337,19 +337,20 @@ apnsPushProviderClient c@APNSPushClient {nonceDrg, apnsCfg} tkn@NtfTknData {toke
result status reason'
| status == Just N.ok200 = pure ()
| status == Just N.badRequest400 =
case reason' of
"BadDeviceToken" -> throwE PPTokenInvalid
"DeviceTokenNotForTopic" -> throwE PPTokenInvalid
"TopicDisallowed" -> throwE PPPermanentError
_ -> err status reason'
| status == Just N.forbidden403 = case reason' of
"ExpiredProviderToken" -> throwE PPPermanentError -- there should be no point retrying it as the token was refreshed
"InvalidProviderToken" -> throwE PPPermanentError
_ -> err status reason'
| status == Just N.gone410 = throwE PPTokenInvalid
throwE $ case reason' of
"BadDeviceToken" -> PPTokenInvalid NTIRBadToken
"DeviceTokenNotForTopic" -> PPTokenInvalid NTIRTokenNotForTopic
"TopicDisallowed" -> PPPermanentError
_ -> PPResponseError status reason'
| status == Just N.forbidden403 = throwE $ case reason' of
"ExpiredProviderToken" -> PPPermanentError -- there should be no point retrying it as the token was refreshed
"InvalidProviderToken" -> PPPermanentError
_ -> PPResponseError status reason'
| status == Just N.gone410 = throwE $ case reason' of
"ExpiredToken" -> PPTokenInvalid NTIRExpiredToken
"Unregistered" -> PPTokenInvalid NTIRUnregistered
_ -> PPRetryLater
| status == Just N.serviceUnavailable503 = liftIO (disconnectApnsHTTP2Client c) >> throwE PPRetryLater
-- Just tooManyRequests429 -> TooManyRequests - too many requests for the same token
| otherwise = err status reason'
err :: Maybe Status -> Text -> ExceptT PushProviderError IO ()
err s r = throwE $ PPResponseError s r
| otherwise = throwE $ PPResponseError status reason'
liftHTTPS2 a = ExceptT $ first PPConnection <$> a
@@ -4,7 +4,7 @@
module Simplex.Messaging.Notifications.Server.Stats where
import Control.Applicative (optional)
import Control.Applicative (optional, (<|>))
import qualified Data.Attoparsec.ByteString.Char8 as A
import qualified Data.ByteString.Char8 as B
import Data.IORef
@@ -17,10 +17,18 @@ data NtfServerStats = NtfServerStats
tknCreated :: IORef Int,
tknVerified :: IORef Int,
tknDeleted :: IORef Int,
tknReplaced :: IORef Int,
subCreated :: IORef Int,
subDeleted :: IORef Int,
ntfReceived :: IORef Int,
ntfDelivered :: IORef Int,
ntfFailed :: IORef Int,
ntfCronDelivered :: IORef Int,
ntfCronFailed :: IORef Int,
ntfVrfQueued :: IORef Int,
ntfVrfDelivered :: IORef Int,
ntfVrfFailed :: IORef Int,
ntfVrfInvalidTkn :: IORef Int,
activeTokens :: PeriodStats,
activeSubs :: PeriodStats
}
@@ -30,10 +38,18 @@ data NtfServerStatsData = NtfServerStatsData
_tknCreated :: Int,
_tknVerified :: Int,
_tknDeleted :: Int,
_tknReplaced :: Int,
_subCreated :: Int,
_subDeleted :: Int,
_ntfReceived :: Int,
_ntfDelivered :: Int,
_ntfFailed :: Int,
_ntfCronDelivered :: Int,
_ntfCronFailed :: Int,
_ntfVrfQueued :: Int,
_ntfVrfDelivered :: Int,
_ntfVrfFailed :: Int,
_ntfVrfInvalidTkn :: Int,
_activeTokens :: PeriodStatsData,
_activeSubs :: PeriodStatsData
}
@@ -44,13 +60,41 @@ newNtfServerStats ts = do
tknCreated <- newIORef 0
tknVerified <- newIORef 0
tknDeleted <- newIORef 0
tknReplaced <- newIORef 0
subCreated <- newIORef 0
subDeleted <- newIORef 0
ntfReceived <- newIORef 0
ntfDelivered <- newIORef 0
ntfFailed <- newIORef 0
ntfCronDelivered <- newIORef 0
ntfCronFailed <- newIORef 0
ntfVrfQueued <- newIORef 0
ntfVrfDelivered <- newIORef 0
ntfVrfFailed <- newIORef 0
ntfVrfInvalidTkn <- newIORef 0
activeTokens <- newPeriodStats
activeSubs <- newPeriodStats
pure NtfServerStats {fromTime, tknCreated, tknVerified, tknDeleted, subCreated, subDeleted, ntfReceived, ntfDelivered, activeTokens, activeSubs}
pure
NtfServerStats
{ fromTime,
tknCreated,
tknVerified,
tknDeleted,
tknReplaced,
subCreated,
subDeleted,
ntfReceived,
ntfDelivered,
ntfFailed,
ntfCronDelivered,
ntfCronFailed,
ntfVrfQueued,
ntfVrfDelivered,
ntfVrfFailed,
ntfVrfInvalidTkn,
activeTokens,
activeSubs
}
getNtfServerStatsData :: NtfServerStats -> IO NtfServerStatsData
getNtfServerStatsData s@NtfServerStats {fromTime} = do
@@ -58,13 +102,41 @@ getNtfServerStatsData s@NtfServerStats {fromTime} = do
_tknCreated <- readIORef $ tknCreated s
_tknVerified <- readIORef $ tknVerified s
_tknDeleted <- readIORef $ tknDeleted s
_tknReplaced <- readIORef $ tknReplaced s
_subCreated <- readIORef $ subCreated s
_subDeleted <- readIORef $ subDeleted s
_ntfReceived <- readIORef $ ntfReceived s
_ntfDelivered <- readIORef $ ntfDelivered s
_ntfFailed <- readIORef $ ntfFailed s
_ntfCronDelivered <- readIORef $ ntfCronDelivered s
_ntfCronFailed <- readIORef $ ntfCronFailed s
_ntfVrfQueued <- readIORef $ ntfVrfQueued s
_ntfVrfDelivered <- readIORef $ ntfVrfDelivered s
_ntfVrfFailed <- readIORef $ ntfVrfFailed s
_ntfVrfInvalidTkn <- readIORef $ ntfVrfInvalidTkn s
_activeTokens <- getPeriodStatsData $ activeTokens s
_activeSubs <- getPeriodStatsData $ activeSubs s
pure NtfServerStatsData {_fromTime, _tknCreated, _tknVerified, _tknDeleted, _subCreated, _subDeleted, _ntfReceived, _ntfDelivered, _activeTokens, _activeSubs}
pure
NtfServerStatsData
{ _fromTime,
_tknCreated,
_tknVerified,
_tknDeleted,
_tknReplaced,
_subCreated,
_subDeleted,
_ntfReceived,
_ntfDelivered,
_ntfFailed,
_ntfCronDelivered,
_ntfCronFailed,
_ntfVrfQueued,
_ntfVrfDelivered,
_ntfVrfFailed,
_ntfVrfInvalidTkn,
_activeTokens,
_activeSubs
}
-- this function is not thread safe, it is used on server start only
setNtfServerStats :: NtfServerStats -> NtfServerStatsData -> IO ()
@@ -73,24 +145,60 @@ setNtfServerStats s@NtfServerStats {fromTime} d@NtfServerStatsData {_fromTime} =
writeIORef (tknCreated s) $! _tknCreated d
writeIORef (tknVerified s) $! _tknVerified d
writeIORef (tknDeleted s) $! _tknDeleted d
writeIORef (tknReplaced s) $! _tknReplaced d
writeIORef (subCreated s) $! _subCreated d
writeIORef (subDeleted s) $! _subDeleted d
writeIORef (ntfReceived s) $! _ntfReceived d
writeIORef (ntfDelivered s) $! _ntfDelivered d
writeIORef (ntfFailed s) $! _ntfFailed d
writeIORef (ntfCronDelivered s) $! _ntfCronDelivered d
writeIORef (ntfCronFailed s) $! _ntfCronFailed d
writeIORef (ntfVrfQueued s) $! _ntfVrfQueued d
writeIORef (ntfVrfDelivered s) $! _ntfVrfDelivered d
writeIORef (ntfVrfFailed s) $! _ntfVrfFailed d
writeIORef (ntfVrfInvalidTkn s) $! _ntfVrfInvalidTkn d
setPeriodStats (activeTokens s) (_activeTokens d)
setPeriodStats (activeSubs s) (_activeSubs d)
instance StrEncoding NtfServerStatsData where
strEncode NtfServerStatsData {_fromTime, _tknCreated, _tknVerified, _tknDeleted, _subCreated, _subDeleted, _ntfReceived, _ntfDelivered, _activeTokens, _activeSubs} =
strEncode
NtfServerStatsData
{ _fromTime,
_tknCreated,
_tknVerified,
_tknDeleted,
_tknReplaced,
_subCreated,
_subDeleted,
_ntfReceived,
_ntfDelivered,
_ntfFailed,
_ntfCronDelivered,
_ntfCronFailed,
_ntfVrfQueued,
_ntfVrfDelivered,
_ntfVrfFailed,
_ntfVrfInvalidTkn,
_activeTokens,
_activeSubs
} =
B.unlines
[ "fromTime=" <> strEncode _fromTime,
"tknCreated=" <> strEncode _tknCreated,
"tknVerified=" <> strEncode _tknVerified,
"tknDeleted=" <> strEncode _tknDeleted,
"tknReplaced=" <> strEncode _tknReplaced,
"subCreated=" <> strEncode _subCreated,
"subDeleted=" <> strEncode _subDeleted,
"ntfReceived=" <> strEncode _ntfReceived,
"ntfDelivered=" <> strEncode _ntfDelivered,
"ntfFailed=" <> strEncode _ntfFailed,
"ntfCronDelivered=" <> strEncode _ntfCronDelivered,
"ntfCronFailed=" <> strEncode _ntfCronFailed,
"ntfVrfQueued=" <> strEncode _ntfVrfQueued,
"ntfVrfDelivered=" <> strEncode _ntfVrfDelivered,
"ntfVrfFailed=" <> strEncode _ntfVrfFailed,
"ntfVrfInvalidTkn=" <> strEncode _ntfVrfInvalidTkn,
"activeTokens:",
strEncode _activeTokens,
"activeSubs:",
@@ -101,12 +209,42 @@ instance StrEncoding NtfServerStatsData where
_tknCreated <- "tknCreated=" *> strP <* A.endOfLine
_tknVerified <- "tknVerified=" *> strP <* A.endOfLine
_tknDeleted <- "tknDeleted=" *> strP <* A.endOfLine
_tknReplaced <- opt "tknReplaced="
_subCreated <- "subCreated=" *> strP <* A.endOfLine
_subDeleted <- "subDeleted=" *> strP <* A.endOfLine
_ntfReceived <- "ntfReceived=" *> strP <* A.endOfLine
_ntfDelivered <- "ntfDelivered=" *> strP <* A.endOfLine
_ntfFailed <- opt "ntfFailed="
_ntfCronDelivered <- opt "ntfCronDelivered="
_ntfCronFailed <- opt "ntfCronFailed="
_ntfVrfQueued <- opt "ntfVrfQueued="
_ntfVrfDelivered <- opt "ntfVrfDelivered="
_ntfVrfFailed <- opt "ntfVrfFailed="
_ntfVrfInvalidTkn <- opt "ntfVrfInvalidTkn="
_ <- "activeTokens:" <* A.endOfLine
_activeTokens <- strP <* A.endOfLine
_ <- "activeSubs:" <* A.endOfLine
_activeSubs <- strP <* optional A.endOfLine
pure NtfServerStatsData {_fromTime, _tknCreated, _tknVerified, _tknDeleted, _subCreated, _subDeleted, _ntfReceived, _ntfDelivered, _activeTokens, _activeSubs}
pure
NtfServerStatsData
{ _fromTime,
_tknCreated,
_tknVerified,
_tknDeleted,
_tknReplaced,
_subCreated,
_subDeleted,
_ntfReceived,
_ntfDelivered,
_ntfFailed,
_ntfCronDelivered,
_ntfCronFailed,
_ntfVrfQueued,
_ntfVrfDelivered,
_ntfVrfFailed,
_ntfVrfInvalidTkn,
_activeTokens,
_activeSubs
}
where
opt s = A.string s *> strP <* A.endOfLine <|> pure 0
@@ -25,6 +25,7 @@ import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Notifications.Protocol
import Simplex.Messaging.Protocol (NtfPrivateAuthKey, NtfPublicAuthKey, SMPServer)
import Simplex.Messaging.Server.QueueStore (RoundedSystemTime)
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Util (whenM, ($>>=))
@@ -57,14 +58,16 @@ data NtfTknData = NtfTknData
tknDhKeys :: C.KeyPair 'C.X25519,
tknDhSecret :: C.DhSecretX25519,
tknRegCode :: NtfRegCode,
tknCronInterval :: TVar Word16
tknCronInterval :: TVar Word16,
tknUpdatedAt :: TVar (Maybe RoundedSystemTime)
}
mkNtfTknData :: NtfTokenId -> NewNtfEntity 'Token -> C.KeyPair 'C.X25519 -> C.DhSecretX25519 -> NtfRegCode -> STM NtfTknData
mkNtfTknData ntfTknId (NewNtfTkn token tknVerifyKey _) tknDhKeys tknDhSecret tknRegCode = do
tknStatus <- newTVar NTRegistered
tknCronInterval <- newTVar 0
pure NtfTknData {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval}
mkNtfTknData :: NtfTokenId -> NewNtfEntity 'Token -> C.KeyPair 'C.X25519 -> C.DhSecretX25519 -> NtfRegCode -> RoundedSystemTime -> IO NtfTknData
mkNtfTknData ntfTknId (NewNtfTkn token tknVerifyKey _) tknDhKeys tknDhSecret tknRegCode ts = do
tknStatus <- newTVarIO NTRegistered
tknCronInterval <- newTVarIO 0
tknUpdatedAt <- newTVarIO $ Just ts
pure NtfTknData {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval, tknUpdatedAt}
data NtfSubData = NtfSubData
{ ntfSubId :: NtfSubscriptionId,
@@ -156,9 +159,8 @@ deleteTokenSubs st tknId = do
$>>= \NtfSubData {smpQueue} ->
TM.delete smpQueue (subscriptionLookup st) $> Just smpQueue
getNtfSubscription :: NtfStore -> NtfSubscriptionId -> STM (Maybe NtfSubData)
getNtfSubscription st subId =
TM.lookup subId (subscriptions st)
getNtfSubscriptionIO :: NtfStore -> NtfSubscriptionId -> IO (Maybe NtfSubData)
getNtfSubscriptionIO st subId = TM.lookupIO subId (subscriptions st)
findNtfSubscription :: NtfStore -> SMPQueueNtf -> STM (Maybe NtfSubData)
findNtfSubscription st smpQueue = do
@@ -16,6 +16,7 @@ module Simplex.Messaging.Notifications.Server.StoreLog
logUpdateToken,
logTokenCron,
logDeleteToken,
logUpdateTokenTime,
logCreateSubscription,
logSubscriptionStatus,
logDeleteSubscription,
@@ -23,6 +24,7 @@ module Simplex.Messaging.Notifications.Server.StoreLog
)
where
import Control.Applicative (optional)
import Control.Concurrent.STM
import Control.Logger.Simple
import Control.Monad
@@ -36,6 +38,7 @@ import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Notifications.Protocol
import Simplex.Messaging.Notifications.Server.Store
import Simplex.Messaging.Protocol (NtfPrivateAuthKey)
import Simplex.Messaging.Server.QueueStore (RoundedSystemTime)
import Simplex.Messaging.Server.StoreLog
import Simplex.Messaging.Util (safeDecodeUtf8)
import System.IO
@@ -46,6 +49,7 @@ data NtfStoreLogRecord
| UpdateToken NtfTokenId DeviceToken NtfRegCode
| TokenCron NtfTokenId Word16
| DeleteToken NtfTokenId
| UpdateTokenTime NtfTokenId RoundedSystemTime
| CreateSubscription NtfSubRec
| SubscriptionStatus NtfSubscriptionId NtfSubStatus
| DeleteSubscription NtfSubscriptionId
@@ -59,21 +63,24 @@ data NtfTknRec = NtfTknRec
tknDhKeys :: C.KeyPair 'C.X25519,
tknDhSecret :: C.DhSecretX25519,
tknRegCode :: NtfRegCode,
tknCronInterval :: Word16
tknCronInterval :: Word16,
tknUpdatedAt :: Maybe RoundedSystemTime
}
deriving (Show)
mkTknData :: NtfTknRec -> STM NtfTknData
mkTknData NtfTknRec {ntfTknId, token, tknStatus = status, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval = cronInt} = do
tknStatus <- newTVar status
tknCronInterval <- newTVar cronInt
pure NtfTknData {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval}
mkTknData :: NtfTknRec -> IO NtfTknData
mkTknData NtfTknRec {ntfTknId, token, tknStatus = status, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval = cronInt, tknUpdatedAt = updatedAt} = do
tknStatus <- newTVarIO status
tknCronInterval <- newTVarIO cronInt
tknUpdatedAt <- newTVarIO updatedAt
pure NtfTknData {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval, tknUpdatedAt}
mkTknRec :: NtfTknData -> STM NtfTknRec
mkTknRec NtfTknData {ntfTknId, token, tknStatus = status, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval = cronInt} = do
tknStatus <- readTVar status
tknCronInterval <- readTVar cronInt
pure NtfTknRec {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval}
mkTknRec :: NtfTknData -> IO NtfTknRec
mkTknRec NtfTknData {ntfTknId, token, tknStatus = status, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval = cronInt, tknUpdatedAt = updatedAt} = do
tknStatus <- readTVarIO status
tknCronInterval <- readTVarIO cronInt
tknUpdatedAt <- readTVarIO updatedAt
pure NtfTknRec {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval, tknUpdatedAt}
data NtfSubRec = NtfSubRec
{ ntfSubId :: NtfSubscriptionId,
@@ -84,9 +91,9 @@ data NtfSubRec = NtfSubRec
}
deriving (Show)
mkSubData :: NtfSubRec -> STM NtfSubData
mkSubData :: NtfSubRec -> IO NtfSubData
mkSubData NtfSubRec {ntfSubId, smpQueue, notifierKey, tokenId, subStatus = status} = do
subStatus <- newTVar status
subStatus <- newTVarIO status
pure NtfSubData {ntfSubId, smpQueue, notifierKey, tokenId, subStatus}
mkSubRec :: NtfSubData -> STM NtfSubRec
@@ -101,6 +108,7 @@ instance StrEncoding NtfStoreLogRecord where
UpdateToken tknId token regCode -> strEncode (Str "TUPDATE", tknId, token, regCode)
TokenCron tknId cronInt -> strEncode (Str "TCRON", tknId, cronInt)
DeleteToken tknId -> strEncode (Str "TDELETE", tknId)
UpdateTokenTime tknId ts -> strEncode (Str "TTIME", tknId, ts)
CreateSubscription subRec -> strEncode (Str "SCREATE", subRec)
SubscriptionStatus subId subStatus -> strEncode (Str "SSTATUS", subId, subStatus)
DeleteSubscription subId -> strEncode (Str "SDELETE", subId)
@@ -111,13 +119,14 @@ instance StrEncoding NtfStoreLogRecord where
"TUPDATE " *> (UpdateToken <$> strP_ <*> strP_ <*> strP),
"TCRON " *> (TokenCron <$> strP_ <*> strP),
"TDELETE " *> (DeleteToken <$> strP),
"TTIME " *> (UpdateTokenTime <$> strP_ <*> strP),
"SCREATE " *> (CreateSubscription <$> strP),
"SSTATUS " *> (SubscriptionStatus <$> strP_ <*> strP),
"SDELETE " *> (DeleteSubscription <$> strP)
]
instance StrEncoding NtfTknRec where
strEncode NtfTknRec {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval} =
strEncode NtfTknRec {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval, tknUpdatedAt} =
B.unwords
[ "tknId=" <> strEncode ntfTknId,
"token=" <> strEncode token,
@@ -128,6 +137,9 @@ instance StrEncoding NtfTknRec where
"regCode=" <> strEncode tknRegCode,
"cron=" <> strEncode tknCronInterval
]
<> maybe "" updatedAtStr tknUpdatedAt
where
updatedAtStr t = " updatedAt=" <> strEncode t
strP = do
ntfTknId <- "tknId=" *> strP_
token <- "token=" *> strP_
@@ -137,7 +149,8 @@ instance StrEncoding NtfTknRec where
tknDhSecret <- "dhSecret=" *> strP_
tknRegCode <- "regCode=" *> strP_
tknCronInterval <- "cron=" *> strP
pure NtfTknRec {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval}
tknUpdatedAt <- optional $ " updatedAt=" *> strP
pure NtfTknRec {ntfTknId, token, tknStatus, tknVerifyKey, tknDhKeys, tknDhSecret, tknRegCode, tknCronInterval, tknUpdatedAt}
instance StrEncoding NtfSubRec where
strEncode NtfSubRec {ntfSubId, smpQueue, notifierKey, tokenId, subStatus} =
@@ -161,7 +174,7 @@ logNtfStoreRecord = writeStoreLogRecord
{-# INLINE logNtfStoreRecord #-}
logCreateToken :: StoreLog 'WriteMode -> NtfTknData -> IO ()
logCreateToken s tkn = logNtfStoreRecord s . CreateToken =<< atomically (mkTknRec tkn)
logCreateToken s tkn = logNtfStoreRecord s . CreateToken =<< mkTknRec tkn
logTokenStatus :: StoreLog 'WriteMode -> NtfTokenId -> NtfTknStatus -> IO ()
logTokenStatus s tknId tknStatus = logNtfStoreRecord s $ TokenStatus tknId tknStatus
@@ -175,6 +188,9 @@ logTokenCron s tknId cronInt = logNtfStoreRecord s $ TokenCron tknId cronInt
logDeleteToken :: StoreLog 'WriteMode -> NtfTokenId -> IO ()
logDeleteToken s tknId = logNtfStoreRecord s $ DeleteToken tknId
logUpdateTokenTime :: StoreLog 'WriteMode -> NtfTokenId -> RoundedSystemTime -> IO ()
logUpdateTokenTime s tknId t = logNtfStoreRecord s $ UpdateTokenTime tknId t
logCreateSubscription :: StoreLog 'WriteMode -> NtfSubData -> IO ()
logCreateSubscription s sub = logNtfStoreRecord s . CreateSubscription =<< atomically (mkSubRec sub)
@@ -192,36 +208,39 @@ readNtfStore f st = mapM_ (addNtfLogRecord . LB.toStrict) . LB.lines =<< LB.read
where
addNtfLogRecord s = case strDecode s of
Left e -> logError $ "Log parsing error (" <> T.pack e <> "): " <> safeDecodeUtf8 (B.take 100 s)
Right lr -> atomically $ case lr of
Right lr -> case lr of
CreateToken r@NtfTknRec {ntfTknId} -> do
tkn <- mkTknData r
addNtfToken st ntfTknId tkn
atomically $ addNtfToken st ntfTknId tkn
TokenStatus tknId status -> do
tkn_ <- getNtfToken st tknId
tkn_ <- getNtfTokenIO st tknId
forM_ tkn_ $ \tkn@NtfTknData {tknStatus} -> do
writeTVar tknStatus status
when (status == NTActive) $ void $ removeInactiveTokenRegistrations st tkn
UpdateToken tknId token' tknRegCode ->
getNtfToken st tknId
atomically $ writeTVar tknStatus status
when (status == NTActive) $ void $ atomically $ removeInactiveTokenRegistrations st tkn
UpdateToken tknId token' tknRegCode -> do
getNtfTokenIO st tknId
>>= mapM_
( \tkn@NtfTknData {tknStatus} -> do
removeTokenRegistration st tkn
writeTVar tknStatus NTRegistered
addNtfToken st tknId tkn {token = token', tknRegCode}
atomically $ removeTokenRegistration st tkn
atomically $ writeTVar tknStatus NTRegistered
atomically $ addNtfToken st tknId tkn {token = token', tknRegCode}
)
TokenCron tknId cronInt ->
getNtfToken st tknId
>>= mapM_ (\NtfTknData {tknCronInterval} -> writeTVar tknCronInterval cronInt)
getNtfTokenIO st tknId
>>= mapM_ (\NtfTknData {tknCronInterval} -> atomically $ writeTVar tknCronInterval cronInt)
DeleteToken tknId ->
void $ deleteNtfToken st tknId
atomically $ void $ deleteNtfToken st tknId
UpdateTokenTime tknId t ->
getNtfTokenIO st tknId
>>= mapM_ (\NtfTknData {tknUpdatedAt} -> atomically $ writeTVar tknUpdatedAt $ Just t)
CreateSubscription r@NtfSubRec {ntfSubId} -> do
sub <- mkSubData r
void $ addNtfSubscription st ntfSubId sub
SubscriptionStatus subId status ->
getNtfSubscription st subId
>>= mapM_ (\NtfSubData {subStatus} -> writeTVar subStatus status)
void $ atomically $ addNtfSubscription st ntfSubId sub
SubscriptionStatus subId status -> do
getNtfSubscriptionIO st subId
>>= mapM_ (\NtfSubData {subStatus} -> atomically $ writeTVar subStatus status)
DeleteSubscription subId ->
deleteNtfSubscription st subId
atomically $ deleteNtfSubscription st subId
writeNtfStore :: StoreLog 'WriteMode -> NtfStore -> IO ()
writeNtfStore s NtfStore {tokens, subscriptions} = do
@@ -44,11 +44,14 @@ initialNTFVersion = VersionNTF 1
authBatchCmdsNTFVersion :: VersionNTF
authBatchCmdsNTFVersion = VersionNTF 2
invalidReasonNTFVersion :: VersionNTF
invalidReasonNTFVersion = VersionNTF 3
currentClientNTFVersion :: VersionNTF
currentClientNTFVersion = VersionNTF 2
currentClientNTFVersion = VersionNTF 3
currentServerNTFVersion :: VersionNTF
currentServerNTFVersion = VersionNTF 2
currentServerNTFVersion = VersionNTF 3
supportedClientNTFVRange :: VersionRangeNTF
supportedClientNTFVRange = mkVersionRange initialNTFVersion currentClientNTFVersion
+49 -50
View File
@@ -397,8 +397,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
atomicModifyIORef'_ (msgExpired stats) (+ expired)
printMessageStats "STORE: messages" msgStats
where
expireQueueMsgs now ms old rId q = fmap (fromRight newMessageStats) . runExceptT $ do
(expired_, stored) <- idleDeleteExpiredMsgs now ms rId q old
expireQueueMsgs now ms old q = fmap (fromRight newMessageStats) . runExceptT $ do
(expired_, stored) <- idleDeleteExpiredMsgs now ms q old
pure MessageStats {storedMsgsCount = stored, expiredMsgsCount = fromMaybe 0 expired_, storedQueues = 1}
expireNtfsThread :: ServerConfig -> M ()
@@ -429,8 +429,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
ss@ServerStats {fromTime, qCreated, qSecured, qDeletedAll, qDeletedAllB, qDeletedNew, qDeletedSecured, qSub, qSubAllB, qSubAuth, qSubDuplicate, qSubProhibited, qSubEnd, qSubEndB, ntfCreated, ntfDeleted, ntfDeletedB, ntfSub, ntfSubB, ntfSubAuth, ntfSubDuplicate, msgSent, msgSentAuth, msgSentQuota, msgSentLarge, msgRecv, msgRecvGet, msgGet, msgGetNoMsg, msgGetAuth, msgGetDuplicate, msgGetProhibited, msgExpired, activeQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount, ntfCount, pRelays, pRelaysOwn, pMsgFwds, pMsgFwdsOwn, pMsgFwdsRecv}
<- asks serverStats
AMS _ st <- asks msgStore
let queues = activeMsgQueues st
notifiers = notifiers' st
let STMQueueStore {queues, notifiers} = stmQueueStore st
interval = 1000000 * logInterval
forever $ do
withFile statsFilePath AppendMode $ \h -> liftIO $ do
@@ -581,13 +580,14 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
rtm <- getRealTimeMetrics env
T.writeFile metricsFile $ prometheusMetrics sm rtm ts
getServerMetrics :: STMQueueStore s => s -> ServerStats -> IO ServerMetrics
getServerMetrics :: STMStoreClass s => s -> ServerStats -> IO ServerMetrics
getServerMetrics st ss = do
d <- getServerStatsData ss
let ps = periodStatDataCounts $ _activeQueues d
psNtf = periodStatDataCounts $ _activeQueuesNtf d
queueCount <- M.size <$> readTVarIO (activeMsgQueues st)
notifierCount <- M.size <$> readTVarIO (notifiers' st)
STMQueueStore {queues, notifiers} = stmQueueStore st
queueCount <- M.size <$> readTVarIO queues
notifierCount <- M.size <$> readTVarIO notifiers
pure ServerMetrics {statsData = d, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount}
getRealTimeMetrics :: Env -> IO RealTimeMetrics
@@ -675,8 +675,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
CPStats -> withUserRole $ do
ss <- unliftIO u $ asks serverStats
AMS _ st <- unliftIO u $ asks msgStore
let queues = activeMsgQueues st
notifiers = notifiers' st
let STMQueueStore {queues, notifiers} = stmQueueStore st
getStat :: (ServerStats -> IORef a) -> IO a
getStat var = readIORef (var ss)
putStat :: Show a => String -> (ServerStats -> IORef a) -> IO ()
@@ -852,8 +851,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
CPDelete sId -> withUserRole $ unliftIO u $ do
AMS _ st <- asks msgStore
r <- liftIO $ runExceptT $ do
(q, qr) <- ExceptT $ getQueueRec st SSender sId
ExceptT $ deleteQueueSize st (recipientId qr) q
q <- ExceptT $ getQueue st SSender sId
ExceptT $ deleteQueueSize st q
case r of
Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
Right (qr, numDeleted) -> do
@@ -916,7 +915,7 @@ runClientTransport h@THandle {params = thParams@THandleParams {thVersion, sessio
c <- liftIO $ newClient msType clientId q thVersion sessionId ts
runClientThreads msType ms active c clientId `finally` clientDisconnected c
where
runClientThreads :: STMQueueStore (MsgStore s) => SMSType s -> MsgStore s -> TVar (IM.IntMap (Maybe AClient)) -> Client (MsgStore s) -> IS.Key -> M ()
runClientThreads :: STMStoreClass (MsgStore s) => SMSType s -> MsgStore s -> TVar (IM.IntMap (Maybe AClient)) -> Client (MsgStore s) -> IS.Key -> M ()
runClientThreads msType ms active c clientId = do
atomically $ modifyTVar' active $ IM.insert clientId $ Just (AClient msType c)
s <- asks server
@@ -972,7 +971,7 @@ cancelSub s = case subThread s of
_ -> pure ()
ProhibitSub -> pure ()
receive :: forall c s. (Transport c, STMQueueStore s) => THandleSMP c 'TServer -> s -> Client s -> M ()
receive :: forall c s. (Transport c, STMStoreClass s) => THandleSMP c 'TServer -> s -> Client s -> M ()
receive h@THandle {params = THandleParams {thAuth}} ms Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
sa <- asks serverActive
@@ -1072,7 +1071,7 @@ data VerificationResult s = VRVerified (Maybe (StoreQueue s, QueueRec)) | VRFail
-- - the queue or party key do not exist.
-- In all cases, the time of the verification should depend only on the provided authorization type,
-- a dummy key is used to run verification in the last two cases, and failure is returned irrespective of the result.
verifyTransmission :: forall s. STMQueueStore s => s -> Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> QueueId -> Cmd -> M (VerificationResult s)
verifyTransmission :: forall s. STMStoreClass s => s -> Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> QueueId -> Cmd -> M (VerificationResult s)
verifyTransmission ms auth_ tAuth authorized queueId cmd =
case cmd of
Cmd SRecipient (NEW k _ _ _ _) -> pure $ Nothing `verifiedWith` k
@@ -1149,7 +1148,7 @@ forkClient Client {endThreads, endThreadSeq} label action = do
action `finally` atomically (modifyTVar' endThreads $ IM.delete tId)
mkWeakThreadId t >>= atomically . modifyTVar' endThreads . IM.insert tId
client :: forall s. STMQueueStore s => THandleParams SMPVersion 'TServer -> Server -> s -> Client s -> M ()
client :: forall s. STMStoreClass s => THandleParams SMPVersion 'TServer -> Server -> s -> Client s -> M ()
client
thParams'
Server {subscribedQ, ntfSubscribedQ, subscribers}
@@ -1282,10 +1281,9 @@ client
updatedAt <- Just <$> liftIO getSystemDate
let rcvDhSecret = C.dh' dhKey privDhKey
qik (rcvId, sndId) = QIK {rcvId, sndId, rcvPublicDhKey, sndSecure}
qRec (recipientId, senderId) =
qRec senderId =
QueueRec
{ recipientId,
senderId,
{ senderId,
recipientKey,
rcvDhSecret,
senderKey = Nothing,
@@ -1297,12 +1295,12 @@ client
(corrId,entId,) <$> addQueueRetry 3 qik qRec
where
addQueueRetry ::
Int -> ((RecipientId, SenderId) -> QueueIdsKeys) -> ((RecipientId, SenderId) -> QueueRec) -> M BrokerMsg
Int -> ((RecipientId, SenderId) -> QueueIdsKeys) -> (SenderId -> QueueRec) -> M BrokerMsg
addQueueRetry 0 _ _ = pure $ ERR INTERNAL
addQueueRetry n qik qRec = do
ids <- getIds
let qr = qRec ids
liftIO (addQueue ms qr) >>= \case
ids@(rId, sId) <- getIds
let qr = qRec sId
liftIO (addQueue ms rId qr) >>= \case
Left DUPLICATE_ -> addQueueRetry (n - 1) qik qRec
Left e -> pure $ ERR e
Right q -> do
@@ -1379,7 +1377,7 @@ client
incStat $ qSubDuplicate stats
atomically (tryTakeTMVar $ delivered s) >> deliver False s
where
rId = recipientId qr
rId = recipientId' q
newSub :: M Sub
newSub = time "SUB newSub" . atomically $ do
writeTQueue subscribedQ (rId, clientId, True)
@@ -1390,7 +1388,7 @@ client
deliver inc sub = do
stats <- asks serverStats
fmap (either (\e -> (corrId, rId, ERR e)) id) $ liftIO $ runExceptT $ do
msg_ <- tryPeekMsg ms rId q
msg_ <- tryPeekMsg ms q
liftIO $ when (inc && isJust msg_) $ incStat (qSub stats)
liftIO $ deliverMessage "SUB" qr rId sub msg_
@@ -1424,7 +1422,7 @@ client
getMessage_ s delivered_ = do
stats <- asks serverStats
fmap (either err id) $ liftIO $ runExceptT $
tryPeekMsg ms (recipientId qr) q >>= \case
tryPeekMsg ms q >>= \case
Just msg -> do
let encMsg = encryptMsg qr msg
incStat $ (if isJust delivered_ then msgGetDuplicate else msgGet) stats
@@ -1471,11 +1469,11 @@ client
fmap (either err id) $ liftIO $ runExceptT $ do
case st of
ProhibitSub -> do
deletedMsg_ <- tryDelMsg ms (recipientId qr) q msgId
deletedMsg_ <- tryDelMsg ms q msgId
liftIO $ mapM_ (updateStats stats True) deletedMsg_
pure ok
_ -> do
(deletedMsg_, msg_) <- tryDelPeekMsg ms (recipientId qr) q msgId
(deletedMsg_, msg_) <- tryDelPeekMsg ms q msgId
liftIO $ mapM_ (updateStats stats False) deletedMsg_
liftIO $ deliverMessage "ACK" qr entId sub msg_
_ -> pure $ err NO_MSG
@@ -1529,7 +1527,7 @@ client
msg_ <- liftIO $ time "SEND" $ runExceptT $ do
expireMessages messageExpiration stats
msg <- liftIO $ mkMessage msgId body
writeMsg ms (recipientId qr) q True msg
writeMsg ms q True msg
case msg_ of
Left e -> pure $ err e
Right Nothing -> do
@@ -1540,10 +1538,10 @@ client
when (notification msgFlags) $ do
mapM_ (`enqueueNotification` msg) (notifier qr)
incStat $ msgSentNtf stats
liftIO $ updatePeriodStats (activeQueuesNtf stats) (recipientId qr)
liftIO $ updatePeriodStats (activeQueuesNtf stats) (recipientId' q)
incStat $ msgSent stats
incStat $ msgCount stats
liftIO $ updatePeriodStats (activeQueues stats) (recipientId qr)
liftIO $ updatePeriodStats (activeQueues stats) (recipientId' q)
pure ok
where
mkMessage :: MsgId -> C.MaxLenBS MaxMessageLen -> IO Message
@@ -1553,7 +1551,7 @@ client
expireMessages :: Maybe ExpirationConfig -> ServerStats -> ExceptT ErrorType IO ()
expireMessages msgExp stats = do
deleted <- maybe (pure 0) (deleteExpiredMsgs ms (recipientId qr) q <=< liftIO . expireBeforeEpoch) msgExp
deleted <- maybe (pure 0) (deleteExpiredMsgs ms q <=< liftIO . expireBeforeEpoch) msgExp
liftIO $ when (deleted > 0) $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
-- The condition for delivery of the message is:
@@ -1571,7 +1569,7 @@ client
whenM (TM.memberIO rId subscribers) $
atomically deliverToSub >>= mapM_ forkDeliver
where
rId = recipientId qr
rId = recipientId' q
deliverToSub =
-- lookup has ot be in the same transaction,
-- so that if subscription ends, it re-evalutates
@@ -1715,7 +1713,7 @@ client
delQueueAndMsgs :: (StoreQueue s, QueueRec) -> M (Transmission BrokerMsg)
delQueueAndMsgs (q, _) = do
liftIO (deleteQueue ms entId q) >>= \case
liftIO (deleteQueue ms q) >>= \case
Right qr -> do
-- Possibly, the same should be done if the queue is suspended, but currently we do not use it
atomically $ do
@@ -1735,11 +1733,11 @@ client
Left e -> pure $ err e
getQueueInfo :: StoreQueue s -> QueueRec -> M BrokerMsg
getQueueInfo q QueueRec {recipientId = rId, senderKey, notifier} = do
getQueueInfo q QueueRec {senderKey, notifier} = do
fmap (either ERR id) $ liftIO $ runExceptT $ do
qiSub <- liftIO $ TM.lookupIO entId subscriptions >>= mapM mkQSub
qiSize <- getQueueSize ms rId q
qiMsg <- toMsgInfo <$$> tryPeekMsg ms rId q
qiSize <- getQueueSize ms q
qiMsg <- toMsgInfo <$$> tryPeekMsg ms q
let info = QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg}
pure $ INFO info
where
@@ -1809,8 +1807,9 @@ exportMessages tty ms f drainMsgs = do
logError $ "error exporting messages: " <> tshow e
exitFailure
where
saveQueueMsgs h rId q =
runExceptT (getQueueMessages drainMsgs ms rId q) >>= \case
saveQueueMsgs h q = do
let rId = recipientId' q
runExceptT (getQueueMessages drainMsgs ms q) >>= \case
Right msgs -> Sum (length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs)
Left e -> do
logError $ "STORE: saveQueueMsgs, error exporting messages from queue " <> decodeLatin1 (strEncode rId) <> ", " <> tshow e
@@ -1838,7 +1837,7 @@ processServerMessages = do
withAllMsgQueues False ms $ processValidateQueue
| otherwise -> logWarn "skipping message expiration" $> Nothing
where
processExpireQueue old rId q =
processExpireQueue old q =
runExceptT expireQueue >>= \case
Right (storedMsgsCount, expiredMsgsCount) ->
pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1}
@@ -1847,20 +1846,20 @@ processServerMessages = do
exitFailure
where
expireQueue = do
expired'' <- deleteExpiredMsgs ms rId q old
stored'' <- getQueueSize ms rId q
expired'' <- deleteExpiredMsgs ms q old
stored'' <- getQueueSize ms q
liftIO $ closeMsgQueue q
pure (stored'', expired'')
processValidateQueue :: RecipientId -> JournalQueue -> IO MessageStats
processValidateQueue rId q =
runExceptT (getQueueSize ms rId q) >>= \case
processValidateQueue :: JournalQueue -> IO MessageStats
processValidateQueue q =
runExceptT (getQueueSize ms q) >>= \case
Right storedMsgsCount -> pure newMessageStats {storedMsgsCount, storedQueues = 1}
Left e -> do
logError $ "STORE: processValidateQueue, failed opening message queue, " <> tshow e
exitFailure
-- TODO this function should be called after importing queues from store log
importMessages :: forall s. STMQueueStore s => Bool -> s -> FilePath -> Maybe Int64 -> IO MessageStats
importMessages :: forall s. STMStoreClass s => Bool -> s -> FilePath -> Maybe Int64 -> IO MessageStats
importMessages tty ms f old_ = do
logInfo $ "restoring messages from file " <> T.pack f
LB.readFile f >>= runExceptT . foldM restoreMsg (0, Nothing, (0, 0, M.empty)) . LB.lines >>= \case
@@ -1873,7 +1872,7 @@ importMessages tty ms f old_ = do
renameFile f $ f <> ".bak"
mapM_ setOverQuota_ overQuota
logQueueStates ms
storedQueues <- M.size <$> readTVarIO (activeMsgQueues ms)
storedQueues <- M.size <$> readTVarIO (queues $ stmQueueStore ms)
pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
where
progress i = "Processed " <> show i <> " lines"
@@ -1892,7 +1891,7 @@ importMessages tty ms f old_ = do
(i + 1,Just (rId, q),) <$> case msg of
Message {msgTs}
| maybe True (systemSeconds msgTs >=) old_ -> do
writeMsg ms rId q False msg >>= \case
writeMsg ms q False msg >>= \case
Just _ -> pure (stored + 1, expired, overQuota)
Nothing -> do
logError $ decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (messageId msg)
@@ -1901,11 +1900,11 @@ importMessages tty ms f old_ = do
MessageQuota {} ->
-- queue was over quota at some point,
-- it will be set as over quota once fully imported
mergeQuotaMsgs >> writeMsg ms rId q False msg $> (stored, expired, M.insert rId q overQuota)
mergeQuotaMsgs >> writeMsg ms q False msg $> (stored, expired, M.insert rId q overQuota)
where
-- if the first message in queue head is "quota", remove it.
mergeQuotaMsgs =
withPeekMsgQueue ms rId q "mergeQuotaMsgs" $ maybe (pure ()) $ \case
withPeekMsgQueue ms q "mergeQuotaMsgs" $ maybe (pure ()) $ \case
(mq, MessageQuota {}) -> tryDeleteMsg_ q mq False
_ -> pure ()
msgErr :: Show e => String -> e -> String
@@ -1982,7 +1981,7 @@ restoreServerStats msgStats_ ntfStats = asks (serverStatsBackupFile . config) >>
Right d@ServerStatsData {_qCount = statsQCount, _msgCount = statsMsgCount, _ntfCount = statsNtfCount} -> do
s <- asks serverStats
AMS _ st <- asks msgStore
_qCount <- M.size <$> readTVarIO (activeMsgQueues st)
_qCount <- M.size <$> readTVarIO (queues $ stmQueueStore st)
let _msgCount = maybe statsMsgCount storedMsgsCount msgStats_
_ntfCount = storedMsgsCount ntfStats
_msgExpired' = _msgExpired d + maybe 0 expiredMsgsCount msgStats_
+2 -2
View File
@@ -189,7 +189,7 @@ type family MsgStore s where
MsgStore 'MSMemory = STMMsgStore
MsgStore 'MSJournal = JournalMsgStore
data AMsgStore = forall s. (STMQueueStore (MsgStore s), MsgStoreClass (MsgStore s)) => AMS (SMSType s) (MsgStore s)
data AMsgStore = forall s. (STMStoreClass (MsgStore s), MsgStoreClass (MsgStore s)) => AMS (SMSType s) (MsgStore s)
data AStoreQueue = forall s. MsgStoreClass (MsgStore s) => ASQ (SMSType s) (StoreQueue (MsgStore s))
@@ -362,5 +362,5 @@ newSMPProxyAgent smpAgentCfg random = do
smpAgent <- newSMPClientAgent smpAgentCfg random
pure ProxyAgent {smpAgent}
readWriteQueueStore :: STMQueueStore s => FilePath -> s -> IO (StoreLog 'WriteMode)
readWriteQueueStore :: STMStoreClass s => FilePath -> s -> IO (StoreLog 'WriteMode)
readWriteQueueStore = readWriteStoreLog readQueueStore writeQueueStore
@@ -15,7 +15,7 @@
{-# LANGUAGE TupleSections #-}
module Simplex.Messaging.Server.MsgStore.Journal
( JournalMsgStore (queues, senders, notifiers, random),
( JournalMsgStore (queueStore, random),
JournalQueue,
JournalMsgQueue (queue, state),
JMQueue (queueDirectory, statePath),
@@ -77,10 +77,7 @@ data JournalMsgStore = JournalMsgStore
{ config :: JournalStoreConfig,
random :: TVar StdGen,
queueLocks :: TMap RecipientId Lock,
queues :: TMap RecipientId JournalQueue,
senders :: TMap SenderId RecipientId,
notifiers :: TMap NotifierId RecipientId,
storeLog :: TVar (Maybe (StoreLog 'WriteMode))
queueStore :: STMQueueStore JournalQueue
}
data JournalStoreConfig = JournalStoreConfig
@@ -98,7 +95,8 @@ data JournalStoreConfig = JournalStoreConfig
}
data JournalQueue = JournalQueue
{ queueLock :: Lock,
{ recipientId :: RecipientId,
queueLock :: Lock,
-- To avoid race conditions and errors when restoring queues,
-- Nothing is written to TVar when queue is deleted.
queueRec :: TVar (Maybe QueueRec),
@@ -218,18 +216,15 @@ logFileExt = ".log"
newtype StoreIO a = StoreIO {unStoreIO :: IO a}
deriving newtype (Functor, Applicative, Monad)
instance STMQueueStore JournalMsgStore where
queues' = queues
senders' = senders
notifiers' = notifiers
storeLog' = storeLog
mkQueue st qr = do
lock <- getMapLock (queueLocks st) $ recipientId qr
instance STMStoreClass JournalMsgStore where
stmQueueStore JournalMsgStore {queueStore} = queueStore
mkQueue st rId qr = do
lock <- getMapLock (queueLocks st) rId
q <- newTVar $ Just qr
mq <- newTVar Nothing
activeAt <- newTVar 0
isEmpty <- newTVar Nothing
pure $ JournalQueue lock q mq activeAt isEmpty
pure $ JournalQueue rId lock q mq activeAt isEmpty
msgQueue_' = msgQueue_
instance MsgStoreClass JournalMsgStore where
@@ -242,27 +237,21 @@ instance MsgStoreClass JournalMsgStore where
newMsgStore config = do
random <- newTVarIO =<< newStdGen
queueLocks <- TM.emptyIO
queues <- TM.emptyIO
senders <- TM.emptyIO
notifiers <- TM.emptyIO
storeLog <- newTVarIO Nothing
pure JournalMsgStore {config, random, queueLocks, queues, senders, notifiers, storeLog}
queueStore <- newQueueStore
pure JournalMsgStore {config, random, queueLocks, queueStore}
setStoreLog :: JournalMsgStore -> StoreLog 'WriteMode -> IO ()
setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
setStoreLog st sl = atomically $ writeTVar (storeLog $ queueStore st) (Just sl)
closeMsgStore st = do
closeMsgStore JournalMsgStore {queueStore = st} = do
readTVarIO (storeLog st) >>= mapM_ closeStoreLog
readTVarIO (queues st) >>= mapM_ closeMsgQueue
activeMsgQueues = queues
{-# INLINE activeMsgQueues #-}
-- This function is a "foldr" that opens and closes all queues, processes them as defined by action and accumulates the result.
-- It is used to export storage to a single file and also to expire messages and validate all queues when server is started.
-- TODO this function requires case-sensitive file system, because it uses queue directory as recipient ID.
-- It can be made to support case-insensite FS by supporting more than one queue per directory, by getting recipient ID from state file name.
withAllMsgQueues :: forall a. Monoid a => Bool -> JournalMsgStore -> (RecipientId -> JournalQueue -> IO a) -> IO a
withAllMsgQueues :: forall a. Monoid a => Bool -> JournalMsgStore -> (JournalQueue -> IO a) -> IO a
withAllMsgQueues tty ms@JournalMsgStore {config} action = ifM (doesDirectoryExist storePath) processStore (pure mempty)
where
processStore = do
@@ -276,7 +265,7 @@ instance MsgStoreClass JournalMsgStore where
r' <- case strDecode $ B.pack queueId of
Right rId ->
getQueue ms SRecipient rId >>= \case
Right q -> unStoreIO (getMsgQueue ms rId q) *> action rId q <* closeMsgQueue q
Right q -> unStoreIO (getMsgQueue ms q) *> action q <* closeMsgQueue q
Left AUTH -> do
logWarn $ "STORE: processQueue, queue " <> T.pack queueId <> " was removed, removing " <> T.pack dir
removeQueueDirectory_ dir
@@ -303,7 +292,7 @@ instance MsgStoreClass JournalMsgStore where
(Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
logQueueStates :: JournalMsgStore -> IO ()
logQueueStates ms = withActiveMsgQueues ms $ \_ -> unStoreIO . logQueueState
logQueueStates ms = withActiveMsgQueues ms $ unStoreIO . logQueueState
logQueueState :: JournalQueue -> StoreIO ()
logQueueState q =
@@ -312,11 +301,14 @@ instance MsgStoreClass JournalMsgStore where
$>>= \mq -> readTVarIO (handles mq)
$>>= (\hs -> (readTVarIO (state mq) >>= appendState (stateHandle hs)) $> Just ())
recipientId' = recipientId
{-# INLINE recipientId' #-}
queueRec' = queueRec
{-# INLINE queueRec' #-}
getMsgQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> StoreIO JournalMsgQueue
getMsgQueue ms@JournalMsgStore {random} rId JournalQueue {msgQueue_} =
getMsgQueue :: JournalMsgStore -> JournalQueue -> StoreIO JournalMsgQueue
getMsgQueue ms@JournalMsgStore {random} JournalQueue {recipientId = rId, msgQueue_} =
StoreIO $ readTVarIO msgQueue_ >>= maybe newQ pure
where
newQ = do
@@ -334,8 +326,8 @@ instance MsgStoreClass JournalMsgStore where
journalId <- newJournalId random
mkJournalQueue queue (newMsgQueueState journalId) Nothing
getPeekMsgQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> StoreIO (Maybe (JournalMsgQueue, Message))
getPeekMsgQueue ms rId q@JournalQueue {isEmpty} =
getPeekMsgQueue :: JournalMsgStore -> JournalQueue -> StoreIO (Maybe (JournalMsgQueue, Message))
getPeekMsgQueue ms q@JournalQueue {isEmpty} =
StoreIO (readTVarIO isEmpty) >>= \case
Just True -> pure Nothing
Just False -> peek
@@ -350,16 +342,16 @@ instance MsgStoreClass JournalMsgStore where
pure r
where
peek = do
mq <- getMsgQueue ms rId q
mq <- getMsgQueue ms q
(mq,) <$$> tryPeekMsg_ q mq
-- only runs action if queue is not empty
withIdleMsgQueue :: Int64 -> JournalMsgStore -> RecipientId -> JournalQueue -> (JournalMsgQueue -> StoreIO a) -> StoreIO (Maybe a, Int)
withIdleMsgQueue now ms@JournalMsgStore {config} rId q action =
withIdleMsgQueue :: Int64 -> JournalMsgStore -> JournalQueue -> (JournalMsgQueue -> StoreIO a) -> StoreIO (Maybe a, Int)
withIdleMsgQueue now ms@JournalMsgStore {config} q action =
StoreIO $ readTVarIO (msgQueue_ q) >>= \case
Nothing ->
E.bracket
(unStoreIO $ getPeekMsgQueue ms rId q)
(unStoreIO $ getPeekMsgQueue ms q)
(mapM_ $ \_ -> closeMsgQueue q)
(maybe (pure (Nothing, 0)) (unStoreIO . run))
where
@@ -375,13 +367,12 @@ instance MsgStoreClass JournalMsgStore where
sz <- unStoreIO $ getQueueSize_ mq
pure (r, sz)
deleteQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType QueueRec)
deleteQueue ms rId q =
fst <$$> deleteQueue_ ms rId q
deleteQueue :: JournalMsgStore -> JournalQueue -> IO (Either ErrorType QueueRec)
deleteQueue ms q = fst <$$> deleteQueue_ ms q
deleteQueueSize :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType (QueueRec, Int))
deleteQueueSize ms rId q =
deleteQueue_ ms rId q >>= mapM (traverse getSize)
deleteQueueSize :: JournalMsgStore -> JournalQueue -> IO (Either ErrorType (QueueRec, Int))
deleteQueueSize ms q =
deleteQueue_ ms q >>= mapM (traverse getSize)
-- traverse operates on the second tuple element
where
getSize = maybe (pure (-1)) (fmap size . readTVarIO . state)
@@ -397,9 +388,9 @@ instance MsgStoreClass JournalMsgStore where
updateReadPos q drainMsgs len hs
(msg :) <$> run msgs
writeMsg :: JournalMsgStore -> RecipientId -> JournalQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
writeMsg ms rId q' logState msg = isolateQueue rId q' "writeMsg" $ do
q <- getMsgQueue ms rId q'
writeMsg :: JournalMsgStore -> JournalQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
writeMsg ms q' logState msg = isolateQueue q' "writeMsg" $ do
q <- getMsgQueue ms q'
StoreIO $ (`E.finally` updateActiveAt q') $ do
st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
let empty = size == 0
@@ -476,9 +467,9 @@ instance MsgStoreClass JournalMsgStore where
$>>= \len -> readTVarIO handles
$>>= \hs -> updateReadPos mq logState len hs $> Just ()
isolateQueue :: RecipientId -> JournalQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
isolateQueue rId JournalQueue {queueLock} op =
tryStore' op rId . withLock' queueLock op . unStoreIO
isolateQueue :: JournalQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
isolateQueue JournalQueue {recipientId, queueLock} op =
tryStore' op recipientId . withLock' queueLock op . unStoreIO
updateActiveAt :: JournalQueue -> IO ()
updateActiveAt q = atomically . writeTVar (activeAt q) . systemSeconds =<< getSystemTime
@@ -721,11 +712,12 @@ validQueueState MsgQueueState {readState = rs, writeState = ws, size}
&& msgPos ws == msgCount ws
&& bytePos ws == byteCount ws
deleteQueue_ :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType (QueueRec, Maybe JournalMsgQueue))
deleteQueue_ ms rId q =
deleteQueue_ :: JournalMsgStore -> JournalQueue -> IO (Either ErrorType (QueueRec, Maybe JournalMsgQueue))
deleteQueue_ ms q =
runExceptT $ isolateQueueId "deleteQueue_" ms rId $
deleteQueue' ms rId q >>= mapM remove
deleteQueue' ms q >>= mapM remove
where
rId = recipientId q
remove r@(_, mq_) = do
mapM_ closeMsgQueueHandles mq_
removeQueueDirectory ms rId
+29 -37
View File
@@ -26,22 +26,18 @@ import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.Server.QueueStore
import Simplex.Messaging.Server.QueueStore.STM
import Simplex.Messaging.Server.StoreLog
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Util ((<$$>), ($>>=))
import System.IO (IOMode (..))
data STMMsgStore = STMMsgStore
{ storeConfig :: STMStoreConfig,
queues :: TMap RecipientId STMQueue,
senders :: TMap SenderId RecipientId,
notifiers :: TMap NotifierId RecipientId,
storeLog :: TVar (Maybe (StoreLog 'WriteMode))
queueStore :: STMQueueStore STMQueue
}
data STMQueue = STMQueue
{ -- To avoid race conditions and errors when restoring queues,
-- Nothing is written to TVar when queue is deleted.
recipientId :: RecipientId,
queueRec :: TVar (Maybe QueueRec),
msgQueue_ :: TVar (Maybe STMMsgQueue)
}
@@ -57,12 +53,9 @@ data STMStoreConfig = STMStoreConfig
quota :: Int
}
instance STMQueueStore STMMsgStore where
queues' = queues
senders' = senders
notifiers' = notifiers
storeLog' = storeLog
mkQueue _ qr = STMQueue <$> newTVar (Just qr) <*> newTVar Nothing
instance STMStoreClass STMMsgStore where
stmQueueStore = queueStore
mkQueue _ rId qr = STMQueue rId <$> newTVar (Just qr) <*> newTVar Nothing
msgQueue_' = msgQueue_
instance MsgStoreClass STMMsgStore where
@@ -73,32 +66,31 @@ instance MsgStoreClass STMMsgStore where
newMsgStore :: STMStoreConfig -> IO STMMsgStore
newMsgStore storeConfig = do
queues <- TM.emptyIO
senders <- TM.emptyIO
notifiers <- TM.emptyIO
storeLog <- newTVarIO Nothing
pure STMMsgStore {storeConfig, queues, senders, notifiers, storeLog}
queueStore <- newQueueStore
pure STMMsgStore {storeConfig, queueStore}
setStoreLog :: STMMsgStore -> StoreLog 'WriteMode -> IO ()
setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
setStoreLog st sl = atomically $ writeTVar (storeLog $ queueStore st) (Just sl)
closeMsgStore st = readTVarIO (storeLog st) >>= mapM_ closeStoreLog
activeMsgQueues = queues
{-# INLINE activeMsgQueues #-}
closeMsgStore st = readTVarIO (storeLog $ queueStore st) >>= mapM_ closeStoreLog
withAllMsgQueues _ = withActiveMsgQueues
{-# INLINE withAllMsgQueues #-}
logQueueStates _ = pure ()
{-# INLINE logQueueStates #-}
logQueueState _ = pure ()
{-# INLINE logQueueState #-}
recipientId' = recipientId
{-# INLINE recipientId' #-}
queueRec' = queueRec
{-# INLINE queueRec' #-}
getMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM STMMsgQueue
getMsgQueue _ _ STMQueue {msgQueue_} = readTVar msgQueue_ >>= maybe newQ pure
getMsgQueue :: STMMsgStore -> STMQueue -> STM STMMsgQueue
getMsgQueue _ STMQueue {msgQueue_} = readTVar msgQueue_ >>= maybe newQ pure
where
newQ = do
msgQueue <- newTQueue
@@ -108,23 +100,23 @@ instance MsgStoreClass STMMsgStore where
writeTVar msgQueue_ (Just q)
pure q
getPeekMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM (Maybe (STMMsgQueue, Message))
getPeekMsgQueue _ _ q@STMQueue {msgQueue_} = readTVar msgQueue_ $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq
getPeekMsgQueue :: STMMsgStore -> STMQueue -> STM (Maybe (STMMsgQueue, Message))
getPeekMsgQueue _ q@STMQueue {msgQueue_} = readTVar msgQueue_ $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq
-- does not create queue if it does not exist, does not delete it if it does (can't just close in-memory queue)
withIdleMsgQueue :: Int64 -> STMMsgStore -> RecipientId -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a, Int)
withIdleMsgQueue _ _ _ STMQueue {msgQueue_} action = readTVar msgQueue_ >>= \case
withIdleMsgQueue :: Int64 -> STMMsgStore -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a, Int)
withIdleMsgQueue _ _ STMQueue {msgQueue_} action = readTVar msgQueue_ >>= \case
Just q -> do
r <- action q
sz <- getQueueSize_ q
pure (Just r, sz)
Nothing -> pure (Nothing, 0)
deleteQueue :: STMMsgStore -> RecipientId -> STMQueue -> IO (Either ErrorType QueueRec)
deleteQueue ms rId q = fst <$$> deleteQueue' ms rId q
deleteQueue :: STMMsgStore -> STMQueue -> IO (Either ErrorType QueueRec)
deleteQueue ms q = fst <$$> deleteQueue' ms q
deleteQueueSize :: STMMsgStore -> RecipientId -> STMQueue -> IO (Either ErrorType (QueueRec, Int))
deleteQueueSize ms rId q = deleteQueue' ms rId q >>= mapM (traverse getSize)
deleteQueueSize :: STMMsgStore -> STMQueue -> IO (Either ErrorType (QueueRec, Int))
deleteQueueSize ms q = deleteQueue' ms q >>= mapM (traverse getSize)
-- traverse operates on the second tuple element
where
getSize = maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size)
@@ -137,9 +129,9 @@ instance MsgStoreClass STMMsgStore where
mapM_ (writeTQueue q) msgs
pure msgs
writeMsg :: STMMsgStore -> RecipientId -> STMQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
writeMsg ms rId q' _logState msg = liftIO $ atomically $ do
STMMsgQueue {msgQueue = q, canWrite, size} <- getMsgQueue ms rId q'
writeMsg :: STMMsgStore -> STMQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
writeMsg ms q' _logState msg = liftIO $ atomically $ do
STMMsgQueue {msgQueue = q, canWrite, size} <- getMsgQueue ms q'
canWrt <- readTVar canWrite
empty <- isEmptyTQueue q
if canWrt || empty
@@ -171,5 +163,5 @@ instance MsgStoreClass STMMsgStore where
Just _ -> modifyTVar' size (subtract 1)
_ -> pure ()
isolateQueue :: RecipientId -> STMQueue -> String -> STM a -> ExceptT ErrorType IO a
isolateQueue _ _ _ = liftIO . atomically
isolateQueue :: STMQueue -> String -> STM a -> ExceptT ErrorType IO a
isolateQueue _ _ = liftIO . atomically
+44 -43
View File
@@ -20,7 +20,6 @@ import Control.Monad.Trans.Except
import Data.Functor (($>))
import Data.Int (Int64)
import Data.Kind
import qualified Data.Map.Strict as M
import Data.Time.Clock.System (SystemTime (systemSeconds))
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server.QueueStore
@@ -29,12 +28,16 @@ import Simplex.Messaging.TMap (TMap)
import Simplex.Messaging.Util ((<$$>))
import System.IO (IOMode (..))
class MsgStoreClass s => STMQueueStore s where
queues' :: s -> TMap RecipientId (StoreQueue s)
senders' :: s -> TMap SenderId RecipientId
notifiers' :: s -> TMap NotifierId RecipientId
storeLog' :: s -> TVar (Maybe (StoreLog 'WriteMode))
mkQueue :: s -> QueueRec -> STM (StoreQueue s)
data STMQueueStore q = STMQueueStore
{ queues :: TMap RecipientId q,
senders :: TMap SenderId RecipientId,
notifiers :: TMap NotifierId RecipientId,
storeLog :: TVar (Maybe (StoreLog 'WriteMode))
}
class MsgStoreClass s => STMStoreClass s where
stmQueueStore :: s -> STMQueueStore (StoreQueue s)
mkQueue :: s -> RecipientId -> QueueRec -> STM (StoreQueue s)
msgQueue_' :: StoreQueue s -> TVar (Maybe (MsgQueue s))
class Monad (StoreMonad s) => MsgStoreClass s where
@@ -45,25 +48,25 @@ class Monad (StoreMonad s) => MsgStoreClass s where
newMsgStore :: MsgStoreConfig s -> IO s
setStoreLog :: s -> StoreLog 'WriteMode -> IO ()
closeMsgStore :: s -> IO ()
activeMsgQueues :: s -> TMap RecipientId (StoreQueue s)
withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
withAllMsgQueues :: Monoid a => Bool -> s -> (StoreQueue s -> IO a) -> IO a
logQueueStates :: s -> IO ()
logQueueState :: StoreQueue s -> StoreMonad s ()
recipientId' :: StoreQueue s -> RecipientId
queueRec' :: StoreQueue s -> TVar (Maybe QueueRec)
getPeekMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message))
getMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (MsgQueue s)
getPeekMsgQueue :: s -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message))
getMsgQueue :: s -> StoreQueue s -> StoreMonad s (MsgQueue s)
-- the journal queue will be closed after action if it was initially closed or idle longer than interval in config
withIdleMsgQueue :: Int64 -> s -> RecipientId -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
deleteQueue :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType QueueRec)
deleteQueueSize :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int))
withIdleMsgQueue :: Int64 -> s -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
deleteQueue :: s -> StoreQueue s -> IO (Either ErrorType QueueRec)
deleteQueueSize :: s -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int))
getQueueMessages_ :: Bool -> MsgQueue s -> StoreMonad s [Message]
writeMsg :: s -> RecipientId -> StoreQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
writeMsg :: s -> StoreQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
setOverQuota_ :: StoreQueue s -> IO () -- can ONLY be used while restoring messages, not while server running
getQueueSize_ :: MsgQueue s -> StoreMonad s Int
tryPeekMsg_ :: StoreQueue s -> MsgQueue s -> StoreMonad s (Maybe Message)
tryDeleteMsg_ :: StoreQueue s -> MsgQueue s -> Bool -> StoreMonad s ()
isolateQueue :: RecipientId -> StoreQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
isolateQueue :: StoreQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
data MSType = MSMemory | MSJournal
@@ -73,28 +76,26 @@ data SMSType :: MSType -> Type where
data AMSType = forall s. AMSType (SMSType s)
withActiveMsgQueues :: (MsgStoreClass s, Monoid a) => s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
withActiveMsgQueues st f = readTVarIO (activeMsgQueues st) >>= foldM run mempty . M.assocs
withActiveMsgQueues :: (STMStoreClass s, Monoid a) => s -> (StoreQueue s -> IO a) -> IO a
withActiveMsgQueues st f = readTVarIO (queues $ stmQueueStore st) >>= foldM run mempty
where
run !acc (k, v) = do
r <- f k v
pure $! acc <> r
run !acc = fmap (acc <>) . f
getQueueMessages :: MsgStoreClass s => Bool -> s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO [Message]
getQueueMessages drainMsgs st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs . fst)
getQueueMessages :: MsgStoreClass s => Bool -> s -> StoreQueue s -> ExceptT ErrorType IO [Message]
getQueueMessages drainMsgs st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs . fst)
{-# INLINE getQueueMessages #-}
getQueueSize :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO Int
getQueueSize st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
getQueueSize :: MsgStoreClass s => s -> StoreQueue s -> ExceptT ErrorType IO Int
getQueueSize st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
{-# INLINE getQueueSize #-}
tryPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message)
tryPeekMsg st rId q = snd <$$> withPeekMsgQueue st rId q "tryPeekMsg" pure
tryPeekMsg :: MsgStoreClass s => s -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message)
tryPeekMsg st q = snd <$$> withPeekMsgQueue st q "tryPeekMsg" pure
{-# INLINE tryPeekMsg #-}
tryDelMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
tryDelMsg st rId q msgId' =
withPeekMsgQueue st rId q "tryDelMsg" $
tryDelMsg :: MsgStoreClass s => s -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
tryDelMsg st q msgId' =
withPeekMsgQueue st q "tryDelMsg" $
maybe (pure Nothing) $ \(mq, msg) ->
if
| messageId msg == msgId' ->
@@ -102,30 +103,30 @@ tryDelMsg st rId q msgId' =
| otherwise -> pure Nothing
-- atomic delete (== read) last and peek next message if available
tryDelPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
tryDelPeekMsg st rId q msgId' =
withPeekMsgQueue st rId q "tryDelPeekMsg" $
tryDelPeekMsg :: MsgStoreClass s => s -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
tryDelPeekMsg st q msgId' =
withPeekMsgQueue st q "tryDelPeekMsg" $
maybe (pure (Nothing, Nothing)) $ \(mq, msg) ->
if
| messageId msg == msgId' -> (Just msg,) <$> (tryDeleteMsg_ q mq True >> tryPeekMsg_ q mq)
| otherwise -> pure (Nothing, Just msg)
-- The action is called with Nothing when it is known that the queue is empty
withPeekMsgQueue :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> String -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a
withPeekMsgQueue st rId q op a = isolateQueue rId q op $ getPeekMsgQueue st rId q >>= a
withPeekMsgQueue :: MsgStoreClass s => s -> StoreQueue s -> String -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a
withPeekMsgQueue st q op a = isolateQueue q op $ getPeekMsgQueue st q >>= a
{-# INLINE withPeekMsgQueue #-}
deleteExpiredMsgs :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
deleteExpiredMsgs st rId q old =
isolateQueue rId q "deleteExpiredMsgs" $
getMsgQueue st rId q >>= deleteExpireMsgs_ old q
deleteExpiredMsgs :: MsgStoreClass s => s -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
deleteExpiredMsgs st q old =
isolateQueue q "deleteExpiredMsgs" $
getMsgQueue st q >>= deleteExpireMsgs_ old q
-- closed and idle queues will be closed after expiration
-- returns (expired count, queue size after expiration)
idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO (Maybe Int, Int)
idleDeleteExpiredMsgs now st rId q old =
isolateQueue rId q "idleDeleteExpiredMsgs" $
withIdleMsgQueue now st rId q (deleteExpireMsgs_ old q)
idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> StoreQueue s -> Int64 -> ExceptT ErrorType IO (Maybe Int, Int)
idleDeleteExpiredMsgs now st q old =
isolateQueue q "idleDeleteExpiredMsgs" $
withIdleMsgQueue now st q (deleteExpireMsgs_ old q)
deleteExpireMsgs_ :: MsgStoreClass s => Int64 -> StoreQueue s -> MsgQueue s -> StoreMonad s Int
deleteExpireMsgs_ old q mq = do
+1 -2
View File
@@ -16,8 +16,7 @@ import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol
data QueueRec = QueueRec
{ recipientId :: !RecipientId,
recipientKey :: !RcvPublicAuthKey,
{ recipientKey :: !RcvPublicAuthKey,
rcvDhSecret :: !RcvDhSecret,
senderId :: !SenderId,
senderKey :: !(Maybe SndPublicAuthKey),
+69 -60
View File
@@ -25,6 +25,7 @@ module Simplex.Messaging.Server.QueueStore.STM
unblockQueue,
updateQueueTime,
deleteQueue',
newQueueStore,
readQueueStore,
withLog',
)
@@ -51,99 +52,106 @@ import Simplex.Messaging.Util (ifM, tshow, ($>>=), (<$$))
import System.IO
import UnliftIO.STM
addQueue :: STMQueueStore s => s -> QueueRec -> IO (Either ErrorType (StoreQueue s))
addQueue st qr@QueueRec {recipientId = rId, senderId = sId, notifier}=
atomically add
$>>= \q -> q <$$ withLog "addQueue" st (`logCreateQueue` qr)
where
add = ifM hasId (pure $ Left DUPLICATE_) $ do
q <- mkQueue st qr -- STMQueue lock <$> (newTVar $! Just qr) <*> newTVar Nothing
TM.insert rId q $ queues' st
TM.insert sId rId $ senders' st
forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId $ notifiers' st
pure $ Right q
hasId = or <$> sequence [TM.member rId $ queues' st, TM.member sId $ senders' st, hasNotifier]
hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId (notifiers' st)) notifier
newQueueStore :: IO (STMQueueStore q)
newQueueStore = do
queues <- TM.emptyIO
senders <- TM.emptyIO
notifiers <- TM.emptyIO
storeLog <- newTVarIO Nothing
pure STMQueueStore {queues, senders, notifiers, storeLog}
getQueue :: (STMQueueStore s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s))
addQueue :: STMStoreClass s => s -> RecipientId -> QueueRec -> IO (Either ErrorType (StoreQueue s))
addQueue st rId qr@QueueRec {senderId = sId, notifier}=
atomically add
$>>= \q -> q <$$ withLog "addQueue" st (\s -> logCreateQueue s rId qr)
where
STMQueueStore {queues, senders, notifiers} = stmQueueStore st
add = ifM hasId (pure $ Left DUPLICATE_) $ do
q <- mkQueue st rId qr
TM.insert rId q queues
TM.insert sId rId senders
forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId notifiers
pure $ Right q
hasId = or <$> sequence [TM.member rId queues, TM.member sId senders, hasNotifier]
hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId notifiers) notifier
getQueue :: (STMStoreClass s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s))
getQueue st party qId =
maybe (Left AUTH) Right <$> case party of
SRecipient -> TM.lookupIO qId $ queues' st
SSender -> TM.lookupIO qId (senders' st) $>>= (`TM.lookupIO` queues' st)
SNotifier -> TM.lookupIO qId (notifiers' st) $>>= (`TM.lookupIO` queues' st)
SRecipient -> TM.lookupIO qId queues
SSender -> TM.lookupIO qId senders $>>= (`TM.lookupIO` queues)
SNotifier -> TM.lookupIO qId notifiers $>>= (`TM.lookupIO` queues)
where
STMQueueStore {queues, senders, notifiers} = stmQueueStore st
getQueueRec :: (STMQueueStore s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s, QueueRec))
getQueueRec :: (STMStoreClass s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s, QueueRec))
getQueueRec st party qId =
getQueue st party qId
$>>= (\q -> maybe (Left AUTH) (Right . (q,)) <$> readTVarIO (queueRec' q))
secureQueue :: STMQueueStore s => s -> StoreQueue s -> SndPublicAuthKey -> IO (Either ErrorType ())
secureQueue :: STMStoreClass s => s -> StoreQueue s -> SndPublicAuthKey -> IO (Either ErrorType ())
secureQueue st sq sKey =
atomically (readQueueRec qr $>>= secure)
$>>= \rId -> withLog "secureQueue" st $ \s -> logSecureQueue s rId sKey
$>>= \_ -> withLog "secureQueue" st $ \s -> logSecureQueue s (recipientId' sq) sKey
where
qr = queueRec' sq
secure q@QueueRec {recipientId = rId} = case senderKey q of
Just k -> pure $ if sKey == k then Right rId else Left AUTH
secure q = case senderKey q of
Just k -> pure $ if sKey == k then Right () else Left AUTH
Nothing -> do
writeTVar qr $ Just q {senderKey = Just sKey}
pure $ Right rId
pure $ Right ()
addQueueNotifier :: STMQueueStore s => s -> StoreQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
addQueueNotifier :: STMStoreClass s => s -> StoreQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId} =
atomically (readQueueRec qr $>>= add)
$>>= \(rId, nId_) -> nId_ <$$ withLog "addQueueNotifier" st (\s -> logAddNotifier s rId ntfCreds)
$>>= \nId_ -> nId_ <$$ withLog "addQueueNotifier" st (\s -> logAddNotifier s rId ntfCreds)
where
rId = recipientId' sq
qr = queueRec' sq
add q@QueueRec {recipientId = rId} = ifM (TM.member nId (notifiers' st)) (pure $ Left DUPLICATE_) $ do
nId_ <- forM (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId (notifiers' st) $> notifierId
STMQueueStore {notifiers} = stmQueueStore st
add q = ifM (TM.member nId notifiers) (pure $ Left DUPLICATE_) $ do
nId_ <- forM (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId notifiers $> notifierId
let !q' = q {notifier = Just ntfCreds}
writeTVar qr $ Just q'
TM.insert nId rId $ notifiers' st
pure $ Right (rId, nId_)
TM.insert nId rId notifiers
pure $ Right nId_
deleteQueueNotifier :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType (Maybe NotifierId))
deleteQueueNotifier :: STMStoreClass s => s -> StoreQueue s -> IO (Either ErrorType (Maybe NotifierId))
deleteQueueNotifier st sq =
atomically (readQueueRec qr >>= mapM delete)
$>>= \(rId, nId_) -> nId_ <$$ withLog "deleteQueueNotifier" st (`logDeleteNotifier` rId)
$>>= \nId_ -> nId_ <$$ withLog "deleteQueueNotifier" st (`logDeleteNotifier` recipientId' sq)
where
qr = queueRec' sq
delete q = fmap (recipientId q,) $ forM (notifier q) $ \NtfCreds {notifierId} -> do
TM.delete notifierId $ notifiers' st
delete q = forM (notifier q) $ \NtfCreds {notifierId} -> do
TM.delete notifierId $ notifiers $ stmQueueStore st
writeTVar qr $! Just q {notifier = Nothing}
pure notifierId
suspendQueue :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType ())
suspendQueue :: STMStoreClass s => s -> StoreQueue s -> IO (Either ErrorType ())
suspendQueue st sq =
atomically (readQueueRec qr >>= mapM suspend)
$>>= \rId -> withLog "suspendQueue" st (`logSuspendQueue` rId)
$>>= \_ -> withLog "suspendQueue" st (`logSuspendQueue` recipientId' sq)
where
qr = queueRec' sq
suspend q = do
writeTVar qr $! Just q {status = EntityOff}
pure $ recipientId q
suspend q = writeTVar qr $! Just q {status = EntityOff}
blockQueue :: STMQueueStore s => s -> StoreQueue s -> BlockingInfo -> IO (Either ErrorType ())
blockQueue :: STMStoreClass s => s -> StoreQueue s -> BlockingInfo -> IO (Either ErrorType ())
blockQueue st sq info =
atomically (readQueueRec qr >>= mapM block)
$>>= \rId -> withLog "blockQueue" st (\sl -> logBlockQueue sl rId info)
$>>= \_ -> withLog "blockQueue" st (\sl -> logBlockQueue sl (recipientId' sq) info)
where
qr = queueRec' sq
block q = do
writeTVar qr $ Just q {status = EntityBlocked info}
pure $ recipientId q
block q = writeTVar qr $ Just q {status = EntityBlocked info}
unblockQueue :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType ())
unblockQueue :: STMStoreClass s => s -> StoreQueue s -> IO (Either ErrorType ())
unblockQueue st sq =
atomically (readQueueRec qr >>= mapM unblock)
$>>= \rId -> withLog "unblockQueue" st (`logUnblockQueue` rId)
$>>= \_ -> withLog "unblockQueue" st (`logUnblockQueue` recipientId' sq)
where
qr = queueRec' sq
unblock q = do
writeTVar qr $ Just q {status = EntityActive}
pure $ recipientId q
unblock q = writeTVar qr $ Just q {status = EntityActive}
updateQueueTime :: STMQueueStore s => s -> StoreQueue s -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
updateQueueTime :: STMStoreClass s => s -> StoreQueue s -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
updateQueueTime st sq t = atomically (readQueueRec qr >>= mapM update) $>>= log'
where
qr = queueRec' sq
@@ -153,20 +161,21 @@ updateQueueTime st sq t = atomically (readQueueRec qr >>= mapM update) $>>= log'
let !q' = q {updatedAt = Just t}
in (writeTVar qr $! Just q') $> (q', True)
log' (q, changed)
| changed = q <$$ withLog "updateQueueTime" st (\sl -> logUpdateQueueTime sl (recipientId q) t)
| changed = q <$$ withLog "updateQueueTime" st (\sl -> logUpdateQueueTime sl (recipientId' sq) t)
| otherwise = pure $ Right q
deleteQueue' :: STMQueueStore s => s -> RecipientId -> StoreQueue s -> IO (Either ErrorType (QueueRec, Maybe (MsgQueue s)))
deleteQueue' st rId sq =
deleteQueue' :: STMStoreClass s => s -> StoreQueue s -> IO (Either ErrorType (QueueRec, Maybe (MsgQueue s)))
deleteQueue' st sq =
atomically (readQueueRec qr >>= mapM delete)
$>>= \q -> withLog "deleteQueue" st (`logDeleteQueue` rId)
$>>= \q -> withLog "deleteQueue" st (`logDeleteQueue` recipientId' sq)
>>= bimapM pure (\_ -> (q,) <$> atomically (swapTVar (msgQueue_' sq) Nothing))
where
qr = queueRec' sq
STMQueueStore {senders, notifiers} = stmQueueStore st
delete q = do
writeTVar qr Nothing
TM.delete (senderId q) $ senders' st
forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId $ notifiers' st
TM.delete (senderId q) senders
forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId notifiers
pure q
readQueueRec :: TVar (Maybe QueueRec) -> STM (Either ErrorType QueueRec)
@@ -183,10 +192,10 @@ withLog' name sl action =
where
err = name <> ", withLog, " <> show e
withLog :: STMQueueStore s => String -> s -> (StoreLog 'WriteMode -> IO ()) -> IO (Either ErrorType ())
withLog name = withLog' name . storeLog'
withLog :: STMStoreClass s => String -> s -> (StoreLog 'WriteMode -> IO ()) -> IO (Either ErrorType ())
withLog name = withLog' name . storeLog . stmQueueStore
readQueueStore :: forall s. STMQueueStore s => FilePath -> s -> IO ()
readQueueStore :: forall s. STMStoreClass s => FilePath -> s -> IO ()
readQueueStore f st = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLine . LB.lines
where
processLine :: LB.ByteString -> IO ()
@@ -195,13 +204,13 @@ readQueueStore f st = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLin
s = LB.toStrict s'
procLogRecord :: StoreLogRecord -> IO ()
procLogRecord = \case
CreateQueue q -> addQueue st q >>= qError (recipientId q) "CreateQueue"
CreateQueue rId q -> addQueue st rId q >>= qError rId "CreateQueue"
SecureQueue qId sKey -> withQueue qId "SecureQueue" $ \q -> secureQueue st q sKey
AddNotifier qId ntfCreds -> withQueue qId "AddNotifier" $ \q -> addQueueNotifier st q ntfCreds
SuspendQueue qId -> withQueue qId "SuspendQueue" $ suspendQueue st
BlockQueue qId info -> withQueue qId "BlockQueue" $ \q -> blockQueue st q info
UnblockQueue qId -> withQueue qId "UnblockQueue" $ unblockQueue st
DeleteQueue qId -> withQueue qId "DeleteQueue" $ deleteQueue st qId
DeleteQueue qId -> withQueue qId "DeleteQueue" $ deleteQueue st
DeleteNotifier qId -> withQueue qId "DeleteNotifier" $ deleteQueueNotifier st
UpdateTime qId t -> withQueue qId "UpdateTime" $ \q -> updateQueueTime st q t
printError :: String -> IO ()
+13 -14
View File
@@ -54,7 +54,7 @@ import System.Directory (doesFileExist, renameFile)
import System.IO
data StoreLogRecord
= CreateQueue QueueRec
= CreateQueue RecipientId QueueRec
| SecureQueue QueueId SndPublicAuthKey
| AddNotifier QueueId NtfCreds
| SuspendQueue QueueId
@@ -77,10 +77,9 @@ data SLRTag
| UpdateTime_
instance StrEncoding QueueRec where
strEncode QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt} =
strEncode QueueRec {recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt} =
B.unwords
[ "rid=" <> strEncode recipientId,
"rk=" <> strEncode recipientKey,
[ "rk=" <> strEncode recipientKey,
"rdh=" <> strEncode rcvDhSecret,
"sid=" <> strEncode senderId,
"sk=" <> strEncode senderKey
@@ -98,7 +97,6 @@ instance StrEncoding QueueRec where
_ -> " status=" <> strEncode status
strP = do
recipientId <- "rid=" *> strP_
recipientKey <- "rk=" *> strP_
rcvDhSecret <- "rdh=" *> strP_
senderId <- "sid=" *> strP_
@@ -107,7 +105,7 @@ instance StrEncoding QueueRec where
notifier <- optional $ " notifier=" *> strP
updatedAt <- optional $ " updated_at=" *> strP
status <- (" status=" *> strP) <|> pure EntityActive
pure QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt}
pure QueueRec {recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt}
instance StrEncoding SLRTag where
strEncode = \case
@@ -136,7 +134,7 @@ instance StrEncoding SLRTag where
instance StrEncoding StoreLogRecord where
strEncode = \case
CreateQueue q -> strEncode (CreateQueue_, q)
CreateQueue rId q -> B.unwords [strEncode CreateQueue_, "rid=" <> strEncode rId, strEncode q]
SecureQueue rId sKey -> strEncode (SecureQueue_, rId, sKey)
AddNotifier rId ntfCreds -> strEncode (AddNotifier_, rId, ntfCreds)
SuspendQueue rId -> strEncode (SuspendQueue_, rId)
@@ -148,7 +146,7 @@ instance StrEncoding StoreLogRecord where
strP =
strP_ >>= \case
CreateQueue_ -> CreateQueue <$> strP
CreateQueue_ -> CreateQueue <$> ("rid=" *> strP_) <*> strP
SecureQueue_ -> SecureQueue <$> strP_ <*> strP
AddNotifier_ -> AddNotifier <$> strP_ <*> strP
SuspendQueue_ -> SuspendQueue <$> strP
@@ -186,8 +184,8 @@ writeStoreLogRecord (WriteStoreLog _ h) r = E.uninterruptibleMask_ $ do
B.hPut h $ strEncode r `B.snoc` '\n' -- hPutStrLn makes write non-atomic for length > 1024
hFlush h
logCreateQueue :: StoreLog 'WriteMode -> QueueRec -> IO ()
logCreateQueue s = writeStoreLogRecord s . CreateQueue
logCreateQueue :: StoreLog 'WriteMode -> RecipientId -> QueueRec -> IO ()
logCreateQueue s rId q = writeStoreLogRecord s $ CreateQueue rId q
logSecureQueue :: StoreLog 'WriteMode -> QueueId -> SndPublicAuthKey -> IO ()
logSecureQueue s qId sKey = writeStoreLogRecord s $ SecureQueue qId sKey
@@ -248,10 +246,11 @@ readWriteStoreLog readStore writeStore f st =
renameFile tempBackup timedBackup
logInfo $ "original state preserved as " <> T.pack timedBackup
writeQueueStore :: STMQueueStore s => StoreLog 'WriteMode -> s -> IO ()
writeQueueStore s st = readTVarIO (activeMsgQueues st) >>= mapM_ writeQueue . M.assocs
writeQueueStore :: STMStoreClass s => StoreLog 'WriteMode -> s -> IO ()
writeQueueStore s st = readTVarIO qs >>= mapM_ writeQueue . M.assocs
where
qs = queues $ stmQueueStore st
writeQueue (rId, q) =
readTVarIO (queueRec' q) >>= \case
Just q' -> logCreateQueue s q'
Nothing -> atomically $ TM.delete rId $ activeMsgQueues st
Just q' -> logCreateQueue s rId q'
Nothing -> atomically $ TM.delete rId qs
+63 -2
View File
@@ -53,18 +53,21 @@ import Data.ByteString.Char8 (ByteString)
import qualified Data.ByteString.Char8 as B
import Data.List.NonEmpty (NonEmpty (..))
import qualified Data.List.NonEmpty as L
import Data.Text (Text)
import qualified Data.Text as T
import Data.Text.Encoding (encodeUtf8)
import qualified Data.Text.IO as TIO
import NtfClient
import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testNtfServer, testNtfServer2)
import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn, xit'')
import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore')
import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers)
import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, SENT)
import Simplex.Messaging.Agent.Store.AgentStore (getSavedNtfToken)
import Simplex.Messaging.Agent.Store.Common (withTransaction)
import Simplex.Messaging.Agent.Store.Interface (closeDBStore, reopenDBStore)
import qualified Simplex.Messaging.Agent.Store.DB as DB
import Simplex.Messaging.Agent.Store.Interface (closeDBStore, reopenDBStore)
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Notifications.Protocol
@@ -117,6 +120,12 @@ notificationTests t = do
it "should keep working with active token until replaced" $
withAPNSMockServer $ \apns ->
testNtfTokenChangeServers t apns
xit'' "should re-register token in NTInvalid status after register attempt" $
withAPNSMockServer $ \apns ->
testNtfTokenReRegisterInvalid t apns
xit'' "should re-register token in NTInvalid status after checking token" $
withAPNSMockServer $ \apns ->
testNtfTokenReRegisterInvalidOnCheck t apns
describe "notification server tests" $ do
it "should pass" $ testRunNTFServerTests t testNtfServer `shouldReturn` Nothing
let srv1 = testNtfServer {keyHash = "1234"}
@@ -459,6 +468,58 @@ testNtfTokenChangeServers t apns =
tkn <- registerTestToken a "qwer" NMInstant apns
checkNtfToken a tkn >>= \r -> liftIO $ r `shouldBe` NTActive
testNtfTokenReRegisterInvalid :: ATransport -> APNSMockServer -> IO ()
testNtfTokenReRegisterInvalid t apns = do
tkn <- withNtfServerStoreLog t $ \_ -> do
withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight $ do
tkn <- registerTestToken a "abcd" NMInstant apns
NTActive <- checkNtfToken a tkn
pure tkn
threadDelay 250000
-- start server to compact
withNtfServerStoreLog t $ \_ -> pure ()
threadDelay 250000
replaceSubstringInFile ntfTestStoreLogFile "tokenStatus=ACTIVE" "tokenStatus=INVALID"
threadDelay 250000
withNtfServerStoreLog t $ \_ -> do
withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight_ $ do
NTInvalid Nothing <- registerNtfToken a tkn NMInstant
tkn1 <- registerTestToken a "abcd" NMInstant apns
NTActive <- checkNtfToken a tkn1
pure ()
replaceSubstringInFile :: FilePath -> Text -> Text -> IO ()
replaceSubstringInFile filePath oldText newText = do
content <- TIO.readFile filePath
let newContent = T.replace oldText newText content
TIO.writeFile filePath newContent
testNtfTokenReRegisterInvalidOnCheck :: ATransport -> APNSMockServer -> IO ()
testNtfTokenReRegisterInvalidOnCheck t apns = do
tkn <- withNtfServerStoreLog t $ \_ -> do
withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight $ do
tkn <- registerTestToken a "abcd" NMInstant apns
NTActive <- checkNtfToken a tkn
pure tkn
threadDelay 250000
-- start server to compact
withNtfServerStoreLog t $ \_ -> pure ()
threadDelay 250000
replaceSubstringInFile ntfTestStoreLogFile "tokenStatus=ACTIVE" "tokenStatus=INVALID"
threadDelay 250000
withNtfServerStoreLog t $ \_ -> do
withAgent 1 agentCfg initAgentServers testDB $ \a -> runRight_ $ do
NTInvalid Nothing <- checkNtfToken a tkn
tkn1 <- registerTestToken a "abcd" NMInstant apns
NTActive <- checkNtfToken a tkn1
pure ()
testRunNTFServerTests :: ATransport -> NtfServer -> IO (Maybe ProtocolTestFailure)
testRunNTFServerTests t srv =
withNtfServerOn t ntfTestPort $
+3 -2
View File
@@ -45,6 +45,7 @@ import Simplex.Messaging.Agent.Store.AgentStore
import Simplex.Messaging.Agent.Store.SQLite
import Simplex.Messaging.Agent.Store.SQLite.Common (DBStore (..), withTransaction')
import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
import Simplex.Messaging.Agent.Store.SQLite.Migrations (appMigrations, getCurrentMigrations)
import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
import qualified Simplex.Messaging.Crypto as C
@@ -81,7 +82,7 @@ createEncryptedStore key keepKey = do
-- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
-- IO operations on multiple similarly named files; error seems to be environment specific
r <- randomIO :: IO Word32
Right st <- createDBStore (DBOpts (testDB <> show r) key keepKey True DB.TQOff) Migrations.app MCError
Right st <- createDBStore (DBOpts (testDB <> show r) key keepKey True DB.TQOff) appMigrations MCError
withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1);")
pure st
@@ -635,7 +636,7 @@ testReopenEncryptedStoreKeepKey = do
hasMigrations st
getMigrations :: DBStore -> IO Bool
getMigrations st = not . null <$> withTransaction st Migrations.getCurrent
getMigrations st = not . null <$> withTransaction st getCurrentMigrations
hasMigrations :: DBStore -> Expectation
hasMigrations st = getMigrations st `shouldReturn` True
+7 -7
View File
@@ -50,7 +50,7 @@ testVerifySchemaDump :: IO ()
testVerifySchemaDump = do
savedSchema <- ifM (doesFileExist appSchema) (readFile appSchema) (pure "")
savedSchema `deepseq` pure ()
void $ createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCConsole
void $ createDBStore (DBOpts testDB "" False True TQOff) appMigrations MCConsole
getSchema testDB appSchema `shouldReturn` savedSchema
removeFile testDB
@@ -58,7 +58,7 @@ testVerifyLintFKeyIndexes :: IO ()
testVerifyLintFKeyIndexes = do
savedLint <- ifM (doesFileExist appLint) (readFile appLint) (pure "")
savedLint `deepseq` pure ()
void $ createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCConsole
void $ createDBStore (DBOpts testDB "" False True TQOff) appMigrations MCConsole
getLintFKeyIndexes testDB "tests/tmp/agent_lint.sql" `shouldReturn` savedLint
removeFile testDB
@@ -70,9 +70,9 @@ withTmpFiles =
testSchemaMigrations :: IO ()
testSchemaMigrations = do
let noDownMigrations = dropWhileEnd (\Migration {down} -> isJust down) Migrations.app
let noDownMigrations = dropWhileEnd (\Migration {down} -> isJust down) appMigrations
Right st <- createDBStore (DBOpts testDB "" False True TQOff) noDownMigrations MCError
mapM_ (testDownMigration st) $ drop (length noDownMigrations) Migrations.app
mapM_ (testDownMigration st) $ drop (length noDownMigrations) appMigrations
closeDBStore st
removeFile testDB
removeFile testSchema
@@ -94,19 +94,19 @@ testSchemaMigrations = do
testUsersMigrationNew :: IO ()
testUsersMigrationNew = do
Right st <- createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCError
Right st <- createDBStore (DBOpts testDB "" False True TQOff) appMigrations MCError
withTransaction' st (`SQL.query_` "SELECT user_id FROM users;")
`shouldReturn` ([] :: [Only Int])
closeDBStore st
testUsersMigrationOld :: IO ()
testUsersMigrationOld = do
let beforeUsers = takeWhile (("m20230110_users" /=) . name) Migrations.app
let beforeUsers = takeWhile (("m20230110_users" /=) . name) appMigrations
Right st <- createDBStore (DBOpts testDB "" False True TQOff) beforeUsers MCError
withTransaction' st (`SQL.query_` "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'users';")
`shouldReturn` ([] :: [Only String])
closeDBStore st
Right st' <- createDBStore (DBOpts testDB "" False True TQOff) Migrations.app MCYesUp
Right st' <- createDBStore (DBOpts testDB "" False True TQOff) appMigrations MCYesUp
withTransaction' st' (`SQL.query_` "SELECT user_id FROM users;")
`shouldReturn` ([Only (1 :: Int)])
closeDBStore st'
+80 -81
View File
@@ -58,12 +58,12 @@ msgStoreTests = do
it "should create write file when missing" testWriteFileMissing
it "should create read file when read and write files are missing" testReadAndWriteFilesMissing
where
someMsgStoreTests :: STMQueueStore s => SpecWith s
someMsgStoreTests :: STMStoreClass s => SpecWith s
someMsgStoreTests = do
it "should get queue and store/read messages" testGetQueue
it "should not fail on EOF when changing read journal" testChangeReadJournal
withMsgStore :: STMQueueStore s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
withMsgStore :: STMStoreClass s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
withMsgStore cfg = bracket (newMsgStore cfg) closeMsgStore
testSMTStoreConfig :: STMStoreConfig
@@ -105,8 +105,7 @@ testNewQueueRec g sndSecure = do
(k, pk) <- atomically $ C.generateKeyPair @'C.X25519 g
let qr =
QueueRec
{ recipientId = rId,
recipientKey,
{ recipientKey,
rcvDhSecret = C.dh' k pk,
senderId,
senderKey = Nothing,
@@ -117,66 +116,66 @@ testNewQueueRec g sndSecure = do
}
pure (rId, qr)
testGetQueue :: STMQueueStore s => s -> IO ()
testGetQueue :: STMStoreClass s => s -> IO ()
testGetQueue ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g True
runRight_ $ do
q <- ExceptT $ addQueue ms qr
let write s = writeMsg ms rId q True =<< mkMessage s
q <- ExceptT $ addQueue ms rId qr
let write s = writeMsg ms q True =<< mkMessage s
Just (Message {msgId = mId1}, True) <- write "message 1"
Just (Message {msgId = mId2}, False) <- write "message 2"
Just (Message {msgId = mId3}, False) <- write "message 3"
Msg "message 1" <- tryPeekMsg ms rId q
Msg "message 1" <- tryPeekMsg ms rId q
Nothing <- tryDelMsg ms rId q mId2
Msg "message 1" <- tryDelMsg ms rId q mId1
Nothing <- tryDelMsg ms rId q mId1
Msg "message 2" <- tryPeekMsg ms rId q
Nothing <- tryDelMsg ms rId q mId1
(Nothing, Msg "message 2") <- tryDelPeekMsg ms rId q mId1
(Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms rId q mId2
(Nothing, Msg "message 3") <- tryDelPeekMsg ms rId q mId2
Msg "message 3" <- tryPeekMsg ms rId q
(Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
Nothing <- tryDelMsg ms rId q mId2
Nothing <- tryDelMsg ms rId q mId3
Nothing <- tryPeekMsg ms rId q
Msg "message 1" <- tryPeekMsg ms q
Msg "message 1" <- tryPeekMsg ms q
Nothing <- tryDelMsg ms q mId2
Msg "message 1" <- tryDelMsg ms q mId1
Nothing <- tryDelMsg ms q mId1
Msg "message 2" <- tryPeekMsg ms q
Nothing <- tryDelMsg ms q mId1
(Nothing, Msg "message 2") <- tryDelPeekMsg ms q mId1
(Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms q mId2
(Nothing, Msg "message 3") <- tryDelPeekMsg ms q mId2
Msg "message 3" <- tryPeekMsg ms q
(Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3
Nothing <- tryDelMsg ms q mId2
Nothing <- tryDelMsg ms q mId3
Nothing <- tryPeekMsg ms q
Just (Message {msgId = mId4}, True) <- write "message 4"
Msg "message 4" <- tryPeekMsg ms rId q
Msg "message 4" <- tryPeekMsg ms q
Just (Message {msgId = mId5}, False) <- write "message 5"
(Nothing, Msg "message 4") <- tryDelPeekMsg ms rId q mId3
(Msg "message 4", Msg "message 5") <- tryDelPeekMsg ms rId q mId4
(Nothing, Msg "message 4") <- tryDelPeekMsg ms q mId3
(Msg "message 4", Msg "message 5") <- tryDelPeekMsg ms q mId4
Just (Message {msgId = mId6}, False) <- write "message 6"
Just (Message {msgId = mId7}, False) <- write "message 7"
Nothing <- write "message 8"
Msg "message 5" <- tryPeekMsg ms rId q
(Nothing, Msg "message 5") <- tryDelPeekMsg ms rId q mId4
(Msg "message 5", Msg "message 6") <- tryDelPeekMsg ms rId q mId5
(Msg "message 6", Msg "message 7") <- tryDelPeekMsg ms rId q mId6
(Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg ms rId q mId7
(Just MessageQuota {}, Nothing) <- tryDelPeekMsg ms rId q mId8
(Nothing, Nothing) <- tryDelPeekMsg ms rId q mId8
void $ ExceptT $ deleteQueue ms rId q
Msg "message 5" <- tryPeekMsg ms q
(Nothing, Msg "message 5") <- tryDelPeekMsg ms q mId4
(Msg "message 5", Msg "message 6") <- tryDelPeekMsg ms q mId5
(Msg "message 6", Msg "message 7") <- tryDelPeekMsg ms q mId6
(Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg ms q mId7
(Just MessageQuota {}, Nothing) <- tryDelPeekMsg ms q mId8
(Nothing, Nothing) <- tryDelPeekMsg ms q mId8
void $ ExceptT $ deleteQueue ms q
testChangeReadJournal :: STMQueueStore s => s -> IO ()
testChangeReadJournal :: STMStoreClass s => s -> IO ()
testChangeReadJournal ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g True
runRight_ $ do
q <- ExceptT $ addQueue ms qr
let write s = writeMsg ms rId q True =<< mkMessage s
q <- ExceptT $ addQueue ms rId qr
let write s = writeMsg ms q True =<< mkMessage s
Just (Message {msgId = mId1}, True) <- write "message 1"
(Msg "message 1", Nothing) <- tryDelPeekMsg ms rId q mId1
(Msg "message 1", Nothing) <- tryDelPeekMsg ms q mId1
Just (Message {msgId = mId2}, True) <- write "message 2"
(Msg "message 2", Nothing) <- tryDelPeekMsg ms rId q mId2
(Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2
Just (Message {msgId = mId3}, True) <- write "message 3"
(Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
(Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3
Just (Message {msgId = mId4}, True) <- write "message 4"
(Msg "message 4", Nothing) <- tryDelPeekMsg ms rId q mId4
(Msg "message 4", Nothing) <- tryDelPeekMsg ms q mId4
Just (Message {msgId = mId5}, True) <- write "message 5"
(Msg "message 5", Nothing) <- tryDelPeekMsg ms rId q mId5
void $ ExceptT $ deleteQueue ms rId q
(Msg "message 5", Nothing) <- tryDelPeekMsg ms q mId5
void $ ExceptT $ deleteQueue ms q
testExportImportStore :: JournalMsgStore -> IO ()
testExportImportStore ms = do
@@ -185,21 +184,21 @@ testExportImportStore ms = do
(rId2, qr2) <- testNewQueueRec g True
sl <- readWriteQueueStore testStoreLogFile ms
runRight_ $ do
let write rId q s = writeMsg ms rId q True =<< mkMessage s
q1 <- ExceptT $ addQueue ms qr1
liftIO $ logCreateQueue sl qr1
Just (Message {}, True) <- write rId1 q1 "message 1"
Just (Message {}, False) <- write rId1 q1 "message 2"
q2 <- ExceptT $ addQueue ms qr2
liftIO $ logCreateQueue sl qr2
Just (Message {msgId = mId3}, True) <- write rId2 q2 "message 3"
Just (Message {msgId = mId4}, False) <- write rId2 q2 "message 4"
(Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms rId2 q2 mId3
(Msg "message 4", Nothing) <- tryDelPeekMsg ms rId2 q2 mId4
Just (Message {}, True) <- write rId2 q2 "message 5"
Just (Message {}, False) <- write rId2 q2 "message 6"
Just (Message {}, False) <- write rId2 q2 "message 7"
Nothing <- write rId2 q2 "message 8"
let write q s = writeMsg ms q True =<< mkMessage s
q1 <- ExceptT $ addQueue ms rId1 qr1
liftIO $ logCreateQueue sl rId1 qr1
Just (Message {}, True) <- write q1 "message 1"
Just (Message {}, False) <- write q1 "message 2"
q2 <- ExceptT $ addQueue ms rId2 qr2
liftIO $ logCreateQueue sl rId2 qr2
Just (Message {msgId = mId3}, True) <- write q2 "message 3"
Just (Message {msgId = mId4}, False) <- write q2 "message 4"
(Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms q2 mId3
(Msg "message 4", Nothing) <- tryDelPeekMsg ms q2 mId4
Just (Message {}, True) <- write q2 "message 5"
Just (Message {}, False) <- write q2 "message 6"
Just (Message {}, False) <- write q2 "message 7"
Nothing <- write q2 "message 8"
pure ()
length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 3
@@ -297,10 +296,10 @@ testMessageState ms = do
(rId, qr) <- testNewQueueRec g True
let dir = msgQueueDirectory ms rId
statePath = msgQueueStatePath dir $ B.unpack (B64.encode $ unEntityId rId)
write q s = writeMsg ms rId q True =<< mkMessage s
write q s = writeMsg ms q True =<< mkMessage s
mId1 <- runRight $ do
q <- ExceptT $ addQueue ms qr
q <- ExceptT $ addQueue ms rId qr
Just (Message {msgId = mId1}, True) <- write q "message 1"
Just (Message {}, False) <- write q "message 2"
liftIO $ closeMsgQueue q
@@ -312,19 +311,19 @@ testMessageState ms = do
runRight_ $ do
q <- ExceptT $ getQueue ms SRecipient rId
Just (Message {msgId = mId3}, False) <- write q "message 3"
(Msg "message 1", Msg "message 3") <- tryDelPeekMsg ms rId q mId1
(Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
(Msg "message 1", Msg "message 3") <- tryDelPeekMsg ms q mId1
(Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3
liftIO $ closeMsgQueue q
testReadFileMissing :: JournalMsgStore -> IO ()
testReadFileMissing ms = do
g <- C.newRandom
(rId, qr) <- testNewQueueRec g True
let write q s = writeMsg ms rId q True =<< mkMessage s
let write q s = writeMsg ms q True =<< mkMessage s
q <- runRight $ do
q <- ExceptT $ addQueue ms qr
q <- ExceptT $ addQueue ms rId qr
Just (Message {}, True) <- write q "message 1"
Msg "message 1" <- tryPeekMsg ms rId q
Msg "message 1" <- tryPeekMsg ms q
pure q
mq <- fromJust <$> readTVarIO (msgQueue_' q)
@@ -335,9 +334,9 @@ testReadFileMissing ms = do
runRight_ $ do
q' <- ExceptT $ getQueue ms SRecipient rId
Nothing <- tryPeekMsg ms rId q'
Nothing <- tryPeekMsg ms q'
Just (Message {}, True) <- write q' "message 2"
Msg "message 2" <- tryPeekMsg ms rId q'
Msg "message 2" <- tryPeekMsg ms q'
pure ()
testReadFileMissingSwitch :: JournalMsgStore -> IO ()
@@ -354,8 +353,8 @@ testReadFileMissingSwitch ms = do
runRight_ $ do
q' <- ExceptT $ getQueue ms SRecipient rId
Just (Message {}, False) <- writeMsg ms rId q' True =<< mkMessage "message 6"
Msg "message 5" <- tryPeekMsg ms rId q'
Just (Message {}, False) <- writeMsg ms q' True =<< mkMessage "message 6"
Msg "message 5" <- tryPeekMsg ms q'
pure ()
testWriteFileMissing :: JournalMsgStore -> IO ()
@@ -373,12 +372,12 @@ testWriteFileMissing ms = do
runRight_ $ do
q' <- ExceptT $ getQueue ms SRecipient rId
Just Message {msgId = mId3} <- tryPeekMsg ms rId q'
(Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms rId q' mId3
Just Message {msgId = mId4} <- tryPeekMsg ms rId q'
(Msg "message 4", Nothing) <- tryDelPeekMsg ms rId q' mId4
Just (Message {}, True) <- writeMsg ms rId q' True =<< mkMessage "message 6"
Msg "message 6" <- tryPeekMsg ms rId q'
Just Message {msgId = mId3} <- tryPeekMsg ms q'
(Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms q' mId3
Just Message {msgId = mId4} <- tryPeekMsg ms q'
(Msg "message 4", Nothing) <- tryDelPeekMsg ms q' mId4
Just (Message {}, True) <- writeMsg ms q' True =<< mkMessage "message 6"
Msg "message 6" <- tryPeekMsg ms q'
pure ()
testReadAndWriteFilesMissing :: JournalMsgStore -> IO ()
@@ -395,20 +394,20 @@ testReadAndWriteFilesMissing ms = do
runRight_ $ do
q' <- ExceptT $ getQueue ms SRecipient rId
Nothing <- tryPeekMsg ms rId q'
Just (Message {}, True) <- writeMsg ms rId q' True =<< mkMessage "message 6"
Msg "message 6" <- tryPeekMsg ms rId q'
Nothing <- tryPeekMsg ms q'
Just (Message {}, True) <- writeMsg ms q' True =<< mkMessage "message 6"
Msg "message 6" <- tryPeekMsg ms q'
pure ()
writeMessages :: JournalMsgStore -> RecipientId -> QueueRec -> IO JournalQueue
writeMessages ms rId qr = runRight $ do
q <- ExceptT $ addQueue ms qr
let write s = writeMsg ms rId q True =<< mkMessage s
q <- ExceptT $ addQueue ms rId qr
let write s = writeMsg ms q True =<< mkMessage s
Just (Message {msgId = mId1}, True) <- write "message 1"
Just (Message {msgId = mId2}, False) <- write "message 2"
Just (Message {}, False) <- write "message 3"
(Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms rId q mId1
(Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms rId q mId2
(Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1
(Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms q mId2
Just (Message {}, False) <- write "message 4"
Just (Message {}, False) <- write "message 5"
pure q
+12 -12
View File
@@ -60,38 +60,38 @@ storeLogTests =
("SMP server store log, sndSecure = " <> show sndSecure)
[ SLTC
{ name = "create new queue",
saved = [CreateQueue qr],
compacted = [CreateQueue qr],
saved = [CreateQueue rId qr],
compacted = [CreateQueue rId qr],
state = M.fromList [(rId, qr)]
},
SLTC
{ name = "secure queue",
saved = [CreateQueue qr, SecureQueue rId testPublicAuthKey],
compacted = [CreateQueue qr {senderKey = Just testPublicAuthKey}],
saved = [CreateQueue rId qr, SecureQueue rId testPublicAuthKey],
compacted = [CreateQueue rId qr {senderKey = Just testPublicAuthKey}],
state = M.fromList [(rId, qr {senderKey = Just testPublicAuthKey})]
},
SLTC
{ name = "create and delete queue",
saved = [CreateQueue qr, DeleteQueue rId],
saved = [CreateQueue rId qr, DeleteQueue rId],
compacted = [],
state = M.fromList []
},
SLTC
{ name = "create queue and add notifier",
saved = [CreateQueue qr, AddNotifier rId ntfCreds],
compacted = [CreateQueue $ qr {notifier = Just ntfCreds}],
saved = [CreateQueue rId qr, AddNotifier rId ntfCreds],
compacted = [CreateQueue rId qr {notifier = Just ntfCreds}],
state = M.fromList [(rId, qr {notifier = Just ntfCreds})]
},
SLTC
{ name = "delete notifier",
saved = [CreateQueue qr, AddNotifier rId ntfCreds, DeleteNotifier rId],
compacted = [CreateQueue qr],
saved = [CreateQueue rId qr, AddNotifier rId ntfCreds, DeleteNotifier rId],
compacted = [CreateQueue rId qr],
state = M.fromList [(rId, qr)]
},
SLTC
{ name = "update time",
saved = [CreateQueue qr, UpdateTime rId date],
compacted = [CreateQueue qr {updatedAt = Just date}],
saved = [CreateQueue rId qr, UpdateTime rId date],
compacted = [CreateQueue rId qr {updatedAt = Just date}],
state = M.fromList [(rId, qr {updatedAt = Just date})]
}
]
@@ -112,4 +112,4 @@ testSMPStoreLog testSuite tests =
([], compacted') <- partitionEithers . map strDecode . B.lines <$> B.readFile testStoreLogFile
compacted' `shouldBe` compacted
storeState :: JournalMsgStore -> IO (M.Map RecipientId QueueRec)
storeState st = M.mapMaybe id <$> (readTVarIO (queues st) >>= mapM (readTVarIO . queueRec'))
storeState st = M.mapMaybe id <$> (readTVarIO (queues $ stmQueueStore st) >>= mapM (readTVarIO . queueRec'))