4.3.0

* refactor tests to improve errors

* fix test descriptions

* 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.1
+version: 4.3.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        4.2.1
+version:        4.3.0
 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.hs

@@ -905,7 +905,7 @@ runCommandProcessing c@AgentClient {subQ} server_ = do
           atomically $ do
             srvs <- readTVar $ smpServers c
             let used' = if length used + 1 >= L.length srvs then initUsed else srv : used
-            writeTVar usedSrvs used'
+            writeTVar usedSrvs $! used'
           action srvAuth
 -- ^ ^ ^ async command processing /
 

src/Simplex/Messaging/Agent/Client.hs

@@ -607,7 +607,7 @@ subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> m ()
 subscribeQueue c rq@RcvQueue {connId, server, rcvPrivateKey, rcvId} = do
   whenM (atomically . TM.member (server, rcvId) $ getMsgLocks c) . throwError $ CMD PROHIBITED
   atomically $ do
-    modifyTVar (subscrConns c) $ S.insert connId
+    modifyTVar' (subscrConns c) $ S.insert connId
     RQ.addQueue rq $ pendingSubs c
   withLogClient c server rcvId "SUB" $ \smp ->
     liftIO (runExceptT (subscribeSMPQueue smp rcvPrivateKey rcvId) >>= processSubResult c rq)
@@ -644,7 +644,7 @@ subscribeQueues :: AgentMonad m => AgentClient -> SMPServer -> [RcvQueue] -> m (
 subscribeQueues c srv qs = do
   (errs, qs_) <- partitionEithers <$> mapM checkQueue qs
   forM_ qs_ $ \rq@RcvQueue {connId} -> atomically $ do
-    modifyTVar (subscrConns c) $ S.insert connId
+    modifyTVar' (subscrConns c) $ S.insert connId
     RQ.addQueue rq $ pendingSubs c
   case L.nonEmpty qs_ of
     Just qs' -> do
@@ -671,7 +671,7 @@ subscribeQueues c srv qs = do
 
 addSubscription :: MonadIO m => AgentClient -> RcvQueue -> m ()
 addSubscription c rq@RcvQueue {connId} = atomically $ do
-  modifyTVar (subscrConns c) $ S.insert connId
+  modifyTVar' (subscrConns c) $ S.insert connId
   RQ.addQueue rq $ activeSubs c
   RQ.deleteQueue rq $ pendingSubs c
 
@@ -680,7 +680,7 @@ hasActiveSubscription c connId = RQ.hasConn connId $ activeSubs c
 
 removeSubscription :: AgentClient -> ConnId -> STM ()
 removeSubscription c connId = do
-  modifyTVar (subscrConns c) $ S.delete connId
+  modifyTVar' (subscrConns c) $ S.delete connId
   RQ.deleteConn connId $ activeSubs c
   RQ.deleteConn connId $ pendingSubs c
 
@@ -945,7 +945,7 @@ storeError = \case
 incStat :: AgentClient -> Int -> AgentStatsKey -> STM ()
 incStat AgentClient {agentStats} n k = do
   TM.lookup k agentStats >>= \case
-    Just v -> modifyTVar v (+ n)
+    Just v -> modifyTVar' v (+ n)
     _ -> newTVar n >>= \v -> TM.insert k v agentStats
 
 incClientStat :: AgentClient -> ProtocolClient msg -> ByteString -> ByteString -> IO ()

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/Crypto.hs

@@ -964,11 +964,7 @@ pseudoRandomCbNonce :: TVar ChaChaDRG -> STM CbNonce
 pseudoRandomCbNonce gVar = CbNonce <$> pseudoRandomBytes 24 gVar
 
 pseudoRandomBytes :: Int -> TVar ChaChaDRG -> STM ByteString
-pseudoRandomBytes n gVar = do
-  g <- readTVar gVar
-  let (bytes, g') = randomBytesGenerate n g
-  writeTVar gVar g'
-  return bytes
+pseudoRandomBytes n gVar = stateTVar gVar $ randomBytesGenerate n
 
 instance Encoding CbNonce where
   smpEncode = unCbNonce

src/Simplex/Messaging/Notifications/Server.hs

@@ -548,7 +548,7 @@ withNtfLog action = liftIO . mapM_ action =<< asks storeLog
 incNtfStat :: (NtfServerStats -> TVar Int) -> M ()
 incNtfStat statSel = do
   stats <- asks serverStats
-  atomically $ modifyTVar (statSel stats) (+ 1)
+  atomically $ modifyTVar' (statSel stats) (+ 1)
 
 saveServerStats :: M ()
 saveServerStats =

src/Simplex/Messaging/Notifications/Server/Stats.hs

@@ -70,14 +70,14 @@ getNtfServerStatsData s = do
 
 setNtfServerStats :: NtfServerStats -> NtfServerStatsData -> STM ()
 setNtfServerStats s d = do
-  writeTVar (fromTime (s :: NtfServerStats)) (_fromTime (d :: NtfServerStatsData))
-  writeTVar (tknCreated s) (_tknCreated d)
-  writeTVar (tknVerified s) (_tknVerified d)
-  writeTVar (tknDeleted s) (_tknDeleted d)
-  writeTVar (subCreated s) (_subCreated d)
-  writeTVar (subDeleted s) (_subDeleted d)
-  writeTVar (ntfReceived s) (_ntfReceived d)
-  writeTVar (ntfDelivered s) (_ntfDelivered d)
+  writeTVar (fromTime (s :: NtfServerStats)) $! _fromTime (d :: NtfServerStatsData)
+  writeTVar (tknCreated s) $! _tknCreated d
+  writeTVar (tknVerified s) $! _tknVerified d
+  writeTVar (tknDeleted s) $! _tknDeleted d
+  writeTVar (subCreated s) $! _subCreated d
+  writeTVar (subDeleted s) $! _subDeleted d
+  writeTVar (ntfReceived s) $! _ntfReceived d
+  writeTVar (ntfDelivered s) $! _ntfDelivered d
   setPeriodStats (activeTokens s) (_activeTokens d)
   setPeriodStats (activeSubs s) (_activeSubs d)
 

src/Simplex/Messaging/Server.hs

@@ -65,9 +65,9 @@ import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore
-import Simplex.Messaging.Server.MsgStore.STM (MsgQueue)
+import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.QueueStore
-import Simplex.Messaging.Server.QueueStore.STM (QueueStore)
+import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
@@ -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 ()
@@ -386,7 +410,8 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
                 Right _ -> do
                   withLog (`logCreateById` rId)
                   stats <- asks serverStats
-                  atomically $ modifyTVar (qCreated stats) (+ 1)
+                  atomically $ modifyTVar' (qCreated stats) (+ 1)
+                  atomically $ modifyTVar' (qCount stats) (+ 1)
                   subscribeQueue qr rId $> IDS (qik ids)
 
             logCreateById :: StoreLog 'WriteMode -> RecipientId -> IO ()
@@ -404,7 +429,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
         secureQueue_ st sKey = time "KEY" $ do
           withLog $ \s -> logSecureQueue s queueId sKey
           stats <- asks serverStats
-          atomically $ modifyTVar (qSecured stats) (+ 1)
+          atomically $ modifyTVar' (qSecured stats) (+ 1)
           atomically $ (corrId,queueId,) . either ERR (const OK) <$> secureQueue st queueId sKey
 
         addQueueNotifier_ :: QueueStore -> NtfPublicVerifyKey -> RcvNtfPublicDhKey -> m (Transmission BrokerMsg)
@@ -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
-                      atomically $ modifyTVar (msgSent stats) (+ 1)
+                      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
@@ -599,9 +633,9 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
           where
             forkSub :: m ()
             forkSub = do
-              atomically . modifyTVar sub $ \s -> s {subThread = SubPending}
+              atomically . modifyTVar' sub $ \s -> s {subThread = SubPending}
               t <- mkWeakThreadId =<< forkIO subscriber
-              atomically . modifyTVar sub $ \case
+              atomically . modifyTVar' sub $ \case
                 s@Sub {subThread = SubPending} -> s {subThread = SubThread t}
                 s -> s
               where
@@ -612,7 +646,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
                     writeTBQueue sndQ [(CorrId "", rId, MSG encMsg)]
                     s <- readTVar sub
                     void $ setDelivered s msg
-                    writeTVar sub s {subThread = NoSub}
+                    writeTVar sub $! s {subThread = NoSub}
 
         time :: T.Text -> m a -> m a
         time name = timed name queueId
@@ -646,7 +680,8 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
           withLog (`logDeleteQueue` queueId)
           ms <- asks msgStore
           stats <- asks serverStats
-          atomically $ modifyTVar (qDeleted stats) (+ 1)
+          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/Env/STM.hs

@@ -164,8 +164,8 @@ newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile,
       (qs, s') <- liftIO $ readWriteStoreLog s
       atomically $ do
         writeTVar queues =<< mapM newTVar qs
-        writeTVar senders $ M.foldr' addSender M.empty qs
-        writeTVar notifiers $ M.foldr' addNotifier M.empty qs
+        writeTVar senders $! M.foldr' addSender M.empty qs
+        writeTVar notifiers $! M.foldr' addNotifier M.empty qs
       pure s'
     addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId
     addSender q = M.insert (senderId q) (recipientId q)

src/Simplex/Messaging/Server/MsgStore.hs

@@ -1,13 +1,11 @@
-{-# LANGUAGE FunctionalDependencies #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
 
 module Simplex.Messaging.Server.MsgStore where
 
 import Control.Applicative ((<|>))
-import Data.Int (Int64)
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (Message (..), MsgId, RcvMessage (..), RecipientId)
+import Simplex.Messaging.Protocol (Message (..), RcvMessage (..), RecipientId)
 
 data MsgLogRecord = MLRv3 RecipientId Message | MLRv1 RecipientId RcvMessage
 
@@ -16,16 +14,3 @@ instance StrEncoding MsgLogRecord where
     MLRv3 rId msg -> strEncode (Str "v3", rId, msg)
     MLRv1 rId msg -> strEncode (rId, msg)
   strP = "v3 " *> (MLRv3 <$> strP_ <*> strP) <|> MLRv1 <$> strP_ <*> strP
-
-class MonadMsgStore s q m | s -> q where
-  getMsgQueue :: s -> RecipientId -> Int -> m q
-  delMsgQueue :: s -> RecipientId -> m ()
-  flushMsgQueue :: s -> RecipientId -> m [Message]
-
-class MonadMsgQueue q m where
-  writeMsg :: q -> Message -> m (Maybe Message) -- non blocking
-  tryPeekMsg :: q -> m (Maybe Message) -- non blocking
-  peekMsg :: q -> m Message -- blocking
-  tryDelMsg :: q -> MsgId -> m Bool -- non blocking
-  tryDelPeekMsg :: q -> MsgId -> m (Bool, Maybe Message) -- atomic delete (== read) last and peek next message, if available
-  deleteExpiredMsgs :: q -> Int64 -> m ()

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

@@ -1,7 +1,6 @@
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
-{-# LANGUAGE InstanceSigs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -9,8 +8,17 @@
 
 module Simplex.Messaging.Server.MsgStore.STM
   ( STMMsgStore,
-    MsgQueue,
+    MsgQueue (..),
     newMsgStore,
+    getMsgQueue,
+    delMsgQueue,
+    flushMsgQueue,
+    writeMsg,
+    tryPeekMsg,
+    peekMsg,
+    tryDelMsg,
+    tryDelPeekMsg,
+    deleteExpiredMsgs,
   )
 where
 
@@ -21,7 +29,6 @@ import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Time.Clock.System (SystemTime (systemSeconds))
 import Simplex.Messaging.Protocol (Message (..), MsgId, RecipientId)
-import Simplex.Messaging.Server.MsgStore
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import UnliftIO.STM
@@ -38,75 +45,73 @@ type STMMsgStore = TMap RecipientId MsgQueue
 newMsgStore :: STM STMMsgStore
 newMsgStore = TM.empty
 
-instance MonadMsgStore STMMsgStore MsgQueue STM where
-  getMsgQueue :: STMMsgStore -> RecipientId -> Int -> STM MsgQueue
-  getMsgQueue st rId quota = maybe newQ pure =<< TM.lookup rId st
-    where
-      newQ = do
-        msgQueue <- newTQueue
-        canWrite <- newTVar True
-        size <- newTVar 0
-        let q = MsgQueue {msgQueue, quota, canWrite, size}
-        TM.insert rId q st
-        pure q
+getMsgQueue :: STMMsgStore -> RecipientId -> Int -> STM MsgQueue
+getMsgQueue st rId quota = maybe newQ pure =<< TM.lookup rId st
+  where
+    newQ = do
+      msgQueue <- newTQueue
+      canWrite <- newTVar True
+      size <- newTVar 0
+      let q = MsgQueue {msgQueue, quota, canWrite, size}
+      TM.insert rId q st
+      pure q
 
-  delMsgQueue :: STMMsgStore -> RecipientId -> STM ()
-  delMsgQueue st rId = TM.delete rId st
+delMsgQueue :: STMMsgStore -> RecipientId -> STM ()
+delMsgQueue st rId = TM.delete rId st
 
-  flushMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
-  flushMsgQueue st rId = TM.lookupDelete rId st >>= maybe (pure []) (flushTQueue . msgQueue)
+flushMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
+flushMsgQueue st rId = TM.lookupDelete rId st >>= maybe (pure []) (flushTQueue . msgQueue)
 
-instance MonadMsgQueue MsgQueue STM where
-  writeMsg :: MsgQueue -> Message -> STM (Maybe Message)
-  writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} msg = do
-    canWrt <- readTVar canWrite
-    empty <- isEmptyTQueue q
-    if canWrt || empty
-      then do
-        canWrt' <- (quota >) <$> readTVar size
-        writeTVar canWrite canWrt'
-        modifyTVar' size (+ 1)
-        if canWrt'
-          then writeTQueue q msg $> Just msg
-          else writeTQueue q msgQuota $> Nothing
-      else pure Nothing
-    where
-      msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
+writeMsg :: MsgQueue -> Message -> STM (Maybe Message)
+writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} msg = do
+  canWrt <- readTVar canWrite
+  empty <- isEmptyTQueue q
+  if canWrt || empty
+    then do
+      canWrt' <- (quota >) <$> readTVar size
+      writeTVar canWrite $! canWrt'
+      modifyTVar' size (+ 1)
+      if canWrt'
+        then writeTQueue q msg $> Just msg
+        else writeTQueue q msgQuota $> Nothing
+    else pure Nothing
+  where
+    msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
 
-  tryPeekMsg :: MsgQueue -> STM (Maybe Message)
-  tryPeekMsg = tryPeekTQueue . msgQueue
-  {-# INLINE tryPeekMsg #-}
+tryPeekMsg :: MsgQueue -> STM (Maybe Message)
+tryPeekMsg = tryPeekTQueue . msgQueue
+{-# INLINE tryPeekMsg #-}
 
-  peekMsg :: MsgQueue -> STM Message
-  peekMsg = peekTQueue . msgQueue
-  {-# INLINE peekMsg #-}
+peekMsg :: MsgQueue -> STM Message
+peekMsg = peekTQueue . msgQueue
+{-# INLINE peekMsg #-}
 
-  tryDelMsg :: MsgQueue -> MsgId -> STM Bool
-  tryDelMsg mq msgId' =
-    tryPeekMsg mq >>= \case
-      Just msg
-        | msgId msg == msgId' || B.null msgId' -> tryDeleteMsg mq >> pure True
-        | otherwise -> pure False
-      _ -> pure False
+tryDelMsg :: MsgQueue -> MsgId -> STM (Maybe Message)
+tryDelMsg mq msgId' =
+  tryPeekMsg mq >>= \case
+    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 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)
+-- atomic delete (== read) last and peek next message if available
+tryDelPeekMsg :: MsgQueue -> MsgId -> STM (Maybe Message, Maybe Message)
+tryDelPeekMsg mq msgId' =
+  tryPeekMsg mq >>= \case
+    msg_@(Just msg)
+      | 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
-    where
-      loop = tryPeekMsg mq >>= mapM_ delOldMsg
-      delOldMsg = \case
-        Message {msgTs} ->
-          when (systemSeconds msgTs < old) $
-            tryDeleteMsg mq >> loop
-        _ -> pure ()
+deleteExpiredMsgs :: MsgQueue -> Int64 -> STM ()
+deleteExpiredMsgs mq old = loop
+  where
+    loop = tryPeekMsg mq >>= mapM_ delOldMsg
+    delOldMsg = \case
+      Message {msgTs} ->
+        when (systemSeconds msgTs < old) $
+          tryDeleteMsg mq >> loop
+      _ -> pure ()
 
 tryDeleteMsg :: MsgQueue -> STM ()
 tryDeleteMsg MsgQueue {msgQueue = q, size} =

src/Simplex/Messaging/Server/QueueStore.hs

@@ -9,20 +9,20 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 
 data QueueRec = QueueRec
-  { recipientId :: RecipientId,
-    recipientKey :: RcvPublicVerifyKey,
-    rcvDhSecret :: RcvDhSecret,
-    senderId :: SenderId,
-    senderKey :: Maybe SndPublicVerifyKey,
-    notifier :: Maybe NtfCreds,
-    status :: ServerQueueStatus
+  { recipientId :: !RecipientId,
+    recipientKey :: !RcvPublicVerifyKey,
+    rcvDhSecret :: !RcvDhSecret,
+    senderId :: !SenderId,
+    senderKey :: !(Maybe SndPublicVerifyKey),
+    notifier :: !(Maybe NtfCreds),
+    status :: !ServerQueueStatus
   }
   deriving (Eq, Show)
 
 data NtfCreds = NtfCreds
-  { notifierId :: NotifierId,
-    notifierKey :: NtfPublicVerifyKey,
-    rcvNtfDhSecret :: RcvNtfDhSecret
+  { notifierId :: !NotifierId,
+    notifierKey :: !NtfPublicVerifyKey,
+    rcvNtfDhSecret :: !RcvNtfDhSecret
   }
   deriving (Eq, Show)
 
@@ -33,12 +33,3 @@ instance StrEncoding NtfCreds where
     pure NtfCreds {notifierId, notifierKey, rcvNtfDhSecret}
 
 data ServerQueueStatus = QueueActive | QueueOff deriving (Eq, Show)
-
-class MonadQueueStore s m where
-  addQueue :: s -> QueueRec -> m (Either ErrorType ())
-  getQueue :: s -> SParty p -> QueueId -> m (Either ErrorType QueueRec)
-  secureQueue :: s -> RecipientId -> SndPublicVerifyKey -> m (Either ErrorType QueueRec)
-  addQueueNotifier :: s -> RecipientId -> NtfCreds -> m (Either ErrorType QueueRec)
-  deleteQueueNotifier :: s -> RecipientId -> m (Either ErrorType ())
-  suspendQueue :: s -> RecipientId -> m (Either ErrorType ())
-  deleteQueue :: s -> RecipientId -> m (Either ErrorType ())

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

@@ -1,7 +1,6 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GADTs #-}
-{-# LANGUAGE InstanceSigs #-}
 {-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
@@ -10,7 +9,18 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE UndecidableInstances #-}
 
-module Simplex.Messaging.Server.QueueStore.STM where
+module Simplex.Messaging.Server.QueueStore.STM
+  ( QueueStore (..),
+    newQueueStore,
+    addQueue,
+    getQueue,
+    secureQueue,
+    addQueueNotifier,
+    deleteQueueNotifier,
+    suspendQueue,
+    deleteQueue,
+  )
+where
 
 import Control.Monad
 import Data.Functor (($>))
@@ -34,66 +44,65 @@ newQueueStore = do
   notifiers <- TM.empty
   pure QueueStore {queues, senders, notifiers}
 
-instance MonadQueueStore QueueStore STM where
-  addQueue :: QueueStore -> QueueRec -> STM (Either ErrorType ())
-  addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = do
-    ifM hasId (pure $ Left DUPLICATE_) $ do
-      qVar <- newTVar q
-      TM.insert rId qVar queues
-      TM.insert sId rId senders
-      pure $ Right ()
-    where
-      hasId = (||) <$> TM.member rId queues <*> TM.member sId senders
+addQueue :: QueueStore -> QueueRec -> STM (Either ErrorType ())
+addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = do
+  ifM hasId (pure $ Left DUPLICATE_) $ do
+    qVar <- newTVar q
+    TM.insert rId qVar queues
+    TM.insert sId rId senders
+    pure $ Right ()
+  where
+    hasId = (||) <$> TM.member rId queues <*> TM.member sId senders
 
-  getQueue :: QueueStore -> SParty p -> QueueId -> STM (Either ErrorType QueueRec)
-  getQueue QueueStore {queues, senders, notifiers} party qId =
-    toResult <$> (mapM readTVar =<< getVar)
-    where
-      getVar = case party of
-        SRecipient -> TM.lookup qId queues
-        SSender -> TM.lookup qId senders $>>= (`TM.lookup` queues)
-        SNotifier -> TM.lookup qId notifiers $>>= (`TM.lookup` queues)
+getQueue :: QueueStore -> SParty p -> QueueId -> STM (Either ErrorType QueueRec)
+getQueue QueueStore {queues, senders, notifiers} party qId =
+  toResult <$> (mapM readTVar =<< getVar)
+  where
+    getVar = case party of
+      SRecipient -> TM.lookup qId queues
+      SSender -> TM.lookup qId senders $>>= (`TM.lookup` queues)
+      SNotifier -> TM.lookup qId notifiers $>>= (`TM.lookup` queues)
 
-  secureQueue :: QueueStore -> RecipientId -> SndPublicVerifyKey -> STM (Either ErrorType QueueRec)
-  secureQueue QueueStore {queues} rId sKey =
-    withQueue rId queues $ \qVar ->
-      readTVar qVar >>= \q -> case senderKey q of
-        Just k -> pure $ if sKey == k then Just q else Nothing
-        _ ->
-          let q' = q {senderKey = Just sKey}
-          in writeTVar qVar q' $> Just q'
+secureQueue :: QueueStore -> RecipientId -> SndPublicVerifyKey -> STM (Either ErrorType QueueRec)
+secureQueue QueueStore {queues} rId sKey =
+  withQueue rId queues $ \qVar ->
+    readTVar qVar >>= \q -> case senderKey q of
+      Just k -> pure $ if sKey == k then Just q else Nothing
+      _ ->
+        let q' = q {senderKey = Just sKey}
+         in writeTVar qVar q' $> Just q'
 
-  addQueueNotifier :: QueueStore -> RecipientId -> NtfCreds -> STM (Either ErrorType QueueRec)
-  addQueueNotifier QueueStore {queues, notifiers} rId ntfCreds@NtfCreds {notifierId = nId} = do
-    ifM (TM.member nId notifiers) (pure $ Left DUPLICATE_) $
-      withQueue rId queues $ \qVar -> do
-        q <- readTVar qVar
-        forM_ (notifier q) $ (`TM.delete` notifiers) . notifierId
-        writeTVar qVar q {notifier = Just ntfCreds}
-        TM.insert nId rId notifiers
-        pure $ Just q
-
-  deleteQueueNotifier :: QueueStore -> RecipientId -> STM (Either ErrorType ())
-  deleteQueueNotifier QueueStore {queues, notifiers} rId =
+addQueueNotifier :: QueueStore -> RecipientId -> NtfCreds -> STM (Either ErrorType QueueRec)
+addQueueNotifier QueueStore {queues, notifiers} rId ntfCreds@NtfCreds {notifierId = nId} = do
+  ifM (TM.member nId notifiers) (pure $ Left DUPLICATE_) $
     withQueue rId queues $ \qVar -> do
       q <- readTVar qVar
-      forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId notifiers
-      writeTVar qVar q {notifier = Nothing}
-      pure $ Just ()
+      forM_ (notifier q) $ (`TM.delete` notifiers) . notifierId
+      writeTVar qVar $! q {notifier = Just ntfCreds}
+      TM.insert nId rId notifiers
+      pure $ Just q
 
-  suspendQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
-  suspendQueue QueueStore {queues} rId =
-    withQueue rId queues $ \qVar -> modifyTVar' qVar (\q -> q {status = QueueOff}) $> Just ()
+deleteQueueNotifier :: QueueStore -> RecipientId -> STM (Either ErrorType ())
+deleteQueueNotifier QueueStore {queues, notifiers} rId =
+  withQueue rId queues $ \qVar -> do
+    q <- readTVar qVar
+    forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId notifiers
+    writeTVar qVar $! q {notifier = Nothing}
+    pure $ Just ()
 
-  deleteQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
-  deleteQueue QueueStore {queues, senders, notifiers} rId = do
-    TM.lookupDelete rId queues >>= \case
-      Just qVar ->
-        readTVar qVar >>= \q -> do
-          TM.delete (senderId q) senders
-          forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId notifiers
-          pure $ Right ()
-      _ -> pure $ Left AUTH
+suspendQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
+suspendQueue QueueStore {queues} rId =
+  withQueue rId queues $ \qVar -> modifyTVar' qVar (\q -> q {status = QueueOff}) $> Just ()
+
+deleteQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
+deleteQueue QueueStore {queues, senders, notifiers} rId = do
+  TM.lookupDelete rId queues >>= \case
+    Just qVar ->
+      readTVar qVar >>= \q -> do
+        TM.delete (senderId q) senders
+        forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId notifiers
+        pure $ Right ()
+    _ -> pure $ Left AUTH
 
 toResult :: Maybe a -> Either ErrorType a
 toResult = maybe (Left AUTH) Right

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,7 +40,12 @@ 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
   }
 
 newServerStats :: UTCTime -> STM ServerStats
@@ -46,7 +57,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,20 +73,30 @@ 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
-  writeTVar (fromTime s) (_fromTime d)
-  writeTVar (qCreated s) (_qCreated d)
-  writeTVar (qSecured s) (_qSecured d)
-  writeTVar (qDeleted s) (_qDeleted d)
-  writeTVar (msgSent s) (_msgSent d)
-  writeTVar (msgRecv s) (_msgRecv d)
-  setPeriodStats (activeQueues s) (_activeQueues d)
+  writeTVar (fromTime s) $! _fromTime d
+  writeTVar (qCreated s) $! _qCreated d
+  writeTVar (qSecured s) $! _qSecured d
+  writeTVar (qDeleted s) $! _qDeleted d
+  writeTVar (msgSent s) $! _msgSent d
+  writeTVar (msgRecv s) $! _msgRecv d
+  setPeriodStats (activeQueuesNtf s) (_activeQueuesNtf 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 +104,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 +118,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),
@@ -117,6 +153,9 @@ data PeriodStatsData a = PeriodStatsData
     _month :: Set a
   }
 
+newPeriodStatsData :: PeriodStatsData a
+newPeriodStatsData = PeriodStatsData {_day = S.empty, _week = S.empty, _month = S.empty}
+
 getPeriodStatsData :: PeriodStats a -> STM (PeriodStatsData a)
 getPeriodStatsData s = do
   _day <- readTVar $ day s
@@ -126,9 +165,9 @@ getPeriodStatsData s = do
 
 setPeriodStats :: PeriodStats a -> PeriodStatsData a -> STM ()
 setPeriodStats s d = do
-  writeTVar (day s) (_day d)
-  writeTVar (week s) (_week d)
-  writeTVar (month s) (_month d)
+  writeTVar (day s) $! _day d
+  writeTVar (week s) $! _week d
+  writeTVar (month s) $! _month d
 
 instance (Ord a, StrEncoding a) => StrEncoding (PeriodStatsData a) where
   strEncode PeriodStatsData {_day, _week, _month} =
@@ -165,4 +204,4 @@ updatePeriodStats stats pId = do
   updatePeriod week
   updatePeriod month
   where
-    updatePeriod pSel = modifyTVar (pSel stats) (S.insert pId)
+    updatePeriod pSel = modifyTVar' (pSel stats) (S.insert pId)

src/Simplex/Messaging/Transport.hs

@@ -75,12 +75,14 @@ import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy as BL
 import Data.Default (def)
 import Data.Functor (($>))
+import Data.Version (showVersion)
 import GHC.Generics (Generic)
 import GHC.IO.Handle.Internals (ioe_EOF)
 import Generic.Random (genericArbitraryU)
 import Network.Socket
 import qualified Network.TLS as T
 import qualified Network.TLS.Extra as TE
+import qualified Paths_simplexmq as SMQ
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON)
@@ -100,7 +102,7 @@ supportedSMPServerVRange :: VersionRange
 supportedSMPServerVRange = mkVersionRange 1 5
 
 simplexMQVersion :: String
-simplexMQVersion = "4.2.1"
+simplexMQVersion = showVersion SMQ.version
 
 -- * Transport connection class
 
@@ -214,7 +216,7 @@ instance Transport TLS where
       $ do
         b <- readChunks =<< readTVarIO buffer
         let (s, b') = B.splitAt n b
-        atomically $ writeTVar buffer b'
+        atomically $ writeTVar buffer $! b'
         pure s
     where
       readChunks :: ByteString -> IO ByteString
@@ -237,7 +239,7 @@ instance Transport TLS where
       $ do
         b <- readChunks =<< readTVarIO buffer
         let (s, b') = B.break (== '\n') b
-        atomically $ writeTVar buffer (B.drop 1 b') -- drop '\n' we made a break at
+        atomically $ writeTVar buffer $! B.drop 1 b' -- drop '\n' we made a break at
         pure $ trimCR s
     where
       readChunks :: ByteString -> IO ByteString

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

@@ -52,7 +52,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
@@ -628,10 +628,12 @@ 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 ()
 
     logSize testStoreLogFile `shouldReturn` 2
-    logSize testStoreMsgsFile `shouldReturn` 3
+    logSize testStoreMsgsFile `shouldReturn` 5
 
     withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
       rId <- readTVarIO recipientId
@@ -647,15 +649,21 @@ 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
 
     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

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"