diff --git a/backend/.sqlx/query-061ff848f258dc880bec81d923370c905e689f37c6d931ee4559c3cfd394e168.json b/backend/.sqlx/query-061ff848f258dc880bec81d923370c905e689f37c6d931ee4559c3cfd394e168.json new file mode 100644 index 0000000000..b765d8bfdd --- /dev/null +++ b/backend/.sqlx/query-061ff848f258dc880bec81d923370c905e689f37c6d931ee4559c3cfd394e168.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "flow_status", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "061ff848f258dc880bec81d923370c905e689f37c6d931ee4559c3cfd394e168" +} diff --git a/backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json b/backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json index dcd14131c0..67bdf6ea9d 100644 --- a/backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json +++ b/backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json b/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json index ad66cfa646..272aefc9d2 100644 --- a/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json +++ b/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json @@ -67,7 +67,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json b/backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json index 2e4467041b..4df35d6cf7 100644 --- a/backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json +++ b/backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json @@ -28,7 +28,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json b/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json index c84e2ab8e2..e6282c7e80 100644 --- a/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json +++ b/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json @@ -60,7 +60,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json b/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json index 8b139967a6..61da7601af 100644 --- a/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json +++ b/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json @@ -40,7 +40,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json b/backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json index b0437fe128..de4127eb14 100644 --- a/backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json +++ b/backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json @@ -46,7 +46,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json b/backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json index 0385dfd1d5..40ab7503e5 100644 --- a/backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json +++ b/backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json b/backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json index 091c4160e3..3b43ee29f5 100644 --- a/backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json +++ b/backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json b/backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json index 7df86d1a0e..c0c9ef873f 100644 --- a/backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json +++ b/backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json @@ -37,7 +37,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 6084bf9c77..506ce3e848 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -276,7 +276,7 @@ mod suspend_resume { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None }) + let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) .arg("n", json!(1)) .arg("port", json!(port)) .push(&db) @@ -357,7 +357,7 @@ mod suspend_resume { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) .arg("n", json!(1)) .arg("op", json!("cancel")) .arg("port", json!(port)) @@ -381,7 +381,7 @@ mod suspend_resume { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None }) + let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) .arg("n", json!(1)) .arg("port", json!(port)) .push(&db) @@ -576,7 +576,7 @@ def main(last, port): .into_iter() .unzip::<_, _, Vec<_>, Vec<_>>(); let server = Server::start(responses).await; - let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None }) .arg("items", json!(["unused", "unused", "unused"])) .arg("port", json!(server.addr.port())) .run_until_complete(&db, server.addr.port()) @@ -605,7 +605,7 @@ def main(last, port): .into_iter() .unzip::<_, _, Vec<_>, Vec<_>>(); let server = Server::start(responses).await; - let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None }) .arg("items", json!(["unused", "unused", "unused"])) .arg("port", json!(server.addr.port())) .run_until_complete(&db, server.addr.port()) @@ -646,7 +646,7 @@ def main(last, port): .into_iter() .unzip::<_, _, Vec<_>, Vec<_>>(); let server = Server::start(responses).await; - let job = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None }) + let job = RunJob::from(JobPayload::RawFlow { value: flow_forloop_retry(), path: None, restarted_from: None }) .arg("items", json!(["unused", "unused", "unused"])) .arg("port", json!(server.addr.port())) .run_until_complete(&db, server.addr.port()) @@ -712,7 +712,7 @@ def main(error, port): .into_iter() .unzip::<_, _, Vec<_>, Vec<_>>(); let server = Server::start(responses).await; - let cjob = RunJob::from(JobPayload::RawFlow { value, path: None }) + let cjob = RunJob::from(JobPayload::RawFlow { value, path: None, restarted_from: None }) .arg("port", json!(server.addr.port())) .run_until_complete(&db, server.addr.port()) .await; @@ -770,7 +770,7 @@ async fn test_iteration(db: Pool) { })) .unwrap(); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("items", json!([])) .run_until_complete(&db, server.addr.port()) .await @@ -779,7 +779,7 @@ async fn test_iteration(db: Pool) { assert_eq!(result, serde_json::json!([])); /* Don't actually test that this does 257 jobs or that will take forever. */ - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("items", json!((0..257).collect::>())) .run_until_complete(&db, server.addr.port()) .await @@ -827,7 +827,7 @@ async fn test_iteration_parallel(db: Pool) { })) .unwrap(); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("items", json!([])) .run_until_complete(&db, server.addr.port()) .await @@ -836,7 +836,7 @@ async fn test_iteration_parallel(db: Pool) { assert_eq!(result, serde_json::json!([])); /* Don't actually test that this does 257 jobs or that will take forever. */ - let job = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let job = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("items", json!((0..50).collect::>())) .run_until_complete(&db, server.addr.port()) .await; @@ -1136,7 +1136,7 @@ async fn test_deno_flow(db: Pool) { } }; - let job = JobPayload::RawFlow { value: flow, path: None }; + let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let port = server.addr.port(); for i in 0..50 { @@ -1176,7 +1176,7 @@ async fn test_identity(db: Pool) { })) .unwrap(); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .run_until_complete(&db, server.addr.port()) .await .json_result() @@ -1365,7 +1365,7 @@ async fn test_deno_flow_same_worker(db: Pool) { ..Default::default() }; - let job = JobPayload::RawFlow { value: flow, path: None }; + let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, job.clone(), server.addr.port()) .await @@ -1421,7 +1421,7 @@ async fn test_flow_result_by_id(db: Pool) { })) .unwrap(); - let job = JobPayload::RawFlow { value: flow, path: None }; + let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, job.clone(), port) .await .json_result() @@ -1463,7 +1463,7 @@ async fn test_stop_after_if(db: Pool) { ], })) .unwrap(); - let job = JobPayload::RawFlow { value: flow, path: None }; + let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = RunJob::from(job.clone()) .arg("n", json!(123)) @@ -1521,7 +1521,7 @@ async fn test_stop_after_if_nested(db: Pool) { ], })) .unwrap(); - let job = JobPayload::RawFlow { value: flow, path: None }; + let job = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = RunJob::from(job.clone()) .arg("n", json!(123)) @@ -1586,7 +1586,7 @@ async fn test_python_flow(db: Pool) { println!("python flow iteration: {}", i); let result = run_job_in_new_worker_until_complete( &db, - JobPayload::RawFlow { value: flow.clone(), path: None }, + JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }, port, ) .await @@ -1621,7 +1621,7 @@ async fn test_python_flow_2(db: Pool) { println!("python flow iteration: {}", i); let result = run_job_in_new_worker_until_complete( &db, - JobPayload::RawFlow { value: flow.clone(), path: None }, + JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }, port, ) .await @@ -1837,7 +1837,7 @@ async fn test_empty_loop(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -1876,7 +1876,7 @@ async fn test_invalid_first_step(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let job = run_job_in_new_worker_until_complete(&db, flow, port).await; assert_eq!( @@ -1918,7 +1918,7 @@ async fn test_empty_loop_2(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -1973,7 +1973,7 @@ async fn test_step_after_loop(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2041,7 +2041,7 @@ async fn test_branchone_simple(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2077,7 +2077,7 @@ async fn test_branchone_with_cond(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2115,7 +2115,7 @@ async fn test_branchall_sequential(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2152,7 +2152,7 @@ async fn test_branchall_simple(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2199,7 +2199,7 @@ async fn test_branchall_skip_failure(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2236,7 +2236,7 @@ async fn test_branchall_skip_failure(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2300,7 +2300,7 @@ async fn test_branchone_nested(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2357,7 +2357,7 @@ async fn test_branchall_nested(db: Pool) { })) .unwrap(); - let flow = JobPayload::RawFlow { value: flow, path: None }; + let flow = JobPayload::RawFlow { value: flow, path: None, restarted_from: None }; let result = run_job_in_new_worker_until_complete(&db, flow, port) .await .json_result() @@ -2421,7 +2421,7 @@ async fn test_failure_module(db: Pool) { })) .unwrap(); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("n", json!(0)) .run_until_complete(&db, port) .await @@ -2437,7 +2437,7 @@ async fn test_failure_module(db: Pool) { .unwrap() .contains("[]")); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("n", json!(1)) .run_until_complete(&db, port) .await @@ -2453,7 +2453,7 @@ async fn test_failure_module(db: Pool) { .unwrap() .contains("[0]")); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("n", json!(2)) .run_until_complete(&db, port) .await @@ -2469,7 +2469,7 @@ async fn test_failure_module(db: Pool) { .unwrap() .contains("[0,1]")); - let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None }) + let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) .arg("n", json!(3)) .run_until_complete(&db, port) .await diff --git a/backend/windmill-api/openapi-deref.yaml b/backend/windmill-api/openapi-deref.yaml index ebd60169c6..5b7dcfd76a 100644 --- a/backend/windmill-api/openapi-deref.yaml +++ b/backend/windmill-api/openapi-deref.yaml @@ -1,6 +1,6 @@ openapi: 3.0.3 info: - version: 1.188.1 + version: 1.192.0 title: Windmill API contact: name: Windmill Team @@ -2329,7 +2329,7 @@ paths: application/json: schema: type: object - properties: &ref_151 + properties: &ref_152 access_token: type: string expires_in: @@ -2340,7 +2340,7 @@ paths: type: array items: type: string - required: &ref_152 + required: &ref_153 - access_token /w/{workspace}/oauth/create_account: post: @@ -3189,32 +3189,32 @@ paths: id: type: string value: - oneOf: &ref_173 + oneOf: &ref_174 - type: object - properties: &ref_157 + properties: &ref_158 input_transforms: type: object additionalProperties: oneOf: &ref_23 - type: object - properties: &ref_153 + properties: &ref_154 value: {} type: type: string enum: - javascript - required: &ref_154 + required: &ref_155 - expr - type - type: object - properties: &ref_155 + properties: &ref_156 expr: type: string type: type: string enum: - javascript - required: &ref_156 + required: &ref_157 - expr - type discriminator: &ref_24 @@ -3253,13 +3253,13 @@ paths: type: number concurrency_time_window_s: type: number - required: &ref_158 + required: &ref_159 - type - content - language - input_transforms - type: object - properties: &ref_159 + properties: &ref_160 input_transforms: type: object additionalProperties: @@ -3273,12 +3273,12 @@ paths: type: string enum: - script - required: &ref_160 + required: &ref_161 - type - path - input_transforms - type: object - properties: &ref_161 + properties: &ref_162 input_transforms: type: object additionalProperties: @@ -3290,12 +3290,12 @@ paths: type: string enum: - flow - required: &ref_162 + required: &ref_163 - type - path - input_transforms - type: object - properties: &ref_163 + properties: &ref_164 modules: type: array items: @@ -3317,13 +3317,13 @@ paths: type: boolean parallelism: type: integer - required: &ref_164 + required: &ref_165 - modules - iterator - skip_failures - type - type: object - properties: &ref_165 + properties: &ref_166 branches: type: array items: @@ -3354,12 +3354,12 @@ paths: type: string enum: - branchone - required: &ref_166 + required: &ref_167 - branches - default - type - type: object - properties: &ref_167 + properties: &ref_168 branches: type: array items: @@ -3384,28 +3384,28 @@ paths: - branchall parallel: type: boolean - required: &ref_168 + required: &ref_169 - branches - type - type: object - properties: &ref_169 + properties: &ref_170 type: type: string enum: - identity flow: type: boolean - required: &ref_170 + required: &ref_171 - type - type: object - properties: &ref_171 + properties: &ref_172 type: type: string enum: - graphql - required: &ref_172 + required: &ref_173 - type - discriminator: &ref_174 + discriminator: &ref_175 propertyName: type mapping: rawscript: '#/components/schemas/RawScript' @@ -3461,7 +3461,7 @@ paths: type: number retry: type: object - properties: &ref_175 + properties: &ref_176 constant: type: object properties: @@ -3981,13 +3981,9 @@ paths: items: type: string concurrent_limit: - type: array - items: - type: integer + type: integer concurrency_time_window_s: - type: array - items: - type: integer + type: integer cache_ttl: type: number dedicated_worker: @@ -4169,13 +4165,9 @@ paths: items: type: string concurrent_limit: - type: array - items: - type: integer + type: integer concurrency_time_window_s: - type: array - items: - type: integer + type: integer cache_ttl: type: number dedicated_worker: @@ -5411,7 +5403,7 @@ paths: type: array items: type: object - properties: &ref_148 + properties: &ref_149 workspace_id: type: string path: @@ -5429,7 +5421,7 @@ paths: edited_at: type: string format: date-time - required: &ref_149 + required: &ref_150 - workspace_id - path - summary @@ -5565,7 +5557,7 @@ paths: type: array items: type: object - properties: &ref_146 + properties: &ref_147 id: type: integer workspace_id: @@ -5591,7 +5583,7 @@ paths: - viewer - publisher - anonymous - required: &ref_147 + required: &ref_148 - id - workspace_id - path @@ -5764,7 +5756,7 @@ paths: content: application/json: schema: - allOf: &ref_150 + allOf: &ref_151 - type: object properties: *ref_45 required: *ref_46 @@ -6142,6 +6134,84 @@ paths: schema: type: string format: uuid + /w/{workspace}/jobs/restart/f/{id}/from/{step_id}: + post: + summary: restart a completed flow at a given step + operationId: restartFlowAtStep + tags: + - job + parameters: + - name: workspace + in: path + required: true + schema: *ref_0 + - name: id + in: path + required: true + schema: &ref_69 + type: string + format: uuid + - name: step_id + description: step id to restart the flow from + required: true + in: path + schema: + type: string + - name: scheduled_for + description: when to schedule this job (leave empty for immediate run) + in: query + schema: + type: string + format: date-time + - name: scheduled_in_secs + description: schedule the script to execute in the number of seconds starting now + in: query + schema: + type: integer + - name: parent_job + description: >- + The parent job that is at the origin and responsible for the + execution of this script if any + in: query + schema: *ref_34 + - name: job_id + description: >- + The job id to assign to the created job. if missing, job is chosen + randomly using the ULID scheme. If a job id already exists in the + queue or as a completed job, the request to create one will fail + (Bad Request) + in: query + schema: *ref_35 + - name: include_header + description: > + List of headers's keys (separated with ',') whove value are added to + the args + + Header's key lowercased and '-'' replaced to '_' such that + 'Content-Type' becomes the 'content_type' arg key + in: query + schema: *ref_36 + - name: invisible_to_owner + description: make the run invisible to the the flow owner (default false) + in: query + schema: + type: boolean + requestBody: + description: flow args + required: true + content: + application/json: + schema: + type: object + additionalProperties: *ref_14 + responses: + '201': + description: job created + content: + text/plain: + schema: + type: string + format: uuid /w/{workspace}/jobs/run/h/{hash}: post: summary: run script by hash @@ -6344,6 +6414,14 @@ paths: additionalProperties: *ref_14 tag: type: string + restarted_from: + type: object + properties: &ref_146 + flow_job_id: + type: string + format: uuid + step_id: + type: string required: &ref_145 - value - content @@ -7049,7 +7127,7 @@ paths: schema: type: array items: - allOf: &ref_69 + allOf: &ref_70 - oneOf: - type: object properties: *ref_65 @@ -7064,7 +7142,7 @@ paths: enum: - CompletedJob - QueuedJob - discriminator: &ref_70 + discriminator: &ref_71 propertyName: type /jobs/db_clock: get: @@ -7093,17 +7171,15 @@ paths: - name: id in: path required: true - schema: &ref_71 - type: string - format: uuid + schema: *ref_69 responses: '200': description: job details content: application/json: schema: - allOf: *ref_69 - discriminator: *ref_70 + allOf: *ref_70 + discriminator: *ref_71 /w/{workspace}/jobs_u/get_logs/{id}: get: summary: get job logs @@ -7118,7 +7194,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 responses: '200': description: job details @@ -7140,7 +7216,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: running in: query schema: @@ -7179,7 +7255,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 responses: '200': description: job details @@ -7203,7 +7279,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 responses: '200': description: result @@ -7224,7 +7300,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: get_started in: query schema: &ref_98 @@ -7259,7 +7335,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 responses: '200': description: job details @@ -7283,7 +7359,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 requestBody: description: reason required: true @@ -7315,7 +7391,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 requestBody: description: reason required: true @@ -7347,7 +7423,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: resume_id in: path required: true @@ -7378,7 +7454,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: resume_id in: path required: true @@ -7420,7 +7496,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: payload description: > The base64 encoded payload that has been encoded as a JSON. e.g how @@ -7463,7 +7539,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: resume_id in: path required: true @@ -7505,7 +7581,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 requestBody: required: true content: @@ -7533,7 +7609,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: resume_id in: path required: true @@ -7568,7 +7644,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: resume_id in: path required: true @@ -7610,7 +7686,7 @@ paths: - name: id in: path required: true - schema: *ref_71 + schema: *ref_69 - name: resume_id in: path required: true @@ -7634,8 +7710,8 @@ paths: type: object properties: job: - allOf: *ref_69 - discriminator: *ref_70 + allOf: *ref_70 + discriminator: *ref_71 approvers: type: array items: @@ -9249,7 +9325,7 @@ components: name: id in: path required: true - schema: *ref_71 + schema: *ref_69 Path: name: path in: path @@ -9492,8 +9568,8 @@ components: properties: *ref_65 required: *ref_66 Job: - allOf: *ref_69 - discriminator: *ref_70 + allOf: *ref_70 + discriminator: *ref_71 User: type: object properties: *ref_8 @@ -9747,23 +9823,26 @@ components: type: object properties: *ref_144 required: *ref_145 + RestartedFrom: + type: object + properties: *ref_146 Policy: type: object properties: *ref_44 ListableApp: type: object - properties: *ref_146 - required: *ref_147 + properties: *ref_147 + required: *ref_148 ListableRawApp: type: object - properties: *ref_148 - required: *ref_149 + properties: *ref_149 + required: *ref_150 AppWithLastVersion: type: object properties: *ref_45 required: *ref_46 AppWithLastVersionWDraft: - allOf: *ref_150 + allOf: *ref_151 SlackToken: type: object properties: @@ -9785,64 +9864,64 @@ components: - bot TokenResponse: type: object - properties: *ref_151 - required: *ref_152 + properties: *ref_152 + required: *ref_153 HubScriptKind: name: kind schema: *ref_28 StaticTransform: type: object - properties: *ref_153 - required: *ref_154 + properties: *ref_154 + required: *ref_155 JavascriptTransform: type: object - properties: *ref_155 - required: *ref_156 + properties: *ref_156 + required: *ref_157 InputTransform: oneOf: *ref_23 discriminator: *ref_24 RawScript: type: object - properties: *ref_157 - required: *ref_158 + properties: *ref_158 + required: *ref_159 PathScript: type: object - properties: *ref_159 - required: *ref_160 + properties: *ref_160 + required: *ref_161 PathFlow: type: object - properties: *ref_161 - required: *ref_162 + properties: *ref_162 + required: *ref_163 FlowModule: type: object properties: *ref_25 required: *ref_26 ForloopFlow: type: object - properties: *ref_163 - required: *ref_164 + properties: *ref_164 + required: *ref_165 BranchOne: type: object - properties: *ref_165 - required: *ref_166 + properties: *ref_166 + required: *ref_167 BranchAll: type: object - properties: *ref_167 - required: *ref_168 + properties: *ref_168 + required: *ref_169 Identity: type: object - properties: *ref_169 - required: *ref_170 + properties: *ref_170 + required: *ref_171 Graphql: type: object - properties: *ref_171 - required: *ref_172 + properties: *ref_172 + required: *ref_173 FlowModuleValue: - oneOf: *ref_173 - discriminator: *ref_174 + oneOf: *ref_174 + discriminator: *ref_175 Retry: type: object - properties: *ref_175 + properties: *ref_176 FlowValue: type: object properties: *ref_47 diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 550f791a28..2177160392 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -4229,6 +4229,58 @@ paths: type: string format: uuid + /w/{workspace}/jobs/restart/f/{id}/from/{step_id}: + post: + summary: restart a completed flow at a given step + operationId: restartFlowAtStep + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/JobId" + - name: step_id + description: step id to restart the flow from + required: true + in: path + schema: + type: string + - name: scheduled_for + description: when to schedule this job (leave empty for immediate run) + in: query + schema: + type: string + format: date-time + - name: scheduled_in_secs + description: schedule the script to execute in the number of seconds starting now + in: query + schema: + type: integer + - $ref: "#/components/parameters/ParentJob" + - $ref: "#/components/parameters/NewJobId" + - $ref: "#/components/parameters/IncludeHeader" + - name: invisible_to_owner + description: make the run invisible to the the flow owner (default false) + in: query + schema: + type: boolean + + requestBody: + description: flow args + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ScriptArgs" + + responses: + "201": + description: job created + content: + text/plain: + schema: + type: string + format: uuid + /w/{workspace}/jobs/run/h/{hash}: post: summary: run script by hash @@ -7865,12 +7917,23 @@ components: $ref: "#/components/schemas/ScriptArgs" tag: type: string + restarted_from: + $ref: "#/components/schemas/RestartedFrom" required: - value - content - args + RestartedFrom: + type: object + properties: + flow_job_id: + type: string + format: uuid + step_id: + type: string + Policy: type: object properties: diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 512f4263df..48b4712531 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -8,6 +8,7 @@ use serde_json::value::RawValue; use std::collections::HashMap; +use windmill_common::flow_status::RestartedFrom; use crate::db::ApiAuthed; @@ -41,7 +42,7 @@ use windmill_common::{ error::{self, to_anyhow, Error}, flow_status::{Approval, FlowStatus, FlowStatusModule}, flows::FlowValue, - jobs::{script_path_to_payload, JobKind, JobPayload, QueuedJob, RawCode}, + jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode}, oauth2::HmacSha256, scripts::{ScriptHash, ScriptLang}, users::username_to_permissioned_as, @@ -63,6 +64,10 @@ pub fn workspaced_service() -> Router { .head(|| async { "" }) .layer(cors.clone()), ) + .route( + "/restart/f/:job_id/from/*step_id", + post(restart_flow).head(|| async { "" }).layer(cors.clone()), + ) .route( "/run/p/*script_path", post(run_job_by_path) @@ -398,64 +403,6 @@ async fn get_job_logs( Ok(text) } -#[derive(Debug, sqlx::FromRow, Serialize)] -pub struct CompletedJob { - pub workspace_id: String, - pub id: Uuid, - #[serde(skip_serializing_if = "Option::is_none")] - pub parent_job: Option, - pub created_by: String, - pub created_at: chrono::DateTime, - pub started_at: chrono::DateTime, - pub duration_ms: i64, - pub success: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub script_hash: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub script_path: Option, - pub args: Option>>>, - #[serde(skip_serializing_if = "Option::is_none")] - pub result: Option>>, - #[serde(skip_serializing_if = "Option::is_none")] - pub logs: Option, - pub deleted: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub raw_code: Option, - pub canceled: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub canceled_by: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub canceled_reason: Option, - pub job_kind: JobKind, - #[serde(skip_serializing_if = "Option::is_none")] - pub schedule_path: Option, - pub permissioned_as: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub flow_status: Option>>, - #[serde(skip_serializing_if = "Option::is_none")] - pub raw_flow: Option>>, - pub is_flow_step: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub language: Option, - pub is_skipped: bool, - pub email: String, - pub visible_to_owner: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub mem_peak: Option, - pub tag: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub priority: Option, -} - -impl CompletedJob { - pub fn json_result(&self) -> Option { - self.result - .as_ref() - .map(|r| serde_json::from_str(r.get()).ok()) - .flatten() - } -} - #[derive(Debug, sqlx::FromRow, Serialize)] pub struct ListableCompletedJob { pub r#type: String, @@ -501,12 +448,6 @@ pub struct ListableCompletedJob { pub priority: Option, } -impl<'a> IntoResponse for CompletedJob { - fn into_response(self) -> Response { - Json(self).into_response() - } -} - #[derive(Deserialize, Clone)] pub struct RunJobQuery { scheduled_for: Option>, @@ -1531,6 +1472,7 @@ struct PreviewFlow { path: Option, args: Option>, tag: Option, + restarted_from: Option, } pub struct QueryOrBody(pub Option); @@ -1686,6 +1628,73 @@ pub async fn run_flow_by_path( Ok((StatusCode::CREATED, uuid.to_string())) } +pub async fn restart_flow( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Extension(rsmq): Extension>, + Path((w_id, job_id, step_id)): Path<(String, Uuid, String)>, + Query(run_query): Query, +) -> error::Result<(StatusCode, String)> { + #[cfg(not(feature = "enterprise"))] + { + return Err(Error::BadRequest( + "Restarting a flow is a feature only available in enterprise version".to_string(), + )); + } + + #[cfg(feature = "enterprise")] + check_license_key_valid().await?; + + let completed_job = sqlx::query_as::<_, CompletedJob>( + "SELECT * from completed_job WHERE id = $1 and workspace_id = $2", + ) + .bind(job_id) + .bind(&w_id) + .fetch_optional(&db) + .await? + .with_context(|| "Unable to find completed job with the given job UUID")?; + + let flow_path = completed_job + .script_path + .with_context(|| "No flow path set for completed flow job")?; + check_scopes(&authed, || format!("run:flow/{flow_path}"))?; + + let push_args = completed_job + .args + .map(|json| PushArgs { args: json.clone(), extra: json.0 }); + + let scheduled_for = run_query.get_scheduled_for(&db).await?; + let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq); + + let (uuid, tx) = push( + &db, + tx, + &w_id, + JobPayload::RestartedFlow { completed_job_id: job_id, step_id: step_id }, + push_args, + &authed.username, + &authed.email, + username_to_permissioned_as(&authed.username), + scheduled_for, + None, + run_query.parent_job, + run_query.parent_job, + run_query.job_id, + false, + false, + None, + !run_query.invisible_to_owner.unwrap_or(false), + Some(completed_job.tag), + None, + None, + completed_job.priority, + ) + .await?; + tx.commit().await?; + Ok((StatusCode::CREATED, uuid.to_string())) +} + pub async fn run_job_by_path( authed: ApiAuthed, Extension(db): Extension, @@ -2324,7 +2333,7 @@ async fn add_batch_jobs( let mut uuids: Vec = Vec::new(); let payload = if let Some(ref fv) = batch_info.flow_value { - JobPayload::RawFlow { value: fv.clone(), path: None } + JobPayload::RawFlow { value: fv.clone(), path: None, restarted_from: None } } else { if let Some(path) = batch_info.path.as_ref() { JobPayload::Flow(path.to_string()) @@ -2442,7 +2451,11 @@ async fn run_preview_flow_job( &db, tx, &w_id, - JobPayload::RawFlow { value: raw_flow.value, path: raw_flow.path }, + JobPayload::RawFlow { + value: raw_flow.value, + path: raw_flow.path, + restarted_from: raw_flow.restarted_from, + }, raw_flow.args.unwrap_or_default(), &authed.username, &authed.email, @@ -2769,7 +2782,8 @@ async fn get_completed_job<'a>( .await?; let job = not_found_if_none(job_o, "Completed Job", id.to_string())?; - Ok(CompletedJob::from_row(&job)?.into_response()) + let response = Json(CompletedJob::from_row(&job)?).into_response(); + Ok(response) } #[derive(FromRow)] @@ -2900,5 +2914,6 @@ async fn delete_completed_job<'a>( .await?; tx.commit().await?; - Ok(CompletedJob::from_row(&job)?.into_response()) + let response = Json(CompletedJob::from_row(&job)?).into_response(); + Ok(response) } diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index fc6c6b6dc7..17e27081f9 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -34,6 +34,8 @@ pub struct FlowStatus { pub retry: RetryStatus, #[serde(skip_serializing_if = "Option::is_none")] pub approval_conditions: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub restarted_from: Option, } #[derive(Serialize, Deserialize, Debug, Clone, Default)] @@ -50,6 +52,13 @@ pub struct ApprovalConditions { pub user_groups_required: Vec, } +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +#[serde(default)] +pub struct RestartedFrom { + pub flow_job_id: Uuid, + pub step_id: String, +} + #[derive(Serialize, Deserialize, Debug, Clone)] pub struct Iterator { pub index: usize, @@ -179,6 +188,13 @@ impl FlowStatusModule { FlowStatusModule::Failure { id, .. } => id.clone(), } } + + pub fn is_failure(&self) -> bool { + match self { + FlowStatusModule::Failure { .. } => true, + _ => false, + } + } } impl FlowStatus { @@ -202,6 +218,7 @@ impl FlowStatus { }, }, retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, + restarted_from: None, } } diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index e53e00c52e..0681b97967 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -7,7 +7,7 @@ use uuid::Uuid; use crate::{ error::{self, Error}, - flow_status::FlowStatus, + flow_status::{FlowStatus, RestartedFrom}, flows::FlowValue, get_latest_deployed_hash_for_path, scripts::{ScriptHash, ScriptLang}, @@ -186,6 +186,82 @@ impl Default for QueuedJob { } } +#[derive(Debug, sqlx::FromRow, Serialize, Clone)] +pub struct CompletedJob { + pub workspace_id: String, + pub id: Uuid, + #[serde(skip_serializing_if = "Option::is_none")] + pub parent_job: Option, + pub created_by: String, + pub created_at: chrono::DateTime, + pub started_at: chrono::DateTime, + pub duration_ms: i64, + pub success: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub script_hash: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub script_path: Option, + pub args: Option>>>, + #[serde(skip_serializing_if = "Option::is_none")] + pub result: Option>>, + #[serde(skip_serializing_if = "Option::is_none")] + pub logs: Option, + pub deleted: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub raw_code: Option, + pub canceled: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub canceled_by: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub canceled_reason: Option, + pub job_kind: JobKind, + #[serde(skip_serializing_if = "Option::is_none")] + pub schedule_path: Option, + pub permissioned_as: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub flow_status: Option>>, + #[serde(skip_serializing_if = "Option::is_none")] + pub raw_flow: Option>>, + pub is_flow_step: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub language: Option, + pub is_skipped: bool, + pub email: String, + pub visible_to_owner: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub mem_peak: Option, + pub tag: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub priority: Option, +} + +impl CompletedJob { + pub fn json_result(&self) -> Option { + self.result + .as_ref() + .map(|r| serde_json::from_str(r.get()).ok()) + .flatten() + } + + pub fn parse_raw_flow(&self) -> Option { + self.raw_flow + .as_ref() + .and_then(|v| serde_json::from_str::((**v).get()).ok()) + } + + pub fn parse_flow_status(&self) -> Option { + self.flow_status + .as_ref() + .and_then(|v| serde_json::from_str::((**v).get()).ok()) + } +} + +#[derive(sqlx::FromRow)] +pub struct BranchResults<'a> { + pub result: &'a RawValue, + pub id: Uuid, +} + #[derive(Debug, Clone)] pub enum JobPayload { ScriptHub { @@ -216,9 +292,14 @@ pub enum JobPayload { version: i64, }, Flow(String), + RestartedFlow { + completed_job_id: Uuid, + step_id: String, + }, RawFlow { value: FlowValue, path: Option, + restarted_from: Option, }, Identity, Noop, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 87acf1a9fd..3d7f39600b 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6,7 +6,7 @@ * LICENSE-AGPL for a copy of the license. */ -use std::{collections::HashMap, sync::Arc, vec}; +use std::{collections::HashMap, iter, sync::Arc, vec}; use anyhow::Context; use async_recursion::async_recursion; @@ -19,6 +19,7 @@ use axum::{ }; use bigdecimal::ToPrimitive; use chrono::{DateTime, Duration, Utc}; +use itertools::Itertools; use prometheus::IntCounter; use reqwest::{ header::{HeaderMap, CONTENT_TYPE}, @@ -41,12 +42,13 @@ use windmill_common::{ db::{Authed, UserDB}, error::{self, Error}, flow_status::{ - FlowStatus, FlowStatusModule, JobResult, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL, + FlowStatus, FlowStatusModule, FlowStatusModuleWParent, JobResult, RestartedFrom, + RetryStatus, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL, }, flows::{FlowModule, FlowModuleValue, FlowValue}, jobs::{ - get_payload_tag_from_prefixed_path, script_path_to_payload, JobKind, JobPayload, QueuedJob, - RawCode, + get_payload_tag_from_prefixed_path, script_path_to_payload, CompletedJob, JobKind, + JobPayload, QueuedJob, RawCode, }, schedule::{schedule_to_user, Schedule}, scripts::{ScriptHash, ScriptLang}, @@ -1467,13 +1469,60 @@ struct ResultR { result: Option>>, } -#[async_recursion] pub async fn get_result_by_id( db: Pool, w_id: String, flow_id: Uuid, node_id: String, json_path: Option, +) -> error::Result> { + match get_result_by_id_from_running_flow( + &db, + w_id.as_str(), + &flow_id, + node_id.as_str(), + json_path.clone(), + ) + .await + { + Ok(res) => Ok(res), + Err(_) => { + let running_flow_job = sqlx::query_as::<_, QueuedJob>( + "SELECT * FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $1), $1) = id AND workspace_id = $2" + ).bind(flow_id) + .bind(&w_id) + .fetch_optional(&db) + .await?; + + let restarted_from = windmill_common::utils::not_found_if_none( + running_flow_job + .map(|fj| fj.parse_flow_status()) + .flatten() + .map(|status| status.restarted_from) + .flatten(), + "Flow result by id in leaf jobs", + format!("{}, {}", flow_id, node_id), + )?; + + get_result_by_id_from_original_flow( + &db, + w_id.as_str(), + &restarted_from.flow_job_id, + node_id.as_str(), + json_path.clone(), + ) + .await + } + } +} + +#[async_recursion] +async fn get_result_by_id_from_running_flow( + db: &Pool, + w_id: &str, + flow_id: &Uuid, + node_id: &str, + json_path: Option, ) -> error::Result> { let flow_job_result = sqlx::query!( "SELECT leaf_jobs->$1::text as leaf_jobs, parent_job FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3", @@ -1481,7 +1530,7 @@ pub async fn get_result_by_id( flow_id, w_id, ) - .fetch_optional(&db) + .fetch_optional(db) .await?; let flow_job_result = windmill_common::utils::not_found_if_none( @@ -1498,11 +1547,11 @@ pub async fn get_result_by_id( if job_result.is_none() && flow_job_result.parent_job.is_some() { let parent_job = flow_job_result.parent_job.unwrap(); let root_job = sqlx::query_scalar!("SELECT root_job FROM queue WHERE id = $1", parent_job) - .fetch_optional(&db) + .fetch_optional(db) .await? .flatten() .unwrap_or(parent_job); - return get_result_by_id(db, w_id, root_job, node_id, json_path).await; + return get_result_by_id_from_running_flow(db, w_id, &root_job, node_id, json_path).await; } let result_id = windmill_common::utils::not_found_if_none( @@ -1511,21 +1560,181 @@ pub async fn get_result_by_id( format!("{}, {}", flow_id, node_id), )?; - let value = match result_id { - JobResult::ListJob(x) => { + extract_result_from_job_result(db, w_id, result_id, json_path).await +} + +#[async_recursion] +async fn get_result_by_id_from_original_flow( + db: &Pool, + w_id: &str, + completed_flow_id: &Uuid, + node_id: &str, + json_path: Option, +) -> error::Result> { + let flow_job = sqlx::query_as::<_, CompletedJob>( + "SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2", + ) + .bind(completed_flow_id) + .bind(w_id) + .fetch_optional(db) + .await?; + + let flow_job = windmill_common::utils::not_found_if_none( + flow_job, + "Flow result by id in leaf jobs", + format!("{}", completed_flow_id), + )?; + + let mut leaf_jobs_for_flow = HashMap::::new(); + compute_leaf_jobs_for_completed_flow( + db, + w_id, + flow_job.id, + Some(flow_job.clone()), + &mut leaf_jobs_for_flow, + ) + .await?; + + tracing::debug!( + "Fetching leaf jobs for flow {} : {:?}", + flow_job.id, + leaf_jobs_for_flow + ); + if !leaf_jobs_for_flow.contains_key(&node_id.to_string()) { + // if the flow is itself a restart flow, the step job might be from the upstream flow + let restarted_from = windmill_common::utils::not_found_if_none( + flow_job + .parse_flow_status() + .map(|status| status.restarted_from) + .flatten(), + "Flow result by id in leaf jobs", + format!("{}", completed_flow_id), + )?; + return get_result_by_id_from_original_flow( + db, + w_id, + &restarted_from.flow_job_id, + node_id, + json_path, + ) + .await; + } + + // if the job is in the leaf_jobs map, then fetch its result and return + let leaf_job_uuid = leaf_jobs_for_flow.get(&node_id.to_string()).unwrap(); + extract_result_from_job_result(db, w_id, leaf_job_uuid.to_owned(), json_path).await +} + +#[async_recursion] +async fn compute_leaf_jobs_for_completed_flow( + db: &Pool, + w_id: &str, + completed_flow_id: Uuid, + completed_flow_row: Option, // if provided, will be used, otherwise completed_flow_id must be set and the job definition will be pulled from DB + recursive_result: &mut HashMap, +) -> error::Result<()> { + let flow_status = match completed_flow_row { + Some(job) => job.parse_flow_status(), + None => { + let job_status_raw = sqlx::query_scalar!( + "SELECT flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2", + completed_flow_id, + w_id, + ) + .fetch_one(db) + .await?; + job_status_raw + .map(|raw| serde_json::from_value::(raw).ok()) + .flatten() + } + }; + + let flow_job_status_modules = windmill_common::utils::not_found_if_none( + flow_status.map(|fs| fs.modules), + "Flow result by id in leaf jobs", + format!("{}", completed_flow_id), + )?; + + let children_jobs = sqlx::query_as::<_, (Uuid, JobKind)>( + "SELECT id, job_kind FROM completed_job WHERE parent_job = $1 AND workspace_id = $2", + ) + .bind(completed_flow_id) + .bind(w_id) + .fetch_all(db) + .await?; + + for child_job in children_jobs { + let child_job_id = child_job.0; + let child_job_kind = child_job.1; + match child_job_kind { + JobKind::Script | JobKind::Preview | JobKind::Script_Hub => { + // if is potentially a leaf job. Get its step_id from the initial flow definition and add it to the result map + for module in &flow_job_status_modules { + if module.job().map(|id| id == child_job_id).unwrap_or(false) { + recursive_result.insert(module.id(), JobResult::SingleJob(child_job_id)); + } + } + } + JobKind::Flow | JobKind::FlowPreview => { + // Extract the leaf job for this flow and add them to the result map + for module in &flow_job_status_modules { + // we add the module as an element of ListJob for this step ID and recursiively extract leaf job of the sub-flow + match module { + FlowStatusModule::Success { flow_jobs: Some(jobs), .. } => { + for job in jobs { + if *job == child_job_id { + let new_list_job = match recursive_result.get(&module.id()) { + Some(JobResult::ListJob(jobs_list)) => jobs_list + .into_iter() + .chain(iter::once(&child_job_id)) + .cloned() + .collect_vec(), + _ => iter::once(&child_job_id).cloned().collect_vec(), + }; + recursive_result + .insert(module.id(), JobResult::ListJob(new_list_job)); + } + } + } + _ => {} + } + } + compute_leaf_jobs_for_completed_flow( + db, + w_id, + child_job_id, + None, + recursive_result, + ) + .await?; + } + _ => {} // do nothing + } + } + Ok(()) +} + +async fn extract_result_from_job_result( + db: &Pool, + w_id: &str, + job_result: JobResult, + json_path: Option, +) -> error::Result> { + match job_result { + JobResult::ListJob(job_ids) => { let rows = sqlx::query( "SELECT result FROM completed_job WHERE id = ANY($1) AND workspace_id = $2", ) - .bind(x.as_slice()) + .bind(job_ids.as_slice()) .bind(w_id) - .fetch_all(&db) + .fetch_all(db) .await? .into_iter() .filter_map(|x| ResultR::from_row(&x).ok().and_then(|x| x.result)) .collect::>>>(); - to_raw_value(&rows) + Ok(to_raw_value(&rows)) } - JobResult::SingleJob(x) => sqlx::query( + JobResult::SingleJob(x) => Ok(sqlx::query( "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2", ) .bind(x) @@ -1535,7 +1744,7 @@ pub async fn get_result_by_id( .map(|x| x.split(".").map(|x| x.to_string()).collect::>()) .unwrap_or_default(), ) - .fetch_optional(&db) + .fetch_optional(db) .await? .map(|r| { ResultR::from_row(&r) @@ -1543,10 +1752,8 @@ pub async fn get_result_by_id( .and_then(|x| x.result.map(|x| x.0)) }) .flatten() - .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)), - }; - - Ok(value) + .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null))), + } } #[instrument(level = "trace", skip_all)] @@ -1896,6 +2103,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection raw_code_tuple, job_kind, mut raw_flow, + flow_status, language, concurrent_limit, concurrency_time_window_s, @@ -1918,6 +2126,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, JobKind::Script, None, + None, Some(language), concurrent_limit, concurrency_time_window_s, @@ -1939,6 +2148,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ) } JobPayload::Code(RawCode { @@ -1955,6 +2165,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection Some((content, lock)), JobKind::Preview, None, + None, Some(language), concurrent_limit, concurrency_time_window_s, @@ -1968,6 +2179,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection Some((dependencies, None)), JobKind::Dependencies, None, + None, Some(language), None, None, @@ -1995,7 +2207,8 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection Some(path), None, JobKind::FlowDependencies, - Some(value), + Some(value.clone()), + Some(FlowStatus::new(&value)), // this is a new flow being pushed, flow_status is set to flow_value None, None, None, @@ -2016,20 +2229,56 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ), - JobPayload::RawFlow { value, path } => ( - None, - path, - None, - JobKind::FlowPreview, - Some(value.clone()), - None, - value.concurrent_limit.clone(), - value.concurrency_time_window_s, - value.cache_ttl.map(|x| x as i32), - None, - value.priority, - ), + JobPayload::RawFlow { value, path, restarted_from } => { + let flow_status: FlowStatus = match restarted_from { + Some(restarted_from_val) => { + let (_, _, step_n, truncated_modules, _) = restarted_flows_resolution( + _db, + workspace_id, + Some(value.clone()), + restarted_from_val.flow_job_id, + restarted_from_val.step_id.as_str(), + ) + .await?; + FlowStatus { + step: step_n, + modules: truncated_modules, + // failure_module is reset + failure_module: FlowStatusModuleWParent { + parent_module: None, + module_status: FlowStatusModule::WaitingForPriorSteps { + id: "failure".to_string(), + }, + }, + // retry status is reset + retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, + // TODO: for now, flows with approval conditions aren't supported for restart + approval_conditions: None, + restarted_from: Some(RestartedFrom { + flow_job_id: restarted_from_val.flow_job_id, + step_id: restarted_from_val.step_id, + }), + } + } + _ => FlowStatus::new(&value), // this is a new flow being pushed, flow_status is set to flow_value + }; + ( + None, + path, + None, + JobKind::FlowPreview, + Some(value.clone()), + Some(flow_status), + None, + value.concurrent_limit.clone(), + value.concurrency_time_window_s, + value.cache_ttl.map(|x| x as i32), + None, + value.priority, + ) + } JobPayload::Flow(flow) => { let value_json = fetch_scalar_isolated!( sqlx::query_scalar!( @@ -2051,6 +2300,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, JobKind::Flow, Some(value.clone()), + Some(FlowStatus::new(&value)), // this is a new flow being pushed, flow_status is set to flow_value None, value.concurrent_limit.clone(), value.concurrency_time_window_s, @@ -2059,6 +2309,47 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection value.priority, ) } + JobPayload::RestartedFlow { completed_job_id, step_id } => { + let (flow_path, raw_flow, step_n, truncated_modules, priority) = + restarted_flows_resolution( + _db, + workspace_id, + None, + completed_job_id, + step_id.as_str(), + ) + .await?; + let restarted_flow_status = FlowStatus { + step: step_n, + modules: truncated_modules, + // failure_module is reset + failure_module: FlowStatusModuleWParent { + parent_module: None, + module_status: FlowStatusModule::WaitingForPriorSteps { + id: "failure".to_string(), + }, + }, + // retry status is reset + retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, + // TODO: for now, flows with approval conditions aren't supported for restart + approval_conditions: None, + restarted_from: Some(RestartedFrom { flow_job_id: completed_job_id, step_id }), + }; + ( + None, + flow_path, + None, + JobKind::Flow, + Some(raw_flow.clone()), + Some(restarted_flow_status), + None, + raw_flow.concurrent_limit, + raw_flow.concurrency_time_window_s, + raw_flow.cache_ttl.map(|x| x as i32), + None, + priority, + ) + } JobPayload::Identity => ( None, None, @@ -2071,6 +2362,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ), JobPayload::Noop => ( None, @@ -2084,6 +2376,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, None, None, + None, ), }; @@ -2156,8 +2449,6 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection .map(|e| (Some(e.0), e.1)) .unwrap_or_else(|| (None, None)); - let flow_status = raw_flow.as_ref().map(FlowStatus::new); - let tag = if dedicated_worker.is_some_and(|x| x) { format!( "{}:{}", @@ -2317,3 +2608,102 @@ pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value { let canceler = job.canceled_by.as_deref().unwrap_or_else(|| "unknown"); serde_json::json!({"message": format!("Job canceled: {reason} by {canceler}"), "name": "Canceled", "reason": reason, "canceler": canceler}) } + +async fn restarted_flows_resolution( + db: &Pool, + workspace_id: &str, + flow_value_if_any: Option, + completed_flow_id: Uuid, + restart_step_id: &str, +) -> Result< + ( + Option, + FlowValue, + i32, + Vec, + Option, + ), + Error, +> { + let completed_job = sqlx::query_as::<_, CompletedJob>( + "SELECT * FROM completed_job WHERE id = $1 and workspace_id = $2", + ) + .bind(completed_flow_id) + .bind(workspace_id) + .fetch_one(db) // TODO: should we try to use the passed-in `tx` here? + .await + .map_err(|err| { + Error::InternalErr(format!( + "completed job not found for UUID {} in workspace {}: {}", + completed_flow_id, workspace_id, err + )) + })?; + + let raw_flow = completed_job + .parse_raw_flow() + .ok_or(Error::InternalErr(format!( + "Unable to parse raw definition for job {} in workspace {}", + completed_flow_id, workspace_id, + )))?; + let flow_status = completed_job + .parse_flow_status() + .ok_or(Error::InternalErr(format!( + "Unable to parse flow status for job {} in workspace {}", + completed_flow_id, workspace_id, + )))?; + + let mut step_n = 0; + let mut dependent_module = false; + let mut truncated_modules: Vec = vec![]; + for module in flow_status.modules { + if flow_value_if_any + .clone() + .map(|fv| { + fv.modules + .iter() + .find(|flow_value_module| flow_value_module.id == module.id()) + .is_none() + }) + .unwrap_or(false) + { + // skip module as it doesn't appear in the flow_value anymore + continue; + } + if module.id() == restart_step_id || dependent_module { + // if the module ID is the one we want to restart the flow at, or if it's past it in the flow, + // set the module as WaitingForPriorSteps as it needs to be re-run + truncated_modules.push(FlowStatusModule::WaitingForPriorSteps { id: module.id() }); + dependent_module = true; + } else { + // else we simply "transfer" the module from the completed flow to the new one if it's a success + step_n = step_n + 1; + match module.clone() { + FlowStatusModule::Success { + id: _, + job: _, + flow_jobs: _, + branch_chosen: _, + approvers: _, + } => Ok(truncated_modules.push(module)), + _ => Err(Error::InternalErr(format!( + "Flow cannot be restarted from a non successful module", + ))), + }?; + } + } + if !dependent_module { + // step not found in flow. + return Err(Error::InternalErr(format!( + "Flow cannot be restarted from step {} as it could not be found.", + restart_step_id + ))); + } + + return Ok(( + completed_job.script_path, + raw_flow, + step_n, + truncated_modules, + completed_job.priority, + )); +} diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 067c254f61..e85d3d3d26 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2150,13 +2150,12 @@ async fn handle_queued_job( }; match job.job_kind { JobKind::FlowPreview | JobKind::Flow => { - let args = job.get_args(); let timer = worker_flow_initial_transition_duration.map(|x| x.start_timer()); handle_flow( &job, db, &client.get_authed().await, - to_raw_value(&args), + None, same_worker_tx, worker_dir, rsmq, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 6e1e800123..e43848c3dc 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -28,7 +28,8 @@ use windmill_common::flow_status::{ ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, }; use windmill_common::jobs::{ - script_hash_to_tag_and_limits, script_path_to_payload, JobPayload, QueuedJob, RawCode, + script_hash_to_tag_and_limits, script_path_to_payload, BranchResults, JobPayload, QueuedJob, + RawCode, }; use windmill_common::worker::to_raw_value; use windmill_common::{ @@ -137,12 +138,6 @@ pub struct RecUpdateFlowStatusAfterJobCompletion { skip_error_handler: bool, } -#[derive(FromRow)] -pub struct BranchResults<'a> { - pub result: &'a RawValue, - pub id: Uuid, -} - #[derive(FromRow)] pub struct SkipIfStopped { pub skip_if_stopped: Option, @@ -508,35 +503,7 @@ pub async fn update_flow_status_after_job_completion_internal< let nresult = match &new_status { Some(FlowStatusModule::Success { flow_jobs: Some(jobs), .. }) | Some(FlowStatusModule::Failure { flow_jobs: Some(jobs), .. }) => { - let results = sqlx::query( - " - SELECT result, id - FROM completed_job - WHERE id = ANY($1) - AND workspace_id = $2 - ", - ) - .bind(jobs.as_slice()) - .bind(w_id) - .fetch_all(&mut tx) - .await? - .into_iter() - .map(|r| { - let br = BranchResults::from_row(&r).unwrap(); - (br.id, br.result.to_owned()) - }) - .collect::>(); - - let results = jobs - .iter() - .map(|j| { - results.get(j).ok_or_else(|| { - Error::InternalErr(format!("missing job result for {}", j)) - }) - }) - .collect::, _>>()?; - - to_raw_value(&results) + retrieve_flow_jobs_results(db, w_id, jobs).await? } _ => result.to_owned(), }; @@ -690,7 +657,7 @@ pub async fn update_flow_status_after_job_completion_internal< &flow_job, db, client, - nresult.to_owned(), + Some(nresult.to_owned()), same_worker_tx.clone(), worker_dir, rsmq.clone(), @@ -741,6 +708,42 @@ pub async fn update_flow_status_after_job_completion_internal< } } +async fn retrieve_flow_jobs_results( + db: &DB, + w_id: &str, + job_uuids: &Vec, +) -> error::Result> { + let results = sqlx::query( + " + SELECT result, id + FROM completed_job + WHERE id = ANY($1) + AND workspace_id = $2 + ", + ) + .bind(job_uuids.as_slice()) + .bind(w_id) + .fetch_all(db) + .await? + .into_iter() + .map(|r| { + let br = BranchResults::from_row(&r).unwrap(); + (br.id, br.result.to_owned()) + }) + .collect::>(); + + let results = job_uuids + .iter() + .map(|j| { + results + .get(j) + .ok_or_else(|| Error::InternalErr(format!("missing job result for {}", j))) + }) + .collect::, _>>()?; + + Ok(to_raw_value(&results)) +} + fn get_module(flow_job: &QueuedJob, module_index: Option) -> Option { let raw_flow = flow_job.parse_raw_flow(); if let Some(raw_flow) = raw_flow { @@ -835,7 +838,7 @@ async fn compute_bool_from_expr( ) -> error::Result { let mut context = HashMap::with_capacity(if resumes.is_some() { 7 } else { 3 }); context.insert("result".to_string(), result.clone()); - context.insert("previous_result".to_string(), result); + context.insert("previous_result".to_string(), result.clone()); if let Some(resumes) = resumes { context.insert("resume".to_string(), resumes.1); @@ -984,25 +987,19 @@ pub async fn handle_flow( flow_job: &QueuedJob, db: &sqlx::Pool, client: &AuthedClient, - last_result: Box, + last_result: Option>, same_worker_tx: Sender, worker_dir: &str, rsmq: Option, worker_name: &str, ) -> anyhow::Result<()> { - let value = flow_job - .raw_flow - .as_ref() - .ok_or_else(|| Error::InternalErr(format!("requiring a raw flow value")))? - .to_owned(); - let flow = serde_json::from_str::((*value.0).get())?; + let flow = flow_job + .parse_raw_flow() + .with_context(|| "Unable to parse flow definition")?; + let status = flow_job + .parse_flow_status() + .with_context(|| "Unable to parse flow status")?; - let status: FlowStatus = serde_json::from_str::( - (*flow_job.flow_status.clone().unwrap_or_default().0).get(), - ) - .with_context(|| format!("parse flow status {}", flow_job.id))?; - - tracing::debug!("handle_flow: {:#?}", flow_job.flow_status); push_next_flow_job( flow_job, status, @@ -1054,7 +1051,7 @@ async fn push_next_flow_job flow: FlowValue, db: &sqlx::Pool, client: &AuthedClient, - last_result: Box, + last_job_result: Option>, same_worker_tx: Sender, worker_dir: &str, rsmq: Option, @@ -1105,14 +1102,13 @@ async fn push_next_flow_job } let arc_flow_job_args = Arc::new(flow_job_args.clone()); - let mut arc_result = Arc::new(last_result); if i == 0 { if let Some(skip_expr) = &flow.skip_expr { let skip = compute_bool_from_expr( skip_expr.to_string(), arc_flow_job_args.clone(), - arc_result.clone(), + Arc::new(to_raw_value(&json!("{}"))), None, Some(client), None, @@ -1139,6 +1135,27 @@ async fn push_next_flow_job } } + // Compute and initialize last_job_result + let arc_last_job_result = if status_module.is_failure() { + // if job is being retried, pass the result of its previous failure + Arc::new(last_job_result.unwrap_or(to_raw_value(&json!("{}")))) + } else if i == 0 { + // if it's the first job executed in the flow, pass the flow args + Arc::new(to_raw_value(&flow_job.args)) + } else { + // else pass the last job result. Either from the function arg if it's set, or manually fetch it from the previous job + // having last_job_result empty can happen either when the job was suspended and is being restarted, or if it's a + // flow restart from a specific step + if last_job_result.is_some() { + Arc::new(last_job_result.unwrap()) + } else { + match get_previous_job_result(db, flow_job.workspace_id.as_str(), &status).await? { + None => Arc::new(to_raw_value(&json!("{}"))), + Some(previous_job_result) => Arc::new(previous_job_result), + } + } + }; + let mut resume_messages: Vec> = vec![]; let mut approvers: Vec = vec![]; @@ -1202,8 +1219,9 @@ async fn push_next_flow_job } InputTransform::Javascript { expr } => { let mut context = HashMap::with_capacity(2); - context.insert("result".to_string(), arc_result.clone()); - context.insert("previous_result".to_string(), arc_result.clone()); + context.insert("result".to_string(), arc_last_job_result.clone()); + context + .insert("previous_result".to_string(), arc_last_job_result.clone()); let eval_result = serde_json::from_str::>( eval_timeout( @@ -1275,20 +1293,6 @@ async fn push_next_flow_job .execute(&mut *tx) .await?; - /* If we are woken up after suspending, last_result will be the flow args, but we - * should use the result from the last job */ - if let FlowStatusModule::WaitingForEvents { .. } = &status_module { - arc_result = Arc::new( - sqlx::query_scalar::<_, Json>>( - "SELECT result FROM completed_job WHERE id = $1", - ) - .bind(last) - .fetch_one(&mut *tx) - .await? - .0, - ) - } - // Remove the approval conditions from the flow status sqlx::query( " @@ -1386,8 +1390,8 @@ async fn push_next_flow_job InputTransform::Static { value } => Ok(value), InputTransform::Javascript { expr } => { let mut context = HashMap::with_capacity(2); - context.insert("result".to_string(), arc_result.clone()); - context.insert("previous_result".to_string(), arc_result.clone()); + context.insert("result".to_string(), arc_last_job_result.clone()); + context.insert("previous_result".to_string(), arc_last_job_result.clone()); serde_json::from_str( eval_timeout( @@ -1562,7 +1566,7 @@ async fn push_next_flow_job let by_id = transform_context.as_ref().unwrap(); transform_input( arc_flow_job_args.clone(), - arc_result.clone(), + arc_last_job_result.clone(), input_transforms, resumes.clone(), resume.clone(), @@ -1573,8 +1577,10 @@ async fn push_next_flow_job .await } FlowModuleValue::Identity => serde_json::from_str( - &serde_json::to_string(&PreviousResult { previous_result: Some(&arc_result) }) - .unwrap(), + &serde_json::to_string(&PreviousResult { + previous_result: Some(&arc_last_job_result), + }) + .unwrap(), ) .map_err(|e| error::Error::InternalErr(format!("identity: {e}"))), @@ -1584,7 +1590,7 @@ async fn push_next_flow_job let next_flow_transform = compute_next_flow_transform( arc_flow_job_args.clone(), - arc_result.clone(), + arc_last_job_result.clone(), flow_job, &flow, transform_context, @@ -1701,7 +1707,7 @@ async fn push_next_flow_job let ctx = get_transform_context(&flow_job, &previous_id, &status).await?; transform_inp = transform_input( Arc::new(hm), - arc_result.clone(), + arc_last_job_result.clone(), input_transforms, resumes.clone(), resume.clone(), @@ -2013,7 +2019,7 @@ enum NextFlowTransform { async fn compute_next_flow_transform( arc_flow_job_args: Arc>>, - arc_result: Arc>, + arc_last_job_result: Arc>, flow_job: &QueuedJob, flow: &FlowValue, by_id: Option, @@ -2106,8 +2112,8 @@ async fn compute_next_flow_transform( InputTransform::Static { value } => to_raw_value(value), InputTransform::Javascript { expr } => { let mut context = HashMap::with_capacity(3); - context.insert("result".to_string(), arc_result.clone()); - context.insert("previous_result".to_string(), arc_result); + context.insert("result".to_string(), arc_last_job_result.clone()); + context.insert("previous_result".to_string(), arc_last_job_result); context.insert("resumes".to_string(), resumes); context.insert("resume".to_string(), resume); context.insert("approvers".to_string(), approvers); @@ -2197,6 +2203,7 @@ async fn compute_next_flow_transform( priority: None, }, path: inner_path, + restarted_from: None, }, tag: None, }); @@ -2256,6 +2263,7 @@ async fn compute_next_flow_transform( priority: None, }, path: Some(format!("{}/forloop", flow_job.script_path())), + restarted_from: None, }, tag: None, } @@ -2298,7 +2306,7 @@ async fn compute_next_flow_transform( let pred = compute_bool_from_expr( b.expr.to_string(), arc_flow_job_args.clone(), - arc_result.clone(), + arc_last_job_result.clone(), Some(idcontext.clone()), Some(client), Some((resumes.clone(), resume.clone(), approvers.clone())), @@ -2353,6 +2361,7 @@ async fn compute_next_flow_transform( flow_job.script_path(), status.step )), + restarted_from: None, }, tag: None, }), @@ -2399,6 +2408,7 @@ async fn compute_next_flow_transform( flow_job.script_path(), i )), + restarted_from: None, }, tag: None, } @@ -2462,6 +2472,7 @@ async fn compute_next_flow_transform( flow_job.script_path(), branch_status.branch )), + restarted_from: None, }, tag: None, }), @@ -2574,8 +2585,15 @@ fn from_now(duration: Duration) -> chrono::DateTime { .unwrap_or(chrono::DateTime::::MAX_UTC) } -/// returns previous module non-zero suspend count and job +/// returns previous module non-zero suspend count and job, if relevant fn needs_resume(flow: &FlowValue, status: &FlowStatus) -> Option<(Suspend, Uuid)> { + // for a restarted job, if the restarted step is just after a suspend, don't run the suspend + if status.restarted_from.is_some() { + let current_step_id = flow.modules.get(status.step as usize)?.id.clone(); + if status.restarted_from.as_ref().unwrap().step_id == current_step_id { + return None; + } + } let prev = usize::try_from(status.step) .ok() .and_then(|s| s.checked_sub(1))?; @@ -2596,3 +2614,32 @@ fn needs_resume(flow: &FlowValue, status: &FlowStatus) -> Option<(Suspend, Uuid) None } } + +// returns the result of the previous step of a running flow (if the job was successful) +async fn get_previous_job_result( + db: &sqlx::Pool, + w_id: &str, + flow_status: &FlowStatus, +) -> error::Result>> { + let prev = usize::try_from(flow_status.step) + .ok() + .and_then(|s| s.checked_sub(1)) + .with_context(|| "No step preceding the current one")?; + + match flow_status.modules.get(prev) { + Some(FlowStatusModule::Success { flow_jobs: Some(flow_jobs), .. }) => { + Ok(Some(retrieve_flow_jobs_results(db, w_id, flow_jobs).await?)) + } + Some(FlowStatusModule::Success { job, .. }) => Ok(Some( + sqlx::query_scalar::<_, Json>>( + "SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2", + ) + .bind(job) + .bind(w_id) + .fetch_one(db) + .await? + .0, + )), + _ => Ok(None), + } +} diff --git a/frontend/src/lib/components/FlowPreviewContent.svelte b/frontend/src/lib/components/FlowPreviewContent.svelte index b48f61ed9e..eec630219c 100644 --- a/frontend/src/lib/components/FlowPreviewContent.svelte +++ b/frontend/src/lib/components/FlowPreviewContent.svelte @@ -1,8 +1,8 @@