laminar_db/pipeline_lifecycle/
topology_recovery_startup.rs

1//! Cold committed-target startup initializes control state and delegates actors to recovery.
2
3use std::sync::atomic::Ordering;
4
5use laminar_core::cluster::control::{TopologyCatalogState, TopologyError};
6
7use super::{DbError, DbState, LaminarDB, RuntimeMode};
8
9impl LaminarDB {
10    /// Prepare a cold committed target for the existing coordinated recovery supervisor.
11    /// Returns true when that supervisor must restore and release the actual target actors;
12    /// false means there is no live topology Commit and ordinary startup remains applicable: a
13    /// settled target restores its own committed checkpoint like any other deployment.
14    /// No actors, sink epochs or intake are started here. The DB remains Created and fenced.
15    ///
16    /// # Errors
17    /// Rejects live/closed/terminal instances, missing deployment identity, divergent target
18    /// inventory/environment, absent checkpoint storage or stale process/catalog authority.
19    pub async fn prepare_committed_cluster_topology_startup(&self) -> Result<bool, DbError> {
20        if !self.is_cluster_runtime() {
21            return Ok(false);
22        }
23        let catalog = self
24            .catalog_manifest_store
25            .lock()
26            .clone()
27            .ok_or(TopologyError::Fenced)?;
28        let state = catalog.topology_state().await?;
29        // Any other live Commit makes ordinary startup fail closed on its own recheck.
30        let Some(commit) = catalog.live_topology_commit().await?.filter(|live| {
31            matches!(&state, TopologyCatalogState::Versioned { committed: Some(commit), .. } if commit == live)
32        }) else {
33            return Ok(false);
34        };
35        let controller = self
36            .cluster_controller
37            .lock()
38            .clone()
39            .ok_or(TopologyError::Fenced)?;
40        let authority = controller
41            .checkpoint_authority()
42            .map_err(|error| TopologyError::Protocol(error.to_string()))?;
43        let (operation, _, target, descriptor) = authority
44            .topology_preparation_input(commit.operation_id)
45            .await?;
46        if operation.commit.as_ref() != Some(&commit)
47            || !controller.process_lease_is_live()
48            || controller.is_draining()
49        {
50            return Err(TopologyError::Fenced.into());
51        }
52        self.fence_coordinated_recovery_lifecycle();
53        controller.set_recovering(true);
54        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(45);
55        tokio::time::timeout_at(deadline, async {
56            let _topology = self.topology_ddl_lock.write().await;
57            let _lifecycle = self.lifecycle_lock.lock().await;
58            self.ensure_catalog_cleanup_unfenced("cold committed topology startup")?;
59            if self.is_closed()
60                || DbState::load(&self.state) != DbState::Created
61                || self.cluster_authority_revoked.load(Ordering::Acquire)
62                || self.durable_terminal_recovery_fence.load(Ordering::Acquire)
63                || self.terminal_pipeline_halt.load(Ordering::Acquire)
64                || self
65                    .startup_attempt
66                    .lock()
67                    .as_ref()
68                    .is_some_and(|attempt| !attempt.is_complete())
69                || self.runtime_handle.lock().await.is_some()
70                || !self.owned_source_tasks.lock().is_empty()
71                || !self.owned_sink_handles.lock().is_empty()
72                || !self.owned_connector_task_fences.lock().is_empty()
73            {
74                return Err(TopologyError::Fenced.into());
75            }
76            // Cold catalog bootstrap may replay the committed inventory, but runtime migration
77            // never uses bootstrap mutation authority. The following round still owns all actors.
78            self.restore_catalog_from_manifest().await?;
79            let identities = self.topology_definition_identities().await?;
80            if self.catalog_manifest_inventory()? != target.entries
81                || identities.pipeline != descriptor.target_pipeline
82                || identities.environment_sha256 != descriptor.environment_sha256
83            {
84                return Err(TopologyError::Fenced.into());
85            }
86            let (sources, sinks, streams, tables) = {
87                let manager = self.connector_manager.lock();
88                (
89                    manager.sources().clone(),
90                    manager.sinks().clone(),
91                    manager.streams().clone(),
92                    manager.tables().clone(),
93                )
94            };
95            let backing = self.validate_startup_durability(RuntimeMode::Cluster)?;
96            let exchange_plan = self.cluster_exchange_plan().await?;
97            let _assignment = self.assignment_adoption_lock.lock().await;
98            let pipeline = self
99                .initialize_checkpointing(
100                    crate::pipeline_identity::PipelineRegistrations::new(
101                        sources.values(),
102                        sinks.values(),
103                        streams.values(),
104                        tables.values(),
105                    ),
106                    &exchange_plan,
107                    RuntimeMode::Cluster,
108                    backing,
109                    Some(&descriptor.deployment_id),
110                )
111                .await?;
112            if pipeline.as_ref() != Some(&descriptor.target_pipeline)
113                || self.coordinator.lock().await.is_none()
114                || catalog.topology_state().await? != state
115            {
116                return Err(TopologyError::Fenced.into());
117            }
118            if catalog.topology_state().await?.committed_version() != Some(commit.topology_version)
119                || authority
120                    .topology_operation_status(commit.operation_id)
121                    .await?
122                    .as_ref()
123                    .and_then(|status| status.commit.as_ref())
124                    != Some(&commit)
125                || !controller.process_lease_is_live()
126                || controller.is_draining()
127            {
128                return Err(TopologyError::Fenced.into());
129            }
130            {
131                let _transition = self.cluster_authority_transition.lock();
132                self.topology_cut_hold.store(true, Ordering::Release);
133                self.source_gate.store(true, Ordering::Release);
134            }
135            crate::coordinated_recovery::queue_local_fault(
136                &controller,
137                &self.pending_recovery_fault,
138            )
139            .map_err(DbError::Checkpoint)?;
140            Ok(true)
141        })
142        .await
143        .map_err(|_| TopologyError::Contended)?
144    }
145}
laminar_db/pipeline_lifecycle/topology_recovery_startup.rs

laminar_db/pipeline_lifecycle/
topology_recovery_startup.rs
