refactor(streaming)!: one streaming core, correct cursors, faster consumers #168

Merged

Grandiras merged 1 commit from issue-126-streaming into main 2026-09-26 01:24:45 +00:00

Conversation 5 Commits 1 Files changed 71 +6190 -2720

Grandiras commented 2026-09-26 01:20:43 +00:00

Owner

Copy link

Closes #126.

One streaming core for the firehose, label, chat moderation and Jetstream streams: a single WebSocket helper, one reconnect policy, one cursor tracker and one cursor store; correct cursors; typed error frames; and less CPU per event. Also covers the scope added in the issue comments (typed CollectionFilter, IRecordEvent, one operation enum, the docs/firehose.md cursor example, a drop hook, the $bytes padding bug, subscribeModEvents) and the entries #123 deferred to this issue.

Consolidation

• One WebSocket helper (StreamSocket, internal): connect (subprotocol, Authorization), receive with reassembly into one reused buffer (capped at 64 MiB), graceful close on dispose. Every stream client and consumer goes through it; each subscription owns its socket, which fixes the leak when FirehoseClient.SubscribeAsync was called twice. A StreamConnector delegate is the test seam.
• One reconnect loop for the atproto-framed streams (EventStreamLoop + EventStreamHandler<T>, internal): header parsing, error frames, reconnect and cancellation live once; the firehose, label and chat moderation handlers only turn bodies into messages and track their position.
• StreamReconnectPolicy (InitialDelay 5 s, MaxDelay 30 s, MaxAttempts 10, null = unlimited), exponential with up to 10 % jitter, on every consumer through the new StreamConsumerOptions base. The attempt count resets when a connection delivers a frame. Exhausted attempts throw EventStreamException (inner = last failure). A non-retryable failure (FutureCursor, a 4xx other than 408/429) throws at once.
• CursorTracker (internal): monotonic; filtered and failed-verification events advance it; saves are single-flight on a background task every CursorPersistInterval events; the final save runs in finally and never throws (a failed save is logged). Delivered events are recorded when the caller asks for the next one (at-least-once).
• IStreamCursorStore (ValueTask) replaces IFirehoseCursorStore; InMemoryStreamCursorStore.
• FirehoseConsumer folded into TypedFirehoseConsumer: each frame is parsed once. Frames that are skipped (filtered, unknown type, unparseable) move the cursor with a seq-only CBOR scan.
• TypedFirehoseConsumer: with a CollectionFilter (now IReadOnlySet<Nsid>), ops[].path and seq are read straight from the CBOR before deserializing; a verifier verifies exactly when set (the VerifySignatures bool is gone) and its signature path also checks every block's CID, so the CAR is parsed and hashed once; #identity still invalidates the verifier's DID cache.
• Error frames (op = -1) surface as EventStreamException (Error, IsRetryable) and are reported to OnStreamError; consumers reconnect after ConsumerTooSlow and throw FutureCursor. FirehoseEventParser.Parse throws for an error frame.
• subscribeLabels finished: LabelStreamMessage → LabelsEvent / LabelInfoEvent, typed FirehoseClient.SubscribeLabelsAsync, and a reconnecting LabelStreamConsumer (a separate options type, so the label-signing issue #135 can add a verifier to it).
• Jetstream: validation in one JetstreamConsumerOptions.Validate() and kind names in one JetstreamKinds helper; JetstreamDictionaryClient folded into JetstreamArchiveClient.GetZstdDictionaryAsync (no API key sent to the public endpoint); query strings through XrpcParams (new internal ToQueryString()); one JetstreamException (subclass of EventStreamException); shared event builders (JetstreamEvents) for the live parser and the segment reader, since a segment row's payload has the same fields the live frame nests; JetstreamEventParser keeps ParseFrame(ReadOnlyMemory<byte>, protocol); dead members removed (JetstreamPlanStats/Stats, IsConnected, PinnedTipSeq, the no-op Dispose, the hand-rolled ReadExactlyAsync; DecodeBlock is internal).
• FirehoseClient and JetstreamClient are IAsyncDisposable; disposing ends their running subscriptions.

Protocol

• Jetstream v2 timestamp cursors (jetstream #350/#352): JetstreamCursor.FromTimestamp / IsTimestamp / TimestampThreshold (10^15); FromTimestamp refuses times that would read as a sequence number. A v2 timestamp seek sets no dedup floor (never deduplicated by time_us); the consumer continues by sequence number from the first event, so a stored v1 cursor migrates to v2 unchanged (documented). A v2 sequence start cursor now also dedupes the server's inclusive replay of it. JetstreamReplayConsumer rejects a timestamp cursor, since the archive is sequence-addressed.
• Archive HEAD probes (#347): ProbeSegmentAsync / ProbeBlockAsync → JetstreamArchiveProbe(ContentLength, ETag). Health: GetHealthAsync (/xrpc/_health → {version}), plus JetstreamHealthCheck and AddHealthChecks().AddJetstream(url) in ATProtoNet.Server/Aspire (next to the PDS check).
• Firehose models (deferred from #123): HandleEvent / TombstoneEvent removed; Seq/Time move to a new FirehoseEvent base, so InfoEvent no longer declares fields #info lacks; TooBig, Rebase, Blobs are [Obsolete] (still bound). The #126 drift allow-list entries are removed; new entries cover the new union bases.
• chat.bsky.moderation.subscribeModEvents: ChatModerationEventConsumer with a per-connection bearer token provider (GetAccessTokenAsync, so each reconnect can mint a fresh service-auth token), resume after the last Rev, and ChatModerationEvent models for all 11 variants plus UnknownChatModerationEvent (open [AtProtoUnion]).
• $bytes padding: DagCborDecoder (the Wrapped form of the shared transcoder) now writes unpadded base64, pinned against RFC 4648 vectors and the atproto-interop data-model fixtures' reference JSON.

Also from the issue comments

• IRecordEvent (Did, Collection, Rkey, Operation, Rev, Cid, Record, Uri, GetRecord<T>), implemented by JetstreamCommitEvent and by FirehoseRecordEvent from CommitEvent.GetRecordEvents() (records decoded from the commit's blocks).
• JetstreamOperation removed in favour of RepoOpAction.
• OnEventDropped hook with DroppedStreamEvent(Reason, Cursor, Detail) on every consumer (and JetstreamClient).
• docs/firehose.md cursor example fixed (it read message.Seq off a raw frame); both streaming docs, chat.md, error-handling.md and api-reference.md updated; both samples updated.

Performance

• Firehose: collection-filtered commits are skipped after a path peek; skipped frames advance the cursor via a seq-only scan; a verified commit's CAR is parsed once.
• Jetstream archive: rows are filtered on their DID/kind/collection columns before ToEvent(); matching rows alias the decompressed block instead of copying payloads; DID and collection (and rev) strings are interned per block; blocks are decoded in the download stage (getBlock units in their download task, whole segments DownloadParallelism blocks ahead), still delivered in order.
• Jetstream live: frames are parsed straight from the socket's receive buffer (no per-frame copy); only a compressed frame is copied, by the decompressor.
• Cursor persistence is off the read loop.

Breaking changes (migration)

• ReconnectDelay / MaxReconnectAttempts → Reconnect = new StreamReconnectPolicy { InitialDelay = …, MaxAttempts = null }.
• Exhausted reconnects throw EventStreamException; cancellation ends every consumer normally (the replay consumer threw OperationCanceledException mid-backfill before).
• IFirehoseCursorStore / InMemoryFirehoseCursorStore → IStreamCursorStore / InMemoryStreamCursorStore, returning ValueTask.
• FirehoseConsumer, FirehoseFrame, FirehoseEventParser.Parse(FirehoseFrame) / TryParse, FirehoseClient.DisconnectAsync, JetstreamClient.DisconnectAsync removed; FirehoseClient.SubscribeAsync yields FirehoseMessage; the clients are IAsyncDisposable (await using); TypedFirehoseConsumer / JetstreamConsumer are no longer IDisposable.
• VerifySignatures removed (set Verifier); CollectionFilter is IReadOnlySet<Nsid>.
• HandleEvent / TombstoneEvent removed; Seq/Time on FirehoseEvent (message is FirehoseEvent e ? e.Seq); LabelsEvent derives from LabelStreamMessage.
• JetstreamOperation → RepoOpAction.
• JetstreamConnectException / JetstreamArchiveException → JetstreamException; JetstreamStreamError → EventStreamError; JetstreamClient throws on an error frame.
• JetstreamDictionaryClient → JetstreamArchiveClient.GetZstdDictionaryAsync.
• JetstreamEventParser.Parse(...) and the span/string ParseFrame overloads removed.
• JetstreamPlanStats, JetstreamSnapshotPlan.Stats, JetstreamConsumer.IsConnected, JetstreamReplayConsumer.PinnedTipSeq removed; JetstreamSegmentReader.DecodeBlock internal.

Each has a Breaking changes bullet with its migration note in the CHANGELOG.

Verification

• dotnet build (whole solution): 0 warnings, 0 errors.
• Unit tests: 3,987 passed, 0 failed, 5 skipped (the pre-existing gated skips). Streaming + DAG-CBOR subset re-run 3× to check the background-save and socket tests for flakiness.
• New tests: TypedFirehoseConsumerTests (filter advances the stored cursor, path peek before parse, break/cancellation keep the cursor, resume/explicit cursor, reconnect dedupe, exhausted attempts throw, FutureCursor throws / ConsumerTooSlow reconnects, drop reporting, verifier/VerifyCids/no verification); EventStreamClientTests (single-connection client, per-subscription sockets, dispose, labels, LabelStreamConsumer, chat moderation typed + unknown events, per-connection tokens, rev resume, backoff curve, retryability); StreamSocketTests against a real Kestrel WebSocket on loopback (fragment reassembly past the buffer, empty messages, text/binary, subprotocol and Authorization, refused upgrade status, close on dispose); CursorTrackerTests (monotonic, final flush, background interval save, slow store does not block and saves the newest, failing store does not throw); JetstreamClientTests (info out of band, receive-buffer parsing, error frame, refused upgrade, drop hook, decompressor); Jetstream timestamp-cursor, migration and error-frame consumer tests; replay cutover (inclusive seam dedupe, refused cutover re-plans from the last delivered event), timestamp rejection, cancellation, column filter; RecordEventTests; archive dictionary/health/HEAD probe tests; JetstreamHealthCheck.
• Pinned vectors: RFC 4648 base64 vectors for $bytes (Zg, Zm8, Zm9v, +/8), and the three atproto-interop data-model-fixtures.json entries now decode to their reference JSON exactly.
• Integration tests compile and gain live checks for a v2 timestamp seek, /xrpc/_health, and a HEAD probe matching listSegments (all gated; skipped locally).
• Benchmarks (BenchmarkDotNet ShortRun, .NET 10, this branch; a 1.8 KB like commit with a 1.45 KB CAR; a 4,096-row archive block, 200 accounts, filter wants one):

  	Mean	Allocated
  Full parse of a #commit frame (old cursor path, and every frame a filter dropped)	2,576 ns	10,256 B
  Seq-only scan of a #commit body	237 ns	416 B
  Consumer, per commit, filter excludes it (path peek)	713 ns	937 B
  Consumer, per commit, full parse	2,837 ns	10,264 B
  Archive block: decode rows, ToEvent() each, then filter (old order)	6,470 µs	9,803 KB
  Archive block: filter on columns, then ToEvent()	389 µs	855 KB

Deliberately deferred / decisions to flag

• Sync 1.1 inductive verification (MST inversion against prevData, per-repo rev tracking) stays with the follow-up issue; the FirehoseVerifier doc no longer claims MST-proof verification.
• Chat moderation cursor is not persisted through IStreamCursorStore: it is a revision string, not a sequence number. The consumer resumes after its last Rev across reconnects and exposes LastRev; persisting it across restarts is the caller's job (documented).
• Deprecated CommitEvent.TooBig/Rebase/Blobs are [Obsolete], not removed, since the Lexicon still lists them as required and the drift test expects them declared.
• #sync events are CID-checked when either VerifyCids or a Verifier is set; their signature check is part of the Sync 1.1 follow-up.
• Firehose/label stream frames keep [JsonPolymorphic] with short #type discriminators (the header names the type), excluded from the open-union guard like FirehoseMessage already was; the chat moderation union uses full NSID discriminators and [AtProtoUnion].
• Options keep ILogger rather than moving to ILoggerFactory (WP15.1 mentions it; not in this issue's checklist).
• PlcClient.StreamExportAsync still has its own WebSocket code; moving it onto StreamSocket is a possible follow-up outside this issue's area.
• Edits outside Streaming/: Http/XrpcParams.cs (internal ToQueryString()), Repo/DagCborJson.cs (unpadded $bytes), the Sync/Label/Chat-moderation model files, AtProtoException.cs (doc), ATProtoNet.Server/Aspire/JetstreamHealthCheck.cs (new). No files owned by #125 or #132 were touched.

🤖 Generated with Claude Code

Closes #126. One streaming core for the firehose, label, chat moderation and Jetstream streams: a single WebSocket helper, one reconnect policy, one cursor tracker and one cursor store; correct cursors; typed error frames; and less CPU per event. Also covers the scope added in the issue comments (typed `CollectionFilter`, `IRecordEvent`, one operation enum, the `docs/firehose.md` cursor example, a drop hook, the `$bytes` padding bug, `subscribeModEvents`) and the entries #123 deferred to this issue. ## Consolidation - **One WebSocket helper** (`StreamSocket`, internal): connect (subprotocol, `Authorization`), receive with reassembly into one reused buffer (capped at 64 MiB), graceful close on dispose. Every stream client and consumer goes through it; each subscription owns its socket, which fixes the leak when `FirehoseClient.SubscribeAsync` was called twice. A `StreamConnector` delegate is the test seam. - **One reconnect loop for the atproto-framed streams** (`EventStreamLoop` + `EventStreamHandler<T>`, internal): header parsing, error frames, reconnect and cancellation live once; the firehose, label and chat moderation handlers only turn bodies into messages and track their position. - **`StreamReconnectPolicy`** (`InitialDelay` 5 s, `MaxDelay` 30 s, `MaxAttempts` 10, `null` = unlimited), exponential with up to 10 % jitter, on every consumer through the new `StreamConsumerOptions` base. The attempt count resets when a connection delivers a frame. **Exhausted attempts throw** `EventStreamException` (inner = last failure). A non-retryable failure (`FutureCursor`, a 4xx other than 408/429) throws at once. - **`CursorTracker`** (internal): monotonic; filtered and failed-verification events advance it; saves are single-flight on a background task every `CursorPersistInterval` events; the final save runs in `finally` and never throws (a failed save is logged). Delivered events are recorded when the caller asks for the next one (at-least-once). - **`IStreamCursorStore`** (`ValueTask`) replaces `IFirehoseCursorStore`; `InMemoryStreamCursorStore`. - **`FirehoseConsumer` folded into `TypedFirehoseConsumer`**: each frame is parsed once. Frames that are skipped (filtered, unknown type, unparseable) move the cursor with a `seq`-only CBOR scan. - **`TypedFirehoseConsumer`**: with a `CollectionFilter` (now `IReadOnlySet<Nsid>`), `ops[].path` and `seq` are read straight from the CBOR before deserializing; a verifier verifies exactly when set (the `VerifySignatures` bool is gone) and its signature path also checks every block's CID, so the CAR is parsed and hashed once; `#identity` still invalidates the verifier's DID cache. - **Error frames** (`op = -1`) surface as `EventStreamException` (`Error`, `IsRetryable`) and are reported to `OnStreamError`; consumers reconnect after `ConsumerTooSlow` and throw `FutureCursor`. `FirehoseEventParser.Parse` throws for an error frame. - **`subscribeLabels` finished**: `LabelStreamMessage` → `LabelsEvent` / `LabelInfoEvent`, typed `FirehoseClient.SubscribeLabelsAsync`, and a reconnecting `LabelStreamConsumer` (a separate options type, so the label-signing issue #135 can add a verifier to it). - **Jetstream**: validation in one `JetstreamConsumerOptions.Validate()` and kind names in one `JetstreamKinds` helper; `JetstreamDictionaryClient` folded into `JetstreamArchiveClient.GetZstdDictionaryAsync` (no API key sent to the public endpoint); query strings through `XrpcParams` (new internal `ToQueryString()`); one `JetstreamException` (subclass of `EventStreamException`); shared event builders (`JetstreamEvents`) for the live parser and the segment reader, since a segment row's payload has the same fields the live frame nests; `JetstreamEventParser` keeps `ParseFrame(ReadOnlyMemory<byte>, protocol)`; dead members removed (`JetstreamPlanStats`/`Stats`, `IsConnected`, `PinnedTipSeq`, the no-op `Dispose`, the hand-rolled `ReadExactlyAsync`; `DecodeBlock` is internal). - `FirehoseClient` and `JetstreamClient` are `IAsyncDisposable`; disposing ends their running subscriptions. ## Protocol - **Jetstream v2 timestamp cursors** (jetstream #350/#352): `JetstreamCursor.FromTimestamp` / `IsTimestamp` / `TimestampThreshold` (10^15); `FromTimestamp` refuses times that would read as a sequence number. A v2 timestamp seek sets no dedup floor (never deduplicated by `time_us`); the consumer continues by sequence number from the first event, so a stored v1 cursor migrates to v2 unchanged (documented). A v2 sequence start cursor now also dedupes the server's inclusive replay of it. `JetstreamReplayConsumer` rejects a timestamp cursor, since the archive is sequence-addressed. - **Archive `HEAD` probes** (#347): `ProbeSegmentAsync` / `ProbeBlockAsync` → `JetstreamArchiveProbe(ContentLength, ETag)`. **Health**: `GetHealthAsync` (`/xrpc/_health` → `{version}`), plus `JetstreamHealthCheck` and `AddHealthChecks().AddJetstream(url)` in `ATProtoNet.Server/Aspire` (next to the PDS check). - **Firehose models** (deferred from #123): `HandleEvent` / `TombstoneEvent` removed; `Seq`/`Time` move to a new `FirehoseEvent` base, so `InfoEvent` no longer declares fields `#info` lacks; `TooBig`, `Rebase`, `Blobs` are `[Obsolete]` (still bound). The #126 drift allow-list entries are removed; new entries cover the new union bases. - **`chat.bsky.moderation.subscribeModEvents`**: `ChatModerationEventConsumer` with a per-connection bearer token provider (`GetAccessTokenAsync`, so each reconnect can mint a fresh service-auth token), resume after the last `Rev`, and `ChatModerationEvent` models for all 11 variants plus `UnknownChatModerationEvent` (open `[AtProtoUnion]`). - **`$bytes` padding**: `DagCborDecoder` (the Wrapped form of the shared transcoder) now writes unpadded base64, pinned against RFC 4648 vectors and the atproto-interop data-model fixtures' reference JSON. ## Also from the issue comments - `IRecordEvent` (Did, Collection, Rkey, Operation, Rev, Cid, Record, Uri, `GetRecord<T>`), implemented by `JetstreamCommitEvent` and by `FirehoseRecordEvent` from `CommitEvent.GetRecordEvents()` (records decoded from the commit's blocks). - `JetstreamOperation` removed in favour of `RepoOpAction`. - `OnEventDropped` hook with `DroppedStreamEvent(Reason, Cursor, Detail)` on every consumer (and `JetstreamClient`). - `docs/firehose.md` cursor example fixed (it read `message.Seq` off a raw frame); both streaming docs, `chat.md`, `error-handling.md` and `api-reference.md` updated; both samples updated. ## Performance - Firehose: collection-filtered commits are skipped after a path peek; skipped frames advance the cursor via a seq-only scan; a verified commit's CAR is parsed once. - Jetstream archive: rows are filtered on their DID/kind/collection columns before `ToEvent()`; matching rows alias the decompressed block instead of copying payloads; DID and collection (and rev) strings are interned per block; blocks are decoded in the download stage (`getBlock` units in their download task, whole segments `DownloadParallelism` blocks ahead), still delivered in order. - Jetstream live: frames are parsed straight from the socket's receive buffer (no per-frame copy); only a compressed frame is copied, by the decompressor. - Cursor persistence is off the read loop. ## Breaking changes (migration) - `ReconnectDelay` / `MaxReconnectAttempts` → `Reconnect = new StreamReconnectPolicy { InitialDelay = …, MaxAttempts = null }`. - Exhausted reconnects throw `EventStreamException`; cancellation ends every consumer normally (the replay consumer threw `OperationCanceledException` mid-backfill before). - `IFirehoseCursorStore` / `InMemoryFirehoseCursorStore` → `IStreamCursorStore` / `InMemoryStreamCursorStore`, returning `ValueTask`. - `FirehoseConsumer`, `FirehoseFrame`, `FirehoseEventParser.Parse(FirehoseFrame)` / `TryParse`, `FirehoseClient.DisconnectAsync`, `JetstreamClient.DisconnectAsync` removed; `FirehoseClient.SubscribeAsync` yields `FirehoseMessage`; the clients are `IAsyncDisposable` (`await using`); `TypedFirehoseConsumer` / `JetstreamConsumer` are no longer `IDisposable`. - `VerifySignatures` removed (set `Verifier`); `CollectionFilter` is `IReadOnlySet<Nsid>`. - `HandleEvent` / `TombstoneEvent` removed; `Seq`/`Time` on `FirehoseEvent` (`message is FirehoseEvent e ? e.Seq`); `LabelsEvent` derives from `LabelStreamMessage`. - `JetstreamOperation` → `RepoOpAction`. - `JetstreamConnectException` / `JetstreamArchiveException` → `JetstreamException`; `JetstreamStreamError` → `EventStreamError`; `JetstreamClient` throws on an error frame. - `JetstreamDictionaryClient` → `JetstreamArchiveClient.GetZstdDictionaryAsync`. - `JetstreamEventParser.Parse(...)` and the span/string `ParseFrame` overloads removed. - `JetstreamPlanStats`, `JetstreamSnapshotPlan.Stats`, `JetstreamConsumer.IsConnected`, `JetstreamReplayConsumer.PinnedTipSeq` removed; `JetstreamSegmentReader.DecodeBlock` internal. Each has a `Breaking changes` bullet with its migration note in the CHANGELOG. ## Verification - `dotnet build` (whole solution): 0 warnings, 0 errors. - Unit tests: 3,987 passed, 0 failed, 5 skipped (the pre-existing gated skips). Streaming + DAG-CBOR subset re-run 3× to check the background-save and socket tests for flakiness. - New tests: `TypedFirehoseConsumerTests` (filter advances the stored cursor, path peek before parse, `break`/cancellation keep the cursor, resume/explicit cursor, reconnect dedupe, exhausted attempts throw, `FutureCursor` throws / `ConsumerTooSlow` reconnects, drop reporting, verifier/`VerifyCids`/no verification); `EventStreamClientTests` (single-connection client, per-subscription sockets, dispose, labels, `LabelStreamConsumer`, chat moderation typed + unknown events, per-connection tokens, rev resume, backoff curve, retryability); `StreamSocketTests` against a real Kestrel WebSocket on loopback (fragment reassembly past the buffer, empty messages, text/binary, subprotocol and `Authorization`, refused upgrade status, close on dispose); `CursorTrackerTests` (monotonic, final flush, background interval save, slow store does not block and saves the newest, failing store does not throw); `JetstreamClientTests` (info out of band, receive-buffer parsing, error frame, refused upgrade, drop hook, decompressor); Jetstream timestamp-cursor, migration and error-frame consumer tests; replay cutover (inclusive seam dedupe, refused cutover re-plans from the last delivered event), timestamp rejection, cancellation, column filter; `RecordEventTests`; archive dictionary/health/`HEAD` probe tests; `JetstreamHealthCheck`. - Pinned vectors: RFC 4648 base64 vectors for `$bytes` (`Zg`, `Zm8`, `Zm9v`, `+/8`), and the three atproto-interop `data-model-fixtures.json` entries now decode to their reference JSON exactly. - Integration tests compile and gain live checks for a v2 timestamp seek, `/xrpc/_health`, and a `HEAD` probe matching `listSegments` (all gated; skipped locally). - Benchmarks (BenchmarkDotNet ShortRun, .NET 10, this branch; a 1.8 KB like commit with a 1.45 KB CAR; a 4,096-row archive block, 200 accounts, filter wants one): | | Mean | Allocated | |---|---:|---:| | Full parse of a `#commit` frame (old cursor path, and every frame a filter dropped) | 2,576 ns | 10,256 B | | Seq-only scan of a `#commit` body | 237 ns | 416 B | | Consumer, per commit, filter excludes it (path peek) | 713 ns | 937 B | | Consumer, per commit, full parse | 2,837 ns | 10,264 B | | Archive block: decode rows, `ToEvent()` each, then filter (old order) | 6,470 µs | 9,803 KB | | Archive block: filter on columns, then `ToEvent()` | 389 µs | 855 KB | ## Deliberately deferred / decisions to flag - **Sync 1.1 inductive verification** (MST inversion against `prevData`, per-repo rev tracking) stays with the follow-up issue; the `FirehoseVerifier` doc no longer claims MST-proof verification. - **Chat moderation cursor is not persisted** through `IStreamCursorStore`: it is a revision string, not a sequence number. The consumer resumes after its last `Rev` across reconnects and exposes `LastRev`; persisting it across restarts is the caller's job (documented). - **Deprecated `CommitEvent.TooBig`/`Rebase`/`Blobs` are `[Obsolete]`, not removed**, since the Lexicon still lists them as required and the drift test expects them declared. - **`#sync` events** are CID-checked when either `VerifyCids` or a `Verifier` is set; their signature check is part of the Sync 1.1 follow-up. - **Firehose/label stream frames keep `[JsonPolymorphic]` with short `#type` discriminators** (the header names the type), excluded from the open-union guard like `FirehoseMessage` already was; the chat moderation union uses full NSID discriminators and `[AtProtoUnion]`. - **Options keep `ILogger`** rather than moving to `ILoggerFactory` (WP15.1 mentions it; not in this issue's checklist). - `PlcClient.StreamExportAsync` still has its own WebSocket code; moving it onto `StreamSocket` is a possible follow-up outside this issue's area. - Edits outside `Streaming/`: `Http/XrpcParams.cs` (internal `ToQueryString()`), `Repo/DagCborJson.cs` (unpadded `$bytes`), the Sync/Label/Chat-moderation model files, `AtProtoException.cs` (doc), `ATProtoNet.Server/Aspire/JetstreamHealthCheck.cs` (new). No files owned by #125 or #132 were touched. 🤖 Generated with [Claude Code](https://claude.com/claude-code)

Grandiras added 1 commit 2026-09-26 01:20:43 +00:00

refactor(streaming)!: one streaming core, correct cursors, faster consumers (closes #126) ... 0153ac2bd8

The firehose, label, chat moderation and Jetstream streams each carried
their own socket, reconnect and cursor code, which had drifted apart: a
collection-filtered firehose consumer did not advance its cursor past the
commits it dropped, the final save could be skipped, VerifySignatures
without a Verifier verified nothing, error frames vanished, and a second
SubscribeAsync leaked a socket.

They now share one WebSocket helper, one reconnect loop and backoff policy
(which throws when attempts run out), one cursor tracker that saves in the
background and in finally, and one IStreamCursorStore. The typed firehose
consumer parses each frame once and peeks operation paths before parsing;
the Jetstream archive filters rows on their columns before decoding and
decodes in parallel. Jetstream gains v2 timestamp cursors, HEAD probes and
a health check; subscribeLabels and chat.bsky.moderation.subscribeModEvents
get typed consumers; DagCborDecoder writes unpadded $bytes; and the
firehose models follow the current subscribeRepos Lexicon.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

opencode-bot commented 2026-09-26 01:20:44 +00:00

Collaborator

Copy link

🤖 Reviewing this PR — the review will link the full session transcript.

📜 Live transcript — updating every ~5 minutes while the review runs.

🤖 Reviewing this PR — the review will link the full session transcript. <!-- opencode-run:review --> 📜 [Live transcript](https://git.grandiras.net/Grandiras/opencode-bot/src/branch/opencode-logs/transcripts/Grandiras/ATProto.NET/pr-168/run-2026-09-26T01-20-44-044Z.md) — updating every ~5 minutes while the review runs.

opencode-bot commented 2026-09-26 01:20:46 +00:00

Collaborator

Copy link

⏳ Held on the OpenCode Go window — the 5-hour window is at 0% (the bot holds at 95%).
This continues automatically after the window resets (2026-09-26T06:20:44.459Z) — nothing to re-trigger.

⏳ **Held on the OpenCode Go window** — the 5-hour window is at 0% (the bot holds at 95%). This continues automatically after the window resets (2026-09-26T06:20:44.459Z) — nothing to re-trigger. <!-- opencode-run:paused --> <!-- opencode-session:ses_f24b31733ffeN7DSTmi9bRjmkz -->

Grandiras commented 2026-09-26 01:20:51 +00:00

Author

Owner

Copy link

@claude /review — large (71 files); the description has the migration notes and benchmarks. The diff will be truncated, so please read these in full from the branch: Streaming/StreamSocket.cs, EventStreamLoop.cs, StreamReconnectPolicy.cs, CursorTracker.cs (cursor correctness: filtered events advance, saves serialized, final save in finally), EventStreamFrame.cs, TypedFirehoseConsumer.cs (parse-once path, verification when a Verifier is set, #sync CID check), FirehoseClient.cs, JetstreamClient.cs, JetstreamConsumer.cs, JetstreamReplayConsumer.cs, JetstreamSegmentReader.cs (column filtering before decode, parallel decode, interning), LabelStreamConsumer.cs, ChatModerationEventConsumer.cs, and Repo/DagCborJson.cs (unpadded $bytes). Focus on socket/subscription lifetime and disposal, reconnect/backoff and cancellation, cursor loss or regression on reconnect, concurrency in the background cursor saves, and unbounded memory (buffers, interning tables, parallel decode).

@claude /review — large (71 files); the description has the migration notes and benchmarks. The diff will be truncated, so please read these in full from the branch: `Streaming/StreamSocket.cs`, `EventStreamLoop.cs`, `StreamReconnectPolicy.cs`, `CursorTracker.cs` (cursor correctness: filtered events advance, saves serialized, final save in finally), `EventStreamFrame.cs`, `TypedFirehoseConsumer.cs` (parse-once path, verification when a Verifier is set, #sync CID check), `FirehoseClient.cs`, `JetstreamClient.cs`, `JetstreamConsumer.cs`, `JetstreamReplayConsumer.cs`, `JetstreamSegmentReader.cs` (column filtering before decode, parallel decode, interning), `LabelStreamConsumer.cs`, `ChatModerationEventConsumer.cs`, and `Repo/DagCborJson.cs` (unpadded `$bytes`). Focus on socket/subscription lifetime and disposal, reconnect/backoff and cancellation, cursor loss or regression on reconnect, concurrency in the background cursor saves, and unbounded memory (buffers, interning tables, parallel decode).

claude-bot commented 2026-09-26 01:20:54 +00:00

Collaborator

Copy link

🤖 Reviewing this PR — follow along in the action run; the reply will link the full session transcript.

📜 Live transcript — updating every ~5 minutes while the task runs.

🤖 Reviewing this PR — follow along in [the action run](https://git.grandiras.net/Grandiras/claude-bot/actions/runs/612); the reply will link the full session transcript. <!-- claude-run:pr_review --> 📜 [Live transcript](https://git.grandiras.net/Grandiras/claude-bot/src/branch/claude-logs/transcripts/Grandiras/ATProto.NET/pr-168/run-612.md) — updating every ~5 minutes while the task runs.

claude-reviewer approved these changes 2026-09-26 01:23:32 +00:00

claude-reviewer left a comment

Copy link

✅ Verdict: approve — no blocking issues found.

Reviewed the core streaming primitives introduced by this refactor in depth: StreamSocket (connect/reassembly/dispose), EventStreamFrame/EventStreamLoop (header parsing, error-frame handling, reconnect vs. single-connection semantics), StreamReconnectPolicy/ReconnectBackoff (backoff math, MaxAttempts = 0 giving up immediately, attempt reset on frame receipt), CursorTracker (monotonic advance, single-flight background saves, final flush-in-finally), TypedFirehoseConsumer (CBOR path-peek filtering before full parse, inclusive-cursor dedup via Seq <= cursor.Current, drop reporting), JetstreamConsumer (v1 rewind vs. v2 inclusive-replay dedup, the timestamp-seek-then-sequence-number transition logic), JetstreamReplayConsumer (pinned-tip cutover, IsNew dedup shared between backfill and live tail, re-planning after a refused cutover), RecordEvents.cs, and the new unpadded-base64 writer in DagCborJson.

Everything checked out. The unpadded-base64 routine (WriteUnpaddedBase64) looks confusing at first glance (index arithmetic against a buffer offset by 1) but is correct: buffer[written] after encoding into buffer[1..] really is the last encoded byte, and the empty-input case produces "" correctly. The cursor dedup/backoff/reconnect state machines handle the edge cases the PR description calls out (inclusive replay, timestamp-to-sequence migration, cutover refusal, cancellation-is-not-failure) consistently with their documentation, and Advance/IsNew are defensively monotonic so a stale or out-of-order frame can't rewind the stored cursor.

This is a very large, carefully engineered refactor (~3100 lines touched in Streaming/ alone, plus docs and samples kept in sync) and I could not exhaustively re-derive every code path in the remaining files (JetstreamEventParser, JetstreamSegmentReader, FirehoseVerifier, JetstreamArchiveClient, and the full test suite) given the scope, but the primitives that most other code depends on are sound, the breaking-change/CHANGELOG bookkeeping matches the actual diff, and the docs were updated consistently with the API changes. No blocking issues found.

---

⏱ 2m 23s · 16 turns · tokens: 118.7k in / 13.8k out (+1864.2k cache-read)
📜 Full session transcript

✅ **Verdict: approve** — no blocking issues found. Reviewed the core streaming primitives introduced by this refactor in depth: `StreamSocket` (connect/reassembly/dispose), `EventStreamFrame`/`EventStreamLoop` (header parsing, error-frame handling, reconnect vs. single-connection semantics), `StreamReconnectPolicy`/`ReconnectBackoff` (backoff math, `MaxAttempts = 0` giving up immediately, attempt reset on frame receipt), `CursorTracker` (monotonic advance, single-flight background saves, final flush-in-finally), `TypedFirehoseConsumer` (CBOR path-peek filtering before full parse, inclusive-cursor dedup via `Seq <= cursor.Current`, drop reporting), `JetstreamConsumer` (v1 rewind vs. v2 inclusive-replay dedup, the timestamp-seek-then-sequence-number transition logic), `JetstreamReplayConsumer` (pinned-tip cutover, `IsNew` dedup shared between backfill and live tail, re-planning after a refused cutover), `RecordEvents.cs`, and the new unpadded-base64 writer in `DagCborJson`. Everything checked out. The unpadded-base64 routine (`WriteUnpaddedBase64`) looks confusing at first glance (index arithmetic against a buffer offset by 1) but is correct: `buffer[written]` after encoding into `buffer[1..]` really is the last encoded byte, and the empty-input case produces `""` correctly. The cursor dedup/backoff/reconnect state machines handle the edge cases the PR description calls out (inclusive replay, timestamp-to-sequence migration, cutover refusal, cancellation-is-not-failure) consistently with their documentation, and `Advance`/`IsNew` are defensively monotonic so a stale or out-of-order frame can't rewind the stored cursor. This is a very large, carefully engineered refactor (~3100 lines touched in `Streaming/` alone, plus docs and samples kept in sync) and I could not exhaustively re-derive every code path in the remaining files (`JetstreamEventParser`, `JetstreamSegmentReader`, `FirehoseVerifier`, `JetstreamArchiveClient`, and the full test suite) given the scope, but the primitives that most other code depends on are sound, the breaking-change/CHANGELOG bookkeeping matches the actual diff, and the docs were updated consistently with the API changes. No blocking issues found. --- ⏱ 2m 23s · 16 turns · tokens: 118.7k in / 13.8k out (+1864.2k cache-read) 📜 [Full session transcript](https://git.grandiras.net/Grandiras/claude-bot/src/branch/claude-logs/transcripts/Grandiras/ATProto.NET/pr-168/run-612.md) <!-- claude-verdict:approve --> <!-- claude-reviewed-head:0153ac2bd8c270d3d4fce37fffcc432e693af545 -->

Grandiras merged commit 810e2deff1 into main 2026-09-26 01:24:45 +00:00

Grandiras deleted branch issue-126-streaming 2026-09-26 01:24:45 +00:00

Grandiras referenced this pull request from a commit 2026-09-26 01:24:46 +00:00

refactor(streaming)!: one streaming core, correct cursors, faster consumers (#168)

Grandiras referenced this pull request 2026-09-26 15:45:54 +00:00

feat(oauth)!: hardened OAuth client with confidential-client support #167

Sign in to join this conversation.

Reviewers

No reviewers

claude-reviewer

Labels

Clear labels   

breaking-change

Introduces a breaking API change

  

bug

Something is not working correctly

  

documentation

Documentation improvements

  

duplicate

This issue or PR already exists

  

enhancement

New feature or improvement

  

good first issue

Good for newcomers to the project

  

help wanted

Extra attention is needed

  

performance

Performance-related issue or improvement

  

question

Further information is requested

  

wontfix

This will not be worked on

No labels

breaking-change

bug

documentation

duplicate

enhancement

good first issue

help wanted

performance

question

wontfix

Milestone

Clear milestone

No items

No milestone

Projects

Clear projects

No items

No project

Assignees

Clear assignees

No assignees

4 participants

Notifications

Subscribe

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference

Grandiras/ATProto.NET!168

No description provided.