fidius_guest/
frame.rs

1// Copyright 2026 Colliery, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Streaming frame format for the server-streaming plugin boundary
16//! (FIDIUS-I-0026, design decision D2).
17//!
18//! A streaming method call yields a *sequence of frames* rather than one buffer.
19//! Each frame is length-delimited and self-identifying:
20//!
21//! ```text
22//! [ tag: u8 ][ len: u32 little-endian ][ payload: len bytes ]
23//! ```
24//!
25//! Three tags exist:
26//!
27//! - [`FRAME_ITEM`] — one streamed item; `payload` is the existing bincode/`Value`
28//!   wire bytes, identical to a unary return value. Streaming therefore inherits
29//!   the unary wire format verbatim (one item == one unary payload).
30//! - [`FRAME_END`] — clean end of stream; `len == 0`, no payload.
31//! - [`FRAME_ERROR`] — the producer failed mid-stream; `payload` is a bincode
32//!   [`PluginError`]. An error frame *terminates* the stream: no frames may
33//!   follow it (enforced by the host-side decoder, [`crate::frame`] consumers).
34//!
35//! This framing is identical across every backend — a WIT `list<u8>` returned
36//! per `next()` (WASM), a `bytes` object yielded by a generator (Python), or a
37//! host-allocated buffer filled by an FFI `next()` export (cdylib). The scheme
38//! is deliberately batch-compatible: a future `next_batch(n)` (design decision
39//! D5) is simply *n* `ITEM` frames concatenated, needing no wire change.
40
41use crate::error::PluginError;
42use crate::wire::{self, WireError};
43
44/// Frame tag: one streamed item. Payload is bincode/`Value` wire bytes.
45pub const FRAME_ITEM: u8 = 0;
46/// Frame tag: clean end of stream. No payload.
47pub const FRAME_END: u8 = 1;
48/// Frame tag: producer error. Payload is a bincode [`PluginError`].
49pub const FRAME_ERROR: u8 = 2;
50
51/// Fixed size of a frame header: one tag byte plus a `u32` length.
52pub const FRAME_HEADER_LEN: usize = 1 + 4;
53
54/// One frame crossing the streaming boundary.
55///
56/// `Item` carries opaque payload bytes (the caller decides how to decode them —
57/// typically `wire::deserialize::<T>()` or a `Value`), so this type stays
58/// backend- and schema-neutral.
59#[derive(Debug, Clone, PartialEq)]
60pub enum Frame {
61    /// One streamed item; bytes are the bincode/`Value` payload.
62    Item(Vec<u8>),
63    /// Clean end of stream.
64    End,
65    /// Producer failed mid-stream; the stream terminates here.
66    Error(PluginError),
67}
68
69/// Errors decoding a [`Frame`] from bytes.
70#[derive(Debug, thiserror::Error)]
71pub enum FrameError {
72    /// Buffer ended before a full header or the declared payload was read.
73    #[error("truncated frame: needed {needed} bytes, had {had}")]
74    Truncated { needed: usize, had: usize },
75
76    /// The tag byte was not one of the known frame tags.
77    #[error("unknown frame tag: {0}")]
78    UnknownTag(u8),
79
80    /// The `ERROR` frame payload failed to decode as a `PluginError`.
81    #[error("malformed error-frame payload: {0}")]
82    Payload(#[from] WireError),
83
84    /// An `END` or `ERROR` frame declared a payload length it must not have, or
85    /// trailing bytes followed a terminal frame in a single-frame decode.
86    #[error("malformed frame: {0}")]
87    Malformed(String),
88}
89
90impl Frame {
91    /// Encode this frame as `[tag][len][payload]`.
92    pub fn encode(&self) -> Result<Vec<u8>, WireError> {
93        let (tag, payload): (u8, Vec<u8>) = match self {
94            Frame::Item(bytes) => (FRAME_ITEM, bytes.clone()),
95            Frame::End => (FRAME_END, Vec::new()),
96            Frame::Error(err) => (FRAME_ERROR, wire::serialize(err)?),
97        };
98        let mut out = Vec::with_capacity(FRAME_HEADER_LEN + payload.len());
99        out.push(tag);
100        out.extend_from_slice(&(payload.len() as u32).to_le_bytes());
101        out.extend_from_slice(&payload);
102        Ok(out)
103    }
104
105    /// Decode exactly one frame from `bytes`, which must contain a single frame
106    /// and nothing more. Use [`Frame::read`] to pull one frame from a larger
107    /// buffer and learn how many bytes were consumed.
108    pub fn decode(bytes: &[u8]) -> Result<Frame, FrameError> {
109        let (frame, consumed) = Frame::read(bytes)?;
110        if consumed != bytes.len() {
111            return Err(FrameError::Malformed(format!(
112                "{} trailing byte(s) after frame",
113                bytes.len() - consumed
114            )));
115        }
116        Ok(frame)
117    }
118
119    /// Read one frame from the front of `bytes`, returning the frame and the
120    /// number of bytes consumed. Lets a decoder walk a buffer holding several
121    /// concatenated frames (the cdylib `next_batch` / `Value`-rail path).
122    pub fn read(bytes: &[u8]) -> Result<(Frame, usize), FrameError> {
123        if bytes.len() < FRAME_HEADER_LEN {
124            return Err(FrameError::Truncated {
125                needed: FRAME_HEADER_LEN,
126                had: bytes.len(),
127            });
128        }
129        let tag = bytes[0];
130        let len = u32::from_le_bytes([bytes[1], bytes[2], bytes[3], bytes[4]]) as usize;
131        let end = FRAME_HEADER_LEN + len;
132        if bytes.len() < end {
133            return Err(FrameError::Truncated {
134                needed: end,
135                had: bytes.len(),
136            });
137        }
138        let payload = &bytes[FRAME_HEADER_LEN..end];
139        let frame = match tag {
140            FRAME_ITEM => Frame::Item(payload.to_vec()),
141            FRAME_END => {
142                if len != 0 {
143                    return Err(FrameError::Malformed(format!(
144                        "END frame carries {len} payload byte(s)"
145                    )));
146                }
147                Frame::End
148            }
149            FRAME_ERROR => Frame::Error(wire::deserialize::<PluginError>(payload)?),
150            other => return Err(FrameError::UnknownTag(other)),
151        };
152        Ok((frame, end))
153    }
154}
155
156#[cfg(test)]
157mod tests {
158    use super::*;
159
160    fn item(payload: &[u8]) -> Frame {
161        Frame::Item(payload.to_vec())
162    }
163
164    #[test]
165    fn item_round_trip() {
166        let f = item(&[1, 2, 3, 4]);
167        let bytes = f.encode().unwrap();
168        assert_eq!(bytes[0], FRAME_ITEM);
169        assert_eq!(Frame::decode(&bytes).unwrap(), f);
170    }
171
172    #[test]
173    fn end_round_trip() {
174        let bytes = Frame::End.encode().unwrap();
175        assert_eq!(bytes[0], FRAME_END);
176        assert_eq!(bytes.len(), FRAME_HEADER_LEN);
177        assert_eq!(Frame::decode(&bytes).unwrap(), Frame::End);
178    }
179
180    #[test]
181    fn error_round_trip() {
182        let err = PluginError::new("BOOM", "it broke");
183        let f = Frame::Error(err.clone());
184        let bytes = f.encode().unwrap();
185        assert_eq!(bytes[0], FRAME_ERROR);
186        assert_eq!(Frame::decode(&bytes).unwrap(), Frame::Error(err));
187    }
188
189    #[test]
190    fn empty_item_is_valid() {
191        let f = item(&[]);
192        let bytes = f.encode().unwrap();
193        assert_eq!(Frame::decode(&bytes).unwrap(), f);
194    }
195
196    #[test]
197    fn read_walks_concatenated_frames() {
198        let mut buf = Vec::new();
199        buf.extend_from_slice(&item(&[9]).encode().unwrap());
200        buf.extend_from_slice(&item(&[8, 7]).encode().unwrap());
201        buf.extend_from_slice(&Frame::End.encode().unwrap());
202
203        let (f1, n1) = Frame::read(&buf).unwrap();
204        assert_eq!(f1, item(&[9]));
205        let (f2, n2) = Frame::read(&buf[n1..]).unwrap();
206        assert_eq!(f2, item(&[8, 7]));
207        let (f3, n3) = Frame::read(&buf[n1 + n2..]).unwrap();
208        assert_eq!(f3, Frame::End);
209        assert_eq!(n1 + n2 + n3, buf.len());
210    }
211
212    #[test]
213    fn truncated_header_is_rejected() {
214        let err = Frame::read(&[FRAME_ITEM, 0, 0]).unwrap_err();
215        assert!(matches!(err, FrameError::Truncated { .. }));
216    }
217
218    #[test]
219    fn truncated_payload_is_rejected() {
220        // Declares 8 payload bytes but only supplies 2.
221        let mut bytes = vec![FRAME_ITEM];
222        bytes.extend_from_slice(&8u32.to_le_bytes());
223        bytes.extend_from_slice(&[1, 2]);
224        let err = Frame::decode(&bytes).unwrap_err();
225        assert!(matches!(err, FrameError::Truncated { needed: 13, had: 7 }));
226    }
227
228    #[test]
229    fn unknown_tag_is_rejected() {
230        let mut bytes = vec![99u8];
231        bytes.extend_from_slice(&0u32.to_le_bytes());
232        assert!(matches!(
233            Frame::decode(&bytes).unwrap_err(),
234            FrameError::UnknownTag(99)
235        ));
236    }
237
238    #[test]
239    fn end_with_payload_is_rejected() {
240        let mut bytes = vec![FRAME_END];
241        bytes.extend_from_slice(&3u32.to_le_bytes());
242        bytes.extend_from_slice(&[1, 2, 3]);
243        assert!(matches!(
244            Frame::decode(&bytes).unwrap_err(),
245            FrameError::Malformed(_)
246        ));
247    }
248
249    #[test]
250    fn trailing_bytes_after_single_decode_rejected() {
251        let mut bytes = item(&[1]).encode().unwrap();
252        bytes.push(0xFF);
253        assert!(matches!(
254            Frame::decode(&bytes).unwrap_err(),
255            FrameError::Malformed(_)
256        ));
257    }
258
259    #[test]
260    fn garbage_is_rejected_not_panicking() {
261        for g in [vec![], vec![0u8], vec![2u8, 0, 0, 0, 1]] {
262            let _ = Frame::read(&g); // must return Err/Ok, never panic
263        }
264    }
265}
fidius_guest/frame.rs

fidius_guest/
frame.rs
