perf: parse WebSocket ingress with sonic-rs

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,13 +105,20 @@ 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 }
 
 # 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

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::<usize>();
+
+    // Validate compatibility once, outside all timed sections.
+    for payload in &messages {
+        let sonic = sonic_rs::from_slice::<Value>(payload).expect("sonic-rs rejected corpus line");
+        let serde =
+            serde_json::from_slice::<Value>(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::<Value>(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::<Value>(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::<Value>(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::<Value>(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::<Value>(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::<Value>(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::<Value>(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::<Value>(black_box(payload)).unwrap()),
+    );
+}
+
+fn load_ndjson(path: &Path) -> Vec<Vec<u8>> {
+    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);

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::<Vec<_>>();
+    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::<usize>();
+    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::<Value>(payload).unwrap());
+            }
+        })
+    });
+    group.bench_function("serde_json", |b| {
+        b.iter(|| {
+            for payload in black_box(&messages) {
+                black_box(serde_json::from_slice::<Value>(payload).unwrap());
+            }
+        })
+    });
+    group.finish();
+}
+
+async fn capture(stream_path: &str, target: usize, timeout_secs: u64) -> Vec<Vec<u8>> {
+    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<u8>]) {
+    assert!(!messages.is_empty(), "no messages captured");
+    for (index, payload) in messages.iter().enumerate() {
+        let sonic = sonic_rs::from_slice::<Value>(payload)
+            .unwrap_or_else(|error| panic!("sonic-rs rejected message {index}: {error}"));
+        let serde = serde_json::from_slice::<Value>(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::<Vec<_>>()
+        .join("/");
+    format!("/stream?streams={streams}")
+}
+
+fn symbols_from_env() -> Vec<String> {
+    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::<Vec<_>>();
+    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);