fix: stand the WAC park down for a cancel that beat it to the row (#10990)

* fix: stand the WAC park down for a cancel that beat it to the row

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GHfNFFJh3ozZYgyyoaEepu

* refactor: share the cancel result payload with canceled_job_to_result

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GHfNFFJh3ozZYgyyoaEepu

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-5dc28b4609bff15ca4895e726c27cfd7ac33ac2616d9f68ab4cb62da105cae57.json

@@ -1,25 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "WITH prev AS (\n            SELECT started_at FROM v2_job_queue WHERE id = $1 AND workspace_id = $2\n         )\n         UPDATE v2_job_queue q\n         SET suspend = $3, suspend_until = now() + make_interval(secs => $4), started_at = null\n         FROM prev\n         WHERE q.id = $1 AND q.workspace_id = $2\n         RETURNING (extract(epoch FROM now() - prev.started_at) * 1000)::bigint",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "int8",
-        "type_info": "Int8"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Text",
-        "Int4",
-        "Float8"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "5dc28b4609bff15ca4895e726c27cfd7ac33ac2616d9f68ab4cb62da105cae57"
-}

backend/.sqlx/query-e749663a4b9248a9d120f77e86fd3413ad3c0fbe8f3dca9aea11cb49fd4a28ef.json

@@ -0,0 +1,35 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT canceled_by, canceled_reason,\n                (extract(epoch FROM now() - started_at) * 1000)::bigint AS segment_ms\n         FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "canceled_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "canceled_reason",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 2,
+        "name": "segment_ms",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      true,
+      true,
+      null
+    ]
+  },
+  "hash": "e749663a4b9248a9d120f77e86fd3413ad3c0fbe8f3dca9aea11cb49fd4a28ef"
+}

backend/.sqlx/query-f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_queue\n         SET suspend = $3, suspend_until = now() + make_interval(secs => $4), started_at = null\n         WHERE id = $1 AND workspace_id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text",
+        "Int4",
+        "Float8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed"
+}

backend/tests/wac_suspend_started_at.rs

@@ -1,9 +1,10 @@
-//! Guards the two things `suspend_wac_parent` promises: the `started_at` invariant
-//! documented on it, and the segment length it hands back for metering.
+//! Guards what `suspend_wac_parent` promises: the `started_at` invariant documented on
+//! it, the segment length it hands back for metering, and that it stands down for a
+//! cancel already on the row.
 
 use sqlx::{Pool, Postgres};
 use uuid::Uuid;
-use windmill_worker::wac_executor::suspend_wac_parent;
+use windmill_worker::wac_executor::{suspend_wac_parent, WacPark};
 
 #[sqlx::test]
 async fn wac_suspend_clears_started_at(db: Pool<Postgres>) -> anyhow::Result<()> {
@@ -17,7 +18,11 @@ async fn wac_suspend_clears_started_at(db: Pool<Postgres>) -> anyhow::Result<()>
     .await?;
 
     let mut tx = db.begin().await?;
-    let segment_ms = suspend_wac_parent(&mut tx, &job_id, "test-workspace", 1, 3600.0).await?;
+    let WacPark::Parked(segment_ms) =
+        suspend_wac_parent(&mut tx, &job_id, "test-workspace", 1, 3600.0).await?
+    else {
+        panic!("an uncancelled parent must park");
+    };
     tx.commit().await?;
 
     // The segment is what gets billed, so it must be the run that just ended, measured
@@ -53,3 +58,54 @@ async fn wac_suspend_clears_started_at(db: Pool<Postgres>) -> anyhow::Result<()>
 
     Ok(())
 }
+
+/// A soft cancel sets `canceled_by` and `suspend = 0` and leaves acting on it to the next
+/// pull. Parking over that holds the row until `suspend_until` — a whole day on a
+/// `sleep(86400)` — so the park has to stand down and let the job complete instead.
+#[sqlx::test]
+async fn wac_suspend_stands_down_for_a_cancel(db: Pool<Postgres>) -> anyhow::Result<()> {
+    let job_id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO v2_job_queue \
+           (id, workspace_id, scheduled_for, running, started_at, suspend, canceled_by, canceled_reason) \
+         VALUES ($1, 'test-workspace', now(), true, now() - interval '30 seconds', 0, 'alice', 'no longer needed')",
+    )
+    .bind(job_id)
+    .execute(&db)
+    .await?;
+
+    let mut tx = db.begin().await?;
+    let parked = suspend_wac_parent(&mut tx, &job_id, "test-workspace", 1, 86400.0).await?;
+    tx.commit().await?;
+
+    match &parked {
+        WacPark::Cancelled(cancel) => {
+            assert_eq!(cancel.username.as_deref(), Some("alice"));
+            assert_eq!(cancel.reason.as_deref(), Some("no longer needed"));
+        }
+        other => panic!("a cancelled parent must not park, got {other:?}"),
+    }
+
+    let (suspend, suspend_until, started_at): (
+        i32,
+        Option<chrono::DateTime<chrono::Utc>>,
+        Option<chrono::DateTime<chrono::Utc>>,
+    ) = sqlx::query_as(
+        "SELECT suspend, suspend_until, started_at FROM v2_job_queue WHERE id = $1",
+    )
+    .bind(job_id)
+    .fetch_one(&db)
+    .await?;
+
+    assert_eq!(suspend, 0, "the cancel's suspend = 0 must survive");
+    assert_eq!(
+        suspend_until, None,
+        "a suspend_until would hold the row back for the whole park window"
+    );
+    assert!(
+        started_at.is_some(),
+        "the segment ran, so its start must stay for the completion's duration"
+    );
+
+    Ok(())
+}

backend/windmill-queue/src/jobs.rs

@@ -7308,15 +7308,19 @@ async fn check_workspace_queue_cap<'c>(
 //     Ok(())
 // }
 
-pub fn canceled_job_to_result(job: &MiniPulledJob) -> serde_json::Value {
-    let reason = job
-        .canceled_reason
-        .as_deref()
-        .unwrap_or_else(|| "no reason given");
-    let canceler = job.canceled_by.as_deref().unwrap_or_else(|| "unknown");
+/// The result payload a job cancelled anywhere carries. Callers that hold the cancel
+/// outside a `MiniPulledJob` — a row read after the pull, say — go through this rather
+/// than rebuilding the shape.
+pub fn canceled_result(reason: Option<&str>, canceler: Option<&str>) -> serde_json::Value {
+    let reason = reason.unwrap_or("no reason given");
+    let canceler = canceler.unwrap_or("unknown");
     serde_json::json!({"message": format!("Job canceled: {reason} by {canceler}"), "name": "Canceled", "reason": reason, "canceler": canceler})
 }
 
+pub fn canceled_job_to_result(job: &MiniPulledJob) -> serde_json::Value {
+    canceled_result(job.canceled_reason.as_deref(), job.canceled_by.as_deref())
+}
+
 /// Helper function to create a restarted module for branch/iteration restart
 fn create_restarted_module(
     module: &FlowStatusModule,

backend/windmill-worker/src/bun_executor.rs

@@ -2572,7 +2572,8 @@ try {{
 
     // WAC v2 post-execution: parse output and handle dispatch/suspend
     if is_wac_v2 {
-        return handle_wac_v2_output(result, job, conn, modules, new_args.as_ref()).await;
+        return handle_wac_v2_output(result, job, conn, canceled_by, modules, new_args.as_ref())
+            .await;
     }
 
     Ok(result)
@@ -2602,11 +2603,13 @@ pub async fn handle_wac_v2_output(
     result: Box<RawValue>,
     job: &MiniPulledJob,
     conn: &Connection,
+    canceled_by: &mut Option<CanceledBy>,
     modules: &Option<std::collections::HashMap<String, windmill_common::scripts::ScriptModule>>,
     preprocessed_args: Option<&HashMap<String, Box<RawValue>>>,
 ) -> error::Result<Box<RawValue>> {
     use crate::wac_executor::{
-        load_checkpoint, parse_wac_output, update_checkpoint_for_dispatch, WacOutput,
+        load_checkpoint, parse_wac_output, update_checkpoint_for_dispatch,
+        wac_cancelled_mid_segment, WacOutput, WacPark,
     };
     use serde_json::Value;
     use windmill_common::get_latest_flow_version_info_for_path;
@@ -2874,14 +2877,22 @@ pub async fn handle_wac_v2_output(
 
                 // Suspend parent before children become visible, so a child that
                 // completes immediately finds a parked parent to decrement.
-                segment_ms = crate::wac_executor::suspend_wac_parent(
+                match crate::wac_executor::suspend_wac_parent(
                     &mut tx,
                     &job.id,
                     &job.workspace_id,
                     num_steps as i32,
                     14.0 * 24.0 * 3600.0,
                 )
-                .await?;
+                .await?
+                {
+                    WacPark::Parked(ms) => segment_ms = ms,
+                    // Returning here drops `tx`, unwriting the checkpoint and the timeline
+                    // entries, so no child is ever pushed against a parent that never parked.
+                    WacPark::Cancelled(cancel) => {
+                        return Err(wac_cancelled_mid_segment(cancel, canceled_by))
+                    }
+                }
 
                 tx.commit().await?;
             }
@@ -3362,14 +3373,20 @@ pub async fn handle_wac_v2_output(
             }
 
             // Suspend parent with suspend=1 (waiting for 1 approval event)
-            let segment_ms = crate::wac_executor::suspend_wac_parent(
+            let segment_ms = match crate::wac_executor::suspend_wac_parent(
                 &mut tx,
                 &job.id,
                 &job.workspace_id,
                 1,
                 timeout_secs,
             )
-            .await?;
+            .await?
+            {
+                WacPark::Parked(ms) => ms,
+                WacPark::Cancelled(cancel) => {
+                    return Err(wac_cancelled_mid_segment(cancel, canceled_by))
+                }
+            };
 
             tx.commit().await?;
             crate::wac_executor::end_wac_segment(conn, job, segment_ms);
@@ -3458,14 +3475,20 @@ pub async fn handle_wac_v2_output(
 
             // Use suspend=1 (not 0) so the suspended pull query only picks it up
             // when `suspend_until <= now()`, not via `suspend <= 0`.
-            let segment_ms = crate::wac_executor::suspend_wac_parent(
+            let segment_ms = match crate::wac_executor::suspend_wac_parent(
                 &mut tx,
                 &job.id,
                 &job.workspace_id,
                 1,
                 sleep_secs,
             )
-            .await?;
+            .await?
+            {
+                WacPark::Parked(ms) => ms,
+                WacPark::Cancelled(cancel) => {
+                    return Err(wac_cancelled_mid_segment(cancel, canceled_by))
+                }
+            };
 
             tx.commit().await?;
             crate::wac_executor::end_wac_segment(conn, job, segment_ms);

backend/windmill-worker/src/python_executor.rs

@@ -1223,6 +1223,7 @@ mount {{
             result,
             job,
             conn,
+            canceled_by,
             modules,
             new_args.as_ref(),
         ))

backend/windmill-worker/src/wac_executor.rs

@@ -7,6 +7,7 @@ use uuid::Uuid;
 use windmill_common::error::{self, Error};
 use windmill_common::scripts::ScriptLang;
 use windmill_common::DB;
+use windmill_queue::CanceledBy;
 
 // Checkpoint model + persistence primitives live in windmill-common so the
 // API server can use them without pulling in the full worker crate. Re-export
@@ -86,6 +87,16 @@ fn default_dispatch_type() -> String {
     "inline".to_string()
 }
 
+/// What `suspend_wac_parent` did with the parent's queue row.
+#[derive(Debug)]
+pub enum WacPark {
+    /// Parked. Carries the segment that just ended, in milliseconds, for `end_wac_segment`.
+    Parked(Option<i64>),
+    /// A cancel reached the row while this segment was running, so the park was skipped.
+    /// Carries who cancelled, for the completion that must happen instead.
+    Cancelled(CanceledBy),
+}
+
 /// Park a WAC v2 parent in the queue until `suspend` reaches 0 or `suspend_secs`
 /// elapses, whichever comes first. `running` stays true so the normal pull query
 /// skips the row; only the suspended pull query takes it back. The `id`/`workspace_id`
@@ -97,45 +108,76 @@ fn default_dispatch_type() -> String {
 /// completes a job without a worker-measured duration — a cancel, the child-failure
 /// handler — falls back to `now() - started_at`. Left pointing at the first segment,
 /// that fallback reports the whole sleep or approval wait as execution time.
-///
-/// Returns the segment that just ended, in milliseconds, for the caller to hand to
-/// `end_wac_segment`.
 pub async fn suspend_wac_parent(
     tx: &mut Transaction<'_, Postgres>,
     job_id: &Uuid,
     w_id: &str,
     suspend: i32,
     suspend_secs: f64,
-) -> error::Result<Option<i64>> {
-    // `prev` holds the pre-update row: RETURNING sees the new one, where `started_at`
-    // has already been cleared.
-    let parked = sqlx::query_scalar!(
-        "WITH prev AS (
-            SELECT started_at FROM v2_job_queue WHERE id = $1 AND workspace_id = $2
-         )
-         UPDATE v2_job_queue q
+) -> error::Result<WacPark> {
+    // `FOR UPDATE` orders this against a concurrent soft cancel, which writes `suspend = 0`
+    // and leaves acting on `canceled_by` to the next pull. Parking on top of that keeps the
+    // row unpullable until `suspend_until` — up to the full `sleep()` — so a cancel already
+    // on the row has to stand the park down rather than be overwritten by it.
+    let prev = sqlx::query!(
+        "SELECT canceled_by, canceled_reason,
+                (extract(epoch FROM now() - started_at) * 1000)::bigint AS segment_ms
+         FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 FOR UPDATE",
+        job_id,
+        w_id,
+    )
+    .fetch_optional(&mut **tx)
+    .await
+    .map_err(|e| Error::internal_err(format!("Failed to read WAC parent job {job_id}: {e}")))?
+    // Silently parking nothing is unrecoverable on the dispatch arm: the children are
+    // pushed right after and decrement a `suspend` that was never set, so the parent
+    // sits out its whole suspend window instead of resuming.
+    .ok_or_else(|| {
+        Error::internal_err(format!(
+            "WAC parent job {job_id} not in the queue of workspace {w_id} to suspend"
+        ))
+    })?;
+
+    if let Some(username) = prev.canceled_by {
+        return Ok(WacPark::Cancelled(CanceledBy {
+            username: Some(username),
+            reason: prev.canceled_reason,
+        }));
+    }
+
+    sqlx::query!(
+        "UPDATE v2_job_queue
          SET suspend = $3, suspend_until = now() + make_interval(secs => $4), started_at = null
-         FROM prev
-         WHERE q.id = $1 AND q.workspace_id = $2
-         RETURNING (extract(epoch FROM now() - prev.started_at) * 1000)::bigint",
+         WHERE id = $1 AND workspace_id = $2",
         job_id,
         w_id,
         suspend,
         suspend_secs,
     )
-    .fetch_optional(&mut **tx)
+    .execute(&mut **tx)
     .await
     .map_err(|e| Error::internal_err(format!("Failed to suspend WAC parent job {job_id}: {e}")))?;
 
-    // Silently parking nothing is unrecoverable on the dispatch arm: the children are
-    // pushed right after and decrement a `suspend` that was never set, so the parent
-    // sits out its whole suspend window instead of resuming.
-    match parked {
-        Some(segment_ms) => Ok(segment_ms),
-        None => Err(Error::internal_err(format!(
-            "WAC parent job {job_id} not in the queue of workspace {w_id} to suspend"
-        ))),
-    }
+    Ok(WacPark::Parked(prev.segment_ms))
+}
+
+/// Turn a cancel that landed mid-segment into the error the executor returns, so the job
+/// completes on this pass instead of parking. Setting the worker's `canceled_by` is what
+/// makes it land as `canceled` rather than `failure`: the row was cancelled after this
+/// worker pulled the job, so the in-memory copy still reads as uncancelled.
+///
+/// The completion charges the segment that just ended, so callers must not also hand it to
+/// `end_wac_segment`.
+pub(crate) fn wac_cancelled_mid_segment(
+    cancel: CanceledBy,
+    canceled_by: &mut Option<CanceledBy>,
+) -> Error {
+    let payload = windmill_common::worker::to_raw_value(&windmill_queue::canceled_result(
+        cancel.reason.as_deref(),
+        cancel.username.as_deref(),
+    ));
+    *canceled_by = Some(cancel);
+    Error::ExecutionRawError(payload)
 }
 
 /// Charge the execution segment a WAC parent just finished. Segments are metered as they