starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add toggle to optionally mute error handler for cancelled jobs (#2567)

Browse Source 
* fix: Fix canceled flag in completed_job table

* Add toggle to workspace settings
This commit is contained in:
Guillaume Bouvignies
2023-11-06 18:24:51 +01:00
committed by GitHub
parent 1ed52ab4c9
commit 83f9ef34e6
33 changed files with 245 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json
Generated
-1
View File
@@ -42,7 +42,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-1730f39fd1793d45fbb41b21389c61296a3ff7489ae12f52a19f9543173ac597.json
Generated
+7 -1
View File
@@ -77,6 +77,11 @@
"ordinal": 14,
"name": "error_handler_extra_args",
"type_info": "Json"
},
{
"ordinal": 15,
"name": "error_handler_muted_on_cancel",
"type_info": "Bool"
}
],
"parameters": {
@@ -99,7 +104,8 @@
true,
true,
false,
true
true,
false
]
},
"hash": "1730f39fd1793d45fbb41b21389c61296a3ff7489ae12f52a19f9543173ac597"
backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json
Generated
-1
View File
@@ -67,7 +67,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json
Generated
-1
View File
@@ -28,7 +28,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-5445083864b2b092b012e894bff7630a1d7b9deb8d33e9f909061f351f96844e.json
Generated
+7 -1
View File
@@ -77,6 +77,11 @@
"ordinal": 14,
"name": "error_handler_extra_args",
"type_info": "Json"
},
{
"ordinal": 15,
"name": "error_handler_muted_on_cancel",
"type_info": "Bool"
}
],
"parameters": {
@@ -99,7 +104,8 @@
true,
true,
false,
true
true,
false
]
},
"hash": "5445083864b2b092b012e894bff7630a1d7b9deb8d33e9f909061f351f96844e"
backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json
Generated
-1
View File
@@ -60,7 +60,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-8b221f0d08f3304364e56f5c4894fea42975b03d21904655d724a35163413d29.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue SET mem_peak = $1, last_ping = now() WHERE id = $2 RETURNING canceled",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "canceled",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Int4",
"Uuid"
]
},
"nullable": [
false
]
},
"hash": "8b221f0d08f3304364e56f5c4894fea42975b03d21904655d724a35163413d29"
}
backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json
Generated
-1
View File
@@ -40,7 +40,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json
Generated
-1
View File
@@ -46,7 +46,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json
Generated
-1
View File
@@ -42,7 +42,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json
Generated
-1
View File
@@ -42,7 +42,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-ea2ad5213bb6de8eab502fbb76149fa7964cd11e74e5ed1a30e857aad8ea5595.json → backend/.sqlx/query-d21e5be1ac26db926b7196316dae0c9fa82a865b95bb55de5370e4ee34889ba3.json
Generated
+3 -2
View File
@@ -1,16 +1,17 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE workspace_settings SET error_handler = $1, error_handler_extra_args = $2 WHERE workspace_id = $3",
"query": "UPDATE workspace_settings SET error_handler = $1, error_handler_extra_args = $2, error_handler_muted_on_cancel = $3 WHERE workspace_id = $4",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Json",
"Bool",
"Text"
]
},
"nullable": []
},
"hash": "ea2ad5213bb6de8eab502fbb76149fa7964cd11e74e5ed1a30e857aad8ea5595"
"hash": "d21e5be1ac26db926b7196316dae0c9fa82a865b95bb55de5370e4ee34889ba3"
}
backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json
Generated
-1
View File
@@ -37,7 +37,6 @@
"bash",
"postgresql",
"nativets",
"Nativets",
"bun",
"mysql",
"bigquery",
backend/.sqlx/query-960dce02871e9697d7aca2fcd5a75eefc3db7179e49fbf20c7e62cbcb235683e.json → backend/.sqlx/query-f1f7a27738deae46de6acc10cbff701ff0c3872cc5b387e304bd9a058bb0c8a8.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 RETURNING id",
"query": "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 RETURNING id",
"describe": {
"columns": [
{
@@ -21,5 +21,5 @@
false
]
},
"hash": "960dce02871e9697d7aca2fcd5a75eefc3db7179e49fbf20c7e62cbcb235683e"
"hash": "f1f7a27738deae46de6acc10cbff701ff0c3872cc5b387e304bd9a058bb0c8a8"
}
backend/migrations/20231106131265_error_handler_skipped_on_cancel.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add down migration script here
ALTER TABLE workspace_settings DROP COLUMN error_handler_muted_on_cancel;
backend/migrations/20231106131265_error_handler_skipped_on_cancel.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE workspace_settings ADD COLUMN IF NOT EXISTS error_handler_muted_on_cancel BOOL NOT NULL DEFAULT false;
backend/openapi-bundled.yaml
View File
backend/openapi-deref.yaml
View File
backend/src/monitor.rs
+1
View File
@@ -714,6 +714,7 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
&client,
&job,
0,
None,
error::Error::ExecutionErr(format!(
"Job timed out after no ping from job since {} (ZOMBIE_JOB_TIMEOUT: {})",
last_ping
backend/windmill-api/openapi-deref.yaml
+10 -1
View File
@@ -1,6 +1,6 @@
openapi: 3.0.3
info:
version: 1.197.1
version: 1.200.0
title: Windmill API
contact:
name: Windmill Team
@@ -1487,6 +1487,8 @@ paths:
error_handler_extra_args:
type: object
additionalProperties: &ref_14 {}
error_handler_muted_on_cancel:
type: boolean
required:
- code_completion_enabled
/w/{workspace}/workspaces/get_deploy_to:
@@ -1763,6 +1765,8 @@ paths:
error_handler_extra_args:
type: object
additionalProperties: *ref_14
error_handler_muted_on_cancel:
type: boolean
responses:
'200':
description: status
@@ -6364,6 +6368,8 @@ paths:
- code
- identity
- http
dedicated_worker:
type: boolean
required: &ref_122
- args
responses:
@@ -8768,6 +8774,8 @@ paths:
type: string
worker_group:
type: string
wm_version:
type: string
required: &ref_137
- worker
- worker_instance
@@ -8776,6 +8784,7 @@ paths:
- ip
- jobs_executed
- worker_group
- wm_version
/configs/list_worker_groups:
get:
summary: list worker groups
backend/windmill-api/openapi.yaml
+25 -42
View File
@@ -1174,6 +1174,8 @@ paths:
type: string
error_handler_extra_args:
$ref: "#/components/schemas/ScriptArgs"
error_handler_muted_on_cancel:
type: boolean
required:
- code_completion_enabled
@@ -1437,6 +1439,8 @@ paths:
type: string
error_handler_extra_args:
$ref: "#/components/schemas/ScriptArgs"
error_handler_muted_on_cancel:
type: boolean
responses:
"200":
@@ -2108,8 +2112,7 @@ paths:
/w/{workspace}/resources/get_value_interpolated/{path}:
get:
summary:
get resource interpolated (variables and resources are fully unrolled)
summary: get resource interpolated (variables and resources are fully unrolled)
operationId: getResourceValueInterpolated
tags:
- resource
@@ -2797,8 +2800,7 @@ paths:
schema:
type: string
- name: first_parent_hash
description:
mask to filter scripts whom first direct parent has exact hash
description: mask to filter scripts whom first direct parent has exact hash
in: query
schema:
type: string
@@ -2993,8 +2995,7 @@ paths:
/workers/custom_tags:
get:
summary:
get all instance custom tags (tags are used to dispatch jobs to
summary: get all instance custom tags (tags are used to dispatch jobs to
different worker groups)
operationId: getCustomTags
tags:
@@ -3045,8 +3046,7 @@ paths:
/w/{workspace}/scripts/delete/h/{hash}:
post:
summary:
delete script by hash (erase content but keep hash, require admin)
summary: delete script by hash (erase content but keep hash, require admin)
operationId: deleteScriptByHash
tags:
- script
@@ -3238,16 +3238,14 @@ paths:
type: string
format: date-time
- name: scheduled_in_secs
description:
schedule the script to execute in the number of seconds starting now
description: schedule the script to execute in the number of seconds starting now
in: query
schema:
type: integer
- $ref: "#/components/parameters/ParentJob"
- $ref: "#/components/parameters/NewJobId"
- name: invisible_to_owner
description:
make the run invisible to the the script owner (default false)
description: make the run invisible to the the script owner (default false)
in: query
schema:
type: boolean
@@ -4220,8 +4218,7 @@ paths:
type: string
format: date-time
- name: scheduled_in_secs
description:
schedule the script to execute in the number of seconds starting now
description: schedule the script to execute in the number of seconds starting now
in: query
schema:
type: integer
@@ -4229,8 +4226,7 @@ paths:
- $ref: "#/components/parameters/NewJobId"
- $ref: "#/components/parameters/IncludeHeader"
- name: invisible_to_owner
description:
make the run invisible to the the flow owner (default false)
description: make the run invisible to the the flow owner (default false)
in: query
schema:
type: boolean
@@ -4282,8 +4278,7 @@ paths:
type: string
format: date-time
- name: scheduled_in_secs
description:
schedule the script to execute in the number of seconds starting now
description: schedule the script to execute in the number of seconds starting now
in: query
schema:
type: integer
@@ -4291,8 +4286,7 @@ paths:
- $ref: "#/components/parameters/NewJobId"
- $ref: "#/components/parameters/IncludeHeader"
- name: invisible_to_owner
description:
make the run invisible to the the flow owner (default false)
description: make the run invisible to the the flow owner (default false)
in: query
schema:
type: boolean
@@ -4330,8 +4324,7 @@ paths:
type: string
format: date-time
- name: scheduled_in_secs
description:
schedule the script to execute in the number of seconds starting now
description: schedule the script to execute in the number of seconds starting now
in: query
schema:
type: integer
@@ -4339,8 +4332,7 @@ paths:
- $ref: "#/components/parameters/NewJobId"
- $ref: "#/components/parameters/IncludeHeader"
- name: invisible_to_owner
description:
make the run invisible to the the script owner (default false)
description: make the run invisible to the the script owner (default false)
in: query
schema:
type: boolean
@@ -4371,8 +4363,7 @@ paths:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/IncludeHeader"
- name: invisible_to_owner
description:
make the run invisible to the the script owner (default false)
description: make the run invisible to the the script owner (default false)
in: query
schema:
type: boolean
@@ -4405,8 +4396,7 @@ paths:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/IncludeHeader"
- name: invisible_to_owner
description:
make the run invisible to the the script owner (default false)
description: make the run invisible to the the script owner (default false)
in: query
schema:
type: boolean
@@ -4621,8 +4611,7 @@ paths:
- job
responses:
"200":
description:
the timestamp of the db that can be used to compute the drift
description: the timestamp of the db that can be used to compute the drift
content:
application/json:
schema:
@@ -4858,8 +4847,7 @@ paths:
/w/{workspace}/jobs/resume_urls/{id}/{resume_id}:
get:
summary:
get resume urls given a job_id, resume_id and a nonce to resume a flow
summary: get resume urls given a job_id, resume_id and a nonce to resume a flow
operationId: getResumeUrls
tags:
- job
@@ -5372,8 +5360,7 @@ paths:
- $ref: "#/components/parameters/WorkspaceId"
- name: only_member_of
in: query
description:
only list the groups the user is member of (default false)
description: only list the groups the user is member of (default false)
schema:
type: boolean
responses:
@@ -5561,8 +5548,7 @@ paths:
- $ref: "#/components/parameters/WorkspaceId"
- name: only_member_of
in: query
description:
only list the folders the user is member of (default false)
description: only list the folders the user is member of (default false)
schema:
type: boolean
responses:
@@ -6280,8 +6266,7 @@ components:
type: integer
PerPage:
name: per_page
description:
number of items to return for a given page (default 30, max 100)
description: number of items to return for a given page (default 30, max 100)
in: query
schema:
type: integer
@@ -6415,8 +6400,7 @@ components:
type: boolean
ArgsFilter:
name: args
description:
filter on jobs containing those args as a json subset (@> in postgres)
description: filter on jobs containing those args as a json subset (@> in postgres)
in: query
schema:
type: string
@@ -6428,8 +6412,7 @@ components:
type: string
ResultFilter:
name: result
description:
filter on jobs containing those result as a json subset (@> in postgres)
description: filter on jobs containing those result as a json subset (@> in postgres)
in: query
schema:
type: string
backend/windmill-api/src/workspaces.rs
+4 -1
View File
@@ -134,6 +134,7 @@ pub struct WorkspaceSettings {
pub code_completion_enabled: bool,
pub error_handler: Option<String>,
pub error_handler_extra_args: Option<serde_json::Value>,
pub error_handler_muted_on_cancel: Option<bool>,
}
#[derive(FromRow, Serialize, Debug)]
@@ -246,6 +247,7 @@ pub struct NewWorkspaceUser {
pub struct EditErrorHandler {
pub error_handler: Option<String>,
pub error_handler_extra_args: Option<serde_json::Value>,
pub error_handler_muted_on_cancel: Option<bool>,
}
async fn list_pending_invites(
@@ -833,9 +835,10 @@ async fn edit_error_handler(
if let Some(error_handler) = &ee.error_handler {
sqlx::query!(
"UPDATE workspace_settings SET error_handler = $1, error_handler_extra_args = $2 WHERE workspace_id = $3",
"UPDATE workspace_settings SET error_handler = $1, error_handler_extra_args = $2, error_handler_muted_on_cancel = $3 WHERE workspace_id = $4",
error_handler,
ee.error_handler_extra_args,
ee.error_handler_muted_on_cancel.unwrap_or(false),
&w_id
)
.execute(&mut *tx)
backend/windmill-queue/src/jobs.rs
+29 -14
View File
@@ -107,6 +107,12 @@ const ERROR_HANDLER_USERNAME: &str = "error_handler";
const ERROR_HANDLER_USER_GROUP: &str = "g/error_handler";
const ERROR_HANDLER_USER_EMAIL: &str = "error_handler@windmill.dev";
#[derive(Clone, Debug)]
pub struct CanceledBy {
pub username: Option<String>,
pub reason: Option<String>,
}
#[async_recursion]
pub async fn cancel_job<'c: 'async_recursion>(
username: &str,
@@ -130,14 +136,14 @@ pub async fn cancel_job<'c: 'async_recursion>(
&& !force_cancel
{
let id = sqlx::query_scalar!(
"UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 RETURNING id",
username,
reason,
id,
w_id
)
.fetch_optional(&mut *tx)
.await?;
"UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 RETURNING id",
username,
reason,
id,
w_id
)
.fetch_optional(&mut *tx)
.await?;
if let Some(id) = id {
tracing::info!("Soft cancelling job {}", id);
}
@@ -151,6 +157,7 @@ pub async fn cancel_job<'c: 'async_recursion>(
&job_running,
format!("canceled by {username}: (force cancel: {force_cancel})"),
job_running.mem_peak.unwrap_or(0),
Some(CanceledBy { username: Some(username.to_string()), reason: Some(reason) }),
e,
rsmq.clone(),
"server",
@@ -238,6 +245,7 @@ pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Sen
queued_job: &QueuedJob,
logs: String,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
e: serde_json::Value,
rsmq: Option<R>,
worker_name: &str,
@@ -274,6 +282,7 @@ pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Sen
Json(&result),
logs,
mem_peak,
canceled_by,
rsmq,
)
.await?;
@@ -314,6 +323,7 @@ pub async fn add_completed_job<
result: Json<&T>,
logs: String,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
rsmq: Option<R>,
) -> Result<Uuid, Error> {
// tracing::error!("Start");
@@ -402,9 +412,9 @@ pub async fn add_completed_job<
logs,
queued_job.raw_code,
queued_job.raw_lock,
queued_job.canceled,
queued_job.canceled_by,
queued_job.canceled_reason,
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,
@@ -514,7 +524,7 @@ pub async fn add_completed_job<
);
}
if let Err(e) = send_error_to_workspace_handler(rsmq.clone(), &queued_job, db, result).await
if let Err(e) = send_error_to_workspace_handler(rsmq.clone(), &queued_job, canceled_by.is_some(), db, result).await
{
tracing::error!(
"Could not run workspace error handler for job {}: {}",
@@ -670,19 +680,24 @@ pub async fn send_error_to_workspace_handler<
>(
rsmq: Option<R>,
queued_job: &QueuedJob,
is_canceled: bool,
db: &Pool<Postgres>,
result: Json<&'a T>,
) -> Result<(), Error> {
let w_id = &queued_job.workspace_id;
let mut tx = db.begin().await?;
let (error_handler, error_handler_extra_args) = sqlx::query_as::<_, (Option<String>, Option<serde_json::Value>)>(
"SELECT error_handler, error_handler_extra_args FROM workspace_settings WHERE workspace_id = $1",
let (error_handler, error_handler_extra_args, error_handler_muted_on_cancel) = sqlx::query_as::<_, (Option<String>, Option<serde_json::Value>, bool)>(
"SELECT error_handler, error_handler_extra_args, error_handler_muted_on_cancel FROM workspace_settings WHERE workspace_id = $1",
).bind(&w_id)
.fetch_optional(&mut *tx)
.await
.context("sending error to global handler")?
.ok_or_else(|| Error::InternalErr(format!("no workspace settings for id {w_id}")))?;
if is_canceled && error_handler_muted_on_cancel {
return Ok(());
}
if let Some(error_handler) = error_handler {
let ws_error_handler_muted: Option<bool> = match queued_job.job_kind {
JobKind::Script => {
backend/windmill-worker/src/bash_executor.rs
+5
View File
@@ -6,6 +6,7 @@ use serde_json::{json, value::RawValue};
use sqlx::types::Json;
use tokio::process::Command;
use windmill_common::{error::Error, jobs::QueuedJob, worker::to_raw_value};
use windmill_queue::CanceledBy;
const BIN_BASH: &str = "/bin/bash";
const NSJAIL_CONFIG_RUN_BASH_CONTENT: &str = include_str!("../nsjail/run.bash.config.proto");
@@ -28,6 +29,7 @@ lazy_static::lazy_static! {
pub async fn handle_bash_job(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
@@ -115,6 +117,7 @@ pub async fn handle_bash_job(
db,
logs,
mem_peak,
canceled_by,
child,
!*DISABLE_NSJAIL,
worker_name,
@@ -160,6 +163,7 @@ fn raw_to_string(x: &str) -> String {
pub async fn handle_powershell_job(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
@@ -253,6 +257,7 @@ pub async fn handle_powershell_job(
db,
logs,
mem_peak,
canceled_by,
child,
!*DISABLE_NSJAIL,
worker_name,
backend/windmill-worker/src/bun_executor.rs
+13
View File
@@ -5,6 +5,7 @@ use itertools::Itertools;
use regex::Regex;
use serde_json::value::RawValue;
use uuid::Uuid;
use windmill_queue::CanceledBy;
#[cfg(feature = "enterprise")]
use crate::common::build_envs_map;
@@ -53,6 +54,7 @@ lazy_static::lazy_static! {
pub async fn gen_lockfile(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_id: &Uuid,
w_id: &str,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -100,6 +102,7 @@ pub async fn gen_lockfile(
db,
logs,
mem_peak,
canceled_by,
child_process,
false,
worker_name,
@@ -144,6 +147,7 @@ pub async fn gen_lockfile(
install_lockfile(
logs,
mem_peak,
canceled_by,
job_id,
w_id,
db,
@@ -180,6 +184,7 @@ pub async fn gen_lockfile(
pub async fn install_lockfile(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_id: &Uuid,
w_id: &str,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -202,6 +207,7 @@ pub async fn install_lockfile(
db,
logs,
mem_peak,
canceled_by,
child_process,
false,
worker_name,
@@ -236,6 +242,7 @@ pub async fn handle_bun_job(
requirements_o: Option<String>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
@@ -279,6 +286,7 @@ pub async fn handle_bun_job(
install_lockfile(
logs,
mem_peak,
canceled_by,
&job.id,
&job.workspace_id,
db,
@@ -302,6 +310,7 @@ pub async fn handle_bun_job(
let _ = gen_lockfile(
logs,
mem_peak,
canceled_by,
&job.id,
&job.workspace_id,
db,
@@ -489,6 +498,7 @@ plugin(p)
db,
logs,
mem_peak,
canceled_by,
child,
false,
worker_name,
@@ -549,6 +559,7 @@ pub async fn start_worker(
) -> Result<()> {
let mut logs = "".to_string();
let mut mem_peak: i32 = 0;
let mut canceled_by: Option<CanceledBy> = None;
let _ = write_file(job_dir, "main.ts", inner_content).await?;
let common_bun_proc_envs: HashMap<String, String> =
get_common_bun_proc_envs(&base_internal_url).await;
@@ -595,6 +606,7 @@ pub async fn start_worker(
install_lockfile(
&mut logs,
&mut mem_peak,
&mut canceled_by,
&Uuid::nil(),
&w_id,
db,
@@ -614,6 +626,7 @@ pub async fn start_worker(
let _ = gen_lockfile(
&mut logs,
&mut mem_peak,
&mut canceled_by,
&Uuid::nil(),
&w_id,
db,
backend/windmill-worker/src/common.rs
+14 -6
View File
@@ -17,6 +17,7 @@ use windmill_common::{
};
use anyhow::Result;
use windmill_queue::CanceledBy;
use std::{
borrow::Borrow,
@@ -422,6 +423,7 @@ pub async fn handle_child(
db: &Pool<Postgres>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by_ref: &mut Option<CanceledBy>,
mut child: Child,
nsjail: bool,
worker_name: &str,
@@ -484,15 +486,21 @@ pub async fn handle_child(
*mem_peak = current_mem
}
tracing::info!("{worker_name}/{job_id} in {_w_id} still running. mem: {current_mem}kB, peak mem: {mem_peak}kB");
if sqlx::query_scalar!("UPDATE queue SET mem_peak = $1, last_ping = now() WHERE id = $2 RETURNING canceled", *mem_peak, job_id)
let (canceled, canceled_by, canceled_reason) = sqlx::query_as::<_, (bool, Option<String>, Option<String>)>("UPDATE queue SET mem_peak = $1, last_ping = now() WHERE id = $2 RETURNING canceled, canceled_by, canceled_reason")
.bind(*mem_peak)
.bind(job_id)
.fetch_optional(&db)
.await
.map(|v| Some(true) == v)
.unwrap_or_else(|err| {
tracing::error!(%job_id, %err, "error checking cancelation for job {job_id}: {err}");
false
.unwrap_or_else(|e| {
tracing::error!(%e, "error updating job {job_id}: {e}");
Some((false, None, None))
})
{
.unwrap_or((false, None, None));
if canceled {
canceled_by_ref.replace(CanceledBy {
username: canceled_by.clone(),
reason: canceled_reason.clone(),
});
break;
}
},
backend/windmill-worker/src/dedicated_worker.rs
+2 -2
View File
@@ -142,10 +142,10 @@ pub async fn handle_dedicated_process(
let job: Arc<QueuedJob> = jobs.pop_front().expect("pop");
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&line) {
Ok(result) => job_completed_tx.send(JobCompleted { job , result, logs: "".to_string(), mem_peak: 0, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap(),
Ok(result) => job_completed_tx.send(JobCompleted { job , result, logs: "".to_string(), mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap(),
Err(e) => {
tracing::error!("Could not deserialize job result `{line}`: {e:?}");
job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), logs: "".to_string(), mem_peak: 0, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap();
job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), logs: "".to_string(), mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap();
},
};
} else {
backend/windmill-worker/src/deno_executor.rs
+5
View File
@@ -3,6 +3,7 @@ use std::{collections::HashMap, process::Stdio};
use itertools::Itertools;
use serde_json::value::RawValue;
use uuid::Uuid;
use windmill_queue::CanceledBy;
use crate::{
common::{
@@ -75,6 +76,7 @@ pub async fn generate_deno_lock(
code: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
w_id: &str,
@@ -120,6 +122,7 @@ pub async fn generate_deno_lock(
db,
logs,
mem_peak,
canceled_by,
child_process,
false,
worker_name,
@@ -142,6 +145,7 @@ pub async fn handle_deno_job(
requirements_o: Option<String>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
@@ -330,6 +334,7 @@ run().catch(async (e) => {{
db,
logs,
mem_peak,
canceled_by,
child,
false,
worker_name,
backend/windmill-worker/src/go_executor.rs
+8
View File
@@ -14,6 +14,7 @@ use windmill_common::{
utils::calculate_hash,
};
use windmill_parser_go::{parse_go_imports, REQUIRE_PARSE};
use windmill_queue::CanceledBy;
use crate::{
common::{
@@ -35,6 +36,7 @@ lazy_static::lazy_static! {
pub async fn handle_go_job(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
@@ -79,6 +81,7 @@ pub async fn handle_go_job(
inner_content,
logs,
mem_peak,
canceled_by,
job_dir,
db,
true,
@@ -195,6 +198,7 @@ func Run(req Req) (interface{{}}, error){{
db,
logs,
mem_peak,
canceled_by,
build_go_process,
false,
worker_name,
@@ -275,6 +279,7 @@ func Run(req Req) (interface{{}}, error){{
db,
logs,
mem_peak,
canceled_by,
child,
!*DISABLE_NSJAIL,
worker_name,
@@ -313,6 +318,7 @@ pub async fn install_go_dependencies(
code: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
non_dep_job: bool,
@@ -336,6 +342,7 @@ pub async fn install_go_dependencies(
db,
logs,
mem_peak,
canceled_by,
child_process,
false,
worker_name,
@@ -400,6 +407,7 @@ pub async fn install_go_dependencies(
db,
logs,
mem_peak,
canceled_by,
child_process,
false,
worker_name,
backend/windmill-worker/src/python_executor.rs
+13
View File
@@ -17,6 +17,7 @@ use windmill_common::{
worker::WORKER_CONFIG,
DB,
};
use windmill_queue::CanceledBy;
lazy_static::lazy_static! {
static ref PYTHON_PATH: String =
@@ -67,6 +68,7 @@ pub async fn pip_compile(
requirements: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &Pool<Postgres>,
worker_name: &str,
@@ -132,6 +134,7 @@ pub async fn pip_compile(
db,
logs,
mem_peak,
canceled_by,
child_process,
false,
worker_name,
@@ -169,6 +172,7 @@ pub async fn handle_python_job(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
db: &sqlx::Pool<sqlx::Postgres>,
client: &AuthedClientBackgroundTask,
inner_content: &String,
@@ -189,6 +193,7 @@ pub async fn handle_python_job(
worker_dir,
logs,
mem_peak,
canceled_by,
)
.await?;
@@ -352,6 +357,7 @@ mount {{
db,
logs,
mem_peak,
canceled_by,
child,
!*DISABLE_NSJAIL,
worker_name,
@@ -502,6 +508,7 @@ async fn handle_python_deps(
worker_dir: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
) -> error::Result<Vec<String>> {
create_dependencies_dir(job_dir).await;
@@ -532,6 +539,7 @@ async fn handle_python_deps(
&requirements,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -555,6 +563,7 @@ async fn handle_python_deps(
w_id,
logs,
mem_peak,
canceled_by,
db,
worker_name,
job_dir,
@@ -571,6 +580,7 @@ pub async fn handle_python_reqs(
w_id: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
job_dir: &str,
@@ -721,6 +731,7 @@ pub async fn handle_python_reqs(
db,
logs,
mem_peak,
canceled_by,
child,
false,
worker_name,
@@ -780,6 +791,7 @@ pub async fn start_worker(
) -> error::Result<()> {
let mut logs = "".to_string();
let mut mem_peak: i32 = 0;
let mut canceled_by: Option<CanceledBy> = None;
let context = variables::get_reserved_variables(
w_id,
&token,
@@ -809,6 +821,7 @@ pub async fn start_worker(
job_dir,
&mut logs,
&mut mem_peak,
&mut canceled_by,
)
.await?;
backend/windmill-worker/src/worker.rs
+61 -3
View File
@@ -41,8 +41,8 @@ use windmill_common::{
DB, IS_READY, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
};
use windmill_queue::{
canceled_job_to_result, empty_args, get_queued_job, pull, push, register_metric, PushArgs,
PushIsolationLevel, WrappedError, HTTP_CLIENT,
canceled_job_to_result, empty_args, get_queued_job, pull, push, register_metric, CanceledBy,
PushArgs, PushIsolationLevel, WrappedError, HTTP_CLIENT,
};
use serde_json::{json, value::RawValue, Value};
@@ -473,6 +473,7 @@ 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();
if let Err(err) = process_completed_job(
jc,
&client,
@@ -491,6 +492,7 @@ async fn handle_receive_completed_job<
&client,
job.as_ref(),
mem_peak,
canceled_by,
err,
false,
same_worker_tx.clone(),
@@ -1499,6 +1501,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
mem_peak: 0,
cached_res_path: None,
token: "".to_string(),
canceled_by: None,
})
.await
.expect("send job completed");
@@ -1619,6 +1622,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
&authed_client.get_authed().await,
arc_job.as_ref(),
0,
None,
err,
false,
same_worker_tx.clone(),
@@ -1786,7 +1790,16 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
// ) -> error::Result<()> {
pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
JobCompleted { job, result, logs, mem_peak, success, cached_res_path, .. }: JobCompleted,
JobCompleted {
job,
result,
logs,
mem_peak,
success,
cached_res_path,
canceled_by,
..
}: JobCompleted,
client: &AuthedClient,
db: &DB,
worker_dir: &str,
@@ -1813,6 +1826,7 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
Json(&result),
logs,
mem_peak.to_owned(),
canceled_by,
rsmq.clone(),
)
.await?;
@@ -1848,6 +1862,7 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
&job,
logs.to_string(),
mem_peak.to_owned(),
canceled_by,
serde_json::from_str(result.get()).unwrap_or_else(
|_| json!({ "message": format!("Non serializable error: {}", result.get()) }),
),
@@ -1919,6 +1934,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
client: &AuthedClient,
job: &QueuedJob,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
err: Error,
unrecoverable: bool,
same_worker_tx: Sender<Uuid>,
@@ -1938,6 +1954,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
job,
format!("Unexpected error during job execution:\n{err:#?}"),
mem_peak,
canceled_by.clone(),
err.clone(),
rsmq_2,
worker_name,
@@ -1986,6 +2003,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
&parent_job,
format!("Unexpected error during flow job error handling:\n{err}"),
mem_peak,
canceled_by.clone(),
e,
rsmq,
worker_name,
@@ -2021,6 +2039,7 @@ pub struct JobCompleted {
pub success: bool,
pub cached_res_path: Option<String>,
pub token: String,
pub canceled_by: Option<CanceledBy>,
}
pub async fn get_content(job: &QueuedJob, db: &Pool<Postgres>) -> Result<String, Error> {
@@ -2172,6 +2191,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
result,
logs,
mem_peak: 0,
canceled_by: None,
success: true,
cached_res_path: None,
token: authed_client.token,
@@ -2202,6 +2222,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
_ => {
let mut logs = "".to_string();
let mut mem_peak: i32 = 0;
let mut canceled_by: Option<CanceledBy> = None;
// println!("handle queue {:?}", SystemTime::now());
if let Some(log_str) = &job.logs {
logs.push_str(&log_str);
@@ -2234,6 +2255,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
&job,
&mut logs,
&mut mem_peak,
&mut canceled_by,
job_dir,
db,
worker_name,
@@ -2247,6 +2269,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
&job,
&mut logs,
&mut mem_peak,
&mut canceled_by,
job_dir,
db,
worker_name,
@@ -2260,6 +2283,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
&job,
&mut logs,
&mut mem_peak,
&mut canceled_by,
job_dir,
db,
worker_name,
@@ -2286,6 +2310,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
worker_dir,
&mut logs,
&mut mem_peak,
&mut canceled_by,
base_internal_url,
worker_name,
)
@@ -2306,6 +2331,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job_completed_tx,
logs,
mem_peak,
canceled_by,
cached_res_path,
client.get_token().await,
)
@@ -2322,6 +2348,7 @@ async fn process_result(
job_completed_tx: JobCompletedSender,
logs: String,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
cached_res_path: Option<String>,
token: String,
) -> error::Result<()> {
@@ -2333,6 +2360,7 @@ async fn process_result(
result: r,
logs,
mem_peak,
canceled_by,
success: true,
cached_res_path,
token: token,
@@ -2375,6 +2403,7 @@ async fn process_result(
result: to_raw_value(&error_value),
logs: logs,
mem_peak,
canceled_by,
success: false,
cached_res_path,
token: token,
@@ -2426,6 +2455,7 @@ async fn handle_code_execution_job(
worker_dir: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
base_internal_url: &str,
worker_name: &str,
) -> error::Result<Box<RawValue>> {
@@ -2564,6 +2594,7 @@ mount {{
job,
logs,
mem_peak,
canceled_by,
db,
client,
&inner_content,
@@ -2578,6 +2609,7 @@ mount {{
requirements_o,
logs,
mem_peak,
canceled_by,
job,
db,
client,
@@ -2594,6 +2626,7 @@ mount {{
requirements_o,
logs,
mem_peak,
canceled_by,
job,
db,
client,
@@ -2610,6 +2643,7 @@ mount {{
handle_go_job(
logs,
mem_peak,
canceled_by,
job,
db,
client,
@@ -2627,6 +2661,7 @@ mount {{
handle_bash_job(
logs,
mem_peak,
canceled_by,
job,
db,
client,
@@ -2643,6 +2678,7 @@ mount {{
handle_powershell_job(
logs,
mem_peak,
canceled_by,
job,
db,
client,
@@ -2676,6 +2712,7 @@ async fn handle_dependency_job(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
@@ -2696,6 +2733,7 @@ async fn handle_dependency_job(
.unwrap_or_else(|| "no raw code"),
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -2738,6 +2776,7 @@ async fn handle_flow_dependency_job(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
@@ -2762,6 +2801,7 @@ async fn handle_flow_dependency_job(
job,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -2803,6 +2843,7 @@ async fn lock_modules(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
@@ -2839,6 +2880,7 @@ async fn lock_modules(
job,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -2861,6 +2903,7 @@ async fn lock_modules(
job,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -2882,6 +2925,7 @@ async fn lock_modules(
job,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -2898,6 +2942,7 @@ async fn lock_modules(
job,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -2932,6 +2977,7 @@ async fn lock_modules(
&dependencies,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -2990,6 +3036,7 @@ async fn lock_modules_app(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
@@ -3033,6 +3080,7 @@ async fn lock_modules_app(
&dependencies,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -3073,6 +3121,7 @@ async fn lock_modules_app(
job,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -3095,6 +3144,7 @@ async fn lock_modules_app(
job,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -3116,6 +3166,7 @@ async fn handle_app_dependency_job(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
@@ -3144,6 +3195,7 @@ async fn handle_app_dependency_job(
job,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -3182,6 +3234,7 @@ async fn capture_dependency_job(
job_raw_code: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
@@ -3199,6 +3252,7 @@ async fn capture_dependency_job(
job_raw_code,
logs,
mem_peak,
canceled_by,
job_dir,
db,
worker_name,
@@ -3213,6 +3267,7 @@ async fn capture_dependency_job(
w_id,
logs,
mem_peak,
canceled_by,
db,
worker_name,
job_dir,
@@ -3236,6 +3291,7 @@ async fn capture_dependency_job(
job_raw_code,
logs,
mem_peak,
canceled_by,
job_dir,
db,
false,
@@ -3252,6 +3308,7 @@ async fn capture_dependency_job(
job_raw_code,
logs,
mem_peak,
canceled_by,
job_dir,
db,
w_id,
@@ -3267,6 +3324,7 @@ async fn capture_dependency_job(
let req = gen_lockfile(
logs,
mem_peak,
canceled_by,
job_id,
w_id,
db,
backend/windmill-worker/src/worker_flow.rs
+18 -2
View File
@@ -43,8 +43,8 @@ use windmill_common::{
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
};
use windmill_queue::{
add_completed_job, add_completed_job_error, handle_maybe_scheduled_job, PushIsolationLevel,
WrappedError,
add_completed_job, add_completed_job_error, handle_maybe_scheduled_job, CanceledBy,
PushIsolationLevel, WrappedError,
};
type DB = sqlx::Pool<sqlx::Postgres>;
@@ -607,6 +607,10 @@ pub async fn update_flow_status_after_job_completion_internal<
&flow_job,
logs,
0,
Some(CanceledBy {
username: flow_job.canceled_by.clone(),
reason: flow_job.canceled_reason.clone(),
}),
canceled_job_to_result(&flow_job),
rsmq.clone(),
worker_name,
@@ -642,6 +646,7 @@ pub async fn update_flow_status_after_job_completion_internal<
Json(&nresult),
logs,
0,
None,
rsmq.clone(),
)
.await?;
@@ -658,6 +663,7 @@ pub async fn update_flow_status_after_job_completion_internal<
),
logs,
0,
None,
rsmq.clone(),
)
.await?;
@@ -684,6 +690,7 @@ pub async fn update_flow_status_after_job_completion_internal<
&flow_job,
"Unexpected error during flow chaining:\n".to_string(),
0,
None,
e,
rsmq.clone(),
worker_name,
@@ -1352,6 +1359,14 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
let skipped = false;
let logs = "Timed out waiting to be resumed".to_string();
let result = json!({ "error": {"message": logs, "name": "SuspendedTimeout"}});
let canceled_by = if flow_job.canceled {
Some(CanceledBy {
username: flow_job.canceled_by.clone(),
reason: flow_job.canceled_reason.clone(),
})
} else {
None
};
let _uuid = add_completed_job(
db,
&flow_job,
@@ -1360,6 +1375,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
Json(&result),
logs,
0,
canceled_by,
rsmq,
)
.await?;
frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte
+14 -3
View File
@@ -48,6 +48,7 @@
let errorHandlerScriptPath: string
let errorHandlerItemKind: 'flow' | 'script' = 'script'
let errorHandlerExtraArgs: Record<string, any> = {}
let errorHandlerMutedOnCancel: boolean | undefined = undefined
let openaiResourceInitialPath: string | undefined = undefined
let codeCompletionEnabled: boolean = false
let tab =
@@ -166,6 +167,7 @@
openaiResourceInitialPath = settings.openai_resource_path
errorHandlerScriptPath = (settings.error_handler ?? '').split('/').slice(1).join('/')
errorHandlerInitialScriptPath = errorHandlerScriptPath
errorHandlerMutedOnCancel = settings.error_handler_muted_on_cancel
if (emptyString($enterpriseLicense)) {
errorHandlerSelected = 'custom'
} else {
@@ -190,7 +192,8 @@
workspace: $workspaceStore!,
requestBody: {
error_handler: `${errorHandlerItemKind}/${errorHandlerScriptPath}`,
error_handler_extra_args: errorHandlerExtraArgs
error_handler_extra_args: errorHandlerExtraArgs,
error_handler_muted_on_cancel: errorHandlerMutedOnCancel
}
})
sendUserToast(`workspace error handler set to ${errorHandlerScriptPath}`)
@@ -199,7 +202,8 @@
workspace: $workspaceStore!,
requestBody: {
error_handler: undefined,
error_handler_extra_args: undefined
error_handler_extra_args: undefined,
error_handler_muted_on_cancel: undefined
}
})
sendUserToast(`workspace error handler removed`)
@@ -490,7 +494,14 @@
</svelte:fragment>
</ErrorOrRecoveryHandler>
<div class="flex mt-5 justify-start">
<div class="flex flex-col mt-5 gap-5 items-start">
<Toggle
disabled={errorHandlerSelected === 'slack' &&
!emptyString(errorHandlerScriptPath) &&
emptyString(errorHandlerExtraArgs['channel'])}
bind:checked={errorHandlerMutedOnCancel}
options={{ right: 'Do not run error handler for canceled jobs' }}
/>
<Button
disabled={errorHandlerSelected === 'slack' &&
!emptyString(errorHandlerScriptPath) &&