laminar_core/cluster/control/topology/
commit.rs

1//! Irreversible catalog decision, separate from installation and intake release.
2
3use serde::{Deserialize, Serialize};
4
5use super::{
6    TopologyAdmissionPhase, TopologyAdmissionStatus, TopologyError, TopologyOperationId,
7    TopologyVersion,
8};
9use crate::cluster::control::CatalogManifestRef;
10
11/// Every frozen participant must support Commit and explicit migration-root reconstruction.
12pub const TOPOLOGY_COMMIT_PROTOCOL_VERSION: u16 = 4;
13
14/// The exact shared append that changed the committed catalog. The containing operation's
15/// immutable plan, root, cut, descriptor and complete preparation roster bind this decision.
16/// This does not prove installation or authorize output, sink succession or intake release.
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(deny_unknown_fields)]
19pub struct TopologyCommit {
20    /// Required Commit/reconstruction protocol.
21    pub protocol_version: u16,
22    /// Original request identity, never replaced on an uncertain write.
23    pub operation_id: TopologyOperationId,
24    /// Previously committed logical version.
25    pub parent_version: TopologyVersion,
26    /// Exact previously committed inventory.
27    pub parent_manifest: CatalogManifestRef,
28    /// Exact successor logical version.
29    pub topology_version: TopologyVersion,
30    /// Inventory published atomically with this decision in the authority lease.
31    pub manifest: CatalogManifestRef,
32    /// Immutable decision append, retained across leader changes and recovery faults.
33    pub authority_sequence: u64,
34}
35
36impl TopologyCommit {
37    pub(crate) fn validate(&self) -> Result<(), TopologyError> {
38        self.parent_manifest.validate()?;
39        self.manifest.validate()?;
40        if self.protocol_version != TOPOLOGY_COMMIT_PROTOCOL_VERSION
41            || self.topology_version != self.parent_version.successor()?
42            || self.manifest == self.parent_manifest
43            || self.authority_sequence == 0
44        {
45            return Err(TopologyError::Invalid(
46                "invalid topology Commit binding".into(),
47            ));
48        }
49        Ok(())
50    }
51}
52
53impl TopologyAdmissionStatus {
54    pub(crate) fn has_target_commit(&self) -> bool {
55        self.commit.is_some()
56    }
57
58    // A committed target reserves execution until a separately certified installation/Release.
59    pub(crate) fn blocks_admission(&self) -> bool {
60        self.is_preparing()
61            || (self.has_target_commit() && self.phase != TopologyAdmissionPhase::Active)
62    }
63
64    pub(crate) fn validate_commit(&self, head: u64) -> Result<(), TopologyError> {
65        match (&self.commit, self.phase) {
66            (
67                None,
68                TopologyAdmissionPhase::Committed
69                | TopologyAdmissionPhase::Activating
70                | TopologyAdmissionPhase::Active,
71            )
72            | (Some(_), TopologyAdmissionPhase::Aborted { .. }) => Err(TopologyError::Invalid(
73                "topology phase and Commit disagree".into(),
74            )),
75            (None, _) => Ok(()),
76            (
77                Some(commit),
78                TopologyAdmissionPhase::Committed
79                | TopologyAdmissionPhase::Activating
80                | TopologyAdmissionPhase::Active,
81            ) => {
82                commit.validate()?;
83                if commit.operation_id != self.operation_id
84                    || commit.authority_sequence > self.status_sequence
85                    || (self.phase == TopologyAdmissionPhase::Committed
86                        && commit.authority_sequence != self.status_sequence)
87                    || commit.authority_sequence > head
88                    || !self.target_preparation_roster_complete()
89                    || self.target_preparations.iter().any(|receipt| {
90                        receipt.protocol_version != TOPOLOGY_COMMIT_PROTOCOL_VERSION
91                            || receipt.authority_sequence >= commit.authority_sequence
92                    })
93                {
94                    return Err(TopologyError::Invalid(
95                        "Commit requires the complete protocol-four target preparation roster"
96                            .into(),
97                    ));
98                }
99                Ok(())
100            }
101            (Some(_), _) => Err(TopologyError::Invalid(
102                "Commit outside a committed phase".into(),
103            )),
104        }
105    }
106
107    pub(crate) fn validate_commit_successor(
108        &self,
109        after: &Self,
110        sequence: u64,
111    ) -> Result<(), TopologyError> {
112        match (&self.commit, &after.commit) {
113            (None, None) => Ok(()),
114            (Some(prior), Some(next))
115                if prior == next
116                    && self.same_migration_binding(after)
117                    && self.target_preparations == after.target_preparations =>
118            {
119                Ok(())
120            }
121            (None, Some(commit))
122                if self.phase == TopologyAdmissionPhase::CutPrepared
123                    && after.phase == TopologyAdmissionPhase::Committed
124                    && commit.authority_sequence == sequence
125                    && self.operation_id == after.operation_id
126                    && self.plan == after.plan
127                    && self.admitted_by == after.admitted_by
128                    && self.admitted_sequence == after.admitted_sequence
129                    && self.cut == after.cut
130                    && self.preparation == after.preparation
131                    && self.migration_root == after.migration_root
132                    && self.target_preparations == after.target_preparations =>
133            {
134                after.validate_commit(sequence)
135            }
136            _ => Err(TopologyError::Invalid(
137                "authority cannot rewrite or roll back a topology Commit".into(),
138            )),
139        }
140    }
141}
laminar_core/cluster/control/topology/commit.rs

laminar_core/cluster/control/topology/
commit.rs
