laminar_core/cluster/control/topology/
mod.rs

1//! Control-path identities and explicit adoption of a legacy sealed catalog.
2//!
3//! Adoption changes the authority encoding, not the processing graph. Runtime migrations must
4//! still establish a checkpoint cut, validate state mappings and commit coordinated release.
5
6use std::num::NonZeroU64;
7
8use serde::{Deserialize, Serialize};
9use uuid::Uuid;
10
11use super::{CatalogManifestError, CatalogManifestRef, LeaseError};
12use crate::error_codes;
13
14mod activation;
15mod admission;
16mod checkpoint_continuity;
17mod commit;
18mod compatibility;
19mod migration_root;
20mod preparation;
21mod recovery;
22mod restore;
23mod source_initialization;
24mod target_preparation;
25pub use activation::{
26    TopologyActivation, TopologyInstallationReceipt, TopologyRelease,
27    TOPOLOGY_INSTALLATION_PROTOCOL_VERSION,
28};
29pub(crate) use admission::MAX_TOPOLOGY_PLAN_BYTES;
30pub use admission::{
31    TopologyAbortReason, TopologyAdmissionPhase, TopologyAdmissionPlan, TopologyAdmissionStatus,
32    TopologyCheckpointCut, TopologyCutCommit, TopologyJournalUsage, TopologyPlanRef,
33    MAX_TOPOLOGY_OPERATIONS,
34};
35pub use commit::{TopologyCommit, TOPOLOGY_COMMIT_PROTOCOL_VERSION};
36pub use compatibility::{
37    ClusterTopologyObjectPlan, ClusterTopologyObjectTransition, ClusterTopologyValidation,
38    TopologyActivationRequirement, TopologyCompatibilityRef, TopologyInitialization,
39    TopologyValidationScope,
40};
41pub(crate) use migration_root::MAX_TOPOLOGY_ROOT_BYTES;
42pub use migration_root::{
43    TopologyMigrationRoot, TopologyMigrationRootBinding, TopologyMigrationRootRef,
44    TopologyPreservedObject, TopologySubscriptionRoot, MAX_TOPOLOGY_ROOT_MANIFEST_BYTES,
45};
46pub use preparation::{
47    TopologyParticipantCertificate, TopologyPreparation, TOPOLOGY_PREPARATION_PROTOCOL_VERSION,
48    TOPOLOGY_SUBMISSION_PROTOCOL_VERSION,
49};
50pub use recovery::{
51    TopologyRecoveryBinding, TopologyRecoveryCut, TopologyRecoveryInput,
52    TOPOLOGY_RECOVERY_PROTOCOL_VERSION,
53};
54pub use restore::TopologyRestoreInput;
55pub use source_initialization::{TopologySourceInitialization, MAX_TOPOLOGY_SOURCE_CHANNELS};
56pub use target_preparation::{
57    TopologyTargetPreparationReceipt, TOPOLOGY_TARGET_PREPARATION_PROTOCOL_VERSION,
58};
59
60/// Version of the topology adoption protocol understood by this binary.
61pub const TOPOLOGY_PROTOCOL_VERSION: u16 = 1;
62
63/// Logical topology version, independent of catalog/checkpoint serialization versions.
64#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
65#[serde(transparent)]
66pub struct TopologyVersion(NonZeroU64);
67
68impl TopologyVersion {
69    /// The explicitly adopted legacy inventory is topology one.
70    pub const LEGACY_BASELINE: Self = Self(NonZeroU64::MIN);
71
72    /// Construct a nonzero logical version.
73    ///
74    /// # Errors
75    /// Zero never identifies a topology.
76    pub fn new(value: u64) -> Result<Self, TopologyError> {
77        NonZeroU64::new(value)
78            .map(Self)
79            .ok_or_else(|| TopologyError::Invalid("topology version must be nonzero".into()))
80    }
81
82    /// Numeric logical version.
83    #[must_use]
84    pub const fn get(self) -> u64 {
85        self.0.get()
86    }
87
88    /// Allocate the exact successor without wrapping.
89    ///
90    /// # Errors
91    /// Fails when the topology version domain is exhausted.
92    pub fn successor(self) -> Result<Self, TopologyError> {
93        self.get()
94            .checked_add(1)
95            .ok_or_else(|| TopologyError::Invalid("topology version exhausted".into()))
96            .and_then(Self::new)
97    }
98}
99
100/// Durable request identity for a topology operation.
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
102#[serde(try_from = "Uuid")]
103pub struct TopologyOperationId(Uuid);
104
105impl TryFrom<Uuid> for TopologyOperationId {
106    type Error = TopologyError;
107
108    fn try_from(value: Uuid) -> Result<Self, Self::Error> {
109        if value.is_nil() {
110            return Err(TopologyError::Invalid(
111                "topology operation identity must be nonzero".into(),
112            ));
113        }
114        Ok(Self(value))
115    }
116}
117
118impl TopologyOperationId {
119    /// UUID supplied by the request owner and reused on every retry.
120    #[must_use]
121    pub const fn get(self) -> Uuid {
122        self.0
123    }
124}
125
126/// Immutable identity of the existing inventory at the authority format upgrade.
127///
128/// This record always represents the original topology one. Later Commit decisions carry the
129/// current catalog separately, retaining this exact adoption and all historical identities.
130#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
131#[serde(deny_unknown_fields)]
132pub struct LegacyTopologyBaseline {
133    /// Protocol the upgrading binary must understand.
134    pub protocol_version: u16,
135    /// Explicitly assigned baseline version, always one.
136    pub topology_version: TopologyVersion,
137    /// Original manifest reference; bytes, hash and object generations are not rewritten.
138    pub manifest: CatalogManifestRef,
139    /// Existing checkpoint/control deployment identity.
140    pub deployment_id: String,
141    /// Idempotency identity of the successful upgrade request.
142    pub operation_id: TopologyOperationId,
143    /// The exact shared authority append that admitted this baseline.
144    pub authority_sequence: u64,
145}
146
147impl LegacyTopologyBaseline {
148    pub(super) fn validate(&self) -> Result<(), TopologyError> {
149        if self.protocol_version != TOPOLOGY_PROTOCOL_VERSION {
150            return Err(TopologyError::Protocol(format!(
151                "unsupported topology protocol {}",
152                self.protocol_version
153            )));
154        }
155        if self.topology_version != TopologyVersion::LEGACY_BASELINE || self.authority_sequence == 0
156        {
157            return Err(TopologyError::Invalid(
158                "legacy adoption must bind topology one to a nonzero authority sequence".into(),
159            ));
160        }
161        let deployment = Uuid::parse_str(&self.deployment_id)
162            .map_err(|_| TopologyError::Invalid("invalid topology deployment identity".into()))?;
163        if deployment.is_nil() || deployment.to_string() != self.deployment_id {
164            return Err(TopologyError::Invalid(
165                "topology deployment identity must be a canonical nonzero UUID".into(),
166            ));
167        }
168        self.manifest.validate()?;
169        Ok(())
170    }
171}
172
173/// Durable catalog state; missing version metadata is never an implicit current version.
174#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
175#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
176pub enum TopologyCatalogState {
177    /// No catalog inventory has been sealed.
178    Uninitialized,
179    /// A sealed inventory exists, but the format upgrade has not been admitted.
180    LegacySealed {
181        /// Exact existing inventory, with no inferred logical topology version.
182        manifest: CatalogManifestRef,
183    },
184    /// Explicitly adopted inventory. Runtime activation is separate evidence.
185    Versioned {
186        /// Durable baseline identity.
187        baseline: LegacyTopologyBaseline,
188        /// Latest irreversible catalog decision, absent while topology one remains committed.
189        #[serde(default, skip_serializing_if = "Option::is_none")]
190        committed: Option<TopologyCommit>,
191    },
192}
193
194impl TopologyCatalogState {
195    /// Current committed logical version; local installation and Release are separate evidence.
196    #[must_use]
197    pub fn committed_version(&self) -> Option<TopologyVersion> {
198        match self {
199            Self::Versioned {
200                baseline,
201                committed,
202            } => Some(
203                committed
204                    .as_ref()
205                    .map_or(baseline.topology_version, |commit| commit.topology_version),
206            ),
207            Self::Uninitialized | Self::LegacySealed { .. } => None,
208        }
209    }
210}
211
212/// Result of one explicit format upgrade, including the original winner on an identical retry.
213#[derive(Debug, Clone, PartialEq, Eq)]
214pub enum TopologyAdoptionOutcome {
215    /// This request admitted the baseline.
216    Created(LegacyTopologyBaseline),
217    /// The same baseline was already admitted, possibly by another request.
218    Existing(LegacyTopologyBaseline),
219}
220
221/// Typed, stable control-path failures. No error authorizes a topology or state fallback.
222#[derive(Debug, thiserror::Error)]
223pub enum TopologyError {
224    /// Shared authority I/O or validation failed.
225    #[error("[{code}] topology authority: {0}", code = error_codes::TOPOLOGY_AUTHORITY_FAILED)]
226    Authority(#[from] LeaseError),
227    /// Catalog content failed validation.
228    #[error("[{code}] topology catalog: {0}", code = error_codes::TOPOLOGY_INVALID)]
229    Catalog(CatalogManifestError),
230    /// Malformed proposal, durable state or exhausted identity domain.
231    #[error("[{code}] invalid topology evidence: {0}", code = error_codes::TOPOLOGY_INVALID)]
232    Invalid(String),
233    /// Expected parent/reference/deployment does not match durable authority.
234    #[error("[{code}] topology parent conflict: {0}", code = error_codes::TOPOLOGY_PARENT_CONFLICT)]
235    Conflict(String),
236    /// Supplied leader proof is stale.
237    #[error("[{code}] topology operation was fenced by current leader authority", code = error_codes::TOPOLOGY_FENCED)]
238    Fenced,
239    /// A required protocol/storage capability is not available.
240    #[error("[{code}] topology protocol unavailable: {0}", code = error_codes::TOPOLOGY_PROTOCOL_UNSUPPORTED)]
241    Protocol(String),
242    /// The bounded CAS/deadline budget was exhausted; authoritative status must be reread.
243    #[error("[{code}] topology authority is contended or its outcome is uncertain; reread status and retry the same operation identity", code = error_codes::TOPOLOGY_AUTHORITY_CONTENDED)]
244    Contended,
245    /// A status read exceeded its control-path deadline.
246    #[error("[{code}] topology status read exceeded its 15 second deadline; retry this read", code = error_codes::TOPOLOGY_AUTHORITY_CONTENDED)]
247    ReadTimedOut,
248    /// A candidate requires an unsupported semantic transformation or execution contract.
249    #[error("[{code}] unsupported topology change: {0}", code = error_codes::TOPOLOGY_CHANGE_UNSUPPORTED)]
250    Unsupported(String),
251    /// Another bounded local candidate compilation owns this process's planning slot.
252    #[error("[{code}] topology validation is busy on this process; retry validation", code = error_codes::TOPOLOGY_AUTHORITY_CONTENDED)]
253    PlanningBusy,
254    /// Local validation exceeded its bounded control-path budget without admitting an operation.
255    #[error("[{code}] topology validation exceeded its 30 second deadline; no operation was admitted", code = error_codes::TOPOLOGY_AUTHORITY_CONTENDED)]
256    PlanningTimedOut,
257}
258
259impl From<CatalogManifestError> for TopologyError {
260    fn from(error: CatalogManifestError) -> Self {
261        match error {
262            CatalogManifestError::Authority(error) => Self::Authority(error),
263            error => Self::Catalog(error),
264        }
265    }
266}
267
268impl TopologyError {
269    /// Stable registry code for this typed failure.
270    #[must_use]
271    pub const fn code(&self) -> &'static str {
272        match self {
273            Self::Authority(_) => error_codes::TOPOLOGY_AUTHORITY_FAILED,
274            Self::Catalog(_) | Self::Invalid(_) => error_codes::TOPOLOGY_INVALID,
275            Self::Conflict(_) => error_codes::TOPOLOGY_PARENT_CONFLICT,
276            Self::Fenced => error_codes::TOPOLOGY_FENCED,
277            Self::Protocol(_) => error_codes::TOPOLOGY_PROTOCOL_UNSUPPORTED,
278            Self::Unsupported(_) => error_codes::TOPOLOGY_CHANGE_UNSUPPORTED,
279            Self::Contended | Self::ReadTimedOut | Self::PlanningBusy | Self::PlanningTimedOut => {
280                error_codes::TOPOLOGY_AUTHORITY_CONTENDED
281            }
282        }
283    }
284}
285
286#[cfg(test)]
287mod tests;
laminar_core/cluster/control/topology/mod.rs

laminar_core/cluster/control/topology/
mod.rs
