laminar_core/cluster/control/leader_lease/
attempt_status.rs

1//! Audited checkpoint settlement and commit-authority observation.
2
3use std::sync::Arc;
4
5use crate::checkpoint::{
6    CheckpointAssignmentFence, CheckpointAttempt, CommittedCheckpointIndex, CommittedCheckpointRef,
7    LeaderProof,
8};
9use crate::checkpoint_decision::{CheckpointDecisionStore, CheckpointOutcome, DecisionError};
10
11use super::{AssignmentHandoffPin, ClusterCheckpointAuthorityError, LeaderLeaseStore};
12
13/// Greatest live cluster Commit, its committed checkpoint index, and the assignment handoff pin,
14/// all observed through one authority record.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct ClusterCommitHead {
17    /// Greatest live cluster Commit.
18    pub outcome: CheckpointOutcome,
19    /// Committed checkpoint index loaded through the Commit's own reference.
20    pub index: CommittedCheckpointIndex,
21    handoff_pin: Option<AssignmentHandoffPin>,
22}
23
24impl ClusterCommitHead {
25    /// Checkpoint pinned for `target` in the authority record that selected this Commit.
26    #[must_use]
27    pub fn handoff_checkpoint(
28        &self,
29        target: &CheckpointAssignmentFence,
30    ) -> Option<&CommittedCheckpointRef> {
31        self.handoff_pin
32            .as_ref()
33            .filter(|pin| pin.target == *target)
34            .map(|pin| &pin.checkpoint)
35    }
36}
37
38/// Durable settlement state for one exact cluster checkpoint attempt.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub enum ClusterAttemptStatus {
41    /// No terminal outcome exists and the admitting leader term can still publish one.
42    Pending,
43    /// An exact or newer terminal outcome durably settles the attempt.
44    Settled(Box<CheckpointOutcome>),
45    /// No terminal outcome exists, but the leader term that admitted the artifacts was superseded.
46    ///
47    /// Takeover and outcome publication serialize through the same authority sequence. Once this
48    /// state is observed, the old term cannot publish a Commit; coordinated recovery still owns
49    /// publication of the authoritative Abort and artifact cleanup.
50    CommitFenced,
51}
52
53fn settlement_from_outcomes(
54    attempt: CheckpointAttempt,
55    outcomes: &[CheckpointOutcome],
56) -> Option<CheckpointOutcome> {
57    if let Ok(index) = outcomes.binary_search_by_key(&attempt.epoch, |outcome| outcome.epoch) {
58        return Some(outcomes[index].clone());
59    }
60    outcomes
61        .last()
62        .filter(|highest| highest.checkpoint_id > attempt.checkpoint_id)
63        .cloned()
64}
65
66impl LeaderLeaseStore {
67    /// Return the exact immutable outcome for `attempt`, or the first audited terminal outcome
68    /// known to close that older checkpoint. Compacted continuity anchors are included in the
69    /// audit.
70    ///
71    /// # Errors
72    /// Returns an error for a noncanonical attempt identity or an unavailable or invalid durable
73    /// authority chain.
74    pub async fn cluster_attempt_settlement(
75        &self,
76        attempt: CheckpointAttempt,
77    ) -> Result<Option<CheckpointOutcome>, ClusterCheckpointAuthorityError> {
78        if !attempt.is_canonical() {
79            return Err(DecisionError::Conflict(
80                "cluster checkpoint settlement requires one nonzero canonical checkpoint ID".into(),
81            )
82            .into());
83        }
84        let outcomes = self.audited_cluster_outcomes().await?.1;
85        Ok(settlement_from_outcomes(attempt, &outcomes))
86    }
87
88    /// Greatest live cluster Commit together with its validated committed checkpoint index and the
89    /// assignment handoff pin of the same authority record.
90    ///
91    /// The index is loaded through the selected Commit's own reference. A second lookup by epoch
92    /// can miss a head that retention retired as soon as a newer Commit landed between the reads.
93    /// A target Commit clears the handoff pin in the write that moves the head, so a pin read from
94    /// a later record can describe a newer head than the one returned.
95    ///
96    /// # Errors
97    /// Returns an error for an unavailable or invalid authority chain, or for a committed index
98    /// that is missing or mismatched while its Commit is still the head.
99    pub async fn highest_cluster_commit_with_checkpoint(
100        &self,
101    ) -> Result<Option<ClusterCommitHead>, ClusterCheckpointAuthorityError> {
102        const LOOKUP_RETRIES: usize = 3;
103        let decisions = CheckpointDecisionStore::new(Arc::clone(&self.store));
104        for _ in 0..LOOKUP_RETRIES {
105            let (record, outcomes) = self.audited_cluster_outcomes().await?;
106            let Some(outcome) =
107                Self::cluster_outcome_inventory_from_audit(record.as_ref(), &outcomes)
108                    .outcomes
109                    .into_iter()
110                    .rev()
111                    .find(CheckpointOutcome::is_commit)
112            else {
113                return Ok(None);
114            };
115            let error = match decisions
116                .validate_committed_checkpoint_for_outcome(&outcome)
117                .await
118            {
119                Ok(index) => {
120                    return Ok(Some(ClusterCommitHead {
121                        outcome,
122                        index,
123                        handoff_pin: record.and_then(|record| record.assignment_handoff_pin),
124                    }))
125                }
126                Err(error) => error,
127            };
128            // Cleanup deletes an index only after a newer Commit has retired it as head.
129            let replaced = self
130                .highest_cluster_committed_outcome()
131                .await?
132                .is_some_and(|current| current.epoch > outcome.epoch);
133            if !replaced {
134                return Err(error.into());
135            }
136        }
137        Err(DecisionError::InventoryChanged(
138            "cluster Commit head lookup exhausted stability retries".into(),
139        )
140        .into())
141    }
142
143    /// Audit the settlement and commit authority of one admitted cluster checkpoint attempt.
144    ///
145    /// Unlike [`Self::cluster_attempt_settlement`], this distinguishes a live pending attempt from
146    /// an attempt whose admitting leader term has been durably superseded. The distinction is
147    /// emitted only when the same audited authority head both retains the exact artifact inventory
148    /// and proves that its admitting leader proof no longer owns the lease.
149    ///
150    /// # Errors
151    /// Returns an error for malformed identities, an unavailable or invalid authority chain, or an
152    /// unsettled attempt that does not exactly match the retained artifact inventory.
153    pub async fn cluster_attempt_status(
154        &self,
155        attempt: CheckpointAttempt,
156        assignment_fence: &CheckpointAssignmentFence,
157        leader_proof: &LeaderProof,
158    ) -> Result<ClusterAttemptStatus, ClusterCheckpointAuthorityError> {
159        if !attempt.is_canonical() {
160            return Err(DecisionError::Conflict(
161                "cluster checkpoint status requires one nonzero canonical checkpoint ID".into(),
162            )
163            .into());
164        }
165        if !assignment_fence.is_canonical() {
166            return Err(DecisionError::Conflict(
167                "cluster checkpoint status requires a canonical assignment fence".into(),
168            )
169            .into());
170        }
171        if !leader_proof.is_canonical() {
172            return Err(ClusterCheckpointAuthorityError::Fenced);
173        }
174        if assignment_fence.participant_incarnation(leader_proof.owner.node_id)
175            != Some(leader_proof.owner.boot_id)
176        {
177            return Err(DecisionError::Conflict(
178                "cluster checkpoint status leader is outside the assignment fence".into(),
179            )
180            .into());
181        }
182
183        let (head, outcomes) = self.audited_cluster_outcomes().await?;
184        if let Some(settlement) = settlement_from_outcomes(attempt, &outcomes) {
185            return Ok(ClusterAttemptStatus::Settled(Box::new(settlement)));
186        }
187        let head = head.ok_or(ClusterCheckpointAuthorityError::Fenced)?;
188        let active = head.active_checkpoint_artifacts.as_ref().ok_or_else(|| {
189            DecisionError::Conflict(format!(
190                "unsettled cluster checkpoint {} has no active artifact inventory",
191                attempt.checkpoint_id
192            ))
193        })?;
194        if active.attempt != attempt
195            || active.assignment_fence.as_ref() != Some(assignment_fence)
196            || head.active_checkpoint_artifact_leader_proof.as_ref() != Some(leader_proof)
197        {
198            return Err(DecisionError::Conflict(format!(
199                "unsettled cluster checkpoint {} does not match its admitted artifact authority",
200                attempt.checkpoint_id
201            ))
202            .into());
203        }
204        if head.lease.matches_proof(leader_proof) {
205            Ok(ClusterAttemptStatus::Pending)
206        } else {
207            Ok(ClusterAttemptStatus::CommitFenced)
208        }
209    }
210}
laminar_core/cluster/control/leader_lease/attempt_status.rs

laminar_core/cluster/control/leader_lease/
attempt_status.rs
