diff --git a/backend/migrations/20221023162721_remove_iscancel_resumejob.down.sql b/backend/migrations/20221023162721_remove_iscancel_resumejob.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20221023162721_remove_iscancel_resumejob.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20221023162721_remove_iscancel_resumejob.up.sql b/backend/migrations/20221023162721_remove_iscancel_resumejob.up.sql new file mode 100644 index 0000000000..f74ce8587e --- /dev/null +++ b/backend/migrations/20221023162721_remove_iscancel_resumejob.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here +ALTER TABLE resume_job DROP COLUMN is_cancel; +ALTER TABLE resume_job ADD COLUMN approver VARCHAR(50); \ No newline at end of file diff --git a/backend/openapi.yaml b/backend/openapi.yaml index dbdc7be04f..cc1b380b3d 100644 --- a/backend/openapi.yaml +++ b/backend/openapi.yaml @@ -2589,6 +2589,10 @@ paths: required: true schema: type: integer + - name: approver + in: query + schema: + type: string responses: "200": description: job signature @@ -2621,6 +2625,10 @@ paths: in: query schema: type: object + - name: approver + in: query + schema: + type: string responses: "201": description: job resumed @@ -2647,6 +2655,10 @@ paths: required: true schema: type: string + - name: approver + in: query + schema: + type: string requestBody: required: true content: @@ -2684,6 +2696,10 @@ paths: in: query schema: type: object + - name: approver + in: query + schema: + type: string responses: "201": description: job resumed @@ -2710,6 +2726,10 @@ paths: required: true schema: type: string + - name: approver + in: query + schema: + type: string requestBody: required: true content: @@ -2743,13 +2763,27 @@ paths: required: true schema: type: string + - name: approver + in: query + schema: + type: string responses: "200": description: parent flow details content: application/json: schema: - $ref: "#/components/schemas/Job" + type: object + properties: + job: + $ref: "#/components/schemas/Job" + approvers: + type: array + items: + type: string + required: + - job + - approvers /schedules/preview: post: diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index b8f3d94ff1..2822499b54 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -384,32 +384,6 @@ }, "query": "SELECT * FROM workspace LIMIT $1 OFFSET $2" }, - "13a10a404e892d6975d4913424880a75bfb2c6dfb5134c270c7c7225acb05db4": { - "describe": { - "columns": [ - { - "name": "value", - "ordinal": 0, - "type_info": "Jsonb" - }, - { - "name": "is_cancel", - "ordinal": 1, - "type_info": "Bool" - } - ], - "nullable": [ - false, - false - ], - "parameters": { - "Left": [ - "Uuid" - ] - } - }, - "query": "SELECT value, is_cancel FROM resume_job WHERE job = $1 ORDER BY created_at ASC" - }, "15de975d9be141c9ed9647935a508492aabbbddbf986d5c5c0f0c415293c432d": { "describe": { "columns": [], @@ -778,22 +752,6 @@ }, "query": "INSERT INTO variable\n (workspace_id, path, value, is_secret, description, account, is_oauth)\n VALUES ($1, $2, $3, $4, $5, $6, $7)" }, - "30d85ec4bfa7e81d2803ea9c0b638a01257ae2e9138dd82df526d28d4467df26": { - "describe": { - "columns": [], - "nullable": [], - "parameters": { - "Left": [ - "Uuid", - "Uuid", - "Uuid", - "Jsonb", - "Bool" - ] - } - }, - "query": "\n INSERT INTO resume_job\n (id, job, flow, value, is_cancel)\n VALUES ($1, $2, $3, $4, $5)\n " - }, "355dcb2cbebd13f0e3bdd4929b9e431b0e6d72716d1c4f9ab6af6adce5b5e4b3": { "describe": { "columns": [ @@ -1226,6 +1184,22 @@ }, "query": "SELECT * from resource_type WHERE name = $1 AND (workspace_id = $2 OR workspace_id = 'starter')" }, + "5a9adf5eec1d51060e6f5ecebea60b511189a22019d6c9249bb30a0cf4ae23dd": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Uuid", + "Uuid", + "Uuid", + "Jsonb", + "Varchar" + ] + } + }, + "query": "\n INSERT INTO resume_job\n (id, job, flow, value, approver)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (id) DO NOTHING\n " + }, "5b7a1d16d8109a65479ab33d411c60d14ea91d870fdff8606d7aa4ad39f0ba00": { "describe": { "columns": [ @@ -1959,6 +1933,32 @@ }, "query": "\n SELECT id, flow_status, suspend\n FROM queue\n WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1)\n FOR UPDATE\n " }, + "860ecd7137bf7963392e1102c642cfdb8f80db156d90229b0919c025ba4bab9c": { + "describe": { + "columns": [ + { + "name": "value", + "ordinal": 0, + "type_info": "Jsonb" + }, + { + "name": "approver", + "ordinal": 1, + "type_info": "Varchar" + } + ], + "nullable": [ + false, + true + ], + "parameters": { + "Left": [ + "Uuid" + ] + } + }, + "query": "SELECT value, approver FROM resume_job WHERE job = $1 ORDER BY created_at ASC" + }, "88a3f58a1a315200fdd2e4bb8638246ee21818f8aaaf56f6e9d7ddce1490d886": { "describe": { "columns": [ @@ -3272,6 +3272,26 @@ }, "query": "\n SELECT SUM(duration_ms)\n FROM completed_job\n WHERE permissioned_as = $1\n AND created_at > NOW() - INTERVAL '1200 seconds'\n AND workspace_id = $2" }, + "eceaa2cd70d08ec603072d7d49379ed19cfc78bd1f4a3a9fcbe33ed66d1be4be": { + "describe": { + "columns": [ + { + "name": "approver", + "ordinal": 0, + "type_info": "Varchar" + } + ], + "nullable": [ + true + ], + "parameters": { + "Left": [ + "Uuid" + ] + } + }, + "query": "\n SELECT approver\n FROM resume_job\n WHERE job = $1\n " + }, "ee99cb974fb35d2d92fa96cecb53b8ad3a729ac710f362dc55a576613b66f9be": { "describe": { "columns": [ diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index 03204ef0dc..0da51eeab4 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -65,7 +65,7 @@ pub fn workspaced_service() -> Router { .route("/run/preview_flow", post(run_preview_flow_job)) .route("/list", get(list_jobs)) .route("/queue/list", get(list_queue_jobs)) - .route("/queue/cancel/:id", post(cancel_job)) + .route("/queue/cancel/:id", post(cancel_job_api)) .route("/completed/list", get(list_completed_jobs)) .route("/completed/get/:id", get(get_completed_job)) .route("/completed/get_result/:id", get(get_completed_job_result)) @@ -802,41 +802,15 @@ async fn get_completed_job_result( Ok(Json(result)) } -async fn cancel_job( +async fn cancel_job_api( authed: Authed, Extension(user_db): Extension, Path((w_id, id)): Path<(String, Uuid)>, Json(CancelJob { reason }): Json, ) -> error::Result { - let mut tx = user_db.begin(&authed).await?; + let tx = user_db.begin(&authed).await?; - let job_option = sqlx::query_scalar!( - "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 \ - AND workspace_id = $4 RETURNING id", - &authed.username, - reason, - id, - w_id - ) - .fetch_optional(&mut tx) - .await?; - - let mut jobs = job_option.map(|j| vec![j]).unwrap_or_default(); - - while !jobs.is_empty() { - let p_job = jobs.pop(); - let new_jobs = sqlx::query_scalar!( - "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2 WHERE parent_job = $3 \ - AND workspace_id = $4 RETURNING id", - &authed.username, - reason, - p_job, - w_id - ) - .fetch_all(&mut tx) - .await?; - jobs.extend(new_jobs); - } + let (mut tx, job_option) = cancel_job(&authed.username, reason, id, &w_id, tx).await?; if let Some(id) = job_option { audit_log( @@ -865,6 +839,41 @@ async fn cancel_job( } } +async fn cancel_job<'c>( + username: &str, + reason: Option, + id: Uuid, + w_id: &str, + mut tx: Transaction<'c, Postgres>, +) -> error::Result<(Transaction<'c, Postgres>, Option)> { + let job_option = sqlx::query_scalar!( + "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 \ + AND workspace_id = $4 RETURNING id", + username, + reason, + id, + w_id + ) + .fetch_optional(&mut tx) + .await?; + let mut jobs = job_option.map(|j| vec![j]).unwrap_or_default(); + while !jobs.is_empty() { + let p_job = jobs.pop(); + let new_jobs = sqlx::query_scalar!( + "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2 WHERE parent_job = $3 \ + AND workspace_id = $4 RETURNING id", + username, + reason, + p_job, + w_id + ) + .fetch_all(&mut tx) + .await?; + jobs.extend(new_jobs); + } + Ok((tx, job_option)) +} + async fn delete_completed_job( authed: Authed, Extension(user_db): Extension, @@ -1066,6 +1075,10 @@ pub async fn get_job_by_id<'c>( } } +#[derive(Deserialize)] +pub struct QueryApprover { + pub approver: Option, +} pub async fn get_queued_job<'c>( id: Uuid, w_id: &str, @@ -1085,85 +1098,19 @@ pub async fn get_queued_job<'c>( pub async fn resume_suspended_job( /* unauthed */ Extension(db): Extension, - Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>, + Path((w_id, job_id, resume_id, secret)): Path<(String, Uuid, u32, String)>, QueryOrBody(value): QueryOrBody, + Query(approver): Query, ) -> error::Result { let value = value.unwrap_or(serde_json::Value::Null); - insert_resume_job(&db, &w_id, job, resume_id, secret, false, value).await?; - Ok(StatusCode::CREATED) -} - -pub async fn cancel_suspended_job( - /* unauthed */ - Extension(db): Extension, - Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>, - QueryOrBody(value): QueryOrBody, -) -> error::Result { - let value = value.unwrap_or(serde_json::Value::Null); - insert_resume_job(&db, &w_id, job, resume_id, secret, true, value).await?; - Ok(StatusCode::CREATED) -} - -pub async fn get_suspended_job_flow( - /* unauthed */ - Extension(db): Extension, - Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>, -) -> error::JsonResult { - let mut tx = db.begin().await?; - let key = get_workspace_key(&w_id, &mut tx).await?; - let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?; - mac.update(job.as_bytes()); - mac.update(resume_id.to_be_bytes().as_ref()); - mac.verify_slice(hex::decode(secret)?.as_ref()) - .map_err(|_| anyhow::anyhow!("Invalid signature"))?; - let flow_id = sqlx::query_scalar!( - r#" - SELECT parent_job - FROM queue - WHERE id = $1 AND workspace_id = $2 - UNION ALL - SELECT parent_job - FROM completed_job - WHERE id = $1 AND workspace_id = $2 - "#, - job, - w_id - ) - .fetch_optional(&mut tx) - .await? - .flatten() - .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; - let flow_o = get_job_by_id(tx, &w_id, flow_id).await?.0; - let flow = crate::utils::not_found_if_none(flow_o, "Parent Flow", job.to_string())?; - Ok(Json(flow)) -} - -pub async fn create_job_signature( - authed: Authed, - Extension(user_db): Extension, - Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>, -) -> error::Result { - let key = get_workspace_key(&w_id, &mut user_db.begin(&authed).await?).await?; - let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?; - mac.update(job_id.as_bytes()); - mac.update(resume_id.to_be_bytes().as_ref()); - Ok(hex::encode(mac.finalize().into_bytes())) -} - -async fn insert_resume_job( - db: &DB, - w_id: &str, - job_id: Uuid, - resume_id: u32, - secret: String, - is_cancel: bool, - value: serde_json::Value, -) -> error::Result<()> { let mut tx = db.begin().await?; let key = get_workspace_key(&w_id, &mut tx).await?; let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?; mac.update(job_id.as_bytes()); mac.update(resume_id.to_be_bytes().as_ref()); + if let Some(approver) = approver.approver.clone() { + mac.update(approver.as_bytes()); + } mac.verify_slice(hex::decode(secret)?.as_ref()) .map_err(|_| anyhow::anyhow!("Invalid signature"))?; let flow = sqlx::query!( @@ -1182,14 +1129,15 @@ async fn insert_resume_job( sqlx::query!( r#" INSERT INTO resume_job - (id, job, flow, value, is_cancel) + (id, job, flow, value, approver) VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (id) DO NOTHING "#, Uuid::from_u128(job_id.as_u128() ^ resume_id as u128), job_id, flow.id, value, - is_cancel, + approver.approver ) .execute(&mut tx) .await?; @@ -1211,7 +1159,7 @@ async fn insert_resume_job( { sqlx::query!( "UPDATE queue SET suspend = $1 WHERE id = $2", - if is_cancel { 0 } else { suspend }, + suspend, flow.id, ) .execute(&mut tx) @@ -1220,7 +1168,159 @@ async fn insert_resume_job( } tx.commit().await?; - Ok(()) + Ok(StatusCode::CREATED) +} + +pub async fn cancel_suspended_job( + /* unauthed */ + Extension(db): Extension, + Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>, + Query(approver): Query, +) -> error::Result { + let mut tx = db.begin().await?; + let key = get_workspace_key(&w_id, &mut tx).await?; + let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?; + mac.update(job.as_bytes()); + mac.update(resume_id.to_be_bytes().as_ref()); + if let Some(approver) = approver.approver.clone() { + mac.update(approver.as_bytes()); + } + mac.verify_slice(hex::decode(secret)?.as_ref()) + .map_err(|_| anyhow::anyhow!("Invalid signature"))?; + + let whom = approver.approver.unwrap_or_else(|| "unknown".to_string()); + let parent_flow = get_root_job(db, &w_id, job).await?; + let (mut tx, job) = cancel_job( + &whom, + Some("approval request disapproved".to_string()), + parent_flow, + &w_id, + tx, + ) + .await?; + if job.is_some() { + audit_log( + &mut tx, + &whom, + "jobs.disapproval", + ActionKind::Delete, + &w_id, + Some(&parent_flow.to_string()), + None, + ) + .await?; + } + tx.commit().await?; + Ok("Flow of job cancelled".to_string()) +} + +pub async fn get_root_job(db: DB, w_id: &str, job: Uuid) -> error::Result { + let mut tx = db.begin().await?; + let mut job_id = job; + loop { + let (job, ntx) = get_job_by_id(tx, w_id, job_id).await?; + tx = ntx; + let p_job = job.and_then(|x| match x { + Job::QueuedJob(job) => job.parent_job, + Job::CompletedJob(job) => job.parent_job, + }); + if let Some(p_job) = p_job { + job_id = p_job; + } else { + return Ok(job_id); + } + } +} + +#[derive(Serialize)] +pub struct SuspendedJobFlow { + pub job: Job, + pub approvers: Vec, +} + +pub async fn get_suspended_job_flow( + /* unauthed */ + Extension(db): Extension, + Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>, + Query(approver): Query, +) -> error::JsonResult { + let mut tx = db.begin().await?; + let key = get_workspace_key(&w_id, &mut tx).await?; + let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?; + mac.update(job.as_bytes()); + mac.update(resume_id.to_be_bytes().as_ref()); + if let Some(approver) = approver.approver { + mac.update(approver.as_bytes()); + } + mac.verify_slice(hex::decode(secret)?.as_ref()) + .map_err(|_| anyhow::anyhow!("Invalid signature"))?; + let flow_id = sqlx::query_scalar!( + r#" + SELECT parent_job + FROM queue + WHERE id = $1 AND workspace_id = $2 + UNION ALL + SELECT parent_job + FROM completed_job + WHERE id = $1 AND workspace_id = $2 + "#, + job, + w_id + ) + .fetch_optional(&mut tx) + .await? + .flatten() + .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; + let (flow_o, mut tx) = get_job_by_id(tx, &w_id, flow_id).await?; + let flow = crate::utils::not_found_if_none(flow_o, "Parent Flow", job.to_string())?; + + let flow_status = flow + .raw_flow() + .ok_or_else(|| anyhow::anyhow!("unable to deserialize the flow"))?; + let flow_module_status = flow_status + .modules + .iter() + .find(|p| p.id() == job.to_string()) + .ok_or_else(|| anyhow::anyhow!("unable to find the module"))?; + let approvers_from_status = match flow_module_status { + FlowStatusModule::Success { approvers, .. } => approvers.to_owned(), + _ => vec![], + }; + let approvers = if approvers_from_status.is_empty() { + sqlx::query_scalar!( + r#" + SELECT approver + FROM resume_job + WHERE job = $1 + "#, + job, + ) + .fetch_all(&mut tx) + .await? + .into_iter() + .map(|x| x.unwrap_or_else(|| "anonymous".to_string())) + .collect() + } else { + approvers_from_status + }; + + Ok(Json(SuspendedJobFlow { job: flow, approvers })) +} + +pub async fn create_job_signature( + authed: Authed, + Extension(user_db): Extension, + Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>, + Query(approver): Query, +) -> error::Result { + let key = get_workspace_key(&w_id, &mut user_db.begin(&authed).await?).await?; + let mut mac = HmacSha256::new_from_slice(key.as_bytes()).map_err(to_anyhow)?; + mac.update(job_id.as_bytes()); + mac.update(resume_id.to_be_bytes().as_ref()); + if let Some(approver) = approver.approver { + mac.update(approver.as_bytes()); + } + Ok(hex::encode(mac.finalize().into_bytes())) } #[derive(Serialize, Debug)] @@ -1230,6 +1330,15 @@ pub enum Job { CompletedJob(CompletedJob), } +impl Job { + pub fn raw_flow(&self) -> Option { + let value = match self { + Job::QueuedJob(job) => job.raw_flow.clone(), + Job::CompletedJob(job) => job.raw_flow.clone(), + }; + value.map(|v| serde_json::from_value(v).ok()).flatten() + } +} #[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Clone)] #[sqlx(type_name = "JOB_KIND", rename_all = "lowercase")] #[serde(rename_all(serialize = "lowercase"))] @@ -1646,6 +1755,14 @@ pub async fn get_hub_script( .map(|e| e.0) } +pub fn canceled_job_to_result(job: &QueuedJob) -> String { + let reason = job + .canceled_reason + .as_deref() + .unwrap_or_else(|| "no reason given"); + let canceler = job.canceled_by.as_deref().unwrap_or_else(|| "unknown"); + format!("Job canceled: {reason} by {canceler}") +} #[instrument(level = "trace", skip_all)] pub async fn add_completed_job_error( db: &DB, diff --git a/backend/src/worker.rs b/backend/src/worker.rs index 8c6e6e2d93..363f23935b 100644 --- a/backend/src/worker.rs +++ b/backend/src/worker.rs @@ -15,8 +15,8 @@ use crate::{ db::DB, error::{self, Error}, jobs::{ - add_completed_job, add_completed_job_error, get_hub_script, get_queued_job, pull, JobKind, - QueuedJob, + add_completed_job, add_completed_job_error, canceled_job_to_result, get_hub_script, + get_queued_job, pull, JobKind, QueuedJob, }, parser::Typ, parser_go::otyp_to_string, @@ -479,7 +479,7 @@ async fn handle_queued_job( base_internal_url: &str, ) -> crate::error::Result<()> { if job.canceled { - return Err(Error::ExecutionErr(format!("Job canceled")))?; + return Err(Error::ExecutionErr(canceled_job_to_result(&job)))?; } match job.job_kind { JobKind::FlowPreview | JobKind::Flow => { @@ -3353,7 +3353,10 @@ def main(): server.close().await.unwrap(); - assert_eq!(json!("from job"), result); + assert_eq!( + json!({"error": "Job canceled: approval request disapproved by unknown" }), + result + ); } #[sqlx::test(fixtures("base"))] @@ -3410,7 +3413,10 @@ def main(): let result = completed_job_result(flow, &db).await; - assert_eq!(json!("from test"), result); + assert_eq!( + json!({"error": "Job canceled: approval request disapproved by unknown" }), + result + ); } } diff --git a/backend/src/worker_flow.rs b/backend/src/worker_flow.rs index 43e25d9b49..7140e2a7cc 100644 --- a/backend/src/worker_flow.rs +++ b/backend/src/worker_flow.rs @@ -6,7 +6,7 @@ use crate::{ error::{self, Error}, flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend}, jobs::{ - add_completed_job, add_completed_job_error, get_queued_job, push, + add_completed_job, add_completed_job_error, canceled_job_to_result, get_queued_job, push, schedule_again_if_scheduled, script_path_to_payload, JobPayload, QueuedJob, RawCode, }, js_eval::{eval_timeout, EvalCreds, IdContext}, @@ -104,6 +104,9 @@ pub enum FlowStatusModule { flow_jobs: Option>, #[serde(skip_serializing_if = "Option::is_none")] branch_chosen: Option, + #[serde(default)] + #[serde(skip_serializing_if = "Vec::is_empty")] + approvers: Vec, }, Failure { id: String, @@ -258,6 +261,7 @@ pub async fn update_flow_status_after_job_completion( job: job.id, flow_jobs, branch_chosen, + approvers: vec![], }, ) } else { @@ -398,15 +402,26 @@ pub async fn update_flow_status_after_job_completion( "Flow job completed".to_string() }; tracing::debug!("{skip_if_stop_early:?}"); - add_completed_job( - db, - &flow_job, - success, - stop_early && skip_if_stop_early.unwrap_or(false), - result.clone(), - logs, - ) - .await?; + if flow_job.canceled { + add_completed_job_error( + db, + &flow_job, + logs, + &canceled_job_to_result(&flow_job), + metrics.clone(), + ) + .await?; + } else { + add_completed_job( + db, + &flow_job, + success, + stop_early && skip_if_stop_early.unwrap_or(false), + result.clone(), + logs, + ) + .await?; + } true } else { match handle_flow( @@ -818,21 +833,33 @@ async fn push_next_flow_job( .context("lock flow in queue")?; let resumes = sqlx::query!( - "SELECT value, is_cancel FROM resume_job WHERE job = $1 ORDER BY created_at ASC", + "SELECT value, approver FROM resume_job WHERE job = $1 ORDER BY created_at ASC", last ) .fetch_all(&mut tx) .await?; - let is_cancelled = resumes - .iter() - .find(|r| r.is_cancel) - .map(|r| r.value.clone()); - - resume_messages.extend(resumes.into_iter().map(|r| r.value)); + resume_messages.extend(resumes.iter().map(|r| r.value.clone())); let required_events = suspend.required_events.unwrap() as u16; - if is_cancelled.is_none() && resume_messages.len() >= required_events as usize { + if resume_messages.len() >= required_events as usize { + sqlx::query( + " + UPDATE queue + SET flow_status = + JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'approvers'], $2) + WHERE id = $3 + ", + ) + .bind(status.step - 1) + .bind(json!(resumes + .into_iter() + .map(|r| r.approver.unwrap_or_else(|| "unknown".to_string())) + .collect::>())) + .bind(flow_job.id) + .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 { @@ -847,12 +874,10 @@ async fn push_next_flow_job( tx.commit().await?; /* not enough messages to do this job, "park"/suspend until there are */ - } else if is_cancelled.is_none() - && matches!( - &status_module, - FlowStatusModule::WaitingForPriorSteps { .. } - ) - { + } else if matches!( + &status_module, + FlowStatusModule::WaitingForPriorSteps { .. } + ) { sqlx::query( " UPDATE queue @@ -878,16 +903,10 @@ async fn push_next_flow_job( let success = false; let skipped = false; - let logs = if is_cancelled.is_some() { - "Cancelled while waiting to be resumed" - } else { - "Timed out waiting to be resumed" - } - .to_string(); - let result = is_cancelled.unwrap_or(json!({ "error": logs })); + let logs = "Timed out waiting to be resumed".to_string(); + let result = json!({ "error": logs }); let _uuid = add_completed_job(db, &flow_job, success, skipped, result, logs).await?; - return Ok(()); } } @@ -1062,6 +1081,7 @@ async fn push_next_flow_job( job: flow_job.id, flow_jobs: Some(vec![]), branch_chosen: None, + approvers: vec![], }, json!([]), same_worker_tx, diff --git a/deno-client/mod.ts b/deno-client/mod.ts index f89d50d4bf..622ab016f1 100644 --- a/deno-client/mod.ts +++ b/deno-client/mod.ts @@ -149,7 +149,7 @@ export interface NonceAndHmac { * @param jobId * @returns HMAC and nonce needed to authorize approval script actions */ -export async function genNounceAndHmac(workspace: string, jobId: string): Promise { +export async function genNounceAndHmac(workspace: string, jobId: string, approver?: string): Promise { const nonce = Math.floor(Math.random() * 4294967295); const sig = await fetch(Deno.env.get("WM_BASE_URL") + `/api/w/${workspace}/jobs/job_signature/${jobId}/${nonce}?token=${Deno.env.get("WM_TOKEN")}`) @@ -169,25 +169,26 @@ export interface ResumeEndpoints { * Get URLs needed for approval script * @returns approval page UI URL, resume and cancel API URLs for approval script */ -export async function getResumeEndpoints(): Promise { +export async function getResumeEndpoints(approver?: string): Promise { const workspace = getWorkspace() const { nonce, signature } = await genNounceAndHmac( workspace, Deno.env.get("WM_JOB_ID") ?? "no_job_id", + approver ); const url_prefix = Deno.env.get("WM_BASE_URL") + `/api/w/${workspace}/jobs/`; - function getResumeUrl(op: string): string { + function getResumeUrl(op: string, approver?: string): string { return url_prefix + - `${op}/${Deno.env.get("WM_JOB_ID")}/${nonce}/${signature}`; + `${op}/${Deno.env.get("WM_JOB_ID")}/${nonce}/${signature}${approver ? `?approver=${approver}` : ''}`; } return { - approvalPage: Deno.env.get("WM_BASE_URL") + `/approve/${workspace}/${Deno.env.get("WM_JOB_ID")}/${nonce}/${signature}`, - resume: getResumeUrl("resume"), - cancel: getResumeUrl("cancel"), + approvalPage: Deno.env.get("WM_BASE_URL") + `/approve/${workspace}/${Deno.env.get("WM_JOB_ID")}/${nonce}/${signature}${approver ? `?approver=${approver}` : ''}`, + resume: getResumeUrl("resume", approver), + cancel: getResumeUrl("cancel", approver), }; } diff --git a/frontend/src/lib/components/DisplayResult.svelte b/frontend/src/lib/components/DisplayResult.svelte index d95969b2f2..30fbce6b43 100644 --- a/frontend/src/lib/components/DisplayResult.svelte +++ b/frontend/src/lib/components/DisplayResult.svelte @@ -55,7 +55,7 @@ return 'jpeg' } else if (keys.length == 1 && keys[0] == 'file') { return 'file' - } else if (keys.length == 1 && keys[0] == 'error') { + } else if (keys.length == 1 && keys[0] == 'error' && typeof result['error'] == 'string') { return 'error' } } catch (err) {} diff --git a/frontend/src/lib/components/flows/content/FlowModuleSuspend.svelte b/frontend/src/lib/components/flows/content/FlowModuleSuspend.svelte index 59ee232cac..dca09d7d5d 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleSuspend.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleSuspend.svelte @@ -44,11 +44,11 @@ } }} options={{ - right: 'Suspend flow execution until events received' + right: 'Suspend flow execution until approvals received' }} />
- Number of events to wait for + Number of approvals required for resuming flow {#if flowModule.suspend} diff --git a/frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte b/frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte index 1feae6005d..127b1d28b7 100644 --- a/frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte +++ b/frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte @@ -7,16 +7,19 @@ import FlowMetadata from '$lib/components/FlowMetadata.svelte' let job: Job | undefined = undefined + let currentApprovers: string[] = [] getJob() async function getJob() { - job = await JobService.getSuspendedJobFlow({ + const suspendedJobFlow = await JobService.getSuspendedJobFlow({ workspace: $page.params.workspace, id: $page.params.job, resumeId: new Number($page.params.resume).valueOf(), signature: $page.params.hmac }) + job = suspendedJobFlow.job + currentApprovers = suspendedJobFlow.approvers } async function resume() { diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 9161bd99cb..6fe8f14df5 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -362,5 +362,9 @@ components: required: - branch - len + approvers: + type: array + items: + type: string required: [type]