desi_explorer/resources/ai/research/data-streaming/05-streaming-protocols.md

05 — Streaming Protocols & Transport

The wire isn't just the serialization — it's the transport + framing. This doc covers how to move framed FlatBuffer messages from the Rust/axum API to the Odin GUI.

High-level options

Transport	Full-duplex	Server push	Browser (WASM) support	Native Odin support	Fit for DESI Explorer
HTTP GET + paginated responses	No	No	Native fetch (works via JS bridging)	core:net / curl	Baseline; simplest start
HTTP chunked streaming (SSE-like)	No	One-way stream	fetch streaming / SSE	core:net	Good for one-shot bulk pushes, no client→server mid-stream control
WebSocket	Yes	Yes	Native browser API	Hand-rolled or FFI	Best overall for interactive pan/zoom + chunked object streaming
Raw TCP	Yes	Yes	N/A (browser can't do raw TCP)	core:net	Best perf but breaks the web target

Recommendation: WebSocket for the interactive streaming path. It supports bidirectional messaging (client asks for a catalog region; server pushes chunks), works in both native Odin and the browser, and axum has first-class ws support.

Framing — where one message ends and the next begins

A byte stream gives you no message boundaries. You need framing. FlatBuffers offers two built-in mechanisms plus the community pattern:

Option 1: Size-prefixed FlatBuffers (built-in)

builder.finish_size_prefixed(root, Some("DESI"));
// +---------------------------+
// | u32 LE: total buffer len  |  <-- size prefix
// | u32 LE: root table offset |
// | file identifier (4 bytes) |
// | ... data ...              |
// +---------------------------+

Reader side:

// Rust
let msg = flatbuffers::size_prefixed_root::<ObjectBatch>(&buf).unwrap();

// Odin (hand-rolled)
length := read_u32_le(buf)              // skip this many bytes for the payload
root_offset := read_u32_le(buf + 4)     // root table starts here

Pros: zero framing code; built into the format. Cons: doesn't carry an explicit checksum; one message type per stream (fine here unless you multiplex multiple message kinds).

Option 2: Custom length-prefix framing (like flatstream)

[ u32 LE: message_len ] [ optional checksum (e.g. u32 crc/xxhash) ] [ flatbuffer payload ]

• flatstream-rs (see 03-rust-integration.md) is a reference implementation of exactly this pattern: length prefix + optional checksum + zero-copy reads.
• For a WebSocket you could rely on WebSocket frames as boundaries — each Message::Binary is already a discrete frame. Simpler than custom framing: one WebSocket message = one FlatBuffer. This is the lowest-effort and totally viable option. FlatBuffers even recommends size-prefixed buffers for streams, but if your transport already frames (WebSocket, HTTP chunked), the extra size prefix is optional — though harmless to keep for uniformity.

Option 3: Batch as a single buffer (no framing needed)

If each HTTP response / WebSocket message is a complete ObjectBatch (columnar) FlatBuffer, you don't need in-band framing at all — the response is one message. Framing only matters when you stream multiple messages per connection. For DESI Explorer:

• HTTP path: one response = one batch → no extra framing.
• WebSocket path: one Binary message = one batch → no extra framing (WebSocket frames give you boundaries).
• Raw TCP path: you need framing (Option 1 or 2).

Transport deep-dive

WebSocket (axum)

Server:

use axum::{
    extract::ws::{Message, WebSocket, WebSocketUpgrade},
    extract::State,
    routing::any,
    Router,
};

async fn ws_handler(ws: WebSocketUpgrade, State(state): State<Arc<AppState>>) -> Response {
    ws.max_message_size(64 * 1024 * 1024) // 64 MB
      .on_upgrade(|socket| handle_socket(socket, state))
}

async fn handle_socket(socket: WebSocket, state: Arc<AppState>) {
    let (mut sender, mut receiver) = socket.split();

    // Writer task: sends FlatBuffer batches from a broadcast channel
    let mut rx = state.tx.subscribe();
    let send_task = tokio::spawn(async move {
        while let Ok(batch) = rx.recv().await {
            if sender.send(Message::Binary(batch.into())).await.is_err() {
                break; // client disconnected
            }
        }
    });

    // Reader task: handles client requests (region query, filters)
    while let Some(Ok(msg)) = receiver.next().await {
        match msg {
            Message::Text(q) => handle_query(q, &state),  // triggers a push
            Message::Close(_) => break,
            _ => {}
        }
    }
    send_task.abort();
}

Key points:

• Message::Binary(Bytes) is the native carrier for FlatBuffer payloads.
• Use tokio::sync::broadcast for fan-out to many clients (window Lagged errors; a slow client falls behind → notify instead of blocking).
• max_message_size and max_frame_size tunables matter for large batches.
• WebSocketUpgrade works over HTTP/1.1 GET upgrades; in a browser, the connection is ws:///wss://.

Client (Odin, native):

import "core:net"

// 1. Open TCP to host:port
sock, err := net.dial_tcp(endpoint)
// 2. Send the HTTP upgrade handshake:
//    GET /ws HTTP/1.1
//    Host: ...
//    Upgrade: websocket
//    Connection: Upgrade
//    Sec-WebSocket-Key: <base64 16 bytes>
//    Sec-WebSocket-Version: 13
// 3. Parse 101 Switching Protocols response.
// 4. Frame format (client→server): [fin|opcode=2 (binary)] [mask=1] [len] [mask-4B] [payload]
// 5. Server→client frames: [fin|opcode=2] [len] [payload]  (no mask requirement)

• Odin core:net has no WebSocket built-in; hand-rolling the upgrade + simple frame reader is ~150-250 lines for the subset you need (text in, binary out). There are community implementations (e.g. furbs has a networking lib).
• For the WASM build, JS owns the WebSocket (new WebSocket(url)), delivers ArrayBuffers — you then copy the byte slice into WASM heap. See 04-odin-client-integration.md Path D.

HTTP streaming (alternative, simpler)

Server:

use axum::body::Body;
use axum::http::header::CONTENT_TYPE;
use futures_util::StreamExt;
use tokio::sync::mpsc;

async fn stream_objects(Query(q): Query<ObjectQuery>) -> Response {
    let (tx, rx) = mpsc::channel::<flatbuffers::Vec<u8>>(8);
    tokio::spawn(async move {
        for chunk in paginate_desi(q).await.unwrap().chunks(50_000) {
            let _ = tx.send(build_object_batch_flatbuffer(chunk)).await; // backpressure via channel
        }
    });
    let stream = tokio_stream::wrappers::ReceiverStream::new(rx)
        .map(|bytes| Ok::<_, std::io::Error>(Bytes::from(bytes)));
    Response::new(Body::from_stream(stream))
}

• Client reads length-prefixed messages from the response body.
• Simpler than WebSocket (no handshake/frames), but one-directional and no mid-stream client control.

Raw TCP (maximum perf, breaks browser)

• core:net TCP + length-prefixed frames. Low overhead, but the web GUI can't participate. Only worth it if the native desktop build must outperform WebSocket by a large margin (it won't in practice for this workload — network bandwidth dominates, and WebSocket adds ~2-14 bytes/frame depending on size).

Message envelope design

Even with transport framing, you'll likely multiplex message kinds over the same stream (catalog list, object batch, error, ping). Two clean patterns:

Pattern A — size-prefixed buffer + ServerMessage union (fully self-describing)

union Message {
  CatalogList,
  ObjectBatch,
  APIError,
}

table ServerMessage {
  msg: Message;        // disc
  catalog_list: CatalogList;
  object_batch: ObjectBatch;
  error: APIError;
}

root_type ServerMessage;

The client verifys the whole buffer and pattern-matches the union tag. One buffer format for everything.

Pattern B — per-kind size-prefixed buffers (simpler decoder)

root_type CatalogList;   // separate finish buffer
root_type ObjectBatch;   // separate finish buffer
root_type APIError;      // separate finish buffer

Each message is a different root type. The client sniffs the file identifier (sector size prefix size_prefixed_root_with_opts) to dispatch. Slightly less structural safety, but each decoder is trivial.

Recommendation: Pattern A if the stream carries mixed traffic (interactive control + data + errors); Pattern B if the path is a pure object stream (e.g. dedicated /ws/objects endpoints per kind).

Backpressure & chunking

• Chunk size: batch ~ tens of thousands of objects (~1-4 MB) keeps latency tight for interactive pan/zoom while amortizing per-buffer overhead.
• mpsc channel: cap in-flight buffered batches (e.g. 8) so a slow client never piles up unbounded memory server-side. tokio::sync::broadcast instead for fan-out, with Lagged handling.
• Client: buffer at most N in-flight frames; drop old detail when a new region query supersedes an in-flight one. With zero-copy reads this is cheap: each frame is one buffer you can free wholesale.
• Cadence: push an initial batch immediately on connect; then respond to client region moves, not a blind rate-limited firehose. (Unless "streaming" means full-survey playback — then a T-state machine with cursor/seq is better.)

HTTP-range / mmap alternative (worth remembering)

If catalogs become static cloud files (like Apache Arrow / FlatCityBuf's model), you can skip the API entirely for bulk data: ship pre-built .fbs files and let the client do HTTP range requests straight from object storage. The FlatCityBuf project (see references) does exactly this and is a working proof of concept: spatial index stored beside the FlatBuffer file, Range requests fetch only the slabs you need.

This is probably a later-stage architecture, but it's the strongest argument for FlatBuffers long-term (mmap-friendly, page-in-what-you-touch).

Decision summary for this repo

1. Start with HTTP GET → one FlatBuffer body per batch to validate the Rust builder + Odin reader (no protocol work at all).
2. Then add WebSocket with one Binary message per batch (no custom framing) for the interactive path.
3. Keep messages self-identifying via file identifiers (file_identifier "DESI").
4. Broaden later: hot spots can move to size-prefixed/flatstream-style framing or even HTTP-range + mmap if catalogs go static.

References

• axum ws docs: https://docs.rs/axum/latest/axum/extract/ws/index.html
• flatstream-rs (framing reference): https://github.com/dallasmarlow/flatstream-rs
• FlatCityBuf (HTTP-range + FlatBuffers + WASM proof-of-concept): https://github.com/cityjson/flatcitybuf
• Apache Arrow IPC framing spec (continuation + length prefix pattern): https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format