laminar_core/checkpoint/checkpoint_manifest/
mod.rs

1//! Versioned checkpoint manifest.
2
3#![allow(clippy::disallowed_types)] // cold path: manifest serialization
4
5use std::collections::{BTreeMap, HashMap};
6use std::num::{NonZeroU32, NonZeroU64};
7
8use sha2::{Digest, Sha256};
9
10use crate::checkpoint::assignment::CheckpointAssignmentFence;
11use crate::checkpoint::NodeSubscriptionManifest;
12use crate::state::{
13    KeyGroupCount, DEFAULT_KEY_GROUP_COUNT, LOCAL_NODE_ID, PARTITIONING_ABI_VERSION,
14};
15
16mod sink_artifacts;
17pub use sink_artifacts::{
18    checkpoint_artifact_intent_sha256, PreparedSinkArtifactIntent, PreparedSinkDescriptor,
19};
20
21/// Current checkpoint manifest format. Every other version is rejected.
22pub const CHECKPOINT_MANIFEST_VERSION: u32 = 10;
23
24/// Canonical pipeline-identity payload version.
25pub const PIPELINE_IDENTITY_VERSION: u16 = 7;
26
27/// Runtime envelope used for prepared sink descriptors.
28pub const PREPARED_SINK_DESCRIPTOR_VERSION: u16 = 1;
29
30const EMPTY_SHA256: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
31
32/// SHA-256 identity of the logical pipeline and recovery-state ABI.
33#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
34#[serde(deny_unknown_fields)]
35pub struct PipelineIdentity {
36    /// Version of the canonical payload format.
37    pub canonical_version: u16,
38    /// Exactly 64 lowercase hexadecimal characters.
39    pub sha256: String,
40}
41
42impl PipelineIdentity {
43    /// Identity of an empty canonical payload.
44    #[must_use]
45    pub fn empty() -> Self {
46        Self {
47            canonical_version: PIPELINE_IDENTITY_VERSION,
48            sha256: EMPTY_SHA256.into(),
49        }
50    }
51
52    pub(crate) fn validation_error(&self) -> Option<String> {
53        if self.canonical_version != PIPELINE_IDENTITY_VERSION {
54            return Some(format!(
55                "unsupported pipeline identity version {}; expected {PIPELINE_IDENTITY_VERSION}",
56                self.canonical_version
57            ));
58        }
59        (!is_sha256(&self.sha256))
60            .then(|| "pipeline identity must be 64 lowercase hexadecimal characters".into())
61    }
62
63    /// Whether this identity uses the current canonical version and digest encoding.
64    #[must_use]
65    pub fn is_canonical(&self) -> bool {
66        self.validation_error().is_none()
67    }
68}
69
70/// Stable identity of one immutable node data object.
71#[derive(
72    Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash, PartialOrd, Ord,
73)]
74#[serde(deny_unknown_fields)]
75pub struct StateChunkId {
76    /// Node that wrote the object.
77    pub participant_id: u64,
78    /// Checkpoint that created the object.
79    pub checkpoint_id: u64,
80}
81
82/// A byte range within an immutable node data object.
83#[derive(
84    Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash, PartialOrd, Ord,
85)]
86#[serde(deny_unknown_fields)]
87pub struct ByteRange {
88    /// First byte in the range.
89    pub offset: u64,
90    /// Number of bytes in the range.
91    pub length: u64,
92}
93
94impl ByteRange {
95    /// Return the exclusive range end, or `None` on overflow.
96    #[must_use]
97    pub fn end(self) -> Option<u64> {
98        self.offset.checked_add(self.length)
99    }
100}
101
102/// The one data object written by this participant for this checkpoint.
103#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
104#[serde(deny_unknown_fields)]
105pub struct NodeDataObject {
106    /// Identity used to derive the provider-neutral object path.
107    pub chunk: StateChunkId,
108    /// Exact object length.
109    pub object_length: u64,
110    /// SHA-256 of the complete object.
111    pub sha256: String,
112}
113
114/// State-frame identity. Whole-operator metadata and vnode-keyed state are distinct.
115#[derive(
116    Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash, PartialOrd, Ord,
117)]
118#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
119pub enum StateFrameKey {
120    /// Non-vnode state such as graph progress, channel watermarks, idleness, and timers.
121    OperatorWhole {
122        /// Stable operator identifier.
123        operator_id: String,
124    },
125    /// State owned by one vnode of an operator.
126    Vnode {
127        /// Stable operator identifier.
128        operator_id: String,
129        /// Vnode within the manifest's partition domain.
130        vnode: u16,
131    },
132}
133
134/// A checksummed state payload in the current or a prior node data object.
135#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
136#[serde(deny_unknown_fields)]
137pub struct StateFrame {
138    /// Logical state slot restored from this payload.
139    pub key: StateFrameKey,
140    /// Object containing the payload.
141    pub chunk: StateChunkId,
142    /// Exact payload range.
143    pub range: ByteRange,
144    /// SHA-256 of only this frame.
145    pub sha256: String,
146}
147
148/// Watermark and idleness state for one stable input channel.
149#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
150#[serde(deny_unknown_fields)]
151pub struct ChannelProgress {
152    /// Participant owning this channel state.
153    pub participant_id: u64,
154    /// Stable registered source name.
155    pub source_name: String,
156    /// Reserved logical singleton or opaque physical input-channel identity.
157    pub input_channel: Vec<u8>,
158    /// Watermark when initialized; `None` before the channel emits one.
159    pub watermark: Option<i64>,
160    /// Whether the channel is excluded from the active watermark minimum.
161    pub idle: bool,
162}
163
164/// Metadata and exact logical reference count for an older immutable object.
165#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
166#[serde(deny_unknown_fields)]
167pub struct ReferencedStateChunk {
168    /// Prior object identity. The object path is derived without listing.
169    pub chunk: StateChunkId,
170    /// Exact immutable object length.
171    pub object_length: u64,
172    /// SHA-256 of the complete object.
173    pub sha256: String,
174    /// Number of state frames in this manifest that reference the object. GC aggregates this
175    /// over the bounded committed manifest/index inventory; it never discovers references by
176    /// listing objects.
177    pub ref_count: NonZeroU32,
178}
179
180/// A point-in-time snapshot of one participant's pipeline state.
181#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
182#[serde(deny_unknown_fields)]
183pub struct CheckpointManifest {
184    /// Exact manifest format version.
185    pub version: u32,
186    /// Unique, monotonically increasing checkpoint ID; must equal `epoch`.
187    pub checkpoint_id: u64,
188    /// Persisted protocol epoch; must equal `checkpoint_id`.
189    pub epoch: u64,
190    /// Creation time in milliseconds since the Unix epoch.
191    pub timestamp_ms: u64,
192    /// Node owning this manifest and its current data object.
193    pub participant_id: u64,
194    /// Per-source connector offsets.
195    pub source_offsets: HashMap<String, ConnectorCheckpoint>,
196    /// Canonically sorted registered source names.
197    pub source_names: Vec<String>,
198    /// Canonically sorted registered sink names.
199    pub sink_names: Vec<String>,
200    /// Deterministic logical topology and state ABI identity.
201    pub pipeline_identity: PipelineIdentity,
202    /// Create-once checkpoint namespace incarnation.
203    pub deployment_id: String,
204    /// Durable key encoding, hashing, and key-group mapping contract.
205    pub partitioning_abi_version: u16,
206    /// Virtual partition count.
207    pub vnode_count: u16,
208    /// Exact cluster assignment generation and participant fence; local manifests use `None`.
209    pub assignment_fence: Option<CheckpointAssignmentFence>,
210    /// Whether this cut can be restored under a different cluster vnode assignment.
211    ///
212    /// Cluster manifests must carry an affirmative, capture-time proof. Local manifests cannot
213    /// claim reassignment portability because they have no assignment-fenced recovery domain.
214    pub reassignment_portable: bool,
215    /// Canonically sorted vnodes whose state inventory this participant supplies.
216    pub owned_vnodes: Vec<u16>,
217    /// Derived checkpoint watermark retained with the exact cut.
218    pub checkpoint_watermark: Option<i64>,
219    /// Canonically ordered per-channel watermarks and idle/uninitialized states.
220    pub channel_progress: Vec<ChannelProgress>,
221    /// The only data object written by this node for this checkpoint.
222    pub node_data: NodeDataObject,
223    /// Canonically ordered complete logical state inventory.
224    pub state_frames: Vec<StateFrame>,
225    /// Canonically ordered phase-one sink descriptor inventory.
226    pub prepared_sinks: Vec<PreparedSinkDescriptor>,
227    /// Canonically ordered recovery intents captured before sink epoch begin.
228    #[serde(default, skip_serializing_if = "Vec::is_empty")]
229    pub sink_artifact_intents: Vec<PreparedSinkArtifactIntent>,
230    /// Canonically ordered older objects retained by `state_frames`.
231    pub referenced_chunks: Vec<ReferencedStateChunk>,
232    /// Participant-local subscription output ranges and immutable segment references.
233    #[serde(default, skip_serializing_if = "Option::is_none")]
234    pub subscription_output: Option<NodeSubscriptionManifest>,
235}
236
237/// Errors found during manifest validation.
238#[derive(Debug, Clone, PartialEq, Eq)]
239pub struct ManifestValidationError {
240    /// Human-readable description of the issue.
241    pub message: String,
242}
243
244impl std::fmt::Display for ManifestValidationError {
245    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
246        f.write_str(&self.message)
247    }
248}
249
250impl CheckpointManifest {
251    /// Create a prepared manifest with the common vnode topology.
252    #[must_use]
253    pub fn new(checkpoint_id: u64, epoch: u64) -> Self {
254        Self::new_with_key_group_count(checkpoint_id, epoch, DEFAULT_KEY_GROUP_COUNT)
255    }
256
257    /// Create a prepared manifest with an explicit vnode topology.
258    #[must_use]
259    pub fn new_with_key_group_count(
260        checkpoint_id: u64,
261        epoch: u64,
262        key_group_count: KeyGroupCount,
263    ) -> Self {
264        #[allow(clippy::cast_possible_truncation)]
265        let timestamp_ms = std::time::SystemTime::now()
266            .duration_since(std::time::UNIX_EPOCH)
267            .unwrap_or_default()
268            .as_millis() as u64;
269
270        Self {
271            version: CHECKPOINT_MANIFEST_VERSION,
272            checkpoint_id,
273            epoch,
274            timestamp_ms,
275            participant_id: LOCAL_NODE_ID.0,
276            source_offsets: HashMap::new(),
277            source_names: Vec::new(),
278            sink_names: Vec::new(),
279            pipeline_identity: PipelineIdentity::empty(),
280            deployment_id: String::new(),
281            partitioning_abi_version: PARTITIONING_ABI_VERSION,
282            vnode_count: key_group_count.get(),
283            assignment_fence: None,
284            reassignment_portable: false,
285            owned_vnodes: (0..key_group_count.get()).collect(),
286            checkpoint_watermark: None,
287            channel_progress: Vec::new(),
288            node_data: NodeDataObject {
289                chunk: StateChunkId {
290                    participant_id: LOCAL_NODE_ID.0,
291                    checkpoint_id,
292                },
293                object_length: 0,
294                sha256: EMPTY_SHA256.into(),
295            },
296            state_frames: Vec::new(),
297            prepared_sinks: Vec::new(),
298            sink_artifact_intents: Vec::new(),
299            referenced_chunks: Vec::new(),
300            subscription_output: None,
301        }
302    }
303
304    /// Bind the manifest and current data-object identity to one participant.
305    pub fn bind_participant(&mut self, participant_id: u64) {
306        self.participant_id = participant_id;
307        self.node_data.chunk.participant_id = participant_id;
308    }
309
310    /// Validate the exact v10 recovery contract.
311    #[must_use]
312    pub fn validate(
313        &self,
314        expected_key_group_count: KeyGroupCount,
315    ) -> Vec<ManifestValidationError> {
316        let mut errors = Vec::new();
317        let mut error = |message| errors.push(ManifestValidationError { message });
318
319        if self.version != CHECKPOINT_MANIFEST_VERSION {
320            error(format!(
321                "unsupported manifest version {}; expected {CHECKPOINT_MANIFEST_VERSION}",
322                self.version
323            ));
324        }
325        if self.partitioning_abi_version != PARTITIONING_ABI_VERSION {
326            error(format!(
327                "partitioning ABI mismatch: checkpoint has {}, runtime expects {PARTITIONING_ABI_VERSION}",
328                self.partitioning_abi_version
329            ));
330        }
331        if self.checkpoint_id == 0 || self.epoch != self.checkpoint_id {
332            error("checkpoint attempt must use one nonzero canonical checkpoint ID".into());
333        }
334        if self.timestamp_ms == 0 {
335            error("timestamp_ms must be nonzero".into());
336        }
337        if let Some(message) = self.pipeline_identity.validation_error() {
338            error(message);
339        }
340        let valid_deployment = uuid::Uuid::parse_str(&self.deployment_id)
341            .is_ok_and(|id| !id.is_nil() && id.to_string() == self.deployment_id);
342        if !valid_deployment {
343            error("deployment_id must be a canonical non-nil UUID".into());
344        }
345        validate_topology(self, expected_key_group_count, &mut error);
346        validate_sorted_unique("source_names", &self.source_names, &mut error);
347        validate_sorted_unique("sink_names", &self.sink_names, &mut error);
348        if !self.channel_progress.windows(2).all(|pair| {
349            (
350                &pair[0].participant_id,
351                &pair[0].source_name,
352                &pair[0].input_channel,
353            ) < (
354                &pair[1].participant_id,
355                &pair[1].source_name,
356                &pair[1].input_channel,
357            )
358        }) {
359            error(
360                "channel_progress must be strictly ordered by participant, source, and input channel"
361                    .into(),
362            );
363        }
364        let mut channel_modes = BTreeMap::new();
365        for channel in &self.channel_progress {
366            if channel.participant_id != self.participant_id {
367                error("channel_progress participant_id must match the manifest participant".into());
368            }
369            if channel.source_name.is_empty() {
370                error("channel_progress source_name must not be empty".into());
371            } else if self
372                .source_names
373                .binary_search(&channel.source_name)
374                .is_err()
375            {
376                error(format!(
377                    "channel_progress source '{}' not in source_names",
378                    channel.source_name
379                ));
380            }
381            if channel.input_channel.is_empty() {
382                error("channel_progress input_channel must not be empty".into());
383            }
384            let modes = channel_modes
385                .entry(channel.source_name.as_str())
386                .or_insert((false, false));
387            if channel.input_channel == super::SINGLETON_WATERMARK_CHANNEL {
388                modes.0 = true;
389            } else {
390                modes.1 = true;
391            }
392        }
393        for (source, (logical, physical)) in channel_modes {
394            if logical && physical {
395                error(format!(
396                    "channel_progress source '{source}' mixes logical and physical input channels"
397                ));
398            }
399        }
400        match super::classify_channel_progress(&self.channel_progress) {
401            Ok(classification) if self.checkpoint_watermark != classification.active_value() => {
402                error("checkpoint_watermark does not match channel progress".into());
403            }
404            Err(message) => error(message),
405            Ok(_) => {}
406        }
407        for name in self.source_offsets.keys() {
408            if self.source_names.binary_search(name).is_err() {
409                error(format!(
410                    "source_offsets contains '{name}' not in source_names"
411                ));
412            }
413        }
414        for (name, checkpoint) in &self.source_offsets {
415            if let Some(channels) = &checkpoint.input_channels {
416                if channels.iter().any(Vec::is_empty)
417                    || !channels.windows(2).all(|pair| pair[0] < pair[1])
418                {
419                    error(format!(
420                        "source '{name}' input_channels must contain non-empty, strictly ordered unique identities"
421                    ));
422                }
423                if channels
424                    .iter()
425                    .any(|channel| channel == super::SINGLETON_WATERMARK_CHANNEL)
426                {
427                    error(format!(
428                        "source '{name}' input_channels contains the reserved logical watermark channel"
429                    ));
430                }
431                let mut progress = self
432                    .channel_progress
433                    .iter()
434                    .filter(|channel| channel.source_name == *name)
435                    .map(|channel| channel.input_channel.as_slice())
436                    .collect::<Vec<_>>();
437                if !progress.is_empty()
438                    && !progress
439                        .iter()
440                        .all(|channel| *channel == super::SINGLETON_WATERMARK_CHANNEL)
441                {
442                    progress.sort_unstable();
443                    if !progress.into_iter().eq(channels.iter().map(Vec::as_slice)) {
444                        error(format!(
445                            "source '{name}' input_channels do not match its channel_progress roster"
446                        ));
447                    }
448                }
449            }
450        }
451
452        let current = self.node_data.chunk;
453        if current.participant_id != self.participant_id
454            || current.checkpoint_id != self.checkpoint_id
455        {
456            error("node_data chunk must match the manifest participant and checkpoint".into());
457        }
458        if !is_sha256(&self.node_data.sha256) {
459            error("node_data digest must be lowercase SHA-256".into());
460        }
461
462        let referenced_by_id = self
463            .referenced_chunks
464            .iter()
465            .map(|reference| (reference.chunk, reference))
466            .collect::<HashMap<_, _>>();
467        if referenced_by_id.len() != self.referenced_chunks.len()
468            || !self
469                .referenced_chunks
470                .windows(2)
471                .all(|pair| pair[0].chunk < pair[1].chunk)
472        {
473            error("referenced_chunks must be strictly ordered and unique".into());
474        }
475        for reference in &self.referenced_chunks {
476            if reference.chunk == current
477                || reference.chunk.checkpoint_id >= self.checkpoint_id
478                || reference.chunk.checkpoint_id == 0
479            {
480                error(format!(
481                    "referenced chunk {:?} is not an older immutable object",
482                    reference.chunk
483                ));
484            }
485            if !is_sha256(&reference.sha256) {
486                error(format!(
487                    "referenced chunk {:?} digest must be lowercase SHA-256",
488                    reference.chunk
489                ));
490            }
491        }
492
493        let mut prior_counts = HashMap::<StateChunkId, u32>::new();
494        let mut current_ranges = Vec::new();
495        if !self
496            .state_frames
497            .windows(2)
498            .all(|pair| pair[0].key < pair[1].key)
499        {
500            error("state_frames must be strictly ordered by logical key".into());
501        }
502        for frame in &self.state_frames {
503            if let StateFrameKey::Vnode { vnode, .. } = &frame.key {
504                if *vnode >= self.vnode_count {
505                    error(format!(
506                        "state frame vnode {vnode} is outside the vnode domain"
507                    ));
508                }
509                if self.owned_vnodes.binary_search(vnode).is_err() {
510                    error(format!(
511                        "state frame vnode {vnode} is not owned by participant {}",
512                        self.participant_id
513                    ));
514                }
515            }
516            if frame.range.length == 0 {
517                error(format!("state frame {:?} has an empty payload", frame.key));
518            }
519            if !is_sha256(&frame.sha256) {
520                error(format!(
521                    "state frame {:?} digest must be lowercase SHA-256",
522                    frame.key
523                ));
524            }
525            let object_length = if frame.chunk == current {
526                current_ranges.push((frame.range, format!("state frame {:?}", frame.key)));
527                Some(self.node_data.object_length)
528            } else {
529                let reference = referenced_by_id.get(&frame.chunk).copied();
530                if let Some(reference) = reference {
531                    let count = prior_counts.entry(frame.chunk).or_default();
532                    *count = count.saturating_add(1);
533                    Some(reference.object_length)
534                } else {
535                    error(format!(
536                        "state frame {:?} references an untracked chunk {:?}",
537                        frame.key, frame.chunk
538                    ));
539                    None
540                }
541            };
542            validate_range(frame.range, object_length, "state frame", &mut error);
543        }
544        for reference in &self.referenced_chunks {
545            let actual = prior_counts.get(&reference.chunk).copied().unwrap_or(0);
546            if actual != reference.ref_count.get() {
547                error(format!(
548                    "referenced chunk {:?} declares ref_count {}, but {actual} state frames reference it",
549                    reference.chunk,
550                    reference.ref_count
551                ));
552            }
553        }
554
555        sink_artifacts::validate_sink_artifacts(self, &mut current_ranges, &mut error);
556
557        let mut expected_offset = 0_u64;
558        for (range, owner) in current_ranges {
559            if range.offset != expected_offset {
560                error(format!(
561                    "{owner} starts at {}, expected {expected_offset} in the canonical node object",
562                    range.offset
563                ));
564            }
565            if let Some(end) = range.end() {
566                expected_offset = end;
567            }
568        }
569        if expected_offset != self.node_data.object_length {
570            error(format!(
571                "current frame and sink ranges cover {expected_offset} bytes, but node_data declares {}",
572                self.node_data.object_length
573            ));
574        }
575
576        errors
577    }
578}
579
580fn validate_topology(
581    manifest: &CheckpointManifest,
582    expected_key_group_count: KeyGroupCount,
583    error: &mut impl FnMut(String),
584) {
585    if manifest.vnode_count != expected_key_group_count.get() {
586        error(format!(
587            "vnode_count mismatch: checkpoint has {}, runtime expects {expected_key_group_count}",
588            manifest.vnode_count
589        ));
590    }
591    if manifest.participant_id == 0 {
592        error("participant_id must be nonzero".into());
593    }
594    if manifest.owned_vnodes.is_empty() {
595        error("owned_vnodes must not be empty".into());
596    } else if !manifest
597        .owned_vnodes
598        .windows(2)
599        .all(|pair| pair[0] < pair[1])
600    {
601        error("owned_vnodes must be strictly ordered and unique".into());
602    }
603    if manifest
604        .owned_vnodes
605        .iter()
606        .any(|vnode| *vnode >= manifest.vnode_count)
607    {
608        error("owned_vnodes contains a vnode outside the manifest domain".into());
609    }
610    match &manifest.assignment_fence {
611        Some(fence) => validate_cluster_topology(manifest, fence, error),
612        None => validate_local_topology(manifest, error),
613    }
614    if let Some(message) = subscription_output_validation_error(manifest) {
615        error(message);
616    }
617}
618
619fn validate_cluster_topology(
620    manifest: &CheckpointManifest,
621    fence: &CheckpointAssignmentFence,
622    error: &mut impl FnMut(String),
623) {
624    if fence.vnode_count != u32::from(manifest.vnode_count)
625        || fence.partitioning_abi_version != manifest.partitioning_abi_version
626        || !fence.contains(manifest.participant_id)
627    {
628        error("assignment_fence does not cover this manifest topology and participant".into());
629    }
630    if !manifest.reassignment_portable {
631        error("a cluster manifest must be proven portable across vnode reassignment".into());
632    }
633}
634
635fn validate_local_topology(manifest: &CheckpointManifest, error: &mut impl FnMut(String)) {
636    let owns_complete_domain = manifest.owned_vnodes.len() == usize::from(manifest.vnode_count)
637        && manifest
638            .owned_vnodes
639            .iter()
640            .copied()
641            .eq(0..manifest.vnode_count);
642    if manifest.participant_id != LOCAL_NODE_ID.0 || !owns_complete_domain {
643        error("a local manifest must use LOCAL_NODE_ID and own every vnode".into());
644    }
645    if manifest.reassignment_portable {
646        error("a local manifest cannot claim vnode reassignment portability".into());
647    }
648}
649
650fn subscription_output_validation_error(manifest: &CheckpointManifest) -> Option<String> {
651    let output = manifest.subscription_output.as_ref()?;
652    let Some(fence) = manifest.assignment_fence.as_ref() else {
653        return Some("a local manifest cannot carry cluster subscription output".into());
654    };
655    if output.epoch != manifest.epoch
656        || output.checkpoint_id != manifest.checkpoint_id
657        || output.participant_id != manifest.participant_id
658        || output.assignment_certificate != *fence
659    {
660        return Some("subscription output manifest belongs to a different participant cut".into());
661    }
662    output
663        .validate(&manifest.owned_vnodes)
664        .err()
665        .map(|error| format!("subscription output manifest is invalid: {error}"))
666}
667
668/// Connector-owned offset map stored at the exact checkpoint cut.
669#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
670#[serde(deny_unknown_fields)]
671pub struct ConnectorCheckpoint {
672    /// Connector-specific offsets.
673    pub offsets: HashMap<String, String>,
674    /// Connector metadata required to interpret the offsets.
675    pub metadata: HashMap<String, String>,
676    /// Canonically ordered opaque identities of the input channels owned by this cut.
677    pub input_channels: Option<Vec<Vec<u8>>>,
678    /// Source-assignment version owning this cut, when applicable.
679    pub source_assignment_version: Option<NonZeroU64>,
680}
681
682impl ConnectorCheckpoint {
683    /// Create an empty connector checkpoint.
684    #[must_use]
685    pub fn new() -> Self {
686        Self::default()
687    }
688
689    /// Create a checkpoint from an offset map.
690    #[must_use]
691    pub fn with_offsets(offsets: HashMap<String, String>) -> Self {
692        Self {
693            offsets,
694            ..Self::default()
695        }
696    }
697}
698
699fn is_sha256(value: &str) -> bool {
700    value.len() == 64
701        && value
702            .bytes()
703            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
704}
705
706fn validate_sorted_unique(field: &str, values: &[String], error: &mut impl FnMut(String)) {
707    if !values.windows(2).all(|pair| pair[0] < pair[1]) {
708        error(format!("{field} must be strictly ordered and unique"));
709    }
710}
711
712fn validate_range(
713    range: ByteRange,
714    object_length: Option<u64>,
715    owner: &str,
716    error: &mut impl FnMut(String),
717) {
718    let Some(end) = range.end() else {
719        error(format!("{owner} byte range overflows"));
720        return;
721    };
722    if object_length.is_some_and(|length| end > length) {
723        error(format!("{owner} byte range ends beyond its object"));
724    }
725}
726
727/// SHA-256 helper used by capture and focused format tests.
728#[must_use]
729pub fn checkpoint_sha256(bytes: &[u8]) -> String {
730    format!("{:x}", Sha256::digest(bytes))
731}
732
733/// Domain-separated digest for an optional prepared sink descriptor.
734#[must_use]
735pub fn checkpoint_descriptor_sha256(payload: Option<&[u8]>) -> String {
736    let mut hasher = Sha256::new();
737    hasher.update(b"laminardb-prepared-sink-descriptor-v1\0");
738    match payload {
739        None => hasher.update([0]),
740        Some(bytes) => {
741            hasher.update([1]);
742            hasher.update(u64::try_from(bytes.len()).unwrap_or(u64::MAX).to_le_bytes());
743            hasher.update(bytes);
744        }
745    }
746    format!("{:x}", hasher.finalize())
747}
748
749#[cfg(test)]
750mod tests;
laminar_core/checkpoint/checkpoint_manifest/mod.rs

laminar_core/checkpoint/checkpoint_manifest/
mod.rs
