Grandiras/ATProto.NET
1
1
Fork 0
Code Issues Pull requests Projects Releases 9 Packages 8 Wiki Activity Actions

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

New issue
Closed
opened 2026-09-24 12:39:07 +00:00 by Grandiras · 3 comments
Grandiras commented 2026-09-24 12:39:07 +00:00
Owner
Copy link

Part of the v0.7.0 update cycle (wave 5). Firehose and Jetstream: one streaming core, correct cursors, and less CPU per event. Sync 1.1 verification builds on this in a follow-up issue.

Consolidation

  • One internal WebSocket helper (connect, receive, reassemble, dispose) shared by FirehoseClient and JetstreamClient. This also stops a second SubscribeAsync leaking a socket.
  • One StreamReconnectPolicy (InitialDelay, MaxDelay, int? MaxAttempts, where null means unlimited instead of -1) on every consumer's options. When attempts run out, ConsumeAsync throws instead of completing normally as if the stream had ended.
  • One CursorTracker: the final save happens in finally, and commits dropped by a collection filter still advance the cursor. Today a rarely matching filter replays everything since the last match after a restart. Persist cursors off the read loop (single-flight), so a slow store doesn't stall the socket.
  • IFirehoseCursorStore → IStreamCursorStore (with ValueTask methods), shared by both streams. Stream clients implement IAsyncDisposable.
  • Fold FirehoseConsumer into the typed consumer so each frame is parsed once; today TryReadSeq fully re-parses every frame. Otherwise use a seq-only CBOR scan, which is 11× cheaper.
  • TypedFirehoseConsumer:
    • peek ops[].path before deserializing when a collection filter is set;
    • verify the CAR once when both CID and signature checks are on;
    • VerifySignatures = true with no Verifier currently skips verification silently: verify exactly when a verifier is configured and delete the bool.
  • Surface event-stream error frames (op = -1, for example FutureCursor and ConsumerTooSlow) as a typed error instead of dropping them.
  • Finish subscribeLabels: parse #labels and #info into typed messages. It has 0 callers today and LabelsEvent is dead; the label-signing issue builds on this.
  • Jetstream:
    • one kind-mapping and validation helper;
    • JetstreamDictionaryClient folded into the archive client;
    • one JetstreamException;
    • shared event builders between the live parser and the segment reader;
    • fewer public parser entry points;
    • dead members removed (JetstreamPlanStats, IsConnected, PinnedTipSeq, the no-op Dispose, …).

Protocol

  • Jetstream v2 timestamp cursors (bluesky-social/jetstream #350/#352, 2026-09-21/23): a cursor ≥ 1e15 is a unix-µs seek. Add JetstreamCursor.FromTimestamp, document v1 → v2 cursor migration, and don't dedupe timestamp seeks by time_us.
  • Archive HEAD size probing (#347) and a Jetstream /xrpc/_health health check.

Performance

  • Archive replay filters on the row columns (DID, collection, kind) before ToEvent() (5.1 µs / 11.8 KB per row today). Decode blocks in the download stage (parallel, still in order). Intern the repetitive strings per block.
  • Jetstream live: parse single-receive frames straight from the receive buffer.

CHANGELOG Breaking changes (options, the exhausted-attempts behaviour, renamed interfaces, removed types), Fixed (cursor bugs, verification skip), Added (timestamp cursors, error frames, labels stream).

Part of the v0.7.0 update cycle (wave 5). Firehose and Jetstream: one streaming core, correct cursors, and less CPU per event. Sync 1.1 verification builds on this in a follow-up issue. ## Consolidation - [ ] One internal WebSocket helper (connect, receive, reassemble, dispose) shared by `FirehoseClient` and `JetstreamClient`. This also stops a second `SubscribeAsync` leaking a socket. - [ ] One `StreamReconnectPolicy` (`InitialDelay`, `MaxDelay`, `int? MaxAttempts`, where `null` means unlimited instead of `-1`) on every consumer's options. When attempts run out, `ConsumeAsync` **throws** instead of completing normally as if the stream had ended. - [ ] One `CursorTracker`: the final save happens in `finally`, and commits dropped by a collection filter still advance the cursor. Today a rarely matching filter replays everything since the last match after a restart. Persist cursors off the read loop (single-flight), so a slow store doesn't stall the socket. - [ ] `IFirehoseCursorStore` → `IStreamCursorStore` (with `ValueTask` methods), shared by both streams. Stream clients implement `IAsyncDisposable`. - [ ] Fold `FirehoseConsumer` into the typed consumer so each frame is parsed once; today `TryReadSeq` fully re-parses every frame. Otherwise use a seq-only CBOR scan, which is 11× cheaper. - [ ] `TypedFirehoseConsumer`: - peek `ops[].path` before deserializing when a collection filter is set; - verify the CAR once when both CID and signature checks are on; - `VerifySignatures = true` with no `Verifier` currently skips verification silently: verify exactly when a verifier is configured and delete the bool. - [ ] Surface event-stream error frames (`op = -1`, for example `FutureCursor` and `ConsumerTooSlow`) as a typed error instead of dropping them. - [ ] Finish `subscribeLabels`: parse `#labels` and `#info` into typed messages. It has 0 callers today and `LabelsEvent` is dead; the label-signing issue builds on this. - [ ] Jetstream: - one kind-mapping and validation helper; - `JetstreamDictionaryClient` folded into the archive client; - one `JetstreamException`; - shared event builders between the live parser and the segment reader; - fewer public parser entry points; - dead members removed (`JetstreamPlanStats`, `IsConnected`, `PinnedTipSeq`, the no-op `Dispose`, …). ## Protocol - [ ] Jetstream v2 **timestamp cursors** (bluesky-social/jetstream #350/#352, 2026-09-21/23): a cursor ≥ 1e15 is a unix-µs seek. Add `JetstreamCursor.FromTimestamp`, document v1 → v2 cursor migration, and don't dedupe timestamp seeks by `time_us`. - [ ] Archive `HEAD` size probing (#347) and a Jetstream `/xrpc/_health` health check. ## Performance - [ ] Archive replay filters on the row columns (DID, collection, kind) **before** `ToEvent()` (5.1 µs / 11.8 KB per row today). Decode blocks in the download stage (parallel, still in order). Intern the repetitive strings per block. - [ ] Jetstream live: parse single-receive frames straight from the receive buffer. CHANGELOG `Breaking changes` (options, the exhausted-attempts behaviour, renamed interfaces, removed types), `Fixed` (cursor bugs, verification skip), `Added` (timestamp cursors, error frames, labels stream).
Grandiras commented 2026-09-24 23:35:58 +00:00
Author
Owner
Copy link

From #119 part 4 (#157), left to this issue:

  • TypedFirehoseConsumerOptions.CollectionFilter is still a set of strings.
  • There is no shared IRecordEvent across firehose and Jetstream.
  • RepoOpAction and JetstreamOperation are separate enums with the same members; consider unifying them.
  • The docs/firehose.md "With Cursor" example reads message.Seq from a raw frame that has no such member.
  • Behaviour note: firehose frames and Jetstream commits whose identifiers don't parse are now dropped (an unparseable rev/handle becomes null and the event is kept). Surface drops through a counter or log hook when consolidating.
From #119 part 4 (#157), left to this issue: - `TypedFirehoseConsumerOptions.CollectionFilter` is still a set of strings. - There is no shared `IRecordEvent` across firehose and Jetstream. - `RepoOpAction` and `JetstreamOperation` are separate enums with the same members; consider unifying them. - The `docs/firehose.md` "With Cursor" example reads `message.Seq` from a raw frame that has no such member. - **Behaviour note:** firehose frames and Jetstream commits whose identifiers don't parse are now dropped (an unparseable `rev`/`handle` becomes null and the event is kept). Surface drops through a counter or log hook when consolidating.
Grandiras commented 2026-09-25 20:54:41 +00:00
Author
Owner
Copy link

Also from #129: DagCborDecoder (the shared DAG-CBOR→JSON transcoder from #111) writes $bytes values with base64 padding, but the atproto data model specifies base64 without padding. Fix it while consolidating the transcoder, with a pinned test.

Also from #129: `DagCborDecoder` (the shared DAG-CBOR→JSON transcoder from #111) writes `$bytes` values with base64 **padding**, but the atproto data model specifies base64 without padding. Fix it while consolidating the transcoder, with a pinned test.
Grandiras commented 2026-09-25 21:00:33 +00:00
Author
Owner
Copy link

Also from #128: chat.bsky.moderation.subscribeModEvents (an authenticated WebSocket subscription) was deferred to the consolidated streaming core. Add it here, with typed events and an unknown fallback.

Also from #128: `chat.bsky.moderation.subscribeModEvents` (an authenticated WebSocket subscription) was deferred to the consolidated streaming core. Add it here, with typed events and an unknown fallback.
Sign in to join this conversation.
No Branch/Tag specified
Branches Tags
main
v0.8.0
v0.7.0
v0.6.0
v0.5.0
v0.4.0
v0.3.0
v0.2.0
v0.1.1
v0.1.0
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
1 participant
Notifications
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#126
No description provided.