diff --git a/Cargo.toml b/Cargo.toml index d279398..8152f59 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,10 +29,13 @@ async-trait = "0.1" # HTTP & WebSocket reqwest = { version = "0.13", default-features = false, features = ["json", "rustls", "gzip", "stream"] } tokio-tungstenite = { version = "0.28", default-features = false, features = ["rustls-tls-native-roots", "connect"] } +ratatui = "0.30.2" +crossterm = { version = "0.29.0", features = ["event-stream"] } # Serialization serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +sonic-rs = "0.5.8" # Error handling thiserror = "2.0.17" @@ -102,6 +105,8 @@ ccxt-exchanges = { path = "ccxt-exchanges", version = "0.1.5" } # Async runtime tokio = { workspace = true } tokio-util = { workspace = true } +futures-util = { workspace = true } +tokio-tungstenite = { workspace = true } # Error handling (for examples) anyhow = { workspace = true } @@ -109,6 +114,11 @@ anyhow = { workspace = true } # Serialization serde = { workspace = true } serde_json = { workspace = true } +sonic-rs = { workspace = true } + +# Terminal benchmark UI +ratatui = { workspace = true } +crossterm = { workspace = true } # Decimal precision rust_decimal = { workspace = true } @@ -181,6 +191,10 @@ inherits = "release" # name = "orderbook" # harness = false -# [[bench]] -# name = "parsing" -# harness = false +[[bench]] +name = "binance_ingress" +harness = false + +[[bench]] +name = "binance_kline_1s_top10" +harness = false diff --git a/benches/binance_ingress.rs b/benches/binance_ingress.rs new file mode 100644 index 0000000..a869b0f --- /dev/null +++ b/benches/binance_ingress.rs @@ -0,0 +1,185 @@ +use std::{env, fs, hint::black_box, path::Path}; + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use serde_json::Value; + +const MAX_DEPTH: usize = 127; +const AGG_TRADE: &[u8] = br#"{"stream":"btcusdt@aggTrade","data":{"e":"aggTrade","E":1672515782136,"s":"BTCUSDT","a":12345,"p":"30000.1","q":"0.01","f":100,"l":105,"T":1672515782136,"m":true}}"#; +const KLINE_1M: &[u8] = br#"{"stream":"btcusdt@kline_1m","data":{"e":"kline","E":1672515782136,"s":"BTCUSDT","k":{"t":1672515780000,"T":1672515839999,"s":"BTCUSDT","i":"1m","f":100,"L":105,"o":"30000.0","c":"30001.0","h":"30002.0","l":"29999.0","v":"1.5","n":6,"x":false,"q":"45000.0","V":"0.8","Q":"24000.0","B":"0"}}}"#; + +fn bench_fixtures(c: &mut Criterion) { + let mut group = c.benchmark_group("binance_ingress_fixtures"); + group.sample_size(100); + for (name, payload) in [("aggTrade", AGG_TRADE), ("kline_1m", KLINE_1M)] { + group.throughput(Throughput::Bytes(payload.len() as u64)); + add_parser_benches(&mut group, name, payload); + } + group.finish(); +} + +fn bench_corpus(c: &mut Criterion) { + let Ok(path) = env::var("BINANCE_CORPUS") else { + eprintln!("BINANCE_CORPUS is unset; skipping NDJSON corpus replay"); + return; + }; + let messages = load_ndjson(Path::new(&path)); + assert!( + !messages.is_empty(), + "BINANCE_CORPUS contains no JSON messages" + ); + let bytes = messages.iter().map(Vec::len).sum::(); + + // Validate compatibility once, outside all timed sections. + for payload in &messages { + let sonic = sonic_rs::from_slice::(payload).expect("sonic-rs rejected corpus line"); + let serde = + serde_json::from_slice::(payload).expect("serde_json rejected corpus line"); + assert_eq!(sonic, serde, "parser value mismatch in corpus"); + } + + let mut group = c.benchmark_group("binance_ingress_corpus"); + group.sample_size(10); + group.throughput(Throughput::Bytes(bytes as u64)); + group.bench_function("sonic-native-limit-255", |b| { + b.iter(|| { + for payload in black_box(&messages) { + black_box(sonic_rs::from_slice::(payload).unwrap()); + } + }) + }); + group.bench_function("sonic-pre-scan-limit-127", |b| { + b.iter(|| { + for payload in black_box(&messages) { + validate_ingress_depth(payload).unwrap(); + black_box(sonic_rs::from_slice::(payload).unwrap()); + } + }) + }); + group.bench_function("sonic-post-dom-limit-127", |b| { + b.iter(|| { + for payload in black_box(&messages) { + let value = sonic_rs::from_slice::(payload).unwrap(); + validate_value_depth(&value).unwrap(); + black_box(value); + } + }) + }); + group.bench_function("serde_json", |b| { + b.iter(|| { + for payload in black_box(&messages) { + black_box(serde_json::from_slice::(payload).unwrap()); + } + }) + }); + group.finish(); +} + +fn add_parser_benches( + group: &mut criterion::BenchmarkGroup<'_, criterion::measurement::WallTime>, + name: &str, + payload: &[u8], +) { + group.bench_with_input( + BenchmarkId::new("sonic-native-limit-255", name), + payload, + |b, payload| b.iter(|| sonic_rs::from_slice::(black_box(payload)).unwrap()), + ); + group.bench_with_input( + BenchmarkId::new("sonic-pre-scan-limit-127", name), + payload, + |b, payload| { + b.iter(|| { + validate_ingress_depth(black_box(payload)).unwrap(); + sonic_rs::from_slice::(payload).unwrap() + }) + }, + ); + group.bench_with_input( + BenchmarkId::new("sonic-post-dom-limit-127", name), + payload, + |b, payload| { + b.iter(|| { + let value = sonic_rs::from_slice::(black_box(payload)).unwrap(); + validate_value_depth(&value).unwrap(); + value + }) + }, + ); + group.bench_with_input( + BenchmarkId::new("serde_json", name), + payload, + |b, payload| b.iter(|| serde_json::from_slice::(black_box(payload)).unwrap()), + ); +} + +fn load_ndjson(path: &Path) -> Vec> { + let data = + fs::read(path).unwrap_or_else(|error| panic!("failed to read {}: {error}", path.display())); + data.split(|byte| *byte == b'\n') + .map(|line| line.strip_suffix(b"\r").unwrap_or(line)) + .filter(|line| !line.iter().all(u8::is_ascii_whitespace)) + .map(<[u8]>::to_vec) + .collect() +} + +// Byte-for-byte equivalent to WsClient::validate_ingress_json_depth. +fn validate_ingress_depth(payload: &[u8]) -> Result<(), ()> { + let mut depth = 0_usize; + let mut in_string = false; + let mut escaped = false; + for &byte in payload { + if in_string { + if escaped { + escaped = false; + } else { + match byte { + b'\\' => escaped = true, + b'"' => in_string = false, + _ => {} + } + } + continue; + } + match byte { + b'"' => in_string = true, + b'{' | b'[' => { + depth += 1; + if depth > MAX_DEPTH { + return Err(()); + } + } + b'}' | b']' => depth = depth.saturating_sub(1), + _ => {} + } + } + Ok(()) +} + +// Relies on sonic-rs's built-in 255-level parse limit, then enforces the +// fork's stricter 127-level policy iteratively without rescanning JSON text. +fn validate_value_depth(root: &Value) -> Result<(), ()> { + let mut stack = vec![(root, 0_usize)]; + while let Some((value, depth)) = stack.pop() { + match value { + Value::Array(values) => { + let child_depth = depth + 1; + if child_depth > MAX_DEPTH { + return Err(()); + } + stack.extend(values.iter().map(|value| (value, child_depth))); + } + Value::Object(values) => { + let child_depth = depth + 1; + if child_depth > MAX_DEPTH { + return Err(()); + } + stack.extend(values.values().map(|value| (value, child_depth))); + } + _ => {} + } + } + Ok(()) +} + +criterion_group!(benches, bench_fixtures, bench_corpus); +criterion_main!(benches); diff --git a/benches/binance_kline_1s_top10.rs b/benches/binance_kline_1s_top10.rs new file mode 100644 index 0000000..4f9185c --- /dev/null +++ b/benches/binance_kline_1s_top10.rs @@ -0,0 +1,161 @@ +use std::{env, fs, hint::black_box, path::PathBuf, time::Duration}; + +use criterion::{Criterion, Throughput, criterion_group, criterion_main}; +use futures_util::StreamExt; +use serde_json::Value; +use tokio::time::{Instant, timeout_at}; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +const DEFAULT_SYMBOLS: &str = + "btcusdt,ethusdt,solusdt,bnbusdt,xrpusdt,dogeusdt,adausdt,trxusdt,avaxusdt,linkusdt"; +const DEFAULT_MESSAGES: usize = 600; +const DEFAULT_CAPTURE_TIMEOUT_SECS: u64 = 180; + +fn bench_top10_kline_1s(c: &mut Criterion) { + let symbols = symbols_from_env(); + let target_messages = env_usize("KLINE_CAPTURE_MESSAGES", DEFAULT_MESSAGES); + let timeout_secs = env_u64("KLINE_CAPTURE_TIMEOUT_SECS", DEFAULT_CAPTURE_TIMEOUT_SECS); + let stream_path = combined_stream_path(&symbols); + + eprintln!("capturing {target_messages} Binance kline_1s messages"); + eprintln!("symbols: {}", symbols.join(",")); + let runtime = tokio::runtime::Runtime::new().expect("failed to create Tokio runtime"); + let messages = runtime.block_on(capture(&stream_path, target_messages, timeout_secs)); + let corpus_path = env::var_os("BINANCE_KLINE_CORPUS") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("target/binance_kline_1s_top10.ndjson")); + let corpus = messages + .iter() + .flat_map(|message| message.iter().copied().chain(std::iter::once(b'\n'))) + .collect::>(); + fs::write(&corpus_path, corpus) + .unwrap_or_else(|error| panic!("failed to write {}: {error}", corpus_path.display())); + let total_bytes = messages.iter().map(Vec::len).sum::(); + eprintln!( + "captured {} messages ({} bytes); starting replay benchmark", + messages.len(), + total_bytes + ); + eprintln!("saved corpus: {}", corpus_path.display()); + + validate_corpus(&messages); + + let mut group = c.benchmark_group("binance_kline_1s_top10"); + group.sample_size(20); + group.throughput(Throughput::Bytes(total_bytes as u64)); + group.bench_function("sonic-rs", |b| { + b.iter(|| { + for payload in black_box(&messages) { + black_box(sonic_rs::from_slice::(payload).unwrap()); + } + }) + }); + group.bench_function("serde_json", |b| { + b.iter(|| { + for payload in black_box(&messages) { + black_box(serde_json::from_slice::(payload).unwrap()); + } + }) + }); + group.finish(); +} + +async fn capture(stream_path: &str, target: usize, timeout_secs: u64) -> Vec> { + let endpoint = env::var("BINANCE_WS_ENDPOINT") + .unwrap_or_else(|_| "wss://data-stream.binance.vision".into()); + let url = format!("{}{stream_path}", endpoint.trim_end_matches('/')); + eprintln!("endpoint: {url}"); + + let (mut socket, _) = timeout_at( + Instant::now() + Duration::from_secs(20), + connect_async(&url), + ) + .await + .unwrap_or_else(|_| { + panic!( + "timed out connecting to {endpoint}; this host is unreachable from the current network" + ) + }) + .unwrap_or_else(|error| panic!("failed to connect to {endpoint}: {error}")); + let deadline = Instant::now() + Duration::from_secs(timeout_secs); + let mut messages = Vec::with_capacity(target); + + while messages.len() < target { + let message = timeout_at(deadline, socket.next()) + .await + .unwrap_or_else(|_| { + panic!( + "capture timed out after {timeout_secs}s with {}/{} messages", + messages.len(), + target + ) + }) + .expect("Binance closed the stream") + .expect("Binance WebSocket read failed"); + match message { + Message::Text(text) => messages.push(text.as_bytes().to_vec()), + Message::Binary(data) => messages.push(data.to_vec()), + Message::Close(frame) => panic!("Binance closed the stream: {frame:?}"), + _ => {} + } + } + messages +} + +fn validate_corpus(messages: &[Vec]) { + assert!(!messages.is_empty(), "no messages captured"); + for (index, payload) in messages.iter().enumerate() { + let sonic = sonic_rs::from_slice::(payload) + .unwrap_or_else(|error| panic!("sonic-rs rejected message {index}: {error}")); + let serde = serde_json::from_slice::(payload) + .unwrap_or_else(|error| panic!("serde_json rejected message {index}: {error}")); + assert_eq!(sonic, serde, "parser mismatch at message {index}"); + assert_eq!(sonic["data"]["e"], "kline", "non-kline message at {index}"); + assert_eq!(sonic["data"]["k"]["i"], "1s", "non-1s kline at {index}"); + } +} + +fn combined_stream_path(symbols: &[String]) -> String { + let streams = symbols + .iter() + .map(|symbol| format!("{symbol}@kline_1s")) + .collect::>() + .join("/"); + format!("/stream?streams={streams}") +} + +fn symbols_from_env() -> Vec { + let raw = env::var("BINANCE_SYMBOLS").unwrap_or_else(|_| DEFAULT_SYMBOLS.into()); + let symbols = raw + .split(',') + .map(|symbol| symbol.trim().to_ascii_lowercase()) + .filter(|symbol| !symbol.is_empty()) + .collect::>(); + assert_eq!( + symbols.len(), + 10, + "BINANCE_SYMBOLS must contain exactly 10 comma-separated symbols" + ); + assert!( + symbols + .iter() + .all(|symbol| symbol.chars().all(|c| c.is_ascii_alphanumeric())), + "symbols may contain only ASCII letters and digits" + ); + symbols +} + +fn env_usize(name: &str, default: usize) -> usize { + env::var(name) + .map(|value| value.parse().unwrap_or_else(|_| panic!("invalid {name}"))) + .unwrap_or(default) +} + +fn env_u64(name: &str, default: u64) -> u64 { + env::var(name) + .map(|value| value.parse().unwrap_or_else(|_| panic!("invalid {name}"))) + .unwrap_or(default) +} + +criterion_group!(benches, bench_top10_kline_1s); +criterion_main!(benches); diff --git a/ccxt-core/Cargo.toml b/ccxt-core/Cargo.toml index 09f2911..c8cb983 100644 --- a/ccxt-core/Cargo.toml +++ b/ccxt-core/Cargo.toml @@ -24,6 +24,7 @@ tokio-tungstenite = { workspace = true, default-features = false, optional = tru # Serialization serde = { workspace = true } serde_json = { workspace = true } +sonic-rs = { workspace = true, optional = true } # Error handling thiserror = { workspace = true } @@ -83,7 +84,11 @@ proptest = { workspace = true } criterion = { workspace = true } [features] -default = ["rustls-tls", "websocket"] +default = ["rustls-tls", "websocket", "sonic-ingress"] + +# Inbound WebSocket JSON parser (mutually exclusive) +sonic-ingress = ["dep:sonic-rs"] +serde-ingress = [] # TLS implementations (mutually exclusive, choose one) rustls-tls = [ @@ -115,3 +120,7 @@ backtrace = [] [[bench]] name = "timestamp_benchmark" harness = false + +[[bench]] +name = "binance_kline_ingress" +harness = false diff --git a/ccxt-core/benches/binance_kline_ingress.rs b/ccxt-core/benches/binance_kline_ingress.rs new file mode 100644 index 0000000..9e76984 --- /dev/null +++ b/ccxt-core/benches/binance_kline_ingress.rs @@ -0,0 +1,47 @@ +use std::{env, fs, hint::black_box, path::PathBuf}; + +use ccxt_core::ws_client::WsClient; +use criterion::{Criterion, Throughput, criterion_group, criterion_main}; + +fn bench_ingress(c: &mut Criterion) { + let path = env::var_os("BINANCE_KLINE_CORPUS") + .map(PathBuf::from) + .expect("set BINANCE_KLINE_CORPUS to an NDJSON capture"); + let data = fs::read(&path) + .unwrap_or_else(|error| panic!("failed to read {}: {error}", path.display())); + let messages = data + .split(|byte| *byte == b'\n') + .map(|line| line.strip_suffix(b"\r").unwrap_or(line)) + .filter(|line| !line.iter().all(u8::is_ascii_whitespace)) + .collect::>(); + assert!(!messages.is_empty(), "corpus contains no messages"); + for payload in &messages { + WsClient::parse_ingress_json(payload).expect("fork ingress rejected corpus payload"); + } + + let total_bytes = messages.iter().map(|payload| payload.len()).sum::(); + let backend = if cfg!(feature = "sonic-ingress") { + "sonic-ingress" + } else { + "serde-ingress" + }; + eprintln!( + "backend={backend}, messages={}, bytes={total_bytes}", + messages.len() + ); + + let mut group = c.benchmark_group("ccxt_core_wsclient_kline_ingress"); + group.sample_size(20); + group.throughput(Throughput::Bytes(total_bytes as u64)); + group.bench_function(backend, |b| { + b.iter(|| { + for payload in black_box(&messages) { + black_box(WsClient::parse_ingress_json(payload).unwrap()); + } + }) + }); + group.finish(); +} + +criterion_group!(benches, bench_ingress); +criterion_main!(benches); diff --git a/ccxt-core/src/ws_client/mod.rs b/ccxt-core/src/ws_client/mod.rs index 43beda2..fad1058 100644 --- a/ccxt-core/src/ws_client/mod.rs +++ b/ccxt-core/src/ws_client/mod.rs @@ -29,7 +29,7 @@ use futures_util::{SinkExt, StreamExt, stream::SplitSink}; use serde_json::Value; use std::collections::HashMap; use std::sync::Arc; -use std::sync::atomic::{AtomicU8, AtomicU32, Ordering}; +use std::sync::atomic::{AtomicU8, AtomicU32, AtomicU64, Ordering}; use tokio::net::TcpStream; use tokio::sync::{Mutex, RwLock, mpsc}; use tokio::time::{Duration, interval}; @@ -40,10 +40,35 @@ use tokio_tungstenite::{ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, instrument, warn}; +#[cfg(all(feature = "sonic-ingress", feature = "serde-ingress"))] +compile_error!("features `sonic-ingress` and `serde-ingress` are mutually exclusive"); +#[cfg(not(any(feature = "sonic-ingress", feature = "serde-ingress")))] +compile_error!("enable exactly one of `sonic-ingress` or `serde-ingress`"); + /// Type alias for WebSocket write half. #[allow(dead_code)] type WsWriter = SplitSink>, Message>; +// Preserve serde_json's effective default recursion ceiling while protecting +// sonic-rs from deeply nested inputs on the WebSocket runtime thread. +const MAX_INGRESS_JSON_DEPTH: usize = 127; + +#[derive(Debug, thiserror::Error)] +/// Error returned when an inbound WebSocket JSON frame is rejected. +pub enum IngressJsonError { + /// The payload exceeds the fork's defensive nesting limit. + #[error("JSON nesting exceeds the {MAX_INGRESS_JSON_DEPTH}-level ingress limit")] + NestingLimitExceeded, + #[error(transparent)] + #[cfg(feature = "sonic-ingress")] + /// sonic-rs rejected the payload. + ParseSonic(#[from] sonic_rs::Error), + #[error(transparent)] + #[cfg(feature = "serde-ingress")] + /// serde_json rejected the payload. + ParseSerde(#[from] serde_json::Error), +} + /// Async WebSocket client for exchange streaming APIs. /// /// # Backpressure Handling @@ -71,6 +96,8 @@ pub struct WsClient { event_callback: Arc>>, /// Counter for dropped messages due to backpressure dropped_messages: Arc, + /// Counter for inbound payloads rejected by the JSON parser. + ingress_parse_errors: Arc, } impl WsClient { @@ -95,6 +122,7 @@ impl WsClient { cancel_token: Arc::new(Mutex::new(None)), event_callback: Arc::new(Mutex::new(None)), dropped_messages: Arc::new(AtomicU32::new(0)), + ingress_parse_errors: Arc::new(AtomicU64::new(0)), } } @@ -320,6 +348,7 @@ impl WsClient { self.subscription_manager.clear(); self.reconnect_count.store(0, Ordering::Release); self.dropped_messages.store(0, Ordering::Relaxed); + self.ingress_parse_errors.store(0, Ordering::Relaxed); self.stats.reset(); } @@ -460,6 +489,16 @@ impl WsClient { self.dropped_messages.store(0, Ordering::Relaxed); } + /// Returns the number of inbound WebSocket payloads rejected as invalid JSON. + pub fn ingress_parse_errors(&self) -> u64 { + self.ingress_parse_errors.load(Ordering::Relaxed) + } + + /// Resets the ingress JSON parse error counter. + pub fn reset_ingress_parse_errors(&self) { + self.ingress_parse_errors.store(0, Ordering::Relaxed); + } + /// Creates an automatic reconnection coordinator. pub fn create_auto_reconnect_coordinator(self: Arc) -> AutoReconnectCoordinator { AutoReconnectCoordinator::new(self) @@ -647,6 +686,7 @@ impl WsClient { let ping_interval_ms = self.config.ping_interval; let backpressure_strategy = self.config.backpressure_strategy; let dropped_messages = Arc::clone(&self.dropped_messages); + let ingress_parse_errors = Arc::clone(&self.ingress_parse_errors); let write_handle = tokio::spawn(async move { let mut write = write; @@ -673,29 +713,42 @@ impl WsClient { match msg_result { Ok(Message::Text(text)) => { ws_stats.record_received(text.len() as u64); - if let Ok(json) = serde_json::from_str::(&text) { - Self::send_with_backpressure( - &message_tx, - json, - backpressure_strategy, - &dropped_messages, - ) - .await; + match Self::parse_ingress_json(text.as_bytes()) { + Ok(json) => { + Self::send_with_backpressure( + &message_tx, + json, + backpressure_strategy, + &dropped_messages, + ) + .await; + } + Err(parse_error) => Self::record_ingress_parse_error( + &ingress_parse_errors, + "text", + text.len(), + &parse_error, + ), } } Ok(Message::Binary(data)) => { ws_stats.record_received(data.len() as u64); - if let Some(json) = String::from_utf8(data.to_vec()) - .ok() - .and_then(|text| serde_json::from_str::(&text).ok()) - { - Self::send_with_backpressure( - &message_tx, - json, - backpressure_strategy, - &dropped_messages, - ) - .await; + match Self::parse_ingress_json(data.as_ref()) { + Ok(json) => { + Self::send_with_backpressure( + &message_tx, + json, + backpressure_strategy, + &dropped_messages, + ) + .await; + } + Err(parse_error) => Self::record_ingress_parse_error( + &ingress_parse_errors, + "binary", + data.len(), + &parse_error, + ), } } Ok(Message::Pong(_)) => { @@ -769,6 +822,70 @@ impl WsClient { }); } + /// Parses an inbound WebSocket payload using the selected ingress backend. + #[inline] + pub fn parse_ingress_json(payload: &[u8]) -> std::result::Result { + Self::validate_ingress_json_depth(payload)?; + #[cfg(feature = "sonic-ingress")] + return Ok(sonic_rs::from_slice(payload)?); + #[cfg(feature = "serde-ingress")] + return Ok(serde_json::from_slice(payload)?); + } + + #[inline] + fn validate_ingress_json_depth(payload: &[u8]) -> std::result::Result<(), IngressJsonError> { + let mut depth = 0_usize; + let mut in_string = false; + let mut escaped = false; + + for &byte in payload { + if in_string { + if escaped { + escaped = false; + } else { + match byte { + b'\\' => escaped = true, + b'"' => in_string = false, + _ => {} + } + } + continue; + } + + match byte { + b'"' => in_string = true, + b'{' | b'[' => { + depth += 1; + if depth > MAX_INGRESS_JSON_DEPTH { + return Err(IngressJsonError::NestingLimitExceeded); + } + } + b'}' | b']' => depth = depth.saturating_sub(1), + _ => {} + } + } + + Ok(()) + } + + fn record_ingress_parse_error( + counter: &AtomicU64, + frame_type: &'static str, + payload_bytes: usize, + parse_error: &IngressJsonError, + ) { + let count = counter.fetch_add(1, Ordering::Relaxed) + 1; + if count % 100 == 1 { + warn!( + error = %parse_error, + frame_type, + payload_bytes, + parse_error_count = count, + "Rejected invalid WebSocket JSON payload" + ); + } + } + /// Sends a message with backpressure handling. /// /// This method implements the configured backpressure strategy when the @@ -916,6 +1033,110 @@ mod tests { assert_eq!(key2, "trades"); } + #[test] + fn test_parse_ingress_json() { + let payload = br#"{ + "stream":"btcusdt@trade", + "data":{"price":"64123.45","quantity":2,"active":true,"note":"\u4ea4\u6613"} + }"#; + + let value = WsClient::parse_ingress_json(payload).expect("valid JSON should be parsed"); + + assert_eq!(value["stream"], "btcusdt@trade"); + assert_eq!(value["data"]["price"], "64123.45"); + assert_eq!(value["data"]["quantity"], 2); + assert_eq!(value["data"]["active"], true); + assert_eq!(value["data"]["note"], "交易"); + + let serde_value: Value = serde_json::from_slice(payload).unwrap(); + assert_eq!(value, serde_value); + } + + #[test] + fn test_parse_ingress_json_rejects_invalid_payloads() { + assert!(WsClient::parse_ingress_json(br#"{"incomplete":true"#).is_err()); + assert!(WsClient::parse_ingress_json(&[0xff, 0xfe, 0xfd]).is_err()); + + let deeply_nested = format!("{}0{}", "[".repeat(128), "]".repeat(128)); + assert!(serde_json::from_str::(&deeply_nested).is_err()); + assert!(matches!( + WsClient::parse_ingress_json(deeply_nested.as_bytes()), + Err(IngressJsonError::NestingLimitExceeded) + )); + + let brackets_in_string = format!(r#"{{"text":"{}"}}"#, "[".repeat(256)); + assert!(WsClient::parse_ingress_json(brackets_in_string.as_bytes()).is_ok()); + } + + #[test] + fn test_ingress_parse_error_counter() { + let client = WsClient::new(WsConfig::default()); + + WsClient::record_ingress_parse_error( + &client.ingress_parse_errors, + "text", + 1, + &WsClient::parse_ingress_json(b"{").unwrap_err(), + ); + + assert_eq!(client.ingress_parse_errors(), 1); + client.reset_ingress_parse_errors(); + assert_eq!(client.ingress_parse_errors(), 0); + } + + #[tokio::test] + async fn test_message_loop_recovers_after_invalid_text_and_binary_json() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut websocket = tokio_tungstenite::accept_async(stream).await.unwrap(); + + websocket + .send(Message::Text(r#"{"broken":true"#.into())) + .await + .unwrap(); + websocket + .send(Message::Binary(vec![0xff, 0xfe, 0xfd].into())) + .await + .unwrap(); + websocket + .send(Message::Text(r#"{"kind":"text","sequence":1}"#.into())) + .await + .unwrap(); + websocket + .send(Message::Binary( + br#"{"kind":"binary","sequence":2}"#.to_vec().into(), + )) + .await + .unwrap(); + }); + + let client = WsClient::new(WsConfig { + url: format!("ws://{address}"), + ping_interval: 0, + ..WsConfig::default() + }); + client.connect().await.unwrap(); + + let text = tokio::time::timeout(Duration::from_secs(1), client.receive()) + .await + .unwrap() + .unwrap(); + let binary = tokio::time::timeout(Duration::from_secs(1), client.receive()) + .await + .unwrap() + .unwrap(); + + assert_eq!(text, serde_json::json!({"kind": "text", "sequence": 1})); + assert_eq!(binary, serde_json::json!({"kind": "binary", "sequence": 2})); + assert_eq!(client.ingress_parse_errors(), 2); + + client.disconnect().await.unwrap(); + server.await.unwrap(); + } + #[tokio::test] async fn test_ws_client_creation() { let config = WsConfig { diff --git a/src/bin/binance_ws_bench.rs b/src/bin/binance_ws_bench.rs new file mode 100644 index 0000000..be166f1 --- /dev/null +++ b/src/bin/binance_ws_bench.rs @@ -0,0 +1,596 @@ +use std::{ + collections::VecDeque, + io::{self, stdout}, + sync::{Arc, Mutex}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; + +use anyhow::Result; +use crossterm::{ + event::{Event, EventStream, KeyCode, KeyEventKind}, + execute, + terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode}, +}; +use futures_util::{SinkExt, StreamExt}; +use ratatui::{ + Terminal, + backend::CrosstermBackend, + layout::{Constraint, Direction, Layout}, + style::{Color, Modifier, Style}, + text::{Line, Span}, + widgets::{Block, Borders, Cell, Paragraph, Row, Table, Wrap}, +}; +use serde_json::Value; +use tokio::time::MissedTickBehavior; +use tokio_tungstenite::{connect_async, tungstenite::Message}; +use tokio_util::sync::CancellationToken; + +const SAMPLE_CAPACITY: usize = 4096; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum Feed { + AggTrade, + Kline, +} + +impl Feed { + const ALL: [Self; 2] = [Self::AggTrade, Self::Kline]; + + fn label(self) -> &'static str { + match self { + Self::AggTrade => "aggTrade", + Self::Kline => "kline_1m", + } + } + + fn index(self) -> usize { + match self { + Self::AggTrade => 0, + Self::Kline => 1, + } + } +} + +#[derive(Default)] +struct Samples { + values: VecDeque, + total: u128, + count: u64, + last: u64, +} + +impl Samples { + fn push(&mut self, value: u64) { + self.last = value; + self.total += u128::from(value); + self.count += 1; + if self.values.len() == SAMPLE_CAPACITY { + self.values.pop_front(); + } + self.values.push_back(value); + } + + fn snapshot(&self) -> SampleView { + let mut values: Vec<_> = self.values.iter().copied().collect(); + values.sort_unstable(); + SampleView { + count: self.count, + last: self.last, + average: if self.count == 0 { + 0 + } else { + (self.total / u128::from(self.count)) as u64 + }, + p50: percentile(&values, 50), + p95: percentile(&values, 95), + p99: percentile(&values, 99), + } + } +} + +#[derive(Default)] +struct SampleView { + count: u64, + last: u64, + average: u64, + p50: u64, + p95: u64, + p99: u64, +} + +fn percentile(sorted: &[u64], percentile: usize) -> u64 { + if sorted.is_empty() { + return 0; + } + let index = ((sorted.len() - 1) * percentile).div_ceil(100); + sorted[index] +} + +struct AppState { + started: Instant, + status: String, + sonic: [Samples; 2], + serde: [Samples; 2], + event_latency_ms: [Samples; 2], + parser_errors: [u64; 2], + mismatches: u64, + messages: u64, + latest_trade: String, + latest_kline: String, +} + +impl Default for AppState { + fn default() -> Self { + Self { + started: Instant::now(), + status: "starting".into(), + sonic: Default::default(), + serde: Default::default(), + event_latency_ms: Default::default(), + parser_errors: [0; 2], + mismatches: 0, + messages: 0, + latest_trade: "waiting for aggTrade".into(), + latest_kline: "waiting for kline_1m".into(), + } + } +} + +struct TerminalGuard; + +impl TerminalGuard { + fn enter() -> Result { + enable_raw_mode()?; + execute!(stdout(), EnterAlternateScreen)?; + Ok(Self) + } +} + +impl Drop for TerminalGuard { + fn drop(&mut self) { + let _ = disable_raw_mode(); + let _ = execute!(stdout(), LeaveAlternateScreen); + } +} + +#[tokio::main] +async fn main() -> Result<()> { + let symbol = normalize_symbol(std::env::args().nth(1).as_deref().unwrap_or("btcusdt"))?; + let url = + format!("wss://stream.binance.com:9443/stream?streams={symbol}@aggTrade/{symbol}@kline_1m"); + let state = Arc::new(Mutex::new(AppState::default())); + let cancel = CancellationToken::new(); + let reader = tokio::spawn(websocket_loop(url.clone(), state.clone(), cancel.clone())); + + let guard = TerminalGuard::enter()?; + let mut terminal = Terminal::new(CrosstermBackend::new(stdout()))?; + terminal.clear()?; + let ui_result = run_ui(&mut terminal, &symbol, &url, &state, &cancel).await; + cancel.cancel(); + let _ = reader.await; + drop(guard); + ui_result +} + +fn normalize_symbol(raw: &str) -> Result { + let symbol = raw.replace(['/', '-', '_'], "").to_ascii_lowercase(); + anyhow::ensure!( + !symbol.is_empty() && symbol.chars().all(|c| c.is_ascii_alphanumeric()), + "symbol must contain only ASCII letters and digits" + ); + Ok(symbol) +} + +async fn run_ui( + terminal: &mut Terminal>, + symbol: &str, + url: &str, + state: &Arc>, + cancel: &CancellationToken, +) -> Result<()> { + let mut events = EventStream::new(); + let mut ticker = tokio::time::interval(Duration::from_millis(250)); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + + loop { + tokio::select! { + _ = ticker.tick() => { + let snapshot = snapshot(state); + terminal.draw(|frame| draw(frame, symbol, url, &snapshot))?; + } + event = events.next() => { + match event.transpose()? { + Some(Event::Key(key)) if key.kind == KeyEventKind::Press + && matches!(key.code, KeyCode::Char('q') | KeyCode::Esc) => break, + Some(_) => {} + None => break, + } + } + result = tokio::signal::ctrl_c() => { + result?; + break; + } + _ = cancel.cancelled() => break, + } + } + Ok(()) +} + +async fn websocket_loop(url: String, state: Arc>, cancel: CancellationToken) { + loop { + set_status(&state, "connecting"); + match connect_async(&url).await { + Ok((mut socket, _)) => { + set_status(&state, "connected"); + loop { + tokio::select! { + _ = cancel.cancelled() => { + let _ = socket.close(None).await; + return; + } + message = socket.next() => match message { + Some(Ok(Message::Text(text))) => benchmark_message(text.as_bytes(), &state), + Some(Ok(Message::Binary(bytes))) => benchmark_message(&bytes, &state), + Some(Ok(Message::Ping(payload))) => { + if socket.send(Message::Pong(payload)).await.is_err() { break; } + } + Some(Ok(Message::Close(_))) | None => break, + Some(Ok(_)) => {} + Some(Err(error)) => { + set_status(&state, &format!("socket error: {error}")); + break; + } + } + } + } + } + Err(error) => set_status(&state, &format!("connect error: {error}")), + } + + if cancel.is_cancelled() { + return; + } + set_status(&state, "reconnecting in 1s"); + tokio::select! { + _ = cancel.cancelled() => return, + _ = tokio::time::sleep(Duration::from_secs(1)) => {} + } + } +} + +fn benchmark_message(bytes: &[u8], state: &Arc>) { + let sequence = state.lock().unwrap_or_else(|e| e.into_inner()).messages; + + let (sonic, sonic_ns, serde, serde_ns) = if sequence.is_multiple_of(2) { + let (sonic, sonic_ns) = timed_sonic(bytes); + let (serde, serde_ns) = timed_serde(bytes); + (sonic, sonic_ns, serde, serde_ns) + } else { + let (serde, serde_ns) = timed_serde(bytes); + let (sonic, sonic_ns) = timed_sonic(bytes); + (sonic, sonic_ns, serde, serde_ns) + }; + + let canonical = sonic.as_ref().ok().or_else(|| serde.as_ref().ok()); + let feed = canonical.and_then(classify); + let mut app = state.lock().unwrap_or_else(|e| e.into_inner()); + app.messages += 1; + let Some(feed) = feed else { + app.parser_errors[0] += u64::from(sonic.is_err()); + app.parser_errors[1] += u64::from(serde.is_err()); + return; + }; + let index = feed.index(); + if sonic.is_ok() { + app.sonic[index].push(sonic_ns); + } else { + app.parser_errors[0] += 1; + } + if serde.is_ok() { + app.serde[index].push(serde_ns); + } else { + app.parser_errors[1] += 1; + } + if let (Ok(left), Ok(right)) = (&sonic, &serde) + && left != right + { + app.mismatches += 1; + } + if let Some(value) = canonical { + update_market_view(&mut app, feed, value); + } +} + +fn timed_sonic(bytes: &[u8]) -> (sonic_rs::Result, u64) { + let started = Instant::now(); + let result = sonic_rs::from_slice(bytes); + (result, nanos(started.elapsed())) +} + +fn timed_serde(bytes: &[u8]) -> (serde_json::Result, u64) { + let started = Instant::now(); + let result = serde_json::from_slice(bytes); + (result, nanos(started.elapsed())) +} + +fn nanos(duration: Duration) -> u64 { + duration.as_nanos().min(u128::from(u64::MAX)) as u64 +} + +fn classify(value: &Value) -> Option { + let event = value.get("data")?.get("e")?.as_str()?; + match event { + "aggTrade" => Some(Feed::AggTrade), + "kline" => Some(Feed::Kline), + _ => None, + } +} + +fn update_market_view(app: &mut AppState, feed: Feed, value: &Value) { + let data = &value["data"]; + if let Some(event_ms) = data.get("E").and_then(Value::as_u64) { + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64; + app.event_latency_ms[feed.index()].push(now_ms.saturating_sub(event_ms)); + } + match feed { + Feed::AggTrade => { + app.latest_trade = format!( + "price={} qty={} maker={} id={}", + text_field(data, "p"), + text_field(data, "q"), + text_field(data, "m"), + text_field(data, "a") + ); + } + Feed::Kline => { + let kline = &data["k"]; + app.latest_kline = format!( + "O={} H={} L={} C={} volume={} closed={}", + text_field(kline, "o"), + text_field(kline, "h"), + text_field(kline, "l"), + text_field(kline, "c"), + text_field(kline, "v"), + text_field(kline, "x") + ); + } + } +} + +fn text_field(value: &Value, key: &str) -> String { + value + .get(key) + .map(|v| { + v.as_str() + .map(str::to_owned) + .unwrap_or_else(|| v.to_string()) + }) + .unwrap_or_else(|| "-".into()) +} + +fn set_status(state: &Arc>, status: &str) { + state.lock().unwrap_or_else(|e| e.into_inner()).status = status.into(); +} + +struct AppSnapshot { + uptime: Duration, + status: String, + sonic: [SampleView; 2], + serde: [SampleView; 2], + latency: [SampleView; 2], + parser_errors: [u64; 2], + mismatches: u64, + messages: u64, + latest_trade: String, + latest_kline: String, +} + +fn snapshot(state: &Arc>) -> AppSnapshot { + let app = state.lock().unwrap_or_else(|e| e.into_inner()); + AppSnapshot { + uptime: app.started.elapsed(), + status: app.status.clone(), + sonic: [app.sonic[0].snapshot(), app.sonic[1].snapshot()], + serde: [app.serde[0].snapshot(), app.serde[1].snapshot()], + latency: [ + app.event_latency_ms[0].snapshot(), + app.event_latency_ms[1].snapshot(), + ], + parser_errors: app.parser_errors, + mismatches: app.mismatches, + messages: app.messages, + latest_trade: app.latest_trade.clone(), + latest_kline: app.latest_kline.clone(), + } +} + +fn draw(frame: &mut ratatui::Frame<'_>, symbol: &str, url: &str, app: &AppSnapshot) { + let areas = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Length(3), + Constraint::Length(9), + Constraint::Length(6), + Constraint::Min(5), + ]) + .split(frame.area()); + + let header = Paragraph::new(Line::from(vec![ + Span::styled( + format!(" {symbol} "), + Style::default() + .fg(Color::Cyan) + .add_modifier(Modifier::BOLD), + ), + Span::raw(format!( + "{} uptime {:02}:{:02} messages {} ({:.1}/s) ", + app.status, + app.uptime.as_secs() / 60, + app.uptime.as_secs() % 60, + app.messages, + app.messages as f64 / app.uptime.as_secs_f64().max(0.001), + )), + Span::styled("q/Esc quit", Style::default().fg(Color::Yellow)), + ])) + .block( + Block::default() + .borders(Borders::ALL) + .title("Binance WS parser benchmark"), + ); + frame.render_widget(header, areas[0]); + + let mut rows = Vec::new(); + for feed in Feed::ALL { + let i = feed.index(); + rows.push(metric_row( + feed.label(), + "sonic-rs", + &app.sonic[i], + Color::Green, + )); + rows.push(metric_row( + feed.label(), + "serde_json", + &app.serde[i], + Color::Blue, + )); + } + let parser_table = Table::new( + rows, + [ + Constraint::Length(11), + Constraint::Length(12), + Constraint::Length(10), + Constraint::Length(11), + Constraint::Length(11), + Constraint::Length(11), + Constraint::Length(11), + Constraint::Length(11), + ], + ) + .header( + Row::new([ + "feed", "parser", "count", "last", "mean", "p50", "p95", "p99", + ]) + .style(Style::default().add_modifier(Modifier::BOLD)), + ) + .block( + Block::default() + .borders(Borders::ALL) + .title("JSON parse duration (rolling 4096 samples)"), + ); + frame.render_widget(parser_table, areas[1]); + + let latency_rows = Feed::ALL.map(|feed| { + let view = &app.latency[feed.index()]; + Row::new([ + Cell::from(feed.label()), + Cell::from(view.count.to_string()), + Cell::from(format!("{} ms", view.last)), + Cell::from(format!("{} ms", view.p50)), + Cell::from(format!("{} ms", view.p95)), + Cell::from(format!("{} ms", view.p99)), + ]) + }); + let latency_table = Table::new( + latency_rows, + [ + Constraint::Length(12), + Constraint::Length(12), + Constraint::Length(12), + Constraint::Length(12), + Constraint::Length(12), + Constraint::Length(12), + ], + ) + .header( + Row::new(["feed", "count", "last", "p50", "p95", "p99"]) + .style(Style::default().add_modifier(Modifier::BOLD)), + ) + .block( + Block::default() + .borders(Borders::ALL) + .title("Binance event time → local receive time (clock-skew sensitive)"), + ); + frame.render_widget(latency_table, areas[2]); + + let details = Paragraph::new(vec![ + Line::from(format!("trade {}", app.latest_trade)), + Line::from(format!("kline {}", app.latest_kline)), + Line::from(format!( + "errors sonic={} serde={} value mismatches={} parser order alternates per message", + app.parser_errors[0], app.parser_errors[1], app.mismatches + )), + Line::from(format!("source {url}")), + ]) + .wrap(Wrap { trim: true }) + .block( + Block::default() + .borders(Borders::ALL) + .title("Live data / validity"), + ); + frame.render_widget(details, areas[3]); +} + +fn metric_row<'a>(feed: &'a str, parser: &'a str, view: &SampleView, color: Color) -> Row<'a> { + Row::new([ + Cell::from(feed), + Cell::from(parser).style(Style::default().fg(color)), + Cell::from(view.count.to_string()), + Cell::from(format_ns(view.last)), + Cell::from(format_ns(view.average)), + Cell::from(format_ns(view.p50)), + Cell::from(format_ns(view.p95)), + Cell::from(format_ns(view.p99)), + ]) +} + +fn format_ns(value: u64) -> String { + if value >= 1_000_000 { + format!("{:.2} ms", value as f64 / 1_000_000.0) + } else if value >= 1_000 { + format!("{:.2} us", value as f64 / 1_000.0) + } else { + format!("{value} ns") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn benchmarks_combined_agg_trade_payload_with_both_parsers() { + let state = Arc::new(Mutex::new(AppState::default())); + benchmark_message( + br#"{"stream":"btcusdt@aggTrade","data":{"e":"aggTrade","E":1672515782136,"s":"BTCUSDT","a":12345,"p":"30000.1","q":"0.01","f":100,"l":105,"T":1672515782136,"m":true}}"#, + &state, + ); + + let app = state.lock().unwrap(); + assert_eq!(app.sonic[Feed::AggTrade.index()].count, 1); + assert_eq!(app.serde[Feed::AggTrade.index()].count, 1); + assert_eq!(app.mismatches, 0); + assert_eq!(app.parser_errors, [0, 0]); + assert!(app.latest_trade.contains("30000.1")); + } + + #[test] + fn benchmarks_combined_one_minute_kline_payload_with_both_parsers() { + let state = Arc::new(Mutex::new(AppState::default())); + benchmark_message( + br#"{"stream":"btcusdt@kline_1m","data":{"e":"kline","E":1672515782136,"s":"BTCUSDT","k":{"t":1672515780000,"T":1672515839999,"s":"BTCUSDT","i":"1m","o":"30000.0","c":"30001.0","h":"30002.0","l":"29999.0","v":"1.5","x":false}}}"#, + &state, + ); + + let app = state.lock().unwrap(); + assert_eq!(app.sonic[Feed::Kline.index()].count, 1); + assert_eq!(app.serde[Feed::Kline.index()].count, 1); + assert_eq!(app.mismatches, 0); + assert_eq!(app.parser_errors, [0, 0]); + assert!(app.latest_kline.contains("C=30001.0")); + } +}