faucet_cli/notify/
event.rs

1//! The runtime event a pipeline emits toward the notifier (#280).
2//!
3//! [`NotifyEvent`] is the pure, channel-agnostic description of something that
4//! happened. Constructors fix the canonical severity per kind so callers at the
5//! emit sites (executor, SLA pass, scheduler) don't have to. Rendering into a
6//! Slack / PagerDuty / webhook body lives in [`crate::notify::render`].
7
8use super::spec::{EventKind, Severity};
9use serde_json::{Map, Value};
10
11/// Scrub every resolved secret from text bound for an external channel.
12fn redact(s: String) -> String {
13    crate::secrets::registry::redact(&s).into_owned()
14}
15
16/// A single thing worth notifying about.
17#[derive(Debug, Clone)]
18pub struct NotifyEvent {
19    pub kind: EventKind,
20    pub severity: Severity,
21    /// Pipeline name (metric-label identity).
22    pub pipeline: String,
23    /// Matrix row id (`""` for non-matrix runs).
24    pub row: String,
25    /// Short one-line summary.
26    pub title: String,
27    /// Human-readable detail.
28    pub message: String,
29    /// Structured context rendered into channel payloads (never contains
30    /// secrets — the emit sites pass only safe scalars).
31    pub details: Map<String, Value>,
32}
33
34impl NotifyEvent {
35    /// The one constructor every event goes through — and therefore the single
36    /// place to scrub secrets.
37    ///
38    /// A notification leaves the trust boundary entirely: Slack, PagerDuty, or a
39    /// customer webhook. `message` is frequently a raw `FaucetError`, whose
40    /// `Display` can carry the material that produced it — `reqwest` includes the
41    /// full request URL, so a REST source with its API key in a query parameter
42    /// would post that key to a third party, and connection-string leakage in a
43    /// CDC error has already been a filed bug (#84). Every other outbound-ish
44    /// surface already redacts (MCP tool errors, the serve log layer, doctor
45    /// probe output); notifications did not (#456 H5).
46    fn base(
47        kind: EventKind,
48        severity: Severity,
49        pipeline: impl Into<String>,
50        row: impl Into<String>,
51        title: impl Into<String>,
52        message: impl Into<String>,
53    ) -> Self {
54        Self {
55            kind,
56            severity,
57            pipeline: pipeline.into(),
58            row: row.into(),
59            title: redact(title.into()),
60            message: redact(message.into()),
61            details: Map::new(),
62        }
63    }
64
65    fn with(mut self, key: &str, value: Value) -> Self {
66        // Detail values are rendered into the same outbound payload as `message`,
67        // so a string detail is scrubbed too.
68        let value = match value {
69            Value::String(s) => Value::String(redact(s)),
70            other => other,
71        };
72        self.details.insert(key.to_string(), value);
73        self
74    }
75
76    /// Correlation key for PagerDuty incident open/resolve pairing and for
77    /// coalescing per (pipeline, row). A `run_success` resolves the incident a
78    /// prior `run_failure` opened on the same key.
79    pub fn incident_key(&self) -> String {
80        format!("{}:{}", self.pipeline, self.row)
81    }
82
83    /// Per-rule coalesce key: the same kind repeating for the same (pipeline,
84    /// row) coalesces within a rule's `dedupe_window_secs`.
85    pub fn dedupe_key(&self) -> String {
86        format!("{}:{}:{}", self.kind.as_str(), self.pipeline, self.row)
87    }
88
89    /// True when this event opens an incident (a failure-class event that a
90    /// later success should resolve).
91    pub fn opens_incident(&self) -> bool {
92        matches!(
93            self.kind,
94            EventKind::RunFailure | EventKind::CircuitOpen | EventKind::ContractAbort
95        )
96    }
97
98    /// True when this event closes any open incident for its `incident_key`.
99    pub fn closes_incident(&self) -> bool {
100        matches!(self.kind, EventKind::RunSuccess)
101    }
102
103    // ── Constructors (canonical severity per kind) ───────────────────────────
104
105    pub fn run_failure(
106        pipeline: impl Into<String>,
107        row: impl Into<String>,
108        error_kind: &str,
109        message: impl Into<String>,
110    ) -> Self {
111        let p = pipeline.into();
112        Self::base(
113            EventKind::RunFailure,
114            Severity::Error,
115            p.clone(),
116            row,
117            format!("Pipeline `{p}` failed"),
118            message,
119        )
120        .with("error_kind", Value::String(error_kind.to_string()))
121    }
122
123    pub fn run_success(
124        pipeline: impl Into<String>,
125        row: impl Into<String>,
126        rows_written: u64,
127    ) -> Self {
128        let p = pipeline.into();
129        Self::base(
130            EventKind::RunSuccess,
131            Severity::Info,
132            p.clone(),
133            row,
134            format!("Pipeline `{p}` succeeded"),
135            format!("Run completed, {rows_written} records written."),
136        )
137        .with("records_written", Value::from(rows_written))
138    }
139
140    pub fn sla_breach(
141        pipeline: impl Into<String>,
142        row: impl Into<String>,
143        sla_kind: &str,
144        message: impl Into<String>,
145    ) -> Self {
146        let p = pipeline.into();
147        Self::base(
148            EventKind::SlaBreach,
149            Severity::Warning,
150            p.clone(),
151            row,
152            format!("SLA breach ({sla_kind}) on `{p}`"),
153            message,
154        )
155        .with("sla_kind", Value::String(sla_kind.to_string()))
156    }
157
158    pub fn circuit_open(
159        pipeline: impl Into<String>,
160        row: impl Into<String>,
161        failures: u32,
162        cooldown_secs: u64,
163    ) -> Self {
164        let p = pipeline.into();
165        Self::base(
166            EventKind::CircuitOpen,
167            Severity::Critical,
168            p.clone(),
169            row,
170            format!("Circuit breaker open on `{p}`"),
171            format!(
172                "Tripped after {failures} consecutive failures; cooling down {cooldown_secs}s."
173            ),
174        )
175        .with("failures", Value::from(failures))
176        .with("cooldown_secs", Value::from(cooldown_secs))
177    }
178
179    pub fn contract_abort(
180        pipeline: impl Into<String>,
181        row: impl Into<String>,
182        message: impl Into<String>,
183    ) -> Self {
184        let p = pipeline.into();
185        Self::base(
186            EventKind::ContractAbort,
187            Severity::Error,
188            p.clone(),
189            row,
190            format!("Data contract breach aborted `{p}`"),
191            message,
192        )
193    }
194
195    pub fn dlq_threshold(
196        pipeline: impl Into<String>,
197        row: impl Into<String>,
198        records_dlq: u64,
199    ) -> Self {
200        let p = pipeline.into();
201        Self::base(
202            EventKind::DlqThreshold,
203            Severity::Warning,
204            p.clone(),
205            row,
206            format!("DLQ threshold reached on `{p}`"),
207            format!("{records_dlq} records were routed to the dead-letter queue."),
208        )
209        .with("records_dlq", Value::from(records_dlq))
210    }
211
212    pub fn scheduler_stuck(pipeline: impl Into<String>, message: impl Into<String>) -> Self {
213        let p = pipeline.into();
214        Self::base(
215            EventKind::SchedulerStuck,
216            Severity::Critical,
217            p.clone(),
218            String::new(),
219            format!("Scheduler stuck for `{p}`"),
220            message,
221        )
222    }
223}
224
225#[cfg(test)]
226mod tests {
227    use super::*;
228
229    #[test]
230    fn constructors_fix_severity_and_kind() {
231        assert_eq!(
232            NotifyEvent::run_failure("p", "", "sink", "boom").severity,
233            Severity::Error
234        );
235        assert_eq!(
236            NotifyEvent::circuit_open("p", "", 5, 30).severity,
237            Severity::Critical
238        );
239        assert_eq!(
240            NotifyEvent::run_success("p", "", 10).severity,
241            Severity::Info
242        );
243        assert_eq!(
244            NotifyEvent::sla_breach("p", "", "staleness", "old").severity,
245            Severity::Warning
246        );
247        assert_eq!(
248            NotifyEvent::scheduler_stuck("p", "no beat").kind,
249            EventKind::SchedulerStuck
250        );
251    }
252
253    #[test]
254    fn incident_and_dedupe_keys() {
255        let f = NotifyEvent::run_failure("p", "r1", "sink", "boom");
256        assert_eq!(f.incident_key(), "p:r1");
257        assert_eq!(f.dedupe_key(), "run_failure:p:r1");
258        assert!(f.opens_incident());
259        assert!(!f.closes_incident());
260
261        let s = NotifyEvent::run_success("p", "r1", 3);
262        assert_eq!(s.incident_key(), "p:r1"); // matches the failure it resolves
263        assert!(s.closes_incident());
264        assert!(!s.opens_incident());
265    }
266
267    #[test]
268    fn details_carry_structured_context() {
269        let e = NotifyEvent::dlq_threshold("p", "", 42);
270        assert_eq!(e.details.get("records_dlq").unwrap(), &Value::from(42u64));
271    }
272}
273
274#[cfg(test)]
275mod redaction_tests {
276    use super::*;
277
278    /// #456 H5: a notification is delivered to Slack / PagerDuty / a customer
279    /// webhook, i.e. outside the trust boundary, so a resolved secret that landed
280    /// in the error text must not ride along.
281    #[test]
282    fn secrets_are_scrubbed_from_every_outbound_field() {
283        // A realistic leak: the API key is a query parameter, and reqwest's error
284        // Display embeds the whole URL.
285        let secret = "sk-live-456-audit-secret";
286        crate::secrets::registry::register(secret);
287
288        let ev = NotifyEvent::run_failure(
289            "p",
290            "row",
291            "http",
292            format!("HTTP error for url (https://api.example.com/v1?api_key={secret})"),
293        );
294        assert!(
295            !ev.message.contains(secret),
296            "message leaked: {}",
297            ev.message
298        );
299        assert!(ev.message.contains("***"), "{}", ev.message);
300
301        // Titles and string details go into the same payload.
302        let ev = NotifyEvent::sla_breach("p", "row", "staleness", format!("token {secret} stale"));
303        assert!(!ev.message.contains(secret));
304
305        let ev = NotifyEvent::run_failure("p", "row", "cfg", "boom")
306            .with("detail", Value::String(format!("url={secret}")));
307        assert!(
308            !ev.details["detail"].as_str().unwrap().contains(secret),
309            "detail leaked: {:?}",
310            ev.details
311        );
312        // Non-string details pass through untouched.
313        let ev = NotifyEvent::run_success("p", "row", 7);
314        assert_eq!(ev.details["records_written"], Value::from(7u64));
315    }
316}
faucet_cli/notify/event.rs

faucet_cli/notify/
event.rs
