Rust Sop

Rust SOP — joelclaw Standard Operating Procedures

Idiomatic Rust patterns and conventions for all Rust code in joelclaw. Primary consumer: clawnode (embedded PDS + mesh daemon). Written for agent workers — include this skill when delegating Rust work to codex.

Project Conventions

Stack

Project Structure

clawnode/
├── Cargo.toml
├── src/
│   ├── main.rs              # CLI entry, clap dispatch
│   ├── lib.rs               # Public API surface
│   ├── daemon/              # Long-running daemon logic
│   │   ├── mod.rs
│   │   ├── server.rs        # Axum HTTP server (XRPC)
│   │   ├── firehose.rs      # WebSocket subscribeRepos
│   │   └── heartbeat.rs     # Presence + health
│   ├── pds/                 # Embedded PDS
│   │   ├── mod.rs
│   │   ├── repo.rs          # MST / record storage
│   │   ├── records.rs       # CRUD operations
│   │   └── auth.rs          # Session management
│   ├── storage/             # libSQL abstraction
│   │   ├── mod.rs
│   │   ├── migrations.rs    # Schema migrations
│   │   └── vectors.rs       # Vector search helpers
│   ├── mesh/                # Service discovery + proxy
│   │   ├── mod.rs
│   │   ├── registry.rs      # ServiceRegistry trait
│   │   └── proxy.rs         # Redis/Typesense/Inngest proxy
│   ├── socket/              # Unix socket JSON-RPC
│   │   └── mod.rs
│   └── cli/                 # CLI subcommands
│       ├── mod.rs
│       ├── status.rs
│       ├── recall.rs
│       └── send.rs
├── tests/
│   ├── integration/
│   └── common/
└── migrations/
    └── 001_initial.sql

Naming Conventions

• Crate name: clawnode (binary), internal lib crate also clawnode
• Modules: snake_case, flat where possible (storage.rs not storage/mod.rs unless submodules needed)
• Types: PascalCase. Prefix domain types: PdsRecord, MeshNode, ServiceEntry
• Traits: Adjective or noun (Discoverable, ServiceRegistry, RecordStore)
• Error enums: <Module>Error (StorageError, PdsError, MeshError)
• Constants: SCREAMING_SNAKE_CASE

Core Patterns

Error Handling

// Library code: thiserror for typed errors
#[derive(thiserror::Error, Debug)]
pub enum PdsError {
    #[error("record not found: {collection}/{rkey}")]
    NotFound { collection: String, rkey: String },
    #[error("storage error: {0}")]
    Storage(#[from] StorageError),
    #[error("invalid record: {0}")]
    InvalidRecord(String),
}

// Application/binary code: anyhow for ergonomic error chains
use anyhow::{Context, Result};

fn main() -> Result<()> {
    let config = load_config()
        .context("failed to load clawnode config")?;
    Ok(())
}

Rules:

• Never unwrap() in production code. Use expect("reason") only for truly invariant conditions.
• Use ? operator everywhere. Add .context("what failed") for anyhow chains.
• Map errors at boundary layers (e.g., PdsError → axum::response::IntoResponse).

Ownership & Borrowing

// Prefer borrowing for function params
fn process_record(record: &PdsRecord) -> Result<()> { ... }

// Use &str not String for read-only string params
fn find_by_collection(collection: &str) -> Result<Vec<PdsRecord>> { ... }

// Use &[T] not Vec<T> for read-only slice params
fn batch_insert(records: &[PdsRecord]) -> Result<usize> { ... }

// Return owned types from constructors/builders
fn create_record(collection: String, rkey: String, value: serde_json::Value) -> PdsRecord { ... }

// Cow for conditional cloning
use std::borrow::Cow;
fn normalize_handle(handle: &str) -> Cow<str> {
    if handle.starts_with("did:") {
        Cow::Borrowed(handle)
    } else {
        Cow::Owned(format!("at://{handle}"))
    }
}

Async / Tokio

// Default: multi-thread runtime
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // ...
}

// Graceful shutdown pattern (critical for daemon)
async fn run_daemon(config: Config) -> anyhow::Result<()> {
    let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);

    let server = tokio::spawn(run_server(config.clone(), shutdown_rx.clone()));
    let heartbeat = tokio::spawn(run_heartbeat(config.clone(), shutdown_rx.clone()));
    let firehose = tokio::spawn(run_firehose(config.clone(), shutdown_rx));

    // Wait for ctrl-c
    tokio::signal::ctrl_c().await?;
    tracing::info!("shutdown signal received");
    shutdown_tx.send(true)?;

    // Wait for all tasks
    let _ = tokio::join!(server, heartbeat, firehose);
    Ok(())
}

// Use select! for cancellable operations
async fn run_heartbeat(config: Config, mut shutdown: tokio::sync::watch::Receiver<bool>) {
    let mut interval = tokio::time::interval(Duration::from_secs(30));
    loop {
        tokio::select! {
            _ = interval.tick() => {
                if let Err(e) = send_heartbeat(&config).await {
                    tracing::warn!("heartbeat failed: {e}");
                }
            }
            _ = shutdown.changed() => {
                tracing::info!("heartbeat shutting down");
                break;
            }
        }
    }
}

// spawn_blocking for CPU-bound work (embeddings, hashing)
let embedding = tokio::task::spawn_blocking(move || {
    compute_embedding(&text)
}).await?;

// Never hold a Mutex guard across .await
// BAD:
let mut guard = data.lock().await;
some_async_op().await; // ← deadlock risk
// GOOD:
let value = {
    let guard = data.lock().await;
    guard.clone()
};
some_async_op_with(value).await;

Traits & Hexagonal Architecture

// Define ports as traits
#[async_trait::async_trait]
pub trait ServiceRegistry: Send + Sync {
    async fn discover(&self, service: &str) -> Result<Vec<ServiceEntry>>;
    async fn register(&self, entry: ServiceEntry) -> Result<()>;
    async fn deregister(&self, node_id: &str) -> Result<()>;
}

#[async_trait::async_trait]
pub trait RecordStore: Send + Sync {
    async fn get(&self, collection: &str, rkey: &str) -> Result<Option<PdsRecord>>;
    async fn list(&self, collection: &str, limit: usize) -> Result<Vec<PdsRecord>>;
    async fn put(&self, record: PdsRecord) -> Result<()>;
    async fn delete(&self, collection: &str, rkey: &str) -> Result<bool>;
}

// Implement adapters
pub struct LibSqlRecordStore { db: libsql::Database }
pub struct StaticServiceRegistry { services: HashMap<String, Vec<ServiceEntry>> }
pub struct PdsServiceRegistry { client: AtpAgent }

// Wire in main.rs (composition root)
let store = LibSqlRecordStore::new("clawnode.db").await?;
let registry = StaticServiceRegistry::from_config(&config);
let daemon = Daemon::new(store, registry);

Structured Logging

use tracing::{info, warn, error, debug, instrument};

// Instrument async functions
#[instrument(skip(db), fields(collection = %collection))]
async fn list_records(db: &impl RecordStore, collection: &str) -> Result<Vec<PdsRecord>> {
    let records = db.list(collection, 100).await?;
    info!(count = records.len(), "listed records");
    Ok(records)
}

// Subscriber setup
fn init_tracing() {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "clawnode=info,tower_http=info".into())
        )
        .with_target(false)
        .json() // structured output for daemon mode
        .init();
}

libSQL + Vector Search

use libsql::Builder;

async fn init_db(path: &str) -> anyhow::Result<libsql::Database> {
    let db = Builder::new_local(path).build().await?;
    let conn = db.connect()?;

    // Run migrations
    conn.execute_batch("
        CREATE TABLE IF NOT EXISTS observations (
            uri TEXT PRIMARY KEY,
            content TEXT NOT NULL,
            category TEXT,
            tags TEXT,
            embedding F32_BLOB(384),
            created_at TEXT NOT NULL
        );
        CREATE INDEX IF NOT EXISTS obs_vec_idx
            ON observations (libsql_vector_idx(embedding, 'metric=cosine'));
    ").await?;

    Ok(db)
}

// Vector recall
async fn recall(conn: &libsql::Connection, embedding: &[f32], k: usize) -> Result<Vec<Observation>> {
    let embedding_str = format!("[{}]", embedding.iter()
        .map(|f| f.to_string())
        .collect::<Vec<_>>()
        .join(","));

    let mut rows = conn.query(
        "SELECT content, category, tags FROM vector_top_k('obs_vec_idx', vector32(?1), ?2)
         JOIN observations ON observations.rowid = id",
        libsql::params![embedding_str, k as i64],
    ).await?;

    let mut results = Vec::new();
    while let Some(row) = rows.next().await? {
        results.push(Observation {
            content: row.get(0)?,
            category: row.get(1)?,
            tags: row.get(2)?,
        });
    }
    Ok(results)
}

Axum HTTP Server

use axum::{Router, Json, extract::State};

fn xrpc_routes(state: AppState) -> Router {
    Router::new()
        .route("/xrpc/com.atproto.repo.createRecord", post(create_record))
        .route("/xrpc/com.atproto.repo.getRecord", get(get_record))
        .route("/xrpc/com.atproto.repo.listRecords", get(list_records))
        .route("/xrpc/com.atproto.repo.deleteRecord", post(delete_record))
        .route("/xrpc/com.atproto.repo.describeRepo", get(describe_repo))
        .route("/_health", get(health))
        .with_state(state)
}

async fn health(State(state): State<AppState>) -> Json<serde_json::Value> {
    Json(serde_json::json!({
        "status": "ok",
        "did": state.did,
        "uptime_secs": state.start_time.elapsed().as_secs(),
    }))
}

Validation Checklist

Every Rust change must pass:

cargo fmt --check                          # formatting
cargo clippy --all-targets --all-features  # lints (zero warnings)
cargo test                                 # all tests
cargo test --doc                           # doctests

For clawnode specifically:

cargo build --release                      # ensure release builds clean
cargo clippy -- -D warnings                # treat warnings as errors in CI

Anti-Patterns

Library-First Development (MANDATORY)

Before writing non-trivial Rust code, search the pdf-brain library. It contains chunked, indexed content from the best Rust books. This is not optional — the library has authoritative patterns that are better than what you'll generate from training data.

# Search the library
joelclaw docs search "tokio graceful shutdown signal"

# Get full context around a result
joelclaw docs context <chunk-id> --mode snippet-window

Load `references/library.md` for the full search playbook — it has few-shot examples for every Rust domain (ownership, async, traits, axum, testing, systems).

Quick examples:

# Before implementing error types:
joelclaw docs search "thiserror custom error types"

# Before writing async code:
joelclaw docs search "tokio spawn select join"

# Before designing a trait:
joelclaw docs search "trait objects dynamic dispatch dyn"

# When stuck on a compiler error:
joelclaw docs search "borrow checker mutable immutable reference"

References

Load these for deeper guidance on specific topics:

Async Rust — Tokio Patterns for Clawnode

Runtime Setup

// Default: multi-thread (daemon, server)
#[tokio::main]
async fn main() -> anyhow::Result<()> { ... }

// Single-thread for tests
#[tokio::test]
async fn test_something() { ... }

// Custom runtime (when you need control)
let runtime = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(4)
    .thread_name("clawnode-worker")
    .enable_all()
    .build()?;

Graceful Shutdown (Daemon Pattern)

This is the core pattern for clawnode — multiple long-running tasks that need coordinated shutdown:

use tokio::sync::watch;
use tokio::signal;

async fn run_daemon(config: Config) -> anyhow::Result<()> {
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    // Spawn all subsystems
    let server = tokio::spawn(run_xrpc_server(config.clone(), shutdown_rx.clone()));
    let heartbeat = tokio::spawn(run_heartbeat(config.clone(), shutdown_rx.clone()));
    let firehose = tokio::spawn(run_firehose_subscriber(config.clone(), shutdown_rx.clone()));
    let socket = tokio::spawn(run_unix_socket(config.clone(), shutdown_rx));

    // Wait for shutdown signal
    shutdown_signal().await;
    tracing::info!("shutdown signal received, draining...");
    let _ = shutdown_tx.send(true);

    // Wait for all tasks with timeout
    let _ = tokio::time::timeout(
        Duration::from_secs(10),
        futures::future::join_all([server, heartbeat, firehose, socket]),
    ).await;

    tracing::info!("shutdown complete");
    Ok(())
}

async fn shutdown_signal() {
    let ctrl_c = signal::ctrl_c();
    #[cfg(unix)]
    let mut sigterm = signal::unix::signal(signal::unix::SignalKind::terminate())
        .expect("failed to install SIGTERM handler");

    tokio::select! {
        _ = ctrl_c => {},
        #[cfg(unix)]
        _ = sigterm.recv() => {},
    }
}

Cancellable Loop Pattern

Every long-running subsystem uses this shape:

async fn run_heartbeat(config: Config, mut shutdown: watch::Receiver<bool>) {
    let mut interval = tokio::time::interval(Duration::from_secs(30));
    loop {
        tokio::select! {
            _ = interval.tick() => {
                if let Err(e) = send_presence(&config).await {
                    tracing::warn!(error = %e, "heartbeat failed");
                    // Don't crash — retry on next tick
                }
            }
            _ = shutdown.changed() => {
                tracing::info!("heartbeat shutting down");
                break;
            }
        }
    }
}

Concurrent Execution

// join! — run concurrently, wait for all
let (users, posts) = tokio::join!(
    fetch_users(&db),
    fetch_posts(&db),
);

// try_join! — stop on first error
let (users, posts) = tokio::try_join!(
    fetch_users(&db),
    fetch_posts(&db),
)?;

// select! — first to complete wins (others cancelled)
tokio::select! {
    result = fetch_from_primary() => handle(result),
    result = fetch_from_replica() => handle(result),
    _ = tokio::time::sleep(Duration::from_secs(5)) => {
        return Err(anyhow::anyhow!("timeout"));
    }
}

// Spawn for fire-and-forget with JoinHandle
let handle = tokio::spawn(async move {
    expensive_work().await
});
let result = handle.await?; // JoinError if task panics

Channels

// mpsc: event processing pipeline
let (tx, mut rx) = tokio::sync::mpsc::channel::<Event>(256);

// Producer
tokio::spawn(async move {
    tx.send(Event::RecordCreated { ... }).await?;
});

// Consumer
while let Some(event) = rx.recv().await {
    process_event(event).await;
}

// watch: health state (readers always get latest)
let (health_tx, health_rx) = watch::channel(HealthState::Starting);
// Writer: health_tx.send(HealthState::Ready)?;
// Reader: let current = *health_rx.borrow();

Shared State

// Arc<tokio::sync::Mutex<T>> for async-safe shared mutable state
let shared = Arc::new(tokio::sync::Mutex::new(HashMap::new()));

// CRITICAL: never hold lock across .await
// BAD:
let mut guard = shared.lock().await;
some_async_op().await; // ← deadlock risk
// GOOD:
let value = {
    let guard = shared.lock().await;
    guard.get("key").cloned()
};
some_async_op_with(value).await;

// RwLock for read-heavy workloads (service registry)
let registry = Arc::new(tokio::sync::RwLock::new(ServiceRegistry::new()));
// Many readers: registry.read().await
// Rare writer: registry.write().await

Blocking Work

// CPU-bound: spawn_blocking (runs on dedicated thread pool)
let hash = tokio::task::spawn_blocking(move || {
    compute_hash(&data) // CPU-intensive, would block async runtime
}).await?;

// File I/O: tokio::fs (async wrappers around blocking FS ops)
let content = tokio::fs::read_to_string("config.toml").await?;

// External process
let output = tokio::process::Command::new("git")
    .args(["rev-parse", "HEAD"])
    .output()
    .await?;

Timeouts and Retries

// Timeout any future
let result = tokio::time::timeout(
    Duration::from_secs(5),
    fetch_remote_data(),
).await.map_err(|_| anyhow::anyhow!("operation timed out"))?;

// Retry with exponential backoff
async fn with_retry<F, Fut, T, E>(f: F, max_retries: u32) -> Result<T, E>
where
    F: Fn() -> Fut,
    Fut: Future<Result<T, E>>,
    E: std::fmt::Display,
{
    let mut delay = Duration::from_millis(100);
    for attempt in 0..max_retries {
        match f().await {
            Ok(v) => return Ok(v),
            Err(e) if attempt < max_retries - 1 => {
                tracing::warn!(attempt, error = %e, "retrying after {:?}", delay);
                tokio::time::sleep(delay).await;
                delay *= 2;
            }
            Err(e) => return Err(e),
        }
    }
    unreachable!()
}

WebSocket (Firehose Pattern)

use tokio_tungstenite::{connect_async, tungstenite::Message};
use futures_util::{StreamExt, SinkExt};

async fn subscribe_firehose(url: &str, mut shutdown: watch::Receiver<bool>) -> anyhow::Result<()> {
    loop {
        let (mut ws, _) = connect_async(url).await?;
        tracing::info!("connected to firehose");

        loop {
            tokio::select! {
                msg = ws.next() => {
                    match msg {
                        Some(Ok(Message::Text(text))) => {
                            process_firehose_event(&text).await;
                        }
                        Some(Ok(Message::Close(_))) | None => {
                            tracing::warn!("firehose connection closed, reconnecting...");
                            break; // break inner loop, reconnect in outer
                        }
                        Some(Err(e)) => {
                            tracing::error!(error = %e, "firehose error");
                            break;
                        }
                        _ => {}
                    }
                }
                _ = shutdown.changed() => {
                    tracing::info!("firehose shutting down");
                    let _ = ws.close(None).await;
                    return Ok(());
                }
            }
        }

        tokio::time::sleep(Duration::from_secs(5)).await; // backoff before reconnect
    }
}

Async Traits

// Rust 1.75+: native async fn in traits (use this when possible)
trait RecordStore: Send + Sync {
    async fn get(&self, collection: &str, rkey: &str) -> Result<Option<Record>>;
    async fn put(&self, record: Record) -> Result<()>;
}

// When you need dyn dispatch (trait objects): use async-trait
use async_trait::async_trait;

#[async_trait]
trait ServiceRegistry: Send + Sync {
    async fn discover(&self, service: &str) -> Result<Vec<ServiceEntry>>;
}

// async-trait works with Box<dyn ServiceRegistry>
// Native async fn in traits does NOT (yet) support dyn dispatch

Testing Async Code

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_record_store() {
        let db = setup_test_db().await;
        let store = LibSqlRecordStore::new(db);

        store.put(Record { ... }).await.unwrap();
        let found = store.get("collection", "rkey").await.unwrap();
        assert!(found.is_some());
    }

    #[tokio::test]
    async fn test_timeout() {
        let result = tokio::time::timeout(
            Duration::from_millis(100),
            tokio::time::sleep(Duration::from_secs(10)),
        ).await;
        assert!(result.is_err()); // timed out
    }
}

Rules

• Never block the async runtime: Use spawn_blocking for CPU work, tokio::fs for file I/O
• Never hold Mutex across .await: Clone data out, drop lock, then await
• Always timeout external I/O: Network, process, file — all can hang forever
• Use watch for shutdown: Every subsystem gets a watch::Receiver<bool>
• Handle JoinError: Spawned tasks can panic — .await? on JoinHandle
• Prefer channels over shared state: mpsc/oneshot are usually cleaner than Arc<Mutex<T>>
• Log at warn level on retry, error on final failure: Don't spam errors for transient issues

Axum — HTTP Server Patterns for Clawnode

Clawnode uses Axum for XRPC endpoints (AT Proto), health checks, and the WebSocket firehose. These patterns are production-grade.

Application Structure

use axum::{Router, routing::{get, post}, extract::State};
use std::sync::Arc;
use tower_http::{
    compression::CompressionLayer,
    trace::TraceLayer,
    timeout::TimeoutLayer,
};

#[derive(Clone)]
struct AppState {
    db: libsql::Database,
    did: String,
    registry: Arc<dyn ServiceRegistry>,
    start_time: std::time::Instant,
}

fn build_router(state: AppState) -> Router {
    Router::new()
        // XRPC routes
        .route("/xrpc/com.atproto.repo.createRecord", post(create_record))
        .route("/xrpc/com.atproto.repo.getRecord", get(get_record))
        .route("/xrpc/com.atproto.repo.listRecords", get(list_records))
        .route("/xrpc/com.atproto.repo.deleteRecord", post(delete_record))
        .route("/xrpc/com.atproto.repo.describeRepo", get(describe_repo))
        // Firehose WebSocket
        .route("/xrpc/com.atproto.sync.subscribeRepos", get(subscribe_repos_ws))
        // Health
        .route("/_health", get(health))
        // Middleware (bottom-to-top execution order)
        .layer(CompressionLayer::new())
        .layer(TraceLayer::new_for_http())
        .layer(TimeoutLayer::new(std::time::Duration::from_secs(30)))
        .with_state(state)
}

async fn run_server(state: AppState, mut shutdown: tokio::sync::watch::Receiver<bool>) -> anyhow::Result<()> {
    let app = build_router(state);
    let listener = tokio::net::TcpListener::bind("0.0.0.0:2583").await?;
    tracing::info!("XRPC server listening on :2583");

    axum::serve(listener, app)
        .with_graceful_shutdown(async move { let _ = shutdown.changed().await; })
        .await?;
    Ok(())
}

Extractors

Extractors pull typed data from requests. Order matters — body extractors must be last.

use axum::extract::{State, Path, Query, Json};
use serde::Deserialize;

// Path params
async fn get_record(
    State(state): State<AppState>,
    Path((collection, rkey)): Path<(String, String)>,
) -> Result<Json<Record>, AppError> { ... }

// Query params
#[derive(Deserialize)]
struct ListParams {
    collection: String,
    limit: Option<usize>,
    cursor: Option<String>,
}

async fn list_records(
    State(state): State<AppState>,
    Query(params): Query<ListParams>,
) -> Result<Json<ListResponse>, AppError> { ... }

// JSON body (must be last extractor)
async fn create_record(
    State(state): State<AppState>,
    Json(body): Json<CreateRecordRequest>,
) -> Result<Json<CreateRecordResponse>, AppError> { ... }

Custom Extractors

For XRPC auth, create a custom extractor:

use axum::extract::FromRequestParts;
use axum::http::request::Parts;

struct AuthenticatedDid(String);

#[async_trait::async_trait]
impl<S: Send + Sync> FromRequestParts<S> for AuthenticatedDid {
    type Rejection = AppError;

    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
        let auth = parts.headers
            .get("authorization")
            .and_then(|v| v.to_str().ok())
            .ok_or(AppError::Unauthorized)?;

        let token = auth.strip_prefix("Bearer ")
            .ok_or(AppError::Unauthorized)?;

        let did = validate_session_token(token)
            .map_err(|_| AppError::Unauthorized)?;

        Ok(AuthenticatedDid(did))
    }
}

// Use in handler — extracted before body
async fn create_record(
    State(state): State<AppState>,
    auth: AuthenticatedDid,
    Json(body): Json<CreateRecordRequest>,
) -> Result<Json<CreateRecordResponse>, AppError> { ... }

Error Handling

Map domain errors to HTTP responses via IntoResponse:

use axum::response::{IntoResponse, Response};
use axum::http::StatusCode;

#[derive(thiserror::Error, Debug)]
enum AppError {
    #[error("not found: {0}")]
    NotFound(String),
    #[error("unauthorized")]
    Unauthorized,
    #[error("invalid request: {0}")]
    BadRequest(String),
    #[error("internal error: {0}")]
    Internal(#[from] anyhow::Error),
}

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        let (status, message) = match &self {
            AppError::NotFound(msg) => (StatusCode::NOT_FOUND, msg.clone()),
            AppError::Unauthorized => (StatusCode::UNAUTHORIZED, "Unauthorized".into()),
            AppError::BadRequest(msg) => (StatusCode::BAD_REQUEST, msg.clone()),
            AppError::Internal(e) => {
                tracing::error!(error = %e, "internal server error");
                (StatusCode::INTERNAL_SERVER_ERROR, "Internal error".into())
            }
        };

        // XRPC error format
        let body = serde_json::json!({
            "error": status.canonical_reason().unwrap_or("Error"),
            "message": message,
        });

        (status, axum::Json(body)).into_response()
    }
}

WebSocket (Firehose)

use axum::extract::ws::{WebSocket, WebSocketUpgrade, Message};

async fn subscribe_repos_ws(
    State(state): State<AppState>,
    ws: WebSocketUpgrade,
) -> impl IntoResponse {
    ws.on_upgrade(move |socket| handle_firehose(socket, state))
}

async fn handle_firehose(mut socket: WebSocket, state: AppState) {
    let mut rx = state.firehose_tx.subscribe(); // broadcast channel

    loop {
        tokio::select! {
            event = rx.recv() => {
                match event {
                    Ok(data) => {
                        let msg = Message::Text(serde_json::to_string(&data).unwrap());
                        if socket.send(msg).await.is_err() {
                            break; // client disconnected
                        }
                    }
                    Err(_) => break,
                }
            }
            msg = socket.recv() => {
                match msg {
                    Some(Ok(Message::Close(_))) | None => break,
                    _ => {} // ignore client messages
                }
            }
        }
    }
}

Middleware

// Function middleware (simplest)
use axum::{http::Request, middleware::Next, response::Response};

async fn log_request(req: Request, next: Next) -> Response {
    let method = req.method().clone();
    let uri = req.uri().clone();
    let start = std::time::Instant::now();

    let response = next.run(req).await;

    tracing::info!(
        method = %method,
        uri = %uri,
        status = %response.status(),
        duration_ms = %start.elapsed().as_millis(),
    );
    response
}

// Apply: .layer(axum::middleware::from_fn(log_request))

// With state:
// .layer(axum::middleware::from_fn_with_state(state.clone(), auth_middleware))

// Middleware ordering: .layer() calls are bottom-to-top
// ServiceBuilder wraps are top-to-bottom
use tower::ServiceBuilder;
let layers = ServiceBuilder::new()
    .layer(TraceLayer::new_for_http())    // runs first
    .layer(CompressionLayer::new())        // runs second
    .layer(TimeoutLayer::new(Duration::from_secs(30))); // runs third

Testing

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use axum::http::{Request, StatusCode, Method};
    use tower::ServiceExt; // for oneshot()

    #[tokio::test]
    async fn test_health() {
        let state = test_state().await;
        let app = build_router(state);

        let response = app
            .oneshot(Request::builder().uri("/_health").body(Body::empty()).unwrap())
            .await
            .unwrap();

        assert_eq!(response.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn test_create_record() {
        let state = test_state().await;
        let app = build_router(state);

        let body = serde_json::json!({
            "repo": "did:plc:test",
            "collection": "dev.joelclaw.node.presence",
            "record": { "status": "online" }
        });

        let response = app
            .oneshot(
                Request::builder()
                    .method(Method::POST)
                    .uri("/xrpc/com.atproto.repo.createRecord")
                    .header("content-type", "application/json")
                    .header("authorization", "Bearer test-token")
                    .body(Body::from(serde_json::to_string(&body).unwrap()))
                    .unwrap()
            )
            .await
            .unwrap();

        assert_eq!(response.status(), StatusCode::OK);
    }
}

Key Crates

Rules

• State must be Clone: Use Arc<T> for expensive-to-clone fields
• Body extractors last: Json, Form, Bytes must be the final extractor
• Layer order matters: .layer() calls execute bottom-to-top
• Always return proper errors: Implement IntoResponse for your error type
• Test with `oneshot()`: No need to start a real server
• Graceful shutdown: Pass with_graceful_shutdown to axum::serve

Common Rust Compiler Errors — Quick Fixes

Borrow Checker

E0502: Cannot borrow x as mutable because also borrowed as immutable

• Fix 1 (Re-scope): Use {} block to drop immutable borrow before mutation
• Fix 2 (Clone): let val = x[0].clone(); x.push(val);

E0382: Use of moved value

• Fix 1 (Reference): Pass &x if ownership transfer isn't needed
• Fix 2 (Clone): x.clone() if consumer needs ownership
• Fix 3 (Copy): Derive Copy if type is trivial

E0597: x does not live long enough

• Fix 1 (Ownership): Return String (owned) not &str (borrowed)
• Fix 2 (Lifetime lifting): Declare storage outside the scope where ref is taken

E0499: Cannot borrow x as mutable more than once

• Fix 1: Split struct fields into separate variables
• Fix 2: Use indices instead of references for collections
• Fix 3: Interior mutability (RefCell, Mutex) as last resort

Async-Specific

"future cannot be sent between threads safely"

• Cause: Holding a non-Send type (e.g., Rc, RefCell) across .await
• Fix: Use Arc + tokio::sync::Mutex instead of Rc + RefCell
• Fix 2: Scope the non-Send value so it's dropped before .await

"this MutexGuard is held across an await point"

• Cause: Lock guard lives across .await — potential deadlock
• Fix: Clone data out, drop guard, then await:

  let value = { data.lock().await.clone() };
  use_value(value).await;

"impl Trait not allowed in trait method return"

• Fix: Use async-trait crate (until RPITIT stabilizes fully)
• Note: Rust 1.75+ supports async fn in traits natively for many cases

Type System

E0277: Trait bound not satisfied

• Check: Did you derive the trait? (#[derive(Clone, Debug, Serialize)])
• Check: Is the trait in scope? (use std::fmt::Display;)
• Check: All generic params satisfy the bound?

E0308: Mismatched types

• Common: String vs &str — use .as_str() or &*s
• Common: Option<T> vs T — use ? or .unwrap_or_default()
• Common: Result<T, E1> vs Result<T, E2> — add #[from] to thiserror enum

Lifetime

"missing lifetime specifier"

• Add explicit lifetime: fn foo<'a>(x: &'a str) -> &'a str
• If returning owned data, don't return reference: fn foo(x: &str) -> String

"lifetime may not live long enough"

• Check: Are you storing a reference in a struct? Add lifetime param: struct Foo<'a> { bar: &'a str }
• Check: Returning a reference to a local? Return owned instead.

Debug Workflow

1. Read the full error message — Rust's errors are excellent 2. Look at the help line — it often has the exact fix 3. If borrow checker: draw the ownership/borrow timeline on paper 4. If lifetime: ask "who owns this data and how long does it live?" 5. If async: check for non-Send types and lock guards across awaits 6. Use dbg!() for quick value inspection (remove before commit) 7. cargo clippy --fix auto-fixes many common issues

Rust Library — pdf-brain Reference

The joelclaw pdf-brain contains Rust books that have been chunked, classified, and vector-indexed. Always search the library before writing non-trivial Rust code — it contains authoritative patterns from the best Rust books.

How to Search

# Text search — fast, keyword-based
joelclaw docs search "query terms"

# Get full context around a chunk
joelclaw docs context <chunk-id> --mode snippet-window

Few-Shot Search Examples

Ownership & Borrowing

# When dealing with lifetimes, borrow checker errors, ownership transfer
joelclaw docs search "ownership borrowing move semantics"
joelclaw docs search "lifetime annotations struct"
joelclaw docs search "borrow checker mutable immutable reference"
joelclaw docs search "smart pointers Box Rc Arc"
joelclaw docs search "interior mutability RefCell Cell"
joelclaw docs search "clone on write Cow"

Error Handling

# When designing error types or handling Result/Option
joelclaw docs search "thiserror custom error types"
joelclaw docs search "anyhow context error propagation"
joelclaw docs search "Result Option combinators map and_then"
joelclaw docs search "error handling question mark operator"

Async & Concurrency

# When writing tokio async code, channels, shared state
joelclaw docs search "tokio async await runtime"
joelclaw docs search "tokio spawn select join"
joelclaw docs search "channels mpsc oneshot broadcast"
joelclaw docs search "async mutex rwlock shared state"
joelclaw docs search "graceful shutdown signal handling"
joelclaw docs search "futures streams async iterator"
joelclaw docs search "spawn_blocking CPU bound work"

Traits & Generics

# When designing trait hierarchies, generic abstractions
joelclaw docs search "trait objects dynamic dispatch dyn"
joelclaw docs search "generic bounds where clause"
joelclaw docs search "associated types trait design"
joelclaw docs search "impl Trait return position"

Web / Server (Axum)

# When building HTTP services, middleware, routing
joelclaw docs search "axum router handler extractor"
joelclaw docs search "web server middleware tower"
joelclaw docs search "REST API JSON serialization"
joelclaw docs search "websocket server tokio tungstenite"
joelclaw docs search "TLS HTTPS rustls"

Testing

# When writing tests, benchmarks, property tests
joelclaw docs search "rust unit test integration test"
joelclaw docs search "mock testing trait objects"
joelclaw docs search "property based testing proptest"
joelclaw docs search "benchmark criterion performance"

Systems / Low-Level

# When dealing with memory, unsafe, FFI, atomics
joelclaw docs search "unsafe rust safety invariant"
joelclaw docs search "atomic operations memory ordering"
joelclaw docs search "FFI foreign function interface C"
joelclaw docs search "Pin Unpin self-referential"
joelclaw docs search "zero cost abstractions performance"

Project Structure & Patterns

# When scaffolding projects, organizing modules
joelclaw docs search "cargo workspace project structure"
joelclaw docs search "module system pub crate visibility"
joelclaw docs search "builder pattern rust idiomatic"
joelclaw docs search "type state pattern compile time"
joelclaw docs search "newtype pattern wrapper type"

Available Books

These books are indexed (or being indexed) in pdf-brain:

Workflow: Library-Informed Development

1. Before implementing: Search the library for the pattern you need 2. Read the chunk context: joelclaw docs context <id> --mode snippet-window 3. Adapt, don't copy: Book patterns are teaching examples — adapt to clawnode conventions 4. When stuck on a compiler error: Search for the error code or concept 5. When designing an API: Search for trait design, builder patterns, or similar abstractions

Tips

• Search terms should be conceptual, not exact phrases: "ownership move" not "the ownership system moves values"
• Combine domain + concept: "async error handling" not just "error handling"
• Use --mode snippet-window on context to get surrounding chunks for fuller picture
• If text search returns noise, try narrower terms or check the book title filter

Ownership, Borrowing & Lifetimes — Quick Reference

Core Rules

1. Each value has exactly one owner 2. When owner goes out of scope, value is dropped 3. You can have either ONE &mut T OR any number of &T — never both

Borrowing Patterns

// Immutable borrow — read-only, multiple allowed
fn len(s: &str) -> usize { s.len() }

// Mutable borrow — read+write, exclusive
fn push(s: &mut String) { s.push_str(" world"); }

// Move — ownership transfer
fn consume(s: String) { drop(s); }

Smart Pointers

Common Combos

• Arc<Mutex<T>> — shared mutable state across async tasks
• Rc<RefCell<T>> — shared mutable state in single thread (avoid in async)
• Arc<RwLock<T>> — read-heavy shared state across tasks

Lifetime Annotations

// Output lifetime tied to input
fn first<'a>(s: &'a str) -> &'a str { &s[..1] }

// Struct borrowing data
struct View<'a> { data: &'a [u8] }

// Static lifetime — lives forever
const NAME: &'static str = "clawnode";

// Lifetime elision (compiler infers these):
fn foo(s: &str) -> &str { s }  // Same as fn foo<'a>(s: &'a str) -> &'a str

Builder Pattern (Ownership-Friendly)

struct Config { host: String, port: u16 }

struct ConfigBuilder { host: Option<String>, port: Option<u16> }

impl ConfigBuilder {
    fn new() -> Self { Self { host: None, port: None } }
    fn host(mut self, h: impl Into<String>) -> Self { self.host = Some(h.into()); self }
    fn port(mut self, p: u16) -> Self { self.port = Some(p); self }
    fn build(self) -> anyhow::Result<Config> {
        Ok(Config {
            host: self.host.context("host required")?,
            port: self.port.unwrap_or(2583),
        })
    }
}

Decision Tree: Borrow or Own?

1. Does the callee need to store it long-term? → Own (take T) 2. Does the callee need to mutate it? → `&mut T` 3. Just reading? → `&T` 4. Need to return it from a function? → Check if you can return &T with lifetime, otherwise Own 5. Conditional mutation? → `Cow<T>`