opssignal_core/
lib.rs

1#![forbid(unsafe_code)]
2
3mod chain;
4#[cfg(feature = "object-store")]
5mod cloud;
6pub mod config;
7pub mod delivery;
8mod inflight;
9pub mod providers;
10mod reload;
11pub mod routing;
12pub mod secrets;
13pub mod signal;
14pub mod sink;
15pub mod source;
16pub mod template;
17mod throttle;
18
19const MAX_RETRY_ATTEMPTS: u32 = 3;
20const INITIAL_BACKOFF_MS: u64 = 200;
21
22/// How long dropping a [`SignalClient`] waits for fire-and-forget
23/// deliveries still in flight.
24pub const DEFAULT_FLUSH_TIMEOUT: Duration = Duration::from_secs(5);
25
26use config::{Config, SinkKind};
27use delivery::{DeliveryResult, DeliveryStatus, Provider, ProviderOutcome};
28use inflight::InFlight;
29use providers::{
30    console::ConsoleProvider, email::EmailProvider, opsgenie::OpsgenieProvider,
31    pagerduty::PagerDutyProvider, slack::SlackProvider, teams::TeamsProvider,
32    webhook::WebhookProvider,
33};
34use routing::{DeliveryPlan, RoutingConfig};
35use signal::{Signal, SignalError, SignalInput};
36use sink::{SignalSink, SqliteSink};
37use source::{ConfigSource, LoadOptions};
38use std::collections::HashMap;
39use std::sync::{Arc, Mutex, RwLock};
40use std::time::Duration;
41use throttle::{Admission, Throttle};
42
43/// The part of a client that a config reload replaces: the routing
44/// rules and the providers they name. Swapped as one unit, so a signal
45/// never sees new routes with old providers.
46pub struct Pipeline {
47    pub routing: routing::RoutingConfig,
48    pub providers: HashMap<String, Arc<dyn Provider>>,
49}
50
51impl Pipeline {
52    /// Registers the configured providers. `console` is always present
53    /// because routing falls back to it.
54    pub fn from_config(config: &Config) -> Self {
55        let mut providers: HashMap<String, Arc<dyn Provider>> = HashMap::new();
56        let console_templates = config
57            .providers
58            .console
59            .as_ref()
60            .and_then(|c| template::Templates::load("console", &c.template));
61        providers.insert(
62            "console".to_string(),
63            Arc::new(ConsoleProvider::new().with_templates(console_templates)),
64        );
65        if let Some(slack) = &config.providers.slack {
66            providers.insert(
67                "slack".to_string(),
68                Arc::new(
69                    SlackProvider::new(slack.webhook_url.clone())
70                        .with_templates(template::Templates::load("slack", &slack.template)),
71                ),
72            );
73        }
74        if let Some(webhook) = &config.providers.webhook {
75            providers.insert(
76                "webhook".to_string(),
77                Arc::new(
78                    WebhookProvider::new(webhook.url.clone())
79                        .with_templates(template::Templates::load("webhook", &webhook.template)),
80                ),
81            );
82        }
83        if let Some(pagerduty) = &config.providers.pagerduty {
84            providers.insert(
85                "pagerduty".to_string(),
86                Arc::new(
87                    PagerDutyProvider::new(pagerduty.routing_key.clone(), pagerduty.url.clone())
88                        .with_templates(template::Templates::load(
89                            "pagerduty",
90                            &pagerduty.template,
91                        )),
92                ),
93            );
94        }
95        if let Some(teams) = &config.providers.teams {
96            providers.insert(
97                "teams".to_string(),
98                Arc::new(
99                    TeamsProvider::new(teams.webhook_url.clone())
100                        .with_templates(template::Templates::load("teams", &teams.template)),
101                ),
102            );
103        }
104        if let Some(opsgenie) = &config.providers.opsgenie {
105            providers.insert(
106                "opsgenie".to_string(),
107                Arc::new(
108                    OpsgenieProvider::new(
109                        opsgenie.api_key.clone(),
110                        opsgenie.url.clone(),
111                        opsgenie.tags.clone(),
112                    )
113                    .with_templates(template::Templates::load("opsgenie", &opsgenie.template)),
114                ),
115            );
116        }
117        if let Some(email) = &config.providers.email {
118            providers.insert(
119                "email".to_string(),
120                Arc::new(EmailProvider::new(email.clone())),
121            );
122        }
123        Pipeline {
124            routing: config.routing(),
125            providers,
126        }
127    }
128
129    /// What each provider `signal` routes to would be sent, in plan
130    /// order, without sending anything or recording the signal; with
131    /// `only`, just that provider whether or not a route picks it. A
132    /// provider that can't show its payload has `None`.
133    pub fn preview(
134        &self,
135        signal: &Signal,
136        only: Option<&str>,
137    ) -> Result<Vec<(String, Option<serde_json::Value>)>, SignalError> {
138        let names: Vec<String> = match only {
139            Some(name) => vec![name.to_string()],
140            None => self.routing.plan(signal).providers().cloned().collect(),
141        };
142        names
143            .into_iter()
144            .map(|name| match self.providers.get(&name) {
145                Some(provider) => {
146                    let payload = provider.preview(signal);
147                    Ok((name, payload))
148                }
149                None => {
150                    let mut known: Vec<&String> = self.providers.keys().collect();
151                    known.sort();
152                    let known: Vec<&str> = known.iter().map(|s| s.as_str()).collect();
153                    Err(SignalError::Config(format!(
154                        "provider '{name}' is not configured (configured: {})",
155                        known.join(", ")
156                    )))
157                }
158            })
159            .collect()
160    }
161}
162
163type SharedPipeline = Arc<RwLock<Arc<Pipeline>>>;
164
165/// The assembled pipeline: validate -> record -> route -> deliver ->
166/// complete. The signal is saved before delivery starts, so it survives
167/// a process exit mid-delivery as `pending`, and each provider's
168/// outcome is saved as soon as it is known. A sink failure is logged
169/// and never stops delivery.
170///
171/// Dropping the client waits up to [`DEFAULT_FLUSH_TIMEOUT`] for
172/// fire-and-forget deliveries; call [`SignalClient::flush`] to choose
173/// the wait, or when the client lives in a static that is never dropped.
174pub struct SignalClient {
175    pub sink: Arc<dyn SignalSink>,
176    pipeline: SharedPipeline,
177    /// Route `dedup` and `rate_limit` history. `None` if its table
178    /// could not be opened; routes then send without throttling.
179    throttle: Option<Arc<Throttle>>,
180    in_flight: Arc<InFlight>,
181    pub runtime: tokio::runtime::Runtime,
182}
183
184impl SignalClient {
185    /// Builds a client from a parsed `signal.yaml`: registers the
186    /// configured providers (console always), opens the sink, and
187    /// starts the runtime that delivery tasks run on. The config is
188    /// fixed for the client's lifetime; see [`SignalClient::from_source`]
189    /// for one that reloads.
190    pub fn from_config(config: &Config) -> Result<Self, SignalError> {
191        Self::with_runtime(config, build_runtime()?)
192    }
193
194    /// Loads the config from `source` and, if `opts.refresh` is set,
195    /// keeps re-reading it in the background and applies changes
196    /// without a restart. A reload that fails to fetch or parse keeps
197    /// the current config, so a bad edit can't take alerting down.
198    pub fn from_source(source: &ConfigSource, opts: &LoadOptions) -> Result<Self, SignalError> {
199        Self::from_sources(std::slice::from_ref(source), opts)
200    }
201
202    /// [`SignalClient::from_source`] for a layered config: `sources`
203    /// are merged lowest priority first, along with every file they
204    /// `extends`, and any of them changing triggers a reload.
205    pub fn from_sources(sources: &[ConfigSource], opts: &LoadOptions) -> Result<Self, SignalError> {
206        let runtime = build_runtime()?;
207        let loaded = runtime.block_on(source::load_initial(sources, opts))?;
208        let client = Self::with_runtime(&loaded.config, runtime)?;
209        if let Some(every) = opts.refresh {
210            reload::spawn(
211                &client.runtime,
212                reload::Reloader {
213                    roots: sources.to_vec(),
214                    opts: opts.clone(),
215                    every,
216                    last_texts: loaded.texts,
217                    storage: loaded.config.storage.clone(),
218                    pipeline: Arc::clone(&client.pipeline),
219                },
220            );
221        }
222        Ok(client)
223    }
224
225    /// The discovered config (`$SIGNAL_CONFIG`, then `./signal.yaml`,
226    /// then the zero-config default), reloaded on the interval in
227    /// `$SIGNAL_CONFIG_REFRESH_SECS`.
228    pub fn from_env() -> Result<Self, SignalError> {
229        let opts = LoadOptions::from_env()?;
230        match ConfigSource::discover()? {
231            Some(sources) => Self::from_sources(&sources, &opts),
232            None => Self::from_config(&Config::default()),
233        }
234    }
235
236    fn with_runtime(
237        config: &Config,
238        runtime: tokio::runtime::Runtime,
239    ) -> Result<Self, SignalError> {
240        let sink: Arc<dyn SignalSink> = match config.storage.sink {
241            SinkKind::Sqlite => Arc::new(SqliteSink::open(&config.storage.path)?),
242        };
243        let throttle = match Throttle::open(&config.storage.path) {
244            Ok(t) => Some(Arc::new(t)),
245            Err(e) => {
246                tracing::error!("{e}; dedup and rate limits are off");
247                None
248            }
249        };
250        Ok(SignalClient {
251            sink,
252            pipeline: Arc::new(RwLock::new(Arc::new(Pipeline::from_config(config)))),
253            throttle,
254            in_flight: Arc::default(),
255            runtime,
256        })
257    }
258
259    /// The routing rules and providers currently in effect.
260    pub fn pipeline(&self) -> Arc<Pipeline> {
261        current(&self.pipeline)
262    }
263
264    /// Fire-and-forget. Blocks the caller only for the local sink
265    /// write, never on a provider. This is the default an Airflow
266    /// on_failure_callback should use — a flaky Slack webhook must never
267    /// add latency or failure risk to the DAG run itself.
268    pub fn notify_async(&self, input: SignalInput) {
269        if let Err(e) = signal::validate(&input) {
270            tracing::warn!("signal validation failed, dropping: {e}");
271            return;
272        }
273        let signal = Signal::from_input(input);
274        // On the caller's thread, so the signal is on disk before this
275        // returns even if the process exits right after.
276        record(self.sink.as_ref(), &signal);
277        let sink = Arc::clone(&self.sink);
278        let pipeline = self.pipeline();
279        // Also on the caller's thread, so dedup sees signals in the
280        // order they were sent.
281        let (plan, admission) = plan_signal(self.throttle.as_deref(), &pipeline.routing, &signal);
282        let throttle = self.throttle.clone();
283        let guard = self.in_flight.start();
284
285        self.runtime.spawn(async move {
286            let _guard = guard;
287            let tracker = Tracker::new(&signal, sink.as_ref());
288            let result = deliver(&tracker, &plan, &pipeline.providers).await;
289            settle(throttle.as_deref(), &admission, &result);
290            if let Err(e) = sink.complete(&signal, &result) {
291                tracing::error!("sink write failed: {e}");
292            }
293        });
294    }
295
296    /// Waits up to `timeout` for every [`SignalClient::notify_async`]
297    /// delivery started so far to finish. Returns false if some were
298    /// still running; those stay recorded as `pending`. Call it before
299    /// the process exits.
300    pub fn flush(&self, timeout: Duration) -> bool {
301        self.in_flight.wait(timeout) == 0
302    }
303
304    /// Blocking, bounded by timeout. Caller gets a definitive result.
305    /// Same pipeline as notify_async — only the blocking behavior
306    /// differs. On a timeout, providers that already delivered still
307    /// count, so a fan-out cut short reports `Partial` rather than
308    /// `Failed`.
309    #[must_use = "check whether the signal was delivered"]
310    pub fn notify_sync(
311        &self,
312        input: SignalInput,
313        timeout: Duration,
314    ) -> Result<DeliveryResult, SignalError> {
315        signal::validate(&input)?;
316        let signal = Signal::from_input(input);
317        record(self.sink.as_ref(), &signal);
318        let pipeline = self.pipeline();
319        let tracker = Tracker::new(&signal, self.sink.as_ref());
320        let (plan, admission) = plan_signal(self.throttle.as_deref(), &pipeline.routing, &signal);
321
322        let result = self.runtime.block_on(async {
323            match tokio::time::timeout(timeout, deliver(&tracker, &plan, &pipeline.providers)).await
324            {
325                Ok(result) => result,
326                Err(_) => summarize(
327                    &plan,
328                    &tracker.outcomes(),
329                    Some(&format!("timed out after {timeout:?}")),
330                ),
331            }
332        });
333        settle(self.throttle.as_deref(), &admission, &result);
334
335        self.sink.complete(&signal, &result)?;
336        Ok(result)
337    }
338}
339
340impl Drop for SignalClient {
341    fn drop(&mut self) {
342        if !self.flush(DEFAULT_FLUSH_TIMEOUT) {
343            tracing::warn!(
344                "exiting with signals still being delivered; they stay recorded as pending"
345            );
346        }
347    }
348}
349
350/// A sink that can't record must not stop the alert going out;
351/// `complete` writes the whole row again afterwards.
352fn record(sink: &dyn SignalSink, signal: &Signal) {
353    if let Err(e) = sink.record(signal) {
354        tracing::error!("sink write failed, delivering anyway: {e}");
355    }
356}
357
358/// Plans a signal's delivery, holding back the routes whose `dedup` or
359/// `rate_limit` turn it down.
360fn plan_signal(
361    throttle: Option<&Throttle>,
362    routing: &RoutingConfig,
363    signal: &Signal,
364) -> (DeliveryPlan, Admission) {
365    let mut admission = Admission::default();
366    let now = throttle::now_ms();
367    let plan = routing.plan_with(signal, |route| match throttle {
368        Some(t) => t.admit(route, signal, now, &mut admission),
369        None => Ok(()),
370    });
371    for line in &plan.suppressed {
372        tracing::info!("signal {} suppressed on {line}", signal.id);
373    }
374    (plan, admission)
375}
376
377/// A signal that reached no provider must not count as sent, or its
378/// next occurrence would be suppressed as a duplicate of an alert
379/// nobody got.
380fn settle(throttle: Option<&Throttle>, admission: &Admission, result: &DeliveryResult) {
381    if let (Some(t), DeliveryStatus::Failed) = (throttle, result.status) {
382        t.release(admission);
383    }
384}
385
386fn build_runtime() -> Result<tokio::runtime::Runtime, SignalError> {
387    tokio::runtime::Builder::new_multi_thread()
388        .worker_threads(2)
389        .thread_name("opssignal")
390        .enable_all()
391        .build()
392        .map_err(|e| SignalError::Internal(format!("failed to start runtime: {e}")))
393}
394
395/// A poisoned lock only means a reload panicked mid-swap; the value
396/// inside is still a complete pipeline, so keep using it.
397fn current(pipeline: &SharedPipeline) -> Arc<Pipeline> {
398    Arc::clone(&pipeline.read().unwrap_or_else(|e| e.into_inner()))
399}
400
401/// The per-provider outcomes of one delivery. Each is written to the
402/// sink as soon as it changes, so a timeout or exit that cuts delivery
403/// short leaves the providers that already have the signal on record.
404struct Tracker<'a> {
405    signal: &'a Signal,
406    sink: &'a dyn SignalSink,
407    outcomes: Mutex<Vec<ProviderOutcome>>,
408}
409
410impl<'a> Tracker<'a> {
411    fn new(signal: &'a Signal, sink: &'a dyn SignalSink) -> Self {
412        Tracker {
413            signal,
414            sink,
415            outcomes: Mutex::default(),
416        }
417    }
418
419    fn report(&self, outcome: ProviderOutcome) {
420        if let Err(e) = self.sink.provider_outcome(self.signal, &outcome) {
421            tracing::error!("sink write failed: {e}");
422        }
423        let mut outcomes = self.outcomes.lock().unwrap_or_else(|e| e.into_inner());
424        match outcomes.iter_mut().find(|o| o.provider == outcome.provider) {
425            Some(existing) => *existing = outcome,
426            None => outcomes.push(outcome),
427        }
428    }
429
430    fn outcomes(&self) -> Vec<ProviderOutcome> {
431        self.outcomes
432            .lock()
433            .unwrap_or_else(|e| e.into_inner())
434            .clone()
435    }
436}
437
438/// Delivers to every target of the plan at once. Targets never wait on
439/// each other, and each retries only its own providers, so a provider
440/// that delivered is never sent the signal again because another failed.
441async fn deliver(
442    tracker: &Tracker<'_>,
443    plan: &DeliveryPlan,
444    providers: &HashMap<String, Arc<dyn Provider>>,
445) -> DeliveryResult {
446    futures_util::future::join_all(
447        plan.targets
448            .iter()
449            .map(|chain| deliver_chain(tracker, chain, providers)),
450    )
451    .await;
452    summarize(plan, &tracker.outcomes(), None)
453}
454
455/// Tries a target's providers in order until one succeeds.
456async fn deliver_chain(
457    tracker: &Tracker<'_>,
458    chain: &[String],
459    providers: &HashMap<String, Arc<dyn Provider>>,
460) {
461    for name in chain {
462        let Some(provider) = providers.get(name) else {
463            tracing::warn!("route references unknown provider: {name}");
464            continue;
465        };
466        let outcome = |status, error, attempts| ProviderOutcome {
467            provider: name.clone(),
468            status,
469            error,
470            attempts,
471        };
472        tracker.report(outcome(DeliveryStatus::Pending, None, 0));
473        let mut last_error = None;
474        // Basic retry: exponential backoff up to MAX_RETRY_ATTEMPTS.
475        // v0.2 roadmap item: add jitter, make configurable per-route.
476        for attempt in 1..=MAX_RETRY_ATTEMPTS {
477            match provider.send(tracker.signal).await {
478                Ok(()) => {
479                    tracker.report(outcome(DeliveryStatus::Delivered, None, attempt));
480                    return;
481                }
482                Err(e) => {
483                    last_error = Some(e.to_string());
484                    if attempt < MAX_RETRY_ATTEMPTS {
485                        tokio::time::sleep(Duration::from_millis(
486                            INITIAL_BACKOFF_MS * 2u64.pow(attempt),
487                        ))
488                        .await;
489                    }
490                }
491            }
492        }
493        tracker.report(outcome(
494            DeliveryStatus::Failed,
495            last_error,
496            MAX_RETRY_ATTEMPTS,
497        ));
498    }
499}
500
501/// Folds per-provider outcomes into one result. `cut_off` is set when
502/// delivery was stopped early; providers still pending then count as
503/// failed with that message.
504fn summarize(
505    plan: &DeliveryPlan,
506    outcomes: &[ProviderOutcome],
507    cut_off: Option<&str>,
508) -> DeliveryResult {
509    let mut deliveries = Vec::new();
510    let mut delivered = Vec::new();
511    let mut errors = Vec::new();
512    let mut attempts = 0;
513    for chain in &plan.targets {
514        let mut tried: Vec<ProviderOutcome> = chain
515            .iter()
516            .filter_map(|name| outcomes.iter().find(|o| &o.provider == name).cloned())
517            .collect();
518        if let Some(msg) = cut_off {
519            for o in tried
520                .iter_mut()
521                .filter(|o| o.status == DeliveryStatus::Pending)
522            {
523                o.status = DeliveryStatus::Failed;
524                o.error = Some(msg.to_string());
525            }
526        }
527        // A chain stops at its first success, so the provider tried
528        // last decides how the target went.
529        match tried.last() {
530            Some(o) if o.status == DeliveryStatus::Delivered => {
531                delivered.push(o.provider.clone());
532                attempts = attempts.max(o.attempts);
533            }
534            Some(o) => {
535                let error = o.error.clone().unwrap_or_else(|| o.status.to_string());
536                errors.push(format!("{}: {error}", o.provider));
537                if plan.targets.len() == 1 {
538                    errors = vec![error];
539                }
540                attempts = attempts.max(o.attempts);
541            }
542            None => errors.push("no provider delivered successfully".to_string()),
543        }
544        deliveries.extend(tried);
545    }
546
547    let status = if plan.targets.is_empty() {
548        // Every matching route was suppressed; nothing was meant to go.
549        errors = plan.suppressed.clone();
550        DeliveryStatus::Skipped
551    } else if delivered.len() == plan.targets.len() {
552        DeliveryStatus::Delivered
553    } else if delivered.is_empty() {
554        DeliveryStatus::Failed
555    } else {
556        DeliveryStatus::Partial
557    };
558    DeliveryResult {
559        status,
560        provider: (!delivered.is_empty()).then(|| delivered.join(",")),
561        error: (!errors.is_empty()).then(|| errors.join("; ")),
562        attempts,
563        deliveries,
564        suppressed: plan.suppressed.clone(),
565    }
566}
567
568#[cfg(test)]
569mod tests {
570    use super::*;
571    use crate::routing::RoutingConfig;
572    use crate::signal::Severity;
573    use wiremock::matchers::{method, path};
574    use wiremock::{Mock, MockServer, ResponseTemplate};
575
576    fn temp_db() -> String {
577        std::env::temp_dir()
578            .join(format!("opssignal-test-{}.db", uuid::Uuid::new_v4()))
579            .to_string_lossy()
580            .into_owned()
581    }
582
583    fn input(severity: Severity) -> SignalInput {
584        SignalInput {
585            source: "airflow".into(),
586            event_type: "task_failed".into(),
587            title: "dbt model failed".into(),
588            severity,
589            environment: Some("prod".into()),
590            ..Default::default()
591        }
592    }
593
594    fn stored_rows(db: &str) -> Vec<(String, Option<String>)> {
595        let conn = rusqlite::Connection::open(db).expect("open db");
596        let mut stmt = conn
597            .prepare("SELECT delivery_status, provider FROM signals")
598            .expect("prepare");
599        stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
600            .expect("query")
601            .map(|r| r.expect("row"))
602            .collect()
603    }
604
605    #[test]
606    fn zero_config_client_delivers_to_console_and_records_it() {
607        let db = temp_db();
608        let yaml = format!("storage:\n  path: \"{db}\"\n");
609        let client = SignalClient::from_config(&Config::from_yaml_str(&yaml).expect("config"))
610            .expect("client");
611
612        let result = client
613            .notify_sync(input(Severity::Info), Duration::from_secs(5))
614            .expect("notify_sync");
615
616        assert_eq!(result.status, DeliveryStatus::Delivered);
617        assert_eq!(result.provider.as_deref(), Some("console"));
618        assert_eq!(
619            stored_rows(&db),
620            vec![("delivered".to_string(), Some("console".to_string()))]
621        );
622        let _ = std::fs::remove_file(&db);
623    }
624
625    #[test]
626    fn configured_route_delivers_to_webhook() {
627        // The mock server lives on its own runtime; notify_sync blocks
628        // on the client's runtime, so the two must not be nested.
629        let server_rt = tokio::runtime::Runtime::new().expect("server runtime");
630        let server = server_rt.block_on(async {
631            let server = MockServer::start().await;
632            Mock::given(method("POST"))
633                .respond_with(ResponseTemplate::new(200))
634                .expect(1)
635                .mount(&server)
636                .await;
637            server
638        });
639
640        let db = temp_db();
641        let yaml = format!(
642            "providers:\n  webhook:\n    url: \"{}/hook\"\nroutes:\n  - match:\n      severity: [error, critical]\n      environment: prod\n    providers: [webhook]\nstorage:\n  path: \"{db}\"\n",
643            server.uri()
644        );
645        let client = SignalClient::from_config(&Config::from_yaml_str(&yaml).expect("config"))
646            .expect("client");
647
648        let result = client
649            .notify_sync(input(Severity::Error), Duration::from_secs(5))
650            .expect("notify_sync");
651
652        assert_eq!(result.status, DeliveryStatus::Delivered);
653        assert_eq!(result.provider.as_deref(), Some("webhook"));
654        server_rt.block_on(server.verify());
655        let _ = std::fs::remove_file(&db);
656    }
657
658    /// A provider that fails its first `fail_times` sends, then succeeds,
659    /// counting every call so tests can assert on retry behaviour.
660    struct FlakyProvider {
661        name: &'static str,
662        fail_times: u32,
663        calls: std::sync::atomic::AtomicU32,
664    }
665
666    impl FlakyProvider {
667        fn new(name: &'static str, fail_times: u32) -> Arc<Self> {
668            Arc::new(FlakyProvider {
669                name,
670                fail_times,
671                calls: std::sync::atomic::AtomicU32::new(0),
672            })
673        }
674
675        fn calls(&self) -> u32 {
676            self.calls.load(std::sync::atomic::Ordering::SeqCst)
677        }
678    }
679
680    #[async_trait::async_trait]
681    impl Provider for FlakyProvider {
682        fn name(&self) -> &'static str {
683            self.name
684        }
685
686        async fn send(&self, _signal: &Signal) -> Result<(), SignalError> {
687            let call = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
688            if call <= self.fail_times {
689                Err(SignalError::Delivery(format!(
690                    "{} failure #{call}",
691                    self.name
692                )))
693            } else {
694                Ok(())
695            }
696        }
697    }
698
699    /// Never finishes within any reasonable timeout.
700    struct HangingProvider;
701
702    #[async_trait::async_trait]
703    impl Provider for HangingProvider {
704        fn name(&self) -> &'static str {
705            "hanging"
706        }
707
708        async fn send(&self, _signal: &Signal) -> Result<(), SignalError> {
709            tokio::time::sleep(Duration::from_secs(3600)).await;
710            Ok(())
711        }
712    }
713
714    fn route(providers: &[&str], mode: routing::DeliveryMode) -> RoutingConfig {
715        routing::RoutingConfig {
716            routes: vec![routing::Route {
717                match_rule: routing::RouteMatch {
718                    severity: vec![],
719                    environment: None,
720                    source: None,
721                },
722                providers: providers.iter().map(|p| p.to_string()).collect(),
723                mode,
724                dedup: None,
725                rate_limit: None,
726            }],
727        }
728    }
729
730    fn routing_to(providers: &[&str]) -> RoutingConfig {
731        route(providers, routing::DeliveryMode::Failover)
732    }
733
734    fn fanout_to(providers: &[&str]) -> RoutingConfig {
735        route(providers, routing::DeliveryMode::Fanout)
736    }
737
738    /// Ignores every write.
739    struct NoSink;
740
741    impl SignalSink for NoSink {
742        fn record(&self, _: &Signal) -> Result<(), SignalError> {
743            Ok(())
744        }
745        fn complete(&self, _: &Signal, _: &DeliveryResult) -> Result<(), SignalError> {
746            Ok(())
747        }
748    }
749
750    async fn deliver_to(
751        routing: &RoutingConfig,
752        providers: &HashMap<String, Arc<dyn Provider>>,
753    ) -> DeliveryResult {
754        let signal = signal();
755        let tracker = Tracker::new(&signal, &NoSink);
756        deliver(&tracker, &routing.plan(&signal), providers).await
757    }
758
759    fn registry(providers: Vec<Arc<dyn Provider>>) -> HashMap<String, Arc<dyn Provider>> {
760        providers
761            .into_iter()
762            .map(|p| (p.name().to_string(), p))
763            .collect()
764    }
765
766    fn signal() -> Signal {
767        Signal::from_input(input(Severity::Error))
768    }
769
770    #[tokio::test(start_paused = true)]
771    async fn deliver_succeeds_first_try_without_backoff() {
772        let ok = FlakyProvider::new("ok", 0);
773        let start = tokio::time::Instant::now();
774
775        let result = deliver_to(&routing_to(&["ok"]), &registry(vec![ok.clone()])).await;
776
777        assert_eq!(result.status, DeliveryStatus::Delivered);
778        assert_eq!(result.provider.as_deref(), Some("ok"));
779        assert_eq!(result.attempts, 1);
780        assert_eq!(ok.calls(), 1);
781        assert_eq!(start.elapsed(), Duration::ZERO);
782    }
783
784    #[tokio::test(start_paused = true)]
785    async fn deliver_retries_with_exponential_backoff_then_succeeds() {
786        let flaky = FlakyProvider::new("flaky", 2);
787        let start = tokio::time::Instant::now();
788
789        let result = deliver_to(&routing_to(&["flaky"]), &registry(vec![flaky.clone()])).await;
790
791        assert_eq!(result.status, DeliveryStatus::Delivered);
792        assert_eq!(result.attempts, 3);
793        assert_eq!(flaky.calls(), 3);
794        // Backoff after attempt n is INITIAL_BACKOFF_MS * 2^n: 400ms + 800ms.
795        assert_eq!(start.elapsed(), Duration::from_millis(1200));
796    }
797
798    #[tokio::test(start_paused = true)]
799    async fn deliver_falls_back_to_next_provider_after_retries_are_exhausted() {
800        let broken = FlakyProvider::new("broken", u32::MAX);
801        let backup = FlakyProvider::new("backup", 0);
802
803        let result = deliver_to(
804            &routing_to(&["broken", "backup"]),
805            &registry(vec![broken.clone(), backup.clone()]),
806        )
807        .await;
808
809        assert_eq!(result.status, DeliveryStatus::Delivered);
810        assert_eq!(result.provider.as_deref(), Some("backup"));
811        assert_eq!(result.attempts, 1);
812        assert_eq!(broken.calls(), MAX_RETRY_ATTEMPTS);
813        assert_eq!(backup.calls(), 1);
814    }
815
816    #[tokio::test(start_paused = true)]
817    async fn deliver_stops_at_first_successful_provider() {
818        let first = FlakyProvider::new("first", 0);
819        let second = FlakyProvider::new("second", 0);
820
821        let result = deliver_to(
822            &routing_to(&["first", "second"]),
823            &registry(vec![first.clone(), second.clone()]),
824        )
825        .await;
826
827        assert_eq!(result.provider.as_deref(), Some("first"));
828        assert_eq!(second.calls(), 0);
829    }
830
831    #[tokio::test(start_paused = true)]
832    async fn deliver_reports_last_error_when_every_provider_fails() {
833        let a = FlakyProvider::new("a", u32::MAX);
834        let b = FlakyProvider::new("b", u32::MAX);
835
836        let result = deliver_to(
837            &routing_to(&["a", "b"]),
838            &registry(vec![a.clone(), b.clone()]),
839        )
840        .await;
841
842        assert_eq!(result.status, DeliveryStatus::Failed);
843        assert_eq!(result.provider, None);
844        assert_eq!(result.attempts, MAX_RETRY_ATTEMPTS);
845        assert_eq!(
846            result.error.as_deref(),
847            Some("delivery failed: b failure #3")
848        );
849        assert_eq!(a.calls(), MAX_RETRY_ATTEMPTS);
850        assert_eq!(b.calls(), MAX_RETRY_ATTEMPTS);
851    }
852
853    #[tokio::test(start_paused = true)]
854    async fn deliver_skips_unregistered_provider_and_uses_the_next() {
855        let ok = FlakyProvider::new("ok", 0);
856
857        let result = deliver_to(&routing_to(&["missing", "ok"]), &registry(vec![ok.clone()])).await;
858
859        assert_eq!(result.status, DeliveryStatus::Delivered);
860        assert_eq!(result.provider.as_deref(), Some("ok"));
861    }
862
863    #[tokio::test(start_paused = true)]
864    async fn deliver_with_no_registered_provider_fails_with_a_message() {
865        let result = deliver_to(&routing_to(&["missing"]), &HashMap::new()).await;
866
867        assert_eq!(result.status, DeliveryStatus::Failed);
868        assert_eq!(
869            result.error.as_deref(),
870            Some("no provider delivered successfully")
871        );
872    }
873
874    #[tokio::test(start_paused = true)]
875    async fn fanout_delivers_to_every_provider_concurrently() {
876        let a = FlakyProvider::new("a", 2);
877        let b = FlakyProvider::new("b", 2);
878        let start = tokio::time::Instant::now();
879
880        let result = deliver_to(
881            &fanout_to(&["a", "b"]),
882            &registry(vec![a.clone(), b.clone()]),
883        )
884        .await;
885
886        assert_eq!(result.status, DeliveryStatus::Delivered);
887        assert_eq!(result.provider.as_deref(), Some("a,b"));
888        assert_eq!(result.error, None);
889        assert_eq!(result.attempts, 3);
890        assert_eq!((a.calls(), b.calls()), (3, 3));
891        // Both back off at the same time: one provider's 400ms + 800ms,
892        // not two of them back to back.
893        assert_eq!(start.elapsed(), Duration::from_millis(1200));
894    }
895
896    #[tokio::test(start_paused = true)]
897    async fn fanout_retries_never_resend_to_a_provider_that_delivered() {
898        let ok = FlakyProvider::new("ok", 0);
899        let flaky = FlakyProvider::new("flaky", 2);
900
901        let result = deliver_to(
902            &fanout_to(&["ok", "flaky"]),
903            &registry(vec![ok.clone(), flaky.clone()]),
904        )
905        .await;
906
907        assert_eq!(result.status, DeliveryStatus::Delivered);
908        assert_eq!(ok.calls(), 1);
909        assert_eq!(flaky.calls(), 3);
910        assert_eq!(
911            result.deliveries,
912            vec![
913                ProviderOutcome {
914                    provider: "ok".into(),
915                    status: DeliveryStatus::Delivered,
916                    error: None,
917                    attempts: 1,
918                },
919                ProviderOutcome {
920                    provider: "flaky".into(),
921                    status: DeliveryStatus::Delivered,
922                    error: None,
923                    attempts: 3,
924                },
925            ]
926        );
927    }
928
929    #[tokio::test(start_paused = true)]
930    async fn fanout_with_a_failing_provider_is_partial() {
931        let ok = FlakyProvider::new("ok", 0);
932        let broken = FlakyProvider::new("broken", u32::MAX);
933
934        let result = deliver_to(
935            &fanout_to(&["broken", "ok"]),
936            &registry(vec![ok.clone(), broken.clone()]),
937        )
938        .await;
939
940        assert_eq!(result.status, DeliveryStatus::Partial);
941        assert_eq!(result.provider.as_deref(), Some("ok"));
942        assert_eq!(
943            result.error.as_deref(),
944            Some("broken: delivery failed: broken failure #3")
945        );
946        assert_eq!(ok.calls(), 1);
947        assert_eq!(broken.calls(), MAX_RETRY_ATTEMPTS);
948    }
949
950    #[tokio::test(start_paused = true)]
951    async fn fanout_where_every_provider_fails_names_each_error() {
952        let a = FlakyProvider::new("a", u32::MAX);
953        let b = FlakyProvider::new("b", u32::MAX);
954
955        let result = deliver_to(&fanout_to(&["a", "b"]), &registry(vec![a, b])).await;
956
957        assert_eq!(result.status, DeliveryStatus::Failed);
958        assert_eq!(result.provider, None);
959        assert_eq!(
960            result.error.as_deref(),
961            Some("a: delivery failed: a failure #3; b: delivery failed: b failure #3")
962        );
963    }
964
965    #[tokio::test(start_paused = true)]
966    async fn failover_result_lists_every_provider_it_tried() {
967        let broken = FlakyProvider::new("broken", u32::MAX);
968        let backup = FlakyProvider::new("backup", 0);
969
970        let result = deliver_to(
971            &routing_to(&["broken", "backup"]),
972            &registry(vec![broken, backup]),
973        )
974        .await;
975
976        let tried: Vec<_> = result
977            .deliveries
978            .iter()
979            .map(|o| (o.provider.as_str(), o.status))
980            .collect();
981        assert_eq!(
982            tried,
983            vec![
984                ("broken", DeliveryStatus::Failed),
985                ("backup", DeliveryStatus::Delivered)
986            ]
987        );
988    }
989
990    fn delivery_rows(db: &str) -> Vec<(String, String, Option<String>)> {
991        let conn = rusqlite::Connection::open(db).expect("open db");
992        let mut stmt = conn
993            .prepare("SELECT provider, status, error FROM deliveries ORDER BY provider")
994            .expect("prepare");
995        stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
996            .expect("query")
997            .map(|r| r.expect("row"))
998            .collect()
999    }
1000
1001    #[test]
1002    fn notify_sync_timeout_keeps_the_providers_that_already_delivered() {
1003        let db = temp_db();
1004        let ok = FlakyProvider::new("ok", 0);
1005        let client = client_with_routing(
1006            vec![ok.clone(), Arc::new(HangingProvider)],
1007            fanout_to(&["ok", "hanging"]),
1008            &db,
1009        );
1010
1011        let result = client
1012            .notify_sync(input(Severity::Error), Duration::from_millis(100))
1013            .expect("notify_sync");
1014
1015        assert_eq!(result.status, DeliveryStatus::Partial);
1016        assert_eq!(result.provider.as_deref(), Some("ok"));
1017        assert_eq!(
1018            result.error.as_deref(),
1019            Some("hanging: timed out after 100ms")
1020        );
1021        assert_eq!(
1022            stored_rows(&db),
1023            vec![("partial".to_string(), Some("ok".to_string()))]
1024        );
1025        assert_eq!(
1026            delivery_rows(&db),
1027            vec![
1028                (
1029                    "hanging".into(),
1030                    "failed".into(),
1031                    Some("timed out after 100ms".into())
1032                ),
1033                ("ok".into(), "delivered".into(), None),
1034            ]
1035        );
1036        let _ = std::fs::remove_file(&db);
1037    }
1038
1039    #[test]
1040    fn provider_is_saved_as_pending_before_its_send() {
1041        let db = temp_db();
1042        let observer = Arc::new(DeliveriesObserver {
1043            db: db.clone(),
1044            seen: Default::default(),
1045        });
1046        let client = client_with(vec![observer.clone()], &["observer"], &db);
1047
1048        client
1049            .notify_sync(input(Severity::Error), Duration::from_secs(5))
1050            .expect("notify_sync");
1051
1052        assert_eq!(
1053            *observer.seen.lock().expect("seen"),
1054            vec![("observer".to_string(), "pending".to_string(), None)]
1055        );
1056        assert_eq!(
1057            delivery_rows(&db),
1058            vec![("observer".into(), "delivered".into(), None)]
1059        );
1060        let _ = std::fs::remove_file(&db);
1061    }
1062
1063    /// Captures the per-provider rows at the moment of delivery.
1064    struct DeliveriesObserver {
1065        db: String,
1066        seen: std::sync::Mutex<Vec<(String, String, Option<String>)>>,
1067    }
1068
1069    #[async_trait::async_trait]
1070    impl Provider for DeliveriesObserver {
1071        fn name(&self) -> &'static str {
1072            "observer"
1073        }
1074
1075        async fn send(&self, _signal: &Signal) -> Result<(), SignalError> {
1076            *self.seen.lock().expect("seen") = delivery_rows(&self.db);
1077            Ok(())
1078        }
1079    }
1080
1081    #[test]
1082    fn notify_async_fanout_records_each_provider() {
1083        let db = temp_db();
1084        let a = FlakyProvider::new("a", 0);
1085        let b = FlakyProvider::new("b", 0);
1086        let client = client_with_routing(vec![a.clone(), b.clone()], fanout_to(&["a", "b"]), &db);
1087
1088        client.notify_async(input(Severity::Error));
1089
1090        assert!(client.flush(Duration::from_secs(5)));
1091        assert_eq!(
1092            stored_rows(&db),
1093            vec![("delivered".to_string(), Some("a,b".to_string()))]
1094        );
1095        assert_eq!(
1096            delivery_rows(&db),
1097            vec![
1098                ("a".into(), "delivered".into(), None),
1099                ("b".into(), "delivered".into(), None),
1100            ]
1101        );
1102        let _ = std::fs::remove_file(&db);
1103    }
1104
1105    fn client_with(providers: Vec<Arc<dyn Provider>>, route: &[&str], db: &str) -> SignalClient {
1106        client_with_routing(providers, routing_to(route), db)
1107    }
1108
1109    fn client_with_routing(
1110        providers: Vec<Arc<dyn Provider>>,
1111        routing: RoutingConfig,
1112        db: &str,
1113    ) -> SignalClient {
1114        SignalClient {
1115            sink: Arc::new(SqliteSink::open(db).expect("open sink")),
1116            pipeline: Arc::new(RwLock::new(Arc::new(Pipeline {
1117                routing,
1118                providers: registry(providers),
1119            }))),
1120            throttle: Some(Arc::new(Throttle::open(db).expect("open throttle"))),
1121            in_flight: Arc::default(),
1122            runtime: tokio::runtime::Builder::new_multi_thread()
1123                .worker_threads(1)
1124                .enable_all()
1125                .build()
1126                .expect("runtime"),
1127        }
1128    }
1129
1130    #[test]
1131    fn notify_sync_times_out_and_records_the_failure() {
1132        let db = temp_db();
1133        let client = client_with(vec![Arc::new(HangingProvider)], &["hanging"], &db);
1134
1135        let result = client
1136            .notify_sync(input(Severity::Error), Duration::from_millis(50))
1137            .expect("notify_sync returns a result, not an error, on timeout");
1138
1139        assert_eq!(result.status, DeliveryStatus::Failed);
1140        assert_eq!(result.attempts, 0);
1141        assert!(result
1142            .error
1143            .as_deref()
1144            .is_some_and(|e| e.contains("timed out")));
1145        assert_eq!(stored_rows(&db), vec![("failed".to_string(), None)]);
1146        let _ = std::fs::remove_file(&db);
1147    }
1148
1149    #[test]
1150    fn notify_sync_rejects_invalid_input_without_recording_it() {
1151        let db = temp_db();
1152        let client = client_with(vec![FlakyProvider::new("ok", 0)], &["ok"], &db);
1153        let mut bad = input(Severity::Info);
1154        bad.title = "  ".into();
1155
1156        let err = client
1157            .notify_sync(bad, Duration::from_secs(1))
1158            .expect_err("empty title must be rejected");
1159
1160        assert!(matches!(err, SignalError::Validation(_)));
1161        assert!(stored_rows(&db).is_empty());
1162        let _ = std::fs::remove_file(&db);
1163    }
1164
1165    #[test]
1166    fn notify_async_delivers_in_the_background_and_records_it() {
1167        let db = temp_db();
1168        let ok = FlakyProvider::new("ok", 0);
1169        let client = client_with(vec![ok.clone()], &["ok"], &db);
1170
1171        client.notify_async(input(Severity::Error));
1172
1173        assert!(client.flush(Duration::from_secs(5)));
1174        assert_eq!(
1175            stored_rows(&db),
1176            vec![("delivered".to_string(), Some("ok".to_string()))]
1177        );
1178        assert_eq!(ok.calls(), 1);
1179        let _ = std::fs::remove_file(&db);
1180    }
1181
1182    /// Sleeps for `delay` and then succeeds.
1183    struct SlowProvider(Duration);
1184
1185    #[async_trait::async_trait]
1186    impl Provider for SlowProvider {
1187        fn name(&self) -> &'static str {
1188            "slow"
1189        }
1190
1191        async fn send(&self, _signal: &Signal) -> Result<(), SignalError> {
1192            tokio::time::sleep(self.0).await;
1193            Ok(())
1194        }
1195    }
1196
1197    /// Captures what the sink holds at the moment of delivery.
1198    struct SinkObserver {
1199        db: String,
1200        seen: std::sync::Mutex<Vec<(String, Option<String>)>>,
1201    }
1202
1203    #[async_trait::async_trait]
1204    impl Provider for SinkObserver {
1205        fn name(&self) -> &'static str {
1206            "observer"
1207        }
1208
1209        async fn send(&self, _signal: &Signal) -> Result<(), SignalError> {
1210            *self.seen.lock().expect("seen") = stored_rows(&self.db);
1211            Ok(())
1212        }
1213    }
1214
1215    #[test]
1216    fn signal_is_saved_as_pending_before_delivery() {
1217        let db = temp_db();
1218        let observer = Arc::new(SinkObserver {
1219            db: db.clone(),
1220            seen: Default::default(),
1221        });
1222        let client = client_with(vec![observer.clone()], &["observer"], &db);
1223
1224        let result = client
1225            .notify_sync(input(Severity::Error), Duration::from_secs(5))
1226            .expect("notify_sync");
1227
1228        assert_eq!(result.status, DeliveryStatus::Delivered);
1229        assert_eq!(
1230            *observer.seen.lock().expect("seen"),
1231            vec![("pending".to_string(), None)]
1232        );
1233        assert_eq!(
1234            stored_rows(&db),
1235            vec![("delivered".to_string(), Some("observer".to_string()))]
1236        );
1237        let _ = std::fs::remove_file(&db);
1238    }
1239
1240    #[test]
1241    fn notify_async_saves_the_signal_before_returning() {
1242        let db = temp_db();
1243        let client = client_with(
1244            vec![Arc::new(SlowProvider(Duration::from_millis(200)))],
1245            &["slow"],
1246            &db,
1247        );
1248
1249        client.notify_async(input(Severity::Error));
1250
1251        assert_eq!(stored_rows(&db), vec![("pending".to_string(), None)]);
1252        drop(client);
1253        let _ = std::fs::remove_file(&db);
1254    }
1255
1256    #[test]
1257    fn flush_waits_for_background_deliveries() {
1258        let db = temp_db();
1259        let client = client_with(
1260            vec![Arc::new(SlowProvider(Duration::from_millis(100)))],
1261            &["slow"],
1262            &db,
1263        );
1264
1265        client.notify_async(input(Severity::Error));
1266        client.notify_async(input(Severity::Error));
1267
1268        assert!(client.flush(Duration::from_secs(5)));
1269        assert_eq!(
1270            stored_rows(&db),
1271            vec![("delivered".to_string(), Some("slow".to_string())); 2]
1272        );
1273        let _ = std::fs::remove_file(&db);
1274    }
1275
1276    #[test]
1277    fn flush_gives_up_after_the_timeout_and_leaves_the_signal_pending() {
1278        let db = temp_db();
1279        let client = client_with(
1280            vec![Arc::new(SlowProvider(Duration::from_millis(300)))],
1281            &["slow"],
1282            &db,
1283        );
1284
1285        client.notify_async(input(Severity::Error));
1286
1287        assert!(!client.flush(Duration::from_millis(20)));
1288        assert_eq!(stored_rows(&db), vec![("pending".to_string(), None)]);
1289        let _ = std::fs::remove_file(&db);
1290    }
1291
1292    #[test]
1293    fn dropping_the_client_finishes_background_deliveries() {
1294        let db = temp_db();
1295        let client = client_with(
1296            vec![Arc::new(SlowProvider(Duration::from_millis(100)))],
1297            &["slow"],
1298            &db,
1299        );
1300
1301        client.notify_async(input(Severity::Error));
1302        drop(client);
1303
1304        assert_eq!(
1305            stored_rows(&db),
1306            vec![("delivered".to_string(), Some("slow".to_string()))]
1307        );
1308        let _ = std::fs::remove_file(&db);
1309    }
1310
1311    #[test]
1312    fn a_sink_that_cannot_record_still_lets_the_signal_through() {
1313        struct BrokenRecord;
1314        impl SignalSink for BrokenRecord {
1315            fn record(&self, _: &Signal) -> Result<(), SignalError> {
1316                Err(SignalError::Sink("disk full".into()))
1317            }
1318            fn complete(&self, _: &Signal, _: &DeliveryResult) -> Result<(), SignalError> {
1319                Ok(())
1320            }
1321        }
1322        let ok = FlakyProvider::new("ok", 0);
1323        let mut client = client_with(vec![ok.clone()], &["ok"], &temp_db());
1324        client.sink = Arc::new(BrokenRecord);
1325
1326        let result = client
1327            .notify_sync(input(Severity::Error), Duration::from_secs(5))
1328            .expect("notify_sync");
1329
1330        assert_eq!(result.status, DeliveryStatus::Delivered);
1331        assert_eq!(ok.calls(), 1);
1332    }
1333
1334    #[test]
1335    fn notify_async_drops_invalid_input_without_delivering() {
1336        let db = temp_db();
1337        let ok = FlakyProvider::new("ok", 0);
1338        let client = client_with(vec![ok.clone()], &["ok"], &db);
1339        let mut bad = input(Severity::Info);
1340        bad.source = String::new();
1341
1342        client.notify_async(bad);
1343        std::thread::sleep(Duration::from_millis(100));
1344
1345        assert_eq!(ok.calls(), 0);
1346        assert!(stored_rows(&db).is_empty());
1347        let _ = std::fs::remove_file(&db);
1348    }
1349
1350    #[test]
1351    fn from_config_registers_only_configured_providers() {
1352        let db = temp_db();
1353        let yaml = format!(
1354            "providers:\n  slack:\n    webhook_url: \"https://hooks.example.test/x\"\nstorage:\n  path: \"{db}\"\n"
1355        );
1356        let client = SignalClient::from_config(&Config::from_yaml_str(&yaml).expect("config"))
1357            .expect("client");
1358
1359        let pipeline = client.pipeline();
1360        let mut names: Vec<&str> = pipeline.providers.keys().map(String::as_str).collect();
1361        names.sort_unstable();
1362        assert_eq!(names, vec!["console", "slack"]);
1363        let _ = std::fs::remove_file(&db);
1364    }
1365
1366    #[test]
1367    fn from_config_registers_pagerduty_and_teams() {
1368        let db = temp_db();
1369        let yaml = format!(
1370            "providers:\n  pagerduty:\n    routing_key: k\n  teams:\n    webhook_url: \"https://example.test/t\"\nstorage:\n  path: \"{db}\"\n"
1371        );
1372        let client = SignalClient::from_config(&Config::from_yaml_str(&yaml).expect("config"))
1373            .expect("client");
1374
1375        let pipeline = client.pipeline();
1376        let mut names: Vec<&str> = pipeline.providers.keys().map(String::as_str).collect();
1377        names.sort_unstable();
1378        assert_eq!(names, vec!["console", "pagerduty", "teams"]);
1379        let _ = std::fs::remove_file(&db);
1380    }
1381
1382    #[test]
1383    fn failed_pagerduty_falls_back_to_teams() {
1384        let server_rt = tokio::runtime::Runtime::new().expect("server runtime");
1385        let server = server_rt.block_on(async {
1386            let server = MockServer::start().await;
1387            Mock::given(method("POST"))
1388                .and(path("/pd"))
1389                .respond_with(ResponseTemplate::new(400))
1390                .mount(&server)
1391                .await;
1392            Mock::given(method("POST"))
1393                .and(path("/teams"))
1394                .respond_with(ResponseTemplate::new(202))
1395                .expect(1)
1396                .mount(&server)
1397                .await;
1398            server
1399        });
1400
1401        let db = temp_db();
1402        let uri = server.uri();
1403        let yaml = format!(
1404            "providers:\n  pagerduty:\n    routing_key: k\n    url: \"{uri}/pd\"\n  teams:\n    webhook_url: \"{uri}/teams\"\nroutes:\n  - match:\n      severity: [error]\n    providers: [pagerduty, teams]\nstorage:\n  path: \"{db}\"\n"
1405        );
1406        let client = SignalClient::from_config(&Config::from_yaml_str(&yaml).expect("config"))
1407            .expect("client");
1408
1409        let result = client
1410            .notify_sync(input(Severity::Error), Duration::from_secs(30))
1411            .expect("notify_sync");
1412
1413        assert_eq!(result.status, DeliveryStatus::Delivered);
1414        assert_eq!(result.provider.as_deref(), Some("teams"));
1415        server_rt.block_on(server.verify());
1416        let _ = std::fs::remove_file(&db);
1417    }
1418
1419    #[test]
1420    fn fanout_route_pages_pagerduty_and_posts_to_teams() {
1421        let server_rt = tokio::runtime::Runtime::new().expect("server runtime");
1422        let server = server_rt.block_on(async {
1423            let server = MockServer::start().await;
1424            for p in ["/pd", "/teams"] {
1425                Mock::given(method("POST"))
1426                    .and(path(p))
1427                    .respond_with(ResponseTemplate::new(202))
1428                    .expect(1)
1429                    .mount(&server)
1430                    .await;
1431            }
1432            server
1433        });
1434
1435        let db = temp_db();
1436        let uri = server.uri();
1437        let yaml = format!(
1438            "providers:\n  pagerduty:\n    routing_key: k\n    url: \"{uri}/pd\"\n  teams:\n    webhook_url: \"{uri}/teams\"\nroutes:\n  - match:\n      severity: [error]\n    providers: [pagerduty, teams]\n    mode: fanout\nstorage:\n  path: \"{db}\"\n"
1439        );
1440        let client = SignalClient::from_config(&Config::from_yaml_str(&yaml).expect("config"))
1441            .expect("client");
1442
1443        let result = client
1444            .notify_sync(input(Severity::Error), Duration::from_secs(30))
1445            .expect("notify_sync");
1446
1447        assert_eq!(result.status, DeliveryStatus::Delivered);
1448        assert_eq!(result.provider.as_deref(), Some("pagerduty,teams"));
1449        server_rt.block_on(server.verify());
1450        let _ = std::fs::remove_file(&db);
1451    }
1452
1453    #[test]
1454    fn failed_opsgenie_falls_back_to_slack() {
1455        let server_rt = tokio::runtime::Runtime::new().expect("server runtime");
1456        let server = server_rt.block_on(async {
1457            let server = MockServer::start().await;
1458            Mock::given(method("POST"))
1459                .and(path("/og"))
1460                .respond_with(ResponseTemplate::new(422))
1461                .mount(&server)
1462                .await;
1463            Mock::given(method("POST"))
1464                .and(path("/slack"))
1465                .respond_with(ResponseTemplate::new(200))
1466                .expect(1)
1467                .mount(&server)
1468                .await;
1469            server
1470        });
1471
1472        let db = temp_db();
1473        let uri = server.uri();
1474        let yaml = format!(
1475            "providers:\n  opsgenie:\n    api_key: k\n    url: \"{uri}/og\"\n  slack:\n    webhook_url: \"{uri}/slack\"\nroutes:\n  - match:\n      severity: [error]\n    providers: [opsgenie, slack]\nstorage:\n  path: \"{db}\"\n"
1476        );
1477        let client = SignalClient::from_config(&Config::from_yaml_str(&yaml).expect("config"))
1478            .expect("client");
1479
1480        let result = client
1481            .notify_sync(input(Severity::Error), Duration::from_secs(30))
1482            .expect("notify_sync");
1483
1484        assert_eq!(result.status, DeliveryStatus::Delivered);
1485        assert_eq!(result.provider.as_deref(), Some("slack"));
1486        server_rt.block_on(server.verify());
1487        let _ = std::fs::remove_file(&db);
1488    }
1489
1490    #[test]
1491    fn from_config_fails_when_the_sink_cannot_be_opened() {
1492        let yaml = "storage:\n  path: \"/nonexistent-opssignal-dir/sub/signals.db\"\n";
1493        let err = SignalClient::from_config(&Config::from_yaml_str(yaml).expect("config"))
1494            .err()
1495            .expect("unopenable sink must fail");
1496        assert!(matches!(err, SignalError::Sink(_)));
1497    }
1498
1499    fn throttled(route: routing::Route, dedup: Option<&str>, rate: Option<&str>) -> routing::Route {
1500        routing::Route {
1501            dedup: dedup.map(|w| w.parse().expect("window")),
1502            rate_limit: rate.map(|r| serde_yaml::from_str(r).expect("rate_limit")),
1503            ..route
1504        }
1505    }
1506
1507    fn deduped(providers: &[&str], mode: routing::DeliveryMode) -> RoutingConfig {
1508        let mut routing = route(providers, mode);
1509        routing.routes[0] = throttled(routing.routes[0].clone(), Some("10m"), None);
1510        routing
1511    }
1512
1513    #[test]
1514    fn dedup_sends_an_incident_once_per_window() {
1515        let db = temp_db();
1516        let pd = FlakyProvider::new("pagerduty", 0);
1517        let client = client_with_routing(
1518            vec![pd.clone()],
1519            deduped(&["pagerduty"], routing::DeliveryMode::Failover),
1520            &db,
1521        );
1522
1523        let first = client
1524            .notify_sync(input(Severity::Error), Duration::from_secs(5))
1525            .expect("first");
1526        assert_eq!(first.status, DeliveryStatus::Delivered);
1527        assert!(first.suppressed.is_empty());
1528
1529        for _ in 0..5 {
1530            let repeat = client
1531                .notify_sync(input(Severity::Error), Duration::from_secs(5))
1532                .expect("repeat");
1533            assert_eq!(repeat.status, DeliveryStatus::Skipped);
1534            assert!(repeat.deliveries.is_empty());
1535            assert_eq!(repeat.suppressed.len(), 1);
1536            let error = repeat.error.expect("reason");
1537            assert!(
1538                error.starts_with("route [pagerduty]: duplicate of a signal sent"),
1539                "{error}"
1540            );
1541        }
1542        assert_eq!(pd.calls(), 1);
1543
1544        // A different incident is not a duplicate.
1545        let mut other = input(Severity::Error);
1546        other.title = "another model failed".into();
1547        let other = client
1548            .notify_sync(other, Duration::from_secs(5))
1549            .expect("other");
1550        assert_eq!(other.status, DeliveryStatus::Delivered);
1551        assert_eq!(pd.calls(), 2);
1552
1553        // Suppressed signals are still recorded.
1554        let rows = stored_rows(&db);
1555        assert_eq!(rows.len(), 7);
1556        assert_eq!(
1557            rows.iter()
1558                .filter(|(status, _)| status == "skipped")
1559                .count(),
1560            5
1561        );
1562        drop(client);
1563        let _ = std::fs::remove_file(&db);
1564    }
1565
1566    #[test]
1567    fn suppressed_route_leaves_other_matching_routes_sending() {
1568        let db = temp_db();
1569        let pd = FlakyProvider::new("pagerduty", 0);
1570        let slack = FlakyProvider::new("slack", 0);
1571        let mut routing = deduped(&["pagerduty"], routing::DeliveryMode::Fanout);
1572        routing.routes.push(
1573            route(&["slack"], routing::DeliveryMode::Failover)
1574                .routes
1575                .remove(0),
1576        );
1577        let client = client_with_routing(vec![pd.clone(), slack.clone()], routing, &db);
1578
1579        for _ in 0..3 {
1580            let _ = client
1581                .notify_sync(input(Severity::Error), Duration::from_secs(5))
1582                .expect("notify");
1583        }
1584        let last = client
1585            .notify_sync(input(Severity::Error), Duration::from_secs(5))
1586            .expect("notify");
1587
1588        assert_eq!(last.status, DeliveryStatus::Delivered);
1589        assert_eq!(last.provider.as_deref(), Some("slack"));
1590        assert_eq!(last.suppressed.len(), 1);
1591        assert_eq!(pd.calls(), 1);
1592        assert_eq!(slack.calls(), 4);
1593        drop(client);
1594        let _ = std::fs::remove_file(&db);
1595    }
1596
1597    #[test]
1598    fn rate_limit_caps_sends_across_incidents() {
1599        let db = temp_db();
1600        let pd = FlakyProvider::new("pagerduty", 0);
1601        let mut routing = route(&["pagerduty"], routing::DeliveryMode::Failover);
1602        routing.routes[0] = throttled(routing.routes[0].clone(), None, Some("{max: 3, per: 1h}"));
1603        let client = client_with_routing(vec![pd.clone()], routing, &db);
1604
1605        let statuses: Vec<DeliveryStatus> = (0..10)
1606            .map(|i| {
1607                let mut signal = input(Severity::Error);
1608                signal.title = format!("model {i} failed");
1609                client
1610                    .notify_sync(signal, Duration::from_secs(5))
1611                    .expect("notify")
1612                    .status
1613            })
1614            .collect();
1615
1616        assert_eq!(pd.calls(), 3);
1617        assert_eq!(
1618            statuses
1619                .iter()
1620                .filter(|s| **s == DeliveryStatus::Skipped)
1621                .count(),
1622            7
1623        );
1624        drop(client);
1625        let _ = std::fs::remove_file(&db);
1626    }
1627
1628    #[test]
1629    fn a_failed_send_does_not_start_the_dedup_window() {
1630        let db = temp_db();
1631        // Fails all 3 attempts of the first signal, then recovers.
1632        let pd = FlakyProvider::new("pagerduty", MAX_RETRY_ATTEMPTS);
1633        let client = client_with_routing(
1634            vec![pd.clone()],
1635            deduped(&["pagerduty"], routing::DeliveryMode::Failover),
1636            &db,
1637        );
1638
1639        let first = client
1640            .notify_sync(input(Severity::Error), Duration::from_secs(5))
1641            .expect("first");
1642        assert_eq!(first.status, DeliveryStatus::Failed);
1643        let retry = client
1644            .notify_sync(input(Severity::Error), Duration::from_secs(5))
1645            .expect("retry");
1646        assert_eq!(retry.status, DeliveryStatus::Delivered);
1647        let repeat = client
1648            .notify_sync(input(Severity::Error), Duration::from_secs(5))
1649            .expect("repeat");
1650        assert_eq!(repeat.status, DeliveryStatus::Skipped);
1651        drop(client);
1652        let _ = std::fs::remove_file(&db);
1653    }
1654
1655    #[test]
1656    fn notify_async_applies_dedup_too() {
1657        let db = temp_db();
1658        let pd = FlakyProvider::new("pagerduty", 0);
1659        let client = client_with_routing(
1660            vec![pd.clone()],
1661            deduped(&["pagerduty"], routing::DeliveryMode::Failover),
1662            &db,
1663        );
1664        for _ in 0..20 {
1665            client.notify_async(input(Severity::Error));
1666        }
1667        assert!(client.flush(Duration::from_secs(5)));
1668        assert_eq!(pd.calls(), 1);
1669        drop(client);
1670        let _ = std::fs::remove_file(&db);
1671    }
1672}
opssignal_core/lib.rs

opssignal_core/
lib.rs
