laminar_core/checkpoint/
checkpoint_manifest.rs

1//! Checkpoint manifest types.
2//!
3//! Manifests are JSON for debuggability. Large operator state goes into a
4//! separate `state.bin` sidecar referenced by offset/length in the manifest.
5
6#![allow(clippy::disallowed_types)] // cold path: manifest serialization
7
8use std::collections::HashMap;
9
10use crate::state::{KeyGroupCount, LOCAL_KEY_GROUP_COUNT, PARTITIONING_ABI_VERSION};
11
12/// Current checkpoint manifest format. Older manifests are rejected rather
13/// than guessed at recovery time.
14/// Version 6 binds recovery to the durable key-partitioning ABI.
15pub const CHECKPOINT_MANIFEST_VERSION: u32 = 6;
16
17/// Canonical pipeline-identity payload version.
18pub const PIPELINE_IDENTITY_VERSION: u16 = 3;
19
20/// SHA-256 identity of the logical pipeline and recovery-state ABI.
21///
22/// The canonical version is part of the persisted contract: changing canonicalization or state
23/// compatibility requires a new version rather than silently comparing unlike digests.
24#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
25pub struct PipelineIdentity {
26    /// Version of the canonical payload format.
27    pub canonical_version: u16,
28    /// Exactly 64 lowercase hexadecimal characters.
29    pub sha256: String,
30}
31
32impl PipelineIdentity {
33    /// Identity of an empty canonical payload, used by manifest-only tests and empty runtimes.
34    #[must_use]
35    pub fn empty() -> Self {
36        Self {
37            canonical_version: PIPELINE_IDENTITY_VERSION,
38            sha256: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".into(),
39        }
40    }
41
42    /// Validate the persisted identity format.
43    pub(crate) fn validation_error(&self) -> Option<String> {
44        if self.canonical_version != PIPELINE_IDENTITY_VERSION {
45            return Some(format!(
46                "unsupported pipeline identity version {}; expected {PIPELINE_IDENTITY_VERSION}",
47                self.canonical_version
48            ));
49        }
50        if self.sha256.len() != 64
51            || !self
52                .sha256
53                .bytes()
54                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
55        {
56            return Some("pipeline identity must be 64 lowercase hexadecimal characters".into());
57        }
58        None
59    }
60
61    /// Whether this identity uses the current canonical version and digest encoding.
62    #[must_use]
63    pub(crate) fn is_canonical(&self) -> bool {
64        self.validation_error().is_none()
65    }
66}
67
68/// Durable publication state of a checkpoint manifest.
69///
70/// `Prepared` records are inventory, not an independent commit signal. Recovery may promote one
71/// only when the exact durable checkpoint decision exists. `Finalized` records are published
72/// recovery candidates, still subject to that decision whenever a decision store is configured.
73#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
74#[serde(rename_all = "snake_case")]
75pub enum DurableCheckpointPhase {
76    /// State has been captured but the checkpoint has not completed.
77    Prepared,
78    /// The checkpoint completed and is eligible for recovery.
79    Finalized,
80}
81
82/// A point-in-time snapshot of all pipeline state.
83#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
84pub struct CheckpointManifest {
85    /// Manifest format version (for future evolution).
86    pub version: u32,
87    /// Unique, monotonically increasing checkpoint ID; must equal `epoch`.
88    pub checkpoint_id: u64,
89    /// The same checkpoint ID retained in the persisted field named `epoch`.
90    pub epoch: u64,
91    /// Timestamp when checkpoint was created (millis since Unix epoch).
92    pub timestamp_ms: u64,
93    /// Durable publication state. This field is intentionally required.
94    pub durable_phase: DurableCheckpointPhase,
95    /// Writer identity (`0` outside cluster mode).
96    pub participant_id: u64,
97
98    // ── Connector State ──
99    /// Per-source connector offsets (key: source name).
100    #[serde(default)]
101    pub source_offsets: HashMap<String, ConnectorCheckpoint>,
102    /// Per-table source offsets for reference tables (key: table name).
103    #[serde(default)]
104    pub table_offsets: HashMap<String, ConnectorCheckpoint>,
105
106    // ── Operator State ──
107    /// Per-operator checkpoint data (key: operator/node name).
108    ///
109    /// Small state is inlined as base64. Large state is stored in a separate
110    /// `state.bin` file and this map holds only a reference marker.
111    #[serde(default)]
112    pub operator_states: HashMap<String, OperatorCheckpoint>,
113
114    // ── Storage State ──
115    /// Path to the table store checkpoint, if any.
116    #[serde(default)]
117    pub table_store_checkpoint_path: Option<String>,
118    // ── Time State ──
119    /// Global watermark at checkpoint time.
120    #[serde(default)]
121    pub watermark: Option<i64>,
122    /// Per-source watermarks (key: source name).
123    #[serde(default)]
124    pub source_watermarks: HashMap<String, i64>,
125
126    // ── Topology ──
127    /// Sorted names of all registered sources at checkpoint time.
128    ///
129    /// Used during recovery to detect topology changes (added/removed sources)
130    /// and warn the operator.
131    #[serde(default)]
132    pub source_names: Vec<String>,
133    /// Sorted names of all registered sinks at checkpoint time.
134    #[serde(default)]
135    pub sink_names: Vec<String>,
136
137    // ── Pipeline Identity ──
138    /// Required, deterministic identity of the logical topology and state ABI.
139    pub pipeline_identity: PipelineIdentity,
140    /// Create-once checkpoint/decision-store incarnation. A storage reset rotates this value so
141    /// surviving external sink cursors cannot be reused by a fresh checkpoint-id sequence.
142    pub deployment_id: String,
143
144    // ── Metadata ──
145    /// Durable key encoding, hashing, and key-group mapping contract.
146    pub partitioning_abi_version: u16,
147    /// Virtual partition count for state key distribution.
148    #[serde(default)]
149    pub vnode_count: u16,
150
151    // ── Integrity ──
152    /// SHA-256 hex digest of the sidecar `state.bin` file (if any).
153    ///
154    /// Written during checkpoint commit so that recovery can verify the
155    /// sidecar hasn't been corrupted or truncated on disk/S3.
156    #[serde(default)]
157    pub state_checksum: Option<String>,
158}
159
160/// Errors found during manifest validation.
161#[derive(Debug, Clone, PartialEq, Eq)]
162pub struct ManifestValidationError {
163    /// Human-readable description of the issue.
164    pub message: String,
165}
166
167impl std::fmt::Display for ManifestValidationError {
168    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
169        write!(f, "{}", self.message)
170    }
171}
172
173impl CheckpointManifest {
174    /// Validates manifest consistency before recovery.
175    ///
176    /// `expected_key_group_count` is the runtime's configured key-group count;
177    /// a manifest written with a different count can't be safely restored
178    /// because state keys won't map to the same shards.
179    ///
180    /// Returns a list of issues found. An empty list means the manifest is valid.
181    /// Every returned issue makes the manifest ineligible for recovery.
182    #[must_use]
183    pub fn validate(
184        &self,
185        expected_key_group_count: KeyGroupCount,
186    ) -> Vec<ManifestValidationError> {
187        let mut errors = Vec::new();
188
189        if self.version != CHECKPOINT_MANIFEST_VERSION {
190            errors.push(ManifestValidationError {
191                message: format!(
192                    "unsupported manifest version {}; expected {CHECKPOINT_MANIFEST_VERSION}",
193                    self.version
194                ),
195            });
196        }
197
198        if self.partitioning_abi_version != PARTITIONING_ABI_VERSION {
199            errors.push(ManifestValidationError {
200                message: format!(
201                    "partitioning ABI mismatch: checkpoint has {}, runtime expects {PARTITIONING_ABI_VERSION}",
202                    self.partitioning_abi_version
203                ),
204            });
205        }
206
207        if self.checkpoint_id == 0 || self.epoch != self.checkpoint_id {
208            errors.push(ManifestValidationError {
209                message: "checkpoint attempt must use one nonzero canonical checkpoint ID".into(),
210            });
211        }
212
213        if self.timestamp_ms == 0 {
214            errors.push(ManifestValidationError {
215                message: "timestamp_ms is 0 (missing creation time)".into(),
216            });
217        }
218
219        if let Some(message) = self.pipeline_identity.validation_error() {
220            errors.push(ManifestValidationError { message });
221        }
222        if !self.deployment_id.is_empty() {
223            let valid = uuid::Uuid::parse_str(&self.deployment_id)
224                .is_ok_and(|id| !id.is_nil() && id.to_string() == self.deployment_id);
225            if !valid {
226                errors.push(ManifestValidationError {
227                    message: "deployment_id must be a canonical non-nil UUID".into(),
228                });
229            }
230        }
231
232        // Source offsets should reference known sources (if topology is recorded)
233        if !self.source_names.is_empty() {
234            for name in self.source_offsets.keys() {
235                if !self.source_names.contains(name) {
236                    errors.push(ManifestValidationError {
237                        message: format!("source_offsets contains '{name}' not in source_names"),
238                    });
239                }
240            }
241        }
242
243        if self.vnode_count == 0 {
244            errors.push(ManifestValidationError {
245                message: "vnode_count is 0 (missing or legacy checkpoint)".into(),
246            });
247        } else if self.vnode_count != expected_key_group_count.get() {
248            errors.push(ManifestValidationError {
249                message: format!(
250                    "vnode_count mismatch: checkpoint has {}, runtime expects {expected_key_group_count}",
251                    self.vnode_count,
252                ),
253            });
254        }
255
256        if !self.operator_states.is_empty() && self.state_checksum.is_none() {
257            errors.push(ManifestValidationError {
258                message: "operator state is missing its integrity checksum".into(),
259            });
260        }
261
262        errors
263    }
264
265    /// Creates a new manifest for an embedded or single-node runtime.
266    ///
267    /// Validation rejects values where `checkpoint_id` and `epoch` differ.
268    #[must_use]
269    pub fn new(checkpoint_id: u64, epoch: u64) -> Self {
270        Self::new_with_key_group_count(checkpoint_id, epoch, LOCAL_KEY_GROUP_COUNT)
271    }
272
273    /// Creates a new manifest with an explicit stable key-group count.
274    ///
275    /// Validation rejects values where `checkpoint_id` and `epoch` differ.
276    #[must_use]
277    pub fn new_with_key_group_count(
278        checkpoint_id: u64,
279        epoch: u64,
280        key_group_count: KeyGroupCount,
281    ) -> Self {
282        #[allow(clippy::cast_possible_truncation)] // u64 millis won't overflow until year 584M
283        let timestamp_ms = std::time::SystemTime::now()
284            .duration_since(std::time::UNIX_EPOCH)
285            .unwrap_or_default()
286            .as_millis() as u64;
287
288        Self {
289            version: CHECKPOINT_MANIFEST_VERSION,
290            checkpoint_id,
291            epoch,
292            timestamp_ms,
293            durable_phase: DurableCheckpointPhase::Prepared,
294            participant_id: 0,
295            source_offsets: HashMap::new(),
296            table_offsets: HashMap::new(),
297            operator_states: HashMap::new(),
298            table_store_checkpoint_path: None,
299            watermark: None,
300            source_watermarks: HashMap::new(),
301            source_names: Vec::new(),
302            sink_names: Vec::new(),
303            pipeline_identity: PipelineIdentity::empty(),
304            deployment_id: String::new(),
305            partitioning_abi_version: PARTITIONING_ABI_VERSION,
306            vnode_count: key_group_count.get(),
307            state_checksum: None,
308        }
309    }
310}
311
312/// Connector-agnostic offset container.
313///
314/// Uses string key-value pairs to support all connector types:
315/// - **Kafka**: `{"events:0": "1234", "events:1": "5678"}`
316/// - **`PostgreSQL` CDC**: `{"lsn": "0/1234ABCD"}`
317/// - **Delta Lake**: `{"version": "42"}`
318///
319/// The containing [`CheckpointManifest`] supplies the exact attempt identity;
320/// duplicating its epoch in each connector payload would create conflicting
321/// authorities during recovery.
322#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
323pub struct ConnectorCheckpoint {
324    /// Connector-specific offset data.
325    pub offsets: HashMap<String, String>,
326    /// Optional metadata (connector type, topic name, etc.).
327    pub metadata: HashMap<String, String>,
328    /// Provider-neutral source-assignment version that owns this offset cut.
329    ///
330    /// `None` is valid for sources that do not participate in partition assignment. Cluster
331    /// recovery validates populated versions against the checkpoint assignment fence.
332    pub source_assignment_version: Option<std::num::NonZeroU64>,
333}
334
335impl ConnectorCheckpoint {
336    /// Creates an empty connector checkpoint.
337    #[must_use]
338    pub fn new() -> Self {
339        Self::default()
340    }
341
342    /// Creates a connector checkpoint with pre-populated offsets.
343    #[must_use]
344    pub fn with_offsets(offsets: HashMap<String, String>) -> Self {
345        Self {
346            offsets,
347            metadata: HashMap::new(),
348            source_assignment_version: None,
349        }
350    }
351}
352
353/// Serialized operator state stored in the manifest.
354#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
355pub struct OperatorCheckpoint {
356    /// Base64-encoded binary state (for small payloads inlined in JSON).
357    #[serde(default)]
358    pub state_b64: Option<String>,
359    /// If true, state is stored externally in the state.bin sidecar file.
360    #[serde(default)]
361    pub external: bool,
362    /// Byte offset into the state.bin file (if external).
363    #[serde(default)]
364    pub external_offset: u64,
365    /// Byte length of the state in the state.bin file (if external).
366    #[serde(default)]
367    pub external_length: u64,
368}
369
370impl OperatorCheckpoint {
371    /// Creates an inline operator checkpoint from raw bytes.
372    ///
373    /// The bytes are base64-encoded for JSON storage.
374    #[must_use]
375    pub fn inline(data: &[u8]) -> Self {
376        use base64::Engine;
377        Self {
378            state_b64: Some(base64::engine::general_purpose::STANDARD.encode(data)),
379            external: false,
380            external_offset: 0,
381            external_length: 0,
382        }
383    }
384
385    /// Creates an external reference to state in the sidecar file.
386    #[must_use]
387    pub fn external(offset: u64, length: u64) -> Self {
388        Self {
389            state_b64: None,
390            external: true,
391            external_offset: offset,
392            external_length: length,
393        }
394    }
395
396    /// Decodes the inline state, returning the raw bytes.
397    ///
398    /// Returns `None` if the state is external, no inline data is present,
399    /// or if the base64 data is corrupted (logs a warning in that case).
400    #[must_use]
401    pub fn decode_inline(&self) -> Option<Vec<u8>> {
402        use base64::Engine;
403        self.state_b64.as_ref().and_then(|b64| {
404            match base64::engine::general_purpose::STANDARD.decode(b64) {
405                Ok(data) => Some(data),
406                Err(e) => {
407                    tracing::warn!(
408                        error = %e,
409                        b64_len = b64.len(),
410                        "[LDB-4004] Failed to decode inline operator state from base64 — \
411                         operator will start from scratch"
412                    );
413                    None
414                }
415            }
416        })
417    }
418
419    /// Decodes the inline state, returning a `Result` for callers that need
420    /// to distinguish between "no inline state" and "corrupted state".
421    ///
422    /// Returns `Ok(None)` if no inline data is present (external or absent).
423    /// Returns `Ok(Some(bytes))` on successful decode.
424    ///
425    /// # Errors
426    ///
427    /// Returns `Err` if base64 data is present but corrupted.
428    pub fn try_decode_inline(&self) -> Result<Option<Vec<u8>>, String> {
429        use base64::Engine;
430        match &self.state_b64 {
431            None => Ok(None),
432            Some(b64) => base64::engine::general_purpose::STANDARD
433                .decode(b64)
434                .map(Some)
435                .map_err(|e| format!("[LDB-4004] base64 decode failed: {e}")),
436        }
437    }
438
439    /// Creates an `OperatorCheckpoint` from raw bytes using a size threshold.
440    ///
441    /// If `data.len() <= threshold`, the state is inlined as base64.
442    /// If `data.len() > threshold`, the state is marked as external with the
443    /// given offset and length, and the raw data is returned for sidecar storage.
444    ///
445    /// # Arguments
446    ///
447    /// * `data` — Raw operator state bytes
448    /// * `threshold` — Maximum size in bytes for inline storage
449    /// * `current_offset` — Byte offset into the sidecar file for this blob
450    ///
451    /// # Returns
452    ///
453    /// A tuple of the checkpoint entry and optional raw data for the sidecar.
454    #[must_use]
455    #[allow(clippy::cast_possible_truncation)]
456    pub fn from_bytes(
457        data: &[u8],
458        threshold: usize,
459        current_offset: u64,
460    ) -> (Self, Option<Vec<u8>>) {
461        if data.len() <= threshold {
462            (Self::inline(data), None)
463        } else {
464            let length = data.len() as u64;
465            (Self::external(current_offset, length), Some(data.to_vec()))
466        }
467    }
468
469    /// Shared-buffer variant of [`Self::from_bytes`].
470    ///
471    /// Takes an owned [`bytes::Bytes`] and returns the same type on the
472    /// external path, avoiding the `data.to_vec()` copy the `&[u8]`
473    /// version has to make. The checkpoint pipeline passes rkyv output
474    /// through as `Bytes`, so per-operator state no longer doubles in
475    /// memory when crossing this boundary.
476    #[must_use]
477    #[allow(clippy::cast_possible_truncation)]
478    pub fn from_bytes_shared(
479        data: bytes::Bytes,
480        threshold: usize,
481        current_offset: u64,
482    ) -> (Self, Option<bytes::Bytes>) {
483        if data.len() <= threshold {
484            (Self::inline(&data), None)
485        } else {
486            let length = data.len() as u64;
487            (Self::external(current_offset, length), Some(data))
488        }
489    }
490}
491
492#[cfg(test)]
493mod tests {
494    use super::*;
495
496    #[test]
497    fn test_manifest_new() {
498        let m = CheckpointManifest::new(5, 5);
499        assert_eq!(m.version, CHECKPOINT_MANIFEST_VERSION);
500        assert_eq!(m.partitioning_abi_version, PARTITIONING_ABI_VERSION);
501        assert_eq!(m.durable_phase, DurableCheckpointPhase::Prepared);
502        assert_eq!(m.checkpoint_id, 5);
503        assert_eq!(m.epoch, 5);
504        assert_eq!(m.vnode_count, LOCAL_KEY_GROUP_COUNT.get());
505        assert!(m.timestamp_ms > 0);
506        assert!(m.source_offsets.is_empty());
507        assert!(m.operator_states.is_empty());
508    }
509
510    #[test]
511    fn test_manifest_new_with_explicit_key_group_count() {
512        let key_group_count = KeyGroupCount::try_from(256_u16).unwrap();
513        let manifest = CheckpointManifest::new_with_key_group_count(5, 5, key_group_count);
514
515        assert_eq!(manifest.vnode_count, key_group_count.get());
516        assert!(manifest.validate(key_group_count).is_empty());
517        assert!(!manifest.validate(LOCAL_KEY_GROUP_COUNT).is_empty());
518    }
519
520    #[test]
521    fn test_manifest_json_round_trip() {
522        let mut m = CheckpointManifest::new(42, 42);
523        let mut source_checkpoint = ConnectorCheckpoint::with_offsets(HashMap::from([
524            ("events:0".into(), "1234".into()),
525            ("events:1".into(), "5678".into()),
526        ]));
527        source_checkpoint.source_assignment_version = std::num::NonZeroU64::new(12);
528        m.source_offsets
529            .insert("kafka-src".into(), source_checkpoint);
530        m.watermark = Some(999_000);
531        m.operator_states
532            .insert("window-agg".into(), OperatorCheckpoint::inline(b"hello"));
533
534        let json = serde_json::to_string_pretty(&m).unwrap();
535        let restored: CheckpointManifest = serde_json::from_str(&json).unwrap();
536
537        assert_eq!(restored.checkpoint_id, 42);
538        assert_eq!(restored.epoch, 42);
539        assert_eq!(restored.watermark, Some(999_000));
540        let src = restored.source_offsets.get("kafka-src").unwrap();
541        assert_eq!(src.offsets.get("events:0"), Some(&"1234".into()));
542        assert_eq!(src.source_assignment_version, std::num::NonZeroU64::new(12));
543
544        let op = restored.operator_states.get("window-agg").unwrap();
545        assert_eq!(op.decode_inline().unwrap(), b"hello");
546    }
547
548    #[test]
549    fn test_manifest_rejects_previous_version() {
550        let mut manifest = CheckpointManifest::new(1, 1);
551        manifest.version = CHECKPOINT_MANIFEST_VERSION - 1;
552        let restored: CheckpointManifest =
553            serde_json::from_str(&serde_json::to_string(&manifest).unwrap()).unwrap();
554        let errors = restored.validate(LOCAL_KEY_GROUP_COUNT);
555        let previous = CHECKPOINT_MANIFEST_VERSION - 1;
556        assert!(
557            errors.iter().any(|error| error
558                .message
559                .contains(&format!("unsupported manifest version {previous}"))),
560            "{errors:?}"
561        );
562    }
563
564    #[test]
565    fn test_manifest_rejects_noncanonical_attempt_identity() {
566        for (checkpoint_id, epoch) in [(0, 0), (5, 0), (0, 5), (5, 6)] {
567            let manifest = CheckpointManifest::new(checkpoint_id, epoch);
568            let errors = manifest.validate(LOCAL_KEY_GROUP_COUNT);
569            assert!(
570                errors.iter().any(|error| error
571                    .message
572                    .contains("one nonzero canonical checkpoint ID")),
573                "{checkpoint_id}/{epoch}: {errors:?}"
574            );
575        }
576    }
577
578    #[test]
579    fn test_connector_checkpoint_new() {
580        let cp = ConnectorCheckpoint::new();
581        assert!(cp.offsets.is_empty());
582        assert!(cp.metadata.is_empty());
583        assert_eq!(cp.source_assignment_version, None);
584    }
585
586    #[test]
587    fn test_connector_checkpoint_with_offsets() {
588        let offsets = HashMap::from([("lsn".into(), "0/ABCD".into())]);
589        let cp = ConnectorCheckpoint::with_offsets(offsets);
590        assert_eq!(cp.offsets.get("lsn"), Some(&"0/ABCD".into()));
591        assert_eq!(cp.source_assignment_version, None);
592    }
593
594    #[test]
595    fn test_operator_checkpoint_inline() {
596        let op = OperatorCheckpoint::inline(b"state-data");
597        assert!(!op.external);
598        assert!(op.state_b64.is_some());
599        assert_eq!(op.decode_inline().unwrap(), b"state-data");
600    }
601
602    #[test]
603    fn test_operator_checkpoint_external() {
604        let op = OperatorCheckpoint::external(1024, 256);
605        assert!(op.external);
606        assert_eq!(op.external_offset, 1024);
607        assert_eq!(op.external_length, 256);
608        assert!(op.decode_inline().is_none());
609    }
610
611    #[test]
612    fn test_operator_checkpoint_empty_inline() {
613        let op = OperatorCheckpoint::inline(b"");
614        assert_eq!(op.decode_inline().unwrap(), b"");
615    }
616
617    #[test]
618    fn test_manifest_table_offsets() {
619        let mut m = CheckpointManifest::new(1, 1);
620        m.table_offsets.insert(
621            "instruments".into(),
622            ConnectorCheckpoint::with_offsets(HashMap::from([("lsn".into(), "0/ABCD".into())])),
623        );
624        m.table_store_checkpoint_path = Some("/tmp/table_store_cp".into());
625
626        let json = serde_json::to_string(&m).unwrap();
627        let restored: CheckpointManifest = serde_json::from_str(&json).unwrap();
628
629        assert_eq!(restored.table_offsets.len(), 1);
630        assert_eq!(
631            restored.table_store_checkpoint_path.as_deref(),
632            Some("/tmp/table_store_cp")
633        );
634    }
635
636    #[test]
637    fn test_manifest_topology_fields_round_trip() {
638        let mut m = CheckpointManifest::new(1, 1);
639        m.source_names = vec!["kafka-clicks".into(), "ws-prices".into()];
640        m.sink_names = vec!["pg-sink".into()];
641
642        let json = serde_json::to_string(&m).unwrap();
643        let restored: CheckpointManifest = serde_json::from_str(&json).unwrap();
644
645        assert_eq!(restored.source_names, vec!["kafka-clicks", "ws-prices"]);
646        assert_eq!(restored.sink_names, vec!["pg-sink"]);
647    }
648
649    #[test]
650    fn test_manifest_requires_durable_phase() {
651        let json = r#"{
652            "version": 1,
653            "checkpoint_id": 5,
654            "epoch": 3,
655            "timestamp_ms": 1000
656        }"#;
657        assert!(serde_json::from_str::<CheckpointManifest>(json).is_err());
658    }
659
660    #[test]
661    fn test_manifest_requires_pipeline_identity() {
662        let manifest = CheckpointManifest::new(5, 5);
663        let mut value = serde_json::to_value(manifest).unwrap();
664        value.as_object_mut().unwrap().remove("pipeline_identity");
665        assert!(serde_json::from_value::<CheckpointManifest>(value).is_err());
666    }
667
668    #[test]
669    fn test_manifest_requires_partitioning_abi() {
670        let manifest = CheckpointManifest::new(5, 5);
671        let mut value = serde_json::to_value(manifest).unwrap();
672        value
673            .as_object_mut()
674            .unwrap()
675            .remove("partitioning_abi_version");
676        assert!(serde_json::from_value::<CheckpointManifest>(value).is_err());
677    }
678
679    #[test]
680    fn test_manifest_rejects_wrong_partitioning_abi() {
681        let mut manifest = CheckpointManifest::new(5, 5);
682        manifest.partitioning_abi_version = PARTITIONING_ABI_VERSION + 1;
683
684        let errors = manifest.validate(LOCAL_KEY_GROUP_COUNT);
685        assert!(
686            errors
687                .iter()
688                .any(|error| error.message.contains("partitioning ABI mismatch")),
689            "{errors:?}"
690        );
691    }
692
693    #[test]
694    fn test_validate_orphaned_source_offset() {
695        let mut m = CheckpointManifest::new(1, 1);
696        m.source_names = vec!["a".into(), "b".into()];
697        m.source_offsets
698            .insert("c".into(), ConnectorCheckpoint::new());
699
700        let errors = m.validate(LOCAL_KEY_GROUP_COUNT);
701        assert!(
702            errors
703                .iter()
704                .any(|e| e.message.contains("'c' not in source_names")),
705            "expected orphaned source offset error: {errors:?}"
706        );
707    }
708
709    #[test]
710    fn test_manifest_pipeline_identity_round_trip() {
711        let mut m = CheckpointManifest::new(1, 1);
712        m.pipeline_identity = PipelineIdentity {
713            canonical_version: PIPELINE_IDENTITY_VERSION,
714            sha256: "ab".repeat(32),
715        };
716
717        let json = serde_json::to_string(&m).unwrap();
718        let restored: CheckpointManifest = serde_json::from_str(&json).unwrap();
719
720        assert_eq!(restored.pipeline_identity, m.pipeline_identity);
721    }
722
723    #[test]
724    fn test_from_bytes_inline() {
725        let data = b"small-state";
726        let (op, sidecar) = OperatorCheckpoint::from_bytes(data, 1024, 0);
727        assert!(!op.external);
728        assert!(sidecar.is_none());
729        assert_eq!(op.decode_inline().unwrap(), data);
730    }
731
732    #[test]
733    fn test_from_bytes_external() {
734        let data = vec![0xAB; 2048];
735        let (op, sidecar) = OperatorCheckpoint::from_bytes(&data, 1024, 512);
736        assert!(op.external);
737        assert_eq!(op.external_offset, 512);
738        assert_eq!(op.external_length, 2048);
739        assert!(op.decode_inline().is_none());
740        assert_eq!(sidecar.unwrap(), data);
741    }
742
743    #[test]
744    fn test_from_bytes_at_threshold_boundary() {
745        // Exactly at threshold → inline
746        let data = vec![0xFF; 100];
747        let (op, sidecar) = OperatorCheckpoint::from_bytes(&data, 100, 0);
748        assert!(!op.external);
749        assert!(sidecar.is_none());
750        assert_eq!(op.decode_inline().unwrap(), data);
751
752        // One byte over threshold → external
753        let data_over = vec![0xFF; 101];
754        let (op2, sidecar2) = OperatorCheckpoint::from_bytes(&data_over, 100, 0);
755        assert!(op2.external);
756        assert!(sidecar2.is_some());
757    }
758
759    #[test]
760    fn test_from_bytes_empty_data() {
761        let (op, sidecar) = OperatorCheckpoint::from_bytes(b"", 1024, 0);
762        assert!(!op.external);
763        assert!(sidecar.is_none());
764        assert_eq!(op.decode_inline().unwrap(), b"");
765    }
766
767    #[test]
768    fn test_manifest_rejects_missing_v2_fields() {
769        let json = r#"{
770            "version": 1,
771            "checkpoint_id": 1,
772            "epoch": 1,
773            "timestamp_ms": 1000
774        }"#;
775        assert!(serde_json::from_str::<CheckpointManifest>(json).is_err());
776    }
777}
laminar_core/checkpoint/checkpoint_manifest.rs

laminar_core/checkpoint/
checkpoint_manifest.rs
