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 normally wait this long after a terminal outcome. Bounded durable-output
21    /// pressure may accelerate the next automatic admission; requests may also be explicit.
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    /// Per-input-port batch cap. Default: 256.
81    pub max_input_buf_batches: usize,
82
83    /// Per-input-port byte cap. `None` = disabled.
84    pub max_input_buf_bytes: Option<usize>,
85
86    /// What to do when either cap is exceeded.
87    pub backpressure_policy: BackpressurePolicy,
88
89    /// Isolate queries that share a source into independent failure domains.
90    /// Off: they fault and recover together; on: a fault holds back only its own offset.
91    pub shared_source_isolation: bool,
92
93    /// Per-source cap on the in-memory replay buffer used by `shared_source_isolation`.
94    /// On overflow the engine falls back to whole-pipeline recovery. Ignored when the
95    /// flag is off.
96    pub max_replay_buffer_bytes: usize,
97}
98
99impl Default for PipelineConfig {
100    fn default() -> Self {
101        Self {
102            max_poll_records: 1024,
103            channel_capacity: 64,
104            fallback_poll_interval: Duration::from_millis(10),
105            checkpoint_schedule: CheckpointSchedule::Disabled,
106            batch_window: Duration::from_millis(5),
107            checkpoint_timeout: Duration::from_secs(30),
108            delivery_guarantee: DeliveryGuarantee::default(),
109            cycle_budget_ns: 10_000_000, // 10ms
110            drain_budget_ns: 1_000_000,  // 1ms
111            query_budget_ns: 8_000_000,  // 8ms
112            max_input_buf_batches: 256,
113            max_input_buf_bytes: None,
114            backpressure_policy: BackpressurePolicy::default(),
115            shared_source_isolation: false,
116            max_replay_buffer_bytes: 256 * 1024 * 1024, // 256 MiB per source
117        }
118    }
119}
120
121impl PipelineConfig {
122    /// Private rollback budget for a failed connector startup stage. This is deliberately fixed:
123    /// cleanup is fail-safe implementation policy, not another latency tuning dimension.
124    pub(crate) const CONNECTOR_STARTUP_CLEANUP_TIMEOUT: Duration = Duration::from_secs(15);
125}
laminar_db/pipeline/config.rs

laminar_db/pipeline/
config.rs
