4.3.1

* fix SMP server stats

* add server stats test

* fix test

CHANGELOG.md

@@ -1,3 +1,13 @@
+# 4.3.0
+
+SMP server:
+
+- additional server usage statistics.
+
+SMP agent:
+
+- increase retry interval when sending messages after ERR QUOTA.
+
 # 4.2.0
 
 SMP agent and server:

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 4.2.2
+version: 4.3.1
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

rfcs/2022-12-26-simplex-file-transfer.md

@@ -95,10 +95,12 @@ To send the file, the sender will:
 
 - compute SHA512 digest
 - pad the file to match the whole number of chunks in size,
-- encrypt it with a randomly chosen symmetric key and IV (e.g., using NaCL cryptobox),
+- encrypt it with a randomly chosen symmetric key and IV (e.g., using NaCL crypto_secretbox),
 - split into fixed size chunks
 - upload each chunk to a randomly chosen server.
 
+The sending client should generate more per-recipient keys than the actual number of recipients, possibly rounding up to a power of 2, to conceal the actual number of intended recipients.
+
 Then the sending client will combine addresses of all chunks and other information into "file description", different for each file recipient, that will include:
 
 - an encryption key that was used to encrypt the file (the same for all recipients).
@@ -106,22 +108,24 @@ Then the sending client will combine addresses of all chunks and other informati
 - list of chunk descriptions; information for each chunk:
   - private Ed25519 key to sign commands for file transfer server.
   - chunk address (server host and chunk ID).
+  - chunk sha512 digest
 
 To reduce the size, chunk descriptions will be grouped by the server host.
 
-This "file description" itself will be sent as a small file. To estimate its size:
+This "file description" itself will be sent as a small file over an authenticated channel, to prevent file description modification. To estimate its size:
 
 - each chunk \* redundancy per chunk, assuming chunks are grouped per server:
-  - chunk number in the file - 8 bytes (including any overhead)
+  - 1-based chunk number in the file - 8 bytes (including any overhead)
   - Ed25519 key (different for each recipient / chunk combination) - 32 bytes \* 4/3 (base64, assuming text encoding)
   - chunk ID (different for each recipient) - 64 bytes \* 4/3
+  - optional (only in the first chunk occurence) chunk sha512 digests - 64 bytes \* 4/3
 - server addresses - say, 128 bytes per server
 - sha512 digest - 64 bytes \* 4/3
 - encryption key - 32 bytes \* 4/3
 - IV - 32 bytes \* 4/3
 - encoding overhead - say, 256 bytes
 
-For 1gb file, sent via 4 different servers, in 8Mb chunks, with redundancy 2, the size of "file description", assuming text encoding, will be ~34kb (`128 * (8 + 32 + 64) * 2 * 4/3 + 128 * 4 + (64 + 32 + 32) * 4/3 + 256`).
+For 1gb file, sent via 4 different servers, in 8Mb chunks, with redundancy 2, the size of "file description", assuming text encoding, will be ~45kb (`128 * (8 + 32 + 64) * 2 * 4/3 + 128 * 64 * 4/3 + 128 * 4 + (64 + 32 + 32) * 4/3 + 256`).
 
 File description format (yml):
 
@@ -132,11 +136,12 @@ chunk: 8Mb
 hash: abc=
 key: abc=
 iv: abc=
+part_hashes: [def=, def=, def=, def=]
 parts:
   - server: xftp://abc=@example1.com
-    chunks: [1:abc=:def=, 3:abc=:def=]
+    chunks: [1:abc=:def=:ghi=, 3:abc=:def=:ghi=]
   - server: xftp://abc=@example2.com
-    chunks: [2:abc=:def=, 4:abc=:def=]
+    chunks: [2:abc=:def=:ghi=, 4:abc=:def=:ghi=]
   - server: xftp://abc=@example3.com
     chunks: [1:abc=:def=, 4:abc=:def=]
   - server: xftp://abc=@example4.com

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        4.2.2
+version:        4.3.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -122,9 +122,9 @@ defaultMessageRetryInterval =
         -- to resume sending on QCONT messages.
         -- After that local message expiration period should be also increased.
         RetryInterval
-          { initialInterval = 10_000000,
+          { initialInterval = 30_000000,
             increaseAfter = 30_000000,
-            maxInterval = 300_000000
+            maxInterval = 600_000000
           }
     }
 

src/Simplex/Messaging/Server.hs

@@ -104,8 +104,8 @@ type M a = ReaderT Env IO a
 smpServer :: TMVar Bool -> ServerConfig -> M ()
 smpServer started cfg@ServerConfig {transports, logTLSErrors} = do
   s <- asks server
-  restoreServerStats
   restoreServerMessages
+  restoreServerStats
   raceAny_
     ( serverThread s subscribedQ subscribers subscriptions cancelSub :
       serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :
@@ -174,7 +174,7 @@ smpServer started cfg@ServerConfig {transports, logTLSErrors} = do
       initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
       liftIO $ putStrLn $ "server stats log enabled: " <> statsFilePath
       threadDelay $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
-      ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, activeQueues} <- asks serverStats
+      ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, activeQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount} <- asks serverStats
       let interval = 1000000 * logInterval
       withFile statsFilePath AppendMode $ \h -> liftIO $ do
         hSetBuffering h LineBuffering
@@ -187,7 +187,31 @@ smpServer started cfg@ServerConfig {transports, logTLSErrors} = do
           msgSent' <- atomically $ swapTVar msgSent 0
           msgRecv' <- atomically $ swapTVar msgRecv 0
           ps <- atomically $ periodStatCounts activeQueues ts
-          hPutStrLn h $ intercalate "," [iso8601Show $ utctDay fromTime', show qCreated', show qSecured', show qDeleted', show msgSent', show msgRecv', dayCount ps, weekCount ps, monthCount ps]
+          msgSentNtf' <- atomically $ swapTVar msgSentNtf 0
+          msgRecvNtf' <- atomically $ swapTVar msgRecvNtf 0
+          psNtf <- atomically $ periodStatCounts activeQueuesNtf ts
+          qCount' <- readTVarIO qCount
+          msgCount' <- readTVarIO msgCount
+          hPutStrLn h $
+            intercalate
+              ","
+              [ iso8601Show $ utctDay fromTime',
+                show qCreated',
+                show qSecured',
+                show qDeleted',
+                show msgSent',
+                show msgRecv',
+                dayCount ps,
+                weekCount ps,
+                monthCount ps,
+                show msgSentNtf',
+                show msgRecvNtf',
+                dayCount psNtf,
+                weekCount psNtf,
+                monthCount psNtf,
+                show qCount',
+                show msgCount'
+              ]
           threadDelay interval
 
     runClient :: Transport c => TProxy c -> c -> M ()
@@ -387,6 +411,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
                   withLog (`logCreateById` rId)
                   stats <- asks serverStats
                   atomically $ modifyTVar' (qCreated stats) (+ 1)
+                  atomically $ modifyTVar' (qCount stats) (+ 1)
                   subscribeQueue qr rId $> IDS (qik ids)
 
             logCreateById :: StoreLog 'WriteMode -> RecipientId -> IO ()
@@ -509,12 +534,12 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
                   q <- getStoreMsgQueue "ACK" queueId
                   case s of
                     Sub {subThread = ProhibitSub} -> do
-                      msgDeleted <- atomically $ tryDelMsg q msgId
-                      when msgDeleted updateStats
+                      deletedMsg_ <- atomically $ tryDelMsg q msgId
+                      mapM_ updateStats deletedMsg_
                       pure ok
                     _ -> do
-                      (msgDeleted, msg_) <- atomically $ tryDelPeekMsg q msgId
-                      when msgDeleted updateStats
+                      (deletedMsg_, msg_) <- atomically $ tryDelPeekMsg q msgId
+                      mapM_ updateStats deletedMsg_
                       deliverMessage "ACK" qr queueId sub q msg_
                 _ -> pure $ err NO_MSG
           where
@@ -525,11 +550,17 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
                 if msgId == msgId' || B.null msgId
                   then pure $ Just s
                   else putTMVar delivered msgId' $> Nothing
-            updateStats :: m ()
-            updateStats = do
-              stats <- asks serverStats
-              atomically $ modifyTVar' (msgRecv stats) (+ 1)
-              atomically $ updatePeriodStats (activeQueues stats) queueId
+            updateStats :: Message -> m ()
+            updateStats = \case
+              MessageQuota {} -> pure ()
+              Message {msgFlags} -> do
+                stats <- asks serverStats
+                atomically $ modifyTVar' (msgRecv stats) (+ 1)
+                atomically $ modifyTVar' (msgCount stats) (+ 1)
+                atomically $ updatePeriodStats (activeQueues stats) queueId
+                when (notification msgFlags) $ do
+                  atomically $ modifyTVar' (msgRecvNtf stats) (+ 1)
+                  atomically $ updatePeriodStats (activeQueuesNtf stats) queueId
 
         sendMessage :: QueueRec -> MsgFlags -> MsgBody -> m (Transmission BrokerMsg)
         sendMessage qr msgFlags msgBody
@@ -547,10 +578,13 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
                   case msg_ of
                     Nothing -> pure $ err QUOTA
                     Just msg -> time "SEND ok" $ do
-                      when (notification msgFlags) $
-                        atomically . trySendNotification msg =<< asks idsDrg
                       stats <- asks serverStats
+                      when (notification msgFlags) $ do
+                        atomically . trySendNotification msg =<< asks idsDrg
+                        atomically $ modifyTVar' (msgSentNtf stats) (+ 1)
+                        atomically $ updatePeriodStats (activeQueuesNtf stats) (recipientId qr)
                       atomically $ modifyTVar' (msgSent stats) (+ 1)
+                      atomically $ modifyTVar' (msgCount stats) (subtract 1)
                       atomically $ updatePeriodStats (activeQueues stats) (recipientId qr)
                       pure ok
           where
@@ -647,6 +681,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
           ms <- asks msgStore
           stats <- asks serverStats
           atomically $ modifyTVar' (qDeleted stats) (+ 1)
+          atomically $ modifyTVar' (qCount stats) (subtract 1)
           atomically $
             deleteQueue st queueId >>= \case
               Left e -> pure $ err e
@@ -726,7 +761,10 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
               full <- atomically $ do
                 q <- getMsgQueue ms rId quota
                 isNothing <$> writeMsg q msg
-              when full . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
+              case msg of
+                Message {} ->
+                  when full . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
+                MessageQuota {} -> pure ()
             updateMsgV1toV3 QueueRec {rcvDhSecret} RcvMessage {msgId, msgTs, msgFlags, msgBody = EncRcvMsgBody body} = do
               let nonce = C.cbNonce msgId
               msgBody <- liftEither . first (msgErr "v1 message decryption") $ C.maxLenBS =<< C.cbDecrypt rcvDhSecret nonce body
@@ -752,7 +790,9 @@ restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStat
       liftIO (strDecode <$> B.readFile f) >>= \case
         Right d -> do
           s <- asks serverStats
-          atomically $ setServerStats s d
+          _qCount <- fmap (length . M.keys) . readTVarIO . queues =<< asks queueStore
+          _msgCount <- foldM (\n q -> (n +) <$> readTVarIO (size q)) 0 =<< readTVarIO =<< asks msgStore
+          atomically $ setServerStats s d {_qCount, _msgCount}
           renameFile f $ f <> ".bak"
           logInfo "server stats restored"
         Left e -> do

src/Simplex/Messaging/Server/MsgStore/STM.hs

@@ -8,7 +8,7 @@
 
 module Simplex.Messaging.Server.MsgStore.STM
   ( STMMsgStore,
-    MsgQueue,
+    MsgQueue (..),
     newMsgStore,
     getMsgQueue,
     delMsgQueue,
@@ -86,22 +86,22 @@ peekMsg :: MsgQueue -> STM Message
 peekMsg = peekTQueue . msgQueue
 {-# INLINE peekMsg #-}
 
-tryDelMsg :: MsgQueue -> MsgId -> STM Bool
+tryDelMsg :: MsgQueue -> MsgId -> STM (Maybe Message)
 tryDelMsg mq msgId' =
   tryPeekMsg mq >>= \case
-    Just msg
-      | msgId msg == msgId' || B.null msgId' -> tryDeleteMsg mq >> pure True
-      | otherwise -> pure False
-    _ -> pure False
+    msg_@(Just msg)
+      | msgId msg == msgId' || B.null msgId' -> tryDeleteMsg mq >> pure msg_
+      | otherwise -> pure Nothing
+    _ -> pure Nothing
 
 -- atomic delete (== read) last and peek next message if available
-tryDelPeekMsg :: MsgQueue -> MsgId -> STM (Bool, Maybe Message)
+tryDelPeekMsg :: MsgQueue -> MsgId -> STM (Maybe Message, Maybe Message)
 tryDelPeekMsg mq msgId' =
   tryPeekMsg mq >>= \case
     msg_@(Just msg)
-      | msgId msg == msgId' || B.null msgId' -> (True,) <$> (tryDeleteMsg mq >> tryPeekMsg mq)
-      | otherwise -> pure (False, msg_)
-    _ -> pure (False, Nothing)
+      | msgId msg == msgId' || B.null msgId' -> (msg_,) <$> (tryDeleteMsg mq >> tryPeekMsg mq)
+      | otherwise -> pure (Nothing, msg_)
+    _ -> pure (Nothing, Nothing)
 
 deleteExpiredMsgs :: MsgQueue -> Int64 -> STM ()
 deleteExpiredMsgs mq old = loop

src/Simplex/Messaging/Server/Stats.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
@@ -5,7 +6,7 @@
 
 module Simplex.Messaging.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.Set (Set)
@@ -24,7 +25,12 @@ data ServerStats = ServerStats
     qDeleted :: TVar Int,
     msgSent :: TVar Int,
     msgRecv :: TVar Int,
-    activeQueues :: PeriodStats RecipientId
+    activeQueues :: PeriodStats RecipientId,
+    msgSentNtf :: TVar Int,
+    msgRecvNtf :: TVar Int,
+    activeQueuesNtf :: PeriodStats RecipientId,
+    qCount :: TVar Int,
+    msgCount :: TVar Int
   }
 
 data ServerStatsData = ServerStatsData
@@ -34,8 +40,14 @@ data ServerStatsData = ServerStatsData
     _qDeleted :: Int,
     _msgSent :: Int,
     _msgRecv :: Int,
-    _activeQueues :: PeriodStatsData RecipientId
+    _activeQueues :: PeriodStatsData RecipientId,
+    _msgSentNtf :: Int,
+    _msgRecvNtf :: Int,
+    _activeQueuesNtf :: PeriodStatsData RecipientId,
+    _qCount :: Int,
+    _msgCount :: Int
   }
+  deriving (Show)
 
 newServerStats :: UTCTime -> STM ServerStats
 newServerStats ts = do
@@ -46,7 +58,12 @@ newServerStats ts = do
   msgSent <- newTVar 0
   msgRecv <- newTVar 0
   activeQueues <- newPeriodStats
-  pure ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, activeQueues}
+  msgSentNtf <- newTVar 0
+  msgRecvNtf <- newTVar 0
+  activeQueuesNtf <- newPeriodStats
+  qCount <- newTVar 0
+  msgCount <- newTVar 0
+  pure ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, activeQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount}
 
 getServerStatsData :: ServerStats -> STM ServerStatsData
 getServerStatsData s = do
@@ -57,7 +74,12 @@ getServerStatsData s = do
   _msgSent <- readTVar $ msgSent s
   _msgRecv <- readTVar $ msgRecv s
   _activeQueues <- getPeriodStatsData $ activeQueues s
-  pure ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _activeQueues}
+  _msgSentNtf <- readTVar $ msgSentNtf s
+  _msgRecvNtf <- readTVar $ msgRecvNtf s
+  _activeQueuesNtf <- getPeriodStatsData $ activeQueuesNtf s
+  _qCount <- readTVar $ qCount s
+  _msgCount <- readTVar $ msgCount s
+  pure ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _activeQueues, _msgSentNtf, _msgRecvNtf, _activeQueuesNtf, _qCount, _msgCount}
 
 setServerStats :: ServerStats -> ServerStatsData -> STM ()
 setServerStats s d = do
@@ -68,9 +90,14 @@ setServerStats s d = do
   writeTVar (msgSent s) $! _msgSent d
   writeTVar (msgRecv s) $! _msgRecv d
   setPeriodStats (activeQueues s) (_activeQueues d)
+  writeTVar (msgSentNtf s) $! _msgSentNtf d
+  writeTVar (msgRecvNtf s) $! _msgRecvNtf d
+  setPeriodStats (activeQueuesNtf s) (_activeQueuesNtf d)
+  writeTVar (qCount s) $! _qCount d
+  writeTVar (msgCount s) $! _qCount d
 
 instance StrEncoding ServerStatsData where
-  strEncode ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _activeQueues} =
+  strEncode ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _msgSentNtf, _msgRecvNtf, _activeQueues, _activeQueuesNtf} =
     B.unlines
       [ "fromTime=" <> strEncode _fromTime,
         "qCreated=" <> strEncode _qCreated,
@@ -78,8 +105,12 @@ instance StrEncoding ServerStatsData where
         "qDeleted=" <> strEncode _qDeleted,
         "msgSent=" <> strEncode _msgSent,
         "msgRecv=" <> strEncode _msgRecv,
+        "msgSentNtf=" <> strEncode _msgSentNtf,
+        "msgRecvNtf=" <> strEncode _msgRecvNtf,
         "activeQueues:",
-        strEncode _activeQueues
+        strEncode _activeQueues,
+        "activeQueuesNtf:",
+        strEncode _activeQueuesNtf
       ]
   strP = do
     _fromTime <- "fromTime=" *> strP <* A.endOfLine
@@ -88,15 +119,21 @@ instance StrEncoding ServerStatsData where
     _qDeleted <- "qDeleted=" *> strP <* A.endOfLine
     _msgSent <- "msgSent=" *> strP <* A.endOfLine
     _msgRecv <- "msgRecv=" *> strP <* A.endOfLine
-    r <- optional ("activeQueues:" <* A.endOfLine)
-    _activeQueues <- case r of
-      Just _ -> strP <* optional A.endOfLine
-      _ -> do
-        _day <- "dayMsgQueues=" *> strP <* A.endOfLine
-        _week <- "weekMsgQueues=" *> strP <* A.endOfLine
-        _month <- "monthMsgQueues=" *> strP <* optional A.endOfLine
-        pure PeriodStatsData {_day, _week, _month}
-    pure ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _activeQueues}
+    _msgSentNtf <- "msgSentNtf=" *> strP <* A.endOfLine <|> pure 0
+    _msgRecvNtf <- "msgRecvNtf=" *> strP <* A.endOfLine <|> pure 0
+    _activeQueues <-
+      optional ("activeQueues:" <* A.endOfLine) >>= \case
+        Just _ -> strP <* optional A.endOfLine
+        _ -> do
+          _day <- "dayMsgQueues=" *> strP <* A.endOfLine
+          _week <- "weekMsgQueues=" *> strP <* A.endOfLine
+          _month <- "monthMsgQueues=" *> strP <* optional A.endOfLine
+          pure PeriodStatsData {_day, _week, _month}
+    _activeQueuesNtf <-
+      optional ("activeQueuesNtf:" <* A.endOfLine) >>= \case
+        Just _ -> strP <* optional A.endOfLine
+        _ -> pure newPeriodStatsData
+    pure ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _msgSentNtf, _msgRecvNtf, _activeQueues, _activeQueuesNtf, _qCount = 0, _msgCount = 0}
 
 data PeriodStats a = PeriodStats
   { day :: TVar (Set a),
@@ -116,6 +153,10 @@ data PeriodStatsData a = PeriodStatsData
     _week :: Set a,
     _month :: Set a
   }
+  deriving (Show)
+
+newPeriodStatsData :: PeriodStatsData a
+newPeriodStatsData = PeriodStatsData {_day = S.empty, _week = S.empty, _month = S.empty}
 
 getPeriodStatsData :: PeriodStats a -> STM (PeriodStatsData a)
 getPeriodStatsData s = do

tests/AgentTests/FunctionalAPITests.hs

@@ -15,6 +15,8 @@ module AgentTests.FunctionalAPITests
     makeConnection,
     exchangeGreetingsMsgId,
     switchComplete,
+    runRight,
+    runRight_,
     get,
     (##>),
     (=##>),
@@ -80,6 +82,15 @@ agentCfgRatchetV1 = agentCfg {e2eEncryptVRange = vr11}
 vr11 :: VersionRange
 vr11 = mkVersionRange 1 1
 
+runRight_ :: ExceptT AgentErrorType IO () -> Expectation
+runRight_ action = runExceptT action `shouldReturn` Right ()
+
+runRight :: ExceptT AgentErrorType IO a -> IO a
+runRight action =
+  runExceptT action >>= \case
+    Right x -> pure x
+    Left e -> error $ "Unexpected error: " <> show e
+
 functionalAPITests :: ATransport -> Spec
 functionalAPITests t = do
   describe "Establishing duplex connection" $
@@ -217,7 +228,7 @@ runTestCfg2 aliceCfg bobCfg baseMsgId runTest = do
 
 runAgentClientTest :: AgentClient -> AgentClient -> AgentMsgId -> IO ()
 runAgentClientTest alice bob baseId = do
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (bobId, qInfo) <- createConnection alice True SCMInvitation Nothing
     aliceId <- joinConnection bob True qInfo "bob's connInfo"
     ("", _, CONF confId _ "bob's connInfo") <- get alice
@@ -247,13 +258,12 @@ runAgentClientTest alice bob baseId = do
     get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
     deleteConnection alice bobId
     liftIO $ noMessages alice "nothing else should be delivered to alice"
-  pure ()
   where
     msgId = subtract baseId
 
 runAgentClientContactTest :: AgentClient -> AgentClient -> AgentMsgId -> IO ()
 runAgentClientContactTest alice bob baseId = do
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (_, qInfo) <- createConnection alice True SCMContact Nothing
     aliceId <- joinConnection bob True qInfo "bob's connInfo"
     ("", _, REQ invId _ "bob's connInfo") <- get alice
@@ -285,7 +295,6 @@ runAgentClientContactTest alice bob baseId = do
     get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
     deleteConnection alice bobId
     liftIO $ noMessages alice "nothing else should be delivered to alice"
-  pure ()
   where
     msgId = subtract baseId
 
@@ -301,7 +310,7 @@ testAsyncInitiatingOffline :: IO ()
 testAsyncInitiatingOffline = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (bobId, cReq) <- createConnection alice True SCMInvitation Nothing
     disconnectAgentClient alice
     aliceId <- joinConnection bob True cReq "bob's connInfo"
@@ -313,13 +322,12 @@ testAsyncInitiatingOffline = do
     get bob ##> ("", aliceId, INFO "alice's connInfo")
     get bob ##> ("", aliceId, CON)
     exchangeGreetings alice' bobId bob aliceId
-  pure ()
 
 testAsyncJoiningOfflineBeforeActivation :: IO ()
 testAsyncJoiningOfflineBeforeActivation = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (bobId, qInfo) <- createConnection alice True SCMInvitation Nothing
     aliceId <- joinConnection bob True qInfo "bob's connInfo"
     disconnectAgentClient bob
@@ -331,13 +339,12 @@ testAsyncJoiningOfflineBeforeActivation = do
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
     get bob' ##> ("", aliceId, CON)
     exchangeGreetings alice bobId bob' aliceId
-  pure ()
 
 testAsyncBothOffline :: IO ()
 testAsyncBothOffline = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (bobId, cReq) <- createConnection alice True SCMInvitation Nothing
     disconnectAgentClient alice
     aliceId <- joinConnection bob True cReq "bob's connInfo"
@@ -352,22 +359,21 @@ testAsyncBothOffline = do
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
     get bob' ##> ("", aliceId, CON)
     exchangeGreetings alice' bobId bob' aliceId
-  pure ()
 
 testAsyncServerOffline :: ATransport -> IO ()
 testAsyncServerOffline t = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
   -- create connection and shutdown the server
-  Right (bobId, cReq) <- withSmpServerStoreLogOn t testPort $ \_ ->
-    runExceptT $ createConnection alice True SCMInvitation Nothing
+  (bobId, cReq) <- withSmpServerStoreLogOn t testPort $ \_ ->
+    runRight $ createConnection alice True SCMInvitation Nothing
   -- connection fails
   Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob True cReq "bob's connInfo"
   ("", "", DOWN srv conns) <- get alice
   srv `shouldBe` testSMPServer
   conns `shouldBe` [bobId]
   -- connection succeeds after server start
-  Right () <- withSmpServerStoreLogOn t testPort $ \_ -> runExceptT $ do
+  withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
     ("", "", UP srv1 conns1) <- get alice
     liftIO $ do
       srv1 `shouldBe` testSMPServer
@@ -379,27 +385,25 @@ testAsyncServerOffline t = do
     get bob ##> ("", aliceId, INFO "alice's connInfo")
     get bob ##> ("", aliceId, CON)
     exchangeGreetings alice bobId bob aliceId
-  pure ()
 
 testAsyncHelloTimeout :: IO ()
 testAsyncHelloTimeout = do
   -- this test would only work if any of the agent is v1, there is no HELLO timeout in v2
   alice <- getSMPAgentClient agentCfgV1 initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2, helloTimeout = 1} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (_, cReq) <- createConnection alice True SCMInvitation Nothing
     disconnectAgentClient alice
     aliceId <- joinConnection bob True cReq "bob's connInfo"
     get bob ##> ("", aliceId, ERR $ CONN NOT_ACCEPTED)
-  pure ()
 
 testDuplicateMessage :: ATransport -> IO ()
 testDuplicateMessage t = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
   (aliceId, bobId, bob1) <- withSmpServerStoreMsgLogOn t testPort $ \_ -> do
-    Right (aliceId, bobId) <- runExceptT $ makeConnection alice bob
-    Right () <- runExceptT $ do
+    (aliceId, bobId) <- runRight $ makeConnection alice bob
+    runRight_ $ do
       4 <- sendMessage alice bobId SMP.noMsgFlags "hello"
       get alice ##> ("", bobId, SENT 4)
       get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
@@ -407,7 +411,7 @@ testDuplicateMessage t = do
 
     -- if the agent user did not send ACK, the message will be delivered again
     bob1 <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-    Right () <- runExceptT $ do
+    runRight_ $ do
       subscribeConnection bob1 aliceId
       get bob1 =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
       ackMessage bob1 aliceId 4
@@ -419,7 +423,7 @@ testDuplicateMessage t = do
 
   get alice =##> \case ("", "", DOWN _ [c]) -> c == bobId; _ -> False
   get bob1 =##> \case ("", "", DOWN _ [c]) -> c == aliceId; _ -> False
-  -- commenting two lines below and uncommenting further two lines would also pass,
+  -- commenting two lines below and uncommenting further two lines would also runRight_,
   -- it is the scenario tested above, when the message was not acknowledged by the user
   threadDelay 200000
   Left (BROKER _ TIMEOUT) <- runExceptT $ ackMessage bob1 aliceId 5
@@ -431,7 +435,7 @@ testDuplicateMessage t = do
   bob2 <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
 
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
-    Right () <- runExceptT $ do
+    runRight_ $ do
       subscribeConnection bob2 aliceId
       subscribeConnection alice2 bobId
       -- get bob2 =##> \case ("", c, Msg "hello 2") -> c == aliceId; _ -> False
@@ -440,7 +444,6 @@ testDuplicateMessage t = do
       6 <- sendMessage alice2 bobId SMP.noMsgFlags "hello 3"
       get alice2 ##> ("", bobId, SENT 6)
       get bob2 =##> \case ("", c, Msg "hello 3") -> c == aliceId; _ -> False
-    pure ()
 
 makeConnection :: AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnection alice bob = do
@@ -458,10 +461,9 @@ testInactiveClientDisconnected t = do
   let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
     alice <- getSMPAgentClient agentCfg initAgentServers
-    Right () <- runExceptT $ do
+    runRight_ $ do
       (connId, _cReq) <- createConnection alice True SCMInvitation Nothing
       get alice ##> ("", "", DOWN testSMPServer [connId])
-    pure ()
 
 testActiveClientNotDisconnected :: ATransport -> IO ()
 testActiveClientNotDisconnected t = do
@@ -469,10 +471,9 @@ testActiveClientNotDisconnected t = do
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
     alice <- getSMPAgentClient agentCfg initAgentServers
     ts <- getSystemTime
-    Right () <- runExceptT $ do
+    runRight_ $ do
       (connId, _cReq) <- createConnection alice True SCMInvitation Nothing
       keepSubscribing alice connId ts
-    pure ()
   where
     keepSubscribing :: AgentClient -> ConnId -> SystemTime -> ExceptT AgentErrorType IO ()
     keepSubscribing alice connId ts = do
@@ -495,7 +496,7 @@ testSuspendingAgent :: IO ()
 testSuspendingAgent = do
   a <- getSMPAgentClient agentCfg initAgentServers
   b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (aId, bId) <- makeConnection a b
     4 <- sendMessage a bId SMP.noMsgFlags "hello"
     get a ##> ("", bId, SENT 4)
@@ -508,13 +509,12 @@ testSuspendingAgent = do
     Nothing <- 100000 `timeout` get b
     activateAgent b
     get b =##> \case ("", c, Msg "hello 2") -> c == aId; _ -> False
-  pure ()
 
 testSuspendingAgentCompleteSending :: ATransport -> IO ()
 testSuspendingAgentCompleteSending t = do
   a <- getSMPAgentClient agentCfg initAgentServers
   b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runExceptT $ do
+  (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
     (aId, bId) <- makeConnection a b
     4 <- sendMessage a bId SMP.noMsgFlags "hello"
     get a ##> ("", bId, SENT 4)
@@ -522,7 +522,7 @@ testSuspendingAgentCompleteSending t = do
     ackMessage b aId 4
     pure (aId, bId)
 
-  Right () <- runExceptT $ do
+  runRight_ $ do
     ("", "", DOWN {}) <- get a
     ("", "", DOWN {}) <- get b
     5 <- sendMessage b aId SMP.noMsgFlags "hello too"
@@ -530,7 +530,7 @@ testSuspendingAgentCompleteSending t = do
     liftIO $ threadDelay 100000
     suspendAgent b 5000000
 
-  Right () <- withSmpServerStoreLogOn t testPort $ \_ -> runExceptT $ do
+  withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
     get b =##> \case ("", c, SENT 5) -> c == aId; ("", "", UP {}) -> True; _ -> False
     get b =##> \case ("", c, SENT 5) -> c == aId; ("", "", UP {}) -> True; _ -> False
     get b =##> \case ("", c, SENT 6) -> c == aId; ("", "", UP {}) -> True; _ -> False
@@ -544,13 +544,11 @@ testSuspendingAgentCompleteSending t = do
     get a =##> \case ("", c, Msg "how are you?") -> c == bId; _ -> False
     ackMessage a bId 6
 
-  pure ()
-
 testSuspendingAgentTimeout :: ATransport -> IO ()
 testSuspendingAgentTimeout t = do
   a <- getSMPAgentClient agentCfg initAgentServers
   b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right (aId, _) <- withSmpServer t . runExceptT $ do
+  (aId, _) <- withSmpServer t . runRight $ do
     (aId, bId) <- makeConnection a b
     4 <- sendMessage a bId SMP.noMsgFlags "hello"
     get a ##> ("", bId, SENT 4)
@@ -558,7 +556,7 @@ testSuspendingAgentTimeout t = do
     ackMessage b aId 4
     pure (aId, bId)
 
-  Right () <- runExceptT $ do
+  runRight_ $ do
     ("", "", DOWN {}) <- get a
     ("", "", DOWN {}) <- get b
     5 <- sendMessage b aId SMP.noMsgFlags "hello too"
@@ -567,13 +565,11 @@ testSuspendingAgentTimeout t = do
     ("", "", SUSPENDED) <- get b
     pure ()
 
-  pure ()
-
 testBatchedSubscriptions :: ATransport -> IO ()
 testBatchedSubscriptions t = do
   a <- getSMPAgentClient agentCfg initAgentServers2
   b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers2
-  Right conns <- runServers $ do
+  conns <- runServers $ do
     conns <- forM [1 .. 200 :: Int] . const $ makeConnection a b
     forM_ conns $ \(aId, bId) -> exchangeGreetings a bId b aId
     forM_ (take 10 conns) $ \(aId, bId) -> do
@@ -585,7 +581,7 @@ testBatchedSubscriptions t = do
   ("", "", DOWN {}) <- get a
   ("", "", DOWN {}) <- get b
   ("", "", DOWN {}) <- get b
-  Right () <- runServers $ do
+  runServers $ do
     ("", "", UP {}) <- get a
     ("", "", UP {}) <- get a
     ("", "", UP {}) <- get b
@@ -594,7 +590,6 @@ testBatchedSubscriptions t = do
     subscribe a $ map snd conns
     subscribe b $ map fst conns
     forM_ (drop 10 conns) $ \(aId, bId) -> exchangeGreetingsMsgId 6 a bId b aId
-  pure ()
   where
     subscribe :: AgentClient -> [ConnId] -> ExceptT AgentErrorType IO ()
     subscribe c cs = do
@@ -604,13 +599,11 @@ testBatchedSubscriptions t = do
         all (== Right ()) (M.withoutKeys r dc) `shouldBe` True
         all (== Left (CONN NOT_FOUND)) (M.restrictKeys r dc) `shouldBe` True
         M.keys r `shouldMatchList` cs
-    runServers :: ExceptT AgentErrorType IO a -> IO (Either AgentErrorType a)
+    runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
       withSmpServerStoreLogOn t testPort $ \t1 -> do
-        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 -> do
-          res <- runExceptT a
-          killThread t2
-          pure res
+        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+          runRight a `finally` killThread t2
         killThread t1
         pure res
 
@@ -618,7 +611,7 @@ testAsyncCommands :: IO ()
 testAsyncCommands = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     bobId <- createConnectionAsync alice "1" True SCMInvitation
     ("1", bobId', INV (ACR _ qInfo)) <- get alice
     liftIO $ bobId' `shouldBe` bobId
@@ -655,7 +648,6 @@ testAsyncCommands = do
     deleteConnectionAsync alice "8" bobId
     ("8", _, OK) <- get alice
     liftIO $ noMessages alice "nothing else should be delivered to alice"
-  pure ()
   where
     baseId = 3
     msgId = subtract baseId
@@ -663,22 +655,21 @@ testAsyncCommands = do
 testAsyncCommandsRestore :: ATransport -> IO ()
 testAsyncCommandsRestore t = do
   alice <- getSMPAgentClient agentCfg initAgentServers
-  Right bobId <- runExceptT $ createConnectionAsync alice "1" True SCMInvitation
+  bobId <- runRight $ createConnectionAsync alice "1" True SCMInvitation
   liftIO $ noMessages alice "alice doesn't receive INV because server is down"
   disconnectAgentClient alice
   alice' <- liftIO $ getSMPAgentClient agentCfg initAgentServers
   withSmpServerStoreLogOn t testPort $ \_ -> do
-    Right () <- runExceptT $ do
+    runRight_ $ do
       subscribeConnection alice' bobId
       ("1", _, INV _) <- get alice'
       pure ()
-    pure ()
 
 testAcceptContactAsync :: IO ()
 testAcceptContactAsync = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (_, qInfo) <- createConnection alice True SCMContact Nothing
     aliceId <- joinConnection bob True qInfo "bob's connInfo"
     ("", _, REQ invId _ "bob's connInfo") <- get alice
@@ -712,7 +703,6 @@ testAcceptContactAsync = do
     get bob ##> ("", aliceId, MERR (baseId + 5) (SMP AUTH))
     deleteConnection alice bobId
     liftIO $ noMessages alice "nothing else should be delivered to alice"
-  pure ()
   where
     baseId = 3
     msgId = subtract baseId
@@ -721,13 +711,12 @@ testSwitchConnection :: InitialAgentServers -> IO ()
 testSwitchConnection servers = do
   a <- getSMPAgentClient agentCfg servers
   b <- getSMPAgentClient agentCfg {database = testDB2, initialClientId = 1} servers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
     switchConnectionAsync a "" bId
     switchComplete a bId b aId
     exchangeGreetingsMsgId 10 a bId b aId
-  pure ()
 
 switchComplete :: AgentClient -> ByteString -> AgentClient -> ByteString -> ExceptT AgentErrorType IO ()
 switchComplete a bId b aId = do
@@ -749,12 +738,12 @@ phase c connId d p =
       ERR (AGENT A_DUPLICATE) -> phase c connId d p
       r -> do
         liftIO . putStrLn $ "expected: " <> show p <> ", received: " <> show r
-        SWITCH _ _ _ <- pure r
+        SWITCH {} <- pure r
         pure ()
 
 testSwitchAsync :: InitialAgentServers -> IO ()
 testSwitchAsync servers = do
-  Right (aId, bId) <- withA $ \a -> withB $ \b -> runExceptT $ do
+  (aId, bId) <- withA $ \a -> withB $ \b -> runRight $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
     pure (aId, bId)
@@ -769,22 +758,20 @@ testSwitchAsync servers = do
     phase b aId QDSnd SPConfirmed
     phase b aId QDSnd SPCompleted
   withA' $ \a -> phase a bId QDRcv SPCompleted
-  Right () <- withA $ \a -> withB $ \b -> runExceptT $ do
+  withA $ \a -> withB $ \b -> runRight_ $ do
     subscribeConnection a bId
     subscribeConnection b aId
     exchangeGreetingsMsgId 10 a bId b aId
-  pure ()
   where
     withAgent :: AgentConfig -> (AgentClient -> IO a) -> IO a
     withAgent cfg' = bracket (getSMPAgentClient cfg' servers) disconnectAgentClient
     session :: (forall a. (AgentClient -> IO a) -> IO a) -> ConnId -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO ()
-    session withC connId a = do
-      Right () <- withC $ \c -> runExceptT $ do
+    session withC connId a =
+      withC $ \c -> runRight_ $ do
         subscribeConnection c connId
         r <- a c
         liftIO $ threadDelay 500000
         pure r
-      pure ()
     withA = withAgent agentCfg
     withB = withAgent agentCfg {database = testDB2, initialClientId = 1}
 
@@ -792,7 +779,7 @@ testSwitchDelete :: InitialAgentServers -> IO ()
 testSwitchDelete servers = do
   a <- getSMPAgentClient agentCfg servers
   b <- getSMPAgentClient agentCfg {database = testDB2, initialClientId = 1} servers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
     disconnectAgentClient b
@@ -801,13 +788,12 @@ testSwitchDelete servers = do
     deleteConnectionAsync a "1" bId
     ("1", bId', OK) <- get a
     liftIO $ bId `shouldBe` bId'
-  pure ()
 
 testCreateQueueAuth :: (Maybe BasicAuth, Version) -> (Maybe BasicAuth, Version) -> IO Int
 testCreateQueueAuth clnt1 clnt2 = do
   a <- getClient clnt1
   b <- getClient clnt2
-  Right created <- runExceptT $ do
+  runRight $ do
     tryError (createConnection a True SCMInvitation Nothing) >>= \case
       Left (SMP AUTH) -> pure 0
       Left e -> throwError e
@@ -823,7 +809,6 @@ testCreateQueueAuth clnt1 clnt2 = do
             get b ##> ("", aId, CON)
             exchangeGreetings a bId b aId
             pure 2
-  pure created
   where
     getClient (clntAuth, clntVersion) =
       let servers = initAgentServers {smp = [ProtoServerWithAuth testSMPServer clntAuth]}
@@ -834,19 +819,17 @@ testSMPServerConnectionTest :: ATransport -> Maybe BasicAuth -> SMPServerWithAut
 testSMPServerConnectionTest t newQueueBasicAuth srv =
   withSmpServerConfigOn t cfg {newQueueBasicAuth} testPort2 $ \_ -> do
     a <- getSMPAgentClient agentCfg initAgentServers -- initially passed server is not running
-    Right r <- runExceptT $ testSMPServerConnection a srv
-    pure r
+    runRight $ testSMPServerConnection a srv
 
 testRatchetAdHash :: IO ()
 testRatchetAdHash = do
   a <- getSMPAgentClient agentCfg initAgentServers
   b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (aId, bId) <- makeConnection a b
     ad1 <- getConnectionRatchetAdHash a bId
     ad2 <- getConnectionRatchetAdHash b aId
     liftIO $ ad1 `shouldBe` ad2
-  pure ()
 
 exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
 exchangeGreetings = exchangeGreetingsMsgId 4

tests/AgentTests/NotificationTests.hs

@@ -8,7 +8,7 @@
 module AgentTests.NotificationTests where
 
 -- import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging)
-import AgentTests.FunctionalAPITests (exchangeGreetingsMsgId, get, makeConnection, switchComplete, testServerMatrix2, (##>), (=##>), pattern Msg)
+import AgentTests.FunctionalAPITests (exchangeGreetingsMsgId, get, makeConnection, runRight, runRight_, switchComplete, testServerMatrix2, (##>), (=##>), pattern Msg)
 import Control.Concurrent (killThread, threadDelay)
 import Control.Monad.Except
 import qualified Data.Aeson as J
@@ -91,7 +91,7 @@ notificationTests t =
 testNotificationToken :: APNSMockServer -> IO ()
 testNotificationToken APNSMockServer {apnsQ} = do
   a <- getSMPAgentClient agentCfg initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
@@ -105,7 +105,6 @@ testNotificationToken APNSMockServer {apnsQ} = do
     -- agent deleted this token
     Left (CMD PROHIBITED) <- tryE $ checkNtfToken a tkn
     pure ()
-  pure ()
 
 (.->) :: J.Value -> J.Key -> ExceptT AgentErrorType IO ByteString
 v .-> key = do
@@ -120,7 +119,7 @@ testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
   -- setLogLevel LogError -- LogDebug
   -- withGlobalLogging logCfg $ do
   a <- getSMPAgentClient agentCfg initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
@@ -138,7 +137,6 @@ testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
     verifyNtfToken a tkn nonce verification
     NTActive <- checkNtfToken a tkn
     pure ()
-  pure ()
 
 testNtfTokenSecondRegistration :: APNSMockServer -> IO ()
 testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
@@ -146,7 +144,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
   -- withGlobalLogging logCfg $ do
   a <- getSMPAgentClient agentCfg initAgentServers
   a' <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
@@ -175,13 +173,12 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
     -- and the second is active
     NTActive <- checkNtfToken a' tkn
     pure ()
-  pure ()
 
 testNtfTokenServerRestart :: ATransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
   a <- getSMPAgentClient agentCfg initAgentServers
   let tkn = DeviceToken PPApnsTest "abcd"
-  Right ntfData <- withNtfServer t . runExceptT $ do
+  ntfData <- withNtfServer t . runRight $ do
     NTRegistered <- registerNtfToken a tkn NMPeriodic
     APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
       atomically $ readTBQueue apnsQ
@@ -193,7 +190,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
   a' <- getSMPAgentClient agentCfg initAgentServers
   -- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token,
   -- so that repeat verification happens without restarting the clients, when notification arrives
-  Right () <- withNtfServer t . runExceptT $ do
+  withNtfServer t . runRight_ $ do
     verification <- ntfData .-> "verification"
     nonce <- C.cbNonce <$> ntfData .-> "nonce"
     Left (NTF AUTH) <- tryE $ verifyNtfToken a' tkn nonce verification
@@ -205,13 +202,12 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
     verifyNtfToken a' tkn nonce' verification'
     NTActive <- checkNtfToken a' tkn
     pure ()
-  pure ()
 
 testNotificationSubscriptionExistingConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right (bobId, aliceId, nonce, message) <- runExceptT $ do
+  (bobId, aliceId, nonce, message) <- runRight $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice True SCMInvitation Nothing
     aliceId <- joinConnection bob True qInfo "bob's connInfo"
@@ -243,12 +239,12 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
 
   -- aliceNtf client doesn't have subscription and is allowed to get notification message
   aliceNtf <- getSMPAgentClient agentCfg initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (_, [SMPMsgMeta {msgFlags = MsgFlags True}]) <- getNotificationMessage aliceNtf nonce message
     pure ()
   disconnectAgentClient aliceNtf
 
-  Right () <- runExceptT $ do
+  runRight_ $ do
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
     ackMessage alice bobId $ baseId + 1
     -- delete notification subscription
@@ -259,7 +255,6 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 2)
     -- no notifications should follow
     noNotification apnsQ
-  pure ()
   where
     baseId = 3
     msgId = subtract baseId
@@ -268,7 +263,7 @@ testNotificationSubscriptionNewConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     -- alice registers notification token
     DeviceToken {} <- registerTestToken alice "abcd" NMInstant apnsQ
     -- bob registers notification token
@@ -303,7 +298,6 @@ testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
     ackMessage bob aliceId $ baseId + 2
     -- no unexpected notifications should follow
     noNotification apnsQ
-  pure ()
   where
     baseId = 3
     msgId = subtract baseId
@@ -325,7 +319,7 @@ testChangeNotificationsMode :: APNSMockServer -> IO ()
 testChangeNotificationsMode APNSMockServer {apnsQ} = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice True SCMInvitation Nothing
     aliceId <- joinConnection bob True qInfo "bob's connInfo"
@@ -381,7 +375,6 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
     ackMessage alice bobId $ baseId + 5
     -- no notifications should follow
     noNotification apnsQ
-  pure ()
   where
     baseId = 3
     msgId = subtract baseId
@@ -390,7 +383,7 @@ testChangeToken :: APNSMockServer -> IO ()
 testChangeToken APNSMockServer {apnsQ} = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right (aliceId, bobId) <- runExceptT $ do
+  (aliceId, bobId) <- runRight $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice True SCMInvitation Nothing
     aliceId <- joinConnection bob True qInfo "bob's connInfo"
@@ -412,7 +405,7 @@ testChangeToken APNSMockServer {apnsQ} = do
   disconnectAgentClient alice
 
   alice1 <- getSMPAgentClient agentCfg initAgentServers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     subscribeConnection alice1 bobId
     -- change notification token
     void $ registerTestToken alice1 "bcde" NMInstant apnsQ
@@ -425,7 +418,6 @@ testChangeToken APNSMockServer {apnsQ} = do
     ackMessage alice1 bobId $ baseId + 2
     -- no notifications should follow
     noNotification apnsQ
-  pure ()
   where
     baseId = 3
     msgId = subtract baseId
@@ -434,7 +426,7 @@ testNotificationsStoreLog :: ATransport -> APNSMockServer -> IO ()
 testNotificationsStoreLog t APNSMockServer {apnsQ} = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right (aliceId, bobId) <- withNtfServerStoreLog t $ \threadId -> runExceptT $ do
+  (aliceId, bobId) <- withNtfServerStoreLog t $ \threadId -> runRight $ do
     (aliceId, bobId) <- makeConnection alice bob
     _ <- registerTestToken alice "abcd" NMInstant apnsQ
     liftIO $ threadDelay 250000
@@ -448,20 +440,19 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do
 
   liftIO $ threadDelay 250000
 
-  Right () <- withNtfServerStoreLog t $ \threadId -> runExceptT $ do
+  withNtfServerStoreLog t $ \threadId -> runRight_ $ do
     liftIO $ threadDelay 250000
     5 <- sendMessage bob aliceId (SMP.MsgFlags True) "hello again"
     get bob ##> ("", aliceId, SENT 5)
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
-  pure ()
 
 testNotificationsSMPRestart :: ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
   alice <- getSMPAgentClient agentCfg initAgentServers
   bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
-  Right (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \threadId -> runExceptT $ do
+  (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \threadId -> runRight $ do
     (aliceId, bobId) <- makeConnection alice bob
     _ <- registerTestToken alice "abcd" NMInstant apnsQ
     liftIO $ threadDelay 250000
@@ -473,11 +464,11 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
     liftIO $ killThread threadId
     pure (aliceId, bobId)
 
-  Right () <- runExceptT $ do
+  runRight_ $ do
     get alice =##> \case ("", "", DOWN _ [c]) -> c == bobId; _ -> False
     get bob =##> \case ("", "", DOWN _ [c]) -> c == aliceId; _ -> False
 
-  Right () <- withSmpServerStoreLogOn t testPort $ \threadId -> runExceptT $ do
+  withSmpServerStoreLogOn t testPort $ \threadId -> runRight_ $ do
     get alice =##> \case ("", "", UP _ [c]) -> c == bobId; _ -> False
     get bob =##> \case ("", "", UP _ [c]) -> c == aliceId; _ -> False
     liftIO $ threadDelay 1000000
@@ -486,13 +477,12 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
     _ <- messageNotificationData alice apnsQ
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
-  pure ()
 
 testSwitchNotifications :: InitialAgentServers -> APNSMockServer -> IO ()
 testSwitchNotifications servers APNSMockServer {apnsQ} = do
   a <- getSMPAgentClient agentCfg servers
   b <- getSMPAgentClient agentCfg {database = testDB2, initialClientId = 1} servers
-  Right () <- runExceptT $ do
+  runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
     _ <- registerTestToken a "abcd" NMInstant apnsQ
@@ -508,7 +498,6 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do
     switchComplete a bId b aId
     liftIO $ threadDelay 500000
     testMessage "hello again"
-  pure ()
 
 messageNotification :: TBQueue APNSMockRequest -> ExceptT AgentErrorType IO (C.CbNonce, ByteString)
 messageNotification apnsQ = do

tests/ServerTests.hs

@@ -11,6 +11,7 @@
 
 module ServerTests where
 
+import AgentTests.NotificationTests (removeFileIfExists)
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
@@ -20,6 +21,7 @@ import Data.Bifunctor (first)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import qualified Data.Set as S
 import SMPClient
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
@@ -28,6 +30,7 @@ import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
+import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..))
 import Simplex.Messaging.Transport
 import System.Directory (removeFile)
 import System.TimeIt (timeItT)
@@ -52,7 +55,7 @@ serverTests t@(ATransport t') = do
     describe "Exceeding queue quota" $ testExceedQueueQuota t'
   describe "Store log" $ testWithStoreLog t
   describe "Restore messages" $ testRestoreMessages t
-  describe "Restore messages (v2)" $ testRestoreMessagesV2 t
+  describe "Restore messages (old / v2)" $ testRestoreMessagesV2 t
   describe "Timing of AUTH error" $ testTiming t
   describe "Message notifications" $ testMessageNotifications t
   describe "Message expiration" $ do
@@ -605,6 +608,10 @@ logSize f =
 testRestoreMessages :: ATransport -> Spec
 testRestoreMessages at@(ATransport t) =
   it "should store messages on exit and restore on start" $ do
+    removeFileIfExists testStoreLogFile
+    removeFileIfExists testStoreMsgsFile
+    removeFileIfExists testServerStatsBackupFile
+
     (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519
     recipientId <- newTVarIO ""
     recipientKey <- newTVarIO Nothing
@@ -628,13 +635,19 @@ testRestoreMessages at@(ATransport t) =
       Resp "2" _ OK <- signSendRecv h sKey ("2", sId, _SEND "hello 2")
       Resp "3" _ OK <- signSendRecv h sKey ("3", sId, _SEND "hello 3")
       Resp "4" _ OK <- signSendRecv h sKey ("4", sId, _SEND "hello 4")
+      Resp "5" _ OK <- signSendRecv h sKey ("5", sId, _SEND "hello 5")
+      Resp "6" _ (ERR QUOTA) <- signSendRecv h sKey ("6", sId, _SEND "hello 6")
       pure ()
 
+    rId <- readTVarIO recipientId
+
     logSize testStoreLogFile `shouldReturn` 2
-    logSize testStoreMsgsFile `shouldReturn` 3
+    logSize testStoreMsgsFile `shouldReturn` 5
+    logSize testServerStatsBackupFile `shouldReturn` 16
+    Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
+    checkStats stats1 [rId] 5 1
 
     withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
-      rId <- readTVarIO recipientId
       Just rKey <- readTVarIO recipientKey
       Just dh <- readTVarIO dhShared
       let dec = decryptMsgV3 dh
@@ -647,21 +660,33 @@ testRestoreMessages at@(ATransport t) =
 
     logSize testStoreLogFile `shouldReturn` 1
     -- the last message is not removed because it was not ACK'd
-    logSize testStoreMsgsFile `shouldReturn` 1
+    logSize testStoreMsgsFile `shouldReturn` 3
+    logSize testServerStatsBackupFile `shouldReturn` 16
+    Right stats2 <- strDecode <$> B.readFile testServerStatsBackupFile
+    checkStats stats2 [rId] 5 3
 
     withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
-      rId <- readTVarIO recipientId
       Just rKey <- readTVarIO recipientKey
       Just dh <- readTVarIO dhShared
+      let dec = decryptMsgV3 dh
       Resp "4" _ (Msg mId4 msg4) <- signSendRecv h rKey ("4", rId, SUB)
-      Resp "5" _ OK <- signSendRecv h rKey ("5", rId, ACK mId4)
-      (decryptMsgV3 dh mId4 msg4, Right "hello 4") #== "restored message delivered"
+      (dec mId4 msg4, Right "hello 4") #== "restored message delivered"
+      Resp "5" _ (Msg mId5 msg5) <- signSendRecv h rKey ("5", rId, ACK mId4)
+      (dec mId5 msg5, Right "hello 5") #== "restored message delivered"
+      Resp "6" _ (Msg mId6 msg6) <- signSendRecv h rKey ("6", rId, ACK mId5)
+      (dec mId6 msg6, Left "ClientRcvMsgQuota") #== "restored message delivered"
+      Resp "7" _ OK <- signSendRecv h rKey ("7", rId, ACK mId6)
+      pure ()
 
     logSize testStoreLogFile `shouldReturn` 1
     logSize testStoreMsgsFile `shouldReturn` 0
+    logSize testServerStatsBackupFile `shouldReturn` 16
+    Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
+    checkStats stats3 [rId] 5 5
 
     removeFile testStoreLogFile
     removeFile testStoreMsgsFile
+    removeFile testServerStatsBackupFile
   where
     runTest :: Transport c => TProxy c -> (THandle c -> IO ()) -> ThreadId -> Expectation
     runTest _ test' server = do
@@ -671,6 +696,20 @@ testRestoreMessages at@(ATransport t) =
     runClient :: Transport c => TProxy c -> (THandle c -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
+checkStats :: ServerStatsData -> [RecipientId] -> Int -> Int -> Expectation
+checkStats s qs sent received = do
+  _qCreated s `shouldBe` length qs
+  _qSecured s `shouldBe` length qs
+  _qDeleted s `shouldBe` 0
+  _msgSent s `shouldBe` sent
+  _msgRecv s `shouldBe` received
+  _msgSentNtf s `shouldBe` 0
+  _msgRecvNtf s `shouldBe` 0
+  let PeriodStatsData {_day, _week, _month} = _activeQueues s
+  S.toList _day `shouldBe` qs
+  S.toList _week `shouldBe` qs
+  S.toList _month `shouldBe` qs
+
 testRestoreMessagesV2 :: ATransport -> Spec
 testRestoreMessagesV2 at@(ATransport t) =
   it "should store messages on exit and restore on start" $ do

tests/Test.hs

@@ -1,8 +1,8 @@
 {-# LANGUAGE TypeApplications #-}
 
 import AgentTests (agentTests)
--- import Control.Logger.Simple
 import CLITests
+import Control.Logger.Simple
 import CoreTests.CryptoTests
 import CoreTests.EncodingTests
 import CoreTests.ProtocolErrorTests
@@ -16,26 +16,26 @@ import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive)
 import System.Environment (setEnv)
 import Test.Hspec
 
--- logCfg :: LogConfig
--- logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
+logCfg :: LogConfig
+logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 
 main :: IO ()
 main = do
-  -- setLogLevel LogInfo -- LogError
-  -- withGlobalLogging logCfg $ do
-  createDirectoryIfMissing False "tests/tmp"
-  setEnv "APNS_KEY_ID" "H82WD9K9AQ"
-  setEnv "APNS_KEY_FILE" "./tests/fixtures/AuthKey_H82WD9K9AQ.p8"
-  hspec $ do
-    describe "Core tests" $ do
-      describe "Encoding tests" encodingTests
-      describe "Protocol error tests" protocolErrorTests
-      describe "Version range" versionRangeTests
-      describe "Encryption tests" cryptoTests
-      describe "Retry interval tests" retryIntervalTests
-    describe "SMP server via TLS" $ serverTests (transport @TLS)
-    describe "SMP server via WebSockets" $ serverTests (transport @WS)
-    describe "Notifications server" $ ntfServerTests (transport @TLS)
-    describe "SMP client agent" $ agentTests (transport @TLS)
-    describe "Server CLIs" cliTests
-  removeDirectoryRecursive "tests/tmp"
+  setLogLevel LogError -- LogInfo
+  withGlobalLogging logCfg $ do
+    createDirectoryIfMissing False "tests/tmp"
+    setEnv "APNS_KEY_ID" "H82WD9K9AQ"
+    setEnv "APNS_KEY_FILE" "./tests/fixtures/AuthKey_H82WD9K9AQ.p8"
+    hspec $ do
+      describe "Core tests" $ do
+        describe "Encoding tests" encodingTests
+        describe "Protocol error tests" protocolErrorTests
+        describe "Version range" versionRangeTests
+        describe "Encryption tests" cryptoTests
+        describe "Retry interval tests" retryIntervalTests
+      describe "SMP server via TLS" $ serverTests (transport @TLS)
+      describe "SMP server via WebSockets" $ serverTests (transport @WS)
+      describe "Notifications server" $ ntfServerTests (transport @TLS)
+      describe "SMP client agent" $ agentTests (transport @TLS)
+      describe "Server CLIs" cliTests
+    removeDirectoryRecursive "tests/tmp"