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    /// Optional resilience policy (retry/backoff/circuit-breaker/poison).
45    pub resilience: Option<crate::resilience::ResiliencePolicy>,
46    /// Compiled schema-drift policy (issue #194). `None` → no drift handling.
47    pub schema_drift: Option<crate::drift::SchemaDriftPolicy>,
48}
49
50impl RunStreamOptions {
51    pub fn new() -> Self {
52        Self::default()
53    }
54
55    pub fn with_state(mut self, store: Arc<dyn StateStore>, key: impl Into<String>) -> Self {
56        self.state_store = Some(store);
57        self.state_key = Some(key.into());
58        self
59    }
60
61    pub fn with_name(mut self, name: impl Into<String>) -> Self {
62        self.pipeline_name = Some(name.into());
63        self
64    }
65
66    pub fn with_row(mut self, row: impl Into<String>) -> Self {
67        self.row = Some(row.into());
68        self
69    }
70
71    pub fn with_run_id(mut self, id: impl Into<String>) -> Self {
72        self.run_id = Some(id.into());
73        self
74    }
75
76    pub fn with_dlq(mut self, dlq: DlqConfig) -> Self {
77        self.dlq = Some(dlq);
78        self
79    }
80
81    /// Attach a cancellation token for cooperative, flush-completing cancel.
82    pub fn with_cancel(mut self, cancel: CancellationToken) -> Self {
83        self.cancel = Some(cancel);
84        self
85    }
86
87    /// Attach an adaptive batch-size controller config.
88    pub fn with_adaptive(mut self, cfg: crate::adaptive::AdaptiveBatchConfig) -> Self {
89        self.adaptive = Some(cfg);
90        self
91    }
92
93    #[cfg(feature = "quality")]
94    pub fn with_quality(
95        mut self,
96        quality: std::sync::Arc<crate::quality::CompiledQuality>,
97    ) -> Self {
98        self.quality = Some(quality);
99        self
100    }
101
102    /// Attach a compiled data contract. The pass runs per page after the
103    /// quality pass and before the schema-drift pass.
104    #[cfg(feature = "contract")]
105    pub fn with_contract(
106        mut self,
107        contract: std::sync::Arc<crate::contract::CompiledContract>,
108    ) -> Self {
109        self.contract = Some(contract);
110        self
111    }
112
113    /// Attach a compiled masking policy. The pass runs per page *first* —
114    /// before quality/contract/drift and every sink write.
115    #[cfg(feature = "masking")]
116    pub fn with_masking(
117        mut self,
118        masking: std::sync::Arc<crate::masking::CompiledMasking>,
119    ) -> Self {
120        self.masking = Some(masking);
121        self
122    }
123
124    /// Set the delivery mode.
125    pub fn with_delivery(mut self, mode: crate::idempotency::DeliveryMode) -> Self {
126        self.delivery = mode;
127        self
128    }
129
130    /// Set the resume sequence (exactly-once). Normally derived by
131    /// `Pipeline::run` from the unwrapped state value.
132    pub fn with_start_seq(mut self, seq: u64) -> Self {
133        self.start_seq = seq;
134        self
135    }
136
137    /// Attach a resilience policy to the run.
138    pub fn with_resilience(mut self, policy: crate::resilience::ResiliencePolicy) -> Self {
139        self.resilience = Some(policy);
140        self
141    }
142
143    /// Attach a compiled schema-drift policy. The pass runs per page after the
144    /// quality pass and before the sink write.
145    pub fn with_schema_drift(mut self, policy: crate::drift::SchemaDriftPolicy) -> Self {
146        self.schema_drift = Some(policy);
147        self
148    }
149}
faucet_core/observability/options.rs

faucet_core/observability/
options.rs
