From 51ee4158ae172d11ba9e67fa94a47b6c54707001 Mon Sep 17 00:00:00 2001 From: 0xFantomMenace <93462858+0xFantomMenace@users.noreply.github.com> Date: Thu, 8 Oct 2026 02:59:08 +0000 Subject: [PATCH 1/3] Subscribe to pending Hyperliquid transactions on the mempool channel (1.13.0) - ws::MEMPOOL_CHANNEL, served only at ws::STREAM_WS_URL; the symbol is optional on it. ws::DEFAULT_WS_URL names the default endpoint. - MempoolItem and MempoolSignature, decoded with ServerMsg::mempool_items() or MempoolItem::decode(). - Capability::ws_endpoint and Capability::plans. - Version 1.13.0. --- CHANGELOG.md | 23 ++++ Cargo.toml | 2 +- README.md | 103 +++++++++++++++- src/error.rs | 3 +- src/lib.rs | 9 +- src/types.rs | 94 ++++++++++++++ src/ws/client.rs | 111 +++++++++++++++-- src/ws/mod.rs | 2 +- tests/api_contract.rs | 40 ++++++ tests/mempool.rs | 225 ++++++++++++++++++++++++++++++++++ tests/websocket_capability.rs | 4 +- 11 files changed, 597 insertions(+), 19 deletions(-) create mode 100644 tests/mempool.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index fbd6d5a..93f27a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,29 @@ All notable changes to the `oxarchive` Rust SDK are tracked in this file. The format is loosely based on [Keep a Changelog](https://keepachangelog.com). +## [1.13.0] - 2026-10-08 + +### Added +- The `mempool` WebSocket channel (`ws::MEMPOOL_CHANNEL`): signed Hyperliquid + transactions as our Hyperliquid node receives them from its peers, before + they are included in a block, for every Hyperliquid product. It is live + only, served only at `wss://stream.0xarchive.io/ws`, and included with the + Pro, Scale and Enterprise plans. Every other channel stays on every plan. + `subscribe(MEMPOOL_CHANNEL, None)` streams every pending transaction; + with a symbol, only the actions that reference that market. `mempool` is + the only channel whose symbol is optional. +- `MempoolItem` and `MempoolSignature`, decoded from a `mempool` message with + `ServerMsg::mempool_items()` or `MempoolItem::decode()`. `action` is a + `serde_json::Value` exactly as signed; unknown fields are ignored. +- `ws::STREAM_WS_URL` (`wss://stream.0xarchive.io/ws`) and + `ws::DEFAULT_WS_URL` (`wss://api.0xarchive.io/ws`, the default `ws_url`). + Pass `STREAM_WS_URL` to `WsOptions::ws_url()` for a connection that + subscribes to `mempool`. +- `Capability::ws_endpoint` and `Capability::plans`, the fields + `/v1/capabilities` sets on the `mempool` row only. `None` means every + endpoint and every plan. They are left out when serializing a row that + does not have them. + ## [1.12.0] - 2026-10-05 ### Upgrading from 1.8 diff --git a/Cargo.toml b/Cargo.toml index e7d263f..94156c4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "oxarchive" -version = "1.12.0" +version = "1.13.0" edition = "2021" rust-version = "1.85" description = "Rust SDK for async 0xArchive market data clients" diff --git a/README.md b/README.md index 0bc2a44..1fcbc49 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Or add directly to your `Cargo.toml`: ```toml [dependencies] -oxarchive = "1.12" +oxarchive = "1.13" tokio = { version = "1", features = ["rt-multi-thread", "macros"] } chrono = "0.4" # the examples below use it for time windows ``` @@ -26,7 +26,7 @@ chrono = "0.4" # the examples below use it for time windows For WebSocket support (real-time streaming and replay): ```toml -oxarchive = { version = "1.12", features = ["websocket"] } +oxarchive = { version = "1.13", features = ["websocket"] } ``` ## Quick Start @@ -196,8 +196,10 @@ println!("{} symbols, {} on HIP-3", symbols.len(), hip3.len()); one `Capability` row per venue and datatype: its REST routes, its WebSocket channels, whether those stream live (`live`) and replay history (`replay`), the first instant served (`available_from`), the cadence, the largest page -(`page_limit`), and the accepted `interval` values. The route is public and -uses no credits. Read replay availability from it rather than from a fixed +(`page_limit`), and the accepted `interval` values. A row whose channels are +served at one endpoint only, or included with some plans only, names them in +`ws_endpoint` and `plans`; only the `mempool` row has them, and elsewhere both +are `None`. The route is public and uses no credits. Read replay availability from it rather than from a fixed list: ```rust @@ -1850,6 +1852,8 @@ For large historical downloads, use the S3 Parquet bulk export in the [Data Cata > Lighter live subscriptions are available for `lighter_orderbook`, `lighter_trades`, `lighter_open_interest`, and `lighter_funding` on mainnet, and for `rh_lighter_orderbook`, `rh_lighter_trades`, `rh_lighter_open_interest`, and `rh_lighter_funding` on Robinhood Chain. `lighter_candles`, `lighter_l3_orderbook`, and `rh_lighter_candles` remain replay-only. Every Lighter channel on both deployments supports historical replay. +> The `mempool` channel (pending Hyperliquid transactions) is served only at `wss://stream.0xarchive.io/ws` and is included with the Pro, Scale and Enterprise plans. See [Pending Transactions (Mempool)](#pending-transactions-mempool). + The client connects with `version=2026-10-01`. Under that version, error messages carry an `error_code` (see [Error Handling](#error-handling)), and Lighter replay rows have the same shapes as the live payloads. `client.capabilities()` lists which channels stream live and which replay. ### Real-time Streaming @@ -1957,6 +1961,94 @@ If the lag persists, a notice such as `Stopped the lighter_trades stream for BTC ends that subscription; subscribe again to resume. `lighter_orderbook` never sends an older book in place of a newer one. +### Pending Transactions (Mempool) + +The `mempool` channel streams signed Hyperliquid transactions (orders, cancels, +modifies, TWAPs, leverage changes, transfers and every other action type) as our +Hyperliquid node receives them from its peers, before they are included in a block. +It covers every Hyperliquid product: perps, HIP-3, HIP-4 and spot. + +- **Live only.** There is no replay, history, REST route or export. +- **One endpoint.** It is served only at `wss://stream.0xarchive.io/ws` + (`oxarchive::ws::STREAM_WS_URL`), with the same API key and protocol. At + `wss://api.0xarchive.io/ws`, the default `ws_url`, a subscribe is answered with + `ErrorCode::EndpointUnsupported`. Use a separate connection for the channels served + at the default endpoint. +- **Plans.** It is included with the Pro, Scale and Enterprise plans; on other plans a + subscribe is answered with `ErrorCode::Forbidden`. Every other channel stays on every + plan, Free included. Each data message is metered like any other WebSocket message. + +```rust +use oxarchive::ws::{OxArchiveWs, ServerMsg, WsOptions, MEMPOOL_CHANNEL, STREAM_WS_URL}; + +let mut ws = OxArchiveWs::new(WsOptions::new("your-api-key").ws_url(STREAM_WS_URL)); +ws.connect().await?; +let mut rx = ws.rx.take().expect("receiver"); + +ws.subscribe(MEMPOOL_CHANNEL, Some("BTC")).await?; // actions that reference BTC +ws.subscribe(MEMPOOL_CHANNEL, Some("xyz:TSLA")).await?; // HIP-3; spot "HYPE-USDC", HIP-4 "#49720" +// ws.subscribe(MEMPOOL_CHANNEL, None).await?; // every pending transaction (unfiltered) + +while let Some(msg) = rx.recv().await { + match msg.mempool_items() { + Some(Ok(items)) => { + for item in items { + println!("{:?} {} {:?}", item.received_at, item.action["type"], item.symbols); + } + } + Some(Err(e)) => eprintln!("Unexpected mempool payload: {e}"), + None => { + if let ServerMsg::Error { message, error_code } = msg { + eprintln!("{error_code:?}: {message}"); + } + } + } +} +``` + +The symbol is optional on this channel only. Without it you receive every pending +transaction; with it, every action whose asset ids include that market, whole (an +order batch that touches `BTC` and `ETH` reaches both subscriptions). Symbols are +spelled as everywhere else: perps `BTC`, HIP-3 `xyz:TSLA`, spot `HYPE-USDC` +(`HYPE/USDC` is also accepted) and HIP-4 `#49720`. An unknown symbol is answered with +`ErrorCode::InvalidSymbol`. The `subscribed` acknowledgement carries the canonical +symbol, or `None` for the unfiltered stream. + +The server sends one `ServerMsg::Data` per batch of transactions as it arrives, with +`coin` and `symbol` set to the subscription's symbol (`None` when unfiltered). +`msg.mempool_items()` decodes its `data` into one `MempoolItem` per signed action: + +| Field | Description | +|-------|-------------| +| `received_at` | When our node received the transaction: an RFC 3339 UTC string with nanosecond precision. Not a block time. | +| `received_at_ms` | The same time in Unix milliseconds. | +| `symbols` | Markets the action's asset ids reference, in first-seen order without repeats. Empty for actions with no market, such as transfers, `noop`, `scheduleCancel` and validator actions. | +| `action` | The action exactly as signed (`serde_json::Value`), in Hyperliquid's exchange-action format: asset ids (`a` or `asset`) rather than symbols, prices and sizes as strings. `serde_json::Value` sorts object keys unless serde_json's `preserve_order` feature is enabled; enable it in your crate if you need the signed key order. | +| `nonce` | The action's nonce. | +| `vault_address` | The vault or subaccount the action acts for, or `None`. | +| `expires_after_ms` | The action's `expiresAfter` in Unix milliseconds, or `None`. | +| `signature` | `MempoolSignature` with `r`, `s` and `v`. The signer's address is not included. | + +`action["type"]` names the action, for example `order`, `cancel`, `cancelByCloid`, +`modify`, `batchModify`, `scheduleCancel`, `twapOrder`, `twapCancel`, +`updateLeverage`, `updateIsolatedMargin`, `noop`, `evmRawTx`, or a transfer such as +`usdSend`, `spotSend`, `usdClassTransfer` or `sendAsset`. Hyperliquid adds action +types, so handle types you do not recognise. + +A pending transaction is not an executed one: it can still be rejected, expire or +never land in a block. The same signed action can occasionally arrive twice; +deduplicate on `signature` if that matters to you. + +**Volume and limits.** The unfiltered stream is several megabytes per second before +compression, and this client does not negotiate permessage-deflate compression, so +subscribe with a symbol where you can. Unfiltered subscriptions are limited +server-wide, and when they are at capacity a subscribe without a symbol is answered +with `ErrorCode::RateLimited`; symbol subscriptions are not capped this way. A +connection that reads too slowly is disconnected, as on any channel, and the usual +per-connection limits apply (subscriptions per plan, 10 subscribe operations per +second). If the feed is temporarily unavailable, a subscribe is answered with +`ErrorCode::UpstreamUnavailable`. + ### Historical Replay Replay a bounded historical window with original timing preserved. Every replay @@ -2037,6 +2129,7 @@ ws.replay_stop().await?; | `funding` | Funding rate snapshots | Yes | Yes | | `ticker` | Price and 24h volume | Yes | No | | `all_tickers` | All market tickers | Yes | No | +| `mempool` | Pending transactions on every Hyperliquid product, before they are in a block. Symbol optional. Pro, Scale and Enterprise plans; served only at `wss://stream.0xarchive.io/ws`. | Yes | No | | `orderbook_full` | Hyperliquid core full-depth L2 order book, aggregated from order-level data (every price level, no user attribution): an `l4_snapshot` frame (`ServerMsg::L4Snapshot`) with the whole book, then `l4_batch` frames (`ServerMsg::L4Batch`) of price-level changes (`side`, `px`, `sz`, `n`, `bn`; `sz` and `n` are `0` when a level is removed). | Yes | Yes, bulk and single-channel, from 2026-03-11 01:03 UTC | | `lighter_orderbook` | Lighter L2 order book | Yes | Yes | | `lighter_trades` | Lighter trades | Yes | Yes | @@ -2095,7 +2188,7 @@ while let Some(msg) = rx.recv().await { ### Tier Limits -All self-serve tiers reach the published route families; Free covers the most recent rolling 30 days of history (30-day span per request or replay), and Build and above keep the retained archive. Schema availability remains family-specific; plans gate capacity and Free's 30-day history window, not route families, schemas, or served depth. +All self-serve tiers reach the published route families; Free covers the most recent rolling 30 days of history (30-day span per request or replay), and Build and above keep the retained archive. Schema availability remains family-specific; plans gate capacity and Free's 30-day history window, not route families, schemas, or served depth. The one exception is the live `mempool` channel, included with the Pro, Scale and Enterprise plans. | Tier | Max Subscriptions | Max Connections | Max Replay Speed | |------|------------------|-----------------|------------------| diff --git a/src/error.rs b/src/error.rs index 926dffa..ea59137 100644 --- a/src/error.rs +++ b/src/error.rs @@ -52,7 +52,8 @@ pub enum ErrorCode { NotFound, /// Missing or invalid credentials. Unauthorized, - /// The credentials do not grant access to this resource. + /// The credentials do not grant access to this resource, or the plan does + /// not include it (the `mempool` WebSocket channel below Pro). Forbidden, /// The monthly credits are spent. InsufficientCredits, diff --git a/src/lib.rs b/src/lib.rs index 21cc82b..e1fdff7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -151,7 +151,7 @@ //! replay: //! //! ```toml -//! oxarchive = { version = "1.12", features = ["websocket"] } +//! oxarchive = { version = "1.13", features = ["websocket"] } //! ``` //! //! Live subscriptions cover the supported Hyperliquid channels and four @@ -167,6 +167,11 @@ //! replay as bulk, single-channel streams. `client.capabilities()` lists //! which channels stream live and which replay. //! +//! The live-only `mempool` channel carries pending Hyperliquid transactions, +//! before they are in a block; its items decode into [`MempoolItem`]. It is +//! served only at `wss://stream.0xarchive.io/ws` (`ws::STREAM_WS_URL`) and +//! included with the Pro, Scale and Enterprise plans. +//! //! Bulk streaming over WebSocket has been discontinued, so //! `OxArchiveWs::stream` is deprecated. For large historical downloads, use //! the S3 Parquet bulk export at . @@ -197,7 +202,7 @@ pub use types::{ LiquidationLevelBucket, LiquidationLevels, LiquidationLevelsHistoryItem, TriggerLevelBucket, TriggerLevels, TriggerLevelsHistoryItem, LighterLiveAssetCtx, LighterLiveData, LighterLiveLevel, LighterLiveMarketStats, - LighterLiveOrderBook, LighterLiveTrade, + LighterLiveOrderBook, LighterLiveTrade, MempoolItem, MempoolSignature, AccountSummary, CumulativeFunding, LighterL1Account, LighterL1Accounts, LighterLiquidation, LighterLiquidationVolume, MarketPosition, MarketPositionsSummary, MetaResponse, Position, PositionChange, PositionLeverage, PositionsFreshness, ResponseMeta, WalletPositions, diff --git a/src/types.rs b/src/types.rs index 4efea09..883a0ca 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1266,6 +1266,91 @@ impl LighterLiveData { } } +// --------------------------------------------------------------------------- +// Mempool (pending Hyperliquid transactions, live only) +// --------------------------------------------------------------------------- + +/// One signed action from a live `mempool` message. +/// +/// The `mempool` channel streams signed Hyperliquid transactions as our +/// Hyperliquid node receives them from its peers, before they are included in +/// a block, for every Hyperliquid product (perps, HIP-3, HIP-4 and spot). It is +/// live only, served only at `wss://stream.0xarchive.io/ws`, and included +/// with the Pro, Scale and Enterprise plans. +/// +/// A pending transaction is not an executed one: it can still be rejected, +/// expire or never land in a block. The same signed action can occasionally +/// arrive twice; deduplicate on [`signature`](Self::signature) if that matters +/// to you. +/// +/// Decode a message's items with [`MempoolItem::decode`], or with +/// `ServerMsg::mempool_items` when the `websocket` feature is enabled. Fields +/// this version does not know are ignored, and new ones may be added in minor +/// releases, so the struct cannot be built with a literal outside this crate. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[non_exhaustive] +pub struct MempoolItem { + /// When our node received the transaction, as an RFC 3339 UTC string with + /// the node's nanosecond precision, for example + /// `2026-10-08T01:57:23.548737209Z`. Not a block time. + pub received_at: Option, + /// When our node received the transaction, in Unix milliseconds. + pub received_at_ms: Option, + /// The markets the action's asset ids reference, spelled as everywhere + /// else (`BTC`, `xyz:TSLA`, `HYPE-USDC`, `#49720`), in first-seen order + /// without repeats. Empty for actions with no market, such as transfers, + /// `noop`, `scheduleCancel` and validator actions. + #[serde(default)] + pub symbols: Vec, + /// The action exactly as signed, in Hyperliquid's exchange-action format: + /// asset ids (`a` or `asset`) rather than symbols, and prices and sizes as + /// strings. `action["type"]` names it, for example `order`, `cancel`, + /// `cancelByCloid`, `modify`, `batchModify`, `scheduleCancel`, + /// `twapOrder`, `twapCancel`, `updateLeverage`, `updateIsolatedMargin`, + /// `noop`, `evmRawTx`, or a transfer such as `usdSend`, `spotSend`, + /// `usdClassTransfer` or `sendAsset`. Hyperliquid adds action types, so + /// handle types you do not recognise. + /// + /// `serde_json::Value` sorts object keys unless serde_json's + /// `preserve_order` feature is enabled. Enable it in your own crate if you + /// need the keys in the order they were signed. + pub action: serde_json::Value, + /// The action's nonce. + pub nonce: Option, + /// The vault or subaccount the action acts for, or `None`. + pub vault_address: Option, + /// The action's `expiresAfter` in Unix milliseconds, or `None`. + pub expires_after_ms: Option, + /// The signature over the action. The signer's address is not included; + /// it can be recovered from the signature and the action. + pub signature: Option, +} + +/// The `{r, s, v}` signature of a pending action, as signed. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct MempoolSignature { + /// `r` as a hex string. + pub r: String, + /// `s` as a hex string. + pub s: String, + /// Recovery id. + pub v: u64, +} + +impl MempoolItem { + /// Decode the `data` field of a `mempool` message: one item per signed + /// action, in the order received. + /// + /// Returns `None` when `channel` is not `mempool`, and `Some(Err(..))` + /// when the payload does not match the item shape. + pub fn decode(channel: &str, data: &serde_json::Value) -> Option>> { + if channel != "mempool" { + return None; + } + Some(Vec::::deserialize(data).map_err(|e| crate::Error::Deserialize(e.to_string()))) + } +} + // --------------------------------------------------------------------------- // Convenience / summary types // --------------------------------------------------------------------------- @@ -2366,6 +2451,15 @@ pub struct Capability { /// Further detail, such as how replay behaves. #[serde(default)] pub notes: Option, + /// The only WebSocket endpoint that serves this row's channels, when that + /// is not every endpoint: `wss://stream.0xarchive.io/ws` for `mempool`. + /// `None` means every endpoint. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ws_endpoint: Option, + /// The plans that include it, when that is not every plan: `pro`, `scale` + /// and `enterprise` for `mempool`. `None` means every plan, Free included. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub plans: Option>, } impl Capability { diff --git a/src/ws/client.rs b/src/ws/client.rs index 53c7b47..70f6b6d 100644 --- a/src/ws/client.rs +++ b/src/ws/client.rs @@ -6,7 +6,7 @@ //! //! Requires the `websocket` feature: //! ```toml -//! oxarchive = { version = "1.12", features = ["websocket"] } +//! oxarchive = { version = "1.13", features = ["websocket"] } //! ``` use futures_util::{SinkExt, StreamExt}; @@ -18,13 +18,62 @@ use tokio_tungstenite::{connect_async, tungstenite::Message}; use crate::error::{Error, ErrorCode, Result}; use crate::http::API_VERSION; -use crate::types::LighterLiveData; +use crate::types::{LighterLiveData, MempoolItem}; // --------------------------------------------------------------------------- // Configuration // --------------------------------------------------------------------------- +/// The default `ws_url` in [`WsOptions`]. It serves every channel except +/// [`MEMPOOL_CHANNEL`], and every replay. +pub const DEFAULT_WS_URL: &str = "wss://api.0xarchive.io/ws"; + +/// The only endpoint that serves [`MEMPOOL_CHANNEL`], with the same API key +/// and protocol as [`DEFAULT_WS_URL`]. +/// +/// It serves a subset of the live channels; a channel it does not serve, and +/// every replay, is answered with a [`ServerMsg::Error`] that names +/// [`DEFAULT_WS_URL`]. Use one connection per endpoint: +/// +/// ```no_run +/// # use oxarchive::ws::{OxArchiveWs, WsOptions, MEMPOOL_CHANNEL, STREAM_WS_URL}; +/// # async fn example() -> oxarchive::Result<()> { +/// let mut ws = OxArchiveWs::new(WsOptions::new("your-api-key").ws_url(STREAM_WS_URL)); +/// ws.connect().await?; +/// ws.subscribe(MEMPOOL_CHANNEL, Some("BTC")).await?; +/// # Ok(()) +/// # } +/// ``` +pub const STREAM_WS_URL: &str = "wss://stream.0xarchive.io/ws"; + +/// The `mempool` channel: signed Hyperliquid transactions as our Hyperliquid +/// node receives them from its peers, before they are included in a block, +/// for every Hyperliquid product (perps, HIP-3, HIP-4 and spot). +/// +/// - Live only: there is no replay, history or REST route. +/// - Served only at [`STREAM_WS_URL`]. Elsewhere a subscribe is answered with +/// [`ErrorCode::EndpointUnsupported`]. +/// - Included with the Pro, Scale and Enterprise plans. On other plans a +/// subscribe is answered with [`ErrorCode::Forbidden`]. Every other channel +/// is on every plan. +/// +/// The symbol is optional on this channel only. `subscribe(MEMPOOL_CHANNEL, +/// None)` streams every pending transaction; with a symbol (`BTC`, +/// `xyz:TSLA`, `HYPE-USDC`, `#49720`) it streams every action that references +/// that market, whole. The unfiltered stream is several megabytes a second +/// before compression, and the server limits unfiltered subscriptions: when +/// they are at capacity, a subscribe without a symbol is answered with +/// [`ErrorCode::RateLimited`]. Subscribe with a symbol where you can. +/// +/// Each message is a [`ServerMsg::Data`] whose `coin` and `symbol` are the +/// subscription's symbol, or `None` on the unfiltered stream. Decode its +/// items with [`ServerMsg::mempool_items`]. +pub const MEMPOOL_CHANNEL: &str = "mempool"; + /// Options for the WebSocket connection. +/// +/// `ws_url` defaults to [`DEFAULT_WS_URL`]. Set it to [`STREAM_WS_URL`] for +/// [`MEMPOOL_CHANNEL`]. pub struct WsOptions { pub api_key: String, pub ws_url: String, @@ -37,7 +86,7 @@ impl WsOptions { pub fn new(api_key: impl Into) -> Self { Self { api_key: api_key.into(), - ws_url: "wss://api.0xarchive.io/ws".to_string(), + ws_url: DEFAULT_WS_URL.to_string(), auto_reconnect: true, reconnect_delay: Duration::from_secs(1), max_reconnect_attempts: 10, @@ -371,7 +420,8 @@ pub enum ServerMsg { /// replay to resync) and `endpoint_unsupported` (this endpoint does not /// serve the channel or operation; the message names the one that does). /// A channel that does not offer a mode (live, replay, `replay.seek`) - /// answers `unsupported_for_venue`. + /// answers `unsupported_for_venue`, and a channel your plan does not + /// include ([`MEMPOOL_CHANNEL`] below Pro) answers `forbidden`. Error { message: String, /// Stable machine-readable reason, `None` when the server sent none. @@ -382,6 +432,9 @@ pub enum ServerMsg { /// /// For the live Lighter channels of both deployments, /// [`ServerMsg::lighter_live_data`] decodes `data` into [`LighterLiveData`]. + /// For [`MEMPOOL_CHANNEL`], [`ServerMsg::mempool_items`] decodes it into + /// [`MempoolItem`]s; `coin` and `symbol` are `None` on the unfiltered + /// stream. Data { channel: String, coin: Option, @@ -542,6 +595,32 @@ impl ServerMsg { } } + /// Decode the items of a [`MEMPOOL_CHANNEL`] data message: one + /// [`MempoolItem`] per signed action, in the order received. + /// + /// Returns `None` for every other message and channel, and + /// `Some(Err(..))` when the payload does not match the item shape. + /// + /// ``` + /// use oxarchive::ws::ServerMsg; + /// + /// let frame = r#"{"type":"data","channel":"mempool","coin":null,"symbol":null,"data":[ + /// {"received_at":"2026-10-08T01:57:23.548737209Z","received_at_ms":1791424643548, + /// "symbols":["BTC"],"action":{"type":"cancel","cancels":[{"a":0,"o":1}]}, + /// "nonce":1791424643400,"vault_address":null,"expires_after_ms":null, + /// "signature":{"r":"0x1","s":"0x2","v":27}}]}"#; + /// let msg: ServerMsg = serde_json::from_str(frame).unwrap(); + /// let items = msg.mempool_items().unwrap().unwrap(); + /// assert_eq!(items[0].symbols, ["BTC"]); + /// assert_eq!(items[0].action["type"], "cancel"); + /// ``` + pub fn mempool_items(&self) -> Option>> { + match self { + ServerMsg::Data { channel, data, .. } => MempoolItem::decode(channel, data), + _ => None, + } + } + /// The `error_code` of an [`ServerMsg::Error`], or `None` for other /// messages and for errors the server sent without a code. pub fn error_code(&self) -> Option<&ErrorCode> { @@ -671,6 +750,10 @@ impl OxArchiveWs { /// support replay, not live subscriptions, and are rejected before a /// request is sent. Use REST for their current data or a bounded replay /// request for stored history. + /// + /// On [`MEMPOOL_CHANNEL`] the symbol is optional: `None` subscribes to + /// every pending transaction. That channel is served only at + /// [`STREAM_WS_URL`]. pub async fn subscribe(&self, channel: &str, symbol: Option<&str>) -> Result<()> { if is_lighter_replay_only_channel(channel) { return Err(Error::InvalidParam(LIGHTER_SUBSCRIPTION_ERROR.to_string())); @@ -723,7 +806,8 @@ impl OxArchiveWs { .await } - /// Unsubscribe from a real-time channel. + /// Unsubscribe from a real-time channel. On [`MEMPOOL_CHANNEL`], `None` + /// ends the unfiltered subscription and a symbol ends that symbol's. pub async fn unsubscribe(&self, channel: &str, symbol: Option<&str>) -> Result<()> { self.send(ClientMsg::Unsubscribe { channel: channel.to_string(), @@ -748,8 +832,9 @@ impl OxArchiveWs { /// Which channels replay, and from when, is listed by /// `client.capabilities()`. A channel without replay (for example /// `ticker` or `spot_trades`) is answered with a - /// [`ServerMsg::Error`] whose `error_code` is `unsupported_for_venue`. A - /// successful replay terminates with a `replay_completed` server message. + /// [`ServerMsg::Error`] whose `error_code` is `unsupported_for_venue`; + /// [`MEMPOOL_CHANNEL`], which is live only, is answered with an error too. + /// A successful replay terminates with a `replay_completed` server message. pub async fn replay( &self, channel: &str, @@ -885,7 +970,7 @@ impl OxArchiveWs { #[cfg(test)] mod lighter_live_tests { - use super::{ClientMsg, ServerMsg}; + use super::{ClientMsg, ServerMsg, MEMPOOL_CHANNEL}; use crate::types::LighterLiveData; const ORDERBOOK_FRAME: &str = r#"{"type":"data","channel":"lighter_orderbook","coin":"BTC","symbol":"BTC","data":{"coin":"BTC","time":1790294171459,"levels":[[{"px":"84368.7","sz":"0.00020","n":1},{"px":"84368.6","sz":"0.00020","n":1},{"px":"84368.3","sz":"0.00010","n":1}],[{"px":"84368.8","sz":"0.05720","n":1},{"px":"84368.9","sz":"0.14223","n":1},{"px":"84369.1","sz":"0.01198","n":1}]]}}"#; @@ -1070,6 +1155,16 @@ mod lighter_live_tests { serde_json::to_value(&unsubscribe).unwrap(), serde_json::json!({"op": "unsubscribe", "channel": "lighter_trades", "symbol": "BTC"}) ); + + // `mempool` is the one channel whose symbol is optional. + let unfiltered = ClientMsg::Subscribe { + channel: MEMPOOL_CHANNEL.to_string(), + symbol: None, + }; + assert_eq!( + serde_json::to_value(&unfiltered).unwrap(), + serde_json::json!({"op": "subscribe", "channel": "mempool", "symbol": null}) + ); } #[test] diff --git a/src/ws/mod.rs b/src/ws/mod.rs index 276ed1c..72967e8 100644 --- a/src/ws/mod.rs +++ b/src/ws/mod.rs @@ -4,7 +4,7 @@ //! //! Requires the `websocket` feature: //! ```toml -//! oxarchive = { version = "1.12", features = ["websocket"] } +//! oxarchive = { version = "1.13", features = ["websocket"] } //! ``` //! //! For large historical downloads, use the S3 Parquet bulk export at diff --git a/tests/api_contract.rs b/tests/api_contract.rs index 1ff1aa2..d0d1d64 100644 --- a/tests/api_contract.rs +++ b/tests/api_contract.rs @@ -858,6 +858,46 @@ async fn capabilities_are_typed_rows() { assert!(Capability::for_channel(&rows, "nope").is_none()); } +#[test] +fn only_the_mempool_row_names_an_endpoint_and_plans() { + let rows: Vec = serde_json::from_value(json!([ + {"venue": "hyperliquid", "datatype": "mempool", "rest_routes": [], + "ws_channels": ["mempool"], "live": true, "replay": false, + "available_from": null, "cadence": "event", "page_limit": null, "intervals": [], + "notes": "Live only, on wss://stream.0xarchive.io/ws.", + "ws_endpoint": "wss://stream.0xarchive.io/ws", + "plans": ["pro", "scale", "enterprise"]}, + {"venue": "hyperliquid", "datatype": "trades", + "rest_routes": ["/v1/hyperliquid/trades/{symbol}"], "ws_channels": ["trades"], + "live": true, "replay": true, "available_from": "2023-04-15T03:31:00.000Z", + "cadence": "event", "page_limit": 1000, "intervals": [], "notes": null} + ])) + .unwrap(); + + let mempool = Capability::for_channel(&rows, "mempool").unwrap(); + assert!(mempool.live && !mempool.replay); + assert_eq!( + mempool.ws_endpoint.as_deref(), + Some("wss://stream.0xarchive.io/ws") + ); + assert_eq!( + mempool.plans.as_deref(), + Some( + &[ + "pro".to_string(), + "scale".to_string(), + "enterprise".to_string() + ][..] + ) + ); + + let trades = Capability::for_channel(&rows, "trades").unwrap(); + assert_eq!((&trades.ws_endpoint, &trades.plans), (&None, &None)); + // Rows without them serialize as before: the fields are left out. + let value = serde_json::to_value(trades).unwrap(); + assert!(value.get("ws_endpoint").is_none() && value.get("plans").is_none()); +} + // --------------------------------------------------------------------------- // Filters the API honours // --------------------------------------------------------------------------- diff --git a/tests/mempool.rs b/tests/mempool.rs new file mode 100644 index 0000000..41394c6 --- /dev/null +++ b/tests/mempool.rs @@ -0,0 +1,225 @@ +#![cfg(feature = "websocket")] +//! The `mempool` WebSocket channel: pending Hyperliquid transactions. + +use oxarchive::error::Error; +use oxarchive::ws::{ + ClientMsg, OxArchiveWs, ServerMsg, WsOptions, DEFAULT_WS_URL, MEMPOOL_CHANNEL, STREAM_WS_URL, +}; +use oxarchive::{ErrorCode, LighterLiveData, MempoolItem, MempoolSignature}; +use serde_json::json; + +/// A frame shaped as the server sends it: an order on BTC and a transfer that +/// references no market, on the unfiltered stream. +const UNFILTERED_FRAME: &str = r#"{"type":"data","channel":"mempool","coin":null,"symbol":null,"data":[ +{"received_at":"2026-10-08T01:57:23.548737209Z","received_at_ms":1791424643548,"symbols":["BTC"],"action":{"type":"order","orders":[{"a":0,"b":true,"p":"83276","s":"0.40011","r":false,"t":{"limit":{"tif":"Alo"}},"c":"0x7849acc2c6c2f6f0fe4bc80ef13d1504"}],"grouping":"na"},"nonce":1791424643400,"vault_address":null,"expires_after_ms":null,"signature":{"r":"0x5afc","s":"0x57e2","v":28}}, +{"received_at":"2026-10-08T01:57:23.548737209Z","received_at_ms":1791424643548,"symbols":[],"action":{"type":"usdSend","signatureChainId":"0xa4b1","hyperliquidChain":"Mainnet","destination":"0x0000000000000000000000000000000000000001","amount":"10","time":1791424643400},"nonce":1791424643400,"vault_address":null,"expires_after_ms":1791424703400,"signature":{"r":"0x01","s":"0x02","v":27}} +]}"#; + +#[test] +fn mempool_is_served_at_the_stream_endpoint_only() { + assert_eq!(MEMPOOL_CHANNEL, "mempool"); + assert_eq!(STREAM_WS_URL, "wss://stream.0xarchive.io/ws"); + assert_eq!(DEFAULT_WS_URL, "wss://api.0xarchive.io/ws"); + assert_eq!(WsOptions::new("k").ws_url, DEFAULT_WS_URL); + assert_eq!( + WsOptions::new("k").ws_url(STREAM_WS_URL).ws_url, + STREAM_WS_URL + ); +} + +#[test] +fn subscribe_without_a_symbol_is_the_unfiltered_stream() { + let unfiltered = ClientMsg::Subscribe { + channel: MEMPOOL_CHANNEL.to_string(), + symbol: None, + }; + assert_eq!( + serde_json::to_value(&unfiltered).unwrap(), + json!({"op": "subscribe", "channel": "mempool", "symbol": null}) + ); + + for symbol in ["BTC", "xyz:TSLA", "HYPE-USDC", "#49720"] { + let filtered = ClientMsg::Subscribe { + channel: MEMPOOL_CHANNEL.to_string(), + symbol: Some(symbol.to_string()), + }; + assert_eq!( + serde_json::to_value(&filtered).unwrap(), + json!({"op": "subscribe", "channel": "mempool", "symbol": symbol}) + ); + } + + let unsubscribe = ClientMsg::Unsubscribe { + channel: MEMPOOL_CHANNEL.to_string(), + symbol: None, + }; + assert_eq!( + serde_json::to_value(&unsubscribe).unwrap(), + json!({"op": "unsubscribe", "channel": "mempool", "symbol": null}) + ); +} + +#[tokio::test] +async fn mempool_subscriptions_are_left_to_the_server() { + // Endpoint and plan are the server's to decide: nothing is refused here. + let ws = OxArchiveWs::new(WsOptions::new("test-key").ws_url(STREAM_WS_URL)); + ws.subscribe(MEMPOOL_CHANNEL, None).await.unwrap(); + ws.subscribe(MEMPOOL_CHANNEL, Some("BTC")).await.unwrap(); + ws.unsubscribe(MEMPOOL_CHANNEL, Some("BTC")).await.unwrap(); + ws.unsubscribe(MEMPOOL_CHANNEL, None).await.unwrap(); +} + +#[test] +fn unfiltered_frame_decodes_to_items() { + let msg: ServerMsg = serde_json::from_str(UNFILTERED_FRAME).unwrap(); + match &msg { + ServerMsg::Data { + channel, + coin, + symbol, + .. + } => { + assert_eq!(channel, "mempool"); + assert_eq!((coin, symbol), (&None, &None)); + } + other => panic!("expected a data message, got {other:?}"), + } + assert!(msg.lighter_live_data().is_none()); + + let items = msg.mempool_items().unwrap().unwrap(); + assert_eq!(items.len(), 2); + + let order = &items[0]; + assert_eq!( + order.received_at.as_deref(), + Some("2026-10-08T01:57:23.548737209Z") + ); + assert_eq!(order.received_at_ms, Some(1791424643548)); + assert_eq!(order.symbols, ["BTC"]); + assert_eq!(order.action["type"], "order"); + assert_eq!(order.action["orders"][0]["a"], 0); + assert_eq!(order.action["orders"][0]["p"], "83276"); + assert_eq!(order.nonce, Some(1791424643400)); + assert_eq!(order.vault_address, None); + assert_eq!(order.expires_after_ms, None); + assert_eq!( + order.signature, + Some(MempoolSignature { + r: "0x5afc".to_string(), + s: "0x57e2".to_string(), + v: 28, + }) + ); + + let transfer = &items[1]; + assert!(transfer.symbols.is_empty()); + assert_eq!(transfer.action["type"], "usdSend"); + assert_eq!(transfer.expires_after_ms, Some(1791424703400)); +} + +#[test] +fn symbol_frame_carries_the_subscription_symbol() { + let frame = UNFILTERED_FRAME.replace( + r#""coin":null,"symbol":null"#, + r#""coin":"BTC","symbol":"BTC""#, + ); + let msg: ServerMsg = serde_json::from_str(&frame).unwrap(); + let ServerMsg::Data { coin, symbol, .. } = &msg else { + panic!("expected a data message, got {msg:?}"); + }; + assert_eq!(coin.as_deref(), Some("BTC")); + assert_eq!(symbol.as_deref(), Some("BTC")); + assert_eq!(msg.mempool_items().unwrap().unwrap().len(), 2); +} + +#[test] +fn other_messages_are_not_mempool_items() { + let trades = r#"{"type":"data","channel":"trades","coin":"BTC","symbol":"BTC","data":[]}"#; + let msg: ServerMsg = serde_json::from_str(trades).unwrap(); + assert!(msg.mempool_items().is_none()); + + let pong: ServerMsg = serde_json::from_str(r#"{"type":"pong"}"#).unwrap(); + assert!(pong.mempool_items().is_none()); + + assert!(MempoolItem::decode("trades", &json!([])).is_none()); + assert!(LighterLiveData::decode("mempool", &json!([])).is_none()); +} + +#[test] +fn items_tolerate_nulls_new_fields_and_new_action_types() { + let data = json!([ + {"received_at": null, "received_at_ms": null, "symbols": [], + "action": {"type": "someFutureAction", "x": 1}, "nonce": null, + "vault_address": "0x00000000000000000000000000000000000000aa", + "expires_after_ms": null, "signature": null, "a_future_field": true}, + {"action": {"type": "noop"}} + ]); + let items = MempoolItem::decode(MEMPOOL_CHANNEL, &data) + .unwrap() + .unwrap(); + assert_eq!(items[0].received_at, None); + assert_eq!(items[0].nonce, None); + assert_eq!(items[0].signature, None); + assert_eq!(items[0].action["type"], "someFutureAction"); + assert_eq!( + items[0].vault_address.as_deref(), + Some("0x00000000000000000000000000000000000000aa") + ); + assert!(items[1].symbols.is_empty()); + assert_eq!(items[1].action, json!({"type": "noop"})); +} + +#[test] +fn a_payload_that_is_not_a_list_of_items_is_an_error() { + let result = MempoolItem::decode(MEMPOOL_CHANNEL, &json!({"not": "a list"})).unwrap(); + assert!(matches!(result, Err(Error::Deserialize(_)))); +} + +#[test] +fn subscribe_acks_without_a_symbol_parse() { + let ack: ServerMsg = serde_json::from_str( + r#"{"type":"subscribed","channel":"mempool","coin":null,"symbol":null}"#, + ) + .unwrap(); + match ack { + ServerMsg::Subscribed { + channel, + coin, + symbol, + } => { + assert_eq!(channel, "mempool"); + assert_eq!((coin, symbol), (None, None)); + } + other => panic!("expected a subscribed message, got {other:?}"), + } +} + +#[test] +fn mempool_refusals_carry_their_error_code() { + for (frame, expected) in [ + ( + r#"{"type":"error","message":"The mempool channel is included with the Pro, Scale and Enterprise plans. Upgrade at https://0xarchive.io/pricing.","error_code":"forbidden"}"#, + ErrorCode::Forbidden, + ), + ( + r#"{"type":"error","message":"The 'mempool' channel is live only and served on wss://stream.0xarchive.io/ws. Subscribe to it there.","error_code":"endpoint_unsupported"}"#, + ErrorCode::EndpointUnsupported, + ), + ( + r#"{"type":"error","message":"The unfiltered mempool stream is at capacity. Subscribe with a symbol, or try again later.","error_code":"rate_limited"}"#, + ErrorCode::RateLimited, + ), + ( + r#"{"type":"error","message":"The mempool channel is temporarily unavailable. Please try again shortly.","error_code":"upstream_unavailable"}"#, + ErrorCode::UpstreamUnavailable, + ), + ( + r#"{"type":"error","message":"Unknown symbol 'NOPE'.","error_code":"invalid_symbol"}"#, + ErrorCode::InvalidSymbol, + ), + ] { + let msg: ServerMsg = serde_json::from_str(frame).unwrap(); + assert_eq!(msg.error_code(), Some(&expected), "{frame}"); + assert!(msg.mempool_items().is_none()); + } +} diff --git a/tests/websocket_capability.rs b/tests/websocket_capability.rs index 4db77e4..20bda11 100644 --- a/tests/websocket_capability.rs +++ b/tests/websocket_capability.rs @@ -540,7 +540,8 @@ fn lighter_replay_rows_decode_with_the_live_shapes() { assert!(msg.lighter_live_data().is_none()); } -/// Channel rows of `GET /v1/capabilities` as served on 2026-10-04. +/// Channel rows of `GET /v1/capabilities` as served on 2026-10-04, plus the +/// `mempool` row. const CAPABILITY_CHANNELS: &str = r#"[ {"venue":"hyperliquid","datatype":"l2_orderbook","ws_channels":["orderbook"],"live":true,"replay":true,"available_from":"2023-04-15T00:00:00.000Z"}, {"venue":"hyperliquid","datatype":"l2_full_depth","ws_channels":["orderbook_full"],"live":true,"replay":true,"available_from":"2026-03-11T01:03:00.000Z"}, @@ -552,6 +553,7 @@ const CAPABILITY_CHANNELS: &str = r#"[ {"venue":"hyperliquid","datatype":"oi","ws_channels":["open_interest"],"live":true,"replay":true,"available_from":"2023-05-20T02:50:00.000Z"}, {"venue":"hyperliquid","datatype":"liquidations","ws_channels":["liquidations"],"live":true,"replay":true,"available_from":"2025-12-22T00:00:00.000Z"}, {"venue":"hyperliquid","datatype":"ticker","ws_channels":["ticker","all_tickers"],"live":true,"replay":false,"available_from":null}, +{"venue":"hyperliquid","datatype":"mempool","ws_channels":["mempool"],"live":true,"replay":false,"available_from":null,"ws_endpoint":"wss://stream.0xarchive.io/ws","plans":["pro","scale","enterprise"]}, {"venue":"hip3","datatype":"l2_orderbook","ws_channels":["hip3_orderbook"],"live":true,"replay":true,"available_from":"2026-02-16T16:57:00.000Z"}, {"venue":"hip3","datatype":"l2_full_depth","ws_channels":["hip3_orderbook_full"],"live":true,"replay":true,"available_from":"2026-03-11T01:03:00.000Z"}, {"venue":"hip3","datatype":"l4_diffs","ws_channels":["hip3_l4_diffs"],"live":true,"replay":true,"available_from":"2026-03-11T01:03:00.000Z"}, From 870a7da828a89906fb396ffa819d162aa838c87b Mon Sep 17 00:00:00 2001 From: 0xFantomMenace <93462858+0xFantomMenace@users.noreply.github.com> Date: Thu, 8 Oct 2026 03:08:13 +0000 Subject: [PATCH 2/3] Keep each mempool action's exact bytes for signature recovery - MempoolItem::action is a Box holding the action as sent, key order included; action_value() parses it for inspection. - Mempool frames arrive as ServerMsg::Mempool, read from the frame text by ServerMsg::from_text(), which the client now uses for every message. - serde_json's raw_value feature is enabled. --- CHANGELOG.md | 16 ++++-- Cargo.toml | 2 +- README.md | 27 +++++----- src/types.rs | 40 ++++++++------- src/ws/client.rs | 82 ++++++++++++++++++++++++------ tests/mempool.rs | 127 +++++++++++++++++++++++++++++++---------------- 6 files changed, 197 insertions(+), 97 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 93f27a9..ca6883c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,9 +14,15 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com). `subscribe(MEMPOOL_CHANNEL, None)` streams every pending transaction; with a symbol, only the actions that reference that market. `mempool` is the only channel whose symbol is optional. -- `MempoolItem` and `MempoolSignature`, decoded from a `mempool` message with - `ServerMsg::mempool_items()` or `MempoolItem::decode()`. `action` is a - `serde_json::Value` exactly as signed; unknown fields are ignored. +- `ServerMsg::Mempool`, with one `MempoolItem` (and its `MempoolSignature`) + per signed action. `ServerMsg::from_text()`, which the client uses to read + every message, builds it straight from the frame text, so + `MempoolItem::action` (a `Box`) keeps the + exact bytes the server sent, key order included, for signature recovery. + `MempoolItem::action_value()` parses the action into a `serde_json::Value` + for inspection, which sorts the keys. `ServerMsg::mempool_items()` returns + a message's items. Unknown item fields are ignored. +- serde_json's `raw_value` feature is enabled. - `ws::STREAM_WS_URL` (`wss://stream.0xarchive.io/ws`) and `ws::DEFAULT_WS_URL` (`wss://api.0xarchive.io/ws`, the default `ws_url`). Pass `STREAM_WS_URL` to `WsOptions::ws_url()` for a connection that @@ -26,6 +32,10 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com). endpoint and every plan. They are left out when serializing a row that does not have them. +### Changed +- `ServerMsg` gained the `Mempool` variant; an exhaustive match on it needs + a new arm. + ## [1.12.0] - 2026-10-05 ### Upgrading from 1.8 diff --git a/Cargo.toml b/Cargo.toml index 94156c4..9bfcb2a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,7 +63,7 @@ include = [ [dependencies] reqwest = { version = "0.12", features = ["json"] } serde = { version = "1", features = ["derive"] } -serde_json = "1" +serde_json = { version = "1", features = ["raw_value"] } thiserror = "2" chrono = { version = "0.4", features = ["serde"] } tokio = { version = "1", features = ["rt", "macros"] } diff --git a/README.md b/README.md index 1fcbc49..42c8ee9 100644 --- a/README.md +++ b/README.md @@ -1990,18 +1990,16 @@ ws.subscribe(MEMPOOL_CHANNEL, Some("xyz:TSLA")).await?; // HIP-3; spot "HYPE-USD // ws.subscribe(MEMPOOL_CHANNEL, None).await?; // every pending transaction (unfiltered) while let Some(msg) = rx.recv().await { - match msg.mempool_items() { - Some(Ok(items)) => { + match msg { + ServerMsg::Mempool { symbol, items, .. } => { for item in items { - println!("{:?} {} {:?}", item.received_at, item.action["type"], item.symbols); - } - } - Some(Err(e)) => eprintln!("Unexpected mempool payload: {e}"), - None => { - if let ServerMsg::Error { message, error_code } = msg { - eprintln!("{error_code:?}: {message}"); + // item.action.get() is the action's exact bytes, for signature recovery. + let action = item.action_value()?; // parsed for inspection (sorted keys) + println!("{symbol:?} {:?} {} {:?}", item.received_at, action["type"], item.symbols); } } + ServerMsg::Error { message, error_code } => eprintln!("{error_code:?}: {message}"), + _ => {} } } ``` @@ -2014,22 +2012,23 @@ spelled as everywhere else: perps `BTC`, HIP-3 `xyz:TSLA`, spot `HYPE-USDC` `ErrorCode::InvalidSymbol`. The `subscribed` acknowledgement carries the canonical symbol, or `None` for the unfiltered stream. -The server sends one `ServerMsg::Data` per batch of transactions as it arrives, with -`coin` and `symbol` set to the subscription's symbol (`None` when unfiltered). -`msg.mempool_items()` decodes its `data` into one `MempoolItem` per signed action: +The server sends one message per batch of transactions as it arrives. The client +delivers it as `ServerMsg::Mempool`, with `coin` and `symbol` set to the +subscription's symbol (`None` when unfiltered) and one `MempoolItem` per signed +action in `items`: | Field | Description | |-------|-------------| | `received_at` | When our node received the transaction: an RFC 3339 UTC string with nanosecond precision. Not a block time. | | `received_at_ms` | The same time in Unix milliseconds. | | `symbols` | Markets the action's asset ids reference, in first-seen order without repeats. Empty for actions with no market, such as transfers, `noop`, `scheduleCancel` and validator actions. | -| `action` | The action exactly as signed (`serde_json::Value`), in Hyperliquid's exchange-action format: asset ids (`a` or `asset`) rather than symbols, prices and sizes as strings. `serde_json::Value` sorts object keys unless serde_json's `preserve_order` feature is enabled; enable it in your crate if you need the signed key order. | +| `action` | The action exactly as signed, in Hyperliquid's exchange-action format: asset ids (`a` or `asset`) rather than symbols, prices and sizes as strings. It is a `Box` that keeps the exact bytes the server sent, key order included, for signature recovery (`item.action.get()`). `item.action_value()` parses it into a `serde_json::Value` for inspection, which sorts the keys. | | `nonce` | The action's nonce. | | `vault_address` | The vault or subaccount the action acts for, or `None`. | | `expires_after_ms` | The action's `expiresAfter` in Unix milliseconds, or `None`. | | `signature` | `MempoolSignature` with `r`, `s` and `v`. The signer's address is not included. | -`action["type"]` names the action, for example `order`, `cancel`, `cancelByCloid`, +The action's `type` names it, for example `order`, `cancel`, `cancelByCloid`, `modify`, `batchModify`, `scheduleCancel`, `twapOrder`, `twapCancel`, `updateLeverage`, `updateIsolatedMargin`, `noop`, `evmRawTx`, or a transfer such as `usdSend`, `spotSend`, `usdClassTransfer` or `sendAsset`. Hyperliquid adds action diff --git a/src/types.rs b/src/types.rs index 883a0ca..8daed62 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1283,11 +1283,13 @@ impl LighterLiveData { /// arrive twice; deduplicate on [`signature`](Self::signature) if that matters /// to you. /// -/// Decode a message's items with [`MempoolItem::decode`], or with -/// `ServerMsg::mempool_items` when the `websocket` feature is enabled. Fields -/// this version does not know are ignored, and new ones may be added in minor -/// releases, so the struct cannot be built with a literal outside this crate. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +/// With the `websocket` feature, `OxArchiveWs` delivers each message's items +/// in `ServerMsg::Mempool`. Elsewhere, deserialize them with serde_json from +/// the message text, not from a `serde_json::Value`, which has already sorted +/// the action's keys. Fields this version does not know are ignored, and new +/// ones may be added in minor releases, so the struct cannot be built with a +/// literal outside this crate. +#[derive(Debug, Clone, Serialize, Deserialize)] #[non_exhaustive] pub struct MempoolItem { /// When our node received the transaction, as an RFC 3339 UTC string with @@ -1304,17 +1306,19 @@ pub struct MempoolItem { pub symbols: Vec, /// The action exactly as signed, in Hyperliquid's exchange-action format: /// asset ids (`a` or `asset`) rather than symbols, and prices and sizes as - /// strings. `action["type"]` names it, for example `order`, `cancel`, + /// strings. Its `type` names it, for example `order`, `cancel`, /// `cancelByCloid`, `modify`, `batchModify`, `scheduleCancel`, /// `twapOrder`, `twapCancel`, `updateLeverage`, `updateIsolatedMargin`, /// `noop`, `evmRawTx`, or a transfer such as `usdSend`, `spotSend`, /// `usdClassTransfer` or `sendAsset`. Hyperliquid adds action types, so /// handle types you do not recognise. /// - /// `serde_json::Value` sorts object keys unless serde_json's - /// `preserve_order` feature is enabled. Enable it in your own crate if you - /// need the keys in the order they were signed. - pub action: serde_json::Value, + /// This raw form keeps the exact bytes the server sent, key order + /// included, as signature recovery needs them; `action.get()` returns + /// them as a `&str`. To inspect the action, + /// [`action_value`](Self::action_value) parses it into a + /// `serde_json::Value`, which sorts the keys. + pub action: Box, /// The action's nonce. pub nonce: Option, /// The vault or subaccount the action acts for, or `None`. @@ -1338,16 +1342,14 @@ pub struct MempoolSignature { } impl MempoolItem { - /// Decode the `data` field of a `mempool` message: one item per signed - /// action, in the order received. + /// The action parsed into a `serde_json::Value`, for inspection, for + /// example `item.action_value()?["type"]`. /// - /// Returns `None` when `channel` is not `mempool`, and `Some(Err(..))` - /// when the payload does not match the item shape. - pub fn decode(channel: &str, data: &serde_json::Value) -> Option>> { - if channel != "mempool" { - return None; - } - Some(Vec::::deserialize(data).map_err(|e| crate::Error::Deserialize(e.to_string()))) + /// The parsed form sorts object keys, so it no longer has the signed key + /// order. Use the raw [`action`](Self::action) for signature recovery. + pub fn action_value(&self) -> crate::Result { + serde_json::from_str(self.action.get()) + .map_err(|e| crate::Error::Deserialize(e.to_string())) } } diff --git a/src/ws/client.rs b/src/ws/client.rs index 70f6b6d..6bcd281 100644 --- a/src/ws/client.rs +++ b/src/ws/client.rs @@ -65,9 +65,9 @@ pub const STREAM_WS_URL: &str = "wss://stream.0xarchive.io/ws"; /// they are at capacity, a subscribe without a symbol is answered with /// [`ErrorCode::RateLimited`]. Subscribe with a symbol where you can. /// -/// Each message is a [`ServerMsg::Data`] whose `coin` and `symbol` are the -/// subscription's symbol, or `None` on the unfiltered stream. Decode its -/// items with [`ServerMsg::mempool_items`]. +/// [`OxArchiveWs`] delivers each message as a [`ServerMsg::Mempool`], whose +/// `coin` and `symbol` are the subscription's symbol, or `None` on the +/// unfiltered stream, and whose items keep each action's exact bytes. pub const MEMPOOL_CHANNEL: &str = "mempool"; /// Options for the WebSocket connection. @@ -432,15 +432,31 @@ pub enum ServerMsg { /// /// For the live Lighter channels of both deployments, /// [`ServerMsg::lighter_live_data`] decodes `data` into [`LighterLiveData`]. - /// For [`MEMPOOL_CHANNEL`], [`ServerMsg::mempool_items`] decodes it into - /// [`MempoolItem`]s; `coin` and `symbol` are `None` on the unfiltered - /// stream. + /// A [`MEMPOOL_CHANNEL`] message read with [`ServerMsg::from_text`], as + /// [`OxArchiveWs`] reads every message, arrives as [`ServerMsg::Mempool`] + /// instead. Data { channel: String, coin: Option, symbol: Option, data: serde_json::Value, }, + /// A [`MEMPOOL_CHANNEL`] data message: one batch of pending transactions, + /// with one [`MempoolItem`] per signed action, in the order received. + /// `coin` and `symbol` are the subscription's symbol, or `None` on the + /// unfiltered stream. + /// + /// [`ServerMsg::from_text`] (which [`OxArchiveWs`] uses) builds it + /// straight from the frame text, so each item's `action` keeps the exact + /// bytes the server sent. Deserializing a `ServerMsg` with serde gives a + /// [`ServerMsg::Data`] instead, and so does a mempool frame whose items do + /// not match [`MempoolItem`]. + #[serde(skip)] + Mempool { + coin: Option, + symbol: Option, + items: Vec, + }, /// Initial order-level L4 state for a live subscription or a replay. /// /// For an L4 replay (every channel in [`L4_REPLAY_CHANNELS`]), this @@ -595,11 +611,11 @@ impl ServerMsg { } } - /// Decode the items of a [`MEMPOOL_CHANNEL`] data message: one - /// [`MempoolItem`] per signed action, in the order received. + /// Parse one text frame from the server, as [`OxArchiveWs`] does. /// - /// Returns `None` for every other message and channel, and - /// `Some(Err(..))` when the payload does not match the item shape. + /// The same as `serde_json::from_str::`, except that a + /// [`MEMPOOL_CHANNEL`] data message becomes a [`ServerMsg::Mempool`], + /// read from the text so that each action keeps its exact bytes. /// /// ``` /// use oxarchive::ws::ServerMsg; @@ -609,14 +625,34 @@ impl ServerMsg { /// "symbols":["BTC"],"action":{"type":"cancel","cancels":[{"a":0,"o":1}]}, /// "nonce":1791424643400,"vault_address":null,"expires_after_ms":null, /// "signature":{"r":"0x1","s":"0x2","v":27}}]}"#; - /// let msg: ServerMsg = serde_json::from_str(frame).unwrap(); - /// let items = msg.mempool_items().unwrap().unwrap(); + /// let msg = ServerMsg::from_text(frame).unwrap(); + /// let items = msg.mempool_items().unwrap(); /// assert_eq!(items[0].symbols, ["BTC"]); - /// assert_eq!(items[0].action["type"], "cancel"); + /// assert_eq!(items[0].action.get(), r#"{"type":"cancel","cancels":[{"a":0,"o":1}]}"#); + /// assert_eq!(items[0].action_value().unwrap()["type"], "cancel"); /// ``` - pub fn mempool_items(&self) -> Option>> { + pub fn from_text(text: &str) -> Result { + // Only a frame that names the channel can be one; skip the extra + // parse for every other frame. + if text.contains("\"mempool\"") { + if let Ok(frame) = serde_json::from_str::(text) { + if frame.kind == "data" && frame.channel == MEMPOOL_CHANNEL { + return Ok(ServerMsg::Mempool { + coin: frame.coin, + symbol: frame.symbol, + items: frame.data, + }); + } + } + } + serde_json::from_str(text).map_err(|e| Error::Deserialize(e.to_string())) + } + + /// The items of a [`ServerMsg::Mempool`], or `None` for every other + /// message. + pub fn mempool_items(&self) -> Option<&[MempoolItem]> { match self { - ServerMsg::Data { channel, data, .. } => MempoolItem::decode(channel, data), + ServerMsg::Mempool { items, .. } => Some(items), _ => None, } } @@ -631,6 +667,20 @@ impl ServerMsg { } } +/// A [`MEMPOOL_CHANNEL`] data frame, read directly from its text so that each +/// action keeps its exact bytes (see [`ServerMsg::from_text`]). +#[derive(Deserialize)] +struct MempoolFrame { + #[serde(rename = "type")] + kind: String, + channel: String, + #[serde(default)] + coin: Option, + #[serde(default)] + symbol: Option, + data: Vec, +} + // --------------------------------------------------------------------------- // WebSocket client // --------------------------------------------------------------------------- @@ -701,7 +751,7 @@ impl OxArchiveWs { while let Some(msg) = read.next().await { match msg { Ok(Message::Text(text)) => { - if let Ok(server_msg) = serde_json::from_str::(&text) { + if let Ok(server_msg) = ServerMsg::from_text(&text) { let _ = tx.send(server_msg); } } diff --git a/tests/mempool.rs b/tests/mempool.rs index 41394c6..c63d2fb 100644 --- a/tests/mempool.rs +++ b/tests/mempool.rs @@ -69,24 +69,28 @@ async fn mempool_subscriptions_are_left_to_the_server() { ws.unsubscribe(MEMPOOL_CHANNEL, None).await.unwrap(); } -#[test] -fn unfiltered_frame_decodes_to_items() { - let msg: ServerMsg = serde_json::from_str(UNFILTERED_FRAME).unwrap(); - match &msg { - ServerMsg::Data { - channel, +/// The order action exactly as it appears in `UNFILTERED_FRAME`. +const ORDER_ACTION: &str = r#"{"type":"order","orders":[{"a":0,"b":true,"p":"83276","s":"0.40011","r":false,"t":{"limit":{"tif":"Alo"}},"c":"0x7849acc2c6c2f6f0fe4bc80ef13d1504"}],"grouping":"na"}"#; + +fn mempool_items(frame: &str) -> (Option, Option, Vec) { + match ServerMsg::from_text(frame).unwrap() { + ServerMsg::Mempool { coin, symbol, - .. - } => { - assert_eq!(channel, "mempool"); - assert_eq!((coin, symbol), (&None, &None)); - } - other => panic!("expected a data message, got {other:?}"), + items, + } => (coin, symbol, items), + other => panic!("expected a mempool message, got {other:?}"), } +} + +#[test] +fn unfiltered_frame_decodes_to_items() { + let msg = ServerMsg::from_text(UNFILTERED_FRAME).unwrap(); assert!(msg.lighter_live_data().is_none()); + assert_eq!(msg.mempool_items().map(<[MempoolItem]>::len), Some(2)); - let items = msg.mempool_items().unwrap().unwrap(); + let (coin, symbol, items) = mempool_items(UNFILTERED_FRAME); + assert_eq!((coin, symbol), (None, None)); assert_eq!(items.len(), 2); let order = &items[0]; @@ -96,9 +100,10 @@ fn unfiltered_frame_decodes_to_items() { ); assert_eq!(order.received_at_ms, Some(1791424643548)); assert_eq!(order.symbols, ["BTC"]); - assert_eq!(order.action["type"], "order"); - assert_eq!(order.action["orders"][0]["a"], 0); - assert_eq!(order.action["orders"][0]["p"], "83276"); + let action = order.action_value().unwrap(); + assert_eq!(action["type"], "order"); + assert_eq!(action["orders"][0]["a"], 0); + assert_eq!(action["orders"][0]["p"], "83276"); assert_eq!(order.nonce, Some(1791424643400)); assert_eq!(order.vault_address, None); assert_eq!(order.expires_after_ms, None); @@ -113,66 +118,100 @@ fn unfiltered_frame_decodes_to_items() { let transfer = &items[1]; assert!(transfer.symbols.is_empty()); - assert_eq!(transfer.action["type"], "usdSend"); + assert_eq!(transfer.action_value().unwrap()["type"], "usdSend"); assert_eq!(transfer.expires_after_ms, Some(1791424703400)); } +#[test] +fn the_action_keeps_its_exact_bytes_and_key_order() { + let (_, _, items) = mempool_items(UNFILTERED_FRAME); + + // Byte for byte as sent, keys in signing order (not alphabetical). + assert_eq!(items[0].action.get(), ORDER_ACTION); + assert!(items[1] + .action + .get() + .starts_with(r#"{"type":"usdSend","signatureChainId":"0xa4b1","#)); + + // Serializing an item writes the action back unchanged. + let item = serde_json::to_string(&items[0]).unwrap(); + assert!( + item.contains(&format!(r#""action":{ORDER_ACTION}"#)), + "{item}" + ); + + // The parsed form is for inspection: it sorts the keys. + let parsed = serde_json::to_string(&items[0].action_value().unwrap()).unwrap(); + assert!( + parsed.starts_with(r#"{"grouping":"na","orders":"#), + "{parsed}" + ); +} + #[test] fn symbol_frame_carries_the_subscription_symbol() { let frame = UNFILTERED_FRAME.replace( r#""coin":null,"symbol":null"#, r#""coin":"BTC","symbol":"BTC""#, ); - let msg: ServerMsg = serde_json::from_str(&frame).unwrap(); - let ServerMsg::Data { coin, symbol, .. } = &msg else { - panic!("expected a data message, got {msg:?}"); - }; + let (coin, symbol, items) = mempool_items(&frame); assert_eq!(coin.as_deref(), Some("BTC")); assert_eq!(symbol.as_deref(), Some("BTC")); - assert_eq!(msg.mempool_items().unwrap().unwrap().len(), 2); + assert_eq!(items.len(), 2); } #[test] -fn other_messages_are_not_mempool_items() { +fn other_frames_parse_as_before() { let trades = r#"{"type":"data","channel":"trades","coin":"BTC","symbol":"BTC","data":[]}"#; - let msg: ServerMsg = serde_json::from_str(trades).unwrap(); + let msg = ServerMsg::from_text(trades).unwrap(); + assert!(matches!(&msg, ServerMsg::Data { channel, .. } if channel == "trades")); assert!(msg.mempool_items().is_none()); - let pong: ServerMsg = serde_json::from_str(r#"{"type":"pong"}"#).unwrap(); + let pong = ServerMsg::from_text(r#"{"type":"pong"}"#).unwrap(); + assert!(matches!(pong, ServerMsg::Pong)); assert!(pong.mempool_items().is_none()); - assert!(MempoolItem::decode("trades", &json!([])).is_none()); assert!(LighterLiveData::decode("mempool", &json!([])).is_none()); + assert!(matches!( + ServerMsg::from_text("not json"), + Err(Error::Deserialize(_)) + )); +} + +#[test] +fn plain_serde_and_unexpected_items_fall_back_to_data() { + // Deserializing with serde alone cannot keep the raw bytes: it gives Data. + let msg: ServerMsg = serde_json::from_str(UNFILTERED_FRAME).unwrap(); + assert!(matches!(&msg, ServerMsg::Data { channel, .. } if channel == "mempool")); + assert!(msg.mempool_items().is_none()); + + // A mempool frame whose items do not match is delivered unchanged. + let odd = + r#"{"type":"data","channel":"mempool","coin":null,"symbol":null,"data":{"not":"a list"}}"#; + let msg = ServerMsg::from_text(odd).unwrap(); + assert!(matches!(&msg, ServerMsg::Data { data, .. } if data["not"] == "a list")); } #[test] fn items_tolerate_nulls_new_fields_and_new_action_types() { - let data = json!([ - {"received_at": null, "received_at_ms": null, "symbols": [], - "action": {"type": "someFutureAction", "x": 1}, "nonce": null, - "vault_address": "0x00000000000000000000000000000000000000aa", - "expires_after_ms": null, "signature": null, "a_future_field": true}, - {"action": {"type": "noop"}} - ]); - let items = MempoolItem::decode(MEMPOOL_CHANNEL, &data) - .unwrap() - .unwrap(); + let frame = r#"{"type":"data","channel":"mempool","coin":null,"symbol":null,"data":[ +{"received_at":null,"received_at_ms":null,"symbols":[],"action":{"x":1,"type":"someFutureAction"},"nonce":null,"vault_address":"0x00000000000000000000000000000000000000aa","expires_after_ms":null,"signature":null,"a_future_field":true}, +{"action":{"type":"noop"}} +]}"#; + let (_, _, items) = mempool_items(frame); assert_eq!(items[0].received_at, None); assert_eq!(items[0].nonce, None); assert_eq!(items[0].signature, None); - assert_eq!(items[0].action["type"], "someFutureAction"); + assert_eq!( + items[0].action.get(), + r#"{"x":1,"type":"someFutureAction"}"# + ); assert_eq!( items[0].vault_address.as_deref(), Some("0x00000000000000000000000000000000000000aa") ); assert!(items[1].symbols.is_empty()); - assert_eq!(items[1].action, json!({"type": "noop"})); -} - -#[test] -fn a_payload_that_is_not_a_list_of_items_is_an_error() { - let result = MempoolItem::decode(MEMPOOL_CHANNEL, &json!({"not": "a list"})).unwrap(); - assert!(matches!(result, Err(Error::Deserialize(_)))); + assert_eq!(items[1].action_value().unwrap(), json!({"type": "noop"})); } #[test] From c724b6b7a08a4349244bcd9424e61303bb923dcb Mon Sep 17 00:00:00 2001 From: 0xFantomMenace <93462858+0xFantomMenace@users.noreply.github.com> Date: Thu, 8 Oct 2026 03:16:37 +0000 Subject: [PATCH 3/3] Clarify mempool endpoint, plan and ordering wording - A capability row without ws_endpoint is served at the default endpoint, wss://api.0xarchive.io/ws; without plans, on every plan. - The unfiltered stream carries the pending transactions our node receives, and pending transactions are never stored. - The parsed action may not keep the original key order; the raw action does. --- CHANGELOG.md | 12 +++++++----- README.md | 19 ++++++++++--------- src/error.rs | 3 ++- src/types.rs | 12 ++++++------ src/ws/client.rs | 18 ++++++++++-------- tests/mempool.rs | 9 +++------ 6 files changed, 38 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca6883c..b7a7cab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,8 +11,9 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com). they are included in a block, for every Hyperliquid product. It is live only, served only at `wss://stream.0xarchive.io/ws`, and included with the Pro, Scale and Enterprise plans. Every other channel stays on every plan. - `subscribe(MEMPOOL_CHANNEL, None)` streams every pending transaction; - with a symbol, only the actions that reference that market. `mempool` is + `subscribe(MEMPOOL_CHANNEL, None)` streams every pending transaction our + Hyperliquid node receives; with a symbol, only the actions that reference + that market. `mempool` is the only channel whose symbol is optional. - `ServerMsg::Mempool`, with one `MempoolItem` (and its `MempoolSignature`) per signed action. `ServerMsg::from_text()`, which the client uses to read @@ -20,7 +21,7 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com). `MempoolItem::action` (a `Box`) keeps the exact bytes the server sent, key order included, for signature recovery. `MempoolItem::action_value()` parses the action into a `serde_json::Value` - for inspection, which sorts the keys. `ServerMsg::mempool_items()` returns + for inspection, which may not keep the original key order. `ServerMsg::mempool_items()` returns a message's items. Unknown item fields are ignored. - serde_json's `raw_value` feature is enabled. - `ws::STREAM_WS_URL` (`wss://stream.0xarchive.io/ws`) and @@ -28,8 +29,9 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com). Pass `STREAM_WS_URL` to `WsOptions::ws_url()` for a connection that subscribes to `mempool`. - `Capability::ws_endpoint` and `Capability::plans`, the fields - `/v1/capabilities` sets on the `mempool` row only. `None` means every - endpoint and every plan. They are left out when serializing a row that + `/v1/capabilities` sets on the `mempool` row only. For `ws_endpoint`, + `None` means the default endpoint, `wss://api.0xarchive.io/ws`. For + `plans`, `None` means every plan. They are left out when serializing a row that does not have them. ### Changed diff --git a/README.md b/README.md index 42c8ee9..4ff8df0 100644 --- a/README.md +++ b/README.md @@ -196,10 +196,10 @@ println!("{} symbols, {} on HIP-3", symbols.len(), hip3.len()); one `Capability` row per venue and datatype: its REST routes, its WebSocket channels, whether those stream live (`live`) and replay history (`replay`), the first instant served (`available_from`), the cadence, the largest page -(`page_limit`), and the accepted `interval` values. A row whose channels are -served at one endpoint only, or included with some plans only, names them in -`ws_endpoint` and `plans`; only the `mempool` row has them, and elsewhere both -are `None`. The route is public and uses no credits. Read replay availability from it rather than from a fixed +(`page_limit`), and the accepted `interval` values. The `mempool` row is the one +that sets `ws_endpoint` (the endpoint that serves it) and `plans` (the plans that +include it); rows without them use the default endpoint, `wss://api.0xarchive.io/ws`, +and are on every plan. The route is public and uses no credits. Read replay availability from it rather than from a fixed list: ```rust @@ -1968,7 +1968,8 @@ modifies, TWAPs, leverage changes, transfers and every other action type) as our Hyperliquid node receives them from its peers, before they are included in a block. It covers every Hyperliquid product: perps, HIP-3, HIP-4 and spot. -- **Live only.** There is no replay, history, REST route or export. +- **Live only.** Pending transactions are never stored. There is no replay, history, + REST route or export. - **One endpoint.** It is served only at `wss://stream.0xarchive.io/ws` (`oxarchive::ws::STREAM_WS_URL`), with the same API key and protocol. At `wss://api.0xarchive.io/ws`, the default `ws_url`, a subscribe is answered with @@ -1987,14 +1988,14 @@ let mut rx = ws.rx.take().expect("receiver"); ws.subscribe(MEMPOOL_CHANNEL, Some("BTC")).await?; // actions that reference BTC ws.subscribe(MEMPOOL_CHANNEL, Some("xyz:TSLA")).await?; // HIP-3; spot "HYPE-USDC", HIP-4 "#49720" -// ws.subscribe(MEMPOOL_CHANNEL, None).await?; // every pending transaction (unfiltered) +// ws.subscribe(MEMPOOL_CHANNEL, None).await?; // pending transactions our node receives (unfiltered) while let Some(msg) = rx.recv().await { match msg { ServerMsg::Mempool { symbol, items, .. } => { for item in items { // item.action.get() is the action's exact bytes, for signature recovery. - let action = item.action_value()?; // parsed for inspection (sorted keys) + let action = item.action_value()?; // parsed, for inspection only println!("{symbol:?} {:?} {} {:?}", item.received_at, action["type"], item.symbols); } } @@ -2005,7 +2006,7 @@ while let Some(msg) = rx.recv().await { ``` The symbol is optional on this channel only. Without it you receive every pending -transaction; with it, every action whose asset ids include that market, whole (an +transaction our Hyperliquid node receives; with it, every action whose asset ids include that market, whole (an order batch that touches `BTC` and `ETH` reaches both subscriptions). Symbols are spelled as everywhere else: perps `BTC`, HIP-3 `xyz:TSLA`, spot `HYPE-USDC` (`HYPE/USDC` is also accepted) and HIP-4 `#49720`. An unknown symbol is answered with @@ -2022,7 +2023,7 @@ action in `items`: | `received_at` | When our node received the transaction: an RFC 3339 UTC string with nanosecond precision. Not a block time. | | `received_at_ms` | The same time in Unix milliseconds. | | `symbols` | Markets the action's asset ids reference, in first-seen order without repeats. Empty for actions with no market, such as transfers, `noop`, `scheduleCancel` and validator actions. | -| `action` | The action exactly as signed, in Hyperliquid's exchange-action format: asset ids (`a` or `asset`) rather than symbols, prices and sizes as strings. It is a `Box` that keeps the exact bytes the server sent, key order included, for signature recovery (`item.action.get()`). `item.action_value()` parses it into a `serde_json::Value` for inspection, which sorts the keys. | +| `action` | The action exactly as signed, in Hyperliquid's exchange-action format: asset ids (`a` or `asset`) rather than symbols, prices and sizes as strings. It is a `Box` that keeps the exact bytes the server sent, key order included, for signature recovery (`item.action.get()`). `item.action_value()` parses it into a `serde_json::Value` for inspection, which may not keep the original key order. | | `nonce` | The action's nonce. | | `vault_address` | The vault or subaccount the action acts for, or `None`. | | `expires_after_ms` | The action's `expiresAfter` in Unix milliseconds, or `None`. | diff --git a/src/error.rs b/src/error.rs index ea59137..27b33d9 100644 --- a/src/error.rs +++ b/src/error.rs @@ -53,7 +53,8 @@ pub enum ErrorCode { /// Missing or invalid credentials. Unauthorized, /// The credentials do not grant access to this resource, or the plan does - /// not include it (the `mempool` WebSocket channel below Pro). + /// not include it (the `mempool` WebSocket channel on plans other than + /// Pro, Scale and Enterprise). Forbidden, /// The monthly credits are spent. InsufficientCredits, diff --git a/src/types.rs b/src/types.rs index 8daed62..cafc349 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1317,7 +1317,7 @@ pub struct MempoolItem { /// included, as signature recovery needs them; `action.get()` returns /// them as a `&str`. To inspect the action, /// [`action_value`](Self::action_value) parses it into a - /// `serde_json::Value`, which sorts the keys. + /// `serde_json::Value`, which may not keep the original key order. pub action: Box, /// The action's nonce. pub nonce: Option, @@ -1345,8 +1345,8 @@ impl MempoolItem { /// The action parsed into a `serde_json::Value`, for inspection, for /// example `item.action_value()?["type"]`. /// - /// The parsed form sorts object keys, so it no longer has the signed key - /// order. Use the raw [`action`](Self::action) for signature recovery. + /// The parsed form may not keep the original key order. Use the raw + /// [`action`](Self::action) for signature recovery. pub fn action_value(&self) -> crate::Result { serde_json::from_str(self.action.get()) .map_err(|e| crate::Error::Deserialize(e.to_string())) @@ -2453,9 +2453,9 @@ pub struct Capability { /// Further detail, such as how replay behaves. #[serde(default)] pub notes: Option, - /// The only WebSocket endpoint that serves this row's channels, when that - /// is not every endpoint: `wss://stream.0xarchive.io/ws` for `mempool`. - /// `None` means every endpoint. + /// The WebSocket endpoint that serves this row's channels, when it is not + /// the default endpoint: `wss://stream.0xarchive.io/ws` for `mempool`. + /// `None` means the default endpoint, `wss://api.0xarchive.io/ws`. #[serde(default, skip_serializing_if = "Option::is_none")] pub ws_endpoint: Option, /// The plans that include it, when that is not every plan: `pro`, `scale` diff --git a/src/ws/client.rs b/src/ws/client.rs index 6bcd281..088e8a8 100644 --- a/src/ws/client.rs +++ b/src/ws/client.rs @@ -33,7 +33,8 @@ pub const DEFAULT_WS_URL: &str = "wss://api.0xarchive.io/ws"; /// /// It serves a subset of the live channels; a channel it does not serve, and /// every replay, is answered with a [`ServerMsg::Error`] that names -/// [`DEFAULT_WS_URL`]. Use one connection per endpoint: +/// [`DEFAULT_WS_URL`]. [`MEMPOOL_CHANNEL`] has no replay on any endpoint. +/// Use one connection per endpoint: /// /// ```no_run /// # use oxarchive::ws::{OxArchiveWs, WsOptions, MEMPOOL_CHANNEL, STREAM_WS_URL}; @@ -58,9 +59,9 @@ pub const STREAM_WS_URL: &str = "wss://stream.0xarchive.io/ws"; /// is on every plan. /// /// The symbol is optional on this channel only. `subscribe(MEMPOOL_CHANNEL, -/// None)` streams every pending transaction; with a symbol (`BTC`, -/// `xyz:TSLA`, `HYPE-USDC`, `#49720`) it streams every action that references -/// that market, whole. The unfiltered stream is several megabytes a second +/// None)` streams every pending transaction our Hyperliquid node receives; +/// with a symbol (`BTC`, `xyz:TSLA`, `HYPE-USDC`, `#49720`) it streams every +/// action that references that market, whole. The unfiltered stream is several megabytes a second /// before compression, and the server limits unfiltered subscriptions: when /// they are at capacity, a subscribe without a symbol is answered with /// [`ErrorCode::RateLimited`]. Subscribe with a symbol where you can. @@ -421,7 +422,8 @@ pub enum ServerMsg { /// serve the channel or operation; the message names the one that does). /// A channel that does not offer a mode (live, replay, `replay.seek`) /// answers `unsupported_for_venue`, and a channel your plan does not - /// include ([`MEMPOOL_CHANNEL`] below Pro) answers `forbidden`. + /// include ([`MEMPOOL_CHANNEL`] on plans other than Pro, Scale and + /// Enterprise) answers `forbidden`. Error { message: String, /// Stable machine-readable reason, `None` when the server sent none. @@ -442,7 +444,7 @@ pub enum ServerMsg { data: serde_json::Value, }, /// A [`MEMPOOL_CHANNEL`] data message: one batch of pending transactions, - /// with one [`MempoolItem`] per signed action, in the order received. + /// with one [`MempoolItem`] per signed action in the batch. /// `coin` and `symbol` are the subscription's symbol, or `None` on the /// unfiltered stream. /// @@ -802,8 +804,8 @@ impl OxArchiveWs { /// request for stored history. /// /// On [`MEMPOOL_CHANNEL`] the symbol is optional: `None` subscribes to - /// every pending transaction. That channel is served only at - /// [`STREAM_WS_URL`]. + /// every pending transaction our Hyperliquid node receives. That channel + /// is served only at [`STREAM_WS_URL`]. pub async fn subscribe(&self, channel: &str, symbol: Option<&str>) -> Result<()> { if is_lighter_replay_only_channel(channel) { return Err(Error::InvalidParam(LIGHTER_SUBSCRIPTION_ERROR.to_string())); diff --git a/tests/mempool.rs b/tests/mempool.rs index c63d2fb..4d7ea28 100644 --- a/tests/mempool.rs +++ b/tests/mempool.rs @@ -140,12 +140,9 @@ fn the_action_keeps_its_exact_bytes_and_key_order() { "{item}" ); - // The parsed form is for inspection: it sorts the keys. - let parsed = serde_json::to_string(&items[0].action_value().unwrap()).unwrap(); - assert!( - parsed.starts_with(r#"{"grouping":"na","orders":"#), - "{parsed}" - ); + // The parsed form is for inspection and holds the same content. + let parsed: serde_json::Value = serde_json::from_str(ORDER_ACTION).unwrap(); + assert_eq!(items[0].action_value().unwrap(), parsed); } #[test]