Remove the per-write allocations from the chunked object AOF path, and bound chunk buffering - #2182

The --op / --opparams option is parsed in the host process, where BaseConfig
mutates the OperationsBase.Params{None,ACL,AOF,AAD} statics, and the provider
yielded only the selected values. BenchmarkDotNet identifies a parameter value by
its position in the provider's sequence, and the generated per-job process
re-invokes the provider and selects by that position. That process never runs
Program.Main, so its statics keep their default true values and it yields all
four. With --op aof the host yielded [AOF] and recorded position 0, while the
child yielded [None, ACL, AOF, AAD] and selected None at position 0, so a row
labelled AOF measured the None configuration.

OperationParamsProvider now yields all four unconditionally, so the sequence is
identical in both processes, and OperationParamsFilter applies the selection as
an IFilter registered in BaseConfig. BenchmarkDotNet fixes parameter positions
while building the cases and applies filters afterwards, so excluding a case
cannot renumber the remaining ones. Every configured filter must admit a case,
so this composes with the --filter globs.

OperationParams takes useAad explicitly rather than defaulting it, and rejects
more than one aspect per value, which is what every provider already yields and
what lets the filter map a value onto a single Params* flag. The providers pass
the arguments by name. The mutual-exclusion check for the two auth modes is
subsumed by that rule and is retained commented out, to be restored if
combinations are ever allowed. The check throws rather than asserting because
benchmarks run against a Release build.

Verified by inspecting the generated per-job source: with --op aof it selects
position 2, which is AOF in the unconditional sequence, and an unfiltered run
emits positions 0 through 3. CI passes no --op and is unaffected.

The three chunked-record paths that accumulate a serialized object value each
built a List<byte[]>, one ToArray() per arriving segment, and held every array
until the value was complete. Segment sizes follow transport framing, so the
arrays were odd-sized and, at the sizes involved, landed on the large object heap
and were promoted while the whole payload stayed reachable. A 6 GB object
retained roughly 1,536 such arrays.

PooledChunkList packs arriving spans into uniform 64 KB buffers rented from
ArrayPool<byte>.Shared: below the 85,000-byte large-object-heap threshold and
within the pooled size classes, so accumulating an arbitrarily large value
allocates nothing at steady state. Packing also decouples buffer size from the
transport's segment size, so odd-sized segments no longer produce odd-sized
arrays. The buffers are exposed as a ReadOnlySequence<byte> exactly as before, so
a value may still exceed 2 GB and is still deserialized as a stream with no
contiguous copy.

Converted sites:

  AofChunkedRecordReader.AppendChunk        AOF replay
  ChunkedRecordReassembler Phase.ObjectData migration import
  MigrationChunkWriterAccumulator.Consume   migration export

ReadOnlySequenceBuilder now links chunks as ReadOnlyMemory<byte> so a segment can
be bounded to its valid bytes within a larger rented array; its List<byte[]>
overload had no remaining callers and is removed. MigrationChunkWriterAccumulator
exposes ChunkCount/GetChunk in place of the List<byte[]> property, and ValueLength
reads the accumulated total rather than summing per-chunk lengths.

Buffers are returned on reset, and for AOF replay immediately after the value is
deserialized. A missed return is not a correctness problem, since the array is
simply collected instead of reused, but returning a buffer that is still
referenced would be, so returns are placed only where the sequence is provably no
longer in use.

This does not change how much of a value is held at once: the accumulated bytes
are still a full second copy alongside the materialized object.

A chunked object value was accumulated in full and then deserialized, so the
serialized bytes were a complete second copy of the object alongside the
materialized form. This adds a path that deserializes the value while its chunks
arrive, so only a bounded window of serialized bytes is held.

GarnetObjectSerializer deserialization is synchronous - a BinaryReader driving a
constructor - so it cannot be parked on an await and runs on its own thread while
the replay thread feeds it. StreamingObjectValueDeserializer joins the two with a
bounded queue of pooled buffers: the replay thread copies each arriving segment
in, blocking only while the queue is full, and the worker drains it through a
Stream. The copy means the worker never holds a pointer into the replay entry
buffer or the log's own memory, both of which are reclaimed as soon as the record
is processed. Deserialize(Stream) builds a reader local to the call, because
BinaryObjectSerializer keeps its reader in an instance field that interleaved
deserializations would otherwise clobber.

Streaming is offered only where blocking is safe, which is narrower than it
first appears:

  - canStream is passed by the caller and defaults to false. RecoverLogDriver and
    RecoverReplayTask pass !isProtected, because the bulk-consume path runs the
    consumer under the log epoch when the record is resident in the in-memory log
    buffer, and blocking there would stall log truncation and page shifting.
    SingleLogRecover passes true, having copied the entry out. Replication replay
    passes false: it is fed by the network, so the blocked thread would be the one
    that must deliver the remaining chunks.
  - Records that will be deferred are excluded. A value buffered for a
    transaction or the fuzzy region would be held in materialized form for the
    whole deferral, which for a collection is several times the serialized bytes
    it replaces, and would have to be disposed on every discard path.
  - Admission is non-blocking and capped. A replay thread that waited for a slot
    could not reach the chunks that would release one, since the in-flight values
    are fed by that same thread.

IHeapObject is IDisposable, so a materialized value that never reaches the store
is disposed: on the skip and fuzzy-buffer paths, and for records still in
progress when a replay context is torn down. Ownership is released once the store
takes the object, so a published value is not disposed twice.

Note that this is currently inert for the dominant workload. Whole-object upserts
are produced by RENAME, which wraps itself in an internal transaction to lock
both keys, so replay sees an open transaction when the value's chunks arrive and
the deferral exclusion above declines to stream. The tests pin that behaviour
rather than assert an outcome the design cannot produce.

The ring a streamed object value is serialized through was sized per write as
min(HeapMemorySize, aofPageSize / 2), reaching 16 MB on the large object heap for
a single write with the default 32 MB AOF page. Sizing it down to stay under the
LOH threshold would have been the wrong fix: because a chunk record is allocated
per drain, the ring also bounds how much value data one record can carry, so a
small ring multiplies the emitted record count. A 6 GB object drains into roughly
1,536 records at 4 MB and roughly 98,304 at 64 KB.

The ring is now rented from the log's SectorAlignedBufferPool, which is not
GC-heap memory, so its size carries no large-object-heap cost and the question
reduces to picking a good record size. IStreamBuffer.BufferSize (4 MB) is the
size the object-log streaming path already uses for this job. It also sits at or
above MinPartialAllocSize, which matters because page-tail packing only splits an
allocation when both halves reach that size: a ring below 1 MB silently stops
chunk records from filling a page tail.

Renting per write also means a cached serializer holds no buffer between writes.
The AOF path caches its serializer thread-statically and indefinitely, so a
managed ring would stay pinned to that thread for the life of the process even
while idle. The pool is passed per write rather than held, because a cached
instance is keyed only by thread and input type and so may be reused across logs
with different pools.

The network path (migration, diskless replication) keeps a managed ring: those
instances live for one operation and use the ring throughout, so renting would
gain nothing, their sizes are caller-chosen rather than pool size classes, and
they have no log from which to obtain a pool. Structurally the pool arrives
through SetObjectWriteTarget, which only the object drive mode reaches, so the
manual mode has nowhere to receive one; BeginSerialize now asserts this rather
than leaving a null ring to fail later. The constructor documents the full
choice.

The streaming path deserialized a chunked object value while its chunks arrived,
so the serialized bytes were never held as a second whole copy. It never engaged,
and could not: the exclusions it needs and the records it would apply to have no
overlap.

  - Whole-object upserts come from RENAME, which wraps itself in an internal
    transaction to lock both keys. Replay is therefore inside a transaction when
    the chunks arrive, and the record is buffered until commit rather than
    dispatched. A buffered record holds its value in whichever form it was
    accumulated, and for a collection the materialized object is several times
    the size of the bytes it would replace, so streaming is declined.
  - Deserialization is synchronous, so streaming needs a second thread fed by the
    replay thread. That thread cannot block while it holds the log epoch (the
    bulk-consume path runs the consumer under it for records resident in the log
    buffer) or while it is the thread that must deliver the remaining chunks
    (replication replay, fed from the network).

Between them these cover every path a chunked object value currently arrives on,
so the machinery was inert: a worker pool, a bounded handoff, a termination
protocol and an object-ownership model, none of it reachable. Carrying it as dead
code costs more in review and maintenance than it saves, and it can be recovered
from history if a non-transactional whole-object upsert path appears.

The reasoning is recorded as comments where object chunks are accumulated, in the
AOF reader and in both migration directions, so the question does not have to be
re-derived. GarnetObjectSerializer.Deserialize(Stream), added only for this, goes
with it; the ReadOnlySequence overload remains.

PooledChunkList rented 64 KB blocks from ArrayPool<byte>.Shared. The size was
chosen to stay under the large-object-heap threshold, which applies to a plain
byte[] but is the wrong constraint here: SectorAlignedBufferPool hands out pinned
arrays that it reuses, so a large block costs no repeated large-object
allocation. Renting from it instead frees the buffer size to be chosen for what
it actually governs, the number of segments in the resulting ReadOnlySequence.

At 4 MB, matching the size the object-log streaming path uses, a 6 GB value forms
roughly 1,536 segments rather than roughly 98,304, each of which is a
ReadOnlySequenceSegment allocation. Buffers are shared across every chunk
accumulation, so AOF replay and cluster migration draw from the same blocks.

SectorAlignedMemory gains AsReadOnlyMemory(offset, length). Unlike the Span
accessors on that type it is derived from the backing array and aligned_offset
rather than from aligned_pointer: Memory<T> stores an object reference and cannot
point at a raw address, so a pointer-based form is not expressible without a
MemoryManager<T>, and going through the array avoids that object entirely. This
is what lets a sequence segment span pooled buffers at no per-buffer cost.

Also removes three members with no callers: GetArrayAndUnalignedOffset on both
SectorAlignedMemory and BlittableFrame, and AvailableSpan and AvailableValidSpan
on SectorAlignedMemory. (DiskReadBuffer declares its own AvailableSpan, which is
the one ObjectLogReader uses; SectorAlignedMemory's was unreferenced.)

Corrects the chunked-ring comment, which claimed the pool's memory "is not
GC-heap memory". It is: SectorAlignedMemory is a pinned array from
GC.AllocateArray<byte>(size, pinned: true). The properties that justify the ring
size are that the blocks are pooled and reused and land on the pinned object heap
rather than the large object heap, not that they are off-heap.

The AOF and migration record-layout docs still described the write ring as a fixed 64 KB buffer sized to stay under the large-object-heap threshold, and object-value accumulation as a List<byte[]>. Both now describe the pooled 4 MB ring and PooledChunkList, including the two constraints that fix the ring size (it bounds how much value data one chunk entry carries, and must stay at or above MinPartialAllocSize for page-tail packing).

Also records why an object value is accumulated rather than deserialized as its chunks arrive, on both the AOF replay side (RENAME runs as an internal transaction, so the record is buffered until commit; and the feeding thread may hold the log epoch or be the one delivering the chunks) and the migration receive side (chunks span network commands, so blocking the receive thread would deadlock the connection).

The doc went straight from framing into chunk header bit layouts, so how chunks actually move was only recoverable by reading the whole section, and the paths that reuse the same machinery (cluster migration, diskless replication) were documented only in the companion doc.

New section 5 maps every path in one table - producer side, carrier, consumer side - and separates the two buffer roles that are easy to conflate: the producer ring, whose size sets how many chunks are emitted and so affects the on-log/on-wire shape, and the consumer accumulator, whose size affects only segment count. It also records why migration must accumulate while diskless replication can stream (synchronous versus asynchronous send, and the store epoch not surviving an await), and calls out that disk-based replication ships checkpoint file byte ranges rather than records, so none of this applies to it until the replica replays the AOF.

Existing sections 5 and 6 renumber to 6 and 7; internal anchors updated.

The builder had a single real consumer (PooledChunkList.AsSequence); the AOF reader's GetValueSequence called it directly rather than through AsSequence. Move the segment-linking into AsSequence and route the AOF reader through it, so the sequence is built in one place and the chunk buffers stay encapsulated.

The paths that discard a chunk accumulation overlap by design: a chunked
operation releases its buffers as it is dispatched, and the container that held
it is then cleared, which releases them again. Correctness therefore rests on
`PooledChunkList.Reset()` being idempotent, and the failure it must not have is
worse than a leak: returning one block twice pushes it onto a pool free list
twice, and those lists are singly linked through the block itself, so the chain
is corrupted and two renters are later handed the same memory. `CHECK_FREE` is
not compiled in, so nothing would report it.

`Reset()` was idempotent only because `buffers.Clear()` followed the loop.
It now drops each reference before handing the buffer back, so no block can be
returned twice even if the list is reset again while the loop is in progress.

`TransactionGroup.Clear()` is renamed to `Discard()`, naming what it does now
that it releases buffers rather than only dropping references, and matching
`AofReplayContext.DiscardFuzzyRegionBuffer` and
`AofChunkedRecordReader.DiscardInProgress`. The comment at its caller records
why discarding an already-dispatched group is safe, since that is the overlap
the idempotence requirement comes from.

`PooledChunkListTests` covers the ownership contract: reset twice, reset and
reuse across rounds, round-trip across buffer boundaries with irregular append
sizes, and the empty case. Each rents past the pool afterwards and asserts no
block is handed out twice, which is the observable form of the corruption.
Verified to fail (7 of 8 cases) when `Reset` is made to return each buffer
twice.

`BeginReplayOp` was called outside the try that releases the accumulator's
pooled chunk buffers, so a throw from it abandoned the rental. It can throw:
it waits on the vector-replication event, and only `ObjectDisposedException`
is swallowed there. Rather than assert a no-throw contract on code this method
does not own, both `ShouldSkipRecord` and `BeginReplayOp` now run inside the
try.

`isBuffered` is declared outside the try so the finally can read it even if
`ShouldSkipRecord` throws, in which case it retains its initial `false` and the
record is released. That is the correct bias: it is set only after
`AddFuzzyRegionOperation` returns, so a throw from the add means the record
never reached the fuzzy-region buffer and is still this method's to release.
This also removes the duplicate release the old early-return branch carried,
leaving one release site.

`BeginReplayOp` gains a remark recording that it may throw and that a caller
holding a resource must call it from within the releasing scope, so the
constraint is not rediscovered.

Both `ShouldSkipRecord` overloads now say what they do. The name suggests a
single outcome, but returning true covers two: the record belongs to a prior
checkpoint version and is dropped, or it is a new-version record inside the
fuzzy region and is buffered for replay at the end of that region. Only the
chunked overload needs to tell them apart, because only its record owns
something that must be released; the non-chunked one buffers a copy of the
record bytes and owns nothing. Renamed `buffered` to `isBuffered`.

`AofChunkedRecordReader.DiscardInProgress` is renamed to
`DiscardInProgressAccumulations`: the original parses as "a discard is in
progress" rather than "discard the in-progress accumulations".

A checkpoint brackets a span of the AOF with CheckpointStartCommit and
CheckpointEndCommit. A replica replaying that span - the fuzzy region - defers
the records belonging to the new store version, takes its own checkpoint at the
end marker, and only then replays what it deferred. A transaction is deferred
differently from an ordinary write: its operations are collected into a
TransactionGroup, and the group, not the individual records, is what must be
set aside and replayed.

A transaction committing inside that span was lost, and could wedge replay
entirely. Four defects compounded:

- AddToFuzzyRegionBuffer enqueued the group, and the ClearSessionTxn() that
  followed emptied that same object, so the queued group held no operations.
- ProcessFuzzyRegionTransactionGroup, holding the only Dequeue, was called from
  a switch arm that AddOrReplayTransactionOperation's early return made
  unreachable for every TxnCommit, so the queue was never drained.
- The deferred commit marker was dispatched through ReplayOp, whose switch has
  no TxnCommit case, so replay threw "Unknown AOF header operation type
  TxnCommit" and the replica stopped applying the stream.
- Groups were deferred on inFuzzyRegion alone. Ordinary records are deferred
  only when storeVersion > CurrentVersion; a group of the old version was
  therefore deferred past the replica's own checkpoint and then skipped as
  stale.

Each of the four is sufficient on its own to lose the transaction.

The mechanism, in six parts:

1. The defer decision applies the version test ShouldSkipRecord uses for a
   single record, to the group as a unit, keyed on the commit record's version.
   An old-version group replays immediately, as it must: the checkpoint being
   taken captures that version.

2. The two branches no longer share a tail. Deferring removes the session's
   entry from activeTxns without clearing it, so the group reaches the buffer
   intact and ownership passes to the buffer. Replaying immediately still
   clears, releasing what the operations hold.

3. ReplayOperation carries a SequenceNumber, set from the commit's
   logAddressSequenceNumber when the marker is recorded. The value is in hand
   at defer time and not recoverable from the marker bytes later, and
   ProcessTransactionGroup needs it for the multi-log release barrier.

4. ProcessFuzzyRegionOperations recognizes the deferred commit marker - already
   recorded, previously inert - and replays its group at the marker's position
   in the deferred stream. The position is what orders the transaction among
   the surrounding records; draining the queue separately would apply groups
   out of order. This also keeps the marker out of ReplayOpDispatch, which is
   what threw.

5. ProcessFuzzyRegionTransactionGroup becomes that replay's only entry point.
   It keeps the FIFO dequeue, since markers and groups are enqueued in
   lockstep, and adds two things: an empty-queue check that reports a lost
   transaction rather than surfacing Dequeue's InvalidOperationException, and a
   finally that releases the group's operations once nothing else refers to
   them. The now-unreachable arm in ProcessAofRecordInternal is removed.

6. Abandoning a region discards the queued groups along with the markers. A
   group is reachable only through its marker, so dropping markers alone would
   leave the two out of step and the next region's markers would dequeue the
   wrong groups. This is reachable: a CheckpointStartCommit arriving while
   already in a fuzzy region abandons the previous one.

ClusterReplicationFuzzyRegionTxnTests covers this. The region is not the
checkpoint: the markers are written entering IN_PROGRESS and WAIT_FLUSH, so it
spans only the wait for previous-version transactions to drain. Left alone that
is about two transactions wide, and a test built on small transactions passes
against the unfixed code. One writer therefore issues large transactions, each
holding the drain open while the other writers' short transactions commit
inside it. The test fails on unmodified main and passes with the fix.

In cluster mode the keys of each command queued into a MULTI are recorded as
the transaction is built, and verified together at EXEC. They were recorded as
pointers into the network receive buffer, and kept valid by deferring a copy:
when a later queued command observed that recvBufferPtr had changed, it copied
every key recorded so far into the transaction's scratch buffer.

That deferral is unsound, because the buffer is not merely replaced. A receive
buffer that fills is grown by DoubleNetworkReceiveBuffer, which allocates the
larger buffer, copies the bytes, and then DISPOSES the old buffer, returning it
to the pool it came from. The pointer change the copy keys off of is the same
event that releases the memory the copy reads. By the time the next queued
command triggers it, the source bytes are gone.

A 6000-command MULTI of a single key reported

    CROSSSLOT Keys in request do not hash to the same slot

which cannot be true of one key. The deferred copy ran, found 4680 keys
recorded, and every one of the 4680 already read back as zero bytes; those are
the keys EXEC then verified.

SaveKeyArgSlice now copies each key into txnScratchBufferAllocator as it is
recorded, while the bytes are still live, so a recorded key never points into
the receive buffer and no later copy is needed. CopyExistingKeysToScratchBuffer,
saveKeyRecvBufferPtr and both pointer-comparison checks are removed.

This also covers the other ways the buffer moves underneath a retained pointer:
ShrinkNetworkReceiveBuffer likewise disposes the old buffer, and
ShiftNetworkReceiveBuffer compacts in place, moving the bytes without changing
the buffer's address at all - which a pointer comparison cannot detect. Copying
at record time is independent of all three.

Rejecting a valid transaction is the benign symptom. The same bytes feed slot
verification, so keys read out of a recycled buffer that happen to hash
consistently would pass a transaction whose keys really do span slots.

ClusterLargeTransactionSlotTests covers same-slot and single-key transactions
large enough to grow the buffer, and asserts that a genuinely cross-slot
transaction is still rejected. That case is driven through the raw RESP client
because StackExchange.Redis rejects a cross-slot transaction client-side and
never reaches the server. The two large cases fail without this change.

WATCH identifies a key in the version map by its hash alone. The map is a
fixed-size array indexed by hash & sizeMask that stores no key bytes, so keys
whose hashes alias to the same slot share a version and a write to either is
seen by a watcher of the other. That is a deliberate tradeoff, and it errs
conservatively: a collision can add a spurious EXEC abort but can never mask a
real modification.

Record the rationale on WatchedKeySlice.hash, document the remaining
WatchedKeySlice fields, and add brief pointers from WatchVersionMap and
WatchedKeysContainer. Comment-only.

PooledChunkList.GetChunk(int) becomes an indexer. The type is the chunk
collection, so this[i] alongside Count reads as the collection access it
is, and matches how callers already use it, in a loop over Count.

MigrationChunkWriterAccumulator keeps a named GetChunk. That type is not a
collection; it also carries a key and an overflow value, so an unnamed
indexer there would not say which of the three it returns.

AofReplayContext.DiscardFuzzyRegionBuffer and AddToFuzzyRegionBuffer become
internal. Both are new in this branch and both are reached only from
AofReplayCoordinator in this assembly, which is also why the chunked
reader's DiscardInProgressAccumulations was already internal.

ClusterLargeTransactionSlotTests built a ConnectionMultiplexer per test and
never disposed it, so each of the fixture's seven cases left one running.

Every cluster fixture in the assembly binds the same reserved ports, and
the leaked multiplexers have AbortOnConnectFail false, so each kept
retrying node 0's endpoint and reconnected to whichever server bound it
next. ClusterManagementTests.ClusterClientList counts the client
connections on each node and allows at most two, so it saw the seven
zombies plus its own and failed with normal=8 on nodeIx=0.

That test passed alone and passed with its own fixture, and failed only
once the leaking fixture had run, which made it look like load-related
flakiness. Running the two fixtures together reproduces it in seven
seconds.

Hold the multiplexer in a field and dispose it in TearDown, before the
servers so it stops reconnecting while they shut down.

Also drop a duplicated summary block on LargeCrossSlotTransactionIsRejected.