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

feat: add approver to approval step

Browse Source
This commit is contained in:
Ruben Fiszel
2022-10-24 00:40:26 +02:00
parent dfc587b812
commit 7a4bcd98f6
12 changed files with 405 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/migrations/20221023162721_remove_iscancel_resumejob.down.sql Normal file
View File

@@ -0,0 +1 @@
-- Add down migration script here

3
backend/migrations/20221023162721_remove_iscancel_resumejob.up.sql Normal file
View File

@@ -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);

36
backend/openapi.yaml
View File

@@ -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:

104
backend/sqlx-data.json
View File

@@ -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": [

327
backend/src/jobs.rs
View File

@@ -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<UserDB>,
Path((w_id, id)): Path<(String, Uuid)>,
Json(CancelJob { reason }): Json<CancelJob>,
) -> error::Result<String> {
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<String>,
id: Uuid,
w_id: &str,
mut tx: Transaction<'c, Postgres>,
) -> error::Result<(Transaction<'c, Postgres>, Option<Uuid>)> {
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<UserDB>,
@@ -1066,6 +1075,10 @@ pub async fn get_job_by_id<'c>(
}
}
#[derive(Deserialize)]
pub struct QueryApprover {
pub approver: Option<String>,
}
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<DB>,
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<serde_json::Value>,
Query(approver): Query<QueryApprover>,
) -> error::Result<StatusCode> {
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<DB>,
Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>,
QueryOrBody(value): QueryOrBody<serde_json::Value>,
) -> error::Result<StatusCode> {
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<DB>,
Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>,
) -> error::JsonResult<Job> {
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<UserDB>,
Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>,
) -> error::Result<String> {
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<DB>,
Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>,
Query(approver): Query<QueryApprover>,
) -> error::Result<String> {
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<Uuid> {
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<String>,
}
pub async fn get_suspended_job_flow(
/* unauthed */
Extension(db): Extension<DB>,
Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>,
Query(approver): Query<QueryApprover>,
) -> error::JsonResult<SuspendedJobFlow> {
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<UserDB>,
Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>,
Query(approver): Query<QueryApprover>,
) -> error::Result<String> {
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<FlowStatus> {
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<E: ToString + std::fmt::Debug>(
db: &DB,

16
backend/src/worker.rs
View File

@@ -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
);
}
}

84
backend/src/worker_flow.rs
View File

@@ -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<Vec<Uuid>>,
#[serde(skip_serializing_if = "Option::is_none")]
branch_chosen: Option<BranchChosen>,
#[serde(default)]
#[serde(skip_serializing_if = "Vec::is_empty")]
approvers: Vec<String>,
},
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::<Vec<_>>()))
.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,

15
deno-client/mod.ts
View File

@@ -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<NonceAndHmac> {
export async function genNounceAndHmac(workspace: string, jobId: string, approver?: string): Promise<NonceAndHmac> {
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<ResumeEndpoints> {
export async function getResumeEndpoints(approver?: string): Promise<ResumeEndpoints> {
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),
};
}

2
frontend/src/lib/components/DisplayResult.svelte
View File

@@ -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) {}

4
frontend/src/lib/components/flows/content/FlowModuleSuspend.svelte
View File

@@ -44,11 +44,11 @@
}
}}
options={{
right: 'Suspend flow execution until events received'
right: 'Suspend flow execution until approvals received'
}}
/>
<div class="mb-4">
<span class="text-xs font-bold">Number of events to wait for</span>
<span class="text-xs font-bold">Number of approvals required for resuming flow</span>
{#if flowModule.suspend}
<input bind:value={flowModule.suspend.required_events} type="number" min="1" placeholder="1" />

5
frontend/src/routes/approve/[workspace]/[job]/[resume]/[hmac]@none.svelte
View File

@@ -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() {

4
openflow.openapi.yaml
View File

@@ -362,5 +362,9 @@ components:
required:
- branch
- len
approvers:
type: array
items:
type: string
required: [type]