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

faucet_core/
dlq.rs
