fix(collector): shard snapshot bootstrap across N producers and pin covered=0 fail-closed invariant (#857) - #858
Merged
Conversation
…overed=0 fail-closed invariant (#857) The all-or-nothing checkpoint gate, combined with a single sequential REST snapshot producer (RTT-bound to ~2 symbols/s), cannot sync 1377 symbols inside a short shadow segment. Every segment then closes with has_replay_safe_checkpoint =false and covered=0, which the strict verifier rejects after a 37-minute CI + install + soak. - Add SNAPSHOT_PRODUCERS (default 8); shard active_symbols() round-robin across N producers and fan the "all websocket shards connected" notification out over a tokio broadcast channel. Subscribers register synchronously in the spawn loop before any task polls, so no connection notification is missed. - Account every snapshot producer in the rotation barrier (shards + N); emit StreamCoverageVerified once from the lead producer. - Pin the fail-closed invariant with a regression guard: a segment that closes before every declared symbol syncs yields has_replay_safe_checkpoint=false, covered_symbol_count=0, and verify_binance_market_tape_for_strict_gate rejects it.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughThe Binance LOB archiver now supports configurable parallel snapshot producers. It partitions symbols across producers, broadcasts websocket coverage notifications, shares REST rate limiting, synchronizes all producers during rotation, and expands coordination tests. ChangesParallel snapshot production
Estimated code review effort: 4 (Complex) | ~60 minutes Mergeability Score: ⚪ Minimal · up to The PR parallelizes snapshot bootstrap and adds a fail-closed regression guard; based on the supplied validation, no actionable merge-blocking risk remains. Sequence Diagram(s)sequenceDiagram
participant WebsocketReaders
participant BroadcastChannel
participant LeadSnapshotProducer
participant SnapshotProducers
participant RotationBarrier
WebsocketReaders->>BroadcastChannel: Publish verified stream coverage
BroadcastChannel->>LeadSnapshotProducer: Deliver coverage notification
LeadSnapshotProducer->>SnapshotProducers: Start assigned snapshot work
SnapshotProducers->>RotationBarrier: Report completion and synchronize rotation
Possibly related issues
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust_hft/tools/collector/src/bin/binance-lob-archiver.rs`:
- Around line 1254-1266: Aggregate InitialSnapshotsComplete events from every
snapshot producer before initializing sync_deadline; currently the first shard
completion starts the timeout prematurely. Update the event-handling flow around
produce_snapshots_after_streams_connect and sync_deadline to track
expected_shards completions, then begin the deadline only after all shards
report completion. Add a multi-producer test covering one delayed shard.
- Around line 3342-3346: Replace the per-producer interval created near the
snapshot request loop with one shared rate limiter initialized before all
producers start, and have every producer acquire permission from it before
issuing a REST snapshot request. Ensure SNAPSHOT_REQUESTS_PER_SECOND is enforced
globally across producers, then add a test that verifies aggregate request
pacing rather than per-producer pacing.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 28f5f091-eb15-42db-82c4-cdbd7502832a
📒 Files selected for processing (1)
rust_hft/tools/collector/src/bin/binance-lob-archiver.rs
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7b1f839b22
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
… limit - Count InitialSnapshotsComplete across every producer before arming the sync deadline, so a fast shard cannot start the clock early. - Replace per-producer interval sleeps with one shared tokio::time::Interval so N shards cannot multiply SNAPSHOT_REQUESTS_PER_SECOND by N. - Subscribe snapshot producers before spawning websocket tasks so an early coverage notification cannot be dropped into an empty broadcast set.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Change
Nconcurrent producers (SNAPSHOT_PRODUCERS, default8) instead of one sequential REST producer. Active symbols are sharded round-robin; atokio::sync::broadcastchannel fans the "all websocket shards connected" notification out to every producer (subscribers register synchronously in the spawn loop before any task polls, so no notification is missed). The rotation barrier accounts forshards + Nproducers, andStreamCoverageVerifiedis emitted once by the lead producer. This removes the all-or-nothing bottleneck where a single RTT-bound producer (~2 symbols/s) could not sync 1377 symbols inside a short segment.has_replay_safe_checkpoint=false,covered_symbol_count=0, andverify_binance_market_tape_for_strict_gaterejects it.Issue relationship
Refs #857
Validation
cargo test -p hft-collector --features collector-binance --locked --bin binance-lob-archiver→ 103 passed.cargo clippy -p hft-collector --all-targets --features collector-binance --no-deps --locked -- -D warnings→ clean.git diff --check→ clean.binance_fee_uploadfailures on macOS (/varsymlink "directory has an indirect ancestor"), unrelated to this change and green on Linux CI.Runtime impact and rollback
SNAPSHOT_PRODUCERS(default8). Default config is the intended production path; no manifest/schema change. Rollback = revert this commit (restores the single-producer path).Summary by CodeRabbit
New Features
Bug Fixes
Tests