laminar_core/checkpoint/subscription/
identity.rs

1use std::fmt::Write as _;
2
3use sha2::{Digest as _, Sha256};
4
5use super::SubscriptionContractError;
6use crate::checkpoint::PipelineIdentity;
7use crate::state::KeyGroupCount;
8
9/// Current committed-subscription protocol version.
10pub const SUBSCRIPTION_PROTOCOL_VERSION: u16 = 1;
11
12/// Version of the durable subscription metadata and segment envelope.
13#[derive(
14    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
15)]
16#[serde(transparent)]
17pub struct SubscriptionProtocolVersion(u16);
18
19impl SubscriptionProtocolVersion {
20    /// Current protocol version.
21    pub const CURRENT: Self = Self(SUBSCRIPTION_PROTOCOL_VERSION);
22
23    /// Construct a version read from a wire or persisted envelope.
24    #[must_use]
25    pub const fn new(value: u16) -> Self {
26        Self(value)
27    }
28
29    /// Raw protocol value.
30    #[must_use]
31    pub const fn get(self) -> u16 {
32        self.0
33    }
34
35    /// Require the exact production protocol.
36    pub(crate) fn validate(self) -> Result<(), SubscriptionContractError> {
37        if self != Self::CURRENT {
38            return Err(SubscriptionContractError::ProtocolVersion {
39                actual: self.0,
40                expected: SUBSCRIPTION_PROTOCOL_VERSION,
41            });
42        }
43        Ok(())
44    }
45}
46
47/// Fixed-size SHA-256 used by subscription metadata.
48#[derive(
49    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
50)]
51#[serde(transparent)]
52pub struct SubscriptionDigest([u8; 32]);
53
54impl SubscriptionDigest {
55    /// Digest one byte string with a domain separator.
56    #[must_use]
57    pub fn for_bytes(domain: &[u8], value: &[u8]) -> Self {
58        let mut hash = Sha256::new();
59        update_part(&mut hash, domain);
60        update_part(&mut hash, value);
61        Self(hash.finalize().into())
62    }
63
64    /// Construct from an already verified SHA-256.
65    #[must_use]
66    pub const fn from_bytes(bytes: [u8; 32]) -> Self {
67        Self(bytes)
68    }
69
70    /// Borrow the exact digest bytes.
71    #[must_use]
72    pub const fn as_bytes(&self) -> &[u8; 32] {
73        &self.0
74    }
75
76    /// Lowercase hexadecimal encoding for content-addressed object keys.
77    #[must_use]
78    pub fn to_hex(self) -> String {
79        let mut encoded = String::with_capacity(64);
80        for byte in self.0 {
81            let _ = write!(&mut encoded, "{byte:02x}");
82        }
83        encoded
84    }
85
86    pub(crate) fn validate(self, field: &'static str) -> Result<(), SubscriptionContractError> {
87        if self.0 == [0; 32] {
88            return Err(SubscriptionContractError::ZeroDigest { field });
89        }
90        Ok(())
91    }
92}
93
94impl std::fmt::Display for SubscriptionDigest {
95    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96        for byte in self.0 {
97            write!(formatter, "{byte:02x}")?;
98        }
99        Ok(())
100    }
101}
102
103/// Durable identity of one catalog stream incarnation.
104#[derive(
105    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
106)]
107#[serde(transparent)]
108pub struct StreamGeneration(SubscriptionDigest);
109
110impl StreamGeneration {
111    /// Derive a generation from the durable deployment, catalog incarnation, query, and pipeline.
112    #[must_use]
113    pub fn derive(
114        deployment_id: uuid::Uuid,
115        catalog_generation: u64,
116        stream_name: &str,
117        canonical_query: &str,
118        pipeline_identity: &PipelineIdentity,
119    ) -> Self {
120        let mut hash = Sha256::new();
121        update_part(&mut hash, b"laminardb-subscription-stream-generation-v1");
122        update_part(&mut hash, deployment_id.as_bytes());
123        update_part(&mut hash, &catalog_generation.to_le_bytes());
124        update_part(&mut hash, stream_name.as_bytes());
125        update_part(&mut hash, canonical_query.as_bytes());
126        update_part(
127            &mut hash,
128            &pipeline_identity.canonical_version.to_le_bytes(),
129        );
130        update_part(&mut hash, pipeline_identity.sha256.as_bytes());
131        Self(SubscriptionDigest(hash.finalize().into()))
132    }
133
134    /// Construct from a fixed digest, for persisted metadata and tests.
135    #[must_use]
136    pub const fn from_digest(digest: SubscriptionDigest) -> Self {
137        Self(digest)
138    }
139
140    /// Exact generation digest.
141    #[must_use]
142    pub const fn digest(self) -> SubscriptionDigest {
143        self.0
144    }
145
146    pub(crate) fn validate(self) -> Result<(), SubscriptionContractError> {
147        self.0.validate("stream_generation")
148    }
149}
150
151impl std::fmt::Display for StreamGeneration {
152    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
153        self.0.fmt(formatter)
154    }
155}
156
157/// Logical output partition. Keyed aggregate partitions are stable vnodes.
158#[derive(
159    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
160)]
161#[serde(transparent)]
162pub struct OutputPartitionId(u16);
163
164impl OutputPartitionId {
165    /// Construct from a vnode number.
166    #[must_use]
167    pub const fn new(vnode: u16) -> Self {
168        Self(vnode)
169    }
170
171    /// Vnode number represented by this output partition.
172    #[must_use]
173    pub const fn get(self) -> u16 {
174        self.0
175    }
176
177    pub(crate) fn validate(
178        self,
179        key_group_count: KeyGroupCount,
180    ) -> Result<(), SubscriptionContractError> {
181        if self.0 >= key_group_count.get() {
182            return Err(SubscriptionContractError::PartitionOutOfRange {
183                partition: self.0,
184                vnode_count: key_group_count.get(),
185            });
186        }
187        Ok(())
188    }
189}
190
191/// Monotonic frame position within one stream generation and output partition.
192#[derive(
193    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
194)]
195#[serde(transparent)]
196pub struct PartitionSequence(u64);
197
198impl PartitionSequence {
199    /// The first assigned partition sequence.
200    pub const FIRST: Self = Self(0);
201
202    /// Construct a position from persisted metadata.
203    #[must_use]
204    pub const fn new(value: u64) -> Self {
205        Self(value)
206    }
207
208    /// Raw sequence value.
209    #[must_use]
210    pub const fn get(self) -> u64 {
211        self.0
212    }
213
214    /// Advance by one without wrapping.
215    ///
216    /// # Errors
217    /// Returns sequence overflow at `u64::MAX`.
218    pub fn checked_next(self) -> Result<Self, SubscriptionContractError> {
219        self.0
220            .checked_add(1)
221            .map(Self)
222            .ok_or(SubscriptionContractError::SequenceOverflow)
223    }
224}
225
226/// Globally unambiguous identity of one partition-local output frame.
227#[derive(
228    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
229)]
230#[serde(deny_unknown_fields)]
231pub struct OutputFrameId {
232    /// Durable stream incarnation.
233    pub stream_generation: StreamGeneration,
234    /// Stable vnode output partition.
235    pub partition: OutputPartitionId,
236    /// Partition-local sequence.
237    pub sequence: PartitionSequence,
238}
239
240fn update_part(hash: &mut Sha256, bytes: &[u8]) {
241    hash.update(u64::try_from(bytes.len()).unwrap_or(u64::MAX).to_le_bytes());
242    hash.update(bytes);
243}
244
245#[cfg(test)]
246mod tests {
247    use super::*;
248
249    fn pipeline() -> PipelineIdentity {
250        PipelineIdentity::empty()
251    }
252
253    #[test]
254    fn generation_round_trips_and_recreation_changes_it() {
255        let deployment = uuid::Uuid::parse_str("11111111-1111-4111-8111-111111111111").unwrap();
256        let first = StreamGeneration::derive(deployment, 7, "positions", "select 1", &pipeline());
257        let recreated =
258            StreamGeneration::derive(deployment, 8, "positions", "select 1", &pipeline());
259        assert_ne!(first, recreated);
260
261        let encoded = serde_json::to_vec(&first).unwrap();
262        let decoded: StreamGeneration = serde_json::from_slice(&encoded).unwrap();
263        assert_eq!(decoded, first);
264        assert_eq!(first.to_string().len(), 64);
265    }
266
267    #[test]
268    fn partition_sequence_starts_at_zero_and_never_wraps() {
269        assert_eq!(PartitionSequence::FIRST.get(), 0);
270        assert_eq!(PartitionSequence::FIRST.checked_next().unwrap().get(), 1);
271        assert_eq!(
272            PartitionSequence::new(u64::MAX).checked_next(),
273            Err(SubscriptionContractError::SequenceOverflow)
274        );
275    }
276}
laminar_core/checkpoint/subscription/identity.rs

laminar_core/checkpoint/subscription/
identity.rs
