laminar_core/checkpoint_decision/
sink_witness.rs

1//! Create-once sink-open witness transitions.
2
3use super::{
4    Bytes, CheckpointAttempt, CheckpointDecisionStore, CheckpointSinkOpenWitness,
5    CheckpointSinkOpenWitnessSlot, CheckpointSinkOpenWitnessSlotState, DecisionError,
6    DecisionStoreUpdateMode, ObjectStore, PipelineIdentity, PutMode, PutOptions, PutPayload,
7    UpdateVersion, VersionedCheckpointSinkOpenWitnessSlot, CHECKPOINT_SINK_OPEN_WITNESS_MAX_BYTES,
8    CHECKPOINT_SINK_OPEN_WITNESS_MAX_SINKS, CHECKPOINT_SINK_OPEN_WITNESS_SLOT_VERSION,
9    CHECKPOINT_SINK_OPEN_WITNESS_VERSION,
10};
11
12impl CheckpointDecisionStore {
13    fn validate_sink_open_witness_shape(
14        witness: &CheckpointSinkOpenWitness,
15    ) -> Result<(), DecisionError> {
16        if witness.version != CHECKPOINT_SINK_OPEN_WITNESS_VERSION {
17            return Err(DecisionError::Conflict(format!(
18                "sink-open witness version {} is unsupported (expected \
19                 {CHECKPOINT_SINK_OPEN_WITNESS_VERSION})",
20                witness.version
21            )));
22        }
23        let deployment = uuid::Uuid::parse_str(&witness.deployment_id).map_err(|error| {
24            DecisionError::Conflict(format!(
25                "sink-open witness has invalid deployment identity: {error}"
26            ))
27        })?;
28        if deployment.is_nil() || deployment.to_string() != witness.deployment_id {
29            return Err(DecisionError::Conflict(
30                "sink-open witness must use a canonical non-nil deployment identity".into(),
31            ));
32        }
33        if let Some(error) = witness.pipeline_identity.validation_error() {
34            return Err(DecisionError::Conflict(format!(
35                "sink-open witness has an invalid pipeline identity: {error}"
36            )));
37        }
38        if !witness.attempt.is_canonical() {
39            return Err(DecisionError::Conflict(
40                "sink-open witness must use one nonzero canonical checkpoint ID".into(),
41            ));
42        }
43        if witness.committable_sinks.is_empty()
44            || witness.committable_sinks.len() > CHECKPOINT_SINK_OPEN_WITNESS_MAX_SINKS
45        {
46            return Err(DecisionError::Conflict(format!(
47                "sink-open witness must name between 1 and \
48                 {CHECKPOINT_SINK_OPEN_WITNESS_MAX_SINKS} committable sinks"
49            )));
50        }
51        if witness
52            .committable_sinks
53            .iter()
54            .any(|name| name.is_empty() || name.trim() != name)
55        {
56            return Err(DecisionError::Conflict(
57                "sink-open witness contains a non-canonical sink name".into(),
58            ));
59        }
60        if witness
61            .committable_sinks
62            .windows(2)
63            .any(|pair| pair[0] >= pair[1])
64        {
65            return Err(DecisionError::Conflict(
66                "sink-open witness sink names must be strictly sorted and unique".into(),
67            ));
68        }
69        let create_token = uuid::Uuid::parse_str(&witness.create_token).map_err(|error| {
70            DecisionError::Conflict(format!(
71                "sink-open witness has invalid create token: {error}"
72            ))
73        })?;
74        if create_token.is_nil() || create_token.to_string() != witness.create_token {
75            return Err(DecisionError::Conflict(
76                "sink-open witness must use a canonical non-nil create token".into(),
77            ));
78        }
79        Ok(())
80    }
81
82    fn validate_sink_open_witness_slot_shape(
83        slot: &CheckpointSinkOpenWitnessSlot,
84    ) -> Result<(), DecisionError> {
85        if slot.version != CHECKPOINT_SINK_OPEN_WITNESS_SLOT_VERSION {
86            return Err(DecisionError::Conflict(format!(
87                "sink-open witness slot version {} is unsupported (expected \
88                 {CHECKPOINT_SINK_OPEN_WITNESS_SLOT_VERSION})",
89                slot.version
90            )));
91        }
92        Self::validate_sink_open_witness_shape(slot.witness())?;
93        if let CheckpointSinkOpenWitnessSlotState::Closed { close_token, .. } = &slot.state {
94            let token = uuid::Uuid::parse_str(close_token).map_err(|error| {
95                DecisionError::Conflict(format!(
96                    "sink-open witness slot has invalid close token: {error}"
97                ))
98            })?;
99            if token.is_nil() || token.to_string() != *close_token {
100                return Err(DecisionError::Conflict(
101                    "sink-open witness slot must use a canonical non-nil close token".into(),
102                ));
103            }
104        }
105        Ok(())
106    }
107
108    fn encode_sink_open_witness_slot(
109        slot: &CheckpointSinkOpenWitnessSlot,
110    ) -> Result<Bytes, DecisionError> {
111        Self::validate_sink_open_witness_slot_shape(slot)?;
112        Self::encode_control_record(
113            "sink-open witness slot",
114            slot,
115            CHECKPOINT_SINK_OPEN_WITNESS_MAX_BYTES,
116        )
117    }
118
119    async fn read_sink_open_witness_record(
120        &self,
121    ) -> Result<Option<VersionedCheckpointSinkOpenWitnessSlot>, DecisionError> {
122        let Some(result) = self
123            .get_control_record(
124                &Self::sink_open_witness_path(),
125                "sink-open witness",
126                CHECKPOINT_SINK_OPEN_WITNESS_MAX_BYTES,
127            )
128            .await?
129        else {
130            return Ok(None);
131        };
132        let update_version = UpdateVersion {
133            e_tag: result.meta.e_tag.clone(),
134            version: result.meta.version.clone(),
135        };
136        self.require_native_cas_token("sink-open witness slot", &update_version)?;
137        let bytes = Self::read_control_record_bytes(
138            result,
139            "sink-open witness",
140            CHECKPOINT_SINK_OPEN_WITNESS_MAX_BYTES,
141            None,
142        )
143        .await?;
144        let slot: CheckpointSinkOpenWitnessSlot = serde_json::from_slice(&bytes)
145            .map_err(|error| DecisionError::Conflict(format!("sink-open witness slot: {error}")))?;
146        Self::validate_sink_open_witness_slot_shape(&slot)?;
147        let canonical = serde_json::to_vec(&slot)
148            .map_err(|error| DecisionError::Conflict(error.to_string()))?;
149        if canonical.as_slice() != bytes.as_ref() {
150            return Err(DecisionError::Conflict(format!(
151                "sink-open witness slot for checkpoint {} does not use its canonical body",
152                slot.witness().attempt.checkpoint_id
153            )));
154        }
155        Ok(Some(VersionedCheckpointSinkOpenWitnessSlot {
156            slot,
157            update_version,
158        }))
159    }
160
161    fn validate_sink_open_witness_slot_deployment(
162        slot: &CheckpointSinkOpenWitnessSlot,
163        deployment_id: &str,
164    ) -> Result<(), DecisionError> {
165        if slot.witness().deployment_id == deployment_id {
166            return Ok(());
167        }
168        Err(DecisionError::Conflict(format!(
169            "sink-open witness belongs to deployment {}, current deployment is {deployment_id}",
170            slot.witness().deployment_id
171        )))
172    }
173
174    fn sink_open_witness_put_mode(&self, expected: Option<UpdateVersion>) -> PutMode {
175        match (self.update_mode, expected) {
176            (_, None) => PutMode::Create,
177            (DecisionStoreUpdateMode::NativeCas, Some(version)) => PutMode::Update(version),
178            (DecisionStoreUpdateMode::LocalSingleWriter, Some(_)) => PutMode::Overwrite,
179        }
180    }
181
182    async fn put_sink_open_witness_slot(
183        &self,
184        slot: &CheckpointSinkOpenWitnessSlot,
185        expected: Option<UpdateVersion>,
186    ) -> Result<(), object_store::Error> {
187        let payload = Self::encode_sink_open_witness_slot(slot).map_err(|error| {
188            object_store::Error::Generic {
189                store: "CheckpointDecisionStore",
190                source: Box::new(error),
191            }
192        })?;
193        self.store
194            .put_opts(
195                &Self::sink_open_witness_path(),
196                PutPayload::from(payload),
197                PutOptions {
198                    mode: self.sink_open_witness_put_mode(expected),
199                    ..PutOptions::default()
200                },
201            )
202            .await
203            .map(|_| ())
204    }
205
206    /// Read the singleton sink-open owner record.
207    ///
208    /// # Errors
209    /// Object-store I/O, malformed/non-canonical metadata, or foreign deployment state.
210    pub async fn sink_open_witness(
211        &self,
212    ) -> Result<Option<CheckpointSinkOpenWitness>, DecisionError> {
213        let deployment_id = self.load_or_create_deployment_id().await?;
214        let Some(versioned) = self.read_sink_open_witness_record().await? else {
215            return Ok(None);
216        };
217        Self::validate_sink_open_witness_slot_deployment(&versioned.slot, &deployment_id)?;
218        match versioned.slot.state {
219            CheckpointSinkOpenWitnessSlotState::Open { witness } => Ok(Some(witness)),
220            CheckpointSinkOpenWitnessSlotState::Closed { .. } => Ok(None),
221        }
222    }
223
224    /// Create the durable witness before invoking any checkpoint-committable sink begin call.
225    ///
226    /// `committable_sinks` must already be strictly sorted and unique so duplicate runtime names
227    /// cannot be silently collapsed into one recovery participant.
228    ///
229    /// # Errors
230    /// Object-store I/O, invalid input, or any malformed, foreign, or conflicting live witness.
231    pub async fn create_sink_open_witness(
232        &self,
233        pipeline_identity: PipelineIdentity,
234        participant_id: u64,
235        attempt: CheckpointAttempt,
236        committable_sinks: Vec<String>,
237    ) -> Result<CheckpointSinkOpenWitness, DecisionError> {
238        let candidate = CheckpointSinkOpenWitness {
239            version: CHECKPOINT_SINK_OPEN_WITNESS_VERSION,
240            deployment_id: self.load_or_create_deployment_id().await?,
241            pipeline_identity,
242            participant_id,
243            attempt,
244            committable_sinks,
245            create_token: uuid::Uuid::now_v7().to_string(),
246        };
247        Self::validate_sink_open_witness_shape(&candidate)?;
248        Self::encode_sink_open_witness_slot(&CheckpointSinkOpenWitnessSlot::open(
249            candidate.clone(),
250        ))?;
251        // An accepted open must always have enough room for its mandatory close tombstone.
252        Self::encode_sink_open_witness_slot(&CheckpointSinkOpenWitnessSlot::closed(
253            candidate.clone(),
254        ))?;
255
256        match self.update_mode {
257            DecisionStoreUpdateMode::NativeCas => {
258                self.create_sink_open_witness_inner(candidate).await
259            }
260            DecisionStoreUpdateMode::LocalSingleWriter => {
261                let local_lock = self.local_metadata_rmw_lock.as_ref().ok_or_else(|| {
262                    DecisionError::Conflict(
263                        "local decision store is missing its namespace write lock".to_owned(),
264                    )
265                })?;
266                let _guard = local_lock.lock().await;
267                self.create_sink_open_witness_inner(candidate).await
268            }
269        }
270    }
271
272    async fn create_sink_open_witness_inner(
273        &self,
274        candidate: CheckpointSinkOpenWitness,
275    ) -> Result<CheckpointSinkOpenWitness, DecisionError> {
276        let current = self.read_sink_open_witness_record().await?;
277        let (expected, prior_slot) = match current {
278            None => (None, None),
279            Some(versioned) => {
280                Self::validate_sink_open_witness_slot_deployment(
281                    &versioned.slot,
282                    &candidate.deployment_id,
283                )?;
284                match &versioned.slot.state {
285                    CheckpointSinkOpenWitnessSlotState::Open { witness } => {
286                        return Err(DecisionError::Conflict(format!(
287                            "sink-open witness create for checkpoint {} observed conflicting \
288                             checkpoint {}",
289                            candidate.attempt.checkpoint_id, witness.attempt.checkpoint_id
290                        )));
291                    }
292                    CheckpointSinkOpenWitnessSlotState::Closed { witness, .. }
293                        if candidate.attempt.checkpoint_id <= witness.attempt.checkpoint_id =>
294                    {
295                        return Err(DecisionError::Conflict(format!(
296                            "sink-open witness checkpoint {} does not advance closed checkpoint {}",
297                            candidate.attempt.checkpoint_id, witness.attempt.checkpoint_id
298                        )));
299                    }
300                    CheckpointSinkOpenWitnessSlotState::Closed { .. } => {}
301                }
302                (Some(versioned.update_version), Some(versioned.slot))
303            }
304        };
305        let candidate_slot = CheckpointSinkOpenWitnessSlot::open(candidate.clone());
306        let create_error = match self
307            .put_sink_open_witness_slot(&candidate_slot, expected)
308            .await
309        {
310            Ok(()) => return Ok(candidate),
311            Err(error) => error,
312        };
313
314        // Only this proposal's exact create token proves that an ambiguous open transition won.
315        if let Some(observed) = self.read_sink_open_witness_record().await? {
316            Self::validate_sink_open_witness_slot_deployment(
317                &observed.slot,
318                &candidate.deployment_id,
319            )?;
320            if observed.slot == candidate_slot {
321                return Ok(candidate);
322            }
323            let conditional_conflict = matches!(
324                &create_error,
325                object_store::Error::Precondition { .. }
326                    | object_store::Error::AlreadyExists { .. }
327                    | object_store::Error::NotFound { .. }
328            );
329            if !conditional_conflict
330                && prior_slot
331                    .as_ref()
332                    .is_some_and(|prior| prior == &observed.slot)
333            {
334                return Err(DecisionError::Io(create_error.to_string()));
335            }
336            return Err(DecisionError::Conflict(format!(
337                "sink-open witness create for checkpoint {} observed conflicting checkpoint {}",
338                candidate.attempt.checkpoint_id,
339                observed.slot.witness().attempt.checkpoint_id
340            )));
341        }
342
343        match create_error {
344            object_store::Error::Precondition { .. }
345            | object_store::Error::AlreadyExists { .. }
346            | object_store::Error::NotFound { .. } => Err(DecisionError::Conflict(format!(
347                "sink-open witness for checkpoint {} disappeared after create conflict",
348                candidate.attempt.checkpoint_id
349            ))),
350            error => Err(DecisionError::Io(error.to_string())),
351        }
352    }
353
354    /// Close exactly the supplied witness after its attempt is terminal or fully rolled back.
355    ///
356    /// Closure durably replaces the open state. The tombstone makes an old conditional write
357    /// harmless after a successor opens and gives ambiguous responses an exact state to reconcile.
358    ///
359    /// # Errors
360    /// Object-store I/O or a malformed, foreign, or different live witness.
361    pub async fn clear_sink_open_witness(
362        &self,
363        expected: &CheckpointSinkOpenWitness,
364    ) -> Result<(), DecisionError> {
365        Self::validate_sink_open_witness_shape(expected)?;
366        let deployment_id = self.load_or_create_deployment_id().await?;
367        if expected.deployment_id != deployment_id {
368            return Err(DecisionError::Conflict(format!(
369                "cannot clear sink-open witness from deployment {}; current deployment is \
370                 {deployment_id}",
371                expected.deployment_id
372            )));
373        }
374        match self.update_mode {
375            DecisionStoreUpdateMode::NativeCas => {
376                self.clear_sink_open_witness_inner(expected).await
377            }
378            DecisionStoreUpdateMode::LocalSingleWriter => {
379                let local_lock = self.local_metadata_rmw_lock.as_ref().ok_or_else(|| {
380                    DecisionError::Conflict(
381                        "local decision store is missing its namespace write lock".to_owned(),
382                    )
383                })?;
384                let _guard = local_lock.lock().await;
385                self.clear_sink_open_witness_inner(expected).await
386            }
387        }
388    }
389
390    async fn clear_sink_open_witness_inner(
391        &self,
392        expected: &CheckpointSinkOpenWitness,
393    ) -> Result<(), DecisionError> {
394        let Some(current) = self.read_sink_open_witness_record().await? else {
395            return Err(DecisionError::Conflict(format!(
396                "sink-open witness slot for checkpoint {} is missing",
397                expected.attempt.checkpoint_id
398            )));
399        };
400        Self::validate_sink_open_witness_slot_deployment(&current.slot, &expected.deployment_id)?;
401        match &current.slot.state {
402            CheckpointSinkOpenWitnessSlotState::Closed { witness, .. } if witness == expected => {
403                return Ok(());
404            }
405            CheckpointSinkOpenWitnessSlotState::Open { witness } if witness == expected => {}
406            _ => {
407                return Err(DecisionError::Conflict(format!(
408                    "cannot clear sink-open witness for checkpoint {}; current slot names \
409                     checkpoint {} with a different create identity",
410                    expected.attempt.checkpoint_id,
411                    current.slot.witness().attempt.checkpoint_id
412                )));
413            }
414        }
415
416        let closed = CheckpointSinkOpenWitnessSlot::closed(expected.clone());
417        let close_error = match self
418            .put_sink_open_witness_slot(&closed, Some(current.update_version))
419            .await
420        {
421            Ok(()) => return Ok(()),
422            Err(error) => error,
423        };
424        match self.read_sink_open_witness_record().await? {
425            Some(observed) if observed.slot == closed => Ok(()),
426            Some(observed) => {
427                Self::validate_sink_open_witness_slot_deployment(
428                    &observed.slot,
429                    &expected.deployment_id,
430                )?;
431                match &observed.slot.state {
432                    CheckpointSinkOpenWitnessSlotState::Closed { witness, .. }
433                        if witness == expected =>
434                    {
435                        // Another exact close is equivalent even though its token differs.
436                        Ok(())
437                    }
438                    CheckpointSinkOpenWitnessSlotState::Open { witness } if witness == expected => {
439                        Err(DecisionError::Io(close_error.to_string()))
440                    }
441                    _ => {
442                        // A different valid generation can only follow a successful close CAS.
443                        // The stale transition cannot touch it because its object version differs.
444                        Ok(())
445                    }
446                }
447            }
448            None => Err(DecisionError::Conflict(format!(
449                "sink-open witness slot disappeared while closing checkpoint {}: {close_error}",
450                expected.attempt.checkpoint_id
451            ))),
452        }
453    }
454}
laminar_core/checkpoint_decision/sink_witness.rs

laminar_core/checkpoint_decision/
sink_witness.rs
