faucet_core/
dlq.rs

1//! Dead-letter queue (DLQ) wiring shared by the pipeline runner.
2//!
3//! The types defined here are config-shaped: they describe *what* the
4//! pipeline should do with row-level failures, not *how* the routing is
5//! executed. The execution lives in [`run_stream`](crate::run_stream).
6//!
7//! See `docs/superpowers/specs/2026-05-24-dlq-design.md`.
8
9use crate::FaucetError;
10use crate::traits::Sink;
11use schemars::JsonSchema;
12use serde::{Deserialize, Serialize};
13use serde_json::{Value, json};
14use std::fmt;
15use std::sync::Arc;
16use std::time::{SystemTime, UNIX_EPOCH};
17
18/// Policy applied when a sink reports an outer failure (the whole batch
19/// failed, no per-row info).
20#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
21#[serde(rename_all = "snake_case")]
22pub enum OnBatchError {
23    /// Surface the underlying [`FaucetError`] and fail the pipeline (default).
24    #[default]
25    Propagate,
26    /// Treat every row in the failed page as a DLQ candidate. Unsafe with
27    /// best-effort APIs that haven't overridden
28    /// [`Sink::write_batch_partial`] — already-committed rows would land in
29    /// the DLQ as duplicates. Use with atomic sinks (single-statement
30    /// INSERT, file writes) where the failure mode is "nothing landed".
31    DlqAll,
32}
33
34/// Pipeline-level DLQ wiring.
35#[derive(Clone)]
36pub struct DlqConfig {
37    /// Sink that receives DLQ envelopes.
38    pub sink: Arc<dyn Sink>,
39    /// What to do when the main sink fails wholesale.
40    pub on_batch_error: OnBatchError,
41    /// Per-page failure budget. `None` = unlimited.
42    ///
43    /// This budget is **shared across both sink-side row failures and
44    /// quality-check quarantines**: a record routed to the DLQ by a
45    /// `quarantine` quality check counts against it just as a sink-side
46    /// row failure does.
47    pub max_failures_per_page: Option<usize>,
48    /// Cumulative failure budget across the run. `None` = unlimited.
49    ///
50    /// This budget is **shared across both sink-side row failures and
51    /// quality-check quarantines**: records quarantined by the quality pass
52    /// accumulate in this counter alongside sink-side failures.
53    pub max_failures_total: Option<usize>,
54    /// Always `true` in v1. Reserved for a future "headers-only" mode.
55    pub include_original_payload: bool,
56}
57
58impl DlqConfig {
59    /// Convenience constructor: `propagate` policy, no budgets, payload
60    /// included.
61    pub fn new(sink: Arc<dyn Sink>) -> Self {
62        Self {
63            sink,
64            on_batch_error: OnBatchError::Propagate,
65            max_failures_per_page: None,
66            max_failures_total: None,
67            include_original_payload: true,
68        }
69    }
70}
71
72impl fmt::Debug for DlqConfig {
73    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
74        f.debug_struct("DlqConfig")
75            .field("sink", &self.sink.connector_name())
76            .field("on_batch_error", &self.on_batch_error)
77            .field("max_failures_per_page", &self.max_failures_per_page)
78            .field("max_failures_total", &self.max_failures_total)
79            .field("include_original_payload", &self.include_original_payload)
80            .finish()
81    }
82}
83
84/// Counters returned alongside [`PipelineResult`](crate::PipelineResult)
85/// when a DLQ is wired.
86#[derive(Debug, Clone, Default, PartialEq, Eq)]
87pub struct DlqStats {
88    /// Total rows routed to the DLQ across the run.
89    pub records_dlq: usize,
90    /// Pages that produced at least one DLQ record.
91    pub pages_with_failures: usize,
92}
93
94/// Reason a page produced DLQ traffic. Used as a metric label and span
95/// attribute; closed-set enum so cardinality stays bounded.
96#[derive(Debug, Clone, Copy, PartialEq, Eq)]
97pub enum DlqReason {
98    /// At least one per-row outcome was `Err`, surfaced by an overriding
99    /// [`Sink::write_batch_partial`].
100    Partial,
101    /// The whole batch failed and the configured policy was
102    /// [`OnBatchError::DlqAll`].
103    DlqAll,
104    /// A record was quarantined (or batch-quarantined) by a data-quality check.
105    Quality,
106    /// A record was routed to the DLQ by an `on_drift`/`on_incompatible`
107    /// quarantine policy.
108    SchemaDrift,
109}
110
111impl DlqReason {
112    /// Returns the stable Prometheus label value for this reason.
113    /// Closed-set values: `"partial"`, `"dlq_all"`, or `"quality"`.
114    pub fn as_str(self) -> &'static str {
115        match self {
116            DlqReason::Partial => "partial",
117            DlqReason::DlqAll => "dlq_all",
118            DlqReason::Quality => "quality",
119            DlqReason::SchemaDrift => "schema_drift",
120        }
121    }
122}
123
124/// Build a single DLQ envelope.
125///
126/// The schema is fixed; see the design spec for the rationale. `payload`
127/// is included verbatim — no truncation, no transformation.
128pub fn build_envelope(
129    payload: &Value,
130    error: &FaucetError,
131    sink_name: &str,
132    pipeline_name: &str,
133    row: &str,
134    record_index: usize,
135) -> Value {
136    let kind = crate::observability::decorator::error_kind(error);
137    let message = error.to_string();
138    // `as_millis()` returns u128. Convert via TryFrom so we saturate at
139    // i64::MAX instead of silently wrapping to a negative number. The
140    // saturation ceiling (year ~292,000,000) is impossible in practice,
141    // so this only ever fires on a corrupt clock. `unwrap_or(0)` covers
142    // the (also impossible on modern systems) clock-before-epoch case.
143    let ts_ms = SystemTime::now()
144        .duration_since(UNIX_EPOCH)
145        .map(|d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
146        .unwrap_or(0);
147    json!({
148        "error": { "kind": kind, "message": message },
149        "payload": payload,
150        "ts_ms": ts_ms,
151        "sink": sink_name,
152        "pipeline": pipeline_name,
153        "row": row,
154        "record_index": record_index,
155    })
156}
157
158#[cfg(test)]
159mod tests {
160    use super::*;
161
162    #[test]
163    fn envelope_has_all_required_fields() {
164        let payload = json!({"user_id": 7, "name": "Alice"});
165        let err = FaucetError::Sink("row rejected: bad timestamp".into());
166        let env = build_envelope(&payload, &err, "bigquery", "users_etl", "us", 3);
167
168        assert_eq!(env["error"]["kind"], "Sink");
169        assert!(
170            env["error"]["message"]
171                .as_str()
172                .unwrap()
173                .contains("row rejected")
174        );
175        assert_eq!(env["payload"], payload);
176        assert!(env["ts_ms"].as_i64().unwrap() > 0);
177        assert_eq!(env["sink"], "bigquery");
178        assert_eq!(env["pipeline"], "users_etl");
179        assert_eq!(env["row"], "us");
180        assert_eq!(env["record_index"], 3);
181    }
182
183    #[test]
184    fn envelope_preserves_payload_byte_for_byte() {
185        let payload = json!({
186            "nested": { "a": [1, 2, 3], "b": null, "c": true },
187            "unicode": "café — résumé"
188        });
189        let env = build_envelope(&payload, &FaucetError::Sink("x".into()), "s", "p", "", 0);
190        assert_eq!(env["payload"], payload);
191    }
192
193    #[test]
194    fn envelope_empty_row_serializes_as_empty_string() {
195        let env = build_envelope(&json!({}), &FaucetError::Sink("x".into()), "s", "", "", 0);
196        assert_eq!(env["row"], "");
197        assert_eq!(env["pipeline"], "");
198    }
199
200    #[test]
201    fn on_batch_error_defaults_to_propagate() {
202        assert_eq!(OnBatchError::default(), OnBatchError::Propagate);
203    }
204
205    #[test]
206    fn on_batch_error_serializes_snake_case() {
207        let prop = serde_json::to_string(&OnBatchError::Propagate).unwrap();
208        let all = serde_json::to_string(&OnBatchError::DlqAll).unwrap();
209        assert_eq!(prop, "\"propagate\"");
210        assert_eq!(all, "\"dlq_all\"");
211    }
212
213    #[test]
214    fn on_batch_error_deserializes_snake_case() {
215        let prop: OnBatchError = serde_json::from_str("\"propagate\"").unwrap();
216        let all: OnBatchError = serde_json::from_str("\"dlq_all\"").unwrap();
217        assert_eq!(prop, OnBatchError::Propagate);
218        assert_eq!(all, OnBatchError::DlqAll);
219    }
220
221    #[test]
222    fn dlq_reason_strings() {
223        assert_eq!(DlqReason::Partial.as_str(), "partial");
224        assert_eq!(DlqReason::DlqAll.as_str(), "dlq_all");
225    }
226
227    #[test]
228    fn dlq_reason_quality_string() {
229        assert_eq!(DlqReason::Quality.as_str(), "quality");
230    }
231
232    #[test]
233    fn dlq_reason_schema_drift_string() {
234        assert_eq!(DlqReason::SchemaDrift.as_str(), "schema_drift");
235    }
236}
faucet_core/dlq.rs

faucet_core/
dlq.rs
