[ee] feat: job cancel signals (oidc refusal, agent 5s ping, job.canceled) (#11592)

* feat: faster agent cancel detection, job.canceled event, worker_group span attribute Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix: report no unix signal in job.canceled where none is sent Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * refactor: agent workers ping every 5s instead of a 2s read-only cancel check Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix: emit job.canceled on windows when the process exits unsignaled Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * chore: pin ee ref with the queue delay gauge main requires Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-12 08:02:44 +00:00 · 2026-10-07 16:44:57 +00:00
parent f2fb15603b
commit afa9ced84c
7 changed files with 143 additions and 57 deletions
@@ -1,11 +1,11 @@
 {
  "db_name": "PostgreSQL",
-  "query": "SELECT EXISTS (SELECT 1 FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 AND running)",
+  "query": "SELECT canceled_by IS NOT NULL FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 AND running",
  "describe": {
    "columns": [
      {
        "ordinal": 0,
-        "name": "exists",
+        "name": "?column?",
        "type_info": "Bool"
      }
    ],
@@ -19,5 +19,5 @@
      null
    ]
  },
-  "hash": "ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e"
+  "hash": "58ea7e2c6b07ffc7ab7387b2ebb542cba3a5532c64a9594d2e0ccdf21dee90d0"
 }
@@ -1 +1 @@
-acb254ac509aff244849c09e50a8d39aae185138
+0b78e2c6fefcbad3c465bf75ebfd6ebd6912ad19
@@ -410,5 +410,22 @@ async fn test_oidc_claim_carries_the_presented_scopes(db: Pool<Postgres>) -> any
        )?;
        assert_eq!(claims["job_token_scopes"], expected, "{job}");
    }
+
+    // A soft-canceled job is still `running` until its worker kills it.
+    sqlx::query("UPDATE v2_job_queue SET canceled_by = 'test-user' WHERE id = $1")
+        .bind(Uuid::parse_str(OIDC_JOB)?)
+        .execute(&db)
+        .await?;
+    let resp = client
+        .post(format!("{base}/oidc/token/test-audience"))
+        .bearer_auth(job_token(&db, OIDC_JOB).await?)
+        .send()
+        .await?;
+    assert_eq!(
+        resp.status(),
+        StatusCode::FORBIDDEN,
+        "{}",
+        resp.text().await?
+    );
    Ok(())
 }
@@ -49,6 +49,7 @@ pub struct LogContext {

    // Worker / job span (windmill-worker/src/worker.rs)
    pub worker: Option<String>,
+    pub worker_group: Option<String>,
    pub hostname: Option<String>,
    pub tag: Option<String>,
    pub job_id: Option<String>,
@@ -929,8 +929,8 @@ pub async fn eval_fetch_timeout(
                    for line in logged.as_deref().unwrap_or(&log).lines() {
                        tracing::info!(
                            target: "windmill:job_log",
-                            job_id = ?job_id,
-                            workspace_id = ?w_id_for_tracing,
+                            job_id = %job_id,
+                            workspace_id = %w_id_for_tracing,
                            "{line}"
                        );
                        if *OTEL_JOB_LOGS {
@@ -210,6 +210,53 @@ pub async fn handle_child(
        }
    }

+    impl KillReason {
+        /// Reports a kill once the process is gone: a span event on the enclosing span and,
+        /// through the logs bridge, a log record carrying the job's log context.
+        fn emit_canceled_event(
+            &self,
+            job_id: &Uuid,
+            timeout_reason: &str,
+            signal: &'static str,
+            detected_at: chrono::DateTime<chrono::Utc>,
+        ) {
+            // `force` is only known for a soft cancel: a job completed under the worker
+            // may have been force canceled or completed by anything else server-side.
+            let (kill_reason, canceled_by, canceled_reason, force) = match self {
+                KillReason::TooManyLogs => return,
+                KillReason::Timeout { .. } => {
+                    ("timeout", Some("timeout"), Some(timeout_reason), None)
+                }
+                KillReason::Cancelled(by) => (
+                    "cancelled",
+                    by.as_ref().and_then(|x| x.username.as_deref()),
+                    by.as_ref().and_then(|x| x.reason.as_deref()),
+                    Some(false),
+                ),
+                KillReason::AlreadyCompleted => ("already_completed", None, None, None),
+            };
+            let exited_at = chrono::Utc::now();
+            let root_job = windmill_common::log_context::current_log_context()
+                .and_then(|c| c.root_job.clone());
+            let rfc3339 = |t: chrono::DateTime<chrono::Utc>| {
+                t.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
+            };
+            tracing::info!(
+                job_id = %job_id,
+                root_job,
+                kill_reason,
+                canceled_by,
+                canceled_reason,
+                force,
+                kill_signal = signal,
+                cancel_detected_at = %rfc3339(detected_at),
+                process_exited_at = %rfc3339(exited_at),
+                kill_duration_ms = (exited_at - detected_at).num_milliseconds(),
+                "job.canceled"
+            );
+        }
+    }
+
    let (timeout_duration, timeout_warn_msg, timeout_source) =
        resolve_job_timeout(&conn, w_id, job_id, custom_timeout).await;
    if let Some(msg) = timeout_warn_msg {
@@ -232,6 +279,7 @@ pub async fn handle_child(
                UpdateJobPollingExit::AlreadyCompleted => KillReason::AlreadyCompleted,
            },
        };
+        let detected_at = chrono::Utc::now();
        tx.send(()).expect("rx should never be dropped");
        drop(tx);

@@ -268,59 +316,78 @@ pub async fn handle_child(
            }
        };

-        #[allow(unused_variables)]
-        if let Some(id) = child.id() {
-            if *MAX_WAIT_FOR_SIGINT > 0 {
-                #[cfg(any(target_os = "linux", target_os = "macos"))]
-                signal::kill(Pid::from_raw(id as i32), Signal::SIGINT).unwrap();
+        // Resolves once the process is reaped, to the last signal it was sent, or to `None`
+        // when the kill could not be confirmed. SIGINT and SIGTERM are only sent where the
+        // sends below are compiled in: elsewhere an exit during their wait is unsignaled.
+        let unix_signal = |s: &'static str| {
+            if cfg!(any(target_os = "linux", target_os = "macos")) { s } else { "none" }
+        };
+        let kill = async {
+            #[allow(unused_variables)]
+            if let Some(id) = child.id() {
+                if *MAX_WAIT_FOR_SIGINT > 0 {
+                    #[cfg(any(target_os = "linux", target_os = "macos"))]
+                    signal::kill(Pid::from_raw(id as i32), Signal::SIGINT).unwrap();

-                for _ in 0..*MAX_WAIT_FOR_SIGINT {
+                    for _ in 0..*MAX_WAIT_FOR_SIGINT {
+                        if child.try_wait().is_ok_and(|x| x.is_some()) {
+                            break;
+                        }
+                        sleep(Duration::from_secs(1)).await;
+                    }
                    if child.try_wait().is_ok_and(|x| x.is_some()) {
-                        break;
+                        set_reason.await;
+                        return Ok(Some(unix_signal("SIGINT")));
                    }
-                    sleep(Duration::from_secs(1)).await;
                }
-                if child.try_wait().is_ok_and(|x| x.is_some()) {
-                    set_reason.await;
-                    return Ok(Err(kill_reason));
+                if sigterm {
+                    #[cfg(any(target_os = "linux", target_os = "macos"))]
+                    signal::kill(Pid::from_raw(id as i32), Signal::SIGTERM).unwrap();
+
+                    for _ in 0..*MAX_WAIT_FOR_SIGTERM {
+                        if child.try_wait().is_ok_and(|x| x.is_some()) {
+                            break;
+                        }
+                        sleep(Duration::from_secs(1)).await;
+                    }
+                    if child.try_wait().is_ok_and(|x| x.is_some()) {
+                        set_reason.await;
+                        return Ok(Some(unix_signal("SIGTERM")));
+                    }
                }
            }
-            if sigterm {
-                #[cfg(any(target_os = "linux", target_os = "macos"))]
-                signal::kill(Pid::from_raw(id as i32), Signal::SIGTERM).unwrap();
-
-                for _ in 0..*MAX_WAIT_FOR_SIGTERM {
-                    if child.try_wait().is_ok_and(|x| x.is_some()) {
-                        break;
+            #[cfg(windows)]
+            {
+                let pid_to_kill = child.id();
+                let killed = match kill_process_tree(pid_to_kill).await {
+                    Ok(_) => {
+                        tracing::debug!(
+                            "successfully killed process tree with PID: {:?}",
+                            pid_to_kill
+                        );
+                        true
                    }
-                    sleep(Duration::from_secs(1)).await;
-                }
-                if child.try_wait().is_ok_and(|x| x.is_some()) {
-                    set_reason.await;
-                    return Ok(Err(kill_reason));
-                }
+                    Err(e) => {
+                        tracing::error!("failed to kill process tree: {:?}", e);
+                        false
+                    }
+                };
+                set_reason.await;
+                return Ok(killed.then_some("taskkill"));
            }
-        }
-        #[cfg(windows)]
-        {
-            let pid_to_kill = child.id();
-            match kill_process_tree(pid_to_kill).await {
-                Ok(_) => tracing::debug!(
-                    "successfully killed process tree with PID: {:?}",
-                    pid_to_kill
-                ),
-                Err(e) => tracing::error!("failed to kill process tree: {:?}", e),
-            };
-            set_reason.await;
-            return Ok(Err(kill_reason));
-        }

-        #[cfg(unix)]
-        {
-            /* send SIGKILL and reap child process */
-            let (_, kill) = future::join(set_reason, Box::into_pin(child.kill())).await;
-            kill.map(|()| Err(kill_reason))
+            #[cfg(unix)]
+            {
+                /* send SIGKILL and reap child process */
+                let (_, kill) = future::join(set_reason, Box::into_pin(child.kill())).await;
+                kill.map(|()| Some("SIGKILL"))
+            }
+        };
+        let signal: io::Result<Option<&'static str>> = kill.await;
+        if let Some(signal) = signal? {
+            kill_reason.emit_canceled_event(&job_id, &timeout_reason, signal, detected_at);
        }
+        Ok(Err(kill_reason))
    };

    let mut stream_result = Vec::new();
@@ -957,9 +1024,9 @@ where

    let conn = conn.clone();
    // No tick at t=0: a job that finishes within the first interval issues none of the
-    // ping/metric statements below. Cancels are picked up by the first job ping (500 ms in,
-    // 2 s on agent workers). Memory is still sampled at start, without SQL: it is the only
-    // `mem_peak` reading a short job gets, and the completed job row reports it.
+    // ping/metric statements below. Cancels are picked up by the next job ping. Memory is still
+    // sampled at start, without SQL: it is the only `mem_peak` reading a short job gets,
+    // and the completed job row reports it.
    let mut interval = interval_at(Instant::now() + update_job_interval, update_job_interval);
    interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
    *mem_peak = (*mem_peak).max(get_mem.next().await.unwrap_or(0));
@@ -1023,11 +1090,10 @@ where
                            }
                        }
                    }
-                    if matches!(conn, Connection::Http(_)) {
-                        if i % 4 != 0 {
-                            // only ping every 4th time (2s) on http agent mode
-                            continue;
-                        }
+                    // An agent worker pays an HTTP round trip per ping, so it pings every 2 s for
+                    // the first 10 s, then every 5 s.
+                    if matches!(conn, Connection::Http(_)) && i % if i < 20 { 4 } else { 10 } != 0 {
+                        continue;
                    }
                    let ping_job_status = ping_job_status(&conn, &job_id, Some(*mem_peak), if current_mem > 0 { Some(current_mem) } else { None }).await.unwrap_or_else(|e| {
                        tracing::error!("Unable to ping job status for job {job_id}. Error was: {:?}", e);
@@ -2005,6 +2005,7 @@ pub fn create_span_with_name(
        root_job = field::Empty,
        workspace_id = %arc_job.workspace_id,
        worker = %worker_name,
+        worker_group = %*WORKER_GROUP,
        hostname = field::Empty,
        tag = %arc_job.tag,
        language = field::Empty,
@@ -2156,6 +2157,7 @@ pub fn log_context_for_job(
        job_id: Some(arc_job.id.to_string()),
        workspace_id: Some(arc_job.workspace_id.clone()),
        worker: Some(worker_name.to_string()),
+        worker_group: Some(WORKER_GROUP.clone()),
        tag: Some(arc_job.tag.clone()),
        job_kind: Some(arc_job.kind.as_str().to_string()),
        created_by: Some(arc_job.created_by.clone()),