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 diff --git a/.github/workflows/ci-nightly.yml b/.github/workflows/ci-nightly.yml new file mode 100644 index 00000000..e00917b4 --- /dev/null +++ b/.github/workflows/ci-nightly.yml @@ -0,0 +1,76 @@ +name: CI Nightly + +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 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" \ + 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) + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@nightly + - run: | + for i in {1..16}; do + cargo test --workspace --all-features -- -Z unstable-options --shuffle + done + + summary: + name: Sanitizer Summary + runs-on: ubuntu-24.04 + needs: [sanitizer-asan, sanitizer-tsan, unit-tests-flaky] + 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 "- **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" diff --git a/.github/workflows/ci-tests.yml b/.github/workflows/ci-tests.yml new file mode 100644 index 00000000..3a05d1ec --- /dev/null +++ b/.github/workflows/ci-tests.yml @@ -0,0 +1,55 @@ +name: CI Tests + +on: + pull_request: + branches: + - '**' + push: + branches: + - main + +jobs: + clippy: + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@nightly + 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@nightly + - run: cargo test --workspace --all-features -- -Z unstable-options --shuffle + + summary: + name: CI Summary + runs-on: ubuntu-24.04 + needs: [clippy, unit-tests] + 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 "" + 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" diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 00000000..a81c9e71 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,310 @@ +# AGENTS.md + +Comprehensive guidance for AI assistants and developers working in this repository. + +## Quick Start + +```bash +# Build and lint +cargo build --workspace +# 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 --all +# Check only +cargo fmt --all -- --check + +# 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>, +} +``` + +## 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 +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 + +### Making Changes + +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 --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" +- Keep commits focused on single behavior change +- Example history: + ``` + fix: panic in orderbook listener + feat: use yawc as default WebSocket crate supporting compression + docs: update readme + ``` + +### 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 +``` + +## Key Dependencies + +| 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 | + +## Caveats & Limitations + +- 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 + +## Project Context + +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 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/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/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 diff --git a/NEW.md b/NEW.md new file mode 100644 index 00000000..43ef00ff --- /dev/null +++ b/NEW.md @@ -0,0 +1,520 @@ +# Newcomer's Guide to the Order Book Server + +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 + +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 + +--- + +## 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 + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ 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 (debug) +cargo build --workspace +# Build everything (release) +cargo build --release + +# Run linter (required before commits) +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 -- \ + --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 b0fe860d..e499da78 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`. @@ -24,23 +22,118 @@ 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`). + ## 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 +``` + +### 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: + +```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: ```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). -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. +## 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. + ## Caveats - This server does **not** show untriggered trigger orders. diff --git a/binaries/src/bin/websocket_server.rs b/binaries/src/bin/websocket_server.rs index 09a605f7..c11a8752 100644 --- a/binaries/src/bin/websocket_server.rs +++ b/binaries/src/bin/websocket_server.rs @@ -25,6 +25,19 @@ struct Args { /// documentation for for more info. #[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. + #[arg(long)] + inactivity_exit_secs: Option, } #[tokio::main] @@ -36,8 +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); - 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, ignore_spot, compression_level, inactivity_exit_secs).await?; Ok(()) } 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 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..d06aeff0 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,30 @@ pub(crate) trait DirectoryListener { #[cfg(test)] mod tests { - use crate::{ - listeners::directory::{DirectoryListener, EventSource}, - prelude::*, + use std::{ + 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 std::{ - io::{Seek, SeekFrom}, - path::{Path, PathBuf}, - sync::{Arc, Mutex}, - time::Duration, + use tempfile::tempdir; + use tokio::{ + fs::File as TokioFile, + io::AsyncWriteExt, + sync::mpsc::{UnboundedSender, unbounded_channel}, + time::{sleep, timeout}, + }; + + use crate::{ + listeners::directory::{DirectoryListener, EventSource}, + prelude::*, }; - use tokio::{fs::File as TokioFile, io::AsyncWriteExt, sync::mpsc::unbounded_channel, time::sleep}; - const MOCK_HL_DIR: &str = "tmp/ws_listener_test"; + 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}]} @@ -73,7 +82,15 @@ mod tests { "#, ]; - async fn listen(listener: &mut L, event_source: EventSource, dir: &Path) -> Result<()> { + // 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, + 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) @@ -86,6 +103,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)) => { @@ -95,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)?; } } } @@ -146,6 +167,7 @@ mod tests { Ok(()) } + // 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(); @@ -174,10 +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>, + current_path: Option, } impl DirectoryListener for TestListener { @@ -190,8 +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); + self.current_path = Some(new_file); Ok(()) } @@ -203,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>) -> Self { - Self { file: None, history } + Self { file: None, history, current_path: None } } } @@ -213,20 +248,27 @@ 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())); + let (watcher_ready_tx, mut watcher_ready_rx) = unbounded_channel(); let mut test_listener = TestListener::new(history.clone()); { 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).await?; sleep(Duration::from_secs(2)).await; diff --git a/server/src/listeners/order_book/mod.rs b/server/src/listeners/order_book/mod.rs index fcd9425a..d9000752 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) -> 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()?; @@ -64,6 +70,7 @@ pub(crate) async fn hl_listen(listener: Arc>, dir: Path // 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)?; @@ -112,17 +119,28 @@ 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(())) => {} } } _ = 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(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()); @@ -137,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 { @@ -183,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(()) }); } @@ -278,25 +298,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(()) @@ -415,11 +435,14 @@ 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(); @@ -436,14 +459,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/state.rs b/server/src/listeners/order_book/state.rs index 3fb74ae2..c31c5805 100644 --- a/server/src/listeners/order_book/state.rs +++ b/server/src/listeners/order_book/state.rs @@ -1,7 +1,9 @@ +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::*, @@ -10,7 +12,6 @@ use crate::{ node_data::{Batch, NodeDataOrderDiff, NodeDataOrderStatus}, }, }; -use std::collections::{HashMap, HashSet, VecDeque}; #[derive(Clone)] pub(super) struct OrderBookState { @@ -104,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/listeners/order_book/utils.rs b/server/src/listeners/order_book/utils.rs index fdbd3b57..781c4677 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::{ @@ -11,17 +20,36 @@ use crate::{ node_data::{Batch, NodeDataFill, NodeDataOrderDiff, NodeDataOrderStatus}, }, }; -use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; -use reqwest::Client; -use serde_json::json; -use std::collections::VecDeque; -use std::{ - collections::HashMap, - path::{Path, PathBuf}, -}; +use log::info; +use tokio::fs; +/// Fetches an L4 snapshot and writes it to `out.json`. +/// +/// 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. +/// +/// 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(&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, &cli_output_path).await?; + Ok(cli_output_path) +} + +async fn process_via_api(output_path: &Path) -> Result<()> { let payload = json!({ "type": "fileSnapshot", "request": { @@ -33,7 +61,9 @@ pub(super) async fn process_rmp_file(dir: &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") .header("Content-Type", "application/json") @@ -41,7 +71,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( @@ -135,10 +239,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/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 694c0889..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,9 +122,6 @@ mod tests { inner::{InnerL4Order, InnerLevel}, }, }; - use alloy::primitives::Address; - use itertools::Itertools; - use std::{fs::create_dir_all, path::PathBuf}; #[must_use] fn snapshot_to_l2_snapshot( @@ -274,12 +279,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(()) } diff --git a/server/src/order_book/types.rs b/server/src/order_book/types.rs index 447eb3f5..348cb7cf 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 { @@ -54,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/servers/websocket_server.rs b/server/src/servers/websocket_server.rs index 05d04fd6..4a47291a 100644 --- a/server/src/servers/websocket_server.rs +++ b/server/src/servers/websocket_server.rs @@ -1,27 +1,20 @@ -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, + http::StatusCode, + response::{IntoResponse, Response}, + routing::get, +}; +use futures_util::{SinkExt, StreamExt}; +use log::{error, info}; use tokio::{ net::TcpListener, + select, sync::{ Mutex, broadcast::{Sender, channel}, @@ -29,7 +22,26 @@ use tokio::{ }; use yawc::{FrameView, OpCode, WebSocket}; -pub async fn run_websocket_server(address: &str, ignore_spot: bool, compression_level: u32) -> 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 @@ -42,7 +54,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); } @@ -78,8 +90,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, @@ -89,10 +107,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( @@ -281,15 +299,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() { @@ -323,11 +341,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; } } @@ -336,11 +354,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/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 64dbd548..0fde738b 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; @@ -31,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; } @@ -49,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"); @@ -62,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; } @@ -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() {