From 2752fe7711d043ce17eed4285de4e39bd25ec50d Mon Sep 17 00:00:00 2001 From: Jeff Repanich Date: Wed, 12 Aug 2026 13:35:34 -0400 Subject: [PATCH] docs(stream): define read conformance contract --- .../cross-language-conformance-suite.yaml | 13 +++++ docs/clients/spec/notice-stream.md | 30 +++++++++++ .../clients/spec/stream-read-conformance.json | 52 +++++++++++++++++++ 3 files changed, 95 insertions(+) create mode 100644 docs/clients/spec/stream-read-conformance.json diff --git a/docs/clients/cross-language-conformance-suite.yaml b/docs/clients/cross-language-conformance-suite.yaml index f8ee5fb2..c5c01365 100644 --- a/docs/clients/cross-language-conformance-suite.yaml +++ b/docs/clients/cross-language-conformance-suite.yaml @@ -236,6 +236,19 @@ scenarios: - "cursor advances monotonically through filtered offsets" - id: "CS-017" + title: "global filtered multi-page replay" + priority: "P1" + setup: + transport: ["websocket", "tcp"] + auth_mode: ["anonymous", "valid_jwt"] + expected: + - "stream://** filtered replay spans multiple pages across at least two realms" + - "filtered-only pages remain observable" + - "records retain concrete routes and global offsets" + - "next offset advances on the global cursor axis" + - "replay yields the caught-up page and then completes" + + - id: "CS-018" title: "bounded concurrency under burst load" priority: "P1" setup: diff --git a/docs/clients/spec/notice-stream.md b/docs/clients/spec/notice-stream.md index 75ba8177..81f9ec57 100644 --- a/docs/clients/spec/notice-stream.md +++ b/docs/clients/spec/notice-stream.md @@ -606,6 +606,36 @@ READ and SUBSCRIBE accept the same finite selector matrix: **Cursor:** `ReadCursor` is response metadata that advances with every committed offset the broker considers during replay, including filtered markers. It is not a durable broker-side resume token. +#### Client replay and follow contract + +The versioned machine-readable client contract is +[`stream-read-conformance.json`](./stream-read-conformance.json). Clients MUST +validate selector classification and replay progress against that fixture. + +A high-level read API requires an explicit starting offset and mode. `replay` +issues READ until the first `has_more=false` response, yields that terminal +batch, and completes. `follow` establishes SUBSCRIBE before its initial READ, +drains through `has_more=false`, then waits for a commit or reconnect wake. +Subscribing first prevents a lost commit between catch-up and subscription. + +Every successful page is observable as one batch, including filtered-only +pages and empty pages whose cursor advances. A batch contains all tagged items, +the event-only record view, the requested offset, the next durable client +offset derived from the selector's cursor axis, and `caughtUp`. A client MUST +NOT fabricate an offset when the applicable cursor axis is absent; it retains +the requested offset. + +Continuation fingerprint and captured watermark fields are protocol-private. +Clients carry them automatically between global READ pages and do not expose +them as application resume tokens. Two consecutive `has_more=true` responses +with the same applicable cursor position are a non-retryable stalled read even +if private continuation values rotate. This permits one empty continuation page +while bounding a malformed broker loop. + +Cancellation, iterator return, reconnect failure, decode failure, and other +terminal failures release a live subscription. Explicit unsubscribe operations +surface broker errors; automatic iterator disposal is best effort. + #### LAST Request ``` diff --git a/docs/clients/spec/stream-read-conformance.json b/docs/clients/spec/stream-read-conformance.json new file mode 100644 index 00000000..ef24a550 --- /dev/null +++ b/docs/clients/spec/stream-read-conformance.json @@ -0,0 +1,52 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "fixture": "fitz.stream-read-conformance", + "version": 1, + "offsetSemantics": { + "resource": "lastResourceOffset + 1", + "area": "lastAreaOffset + 1", + "realm": "lastRealmOffset + 1", + "global": "lastGlobalOffset + 1", + "missingApplicableAxis": "retain fromOffset", + "filteredMarkersAdvance": true, + "emptyAdvancingPagesVisible": true + }, + "selectors": [ + { "selector": "stream://realm/area/resource", "axis": "resource", "canonical": "stream://realm/area/resource" }, + { "selector": "stream://realm/area/*", "axis": "area", "canonical": "stream://realm/area/*" }, + { "selector": "stream://realm/*/resource", "axis": "realm", "canonical": "stream://realm/*/resource" }, + { "selector": "stream://realm/*/*", "axis": "realm", "canonical": "stream://realm/*/*" }, + { "selector": "stream://realm/**", "axis": "realm", "canonical": "stream://realm/*/*" }, + { "selector": "stream://*/area/resource", "axis": "global", "canonical": "stream://*/area/resource" }, + { "selector": "stream://*/area/*", "axis": "global", "canonical": "stream://*/area/*" }, + { "selector": "stream://*/*/resource", "axis": "global", "canonical": "stream://*/*/resource" }, + { "selector": "stream://*/*/*", "axis": "global", "canonical": "stream://*/*/*" }, + { "selector": "stream://**", "axis": "global", "canonical": "stream://*/*/*" } + ], + "invalidSelectors": [ + "stream://realm/area", + "stream://realm/area/resource/extra", + "stream://realm/**/resource", + "stream://realm/a*ea/resource", + "stream:///area/resource", + "notice://realm/area/resource" + ], + "deliveryItems": [ + { "tag": 0, "name": "event", "advances": true, "isRecord": true }, + { "tag": 1, "name": "filteredOffset", "advances": true, "isRecord": false }, + { "tag": 2, "name": "filteredRange", "advances": true, "isRecord": false } + ], + "progressInvariants": { + "yieldEverySuccessfulPage": true, + "caughtUp": "hasMore == false", + "replayCompletesAfterFirstCaughtUpBatch": true, + "followSubscribesBeforeInitialRead": true, + "privateContinuationFields": ["cursorFingerprint", "capturedWatermark"], + "stallBound": { + "consecutiveHasMorePagesWithoutLogicalProgress": 2, + "continuationRotationDoesNotCountAsProgress": true, + "retryable": false + }, + "releaseSubscriptionOn": ["abort", "return", "reconnectFailure", "decodeFailure", "readFailure"] + } +}