laminar_core/cluster/control/leader_lease/
attempt_status.rs

1//! Audited checkpoint settlement and commit-authority observation.
2
3use crate::checkpoint::{CheckpointAssignmentFence, CheckpointAttempt, LeaderProof};
4use crate::checkpoint_decision::{CheckpointOutcome, DecisionError};
5
6use super::{ClusterCheckpointAuthorityError, LeaderLeaseStore};
7
8/// Durable settlement state for one exact cluster checkpoint attempt.
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub enum ClusterAttemptStatus {
11    /// No terminal outcome exists and the admitting leader term can still publish one.
12    Pending,
13    /// An exact or newer terminal outcome durably settles the attempt.
14    Settled(Box<CheckpointOutcome>),
15    /// No terminal outcome exists, but the leader term that admitted the artifacts was superseded.
16    ///
17    /// Takeover and outcome publication serialize through the same authority sequence. Once this
18    /// state is observed, the old term cannot publish a Commit; coordinated recovery still owns
19    /// publication of the authoritative Abort and artifact cleanup.
20    CommitFenced,
21}
22
23fn settlement_from_outcomes(
24    attempt: CheckpointAttempt,
25    outcomes: &[CheckpointOutcome],
26) -> Option<CheckpointOutcome> {
27    if let Ok(index) = outcomes.binary_search_by_key(&attempt.epoch, |outcome| outcome.epoch) {
28        return Some(outcomes[index].clone());
29    }
30    outcomes
31        .last()
32        .filter(|highest| highest.checkpoint_id > attempt.checkpoint_id)
33        .cloned()
34}
35
36impl LeaderLeaseStore {
37    /// Return the exact immutable outcome for `attempt`, or the first audited terminal outcome
38    /// known to close that older checkpoint. Compacted continuity anchors are included in the
39    /// audit.
40    ///
41    /// # Errors
42    /// Returns an error for a noncanonical attempt identity or an unavailable or invalid durable
43    /// authority chain.
44    pub async fn cluster_attempt_settlement(
45        &self,
46        attempt: CheckpointAttempt,
47    ) -> Result<Option<CheckpointOutcome>, ClusterCheckpointAuthorityError> {
48        if !attempt.is_canonical() {
49            return Err(DecisionError::Conflict(
50                "cluster checkpoint settlement requires one nonzero canonical checkpoint ID".into(),
51            )
52            .into());
53        }
54        let outcomes = self.audited_cluster_outcomes().await?.1;
55        Ok(settlement_from_outcomes(attempt, &outcomes))
56    }
57
58    /// Audit the settlement and commit authority of one admitted cluster checkpoint attempt.
59    ///
60    /// Unlike [`Self::cluster_attempt_settlement`], this distinguishes a live pending attempt from
61    /// an attempt whose admitting leader term has been durably superseded. The distinction is
62    /// emitted only when the same audited authority head both retains the exact artifact inventory
63    /// and proves that its admitting leader proof no longer owns the lease.
64    ///
65    /// # Errors
66    /// Returns an error for malformed identities, an unavailable or invalid authority chain, or an
67    /// unsettled attempt that does not exactly match the retained artifact inventory.
68    pub async fn cluster_attempt_status(
69        &self,
70        attempt: CheckpointAttempt,
71        assignment_fence: &CheckpointAssignmentFence,
72        leader_proof: &LeaderProof,
73    ) -> Result<ClusterAttemptStatus, ClusterCheckpointAuthorityError> {
74        if !attempt.is_canonical() {
75            return Err(DecisionError::Conflict(
76                "cluster checkpoint status requires one nonzero canonical checkpoint ID".into(),
77            )
78            .into());
79        }
80        if !assignment_fence.is_canonical() {
81            return Err(DecisionError::Conflict(
82                "cluster checkpoint status requires a canonical assignment fence".into(),
83            )
84            .into());
85        }
86        if !leader_proof.is_canonical() {
87            return Err(ClusterCheckpointAuthorityError::Fenced);
88        }
89        if assignment_fence.participant_incarnation(leader_proof.owner.node_id)
90            != Some(leader_proof.owner.boot_id)
91        {
92            return Err(DecisionError::Conflict(
93                "cluster checkpoint status leader is outside the assignment fence".into(),
94            )
95            .into());
96        }
97
98        let (head, outcomes) = self.audited_cluster_outcomes().await?;
99        if let Some(settlement) = settlement_from_outcomes(attempt, &outcomes) {
100            return Ok(ClusterAttemptStatus::Settled(Box::new(settlement)));
101        }
102        let head = head.ok_or(ClusterCheckpointAuthorityError::Fenced)?;
103        let active = head.active_checkpoint_artifacts.as_ref().ok_or_else(|| {
104            DecisionError::Conflict(format!(
105                "unsettled cluster checkpoint {} has no active artifact inventory",
106                attempt.checkpoint_id
107            ))
108        })?;
109        if active.attempt != attempt
110            || active.assignment_fence.as_ref() != Some(assignment_fence)
111            || head.active_checkpoint_artifact_leader_proof.as_ref() != Some(leader_proof)
112        {
113            return Err(DecisionError::Conflict(format!(
114                "unsettled cluster checkpoint {} does not match its admitted artifact authority",
115                attempt.checkpoint_id
116            ))
117            .into());
118        }
119        if head.lease.matches_proof(leader_proof) {
120            Ok(ClusterAttemptStatus::Pending)
121        } else {
122            Ok(ClusterAttemptStatus::CommitFenced)
123        }
124    }
125}
laminar_core/cluster/control/leader_lease/attempt_status.rs

laminar_core/cluster/control/leader_lease/
attempt_status.rs
