updated procedures in data.odin to pass url; added research notes for data-streaming
This commit is contained in:
@@ -0,0 +1,278 @@
|
||||
# 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)
|
||||
|
||||
```rust
|
||||
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
|
||||
// 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`)
|
||||
|
||||
```
|
||||
[ 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:
|
||||
|
||||
```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
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user