fix(collector): prevent queued events crossing segment bounds - #346
Conversation
|
Warning Review limit reached
Next review available in: 31 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 (2)
📝 WalkthroughWalkthroughThe archiver now separates manifest timestamps from segment identity timestamps, tracks in-flight senders, and drains queued events before and after sender quiescence before rotating segments. Tests cover timestamp handling, sender cancellation, and hot-queue rotation behavior. ChangesArchiver Boundary Handling
Sequence Diagram(s)sequenceDiagram
participant run_session
participant receiver_queue
participant process_event
participant ACTIVE_SENDS
participant rotate_segment
run_session->>receiver_queue: snapshot queued event count
run_session->>process_event: drain queued events
run_session->>ACTIVE_SENDS: wait for active sends to reach zero
run_session->>receiver_queue: snapshot newly queued event count
run_session->>process_event: drain newly queued events
run_session->>rotate_segment: rotate after drains and quiescence
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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 |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 011528925e
ℹ️ 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".
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
rust_hft/tools/collector/src/bin/binance-lob-archiver.rs (1)
897-947: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated drain-and-check logic before/after quiescence.
The pre-wait and post-wait blocks (compute queue length →
drain_events_before_segment_rotation→ handlerestarts_capture_session()→ handleInitialSnapshotsComplete) are copy-pasted verbatim aside from the variable name. Extracting a small helper would remove ~20 duplicated lines and reduce the risk of the two copies drifting apart on a future edit.♻️ Proposed helper extraction
fn drain_before_rotation( config: &Config, receiver: &mut mpsc::Receiver<Event>, segment: &mut Segment, states: &mut HashMap<String, OrderBookState>, budget: &mut PendingBudget, session_id: &str, process_state: &mut ProcessState, sync_deadline: &mut Option<Instant>, sync_timeout: Duration, ) -> anyhow::Result<bool> { let queued = receiver.len(); let action = drain_events_before_segment_rotation( config, receiver, segment, states, budget, session_id, process_state, queued, )?; if matches!(action, ProcessAction::InitialSnapshotsComplete) { *sync_deadline = Some(Instant::now() + sync_timeout); } Ok(action.restarts_capture_session()) }- let queued_events = receiver.len(); - let rotation_action = match drain_events_before_segment_rotation( - &config, &mut receiver, &mut segment, &mut states, &mut budget, - &session_id, &mut process_state, queued_events, - ) { - Ok(action) => action, - Err(error) => { failure = Some(error); break; } - }; - if rotation_action.restarts_capture_session() { break; } - if matches!(rotation_action, ProcessAction::InitialSnapshotsComplete) { - sync_deadline = Some(Instant::now() + config.sync_timeout); - } + match drain_before_rotation( + &config, &mut receiver, &mut segment, &mut states, &mut budget, + &session_id, &mut process_state, &mut sync_deadline, config.sync_timeout, + ) { + Ok(true) => break, + Ok(false) => {} + Err(error) => { failure = Some(error); break; } + } if let Err(error) = wait_for_active_sends_before_rotation().await { failure = Some(error); break; } - let queued_after_sends = receiver.len(); - let rotation_action = match drain_events_before_segment_rotation( - &config, &mut receiver, &mut segment, &mut states, &mut budget, - &session_id, &mut process_state, queued_after_sends, - ) { - Ok(action) => action, - Err(error) => { failure = Some(error); break; } - }; - if rotation_action.restarts_capture_session() { break; } - if matches!(rotation_action, ProcessAction::InitialSnapshotsComplete) { - sync_deadline = Some(Instant::now() + config.sync_timeout); - } + match drain_before_rotation( + &config, &mut receiver, &mut segment, &mut states, &mut budget, + &session_id, &mut process_state, &mut sync_deadline, config.sync_timeout, + ) { + Ok(true) => break, + Ok(false) => {} + Err(error) => { failure = Some(error); break; } + }🤖 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 897 - 947, Extract the duplicated pre- and post-quiescence logic around drain_events_before_segment_rotation into a helper such as drain_before_rotation, preserving queue-length capture, InitialSnapshotsComplete sync_deadline updates, and restarts_capture_session handling. Replace both inline blocks in the segment rotation flow with the helper and retain existing failure propagation and break behavior.
🤖 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/lob_archiver.rs`:
- Around line 1049-1050: Update send_or_shutdown to manage ACTIVE_SENDS with an
RAII guard created immediately after incrementing the counter, so its Drop
implementation decrements the counter on normal completion or cancellation.
Remove the manual decrement from the async send path while preserving the
existing counter semantics and wait_for_active_sends_before_rotation behavior.
---
Nitpick comments:
In `@rust_hft/tools/collector/src/bin/binance-lob-archiver.rs`:
- Around line 897-947: Extract the duplicated pre- and post-quiescence logic
around drain_events_before_segment_rotation into a helper such as
drain_before_rotation, preserving queue-length capture, InitialSnapshotsComplete
sync_deadline updates, and restarts_capture_session handling. Replace both
inline blocks in the segment rotation flow with the helper and retain existing
failure propagation and break behavior.
🪄 Autofix (Beta)
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: 5106511c-5ed5-4225-813a-1956ee253ad4
📒 Files selected for processing (2)
rust_hft/tools/collector/src/bin/binance-lob-archiver.rsrust_hft/tools/collector/src/lob_archiver.rs
711f1ea to
9811119
Compare
2 participants
Change contract: drain already-received events before scheduled segment rotation and bind manifest receive bounds to actual archived rows, so market-tape strict verification cannot fail on collector-created boundary drift.
Acceptance evidence:
Out of scope: no snapshot logic, research logic, gate policy, service unit, or ECS production cutover changes.
Dependency / merge order: None; based on main b49b379.
Rollout / rollback: merge and publish a new Rust collector artifact, install as shadow candidate, rerun the short diagnostic gate, then require the formal production gate before cutover. Roll back by restoring the prior collector release; production remains unchanged until the gate passes.
Summary by CodeRabbit
Bug Fixes
Tests