diff --git a/backend/.sqlx/query-6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a.json b/backend/.sqlx/query-003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91.json similarity index 76% rename from backend/.sqlx/query-6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a.json rename to backend/.sqlx/query-003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91.json index d4289d89b2..00797f7524 100644 --- a/backend/.sqlx/query-6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a.json +++ b/backend/.sqlx/query-003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind != 'flow' AND job_kind != 'flowpreview' AND job_kind != 'singlescriptflow' AND same_worker = false RETURNING id, workspace_id, last_ping", + "query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", "describe": { "columns": [ { @@ -30,5 +30,5 @@ false ] }, - "hash": "6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a" + "hash": "003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91" } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 6fb92b9fd2..ad00155ceb 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -767,7 +767,7 @@ async fn handle_zombie_jobs if *RESTART_ZOMBIE_JOBS { let restarted = sqlx::query!( "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval - AND running = true AND job_kind != 'flow' AND job_kind != 'flowpreview' AND job_kind != 'singlescriptflow' AND same_worker = false RETURNING id, workspace_id, last_ping", + AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", *ZOMBIE_JOB_TIMEOUT, ) .fetch_all(db) @@ -788,7 +788,7 @@ async fn handle_zombie_jobs } } - let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != 'flow' AND job_kind != 'flowpreview' AND job_kind != 'singlescriptflow'".to_string(); + let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow')".to_string(); if *RESTART_ZOMBIE_JOBS { timeout_query.push_str(" AND same_worker = true"); }; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 7d0c4c342a..dec0b74148 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -665,6 +665,7 @@ pub struct ListQueueQuery { pub tag: Option, pub scheduled_for_before_now: Option, pub all_workspaces: Option, + pub is_flow_step: Option, } fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> SqlBuilder { @@ -708,6 +709,9 @@ fn list_queue_jobs_query(w_id: &str, lq: &ListQueueQuery, fields: &[&str]) -> Sq if let Some(dt) = &lq.started_after { sqlb.and_where_ge("started_at", format!("to_timestamp({})", dt.timestamp())); } + if let Some(fs) = &lq.is_flow_step { + sqlb.and_where_eq("is_flow_step", fs); + } if let Some(dt) = &lq.created_before { sqlb.and_where_le("created_at", format!("to_timestamp({})", dt.timestamp())); @@ -987,6 +991,7 @@ async fn list_jobs( schedule_path: lq.schedule_path, scheduled_for_before_now: lq.scheduled_for_before_now, all_workspaces: lq.all_workspaces, + is_flow_step: lq.is_flow_step, }, &[ "'QueuedJob' as typ", diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 72fb0dab5d..1d8cdfd1fc 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -2766,7 +2766,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, Some(path), None, - JobKind::SingleScriptFlow, + JobKind::Flow, Some(flow_value.clone()), Some(FlowStatus::new(&flow_value)), // this is a new flow being pushed, flow_status is set to flow_value None,