diff --git a/backend/.sqlx/query-07982120afdeefd619777e9600cee9e0520babf241a2fe018ae550b3d13d33a9.json b/backend/.sqlx/query-07982120afdeefd619777e9600cee9e0520babf241a2fe018ae550b3d13d33a9.json new file mode 100644 index 0000000000..e4cf823ad1 --- /dev/null +++ b/backend/.sqlx/query-07982120afdeefd619777e9600cee9e0520babf241a2fe018ae550b3d13d33a9.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COUNT(*) FROM v2_job WHERE id = ANY($1) AND created_by != 'anonymous'", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [ + null + ] + }, + "hash": "07982120afdeefd619777e9600cee9e0520babf241a2fe018ae550b3d13d33a9" +} diff --git a/backend/.sqlx/query-115b35455f0e151386c85f0f239cf6b544ba58040742f46f31cea24389900142.json b/backend/.sqlx/query-115b35455f0e151386c85f0f239cf6b544ba58040742f46f31cea24389900142.json new file mode 100644 index 0000000000..a78c73c951 --- /dev/null +++ b/backend/.sqlx/query-115b35455f0e151386c85f0f239cf6b544ba58040742f46f31cea24389900142.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT \n flow_status->>'step' = '0' \n AND (\n jsonb_array_length(flow_status->'modules') = 0 \n OR flow_status->'modules'->0->>'type' = 'WaitingForPriorSteps' \n OR (\n flow_status->'modules'->0->>'type' = 'Failure' \n AND flow_status->'modules'->0->>'job' = $1\n )\n )\n FROM v2_job_status WHERE id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "115b35455f0e151386c85f0f239cf6b544ba58040742f46f31cea24389900142" +} diff --git a/backend/.sqlx/query-30c23ac8d77cea1d1fb8966da7b0fbdb85df82b240dc41b963e47a26ca324779.json b/backend/.sqlx/query-30c23ac8d77cea1d1fb8966da7b0fbdb85df82b240dc41b963e47a26ca324779.json new file mode 100644 index 0000000000..bba0f46222 --- /dev/null +++ b/backend/.sqlx/query-30c23ac8d77cea1d1fb8966da7b0fbdb85df82b240dc41b963e47a26ca324779.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id as \"id!\" FROM v2_as_queue WHERE workspace_id = $1 AND script_path = $2 AND canceled = false", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "30c23ac8d77cea1d1fb8966da7b0fbdb85df82b240dc41b963e47a26ca324779" +} diff --git a/backend/.sqlx/query-75bd7b3be39b52ed468f9e1ae748f3aac4c5c373819a9a1a0afcf9c6625eae17.json b/backend/.sqlx/query-75bd7b3be39b52ed468f9e1ae748f3aac4c5c373819a9a1a0afcf9c6625eae17.json new file mode 100644 index 0000000000..a25bf6a993 --- /dev/null +++ b/backend/.sqlx/query-75bd7b3be39b52ed468f9e1ae748f3aac4c5c373819a9a1a0afcf9c6625eae17.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT NULL AS \"lock!: ()\" FROM v2_job_queue WHERE id = $1 FOR UPDATE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "lock!: ()", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "75bd7b3be39b52ed468f9e1ae748f3aac4c5c373819a9a1a0afcf9c6625eae17" +} diff --git a/backend/.sqlx/query-cca9425bfac22749f7ca4904227de7fc3c416120d33df76fbb16f1252e81a65c.json b/backend/.sqlx/query-cca9425bfac22749f7ca4904227de7fc3c416120d33df76fbb16f1252e81a65c.json new file mode 100644 index 0000000000..c21671d701 --- /dev/null +++ b/backend/.sqlx/query-cca9425bfac22749f7ca4904227de7fc3c416120d33df76fbb16f1252e81a65c.json @@ -0,0 +1,31 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_queue q SET\n canceled_by = $1,\n canceled_reason = $2,\n scheduled_for = NOW(),\n suspend = 0\n FROM v2_job j\n WHERE q.id = ANY($3) AND j.id = q.id AND q.workspace_id = $4\n AND canceled_by IS NULL\n AND (j.trigger IS DISTINCT FROM 'schedule' OR running = true)\n RETURNING q.id, (q.running = false AND parent_job IS NULL AND trigger IS NULL) AS trivial", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "trivial", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Varchar", + "Text", + "UuidArray", + "Text" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "cca9425bfac22749f7ca4904227de7fc3c416120d33df76fbb16f1252e81a65c" +} diff --git a/backend/.sqlx/query-d3dac23643070890e85c7668bbde3e75f1409bd6effc4da69b78444ddf21992b.json b/backend/.sqlx/query-d3dac23643070890e85c7668bbde3e75f1409bd6effc4da69b78444ddf21992b.json new file mode 100644 index 0000000000..e5d146c927 --- /dev/null +++ b/backend/.sqlx/query-d3dac23643070890e85c7668bbde3e75f1409bd6effc4da69b78444ddf21992b.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT flow_innermost_root_job AS \"root!\" FROM v2_job\n WHERE id = ANY($1) AND flow_innermost_root_job IS NOT NULL", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "root!", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [ + true + ] + }, + "hash": "d3dac23643070890e85c7668bbde3e75f1409bd6effc4da69b78444ddf21992b" +} diff --git a/backend/.sqlx/query-db2c20865d6fe6bd6752ec62a03267aa23c4e035cb0de8b25bada7ebf1c565fe.json b/backend/.sqlx/query-db2c20865d6fe6bd6752ec62a03267aa23c4e035cb0de8b25bada7ebf1c565fe.json new file mode 100644 index 0000000000..31334dc1c2 --- /dev/null +++ b/backend/.sqlx/query-db2c20865d6fe6bd6752ec62a03267aa23c4e035cb0de8b25bada7ebf1c565fe.json @@ -0,0 +1,40 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH queued AS (\n DELETE FROM v2_job_queue\n WHERE id = ANY($1::UUID[])\n RETURNING\n id, workspace_id, started_at, worker, canceled_by, canceled_reason, extras\n ), queued_and_runtime AS (\n SELECT queued.*, GREATEST($7, memory_peak) AS memory_peak, flow_status,\n workflow_as_code_status\n FROM queued\n LEFT JOIN v2_job_runtime USING (id)\n LEFT JOIN v2_job_status USING (id)\n ) INSERT INTO v2_job_completed (\n id, workspace_id, started_at, worker, memory_peak, flow_status, workflow_as_code_status,\n result, result_columns, canceled_by, canceled_reason, extras,\n duration_ms,\n status\n ) SELECT\n id, workspace_id, started_at, worker, memory_peak, flow_status, workflow_as_code_status,\n $5, $6, canceled_by, canceled_reason, extras,\n COALESCE($2::BIGINT, CASE\n WHEN started_at IS NULL THEN 0\n ELSE (EXTRACT('epoch' FROM NOW()) - EXTRACT('epoch' FROM started_at)) * 1000\n END) AS duration_ms,\n CASE\n WHEN canceled_by IS NOT NULL THEN 'canceled'::job_status\n WHEN $3::BOOLEAN THEN 'skipped'::job_status\n WHEN $4::BOOLEAN THEN 'success'::job_status\n ELSE 'failure'::job_status\n END AS status\n FROM queued_and_runtime\n RETURNING id, status = 'canceled' AS \"canceled!\", duration_ms", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "canceled!", + "type_info": "Bool" + }, + { + "ordinal": 2, + "name": "duration_ms", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "UuidArray", + "Int8", + "Bool", + "Bool", + "Jsonb", + "TextArray", + "Int4" + ] + }, + "nullable": [ + false, + null, + false + ] + }, + "hash": "db2c20865d6fe6bd6752ec62a03267aa23c4e035cb0de8b25bada7ebf1c565fe" +} diff --git a/backend/.sqlx/query-db55295a2010e137f8637ac2bd472e171f14984852ac6ed529043f1466bf2481.json b/backend/.sqlx/query-db55295a2010e137f8637ac2bd472e171f14984852ac6ed529043f1466bf2481.json new file mode 100644 index 0000000000..6732914662 --- /dev/null +++ b/backend/.sqlx/query-db55295a2010e137f8637ac2bd472e171f14984852ac6ed529043f1466bf2481.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_status SET flow_status = jsonb_set(\n jsonb_set(\n COALESCE(flow_status, '{}'::jsonb),\n array[$1],\n COALESCE(flow_status->$1, '{}'::jsonb)\n ),\n array[$1, 'duration_ms'],\n to_jsonb($2::bigint)\n ) WHERE id = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Int8", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "db55295a2010e137f8637ac2bd472e171f14984852ac6ed529043f1466bf2481" +} diff --git a/backend/src/main.rs b/backend/src/main.rs index d9ae5c6e58..4f5ec3dddc 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -155,7 +155,6 @@ async fn cache_hub_scripts(file_path: Option) -> anyhow::Result<()> { &Uuid::nil(), &res.content, &mut 0, - &mut None, &job_dir, None, "global", @@ -174,7 +173,6 @@ async fn cache_hub_scripts(file_path: Option) -> anyhow::Result<()> { let envs = windmill_worker::get_common_bun_proc_envs(None).await; let _ = windmill_worker::install_bun_lockfile( &mut 0, - &mut None, &job_id, "admins", None, diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index a3f117721f..962b500dc5 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -66,7 +66,7 @@ use windmill_common::{ MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED, SERVICE_LOG_RETENTION_SECS, }; -use windmill_queue::cancel_job; +use windmill_queue::cancel; use windmill_worker::{ create_token_for_owner, handle_job_error, AuthedClient, SameWorkerPayload, SameWorkerSender, SendResult, BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, @@ -1698,7 +1698,6 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker &client, &job, 0, - None, error::Error::ExecutionErr(format!( "Job timed out after no ping from job since {} (ZOMBIE_JOB_TIMEOUT: {}, same_worker: {})", last_ping @@ -1836,24 +1835,12 @@ async fn cancel_zombie_flow_job( workspace_id: &str, message: String, ) -> Result<(), error::Error> { - let mut tx = db.begin().await?; tracing::error!( "zombie flow detected: {} in workspace {}. Cancelling it.", id, workspace_id ); - (tx, _) = cancel_job( - "monitor", - Some(message), - id, - workspace_id, - tx, - db, - true, - false, - ) - .await?; - tx.commit().await?; + let _ = cancel(db, &[id], workspace_id, "monitor", &message, true, true).await?; Ok(()) } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 90caf92518..bc97d3e067 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -87,8 +87,8 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED}; use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL}; use windmill_queue::{ - cancel_job, get_result_and_success_by_id_from_flow, job_is_complete, push, PushArgs, - PushArgsOwned, PushIsolationLevel, + cancel, get_result_and_success_by_id_from_flow, job_is_complete, push, PushArgs, PushArgsOwned, + PushIsolationLevel, }; #[cfg(feature = "prometheus")] @@ -363,8 +363,6 @@ async fn cancel_job_api( Path((w_id, id)): Path<(String, Uuid)>, Json(CancelJob { reason }): Json, ) -> error::Result { - let tx = db.begin().await?; - let audit_author: AuditAuthor = match opt_authed.as_ref() { Some(authed) => (authed).into(), None => AuditAuthor { @@ -374,17 +372,16 @@ async fn cancel_job_api( }, }; - let (mut tx, job_option) = tokio::time::timeout( + let canceled = tokio::time::timeout( std::time::Duration::from_secs(120), - windmill_queue::cancel_job( - &audit_author.username, - reason, - id, - &w_id, - tx, + cancel( &db, + &[id], + &w_id, + &audit_author.username, + &reason.unwrap_or_else(|| "unknown reason".into()), false, - opt_authed.is_none(), + opt_authed.is_some(), ), ) .await @@ -394,9 +391,9 @@ async fn cancel_job_api( )) })??; - if let Some(id) = job_option { + if let Some(id) = canceled.into_iter().next() { audit_log( - &mut *tx, + &db, &audit_author, "jobs.cancel", ActionKind::Delete, @@ -405,10 +402,8 @@ async fn cancel_job_api( None, ) .await?; - tx.commit().await?; Ok(id.to_string()) } else { - tx.commit().await?; if job_is_complete(&db, id, &w_id).await.unwrap_or(false) { return Ok(format!("queued job id {} is already completed", id)); } else { @@ -474,8 +469,6 @@ async fn force_cancel( Path((w_id, id)): Path<(String, Uuid)>, Json(CancelJob { reason }): Json, ) -> error::Result { - let tx = db.begin().await?; - let audit_author: AuditAuthor = match opt_authed.as_ref() { Some(authed) => (authed).into(), None => AuditAuthor { @@ -485,17 +478,16 @@ async fn force_cancel( }, }; - let (mut tx, job_option) = tokio::time::timeout( + let canceled = tokio::time::timeout( std::time::Duration::from_secs(120), - windmill_queue::cancel_job( - &audit_author.username, - reason, - id, - &w_id, - tx, + cancel( &db, + &[id], + &w_id, + &audit_author.username, + &reason.unwrap_or_else(|| "unknown reason".into()), true, - opt_authed.is_none(), + opt_authed.is_some(), ), ) .await @@ -505,9 +497,9 @@ async fn force_cancel( )) })??; - if let Some(id) = job_option { + if let Some(id) = canceled.into_iter().next() { audit_log( - &mut *tx, + &db, &audit_author, "jobs.force_cancel", ActionKind::Delete, @@ -516,10 +508,8 @@ async fn force_cancel( None, ) .await?; - tx.commit().await?; Ok(id.to_string()) } else { - tx.commit().await?; if job_is_complete(&db, id, &w_id).await.unwrap_or(false) { return Ok(format!("queued job id {} is already completed", id)); } else { @@ -1480,101 +1470,6 @@ async fn list_queue_jobs( Ok(Json(jobs)) } -async fn cancel_jobs( - jobs: Vec, - db: &DB, - username: &str, - w_id: &str, -) -> error::JsonResult> { - let mut uuids = vec![]; - let mut tx = db.begin().await?; - let trivial_jobs = sqlx::query!("INSERT INTO v2_job_completed AS cj - ( workspace_id - , id - , duration_ms - , result - , canceled_by - , canceled_reason - , flow_status - , status - , worker - ) - SELECT q.workspace_id - , q.id - , 0 - , $4 - , $1 - , 'cancel all' - , (SELECT flow_status FROM v2_job_status WHERE id = q.id) - , 'canceled'::job_status - , worker - FROM v2_job_queue q - JOIN v2_job USING (id) - WHERE q.id = any($2) AND running = false AND parent_job IS NULL AND q.workspace_id = $3 AND trigger IS NULL - FOR UPDATE SKIP LOCKED - ON CONFLICT (id) DO NOTHING RETURNING id AS \"id!\"", username, &jobs, w_id, serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}})) - .fetch_all(&mut *tx) - .await?.into_iter().map(|x| x.id).collect::>(); - - sqlx::query!( - "DELETE FROM v2_job_queue WHERE id = any($1) AND workspace_id = $2", - &trivial_jobs, - w_id - ) - .execute(&mut *tx) - .await?; - tx.commit().await?; - - // sqlx::query!( - // "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = 'cancelled all by user' WHERE id IN (SELECT id FROM queue where id = any($2) AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED) RETURNING id", - // username, - // &jobs, - // w_id - // ).execute(db).await?; - for job_id in jobs.into_iter() { - if trivial_jobs.contains(&job_id) { - continue; - } - match tokio::time::timeout(tokio::time::Duration::from_secs(5), async move { - let tx = db.begin().await?; - let (tx, _) = windmill_queue::cancel_job( - username, - None, - job_id.clone(), - w_id, - tx, - db, - false, - false, - ) - .await?; - tx.commit().await?; - Ok::<_, anyhow::Error>(()) - }) - .await - { - Ok(result) => match result { - Ok(_) => { - uuids.push(job_id); - } - Err(e) => { - tracing::error!("Failed to cancel job {:?}: {:?}", job_id, e); - } - }, - Err(_) => { - tracing::error!( - "Timeout while trying to cancel job {:?} after 5 seconds", - job_id - ); - } - } - } - - uuids.extend(trivial_jobs); - - Ok(Json(uuids)) -} - async fn cancel_selection( authed: ApiAuthed, Extension(db): Extension, @@ -1593,7 +1488,17 @@ async fn cancel_selection( .await?; tx.commit().await?; - cancel_jobs(jobs_to_cancel, &db, authed.username.as_str(), w_id.as_str()).await + cancel( + &db, + &jobs_to_cancel, + w_id.as_str(), + authed.username.as_str(), + "cancel selection", + false, + true, + ) + .await + .map(Json) } async fn list_filtered_uuids( @@ -3482,24 +3387,17 @@ impl Drop for Guard { tracing::info!("http connection broke, marking job {id} as canceled"); tokio::spawn(async move { - let cancel_f = async { - let tx = db.begin().await?; - let (tx, _) = cancel_job( - &username, - Some("http connection broke".to_string()), - id, - &w_id, - tx, - &db, - false, - false, - ) - .await?; - tx.commit().await?; - Ok::<_, anyhow::Error>(()) - }; - - if let Err(e) = cancel_f.await { + let jobs = &[id]; + let cancel_fut = cancel( + &db, + jobs, + &w_id, + &username, + "http connection broke", + false, + true, + ); + if let Err(e) = cancel_fut.await { tracing::error!( "Error marking job as canceled after http connection broke: {e}" ); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index f561549af8..dec858b4be 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6,12 +6,12 @@ * LICENSE-AGPL for a copy of the license. */ -use std::{borrow::Borrow, collections::HashMap, sync::Arc, vec}; +use std::{borrow::Borrow, collections::HashMap, collections::HashSet, sync::Arc, vec}; use anyhow::Context; use async_recursion::async_recursion; use chrono::{DateTime, Duration, Utc}; -use futures::future::TryFutureExt; +use futures::{stream, Stream, StreamExt, TryFutureExt, TryStreamExt}; use itertools::Itertools; #[cfg(feature = "prometheus")] use prometheus::IntCounter; @@ -19,14 +19,13 @@ use regex::Regex; use reqwest::Client; use serde::{ser::SerializeMap, Serialize}; use serde_json::{json, value::RawValue}; -use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction}; +use sqlx::{types::Json, FromRow, PgExecutor, Pool, Postgres, Transaction}; use tokio::{sync::RwLock, time::sleep}; use ulid::Ulid; use uuid::Uuid; use windmill_audit::audit_ee::{audit_log, AuditAuthor}; use windmill_audit::ActionKind; -use windmill_common::utils::now_from_db; use windmill_common::{ auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username}, cache::{self, FlowData}, @@ -121,174 +120,169 @@ const SCHEDULE_ERROR_HANDLER_USER_EMAIL: &str = "schedule_error_handler@windmill #[cfg(any(feature = "enterprise", feature = "cloud"))] const SCHEDULE_RECOVERY_HANDLER_USER_EMAIL: &str = "schedule_recovery_handler@windmill.dev"; -#[derive(Clone, Debug)] -pub struct CanceledBy { - pub username: Option, - pub reason: Option, -} - -pub async fn cancel_single_job<'c>( +pub async fn cancel( + db: &DB, + jobs: &[Uuid], + workspace_id: &str, username: &str, - reason: Option, - job_running: Arc, - w_id: &str, - mut tx: Transaction<'c, Postgres>, - db: &Pool, - force_cancel: bool, -) -> error::Result<(Transaction<'c, Postgres>, Option)> { - if force_cancel || (job_running.parent_job.is_none() && !job_running.running) { - let username = username.to_string(); - let w_id = w_id.to_string(); - let db = db.clone(); - let job_running = job_running.clone(); - tokio::task::spawn(async move { - let reason: String = reason - .clone() - .unwrap_or_else(|| "unexplicited reasons".to_string()); - let e = serde_json::json!({"message": format!("Job canceled: {reason} by {username}"), "name": "Canceled", "reason": reason, "canceler": username}); - append_logs( - &job_running.id, - w_id.to_string(), - format!("canceled by {username}: (force cancel: {force_cancel})"), - &db, - ) - .await; - let add_job = add_completed_job_error( - &db, - &job_running, - job_running.mem_peak.unwrap_or(0), - Some(CanceledBy { username: Some(username.to_string()), reason: Some(reason) }), - e, - "server", - false, - None, - ) - .await; - - if let Err(e) = add_job { - tracing::error!("Failed to add canceled job: {}", e); - } - }); + reason: &str, + force: bool, + authed: bool, +) -> error::Result> { + let mut roots; + // On force cancel, prepend flow root jobs to the list: + let jobs = if !force { + jobs } else { - let id: Option = sqlx::query_scalar!( - "UPDATE v2_job_queue SET canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 AND (canceled_by IS NULL OR canceled_reason != $2) RETURNING id", - username, - reason, - job_running.id, - w_id + roots = sqlx::query_scalar!( + "SELECT flow_innermost_root_job AS \"root!\" FROM v2_job + WHERE id = ANY($1) AND flow_innermost_root_job IS NOT NULL", + jobs ) - .fetch_optional(&mut *tx) + .fetch_all(db) .await?; - if let Some(id) = id { - tracing::info!("Soft cancelling job {}", id); - } - } - - Ok((tx, Some(job_running.id))) -} - -pub async fn cancel_job<'c>( - username: &str, - reason: Option, - id: Uuid, - w_id: &str, - mut tx: Transaction<'c, Postgres>, - db: &Pool, - force_cancel: bool, - require_anonymous: bool, -) -> error::Result<(Transaction<'c, Postgres>, Option)> { - let job = get_queued_job_tx(id, &w_id, &mut tx).await?; - - if job.is_none() { - return Ok((tx, None)); - } - - if require_anonymous && job.as_ref().unwrap().created_by != "anonymous" { - return Err(Error::BadRequest( - "You are not logged in and this job was not created by an anonymous user like you so you cannot cancel it".to_string(), - )); - } - - let mut job = job.unwrap(); - if force_cancel { - // if force canceling a flow step, make sure we force cancel from the highest parent - loop { - if job.parent_job.is_none() { - break; - } - match get_queued_job_tx(job.parent_job.unwrap(), &w_id, &mut tx).await? { - Some(j) => { - job = j; - } - None => break, - } - } - } - - // prevent cancelling a future tick of a schedule - if let Some(schedule_path) = job.schedule_path.as_ref() { - let now = now_from_db(&mut *tx).await?; - if job.scheduled_for > now { - return Err(Error::BadRequest( - format!( - "Cannot cancel a future tick of a schedule, cancel the schedule direcly ({})", - schedule_path - ) - .to_string(), + roots.extend(jobs.iter().cloned()); + &roots + }; + // Check if the user is authorized to cancel authed jobs: + if !authed { + let authed_jobs = sqlx::query_scalar!( + "SELECT COUNT(*) FROM v2_job WHERE id = ANY($1) AND created_by != 'anonymous'", + jobs + ) + .fetch_one(db) + .await? + .unwrap_or_default(); + if authed_jobs > 0 { + return Err(Error::NotAuthorized( + "Only authenticated users can cancel authed jobs".into(), )); } } - - let job = Arc::new(job); - - // get all children - let mut jobs = vec![job.id]; - let mut jobs_to_cancel = vec![]; - while !jobs.is_empty() { - let p_job = jobs.pop(); - let new_jobs = sqlx::query_scalar!( - "SELECT id AS \"id!\" FROM v2_job WHERE parent_job = $1 AND workspace_id = $2", - p_job, - w_id - ) - .fetch_all(&mut *tx) + let batches = jobs.chunks(5000).map(|batch| batch.to_vec()).collect_vec(); + let batches = stream::iter(batches.into_iter()) + .map(|batch| cancel_inner(db, batch, workspace_id, username, reason, force, authed)) + .buffer_unordered(32) + .try_collect::>() .await?; - jobs.extend(new_jobs.clone()); - jobs_to_cancel.extend(new_jobs); - } - jobs.reverse(); + Ok(batches.into_iter().flatten().collect()) +} - let (ntx, _) = cancel_single_job( +async fn cancel_inner( + db: &DB, + jobs: Vec, + workspace_id: &str, + username: &str, + reason: &str, + force: bool, + authed: bool, +) -> error::Result> { + // 1. Lock & soft cancel all queued jobs: + // This query both acquires an exclusive row-level lock on the job and soft cancels it. + // It also returns whenever the job is trivial (i.e. not running, no parent, no trigger). + // NOTE: + // - The `canceled_by` field is used to prevent the job from being canceled multiple times. + // - Schedule next ticks aren't cancelable. + let mut tx = db.begin().await?; + let mut st = sqlx::query!( + "UPDATE v2_job_queue q SET + canceled_by = $1, + canceled_reason = $2, + scheduled_for = NOW(), + suspend = 0 + FROM v2_job j + WHERE q.id = ANY($3) AND j.id = q.id AND q.workspace_id = $4 + AND canceled_by IS NULL + AND (j.trigger IS DISTINCT FROM 'schedule' OR running = true) + RETURNING q.id, (q.running = false AND parent_job IS NULL AND trigger IS NULL) AS trivial", username, - reason.clone(), - job.clone(), - w_id, - tx, - db, - force_cancel, + reason, + &jobs, + workspace_id ) - .await?; - tx = ntx; - - // cancel children - for job_id in jobs_to_cancel { - let job = get_queued_job_tx(job_id, &w_id, &mut tx).await?; - - if let Some(job) = job { - let (ntx, _) = cancel_single_job( - username, - reason.clone(), - Arc::new(job), - w_id, - tx, - db, - force_cancel, - ) - .await?; - tx = ntx; + .fetch(&mut *tx) + .map(|result| result.map(|row| (row.id, row.trivial))); + let (mut jobs, mut trivial_jobs) = ( + Vec::with_capacity(jobs.len()), + Vec::with_capacity(jobs.len()), + ); + while let Some((id, trivial)) = st.try_next().await? { + jobs.push(id); + if trivial.unwrap_or(false) { + trivial_jobs.push(id); } } - Ok((tx, Some(id))) + if jobs.is_empty() { + return Ok(jobs); + } + drop(st); + // 2. Terminate trivial jobs in batch: + let error = json!({ + "error": { + "message": format!("Job canceled by {username}"), + "name": "Canceled", + "reason": reason, + "canceler": username + } + }); + let result = Json(&error); + let trivial_jobs = terminate( + &mut *tx, + &trivial_jobs, + None, + false, + false, + result, + None, + None, + ) + .map_ok(|status| status.id) + .try_collect::>() + .await?; + // 3. Commit: + tx.commit().await?; + // 4. Complete non-trivial jobs in parallel: + // This help workers draining out canceled jobs faster. + let cancel = |id: Uuid| async move { + // 4.1. Cancel children jobs: + let children = sqlx::query_scalar!("SELECT id FROM v2_job WHERE parent_job = $1", id) + .fetch_all(db) + .await?; + if !children.is_empty() { + let _ = cancel(db, &children, workspace_id, username, reason, force, authed).await?; + } + if !force { + // 4.2. Wait for the job to complete: + use tokio::time; + let wait_completed_fut = async move { + while !job_is_complete(db, id, workspace_id).await? { + sleep(time::Duration::from_millis(100)).await; + } + Ok::<_, Error>(()) + }; + match time::timeout(time::Duration::from_secs(2), wait_completed_fut).await { + Ok(Ok(_)) => return Ok::<_, Error>(()), + Ok(Err(err)) => return Err(err), + Err(_) => { /* ignore timeout */ } + } + } + // 4.2. Force complete the job: + if let Some(job) = get_queued_job(&id, workspace_id, db).await? { + add_completed_job(&db, &job, false, false, result, None, 0, false, None).await?; + } + Ok::<_, Error>(()) + }; + let _ = stream::iter(jobs.iter().cloned().filter(|id| !trivial_jobs.contains(id))) + .map(|id| async move { + if let Err(err) = cancel(id).await { + tracing::error!("Failed to complete job {id}: {err:#}"); + } + }) + .buffer_unordered(64) + .collect::<()>() + .await; + // 5. Return successfully canceled jobs: + Ok(jobs) } /* TODO retry this? */ @@ -355,35 +349,17 @@ async fn cancel_persistent_script_jobs_internal<'c>( w_id: &str, db: &Pool, ) -> error::Result> { - let mut tx = db.begin().await?; - // we could have retrieved the job IDs in the first query where we retrieve the hashes, but just in case a job was inserted in the queue right in-between the two above query, we re-do the fetch here - let jobs_to_cancel = sqlx::query_scalar::<_, Uuid>( - "SELECT id FROM v2_as_queue WHERE workspace_id = $1 AND script_path = $2 AND canceled = false", + let jobs_to_cancel = sqlx::query_scalar!( + "SELECT id as \"id!\" FROM v2_as_queue WHERE workspace_id = $1 AND script_path = $2 AND canceled = false", + w_id, + script_path ) - .bind(w_id) - .bind(script_path) - .fetch_all(&mut *tx) + .fetch_all(db) .await?; - // Then we cancel all the jobs currently in the queue, one by one - for queued_job_id in jobs_to_cancel.clone() { - let (new_tx, _) = cancel_job( - username, - reason.clone(), - queued_job_id, - w_id, - tx, - db, - false, - false, - ) - .await?; - tx = new_tx; - } - tx.commit().await?; - - return Ok(jobs_to_cancel); + let reason = reason.unwrap_or_else(|| "cancel persistent scripts".into()); + cancel(db, &jobs_to_cancel, w_id, username, &reason, false, true).await } #[derive(Serialize, Debug)] @@ -489,11 +465,9 @@ pub async fn add_completed_job_error( db: &Pool, queued_job: &QueuedJob, mem_peak: i32, - canceled_by: Option, e: serde_json::Value, _worker_name: &str, flow_is_done: bool, - duration: Option, ) -> Result { #[cfg(feature = "prometheus")] register_metric( @@ -529,9 +503,8 @@ pub async fn add_completed_job_error( Json(&result), None, mem_peak, - canceled_by, flow_is_done, - duration, + None, ) .await?; Ok(result) @@ -541,6 +514,66 @@ lazy_static::lazy_static! { pub static ref GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE: Option = std::env::var("GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE").ok(); } +struct TerminationStatus { + pub id: Uuid, + pub canceled: bool, + pub duration_ms: i64, +} + +fn terminate<'c, T: Serialize + ValidableJson>( + e: impl PgExecutor<'c> + 'c, + id: &[Uuid], + duration: Option, + skipped: bool, + success: bool, + result: Json<&T>, + result_columns: Option<&[String]>, + memory_peak: Option, +) -> impl Stream> + 'c { + sqlx::query_as!( + TerminationStatus, + "WITH queued AS ( + DELETE FROM v2_job_queue + WHERE id = ANY($1::UUID[]) + RETURNING + id, workspace_id, started_at, worker, canceled_by, canceled_reason, extras + ), queued_and_runtime AS ( + SELECT queued.*, GREATEST($7, memory_peak) AS memory_peak, flow_status, + workflow_as_code_status + FROM queued + LEFT JOIN v2_job_runtime USING (id) + LEFT JOIN v2_job_status USING (id) + ) INSERT INTO v2_job_completed ( + id, workspace_id, started_at, worker, memory_peak, flow_status, workflow_as_code_status, + result, result_columns, canceled_by, canceled_reason, extras, + duration_ms, + status + ) SELECT + id, workspace_id, started_at, worker, memory_peak, flow_status, workflow_as_code_status, + $5, $6, canceled_by, canceled_reason, extras, + COALESCE($2::BIGINT, CASE + WHEN started_at IS NULL THEN 0 + ELSE (EXTRACT('epoch' FROM NOW()) - EXTRACT('epoch' FROM started_at)) * 1000 + END) AS duration_ms, + CASE + WHEN canceled_by IS NOT NULL THEN 'canceled'::job_status + WHEN $3::BOOLEAN THEN 'skipped'::job_status + WHEN $4::BOOLEAN THEN 'success'::job_status + ELSE 'failure'::job_status + END AS status + FROM queued_and_runtime + RETURNING id, status = 'canceled' AS \"canceled!\", duration_ms", + /* $1 */ id, + /* $2 */ duration, + /* $3 */ skipped, + /* $4 */ success, + /* $5 */ result as Json<&T>, + /* $6 */ result_columns as Option<&[String]>, + /* $7 */ memory_peak, + ) + .fetch(e) +} + pub async fn add_completed_job( db: &Pool, queued_job: &QueuedJob, @@ -549,7 +582,6 @@ pub async fn add_completed_job( result: Json<&T>, result_columns: Option>, mem_peak: i32, - canceled_by: Option, flow_is_done: bool, duration: Option, ) -> Result { @@ -564,11 +596,9 @@ pub async fn add_completed_job( } let result_columns = result_columns.as_ref(); - let _job_id = queued_job.id; - let (opt_uuid, _duration, _skip_downstream_error_handlers) = (|| async { + let job_id = queued_job.id; + let (opt_uuid, canceled, _duration, _skip_downstream_error_handlers) = (|| async { let mut tx = db.begin().await?; - - let job_id = queued_job.id; // tracing::error!("1 {:?}", start.elapsed()); tracing::debug!( @@ -577,47 +607,36 @@ pub async fn add_completed_job( serde_json::to_string(&result).unwrap_or_else(|_| "".to_string()) ); - let mem_peak = mem_peak.max(queued_job.mem_peak.unwrap_or(0)); // add_time!(bench, "add_completed_job query START"); - - let _duration = sqlx::query_scalar!( - "INSERT INTO v2_job_completed AS cj - ( workspace_id - , id - , started_at - , duration_ms - , result - , result_columns - , canceled_by - , canceled_reason - , flow_status - , workflow_as_code_status - , memory_peak - , status - , worker - ) - SELECT q.workspace_id, q.id, started_at, COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000), $3, $10, $5, $6, - flow_status, workflow_as_code_status, - $8, CASE WHEN $4::BOOL THEN 'canceled'::job_status - WHEN $7::BOOL THEN 'skipped'::job_status - WHEN $2::BOOL THEN 'success'::job_status - ELSE 'failure'::job_status END AS status, - q.worker - FROM v2_job_queue q LEFT JOIN v2_job_status USING (id) WHERE q.id = $1 - ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3 RETURNING duration_ms AS \"duration_ms!\"", - /* $1 */ queued_job.id, - /* $2 */ success, - /* $3 */ result as Json<&T>, - /* $4 */ canceled_by.is_some(), - /* $5 */ canceled_by.clone().map(|cb| cb.username).flatten(), - /* $6 */ canceled_by.clone().map(|cb| cb.reason).flatten(), - /* $7 */ skipped, - /* $8 */ if mem_peak > 0 { Some(mem_peak) } else { None }, - /* $9 */ duration, - /* $10 */ result_columns as Option<&Vec>, + // acquire an exclusive row-level lock for this job: + let lock = sqlx::query_scalar!( + "SELECT NULL AS \"lock!: ()\" FROM v2_job_queue WHERE id = $1 FOR UPDATE", + job_id ) - .fetch_one(&mut *tx) + .fetch_optional(&mut *tx) + .await?; + if let None = lock { + tracing::warn!( + "Job {} already completed, discarding result `{}`", + job_id, + serde_json::to_string(&result).unwrap_or_default() + ); + return Ok((None, false, 0, false)); + } + + let TerminationStatus { canceled, duration_ms: _duration, .. } = terminate( + &mut *tx, + &[job_id], + duration, + skipped, + success, + result, + result_columns.map(Vec::as_slice), + if mem_peak > 0 { Some(mem_peak) } else { None }, + ) + .next() .await + .ok_or_else(|| Error::NotFound(format!("Job {job_id}")))? .map_err(|e| Error::internal_err(format!("Could not add completed job {job_id}: {e:#}")))?; if let Some(labels) = result.wm_labels() { @@ -634,20 +653,19 @@ pub async fn add_completed_job( .map_err(|e| Error::InternalErr(format!("Could not update job labels: {e:#}")))?; } + // `workflow_as_code`: if !queued_job.is_flow_step { if let Some(parent_job) = queued_job.parent_job { let _ = sqlx::query_scalar!( - "UPDATE v2_job_status SET - workflow_as_code_status = jsonb_set( - jsonb_set( - COALESCE(workflow_as_code_status, '{}'::jsonb), - array[$1], - COALESCE(workflow_as_code_status->$1, '{}'::jsonb) - ), - array[$1, 'duration_ms'], - to_jsonb($2::bigint) - ) - WHERE id = $3", + "UPDATE v2_job_status SET workflow_as_code_status = jsonb_set( + jsonb_set( + COALESCE(workflow_as_code_status, '{}'::jsonb), + array[$1], + COALESCE(workflow_as_code_status->$1, '{}'::jsonb) + ), + array[$1, 'duration_ms'], + to_jsonb($2::bigint) + ) WHERE id = $3", &queued_job.id.to_string(), _duration, parent_job @@ -663,7 +681,6 @@ pub async fn add_completed_job( // tracing::error!("Added completed job {:#?}", queued_job); let mut _skip_downstream_error_handlers = false; - tx = delete_job(tx, &queued_job.workspace_id, job_id).await?; // tracing::error!("3 {:?}", start.elapsed()); if queued_job.is_flow_step { @@ -728,10 +745,9 @@ pub async fn add_completed_job( AND flow_status->'modules'->0->>'job' = $1 ) ) - FROM v2_job_completed WHERE id = $2 AND workspace_id = $3", + FROM v2_job_status WHERE id = $2", Uuid::nil().to_string(), - &queued_job.id, - &queued_job.workspace_id + &queued_job.id ).fetch_optional(&mut *tx).await?.flatten().unwrap_or(false); if schedule_next_tick { @@ -747,7 +763,7 @@ pub async fn add_completed_job( match err { Error::QuotaExceeded(_) => (), // scheduling next job failed and could not disable schedule => make zombie job to retry - _ => return Ok((Some(job_id), 0, true)), + _ => return Ok((Some(job_id), canceled, 0, true)), } }; } @@ -858,7 +874,7 @@ pub async fn add_completed_job( "inserted completed job: {} (success: {success})", queued_job.id ); - Ok((None, _duration, _skip_downstream_error_handlers)) as windmill_common::error::Result<(Option, i64, bool)> + Result::Ok((None, canceled, _duration, _skip_downstream_error_handlers)) }) .retry( ConstantBuilder::default() @@ -996,13 +1012,8 @@ pub async fn add_completed_job( ); } - if let Err(err) = send_error_to_workspace_handler( - &queued_job, - canceled_by.is_some(), - db, - Json(&result), - ) - .await + if let Err(err) = + send_error_to_workspace_handler(&queued_job, canceled, db, Json(&result)).await { match err { Error::QuotaExceeded(_) => {} @@ -1021,7 +1032,7 @@ pub async fn add_completed_job( } } - if !queued_job.is_flow_step && queued_job.job_kind == JobKind::Script && canceled_by.is_none() { + if !queued_job.is_flow_step && queued_job.job_kind == JobKind::Script && !canceled { if let Some(hash) = queued_job.script_hash { let p = sqlx::query_scalar!( "SELECT restart_unless_cancelled FROM script WHERE hash = $1 AND workspace_id = $2", @@ -2579,39 +2590,6 @@ async fn extract_result_from_job_result( } } -pub async fn delete_job<'c>( - mut tx: Transaction<'c, Postgres>, - w_id: &str, - job_id: Uuid, -) -> windmill_common::error::Result> { - #[cfg(feature = "prometheus")] - if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { - QUEUE_DELETE_COUNT.inc(); - } - - let job_removed = sqlx::query_scalar!( - "DELETE FROM v2_job_queue WHERE workspace_id = $1 AND id = $2 RETURNING 1", - w_id, - job_id - ) - .fetch_optional(&mut *tx) - .await; - - if let Err(job_removed) = job_removed { - tracing::error!( - "Job {job_id} could not be deleted: {job_removed}. This is not necessarily an error, as the job might have been deleted by another process such as in the case of cancelling" - ); - } else { - let job_removed = job_removed.unwrap().flatten().unwrap_or(0); - if job_removed != 1 { - tracing::error!("Job {job_id} could not be deleted, returned not 1: {job_removed}. This is not necessarily an error, as the job might have been deleted by another process such as in the case of cancelling"); - } - } - - tracing::debug!("Job {job_id} deleted"); - Ok(tx) -} - pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result { Ok(sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM v2_job_completed WHERE id = $1 AND workspace_id = $2)", @@ -2623,22 +2601,6 @@ pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result( - id: Uuid, - w_id: &str, - tx: &mut Transaction<'c, Postgres>, -) -> error::Result> { - sqlx::query_as::<_, QueuedJob>( - "SELECT * - FROM v2_as_queue WHERE id = $1 AND workspace_id = $2", - ) - .bind(id) - .bind(w_id) - .fetch_optional(&mut **tx) - .await - .map_err(Into::into) -} - pub async fn get_queued_job(id: &Uuid, w_id: &str, db: &DB) -> error::Result> { sqlx::query_as::<_, QueuedJob>( "SELECT * diff --git a/backend/windmill-worker/src/ansible_executor.rs b/backend/windmill-worker/src/ansible_executor.rs index fee2eaa119..98dc0dc6a4 100644 --- a/backend/windmill-worker/src/ansible_executor.rs +++ b/backend/windmill-worker/src/ansible_executor.rs @@ -19,7 +19,7 @@ use windmill_common::{ worker::{to_raw_value, write_file, write_file_at_user_defined_location, WORKER_CONFIG}, }; use windmill_parser_yaml::{AnsibleRequirements, ResourceOrVariablePath}; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; use crate::{ bash_executor::BIN_BASH, @@ -53,7 +53,6 @@ async fn handle_ansible_python_deps( worker_name: &str, worker_dir: &str, mem_peak: &mut i32, - canceled_by: &mut Option, occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { create_dependencies_dir(job_dir).await; @@ -78,7 +77,6 @@ async fn handle_ansible_python_deps( job_id, &requirements, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -106,7 +104,6 @@ async fn handle_ansible_python_deps( job_id, w_id, mem_peak, - canceled_by, db, worker_name, job_dir, @@ -128,7 +125,6 @@ async fn install_galaxy_collections( worker_name: &str, w_id: &str, mem_peak: &mut i32, - canceled_by: &mut Option, db: &sqlx::Pool, occupancy_metrics: &mut OccupancyMetrics, ) -> anyhow::Result<()> { @@ -166,7 +162,6 @@ async fn install_galaxy_collections( job_id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, @@ -188,7 +183,6 @@ pub async fn handle_ansible_job( worker_name: &str, job: &QueuedJob, mem_peak: &mut i32, - canceled_by: &mut Option, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, inner_content: &String, @@ -217,7 +211,6 @@ pub async fn handle_ansible_job( worker_name, worker_dir, mem_peak, - canceled_by, occupancy_metrics, ) .await?; @@ -290,7 +283,6 @@ pub async fn handle_ansible_job( worker_name, &job.workspace_id, mem_peak, - canceled_by, db, occupancy_metrics, ) @@ -426,7 +418,6 @@ fi &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 7b536afdae..5ee41fa94a 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -25,7 +25,7 @@ use windmill_common::DB; #[cfg(feature = "dind")] use windmill_common::error::to_anyhow; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; lazy_static::lazy_static! { pub static ref BIN_BASH: String = std::env::var("BASH_PATH").unwrap_or_else(|_| "/bin/bash".to_string()); @@ -62,7 +62,6 @@ lazy_static::lazy_static! { #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_bash_job( mem_peak: &mut i32, - canceled_by: &mut Option, job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -206,7 +205,6 @@ exit $exit_status &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, @@ -226,7 +224,6 @@ exit $exit_status db, job.timeout, mem_peak, - canceled_by, worker_name, occupancy_metrics, _killpill_rx, @@ -271,7 +268,6 @@ async fn handle_docker_job( db: &DB, job_timeout: Option, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, occupancy_metrics: &mut OccupancyMetrics, killpill_rx: &mut tokio::sync::broadcast::Receiver<()>, @@ -361,7 +357,6 @@ async fn handle_docker_job( job_timeout, db, mem_peak, - canceled_by, wait_f, worker_name, workspace_id, @@ -458,7 +453,6 @@ fn raw_to_string(x: &str) -> String { #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_powershell_job( mem_peak: &mut i32, - canceled_by: &mut Option, job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -548,7 +542,6 @@ pub async fn handle_powershell_job( &job.id, db, mem_peak, - canceled_by, child, false, worker_name, @@ -759,7 +752,6 @@ $env:PSModulePath = \"{};$PSModulePathBackup\"", &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, diff --git a/backend/windmill-worker/src/bigquery_executor.rs b/backend/windmill-worker/src/bigquery_executor.rs index e4674ed6ee..f965f73360 100644 --- a/backend/windmill-worker/src/bigquery_executor.rs +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -10,7 +10,6 @@ use windmill_common::{error::Error, worker::to_raw_value}; use windmill_parser_sql::{ parse_bigquery_sig, parse_db_resource, parse_sql_blocks, parse_sql_statement_named_params, }; -use windmill_queue::CanceledBy; use serde::Deserialize; @@ -209,7 +208,6 @@ pub async fn do_bigquery( query: &str, db: &sqlx::Pool, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, column_order: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, @@ -362,7 +360,6 @@ pub async fn do_bigquery( job.timeout, db, mem_peak, - canceled_by, result_f.map_err(to_anyhow), worker_name, &job.workspace_id, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 4fc471df91..c644c5803f 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -11,7 +11,7 @@ use serde_json::value::RawValue; use sha2::Digest; use uuid::Uuid; use windmill_parser_ts::remove_pinned_imports; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; #[cfg(feature = "enterprise")] use crate::common::build_envs_map; @@ -96,7 +96,6 @@ fn split_lockfile(lockfile: &str) -> (&str, Option<&str>, bool, bool) { pub async fn gen_bun_lockfile( mem_peak: &mut i32, - canceled_by: &mut Option, job_id: &Uuid, w_id: &str, db: Option<&sqlx::Pool>, @@ -160,7 +159,6 @@ pub async fn gen_bun_lockfile( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -185,7 +183,6 @@ pub async fn gen_bun_lockfile( if !empty_deps { install_bun_lockfile( mem_peak, - canceled_by, job_id, w_id, db, @@ -272,7 +269,6 @@ registry = {} pub async fn install_bun_lockfile( mem_peak: &mut i32, - canceled_by: &mut Option, job_id: &Uuid, w_id: &str, db: Option<&sqlx::Pool>, @@ -343,7 +339,6 @@ pub async fn install_bun_lockfile( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -492,7 +487,6 @@ pub async fn generate_wrapper_mjs( db: &sqlx::Pool, timeout: Option, mem_peak: &mut i32, - canceled_by: &mut Option, common_bun_proc_envs: &HashMap, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, ) -> Result<()> { @@ -514,7 +508,6 @@ pub async fn generate_wrapper_mjs( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -541,7 +534,6 @@ pub async fn generate_bun_bundle( db: Option>, timeout: Option, mem_peak: &mut i32, - canceled_by: &mut Option, common_bun_proc_envs: &HashMap, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, ) -> Result<()> { @@ -564,7 +556,6 @@ pub async fn generate_bun_bundle( job_id, &db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -739,7 +730,6 @@ pub async fn prebundle_bun_script( db.clone(), None, &mut 0, - &mut None, &common_bun_proc_envs, occupancy_metrics, ) @@ -834,7 +824,6 @@ pub async fn handle_bun_job( requirements_o: Option<&String>, codebase: Option<&String>, mem_peak: &mut i32, - canceled_by: &mut Option, job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -954,7 +943,6 @@ pub async fn handle_bun_job( if !skip_install { install_bun_lockfile( mem_peak, - canceled_by, &job.id, &job.workspace_id, Some(db), @@ -992,7 +980,6 @@ pub async fn handle_bun_job( append_logs(&job.id, &job.workspace_id, logs1, db).await; let _ = gen_bun_lockfile( mem_peak, - canceled_by, &job.id, &job.workspace_id, Some(db), @@ -1241,7 +1228,6 @@ try {{ Some(db.clone()), job.timeout, mem_peak, - canceled_by, &common_bun_proc_envs, &mut Some(occupancy_metrics), ) @@ -1283,7 +1269,6 @@ try {{ db, job.timeout, mem_peak, - canceled_by, &common_bun_proc_envs, &mut Some(occupancy_metrics), ) @@ -1330,7 +1315,6 @@ try {{ job.timeout, db, mem_peak, - canceled_by, worker_name, &job.workspace_id, false, @@ -1478,7 +1462,6 @@ try {{ &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, @@ -1570,7 +1553,6 @@ pub async fn start_worker( ) -> Result<()> { let mut logs = "".to_string(); let mut mem_peak: i32 = 0; - let mut canceled_by: Option = None; tracing::info!("Starting worker {w_id};{script_path} (codebase: {codebase:?}"); if !codebase.is_some() { let _ = write_file(job_dir, "main.ts", inner_content)?; @@ -1633,7 +1615,6 @@ pub async fn start_worker( install_bun_lockfile( &mut mem_peak, - &mut canceled_by, &Uuid::nil(), &w_id, Some(db), @@ -1650,7 +1631,6 @@ pub async fn start_worker( logs.push_str("\n\n--- BUN INSTALL ---\n"); let _ = gen_bun_lockfile( &mut mem_peak, - &mut canceled_by, &Uuid::nil(), &w_id, Some(db), @@ -1757,7 +1737,6 @@ for await (const line of Readline.createInterface({{ input: process.stdin }})) { db, None, &mut mem_peak, - &mut canceled_by, &common_bun_proc_envs, &mut None, ) diff --git a/backend/windmill-worker/src/csharp_executor.rs b/backend/windmill-worker/src/csharp_executor.rs index 61f53ab65a..07c740df2a 100644 --- a/backend/windmill-worker/src/csharp_executor.rs +++ b/backend/windmill-worker/src/csharp_executor.rs @@ -23,8 +23,6 @@ use windmill_common::jobs::QueuedJob; #[cfg(feature = "csharp")] use windmill_queue::append_logs; -use windmill_queue::CanceledBy; - #[cfg(feature = "csharp")] use crate::{ common::{ @@ -66,7 +64,6 @@ pub async fn generate_nuget_lockfile( job_id: &Uuid, code: &str, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -118,7 +115,6 @@ pub async fn generate_nuget_lockfile( job_id, db, mem_peak, - canceled_by, gen_lockfile_process, false, worker_name, @@ -148,7 +144,6 @@ pub async fn generate_nuget_lockfile( _job_id: &Uuid, _code: &str, _mem_peak: &mut i32, - _canceled_by: &mut Option, _job_dir: &str, _db: &sqlx::Pool, _worker_name: &str, @@ -309,7 +304,6 @@ namespace WindmillScriptCSharpInternal {{ async fn build_cs_proj( job_id: &Uuid, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -373,7 +367,6 @@ async fn build_cs_proj( job_id, db, mem_peak, - canceled_by, build_cs_process, false, worker_name, @@ -429,7 +422,6 @@ fn remove_lines_from_text(contents: &str, indices_to_remove: Vec) -> Stri #[cfg(not(feature = "csharp"))] pub async fn handle_csharp_job( _mem_peak: &mut i32, - _canceled_by: &mut Option, _job: &QueuedJob, _db: &sqlx::Pool, _client: &AuthedClientBackgroundTask, @@ -448,7 +440,6 @@ pub async fn handle_csharp_job( #[cfg(feature = "csharp")] pub async fn handle_csharp_job( mem_peak: &mut i32, - canceled_by: &mut Option, job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -516,7 +507,6 @@ pub async fn handle_csharp_job( build_cs_proj( &job.id, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -621,7 +611,6 @@ pub async fn handle_csharp_job( &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index fb3a30176c..6ab97c6e3e 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -185,14 +185,14 @@ pub async fn handle_dedicated_process( let result = Arc::new(result); append_logs(&job.id, &job.workspace_id, logs.clone(), db).await; if line.starts_with("wm_res[success]:") { - job_completed_tx.send(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string(), duration: None }).await.unwrap() + job_completed_tx.send(JobCompleted { job , result, result_columns: None, mem_peak: 0, success: true, cached_res_path: None, token: token.to_string(), duration: None }).await.unwrap() } else { - job_completed_tx.send(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None }).await.unwrap() + job_completed_tx.send(JobCompleted { job , result, result_columns: None, mem_peak: 0, success: false, cached_res_path: None, token: token.to_string(), duration: None }).await.unwrap() } }, Err(e) => { tracing::error!("Could not deserialize job result `{line}`: {e:?}"); - job_completed_tx.send(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None }).await.unwrap(); + job_completed_tx.send(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), result_columns: None, mem_peak: 0, success: false, cached_res_path: None, token: token.to_string(), duration: None }).await.unwrap(); }, }; logs = init_log.clone(); diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 6a7bed5c38..d786629529 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -3,7 +3,7 @@ use std::{collections::HashMap, process::Stdio}; use itertools::Itertools; use serde_json::value::RawValue; use uuid::Uuid; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; use crate::{ common::{ @@ -98,7 +98,6 @@ pub async fn generate_deno_lock( job_id: &Uuid, code: &str, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: Option<&sqlx::Pool>, w_id: &str, @@ -150,7 +149,6 @@ pub async fn generate_deno_lock( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -179,7 +177,6 @@ pub async fn generate_deno_lock( pub async fn handle_deno_job( requirements_o: Option<&String>, mem_peak: &mut i32, - canceled_by: &mut Option, job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -405,7 +402,6 @@ try {{ &job.id, db, mem_peak, - canceled_by, child, false, worker_name, diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 1a8156a1bb..1e91476355 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -12,7 +12,7 @@ use windmill_common::{ worker::{save_cache, write_file}, }; use windmill_parser_go::{parse_go_imports, REQUIRE_PARSE}; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; use crate::{ common::{ @@ -35,7 +35,6 @@ pub const GO_OBJECT_STORE_PREFIX: &str = "gobin/"; #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_go_job( mem_peak: &mut i32, - canceled_by: &mut Option, job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -83,7 +82,6 @@ pub async fn handle_go_job( &job.id, inner_content, mem_peak, - canceled_by, job_dir, db, true, @@ -203,7 +201,6 @@ func Run(req Req) (interface{{}}, error){{ &job.id, db, mem_peak, - canceled_by, build_go_process, false, worker_name, @@ -306,7 +303,6 @@ func Run(req Req) (interface{{}}, error){{ &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, @@ -346,7 +342,6 @@ pub async fn install_go_dependencies( job_id: &Uuid, code: &str, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, non_dep_job: bool, @@ -370,7 +365,6 @@ pub async fn install_go_dependencies( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -436,7 +430,6 @@ pub async fn install_go_dependencies( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, diff --git a/backend/windmill-worker/src/graphql_executor.rs b/backend/windmill-worker/src/graphql_executor.rs index 929235cccf..648cdc4c72 100644 --- a/backend/windmill-worker/src/graphql_executor.rs +++ b/backend/windmill-worker/src/graphql_executor.rs @@ -8,7 +8,6 @@ use windmill_common::jobs::QueuedJob; use windmill_common::worker::to_raw_value; use windmill_common::{error::Error, worker::CLOUD_HOSTED}; use windmill_parser_graphql::parse_graphql_sig; -use windmill_queue::CanceledBy; use serde::Deserialize; @@ -40,7 +39,6 @@ pub async fn do_graphql( query: &str, db: &sqlx::Pool, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, occupation_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { @@ -153,7 +151,6 @@ pub async fn do_graphql( job.timeout, db, mem_peak, - canceled_by, result_f, worker_name, &job.workspace_id, diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 7294950363..d2ba2d5654 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -17,7 +17,7 @@ use windmill_common::error::{self, Error}; use windmill_common::worker::{get_windmill_memory_usage, get_worker_memory_usage, CLOUD_HOSTED}; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; #[cfg(any(target_os = "linux", target_os = "macos"))] use std::os::unix::process::ExitStatusExt; @@ -94,7 +94,6 @@ pub async fn handle_child( job_id: &Uuid, db: &Pool, mem_peak: &mut i32, - canceled_by_ref: &mut Option, mut child: Child, nsjail: bool, worker: &str, @@ -138,7 +137,6 @@ pub async fn handle_child( job_id, db, mem_peak, - canceled_by_ref, Box::pin(stream::unfold((), move |_| async move { Some((get_mem_peak(pid, nsjail).await, ())) })), @@ -492,7 +490,6 @@ pub async fn run_future_with_polling_update_job_poller( timeout: Option, db: &DB, mem_peak: &mut i32, - canceled_by_ref: &mut Option, result_f: Fut, worker_name: &str, w_id: &str, @@ -509,7 +506,6 @@ where job_id, db, mem_peak, - canceled_by_ref, get_mem, worker_name, w_id, @@ -546,6 +542,11 @@ where Ok(rows) } +pub struct CanceledBy { + pub username: Option, + pub reason: Option, +} + pub enum UpdateJobPollingExit { Done(Option), AlreadyCompleted, @@ -555,7 +556,6 @@ pub async fn update_job_poller( job_id: Uuid, db: &DB, mem_peak: &mut i32, - canceled_by_ref: &mut Option, mut get_mem: S, worker_name: &str, w_id: &str, @@ -566,6 +566,7 @@ where S: stream::Stream + Unpin, { let update_job_interval = Duration::from_millis(500); + let mut cancellation = None; let db = db.clone(); @@ -666,9 +667,9 @@ where return UpdateJobPollingExit::AlreadyCompleted } if canceled_by.is_some() { - canceled_by_ref.replace(CanceledBy { - username: canceled_by.clone(), - reason: canceled_reason.clone(), + cancellation = Some(CanceledBy { + username: canceled_by, + reason: canceled_reason, }); break } @@ -679,7 +680,7 @@ where } tracing::info!("job {job_id} finished"); - UpdateJobPollingExit::Done(canceled_by_ref.clone()) + UpdateJobPollingExit::Done(cancellation) } /// takes stdout and stderr from Child, panics if either are not present diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index 9aab144296..d84a289bf1 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -46,7 +46,6 @@ use windmill_common::error::Error; use windmill_common::worker::{write_file, TMP_DIR}; use windmill_common::{flow_status::JobResult, DB}; -use windmill_queue::CanceledBy; use crate::{common::OccupancyMetrics, AuthedClient}; @@ -745,7 +744,6 @@ pub async fn eval_fetch_timeout( _job_timeout: Option, _db: &DB, _mem_peak: &mut i32, - _canceled_by: &mut Option, _worker_name: &str, _w_id: &str, _load_client: bool, @@ -765,7 +763,6 @@ pub async fn eval_fetch_timeout( job_timeout: Option, db: &DB, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, w_id: &str, load_client: bool, @@ -915,7 +912,6 @@ pub async fn eval_fetch_timeout( job_timeout, db, mem_peak, - canceled_by, async { result_f.await? }, worker_name, w_id, diff --git a/backend/windmill-worker/src/mssql_executor.rs b/backend/windmill-worker/src/mssql_executor.rs index 42d9c3aa8c..ac8a0cfb61 100644 --- a/backend/windmill-worker/src/mssql_executor.rs +++ b/backend/windmill-worker/src/mssql_executor.rs @@ -12,7 +12,7 @@ use windmill_common::error::{self, Error}; use windmill_common::worker::to_raw_value; use windmill_common::{error::to_anyhow, jobs::QueuedJob}; use windmill_parser_sql::{parse_db_resource, parse_mssql_sig}; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; use crate::common::{build_args_values, OccupancyMetrics}; use crate::handle_child::run_future_with_polling_update_job_poller; @@ -38,7 +38,6 @@ pub async fn do_mssql( query: &str, db: &sqlx::Pool, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { @@ -174,7 +173,6 @@ pub async fn do_mssql( job.timeout, db, mem_peak, - canceled_by, result_f, worker_name, &job.workspace_id, diff --git a/backend/windmill-worker/src/mysql_executor.rs b/backend/windmill-worker/src/mysql_executor.rs index d733901f93..24d8c1448e 100644 --- a/backend/windmill-worker/src/mysql_executor.rs +++ b/backend/windmill-worker/src/mysql_executor.rs @@ -19,7 +19,6 @@ use windmill_parser_sql::{ parse_db_resource, parse_mysql_sig, parse_sql_blocks, parse_sql_statement_named_params, RE_ARG_MYSQL_NAMED, }; -use windmill_queue::CanceledBy; use crate::{ common::{build_args_map, OccupancyMetrics}, @@ -108,7 +107,6 @@ pub async fn do_mysql( query: &str, db: &sqlx::Pool, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, column_order: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, @@ -293,7 +291,6 @@ pub async fn do_mysql( job.timeout, db, mem_peak, - canceled_by, result_f, worker_name, &job.workspace_id, diff --git a/backend/windmill-worker/src/oracledb_executor.rs b/backend/windmill-worker/src/oracledb_executor.rs index e6292181b2..5f14e80f72 100644 --- a/backend/windmill-worker/src/oracledb_executor.rs +++ b/backend/windmill-worker/src/oracledb_executor.rs @@ -17,7 +17,6 @@ use windmill_common::{ use windmill_parser_sql::{ parse_db_resource, parse_oracledb_sig, parse_sql_blocks, parse_sql_statement_named_params, }; -use windmill_queue::CanceledBy; use crate::{ common::{build_args_map, check_executor_binary_exists, OccupancyMetrics}, @@ -297,7 +296,6 @@ pub async fn do_oracledb( query: &str, db: &sqlx::Pool, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, column_order: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, @@ -406,7 +404,6 @@ pub async fn do_oracledb( job.timeout, db, mem_peak, - canceled_by, result_f, worker_name, &job.workspace_id, diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 2ad6eb4038..015f12bbc7 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -32,7 +32,6 @@ use windmill_parser::{Arg, Typ}; use windmill_parser_sql::{ parse_db_resource, parse_pg_statement_arg_indices, parse_pgsql_sig, parse_sql_blocks, }; -use windmill_queue::CanceledBy; use crate::common::{build_args_values, sizeof_val, OccupancyMetrics}; use crate::handle_child::run_future_with_polling_update_job_poller; @@ -161,7 +160,6 @@ pub async fn do_postgresql( query: &str, db: &sqlx::Pool, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, column_order: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, @@ -350,7 +348,6 @@ pub async fn do_postgresql( job.timeout, db, mem_peak, - canceled_by, result_f, worker_name, &job.workspace_id, diff --git a/backend/windmill-worker/src/php_executor.rs b/backend/windmill-worker/src/php_executor.rs index 4f579a54a8..5211a36647 100644 --- a/backend/windmill-worker/src/php_executor.rs +++ b/backend/windmill-worker/src/php_executor.rs @@ -11,7 +11,7 @@ use windmill_common::{ worker::write_file, }; use windmill_parser::Typ; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; use crate::{ common::{ @@ -63,7 +63,6 @@ pub fn parse_php_imports(code: &str) -> anyhow::Result> { pub async fn composer_install( mem_peak: &mut i32, - canceled_by: &mut Option, job_id: &Uuid, w_id: &str, db: &sqlx::Pool, @@ -95,7 +94,6 @@ pub async fn composer_install( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -135,7 +133,6 @@ $args->{arg_name} = new {rt_name}($args->{arg_name});" pub async fn handle_php_job( requirements_o: Option<&String>, mem_peak: &mut i32, - canceled_by: &mut Option, job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -168,7 +165,6 @@ pub async fn handle_php_job( composer_install( mem_peak, - canceled_by, &job.id, &job.workspace_id, db, @@ -328,7 +324,6 @@ try {{ &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index d86c84dd0d..242c0950c3 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -36,7 +36,7 @@ use windmill_common::{ use windmill_common::variables::get_secret_value_as_admin; use std::env::var; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; lazy_static::lazy_static! { static ref PYTHON_PATH: String = @@ -335,7 +335,6 @@ impl PyVersion { job_id, db, mem_peak, - &mut None, child_process, false, worker_name, @@ -442,7 +441,6 @@ pub async fn uv_pip_compile( job_id: &Uuid, requirements: &str, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &Pool, worker_name: &str, @@ -587,7 +585,6 @@ pub async fn uv_pip_compile( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -698,7 +695,6 @@ pub async fn uv_pip_compile( job_id, db, mem_peak, - canceled_by, child_process, false, worker_name, @@ -876,7 +872,6 @@ pub async fn handle_python_job( worker_name: &str, job: &QueuedJob, mem_peak: &mut i32, - canceled_by: &mut Option, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, inner_content: &String, @@ -899,7 +894,6 @@ pub async fn handle_python_job( worker_name, worker_dir, mem_peak, - canceled_by, &mut Some(occupancy_metrics), ) .await?; @@ -1201,7 +1195,6 @@ mount {{ &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, @@ -1473,7 +1466,6 @@ async fn handle_python_deps( worker_name: &str, worker_dir: &str, mem_peak: &mut i32, - canceled_by: &mut Option, occupancy_metrics: &mut Option<&mut OccupancyMetrics>, ) -> error::Result<(PyVersion, Vec)> { create_dependencies_dir(job_dir).await; @@ -1515,7 +1507,6 @@ async fn handle_python_deps( job_id, &requirements, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1578,7 +1569,6 @@ async fn handle_python_deps( job_id, w_id, mem_peak, - canceled_by, db, worker_name, job_dir, @@ -1844,7 +1834,6 @@ pub async fn handle_python_reqs( job_id: &Uuid, w_id: &str, mem_peak: &mut i32, - _canceled_by: &mut Option, db: &sqlx::Pool, _worker_name: &str, job_dir: &str, @@ -2433,7 +2422,6 @@ pub async fn start_worker( killpill_rx: tokio::sync::broadcast::Receiver<()>, ) -> error::Result<()> { let mut mem_peak: i32 = 0; - let mut canceled_by: Option = None; let context = variables::get_reserved_variables( db, w_id, @@ -2466,7 +2454,6 @@ pub async fn start_worker( worker_name, job_dir, &mut mem_peak, - &mut canceled_by, &mut None, ) .await?; diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 1b7ec63a1c..d3b9b8b780 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -28,7 +28,7 @@ use windmill_common::{ #[cfg(feature = "benchmark")] use crate::bench::{BenchmarkInfo, BenchmarkIter}; -use windmill_queue::{append_logs, get_queued_job, CanceledBy, WrappedError}; +use windmill_queue::{append_logs, get_queued_job, WrappedError}; use serde_json::{json, value::RawValue}; @@ -234,7 +234,6 @@ async fn send_job_completed( result: Arc>, result_columns: Option>, mem_peak: i32, - canceled_by: Option, success: bool, cached_res_path: Option, token: String, @@ -245,12 +244,7 @@ async fn send_job_completed( result, result_columns, mem_peak, - canceled_by, - success, - cached_res_path, - token, - duration, - }; + success, cached_res_path, token, duration }; job_completed_tx .send(jc) .with_context(windmill_common::otel_ee::otel_ctx()) @@ -264,7 +258,6 @@ pub async fn process_result( job_dir: &str, job_completed_tx: JobCompletedSender, mem_peak: i32, - canceled_by: Option, cached_res_path: Option, token: String, column_order: Option>, @@ -291,7 +284,6 @@ pub async fn process_result( r, column_order, mem_peak, - canceled_by, true, cached_res_path, token, @@ -337,7 +329,6 @@ pub async fn process_result( Arc::new(to_raw_value(&error_value)), None, mem_peak, - canceled_by, false, cached_res_path, token, @@ -370,7 +361,6 @@ pub async fn handle_receive_completed_job( }; let job = jc.job.clone(); let mem_peak = jc.mem_peak.clone(); - let canceled_by = jc.canceled_by.clone(); match process_completed_job( jc, &client, @@ -390,7 +380,6 @@ pub async fn handle_receive_completed_job( &client, job.as_ref(), mem_peak, - canceled_by, err, false, same_worker_tx.clone(), @@ -414,7 +403,7 @@ pub async fn process_completed_job( mem_peak, success, cached_res_path, - canceled_by, + duration, result_columns, .. @@ -465,7 +454,6 @@ pub async fn process_completed_job( Json(&result), result_columns, mem_peak.to_owned(), - canceled_by, false, duration, ) @@ -505,13 +493,11 @@ pub async fn process_completed_job( db, &job, mem_peak.to_owned(), - canceled_by, serde_json::from_str(result.get()).unwrap_or_else( |_| json!({ "message": format!("Non serializable error: {}", result.get()) }), ), worker_name, false, - None, ) .await?; if job.is_flow_step { @@ -549,7 +535,6 @@ pub async fn handle_job_error( client: &AuthedClient, job: &QueuedJob, mem_peak: i32, - canceled_by: Option, err: Error, unrecoverable: bool, same_worker_tx: SameWorkerSender, @@ -571,17 +556,7 @@ pub async fn handle_job_error( db, ) .await; - add_completed_job_error( - db, - job, - mem_peak, - canceled_by.clone(), - err.clone(), - worker_name, - false, - None, - ) - .await + add_completed_job_error(db, job, mem_peak, err.clone(), worker_name, false).await }; let update_job_future = if job.is_flow_step || job.is_flow() { @@ -631,17 +606,9 @@ pub async fn handle_job_error( db, ) .await; - let _ = add_completed_job_error( - db, - &parent_job, - mem_peak, - canceled_by.clone(), - e, - worker_name, - false, - None, - ) - .await; + let _ = + add_completed_job_error(db, &parent_job, mem_peak, e, worker_name, false) + .await; } } } diff --git a/backend/windmill-worker/src/rust_executor.rs b/backend/windmill-worker/src/rust_executor.rs index 639adc709a..d2956b927b 100644 --- a/backend/windmill-worker/src/rust_executor.rs +++ b/backend/windmill-worker/src/rust_executor.rs @@ -11,7 +11,7 @@ use windmill_common::{ utils::calculate_hash, worker::{save_cache, write_file}, }; -use windmill_queue::{append_logs, CanceledBy}; +use windmill_queue::append_logs; use crate::{ common::{ @@ -125,7 +125,6 @@ pub async fn generate_cargo_lockfile( job_id: &Uuid, code: &str, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -155,7 +154,6 @@ pub async fn generate_cargo_lockfile( job_id, db, mem_peak, - canceled_by, gen_lockfile_process, false, worker_name, @@ -177,7 +175,6 @@ pub async fn generate_cargo_lockfile( pub async fn build_rust_crate( job_id: &Uuid, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -217,7 +214,6 @@ pub async fn build_rust_crate( job_id, db, mem_peak, - canceled_by, build_rust_process, false, worker_name, @@ -274,7 +270,6 @@ pub fn compute_rust_hash(code: &str, requirements_o: Option<&String>) -> String #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_rust_job( mem_peak: &mut i32, - canceled_by: &mut Option, job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClientBackgroundTask, @@ -328,7 +323,6 @@ pub async fn handle_rust_job( build_rust_crate( &job.id, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -396,7 +390,6 @@ pub async fn handle_rust_job( &job.id, db, mem_peak, - canceled_by, child, !*DISABLE_NSJAIL, worker_name, diff --git a/backend/windmill-worker/src/snowflake_executor.rs b/backend/windmill-worker/src/snowflake_executor.rs index 3315df6bc1..a5544c9246 100644 --- a/backend/windmill-worker/src/snowflake_executor.rs +++ b/backend/windmill-worker/src/snowflake_executor.rs @@ -13,7 +13,7 @@ use windmill_common::error::to_anyhow; use windmill_common::jobs::QueuedJob; use windmill_common::{error::Error, worker::to_raw_value}; use windmill_parser_sql::{parse_db_resource, parse_snowflake_sig, parse_sql_blocks}; -use windmill_queue::{CanceledBy, HTTP_CLIENT}; +use windmill_queue::HTTP_CLIENT; use serde::{Deserialize, Serialize}; @@ -245,7 +245,6 @@ pub async fn do_snowflake( query: &str, db: &sqlx::Pool, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, column_order: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, @@ -422,7 +421,6 @@ pub async fn do_snowflake( job.timeout, db, mem_peak, - canceled_by, result_f.map_err(to_anyhow), worker_name, &job.workspace_id, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 90b0444901..c1d77e9a7b 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -63,7 +63,7 @@ use windmill_common::{ }; use windmill_queue::{ - append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy, PulledJob, PushArgs, + append_logs, canceled_job_to_result, empty_result, pull, push, PulledJob, PushArgs, PushIsolationLevel, HTTP_CLIENT, }; @@ -1504,7 +1504,6 @@ pub async fn run_worker( mem_peak: 0, cached_res_path: None, token: "".to_string(), - canceled_by: None, duration: None, }) .await @@ -1686,7 +1685,6 @@ pub async fn run_worker( &authed_client.get_authed().await, arc_job.as_ref(), 0, - None, err, false, same_worker_tx.clone(), @@ -1902,7 +1900,6 @@ pub struct JobCompleted { pub success: bool, pub cached_res_path: Option, pub token: String, - pub canceled_by: Option, pub duration: Option, } @@ -1913,7 +1910,6 @@ async fn do_nativets( code: String, db: &Pool, mem_peak: &mut i32, - canceled_by: &mut Option, worker_name: &str, occupancy_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { @@ -1933,7 +1929,6 @@ async fn do_nativets( job.timeout, db, mem_peak, - canceled_by, worker_name, &job.workspace_id, true, @@ -2089,7 +2084,6 @@ async fn handle_queued_job( result, result_columns: None, mem_peak: 0, - canceled_by: None, success: true, cached_res_path: None, token: authed_client.token, @@ -2124,7 +2118,6 @@ async fn handle_queued_job( } else { let mut logs = "".to_string(); let mut mem_peak: i32 = 0; - let mut canceled_by: Option = None; // println!("handle queue {:?}", SystemTime::now()); logs.push_str(&format!( @@ -2166,7 +2159,6 @@ async fn handle_queued_job( &job, preview_data.as_ref(), &mut mem_peak, - &mut canceled_by, job_dir, db, worker_name, @@ -2182,7 +2174,6 @@ async fn handle_queued_job( &job, preview_data.as_ref(), &mut mem_peak, - &mut canceled_by, job_dir, db, worker_name, @@ -2196,7 +2187,6 @@ async fn handle_queued_job( JobKind::AppDependencies => handle_app_dependency_job( &job, &mut mem_peak, - &mut canceled_by, job_dir, db, worker_name, @@ -2228,7 +2218,6 @@ async fn handle_queued_job( job_dir, worker_dir, &mut mem_peak, - &mut canceled_by, base_internal_url, worker_name, &mut column_order, @@ -2261,7 +2250,6 @@ async fn handle_queued_job( job_dir, job_completed_tx, mem_peak, - canceled_by, cached_res_path, client.get_token().await, column_order, @@ -2354,7 +2342,6 @@ async fn handle_code_execution_job( job_dir: &str, #[allow(unused_variables)] worker_dir: &str, mem_peak: &mut i32, - canceled_by: &mut Option, base_internal_url: &str, worker_name: &str, column_order: &mut Option>, @@ -2447,7 +2434,6 @@ async fn handle_code_execution_job( &code, db, mem_peak, - canceled_by, worker_name, column_order, occupancy_metrics, @@ -2466,7 +2452,6 @@ async fn handle_code_execution_job( &code, db, mem_peak, - canceled_by, worker_name, column_order, occupancy_metrics, @@ -2496,7 +2481,6 @@ async fn handle_code_execution_job( &code, db, mem_peak, - canceled_by, worker_name, column_order, occupancy_metrics, @@ -2519,7 +2503,6 @@ async fn handle_code_execution_job( &code, db, mem_peak, - canceled_by, worker_name, column_order, occupancy_metrics, @@ -2550,7 +2533,6 @@ async fn handle_code_execution_job( &code, db, mem_peak, - canceled_by, worker_name, occupancy_metrics, ) @@ -2580,7 +2562,6 @@ async fn handle_code_execution_job( &code, db, mem_peak, - canceled_by, worker_name, column_order, occupancy_metrics, @@ -2594,7 +2575,6 @@ async fn handle_code_execution_job( &code, db, mem_peak, - canceled_by, worker_name, occupancy_metrics, ) @@ -2625,7 +2605,6 @@ async fn handle_code_execution_job( code.clone(), db, mem_peak, - canceled_by, worker_name, occupancy_metrics, ) @@ -2690,7 +2669,6 @@ mount {{ worker_name, job, mem_peak, - canceled_by, db, client, &code, @@ -2706,7 +2684,6 @@ mount {{ handle_deno_job( lock.as_ref(), mem_peak, - canceled_by, job, db, client, @@ -2725,7 +2702,6 @@ mount {{ lock.as_ref(), codebase.as_ref(), mem_peak, - canceled_by, job, db, client, @@ -2743,7 +2719,6 @@ mount {{ Some(ScriptLang::Go) => { handle_go_job( mem_peak, - canceled_by, job, db, client, @@ -2761,7 +2736,6 @@ mount {{ Some(ScriptLang::Bash) => { handle_bash_job( mem_peak, - canceled_by, job, db, client, @@ -2779,7 +2753,6 @@ mount {{ Some(ScriptLang::Powershell) => { handle_powershell_job( mem_peak, - canceled_by, job, db, client, @@ -2803,7 +2776,6 @@ mount {{ handle_php_job( lock.as_ref(), mem_peak, - canceled_by, job, db, client, @@ -2826,7 +2798,6 @@ mount {{ #[cfg(feature = "rust")] handle_rust_job( mem_peak, - canceled_by, job, db, client, @@ -2855,7 +2826,6 @@ mount {{ worker_name, job, mem_peak, - canceled_by, db, client, &code, @@ -2869,7 +2839,6 @@ mount {{ Some(ScriptLang::CSharp) => { handle_csharp_job( mem_peak, - canceled_by, job, db, client, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 42dd66c217..dbd632a392 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -56,8 +56,8 @@ use windmill_common::{ }; use windmill_queue::schedule::get_schedule_opt; use windmill_queue::{ - add_completed_job, add_completed_job_error, append_logs, handle_maybe_scheduled_job, - CanceledBy, PushArgs, PushIsolationLevel, WrappedError, + add_completed_job, add_completed_job_error, append_logs, handle_maybe_scheduled_job, PushArgs, + PushIsolationLevel, WrappedError, }; type DB = sqlx::Pool; @@ -1065,14 +1065,9 @@ pub async fn update_flow_status_after_job_completion_internal( db, &flow_job, 0, - Some(CanceledBy { - username: flow_job.canceled_by.clone(), - reason: flow_job.canceled_reason.clone(), - }), canceled_job_to_result(&flow_job), worker_name, true, - None, ) .await?; } else { @@ -1100,7 +1095,6 @@ pub async fn update_flow_status_after_job_completion_internal( Json(&nresult), None, 0, - None, true, None, ) @@ -1118,7 +1112,6 @@ pub async fn update_flow_status_after_job_completion_internal( ), None, 0, - None, true, None, ) @@ -1151,8 +1144,7 @@ pub async fn update_flow_status_after_job_completion_internal( db, ) .await; - let _ = add_completed_job_error(db, &flow_job, 0, None, e, worker_name, true, None) - .await; + let _ = add_completed_job_error(db, &flow_job, 0, e, worker_name, true).await; true } Ok(_) => false, diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 3888d8eb63..90642d689b 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -30,7 +30,7 @@ use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; #[cfg(feature = "python")] use windmill_parser_py_imports::parse_relative_imports; use windmill_parser_ts::parse_expr_for_imports; -use windmill_queue::{append_logs, CanceledBy, PushIsolationLevel}; +use windmill_queue::{append_logs, PushIsolationLevel}; use crate::common::OccupancyMetrics; use crate::csharp_executor::generate_nuget_lockfile; @@ -221,7 +221,6 @@ pub async fn handle_dependency_job( job: &QueuedJob, preview_data: Option<&RawData>, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -277,7 +276,6 @@ pub async fn handle_dependency_job( })?, &script_data.code, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -541,7 +539,6 @@ pub async fn handle_flow_dependency_job( job: &QueuedJob, preview_data: Option<&RawData>, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -617,7 +614,6 @@ pub async fn handle_flow_dependency_job( flow.modules, job, mem_peak, - canceled_by, job_dir, db, tx, @@ -743,7 +739,6 @@ async fn lock_modules<'c>( modules: Vec, job: &QueuedJob, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, mut tx: sqlx::Transaction<'c, sqlx::Postgres>, @@ -791,7 +786,6 @@ async fn lock_modules<'c>( modules, job, mem_peak, - canceled_by, job_dir, db, tx, @@ -824,7 +818,6 @@ async fn lock_modules<'c>( b.modules, job, mem_peak, - canceled_by, job_dir, db, tx, @@ -849,7 +842,6 @@ async fn lock_modules<'c>( modules, job, mem_peak, - canceled_by, job_dir, db, tx, @@ -880,7 +872,6 @@ async fn lock_modules<'c>( b.modules, job, mem_peak, - canceled_by, job_dir, db, tx, @@ -902,7 +893,6 @@ async fn lock_modules<'c>( default, job, mem_peak, - canceled_by, job_dir, db, tx, @@ -951,7 +941,6 @@ async fn lock_modules<'c>( &language, &content, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1321,7 +1310,6 @@ async fn lock_modules_app( value: Value, job: &QueuedJob, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -1363,7 +1351,6 @@ async fn lock_modules_app( &language, &content, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1424,7 +1411,6 @@ async fn lock_modules_app( b, job, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1447,7 +1433,6 @@ async fn lock_modules_app( b, job, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1469,7 +1454,6 @@ async fn lock_modules_app( pub async fn handle_app_dependency_job( job: &QueuedJob, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -1499,7 +1483,6 @@ pub async fn handle_app_dependency_job( value, job, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1593,7 +1576,6 @@ async fn python_dep( reqs: String, job_id: &Uuid, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -1626,7 +1608,6 @@ async fn python_dep( job_id, &reqs, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1644,7 +1625,6 @@ async fn python_dep( job_id, w_id, mem_peak, - canceled_by, db, worker_name, job_dir, @@ -1670,7 +1650,6 @@ async fn capture_dependency_job( job_language: &ScriptLang, job_raw_code: &str, mem_peak: &mut i32, - canceled_by: &mut Option, job_dir: &str, db: &sqlx::Pool, worker_name: &str, @@ -1736,7 +1715,6 @@ async fn capture_dependency_job( reqs, job_id, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1786,7 +1764,6 @@ async fn capture_dependency_job( reqs, job_id, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1811,7 +1788,6 @@ async fn capture_dependency_job( job_id, job_raw_code, mem_peak, - canceled_by, job_dir, db, false, @@ -1833,7 +1809,6 @@ async fn capture_dependency_job( job_id, job_raw_code, mem_peak, - canceled_by, job_dir, Some(db), w_id, @@ -1852,7 +1827,6 @@ async fn capture_dependency_job( } let req = gen_bun_lockfile( mem_peak, - canceled_by, job_id, w_id, Some(db), @@ -1912,7 +1886,6 @@ async fn capture_dependency_job( }; composer_install( mem_peak, - canceled_by, job_id, w_id, db, @@ -1942,7 +1915,6 @@ async fn capture_dependency_job( job_id, job_raw_code, mem_peak, - canceled_by, job_dir, db, worker_name, @@ -1965,7 +1937,6 @@ async fn capture_dependency_job( job_id, job_raw_code, mem_peak, - canceled_by, job_dir, db, worker_name,