shep_core/protocol/
wire.rs

1//! Frame encoding: u32 length prefix + JSON payload
2//!
3//! One codec constructor + encode/decode helpers shared by daemon and
4//! client so framing parameters can never drift between the two.
5
6use core::fmt;
7
8use bytes::Bytes;
9use serde::Serialize;
10use serde::de::DeserializeOwned;
11use tokio_util::codec::LengthDelimitedCodec;
12
13/// Hard ceiling per frame; larger is a protocol violation
14pub const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
15
16/// Builds the shared length-delimited codec (u32 BE prefix, 16 MiB cap)
17#[must_use]
18pub fn codec() -> LengthDelimitedCodec {
19    LengthDelimitedCodec::builder()
20        .length_field_type::<u32>()
21        .max_frame_length(MAX_FRAME_BYTES)
22        .new_codec()
23}
24
25/// Serializes one value to a frame payload
26///
27/// # Errors
28///
29/// - [`WireError::Json`]: serialization failed (carries serde's message).
30/// - [`WireError::FrameTooLarge`]: payload exceeds [`MAX_FRAME_BYTES`].
31pub fn encode_frame<T: Serialize>(value: &T) -> Result<Bytes, WireError> {
32    let vec = serde_json::to_vec(value).map_err(|e| WireError::Json(e.to_string()))?;
33    if vec.len() > MAX_FRAME_BYTES {
34        return Err(WireError::FrameTooLarge(vec.len()));
35    }
36    Ok(Bytes::from(vec)) // zero-copy: Bytes takes the Vec's buffer
37}
38
39/// Deserializes one frame payload
40///
41/// # Errors
42///
43/// - [`WireError::Json`]: the payload is not valid JSON for `T`.
44pub fn decode_frame<T: DeserializeOwned>(frame: &[u8]) -> Result<T, WireError> {
45    serde_json::from_slice(frame).map_err(|e| WireError::Json(e.to_string()))
46}
47
48/// Error type returned from [`encode_frame`] and [`decode_frame`]
49///
50/// `#[non_exhaustive]`: this type is on the peer-facing surface, and the
51/// protocol will grow past a JSON payload and a size cap eventually.
52#[non_exhaustive]
53#[derive(Debug, Clone, PartialEq, Eq)]
54pub enum WireError {
55    /// JSON (de)serialization failed (carries the serde message)
56    Json(String),
57    /// Encoded payload exceeds [`MAX_FRAME_BYTES`] (carries actual size)
58    FrameTooLarge(usize),
59}
60
61impl fmt::Display for WireError {
62    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
63        match self {
64            Self::Json(m) => write!(f, "wire frame JSON error: {m}"),
65            Self::FrameTooLarge(n) => {
66                write!(
67                    f,
68                    "frame of {n} bytes exceeds the {MAX_FRAME_BYTES}-byte limit"
69                )
70            }
71        }
72    }
73}
74
75impl core::error::Error for WireError {}
76
77#[cfg(test)]
78mod tests {
79    use super::*;
80    use crate::protocol::{Envelope, Request};
81
82    #[test]
83    fn encode_decode_round_trip() {
84        let env = Envelope {
85            id: 9,
86            deadline_ms: Some(5000),
87            body: Request::Ping,
88        };
89        let bytes = encode_frame(&env).unwrap();
90        let back: Envelope = decode_frame(&bytes).unwrap();
91        assert_eq!(back, env);
92    }
93
94    #[test]
95    fn decode_rejects_garbage_with_json_error() {
96        assert!(matches!(
97            decode_frame::<Envelope>(b"not json"),
98            Err(WireError::Json(_))
99        ));
100    }
101
102    #[test]
103    fn codec_uses_u32_prefix_and_max_frame() {
104        let c = codec();
105        assert_eq!(c.max_frame_length(), MAX_FRAME_BYTES);
106    }
107
108    #[tokio::test]
109    async fn framed_stream_round_trip() {
110        use futures_util::{SinkExt, StreamExt};
111        use tokio_util::codec::{FramedRead, FramedWrite};
112
113        let (client, server) = tokio::io::duplex(64 * 1024);
114        let mut writer = FramedWrite::new(client, codec());
115        let mut reader = FramedRead::new(server, codec());
116
117        let env = Envelope {
118            id: 1,
119            deadline_ms: None,
120            body: Request::ListFlock,
121        };
122        writer.send(encode_frame(&env).unwrap()).await.unwrap();
123
124        let frame = reader.next().await.unwrap().unwrap();
125        let back: Envelope = decode_frame(&frame).unwrap();
126        assert_eq!(back, env);
127    }
128}
shep_core/protocol/wire.rs

shep_core/protocol/
wire.rs
