feat(collector): complete Rust full-catalog parity - #23
Conversation
📝 WalkthroughWalkthroughThe Rust Binance LOB archiver adds runtime symbol exclusion, bounded snapshot retries, watchdog monitoring, centralized session actions, dynamic segment catalogs, expanded tests, and deployment safeguards for shadow spool directories and ChangesRust archiver resilience
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant BinanceWebSocketShards
participant run_session
participant produce_snapshots
participant BinanceRESTAPI
participant Segment
BinanceWebSocketShards->>run_session: Diff and Snapshot events
run_session->>produce_snapshots: Schedule initial or resnapshot fetch
produce_snapshots->>BinanceRESTAPI: Fetch snapshot with retries
BinanceRESTAPI-->>produce_snapshots: Snapshot or invalid-symbol response
produce_snapshots->>run_session: InitialSnapshotsComplete or ExcludeSymbol
run_session->>Segment: Rotate and update catalog
🚥 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 |
| snapshot_limit: 100, | ||
| snapshot_requests_per_second: 15.0, | ||
| segment_seconds: 3600, | ||
| spool_dir: env::temp_dir(), |
| ) | ||
| .unwrap(); | ||
| }); | ||
| let root = env::temp_dir().join(format!("monday-exclusion-test-{}", now_ns().unwrap())); |
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 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 `@deployment/aliyun/README.md`:
- Around line 118-119: Add a mountpoint -q /data preflight immediately before
the install command in the setup instructions, aborting when /data is not
mounted; only run install -d for the spool directories after this check
succeeds.
In `@rust_hft/tools/collector/src/bin/binance-lob-archiver.rs`:
- Around line 698-707: Update archive_only and its callers to accept Config, and
in the Event::ExcludeSymbol branch call config.exclude_symbol for the excluded
symbol while preserving the existing symbol_excluded event recording. Add a
regression test covering the shutdown path that verifies the catalog/manifest no
longer lists the excluded symbol as active.
- Around line 273-285: Disarm the watchdog before graceful shutdown begins,
including the shutdown-only teardown in run_session, so segment draining and
closing cannot trigger an expiry exit. Update the monitor loop around stop() and
the armed/process_watchdog_expired check to re-check shutdown after sleeping and
before evaluating expiry, preventing a stop during sleep from reaching the
watchdog exit path.
- Around line 466-469: Update the ProcessAction::Excluded handling around
rotate_segment to detect when removing the final symbol leaves no states. Fail
or terminate the session before creating another segment, rather than allowing
empty-map all() checks to mark it synced; preserve normal segment rotation when
other symbols remain.
🪄 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
Run ID: 474c90bc-d61e-474c-8f7a-737fad59a2d4
📒 Files selected for processing (4)
deployment/aliyun/README.mddeployment/aliyun/binance-lob-archiver-rust@.servicerust_hft/tools/collector/src/bin/binance-lob-archiver.rsrust_hft/tools/collector/src/lob_archiver.rs
| sudo install -d -m 0750 -o hftcollector -g hftcollector \ | ||
| /data/monday/spool/binance-lob-rust-shadow/{spot,usdm} |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Verify /data is mounted before creating the spool directories.
If the mount is absent, install -d creates these paths on the root filesystem; the later /data mount hides them, leaving the real spool directories unprepared. Add a mountpoint -q /data preflight that aborts before running install -d.
🤖 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 `@deployment/aliyun/README.md` around lines 118 - 119, Add a mountpoint -q
/data preflight immediately before the install command in the setup
instructions, aborting when /data is not mounted; only run install -d for the
spool directories after this check succeeds.
| while !monitor.inner.shutdown.load(Ordering::Relaxed) { | ||
| thread::sleep(interval); | ||
| let now_ms = monitor.elapsed_ms(); | ||
| let last_ms = monitor.inner.last_data_ms.load(Ordering::Relaxed); | ||
| if monitor.inner.armed.load(Ordering::Relaxed) | ||
| && process_watchdog_expired(last_ms, now_ms, timeout) | ||
| { | ||
| error!( | ||
| silent_ms = now_ms.saturating_sub(last_ms), | ||
| "process watchdog exiting after market-data stall" | ||
| ); | ||
| std::process::exit(75); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Disarm the watchdog before graceful teardown.
The watchdog remains armed while run_session drains and closes the segment, potentially exiting after the 180-second default during a valid shutdown. Additionally, stop() during the monitor sleep does not prevent the subsequent expiry check.
Proposed fix
while !monitor.inner.shutdown.load(Ordering::Relaxed) {
thread::sleep(interval);
+ if monitor.inner.shutdown.load(Ordering::Relaxed) {
+ break;
+ }
let now_ms = monitor.elapsed_ms();Also stop it before shutdown-only session teardown:
let _ = session_stop_tx.send(true);
+ if *shutdown.borrow() {
+ watchdog.stop();
+ }
let final_queue_health = QueueHealth::from_sender(&sender);Also applies to: 299-301, 328-346, 525-525
🤖 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 273 -
285, Disarm the watchdog before graceful shutdown begins, including the
shutdown-only teardown in run_session, so segment draining and closing cannot
trigger an expiry exit. Update the monitor loop around stop() and the
armed/process_watchdog_expired check to re-check shutdown after sleeping and
before evaluating expiry, preventing a stop during sleep from reaching the
watchdog exit path.
| ProcessAction::Excluded => { | ||
| segment = | ||
| rotate_segment(segment, &config, &states, &session_id, "symbol_excluded")?; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Terminate the session when the final symbol is excluded.
After removing the last state, empty-map all() checks report the session as synced. The collector then remains healthy while archiving no symbols. Fail the session before creating another segment.
Proposed fix
ProcessAction::Excluded => {
+ if states.is_empty() {
+ failure = Some(anyhow::anyhow!(
+ "no active symbols remain after runtime exclusion"
+ ));
+ break;
+ }
segment =
rotate_segment(segment, &config, &states, &session_id, "symbol_excluded")?;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| ProcessAction::Excluded => { | |
| segment = | |
| rotate_segment(segment, &config, &states, &session_id, "symbol_excluded")?; | |
| } | |
| ProcessAction::Excluded => { | |
| if states.is_empty() { | |
| failure = Some(anyhow::anyhow!( | |
| "no active symbols remain after runtime exclusion" | |
| )); | |
| break; | |
| } | |
| segment = | |
| rotate_segment(segment, &config, &states, &session_id, "symbol_excluded")?; | |
| } |
🤖 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 466 -
469, Update the ProcessAction::Excluded handling around rotate_segment to detect
when removing the final symbol leaves no states. Fail or terminate the session
before creating another segment, rather than allowing empty-map all() checks to
mark it synced; preserve normal segment rotation when other symbols remain.
| Event::ExcludeSymbol { symbol, reason } => segment.write( | ||
| "symbol_excluded", | ||
| json!({ | ||
| "session_id":session_id, | ||
| "archived_only":true, | ||
| "symbol":symbol, | ||
| "reason":reason, | ||
| }), | ||
| now_ns()?, | ||
| ), |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Apply archived shutdown exclusions to the catalog.
When send_or_shutdown returns an ExcludeSymbol, archive_only records the event but never calls config.exclude_symbol. The final manifest therefore still lists the symbol as active despite containing a symbol_excluded event.
Pass Config into archive_only, call exclude_symbol for this variant, and add a shutdown-path regression test.
🤖 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 698 -
707, Update archive_only and its callers to accept Config, and in the
Event::ExcludeSymbol branch call config.exclude_symbol for the excluded symbol
while preserving the existing symbol_excluded event recording. Add a regression
test covering the shutdown path that verifies the catalog/manifest no longer
lists the excluded symbol as active.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7dfd222c19
ℹ️ 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".
| silent_ms = now_ms.saturating_sub(last_ms), | ||
| "process watchdog exiting after market-data stall" | ||
| ); | ||
| std::process::exit(75); |
There was a problem hiding this comment.
When the watchdog fires, for example if the Tokio loop is blocked or no frames arrive for PROCESS_WATCHDOG_SECONDS, std::process::exit terminates without running destructors, so the current Segment's 1 MiB BufWriter is never flushed or closed and recover_parts can only finalize bytes already in the .part file. That can silently drop the latest market-data events; use a graceful shutdown/flush path or periodic segment flushes before relying on a hard process exit.
Useful? React with 👍 / 👎.
| ProcessAction::None => {} | ||
| ProcessAction::Resnapshot(symbol) => { | ||
| segment = rotate_segment(segment, &config, &states, &session_id, "sequence_gap")?; | ||
| sync_deadline = Some(Instant::now() + config.sync_timeout); |
There was a problem hiding this comment.
For a sequence gap, this arms sync_deadline before produce_snapshot has fetched the replacement snapshot. If REST is rate-limited and fetch_snapshot honors Retry-After or exponential backoff longer than SYNC_TIMEOUT_SECONDS, the main loop can hit the bridge timeout while the snapshot task is still waiting, aborting the configured retries and reconnecting repeatedly; initial sync starts the deadline after the last snapshot completes, and resync should use the same ordering or a separate request timeout.
Useful? React with 👍 / 👎.
| }), | ||
| received_at_ns, | ||
| ), | ||
| Event::ExcludeSymbol { symbol, reason } => segment.write( |
There was a problem hiding this comment.
When a session is being torn down after another task failure, a snapshot task can return Stopped(Some(Event::ExcludeSymbol)); this branch only writes the event as archived data and never updates config.excluded_symbols or removes the state. The next reconnect will therefore include the same stale symbol again even though this session already discovered it was unavailable, causing repeated failed snapshot cycles and manifests that contain symbol_excluded events without matching excluded_symbols metadata.
Useful? React with 👍 / 👎.
| ProcessAction::Excluded => { | ||
| segment = | ||
| rotate_segment(segment, &config, &states, &session_id, "symbol_excluded")?; |
There was a problem hiding this comment.
If the last active symbol is excluded at runtime, this branch only rotates the segment and the loop continues with an empty states map. Because states.values().all(...) is true for an empty map, health can report synced with symbol_count=0 until a websocket stall or reconnect happens, instead of taking the existing no active symbols remain failure path immediately.
Useful? React with 👍 / 👎.
| segment.update_catalog( | ||
| catalog.symbols, | ||
| catalog.security_token_symbols, | ||
| catalog.excluded_symbols, |
There was a problem hiding this comment.
When a symbol is excluded after this segment has already recorded diffs or snapshots for it, config has already been mutated before close_segment runs, so this overwrites the open segment's manifest with the post-exclusion active symbol list. The resulting .manifest.json can omit a symbol whose events are still present in the .jsonl.zst, causing downstream readers that select segments by manifest symbols to miss data; keep the segment's original catalog or rotate before applying the exclusion to the catalog used for the closing segment.
Useful? React with 👍 / 👎.
| if states.values().all(|state| state.synced && state.bridged) { | ||
| sync_deadline = None; | ||
| } else if sync_timed_out(&states, sync_deadline, Instant::now()) { |
There was a problem hiding this comment.
With SYMBOLS=ALL, an active but illiquid Binance symbol can legitimately receive no depth update for more than SYNC_TIMEOUT_SECONDS after its snapshot; install_snapshot sets synced=true but bridged=false until a later diff arrives, so this timeout treats quiet markets as failed bridges and reconnects forever even though no sequence gap occurred. This undermines full-catalog collection; the deadline should not require every symbol to bridge on live traffic within a fixed wall-clock window.
Useful? React with 👍 / 👎.
Summary
Verification
Production Python collectors remain active. Rust shadow is paused pending this PR, CI, republish, and repeat shadow verification.
Summary by CodeRabbit
New Features
Bug Fixes
/datamount is available.Documentation