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    /// Compiled data contract (issue #204). The pass runs per page after the
21    /// quality pass and before the schema-drift pass.
22    #[cfg(feature = "contract")]
23    pub contract: Option<std::sync::Arc<crate::contract::CompiledContract>>,
24    /// Compiled masking policy (issue #206). The pass runs per page *first* —
25    /// before quality/contract/drift and every sink write — so PII never
26    /// reaches a sink, the DLQ, or a lineage sample unmasked.
27    #[cfg(feature = "masking")]
28    pub masking: Option<std::sync::Arc<crate::masking::CompiledMasking>>,
29    /// Adaptive batch-size controller config; `None` (or `enabled = false`)
30    /// leaves the per-page write path unchanged.
31    pub adaptive: Option<crate::adaptive::AdaptiveBatchConfig>,
32    /// Cooperative cancellation. When set and cancelled mid-run, the streaming
33    /// loop stops polling new pages, **flushes the sinks** (so a buffered sink
34    /// like Parquet writes its footer / completes its upload rather than
35    /// orphaning the file), and returns the partial result. Without this, a
36    /// dropped run future loses everything written-but-unflushed (#146 H16).
37    pub cancel: Option<CancellationToken>,
38    /// Delivery guarantee. `AtLeastOnce` (default) leaves the write path
39    /// unchanged. `ExactlyOnce` enables the resume/skip + atomic-token path.
40    pub delivery: crate::idempotency::DeliveryMode,
41    /// Resume sequence read from the (unwrapped) exactly-once state value.
42    /// Ignored unless `delivery == ExactlyOnce`. Defaults to 0.
43    pub start_seq: u64,
44    /// The source's replay capability, when the caller knows it.
45    /// `Some(NonDeterministic)` steers an `ExactlyOnce` run away from the
46    /// atomic-watermark skip path (whose correctness depends on positional
47    /// replay) and onto the keyed-upsert mechanism when the sink is configured
48    /// for it. `None` (default) trusts the caller — today's behaviour for
49    /// library users driving `run_stream` directly. `Pipeline::run` always
50    /// sets it from `Source::replay_guarantee()`.
51    pub replay: Option<crate::idempotency::ReplayGuarantee>,
52    /// Optional resilience policy (retry/backoff/circuit-breaker/poison).
53    pub resilience: Option<crate::resilience::ResiliencePolicy>,
54    /// Compiled schema-drift policy (issue #194). `None` → no drift handling.
55    pub schema_drift: Option<crate::drift::SchemaDriftPolicy>,
56}
57
58impl RunStreamOptions {
59    pub fn new() -> Self {
60        Self::default()
61    }
62
63    pub fn with_state(mut self, store: Arc<dyn StateStore>, key: impl Into<String>) -> Self {
64        self.state_store = Some(store);
65        self.state_key = Some(key.into());
66        self
67    }
68
69    pub fn with_name(mut self, name: impl Into<String>) -> Self {
70        self.pipeline_name = Some(name.into());
71        self
72    }
73
74    pub fn with_row(mut self, row: impl Into<String>) -> Self {
75        self.row = Some(row.into());
76        self
77    }
78
79    pub fn with_run_id(mut self, id: impl Into<String>) -> Self {
80        self.run_id = Some(id.into());
81        self
82    }
83
84    pub fn with_dlq(mut self, dlq: DlqConfig) -> Self {
85        self.dlq = Some(dlq);
86        self
87    }
88
89    /// Attach a cancellation token for cooperative, flush-completing cancel.
90    pub fn with_cancel(mut self, cancel: CancellationToken) -> Self {
91        self.cancel = Some(cancel);
92        self
93    }
94
95    /// Attach an adaptive batch-size controller config.
96    pub fn with_adaptive(mut self, cfg: crate::adaptive::AdaptiveBatchConfig) -> Self {
97        self.adaptive = Some(cfg);
98        self
99    }
100
101    #[cfg(feature = "quality")]
102    pub fn with_quality(
103        mut self,
104        quality: std::sync::Arc<crate::quality::CompiledQuality>,
105    ) -> Self {
106        self.quality = Some(quality);
107        self
108    }
109
110    /// Attach a compiled data contract. The pass runs per page after the
111    /// quality pass and before the schema-drift pass.
112    #[cfg(feature = "contract")]
113    pub fn with_contract(
114        mut self,
115        contract: std::sync::Arc<crate::contract::CompiledContract>,
116    ) -> Self {
117        self.contract = Some(contract);
118        self
119    }
120
121    /// Attach a compiled masking policy. The pass runs per page *first* —
122    /// before quality/contract/drift and every sink write.
123    #[cfg(feature = "masking")]
124    pub fn with_masking(
125        mut self,
126        masking: std::sync::Arc<crate::masking::CompiledMasking>,
127    ) -> Self {
128        self.masking = Some(masking);
129        self
130    }
131
132    /// Set the delivery mode.
133    pub fn with_delivery(mut self, mode: crate::idempotency::DeliveryMode) -> Self {
134        self.delivery = mode;
135        self
136    }
137
138    /// Set the resume sequence (exactly-once). Normally derived by
139    /// `Pipeline::run` from the unwrapped state value.
140    /// Declare the source's replay capability (see the `replay` field).
141    pub fn with_replay_guarantee(mut self, replay: crate::idempotency::ReplayGuarantee) -> Self {
142        self.replay = Some(replay);
143        self
144    }
145
146    pub fn with_start_seq(mut self, seq: u64) -> Self {
147        self.start_seq = seq;
148        self
149    }
150
151    /// Attach a resilience policy to the run.
152    pub fn with_resilience(mut self, policy: crate::resilience::ResiliencePolicy) -> Self {
153        self.resilience = Some(policy);
154        self
155    }
156
157    /// Attach a compiled schema-drift policy. The pass runs per page after the
158    /// quality pass and before the sink write.
159    pub fn with_schema_drift(mut self, policy: crate::drift::SchemaDriftPolicy) -> Self {
160        self.schema_drift = Some(policy);
161        self
162    }
163}
164
165#[cfg(test)]
166mod tests {
167    use super::RunStreamOptions;
168    use crate::idempotency::{DeliveryMode, ReplayGuarantee};
169    use crate::state::MemoryStateStore;
170    use std::sync::Arc;
171    use tokio_util::sync::CancellationToken;
172
173    #[test]
174    fn default_is_empty_at_least_once() {
175        let o = RunStreamOptions::new();
176        assert!(o.pipeline_name.is_none());
177        assert!(o.row.is_none());
178        assert!(o.run_id.is_none());
179        assert!(o.state_store.is_none());
180        assert!(o.state_key.is_none());
181        assert!(o.cancel.is_none());
182        assert_eq!(o.delivery, DeliveryMode::AtLeastOnce);
183        assert_eq!(o.start_seq, 0);
184        assert!(o.replay.is_none());
185        assert!(o.resilience.is_none());
186    }
187
188    #[test]
189    fn each_builder_sets_only_its_own_field() {
190        let o = RunStreamOptions::new().with_name("p");
191        assert_eq!(o.pipeline_name.as_deref(), Some("p"));
192        assert!(o.row.is_none(), "with_name must not touch row");
193
194        let o = RunStreamOptions::new().with_row("r1");
195        assert_eq!(o.row.as_deref(), Some("r1"));
196        assert!(o.pipeline_name.is_none());
197
198        let o = RunStreamOptions::new().with_run_id("run-42");
199        assert_eq!(o.run_id.as_deref(), Some("run-42"));
200
201        let o = RunStreamOptions::new().with_state(Arc::new(MemoryStateStore::new()), "k");
202        assert!(o.state_store.is_some());
203        assert_eq!(o.state_key.as_deref(), Some("k"));
204
205        let o = RunStreamOptions::new().with_cancel(CancellationToken::new());
206        assert!(o.cancel.is_some());
207
208        let o = RunStreamOptions::new().with_delivery(DeliveryMode::ExactlyOnce);
209        assert_eq!(o.delivery, DeliveryMode::ExactlyOnce);
210
211        let o = RunStreamOptions::new().with_start_seq(7);
212        assert_eq!(o.start_seq, 7);
213
214        let o = RunStreamOptions::new().with_replay_guarantee(ReplayGuarantee::Deterministic);
215        assert_eq!(o.replay, Some(ReplayGuarantee::Deterministic));
216
217        let o = RunStreamOptions::new().with_resilience(Default::default());
218        assert!(o.resilience.is_some());
219    }
220
221    #[test]
222    fn chaining_composes_all_set_fields() {
223        let o = RunStreamOptions::new()
224            .with_name("pipe")
225            .with_row("row0")
226            .with_run_id("rid")
227            .with_state(Arc::new(MemoryStateStore::new()), "state-key")
228            .with_delivery(DeliveryMode::ExactlyOnce)
229            .with_start_seq(3);
230        assert_eq!(o.pipeline_name.as_deref(), Some("pipe"));
231        assert_eq!(o.row.as_deref(), Some("row0"));
232        assert_eq!(o.run_id.as_deref(), Some("rid"));
233        assert!(o.state_store.is_some());
234        assert_eq!(o.state_key.as_deref(), Some("state-key"));
235        assert_eq!(o.delivery, DeliveryMode::ExactlyOnce);
236        assert_eq!(o.start_seq, 3);
237    }
238}
faucet_core/observability/options.rs

faucet_core/observability/
options.rs
