kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

flow step retry feature (#493)

Browse Source 
* flow step retry feature

* comparison constant on right side for clarity

* raise high retry values when starting a flow

also renamed duration to interval to be more specific about the retry
interval/period between tries or attempts

* add flow retry to openflow openapi

Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com>
This commit is contained in:
sqwishy
2022-09-03 11:38:31 -07:00
committed by GitHub
parent e8aec1c93a
commit d69d002b82
8 changed files with 763 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
backend/src/error.rs
View File

@@ -49,6 +49,13 @@ pub enum Error {
Anyhow(#[from] anyhow::Error),
}
impl Error {
/// https://docs.rs/anyhow/1/anyhow/struct.Error.html#display-representations
pub fn alt(&self) -> String {
format!("{:#}", self)
}
}
pub fn to_anyhow<T: 'static + std::error::Error + Send + Sync>(e: T) -> anyhow::Error {
From::from(e)
}

252
backend/src/flows.rs
View File

@@ -7,6 +7,7 @@
*/
use std::collections::HashMap;
use std::time::Duration;
use reqwest::Client;
use sql_builder::prelude::*;
@@ -25,6 +26,7 @@ use crate::{
db::{UserDB, DB},
error::{self, to_anyhow, Error, JsonResult, Result},
jobs::RawCode,
more_serde::{default_true, is_default},
scripts::Schema,
users::Authed,
utils::{http_get_from_hub, list_elems_from_hub, Pagination, StripPath},
@@ -69,17 +71,86 @@ pub struct NewFlow {
pub schema: Option<Schema>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
#[derive(Deserialize, Serialize, Debug, Clone, Default)]
pub struct FlowValue {
pub modules: Vec<FlowModule>,
#[serde(default)]
#[serde(skip_serializing_if = "is_default")]
pub retry: Retry,
#[serde(default)]
pub failure_module: Option<FlowModule>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct StopAfterIf {
pub expr: String,
pub skip_if_stopped: bool,
}
#[derive(Deserialize, Serialize, Debug, Clone, Default, PartialEq)]
#[serde(default)]
pub struct Retry {
constant: ConstantDelay,
exponential: ExponentialDelay,
}
impl Retry {
/// Takes the number of previous retries and returns the interval until the next retry if any.
///
/// May return [`Duration::ZERO`] to retry immediately.
pub fn interval(&self, previous_attempts: u16) -> Option<Duration> {
let Self { constant, exponential } = self;
if previous_attempts < constant.attempts {
Some(Duration::from_secs(constant.seconds as u64))
} else if previous_attempts - constant.attempts < exponential.attempts {
let exp = previous_attempts.saturating_add(1) as u32;
let secs = exponential.multiplier * exponential.seconds.saturating_pow(exp);
Some(Duration::from_secs(secs as u64))
} else {
None
}
}
pub fn has_attempts(&self) -> bool {
self.constant.attempts != 0 || self.exponential.attempts != 0
}
pub fn max_attempts(&self) -> u16 {
self.constant
.attempts
.saturating_add(self.exponential.attempts)
}
pub fn max_interval(&self) -> Option<Duration> {
self.max_attempts()
.checked_sub(1)
.and_then(|p| self.interval(p))
}
}
#[derive(Deserialize, Serialize, Debug, Clone, Default, PartialEq)]
#[serde(default)]
pub struct ConstantDelay {
pub attempts: u16,
pub seconds: u16,
}
/// multiplier * seconds ^ failures
#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
#[serde(default)]
pub struct ExponentialDelay {
pub attempts: u16,
pub multiplier: u16,
pub seconds: u16,
}
impl Default for ExponentialDelay {
fn default() -> Self {
Self { attempts: 0, multiplier: 1, seconds: 0 }
}
}
#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct FlowModule {
#[serde(default)]
@@ -121,10 +192,6 @@ pub enum FlowModuleValue {
RawScript(RawCode),
}
fn default_true() -> bool {
true
}
#[derive(Deserialize)]
pub struct ListFlowQuery {
pub path_start: Option<String>,
@@ -413,17 +480,18 @@ mod tests {
// Note this useful idiom: importing names from outer (for mod tests) scope.
use super::*;
const SECOND: Duration = Duration::from_secs(1);
#[test]
fn test_serialize() -> anyhow::Result<()> {
let mut hm = HashMap::new();
hm.insert(
"test".to_owned(),
InputTransform::Static { value: serde_json::json!("test2") },
);
fn flowmodule_serde() {
let fv = FlowValue {
modules: vec![
FlowModule {
input_transforms: hm,
input_transforms: [(
"test".to_string(),
InputTransform::Static { value: serde_json::json!("test2") },
)]
.into(),
value: FlowModuleValue::Script { path: "test".to_string() },
stop_after_if: None,
summary: None,
@@ -468,9 +536,83 @@ mod tests {
}),
summary: None,
}),
retry: Default::default(),
};
println!("{}", serde_json::json!(fv).to_string());
Ok(())
let expect = serde_json::json!({
"modules": [
{
"input_transforms": {
"test": {
"type": "static",
"value": "test2"
}
},
"value": {
"type": "script",
"path": "test"
},
"stop_after_if": null,
"summary": null
},
{
"input_transforms": {},
"value": {
"type": "rawscript",
"content": "test",
"path": null,
"language": "deno"
},
"stop_after_if": {
"expr": "foo = 'bar'",
"skip_if_stopped": false
},
"summary": null
},
{
"input_transforms": {
"iterand": {
"type": "static",
"value": [
1,
2,
3
]
}
},
"value": {
"type": "forloopflow",
"iterator": {
"type": "static",
"value": [
1,
2,
3
]
},
"skip_failures": true,
"modules": []
},
"stop_after_if": {
"expr": "previous.isEmpty()",
"skip_if_stopped": false,
},
"summary": null
}
],
"failure_module": {
"input_transforms": {},
"value": {
"type": "flow",
"path": "test"
},
"stop_after_if": {
"expr": "previous.isEmpty()",
"skip_if_stopped": false
},
"summary": null
}
});
assert_eq!(dbg!(serde_json::json!(fv)), dbg!(expect));
}
#[test]
@@ -497,4 +639,86 @@ mod tests {
InputTransform::Javascript { expr: "flow_input.iter.value".to_string() }
);
}
#[test]
fn retry_serde() {
assert_eq!(Retry::default(), serde_json::from_str(r#"{}"#).unwrap());
assert_eq!(
Retry::default(),
serde_json::from_str(
r#"
{
"constant": {
"seconds": 0
},
"exponential": {
"multiplier": 1,
"seconds": 0
}
}
"#
)
.unwrap()
);
assert_eq!(
Retry {
constant: Default::default(),
exponential: ExponentialDelay { attempts: 0, multiplier: 1, seconds: 123 }
},
serde_json::from_str(
r#"
{
"constant": {},
"exponential": { "seconds": 123 }
}
"#
)
.unwrap()
);
}
#[test]
fn retry_exponential() {
let retry = Retry {
constant: ConstantDelay::default(),
exponential: ExponentialDelay { attempts: 3, multiplier: 4, seconds: 3 },
};
assert_eq!(
vec![
Some(12 * SECOND),
Some(36 * SECOND),
Some(108 * SECOND),
None
],
(0..4)
.map(|previous_attempts| retry.interval(previous_attempts))
.collect::<Vec<_>>()
);
assert_eq!(Some(108 * SECOND), retry.max_interval());
}
#[test]
fn retry_both() {
let retry = Retry {
constant: ConstantDelay { attempts: 2, seconds: 4 },
exponential: ExponentialDelay { attempts: 2, multiplier: 1, seconds: 3 },
};
assert_eq!(
vec![
Some(4 * SECOND),
Some(4 * SECOND),
Some(27 * SECOND),
Some(81 * SECOND),
None,
],
(0..5)
.map(|previous_attempts| retry.interval(previous_attempts))
.collect::<Vec<_>>()
);
assert_eq!(Some(81 * SECOND), retry.max_interval());
}
}

14
backend/src/jobs.rs
View File

@@ -24,7 +24,7 @@ use crate::{
users::{owner_to_token_owner, Authed},
utils::{require_admin, Pagination, StripPath, now_from_db},
worker,
worker_flow::init_flow_status,
worker_flow::{init_flow_status, FlowStatus},
};
use axum::{
extract::{Extension, Path, Query},
@@ -99,6 +99,18 @@ impl QueuedJob {
.map(String::as_str)
.unwrap_or("NO_FLOW_PATH")
}
pub fn parse_raw_flow(&self) -> Option<FlowValue> {
self.raw_flow
.as_ref()
.and_then(|v| serde_json::from_value::<FlowValue>(v.clone()).ok())
}
pub fn parse_flow_status(&self) -> Option<FlowStatus> {
self.flow_status
.as_ref()
.and_then(|v| serde_json::from_value::<FlowStatus>(v.clone()).ok())
}
}
#[derive(Debug, sqlx::FromRow, Serialize)]

1
backend/src/lib.rs
View File

@@ -34,6 +34,7 @@ mod granular_acls;
mod groups;
mod jobs;
mod js_eval;
mod more_serde;
mod oauth2;
mod parser;
mod parser_py;

9
backend/src/more_serde.rs Normal file
View File

@@ -0,0 +1,9 @@
//! helpers for serde + serde derive attributes
pub fn default_true() -> bool {
true
}
pub fn is_default<T: Default + std::cmp::PartialEq>(t: &T) -> bool {
&T::default() == t
}

298
backend/src/worker.rs
View File

@@ -68,6 +68,7 @@ pub struct WorkerConfig {
pub disable_nuser: bool,
pub disable_nsjail: bool,
}
pub async fn run_worker(
db: &DB,
timeout: i32,
@@ -240,7 +241,7 @@ pub async fn run_worker(
}
}
}
tracing::error!(job_id = %job.id, "Error handling job: {err}");
tracing::error!(job_id = %job.id, err = err.alt(), "error handling job");
};
}
Ok(None) => (),
@@ -292,12 +293,8 @@ async fn handle_queued_job(
match job.job_kind {
JobKind::FlowPreview | JobKind::Flow => {
handle_flow(
&job,
db,
job.args.clone().unwrap_or_else(|| serde_json::Value::Null),
)
.await?;
let args = job.args.clone().unwrap_or(Value::Null);
handle_flow(&job, db, args).await?;
}
_ => {
let mut logs = "".to_string();
@@ -1301,7 +1298,7 @@ mod tests {
summary: Default::default(),
},
],
failure_module: Default::default(),
..Default::default()
};
let job = JobPayload::RawFlow { value: flow, path: None };
@@ -1694,6 +1691,291 @@ def main():
assert_eq!(json!({ "l": [0, 1, 2] }), result);
}
mod retry {
use super::*;
/// test helper provides some external state to help steps fail at specific points
struct Server {
addr: std::net::SocketAddr,
tx: tokio::sync::oneshot::Sender<()>,
task: tokio::task::JoinHandle<Vec<u8>>,
}
impl Server {
async fn start(responses: Vec<Option<u8>>) -> Self {
use tokio::net::TcpListener;
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
let sock = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = sock.local_addr().unwrap();
let task = tokio::task::spawn(async move {
tokio::pin!(rx);
let mut results = vec![];
for next in responses {
let (mut peer, _) = tokio::select! {
_ = &mut rx => break,
r = sock.accept() => r,
}
.unwrap();
let n = peer.read_u8().await.unwrap();
results.push(n);
if let Some(next) = next {
peer.write_u8(next).await.unwrap();
}
}
results
});
return Self { addr, tx, task };
}
async fn close(self) -> Vec<u8> {
let Self { task, tx, .. } = self;
drop(tx);
task.await.unwrap()
}
}
fn flow() -> FlowValue {
let inner = r#"
export async function main(index, port) {
const buf = new Uint8Array([0]);
const sock = await Deno.connect({ port });
await sock.write(new Uint8Array([index]));
if (await sock.read(buf) != 1) throw "read";
return buf[0];
}
"#;
let last = r#"
def main(last, port):
with __import__("socket").create_connection((None, port)) as sock:
sock.send(b'\xff')
return last + [sock.recv(1)[0]]
"#;
serde_json::from_value(serde_json::json!({
"modules": [{
"value": {
"type": "forloopflow",
"iterator": { "type": "javascript", "expr": "result.items" },
"skip_failures": false,
"modules": [{
"input_transform": {
"index": { "type": "javascript", "expr": "previous_result.iter.index" },
"port": { "type": "javascript", "expr": "flow_input.port" },
},
"value": {
"type": "rawscript",
"language": "deno",
"content": inner,
},
}]
}
}, {
"input_transform": {
"last": { "type": "javascript", "expr": "previous_result" },
"port": { "type": "javascript", "expr": "flow_input.port" },
},
"value": {
"type": "rawscript",
"language": "python3",
"content": last,
},
}],
"retry": { "constant": { "attempts": 2, "seconds": 0 } },
})).unwrap()
}
#[sqlx::test(fixtures("base"))]
async fn test_pass(db: DB) {
initialize_tracing().await;
/* fails twice in the loop, then once on the last step
* retry attempts is measured per-step, so it _retries_ at most two times on each step,
* which means it may run the step three times in total */
let (attempts, responses) = [
/* pass fail */
(0, Some(99)),
(1, None),
/* pass pass fail */
(0, Some(99)),
(1, Some(99)),
(2, None),
/* pass pass pass */
(0, Some(3)),
(1, Some(5)),
(2, Some(7)),
/* fail the last step once */
(0xff, None),
(0xff, Some(9)),
]
.into_iter()
.unzip::<_, _, Vec<_>, Vec<_>>();
let server = Server::start(responses).await;
let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None })
.arg("items", json!(["unused", "unused", "unused"]))
.arg("port", json!(server.addr.port()))
.wait_until_complete(&db)
.await;
assert_eq!(server.close().await, attempts);
assert_eq!(json!([3, 5, 7, 9]), result);
}
#[sqlx::test(fixtures("base"))]
async fn test_fail_step_zero(db: DB) {
initialize_tracing().await;
/* attempt and fail the first step three times and stop */
let (attempts, responses) = [
/* pass fail x3 */
(0, Some(99)),
(1, None),
(0, Some(99)),
(1, None),
(0, Some(99)),
(1, None),
]
.into_iter()
.unzip::<_, _, Vec<_>, Vec<_>>();
let server = Server::start(responses).await;
let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None })
.arg("items", json!(["unused", "unused", "unused"]))
.arg("port", json!(server.addr.port()))
.wait_until_complete(&db)
.await;
assert_eq!(server.close().await, attempts);
assert!(result["error"]
.as_str()
.unwrap()
.contains(r#"Uncaught (in promise) "read""#));
}
#[sqlx::test(fixtures("base"))]
async fn test_fail_step_one(db: DB) {
initialize_tracing().await;
/* attempt and fail the first step three times and stop */
let (attempts, responses) = [
/* fail once, then pass */
(0, None),
(0, Some(1)),
(1, Some(2)),
(2, Some(3)),
/* fail three times */
(0xff, None),
(0xff, None),
(0xff, None),
]
.into_iter()
.unzip::<_, _, Vec<_>, Vec<_>>();
let server = Server::start(responses).await;
let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None })
.arg("items", json!(["unused", "unused", "unused"]))
.arg("port", json!(server.addr.port()))
.wait_until_complete(&db)
.await;
assert_eq!(server.close().await, attempts);
assert!(result["error"]
.as_str()
.unwrap()
.contains("index out of range"));
}
#[sqlx::test(fixtures("base"))]
async fn test_with_failure_module(db: DB) {
let value = serde_json::from_value(json!({
"modules": [{
"input_transform": { "port": { "type": "javascript", "expr": "flow_input.port" } },
"value": {
"type": "rawscript",
"language": "python3",
"content": r#"
def main(port):
with __import__("socket").create_connection((None, port)) as sock:
sock.send(b'\x00')
return sock.recv(1)[0]"#,
},
}],
"failure_module": {
"input_transform": { "error": { "type": "javascript", "expr": "previous_result", },
"port": { "type": "javascript", "expr": "flow_input.port" } },
"value": {
"type": "rawscript",
"language": "python3",
"content": r#"
def main(error, port):
with __import__("socket").create_connection((None, port)) as sock:
sock.send(b'\xff')
return { "recv": sock.recv(1)[0], "from failure module": error }"#,
}
},
"retry": { "constant": { "attempts": 1, "seconds": 0 } },
}))
.unwrap();
let (attempts, responses) = [
/* fail the first step twice */
(0x00, None),
(0x00, None),
/* and the failure module once */
(0xff, None),
(0xff, Some(42)),
]
.into_iter()
.unzip::<_, _, Vec<_>, Vec<_>>();
let server = Server::start(responses).await;
let result = RunJob::from(JobPayload::RawFlow { value, path: None })
.arg("port", json!(server.addr.port()))
.wait_until_complete(&db)
.await;
assert_eq!(server.close().await, attempts);
assert_eq!(
result,
json!({
"recv": 42,
"from failure module": {
"error": "\
Error during execution of the script\nlast 5 logs lines:\n \
File \"/tmp/main.py\", line 14, in <module>\n \
res = inner_script.main(**kwargs)\n \
File \"/tmp/inner.py\", line 5, in main\n \
return sock.recv(1)[0]\nIndexError: index out of range"
}
})
);
}
#[sqlx::test(fixtures("base"))]
async fn bad_values_max(db: DB) {
let value = serde_json::from_value(json!({
"modules": [{
"value": { "type": "rawscript", "language": "python3", "content": "asdf" },
}],
"retry": { "exponential": { "attempts": 50, "seconds": 60 } },
}))
.unwrap();
let result = RunJob::from(JobPayload::RawFlow { value, path: None })
.wait_until_complete(&db)
.await;
assert_eq!(
result,
json!({"error": "Bad request: retry interval exceeds the maximum of 21600 seconds"})
)
}
}
#[sqlx::test(fixtures("base"))]
async fn test_iteration(db: DB) {
initialize_tracing().await;

207
backend/src/worker_flow.rs
View File

@@ -1,14 +1,16 @@
use std::collections::HashMap;
use std::time::Duration;
use crate::{
db::DB,
error::{self, Error},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry},
jobs::{
add_completed_job, add_completed_job_error, get_queued_job, postprocess_queued_job, push,
script_path_to_payload, JobPayload, QueuedJob,
},
js_eval::{eval_timeout, EvalCreds},
more_serde::is_default,
users::create_token_for_owner,
worker,
};
@@ -20,11 +22,24 @@ use serde_json::{json, Map, Value};
use tracing::instrument;
use uuid::Uuid;
const MAX_RETRY_ATTEMPTS: u16 = 1000;
const MAX_RETRY_INTERVAL: Duration = /* six hours */ Duration::from_secs(6 * 60 * 60);
#[derive(Serialize, Deserialize, Debug)]
pub struct FlowStatus {
pub step: i32,
pub modules: Vec<FlowStatusModule>,
pub failure_module: FlowStatusModule,
#[serde(default)]
#[serde(skip_serializing_if = "is_default")]
pub retry: RetryStatus,
}
#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq)]
#[serde(default)]
pub struct RetryStatus {
pub fail_count: u16,
pub previous_result: Option<Value>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
@@ -51,6 +66,7 @@ impl FlowStatus {
step: 0,
modules: vec![FlowStatusModule::WaitingForPriorSteps; f.modules.len()],
failure_module: FlowStatusModule::WaitingForPriorSteps,
retry: RetryStatus { fail_count: 0, previous_result: None },
}
}
}
@@ -128,6 +144,7 @@ pub async fn update_flow_status_after_job_completion(
}
};
/* is_last_step is true when the step_counter (the next step index) is an invalid index */
let is_last_step = usize::try_from(step_counter)
.map(|i| !(..old_status.modules.len()).contains(&i))
.unwrap_or(true);
@@ -161,10 +178,6 @@ pub async fn update_flow_status_after_job_completion(
tracing::debug!("UPDATE: {:?}", new_status);
let flow_job = get_queued_job(flow, w_id, &mut tx)
.await?
.ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
let stop_early = success
&& if let Some(expr) = stop_early_expr.clone() {
compute_stop_early(expr, result.clone()).await?
@@ -172,7 +185,7 @@ pub async fn update_flow_status_after_job_completion(
false
};
let result = match new_status {
let result = match &new_status {
FlowStatusModule::Success { forloop_jobs: Some(jobs), .. } => {
let results = sqlx::query_as(
"
@@ -192,14 +205,48 @@ pub async fn update_flow_status_after_job_completion(
.await?;
json!(results)
}
_ => result.clone(),
_ => result,
};
if matches!(&new_status, FlowStatusModule::Success { .. }) {
sqlx::query(
"
UPDATE queue
SET flow_status = flow_status - 'retry'
WHERE id = $1
RETURNING flow_status
",
)
.bind(flow)
.execute(&mut tx)
.await
.context("remove flow status retry")?;
}
let flow_job = get_queued_job(flow, w_id, &mut tx)
.await?
.ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
let should_continue_flow = match success {
_ if stop_early => false,
_ if flow_job.canceled => false,
true => !is_last_step,
false if skip_loop_failures => !is_last_step,
false
if next_retry(
&flow_job
.parse_raw_flow()
.map(|module| module.retry)
.unwrap_or_default(),
&flow_job
.parse_flow_status()
.map(|status| status.retry)
.unwrap_or_default(),
)
.is_some() =>
{
true
}
false if has_failure_module(flow, &mut tx).await? => true,
false => false,
};
@@ -302,6 +349,13 @@ async fn has_failure_module<'c>(
.map_err(|e| Error::InternalErr(format!("error during retrieval of has_failure_module: {e}")))
}
fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> {
(status.fail_count <= MAX_RETRY_ATTEMPTS)
.then(|| &retry)
.and_then(|retry| retry.interval(status.fail_count))
.map(|d| (status.fail_count + 1, std::cmp::min(d, MAX_RETRY_INTERVAL)))
}
async fn compute_stop_early(expr: String, result: serde_json::Value) -> error::Result<bool> {
match eval_timeout(expr, [("result".to_string(), result)].into(), None, vec![]).await? {
serde_json::Value::Bool(true) => Ok(true),
@@ -417,6 +471,19 @@ pub async fn handle_flow(
)))?;
}
if flow.retry.max_attempts() > MAX_RETRY_ATTEMPTS {
Err(Error::BadRequest(format!(
"retry attempts exceeds the maximum of {MAX_RETRY_ATTEMPTS}"
)))?
}
if matches!(flow.retry.max_interval(), Some(interval) if interval > MAX_RETRY_INTERVAL) {
let max = MAX_RETRY_INTERVAL.as_secs();
Err(Error::BadRequest(format!(
"retry interval exceeds the maximum of {max} seconds"
)))?
}
push_next_flow_job(
flow_job,
flow,
@@ -435,7 +502,7 @@ async fn push_next_flow_job(
flow: FlowValue,
schedule_path: Option<String>,
db: &sqlx::Pool<sqlx::Postgres>,
last_result: serde_json::Value,
mut last_result: serde_json::Value,
) -> anyhow::Result<()> {
let status: FlowStatus =
serde_json::from_value::<FlowStatus>(flow_job.flow_status.clone().unwrap_or_default())
@@ -449,13 +516,14 @@ async fn push_next_flow_job(
let mut module: &FlowModule = flow
.modules
.get(i)
.or_else(|| flow.failure_module.as_ref())
.with_context(|| format!("no module at index {}", status.step))?;
let mut status_module: FlowStatusModule = status
.modules
.get(i)
.cloned()
.with_context(|| format!("no status at index {}", status.step))?;
.unwrap_or_else(|| status.failure_module.clone());
tracing::debug!(
"PUSH: module: {:#?}, status: {:#?}",
@@ -463,21 +531,98 @@ async fn push_next_flow_job(
status_module
);
if matches!(&status_module, FlowStatusModule::Success { .. }) {
anyhow::bail!("no job for {status_module:?}")
} else if matches!(&status_module, FlowStatusModule::Failure { .. }) {
/* To run to the failure module, call push_next_flow_job 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. */
i = flow.modules.len();
module = flow
.failure_module
.as_ref()
/* If this fails, it's a update_flow_status_in_progress shouldn't have called
* handle_flow to get here. */
.context("missing failure module")?;
status_module = status.failure_module.clone();
};
let mut scheduled_for_o = None;
if matches!(&status_module, FlowStatusModule::Failure { .. }) {
if let Some((fail_count, retry_in)) = next_retry(&flow.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;
}
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 */
if flow.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,
}))
.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)
&& flow.retry.has_attempts()
&& 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")?;
}
/* Don't evaluate `module.input_transforms` after iteration has begun. Instead, args are
* carried through the Iterator by the InProgress variant. */
@@ -554,7 +699,7 @@ async fn push_next_flow_job(
ARRAY['step'], $3)
WHERE id = $4
RETURNING *
"#,
"#,
)
.bind(status.step)
.bind(json!(FlowStatusModule::Success {
@@ -651,6 +796,7 @@ async fn push_next_flow_job(
value: FlowValue {
modules: (*modules).clone(),
failure_module: flow.failure_module.clone(),
retry: Default::default(),
},
path: Some(format!("{}/{}", flow_job.script_path(), status.step)),
},
@@ -670,7 +816,7 @@ async fn push_next_flow_job(
Some(args.clone()),
&flow_job.created_by,
flow_job.permissioned_as.to_owned(),
None,
scheduled_for_o,
schedule_path,
Some(flow_job.id),
true,
@@ -743,3 +889,12 @@ impl IntoArray for Value {
}
}
}
fn from_now(duration: Duration) -> chrono::DateTime<chrono::Utc> {
// "This function errors when original duration is larger than
// the maximum value supported for this type."
chrono::Duration::from_std(duration)
.ok()
.and_then(|d| chrono::Utc::now().checked_add_signed(d))
.unwrap_or(chrono::DateTime::<chrono::Utc>::MAX_UTC)
}

24
openflow.openapi.yaml
View File

@@ -42,6 +42,25 @@ components:
$ref: "#/components/schemas/FlowModule"
failure_module:
$ref: "#/components/schemas/FlowModule"
retry:
type: object
properties:
constant:
type: object
properties:
attempts:
type: integer
seconds:
type: integer
exponential:
type: object
properties:
attempts:
type: integer
multiplier:
type: integer
seconds:
type: integer
required:
- modules
@@ -194,6 +213,11 @@ components:
$ref: "#/components/schemas/FlowStatusModule"
failure_module:
$ref: "#/components/schemas/FlowStatusModule"
retry:
type: object
properties:
fail_count:
type: integer
required:
- step
- modules