Stream Mediator Framework message serialization

[Copilot]

Mediator messaging buffered payloads repeatedly across serialization, compression, DataBus offload, and transport encoding. This change introduces bounded streaming pipelines while retaining only threshold-sized pooled buffers.

Changes

• Streaming codecs

  • Replace buffered codec APIs with asynchronous PipeReader/PipeWriter contracts.
  • Stream deserialization directly from inline, compressed, or DataBus sources.
  • Update generated dispatch to await payload deserialization.

• Bounded outgoing pipeline

  • Connect serialization, optional compression, and destination writing through bounded pipes.
  • Buffer only until the compression and DataBus thresholds.
  • Replay the threshold prefix into DataBus, then stream remaining bytes directly.
  • Use pooled arrays for retained inline payloads.

• Transactional DataBus

  • Add streamed write sessions with commit metadata containing length and SHA-256.
  • Validate attachment integrity while reading.
  • Remove incomplete or rejected attachments, including post-commit envelope failures.

• Payload ownership

  • Remove payload storage from MessagingOutgoingContext.
  • Introduce explicit ownership and disposal for pooled inline payloads.
  • Preserve buffering only where required by transport or outbox contracts.

• Transport efficiency

  • Account for native envelope overhead when determining inline limits.
  • Avoid redundant Service Bus payload copies.
  • Encode Storage Queue Base64 through a pooled canonical buffer.

await codec.SerializeAsync(message, pipe.Writer, cancellationToken);
var message = await codec.DeserializeAsync<T>(pipe.Reader, cancellationToken);

[Copilot]

Pull request overview

Introduces streaming, bounded message serialization/deserialization for MediatorFramework messaging to reduce repeated buffering across codec, compression, DataBus offload, and transport encoding.

Changes:

• Reworks IMessagingCodec and dispatch to async PipeReader/PipeWriter streaming APIs.
• Adds transactional DataBus write sessions with committed integrity metadata (length + SHA-256) and cleanup on failure.
• Updates transports and envelope codecs to better account for native overhead and reduce payload copying.

Reviewed changes

Copilot reviewed 29 out of 29 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
tests/Ark.Tools.MediatorFramework.Tests/MessagingStreamingTestExtensions.cs	Adds test helpers to adapt streaming codecs to existing buffered test patterns.
tests/Ark.Tools.MediatorFramework.Tests/MessagingRuntimeTests.cs	Updates dispatcher tests to await async payload deserialization.
tests/Ark.Tools.MediatorFramework.Tests/MessagingCompressionAndDataBusTests.cs	Expands coverage for claim-check cleanup + async stream payload reader behavior.
tests/Ark.Tools.MediatorFramework.Tests/GeneratorSnapshotTests.cs	Updates generator snapshot expectations to async payload deserialization calls.
src/mediator-framework/Ark.Tools.MediatorFramework/MessagingPipelineContracts.cs	Removes buffered payload from incoming/outgoing contexts to support streaming ownership.
src/mediator-framework/Ark.Tools.MediatorFramework/IMessagingDataBus.cs	Introduces transactional write sessions + delete API; adjusts DataBus contract docs.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/StorageQueueMessagingTransport.cs	Computes inline payload budget accounting for Storage Queue envelope overhead.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/StorageQueueEnvelopeCodec.cs	Switches canonical encoding to a pooled fixed buffer writer and direct Base64 string generation.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/ServiceBusMessagingTransport.cs	Adds inline payload budget accounting and reduces redundant payload copies for single-segment payloads.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/ProtobufMessagingCodec.cs	Updates protobuf codec to streaming PipeReader/PipeWriter contract.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/ProtobufContractRegistry.cs	Changes protobuf parser registry to stream-based parsing.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagingStreamPayloadReader.cs	Implements async streaming deserialization by opening/disposing payload streams per read.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagingPayloadSender.cs	Implements bounded streaming pipeline: serialize → optional compress → inline/DataBus destination, with pooled inline ownership.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagingPayloadReceiver.cs	Returns replayable stream factory-backed payload reader rather than buffering upfront.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagingDispatcher.cs	Stops passing buffered payload into MessagingIncomingContext.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagingBus.cs	Adopts MessagingOutgoingPayload ownership/disposal and updates transport/outbox writes.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagePackMessagingCodec.cs	Updates MessagePack codec to async stream-based serialization/deserialization.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/JsonMessagingCodec.cs	Updates JSON codec to async stream-based serialization/deserialization.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/InMemoryMessagingTransport.cs	Implements inline payload budget API (unbounded).
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/InMemoryMessagingDataBus.cs	Implements transactional write sessions and delete for in-memory DataBus with integrity validation.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/IMessagingTransport.cs	Adds GetMaximumInlinePayloadBytes API for transports to report header-adjusted payload budget.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/IMessagingPayloadReader.cs	Replaces buffered payload reads with async typed deserialization.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/IMessagingCodec.cs	Replaces buffered codec APIs with streaming PipeReader/PipeWriter contracts.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/HashingWriteStream.cs	Adds streaming SHA-256 hashing writer used by DataBus write sessions.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/CompressionSwitchingBufferWriter.cs	Removes buffered compression-switching writer in favor of streaming pipeline.
src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/AzureBlobMessagingDataBus.cs	Implements transactional streamed writes with post-write metadata commit and abort cleanup.
src/mediator-framework/Ark.Tools.MediatorFramework.Generators/MessagingNetworkGenerator.cs	Updates generated dispatch to await payload deserialization.
samples/Ark.MediatorFramework.Sample/test/Ark.MediatorFramework.Sample.Tests/MessagingBusSampleTests.cs	Updates sample tests to deserialize via streaming codec APIs.
docs/mediator-framework/guide/serialization.md	Updates documentation to describe streaming codec APIs and bounded send/receive behavior.

Suppressed comments (2)

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagingPayloadSender.cs:25

• Public constructor XML docs lost the <param> descriptions in this change. This repo requires XML docs for public APIs; restore parameter documentation so generated docs and analyzers remain consistent.

    /// <summary>Creates a payload sender.</summary>
    public MessagingPayloadSender(
        IMessagingDataBus dataBus,
        MessagingNetworkOptions network,
        CompressionAlgorithm algorithm,
        int compressionMinimumSizeBytes)
    {

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagingPayloadSender.cs:46

• BuildOutgoingPayloadAsync now returns an owned MessagingOutgoingPayload that must be disposed to return pooled buffers, but the method no longer documents parameters/return semantics. Restore XML docs (including disposal requirement) to keep the public API contract clear.

    /// <summary>Serializes, optionally compresses, and claim-checks a message.</summary>
    public async Task<MessagingOutgoingPayload> BuildOutgoingPayloadAsync<T>(
        T message,
        IMessagingCodec codec,
        IMessagingTransport transport,
        IDictionary<string, string> headers,
        CancellationToken ctk)
        where T : class
    {

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.

[github-code-quality]

Code Coverage Overview

Languages: C#

C# / Ark.Tools

The overall line coverage in commit 0ee0f53 in the copilot/review-end2e... branch remains at 47%, unchanged from commit 570f997 in the master branch.

Show a line coverage summary of the most impacted files.

File	master 570f997	copilot/review-end2e... 0ee0f53	+/-
/home/runner/wo...ayloadReader.cs	79%	54%	-25%
/home/runner/wo...ayloadSender.cs	91%	78%	-13%
/home/runner/wo...ineContracts.cs	69%	63%	-6%
/home/runner/wo...agingDataBus.cs	83%	78%	-5%
/home/runner/wo...loadReceiver.cs	72%	68%	-4%
/home/runner/wo...nvelopeCodec.cs	84%	83%	-1%
/home/runner/wo...ingTransport.cs	78%	79%	+1%
/home/runner/wo...agingDataBus.cs	70%	74%	+4%
/home/runner/wo...MemTransport.cs	86%	93%	+7%
/home/runner/wo...gWriteStream.cs	0%	49%	+49%

---

_{Updated August 30, 2026 17:35 UTC}

[Copilot]

Code Coverage Overview

...

Reviewed. Overall line coverage remains unchanged at 47%; focused Mediator Framework validation passes all 259 tests.

[Copilot]

Pull request overview

Copilot reviewed 29 out of 29 changed files in this pull request and generated 2 comments.

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/HashingWriteStream.cs:58

• After _completeHash() is called, HashingWriteStream still allows further writes, which can mutate the underlying stream after the committed hash/length were captured. Guard Write/WriteAsync to throw once hashing has completed.

    public override void Write(ReadOnlySpan<byte> buffer)
    {
        _inner.Write(buffer);
        _hash.AppendData(buffer);
        _bytesWritten += buffer.Length;
    }

[Copilot]

Pull request overview

Copilot reviewed 29 out of 29 changed files in this pull request and generated 2 comments....

Reviewed the suppressed finding and confirmed it. HashingWriteStream now reports CanWrite == false and rejects synchronous/asynchronous writes after hash completion. Added CompletedDataBusSessionRejectsFurtherWrites regression coverage.

[Copilot]

Pull request overview

Copilot reviewed 29 out of 29 changed files in this pull request and generated no new comments.

Suppressed comments (6)

Previously missed (5) — in code that hasn't changed since the last review.

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/ServiceBusMessagingTransport.cs:45

• GetMaximumInlinePayloadBytes can return a negative payload budget when headers alone exceed the Service Bus ceiling. That negative value then propagates into sender inline buffering decisions; clamp the budget at 0 to match the interface default behavior.

    /// <inheritdoc />
    public long? GetMaximumInlinePayloadBytes(IReadOnlyDictionary<string, string> headers)
    {
        return _maximumMessageBytes - MeasureNative(headers, ReadOnlySequence<byte>.Empty);
    }

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/StorageQueueMessagingTransport.cs:90

• GetMaximumInlinePayloadBytes can go negative when the canonical header footprint exceeds the Storage Queue canonical limit. A negative payload budget is not meaningful and causes downstream inline-limit calculations to underflow; clamp to 0.

    /// <inheritdoc />
    public long? GetMaximumInlinePayloadBytes(IReadOnlyDictionary<string, string> headers)
    {
        return StorageQueueLimits.MaximumNormalCanonicalBytes
            - StorageQueueEnvelopeCodec._measureCanonical(headers, ReadOnlySequence<byte>.Empty);
    }

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/ProtobufMessagingCodec.cs:49

• DeserializeAsync uses await Task.CompletedTask solely to satisfy the compiler, which adds an unnecessary async state machine. Since parsing is synchronous, make this method non-async and return Task.FromResult(...) instead.

        var result = parse(reader.AsStream(leaveOpen: true));
        await Task.CompletedTask.ConfigureAwait(false);
        return result;

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/InMemoryMessagingDataBus.cs:57

• OpenWriteAsync is implemented with await Task.CompletedTask even though it performs no async work. This creates an unnecessary async state machine; return a completed task instead.

This issue also appears on line 113 of the same file.

        _removeExpired();
        var id = Guid.NewGuid().ToString("N");
        await Task.CompletedTask.ConfigureAwait(false);
        return new WriteSession(this, id);

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/MessagingPayloadReceiver.cs:124

• PreparePayloadReaderAsync currently awaits Task.CompletedTask but does no async work. Removing async/await avoids an unnecessary state machine while keeping the same API shape.

    public async Task<MessagingStreamPayloadReader> PreparePayloadReaderAsync(
        IReadOnlyDictionary<string, string> headers,
        ReadOnlySequence<byte> transportPayload,
        IMessagingCodec codec,
        CancellationToken ctk)
    {
        ArgumentNullException.ThrowIfNull(codec);
        await Task.CompletedTask.ConfigureAwait(false);
        return new MessagingStreamPayloadReader(
            token => PreparePayloadAsync(headers, transportPayload, token),
            codec);
    }

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/InMemoryMessagingDataBus.cs:119

• DeleteAsync uses await Task.CompletedTask even though it has no asynchronous operations. Return Task.CompletedTask directly to avoid the async state machine.

    public async Task DeleteAsync(string attachmentId, CancellationToken ctk)
    {
        ArgumentException.ThrowIfNullOrEmpty(attachmentId);
        ctk.ThrowIfCancellationRequested();
        _attachments.TryRemove(attachmentId, out _);
        await Task.CompletedTask.ConfigureAwait(false);
    }

[Copilot]

Pull request overview

Copilot reviewed 28 out of 28 changed files in this pull request and generated 2 comments.

[Copilot]

Pull request overview

Copilot reviewed 28 out of 28 changed files in this pull request and generated no new comments.

Suppressed comments (2)

Previously missed (1) — in code that hasn't changed since the last review.

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/InMemoryMessagingDataBus.cs:169

• CompleteAsync stores MemoryStream.GetBuffer() into the attachment, which retains the stream's capacity (often > Length) and can significantly inflate memory usage for large attachments. Store a right-sized array so in-memory DataBus doesn't keep extra unused bytes alive.

            var hash = _stream._completeHash();
            var content = _buffer.GetBuffer();
            var length = checked((int)_buffer.Length);
            _owner._attachments[_id] = new Attachment(
                content,

src/mediator-framework/Ark.Tools.MediatorFramework.Messaging/ProtobufMessagingCodec.cs:49

• DeserializeAsync uses await Task.CompletedTask solely to satisfy the async signature, which is a no-op and obscures intent. Prefer an awaited Task.FromResult(...) return so the method remains async without dummy work.

        var result = parse(reader.AsStream(leaveOpen: true));
        await Task.CompletedTask.ConfigureAwait(false);
        return result;