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
10use crate::error::DbError;
11
12/// Default pipeline-wide lower-bound charge allowed for managed operator working state.
13///
14/// This execution budget is independent of checkpoint storage.
15pub const DEFAULT_MAX_MANAGED_STATE_BYTES: usize = 256 * 1024 * 1024;
16
17/// Default per-DB limit for participating `DataFusion` reservations (256 MiB).
18/// This does not limit direct Arrow allocations or process RSS.
19pub const DEFAULT_DATAFUSION_MEMORY_LIMIT_BYTES: usize = 256 * 1024 * 1024;
20
21/// Default shared connector-to-coordinator Arrow-byte budget (64 MiB).
22pub const DEFAULT_SOURCE_QUEUE_MAX_BYTES: usize = 64 * 1024 * 1024;
23
24/// Default maximum number of live rows in each local reference table.
25pub const DEFAULT_REFERENCE_TABLE_MAX_ROWS: usize = 1_000_000;
26
27/// Default retained-memory charge limit for each local reference table (256 MiB).
28pub const DEFAULT_REFERENCE_TABLE_MAX_BYTES: usize = 256 * 1024 * 1024;
29
30/// Default live row limit per local materialized view (distinct rows for multisets).
31pub const DEFAULT_MATERIALIZED_VIEW_MAX_ROWS: usize = 1_000_000;
32
33/// Default retained-memory charge limit per local materialized view (256 MiB).
34pub const DEFAULT_MATERIALIZED_VIEW_MAX_BYTES: usize = 256 * 1024 * 1024;
35
36/// Default byte budget of decoded committed view state cached for reads (256 MiB).
37pub const DEFAULT_MATERIALIZED_VIEW_READ_CACHE_BYTES: usize = 256 * 1024 * 1024;
38
39/// Validate the live row and retained-memory limits for each local materialized view.
40///
41/// # Errors
42/// Returns an error if either limit is zero.
43pub fn validate_materialized_view_limits(rows: usize, bytes: usize) -> Result<(), &'static str> {
44    if rows == 0 || bytes == 0 {
45        return Err(
46            "materialized_view_max_rows and materialized_view_max_bytes must be greater than zero",
47        );
48    }
49    Ok(())
50}
51
52/// Validate per-table live row and retained-memory limits.
53///
54/// # Errors
55/// Returns an error if either limit is zero.
56pub fn validate_reference_table_limits(rows: usize, bytes: usize) -> Result<(), &'static str> {
57    if rows == 0 || bytes == 0 {
58        return Err(
59            "reference_table_max_rows and reference_table_max_bytes must be greater than zero",
60        );
61    }
62    Ok(())
63}
64
65/// Validate the pipeline-wide managed operator state budget.
66///
67/// # Errors
68/// Returns an error if the budget is zero.
69pub fn validate_managed_state_max_bytes(bytes: usize) -> Result<(), &'static str> {
70    if bytes == 0 {
71        return Err("pipeline_max_managed_state_bytes must be greater than zero");
72    }
73    Ok(())
74}
75
76/// Largest source queue budget supported by byte-sized semaphore reservations.
77pub const MAX_SOURCE_QUEUE_BYTES: usize = if tokio::sync::Semaphore::MAX_PERMITS < u32::MAX as usize
78{
79    tokio::sync::Semaphore::MAX_PERMITS
80} else {
81    u32::MAX as usize
82};
83
84/// Validates the shared source queue byte budget before connector I/O.
85///
86/// # Errors
87/// Returns an error for zero or values above [`MAX_SOURCE_QUEUE_BYTES`].
88pub fn validate_source_queue_max_bytes(bytes: usize) -> Result<(), &'static str> {
89    if bytes == 0 || bytes > MAX_SOURCE_QUEUE_BYTES {
90        return Err("source_queue_max_bytes must be nonzero and at most MAX_SOURCE_QUEUE_BYTES");
91    }
92    Ok(())
93}
94
95pub(crate) fn event_time_max_future_skew_ms(
96    skew: std::time::Duration,
97) -> Result<i64, &'static str> {
98    let skew_ms = i64::try_from(skew.as_millis())
99        .map_err(|_| "event_time_max_future_skew exceeds the supported millisecond range")?;
100    if !skew.is_zero() && skew_ms == 0 {
101        return Err("event_time_max_future_skew must be zero or at least 1ms");
102    }
103    Ok(skew_ms)
104}
105
106pub(crate) fn source_idle_timeout_ms(
107    timeout: Option<std::time::Duration>,
108) -> Result<Option<u64>, &'static str> {
109    let Some(timeout) = timeout else {
110        return Ok(None);
111    };
112    let timeout_ms = u64::try_from(timeout.as_millis())
113        .map_err(|_| "source_idle_timeout exceeds the supported millisecond range")?;
114    if timeout_ms == 0 {
115        return Err("source_idle_timeout must be at least 1ms");
116    }
117    Ok(Some(timeout_ms))
118}
119
120pub(crate) fn temporal_join_idle_history_retention_ms(
121    retention: Option<std::time::Duration>,
122) -> Result<i64, &'static str> {
123    let retention = retention
124        .ok_or("temporal_join_idle_history_retention must be configured for temporal joins")?;
125    let retention_ms = i64::try_from(retention.as_millis()).map_err(|_| {
126        "temporal_join_idle_history_retention exceeds the supported millisecond range"
127    })?;
128    if retention_ms == 0 {
129        return Err("temporal_join_idle_history_retention must be at least 1ms");
130    }
131    Ok(retention_ms)
132}
133
134/// What to do when an operator's input buffer exceeds its cap.
135#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
136pub enum BackpressurePolicy {
137    /// Defer producers before execution when a downstream port cannot accept input.
138    /// An executed result that exceeds a port's prospective limit halts the pipeline;
139    /// it cannot safely be retried against already mutated operator state.
140    #[default]
141    Backpressure,
142    /// Drop oldest batches; counted in `shed_records_total`.
143    /// Available only with [`DeliveryGuarantee::BestEffort`].
144    ShedOldest,
145    /// Error out the cycle.
146    Fail,
147}
148
149/// String wrapper whose `Debug` redacts the value, for credentials in [`LaminarConfig`].
150#[derive(Clone)]
151pub struct SecretString(String);
152
153impl SecretString {
154    /// Wrap a secret value.
155    pub fn new(value: impl Into<String>) -> Self {
156        Self(value.into())
157    }
158
159    /// Borrow the underlying secret. Call only at the point of use.
160    #[must_use]
161    pub fn expose(&self) -> &str {
162        &self.0
163    }
164}
165
166impl std::fmt::Debug for SecretString {
167    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
168        f.write_str("\"[REDACTED]\"")
169    }
170}
171
172/// Auto-restart policy for the fault supervisor (see `LaminarDB::enable_supervision`).
173#[derive(Debug, Clone)]
174pub struct RestartPolicy {
175    /// Max restarts within `window` before the pipeline is left hard-faulted.
176    pub max_restarts: usize,
177    /// Sliding window over which `max_restarts` is counted.
178    pub window: std::time::Duration,
179    /// Backoff before the first restart in a window.
180    pub initial_backoff: std::time::Duration,
181    /// Cap on the exponential backoff.
182    pub max_backoff: std::time::Duration,
183}
184
185impl Default for RestartPolicy {
186    fn default() -> Self {
187        Self {
188            max_restarts: 5,
189            window: std::time::Duration::from_secs(60),
190            initial_backoff: std::time::Duration::from_millis(500),
191            max_backoff: std::time::Duration::from_secs(30),
192        }
193    }
194}
195
196/// Configuration for a `LaminarDB` instance.
197#[derive(Debug, Clone)]
198pub struct LaminarConfig {
199    /// Live row limit per local MV; multiset mode counts distinct rows (default 1,000,000).
200    /// Applies to embedded and single-node execution; must be nonzero.
201    /// Keyed staging and each input batch allow at most twice this many rows.
202    pub materialized_view_max_rows: usize,
203    /// Per-MV retained-memory charge limit (default 256 MiB); must be nonzero.
204    /// Counts Arrow backing capacity, owned scalar/key storage and fixed entry metadata.
205    /// Shared buffers are conservatively charged per stored batch/scalar. Map spare capacity,
206    /// schema/converter/allocator overhead, snapshots and checkpoint scratch are separate.
207    /// Keyed staging and input conversion have separate limits of twice this charge plus
208    /// bounded staging metadata. Oversized net-neutral cycles can fail before live admission.
209    /// An append view counts the rows it holds in memory: those since the last committed
210    /// checkpoint, or its whole retained log without checkpointing. A cycle over the limit fails.
211    pub materialized_view_max_bytes: usize,
212    /// Byte budget of the committed view state decoded for reads and cached by digest (default
213    /// 256 MiB), shared by every view. Zero disables the cache.
214    pub materialized_view_read_cache_bytes: usize,
215    /// Maximum live rows in each reference table (default 1,000,000).
216    /// Applies to embedded and single-node tables; must be nonzero.
217    pub reference_table_max_rows: usize,
218    /// Per-table encoded-key, row metadata and retained Arrow capacity charge (default 256 MiB).
219    /// Shared buffers count once per table, using Arrow-reported allocation capacity.
220    /// Hash-map spare capacity, schema/allocator/opaque-owner overhead, caller-held data, prepared
221    /// replacements, query snapshots and checkpoint scratch are separate from this live limit.
222    pub reference_table_max_bytes: usize,
223    /// Shared limit for participating `DataFusion` reservations across this DB's contexts.
224    /// Must be greater than zero; defaults to [`DEFAULT_DATAFUSION_MEMORY_LIMIT_BYTES`].
225    /// DB-owned contexts cannot spill to disk. Direct Arrow allocations, managed state,
226    /// queues, connector-owned contexts and process RSS are outside this budget.
227    pub datafusion_memory_limit_bytes: usize,
228    /// Streaming channel buffer size.
229    pub default_buffer_size: usize,
230    /// Per-source Arrow-byte limit for in-process push queues, held until the pipeline reads.
231    /// Each source's snapshot history has a separate cap of the same size (default 64 MiB).
232    /// Applies to typed handles after Arrow conversion. Caller-held batches, conversion
233    /// scratch, query snapshots and downstream buffers are outside these two limits.
234    pub push_source_max_bytes: usize,
235    /// Backpressure strategy.
236    pub default_backpressure: BackpressureStrategy,
237    /// Checkpoint directory. `None` = in-memory only.
238    pub storage_dir: Option<PathBuf>,
239    /// Checkpoint config. `None` = disabled.
240    pub checkpoint: Option<StreamCheckpointConfig>,
241    /// Cloud checkpoint URL, e.g. `s3://bucket/prefix`.
242    pub object_store_url: Option<String>,
243    /// Credential/config overrides for the object store.
244    pub object_store_options: HashMap<String, String>,
245    /// Bearer token presented when forwarding requests to the cluster leader's
246    /// HTTP API (set when the server gates `/api/v1` with `console_token`).
247    pub http_auth_token: Option<SecretString>,
248    /// Delivery guarantee.
249    pub delivery_guarantee: DeliveryGuarantee,
250    /// Source-to-coordinator channel capacity. `None` = 64.
251    pub pipeline_channel_capacity: Option<usize>,
252    /// Shared connector FIFO Arrow-byte budget, including parked messages (default 64 MiB).
253    /// Also caps each source's one batch waiting for capacity or a cursor. Does not cover
254    /// connector decode scratch, embedded push rings, staged cycles, graph buffers or RSS.
255    pub source_queue_max_bytes: usize,
256    /// Micro-batch coalescing window. `None` = 5ms connectors / 0 embedded.
257    pub pipeline_batch_window: Option<std::time::Duration>,
258    /// Drain budget per cycle (ns). `None` = 1ms.
259    pub pipeline_drain_budget_ns: Option<u64>,
260    /// Per-query budget (ns). `None` = 8ms.
261    pub pipeline_query_budget_ns: Option<u64>,
262    /// Per-port operator input-buffer cap (batches). `None` = 256.
263    pub pipeline_max_input_buf_batches: Option<usize>,
264    /// Per-port retained Arrow-byte cap, including source priming. `None` = disabled.
265    /// Slices charge their backing storage; each fan-out port charges independently.
266    pub pipeline_max_input_buf_bytes: Option<usize>,
267    /// Pipeline-wide managed working-state budget in charged bytes. `None` resolves to
268    /// [`DEFAULT_MAX_MANAGED_STATE_BYTES`] when the database is constructed.
269    pub pipeline_max_managed_state_bytes: Option<usize>,
270    /// Retention contract for right-side history while a temporal join input is idle.
271    /// Required only when the pipeline contains a temporal join.
272    pub temporal_join_idle_history_retention: Option<std::time::Duration>,
273    /// Mark inactive watermarked sources and input channels idle after this duration.
274    /// `None` disables automatic idle detection.
275    pub source_idle_timeout: Option<std::time::Duration>,
276    /// Event timestamps farther ahead of wall clock do not advance source watermarks.
277    /// Zero disables the guard.
278    pub event_time_max_future_skew: std::time::Duration,
279    /// Backpressure policy. [`BackpressurePolicy::ShedOldest`] is `BestEffort` only.
280    pub pipeline_backpressure_policy: BackpressurePolicy,
281    /// Auto-restart policy applied when supervision is enabled.
282    pub restart_policy: RestartPolicy,
283    /// Isolate queries that share a source into independent failure domains.
284    /// Default off; when off, shared-source queries fault and recover together.
285    pub shared_source_isolation: bool,
286}
287
288impl LaminarConfig {
289    pub(crate) fn validate_and_normalize(&mut self) -> Result<(), DbError> {
290        validate_materialized_view_limits(
291            self.materialized_view_max_rows,
292            self.materialized_view_max_bytes,
293        )
294        .map_err(|error| DbError::Config(error.into()))?;
295        validate_reference_table_limits(
296            self.reference_table_max_rows,
297            self.reference_table_max_bytes,
298        )
299        .map_err(|error| DbError::Config(error.into()))?;
300        if self.datafusion_memory_limit_bytes == 0 {
301            return Err(DbError::Config(
302                "datafusion_memory_limit_bytes must be greater than zero".into(),
303            ));
304        }
305        validate_source_queue_max_bytes(self.source_queue_max_bytes)
306            .map_err(|error| DbError::Config(error.into()))?;
307        laminar_core::streaming::validate_source_max_queued_bytes(self.push_source_max_bytes)
308            .map_err(|error| DbError::Config(format!("push_source_max_bytes: {error}")))?;
309        self.source_idle_timeout = source_idle_timeout_ms(self.source_idle_timeout)
310            .map_err(|error| DbError::Config(error.to_string()))?
311            .map(std::time::Duration::from_millis);
312        let future_skew_ms = event_time_max_future_skew_ms(self.event_time_max_future_skew)
313            .map_err(|error| DbError::Config(error.to_string()))?;
314        self.event_time_max_future_skew =
315            std::time::Duration::from_millis(future_skew_ms.unsigned_abs());
316        let max_managed_state_bytes = self
317            .pipeline_max_managed_state_bytes
318            .unwrap_or(DEFAULT_MAX_MANAGED_STATE_BYTES);
319        validate_managed_state_max_bytes(max_managed_state_bytes)
320            .map_err(|error| DbError::Config(error.into()))?;
321        self.pipeline_max_managed_state_bytes = Some(max_managed_state_bytes);
322
323        if let Some(checkpoint) = self.checkpoint.as_mut() {
324            let max_node_data_bytes = checkpoint.max_node_data_bytes.unwrap_or(
325                laminar_core::checkpoint::checkpoint_store::DEFAULT_MAX_CHECKPOINT_NODE_DATA_BYTES,
326            );
327            laminar_core::checkpoint::checkpoint_store::validate_max_checkpoint_node_data_bytes(
328                max_node_data_bytes,
329            )
330            .map_err(|error| DbError::Config(format!("checkpoint.max_node_data_bytes: {error}")))?;
331            checkpoint.max_node_data_bytes = Some(max_node_data_bytes);
332        }
333
334        self.validate_backpressure_policy()
335    }
336
337    pub(crate) fn validate_backpressure_policy(&self) -> Result<(), DbError> {
338        if self.pipeline_max_input_buf_bytes == Some(0) {
339            return Err(DbError::Config(
340                "pipeline_max_input_buf_bytes must be greater than zero when configured".into(),
341            ));
342        }
343        let policy = self.pipeline_backpressure_policy;
344        if policy == BackpressurePolicy::Backpressure {
345            return Ok(());
346        }
347
348        let has_count_cap = self.pipeline_max_input_buf_batches.is_none_or(|c| c > 0);
349        let has_byte_cap = self.pipeline_max_input_buf_bytes.is_some_and(|b| b > 0);
350        if !has_count_cap && !has_byte_cap {
351            return Err(DbError::Config(format!(
352                "backpressure_policy={policy:?} requires at least one of \
353                 pipeline_max_input_buf_batches (>0) or pipeline_max_input_buf_bytes"
354            )));
355        }
356
357        if policy == BackpressurePolicy::ShedOldest
358            && self.delivery_guarantee != DeliveryGuarantee::BestEffort
359        {
360            return Err(DbError::Config(
361                "ShedOldest drops data and supports BestEffort only; at-least-once and \
362                 exactly-once delivery require Backpressure or Fail."
363                    .into(),
364            ));
365        }
366        Ok(())
367    }
368}
369
370impl Default for LaminarConfig {
371    fn default() -> Self {
372        Self {
373            reference_table_max_rows: DEFAULT_REFERENCE_TABLE_MAX_ROWS,
374            reference_table_max_bytes: DEFAULT_REFERENCE_TABLE_MAX_BYTES,
375            materialized_view_max_rows: DEFAULT_MATERIALIZED_VIEW_MAX_ROWS,
376            materialized_view_max_bytes: DEFAULT_MATERIALIZED_VIEW_MAX_BYTES,
377            materialized_view_read_cache_bytes: DEFAULT_MATERIALIZED_VIEW_READ_CACHE_BYTES,
378            datafusion_memory_limit_bytes: DEFAULT_DATAFUSION_MEMORY_LIMIT_BYTES,
379            default_buffer_size: 65536,
380            push_source_max_bytes: laminar_core::streaming::DEFAULT_SOURCE_MAX_QUEUED_BYTES,
381            default_backpressure: BackpressureStrategy::Block,
382            storage_dir: None,
383            checkpoint: None,
384            object_store_url: None,
385            object_store_options: HashMap::new(),
386            http_auth_token: None,
387            delivery_guarantee: DeliveryGuarantee::default(),
388            pipeline_channel_capacity: None,
389            source_queue_max_bytes: DEFAULT_SOURCE_QUEUE_MAX_BYTES,
390            pipeline_batch_window: None,
391            pipeline_drain_budget_ns: None,
392            pipeline_query_budget_ns: None,
393            pipeline_max_input_buf_batches: None,
394            pipeline_max_input_buf_bytes: None,
395            pipeline_max_managed_state_bytes: None,
396            temporal_join_idle_history_retention: None,
397            source_idle_timeout: None,
398            event_time_max_future_skew: std::time::Duration::from_millis(
399                laminar_core::time::DEFAULT_MAX_FUTURE_SKEW_MS.unsigned_abs(),
400            ),
401            pipeline_backpressure_policy: BackpressurePolicy::default(),
402            restart_policy: RestartPolicy::default(),
403            shared_source_isolation: false,
404        }
405    }
406}
407
408#[cfg(test)]
409mod tests {
410    #[tokio::test]
411    async fn materialized_view_limits_apply_through_config_and_builder() {
412        use super::*;
413        assert_eq!(
414            LaminarConfig::default().materialized_view_max_rows,
415            1_000_000
416        );
417        assert_eq!(
418            LaminarConfig::default().materialized_view_max_bytes,
419            256 * 1024 * 1024
420        );
421        for (rows, bytes) in [(0, 1024), (1, 0)] {
422            assert!(matches!(
423                crate::LaminarDB::open_with_config(LaminarConfig {
424                    materialized_view_max_rows: rows,
425                    materialized_view_max_bytes: bytes,
426                    ..Default::default()
427                }),
428                Err(DbError::Config(_))
429            ));
430            assert!(matches!(
431                crate::LaminarDB::builder()
432                    .materialized_view_max_rows(rows)
433                    .materialized_view_max_bytes(bytes)
434                    .build()
435                    .await,
436                Err(DbError::Config(_))
437            ));
438        }
439        let db = crate::LaminarDB::builder()
440            .materialized_view_max_rows(1)
441            .materialized_view_max_bytes(8192)
442            .build()
443            .await
444            .unwrap();
445        let batch = arrow::record_batch::RecordBatch::try_new(
446            std::sync::Arc::new(arrow::datatypes::Schema::new(vec![
447                arrow::datatypes::Field::new("id", arrow::datatypes::DataType::Int64, false),
448            ])),
449            vec![std::sync::Arc::new(arrow::array::Int64Array::from(vec![
450                1, 2,
451            ]))],
452        )
453        .unwrap();
454        let mut store = db.mv_store.write();
455        store
456            .create_mv(
457                "v",
458                batch.schema(),
459                crate::mv_store::MvStorageMode::append_rows(1_000),
460            )
461            .unwrap();
462        assert!(matches!(
463            store.update_cycle("v", &[batch]),
464            Err(DbError::MaterializedViewQuotaExceeded { max_rows: 1, .. })
465        ));
466    }
467
468    use super::*;
469
470    #[tokio::test]
471    async fn reference_table_limits_apply_through_config_and_builder() {
472        assert_eq!(LaminarConfig::default().reference_table_max_rows, 1_000_000);
473        assert_eq!(
474            LaminarConfig::default().reference_table_max_bytes,
475            256 * 1024 * 1024
476        );
477        for (rows, bytes) in [(0, 1024), (1, 0)] {
478            let error = crate::LaminarDB::open_with_config(LaminarConfig {
479                reference_table_max_rows: rows,
480                reference_table_max_bytes: bytes,
481                ..Default::default()
482            })
483            .err()
484            .unwrap();
485            assert!(matches!(error, DbError::Config(_)));
486            let error = crate::LaminarDB::builder()
487                .reference_table_max_rows(rows)
488                .reference_table_max_bytes(bytes)
489                .build()
490                .await
491                .err()
492                .unwrap();
493            assert!(matches!(error, DbError::Config(_)));
494        }
495        let db = crate::LaminarDB::builder()
496            .reference_table_max_rows(1)
497            .reference_table_max_bytes(8192)
498            .build()
499            .await
500            .unwrap();
501        db.execute("CREATE TABLE dimensions (id BIGINT PRIMARY KEY, value VARCHAR)")
502            .await
503            .unwrap();
504        db.execute("INSERT INTO dimensions VALUES (1, 'first')")
505            .await
506            .unwrap();
507        assert!(matches!(
508            db.execute("INSERT INTO dimensions VALUES (1, 'changed'), (2, 'new')")
509                .await,
510            Err(DbError::ReferenceTableQuotaExceeded {
511                rows: 2,
512                max_rows: 1,
513                ..
514            })
515        ));
516        let result = db.execute("SELECT value FROM dimensions").await.unwrap();
517        let crate::ExecuteResult::Query(mut query) = result else {
518            panic!("expected query");
519        };
520        let mut rows = query.subscribe_raw().unwrap();
521        let output = tokio::time::timeout(std::time::Duration::from_secs(5), rows.next_batch())
522            .await
523            .unwrap()
524            .unwrap()
525            .unwrap();
526        let value = output
527            .column(0)
528            .as_any()
529            .downcast_ref::<arrow::array::StringArray>()
530            .unwrap();
531        assert_eq!(value.value(0), "first");
532    }
533
534    #[tokio::test]
535    async fn push_source_limits_are_validated_at_db_entry() {
536        assert_eq!(
537            LaminarConfig::default().push_source_max_bytes,
538            64 * 1024 * 1024
539        );
540        for bytes in [0, laminar_core::streaming::MAX_SOURCE_QUEUED_BYTES + 1] {
541            let error = crate::LaminarDB::open_with_config(LaminarConfig {
542                push_source_max_bytes: bytes,
543                ..Default::default()
544            })
545            .err()
546            .unwrap();
547            assert!(
548                matches!(error, DbError::Config(ref message) if message.contains("push_source_max_bytes"))
549            );
550            let error = crate::LaminarDB::builder()
551                .push_source_max_bytes(bytes)
552                .build()
553                .await
554                .err()
555                .unwrap();
556            assert!(
557                matches!(error, DbError::Config(ref message) if message.contains("push_source_max_bytes"))
558            );
559        }
560    }
561
562    #[tokio::test]
563    async fn source_queue_limits_are_validated_at_db_entry() {
564        assert_eq!(
565            LaminarConfig::default().source_queue_max_bytes,
566            64 * 1024 * 1024
567        );
568        for bytes in [0, MAX_SOURCE_QUEUE_BYTES + 1] {
569            let error = crate::LaminarDB::open_with_config(LaminarConfig {
570                source_queue_max_bytes: bytes,
571                ..Default::default()
572            })
573            .err()
574            .unwrap();
575            assert!(
576                matches!(error, DbError::Config(ref message) if message.contains("source_queue_max_bytes"))
577            );
578            let error = crate::LaminarDB::builder()
579                .source_queue_max_bytes(bytes)
580                .build()
581                .await
582                .err()
583                .unwrap();
584            assert!(
585                matches!(error, DbError::Config(ref message) if message.contains("source_queue_max_bytes"))
586            );
587        }
588        for bytes in [1, MAX_SOURCE_QUEUE_BYTES] {
589            assert!(validate_source_queue_max_bytes(bytes).is_ok());
590        }
591    }
592}
laminar_db/config.rs

laminar_db/
config.rs
