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_system::RegisteredComponents;
994    use proptest::{prelude::*, test_runner::Config as ProptestConfig};
995    use rstest::rstest;
996    use serde::{Deserialize, Serializer};
997    use ustr::Ustr;
998
999    use super::*;
1000    use crate::{
1001        EventStoreEntry, Headers, RunManifest, RunStatus, SnapshotAnchor,
1002        hash::{EntryHash, compute_entry_hash},
1003        markers::{
1004            DataClass, DataCursorSnapshot, HiFiMarker, MarkerGap, MarkerGapReason, StreamCursor,
1005            StreamDictEntry,
1006        },
1007    };
1008
1009    #[derive(Debug, Serialize, Deserialize)]
1010    struct ScalarProbe {
1011        b: bool,
1012        i8s: [i8; 3],
1013        i16s: [i16; 3],
1014        i32s: [i32; 3],
1015        i64s: [i64; 3],
1016        u16: u16,
1017        f32s: [f32; 3],
1018        f64s: [f64; 3],
1019        ch: char,
1020    }
1021
1022    impl PartialEq for ScalarProbe {
1023        fn eq(&self, other: &Self) -> bool {
1024            self.b == other.b
1025                && self.i8s == other.i8s
1026                && self.i16s == other.i16s
1027                && self.i32s == other.i32s
1028                && self.i64s == other.i64s
1029                && self.u16 == other.u16
1030                && self
1031                    .f32s
1032                    .iter()
1033                    .zip(other.f32s)
1034                    .all(|(left, right)| left.to_bits() == right.to_bits())
1035                && self
1036                    .f64s
1037                    .iter()
1038                    .zip(other.f64s)
1039                    .all(|(left, right)| left.to_bits() == right.to_bits())
1040                && self.ch == other.ch
1041        }
1042    }
1043
1044    #[derive(Debug, PartialEq, Serialize, Deserialize)]
1045    struct BoolProbe {
1046        b: bool,
1047    }
1048
1049    struct TooManySeq;
1050
1051    impl Serialize for TooManySeq {
1052        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1053        where
1054            S: Serializer,
1055        {
1056            let seq = serializer.serialize_seq(Some(MAX_COLLECTION_COUNT + 1))?;
1057            SerializeSeq::end(seq)
1058        }
1059    }
1060
1061    fn roundtrip<T>(value: &T) -> T
1062    where
1063        T: Serialize + DeserializeOwned,
1064    {
1065        decode_from_slice(&encode_to_vec(value).expect("encode")).expect("decode")
1066    }
1067
1068    fn headers_populated() -> Headers {
1069        Headers {
1070            correlation_id: Some(UUID4::from_bytes([1; 16])),
1071            causation_id: Some(UUID4::from_bytes([2; 16])),
1072        }
1073    }
1074
1075    fn entry(headers: Headers) -> EventStoreEntry {
1076        let topic = "exec.command".into();
1077        let payload_type = Ustr::from("SubmitOrder");
1078        let payload = Bytes::from_static(b"\x01\x02\x03\x04");
1079        let seq = 42;
1080        let ts_init = UnixNanos::from(1_700_000_000_000_000_000);
1081        let ts_publish = UnixNanos::from(1_700_000_000_000_000_001);
1082        let entry_hash = compute_entry_hash(
1083            seq,
1084            ts_init,
1085            ts_publish,
1086            "exec.command",
1087            payload_type.as_str(),
1088            &payload,
1089            &headers,
1090        );
1091
1092        EventStoreEntry::new(
1093            entry_hash,
1094            seq,
1095            headers,
1096            topic,
1097            payload_type,
1098            payload,
1099            ts_init,
1100            ts_publish,
1101        )
1102    }
1103
1104    fn registered_components() -> RegisteredComponents {
1105        let mut components = RegisteredComponents::default();
1106        components
1107            .actors
1108            .insert("actor-1".to_string(), "hash-a".to_string());
1109        components
1110            .strategies
1111            .insert("strategy-1".to_string(), "hash-s".to_string());
1112        components
1113            .algorithms
1114            .insert("algo-1".to_string(), "hash-g".to_string());
1115        components.subscriptions.push("data.quotes".to_string());
1116        components.endpoints.push("exec.command".to_string());
1117        components
1118    }
1119
1120    fn running_manifest() -> RunManifest {
1121        RunManifest {
1122            run_id: "1700000000-abcd1234".to_string(),
1123            parent_run_id: None,
1124            instance_id: "trader-001".to_string(),
1125            binary_hash: "deadbeef".to_string(),
1126            schema_version: 1,
1127            crate_versions: "feedface".to_string(),
1128            feature_flags: Vec::new(),
1129            adapter_versions: IndexMap::new(),
1130            config_hash: "cafebabe".to_string(),
1131            registered_components: RegisteredComponents::default(),
1132            seed: None,
1133            start_ts_init: UnixNanos::from(10),
1134            end_ts_init: None,
1135            high_watermark: 0,
1136            status: RunStatus::Running,
1137        }
1138    }
1139
1140    fn sealed_manifest() -> RunManifest {
1141        let mut adapter_versions = IndexMap::new();
1142        adapter_versions.insert("binance".to_string(), "1.2.3".to_string());
1143        adapter_versions.insert("okx".to_string(), "2.3.4".to_string());
1144
1145        RunManifest {
1146            run_id: "1700000010-cafe1234".to_string(),
1147            parent_run_id: Some("1700000000-abcd1234".to_string()),
1148            instance_id: "trader-001".to_string(),
1149            binary_hash: "deadbeef".to_string(),
1150            schema_version: 2,
1151            crate_versions: "feedface".to_string(),
1152            feature_flags: vec!["live".to_string(), "persistence".to_string()],
1153            adapter_versions,
1154            config_hash: "cafebabe".to_string(),
1155            registered_components: registered_components(),
1156            seed: Some(7),
1157            start_ts_init: UnixNanos::from(10),
1158            end_ts_init: Some(UnixNanos::from(20)),
1159            high_watermark: 99,
1160            status: RunStatus::Ended,
1161        }
1162    }
1163
1164    fn snapshot_with_cursors() -> DataCursorSnapshot {
1165        DataCursorSnapshot {
1166            marker_seq: 7,
1167            event_seq_before: 42,
1168            ts_init: UnixNanos::from(100),
1169            advanced: vec![
1170                StreamCursor {
1171                    slot: 1,
1172                    ts_init_hi: UnixNanos::from(101),
1173                    count: 10,
1174                },
1175                StreamCursor {
1176                    slot: 2,
1177                    ts_init_hi: UnixNanos::from(102),
1178                    count: 11,
1179                },
1180            ],
1181        }
1182    }
1183
1184    fn hifi_marker() -> HiFiMarker {
1185        HiFiMarker {
1186            marker_seq: 1,
1187            event_seq_before: 42,
1188            slot: 3,
1189            ts_event: UnixNanos::from(1000),
1190            ts_init: UnixNanos::from(1001),
1191            same_ts_ordinal: 2,
1192            record_fingerprint: array::from_fn(|idx| {
1193                u8::try_from(idx).expect("fingerprint index is in 0..32")
1194            }),
1195        }
1196    }
1197
1198    #[rstest]
1199    #[case::empty(Headers::empty())]
1200    #[case::populated(headers_populated())]
1201    fn roundtrip_event_store_entry(#[case] headers: Headers) {
1202        let decoded = roundtrip(&entry(headers));
1203
1204        assert_eq!(decoded.recompute_hash(), decoded.entry_hash);
1205    }
1206
1207    #[rstest]
1208    #[case::running(running_manifest())]
1209    #[case::sealed(sealed_manifest())]
1210    fn roundtrip_run_manifest(#[case] manifest: RunManifest) {
1211        assert_eq!(roundtrip(&manifest), manifest);
1212    }
1213
1214    #[rstest]
1215    fn roundtrip_registered_components() {
1216        let components = registered_components();
1217
1218        assert_eq!(roundtrip(&components), components);
1219    }
1220
1221    #[rstest]
1222    fn roundtrip_snapshot_anchor() {
1223        let anchor = SnapshotAnchor::new(10, "cache://run/10", "blake3:abcd");
1224
1225        assert_eq!(roundtrip(&anchor), anchor);
1226    }
1227
1228    #[rstest]
1229    #[case::advanced(snapshot_with_cursors())]
1230    #[case::empty(DataCursorSnapshot {
1231        marker_seq: 8,
1232        event_seq_before: 43,
1233        ts_init: UnixNanos::from(101),
1234        advanced: Vec::new(),
1235    })]
1236    fn roundtrip_data_cursor_snapshot(#[case] snapshot: DataCursorSnapshot) {
1237        assert_eq!(roundtrip(&snapshot), snapshot);
1238    }
1239
1240    #[rstest]
1241    fn roundtrip_hifi_marker() {
1242        let marker = hifi_marker();
1243
1244        assert_eq!(roundtrip(&marker), marker);
1245    }
1246
1247    #[rstest]
1248    #[case(MarkerGapReason::Overflow)]
1249    #[case(MarkerGapReason::WriterClosed)]
1250    fn roundtrip_marker_gap(#[case] reason: MarkerGapReason) {
1251        let gap = MarkerGap {
1252            from_marker_seq: 1,
1253            to_marker_seq: 2,
1254            reason,
1255        };
1256
1257        assert_eq!(roundtrip(&gap), gap);
1258    }
1259
1260    #[rstest]
1261    fn roundtrip_stream_dict_entry() {
1262        let entry = StreamDictEntry {
1263            slot: 3,
1264            data_cls: DataClass::Quote,
1265            identifier: "ETHUSDT.BINANCE".to_string(),
1266        };
1267
1268        assert_eq!(roundtrip(&entry), entry);
1269    }
1270
1271    #[rstest]
1272    #[case(RunStatus::Running)]
1273    #[case(RunStatus::Ended)]
1274    #[case(RunStatus::CrashedRecovered)]
1275    #[case(RunStatus::Quarantined)]
1276    fn roundtrip_run_status_all_variants(#[case] status: RunStatus) {
1277        assert_eq!(roundtrip(&status), status);
1278    }
1279
1280    #[rstest]
1281    #[case(DataClass::BookDeltas)]
1282    #[case(DataClass::BookDepth10)]
1283    #[case(DataClass::Quote)]
1284    #[case(DataClass::Trade)]
1285    #[case(DataClass::Bar)]
1286    fn roundtrip_data_class_all_variants(#[case] data_class: DataClass) {
1287        assert_eq!(roundtrip(&data_class), data_class);
1288    }
1289
1290    #[rstest]
1291    #[case(MarkerGapReason::Overflow)]
1292    #[case(MarkerGapReason::WriterClosed)]
1293    fn roundtrip_marker_gap_reason_all_variants(#[case] reason: MarkerGapReason) {
1294        assert_eq!(roundtrip(&reason), reason);
1295    }
1296
1297    #[rstest]
1298    #[case::false_value(false)]
1299    #[case::true_value(true)]
1300    fn roundtrip_scalars(#[case] value: bool) {
1301        let probe = ScalarProbe {
1302            b: value,
1303            i8s: [i8::MIN, -1, i8::MAX],
1304            i16s: [i16::MIN, -2, i16::MAX],
1305            i32s: [i32::MIN, -3, i32::MAX],
1306            i64s: [i64::MIN, -4, i64::MAX],
1307            u16: u16::MAX,
1308            f32s: [0.0, 1.25, f32::NAN],
1309            f64s: [0.0, f64::INFINITY, f64::NAN],
1310            ch: '∞',
1311        };
1312
1313        assert_eq!(roundtrip(&probe), probe);
1314    }
1315
1316    #[rstest]
1317    fn roundtrip_zero_width_sequence() {
1318        let value = vec![(); 10];
1319
1320        assert_eq!(roundtrip(&value), value);
1321    }
1322
1323    #[rstest]
1324    fn encode_is_deterministic() {
1325        let entry = entry(headers_populated());
1326        let manifest = sealed_manifest();
1327
1328        assert_eq!(
1329            encode_to_vec(&entry).unwrap(),
1330            encode_to_vec(&entry).unwrap()
1331        );
1332        assert_eq!(
1333            encode_to_vec(&manifest).unwrap(),
1334            encode_to_vec(&manifest).unwrap()
1335        );
1336    }
1337
1338    #[rstest]
1339    fn header_present_and_correct() {
1340        let bytes = encode_to_vec(&RunStatus::Running).unwrap();
1341
1342        assert_eq!(&bytes[..4], b"NESC");
1343        assert_eq!(bytes[4], 1);
1344    }
1345
1346    #[rstest]
1347    fn header_bad_magic_rejected() {
1348        let mut bytes = encode_to_vec(&entry(Headers::empty())).unwrap();
1349        bytes[0] ^= 0xFF;
1350
1351        assert!(matches!(
1352            decode_from_slice::<EventStoreEntry>(&bytes),
1353            Err(CodecError::BadMagic)
1354        ));
1355    }
1356
1357    #[rstest]
1358    fn header_unsupported_version_rejected() {
1359        let mut bytes = encode_to_vec(&entry(Headers::empty())).unwrap();
1360        bytes[4] = 2;
1361
1362        assert!(matches!(
1363            decode_from_slice::<EventStoreEntry>(&bytes),
1364            Err(CodecError::UnsupportedVersion(2))
1365        ));
1366    }
1367
1368    #[rstest]
1369    fn header_truncated_rejected() {
1370        let bytes = encode_to_vec(&entry(Headers::empty())).unwrap();
1371
1372        assert!(matches!(
1373            decode_from_slice::<EventStoreEntry>(&bytes[..3]),
1374            Err(CodecError::UnexpectedEof { .. })
1375        ));
1376    }
1377
1378    #[rstest]
1379    fn decodes_a_bincode_blob_as_bad_magic() {
1380        // bincode standard() encoding of String "old-format", frozen so the test needs
1381        // no bincode dependency: varint len 10 (0x0A) ++ b"old-format".
1382        const OLD_BINCODE_STRING: &[u8] = b"\x0aold-format";
1383
1384        assert!(matches!(
1385            decode_from_slice::<String>(OLD_BINCODE_STRING),
1386            Err(CodecError::BadMagic)
1387        ));
1388    }
1389
1390    #[rstest]
1391    fn truncated_body_rejected() {
1392        let bytes = encode_to_vec(&entry(Headers::empty())).unwrap();
1393
1394        assert!(decode_from_slice::<EventStoreEntry>(&bytes[..bytes.len() - 1]).is_err());
1395    }
1396
1397    #[rstest]
1398    fn forged_length_prefix_rejected() {
1399        let mut bytes = encode_to_vec(&"ok".to_string()).unwrap();
1400        bytes[HEADER_LEN..HEADER_LEN + 8].copy_from_slice(&u64::MAX.to_le_bytes());
1401
1402        assert!(matches!(
1403            decode_from_slice::<String>(&bytes),
1404            Err(CodecError::LengthOverflow { .. })
1405        ));
1406    }
1407
1408    #[rstest]
1409    fn forged_sequence_count_rejected_on_eof() {
1410        let mut bytes = encode_to_vec(&vec![1_u8, 2, 3]).unwrap();
1411        bytes[HEADER_LEN..HEADER_LEN + 8].copy_from_slice(&99_u64.to_le_bytes());
1412
1413        assert!(matches!(
1414            decode_from_slice::<Vec<u8>>(&bytes),
1415            Err(CodecError::UnexpectedEof { .. })
1416        ));
1417    }
1418
1419    #[rstest]
1420    fn collection_count_overflow_rejected() {
1421        let mut bytes = encode_to_vec(&vec![1_u8, 2, 3]).unwrap();
1422        let too_many = (MAX_COLLECTION_COUNT as u64) + 1;
1423        bytes[HEADER_LEN..HEADER_LEN + 8].copy_from_slice(&too_many.to_le_bytes());
1424
1425        assert!(matches!(
1426            decode_from_slice::<Vec<u8>>(&bytes),
1427            Err(CodecError::CountOverflow {
1428                claimed,
1429                max: MAX_COLLECTION_COUNT
1430            }) if claimed == too_many
1431        ));
1432    }
1433
1434    #[rstest]
1435    fn collection_count_overflow_rejected_on_encode() {
1436        assert!(matches!(
1437            encode_to_vec(&TooManySeq),
1438            Err(CodecError::CountOverflow {
1439                claimed,
1440                max: MAX_COLLECTION_COUNT
1441            }) if claimed == (MAX_COLLECTION_COUNT as u64) + 1
1442        ));
1443    }
1444
1445    #[rstest]
1446    fn bad_bool_discriminant_rejected() {
1447        let mut bytes = encode_to_vec(&BoolProbe { b: false }).unwrap();
1448        bytes[HEADER_LEN] = 2;
1449
1450        assert!(matches!(
1451            decode_from_slice::<BoolProbe>(&bytes),
1452            Err(CodecError::InvalidBool(2))
1453        ));
1454    }
1455
1456    #[rstest]
1457    fn bad_option_discriminant_rejected() {
1458        let mut bytes = encode_to_vec(&Headers::empty()).unwrap();
1459        bytes[HEADER_LEN] = 2;
1460
1461        assert!(matches!(
1462            decode_from_slice::<Headers>(&bytes),
1463            Err(CodecError::InvalidOption(2))
1464        ));
1465    }
1466
1467    #[rstest]
1468    fn invalid_utf8_rejected() {
1469        let mut bytes = encode_to_vec(&"ok".to_string()).unwrap();
1470        bytes[HEADER_LEN + 8] = 0xFF;
1471
1472        assert!(matches!(
1473            decode_from_slice::<String>(&bytes),
1474            Err(CodecError::InvalidUtf8)
1475        ));
1476    }
1477
1478    #[rstest]
1479    fn unknown_enum_variant_rejected() {
1480        let mut bytes = encode_to_vec(&RunStatus::Running).unwrap();
1481        bytes[HEADER_LEN..HEADER_LEN + 4].copy_from_slice(&99_u32.to_le_bytes());
1482
1483        assert!(matches!(
1484            decode_from_slice::<RunStatus>(&bytes),
1485            Err(CodecError::UnknownVariant(99))
1486        ));
1487    }
1488
1489    #[rstest]
1490    fn trailing_bytes_rejected() {
1491        let mut bytes = encode_to_vec(&RunStatus::Running).unwrap();
1492        bytes.push(0xFF);
1493
1494        assert!(matches!(
1495            decode_from_slice::<RunStatus>(&bytes),
1496            Err(CodecError::TrailingBytes(1))
1497        ));
1498    }
1499
1500    #[rstest]
1501    fn rejects_self_describing() {
1502        #[allow(dead_code)]
1503        #[derive(Debug, Deserialize)]
1504        #[serde(untagged)]
1505        enum Probe {
1506            A(u64),
1507            B(String),
1508        }
1509
1510        let bytes = encode_to_vec(&1_u64).unwrap();
1511        let err = decode_from_slice::<Probe>(&bytes).expect_err("untagged enum must reject any");
1512        assert!(matches!(err, CodecError::SelfDescribing));
1513    }
1514
1515    #[rstest]
1516    fn serialize_without_known_len_is_rejected() {
1517        let mut encoder = Encoder { out: Vec::new() };
1518        assert!(matches!(
1519            ser::Serializer::serialize_seq(&mut encoder, None),
1520            Err(CodecError::MissingLen)
1521        ));
1522
1523        let mut encoder = Encoder { out: Vec::new() };
1524        assert!(matches!(
1525            ser::Serializer::serialize_map(&mut encoder, None),
1526            Err(CodecError::MissingLen)
1527        ));
1528    }
1529
1530    struct U32Visitor;
1531
1532    impl Visitor<'_> for U32Visitor {
1533        type Value = u32;
1534
1535        fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1536            formatter.write_str("a little-endian u32 identifier")
1537        }
1538
1539        fn visit_u32<E>(self, value: u32) -> Result<Self::Value, E>
1540        where
1541            E: de::Error,
1542        {
1543            Ok(value)
1544        }
1545    }
1546
1547    #[rstest]
1548    fn deserialize_identifier_reads_u32() {
1549        // `deserialize_identifier` is implemented for totality only (struct
1550        // fields decode positionally and variant indices flow through serde's
1551        // `U32Deserializer`, so the derive never reaches it). Drive it directly
1552        // to prove the laid brick: it forwards to the little-endian `u32` reader.
1553        let body = 7_u32.to_le_bytes();
1554        let mut decoder = Decoder {
1555            input: &body,
1556            pos: 0,
1557        };
1558
1559        let value = de::Deserializer::deserialize_identifier(&mut decoder, U32Visitor)
1560            .expect("identifier decodes as a u32");
1561
1562        assert_eq!(value, 7);
1563        assert_eq!(decoder.pos, body.len());
1564    }
1565
1566    #[rstest]
1567    fn deserialize_ignored_any_rejected() {
1568        // `deserialize_ignored_any` is implemented for totality only and is hard
1569        // to reach via the positional model, but it carries the same
1570        // self-describing-rejection load as `deserialize_any` (the `wire.rs`
1571        // invariant). `IgnoredAny::deserialize` drives it directly; assert it
1572        // rejects rather than silently skipping.
1573        let mut decoder = Decoder { input: &[], pos: 0 };
1574
1575        let err = de::IgnoredAny::deserialize(&mut decoder)
1576            .expect_err("ignored_any must reject as self-describing");
1577
1578        assert!(matches!(err, CodecError::SelfDescribing));
1579    }
1580
1581    #[rstest]
1582    fn entry_hash_newtype_roundtrips() {
1583        let hash = EntryHash(array::from_fn(|idx| {
1584            u8::try_from(idx).expect("hash index is in 0..32")
1585        }));
1586
1587        assert_eq!(roundtrip(&hash), hash);
1588    }
1589
1590    proptest! {
1591        #![proptest_config(ProptestConfig { cases: 64, ..ProptestConfig::default() })]
1592
1593        #[rstest]
1594        fn prop_roundtrip_data_cursor_snapshot(
1595            marker_seq in any::<u64>(),
1596            event_seq_before in any::<u64>(),
1597            ts_init in any::<u64>(),
1598            cursors in proptest::collection::vec((any::<u32>(), any::<u64>(), any::<u64>()), 0..8),
1599        ) {
1600            let snap = DataCursorSnapshot {
1601                marker_seq,
1602                event_seq_before,
1603                ts_init: UnixNanos::from(ts_init),
1604                advanced: cursors
1605                    .into_iter()
1606                    .map(|(slot, hi, count)| StreamCursor {
1607                        slot,
1608                        ts_init_hi: UnixNanos::from(hi),
1609                        count,
1610                    })
1611                    .collect(),
1612            };
1613
1614            let bytes = encode_to_vec(&snap).expect("encode");
1615            let decoded: DataCursorSnapshot = decode_from_slice(&bytes).expect("decode");
1616            prop_assert_eq!(snap, decoded);
1617        }
1618
1619        #[rstest]
1620        fn prop_roundtrip_hifi_marker(
1621            marker_seq in any::<u64>(),
1622            event_seq_before in any::<u64>(),
1623            slot in any::<u32>(),
1624            ts_event in any::<u64>(),
1625            ts_init in any::<u64>(),
1626            same_ts_ordinal in any::<u32>(),
1627            fingerprint in proptest::array::uniform32(any::<u8>()),
1628        ) {
1629            let marker = HiFiMarker {
1630                marker_seq,
1631                event_seq_before,
1632                slot,
1633                ts_event: UnixNanos::from(ts_event),
1634                ts_init: UnixNanos::from(ts_init),
1635                same_ts_ordinal,
1636                record_fingerprint: fingerprint,
1637            };
1638
1639            let bytes = encode_to_vec(&marker).expect("encode");
1640            let decoded: HiFiMarker = decode_from_slice(&bytes).expect("decode");
1641            prop_assert_eq!(marker, decoded);
1642        }
1643    }
1644}
nautilus_event_store/codec.rs

nautilus_event_store/
codec.rs
