Rust SDK

The oracle-client crate is the consumer SDK. It connects to every node in the registry, verifies each signed frame, and resolves one price for each feed across the nodes. Use it when your Rust service must not trust one node or one operator.

The crate has two pure, synchronous parts and a thin async layer:

Add the crate

The crate name is oracle-client. It is not on crates.io, so add it as a git dependency on a release tag. The API also uses types from oracle-wire and oracle-types, so add those two crates from the same tag.

[dependencies]
oracle-client = { git = "https://github.com/markoutlabs/oracle-node", tag = "v0.1.0" }
oracle-wire = { git = "https://github.com/markoutlabs/oracle-node", tag = "v0.1.0" }  # Signer, Address, Registry
oracle-types = { git = "https://github.com/markoutlabs/oracle-node", tag = "v0.1.0" } # Fixed, Symbol, InstrumentId
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

Use the same tag for the three crates. Different tags give two copies of the shared types, and the compiler rejects the mix.

Requirements:

• Rust 1.85 or newer. The workspace uses edition 2024.
• A Tokio runtime. subscribe starts one task for each node on the current runtime.
• TLS through rustls with the ring provider. HttpRegistrySource::new installs ring as the process default, unless your application installed a provider first.

Connect to mainnet

The module oracle_client::mainnet holds the trust root of the mainnet registry canister.

The first two constants are Options. A build without a mainnet deployment has None.

use std::sync::Arc;

use oracle_client::{HttpRegistrySource, OracleClient, Status, mainnet};
use oracle_wire::{Address, Signer};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let canister = mainnet::CANISTER_ID.ok_or("this build has no mainnet canister")?;
    let governance: Address = mainnet::GOVERNANCE_ADDRESS
        .ok_or("this build has no mainnet governance address")?
        .parse()?;
    let source = Arc::new(HttpRegistrySource::canister(canister)?);

    // Hex, as scripts/client-key.sh writes it.
    let api_key = Signer::from_hex(std::fs::read_to_string("client.key")?.trim())?;

    let mut client = OracleClient::builder_from_source(source, governance)
        .await?
        .client_key(api_key)
        .build()?;

    client.subscribe(&["CRYPTO.BTC/USD"]).await?;
    while let Some(price) = client.next().await {
        match price.status {
            Status::Ok | Status::Degraded => {
                // Act on `price.conf`, not only on `price.price`.
                println!("{:?} ± {:?}", price.price, price.conf);
            }
            Status::Halted => println!("halted: {:?}", price.concerns),
        }
    }
    Ok(())
}

The example does these steps:

1. HttpRegistrySource::canister points at https://<canister>.raw.icp0.io/registry.json. Each fetch checks the IC certificate against the mainnet root key. The certificate must be at most 5 minutes old.
2. builder_from_source fetches the registry and verifies the governance signature against the pinned address.
3. build checks the quorum against the registry.
4. subscribe connects to each node, runs the handshake with the client key, and sends the subscription.

Pin the governance address in your code or your config. Never read it from the document that you fetch. The certificate shows that the document is current. The signature shows that the document is genuine.

Connect to a local network

On a local network, give the canister id, the gateway URL and the root key of that network. icp network status --json prints the gateway URL and the root key.

use std::sync::Arc;

use oracle_client::{Certification, HttpRegistrySource, OracleClient};
use oracle_wire::{Address, Signer};

async fn local_client(
    canister_id: &str,
    root_key: &str,
    governance: Address,
    api_key: Signer,
) -> Result<OracleClient, Box<dyn std::error::Error>> {
    let certification = Certification::new(canister_id, root_key)?;
    let url = format!("http://127.0.0.1:8000/registry.json?canisterId={canister_id}");
    let source = Arc::new(HttpRegistrySource::new(&url, certification)?);

    let client = OracleClient::builder_from_source(source, governance)
        .await?
        .client_key(api_key)
        // One local node: no protection against a bad node.
        .quorum(1)
        .i_understand_this_provides_no_protection_against_a_bad_node()
        .build()?;
    Ok(client)
}

You can also pin a registry file. A file has no source, so the client does not refresh it. The client then loses a node that rotates its key, until you call refresh_registry with the new document.

let registry: oracle_wire::Registry =
    serde_json::from_str(&std::fs::read_to_string("registry.json")?)?;
let builder = OracleClient::builder(registry.verify(governance, 0)?);

The client API key

Nodes serve only listed clients. Without a key, subscribe, subscribe_venues, subscribe_books and subscribe_funding return ClientError::NoClientKey.

• The key is a secp256k1 key, an oracle_wire::Signer. It only signs.
• Governance lists the address of the key with add_client on the registry canister.
• Signer::from_hex reads the hex file that scripts/client-key.sh writes.
• Use the key only for this protocol.

The SDK runs the handshake on each connection. It checks that the node signed auth_ok with the address that the client dialled. Only then does it send the subscription. Every frame that it decrypts still goes through the Verifier. Decryption proves nothing, because every listed client holds the broadcast key.

Client access tells how to get a key listed and what the transport does.

Builder settings

OracleClient::builder(registry) and OracleClient::builder_from_source(source, governance) return an OracleClientBuilder.

use std::time::Duration;

use oracle_types::Fixed;

let mut client = OracleClient::builder_from_source(source, governance)
    .await?
    .client_key(api_key)
    .quorum(3)
    .max_age(Duration::from_secs(2))
    .max_skew(Duration::from_millis(500))
    .max_divergence_bps(Fixed::from_int(50)?)
    .refresh_interval(Duration::from_secs(10))
    .registry_poll_interval(Duration::from_secs(60))
    .max_registry_age(Duration::from_secs(48 * 3_600))
    .build()?;

build returns one of these errors:

Quorum

The quorum is the number of distinct nodes that must report a feed before the SDK gives a usable price. The default is 3.

build refuses a quorum below 3 unless you call i_understand_this_provides_no_protection_against_a_bad_node(). Signatures from one node confirm who is talking. They tell you nothing about whether to believe that node. One compromised node is then one compromised price. This mistake is easy to make during integration and easy to never revisit, so the API makes it loud. Use the acknowledgement only for local development against one node.

Other quorum rules:

• The registry carries quorum.min_nodes and quorum.recommended as guidance from governance. The SDK does not apply them. You set the quorum.
• refresh_registry refuses a new registry version that lists fewer nodes than your quorum. The client keeps its current registry.
• Five nodes give a quorum of 3 room for one planned and one unplanned outage.

Why a quorum below 3 is refused has the full argument.

Subscribe and read events

There are four subscriptions. Each one takes feed symbols.

Each call adds symbols and reconnects every node with the full set. subscribe_all takes every list in one Subscription, with one reconnect of each node. A bad symbol in any list changes nothing.

next_event() returns the next Event. next() returns only prices and skips venue reports, books and funding. Keep one of them in a loop at all times. The client adopts scheduled registry fetches inside that loop. Both return None only after shutdown.

next_event() is cancel-safe: a call that you drop loses no frame. A registry fetch, for a key that the client does not know yet, runs on its own task. So next_event() can share a tokio::select! with a timer.

use oracle_client::{Event, Subscription};
use oracle_types::Fixed;

// One reconnect of each node for every list.
client
    .subscribe_all(Subscription {
        prices: &["CRYPTO.BTC/USD"],
        venues: &["CRYPTO.BTC/USDT"],
        books: &["CRYPTO.BTC/USDT"],
        ..Subscription::default()
    })
    .await?;

while let Some(event) = client.next_event().await {
    match event {
        Event::Price(resolved) => {
            if !resolved.is_usable() {
                // Halted: never settle or liquidate on this feed now.
                continue;
            }
            // Ok or Degraded: size the risk from `conf`.
        }
        Event::Venues(accepted) => {
            // One node's view. The SDK does not combine reports.
            let report = &accepted.report;
            for entry in &report.entries {
                let bid = entry
                    .bid
                    .map(|mantissa| Fixed::from_scaled(mantissa, report.expo))
                    .transpose()?;
                println!(
                    "{} {}:{} {} {} bps bid={bid:?}",
                    accepted.signer, entry.venue, entry.market, entry.status, entry.weight_bps
                );
            }
        }
        Event::Book(_) => {
            // Already applied. Read the book with `client.book()`.
        }
    }
}

The Event enum:

Rules for the event loop:

• The client yields one resolution for each accepted frame. You see a change as soon as one node reports it.
• You observe state, not transitions. A node conflates its fanout, so a halt shorter than one second can disappear. Read status on each event. Do not count transitions.
• A rejected frame never reaches you. The client logs it at debug level and counts it in stats().
• resolve(&symbol, now_ns) resolves a feed again without a new frame. Use it on a timer, so that a silent upstream shows as Halted.

Resolved

Resolved is the price for one feed, resolved across the nodes. It serializes with serde.

Resolved::is_usable() returns true for Ok and Degraded.

A healthy price as JSON:

{
  "feed_id": "0x0523e79c7ed211911a19afaa4def68c896bc72d674bf33c5c64d4d7089e320e2",
  "symbol": "CRYPTO.BTC/USD",
  "status": "ok",
  "price": "86599.95",
  "conf": "19.71359135",
  "nodes": 3,
  "halted_nodes": 0,
  "concerns": [],
  "publish_time_ns": "1791299498663252000",
  "spread_bps": "0"
}

A halted price as JSON. One node reported its feed halted, and too few nodes remain:

{
  "feed_id": "0x0523e79c7ed211911a19afaa4def68c896bc72d674bf33c5c64d4d7089e320e2",
  "symbol": "CRYPTO.BTC/USD",
  "status": "halted",
  "price": null,
  "conf": null,
  "nodes": 2,
  "halted_nodes": 1,
  "concerns": ["node_reported_halted", "insufficient_nodes"],
  "publish_time_ns": "0",
  "spread_bps": null
}

Decimals and publish_time_ns are strings. A JavaScript number loses precision on a nanosecond timestamp. A Fixed prints with up to 12 decimal places, without trailing zeros.

Status and concerns

Concern is #[non_exhaustive]. A match outside the crate needs a wildcard arm:

use oracle_client::{Concern, Resolved};

fn review(resolved: &Resolved) {
    for concern in &resolved.concerns {
        match concern {
            Concern::InsufficientNodes => { /* halted: stop */ }
            Concern::NodeDivergence => { /* watch spread_bps */ }
            Concern::NodeReportedDegraded => { /* a node degraded its own feed */ }
            Concern::NodeReportedHalted => { /* a node halted and was left out */ }
            Concern::RegistryStale => { /* check the path to the canister */ }
            // `Concern` is `#[non_exhaustive]`.
            _ => {}
        }
    }
}

Status flags explains why a node sets DEGRADED or HALTED on its own update.

Resolved and PriceUpdate

A PriceUpdate is the payload that one node signs. A Resolved is what the SDK makes from the latest accepted update of each node. Use Resolved for decisions.

client.view(&symbol) returns the FeedView behind a feed. FeedView::latest() gives each node's latest accepted PriceUpdate, keyed by node address. Use it to monitor which node says what. Do not use one node's update as a price.

use oracle_client::{OracleClient, now_unix_nanos};
use oracle_types::Symbol;

fn per_node(client: &OracleClient) -> Result<(), Box<dyn std::error::Error>> {
    let symbol: Symbol = "CRYPTO.BTC/USD".parse()?;
    if let Some(view) = client.view(&symbol) {
        for (node, update) in view.latest() {
            println!("{node} seq={} price={}", update.seq, update.price_fixed()?);
        }
    }
    // The same feed, resolved now without a new frame.
    let _now = client.resolve(&symbol, now_unix_nanos());
    Ok(())
}

Books

Each node sends the venue books that it keeps. A snapshot of the top 50 levels goes out every second. The changed levels go out in between, at most every 50 ms. The format is the same for every venue, because the node computes the changes from its own book. Streams describes the frames.

The SDK keeps one Book for each pair of node and venue instrument:

• It verifies each frame like a price: signature, registry, a rising seq, and the clock.
• A missed seq makes that node's book invalid until its next snapshot.
• client.book(&instrument, now_ns) returns the freshest valid book whose last frame is younger than max_age. It is always one node's whole book, never a mix of two nodes.
• No quorum applies to a book. It is the view of one node.

use oracle_client::{Book, OracleClient, now_unix_nanos};
use oracle_types::{InstrumentId, MarketKind, Venue};

fn top_of_book(client: &OracleClient) -> Result<(), Box<dyn std::error::Error>> {
    let binance = InstrumentId::new(Venue::Binance, MarketKind::Spot, "BTCUSDT");
    if let Some(book) = client.book(&binance, now_unix_nanos()) {
        if let (Some(bid), Some(ask)) = (book.best_bid(), book.best_ask()) {
            let (bid_price, bid_size) = Book::level_fixed(bid)?;
            let (ask_price, ask_size) = Book::level_fixed(ask)?;
            println!(
                "{binance} from {}: {bid_size} @ {bid_price} / {ask_size} @ {ask_price}",
                book.node
            );
        }
    }
    // Every book that a node has sent.
    for instrument in client.book_instruments() {
        println!("{instrument}");
    }
    Ok(())
}

An instrument is a venue, a market and the venue's own symbol, such as BTCUSDT on Binance or BTC-USDT on OKX. The node config sets them. book_instruments() lists the instruments that arrived.

A BookLevel holds price: i64 and qty: u64. Both are mantissas at BOOK_EXPO, which is −8. Sizes are in base units.

Venue reports

A venue report shows what one node saw for each source of a feed. The node signs it with its node key. It contains:

• the best bid and ask, with their sizes;
• the impact prices for the feed's impact_notional;
• the mark, the index and the funding rate, when the venue sends them;
• the status of the source in the price, and its weight.

The SDK verifies each report like a price: signature, registry, the report's own seq, and the clock. It does not combine reports across nodes, because each node reports what it saw.

Each VenueEntry in report.entries has these fields:

A field that the venue does not send for the instrument is None. The prices are the venue's own, before the USDT to USD conversion. The status and the weight refer to the converted price.

A node signs reports only for a feed that some client subscribed to. The first report comes at the next evaluation after the subscription.

Latency

client.latency() returns a LatencyReport for the updates that the client accepted. It has three spans. The client stamps each frame when it arrives, before it decrypts the frame.

Each span is a Percentiles value with count, p50, p90, p99 and max. Its Display prints a line such as p50 0.28 ms, p90 0.59 ms, p99 1.51 ms, max 6.33 ms (n=1059).

let latency = client.latency();
println!("delivery    {}", latency.delivery);
println!("data age    {}", latency.data_age);
println!("processing  {}", latency.processing);

// Start again after a warm-up.
client.reset_latency();

• delivery compares two clocks. It is exact only on one machine, or with synchronized clocks.
• data_age includes heartbeats, which send old data on purpose. It skips halted updates.
• A value above one minute counts as one minute.

Latency gives measured values.

Counters and the registry

client.stats() returns a ClientStats value. Export it to your monitoring.

A sudden rise in replayed or out_of_time means that something sits between you and the node. No signature check tells you that. A few retired frames right after a rotation are frames in flight. A steady stream of them is a stolen key in use.

Registry state:

A node rotation reaches the client without a restart. An unknown signer causes one registry fetch, at most once each refresh_interval. The client adopts the newer version in place and moves the node's quorum seat to the new address. Key rotation has the full sequence.

Funding

Each node sends the funding of each perpetual source. A frame holds the rate as the venue states it, with its basis. Where the venue sends them, it also holds the predicted rate, the interval, the next funding time, the last settled rate and the open interest. A frame goes out on a change, at most once a second, and at least once a minute. FundingUpdate v1 lists the fields of each venue.

• The SDK verifies each frame like a book frame: signature, registry, a rising seq, and the clock. The age limit is two minutes, because a new subscriber gets the latest frame, which can be a minute old.
• Each frame carries the whole state. client.funding(&perpetual, now_ns) returns the freshest frame of any node, as AcceptedFunding, with the node that signed it.
• No quorum applies. It is the view of one node.

use oracle_client::{OracleClient, now_unix_nanos};
use oracle_types::{FUNDING_EXPO, InstrumentId, MarketKind, Venue};

fn okx_funding(client: &OracleClient) {
    let okx = InstrumentId::new(Venue::Okx, MarketKind::LinearPerp, "BTC-USDT-SWAP");
    if let Some(funding) = client.funding(&okx, now_unix_nanos()) {
        let update = &funding.update;
        // A mantissa at FUNDING_EXPO (−12), over the venue's interval.
        println!("{} e{FUNDING_EXPO}, every {:?} ns", update.rate, update.interval_ns);
    }
    // Every perpetual that a node has sent funding for.
    for perpetual in client.funding_instruments() {
        println!("{perpetual}");
    }
}

Frame copies for an archive

client.frames(capacity) opens a bounded channel. It carries a copy of each frame whose signature and registry check pass, also when the client then refuses the frame. Each copy is a TappedFrame:

The client never waits on the channel. When the channel is full, the client drops the copy and counts it in tap_dropped, so a slow archive never slows the prices. client.frames_of(capacity, &[FrameKind::VenueReport]) copies only the frames of the listed kinds. tap_dropped then counts only those. A Replayed copy with other bytes than the stored frame is a node that signed two frames with one seq.

Node reports

client.nodes(now_ns) reports each node in the registry, for monitoring. The quorum, not this report, decides a price.

use oracle_client::{ConnectionState, NodeFeedState, now_unix_nanos};

for node in client.nodes(now_unix_nanos()) {
    let connected = matches!(node.connection, ConnectionState::Connected { .. });
    for feed in &node.feeds {
        if feed.state == Some(NodeFeedState::Stale) {
            println!("{} is stale on {}", node.address, feed.symbol);
        }
    }
    println!("{} connected={connected} accepted={}", node.address, node.frames.accepted);
}

A frame from an unknown or retired key belongs to no node, so only stats() counts it. NodeReport serializes with serde. oracle-proxy sends it as the nodes message.

Test your integration

The module oracle_client::testing has fake nodes and registries. Use it to test how your app handles halted and degraded feeds without five live nodes. Its keys come from small integers, so anyone can guess them. Never use them in production.

use oracle_client::testing::{REGISTRY_ISSUED_AT, TestNode, registry_with, update_at};
use oracle_client::{OracleClient, Status};
use oracle_types::NANOS_PER_SEC;

#[test]
fn three_agreeing_nodes_resolve_to_ok() -> Result<(), Box<dyn std::error::Error>> {
    let nodes: Vec<TestNode> = (1..=3_u8).map(TestNode::new).collect();
    let registry = registry_with(&nodes.iter().collect::<Vec<_>>());
    let mut client = OracleClient::builder(registry).quorum(3).build()?;

    // Test registries are issued at this time. Use it as the clock.
    let now = REGISTRY_ISSUED_AT * NANOS_PER_SEC;
    let mut last = None;
    for node in &nodes {
        last = Some(client.ingest(&node.sign(&update_at(now, 1)), now)?);
    }
    assert_eq!(last.map(|resolved| resolved.status), Some(Status::Ok));
    Ok(())
}

ingest and ingest_event feed one frame in directly, without the network. They return the RejectReason when a frame fails verification.