fix(synthetics): alert on the merged vote of a confirm run

- A run that held a confirm job alerts on its merged class, not on the
  worst raw job or the last ack's error source.
- A confirm run that finalizes in the reaper or dispatcher resolves its
  alert there and notifies through a registered run notifier.
- A confirmed failure below quorum is partial, even with a healthy
  majority.
- The run row stores the merged run result.

src/api/management/src/request/synthetics/mod.rs

@@ -1705,46 +1705,23 @@ async fn process_ack(
     // 3` alerted on the first failure and a 30-minute cooldown sent thirty
     // notifications.
     #[cfg(feature = "enterprise")]
-    use openobserve_synthetics::job_api::AlertDecision;
-    #[cfg(feature = "enterprise")]
-    let recovery = matches!(resp.alert, AlertDecision::Recovered);
-    #[cfg(feature = "enterprise")]
-    let flaky = matches!(resp.alert, AlertDecision::Flaky);
-    // A degrading target is `warning` on every run for as long as the condition
-    // lasts, so this one is throttled by transition upstream, not by cooldown.
-    #[cfg(feature = "enterprise")]
-    let degraded = matches!(resp.alert, AlertDecision::Degraded);
-    #[cfg(feature = "enterprise")]
-    let should_notify =
-        !matches!(resp.alert, AlertDecision::Silent) && resp.suppressed_by.is_none();
+    let should_notify = !matches!(
+        resp.alert,
+        openobserve_synthetics::job_api::AlertDecision::Silent
+    ) && resp.suppressed_by.is_none();
     // Enterprise-gated because alert *destinations* are: the dispatch it ends in
     // (`alerts::alert::dispatch_notification`) is not built in OSS. An OSS build
     // runs the check, records the run and serves the result — it just has
     // nowhere to send a notification, exactly as it has for alerts.
     #[cfg(feature = "enterprise")]
     if should_notify && !resp.destinations.is_empty() {
-        let notification = openobserve_core::synthetics::CheckNotification {
-            org_id: resp.org_id.clone(),
-            check_name: resp.synthetics_name.clone(),
-            check_id: resp.synthetics_id.clone(),
-            check_type: resp.synthetic_type.clone(),
-            target: resp.target.clone(),
-            destinations: resp.destinations.clone(),
-            run_id: resp.run_id.clone(),
-            status: resp.run_status.clone().unwrap_or_else(|| status.clone()),
-            job_count: resp.job_count as i64,
-            error: error.clone(),
+        let notification = openobserve_core::synthetics::CheckNotification::from_ack(
+            &resp,
+            &status,
+            error.clone(),
+            error_source.clone(),
             checked_at,
-            recovery,
-            consecutive_failures: resp.consecutive_failures,
-            flaky,
-            degraded,
-            status_reason: resp.status_reason.clone(),
-            error_source: error_source.clone(),
-            failing_locations: resp.failing_locations.clone(),
-            failing_environments: resp.failing_environments.clone(),
-            passing_locations: resp.passing_locations.clone(),
-        };
+        );
         tokio::spawn(async move {
             openobserve_core::synthetics::notify_check_result(notification).await;
         });

src/core/src/synthetics.rs

@@ -76,6 +76,65 @@ pub struct CheckNotification {
     pub failing_environments: Vec<String>,
 }
 
+#[cfg(feature = "enterprise")]
+impl CheckNotification {
+    /// The notification for a completed run. `status`, `error` and `error_source` describe the
+    /// ack that completed it, and `status` is the fallback when the run has no aggregate status.
+    pub fn from_ack(
+        resp: &openobserve_synthetics::job_api::AckResponse,
+        status: &str,
+        error: Option<String>,
+        error_source: String,
+        checked_at: i64,
+    ) -> Self {
+        use openobserve_synthetics::job_api::AlertDecision;
+        Self {
+            org_id: resp.org_id.clone(),
+            check_name: resp.synthetics_name.clone(),
+            check_id: resp.synthetics_id.clone(),
+            check_type: resp.synthetic_type.clone(),
+            target: resp.target.clone(),
+            destinations: resp.destinations.clone(),
+            run_id: resp.run_id.clone(),
+            status: resp
+                .run_status
+                .clone()
+                .unwrap_or_else(|| status.to_string()),
+            job_count: resp.job_count as i64,
+            error,
+            checked_at,
+            recovery: matches!(resp.alert, AlertDecision::Recovered),
+            consecutive_failures: resp.consecutive_failures,
+            flaky: matches!(resp.alert, AlertDecision::Flaky),
+            // A degrading target is `warning` on every run for as long as the condition lasts,
+            // so this one is throttled by transition upstream, not by cooldown.
+            degraded: matches!(resp.alert, AlertDecision::Degraded),
+            status_reason: resp.status_reason.clone(),
+            error_source,
+            failing_locations: resp.failing_locations.clone(),
+            failing_environments: resp.failing_environments.clone(),
+            passing_locations: resp.passing_locations.clone(),
+        }
+    }
+}
+
+/// Sends the notification of a run that finalized outside an ack, for
+/// `openobserve_synthetics::job_api::register_run_notifier`.
+#[cfg(feature = "enterprise")]
+pub fn notify_finalized_run(
+    resp: openobserve_synthetics::job_api::AckResponse,
+) -> openobserve_synthetics::alerting::BoxFuture<'static, ()> {
+    Box::pin(async move {
+        if resp.destinations.is_empty() {
+            return;
+        }
+        let checked_at = config::utils::time::now_micros();
+        let status = resp.run_status.clone().unwrap_or_default();
+        let n = CheckNotification::from_ack(&resp, &status, None, String::new(), checked_at);
+        notify_check_result(n).await;
+    })
+}
+
 /// Fires once per run (when all jobs have completed) for non-passing runs.
 /// Passing runs are suppressed — operators want alerts, not confirmations.
 ///

src/infra/src/table/synthetics_runs.rs

@@ -111,6 +111,8 @@ pub struct RunFinalized {
     pub run_class: RunClass,
     /// A confirm job passed, so a failed location counts as up (`flaky`).
     pub confirmed_flaky: bool,
+    /// The run held a confirm job, so its answer is the merged vote, not the worst raw result.
+    pub had_confirm: bool,
     /// The run earns one extra run: it is a scheduled run that measured nothing because the probe
     /// could not reach the job API or could not be dispatched.
     pub gap_retry: bool,
@@ -139,6 +141,7 @@ pub struct Tally {
     /// the run has no confirm job, so the stored `run_result` stands.
     pub run_result: Option<i32>,
     pub confirmed_flaky: bool,
+    pub has_confirm: bool,
     /// Some job measured nothing because of [`GAP_RETRY_SOURCES`].
     pub gap_cause: bool,
 }
@@ -220,16 +223,17 @@ pub fn tally(job_count: i32, jobs: &[JobOutcome]) -> Tally {
         _ if failed + healthy == 0 && config > 0 => RunClass::Misconfigured,
         _ if failed + healthy == 0 => RunClass::NotMeasured,
         Verdict::Down => RunClass::Down,
-        Verdict::Up => RunClass::Up,
-        Verdict::NoVerdict if failed > 0 => RunClass::Partial,
-        // Only healthy locations tested the site, too few for a verdict.
-        Verdict::NoVerdict => RunClass::Up,
+        // A failure below quorum is partial, even when a healthy majority makes the verdict Up.
+        _ if failed > 0 => RunClass::Partial,
+        // Only healthy locations tested the site.
+        _ => RunClass::Up,
     };
     Tally {
         verdict,
         class,
         run_result: (!confirms.is_empty() && worst > 0).then_some(worst),
         confirmed_flaky,
+        has_confirm: !confirms.is_empty(),
         gap_cause: jobs.iter().any(JobOutcome::is_gap_cause),
     }
 }
@@ -518,6 +522,17 @@ async fn finalize<C: ConnectionTrait + TransactionTrait>(
         .or(run.run_result)
         .or(job_result)
         .unwrap_or(STATUS_ERROR);
+    if t.run_result
+        .is_some_and(|merged| Some(merged) != run.run_result)
+    {
+        // The run history shows the merged vote, as the check row does.
+        txn.execute(Statement::from_sql_and_values(
+            txn.get_database_backend(),
+            "UPDATE synthetics_runs SET run_result = $1 WHERE id = $2",
+            [Value::from(run_result), Value::from(run_id.to_owned())],
+        ))
+        .await?;
+    }
     let check_update = write_check_row(
         &txn,
         &run.synthetics_id,
@@ -537,6 +552,7 @@ async fn finalize<C: ConnectionTrait + TransactionTrait>(
         verdict: t.verdict,
         run_class: t.class,
         confirmed_flaky: t.confirmed_flaky,
+        had_confirm: t.has_confirm,
         gap_retry: t.class == RunClass::NotMeasured
             && t.gap_cause
             && run.trigger_type == TRIGGER_SCHEDULE,
@@ -1508,6 +1524,21 @@ mod tests {
             ],
         );
         assert_eq!((two.verdict, two.class), (Verdict::Up, RunClass::Up));
+        // A confirmed failure below quorum is partial, even with a healthy majority.
+        let regional = tally(
+            4,
+            &[
+                job("a", JOB_FAILED),
+                job("b", JOB_PASSED),
+                job("d", JOB_PASSED),
+                confirm("c", "a", JOB_FAILED),
+            ],
+        );
+        assert_eq!(
+            (regional.verdict, regional.class),
+            (Verdict::Up, RunClass::Partial)
+        );
+        assert!(regional.has_confirm);
         // No confirm job: the stored run result stands.
         assert_eq!(tally(1, &[job("a", JOB_FAILED)]).run_result, None);
     }
@@ -1651,7 +1682,13 @@ mod tests {
         let done = finalize_run(&db, "r", 1, 300).await.unwrap().unwrap();
         assert_eq!(done.run_class, RunClass::Up);
         assert!(done.confirmed_flaky);
+        assert!(done.had_confirm);
         assert_eq!(done.run_result, 2);
+        assert_eq!(
+            get_run(&db, "r").await.unwrap().unwrap().run_result,
+            Some(2),
+            "the run history shows the merged vote"
+        );
     }
 
     #[tokio::test]

src/jobs/src/job/mod.rs

@@ -942,6 +942,12 @@ pub async fn init() -> Result<(), anyhow::Error> {
             openobserve_core::synthetics::downtime_may_mute,
         );
     }
+    // A run a confirm job kept open can finalize in the reaper or the dispatcher, where no ack
+    // sends its alert.
+    #[cfg(feature = "enterprise")]
+    openobserve_synthetics::job_api::register_run_notifier(
+        openobserve_core::synthetics::notify_finalized_run,
+    );
     // Warm the cache on queriers (UI APIs) and on whichever node role is the configured
     // processing node so that get_coverage_deficit returns accurate data from startup
     // rather than always returning (0, 0) until files happen to be processed.

src/synthetics/src/dispatcher.rs

@@ -356,7 +356,10 @@ async fn mark_failure(row: &synthetics_jobs::LeasedRow, reason: Option<&str>) ->
         return false;
     }
     match synthetics_runs::finalize_run(db, &row.run_id, 4, now_us).await {
-        Ok(Some(done)) => crate::job_api::after_run_finalized(&done).await,
+        Ok(Some(done)) => {
+            crate::job_api::after_run_finalized(&done).await;
+            crate::job_api::alert_for_swept_run(&done).await;
+        }
         Ok(None) => {}
         Err(e) => {
             tracing::error!(synthetics_id = %row.synthetics_id, run_id = %row.run_id, job_id = %row.id, "[synthetics dispatcher] finalize_run: {e}");

src/synthetics/src/job_api.rs

@@ -1680,6 +1680,102 @@ pub(crate) fn wants_confirm_job(
         && !synthetics_runs::is_not_measured_source(error_source, status_reason)
 }
 
+/// Sends the notification for a run that finalized outside an ack; the server registers it, as
+/// notification delivery lives above this crate.
+pub type RunNotifier = fn(AckResponse) -> crate::alerting::BoxFuture<'static, ()>;
+
+static RUN_NOTIFIER: std::sync::OnceLock<RunNotifier> = std::sync::OnceLock::new();
+
+pub fn register_run_notifier(f: RunNotifier) {
+    let _ = RUN_NOTIFIER.set(f);
+}
+
+/// The alert inputs of a run that held a confirm job: `(status, status_reason, error_source)`
+/// from the merged vote, so neither the worst raw job nor the last ack decides the alert.
+pub(crate) fn merged_alert_input(
+    class: config::meta::synthetics::RunClass,
+    confirmed_flaky: bool,
+) -> (&'static str, Option<&'static str>, &'static str) {
+    use config::meta::synthetics::RunClass as C;
+    match class {
+        C::Up if confirmed_flaky => ("warning", Some(crate::alerting::REASON_FLAKY), ""),
+        C::Up => ("passed", None, ""),
+        C::Partial | C::Down => ("failed", None, ""),
+        C::NotMeasured => ("error", None, crate::alerting::ERROR_SOURCE_CONTROL_PLANE),
+        C::Misconfigured => ("error", None, crate::alerting::ERROR_SOURCE_CONFIG),
+        C::Unknown => ("error", None, ""),
+    }
+}
+
+/// Decides and sends the alert of a run that a confirm job kept open and that finalized outside
+/// an ack: a dead letter, a failed dispatch or the stuck-run sweep. No ack runs the alert for it.
+pub(crate) async fn alert_for_swept_run(done: &synthetics_runs::RunFinalized) {
+    if !done.had_confirm {
+        return;
+    }
+    let conn = get_orm_client_rw().await;
+    let Ok(Some(synthetic)) =
+        synthetics_checks::get_cached(conn, &done.org_id, &done.synthetics_id).await
+    else {
+        return;
+    };
+    let (status, status_reason, error_source) =
+        merged_alert_input(done.run_class, done.confirmed_flaky);
+    let now_us = config::utils::time::now_micros();
+    let (alert, consecutive_failures, suppressed_by) = resolve_alert(
+        conn,
+        &done.synthetics_id,
+        RunOutcome {
+            status: Some(status),
+            attempts: 1,
+            status_reason,
+            error_source,
+        },
+        &synthetic,
+        now_us,
+    )
+    .await;
+    if matches!(alert, AlertDecision::Silent) || suppressed_by.is_some() {
+        return;
+    }
+    let Some(notify) = RUN_NOTIFIER.get() else {
+        return;
+    };
+    let outcomes = synthetics_jobs::run_location_outcomes(conn, &done.run_id)
+        .await
+        .unwrap_or_default();
+    let failing_environments = environment_names(&done.org_id, &done.run_id).await;
+    let synthetic_type = serde_json::to_value(&synthetic.check_type)
+        .ok()
+        .and_then(|v| v.as_str().map(str::to_owned))
+        .unwrap_or_default();
+    notify(AckResponse {
+        run_complete: true,
+        run_status: Some(status.to_string()),
+        job_count: done.job_count,
+        org_id: done.org_id.clone(),
+        job_id: String::new(),
+        run_id: done.run_id.clone(),
+        synthetics_id: done.synthetics_id.clone(),
+        synthetics_name: synthetic.name,
+        synthetic_type,
+        target: synthetic.target,
+        destinations: synthetic.destinations,
+        location: String::new(),
+        pool: String::new(),
+        trigger_type: "scheduled".to_string(),
+        alert,
+        status_reason: status_reason.map(str::to_owned),
+        consecutive_failures,
+        failing_locations: outcomes.failing,
+        passing_locations: outcomes.passing,
+        usage_events: Vec::new(),
+        failing_environments,
+        suppressed_by: None,
+    })
+    .await;
+}
+
 /// Delay before the one extra run that fills a Not measured gap.
 const GAP_RETRY_DELAY_US: i64 = 60_000_000;
 
@@ -1870,7 +1966,12 @@ pub async fn ack(req: AckRequest, token_org: &str) -> anyhow::Result<AckResponse
         .await
         .map_err(|e| anyhow::anyhow!(e.to_string()))?;
     let run_complete = run_completion.is_some();
+    let merged = run_completion
+        .as_ref()
+        .filter(|d| d.had_confirm)
+        .map(|d| merged_alert_input(d.run_class, d.confirmed_flaky));
     let (run_status, job_count) = match &run_completion {
+        Some(done) if merged.is_some() => (merged.map(|m| m.0.to_string()), done.job_count),
         Some(done) => {
             let status =
                 match config::meta::synthetics::SyntheticStatus::from_db(done.run_result) {
@@ -1935,26 +2036,16 @@ pub async fn ack(req: AckRequest, token_org: &str) -> anyhow::Result<AckResponse
     // location for the same outage, and would advance the failure streak by the
     // fan-out factor rather than by one.
     let (alert, consecutive_failures, suppressed_by) = if run_complete {
-        // A confirm job's ack completes the run, but the run's answer is the merged vote: its own
-        // error source must not silence the failure it confirms.
-        let confirmed_flaky = run_completion.as_ref().is_some_and(|d| d.confirmed_flaky);
-        let completed_by_confirm = is_confirm_job(&check);
+        // A run that held a confirm job answers with its merged vote, so neither its worst raw
+        // job nor the last ack's error source decides the alert.
         resolve_alert(
             conn,
             &check.synthetics_id,
             RunOutcome {
                 status: run_status.as_deref(),
                 attempts: req.attempts,
-                status_reason: if confirmed_flaky {
-                    Some(crate::alerting::REASON_FLAKY)
-                } else {
-                    req.status_reason.as_deref()
-                },
-                error_source: if completed_by_confirm {
-                    ""
-                } else {
-                    &req.error_source
-                },
+                status_reason: merged.map_or(req.status_reason.as_deref(), |m| m.1),
+                error_source: merged.map_or(req.error_source.as_str(), |m| m.2),
             },
             &synthetic,
             now_us,
@@ -2002,7 +2093,7 @@ pub async fn ack(req: AckRequest, token_org: &str) -> anyhow::Result<AckResponse
         trigger_type: req.trigger_type,
         alert,
         // Echoed straight back so the notification can name the condition.
-        status_reason: req.status_reason.clone(),
+        status_reason: merged.map_or(req.status_reason.clone(), |m| m.1.map(str::to_owned)),
         consecutive_failures,
         failing_locations,
         passing_locations,
@@ -2361,6 +2452,24 @@ mod tests {
     }
 
     // Step billing. Every guard is a pure function, so these need no database.
+    #[test]
+    fn a_confirm_run_alerts_on_its_merged_vote() {
+        use config::meta::synthetics::RunClass as C;
+        let classify = |class, flaky| {
+            let (status, reason, source) = super::merged_alert_input(class, flaky);
+            crate::alerting::classify(Some(status), 1, source, reason)
+        };
+        use crate::alerting::RunClass as A;
+        assert_eq!(classify(C::Up, true), A::Flaky);
+        assert_eq!(classify(C::Up, false), A::Healthy);
+        // A not measured location beside a confirmed pass must not page the owner.
+        assert_eq!(classify(C::Up, false), A::Healthy);
+        assert_eq!(classify(C::Down, false), A::Failing);
+        assert_eq!(classify(C::Partial, false), A::Failing);
+        assert_eq!(classify(C::NotMeasured, false), A::NotMeasured);
+        assert_eq!(classify(C::Misconfigured, false), A::Failing);
+    }
+
     #[test]
     fn only_a_scheduled_target_failure_gets_a_confirm_job() {
         let want = |status, source: &str, reason, probe, run, on, confirm| {

src/synthetics/src/reaper/mod.rs

@@ -95,6 +95,7 @@ pub async fn run(stop: crate::StopToken) {
                         "[synthetics reaper] finalized a run whose finalization never committed"
                     );
                     crate::job_api::after_run_finalized(run).await;
+                    crate::job_api::alert_for_swept_run(run).await;
                 }
             }
             Err(e) => tracing::error!("[synthetics reaper] finalize_stuck_runs: {e}"),
@@ -163,7 +164,10 @@ async fn handle_dead_letter(
     // Run accounting first: the ingest-token lookup below can return early, and an unfinalized
     // run never completes. 4 = Error on the SyntheticStatus scale, as the dispatcher reports.
     match synthetics_runs::finalize_run(db, &row.run_id, 4, now_us).await {
-        Ok(Some(done)) => crate::job_api::after_run_finalized(&done).await,
+        Ok(Some(done)) => {
+            crate::job_api::after_run_finalized(&done).await;
+            crate::job_api::alert_for_swept_run(&done).await;
+        }
         Ok(None) => {}
         Err(e) => {
             tracing::error!(