laminar_db/db/
catalog_bootstrap.rs

1//! Resolve and admit a private catalog, seal it, then install its committed contracts.
2
3use std::collections::HashSet;
4use std::sync::atomic::Ordering;
5use std::sync::Arc;
6
7use futures::FutureExt;
8use laminar_core::cluster::control::{CatalogManifest, CatalogObjectKind, LeaderProof};
9use laminar_sql::parser::{parse_streaming_sql, StreamingStatement};
10
11use super::{
12    validate_cluster_catalog_create, CatalogBootstrapGuard, DbState, LaminarDB, CATALOG_BOOTSTRAP,
13};
14use crate::error::DbError;
15use crate::handle::ExecuteResult;
16
17type BootstrapEntry = (String, StreamingStatement, String, CatalogObjectKind);
18
19impl LaminarDB {
20    /// Resolve, validate and durably seal the complete startup catalog before activation.
21    /// Existing sealed catalogs accept their unchanged original definitions without discovery.
22    ///
23    /// # Errors
24    /// Rejects divergent definitions, unsupported plans, stale leader authority and failed sealing.
25    /// Authorized external preparation may leave unused artifacts if publication fails.
26    pub async fn execute_cluster_bootstrap_batch(
27        &self,
28        sql: &[String],
29    ) -> Result<Vec<ExecuteResult>, DbError> {
30        let _creation = self.schema_creation_lock.lock().await;
31        {
32            let _catalog = self.topology_ddl_lock.read().await;
33            self.ensure_bootstrap_available()?;
34        }
35        self.connector_registry.freeze();
36        let parsed = parse_bootstrap(self, sql)?;
37        // No catalog lock spans discovery or object-store I/O. The creation gate serializes DDL.
38        if let Some(manifest) = self.restore_catalog_from_manifest().boxed().await? {
39            return self
40                .validate_sealed_topology_bootstrap(&manifest, &parsed)
41                .await;
42        }
43        let store = self.catalog_manifest_store.lock().clone().ok_or_else(|| {
44            DbError::Pipeline("cluster catalog manifest store is not configured".into())
45        })?;
46        let controller = self.cluster_controller.lock().clone().ok_or_else(|| {
47            DbError::Pipeline(
48                "[LDB-6043] cluster catalog bootstrap requires a cluster controller".into(),
49            )
50        })?;
51        let proof = controller
52            .capture_catalog_bootstrap_proof()
53            .ok_or_else(|| {
54                DbError::Pipeline(
55                    "[LDB-6043] cluster catalog bootstrap requires the active durable leader lease"
56                        .into(),
57                )
58            })?;
59        self.recheck_unsealed_bootstrap(&proof)?;
60        let candidate = self.isolated_topology_catalog()?;
61        for (sql, statement, name, _) in &parsed {
62            let result = CATALOG_BOOTSTRAP
63                .scope((), candidate.execute_parsed_single(sql, statement).boxed())
64                .await?;
65            if !matches!(result, ExecuteResult::Ddl(ref info) if info.applied && info.object_name == *name)
66            {
67                return Err(DbError::Pipeline(format!(
68                    "candidate definition '{name}' did not apply exactly once"
69                )));
70            }
71            self.recheck_unsealed_bootstrap(&proof)?;
72        }
73        // Reuse the startup compiler, including exact delivery, topology and mutation admission.
74        let complete_connector_graph = {
75            let manager = candidate.connector_manager.lock();
76            manager.sources().len() == candidate.catalog.list_sources().len()
77                && manager
78                    .sinks()
79                    .values()
80                    .all(|sink| sink.connector_type.is_some())
81        };
82        if complete_connector_graph {
83            candidate.plan_topology_graph().await?;
84        } else if candidate
85            .connector_manager
86            .lock()
87            .sinks()
88            .values()
89            .any(|sink| sink.connector_type.is_some())
90        {
91            return Err(DbError::Unsupported("external sink preparation requires a complete admitted connector graph; catalog ingress sources cannot prove external delivery".into()));
92        }
93        self.recheck_unsealed_bootstrap(&proof)?;
94        self.prepare_bootstrap_sinks(&candidate, &proof).await?;
95        let manifest = CatalogManifest::new(candidate.catalog_manifest_inventory()?)
96            .map_err(|error| DbError::Pipeline(format!("invalid resolved catalog: {error}")))?;
97        #[cfg(test)]
98        let gate = { self.catalog_seal_gate.lock().clone() };
99        #[cfg(test)]
100        if let Some((entered, release)) = gate {
101            entered.notify_one();
102            release.notified().await;
103        }
104        self.recheck_unsealed_bootstrap(&proof)?;
105        store
106            .seal(&manifest, &proof)
107            .boxed()
108            .await
109            .map_err(|error| DbError::Pipeline(format!("catalog manifest seal failed: {error}")))?;
110        self.recheck_unsealed_bootstrap(&proof)?;
111        let _catalog = self.topology_ddl_lock.write().await;
112        self.recheck_unsealed_bootstrap(&proof)?;
113        self.install_bootstrap(&manifest, &parsed).await
114    }
115
116    fn ensure_bootstrap_available(&self) -> Result<(), DbError> {
117        self.ensure_catalog_cleanup_unfenced("cluster catalog bootstrap")?;
118        self.ensure_coordinated_recovery_mutation_unfenced("cluster catalog bootstrap")?;
119        if self.shutdown.load(Ordering::Acquire) {
120            return Err(DbError::Shutdown);
121        }
122        if DbState::load(&self.state) != DbState::Created {
123            return Err(DbError::InvalidOperation(
124                "cluster catalog bootstrap is only valid before pipeline startup".into(),
125            ));
126        }
127        Ok(())
128    }
129
130    fn recheck_unsealed_bootstrap(&self, proof: &LeaderProof) -> Result<(), DbError> {
131        self.ensure_bootstrap_available()?;
132        self.validate_catalog_seal_authority(Some(proof))?;
133        if !self.catalog_manifest_inventory()?.is_empty() {
134            return Err(DbError::Pipeline(
135                "cannot seal a new cluster catalog over uncommitted local topology".into(),
136            ));
137        }
138        Ok(())
139    }
140
141    async fn prepare_bootstrap_sinks(
142        &self,
143        candidate: &LaminarDB,
144        proof: &LeaderProof,
145    ) -> Result<(), DbError> {
146        let registrations = candidate.connector_manager.lock().sinks().clone();
147        let mut names: Vec<_> = registrations.keys().collect();
148        names.sort_unstable();
149        for name in names {
150            self.recheck_unsealed_bootstrap(proof)?;
151            let registration = &registrations[name];
152            let Some(resolved_binding) = &registration.schema_binding else {
153                continue;
154            };
155            let mut binding = resolved_binding.clone();
156            let mut config = crate::connector_manager::build_sink_config(
157                registration,
158                candidate.config.delivery_guarantee,
159            )?;
160            config.set(
161                "_arrow_schema",
162                crate::pipeline_callback::encode_arrow_schema(&Arc::new(binding.logical.clone())),
163            );
164            let mut sink = candidate.connector_registry.create_sink(&config, None)?;
165            tokio::time::timeout(
166                std::time::Duration::from_secs(30),
167                sink.prepare_schema(&config, &mut binding),
168            )
169            .await
170            .map_err(|_| {
171                DbError::Connector(format!(
172                    "sink '{name}' schema preparation exceeded 30 seconds"
173                ))
174            })??;
175            self.recheck_unsealed_bootstrap(proof)?;
176            laminar_connectors::schema::resolution::validate_prepared_writer(
177                &config,
178                resolved_binding,
179                &binding,
180            )
181            .map_err(|error| {
182                DbError::Connector(format!("sink '{name}' schema preparation: {error}"))
183            })?;
184            candidate
185                .connector_manager
186                .lock()
187                .set_schema_binding(name, binding)?;
188        }
189        Ok(())
190    }
191
192    async fn install_bootstrap(
193        &self,
194        manifest: &CatalogManifest,
195        parsed: &[BootstrapEntry],
196    ) -> Result<Vec<ExecuteResult>, DbError> {
197        let mut guard = CatalogBootstrapGuard {
198            db: self,
199            created: Vec::with_capacity(parsed.len()),
200            sealed: false,
201        };
202        let mut results = Vec::with_capacity(parsed.len());
203        for ((sql, statement, name, kind), entry) in parsed.iter().zip(&manifest.entries) {
204            let result = CATALOG_BOOTSTRAP
205                .scope(
206                    (),
207                    crate::ddl::schema_resolution::RESOLVED_SCHEMA.scope(
208                        entry.schema_binding.clone(),
209                        self.execute_parsed_single(sql, statement).boxed(),
210                    ),
211                )
212                .await?;
213            if !matches!(&result, ExecuteResult::Ddl(info) if info.applied && info.object_name == *name)
214            {
215                return Err(DbError::Pipeline(format!(
216                    "cluster catalog create '{name}' did not apply exactly once"
217                )));
218            }
219            guard.record(name.clone(), *kind);
220            results.push(result);
221        }
222        if self.reconcile_catalog_manifest_inventory(manifest)? != manifest.entries {
223            return Err(DbError::Pipeline(
224                "installed catalog differs from its sealed schema contracts".into(),
225            ));
226        }
227        guard.sealed();
228        Ok(results)
229    }
230}
231
232fn parse_bootstrap(db: &LaminarDB, sql: &[String]) -> Result<Vec<BootstrapEntry>, DbError> {
233    if sql.len() > 4096
234        || sql
235            .iter()
236            .map(String::len)
237            .try_fold(0_usize, usize::checked_add)
238            .is_none_or(|size| size > 8 * 1024 * 1024)
239    {
240        return Err(DbError::InvalidOperation(
241            "catalog bootstrap exceeds 4096 entries or 8 MiB of original SQL".into(),
242        ));
243    }
244    let mut entries = Vec::new();
245    let mut names = HashSet::new();
246    for batch in sql {
247        for sql in crate::sql_utils::split_statements(batch) {
248            let mut statements = parse_streaming_sql(sql)?;
249            if statements.len() != 1 {
250                return Err(DbError::InvalidOperation(
251                    "cluster bootstrap entries must contain exactly one SQL statement".into(),
252                ));
253            }
254            let statement = statements
255                .pop()
256                .ok_or_else(|| DbError::InvalidOperation("empty bootstrap statement".into()))?;
257            let (name, kind, _) = validate_cluster_catalog_create(db, sql, &statement)?;
258            if !names.insert(name.clone()) {
259                return Err(DbError::InvalidOperation(format!(
260                    "cluster bootstrap defines '{name}' more than once"
261                )));
262            }
263            if entries.len() >= 4096 {
264                return Err(DbError::InvalidOperation(
265                    "catalog bootstrap exceeds 4096 parsed entries".into(),
266                ));
267            }
268            entries.push((sql.to_owned(), statement, name, kind));
269        }
270    }
271    Ok(entries)
272}
273
274impl LaminarDB {
275    pub(crate) fn preflight_cluster_catalog_mutation(
276        &self,
277        sql: &str,
278        statement: &StreamingStatement,
279    ) -> Result<(), DbError> {
280        if !super::is_topology_ddl(statement) || super::catalog_manifest_replay_active() {
281            return Ok(());
282        }
283        let store_configured = self.catalog_manifest_store.lock().is_some();
284        if !store_configured && !self.is_cluster_runtime() {
285            return Ok(());
286        }
287        validate_cluster_catalog_create(self, sql, statement)?;
288        if !store_configured && !super::catalog_bootstrap_active() {
289            return Err(DbError::Pipeline(
290                "cluster topology DDL requires a catalog manifest store".into(),
291            ));
292        }
293        if !super::catalog_bootstrap_active() {
294            return Err(DbError::Pipeline(
295                "[LDB-6043] configured cluster topology can change only through startup bootstrap/replay until a replicated topology-version barrier is implemented".into(),
296            ));
297        }
298        Ok(())
299    }
300}
laminar_db/db/catalog_bootstrap.rs

laminar_db/db/
catalog_bootstrap.rs
