laminar_db/db/topology/
preparation.rs

1//! Local compilation owns certificate publication; caller-supplied reports are never trusted.
2
3use super::{DbError, DbState, LaminarDB, TopologyAdmissionStatus, TopologyOperationId};
4use laminar_core::cluster::control::{TopologyAdmissionPhase, TopologyError};
5use std::sync::atomic::Ordering;
6
7impl LaminarDB {
8    /// Independently compile and durably certify an already admitted operation on this process.
9    /// This neither submits a mutation nor closes intake, restores state or authorizes target output.
10    /// The exact target and descriptor come from immutable shared authority, never the caller.
11    /// A disconnect may leave a successful certificate; query status or retry the same operation.
12    ///
13    /// # Errors
14    /// Requires Running state, live process/assignment authority, an adopted exact parent and a
15    /// matching compiled candidate. One existing compiler slot and a 45 second end-to-end deadline
16    /// bound read/compile/publication; publication retains the authority's 15 second/16-CAS bounds.
17    pub async fn prepare_cluster_topology_operation(
18        &self,
19        operation_id: TopologyOperationId,
20    ) -> Result<TopologyAdmissionStatus, DbError> {
21        tokio::time::timeout(std::time::Duration::from_secs(45), async {
22            self.ensure_topology_preparation_available()?;
23            let controller = self.cluster_controller.lock().clone().ok_or_else(|| TopologyError::Protocol("preparation requires the configured cluster controller".into()))?;
24            let before = controller.try_live_local_process_authority_identity().map_err(|_| TopologyError::Fenced)?;
25            let authority = controller.checkpoint_authority().map_err(|error| TopologyError::Protocol(error.to_string()))?;
26            let (operation, plan, _target, descriptor) = authority.topology_preparation_input(operation_id).await?;
27            if !matches!(operation.phase, TopologyAdmissionPhase::Planned | TopologyAdmissionPhase::Preparing | TopologyAdmissionPhase::Quiescing | TopologyAdmissionPhase::CutPrepared) {
28                return Err(TopologyError::Conflict("aborted topology operation cannot prepare participants".into()).into());
29            }
30            if plan.assignment.participant_incarnation(before.participant.node_id) != Some(before.participant.boot_incarnation) {
31                return Err(TopologyError::Fenced.into());
32            }
33            let compiled = self.validate_cluster_topology_change(plan.expected_parent, &descriptor.statements).await?;
34            if compiled != descriptor {
35                return Err(TopologyError::Conflict("this process compiled a divergent candidate; check binary, config, catalog and connector versions".into()).into());
36            }
37            self.ensure_topology_preparation_available()?;
38            let status = controller.certify_topology_candidate(operation_id, &operation.plan, before, &compiled).await?;
39            self.ensure_topology_preparation_available()?;
40            Ok(status)
41        }).await.map_err(|_| TopologyError::Contended)?
42    }
43
44    pub(super) fn ensure_topology_preparation_available(&self) -> Result<(), DbError> {
45        if self.shutdown.load(Ordering::Acquire) {
46            return Err(DbError::Shutdown);
47        }
48        if !self.is_cluster_runtime()
49            || DbState::load(&self.state) != DbState::Running
50            || self.cluster_authority_revoked.load(Ordering::Acquire)
51            || self.durable_terminal_recovery_fence.load(Ordering::Acquire)
52            || self.terminal_pipeline_halt.load(Ordering::Acquire)
53            || self.coordinated_recovery_in_progress()
54        {
55            return Err(TopologyError::Conflict("participant preparation requires a running parent with live process and recovery authority".into()).into());
56        }
57        self.ensure_catalog_cleanup_unfenced("topology preparation")
58    }
59}
laminar_db/db/topology/preparation.rs

laminar_db/db/topology/
preparation.rs
