faucet_core/observability/
options.rs

1//! Options struct passed to `run_stream`. Replaces the prior positional
2//! argument list so future observability additions don't keep breaking the
3//! function signature.
4
5use crate::dlq::DlqConfig;
6use crate::state::StateStore;
7use std::sync::Arc;
8use tokio_util::sync::CancellationToken;
9
10#[derive(Default, Clone)]
11pub struct RunStreamOptions {
12    pub state_store: Option<Arc<dyn StateStore>>,
13    pub state_key: Option<String>,
14    pub pipeline_name: Option<String>,
15    pub row: Option<String>,
16    pub run_id: Option<String>,
17    pub dlq: Option<DlqConfig>,
18    #[cfg(feature = "quality")]
19    pub quality: Option<std::sync::Arc<crate::quality::CompiledQuality>>,
20    /// Adaptive batch-size controller config; `None` (or `enabled = false`)
21    /// leaves the per-page write path unchanged.
22    pub adaptive: Option<crate::adaptive::AdaptiveBatchConfig>,
23    /// Cooperative cancellation. When set and cancelled mid-run, the streaming
24    /// loop stops polling new pages, **flushes the sinks** (so a buffered sink
25    /// like Parquet writes its footer / completes its upload rather than
26    /// orphaning the file), and returns the partial result. Without this, a
27    /// dropped run future loses everything written-but-unflushed (#146 H16).
28    pub cancel: Option<CancellationToken>,
29    /// Delivery guarantee. `AtLeastOnce` (default) leaves the write path
30    /// unchanged. `ExactlyOnce` enables the resume/skip + atomic-token path.
31    pub delivery: crate::idempotency::DeliveryMode,
32    /// Resume sequence read from the (unwrapped) exactly-once state value.
33    /// Ignored unless `delivery == ExactlyOnce`. Defaults to 0.
34    pub start_seq: u64,
35    /// Optional resilience policy (retry/backoff/circuit-breaker/poison).
36    pub resilience: Option<crate::resilience::ResiliencePolicy>,
37    /// Compiled schema-drift policy (issue #194). `None` → no drift handling.
38    pub schema_drift: Option<crate::drift::SchemaDriftPolicy>,
39}
40
41impl RunStreamOptions {
42    pub fn new() -> Self {
43        Self::default()
44    }
45
46    pub fn with_state(mut self, store: Arc<dyn StateStore>, key: impl Into<String>) -> Self {
47        self.state_store = Some(store);
48        self.state_key = Some(key.into());
49        self
50    }
51
52    pub fn with_name(mut self, name: impl Into<String>) -> Self {
53        self.pipeline_name = Some(name.into());
54        self
55    }
56
57    pub fn with_row(mut self, row: impl Into<String>) -> Self {
58        self.row = Some(row.into());
59        self
60    }
61
62    pub fn with_run_id(mut self, id: impl Into<String>) -> Self {
63        self.run_id = Some(id.into());
64        self
65    }
66
67    pub fn with_dlq(mut self, dlq: DlqConfig) -> Self {
68        self.dlq = Some(dlq);
69        self
70    }
71
72    /// Attach a cancellation token for cooperative, flush-completing cancel.
73    pub fn with_cancel(mut self, cancel: CancellationToken) -> Self {
74        self.cancel = Some(cancel);
75        self
76    }
77
78    /// Attach an adaptive batch-size controller config.
79    pub fn with_adaptive(mut self, cfg: crate::adaptive::AdaptiveBatchConfig) -> Self {
80        self.adaptive = Some(cfg);
81        self
82    }
83
84    #[cfg(feature = "quality")]
85    pub fn with_quality(
86        mut self,
87        quality: std::sync::Arc<crate::quality::CompiledQuality>,
88    ) -> Self {
89        self.quality = Some(quality);
90        self
91    }
92
93    /// Set the delivery mode.
94    pub fn with_delivery(mut self, mode: crate::idempotency::DeliveryMode) -> Self {
95        self.delivery = mode;
96        self
97    }
98
99    /// Set the resume sequence (exactly-once). Normally derived by
100    /// `Pipeline::run` from the unwrapped state value.
101    pub fn with_start_seq(mut self, seq: u64) -> Self {
102        self.start_seq = seq;
103        self
104    }
105
106    /// Attach a resilience policy to the run.
107    pub fn with_resilience(mut self, policy: crate::resilience::ResiliencePolicy) -> Self {
108        self.resilience = Some(policy);
109        self
110    }
111
112    /// Attach a compiled schema-drift policy. The pass runs per page after the
113    /// quality pass and before the sink write.
114    pub fn with_schema_drift(mut self, policy: crate::drift::SchemaDriftPolicy) -> Self {
115        self.schema_drift = Some(policy);
116        self
117    }
118}
faucet_core/observability/options.rs

faucet_core/observability/
options.rs
