diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index d0055bbe3a..dabedefc00 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -143,6 +143,7 @@ pub struct JobCompleted { pub canceled_by: Option, pub duration: Option, pub has_stream: Option, + pub from_cache: Option, } 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( flow_is_done: bool, duration: Option, 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( flow_is_done, duration, has_stream, + from_cache, ) }) .retry( @@ -881,6 +885,7 @@ async fn commit_completed_job( flow_is_done: bool, duration: Option, has_stream: bool, + from_cache: bool, ) -> windmill_common::error::Result<(Option, i64, bool)> { // let start = std::time::Instant::now(); @@ -1061,9 +1066,11 @@ async fn commit_completed_job( } // 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), diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index df57c5e67c..4ae3621b73 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -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 diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index a367c818a3..6ce2717d38 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -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(); diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 3b0645ef06..9f4fd7907d 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -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, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index aed1f82d93..50b1f8a967 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -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, ) diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 2b8081651d..6889f487cc 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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