1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
#![cfg(not(target_arch = "wasm32"))]
//! Phase ι.4 — orderbook delta continuity / sequence-gap test.
//!
//! Validates that orderbook delta streams produce contiguous sequence numbers.
//! Gap detection uses `OrderbookDelta::last_update_id` (u64, Option).
//!
//! Per-exchange sequence field mapping:
//! Binance: last_update_id (u) — final update ID of delta message.
//! OKX: last_update_id — seq_id exposed as last_update_id in parser.
//! Bybit: last_update_id — update_id / u exposed as last_update_id in parser.
//!
//! If an exchange parser does not fill last_update_id, total > 0 is asserted only
//! and gap detection is skipped (documented as deferred in that exchange's parser).
//!
//! Run with: cargo test --test sequence_gap -- --ignored --nocapture
#[path = "common/mod.rs"]
mod common;
use common::{run_jobs, JobOutcome};
use digdigdig3::connector_manager::ExchangeHub;
use digdigdig3::{AccountType, ExchangeId, StreamEvent};
use futures_util::StreamExt;
use std::time::Duration;
use tokio::time::timeout;
/// (exchange, account_type, exchange-native symbol)
const TARGETS: &[(ExchangeId, AccountType, &str)] = &[
(ExchangeId::Binance, AccountType::Spot, "BTCUSDT"),
(ExchangeId::OKX, AccountType::Spot, "BTC-USDT"),
(ExchangeId::Bybit, AccountType::Spot, "BTCUSDT"),
];
#[tokio::test]
#[ignore] // live API — run with --ignored
async fn orderbook_delta_no_gaps() {
let labels: Vec<String> = TARGETS
.iter()
.map(|(id, _, _)| format!("{id:?}"))
.collect();
let results = run_jobs(
labels,
Duration::from_secs(45),
|_job_id, label| async move {
let (id, acct, sym_str) = TARGETS
.iter()
.find(|(id, _, _)| format!("{id:?}") == label)
.ok_or_else(|| format!("unknown target: {label}"))?;
let hub = ExchangeHub::new();
hub.connect_full(*id, &[*acct], false)
.await
.map_err(|e| e.to_string())?;
let ws = hub
.ws(*id, *acct)
.ok_or_else(|| format!("{label}: no ws handle"))?;
// Subscribe to orderbook delta stream
ws.subscribe(digdigdig3::SubscriptionRequest {
symbol: digdigdig3::Symbol::with_raw("", "", sym_str.to_string()),
stream_type: digdigdig3::StreamType::OrderbookDelta,
account_type: *acct,
depth: None,
update_speed_ms: None,
})
.await
.map_err(|e| format!("{label}: subscribe failed: {e}"))?;
let mut stream = ws.event_stream();
let mut last_seq: Option<u64> = None;
let mut gaps = 0u64;
let mut total = 0u64;
// Track whether last_update_id was ever populated
let mut seq_available = false;
let deadline = tokio::time::Instant::now() + Duration::from_secs(25);
while tokio::time::Instant::now() < deadline {
let remaining =
deadline.saturating_duration_since(tokio::time::Instant::now());
match timeout(remaining.max(Duration::from_millis(100)), stream.next()).await {
Ok(Some(Ok(StreamEvent::OrderbookDelta { delta: d, .. }))) => {
total += 1;
// last_update_id is the canonical sequence field populated by
// parsers for Binance (U/u fields), OKX (seqId), Bybit (u).
if let Some(seq) = d.last_update_id {
seq_available = true;
if let Some(prev) = last_seq {
// Allow +1 contiguity; allow skip if first_update_id > prev+1
// (some exchanges may legitimately skip on snapshot boundary).
let first = d.first_update_id.unwrap_or(seq);
if first > prev + 1 {
gaps += 1;
}
}
last_seq = Some(seq);
}
}
Ok(Some(Ok(_))) => {} // other event types, ignore
Ok(Some(Err(_))) | Ok(None) => {}
Err(_) => break,
}
}
Ok((total, gaps, seq_available))
},
)
.await;
println!("\n=== orderbook_delta_no_gaps ===");
for r in &results {
match &r.outcome {
JobOutcome::Ok((total, gaps, seq_available)) => {
if *seq_available {
let gap_pct = if *total > 0 { gaps * 100 / total } else { 0 };
println!(
"{:<15} job#{:>3} total={total} gaps={gaps} gap%={gap_pct}%",
r.label, r.job_id.0
);
assert!(
*total > 0,
"{}: zero orderbook deltas received in 25s",
r.label
);
assert!(
gap_pct < 5,
"{}: gap rate too high ({gaps}/{total} = {gap_pct}%)",
r.label
);
} else {
// Sequence field not populated by parser — assert events flow only.
// Gap detection deferred: parser does not fill last_update_id for this exchange.
println!(
"{:<15} job#{:>3} total={total} seq=N/A (gap detection deferred)",
r.label, r.job_id.0
);
assert!(
*total > 0,
"{}: zero orderbook deltas received in 25s (seq detection deferred)",
r.label
);
}
}
JobOutcome::TimedOut => {
panic!("{}: timed out waiting for orderbook deltas", r.label)
}
JobOutcome::Failed(e) => {
panic!("{}: {e}", r.label)
}
}
}
}