feat(sdk): add a non-terminal Producer flush/barrier

[shikhar]

Motivation

Producer currently emits a partial batch when a size/count threshold or linger deadline is reached, and close() forces the final partial batch before terminating. There is no way to force the current partial batch at an application-defined durability boundary while keeping the Producer open.

This matters for applications that submit records speculatively and then encounter a causal boundary that must wait for the complete prefix to become durable. A fault-tolerant WASM runtime using S2 as its journal is one concrete example:

1. Submit granular observation records without blocking execution.
2. Reach an externally visible output boundary.
3. Immediately seal the partial batch and wait for the submitted prefix to become durable.
4. Continue using the same Producer and append session.

The available choices all have an avoidable cost:

• use the default 5 ms linger and add up to 5 ms at each latency-sensitive boundary;
• use zero linger and risk producing many small batches;
• call close() and recreate Producer/session at every boundary; or
• reimplement Producer's batching over AppendSession to gain an explicit flush operation.

The last option is what the downstream runtime currently does. Producer otherwise already has the needed correctness primitives: initial match_seq_num with automatic progression, a fixed fencing token, ordered append-session submission, and per-record tickets with exact sequence positions.

Requested semantics

Please add a non-terminal flush or barrier operation. The exact API shape is open, but the required behavior is:

• force any partial batch containing records ordered before the barrier into the underlying append session immediately, without waiting for linger;
• allow the caller to wait until every record ordered before the barrier is durably acknowledged;
• do not wait for records ordered after the barrier;
• leave Producer open and usable for subsequent submissions;
• preserve existing per-record ticket results and ordering;
• complete immediately when there is no preceding work; and
• propagate a terminal submission/acknowledgement error consistently to the barrier and affected record tickets.

For example, either a directly durable flush:

let a = producer.submit(record_a).await?;
let b = producer.submit(record_b).await?;

producer.flush().await?; // a and b are durable; Producer remains open

or a split submission/durability ticket:

let barrier = producer.flush().await?; // partial batch is submitted now
barrier.await?;                        // preceding prefix is durable

Concurrent submission semantics should be explicit. One reasonable definition is that the barrier covers commands ordered before it in Producer's command stream; records ordered afterward belong to a later batch/barrier.

Acceptance coverage

• A Producer configured with a very long linger flushes a one-record partial batch promptly without closing.
• Multiple records before the barrier retain order and receive their exact indexed acknowledgements.
• Records submitted after the barrier are not included in its durability wait.
• New records can be submitted and flushed after the first barrier.
• Empty barriers complete without creating an append batch.
• Condition failures, ambiguous session termination, and Producer shutdown cannot report the barrier successful while a covered record ticket fails.

This is the reusable, non-terminal counterpart to the immediate final flush added for Producer::close() in #233. It requires no S2 protocol change; it is a Producer batching/lifecycle primitive.

[shikhar]

Cross-SDK follow-up: I reviewed the current Producer/batcher/session implementations in the four supported SDKs. The Rust feature in this issue can stand alone, but there is a useful cross-cutting workstream someone can pick up so the resulting API has one behavioral contract everywhere.

Current shape

SDK	Producer/session ownership	Non-terminal partial flush	close() behavior
Rust	stream.producer(config) creates and owns its session	None	Flushes, drains, closes session
Python	stream.producer(...) creates and owns its session	Private emission helper only	Flushes, drains, closes session
TypeScript	Session is supplied externally	Batcher flush is private/close-only	Producer closes the supplied session
Go	Session is supplied externally	Batcher.Flush() only emits a batch; it is not a durability barrier	Drains Producer work but leaves session open

So none currently exposes a Producer-level, durable, non-terminal barrier. Recreating a Producer on the same session is only naturally expressible in Go today, and is a poor portable substitute because it moves batching state, match_seq_num progression, failure state, and resource ownership across objects.

Suggested common Producer contract

Keep syntax language-idiomatic, but standardize these semantics:

submit(record) -> per-record durability ticket
flush()        -> durable prefix barrier; Producer remains open
close()        -> flush + drain + release resources owned by Producer

flush() should:

• be linearized in the same ordered command stream as submit;
• immediately seal any partial batch before it;
• wait until every record ordered before it is durably acknowledged;
• not wait for records ordered after it;
• complete immediately without an append when the covered prefix is empty;
• preserve all per-record ticket results and exact sequence positions;
• leave the Producer usable afterward; and
• fail with the sticky terminal cause if any covered submission/ack fails.

A submit that has completed before flush() begins must be covered. Submits concurrent with the barrier may land on either side, but the cut must be unambiguous. The initial API can return only success/error; exact positions remain available through record tickets.

Suggested language shapes:

producer.flush().await?;

await producer.flush();

await producer.flush()

err := producer.Flush(ctx)

Ownership recommendation

The normal high-level API should be stream.producer(options), with Producer owning its batcher and logical append session. If constructing a Producer over an existing session remains supported, make that mode explicitly non-owning (for example fromSession/attachProducer) rather than using an ambiguous ownership boolean. Closing an attached Producer should durably drain Producer work but leave the caller-owned session open.

That can be handled separately from this Rust issue because changing TypeScript's current close behavior may need a deprecation/major-version path.

Other cross-SDK drift found

• linger = 0 currently means immediate scheduling in Rust/TypeScript, disables time-based flushing in Python, and selects the 5 ms default in Go.
• TypeScript uses a 3 MiB default inflight-byte limit while shared docs and the other SDKs use 5 MiB.
• Invalid batch limits are rejected in some SDKs but often clamped/defaulted in Go.
• Go Producer.Close() currently returns nil even when record tickets observed errors.
• Concurrent/idempotent close behavior is not equally strong across implementations.

Suggested normalization: default linger 5 ms; zero means no intentional delay; use an explicit None/disabled value for no timer; default inflight bytes 5 MiB; reject invalid limits; and make terminal Producer errors sticky across tickets, flush, and close.

Possible breakdown

1. Land the Rust implementation and tests described in this issue.
2. Extract these semantics into a short language-neutral Producer contract/conformance checklist.
3. Port the barrier tests and implementation to Python, TypeScript, and Go.
4. Address ownership constructors and configuration defaults as a separate compatibility-focused pass.

Cross-SDK conformance cases should include: long-linger one-record flush, multiple records retaining exact acks, post-barrier records excluded from the wait, repeated flush/use, empty flush, a concurrent submit cut, condition failure, ambiguous session termination, and close racing with a barrier.