[moetemp]

What changed?

This PR adds the extension call WriteRecords, the awaited publish of an Activity or a client.
WriteRecordsRequest { writer_id, records } appends the Records at once for an open writer and answers WriteRecordsResponse { first_offset, last_offset } when they are in the Store. It first sends what Publish buffered, under the same flush lock, so the order of one writer holds. It keeps the writer's Producer epoch and sequence order. A Store failure resends the same batch with the same sequence, so a batch whose answer was lost goes in once. Records past the Store's batch size or byte limit go as more batches, in order, and the offsets span all of them. An empty request answers zeros with no write.
A Record over the Store's Record limit, or a closed writer, is refused before the Store, and the writer stays usable after the size refusal. A failure after the Store retries fails the writer, as a failed flush does, since the batch may be in the Store.
I chose to keep Publish and the interval writer for compatibility. An SDK does not need them for an awaited publish: it opens a writer per Producer epoch (which still sends the empty batch of an Activity retry), calls WriteRecords per publish, and CloseWriter at the end. BufferedPublisher in lang can sit on WriteRecords with its own timer.

Why?

Decision S71 (and S47). An Activity or client await publish must have published when it returns, and it returns the offset. A hidden timer would make the return mean only "buffered".

How did you test it?

cargo fmt --all -- --check, cargo clippy --workspace --tests -- -D warnings (also with --no-default-features --features tls-aws-lc), and STREAMS_REQUIRE=redis cargo test --workspace on standalone Redis, a 3-primary Redis cluster and the memory Store.
New writer tests check that a write is readable when it answers and that its offsets match, that a write goes after buffered Records in one order and splits at the batch size, that an Activity retry's write withdraws the earlier attempt and fences it, and the refusals of an oversize Record and a closed writer. A test on a memory Store that loses the first append answer checks that the retry goes in once with the first offsets.

[moetemp]

Review fix in 4698c5f: a WriteRecords that waited behind a CloseWriter of the same writer is refused with stream_unknown_writer, with a test.

[moetemp]

Closing as stale. This layer is folded into the v5 tip, #41, and the upstream review uses a fresh series instead. Plan: https://www.notion.so/4048fc567738814990b7c796ac28fe3f