salamander/format/
mod.rs

1//! Stable post-v0.1 framing and engine envelope.
2//!
3//! This module owns byte-level engine metadata while treating application
4//! payloads as opaque bytes. It intentionally has no dependency on the log,
5//! projections, agent vocabulary, or database API.
6//!
7//! `#[doc(hidden)]` at the crate root: the framing internals here are not a
8//! stable API. The engine identity and codec types users actually need are
9//! re-exported at the crate root and documented there.
10#![allow(missing_docs)]
11
12use std::collections::BTreeMap;
13use std::fmt;
14use std::sync::atomic::{AtomicU64, Ordering};
15use std::time::{SystemTime, UNIX_EPOCH};
16
17use serde::de::DeserializeOwned;
18use serde::Serialize;
19
20use crate::{Result, SalamanderError};
21
22pub const FRAME_MAGIC: [u8; 4] = *b"SDB2";
23pub const FRAME_VERSION: u16 = 2;
24pub(crate) const FRAME_HEADER_LEN: usize = 32;
25const ENVELOPE_VERSION: u16 = 1;
26const KNOWN_FLAGS: u8 = 0;
27const RESERVED_METADATA_PREFIX: &str = "salamander.";
28
29macro_rules! id_type {
30    ($name:ident) => {
31        #[derive(
32            Clone,
33            Copy,
34            PartialEq,
35            Eq,
36            PartialOrd,
37            Ord,
38            Hash,
39            Default,
40            serde::Serialize,
41            serde::Deserialize,
42        )]
43        pub struct $name([u8; 16]);
44
45        impl $name {
46            pub const ZERO: Self = Self([0; 16]);
47
48            pub const fn from_bytes(bytes: [u8; 16]) -> Self {
49                Self(bytes)
50            }
51
52            pub const fn as_bytes(&self) -> &[u8; 16] {
53                &self.0
54            }
55
56            pub const fn into_bytes(self) -> [u8; 16] {
57                self.0
58            }
59        }
60
61        impl fmt::Debug for $name {
62            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
63                for byte in self.0 {
64                    write!(f, "{byte:02x}")?;
65                }
66                Ok(())
67            }
68        }
69    };
70}
71
72id_type!(DatabaseId);
73id_type!(EventId);
74id_type!(BranchId);
75id_type!(StreamId);
76id_type!(BatchId);
77
78#[derive(
79    Debug,
80    Clone,
81    Copy,
82    PartialEq,
83    Eq,
84    PartialOrd,
85    Ord,
86    Hash,
87    Default,
88    serde::Serialize,
89    serde::Deserialize,
90)]
91pub struct StreamRevision(pub u64);
92
93#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
94pub struct CodecId(pub u32);
95
96impl CodecId {
97    pub const JSON_UTF8: Self = Self(1);
98    pub const RUST_BINCODE_V1: Self = Self(0x8000_0001);
99}
100
101#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
102pub struct EventType(String);
103
104impl EventType {
105    pub fn new(value: impl Into<String>) -> Result<Self> {
106        let value = value.into();
107        if value.is_empty() {
108            return Err(SalamanderError::InvalidFormat(
109                "event type must not be empty".into(),
110            ));
111        }
112        if value.as_bytes().contains(&0) {
113            return Err(SalamanderError::InvalidFormat(
114                "event type must not contain NUL".into(),
115            ));
116        }
117        Ok(Self(value))
118    }
119
120    pub fn as_str(&self) -> &str {
121        &self.0
122    }
123}
124
125pub type Metadata = BTreeMap<String, Vec<u8>>;
126
127#[derive(Debug, Clone, Copy, PartialEq, Eq)]
128#[repr(u8)]
129pub enum FrameKind {
130    Event = 1,
131    System = 2,
132    BatchBegin = 3,
133    BatchCommit = 4,
134}
135
136impl TryFrom<u8> for FrameKind {
137    type Error = SalamanderError;
138
139    fn try_from(value: u8) -> Result<Self> {
140        match value {
141            1 => Ok(Self::Event),
142            2 => Ok(Self::System),
143            3 => Ok(Self::BatchBegin),
144            4 => Ok(Self::BatchCommit),
145            other => Err(SalamanderError::InvalidFormat(format!(
146                "unknown frame kind {other}"
147            ))),
148        }
149    }
150}
151
152#[derive(Debug, Clone, PartialEq, Eq)]
153pub struct RecordEnvelopeV2 {
154    pub event_id: EventId,
155    pub database_id: DatabaseId,
156    pub branch_id: BranchId,
157    pub stream_id: StreamId,
158    pub stream_revision: StreamRevision,
159    pub timestamp_unix_nanos: i64,
160    pub event_type: EventType,
161    pub schema_version: u32,
162    pub codec: CodecId,
163    pub batch_id: BatchId,
164    pub batch_index: u32,
165    pub metadata: Metadata,
166}
167
168#[derive(Debug, Clone, PartialEq, Eq)]
169pub struct StoredRecord<'a> {
170    pub kind: FrameKind,
171    pub flags: u8,
172    pub position: u64,
173    pub envelope: RecordEnvelopeV2,
174    pub payload: &'a [u8],
175}
176
177#[derive(Debug, Clone, PartialEq, Eq)]
178pub struct OwnedStoredRecord {
179    pub kind: FrameKind,
180    pub flags: u8,
181    pub position: u64,
182    pub envelope: RecordEnvelopeV2,
183    pub payload: Vec<u8>,
184}
185
186impl OwnedStoredRecord {
187    pub fn as_borrowed(&self) -> StoredRecord<'_> {
188        StoredRecord {
189            kind: self.kind,
190            flags: self.flags,
191            position: self.position,
192            envelope: self.envelope.clone(),
193            payload: &self.payload,
194        }
195    }
196}
197
198impl From<StoredRecord<'_>> for OwnedStoredRecord {
199    fn from(value: StoredRecord<'_>) -> Self {
200        Self {
201            kind: value.kind,
202            flags: value.flags,
203            position: value.position,
204            envelope: value.envelope,
205            payload: value.payload.to_vec(),
206        }
207    }
208}
209
210static ID_COUNTER: AtomicU64 = AtomicU64::new(1);
211
212/// Generate a process-unique 128-bit identifier without requiring an ambient
213/// random-number service. The timestamp half and process/counter half make
214/// collisions across normal embedded-engine operation vanishingly unlikely.
215pub fn generate_id_bytes() -> [u8; 16] {
216    let nanos = SystemTime::now()
217        .duration_since(UNIX_EPOCH)
218        .map(|duration| duration.as_nanos())
219        .unwrap_or(0);
220    let counter = ID_COUNTER.fetch_add(1, Ordering::Relaxed);
221    let process = u64::from(std::process::id());
222    let mut bytes = [0; 16];
223    bytes[..8].copy_from_slice(&(nanos as u64).to_le_bytes());
224    bytes[8..].copy_from_slice(&(counter ^ process.rotate_left(32)).to_le_bytes());
225    bytes
226}
227
228/// Deterministically map a stream name into a compact v2 placeholder ID.
229/// WP-02 replaces this with the first-class stream catalog while preserving
230/// the stable ID already written in v2 envelopes.
231pub fn derive_stream_id(database: DatabaseId, branch: BranchId, name: &str) -> StreamId {
232    let mut left = 0xcbf2_9ce4_8422_2325u64;
233    let mut right = 0x8422_2325_cbf2_9ce4u64;
234    for byte in database
235        .as_bytes()
236        .iter()
237        .chain(branch.as_bytes())
238        .chain(name.as_bytes())
239    {
240        left ^= u64::from(*byte);
241        left = left.wrapping_mul(0x0000_0100_0000_01b3);
242        right ^= u64::from(*byte).rotate_left(1);
243        right = right.wrapping_mul(0x9e37_79b1_85eb_ca87);
244    }
245    let mut bytes = [0; 16];
246    bytes[..8].copy_from_slice(&left.to_le_bytes());
247    bytes[8..].copy_from_slice(&right.to_le_bytes());
248    StreamId::from_bytes(bytes)
249}
250
251pub(crate) fn derive_import_id(source_identity: [u8; 16], offset: u64) -> [u8; 16] {
252    let mut bytes = source_identity;
253    for (index, byte) in offset.to_le_bytes().iter().enumerate() {
254        bytes[index] ^= *byte;
255        bytes[index + 8] ^= byte.rotate_left(1);
256    }
257    for round in 0..4u8 {
258        for index in 0..16 {
259            let next = bytes[(index + 1) % 16];
260            bytes[index] = bytes[index]
261                .wrapping_add(next)
262                .rotate_left(u32::from((round + index as u8) % 7 + 1));
263        }
264    }
265    bytes
266}
267
268#[derive(Debug, Clone, Copy)]
269pub struct FormatLimits {
270    pub max_envelope_bytes: u32,
271    pub max_payload_bytes: u32,
272    pub max_metadata_entries: u16,
273    pub max_metadata_key_bytes: u16,
274    pub max_metadata_value_bytes: u32,
275    pub max_event_type_bytes: u16,
276}
277
278impl Default for FormatLimits {
279    fn default() -> Self {
280        Self {
281            max_envelope_bytes: 1024 * 1024,
282            max_payload_bytes: 16 * 1024 * 1024,
283            max_metadata_entries: 256,
284            max_metadata_key_bytes: 1024,
285            max_metadata_value_bytes: 64 * 1024,
286            max_event_type_bytes: 1024,
287        }
288    }
289}
290
291pub fn encode(record: &StoredRecord<'_>, limits: FormatLimits) -> Result<Vec<u8>> {
292    validate_flags(record.flags)?;
293    enforce_limit(
294        "payload bytes",
295        record.payload.len(),
296        limits.max_payload_bytes as usize,
297    )?;
298    let envelope = encode_envelope(&record.envelope, limits)?;
299    enforce_limit(
300        "envelope bytes",
301        envelope.len(),
302        limits.max_envelope_bytes as usize,
303    )?;
304
305    let mut out = Vec::with_capacity(FRAME_HEADER_LEN + envelope.len() + record.payload.len());
306    out.extend_from_slice(&FRAME_MAGIC);
307    out.extend_from_slice(&FRAME_VERSION.to_le_bytes());
308    out.push(record.kind as u8);
309    out.push(record.flags);
310    out.extend_from_slice(&(envelope.len() as u32).to_le_bytes());
311    out.extend_from_slice(&(record.payload.len() as u32).to_le_bytes());
312    out.extend_from_slice(&record.position.to_le_bytes());
313    out.extend_from_slice(&0u32.to_le_bytes());
314    out.extend_from_slice(&0u32.to_le_bytes());
315    out.extend_from_slice(&envelope);
316    out.extend_from_slice(record.payload);
317
318    let crc = frame_crc(&out);
319    out[24..28].copy_from_slice(&crc.to_le_bytes());
320    Ok(out)
321}
322
323/// Header-only probe: how many bytes the frame starting at `buf[0]` spans,
324/// or `None` when fewer than a full header is available. Validates the
325/// fixed header (magic, version, kind, flags, limits) but reads no envelope
326/// or payload bytes and computes no checksum — the streaming reader uses it
327/// to size its buffer fill before a full `decode`.
328pub(crate) fn frame_total_len(buf: &[u8], limits: FormatLimits) -> Result<Option<usize>> {
329    if buf.len() < FRAME_HEADER_LEN {
330        return Ok(None);
331    }
332    if buf[0..4] != FRAME_MAGIC {
333        return Err(SalamanderError::InvalidFormat("bad v2 frame magic".into()));
334    }
335    let version = read_u16(&buf[4..6]);
336    if version != FRAME_VERSION {
337        return Err(SalamanderError::InvalidFormat(format!(
338            "unsupported frame version {version}"
339        )));
340    }
341    FrameKind::try_from(buf[6])?;
342    validate_flags(buf[7])?;
343    let envelope_len = read_u32(&buf[8..12]) as usize;
344    let payload_len = read_u32(&buf[12..16]) as usize;
345    enforce_limit(
346        "envelope bytes",
347        envelope_len,
348        limits.max_envelope_bytes as usize,
349    )?;
350    enforce_limit(
351        "payload bytes",
352        payload_len,
353        limits.max_payload_bytes as usize,
354    )?;
355    FRAME_HEADER_LEN
356        .checked_add(envelope_len)
357        .and_then(|n| n.checked_add(payload_len))
358        .map(Some)
359        .ok_or_else(|| SalamanderError::InvalidFormat("frame length overflow".into()))
360}
361
362pub fn decode(buf: &[u8], limits: FormatLimits) -> Result<Option<(StoredRecord<'_>, usize)>> {
363    if buf.len() < FRAME_HEADER_LEN {
364        return Ok(None);
365    }
366    if buf[0..4] != FRAME_MAGIC {
367        return Err(SalamanderError::InvalidFormat("bad v2 frame magic".into()));
368    }
369    let version = read_u16(&buf[4..6]);
370    if version != FRAME_VERSION {
371        return Err(SalamanderError::InvalidFormat(format!(
372            "unsupported frame version {version}"
373        )));
374    }
375    let kind = FrameKind::try_from(buf[6])?;
376    let flags = buf[7];
377    validate_flags(flags)?;
378    let envelope_len = read_u32(&buf[8..12]) as usize;
379    let payload_len = read_u32(&buf[12..16]) as usize;
380    enforce_limit(
381        "envelope bytes",
382        envelope_len,
383        limits.max_envelope_bytes as usize,
384    )?;
385    enforce_limit(
386        "payload bytes",
387        payload_len,
388        limits.max_payload_bytes as usize,
389    )?;
390    let position = read_u64(&buf[16..24]);
391    let stored_crc = read_u32(&buf[24..28]);
392    if read_u32(&buf[28..32]) != 0 {
393        return Err(SalamanderError::InvalidFormat(
394            "nonzero reserved frame header field".into(),
395        ));
396    }
397    let total = FRAME_HEADER_LEN
398        .checked_add(envelope_len)
399        .and_then(|n| n.checked_add(payload_len))
400        .ok_or_else(|| SalamanderError::InvalidFormat("frame length overflow".into()))?;
401    if buf.len() < total {
402        return Ok(None);
403    }
404    let computed_crc = frame_crc(&buf[..total]);
405    if stored_crc != computed_crc {
406        return Err(SalamanderError::Corrupt {
407            offset: position,
408            reason: format!(
409                "v2 crc mismatch: stored {stored_crc:#010x}, computed {computed_crc:#010x}"
410            ),
411        });
412    }
413    let envelope_end = FRAME_HEADER_LEN + envelope_len;
414    let envelope = decode_envelope(&buf[FRAME_HEADER_LEN..envelope_end], limits)?;
415    let payload = &buf[envelope_end..total];
416    Ok(Some((
417        StoredRecord {
418            kind,
419            flags,
420            position,
421            envelope,
422            payload,
423        },
424        total,
425    )))
426}
427
428fn encode_envelope(envelope: &RecordEnvelopeV2, limits: FormatLimits) -> Result<Vec<u8>> {
429    enforce_limit(
430        "event type bytes",
431        envelope.event_type.as_str().len(),
432        limits.max_event_type_bytes as usize,
433    )?;
434    enforce_limit(
435        "metadata entries",
436        envelope.metadata.len(),
437        limits.max_metadata_entries as usize,
438    )?;
439
440    let mut out = Vec::new();
441    out.extend_from_slice(&ENVELOPE_VERSION.to_le_bytes());
442    out.extend_from_slice(envelope.event_id.as_bytes());
443    out.extend_from_slice(envelope.database_id.as_bytes());
444    out.extend_from_slice(envelope.branch_id.as_bytes());
445    out.extend_from_slice(envelope.stream_id.as_bytes());
446    out.extend_from_slice(&envelope.stream_revision.0.to_le_bytes());
447    out.extend_from_slice(&envelope.timestamp_unix_nanos.to_le_bytes());
448    write_short_bytes(&mut out, envelope.event_type.as_str().as_bytes())?;
449    out.extend_from_slice(&envelope.schema_version.to_le_bytes());
450    out.extend_from_slice(&envelope.codec.0.to_le_bytes());
451    out.extend_from_slice(envelope.batch_id.as_bytes());
452    out.extend_from_slice(&envelope.batch_index.to_le_bytes());
453    out.extend_from_slice(&(envelope.metadata.len() as u16).to_le_bytes());
454    for (key, value) in &envelope.metadata {
455        validate_metadata_key(key)?;
456        enforce_limit(
457            "metadata key bytes",
458            key.len(),
459            limits.max_metadata_key_bytes as usize,
460        )?;
461        enforce_limit(
462            "metadata value bytes",
463            value.len(),
464            limits.max_metadata_value_bytes as usize,
465        )?;
466        write_short_bytes(&mut out, key.as_bytes())?;
467        out.extend_from_slice(&(value.len() as u32).to_le_bytes());
468        out.extend_from_slice(value);
469    }
470    Ok(out)
471}
472
473fn decode_envelope(buf: &[u8], limits: FormatLimits) -> Result<RecordEnvelopeV2> {
474    let mut cursor = Cursor::new(buf);
475    let version = cursor.u16()?;
476    if version != ENVELOPE_VERSION {
477        return Err(SalamanderError::InvalidFormat(format!(
478            "unsupported envelope version {version}"
479        )));
480    }
481    let event_id = EventId::from_bytes(cursor.array()?);
482    let database_id = DatabaseId::from_bytes(cursor.array()?);
483    let branch_id = BranchId::from_bytes(cursor.array()?);
484    let stream_id = StreamId::from_bytes(cursor.array()?);
485    let stream_revision = StreamRevision(cursor.u64()?);
486    let timestamp_unix_nanos = cursor.i64()?;
487    let event_type_bytes = cursor.short_bytes(limits.max_event_type_bytes as usize)?;
488    let event_type = EventType::new(
489        std::str::from_utf8(event_type_bytes)
490            .map_err(|_| SalamanderError::InvalidFormat("event type is not UTF-8".into()))?,
491    )?;
492    let schema_version = cursor.u32()?;
493    let codec = CodecId(cursor.u32()?);
494    let batch_id = BatchId::from_bytes(cursor.array()?);
495    let batch_index = cursor.u32()?;
496    let metadata_count = cursor.u16()? as usize;
497    enforce_limit(
498        "metadata entries",
499        metadata_count,
500        limits.max_metadata_entries as usize,
501    )?;
502    let mut metadata = Metadata::new();
503    for _ in 0..metadata_count {
504        let key_bytes = cursor.short_bytes(limits.max_metadata_key_bytes as usize)?;
505        let key = std::str::from_utf8(key_bytes)
506            .map_err(|_| SalamanderError::InvalidFormat("metadata key is not UTF-8".into()))?
507            .to_string();
508        validate_metadata_key(&key)?;
509        let value_len = cursor.u32()? as usize;
510        enforce_limit(
511            "metadata value bytes",
512            value_len,
513            limits.max_metadata_value_bytes as usize,
514        )?;
515        let value = cursor.take(value_len)?.to_vec();
516        if metadata.insert(key, value).is_some() {
517            return Err(SalamanderError::InvalidFormat(
518                "duplicate metadata key".into(),
519            ));
520        }
521    }
522    if !cursor.is_empty() {
523        return Err(SalamanderError::InvalidFormat(
524            "trailing bytes in v2 envelope".into(),
525        ));
526    }
527    Ok(RecordEnvelopeV2 {
528        event_id,
529        database_id,
530        branch_id,
531        stream_id,
532        stream_revision,
533        timestamp_unix_nanos,
534        event_type,
535        schema_version,
536        codec,
537        batch_id,
538        batch_index,
539        metadata,
540    })
541}
542
543fn frame_crc(bytes: &[u8]) -> u32 {
544    let mut crc = crc32c::crc32c(&bytes[..24]);
545    crc = crc32c::crc32c_append(crc, &bytes[28..]);
546    crc
547}
548
549fn validate_flags(flags: u8) -> Result<()> {
550    if flags & !KNOWN_FLAGS != 0 {
551        return Err(SalamanderError::InvalidFormat(format!(
552            "unknown mandatory frame flags {flags:#04x}"
553        )));
554    }
555    Ok(())
556}
557
558fn validate_metadata_key(key: &str) -> Result<()> {
559    if key.is_empty() || key.as_bytes().contains(&0) {
560        return Err(SalamanderError::InvalidFormat(
561            "metadata key must be nonempty UTF-8 without NUL".into(),
562        ));
563    }
564    if key.starts_with(RESERVED_METADATA_PREFIX) && !is_known_reserved_key(key) {
565        return Err(SalamanderError::InvalidFormat(format!(
566            "unknown reserved metadata key {key}"
567        )));
568    }
569    Ok(())
570}
571
572fn is_known_reserved_key(key: &str) -> bool {
573    matches!(
574        key,
575        "salamander.stream_name"
576            | "salamander.v1_offset"
577            | "salamander.v1_namespace"
578            | "salamander.batch_count"
579            | "salamander.batch_digest"
580            | "salamander.idempotency_key"
581            | "salamander.request_digest"
582    )
583}
584
585fn enforce_limit(resource: &'static str, actual: usize, maximum: usize) -> Result<()> {
586    if actual > maximum {
587        return Err(SalamanderError::ResourceLimit {
588            resource,
589            actual: actual as u64,
590            maximum: maximum as u64,
591        });
592    }
593    Ok(())
594}
595
596fn write_short_bytes(out: &mut Vec<u8>, bytes: &[u8]) -> Result<()> {
597    let len = u16::try_from(bytes.len()).map_err(|_| SalamanderError::ResourceLimit {
598        resource: "short byte string",
599        actual: bytes.len() as u64,
600        maximum: u16::MAX as u64,
601    })?;
602    out.extend_from_slice(&len.to_le_bytes());
603    out.extend_from_slice(bytes);
604    Ok(())
605}
606
607fn read_u16(bytes: &[u8]) -> u16 {
608    u16::from_le_bytes(bytes.try_into().expect("two-byte slice"))
609}
610
611fn read_u32(bytes: &[u8]) -> u32 {
612    u32::from_le_bytes(bytes.try_into().expect("four-byte slice"))
613}
614
615fn read_u64(bytes: &[u8]) -> u64 {
616    u64::from_le_bytes(bytes.try_into().expect("eight-byte slice"))
617}
618
619struct Cursor<'a> {
620    remaining: &'a [u8],
621}
622
623impl<'a> Cursor<'a> {
624    fn new(bytes: &'a [u8]) -> Self {
625        Self { remaining: bytes }
626    }
627
628    fn take(&mut self, len: usize) -> Result<&'a [u8]> {
629        if self.remaining.len() < len {
630            return Err(SalamanderError::InvalidFormat(
631                "truncated v2 envelope".into(),
632            ));
633        }
634        let (head, tail) = self.remaining.split_at(len);
635        self.remaining = tail;
636        Ok(head)
637    }
638
639    fn array<const N: usize>(&mut self) -> Result<[u8; N]> {
640        Ok(self
641            .take(N)?
642            .try_into()
643            .expect("slice has requested length"))
644    }
645
646    fn u16(&mut self) -> Result<u16> {
647        Ok(u16::from_le_bytes(self.array()?))
648    }
649
650    fn u32(&mut self) -> Result<u32> {
651        Ok(u32::from_le_bytes(self.array()?))
652    }
653
654    fn u64(&mut self) -> Result<u64> {
655        Ok(u64::from_le_bytes(self.array()?))
656    }
657
658    fn i64(&mut self) -> Result<i64> {
659        Ok(i64::from_le_bytes(self.array()?))
660    }
661
662    fn short_bytes(&mut self, maximum: usize) -> Result<&'a [u8]> {
663        let len = self.u16()? as usize;
664        enforce_limit("short byte string", len, maximum)?;
665        self.take(len)
666    }
667
668    fn is_empty(&self) -> bool {
669        self.remaining.is_empty()
670    }
671}
672
673pub trait TypedCodec<T> {
674    fn id(&self) -> CodecId;
675    fn encode(&self, value: &T) -> Result<Vec<u8>>;
676    fn decode(&self, bytes: &[u8]) -> Result<T>;
677}
678
679#[derive(Debug, Clone, Copy, Default)]
680pub struct JsonCodec;
681
682impl<T> TypedCodec<T> for JsonCodec
683where
684    T: Serialize + DeserializeOwned,
685{
686    fn id(&self) -> CodecId {
687        CodecId::JSON_UTF8
688    }
689
690    fn encode(&self, value: &T) -> Result<Vec<u8>> {
691        serde_json::to_vec(value).map_err(|error| SalamanderError::Codec(error.to_string()))
692    }
693
694    fn decode(&self, bytes: &[u8]) -> Result<T> {
695        serde_json::from_slice(bytes).map_err(|error| SalamanderError::Codec(error.to_string()))
696    }
697}
698
699#[derive(Debug, Clone, Copy, Default)]
700pub struct BincodeCodec;
701
702impl<T> TypedCodec<T> for BincodeCodec
703where
704    T: Serialize + DeserializeOwned,
705{
706    fn id(&self) -> CodecId {
707        CodecId::RUST_BINCODE_V1
708    }
709
710    fn encode(&self, value: &T) -> Result<Vec<u8>> {
711        bincode::serialize(value).map_err(|error| SalamanderError::Codec(error.to_string()))
712    }
713
714    fn decode(&self, bytes: &[u8]) -> Result<T> {
715        bincode::deserialize(bytes).map_err(|error| SalamanderError::Codec(error.to_string()))
716    }
717}
718
719#[cfg(test)]
720mod tests {
721    use super::*;
722    use proptest::prelude::*;
723
724    fn id(byte: u8) -> [u8; 16] {
725        [byte; 16]
726    }
727
728    fn record<'a>(payload: &'a [u8]) -> StoredRecord<'a> {
729        let mut metadata = Metadata::new();
730        metadata.insert("actor".into(), b"agent-7".to_vec());
731        StoredRecord {
732            kind: FrameKind::Event,
733            flags: 0,
734            position: 42,
735            envelope: RecordEnvelopeV2 {
736                event_id: EventId::from_bytes(id(1)),
737                database_id: DatabaseId::from_bytes(id(2)),
738                branch_id: BranchId::from_bytes(id(3)),
739                stream_id: StreamId::from_bytes(id(4)),
740                stream_revision: StreamRevision(9),
741                timestamp_unix_nanos: -123,
742                event_type: EventType::new("chat.message").unwrap(),
743                schema_version: 7,
744                codec: CodecId::JSON_UTF8,
745                batch_id: BatchId::from_bytes(id(5)),
746                batch_index: 2,
747                metadata,
748            },
749            payload,
750        }
751    }
752
753    fn decode_hex(text: &str) -> Vec<u8> {
754        let compact: String = text.chars().filter(|c| !c.is_whitespace()).collect();
755        assert_eq!(compact.len() % 2, 0, "hex fixture has an odd digit count");
756        compact
757            .as_bytes()
758            .chunks_exact(2)
759            .map(|pair| {
760                let pair = std::str::from_utf8(pair).unwrap();
761                u8::from_str_radix(pair, 16).unwrap()
762            })
763            .collect()
764    }
765
766    #[test]
767    fn round_trip_preserves_envelope_and_opaque_payload() {
768        let original = record(&[0, 159, 255, 1]);
769        let bytes = encode(&original, FormatLimits::default()).unwrap();
770        let (decoded, consumed) = decode(&bytes, FormatLimits::default()).unwrap().unwrap();
771        assert_eq!(decoded, original);
772        assert_eq!(consumed, bytes.len());
773    }
774
775    #[test]
776    fn checked_in_golden_event_frame_is_stable() {
777        let expected = encode(&record(&[0, 1, 159, 255]), FormatLimits::default()).unwrap();
778        let fixture = decode_hex(include_str!(
779            "../../tests/fixtures/format-v2/event-minimal.hex"
780        ));
781        assert_eq!(fixture, expected, "v2 byte layout changed");
782        let (decoded, consumed) = decode(&fixture, FormatLimits::default()).unwrap().unwrap();
783        assert_eq!(consumed, fixture.len());
784        assert_eq!(decoded, record(&[0, 1, 159, 255]));
785    }
786
787    #[test]
788    fn short_frame_is_incomplete_not_corrupt() {
789        let bytes = encode(&record(b"body"), FormatLimits::default()).unwrap();
790        for length in 0..bytes.len() {
791            let result = decode(&bytes[..length], FormatLimits::default());
792            assert!(matches!(result, Ok(None)));
793        }
794    }
795
796    #[test]
797    fn rejects_unknown_flags_before_payload_access() {
798        let mut bytes = encode(&record(b"body"), FormatLimits::default()).unwrap();
799        bytes[7] = 0x80;
800        assert!(matches!(
801            decode(&bytes, FormatLimits::default()),
802            Err(SalamanderError::InvalidFormat(_))
803        ));
804    }
805
806    #[test]
807    fn rejects_bad_magic_version_and_reserved_header() {
808        let original = encode(&record(b"body"), FormatLimits::default()).unwrap();
809
810        let mut bad_magic = original.clone();
811        bad_magic[0] ^= 1;
812        assert!(matches!(
813            decode(&bad_magic, FormatLimits::default()),
814            Err(SalamanderError::InvalidFormat(_))
815        ));
816
817        let mut bad_version = original.clone();
818        bad_version[4..6].copy_from_slice(&99u16.to_le_bytes());
819        assert!(matches!(
820            decode(&bad_version, FormatLimits::default()),
821            Err(SalamanderError::InvalidFormat(_))
822        ));
823
824        let mut bad_reserved = original;
825        bad_reserved[28] = 1;
826        assert!(matches!(
827            decode(&bad_reserved, FormatLimits::default()),
828            Err(SalamanderError::InvalidFormat(_))
829        ));
830    }
831
832    #[test]
833    fn rejects_checksum_mutation() {
834        let mut bytes = encode(&record(b"body"), FormatLimits::default()).unwrap();
835        *bytes.last_mut().unwrap() ^= 1;
836        assert!(matches!(
837            decode(&bytes, FormatLimits::default()),
838            Err(SalamanderError::Corrupt { offset: 42, .. })
839        ));
840    }
841
842    #[test]
843    fn limits_are_checked_before_waiting_for_declared_body() {
844        let mut bytes = encode(&record(b"body"), FormatLimits::default()).unwrap();
845        bytes[12..16].copy_from_slice(&u32::MAX.to_le_bytes());
846        assert!(matches!(
847            decode(&bytes, FormatLimits::default()),
848            Err(SalamanderError::ResourceLimit {
849                resource: "payload bytes",
850                ..
851            })
852        ));
853    }
854
855    #[test]
856    fn json_codec_is_portable_utf8_json() {
857        let value = serde_json::json!({"hello": [1, true, null]});
858        let bytes = JsonCodec.encode(&value).unwrap();
859        assert_eq!(
860            std::str::from_utf8(&bytes).unwrap(),
861            r#"{"hello":[1,true,null]}"#
862        );
863        let decoded: serde_json::Value = JsonCodec.decode(&bytes).unwrap();
864        assert_eq!(decoded, value);
865    }
866
867    proptest! {
868        #[test]
869        fn arbitrary_payload_round_trips(payload in prop::collection::vec(any::<u8>(), 0..4096)) {
870            let original = record(&payload);
871            let bytes = encode(&original, FormatLimits::default()).unwrap();
872            let (decoded, consumed) = decode(&bytes, FormatLimits::default()).unwrap().unwrap();
873            prop_assert_eq!(decoded.payload, payload.as_slice());
874            prop_assert_eq!(decoded.envelope, original.envelope);
875            prop_assert_eq!(consumed, bytes.len());
876        }
877    }
878}
salamander/format/mod.rs

salamander/format/
mod.rs
