laminar_db/
builder.rs

1//! Fluent builder for `LaminarDB` construction.
2#![allow(clippy::disallowed_types)] // cold path
3
4use std::collections::HashMap;
5use std::path::PathBuf;
6use std::sync::Arc;
7
8use datafusion_expr::{AggregateUDF, ScalarUDF};
9use laminar_core::streaming::{BackpressureStrategy, StreamCheckpointConfig};
10
11use crate::config::LaminarConfig;
12use crate::db::{LaminarDB, RuntimeMode};
13use crate::error::DbError;
14use crate::profile::Profile;
15
16/// Callback for registering custom connectors.
17type ConnectorCallback = Box<
18    dyn FnOnce(
19            &laminar_connectors::registry::ConnectorRegistry,
20        ) -> Result<(), laminar_connectors::error::ConnectorError>
21        + Send,
22>;
23
24/// Fluent builder for constructing a [`LaminarDB`] instance.
25///
26/// # Example
27///
28/// ```rust,ignore
29/// let db = LaminarDB::builder()
30///     .config_var("KAFKA_BROKERS", "localhost:9092")
31///     .buffer_size(131072)
32///     .build()
33///     .await?;
34/// ```
35pub struct LaminarDbBuilder {
36    config: LaminarConfig,
37    config_vars: HashMap<String, String>,
38    connector_callbacks: Vec<ConnectorCallback>,
39    profile: Profile,
40    profile_explicit: bool,
41    delivery_explicit: bool,
42    object_store_url: Option<String>,
43    object_store_options: HashMap<String, String>,
44    custom_udfs: Vec<ScalarUDF>,
45    custom_udafs: Vec<AggregateUDF>,
46    #[cfg(feature = "cluster")]
47    cluster_controller: Option<std::sync::Arc<laminar_core::cluster::control::ClusterController>>,
48    #[cfg(feature = "cluster")]
49    shuffle_sender: Option<std::sync::Arc<laminar_core::shuffle::ShuffleSender>>,
50    #[cfg(feature = "cluster")]
51    shuffle_receiver: Option<std::sync::Arc<laminar_core::shuffle::ShuffleReceiver>>,
52    #[cfg(feature = "cluster")]
53    decision_store: Option<std::sync::Arc<laminar_core::cluster::control::CheckpointDecisionStore>>,
54    #[cfg(feature = "cluster")]
55    assignment_snapshot_store:
56        Option<std::sync::Arc<laminar_core::cluster::control::AssignmentSnapshotStore>>,
57    #[cfg(feature = "cluster")]
58    catalog_manifest_store:
59        Option<std::sync::Arc<laminar_core::cluster::control::CatalogManifestStore>>,
60    #[cfg(feature = "cluster")]
61    verified_cluster_namespaces: Option<laminar_core::cluster::control::VerifiedClusterNamespaces>,
62    #[cfg(all(feature = "cluster", test))]
63    unchecked_cluster_checkpoint_store: Option<Arc<dyn object_store::ObjectStore>>,
64    state_backend: Option<std::sync::Arc<dyn laminar_core::state::StateBackend>>,
65    vnode_registry: Option<std::sync::Arc<laminar_core::state::VnodeRegistry>>,
66    physical_optimizer_rules: Vec<
67        std::sync::Arc<dyn datafusion::physical_optimizer::PhysicalOptimizerRule + Send + Sync>,
68    >,
69    target_partitions: Option<usize>,
70    ai_runtime: Option<std::sync::Arc<crate::ai::AiRuntime>>,
71}
72
73impl LaminarDbBuilder {
74    /// Create a new builder with default settings.
75    #[must_use]
76    pub fn new() -> Self {
77        Self {
78            config: LaminarConfig::default(),
79            config_vars: HashMap::new(),
80            connector_callbacks: Vec::new(),
81            profile: Profile::default(),
82            profile_explicit: false,
83            delivery_explicit: false,
84            object_store_url: None,
85            object_store_options: HashMap::new(),
86            custom_udfs: Vec::new(),
87            custom_udafs: Vec::new(),
88            #[cfg(feature = "cluster")]
89            cluster_controller: None,
90            #[cfg(feature = "cluster")]
91            shuffle_sender: None,
92            #[cfg(feature = "cluster")]
93            shuffle_receiver: None,
94            #[cfg(feature = "cluster")]
95            decision_store: None,
96            #[cfg(feature = "cluster")]
97            assignment_snapshot_store: None,
98            #[cfg(feature = "cluster")]
99            catalog_manifest_store: None,
100            #[cfg(feature = "cluster")]
101            verified_cluster_namespaces: None,
102            #[cfg(all(feature = "cluster", test))]
103            unchecked_cluster_checkpoint_store: None,
104            state_backend: None,
105            vnode_registry: None,
106            physical_optimizer_rules: Vec::new(),
107            target_partitions: None,
108            ai_runtime: None,
109        }
110    }
111
112    /// Install the AI subsystem; required for `ai_*` SQL functions.
113    #[must_use]
114    pub fn ai(mut self, runtime: std::sync::Arc<crate::ai::AiRuntime>) -> Self {
115        self.ai_runtime = Some(runtime);
116        self
117    }
118
119    /// Override `target_partitions`.
120    #[must_use]
121    pub fn target_partitions(mut self, n: usize) -> Self {
122        self.target_partitions = Some(n);
123        self
124    }
125
126    /// Register an additional `PhysicalOptimizerRule` on the session state.
127    #[must_use]
128    pub fn physical_optimizer_rule(
129        mut self,
130        rule: std::sync::Arc<
131            dyn datafusion::physical_optimizer::PhysicalOptimizerRule + Send + Sync,
132        >,
133    ) -> Self {
134        self.physical_optimizer_rules.push(rule);
135        self
136    }
137
138    /// Install a state backend; must be paired with [`Self::vnode_registry`].
139    #[must_use]
140    pub fn state_backend(
141        mut self,
142        backend: std::sync::Arc<dyn laminar_core::state::StateBackend>,
143    ) -> Self {
144        self.state_backend = Some(backend);
145        self
146    }
147
148    /// Install a vnode registry; must be paired with [`Self::state_backend`].
149    #[must_use]
150    pub fn vnode_registry(
151        mut self,
152        registry: std::sync::Arc<laminar_core::state::VnodeRegistry>,
153    ) -> Self {
154        self.vnode_registry = Some(registry);
155        self
156    }
157
158    /// Install the cluster control facade; selects cluster runtime semantics when the profile is
159    /// inferred. An explicitly selected profile must be [`Profile::Cluster`].
160    #[cfg(feature = "cluster")]
161    #[must_use]
162    pub fn cluster_controller(
163        mut self,
164        controller: std::sync::Arc<laminar_core::cluster::control::ClusterController>,
165    ) -> Self {
166        self.cluster_controller = Some(controller);
167        self
168    }
169
170    /// Install the outbound shuffle handle; pair with [`Self::shuffle_receiver`].
171    #[cfg(feature = "cluster")]
172    #[must_use]
173    pub fn shuffle_sender(
174        mut self,
175        sender: std::sync::Arc<laminar_core::shuffle::ShuffleSender>,
176    ) -> Self {
177        self.shuffle_sender = Some(sender);
178        self
179    }
180
181    /// Install the inbound shuffle handle; pair with [`Self::shuffle_sender`].
182    #[cfg(feature = "cluster")]
183    #[must_use]
184    pub fn shuffle_receiver(
185        mut self,
186        receiver: std::sync::Arc<laminar_core::shuffle::ShuffleReceiver>,
187    ) -> Self {
188        self.shuffle_receiver = Some(receiver);
189        self
190    }
191
192    /// Install the commit-marker store for cross-instance 2PC.
193    #[cfg(feature = "cluster")]
194    #[must_use]
195    pub fn decision_store(
196        mut self,
197        store: std::sync::Arc<laminar_core::cluster::control::CheckpointDecisionStore>,
198    ) -> Self {
199        self.decision_store = Some(store);
200        self
201    }
202
203    /// Install the assignment-snapshot store for dynamic rebalance.
204    #[cfg(feature = "cluster")]
205    #[must_use]
206    pub fn assignment_snapshot_store(
207        mut self,
208        store: std::sync::Arc<laminar_core::cluster::control::AssignmentSnapshotStore>,
209    ) -> Self {
210        self.assignment_snapshot_store = Some(store);
211        self
212    }
213
214    /// Install the catalog-manifest store for cluster-wide DDL replay.
215    #[cfg(feature = "cluster")]
216    #[must_use]
217    pub fn catalog_manifest_store(
218        mut self,
219        store: std::sync::Arc<laminar_core::cluster::control::CatalogManifestStore>,
220    ) -> Self {
221        self.catalog_manifest_store = Some(store);
222        self
223    }
224
225    /// Install the exact checkpoint handle admitted by the cluster namespace proof.
226    #[cfg(feature = "cluster")]
227    #[must_use]
228    pub fn verified_cluster_namespaces(
229        mut self,
230        namespaces: laminar_core::cluster::control::VerifiedClusterNamespaces,
231    ) -> Self {
232        self.verified_cluster_namespaces = Some(namespaces);
233        self
234    }
235
236    /// Install an unchecked checkpoint store in crate unit tests.
237    #[cfg(all(feature = "cluster", test))]
238    #[must_use]
239    pub(crate) fn cluster_checkpoint_object_store(
240        mut self,
241        store: Arc<dyn object_store::ObjectStore>,
242    ) -> Self {
243        self.unchecked_cluster_checkpoint_store = Some(store);
244        self
245    }
246
247    /// Set a config variable for `${VAR}` substitution in SQL.
248    #[must_use]
249    pub fn config_var(mut self, key: &str, value: &str) -> Self {
250        self.config_vars.insert(key.to_string(), value.to_string());
251        self
252    }
253
254    /// Set the bearer token presented when forwarding requests to the cluster
255    /// leader's HTTP API.
256    #[must_use]
257    pub fn http_auth_token(mut self, token: impl Into<String>) -> Self {
258        self.config.http_auth_token = Some(crate::config::SecretString::new(token));
259        self
260    }
261
262    /// Set the default buffer size for streaming channels.
263    #[must_use]
264    pub fn buffer_size(mut self, size: usize) -> Self {
265        self.config.default_buffer_size = size;
266        self
267    }
268
269    /// Set the default backpressure strategy.
270    #[must_use]
271    pub fn backpressure(mut self, strategy: BackpressureStrategy) -> Self {
272        self.config.default_backpressure = strategy;
273        self
274    }
275
276    /// Set the storage directory for WAL and checkpoints.
277    #[must_use]
278    pub fn storage_dir(mut self, path: impl Into<PathBuf>) -> Self {
279        self.config.storage_dir = Some(path.into());
280        self
281    }
282
283    /// Set checkpoint configuration.
284    #[must_use]
285    pub fn checkpoint(mut self, config: StreamCheckpointConfig) -> Self {
286        self.config.checkpoint = Some(config);
287        self
288    }
289
290    /// Select dirty-only changelog emission for keyed running aggregates.
291    #[must_use]
292    pub fn incremental_emit(mut self, enabled: bool) -> Self {
293        self.config.incremental_emit = enabled;
294        self
295    }
296
297    /// Set the deployment profile.
298    ///
299    /// See [`Profile`] for the available tiers. [`Profile::Cluster`] also requires a cluster
300    /// controller at build time.
301    #[must_use]
302    pub fn profile(mut self, profile: Profile) -> Self {
303        self.profile = profile;
304        self.profile_explicit = true;
305        self
306    }
307
308    /// Set the object-store URL for durable checkpoints.
309    ///
310    /// Required when using [`Profile::Durable`] or
311    /// [`Profile::Cluster`].
312    #[must_use]
313    pub fn object_store_url(mut self, url: impl Into<String>) -> Self {
314        self.object_store_url = Some(url.into());
315        self
316    }
317
318    /// Set explicit credential/config overrides for the object store.
319    ///
320    /// Keys are backend-specific (e.g., `aws_access_key_id`, `aws_region`).
321    /// These supplement environment-variable-based credential resolution.
322    #[must_use]
323    pub fn object_store_options(mut self, opts: HashMap<String, String>) -> Self {
324        self.object_store_options = opts;
325        self
326    }
327
328    /// Set the end-to-end delivery guarantee for the pipeline.
329    #[must_use]
330    pub fn delivery_guarantee(
331        mut self,
332        guarantee: laminar_connectors::connector::DeliveryGuarantee,
333    ) -> Self {
334        self.config.delivery_guarantee = guarantee;
335        self.delivery_explicit = true;
336        self
337    }
338
339    /// Register a custom scalar UDF; available in SQL after `build()`.
340    #[must_use]
341    pub fn register_udf(mut self, udf: ScalarUDF) -> Self {
342        self.custom_udfs.push(udf);
343        self
344    }
345
346    /// Register a custom aggregate UDF; available in SQL after `build()`.
347    #[must_use]
348    pub fn register_udaf(mut self, udaf: AggregateUDF) -> Self {
349        self.custom_udafs.push(udaf);
350        self
351    }
352
353    /// Source → coordinator channel capacity (default 64).
354    #[must_use]
355    pub fn pipeline_channel_capacity(mut self, capacity: usize) -> Self {
356        self.config.pipeline_channel_capacity = Some(capacity);
357        self
358    }
359
360    /// Micro-batch coalescing window (default 5ms for connectors, 0 for embedded).
361    #[must_use]
362    pub fn pipeline_batch_window(mut self, window: std::time::Duration) -> Self {
363        self.config.pipeline_batch_window = Some(window);
364        self
365    }
366
367    /// Max drain time per cycle in nanoseconds (default 1ms).
368    #[must_use]
369    pub fn pipeline_drain_budget_ns(mut self, ns: u64) -> Self {
370        self.config.pipeline_drain_budget_ns = Some(ns);
371        self
372    }
373
374    /// Per-query execution budget in nanoseconds (default 8ms).
375    #[must_use]
376    pub fn pipeline_query_budget_ns(mut self, ns: u64) -> Self {
377        self.config.pipeline_query_budget_ns = Some(ns);
378        self
379    }
380
381    /// Per-port operator input-buffer cap in batches (default 256).
382    #[must_use]
383    pub fn pipeline_max_input_buf_batches(mut self, batches: usize) -> Self {
384        self.config.pipeline_max_input_buf_batches = Some(batches);
385        self
386    }
387
388    /// Per-port operator input-buffer cap in bytes.
389    #[must_use]
390    pub fn pipeline_max_input_buf_bytes(mut self, bytes: usize) -> Self {
391        self.config.pipeline_max_input_buf_bytes = Some(bytes);
392        self
393    }
394
395    /// Backpressure policy (default `Backpressure`).
396    #[must_use]
397    pub fn pipeline_backpressure_policy(
398        mut self,
399        policy: crate::config::BackpressurePolicy,
400    ) -> Self {
401        self.config.pipeline_backpressure_policy = policy;
402        self
403    }
404
405    /// Auto-restart policy used when supervision is enabled.
406    #[must_use]
407    pub fn restart_policy(mut self, policy: crate::config::RestartPolicy) -> Self {
408        self.config.restart_policy = policy;
409        self
410    }
411
412    /// Register custom connectors; the callback runs after built-ins are wired and must propagate
413    /// registration errors. The registry is frozen before [`Self::build`] returns.
414    #[must_use]
415    pub fn register_connector(
416        mut self,
417        f: impl FnOnce(
418                &laminar_connectors::registry::ConnectorRegistry,
419            ) -> Result<(), laminar_connectors::error::ConnectorError>
420            + Send
421            + 'static,
422    ) -> Self {
423        self.connector_callbacks.push(Box::new(f));
424        self
425    }
426
427    /// Build the `LaminarDB` instance.
428    ///
429    /// # Errors
430    ///
431    /// Returns `DbError` if database creation fails.
432    #[allow(clippy::unused_async)]
433    pub async fn build(mut self) -> Result<Arc<LaminarDB>, DbError> {
434        let has_checkpoint_data_dir = self
435            .config
436            .checkpoint
437            .as_ref()
438            .and_then(|checkpoint| checkpoint.data_dir.as_ref())
439            .is_some();
440        if self.object_store_url.is_some() && has_checkpoint_data_dir {
441            return Err(DbError::Config(
442                "object_store_url conflicts with checkpoint.data_dir; configure exactly one checkpoint store"
443                    .into(),
444            ));
445        }
446        #[cfg(feature = "cluster")]
447        if self.has_cluster_checkpoint_store() && self.object_store_url.is_some() {
448            return Err(DbError::Config(
449                "verified cluster namespaces conflict with object_store_url; configure exactly one checkpoint object store"
450                    .into(),
451            ));
452        }
453        #[cfg(feature = "cluster")]
454        if self.has_cluster_checkpoint_store() && has_checkpoint_data_dir {
455            return Err(DbError::Config(
456                "verified cluster namespaces conflict with checkpoint.data_dir; configure exactly one checkpoint store"
457                    .into(),
458            ));
459        }
460
461        self.config.object_store_url = self.object_store_url.take();
462        self.config.object_store_options = std::mem::take(&mut self.object_store_options);
463        if let Some(url) = self
464            .config
465            .object_store_url
466            .as_deref()
467            .filter(|url| url.starts_with("file://"))
468        {
469            laminar_core::storage::object_store_builder::file_url_path(url)
470                .map_err(|error| DbError::Config(format!("checkpoint storage URL: {error}")))?;
471        }
472
473        if !self.profile_explicit {
474            self.profile = Profile::from_config(&self.config, false);
475        }
476
477        let runtime_mode = self.resolve_runtime_mode()?;
478        if runtime_mode.is_cluster() && !self.profile_explicit {
479            self.profile = Profile::Cluster;
480        }
481        if runtime_mode.is_cluster() && !self.delivery_explicit {
482            self.config.delivery_guarantee =
483                laminar_connectors::connector::DeliveryGuarantee::AtLeastOnce;
484        }
485
486        self.profile
487            .validate_features()
488            .map_err(|e| DbError::Config(e.to_string()))?;
489        Self::validate_cluster_delivery(runtime_mode, self.config.delivery_guarantee)?;
490        #[cfg(feature = "cluster")]
491        if runtime_mode == RuntimeMode::Cluster {
492            if self.config.object_store_url.is_some() || has_checkpoint_data_dir {
493                return Err(DbError::Config(
494                    "[LDB-0011] cluster checkpoint storage must be supplied by a successful shared namespace proof; object_store_url and checkpoint.data_dir are not cluster authorities"
495                        .into(),
496                ));
497            }
498            if !self.has_cluster_checkpoint_store() {
499                return Err(DbError::Config(
500                    "[LDB-0011] cluster runtime requires VerifiedClusterNamespaces from the shared checkpoint/state namespace proof"
501                        .into(),
502                ));
503            }
504            if let Some(namespaces) = &self.verified_cluster_namespaces {
505                let controller = self.cluster_controller.as_ref().ok_or_else(|| {
506                    DbError::Config("cluster namespace proof requires a cluster controller".into())
507                })?;
508                let expected = laminar_core::checkpoint::CheckpointParticipant {
509                    node_id: controller.instance_id().0,
510                    boot_incarnation: controller.recovery_incarnation(),
511                };
512                if namespaces.local_participant() != expected {
513                    let proved = namespaces.local_participant();
514                    return Err(DbError::Config(format!(
515                        "[LDB-0011] verified cluster namespaces belong to node {} boot {}, but the controller is node {} boot {}",
516                        proved.node_id,
517                        proved.boot_incarnation,
518                        expected.node_id,
519                        expected.boot_incarnation,
520                    )));
521                }
522                let state_backend = self.state_backend.as_ref().ok_or_else(|| {
523                    DbError::Config(
524                        "[LDB-0011] cluster runtime requires a ClusterShared state_backend built from the verified state namespace"
525                            .into(),
526                    )
527                })?;
528                let durability = state_backend.durability_scope();
529                if durability != laminar_core::state::StateBackendDurability::ClusterShared {
530                    return Err(DbError::Config(format!(
531                        "[LDB-0011] cluster state_backend must be ClusterShared, but is {durability:?}"
532                    )));
533                }
534                let verified_state_store = namespaces.state_store();
535                if !state_backend.uses_exact_object_store(&verified_state_store) {
536                    return Err(DbError::Config(
537                        "[LDB-0011] cluster state_backend does not use the exact object-store handle admitted by the namespace proof"
538                            .into(),
539                    ));
540                }
541            }
542        }
543        #[cfg(feature = "cluster")]
544        let profile_object_store = self.config.object_store_url.as_deref().or_else(|| {
545            self.has_cluster_checkpoint_store()
546                .then_some("verified-cluster-checkpoint-store")
547        });
548        #[cfg(not(feature = "cluster"))]
549        let profile_object_store = self.config.object_store_url.as_deref();
550        self.profile
551            .validate_config(&self.config, profile_object_store)
552            .map_err(|e| DbError::Config(e.to_string()))?;
553
554        if runtime_mode == RuntimeMode::Local
555            && self.config.delivery_guarantee
556                != laminar_connectors::connector::DeliveryGuarantee::BestEffort
557            && self
558                .config
559                .object_store_url
560                .as_deref()
561                .is_some_and(|url| !url.starts_with("file://"))
562        {
563            return Err(DbError::Config(
564                "[LDB-0014] a local replay-capable deployment with a shared cloud checkpoint namespace is not admitted until its writer lease is term-fenced; use a built-in or file:// local checkpoint directory, or best_effort delivery"
565                    .into(),
566            ));
567        }
568
569        Self::validate_backpressure(&self.config)?;
570        self.validate_state_topology(runtime_mode)?;
571        #[cfg(feature = "cluster")]
572        self.bind_cluster_process_lease(runtime_mode)?;
573
574        self.profile.apply_defaults(&mut self.config);
575
576        #[cfg(feature = "cluster")]
577        let cluster_checkpoint_store = self.cluster_checkpoint_store();
578        let mut db = LaminarDB::open_with_config_and_vars_and_rules(
579            self.config,
580            self.config_vars,
581            &self.physical_optimizer_rules,
582            self.target_partitions,
583            runtime_mode,
584        )?;
585        if let Some(runtime) = self.ai_runtime {
586            let handle = tokio::runtime::Handle::try_current().map_err(|_| {
587                DbError::InvalidOperation(
588                    "LaminarDB::build() with an AI runtime must run inside a Tokio runtime"
589                        .to_string(),
590                )
591            })?;
592            db.set_ai_runtime(runtime, handle);
593        }
594        for callback in self.connector_callbacks {
595            callback(db.connector_registry())?;
596        }
597        db.connector_registry().freeze();
598        for udf in self.custom_udfs {
599            db.register_custom_udf(udf);
600        }
601        for udaf in self.custom_udafs {
602            db.register_custom_udaf(udaf);
603        }
604        #[cfg(feature = "cluster")]
605        if let Some(controller) = self.cluster_controller {
606            db.set_cluster_controller(controller)?;
607        }
608        #[cfg(feature = "cluster")]
609        if let Some(store) = cluster_checkpoint_store {
610            db.set_cluster_checkpoint_object_store(store)?;
611        }
612        #[cfg(feature = "cluster")]
613        if let Some(sender) = self.shuffle_sender {
614            db.set_shuffle_sender(sender);
615        }
616        #[cfg(feature = "cluster")]
617        if let Some(receiver) = self.shuffle_receiver {
618            db.set_shuffle_receiver(receiver);
619        }
620        #[cfg(feature = "cluster")]
621        if let Some(store) = self.decision_store {
622            db.set_decision_store(store);
623        }
624        #[cfg(feature = "cluster")]
625        if let Some(store) = self.assignment_snapshot_store {
626            db.set_assignment_snapshot_store(store);
627        }
628        #[cfg(feature = "cluster")]
629        if let Some(store) = self.catalog_manifest_store {
630            db.set_catalog_manifest_store(store);
631        }
632        if let Some(backend) = self.state_backend {
633            db.set_state_backend(backend);
634        }
635        if let Some(registry) = self.vnode_registry {
636            db.set_vnode_registry(registry);
637        }
638        Ok(Arc::new(db))
639    }
640
641    fn validate_state_topology(&self, runtime_mode: RuntimeMode) -> Result<(), DbError> {
642        match (&self.state_backend, &self.vnode_registry) {
643            (Some(_), None) => {
644                return Err(DbError::Config(
645                    "state_backend is set but vnode_registry is missing".into(),
646                ));
647            }
648            (None, Some(_)) => {
649                return Err(DbError::Config(
650                    "vnode_registry is set but state_backend is missing".into(),
651                ));
652            }
653            _ => {}
654        }
655        if let (Some(backend), Some(registry)) =
656            (self.state_backend.as_ref(), self.vnode_registry.as_ref())
657        {
658            let backend_capacity = backend.key_group_capacity();
659            let backend_key_groups =
660                laminar_core::state::KeyGroupCount::try_from(backend_capacity).map_err(|_| {
661                    DbError::Config(format!(
662                        "state_backend key-group capacity must be between 1 and {}, got {backend_capacity}",
663                        laminar_core::state::MAX_KEY_GROUP_COUNT
664                    ))
665                })?;
666            let registry_count = registry.vnode_count();
667            let registry_key_groups = laminar_core::state::KeyGroupCount::try_from(registry_count)
668                .map_err(|_| {
669                    DbError::Config(format!(
670                        "vnode_registry count must be between 1 and {}, got {registry_count}",
671                        laminar_core::state::MAX_KEY_GROUP_COUNT
672                    ))
673                })?;
674            if backend_key_groups != registry_key_groups {
675                return Err(DbError::Config(format!(
676                    "state_backend key-group capacity {backend_key_groups} does not match vnode_registry count {registry_key_groups}"
677                )));
678            }
679            if !runtime_mode.is_cluster()
680                && registry_key_groups != laminar_core::state::LOCAL_KEY_GROUP_COUNT
681            {
682                return Err(DbError::Config(format!(
683                    "local runtime requires exactly {} key group; got {registry_key_groups}. Multi-key-group ownership requires cluster mode",
684                    laminar_core::state::LOCAL_KEY_GROUP_COUNT
685                )));
686            }
687        }
688        Ok(())
689    }
690
691    #[cfg(feature = "cluster")]
692    fn bind_cluster_process_lease(&self, runtime_mode: RuntimeMode) -> Result<(), DbError> {
693        if !runtime_mode.is_cluster() {
694            return Ok(());
695        }
696        let controller = self.cluster_controller.as_ref().ok_or_else(|| {
697            DbError::Config("cluster runtime requires a cluster controller".into())
698        })?;
699        let deadline = controller.process_lease_deadline().ok_or_else(|| {
700            DbError::Config(
701                "cluster runtime requires one shared process lease deadline before construction"
702                    .into(),
703            )
704        })?;
705        if !deadline.is_live() {
706            return Err(DbError::Config(
707                "cluster runtime process lease deadline is already expired".into(),
708            ));
709        }
710        match (&self.shuffle_sender, &self.shuffle_receiver) {
711            (Some(sender), Some(receiver)) => {
712                sender
713                    .bind_process_lease_deadline_pair(receiver, deadline)
714                    .map_err(|error| {
715                        DbError::Config(format!(
716                            "shuffle process lease does not match the controller: {error}"
717                        ))
718                    })?;
719            }
720            (None, None) => {}
721            _ => {
722                return Err(DbError::Config(
723                    "cluster shuffle sender and receiver must be installed together".into(),
724                ));
725            }
726        }
727        Ok(())
728    }
729
730    fn validate_backpressure(config: &LaminarConfig) -> Result<(), DbError> {
731        use crate::config::BackpressurePolicy;
732        use laminar_connectors::connector::DeliveryGuarantee;
733
734        let policy = config.pipeline_backpressure_policy;
735        if policy == BackpressurePolicy::Backpressure {
736            return Ok(());
737        }
738
739        let has_count_cap = config.pipeline_max_input_buf_batches.is_none_or(|c| c > 0);
740        let has_byte_cap = config.pipeline_max_input_buf_bytes.is_some_and(|b| b > 0);
741        if !has_count_cap && !has_byte_cap {
742            return Err(DbError::Config(format!(
743                "backpressure_policy={policy:?} requires at least one of \
744                 pipeline_max_input_buf_batches (>0) or pipeline_max_input_buf_bytes"
745            )));
746        }
747
748        if policy == BackpressurePolicy::ShedOldest
749            && config.delivery_guarantee == DeliveryGuarantee::ExactlyOnce
750        {
751            return Err(DbError::Config(
752                "ShedOldest drops data; it is incompatible with exactly-once \
753                 delivery. Use Backpressure or Fail, or downgrade the guarantee."
754                    .into(),
755            ));
756        }
757        Ok(())
758    }
759
760    #[cfg(feature = "cluster")]
761    fn has_cluster_checkpoint_store(&self) -> bool {
762        let has_store = self.verified_cluster_namespaces.is_some();
763        #[cfg(test)]
764        let has_store = has_store || self.unchecked_cluster_checkpoint_store.is_some();
765        has_store
766    }
767
768    #[cfg(feature = "cluster")]
769    fn cluster_checkpoint_store(&self) -> Option<Arc<dyn object_store::ObjectStore>> {
770        let store = self
771            .verified_cluster_namespaces
772            .as_ref()
773            .map(laminar_core::cluster::control::VerifiedClusterNamespaces::checkpoint_store);
774        #[cfg(test)]
775        let store = store.or_else(|| self.unchecked_cluster_checkpoint_store.clone());
776        store
777    }
778
779    /// Resolve the distributed execution scope exactly once and reject partial or contradictory
780    /// cluster wiring before constructing the database.
781    fn resolve_runtime_mode(&self) -> Result<RuntimeMode, DbError> {
782        #[cfg(not(feature = "cluster"))]
783        {
784            if self.profile == Profile::Cluster {
785                return Err(DbError::Config(
786                    "Profile::Cluster requires a cluster controller and a cluster-enabled build"
787                        .into(),
788                ));
789            }
790            Ok(RuntimeMode::Local)
791        }
792
793        #[cfg(feature = "cluster")]
794        {
795            let has_controller = self.cluster_controller.is_some();
796            let has_cluster_only_handle = self.shuffle_sender.is_some()
797                || self.shuffle_receiver.is_some()
798                || self.decision_store.is_some()
799                || self.assignment_snapshot_store.is_some()
800                || self.catalog_manifest_store.is_some()
801                || self.has_cluster_checkpoint_store();
802
803            if has_cluster_only_handle && !has_controller {
804                return Err(DbError::Config(
805                    "cluster-only stores and shuffle handles require a cluster controller".into(),
806                ));
807            }
808            if self.profile == Profile::Cluster && !has_controller {
809                return Err(DbError::Config(
810                    "Profile::Cluster requires a cluster controller".into(),
811                ));
812            }
813            if self.profile_explicit && self.profile != Profile::Cluster && has_controller {
814                return Err(DbError::Config(format!(
815                    "profile {} cannot be combined with a cluster controller; select Profile::Cluster or omit the explicit profile",
816                    self.profile
817                )));
818            }
819
820            Ok(if has_controller {
821                RuntimeMode::Cluster
822            } else {
823                RuntimeMode::Local
824            })
825        }
826    }
827
828    /// Validate delivery semantics whose correctness depends on the runtime mode.
829    ///
830    /// Checkpoint decisions are term-fenced. Cluster exactly-once remains closed until supported
831    /// connectors also provide certified term-fenced source handoff and external sink cursor
832    /// commit across reassignment.
833    fn validate_cluster_delivery(
834        runtime_mode: RuntimeMode,
835        guarantee: laminar_connectors::connector::DeliveryGuarantee,
836    ) -> Result<(), DbError> {
837        use laminar_connectors::connector::DeliveryGuarantee;
838
839        if !runtime_mode.is_cluster() {
840            return Ok(());
841        }
842        match guarantee {
843            DeliveryGuarantee::BestEffort => Err(DbError::Config(
844                "cluster mode requires at_least_once delivery; best_effort has no defined \
845                 rebalance/state-loss contract"
846                    .into(),
847            )),
848            DeliveryGuarantee::AtLeastOnce => Ok(()),
849            DeliveryGuarantee::ExactlyOnce => Err(DbError::Config(
850                "[LDB-0013] cluster exactly-once is not admitted: checkpoint decisions are \
851                 term-fenced, but supported connectors do not yet provide a certified \
852                 term-fenced source handoff and external sink cursor commit. Use cluster \
853                 at_least_once, or exactly_once in embedded/single-node mode"
854                    .into(),
855            )),
856        }
857    }
858}
859
860impl Default for LaminarDbBuilder {
861    fn default() -> Self {
862        Self::new()
863    }
864}
865
866impl std::fmt::Debug for LaminarDbBuilder {
867    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
868        f.debug_struct("LaminarDbBuilder")
869            .field("config", &self.config)
870            .field("profile", &self.profile)
871            .field("profile_explicit", &self.profile_explicit)
872            .field("delivery_explicit", &self.delivery_explicit)
873            .field("object_store_url", &self.object_store_url)
874            .field(
875                "object_store_options_count",
876                &self.object_store_options.len(),
877            )
878            .field("config_vars_count", &self.config_vars.len())
879            .field("connector_callbacks", &self.connector_callbacks.len())
880            .field("custom_udfs", &self.custom_udfs.len())
881            .field("custom_udafs", &self.custom_udafs.len())
882            .finish_non_exhaustive()
883    }
884}
885
886#[cfg(test)]
887mod tests {
888    use super::*;
889
890    #[cfg(feature = "cluster")]
891    fn test_cluster_controller_without_deadline(
892    ) -> Arc<laminar_core::cluster::control::ClusterController> {
893        use laminar_core::cluster::control::{ClusterController, ClusterKv, InMemoryKv};
894        use laminar_core::cluster::discovery::{NodeId, NodeInfo};
895
896        let node = NodeId(1);
897        let kv: Arc<dyn ClusterKv> = Arc::new(InMemoryKv::new(node));
898        let (_members_tx, members_rx) = tokio::sync::watch::channel(Vec::<NodeInfo>::new());
899        Arc::new(ClusterController::new(node, kv, None, members_rx))
900    }
901
902    #[cfg(feature = "cluster")]
903    fn test_cluster_controller() -> Arc<laminar_core::cluster::control::ClusterController> {
904        use laminar_core::cluster::control::LeaseDeadline;
905
906        let controller = test_cluster_controller_without_deadline();
907        controller
908            .set_process_lease_deadline(Arc::new(LeaseDeadline::live_for(
909                std::time::Duration::from_secs(60),
910            )))
911            .unwrap();
912        controller
913    }
914
915    #[cfg(feature = "cluster")]
916    fn test_cluster_checkpoint_store() -> Arc<dyn object_store::ObjectStore> {
917        Arc::new(object_store::memory::InMemory::new())
918    }
919
920    #[cfg(feature = "cluster")]
921    async fn test_verified_cluster_namespaces(
922        controller: &laminar_core::cluster::control::ClusterController,
923        checkpoint_store: Arc<dyn object_store::ObjectStore>,
924    ) -> laminar_core::cluster::control::VerifiedClusterNamespaces {
925        use laminar_core::checkpoint::CheckpointParticipant;
926        use laminar_core::cluster::control::{
927            prove_shared_object_store_namespaces, ClusterKv, InMemoryKv,
928        };
929
930        let participant = CheckpointParticipant {
931            node_id: controller.instance_id().0,
932            boot_incarnation: controller.recovery_incarnation(),
933        };
934        let control: Arc<dyn ClusterKv> = Arc::new(InMemoryKv::new(controller.instance_id()));
935        prove_shared_object_store_namespaces(
936            participant,
937            &[participant],
938            control,
939            checkpoint_store,
940            test_cluster_checkpoint_store(),
941            std::time::Duration::from_secs(1),
942        )
943        .await
944        .unwrap()
945    }
946
947    #[cfg(feature = "cluster")]
948    fn test_verified_state_topology(
949        namespaces: &laminar_core::cluster::control::VerifiedClusterNamespaces,
950    ) -> (
951        Arc<dyn laminar_core::state::StateBackend>,
952        Arc<laminar_core::state::VnodeRegistry>,
953    ) {
954        let vnode_count = u32::from(laminar_core::state::DEFAULT_CLUSTER_KEY_GROUP_COUNT);
955        (
956            Arc::new(laminar_core::state::ObjectStoreBackend::cluster_shared(
957                namespaces.state_store(),
958                "test-node",
959                vnode_count,
960            )),
961            Arc::new(laminar_core::state::VnodeRegistry::new(vnode_count)),
962        )
963    }
964
965    #[tokio::test]
966    async fn test_default_builder() {
967        let db = LaminarDbBuilder::new().build().await.unwrap();
968        assert!(!db.is_closed());
969        assert!(!db.is_cluster_runtime());
970        #[cfg(feature = "cluster")]
971        assert!(!db.cluster_intake_fenced());
972    }
973
974    #[tokio::test]
975    async fn test_shed_oldest_requires_cap() {
976        use crate::config::BackpressurePolicy;
977        let err = LaminarDbBuilder::new()
978            .pipeline_backpressure_policy(BackpressurePolicy::ShedOldest)
979            .pipeline_max_input_buf_batches(0)
980            .build()
981            .await
982            .expect_err("ShedOldest with no caps must be rejected");
983        assert!(err.to_string().contains("requires at least one"), "{err}");
984    }
985
986    #[tokio::test]
987    async fn test_shed_oldest_rejects_exactly_once() {
988        use crate::config::BackpressurePolicy;
989        use laminar_connectors::connector::DeliveryGuarantee;
990        let err = LaminarDbBuilder::new()
991            .pipeline_backpressure_policy(BackpressurePolicy::ShedOldest)
992            .pipeline_max_input_buf_batches(64)
993            .delivery_guarantee(DeliveryGuarantee::ExactlyOnce)
994            .build()
995            .await
996            .expect_err("ShedOldest + ExactlyOnce must be rejected");
997        assert!(err.to_string().contains("exactly-once"), "{err}");
998    }
999
1000    #[tokio::test]
1001    async fn local_replay_delivery_rejects_cloud_checkpoint_storage_at_build() {
1002        use laminar_connectors::connector::DeliveryGuarantee;
1003
1004        let error = LaminarDbBuilder::new()
1005            .checkpoint(StreamCheckpointConfig::default())
1006            .delivery_guarantee(DeliveryGuarantee::AtLeastOnce)
1007            .object_store_url("s3://shared-checkpoints/deployment")
1008            .build()
1009            .await
1010            .expect_err("an unfenced local cloud writer must fail before startup");
1011        assert!(error.to_string().contains("[LDB-0014]"), "{error}");
1012    }
1013
1014    #[tokio::test]
1015    async fn checkpoint_url_conflicts_with_checkpoint_data_dir() {
1016        let error = LaminarDbBuilder::new()
1017            .checkpoint(StreamCheckpointConfig {
1018                data_dir: Some(std::path::PathBuf::from("checkpoint-data")),
1019                ..StreamCheckpointConfig::default()
1020            })
1021            .object_store_url("memory://checkpoint-test")
1022            .build()
1023            .await
1024            .expect_err("two checkpoint authorities must be rejected");
1025        assert!(error.to_string().contains("conflicts"), "{error}");
1026    }
1027
1028    #[tokio::test]
1029    async fn checkpoint_state_budget_is_resolved_and_propagated_to_status_stores() {
1030        for max_staged_bytes in [17, isize::MAX as u64] {
1031            let directory = tempfile::tempdir().unwrap();
1032            let db = LaminarDbBuilder::new()
1033                .storage_dir(directory.path())
1034                .checkpoint(StreamCheckpointConfig {
1035                    max_staged_bytes: Some(max_staged_bytes),
1036                    ..StreamCheckpointConfig::default()
1037                })
1038                .build()
1039                .await
1040                .unwrap();
1041
1042            assert_eq!(
1043                db.config
1044                    .checkpoint
1045                    .as_ref()
1046                    .and_then(|checkpoint| checkpoint.max_staged_bytes),
1047                Some(max_staged_bytes)
1048            );
1049            assert_eq!(
1050                db.checkpoint_store()
1051                    .unwrap()
1052                    .unwrap()
1053                    .max_state_data_bytes(),
1054                max_staged_bytes
1055            );
1056        }
1057    }
1058
1059    #[tokio::test]
1060    async fn checkpoint_state_budget_defaults_to_the_core_storage_limit() {
1061        let directory = tempfile::tempdir().unwrap();
1062        let db = LaminarDbBuilder::new()
1063            .storage_dir(directory.path())
1064            .checkpoint(StreamCheckpointConfig::default())
1065            .build()
1066            .await
1067            .unwrap();
1068        let expected = laminar_core::storage::checkpoint_store::DEFAULT_MAX_CHECKPOINT_STATE_BYTES;
1069
1070        assert_eq!(
1071            db.config
1072                .checkpoint
1073                .as_ref()
1074                .and_then(|checkpoint| checkpoint.max_staged_bytes),
1075            Some(expected)
1076        );
1077        assert_eq!(
1078            db.checkpoint_store()
1079                .unwrap()
1080                .unwrap()
1081                .max_state_data_bytes(),
1082            expected
1083        );
1084    }
1085
1086    #[tokio::test]
1087    async fn checkpoint_state_budget_rejects_an_unaddressable_limit_at_build() {
1088        let unaddressable = (isize::MAX as u64) + 1;
1089        let error = LaminarDbBuilder::new()
1090            .checkpoint(StreamCheckpointConfig {
1091                max_staged_bytes: Some(unaddressable),
1092                ..StreamCheckpointConfig::default()
1093            })
1094            .build()
1095            .await
1096            .expect_err("checkpoint recovery must not accept an unaddressable sidecar limit");
1097
1098        assert!(
1099            error
1100                .to_string()
1101                .contains("exceeds this process address space"),
1102            "{error}"
1103        );
1104    }
1105
1106    #[tokio::test]
1107    async fn explicit_profile_cannot_bypass_file_url_validation() {
1108        let error = LaminarDbBuilder::new()
1109            .profile(Profile::BareMetal)
1110            .object_store_url("file://./relative")
1111            .build()
1112            .await
1113            .expect_err("malformed checkpoint URLs must fail independently of tuning profile");
1114        assert!(
1115            error.to_string().contains("invalid object store URL"),
1116            "{error}"
1117        );
1118    }
1119
1120    #[test]
1121    fn cluster_delivery_admission_fails_closed_without_durable_term_binding() {
1122        use laminar_connectors::connector::DeliveryGuarantee;
1123
1124        assert!(LaminarDbBuilder::validate_cluster_delivery(
1125            RuntimeMode::Cluster,
1126            DeliveryGuarantee::AtLeastOnce,
1127        )
1128        .is_ok());
1129
1130        let error = LaminarDbBuilder::validate_cluster_delivery(
1131            RuntimeMode::Cluster,
1132            DeliveryGuarantee::ExactlyOnce,
1133        )
1134        .expect_err("cluster EO must remain closed until connectors consume the fence end to end");
1135        assert!(error.to_string().contains("[LDB-0013]"), "{error}");
1136        assert!(
1137            error.to_string().contains("external sink cursor commit"),
1138            "{error}"
1139        );
1140
1141        assert!(LaminarDbBuilder::validate_cluster_delivery(
1142            RuntimeMode::Local,
1143            DeliveryGuarantee::ExactlyOnce,
1144        )
1145        .is_ok());
1146    }
1147
1148    #[tokio::test]
1149    async fn cluster_profile_without_controller_is_rejected() {
1150        let error = LaminarDbBuilder::new()
1151            .profile(Profile::Cluster)
1152            .object_store_url("memory://checkpoint-test")
1153            .build()
1154            .await
1155            .expect_err("cluster profile must not degrade into a local runtime");
1156        assert!(error.to_string().contains("cluster controller"), "{error}");
1157    }
1158
1159    #[cfg(feature = "cluster")]
1160    #[tokio::test]
1161    async fn controller_selects_cluster_runtime_when_profile_is_inferred() {
1162        let checkpoint_store = test_cluster_checkpoint_store();
1163        let controller = test_cluster_controller();
1164        let verified_namespaces =
1165            test_verified_cluster_namespaces(controller.as_ref(), Arc::clone(&checkpoint_store))
1166                .await;
1167        let verified_state_store = verified_namespaces.state_store();
1168        let (state_backend, vnode_registry) = test_verified_state_topology(&verified_namespaces);
1169        let db = LaminarDbBuilder::new()
1170            .cluster_controller(controller)
1171            .verified_cluster_namespaces(verified_namespaces)
1172            .state_backend(state_backend)
1173            .vnode_registry(vnode_registry)
1174            .build()
1175            .await
1176            .unwrap();
1177        assert!(db.is_cluster_runtime());
1178        assert_eq!(db.config.default_buffer_size, 262_144);
1179        assert!(Arc::ptr_eq(
1180            &db.cluster_checkpoint_object_store().unwrap(),
1181            &checkpoint_store
1182        ));
1183        assert!(db
1184            .state_backend
1185            .lock()
1186            .as_ref()
1187            .unwrap()
1188            .uses_exact_object_store(&verified_state_store));
1189        assert_eq!(
1190            db.config.delivery_guarantee,
1191            laminar_connectors::connector::DeliveryGuarantee::AtLeastOnce
1192        );
1193    }
1194
1195    #[cfg(feature = "cluster")]
1196    #[tokio::test]
1197    async fn injected_checkpoint_store_conflicts_with_checkpoint_data_dir() {
1198        let error = LaminarDbBuilder::new()
1199            .cluster_controller(test_cluster_controller())
1200            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1201            .checkpoint(StreamCheckpointConfig {
1202                data_dir: Some(std::path::PathBuf::from("checkpoint-data")),
1203                ..StreamCheckpointConfig::default()
1204            })
1205            .build()
1206            .await
1207            .expect_err("two checkpoint authorities must be rejected");
1208        assert!(error.to_string().contains("conflict"), "{error}");
1209    }
1210
1211    #[cfg(feature = "cluster")]
1212    #[tokio::test]
1213    async fn checkpoint_status_reads_the_injected_participant_store() {
1214        use arrow::array::UInt64Array;
1215        use laminar_core::checkpoint::checkpoint_manifest::DurableCheckpointPhase;
1216        use laminar_core::storage::CheckpointStore;
1217
1218        let controller = test_cluster_controller();
1219        let participant_id = controller.instance_id().0;
1220        let object_store = test_cluster_checkpoint_store();
1221        let max_staged_bytes = 97;
1222        let db = LaminarDbBuilder::new()
1223            .cluster_controller(controller)
1224            .cluster_checkpoint_object_store(Arc::clone(&object_store))
1225            .checkpoint(StreamCheckpointConfig {
1226                max_staged_bytes: Some(max_staged_bytes),
1227                ..StreamCheckpointConfig::default()
1228            })
1229            .build()
1230            .await
1231            .unwrap();
1232        let status_store = db.checkpoint_store().unwrap().unwrap();
1233        assert_eq!(status_store.participant_id(), participant_id);
1234        assert_eq!(status_store.max_state_data_bytes(), max_staged_bytes);
1235        let participant_store =
1236            laminar_core::storage::checkpoint_store::ObjectStoreCheckpointStore::new(
1237                object_store,
1238                format!("nodes/{participant_id}/"),
1239            )
1240            .with_max_state_data_bytes(max_staged_bytes)
1241            .unwrap()
1242            .with_key_group_count(db.checkpoint_key_groups())
1243            .with_participant_id(participant_id);
1244        let mut manifest = laminar_core::checkpoint::CheckpointManifest::new_with_key_group_count(
1245            7,
1246            7,
1247            db.checkpoint_key_groups(),
1248        );
1249        manifest.participant_id = participant_id;
1250        manifest.durable_phase = DurableCheckpointPhase::Finalized;
1251        participant_store.save(&manifest).await.unwrap();
1252
1253        let status = db.build_show_checkpoint_status().await.unwrap();
1254        let checkpoint_ids = status
1255            .column(0)
1256            .as_any()
1257            .downcast_ref::<UInt64Array>()
1258            .unwrap();
1259        let total = status
1260            .column(5)
1261            .as_any()
1262            .downcast_ref::<UInt64Array>()
1263            .unwrap();
1264        assert_eq!(checkpoint_ids.value(0), 7);
1265        assert_eq!(total.value(0), 1);
1266    }
1267
1268    #[cfg(feature = "cluster")]
1269    #[tokio::test]
1270    async fn cluster_runtime_rejects_a_controller_without_process_lease_authority() {
1271        let error = LaminarDbBuilder::new()
1272            .cluster_controller(test_cluster_controller_without_deadline())
1273            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1274            .build()
1275            .await
1276            .expect_err("cluster construction must not fail open without a process lease clock");
1277
1278        assert!(
1279            error.to_string().contains("shared process lease deadline"),
1280            "{error}"
1281        );
1282    }
1283
1284    #[cfg(feature = "cluster")]
1285    #[tokio::test]
1286    async fn cluster_builder_binds_shuffle_to_the_controller_deadline() {
1287        use laminar_core::cluster::control::LeaseDeadline;
1288
1289        let controller = test_cluster_controller();
1290        let sender = Arc::new(laminar_core::shuffle::ShuffleSender::new(
1291            1,
1292            uuid::Uuid::from_u128(1),
1293        ));
1294        let receiver = Arc::new(
1295            laminar_core::shuffle::ShuffleReceiver::bind(
1296                1,
1297                "127.0.0.1:0".parse().unwrap(),
1298                uuid::Uuid::from_u128(1),
1299            )
1300            .await
1301            .unwrap(),
1302        );
1303        let _db = LaminarDbBuilder::new()
1304            .cluster_controller(controller)
1305            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1306            .shuffle_sender(Arc::clone(&sender))
1307            .shuffle_receiver(Arc::clone(&receiver))
1308            .build()
1309            .await
1310            .unwrap();
1311        let different = Arc::new(LeaseDeadline::live_for(std::time::Duration::from_secs(60)));
1312
1313        assert!(sender
1314            .install_process_lease_deadline(Arc::clone(&different))
1315            .is_err());
1316        assert!(receiver.install_process_lease_deadline(different).is_err());
1317    }
1318
1319    #[cfg(feature = "cluster")]
1320    #[tokio::test]
1321    async fn cluster_shuffle_binding_failure_does_not_partially_bind_sender() {
1322        use laminar_core::cluster::control::LeaseDeadline;
1323
1324        let sender = Arc::new(laminar_core::shuffle::ShuffleSender::new(
1325            1,
1326            uuid::Uuid::from_u128(1),
1327        ));
1328        let receiver = Arc::new(
1329            laminar_core::shuffle::ShuffleReceiver::bind(
1330                1,
1331                "127.0.0.1:0".parse().unwrap(),
1332                uuid::Uuid::from_u128(1),
1333            )
1334            .await
1335            .unwrap(),
1336        );
1337        receiver
1338            .install_process_lease_deadline(Arc::new(LeaseDeadline::live_for(
1339                std::time::Duration::from_secs(60),
1340            )))
1341            .unwrap();
1342
1343        let error = LaminarDbBuilder::new()
1344            .cluster_controller(test_cluster_controller())
1345            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1346            .shuffle_sender(Arc::clone(&sender))
1347            .shuffle_receiver(receiver)
1348            .build()
1349            .await
1350            .expect_err("an incompatible receiver lease must reject the pair");
1351        assert!(error.to_string().contains("already installed"), "{error}");
1352
1353        sender
1354            .install_process_lease_deadline(Arc::new(LeaseDeadline::live_for(
1355                std::time::Duration::from_secs(60),
1356            )))
1357            .expect("failed pair validation must leave the sender unbound");
1358    }
1359
1360    #[cfg(feature = "cluster")]
1361    #[tokio::test]
1362    async fn stateless_cluster_uses_default_key_group_topology() {
1363        let db = LaminarDbBuilder::new()
1364            .cluster_controller(test_cluster_controller())
1365            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1366            .build()
1367            .await
1368            .unwrap();
1369
1370        assert_eq!(
1371            db.checkpoint_key_groups(),
1372            laminar_core::state::DEFAULT_CLUSTER_KEY_GROUP_COUNT
1373        );
1374        assert!(db.state_backend.lock().is_none());
1375        assert!(db.vnode_registry.lock().is_none());
1376    }
1377
1378    #[cfg(feature = "cluster")]
1379    #[tokio::test]
1380    async fn fresh_cluster_database_keeps_intake_fenced() {
1381        let db = LaminarDbBuilder::new()
1382            .cluster_controller(test_cluster_controller())
1383            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1384            .build()
1385            .await
1386            .unwrap();
1387
1388        assert!(db.cluster_intake_fenced());
1389        db.fence_cluster_startup();
1390        db.fence_cluster_startup();
1391        assert!(db.cluster_intake_fenced());
1392    }
1393
1394    #[cfg(feature = "cluster")]
1395    #[tokio::test]
1396    async fn inferred_cluster_requires_shared_checkpoint_storage() {
1397        let error = LaminarDbBuilder::new()
1398            .cluster_controller(test_cluster_controller())
1399            .build()
1400            .await
1401            .expect_err("cluster runtime must not retain bare-metal storage admission");
1402        assert!(
1403            error.to_string().contains("VerifiedClusterNamespaces"),
1404            "{error}"
1405        );
1406    }
1407
1408    #[cfg(feature = "cluster")]
1409    #[tokio::test]
1410    async fn complete_cluster_profile_selects_cluster_runtime() {
1411        let controller = test_cluster_controller();
1412        let verified_namespaces =
1413            test_verified_cluster_namespaces(controller.as_ref(), test_cluster_checkpoint_store())
1414                .await;
1415        let (state_backend, vnode_registry) = test_verified_state_topology(&verified_namespaces);
1416        let db = LaminarDbBuilder::new()
1417            .profile(Profile::Cluster)
1418            .cluster_controller(controller)
1419            .verified_cluster_namespaces(verified_namespaces)
1420            .state_backend(state_backend)
1421            .vnode_registry(vnode_registry)
1422            .build()
1423            .await
1424            .unwrap();
1425        assert!(db.is_cluster_runtime());
1426    }
1427
1428    #[cfg(feature = "cluster")]
1429    #[tokio::test]
1430    async fn cluster_runtime_rejects_node_local_checkpoint_url() {
1431        let directory = tempfile::tempdir().unwrap();
1432        let path = directory.path().display().to_string().replace('\\', "/");
1433        let url = if path.starts_with('/') {
1434            format!("file://{path}")
1435        } else {
1436            format!("file:///{path}")
1437        };
1438
1439        let error = LaminarDbBuilder::new()
1440            .profile(Profile::Cluster)
1441            .object_store_url(url)
1442            .cluster_controller(test_cluster_controller())
1443            .build()
1444            .await
1445            .expect_err("cluster recovery must not depend on node-local checkpoint storage");
1446        assert!(error.to_string().contains("[LDB-0011]"), "{error}");
1447        assert!(
1448            error
1449                .to_string()
1450                .contains("successful shared namespace proof"),
1451            "{error}"
1452        );
1453    }
1454
1455    #[cfg(feature = "cluster")]
1456    #[tokio::test]
1457    async fn cluster_runtime_rejects_shared_checkpoint_url_without_proof() {
1458        let error = LaminarDbBuilder::new()
1459            .profile(Profile::Cluster)
1460            .object_store_url("s3://checkpoint-test/cluster")
1461            .cluster_controller(test_cluster_controller())
1462            .build()
1463            .await
1464            .expect_err("a shared URL must not bypass the exact namespace proof");
1465        assert!(error.to_string().contains("[LDB-0011]"), "{error}");
1466        assert!(
1467            error
1468                .to_string()
1469                .contains("successful shared namespace proof"),
1470            "{error}"
1471        );
1472    }
1473
1474    #[cfg(feature = "cluster")]
1475    #[tokio::test]
1476    async fn cluster_runtime_rejects_namespace_proof_from_another_process() {
1477        let proved_controller = test_cluster_controller();
1478        let verified_namespaces = test_verified_cluster_namespaces(
1479            proved_controller.as_ref(),
1480            test_cluster_checkpoint_store(),
1481        )
1482        .await;
1483        let runtime_controller = test_cluster_controller();
1484
1485        let error = LaminarDbBuilder::new()
1486            .cluster_controller(runtime_controller)
1487            .verified_cluster_namespaces(verified_namespaces)
1488            .build()
1489            .await
1490            .expect_err("a namespace proof must remain bound to its process incarnation");
1491        assert!(error.to_string().contains("[LDB-0011]"), "{error}");
1492        assert!(error.to_string().contains("belong to node"), "{error}");
1493    }
1494
1495    #[cfg(feature = "cluster")]
1496    #[tokio::test]
1497    async fn cluster_runtime_rejects_in_process_state_backend() {
1498        let controller = test_cluster_controller();
1499        let verified_namespaces =
1500            test_verified_cluster_namespaces(controller.as_ref(), test_cluster_checkpoint_store())
1501                .await;
1502        let error = LaminarDbBuilder::new()
1503            .cluster_controller(controller)
1504            .verified_cluster_namespaces(verified_namespaces)
1505            .state_backend(Arc::new(laminar_core::state::InProcessBackend::new(1)))
1506            .vnode_registry(Arc::new(laminar_core::state::VnodeRegistry::new(1)))
1507            .build()
1508            .await
1509            .expect_err("volatile state must not enter cluster runtime");
1510        assert!(error.to_string().contains("[LDB-0011]"), "{error}");
1511        assert!(error.to_string().contains("ClusterShared"), "{error}");
1512    }
1513
1514    #[cfg(feature = "cluster")]
1515    #[tokio::test]
1516    async fn cluster_runtime_rejects_different_state_object_store() {
1517        let controller = test_cluster_controller();
1518        let verified_namespaces =
1519            test_verified_cluster_namespaces(controller.as_ref(), test_cluster_checkpoint_store())
1520                .await;
1521        let wrong_backend: Arc<dyn laminar_core::state::StateBackend> =
1522            Arc::new(laminar_core::state::ObjectStoreBackend::cluster_shared(
1523                test_cluster_checkpoint_store(),
1524                "wrong-state-store",
1525                1,
1526            ));
1527        let error = LaminarDbBuilder::new()
1528            .cluster_controller(controller)
1529            .verified_cluster_namespaces(verified_namespaces)
1530            .state_backend(wrong_backend)
1531            .vnode_registry(Arc::new(laminar_core::state::VnodeRegistry::new(1)))
1532            .build()
1533            .await
1534            .expect_err("a different ClusterShared handle must not bypass namespace proof");
1535        assert!(error.to_string().contains("[LDB-0011]"), "{error}");
1536        assert!(
1537            error.to_string().contains("exact object-store handle"),
1538            "{error}"
1539        );
1540    }
1541
1542    #[cfg(feature = "cluster")]
1543    #[tokio::test]
1544    async fn explicit_local_profile_rejects_cluster_controller() {
1545        let error = LaminarDbBuilder::new()
1546            .profile(Profile::BareMetal)
1547            .cluster_controller(test_cluster_controller())
1548            .build()
1549            .await
1550            .expect_err("explicit local profile and cluster wiring must not disagree");
1551        assert!(error.to_string().contains("cannot be combined"), "{error}");
1552    }
1553
1554    #[cfg(feature = "cluster")]
1555    #[tokio::test]
1556    async fn explicit_cluster_best_effort_is_rejected() {
1557        let error = LaminarDbBuilder::new()
1558            .cluster_controller(test_cluster_controller())
1559            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1560            .delivery_guarantee(laminar_connectors::connector::DeliveryGuarantee::BestEffort)
1561            .build()
1562            .await
1563            .expect_err("explicit cluster best-effort must fail closed");
1564        assert!(
1565            error.to_string().contains("requires at_least_once"),
1566            "{error}"
1567        );
1568    }
1569
1570    #[cfg(feature = "cluster")]
1571    #[tokio::test]
1572    async fn cluster_only_handle_without_controller_is_rejected() {
1573        let error = LaminarDbBuilder::new()
1574            .shuffle_sender(Arc::new(laminar_core::shuffle::ShuffleSender::new(
1575                1,
1576                uuid::Uuid::from_u128(1),
1577            )))
1578            .build()
1579            .await
1580            .expect_err("partial cluster wiring must not create a local runtime");
1581        assert!(
1582            error.to_string().contains("require a cluster controller"),
1583            "{error}"
1584        );
1585    }
1586
1587    #[cfg(feature = "cluster")]
1588    #[tokio::test]
1589    async fn cluster_checkpoint_store_without_controller_is_rejected() {
1590        let error = LaminarDbBuilder::new()
1591            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1592            .build()
1593            .await
1594            .expect_err("shared cluster checkpoint wiring must select a controller authority");
1595        assert!(
1596            error.to_string().contains("require a cluster controller"),
1597            "{error}"
1598        );
1599    }
1600
1601    #[cfg(feature = "cluster")]
1602    #[tokio::test]
1603    async fn cluster_checkpoint_store_conflicts_with_url() {
1604        let error = LaminarDbBuilder::new()
1605            .cluster_controller(test_cluster_controller())
1606            .cluster_checkpoint_object_store(test_cluster_checkpoint_store())
1607            .object_store_url("memory://checkpoint-test")
1608            .build()
1609            .await
1610            .expect_err("two checkpoint namespace authorities must be rejected");
1611        assert!(error.to_string().contains("conflict"), "{error}");
1612    }
1613
1614    #[cfg(feature = "cluster")]
1615    #[tokio::test]
1616    async fn cluster_exactly_once_build_is_rejected_before_runtime_start() {
1617        use laminar_connectors::connector::DeliveryGuarantee;
1618
1619        let error = LaminarDbBuilder::new()
1620            .profile(Profile::Cluster)
1621            .object_store_url("memory://checkpoint-test")
1622            .cluster_controller(test_cluster_controller())
1623            .delivery_guarantee(DeliveryGuarantee::ExactlyOnce)
1624            .build()
1625            .await
1626            .expect_err("cluster EO must fail at database admission");
1627        assert!(error.to_string().contains("[LDB-0013]"), "{error}");
1628    }
1629
1630    #[tokio::test]
1631    async fn injected_vnode_registry_must_fit_checkpoint_abi() {
1632        let vnode_count = laminar_core::state::MAX_KEY_GROUP_COUNT + 1;
1633        let error = LaminarDbBuilder::new()
1634            .state_backend(Arc::new(laminar_core::state::InProcessBackend::new(1)))
1635            .vnode_registry(Arc::new(laminar_core::state::VnodeRegistry::new(
1636                vnode_count,
1637            )))
1638            .build()
1639            .await
1640            .expect_err("oversized injected vnode registry must be rejected");
1641        assert!(
1642            error
1643                .to_string()
1644                .contains("vnode_registry count must be between 1"),
1645            "{error}"
1646        );
1647    }
1648
1649    #[tokio::test]
1650    async fn injected_state_backend_capacity_must_fit_checkpoint_abi() {
1651        let error = LaminarDbBuilder::new()
1652            .state_backend(Arc::new(laminar_core::state::InProcessBackend::new(0)))
1653            .vnode_registry(Arc::new(laminar_core::state::VnodeRegistry::new(1)))
1654            .build()
1655            .await
1656            .expect_err("invalid backend topology must be rejected at admission");
1657        assert!(
1658            error
1659                .to_string()
1660                .contains("state_backend key-group capacity must be between 1"),
1661            "{error}"
1662        );
1663    }
1664
1665    #[tokio::test]
1666    async fn state_backend_capacity_must_match_registry() {
1667        let error = LaminarDbBuilder::new()
1668            .state_backend(Arc::new(laminar_core::state::InProcessBackend::new(2)))
1669            .vnode_registry(Arc::new(laminar_core::state::VnodeRegistry::new(1)))
1670            .build()
1671            .await
1672            .expect_err("backend and registry must describe one topology");
1673        assert!(
1674            error.to_string().contains(
1675                "state_backend key-group capacity 2 does not match vnode_registry count 1"
1676            ),
1677            "{error}"
1678        );
1679    }
1680
1681    #[tokio::test]
1682    async fn local_runtime_rejects_multi_key_group_registry() {
1683        let error = LaminarDbBuilder::new()
1684            .state_backend(Arc::new(laminar_core::state::InProcessBackend::new(2)))
1685            .vnode_registry(Arc::new(laminar_core::state::VnodeRegistry::new(2)))
1686            .build()
1687            .await
1688            .expect_err("local runtime must not expose cluster rescaling dimensions");
1689        assert!(
1690            error.to_string().contains("requires exactly 1 key group"),
1691            "{error}"
1692        );
1693    }
1694
1695    #[tokio::test]
1696    async fn test_valid_shed_oldest_builds() {
1697        use crate::config::BackpressurePolicy;
1698        let db = LaminarDbBuilder::new()
1699            .pipeline_backpressure_policy(BackpressurePolicy::ShedOldest)
1700            .pipeline_max_input_buf_batches(64)
1701            .build()
1702            .await
1703            .unwrap();
1704        assert!(!db.is_closed());
1705    }
1706
1707    #[tokio::test]
1708    async fn test_builder_with_config_vars() {
1709        let db = LaminarDbBuilder::new()
1710            .config_var("KAFKA_BROKERS", "localhost:9092")
1711            .config_var("GROUP_ID", "test-group")
1712            .build()
1713            .await
1714            .unwrap();
1715        assert!(!db.is_closed());
1716    }
1717
1718    #[tokio::test]
1719    async fn test_builder_with_options() {
1720        let db = LaminarDbBuilder::new()
1721            .buffer_size(131_072)
1722            .build()
1723            .await
1724            .unwrap();
1725        assert!(!db.is_closed());
1726    }
1727
1728    #[tokio::test]
1729    async fn test_builder_from_laminardb() {
1730        let db = LaminarDB::builder().build().await.unwrap();
1731        assert!(!db.is_closed());
1732    }
1733
1734    #[test]
1735    fn test_builder_debug() {
1736        let builder = LaminarDbBuilder::new().config_var("K", "V");
1737        let debug = format!("{builder:?}");
1738        assert!(debug.contains("LaminarDbBuilder"));
1739        assert!(debug.contains("config_vars_count: 1"));
1740    }
1741
1742    #[tokio::test]
1743    async fn test_builder_register_udf() {
1744        use std::any::Any;
1745        use std::hash::{Hash, Hasher};
1746
1747        use arrow::datatypes::DataType;
1748        use datafusion_expr::{
1749            ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
1750        };
1751
1752        /// Trivial UDF that returns 42.
1753        #[derive(Debug)]
1754        struct FortyTwo {
1755            signature: Signature,
1756        }
1757
1758        impl FortyTwo {
1759            fn new() -> Self {
1760                Self {
1761                    signature: Signature::new(TypeSignature::Nullary, Volatility::Immutable),
1762                }
1763            }
1764        }
1765
1766        impl PartialEq for FortyTwo {
1767            fn eq(&self, _: &Self) -> bool {
1768                true
1769            }
1770        }
1771
1772        impl Eq for FortyTwo {}
1773
1774        impl Hash for FortyTwo {
1775            fn hash<H: Hasher>(&self, state: &mut H) {
1776                "forty_two".hash(state);
1777            }
1778        }
1779
1780        impl ScalarUDFImpl for FortyTwo {
1781            fn as_any(&self) -> &dyn Any {
1782                self
1783            }
1784            fn name(&self) -> &'static str {
1785                "forty_two"
1786            }
1787            fn signature(&self) -> &Signature {
1788                &self.signature
1789            }
1790            fn return_type(&self, _: &[DataType]) -> datafusion_common::Result<DataType> {
1791                Ok(DataType::Int64)
1792            }
1793            fn invoke_with_args(
1794                &self,
1795                _args: ScalarFunctionArgs,
1796            ) -> datafusion_common::Result<ColumnarValue> {
1797                Ok(ColumnarValue::Scalar(
1798                    datafusion_common::ScalarValue::Int64(Some(42)),
1799                ))
1800            }
1801        }
1802
1803        let udf = ScalarUDF::new_from_impl(FortyTwo::new());
1804        let db = LaminarDB::builder()
1805            .register_udf(udf)
1806            .build()
1807            .await
1808            .unwrap();
1809
1810        // Verify the UDF is queryable
1811        let result = db.execute("SELECT forty_two()").await;
1812        assert!(result.is_ok(), "UDF should be callable: {result:?}");
1813    }
1814}
laminar_db/builder.rs

laminar_db/
builder.rs
