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/// What to do when an operator's input buffer exceeds its cap.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
12pub enum BackpressurePolicy {
13    /// Defer the producer; sources block on `send`. No data loss.
14    #[default]
15    Backpressure,
16    /// Drop oldest batches; counted in `shed_records_total`.
17    ShedOldest,
18    /// Error out the cycle.
19    Fail,
20}
21
22/// String wrapper whose `Debug` redacts the value, for credentials in [`LaminarConfig`].
23#[derive(Clone)]
24pub struct SecretString(String);
25
26impl SecretString {
27    /// Wrap a secret value.
28    pub fn new(value: impl Into<String>) -> Self {
29        Self(value.into())
30    }
31
32    /// Borrow the underlying secret. Call only at the point of use.
33    #[must_use]
34    pub fn expose(&self) -> &str {
35        &self.0
36    }
37}
38
39impl std::fmt::Debug for SecretString {
40    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41        f.write_str("\"[REDACTED]\"")
42    }
43}
44
45/// Auto-restart policy for the fault supervisor (see `LaminarDB::enable_supervision`).
46#[derive(Debug, Clone)]
47pub struct RestartPolicy {
48    /// Max restarts within `window` before the pipeline is left hard-faulted.
49    pub max_restarts: usize,
50    /// Sliding window over which `max_restarts` is counted.
51    pub window: std::time::Duration,
52    /// Backoff before the first restart in a window.
53    pub initial_backoff: std::time::Duration,
54    /// Cap on the exponential backoff.
55    pub max_backoff: std::time::Duration,
56}
57
58impl Default for RestartPolicy {
59    fn default() -> Self {
60        Self {
61            max_restarts: 5,
62            window: std::time::Duration::from_secs(60),
63            initial_backoff: std::time::Duration::from_millis(500),
64            max_backoff: std::time::Duration::from_secs(30),
65        }
66    }
67}
68
69/// Configuration for a `LaminarDB` instance.
70#[derive(Debug, Clone)]
71pub struct LaminarConfig {
72    /// Streaming channel buffer size.
73    pub default_buffer_size: usize,
74    /// Backpressure strategy.
75    pub default_backpressure: BackpressureStrategy,
76    /// Checkpoint directory. `None` = in-memory only.
77    pub storage_dir: Option<PathBuf>,
78    /// Checkpoint config. `None` = disabled.
79    pub checkpoint: Option<StreamCheckpointConfig>,
80    /// Emit dirty-only changelogs for keyed non-windowed aggregate materialized views instead of
81    /// re-materializing every group each cycle. This is query execution policy, not checkpointing.
82    pub incremental_emit: bool,
83    /// Cloud checkpoint URL, e.g. `s3://bucket/prefix`.
84    pub object_store_url: Option<String>,
85    /// Credential/config overrides for the object store.
86    pub object_store_options: HashMap<String, String>,
87    /// Bearer token presented when forwarding requests to the cluster leader's
88    /// HTTP API (set when the server gates `/api/v1` with `console_token`).
89    pub http_auth_token: Option<SecretString>,
90    /// Delivery guarantee.
91    pub delivery_guarantee: DeliveryGuarantee,
92    /// Source-to-coordinator channel capacity. `None` = 64.
93    pub pipeline_channel_capacity: Option<usize>,
94    /// Micro-batch coalescing window. `None` = 5ms connectors / 0 embedded.
95    pub pipeline_batch_window: Option<std::time::Duration>,
96    /// Drain budget per cycle (ns). `None` = 1ms.
97    pub pipeline_drain_budget_ns: Option<u64>,
98    /// Per-query budget (ns). `None` = 8ms.
99    pub pipeline_query_budget_ns: Option<u64>,
100    /// Per-port operator input-buffer cap (batches). `None` = 256.
101    pub pipeline_max_input_buf_batches: Option<usize>,
102    /// Per-port operator input-buffer cap (bytes). `None` = disabled.
103    pub pipeline_max_input_buf_bytes: Option<usize>,
104    /// Backpressure policy. See [`BackpressurePolicy`].
105    pub pipeline_backpressure_policy: BackpressurePolicy,
106    /// Auto-restart policy applied when supervision is enabled.
107    pub restart_policy: RestartPolicy,
108    /// Isolate queries that share a source into independent failure domains.
109    /// Default off; when off, shared-source queries fault and recover together.
110    pub shared_source_isolation: bool,
111}
112
113impl Default for LaminarConfig {
114    fn default() -> Self {
115        Self {
116            default_buffer_size: 65536,
117            default_backpressure: BackpressureStrategy::Block,
118            storage_dir: None,
119            checkpoint: None,
120            incremental_emit: true,
121            object_store_url: None,
122            object_store_options: HashMap::new(),
123            http_auth_token: None,
124            delivery_guarantee: DeliveryGuarantee::default(),
125            pipeline_channel_capacity: None,
126            pipeline_batch_window: None,
127            pipeline_drain_budget_ns: None,
128            pipeline_query_budget_ns: None,
129            pipeline_max_input_buf_batches: None,
130            pipeline_max_input_buf_bytes: None,
131            pipeline_backpressure_policy: BackpressurePolicy::default(),
132            restart_policy: RestartPolicy::default(),
133            shared_source_isolation: false,
134        }
135    }
136}
laminar_db/config.rs

laminar_db/
config.rs
