laminar_db/db/topology/
mod.rs

1//! Durable topology status, admission and exact committed runtime lifecycle.
2
3use laminar_core::cluster::control::{
4    TopologyAdmissionStatus, TopologyCatalogState, TopologyJournalUsage, TopologyOperationId,
5    TopologyVersion,
6};
7use std::sync::atomic::Ordering;
8
9use super::{DbError, DbState, LaminarDB};
10
11mod activation;
12pub(super) mod catalog_changes;
13mod commit;
14pub(crate) use activation::InstalledTopologyRuntime;
15mod forwarding;
16mod installation;
17mod migration_root;
18mod planning;
19mod preparation;
20mod recovery;
21mod recovery_runtime;
22mod restore;
23mod retirement;
24mod submission;
25pub use submission::{ClusterTopologyAdoptionRequest, ClusterTopologyRequest};
26mod target_preparation;
27mod transport;
28pub use planning::{
29    ClusterTopologyObjectPlan, ClusterTopologyObjectTransition, ClusterTopologyValidation,
30    TopologyActivationRequirement, TopologyInitialization, TopologyValidationScope,
31};
32pub(crate) use restore::TopologyRuntimeMetadata;
33pub use restore::{PreparedTopologyRestore, PreparedTopologySourcePosition};
34
35/// Committed catalog version this process replayed, with the shuffle generation peers expect.
36/// Topology one is the legacy fabric and carries no fence.
37#[derive(Debug, Clone, Copy, PartialEq, Eq)]
38pub(crate) struct ReplayedTopology {
39    pub(crate) version: TopologyVersion,
40    pub(crate) fence: Option<laminar_core::shuffle::ShuffleTopologyFence>,
41}
42
43impl ReplayedTopology {
44    pub(crate) const LEGACY_BASELINE: Self = Self {
45        version: TopologyVersion::LEGACY_BASELINE,
46        fence: None,
47    };
48
49    pub(crate) fn from_catalog(catalog: &TopologyCatalogState) -> Result<Option<Self>, DbError> {
50        let TopologyCatalogState::Versioned { committed, .. } = catalog else {
51            return Ok(None);
52        };
53        let Some(commit) = committed else {
54            return Ok(Some(Self::LEGACY_BASELINE));
55        };
56        let fence = laminar_core::shuffle::ShuffleTopologyFence::from_manifest(
57            commit.topology_version,
58            &commit.manifest,
59        )
60        .map_err(|error| {
61            laminar_core::cluster::control::TopologyError::Invalid(error.to_string())
62        })?;
63        Ok(Some(Self {
64            version: commit.topology_version,
65            fence: Some(fence),
66        }))
67    }
68}
69
70/// Durable catalog version and this process's independently observed runtime activation.
71#[derive(Debug, Clone, serde::Serialize)]
72pub struct ClusterTopologyStatus {
73    /// Authority state, including an explicit unversioned legacy state.
74    pub catalog: TopologyCatalogState,
75    /// Committed logical version, absent for an unadopted or uninitialized catalog.
76    pub committed_version: Option<TopologyVersion>,
77    /// Version replayed by this process and running with live authority and intake released.
78    /// A persisted inventory alone never sets this field.
79    pub locally_active_version: Option<TopologyVersion>,
80    /// Committed version whose migration still freezes the complete owner map: it is not yet
81    /// Active with its own committed checkpoint. Rescale only when this is absent.
82    pub live_migration: Option<TopologyVersion>,
83    /// Bounded request journal occupancy; new requests are rejected once it is full.
84    pub journal: TopologyJournalUsage,
85}
86
87impl LaminarDB {
88    /// Commit of a migration that still owns startup, assignment and recovery authority.
89    pub(crate) async fn live_topology_commit(
90        &self,
91    ) -> Result<Option<laminar_core::cluster::control::TopologyCommit>, DbError> {
92        let store = self
93            .catalog_manifest_store
94            .lock()
95            .clone()
96            .ok_or(laminar_core::cluster::control::TopologyError::Fenced)?;
97        Ok(store.live_topology_commit().await?)
98    }
99
100    /// Read an admitted request's definitive status; no request ownership depends on this call.
101    ///
102    /// # Errors
103    /// Fails outside cluster mode or for unavailable/corrupt durable evidence.
104    pub async fn cluster_topology_operation_status(
105        &self,
106        operation_id: TopologyOperationId,
107    ) -> Result<Option<TopologyAdmissionStatus>, DbError> {
108        if !self.is_cluster_runtime() {
109            return Err(DbError::InvalidOperation(
110                "cluster topology status requires cluster mode".into(),
111            ));
112        }
113        let store = self.catalog_manifest_store.lock().clone().ok_or_else(|| {
114            DbError::InvalidOperation("cluster topology status requires a catalog authority".into())
115        })?;
116        Ok(store.operation_status(operation_id).await?)
117    }
118
119    /// Read the cluster's durable topology and this process's local activation evidence.
120    ///
121    /// This is a control-path read, with no catalog mutation, checkpoint allocation or source
122    /// effects. It does not declare that other participants have activated.
123    ///
124    /// # Errors
125    /// Fails outside cluster mode or for unavailable, corrupt or inconsistent authority.
126    pub async fn cluster_topology_status(&self) -> Result<ClusterTopologyStatus, DbError> {
127        if !self.is_cluster_runtime() {
128            return Err(DbError::InvalidOperation(
129                "cluster topology status requires cluster mode".into(),
130            ));
131        }
132        let store = self.catalog_manifest_store.lock().clone().ok_or_else(|| {
133            DbError::InvalidOperation("cluster topology status requires a catalog authority".into())
134        })?;
135        let catalog = store.topology_state().await?;
136        let committed_version = catalog.committed_version();
137        let replayed_version = self
138            .replayed_topology
139            .lock()
140            .map(|replayed| replayed.version);
141        let controller = self.cluster_controller.lock().clone();
142        let live = store.live_topology_commit().await?;
143        // A settled target is an ordinary deployment: running with open gates is its activation.
144        let release_observed = match &catalog {
145            TopologyCatalogState::Versioned {
146                committed: Some(commit),
147                ..
148            } if live.as_ref() == Some(commit) => {
149                let binding = self.installed_topology_runtime.lock().clone();
150                match binding.filter(|binding| {
151                    binding.input.operation().commit.as_ref() == Some(commit)
152                        && !binding.shutdown.is_cancelled()
153                        && self.owned_source_tasks.lock().iter().all(
154                            crate::pipeline::streaming_coordinator::SourceTaskLease::is_running,
155                        )
156                        && self
157                            .owned_sink_handles
158                            .lock()
159                            .iter()
160                            .all(crate::sink_task::SinkTaskHandle::is_ready)
161                }) {
162                    Some(binding) if binding.recovery.is_some() => {
163                        self.recovered_topology_runtime_is_active(&binding).await?
164                    }
165                    Some(binding) => store
166                        .operation_status(commit.operation_id)
167                        .await?
168                        .is_some_and(|status| {
169                            status.phase
170                                == laminar_core::cluster::control::TopologyAdmissionPhase::Active
171                                && status.activation.as_ref().is_some_and(|round| {
172                                    round.release.as_ref().is_some_and(|release| {
173                                        Some(release.authority_sequence)
174                                            == binding.released_sequence
175                                    }) && round.installations.iter().any(|receipt| {
176                                        receipt.runtime_id == binding.runtime_id
177                                            && receipt.process == binding.input.process()
178                                    })
179                                })
180                        }),
181                    None => false,
182                }
183            }
184            _ => true,
185        };
186        let locally_active_version = if DbState::load(&self.state) == DbState::Running
187            && release_observed
188            && !self.source_gate.load(Ordering::Acquire)
189            && !self.topology_cut_hold.load(Ordering::Acquire)
190            && !self.cluster_authority_revoked.load(Ordering::Acquire)
191            && !self.durable_terminal_recovery_fence.load(Ordering::Acquire)
192            && !self.terminal_pipeline_halt.load(Ordering::Acquire)
193            && !self.coordinated_recovery_in_progress()
194            && controller.as_ref().is_some_and(|controller| {
195                controller.process_lease_is_live() && !controller.is_recovering()
196            })
197            && replayed_version == committed_version
198        {
199            replayed_version
200        } else {
201            None
202        };
203        Ok(ClusterTopologyStatus {
204            catalog,
205            committed_version,
206            locally_active_version,
207            live_migration: live.map(|commit| commit.topology_version),
208            journal: store.topology_journal_usage().await?,
209        })
210    }
211}
212
213impl LaminarDB {
214    pub(super) async fn validate_sealed_topology_bootstrap(
215        &self,
216        manifest: &laminar_core::cluster::control::CatalogManifest,
217        parsed: &[(
218            String,
219            laminar_sql::parser::StreamingStatement,
220            String,
221            laminar_core::cluster::control::CatalogObjectKind,
222        )],
223    ) -> Result<Vec<crate::handle::ExecuteResult>, DbError> {
224        let matches = |catalog: &laminar_core::cluster::control::CatalogManifest| {
225            parsed.len() == catalog.entries.len()
226                && parsed
227                    .iter()
228                    .zip(&catalog.entries)
229                    .all(|((ddl, _, name, kind), entry)| {
230                        ddl == &entry.ddl && name == &entry.canonical_name && kind == &entry.kind
231                    })
232        };
233        if !matches(manifest) {
234            let store = self.catalog_manifest_store.lock().clone().ok_or_else(|| {
235                DbError::Pipeline("cluster catalog manifest store is not configured".into())
236            })?;
237            let original = store
238                .original_topology_catalog(
239                    &manifest
240                        .reference()
241                        .map_err(laminar_core::cluster::control::TopologyError::from)?,
242                )
243                .await?;
244            if !original.as_ref().is_some_and(matches) {
245                return Err(DbError::Pipeline(format!(
246                    "configured cluster catalog must exactly match the complete ordered sealed inventory or its original adopted bootstrap (configured entries: {}, sealed entries: {})",
247                    parsed.len(), manifest.entries.len()
248                )));
249            }
250        }
251        parsed
252            .iter()
253            .map(|(_, statement, _, _)| {
254                let (name, _, statement_type) = super::catalog_create_identity(statement)?
255                    .ok_or_else(|| {
256                        DbError::InvalidOperation(
257                            "bootstrap assertion requires typed CREATE statements".into(),
258                        )
259                    })?;
260                Ok(crate::handle::ExecuteResult::Ddl(crate::handle::DdlInfo {
261                    statement_type: statement_type.to_string(),
262                    object_name: name,
263                    topology_operation: None,
264                    applied: false,
265                }))
266            })
267            .collect()
268    }
269}
laminar_db/db/topology/mod.rs

laminar_db/db/topology/
mod.rs
