laminar_db/db/topology/
installation.rs

1//! Transfer an exact committed image into the existing, still-held startup lifecycle.
2
3use std::sync::atomic::Ordering;
4use std::sync::Arc;
5
6use laminar_core::cluster::control::{TopologyError, TopologyRestoreInput};
7use laminar_core::shuffle::ShuffleTopologyFence;
8
9use super::{DbError, DbState, LaminarDB, PreparedTopologyRestore};
10
11impl LaminarDB {
12    /// Install the exact committed catalog, restored graph and connector actors with intake held.
13    /// The existing startup owner outlives caller cancellation. No target checkpoint, source
14    /// acknowledgement, readiness receipt, external epoch admission or Release is invented.
15    /// Success means the local control loop is installed, never that the topology is active.
16    ///
17    /// # Errors
18    /// Rejects foreign/uncommitted/executed images, stale authority, unresolved predecessor work,
19    /// unsupported atomic source starts, local faults and the 45-second installation budget.
20    /// Failed installation retains Commit, the catalog and namespace ownership for root recovery.
21    pub async fn install_committed_cluster_topology(
22        self: &Arc<Self>,
23        mut image: PreparedTopologyRestore,
24    ) -> Result<(), DbError> {
25        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(45);
26        tokio::time::timeout_at(
27            deadline,
28            self.prepare_cluster_topology_transport(&mut image),
29        )
30        .await
31        .map_err(|_| TopologyError::Contended)??;
32        self.start_with_topology_image(image, deadline).await
33    }
34
35    pub(crate) async fn validate_topology_installation(
36        &self,
37        input: &TopologyRestoreInput,
38    ) -> Result<(), DbError> {
39        self.ensure_topology_runtime_held()?;
40        let controller = self
41            .cluster_controller
42            .lock()
43            .clone()
44            .ok_or(TopologyError::Fenced)?;
45        let fresh = controller
46            .committed_topology_restore_input(input.operation().operation_id)
47            .await?;
48        if if DbState::load(&self.state) == DbState::Starting {
49            !fresh.same_restore_requirements(input)
50        } else {
51            !fresh.same_installed_generation(input)
52        } {
53            return Err(TopologyError::Fenced.into());
54        }
55        if controller.is_recovering() {
56            return Err(TopologyError::Fenced.into());
57        }
58        self.validate_topology_transport_identity(input)?;
59        self.ensure_topology_runtime_held()
60    }
61
62    pub(crate) fn validate_topology_transport_identity(
63        &self,
64        input: &TopologyRestoreInput,
65    ) -> Result<(), DbError> {
66        let controller = self
67            .cluster_controller
68            .lock()
69            .clone()
70            .ok_or(TopologyError::Fenced)?;
71        let target = ShuffleTopologyFence::from_manifest(
72            input.descriptor().target_version,
73            &input.plan().target_manifest,
74        )
75        .map_err(|error| TopologyError::Invalid(error.to_string()))?;
76        let sender = self
77            .shuffle_sender
78            .lock()
79            .clone()
80            .ok_or(TopologyError::Fenced)?;
81        let receiver = self
82            .shuffle_receiver
83            .lock()
84            .clone()
85            .ok_or(TopologyError::Fenced)?;
86        let registry = self
87            .vnode_registry
88            .lock()
89            .clone()
90            .ok_or(TopologyError::Fenced)?;
91        let assignment = registry.versioned_snapshot();
92        let owners = assignment
93            .owners()
94            .iter()
95            .map(|owner| owner.0)
96            .collect::<Vec<_>>();
97        if assignment.version() != input.assignment().assignment_version
98            || !input.assignment().matches_owner_map(&owners)
99            || sender.topology_fence() != Some(target)
100            || receiver.topology_fence() != Some(target)
101            || sender.active_assignment_digest() != Some(input.assignment().digest())
102            || receiver.active_assignment_digest() != Some(input.assignment().digest())
103            || controller.try_live_local_process_authority_identity().ok() != Some(input.process())
104            || controller.is_draining()
105        {
106            return Err(TopologyError::Fenced.into());
107        }
108        Ok(())
109    }
110
111    pub(crate) fn ensure_topology_installation_held(&self) -> Result<(), DbError> {
112        if DbState::load(&self.state) != DbState::Starting {
113            return Err(TopologyError::Fenced.into());
114        }
115        self.ensure_topology_runtime_held()
116    }
117
118    pub(crate) fn ensure_topology_runtime_held(&self) -> Result<(), DbError> {
119        if self.is_closed() {
120            return Err(DbError::Shutdown);
121        }
122        if !self.is_cluster_runtime()
123            || !matches!(
124                DbState::load(&self.state),
125                DbState::Starting | DbState::Running
126            )
127            || !self.source_gate.load(Ordering::Acquire)
128            || !self.topology_cut_hold.load(Ordering::Acquire)
129            || self.cluster_authority_revoked.load(Ordering::Acquire)
130            || self.durable_terminal_recovery_fence.load(Ordering::Acquire)
131            || self.terminal_pipeline_halt.load(Ordering::Acquire)
132            || self.coordinated_recovery_in_progress()
133            || self.pending_recovery_fault.load(Ordering::Acquire) != 0
134            || self.last_fault.lock().is_some()
135        {
136            return Err(TopologyError::Fenced.into());
137        }
138        self.ensure_catalog_cleanup_unfenced("committed topology installation")
139    }
140}
laminar_db/db/topology/installation.rs

laminar_db/db/topology/
installation.rs
