feat: s2-stream-config header for auto-created streams

[infiniteregrets]

Summary

When a basin has create_stream_on_append enabled, an append can carry an s2-stream-config header whose value is a compact JSON StreamConfig. If that request is the one that creates the stream, the config is layered over the basin's default_stream_config; unset fields inherit the defaults. It is ignored once the stream exists, so clients can attach it to every append without tracking whether the stream has been created.

This gives per-stream config (e.g. retention, delete-on-empty) on auto-created streams without a control-plane round trip.

Spec half: s2-streamstore/s2-specs#21 (this PR bumps the api/specs submodule to that branch's commit; re-bump to the merge commit once it lands).

API

One header, same for JSON, proto and S2S (an append session is a single request, so the header covers the whole session):

curl -X POST "https://$BASIN.b.s2.dev/v1/streams/tenant-42%2Fevents/records" \
  -H "Authorization: Bearer $S2_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -H 's2-stream-config: {"retention_policy":{"age":3600},"delete_on_empty":{"min_age_secs":300}}' \
  -d '{"records": [{"body": "hello"}]}'

The value is validated exactly like a CreateStream config; an invalid value is rejected with 400 bad_header before any lookup and no stream is created:

{"code":"bad_header","message":"Invalid header `s2-stream-config`: age must be greater than 0 seconds"}

SDK: the option lives on the stream handle, like the encryption key, and applies to unary appends, append sessions and producers:

let stream = basin
    .stream(name)
    .with_stream_config(
        StreamConfig::new()
            .with_retention_policy(RetentionPolicy::Age(3600))
            .with_delete_on_empty(DeleteOnEmptyConfig::new().with_min_age(Duration::from_secs(300))),
    );
stream.append(input).await?;                       // or
stream.producer(ProducerConfig::default());        // or stream.append_session(..)

CLI:

echo hello | s2 append s2://my-basin/tenant-42/events --format text \
  --retention-policy 1h --delete-on-empty-min-age 5m

Why a header

• No proto change: the same header carries the config for unary appends and S2S sessions, so AppendInput (which flows into storage) stays untouched.
• Known before the server responds. With a body/frame field, S2S sessions needed the server to wait for the first frame before creating the stream, while clients wait for response headers before sending it; the header removes that ordering problem entirely.
• Reusable for read paths (create_stream_on_read) later, since GET has no body.

Changes

• api
  • v1::config::STREAM_CONFIG_HEADER (s2-stream-config) and StreamConfigHeader, a ParseableHeader that deserializes the JSON StreamConfig and reuses TryFrom<StreamConfig> for OptionalStreamConfig so validation lives in one place. to_header_value for clients.
  • data::S2StreamConfigHeader documents the header in OpenAPI (string schema, with an example value; utoipa cannot express content on a parameter).
  • AppendRequest::Unary / S2s gain stream_config: OptionalStreamConfig, parsed once in the extractor.
• lite
  • stream_handle_with_auto_create takes an AutoCreateOn (Append / Read) and the OptionalStreamConfig to layer over the basin defaults when creating.
  • Backend::open_for_append(.., stream_config) serves both unary appends and sessions; the stream is created (or the request fails) before the response, as before this feature.
• sdk: S2Stream::with_stream_config, mirroring with_encryption_key. Internally, AppendHeaders { encryption, stream_config } is threaded through sessions/producers and set on every (re)connect.
• cli: s2 append accepts the same stream config flags as create-stream (--retention-policy, --storage-class, --timestamping-*, --delete-on-empty-min-age), listed under their own help heading; set on the stream handle.

Compatibility

• Old clients never send the header; old servers ignore unknown headers.
• s2-api public API change: AppendRequest variants gain a field.

Testing

• s2-api unit: header parse/validate (valid, {}, invalid JSON, age: 0) and to_header_value roundtrip.
• Backend-level: applies + merges with basin defaults; existing stream ignores config.
• HTTP-level: JSON unary with header; invalid header -> 400 bad_header with no stream created (both invalid config and non-JSON); S2S session with header.
• SDK integration against s2 lite: unary + producer create with config, existing stream unchanged.
• CLI integration against s2 lite: 47/47 pass.
• clippy -D warnings clean; workspace unit suites pass.

Made with Cursor

[greptile-apps]

Greptile Summary

The PR adds an s2-stream-config request header for configuring streams created automatically by append or read operations.

• Parses and validates partial stream configuration at the API boundary.
• Layers supplied fields over basin defaults during automatic stream creation and checks supplied fields against existing stream configuration.
• Propagates configuration through Lite, SDK sessions and producers, and CLI append/read commands.
• Adds backend, HTTP, SDK, and CLI coverage for creation and mismatch behavior.

Confidence Score: 5/5

The PR appears safe to merge.

No blocking failure remains.

Important Files Changed

Filename	Overview
api/src/v1/config.rs	Defines parsing, validation, and client-side serialization for the JSON stream-configuration header.
api/src/v1/stream/extract.rs	Extracts the optional stream configuration consistently across append and read request variants.
common/src/config.rs	Adds partial-configuration comparison and detailed mismatch reporting.
lite/src/backend/core.rs	Applies requested configuration during automatic creation and validates configured fields when resolving existing streams.
lite/src/handlers/v1/records.rs	Threads extracted stream configuration into Lite append and read backend operations.
sdk/src/api.rs	Serializes configured SDK operations into the new request header.
sdk/src/session/append.rs	Retains stream configuration across append-session connections and reconnects.
sdk/src/session/read.rs	Retains stream configuration across read-session connections and reconnects.
sdk/src/producer.rs	Propagates producer-level stream configuration into its append session.
cli/src/ops.rs	Converts append and read configuration flags into SDK operation options.

Sequence Diagram

sequenceDiagram
    participant Client
    participant API
    participant Lite as S2 Lite
    participant Metadata

    Client->>API: Append/read + s2-stream-config
    API->>API: Parse and validate JSON header
    API->>Lite: Request + OptionalStreamConfig
    Lite->>Metadata: Look up stream
    alt Stream is missing and auto-create is enabled
        Lite->>Metadata: Create using basin defaults layered with supplied fields
    else Stream exists
        Lite->>Metadata: Compare supplied fields with existing config
    end
    Lite-->>Client: Append/read response

Loading

_{Reviews (3): Last reviewed commit: "cli tests: bound read, assert mismatch r..." | Re-trigger Greptile}

[infiniteregrets]

@greptile-apps review