# 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)

```mermaid
flowchart LR
    A["u32 LE<br/>total buffer len<br/><i>size prefix</i>"]
    B["u32 LE<br/>root table offset"]
    C["file identifier<br/>(4 bytes)"]
    D["tables · vtables · data"]
    A --> B --> C --> D
```

```rust
builder.finish_size_prefixed(root, Some("DESI"));
```

Reader side:

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

```odin
// 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`)

```mermaid
flowchart LR
    A["u32 LE<br/>message_len"]
    B["optional checksum<br/>(u32 crc / xxhash)"]
    C["FlatBuffer payload"]
    A --> B --> C
```

- `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:

```rust
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):

```odin
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
  `ArrayBuffer`s — you then copy the byte slice into WASM heap. See
  `04-odin-client-integration.md` Path D.

### HTTP streaming (alternative, simpler)

Server:

```rust
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)

```fbs
union Message {
  CatalogList,
  ObjectBatch,
  APIError,
}

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

root_type ServerMessage;
```

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

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

```fbs
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

```mermaid
flowchart TD
    A["HTTP GET → one FlatBuffer body per batch<br/>validate Rust builder + Odin reader"]
    B["WebSocket → one Binary message per batch<br/>interactive path · no custom framing"]
    C["Self-identifying messages<br/>file_identifier &#34;DESI&#34;"]
    D["size-prefixed / flatstream-style framing<br/>or HTTP-range + mmap for static catalogs"]
    A --> B --> C
    C -. "later, if needed" .-> D
```

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