laminar_db/pipeline/
config.rs

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

laminar_db/pipeline/
config.rs
