Merge branch 'master' into f/files-tests

2026-07-11 22:49:14 +00:00 · 2023-07-17 20:30:18 +04:00
parent f0302e9d35 de95119ca6
commit 7e002d784b
11 changed files with 179 additions and 74 deletions
@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.2.1
+version: 5.3.0.0
 synopsis: SimpleXMQ message broker
 description: |
  This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack

 name:           simplexmq
-version:        5.2.1
+version:        5.3.0.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -109,6 +109,7 @@ library
      Simplex.Messaging.Protocol
      Simplex.Messaging.Server
      Simplex.Messaging.Server.CLI
+      Simplex.Messaging.Server.Control
      Simplex.Messaging.Server.Env.STM
      Simplex.Messaging.Server.Expiration
      Simplex.Messaging.Server.Main
@@ -1095,7 +1095,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                SMP SMP.QUOTA -> case msgType of
                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                  AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
-                  _ -> retrySndMsg RISlow
+                  _ -> retryIfNotExpired RISlow err
                SMP SMP.AUTH -> case msgType of
                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                  AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -1124,11 +1124,12 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                _
                  -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server),
                  -- the message sending would be retried
-                  | temporaryOrHostError e -> do
-                    let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg RIFast)
+                  | temporaryOrHostError e -> retryIfNotExpired RIFast err
                  | otherwise -> notifyDel msgId err
              where
+                retryIfNotExpired riMode err = do
+                  let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
+                  ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg riMode)
                msgExpired timeoutSel = do
                  msgTimeout <- asks $ timeoutSel . config
                  currentTime <- liftIO getCurrentTime
@@ -58,11 +58,13 @@ import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Data.Type.Equality
 import GHC.TypeLits (KnownNat)
-import Network.Socket (ServiceName)
+import Network.Socket (ServiceName, Socket, socketToHandle)
+import Simplex.Messaging.Agent.Lock
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding (Encoding (smpEncode))
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.Control
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore
@@ -74,10 +76,11 @@ import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Buffer (trimCR)
 import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Util
 import System.Exit (exitFailure)
-import System.IO (hPutStrLn)
+import System.IO (hPutStrLn, hSetNewlineMode, universalNewlineMode)
 import System.Mem.Weak (deRefWeak)
 import UnliftIO.Concurrent
 import UnliftIO.Directory (doesFileExist, renameFile)
@@ -110,19 +113,22 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
  raceAny_
    ( serverThread s subscribedQ subscribers subscriptions cancelSub :
      serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :
-      map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg
+      map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
    )
-    `finally` (withLog closeStoreLog >> saveServerMessages >> saveServerStats)
+    `finally` withLock (savingLock s) "final" (saveServer False)
  where
    runServer :: (ServiceName, ATransport) -> M ()
    runServer (tcpPort, ATransport t) = do
      serverParams <- asks tlsServerParams
      runTransportServer started tcpPort serverParams tCfg (runClient t)

+    saveServer :: Bool -> M ()
+    saveServer keepMsgs = withLog closeStoreLog >> saveServerMessages keepMsgs >> saveServerStats
+
    serverThread ::
      forall s.
      Server ->
-      (Server -> TBQueue (QueueId, Client)) ->
+      (Server -> TQueue (QueueId, Client)) ->
      (Server -> TMap QueueId Client) ->
      (Client -> TMap QueueId s) ->
      (s -> IO ()) ->
@@ -134,7 +140,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
      where
        updateSubscribers :: STM (Maybe (QueueId, Client))
        updateSubscribers = do
-          (qId, clnt) <- readTBQueue $ subQ s
+          (qId, clnt) <- readTQueue $ subQ s
          let clientToBeNotified = \c' ->
                if sameClientSession clnt c'
                  then pure Nothing
@@ -223,6 +229,57 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
        Right th -> runClientTransport th
        Left _ -> pure ()

+    controlPortThread_ :: ServerConfig -> [M ()]
+    controlPortThread_ ServerConfig {controlPort = Just port} = [runCPServer port]
+    controlPortThread_ _ = []
+
+    runCPServer :: ServiceName -> M ()
+    runCPServer port = do
+      srv <- asks server
+      cpStarted <- newEmptyTMVarIO
+      u <- askUnliftIO
+      liftIO $ runTCPServer cpStarted port $ runCPClient u srv
+      where
+        runCPClient :: UnliftIO (ReaderT Env IO) -> Server -> Socket -> IO ()
+        runCPClient u srv sock = do
+          h <- socketToHandle sock ReadWriteMode
+          hSetBuffering h LineBuffering
+          hSetNewlineMode h universalNewlineMode
+          hPutStrLn h "SMP server control port\n'help' for supported commands"
+          cpLoop h
+          where
+            cpLoop h = do
+              s <- B.hGetLine h
+              case strDecode $ trimCR s of
+                Right CPQuit -> hClose h
+                Right cmd -> processCP h cmd >> cpLoop h
+                Left err -> hPutStrLn h ("error: " <> err) >> cpLoop h
+            processCP h = \case
+              CPSuspend -> hPutStrLn h "suspend not implemented"
+              CPResume -> hPutStrLn h "resume not implemented"
+              CPClients -> hPutStrLn h "clients not implemented"
+              CPStats -> do
+                ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, msgSentNtf, msgRecvNtf, qCount, msgCount} <- unliftIO u $ asks serverStats
+                putStat "fromTime" fromTime
+                putStat "qCreated" qCreated
+                putStat "qSecured" qSecured
+                putStat "qDeleted" qDeleted
+                putStat "msgSent" msgSent
+                putStat "msgRecv" msgRecv
+                putStat "msgSentNtf" msgSentNtf
+                putStat "msgRecvNtf" msgRecvNtf
+                putStat "qCount" qCount
+                putStat "msgCount" msgCount
+                where
+                  putStat :: Show a => String -> TVar a -> IO ()
+                  putStat label var = readTVarIO var >>= \v -> hPutStrLn h $ label <> ": " <> show v
+              CPSave -> withLock (savingLock srv) "control" $ do
+                hPutStrLn h "saving server state..."
+                unliftIO u $ saveServer True
+                hPutStrLn h "server state saved!"
+              CPHelp -> hPutStrLn h "commands: stats, save, help, quit"
+              CPQuit -> pure ()
+
 runClientTransport :: Transport c => THandle c -> M ()
 runClientTransport th@THandle {thVersion, sessionId} = do
  q <- asks $ tbqSize . config
@@ -477,7 +534,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
          where
            newSub :: m (TVar Sub)
            newSub = time "SUB newSub" . atomically $ do
-              writeTBQueue subscribedQ (rId, clnt)
+              writeTQueue subscribedQ (rId, clnt)
              sub <- newTVar =<< newSubscription NoSub
              TM.insert rId sub subscriptions
              pure sub
@@ -522,7 +579,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
        subscribeNotifications :: m (Transmission BrokerMsg)
        subscribeNotifications = time "NSUB" . atomically $ do
          unlessM (TM.member queueId ntfSubscriptions) $ do
-            writeTBQueue ntfSubscribedQ (queueId, clnt)
+            writeTQueue ntfSubscribedQ (queueId, clnt)
            TM.insert queueId () ntfSubscriptions
          pure ok

@@ -720,8 +777,8 @@ randomId n = do
  gVar <- asks idsDrg
  atomically (C.pseudoRandomBytes n gVar)

-saveServerMessages :: (MonadUnliftIO m, MonadReader Env m) => m ()
-saveServerMessages = asks (storeMsgsFile . config) >>= mapM_ saveMessages
+saveServerMessages :: (MonadUnliftIO m, MonadReader Env m) => Bool -> m ()
+saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessages
  where
    saveMessages f = do
      logInfo $ "saving messages to file " <> T.pack f
@@ -730,8 +787,9 @@ saveServerMessages = asks (storeMsgsFile . config) >>= mapM_ saveMessages
        readTVarIO ms >>= mapM_ (saveQueueMsgs ms h) . M.keys
      logInfo "messages saved"
      where
+        getMessages = if keepMsgs then snapshotMsgQueue else flushMsgQueue
        saveQueueMsgs ms h rId =
-          atomically (flushMsgQueue ms rId)
+          atomically (getMessages ms rId)
            >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId)

 restoreServerMessages :: forall m. (MonadUnliftIO m, MonadReader Env m) => m ()
@@ -0,0 +1,36 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Server.Control where
+
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import Simplex.Messaging.Encoding.String
+
+data ControlProtocol
+  = CPSuspend
+  | CPResume
+  | CPClients
+  | CPStats
+  | CPSave
+  | CPHelp
+  | CPQuit
+
+instance StrEncoding ControlProtocol where
+  strEncode = \case
+    CPSuspend -> "suspend"
+    CPResume -> "resume"
+    CPClients -> "clients"
+    CPStats -> "stats"
+    CPSave -> "save"
+    CPHelp -> "help"
+    CPQuit -> "quit"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "suspend" -> pure CPSuspend
+      "resume" -> pure CPResume
+      "clients" -> pure CPClients
+      "stats" -> pure CPStats
+      "save" -> pure CPSave
+      "help" -> pure CPHelp
+      "quit" -> pure CPQuit
+      _ -> fail "bad ControlProtocol command"
@@ -19,6 +19,7 @@ import Data.X509.Validation (Fingerprint (..))
 import Network.Socket (ServiceName)
 import qualified Network.TLS as T
 import Numeric.Natural
+import Simplex.Messaging.Agent.Lock
 import Simplex.Messaging.Crypto (KeyHash (..))
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Expiration
@@ -39,7 +40,7 @@ import UnliftIO.STM
 data ServerConfig = ServerConfig
  { transports :: [(ServiceName, ATransport)],
    tbqSize :: Natural,
-    serverTbqSize :: Natural,
+    -- serverTbqSize :: Natural,
    msgQueueQuota :: Int,
    queueIdBytes :: Int,
    msgIdBytes :: Int,
@@ -70,7 +71,9 @@ data ServerConfig = ServerConfig
    -- | SMP client-server protocol version range
    smpServerVRange :: VersionRange,
    -- | TCP transport config
-    transportConfig :: TransportServerConfig
+    transportConfig :: TransportServerConfig,
+    -- | run listener on control port
+    controlPort :: Maybe ServiceName
  }

 defMsgExpirationDays :: Int64
@@ -103,10 +106,11 @@ data Env = Env
  }

 data Server = Server
-  { subscribedQ :: TBQueue (RecipientId, Client),
+  { subscribedQ :: TQueue (RecipientId, Client),
    subscribers :: TMap RecipientId Client,
-    ntfSubscribedQ :: TBQueue (NotifierId, Client),
-    notifiers :: TMap NotifierId Client
+    ntfSubscribedQ :: TQueue (NotifierId, Client),
+    notifiers :: TMap NotifierId Client,
+    savingLock :: Lock
  }

 data Client = Client
@@ -127,13 +131,14 @@ data Sub = Sub
    delivered :: TMVar MsgId
  }

-newServer :: Natural -> STM Server
-newServer qSize = do
-  subscribedQ <- newTBQueue qSize
+newServer :: STM Server
+newServer = do
+  subscribedQ <- newTQueue
  subscribers <- TM.empty
-  ntfSubscribedQ <- newTBQueue qSize
+  ntfSubscribedQ <- newTQueue
  notifiers <- TM.empty
-  return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers}
+  savingLock <- createLock
+  return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, savingLock}

 newClient :: Natural -> Version -> ByteString -> SystemTime -> STM Client
 newClient qSize thVersion sessionId ts = do
@@ -152,26 +157,25 @@ newSubscription subThread = do

 newEnv :: forall m. (MonadUnliftIO m, MonadRandom m) => ServerConfig -> m Env
 newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile} = do
-  server <- atomically $ newServer (serverTbqSize config)
+  server <- atomically newServer
  queueStore <- atomically newQueueStore
  msgStore <- atomically newMsgStore
  idsDrg <- drgNew >>= newTVarIO
-  storeLog <- liftIO $ openReadStoreLog `mapM` storeLogFile
-  s' <- restoreQueues queueStore `mapM` storeLog
+  storeLog <- restoreQueues queueStore `mapM` storeLogFile
  tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
  Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile
  let serverIdentity = KeyHash fp
  serverStats <- atomically . newServerStats =<< liftIO getCurrentTime
-  return Env {config, server, serverIdentity, queueStore, msgStore, idsDrg, storeLog = s', tlsServerParams, serverStats}
+  return Env {config, server, serverIdentity, queueStore, msgStore, idsDrg, storeLog, tlsServerParams, serverStats}
  where
-    restoreQueues :: QueueStore -> StoreLog 'ReadMode -> m (StoreLog 'WriteMode)
-    restoreQueues QueueStore {queues, senders, notifiers} s = do
-      (qs, s') <- liftIO $ readWriteStoreLog s
+    restoreQueues :: QueueStore -> FilePath -> m (StoreLog 'WriteMode)
+    restoreQueues QueueStore {queues, senders, notifiers} f = do
+      (qs, s) <- liftIO $ readWriteStoreLog f
      atomically $ do
        writeTVar queues =<< mapM newTVar qs
        writeTVar senders $! M.foldr' addSender M.empty qs
        writeTVar notifiers $! M.foldr' addNotifier M.empty qs
-      pure s'
+      pure s
    addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId
    addSender q = M.insert (senderId q) (recipientId q)
    addNotifier :: QueueRec -> Map NotifierId RecipientId -> Map NotifierId RecipientId
@@ -130,7 +130,8 @@ smpServerCLI cfgPath logPath =
                <> ("host: " <> host <> "\n")
                <> ("port: " <> defaultServerPort <> "\n")
                <> "log_tls_errors: off\n\
-                   \websockets: off\n\n\
+                   \websockets: off\n\
+                   \# control_port: 5224\n\n\
                   \[INACTIVE_CLIENTS]\n\
                   \# TTL and interval to check inactive clients\n\
                   \disconnect: off\n"
@@ -166,7 +167,7 @@ smpServerCLI cfgPath logPath =
          ServerConfig
            { transports = iniTransports ini,
              tbqSize = 64,
-              serverTbqSize = 1024,
+              -- serverTbqSize = 1024,
              msgQueueQuota = 128,
              queueIdBytes = 24,
              msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
@@ -202,7 +203,8 @@ smpServerCLI cfgPath logPath =
              transportConfig =
                defaultTransportServerConfig
                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
-                  }
+                  },
+              controlPort = either (const Nothing) (Just . T.unpack) $ lookupValue "TRANSPORT" "control_port" ini
            }

 data CliCommand
@@ -13,6 +13,7 @@ module Simplex.Messaging.Server.MsgStore.STM
    getMsgQueue,
    delMsgQueue,
    flushMsgQueue,
+    snapshotMsgQueue,
    writeMsg,
    tryPeekMsg,
    peekMsg,
@@ -62,6 +63,14 @@ delMsgQueue st rId = TM.delete rId st
 flushMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
 flushMsgQueue st rId = TM.lookupDelete rId st >>= maybe (pure []) (flushTQueue . msgQueue)

+snapshotMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
+snapshotMsgQueue st rId  = TM.lookup rId st >>= maybe (pure []) (snapshotTQueue . msgQueue)
+  where
+    snapshotTQueue q = do
+      msgs <- flushTQueue q
+      mapM_ (writeTQueue q) msgs
+      pure msgs
+
 writeMsg :: MsgQueue -> Message -> STM (Maybe Message)
 writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} msg = do
  canWrt <- readTVar canWrite
@@ -94,7 +94,7 @@ setServerStats s d = do
  writeTVar (msgRecvNtf s) $! _msgRecvNtf d
  setPeriodStats (activeQueuesNtf s) (_activeQueuesNtf d)
  writeTVar (qCount s) $! _qCount d
-  writeTVar (msgCount s) $! _qCount d
+  writeTVar (msgCount s) $! _msgCount d

 instance StrEncoding ServerStatsData where
  strEncode ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _msgSentNtf, _msgRecvNtf, _activeQueues, _activeQueuesNtf} =
@@ -25,20 +25,17 @@ module Simplex.Messaging.Server.StoreLog
 where

 import Control.Applicative (optional, (<|>))
-import Control.Monad (unless)
-import Data.Bifunctor (first, second)
+import Control.Monad (foldM, unless, when)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy.Char8 as LB
-import Data.Either (partitionEithers)
 import Data.Functor (($>))
-import Data.List (foldl')
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), ServerQueueStatus (..))
 import Simplex.Messaging.Transport.Buffer (trimCR)
+import Simplex.Messaging.Util (ifM)
 import System.Directory (doesFileExist)
 import System.IO

@@ -139,37 +136,33 @@ logDeleteQueue s = writeStoreLogRecord s . DeleteQueue
 logDeleteNotifier :: StoreLog 'WriteMode -> QueueId -> IO ()
 logDeleteNotifier s = writeStoreLogRecord s . DeleteNotifier

-readWriteStoreLog :: StoreLog 'ReadMode -> IO (Map RecipientId QueueRec, StoreLog 'WriteMode)
-readWriteStoreLog s@(ReadStoreLog f _) = do
-  qs <- readQueues s
-  closeStoreLog s
-  s' <- openWriteStoreLog f
-  writeQueues s' qs
-  pure (qs, s')
+readWriteStoreLog :: FilePath -> IO (Map RecipientId QueueRec, StoreLog 'WriteMode)
+readWriteStoreLog f = do
+  qs <- ifM (doesFileExist f) (readQueues f) (pure M.empty)
+  s <- openWriteStoreLog f
+  writeQueues s qs
+  pure (qs, s)

 writeQueues :: StoreLog 'WriteMode -> Map RecipientId QueueRec -> IO ()
-writeQueues s = mapM_ (writeStoreLogRecord s . CreateQueue) . M.filter active
+writeQueues s = mapM_ $ \q -> when (active q) $ logCreateQueue s q
  where
    active QueueRec {status} = status == QueueActive

-type LogParsingError = (String, ByteString)
-
-readQueues :: StoreLog 'ReadMode -> IO (Map RecipientId QueueRec)
-readQueues (ReadStoreLog _ h) = LB.hGetContents h >>= returnResult . procStoreLog
+readQueues :: FilePath -> IO (Map RecipientId QueueRec)
+readQueues f = foldM processLine M.empty . B.lines =<< B.readFile f
  where
-    procStoreLog :: LB.ByteString -> ([LogParsingError], Map RecipientId QueueRec)
-    procStoreLog = second (foldl' procLogRecord M.empty) . partitionEithers . map parseLogRecord . LB.lines
-    returnResult :: ([LogParsingError], Map RecipientId QueueRec) -> IO (Map RecipientId QueueRec)
-    returnResult (errs, res) = mapM_ printError errs $> res
-    parseLogRecord :: LB.ByteString -> Either LogParsingError StoreLogRecord
-    parseLogRecord = (\s -> first (,s) $ strDecode s) . trimCR . LB.toStrict
-    procLogRecord :: Map RecipientId QueueRec -> StoreLogRecord -> Map RecipientId QueueRec
-    procLogRecord m = \case
-      CreateQueue q -> M.insert (recipientId q) q m
-      SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m
-      AddNotifier qId ntfCreds -> M.adjust (\q -> q {notifier = Just ntfCreds}) qId m
-      SuspendQueue qId -> M.adjust (\q -> q {status = QueueOff}) qId m
-      DeleteQueue qId -> M.delete qId m
-      DeleteNotifier qId -> M.adjust (\q -> q {notifier = Nothing}) qId m
-    printError :: LogParsingError -> IO ()
-    printError (e, s) = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
+    processLine :: Map RecipientId QueueRec -> ByteString -> IO (Map RecipientId QueueRec)
+    processLine m s = case strDecode $ trimCR s of
+      Right r -> pure $ procLogRecord r
+      Left e -> printError e $> m
+      where
+        procLogRecord :: StoreLogRecord -> Map RecipientId QueueRec
+        procLogRecord = \case
+          CreateQueue q -> M.insert (recipientId q) q m
+          SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m
+          AddNotifier qId ntfCreds -> M.adjust (\q -> q {notifier = Just ntfCreds}) qId m
+          SuspendQueue qId -> M.adjust (\q -> q {status = QueueOff}) qId m
+          DeleteQueue qId -> M.delete qId m
+          DeleteNotifier qId -> M.adjust (\q -> q {notifier = Nothing}) qId m
+        printError :: String -> IO ()
+        printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
@@ -82,7 +82,7 @@ cfg =
  ServerConfig
    { transports = undefined,
      tbqSize = 1,
-      serverTbqSize = 1,
+      -- serverTbqSize = 1,
      msgQueueQuota = 4,
      queueIdBytes = 24,
      msgIdBytes = 24,
@@ -100,7 +100,8 @@ cfg =
      privateKeyFile = "tests/fixtures/server.key",
      certificateFile = "tests/fixtures/server.crt",
      smpServerVRange = supportedSMPServerVRange,
-      transportConfig = defaultTransportServerConfig
+      transportConfig = defaultTransportServerConfig,
+      controlPort = Nothing
    }

 withSmpServerStoreMsgLogOnV2 :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a