mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-07 08:02:40 +00:00
add failed_jobs to retrystatus
This commit is contained in:
Ruben Fiszel
+86
-79
@@ -44,6 +44,7 @@ pub struct FlowStatus {
|
||||
pub struct RetryStatus {
|
||||
pub fail_count: u16,
|
||||
pub previous_result: Option<Value>,
|
||||
pub failed_jobs: Vec<Uuid>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
@@ -105,7 +106,7 @@ impl FlowStatus {
|
||||
step: 0,
|
||||
modules: vec![FlowStatusModule::WaitingForPriorSteps; f.modules.len()],
|
||||
failure_module: FlowStatusModule::WaitingForPriorSteps,
|
||||
retry: RetryStatus { fail_count: 0, previous_result: None },
|
||||
retry: RetryStatus { fail_count: 0, previous_result: None, failed_jobs: vec![] },
|
||||
}
|
||||
}
|
||||
|
||||
@@ -718,59 +719,18 @@ async fn push_next_flow_job(
|
||||
}
|
||||
}
|
||||
|
||||
if matches!(&status_module, FlowStatusModule::Failure { .. }) {
|
||||
let retry = &module.retry.clone().unwrap_or_default();
|
||||
if let Some((fail_count, retry_in)) = next_retry(retry, &status.retry) {
|
||||
tracing::debug!(
|
||||
retry_in_seconds = retry_in.as_secs(),
|
||||
fail_count = fail_count,
|
||||
"retrying"
|
||||
);
|
||||
|
||||
scheduled_for_o = Some(from_now(retry_in));
|
||||
|
||||
sqlx::query(
|
||||
"
|
||||
UPDATE queue
|
||||
SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1)
|
||||
WHERE id = $2
|
||||
",
|
||||
)
|
||||
.bind(json!(RetryStatus { fail_count, ..status.retry.clone() }))
|
||||
.bind(flow_job.id)
|
||||
.execute(db)
|
||||
.await
|
||||
.context("update flow retry")?;
|
||||
|
||||
/* it might be better to retry the job using the previous args instead of determining
|
||||
* them again from the last result, but that seemed to not play well with the forloop
|
||||
* logic and I couldn't figure out why. */
|
||||
if let Some(v) = &status.retry.previous_result {
|
||||
last_result = v.clone();
|
||||
}
|
||||
status_module = FlowStatusModule::WaitingForPriorSteps;
|
||||
|
||||
/* Start the failure module ... */
|
||||
} else {
|
||||
/* push_next_flow_job is called with the current step on FlowStatusModule::Failure.
|
||||
* This must update the step index to the end so that no subsequent steps are run after
|
||||
* the failure module.
|
||||
*
|
||||
* The failure module may also run again if it fails and the retry feature is used.
|
||||
* In that case, `i` will index past `flow.modules`. The above should handle that and
|
||||
* re-run the failure module. */
|
||||
i = flow.modules.len();
|
||||
module = flow
|
||||
.failure_module
|
||||
.as_ref()
|
||||
/* If this fails, it's a update_flow_status_after_job_completion shouldn't have called
|
||||
* handle_flow to get here. */
|
||||
.context("missing failure module")?;
|
||||
status_module = status.failure_module.clone();
|
||||
|
||||
/* (retry feature) save the previous_result the first time this step is run */
|
||||
match &status_module {
|
||||
FlowStatusModule::Failure { job, .. } => {
|
||||
let retry = &module.retry.clone().unwrap_or_default();
|
||||
if retry.has_attempts() {
|
||||
if let Some((fail_count, retry_in)) = next_retry(retry, &status.retry) {
|
||||
tracing::debug!(
|
||||
retry_in_seconds = retry_in.as_secs(),
|
||||
fail_count = fail_count,
|
||||
"retrying"
|
||||
);
|
||||
|
||||
scheduled_for_o = Some(from_now(retry_in));
|
||||
status.retry.failed_jobs.push(job.clone());
|
||||
sqlx::query(
|
||||
"
|
||||
UPDATE queue
|
||||
@@ -778,41 +738,88 @@ async fn push_next_flow_job(
|
||||
WHERE id = $2
|
||||
",
|
||||
)
|
||||
.bind(json!(RetryStatus {
|
||||
previous_result: Some(last_result.clone()),
|
||||
fail_count: 0,
|
||||
}))
|
||||
.bind(json!(RetryStatus { fail_count, ..status.retry.clone() }))
|
||||
.bind(flow_job.id)
|
||||
.execute(db)
|
||||
.await
|
||||
.context("update flow retry")?;
|
||||
};
|
||||
}
|
||||
|
||||
/* (retry feature) save the previous_result the first time this step is run */
|
||||
} else if matches!(&status_module, FlowStatusModule::WaitingForPriorSteps)
|
||||
&& module
|
||||
.retry
|
||||
.as_ref()
|
||||
.map(|x| x.has_attempts())
|
||||
.unwrap_or(false)
|
||||
&& status.retry.fail_count == 0
|
||||
{
|
||||
sqlx::query(
|
||||
"
|
||||
/* it might be better to retry the job using the previous args instead of determining
|
||||
* them again from the last result, but that seemed to not play well with the forloop
|
||||
* logic and I couldn't figure out why. */
|
||||
if let Some(v) = &status.retry.previous_result {
|
||||
last_result = v.clone();
|
||||
}
|
||||
status_module = FlowStatusModule::WaitingForPriorSteps;
|
||||
|
||||
/* Start the failure module ... */
|
||||
} else {
|
||||
/* push_next_flow_job is called with the current step on FlowStatusModule::Failure.
|
||||
* This must update the step index to the end so that no subsequent steps are run after
|
||||
* the failure module.
|
||||
*
|
||||
* The failure module may also run again if it fails and the retry feature is used.
|
||||
* In that case, `i` will index past `flow.modules`. The above should handle that and
|
||||
* re-run the failure module. */
|
||||
i = flow.modules.len();
|
||||
module = flow
|
||||
.failure_module
|
||||
.as_ref()
|
||||
/* If this fails, it's a update_flow_status_after_job_completion shouldn't have called
|
||||
* handle_flow to get here. */
|
||||
.context("missing failure module")?;
|
||||
status_module = status.failure_module.clone();
|
||||
|
||||
/* (retry feature) save the previous_result the first time this step is run */
|
||||
let retry = &module.retry.clone().unwrap_or_default();
|
||||
if retry.has_attempts() {
|
||||
sqlx::query(
|
||||
"
|
||||
UPDATE queue
|
||||
SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1)
|
||||
WHERE id = $2
|
||||
",
|
||||
)
|
||||
.bind(json!(RetryStatus {
|
||||
previous_result: Some(last_result.clone()),
|
||||
fail_count: 0,
|
||||
failed_jobs: vec![],
|
||||
}))
|
||||
.bind(flow_job.id)
|
||||
.execute(db)
|
||||
.await
|
||||
.context("update flow retry")?;
|
||||
};
|
||||
}
|
||||
|
||||
/* (retry feature) save the previous_result the first time this step is run */
|
||||
}
|
||||
FlowStatusModule::WaitingForPriorSteps
|
||||
if module
|
||||
.retry
|
||||
.as_ref()
|
||||
.map(|x| x.has_attempts())
|
||||
.unwrap_or(false)
|
||||
&& status.retry.fail_count == 0 =>
|
||||
{
|
||||
sqlx::query(
|
||||
"
|
||||
UPDATE queue
|
||||
SET flow_status = JSONB_SET(flow_status, ARRAY['retry'], $1)
|
||||
WHERE id = $2
|
||||
",
|
||||
)
|
||||
.bind(json!(RetryStatus {
|
||||
previous_result: Some(last_result.clone()),
|
||||
fail_count: 0,
|
||||
}))
|
||||
.bind(flow_job.id)
|
||||
.execute(db)
|
||||
.await
|
||||
.context("update flow retry")?;
|
||||
)
|
||||
.bind(json!(RetryStatus {
|
||||
previous_result: Some(last_result.clone()),
|
||||
fail_count: 0,
|
||||
failed_jobs: vec![],
|
||||
}))
|
||||
.bind(flow_job.id)
|
||||
.execute(db)
|
||||
.await
|
||||
.context("update flow retry")?;
|
||||
}
|
||||
_ => (),
|
||||
}
|
||||
|
||||
/* Don't evaluate `module.input_transforms` after iteration has begun. Instead, args are
|
||||
|
||||
@@ -253,6 +253,11 @@ components:
|
||||
properties:
|
||||
fail_count:
|
||||
type: integer
|
||||
failed_jobs:
|
||||
type: array
|
||||
items:
|
||||
type: string
|
||||
format: uuid
|
||||
required:
|
||||
- step
|
||||
- modules
|
||||
|
||||
Reference in New Issue
Block a user