fix(collector): reconnect Binance shards without masking gaps - #341
Conversation
|
Warning Review limit reached
Next review available in: 44 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughThe Binance LOB archiver now reports websocket reconnect state, tracks recovery separately for all and depth streams, signals reconnect cycles, excludes lifecycle events from archives, and prevents replay-safe checkpoint publication until depth streams recover. ChangesReconnect health and replay safety
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Websocket
participant receive_url
participant process_event
participant ProcessState
participant Checkpoint
Websocket-->>receive_url: connection or proof failure
receive_url->>process_event: StreamDisconnected
process_event->>ProcessState: mark shard reconnecting
receive_url->>Websocket: reconnect with backoff
Websocket-->>receive_url: subscription proof succeeds
receive_url->>process_event: StreamReconnected
Websocket-->>process_event: depth and aggregate-trade events
process_event->>ProcessState: mark streams observed
Checkpoint->>ProcessState: check depth stream health
ProcessState-->>Checkpoint: allow or block replay-safe checkpoint
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 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.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 407c46fc4b
ℹ️ 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".
| .reconnecting_depth_shards | ||
| .iter() | ||
| .position(|shard| shard.contains(stream)) | ||
| { | ||
| self.reconnecting_depth_shards.remove(index); |
There was a problem hiding this comment.
When a depth shard contains multiple symbols, the first valid post-reconnect depth event removes the entire shard from reconnecting_depth_shards. replay_checkpoint_ready can then report readiness while the other symbols still retain pre-disconnect state and have not received an event that could reveal missed update IDs; a rotation in that interval publishes a falsely replay-safe checkpoint. Track pending depth streams or symbols individually and add a multi-symbol reconnect counterexample.
AGENTS.md reference: AGENTS.md:L93-L96
Useful? React with 👍 / 👎.
| process_state.depth_streams_healthy() | ||
| && segment.is_replay_safe() | ||
| && process_state.stream_coverage_trusted | ||
| && segment.event_count("agg_trade") > 0 |
There was a problem hiding this comment.
When the separately connected aggregate-trade shard disconnects, depth_streams_healthy() remains true and an aggregate trade recorded earlier in the segment satisfies the only trade check here. A scheduled rotation during that outage can therefore publish and upload a replay-safe segment before the next trade exposes any missed IDs, even though the declared replay scope includes captured aggregate trades. Gate readiness on all disconnected market-data shards, not only depth shards, and cover an aggregate-trade-only disconnect.
AGENTS.md reference: AGENTS.md:L93-L96
Useful? React with 👍 / 👎.
| let event = event_from_frame(frame, received_at_ns)?; | ||
| match receive_before_subscription_proof_deadline( | ||
| subscription_proof_deadline, | ||
| send_or_shutdown(&sender, event, &mut shutdown), | ||
| ) |
There was a problem hiding this comment.
On a reconnect, market frames received before the LIST_SUBSCRIPTIONS response are immediately sent to process_event, while the original stream_coverage_trusted flag is still true. Such a frame can clear the reconnect marker and permit a replay-safe rotation during the 20-second proof window even if the eventual listing is incomplete or times out; the later disconnect event cannot retract an already sealed segment. Buffer these frames or prevent them from restoring readiness until exact subscription coverage has been revalidated.
AGENTS.md reference: AGENTS.md:L93-L96
Useful? React with 👍 / 👎.
| process_state.depth_streams_healthy() | ||
| && segment.is_replay_safe() |
There was a problem hiding this comment.
If a disconnect spans a scheduled rotation, this new health gate makes seed_next false, so rotate_segment marks the next segment replay-unsafe. Even after every stream reconnects and validates continuity, segment.is_replay_safe() remains false; every later rotation repeats the same transition, and the new internal reconnect loop means the session never restarts to clear that latch. One transient outage crossing a boundary therefore makes all subsequent segments unusable for strict replay until the process is restarted. Establish a new trusted segment after complete recovery or restart the capture session at that point.
AGENTS.md reference: AGENTS.md:L93-L96
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rust_hft/tools/collector/src/bin/binance-lob-archiver.rs (1)
405-460: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftRequire every depth stream in a multi-symbol shard to be observed before clearing depth health.
stream_shards()batchesws_shard_sizesymbols per shard by default, and both depth and aggregate streams are tracked together.mark_stream_observed()removes a whole depth shard as soon as one@depth@100mssymbol is observed, whiledepth_streams_healthy()gatesreplay_checkpoint_ready, so a reconnect can seed the next replay-safe checkpoint even when the other symbol depth feeds on the same shard have not yet reported post-reconnect depth data. Track only depth streams inreconnecting_depth_shards, and remove an entry only when its tracked set becomes empty.🤖 Prompt for 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. In `@rust_hft/tools/collector/src/bin/binance-lob-archiver.rs` around lines 405 - 460, Update ProcessState::mark_shard_disconnected to track only `@depth`@100ms streams in reconnecting_depth_shards, while retaining the full shard in reconnecting_shards. In mark_stream_observed, remove the observed depth stream from the matching tracked set and delete that set only when empty, so depth_streams_healthy cannot recover until every depth stream in the shard has been observed.
🤖 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.
Outside diff comments:
In `@rust_hft/tools/collector/src/bin/binance-lob-archiver.rs`:
- Around line 405-460: Update ProcessState::mark_shard_disconnected to track
only `@depth`@100ms streams in reconnecting_depth_shards, while retaining the full
shard in reconnecting_shards. In mark_stream_observed, remove the observed depth
stream from the matching tracked set and delete that set only when empty, so
depth_streams_healthy cannot recover until every depth stream in the shard has
been observed.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 4e358cbd-64ce-4346-b948-93bed9eeede3
📒 Files selected for processing (1)
rust_hft/tools/collector/src/bin/binance-lob-archiver.rs
Change contract
Keep the Rust Binance market-tape collector alive across transient WebSocket disconnects while preserving fail-closed sequence-gap handling and requiring complete post-reconnect market-stream recovery before replay-safe checkpoints.
Acceptance evidence
cargo test -p hft-collector(214 library, 52 archiver, all passing; 2 ignored)cargo check -p hft-collector --features collector-binance --lockedcargo clippy -p hft-collector --all-targets --features collector-binance --no-deps --locked -- -D warningscargo test -p hft-collector --features collector-binance --lockedbash deployment/aliyun/test-rust-lob-control-plane.shOut of scope
No snapshot logic, research/evaluator logic, historical backfill, or production cutover.
Dependency / merge order
None.
Rollout / rollback impact
Shadow-only candidate verification first. Production remains on the existing binary until the required gate passes. Rollback is the existing candidate-to-previous-release pointer rollback; no production pointer is changed by this PR.