diff --git a/backend/.sqlx/query-454a611a5a162b2ace137c139bd5383bc7fe142c515dac3edd47991839485e51.json b/backend/.sqlx/query-454a611a5a162b2ace137c139bd5383bc7fe142c515dac3edd47991839485e51.json new file mode 100644 index 0000000000..2f62420f87 --- /dev/null +++ b/backend/.sqlx/query-454a611a5a162b2ace137c139bd5383bc7fe142c515dac3edd47991839485e51.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM job_result_stream_v2\n WHERE job_id NOT IN (SELECT id FROM v2_job_queue)\n AND job_id NOT IN (\n SELECT id FROM v2_job_completed\n WHERE completed_at > NOW() - INTERVAL '60 seconds'\n )\n RETURNING job_id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "job_id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false + ] + }, + "hash": "454a611a5a162b2ace137c139bd5383bc7fe142c515dac3edd47991839485e51" +} diff --git a/backend/.sqlx/query-a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec.json b/backend/.sqlx/query-a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec.json deleted file mode 100644 index 66af36c0fd..0000000000 --- a/backend/.sqlx/query-a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM job_result_stream_v2 WHERE job_id NOT IN (SELECT id FROM v2_job_queue) RETURNING job_id", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "job_id", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - false - ] - }, - "hash": "a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec" -} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index bfbfaa2d87..b854516b69 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -3092,7 +3092,13 @@ RETURNING job_id" async fn cleanup_job_result_stream_orphaned_jobs(db: &DB) -> error::Result<()> { let result = sqlx::query!( - "DELETE FROM job_result_stream_v2 WHERE job_id NOT IN (SELECT id FROM v2_job_queue) RETURNING job_id", + "DELETE FROM job_result_stream_v2 + WHERE job_id NOT IN (SELECT id FROM v2_job_queue) + AND job_id NOT IN ( + SELECT id FROM v2_job_completed + WHERE completed_at > NOW() - INTERVAL '60 seconds' + ) + RETURNING job_id", ) .fetch_all(db) .await?; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 6974c5f244..4116049a40 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -114,7 +114,10 @@ use windmill_common::{ get_latest_flow_version_info_for_path, get_script_info_for_hash, utils::empty_as_none, ScriptHashInfo, BASE_URL, }; -use windmill_queue::{job_is_complete, push, PushArgs, PushArgsOwned, PushIsolationLevel}; +use windmill_queue::{ + get_result_and_success_by_id_from_flow, job_is_complete, push, PushArgs, PushArgsOwned, + PushIsolationLevel, +}; pub fn workspaced_service() -> Router { let cors = CorsLayer::new() @@ -3471,8 +3474,11 @@ pub async fn run_workflow_as_code( let JobExtended { inner: job, raw_code, raw_lock, .. } = job; let (_debouncing_settings, concurrency_settings) = - windmill_common::runnable_settings::prefetch_cached_from_handle(job.runnable_settings_handle, &db) - .await?; + windmill_common::runnable_settings::prefetch_cached_from_handle( + job.runnable_settings_handle, + &db, + ) + .await?; let (job_payload, tag, _delete_after_use, timeout, on_behalf_of) = match job.job_kind { JobKind::Preview => ( @@ -4232,7 +4238,7 @@ pub async fn stream_job( }; let poll_delay_ms = run_query.poll_delay_ms; - let uuid = match runnable_id { + let (uuid, early_return) = match runnable_id { RunnableId::ScriptId(ScriptId::ScriptPath(script_path)) | RunnableId::HubScript(script_path) => { let (uuid, _, _) = push_script_job_by_path_into_queue( @@ -4247,10 +4253,10 @@ pub async fn stream_job( None, ) .await?; - uuid + (uuid, None) } RunnableId::ScriptId(ScriptId::ScriptHash(script_hash)) => { - run_job_by_hash_inner( + let (uuid, _) = run_job_by_hash_inner( authed.clone(), db.clone(), user_db, @@ -4260,11 +4266,11 @@ pub async fn stream_job( args, None, ) - .await? - .0 + .await?; + (uuid, None) } RunnableId::FlowId(FlowId::FlowPath(flow_path)) => { - push_flow_job_by_path_into_queue( + let (uuid, early_return, _) = push_flow_job_by_path_into_queue( authed.clone(), db.clone(), None, @@ -4275,11 +4281,11 @@ pub async fn stream_job( args, None, ) - .await? - .0 + .await?; + (uuid, early_return) } RunnableId::FlowId(FlowId::FlowVersion(version)) => { - run_flow_by_version_inner( + let (uuid, early_return) = run_flow_by_version_inner( authed.clone(), db.clone(), user_db, @@ -4289,8 +4295,8 @@ pub async fn stream_job( args, None, ) - .await? - .0 + .await?; + (uuid, early_return) } }; @@ -4321,6 +4327,7 @@ pub async fn stream_job( None, tx, poll_delay_ms, + early_return, ); let body = axum::body::Body::from_stream(stream.map(Result::<_, std::convert::Infallible>::Ok)); @@ -4643,12 +4650,19 @@ fn register_potential_assets_on_inline_execution( let columns = asset.columns.as_ref().map(|cols| { cols.iter() .map(|(col_name, col_access_type)| { - (col_name.clone(), windmill_common::assets::asset_access_type_from_parser(*col_access_type)) + ( + col_name.clone(), + windmill_common::assets::asset_access_type_from_parser( + *col_access_type, + ), + ) }) .collect() }); register_runtime_asset(InsertRuntimeAssetParams { - access_type: asset.access_type.map(windmill_common::assets::asset_access_type_from_parser), + access_type: asset + .access_type + .map(windmill_common::assets::asset_access_type_from_parser), asset_kind: windmill_common::assets::asset_kind_from_parser(asset.kind), asset_path: asset.path, columns, @@ -5825,6 +5839,7 @@ async fn get_job_update( no_logs, is_flow, None, + None, ) .await?, )) @@ -5865,6 +5880,7 @@ async fn get_job_update_sse( is_flow, tx, poll_delay_ms, + None, ); let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(|x| { @@ -5902,6 +5918,7 @@ pub fn start_job_update_sse_stream( is_flow: Option, tx: tokio::sync::mpsc::Sender, poll_delay_ms: Option, + early_return: Option, ) -> () { tokio::spawn(async move { let mut log_offset = initial_log_offset; @@ -5929,6 +5946,7 @@ pub fn start_job_update_sse_stream( no_logs, is_flow, flow_stream_job_id, + early_return.as_deref(), ) .await { @@ -6046,6 +6064,7 @@ pub fn start_job_update_sse_stream( no_logs, is_flow, flow_stream_job_id, + early_return.as_deref(), ) .await { @@ -6176,6 +6195,7 @@ async fn get_job_update_data( no_logs: Option, is_flow: Option, flow_stream_job_id: Option, + early_return: Option<&str>, ) -> error::Result { let tags = if log_view { log_job_view( @@ -6339,6 +6359,16 @@ async fn get_job_update_data( let flow_stream_job_id = flow_stream_job_id.or(new_flow_stream_job_id); + let result = if let Some(early_return) = early_return { + match get_result_and_success_by_id_from_flow(db, w_id, job_id, early_return, None).await + { + Ok((early_result, _)) => Some(early_result), + Err(_) => result, + } + } else { + result + }; + let flow_stream_delta = get_flow_stream_delta(db, flow_stream_job_id, stream_offset).await?; diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index 211861f17e..44985ff8d6 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/backend/windmill-api/src/triggers/http/handler.rs @@ -505,7 +505,7 @@ async fn route_job( match trigger.request_type { RequestType::SyncSse => { // Trigger the job (always async when streaming) - let (uuid, _, _, _) = trigger_runnable_inner( + let (uuid, _, early_return, _) = trigger_runnable_inner( &db, None, Some(user_db.clone()), @@ -553,6 +553,7 @@ async fn route_job( None, tx, None, + early_return, ); let body = axum::body::Body::from_stream(