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

laminar_core/checkpoint/checkpoint_manifest/
mod.rs
