mongreldb_log/
envelope.rs

1//! Versioned durable command envelope (spec section 9.3, FND-003).
2//!
3//! Every new cluster command is persisted inside a [`CommandEnvelope`], never
4//! as an unversioned `bincode` enum. The canonical encoding is:
5//!
6//! ```text
7//! format_version : u32 LE
8//! command_id     : [u8; 16]
9//! command_type   : u32 LE
10//! payload_len    : u32 LE
11//! payload        : [u8; payload_len]
12//! payload_sha256 : [u8; 32]
13//! ```
14//!
15//! The checksum covers the format version, the command type, the payload
16//! length, and the payload. Decoding fails closed: unknown format versions,
17//! oversized payloads, truncated frames, trailing bytes, and checksum
18//! mismatches are all errors.
19//!
20//! Encoding is deterministic: [`CommandEnvelope::encode`] is a pure function
21//! of the envelope fields, so equal envelopes produce byte-identical frames
22//! and `decode(encode(e))` is the identity.
23//!
24//! Payload evolution rules (spec section 9.3):
25//!
26//! - `command_type` discriminants and payload field numbers are never reused.
27//! - Unknown optional fields are preserved or ignored safely at the payload
28//!   layer; the envelope itself never interprets payload bytes.
29//! - Unknown required command versions fail closed with
30//!   [`EnvelopeError::UnsupportedVersion`].
31//! - Payloads use a schema-evolution-safe encoding (Protobuf per
32//!   `docs/architecture/adr/0005`, network protocol and serialization).
33
34use core::fmt;
35use sha2::{Digest, Sha256};
36
37/// The envelope format version this build writes.
38pub const COMMAND_ENVELOPE_FORMAT_VERSION: u32 = 1;
39/// The oldest envelope format version this build accepts.
40pub const MIN_SUPPORTED_FORMAT_VERSION: u32 = 1;
41/// Upper bound on a single command payload.
42pub const MAX_COMMAND_PAYLOAD_BYTES: usize = 64 * 1024 * 1024;
43/// Encoded length of the fixed header preceding the payload.
44pub const HEADER_LEN: usize = 4 + 16 + 4 + 4;
45/// Encoded length of the trailing checksum.
46pub const CHECKSUM_LEN: usize = 32;
47
48/// Errors produced while verifying or decoding a [`CommandEnvelope`].
49#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
50pub enum EnvelopeError {
51    /// The format version is outside the supported range.
52    #[error("unsupported command-envelope format version {found} (supported {min}..={max})")]
53    UnsupportedVersion {
54        /// Version found in the envelope.
55        found: u32,
56        /// Oldest version this build accepts.
57        min: u32,
58        /// Newest version this build accepts.
59        max: u32,
60    },
61    /// The stored checksum does not match the recomputed one.
62    #[error("command-envelope checksum mismatch")]
63    ChecksumMismatch,
64    /// The byte slice ended before a complete envelope was read.
65    #[error("command envelope truncated: expected at least {expected} bytes, got {actual}")]
66    Truncated {
67        /// Minimum number of bytes required.
68        expected: usize,
69        /// Number of bytes actually present.
70        actual: usize,
71    },
72    /// Extra bytes followed an otherwise complete envelope.
73    #[error("command envelope has {0} trailing bytes")]
74    TrailingBytes(usize),
75    /// The payload exceeds [`MAX_COMMAND_PAYLOAD_BYTES`].
76    #[error("command payload too large: {0} bytes")]
77    PayloadTooLarge(usize),
78}
79
80/// The versioned, checksummed form of every persisted command.
81#[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
82pub struct CommandEnvelope {
83    /// Envelope format version; see [`COMMAND_ENVELOPE_FORMAT_VERSION`].
84    pub format_version: u32,
85    /// Unique identifier of this command (for idempotent apply).
86    pub command_id: [u8; 16],
87    /// Discriminant of the payload command; numbers are never reused.
88    pub command_type: u32,
89    /// Schema-evolution-safe encoded command payload. The envelope never
90    /// interprets these bytes; unknown optional payload fields are preserved
91    /// or ignored safely by the payload decoder.
92    pub payload: Vec<u8>,
93    /// SHA-256 over version, type, length, and payload.
94    pub payload_sha256: [u8; 32],
95}
96
97impl CommandEnvelope {
98    /// Builds an envelope at the current format version, computing the checksum.
99    pub fn new(command_type: u32, command_id: [u8; 16], payload: Vec<u8>) -> Self {
100        let payload_sha256 =
101            Self::checksum(COMMAND_ENVELOPE_FORMAT_VERSION, command_type, &payload);
102        Self {
103            format_version: COMMAND_ENVELOPE_FORMAT_VERSION,
104            command_id,
105            command_type,
106            payload,
107            payload_sha256,
108        }
109    }
110
111    /// Computes the checksum covering type, version, length, and payload.
112    pub fn checksum(format_version: u32, command_type: u32, payload: &[u8]) -> [u8; 32] {
113        let mut hasher = Sha256::new();
114        hasher.update(format_version.to_le_bytes());
115        hasher.update(command_type.to_le_bytes());
116        hasher.update((payload.len() as u64).to_le_bytes());
117        hasher.update(payload);
118        hasher.finalize().into()
119    }
120
121    /// Fails closed unless the version is supported and the checksum matches.
122    pub fn verify(&self) -> Result<(), EnvelopeError> {
123        if !(MIN_SUPPORTED_FORMAT_VERSION..=COMMAND_ENVELOPE_FORMAT_VERSION)
124            .contains(&self.format_version)
125        {
126            return Err(EnvelopeError::UnsupportedVersion {
127                found: self.format_version,
128                min: MIN_SUPPORTED_FORMAT_VERSION,
129                max: COMMAND_ENVELOPE_FORMAT_VERSION,
130            });
131        }
132        if self.payload.len() > MAX_COMMAND_PAYLOAD_BYTES {
133            return Err(EnvelopeError::PayloadTooLarge(self.payload.len()));
134        }
135        let expected = Self::checksum(self.format_version, self.command_type, &self.payload);
136        if expected != self.payload_sha256 {
137            return Err(EnvelopeError::ChecksumMismatch);
138        }
139        Ok(())
140    }
141
142    /// Serializes to the canonical deterministic encoding: equal envelopes
143    /// always produce byte-identical output.
144    pub fn encode(&self) -> Vec<u8> {
145        debug_assert!(
146            self.payload.len() <= MAX_COMMAND_PAYLOAD_BYTES,
147            "payload exceeds MAX_COMMAND_PAYLOAD_BYTES"
148        );
149        let mut out = Vec::with_capacity(HEADER_LEN + self.payload.len() + CHECKSUM_LEN);
150        out.extend_from_slice(&self.format_version.to_le_bytes());
151        out.extend_from_slice(&self.command_id);
152        out.extend_from_slice(&self.command_type.to_le_bytes());
153        out.extend_from_slice(&(self.payload.len() as u32).to_le_bytes());
154        out.extend_from_slice(&self.payload);
155        out.extend_from_slice(&self.payload_sha256);
156        out
157    }
158
159    /// Parses one envelope, verifying version and checksum (fails closed).
160    pub fn decode(bytes: &[u8]) -> Result<Self, EnvelopeError> {
161        if bytes.len() < HEADER_LEN + CHECKSUM_LEN {
162            return Err(EnvelopeError::Truncated {
163                expected: HEADER_LEN + CHECKSUM_LEN,
164                actual: bytes.len(),
165            });
166        }
167        let format_version = u32::from_le_bytes(bytes[0..4].try_into().expect("slice len"));
168        let command_id: [u8; 16] = bytes[4..20].try_into().expect("slice len");
169        let command_type = u32::from_le_bytes(bytes[20..24].try_into().expect("slice len"));
170        let payload_len = u32::from_le_bytes(bytes[24..28].try_into().expect("slice len")) as usize;
171        // Fail closed on an oversize length prefix before copying any payload
172        // bytes, regardless of how long the supplied frame actually is.
173        if payload_len > MAX_COMMAND_PAYLOAD_BYTES {
174            return Err(EnvelopeError::PayloadTooLarge(payload_len));
175        }
176        let expected_total = HEADER_LEN + payload_len + CHECKSUM_LEN;
177        if bytes.len() < expected_total {
178            return Err(EnvelopeError::Truncated {
179                expected: expected_total,
180                actual: bytes.len(),
181            });
182        }
183        if bytes.len() > expected_total {
184            return Err(EnvelopeError::TrailingBytes(bytes.len() - expected_total));
185        }
186        let payload = bytes[HEADER_LEN..HEADER_LEN + payload_len].to_vec();
187        let payload_sha256: [u8; 32] = bytes[HEADER_LEN + payload_len..expected_total]
188            .try_into()
189            .expect("slice len");
190        let envelope = Self {
191            format_version,
192            command_id,
193            command_type,
194            payload,
195            payload_sha256,
196        };
197        envelope.verify()?;
198        Ok(envelope)
199    }
200}
201
202impl fmt::Debug for CommandEnvelope {
203    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
204        f.debug_struct("CommandEnvelope")
205            .field("format_version", &self.format_version)
206            .field("command_id", &self.command_id)
207            .field("command_type", &self.command_type)
208            .field("payload_len", &self.payload.len())
209            .finish_non_exhaustive()
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216
217    #[test]
218    fn round_trip() {
219        let envelope = CommandEnvelope::new(7, [42u8; 16], b"hello".to_vec());
220        let bytes = envelope.encode();
221        assert_eq!(CommandEnvelope::decode(&bytes).unwrap(), envelope);
222    }
223
224    #[test]
225    fn bit_flip_breaks_checksum() {
226        let envelope = CommandEnvelope::new(1, [1u8; 16], vec![9u8; 64]);
227        let mut bytes = envelope.encode();
228        bytes[HEADER_LEN] ^= 0x01;
229        assert_eq!(
230            CommandEnvelope::decode(&bytes),
231            Err(EnvelopeError::ChecksumMismatch)
232        );
233    }
234
235    #[test]
236    fn unknown_version_fails_closed() {
237        let mut envelope = CommandEnvelope::new(1, [1u8; 16], vec![]);
238        envelope.format_version = COMMAND_ENVELOPE_FORMAT_VERSION + 1;
239        envelope.payload_sha256 =
240            CommandEnvelope::checksum(envelope.format_version, 1, &envelope.payload);
241        assert!(matches!(
242            envelope.verify(),
243            Err(EnvelopeError::UnsupportedVersion { .. })
244        ));
245    }
246
247    #[test]
248    fn truncation_and_trailing_bytes_fail() {
249        let envelope = CommandEnvelope::new(3, [7u8; 16], vec![1, 2, 3]);
250        let bytes = envelope.encode();
251        assert!(matches!(
252            CommandEnvelope::decode(&bytes[..bytes.len() - 1]),
253            Err(EnvelopeError::Truncated { .. })
254        ));
255        let mut longer = bytes.clone();
256        longer.push(0);
257        assert_eq!(
258            CommandEnvelope::decode(&longer),
259            Err(EnvelopeError::TrailingBytes(1))
260        );
261    }
262}
mongreldb_log/envelope.rs

mongreldb_log/
envelope.rs
