diff --git a/backend/.sqlx/query-5dc28b4609bff15ca4895e726c27cfd7ac33ac2616d9f68ab4cb62da105cae57.json b/backend/.sqlx/query-5dc28b4609bff15ca4895e726c27cfd7ac33ac2616d9f68ab4cb62da105cae57.json deleted file mode 100644 index a93f92c4e2..0000000000 --- a/backend/.sqlx/query-5dc28b4609bff15ca4895e726c27cfd7ac33ac2616d9f68ab4cb62da105cae57.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-e749663a4b9248a9d120f77e86fd3413ad3c0fbe8f3dca9aea11cb49fd4a28ef.json b/backend/.sqlx/query-e749663a4b9248a9d120f77e86fd3413ad3c0fbe8f3dca9aea11cb49fd4a28ef.json new file mode 100644 index 0000000000..328465fdf1 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed.json b/backend/.sqlx/query-f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed.json new file mode 100644 index 0000000000..5f86fb97d8 --- /dev/null +++ b/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" +} diff --git a/backend/tests/wac_suspend_started_at.rs b/backend/tests/wac_suspend_started_at.rs index f374a25220..5a83730d24 100644 --- a/backend/tests/wac_suspend_started_at.rs +++ b/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) -> anyhow::Result<()> { @@ -17,7 +18,11 @@ async fn wac_suspend_clears_started_at(db: Pool) -> 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) -> 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) -> 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>, + Option>, + ) = 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(()) +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index d7f847a73d..5bb9a1120f 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/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, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index ac1eebe723..7c112f5985 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/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, job: &MiniPulledJob, conn: &Connection, + canceled_by: &mut Option, modules: &Option>, preprocessed_args: Option<&HashMap>>, ) -> error::Result> { 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); diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 883b035a32..38fb4bca32 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -1223,6 +1223,7 @@ mount {{ result, job, conn, + canceled_by, modules, new_args.as_ref(), )) diff --git a/backend/windmill-worker/src/wac_executor.rs b/backend/windmill-worker/src/wac_executor.rs index 1402c7aed9..cecf0e316c 100644 --- a/backend/windmill-worker/src/wac_executor.rs +++ b/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), + /// 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> { - // `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 { + // `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, +) -> 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