From 73de01f4df425d2db173397ffc3ea1811c925cfc Mon Sep 17 00:00:00 2001 From: yangwao Date: Sun, 8 Mar 2026 20:35:18 +0100 Subject: [PATCH 01/29] fix: support --serve-info nodes by falling back to periodic ABCI state snapshots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `fileSnapshot` API endpoint is not available when the node is started with `--serve-info`, causing the order book server to crash with "No such file or directory" (issue #1). This adds a fallback: when the API returns an error, the server reads the latest `.rmp` file from `periodic_abci_states/` and uses `hl-node compute-l4-snapshots` to generate the L4 snapshot JSON. The `hl-node` binary path and chain can be configured via `HL_NODE_PATH` and `HL_CHAIN` environment variables. Additionally, snapshot fetch errors are now non-fatal — the listener logs the error and retries on the next tick instead of crashing. Co-Authored-By: Claude Opus 4.6 --- server/src/listeners/order_book/mod.rs | 9 +- server/src/listeners/order_book/utils.rs | 100 ++++++++++++++++++++++- 2 files changed, 107 insertions(+), 2 deletions(-) diff --git a/server/src/listeners/order_book/mod.rs b/server/src/listeners/order_book/mod.rs index fcd9425a..361e4897 100644 --- a/server/src/listeners/order_book/mod.rs +++ b/server/src/listeners/order_book/mod.rs @@ -112,7 +112,14 @@ pub(crate) async fn hl_listen(listener: Arc>, dir: Path return Err("Snapshot fetch task sender dropped".into()); } Some(Err(err)) => { - return Err(format!("Abci state reading error: {err}").into()); + let is_initialized = listener.lock().await.is_ready(); + if is_initialized { + // Already have a working order book — skip validation, don't crash + error!("Snapshot validation error (non-fatal): {err}"); + } else { + // Not yet initialized — log and wait for next tick to retry + error!("Snapshot fetch error (will retry): {err}"); + } } Some(Ok(())) => {} } diff --git a/server/src/listeners/order_book/utils.rs b/server/src/listeners/order_book/utils.rs index fdbd3b57..8c0893b1 100644 --- a/server/src/listeners/order_book/utils.rs +++ b/server/src/listeners/order_book/utils.rs @@ -11,6 +11,7 @@ use crate::{ node_data::{Batch, NodeDataFill, NodeDataOrderDiff, NodeDataOrderStatus}, }, }; +use log::info; use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; use reqwest::Client; use serde_json::json; @@ -19,9 +20,32 @@ use std::{ collections::HashMap, path::{Path, PathBuf}, }; +use tokio::fs; +/// Fetches an L4 snapshot and writes it to `out.json` in the given directory. +/// +/// First attempts the `fileSnapshot` API (available on full nodes). If that fails +/// (e.g. when the node is started with `--serve-info`), falls back to reading +/// the latest periodic ABCI state file and computing L4 snapshots via the +/// `hl-node compute-l4-snapshots` CLI command. +/// +/// The fallback requires `hl-node` to be available (on PATH or via `HL_NODE_PATH` +/// env var) and periodic ABCI state files to exist on disk. pub(super) async fn process_rmp_file(dir: &Path) -> Result { let output_path = dir.join("out.json"); + + match process_via_api(&output_path).await { + Ok(()) => return Ok(output_path), + Err(err) => { + info!("fileSnapshot API unavailable ({err}), trying periodic ABCI state fallback"); + } + } + + process_via_cli(dir, &output_path).await?; + Ok(output_path) +} + +async fn process_via_api(output_path: &Path) -> Result<()> { let payload = json!({ "type": "fileSnapshot", "request": { @@ -41,7 +65,81 @@ pub(super) async fn process_rmp_file(dir: &Path) -> Result { .send() .await? .error_for_status()?; - Ok(output_path) + Ok(()) +} + +async fn process_via_cli(dir: &Path, output_path: &Path) -> Result<()> { + let abci_states_dir = dir.join("hl/data/periodic_abci_states"); + let (rmp_path, height) = find_latest_rmp(&abci_states_dir).await?; + info!("Using periodic ABCI state at height {height}: {}", rmp_path.display()); + + let hl_node = std::env::var("HL_NODE_PATH").unwrap_or_else(|_| "hl-node".to_string()); + let chain = std::env::var("HL_CHAIN").unwrap_or_else(|_| "Mainnet".to_string()); + let raw_output = dir.join("l4_raw.json"); + + let result = tokio::process::Command::new(&hl_node) + .args(["--chain", &chain, "compute-l4-snapshots", "--include-users"]) + .arg(&rmp_path) + .arg(&raw_output) + .output() + .await + .map_err(|e| format!("Failed to run '{hl_node}': {e}. Set HL_NODE_PATH to the hl-node binary location."))?; + + if !result.status.success() { + let stderr = String::from_utf8_lossy(&result.stderr); + return Err(format!("hl-node compute-l4-snapshots failed: {stderr}").into()); + } + + // The CLI outputs [[coin, [bids, asks]], ...] without a height prefix. + // Wrap it as [height, data] to match the fileSnapshot API format. + let raw_content = fs::read_to_string(&raw_output).await?; + let wrapped = format!("[{height},{raw_content}]"); + fs::write(output_path, wrapped).await?; + + // Clean up temporary file + drop(fs::remove_file(&raw_output).await); + + Ok(()) +} + +async fn find_latest_rmp(abci_states_dir: &Path) -> Result<(PathBuf, u64)> { + let mut latest_date: Option = None; + let mut entries = fs::read_dir(abci_states_dir) + .await + .map_err(|e| format!("Cannot read periodic_abci_states directory: {e}"))?; + + while let Some(entry) = entries.next_entry().await? { + if entry.file_type().await?.is_dir() { + let name = entry.file_name().to_string_lossy().to_string(); + if latest_date.as_ref().is_none_or(|d| name > *d) { + latest_date = Some(name); + } + } + } + + let date_dir = latest_date.ok_or("No date directories found in periodic_abci_states")?; + let date_path = abci_states_dir.join(&date_dir); + + let mut latest_height: Option = None; + let mut latest_path: Option = None; + let mut entries = fs::read_dir(&date_path).await?; + + while let Some(entry) = entries.next_entry().await? { + let name = entry.file_name().to_string_lossy().to_string(); + if let Some(height_str) = name.strip_suffix(".rmp") { + if let Ok(height) = height_str.parse::() { + if latest_height.is_none_or(|h| height > h) { + latest_height = Some(height); + latest_path = Some(entry.path()); + } + } + } + } + + match (latest_path, latest_height) { + (Some(path), Some(height)) => Ok((path, height)), + _ => Err("No .rmp files found in periodic_abci_states".into()), + } } pub(super) fn validate_snapshot_consistency( From d6cebff081e1d1bd5b2afe38e1e3631114e07af0 Mon Sep 17 00:00:00 2001 From: Jonathan Udd Date: Fri, 5 Dec 2025 15:27:40 +0100 Subject: [PATCH 02/29] fix: panic in orderbook listener --- server/src/listeners/order_book/mod.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/server/src/listeners/order_book/mod.rs b/server/src/listeners/order_book/mod.rs index 361e4897..3f37789a 100644 --- a/server/src/listeners/order_book/mod.rs +++ b/server/src/listeners/order_book/mod.rs @@ -422,15 +422,20 @@ impl DirectoryListener for OrderBookListener { let (height, event_batch) = match res { Ok(data) => data, Err(err) => { + // Build a safe preview of the line (up to 100 *characters*). + let preview: String = line.chars().take(100).collect(); + // if we run into a serialization error (hitting EOF), just return to last line. error!( "{event_source} serialization error {err}, height: {:?}, line: {:?}", self.order_book_state.as_ref().map(OrderBookState::height), - &line[..100], + preview, ); #[allow(clippy::unwrap_used)] let total_len: i64 = total_len.try_into().unwrap(); - self.file_mut(event_source).as_mut().map(|f| f.seek_relative(-total_len)); + self.file_mut(event_source) + .as_mut() + .map(|f| f.seek_relative(-total_len)); break; } }; From 0dc7f7743c777caab7d92b15d85878637369bf53 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 14:03:40 +0100 Subject: [PATCH 03/29] docs: add initial AGENTS.md --- AGENTS.md | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) create mode 100644 AGENTS.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 00000000..5004a1ac --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,50 @@ +# Repository Guidelines + +## Project Structure & Module Organization + +- Workspace root contains shared config (`Cargo.toml`, `rustfmt.toml`) and documentation (`README.md`). +- `server/` is the core library crate. Main modules live under `server/src/` with submodules like `listeners/`, `order_book/`, `servers/`, and `types/`. +- `binaries/` contains runnable entry points in `binaries/src/bin/`: + - `websocket_server.rs` (primary server binary) + - `example_client.rs` (client example) +- Tests are colocated with code in module files via `#[test]` blocks (no top-level `tests/` directory). + +## Build, Test, and Development Commands + +```bash +cargo build --workspace +``` +Builds all crates in the workspace. + +```bash +cargo test --workspace +``` +Runs all unit tests across `server` and `binaries`. + +```bash +RUST_LOG=info cargo run --release --bin websocket_server -- --address 0.0.0.0 --port 8000 +``` +Runs the local WebSocket server with logging enabled. + +## Coding Style & Naming Conventions + +- Rust 2024 edition; format with `cargo fmt` (see `rustfmt.toml`, max width 120). +- Clippy lints are configured at the workspace level; keep code clippy-clean and avoid `unwrap`/`expect` where possible. +- Follow Rust naming conventions: `snake_case` for modules/functions, `CamelCase` for types, `SCREAMING_SNAKE_CASE` for constants. + +## Testing Guidelines + +- Prefer module-level unit tests near the code under test. +- Keep tests deterministic and focused on order book behavior and subscription parsing. +- Run targeted tests with `cargo test -p server` when working only on the library crate. + +## Commit & Pull Request Guidelines + +- Commit history uses short, imperative messages and sometimes a `type:` prefix (e.g., `fix: panic in orderbook listener`). +- Keep commits small and focused; include the primary behavior change in the subject line. +- PRs should include: a concise summary, test commands run, and notes about any API or protocol changes. + +## Runtime & Configuration Notes + +- The server expects a locally running non-validating Hyperliquid node (see `README.md` for setup details). +- If the node stops emitting events, the server may exit after a short timeout; plan local workflows accordingly. From 4e35b632dc19137d62aa39f61fdd29afc83a64a1 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 14:08:27 +0100 Subject: [PATCH 04/29] test: make tests use tempfile --- Cargo.lock | 1 + server/Cargo.toml | 1 + server/src/listeners/directory.rs | 5 +++-- server/src/order_book/multi_book.rs | 15 ++++++++------- 4 files changed, 13 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b2917ab5..a74ac49c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3554,6 +3554,7 @@ dependencies = [ "serde_json", "slab", "strum_macros", + "tempfile", "tokio", "yawc", ] diff --git a/server/Cargo.toml b/server/Cargo.toml index 2eb3e448..5da8dcdb 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -25,3 +25,4 @@ workspace = true [dev-dependencies] rand = "0.9.1" +tempfile = "3" diff --git a/server/src/listeners/directory.rs b/server/src/listeners/directory.rs index 471aa74f..05b477ae 100644 --- a/server/src/listeners/directory.rs +++ b/server/src/listeners/directory.rs @@ -40,9 +40,9 @@ mod tests { sync::{Arc, Mutex}, time::Duration, }; + use tempfile::tempdir; use tokio::{fs::File as TokioFile, io::AsyncWriteExt, sync::mpsc::unbounded_channel, time::sleep}; - const MOCK_HL_DIR: &str = "tmp/ws_listener_test"; const DATA: [&str; 2] = [ r#"{"coin":"@151","side":"A","time":"2025-06-24T02:56:36.172847427","px":"2393.9","sz":"0.1539","hash":"0x2b21750229be769650b604261eaac1018c00c45812652efbbdd35fe0ecb201a1","trade_dir_override":"Na","side_info":[{"user":"0xecb63caa47c7c4e77f60f1ce858cf28dc2b82b00","start_pos":"1166.565307356","oid":105686971733,"twap_id":null,"cloid":"0x1070fff92506b3ab5e5aec135e5a5ddd"},{"user":"0xb65117c1e1006e7b2413fa90e96fcbe3fa83ed75","start_pos":"0.153928559","oid":105686976226,"twap_id":null,"cloid":null}]} {"coin":"@166","side":"A","time":"2025-06-24T02:56:36.172847427","px":"1.0003","sz":"184.11","hash":"0x0ffc6896b2147680820e04261eaac1018c0101735014e44b56f038478b13ad8f","trade_dir_override":"Na","side_info":[{"user":"0x107332a1729ba0bcf6171117815a87b72a7e6082","start_pos":"36301.55539655","oid":105686050113,"twap_id":null,"cloid":null},{"user":"0xb65117c1e1006e7b2413fa90e96fcbe3fa83ed75","start_pos":"184.12704003","oid":105686976227,"twap_id":null,"cloid":null}]} @@ -213,7 +213,8 @@ mod tests { #[allow(clippy::significant_drop_tightening)] #[tokio::test] async fn test_trade_listener() -> Result<()> { - let mock_path = PathBuf::from(MOCK_HL_DIR); + let temp_dir = tempdir()?; + let mock_path = temp_dir.path().to_path_buf(); let event_source = EventSource::Fills; create_dir_all(event_source.event_source_dir(&mock_path))?; let history = Arc::new(Mutex::new(String::new())); diff --git a/server/src/order_book/multi_book.rs b/server/src/order_book/multi_book.rs index 694c0889..6ef7746c 100644 --- a/server/src/order_book/multi_book.rs +++ b/server/src/order_book/multi_book.rs @@ -116,7 +116,8 @@ mod tests { }; use alloy::primitives::Address; use itertools::Itertools; - use std::{fs::create_dir_all, path::PathBuf}; + use std::fs::create_dir_all; + use tempfile::tempdir; #[must_use] fn snapshot_to_l2_snapshot( @@ -274,12 +275,12 @@ mod tests { #[tokio::test] async fn test_deserialization_from_json() -> Result<()> { - create_dir_all("tmp/deserialization_test")?; - fs::write("tmp/deserialization_test/out.json", SNAPSHOT_JSON)?; - load_snapshots_from_json::(&PathBuf::from( - "tmp/deserialization_test/out.json", - )) - .await?; + let temp_dir = tempdir()?; + let test_dir = temp_dir.path().join("deserialization_test"); + create_dir_all(&test_dir)?; + let snapshot_path = test_dir.join("out.json"); + fs::write(&snapshot_path, SNAPSHOT_JSON)?; + load_snapshots_from_json::(&snapshot_path).await?; Ok(()) } From 8c5445589b22b4b81e1c49adc59ce0b7ebb7df6f Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 14:10:02 +0100 Subject: [PATCH 05/29] docs: architecture overview + test section in readme --- README.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/README.md b/README.md index b0fe860d..8ef94431 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,12 @@ The `l4book` subscription first sends a snapshot of the entire book and then for } ``` +## Architecture Overview + +- `server/` is the core library crate. It ingests node event files via `listeners/`, maintains order book state in `order_book/`, and exposes WebSocket/http handlers in `servers/`. +- `binaries/` contains runnable entry points such as `websocket_server.rs`, which wires configuration, logging, and the `server` crate together. +- Data flow: node event files -> listener parsing -> in-memory order books -> websocket subscriptions (`l2book`, `trades`, `l4book`). + ## Setup 1. Run a non-validating node (from [`hyperliquid-dex/node`](https://github.com/hyperliquid-dex/node)). Requires batching by block. Requires recording fills, order statuses, and raw book diffs. @@ -41,6 +47,14 @@ If you want logging, prepend the command with `RUST_LOG=info`. The WebSocket server comes with compression built-in. The compression ratio can be tuned using the `--websocket-compression-level` flag. +## Tests + +```bash +cargo test --workspace +``` + +Runs all unit tests for the `server` and `binaries` crates. + ## Caveats - This server does **not** show untriggered trigger orders. From ea8ce1f13fe4e40081e5e2ca54b5d1769b7f5c01 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 15:18:34 +0100 Subject: [PATCH 06/29] docs: add initial CLAUDE.md --- CLAUDE.md | 56 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 CLAUDE.md diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 00000000..33ea1a09 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,56 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Build and Test Commands + +```bash +cargo build --workspace # Build all crates +cargo test --workspace # Run all tests +cargo test -p server # Test library crate only +cargo test # Run single test by name +cargo fmt # Format (max width 120) +cargo clippy --workspace # Lint check (must pass before commits) + +# Run the server +RUST_LOG=info cargo run --release --bin websocket_server -- \ + --address 0.0.0.0 --port 8000 --websocket-compression-level 1 +``` + +## Architecture + +Local WebSocket server for Hyperliquid order book data. Ingests events from a non-validating Hyperliquid node and serves `l2book`, `trades`, and `l4book` subscriptions. + +**Data flow:** +``` +Node event files (~/.hl-node/events/{order_statuses,fills,order_diffs}/) + ↓ +listeners/order_book/ (file watchers + parsers) + ↓ +order_book/ (in-memory state: OrderBook, OrderBooks) + ↓ +servers/websocket_server.rs (axum + yawc) + ↓ +Client WebSocket subscriptions +``` + +**Workspace structure:** +- `server/` - Core library: `listeners/`, `order_book/`, `servers/`, `types/` +- `binaries/src/bin/` - Executables: `websocket_server.rs`, `example_client.rs` + +**Key abstractions:** +- `InnerOrder` trait (`order_book/types.rs`) - Generic contract for order types +- `OrderBook` (`order_book/mod.rs`) - Single coin book (BTreeMap of LinkedLists) +- `OrderBooks` (`order_book/multi_book.rs`) - Multi-coin container +- `OrderBookListener` (`listeners/order_book/mod.rs`) - Event ingestion + state management +- `SubscriptionManager` (`types/subscription.rs`) - Client subscription routing + +**Concurrency:** `Arc>` for shared state, `tokio::broadcast` for updates. + +## Code Conventions + +- Rust 2024 edition with strict clippy (all warn-level categories enabled) +- Avoid `unwrap`/`expect` except in `#[cfg(test)]` blocks +- `unsafe` code is warned - avoid unless necessary +- Error type: `Box` (see `prelude.rs`) +- Tests colocated in module files, not separate `tests/` directory From 2c698947dc96ad7273089e2c90f74e8d39a0bd8b Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 15:21:43 +0100 Subject: [PATCH 07/29] docs: merge all AGENT files into AGENTS.md --- AGENTS.md | 314 ++++++++++++++++++++++++++++++++++++++++++++++++------ CLAUDE.md | 57 +--------- GEMINI.md | 1 + 3 files changed, 285 insertions(+), 87 deletions(-) mode change 100644 => 120000 CLAUDE.md create mode 120000 GEMINI.md diff --git a/AGENTS.md b/AGENTS.md index 5004a1ac..38b68b0b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,50 +1,302 @@ -# Repository Guidelines +# AGENTS.md -## Project Structure & Module Organization +Comprehensive guidance for AI assistants and developers working in this repository. -- Workspace root contains shared config (`Cargo.toml`, `rustfmt.toml`) and documentation (`README.md`). -- `server/` is the core library crate. Main modules live under `server/src/` with submodules like `listeners/`, `order_book/`, `servers/`, and `types/`. -- `binaries/` contains runnable entry points in `binaries/src/bin/`: - - `websocket_server.rs` (primary server binary) - - `example_client.rs` (client example) -- Tests are colocated with code in module files via `#[test]` blocks (no top-level `tests/` directory). - -## Build, Test, and Development Commands +## Quick Start ```bash +# Build and lint cargo build --workspace -``` -Builds all crates in the workspace. +cargo clippy --workspace # Must pass before commits -```bash +# Run all tests cargo test --workspace + +# Run specific tests +cargo test -p server # Library crate only +cargo test test_trade_listener # Single test by name + +# Format code (max width 120) +cargo fmt + +# Run the server +RUST_LOG=info cargo run --release --bin websocket_server -- \ + --address 0.0.0.0 \ + --port 8000 \ + --websocket-compression-level 1 +``` + +## Architecture + +**Purpose:** Local WebSocket server for Hyperliquid order book data. Ingests events from a non-validating Hyperliquid node and serves `l2book`, `trades`, and `l4book` subscriptions with ~100ms latency (batched by block). + +### Data Flow + +``` +Node event files ($HOME/hl-node/events/) + ├── order_statuses/hourly/YYYYMMDD/*.jsonl + ├── fills/hourly/YYYYMMDD/*.jsonl + └── order_diffs/hourly/YYYYMMDD/*.jsonl + ↓ +listeners/order_book/ (DirectoryListener trait) + - File watchers (notify crate) + - Event parsers (rmp + serde_json) + - Batch synchronization by block number + ↓ +order_book/ (in-memory state) + - OrderBook: BTreeMap> + - OrderBooks: HashMap> + - Snapshot validation every 10s via reqwest + ↓ +servers/websocket_server.rs (axum + yawc) + - tokio::broadcast channels + - Per-client subscription filtering + ↓ +Client WebSocket subscriptions + - l2book: Aggregated levels (up to 100) + - trades: Fill events + - l4book: Full order-level book + diffs +``` + +### Workspace Structure + +``` +order_book_server/ +├── server/ # Core library crate +│ └── src/ +│ ├── listeners/ # Event ingestion layer +│ │ ├── directory.rs # DirectoryListener trait +│ │ └── order_book/ # OrderBookListener implementation +│ │ ├── mod.rs # Main listener + file watching +│ │ ├── state.rs # OrderBookState (update logic) +│ │ └── utils.rs # RMP processing, validation +│ ├── order_book/ # In-memory data structures +│ │ ├── mod.rs # OrderBook generic book +│ │ ├── types.rs # InnerOrder trait, Coin, Px, Sz +│ │ ├── multi_book.rs # OrderBooks, Snapshots +│ │ ├── linked_list.rs # Custom list for O(1) removal +│ │ └── levels.rs # Price level aggregation +│ ├── servers/ # HTTP/WebSocket layer +│ │ └── websocket_server.rs # axum handlers + routing +│ ├── types/ # Public API types +│ │ ├── mod.rs # Trade, L2Book, L4Book, Level +│ │ ├── subscription.rs # SubscriptionManager, validation +│ │ ├── node_data.rs # Node event schemas +│ │ └── inner.rs # Serialization wrappers +│ └── prelude.rs # Common imports + type aliases +└── binaries/ # Executable crates + └── src/bin/ + ├── websocket_server.rs # Main server binary + └── example_client.rs # Client example +``` + +### Key Abstractions + +| Component | Location | Purpose | +|-----------|----------|---------| +| `InnerOrder` trait | `order_book/types.rs` | Generic contract for order types (oid, side, price, size, matching) | +| `OrderBook` | `order_book/mod.rs` | Single-coin book: BTreeMap of LinkedLists for bid/ask | +| `OrderBooks` | `order_book/multi_book.rs` | Multi-coin container with concurrent access | +| `OrderBookListener` | `listeners/order_book/mod.rs` | Event ingestion + file watching + state updates | +| `OrderBookState` | `listeners/order_book/state.rs` | Apply order diffs, compute snapshots, validate consistency | +| `SubscriptionManager` | `types/subscription.rs` | Client subscription routing and filtering | +| `DirectoryListener` | `listeners/directory.rs` | Generic trait for file system watchers | + +### Concurrency Model + +- **Shared state:** `Arc>` for order book access +- **Broadcasting:** `tokio::broadcast::channel` for snapshot/fill/L4 updates +- **Async runtime:** Tokio with separate tasks for: + - File system watching (`notify` crate) + - Periodic snapshot validation (every 10s) + - WebSocket connection handlers +- **Parallelism:** `rayon` for parallel snapshot processing + +### Subscription Types + +1. **`trades`** - Fill events for a single coin +2. **`l2book`** - Aggregated price levels + - Optional `n_levels` (1-100, default 20) + - Optional `sig_figs` and `mantissa` for rounding +3. **`l4book`** - Full order-level book (new endpoint) + - Initial snapshot of all orders + - Block-by-block diffs (adds, cancels, size changes) + +## Code Conventions + +### Language & Formatting + +- **Edition:** Rust 2024 +- **Max line width:** 120 characters (`rustfmt.toml`) +- **Import style:** Crate-level granularity, grouped as StdExternalCrate +- **Naming:** Standard Rust conventions (`snake_case`, `CamelCase`, `SCREAMING_SNAKE_CASE`) + +### Linting + +Workspace-level clippy with **strict** configuration: +- All categories enabled at warn level: correctness, suspicious, complexity, perf, style, pedantic, nursery, cargo +- `unwrap_used` and `expect_used` warned (allowed only in `#[cfg(test)]`) +- `unsafe_code` warned (avoid unless necessary) +- See `Cargo.toml` for 20+ allowed exceptions (e.g., `cast_precision_loss`, `missing_panics_doc`) + +### Error Handling + +```rust +// Standard error type (see prelude.rs) +pub type Error = Box; +pub type Result = std::result::Result; + +// Avoid unwrap/expect except in tests +#[cfg(test)] +fn test_something() { + let val = maybe_value.unwrap(); // OK in tests +} +``` + +### Generic Design + +The codebase uses trait-based generics for flexibility: +```rust +pub trait InnerOrder: Clone + Send + Sync { + fn oid(&self) -> Oid; + fn side(&self) -> Side; + fn price(&self) -> Px; + fn size(&self) -> Sz; + // ... matching logic +} + +pub struct OrderBook { + bids: BTreeMap>, + asks: BTreeMap>, +} ``` -Runs all unit tests across `server` and `binaries`. + +## Testing + +### Test Organization + +- **Location:** Colocated in module files using `#[test]` blocks (no top-level `tests/` directory) +- **Async tests:** Use `#[tokio::test]` for async functions +- **Deterministic:** Fixed random seeds (`[42; 32]`), explicit wait durations (100ms) +- **Focus:** Order book operations, subscription validation, file system events + +### Test Patterns + +```rust +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_trade_listener() { + let temp_dir = tempfile::tempdir().unwrap(); + // ... setup with mock node data + } + + #[test] + fn test_order_book_add() { + // ... deterministic unit test + } +} +``` + +### Running Tests ```bash -RUST_LOG=info cargo run --release --bin websocket_server -- --address 0.0.0.0 --port 8000 +cargo test --workspace # All tests +cargo test -p server # Library only +cargo test test_order_book_add # Single test +cargo test -- --nocapture # Show println! output +RUST_LOG=debug cargo test # With logging ``` -Runs the local WebSocket server with logging enabled. -## Coding Style & Naming Conventions +## Development Workflow + +### Making Changes -- Rust 2024 edition; format with `cargo fmt` (see `rustfmt.toml`, max width 120). -- Clippy lints are configured at the workspace level; keep code clippy-clean and avoid `unwrap`/`expect` where possible. -- Follow Rust naming conventions: `snake_case` for modules/functions, `CamelCase` for types, `SCREAMING_SNAKE_CASE` for constants. +1. Read existing code to understand patterns before modifying +2. Maintain consistency with current architecture (generic traits, error types, async patterns) +3. Ensure clippy passes: `cargo clippy --workspace` +4. Format code: `cargo fmt` +5. Run relevant tests: `cargo test -p server` or `cargo test ` +6. Update documentation if public API changes + +### Commit Guidelines + +- Use short, imperative messages: "fix panic in orderbook listener" +- Optional type prefix: `fix:`, `feat:`, `test:`, `docs:` +- Keep commits focused on single behavior change +- Example history: + ``` + fix: panic in orderbook listener + use yawc as default WebSocket crate supporting compression + update docs + ``` + +### Pull Request Guidelines + +Include in PR description: +1. Concise summary of changes +2. Test commands executed +3. Notes on any API or protocol changes +4. Breaking changes (if any) + +## Runtime Behavior + +### Node Dependency + +- Requires locally running non-validating Hyperliquid node from [`hyperliquid-dex/node`](https://github.com/hyperliquid-dex/node) +- Node must be configured with: + - Batching by block enabled + - Recording: fills, order statuses, raw book diffs +- Default event directory: `$HOME/hl-node/events/` + +### Exit Conditions + +Server will automatically exit when: +1. **No events for 5 seconds** - Node stopped emitting data +2. **Snapshot validation fails** - Local state diverged from node (checked every 10s) +3. **Fatal errors** - `std::process::exit(1)` or `exit(2)` + +### Performance Characteristics + +- **Latency:** ~100ms behind live (batched by block, not streaming) +- **Update frequency:** Per-block updates (every ~1 second in practice) +- **Memory:** Full in-memory order books for all subscribed coins +- **Compression:** Tunable WebSocket compression (0-9, default 1) +- **Snapshot sync:** Periodic validation via HTTP to node (every 10s) + +### Logging + +```bash +RUST_LOG=info cargo run --bin websocket_server # Info level +RUST_LOG=debug cargo run --bin websocket_server # Debug level +RUST_LOG=server::listeners=trace # Module-specific +``` -## Testing Guidelines +## Key Dependencies -- Prefer module-level unit tests near the code under test. -- Keep tests deterministic and focused on order book behavior and subscription parsing. -- Run targeted tests with `cargo test -p server` when working only on the library crate. +| Crate | Purpose | +|-------|---------| +| `axum` | HTTP routing and WebSocket upgrade | +| `yawc` | WebSocket with compression support (replaces tungstenite) | +| `tokio` | Async runtime with full features | +| `notify` | File system event watching | +| `serde` + `serde_json` | JSON serialization | +| `alloy` | Ethereum primitives (addresses, signatures) | +| `rayon` | Parallel iteration for snapshot processing | +| `reqwest` | HTTP client for snapshot fetching | +| `itertools` | Iterator extensions | +| `chrono` | Date/time handling | -## Commit & Pull Request Guidelines +## Caveats & Limitations -- Commit history uses short, imperative messages and sometimes a `type:` prefix (e.g., `fix: panic in orderbook listener`). -- Keep commits small and focused; include the primary behavior change in the subject line. -- PRs should include: a concise summary, test commands run, and notes about any API or protocol changes. +- Does **not** show untriggered trigger orders +- Does **not** support spot order books currently +- Batched updates (not streaming) introduce ~100ms latency +- Requires node with specific configuration (batching, event recording) +- No persistence - state rebuilt on restart from node snapshots -## Runtime & Configuration Notes +## Project Context -- The server expects a locally running non-validating Hyperliquid node (see `README.md` for setup details). -- If the node stops emitting events, the server may exit after a short timeout; plan local workflows accordingly. +This is a standalone educational project, not maintained by Hyperliquid Labs. See `README.md` disclaimer for full context. Use at your own risk. diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 100644 index 33ea1a09..00000000 --- a/CLAUDE.md +++ /dev/null @@ -1,56 +0,0 @@ -# CLAUDE.md - -This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. - -## Build and Test Commands - -```bash -cargo build --workspace # Build all crates -cargo test --workspace # Run all tests -cargo test -p server # Test library crate only -cargo test # Run single test by name -cargo fmt # Format (max width 120) -cargo clippy --workspace # Lint check (must pass before commits) - -# Run the server -RUST_LOG=info cargo run --release --bin websocket_server -- \ - --address 0.0.0.0 --port 8000 --websocket-compression-level 1 -``` - -## Architecture - -Local WebSocket server for Hyperliquid order book data. Ingests events from a non-validating Hyperliquid node and serves `l2book`, `trades`, and `l4book` subscriptions. - -**Data flow:** -``` -Node event files (~/.hl-node/events/{order_statuses,fills,order_diffs}/) - ↓ -listeners/order_book/ (file watchers + parsers) - ↓ -order_book/ (in-memory state: OrderBook, OrderBooks) - ↓ -servers/websocket_server.rs (axum + yawc) - ↓ -Client WebSocket subscriptions -``` - -**Workspace structure:** -- `server/` - Core library: `listeners/`, `order_book/`, `servers/`, `types/` -- `binaries/src/bin/` - Executables: `websocket_server.rs`, `example_client.rs` - -**Key abstractions:** -- `InnerOrder` trait (`order_book/types.rs`) - Generic contract for order types -- `OrderBook` (`order_book/mod.rs`) - Single coin book (BTreeMap of LinkedLists) -- `OrderBooks` (`order_book/multi_book.rs`) - Multi-coin container -- `OrderBookListener` (`listeners/order_book/mod.rs`) - Event ingestion + state management -- `SubscriptionManager` (`types/subscription.rs`) - Client subscription routing - -**Concurrency:** `Arc>` for shared state, `tokio::broadcast` for updates. - -## Code Conventions - -- Rust 2024 edition with strict clippy (all warn-level categories enabled) -- Avoid `unwrap`/`expect` except in `#[cfg(test)]` blocks -- `unsafe` code is warned - avoid unless necessary -- Error type: `Box` (see `prelude.rs`) -- Tests colocated in module files, not separate `tests/` directory diff --git a/CLAUDE.md b/CLAUDE.md new file mode 120000 index 00000000..47dc3e3d --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file diff --git a/GEMINI.md b/GEMINI.md new file mode 120000 index 00000000..47dc3e3d --- /dev/null +++ b/GEMINI.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file From 4ff9bf59a4832184bd97ec95373e9aabe5882daf Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 15:30:02 +0100 Subject: [PATCH 08/29] docs: add guide for the new people --- NEW.md | 496 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ README.md | 2 + 2 files changed, 498 insertions(+) create mode 100644 NEW.md diff --git a/NEW.md b/NEW.md new file mode 100644 index 00000000..655c21ae --- /dev/null +++ b/NEW.md @@ -0,0 +1,496 @@ +# Newcomer's Guide to the Order Book Server + +A practical guide for developers experienced in Python/Go who are new to Rust. + +--- + +## What This Application Does + +This server streams real-time order book data from a [Hyperliquid](https://hyperliquid.xyz) node to clients via WebSocket. Think of it as a data pipeline: + +``` +┌──────────────────────┐ ┌──────────────────────┐ ┌─────────────────┐ +│ Hyperliquid Node │────▶│ This Server │────▶│ Your Clients │ +│ (produces events) │ │ (processes & serves)│ │ (consume data) │ +└──────────────────────┘ └──────────────────────┘ └─────────────────┘ +``` + +**Key features:** +- Maintains in-memory order books for all traded assets +- Serves three subscription types: `l2book`, `trades`, and `l4book` +- Validates state against node snapshots every 10 seconds +- Supports WebSocket compression for bandwidth efficiency + +--- + +## Architecture at a Glance + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ WORKSPACE LAYOUT │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ order_book_server/ │ +│ ├── server/ # Core library (all the logic) │ +│ │ └── src/ │ +│ │ ├── listeners/ # File watching & event parsing │ +│ │ ├── order_book/ # In-memory data structures │ +│ │ ├── servers/ # WebSocket handler │ +│ │ ├── types/ # API types & subscriptions │ +│ │ ├── lib.rs # Library entry point │ +│ │ └── prelude.rs # Common imports │ +│ │ │ +│ └── binaries/ # Executable crates │ +│ └── src/bin/ │ +│ ├── websocket_server.rs # Main binary │ +│ └── example_client.rs # Client example │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ +``` + +--- + +## Data Flow: The Big Picture + +The server processes three types of events from the node: + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ NODE EVENT FILES │ +│ ($HOME/hl-node/events/) │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ fills/hourly/YYYYMMDD/*.jsonl │ +│ └─▶ Trade executions (who bought/sold, price, size) │ +│ │ +│ order_statuses/hourly/YYYYMMDD/*.jsonl │ +│ └─▶ Order lifecycle events (opened, triggered, cancelled) │ +│ │ +│ order_diffs/hourly/YYYYMMDD/*.jsonl │ +│ └─▶ Raw book changes (add order, update size, remove order) │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────────────────┐ +│ DIRECTORY LISTENERS │ +│ (server/src/listeners/) │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ Uses `notify` crate to watch for file changes │ +│ Parses MessagePack (RMP) encoded batches │ +│ Synchronizes events by block number │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────────────────┐ +│ ORDER BOOK STATE │ +│ (server/src/order_book/) │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ OrderBook Single-coin book with bid/ask sides │ +│ OrderBooks Multi-coin container │ +│ LinkedList O(1) order cancellation │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────────────────┐ +│ BROADCAST CHANNELS │ +│ (tokio::broadcast) │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ InternalMessage::Snapshot → L2 book updates │ +│ InternalMessage::Fills → Trade events │ +│ InternalMessage::L4BookUpdates → Full book diffs │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────────────────┐ +│ WEBSOCKET CLIENTS │ +│ (filtered by subscription) │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ Client A: subscribed to l2book ETH │ +│ Client B: subscribed to trades BTC, l4book ETH │ +│ Client C: subscribed to l2book BTC (5 sig figs) │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ +``` + +--- + +## Core Components Explained + +### 1. The Order Book (`server/src/order_book/`) + +**Rust vs Python/Go context:** +In Python, you might use a `dict` with price as key and a list of orders as value. +In Go, you'd use a `map[float64][]Order`. +In Rust, we use `BTreeMap>` for sorted iteration by price. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ OrderBook │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ bids: BTreeMap> (sorted high→low) │ +│ ┌─────────┬──────────────────────────────────────┐ │ +│ │ $100.50 │ [Order1] ←→ [Order2] ←→ [Order3] │ │ +│ │ $100.40 │ [Order4] ←→ [Order5] │ │ +│ │ $100.30 │ [Order6] │ │ +│ └─────────┴──────────────────────────────────────┘ │ +│ │ +│ asks: BTreeMap> (sorted low→high) │ +│ ┌─────────┬──────────────────────────────────────┐ │ +│ │ $100.60 │ [Order7] ←→ [Order8] │ │ +│ │ $100.70 │ [Order9] │ │ +│ └─────────┴──────────────────────────────────────┘ │ +│ │ +│ oid_to_side_px: HashMap (for O(1) lookup) │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ +``` + +**Why a custom LinkedList?** +Standard library linked lists don't support O(1) removal by key. +Our `LinkedList` uses a `slab::Slab` (arena allocator) plus a `HashMap` +to enable instant order cancellation without traversing the list. + +### 2. The Listener (`server/src/listeners/order_book/`) + +The listener watches node event files and applies updates to the order book. + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ OrderBookListener │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ order_book_state: Option │ +│ └─▶ None until first snapshot fetched │ +│ │ +│ order_status_cache: BatchQueue │ +│ order_diff_cache: BatchQueue │ +│ └─▶ Buffer events until paired by block number │ +│ │ +│ internal_message_tx: Sender │ +│ └─▶ Broadcast channel to WebSocket clients │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ + │ + ┌───────────────────────────┼───────────────────────────┐ + │ │ │ + ▼ ▼ ▼ + ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ + │ Fills │ │ Statuses │ │ Diffs │ + │ Watcher │ │ Watcher │ │ Watcher │ + └─────────────┘ └─────────────┘ └─────────────┘ + │ │ │ + │ └─────────┬─────────────────┘ + │ │ + ▼ ▼ + Broadcast Pair by block, + immediately then apply to + as Trades OrderBookState +``` + +### 3. The WebSocket Server (`server/src/servers/websocket_server.rs`) + +Each connected client gets its own async task that: +1. Subscribes to the internal broadcast channel +2. Filters messages based on client subscriptions +3. Serializes and sends matching data + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ WebSocket Handler Flow │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ tokio::select! { │ +│ msg = internal_rx.recv() => { │ +│ // Filter by client's SubscriptionManager │ +│ // Send matching L2/L4/Trades to client │ +│ } │ +│ msg = client_socket.recv() => { │ +│ // Parse subscribe/unsubscribe requests │ +│ // Update SubscriptionManager │ +│ } │ +│ } │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ +``` + +--- + +## Key Rust Patterns (for Python/Go developers) + +### Generics with Traits + +**Python equivalent:** Duck typing / Protocol classes +**Go equivalent:** Interfaces + +This codebase uses a trait called `InnerOrder` to abstract over different order types: + +```rust +// server/src/order_book/types.rs +pub trait InnerOrder: Clone + Send + Sync { + fn oid(&self) -> Oid; + fn side(&self) -> Side; + fn limit_px(&self) -> Px; + fn sz(&self) -> Sz; + fn fill(&mut self, maker_order: &mut Self) -> Sz; + fn modify_sz(&mut self, sz: Sz); +} +``` + +Then `OrderBook` works with any type implementing this trait. +This is similar to Go's `interface{}` but with compile-time guarantees. + +### Error Handling + +**Python:** Exceptions with try/except +**Go:** Multiple returns with `(value, error)` +**Rust:** `Result` enum + +```rust +// server/src/prelude.rs +pub type Error = Box; +pub type Result = std::result::Result; + +// Usage (with ? operator for propagation, like Go's if err != nil) +fn do_something() -> Result { + let file = std::fs::read_to_string(path)?; // Returns early on error + Ok(parse(file)?) +} +``` + +### Ownership and Borrowing + +**The key concept Python/Go developers need to understand:** + +```rust +// In Python/Go, this would just copy a reference: +let book = OrderBook::new(); +process(book); // In Rust: book is MOVED, can't use it anymore! + +// To keep using it, you borrow: +process(&book); // Immutable borrow (like passing a const pointer) +process(&mut book); // Mutable borrow (exclusive access) + +// Or clone (explicit copy, like Python's copy.deepcopy): +process(book.clone()); +``` + +### Async/Await + +**Python:** `async def` / `await` +**Go:** Goroutines / channels +**Rust:** `async fn` / `.await` with an executor (Tokio) + +```rust +// Rust async looks like Python's +async fn fetch_data() -> Result { + let response = client.get(url).await?; + Ok(response.json().await?) +} + +// But spawning is explicit (like Go's go keyword) +tokio::spawn(async move { + loop { + process().await; + } +}); +``` + +--- + +## The Subscription System + +Clients send JSON messages to subscribe: + +```json +{"method": "subscribe", "subscription": {"type": "l2Book", "coin": "ETH"}} +{"method": "subscribe", "subscription": {"type": "trades", "coin": "BTC"}} +{"method": "subscribe", "subscription": {"type": "l4Book", "coin": "ETH"}} +``` + +**L2 Book options:** +- `n_levels`: 1-100 (default 20) - depth of book +- `n_sig_figs`: 2-5 - price grouping by significant figures +- `mantissa`: 2 or 5 - rounding factor + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ Subscription Data Flow │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ ┌─────────────┐ │ +│ │ Client │─── subscribe(l2book, ETH) ───▶ SubscriptionManager │ +│ │ WebSocket │ │ │ +│ └─────────────┘ │ │ +│ ▲ ▼ │ +│ │ ┌──────────────────┐ │ +│ │ │ HashSet │ │ +│ │ │ - l2Book(ETH) │ │ +│ │ └──────────────────┘ │ +│ │ │ │ +│ │ │ filter │ +│ │ ▼ │ +│ └──── L2Book { coin: "ETH", ... } ────────────┘ │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ +``` + +--- + +## Running and Testing + +### Quick Start + +```bash +# Build everything +cargo build --workspace + +# Run linter (required before commits) +cargo clippy --workspace + +# Run tests +cargo test --workspace + +# Run server (requires node running) +RUST_LOG=info cargo run --release --bin websocket_server -- \ + --address 0.0.0.0 \ + --port 8000 +``` + +### Testing a Subscription + +**Using websocat (simple interactive client):** +```bash +websocat ws://localhost:8000/ws + +# Then send: +{"method":"subscribe","subscription":{"type":"l2Book","coin":"ETH"}} +``` + +**Using wsstat (performance testing and stats):** +```bash +# Install wsstat: https://github.com/jkbrsn/wsstat +# Test connection and measure latency +wsstat \ + -s -t '{"method":"subscribe","subscription":{"type":"l2Book","coin":"ETH"}}' \ + --summary-interval 10s \ + ws://localhost:8000/ws +``` + +--- + +## File-by-File Guide + +| File | What it does | Read this if... | +|------|--------------|-----------------| +| `server/src/prelude.rs` | Type aliases (`Error`, `Result`) | You want to understand error handling | +| `server/src/order_book/types.rs` | Core types (`Px`, `Sz`, `Coin`, `InnerOrder`) | You want to understand the data model | +| `server/src/order_book/mod.rs` | `OrderBook` implementation | You want to understand order matching | +| `server/src/order_book/linked_list.rs` | Custom linked list with O(1) removal | You want to understand the performance optimization | +| `server/src/listeners/order_book/mod.rs` | File watching and event processing | You want to understand data ingestion | +| `server/src/listeners/order_book/state.rs` | Order book state updates | You want to understand how updates are applied | +| `server/src/types/subscription.rs` | Subscription types and validation | You want to understand the client API | +| `server/src/servers/websocket_server.rs` | WebSocket routing and handlers | You want to understand the server layer | +| `binaries/src/bin/websocket_server.rs` | CLI entry point | You want to understand startup configuration | + +--- + +## Common Tasks + +### Adding a New Subscription Type + +1. Add variant to `Subscription` enum in `types/subscription.rs` +2. Add variant to `InternalMessage` in `listeners/order_book/mod.rs` +3. Add handling in `handle_socket()` in `servers/websocket_server.rs` +4. Add serialization type in `types/mod.rs` + +### Modifying Order Book Logic + +1. Update `InnerOrder` trait if changing order properties +2. Modify `OrderBook::add_order()` for matching logic +3. Update `OrderBookState::apply_updates()` for state transitions +4. Add tests in the same file using `#[cfg(test)]` blocks + +### Debugging + +```bash +# Full debug logging +RUST_LOG=debug cargo run --bin websocket_server + +# Module-specific logging +RUST_LOG=server::listeners=trace cargo run --bin websocket_server + +# Run specific test with output +cargo test test_order_book_add -- --nocapture +``` + +--- + +## Concurrency Model Summary + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ Concurrency Architecture │ +├─────────────────────────────────────────────────────────────────────────────┤ +│ │ +│ Main Thread │ +│ └─▶ Spawns Tokio runtime │ +│ │ +│ Tokio Tasks: │ +│ ┌─────────────────────────────────────────────────────────────────────┐ │ +│ │ hl_listen() task │ │ +│ │ └─▶ File watchers (notify crate, 3 directories) │ │ +│ │ └─▶ Snapshot fetch timer (every 10s) │ │ +│ │ └─▶ Heartbeat checker (exit if no events for 5s) │ │ +│ └─────────────────────────────────────────────────────────────────────┘ │ +│ │ +│ ┌─────────────────────────────────────────────────────────────────────┐ │ +│ │ Per-client WebSocket tasks (spawned on connection) │ │ +│ │ └─▶ Receives from broadcast channel │ │ +│ │ └─▶ Sends filtered messages to client │ │ +│ └─────────────────────────────────────────────────────────────────────┘ │ +│ │ +│ Shared State: │ +│ └─▶ Arc> (async mutex, short hold times) │ +│ │ +│ Communication: │ +│ └─▶ tokio::broadcast::channel (100 capacity, multi-consumer) │ +│ │ +└─────────────────────────────────────────────────────────────────────────────┘ +``` + +**Go comparison:** +- `tokio::spawn` is like `go func()` +- `tokio::broadcast` is like a buffered channel with multiple receivers +- `Arc>` is like `sync.Mutex` protecting shared data + +--- + +## Exit Conditions + +The server will exit automatically when: + +| Condition | Exit Code | Meaning | +|-----------|-----------|---------| +| No events for 5 seconds | 1 | Node stopped producing data | +| Snapshot validation fails | 2 | Local state diverged from node | +| Channel closed | 1 | Internal communication failure | + +This is intentional: the server should be restarted by a supervisor (systemd, Docker, etc.) if it exits. + +--- + +## Further Reading + +- [The Rust Book](https://doc.rust-lang.org/book/) - Official Rust tutorial +- [Tokio Tutorial](https://tokio.rs/tokio/tutorial) - Async Rust +- [AGENTS.md](./AGENTS.md) - Detailed technical reference for this codebase +- [README.md](./README.md) - Quick start and API reference diff --git a/README.md b/README.md index 8ef94431..226dea69 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,8 @@ The `l4book` subscription first sends a snapshot of the entire book and then for ## Architecture Overview +For a detailed guide with diagrams aimed at developers new to Rust, see [NEW.md](./NEW.md). + - `server/` is the core library crate. It ingests node event files via `listeners/`, maintains order book state in `order_book/`, and exposes WebSocket/http handlers in `servers/`. - `binaries/` contains runnable entry points such as `websocket_server.rs`, which wires configuration, logging, and the `server` crate together. - Data flow: node event files -> listener parsing -> in-memory order books -> websocket subscriptions (`l2book`, `trades`, `l4book`). From 11b62b2bcfd3c7f0fe22b1898d4dc44581aa975d Mon Sep 17 00:00:00 2001 From: Jonathan Udd Date: Thu, 25 Sep 2025 14:06:47 +0200 Subject: [PATCH 09/29] feat(add): CLI arg inactivity_exit_secs for tuning how long the server waits until it exits when there is no new data --- README.md | 4 +++- binaries/src/bin/websocket_server.rs | 9 ++++++++- server/src/listeners/order_book/mod.rs | 4 ++-- server/src/servers/websocket_server.rs | 4 ++-- 4 files changed, 15 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 226dea69..cc943983 100644 --- a/README.md +++ b/README.md @@ -40,9 +40,11 @@ For a detailed guide with diagrams aimed at developers new to Rust, see [NEW.md] ```bash cargo run --release --bin websocket_server -- --address 0.0.0.0 --port 8000 +# With custom inactivity timeout (e.g., 30s): +cargo run --release --bin websocket_server -- --address 0.0.0.0 --port 8000 --inactivity-exit-secs 30 ``` -If this local server does not detect the node writing down any new events, it will automatically exit after some amount of time (currently set to 5 seconds). +If this local server does not detect the node writing down any new events, it will automatically exit after some amount of time (default 5 seconds; configurable via `--inactivity-exit-secs `). In addition, the local server periodically fetches order book snapshots from the node, and compares to its own internal state. If a difference is detected, it will exit. If you want logging, prepend the command with `RUST_LOG=info`. diff --git a/binaries/src/bin/websocket_server.rs b/binaries/src/bin/websocket_server.rs index 09a605f7..ad855393 100644 --- a/binaries/src/bin/websocket_server.rs +++ b/binaries/src/bin/websocket_server.rs @@ -25,6 +25,12 @@ struct Args { /// documentation for for more info. #[arg(long)] websocket_compression_level: Option, + + /// Inactivity timeout in seconds before server exits. + /// If no node events are observed for this duration, the process exits. + /// Default and minimum is 5 seconds. + #[arg(long)] + inactivity_exit_secs: Option, } #[tokio::main] @@ -37,7 +43,8 @@ async fn main() -> Result<()> { println!("Running websocket server on {full_address}"); let compression_level = args.websocket_compression_level.unwrap_or(/* Some compression */ 1); - run_websocket_server(&full_address, true, compression_level).await?; + let inactivity_exit_secs = args.inactivity_exit_secs.unwrap_or(5).max(5); + run_websocket_server(&full_address, true, compression_level, inactivity_exit_secs).await?; Ok(()) } diff --git a/server/src/listeners/order_book/mod.rs b/server/src/listeners/order_book/mod.rs index 3f37789a..164025e4 100644 --- a/server/src/listeners/order_book/mod.rs +++ b/server/src/listeners/order_book/mod.rs @@ -39,7 +39,7 @@ mod utils; // WARNING - this code assumes no other file system operations are occurring in the watched directories // if there are scripts running, this may not work as intended -pub(crate) async fn hl_listen(listener: Arc>, dir: PathBuf) -> Result<()> { +pub(crate) async fn hl_listen(listener: Arc>, dir: PathBuf, inactivity_exit_secs: u64) -> Result<()> { let order_statuses_dir = EventSource::OrderStatuses.event_source_dir(&dir).canonicalize()?; let fills_dir = EventSource::Fills.event_source_dir(&dir).canonicalize()?; let order_diffs_dir = EventSource::OrderDiffs.event_source_dir(&dir).canonicalize()?; @@ -129,7 +129,7 @@ pub(crate) async fn hl_listen(listener: Arc>, dir: Path let snapshot_fetch_task_tx = snapshot_fetch_task_tx.clone(); fetch_snapshot(dir.clone(), listener, snapshot_fetch_task_tx, ignore_spot); } - () = sleep(Duration::from_secs(5)) => { + () = sleep(Duration::from_secs(inactivity_exit_secs)) => { let listener = listener.lock().await; if listener.is_ready() { return Err(format!("Stream has fallen behind ({HL_NODE} failed?)").into()); diff --git a/server/src/servers/websocket_server.rs b/server/src/servers/websocket_server.rs index 05d04fd6..acf7d0da 100644 --- a/server/src/servers/websocket_server.rs +++ b/server/src/servers/websocket_server.rs @@ -29,7 +29,7 @@ use tokio::{ }; use yawc::{FrameView, OpCode, WebSocket}; -pub async fn run_websocket_server(address: &str, ignore_spot: bool, compression_level: u32) -> Result<()> { +pub async fn run_websocket_server(address: &str, ignore_spot: bool, compression_level: u32, inactivity_exit_secs: u64) -> Result<()> { let (internal_message_tx, _) = channel::>(100); // Central task: listen to messages and forward them for distribution @@ -42,7 +42,7 @@ pub async fn run_websocket_server(address: &str, ignore_spot: bool, compression_ { let listener = listener.clone(); tokio::spawn(async move { - if let Err(err) = hl_listen(listener, home_dir).await { + if let Err(err) = hl_listen(listener, home_dir, inactivity_exit_secs).await { error!("Listener fatal error: {err}"); std::process::exit(1); } From ed423be203ac60e408ca4fbe1a0a0ab8744e35b4 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 16:05:50 +0100 Subject: [PATCH 10/29] docs: add date comment to NEW.md --- NEW.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/NEW.md b/NEW.md index 655c21ae..b0df6bca 100644 --- a/NEW.md +++ b/NEW.md @@ -2,6 +2,8 @@ A practical guide for developers experienced in Python/Go who are new to Rust. +**NOTE**: this guide was created on 2026-01-19, commit `cbeb1b5`, the project may have changed since then. + --- ## What This Application Does From a23faeb8080d0e9160f6098302630237c82a0b2d Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 16:25:00 +0100 Subject: [PATCH 11/29] ci: add CI tests workflow --- .github/workflows/ci-tests.yml | 70 ++++++++++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) create mode 100644 .github/workflows/ci-tests.yml diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml new file mode 100644 index 00000000..0d4b71c6 --- /dev/null +++ b/.github/workflows/ci-tests.yml @@ -0,0 +1,70 @@ +name: CI Tests + +on: + pull_request: + branches: + - '**' + push: + branches: + - main + +env: + RUST_TOOLCHAIN: stable + +jobs: + clippy: + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@stable + with: + components: clippy, rustfmt + + - name: fmt + run: cargo fmt --all -- --check + + - name: clippy + run: cargo clippy --workspace --all-targets --all-features -- -D warnings + + unit-tests: + name: Unit tests + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@stable + - run: cargo test --workspace --all-features -- --shuffle + + unit-tests-flaky: + name: Unit tests (flaky) + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@stable + - run: | + for i in {1..16}; do + cargo test --workspace --all-features -- --shuffle + done + + summary: + name: CI Summary + runs-on: ubuntu-24.04 + needs: [clippy, unit-tests, unit-tests-flaky] + if: always() # Always run, even if earlier jobs fail + steps: + - name: Write CI summary + run: | + { + echo "## CI Test Summary" + echo "" + echo "- **Clippy + fmt:** ${{ needs.clippy.result }}" + echo "- **Unit:** ${{ needs.unit-tests.result }}" + echo "- **Unit (flaky):** ${{ needs.unit-tests-flaky.result }}" + echo "" + echo "### Commit" + echo "[${{ github.sha }}](${{ github.server_url }}/${{ github.repository }}/commit/${{ github.sha }})" + echo "" + echo "Triggered by: @${{ github.actor }}" + echo "Event: \`${{ github.event_name }}\`" + echo "" + echo "[View workflow run](${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }})" + } >> "$GITHUB_STEP_SUMMARY" From ac70bd70e0b745f0ba75c49d0fb2f4fa227ed945 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 16:33:46 +0100 Subject: [PATCH 12/29] ci: add CI sanitizers workflow --- .github/workflows/ci-sanitizers.yml | 63 +++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 .github/workflows/ci-sanitizers.yml diff --git a/.github/workflows/ci-sanitizers.yml b/.github/workflows/ci-sanitizers.yml new file mode 100644 index 00000000..ad6cc01f --- /dev/null +++ b/.github/workflows/ci-sanitizers.yml @@ -0,0 +1,63 @@ +name: CI Sanitizers + +on: + workflow_dispatch: + schedule: + - cron: "0 3 * * 1" + push: + branches: + - main + +env: + RUST_TOOLCHAIN: nightly + +jobs: + sanitizer-asan: + name: Sanitizer (ASan) + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@nightly + with: + components: rust-src + - name: AddressSanitizer + run: | + ASAN_OPTIONS=detect_leaks=0 \ + RUSTFLAGS="-Z sanitizer=address" \ + cargo +${{ env.RUST_TOOLCHAIN }} test --workspace --all-features -Z sanitizer=address -Z build-std --target x86_64-unknown-linux-gnu + + sanitizer-tsan: + name: Sanitizer (TSan) + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@nightly + with: + components: rust-src + - name: ThreadSanitizer + run: | + RUSTFLAGS="-Z sanitizer=thread" \ + cargo +${{ env.RUST_TOOLCHAIN }} test --workspace --all-features -Z sanitizer=thread -Z build-std --target x86_64-unknown-linux-gnu + + summary: + name: Sanitizer Summary + runs-on: ubuntu-24.04 + needs: [sanitizer-asan, sanitizer-tsan] + if: always() + steps: + - name: Write CI summary + run: | + { + echo "## Sanitizer Summary" + echo "" + echo "- **ASan:** ${{ needs.sanitizer-asan.result }}" + echo "- **TSan:** ${{ needs.sanitizer-tsan.result }}" + echo "" + echo "### Commit" + echo "[${{ github.sha }}](${{ github.server_url }}/${{ github.repository }}/commit/${{ github.sha }})" + echo "" + echo "Triggered by: @${{ github.actor }}" + echo "Event: \`${{ github.event_name }}\`" + echo "" + echo "[View workflow run](${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }})" + } >> "$GITHUB_STEP_SUMMARY" From 630934ea6a8f55040618cc37f8db2d6ed4f83cb4 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 16:43:51 +0100 Subject: [PATCH 13/29] ci: make CI tests workflow use nightly toolchain --- .github/workflows/ci-tests.yml | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml index 0d4b71c6..2da70497 100644 --- a/.github/workflows/ci-tests.yml +++ b/.github/workflows/ci-tests.yml @@ -8,15 +8,12 @@ on: branches: - main -env: - RUST_TOOLCHAIN: stable - jobs: clippy: runs-on: ubuntu-24.04 steps: - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@nightly with: components: clippy, rustfmt @@ -31,18 +28,18 @@ jobs: runs-on: ubuntu-24.04 steps: - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@stable - - run: cargo test --workspace --all-features -- --shuffle + - uses: dtolnay/rust-toolchain@nightly + - run: cargo test -Z unstable-options --workspace --all-features -- --shuffle unit-tests-flaky: name: Unit tests (flaky) runs-on: ubuntu-24.04 steps: - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@nightly - run: | for i in {1..16}; do - cargo test --workspace --all-features -- --shuffle + cargo test -Z unstable-options --workspace --all-features -- --shuffle done summary: From b0c732b68546f86e99f923f28f722637369fc252 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 16:51:16 +0100 Subject: [PATCH 14/29] ci: move flaky tests to nightly --- .../{ci-sanitizers.yml => ci-nightly.yml} | 16 ++++++++++++++-- .github/workflows/ci-tests.yml | 14 +------------- 2 files changed, 15 insertions(+), 15 deletions(-) rename .github/workflows/{ci-sanitizers.yml => ci-nightly.yml} (80%) diff --git a/.github/workflows/ci-sanitizers.yml b/.github/workflows/ci-nightly.yml similarity index 80% rename from .github/workflows/ci-sanitizers.yml rename to .github/workflows/ci-nightly.yml index ad6cc01f..3e2e4aad 100644 --- a/.github/workflows/ci-sanitizers.yml +++ b/.github/workflows/ci-nightly.yml @@ -1,4 +1,4 @@ -name: CI Sanitizers +name: CI Nightly on: workflow_dispatch: @@ -39,10 +39,21 @@ jobs: RUSTFLAGS="-Z sanitizer=thread" \ cargo +${{ env.RUST_TOOLCHAIN }} test --workspace --all-features -Z sanitizer=thread -Z build-std --target x86_64-unknown-linux-gnu + unit-tests-flaky: + name: Unit tests (flaky) + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@nightly + - run: | + for i in {1..16}; do + cargo test -Z unstable-options --workspace --all-features -- --shuffle + done + summary: name: Sanitizer Summary runs-on: ubuntu-24.04 - needs: [sanitizer-asan, sanitizer-tsan] + needs: [sanitizer-asan, sanitizer-tsan, unit-tests-flaky] if: always() steps: - name: Write CI summary @@ -52,6 +63,7 @@ jobs: echo "" echo "- **ASan:** ${{ needs.sanitizer-asan.result }}" echo "- **TSan:** ${{ needs.sanitizer-tsan.result }}" + echo "- **Unit (flaky):** ${{ needs.unit-tests-flaky.result }}" echo "" echo "### Commit" echo "[${{ github.sha }}](${{ github.server_url }}/${{ github.repository }}/commit/${{ github.sha }})" diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml index 2da70497..b2548ccd 100644 --- a/.github/workflows/ci-tests.yml +++ b/.github/workflows/ci-tests.yml @@ -31,21 +31,10 @@ jobs: - uses: dtolnay/rust-toolchain@nightly - run: cargo test -Z unstable-options --workspace --all-features -- --shuffle - unit-tests-flaky: - name: Unit tests (flaky) - runs-on: ubuntu-24.04 - steps: - - uses: actions/checkout@v6 - - uses: dtolnay/rust-toolchain@nightly - - run: | - for i in {1..16}; do - cargo test -Z unstable-options --workspace --all-features -- --shuffle - done - summary: name: CI Summary runs-on: ubuntu-24.04 - needs: [clippy, unit-tests, unit-tests-flaky] + needs: [clippy, unit-tests] if: always() # Always run, even if earlier jobs fail steps: - name: Write CI summary @@ -55,7 +44,6 @@ jobs: echo "" echo "- **Clippy + fmt:** ${{ needs.clippy.result }}" echo "- **Unit:** ${{ needs.unit-tests.result }}" - echo "- **Unit (flaky):** ${{ needs.unit-tests-flaky.result }}" echo "" echo "### Commit" echo "[${{ github.sha }}](${{ github.server_url }}/${{ github.repository }}/commit/${{ github.sha }})" From a19067beb125897aa314eca3abcc47ead9ebd754 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Mon, 19 Jan 2026 16:53:23 +0100 Subject: [PATCH 15/29] style: cargo fmt --- server/src/listeners/directory.rs | 24 +++++++----- server/src/listeners/order_book/mod.rs | 48 +++++++++++++----------- server/src/listeners/order_book/state.rs | 3 +- server/src/listeners/order_book/utils.rs | 17 +++++---- server/src/order_book/levels.rs | 8 ++-- server/src/order_book/linked_list.rs | 12 ++++-- server/src/order_book/mod.rs | 10 +++-- server/src/order_book/multi_book.rs | 24 +++++++----- server/src/order_book/types.rs | 10 +++-- server/src/servers/websocket_server.rs | 43 ++++++++++++--------- server/src/types/subscription.rs | 9 +++-- 11 files changed, 121 insertions(+), 87 deletions(-) diff --git a/server/src/listeners/directory.rs b/server/src/listeners/directory.rs index 05b477ae..0396a8d1 100644 --- a/server/src/listeners/directory.rs +++ b/server/src/listeners/directory.rs @@ -1,7 +1,9 @@ -use crate::{prelude::*, types::node_data::EventSource}; +use std::path::PathBuf; + use fs::File; use io::Read; -use std::path::PathBuf; + +use crate::{prelude::*, types::node_data::EventSource}; // We want all of these functions to be synchronous just for ease of use since they are fast (for now) // Asynchronous stuff can be done in the listen function (waiting for next file event) @@ -26,23 +28,25 @@ pub(crate) trait DirectoryListener { #[cfg(test)] mod tests { - use crate::{ - listeners::directory::{DirectoryListener, EventSource}, - prelude::*, - }; - use fs::{File, create_dir_all, read_dir, remove_dir_all, remove_file}; - use log::{error, info}; - use notify::{RecursiveMode, Watcher, recommended_watcher}; - use rand::{Rng, SeedableRng, rngs::StdRng}; use std::{ io::{Seek, SeekFrom}, path::{Path, PathBuf}, sync::{Arc, Mutex}, time::Duration, }; + + use fs::{File, create_dir_all, read_dir, remove_dir_all, remove_file}; + use log::{error, info}; + use notify::{RecursiveMode, Watcher, recommended_watcher}; + use rand::{Rng, SeedableRng, rngs::StdRng}; use tempfile::tempdir; use tokio::{fs::File as TokioFile, io::AsyncWriteExt, sync::mpsc::unbounded_channel, time::sleep}; + use crate::{ + listeners::directory::{DirectoryListener, EventSource}, + prelude::*, + }; + const DATA: [&str; 2] = [ r#"{"coin":"@151","side":"A","time":"2025-06-24T02:56:36.172847427","px":"2393.9","sz":"0.1539","hash":"0x2b21750229be769650b604261eaac1018c00c45812652efbbdd35fe0ecb201a1","trade_dir_override":"Na","side_info":[{"user":"0xecb63caa47c7c4e77f60f1ce858cf28dc2b82b00","start_pos":"1166.565307356","oid":105686971733,"twap_id":null,"cloid":"0x1070fff92506b3ab5e5aec135e5a5ddd"},{"user":"0xb65117c1e1006e7b2413fa90e96fcbe3fa83ed75","start_pos":"0.153928559","oid":105686976226,"twap_id":null,"cloid":null}]} {"coin":"@166","side":"A","time":"2025-06-24T02:56:36.172847427","px":"1.0003","sz":"184.11","hash":"0x0ffc6896b2147680820e04261eaac1018c0101735014e44b56f038478b13ad8f","trade_dir_override":"Na","side_info":[{"user":"0x107332a1729ba0bcf6171117815a87b72a7e6082","start_pos":"36301.55539655","oid":105686050113,"twap_id":null,"cloid":null},{"user":"0xb65117c1e1006e7b2413fa90e96fcbe3fa83ed75","start_pos":"184.12704003","oid":105686976227,"twap_id":null,"cloid":null}]} diff --git a/server/src/listeners/order_book/mod.rs b/server/src/listeners/order_book/mod.rs index 164025e4..c3e093df 100644 --- a/server/src/listeners/order_book/mod.rs +++ b/server/src/listeners/order_book/mod.rs @@ -1,21 +1,3 @@ -use crate::{ - HL_NODE, - listeners::{directory::DirectoryListener, order_book::state::OrderBookState}, - order_book::{ - Coin, Snapshot, - multi_book::{Snapshots, load_snapshots_from_json}, - }, - prelude::*, - types::{ - L4Order, - inner::{InnerL4Order, InnerLevel}, - node_data::{Batch, EventSource, NodeDataFill, NodeDataOrderDiff, NodeDataOrderStatus}, - }, -}; -use alloy::primitives::Address; -use fs::File; -use log::{error, info}; -use notify::{Event, RecursiveMode, Watcher, recommended_watcher}; use std::{ cmp::Ordering, collections::{HashMap, HashSet, VecDeque}, @@ -24,6 +6,11 @@ use std::{ sync::Arc, time::Duration, }; + +use alloy::primitives::Address; +use fs::File; +use log::{error, info}; +use notify::{Event, RecursiveMode, Watcher, recommended_watcher}; use tokio::{ sync::{ Mutex, @@ -34,12 +21,31 @@ use tokio::{ }; use utils::{BatchQueue, EventBatch, process_rmp_file, validate_snapshot_consistency}; +use crate::{ + HL_NODE, + listeners::{directory::DirectoryListener, order_book::state::OrderBookState}, + order_book::{ + Coin, Snapshot, + multi_book::{Snapshots, load_snapshots_from_json}, + }, + prelude::*, + types::{ + L4Order, + inner::{InnerL4Order, InnerLevel}, + node_data::{Batch, EventSource, NodeDataFill, NodeDataOrderDiff, NodeDataOrderStatus}, + }, +}; + mod state; mod utils; // WARNING - this code assumes no other file system operations are occurring in the watched directories // if there are scripts running, this may not work as intended -pub(crate) async fn hl_listen(listener: Arc>, dir: PathBuf, inactivity_exit_secs: u64) -> Result<()> { +pub(crate) async fn hl_listen( + listener: Arc>, + dir: PathBuf, + inactivity_exit_secs: u64, +) -> Result<()> { let order_statuses_dir = EventSource::OrderStatuses.event_source_dir(&dir).canonicalize()?; let fills_dir = EventSource::Fills.event_source_dir(&dir).canonicalize()?; let order_diffs_dir = EventSource::OrderDiffs.event_source_dir(&dir).canonicalize()?; @@ -433,9 +439,7 @@ impl DirectoryListener for OrderBookListener { ); #[allow(clippy::unwrap_used)] let total_len: i64 = total_len.try_into().unwrap(); - self.file_mut(event_source) - .as_mut() - .map(|f| f.seek_relative(-total_len)); + self.file_mut(event_source).as_mut().map(|f| f.seek_relative(-total_len)); break; } }; diff --git a/server/src/listeners/order_book/state.rs b/server/src/listeners/order_book/state.rs index 3fb74ae2..067f8c3d 100644 --- a/server/src/listeners/order_book/state.rs +++ b/server/src/listeners/order_book/state.rs @@ -1,3 +1,5 @@ +use std::collections::{HashMap, HashSet, VecDeque}; + use crate::{ listeners::order_book::{L2Snapshots, TimedSnapshots, utils::compute_l2_snapshots}, order_book::{ @@ -10,7 +12,6 @@ use crate::{ node_data::{Batch, NodeDataOrderDiff, NodeDataOrderStatus}, }, }; -use std::collections::{HashMap, HashSet, VecDeque}; #[derive(Clone)] pub(super) struct OrderBookState { diff --git a/server/src/listeners/order_book/utils.rs b/server/src/listeners/order_book/utils.rs index 8c0893b1..831e01df 100644 --- a/server/src/listeners/order_book/utils.rs +++ b/server/src/listeners/order_book/utils.rs @@ -1,3 +1,12 @@ +use std::{ + collections::{HashMap, VecDeque}, + path::{Path, PathBuf}, +}; + +use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; +use reqwest::Client; +use serde_json::json; + use crate::{ listeners::order_book::{L2SnapshotParams, L2Snapshots}, order_book::{ @@ -12,14 +21,6 @@ use crate::{ }, }; use log::info; -use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; -use reqwest::Client; -use serde_json::json; -use std::collections::VecDeque; -use std::{ - collections::HashMap, - path::{Path, PathBuf}, -}; use tokio::fs; /// Fetches an L4 snapshot and writes it to `out.json` in the given directory. diff --git a/server/src/order_book/levels.rs b/server/src/order_book/levels.rs index 23aad1e5..d4a6c9cf 100644 --- a/server/src/order_book/levels.rs +++ b/server/src/order_book/levels.rs @@ -1,8 +1,10 @@ -use crate::order_book::{InnerOrder, Oid, OrderBook, Px, Side, Snapshot, Sz, linked_list::LinkedList}; -use crate::types::Level; -use crate::types::inner::InnerLevel; use std::collections::BTreeMap; +use crate::{ + order_book::{InnerOrder, Oid, OrderBook, Px, Side, Snapshot, Sz, linked_list::LinkedList}, + types::{Level, inner::InnerLevel}, +}; + #[must_use] fn bucket(px: Px, side: Side, n_sig_figs: Option, mantissa: Option) -> Px { let m = mantissa.unwrap_or(1); diff --git a/server/src/order_book/linked_list.rs b/server/src/order_book/linked_list.rs index d4e4de14..eeca8026 100644 --- a/server/src/order_book/linked_list.rs +++ b/server/src/order_book/linked_list.rs @@ -1,7 +1,9 @@ -use crate::prelude::*; -use slab::Slab; use std::{collections::HashMap, hash::Hash, marker::PhantomData}; +use slab::Slab; + +use crate::prelude::*; + #[derive(Clone)] struct Node { key: K, @@ -145,10 +147,12 @@ impl LinkedList { #[cfg(test)] mod tests { - use super::*; - use itertools::Itertools; use std::collections::VecDeque; + use itertools::Itertools; + + use super::*; + #[must_use] fn to_rev_vec(list: &LinkedList) -> Vec<&T> { let mut res = Vec::new(); diff --git a/server/src/order_book/mod.rs b/server/src/order_book/mod.rs index 908adadc..45d64fb1 100644 --- a/server/src/order_book/mod.rs +++ b/server/src/order_book/mod.rs @@ -1,7 +1,9 @@ -use crate::prelude::*; +use std::collections::{BTreeMap, HashMap, HashSet}; + use itertools::Itertools; use linked_list::LinkedList; -use std::collections::{BTreeMap, HashMap, HashSet}; + +use crate::prelude::*; pub(crate) mod levels; mod linked_list; @@ -181,10 +183,10 @@ fn match_order(maker_orders: &mut BTreeMap #[cfg(test)] mod tests { - use crate::order_book::types::{Coin, Sz}; + use std::collections::BTreeSet; use super::*; - use std::collections::BTreeSet; + use crate::order_book::types::{Coin, Sz}; #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] struct MinimalOrder { diff --git a/server/src/order_book/multi_book.rs b/server/src/order_book/multi_book.rs index 6ef7746c..353dbc76 100644 --- a/server/src/order_book/multi_book.rs +++ b/server/src/order_book/multi_book.rs @@ -1,15 +1,17 @@ -use crate::{ - order_book::{Coin, InnerOrder, Oid, OrderBook, Snapshot, Sz}, - prelude::*, -}; -use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; -use serde::{Deserialize, Serialize}; use std::{ collections::{BTreeMap, HashMap}, path::Path, }; + +use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; +use serde::{Deserialize, Serialize}; use tokio::fs::read_to_string; +use crate::{ + order_book::{Coin, InnerOrder, Oid, OrderBook, Snapshot, Sz}, + prelude::*, +}; + pub(crate) struct Snapshots(HashMap>); impl Snapshots { @@ -102,6 +104,12 @@ where #[cfg(test)] mod tests { + use std::fs::create_dir_all; + + use alloy::primitives::Address; + use itertools::Itertools; + use tempfile::tempdir; + use crate::{ order_book::{ InnerOrder, OrderBook, Px, Side, Snapshot, Sz, @@ -114,10 +122,6 @@ mod tests { inner::{InnerL4Order, InnerLevel}, }, }; - use alloy::primitives::Address; - use itertools::Itertools; - use std::fs::create_dir_all; - use tempfile::tempdir; #[must_use] fn snapshot_to_l2_snapshot( diff --git a/server/src/order_book/types.rs b/server/src/order_book/types.rs index 447eb3f5..a36db079 100644 --- a/server/src/order_book/types.rs +++ b/server/src/order_book/types.rs @@ -1,7 +1,11 @@ -use crate::prelude::*; +use std::{ + fmt::{Debug, Formatter}, + ops::Add, +}; + use serde::{Deserialize, Serialize}; -use std::fmt::{Debug, Formatter}; -use std::ops::Add; + +use crate::prelude::*; #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] pub(crate) enum Side { diff --git a/server/src/servers/websocket_server.rs b/server/src/servers/websocket_server.rs index acf7d0da..a529c620 100644 --- a/server/src/servers/websocket_server.rs +++ b/server/src/servers/websocket_server.rs @@ -1,27 +1,15 @@ -use crate::{ - listeners::order_book::{ - InternalMessage, L2SnapshotParams, L2Snapshots, OrderBookListener, TimedSnapshots, hl_listen, - }, - order_book::{Coin, Snapshot}, - prelude::*, - types::{ - L2Book, L4Book, L4BookUpdates, L4Order, Trade, - inner::InnerLevel, - node_data::{Batch, NodeDataFill, NodeDataOrderDiff, NodeDataOrderStatus}, - subscription::{ClientMessage, DEFAULT_LEVELS, ServerResponse, Subscription, SubscriptionManager}, - }, -}; -use axum::{Router, response::IntoResponse, routing::get}; -use futures_util::{SinkExt, StreamExt}; -use log::{error, info}; use std::{ collections::{HashMap, HashSet}, env::home_dir, sync::Arc, }; -use tokio::select; + +use axum::{Router, response::IntoResponse, routing::get}; +use futures_util::{SinkExt, StreamExt}; +use log::{error, info}; use tokio::{ net::TcpListener, + select, sync::{ Mutex, broadcast::{Sender, channel}, @@ -29,7 +17,26 @@ use tokio::{ }; use yawc::{FrameView, OpCode, WebSocket}; -pub async fn run_websocket_server(address: &str, ignore_spot: bool, compression_level: u32, inactivity_exit_secs: u64) -> Result<()> { +use crate::{ + listeners::order_book::{ + InternalMessage, L2SnapshotParams, L2Snapshots, OrderBookListener, TimedSnapshots, hl_listen, + }, + order_book::{Coin, Snapshot}, + prelude::*, + types::{ + L2Book, L4Book, L4BookUpdates, L4Order, Trade, + inner::InnerLevel, + node_data::{Batch, NodeDataFill, NodeDataOrderDiff, NodeDataOrderStatus}, + subscription::{ClientMessage, DEFAULT_LEVELS, ServerResponse, Subscription, SubscriptionManager}, + }, +}; + +pub async fn run_websocket_server( + address: &str, + ignore_spot: bool, + compression_level: u32, + inactivity_exit_secs: u64, +) -> Result<()> { let (internal_message_tx, _) = channel::>(100); // Central task: listen to messages and forward them for distribution diff --git a/server/src/types/subscription.rs b/server/src/types/subscription.rs index 64dbd548..6701811d 100644 --- a/server/src/types/subscription.rs +++ b/server/src/types/subscription.rs @@ -1,7 +1,9 @@ -use crate::types::{L2Book, L4Book, Trade}; +use std::collections::HashSet; + use log::info; use serde::{Deserialize, Serialize}; -use std::collections::HashSet; + +use crate::types::{L2Book, L4Book, Trade}; const MAX_LEVELS: usize = 100; pub(crate) const DEFAULT_LEVELS: usize = 20; @@ -105,9 +107,8 @@ impl SubscriptionManager { #[cfg(test)] mod test { - use crate::types::subscription::Subscription; - use super::{ClientMessage, ServerResponse}; + use crate::types::subscription::Subscription; #[test] fn test_message_deserialization_subscription_response() { From 900ae29aba1e35c4241ac79f34eccf47a0bc8e63 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 08:58:55 +0100 Subject: [PATCH 16/29] fix(linter): various --- server/src/listeners/directory.rs | 25 ++++++++--- server/src/listeners/order_book/mod.rs | 52 +++++++++++------------ server/src/listeners/order_book/utils.rs | 8 ++-- server/src/servers/websocket_server.rs | 54 +++++++++++++----------- server/src/types/subscription.rs | 8 ++-- 5 files changed, 83 insertions(+), 64 deletions(-) diff --git a/server/src/listeners/directory.rs b/server/src/listeners/directory.rs index 0396a8d1..648f3237 100644 --- a/server/src/listeners/directory.rs +++ b/server/src/listeners/directory.rs @@ -40,7 +40,12 @@ mod tests { use notify::{RecursiveMode, Watcher, recommended_watcher}; use rand::{Rng, SeedableRng, rngs::StdRng}; use tempfile::tempdir; - use tokio::{fs::File as TokioFile, io::AsyncWriteExt, sync::mpsc::unbounded_channel, time::sleep}; + use tokio::{ + fs::File as TokioFile, + io::AsyncWriteExt, + sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}, + time::sleep, + }; use crate::{ listeners::directory::{DirectoryListener, EventSource}, @@ -150,7 +155,11 @@ mod tests { Ok(()) } - async fn create_mock_data(event_source: EventSource, mock_dir: &Path) -> Result { + async fn create_mock_data( + event_source: EventSource, + mock_dir: &Path, + ready_rx: &mut UnboundedReceiver<()>, + ) -> Result { // set up so that the directory is initially empty let mut res = String::new(); sleep(Duration::from_millis(100)).await; @@ -165,6 +174,7 @@ mod tests { res += data; let lines = data.split_whitespace(); let mut mock_file = TokioFile::create(mock_dir.join((i + 1).to_string())).await?; + ready_rx.recv().await.ok_or("Listener readiness channel closed")?; for line in lines { mock_file.write_all((line.to_string() + "\n").as_bytes()).await?; mock_file.flush().await?; @@ -182,6 +192,7 @@ mod tests { struct TestListener { file: Option, history: Arc>, + ready_tx: UnboundedSender<()>, } impl DirectoryListener for TestListener { @@ -196,6 +207,7 @@ mod tests { fn on_file_creation(&mut self, new_file: PathBuf, _event_source: EventSource) -> Result<()> { let file = File::open(new_file)?; self.file = Some(file); + let _unused = self.ready_tx.send(()); Ok(()) } @@ -208,8 +220,8 @@ mod tests { } impl TestListener { - fn new(history: Arc>) -> Self { - Self { file: None, history } + fn new(history: Arc>, ready_tx: UnboundedSender<()>) -> Self { + Self { file: None, history, ready_tx } } } @@ -222,7 +234,8 @@ mod tests { let event_source = EventSource::Fills; create_dir_all(event_source.event_source_dir(&mock_path))?; let history = Arc::new(Mutex::new(String::new())); - let mut test_listener = TestListener::new(history.clone()); + let (ready_tx, mut ready_rx) = unbounded_channel(); + let mut test_listener = TestListener::new(history.clone(), ready_tx); { let mock_path = mock_path.clone(); tokio::spawn(async move { @@ -233,7 +246,7 @@ mod tests { } // get desired output - let expected = create_mock_data(event_source, &mock_path).await?; + let expected = create_mock_data(event_source, &mock_path, &mut ready_rx).await?; sleep(Duration::from_secs(2)).await; let history = history.lock().unwrap(); assert_eq!(*history, expected); diff --git a/server/src/listeners/order_book/mod.rs b/server/src/listeners/order_book/mod.rs index c3e093df..3c593223 100644 --- a/server/src/listeners/order_book/mod.rs +++ b/server/src/listeners/order_book/mod.rs @@ -291,25 +291,25 @@ impl OrderBookListener { } } } - if self.is_ready() { - if let Some((order_statuses, order_diffs)) = self.pop_cache() { - self.order_book_state - .as_mut() - .map(|book| book.apply_updates(order_statuses.clone(), order_diffs.clone())) - .transpose()?; - if let Some(cache) = &mut self.fetched_snapshot_cache { - cache.push_back((order_statuses.clone(), order_diffs.clone())); - } - if let Some(tx) = &self.internal_message_tx { - let tx = tx.clone(); - tokio::spawn(async move { - let updates = Arc::new(InternalMessage::L4BookUpdates { - diff_batch: order_diffs, - status_batch: order_statuses, - }); - let _unused = tx.send(updates); + if self.is_ready() + && let Some((order_statuses, order_diffs)) = self.pop_cache() + { + self.order_book_state + .as_mut() + .map(|book| book.apply_updates(order_statuses.clone(), order_diffs.clone())) + .transpose()?; + if let Some(cache) = &mut self.fetched_snapshot_cache { + cache.push_back((order_statuses.clone(), order_diffs.clone())); + } + if let Some(tx) = &self.internal_message_tx { + let tx = tx.clone(); + tokio::spawn(async move { + let updates = Arc::new(InternalMessage::L4BookUpdates { + diff_batch: order_diffs, + status_batch: order_statuses, }); - } + let _unused = tx.send(updates); + }); } } Ok(()) @@ -452,14 +452,14 @@ impl DirectoryListener for OrderBookListener { } } let snapshot = self.l2_snapshots(true); - if let Some(snapshot) = snapshot { - if let Some(tx) = &self.internal_message_tx { - let tx = tx.clone(); - tokio::spawn(async move { - let snapshot = Arc::new(InternalMessage::Snapshot { l2_snapshots: snapshot.1, time: snapshot.0 }); - let _unused = tx.send(snapshot); - }); - } + if let Some(snapshot) = snapshot + && let Some(tx) = &self.internal_message_tx + { + let tx = tx.clone(); + tokio::spawn(async move { + let snapshot = Arc::new(InternalMessage::Snapshot { l2_snapshots: snapshot.1, time: snapshot.0 }); + let _unused = tx.send(snapshot); + }); } Ok(()) } diff --git a/server/src/listeners/order_book/utils.rs b/server/src/listeners/order_book/utils.rs index 831e01df..5dc4c762 100644 --- a/server/src/listeners/order_book/utils.rs +++ b/server/src/listeners/order_book/utils.rs @@ -234,10 +234,10 @@ impl BatchQueue { } pub(super) fn push(&mut self, block: Batch) -> bool { - if let Some(last_ts) = self.last_ts { - if last_ts >= block.block_number() { - return false; - } + if let Some(last_ts) = self.last_ts + && last_ts >= block.block_number() + { + return false; } self.last_ts = Some(block.block_number()); self.deque.push_back(block); diff --git a/server/src/servers/websocket_server.rs b/server/src/servers/websocket_server.rs index a529c620..8d45e864 100644 --- a/server/src/servers/websocket_server.rs +++ b/server/src/servers/websocket_server.rs @@ -4,7 +4,7 @@ use std::{ sync::Arc, }; -use axum::{Router, response::IntoResponse, routing::get}; +use axum::{Router, http::StatusCode, response::{IntoResponse, Response}, routing::get}; use futures_util::{SinkExt, StreamExt}; use log::{error, info}; use tokio::{ @@ -85,8 +85,14 @@ fn ws_handler( listener: Arc>, ignore_spot: bool, websocket_opts: yawc::Options, -) -> impl IntoResponse { - let (resp, fut) = incoming.upgrade(websocket_opts).unwrap(); +) -> Response { + let (resp, fut) = match incoming.upgrade(websocket_opts) { + Ok(ok) => ok, + Err(err) => { + error!("failed to start websocket upgrade: {err}"); + return StatusCode::BAD_REQUEST.into_response(); + } + }; tokio::spawn(async move { let ws = match fut.await { Ok(ok) => ok, @@ -96,10 +102,10 @@ fn ws_handler( } }; - handle_socket(ws, internal_message_tx, listener, ignore_spot).await + handle_socket(ws, internal_message_tx, listener, ignore_spot).await; }); - resp + resp.into_response() } async fn handle_socket( @@ -288,15 +294,15 @@ fn coin_to_trades(batch: &Batch) -> HashMap> { while fills.len() >= 2 { let f2 = fills.pop(); let f1 = fills.pop(); - if let Some(f1) = f1 { - if let Some(f2) = f2 { - let mut fills = HashMap::new(); - fills.insert(f1.1.side, f1); - fills.insert(f2.1.side, f2); - let trade = Trade::from_fills(fills); - let coin = trade.coin.clone(); - trades.entry(coin).or_insert_with(Vec::new).push(trade); - } + if let Some(f1) = f1 + && let Some(f2) = f2 + { + let mut fills = HashMap::new(); + fills.insert(f1.1.side, f1); + fills.insert(f2.1.side, f2); + let trade = Trade::from_fills(fills); + let coin = trade.coin.clone(); + trades.entry(coin).or_insert_with(Vec::new).push(trade); } } for list in trades.values_mut() { @@ -330,11 +336,11 @@ async fn send_ws_data_from_book_updates( subscription: &Subscription, book_updates: &mut HashMap, ) { - if let Subscription::L4Book { coin } = subscription { - if let Some(updates) = book_updates.remove(coin) { - let msg = ServerResponse::L4Book(L4Book::Updates(updates)); - send_socket_message(socket, msg).await; - } + if let Subscription::L4Book { coin } = subscription + && let Some(updates) = book_updates.remove(coin) + { + let msg = ServerResponse::L4Book(L4Book::Updates(updates)); + send_socket_message(socket, msg).await; } } @@ -343,11 +349,11 @@ async fn send_ws_data_from_trades( subscription: &Subscription, trades: &mut HashMap>, ) { - if let Subscription::Trades { coin } = subscription { - if let Some(trades) = trades.remove(coin) { - let msg = ServerResponse::Trades(trades); - send_socket_message(socket, msg).await; - } + if let Subscription::Trades { coin } = subscription + && let Some(trades) = trades.remove(coin) + { + let msg = ServerResponse::Trades(trades); + send_socket_message(socket, msg).await; } } diff --git a/server/src/types/subscription.rs b/server/src/types/subscription.rs index 6701811d..71eaa2db 100644 --- a/server/src/types/subscription.rs +++ b/server/src/types/subscription.rs @@ -51,10 +51,10 @@ impl Subscription { info!("Invalid subscription: sig figs aren't set correctly"); return false; } - if let Some(m) = *mantissa { - if n_sig_figs < 5 || (m != 5 && m != 2) { - return false; - } + if let Some(m) = *mantissa + && (n_sig_figs < 5 || (m != 5 && m != 2)) + { + return false; } } else if mantissa.is_some() { info!("Invalid subscription: mantissa can not be some if sig figs are not set"); From 4273ab3c5d07b2e0c451c42debc2395464f0f9eb Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 09:24:03 +0100 Subject: [PATCH 17/29] ci: update test calls --- .github/workflows/ci-nightly.yml | 2 +- .github/workflows/ci-tests.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci-nightly.yml b/.github/workflows/ci-nightly.yml index 3e2e4aad..7204336e 100644 --- a/.github/workflows/ci-nightly.yml +++ b/.github/workflows/ci-nightly.yml @@ -47,7 +47,7 @@ jobs: - uses: dtolnay/rust-toolchain@nightly - run: | for i in {1..16}; do - cargo test -Z unstable-options --workspace --all-features -- --shuffle + cargo test --workspace --all-features -- -Z unstable-options --shuffle done summary: diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml index b2548ccd..3a05d1ec 100644 --- a/.github/workflows/ci-tests.yml +++ b/.github/workflows/ci-tests.yml @@ -29,7 +29,7 @@ jobs: steps: - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@nightly - - run: cargo test -Z unstable-options --workspace --all-features -- --shuffle + - run: cargo test --workspace --all-features -- -Z unstable-options --shuffle summary: name: CI Summary From 744e1433ada0cab45b4bcab80d63db8988467d51 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 09:29:12 +0100 Subject: [PATCH 18/29] docs: AGENTS.md + NEW.md updates --- AGENTS.md | 34 +++++++++++++++++++++------------- NEW.md | 4 +++- 2 files changed, 24 insertions(+), 14 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 38b68b0b..a81c9e71 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -7,17 +7,22 @@ Comprehensive guidance for AI assistants and developers working in this reposito ```bash # Build and lint cargo build --workspace -cargo clippy --workspace # Must pass before commits +# Must pass before commits +cargo clippy --workspace --all-targets --all-features -- -D warnings # Run all tests cargo test --workspace +# Run deeper tests (requires nightly) +cargo test --workspace --all-features -- -Z unstable-options --shuffle # Run specific tests cargo test -p server # Library crate only cargo test test_trade_listener # Single test by name # Format code (max width 120) -cargo fmt +cargo fmt --all +# Check only +cargo fmt --all -- --check # Run the server RUST_LOG=info cargo run --release --bin websocket_server -- \ @@ -203,11 +208,12 @@ mod tests { ### Running Tests ```bash -cargo test --workspace # All tests -cargo test -p server # Library only -cargo test test_order_book_add # Single test -cargo test -- --nocapture # Show println! output -RUST_LOG=debug cargo test # With logging +cargo test --workspace # All tests +cargo test --workspace --all-features -- -Z unstable-options --shuffle # Deeper tests (nightly) +cargo test -p server # Library only +cargo test test_order_book_add # Single test +cargo test -- --nocapture # Show println! output +RUST_LOG=debug cargo test # With logging ``` ## Development Workflow @@ -216,21 +222,23 @@ RUST_LOG=debug cargo test # With logging 1. Read existing code to understand patterns before modifying 2. Maintain consistency with current architecture (generic traits, error types, async patterns) -3. Ensure clippy passes: `cargo clippy --workspace` -4. Format code: `cargo fmt` -5. Run relevant tests: `cargo test -p server` or `cargo test ` +3. Ensure clippy passes: `cargo clippy --workspace --all-targets --all-features -- -D warnings` +4. Format code: `cargo fmt --all` +5. Run relevant tests as per [Running Tests](#running-tests) 6. Update documentation if public API changes ### Commit Guidelines +Assume the user will be handling any git interactions. If you are asked to help: + +- Use conventional commit prefixes - Use short, imperative messages: "fix panic in orderbook listener" -- Optional type prefix: `fix:`, `feat:`, `test:`, `docs:` - Keep commits focused on single behavior change - Example history: ``` fix: panic in orderbook listener - use yawc as default WebSocket crate supporting compression - update docs + feat: use yawc as default WebSocket crate supporting compression + docs: update readme ``` ### Pull Request Guidelines diff --git a/NEW.md b/NEW.md index b0df6bca..02240e84 100644 --- a/NEW.md +++ b/NEW.md @@ -356,10 +356,12 @@ Clients send JSON messages to subscribe: cargo build --workspace # Run linter (required before commits) -cargo clippy --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings # Run tests cargo test --workspace +# Run deeper tests (nightly, shuffled) +cargo test --workspace --all-features -- -Z unstable-options --shuffle # Run server (requires node running) RUST_LOG=info cargo run --release --bin websocket_server -- \ From 0c2078b6806d20c6d12a739fd8005cf9a05faf2f Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 09:38:51 +0100 Subject: [PATCH 19/29] style: cargo fmt --- server/src/servers/websocket_server.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/server/src/servers/websocket_server.rs b/server/src/servers/websocket_server.rs index 8d45e864..4a47291a 100644 --- a/server/src/servers/websocket_server.rs +++ b/server/src/servers/websocket_server.rs @@ -4,7 +4,12 @@ use std::{ sync::Arc, }; -use axum::{Router, http::StatusCode, response::{IntoResponse, Response}, routing::get}; +use axum::{ + Router, + http::StatusCode, + response::{IntoResponse, Response}, + routing::get, +}; use futures_util::{SinkExt, StreamExt}; use log::{error, info}; use tokio::{ From da8428431ea280bbe1c4ba0e39da91a9883cf4f1 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 09:50:17 +0100 Subject: [PATCH 20/29] ci: add git hook pre-commit --- .githooks/pre-commit | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100755 .githooks/pre-commit diff --git a/.githooks/pre-commit b/.githooks/pre-commit new file mode 100755 index 00000000..9ec85804 --- /dev/null +++ b/.githooks/pre-commit @@ -0,0 +1,9 @@ +#!/bin/sh + +echo "> Running pre-commit hook (fmt, lint)" + +cargo fmt --all -- --check +cargo clippy --workspace --all-targets --all-features -- -D warnings + +echo "> pre-commit hook done" +exit 0 From 4829c75699a181fd84087df7e2b633f2d0402a37 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 09:51:04 +0100 Subject: [PATCH 21/29] docs: README.md and NEW.md updates --- NEW.md | 22 +++++++++++++++- README.md | 76 +++++++++++++++++++++++++++++++++++++++++++++++-------- 2 files changed, 86 insertions(+), 12 deletions(-) diff --git a/NEW.md b/NEW.md index 02240e84..43ef00ff 100644 --- a/NEW.md +++ b/NEW.md @@ -25,6 +25,24 @@ This server streams real-time order book data from a [Hyperliquid](https://hyper --- +## API Overview + +Clients connect to `ws://:/ws` and send JSON messages using a simple subscribe envelope: + +```json +{"method":"subscribe","subscription":{"type":"l2Book","coin":"ETH"}} +{"method":"subscribe","subscription":{"type":"trades","coin":"BTC"}} +{"method":"subscribe","subscription":{"type":"l4Book","coin":"ETH"}} +``` + +**Subscriptions:** +- `l2Book`: Aggregated levels for a coin. + - Optional fields: `n_levels` (1-100, default 20), `sig_figs`, `mantissa`. +- `trades`: Fill events for a coin. +- `l4Book`: Full order-level book. Sends an initial snapshot, then block-by-block diffs. + +This server follows Hyperliquid’s WebSocket subscription format for `l2book` and `trades`, and adds `l4book` as an extra endpoint. + ## Architecture at a Glance ``` @@ -352,8 +370,10 @@ Clients send JSON messages to subscribe: ### Quick Start ```bash -# Build everything +# Build everything (debug) cargo build --workspace +# Build everything (release) +cargo build --release # Run linter (required before commits) cargo clippy --workspace --all-targets --all-features -- -D warnings diff --git a/README.md b/README.md index cc943983..1812ec31 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,12 @@ -# Local WebSocket Server +# Hyperliquid Orderbook WebSocket Server -## Disclaimer - -This was a standalone project, not written by the Hyperliquid Labs core team. It is made available "as is", without warranty of any kind, express or implied, including but not limited to warranties of merchantability, fitness for a particular purpose, or noninfringement. Use at your own risk. It is intended for educational or illustrative purposes only and may be incomplete, insecure, or incompatible with future systems. No commitment is made to maintain, update, or fix any issues in this repository. +This server provides the `l2book` and `trades` endpoints from [Hyperliquid’s official API](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions), with roughly the same API. -## Functionality +## API -This server provides the `l2book` and `trades` endpoints from [Hyperliquid’s official API](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions), with roughly the same API. +Custom adaptions not included in the [official API](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions): -- The `l2book` subscription now includes an optional field: +- The `l2book` subscription includes an optional field: `n_levels`, which can be up to `100` and defaults to `20`. - This server also introduces a new endpoint: `l4book`. @@ -34,7 +32,30 @@ For a detailed guide with diagrams aimed at developers new to Rust, see [NEW.md] ## Setup -1. Run a non-validating node (from [`hyperliquid-dex/node`](https://github.com/hyperliquid-dex/node)). Requires batching by block. Requires recording fills, order statuses, and raw book diffs. +### Rust + +This project uses nightly Rust for formatting and test shuffling. We recommend `rustup` as the toolchain manager. + +```bash +# Install rustup (see rustup.rs for platform-specific instructions) +rustup toolchain install nightly +rustup component add rustfmt clippy --toolchain nightly + +# Use nightly for this repo +rustup default nightly +``` + +### Git Hooks + +This repo ships a [pre-commit](.githooks/pre-commit) hook for formatting and linting. To enable it: + +```bash +git config core.hooksPath .githooks +``` + +### Running the Server + +1. Run a non-validating node (from [`hyperliquid-dex/node`](https://github.com/hyperliquid-dex/node)). Ensure batching by block is enabled and the node records fills, order statuses, and raw book diffs. 2. Then run this local server: @@ -44,17 +65,50 @@ cargo run --release --bin websocket_server -- --address 0.0.0.0 --port 8000 cargo run --release --bin websocket_server -- --address 0.0.0.0 --port 8000 --inactivity-exit-secs 30 ``` -If this local server does not detect the node writing down any new events, it will automatically exit after some amount of time (default 5 seconds; configurable via `--inactivity-exit-secs `). -In addition, the local server periodically fetches order book snapshots from the node, and compares to its own internal state. If a difference is detected, it will exit. +If this local server does not detect the node writing down any new events, it will automatically exit after some amount of time (default 5 seconds; configurable via `--inactivity-exit-secs `). In addition, the local server periodically fetches order book snapshots from the node, and compares to its own internal state. If a difference is detected, it will exit. If you want logging, prepend the command with `RUST_LOG=info`. The WebSocket server comes with compression built-in. The compression ratio can be tuned using the `--websocket-compression-level` flag. -## Tests +## Development + +### Format and Lint + +```bash +# Format +cargo fmt --all + +# Lint +cargo clippy --workspace --all-targets --all-features -- -D warnings +``` + +Note: formatting line length set to max 120 columns in [rustfmt.toml](./rustfmt.toml). + +### Build ```bash +# Build everything (debug) +cargo build --workspace +# Build everything (release) +cargo build --release +``` + +Build artifacts are written under `target/`: +- Debug binaries in `target/debug/` +- Release binaries in `target/release/` + +### Tests + +```bash +# Run all tests cargo test --workspace +# Run deeper tests (requires nightly) +cargo test --workspace --all-features -- -Z unstable-options --shuffle + +# Run specific tests +cargo test -p server # Library crate only +cargo test test_trade_listener # Single test by name ``` Runs all unit tests for the `server` and `binaries` crates. From 324d619fe61ff4085397708a8d181b3b20790465 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 11:05:36 +0100 Subject: [PATCH 22/29] ci: fix sanitizer test calls --- .github/workflows/ci-nightly.yml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci-nightly.yml b/.github/workflows/ci-nightly.yml index 7204336e..e00917b4 100644 --- a/.github/workflows/ci-nightly.yml +++ b/.github/workflows/ci-nightly.yml @@ -24,7 +24,7 @@ jobs: run: | ASAN_OPTIONS=detect_leaks=0 \ RUSTFLAGS="-Z sanitizer=address" \ - cargo +${{ env.RUST_TOOLCHAIN }} test --workspace --all-features -Z sanitizer=address -Z build-std --target x86_64-unknown-linux-gnu + cargo +${{ env.RUST_TOOLCHAIN }} test --workspace --all-features -Z build-std --target x86_64-unknown-linux-gnu sanitizer-tsan: name: Sanitizer (TSan) @@ -37,7 +37,8 @@ jobs: - name: ThreadSanitizer run: | RUSTFLAGS="-Z sanitizer=thread" \ - cargo +${{ env.RUST_TOOLCHAIN }} test --workspace --all-features -Z sanitizer=thread -Z build-std --target x86_64-unknown-linux-gnu + RUSTDOCFLAGS="-Z sanitizer=thread" \ + cargo +${{ env.RUST_TOOLCHAIN }} test --workspace --all-features -Z build-std --target x86_64-unknown-linux-gnu unit-tests-flaky: name: Unit tests (flaky) From d470a49f5b18bb8079538f5a745be54a4ce1f3c0 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 11:11:33 +0100 Subject: [PATCH 23/29] test: fix TSan error --- server/src/listeners/directory.rs | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/server/src/listeners/directory.rs b/server/src/listeners/directory.rs index 648f3237..06607adc 100644 --- a/server/src/listeners/directory.rs +++ b/server/src/listeners/directory.rs @@ -44,7 +44,7 @@ mod tests { fs::File as TokioFile, io::AsyncWriteExt, sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}, - time::sleep, + time::{sleep, timeout}, }; use crate::{ @@ -52,6 +52,7 @@ mod tests { prelude::*, }; + const READY_TIMEOUT: Duration = Duration::from_secs(5); const DATA: [&str; 2] = [ r#"{"coin":"@151","side":"A","time":"2025-06-24T02:56:36.172847427","px":"2393.9","sz":"0.1539","hash":"0x2b21750229be769650b604261eaac1018c00c45812652efbbdd35fe0ecb201a1","trade_dir_override":"Na","side_info":[{"user":"0xecb63caa47c7c4e77f60f1ce858cf28dc2b82b00","start_pos":"1166.565307356","oid":105686971733,"twap_id":null,"cloid":"0x1070fff92506b3ab5e5aec135e5a5ddd"},{"user":"0xb65117c1e1006e7b2413fa90e96fcbe3fa83ed75","start_pos":"0.153928559","oid":105686976226,"twap_id":null,"cloid":null}]} {"coin":"@166","side":"A","time":"2025-06-24T02:56:36.172847427","px":"1.0003","sz":"184.11","hash":"0x0ffc6896b2147680820e04261eaac1018c0101735014e44b56f038478b13ad8f","trade_dir_override":"Na","side_info":[{"user":"0x107332a1729ba0bcf6171117815a87b72a7e6082","start_pos":"36301.55539655","oid":105686050113,"twap_id":null,"cloid":null},{"user":"0xb65117c1e1006e7b2413fa90e96fcbe3fa83ed75","start_pos":"184.12704003","oid":105686976227,"twap_id":null,"cloid":null}]} @@ -82,7 +83,12 @@ mod tests { "#, ]; - async fn listen(listener: &mut L, event_source: EventSource, dir: &Path) -> Result<()> { + async fn listen( + listener: &mut L, + event_source: EventSource, + dir: &Path, + watcher_ready_tx: UnboundedSender<()>, + ) -> Result<()> { let event_source_dir = event_source.event_source_dir(dir).canonicalize()?; info!("Monitoring directory: {}", event_source_dir.display()); // monitoring the directory via the notify crate (gives file system events) @@ -95,6 +101,7 @@ mod tests { })?; watcher.watch(&event_source_dir, RecursiveMode::Recursive)?; + let _unused = watcher_ready_tx.send(()); loop { match fs_event_rx.recv().await { Some(Ok(event)) => { @@ -174,7 +181,9 @@ mod tests { res += data; let lines = data.split_whitespace(); let mut mock_file = TokioFile::create(mock_dir.join((i + 1).to_string())).await?; - ready_rx.recv().await.ok_or("Listener readiness channel closed")?; + let ready = + timeout(READY_TIMEOUT, ready_rx.recv()).await.map_err(|_| "Listener readiness channel timed out")?; + ready.ok_or("Listener readiness channel closed")?; for line in lines { mock_file.write_all((line.to_string() + "\n").as_bytes()).await?; mock_file.flush().await?; @@ -235,16 +244,22 @@ mod tests { create_dir_all(event_source.event_source_dir(&mock_path))?; let history = Arc::new(Mutex::new(String::new())); let (ready_tx, mut ready_rx) = unbounded_channel(); + let (watcher_ready_tx, mut watcher_ready_rx) = unbounded_channel(); let mut test_listener = TestListener::new(history.clone(), ready_tx); { let mock_path = mock_path.clone(); + let watcher_ready_tx = watcher_ready_tx.clone(); tokio::spawn(async move { - if let Err(err) = listen(&mut test_listener, event_source, &mock_path).await { + if let Err(err) = listen(&mut test_listener, event_source, &mock_path, watcher_ready_tx).await { error!("Listener error: {err}"); } }); } + let watcher_ready = + timeout(READY_TIMEOUT, watcher_ready_rx.recv()).await.map_err(|_| "Watcher readiness channel timed out")?; + watcher_ready.ok_or("Watcher readiness channel closed")?; + // get desired output let expected = create_mock_data(event_source, &mock_path, &mut ready_rx).await?; sleep(Duration::from_secs(2)).await; From e33e41e669b559fc7f853f0d121129eac815cd05 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 11:25:15 +0100 Subject: [PATCH 24/29] chore: convenience tool justfile added --- README.md | 21 +++++++++++++++++++++ justfile | 27 +++++++++++++++++++++++++++ 2 files changed, 48 insertions(+) create mode 100644 justfile diff --git a/README.md b/README.md index 1812ec31..e499da78 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,27 @@ rustup component add rustfmt clippy --toolchain nightly rustup default nightly ``` +### justfile + +Optional convenience commands are provided via `just`. Install it once: + +```bash +cargo install just +``` + +Usage examples: + +```bash +just --list # --unsorted (optional) +just build +just test +just fmt +just lint +just clean +just asan +just tsan +``` + ### Git Hooks This repo ships a [pre-commit](.githooks/pre-commit) hook for formatting and linting. To enable it: diff --git a/justfile b/justfile new file mode 100644 index 00000000..93a4a648 --- /dev/null +++ b/justfile @@ -0,0 +1,27 @@ +set shell := ["bash", "-lc"] + +build: + cargo build --release + +test: + cargo test --workspace --all-features -- -Z unstable-options --shuffle + +fmt: + cargo fmt --all + +lint: + cargo clippy --workspace --all-targets --all-features -- -D warnings + +clean: + cargo clean + +asan: + ASAN_OPTIONS=detect_leaks=0 \ + RUSTFLAGS="-Zsanitizer=address" \ + RUSTDOCFLAGS="-Zsanitizer=address" \ + cargo +nightly test --workspace --all-features -Z build-std --target x86_64-unknown-linux-gnu + +tsan: + RUSTFLAGS="-Zsanitizer=thread" \ + RUSTDOCFLAGS="-Zsanitizer=thread" \ + cargo +nightly test --workspace --all-features -Z build-std --target x86_64-unknown-linux-gnu From c7d82eccf33ca7fa85241eb962be2e68fba76c38 Mon Sep 17 00:00:00 2001 From: Jakob Ersson <11717405+jakobilobi@users.noreply.github.com> Date: Tue, 20 Jan 2026 12:44:30 +0100 Subject: [PATCH 25/29] test: fix sanitizer errors --- server/src/listeners/directory.rs | 65 ++++++++++++++++++------------- 1 file changed, 37 insertions(+), 28 deletions(-) diff --git a/server/src/listeners/directory.rs b/server/src/listeners/directory.rs index 06607adc..d06aeff0 100644 --- a/server/src/listeners/directory.rs +++ b/server/src/listeners/directory.rs @@ -29,7 +29,6 @@ pub(crate) trait DirectoryListener { #[cfg(test)] mod tests { use std::{ - io::{Seek, SeekFrom}, path::{Path, PathBuf}, sync::{Arc, Mutex}, time::Duration, @@ -43,7 +42,7 @@ mod tests { use tokio::{ fs::File as TokioFile, io::AsyncWriteExt, - sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}, + sync::mpsc::{UnboundedSender, unbounded_channel}, time::{sleep, timeout}, }; @@ -83,7 +82,10 @@ mod tests { "#, ]; - async fn listen( + // Watch the directory and forward events to the listener. + // We read immediately after create/first modify because notify can emit modify without create, + // especially under sanitizers or slow scheduling. + async fn listen( listener: &mut L, event_source: EventSource, dir: &Path, @@ -111,25 +113,28 @@ mod tests { if new_path.is_file() { info!("-- Event: {} created --", new_path.display()); listener.on_file_creation(new_path.clone(), event_source)?; + // Read immediately to avoid missing the first write. + listener.on_file_modification(event_source)?; } } // Check for `Modify` event (only if the file is already initialized) else if event.kind.is_modify() { let new_path = &event.paths[0]; if new_path.is_file() { - // If we are not tracking anything right now, we treat a file update as declaring that it has been created. - // Unfortunately, we miss the update that occurs at this time step. - // We go to the end of the file to read for updates after that. + // If we are not tracking anything right now, treat the modify event as a creation + // and read immediately so the test doesn't miss the first write. if listener.is_reading(event_source) { + if listener.current_path().is_none_or(|path| path != new_path) { + info!("-- Event: {} created --", new_path.display()); + listener.on_file_creation(new_path.clone(), event_source)?; + } + // Modify can arrive before create; treat it as a read signal. info!("-- Event: {} modified --", new_path.display()); - listener.on_file_modification(event_source)?; } else { info!("-- Event: {} created --", new_path.display()); - let file = listener.file_mut(event_source); - let mut new_file = File::open(new_path)?; - new_file.seek(SeekFrom::End(0))?; - *file = Some(new_file); + listener.on_file_creation(new_path.clone(), event_source)?; } + listener.on_file_modification(event_source)?; } } } @@ -162,11 +167,8 @@ mod tests { Ok(()) } - async fn create_mock_data( - event_source: EventSource, - mock_dir: &Path, - ready_rx: &mut UnboundedReceiver<()>, - ) -> Result { + // Write mock data without per-file readiness waits so sanitizer timing can't deadlock the test. + async fn create_mock_data(event_source: EventSource, mock_dir: &Path) -> Result { // set up so that the directory is initially empty let mut res = String::new(); sleep(Duration::from_millis(100)).await; @@ -181,9 +183,6 @@ mod tests { res += data; let lines = data.split_whitespace(); let mut mock_file = TokioFile::create(mock_dir.join((i + 1).to_string())).await?; - let ready = - timeout(READY_TIMEOUT, ready_rx.recv()).await.map_err(|_| "Listener readiness channel timed out")?; - ready.ok_or("Listener readiness channel closed")?; for line in lines { mock_file.write_all((line.to_string() + "\n").as_bytes()).await?; mock_file.flush().await?; @@ -197,11 +196,16 @@ mod tests { Ok(res) } - // will listen to file events and collect their results in the history field + // Test listener needs to expose the current path to handle out-of-order events. + trait TestDirectoryListener: DirectoryListener { + fn current_path(&self) -> Option<&Path>; + } + + // Test listener used to validate filesystem event handling. struct TestListener { file: Option, history: Arc>, - ready_tx: UnboundedSender<()>, + current_path: Option, } impl DirectoryListener for TestListener { @@ -214,9 +218,9 @@ mod tests { } fn on_file_creation(&mut self, new_file: PathBuf, _event_source: EventSource) -> Result<()> { - let file = File::open(new_file)?; + let file = File::open(&new_file)?; self.file = Some(file); - let _unused = self.ready_tx.send(()); + self.current_path = Some(new_file); Ok(()) } @@ -228,9 +232,15 @@ mod tests { } } + impl TestDirectoryListener for TestListener { + fn current_path(&self) -> Option<&Path> { + self.current_path.as_deref() + } + } + impl TestListener { - fn new(history: Arc>, ready_tx: UnboundedSender<()>) -> Self { - Self { file: None, history, ready_tx } + fn new(history: Arc>) -> Self { + Self { file: None, history, current_path: None } } } @@ -243,9 +253,8 @@ mod tests { let event_source = EventSource::Fills; create_dir_all(event_source.event_source_dir(&mock_path))?; let history = Arc::new(Mutex::new(String::new())); - let (ready_tx, mut ready_rx) = unbounded_channel(); let (watcher_ready_tx, mut watcher_ready_rx) = unbounded_channel(); - let mut test_listener = TestListener::new(history.clone(), ready_tx); + let mut test_listener = TestListener::new(history.clone()); { let mock_path = mock_path.clone(); let watcher_ready_tx = watcher_ready_tx.clone(); @@ -261,7 +270,7 @@ mod tests { watcher_ready.ok_or("Watcher readiness channel closed")?; // get desired output - let expected = create_mock_data(event_source, &mock_path, &mut ready_rx).await?; + let expected = create_mock_data(event_source, &mock_path).await?; sleep(Duration::from_secs(2)).await; let history = history.lock().unwrap(); assert_eq!(*history, expected); From 16015beea5786a4d8713860720394047370f0b4b Mon Sep 17 00:00:00 2001 From: yangwao Date: Sun, 8 Mar 2026 20:50:11 +0100 Subject: [PATCH 26/29] feat: add support for spot markets via --include-spot-unsafe Handles special addresses (Assistance Fund 0xFE, HIP-2 0xFF) that don't emit order_status events by creating synthetic Alo Limit order entries. Removes the spot coin subscription filter. Opt-in via --include-spot-unsafe CLI flag since the order type assumption is an approximation. Cherry-picked from hyperliquid-dex/order_book_server#10. Co-Authored-By: Claude Opus 4.6 --- binaries/src/bin/websocket_server.rs | 10 +++++++++- server/src/listeners/order_book/state.rs | 22 +++++++++++++++++++++- server/src/order_book/types.rs | 3 +++ server/src/types/node_data.rs | 18 +++++++++++++++++- server/src/types/subscription.rs | 4 ++-- 5 files changed, 52 insertions(+), 5 deletions(-) diff --git a/binaries/src/bin/websocket_server.rs b/binaries/src/bin/websocket_server.rs index ad855393..c11a8752 100644 --- a/binaries/src/bin/websocket_server.rs +++ b/binaries/src/bin/websocket_server.rs @@ -26,6 +26,13 @@ struct Args { #[arg(long)] websocket_compression_level: Option, + /// Enable to include orders from spot markets. This is "unsafe" because order statuses for new + /// orders from special addresses (e.g. the assistance fund and HIP-2) are not emitted so we + /// **unsafely** assume that these orders are all "Alo" limit orders. Default is false, meaning + /// that spot orders are ignored. + #[arg(long)] + include_spot_unsafe: Option, + /// Inactivity timeout in seconds before server exits. /// If no node events are observed for this duration, the process exits. /// Default and minimum is 5 seconds. @@ -42,9 +49,10 @@ async fn main() -> Result<()> { let full_address = format!("{}:{}", args.address, args.port); println!("Running websocket server on {full_address}"); + let ignore_spot = !args.include_spot_unsafe.unwrap_or(false); let compression_level = args.websocket_compression_level.unwrap_or(/* Some compression */ 1); let inactivity_exit_secs = args.inactivity_exit_secs.unwrap_or(5).max(5); - run_websocket_server(&full_address, true, compression_level, inactivity_exit_secs).await?; + run_websocket_server(&full_address, ignore_spot, compression_level, inactivity_exit_secs).await?; Ok(()) } diff --git a/server/src/listeners/order_book/state.rs b/server/src/listeners/order_book/state.rs index 067f8c3d..c31c5805 100644 --- a/server/src/listeners/order_book/state.rs +++ b/server/src/listeners/order_book/state.rs @@ -3,7 +3,7 @@ use std::collections::{HashMap, HashSet, VecDeque}; use crate::{ listeners::order_book::{L2Snapshots, TimedSnapshots, utils::compute_l2_snapshots}, order_book::{ - Coin, InnerOrder, Oid, + Coin, InnerOrder, Oid, Px, multi_book::{OrderBooks, Snapshots}, }, prelude::*, @@ -105,6 +105,26 @@ impl OrderBookState { #[allow(clippy::unwrap_used)] inner_order.convert_trigger(time.try_into().unwrap()); self.order_book.add_order(inner_order); + } else if diff.special_address() { + // Assume all orders from special addresses are Alo, Limit orders + let inner_order = InnerL4Order { + user: diff.user(), + coin, + side: diff.side(), + limit_px: Px::parse_from_str(diff.px().as_str())?, + sz, + oid: oid.value(), + timestamp: time, + trigger_condition: "N/A".to_string(), + is_trigger: false, + trigger_px: "0.0".to_string(), + is_position_tpsl: false, + reduce_only: false, + order_type: "Limit".to_string(), + tif: Some("Alo".to_string()), + cloid: None, + }; + self.order_book.add_order(inner_order); } else { return Err(format!("Unable to find order opening status {diff:?}").into()); } diff --git a/server/src/order_book/types.rs b/server/src/order_book/types.rs index a36db079..348cb7cf 100644 --- a/server/src/order_book/types.rs +++ b/server/src/order_book/types.rs @@ -58,6 +58,9 @@ impl Oid { pub(crate) const fn new(value: u64) -> Self { Self(value) } + pub(crate) const fn value(self) -> u64 { + self.0 + } } pub(crate) trait InnerOrder: Clone { diff --git a/server/src/types/node_data.rs b/server/src/types/node_data.rs index acb6ed22..99590003 100644 --- a/server/src/types/node_data.rs +++ b/server/src/types/node_data.rs @@ -5,14 +5,18 @@ use chrono::NaiveDateTime; use serde::{Deserialize, Serialize}; use crate::{ - order_book::{Coin, Oid}, + order_book::{Coin, Oid, Side}, types::{Fill, L4Order, OrderDiff}, }; +const ASSISTANCE_FUND: Address = Address::repeat_byte(0xFE); +const HIP_2: Address = Address::repeat_byte(0xFF); + #[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct NodeDataOrderDiff { user: Address, oid: u64, + side: Side, px: String, coin: String, pub(crate) raw_book_diff: OrderDiff, @@ -29,6 +33,18 @@ impl NodeDataOrderDiff { pub(crate) fn coin(&self) -> Coin { Coin::new(&self.coin) } + pub(crate) fn user(&self) -> Address { + self.user + } + pub(crate) fn side(&self) -> Side { + self.side + } + pub(crate) fn px(&self) -> String { + self.px.clone() + } + pub(crate) fn special_address(&self) -> bool { + self.user == ASSISTANCE_FUND || self.user == HIP_2 + } } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/server/src/types/subscription.rs b/server/src/types/subscription.rs index 71eaa2db..0fde738b 100644 --- a/server/src/types/subscription.rs +++ b/server/src/types/subscription.rs @@ -33,7 +33,7 @@ impl Subscription { match self { Self::Trades { coin } => universe.contains(coin), Self::L2Book { coin, n_sig_figs, n_levels, mantissa } => { - if !universe.contains(coin) || coin.starts_with('@') { + if !universe.contains(coin) { info!("Invalid subscription: coin not found"); return false; } @@ -64,7 +64,7 @@ impl Subscription { true } Self::L4Book { coin } => { - if !universe.contains(coin) || coin.starts_with('@') { + if !universe.contains(coin) { info!("Invalid subscription: coin not found"); return false; } From e024f3b2b83fd6dd890ba42d4d90a4387a62f275 Mon Sep 17 00:00:00 2001 From: yangwao Date: Sun, 8 Mar 2026 21:42:07 +0100 Subject: [PATCH 27/29] fix: use shared volume path for fileSnapshot API output When running in separate Docker containers, the node and OBS have isolated filesystems. The fileSnapshot API writes to an outPath on the node's filesystem, but the OBS tries to read it from its own. By using a path inside the shared volume (hl/out.json), both containers can access the same file. The CLI fallback still writes to the local filesystem since hl-node runs as a subprocess of the OBS. Co-Authored-By: Claude Opus 4.6 --- server/src/listeners/order_book/utils.rs | 27 +++++++++++++----------- 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/server/src/listeners/order_book/utils.rs b/server/src/listeners/order_book/utils.rs index 5dc4c762..7c30a3b2 100644 --- a/server/src/listeners/order_book/utils.rs +++ b/server/src/listeners/order_book/utils.rs @@ -23,27 +23,30 @@ use crate::{ use log::info; use tokio::fs; -/// Fetches an L4 snapshot and writes it to `out.json` in the given directory. +/// Fetches an L4 snapshot and writes it to `out.json`. /// -/// First attempts the `fileSnapshot` API (available on full nodes). If that fails -/// (e.g. when the node is started with `--serve-info`), falls back to reading -/// the latest periodic ABCI state file and computing L4 snapshots via the -/// `hl-node compute-l4-snapshots` CLI command. +/// First attempts the `fileSnapshot` API. The output path for the API is set +/// inside `hl/` (the shared volume) so the node process can write it and the +/// order book server can read it — important when running in separate containers. /// -/// The fallback requires `hl-node` to be available (on PATH or via `HL_NODE_PATH` -/// env var) and periodic ABCI state files to exist on disk. +/// If the API fails, falls back to reading the latest periodic ABCI state file +/// and computing L4 snapshots via `hl-node compute-l4-snapshots`. The fallback +/// requires `hl-node` on PATH (or via `HL_NODE_PATH` env var). pub(super) async fn process_rmp_file(dir: &Path) -> Result { - let output_path = dir.join("out.json"); + // API: write into the shared volume so the node and OBS can both access it + let api_output_path = dir.join("hl/out.json"); + // CLI: write to local filesystem (OBS runs hl-node as a subprocess) + let cli_output_path = dir.join("out.json"); - match process_via_api(&output_path).await { - Ok(()) => return Ok(output_path), + match process_via_api(&api_output_path).await { + Ok(()) => return Ok(api_output_path), Err(err) => { info!("fileSnapshot API unavailable ({err}), trying periodic ABCI state fallback"); } } - process_via_cli(dir, &output_path).await?; - Ok(output_path) + process_via_cli(dir, &cli_output_path).await?; + Ok(cli_output_path) } async fn process_via_api(output_path: &Path) -> Result<()> { From bf6b62a6938bd10b783b80eb5d9de059a85cec97 Mon Sep 17 00:00:00 2001 From: yangwao Date: Tue, 17 Mar 2026 16:20:24 +0100 Subject: [PATCH 28/29] fix: correct API port, add timeout, and prevent concurrent snapshot fetches The fileSnapshot API was hitting port 3001 but the node serves on 4001. Added a 30s timeout to the reqwest client so the OBS falls back quickly when the node is unavailable instead of hanging indefinitely. Also added an AtomicBool guard to prevent the 10s ticker from spawning overlapping fetch_snapshot tasks that pile up hl-node subprocesses. Co-Authored-By: Claude Opus 4.6 --- server/src/listeners/order_book/mod.rs | 13 ++++++++++--- server/src/listeners/order_book/utils.rs | 6 ++++-- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/server/src/listeners/order_book/mod.rs b/server/src/listeners/order_book/mod.rs index 3c593223..d9000752 100644 --- a/server/src/listeners/order_book/mod.rs +++ b/server/src/listeners/order_book/mod.rs @@ -70,6 +70,7 @@ pub(crate) async fn hl_listen( // every so often, we fetch a new snapshot and the snapshot_fetch_task starts running. // Result is sent back along this channel (if error, we want to return to top level) let (snapshot_fetch_task_tx, mut snapshot_fetch_task_rx) = unbounded_channel::>(); + let fetch_in_progress = Arc::new(std::sync::atomic::AtomicBool::new(false)); watcher.watch(&order_statuses_dir, RecursiveMode::Recursive)?; watcher.watch(&fills_dir, RecursiveMode::Recursive)?; @@ -131,9 +132,13 @@ pub(crate) async fn hl_listen( } } _ = ticker.tick() => { - let listener = listener.clone(); - let snapshot_fetch_task_tx = snapshot_fetch_task_tx.clone(); - fetch_snapshot(dir.clone(), listener, snapshot_fetch_task_tx, ignore_spot); + if !fetch_in_progress.load(std::sync::atomic::Ordering::Relaxed) { + let listener = listener.clone(); + let snapshot_fetch_task_tx = snapshot_fetch_task_tx.clone(); + let fetch_in_progress = fetch_in_progress.clone(); + fetch_in_progress.store(true, std::sync::atomic::Ordering::Relaxed); + fetch_snapshot(dir.clone(), listener, snapshot_fetch_task_tx, ignore_spot, fetch_in_progress); + } } () = sleep(Duration::from_secs(inactivity_exit_secs)) => { let listener = listener.lock().await; @@ -150,6 +155,7 @@ fn fetch_snapshot( listener: Arc>, tx: UnboundedSender>, ignore_spot: bool, + fetch_in_progress: Arc, ) { let tx = tx.clone(); tokio::spawn(async move { @@ -196,6 +202,7 @@ fn fetch_snapshot( Err(err) => Err(err), }; let _unused = tx.send(res); + fetch_in_progress.store(false, std::sync::atomic::Ordering::Relaxed); Ok(()) }); } diff --git a/server/src/listeners/order_book/utils.rs b/server/src/listeners/order_book/utils.rs index 7c30a3b2..b1a30022 100644 --- a/server/src/listeners/order_book/utils.rs +++ b/server/src/listeners/order_book/utils.rs @@ -61,9 +61,11 @@ async fn process_via_api(output_path: &Path) -> Result<()> { "includeHeightInOutput": true }); - let client = Client::new(); + let client = Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build()?; client - .post("http://localhost:3001/info") + .post("http://localhost:4001/info") .header("Content-Type", "application/json") .json(&payload) .send() From 00026cf18831569dd3c379e7d141725a8dfc0369 Mon Sep 17 00:00:00 2001 From: yangwao Date: Wed, 18 Mar 2026 10:05:11 +0100 Subject: [PATCH 29/29] fix: correct API port to 3001 (info API), not 4001 (gossip) Port 4001 is the binary gossip protocol, not HTTP. The info API (--serve-info) listens on port 3001. Co-Authored-By: Claude Opus 4.6 --- server/src/listeners/order_book/utils.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/src/listeners/order_book/utils.rs b/server/src/listeners/order_book/utils.rs index b1a30022..781c4677 100644 --- a/server/src/listeners/order_book/utils.rs +++ b/server/src/listeners/order_book/utils.rs @@ -65,7 +65,7 @@ async fn process_via_api(output_path: &Path) -> Result<()> { .timeout(std::time::Duration::from_secs(30)) .build()?; client - .post("http://localhost:4001/info") + .post("http://localhost:3001/info") .header("Content-Type", "application/json") .json(&payload) .send()