etl_core/
deser.rs

1//! Deserialization contract: one borrowed payload in, 0..N records out,
2//! push-style.
3//!
4//! The [`RecFamily`] lifetime→type family is what lets a
5//! lifetime-parameterized record type cross generic and dyn boundaries
6//! (validated by the seam prototype — see `docs/DESIGN.md` § Frozen v1
7//! contracts). Deserializers for borrowing record types implement
8//! `Deserializer<F>` for a family `F` whose `Rec<'buf>` borrows the payload
9//! buffer; owned record types use the provided [`Owned`] family.
10
11use crate::checkpoint::AckRef;
12use crate::error::DeserError;
13use crate::record::{Flow, RawPayload, Record};
14
15/// Maps a payload-buffer lifetime to the deserialized record type.
16///
17/// The family tag itself is `'static` — it is a type-level function, never
18/// instantiated — which is what keeps chains nameable and `Send` while the
19/// records they process borrow ephemeral buffers.
20pub trait RecFamily: 'static {
21    /// The record type produced for payloads borrowing `'buf`.
22    type Rec<'buf>: Send;
23}
24
25/// Family for record types that do not borrow (owned payloads).
26#[derive(Debug)]
27pub struct Owned<T>(std::marker::PhantomData<fn() -> T>);
28
29impl<T: Send + 'static> RecFamily for Owned<T> {
30    type Rec<'buf> = T;
31}
32
33/// Push-style receiver for deserialized records (0..N per payload).
34pub trait EmitRecord<'buf, T> {
35    /// Hand one record to the chain. A [`Flow::Blocked`] return means
36    /// downstream is full; the deserializer stops emitting and the driver
37    /// retries the batch from the current payload cursor.
38    fn emit(&mut self, rec: Record<T>) -> Flow;
39}
40
41/// One borrowed payload in, 0..N records out through `out`.
42///
43/// Dyn-compatible for a concrete family (the only generic on the method is
44/// a lifetime). Zero emissions is valid (tombstones, empty envelopes);
45/// errors are subject to the stage's `ErrorPolicy`.
46pub trait Deserializer<F: RecFamily>: Send {
47    /// Decode `raw`, emitting each resulting record with the payload's
48    /// metadata and a clone of `ack`.
49    fn deserialize<'buf>(
50        &mut self,
51        raw: &RawPayload<'buf>,
52        ack: &AckRef,
53        out: &mut dyn EmitRecord<'buf, F::Rec<'buf>>,
54    ) -> Result<(), DeserError>;
55}
56
57/// Passthrough deserializer: yields the raw bytes as owned `Vec<u8>`
58/// records. Useful for byte-oriented pipelines and tests.
59#[derive(Clone, Copy, Debug, Default)]
60pub struct BytesPassthrough;
61
62impl Deserializer<Owned<Vec<u8>>> for BytesPassthrough {
63    fn deserialize<'buf>(
64        &mut self,
65        raw: &RawPayload<'buf>,
66        ack: &AckRef,
67        out: &mut dyn EmitRecord<'buf, Vec<u8>>,
68    ) -> Result<(), DeserError> {
69        let _ = out.emit(Record {
70            payload: raw.bytes.to_vec(),
71            meta: raw.meta(),
72            ack: ack.clone(),
73        });
74        Ok(())
75    }
76}
77
78#[cfg(all(test, not(loom)))]
79mod tests {
80    use super::*;
81    use crate::record::PartitionId;
82
83    struct Sink(Vec<Vec<u8>>);
84    impl EmitRecord<'_, Vec<u8>> for Sink {
85        fn emit(&mut self, rec: Record<Vec<u8>>) -> Flow {
86            self.0.push(rec.payload);
87            Flow::Continue
88        }
89    }
90
91    #[test]
92    fn passthrough_emits_one_owned_record() {
93        let (ack, _rx) = AckRef::test_pair();
94        let raw = RawPayload {
95            bytes: b"abc",
96            key: None,
97            partition: PartitionId(0),
98            offset: 1,
99            timestamp_ms: 2,
100        };
101        let mut sink = Sink(Vec::new());
102        BytesPassthrough.deserialize(&raw, &ack, &mut sink).unwrap();
103        assert_eq!(sink.0, vec![b"abc".to_vec()]);
104    }
105}
etl_core/deser.rs

etl_core/
deser.rs
