nautilus_event_store/
codec.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Hardened positional serde codec for event-store records.
17
18use std::{char, fmt::Display, str};
19
20use serde::{
21    Serialize,
22    de::{
23        self, DeserializeOwned, DeserializeSeed, EnumAccess, IntoDeserializer, MapAccess,
24        SeqAccess, VariantAccess, Visitor,
25    },
26    ser::{
27        self, SerializeMap, SerializeSeq, SerializeStruct, SerializeStructVariant, SerializeTuple,
28        SerializeTupleStruct, SerializeTupleVariant,
29    },
30};
31use thiserror::Error;
32
33const MAGIC: [u8; 4] = *b"NESC";
34const VERSION: u8 = 1;
35const HEADER_LEN: usize = MAGIC.len() + 1;
36const MAX_COLLECTION_COUNT: usize = 1_048_576;
37
38/// Serializes `value` into a freshly allocated, framed codec buffer.
39///
40/// The output is `MAGIC ++ VERSION ++ body`. Deterministic: equal values produce
41/// byte-identical output.
42///
43/// # Errors
44///
45/// Returns [`CodecError`] if `value`'s [`Serialize`] impl drives the format outside the
46/// supported positional model, such as an unbounded sequence whose length is not known up front.
47pub fn encode_to_vec<T: Serialize>(value: &T) -> Result<Vec<u8>, CodecError> {
48    let mut encoder = Encoder { out: Vec::new() };
49    encoder.out.extend_from_slice(&MAGIC);
50    encoder.out.push(VERSION);
51    value.serialize(&mut encoder)?;
52    Ok(encoder.out)
53}
54
55/// Decodes a `T` from a complete framed codec buffer.
56///
57/// Validates the frame header, decodes the body positionally, then requires the input to be fully
58/// consumed.
59///
60/// # Errors
61///
62/// Returns [`CodecError`] on a bad or short header, malformed body, a self-describing deserialize
63/// request, or unconsumed trailing bytes.
64pub fn decode_from_slice<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, CodecError> {
65    let mut decoder = Decoder {
66        input: bytes,
67        pos: 0,
68    };
69    decoder.read_header()?;
70    let value = T::deserialize(&mut decoder)?;
71    if decoder.pos != decoder.input.len() {
72        return Err(CodecError::TrailingBytes(decoder.input.len() - decoder.pos));
73    }
74    Ok(value)
75}
76
77/// Errors returned by the event-store positional codec.
78#[derive(Debug, Error)]
79#[non_exhaustive]
80pub enum CodecError {
81    /// The input ended before the requested bytes could be read.
82    #[error("unexpected end of codec input: needed {needed} more byte(s)")]
83    UnexpectedEof {
84        /// Number of bytes the decoder attempted to read.
85        needed: usize,
86    },
87    /// The frame did not start with the event-store codec magic.
88    #[error("bad codec magic: input is not a nautilus event-store codec frame")]
89    BadMagic,
90    /// The frame version is not supported by this decoder.
91    #[error("unsupported codec version: {0}")]
92    UnsupportedVersion(u8),
93    /// An encoded byte length would exceed the remaining input.
94    #[error("encoded length {claimed} exceeds remaining input {remaining}")]
95    LengthOverflow {
96        /// Claimed encoded byte length, as the raw on-wire `u64`.
97        ///
98        /// Kept as `u64` (not `usize`) so the reported value is exact on every
99        /// target, including a length that overflows `usize` on a sub-64-bit
100        /// build - the case that would otherwise be reported with a lossy
101        /// sentinel.
102        claimed: u64,
103        /// Bytes remaining in the input when the length was checked.
104        remaining: usize,
105    },
106    /// An encoded collection count exceeded the supported maximum.
107    #[error("encoded collection count {claimed} exceeds maximum {max}")]
108    CountOverflow {
109        /// Claimed collection count, as the raw on-wire `u64`.
110        claimed: u64,
111        /// Maximum supported collection count.
112        max: usize,
113    },
114    /// A string was not valid UTF-8.
115    #[error("invalid utf-8 in encoded string")]
116    InvalidUtf8,
117    /// A bool discriminant was not `0x00` or `0x01`.
118    #[error("invalid bool discriminant: {0:#04x}")]
119    InvalidBool(u8),
120    /// An option discriminant was not `0x00` or `0x01`.
121    #[error("invalid option discriminant: {0:#04x}")]
122    InvalidOption(u8),
123    /// A decoded char value was not a valid Unicode scalar value.
124    #[error("invalid char scalar value: {0:#010x}")]
125    InvalidChar(u32),
126    /// An enum variant index was outside the type's variant set.
127    #[error("unknown enum variant index: {0}")]
128    UnknownVariant(u32),
129    /// Serde did not provide a sequence or map length.
130    #[error("sequence length was not provided by the serializer")]
131    MissingLen,
132    /// The input had bytes remaining after a complete value was decoded.
133    #[error("{0} unconsumed trailing byte(s) after decode")]
134    TrailingBytes(usize),
135    /// The caller attempted self-describing deserialization.
136    #[error("self-describing deserialization is not supported by this format")]
137    SelfDescribing,
138    /// Serde-generated error message.
139    #[error("{0}")]
140    Message(String),
141}
142
143impl ser::Error for CodecError {
144    fn custom<T: Display>(msg: T) -> Self {
145        Self::Message(msg.to_string())
146    }
147}
148
149impl de::Error for CodecError {
150    fn custom<T: Display>(msg: T) -> Self {
151        Self::Message(msg.to_string())
152    }
153}
154
155#[derive(Debug)]
156struct Encoder {
157    out: Vec<u8>,
158}
159
160impl Encoder {
161    fn write_len(&mut self, len: usize) {
162        self.out.extend_from_slice(&(len as u64).to_le_bytes());
163    }
164
165    fn write_count(&mut self, count: usize) -> Result<(), CodecError> {
166        if count > MAX_COLLECTION_COUNT {
167            return Err(CodecError::CountOverflow {
168                claimed: count as u64,
169                max: MAX_COLLECTION_COUNT,
170            });
171        }
172
173        self.write_len(count);
174        Ok(())
175    }
176
177    fn write_variant(&mut self, variant_index: u32) {
178        self.out.extend_from_slice(&variant_index.to_le_bytes());
179    }
180}
181
182impl ser::Serializer for &mut Encoder {
183    type Ok = ();
184    type Error = CodecError;
185    type SerializeSeq = Self;
186    type SerializeTuple = Self;
187    type SerializeTupleStruct = Self;
188    type SerializeTupleVariant = Self;
189    type SerializeMap = Self;
190    type SerializeStruct = Self;
191    type SerializeStructVariant = Self;
192
193    fn serialize_bool(self, v: bool) -> Result<Self::Ok, Self::Error> {
194        self.out.push(u8::from(v));
195        Ok(())
196    }
197
198    fn serialize_i8(self, v: i8) -> Result<Self::Ok, Self::Error> {
199        self.out.extend_from_slice(&v.to_le_bytes());
200        Ok(())
201    }
202
203    fn serialize_i16(self, v: i16) -> Result<Self::Ok, Self::Error> {
204        self.out.extend_from_slice(&v.to_le_bytes());
205        Ok(())
206    }
207
208    fn serialize_i32(self, v: i32) -> Result<Self::Ok, Self::Error> {
209        self.out.extend_from_slice(&v.to_le_bytes());
210        Ok(())
211    }
212
213    fn serialize_i64(self, v: i64) -> Result<Self::Ok, Self::Error> {
214        self.out.extend_from_slice(&v.to_le_bytes());
215        Ok(())
216    }
217
218    fn serialize_u8(self, v: u8) -> Result<Self::Ok, Self::Error> {
219        self.out.push(v);
220        Ok(())
221    }
222
223    fn serialize_u16(self, v: u16) -> Result<Self::Ok, Self::Error> {
224        self.out.extend_from_slice(&v.to_le_bytes());
225        Ok(())
226    }
227
228    fn serialize_u32(self, v: u32) -> Result<Self::Ok, Self::Error> {
229        self.out.extend_from_slice(&v.to_le_bytes());
230        Ok(())
231    }
232
233    fn serialize_u64(self, v: u64) -> Result<Self::Ok, Self::Error> {
234        self.out.extend_from_slice(&v.to_le_bytes());
235        Ok(())
236    }
237
238    fn serialize_f32(self, v: f32) -> Result<Self::Ok, Self::Error> {
239        self.out.extend_from_slice(&v.to_le_bytes());
240        Ok(())
241    }
242
243    fn serialize_f64(self, v: f64) -> Result<Self::Ok, Self::Error> {
244        self.out.extend_from_slice(&v.to_le_bytes());
245        Ok(())
246    }
247
248    fn serialize_char(self, v: char) -> Result<Self::Ok, Self::Error> {
249        self.out.extend_from_slice(&(v as u32).to_le_bytes());
250        Ok(())
251    }
252
253    fn serialize_str(self, v: &str) -> Result<Self::Ok, Self::Error> {
254        self.write_len(v.len());
255        self.out.extend_from_slice(v.as_bytes());
256        Ok(())
257    }
258
259    fn serialize_bytes(self, v: &[u8]) -> Result<Self::Ok, Self::Error> {
260        self.write_len(v.len());
261        self.out.extend_from_slice(v);
262        Ok(())
263    }
264
265    fn serialize_none(self) -> Result<Self::Ok, Self::Error> {
266        self.out.push(0);
267        Ok(())
268    }
269
270    fn serialize_some<T>(self, value: &T) -> Result<Self::Ok, Self::Error>
271    where
272        T: ?Sized + Serialize,
273    {
274        self.out.push(1);
275        value.serialize(self)
276    }
277
278    fn serialize_unit(self) -> Result<Self::Ok, Self::Error> {
279        Ok(())
280    }
281
282    fn serialize_unit_struct(self, _name: &'static str) -> Result<Self::Ok, Self::Error> {
283        Ok(())
284    }
285
286    fn serialize_unit_variant(
287        self,
288        _name: &'static str,
289        variant_index: u32,
290        _variant: &'static str,
291    ) -> Result<Self::Ok, Self::Error> {
292        self.write_variant(variant_index);
293        Ok(())
294    }
295
296    fn serialize_newtype_struct<T>(
297        self,
298        _name: &'static str,
299        value: &T,
300    ) -> Result<Self::Ok, Self::Error>
301    where
302        T: ?Sized + Serialize,
303    {
304        value.serialize(self)
305    }
306
307    fn serialize_newtype_variant<T>(
308        self,
309        _name: &'static str,
310        variant_index: u32,
311        _variant: &'static str,
312        value: &T,
313    ) -> Result<Self::Ok, Self::Error>
314    where
315        T: ?Sized + Serialize,
316    {
317        self.write_variant(variant_index);
318        value.serialize(self)
319    }
320
321    fn serialize_seq(self, len: Option<usize>) -> Result<Self::SerializeSeq, Self::Error> {
322        let len = len.ok_or(CodecError::MissingLen)?;
323        self.write_count(len)?;
324        Ok(self)
325    }
326
327    fn serialize_tuple(self, _len: usize) -> Result<Self::SerializeTuple, Self::Error> {
328        Ok(self)
329    }
330
331    fn serialize_tuple_struct(
332        self,
333        _name: &'static str,
334        _len: usize,
335    ) -> Result<Self::SerializeTupleStruct, Self::Error> {
336        Ok(self)
337    }
338
339    fn serialize_tuple_variant(
340        self,
341        _name: &'static str,
342        variant_index: u32,
343        _variant: &'static str,
344        _len: usize,
345    ) -> Result<Self::SerializeTupleVariant, Self::Error> {
346        self.write_variant(variant_index);
347        Ok(self)
348    }
349
350    fn serialize_map(self, len: Option<usize>) -> Result<Self::SerializeMap, Self::Error> {
351        let len = len.ok_or(CodecError::MissingLen)?;
352        self.write_count(len)?;
353        Ok(self)
354    }
355
356    fn serialize_struct(
357        self,
358        _name: &'static str,
359        _len: usize,
360    ) -> Result<Self::SerializeStruct, Self::Error> {
361        Ok(self)
362    }
363
364    fn serialize_struct_variant(
365        self,
366        _name: &'static str,
367        variant_index: u32,
368        _variant: &'static str,
369        _len: usize,
370    ) -> Result<Self::SerializeStructVariant, Self::Error> {
371        self.write_variant(variant_index);
372        Ok(self)
373    }
374
375    fn is_human_readable(&self) -> bool {
376        false
377    }
378}
379
380impl SerializeSeq for &mut Encoder {
381    type Ok = ();
382    type Error = CodecError;
383
384    fn serialize_element<T>(&mut self, value: &T) -> Result<(), Self::Error>
385    where
386        T: ?Sized + Serialize,
387    {
388        value.serialize(&mut **self)
389    }
390
391    fn end(self) -> Result<Self::Ok, Self::Error> {
392        Ok(())
393    }
394}
395
396impl SerializeTuple for &mut Encoder {
397    type Ok = ();
398    type Error = CodecError;
399
400    fn serialize_element<T>(&mut self, value: &T) -> Result<(), Self::Error>
401    where
402        T: ?Sized + Serialize,
403    {
404        value.serialize(&mut **self)
405    }
406
407    fn end(self) -> Result<Self::Ok, Self::Error> {
408        Ok(())
409    }
410}
411
412impl SerializeTupleStruct for &mut Encoder {
413    type Ok = ();
414    type Error = CodecError;
415
416    fn serialize_field<T>(&mut self, value: &T) -> Result<(), Self::Error>
417    where
418        T: ?Sized + Serialize,
419    {
420        value.serialize(&mut **self)
421    }
422
423    fn end(self) -> Result<Self::Ok, Self::Error> {
424        Ok(())
425    }
426}
427
428impl SerializeTupleVariant for &mut Encoder {
429    type Ok = ();
430    type Error = CodecError;
431
432    fn serialize_field<T>(&mut self, value: &T) -> Result<(), Self::Error>
433    where
434        T: ?Sized + Serialize,
435    {
436        value.serialize(&mut **self)
437    }
438
439    fn end(self) -> Result<Self::Ok, Self::Error> {
440        Ok(())
441    }
442}
443
444impl SerializeMap for &mut Encoder {
445    type Ok = ();
446    type Error = CodecError;
447
448    fn serialize_key<T>(&mut self, key: &T) -> Result<(), Self::Error>
449    where
450        T: ?Sized + Serialize,
451    {
452        key.serialize(&mut **self)
453    }
454
455    fn serialize_value<T>(&mut self, value: &T) -> Result<(), Self::Error>
456    where
457        T: ?Sized + Serialize,
458    {
459        value.serialize(&mut **self)
460    }
461
462    fn end(self) -> Result<Self::Ok, Self::Error> {
463        Ok(())
464    }
465}
466
467impl SerializeStruct for &mut Encoder {
468    type Ok = ();
469    type Error = CodecError;
470
471    fn serialize_field<T>(&mut self, _key: &'static str, value: &T) -> Result<(), Self::Error>
472    where
473        T: ?Sized + Serialize,
474    {
475        value.serialize(&mut **self)
476    }
477
478    fn end(self) -> Result<Self::Ok, Self::Error> {
479        Ok(())
480    }
481}
482
483impl SerializeStructVariant for &mut Encoder {
484    type Ok = ();
485    type Error = CodecError;
486
487    fn serialize_field<T>(&mut self, _key: &'static str, value: &T) -> Result<(), Self::Error>
488    where
489        T: ?Sized + Serialize,
490    {
491        value.serialize(&mut **self)
492    }
493
494    fn end(self) -> Result<Self::Ok, Self::Error> {
495        Ok(())
496    }
497}
498
499#[derive(Debug)]
500struct Decoder<'de> {
501    input: &'de [u8],
502    pos: usize,
503}
504
505impl<'de> Decoder<'de> {
506    fn read_header(&mut self) -> Result<(), CodecError> {
507        if self.input.len() < HEADER_LEN {
508            return Err(CodecError::UnexpectedEof { needed: HEADER_LEN });
509        }
510
511        let magic = self.take(MAGIC.len())?;
512        if magic != MAGIC {
513            return Err(CodecError::BadMagic);
514        }
515
516        let version = self.take(1)?[0];
517        if version != VERSION {
518            return Err(CodecError::UnsupportedVersion(version));
519        }
520
521        Ok(())
522    }
523
524    fn remaining(&self) -> usize {
525        self.input.len() - self.pos
526    }
527
528    fn take(&mut self, n: usize) -> Result<&'de [u8], CodecError> {
529        let end = self
530            .pos
531            .checked_add(n)
532            .ok_or(CodecError::UnexpectedEof { needed: n })?;
533        let slice = self
534            .input
535            .get(self.pos..end)
536            .ok_or(CodecError::UnexpectedEof { needed: n })?;
537        self.pos = end;
538        Ok(slice)
539    }
540
541    fn read_byte_len(&mut self) -> Result<usize, CodecError> {
542        let raw = self.read_u64()?;
543        let remaining = self.remaining();
544        // A length is valid only if it both fits `usize` and is within the
545        // remaining input. Any failure reports the exact `u64` wire value via
546        // `claimed`, so the overflow case needs no lossy `usize` sentinel.
547        match usize::try_from(raw) {
548            Ok(claimed) if claimed <= remaining => Ok(claimed),
549            _ => Err(CodecError::LengthOverflow {
550                claimed: raw,
551                remaining,
552            }),
553        }
554    }
555
556    fn read_count(&mut self) -> Result<usize, CodecError> {
557        let raw = self.read_u64()?;
558        match usize::try_from(raw) {
559            Ok(count) if count <= MAX_COLLECTION_COUNT => Ok(count),
560            _ => Err(CodecError::CountOverflow {
561                claimed: raw,
562                max: MAX_COLLECTION_COUNT,
563            }),
564        }
565    }
566
567    fn read_u16(&mut self) -> Result<u16, CodecError> {
568        let bytes: [u8; 2] = self
569            .take(2)?
570            .try_into()
571            .expect("take returned the exact requested width");
572        Ok(u16::from_le_bytes(bytes))
573    }
574
575    fn read_u32(&mut self) -> Result<u32, CodecError> {
576        let bytes: [u8; 4] = self
577            .take(4)?
578            .try_into()
579            .expect("take returned the exact requested width");
580        Ok(u32::from_le_bytes(bytes))
581    }
582
583    fn read_u64(&mut self) -> Result<u64, CodecError> {
584        let bytes: [u8; 8] = self
585            .take(8)?
586            .try_into()
587            .expect("take returned the exact requested width");
588        Ok(u64::from_le_bytes(bytes))
589    }
590}
591
592macro_rules! deserialize_integer {
593    ($method:ident, $visit:ident, $ty:ty, $width:literal) => {
594        fn $method<V>(self, visitor: V) -> Result<V::Value, Self::Error>
595        where
596            V: Visitor<'de>,
597        {
598            let bytes: [u8; $width] = self
599                .take($width)?
600                .try_into()
601                .expect("take returned the exact requested width");
602            visitor.$visit(<$ty>::from_le_bytes(bytes))
603        }
604    };
605}
606
607impl<'de> de::Deserializer<'de> for &mut Decoder<'de> {
608    type Error = CodecError;
609
610    fn deserialize_any<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
611    where
612        V: Visitor<'de>,
613    {
614        Err(CodecError::SelfDescribing)
615    }
616
617    fn deserialize_bool<V>(self, visitor: V) -> Result<V::Value, Self::Error>
618    where
619        V: Visitor<'de>,
620    {
621        match self.take(1)?[0] {
622            0 => visitor.visit_bool(false),
623            1 => visitor.visit_bool(true),
624            other => Err(CodecError::InvalidBool(other)),
625        }
626    }
627
628    fn deserialize_i8<V>(self, visitor: V) -> Result<V::Value, Self::Error>
629    where
630        V: Visitor<'de>,
631    {
632        let bytes: [u8; 1] = self
633            .take(1)?
634            .try_into()
635            .expect("take returned the exact requested width");
636        visitor.visit_i8(i8::from_le_bytes(bytes))
637    }
638
639    deserialize_integer!(deserialize_i16, visit_i16, i16, 2);
640    deserialize_integer!(deserialize_i32, visit_i32, i32, 4);
641    deserialize_integer!(deserialize_i64, visit_i64, i64, 8);
642
643    fn deserialize_u8<V>(self, visitor: V) -> Result<V::Value, Self::Error>
644    where
645        V: Visitor<'de>,
646    {
647        visitor.visit_u8(self.take(1)?[0])
648    }
649
650    fn deserialize_u16<V>(self, visitor: V) -> Result<V::Value, Self::Error>
651    where
652        V: Visitor<'de>,
653    {
654        visitor.visit_u16(self.read_u16()?)
655    }
656
657    fn deserialize_u32<V>(self, visitor: V) -> Result<V::Value, Self::Error>
658    where
659        V: Visitor<'de>,
660    {
661        visitor.visit_u32(self.read_u32()?)
662    }
663
664    fn deserialize_u64<V>(self, visitor: V) -> Result<V::Value, Self::Error>
665    where
666        V: Visitor<'de>,
667    {
668        visitor.visit_u64(self.read_u64()?)
669    }
670
671    fn deserialize_f32<V>(self, visitor: V) -> Result<V::Value, Self::Error>
672    where
673        V: Visitor<'de>,
674    {
675        let bytes: [u8; 4] = self
676            .take(4)?
677            .try_into()
678            .expect("take returned the exact requested width");
679        visitor.visit_f32(f32::from_le_bytes(bytes))
680    }
681
682    fn deserialize_f64<V>(self, visitor: V) -> Result<V::Value, Self::Error>
683    where
684        V: Visitor<'de>,
685    {
686        let bytes: [u8; 8] = self
687            .take(8)?
688            .try_into()
689            .expect("take returned the exact requested width");
690        visitor.visit_f64(f64::from_le_bytes(bytes))
691    }
692
693    fn deserialize_char<V>(self, visitor: V) -> Result<V::Value, Self::Error>
694    where
695        V: Visitor<'de>,
696    {
697        let raw = self.read_u32()?;
698        let value = char::from_u32(raw).ok_or(CodecError::InvalidChar(raw))?;
699        visitor.visit_char(value)
700    }
701
702    fn deserialize_str<V>(self, visitor: V) -> Result<V::Value, Self::Error>
703    where
704        V: Visitor<'de>,
705    {
706        let len = self.read_byte_len()?;
707        let bytes = self.take(len)?;
708        let value = str::from_utf8(bytes).map_err(|_| CodecError::InvalidUtf8)?;
709        visitor.visit_borrowed_str(value)
710    }
711
712    fn deserialize_string<V>(self, visitor: V) -> Result<V::Value, Self::Error>
713    where
714        V: Visitor<'de>,
715    {
716        self.deserialize_str(visitor)
717    }
718
719    fn deserialize_bytes<V>(self, visitor: V) -> Result<V::Value, Self::Error>
720    where
721        V: Visitor<'de>,
722    {
723        let len = self.read_byte_len()?;
724        let bytes = self.take(len)?;
725        visitor.visit_borrowed_bytes(bytes)
726    }
727
728    fn deserialize_byte_buf<V>(self, visitor: V) -> Result<V::Value, Self::Error>
729    where
730        V: Visitor<'de>,
731    {
732        self.deserialize_bytes(visitor)
733    }
734
735    fn deserialize_option<V>(self, visitor: V) -> Result<V::Value, Self::Error>
736    where
737        V: Visitor<'de>,
738    {
739        match self.take(1)?[0] {
740            0 => visitor.visit_none(),
741            1 => visitor.visit_some(self),
742            other => Err(CodecError::InvalidOption(other)),
743        }
744    }
745
746    fn deserialize_unit<V>(self, visitor: V) -> Result<V::Value, Self::Error>
747    where
748        V: Visitor<'de>,
749    {
750        visitor.visit_unit()
751    }
752
753    fn deserialize_unit_struct<V>(
754        self,
755        _name: &'static str,
756        visitor: V,
757    ) -> Result<V::Value, Self::Error>
758    where
759        V: Visitor<'de>,
760    {
761        visitor.visit_unit()
762    }
763
764    fn deserialize_newtype_struct<V>(
765        self,
766        _name: &'static str,
767        visitor: V,
768    ) -> Result<V::Value, Self::Error>
769    where
770        V: Visitor<'de>,
771    {
772        visitor.visit_newtype_struct(self)
773    }
774
775    fn deserialize_seq<V>(self, visitor: V) -> Result<V::Value, Self::Error>
776    where
777        V: Visitor<'de>,
778    {
779        let remaining = self.read_count()?;
780        visitor.visit_seq(SeqReader {
781            dec: self,
782            remaining,
783        })
784    }
785
786    fn deserialize_tuple<V>(self, len: usize, visitor: V) -> Result<V::Value, Self::Error>
787    where
788        V: Visitor<'de>,
789    {
790        visitor.visit_seq(SeqReader {
791            dec: self,
792            remaining: len,
793        })
794    }
795
796    fn deserialize_tuple_struct<V>(
797        self,
798        _name: &'static str,
799        len: usize,
800        visitor: V,
801    ) -> Result<V::Value, Self::Error>
802    where
803        V: Visitor<'de>,
804    {
805        self.deserialize_tuple(len, visitor)
806    }
807
808    fn deserialize_map<V>(self, visitor: V) -> Result<V::Value, Self::Error>
809    where
810        V: Visitor<'de>,
811    {
812        let remaining = self.read_count()?;
813        visitor.visit_map(MapReader {
814            dec: self,
815            remaining,
816        })
817    }
818
819    fn deserialize_struct<V>(
820        self,
821        _name: &'static str,
822        fields: &'static [&'static str],
823        visitor: V,
824    ) -> Result<V::Value, Self::Error>
825    where
826        V: Visitor<'de>,
827    {
828        visitor.visit_seq(SeqReader {
829            dec: self,
830            remaining: fields.len(),
831        })
832    }
833
834    fn deserialize_enum<V>(
835        self,
836        _name: &'static str,
837        variants: &'static [&'static str],
838        visitor: V,
839    ) -> Result<V::Value, Self::Error>
840    where
841        V: Visitor<'de>,
842    {
843        let index = self.read_u32()?;
844        if index as usize >= variants.len() {
845            return Err(CodecError::UnknownVariant(index));
846        }
847        visitor.visit_enum(EnumReader { dec: self, index })
848    }
849
850    fn deserialize_identifier<V>(self, visitor: V) -> Result<V::Value, Self::Error>
851    where
852        V: Visitor<'de>,
853    {
854        self.deserialize_u32(visitor)
855    }
856
857    fn deserialize_ignored_any<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
858    where
859        V: Visitor<'de>,
860    {
861        Err(CodecError::SelfDescribing)
862    }
863
864    fn is_human_readable(&self) -> bool {
865        false
866    }
867}
868
869#[derive(Debug)]
870struct SeqReader<'a, 'de> {
871    dec: &'a mut Decoder<'de>,
872    remaining: usize,
873}
874
875impl<'de> SeqAccess<'de> for SeqReader<'_, 'de> {
876    type Error = CodecError;
877
878    fn next_element_seed<T>(&mut self, seed: T) -> Result<Option<T::Value>, Self::Error>
879    where
880        T: DeserializeSeed<'de>,
881    {
882        if self.remaining == 0 {
883            return Ok(None);
884        }
885
886        self.remaining -= 1;
887        seed.deserialize(&mut *self.dec).map(Some)
888    }
889
890    fn size_hint(&self) -> Option<usize> {
891        Some(self.remaining)
892    }
893}
894
895#[derive(Debug)]
896struct MapReader<'a, 'de> {
897    dec: &'a mut Decoder<'de>,
898    remaining: usize,
899}
900
901impl<'de> MapAccess<'de> for MapReader<'_, 'de> {
902    type Error = CodecError;
903
904    fn next_key_seed<K>(&mut self, seed: K) -> Result<Option<K::Value>, Self::Error>
905    where
906        K: DeserializeSeed<'de>,
907    {
908        if self.remaining == 0 {
909            return Ok(None);
910        }
911
912        self.remaining -= 1;
913        seed.deserialize(&mut *self.dec).map(Some)
914    }
915
916    fn next_value_seed<V>(&mut self, seed: V) -> Result<V::Value, Self::Error>
917    where
918        V: DeserializeSeed<'de>,
919    {
920        seed.deserialize(&mut *self.dec)
921    }
922
923    fn size_hint(&self) -> Option<usize> {
924        Some(self.remaining)
925    }
926}
927
928#[derive(Debug)]
929struct EnumReader<'a, 'de> {
930    dec: &'a mut Decoder<'de>,
931    index: u32,
932}
933
934impl<'de> EnumAccess<'de> for EnumReader<'_, 'de> {
935    type Error = CodecError;
936    type Variant = Self;
937
938    fn variant_seed<V>(self, seed: V) -> Result<(V::Value, Self::Variant), Self::Error>
939    where
940        V: DeserializeSeed<'de>,
941    {
942        let value = seed.deserialize(self.index.into_deserializer())?;
943        Ok((value, self))
944    }
945}
946
947impl<'de> VariantAccess<'de> for EnumReader<'_, 'de> {
948    type Error = CodecError;
949
950    fn unit_variant(self) -> Result<(), Self::Error> {
951        Ok(())
952    }
953
954    fn newtype_variant_seed<T>(self, seed: T) -> Result<T::Value, Self::Error>
955    where
956        T: DeserializeSeed<'de>,
957    {
958        seed.deserialize(self.dec)
959    }
960
961    fn tuple_variant<V>(self, len: usize, visitor: V) -> Result<V::Value, Self::Error>
962    where
963        V: Visitor<'de>,
964    {
965        visitor.visit_seq(SeqReader {
966            dec: self.dec,
967            remaining: len,
968        })
969    }
970
971    fn struct_variant<V>(
972        self,
973        fields: &'static [&'static str],
974        visitor: V,
975    ) -> Result<V::Value, Self::Error>
976    where
977        V: Visitor<'de>,
978    {
979        visitor.visit_seq(SeqReader {
980            dec: self.dec,
981            remaining: fields.len(),
982        })
983    }
984}
985
986#[cfg(test)]
987mod tests {
988    use std::array;
989
990    use bytes::Bytes;
991    use indexmap::IndexMap;
992    use nautilus_core::{UUID4, UnixNanos};
993    use nautilus_model::data::DataType;
994    use nautilus_system::RegisteredComponents;
995    use proptest::{prelude::*, test_runner::Config as ProptestConfig};
996    use rstest::rstest;
997    use serde::{Deserialize, Serializer};
998    use ustr::Ustr;
999
1000    use super::*;
1001    use crate::{
1002        EventStoreEntry, Headers, RunManifest, RunStatus, SnapshotAnchor,
1003        hash::{EntryHash, compute_entry_hash},
1004        markers::{
1005            DataClass, DataCursorSnapshot, HiFiMarker, MarkerGap, MarkerGapReason, StreamCursor,
1006            StreamDictEntry,
1007        },
1008    };
1009
1010    #[derive(Debug, Serialize, Deserialize)]
1011    struct ScalarProbe {
1012        b: bool,
1013        i8s: [i8; 3],
1014        i16s: [i16; 3],
1015        i32s: [i32; 3],
1016        i64s: [i64; 3],
1017        u16: u16,
1018        f32s: [f32; 3],
1019        f64s: [f64; 3],
1020        ch: char,
1021    }
1022
1023    impl PartialEq for ScalarProbe {
1024        fn eq(&self, other: &Self) -> bool {
1025            self.b == other.b
1026                && self.i8s == other.i8s
1027                && self.i16s == other.i16s
1028                && self.i32s == other.i32s
1029                && self.i64s == other.i64s
1030                && self.u16 == other.u16
1031                && self
1032                    .f32s
1033                    .iter()
1034                    .zip(other.f32s)
1035                    .all(|(left, right)| left.to_bits() == right.to_bits())
1036                && self
1037                    .f64s
1038                    .iter()
1039                    .zip(other.f64s)
1040                    .all(|(left, right)| left.to_bits() == right.to_bits())
1041                && self.ch == other.ch
1042        }
1043    }
1044
1045    #[derive(Debug, PartialEq, Serialize, Deserialize)]
1046    struct BoolProbe {
1047        b: bool,
1048    }
1049
1050    struct TooManySeq;
1051
1052    impl Serialize for TooManySeq {
1053        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1054        where
1055            S: Serializer,
1056        {
1057            let seq = serializer.serialize_seq(Some(MAX_COLLECTION_COUNT + 1))?;
1058            SerializeSeq::end(seq)
1059        }
1060    }
1061
1062    fn roundtrip<T>(value: &T) -> T
1063    where
1064        T: Serialize + DeserializeOwned,
1065    {
1066        decode_from_slice(&encode_to_vec(value).expect("encode")).expect("decode")
1067    }
1068
1069    fn headers_populated() -> Headers {
1070        Headers {
1071            correlation_id: Some(UUID4::from_bytes([1; 16])),
1072            causation_id: Some(UUID4::from_bytes([2; 16])),
1073        }
1074    }
1075
1076    fn entry(headers: Headers) -> EventStoreEntry {
1077        let topic = "exec.command".into();
1078        let payload_type = Ustr::from("SubmitOrder");
1079        let payload = Bytes::from_static(b"\x01\x02\x03\x04");
1080        let seq = 42;
1081        let ts_init = UnixNanos::from(1_700_000_000_000_000_000);
1082        let ts_publish = UnixNanos::from(1_700_000_000_000_000_001);
1083        let entry_hash = compute_entry_hash(
1084            seq,
1085            ts_init,
1086            ts_publish,
1087            "exec.command",
1088            payload_type.as_str(),
1089            &payload,
1090            &headers,
1091        );
1092
1093        EventStoreEntry::new(
1094            entry_hash,
1095            seq,
1096            headers,
1097            topic,
1098            payload_type,
1099            payload,
1100            ts_init,
1101            ts_publish,
1102        )
1103    }
1104
1105    fn registered_components() -> RegisteredComponents {
1106        let mut components = RegisteredComponents::default();
1107        components
1108            .actors
1109            .insert("actor-1".to_string(), "hash-a".to_string());
1110        components
1111            .strategies
1112            .insert("strategy-1".to_string(), "hash-s".to_string());
1113        components
1114            .algorithms
1115            .insert("algo-1".to_string(), "hash-g".to_string());
1116        components.subscriptions.push("data.quotes".to_string());
1117        components.endpoints.push("exec.command".to_string());
1118        components
1119    }
1120
1121    fn running_manifest() -> RunManifest {
1122        RunManifest {
1123            run_id: "1700000000-abcd1234".to_string(),
1124            parent_run_id: None,
1125            instance_id: "trader-001".to_string(),
1126            binary_hash: "deadbeef".to_string(),
1127            schema_version: 1,
1128            crate_versions: "feedface".to_string(),
1129            feature_flags: Vec::new(),
1130            adapter_versions: IndexMap::new(),
1131            config_hash: "cafebabe".to_string(),
1132            registered_components: RegisteredComponents::default(),
1133            seed: None,
1134            start_ts_init: UnixNanos::from(10),
1135            end_ts_init: None,
1136            high_watermark: 0,
1137            status: RunStatus::Running,
1138        }
1139    }
1140
1141    fn sealed_manifest() -> RunManifest {
1142        let mut adapter_versions = IndexMap::new();
1143        adapter_versions.insert("binance".to_string(), "1.2.3".to_string());
1144        adapter_versions.insert("okx".to_string(), "2.3.4".to_string());
1145
1146        RunManifest {
1147            run_id: "1700000010-cafe1234".to_string(),
1148            parent_run_id: Some("1700000000-abcd1234".to_string()),
1149            instance_id: "trader-001".to_string(),
1150            binary_hash: "deadbeef".to_string(),
1151            schema_version: 2,
1152            crate_versions: "feedface".to_string(),
1153            feature_flags: vec!["live".to_string(), "persistence".to_string()],
1154            adapter_versions,
1155            config_hash: "cafebabe".to_string(),
1156            registered_components: registered_components(),
1157            seed: Some(7),
1158            start_ts_init: UnixNanos::from(10),
1159            end_ts_init: Some(UnixNanos::from(20)),
1160            high_watermark: 99,
1161            status: RunStatus::Ended,
1162        }
1163    }
1164
1165    fn snapshot_with_cursors() -> DataCursorSnapshot {
1166        DataCursorSnapshot {
1167            marker_seq: 7,
1168            event_seq_before: 42,
1169            ts_init: UnixNanos::from(100),
1170            advanced: vec![
1171                StreamCursor {
1172                    slot: 1,
1173                    ts_init_hi: UnixNanos::from(101),
1174                    count: 10,
1175                },
1176                StreamCursor {
1177                    slot: 2,
1178                    ts_init_hi: UnixNanos::from(102),
1179                    count: 11,
1180                },
1181            ],
1182        }
1183    }
1184
1185    fn hifi_marker() -> HiFiMarker {
1186        HiFiMarker {
1187            marker_seq: 1,
1188            event_seq_before: 42,
1189            slot: 3,
1190            ts_event: UnixNanos::from(1000),
1191            ts_init: UnixNanos::from(1001),
1192            same_ts_ordinal: 2,
1193            record_fingerprint: array::from_fn(|idx| {
1194                u8::try_from(idx).expect("fingerprint index is in 0..32")
1195            }),
1196        }
1197    }
1198
1199    #[rstest]
1200    #[case::empty(Headers::empty())]
1201    #[case::populated(headers_populated())]
1202    fn roundtrip_event_store_entry(#[case] headers: Headers) {
1203        let decoded = roundtrip(&entry(headers));
1204
1205        assert_eq!(decoded.recompute_hash(), decoded.entry_hash);
1206    }
1207
1208    #[rstest]
1209    #[case::running(running_manifest())]
1210    #[case::sealed(sealed_manifest())]
1211    fn roundtrip_run_manifest(#[case] manifest: RunManifest) {
1212        assert_eq!(roundtrip(&manifest), manifest);
1213    }
1214
1215    #[rstest]
1216    fn roundtrip_registered_components() {
1217        let components = registered_components();
1218
1219        assert_eq!(roundtrip(&components), components);
1220    }
1221
1222    #[rstest]
1223    fn roundtrip_snapshot_anchor() {
1224        let anchor = SnapshotAnchor::new(10, "cache://run/10", "blake3:abcd");
1225
1226        assert_eq!(roundtrip(&anchor), anchor);
1227    }
1228
1229    #[rstest]
1230    #[case::advanced(snapshot_with_cursors())]
1231    #[case::empty(DataCursorSnapshot {
1232        marker_seq: 8,
1233        event_seq_before: 43,
1234        ts_init: UnixNanos::from(101),
1235        advanced: Vec::new(),
1236    })]
1237    fn roundtrip_data_cursor_snapshot(#[case] snapshot: DataCursorSnapshot) {
1238        assert_eq!(roundtrip(&snapshot), snapshot);
1239    }
1240
1241    #[rstest]
1242    fn roundtrip_hifi_marker() {
1243        let marker = hifi_marker();
1244
1245        assert_eq!(roundtrip(&marker), marker);
1246    }
1247
1248    #[rstest]
1249    #[case(MarkerGapReason::Overflow)]
1250    #[case(MarkerGapReason::WriterClosed)]
1251    fn roundtrip_marker_gap(#[case] reason: MarkerGapReason) {
1252        let gap = MarkerGap {
1253            from_marker_seq: 1,
1254            to_marker_seq: 2,
1255            reason,
1256        };
1257
1258        assert_eq!(roundtrip(&gap), gap);
1259    }
1260
1261    #[rstest]
1262    fn roundtrip_stream_dict_entry() {
1263        let entry = StreamDictEntry {
1264            slot: 3,
1265            data_cls: DataClass::Quote,
1266            identifier: "ETHUSDT.BINANCE".to_string(),
1267        };
1268
1269        assert_eq!(roundtrip(&entry), entry);
1270    }
1271
1272    #[rstest]
1273    #[case(RunStatus::Running)]
1274    #[case(RunStatus::Ended)]
1275    #[case(RunStatus::CrashedRecovered)]
1276    #[case(RunStatus::Quarantined)]
1277    fn roundtrip_run_status_all_variants(#[case] status: RunStatus) {
1278        assert_eq!(roundtrip(&status), status);
1279    }
1280
1281    #[rstest]
1282    #[case(DataClass::BookDeltas)]
1283    #[case(DataClass::BookDepth10)]
1284    #[case(DataClass::Quote)]
1285    #[case(DataClass::Trade)]
1286    #[case(DataClass::Bar)]
1287    fn roundtrip_data_class_all_variants(#[case] data_class: DataClass) {
1288        assert_eq!(roundtrip(&data_class), data_class);
1289    }
1290
1291    #[rstest]
1292    #[case(MarkerGapReason::Overflow)]
1293    #[case(MarkerGapReason::WriterClosed)]
1294    fn roundtrip_marker_gap_reason_all_variants(#[case] reason: MarkerGapReason) {
1295        assert_eq!(roundtrip(&reason), reason);
1296    }
1297
1298    #[rstest]
1299    #[case::false_value(false)]
1300    #[case::true_value(true)]
1301    fn roundtrip_scalars(#[case] value: bool) {
1302        let probe = ScalarProbe {
1303            b: value,
1304            i8s: [i8::MIN, -1, i8::MAX],
1305            i16s: [i16::MIN, -2, i16::MAX],
1306            i32s: [i32::MIN, -3, i32::MAX],
1307            i64s: [i64::MIN, -4, i64::MAX],
1308            u16: u16::MAX,
1309            f32s: [0.0, 1.25, f32::NAN],
1310            f64s: [0.0, f64::INFINITY, f64::NAN],
1311            ch: '∞',
1312        };
1313
1314        assert_eq!(roundtrip(&probe), probe);
1315    }
1316
1317    #[rstest]
1318    fn roundtrip_zero_width_sequence() {
1319        let value = vec![(); 10];
1320
1321        assert_eq!(roundtrip(&value), value);
1322    }
1323
1324    #[rstest]
1325    fn encode_is_deterministic() {
1326        let entry = entry(headers_populated());
1327        let manifest = sealed_manifest();
1328
1329        assert_eq!(
1330            encode_to_vec(&entry).unwrap(),
1331            encode_to_vec(&entry).unwrap()
1332        );
1333        assert_eq!(
1334            encode_to_vec(&manifest).unwrap(),
1335            encode_to_vec(&manifest).unwrap()
1336        );
1337    }
1338
1339    #[rstest]
1340    fn data_type_roundtrip_preserves_fields_and_repairs_hash() {
1341        // Absent metadata isolates the sequence framing under test. Params stores
1342        // serde_json::Value, whose Deserialize requires deserialize_any, so a non-empty
1343        // Params cannot be decoded by this codec at all - a pre-existing Params limitation
1344        // unrelated to the DataType visitor.
1345        let expected = DataType::new("ExampleType", None, Some("catalog/path".to_string()));
1346        let expected_hash = expected.precomputed_hash();
1347        let mut bytes = encode_to_vec(&expected).unwrap();
1348
1349        let hash_offset = HEADER_LEN
1350            + size_of::<u64>()
1351            + expected.type_name().len()
1352            + size_of::<u8>()
1353            + size_of::<u64>()
1354            + expected.topic().len();
1355        bytes[hash_offset..hash_offset + size_of::<u64>()]
1356            .copy_from_slice(&(expected_hash ^ u64::MAX).to_le_bytes());
1357
1358        let roundtripped: DataType = decode_from_slice(&bytes).unwrap();
1359
1360        assert_eq!(roundtripped.type_name(), expected.type_name());
1361        assert_eq!(roundtripped.metadata(), expected.metadata());
1362        assert_eq!(roundtripped.topic(), expected.topic());
1363        assert_eq!(roundtripped.identifier(), expected.identifier());
1364        assert_eq!(roundtripped.precomputed_hash(), expected_hash);
1365    }
1366
1367    #[rstest]
1368    fn header_present_and_correct() {
1369        let bytes = encode_to_vec(&RunStatus::Running).unwrap();
1370
1371        assert_eq!(&bytes[..4], b"NESC");
1372        assert_eq!(bytes[4], 1);
1373    }
1374
1375    #[rstest]
1376    fn header_bad_magic_rejected() {
1377        let mut bytes = encode_to_vec(&entry(Headers::empty())).unwrap();
1378        bytes[0] ^= 0xFF;
1379
1380        assert!(matches!(
1381            decode_from_slice::<EventStoreEntry>(&bytes),
1382            Err(CodecError::BadMagic)
1383        ));
1384    }
1385
1386    #[rstest]
1387    fn header_unsupported_version_rejected() {
1388        let mut bytes = encode_to_vec(&entry(Headers::empty())).unwrap();
1389        bytes[4] = 2;
1390
1391        assert!(matches!(
1392            decode_from_slice::<EventStoreEntry>(&bytes),
1393            Err(CodecError::UnsupportedVersion(2))
1394        ));
1395    }
1396
1397    #[rstest]
1398    fn header_truncated_rejected() {
1399        let bytes = encode_to_vec(&entry(Headers::empty())).unwrap();
1400
1401        assert!(matches!(
1402            decode_from_slice::<EventStoreEntry>(&bytes[..3]),
1403            Err(CodecError::UnexpectedEof { .. })
1404        ));
1405    }
1406
1407    #[rstest]
1408    fn decodes_a_bincode_blob_as_bad_magic() {
1409        // bincode standard() encoding of String "old-format", frozen so the test needs
1410        // no bincode dependency: varint len 10 (0x0A) ++ b"old-format".
1411        const OLD_BINCODE_STRING: &[u8] = b"\x0aold-format";
1412
1413        assert!(matches!(
1414            decode_from_slice::<String>(OLD_BINCODE_STRING),
1415            Err(CodecError::BadMagic)
1416        ));
1417    }
1418
1419    #[rstest]
1420    fn truncated_body_rejected() {
1421        let bytes = encode_to_vec(&entry(Headers::empty())).unwrap();
1422
1423        assert!(decode_from_slice::<EventStoreEntry>(&bytes[..bytes.len() - 1]).is_err());
1424    }
1425
1426    #[rstest]
1427    fn forged_length_prefix_rejected() {
1428        let mut bytes = encode_to_vec(&"ok".to_string()).unwrap();
1429        bytes[HEADER_LEN..HEADER_LEN + 8].copy_from_slice(&u64::MAX.to_le_bytes());
1430
1431        assert!(matches!(
1432            decode_from_slice::<String>(&bytes),
1433            Err(CodecError::LengthOverflow { .. })
1434        ));
1435    }
1436
1437    #[rstest]
1438    fn forged_sequence_count_rejected_on_eof() {
1439        let mut bytes = encode_to_vec(&vec![1_u8, 2, 3]).unwrap();
1440        bytes[HEADER_LEN..HEADER_LEN + 8].copy_from_slice(&99_u64.to_le_bytes());
1441
1442        assert!(matches!(
1443            decode_from_slice::<Vec<u8>>(&bytes),
1444            Err(CodecError::UnexpectedEof { .. })
1445        ));
1446    }
1447
1448    #[rstest]
1449    fn collection_count_overflow_rejected() {
1450        let mut bytes = encode_to_vec(&vec![1_u8, 2, 3]).unwrap();
1451        let too_many = (MAX_COLLECTION_COUNT as u64) + 1;
1452        bytes[HEADER_LEN..HEADER_LEN + 8].copy_from_slice(&too_many.to_le_bytes());
1453
1454        assert!(matches!(
1455            decode_from_slice::<Vec<u8>>(&bytes),
1456            Err(CodecError::CountOverflow {
1457                claimed,
1458                max: MAX_COLLECTION_COUNT
1459            }) if claimed == too_many
1460        ));
1461    }
1462
1463    #[rstest]
1464    fn collection_count_overflow_rejected_on_encode() {
1465        assert!(matches!(
1466            encode_to_vec(&TooManySeq),
1467            Err(CodecError::CountOverflow {
1468                claimed,
1469                max: MAX_COLLECTION_COUNT
1470            }) if claimed == (MAX_COLLECTION_COUNT as u64) + 1
1471        ));
1472    }
1473
1474    #[rstest]
1475    fn bad_bool_discriminant_rejected() {
1476        let mut bytes = encode_to_vec(&BoolProbe { b: false }).unwrap();
1477        bytes[HEADER_LEN] = 2;
1478
1479        assert!(matches!(
1480            decode_from_slice::<BoolProbe>(&bytes),
1481            Err(CodecError::InvalidBool(2))
1482        ));
1483    }
1484
1485    #[rstest]
1486    fn bad_option_discriminant_rejected() {
1487        let mut bytes = encode_to_vec(&Headers::empty()).unwrap();
1488        bytes[HEADER_LEN] = 2;
1489
1490        assert!(matches!(
1491            decode_from_slice::<Headers>(&bytes),
1492            Err(CodecError::InvalidOption(2))
1493        ));
1494    }
1495
1496    #[rstest]
1497    fn invalid_utf8_rejected() {
1498        let mut bytes = encode_to_vec(&"ok".to_string()).unwrap();
1499        bytes[HEADER_LEN + 8] = 0xFF;
1500
1501        assert!(matches!(
1502            decode_from_slice::<String>(&bytes),
1503            Err(CodecError::InvalidUtf8)
1504        ));
1505    }
1506
1507    #[rstest]
1508    fn unknown_enum_variant_rejected() {
1509        let mut bytes = encode_to_vec(&RunStatus::Running).unwrap();
1510        bytes[HEADER_LEN..HEADER_LEN + 4].copy_from_slice(&99_u32.to_le_bytes());
1511
1512        assert!(matches!(
1513            decode_from_slice::<RunStatus>(&bytes),
1514            Err(CodecError::UnknownVariant(99))
1515        ));
1516    }
1517
1518    #[rstest]
1519    fn trailing_bytes_rejected() {
1520        let mut bytes = encode_to_vec(&RunStatus::Running).unwrap();
1521        bytes.push(0xFF);
1522
1523        assert!(matches!(
1524            decode_from_slice::<RunStatus>(&bytes),
1525            Err(CodecError::TrailingBytes(1))
1526        ));
1527    }
1528
1529    #[rstest]
1530    fn rejects_self_describing() {
1531        #[allow(dead_code)]
1532        #[derive(Debug, Deserialize)]
1533        #[serde(untagged)]
1534        enum Probe {
1535            A(u64),
1536            B(String),
1537        }
1538
1539        let bytes = encode_to_vec(&1_u64).unwrap();
1540        let err = decode_from_slice::<Probe>(&bytes).expect_err("untagged enum must reject any");
1541        assert!(matches!(err, CodecError::SelfDescribing));
1542    }
1543
1544    #[rstest]
1545    fn serialize_without_known_len_is_rejected() {
1546        let mut encoder = Encoder { out: Vec::new() };
1547        assert!(matches!(
1548            ser::Serializer::serialize_seq(&mut encoder, None),
1549            Err(CodecError::MissingLen)
1550        ));
1551
1552        let mut encoder = Encoder { out: Vec::new() };
1553        assert!(matches!(
1554            ser::Serializer::serialize_map(&mut encoder, None),
1555            Err(CodecError::MissingLen)
1556        ));
1557    }
1558
1559    struct U32Visitor;
1560
1561    impl Visitor<'_> for U32Visitor {
1562        type Value = u32;
1563
1564        fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1565            formatter.write_str("a little-endian u32 identifier")
1566        }
1567
1568        fn visit_u32<E>(self, value: u32) -> Result<Self::Value, E>
1569        where
1570            E: de::Error,
1571        {
1572            Ok(value)
1573        }
1574    }
1575
1576    #[rstest]
1577    fn deserialize_identifier_reads_u32() {
1578        // `deserialize_identifier` is implemented for totality only (struct
1579        // fields decode positionally and variant indices flow through serde's
1580        // `U32Deserializer`, so the derive never reaches it). Drive it directly
1581        // to prove the laid brick: it forwards to the little-endian `u32` reader.
1582        let body = 7_u32.to_le_bytes();
1583        let mut decoder = Decoder {
1584            input: &body,
1585            pos: 0,
1586        };
1587
1588        let value = de::Deserializer::deserialize_identifier(&mut decoder, U32Visitor)
1589            .expect("identifier decodes as a u32");
1590
1591        assert_eq!(value, 7);
1592        assert_eq!(decoder.pos, body.len());
1593    }
1594
1595    #[rstest]
1596    fn deserialize_ignored_any_rejected() {
1597        // `deserialize_ignored_any` is implemented for totality only and is hard
1598        // to reach via the positional model, but it carries the same
1599        // self-describing-rejection load as `deserialize_any` (the `wire.rs`
1600        // invariant). `IgnoredAny::deserialize` drives it directly; assert it
1601        // rejects rather than silently skipping.
1602        let mut decoder = Decoder { input: &[], pos: 0 };
1603
1604        let err = de::IgnoredAny::deserialize(&mut decoder)
1605            .expect_err("ignored_any must reject as self-describing");
1606
1607        assert!(matches!(err, CodecError::SelfDescribing));
1608    }
1609
1610    #[rstest]
1611    fn entry_hash_newtype_roundtrips() {
1612        let hash = EntryHash(array::from_fn(|idx| {
1613            u8::try_from(idx).expect("hash index is in 0..32")
1614        }));
1615
1616        assert_eq!(roundtrip(&hash), hash);
1617    }
1618
1619    proptest! {
1620        #![proptest_config(ProptestConfig { cases: 64, ..ProptestConfig::default() })]
1621
1622        #[rstest]
1623        fn prop_roundtrip_data_cursor_snapshot(
1624            marker_seq in any::<u64>(),
1625            event_seq_before in any::<u64>(),
1626            ts_init in any::<u64>(),
1627            cursors in proptest::collection::vec((any::<u32>(), any::<u64>(), any::<u64>()), 0..8),
1628        ) {
1629            let snap = DataCursorSnapshot {
1630                marker_seq,
1631                event_seq_before,
1632                ts_init: UnixNanos::from(ts_init),
1633                advanced: cursors
1634                    .into_iter()
1635                    .map(|(slot, hi, count)| StreamCursor {
1636                        slot,
1637                        ts_init_hi: UnixNanos::from(hi),
1638                        count,
1639                    })
1640                    .collect(),
1641            };
1642
1643            let bytes = encode_to_vec(&snap).expect("encode");
1644            let decoded: DataCursorSnapshot = decode_from_slice(&bytes).expect("decode");
1645            prop_assert_eq!(snap, decoded);
1646        }
1647
1648        #[rstest]
1649        fn prop_roundtrip_hifi_marker(
1650            marker_seq in any::<u64>(),
1651            event_seq_before in any::<u64>(),
1652            slot in any::<u32>(),
1653            ts_event in any::<u64>(),
1654            ts_init in any::<u64>(),
1655            same_ts_ordinal in any::<u32>(),
1656            fingerprint in proptest::array::uniform32(any::<u8>()),
1657        ) {
1658            let marker = HiFiMarker {
1659                marker_seq,
1660                event_seq_before,
1661                slot,
1662                ts_event: UnixNanos::from(ts_event),
1663                ts_init: UnixNanos::from(ts_init),
1664                same_ts_ordinal,
1665                record_fingerprint: fingerprint,
1666            };
1667
1668            let bytes = encode_to_vec(&marker).expect("encode");
1669            let decoded: HiFiMarker = decode_from_slice(&bytes).expect("decode");
1670            prop_assert_eq!(marker, decoded);
1671        }
1672    }
1673}
nautilus_event_store/codec.rs

nautilus_event_store/
codec.rs
