feat: Publish ReadOnlySequence

[PauloHMattos]

Note

This PR was mostly written by Claude Opus 5 with occasional input from me.
The PR description with the changes summary were written just by Claude.

Why

Message bodies are frequently produced in pieces: PipeReader.ReadAsync hands back a ReadOnlySequence<byte>, serializers write into chained buffers, and pooled allocators hand out fixed-size blocks. To publish such a body today the caller must gather it into one contiguous byte[]/IMemoryOwner<byte> first, which costs an extra full-body copy and an allocation on every publish.

That copy is not required by the protocol. AMQP 0-9-1 already frames a body as a series of body frames, so the wire format never needed contiguity - only the client API did.

This adds ReadOnlySequence<byte> publish overloads that write the segments straight to the wire, following the shape of the ReadOnlyMemory<byte> + IDisposable bodyOwner overloads added in #1922 and reusing their memory-ownership semantics.

// body arrives as a multi-segment sequence, e.g. from a Pipe
ReadResult read = await pipeReader.ReadAsync();
await channel.BasicPublishAsync(exchange, routingKey, read.Buffer, bodyOwner: owner);

What changed

Public API (additive only, 7 new members)

Two IChannel overloads, mirroring the existing memory + bodyOwner pair:

ValueTask BasicPublishAsync<TProperties>(string exchange, string routingKey,
    bool mandatory, TProperties basicProperties, ReadOnlySequence<byte> body,
    IDisposable? bodyOwner, CancellationToken cancellationToken = default)
    where TProperties : IReadOnlyBasicProperties, IAmqpHeader;

// ...and the CachedString variant

Plus the five matching IChannelExtensions conveniences (PublicationAddress, and string/CachedString with and without mandatory). Entries were added to both PublicAPI.Unshipped.net8.0.txt and PublicAPI.Unshipped.netstandard2.0.txt. No existing signature changed.

Ownership contract (documented on every new overload, identical to the memory overloads): ownership of bodyOwner transfers to the client, which disposes it exactly once - after the message reaches the wire, or when publication fails. Every segment must stay valid and unmodified until then, and the caller must not reuse, mutate, or release that memory itself. Passing null is valid and selects the copy path.

Internals

OutgoingFrame._body changed from ReadOnlyMemory<byte> to ReadOnlySequence<byte>, and the publish path (ISession.TransmitAsync → Channel.ModelSendAsync → BasicPublishCoreAsync) was unified on that one representation rather than duplicated. All of it is internal.

The existing ReadOnlyMemory path retains a dedicated TransmitAsync/SerializeToFrames overload and incurs no wrapping cost.

For the ReadOnlySequence path, every hot spot short-circuits on IsSingleSegment: Framing.SerializeToFrames delegates single-segment sequences to the existing ReadOnlyMemory implementation, and OutgoingFrame.WriteTo keeps its original span loop. Multi-segment bodies use a new BodySegment.WriteTo(IBufferWriter<byte>, …) that writes the 7-byte frame header, each segment, then the end marker independently - so it never asks the PipeWriter for a span covering more than one segment. Body frame boundaries are sliced out of the sequence independently of segment boundaries.

Both branches are preserved for sequences: with no bodyOwner the body is copied into a single pooled buffer (the caller still owns it); with a bodyOwner only the method + header are pooled and the segments go straight out, then the owner is disposed.

Struct size

OutgoingFrame._body changed from ReadOnlyMemory<byte> (16 bytes) to ReadOnlySequence<byte> (24 bytes), widening the struct from 48 to 56 bytes. This also widens the bounded 128-entry Channel<OutgoingFrame> queue at SocketFrameHandler.cs by 1 KB total.

Validation

body.Length is a long, while AMQP content headers, the tracing APIs, and the pooled frame buffer are all int-based. Publishing now rejects bodies that cannot be framed, disposing the owner first:

• ArgumentOutOfRangeException when body.Length > int.MaxValue, thrown at the public entry point before any publisher-confirm sequence number is consumed.
• ArgumentOutOfRangeException when the total frame set (method + header + body + per-frame overhead) would exceed int.MaxValue, checked against the negotiated MaxPayloadSize. SessionBase.TransmitAsync's existing bytes.Size == 0 branch already disposes the owner on this path.

Incidental fixes

Both are in Framing, both were reachable before this PR, and both are covered by new tests:

1. int overflow in the frame-count math. (length + maxPayloadBytes - 1) / maxPayloadBytes overflows for a body near int.MaxValue, producing a negative count and a bogus buffer size. Rewritten as ((length - 1) / maxPayloadBytes) + 1, which cannot overflow.
2. Empty body with unlimited frame size. GetBodyFrameCount returned 1 for a zero-length body when maxPayloadBytes == int.MaxValue (i.e. a negotiated frame_max of 0). The frame set was sized 8 bytes too large while the write loop emitted no body frame, so the client sent 8 bytes of uninitialized pooled memory and tripped the offset == size assertion in debug builds. It now returns 0.

Benchmarks

projects/Benchmarks/WireFormatting/MethodFraming.cs gained single-segment and multi-segment ReadOnlySequence variants (BasicPublishWriteSingleSegmentSequence, BasicPublishWriteMultiSegmentSequence, and their WithOwner counterparts).

What the code paths guarantee, independent of measured numbers:

• The ReadOnlyMemory overloads keep their own SerializeToFrames/TransmitAsync path and are never routed through a ReadOnlySequence, so the pre-existing publish paths take on no wrapping cost from this change.
• A single-segment ReadOnlySequence short-circuits on IsSingleSegment and delegates to that same ReadOnlyMemory implementation, so it runs the memory path rather than the multi-segment one.
• A multi-segment body with a bodyOwner writes its segments straight to the wire instead of gathering them into one contiguous buffer, trading the gather-copy for independent per-segment frame writes; with no bodyOwner it falls back to a single gather-copy that the caller still owns.

Numeric results are intentionally omitted for now: the earlier table predates the split of the ReadOnlyMemory and ReadOnlySequence paths, and I would rather add fresh figures from a clean benchmark run than quote stale ones.

Testing

New unit tests (no broker required):

• TestSegmentedBodyFraming - a multi-segment body produces byte-identical output to its contiguous equivalent, across maxBodyPayloadBytes smaller than, equal to, and larger than the segment size, plus ragged lengths and int.MaxValue; correct body-frame count, payload limits, and end markers; Size equals bytes actually written; empty sequences, all-empty-segment sequences, and empty segments interleaved with data.
• TestSerializeToFramesWithSequence - copy path and zero-copy path emit identical bytes; single-segment sequences match the memory overload; owner disposed exactly once by frame.Dispose(); both oversize guards reject without leaking the owner (using a MemoryManager<byte>-backed sequence that reports a huge length without allocating).

New integration tests in TestBasicPublishAsync - round trip via the IChannel, extension, and CachedString overloads; a body spanning multiple body frames and multiple segments; confirms disabled; channel already closed; pre-cancelled token; mandatory: true producing a basic.return; oversize rejection. Each asserts the owner was disposed exactly once (waiting for the write loop rather than racing it), and the three pre-existing memory tests were upgraded to the same assertion.

[PauloHMattos]

Question for reviewers: Should a no-owner ReadOnlySequence<byte> overload be added for symmetry with the plain
ReadOnlyMemory<byte> API? Right now callers who don't need ownership transfer pass
bodyOwner: null, which works and takes the copy path, but there's no bodyOwner-free signature as
there is for memory bodies. I left it out to keep the added surface minimal and happy to add it.

[PauloHMattos]

@lukebakken finally got time to contribute this change, but I saw you are putting the work to release 7.3.0 and I don't want to add to the burden, so if you think it's better we can leave this for 7.4.0

cc @danielmarbach I think you might be interested in this

[danielmarbach]

I need to look into it with a fresh mind. Did a quick squim only for now

[danielmarbach]

~9 ns more, from wrapping the ReadOnlyMemory into a ReadOnlySequence inside OutgoingFrame;

I guess having a dual path in there would also add overhead right? And if so do we know how much? So should this be seen as the cost of supporting this additional machinery or how do you see it?

[lukebakken]

@PauloHMattos hello! I added this to 7.3.0 as I expect the OpenTelemetry work to take a while.

[paulomorgado]

Question for reviewers: Should a no-owner ReadOnlySequence<byte> overload be added for symmetry with the plain ReadOnlyMemory<byte> API? Right now callers who don't need ownership transfer pass bodyOwner: null, which works and takes the copy path, but there's no bodyOwner-free signature as there is for memory bodies. I left it out to keep the added surface minimal and happy to add it.

The using statement does not require that the returned disposable is not null.

So, it's better to have a bodyOwner parameter that may be null. And make that the argument's default value.

[lukebakken]

Note

Written by Claude (Anthropic's Claude Code), directed and approved by @lukebakken. The findings were verified against 5f9e25f by build, test runs, and mutation testing.

Nice work. I could not find a correctness bug, and I tried reasonably hard to: beyond reading the diff I ran two mutation tests to check whether the new tests actually have teeth.

What I verified locally at 5f9e25f: client builds 0 warnings / 0 errors with -warnaserror, with both net8.0 and netstandard2.0 assemblies actually produced. Unit 186/186, TestBasicPublishAsync 14/14. I could not build net472 here (no targeting pack), so CI's integration-win32 and sequential-integration-win32 are what cover that TFM - both green.

The dispose contract is genuinely pinned. I replaced the guarded if (!bodyOwnerTransferred) in the finally with an unconditional bodyOwner?.Dispose() and 9 of 14 tests failed. DisposalTracker is the reason: Interlocked.Increment plus a 250ms window after the first disposal to catch a second one. That is a real improvement over the Assert.True(body.Disposed) on a plain bool that main had, which could not detect a double-dispose at all. The ownership handoff reads correctly too - the flag is set immediately before ModelSendAsync, and TransmitAsync discriminates on bytes.Size == 0 for the never-captured case.

The one thing I would like fixed before merge is an untested invariant in the multi-segment writer - inline below. Everything else is minor.

Also confirmed: the empty-body fix is real (on main, GetBodyFrameCount(int.MaxValue, 0) returns 1 because the maxPayloadBytes == int.MaxValue early-return precedes any length check, and Connection.Commands.cs:180 sets MaxPayloadSize = int.MaxValue whenever the broker negotiates frame_max = 0), and it is pinned by EmptyBodyWithUnlimitedFrameSizeProducesNoBodyFrames. Your "Incidental fixes" write-up matches what I found in the code.

[PauloHMattos]

Thank you all for the thorough review! It's genuinely a pleasure working with this level of collaboration.
I just pushed a few commits that I hope address all the concerns for now. I just wasn't able to re run the benchmarks yet, but will do later today and update the results in the PR body.

Looking forward to the next round of reviews