vgi_rpc/
stream_codec.rs

1//! Serialization codec for streaming state.
2//!
3//! The HTTP transport is stateless: every continuation request carries
4//! the serialized [`ProducerState`](crate::ProducerState) /
5//! [`ExchangeState`](crate::ExchangeState) back to the server inside an
6//! HMAC-signed token, so any worker behind a load balancer can resume
7//! any stream. This module defines the trait for per-state-type
8//! encode/decode + a bincode-backed helper for the common case.
9//!
10//! Pipe and unix transports hold state in memory and skip the codec
11//! entirely.
12
13use serde::{de::DeserializeOwned, Serialize};
14
15use crate::errors::{Result, RpcError};
16
17/// Round-trip a streaming-state value through a byte representation.
18///
19/// Implementations choose their own format (bincode, Arrow IPC, etc.);
20/// the bytes are opaque to the HTTP transport which only signs and
21/// carries them.
22pub trait StreamStateCodec: Sized {
23    fn encode(&self) -> Result<Vec<u8>>;
24    fn decode(bytes: &[u8]) -> Result<Self>;
25}
26
27/// Encode a `serde::Serialize` value with bincode.
28///
29/// **Internal:** used by `#[derive(StreamState)]` expansion. Most
30/// users should implement [`StreamStateCodec`] manually or via the
31/// derive macro rather than calling this directly.
32#[doc(hidden)]
33pub fn bincode_encode<T: Serialize>(value: &T) -> Result<Vec<u8>> {
34    bincode::serialize(value).map_err(|e| RpcError::runtime_error(format!("bincode encode: {e}")))
35}
36
37/// Hard ceiling on the number of bytes `bincode_decode` will allocate
38/// from a length-prefixed field. Streaming-state values are small
39/// (counters, cursors, small buffers); 16 MiB is generous headroom.
40/// The HTTP transport already seals these bytes in an AEAD token, so
41/// they are integrity-protected — this is defence-in-depth against a
42/// crafted length prefix should the codec ever be fed untrusted bytes.
43const MAX_STATE_DECODE_BYTES: u64 = 16 * 1024 * 1024;
44
45/// Decode bytes produced by [`bincode_encode`].
46///
47/// **Internal:** used by `#[derive(StreamState)]` expansion.
48#[doc(hidden)]
49pub fn bincode_decode<T: DeserializeOwned>(bytes: &[u8]) -> Result<T> {
50    use bincode::Options;
51    // `DefaultOptions` matches the byte layout of `bincode::serialize`
52    // (used by `bincode_encode`); `.with_limit` bounds allocation
53    // without changing the wire format.
54    bincode::DefaultOptions::new()
55        .with_fixint_encoding()
56        .with_limit(MAX_STATE_DECODE_BYTES)
57        .deserialize(bytes)
58        .map_err(|e| RpcError::runtime_error(format!("bincode decode: {e}")))
59}
vgi_rpc/stream_codec.rs

vgi_rpc/
stream_codec.rs
