[merlimat]

Fixes #397.

Problem

ManagedWriteStream.send() takes the per-shard stream lock and then calls onNext(), where the generated gRPC marshaller serializes the WriteRequest. handleResponse() takes the same lock for every response, on the gRPC event loop (the channels use directExecutor()). So a response can wait behind another batch's serialization, and so can everything else on that event loop. The batcher thread can also park on the lock while the event loop holds it: when a shard's dispatch window is full, DispatchWindow.release() sends the next batch from the event loop.

#433 already moved building the request out of the lock: WriteBatch calls toProto() before send(). Serializing it is what's left.

Example

// ManagedWriteStream.send()
lock.lock();
...
inflightWrites.addLast(inflightWrite);
subStreamObserver.send(inflightWrite.request); // onNext(): the marshaller serializes the request here

// ManagedWriteStream.handleResponse(), on the event loop
lock.lock(); // waits for that serialization

The work done under the lock in onNext() for a full batch (128 KB by default), from a throwaway JMH benchmark (Apple M1 Max, JDK 26):

Batch	Serialize + copy into a transport buffer (before)	Copy only (after)
128 KB of 100-byte values (~1,050 puts)	18.7 µs	3.0 µs
128 KB of 1 KB values (~125 puts)	7.2 µs	2.7 µs

Change

send() serializes the request before taking the lock, with the generated marshaller, and the write stream call takes the serialized stream through a pass-through marshaller:

// GrpcRpcProvider
private static final MethodDescriptor<InputStream, WriteResponse> WRITE_STREAM_METHOD =
        OxiaClientGrpc.getWriteStreamMethod()
                .toBuilder(/* stream() and parse() return their argument */, responseMarshaller)
                .build();
...
ClientCalls.asyncBidiStreamingCall(
        stub.getChannel().newCall(WRITE_STREAM_METHOD, stub.getCallOptions()), barrierObserver);

// ManagedWriteStream.send()
final InputStream serializedRequest = ManagedSubWriteStream.serialize(request); // streamRequest()
final CompletableFuture<WriteResponse> future = new CompletableFuture<>();
final InflightWrite inflightWrite = new InflightWrite(request, future, System.nanoTime());
lock.lock();
...
subStreamObserver.send(serializedRequest);

Under the lock, send() now only appends the in-flight entry and calls onNext(), which copies the bytes into gRPC's buffer and queues the frame.

• Allocations don't change: the generated marshaller still serializes into its per-thread spare array, and gets it back when gRPC closes the stream after copying it. Serializing into a new byte[] instead would add about 140 KB of allocation per 128 KB batch.
• A serialized stream can be read once. A replay after a reconnect serializes the in-flight requests again from the WriteRequest each entry keeps, under the lock, as before. gRPC never streams a message twice here, as the channels are built with disableRetry().
• A request is still never serialized by two threads at once: send() serializes it before it's in the in-flight queue, and replays only serialize queued requests, under the lock.
• RpcProvider.writeStream() returns a StreamObserver<InputStream>.

Testing

• New ManagedWriteStreamTest.handlesResponseWhileAnotherWriteIsSerialized: a put whose value blocks in the middle of its serialization is sent from another thread, and the server then answers an earlier write. Before the change, that response doesn't complete the earlier write within 5 s (TimeoutException: the event loop waits on the lock); after it, it does, and the blocked write then reaches the server with its value.
• The tests that mock RpcProvider.writeStream() take InputStream requests, and parse them where they check the keys.
• ./gradlew :client:test: all 406 tests pass locally, including the Docker integration tests (OxiaClientIT, ClientReconnectIT, ...). The new test passes 10 out of 10 repeated runs.