worker: move ping to root execution

The cancellation informations where moved from bottom to top where the
infromations are already set within the queue table.

backend/.sqlx/query-c6423de98ffedc4e58cca0bd7883ef87f04a3daa054882a0c4e81c723196dbb6.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE worker_ping SET\n                                          ping_at = now()\n                                        , current_job_id = $1\n                                        , current_job_workspace_id = $2\n                                        , memory_usage = $3\n                                        , wm_memory_usage = $4\n                                        , occupancy_rate = $6\n                                        , occupancy_rate_15s = $7\n                                        , occupancy_rate_5m = $8\n                                        , occupancy_rate_30m = $9\n                                    WHERE worker = $5",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Varchar",
+        "Int8",
+        "Int8",
+        "Text",
+        "Float4",
+        "Float4",
+        "Float4",
+        "Float4"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c6423de98ffedc4e58cca0bd7883ef87f04a3daa054882a0c4e81c723196dbb6"
+}

backend/.sqlx/query-db0b7acb2beeb6bbd22af959fbd6d889dd2b7d226d44c92abbc9d1761c2b230e.json

@@ -0,0 +1,35 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            INSERT INTO completed_job\n                ( workspace_id\n                , id\n                , parent_job\n                , created_by\n                , created_at\n                , started_at\n                , duration_ms\n                , success\n                , script_hash\n                , script_path\n                , args\n                , result\n                , raw_code\n                , raw_lock\n                , canceled\n                , canceled_by\n                , canceled_reason\n                , job_kind\n                , schedule_path\n                , permissioned_as\n                , flow_status\n                , raw_flow\n                , is_flow_step\n                , is_skipped\n                , language\n                , email\n                , visible_to_owner\n                , mem_peak\n                , tag\n                , priority\n                )\n            SELECT \n                  queue.workspace_id\n                , queue.id\n                , queue.parent_job\n                , queue.created_by\n                , queue.created_at\n                , queue.started_at\n                , COALESCE($2::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(queue.started_at, now())))) * 1000)\n                , $3 AS success\n                , queue.script_hash\n                , queue.script_path\n                , queue.args\n                , $4 AS result\n                , queue.raw_code\n                , queue.raw_lock\n                , queue.canceled OR queue.canceled_by IS NOT NULL as canceled\n                , COALESCE($7, queue.canceled_by) as canceled_by\n                , COALESCE($8, queue.canceled_reason) as canceled_reason\n                , queue.job_kind\n                , queue.schedule_path\n                , queue.permissioned_as\n                , queue.flow_status\n                , queue.raw_flow\n                , queue.is_flow_step\n                , $5 AS is_skipped\n                , queue.language\n                , queue.email\n                , queue.visible_to_owner\n                , CASE WHEN $6 > 0 THEN $6 ELSE NULL END AS mem_peak\n                , queue.tag\n                , queue.priority\n            FROM queue\n            WHERE queue.id = $1\n            LIMIT 1\n            ON CONFLICT (id) DO UPDATE SET \n                success = $3, \n                result = $4\n            RETURNING duration_ms, canceled\n            ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "duration_ms",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "canceled",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Int8",
+        "Bool",
+        "Jsonb",
+        "Bool",
+        "Int4",
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "db0b7acb2beeb6bbd22af959fbd6d889dd2b7d226d44c92abbc9d1761c2b230e"
+}

backend/src/main.rs

@@ -154,13 +154,11 @@ async fn cache_hub_scripts(file_path: Option<String>) -> anyhow::Result<()> {
                 &Uuid::nil(),
                 &res.content,
                 &mut 0,
-                &mut None,
                 &job_dir,
                 None,
                 "global",
                 "global",
                 "",
-                &mut None,
             )
             .await?;
             tokio::fs::remove_dir_all(job_dir).await?;
@@ -173,7 +171,6 @@ async fn cache_hub_scripts(file_path: Option<String>) -> 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,
@@ -181,7 +178,6 @@ async fn cache_hub_scripts(file_path: Option<String>) -> anyhow::Result<()> {
                     "cache_init",
                     envs.clone(),
                     false,
-                    &mut None,
                 )
                 .await?;
 
@@ -198,7 +194,6 @@ async fn cache_hub_scripts(file_path: Option<String>) -> anyhow::Result<()> {
                     "",
                     "cache_init",
                     "",
-                    &mut None,
                 )
                 .await
                 {

backend/src/monitor.rs

@@ -1592,7 +1592,6 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, 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: {})",
                 last_ping

backend/windmill-queue/src/jobs.rs

@@ -521,7 +521,9 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
     }
 
     let _job_id = queued_job.id;
-    let (opt_uuid, _duration, _skip_downstream_error_handlers) = (|| async {
+    let (canceled_by, canceled_reason) =
+        canceled_by.map_or((None, None), |c| (c.username, c.reason));
+    let (opt_uuid, _duration, canceled, _skip_downstream_error_handlers) = (|| async {
         let mut tx = db.begin().await?;
 
         let job_id = queued_job.id;
@@ -533,103 +535,94 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
             serde_json::to_string(&result).unwrap_or_else(|_| "".to_string())
         );
 
-        let (raw_code, raw_lock, raw_flow) = if !*MIN_VERSION_IS_AT_LEAST_1_427.read().await {
-            sqlx::query!(
-                "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json<Box<JsonRawValue>>\"
-            FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
-                &job_id,
-                &queued_job.workspace_id
-            )
-            .fetch_one(db)
-            .map_ok(|record| (record.raw_code, record.raw_lock, record.raw_flow))
-            .or_else(|_| {
-                sqlx::query!(
-                    "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json<Box<JsonRawValue>>\"
-                FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1",
-                    &job_id,
-                    &queued_job.workspace_id
-                )
-                .fetch_one(db)
-                .map_ok(|record| (record.raw_code, record.raw_lock, record.raw_flow))
-            })
-            .await
-            .unwrap_or_default()
-        } else {
-            (None, None, None)
-        };
-
         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 completed_job AS cj
-                    ( workspace_id
-                    , id
-                    , parent_job
-                    , created_by
-                    , created_at
-                    , started_at
-                    , duration_ms
-                    , success
-                    , script_hash
-                    , script_path
-                    , args
-                    , result
-                    , raw_code
-                    , raw_lock
-                    , canceled
-                    , canceled_by
-                    , canceled_reason
-                    , job_kind
-                    , schedule_path
-                    , permissioned_as
-                    , flow_status
-                    , raw_flow
-                    , is_flow_step
-                    , is_skipped
-                    , language
-                    , email
-                    , visible_to_owner
-                    , mem_peak
-                    , tag
-                    , priority
-                    )
-                VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($30::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,\
-                        $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)
-            ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms",
-            queued_job.workspace_id,
-            queued_job.id,
-            queued_job.parent_job,
-            queued_job.created_by,
-            queued_job.created_at,
-            queued_job.started_at,
-            success,
-            queued_job.script_hash.map(|x| x.0),
-            queued_job.script_path,
-            &queued_job.args as &Option<Json<HashMap<String, Box<RawValue>>>>,
-            result as Json<&T>,
-            raw_code,
-            raw_lock,
-            canceled_by.is_some(),
-            canceled_by.clone().map(|cb| cb.username).flatten(),
-            canceled_by.clone().map(|cb| cb.reason).flatten(),
-            queued_job.job_kind.clone() as JobKind,
-            queued_job.schedule_path,
-            queued_job.permissioned_as,
-            &queued_job.flow_status as &Option<Json<Box<RawValue>>>,
-            &raw_flow as &Option<Json<Box<RawValue>>>,
-            queued_job.is_flow_step,
-            skipped,
-            queued_job.language.clone() as Option<ScriptLang>,
-            queued_job.email,
-            queued_job.visible_to_owner,
-            if mem_peak > 0 { Some(mem_peak) } else { None },
-            queued_job.tag,
-            queued_job.priority,
-            duration,
+        let (_duration, canceled) =  sqlx::query!(
+            r#"
+            INSERT INTO completed_job
+                ( workspace_id
+                , id
+                , parent_job
+                , created_by
+                , created_at
+                , started_at
+                , duration_ms
+                , success
+                , script_hash
+                , script_path
+                , args
+                , result
+                , raw_code
+                , raw_lock
+                , canceled
+                , canceled_by
+                , canceled_reason
+                , job_kind
+                , schedule_path
+                , permissioned_as
+                , flow_status
+                , raw_flow
+                , is_flow_step
+                , is_skipped
+                , language
+                , email
+                , visible_to_owner
+                , mem_peak
+                , tag
+                , priority
+                )
+            SELECT 
+                  queue.workspace_id
+                , queue.id
+                , queue.parent_job
+                , queue.created_by
+                , queue.created_at
+                , queue.started_at
+                , COALESCE($2::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(queue.started_at, now())))) * 1000)
+                , $3 AS success
+                , queue.script_hash
+                , queue.script_path
+                , queue.args
+                , $4 AS result
+                , queue.raw_code
+                , queue.raw_lock
+                , queue.canceled OR queue.canceled_by IS NOT NULL as canceled
+                , COALESCE($7, queue.canceled_by) as canceled_by
+                , COALESCE($8, queue.canceled_reason) as canceled_reason
+                , queue.job_kind
+                , queue.schedule_path
+                , queue.permissioned_as
+                , queue.flow_status
+                , queue.raw_flow
+                , queue.is_flow_step
+                , $5 AS is_skipped
+                , queue.language
+                , queue.email
+                , queue.visible_to_owner
+                , CASE WHEN $6 > 0 THEN $6 ELSE NULL END AS mem_peak
+                , queue.tag
+                , queue.priority
+            FROM queue
+            WHERE queue.id = $1
+            LIMIT 1
+            ON CONFLICT (id) DO UPDATE SET 
+                success = $3, 
+                result = $4
+            RETURNING duration_ms, canceled
+            "#,
+            /* $1 */ queued_job.id,
+            /* $2 */ duration,
+            /* $3 */ success,
+            /* $4 */ result as Json<&T>,
+            /* $5 */ skipped,
+            /* $6 */ mem_peak,
+            /* $7 */ canceled_by,
+            /* $8 */ canceled_reason,
         )
         .fetch_one(&mut *tx)
         .await
+        .map(|record| (record.duration_ms, record.canceled))
         .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?;
 
 
@@ -735,7 +728,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
                             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), 0, canceled, true)),
                             }
                         };
                     }
@@ -846,7 +839,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
             "inserted completed job: {} (success: {success})",
             queued_job.id
         );
-        Ok((None, _duration, _skip_downstream_error_handlers)) as windmill_common::error::Result<(Option<Uuid>, i64, bool)>
+        Ok((None, _duration, canceled, _skip_downstream_error_handlers)) as error::Result<(Option<Uuid>, i64, bool, bool)>
     })
     .retry(
         ConstantBuilder::default()
@@ -978,13 +971,8 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
                 );
             }
 
-            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(_) => {}
@@ -1003,7 +991,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
         }
     }
 
-    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",

backend/windmill-worker/src/ansible_executor.rs

@@ -1,10 +1,5 @@
 #[cfg(unix)]
-use std::{
-    collections::HashMap,
-    os::unix::fs::PermissionsExt,
-    path::PathBuf,
-    process::Stdio,
-};
+use std::{collections::HashMap, os::unix::fs::PermissionsExt, path::PathBuf, process::Stdio};
 
 #[cfg(windows)]
 use std::{
@@ -24,12 +19,13 @@ 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,
     common::{
-        check_executor_binary_exists, get_reserved_variables, read_and_check_result, start_child_process, transform_json, OccupancyMetrics
+        check_executor_binary_exists, get_reserved_variables, read_and_check_result,
+        start_child_process, transform_json,
     },
     handle_child::handle_child,
     python_executor::{create_dependencies_dir, handle_python_reqs, uv_pip_compile},
@@ -57,8 +53,6 @@ async fn handle_ansible_python_deps(
     worker_name: &str,
     worker_dir: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<Vec<String>> {
     create_dependencies_dir(job_dir).await;
 
@@ -82,12 +76,10 @@ async fn handle_ansible_python_deps(
                     job_id,
                     &requirements,
                     mem_peak,
-                    canceled_by,
                     job_dir,
                     db,
                     worker_name,
                     w_id,
-                    &mut Some(occupancy_metrics),
                     false,
                     false,
                 )
@@ -109,12 +101,10 @@ async fn handle_ansible_python_deps(
             job_id,
             w_id,
             mem_peak,
-            canceled_by,
             db,
             worker_name,
             job_dir,
             worker_dir,
-            &mut Some(occupancy_metrics),
             false,
             false,
         )
@@ -131,9 +121,7 @@ async fn install_galaxy_collections(
     worker_name: &str,
     w_id: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     db: &sqlx::Pool<sqlx::Postgres>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> anyhow::Result<()> {
     write_file(job_dir, "requirements.yml", collections_yml)?;
 
@@ -169,7 +157,6 @@ async fn install_galaxy_collections(
         job_id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -177,7 +164,6 @@ async fn install_galaxy_collections(
         "ansible galaxy install",
         None,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 
@@ -191,16 +177,18 @@ pub async fn handle_ansible_job(
     worker_name: &str,
     job: &QueuedJob,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
     inner_content: &String,
     shared_mount: &str,
     base_internal_url: &str,
     envs: HashMap<String, String>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> windmill_common::error::Result<Box<RawValue>> {
-    check_executor_binary_exists("ansible-playbook", ANSIBLE_PLAYBOOK_PATH.as_str(), "ansible")?;
+    check_executor_binary_exists(
+        "ansible-playbook",
+        ANSIBLE_PLAYBOOK_PATH.as_str(),
+        "ansible",
+    )?;
 
     let (logs, reqs, playbook) = windmill_parser_yaml::parse_ansible_reqs(inner_content)?;
     append_logs(&job.id, &job.workspace_id, logs, db).await;
@@ -216,8 +204,6 @@ pub async fn handle_ansible_job(
         worker_name,
         worker_dir,
         mem_peak,
-        canceled_by,
-        occupancy_metrics,
     )
     .await?;
 
@@ -289,9 +275,7 @@ pub async fn handle_ansible_job(
                 worker_name,
                 &job.workspace_id,
                 mem_peak,
-                canceled_by,
                 db,
-                occupancy_metrics,
             )
             .await?;
         }
@@ -425,7 +409,6 @@ fi
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -433,7 +416,6 @@ fi
         "python run",
         job.timeout,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
     read_and_check_result(job_dir).await

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());
@@ -44,7 +44,6 @@ use crate::handle_child::run_future_with_polling_update_job_poller;
 use crate::{
     common::{
         build_args_map, get_reserved_variables, read_file, read_file_content, start_child_process,
-        OccupancyMetrics,
     },
     handle_child::handle_child,
     AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV,
@@ -62,7 +61,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<CanceledBy>,
     job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
@@ -72,7 +70,6 @@ pub async fn handle_bash_job(
     base_internal_url: &str,
     worker_name: &str,
     envs: HashMap<String, String>,
-    occupancy_metrics: &mut OccupancyMetrics,
     _killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
 ) -> Result<Box<RawValue>, Error> {
     let annotation = windmill_common::worker::BashAnnotations::parse(&content);
@@ -206,7 +203,6 @@ exit $exit_status
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -214,7 +210,6 @@ exit $exit_status
         "bash run",
         job.timeout,
         true,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 
@@ -226,9 +221,7 @@ exit $exit_status
             db,
             job.timeout,
             mem_peak,
-            canceled_by,
             worker_name,
-            occupancy_metrics,
             _killpill_rx,
         )
         .await;
@@ -271,9 +264,7 @@ async fn handle_docker_job(
     db: &DB,
     job_timeout: Option<i32>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
     killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
 ) -> Result<Box<RawValue>, Error> {
     let client = bollard::Docker::connect_with_unix_defaults().map_err(to_anyhow)?;
@@ -361,11 +352,9 @@ async fn handle_docker_job(
         job_timeout,
         db,
         mem_peak,
-        canceled_by,
         wait_f,
         worker_name,
         workspace_id,
-        &mut Some(occupancy_metrics),
         Box::pin(match mem_client {
             Ok(client) => client
                 .stats(
@@ -458,7 +447,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<CanceledBy>,
     job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
@@ -468,7 +456,6 @@ pub async fn handle_powershell_job(
     base_internal_url: &str,
     worker_name: &str,
     envs: HashMap<String, String>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> Result<Box<RawValue>, Error> {
     let pwsh_args = {
         let args = build_args_map(job, client, db).await?.map(Json);
@@ -548,7 +535,6 @@ pub async fn handle_powershell_job(
             &job.id,
             db,
             mem_peak,
-            canceled_by,
             child,
             false,
             worker_name,
@@ -556,7 +542,6 @@ pub async fn handle_powershell_job(
             "powershell install",
             job.timeout,
             false,
-            &mut Some(occupancy_metrics),
         )
         .await?;
     }
@@ -759,7 +744,6 @@ $env:PSModulePath = \"{};$PSModulePathBackup\"",
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -767,7 +751,6 @@ $env:PSModulePath = \"{};$PSModulePathBackup\"",
         "powershell run",
         job.timeout,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 

backend/windmill-worker/src/bigquery_executor.rs

@@ -9,7 +9,7 @@ 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, HTTP_CLIENT};
+use windmill_queue::HTTP_CLIENT;
 
 use serde::Deserialize;
 
@@ -207,10 +207,8 @@ pub async fn do_bigquery(
     query: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
     column_order: &mut Option<Vec<String>>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> windmill_common::error::Result<Box<RawValue>> {
     let bigquery_args = build_args_values(job, client, db).await?;
 
@@ -361,11 +359,9 @@ pub async fn do_bigquery(
         job.timeout,
         db,
         mem_peak,
-        canceled_by,
         result_f.map_err(to_anyhow),
         worker_name,
         &job.workspace_id,
-        &mut Some(occupancy_metrics),
         Box::pin(futures::stream::once(async { 0 })),
     )
     .await?;

backend/windmill-worker/src/bun_executor.rs

@@ -12,7 +12,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;
@@ -21,7 +21,6 @@ use crate::{
     common::{
         create_args_and_out_file, get_main_override, get_reserved_variables, parse_npm_config,
         read_file, read_file_content, read_result, start_child_process, write_file_binary,
-        OccupancyMetrics,
     },
     handle_child::handle_child,
     AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR,
@@ -85,7 +84,6 @@ fn split_lockfile(lockfile: &str) -> (&str, Option<&str>, bool) {
 
 pub async fn gen_bun_lockfile(
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_id: &Uuid,
     w_id: &str,
     db: Option<&sqlx::Pool<sqlx::Postgres>>,
@@ -97,7 +95,6 @@ pub async fn gen_bun_lockfile(
     export_pkg: bool,
     raw_deps: Option<String>,
     npm_mode: bool,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> Result<Option<String>> {
     let common_bun_proc_envs: HashMap<String, String> = get_common_bun_proc_envs(None).await;
 
@@ -149,7 +146,6 @@ pub async fn gen_bun_lockfile(
                 job_id,
                 db,
                 mem_peak,
-                canceled_by,
                 child_process,
                 false,
                 worker_name,
@@ -157,7 +153,6 @@ pub async fn gen_bun_lockfile(
                 "bun build",
                 None,
                 false,
-                occupancy_metrics,
             )
             .await?;
         } else {
@@ -174,7 +169,6 @@ pub async fn gen_bun_lockfile(
     if !empty_deps {
         install_bun_lockfile(
             mem_peak,
-            canceled_by,
             job_id,
             w_id,
             db,
@@ -182,7 +176,6 @@ pub async fn gen_bun_lockfile(
             worker_name,
             common_bun_proc_envs,
             npm_mode,
-            occupancy_metrics,
         )
         .await?;
     } else {
@@ -261,7 +254,6 @@ registry = {}
 
 pub async fn install_bun_lockfile(
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_id: &Uuid,
     w_id: &str,
     db: Option<&sqlx::Pool<sqlx::Postgres>>,
@@ -269,7 +261,6 @@ pub async fn install_bun_lockfile(
     worker_name: &str,
     common_bun_proc_envs: HashMap<String, String>,
     npm_mode: bool,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> Result<()> {
     let mut child_cmd = Command::new(if npm_mode { &*NPM_PATH } else { &*BUN_PATH });
     child_cmd
@@ -332,7 +323,6 @@ pub async fn install_bun_lockfile(
             job_id,
             db,
             mem_peak,
-            canceled_by,
             child_process,
             false,
             worker_name,
@@ -340,7 +330,6 @@ pub async fn install_bun_lockfile(
             "bun install",
             None,
             false,
-            occupancy_metrics,
         )
         .await?
     } else {
@@ -479,9 +468,7 @@ pub async fn generate_wrapper_mjs(
     db: &sqlx::Pool<sqlx::Postgres>,
     timeout: Option<i32>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     common_bun_proc_envs: &HashMap<String, String>,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> Result<()> {
     let mut child = Command::new(&*BUN_PATH);
     child
@@ -501,7 +488,6 @@ pub async fn generate_wrapper_mjs(
         job_id,
         db,
         mem_peak,
-        canceled_by,
         child_process,
         false,
         worker_name,
@@ -509,7 +495,6 @@ pub async fn generate_wrapper_mjs(
         "bun build",
         timeout,
         false,
-        occupancy_metrics,
     )
     .await?;
     fs::rename(
@@ -528,9 +513,7 @@ pub async fn generate_bun_bundle(
     db: Option<sqlx::Pool<sqlx::Postgres>>,
     timeout: Option<i32>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     common_bun_proc_envs: &HashMap<String, String>,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> Result<()> {
     let mut child = Command::new(&*BUN_PATH);
     child
@@ -551,7 +534,6 @@ pub async fn generate_bun_bundle(
             job_id,
             &db,
             mem_peak,
-            canceled_by,
             child_process,
             false,
             worker_name,
@@ -559,7 +541,6 @@ pub async fn generate_bun_bundle(
             "bun build",
             timeout,
             false,
-            occupancy_metrics,
         )
         .await?;
     } else {
@@ -674,7 +655,6 @@ pub async fn prebundle_bun_script(
     base_internal_url: &str,
     worker_name: &str,
     token: &str,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> Result<()> {
     let (local_path, remote_path) = compute_bundle_local_and_remote_path(
         inner_content,
@@ -720,9 +700,7 @@ pub async fn prebundle_bun_script(
         db.clone(),
         None,
         &mut 0,
-        &mut None,
         &common_bun_proc_envs,
-        occupancy_metrics,
     )
     .await?;
 
@@ -811,7 +789,6 @@ pub async fn handle_bun_job(
     requirements_o: Option<&String>,
     codebase: Option<&String>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
@@ -822,7 +799,6 @@ pub async fn handle_bun_job(
     envs: HashMap<String, String>,
     shared_mount: &str,
     new_args: &mut Option<HashMap<String, Box<RawValue>>>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<Box<RawValue>> {
     let mut annotation = windmill_common::worker::TypeScriptAnnotations::parse(inner_content);
 
@@ -939,7 +915,6 @@ pub async fn handle_bun_job(
             if !skip_install {
                 install_bun_lockfile(
                     mem_peak,
-                    canceled_by,
                     &job.id,
                     &job.workspace_id,
                     Some(db),
@@ -947,7 +922,6 @@ pub async fn handle_bun_job(
                     worker_name,
                     common_bun_proc_envs.clone(),
                     annotation.npm,
-                    &mut Some(occupancy_metrics),
                 )
                 .await?;
 
@@ -977,7 +951,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),
@@ -989,7 +962,6 @@ pub async fn handle_bun_job(
             false,
             None,
             annotation.npm,
-            &mut Some(occupancy_metrics),
         )
         .await?;
 
@@ -1223,9 +1195,7 @@ try {{
                 Some(db.clone()),
                 job.timeout,
                 mem_peak,
-                canceled_by,
                 &common_bun_proc_envs,
-                &mut Some(occupancy_metrics),
             )
             .await?;
             if !local_path.is_empty() {
@@ -1265,9 +1235,7 @@ try {{
                 db,
                 job.timeout,
                 mem_peak,
-                canceled_by,
                 &common_bun_proc_envs,
-                &mut Some(occupancy_metrics),
             )
             .await?;
         }
@@ -1312,11 +1280,9 @@ try {{
                 job.timeout,
                 db,
                 mem_peak,
-                canceled_by,
                 worker_name,
                 &job.workspace_id,
                 false,
-                occupancy_metrics,
             )
             .await?;
             tracing::info!(
@@ -1461,7 +1427,6 @@ try {{
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -1469,7 +1434,6 @@ try {{
         "bun run",
         job.timeout,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 
@@ -1553,7 +1517,6 @@ pub async fn start_worker(
 ) -> Result<()> {
     let mut logs = "".to_string();
     let mut mem_peak: i32 = 0;
-    let mut canceled_by: Option<CanceledBy> = None;
     tracing::info!("Starting worker {w_id};{script_path} (codebase: {codebase:?}");
     if !codebase.is_some() {
         let _ = write_file(job_dir, "main.ts", inner_content)?;
@@ -1612,7 +1575,6 @@ pub async fn start_worker(
 
             install_bun_lockfile(
                 &mut mem_peak,
-                &mut canceled_by,
                 &Uuid::nil(),
                 &w_id,
                 Some(db),
@@ -1620,7 +1582,6 @@ pub async fn start_worker(
                 worker_name,
                 common_bun_proc_envs.clone(),
                 annotation.npm,
-                &mut None,
             )
             .await?;
             tracing::info!("dedicated worker requirements installed: {reqs}");
@@ -1629,7 +1590,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),
@@ -1641,7 +1601,6 @@ pub async fn start_worker(
             false,
             None,
             annotation.npm,
-            &mut None,
         )
         .await?;
     }
@@ -1736,9 +1695,7 @@ for await (const line of Readline.createInterface({{ input: process.stdin }})) {
             db,
             None,
             &mut mem_peak,
-            &mut canceled_by,
             &common_bun_proc_envs,
-            &mut None,
         )
         .await?;
     }

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::{
@@ -36,7 +34,6 @@ use crate::{
     NUGET_CONFIG, PATH_ENV, TZ_ENV,
 };
 
-use crate::common::OccupancyMetrics;
 use crate::AuthedClientBackgroundTask;
 
 #[cfg(windows)]
@@ -60,12 +57,10 @@ pub async fn generate_nuget_lockfile(
     job_id: &Uuid,
     code: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
     w_id: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<String> {
     check_executor_binary_exists("dotnet", DOTNET_PATH.as_str(), "C#")?;
 
@@ -88,7 +83,6 @@ pub async fn generate_nuget_lockfile(
         job_id,
         db,
         mem_peak,
-        canceled_by,
         gen_lockfile_process,
         false,
         worker_name,
@@ -96,7 +90,6 @@ pub async fn generate_nuget_lockfile(
         "dotnet restore",
         None,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 
@@ -118,12 +111,10 @@ pub async fn generate_nuget_lockfile(
     _job_id: &Uuid,
     _code: &str,
     _mem_peak: &mut i32,
-    _canceled_by: &mut Option<CanceledBy>,
     _job_dir: &str,
     _db: &sqlx::Pool<sqlx::Postgres>,
     _worker_name: &str,
     _w_id: &str,
-    _occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<String> {
     Err(anyhow!("C# is not available because the feature is not enabled").into())
 }
@@ -261,14 +252,12 @@ namespace WindmillScriptCSharpInternal {{
 async fn build_cs_proj(
     job_id: &Uuid,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
     w_id: &str,
     base_internal_url: &str,
     hash: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<String> {
     if let Some(nuget_config) = NUGET_CONFIG.read().await.clone() {
         write_file(job_dir, "nuget.config", &nuget_config)?;
@@ -310,7 +299,6 @@ async fn build_cs_proj(
         job_id,
         db,
         mem_peak,
-        canceled_by,
         build_cs_process,
         false,
         worker_name,
@@ -318,7 +306,6 @@ async fn build_cs_proj(
         "dotnet publish",
         None,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
     append_logs(job_id, w_id, "\n\n", db).await;
@@ -363,7 +350,6 @@ fn remove_lines_from_text(contents: &str, indices_to_remove: Vec<usize>) -> Stri
 #[cfg(not(feature = "csharp"))]
 pub async fn handle_csharp_job(
     _mem_peak: &mut i32,
-    _canceled_by: &mut Option<CanceledBy>,
     _job: &QueuedJob,
     _db: &sqlx::Pool<sqlx::Postgres>,
     _client: &AuthedClientBackgroundTask,
@@ -374,7 +360,6 @@ pub async fn handle_csharp_job(
     _base_internal_url: &str,
     _worker_name: &str,
     _envs: HashMap<String, String>,
-    _occupancy_metrics: &mut OccupancyMetrics,
 ) -> Result<Box<RawValue>, Error> {
     Err(anyhow!("C# is not available because the feature is not enabled").into())
 }
@@ -382,7 +367,6 @@ pub async fn handle_csharp_job(
 #[cfg(feature = "csharp")]
 pub async fn handle_csharp_job(
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
@@ -393,7 +377,6 @@ pub async fn handle_csharp_job(
     base_internal_url: &str,
     worker_name: &str,
     envs: HashMap<String, String>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> Result<Box<RawValue>, Error> {
     check_executor_binary_exists("dotnet", DOTNET_PATH.as_str(), "C#")?;
 
@@ -452,14 +435,12 @@ pub async fn handle_csharp_job(
         build_cs_proj(
             &job.id,
             mem_peak,
-            canceled_by,
             job_dir,
             db,
             worker_name,
             &job.workspace_id,
             base_internal_url,
             &hash,
-            occupancy_metrics,
         )
         .await?
     };
@@ -523,7 +504,6 @@ pub async fn handle_csharp_job(
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -531,7 +511,6 @@ pub async fn handle_csharp_job(
         "csharp run",
         job.timeout,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
     read_result(job_dir).await

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, 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, 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, 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, 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:?}")}))),  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:?}")}))),  mem_peak: 0, success: false, cached_res_path: None, token: token.to_string(), duration: None }).await.unwrap();
                             },
                         };
                         logs = init_log.clone();

backend/windmill-worker/src/deno_executor.rs

@@ -3,12 +3,12 @@ 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::{
         create_args_and_out_file, get_main_override, get_reserved_variables, parse_npm_config,
-        read_file, read_result, start_child_process, OccupancyMetrics,
+        read_file, read_result, start_child_process,
     },
     handle_child::handle_child,
     AuthedClientBackgroundTask, DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV,
@@ -100,13 +100,11 @@ pub async fn generate_deno_lock(
     job_id: &Uuid,
     code: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: Option<&sqlx::Pool<sqlx::Postgres>>,
     w_id: &str,
     worker_name: &str,
     base_internal_url: &str,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> error::Result<String> {
     let _ = write_file(job_dir, "main.ts", code)?;
 
@@ -152,7 +150,6 @@ pub async fn generate_deno_lock(
             job_id,
             db,
             mem_peak,
-            canceled_by,
             child_process,
             false,
             worker_name,
@@ -160,7 +157,6 @@ pub async fn generate_deno_lock(
             "deno cache",
             None,
             false,
-            occupancy_metrics,
         )
         .await?;
     } else {
@@ -181,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<CanceledBy>,
     job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
@@ -191,7 +186,6 @@ pub async fn handle_deno_job(
     worker_name: &str,
     envs: HashMap<String, String>,
     new_args: &mut Option<HashMap<String, Box<RawValue>>>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<Box<RawValue>> {
     // let mut start = Instant::now();
     let logs1 = "\n\n--- DENO CODE EXECUTION ---\n".to_string();
@@ -412,7 +406,6 @@ try {{
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         false,
         worker_name,
@@ -420,7 +413,6 @@ try {{
         "deno run",
         job.timeout,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
     // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str());

backend/windmill-worker/src/go_executor.rs

@@ -12,12 +12,12 @@ 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::{
         capitalize, create_args_and_out_file, get_reserved_variables, read_result,
-        start_child_process, OccupancyMetrics,
+        start_child_process,
     },
     handle_child::handle_child,
     AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, GOPRIVATE, GOPROXY,
@@ -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<CanceledBy>,
     job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
@@ -46,7 +45,6 @@ pub async fn handle_go_job(
     base_internal_url: &str,
     worker_name: &str,
     envs: HashMap<String, String>,
-    occupation_metrics: &mut OccupancyMetrics,
 ) -> Result<Box<RawValue>, Error> {
     //go does not like executing modules at temp root
     let job_dir = &format!("{job_dir}/go");
@@ -83,7 +81,6 @@ pub async fn handle_go_job(
             &job.id,
             inner_content,
             mem_peak,
-            canceled_by,
             job_dir,
             db,
             true,
@@ -91,7 +88,6 @@ pub async fn handle_go_job(
             skip_tidy,
             worker_name,
             &job.workspace_id,
-            occupation_metrics,
         )
         .await?;
 
@@ -203,7 +199,6 @@ func Run(req Req) (interface{{}}, error){{
             &job.id,
             db,
             mem_peak,
-            canceled_by,
             build_go_process,
             false,
             worker_name,
@@ -211,7 +206,6 @@ func Run(req Req) (interface{{}}, error){{
             "go build",
             None,
             false,
-            &mut Some(occupation_metrics),
         )
         .await?;
 
@@ -307,7 +301,6 @@ func Run(req Req) (interface{{}}, error){{
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -315,7 +308,6 @@ func Run(req Req) (interface{{}}, error){{
         "go run",
         job.timeout,
         false,
-        &mut Some(occupation_metrics),
     )
     .await?;
 
@@ -347,7 +339,6 @@ pub async fn install_go_dependencies(
     job_id: &Uuid,
     code: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     non_dep_job: bool,
@@ -355,7 +346,6 @@ pub async fn install_go_dependencies(
     has_sum: bool,
     worker_name: &str,
     w_id: &str,
-    occupation_metrics: &mut OccupancyMetrics,
 ) -> error::Result<String> {
     if !skip_go_mod {
         gen_go_mymod(code, job_dir).await?;
@@ -371,7 +361,6 @@ pub async fn install_go_dependencies(
             job_id,
             db,
             mem_peak,
-            canceled_by,
             child_process,
             false,
             worker_name,
@@ -379,7 +368,6 @@ pub async fn install_go_dependencies(
             "go init",
             None,
             false,
-            &mut Some(occupation_metrics),
         )
         .await?;
 
@@ -437,7 +425,6 @@ pub async fn install_go_dependencies(
         job_id,
         db,
         mem_peak,
-        canceled_by,
         child_process,
         false,
         worker_name,
@@ -445,7 +432,6 @@ pub async fn install_go_dependencies(
         &format!("go {mod_command}"),
         None,
         false,
-        &mut Some(occupation_metrics),
     )
     .await?;
 

backend/windmill-worker/src/graphql_executor.rs

@@ -8,11 +8,10 @@ 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, HTTP_CLIENT};
+use windmill_queue::HTTP_CLIENT;
 
 use serde::Deserialize;
 
-use crate::common::OccupancyMetrics;
 use crate::handle_child::run_future_with_polling_update_job_poller;
 use crate::{common::build_args_map, AuthedClientBackgroundTask};
 
@@ -40,9 +39,7 @@ pub async fn do_graphql(
     query: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
-    occupation_metrics: &mut OccupancyMetrics,
 ) -> windmill_common::error::Result<Box<RawValue>> {
     let args = build_args_map(job, client, db).await?.map(Json);
     let job_args = if args.is_some() {
@@ -149,11 +146,9 @@ pub async fn do_graphql(
         job.timeout,
         db,
         mem_peak,
-        canceled_by,
         result_f,
         worker_name,
         &job.workspace_id,
-        &mut Some(occupation_metrics),
         Box::pin(stream::once(async { 0 })),
     )
     .await?;

backend/windmill-worker/src/handle_child.rs

@@ -15,7 +15,7 @@ use windmill_common::error::to_anyhow;
 
 use windmill_common::error::{self, Error};
 
-use windmill_common::worker::{get_windmill_memory_usage, get_worker_memory_usage, CLOUD_HOSTED};
+use windmill_common::worker::CLOUD_HOSTED;
 
 use windmill_queue::{append_logs, CanceledBy};
 
@@ -46,7 +46,7 @@ use futures::{
     stream, StreamExt,
 };
 
-use crate::common::{resolve_job_timeout, OccupancyMetrics};
+use crate::common::resolve_job_timeout;
 use crate::job_logger::{append_job_logs, append_with_limit, LARGE_LOG_THRESHOLD_SIZE};
 use crate::job_logger_ee::process_streaming_log_lines;
 use crate::{MAX_RESULT_SIZE, MAX_WAIT_FOR_SIGINT, MAX_WAIT_FOR_SIGTERM};
@@ -92,7 +92,6 @@ pub async fn handle_child(
     job_id: &Uuid,
     db: &Pool<Postgres>,
     mem_peak: &mut i32,
-    canceled_by_ref: &mut Option<CanceledBy>,
     mut child: Child,
     nsjail: bool,
     worker: &str,
@@ -100,7 +99,6 @@ pub async fn handle_child(
     child_name: &str,
     custom_timeout: Option<i32>,
     sigterm: bool,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> error::Result<()> {
     let start = Instant::now();
 
@@ -136,14 +134,12 @@ 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, ()))
         })),
         worker,
         w_id,
         rx,
-        occupancy_metrics,
     );
 
     enum KillReason {
@@ -493,11 +489,9 @@ pub async fn run_future_with_polling_update_job_poller<Fut, T, S>(
     timeout: Option<i32>,
     db: &DB,
     mem_peak: &mut i32,
-    canceled_by_ref: &mut Option<CanceledBy>,
     result_f: Fut,
     worker_name: &str,
     w_id: &str,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
     get_mem: S,
 ) -> error::Result<T>
 where
@@ -506,17 +500,7 @@ where
 {
     let (tx, rx) = broadcast::channel::<()>(3);
 
-    let update_job = update_job_poller(
-        job_id,
-        db,
-        mem_peak,
-        canceled_by_ref,
-        get_mem,
-        worker_name,
-        w_id,
-        rx,
-        occupancy_metrics,
-    );
+    let update_job = update_job_poller(job_id, db, mem_peak, get_mem, worker_name, w_id, rx);
 
     let timeout_ms = u64::try_from(
         resolve_job_timeout(&db, &w_id, job_id, timeout)
@@ -556,17 +540,16 @@ pub async fn update_job_poller<S>(
     job_id: Uuid,
     db: &DB,
     mem_peak: &mut i32,
-    canceled_by_ref: &mut Option<CanceledBy>,
     mut get_mem: S,
     worker_name: &str,
     w_id: &str,
     mut rx: broadcast::Receiver<()>,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> UpdateJobPollingExit
 where
     S: stream::Stream<Item = i32> + Unpin,
 {
     let update_job_interval = Duration::from_millis(500);
+    let mut cancellation = None;
 
     let db = db.clone();
 
@@ -585,37 +568,12 @@ where
             _ = interval.tick() => {
                 // update the last_ping column every 5 seconds
                 i+=1;
-                if i == 1 || i % 10 == 0 {
-                    let memory_usage = get_worker_memory_usage();
-                    let wm_memory_usage = get_windmill_memory_usage();
-                    tracing::info!("job {job_id} on {worker_name} in {w_id} worker memory snapshot {}kB/{}kB", memory_usage.unwrap_or_default()/1024, wm_memory_usage.unwrap_or_default()/1024);
-                    let occupancy = occupancy_metrics.as_mut().map(|x| x.update_occupancy_metrics());
-                    if job_id != Uuid::nil() {
-                        sqlx::query!(
-                            "UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2, memory_usage = $3, wm_memory_usage = $4,
-                            occupancy_rate = $6, occupancy_rate_15s = $7, occupancy_rate_5m = $8, occupancy_rate_30m = $9 WHERE worker = $5",
-                            &job_id,
-                            &w_id,
-                            memory_usage,
-                            wm_memory_usage,
-                            &worker_name,
-                            occupancy.map(|x| x.0),
-                            occupancy.and_then(|x| x.1),
-                            occupancy.and_then(|x| x.2),
-                            occupancy.and_then(|x| x.3),
-                        )
-                        .execute(&db)
-                        .await
-                        .expect("update worker ping");
-                    }
-                }
                 let current_mem = get_mem.next().await.unwrap_or(0);
                 if current_mem > *mem_peak {
                     *mem_peak = current_mem
                 }
                 tracing::info!("job {job_id} on {worker_name} in {w_id} still running.  mem: {current_mem}kB, peak mem: {mem_peak}kB");
 
-
                 let update_job_row = i == 2 || (!*SLOW_LOGS && (i < 20 || (i < 120 && i % 5 == 0) || i % 10 == 0)) || i % 20 == 0;
                 if update_job_row {
                 #[cfg(feature = "enterprise")]
@@ -659,9 +617,9 @@ where
                         return UpdateJobPollingExit::AlreadyCompleted
                     }
                     if canceled {
-                        canceled_by_ref.replace(CanceledBy {
-                            username: canceled_by.clone(),
-                            reason: canceled_reason.clone(),
+                        cancellation = Some(CanceledBy {
+                            username: canceled_by,
+                            reason: canceled_reason,
                         });
                         break
                     }
@@ -672,7 +630,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

backend/windmill-worker/src/js_eval.rs

@@ -44,9 +44,8 @@ use uuid::Uuid;
 use windmill_common::error::Error;
 
 use windmill_common::{flow_status::JobResult, DB};
-use windmill_queue::CanceledBy;
 
-use crate::{common::OccupancyMetrics, AuthedClient};
+use crate::AuthedClient;
 
 #[cfg(feature = "deno_core")]
 use crate::{common::unsafe_raw, handle_child::run_future_with_polling_update_job_poller};
@@ -743,11 +742,9 @@ pub async fn eval_fetch_timeout(
     _job_timeout: Option<i32>,
     _db: &DB,
     _mem_peak: &mut i32,
-    _canceled_by: &mut Option<CanceledBy>,
     _worker_name: &str,
     _w_id: &str,
     _load_client: bool,
-    _occupation_metrics: &mut OccupancyMetrics,
 ) -> anyhow::Result<Box<RawValue>> {
     use serde_json::value::to_raw_value;
     Ok(to_raw_value("require deno_core").unwrap())
@@ -763,11 +760,9 @@ pub async fn eval_fetch_timeout(
     job_timeout: Option<i32>,
     db: &DB,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
     w_id: &str,
     load_client: bool,
-    occupation_metrics: &mut OccupancyMetrics,
 ) -> anyhow::Result<Box<RawValue>> {
     use windmill_queue::append_logs;
 
@@ -913,11 +908,9 @@ pub async fn eval_fetch_timeout(
         job_timeout,
         db,
         mem_peak,
-        canceled_by,
         async { result_f.await? },
         worker_name,
         w_id,
-        &mut Some(occupation_metrics),
         Box::pin(futures::stream::once(async { 0 })),
     )
     .await

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,9 +38,7 @@ pub async fn do_mssql(
     query: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<Box<RawValue>> {
     let mssql_args = build_args_values(job, client, db).await?;
 
@@ -155,11 +153,9 @@ pub async fn do_mssql(
         job.timeout,
         db,
         mem_peak,
-        canceled_by,
         result_f,
         worker_name,
         &job.workspace_id,
-        &mut Some(occupancy_metrics),
         Box::pin(futures::stream::once(async { 0 })),
     )
     .await?;

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,10 +107,8 @@ pub async fn do_mysql(
     query: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
     column_order: &mut Option<Vec<String>>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> windmill_common::error::Result<Box<RawValue>> {
     let args = build_args_map(job, client, db).await?.map(Json);
     let job_args = if args.is_some() {
@@ -293,11 +290,9 @@ pub async fn do_mysql(
         job.timeout,
         db,
         mem_peak,
-        canceled_by,
         result_f,
         worker_name,
         &job.workspace_id,
-        &mut Some(occupancy_metrics),
         Box::pin(futures::stream::once(async { 0 })),
     )
     .await?;

backend/windmill-worker/src/pg_executor.rs

@@ -32,9 +32,8 @@ 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::common::{build_args_values, sizeof_val};
 use crate::handle_child::run_future_with_polling_update_job_poller;
 use crate::{AuthedClientBackgroundTask, MAX_RESULT_SIZE};
 use bytes::Buf;
@@ -161,10 +160,8 @@ pub async fn do_postgresql(
     query: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
     column_order: &mut Option<Vec<String>>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<Box<RawValue>> {
     let pg_args = build_args_values(job, client, db).await?;
 
@@ -350,11 +347,9 @@ pub async fn do_postgresql(
         job.timeout,
         db,
         mem_peak,
-        canceled_by,
         result_f,
         worker_name,
         &job.workspace_id,
-        &mut Some(occupancy_metrics),
         Box::pin(futures::stream::once(async { 0 })),
     )
     .await?;

backend/windmill-worker/src/php_executor.rs

@@ -11,11 +11,12 @@ use windmill_common::{
     worker::write_file,
 };
 use windmill_parser::Typ;
-use windmill_queue::{append_logs, CanceledBy};
+use windmill_queue::append_logs;
 
 use crate::{
     common::{
-        check_executor_binary_exists, create_args_and_out_file, get_main_override, get_reserved_variables, read_result, start_child_process, OccupancyMetrics
+        check_executor_binary_exists, create_args_and_out_file, get_main_override,
+        get_reserved_variables, read_result, start_child_process, OccupancyMetrics,
     },
     handle_child::handle_child,
     AuthedClientBackgroundTask, COMPOSER_CACHE_DIR, COMPOSER_PATH, DISABLE_NSJAIL, DISABLE_NUSER,
@@ -62,7 +63,6 @@ pub fn parse_php_imports(code: &str) -> anyhow::Result<Option<String>> {
 
 pub async fn composer_install(
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_id: &Uuid,
     w_id: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
@@ -70,7 +70,6 @@ pub async fn composer_install(
     worker_name: &str,
     requirements: String,
     lock: Option<String>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> Result<String> {
     check_executor_binary_exists("php", PHP_PATH.as_str(), "php")?;
 
@@ -94,7 +93,6 @@ pub async fn composer_install(
         job_id,
         db,
         mem_peak,
-        canceled_by,
         child_process,
         false,
         worker_name,
@@ -102,7 +100,6 @@ pub async fn composer_install(
         "composer install",
         None,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 
@@ -134,7 +131,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<CanceledBy>,
     job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
@@ -144,7 +140,6 @@ pub async fn handle_php_job(
     worker_name: &str,
     envs: HashMap<String, String>,
     shared_mount: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<Box<RawValue>> {
     check_executor_binary_exists("php", PHP_PATH.as_str(), "php")?;
 
@@ -167,7 +162,6 @@ pub async fn handle_php_job(
 
         composer_install(
             mem_peak,
-            canceled_by,
             &job.id,
             &job.workspace_id,
             db,
@@ -175,7 +169,6 @@ pub async fn handle_php_job(
             worker_name,
             composer_json,
             composer_lock,
-            occupancy_metrics,
         )
         .await?;
         "require './vendor/autoload.php';"
@@ -324,7 +317,6 @@ try {{
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -332,7 +324,6 @@ try {{
         "php run",
         job.timeout,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
     read_result(job_dir).await

backend/windmill-worker/src/python_executor.rs

@@ -32,7 +32,7 @@ use windmill_common::{
 #[cfg(feature = "enterprise")]
 use windmill_common::variables::get_secret_value_as_admin;
 
-use windmill_queue::{append_logs, CanceledBy};
+use windmill_queue::append_logs;
 
 lazy_static::lazy_static! {
     static ref PYTHON_PATH: String =
@@ -80,7 +80,7 @@ use windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS;
 use crate::{
     common::{
         create_args_and_out_file, get_main_override, get_reserved_variables, read_file,
-        read_result, start_child_process, OccupancyMetrics,
+        read_result, start_child_process,
     },
     handle_child::{get_mem_peak, handle_child},
     AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, LOCK_CACHE_DIR,
@@ -124,12 +124,10 @@ pub async fn uv_pip_compile(
     job_id: &Uuid,
     requirements: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &Pool<Postgres>,
     worker_name: &str,
     w_id: &str,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
     // Fallback to pip-compile. Will be removed in future
     mut no_uv: bool,
     // Debug-only flag
@@ -256,7 +254,6 @@ pub async fn uv_pip_compile(
             job_id,
             db,
             mem_peak,
-            canceled_by,
             child_process,
             false,
             worker_name,
@@ -264,7 +261,6 @@ pub async fn uv_pip_compile(
             "pip-compile",
             None,
             false,
-            occupancy_metrics,
         )
         .await
         .map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?;
@@ -341,7 +337,6 @@ pub async fn uv_pip_compile(
             job_id,
             db,
             mem_peak,
-            canceled_by,
             child_process,
             false,
             worker_name,
@@ -350,7 +345,6 @@ pub async fn uv_pip_compile(
             "uv",
             None,
             false,
-            occupancy_metrics,
         )
         .await
         .map_err(|e| {
@@ -514,7 +508,6 @@ pub async fn handle_python_job(
     worker_name: &str,
     job: &QueuedJob,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
     inner_content: &String,
@@ -522,7 +515,6 @@ pub async fn handle_python_job(
     base_internal_url: &str,
     envs: HashMap<String, String>,
     new_args: &mut Option<HashMap<String, Box<RawValue>>>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> windmill_common::error::Result<Box<RawValue>> {
     let script_path = crate::common::use_flow_root_path(job.script_path());
     let mut additional_python_paths = handle_python_deps(
@@ -536,8 +528,6 @@ pub async fn handle_python_job(
         worker_name,
         worker_dir,
         mem_peak,
-        canceled_by,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 
@@ -782,7 +772,6 @@ mount {{
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -790,7 +779,6 @@ mount {{
         "python run",
         job.timeout,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 
@@ -1058,8 +1046,6 @@ async fn handle_python_deps(
     worker_name: &str,
     worker_dir: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
 ) -> error::Result<Vec<String>> {
     create_dependencies_dir(job_dir).await;
 
@@ -1092,12 +1078,10 @@ async fn handle_python_deps(
                     job_id,
                     &requirements,
                     mem_peak,
-                    canceled_by,
                     job_dir,
                     db,
                     worker_name,
                     w_id,
-                    occupancy_metrics,
                     annotations.no_uv || annotations.no_uv_compile,
                     annotations.no_cache,
                 )
@@ -1119,12 +1103,10 @@ async fn handle_python_deps(
             job_id,
             w_id,
             mem_peak,
-            canceled_by,
             db,
             worker_name,
             job_dir,
             worker_dir,
-            occupancy_metrics,
             annotations.no_uv || annotations.no_uv_install,
             false,
         )
@@ -1331,12 +1313,10 @@ pub async fn handle_python_reqs(
     job_id: &Uuid,
     w_id: &str,
     mem_peak: &mut i32,
-    _canceled_by: &mut Option<CanceledBy>,
     db: &sqlx::Pool<sqlx::Postgres>,
     _worker_name: &str,
     job_dir: &str,
     worker_dir: &str,
-    _occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
     // TODO: Remove (Deprecated)
     mut no_uv_install: bool,
     is_ansible: bool,
@@ -1914,7 +1894,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<CanceledBy> = None;
     let context = variables::get_reserved_variables(
         db,
         w_id,
@@ -1947,8 +1926,6 @@ pub async fn start_worker(
         worker_name,
         job_dir,
         &mut mem_peak,
-        &mut canceled_by,
-        &mut None,
     )
     .await?;
 

backend/windmill-worker/src/result_processor.rs

@@ -28,7 +28,7 @@ use windmill_common::{
 #[cfg(feature = "benchmark")]
 use windmill_common::bench::{BenchmarkInfo, BenchmarkIter};
 
-use windmill_queue::{append_logs, get_queued_job, CanceledBy, WrappedError};
+use windmill_queue::{append_logs, get_queued_job, WrappedError};
 
 #[cfg(feature = "prometheus")]
 use windmill_queue::register_metric;
@@ -236,22 +236,12 @@ async fn send_job_completed(
     job: Arc<QueuedJob>,
     result: Arc<Box<RawValue>>,
     mem_peak: i32,
-    canceled_by: Option<CanceledBy>,
     success: bool,
     cached_res_path: Option<String>,
     token: String,
     duration: Option<i64>,
 ) {
-    let jc = JobCompleted {
-        job,
-        result,
-        mem_peak,
-        canceled_by,
-        success,
-        cached_res_path,
-        token,
-        duration,
-    };
+    let jc = JobCompleted { job, result, mem_peak, success, cached_res_path, token, duration };
     job_completed_tx
         .send(jc)
         .with_context(windmill_common::otel_ee::otel_ctx())
@@ -265,7 +255,6 @@ pub async fn process_result(
     job_dir: &str,
     job_completed_tx: JobCompletedSender,
     mem_peak: i32,
-    canceled_by: Option<CanceledBy>,
     cached_res_path: Option<String>,
     token: String,
     column_order: Option<Vec<String>>,
@@ -322,7 +311,6 @@ pub async fn process_result(
                 job,
                 r,
                 mem_peak,
-                canceled_by,
                 true,
                 cached_res_path,
                 token,
@@ -367,7 +355,6 @@ pub async fn process_result(
                 job,
                 Arc::new(to_raw_value(&error_value)),
                 mem_peak,
-                canceled_by,
                 false,
                 cached_res_path,
                 token,
@@ -400,7 +387,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,
@@ -420,7 +406,6 @@ pub async fn handle_receive_completed_job(
                 &client,
                 job.as_ref(),
                 mem_peak,
-                canceled_by,
                 err,
                 false,
                 same_worker_tx.clone(),
@@ -438,7 +423,7 @@ pub async fn handle_receive_completed_job(
 }
 
 pub async fn process_completed_job(
-    JobCompleted { job, result, mem_peak, success, cached_res_path, canceled_by, duration, .. }: JobCompleted,
+    JobCompleted { job, result, mem_peak, success, cached_res_path, duration, .. }: JobCompleted,
     client: &AuthedClient,
     db: &DB,
     worker_dir: &str,
@@ -465,7 +450,7 @@ pub async fn process_completed_job(
             false,
             Json(&result),
             mem_peak.to_owned(),
-            canceled_by,
+            None,
             false,
             duration,
         )
@@ -505,7 +490,7 @@ pub async fn process_completed_job(
             db,
             &job,
             mem_peak.to_owned(),
-            canceled_by,
+            None,
             serde_json::from_str(result.get()).unwrap_or_else(
                 |_| json!({ "message": format!("Non serializable error: {}", result.get()) }),
             ),
@@ -549,7 +534,6 @@ pub async fn handle_job_error(
     client: &AuthedClient,
     job: &QueuedJob,
     mem_peak: i32,
-    canceled_by: Option<CanceledBy>,
     err: Error,
     unrecoverable: bool,
     same_worker_tx: SameWorkerSender,
@@ -575,7 +559,7 @@ pub async fn handle_job_error(
             db,
             job,
             mem_peak,
-            canceled_by.clone(),
+            None,
             err.clone(),
             worker_name,
             false,
@@ -635,7 +619,7 @@ pub async fn handle_job_error(
                         db,
                         &parent_job,
                         mem_peak,
-                        canceled_by.clone(),
+                        None,
                         e,
                         worker_name,
                         false,

backend/windmill-worker/src/rust_executor.rs

@@ -11,11 +11,12 @@ 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::{
-        check_executor_binary_exists, create_args_and_out_file, get_reserved_variables, read_result, start_child_process, OccupancyMetrics
+        check_executor_binary_exists, create_args_and_out_file, get_reserved_variables,
+        read_result, start_child_process,
     },
     handle_child::handle_child,
     AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV,
@@ -124,12 +125,10 @@ pub async fn generate_cargo_lockfile(
     job_id: &Uuid,
     code: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
     w_id: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<String> {
     check_executor_binary_exists("cargo", CARGO_PATH.as_str(), "rust")?;
 
@@ -154,7 +153,6 @@ pub async fn generate_cargo_lockfile(
         job_id,
         db,
         mem_peak,
-        canceled_by,
         gen_lockfile_process,
         false,
         worker_name,
@@ -162,7 +160,6 @@ pub async fn generate_cargo_lockfile(
         "cargo generate-lockfile",
         None,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
 
@@ -176,14 +173,12 @@ pub async fn generate_cargo_lockfile(
 pub async fn build_rust_crate(
     job_id: &Uuid,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
     w_id: &str,
     base_internal_url: &str,
     hash: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<String> {
     let bin_path = format!("{}/{hash}", RUST_CACHE_DIR);
 
@@ -216,7 +211,6 @@ pub async fn build_rust_crate(
         job_id,
         db,
         mem_peak,
-        canceled_by,
         build_rust_process,
         false,
         worker_name,
@@ -224,7 +218,6 @@ pub async fn build_rust_crate(
         "rust build",
         None,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
     append_logs(job_id, w_id, "\n\n", db).await;
@@ -273,7 +266,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<CanceledBy>,
     job: &QueuedJob,
     db: &sqlx::Pool<sqlx::Postgres>,
     client: &AuthedClientBackgroundTask,
@@ -284,7 +276,6 @@ pub async fn handle_rust_job(
     base_internal_url: &str,
     worker_name: &str,
     envs: HashMap<String, String>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> Result<Box<RawValue>, Error> {
     check_executor_binary_exists("cargo", CARGO_PATH.as_str(), "rust")?;
 
@@ -327,14 +318,12 @@ pub async fn handle_rust_job(
         build_rust_crate(
             &job.id,
             mem_peak,
-            canceled_by,
             job_dir,
             db,
             worker_name,
             &job.workspace_id,
             base_internal_url,
             &hash,
-            occupancy_metrics,
         )
         .await?
     };
@@ -395,7 +384,6 @@ pub async fn handle_rust_job(
         &job.id,
         db,
         mem_peak,
-        canceled_by,
         child,
         !*DISABLE_NSJAIL,
         worker_name,
@@ -403,7 +391,6 @@ pub async fn handle_rust_job(
         "rust run",
         job.timeout,
         false,
-        &mut Some(occupancy_metrics),
     )
     .await?;
     read_result(job_dir).await

backend/windmill-worker/src/snowflake_executor.rs

@@ -13,11 +13,11 @@ 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};
 
-use crate::common::{resolve_job_timeout, OccupancyMetrics};
+use crate::common::resolve_job_timeout;
 use crate::handle_child::run_future_with_polling_update_job_poller;
 use crate::{common::build_args_values, AuthedClientBackgroundTask};
 
@@ -244,10 +244,8 @@ pub async fn do_snowflake(
     query: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
     column_order: &mut Option<Vec<String>>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> windmill_common::error::Result<Box<RawValue>> {
     let snowflake_args = build_args_values(job, client, db).await?;
 
@@ -414,11 +412,9 @@ pub async fn do_snowflake(
         job.timeout,
         db,
         mem_peak,
-        canceled_by,
         result_f.map_err(to_anyhow),
         worker_name,
         &job.workspace_id,
-        &mut Some(occupancy_metrics),
         Box::pin(futures::stream::once(async { 0 })),
     )
     .await?;

backend/windmill-worker/src/worker.rs

@@ -35,12 +35,14 @@ use windmill_common::METRICS_DEBUG_ENABLED;
 #[cfg(feature = "prometheus")]
 use windmill_common::METRICS_ENABLED;
 
+use futures::{future, future::Either};
 use reqwest::Response;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use sqlx::{types::Json, Pool, Postgres};
 use std::{
     collections::HashMap,
     fs::DirBuilder,
+    pin::pin,
     sync::{
         atomic::{AtomicBool, AtomicU16, Ordering},
         Arc,
@@ -63,7 +65,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,
 };
 
@@ -84,7 +86,7 @@ use tokio::{
         RwLock,
     },
     task::JoinHandle,
-    time::Instant,
+    time::{interval, Instant, MissedTickBehavior},
 };
 
 use rand::Rng;
@@ -127,11 +129,10 @@ use crate::ansible_executor::handle_ansible_job;
 #[cfg(feature = "mysql")]
 use crate::mysql_executor::do_mysql;
 
-use backon::ConstantBuilder;
-use backon::{BackoffBuilder, Retryable};
-
 #[cfg(feature = "enterprise")]
 use crate::dedicated_worker::create_dedicated_worker_map;
+use backon::ConstantBuilder;
+use backon::{BackoffBuilder, Retryable};
 
 #[cfg(feature = "enterprise")]
 use crate::snowflake_executor::do_snowflake;
@@ -1438,7 +1439,6 @@ pub async fn run_worker(
                             mem_peak: 0,
                             cached_res_path: None,
                             token: "".to_string(),
-                            canceled_by: None,
                             duration: None,
                         })
                         .await
@@ -1592,35 +1592,96 @@ pub async fn run_worker(
                     windmill_common::otel_ee::set_span_parent(&span, &rj);
                     // span.context().span().add_event_with_timestamp("job created".to_string(), arc_job.created_at.into(), vec![]);
 
-                    match handle_queued_job(
-                        arc_job.clone(),
-                        raw_code,
-                        raw_lock,
-                        raw_flow,
-                        db,
-                        &authed_client,
-                        &hostname,
-                        &worker_name,
-                        &worker_dir,
-                        &job_dir,
-                        same_worker_tx.clone(),
-                        base_internal_url,
-                        job_completed_tx.clone(),
-                        &mut occupancy_metrics,
-                        &mut killpill_rx2,
-                        #[cfg(feature = "benchmark")]
-                        &mut bench,
+                    let mut script_duration = 0f32;
+                    let mut ping_interval = interval(Duration::from_secs(5));
+                    ping_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+                    let Either::Left((result, _)) = future::select(
+                        // Handle job:
+                        pin!(handle_queued_job(
+                            arc_job.clone(),
+                            raw_code,
+                            raw_lock,
+                            raw_flow,
+                            db,
+                            &authed_client,
+                            &hostname,
+                            &worker_name,
+                            &worker_dir,
+                            &job_dir,
+                            same_worker_tx.clone(),
+                            base_internal_url,
+                            job_completed_tx.clone(),
+                            &mut script_duration,
+                            &mut killpill_rx2,
+                            #[cfg(feature = "benchmark")]
+                            &mut bench,
+                        )
+                        .instrument(span)),
+                        // Ping worker every 5 seconds:
+                        pin!(async {
+                            loop {
+                                ping_interval.tick().await;
+
+                                let memory_usage = get_worker_memory_usage();
+                                let wm_memory_usage = get_windmill_memory_usage();
+                                tracing::info!(
+                                    "job {} on {worker_name} in {} worker memory snapshot {}kB/{}kB",
+                                    &arc_job.id,
+                                    &arc_job.workspace_id,
+                                    memory_usage.unwrap_or_default() / 1024,
+                                    wm_memory_usage.unwrap_or_default() / 1024
+                                );
+                                let (
+                                    occupancy_rate,
+                                    occupancy_rate_15s,
+                                    occupancy_rate_5m,
+                                    occupancy_rate_30m,
+                                ) = occupancy_metrics.update_occupancy_metrics();
+                                let ping_res = sqlx::query!(
+                                    "UPDATE worker_ping SET
+                                          ping_at = now()
+                                        , current_job_id = $1
+                                        , current_job_workspace_id = $2
+                                        , memory_usage = $3
+                                        , wm_memory_usage = $4
+                                        , occupancy_rate = $6
+                                        , occupancy_rate_15s = $7
+                                        , occupancy_rate_5m = $8
+                                        , occupancy_rate_30m = $9
+                                    WHERE worker = $5",
+                                    &arc_job.id,
+                                    &arc_job.workspace_id,
+                                    memory_usage,
+                                    wm_memory_usage,
+                                    &worker_name,
+                                    occupancy_rate,
+                                    occupancy_rate_15s,
+                                    occupancy_rate_5m,
+                                    occupancy_rate_30m
+                                )
+                                .execute(db)
+                                .await;
+                                if let Err(err) = ping_res {
+                                    tracing::error!("failed to update worker ping: {}", err);
+                                }
+                                last_ping = Instant::now();
+                            }
+                        }),
                     )
-                    .instrument(span)
                     .await
-                    {
+                    else {
+                        unreachable!()
+                    };
+
+                    occupancy_metrics.total_duration_of_running_jobs += script_duration;
+                    match result {
                         Err(err) => {
                             handle_job_error(
                                 db,
                                 &authed_client.get_authed().await,
                                 arc_job.as_ref(),
                                 0,
-                                None,
                                 err,
                                 false,
                                 same_worker_tx.clone(),
@@ -1674,7 +1735,7 @@ pub async fn run_worker(
 
                     if !KEEP_JOB_DIR.load(Ordering::Relaxed) && !(arc_job.is_flow() && same_worker)
                     {
-                        let _ = tokio::fs::remove_dir_all(job_dir).await;
+                        let _ = tokio::fs::remove_dir_all(&job_dir).await;
                     }
                 }
 
@@ -1835,7 +1896,6 @@ pub struct JobCompleted {
     pub success: bool,
     pub cached_res_path: Option<String>,
     pub token: String,
-    pub canceled_by: Option<CanceledBy>,
     pub duration: Option<i64>,
 }
 
@@ -1846,9 +1906,7 @@ async fn do_nativets(
     code: String,
     db: &Pool<Postgres>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     worker_name: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> windmill_common::error::Result<Box<RawValue>> {
     let args = build_args_map(job, client, db).await?.map(Json);
     let job_args = if args.is_some() {
@@ -1866,11 +1924,9 @@ async fn do_nativets(
         job.timeout,
         db,
         mem_peak,
-        canceled_by,
         worker_name,
         &job.workspace_id,
         true,
-        occupancy_metrics,
     )
     .await?)
 }
@@ -1895,7 +1951,7 @@ async fn handle_queued_job(
     same_worker_tx: SameWorkerSender,
     base_internal_url: &str,
     job_completed_tx: JobCompletedSender,
-    occupancy_metrics: &mut OccupancyMetrics,
+    duration: &mut f32,
     killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
     #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
 ) -> windmill_common::error::Result<bool> {
@@ -2008,7 +2064,6 @@ async fn handle_queued_job(
                     job,
                     result,
                     mem_peak: 0,
-                    canceled_by: None,
                     success: true,
                     cached_res_path: None,
                     token: authed_client.token,
@@ -2042,7 +2097,6 @@ async fn handle_queued_job(
     } else {
         let mut logs = "".to_string();
         let mut mem_peak: i32 = 0;
-        let mut canceled_by: Option<CanceledBy> = None;
         // println!("handle queue {:?}",  SystemTime::now());
 
         logs.push_str(&format!(
@@ -2084,14 +2138,12 @@ async fn handle_queued_job(
                     &job,
                     preview_data.as_ref(),
                     &mut mem_peak,
-                    &mut canceled_by,
                     job_dir,
                     db,
                     worker_name,
                     worker_dir,
                     base_internal_url,
                     &client.get_token().await,
-                    occupancy_metrics,
                 )
                 .await
             }
@@ -2100,28 +2152,24 @@ async fn handle_queued_job(
                     &job,
                     preview_data.as_ref(),
                     &mut mem_peak,
-                    &mut canceled_by,
                     job_dir,
                     db,
                     worker_name,
                     worker_dir,
                     base_internal_url,
                     &client.get_token().await,
-                    occupancy_metrics,
                 )
                 .await
             }
             JobKind::AppDependencies => handle_app_dependency_job(
                 &job,
                 &mut mem_peak,
-                &mut canceled_by,
                 job_dir,
                 db,
                 worker_name,
                 worker_dir,
                 base_internal_url,
                 &client.get_token().await,
-                occupancy_metrics,
             )
             .await
             .map(|()| serde_json::from_str("{}").unwrap()),
@@ -2146,17 +2194,14 @@ async fn handle_queued_job(
                     job_dir,
                     worker_dir,
                     &mut mem_peak,
-                    &mut canceled_by,
                     base_internal_url,
                     worker_name,
                     &mut column_order,
                     &mut new_args,
-                    occupancy_metrics,
                     killpill_rx,
                 )
                 .await;
-                occupancy_metrics.total_duration_of_running_jobs +=
-                    metric_timer.elapsed().as_secs_f32();
+                *duration = metric_timer.elapsed().as_secs_f32();
                 r
             }
         };
@@ -2179,7 +2224,6 @@ async fn handle_queued_job(
             job_dir,
             job_completed_tx,
             mem_peak,
-            canceled_by,
             cached_res_path,
             client.get_token().await,
             column_order,
@@ -2272,12 +2316,10 @@ async fn handle_code_execution_job(
     job_dir: &str,
     #[allow(unused_variables)] worker_dir: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     base_internal_url: &str,
     worker_name: &str,
     column_order: &mut Option<Vec<String>>,
     new_args: &mut Option<HashMap<String, Box<RawValue>>>,
-    occupancy_metrics: &mut OccupancyMetrics,
     killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
 ) -> error::Result<Box<RawValue>> {
     let script_hash = || {
@@ -2364,10 +2406,8 @@ async fn handle_code_execution_job(
             &code,
             db,
             mem_peak,
-            canceled_by,
             worker_name,
             column_order,
-            occupancy_metrics,
         )
         .await;
     } else if language == Some(ScriptLang::Mysql) {
@@ -2383,10 +2423,8 @@ async fn handle_code_execution_job(
             &code,
             db,
             mem_peak,
-            canceled_by,
             worker_name,
             column_order,
-            occupancy_metrics,
         )
         .await;
     } else if language == Some(ScriptLang::Bigquery) {
@@ -2413,10 +2451,8 @@ async fn handle_code_execution_job(
                 &code,
                 db,
                 mem_peak,
-                canceled_by,
                 worker_name,
                 column_order,
-                occupancy_metrics,
             )
             .await;
         }
@@ -2436,10 +2472,8 @@ async fn handle_code_execution_job(
                 &code,
                 db,
                 mem_peak,
-                canceled_by,
                 worker_name,
                 column_order,
-                occupancy_metrics,
             )
             .await;
         }
@@ -2461,30 +2495,10 @@ async fn handle_code_execution_job(
 
         #[cfg(all(feature = "enterprise", feature = "mssql"))]
         {
-            return do_mssql(
-                job,
-                &client,
-                &code,
-                db,
-                mem_peak,
-                canceled_by,
-                worker_name,
-                occupancy_metrics,
-            )
-            .await;
+            return do_mssql(job, &client, &code, db, mem_peak, worker_name).await;
         }
     } else if language == Some(ScriptLang::Graphql) {
-        return do_graphql(
-            job,
-            &client,
-            &code,
-            db,
-            mem_peak,
-            canceled_by,
-            worker_name,
-            occupancy_metrics,
-        )
-        .await;
+        return do_graphql(job, &client, &code, db, mem_peak, worker_name).await;
     } else if language == Some(ScriptLang::Nativets) {
         append_logs(
             &job.id,
@@ -2511,9 +2525,7 @@ async fn handle_code_execution_job(
             code.clone(),
             db,
             mem_peak,
-            canceled_by,
             worker_name,
-            occupancy_metrics,
         )
         .await?;
         return Ok(result);
@@ -2576,7 +2588,6 @@ mount {{
                 worker_name,
                 job,
                 mem_peak,
-                canceled_by,
                 db,
                 client,
                 &code,
@@ -2584,7 +2595,6 @@ mount {{
                 base_internal_url,
                 envs,
                 new_args,
-                occupancy_metrics,
             )
             .await
         }
@@ -2592,7 +2602,6 @@ mount {{
             handle_deno_job(
                 lock.as_ref(),
                 mem_peak,
-                canceled_by,
                 job,
                 db,
                 client,
@@ -2602,7 +2611,6 @@ mount {{
                 worker_name,
                 envs,
                 new_args,
-                occupancy_metrics,
             )
             .await
         }
@@ -2611,7 +2619,6 @@ mount {{
                 lock.as_ref(),
                 codebase.as_ref(),
                 mem_peak,
-                canceled_by,
                 job,
                 db,
                 client,
@@ -2622,14 +2629,12 @@ mount {{
                 envs,
                 &shared_mount,
                 new_args,
-                occupancy_metrics,
             )
             .await
         }
         Some(ScriptLang::Go) => {
             handle_go_job(
                 mem_peak,
-                canceled_by,
                 job,
                 db,
                 client,
@@ -2640,14 +2645,12 @@ mount {{
                 base_internal_url,
                 worker_name,
                 envs,
-                occupancy_metrics,
             )
             .await
         }
         Some(ScriptLang::Bash) => {
             handle_bash_job(
                 mem_peak,
-                canceled_by,
                 job,
                 db,
                 client,
@@ -2657,7 +2660,6 @@ mount {{
                 base_internal_url,
                 worker_name,
                 envs,
-                occupancy_metrics,
                 killpill_rx,
             )
             .await
@@ -2665,7 +2667,6 @@ mount {{
         Some(ScriptLang::Powershell) => {
             handle_powershell_job(
                 mem_peak,
-                canceled_by,
                 job,
                 db,
                 client,
@@ -2675,7 +2676,6 @@ mount {{
                 base_internal_url,
                 worker_name,
                 envs,
-                occupancy_metrics,
             )
             .await
         }
@@ -2689,7 +2689,6 @@ mount {{
             handle_php_job(
                 lock.as_ref(),
                 mem_peak,
-                canceled_by,
                 job,
                 db,
                 client,
@@ -2699,7 +2698,6 @@ mount {{
                 worker_name,
                 envs,
                 &shared_mount,
-                occupancy_metrics,
             )
             .await
         }
@@ -2712,7 +2710,6 @@ mount {{
             #[cfg(feature = "rust")]
             handle_rust_job(
                 mem_peak,
-                canceled_by,
                 job,
                 db,
                 client,
@@ -2723,7 +2720,6 @@ mount {{
                 base_internal_url,
                 worker_name,
                 envs,
-                occupancy_metrics,
             )
             .await
         }
@@ -2741,21 +2737,18 @@ mount {{
                 worker_name,
                 job,
                 mem_peak,
-                canceled_by,
                 db,
                 client,
                 &code,
                 &shared_mount,
                 base_internal_url,
                 envs,
-                occupancy_metrics,
             )
             .await
         }
         Some(ScriptLang::CSharp) => {
             handle_csharp_job(
                 mem_peak,
-                canceled_by,
                 job,
                 db,
                 client,
@@ -2766,7 +2759,6 @@ mount {{
                 base_internal_url,
                 worker_name,
                 envs,
-                occupancy_metrics,
             )
             .await
         }

backend/windmill-worker/src/worker_flow.rs

@@ -55,8 +55,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<sqlx::Postgres>;
@@ -1050,10 +1050,7 @@ 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(),
-                }),
+                None,
                 canceled_job_to_result(&flow_job),
                 worker_name,
                 true,

backend/windmill-worker/src/worker_lockfiles.rs

@@ -30,9 +30,8 @@ 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;
 
 #[cfg(feature = "php")]
@@ -220,14 +219,12 @@ pub async fn handle_dependency_job(
     job: &QueuedJob,
     preview_data: Option<&RawData>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
     worker_dir: &str,
     base_internal_url: &str,
     token: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<Box<RawValue>> {
     let script_path = job.script_path();
     let raw_deps = job
@@ -276,7 +273,6 @@ pub async fn handle_dependency_job(
         })?,
         &script_data.code,
         mem_peak,
-        canceled_by,
         job_dir,
         db,
         worker_name,
@@ -287,7 +283,6 @@ pub async fn handle_dependency_job(
         script_path,
         raw_deps,
         npm_mode,
-        occupancy_metrics,
     )
     .await;
 
@@ -539,14 +534,12 @@ pub async fn handle_flow_dependency_job(
     job: &QueuedJob,
     preview_data: Option<&RawData>,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
     worker_dir: &str,
     base_internal_url: &str,
     token: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<Box<serde_json::value::RawValue>> {
     let job_path = job.script_path.clone().ok_or_else(|| {
         error::Error::InternalErr(
@@ -615,7 +608,6 @@ pub async fn handle_flow_dependency_job(
         flow.modules,
         job,
         mem_peak,
-        canceled_by,
         job_dir,
         db,
         tx,
@@ -625,7 +617,6 @@ pub async fn handle_flow_dependency_job(
         base_internal_url,
         token,
         &nodes_to_relock,
-        occupancy_metrics,
     )
     .await?;
     let new_flow_value = Json(serde_json::value::to_raw_value(&flow).map_err(to_anyhow)?);
@@ -739,7 +730,6 @@ async fn lock_modules<'c>(
     modules: Vec<FlowModule>,
     job: &QueuedJob,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
@@ -749,7 +739,6 @@ async fn lock_modules<'c>(
     base_internal_url: &str,
     token: &str,
     locks_to_reload: &Option<Vec<String>>,
-    occupancy_metrics: &mut OccupancyMetrics,
     // (modules to replace old seq (even unmmodified ones), new transaction, modified ids) )
 ) -> Result<(
     Vec<FlowModule>,
@@ -787,7 +776,6 @@ async fn lock_modules<'c>(
                         modules,
                         job,
                         mem_peak,
-                        canceled_by,
                         job_dir,
                         db,
                         tx,
@@ -797,7 +785,6 @@ async fn lock_modules<'c>(
                         base_internal_url,
                         token,
                         locks_to_reload,
-                        occupancy_metrics,
                     ))
                     .await?;
                     e.value = FlowModuleValue::ForloopFlow {
@@ -820,7 +807,6 @@ async fn lock_modules<'c>(
                             b.modules,
                             job,
                             mem_peak,
-                            canceled_by,
                             job_dir,
                             db,
                             tx,
@@ -830,7 +816,6 @@ async fn lock_modules<'c>(
                             base_internal_url,
                             token,
                             locks_to_reload,
-                            occupancy_metrics,
                         ))
                         .await?;
                         nmodified_ids.extend(inner_modified_ids);
@@ -845,7 +830,6 @@ async fn lock_modules<'c>(
                         modules,
                         job,
                         mem_peak,
-                        canceled_by,
                         job_dir,
                         db,
                         tx,
@@ -855,7 +839,6 @@ async fn lock_modules<'c>(
                         base_internal_url,
                         token,
                         locks_to_reload,
-                        occupancy_metrics,
                     ))
                     .await?;
                     e.value = FlowModuleValue::WhileloopFlow {
@@ -876,7 +859,6 @@ async fn lock_modules<'c>(
                             b.modules,
                             job,
                             mem_peak,
-                            canceled_by,
                             job_dir,
                             db,
                             tx,
@@ -886,7 +868,6 @@ async fn lock_modules<'c>(
                             base_internal_url,
                             token,
                             locks_to_reload,
-                            occupancy_metrics,
                         ))
                         .await?;
                         nmodified_ids.extend(inner_modified_ids);
@@ -898,7 +879,6 @@ async fn lock_modules<'c>(
                         default,
                         job,
                         mem_peak,
-                        canceled_by,
                         job_dir,
                         db,
                         tx,
@@ -908,7 +888,6 @@ async fn lock_modules<'c>(
                         base_internal_url,
                         token,
                         locks_to_reload,
-                        occupancy_metrics,
                     ))
                     .await?;
                     e.value = FlowModuleValue::BranchOne {
@@ -947,7 +926,6 @@ async fn lock_modules<'c>(
             &language,
             &content,
             mem_peak,
-            canceled_by,
             job_dir,
             db,
             worker_name,
@@ -961,7 +939,6 @@ async fn lock_modules<'c>(
             ),
             false,
             None,
-            occupancy_metrics,
         )
         .await;
         //
@@ -1317,7 +1294,6 @@ async fn lock_modules_app(
     value: Value,
     job: &QueuedJob,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
@@ -1325,7 +1301,6 @@ async fn lock_modules_app(
     job_path: &str,
     base_internal_url: &str,
     token: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> Result<Value> {
     match value {
         Value::Object(mut m) => {
@@ -1359,7 +1334,6 @@ async fn lock_modules_app(
                                 &language,
                                 &content,
                                 mem_peak,
-                                canceled_by,
                                 job_dir,
                                 db,
                                 worker_name,
@@ -1370,7 +1344,6 @@ async fn lock_modules_app(
                                 &format!("{}/app", job.script_path()),
                                 false,
                                 None,
-                                occupancy_metrics,
                             )
                             .await;
                             match new_lock {
@@ -1420,7 +1393,6 @@ async fn lock_modules_app(
                         b,
                         job,
                         mem_peak,
-                        canceled_by,
                         job_dir,
                         db,
                         worker_name,
@@ -1428,7 +1400,6 @@ async fn lock_modules_app(
                         job_path,
                         base_internal_url,
                         token,
-                        occupancy_metrics,
                     )
                     .await?,
                 );
@@ -1443,7 +1414,6 @@ async fn lock_modules_app(
                         b,
                         job,
                         mem_peak,
-                        canceled_by,
                         job_dir,
                         db,
                         worker_name,
@@ -1451,7 +1421,6 @@ async fn lock_modules_app(
                         job_path,
                         base_internal_url,
                         token,
-                        occupancy_metrics,
                     )
                     .await?,
                 );
@@ -1465,14 +1434,12 @@ async fn lock_modules_app(
 pub async fn handle_app_dependency_job(
     job: &QueuedJob,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
     worker_dir: &str,
     base_internal_url: &str,
     token: &str,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<()> {
     let job_path = job.script_path.clone().ok_or_else(|| {
         error::Error::InternalErr(
@@ -1495,7 +1462,6 @@ pub async fn handle_app_dependency_job(
             value,
             job,
             mem_peak,
-            canceled_by,
             job_dir,
             db,
             worker_name,
@@ -1503,7 +1469,6 @@ pub async fn handle_app_dependency_job(
             &job_path,
             base_internal_url,
             token,
-            occupancy_metrics,
         )
         .await?;
 
@@ -1587,13 +1552,11 @@ async fn python_dep(
     reqs: String,
     job_id: &Uuid,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
     w_id: &str,
     worker_dir: &str,
-    occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
     no_uv_compile: bool,
     no_uv_install: bool,
 ) -> std::result::Result<String, Error> {
@@ -1602,12 +1565,10 @@ async fn python_dep(
         job_id,
         &reqs,
         mem_peak,
-        canceled_by,
         job_dir,
         db,
         worker_name,
         w_id,
-        occupancy_metrics,
         no_uv_compile,
         false,
     )
@@ -1619,12 +1580,10 @@ async fn python_dep(
             job_id,
             w_id,
             mem_peak,
-            canceled_by,
             db,
             worker_name,
             job_dir,
             worker_dir,
-            occupancy_metrics,
             no_uv_install,
             false,
         )
@@ -1645,7 +1604,6 @@ async fn capture_dependency_job(
     job_language: &ScriptLang,
     job_raw_code: &str,
     mem_peak: &mut i32,
-    canceled_by: &mut Option<CanceledBy>,
     job_dir: &str,
     db: &sqlx::Pool<sqlx::Postgres>,
     worker_name: &str,
@@ -1656,7 +1614,6 @@ async fn capture_dependency_job(
     script_path: &str,
     raw_deps: bool,
     npm_mode: Option<bool>,
-    occupancy_metrics: &mut OccupancyMetrics,
 ) -> error::Result<String> {
     match job_language {
         ScriptLang::Python3 => {
@@ -1705,13 +1662,11 @@ async fn capture_dependency_job(
                     reqs,
                     job_id,
                     mem_peak,
-                    canceled_by,
                     job_dir,
                     db,
                     worker_name,
                     w_id,
                     worker_dir,
-                    &mut Some(occupancy_metrics),
                     no_uv_compile | no_uv,
                     no_uv_install | no_uv,
                 )
@@ -1753,13 +1708,11 @@ async fn capture_dependency_job(
                     reqs,
                     job_id,
                     mem_peak,
-                    canceled_by,
                     job_dir,
                     db,
                     worker_name,
                     w_id,
                     worker_dir,
-                    &mut Some(occupancy_metrics),
                     false,
                     false,
                 )
@@ -1776,7 +1729,6 @@ async fn capture_dependency_job(
                 job_id,
                 job_raw_code,
                 mem_peak,
-                canceled_by,
                 job_dir,
                 db,
                 false,
@@ -1784,7 +1736,6 @@ async fn capture_dependency_job(
                 false,
                 worker_name,
                 w_id,
-                occupancy_metrics,
             )
             .await
         }
@@ -1798,13 +1749,11 @@ async fn capture_dependency_job(
                 job_id,
                 job_raw_code,
                 mem_peak,
-                canceled_by,
                 job_dir,
                 Some(db),
                 w_id,
                 worker_name,
                 base_internal_url,
-                &mut Some(occupancy_metrics),
             )
             .await
         }
@@ -1817,7 +1766,6 @@ async fn capture_dependency_job(
             }
             let req = gen_bun_lockfile(
                 mem_peak,
-                canceled_by,
                 job_id,
                 w_id,
                 Some(db),
@@ -1833,7 +1781,6 @@ async fn capture_dependency_job(
                     None
                 },
                 npm_mode,
-                &mut Some(occupancy_metrics),
             )
             .await?;
             if req.is_some() && !raw_deps {
@@ -1848,7 +1795,6 @@ async fn capture_dependency_job(
                     base_internal_url,
                     worker_name,
                     &token,
-                    &mut Some(occupancy_metrics),
                 )
                 .await?;
             }
@@ -1875,19 +1821,7 @@ async fn capture_dependency_job(
                         }
                     }
                 };
-                composer_install(
-                    mem_peak,
-                    canceled_by,
-                    job_id,
-                    w_id,
-                    db,
-                    job_dir,
-                    worker_name,
-                    reqs,
-                    None,
-                    occupancy_metrics,
-                )
-                .await
+                composer_install(mem_peak, job_id, w_id, db, job_dir, worker_name, reqs, None).await
             }
         }
         ScriptLang::Rust => {
@@ -1907,12 +1841,10 @@ async fn capture_dependency_job(
                 job_id,
                 job_raw_code,
                 mem_peak,
-                canceled_by,
                 job_dir,
                 db,
                 worker_name,
                 w_id,
-                occupancy_metrics,
             )
             .await?;
 
@@ -1930,12 +1862,10 @@ async fn capture_dependency_job(
                 job_id,
                 job_raw_code,
                 mem_peak,
-                canceled_by,
                 job_dir,
                 db,
                 worker_name,
                 w_id,
-                occupancy_metrics,
             )
             .await
         }