feat(collector): archive bookTicker/raw trade/forceOrder as tape v2 events - #545
Conversation
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
📝 WalkthroughWalkthroughThe Binance archiver now captures trade, bookTicker, and USD-M forceOrder streams. It archives validated events with market-tape schema v2 metadata, validates stream coverage and raw-trade sequences, tracks reconnect health, and preserves legacy and v1 recovery behavior. ChangesBinance market stream archival
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant BinanceWebSocket
participant event_from_frame
participant process_event
participant SegmentArchive
BinanceWebSocket->>event_from_frame: stream frame
event_from_frame->>process_event: parsed market event
process_event->>process_event: validate symbol, sequence, and market scope
process_event->>SegmentArchive: archive validated event
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 |
1adcaad to
3b3328e
Compare
Spot bookTicker payloads carry no e/E/T fields; pin that the archiver routes them through event_from_frame instead of failing the producer. Refs #536
be4e839 to
47a8888
Compare
There was a problem hiding this comment.
Actionable comments posted: 5
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)
4002-4017: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winMake
trusted_process_statemarket-aware.
trusted_process_statealways builds the four spot streams and never adds{symbol}@forceOrder``.book_ticker_and_force_order_are_archived_without_sequence_stateat Lines 4882-4919 sets `config.market = Market::Usdm`, so production would declare five stream types for that config and the coverage shard set would include `btcusdt@forceOrder`.The test passes only because it never calls
validate_stream_coverage_shards. The USD-M fixture therefore does not represent the state the collector reaches at runtime, and it will mask a real mismatch as soon as a test routes coverage validation through it.Take the market, or the stream-type list, as a parameter and append
forceOrderfor USD-M.🛠️ Proposed fix
- fn trusted_process_state(symbols: &[String]) -> ProcessState { + fn trusted_process_state_for(market: Market, symbols: &[String]) -> ProcessState { let mut state = ProcessState::new(true); + let mut stream_types = vec!["depth@100ms", "aggTrade", "trade", "bookTicker"]; + if market == Market::Usdm { + stream_types.push("forceOrder"); + } state.stream_coverage_shards = vec![symbols .iter() .flat_map(|symbol| { let symbol = symbol.to_ascii_lowercase(); - [ - format!("{symbol}`@depth`@100ms"), - format!("{symbol}`@aggTrade`"), - format!("{symbol}`@trade`"), - format!("{symbol}`@bookTicker`"), - ] + stream_types + .iter() + .map(move |stream_type| format!("{symbol}@{stream_type}")) + .collect::<Vec<_>>() }) .collect()]; state } + + fn trusted_process_state(symbols: &[String]) -> ProcessState { + trusted_process_state_for(Market::Spot, symbols) + }🤖 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 4002 - 4017, Update trusted_process_state to accept the configured market or stream-type list, and include the existing four spot streams plus {symbol}`@forceOrder` when the market is Market::Usdm. Update its callers, including book_ticker_and_force_order_are_archived_without_sequence_state, so the generated coverage shard state matches the runtime configuration.
🧹 Nitpick comments (3)
rust_hft/tools/collector/src/bin/binance-lob-archiver.rs (3)
305-336: 🚀 Performance & Scalability | 🔵 TrivialPlan for the increased websocket connection count.
Each symbol chunk now produces 4 shards for spot and 5 for USD-M, instead of 2. With
SYMBOLS=ALLand the defaultWS_SHARD_SIZE=100, the collector opens 2x to 2.5x more concurrent websocket connections and issues that many more connect attempts per reconnect storm. Binance enforces per-IP connection-attempt limits.Consider raising
WS_SHARD_SIZEfor the new high-rate stream types, or staggering the initial connect across shards. Track connect-rejection and reconnect counts per shard in the health file so a limit breach is visible.🤖 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 305 - 336, Update stream_shards to account for the increased number of streams per symbol by raising the effective WS_SHARD_SIZE or otherwise reducing concurrent shard creation, while preserving complete stream coverage. Also update the shard connection/reconnect handling to stagger initial connections and record per-shard connection rejections and reconnect counts in the health file.
3407-3412: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe declared stream-type catalog is duplicated across four literals.
Config::stream_typesis the authority for the declared catalog, but the same list is written out again in production support code and in three test fixtures. When a stream type is added or renamed, these literals drift silently and the affected tests keep passing against a stale catalog.
rust_hft/tools/collector/src/bin/binance-lob-archiver.rs#L3407-L3412: expose the market-keyed list as one reusable function next toConfig::stream_types, and call it here instead of the inline four-element literal.rust_hft/tools/collector/src/bin/binance-lob-archiver.rs#L4002-L4017: call the shared function with the fixture's market so USD-M fixtures includeforceOrder.rust_hft/tools/collector/src/lob_archiver.rs#L1234-L1239: buildstream_typesinrecovery_configfrom the shared function forMarket::Spot.rust_hft/tools/collector/src/lob_archiver.rs#L1555-L1560: buildstream_typesinsegment_declares_complete_market_tape_surfacefrom the same shared function.🤖 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 3407 - 3412, Eliminate the duplicated stream-type catalogs by exposing one market-keyed reusable function alongside Config::stream_types. In rust_hft/tools/collector/src/bin/binance-lob-archiver.rs lines 3407-3412, replace the inline list with the shared function; at lines 4002-4017, call it with the fixture market so USD-M includes forceOrder; in rust_hft/tools/collector/src/lob_archiver.rs lines 1234-1239 and 1555-1560, build stream_types through the same function for Market::Spot.
1347-1357: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winUse one event type for sequence gaps.
The raw-trade arm writes the event type
sequence_gapwith akindfield ofraw_trade_sequence, which matches the depth path at Lines 1266-1276. The aggregate-trade arm at Lines 1313-1322 writes a dedicated event typeaggregate_trade_gapinstead.The manifest
event_typesmap now mixes both conventions, so a consumer that counts gaps must readsequence_gapandaggregate_trade_gapand then inspectkind. Pick one convention. Thesequence_gappluskindform already covers depth and raw trades, so migrating the aggregate-trade arm to it removes the split.🤖 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 1347 - 1357, The aggregate-trade sequence-gap handling should use the existing unified event type convention. Update the aggregate-trade arm near the sequence-gap write to emit event type sequence_gap and include the appropriate aggregate-trade kind value, matching the depth and raw-trade paths; remove use of aggregate_trade_gap while preserving the existing gap payload.
🤖 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 933-941: Bound the derived queue capacity in the channel setup
around stream_shards and queue_capacity so it cannot grow indefinitely with
high-rate shard count. Add a configurable absolute maximum, use it to clamp the
sum of max_buffered_diffs and high-rate backlog, and preserve saturation-safe
arithmetic and existing behavior below the cap.
- Around line 1344-1359: Update the raw trade handling around
RawTradeSequenceValidator::observe so an observation error remains recorded via
segment.mark_replay_unsafe(), process_state.sequence_gaps, and the sequence_gap
event, but is not returned to run_session. Continue the capture loop after
logging the gap, while preserving normal error propagation for unrelated
failures.
In `@rust_hft/tools/collector/src/lob_archiver.rs`:
- Around line 1894-1902: Extend the test cases surrounding the RAW_SCHEMA
manifest assertions to cover v2 segments with an empty stream_types array and
with a duplicated stream type. Close each segment and assert the operation fails
with the exact error text “v2 market tape requires a declared stream type set,”
while preserving the existing valid v2 and v1 assertions.
- Line 27: Update the external schema gates in host-rust-lob-shadow-gate.sh at
the trade-summary and aggregate-row checks to accept binance.market_tape.v2
alongside binance.market_tape.v1. Preserve existing v1 compatibility while
allowing RAW_SCHEMA-generated v2 segments containing aggregate_trade_count.
- Around line 703-717: Move the v2 `config.stream_types` validation from the
metadata-writing block to the beginning of `finalize_segment`, before
compression, digesting, renaming, or any filesystem mutation. Keep the existing
validation conditions and error message unchanged, and leave only the valid v2
metadata insertion in the later manifest-generation path.
---
Outside diff comments:
In `@rust_hft/tools/collector/src/bin/binance-lob-archiver.rs`:
- Around line 4002-4017: Update trusted_process_state to accept the configured
market or stream-type list, and include the existing four spot streams plus
{symbol}`@forceOrder` when the market is Market::Usdm. Update its callers,
including book_ticker_and_force_order_are_archived_without_sequence_state, so
the generated coverage shard state matches the runtime configuration.
---
Nitpick comments:
In `@rust_hft/tools/collector/src/bin/binance-lob-archiver.rs`:
- Around line 305-336: Update stream_shards to account for the increased number
of streams per symbol by raising the effective WS_SHARD_SIZE or otherwise
reducing concurrent shard creation, while preserving complete stream coverage.
Also update the shard connection/reconnect handling to stagger initial
connections and record per-shard connection rejections and reconnect counts in
the health file.
- Around line 3407-3412: Eliminate the duplicated stream-type catalogs by
exposing one market-keyed reusable function alongside Config::stream_types. In
rust_hft/tools/collector/src/bin/binance-lob-archiver.rs lines 3407-3412,
replace the inline list with the shared function; at lines 4002-4017, call it
with the fixture market so USD-M includes forceOrder; in
rust_hft/tools/collector/src/lob_archiver.rs lines 1234-1239 and 1555-1560,
build stream_types through the same function for Market::Spot.
- Around line 1347-1357: The aggregate-trade sequence-gap handling should use
the existing unified event type convention. Update the aggregate-trade arm near
the sequence-gap write to emit event type sequence_gap and include the
appropriate aggregate-trade kind value, matching the depth and raw-trade paths;
remove use of aggregate_trade_gap while preserving the existing gap payload.
🪄 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: f163cb6a-69ed-414f-8a98-1bbebfaca1fa
📒 Files selected for processing (2)
rust_hft/tools/collector/src/bin/binance-lob-archiver.rsrust_hft/tools/collector/src/lob_archiver.rs
Change contract
binance-lob-archivercaptures<sym>@tradeand<sym>@bookTicker(spot + USD-M) and<sym>@forceOrder(USD-M only) alongside depth/aggTrade, writingbinance.market_tape.v2segments whose manifests and session_start rows declarestream_types. Raw trades get per-symbol id-continuity validation (gap → sequence_gap row + replay-unsafe); high-rate shards get a 4× reconnect-proof budget (65536/20s ≈ 3.2k msgs/s) and dedicated queue backlog so rotation barriers can't starve the consumer.Out of scope
trade_representation/price_surface_derivationmanifest fields — unchanged, the v2 verifier still pins the existing constantsDependency / merge order
Stacked on the tape-v2-schema PR (base branch
codex/tape-v2-schema; retarget to main after it merges). Must land before #537 — a v2-writing candidate cannot pass the old gate (fail-closed interlock, intended).Focused validation
cargo test -p hft-data --lockedgreenRollout / rollback impact
After merge + release, the next candidate gate must run with the #537 policies. Rollback = cut over back to a previously gated v1 digest (v1 binaries remain gateable under #537).
Issue relationship
Closes #536
Summary by CodeRabbit