mirror of
https://github.com/s2-streamstore/s2.git
synced 2026-09-29 18:16:03 +00:00
a4f247aeed## 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): ```sh 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: ```json {"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: ```rust 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: ```sh 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](https://cursor.com) --------- Co-authored-by: Cursor <cursoragent@cursor.com> |
History
|
|
|---|---|---|
| .. | ||
| cli.rs | ||
| integration.rs | ||