core: fix delivery batching (#7065)

src/Simplex/Chat/Library/Subscriber.hs

@@ -3635,13 +3635,13 @@ runDeliveryTaskWorker a deliveryKey Worker {doWork} = do
                   withStore' $ \db -> setDeliveryTaskErrStatus db (deliveryTaskId task) "relay inactive"
               | otherwise ->
                   withWorkItems a doWork (withStore' $ \db -> getNextDeliveryTasks db gInfo task) $ \nextTasks -> do
-                    let (body, acceptedTasks, largeTasks) = batchDeliveryTasks1 (vr cxt) maxEncodedMsgLength nextTasks
+                    let (body_, acceptedTasks, largeTasks) = batchDeliveryTasks1 (vr cxt) maxEncodedMsgLength nextTasks
                         senderGMIds = S.toList . S.fromList $ map (\MessageDeliveryTask {senderGMId} -> senderGMId) acceptedTasks
                     withStore' $ \db -> do
-                      createMsgDeliveryJob db gInfo jobScope senderGMIds body
+                      forM_ body_ $ \body -> createMsgDeliveryJob db gInfo jobScope senderGMIds body
                       forM_ acceptedTasks $ \t -> updateDeliveryTaskStatus db (deliveryTaskId t) DTSProcessed
                       forM_ largeTasks $ \t -> setDeliveryTaskErrStatus db (deliveryTaskId t) "large"
-                    lift . void $ getDeliveryJobWorker True deliveryKey
+                    when (isJust body_) . lift . void $ getDeliveryJobWorker True deliveryKey
             -- DJRelayRemoved is allowed when RSInactive - it forwards XGrpMemDel about relay's own deletion
             DJRelayRemoved
               | workerScope /= DWSGroup ->

src/Simplex/Chat/Messages/Batch.hs

@@ -24,7 +24,6 @@ import qualified Data.ByteString as BS
 import qualified Data.ByteString.Char8 as B
 import Data.Char (ord)
 import Data.Function (on)
-import Data.Foldable (foldr')
 import Data.List (foldl', sortBy)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
@@ -79,15 +78,15 @@ batchMessages mode maxLen = addBatch . foldr addToBatch ([], [], [], 0, 0)
           let encoded = encodeBatch mode bodies
            in Right (MsgBatch encoded msgs) : batches
 
--- | Batches delivery tasks into (batch, accepted, large).
+-- | Batches delivery tasks into (batch if any task was accepted, accepted, large).
 -- Always uses binary batch format for relay groups.
-batchDeliveryTasks1 :: VersionRangeChat -> Int -> NonEmpty MessageDeliveryTask -> (ByteString, [MessageDeliveryTask], [MessageDeliveryTask])
+batchDeliveryTasks1 :: VersionRangeChat -> Int -> NonEmpty MessageDeliveryTask -> (Maybe ByteString, [MessageDeliveryTask], [MessageDeliveryTask])
 batchDeliveryTasks1 _vr maxLen = toResult . foldl' addToBatch ([], [], [], 0, 0) . L.toList
   where
     addToBatch :: ([ByteString], [MessageDeliveryTask], [MessageDeliveryTask], Int, Int) -> MessageDeliveryTask -> ([ByteString], [MessageDeliveryTask], [MessageDeliveryTask], Int, Int)
     addToBatch (msgBodies, accepted, large, len, n) task
-      -- too large: skip, record in large
-      | msgLen > maxLen = (msgBodies, accepted, task : large, len, n)
+      -- element can't fit even a singleton batch (4-byte binary-batch framing)
+      | msgLen + 4 > maxLen = (msgBodies, accepted, task : large, len, n)
       -- fits: include in batch
       -- batch overhead: '=' + count (2) + 2-byte length prefix per element
       | len' + (n + 1) * 2 + 2 <= maxLen = (msgBody : msgBodies, task : accepted, large, len', n + 1)
@@ -98,10 +97,11 @@ batchDeliveryTasks1 _vr maxLen = toResult . foldl' addToBatch ([], [], [], 0, 0)
         msgBody = encodeFwdElement GrpMsgForward {fwdSender, fwdBrokerTs} verifiedMsg
         msgLen = B.length msgBody
         len' = len + msgLen
-    toResult :: ([ByteString], [MessageDeliveryTask], [MessageDeliveryTask], Int, Int) -> (ByteString, [MessageDeliveryTask], [MessageDeliveryTask])
+    toResult :: ([ByteString], [MessageDeliveryTask], [MessageDeliveryTask], Int, Int) -> (Maybe ByteString, [MessageDeliveryTask], [MessageDeliveryTask])
     toResult (msgBodies, accepted, large, _, _) =
       let encoded = encodeBinaryBatch (reverse msgBodies)
-       in (encoded, reverse accepted, reverse large)
+          body = if null accepted then Nothing else Just encoded
+       in (body, reverse accepted, reverse large)
 
 -- | Encode a batch element for relay groups: ><GrpMsgForward>[/<sigs>]<body>.
 encodeFwdElement :: GrpMsgForward -> VerifiedMsg 'Json -> ByteString

tests/MessageBatching.hs

@@ -12,14 +12,31 @@ import qualified Data.ByteString as B
 import Data.ByteString.Internal (c2w)
 import Data.Either (partitionEithers)
 import Data.Int (Int64)
+import Data.List.NonEmpty (NonEmpty (..))
 import Data.String (IsString (..))
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
+import Data.Time.Clock.System (SystemTime (..), systemToUTCTime)
+import Simplex.Chat.Delivery
+  ( DeliveryJobScope (DJSGroup, jobSpec),
+    DeliveryJobSpec (DJDeliveryJob, includePending),
+    MessageDeliveryTask (MessageDeliveryTask, brokerTs, fwdSender, jobScope, senderGMId, taskId, verifiedMsg),
+    deliveryTaskId,
+  )
 import Simplex.Chat.Messages.Batch
 import Simplex.Chat.Controller (ChatError (..), ChatErrorType (..))
 import Simplex.Chat.Messages (SndMessage (..))
-import Simplex.Chat.Protocol (maxEncodedMsgLength)
-import Simplex.Chat.Types (SharedMsgId (..))
+import Simplex.Chat.Protocol
+  ( ChatMessage (ChatMessage),
+    ChatMsgEvent (XMsgNew),
+    FwdSender (FwdChannel),
+    GrpMsgForward (GrpMsgForward),
+    MsgContent (MCText),
+    VerifiedMsg (VMUnsigned),
+    maxEncodedMsgLength,
+    mcSimple,
+  )
+import Simplex.Chat.Types (SharedMsgId (..), chatInitialVRange)
 import Simplex.Messaging.Encoding (Large (..), smpEncodeList)
 import Test.Hspec
 
@@ -28,6 +45,8 @@ batchingTests = describe "message batching tests" $ do
   testBatchingCorrectness
   testBinaryBatchingCorrectness
   it "image x.msg.new and x.msg.file.descr should fit into single batch" testImageFitsSingleBatch
+  it "does not create a relay delivery body when every task is oversized" testRelayBatchAllLarge
+  it "classifies a task that fits raw but not as a framed singleton as large" testRelayBatchSingletonOverflow
 
 instance IsString SndMessage where
   fromString s = SndMessage {msgId, sharedMsgId = SharedMsgId "", msgBody = s', signedMsg_ = Nothing}
@@ -131,6 +150,37 @@ testImageFitsSingleBatch = do
 
   runBatcherTest' BMJson maxEncodedMsgLength [msg xMsgNewStr, msg descrStr] [] [batched]
 
+testRelayBatchAllLarge :: IO ()
+testRelayBatchAllLarge = do
+  let task1 = deliveryTask 1 "one"
+      task2 = deliveryTask 2 "two"
+      (body_, accepted, large) = batchDeliveryTasks1 chatInitialVRange 1 (task1 :| [task2])
+  body_ `shouldBe` Nothing
+  map deliveryTaskId accepted `shouldBe` []
+  map deliveryTaskId large `shouldBe` [1, 2]
+
+deliveryTask :: Int64 -> T.Text -> MessageDeliveryTask
+deliveryTask taskId text =
+  MessageDeliveryTask
+    { taskId,
+      jobScope = DJSGroup {jobSpec = DJDeliveryJob {includePending = False}},
+      senderGMId = 1,
+      fwdSender = FwdChannel,
+      brokerTs = systemToUTCTime $ MkSystemTime 0 0,
+      verifiedMsg =
+        VMUnsigned
+          (ChatMessage chatInitialVRange Nothing $ XMsgNew $ mcSimple $ MCText text)
+    }
+
+testRelayBatchSingletonOverflow :: IO ()
+testRelayBatchSingletonOverflow = do
+  let task = deliveryTask 1 "overflow"
+      elemLen = B.length $ encodeFwdElement (GrpMsgForward (fwdSender task) (brokerTs task)) (verifiedMsg task)
+      (body_, accepted, large) = batchDeliveryTasks1 chatInitialVRange (elemLen + 2) (task :| [])
+  body_ `shouldBe` Nothing
+  map deliveryTaskId accepted `shouldBe` []
+  map deliveryTaskId large `shouldBe` [1]
+
 runBatcherTest :: BatchMode -> Int -> [SndMessage] -> [ChatError] -> [ByteString] -> Spec
 runBatcherTest mode maxLen msgs expectedErrors expectedBatches =
   it