laminar_core/checkpoint/
recovery_capsule.rs

1//! Canonical, content-addressed recovery image for a committed cluster checkpoint.
2
3use std::{collections::BTreeMap, num::NonZeroU64};
4
5use serde::Serialize;
6use sha2::{Digest, Sha256};
7
8use crate::checkpoint::{
9    CheckpointAssignmentFence, ConnectorCheckpoint, PipelineIdentity, PIPELINE_IDENTITY_VERSION,
10};
11use crate::state::CheckpointAttempt;
12
13/// Current canonical cluster recovery-capsule format.
14pub const CLUSTER_RECOVERY_CAPSULE_VERSION: u32 = 5;
15
16/// One participant's event-time position at a checkpoint cut.
17///
18/// An uninitialized participant must block watermark advancement. An explicitly idle participant
19/// has no input that can currently hold back the cut and is excluded from the active minimum.
20#[derive(Debug, Clone, Copy, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
21#[serde(rename_all = "snake_case", tag = "state", content = "watermark_ms")]
22pub enum CheckpointWatermark {
23    /// At least one required input has not established an event-time frontier.
24    #[default]
25    Uninitialized,
26    /// Every required input is explicitly idle.
27    Idle,
28    /// Minimum active input watermark for this participant.
29    Active(i64),
30}
31
32impl CheckpointWatermark {
33    /// Fold two required participants into a safe cluster watermark state.
34    #[must_use]
35    pub const fn cluster_min(self, other: Self) -> Self {
36        match (self, other) {
37            (Self::Uninitialized, _) | (_, Self::Uninitialized) => Self::Uninitialized,
38            (Self::Idle, Self::Idle) => Self::Idle,
39            (Self::Idle, Self::Active(value)) | (Self::Active(value), Self::Idle) => {
40                Self::Active(value)
41            }
42            (Self::Active(left), Self::Active(right)) => {
43                Self::Active(if left < right { left } else { right })
44            }
45        }
46    }
47
48    /// Active watermark value, if this participant has active inputs.
49    #[must_use]
50    pub const fn active_value(self) -> Option<i64> {
51        match self {
52            Self::Active(value) => Some(value),
53            Self::Uninitialized | Self::Idle => None,
54        }
55    }
56
57    /// Reject the reserved uninitialized sentinel as an active watermark.
58    ///
59    /// # Errors
60    /// Returns an error when an active watermark uses the reserved sentinel.
61    pub fn validate(self) -> Result<(), String> {
62        if self == Self::Active(i64::MIN) {
63            return Err("active checkpoint watermark cannot use the uninitialized sentinel".into());
64        }
65        Ok(())
66    }
67}
68
69/// Immutable runtime state for one source at a committed checkpoint cut.
70///
71/// This type is deliberately not serializable. The durable authority remains
72/// [`ClusterRecoveryCapsule`]; this is its validated runtime projection.
73#[derive(Debug, PartialEq, Eq)]
74pub struct SourceHandoffState {
75    checkpoint: ConnectorCheckpoint,
76    watermark_ms: Option<i64>,
77}
78
79impl SourceHandoffState {
80    /// Complete connector-defined offsets and metadata plus its provider-neutral assignment cut.
81    #[must_use]
82    pub const fn checkpoint(&self) -> &ConnectorCheckpoint {
83        &self.checkpoint
84    }
85
86    /// Event-time watermark at the committed cut, when this source had one.
87    #[must_use]
88    pub const fn watermark(&self) -> Option<i64> {
89        self.watermark_ms
90    }
91}
92
93/// Validated source recovery state bound to one committed cluster cut.
94///
95/// The object is immutable and intended to be shared as one `Arc` with an
96/// assignment publication. It is not a second durable or serialized format.
97#[derive(Debug, PartialEq, Eq)]
98pub struct CommittedSourceHandoff {
99    attempt: CheckpointAttempt,
100    assignment_version: u64,
101    sources: BTreeMap<String, SourceHandoffState>,
102    cluster_watermark: CheckpointWatermark,
103    recovery_watermark_frontier: Option<i64>,
104}
105
106impl CommittedSourceHandoff {
107    /// Exact committed checkpoint attempt supplying this handoff.
108    #[must_use]
109    pub const fn attempt(&self) -> CheckpointAttempt {
110        self.attempt
111    }
112
113    /// Assignment version sealed by the committed checkpoint.
114    #[must_use]
115    pub const fn checkpoint_assignment_version(&self) -> u64 {
116        self.assignment_version
117    }
118
119    /// Explicit cluster event-time status at the committed cut.
120    #[must_use]
121    pub const fn cluster_watermark(&self) -> CheckpointWatermark {
122        self.cluster_watermark
123    }
124
125    /// Durable numeric event-time frontier restored with the committed cut.
126    ///
127    /// An idle cut can retain the last active frontier even though it has no
128    /// currently active input. An uninitialized cut has no recovery frontier.
129    #[must_use]
130    pub const fn recovery_watermark_frontier(&self) -> Option<i64> {
131        self.recovery_watermark_frontier
132    }
133
134    /// State for `source`, or `None` when the committed cut did not contain it.
135    #[must_use]
136    pub fn source(&self, source: &str) -> Option<&SourceHandoffState> {
137        self.sources.get(source)
138    }
139
140    /// Iterate the complete source cut in canonical source-name order.
141    #[must_use]
142    pub fn sources(
143        &self,
144    ) -> impl ExactSizeIterator<Item = (&str, &SourceHandoffState)> + DoubleEndedIterator {
145        self.sources
146            .iter()
147            .map(|(name, state)| (name.as_str(), state))
148    }
149
150    /// Number of sources captured in the committed cut.
151    #[must_use]
152    pub fn source_count(&self) -> usize {
153        self.sources.len()
154    }
155
156    /// Whether the committed cut contains no sources.
157    #[must_use]
158    pub fn is_empty(&self) -> bool {
159        self.sources.is_empty()
160    }
161}
162
163/// Hard bound on the exact canonical capsule body.
164pub const MAX_RECOVERY_CAPSULE_BYTES: usize = 8 * 1024 * 1024;
165
166/// Content-addressed reference carried by a durable checkpoint outcome.
167#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
168#[serde(deny_unknown_fields)]
169pub struct RecoveryCapsuleRef {
170    /// Exact epoch encoded by the capsule and its object path.
171    pub epoch: u64,
172    /// Exact checkpoint attempt encoded by the capsule and its object path.
173    pub checkpoint_id: u64,
174    /// Lowercase hexadecimal SHA-256 of the exact canonical JSON body.
175    pub sha256: String,
176    /// Exact canonical JSON body length.
177    pub len: u64,
178}
179
180impl RecoveryCapsuleRef {
181    /// Validate the persisted reference format.
182    ///
183    /// # Errors
184    /// Returns a description when the digest or encoded length is not canonical.
185    pub fn validate(&self) -> Result<(), String> {
186        if !CheckpointAttempt::new(self.epoch, self.checkpoint_id).is_canonical() {
187            return Err(
188                "recovery capsule reference must use one nonzero canonical checkpoint ID".into(),
189            );
190        }
191        validate_digest("recovery capsule", &self.sha256)?;
192        if self.len == 0 || self.len > u64::try_from(MAX_RECOVERY_CAPSULE_BYTES).unwrap_or(u64::MAX)
193        {
194            return Err(format!(
195                "recovery capsule length {} is outside 1..={MAX_RECOVERY_CAPSULE_BYTES}",
196                self.len
197            ));
198        }
199        Ok(())
200    }
201}
202
203/// Exact participant artifacts admitted into a cluster recovery image.
204#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
205#[serde(deny_unknown_fields)]
206pub struct ParticipantRecoveryRef {
207    /// Stable participant node identifier.
208    pub participant_id: u64,
209    /// SHA-256 of the participant's canonical readiness record.
210    pub readiness_sha256: String,
211    /// SHA-256 of the participant's normalized prepared manifest.
212    pub manifest_sha256: String,
213    /// SHA-256 of the participant's portable non-vnode state image.
214    pub portable_state_sha256: String,
215}
216
217impl ParticipantRecoveryRef {
218    fn validate(&self) -> Result<(), String> {
219        if self.participant_id == 0 {
220            return Err("recovery capsule participant ID cannot be 0".into());
221        }
222        validate_digest("participant readiness", &self.readiness_sha256)?;
223        validate_digest("participant manifest", &self.manifest_sha256)?;
224        validate_digest("participant portable state", &self.portable_state_sha256)?;
225        Ok(())
226    }
227}
228
229/// Canonical global recovery image selected by one cluster Commit outcome.
230#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
231#[serde(deny_unknown_fields)]
232pub struct ClusterRecoveryCapsule {
233    /// Capsule payload format.
234    pub version: u32,
235    /// Exact checkpoint attempt represented by this image.
236    pub attempt: CheckpointAttempt,
237    /// Durable deployment incarnation that owns the image.
238    pub deployment_id: String,
239    /// Exact logical pipeline and recovery-state ABI.
240    pub pipeline_identity: PipelineIdentity,
241    /// Exact vnode assignment and process roster covered by the image.
242    pub assignment_fence: CheckpointAssignmentFence,
243    /// SHA-256 of the canonical state-backend seal inventory.
244    pub seal_inventory_sha256: String,
245    /// Participant artifacts, sorted by participant ID and exactly covering the fence.
246    pub participants: Vec<ParticipantRecoveryRef>,
247    /// Complete per-source connector offsets in canonical map order.
248    pub source_offsets: BTreeMap<String, BTreeMap<String, String>>,
249    /// Complete per-source connector metadata in canonical map order.
250    pub source_metadata: BTreeMap<String, BTreeMap<String, String>>,
251    /// Assignment version captured by each partitioned source, in canonical source-name order.
252    ///
253    /// The map may be sparse because runtime topology determines which sources require an
254    /// assignment cut. Every populated version must match `assignment_fence`.
255    pub source_assignment_versions: BTreeMap<String, NonZeroU64>,
256    /// Per-source event-time watermarks in canonical map order.
257    pub source_watermarks: BTreeMap<String, i64>,
258    /// Cluster-wide event-time state at this exact cut.
259    pub cluster_watermark: CheckpointWatermark,
260    /// Durable numeric event-time frontier restored with this cut.
261    ///
262    /// This equals the active watermark for an active cut, is absent for an
263    /// uninitialized cut, and may retain the last active value for an idle cut.
264    pub recovery_watermark_frontier: Option<i64>,
265    /// SHA-256 of the canonical portable non-vnode state image.
266    pub portable_state_sha256: String,
267}
268
269impl ClusterRecoveryCapsule {
270    /// Validate all canonical and cross-record invariants.
271    ///
272    /// # Errors
273    /// Returns a description when the capsule cannot name one exact cluster recovery image.
274    pub fn validate(&self) -> Result<(), String> {
275        if self.version != CLUSTER_RECOVERY_CAPSULE_VERSION {
276            return Err(format!(
277                "unsupported recovery capsule version {}; expected {CLUSTER_RECOVERY_CAPSULE_VERSION}",
278                self.version
279            ));
280        }
281        if !self.attempt.is_canonical() {
282            return Err(
283                "recovery capsule attempt must use one nonzero canonical checkpoint ID".into(),
284            );
285        }
286
287        let deployment = uuid::Uuid::parse_str(&self.deployment_id)
288            .map_err(|error| format!("recovery capsule deployment identity is invalid: {error}"))?;
289        if deployment.is_nil() || deployment.to_string() != self.deployment_id {
290            return Err(
291                "recovery capsule deployment identity must be a canonical non-nil UUID".into(),
292            );
293        }
294
295        if self.pipeline_identity.canonical_version != PIPELINE_IDENTITY_VERSION {
296            return Err(format!(
297                "unsupported recovery capsule pipeline identity version {}; expected {PIPELINE_IDENTITY_VERSION}",
298                self.pipeline_identity.canonical_version
299            ));
300        }
301        validate_digest("pipeline identity", &self.pipeline_identity.sha256)?;
302
303        if !self.assignment_fence.is_canonical() {
304            return Err("recovery capsule assignment fence is not canonical".into());
305        }
306        validate_digest("checkpoint seal inventory", &self.seal_inventory_sha256)?;
307        validate_digest("portable state", &self.portable_state_sha256)?;
308
309        if self.participants.is_empty() {
310            return Err("recovery capsule has no participants".into());
311        }
312        for participant in &self.participants {
313            participant.validate()?;
314            if participant.portable_state_sha256 != self.portable_state_sha256 {
315                return Err(format!(
316                    "recovery capsule participant {} has divergent portable state",
317                    participant.participant_id
318                ));
319            }
320        }
321        if self
322            .participants
323            .windows(2)
324            .any(|pair| pair[0].participant_id >= pair[1].participant_id)
325        {
326            return Err(
327                "recovery capsule participants must be sorted and unique by participant ID".into(),
328            );
329        }
330        let participant_ids: Vec<u64> = self
331            .participants
332            .iter()
333            .map(|participant| participant.participant_id)
334            .collect();
335        if participant_ids != self.assignment_fence.participant_ids() {
336            return Err(
337                "recovery capsule participants do not exactly cover its assignment fence".into(),
338            );
339        }
340        validate_source_maps(&self.source_offsets, &self.source_metadata)?;
341        for (source, assignment_version) in &self.source_assignment_versions {
342            validate_name("source assignment version", source)?;
343            if !self.source_offsets.contains_key(source) {
344                return Err(format!(
345                    "recovery capsule source assignment version names unknown source '{source}'"
346                ));
347            }
348            if assignment_version.get() != self.assignment_fence.assignment_version {
349                return Err(format!(
350                    "recovery capsule source assignment version for '{source}' is {}; expected {}",
351                    assignment_version, self.assignment_fence.assignment_version
352                ));
353            }
354        }
355        for (source, watermark) in &self.source_watermarks {
356            validate_name("source watermark", source)?;
357            if !self.source_offsets.contains_key(source) {
358                return Err(format!(
359                    "recovery capsule source watermark '{source}' has no connector checkpoint"
360                ));
361            }
362            if *watermark == i64::MIN {
363                return Err(format!(
364                    "recovery capsule source watermark '{source}' uses the uninitialized sentinel"
365                ));
366            }
367        }
368        self.cluster_watermark.validate()?;
369        match self.cluster_watermark {
370            CheckpointWatermark::Active(active) => {
371                if self.recovery_watermark_frontier != Some(active) {
372                    return Err(
373                        "active recovery capsule watermark must equal its recovery frontier".into(),
374                    );
375                }
376            }
377            CheckpointWatermark::Uninitialized => {
378                if self.recovery_watermark_frontier.is_some() {
379                    return Err(
380                        "uninitialized recovery capsule watermark cannot have a recovery frontier"
381                            .into(),
382                    );
383                }
384            }
385            CheckpointWatermark::Idle => {}
386        }
387        if let Some(frontier) = self.recovery_watermark_frontier {
388            if frontier == i64::MIN {
389                return Err(
390                    "recovery capsule watermark frontier cannot use the uninitialized sentinel"
391                        .into(),
392                );
393            }
394            if self
395                .source_watermarks
396                .values()
397                .any(|source_watermark| *source_watermark < frontier)
398            {
399                return Err(
400                    "recovery capsule watermark frontier exceeds a source watermark".into(),
401                );
402            }
403        }
404        Ok(())
405    }
406
407    pub(crate) fn encode_and_reference(&self) -> Result<(Vec<u8>, RecoveryCapsuleRef), String> {
408        self.validate()?;
409        let encoded = canonical_json_bytes(self).map_err(|error| error.to_string())?;
410        if encoded.len() > MAX_RECOVERY_CAPSULE_BYTES {
411            return Err(format!(
412                "encoded recovery capsule is {} bytes; maximum is {MAX_RECOVERY_CAPSULE_BYTES}",
413                encoded.len()
414            ));
415        }
416        let reference = RecoveryCapsuleRef {
417            epoch: self.attempt.epoch,
418            checkpoint_id: self.attempt.checkpoint_id,
419            sha256: sha256_hex(&encoded),
420            len: u64::try_from(encoded.len())
421                .map_err(|_| "recovery capsule length overflow".to_string())?,
422        };
423        reference.validate()?;
424        Ok((encoded, reference))
425    }
426}
427
428impl TryFrom<&ClusterRecoveryCapsule> for CommittedSourceHandoff {
429    type Error = String;
430
431    fn try_from(capsule: &ClusterRecoveryCapsule) -> Result<Self, Self::Error> {
432        capsule.validate()?;
433
434        let mut sources = BTreeMap::new();
435        for (source, offsets) in &capsule.source_offsets {
436            let metadata = capsule.source_metadata.get(source).ok_or_else(|| {
437                format!("recovery capsule source '{source}' is missing connector metadata")
438            })?;
439            sources.insert(
440                source.clone(),
441                SourceHandoffState {
442                    checkpoint: ConnectorCheckpoint {
443                        offsets: offsets
444                            .iter()
445                            .map(|(key, value)| (key.clone(), value.clone()))
446                            .collect(),
447                        metadata: metadata
448                            .iter()
449                            .map(|(key, value)| (key.clone(), value.clone()))
450                            .collect(),
451                        source_assignment_version: capsule
452                            .source_assignment_versions
453                            .get(source)
454                            .copied(),
455                    },
456                    watermark_ms: capsule.source_watermarks.get(source).copied(),
457                },
458            );
459        }
460
461        Ok(Self {
462            attempt: capsule.attempt,
463            assignment_version: capsule.assignment_fence.assignment_version,
464            sources,
465            cluster_watermark: capsule.cluster_watermark,
466            recovery_watermark_frontier: capsule.recovery_watermark_frontier,
467        })
468    }
469}
470
471fn validate_source_maps(
472    offsets: &BTreeMap<String, BTreeMap<String, String>>,
473    metadata: &BTreeMap<String, BTreeMap<String, String>>,
474) -> Result<(), String> {
475    if offsets.keys().ne(metadata.keys()) {
476        return Err(
477            "recovery capsule source offsets and metadata must cover the same sources".into(),
478        );
479    }
480    for (source, source_offsets) in offsets {
481        validate_name("source", source)?;
482        for key in source_offsets.keys() {
483            validate_name("source offset key", key)?;
484        }
485        for key in metadata
486            .get(source)
487            .expect("source key sets were compared above")
488            .keys()
489        {
490            validate_name("source metadata key", key)?;
491        }
492    }
493    Ok(())
494}
495
496fn validate_name(kind: &str, name: &str) -> Result<(), String> {
497    if name.is_empty() || name.trim() != name {
498        return Err(format!("recovery capsule has a non-canonical {kind} name"));
499    }
500    Ok(())
501}
502
503fn validate_digest(kind: &str, digest: &str) -> Result<(), String> {
504    if digest.len() != 64
505        || !digest
506            .bytes()
507            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
508    {
509        return Err(format!(
510            "recovery capsule {kind} SHA-256 must be 64 lowercase hexadecimal characters"
511        ));
512    }
513    Ok(())
514}
515
516fn sha256_hex(bytes: &[u8]) -> String {
517    const HEX: &[u8; 16] = b"0123456789abcdef";
518    let digest = Sha256::digest(bytes);
519    let mut encoded = String::with_capacity(digest.len() * 2);
520    for byte in digest {
521        encoded.push(HEX[(byte >> 4) as usize] as char);
522        encoded.push(HEX[(byte & 0x0f) as usize] as char);
523    }
524    encoded
525}
526
527#[derive(Serialize)]
528#[serde(untagged)]
529enum CanonicalJsonValue {
530    Null,
531    Bool(bool),
532    Number(serde_json::Number),
533    String(String),
534    Array(Vec<Self>),
535    Object(BTreeMap<String, Self>),
536}
537
538impl From<serde_json::Value> for CanonicalJsonValue {
539    fn from(value: serde_json::Value) -> Self {
540        match value {
541            serde_json::Value::Null => Self::Null,
542            serde_json::Value::Bool(value) => Self::Bool(value),
543            serde_json::Value::Number(value) => Self::Number(value),
544            serde_json::Value::String(value) => Self::String(value),
545            serde_json::Value::Array(values) => {
546                Self::Array(values.into_iter().map(Self::from).collect())
547            }
548            serde_json::Value::Object(values) => Self::Object(
549                values
550                    .into_iter()
551                    .map(|(key, value)| (key, Self::from(value)))
552                    .collect(),
553            ),
554        }
555    }
556}
557
558/// Serialize any JSON-compatible value with recursively sorted object keys.
559///
560/// This is the canonical digest input for manifests and readiness records containing `HashMap`s.
561///
562/// # Errors
563/// Returns a JSON error when `value` cannot be represented as JSON.
564pub fn canonical_json_bytes<T: Serialize>(value: &T) -> Result<Vec<u8>, serde_json::Error> {
565    let value = serde_json::to_value(value)?;
566    serde_json::to_vec(&CanonicalJsonValue::from(value))
567}
568
569/// SHA-256 of [`canonical_json_bytes`], encoded as lowercase hexadecimal.
570///
571/// # Errors
572/// Returns a JSON error when `value` cannot be represented as JSON.
573pub fn canonical_json_sha256<T: Serialize>(value: &T) -> Result<String, serde_json::Error> {
574    canonical_json_bytes(value).map(|bytes| sha256_hex(&bytes))
575}
576
577#[cfg(test)]
578mod tests {
579    use std::collections::HashMap;
580
581    use super::*;
582    use crate::checkpoint::CheckpointParticipant;
583
584    fn digest(byte: u8) -> String {
585        format!("{byte:02x}").repeat(32)
586    }
587
588    fn capsule() -> ClusterRecoveryCapsule {
589        let assignment_fence = CheckpointAssignmentFence::from_owner_map(
590            7,
591            &[2, 9, 2],
592            vec![
593                CheckpointParticipant {
594                    node_id: 2,
595                    boot_incarnation: uuid::Uuid::from_u128(22),
596                },
597                CheckpointParticipant {
598                    node_id: 9,
599                    boot_incarnation: uuid::Uuid::from_u128(99),
600                },
601            ],
602        )
603        .unwrap();
604        ClusterRecoveryCapsule {
605            version: CLUSTER_RECOVERY_CAPSULE_VERSION,
606            attempt: CheckpointAttempt::canonical(10),
607            deployment_id: uuid::Uuid::from_u128(500).to_string(),
608            pipeline_identity: PipelineIdentity {
609                canonical_version: PIPELINE_IDENTITY_VERSION,
610                sha256: digest(1),
611            },
612            assignment_fence,
613            seal_inventory_sha256: digest(2),
614            participants: vec![
615                ParticipantRecoveryRef {
616                    participant_id: 2,
617                    readiness_sha256: digest(3),
618                    manifest_sha256: digest(4),
619                    portable_state_sha256: digest(9),
620                },
621                ParticipantRecoveryRef {
622                    participant_id: 9,
623                    readiness_sha256: digest(6),
624                    manifest_sha256: digest(7),
625                    portable_state_sha256: digest(9),
626                },
627            ],
628            source_offsets: BTreeMap::from([(
629                "events".into(),
630                BTreeMap::from([("partition:0".into(), "41".into())]),
631            )]),
632            source_metadata: BTreeMap::from([(
633                "events".into(),
634                BTreeMap::from([("topic".into(), "events".into())]),
635            )]),
636            source_assignment_versions: BTreeMap::from([(
637                "events".into(),
638                NonZeroU64::new(7).unwrap(),
639            )]),
640            source_watermarks: BTreeMap::from([("events".into(), 1_000)]),
641            cluster_watermark: CheckpointWatermark::Active(900),
642            recovery_watermark_frontier: Some(900),
643            portable_state_sha256: digest(9),
644        }
645    }
646
647    #[test]
648    fn canonical_json_digest_ignores_hash_map_insertion_order() {
649        let mut first = HashMap::new();
650        first.insert("z", HashMap::from([("b", 2), ("a", 1)]));
651        first.insert("a", HashMap::from([("d", 4), ("c", 3)]));
652
653        let mut second = HashMap::new();
654        second.insert("a", HashMap::from([("c", 3), ("d", 4)]));
655        second.insert("z", HashMap::from([("a", 1), ("b", 2)]));
656
657        assert_eq!(
658            canonical_json_bytes(&first).unwrap(),
659            canonical_json_bytes(&second).unwrap()
660        );
661        assert_eq!(
662            canonical_json_sha256(&first).unwrap(),
663            canonical_json_sha256(&second).unwrap()
664        );
665    }
666
667    #[test]
668    fn capsule_requires_exact_sorted_fence_roster() {
669        let valid = capsule();
670        valid.validate().unwrap();
671
672        let mut noncanonical_attempt = valid.clone();
673        noncanonical_attempt.attempt.checkpoint_id += 1;
674        assert!(noncanonical_attempt
675            .validate()
676            .unwrap_err()
677            .contains("one nonzero canonical checkpoint ID"));
678
679        let mut previous_version = valid.clone();
680        previous_version.version = 4;
681        assert!(previous_version
682            .validate()
683            .unwrap_err()
684            .contains("unsupported recovery capsule version 4"));
685
686        let mut reordered = valid.clone();
687        reordered.participants.reverse();
688        assert!(reordered
689            .validate()
690            .unwrap_err()
691            .contains("sorted and unique"));
692
693        let mut incomplete = valid;
694        incomplete.participants.pop();
695        assert!(incomplete.validate().unwrap_err().contains("exactly cover"));
696    }
697
698    #[test]
699    fn encoded_reference_binds_exact_canonical_body() {
700        let capsule = capsule();
701        let (encoded, reference) = capsule.encode_and_reference().unwrap();
702        let json: serde_json::Value = serde_json::from_slice(&encoded).unwrap();
703        assert_eq!(json["version"], CLUSTER_RECOVERY_CAPSULE_VERSION);
704        assert!(json.get("bootstrap_manifest_participant_id").is_none());
705        assert!(json.get("cluster_min_watermark").is_none());
706        assert_eq!(json["cluster_watermark"]["state"], "active");
707        assert_eq!(json["cluster_watermark"]["watermark_ms"], 900);
708        assert_eq!(json["recovery_watermark_frontier"], 900);
709        assert_eq!(json["source_assignment_versions"]["events"], 7);
710        assert_eq!(reference.len, u64::try_from(encoded.len()).unwrap());
711        assert_eq!(reference.sha256, sha256_hex(&encoded));
712        reference.validate().unwrap();
713
714        let mut noncanonical_reference = reference;
715        noncanonical_reference.checkpoint_id += 1;
716        assert!(noncanonical_reference
717            .validate()
718            .unwrap_err()
719            .contains("one nonzero canonical checkpoint ID"));
720    }
721
722    #[test]
723    fn runtime_handoff_preserves_the_complete_committed_source_cut() {
724        let capsule = capsule();
725        let handoff = CommittedSourceHandoff::try_from(&capsule).unwrap();
726
727        assert_eq!(handoff.attempt(), CheckpointAttempt::canonical(10));
728        assert_eq!(handoff.checkpoint_assignment_version(), 7);
729        assert_eq!(
730            handoff.cluster_watermark(),
731            CheckpointWatermark::Active(900)
732        );
733        assert_eq!(handoff.recovery_watermark_frontier(), Some(900));
734        assert_eq!(handoff.source_count(), 1);
735        assert!(!handoff.is_empty());
736
737        let events = handoff.source("events").unwrap();
738        assert_eq!(
739            events
740                .checkpoint()
741                .offsets
742                .get("partition:0")
743                .map(String::as_str),
744            Some("41")
745        );
746        assert_eq!(
747            events
748                .checkpoint()
749                .metadata
750                .get("topic")
751                .map(String::as_str),
752            Some("events")
753        );
754        assert_eq!(
755            events.checkpoint().source_assignment_version,
756            NonZeroU64::new(7)
757        );
758        assert_eq!(events.watermark(), Some(1_000));
759        assert!(handoff.source("missing").is_none());
760    }
761
762    #[test]
763    fn source_assignment_versions_are_sparse_and_fenced() {
764        let mut sparse = capsule();
765        sparse.source_assignment_versions.clear();
766        sparse.validate().unwrap();
767        let sparse_handoff = CommittedSourceHandoff::try_from(&sparse).unwrap();
768        assert_eq!(
769            sparse_handoff
770                .source("events")
771                .unwrap()
772                .checkpoint()
773                .source_assignment_version,
774            None
775        );
776
777        let mut unknown = capsule();
778        unknown.source_assignment_versions.insert(
779            "missing".into(),
780            NonZeroU64::new(unknown.assignment_fence.assignment_version).unwrap(),
781        );
782        assert!(unknown
783            .validate()
784            .unwrap_err()
785            .contains("unknown source 'missing'"));
786
787        let mut mismatched = capsule();
788        mismatched
789            .source_assignment_versions
790            .insert("events".into(), NonZeroU64::new(8).unwrap());
791        assert!(mismatched
792            .validate()
793            .unwrap_err()
794            .contains("for 'events' is 8; expected 7"));
795    }
796
797    #[test]
798    fn malformed_digest_and_source_metadata_fail_closed() {
799        let mut bad_digest = capsule();
800        bad_digest.participants[0].manifest_sha256 = "AB".repeat(32);
801        assert!(bad_digest.validate().is_err());
802
803        let mut missing_metadata = capsule();
804        missing_metadata.source_metadata.clear();
805        assert!(missing_metadata
806            .validate()
807            .unwrap_err()
808            .contains("same sources"));
809        assert!(CommittedSourceHandoff::try_from(&missing_metadata).is_err());
810
811        let mut divergent_state = capsule();
812        divergent_state.participants[1].portable_state_sha256 = digest(0xaa);
813        assert!(divergent_state
814            .validate()
815            .unwrap_err()
816            .contains("divergent portable state"));
817    }
818
819    #[test]
820    fn capsule_preserves_non_active_watermark_states() {
821        for (watermark, frontier) in [
822            (CheckpointWatermark::Uninitialized, None),
823            (CheckpointWatermark::Idle, None),
824            (CheckpointWatermark::Idle, Some(800)),
825        ] {
826            let mut capsule = capsule();
827            capsule.cluster_watermark = watermark;
828            capsule.recovery_watermark_frontier = frontier;
829            let (encoded, _) = capsule.encode_and_reference().unwrap();
830            let decoded: ClusterRecoveryCapsule = serde_json::from_slice(&encoded).unwrap();
831            assert_eq!(decoded.cluster_watermark, watermark);
832            assert_eq!(decoded.recovery_watermark_frontier, frontier);
833            let handoff = CommittedSourceHandoff::try_from(&decoded).unwrap();
834            assert_eq!(handoff.cluster_watermark(), watermark);
835            assert_eq!(handoff.recovery_watermark_frontier(), frontier);
836        }
837    }
838
839    #[test]
840    fn recovery_watermark_status_and_frontier_must_agree() {
841        let mut reserved = capsule();
842        reserved.cluster_watermark = CheckpointWatermark::Active(i64::MIN);
843        reserved.recovery_watermark_frontier = Some(i64::MIN);
844        assert!(reserved
845            .validate()
846            .unwrap_err()
847            .contains("uninitialized sentinel"));
848
849        let mut missing = capsule();
850        missing.recovery_watermark_frontier = None;
851        assert!(missing
852            .validate()
853            .unwrap_err()
854            .contains("must equal its recovery frontier"));
855
856        let mut mismatched = capsule();
857        mismatched.recovery_watermark_frontier = Some(899);
858        assert!(mismatched
859            .validate()
860            .unwrap_err()
861            .contains("must equal its recovery frontier"));
862
863        let mut uninitialized = capsule();
864        uninitialized.cluster_watermark = CheckpointWatermark::Uninitialized;
865        assert!(uninitialized
866            .validate()
867            .unwrap_err()
868            .contains("cannot have a recovery frontier"));
869
870        let mut idle_reserved = capsule();
871        idle_reserved.cluster_watermark = CheckpointWatermark::Idle;
872        idle_reserved.recovery_watermark_frontier = Some(i64::MIN);
873        assert!(idle_reserved
874            .validate()
875            .unwrap_err()
876            .contains("uninitialized sentinel"));
877    }
878
879    #[test]
880    fn recovery_watermark_frontier_must_not_exceed_sources() {
881        let mut ahead = capsule();
882        ahead.cluster_watermark = CheckpointWatermark::Idle;
883        ahead.recovery_watermark_frontier = Some(1_001);
884        assert!(ahead
885            .validate()
886            .unwrap_err()
887            .contains("frontier exceeds a source watermark"));
888
889        let mut at_source = capsule();
890        at_source.cluster_watermark = CheckpointWatermark::Active(1_000);
891        at_source.recovery_watermark_frontier = Some(1_000);
892        at_source.validate().unwrap();
893    }
894}
laminar_core/checkpoint/recovery_capsule.rs

laminar_core/checkpoint/
recovery_capsule.rs
