remove use of singlescript flow job kinds

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"
 }

backend/src/monitor.rs

@@ -767,7 +767,7 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
     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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         }
     }
 
-    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");
     };

backend/windmill-api/src/jobs.rs

@@ -665,6 +665,7 @@ pub struct ListQueueQuery {
     pub tag: Option<String>,
     pub scheduled_for_before_now: Option<bool>,
     pub all_workspaces: Option<bool>,
+    pub is_flow_step: Option<bool>,
 }
 
 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",

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,