From 22a7da58b1d20721892906cba2dee6fbeb1cc1fd Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 11 Sep 2023 08:00:02 +0200 Subject: [PATCH] feat: zero copy result for job result (#2263) * feat: zero copy result for job result * update --- ...76503cee1b45eba150c3082eb246ea3f98d47.json | 23 ++ backend/Cargo.toml | 2 +- backend/tests/worker.rs | 200 ++++++------ backend/windmill-api/src/inputs.rs | 17 +- backend/windmill-api/src/jobs.rs | 292 +++++++++++------- backend/windmill-queue/src/jobs.rs | 11 + 6 files changed, 333 insertions(+), 212 deletions(-) create mode 100644 backend/.sqlx/query-9f16a61d6a9a42f3fd3e30a1e7776503cee1b45eba150c3082eb246ea3f98d47.json diff --git a/backend/.sqlx/query-9f16a61d6a9a42f3fd3e30a1e7776503cee1b45eba150c3082eb246ea3f98d47.json b/backend/.sqlx/query-9f16a61d6a9a42f3fd3e30a1e7776503cee1b45eba150c3082eb246ea3f98d47.json new file mode 100644 index 0000000000..aa64996dc8 --- /dev/null +++ b/backend/.sqlx/query-9f16a61d6a9a42f3fd3e30a1e7776503cee1b45eba150c3082eb246ea3f98d47.json @@ -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" +} diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 34638cf683..e5acd6ef7c 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -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" diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index b3fd54db8d..ec8daa4f0d 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -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, + pub created_by: String, + pub created_at: chrono::DateTime, + pub started_at: chrono::DateTime, + pub duration_ms: i32, + pub success: bool, + pub script_path: Option, + pub args: Option, + pub result: Option, + pub logs: Option, + pub deleted: bool, + pub raw_code: Option, + pub canceled: bool, + pub canceled_by: Option, + pub canceled_reason: Option, + pub schedule_path: Option, + pub permissioned_as: String, + pub flow_status: Option, + pub raw_flow: Option, + pub is_flow_step: bool, + pub is_skipped: bool, + pub email: String, + pub visible_to_owner: bool, + pub mem_peak: Option, + pub tag: String, + pub script_hash: Option, + pub language: Option, + pub job_kind: JobKind, + +} + +impl CompletedJob { + pub fn json_result(&self) -> Option { + 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, 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) -> 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) -> 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 { 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) { .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) { .arg("items", json!((0..257).collect::>())) .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) { .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) { .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) -> 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) { 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) { 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) { 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) { 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) { .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) { .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) { .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) { .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) { 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) { 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) { 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) { 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) { 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) { 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) { 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) { 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) { 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) { 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) { 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) { 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) { 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) { 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) { .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) { .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) { .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) { .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) { 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) { 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) { 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) { 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(); diff --git a/backend/windmill-api/src/inputs.rs b/backend/windmill-api/src/inputs.rs index 8284b5dd9a..3949bf3544 100644 --- a/backend/windmill-api/src/inputs.rs +++ b/backend/windmill-api/src/inputs.rs @@ -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, + args: Option, + created_by: String, + success: bool, +} + async fn get_input_history( authed: ApiAuthed, Extension(user_db): Extension, @@ -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)?), diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 8299497d4a..b6fbb72260 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -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, Path((w_id, id)): Path<(String, Uuid)>, -) -> error::JsonResult { - 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 { + 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, 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, pub args: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub result: Option, + #[serde(skip_serializing_if = "Option::is_none", borrow)] + pub result: Option<&'rows JsonRawValue>, #[serde(skip_serializing_if = "Option::is_none")] pub logs: Option, pub deleted: bool, @@ -447,6 +434,64 @@ pub struct CompletedJob { pub tag: String, } +impl<'row> CompletedJob<'row> { + pub fn json_result(&self) -> Option { + self.result + .as_ref() + .map(|r| serde_json::from_str(r.get()).ok()) + .flatten() + } +} + +#[derive(Debug, sqlx::FromRow, Serialize)] +pub struct ListableCompletedJob { + pub r#type: String, + pub workspace_id: String, + pub id: Uuid, + #[serde(skip_serializing_if = "Option::is_none")] + pub parent_job: Option, + pub created_by: String, + pub created_at: chrono::DateTime, + pub started_at: chrono::DateTime, + pub duration_ms: i32, + pub success: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub script_hash: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub script_path: Option, + pub deleted: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub raw_code: Option, + pub canceled: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub canceled_by: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub canceled_reason: Option, + pub job_kind: JobKind, + #[serde(skip_serializing_if = "Option::is_none")] + pub schedule_path: Option, + pub permissioned_as: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub flow_status: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub raw_flow: Option, + pub is_flow_step: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub language: Option, + pub is_skipped: bool, + pub email: String, + pub visible_to_owner: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub mem_peak: Option, + pub tag: String, +} + +impl<'a> IntoResponse for CompletedJob<'a> { + fn into_response(self) -> Response { + Json(self).into_response() + } +} + #[derive(Deserialize, Clone)] pub struct RunJobQuery { scheduled_for: Option>, @@ -728,7 +773,7 @@ async fn list_jobs( Path(w_id): Path, Query(pagination): Query, Query(lq): Query, -) -> error::JsonResult> { +) -> error::JsonResult>> { 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, } @@ -1101,7 +1146,7 @@ pub async fn get_suspended_job_flow( Extension(db): Extension, Path((w_id, job, resume_id, secret)): Path<(String, Uuid, u32, String)>, Query(approver): Query, -) -> error::JsonResult { +) -> 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)?; @@ -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 { let value = match self { Job::QueuedJob(job) => job.raw_flow.clone(), @@ -1281,7 +1348,6 @@ struct UnifiedJob { running: Option, script_hash: Option, script_path: Option, - args: Option, duration_ms: Option, success: Option, deleted: bool, @@ -1302,7 +1368,7 @@ struct UnifiedJob { concurrency_time_window_s: Option, } -impl From for Job { +impl<'a> From for Job<'a> { fn from(uj: UnifiedJob) -> Self { match uj.typ.as_ref() { "CompletedJob" => Job::CompletedJob(CompletedJob { @@ -1316,7 +1382,7 @@ impl From 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 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, Query(pagination): Query, Query(lq): Query, -) -> error::JsonResult> { +) -> error::JsonResult> { 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, Path((w_id, id)): Path<(String, Uuid)>, -) -> error::JsonResult { - 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 { + 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, Path((w_id, id)): Path<(String, Uuid)>, -) -> error::JsonResult> { - 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 { + 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, + result: Option<&'c JsonRawValue>, } async fn get_completed_job_result_maybe( Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, -) -> error::JsonResult { - 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 { + 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, Path((w_id, id)): Path<(String, Uuid)>, -) -> error::JsonResult { +) -> error::Result { 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()) } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index ac1918042b..9db9d129f6 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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 { + 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,