laminar_core/checkpoint/
mod.rs

1//! Checkpoint barrier protocol and storage.
2//!
3//! Coordinator-triggered barriers flow through sources to trigger consistent
4//! state snapshots. The fast path is a single `AtomicU64` load (~10ns).
5
6/// Checkpoint barrier types and cross-thread injection.
7pub mod barrier;
8
9/// Identity and ordering for checkpoint attempts.
10pub mod attempt;
11
12/// Feature-neutral assignment certificate retained by exact checkpoint attempts.
13pub mod assignment;
14
15/// Feature-neutral leader authority retained by durable protocol records.
16pub mod authority;
17
18/// Unified checkpoint manifest types
19pub mod checkpoint_manifest;
20
21/// Checkpoint persistence trait and filesystem/object store implementations
22pub mod checkpoint_store;
23
24/// Object store factory — builds S3, GCS, Azure, or local backends from URL schemes.
25pub mod object_store_builder;
26
27/// Canonical global index selected by a committed checkpoint.
28pub mod committed_checkpoint;
29
30/// Durable identities and canonical metadata for committed subscription output.
31pub mod subscription;
32
33pub use assignment::{
34    AssignmentDrainId, AssignmentDrainTransition, CheckpointAssignmentAdoption,
35    CheckpointAssignmentFence, CheckpointParticipant, MAX_CHECKPOINT_PARTICIPANTS,
36};
37pub use attempt::{CheckpointAttempt, CheckpointAttemptRelation};
38pub use authority::{LeaderProof, LeaderProofOwner};
39pub use barrier::{
40    flags, BarrierPollHandle, CheckpointBarrier, CheckpointBarrierInjector, StreamMessage,
41};
42
43pub use checkpoint_manifest::{
44    checkpoint_artifact_intent_sha256, checkpoint_descriptor_sha256, checkpoint_sha256, ByteRange,
45    ChannelProgress, CheckpointManifest, ConnectorCheckpoint, NodeDataObject, PipelineIdentity,
46    PreparedSinkArtifactIntent, PreparedSinkDescriptor, ReferencedStateChunk, StateChunkId,
47    StateFrame, StateFrameKey, PIPELINE_IDENTITY_VERSION, PREPARED_SINK_DESCRIPTOR_VERSION,
48};
49pub use checkpoint_store::{
50    checkpoint_artifact_identity_sha256, checkpoint_manifest_bytes,
51    probe_object_store_conditional_create, probe_object_store_conditional_update,
52    CheckpointManifestAbortSeal, CheckpointSinkArtifactIntent, CheckpointStore,
53    CheckpointStoreError, ObjectStoreCheckpointStore,
54    MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_AGGREGATE_BYTES, MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_BYTES,
55};
56pub use committed_checkpoint::{
57    canonical_json_bytes, canonical_json_sha256, CheckpointScope, CheckpointWatermark,
58    CommittedCheckpointIndex, CommittedCheckpointRef, CommittedParticipantRef,
59    COMMITTED_CHECKPOINT_INDEX_VERSION, MAX_COMMITTED_CHECKPOINT_INDEX_BYTES,
60};
61pub use subscription::{
62    merge_node_subscription_manifests, ChangelogMode, MergedSubscriptionCheckpoint,
63    NodePartitionRange, NodeSubscriptionManifest, NodeSubscriptionStreamManifest,
64    OutputDistribution, OutputDistributionCertificate, OutputFrameId, OutputPartitionId,
65    OutputSegmentRef, PartitionFrontier, PartitionSequence, StreamGeneration,
66    SubscriptionCheckpointManifest, SubscriptionContractError, SubscriptionDigest,
67    SubscriptionProtocolVersion, MAX_OUTPUT_FRAMES_PER_SEGMENT, MAX_OUTPUT_SEGMENT_BYTES,
68    OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION, SUBSCRIPTION_PROTOCOL_VERSION,
69};
70
71/// Reserved input-channel identity for one logical watermark per source and participant.
72pub const SINGLETON_WATERMARK_CHANNEL: &[u8] = b"\0";
73
74/// Reconstruct the event-time state represented by an exact channel cut.
75///
76/// # Errors
77/// Returns an error when a channel uses the reserved uninitialized watermark value.
78pub fn classify_channel_progress(
79    channels: &[ChannelProgress],
80) -> Result<CheckpointWatermark, String> {
81    let mut minimum = None;
82    for channel in channels {
83        if channel.watermark == Some(i64::MIN) {
84            return Err("channel progress uses the reserved uninitialized watermark".into());
85        }
86        if channel.idle {
87            continue;
88        }
89        let Some(watermark) = channel.watermark else {
90            return Ok(CheckpointWatermark::Uninitialized);
91        };
92        minimum = Some(minimum.map_or(watermark, |current: i64| current.min(watermark)));
93    }
94    Ok(minimum.map_or(CheckpointWatermark::Idle, CheckpointWatermark::Active))
95}
96
97/// Numeric frontier retained by a channel cut, including an all-idle cut.
98///
99/// # Errors
100/// Returns an error when the channel cut contains an invalid watermark sentinel.
101pub fn channel_progress_frontier(channels: &[ChannelProgress]) -> Result<Option<i64>, String> {
102    Ok(match classify_channel_progress(channels)? {
103        CheckpointWatermark::Active(watermark) => Some(watermark),
104        CheckpointWatermark::Idle => channels
105            .iter()
106            .filter_map(|channel| channel.watermark)
107            .max(),
108        CheckpointWatermark::Uninitialized => None,
109    })
110}
111
112/// Numeric decision frontier retained for each source in an exact channel cut.
113///
114/// Active channels contribute their minimum. An active uninitialized channel withholds only its
115/// own source, while an all-idle source retains the greatest initialized channel watermark. This
116/// is the source-keyed counterpart of [`channel_progress_frontier`]; callers must not use the
117/// pipeline-wide minimum when advancing a source-specific ordered operator.
118///
119/// # Errors
120/// Returns an error when a channel uses the reserved uninitialized watermark value.
121pub fn channel_progress_frontiers_by_source(
122    channels: &[ChannelProgress],
123) -> Result<std::collections::BTreeMap<&str, Option<i64>>, String> {
124    #[derive(Default)]
125    struct SourceProgress {
126        active_minimum: Option<i64>,
127        active_uninitialized: bool,
128        idle_maximum: Option<i64>,
129    }
130
131    let mut progress = std::collections::BTreeMap::<&str, SourceProgress>::new();
132    for channel in channels {
133        if channel.watermark == Some(i64::MIN) {
134            return Err("channel progress uses the reserved uninitialized watermark".into());
135        }
136        let source = progress.entry(channel.source_name.as_str()).or_default();
137        if channel.idle {
138            if let Some(watermark) = channel.watermark {
139                source.idle_maximum = Some(
140                    source
141                        .idle_maximum
142                        .map_or(watermark, |current| current.max(watermark)),
143                );
144            }
145        } else if let Some(watermark) = channel.watermark {
146            source.active_minimum = Some(
147                source
148                    .active_minimum
149                    .map_or(watermark, |current| current.min(watermark)),
150            );
151        } else {
152            source.active_uninitialized = true;
153        }
154    }
155
156    Ok(progress
157        .into_iter()
158        .map(|(source, progress)| {
159            let frontier = if progress.active_uninitialized {
160                None
161            } else {
162                progress.active_minimum.or(progress.idle_maximum)
163            };
164            (source, frontier)
165        })
166        .collect())
167}
laminar_core/checkpoint/mod.rs

laminar_core/checkpoint/
mod.rs
