fix

src/Simplex/Messaging/Server.hs

@@ -52,7 +52,7 @@ import qualified Data.ByteString.Builder as BLD
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
-import Data.Either (fromRight, partitionEithers)
+import Data.Either (fromRight, partitionEithers, rights)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import qualified Data.IntMap.Strict as IM
@@ -61,7 +61,7 @@ import Data.List (intercalate, mapAccumR)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
-import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing)
+import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, listToMaybe)
 import qualified Data.Set as S
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
@@ -170,9 +170,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
     serverThread s label subQ subs clientSubs unsub = do
       labelMyThread label
       cls <- asks clients
+      stats <- asks serverStats
       forever $
         atomically (updateSubscribers cls)
-          $>>= endPreviousSubscriptions
+          $>>= endPreviousSubscriptions stats
           >>= liftIO . mapM_ unsub
       where
         updateSubscribers :: TVar (IM.IntMap (Maybe Client)) -> STM (Maybe (QueueId, Client))
@@ -189,10 +190,12 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                     yes <- readTVar $ connected c'
                     pure $ if yes then Just (qId, c') else Nothing
           updateSub qId (subs s) $>>= clientToBeNotified
-        endPreviousSubscriptions :: (QueueId, Client) -> M (Maybe s)
-        endPreviousSubscriptions (qId, c) = do
-          forkClient c (label <> ".endPreviousSubscriptions") $
+        endPreviousSubscriptions :: ServerStats -> (QueueId, Client) -> M (Maybe s)
+        endPreviousSubscriptions stats (qId, c) = do
+          forkClient c (label <> ".endPreviousSubscriptions") $ do
             atomically $ writeTBQueue (sndQ c) [(CorrId "", qId, END)]
+            incStat $ qSubEnd stats
+            incStat $ qSubEndB stats
           atomically $ TM.lookupDelete qId (clientSubs c)
 
     receiveFromProxyAgent :: ProxyAgent -> M ()
@@ -238,7 +241,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
       initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
       liftIO $ putStrLn $ "server stats log enabled: " <> statsFilePath
       liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
-      ss@ServerStats {fromTime, qCreated, qSecured, qDeletedAll, qDeletedNew, qDeletedSecured, qSub, qSubNoMsg, qSubAuth, qSubDuplicate, qSubProhibited, ntfCreated, ntfDeleted, ntfSub, ntfSubAuth, ntfSubDuplicate, msgSent, msgSentAuth, msgSentQuota, msgSentLarge, msgRecv, msgRecvGet, msgGet, msgGetNoMsg, msgGetAuth, msgGetDuplicate, msgGetProhibited, msgExpired, activeQueues, subscribedQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount, pRelays, pRelaysOwn, pMsgFwds, pMsgFwdsOwn, pMsgFwdsRecv}
+      ss@ServerStats {fromTime, qCreated, qSecured, qDeletedAll, qDeletedAllB, qDeletedNew, qDeletedSecured, qSub, qSubNoMsg, qSubAllB, qSubAuth, qSubDuplicate, qSubProhibited, qSubEnd, qSubEndB, qSubEndSent, qSubEndSentB, ntfCreated, ntfDeleted, ntfDeletedB, ntfSub, ntfSubB, ntfSubAuth, ntfSubDuplicate, msgSent, msgSentAuth, msgSentQuota, msgSentLarge, msgRecv, msgRecvGet, msgGet, msgGetNoMsg, msgGetAuth, msgGetDuplicate, msgGetProhibited, msgExpired, activeQueues, subscribedQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount, pRelays, pRelaysOwn, pMsgFwds, pMsgFwdsOwn, pMsgFwdsRecv}
         <- asks serverStats
       QueueStore {queues, notifiers} <- asks queueStore
       let interval = 1000000 * logInterval
@@ -250,16 +253,24 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
           qCreated' <- atomically $ swapTVar qCreated 0
           qSecured' <- atomically $ swapTVar qSecured 0
           qDeletedAll' <- atomically $ swapTVar qDeletedAll 0
+          qDeletedAllB' <- atomically $ swapTVar qDeletedAllB 0
           qDeletedNew' <- atomically $ swapTVar qDeletedNew 0
           qDeletedSecured' <- atomically $ swapTVar qDeletedSecured 0
           qSub' <- atomically $ swapTVar qSub 0
           qSubNoMsg' <- atomically $ swapTVar qSubNoMsg 0
+          qSubAllB' <- atomically $ swapTVar qSubAllB 0
           qSubAuth' <- atomically $ swapTVar qSubAuth 0
           qSubDuplicate' <- atomically $ swapTVar qSubDuplicate 0
           qSubProhibited' <- atomically $ swapTVar qSubProhibited 0
+          qSubEnd' <- atomically $ swapTVar qSubEnd 0
+          qSubEndB' <- atomically $ swapTVar qSubEndB 0
+          qSubEndSent' <- atomically $ swapTVar qSubEndSent 0
+          qSubEndSentB' <- atomically $ swapTVar qSubEndSentB 0
           ntfCreated' <- atomically $ swapTVar ntfCreated 0
           ntfDeleted' <- atomically $ swapTVar ntfDeleted 0
+          ntfDeletedB' <- atomically $ swapTVar ntfDeletedB 0
           ntfSub' <- atomically $ swapTVar ntfSub 0
+          ntfSubB' <- atomically $ swapTVar ntfSubB 0
           ntfSubAuth' <- atomically $ swapTVar ntfSubAuth 0
           ntfSubDuplicate' <- atomically $ swapTVar ntfSubDuplicate 0
           msgSent' <- atomically $ swapTVar msgSent 0
@@ -345,7 +356,15 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                        show ntfSub',
                        show ntfSubAuth',
                        show ntfSubDuplicate',
-                       show ntfCount'
+                       show ntfCount',
+                       show qDeletedAllB',
+                       show qSubAllB',
+                       show qSubEnd',
+                       show qSubEndB',
+                       show qSubEndSent',
+                       show qSubEndSentB',
+                       show ntfDeletedB',
+                       show ntfSubB'
                      ]
               )
         liftIO $ threadDelay' interval
@@ -434,14 +453,20 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                 putStat "qCreated" qCreated
                 putStat "qSecured" qSecured
                 putStat "qDeletedAll" qDeletedAll
+                putStat "qDeletedAllB" qDeletedAllB
                 putStat "qDeletedNew" qDeletedNew
                 putStat "qDeletedSecured" qDeletedSecured
                 getStat (day . activeQueues) >>= \v -> hPutStrLn h $ "daily active queues: " <> show (S.size v)
                 getStat (day . subscribedQueues) >>= \v -> hPutStrLn h $ "daily subscribed queues: " <> show (S.size v)
                 putStat "qSub" qSub
                 putStat "qSubNoMsg" qSubNoMsg
+                putStat "qSubAllB" qSubAllB
+                subEnds <- (,,,) <$> getStat qSubEnd <*> getStat qSubEndB <*> getStat qSubEndSent <*> getStat qSubEndSentB
+                hPutStrLn h $ "SUB ENDs (queued, queued batches, sent, sent batches): " <> show subEnds
                 subs <- (,,) <$> getStat qSubAuth <*> getStat qSubDuplicate <*> getStat qSubProhibited
                 hPutStrLn h $ "other SUB events (auth, duplicate, prohibited): " <> show subs
+                putStat "qSubEnd" qSubEnd
+                putStat "qSubEndSent" qSubEndSent
                 putStat "msgSent" msgSent
                 putStat "msgRecv" msgRecv
                 putStat "msgRecvGet" msgRecvGet
@@ -631,9 +656,10 @@ runClientTransport h@THandle {params = thParams@THandleParams {thVersion, sessio
       atomically $ modifyTVar' active $ IM.insert clientId $ Just c
       s <- asks server
       expCfg <- asks $ inactiveClientExpiration . config
+      stats <- asks serverStats
       th <- newMVar h -- put TH under a fair lock to interleave messages and command responses
       labelMyThread . B.unpack $ "client $" <> encode sessionId
-      raceAny_ $ [liftIO $ send th c, liftIO $ sendMsg th c, client thParams c s, receive h c] <> disconnectThread_ c expCfg
+      raceAny_ $ [liftIO $ send th c stats, liftIO $ sendMsg th c, client thParams c s, receive h c] <> disconnectThread_ c expCfg
     disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
     disconnectThread_ _ _ = []
     noSubscriptions c = atomically $ (&&) <$> TM.null (ntfSubscriptions c) <*> (not . hasSubs <$> readTVar (subscriptions c))
@@ -679,10 +705,19 @@ receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiv
     ts <- L.toList <$> liftIO (tGet h)
     atomically . (writeTVar rcvActiveAt $!) =<< liftIO getSystemTime
     stats <- asks serverStats
+    let cmd = listToMaybe $ rights $ map (\(_, _, (_, _, cmdOrError)) -> cmdOrError) ts
+    forM_ (cmd >>= batchStatSel) $ \sel -> incStat $ sel stats
     (errs, cmds) <- partitionEithers <$> mapM (cmdAction stats) ts
     write sndQ errs
     write rcvQ cmds
   where
+    batchStatSel :: Cmd -> Maybe (ServerStats -> TVar Int)
+    batchStatSel (Cmd _ cmd) = case cmd of
+      SUB -> Just qSubAllB
+      DEL -> Just qDeletedAllB
+      NSUB -> Just ntfSubB
+      NDEL -> Just ntfDeletedB
+      _ -> Nothing
     cmdAction :: ServerStats -> SignedTransmission ErrorType Cmd -> M (Either (Transmission BrokerMsg) (Maybe QueueRec, Transmission Cmd))
     cmdAction stats (tAuth, authorized, (corrId, entId, cmdOrError)) =
       case cmdOrError of
@@ -701,17 +736,20 @@ receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiv
                 pure $ Left (corrId, entId, ERR AUTH)
     write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
 
-send :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> IO ()
-send th c@Client {sndQ, msgQ, sessionId} = do
+send :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> ServerStats -> IO ()
+send th c@Client {sndQ, msgQ, sessionId} stats = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
-  forever $ atomically (readTBQueue sndQ) >>= sendTransmissions
+  forever $ do
+    ts <- atomically (readTBQueue sndQ)
+    sendTransmissions ts
+    updateENDStats ts
   where
     sendTransmissions :: NonEmpty (Transmission BrokerMsg) -> IO ()
     sendTransmissions ts
       | L.length ts <= 2 = tSend th c ts
       | otherwise = do
           let (msgs_, ts') = mapAccumR splitMessages [] ts
-          -- If the request had batched subscriptions (L.length ts > 2)
+          -- If the request had batched subscriptions and L.length ts > 2
           -- this will reply OK to all SUBs in the first batched transmission,
           -- to reduce client timeouts.
           tSend th c ts'
@@ -725,7 +763,14 @@ send th c@Client {sndQ, msgQ, sessionId} = do
           -- replace MSG response with OK, accumulating MSG in a separate list.
           MSG {} -> ((CorrId "", entId, cmd) : msgs, (corrId, entId, OK))
           _ -> (msgs, t)
-
+    updateENDStats :: NonEmpty (Transmission BrokerMsg) -> IO ()
+    updateENDStats = \case
+      ts@((_, _, END) :| _) -> do -- END events are not combined with others
+        let len = L.length ts
+        atomically $ modifyTVar' (qSubEndSent stats) (+ len)
+        atomically $ modifyTVar' (qSubEndSentB stats) (+ len `div` 255) -- up to 255 ENDs in the batch
+      _ -> pure ()
+        
 sendMsg :: Transport c => MVar (THandleSMP c 'TServer) -> Client -> IO ()
 sendMsg th c@Client {msgQ, sessionId} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " sendMsg"

src/Simplex/Messaging/Server/Stats.hs

@@ -24,16 +24,24 @@ data ServerStats = ServerStats
     qCreated :: TVar Int,
     qSecured :: TVar Int,
     qDeletedAll :: TVar Int,
+    qDeletedAllB :: TVar Int,
     qDeletedNew :: TVar Int,
     qDeletedSecured :: TVar Int,
     qSub :: TVar Int,
     qSubNoMsg :: TVar Int,
+    qSubAllB :: TVar Int,
     qSubAuth :: TVar Int,
     qSubDuplicate :: TVar Int,
     qSubProhibited :: TVar Int,
+    qSubEnd :: TVar Int,
+    qSubEndB :: TVar Int,
+    qSubEndSent :: TVar Int,
+    qSubEndSentB :: TVar Int,
     ntfCreated :: TVar Int,
     ntfDeleted :: TVar Int,
+    ntfDeletedB :: TVar Int,
     ntfSub :: TVar Int,
+    ntfSubB :: TVar Int,
     ntfSubAuth :: TVar Int,
     ntfSubDuplicate :: TVar Int,
     msgSent :: TVar Int,
@@ -70,16 +78,24 @@ data ServerStatsData = ServerStatsData
     _qCreated :: Int,
     _qSecured :: Int,
     _qDeletedAll :: Int,
+    _qDeletedAllB :: Int,
     _qDeletedNew :: Int,
     _qDeletedSecured :: Int,
     _qSub :: Int,
     _qSubNoMsg :: Int,
+    _qSubAllB :: Int,
     _qSubAuth :: Int,
     _qSubDuplicate :: Int,
     _qSubProhibited :: Int,
+    _qSubEnd :: Int,
+    _qSubEndB :: Int,
+    _qSubEndSent :: Int,
+    _qSubEndSentB :: Int,
     _ntfCreated :: Int,
     _ntfDeleted :: Int,
+    _ntfDeletedB :: Int,
     _ntfSub :: Int,
+    _ntfSubB :: Int,
     _ntfSubAuth :: Int,
     _ntfSubDuplicate :: Int,
     _msgSent :: Int,
@@ -118,16 +134,24 @@ newServerStats ts = do
   qCreated <- newTVarIO 0
   qSecured <- newTVarIO 0
   qDeletedAll <- newTVarIO 0
+  qDeletedAllB <- newTVarIO 0
   qDeletedNew <- newTVarIO 0
   qDeletedSecured <- newTVarIO 0
   qSub <- newTVarIO 0
   qSubNoMsg <- newTVarIO 0
+  qSubAllB <- newTVarIO 0
   qSubAuth <- newTVarIO 0
   qSubDuplicate <- newTVarIO 0
   qSubProhibited <- newTVarIO 0
+  qSubEnd <- newTVarIO 0
+  qSubEndB <- newTVarIO 0
+  qSubEndSent <- newTVarIO 0
+  qSubEndSentB <- newTVarIO 0
   ntfCreated <- newTVarIO 0
   ntfDeleted <- newTVarIO 0
+  ntfDeletedB <- newTVarIO 0
   ntfSub <- newTVarIO 0
+  ntfSubB <- newTVarIO 0
   ntfSubAuth <- newTVarIO 0
   ntfSubDuplicate <- newTVarIO 0
   msgSent <- newTVarIO 0
@@ -163,16 +187,24 @@ newServerStats ts = do
         qCreated,
         qSecured,
         qDeletedAll,
+        qDeletedAllB,
         qDeletedNew,
         qDeletedSecured,
         qSub,
         qSubNoMsg,
+        qSubAllB,
         qSubAuth,
         qSubDuplicate,
         qSubProhibited,
+        qSubEnd,
+        qSubEndB,
+        qSubEndSent,
+        qSubEndSentB,
         ntfCreated,
         ntfDeleted,
+        ntfDeletedB,
         ntfSub,
+        ntfSubB,
         ntfSubAuth,
         ntfSubDuplicate,
         msgSent,
@@ -210,16 +242,24 @@ getServerStatsData s = do
   _qCreated <- readTVarIO $ qCreated s
   _qSecured <- readTVarIO $ qSecured s
   _qDeletedAll <- readTVarIO $ qDeletedAll s
+  _qDeletedAllB <- readTVarIO $ qDeletedAllB s
   _qDeletedNew <- readTVarIO $ qDeletedNew s
   _qDeletedSecured <- readTVarIO $ qDeletedSecured s
   _qSub <- readTVarIO $ qSub s
   _qSubNoMsg <- readTVarIO $ qSubNoMsg s
+  _qSubAllB <- readTVarIO $ qSubAllB s
   _qSubAuth <- readTVarIO $ qSubAuth s
   _qSubDuplicate <- readTVarIO $ qSubDuplicate s
   _qSubProhibited <- readTVarIO $ qSubProhibited s
+  _qSubEnd <- readTVarIO $ qSubEnd s
+  _qSubEndB <- readTVarIO $ qSubEndB s
+  _qSubEndSent <- readTVarIO $ qSubEndSent s
+  _qSubEndSentB <- readTVarIO $ qSubEndSentB s
   _ntfCreated <- readTVarIO $ ntfCreated s
   _ntfDeleted <- readTVarIO $ ntfDeleted s
+  _ntfDeletedB <- readTVarIO $ ntfDeletedB s
   _ntfSub <- readTVarIO $ ntfSub s
+  _ntfSubB <- readTVarIO $ ntfSubB s
   _ntfSubAuth <- readTVarIO $ ntfSubAuth s
   _ntfSubDuplicate <- readTVarIO $ ntfSubDuplicate s
   _msgSent <- readTVarIO $ msgSent s
@@ -255,16 +295,24 @@ getServerStatsData s = do
         _qCreated,
         _qSecured,
         _qDeletedAll,
+        _qDeletedAllB,
         _qDeletedNew,
         _qDeletedSecured,
         _qSub,
         _qSubNoMsg,
+        _qSubAllB,
         _qSubAuth,
         _qSubDuplicate,
         _qSubProhibited,
+        _qSubEnd,
+        _qSubEndB,
+        _qSubEndSent,
+        _qSubEndSentB,
         _ntfCreated,
         _ntfDeleted,
+        _ntfDeletedB,
         _ntfSub,
+        _ntfSubB,
         _ntfSubAuth,
         _ntfSubDuplicate,
         _msgSent,
@@ -302,16 +350,24 @@ setServerStats s d = do
   writeTVar (qCreated s) $! _qCreated d
   writeTVar (qSecured s) $! _qSecured d
   writeTVar (qDeletedAll s) $! _qDeletedAll d
+  writeTVar (qDeletedAllB s) $! _qDeletedAllB d
   writeTVar (qDeletedNew s) $! _qDeletedNew d
   writeTVar (qDeletedSecured s) $! _qDeletedSecured d
   writeTVar (qSub s) $! _qSub d
   writeTVar (qSubNoMsg s) $! _qSubNoMsg d
+  writeTVar (qSubAllB s) $! _qSubAllB d
   writeTVar (qSubAuth s) $! _qSubAuth d
   writeTVar (qSubDuplicate s) $! _qSubDuplicate d
   writeTVar (qSubProhibited s) $! _qSubProhibited d
+  writeTVar (qSubEnd s) $! _qSubEnd d
+  writeTVar (qSubEndB s) $! _qSubEndB d
+  writeTVar (qSubEndSent s) $! _qSubEndSent d
+  writeTVar (qSubEndSentB s) $! _qSubEndSentB d
   writeTVar (ntfCreated s) $! _ntfCreated d
   writeTVar (ntfDeleted s) $! _ntfDeleted d
+  writeTVar (ntfDeletedB s) $! _ntfDeletedB d
   writeTVar (ntfSub s) $! _ntfSub d
+  writeTVar (ntfSubB s) $! _ntfSubB d
   writeTVar (ntfSubAuth s) $! _ntfSubAuth d
   writeTVar (ntfSubDuplicate s) $! _ntfSubDuplicate d
   writeTVar (msgSent s) $! _msgSent d
@@ -351,15 +407,23 @@ instance StrEncoding ServerStatsData where
         "qDeletedAll=" <> strEncode (_qDeletedAll d),
         "qDeletedNew=" <> strEncode (_qDeletedNew d),
         "qDeletedSecured=" <> strEncode (_qDeletedSecured d),
+        "qDeletedAllB=" <> strEncode (_qDeletedAllB d),
         "qCount=" <> strEncode (_qCount d),
         "qSub=" <> strEncode (_qSub d),
         "qSubNoMsg=" <> strEncode (_qSubNoMsg d),
+        "qSubAllB=" <> strEncode (_qSubAllB d),
         "qSubAuth=" <> strEncode (_qSubAuth d),
         "qSubDuplicate=" <> strEncode (_qSubDuplicate d),
         "qSubProhibited=" <> strEncode (_qSubProhibited d),
+        "qSubEnd=" <> strEncode (_qSubEnd d),
+        "qSubEndB=" <> strEncode (_qSubEndB d),
+        "qSubEndSent=" <> strEncode (_qSubEndSent d),
+        "qSubEndSentB=" <> strEncode (_qSubEndSentB d),
         "ntfCreated=" <> strEncode (_ntfCreated d),
         "ntfDeleted=" <> strEncode (_ntfDeleted d),
+        "ntfDeletedB=" <> strEncode (_ntfDeletedB d),
         "ntfSub=" <> strEncode (_ntfSub d),
+        "ntfSubB=" <> strEncode (_ntfSubB d),
         "ntfSubAuth=" <> strEncode (_ntfSubAuth d),
         "ntfSubDuplicate=" <> strEncode (_ntfSubDuplicate d),
         "msgSent=" <> strEncode (_msgSent d),
@@ -402,15 +466,23 @@ instance StrEncoding ServerStatsData where
     (_qDeletedAll, _qDeletedNew, _qDeletedSecured) <-
       (,0,0) <$> ("qDeleted=" *> strP <* A.endOfLine)
         <|> ((,,) <$> ("qDeletedAll=" *> strP <* A.endOfLine) <*> ("qDeletedNew=" *> strP <* A.endOfLine) <*> ("qDeletedSecured=" *> strP <* A.endOfLine))
+    _qDeletedAllB <- opt "qDeletedAllB="
     _qCount <- opt "qCount="
     _qSub <- opt "qSub="
     _qSubNoMsg <- opt "qSubNoMsg="
+    _qSubAllB <- opt "qSubAllB="
     _qSubAuth <- opt "qSubAuth="
     _qSubDuplicate <- opt "qSubDuplicate="
     _qSubProhibited <- opt "qSubProhibited="
+    _qSubEnd <- opt "qSubEnd="
+    _qSubEndB <- opt "qSubEndB="
+    _qSubEndSent <- opt "qSubEndSent="
+    _qSubEndSentB <- opt "qSubEndSentB="
     _ntfCreated <- opt "ntfCreated="
     _ntfDeleted <- opt "ntfDeleted="
+    _ntfDeletedB <- opt "ntfDeletedB="
     _ntfSub <- opt "ntfSub="
+    _ntfSubB <- opt "ntfSubB="
     _ntfSubAuth <- opt "ntfSubAuth="
     _ntfSubDuplicate <- opt "ntfSubDuplicate="
     _msgSent <- "msgSent=" *> strP <* A.endOfLine
@@ -457,16 +529,24 @@ instance StrEncoding ServerStatsData where
           _qCreated,
           _qSecured,
           _qDeletedAll,
+          _qDeletedAllB,
           _qDeletedNew,
           _qDeletedSecured,
           _qSub,
           _qSubNoMsg,
+          _qSubAllB,
           _qSubAuth,
           _qSubDuplicate,
           _qSubProhibited,
+          _qSubEnd,
+          _qSubEndB,
+          _qSubEndSent,
+          _qSubEndSentB,
           _ntfCreated,
           _ntfDeleted,
+          _ntfDeletedB,
           _ntfSub,
+          _ntfSubB,
           _ntfSubAuth,
           _ntfSubDuplicate,
           _msgSent,

tests/CoreTests/BatchingTests.hs

@@ -2,6 +2,7 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 
@@ -40,6 +41,7 @@ batchingTests = do
       it "should break on large message" testClientBatchWithLargeMessage
     describe "v7 (next)" $ do
       it "should batch with 136 subscriptions per batch" testClientBatchSubscriptionsV7
+      it "should batch with N ENDs per batch" testClientBatchENDs
       it "should break on message that does not fit" testClientBatchWithMessageV7
       it "should break on large message" testClientBatchWithLargeMessageV7
 
@@ -165,6 +167,20 @@ testClientBatchSubscriptionsV7 = do
   (length rs1, length rs2, length rs3) `shouldBe` (28, 136, 136)
   all lenOk [s1, s2, s3] `shouldBe` True
 
+testClientBatchENDs :: IO ()
+testClientBatchENDs = do
+  client <- clientStubV7
+  ends <- replicateM 300 randomENDCmd
+  let ends' = map (\t -> Right (Nothing, encodeTransmission (thParams client) t)) ends
+      batches1 = batchTransmissions False smpBlockSize $ L.fromList ends'
+  all lenOk1 batches1 `shouldBe` True
+  let batches = batchTransmissions True smpBlockSize $ L.fromList ends'
+  length batches `shouldBe` 2
+  [TBTransmissions s1 n1 rs1, TBTransmissions s2 n2 rs2] <- pure batches
+  (n1, n2) `shouldBe` (45, 255)
+  (length rs1, length rs2) `shouldBe` (45, 255)
+  all lenOk [s1, s2] `shouldBe` True
+
 testClientBatchWithMessage :: IO ()
 testClientBatchWithMessage = do
   client <- testClientStub
@@ -301,6 +317,12 @@ randomSUBCmd_ a c = do
   (_, rpKey) <- atomically $ C.generateAuthKeyPair a g
   mkTransmission c (Just rpKey, rId, Cmd SRecipient SUB)
 
+randomENDCmd :: IO (Transmission BrokerMsg)
+randomENDCmd = do
+  g <- C.newRandom
+  rId <- atomically $ C.randomBytes 24 g
+  pure (CorrId "", rId, END)
+
 randomSEND :: ByteString -> Int -> IO (Either TransportError (Maybe TransmissionAuth, ByteString))
 randomSEND = randomSEND_ C.SEd25519 subModeSMPVersion
 

tests/ServerTests.hs

@@ -610,7 +610,7 @@ testRestoreMessages at@(ATransport t) =
 
     logSize testStoreLogFile `shouldReturn` 2
     logSize testStoreMsgsFile `shouldReturn` 5
-    logSize testServerStatsBackupFile `shouldReturn` 71
+    logSize testServerStatsBackupFile `shouldReturn` 79
     Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats1 [rId] 5 1
 
@@ -628,7 +628,7 @@ testRestoreMessages at@(ATransport t) =
     logSize testStoreLogFile `shouldReturn` 1
     -- the last message is not removed because it was not ACK'd
     logSize testStoreMsgsFile `shouldReturn` 3
-    logSize testServerStatsBackupFile `shouldReturn` 71
+    logSize testServerStatsBackupFile `shouldReturn` 79
     Right stats2 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats2 [rId] 5 3
 
@@ -647,7 +647,7 @@ testRestoreMessages at@(ATransport t) =
 
     logSize testStoreLogFile `shouldReturn` 1
     logSize testStoreMsgsFile `shouldReturn` 0
-    logSize testServerStatsBackupFile `shouldReturn` 71
+    logSize testServerStatsBackupFile `shouldReturn` 79
     Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats3 [rId] 5 5