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    /// A record was routed to the DLQ by a data-contract `on_breach:
110    /// quarantine` policy.
111    Contract,
112}
113
114impl DlqReason {
115    /// Returns the stable Prometheus label value for this reason.
116    /// Closed-set values: `"partial"`, `"dlq_all"`, or `"quality"`.
117    pub fn as_str(self) -> &'static str {
118        match self {
119            DlqReason::Partial => "partial",
120            DlqReason::DlqAll => "dlq_all",
121            DlqReason::Quality => "quality",
122            DlqReason::SchemaDrift => "schema_drift",
123            DlqReason::Contract => "contract",
124        }
125    }
126
127    /// Every closed-set reason value, for validating a user-supplied
128    /// `--reason` filter against the exact serde strings.
129    pub const ALL: [DlqReason; 5] = [
130        DlqReason::Partial,
131        DlqReason::DlqAll,
132        DlqReason::Quality,
133        DlqReason::SchemaDrift,
134        DlqReason::Contract,
135    ];
136
137    /// Parse a reason from its stable serde string (the inverse of
138    /// [`as_str`](Self::as_str)). Returns `None` for an unknown value.
139    pub fn from_serde_str(s: &str) -> Option<DlqReason> {
140        DlqReason::ALL.into_iter().find(|r| r.as_str() == s)
141    }
142}
143
144/// Build a single DLQ envelope.
145///
146/// The schema is fixed; see the design spec for the rationale. `payload`
147/// is included verbatim — no truncation, no transformation. `reason`
148/// records *which stage* quarantined the row (as the closed-set
149/// [`DlqReason`] serde value) so tools like `faucet dlq inspect` /
150/// `faucet dlq replay` can group and filter without re-deriving it from
151/// the free-form error message. It is written as a top-level `reason`
152/// field alongside the structured `error`.
153pub fn build_envelope(
154    payload: &Value,
155    error: &FaucetError,
156    reason: DlqReason,
157    sink_name: &str,
158    pipeline_name: &str,
159    row: &str,
160    record_index: usize,
161) -> Value {
162    let kind = crate::observability::decorator::error_kind(error);
163    let message = error.to_string();
164    // `as_millis()` returns u128. Convert via TryFrom so we saturate at
165    // i64::MAX instead of silently wrapping to a negative number. The
166    // saturation ceiling (year ~292,000,000) is impossible in practice,
167    // so this only ever fires on a corrupt clock. `unwrap_or(0)` covers
168    // the (also impossible on modern systems) clock-before-epoch case.
169    let ts_ms = SystemTime::now()
170        .duration_since(UNIX_EPOCH)
171        .map(|d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
172        .unwrap_or(0);
173    json!({
174        "error": { "kind": kind, "message": message },
175        "reason": reason.as_str(),
176        "payload": payload,
177        "ts_ms": ts_ms,
178        "sink": sink_name,
179        "pipeline": pipeline_name,
180        "row": row,
181        "record_index": record_index,
182    })
183}
184
185/// A DLQ envelope parsed back into its original payload plus the metadata
186/// needed to inspect and replay it. Produced by [`unwrap_envelope`].
187#[derive(Debug, Clone, PartialEq)]
188pub struct UnwrappedEnvelope {
189    /// The original record that was quarantined — replayed verbatim.
190    pub payload: Value,
191    /// The stage that quarantined the row (`build_envelope`'s `reason`
192    /// field). `None` for envelopes written before the field existed.
193    pub reason: Option<String>,
194    /// The [`FaucetError`] variant name (`error.kind`), e.g. `"Sink"`,
195    /// `"QualityFailure"`. `None` if the envelope omits it.
196    pub error_kind: Option<String>,
197    /// Human-readable failure message (`error.message`), if present.
198    pub error_message: Option<String>,
199    /// Position of the record within its original page.
200    pub record_index: Option<u64>,
201    /// Pipeline name that produced the envelope, if present.
202    pub pipeline: Option<String>,
203    /// Matrix row id that produced the envelope, if present.
204    pub row: Option<String>,
205    /// Sink name the record was destined for, if present.
206    pub sink: Option<String>,
207    /// Epoch-millis timestamp the envelope was written, if present.
208    pub ts_ms: Option<i64>,
209}
210
211/// Error returned by [`unwrap_envelope`] when a value is not a usable DLQ
212/// envelope. Only the *payload* is mandatory — every other field is
213/// optional so envelopes written by older versions still replay.
214#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
215pub enum EnvelopeError {
216    /// The value was not a JSON object.
217    #[error("DLQ envelope is not a JSON object")]
218    NotObject,
219    /// The mandatory `payload` field was absent — nothing to replay.
220    #[error("DLQ envelope has no `payload` field")]
221    MissingPayload,
222}
223
224/// Parse a DLQ envelope produced by [`build_envelope`] back into its
225/// original payload plus metadata.
226///
227/// Only `payload` is required; all other fields are optional so envelopes
228/// written before a field existed still round-trip (forward-compatible
229/// read). Callers reading a DLQ location back (e.g. `faucet dlq inspect`)
230/// should treat an [`EnvelopeError`] as "skip + count", never as fatal —
231/// a DLQ file may legitimately contain arbitrary lines.
232pub fn unwrap_envelope(value: &Value) -> Result<UnwrappedEnvelope, EnvelopeError> {
233    let obj = value.as_object().ok_or(EnvelopeError::NotObject)?;
234    let payload = obj.get("payload").ok_or(EnvelopeError::MissingPayload)?;
235    let error = obj.get("error").and_then(|e| e.as_object());
236    let str_field = |k: &str| obj.get(k).and_then(|v| v.as_str()).map(str::to_owned);
237    Ok(UnwrappedEnvelope {
238        payload: payload.clone(),
239        reason: str_field("reason"),
240        error_kind: error
241            .and_then(|e| e.get("kind"))
242            .and_then(|v| v.as_str())
243            .map(str::to_owned),
244        error_message: error
245            .and_then(|e| e.get("message"))
246            .and_then(|v| v.as_str())
247            .map(str::to_owned),
248        record_index: obj.get("record_index").and_then(Value::as_u64),
249        pipeline: str_field("pipeline"),
250        row: str_field("row"),
251        sink: str_field("sink"),
252        ts_ms: obj.get("ts_ms").and_then(Value::as_i64),
253    })
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259
260    #[test]
261    fn envelope_has_all_required_fields() {
262        let payload = json!({"user_id": 7, "name": "Alice"});
263        let err = FaucetError::Sink("row rejected: bad timestamp".into());
264        let env = build_envelope(
265            &payload,
266            &err,
267            DlqReason::Partial,
268            "bigquery",
269            "users_etl",
270            "us",
271            3,
272        );
273
274        assert_eq!(env["error"]["kind"], "Sink");
275        assert_eq!(env["reason"], "partial");
276        assert!(
277            env["error"]["message"]
278                .as_str()
279                .unwrap()
280                .contains("row rejected")
281        );
282        assert_eq!(env["payload"], payload);
283        assert!(env["ts_ms"].as_i64().unwrap() > 0);
284        assert_eq!(env["sink"], "bigquery");
285        assert_eq!(env["pipeline"], "users_etl");
286        assert_eq!(env["row"], "us");
287        assert_eq!(env["record_index"], 3);
288    }
289
290    #[test]
291    fn envelope_preserves_payload_byte_for_byte() {
292        let payload = json!({
293            "nested": { "a": [1, 2, 3], "b": null, "c": true },
294            "unicode": "café — résumé"
295        });
296        let env = build_envelope(
297            &payload,
298            &FaucetError::Sink("x".into()),
299            DlqReason::Quality,
300            "s",
301            "p",
302            "",
303            0,
304        );
305        assert_eq!(env["payload"], payload);
306    }
307
308    #[test]
309    fn envelope_empty_row_serializes_as_empty_string() {
310        let env = build_envelope(
311            &json!({}),
312            &FaucetError::Sink("x".into()),
313            DlqReason::DlqAll,
314            "s",
315            "",
316            "",
317            0,
318        );
319        assert_eq!(env["row"], "");
320        assert_eq!(env["pipeline"], "");
321    }
322
323    #[test]
324    fn dlq_reason_from_serde_str_round_trips() {
325        for r in DlqReason::ALL {
326            assert_eq!(DlqReason::from_serde_str(r.as_str()), Some(r));
327        }
328        assert_eq!(DlqReason::from_serde_str("nope"), None);
329        assert_eq!(DlqReason::from_serde_str("sink_error"), None);
330    }
331
332    #[test]
333    fn unwrap_envelope_round_trips_build_envelope() {
334        let payload = json!({"id": 42, "name": "Zoe"});
335        let err = FaucetError::QualityFailure {
336            check: "not_null(email)".into(),
337            message: "email is null".into(),
338        };
339        let env = build_envelope(&payload, &err, DlqReason::Quality, "pg", "etl", "eu", 5);
340        let u = unwrap_envelope(&env).expect("valid envelope");
341        assert_eq!(u.payload, payload);
342        assert_eq!(u.reason.as_deref(), Some("quality"));
343        assert_eq!(u.error_kind.as_deref(), Some("QualityFailure"));
344        assert!(u.error_message.unwrap().contains("email is null"));
345        assert_eq!(u.record_index, Some(5));
346        assert_eq!(u.pipeline.as_deref(), Some("etl"));
347        assert_eq!(u.row.as_deref(), Some("eu"));
348        assert_eq!(u.sink.as_deref(), Some("pg"));
349        assert!(u.ts_ms.unwrap() > 0);
350    }
351
352    #[test]
353    fn unwrap_envelope_tolerates_legacy_envelope_without_reason() {
354        // An envelope written before `reason`/`error` existed still yields its
355        // payload; the missing metadata comes back as `None`, never a panic.
356        let legacy = json!({ "payload": { "x": 1 } });
357        let u = unwrap_envelope(&legacy).expect("payload present");
358        assert_eq!(u.payload, json!({ "x": 1 }));
359        assert_eq!(u.reason, None);
360        assert_eq!(u.error_kind, None);
361        assert_eq!(u.record_index, None);
362    }
363
364    #[test]
365    fn unwrap_envelope_errors_on_non_object_and_missing_payload() {
366        assert_eq!(
367            unwrap_envelope(&json!("just a string")),
368            Err(EnvelopeError::NotObject)
369        );
370        assert_eq!(
371            unwrap_envelope(&json!([1, 2, 3])),
372            Err(EnvelopeError::NotObject)
373        );
374        assert_eq!(
375            unwrap_envelope(&json!({ "error": { "kind": "Sink" } })),
376            Err(EnvelopeError::MissingPayload)
377        );
378    }
379
380    #[test]
381    fn on_batch_error_defaults_to_propagate() {
382        assert_eq!(OnBatchError::default(), OnBatchError::Propagate);
383    }
384
385    #[test]
386    fn on_batch_error_serializes_snake_case() {
387        let prop = serde_json::to_string(&OnBatchError::Propagate).unwrap();
388        let all = serde_json::to_string(&OnBatchError::DlqAll).unwrap();
389        assert_eq!(prop, "\"propagate\"");
390        assert_eq!(all, "\"dlq_all\"");
391    }
392
393    #[test]
394    fn on_batch_error_deserializes_snake_case() {
395        let prop: OnBatchError = serde_json::from_str("\"propagate\"").unwrap();
396        let all: OnBatchError = serde_json::from_str("\"dlq_all\"").unwrap();
397        assert_eq!(prop, OnBatchError::Propagate);
398        assert_eq!(all, OnBatchError::DlqAll);
399    }
400
401    #[test]
402    fn dlq_reason_strings() {
403        assert_eq!(DlqReason::Partial.as_str(), "partial");
404        assert_eq!(DlqReason::DlqAll.as_str(), "dlq_all");
405    }
406
407    #[test]
408    fn dlq_reason_quality_string() {
409        assert_eq!(DlqReason::Quality.as_str(), "quality");
410    }
411
412    #[test]
413    fn dlq_reason_schema_drift_string() {
414        assert_eq!(DlqReason::SchemaDrift.as_str(), "schema_drift");
415    }
416
417    #[test]
418    fn dlq_reason_contract_string() {
419        assert_eq!(DlqReason::Contract.as_str(), "contract");
420    }
421}
faucet_core/dlq.rs

faucet_core/
dlq.rs
