laminar_db/
config.rs

1//! Configuration for `LaminarDB`.
2#![allow(clippy::disallowed_types)] // cold path
3
4use std::collections::HashMap;
5use std::path::PathBuf;
6
7use laminar_connectors::connector::DeliveryGuarantee;
8use laminar_core::streaming::{BackpressureStrategy, StreamCheckpointConfig};
9
10/// Default pipeline-wide lower-bound charge allowed for managed operator working state.
11///
12/// This execution budget is independent of checkpoint storage.
13pub const DEFAULT_MAX_MANAGED_STATE_BYTES: usize = 256 * 1024 * 1024;
14
15pub(crate) fn event_time_max_future_skew_ms(
16    skew: std::time::Duration,
17) -> Result<i64, &'static str> {
18    let skew_ms = i64::try_from(skew.as_millis())
19        .map_err(|_| "event_time_max_future_skew exceeds the supported millisecond range")?;
20    if !skew.is_zero() && skew_ms == 0 {
21        return Err("event_time_max_future_skew must be zero or at least 1ms");
22    }
23    Ok(skew_ms)
24}
25
26pub(crate) fn source_idle_timeout_ms(
27    timeout: Option<std::time::Duration>,
28) -> Result<Option<u64>, &'static str> {
29    let Some(timeout) = timeout else {
30        return Ok(None);
31    };
32    let timeout_ms = u64::try_from(timeout.as_millis())
33        .map_err(|_| "source_idle_timeout exceeds the supported millisecond range")?;
34    if timeout_ms == 0 {
35        return Err("source_idle_timeout must be at least 1ms");
36    }
37    Ok(Some(timeout_ms))
38}
39
40pub(crate) fn temporal_join_idle_history_retention_ms(
41    retention: Option<std::time::Duration>,
42) -> Result<i64, &'static str> {
43    let retention = retention
44        .ok_or("temporal_join_idle_history_retention must be configured for temporal joins")?;
45    let retention_ms = i64::try_from(retention.as_millis()).map_err(|_| {
46        "temporal_join_idle_history_retention exceeds the supported millisecond range"
47    })?;
48    if retention_ms == 0 {
49        return Err("temporal_join_idle_history_retention must be at least 1ms");
50    }
51    Ok(retention_ms)
52}
53
54/// What to do when an operator's input buffer exceeds its cap.
55#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
56pub enum BackpressurePolicy {
57    /// Defer the producer; sources block on `send`. No data loss.
58    #[default]
59    Backpressure,
60    /// Drop oldest batches; counted in `shed_records_total`.
61    ShedOldest,
62    /// Error out the cycle.
63    Fail,
64}
65
66/// String wrapper whose `Debug` redacts the value, for credentials in [`LaminarConfig`].
67#[derive(Clone)]
68pub struct SecretString(String);
69
70impl SecretString {
71    /// Wrap a secret value.
72    pub fn new(value: impl Into<String>) -> Self {
73        Self(value.into())
74    }
75
76    /// Borrow the underlying secret. Call only at the point of use.
77    #[must_use]
78    pub fn expose(&self) -> &str {
79        &self.0
80    }
81}
82
83impl std::fmt::Debug for SecretString {
84    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85        f.write_str("\"[REDACTED]\"")
86    }
87}
88
89/// Auto-restart policy for the fault supervisor (see `LaminarDB::enable_supervision`).
90#[derive(Debug, Clone)]
91pub struct RestartPolicy {
92    /// Max restarts within `window` before the pipeline is left hard-faulted.
93    pub max_restarts: usize,
94    /// Sliding window over which `max_restarts` is counted.
95    pub window: std::time::Duration,
96    /// Backoff before the first restart in a window.
97    pub initial_backoff: std::time::Duration,
98    /// Cap on the exponential backoff.
99    pub max_backoff: std::time::Duration,
100}
101
102impl Default for RestartPolicy {
103    fn default() -> Self {
104        Self {
105            max_restarts: 5,
106            window: std::time::Duration::from_secs(60),
107            initial_backoff: std::time::Duration::from_millis(500),
108            max_backoff: std::time::Duration::from_secs(30),
109        }
110    }
111}
112
113/// Configuration for a `LaminarDB` instance.
114#[derive(Debug, Clone)]
115pub struct LaminarConfig {
116    /// Streaming channel buffer size.
117    pub default_buffer_size: usize,
118    /// Backpressure strategy.
119    pub default_backpressure: BackpressureStrategy,
120    /// Checkpoint directory. `None` = in-memory only.
121    pub storage_dir: Option<PathBuf>,
122    /// Checkpoint config. `None` = disabled.
123    pub checkpoint: Option<StreamCheckpointConfig>,
124    /// Emit dirty-only changelogs for keyed non-windowed aggregate materialized views instead of
125    /// re-materializing every group each cycle. This is query execution policy, not checkpointing.
126    pub incremental_emit: bool,
127    /// Cloud checkpoint URL, e.g. `s3://bucket/prefix`.
128    pub object_store_url: Option<String>,
129    /// Credential/config overrides for the object store.
130    pub object_store_options: HashMap<String, String>,
131    /// Bearer token presented when forwarding requests to the cluster leader's
132    /// HTTP API (set when the server gates `/api/v1` with `console_token`).
133    pub http_auth_token: Option<SecretString>,
134    /// Delivery guarantee.
135    pub delivery_guarantee: DeliveryGuarantee,
136    /// Source-to-coordinator channel capacity. `None` = 64.
137    pub pipeline_channel_capacity: Option<usize>,
138    /// Micro-batch coalescing window. `None` = 5ms connectors / 0 embedded.
139    pub pipeline_batch_window: Option<std::time::Duration>,
140    /// Drain budget per cycle (ns). `None` = 1ms.
141    pub pipeline_drain_budget_ns: Option<u64>,
142    /// Per-query budget (ns). `None` = 8ms.
143    pub pipeline_query_budget_ns: Option<u64>,
144    /// Per-port operator input-buffer cap (batches). `None` = 256.
145    pub pipeline_max_input_buf_batches: Option<usize>,
146    /// Per-port operator input-buffer cap (bytes). `None` = disabled.
147    pub pipeline_max_input_buf_bytes: Option<usize>,
148    /// Pipeline-wide managed working-state budget in charged bytes. `None` resolves to
149    /// [`DEFAULT_MAX_MANAGED_STATE_BYTES`] when the database is constructed.
150    pub pipeline_max_managed_state_bytes: Option<usize>,
151    /// Retention contract for right-side history while a temporal join input is idle.
152    /// Required only when the pipeline contains a temporal join.
153    pub temporal_join_idle_history_retention: Option<std::time::Duration>,
154    /// Mark inactive watermarked sources and input channels idle after this duration.
155    /// `None` disables automatic idle detection.
156    pub source_idle_timeout: Option<std::time::Duration>,
157    /// Event timestamps farther ahead of wall clock do not advance source watermarks.
158    /// Zero disables the guard.
159    pub event_time_max_future_skew: std::time::Duration,
160    /// Backpressure policy. See [`BackpressurePolicy`].
161    pub pipeline_backpressure_policy: BackpressurePolicy,
162    /// Auto-restart policy applied when supervision is enabled.
163    pub restart_policy: RestartPolicy,
164    /// Isolate queries that share a source into independent failure domains.
165    /// Default off; when off, shared-source queries fault and recover together.
166    pub shared_source_isolation: bool,
167}
168
169impl Default for LaminarConfig {
170    fn default() -> Self {
171        Self {
172            default_buffer_size: 65536,
173            default_backpressure: BackpressureStrategy::Block,
174            storage_dir: None,
175            checkpoint: None,
176            incremental_emit: true,
177            object_store_url: None,
178            object_store_options: HashMap::new(),
179            http_auth_token: None,
180            delivery_guarantee: DeliveryGuarantee::default(),
181            pipeline_channel_capacity: None,
182            pipeline_batch_window: None,
183            pipeline_drain_budget_ns: None,
184            pipeline_query_budget_ns: None,
185            pipeline_max_input_buf_batches: None,
186            pipeline_max_input_buf_bytes: None,
187            pipeline_max_managed_state_bytes: None,
188            temporal_join_idle_history_retention: None,
189            source_idle_timeout: None,
190            event_time_max_future_skew: std::time::Duration::from_millis(
191                laminar_core::time::DEFAULT_MAX_FUTURE_SKEW_MS.unsigned_abs(),
192            ),
193            pipeline_backpressure_policy: BackpressurePolicy::default(),
194            restart_policy: RestartPolicy::default(),
195            shared_source_isolation: false,
196        }
197    }
198}
laminar_db/config.rs

laminar_db/
config.rs
