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

feat: zero copy result for job result (#2263)

Browse Source 
* feat: zero copy result for job result

* update
This commit is contained in:
Ruben Fiszel
2023-09-11 08:00:02 +02:00
committed by GitHub
parent 164cdaf094
commit 22a7da58b1
6 changed files with 333 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-9f16a61d6a9a42f3fd3e30a1e7776503cee1b45eba150c3082eb246ea3f98d47.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXISTS(SELECT 1 FROM completed_job WHERE id = $1 AND workspace_id = $2)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "exists",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
null
]
},
"hash": "9f16a61d6a9a42f3fd3e30a1e7776503cee1b45eba150c3082eb246ea3f98d47"
}
backend/Cargo.toml
+1 -1
View File
@@ -96,7 +96,7 @@ tower = "^0"
tower-http = { version = "^0", features = ["trace", "cors"] }
tower-cookies = "^0"
serde = "^1"
serde_json = { version = "^1", features = ["preserve_order"] }
serde_json = { version = "^1", features = ["preserve_order", "raw_value"] }
uuid = { version = "^1", features = ["serde", "v4"] }
thiserror = "^1"
anyhow = "^1"
backend/tests/worker.rs
+103 -97
View File
@@ -5,22 +5,63 @@ use futures::StreamExt;
use futures::{stream, Stream};
use serde::Deserialize;
use serde_json::json;
use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres, Transaction};
use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres, query};
use tokio::{
sync::RwLock,
time::{timeout, Duration},
};
use windmill_api::jobs::{CompletedJob, Job};
use windmill_api_client::types::{
CreateFlowBody, EditSchedule, NewSchedule, RawScript, ScriptArgs,
};
use windmill_common::{
flow_status::{FlowStatus, FlowStatusModule},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform},
jobs::{JobPayload, RawCode},
scripts::ScriptLang,
jobs::{JobPayload, RawCode, JobKind},
scripts::{ScriptLang, ScriptHash}
};
use windmill_queue::{get_queued_job, PushIsolationLevel};
use windmill_queue::PushIsolationLevel;
use serde::Serialize;
#[derive(Debug, sqlx::FromRow, Serialize)]
pub struct CompletedJob {
pub workspace_id: String,
pub id: Uuid,
pub parent_job: Option<Uuid>,
pub created_by: String,
pub created_at: chrono::DateTime<chrono::Utc>,
pub started_at: chrono::DateTime<chrono::Utc>,
pub duration_ms: i32,
pub success: bool,
pub script_path: Option<String>,
pub args: Option<serde_json::Value>,
pub result: Option<serde_json::Value>,
pub logs: Option<String>,
pub deleted: bool,
pub raw_code: Option<String>,
pub canceled: bool,
pub canceled_by: Option<String>,
pub canceled_reason: Option<String>,
pub schedule_path: Option<String>,
pub permissioned_as: String,
pub flow_status: Option<serde_json::Value>,
pub raw_flow: Option<serde_json::Value>,
pub is_flow_step: bool,
pub is_skipped: bool,
pub email: String,
pub visible_to_owner: bool,
pub mem_peak: Option<i32>,
pub tag: String,
pub script_hash: Option<ScriptHash>,
pub language: Option<ScriptLang>,
pub job_kind: JobKind,
}
impl CompletedJob {
pub fn json_result(&self) -> Option<serde_json::Value> {
self.result.clone()
}
}
async fn initialize_tracing() {
use std::sync::Once;
@@ -54,37 +95,6 @@ fn next_worker_name() -> String {
format!("{id}/{thread_name}")
}
pub async fn get_job_by_id<'c>(
mut tx: Transaction<'c, Postgres>,
w_id: &str,
id: Uuid,
) -> windmill_common::error::Result<(Option<Job>, Transaction<'c, Postgres>)> {
let cjob_option = sqlx::query_as::<_, CompletedJob>(
"SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2",
)
.bind(id)
.bind(w_id)
.fetch_optional(&mut *tx)
.await?;
let job_option = match cjob_option {
Some(job) => Some(Job::CompletedJob(job)),
None => get_queued_job(id, w_id, &mut tx).await?.map(Job::QueuedJob),
};
if job_option.is_some() {
Ok((job_option, tx))
} else {
// check if a job had been moved in-between queries
let cjob_option = sqlx::query_as::<_, CompletedJob>(
"SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2",
)
.bind(id)
.bind(w_id)
.fetch_optional(&mut *tx)
.await?;
Ok((cjob_option.map(Job::CompletedJob), tx))
}
}
pub struct ApiServer {
pub addr: std::net::SocketAddr,
tx: tokio::sync::broadcast::Sender<()>,
@@ -121,15 +131,15 @@ impl ApiServer {
}
}
async fn _print_job(id: Uuid, db: &Pool<Postgres>) -> Result<(), anyhow::Error> {
tracing::info!(
"{:#?}",
get_job_by_id(db.begin().await?, "test-workspace", id)
.await?
.0
);
Ok(())
}
// async fn _print_job(id: Uuid, db: &Pool<Postgres>) -> Result<(), anyhow::Error> {
// tracing::info!(
// "{:#?}",
// get_job_by_id(db.begin().await?, "test-workspace", id)
// .await?
// .0
// );
// Ok(())
// }
fn get_module(cjob: &CompletedJob, id: &str) -> Option<FlowStatusModule> {
cjob.flow_status.clone().and_then(|fs| {
@@ -299,7 +309,7 @@ mod suspend_resume {
server.close().await.unwrap();
let result = completed_job(flow, &db).await.result.unwrap();
let result = completed_job(flow, &db).await.json_result().unwrap();
assert_eq!(
json!({
@@ -337,7 +347,7 @@ mod suspend_resume {
.arg("port", json!(port))
.run_until_complete(&db, port)
.await
.result
.json_result()
.unwrap();
server.close().await.unwrap();
@@ -400,7 +410,7 @@ mod suspend_resume {
server.close().await.unwrap();
let result = completed_job(flow, &db).await.result.unwrap();
let result = completed_job(flow, &db).await.json_result().unwrap();
assert_eq!(
json!( {"error": {"name": "Canceled", "reason": "approval request disapproved", "message": "Job canceled: approval request disapproved by unknown", "canceler": "unknown"}}),
@@ -555,7 +565,7 @@ def main(last, port):
.arg("port", json!(server.addr.port()))
.run_until_complete(&db, server.addr.port())
.await
.result
.json_result()
.unwrap();
assert_eq!(server.close().await, attempts);
@@ -584,7 +594,7 @@ def main(last, port):
.arg("port", json!(server.addr.port()))
.run_until_complete(&db, server.addr.port())
.await
.result
.json_result()
.unwrap();
assert_eq!(server.close().await, attempts);
@@ -626,7 +636,7 @@ def main(last, port):
.run_until_complete(&db, server.addr.port())
.await;
let result = job.result.unwrap();
let result = job.json_result().unwrap();
assert_eq!(server.close().await, attempts);
assert!(result["error"]
.as_object()
@@ -690,7 +700,7 @@ def main(error, port):
.arg("port", json!(server.addr.port()))
.run_until_complete(&db, server.addr.port())
.await;
let result = cjob.result.clone().unwrap();
let result = cjob.json_result().clone().unwrap();
let failed_module = get_module(&cjob, "a").unwrap();
match failed_module {
FlowStatusModule::Failure { .. } => {}
@@ -748,7 +758,7 @@ async fn test_iteration(db: Pool<Postgres>) {
.arg("items", json!([]))
.run_until_complete(&db, server.addr.port())
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([]));
@@ -757,7 +767,7 @@ async fn test_iteration(db: Pool<Postgres>) {
.arg("items", json!((0..257).collect::<Vec<_>>()))
.run_until_complete(&db, server.addr.port())
.await
.result
.json_result()
.unwrap();
assert!(matches!(result, serde_json::Value::Array(_)));
assert!(result[2]["error"]
@@ -805,7 +815,7 @@ async fn test_iteration_parallel(db: Pool<Postgres>) {
.arg("items", json!([]))
.run_until_complete(&db, server.addr.port())
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([]));
@@ -815,7 +825,7 @@ async fn test_iteration_parallel(db: Pool<Postgres>) {
.run_until_complete(&db, server.addr.port())
.await;
// println!("{:#?}", job);
let result = job.result.unwrap();
let result = job.json_result().unwrap();
assert!(matches!(result, serde_json::Value::Array(_)));
assert!(result[2]["error"]
.as_object()
@@ -995,8 +1005,8 @@ async fn listen_for_uuid_on(
}
async fn completed_job(uuid: Uuid, db: &Pool<Postgres>) -> CompletedJob {
sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
.bind(uuid)
sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1").bind(uuid)
.fetch_one(db)
.await
.unwrap()
@@ -1106,7 +1116,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
println!("deno flow iteration: {}", i);
let job = run_job_in_new_worker_until_complete(&db, job.clone(), port).await;
// println!("job: {:#?}", job.flow_status);
let result = job.result.unwrap();
let result = job.json_result().unwrap();
assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {}", i);
}
}
@@ -1142,7 +1152,7 @@ async fn test_identity(db: Pool<Postgres>) {
let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None })
.run_until_complete(&db, server.addr.port())
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(42));
}
@@ -1331,7 +1341,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
let result = run_job_in_new_worker_until_complete(&db, job.clone(), server.addr.port())
.await
.result
.json_result()
.unwrap();
assert_eq!(
result,
@@ -1386,7 +1396,7 @@ async fn test_flow_result_by_id(db: Pool<Postgres>) {
let job = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, job.clone(), port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([[42]]));
}
@@ -1431,7 +1441,7 @@ async fn test_stop_after_if(db: Pool<Postgres>) {
.arg("n", json!(123))
.run_until_complete(&db, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(json!("last step saw 123"), result);
@@ -1440,7 +1450,7 @@ async fn test_stop_after_if(db: Pool<Postgres>) {
.run_until_complete(&db, port)
.await;
let result = cjob.result.unwrap();
let result = cjob.json_result().unwrap();
assert_eq!(json!(-123), result);
}
@@ -1489,7 +1499,7 @@ async fn test_stop_after_if_nested(db: Pool<Postgres>) {
.arg("n", json!(123))
.run_until_complete(&db, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(json!("last step saw [123]"), result);
@@ -1498,7 +1508,7 @@ async fn test_stop_after_if_nested(db: Pool<Postgres>) {
.run_until_complete(&db, port)
.await;
let result = cjob.result.unwrap();
let result = cjob.json_result().unwrap();
assert_eq!(json!([-123]), result);
}
@@ -1552,7 +1562,7 @@ async fn test_python_flow(db: Pool<Postgres>) {
port,
)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {i}");
@@ -1587,7 +1597,7 @@ async fn test_python_flow_2(db: Pool<Postgres>) {
port,
)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("Hello"), "iteration: {i}");
@@ -1624,7 +1634,7 @@ func main(derp string) (string, error) {
.arg("derp", json!("world"))
.run_until_complete(&db, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("hello world"));
@@ -1655,7 +1665,7 @@ echo "hello $msg"
.run_until_complete(&db, port)
.await;
assert_eq!(job.result, Some(json!("hello world")));
assert_eq!(job.json_result(), Some(json!("hello world")));
}
#[sqlx::test(fixtures("base"))]
@@ -1682,7 +1692,7 @@ def main():
let result = run_job_in_new_worker_until_complete(&db, job, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("hello world"));
@@ -1715,7 +1725,7 @@ def main():
let result = run_job_in_new_worker_until_complete(&db, job, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(3));
@@ -1747,7 +1757,7 @@ def main():
let result = run_job_in_new_worker_until_complete(&db, job, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!("test-workspace"));
@@ -1803,7 +1813,7 @@ async fn test_empty_loop(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(0));
@@ -1843,7 +1853,7 @@ async fn test_invalid_first_step(db: Pool<Postgres>) {
let job = run_job_in_new_worker_until_complete(&db, flow, port).await;
assert_eq!(
job.result.unwrap(),
job.json_result().unwrap(),
serde_json::json!( {"error": {"name": "InternalErr", "message": "Expected an array value, found: {}"}})
);
}
@@ -1884,7 +1894,7 @@ async fn test_empty_loop_2(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([]));
@@ -1939,7 +1949,7 @@ async fn test_step_after_loop(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!(9));
@@ -2007,7 +2017,7 @@ async fn test_branchone_simple(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([1, 2]));
@@ -2043,7 +2053,7 @@ async fn test_branchone_with_cond(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([1, 3]));
@@ -2081,7 +2091,7 @@ async fn test_branchall_sequential(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([[1, 2], [1, 3]]));
@@ -2118,7 +2128,7 @@ async fn test_branchall_simple(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([[1, 2], [1, 3]]));
@@ -2165,7 +2175,7 @@ async fn test_branchall_skip_failure(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(
@@ -2202,7 +2212,7 @@ async fn test_branchall_skip_failure(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(
@@ -2266,7 +2276,7 @@ async fn test_branchone_nested(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(result, serde_json::json!([1, 2, 3]));
@@ -2323,7 +2333,7 @@ async fn test_branchall_nested(db: Pool<Postgres>) {
let flow = JobPayload::RawFlow { value: flow, path: None };
let result = run_job_in_new_worker_until_complete(&db, flow, port)
.await
.result
.json_result()
.unwrap();
println!("{:#?}", result);
@@ -2388,7 +2398,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
.arg("n", json!(0))
.run_until_complete(&db, port)
.await
.result
.json_result()
.unwrap();
assert!(result["from failure module"]["error"]
@@ -2404,7 +2414,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
.arg("n", json!(1))
.run_until_complete(&db, port)
.await
.result
.json_result()
.unwrap();
assert!(result["from failure module"]["error"]
@@ -2420,7 +2430,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
.arg("n", json!(2))
.run_until_complete(&db, port)
.await
.result
.json_result()
.unwrap();
assert!(result["from failure module"]["error"]
@@ -2436,7 +2446,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
.arg("n", json!(3))
.run_until_complete(&db, port)
.await
.result
.json_result()
.unwrap();
assert_eq!(json!({ "l": [0, 1, 2] }), result);
}
@@ -2642,8 +2652,7 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
let uuid = uuid.unwrap().unwrap();
let completed_job =
sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
.bind(uuid)
query!("SELECT script_path FROM completed_job WHERE id = $1", uuid)
.fetch_one(&db2)
.await
.unwrap();
@@ -2702,8 +2711,7 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
let uuid = uuid.unwrap().unwrap();
let completed_job =
sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
.bind(uuid)
query!("SELECT script_path FROM completed_job WHERE id = $1", uuid)
.fetch_one(&db2)
.await
.unwrap();
@@ -2778,8 +2786,7 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
let uuid = uuid.unwrap().unwrap();
let completed_job =
sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
.bind(uuid)
query!("SELECT script_path FROM completed_job WHERE id = $1", uuid)
.fetch_one(&db2)
.await
.unwrap();
@@ -2839,8 +2846,7 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
let uuid = uuid.unwrap().unwrap();
let completed_job =
sqlx::query_as::<_, CompletedJob>("SELECT * FROM completed_job WHERE id = $1")
.bind(uuid)
query!("SELECT script_path FROM completed_job WHERE id = $1", uuid)
.fetch_one(&db2)
.await
.unwrap();
backend/windmill-api/src/inputs.rs
+13 -4
View File
@@ -6,7 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
use crate::{db::ApiAuthed, jobs::CompletedJob};
use crate::db::ApiAuthed;
use axum::{
extract::{Path, Query},
routing::{get, post},
@@ -15,7 +15,7 @@ use axum::{
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::types::Uuid;
use sqlx::{types::Uuid, FromRow};
use std::{
fmt::{Display, Formatter},
vec,
@@ -102,6 +102,15 @@ pub struct Input {
success: bool,
}
#[derive(Debug, Serialize, Deserialize, FromRow)]
pub struct CompletedJobMini {
id: Uuid,
created_at: chrono::DateTime<chrono::Utc>,
args: Option<serde_json::Value>,
created_by: String,
success: bool,
}
async fn get_input_history(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
@@ -114,13 +123,13 @@ async fn get_input_history(
let mut tx = user_db.begin(&authed).await?;
let sql = &format!(
"select * from completed_job \
"select id, created_at, created_by, args, success from completed_job \
where {} = $1 and job_kind = $2 and workspace_id = $3 \
order by created_at desc limit $4 offset $5",
r.runnable_type.column_name()
);
let query = sqlx::query_as::<_, CompletedJob>(sql);
let query = sqlx::query_as::<_, CompletedJobMini>(sql);
let query = match r.runnable_type {
RunnableType::ScriptHash => query.bind(to_i64(&r.runnable_id)?),
backend/windmill-api/src/jobs.rs
+182 -110
View File
@@ -29,6 +29,7 @@ use hmac::Mac;
use hyper::{header::CONTENT_TYPE, http, HeaderMap, Request, StatusCode};
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use sql_builder::{prelude::*, quote, SqlBuilder};
use sqlx::types::JsonRawValue;
use sqlx::{query_scalar, types::Uuid, FromRow, Postgres, Transaction};
use tower_http::cors::{Any, CorsLayer};
use urlencoding::encode;
@@ -44,7 +45,7 @@ use windmill_common::{
users::username_to_permissioned_as,
utils::{not_found_if_none, now_from_db, paginate, require_admin, Pagination, StripPath},
};
use windmill_queue::{get_queued_job, push, PushIsolationLevel};
use windmill_queue::{job_is_complete, push, PushIsolationLevel};
pub fn workspaced_service() -> Router {
let cors = CorsLayer::new()
@@ -231,15 +232,15 @@ async fn cancel_job_api(
tx.commit().await?;
Ok(id.to_string())
} else {
let (job_o, tx) = get_job_by_id(tx, &w_id, id).await?;
tx.commit().await?;
let err = match job_o {
Some(Job::CompletedJob(_)) => {
return Ok(format!("queued job id {} is already completed", id))
}
_ => error::Error::NotFound(format!("queued job id {} does not exist", id)),
};
Err(err)
if job_is_complete(&db, id, &w_id).await.unwrap_or(false) {
return Ok(format!("queued job id {} is already completed", id));
} else {
return Err(error::Error::NotFound(format!(
"queued job id {} does not exist",
id
)));
}
}
}
@@ -274,15 +275,15 @@ async fn force_cancel(
tx.commit().await?;
Ok(id.to_string())
} else {
let (job_o, tx) = get_job_by_id(tx, &w_id, id).await?;
tx.commit().await?;
let err = match job_o {
Some(Job::CompletedJob(_)) => {
return Ok(format!("queued job id {} is already completed", id))
}
_ => error::Error::NotFound(format!("queued job id {} does not exist", id)),
};
Err(err)
if job_is_complete(&db, id, &w_id).await.unwrap_or(false) {
return Ok(format!("queued job id {} is already completed", id));
} else {
return Err(error::Error::NotFound(format!(
"queued job id {} does not exist",
id
)));
}
}
}
@@ -345,12 +346,29 @@ pub async fn get_path_tag_limits_cache_for_hash(
async fn get_job(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::JsonResult<Job> {
let tx = db.begin().await?;
let (job_o, tx) = get_job_by_id(tx, &w_id, id).await?;
let job = not_found_if_none(job_o, "Job", id.to_string())?;
tx.commit().await?;
Ok(Json(job))
) -> error::Result<Response> {
let cjob_option =
sqlx::query("SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2")
.bind(id)
.bind(&w_id)
.fetch_optional(&db)
.await?;
if let Some(job) = cjob_option {
let job = Job::CompletedJob(CompletedJob::from_row(&job)?);
Ok(Json(job).into_response())
} else {
let job_o = sqlx::query_as::<_, QueuedJob>(
"SELECT *
FROM queue WHERE id = $1 AND workspace_id = $2",
)
.bind(id)
.bind(&w_id)
.fetch_optional(&db)
.await?
.map(Job::QueuedJob);
let job: Job<'_> = not_found_if_none(job_o, "Job", id.to_string())?;
Ok(Json(job).into_response())
}
}
async fn get_job_logs(
@@ -369,39 +387,8 @@ async fn get_job_logs(
Ok(text)
}
pub async fn get_job_by_id<'c>(
mut tx: Transaction<'c, Postgres>,
w_id: &str,
id: Uuid,
) -> error::Result<(Option<Job>, Transaction<'c, Postgres>)> {
let cjob_option = sqlx::query_as::<_, CompletedJob>(
"SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2",
)
.bind(id)
.bind(w_id)
.fetch_optional(&mut *tx)
.await?;
let job_option = match cjob_option {
Some(job) => Some(Job::CompletedJob(job)),
None => get_queued_job(id, w_id, &mut tx).await?.map(Job::QueuedJob),
};
if job_option.is_some() {
Ok((job_option, tx))
} else {
// check if a job had been moved in-between queries
let cjob_option = sqlx::query_as::<_, CompletedJob>(
"SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2",
)
.bind(id)
.bind(w_id)
.fetch_optional(&mut *tx)
.await?;
Ok((cjob_option.map(Job::CompletedJob), tx))
}
}
#[derive(Debug, sqlx::FromRow, Serialize)]
pub struct CompletedJob {
pub struct CompletedJob<'rows> {
pub workspace_id: String,
pub id: Uuid,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -416,8 +403,8 @@ pub struct CompletedJob {
#[serde(skip_serializing_if = "Option::is_none")]
pub script_path: Option<String>,
pub args: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub result: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none", borrow)]
pub result: Option<&'rows JsonRawValue>,
#[serde(skip_serializing_if = "Option::is_none")]
pub logs: Option<String>,
pub deleted: bool,
@@ -447,6 +434,64 @@ pub struct CompletedJob {
pub tag: String,
}
impl<'row> CompletedJob<'row> {
pub fn json_result(&self) -> Option<serde_json::Value> {
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,
pub workspace_id: String,
pub id: Uuid,
#[serde(skip_serializing_if = "Option::is_none")]
pub parent_job: Option<Uuid>,
pub created_by: String,
pub created_at: chrono::DateTime<chrono::Utc>,
pub started_at: chrono::DateTime<chrono::Utc>,
pub duration_ms: i32,
pub success: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub script_hash: Option<ScriptHash>,
#[serde(skip_serializing_if = "Option::is_none")]
pub script_path: Option<String>,
pub deleted: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub raw_code: Option<String>,
pub canceled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub canceled_by: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub canceled_reason: Option<String>,
pub job_kind: JobKind,
#[serde(skip_serializing_if = "Option::is_none")]
pub schedule_path: Option<String>,
pub permissioned_as: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub flow_status: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub raw_flow: Option<serde_json::Value>,
pub is_flow_step: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub language: Option<ScriptLang>,
pub is_skipped: bool,
pub email: String,
pub visible_to_owner: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub mem_peak: Option<i32>,
pub tag: String,
}
impl<'a> IntoResponse for CompletedJob<'a> {
fn into_response(self) -> Response {
Json(self).into_response()
}
}
#[derive(Deserialize, Clone)]
pub struct RunJobQuery {
scheduled_for: Option<chrono::DateTime<chrono::Utc>>,
@@ -728,7 +773,7 @@ async fn list_jobs(
Path(w_id): Path<String>,
Query(pagination): Query<Pagination>,
Query(lq): Query<ListCompletedQuery>,
) -> error::JsonResult<Vec<Job>> {
) -> error::JsonResult<Vec<Job<'static>>> {
check_scopes(&authed, || format!("listjobs"))?;
let (per_page, offset) = paginate(pagination);
@@ -1086,8 +1131,8 @@ pub async fn cancel_suspended_job(
}
#[derive(Serialize)]
pub struct SuspendedJobFlow {
pub job: Job,
pub struct SuspendedJobFlow<'a> {
pub job: Job<'a>,
pub approvers: Vec<Approval>,
}
@@ -1101,7 +1146,7 @@ pub async fn get_suspended_job_flow(
Extension(db): Extension<DB>,
Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>,
Query(approver): Query<QueryApprover>,
) -> error::JsonResult<SuspendedJobFlow> {
) -> error::Result<Response> {
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)?;
@@ -1129,7 +1174,29 @@ pub async fn get_suspended_job_flow(
.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 cjob_option =
sqlx::query("SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2")
.bind(flow_id)
.bind(&w_id)
.fetch_optional(&db)
.await?;
let mut _rows = None;
let flow_o = if let Some(job) = cjob_option {
_rows = Some(job);
Some(Job::CompletedJob(CompletedJob::from_row(
_rows.as_ref().unwrap(),
)?))
} else {
sqlx::query_as::<_, QueuedJob>(
"SELECT *
FROM queue WHERE id = $1 AND workspace_id = $2",
)
.bind(flow_id)
.bind(&w_id)
.fetch_optional(&db)
.await?
.map(Job::QueuedJob)
};
let flow = not_found_if_none(flow_o, "Parent Flow", job.to_string())?;
let flow_status = flow
@@ -1166,7 +1233,7 @@ pub async fn get_suspended_job_flow(
approvers_from_status
};
Ok(Json(SuspendedJobFlow { job: flow, approvers }))
Ok(Json(SuspendedJobFlow { job: flow, approvers }).into_response())
}
pub async fn create_job_signature(
@@ -1246,12 +1313,12 @@ pub async fn get_resume_urls(
#[derive(Serialize, Debug)]
#[serde(tag = "type")]
pub enum Job {
pub enum Job<'a> {
QueuedJob(QueuedJob),
CompletedJob(CompletedJob),
CompletedJob(CompletedJob<'a>),
}
impl Job {
impl<'a> Job<'a> {
pub fn raw_flow(&self) -> Option<FlowValue> {
let value = match self {
Job::QueuedJob(job) => job.raw_flow.clone(),
@@ -1281,7 +1348,6 @@ struct UnifiedJob {
running: Option<bool>,
script_hash: Option<ScriptHash>,
script_path: Option<String>,
args: Option<serde_json::Value>,
duration_ms: Option<i32>,
success: Option<bool>,
deleted: bool,
@@ -1302,7 +1368,7 @@ struct UnifiedJob {
concurrency_time_window_s: Option<i32>,
}
impl From<UnifiedJob> for Job {
impl<'a> From<UnifiedJob> for Job<'a> {
fn from(uj: UnifiedJob) -> Self {
match uj.typ.as_ref() {
"CompletedJob" => Job::CompletedJob(CompletedJob {
@@ -1316,7 +1382,7 @@ impl From<UnifiedJob> for Job {
success: uj.success.unwrap(),
script_hash: uj.script_hash,
script_path: uj.script_path,
args: uj.args,
args: None,
result: None,
logs: None,
flow_status: None,
@@ -1346,7 +1412,7 @@ impl From<UnifiedJob> for Job {
started_at: uj.started_at,
script_hash: uj.script_hash,
script_path: uj.script_path,
args: uj.args,
args: None,
running: uj.running.unwrap(),
scheduled_for: uj.scheduled_for.unwrap(),
logs: None,
@@ -2621,7 +2687,7 @@ async fn list_completed_jobs(
Path(w_id): Path<String>,
Query(pagination): Query<Pagination>,
Query(lq): Query<ListCompletedQuery>,
) -> error::JsonResult<Vec<CompletedJob>> {
) -> error::JsonResult<Vec<ListableCompletedJob>> {
check_scopes(&authed, || format!("listjobs"))?;
let (per_page, offset) = paginate(pagination);
@@ -2642,9 +2708,6 @@ async fn list_completed_jobs(
"success",
"script_hash",
"script_path",
"null as args",
"null as result",
"null as logs",
"deleted",
"canceled",
"canceled_by",
@@ -2662,83 +2725,92 @@ async fn list_completed_jobs(
"visible_to_owner",
"mem_peak",
"tag",
"'CompletedJob' as type",
],
)
.sql()?;
let jobs = sqlx::query_as::<_, CompletedJob>(&sql)
let jobs = sqlx::query_as::<_, ListableCompletedJob>(&sql)
.fetch_all(&db)
.await?;
Ok(Json(jobs))
}
async fn get_completed_job(
async fn get_completed_job<'a>(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::JsonResult<CompletedJob> {
let job_o = sqlx::query_as::<_, CompletedJob>(
"SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2",
)
.bind(id)
.bind(w_id)
.fetch_optional(&db)
.await?;
) -> error::Result<Response> {
let job_o = sqlx::query("SELECT * FROM completed_job WHERE id = $1 AND workspace_id = $2")
.bind(id)
.bind(w_id)
.fetch_optional(&db)
.await?;
let job = not_found_if_none(job_o, "Completed Job", id.to_string())?;
Ok(Json(job))
Ok(CompletedJob::from_row(&job)?.into_response())
}
#[derive(FromRow)]
pub struct RawResult<'a> {
pub result: &'a JsonRawValue,
}
impl<'a> IntoResponse for RawResult<'a> {
fn into_response(self) -> Response {
Json(self.result).into_response()
}
}
async fn get_completed_job_result(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::JsonResult<Option<serde_json::Value>> {
let result_o = sqlx::query_scalar!(
"SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2",
id,
w_id,
)
.fetch_optional(&db)
.await?;
) -> error::Result<Response> {
let result_o =
sqlx::query("SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2")
.bind(id)
.bind(w_id)
.fetch_optional(&db)
.await?;
let result = not_found_if_none(result_o, "Completed Job", id.to_string())?;
Ok(Json(result))
Ok(RawResult::from_row(&result)?.into_response())
}
#[derive(Serialize)]
struct CompletedJobResult {
struct CompletedJobResult<'c> {
completed: bool,
result: Option<serde_json::Value>,
result: Option<&'c JsonRawValue>,
}
async fn get_completed_job_result_maybe(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::JsonResult<CompletedJobResult> {
let result_o = sqlx::query_scalar!(
"SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2",
id,
w_id,
)
.fetch_optional(&db)
.await?;
) -> error::Result<Response> {
let result_o =
sqlx::query("SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2")
.bind(id)
.bind(w_id)
.fetch_optional(&db)
.await?;
if let Some(result) = result_o {
Ok(Json(CompletedJobResult { completed: true, result }))
let res = RawResult::from_row(&result)?;
Ok(Json(CompletedJobResult { completed: true, result: Some(res.result) }).into_response())
} else {
Ok(Json(CompletedJobResult { completed: false, result: None }))
Ok(Json(CompletedJobResult { completed: false, result: None }).into_response())
}
}
async fn delete_completed_job(
async fn delete_completed_job<'a>(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::JsonResult<CompletedJob> {
) -> error::Result<Response> {
check_scopes(&authed, || format!("deletejob"))?;
let mut tx = user_db.begin(&authed).await?;
require_admin(authed.is_admin, &authed.username)?;
let job_o = sqlx::query_as::<_, CompletedJob>(
let job_o = sqlx::query(
"UPDATE completed_job SET logs = '', result = null, deleted = true WHERE id = $1 AND workspace_id = $2 \
RETURNING *",
)
@@ -2761,5 +2833,5 @@ async fn delete_completed_job(
.await?;
tx.commit().await?;
Ok(Json(job))
Ok(CompletedJob::from_row(&job)?.into_response())
}
backend/windmill-queue/src/jobs.rs
+11
View File
@@ -1371,6 +1371,17 @@ pub async fn delete_job<'c, R: rsmq_async::RsmqConnection + Clone + Send>(
Ok(tx)
}
pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result<bool> {
Ok(sqlx::query_scalar!(
"SELECT EXISTS(SELECT 1 FROM completed_job WHERE id = $1 AND workspace_id = $2)",
id,
w_id
)
.fetch_one(db)
.await?
.unwrap_or(false))
}
pub async fn get_queued_job<'c>(
id: Uuid,
w_id: &str,