feat(client): keep several write batches of a shard in flight (#1432) - #1537

Fixes the write half of #1432: the Go client sent the next write batch of a shard only after the previous one was answered, so a shard got one batch per round trip no matter how many callers there were.

Changes

Six commits:

1. Batcher window (oxia/batch). A batcher with MaxBatchesInFlight sends AsyncBatch batches without waiting for their responses, up to the window, and completes them in send order on its own goroutine. While the window is full, the ready batch keeps filling up to its limits (H2: Go client keeps one batch in flight per shard #1432's cheap fix comes for free: at linger 0, the calls already queued join the batch). A Barrier fires only once the batches sent before it have completed, so the split hand-off from feat: keep the write order across shard splits #1360 keeps its ordering. Without a window the batcher is unchanged.

2. Async write stream (oxia/internal). ExecuteWriteAsync sends on a per-shard stream and matches responses in send order. When the stream ends, the writes in flight are settled together, in order:

   • Resent, in order on a new stream and ahead of any newer write, only when the end proves the server processed none of them:

     • the stream was rejected at setup: ErrNodeIsNotLeader with a leader hint, which the server only sends before reading anything;
     • or the server marked the end unprocessed (commit 5).

     Repeated rejections back off. An error a retry can't fix (e.g. the shard is gone after a split) fails them, and the batches reroute them.

   • Failed, in order, and never resent, in every other case: broken connection, EOF, or a leader that steps down mid-stream. Those writes may already be appended.

   • Partial resend: a resend whose stream breaks after taking some writes reads the stream status. Unless it proves them unprocessed, they fail instead of being sent a third time.

   • Deadline: a write with no response by its deadline aborts the stream (only while that write is still in flight), so the writes behind it fail at once.

   Only a request that was never sent is retried before sending. The synchronous ExecuteWrite path is untouched.

3. Client wiring and options. Write batches are AsyncBatch. WithMaxWriteBatchesInFlight (default 4, as in the Java client; 0 keeps today's behavior) and WithMaxBatchSize (the byte cap already had a validation error but no option). Reads are unchanged; a read window can follow separately.

4. Resend only what is provably unprocessed (review fix): the rules above, the partial-resend check, and the stale-abort race. A generation check and the abort it guards are now one critical section.

5. Server: end a stream as unprocessed. The leader controller tags a write it rejects before appending it (lead.IsNotAppended). The write stream then:

   • stops reading;
   • answers every write it appended before the rejected one;
   • ends with the rejection plus ErrorMetadata["unprocessed"].

   If one of those earlier writes fails instead, the stream ends with that error, unmarked. Older clients ignore the key, and older servers never send it, so they get the safe fail path.

6. Server: graceful close drains, and setup rejections are marked.

   • A closing leader controller stops accepting writes (rejecting new ones before append) and waits up to 5 s for the ones in flight to complete, instead of abandoning them. A fence (NewTerm) still stops it at once.
   • A stream rejected at setup is marked unprocessed whatever the error.
   • A stream whose leader closed ends marked once every appended write is answered, unless one of them failed.

   Together these make a rolling restart transparent: TestCoordinator_LeaderFailoverWithOpenClient keeps passing, with no write sent twice.

Relation to #1399

This follows #1399's rule: a write that may have reached a leader is never sent again. A resend only happens when the server proves it processed none of the writes. So a leader move, a split freeze or a graceful shutdown stays transparent when the server marks the stream, and a crash fails the writes in flight, without a duplicate in either case. Leader following on assignment changes and cancelling the stream setup on the request deadline are left to #1399; they apply to the async stream the same way.

Benchmarks

darwin/arm64, 10 cores, in-process 3-server cluster, one shard, RF=3, measured at the head of this PR (3 to 4 runs each).

SyncClient with 64 goroutines, 100-byte values. Window 0 is today's path, same build:

BenchmarkE2E from #1443 (async client, 5,000 operations outstanding, fsync off), main vs this branch:

Put throughput is about +60%. The higher latency is queueing at saturation, not slower requests: on main the client blocks the caller once a shard's queue is full, so only about 750 operations are actually outstanding (131k/s × 5.7 ms), while with the window nearly all 5,000 are (213k/s × 21 ms).

Tests

• Batcher: the window overlaps batches; while full, a batch keeps filling; a barrier waits for the batches in flight; close fails the batch being formed and lets the ones in flight complete.
• Stream (real gRPC servers):
  • writes go out before their responses;
  • a broken stream fails the writes in flight in order and never resends them;
  • a leader stepping down mid-stream (no hint) fails them, and never resends them;
  • a setup rejection resends them in order to the hinted leader, ahead of a newer write;
  • an unprocessed end resends them in order;
  • a partial resend that breaks with an unknown status fails them (no third send), and with a setup rejection retries them;
  • repeated rejections back off;
  • rejected writes of a gone shard fail with ErrShardNotFound;
  • a deadline aborts the stream.
• Server:
  • a frozen or fenced leader tags its rejection as not appended;
  • a write stream ends unprocessed only after answering the writes appended before the rejected one, and reads nothing after it;
  • a failed earlier write leaves the end unmarked.
• Client: two batches in flight; a split with writes in flight keeps per-key order (it fails if the barrier only waits for sends).
• The whole oxia module and the tests module pass with -race with the window on by default, including the split, leader-hint and failover e2e tests.