desi_explorer/resources/ai/research/data-streaming/03-rust-integration.md

03 — Rust / API Integration

Crate landscape

flatbuffers (runtime, official)

• crate: flatbuffers — https://crates.io/crates/flatbuffers
• Version: 25.12.19 (Dec 2025). Note: project moved to CalVer in 2025 (24.12, 25.1, 25.2, 25.9, 25.12.19). The old 2.1.x line is legacy.
• Support level: labeled "experimental" historically, but is the official Google runtime; used in production at Google and elsewhere. API stable enough for real usage.
• Features: serde (optional), std (default), also supports no_std.
• Dependencies: bitflags only.
• Downloads: ~17M/month, 174 dependents — well maintained.

flatbuffers-build (build-time codegen, community)

• crate: flatbuffers-build — wraps flatc so .fbs → .rs happens in build.rs instead of manually. Requires flatc on PATH (or FLATC_PATH).
• Caveat: it writes generated code into a symlinked directory under src/ (set_symlink_directory("src/generated")) because Cargo can't include OUT_DIR output as a module. This is a known hack; works but slightly ugly.
• Alternative: the flatc crate (https://crates.io/crates/flatc) vendors a prebuilt flatc binary as a dependency.

flatc (vendored binary, community)

• Crate flatc 0.2.2+23.5.26 bundles the flatc executable. Let's you invoke flatc from build.rs without requiring it system-wide.

flatstream (framing layer, community — very relevant)

• crate: flatstream — https://github.com/dallasmarlow/flatstream-rs
• Adds a framing layer for streams of FlatBuffers: each frame is [4-byte LE length][optional checksum][payload].
• Zero-copy read via StreamReader::process_all(&[u8]).
• Originated as a high-frequency telemetry capture format. Composability: bounded payload caps, checksum via xxhash64/crc32, observers, validators.
• Sync IO only (std Read/Write). Not async — you'd pair it with tokio::task::spawn_blocking for network use, or just write your own framing (it's ~20 lines).
• Note: the crate is young (2025) and small. Treat as a reference pattern, not a hard dependency. A size-prefixed FlatBuffer (builder.finish_size_prefixed(root, Some("ID"))) already gives you framing for free if you only need one message type per stream.

Choice matrix

Need	Crate	Notes
Runtime read/write	flatbuffers	Required regardless
.fbs → .rs manually	flatc binary	Simple, deterministic, scriptable
.fbs → .rs in build.rs	flatbuffers-build	Needs flatc on PATH; symlink hack
Vendored flatc binary	flatc crate	Avoids system install
Stream framing	flatstream (or hand-rolled)	4-byte LE length prefix pattern
Testing support	flatbuffers verification + cargo test, proptest, quickcheck	See testing doc

Codegen options for the API

Option A — Commit generated code (recommended starting point)

flatc --rust -o api/src/generated catalog.fbs

Then mod catalog_generated; in lib.rs. Pros: dead simple, zero build scripts, works with cargo test out of the box, diffable in code review. Cons: regenerate manually on schema change, risk of stale generated code.

Option B — build.rs codegen

// api/build.rs
fn main() {
    let schemas = ["../schema/catalog.fbs"];
    flatbuffers_build::BuilderOptions::new_with_files(&schemas)
        .set_symlink_directory("src/generated")
        .compile()
        .expect("flatbuffer compilation failed");
}

Pros: automatic regeneration. Cons: symlink hack, flatc must be installed (CI needs it too), cargo::rerun-if-changed wiring is manual-ish. For a repo that already pins toolchains in CI, adding flatc to .gitea/workflows/ci.yml is mechanical.

Option C — shared schema + make target

The project's build is make-based and mono-repo style. A root target like

schema-gen:
	flatc --rust -o api/src/generated ../schema/catalog.fbs
	flatcc --c -o gui/src/generated ../schema/catalog.fbs

run before make build/make test keeps both sides in sync from one schema file. This fits the existing "root Makefile delegates" pattern well.

Recipe: serving FlatBuffers from axum

Dependencies

[dependencies]
axum = { version = "0.8", features = ["ws"] }   # ws feature for WebSocket
flatbuffers = "25.12.19"                        # runtime
# for build.rs codegen (if you choose Option B):
[build-dependencies]
flatbuffers-build = "0.2"   # or use `flatc` binary option

Endpoint that returns a FlatBuffer body (content-negotiated)

use axum::{body::Bytes, extract::Query, http::StatusCode, response::{IntoResponse, Response}};
use flatbuffers::{FlatBufferBuilder, WIPOffset};

mod catalog_generated;
use catalog_generated::desi::{CatalogObject, CatalogObjectArgs, ObjectBatch, ObjectBatchArgs,
                               finish_object_batch_buffer, root_as_object_batch};

async fn objects(query: Query<ObjectQuery>) -> Response {
    let rows = fetch_desi_objects(&query.catalog, query.limit.clamp(1, 100_000)).await;

    let mut fbb = FlatBufferBuilder::with_capacity(1 << 20);
    let ra = fbb.create_vector(&rows.ra);
    let dec = fbb.create_vector(&rows.dec);
    let z = fbb.create_vector(&rows.z);
    let ids = fbb.create_vector(&rows.id_offsets);

    let batch = ObjectBatch::create(&mut fbb, &ObjectBatchArgs {
        catalog: Some(fbb.create_string(&rows.catalog)),
        n: rows.len as u64,
        ra: Some(ra),
        dec: Some(dec),
        redshift: Some(z),
        ids: Some(ids),
        ..Default::default()
    });
    finish_object_batch_buffer(&mut fbb, batch);

    ([(header::CONTENT_TYPE, "application/octet-stream")], Bytes::copy_from_slice(fbb.finished_data()))
        .into_response()
}

WebSocket endpoint

use axum::{
    extract::ws::{Message, WebSocket, WebSocketUpgrade},
    response::Response,
    routing::any,
};
use futures_util::{stream::StreamExt, sink::SinkExt};

async fn ws_handler(ws: WebSocketUpgrade) -> Response {
    ws.on_upgrade(handle_socket)
}

async fn handle_socket(mut socket: WebSocket) {
    // On connect, stream catalog batches as Binary(Message::Binary(...)) frames:
    // each frame's payload is a size-prefixed or length-prefixed FlatBuffer.
    while let Some(Ok(msg)) = socket.recv().await {
        match msg {
            Message::Text(query) => {
                // Client asked for a catalog region; stream batch FlatBuffers back.
                let rows = fetch_region(&query).await;
                for chunk in rows.chunks(50_000) {
                    let buf = build_object_batch_flatbuffer(chunk);
                    socket.send(Message::Binary(buf.into())).await.unwrap();
                }
            }
            Message::Close(_) => break,
            _ => {}
        }
    }
}

Notes:

• WebSocket binary messages (Message::Binary) are exactly-typed Vec<u8>/ Bytes — ideal for FlatBuffer payloads with no extra encoding.
• Use WebSocketUpgrade.max_message_size(...) to cap client message size (default 64 MB).
• For a single server pushing to many clients, use tokio::sync::broadcast and split the socket (socket.split() → SinkExt for sending, StreamExt for receiving) — see 05-streaming-protocols.md.
• If the Odin client is WASM (browser), CORS matters: add tower-http's CorsLayer since the WASM origin differs from the API origin. WebSockets aren't CORS-restricted the same way HTTP is, but the HTTP upgrade handshake still goes through the middleware stack.

Building buffers efficiently (Rust specifics)

Build order is back-to-front (children before parents):

flowchart TB
    A["create_string / create_vector<br/>children first"]
    B["create nested child tables"]
    C["create parent table<br/>ObjectBatch::create(&args)"]
    D["builder.finish(root, Some(&#34;DESI&#34;))"]
    E["Bytes::copy_from_slice(fbb.finished_data())<br/>→ HTTP / WebSocket response"]
    A --> B --> C --> D --> E

• FlatBufferBuilder::with_capacity(n) pre-allocates; reset() reuses the buffer across messages. In a loop streaming batches, create one builder, reuse it — avoid repeated reallocation.
• Builder methods have try_* counterparts (try_create_string, try_push, try_finish) that return Result instead of panicking on out-of-capacity — useful on the server under memory pressure.
• with_internal_capacity(size, field_locs, vtables, strings) preallocates all internal data structures — removes allocation-related latency spikes in a latency-sensitive server path. Combined with reset(), a long-lived stream handler does zero allocations after startup.
• For bulk numeric data, prefer vectors of scalars (columnar layout) over create_vector of tables. create_vector on &[f64] is a fast path; on little-endian it back-fills with scalar offsets, enabling safe_slice reads. Result: one contiguous &[f64] slice per column on the client.
• Building is back-to-front: children (strings, vectors, nested tables) must be created before the parent table referencing them. The *Args struct pattern (ObjectBatchArgs { .. }) makes this manageable, but be deliberate about creation order.
• builder.finished_data() returns &[u8] borrowing the builder — copy to an owned Bytes for the response, then reset().

Serialization on the server (axes to optimize)

• Batch size tradeoff: each batch carries one vtable + one table header; amortize over more rows. Choose batch size so the buffer is well under the 2 GiB limit and matches your streaming cadence (e.g. 50k objects ≈ ~3 MB).
• CPU cost: FlatBuffers serialization is fast but not free (it's the read-side where FlatBuffers shines). Measure with criterion; if serialization becomes the bottleneck, move builder work to a dedicated worker thread / spawn_blocking and hand finished batches to the WebSocket sender via a channel.
• Vtables: tables with identical field layouts share vtables automatically — batching objects with the same schema shape is free (they all share).

Integration with existing serde models

You don't have to delete serde. Pragmatic path:

• Keep CatalogObject (serde) for JSON endpoints / debugging / curl-ability.
• Add a separate FlatBuffers representation for the binary path, generated from the shared .fbs.
• Add a From<CatalogObject> for ObjectBatchArgs-style conversion, or better: build FlatBuffers directly from the raw DESI rows (skip the serde struct entirely, avoiding a serialize-then-rebuild pass). The DESI store should hand you vectors of f64; feed those straight into create_vector.

This preserves today's test shape (api/tests/health.rs parses JSON) and adds the binary path without churn.

CI / build concerns for this repo

• The Gitea workflow (ci.yml) runs cargo test + clippy. If you switch to build.rs codegen, CI needs flatc installed (or use the vendored flatc crate). If you commit generated code, CI needs nothing extra — one more reason Option A (commit generated code) is the lowest-friction start.
• Pin the same schema file for both API and GUI. If the schema lives in a shared schema/ directory and both sides' Makefiles depend on it, make test stays the single verification step: Rust tests + Odin tests + Go tests each validate their own pipeline against the shared contract.
• Mind the repo convention: make -C api check is the passthrough for cargo test + clippy. Keep the FlatBuffers stack incrementally addable here.

References

• https://flatbuffers.dev/languages/rust/ — official Rust usage doc
• https://docs.rs/flatbuffers — runtime API reference
• https://github.com/frol/flatc-rust — flatc-as-build-API (alternative)
• https://github.com/dallasmarlow/flatstream-rs — framing reference
• https://crates.io/crates/flatbuffers-build — codegen crate