Rust SDK

The Arete Rust SDK is an asynchronous client for consuming streaming data from Arete. It is designed for backend services, trading bots, and CLI tools that require type-safe access to Solana state projections.

Installation

Add to your Cargo.toml:

[dependencies]
a4-sdk = "0.3.0"
a4-stacks = { version = "0.3.0", optional = true }
tokio = { version = "1", features = ["full"] }
anyhow = "1"
futures = "0.3"

The a4-stacks crate provides pre-built stack definitions for popular Solana protocols (optional but recommended).

TLS Options

By default, the SDK uses rustls for TLS. You can switch to native TLS:

[dependencies]
a4-sdk = { version = "0.3.0", default-features = false, features = ["native-tls"] }

Tokio Runtime

The SDK requires the Tokio runtime. Ensure you have it enabled in your project (specifically the rt-multi-thread, macros, and time features).

Quick Start

Connect and Stream

use a4_sdk::prelude::*;
use a4_stacks::ore::{OreStack, OreRound};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Connect to the ORE stack (URL is defined in OreStack)
    let a4 = Arete::<OreStack>::connect().await?;

    println!("Connected! Streaming ORE rounds...\n");

    // Access views directly
    let mut stream = a4.views.ore_round.latest().listen();

    // Stream updates
    while let Some(round) = stream.next().await {
        println!("Round # {:?}", round.id.round_id);
        println!("  Motherlode: {:?}", round.state.motherlode);
        println!("  Total difficulty: {:?}\n", round.state.total_difficulty);
    }

    Ok(())
}

Run with:

cargo run

Project Setup

1. Create a New Project

cargo new my-arete-app
cd my-arete-app

2. Add Dependencies

[dependencies]
a4-sdk = "0.3.0"
a4-stacks = "0.3.0"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
anyhow = "1"
futures = "0.3"

3. Basic Structure

use a4_sdk::prelude::*;
use a4_stacks::ore::{OreStack, OreRound};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let a4 = Arete::<OreStack>::connect().await?;

    let mut stream = a4.views.ore_round.latest().listen();

    while let Some(round) = stream.next().await {
        println!("Round: {:?}", round.id.round_id);
    }

    Ok(())
}

Connection Management

Basic Connection

Each stack defines its own URL, so connection is simple:

let a4 = Arete::<OreStack>::connect().await?;

With Custom URL

Override the default URL if needed:

let a4 = Arete::<OreStack>::connect_url("wss://custom.endpoint.com").await?;

Builder Pattern

Configure the client with custom reconnection logic and intervals:

use std::time::Duration;

let a4 = Arete::<OreStack>::builder()
    .url("wss://custom.endpoint.com")  // Optional: override default
    .auto_reconnect(true)
    .max_reconnect_attempts(10)
    .reconnect_intervals(vec![
        Duration::from_secs(1),
        Duration::from_secs(2),
        Duration::from_secs(5),
    ])
    .ping_interval(Duration::from_secs(30))
    .initial_data_timeout(Duration::from_secs(5))
    .max_entries_per_view(5000)
    .connect()
    .await?;

Connection State

// Check current state
match a4.connection_state().await {
    ConnectionState::Connected => println!("Connected!"),
    ConnectionState::Connecting => println!("Connecting..."),
    ConnectionState::Disconnected => println!("Disconnected"),
    ConnectionState::Reconnecting { attempt } => println!("Reconnecting (attempt {})...", attempt),
    ConnectionState::Error => println!("Error"),
}

Disconnect

// Graceful disconnect
a4.disconnect().await;

Store Size Limits

By default, each view is limited to 10,000 entries to prevent memory issues on long-running clients. When the limit is reached, oldest entries are evicted (LRU).

// Custom limit
let a4 = Arete::<OreStack>::builder()
    .max_entries_per_view(5000)
    .connect()
    .await?;

// Unlimited (not recommended for long-running clients)
let a4 = Arete::<OreStack>::builder()
    .unlimited_entries()
    .connect()
    .await?;

Views

Views provide typed access to your stack’s data. Access them directly through a4.views:

// Direct field access
let rounds = a4.views.ore_round.latest().get().await;
let all_rounds = a4.views.ore_round.list().get().await;
let specific = a4.views.ore_round.state().get("round_key").await;

Pre-Built Stacks

The a4-stacks crate includes view definitions for popular protocols:

use a4_stacks::ore::OreStack;

let ore = Arete::<OreStack>::connect().await?;

Building Your Own Stack

To create views for your own Solana programs, you’ll need to build a stack. The CLI then generates typed view accessors for you automatically.

Streaming Data

The SDK provides three streaming methods with different levels of detail:

Method	Returns	Description
`.listen()`	`T`	Merged entity directly (filters out removals/deletes)
`.watch()`	`Update<T>`	Upsert/Patch/Remove/Delete events
`.watch_rich()`	`RichUpdate<T>`	Before/after diffs for tracking changes

Simple Streaming with `.listen()`

The simplest API - emits merged entities directly, filtering out removals and deletes:

let mut stream = a4.views.ore_round.latest().listen();

while let Some(round) = stream.next().await {
    println!("Round: {:?}", round.id.round_id);
}

Watch with Event Types

Stream all update types (upsert, patch, remove, delete):

let mut stream = a4.views.ore_round.latest().watch();

while let Some(update) = stream.next().await {
    match update {
        Update::Upsert { data, .. } => {
            println!("New/Updated round: {:?}", data.id.round_id);
        }
        Update::Patch { data, .. } => {
            println!("Patched round: {:?}", data.id.round_id);
        }
        Update::Remove { key } => {
            println!("Round left this query: {:?}", key);
        }
        Update::Delete { key } => {
            println!("Deleted round: {:?}", key);
        }
    }
}

Rich Streaming with Before/After

Track changes over time with before/after diffs:

let mut stream = a4.views.ore_round.latest().watch_rich();

while let Some(update) = stream.next().await {
    match update {
        RichUpdate::Created { data, .. } => {
            println!("Created: {:?}", data.id.round_id);
        }
        RichUpdate::Updated { before, after, .. } => {
            println!("Updated from {:?} to {:?}",
                before.state.motherlode,
                after.state.motherlode);
        }
        RichUpdate::Removed { key, .. } => {
            println!("Left this query: {:?}", key);
        }
        RichUpdate::Deleted { key, .. } => {
            println!("Deleted: {:?}", key);
        }
    }
}

Watch a Specific Entity

Stream updates for a specific key using state views:

let specific_round = "some-round-key";
let mut stream = a4.views.ore_round.state().watch(specific_round);

while let Some(update) = stream.next().await {
    println!("Round updated: {:?}", update);
}

Chainable Options

All stream builders support server-side options:

// Limit to top 10 items
let mut stream = a4.views.ore_round.list().watch().take(10);

// Skip first 5, take next 10
let mut stream = a4.views.ore_round.list().watch().skip(5).take(10);

// Filter by field
let mut stream = a4.views.ore_round.list().watch().filter("status", "active");

// Select a partition and cap only the initial snapshot
let mut stream = a4.views.ore_round.list()
    .watch()
    .partition("solana-mainnet")
    .with_snapshot_limit(100);

// Resume incrementally after an _seq cursor
let mut stream = a4.views.ore_round.list()
    .watch()
    .after("1234:000000000010");

Each builder defines exact protocol v2 query membership. Different windows and filters on the same view coexist independently. Equivalent normalized queries share one wire subscription, and dropping the final equivalent stream sends unsubscribe.

The SDK keeps a stable opaque subscription ID across reconnects. A complete authoritative snapshot atomically replaces only that query’s membership; an after snapshot is incremental and merges without pruning.

Client-Side Filtering

Use standard stream adapters for client-side filtering:

use futures::StreamExt;

let mut stream = a4.views.ore_round.latest().watch().filter(|update| {
    futures::future::ready(matches!(update, Update::Upsert { .. }))
});

// Only receives upsert events
while let Some(update) = stream.next().await {
    println!("Upsert: {:?}", update);
}

Lazy Streams

Streams are lazy - calling watch() returns immediately without subscribing. The subscription happens automatically on first poll. This enables ergonomic method chaining:

use std::collections::HashSet;

let watchlist: HashSet<String> = /* tokens to watch */;

let mut price_alerts = a4.views.ore_round.list()
    .watch_rich()
    .filter(move |u| watchlist.contains(u.key()))
    .filter_map(|update| match update {
        RichUpdate::Updated { before, after, .. } => {
            let prev = before.trading.last_trade_price.flatten().unwrap_or(0.0);
            let curr = after.trading.last_trade_price.flatten().unwrap_or(0.0);
            if prev > 0.0 {
                let pct = (curr - prev) / prev * 100.0;
                if pct.abs() > 0.1 {
                    return Some((after.info.name.clone(), pct));
                }
            }
            None
        }
        _ => None,
    });

while let Some((name, pct)) = price_alerts.next().await {
    println!("[PRICE] {:?} changed by {:.2}%", name, pct);
}

One-Shot Queries

Fetch current state without streaming:

// Get all items from a view
let rounds: Vec<OreRound> = a4.views.ore_round.latest().get().await;
println!("Found {} rounds", rounds.len());

// Get a specific entity by key
let round: Option<OreRound> = a4.views.ore_round.state().get("round-key").await;
if let Some(r) = round {
    println!("Round: {:?}", r.id.round_id);
}

Synchronous Cache Access

For hot paths where you can’t await, use sync methods to read from cache:

// Synchronous - returns cached data immediately
let cached_rounds = a4.views.ore_round.latest().get_sync();
let cached_round = a4.views.ore_round.state().get_sync("round-key");

Note: Sync methods return empty/None if data hasn’t been loaded yet.

Core Methods Reference

ViewHandle Methods (list/derived views)

Method	Returns	Description
`.get().await`	`Vec<T>`	Get all items
`.get_sync()`	`Vec<T>`	Synchronous cache read
`.listen()`	`Stream<T>`	Stream merged entities (no deletes)
`.watch()`	`Stream<Update<T>>`	Stream all update types
`.watch_rich()`	`Stream<RichUpdate<T>>`	Stream with before/after diffs
`.watch_keys(&[keys])`	`Stream<Update<T>>`	Stream updates for specific keys

StateView Methods (keyed access)

Method	Returns	Description
`.get(key).await`	`Option<T>`	Get entity by key
`.get_sync(key)`	`Option<T>`	Synchronous cache read
`.listen(key)`	`Stream<T>`	Stream merged entity values
`.watch(key)`	`Stream<Update<T>>`	Stream updates for key
`.watch_rich(key)`	`Stream<RichUpdate<T>>`	Stream with diffs for key

Stream Builder Options

Method	Description
`.take(n)`	Server-side limit to N items
`.skip(n)`	Server-side offset
`.filter(key, value)`	Server-side filter
`.partition(value)`	Exact `_partition` match
`.with_snapshot(bool)`	Enable the initial snapshot
`.after(cursor)`	Incremental `_seq` cursor
`.with_snapshot_limit(n)`	Cap initial snapshot rows

Update Types

When streaming with watch(), you receive Update<T> variants:

pub enum Update<T> {
    Upsert { key: String, data: T },  // Full entity update
    Patch { key: String, data: T },   // Partial update (merged)
    Remove { key: String },           // Left this query's membership
    Delete { key: String },           // Deleted from the source view
}

Helper methods: key(), data(), is_remove(), is_delete(), has_data(), into_data(), into_key(), map(f)

Rich Updates (Before/After Diffs)

For tracking changes over time, use watch_rich():

pub enum RichUpdate<T> {
    Created { key: String, data: T },
    Updated { key: String, before: T, after: T, patch: Option<Value> },
    Removed { key: String, last_known: Option<T> },
    Deleted { key: String, last_known: Option<T> },
}

Remove/Removed is query-local: the entity still exists and may remain in another active query. Delete/Deleted is global to the source view.

The Updated variant includes patch - the raw JSON of changed fields, useful for checking what specifically changed:

if update.has_patch_field("trading") {
    // The trading field was modified
}

Understanding `Option<Option<T>>` Fields

Generated entity types often have fields typed as Option<Option<T>>. This represents the patch semantics of Arete updates:

Value	Meaning
`None`	Field was not included in this update (no change)
`Some(None)`	Field was explicitly set to null
`Some(Some(value))`	Field has a concrete value

This distinction matters for partial updates (patches). When the server sends a patch, only changed fields are included. An absent field means “keep the previous value”, while an explicit null means “clear this field”.

Working with `Option<Option<T>>`

// Access a nested optional field
let price = token.trading.last_trade_price.flatten().unwrap_or(0.0);

// Check if field was explicitly set (vs absent from patch)
match &token.reserves.current_price_sol {
    None => println!("Price not in this update"),
    Some(None) => println!("Price explicitly cleared"),
    Some(Some(price)) => println!("Price: {}", price),
}

// Compare values in before/after
if before.trading.last_trade_price != after.trading.last_trade_price {
    println!("Price changed!");
}

Generating a Rust SDK

Use the Arete CLI to generate a typed Rust SDK from your spec. See CLI Commands for the full reference and Configuration for arete.toml options.

# Generate SDK crate
a4 sdk create rust settlement-game

# With custom output directory
a4 sdk create rust settlement-game --output ./crates/game-sdk

# With custom crate name
a4 sdk create rust settlement-game --crate-name game-sdk

# Generate as a module instead of a standalone crate
a4 sdk create rust settlement-game --module --output ./src/stacks/game

Crate vs Module Output

By default, the CLI generates a standalone crate with its own Cargo.toml:

generated/settlement-game-stack/
├── Cargo.toml
└── src/
    ├── lib.rs      # Re-exports
    ├── types.rs    # Data structs (with Option<Option<T>> for patchable fields)
    └── entity.rs   # Stack and Views implementations

With the --module flag, the CLI generates a module that can be embedded in an existing crate:

src/stacks/game/
├── mod.rs      # Re-exports
├── types.rs    # Data structs
└── entity.rs   # Stack and Views implementations

Using the Generated Code

Add the generated crate to your Cargo.toml:

[dependencies]
a4-sdk = "0.3.0"
settlement-game-stack = { path = "./generated/settlement-game-stack" }

Then use it:

use a4_sdk::prelude::*;
use settlement_game_stack::{SettlementStack, GameState};

let a4 = Arete::<SettlementStack>::connect().await?;
let scores = a4.views.player_score.leaderboard().get().await;
let game = a4.views.game_state.state().get("game-key").await;

Or if using module mode, add to your lib.rs:

pub mod game;  // Points to src/game/mod.rs

Error Handling

use a4_sdk::AreteError;

match Arete::<OreStack>::connect().await {
    Ok(a4) => println!("Connected!"),
    Err(AreteError::Connection(e)) => {
        eprintln!("Connection failed: {}", e);
    }
    Err(AreteError::Authentication(e)) => {
        eprintln!("Auth failed: {}", e);
    }
    Err(e) => {
        eprintln!("Unexpected error: {:?}", e);
    }
}

Auto-Reconnection

The SDK automatically reconnects on connection loss with configurable backoff:

let a4 = Arete::<OreStack>::builder()
    .auto_reconnect(true)
    .reconnect_intervals(vec![
        Duration::from_secs(1),
        Duration::from_secs(2),
        Duration::from_secs(5),
        Duration::from_secs(10),
    ])
    .max_reconnect_attempts(20)
    .connect()
    .await?;

Complete Example

A full command-line app that streams ORE mining rounds:

use a4_sdk::prelude::*;
use a4_stacks::ore::{OreStack, OreRound};
use std::time::Duration;
use tokio::time::timeout;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    println!("-----------------------------------");
    println!("  Arete ORE Round Monitor     ");
    println!("-----------------------------------\n");

    // Connect with 10 second timeout
    let a4 = match timeout(
        Duration::from_secs(10),
        Arete::<OreStack>::connect()
    ).await {
        Ok(Ok(a4)) => {
            println!("Connected to ORE stack\n");
            a4
        }
        Ok(Err(e)) => {
            eprintln!("Connection error: {}", e);
            return Err(e.into());
        }
        Err(_) => {
            eprintln!("Connection timeout");
            return Err(anyhow::anyhow!("Connection timeout"));
        }
    };

    // Stream with stats
    let mut round_count = 0;
    let mut update_count = 0;
    let mut stream = a4.views.ore_round.latest().watch();

    println!("Streaming live ORE rounds (press Ctrl+C to exit)...\n");

    while let Some(update) = stream.next().await {
        update_count += 1;

        match update {
            Update::Upsert { data, .. } => {
                round_count += 1;
                println!(
                    "[#{:03}] Round #{} - Motherlode: {} SOL",
                    update_count,
                    data.id.round_id,
                    data.state.motherlode as f64 / 1_000_000_000.0
                );
            }
            Update::Patch { data, .. } => {
                println!(
                    "[#{:03}] Updated Round #{} - Difficulty: {}",
                    update_count,
                    data.id.round_id,
                    data.state.total_difficulty
                );
            }
            Update::Remove { key } => {
                println!("[#{:03}] Round left this query: {:?}", update_count, key);
            }
            Update::Delete { key } => {
                println!("[#{:03}] Deleted round: {:?}", update_count, key);
            }
        }

        // Print stats every 10 updates
        if update_count % 10 == 0 {
            println!("\n--- Stats: {} rounds tracked, {} updates received ---\n",
                round_count, update_count);
        }
    }

    Ok(())
}

Run it:

cargo run

Advanced Patterns

Multiple Concurrent Streams

use futures::future::join;

// Watch multiple views concurrently
let latest_stream = a4.views.ore_round.latest().watch();
let list_stream = a4.views.ore_round.list().watch();

let (latest_result, list_result) = join(
    process_stream(latest_stream),
    process_stream(list_stream)
).await;

Reconnection Handling

loop {
    match Arete::<OreStack>::connect().await {
        Ok(a4) => {
            println!("Connected!");

            let mut stream = a4.views.ore_round.latest().watch();

            while let Some(update) = stream.next().await {
                process_update(update);
            }

            // Stream ended - connection lost
            println!("Connection lost, reconnecting...");
        }
        Err(e) => {
            eprintln!("Connection failed: {}, retrying in 5s...", e);
            tokio::time::sleep(Duration::from_secs(5)).await;
        }
    }
}

Graceful Shutdown

use tokio::signal;

let a4 = Arete::<OreStack>::connect().await?;
let mut stream = a4.views.ore_round.latest().watch();

loop {
    tokio::select! {
        Some(update) = stream.next() => {
            process_update(update);
        }
        _ = signal::ctrl_c() => {
            println!("\nShutting down gracefully...");
            a4.disconnect().await;
            break;
        }
    }
}

Examples

Scaffold a complete Rust example that streams ORE mining rounds:

a4 create my-ore-app --template rust-ore
cd my-ore-app
cargo run

Or run a4 create interactively and select the rust-ore template.

Next Steps

TypeScript SDK - Use Arete in Node.js or browsers
React SDK - Build web apps with React hooks
Build Your Own Stack - Create custom data streams for any Solana program
CLI Reference - Deployment and management commands