From d69d002b82e4ce92828251fe1e0c82f9234792d4 Mon Sep 17 00:00:00 2001 From: sqwishy Date: Sat, 3 Sep 2022 11:38:31 -0700 Subject: [PATCH] flow step retry feature (#493) * 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 --- backend/src/error.rs | 7 + backend/src/flows.rs | 252 +++++++++++++++++++++++++++++-- backend/src/jobs.rs | 14 +- backend/src/lib.rs | 1 + backend/src/more_serde.rs | 9 ++ backend/src/worker.rs | 298 ++++++++++++++++++++++++++++++++++++- backend/src/worker_flow.rs | 207 ++++++++++++++++++++++---- openflow.openapi.yaml | 24 +++ 8 files changed, 763 insertions(+), 49 deletions(-) create mode 100644 backend/src/more_serde.rs diff --git a/backend/src/error.rs b/backend/src/error.rs index 963b77bb02..e36dc713f4 100644 --- a/backend/src/error.rs +++ b/backend/src/error.rs @@ -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(e: T) -> anyhow::Error { From::from(e) } diff --git a/backend/src/flows.rs b/backend/src/flows.rs index f373fd0010..7b471f9509 100644 --- a/backend/src/flows.rs +++ b/backend/src/flows.rs @@ -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, } -#[derive(Deserialize, Serialize, Debug, Clone)] +#[derive(Deserialize, Serialize, Debug, Clone, Default)] pub struct FlowValue { pub modules: Vec, + #[serde(default)] + #[serde(skip_serializing_if = "is_default")] + pub retry: Retry, + #[serde(default)] pub failure_module: Option, } + #[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 { + 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 { + 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, @@ -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::>() + ); + + 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::>() + ); + + assert_eq!(Some(81 * SECOND), retry.max_interval()); + } } diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index ba6adaccb0..f82dc1dd4b 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -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 { + self.raw_flow + .as_ref() + .and_then(|v| serde_json::from_value::(v.clone()).ok()) + } + + pub fn parse_flow_status(&self) -> Option { + self.flow_status + .as_ref() + .and_then(|v| serde_json::from_value::(v.clone()).ok()) + } } #[derive(Debug, sqlx::FromRow, Serialize)] diff --git a/backend/src/lib.rs b/backend/src/lib.rs index bd0b6a7124..46f14e57e2 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -34,6 +34,7 @@ mod granular_acls; mod groups; mod jobs; mod js_eval; +mod more_serde; mod oauth2; mod parser; mod parser_py; diff --git a/backend/src/more_serde.rs b/backend/src/more_serde.rs new file mode 100644 index 0000000000..66432d6565 --- /dev/null +++ b/backend/src/more_serde.rs @@ -0,0 +1,9 @@ +//! helpers for serde + serde derive attributes + +pub fn default_true() -> bool { + true +} + +pub fn is_default(t: &T) -> bool { + &T::default() == t +} diff --git a/backend/src/worker.rs b/backend/src/worker.rs index fcf428c193..ee0cdb82ea 100644 --- a/backend/src/worker.rs +++ b/backend/src/worker.rs @@ -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>, + } + + impl Server { + async fn start(responses: Vec>) -> 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 { + 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 \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; diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs index 4cdaf3ade3..1a9db7621b 100644 --- a/backend/src/worker_flow.rs +++ b/backend/src/worker_flow.rs @@ -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, 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, } #[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 { 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, db: &sqlx::Pool, - last_result: serde_json::Value, + mut last_result: serde_json::Value, ) -> anyhow::Result<()> { let status: FlowStatus = serde_json::from_value::(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 { + // "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::::MAX_UTC) +} diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 0f45596342..c795ea3a3d 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -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