starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-08 08:04:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: secure suspended resume event + configurable timeout (#721)

Browse Source 
* fix suspend count for early message + delete resume_job

* all

* done everything

* done everything
This commit is contained in:
Ruben Fiszel
2022-10-11 22:55:14 +02:00
committed by GitHub
parent 9acee22b1f
commit ff7fb0f6f3
16 changed files with 562 additions and 683 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+252 -490
View File
File diff suppressed because it is too large Load Diff
backend/Cargo.toml
+2 -1
View File
@@ -45,7 +45,6 @@ url = "^2"
async-oauth2 = "^0"
reqwest = { version = "^0", features = ["json"] }
time = "^0"
slack-http-verifier = "^0"
serde_urlencoded = "^0"
tokio-tar = "^0"
tempfile = "^3"
@@ -60,6 +59,8 @@ swc_ecma_parser = "^0"
swc_ecma_ast = "^0"
base64 = "^0"
unicode-general-category = "^0"
hmac = "^0"
sha2 = "^0"
sqlx = { version = "^0", features = ["offline", "macros", "migrate", "uuid", "json", "chrono", "postgres", "runtime-tokio-rustls"]}
dotenv = "^0"
backend/openapi.yaml
+22 -2
View File
@@ -2551,7 +2551,7 @@ paths:
schema:
type: string
/w/{workspace}/jobs/resume/{id}:
/w/{workspace}/jobs/resume/{job_id}/{resume_id}/{signature}:
get:
summary: resume a job for a suspended flow
operationId: resumeSuspendedJob
@@ -2560,6 +2560,16 @@ paths:
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/JobId"
- name: resume_id
in: path
required: true
schema:
type: integer
- name: signature
in: path
required: true
schema:
type: string
- name: payload
in: query
schema:
@@ -2593,7 +2603,7 @@ paths:
schema:
type: string
/w/{workspace}/jobs/cancel/{id}:
/w/{workspace}/jobs/cancel/{job_id}/{resume_id}/{signature}:
get:
summary: cancel a job for a suspended flow
operationId: cancelSuspendedJob
@@ -2602,6 +2612,16 @@ paths:
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/JobId"
- name: resume_id
in: path
required: true
schema:
type: integer
- name: signature
in: path
required: true
schema:
type: string
- name: payload
in: query
schema:
backend/sqlx-data.json
+32 -32
View File
@@ -1178,38 +1178,6 @@
},
"query": "SELECT * from resource_type WHERE name = $1 AND (workspace_id = $2 OR workspace_id = 'starter')"
},
"5a8eeff8ac0f52408720f8204bae085045cc4313bb31ab57442b093182633101": {
"describe": {
"columns": [
{
"name": "id",
"ordinal": 0,
"type_info": "Uuid"
},
{
"name": "flow_status",
"ordinal": 1,
"type_info": "Jsonb"
},
{
"name": "suspend",
"ordinal": 2,
"type_info": "Int4"
}
],
"nullable": [
false,
true,
false
],
"parameters": {
"Left": [
"Uuid"
]
}
},
"query": "\n SELECT id, flow_status, suspend\n FROM queue\n WHERE id = ( SELECT parent_job FROM queue WHERE id = $1\n UNION\n SELECT parent_job from completed_job WHERE id = $1 )\n FOR UPDATE\n "
},
"5b7a1d16d8109a65479ab33d411c60d14ea91d870fdff8606d7aa4ad39f0ba00": {
"describe": {
"columns": [
@@ -1887,6 +1855,38 @@
},
"query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , logs\n , raw_code\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language )\n VALUES ($1, $2, $3, $4, $5, $6, EXTRACT(milliseconds FROM (now() - $6)), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12)"
},
"853788436dbe987853433e8dc83665f68bd127de31d4c807abafeead896f6ac4": {
"describe": {
"columns": [
{
"name": "id",
"ordinal": 0,
"type_info": "Uuid"
},
{
"name": "flow_status",
"ordinal": 1,
"type_info": "Jsonb"
},
{
"name": "suspend",
"ordinal": 2,
"type_info": "Int4"
}
],
"nullable": [
false,
true,
false
],
"parameters": {
"Left": [
"Uuid"
]
}
},
"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 "
},
"88a3f58a1a315200fdd2e4bb8638246ee21818f8aaaf56f6e9d7ddce1490d886": {
"describe": {
"columns": [
backend/src/email.rs
-31
View File
@@ -1,31 +0,0 @@
use lettre::{
transport::smtp::authentication::Credentials, AsyncSmtpTransport, AsyncTransport, Message,
Tokio1Executor,
};
use crate::error::{Error, Result};
pub struct EmailSender {
pub from: String,
pub server: String,
pub password: String,
}
impl EmailSender {
pub async fn send_email(&self, email: Message) -> Result<()> {
let creds = Credentials::new(self.from.to_string(), self.password.to_string());
// Open a remote connection to gmail
let mailer: AsyncSmtpTransport<Tokio1Executor> =
AsyncSmtpTransport::<Tokio1Executor>::relay(&self.server)
.unwrap()
.credentials(creds)
.build();
mailer
.send(email)
.await
.map_err(|x| Error::InternalErr(format!("Impossible to send email {x}")))?;
Ok(())
}
}
backend/src/fixtures/base.sql
+8
View File
@@ -5,6 +5,14 @@ INSERT INTO workspace
(id, name, owner, domain)
VALUES ('test-workspace', 'test-workspace', 'test-user', null);
INSERT INTO usr(workspace_id, email, username, is_admin, role) VALUES
('test-workspace', 'test@windmill.dev', 'test-user', true, 'Admin');
INSERT INTO workspace_key(workspace_id, kind, key) VALUES
('test-workspace', 'cloud', 'test-key');
GRANT ALL PRIVILEGES ON TABLE workspace_key TO windmill_admin;
GRANT ALL PRIVILEGES ON TABLE workspace_key TO windmill_user;
CREATE FUNCTION "notify_insert_on_completed_job" ()
RETURNS TRIGGER AS $$
backend/src/flows.rs
+10 -3
View File
@@ -151,6 +151,14 @@ impl Default for ExponentialDelay {
}
}
#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct Suspend {
#[serde(skip_serializing_if = "Option::is_none")]
pub required_events: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<u32>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
pub struct FlowModule {
#[serde(default)]
@@ -159,9 +167,8 @@ pub struct FlowModule {
pub value: FlowModuleValue,
pub stop_after_if: Option<StopAfterIf>,
pub summary: Option<String>,
#[serde(default)]
#[serde(skip_serializing_if = "is_default")]
pub suspend: u16,
#[serde(skip_serializing_if = "Option::is_none")]
pub suspend: Option<Suspend>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<Retry>,
#[serde(skip_serializing_if = "Option::is_none")]
backend/src/jobs.rs
+68 -42
View File
@@ -9,6 +9,7 @@
use axum::extract::Host;
use anyhow::Context;
use hmac::Mac;
use sql_builder::prelude::*;
use sqlx::{query_scalar, Postgres, Transaction};
use std::collections::HashMap;
@@ -17,12 +18,14 @@ use tracing::instrument;
use crate::{
audit::{audit_log, ActionKind},
db::{UserDB, DB},
error::{self, to_anyhow, Error, OrElseNotFound},
error::{self, to_anyhow, Error},
flows::FlowValue,
oauth2::HmacSha256,
schedule::get_schedule_opt,
scripts::{get_hub_script_by_path, ScriptHash, ScriptLang},
users::{owner_to_token_owner, Authed},
utils::{now_from_db, require_admin, Pagination, StripPath},
variables::get_workspace_key,
worker,
worker_flow::{
init_flow_status, FlowStatus, FlowStatusModule, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
@@ -69,14 +72,30 @@ pub fn workspaced_service() -> Router {
.route("/completed/delete/:id", post(delete_completed_job))
.route("/get/:id", get(get_job))
.route("/getupdate/:id", get(get_job_update))
.route(
"/job_signature/:job_id/:resume_id",
get(create_job_signature),
)
}
pub fn global_service() -> Router {
Router::new()
.route("/resume/:id", get(resume_suspended_job))
.route("/resume/:id", post(resume_suspended_job))
.route("/cancel/:id", get(cancel_suspended_job))
.route("/cancel/:id", post(cancel_suspended_job))
.route(
"/resume/:job_id/:resume_id/:secret",
get(resume_suspended_job),
)
.route(
"/resume/:job_id/:resume_id/:secret",
post(resume_suspended_job),
)
.route(
"/cancel/:job_id/:resume_id/:secret",
get(cancel_suspended_job),
)
.route(
"/cancel/:job_id/:resume_id/:secret",
post(cancel_suspended_job),
)
}
#[derive(Debug, sqlx::FromRow, Serialize, Clone)]
@@ -118,7 +137,7 @@ impl QueuedJob {
}
#[derive(Debug, sqlx::FromRow, Serialize)]
struct CompletedJob {
pub struct CompletedJob {
workspace_id: String,
id: Uuid,
parent_job: Option<Uuid>,
@@ -925,7 +944,7 @@ async fn get_job(
Ok(Json(job))
}
async fn get_job_by_id<'c>(
pub async fn get_job_by_id<'c>(
mut tx: Transaction<'c, Postgres>,
w_id: &str,
id: Uuid,
@@ -963,58 +982,65 @@ pub async fn get_queued_job<'c>(
pub async fn resume_suspended_job(
/* unauthed */
Extension(db): Extension<DB>,
Path((_, job)): Path<(String, Uuid)>,
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);
NewResumeJob { job, value, is_cancel: false }
.insert(&db)
.await?;
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((_, job)): Path<(String, Uuid)>,
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);
NewResumeJob { job, value, is_cancel: true }
.insert(&db)
.await?;
insert_resume_job(&db, &w_id, job, resume_id, secret, true, value).await?;
Ok(StatusCode::CREATED)
}
struct NewResumeJob {
pub job: Uuid,
pub value: Value,
pub is_cancel: bool,
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()))
}
impl NewResumeJob {
async fn insert(self, db: &DB) -> error::Result<()> {
insert_resume_job(db, self).await
}
}
async fn insert_resume_job(db: &DB, new: NewResumeJob) -> error::Result<()> {
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());
mac.verify_slice(hex::decode(secret)?.as_ref())
.map_err(|_| anyhow::anyhow!("Invalid signature"))?;
let flow = sqlx::query!(
r#"
SELECT id, flow_status, suspend
FROM queue
WHERE id = ( SELECT parent_job FROM queue WHERE id = $1
UNION
SELECT parent_job from completed_job WHERE id = $1 )
FOR UPDATE
FROM queue
WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1)
FOR UPDATE
"#,
new.job,
job_id,
)
.fetch_optional(&mut tx)
.await?
.or_else_not_found(new.job)?;
.ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?;
sqlx::query!(
r#"
@@ -1022,11 +1048,11 @@ async fn insert_resume_job(db: &DB, new: NewResumeJob) -> error::Result<()> {
(id, job, flow, value, is_cancel)
VALUES ($1, $2, $3, $4, $5)
"#,
Uuid::from(Ulid::new()),
new.job,
Uuid::from_u128(job_id.as_u128() ^ resume_id as u128),
job_id,
flow.id,
new.value,
new.is_cancel,
value,
is_cancel,
)
.execute(&mut tx)
.await?;
@@ -1044,11 +1070,11 @@ async fn insert_resume_job(db: &DB, new: NewResumeJob) -> error::Result<()> {
let status =
serde_json::from_value::<FlowStatus>(flow.flow_status.context("no flow status")?)
.context("deserialize flow status")?;
if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &new.job)
if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &job_id)
{
sqlx::query!(
"UPDATE queue SET suspend = $1 WHERE id = $2",
if new.is_cancel { 0 } else { suspend },
if is_cancel { 0 } else { suspend },
flow.id,
)
.execute(&mut tx)
@@ -1060,9 +1086,9 @@ async fn insert_resume_job(db: &DB, new: NewResumeJob) -> error::Result<()> {
Ok(())
}
#[derive(Serialize)]
#[derive(Serialize, Debug)]
#[serde(tag = "type")]
enum Job {
pub enum Job {
QueuedJob(QueuedJob),
CompletedJob(CompletedJob),
}
backend/src/lib.rs
+4 -11
View File
@@ -12,7 +12,6 @@ use axum::{handler::Handler, middleware::from_extractor, routing::get, Extension
use db::DB;
use futures::FutureExt;
use git_version::git_version;
use slack_http_verifier::SlackVerifier;
use std::{net::SocketAddr, sync::Arc};
use tower::ServiceBuilder;
use tower_cookies::CookieManagerLayer;
@@ -25,7 +24,6 @@ extern crate dotenv;
mod audit;
mod client;
mod db;
mod email;
mod error;
mod external_ip;
mod flows;
@@ -57,11 +55,10 @@ mod workspaces;
use error::Error;
pub use crate::email::EmailSender;
use crate::{
db::UserDB,
error::to_anyhow,
oauth2::build_oauth_clients,
oauth2::{build_oauth_clients, SlackVerifier},
tracing_init::{MyMakeSpan, MyOnResponse},
utils::rd_string,
};
@@ -99,16 +96,14 @@ struct CloudHosted(bool);
pub async fn run_server(
db: DB,
addr: SocketAddr,
base_url: &str,
es: EmailSender,
base_url: String,
mut rx: tokio::sync::broadcast::Receiver<()>,
) -> anyhow::Result<()> {
let user_db = UserDB::new(db.clone());
let auth_cache = Arc::new(users::AuthCache::new(db.clone()));
let argon2 = Arc::new(Argon2::default());
let email_sender = Arc::new(es);
let basic_clients = Arc::new(build_oauth_clients(base_url).await?);
let basic_clients = Arc::new(build_oauth_clients(&base_url).await?);
let slack_verifier = Arc::new(
std::env::var("SLACK_SIGNING_SECRET")
.ok()
@@ -150,9 +145,7 @@ pub async fn run_server(
.nest("/jobs", jobs::workspaced_service())
.nest(
"/users",
users::workspaced_service()
.layer(Extension(argon2.clone()))
.layer(Extension(email_sender)),
users::workspaced_service().layer(Extension(argon2.clone())),
)
.nest("/variables", variables::workspaced_service())
.nest("/oauth", oauth2::workspaced_service())
backend/src/main.rs
+1 -12
View File
@@ -59,18 +59,7 @@ async fn main() -> anyhow::Result<()> {
let base_url_2 = base_url.clone();
let server_f = async {
if server_mode {
windmill::run_server(
db.clone(),
addr,
&base_url_2,
windmill::EmailSender {
from: "bot@windmill.dev".to_string(),
server: "smtp.gmail.com".to_string(),
password: std::env::var("SMTP_PASSWORD").unwrap_or("NOPASS".to_string()),
},
rx.resubscribe(),
)
.await?;
windmill::run_server(db.clone(), addr, base_url_2, rx.resubscribe()).await?;
}
Ok(()) as anyhow::Result<()>
};
backend/src/oauth2.rs
+32 -1
View File
@@ -16,7 +16,6 @@ use itertools::Itertools;
use oauth2::{Client as OClient, *};
use reqwest::Client;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use slack_http_verifier::SlackVerifier;
use sqlx::{Postgres, Transaction};
use tokio::{fs::File, io::AsyncReadExt};
use tower_cookies::{Cookie, Cookies};
@@ -36,6 +35,13 @@ use crate::{
BaseUrl,
};
use std::str;
use hmac::{Hmac, Mac};
use sha2::Sha256;
pub type HmacSha256 = Hmac<Sha256>;
pub fn global_service() -> Router {
Router::new()
.route("/login/:client", get(login))
@@ -959,3 +965,28 @@ fn set_cookie(state: &State, cookies: Cookies) {
cookie.set_path("/");
cookies.add(cookie);
}
#[derive(Clone, Debug)]
pub struct SlackVerifier {
mac: HmacSha256,
}
impl SlackVerifier {
pub fn new<S: AsRef<[u8]>>(secret: S) -> anyhow::Result<SlackVerifier> {
HmacSha256::new_from_slice(secret.as_ref())
.map(|mac| SlackVerifier { mac })
.map_err(|_| anyhow::anyhow!("invalid secret"))
}
pub fn verify(&self, ts: &str, body: &str, exp_sig: &str) -> anyhow::Result<()> {
let basestring = format!("v0:{}:{}", ts, body);
let mut mac = self.mac.clone();
mac.update(basestring.as_bytes());
let sig = format!("v0={}", hex::encode(mac.finalize().into_bytes()));
if sig != exp_sig {
Err(anyhow::anyhow!("signature mismatch"))?;
}
Ok(())
}
}
backend/src/variables.rs
+10 -3
View File
@@ -453,15 +453,22 @@ pub async fn build_crypt<'c>(
db: &mut Transaction<'c, Postgres>,
w_id: &str,
) -> Result<MagicCrypt256> {
let key = get_workspace_key(w_id, db).await?;
Ok(magic_crypt::new_magic_crypt!(key, 256))
}
pub async fn get_workspace_key<'c>(
w_id: &str,
db: &mut Transaction<'c, Postgres>,
) -> Result<String> {
let key = sqlx::query_scalar!(
"SELECT key FROM workspace_key WHERE workspace_id = $1 AND kind = 'cloud'",
w_id
)
.fetch_one(db)
.await
.map_err(|e| Error::InternalErr(format!("fetching crypt key: {e}")))?;
Ok(magic_crypt::new_magic_crypt!(key, 256))
.map_err(|e| Error::InternalErr(format!("fetching workspace key: {e}")))?;
Ok(key)
}
pub fn encrypt(mc: &MagicCrypt256, value: &str) -> String {
backend/src/worker.rs
+86 -38
View File
@@ -2623,46 +2623,42 @@ def main():
}
mod suspend_resume {
use crate::jobs::get_job_by_id;
use super::*;
use futures::{Stream, StreamExt};
struct ApiServer {
addr: std::net::SocketAddr,
tx: tokio::sync::oneshot::Sender<()>,
task: tokio::task::JoinHandle<hyper::Result<()>>,
tx: tokio::sync::broadcast::Sender<()>,
task: tokio::task::JoinHandle<anyhow::Result<()>>,
}
impl ApiServer {
async fn start(db: DB) -> Self {
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
let (tx, rx) = tokio::sync::broadcast::channel::<()>(1);
let sock = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = sock.local_addr().unwrap();
drop(sock);
use crate::jobs::{cancel_suspended_job, resume_suspended_job};
use axum::routing::{get, post};
let router = axum::Router::new()
.route("/w/:workspace/jobs/resume/:id", get(resume_suspended_job))
.route("/w/:workspace/jobs/resume/:id", post(resume_suspended_job))
.route("/w/:workspace/jobs/cancel/:id", get(cancel_suspended_job))
.route("/w/:workspace/jobs/cancel/:id", post(cancel_suspended_job))
.layer(axum::extract::Extension(db));
let serve = axum::Server::from_tcp(sock.into_std().unwrap())
.unwrap()
.serve(router.into_make_service())
.with_graceful_shutdown(async { drop(rx.await) });
let task = tokio::task::spawn(serve);
let task = tokio::task::spawn({
crate::run_server(
db.clone(),
addr,
format!("http://localhost:{}", addr.port()),
rx,
)
});
return Self { addr, tx, task };
}
async fn close(self) -> hyper::Result<()> {
let Self { task, tx, .. } = self;
async fn close(self) -> anyhow::Result<()> {
let Self { tx, task, .. } = self;
drop(tx);
task.await.unwrap()
}
@@ -2686,6 +2682,16 @@ def main():
}
}
async fn print_job(id: Uuid, db: &DB) -> Result<(), anyhow::Error> {
tracing::info!(
"{:#?}",
get_job_by_id(db.begin().await?, "test-workspace", id)
.await?
.0
);
Ok(())
}
fn flow() -> FlowValue {
serde_json::from_value(serde_json::json!({
"modules": [{
@@ -2700,19 +2706,32 @@ def main():
"content": "\
export async function main(n, port, op) {\
const job = Deno.env.get('WM_JOB_ID');
const token = Deno.env.get('WM_TOKEN');
const r = await fetch(
`http://localhost:${port}/w/test-workspace/jobs/${op}/${job}`,\
`http://localhost:${port}/api/w/test-workspace/jobs/job_signature/${job}/0?token=${token}`,\
{\
method: 'GET',\
headers: { 'Authorization': `Bearer ${token}` }\
}\
);\
console.log(r);\
const secret = await r.text();\
console.log('Secret: ' + secret + ' ' + job + ' ' + token);\
const r2 = await fetch(
`http://localhost:${port}/api/w/test-workspace/jobs/${op}/${job}/0/${secret}`,\
{\
method: 'POST',\
body: JSON.stringify('from job'),\
headers: { 'content-type': 'application/json' }\
}\
);\
console.log(r);
console.log(await r2.text());\
return n + 1;\
}",
},
"suspend": 1,
"suspend": {
"required_events": 1
},
}, {
"input_transform": {
"n": { "type": "javascript", "expr": "previous_result", },
@@ -2724,7 +2743,9 @@ def main():
"language": "deno",
"content": "export function main(n, resume, resumes) { return { n: n + 1, resume, resumes } }"
},
"suspend": 1,
"suspend": {
"required_events": 1
},
}, {
"input_transform": {
"last": { "type": "javascript", "expr": "previous_result", },
@@ -2762,19 +2783,34 @@ def main():
let db = db_;
wait_until_flow_suspends(flow, queue, &db).await;
// print_job(flow, &db).await;
/* The first job resumes itself. */
let _first = completed.next().await.unwrap();
// print_job(_first, &db).await;
/* ... and send a request resume it. */
let second = completed.next().await.unwrap();
// print_job(second, &db).await;
let token = create_token_for_owner(&db, "test-workspace", "u/test-user", "", 100, "").await.unwrap();
let secret = reqwest::get(format!(
"http://localhost:{port}/api/w/test-workspace/jobs/job_signature/{second}/0?token={token}"
))
.await
.unwrap()
.error_for_status()
.unwrap()
.text().await.unwrap();
println!("{}", secret);
/* ImZyb20gdGVzdCIK = base64 "from test" */
reqwest::get(format!(
"http://localhost:{port}/w/test-workspace/jobs/resume/{second}?payload=ImZyb20gdGVzdCIK"
))
.await
.unwrap()
.error_for_status()
.unwrap();
"http://localhost:{port}/api/w/test-workspace/jobs/resume/{second}/0/{secret}?payload=ImZyb20gdGVzdCIK"
))
.await
.unwrap()
.error_for_status()
.unwrap();
completed.find(&flow).await.unwrap();
})
@@ -2852,13 +2888,25 @@ def main():
/* ... and send a request resume it. */
let second = completed.next().await.unwrap();
let token = create_token_for_owner(&db, "test-workspace", "u/test-user", "", 100, "").await.unwrap();
let secret = reqwest::get(format!(
"http://localhost:{port}/api/w/test-workspace/jobs/job_signature/{second}/0?token={token}"
))
.await
.unwrap()
.error_for_status()
.unwrap()
.text().await.unwrap();
println!("{}", secret);
/* ImZyb20gdGVzdCIK = base64 "from test" */
reqwest::get(format!(
"http://localhost:{port}/w/test-workspace/jobs/cancel/{second}?payload=ImZyb20gdGVzdCIK"
))
.await
.unwrap()
.error_for_status()
.unwrap();
"http://localhost:{port}/api/w/test-workspace/jobs/cancel/{second}/0/{secret}?payload=ImZyb20gdGVzdCIK"
))
.await
.unwrap()
.error_for_status()
.unwrap();
completed.find(&flow).await.unwrap();
})
@@ -3213,7 +3261,7 @@ def main(error, port):
payload,
Some(args),
/* user */ "test-user",
/* permissioned_as */ Default::default(),
/* permissioned_as */ "u/admin".to_string(),
/* scheduled_for_o */ None,
/* schedule_path */ None,
/* parent_job */ None,
backend/src/worker_flow.rs
+16 -14
View File
@@ -4,7 +4,7 @@ use std::time::Duration;
use crate::{
db::DB,
error::{self, Error},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
jobs::{
add_completed_job, add_completed_job_error, get_queued_job, postprocess_queued_job, push,
script_path_to_payload, JobPayload, QueuedJob,
@@ -623,7 +623,7 @@ async fn push_next_flow_job(
&status_module,
FlowStatusModule::WaitingForPriorSteps | FlowStatusModule::WaitingForEvents { .. }
) {
if let Some((count, last)) = needs_resume(&flow, &status) {
if let Some((suspend, last)) = needs_resume(&flow, &status) {
let mut tx = db.begin().await?;
/* Lock this row to prevent the suspend column getting out out of sync
@@ -652,7 +652,8 @@ async fn push_next_flow_job(
resume_messages.extend(resumes.into_iter().map(|r| r.value));
if is_cancelled.is_none() && resume_messages.len() >= count as usize {
let required_events = suspend.required_events.unwrap() as u16;
if is_cancelled.is_none() && resume_messages.len() >= required_events as usize {
/* 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 {
@@ -663,10 +664,6 @@ async fn push_next_flow_job(
.context("previous job result")?;
}
sqlx::query!("DELETE FROM resume_job WHERE job = $1", last)
.execute(&mut tx)
.await?;
/* continue on and run this job! */
tx.commit().await?;
@@ -683,9 +680,9 @@ async fn push_next_flow_job(
WHERE id = $4
",
)
.bind(json!(FlowStatusModule::WaitingForEvents { count, job: last }))
.bind((count - resume_messages.len() as u16) as i32)
.bind(30 * MINUTES)
.bind(json!(FlowStatusModule::WaitingForEvents { count: required_events, job: last }))
.bind((required_events - resume_messages.len() as u16) as i32)
.bind(suspend.timeout.map(|t| Duration::from_secs(t.into())).unwrap_or_else(|| 30 * MINUTES))
.bind(flow_job.id)
.execute(&mut tx)
.await?;
@@ -1333,18 +1330,23 @@ fn from_now(duration: Duration) -> chrono::DateTime<chrono::Utc> {
}
/// returns previous module non-zero suspend count and job
fn needs_resume(flow: &FlowValue, status: &FlowStatus) -> Option<(u16, Uuid)> {
fn needs_resume(flow: &FlowValue, status: &FlowStatus) -> Option<(Suspend, Uuid)> {
let prev = usize::try_from(status.step)
.ok()
.and_then(|s| s.checked_sub(1))?;
let suspend = flow.modules.get(prev)?.suspend;
if suspend == 0 {
let suspend = flow.modules.get(prev)?.suspend.clone();
if suspend
.as_ref()
.and_then(|s| s.required_events)
.unwrap_or(0)
== 0
{
return None;
}
if let &FlowStatusModule::Success { job, .. } = status.modules.get(prev)? {
Some((suspend, job))
Some((suspend.unwrap(), job))
} else {
None
}
frontend/src/lib/components/flows/content/FlowModuleSuspend.svelte
+13 -2
View File
@@ -37,7 +37,10 @@
if (isSuspendEnabled && flowModule.suspend != undefined) {
flowModule.suspend = undefined
} else {
flowModule.suspend = 1
flowModule.suspend = {
required_events: 1,
timeout: 1800
}
}
}}
options={{
@@ -48,7 +51,15 @@
<span class="text-xs font-bold">Number of events to wait for</span>
{#if flowModule.suspend}
<input bind:value={flowModule.suspend} type="number" min="1" placeholder="1" />
<input bind:value={flowModule.suspend.required_events} type="number" min="1" placeholder="1" />
{:else}
<input type="number" disabled />
{/if}
<span class="text-xs font-bold">Timeout (in seconds)</span>
{#if flowModule.suspend}
<input bind:value={flowModule.suspend.timeout} type="number" min="1" placeholder="1800" />
{:else}
<input type="number" disabled />
{/if}
openflow.openapi.yaml
+6 -1
View File
@@ -91,7 +91,12 @@ components:
summary:
type: string
suspend:
type: integer
type: object
properties:
required_events:
type: integer
timeout:
type: integer
retry:
$ref: "#/components/schemas/Retry"
required: