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,
45    view_log_segment_limit, ByteRange, ChannelProgress, CheckpointManifest, ConnectorCheckpoint,
46    EventTimeBounds, NodeDataObject, PipelineIdentity, PreparedSinkArtifactIntent,
47    PreparedSinkDescriptor, ReferencedStateChunk, StateChunkId, StateFrame, StateFrameKey, ViewLog,
48    ViewLogSegment, MAX_CLUSTER_VIEW_LOG_SEGMENTS, MAX_VIEW_LOG_SEGMENTS,
49    PIPELINE_IDENTITY_VERSION, PREPARED_SINK_DESCRIPTOR_VERSION,
50};
51pub use checkpoint_store::{
52    checkpoint_artifact_identity_sha256, checkpoint_manifest_bytes,
53    probe_object_store_conditional_create, probe_object_store_conditional_update,
54    CheckpointManifestAbortSeal, CheckpointSinkArtifactIntent, CheckpointStore,
55    CheckpointStoreError, ObjectStoreCheckpointStore,
56    MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_AGGREGATE_BYTES, MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_BYTES,
57};
58pub use committed_checkpoint::{
59    adopted_view_logs, canonical_json_bytes, canonical_json_sha256, check_view_log_adoption,
60    fold_view_frontiers, validate_cut_view_logs, validate_departed_view_logs, AdoptedSegment,
61    AdoptedViewLog, CheckpointScope, CheckpointWatermark, CommittedCheckpointIndex,
62    CommittedCheckpointRef, CommittedParticipantRef, COMMITTED_CHECKPOINT_INDEX_VERSION,
63    MAX_COMMITTED_CHECKPOINT_INDEX_BYTES,
64};
65pub use subscription::{
66    merge_node_subscription_manifests, ChangelogMode, MergedSubscriptionCheckpoint,
67    NodePartitionRange, NodeSubscriptionManifest, NodeSubscriptionStreamManifest,
68    OutputDistribution, OutputDistributionCertificate, OutputFrameId, OutputPartitionId,
69    OutputSegmentRef, PartitionFrontier, PartitionSequence, StreamGeneration,
70    SubscriptionCheckpointManifest, SubscriptionContractError, SubscriptionDigest,
71    SubscriptionProtocolVersion, MAX_OUTPUT_FRAMES_PER_SEGMENT, MAX_OUTPUT_SEGMENT_BYTES,
72    OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION, SUBSCRIPTION_PROTOCOL_VERSION,
73};
74
75/// Reserved input-channel identity for one logical watermark per source and participant.
76pub const SINGLETON_WATERMARK_CHANNEL: &[u8] = b"\0";
77
78/// Reconstruct the event-time state represented by an exact channel cut.
79///
80/// # Errors
81/// Returns an error when a channel uses the reserved uninitialized watermark value.
82pub fn classify_channel_progress(
83    channels: &[ChannelProgress],
84) -> Result<CheckpointWatermark, String> {
85    let mut minimum = None;
86    for channel in channels {
87        if channel.watermark == Some(i64::MIN) {
88            return Err("channel progress uses the reserved uninitialized watermark".into());
89        }
90        if channel.idle {
91            continue;
92        }
93        let Some(watermark) = channel.watermark else {
94            return Ok(CheckpointWatermark::Uninitialized);
95        };
96        minimum = Some(minimum.map_or(watermark, |current: i64| current.min(watermark)));
97    }
98    Ok(minimum.map_or(CheckpointWatermark::Idle, CheckpointWatermark::Active))
99}
100
101/// Numeric frontier retained by a channel cut, including an all-idle cut.
102///
103/// # Errors
104/// Returns an error when the channel cut contains an invalid watermark sentinel.
105pub fn channel_progress_frontier(channels: &[ChannelProgress]) -> Result<Option<i64>, String> {
106    Ok(match classify_channel_progress(channels)? {
107        CheckpointWatermark::Active(watermark) => Some(watermark),
108        CheckpointWatermark::Idle => channels
109            .iter()
110            .filter_map(|channel| channel.watermark)
111            .max(),
112        CheckpointWatermark::Uninitialized => None,
113    })
114}
115
116/// Numeric decision frontier retained for each source in an exact channel cut.
117///
118/// Active channels contribute their minimum. An active uninitialized channel withholds only its
119/// own source, while an all-idle source retains the greatest initialized channel watermark. This
120/// is the source-keyed counterpart of [`channel_progress_frontier`]; callers must not use the
121/// pipeline-wide minimum when advancing a source-specific ordered operator.
122///
123/// # Errors
124/// Returns an error when a channel uses the reserved uninitialized watermark value.
125pub fn channel_progress_frontiers_by_source(
126    channels: &[ChannelProgress],
127) -> Result<std::collections::BTreeMap<&str, Option<i64>>, String> {
128    #[derive(Default)]
129    struct SourceProgress {
130        active_minimum: Option<i64>,
131        active_uninitialized: bool,
132        idle_maximum: Option<i64>,
133    }
134
135    let mut progress = std::collections::BTreeMap::<&str, SourceProgress>::new();
136    for channel in channels {
137        if channel.watermark == Some(i64::MIN) {
138            return Err("channel progress uses the reserved uninitialized watermark".into());
139        }
140        let source = progress.entry(channel.source_name.as_str()).or_default();
141        if channel.idle {
142            if let Some(watermark) = channel.watermark {
143                source.idle_maximum = Some(
144                    source
145                        .idle_maximum
146                        .map_or(watermark, |current| current.max(watermark)),
147                );
148            }
149        } else if let Some(watermark) = channel.watermark {
150            source.active_minimum = Some(
151                source
152                    .active_minimum
153                    .map_or(watermark, |current| current.min(watermark)),
154            );
155        } else {
156            source.active_uninitialized = true;
157        }
158    }
159
160    Ok(progress
161        .into_iter()
162        .map(|(source, progress)| {
163            let frontier = if progress.active_uninitialized {
164                None
165            } else {
166                progress.active_minimum.or(progress.idle_maximum)
167            };
168            (source, frontier)
169        })
170        .collect())
171}
laminar_core/checkpoint/mod.rs

laminar_core/checkpoint/
mod.rs
