ozzy_proto/
envelope.rs

1//! Bounded Ozzy framing shared by every command family.
2
3use crate::{LinkSessionId, NodeId, RequestId, layout::Fixed};
4
5/// Current wire version. This pre-release protocol has no compatibility decoder.
6pub const VERSION: u8 = 1;
7
8/// Exact size of the native envelope, excluding OMQ routing metadata.
9pub const ENVELOPE_BYTES: usize = 64;
10
11macro_rules! opcodes {
12    ($($variant:ident = $value:literal => $name:literal),+ $(,)?) => {
13        /// Command allocations in `doc/PROTOCOL.md`, not capability promises.
14        #[derive(Debug, Clone, Copy, PartialEq, Eq)]
15        #[repr(u8)]
16        pub enum Opcode {
17            $(#[doc = $name] $variant = $value,)+
18        }
19
20        impl Opcode {
21            fn decode(value: u8) -> Result<Self, EnvelopeError> {
22                match value {
23                    $($value => Ok(Self::$variant),)+
24                    _ => Err(EnvelopeError::Opcode(value)),
25                }
26            }
27        }
28    };
29}
30
31opcodes! {
32    Hello = 0x01 => "`HELLO`: initiate link negotiation.",
33    Welcome = 0x02 => "`WELCOME`: establish the negotiated link session.",
34    Error = 0x03 => "`ERROR`: bounded protocol error.",
35    Append = 0x10 => "`APPEND`: submit producer records.",
36    Appended = 0x11 => "`APPENDED`: confirm the configured append boundary.",
37    LookupAppend = 0x12 => "`LOOKUP_APPEND`: resolve an uncertain append.",
38    AppendState = 0x13 => "`APPEND_STATE`: report an append outcome.",
39    OpenProducer = 0x14 => "`OPEN_PRODUCER`: open or fence a producer session.",
40    ProducerOpened = 0x15 => "`PRODUCER_OPENED`: confirm producer authority.",
41    AckResults = 0x16 => "`ACK_RESULTS`: advance producer retry-result floors.",
42    ResultsAcked = 0x17 => "`RESULTS_ACKED`: confirm result-floor commit.",
43    Subscribe = 0x20 => "`SUBSCRIBE`: request record delivery.",
44    Subscribed = 0x21 => "`SUBSCRIBED`: describe assigned delivery ranges.",
45    Records = 0x22 => "`RECORDS`: deliver committed records.",
46    Ack = 0x23 => "`ACK`: report consumer receipt or processing observations.",
47    ProgressCommit = 0x24 => "`PROGRESS_COMMIT`: request durable consumer progress.",
48    ProgressCommitted = 0x25 => "`PROGRESS_COMMITTED`: confirm committed progress.",
49    Unsubscribe = 0x27 => "`UNSUBSCRIBE`: cancel a subscription generation.",
50    Unsubscribed = 0x28 => "`UNSUBSCRIBED`: confirm subscription cancellation.",
51    ReplicaOpen = 0x30 => "`REPLICA_OPEN`: open a configured replica group.",
52    ReplicaState = 0x31 => "`REPLICA_STATE`: report replica state and volatile receipt.",
53    Prepare = 0x32 => "`PREPARE`: replicate canonical operations.",
54    PrepareOk = 0x33 => "`PREPARE_OK`: report contiguous replication evidence.",
55    Commit = 0x34 => "`COMMIT`: announce the quorum-committed prefix.",
56    StartViewChange = 0x35 => "`START_VIEW_CHANGE`: initiate a higher view.",
57    DoViewChange = 0x36 => "`DO_VIEW_CHANGE`: report immutable selected history.",
58    StartView = 0x37 => "`START_VIEW`: announce installed view and history.",
59    Recovery = 0x38 => "`RECOVERY`: request nonvoting recovery evidence.",
60    RecoveryState = 0x39 => "`RECOVERY_STATE`: reply to a fresh recovery nonce.",
61    FetchOps = 0x3a => "`FETCH_OPS`: request bounded canonical history.",
62    Ops = 0x3b => "`OPS`: transfer canonical operation bodies.",
63    SnapshotBegin = 0x3c => "`SNAPSHOT_BEGIN`: describe a checkpoint transfer.",
64    SnapshotChunk = 0x3d => "`SNAPSHOT_CHUNK`: transfer a bounded checkpoint chunk.",
65    SnapshotEnd = 0x3e => "`SNAPSHOT_END`: finish a checkpoint transfer.",
66    SnapshotInstalled = 0x3f => "`SNAPSHOT_INSTALLED`: confirm checkpoint installation.",
67    StateSnapshotRequest = 0x40 => "`STATE_SNAPSHOT_REQUEST`: request directory state.",
68    StateSnapshot = 0x41 => "`STATE_SNAPSHOT`: transfer directory state.",
69    StateUpdate = 0x42 => "`STATE_UPDATE`: advance directory state.",
70    StateResync = 0x43 => "`STATE_RESYNC`: request directory resynchronization.",
71    ExitView = 0x50 => "`EXIT_VIEW`: request quorum-supported departure from the current view.",
72    PrepareFlow = 0x51 => "`PREPARE_FLOW`: replicate canonical operations in a fenced receive epoch.",
73    PreparePub = 0x52 => "`PREPARE_PUB`: publish canonical operations to followers.",
74    RecordsPub = 0x53 => "`RECORDS_PUB`: publish confirmed records to every live reader.",
75    ReplicaReceipt = 0x54 => "`REPLICA_RECEIPT`: compact same-channel volatile receipt.",
76    HistoryRetired = 0x55 => "`HISTORY_RETIRED`: require nonvoting checkpoint recovery below retained history.",
77    GroupJoin = 0x60 => "`GROUP_JOIN`: join, heartbeat, or leave a consumer group on its coordinator.",
78    GroupJoined = 0x61 => "`GROUP_JOINED`: describe the committed membership view of one instance.",
79    GroupAssign = 0x62 => "`GROUP_ASSIGN`: relay a committed plan revision to a source partition.",
80    GroupAssigned = 0x63 => "`GROUP_ASSIGNED`: confirm source activation coordinates.",
81    GroupInspect = 0x64 => "`GROUP_INSPECT`: read a group's committed coordinator image.",
82    GroupInspected = 0x65 => "`GROUP_INSPECTED`: describe a group's committed coordinator image.",
83    GroupDelete = 0x66 => "`GROUP_DELETE`: retire one group incarnation on its coordinator.",
84    GroupDeleted = 0x67 => "`GROUP_DELETED`: confirm a group incarnation is retired.",
85    Nack = 0x7f => "`NACK`: report a typed negative outcome.",
86}
87
88/// Directional receive bounds checked before command decoding.
89#[derive(Debug, Clone, Copy, PartialEq, Eq)]
90pub struct EnvelopeLimits {
91    /// Maximum bytes in the metadata frame.
92    pub max_metadata_bytes: usize,
93    /// Maximum bytes in the packed payload frame.
94    pub max_payload_bytes: usize,
95}
96
97impl Default for EnvelopeLimits {
98    fn default() -> Self {
99        Self {
100            max_metadata_bytes: 64 * 1024,
101            max_payload_bytes: 1024 * 1024,
102        }
103    }
104}
105
106/// Routing-independent command identity and link fence.
107#[derive(Debug, Clone, Copy, PartialEq, Eq)]
108pub struct Envelope {
109    /// Allocated command number, still requiring capability and schema checks.
110    pub opcode: Opcode,
111    /// Whether this command responds to the named nonzero request.
112    pub response: bool,
113    /// Correlation identity; absent only for uncorrelated notifications.
114    pub request_id: Option<RequestId>,
115    /// Claimed sender, which the authenticated link must independently verify.
116    pub sender: NodeId,
117    /// Negotiated session; absent for HELLO and the receiver-fenced publications
118    /// `PREPARE_PUB` and `RECORDS_PUB`.
119    pub session: Option<LinkSessionId>,
120}
121
122impl Envelope {
123    /// Validate common identities and exact frame sizes without encoding a
124    /// throwaway header. Command-specific metadata and payload remain unchecked.
125    pub fn validate_frames(
126        self,
127        metadata_bytes: usize,
128        payload_bytes: usize,
129        limits: EnvelopeLimits,
130    ) -> Result<(), EnvelopeError> {
131        validate_lengths(metadata_bytes, payload_bytes, limits)?;
132        self.validate(payload_bytes)?;
133        u32::try_from(metadata_bytes).map_err(|_| EnvelopeError::Length)?;
134        u32::try_from(payload_bytes).map_err(|_| EnvelopeError::Length)?;
135        Ok(())
136    }
137
138    /// Encode into a fixed stack-sized header; no frame or payload allocation.
139    ///
140    /// Lengths must describe the exact frames submitted with this header. This
141    /// does not encode or validate any command-specific metadata or payload.
142    pub fn encode_header(
143        self,
144        metadata_bytes: usize,
145        payload_bytes: usize,
146        limits: EnvelopeLimits,
147    ) -> Result<[u8; ENVELOPE_BYTES], EnvelopeError> {
148        self.validate_frames(metadata_bytes, payload_bytes, limits)?;
149        let mut bytes = [0; ENVELOPE_BYTES];
150        bytes[..4].copy_from_slice(b"OZY\0");
151        bytes[4] = VERSION;
152        bytes[5] = self.opcode as u8;
153        bytes[6..8].copy_from_slice(&u16::from(self.response).to_be_bytes());
154        if let Some(request) = self.request_id {
155            bytes[8..24].copy_from_slice(request.as_bytes());
156        }
157        bytes[24..40].copy_from_slice(self.sender.as_bytes());
158        if let Some(session) = self.session {
159            bytes[40..56].copy_from_slice(session.as_bytes());
160        }
161        bytes[56..60].copy_from_slice(&(metadata_bytes as u32).to_be_bytes());
162        bytes[60..64].copy_from_slice(&(payload_bytes as u32).to_be_bytes());
163        Ok(bytes)
164    }
165
166    fn validate(self, payload_bytes: usize) -> Result<(), EnvelopeError> {
167        if self.sender.as_bytes() == &[0; 16]
168            || self.request_id.is_some_and(|id| id.as_bytes() == &[0; 16])
169            || self.session.is_some_and(|id| id.as_bytes() == &[0; 16])
170        {
171            return Err(EnvelopeError::ZeroIdentity);
172        }
173        if self.response && self.request_id.is_none() {
174            return Err(EnvelopeError::Correlation);
175        }
176        if self.opcode == Opcode::Hello {
177            if self.session.is_some()
178                || self.response
179                || self.request_id.is_none()
180                || payload_bytes != 0
181            {
182                return Err(EnvelopeError::Handshake);
183            }
184        } else if matches!(self.opcode, Opcode::PreparePub | Opcode::RecordsPub) {
185            if self.session.is_some() || self.response || self.request_id.is_some() {
186                return Err(EnvelopeError::Session);
187            }
188        } else if self.session.is_none() {
189            return Err(EnvelopeError::Session);
190        } else if self.opcode == Opcode::Welcome && (!self.response || self.request_id.is_none()) {
191            return Err(EnvelopeError::Handshake);
192        }
193        Ok(())
194    }
195}
196
197/// Structurally valid frames borrowing the caller's packet storage.
198///
199/// Metadata and payload remain untrusted command bytes. Session, capability,
200/// schema, and authority validation must precede any application or quorum effect.
201#[derive(Debug, Clone, Copy, PartialEq, Eq)]
202pub struct Packet<'a> {
203    /// Decoded command and link identity.
204    pub envelope: Envelope,
205    /// Exact bounded metadata frame.
206    pub metadata: &'a [u8],
207    /// Exact bounded packed payload frame.
208    pub payload: &'a [u8],
209}
210
211/// The metadata and payload frames of one data command: caller-reserved
212/// buffers to encode into, or received frames that decoded records share.
213#[derive(Debug, Clone, Copy, PartialEq, Eq)]
214pub struct Frames<T> {
215    /// Metadata frame.
216    pub metadata: T,
217    /// Packed payload frame.
218    pub payload: T,
219}
220
221/// Decode exactly three Ozzy frames, after the transport removes routing frames.
222///
223/// Reject unsupported versions and old frame shapes. All header reads are
224/// allocation-free; no length from the peer controls an allocation here.
225pub fn decode_packet<'a>(
226    frames: &[&'a [u8]],
227    limits: EnvelopeLimits,
228) -> Result<Packet<'a>, EnvelopeError> {
229    let [header, metadata, payload] = frames else {
230        return Err(EnvelopeError::FrameCount);
231    };
232    let header: &[u8; ENVELOPE_BYTES] = (*header)
233        .try_into()
234        .map_err(|_| EnvelopeError::HeaderLength)?;
235    validate_lengths(metadata.len(), payload.len(), limits)?;
236    if header.array::<0, 4>() != *b"OZY\0" {
237        return Err(EnvelopeError::Magic);
238    }
239    if header.u8_at::<4>() != VERSION {
240        return Err(EnvelopeError::Major(header.u8_at::<4>()));
241    }
242    let opcode = Opcode::decode(header.u8_at::<5>())?;
243    let flags = header.u16_at::<6>();
244    if flags & !1 != 0 {
245        return Err(EnvelopeError::Flags);
246    }
247    let metadata_bytes = header.u32_at::<56>();
248    let payload_bytes = header.u32_at::<60>();
249    if usize::try_from(metadata_bytes) != Ok(metadata.len())
250        || usize::try_from(payload_bytes) != Ok(payload.len())
251    {
252        return Err(EnvelopeError::Length);
253    }
254    let request = header.array::<8, 16>();
255    let session = header.array::<40, 16>();
256    let envelope = Envelope {
257        opcode,
258        response: flags == 1,
259        request_id: (request != [0; 16]).then(|| RequestId::from_bytes(request)),
260        sender: NodeId::from_bytes(header.array::<24, 16>()),
261        session: (session != [0; 16]).then(|| LinkSessionId::from_bytes(session)),
262    };
263    envelope.validate(payload.len())?;
264    Ok(Packet {
265        envelope,
266        metadata,
267        payload,
268    })
269}
270
271fn validate_lengths(
272    metadata: usize,
273    payload: usize,
274    limits: EnvelopeLimits,
275) -> Result<(), EnvelopeError> {
276    if metadata > limits.max_metadata_bytes || payload > limits.max_payload_bytes {
277        return Err(EnvelopeError::Limit);
278    }
279    Ok(())
280}
281
282/// Structural native envelope rejection, without interpreting command semantics.
283#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
284pub enum EnvelopeError {
285    /// Exactly three frames are required, including present empty frames.
286    #[error("native packet requires exactly three Ozzy frames")]
287    FrameCount,
288    /// Header is not exactly 64 bytes.
289    #[error("invalid native envelope length")]
290    HeaderLength,
291    /// Magic does not identify Ozzy.
292    #[error("invalid Ozzy magic")]
293    Magic,
294    /// No implicit legacy fallback or future-major interpretation.
295    #[error("unsupported Ozzy major {0}")]
296    Major(u8),
297    /// Unallocated command number.
298    #[error("unknown Ozzy opcode {0}")]
299    Opcode(u8),
300    /// Flags other than the response bit are reserved.
301    #[error("unsupported Ozzy envelope flags")]
302    Flags,
303    /// Claimed sizes differ from actual frames or exceed the wire integer width.
304    #[error("Ozzy frame length mismatch or overflow")]
305    Length,
306    /// Frame exceeds the directional receiver's configured bound.
307    #[error("Ozzy frame exceeds receive limit")]
308    Limit,
309    /// Mandatory sender or explicitly supplied optional ID is zero.
310    #[error("zero Ozzy envelope identity")]
311    ZeroIdentity,
312    /// A response cannot be uncorrelated.
313    #[error("response requires a nonzero request ID")]
314    Correlation,
315    /// Only HELLO can omit the link session.
316    #[error("command requires a negotiated link session")]
317    Session,
318    /// Invalid HELLO/WELCOME direction, correlation, session, or payload.
319    #[error("invalid handshake envelope")]
320    Handshake,
321}
ozzy_proto/envelope.rs

ozzy_proto/
envelope.rs
