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

fix: proper error handling in pulled job preprocessor (#7098)

Browse Source 
* fix: proper error handling in pulled job preprocessor

Signed-off-by: pyranota <pyra@duck.com>

* follow up for merge

Signed-off-by: pyranota <pyra@duck.com>

* make it safe

Signed-off-by: pyranota <pyra@duck.com>

* clippy

Signed-off-by: pyranota <pyra@duck.com>

* remove unused import

Signed-off-by: pyranota <pyra@duck.com>

* use String instead of Value

Signed-off-by: pyranota <pyra@duck.com>

* update ee ref

Signed-off-by: pyranota <pyra@duck.com>

* implement Error for PulledJobResultToErr

Signed-off-by: pyranota <pyra@duck.com>

* updatesqlx

Signed-off-by: pyranota <pyra@duck.com>

---------

Signed-off-by: pyranota <pyra@duck.com>
This commit is contained in:
Pyra
2025-11-10 13:43:50 +00:00
committed by GitHub
parent fc5034e94d
commit 84992cd8ff
12 changed files with 394 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
backend/.sqlx/query-83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO app_version\n (app_id, value, created_by, raw_app)\n SELECT app_id, value, created_by, raw_app\n FROM app_version WHERE id = $1\n RETURNING id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false
]
},
"hash": "83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604"
}
backend/.sqlx/query-f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2.json
Generated
+24
View File
@@ -0,0 +1,24 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow_version\n (workspace_id, path, value, schema, created_by)\n\n SELECT workspace_id, path, value, schema, created_by\n FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3\n\n RETURNING id\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Int8"
]
},
"nullable": [
false
]
},
"hash": "f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2"
}
backend/Cargo.lock
Generated
+1
View File
@@ -15725,6 +15725,7 @@ dependencies = [
"serde_urlencoded",
"sql-builder",
"sqlx",
"thiserror 2.0.17",
"tokio",
"tracing",
"ulid",
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
5b7afe50da442441747e7a8f6ef461c96faa9dc2
1cc227920a3799fe24a4135f1c6ae7026c0abcb5
backend/windmill-common/src/worker.rs
+8 -1
View File
@@ -7,7 +7,7 @@ use regex::Regex;
use reqwest_middleware::ClientWithMiddleware;
use semver::Version;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use serde_json::value::RawValue;
use serde_json::{json, value::RawValue};
use sqlx::{types::Json, Pool, Postgres};
use std::{
cmp::Reverse,
@@ -1701,6 +1701,13 @@ pub fn load_env_vars(
.collect()
}
pub fn error_to_value(err: &error::Error) -> serde_json::Value {
match err {
error::Error::JsonErr(err) => err.clone(),
_ => json!({"message": err.to_string(), "name": err.name()}),
}
}
#[derive(Clone, PartialEq, Debug)]
pub struct WorkspacedPath {
pub workspace_id: String,
backend/windmill-queue/Cargo.toml
+1
View File
@@ -46,3 +46,4 @@ serde_urlencoded.workspace = true
regex.workspace = true
backon.workspace = true
quick_cache.workspace = true
thiserror.workspace = true
backend/windmill-queue/src/jobs.rs
+285 -249
View File
@@ -2456,12 +2456,18 @@ pub struct PulledJobResult {
pub job: Option<PulledJob>,
pub suspended: bool,
pub missing_concurrency_key: bool,
pub error_while_preprocessing: Option<String>
}
#[derive(thiserror::Error, Debug)]
pub enum PulledJobResultToJobErr {
#[error("missing concurrency key")]
MissingConcurrencyKey(JobCompleted),
#[error("pulled job preprocessor error: {}", .0.result)]
ErrorWhilePreprocessing(JobCompleted),
}
impl PulledJobResult {
pub fn to_pulled_job(self) -> Result<Option<PulledJob>, PulledJobResultToJobErr> {
match self {
@@ -2484,9 +2490,278 @@ impl PulledJobResult {
from_cache: None,
}),
),
PulledJobResult { job: Some(job), error_while_preprocessing: Some(e), .. } => Err(
PulledJobResultToJobErr::ErrorWhilePreprocessing(JobCompleted {
preprocessed_args: None,
job: MiniCompletedJob::from(job.job),
success: false,
result: Arc::new(windmill_common::worker::to_raw_value(&json!({
"name": "Pulled job preprocessing error",
"message": e
}))),
result_columns: None,
mem_peak: 0,
cached_res_path: None,
token: "".to_string(),
canceled_by: None,
duration: None,
has_stream: Some(false),
from_cache: None,
}),
),
PulledJobResult { job, .. } => Ok(job),
}
}
/// Generic preprocess function
/// Can be used for any kind of preprocessing
pub async fn preprocess(&mut self, db: &DB) -> error::Result<()> {
let PulledJobResult { job: Some(ref mut pulled_job), .. } = self else {
return Ok(());
};
let kind = pulled_job.kind;
// Handle dependency job debouncing cleanup when a job is pulled for execution
if kind.is_dependency()
&& pulled_job
.args
.as_ref()
.map(|x| x.get("triggered_by_relative_import").is_some())
.unwrap_or_default()
&& !*WMDEBUG_NO_DJOB_DEBOUNCING
{
return Box::pin(async move {
// Only used for testing in tests/relative_imports.rs
// Give us some space to work with.
#[cfg(debug_assertions)]
if let Some(duration) = pulled_job
.args
.as_ref()
.map(|x| {
x.get("dbg_sleep_between_pull_and_debounce_key_removal")
.map(|v| serde_json::from_str::<u32>(v.get()).ok())
.flatten()
})
.flatten()
{
tracing::debug!("going to sleep",);
sleep(std::time::Duration::from_secs(duration as u64)).await;
}
tracing::debug!(
"Processing debounce cleanup for dependency job {} at path {:?}",
&pulled_job.id,
&pulled_job.runnable_path
);
let key = format!("{}:{}:dependency", &pulled_job.workspace_id, pulled_job.runnable_path());
let mut tx = db.begin().await?;
// === DEBOUNCE CLEANUP ===
//
// Clean up the debounce_key entry for this job (if it exists).
//
// IMPORTANT: We delete by key (not job_id) to avoid race conditions:
// If pusher has locked this row then this call will be blocked until all txs are commited.
//
// The idea is that the worker_lockfiles::trigger_dependents_to_recompute_locks will fetch the latest version of the obj.
// This object needs to be created before the djob is executed and it happens right here.
//
// This way the next pusher can fetch the latest version of object and base their djob payload on newest version.
// The concurrency limit on djobs will make sure that by the time next djob is started executing the base version it is referencing
// has already calculated all locks. This way even next djob will always use the fully finalized version of object.
//
//
//
// Note: We don't use a transaction here for performance (it's called during job pull).
// This means there's a tiny window where the job is running but key isn't deleted yet,
// which is acceptable because new requests will just accumulate data to this job.
tracing::debug!(
job_id = %pulled_job.id,
"Cleaning up debounce_key entry for completed/pulled job"
);
// This will either:
// 1. Block until pusher pushed. Which gives us:
// - If there was any stale data in pusher, then we will read it here (couple of lines below)
// 2. Block pusher until we are done here. This gives us:
// - We will clone objects and retrieve the latest version. So when we are done the pusher can read latest version.
sqlx::query!("DELETE FROM debounce_key WHERE key = $1", &key)
.execute(&mut *tx)
.await
.map_err(|e| {
tracing::error!(
error = %e,
job_id = %pulled_job.id,
"Failed to delete debounce_key"
);
e
})?;
let Some(base_hash) = pulled_job.runnable_id else {
return Err(Error::InternalErr(
"Missing runnable_id for dependency job triggered by relative import"
.to_string(),
));
};
tracing::debug!(
job_id = %pulled_job.id,
base_hash = %base_hash,
job_kind = ?kind,
"Creating new version for dependency job triggered by relative import"
);
let new_id = match kind {
JobKind::Dependencies => {
let deployment_message = pulled_job
.args
.clone()
.map(|hashmap| {
hashmap
.get("deployment_message")
.map(|map_value| {
serde_json::from_str::<String>(map_value.get()).ok()
})
.flatten()
})
.flatten();
// This way we tell downstream which script we should archive when the resolution is finished.
// (not used at the moment)
pulled_job.args.as_mut().map(|args| {
args.insert("base_hash".to_owned(), to_raw_value(&*base_hash))
});
let cloned_script = windmill_common::scripts::clone_script(
base_hash,
&pulled_job.workspace_id,
deployment_message,
&mut tx,
)
.await?;
if is_generated_from_raw_requirements(&Some(cloned_script.old_script.language), &cloned_script.old_script.lock.map(|v| v.to_string())) {
return Err(Error::BadRequest(format!(
"Script at path {} is generated from raw requirements, not overriding",
pulled_job.runnable_path()
)));
}
cloned_script.new_hash
}
JobKind::FlowDependencies => {
sqlx::query_scalar!(
"INSERT INTO flow_version
(workspace_id, path, value, schema, created_by)
SELECT workspace_id, path, value, schema, created_by
FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3
RETURNING id
",
pulled_job.runnable_path(),
pulled_job.workspace_id,
*base_hash,
)
.fetch_one(&mut *tx)
.await?
}
JobKind::AppDependencies => {
sqlx::query_scalar!(
"INSERT INTO app_version
(app_id, value, created_by, raw_app)
SELECT app_id, value, created_by, raw_app
FROM app_version WHERE id = $1
RETURNING id",
*base_hash
)
.fetch_one(&mut *tx)
.await?
}
_ => {
return Err(Error::InternalErr(format!(
"Matched unexpected JobKind ({:?}). This is a bug!",
kind
)))
}
};
pulled_job.runnable_id.replace(new_id.into());
if *windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING.read().await {
// === RETRIEVE ACCUMULATED DEBOUNCE DATA ===
//
// For flows and apps, retrieve all nodes/components that were accumulated
// during the debounce window. This data comes from requests that were merged
// into this job instead of creating their own jobs.
//
// Scripts don't need this because they don't have nodes/components to relock.
if let Some(to_relock_field) = match &pulled_job.kind {
JobKind::FlowDependencies => Some("nodes_to_relock"),
JobKind::AppDependencies => Some("components_to_relock"),
_ => None, // Scripts don't use accumulated stale data
} {
tracing::debug!(
job_id = %pulled_job.id,
job_kind = ?pulled_job.kind,
field = %to_relock_field,
"Retrieving accumulated stale data from debounced requests"
);
if let Some(stale_data) = sqlx::query_scalar!(
"DELETE FROM debounce_stale_data WHERE job_id = $1 RETURNING to_relock",
&pulled_job.id
)
.fetch_optional(&mut *tx)
.await
.map_err(|e| {
tracing::error!(
error = %e,
job_id = %pulled_job.id,
"Failed to retrieve debounce_stale_data"
);
e
})?
.flatten()
{
tracing::debug!(
job_id = %pulled_job.id,
node_count = stale_data.len(),
nodes = ?stale_data,
"Retrieved accumulated nodes/components from {} debounced requests",
stale_data.len()
);
// Replace the job's relock list with the accumulated data
// This ensures all nodes from all debounced requests are processed
if let Some(args) = pulled_job.args.as_mut() {
args.insert(to_relock_field.to_owned(), to_raw_value(&stale_data));
tracing::debug!(
field = %to_relock_field,
"Updated job args with accumulated debounce data"
);
}
} else {
tracing::trace!(
job_id = %pulled_job.id,
"No accumulated stale data found (no debounced requests or already cleaned up)"
);
}
}
} else {
tracing::warn!("Debouncing is not supported on this version of Windmill. Minimum version required for debouncing support.");
}
// This will unblock pusher.
tx.commit().await?;
Ok(())
}).await;
}
Ok(())
}
}
/// Pull the job from queue
@@ -2512,6 +2787,7 @@ pub async fn pull(
job: None,
suspended: false,
missing_concurrency_key: false,
error_while_preprocessing: None,
});
}
@@ -2567,20 +2843,21 @@ pub async fn pull(
// Concurrency limit is available for either enterprise job or dependency job
&& (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)) =>
{
let job = crate::jobs_ee::apply_concurrency_limit(
crate::jobs_ee::apply_concurrency_limit(
db,
pull_loop_count,
suspended,
job,
)
.await?;
job.unwrap_or(PulledJobResult {
.await?
.unwrap_or(PulledJobResult {
job: None,
suspended,
missing_concurrency_key: false,
error_while_preprocessing: None,
})
}
_ => PulledJobResult { job, suspended, missing_concurrency_key: false },
_ => PulledJobResult { job, suspended, missing_concurrency_key: false, error_while_preprocessing: None },
};
Ok::<_, Error>(pulled_job_result)
@@ -2598,7 +2875,7 @@ pub async fn pull(
)
.await?;
let Some(job) = job else {
return Ok(PulledJobResult { job: None, suspended, missing_concurrency_key: false });
return Ok(PulledJobResult { job: None, suspended, missing_concurrency_key: false, error_while_preprocessing: None });
};
let has_concurent_limit = job.concurrent_limit.is_some();
@@ -2626,6 +2903,7 @@ pub async fn pull(
job: Some(pulled_job),
suspended,
missing_concurrency_key: false,
error_while_preprocessing: None,
});
}
@@ -2633,11 +2911,11 @@ pub async fn pull(
if cfg!(feature = "enterprise")
|| (pulled_job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)
{
if let Some(pulled_job) =
if let Some(pulled_job_res) =
crate::jobs_ee::apply_concurrency_limit(db, pull_loop_count, suspended, pulled_job)
.await?
{
return Ok(pulled_job);
return Ok(pulled_job_res);
}
}
}
@@ -5547,245 +5825,3 @@ pub async fn get_same_worker_job(
))
})
}
pub async fn preprocess_dependency_job(job: &mut PulledJob, db: &DB) -> error::Result<()> {
let kind = job.kind;
// Handle dependency job debouncing cleanup when a job is pulled for execution
if kind.is_dependency()
&& job
.args
.as_ref()
.map(|x| x.get("triggered_by_relative_import").is_some())
.unwrap_or_default()
&& !*WMDEBUG_NO_DJOB_DEBOUNCING
{
return Box::pin(async move {
// Only used for testing in tests/relative_imports.rs
// Give us some space to work with.
#[cfg(debug_assertions)]
if let Some(duration) = job
.args
.as_ref()
.map(|x| {
x.get("dbg_sleep_between_pull_and_debounce_key_removal")
.map(|v| serde_json::from_str::<u32>(v.get()).ok())
.flatten()
})
.flatten()
{
tracing::debug!("going to sleep",);
sleep(std::time::Duration::from_secs(duration as u64)).await;
}
tracing::debug!(
"Processing debounce cleanup for dependency job {} at path {:?}",
&job.id,
&job.runnable_path
);
let key = format!("{}:{}:dependency", &job.workspace_id, job.runnable_path());
let mut tx = db.begin().await?;
// === DEBOUNCE CLEANUP ===
//
// Clean up the debounce_key entry for this job (if it exists).
//
// IMPORTANT: We delete by key (not job_id) to avoid race conditions:
// If pusher has locked this row then this call will be blocked until all txs are commited.
//
// The idea is that the worker_lockfiles::trigger_dependents_to_recompute_locks will fetch the latest version of the obj.
// This object needs to be created before the djob is executed and it happens right here.
//
// This way the next pusher can fetch the latest version of object and base their djob payload on newest version.
// The concurrency limit on djobs will make sure that by the time next djob is started executing the base version it is referencing
// has already calculated all locks. This way even next djob will always use the fully finalized version of object.
//
//
//
// Note: We don't use a transaction here for performance (it's called during job pull).
// This means there's a tiny window where the job is running but key isn't deleted yet,
// which is acceptable because new requests will just accumulate data to this job.
tracing::debug!(
job_id = %job.id,
"Cleaning up debounce_key entry for completed/pulled job"
);
// This will either:
// 1. Block until pusher pushed. Which gives us:
// - If there was any stale data in pusher, then we will read it here (couple of lines below)
// 2. Block pusher until we are done here. This gives us:
// - We will clone objects and retrieve the latest version. So when we are done the pusher can read latest version.
sqlx::query!("DELETE FROM debounce_key WHERE key = $1", &key)
.execute(&mut *tx)
.await
.map_err(|e| {
tracing::error!(
error = %e,
job_id = %job.id,
"Failed to delete debounce_key"
);
e
})?;
let Some(base_hash) = job.runnable_id else {
return Err(Error::InternalErr(
"Missing runnable_id for dependency job triggered by relative import"
.to_string(),
));
};
tracing::debug!(
job_id = %job.id,
base_hash = %base_hash,
job_kind = ?kind,
"Creating new version for dependency job triggered by relative import"
);
let new_id = match kind {
JobKind::Dependencies => {
let deployment_message = job
.args
.clone()
.map(|hashmap| {
hashmap
.get("deployment_message")
.map(|map_value| {
serde_json::from_str::<String>(map_value.get()).ok()
})
.flatten()
})
.flatten();
// This way we tell downstream which script we should archive when the resolution is finished.
// (not used at the moment)
job.args.as_mut().map(|args| {
args.insert("base_hash".to_owned(), to_raw_value(&*base_hash))
});
let cloned_script = windmill_common::scripts::clone_script(
base_hash,
&job.workspace_id,
deployment_message,
&mut tx,
)
.await?;
if is_generated_from_raw_requirements(&Some(cloned_script.old_script.language), &cloned_script.old_script.lock.map(|v| v.to_string())) {
return Err(Error::BadRequest(format!(
"Script at path {} is generated from raw requirements, not overriding",
job.runnable_path()
)));
}
cloned_script.new_hash
}
JobKind::FlowDependencies => {
sqlx::query_scalar!(
"INSERT INTO flow_version
(workspace_id, path, value, schema, created_by)
SELECT workspace_id, path, value, schema, created_by
FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3
RETURNING id
",
job.runnable_path(),
job.workspace_id,
*base_hash,
)
.fetch_one(&mut *tx)
.await?
}
JobKind::AppDependencies => {
sqlx::query_scalar!(
"INSERT INTO app_version
(app_id, value, created_by, raw_app)
SELECT app_id, value, created_by, raw_app
FROM app_version WHERE id = $1
RETURNING id",
*base_hash
)
.fetch_one(&mut *tx)
.await?
}
_ => {
return Err(Error::InternalErr(format!(
"Matched unexpected JobKind ({:?}). This is a bug!",
kind
)))
}
};
job.runnable_id.replace(new_id.into());
if !*windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING.read().await {
tx.commit().await?;
tracing::warn!("Debouncing is not supported on this version of Windmill. Minimum version required for debouncing support.");
return Ok(());
}
// === RETRIEVE ACCUMULATED DEBOUNCE DATA ===
//
// For flows and apps, retrieve all nodes/components that were accumulated
// during the debounce window. This data comes from requests that were merged
// into this job instead of creating their own jobs.
//
// Scripts don't need this because they don't have nodes/components to relock.
if let Some(to_relock_field) = match &job.kind {
JobKind::FlowDependencies => Some("nodes_to_relock"),
JobKind::AppDependencies => Some("components_to_relock"),
_ => None, // Scripts don't use accumulated stale data
} {
tracing::debug!(
job_id = %job.id,
job_kind = ?job.kind,
field = %to_relock_field,
"Retrieving accumulated stale data from debounced requests"
);
if let Some(stale_data) = sqlx::query_scalar!(
"DELETE FROM debounce_stale_data WHERE job_id = $1 RETURNING to_relock",
&job.id
)
.fetch_optional(&mut *tx)
.await
.map_err(|e| {
tracing::error!(
error = %e,
job_id = %job.id,
"Failed to retrieve debounce_stale_data"
);
e
})?
.flatten()
{
tracing::debug!(
job_id = %job.id,
node_count = stale_data.len(),
nodes = ?stale_data,
"Retrieved accumulated nodes/components from {} debounced requests",
stale_data.len()
);
// Replace the job's relock list with the accumulated data
// This ensures all nodes from all debounced requests are processed
if let Some(args) = job.args.as_mut() {
args.insert(to_relock_field.to_owned(), to_raw_value(&stale_data));
tracing::debug!(
field = %to_relock_field,
"Updated job args with accumulated debounce data"
);
}
} else {
tracing::trace!(
job_id = %job.id,
"No accumulated stale data found (no debounced requests or already cleaned up)"
);
}
}
// This will unblock pusher.
tx.commit().await?;
Ok(())
}).await;
}
Ok(())
}
backend/windmill-worker/src/ai/tools.rs
+2 -2
View File
@@ -6,7 +6,7 @@ use crate::ai::utils::{
update_flow_status_module_with_actions, update_flow_status_module_with_actions_success,
FlowContext,
};
use crate::common::{error_to_value, OccupancyMetrics};
use crate::common::OccupancyMetrics;
use crate::result_processor::handle_non_flow_job_error;
use crate::worker_flow::{
evaluate_input_transform, raw_script_to_payload, script_to_payload, JobPayloadWithTag,
@@ -591,7 +591,7 @@ async fn handle_tool_execution_error(
final_events_str: &mut String,
) -> Result<(), Error> {
let err_string = format!("{}: {}", err.name(), err.to_string());
let err_json = error_to_value(&err);
let err_json = windmill_common::worker::error_to_value(&err);
let _ = handle_non_flow_job_error(
ctx.db,
tool_job,
backend/windmill-worker/src/common.rs
-7
View File
@@ -539,13 +539,6 @@ pub async fn update_worker_ping_for_failed_init_script(
}
}
pub fn error_to_value(err: &Error) -> serde_json::Value {
match err {
Error::JsonErr(err) => err.clone(),
_ => json!({"message": err.to_string(), "name": err.name()}),
}
}
#[derive(Clone)]
pub struct OccupancyMetrics {
pub running_job_started_at: Option<Instant>,
backend/windmill-worker/src/result_processor.rs
+4 -3
View File
@@ -22,7 +22,7 @@ use windmill_common::{
flow_status::FlowJobDuration,
jobs::JobKind,
utils::WarnAfterExt,
worker::{to_raw_value, Connection, WORKER_GROUP},
worker::{error_to_value, to_raw_value, Connection, WORKER_GROUP},
worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap},
KillpillSender, DB,
};
@@ -31,7 +31,8 @@ use windmill_common::{
use windmill_common::bench::{BenchmarkInfo, BenchmarkIter};
use windmill_queue::{
CanceledBy, INIT_SCRIPT_TAG, JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, WrappedError, append_logs, get_mini_completed_job
append_logs, get_mini_completed_job, CanceledBy, JobCompleted, MiniCompletedJob, MiniPulledJob,
ValidableJson, WrappedError, INIT_SCRIPT_TAG,
};
use serde_json::{json, value::RawValue, Value};
@@ -42,7 +43,7 @@ use windmill_queue::{add_completed_job, add_completed_job_error};
use crate::{
bash_executor::ANSI_ESCAPE_RE,
common::{error_to_value, read_result, save_in_cache},
common::{read_result, save_in_cache},
otel_oss::add_root_flow_job_to_otlp,
worker_flow::update_flow_status_after_job_completion,
JobCompletedReceiver, JobCompletedSender, SameWorkerSender, SendResult, SendResultPayload,
backend/windmill-worker/src/worker.rs
+45 -28
View File
@@ -18,6 +18,7 @@ use windmill_common::scripts::hash_to_codebase_id;
use windmill_common::scripts::is_special_codebase_hash;
use windmill_common::utils::report_critical_error;
use windmill_common::utils::retrieve_common_worker_prefix;
use windmill_common::worker::error_to_value;
use windmill_common::{
agent_workers::DECODED_AGENT_TOKEN,
apps::AppScriptId,
@@ -58,7 +59,6 @@ use std::{
time::Duration,
};
use windmill_parser::MainArgSignature;
use windmill_queue::preprocess_dependency_job;
use windmill_queue::MiniCompletedJob;
use windmill_queue::PulledJobResultToJobErr;
@@ -112,7 +112,7 @@ use crate::{
bash_executor::handle_bash_job,
bun_executor::handle_bun_job,
common::{
build_args_map, cached_result_path, error_to_value, get_cached_resource_value_if_valid,
build_args_map, cached_result_path, get_cached_resource_value_if_valid,
get_reserved_variables, update_worker_ping_for_failed_init_script, OccupancyMetrics,
},
csharp_executor::handle_csharp_job,
@@ -841,10 +841,22 @@ pub fn start_interactive_worker_shell(
match job {
Ok(j) => match j.to_pulled_job() {
Ok(j) => Ok(j.map(NextJob::Sql)),
Err(PulledJobResultToJobErr::MissingConcurrencyKey(jc)) => {
if let Err(err) = job_completed_tx.send_job(jc, true).await {
tracing::error!("An error occurred while sending job completed (missing concurrency key): {:#?}", err)
Ok(j) => Ok(j.clone().map(NextJob::Sql)),
ref e @ (Err(PulledJobResultToJobErr::MissingConcurrencyKey(
ref jc,
))
| Err(PulledJobResultToJobErr::ErrorWhilePreprocessing(
ref jc,
))) => {
if let Err(err) =
job_completed_tx.send_job(jc.clone(), true).await
{
let e_fmt = match e {
Ok(_) => "unknown error".to_owned(),
Err(e) => e.to_string(),
};
tracing::error!("An error occurred while sending job completed ({e_fmt}): {:#?}", err)
}
Ok(None)
}
@@ -877,7 +889,7 @@ pub fn start_interactive_worker_shell(
token,
precomputed_agent_info: precomputed_bundle,
} = extract_job_and_perms(job, &conn).await;
let authed_client = AuthedClient::new(
base_internal_url.to_owned(),
job.workspace_id.clone(),
@@ -886,7 +898,7 @@ pub fn start_interactive_worker_shell(
);
let arc_job = Arc::new(job);
let _ = handle_queued_job(
arc_job.clone(),
raw_code,
@@ -1605,30 +1617,23 @@ pub async fn run_worker(
}
};
// Essential debouncing job preprocessing.
if let Ok(windmill_queue::PulledJobResult {
job: Some(ref mut pulled_job),
..
}) = &mut job
{
match timeout(
// Preprocess pulled job result
if let Ok(ref mut pulled_job_res) = job {
if let Err(e) = timeout(
// Will fail if longer than 10 seconds
core::time::Duration::from_secs(10),
preprocess_dependency_job(pulled_job, &db),
pulled_job_res.preprocess(db),
)
.warn_after_seconds(2)
.await
// Flatten result
.map_err(error::Error::from)
.and_then(|r| r)
{
Ok(Err(e)) => {
tracing::error!(worker = %worker_name, hostname = %hostname, "critical: debouncing job preprocessor failed: {e:?}");
job = Err(e.into());
}
Err(e) => {
tracing::error!(worker = %worker_name, hostname = %hostname, "critical: debouncing job preprocessor has timed out: {e:?}");
job = Err(e.into());
}
_ => {}
pulled_job_res.error_while_preprocessing = Some(e.to_string());
}
}
add_time!(bench, "job pulled from DB");
let duration_pull_s = pull_time.elapsed().as_secs_f64();
let err_pull = job.is_ok();
@@ -1688,9 +1693,21 @@ pub async fn run_worker(
match job {
Ok(pulled_job_result) => match pulled_job_result.to_pulled_job() {
Ok(j) => Ok(j.map(NextJob::Sql)),
Err(PulledJobResultToJobErr::MissingConcurrencyKey(jc)) => {
if let Err(err) = job_completed_tx.send_job(jc, true).await {
tracing::error!("An error occurred while sending job completed (missing concurrency key): {:#?}", err)
ref e @ (Err(PulledJobResultToJobErr::MissingConcurrencyKey(
ref jc,
))
| Err(PulledJobResultToJobErr::ErrorWhilePreprocessing(
ref jc,
))) => {
if let Err(err) =
job_completed_tx.send_job(jc.clone(), true).await
{
let e_fmt = match e {
Ok(_) => "unknown error".to_owned(),
Err(e) => e.to_string(),
};
tracing::error!("An error occurred while sending job completed ({e_fmt}): {:#?}", err)
}
Ok(None)
}