etl_core/record.rs
1//! Core record types: payloads, metadata, and the flow-control signal.
2//!
3//! A [`Record`] is what moves through an operator chain: a payload (which
4//! may borrow from the source's buffers), a [`Copy`] metadata struct, and a
5//! clone of the source batch's acknowledgement handle
6//! ([`AckRef`](crate::checkpoint::AckRef)). Records are born inside a chain's
7//! `push_batch` call (deserialization) and die inside the same call
8//! (serialized into shard frames, filtered out, or failed) — they are never
9//! stored across the chain boundary, which is what makes borrowed payloads
10//! sound. See `docs/DESIGN.md` (§ Frozen v1 contracts).
11
12use crate::checkpoint::AckRef;
13
14/// Flow-control result of pushing one record downstream.
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub enum Flow {
17 /// Keep pushing.
18 Continue,
19 /// Downstream is full: stop pushing and resume from the current cursor
20 /// once capacity frees. Never block — see the backpressure invariant.
21 Blocked,
22}
23
24/// Identifier of a source partition (dense, source-assigned).
25#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
26pub struct PartitionId(pub u32);
27
28/// Per-record metadata. `Copy`, no drop glue — moves through operators for
29/// free and never touches the heap.
30#[derive(Clone, Copy, Debug)]
31pub struct RecordMeta {
32 /// Source partition this record was read from.
33 pub partition: PartitionId,
34 /// Source offset (e.g. Kafka offset) of the originating payload.
35 pub offset: i64,
36 /// Event time in milliseconds since the epoch, as reported by the
37 /// source (e.g. the Kafka message timestamp).
38 pub event_time_ms: i64,
39 /// Stable hash of the message key, when the source provides one; used
40 /// by key-based shard routing.
41 pub key_hash: Option<u64>,
42}
43
44/// A record flowing through the chain.
45#[derive(Debug)]
46pub struct Record<T> {
47 /// The payload; may borrow from the source's buffers.
48 pub payload: T,
49 /// Copyable metadata, preserved automatically by operators.
50 pub meta: RecordMeta,
51 /// Acknowledgement handle shared by every record derived from the same
52 /// source poll batch. Dropping it (filter, error-skip, or successful
53 /// write) resolves this record's share of the batch.
54 pub ack: AckRef,
55}
56
57impl<T> Record<T> {
58 /// Transform the payload, preserving metadata and the ack handle.
59 #[inline(always)]
60 pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Record<U> {
61 Record {
62 payload: f(self.payload),
63 meta: self.meta,
64 ack: self.ack,
65 }
66 }
67}
68
69/// A raw payload borrowed from the source's buffers, valid for `'buf`.
70#[derive(Clone, Copy, Debug)]
71pub struct RawPayload<'buf> {
72 /// The message value.
73 pub bytes: &'buf [u8],
74 /// The message key, when the source provides one.
75 pub key: Option<&'buf [u8]>,
76 /// Source partition.
77 pub partition: PartitionId,
78 /// Source offset.
79 pub offset: i64,
80 /// Source-reported event time (ms since epoch).
81 pub timestamp_ms: i64,
82}
83
84impl RawPayload<'_> {
85 /// Metadata for records derived from this payload. The key hash uses a
86 /// stable 64-bit hash so shard routing survives restarts and versions.
87 #[inline]
88 pub fn meta(&self) -> RecordMeta {
89 RecordMeta {
90 partition: self.partition,
91 offset: self.offset,
92 event_time_ms: self.timestamp_ms,
93 key_hash: self.key.map(stable_key_hash),
94 }
95 }
96}
97
98/// Stable, seedless 64-bit hash for shard routing (FNV-1a). Not for
99/// adversarial input — routing only. Stability across processes, versions,
100/// and platforms is a contract: changing this function reshuffles shards.
101#[inline]
102#[must_use]
103pub fn stable_key_hash(key: &[u8]) -> u64 {
104 const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
105 const PRIME: u64 = 0x0000_0100_0000_01b3;
106 let mut h = OFFSET;
107 for &b in key {
108 h ^= u64::from(b);
109 h = h.wrapping_mul(PRIME);
110 }
111 h
112}
113
114#[cfg(test)]
115mod tests {
116 use super::*;
117
118 // Constructs an AckRef, whose internals are loom types under the loom
119 // cfg and may only be touched inside a loom model.
120 #[cfg(not(loom))]
121 #[test]
122 fn record_map_preserves_meta_and_ack() {
123 let (ack, _rx) = crate::checkpoint::AckRef::test_pair();
124 let rec = Record {
125 payload: 21u32,
126 meta: RecordMeta {
127 partition: PartitionId(3),
128 offset: 42,
129 event_time_ms: 1_000,
130 key_hash: Some(7),
131 },
132 ack,
133 };
134 let mapped = rec.map(|v| u64::from(v) * 2);
135 assert_eq!(mapped.payload, 42);
136 assert_eq!(mapped.meta.partition, PartitionId(3));
137 assert_eq!(mapped.meta.offset, 42);
138 assert_eq!(mapped.meta.key_hash, Some(7));
139 }
140
141 #[test]
142 fn key_hash_is_stable() {
143 // Pinned expectations: changing the hash reshuffles shard routing,
144 // so a change here must be a conscious, breaking decision.
145 assert_eq!(stable_key_hash(b""), 0xcbf2_9ce4_8422_2325);
146 assert_eq!(stable_key_hash(b"a"), 0xaf63_dc4c_8601_ec8c);
147 assert_ne!(stable_key_hash(b"user-1"), stable_key_hash(b"user-2"));
148 }
149
150 #[test]
151 fn raw_payload_meta_hashes_key() {
152 let payload = RawPayload {
153 bytes: b"v",
154 key: Some(b"k"),
155 partition: PartitionId(0),
156 offset: 9,
157 timestamp_ms: 5,
158 };
159 assert_eq!(payload.meta().key_hash, Some(stable_key_hash(b"k")));
160 let keyless = RawPayload {
161 key: None,
162 ..payload
163 };
164 assert_eq!(keyless.meta().key_hash, None);
165 }
166
167 #[test]
168 fn meta_is_copy_and_small() {
169 // The hot path passes RecordMeta in registers; keep it lean.
170 assert!(size_of::<RecordMeta>() <= 40);
171 fn assert_copy<T: Copy>() {}
172 assert_copy::<RecordMeta>();
173 assert_copy::<Flow>();
174 assert_copy::<PartitionId>();
175 }
176}