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

fix: fix scheduling of flows with cached results based on inputs (#6774)

Browse Source
This commit is contained in:
Ruben Fiszel
2025-10-08 09:14:13 +00:00
committed by GitHub
parent 27d7809ef9
commit 5ad61829e4
6 changed files with 28 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-queue/src/jobs.rs
+9 -1
View File
@@ -143,6 +143,7 @@ pub struct JobCompleted {
pub canceled_by: Option<CanceledBy>,
pub duration: Option<i64>,
pub has_stream: Option<bool>,
pub from_cache: Option<bool>,
}
pub async fn cancel_single_job<'c>(
@@ -736,6 +737,7 @@ pub async fn add_completed_job_error(
flow_is_done,
duration,
false,
false,
)
.await?;
Ok(result)
@@ -758,6 +760,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
flow_is_done: bool,
duration: Option<i64>,
has_stream: bool,
from_cache: bool,
) -> Result<(Uuid, i64), Error> {
// tracing::error!("Start");
// let start = tokio::time::Instant::now();
@@ -783,6 +786,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
flow_is_done,
duration,
has_stream,
from_cache,
)
})
.retry(
@@ -881,6 +885,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
flow_is_done: bool,
duration: Option<i64>,
has_stream: bool,
from_cache: bool,
) -> windmill_common::error::Result<(Option<Uuid>, i64, bool)> {
// let start = std::time::Instant::now();
@@ -1061,9 +1066,11 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
}
// for scripts, always try to schedule next tick
// for flows, only try to schedule next tick here if flow failed and because first handle_flow failed (step = 0, modules[0] = {type: 'Failure', 'job': uuid::nil()}) or job was cancelled before first handle_flow was called (step = 0, modules = [] OR modules[0].type == 'WaitingForPriorSteps')
// for flows, only try to schedule next tick here if flow failed and because first handle_flow failed (step = 0, modules[0] = {type: 'Failure', 'job': uuid::nil()})
// or job was cancelled before first handle_flow was called (step = 0, modules = [] OR modules[0].type == 'WaitingForPriorSteps')
// otherwise flow rescheduling is done inside handle_flow
let schedule_next_tick = !queued_job.is_flow()
|| from_cache
|| !success
&& sqlx::query_scalar!(
"SELECT
@@ -2218,6 +2225,7 @@ impl PulledJobResult {
canceled_by: None,
duration: None,
has_stream: Some(false),
from_cache: None,
}),
),
PulledJobResult { job, .. } => Ok(job),
backend/windmill-queue/src/schedule.rs
+1 -1
View File
@@ -114,7 +114,7 @@ pub async fn push_scheduled_job<'c>(
.unwrap_or(false);
if already_exists {
tracing::info!(
tracing::warn!(
"Job for schedule {} at {} already exists",
&schedule.path,
next
backend/windmill-worker/src/dedicated_worker.rs
+3 -3
View File
@@ -186,14 +186,14 @@ pub async fn handle_dedicated_process(
let result = Arc::new(result);
append_logs(&job.id, &job.workspace_id, logs.clone(), &db.into()).await;
if line.starts_with("wm_res[success]:") {
job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: Some(false) }, true).await.unwrap()
job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: Some(false), from_cache: None }, true).await.unwrap()
} else {
job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: Some(false) }, true).await.unwrap()
job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: Some(false), from_cache: None }, true).await.unwrap()
}
},
Err(e) => {
tracing::error!("Could not deserialize job result `{line}`: {e:?}");
job_completed_tx.send_job(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: Some(false) }, true).await.unwrap();
job_completed_tx.send_job(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: Some(false), from_cache: None }, true).await.unwrap();
},
};
logs = init_log.clone();
backend/windmill-worker/src/result_processor.rs
+10 -4
View File
@@ -430,6 +430,7 @@ pub async fn process_result(
token: token.to_string(),
duration,
has_stream: Some(has_stream),
from_cache: None,
},
)
.with_context(windmill_common::otel_oss::otel_ctx())
@@ -493,6 +494,7 @@ pub async fn process_result(
token: token.to_string(),
duration,
has_stream: Some(has_stream),
from_cache: None,
},
)
.with_context(windmill_common::otel_oss::otel_ctx())
@@ -568,6 +570,7 @@ pub async fn process_completed_job(
result_columns,
preprocessed_args,
has_stream,
from_cache,
..
}: JobCompleted,
client: &AuthedClient,
@@ -633,6 +636,7 @@ pub async fn process_completed_job(
false,
duration,
has_stream.unwrap_or(false),
from_cache.unwrap_or(false),
)
.await?;
drop(job);
@@ -691,10 +695,12 @@ pub async fn process_completed_job(
&job.workspace_id,
false,
Arc::new(serde_json::value::to_raw_value(&result).unwrap()),
duration.and_then(|d| job.started_at.map(|started_at| FlowJobDuration {
started_at: started_at,
duration_ms: d,
})),
duration.and_then(|d| {
job.started_at.map(|started_at| FlowJobDuration {
started_at: started_at,
duration_ms: d,
})
}),
false,
&same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(),
&worker_dir,
backend/windmill-worker/src/worker.rs
+3
View File
@@ -776,6 +776,7 @@ pub async fn handle_all_job_kind_error(
token: authed_client.token.clone(),
duration: None,
has_stream: Some(false),
from_cache: None,
},
false,
)
@@ -1721,6 +1722,7 @@ pub async fn run_worker(
canceled_by: None,
duration: None,
has_stream: Some(false),
from_cache: None,
},
true,
)
@@ -2448,6 +2450,7 @@ pub async fn handle_queued_job(
token: client.token.clone(),
duration: None,
has_stream: Some(false),
from_cache: Some(true),
},
true,
)
backend/windmill-worker/src/worker_flow.rs
+2
View File
@@ -1474,6 +1474,7 @@ pub async fn update_flow_status_after_job_completion_internal(
true,
None,
false,
false,
)
.await?;
duration
@@ -1494,6 +1495,7 @@ pub async fn update_flow_status_after_job_completion_internal(
true,
None,
false,
false,
)
.await?;
duration