running_process/
frame_v1.rs

1//! Frozen v1 `Frame` envelope primitives.
2//!
3//! This is the smallest running-process surface for a daemon that already
4//! owns its endpoint and its application payload. It contains only the wire
5//! envelope, incremental and stream codecs, and payload-protocol registration
6//! checks. It deliberately does not select an IPC transport, hash a daemon
7//! image, parse configuration, or start a runtime.
8//!
9//! The exact wire layout is `[u8 framing_version=1][u32 LE body_length][prost
10//! Frame]`. These values are frozen for v1; broad broker paths re-export the
11//! literal same items from this module for source and type compatibility.
12
13use std::io::{self, Read, Write};
14
15use prost::Message;
16
17pub use running_process_protocol::broker::v1::{Frame, FrameKind, PayloadEncoding};
18
19/// Framing byte for every v1 broker `Frame`.
20pub const FRAMING_VERSION_V1: u8 = 1;
21
22/// Hard ceiling on one v1 frame body.
23pub const MAX_FRAME_SIZE_BYTES: usize = 16 * 1024 * 1024;
24
25/// Hard ceiling on the initial v1 Hello envelope.
26pub const MAX_HELLO_SIZE_BYTES: usize = 64 * 1024;
27
28/// Framing byte for v1. Alias of [`FRAMING_VERSION_V1`].
29pub const ENVELOPE_VERSION: u8 = FRAMING_VERSION_V1;
30
31/// Default per-frame size cap (16 MiB). Alias of [`MAX_FRAME_SIZE_BYTES`].
32pub const MAX_FRAME_BYTES: usize = MAX_FRAME_SIZE_BYTES;
33
34/// Hello-envelope size cap (64 KiB). Alias of [`MAX_HELLO_SIZE_BYTES`].
35pub const MAX_HELLO_BYTES: usize = MAX_HELLO_SIZE_BYTES;
36
37/// Length of the outer wire header: `[u8 framing_version][u32 LE body_len]`.
38pub const FRAME_HEADER_BYTES: usize = 5;
39
40/// Errors produced while reading, writing, or incrementally decoding a v1
41/// `Frame` envelope.
42#[derive(Debug, thiserror::Error)]
43pub enum FramingError {
44    /// Peer's framing byte did not match [`ENVELOPE_VERSION`].
45    #[error("unsupported framing version: got {got}, expected {expected}")]
46    UnsupportedFramingVersion {
47        /// The framing byte the peer actually sent.
48        got: u8,
49        /// The frozen framing byte this codec expects.
50        expected: u8,
51    },
52
53    /// Body length exceeds the configured per-frame cap.
54    #[error("frame body too large: {body_length} bytes exceeds cap {cap}")]
55    FrameTooLarge {
56        /// The length announced in the four-byte little-endian header.
57        body_length: usize,
58        /// The cap applied by the caller or frozen default.
59        cap: usize,
60    },
61
62    /// The stream ended before its complete frame arrived.
63    #[error("unexpected EOF while reading frame ({context})")]
64    UnexpectedEof {
65        /// Which part of the frame was incomplete.
66        context: &'static str,
67    },
68
69    /// Raw stream I/O failure.
70    #[error("I/O error: {0}")]
71    Io(#[from] io::Error),
72
73    /// The complete frame body was not a valid protobuf `Frame`.
74    #[error("failed to decode Frame body: {0}")]
75    Decode(#[from] prost::DecodeError),
76}
77
78/// Encode one [`Frame`] into complete v1 wire bytes.
79///
80/// # Errors
81///
82/// Returns [`FramingError::FrameTooLarge`] when the encoded body exceeds the
83/// frozen [`MAX_FRAME_BYTES`] cap.
84pub fn encode_framed(frame: &Frame) -> Result<Vec<u8>, FramingError> {
85    let body_len = frame.encoded_len();
86    if body_len > MAX_FRAME_BYTES {
87        return Err(FramingError::FrameTooLarge {
88            body_length: body_len,
89            cap: MAX_FRAME_BYTES,
90        });
91    }
92    let mut wire = Vec::with_capacity(FRAME_HEADER_BYTES + body_len);
93    wire.push(ENVELOPE_VERSION);
94    wire.extend_from_slice(&(body_len as u32).to_le_bytes());
95    frame
96        .encode(&mut wire)
97        .expect("prost encoding into Vec cannot fail because Vec writes are infallible");
98    Ok(wire)
99}
100
101/// One [`Frame`] decoded from the front of a byte buffer.
102#[derive(Debug, Clone, PartialEq)]
103pub struct DecodedFramed {
104    /// Decoded frozen v1 envelope.
105    pub frame: Frame,
106    /// Total bytes occupied by the outer header and protobuf body.
107    pub consumed: usize,
108}
109
110/// Incrementally decode one [`Frame`] from the front of `buf`.
111///
112/// Returns `Ok(None)` without consuming or interpreting a partial header or
113/// body. Foreign framing versions, oversize declarations, and malformed
114/// protobuf bodies are terminal errors for the current connection.
115pub fn try_decode_framed(buf: &[u8]) -> Result<Option<DecodedFramed>, FramingError> {
116    if buf.is_empty() {
117        return Ok(None);
118    }
119    if buf[0] != ENVELOPE_VERSION {
120        return Err(FramingError::UnsupportedFramingVersion {
121            got: buf[0],
122            expected: ENVELOPE_VERSION,
123        });
124    }
125    if buf.len() < FRAME_HEADER_BYTES {
126        return Ok(None);
127    }
128    let body_len = u32::from_le_bytes([buf[1], buf[2], buf[3], buf[4]]) as usize;
129    if body_len > MAX_FRAME_BYTES {
130        return Err(FramingError::FrameTooLarge {
131            body_length: body_len,
132            cap: MAX_FRAME_BYTES,
133        });
134    }
135    let total = FRAME_HEADER_BYTES + body_len;
136    if buf.len() < total {
137        return Ok(None);
138    }
139    let frame = Frame::decode(&buf[FRAME_HEADER_BYTES..total])?;
140    Ok(Some(DecodedFramed {
141        frame,
142        consumed: total,
143    }))
144}
145
146/// Read one v1 frame body with the default 16 MiB cap.
147pub fn read_frame<R: Read>(reader: &mut R) -> Result<Vec<u8>, FramingError> {
148    read_frame_with_cap(reader, MAX_FRAME_BYTES)
149}
150
151/// Read one v1 frame body, rejecting an announced size greater than
152/// `max_bytes` before allocating.
153pub fn read_frame_with_cap<R: Read>(
154    reader: &mut R,
155    max_bytes: usize,
156) -> Result<Vec<u8>, FramingError> {
157    let mut version_buf = [0_u8; 1];
158    read_exact_or_eof(reader, &mut version_buf, "framing byte")?;
159    let version = version_buf[0];
160    if version != ENVELOPE_VERSION {
161        return Err(FramingError::UnsupportedFramingVersion {
162            got: version,
163            expected: ENVELOPE_VERSION,
164        });
165    }
166
167    let mut len_buf = [0_u8; 4];
168    read_exact_or_eof(reader, &mut len_buf, "body length header")?;
169    let body_length = u32::from_le_bytes(len_buf) as usize;
170    if body_length > max_bytes {
171        return Err(FramingError::FrameTooLarge {
172            body_length,
173            cap: max_bytes,
174        });
175    }
176
177    let mut body = vec![0_u8; body_length];
178    if body_length != 0 {
179        read_exact_or_eof(reader, &mut body, "frame body")?;
180    }
181    Ok(body)
182}
183
184/// Write one v1 frame body, flush the stream, and return the byte count.
185pub fn write_frame<W: Write>(writer: &mut W, body: &[u8]) -> Result<usize, FramingError> {
186    if body.len() > MAX_FRAME_BYTES {
187        return Err(FramingError::FrameTooLarge {
188            body_length: body.len(),
189            cap: MAX_FRAME_BYTES,
190        });
191    }
192
193    let body_len = body.len() as u32;
194    let header = [
195        ENVELOPE_VERSION,
196        (body_len & 0xFF) as u8,
197        ((body_len >> 8) & 0xFF) as u8,
198        ((body_len >> 16) & 0xFF) as u8,
199        ((body_len >> 24) & 0xFF) as u8,
200    ];
201    writer.write_all(&header)?;
202    if !body.is_empty() {
203        writer.write_all(body)?;
204    }
205    writer.flush()?;
206    Ok(header.len() + body.len())
207}
208
209fn read_exact_or_eof<R: Read>(
210    reader: &mut R,
211    buf: &mut [u8],
212    context: &'static str,
213) -> Result<(), FramingError> {
214    match reader.read_exact(buf) {
215        Ok(()) => Ok(()),
216        Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => {
217            Err(FramingError::UnexpectedEof { context })
218        }
219        Err(error) => Err(FramingError::Io(error)),
220    }
221}
222
223/// Frozen v1 payload-protocol registry and consumer registration checks.
224pub mod registry {
225    /// Negotiated v1 broker protocol version carried in `Frame` protobuf
226    /// fields. It is distinct from the outer [`super::FRAMING_VERSION_V1`].
227    pub const PROTOCOL_VERSION: u32 = 1;
228
229    /// Control-plane Hello/HelloReply protocol.
230    pub const CONTROL_PAYLOAD_PROTOCOL: u32 = 0x00;
231    /// Admin request/reply protocol.
232    pub const ADMIN_PAYLOAD_PROTOCOL: u32 = 0xAD01;
233    /// Same-endpoint daemon-identity nonce probe protocol.
234    pub const BACKEND_HANDLE_PROBE_PAYLOAD_PROTOCOL: u32 = 0xB232;
235    /// Broker/backend handoff offer and acknowledgment protocol.
236    pub const HANDOFF_PAYLOAD_PROTOCOL: u32 = 0xD0FF;
237    /// SESSION proxy data-plane protocol.
238    pub const SESSION_PAYLOAD_PROTOCOL: u32 = 0x5350;
239
240    /// Inclusive lower bound for centrally registered consumer protocols.
241    pub const CONSUMER_PAYLOAD_PROTOCOL_MIN: u32 = 0x7000;
242    /// Inclusive upper bound for centrally registered consumer protocols.
243    pub const CONSUMER_PAYLOAD_PROTOCOL_MAX: u32 = 0x7EFF;
244    /// Inclusive lower bound for private-use protocols.
245    pub const PRIVATE_USE_PAYLOAD_PROTOCOL_MIN: u32 = 0xF000;
246    /// Inclusive upper bound for private-use protocols.
247    pub const PRIVATE_USE_PAYLOAD_PROTOCOL_MAX: u32 = 0xFFFF;
248
249    /// Registered consumer value for zccache's opaque Frame v1 lane.
250    pub const ZCCACHE_PAYLOAD_PROTOCOL: u32 = 0x7A63;
251    /// Registered consumer value for clud's opaque Frame v1 lane.
252    pub const CLUD_PAYLOAD_PROTOCOL: u32 = 0x7C4C;
253    /// Registered consumer value for fbuild's opaque Frame v1 lane.
254    pub const FBUILD_PAYLOAD_PROTOCOL: u32 = 0x7EB1;
255
256    /// First-party protocols that consumer values must not overlap.
257    pub const FIRST_PARTY_PAYLOAD_PROTOCOLS: [u32; 4] = [
258        CONTROL_PAYLOAD_PROTOCOL,
259        ADMIN_PAYLOAD_PROTOCOL,
260        BACKEND_HANDLE_PROBE_PAYLOAD_PROTOCOL,
261        HANDOFF_PAYLOAD_PROTOCOL,
262    ];
263
264    /// Return whether `id` belongs to a first-party subsystem.
265    pub const fn is_first_party(id: u32) -> bool {
266        let mut index = 0;
267        while index < FIRST_PARTY_PAYLOAD_PROTOCOLS.len() {
268            if FIRST_PARTY_PAYLOAD_PROTOCOLS[index] == id {
269                return true;
270            }
271            index += 1;
272        }
273        false
274    }
275
276    /// Return whether `id` is in the registered-consumer range.
277    pub const fn is_registered_consumer_id(id: u32) -> bool {
278        id >= CONSUMER_PAYLOAD_PROTOCOL_MIN && id <= CONSUMER_PAYLOAD_PROTOCOL_MAX
279    }
280
281    /// Return whether `id` is in the private-use range.
282    pub const fn is_private_use_id(id: u32) -> bool {
283        id >= PRIVATE_USE_PAYLOAD_PROTOCOL_MIN && id <= PRIVATE_USE_PAYLOAD_PROTOCOL_MAX
284    }
285}
286
287pub use registry::{
288    ADMIN_PAYLOAD_PROTOCOL, BACKEND_HANDLE_PROBE_PAYLOAD_PROTOCOL, CLUD_PAYLOAD_PROTOCOL,
289    CONTROL_PAYLOAD_PROTOCOL, FBUILD_PAYLOAD_PROTOCOL, HANDOFF_PAYLOAD_PROTOCOL, PROTOCOL_VERSION,
290    SESSION_PAYLOAD_PROTOCOL, ZCCACHE_PAYLOAD_PROTOCOL,
291};
292
293/// Define a consumer payload-protocol constant with compile-time range and
294/// first-party-collision checks.
295///
296/// The authoritative allocations remain in [`registry`]; this macro verifies
297/// only that a consumer's pinned literal belongs to an allowed allocation
298/// range and cannot collide with a first-party protocol.
299#[macro_export]
300macro_rules! register_payload_protocol {
301    ($(#[$meta:meta])* $vis:vis const $name:ident: u32 = $value:expr;) => {
302        $(#[$meta])*
303        $vis const $name: u32 = $value;
304
305        const _: () = {
306            assert!(
307                !$crate::frame_v1::registry::is_first_party($name),
308                concat!(
309                    stringify!($name),
310                    " collides with a first-party running-process payload protocol",
311                ),
312            );
313            assert!(
314                $crate::frame_v1::registry::is_registered_consumer_id($name)
315                    || $crate::frame_v1::registry::is_private_use_id($name),
316                concat!(
317                    stringify!($name),
318                    " must lie in the registered-consumer range (0x7000..=0x7EFF) ",
319                    "or the private-use range (0xF000..=0xFFFF)",
320                ),
321            );
322        };
323    };
324}
325
326#[cfg(test)]
327mod tests {
328    use super::registry::{
329        is_first_party, is_private_use_id, is_registered_consumer_id, ADMIN_PAYLOAD_PROTOCOL,
330        BACKEND_HANDLE_PROBE_PAYLOAD_PROTOCOL, CLUD_PAYLOAD_PROTOCOL,
331        CONSUMER_PAYLOAD_PROTOCOL_MAX, CONSUMER_PAYLOAD_PROTOCOL_MIN, CONTROL_PAYLOAD_PROTOCOL,
332        FBUILD_PAYLOAD_PROTOCOL, HANDOFF_PAYLOAD_PROTOCOL, PRIVATE_USE_PAYLOAD_PROTOCOL_MAX,
333        PRIVATE_USE_PAYLOAD_PROTOCOL_MIN, PROTOCOL_VERSION, ZCCACHE_PAYLOAD_PROTOCOL,
334    };
335    use super::{
336        encode_framed, try_decode_framed, Frame, FrameKind, FramingError, PayloadEncoding,
337        ENVELOPE_VERSION, MAX_FRAME_BYTES,
338    };
339
340    crate::register_payload_protocol! {
341        /// Registered-consumer-range example checked at compile time.
342        const MACRO_CONSUMER_RANGE_EXAMPLE: u32 = 0x7001;
343    }
344    crate::register_payload_protocol! {
345        /// Private-use-range example checked at compile time.
346        const MACRO_PRIVATE_RANGE_EXAMPLE: u32 = 0xF00D;
347    }
348
349    #[test]
350    fn payload_protocol_ids_are_pairwise_distinct() {
351        let registered: [(u32, &str); 4] = [
352            (CONTROL_PAYLOAD_PROTOCOL, "CONTROL_PAYLOAD_PROTOCOL"),
353            (ADMIN_PAYLOAD_PROTOCOL, "ADMIN_PAYLOAD_PROTOCOL"),
354            (
355                BACKEND_HANDLE_PROBE_PAYLOAD_PROTOCOL,
356                "BACKEND_HANDLE_PROBE_PAYLOAD_PROTOCOL",
357            ),
358            (HANDOFF_PAYLOAD_PROTOCOL, "HANDOFF_PAYLOAD_PROTOCOL"),
359        ];
360        for (left_index, (left_id, left_name)) in registered.iter().enumerate() {
361            for (right_id, right_name) in &registered[left_index + 1..] {
362                assert_ne!(
363                    left_id, right_id,
364                    "{left_name} and {right_name} share payload-protocol id {left_id:#06X}"
365                );
366            }
367        }
368    }
369
370    #[test]
371    fn frozen_v1_wire_values() {
372        assert_eq!(PROTOCOL_VERSION, 1);
373        assert_eq!(CONTROL_PAYLOAD_PROTOCOL, 0x00);
374        assert_eq!(ADMIN_PAYLOAD_PROTOCOL, 0xAD01);
375        assert_eq!(BACKEND_HANDLE_PROBE_PAYLOAD_PROTOCOL, 0xB232);
376        assert_eq!(HANDOFF_PAYLOAD_PROTOCOL, 0xD0FF);
377        assert_eq!(u32::from(super::FRAMING_VERSION_V1), 1);
378    }
379
380    #[test]
381    fn frozen_consumer_registry_values() {
382        assert_eq!(CONSUMER_PAYLOAD_PROTOCOL_MIN, 0x7000);
383        assert_eq!(CONSUMER_PAYLOAD_PROTOCOL_MAX, 0x7EFF);
384        assert_eq!(PRIVATE_USE_PAYLOAD_PROTOCOL_MIN, 0xF000);
385        assert_eq!(PRIVATE_USE_PAYLOAD_PROTOCOL_MAX, 0xFFFF);
386        assert_eq!(ZCCACHE_PAYLOAD_PROTOCOL, 0x7A63);
387        assert_eq!(CLUD_PAYLOAD_PROTOCOL, 0x7C4C);
388        assert_eq!(FBUILD_PAYLOAD_PROTOCOL, 0x7EB1);
389        assert!(is_first_party(BACKEND_HANDLE_PROBE_PAYLOAD_PROTOCOL));
390        assert!(is_registered_consumer_id(ZCCACHE_PAYLOAD_PROTOCOL));
391        assert!(is_private_use_id(0xF412));
392    }
393
394    #[test]
395    fn register_macro_defines_usable_constants() {
396        assert_eq!(MACRO_CONSUMER_RANGE_EXAMPLE, 0x7001);
397        assert_eq!(MACRO_PRIVATE_RANGE_EXAMPLE, 0xF00D);
398    }
399
400    #[test]
401    fn root_client_reexports_keep_frame_extensions_and_codecs() {
402        let frame = Frame::request(0x7A63, b"ping".to_vec()).with_request_id(42);
403        assert_eq!(frame.kind, FrameKind::Request as i32);
404        assert_eq!(frame.payload_encoding, PayloadEncoding::None as i32);
405
406        let wire = encode_framed(&frame).expect("encode");
407        let decoded = try_decode_framed(&wire)
408            .expect("decode")
409            .expect("complete frame");
410        assert_eq!(decoded.frame, frame);
411        assert_eq!(decoded.consumed, wire.len());
412    }
413
414    #[test]
415    fn try_decode_framed_waits_for_complete_frames() {
416        let wire = encode_framed(&Frame::request(0x7001, b"abc".to_vec())).expect("encode");
417
418        assert!(
419            try_decode_framed(&[])
420                .expect("empty buffer is not an error")
421                .is_none(),
422            "an empty buffer must ask for more bytes, not decode"
423        );
424        for cut in 1..wire.len() {
425            assert!(
426                try_decode_framed(&wire[..cut])
427                    .expect("a prefix is not an error")
428                    .is_none(),
429                "partial frame of {cut} of {} bytes must not decode",
430                wire.len()
431            );
432        }
433
434        let mut two = wire.clone();
435        two.extend_from_slice(&wire);
436        let first = try_decode_framed(&two)
437            .expect("decode")
438            .expect("first frame is complete");
439        assert_eq!(first.consumed, wire.len());
440    }
441
442    #[test]
443    fn try_decode_framed_rejects_foreign_version_and_oversize() {
444        let foreign = ENVELOPE_VERSION.wrapping_add(1);
445        assert!(matches!(
446            try_decode_framed(&[foreign, 0, 0, 0, 0]),
447            Err(FramingError::UnsupportedFramingVersion { got, expected })
448                if got == foreign && expected == ENVELOPE_VERSION
449        ));
450
451        let mut oversize = vec![ENVELOPE_VERSION];
452        let claimed = u32::try_from(MAX_FRAME_BYTES).expect("cap fits u32") + 1;
453        oversize.extend_from_slice(&claimed.to_le_bytes());
454        assert!(matches!(
455            try_decode_framed(&oversize),
456            Err(FramingError::FrameTooLarge { body_length, cap })
457                if body_length == claimed as usize && cap == MAX_FRAME_BYTES
458        ));
459    }
460
461    #[cfg(feature = "client")]
462    #[test]
463    fn root_client_reexports_keep_endpoint_extensions_and_errors() {
464        use crate::broker::protocol::{Endpoint, EndpointNameError};
465
466        assert!(Endpoint::unix_socket("svc", "/tmp/svc.sock").is_ok());
467        assert_eq!(
468            Endpoint::windows_pipe("svc", r"\\.\pipe\svc-pipe"),
469            Err(EndpointNameError::PrefixedPipeName {
470                got: r"\\.\pipe\svc-pipe".to_owned(),
471            })
472        );
473    }
474}
running_process/frame_v1.rs

running_process/
frame_v1.rs
