laminar_connectors/connector/
coordinated_commit.rs

1//! Bounded, fenced coordinated-commit batches and cursor validation.
2
3use async_trait::async_trait;
4use sha2::{Digest, Sha256};
5
6use crate::error::ConnectorError;
7
8/// Fixed control-plane bound for one connector's coordinated-commit payload.
9///
10/// Connectors must keep prepared metadata at or below this limit before
11/// returning it to the checkpoint runtime. Bulk records belong in the sink's
12/// data plane, referenced by the bounded payload.
13pub const MAX_COORDINATED_COMMIT_PAYLOAD_BYTES: usize = 16 * 1024 * 1024;
14
15/// Fixed aggregate control-plane bound for one designated commit call.
16pub const MAX_COORDINATED_COMMIT_BATCH_BYTES: usize = 64 * 1024 * 1024;
17
18/// Fixed participant-marker bound for one designated commit call.
19pub const MAX_COORDINATED_COMMIT_BATCH_ENTRIES: usize = 4_096;
20
21/// Stable external commit namespace for one deployment incarnation of a logical pipeline sink.
22///
23/// The configured external target already scopes its metadata. The create-once deployment id
24/// prevents checkpoint-store resets or two
25/// identically configured deployments from sharing a cursor. Pipeline identity plus sink id then
26/// binds that deployment to one recovery-compatible logical writer.
27#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
28pub struct CoordinatedCommitNamespace {
29    /// Canonical logical-pipeline identity used by checkpoint recovery.
30    pub pipeline_identity: laminar_core::checkpoint::checkpoint_manifest::PipelineIdentity,
31    /// Create-once UUID stored with checkpoint decisions and shared by every cluster member.
32    pub deployment_id: String,
33    /// Stable sink registration id within the pipeline.
34    pub sink_id: String,
35}
36
37impl CoordinatedCommitNamespace {
38    /// Construct and validate a namespace before any external metadata lookup.
39    ///
40    /// # Errors
41    /// Returns a configuration error for a non-canonical pipeline identity or
42    /// empty sink id.
43    pub fn try_new(
44        pipeline_identity: laminar_core::checkpoint::checkpoint_manifest::PipelineIdentity,
45        deployment_id: impl Into<String>,
46        sink_id: impl Into<String>,
47    ) -> Result<Self, ConnectorError> {
48        let namespace = Self {
49            pipeline_identity,
50            deployment_id: deployment_id.into(),
51            sink_id: sink_id.into(),
52        };
53        namespace.validate()?;
54        Ok(namespace)
55    }
56
57    /// Validate a namespace received through a public or serialized boundary.
58    ///
59    /// # Errors
60    /// Returns a configuration error when any identity component is not
61    /// canonical.
62    pub fn validate(&self) -> Result<(), ConnectorError> {
63        if !self.pipeline_identity.is_canonical() {
64            return Err(ConnectorError::ConfigurationError(
65                "coordinated commit requires the current canonical pipeline identity".into(),
66            ));
67        }
68        if self.sink_id.is_empty() {
69            return Err(ConnectorError::ConfigurationError(
70                "coordinated commit sink id cannot be empty".into(),
71            ));
72        }
73        let parsed_deployment = uuid::Uuid::parse_str(&self.deployment_id).map_err(|error| {
74            ConnectorError::ConfigurationError(format!(
75                "coordinated commit deployment id is not a UUID: {error}"
76            ))
77        })?;
78        if parsed_deployment.is_nil() || parsed_deployment.to_string() != self.deployment_id {
79            return Err(ConnectorError::ConfigurationError(
80                "coordinated commit deployment id must be a canonical non-nil UUID".into(),
81            ));
82        }
83        Ok(())
84    }
85
86    /// Bounded, filesystem/catalog-safe key for external transaction metadata.
87    #[must_use]
88    pub fn external_key(&self) -> String {
89        let mut digest = Sha256::new();
90        digest.update(self.pipeline_identity.canonical_version.to_be_bytes());
91        digest.update(self.pipeline_identity.sha256.as_bytes());
92        digest.update([0]);
93        digest.update(self.deployment_id.as_bytes());
94        digest.update([0]);
95        digest.update(self.sink_id.as_bytes());
96        let digest = digest.finalize();
97        format!("ldb-c3-{digest:x}")
98    }
99}
100
101/// Exact external commit position and the authority that published it.
102#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
103pub struct CoordinatedCommitCursor {
104    /// Highest globally unique checkpoint id atomically reflected by the sink.
105    pub checkpoint_id: u64,
106    /// Monotonic authority token that fenced earlier designated committers.
107    pub fencing_token: u64,
108}
109
110/// One participant's validated prepared marker for one exact attempt.
111#[derive(Debug, Clone, PartialEq, Eq)]
112pub struct CoordinatedCommitPayload {
113    /// Exact checkpoint attempt that admitted this marker.
114    pub attempt: laminar_core::checkpoint::CheckpointAttempt,
115    /// Stable nonzero runtime participant ID.
116    pub participant_id: u64,
117    /// Connector-specific committable, or `None` for an explicitly empty cut.
118    pub payload: Option<Vec<u8>>,
119}
120
121/// Whether an aborted participant reached durable phase-one preparation.
122#[derive(Debug, Clone, PartialEq, Eq)]
123pub enum CoordinatedAbortDescriptor {
124    /// The participant may have written files but did not durably prepare a descriptor.
125    Open,
126    /// The participant durably prepared its connector descriptor. `None` is an explicit empty cut.
127    Prepared(Option<Vec<u8>>),
128}
129
130/// One participant's durable cleanup evidence for an aborted attempt.
131#[derive(Debug, Clone, PartialEq, Eq)]
132pub struct CoordinatedAbortEntry {
133    /// Exact checkpoint attempt that admitted the participant.
134    pub attempt: laminar_core::checkpoint::CheckpointAttempt,
135    /// Stable nonzero runtime participant ID.
136    pub participant_id: u64,
137    /// Phase-one descriptor state retained by checkpoint persistence.
138    pub descriptor: CoordinatedAbortDescriptor,
139    /// Connector cleanup evidence persisted before `begin_epoch`, if the connector supplied it.
140    pub artifact_intent: Option<Vec<u8>>,
141}
142
143/// Exact batch submitted to a designated external-sink committer.
144#[derive(Debug, Clone, PartialEq, Eq)]
145pub struct CoordinatedCommitBatch {
146    /// External cursor namespace.
147    pub namespace: CoordinatedCommitNamespace,
148    /// Exact external cursor that must precede this batch. The zero cursor names
149    /// an empty target. A different authority at the predecessor checkpoint is
150    /// a conflicting history and must fail closed.
151    pub expected_predecessor: CoordinatedCommitCursor,
152    /// Non-zero authority token that the external commit must persist atomically.
153    pub fencing_token: u64,
154    /// Highest exact attempt atomically covered by this commit.
155    pub target: laminar_core::checkpoint::CheckpointAttempt,
156    /// Every prepared participant marker through `target`, including empty ones.
157    pub entries: Vec<CoordinatedCommitPayload>,
158}
159
160/// Exact prepared descriptors released only after an authoritative Abort.
161#[derive(Debug, Clone, PartialEq, Eq)]
162pub struct CoordinatedAbortBatch {
163    /// External cursor namespace that owned the aborted attempt.
164    pub namespace: CoordinatedCommitNamespace,
165    /// Current non-zero cleanup authority token.
166    pub fencing_token: u64,
167    /// Exact aborted checkpoint attempt.
168    pub target: laminar_core::checkpoint::CheckpointAttempt,
169    /// Every admitted participant's durable cleanup evidence for the aborted attempt.
170    pub entries: Vec<CoordinatedAbortEntry>,
171}
172
173/// Runtime-owned deadline for one designated external publication.
174///
175/// The deadline is created before the command enters the sink actor, so a
176/// connector sees the actual budget left after queueing rather than a second,
177/// connector-local timeout window.
178#[derive(Debug, Clone, Copy, PartialEq, Eq)]
179pub struct CoordinatedCommitContext {
180    deadline: tokio::time::Instant,
181}
182
183impl CoordinatedCommitContext {
184    /// Create a context from the sink actor's absolute end-to-end deadline.
185    #[must_use]
186    pub const fn new(deadline: tokio::time::Instant) -> Self {
187        Self { deadline }
188    }
189
190    /// Absolute monotonic publication deadline.
191    #[must_use]
192    pub const fn deadline(self) -> tokio::time::Instant {
193        self.deadline
194    }
195
196    /// Budget still available at the point the connector starts publication.
197    #[must_use]
198    pub fn remaining(self) -> std::time::Duration {
199        self.deadline
200            .saturating_duration_since(tokio::time::Instant::now())
201    }
202}
203
204impl CoordinatedCommitBatch {
205    /// Collision-resistant identity for one exact ordered publication cut.
206    /// Every variable-length field is length framed so distinct batches cannot
207    /// share an input byte stream before hashing.
208    #[must_use]
209    pub fn exact_fingerprint(&self) -> [u8; 32] {
210        fn update_length(hasher: &mut Sha256, length: usize) {
211            let source = length.to_be_bytes();
212            let mut encoded = [0_u8; 16];
213            let start = encoded.len() - source.len();
214            encoded[start..].copy_from_slice(&source);
215            hasher.update(encoded);
216        }
217
218        fn update_framed(hasher: &mut Sha256, bytes: &[u8]) {
219            update_length(hasher, bytes.len());
220            hasher.update(bytes);
221        }
222
223        let mut hasher = Sha256::new();
224        update_framed(&mut hasher, b"laminardb/coordinated-commit-batch/v1");
225        update_framed(&mut hasher, self.namespace.external_key().as_bytes());
226        hasher.update(self.expected_predecessor.checkpoint_id.to_be_bytes());
227        hasher.update(self.expected_predecessor.fencing_token.to_be_bytes());
228        hasher.update(self.fencing_token.to_be_bytes());
229        hasher.update(self.target.epoch.to_be_bytes());
230        hasher.update(self.target.checkpoint_id.to_be_bytes());
231        update_length(&mut hasher, self.entries.len());
232        for entry in &self.entries {
233            hasher.update(entry.attempt.epoch.to_be_bytes());
234            hasher.update(entry.attempt.checkpoint_id.to_be_bytes());
235            hasher.update(entry.participant_id.to_be_bytes());
236            match &entry.payload {
237                Some(payload) => {
238                    hasher.update([1]);
239                    update_framed(&mut hasher, payload);
240                }
241                None => hasher.update([0]),
242            }
243        }
244        hasher.finalize().into()
245    }
246
247    /// Validate canonical attempt/participant order and all fixed control-plane bounds.
248    /// This check is independent of external state and must run before connector I/O.
249    ///
250    /// # Errors
251    /// Returns a diagnostic when the batch is malformed or exceeds a fixed bound.
252    pub fn validate_shape(&self) -> Result<(), String> {
253        validate_prepared_entries(&self.namespace, self.target, &self.entries)?;
254        if self.expected_predecessor.checkpoint_id >= self.target.checkpoint_id {
255            return Err(format!(
256                "invalid coordinated batch predecessor {} for target {}",
257                self.expected_predecessor.checkpoint_id, self.target.checkpoint_id
258            ));
259        }
260        if (self.expected_predecessor.checkpoint_id == 0)
261            != (self.expected_predecessor.fencing_token == 0)
262        {
263            return Err(
264                "coordinated batch predecessor must be either an exact non-zero cursor or the zero cursor"
265                    .into(),
266            );
267        }
268        if self.fencing_token == 0 {
269            return Err("coordinated batch fencing token must be non-zero".into());
270        }
271        for entry in &self.entries {
272            if entry.attempt.checkpoint_id <= self.expected_predecessor.checkpoint_id
273                || entry.attempt.checkpoint_id > self.target.checkpoint_id
274            {
275                return Err(
276                    "coordinated batch entries do not cover the predecessor-to-target interval"
277                        .into(),
278                );
279            }
280        }
281        Ok(())
282    }
283
284    /// Validate a cursor freshly read from the external target against this
285    /// exact batch. Advancing overlap is safe only at an attempt named by the
286    /// batch; rollback or an unproven gap would skip output.
287    ///
288    /// # Errors
289    ///
290    /// Returns a diagnostic when the batch is malformed, the observed cursor
291    /// proves rollback, or an overlap cannot be tied to an exact batch entry.
292    pub fn validate_observed_cursor(
293        &self,
294        observed: Option<CoordinatedCommitCursor>,
295    ) -> Result<(), String> {
296        self.validate_shape()?;
297        let Some(observed) = observed else {
298            return if self.expected_predecessor.checkpoint_id == 0 {
299                Ok(())
300            } else {
301                Err(format!(
302                    "external cursor is absent below expected predecessor {}",
303                    self.expected_predecessor.checkpoint_id
304                ))
305            };
306        };
307        if observed.fencing_token == 0 {
308            return Err("external cursor contains a zero fencing token".into());
309        }
310        if observed.fencing_token > self.fencing_token {
311            return Err(format!(
312                "external fencing token {} is newer than designated committer token {}",
313                observed.fencing_token, self.fencing_token
314            ));
315        }
316        if observed.checkpoint_id >= self.target.checkpoint_id
317            && observed.fencing_token != self.fencing_token
318        {
319            return Err(format!(
320                "external cursor at or above target {} has fencing token {}, expected {}",
321                self.target.checkpoint_id, observed.fencing_token, self.fencing_token
322            ));
323        }
324        if observed.checkpoint_id < self.expected_predecessor.checkpoint_id {
325            return Err(format!(
326                "external cursor rolled back from expected predecessor {} to {}",
327                self.expected_predecessor.checkpoint_id, observed.checkpoint_id
328            ));
329        }
330        if observed.checkpoint_id == self.expected_predecessor.checkpoint_id
331            && observed != self.expected_predecessor
332        {
333            return Err(format!(
334                "external cursor checkpoint {} has fencing token {}, expected predecessor token {}",
335                observed.checkpoint_id,
336                observed.fencing_token,
337                self.expected_predecessor.fencing_token
338            ));
339        }
340        if observed.checkpoint_id > self.expected_predecessor.checkpoint_id
341            && observed.fencing_token < self.expected_predecessor.fencing_token
342        {
343            return Err(format!(
344                "external cursor advanced past predecessor {} while fencing token regressed from {} to {}",
345                self.expected_predecessor.checkpoint_id,
346                self.expected_predecessor.fencing_token,
347                observed.fencing_token
348            ));
349        }
350        if observed.checkpoint_id < self.target.checkpoint_id
351            && observed.checkpoint_id != self.expected_predecessor.checkpoint_id
352            && !self
353                .entries
354                .iter()
355                .any(|entry| entry.attempt.checkpoint_id == observed.checkpoint_id)
356        {
357            return Err(format!(
358                "external cursor {} is not an exact attempt in batch {}..={}",
359                observed.checkpoint_id,
360                self.expected_predecessor.checkpoint_id,
361                self.target.checkpoint_id
362            ));
363        }
364        Ok(())
365    }
366}
367
368impl CoordinatedAbortBatch {
369    /// Validate exact-attempt membership and fixed control-plane bounds.
370    ///
371    /// # Errors
372    /// Returns a diagnostic for a malformed authority, participant, or payload.
373    pub fn validate_shape(&self) -> Result<(), String> {
374        if self.fencing_token == 0 {
375            return Err("coordinated abort fencing token must be non-zero".into());
376        }
377        self.namespace
378            .validate()
379            .map_err(|error| error.to_string())?;
380        validate_abort_entries(self.target, &self.entries)
381    }
382}
383
384fn validate_abort_entries(
385    target: laminar_core::checkpoint::CheckpointAttempt,
386    entries: &[CoordinatedAbortEntry],
387) -> Result<(), String> {
388    if !target.is_canonical() {
389        return Err("coordinated abort target must use one nonzero canonical checkpoint ID".into());
390    }
391    if entries.is_empty() || entries.len() > MAX_COORDINATED_COMMIT_BATCH_ENTRIES {
392        return Err(format!(
393            "coordinated abort entry count must be in 1..={MAX_COORDINATED_COMMIT_BATCH_ENTRIES}"
394        ));
395    }
396    let mut previous_participant = None;
397    let mut total_payload_bytes = 0_usize;
398    for entry in entries {
399        if entry.participant_id == 0 || entry.attempt != target {
400            return Err(
401                "coordinated abort entries require nonzero participants at the exact target attempt"
402                    .into(),
403            );
404        }
405        if previous_participant.is_some_and(|previous| previous >= entry.participant_id) {
406            return Err(
407                "coordinated abort participants must be strictly ordered and unique".into(),
408            );
409        }
410        previous_participant = Some(entry.participant_id);
411        let descriptor = match &entry.descriptor {
412            CoordinatedAbortDescriptor::Open | CoordinatedAbortDescriptor::Prepared(None) => None,
413            CoordinatedAbortDescriptor::Prepared(Some(payload)) => Some(payload),
414        };
415        for payload in descriptor.into_iter().chain(entry.artifact_intent.as_ref()) {
416            if payload.len() > MAX_COORDINATED_COMMIT_PAYLOAD_BYTES {
417                return Err(format!(
418                    "coordinated abort payload exceeds the fixed {MAX_COORDINATED_COMMIT_PAYLOAD_BYTES} byte limit"
419                ));
420            }
421            total_payload_bytes = total_payload_bytes
422                .checked_add(payload.len())
423                .ok_or_else(|| "coordinated abort payload byte count overflow".to_owned())?;
424            if total_payload_bytes > MAX_COORDINATED_COMMIT_BATCH_BYTES {
425                return Err(format!(
426                    "coordinated abort payloads exceed the fixed {MAX_COORDINATED_COMMIT_BATCH_BYTES} byte limit"
427                ));
428            }
429        }
430    }
431    Ok(())
432}
433
434fn validate_prepared_entries(
435    namespace: &CoordinatedCommitNamespace,
436    target: laminar_core::checkpoint::CheckpointAttempt,
437    entries: &[CoordinatedCommitPayload],
438) -> Result<(), String> {
439    use laminar_core::checkpoint::CheckpointAttemptRelation;
440
441    namespace.validate().map_err(|error| error.to_string())?;
442    if !target.is_canonical() {
443        return Err("coordinated batch target must use one nonzero canonical checkpoint ID".into());
444    }
445    if entries.is_empty() || entries.len() > MAX_COORDINATED_COMMIT_BATCH_ENTRIES {
446        return Err(format!(
447            "coordinated batch entry count must be in 1..={MAX_COORDINATED_COMMIT_BATCH_ENTRIES}"
448        ));
449    }
450    if let Some(entry) = entries
451        .iter()
452        .find(|entry| entry.participant_id == 0 || !entry.attempt.is_canonical())
453    {
454        return Err(format!(
455            "coordinated batch entry must use a nonzero participant and canonical checkpoint ID; got participant {}",
456            entry.participant_id
457        ));
458    }
459
460    let mut total_payload_bytes = 0usize;
461    let mut previous: Option<&CoordinatedCommitPayload> = None;
462    for entry in entries {
463        if let Some(payload) = &entry.payload {
464            if payload.len() > MAX_COORDINATED_COMMIT_PAYLOAD_BYTES {
465                return Err(format!(
466                    "coordinated participant payload exceeds the fixed {MAX_COORDINATED_COMMIT_PAYLOAD_BYTES} byte limit"
467                ));
468            }
469            total_payload_bytes = total_payload_bytes
470                .checked_add(payload.len())
471                .ok_or_else(|| "coordinated batch payload byte count overflow".to_owned())?;
472            if total_payload_bytes > MAX_COORDINATED_COMMIT_BATCH_BYTES {
473                return Err(format!(
474                    "coordinated batch payloads exceed the fixed {MAX_COORDINATED_COMMIT_BATCH_BYTES} byte limit"
475                ));
476            }
477        }
478        if let Some(previous) = previous {
479            match entry.attempt.relation_to(previous.attempt) {
480                CheckpointAttemptRelation::Exact
481                    if entry.participant_id > previous.participant_id => {}
482                CheckpointAttemptRelation::Newer => {}
483                CheckpointAttemptRelation::Exact => {
484                    return Err(
485                        "coordinated batch contains a duplicate or out-of-order attempt/participant key"
486                            .into(),
487                    );
488                }
489                CheckpointAttemptRelation::Older | CheckpointAttemptRelation::Conflict => {
490                    return Err(
491                        "coordinated batch attempts are not in coherent epoch/checkpoint order"
492                            .into(),
493                    );
494                }
495            }
496        }
497        previous = Some(entry);
498    }
499    if previous.map(|entry| entry.attempt) != Some(target) {
500        return Err("coordinated batch target is not its final exact attempt".into());
501    }
502    Ok(())
503}
504
505/// Recovery-only cleanup capability detached from a participant writer.
506#[async_trait]
507pub trait CoordinatedAbortCleaner: Send + Sync {
508    /// Release exact durable participant artifacts after an authoritative Abort.
509    ///
510    /// The cleaner must remain usable after the participant writer is closed. Implementations
511    /// must be idempotent and retain any artifact whose publication status is ambiguous.
512    async fn cleanup_aborted(
513        &self,
514        batch: CoordinatedAbortBatch,
515        context: CoordinatedCommitContext,
516    ) -> Result<(), ConnectorError>;
517}
518
519/// Leader-side commit for checkpoint-committable sinks.
520///
521/// The designated committer aggregates every writer's `pre_commit` descriptor
522/// for an epoch into one external commit. Must be idempotent: re-running with
523/// the same inputs after a leader failover is a no-op once the target already
524/// reflects the epoch.
525#[async_trait]
526pub trait CoordinatedCommitter: Send + Sync {
527    /// Atomically commit the validated participant markers and advance the
528    /// namespaced external cursor to the batch's exact target. Empty markers
529    /// still advance the cursor.
530    async fn commit_aggregated(
531        &self,
532        batch: CoordinatedCommitBatch,
533        context: CoordinatedCommitContext,
534    ) -> Result<(), ConnectorError>;
535
536    /// Highest checkpoint and fencing authority committed in `namespace`.
537    /// A metadata read error must be returned, never converted to an absent
538    /// cursor, because that could duplicate a previously committed batch.
539    /// Implementations must validate the namespace before external I/O.
540    async fn committed_cursor(
541        &self,
542        namespace: &CoordinatedCommitNamespace,
543    ) -> Result<Option<CoordinatedCommitCursor>, ConnectorError>;
544}
laminar_connectors/connector/coordinated_commit.rs

laminar_connectors/connector/
coordinated_commit.rs
