laminar_db/pipeline/
config.rs

1//! Pipeline configuration.
2
3use std::time::Duration;
4
5use laminar_connectors::connector::DeliveryGuarantee;
6
7use crate::config::BackpressurePolicy;
8
9/// Checkpoint triggering configured for the running pipeline.
10///
11/// The three states mirror [`LaminarConfig::checkpoint`](crate::config::LaminarConfig::checkpoint)
12/// without allowing an enabled flag and timer interval to contradict each other.
13#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
14pub enum CheckpointSchedule {
15    /// No durable checkpoint service is configured.
16    #[default]
17    Disabled,
18    /// Checkpoints run only when explicitly requested.
19    Manual,
20    /// Checkpoints wait at least this long after a terminal outcome before the next automatic
21    /// admission; they may also be requested explicitly.
22    Periodic(Duration),
23}
24
25impl CheckpointSchedule {
26    /// Whether the durable checkpoint service is configured.
27    #[must_use]
28    pub const fn is_enabled(self) -> bool {
29        !matches!(self, Self::Disabled)
30    }
31
32    /// The periodic cadence, if automatic checkpoints are configured.
33    #[must_use]
34    pub const fn periodic_interval(self) -> Option<Duration> {
35        match self {
36            Self::Periodic(interval) => Some(interval),
37            Self::Disabled | Self::Manual => None,
38        }
39    }
40}
41
42/// Configuration for the event-driven connector pipeline.
43#[derive(Debug, Clone)]
44pub struct PipelineConfig {
45    /// Maximum records per `poll_batch()` call.
46    pub max_poll_records: usize,
47
48    /// Channel capacity for per-source `mpsc` sender → coordinator.
49    pub channel_capacity: usize,
50
51    /// Fallback poll interval when a source returns no `data_ready_notify`.
52    pub fallback_poll_interval: Duration,
53
54    /// Whether checkpoints are disabled, manual-only, or periodic.
55    pub checkpoint_schedule: CheckpointSchedule,
56
57    /// Sleep after the first event in a cycle to let more data accumulate.
58    ///
59    /// Bounds SQL executions per second without sacrificing data. The bounded
60    /// channel provides natural backpressure during the window. `ZERO` = no batching.
61    pub batch_window: Duration,
62
63    /// Sole checkpoint-derived control-plane budget. Each connector startup stage and checkpoint
64    /// barrier creates one absolute deadline from this duration; individual connectors cannot
65    /// reset it.
66    pub checkpoint_timeout: Duration,
67
68    /// End-to-end delivery guarantee for the pipeline.
69    pub delivery_guarantee: DeliveryGuarantee,
70
71    /// Maximum wall-clock time for a single processing cycle (nanoseconds). Default: 10ms.
72    pub cycle_budget_ns: u64,
73
74    /// Maximum wall-clock time for the message drain phase (nanoseconds). Default: 1ms.
75    pub drain_budget_ns: u64,
76
77    /// Maximum wall-clock time for per-query execution within a cycle (nanoseconds). Default: 8ms.
78    pub query_budget_ns: u64,
79
80    /// Maximum wall-clock time for background work after SQL execution (nanoseconds). Default: 5ms.
81    pub background_budget_ns: u64,
82
83    /// Per-input-port batch cap. Default: 256.
84    pub max_input_buf_batches: usize,
85
86    /// Per-input-port byte cap. `None` = disabled.
87    pub max_input_buf_bytes: Option<usize>,
88
89    /// What to do when either cap is exceeded.
90    pub backpressure_policy: BackpressurePolicy,
91
92    /// Isolate queries that share a source into independent failure domains.
93    /// Off: they fault and recover together; on: a fault holds back only its own offset.
94    pub shared_source_isolation: bool,
95
96    /// Per-source cap on the in-memory replay buffer used by `shared_source_isolation`.
97    /// On overflow the engine falls back to whole-pipeline recovery. Ignored when the
98    /// flag is off.
99    pub max_replay_buffer_bytes: usize,
100}
101
102impl Default for PipelineConfig {
103    fn default() -> Self {
104        Self {
105            max_poll_records: 1024,
106            channel_capacity: 64,
107            fallback_poll_interval: Duration::from_millis(10),
108            checkpoint_schedule: CheckpointSchedule::Disabled,
109            batch_window: Duration::from_millis(5),
110            checkpoint_timeout: Duration::from_secs(30),
111            delivery_guarantee: DeliveryGuarantee::default(),
112            cycle_budget_ns: 10_000_000,     // 10ms
113            drain_budget_ns: 1_000_000,      // 1ms
114            query_budget_ns: 8_000_000,      // 8ms
115            background_budget_ns: 5_000_000, // 5ms
116            max_input_buf_batches: 256,
117            max_input_buf_bytes: None,
118            backpressure_policy: BackpressurePolicy::default(),
119            shared_source_isolation: false,
120            max_replay_buffer_bytes: 256 * 1024 * 1024, // 256 MiB per source
121        }
122    }
123}
124
125impl PipelineConfig {
126    /// Private rollback budget for a failed connector startup stage. This is deliberately fixed:
127    /// cleanup is fail-safe implementation policy, not another latency tuning dimension.
128    pub(crate) const CONNECTOR_STARTUP_CLEANUP_TIMEOUT: Duration = Duration::from_secs(15);
129}
laminar_db/pipeline/config.rs

laminar_db/pipeline/
config.rs
