subscribeRepos accepted the WebSocket and only then enumerated
PdsSequencer.SubscribeAsync, which registers the subscriber lazily on
first read. A peer that acted the instant its connect returned could
have the resulting event published into that gap, where it was in
neither the replay snapshot nor the live channel — a live-only
subscriber then waited forever for a frame that was already gone.

Add PdsSequencer.Subscribe(cursor), which registers eagerly and returns
a disposable PdsFirehoseSubscription carrying the replay snapshot and
the sequence state captured under the same lock. The handler now
subscribes before the handshake, so "connected" and "receiving" are the
same instant, and the FutureCursor/OutdatedCursor checks read the
snapshot instead of racing a concurrent publish.

SubscribeAsync is unchanged and delegates to the new method.

closes #66

claude-bot referenced this pull request

2026-07-25 17:43:10 +00:00

Test failure #66

claude-bot commented

2026-07-25 17:43:14 +00:00

Author

Collaborator

🤖 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/69); the reply will link the full session transcript. 📜 [Live transcript](https://git.grandiras.net/Grandiras/claude-bot/src/branch/claude-logs/transcripts/Grandiras/ATProto.NET/pr-67/run-69.md) — updating every ~5 minutes while the task runs.

claude-bot reviewed

2026-07-25 17:44:14 +00:00

claude-bot left a comment

Clean refactor that hoists subscriber registration out of SubscribeAsync's lazy generator into an eager PdsSequencer.Subscribe(cursor) returning a disposable PdsFirehoseSubscription. The replay snapshot, CurrentSeq, and OldestAvailableSeq are all captured under the same lock at registration time, so the WebSocket handler can register before AcceptWebSocketAsync and validate the cursor against a state snapshot rather than the sequencer's live state — correctly closing the race described in issue #66. SubscribeAsync now delegates to Subscribe + ReadAllAsync, preserving prior behavior (backlog snapshot, single-reader bounded channel, drop-on-full semantics, subscriber removal in finally/Dispose). Dispose is idempotent (tested), double-enumeration is guarded (tested), and the new unit tests cover buffering-before-read, snapshot isolation from a later publish, dispose idempotency, and single-enumeration enforcement. CHANGELOG entries are appropriately placed under Added/Fixed. No correctness, security, or omission issues found; a couple of very minor no-ops (early cursor-error path still transiently registers/disposes a subscription, the seq<=lastReplayed guard in ReadAllAsync is now unreachable since replay and channel seq ranges can't overlap) aren't worth blocking on.

⏱ 0m 39s · 1 turns · tokens: 20.6k in / 3.7k out (+23.7k cache-read)
📜 Full session transcript

Clean refactor that hoists subscriber registration out of `SubscribeAsync`'s lazy generator into an eager `PdsSequencer.Subscribe(cursor)` returning a disposable `PdsFirehoseSubscription`. The replay snapshot, `CurrentSeq`, and `OldestAvailableSeq` are all captured under the same lock at registration time, so the WebSocket handler can register before `AcceptWebSocketAsync` and validate the cursor against a state snapshot rather than the sequencer's live state — correctly closing the race described in issue #66. `SubscribeAsync` now delegates to `Subscribe` + `ReadAllAsync`, preserving prior behavior (backlog snapshot, single-reader bounded channel, drop-on-full semantics, subscriber removal in `finally`/`Dispose`). Dispose is idempotent (tested), double-enumeration is guarded (tested), and the new unit tests cover buffering-before-read, snapshot isolation from a later publish, dispose idempotency, and single-enumeration enforcement. CHANGELOG entries are appropriately placed under Added/Fixed. No correctness, security, or omission issues found; a couple of very minor no-ops (early cursor-error path still transiently registers/disposes a subscription, the seq<=lastReplayed guard in `ReadAllAsync` is now unreachable since replay and channel seq ranges can't overlap) aren't worth blocking on. --- ⏱ 0m 39s · 1 turns · tokens: 20.6k in / 3.7k out (+23.7k cache-read) 📜 [Full session transcript](https://git.grandiras.net/Grandiras/claude-bot/src/branch/claude-logs/transcripts/Grandiras/ATProto.NET/pr-67/run-69.md)