diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index 2e07282036..ba1f7f7d1f 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/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 -> diff --git a/src/Simplex/Chat/Messages/Batch.hs b/src/Simplex/Chat/Messages/Batch.hs index ed65bd4af7..81861aad74 100644 --- a/src/Simplex/Chat/Messages/Batch.hs +++ b/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: >[/]. encodeFwdElement :: GrpMsgForward -> VerifiedMsg 'Json -> ByteString diff --git a/tests/MessageBatching.hs b/tests/MessageBatching.hs index 05322a0834..00cbbd757b 100644 --- a/tests/MessageBatching.hs +++ b/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