laminar_db/db/topology/
migration_root.rs

1//! DB-owned exact-cut metadata staging for the future migration driver.
2
3use super::{DbError, LaminarDB, TopologyAdmissionStatus, TopologyOperationId};
4use laminar_core::checkpoint::ObjectStoreCheckpointStore;
5use laminar_core::cluster::control::topology::ClusterTopologyObjectTransition;
6use laminar_core::cluster::control::{
7    CatalogManifest, CatalogObjectKind, ClusterTopologyValidation, TopologyError,
8    TopologySourceInitialization,
9};
10use std::sync::atomic::Ordering;
11
12impl LaminarDB {
13    /// Durably stage preserved state/progress and source initialization for a prepared addition.
14    /// Requires the current leader's running parent and held cut. The configured checkpoint
15    /// reader supplies metadata only; no state is copied or restored and no target actor starts.
16    /// New sources use their configured connector's read-only initialization contract. The first
17    /// sealed global cursor survives a disconnect before publication and is never resolved again.
18    /// Built-in Kafka supports explicit topics with earliest/latest; other connectors fail closed.
19    ///
20    /// # Errors
21    /// Rejects a missing held cut, shutdown/recovery/leader/process fences, damaged metadata,
22    /// unsupported initialization or a 30 second total deadline (authority publication is bounded
23    /// to 15 seconds/16 CAS attempts). A cancelled caller may have durably staged the root; query
24    /// operation status or retry the same operation identity.
25    pub async fn stage_cluster_topology_migration_root(
26        &self,
27        operation_id: TopologyOperationId,
28    ) -> Result<TopologyAdmissionStatus, DbError> {
29        tokio::time::timeout(std::time::Duration::from_secs(30), async {
30            self.ensure_topology_root_available()?;
31            let _compiler = self.topology_validation_lock.try_lock()
32                .map_err(|_| TopologyError::PlanningBusy)?;
33            let controller = self.cluster_controller.lock().clone().ok_or_else(|| {
34                TopologyError::Protocol("root staging requires the configured controller".into())
35            })?;
36            let authority = controller
37                .checkpoint_authority()
38                .map_err(|e| TopologyError::Protocol(e.to_string()))?;
39            let operation = authority
40                .topology_operation_status(operation_id)
41                .await?
42                .ok_or_else(|| TopologyError::Conflict("unknown topology operation".into()))?;
43            let backing = self.checkpoint_object_store()?.ok_or_else(|| {
44                TopologyError::Protocol("root staging requires checkpoint storage".into())
45            })?;
46            let max_bytes = self
47                .config
48                .checkpoint
49                .as_ref()
50                .and_then(|c| c.max_node_data_bytes)
51                .unwrap_or(laminar_core::checkpoint::checkpoint_store::DEFAULT_MAX_CHECKPOINT_NODE_DATA_BYTES);
52            let store = ObjectStoreCheckpointStore::new(backing, "")
53                .with_key_group_count(self.checkpoint_key_groups())
54                .with_max_node_data_bytes(max_bytes)?;
55            let status = controller
56                .stage_topology_migration_root_with_initialization(
57                    &store, operation_id, &operation.plan,
58                    |target, descriptor| self.resolve_topology_source_initializations(target, descriptor),
59                )
60                .await?;
61            self.ensure_topology_root_available()?;
62            Ok(status)
63        })
64        .await
65        .map_err(|_| TopologyError::Contended)?
66    }
67
68    async fn resolve_topology_source_initializations(
69        &self,
70        target: CatalogManifest,
71        descriptor: ClusterTopologyValidation,
72    ) -> Result<Vec<TopologySourceInitialization>, TopologyError> {
73        let map_error = |error| match error {
74            DbError::Topology(error) => error,
75            error => TopologyError::Invalid(error.to_string()),
76        };
77        self.ensure_topology_root_available().map_err(map_error)?;
78        let candidate = self.isolated_topology_catalog().map_err(map_error)?;
79        for entry in &target.entries {
80            super::planning::replay_entry(&candidate, entry)
81                .await
82                .map_err(map_error)?;
83        }
84        if candidate
85            .reconcile_catalog_manifest_inventory(&target)
86            .map_err(map_error)?
87            != target.entries
88        {
89            return Err(TopologyError::Invalid(
90                "isolated target replay changed its exact inventory or incarnations".into(),
91            ));
92        }
93        // Resolve the immutable target with the exact configured factories/environment. No graph,
94        // source actor or sink is installed, and no historical state/data payload is copied.
95        let identities = candidate
96            .topology_definition_identities()
97            .await
98            .map_err(map_error)?;
99        if identities.pipeline != descriptor.target_pipeline
100            || identities.environment_sha256 != descriptor.environment_sha256
101        {
102            return Err(TopologyError::Conflict(
103                "resolved source configuration differs from the certified target; prepare a new operation".into(),
104            ));
105        }
106        let sources = candidate.connector_manager.lock().sources().clone();
107        let mut initialized = Vec::new();
108        for object in descriptor.objects.iter().filter(|object| {
109            object.kind == CatalogObjectKind::Source
110                && object.transition == ClusterTopologyObjectTransition::AddFutureOnly
111        }) {
112            self.ensure_topology_root_available().map_err(map_error)?;
113            let registration = sources.get(&object.name).ok_or_else(|| {
114                TopologyError::Invalid(format!(
115                    "new source '{}' has no configured connector",
116                    object.name
117                ))
118            })?;
119            let config = candidate
120                .build_registered_source_config(&object.name, registration)
121                .map_err(map_error)?;
122            let mut connector = candidate
123                .connector_registry
124                .create_source(&config, None)
125                .map_err(|e| {
126                    TopologyError::Unsupported(format!("source '{}': {e}", object.name))
127                })?;
128            let checkpoint = connector
129                .resolve_initial_position(&config)
130                .await
131                .map_err(|e| {
132                    TopologyError::Unsupported(format!(
133                        "source '{}' initialization: {e}",
134                        object.name
135                    ))
136                })?;
137            initialized.push(TopologySourceInitialization {
138                name: object.name.clone(),
139                catalog_generation: object.catalog_generation,
140                compatibility_sha256: object.compatibility_sha256.clone(),
141                checkpoint: crate::checkpoint_coordinator::source_to_connector_checkpoint(
142                    &checkpoint,
143                ),
144            });
145            // Connector drop retains its existing terminal-task ownership. Metadata clients are
146            // created/destroyed inside tracked blocking work; no lifecycle cleanup future is added.
147        }
148        self.ensure_topology_root_available().map_err(map_error)?;
149        Ok(initialized)
150    }
151
152    pub(super) fn ensure_topology_root_available(&self) -> Result<(), DbError> {
153        self.ensure_topology_preparation_available()?;
154        if !self.topology_cut_hold.load(Ordering::Acquire)
155            || !self.source_gate.load(Ordering::Acquire)
156        {
157            return Err(TopologyError::Conflict(
158                "root staging requires the prepared old-topology intake hold".into(),
159            )
160            .into());
161        }
162        Ok(())
163    }
164}
laminar_db/db/topology/migration_root.rs

laminar_db/db/topology/
migration_root.rs
