diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index a3f117721f..6f2f210300 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -4,10 +4,7 @@ use std::{ fmt::Display, ops::Mul, str::FromStr, - sync::{ - atomic::{AtomicU16, Ordering}, - Arc, - }, + sync::{atomic::Ordering, Arc}, time::Duration, }; @@ -15,11 +12,7 @@ use chrono::{NaiveDateTime, Utc}; use futures::{stream::FuturesUnordered, StreamExt}; use serde::{de::DeserializeOwned, Deserializer}; use sqlx::{Pool, Postgres}; -use tokio::{ - join, - sync::{mpsc, RwLock}, -}; -use uuid::Uuid; +use tokio::{join, sync::RwLock}; #[cfg(feature = "embedding")] use windmill_api::embeddings::update_embeddings_db; @@ -37,7 +30,6 @@ use windmill_common::{ auth::JWT_SECRET, ee::CriticalErrorChannel, error, - flow_status::{FlowStatus, FlowStatusModule}, global_settings::{ BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, CRITICAL_ERROR_CHANNELS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, @@ -50,12 +42,11 @@ use windmill_common::{ SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING, }, indexer::load_indexer_config, - jobs::QueuedJob, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, server::load_smtp_config, tracing_init::JSON_FMT, users::truncate_token, - utils::{now_from_db, rd_string, report_critical_error, Mode}, + utils::{rd_string, Mode}, worker::{ load_worker_config, make_pull_query, make_suspended_pull_query, reload_custom_tags_setting, update_min_version, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG, @@ -66,11 +57,9 @@ use windmill_common::{ MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED, SERVICE_LOG_RETENTION_SECS, }; -use windmill_queue::cancel_job; use windmill_worker::{ - create_token_for_owner, handle_job_error, AuthedClient, SameWorkerPayload, SameWorkerSender, - SendResult, BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, - NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, SCRIPT_TOKEN_EXPIRY, + BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, + NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, }; #[cfg(feature = "parquet")] @@ -89,18 +78,6 @@ use crate::ee::set_license_key; #[cfg(feature = "prometheus")] lazy_static::lazy_static! { - - static ref QUEUE_ZOMBIE_RESTART_COUNT: prometheus::IntCounter = prometheus::register_int_counter!( - "queue_zombie_restart_count", - "Total number of jobs restarted due to ping timeout." - ) - .unwrap(); - static ref QUEUE_ZOMBIE_DELETE_COUNT: prometheus::IntCounter = prometheus::register_int_counter!( - "queue_zombie_delete_count", - "Total number of jobs deleted due to their ping timing out in an unrecoverable state." - ) - .unwrap(); - static ref QUEUE_COUNT: prometheus::IntGaugeVec = prometheus::register_int_gauge_vec!( "queue_count", "Number of jobs in the queue", @@ -109,26 +86,7 @@ lazy_static::lazy_static! { } lazy_static::lazy_static! { - static ref ZOMBIE_JOB_TIMEOUT: String = std::env::var("ZOMBIE_JOB_TIMEOUT") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or_else(|| "60".to_string()); - - static ref FLOW_ZOMBIE_TRANSITION_TIMEOUT: String = std::env::var("FLOW_ZOMBIE_TRANSITION_TIMEOUT") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or_else(|| "60".to_string()); - - - pub static ref RESTART_ZOMBIE_JOBS: bool = std::env::var("RESTART_ZOMBIE_JOBS") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(true); - - - static ref QUEUE_COUNT_TAGS: Arc>> = Arc::new(RwLock::new(Vec::new())); - } pub async fn initial_load( @@ -1232,13 +1190,15 @@ pub async fn monitor_db( ) { let zombie_jobs_f = async { if server_mode && !initial_load { - handle_zombie_jobs(db, base_internal_url, "server").await; - match handle_zombie_flows(db).await { - Err(err) => { - tracing::error!("Error handling zombie flows: {:?}", err); - } - _ => {} - } + windmill_worker::monitor::monitor_once( + &db, + base_internal_url, + "server", + None, + None, + None, + ) + .await; } }; let expired_items_f = async { @@ -1517,346 +1477,6 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> { Ok(()) } -async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker_name: &str) { - if *RESTART_ZOMBIE_JOBS { - let restarted = sqlx::query!( - "WITH zombie_jobs AS ( - UPDATE v2_job_queue q SET running = false, started_at = null - FROM v2_job j, v2_job_runtime r - WHERE j.id = q.id AND j.id = r.id - AND ping < now() - ($1 || ' seconds')::interval - AND running = true - AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') - AND same_worker = false - RETURNING q.id, q.workspace_id, ping - ), - update_concurrency AS ( - UPDATE concurrency_counter cc - SET job_uuids = job_uuids - zj.id::text - FROM zombie_jobs zj - INNER JOIN concurrency_key ck ON ck.job_id = zj.id - WHERE cc.concurrency_id = ck.key - ) - SELECT id, workspace_id, ping FROM zombie_jobs", - *ZOMBIE_JOB_TIMEOUT, - ) - .fetch_all(db) - .await - .ok() - .unwrap_or_else(|| vec![]); - - #[cfg(feature = "prometheus")] - if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { - QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); - } - - let base_url = BASE_URL.read().await.clone(); - for r in restarted { - let last_ping = if let Some(x) = r.ping { - format!("last ping at {x}") - } else { - "no last ping".to_string() - }; - let url = format!("{}/run/{}?workspace={}", base_url, r.id, r.workspace_id,); - let error_message = format!( - "Zombie job {} on {} ({}) detected, restarting it, {}", - r.id, r.workspace_id, url, last_ping - ); - - let _ = sqlx::query!(" - INSERT INTO job_logs (job_id, logs) - VALUES ($1, 'Restarted job after not receiving job''s ping for too long the ' || now() || '\n\n') - ON CONFLICT (job_id) DO UPDATE SET logs = job_logs.logs || '\n' || EXCLUDED.logs - WHERE job_logs.job_id = $1", - r.id - ) - .execute(db) - .await; - tracing::error!(error_message); - report_critical_error(error_message, db.clone(), Some(&r.workspace_id), None).await; - } - } - - let same_worker_timeout_jobs = { - let long_same_worker_jobs = sqlx::query!( - "SELECT worker, array_agg(v2_job_queue.id) as ids FROM v2_job_queue LEFT JOIN v2_job ON v2_job_queue.id = v2_job.id LEFT JOIN v2_job_runtime ON v2_job_queue.id = v2_job_runtime.id WHERE v2_job_queue.created_at < now() - ('60 seconds')::interval - AND running = true AND ping IS NULL AND same_worker = true AND worker IS NOT NULL GROUP BY worker", - ) - .fetch_all(db) - .await - .ok() - .unwrap_or_else(|| vec![]); - - let worker_ids = long_same_worker_jobs - .iter() - .map(|x| x.worker.clone().unwrap_or_default()) - .collect::>(); - - let long_dead_workers: std::collections::HashSet = sqlx::query_scalar!( - "WITH worker_ids AS (SELECT unnest($1::text[]) as worker) - SELECT worker_ids.worker FROM worker_ids - LEFT JOIN worker_ping ON worker_ids.worker = worker_ping.worker - WHERE worker_ping.worker IS NULL OR worker_ping.ping_at < now() - ('60 seconds')::interval", - &worker_ids[..] - ) - .fetch_all(db) - .await - .ok() - .unwrap_or_else(|| vec![]) - .into_iter() - .filter_map(|x| x) - .collect(); - - let mut timeouts: Vec = vec![]; - for worker in long_same_worker_jobs { - if worker.worker.is_some() && long_dead_workers.contains(&worker.worker.unwrap()) { - if let Some(ids) = worker.ids { - timeouts.extend(ids); - } - } - } - if !timeouts.is_empty() { - tracing::error!( - "Failing same worker zombie jobs: {:?}", - timeouts - .iter() - .map(|x| x.hyphenated().to_string()) - .collect::>() - .join(",") - ); - } - - let jobs = sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE id = ANY($1)") - .bind(&timeouts[..]) - .fetch_all(db) - .await - .map_err(|e| tracing::error!("Error fetching same worker jobs: {:?}", e)) - .unwrap_or_default(); - - jobs - }; - - let non_restartable_jobs = if *RESTART_ZOMBIE_JOBS { - vec![] - } else { - sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE last_ping < now() - ($1 || ' seconds')::interval - AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false") - .bind(ZOMBIE_JOB_TIMEOUT.as_str()) - .fetch_all(db) - .await - .ok() - .unwrap_or_else(|| vec![]) - }; - - let timeouts = non_restartable_jobs - .into_iter() - .chain(same_worker_timeout_jobs) - .collect::>(); - - #[cfg(feature = "prometheus")] - if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { - QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _); - } - - for job in timeouts { - // since the job is unrecoverable, the same worker queue should never be sent anything - let (same_worker_tx_never_used, _same_worker_rx_never_used) = - mpsc::channel::(1); - let same_worker_tx_never_used = - SameWorkerSender(same_worker_tx_never_used, Arc::new(AtomicU16::new(0))); - let (send_result_never_used, _send_result_rx_never_used) = mpsc::channel::(1); - - let label = if job.permissioned_as != format!("u/{}", job.created_by) - && job.permissioned_as != job.created_by - { - format!("ephemeral-script-end-user-{}", job.created_by) - } else { - "ephemeral-script".to_string() - }; - let token = create_token_for_owner( - &db, - &job.workspace_id, - &job.permissioned_as, - &label, - *SCRIPT_TOKEN_EXPIRY, - &job.email, - &job.id, - ) - .await - .expect("could not create job token"); - - let client = AuthedClient { - base_internal_url: base_internal_url.to_string(), - token, - workspace: job.workspace_id.to_string(), - force_client: None, - }; - - let last_ping = job.last_ping.clone(); - let _ = handle_job_error( - db, - &client, - &job, - 0, - None, - error::Error::ExecutionErr(format!( - "Job timed out after no ping from job since {} (ZOMBIE_JOB_TIMEOUT: {}, same_worker: {})", - last_ping - .map(|x| x.to_string()) - .unwrap_or_else(|| "no ping".to_string()), - *ZOMBIE_JOB_TIMEOUT, - job.same_worker - )), - true, - same_worker_tx_never_used, - "", - worker_name, - send_result_never_used, - #[cfg(feature = "benchmark")] - &mut windmill_worker::bench::BenchmarkIter::new(), - ) - .await; - } -} - -async fn handle_zombie_flows(db: &DB) -> error::Result<()> { - let flows = sqlx::query!( - r#" - SELECT - id AS "id!", workspace_id AS "workspace_id!", parent_job, is_flow_step, - flow_status AS "flow_status: Box", last_ping, same_worker - FROM v2_as_queue - WHERE running = true AND suspend = 0 AND suspend_until IS null AND scheduled_for <= now() - AND (job_kind = 'flow' OR job_kind = 'flowpreview' OR job_kind = 'flownode') - AND last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval - AND canceled = false - "#, - FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str() - ) - .fetch_all(db) - .await?; - - for flow in flows { - let status = flow - .flow_status - .as_deref() - .and_then(|x| serde_json::from_str::(x).ok()); - if !flow.same_worker.unwrap_or(false) - && status.is_some_and(|s| { - s.modules - .get(0) - .is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. })) - }) - { - let error_message = format!( - "Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.", - flow.id, flow.workspace_id - ); - tracing::error!(error_message); - report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None).await; - // if the flow hasn't started and is a zombie, we can simply restart it - let mut tx = db.begin().await?; - - let concurrency_key = - sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", flow.id) - .fetch_optional(&mut *tx) - .await?; - - if let Some(key) = concurrency_key { - sqlx::query!( - "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", - key, - flow.id.hyphenated().to_string() - ) - .execute(&mut *tx) - .await?; - } - - sqlx::query!( - "UPDATE v2_job_queue SET running = false, started_at = null - WHERE id = $1 AND canceled_by IS NULL", - flow.id - ) - .execute(&mut *tx) - .await?; - - tx.commit().await?; - } else { - let id = flow.id.clone(); - let last_ping = flow.last_ping.clone(); - let now = now_from_db(db).await?; - let reason = format!( - "{} was hanging in between 2 steps. Last ping: {last_ping:?} (now: {now})", - if flow.is_flow_step.unwrap_or(false) && flow.parent_job.is_some() { - format!("Flow was cancelled because subflow {id}") - } else { - format!("Flow {id} was cancelled because it") - } - ); - report_critical_error(reason.clone(), db.clone(), Some(&flow.workspace_id), None).await; - cancel_zombie_flow_job(db, flow.id, &flow.workspace_id, reason).await?; - } - } - - let flows2 = sqlx::query!( - r#" - DELETE - FROM parallel_monitor_lock - WHERE last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval - RETURNING parent_flow_id, job_id, last_ping, (SELECT workspace_id FROM v2_job_queue q - WHERE q.id = parent_flow_id AND q.running = true AND q.canceled_by IS NULL - ) AS workspace_id - "#, - FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str() - ) - .fetch_all(db) - .await?; - - for flow in flows2 { - if let Some(parent_flow_workspace_id) = flow.workspace_id { - tracing::error!( - "parallel Zombie flow detected: {} in workspace {}. Last ping was: {:?}.", - flow.parent_flow_id, - parent_flow_workspace_id, - flow.last_ping - ); - cancel_zombie_flow_job(db, flow.parent_flow_id, &parent_flow_workspace_id, - format!("Flow {} cancelled as one of the parallel branch {} was unable to make the last transition ", flow.parent_flow_id, flow.job_id)) - .await?; - } else { - tracing::info!("releasing lock for parallel flow: {}", flow.parent_flow_id); - } - } - Ok(()) -} - -async fn cancel_zombie_flow_job( - db: &Pool, - id: Uuid, - workspace_id: &str, - message: String, -) -> Result<(), error::Error> { - let mut tx = db.begin().await?; - tracing::error!( - "zombie flow detected: {} in workspace {}. Cancelling it.", - id, - workspace_id - ); - (tx, _) = cancel_job( - "monitor", - Some(message), - id, - workspace_id, - tx, - db, - true, - false, - ) - .await?; - tx.commit().await?; - Ok(()) -} - pub async fn reload_hub_base_url_setting(db: &DB, server_mode: bool) -> error::Result<()> { let hub_base_url = load_value_from_global_settings(db, HUB_BASE_URL_SETTING).await?; diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 7573e139ee..005f0d700f 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1,6 +1,6 @@ use serde::de::DeserializeOwned; use std::future::Future; -use std::{str::FromStr, sync::Arc}; +use std::{default::Default, str::FromStr, sync::Arc}; use windmill_api_client::types::{NewScript, ScriptLang as NewScriptLanguage}; #[cfg(feature = "enterprise")] @@ -13,7 +13,6 @@ use serde_json::json; use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres}; use tokio::sync::RwLock; -#[cfg(feature = "enterprise")] use tokio::time::{timeout, Duration}; use windmill_api_client::types::{CreateFlowBody, RawScript}; @@ -42,7 +41,7 @@ pub struct CompletedJob { pub parent_job: Option, pub created_by: String, pub created_at: chrono::DateTime, - pub started_at: chrono::DateTime, + pub started_at: Option>, pub duration_ms: i64, pub success: bool, pub script_path: Option, @@ -153,7 +152,7 @@ impl ApiServer { println!("closing api server"); let Self { tx, task, .. } = self; drop(tx); - task.await.unwrap() + task.await? } } @@ -213,7 +212,7 @@ mod suspend_resume { } fn flow() -> FlowValue { - serde_json::from_value(serde_json::json!({ + serde_json::from_value(json!({ "modules": [{ "id": "a", "value": { @@ -302,7 +301,10 @@ mod suspend_resume { let queue = listen_for_queue(&db).await; let db_ = db.clone(); - in_test_worker(&db, async move { + in_test_worker( + &db, + WorkerOptions { port, ..Default::default() }, + async move { let db = db_; wait_until_flow_suspends(flow, queue, &db).await; @@ -336,8 +338,9 @@ mod suspend_resume { .unwrap(); completed.find(&flow).await.unwrap(); - }, port) - .await; + } + ) + .await; server.close().await.unwrap(); @@ -409,7 +412,10 @@ mod suspend_resume { let queue = listen_for_queue(&db).await; let db_ = db.clone(); - in_test_worker(&db, async move { + in_test_worker( + &db, + WorkerOptions { port, ..Default::default() }, + async move { let db = db_; wait_until_flow_suspends(flow, queue, &db).await; @@ -439,8 +445,9 @@ mod suspend_resume { .unwrap(); completed.find(&flow).await.unwrap(); - }, port) - .await; + } + ) + .await; server.close().await.unwrap(); @@ -496,7 +503,7 @@ mod retry { results }); - return Self { addr, tx, task }; + Self { addr, tx, task } } async fn close(self) -> Vec { @@ -528,7 +535,7 @@ def main(last, port): } fn flow_forloop_retry() -> FlowValue { - serde_json::from_value(serde_json::json!({ + serde_json::from_value(json!({ "modules": [{ "id": "a", "value": { @@ -640,15 +647,15 @@ def main(last, port): assert_eq!(server.close().await, attempts); - assert!( + assert_eq!( result[1]["error"] .as_object() .unwrap() .get("message") .unwrap() .as_str() - .unwrap() - == "read" + .unwrap(), + "read" ); } @@ -775,7 +782,7 @@ async fn test_iteration(db: Pool) { let server = ApiServer::start(db.clone()).await; - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [{ "value": { "type": "forloopflow", @@ -806,7 +813,7 @@ async fn test_iteration(db: Pool) { .await .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([])); + assert_eq!(result, json!([])); /* Don't actually test that this does 257 jobs or that will take forever. */ let result = @@ -833,7 +840,7 @@ async fn test_iteration_parallel(db: Pool) { let server = ApiServer::start(db.clone()).await; - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [{ "value": { "type": "forloopflow", @@ -865,7 +872,7 @@ async fn test_iteration_parallel(db: Pool) { .await .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([])); + assert_eq!(result, json!([])); /* Don't actually test that this does 257 jobs or that will take forever. */ let job = @@ -946,7 +953,12 @@ impl RunJob { async fn run_until_complete(self, db: &Pool, port: u16) -> CompletedJob { let uuid = self.push(db).await; let listener = listen_for_completed_jobs(db).await; - in_test_worker(db, listener.find(&uuid), port).await; + in_test_worker( + db, + WorkerOptions { port, ..Default::default() }, + listener.find(&uuid), + ) + .await; let r = completed_job(uuid, db).await; r } @@ -961,7 +973,12 @@ impl RunJob { let uuid = self.push(db).await; let listener = listen_for_completed_jobs(db).await; test(uuid).await; - in_test_worker(db, listener.find(&uuid), port).await; + in_test_worker( + db, + WorkerOptions { port, ..Default::default() }, + listener.find(&uuid), + ) + .await; let r = completed_job(uuid, db).await; r } @@ -975,17 +992,34 @@ async fn run_job_in_new_worker_until_complete( RunJob::from(job).run_until_complete(db, port).await } +#[derive(Default)] +enum MonitorZombiesOption { + #[default] + Disabled, + Enabled { + restart_zombies: bool, + timeout_seconds: &'static str, + flow_timeout_seconds: &'static str, + }, +} + +#[derive(Default)] +struct WorkerOptions { + port: u16, + monitor_zombies: MonitorZombiesOption, +} + /// Start a worker with a timeout and run a future, until the worker quits or we time out. /// /// Cleans up the worker before resolving. -async fn in_test_worker( +async fn in_test_worker( db: &Pool, + options: WorkerOptions, inner: Fut, - port: u16, -) -> ::Output { +) -> ::Output { set_jwt_secret().await; - let (quit, worker) = spawn_test_worker(db, port); - let worker = tokio::time::timeout(std::time::Duration::from_secs(60), worker); + let (quit, worker) = spawn_test_worker(db, options); + let worker = timeout(Duration::from_secs(60), worker); tokio::pin!(worker); let res = tokio::select! { @@ -1010,7 +1044,7 @@ async fn in_test_worker( fn spawn_test_worker( db: &Pool, - port: u16, + options: WorkerOptions, ) -> ( tokio::sync::broadcast::Sender<()>, tokio::task::JoinHandle<()>, @@ -1032,8 +1066,9 @@ fn spawn_test_worker( let ip: &str = Default::default(); let tx2 = tx.clone(); - let future = async move { - let base_internal_url = format!("http://localhost:{}", port); + let db2 = db.clone(); + let worker_fut = async move { + let base_internal_url = format!("http://localhost:{}", options.port); { let mut wc = WORKER_CONFIG.write().await; (*wc).worker_tags = windmill_common::worker::DEFAULT_TAGS.clone(); @@ -1045,7 +1080,7 @@ fn spawn_test_worker( windmill_common::worker::make_pull_query(&wc).await; } windmill_worker::run_worker( - &db, + &db2, worker_instance, worker_name, 1, @@ -1059,6 +1094,35 @@ fn spawn_test_worker( .await }; + let MonitorZombiesOption::Enabled { restart_zombies, timeout_seconds, flow_timeout_seconds } = + options.monitor_zombies + else { + return (tx, tokio::task::spawn(worker_fut)); + }; + + let mut rx = tx.subscribe(); + let monitor_fut = async move { + let base_internal_url = format!("http://localhost:{}", options.port); + loop { + if rx.try_recv().is_ok() { + break; + } + windmill_worker::monitor::monitor_once( + &db, + &base_internal_url, + worker_instance, + Some(timeout_seconds), + Some(restart_zombies), + Some(flow_timeout_seconds), + ) + .await; + tokio::time::sleep(Duration::from_millis(10)).await; + } + }; + + let future = async move { + tokio::select! { biased; _ = worker_fut => (), _ = monitor_fut => () } + }; (tx, tokio::task::spawn(future)) } @@ -1101,10 +1165,10 @@ async fn completed_job(uuid: Uuid, db: &Pool) -> CompletedJob { } #[axum::async_trait(?Send)] -trait StreamFind: futures::Stream + Unpin + Sized { +trait StreamFind: Stream + Unpin + Sized { async fn find(self, item: &Self::Item) -> Option where - for<'l> &'l Self::Item: std::cmp::PartialEq, + for<'l> &'l Self::Item: PartialEq, { use futures::{future::ready, StreamExt}; @@ -1112,7 +1176,7 @@ trait StreamFind: futures::Stream + Unpin + Sized { } } -impl StreamFind for T {} +impl StreamFind for T {} #[sqlx::test(fixtures("base"))] async fn test_deno_flow(db: Pool) { @@ -1228,7 +1292,7 @@ async fn test_deno_flow(db: Pool) { let job = run_job_in_new_worker_until_complete(&db, job.clone(), port).await; // println!("job: {:#?}", job.flow_status); let result = job.json_result().unwrap(); - assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {}", i); + assert_eq!(result, json!([2, 4, 6]), "iteration: {}", i); } } @@ -1238,7 +1302,7 @@ async fn test_identity(db: Pool) { let server = ApiServer::start(db.clone()).await; - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [{ "value": { "type": "rawscript", @@ -1266,7 +1330,7 @@ async fn test_identity(db: Pool) { .await .json_result() .unwrap(); - assert_eq!(result, serde_json::json!(42)); + assert_eq!(result, json!(42)); } #[sqlx::test(fixtures("base"))] @@ -1488,7 +1552,7 @@ async fn test_deno_flow_same_worker(db: Pool) { .unwrap(); assert_eq!( result, - serde_json::json!("false 1,true 1,false 1,true 2,false 1,true 3,false 1,true 3") + json!("false 1,true 1,false 1,true 2,false 1,true 3,false 1,true 3") ); } @@ -1541,7 +1605,7 @@ async fn test_flow_result_by_id(db: Pool) { .await .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([[42]])); + assert_eq!(result, json!([[42]])); } #[sqlx::test(fixtures("base"))] @@ -1551,7 +1615,7 @@ async fn test_stop_after_if(db: Pool) { // let port = server.addr.port(); let port = 123; - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [ { "id": "a", @@ -1604,7 +1668,7 @@ async fn test_stop_after_if_nested(db: Pool) { // let port = server.addr.port(); let port = 123; - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [ { "id": "a", @@ -1664,7 +1728,7 @@ async fn test_python_flow(db: Pool) { let numbers = "def main(): return [1, 2, 3]"; let doubles = "def main(n): return n * 2"; - let flow: FlowValue = serde_json::from_value(serde_json::json!( { + let flow: FlowValue = serde_json::from_value(json!( { "modules": [ { "value": { @@ -1708,7 +1772,7 @@ async fn test_python_flow(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {i}"); + assert_eq!(result, json!([2, 4, 6]), "iteration: {i}"); } } @@ -1718,7 +1782,7 @@ async fn test_python_flow_2(db: Pool) { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [ { "value": { @@ -1743,7 +1807,7 @@ async fn test_python_flow_2(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!("Hello"), "iteration: {i}"); + assert_eq!(result, json!("Hello"), "iteration: {i}"); } } @@ -1783,7 +1847,7 @@ func main(derp string) (string, error) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!("hello world")); + assert_eq!(result, json!("hello world")); } #[sqlx::test(fixtures("base"))] @@ -1818,7 +1882,7 @@ fn main(world: String) -> Result { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!("Hello Hyrule!")); + assert_eq!(result, json!("Hello Hyrule!")); } // #[sqlx::test(fixtures("base"))] @@ -1923,7 +1987,7 @@ def main(): .json_result() .unwrap(); - assert_eq!(result, serde_json::json!("hello world")); + assert_eq!(result, json!("hello world")); } #[sqlx::test(fixtures("base"))] @@ -1959,7 +2023,7 @@ def main(): .json_result() .unwrap(); - assert_eq!(result, serde_json::json!(3)); + assert_eq!(result, json!(3)); } #[sqlx::test(fixtures("base"))] @@ -1994,7 +2058,7 @@ def main(): .json_result() .unwrap(); - assert_eq!(result, serde_json::json!("test-workspace")); + assert_eq!(result, json!("test-workspace")); } #[sqlx::test(fixtures("base"))] @@ -2028,7 +2092,7 @@ export async function main(a: Date) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!("object")); + assert_eq!(result, json!("object")); } #[sqlx::test(fixtures("base"))] @@ -2062,7 +2126,7 @@ export async function main(a: Date) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!("object")); + assert_eq!(result, json!("object")); } #[sqlx::test(fixtures("base"))] @@ -2097,7 +2161,7 @@ def main(a: datetime, b: bytes): .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([true, true])); + assert_eq!(result, json!([true, true])); } #[sqlx::test(fixtures("base"))] @@ -2106,7 +2170,7 @@ async fn test_empty_loop_1(db: Pool) { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [ { "id": "a", @@ -2153,7 +2217,7 @@ async fn test_empty_loop_1(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!(0)); + assert_eq!(result, json!(0)); } #[sqlx::test(fixtures("base"))] @@ -2162,7 +2226,7 @@ async fn test_invalid_first_step(db: Pool) { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [ { "value": { @@ -2200,7 +2264,7 @@ async fn test_empty_loop_2(db: Pool) { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [ { "value": { @@ -2233,7 +2297,7 @@ async fn test_empty_loop_2(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([])); + assert_eq!(result, json!([])); } #[sqlx::test(fixtures("base"))] @@ -2241,7 +2305,7 @@ async fn test_step_after_loop(db: Pool) { initialize_tracing().await; let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [ { "id": "a", @@ -2288,7 +2352,7 @@ async fn test_step_after_loop(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!(9)); + assert_eq!(result, json!(9)); } fn module_add_item_to_list(i: i32, id: &str) -> serde_json::Value { @@ -2356,7 +2420,7 @@ async fn test_branchone_simple(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([1, 2])); + assert_eq!(result, json!([1, 2])); } #[sqlx::test(fixtures("base"))] @@ -2392,7 +2456,7 @@ async fn test_branchone_with_cond(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([1, 3])); + assert_eq!(result, json!([1, 3])); } #[sqlx::test(fixtures("base"))] @@ -2430,7 +2494,7 @@ async fn test_branchall_sequential(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([[1, 2], [1, 3]])); + assert_eq!(result, json!([[1, 2], [1, 3]])); } #[sqlx::test(fixtures("base"))] @@ -2467,7 +2531,7 @@ async fn test_branchall_simple(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([[1, 2], [1, 3]])); + assert_eq!(result, json!([[1, 2], [1, 3]])); } #[derive(Deserialize)] @@ -2615,7 +2679,7 @@ async fn test_branchone_nested(db: Pool) { .json_result() .unwrap(); - assert_eq!(result, serde_json::json!([1, 2, 3])); + assert_eq!(result, json!([1, 2, 3])); } #[sqlx::test(fixtures("base"))] @@ -2675,7 +2739,7 @@ async fn test_branchall_nested(db: Pool) { println!("{:#?}", result); assert_eq!( result, - serde_json::json!([[[[1, 2], [1, 3], 4], [[1, 2], [1, 3], 5]], [1, 6]]) + json!([[[[1, 2], [1, 3], 4], [[1, 2], [1, 3], 5]], [1, 6]]) ); } @@ -2685,7 +2749,7 @@ async fn test_failure_module(db: Pool) { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow: FlowValue = serde_json::from_value(serde_json::json!({ + let flow: FlowValue = serde_json::from_value(json!({ "modules": [{ "id": "a", "value": { @@ -2798,7 +2862,7 @@ async fn test_flow_lock_all(db: Pool) { let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); - let flow: windmill_api_client::types::OpenFlow = serde_json::from_value(serde_json::json!({ + let flow: windmill_api_client::types::OpenFlow = serde_json::from_value(json!({ "summary": "", "description": "", "value": { @@ -2898,7 +2962,12 @@ async fn test_flow_lock_all(db: Pool) { .unwrap(); let mut str = listen_for_completed_jobs(&db).await; let listen_first_job = str.next(); - in_test_worker(&db, listen_first_job, port).await; + in_test_worker( + &db, + WorkerOptions { port, ..Default::default() }, + listen_first_job, + ) + .await; let modules = client .get_flow_by_path("test-workspace", "g/all/flow_lock_all", None) @@ -3193,6 +3262,7 @@ async fn test_script_schedule_handlers(db: Pool) { let db2 = db.clone(); in_test_worker( &db, + WorkerOptions { port, ..Default::default() }, async move { str.next().await; // completed error job @@ -3220,7 +3290,6 @@ async fn test_script_schedule_handlers(db: Pool) { ); } }, - port, ) .await; @@ -3264,6 +3333,7 @@ async fn test_script_schedule_handlers(db: Pool) { let db2 = db.clone(); in_test_worker( &db, + WorkerOptions { port, ..Default::default() }, async move { str.next().await; // completed working job let uuid = timeout(Duration::from_millis(5000), str.next()).await; // recovery handler @@ -3287,7 +3357,6 @@ async fn test_script_schedule_handlers(db: Pool) { panic!("a script was run after main job execution but was not schedule recovery handler"); } }, - port, ) .await; } @@ -3346,6 +3415,7 @@ async fn test_flow_schedule_handlers(db: Pool) { let db2 = db.clone(); in_test_worker( &db, + WorkerOptions { port, ..Default::default() }, async move { str.next().await; // completed error step str.next().await; // completed error flow @@ -3374,7 +3444,6 @@ async fn test_flow_schedule_handlers(db: Pool) { ); } }, - port, ) .await; @@ -3418,6 +3487,7 @@ async fn test_flow_schedule_handlers(db: Pool) { let db2 = db.clone(); in_test_worker( &db, + WorkerOptions { port, ..Default::default() }, async move { str.next().await; // completed working step str.next().await; // completed working flow @@ -3442,7 +3512,6 @@ async fn test_flow_schedule_handlers(db: Pool) { panic!("a script was run after main job execution but was not schedule recovery handler"); } }, - port, ) .await; } @@ -3502,6 +3571,7 @@ async fn run_deployed_relative_imports( let db2 = db.clone(); in_test_worker( &db, + WorkerOptions { port, ..Default::default() }, async move { completed.next().await; // deployed script @@ -3534,7 +3604,7 @@ async fn run_deployed_relative_imports( assert_eq!( result, - serde_json::json!([ + json!([ "f/system/same_folder_script", "f/system/same_folder_script", "f/system_relative/different_folder_script", @@ -3542,7 +3612,6 @@ async fn run_deployed_relative_imports( ]) ); }, - port, ) .await; } @@ -3560,6 +3629,7 @@ async fn run_preview_relative_imports( let db2 = db.clone(); in_test_worker( &db, + WorkerOptions { port, ..Default::default() }, async move { let job = RunJob::from(JobPayload::Code(RawCode { hash: None, @@ -3582,7 +3652,7 @@ async fn run_preview_relative_imports( assert_eq!( result, - serde_json::json!([ + json!([ "f/system/same_folder_script", "f/system/same_folder_script", "f/system_relative/different_folder_script", @@ -3590,7 +3660,6 @@ async fn run_preview_relative_imports( ]) ); }, - port, ) .await; } @@ -3864,6 +3933,7 @@ async fn test_workflow_as_code(db: Pool) { let db = &db; in_test_worker( &db, + WorkerOptions { port, ..Default::default() }, async move { let job = RunJob::from(JobPayload::Code(RawCode { language: ScriptLang::Python3, @@ -3912,7 +3982,6 @@ async fn test_workflow_as_code(db: Pool) { ); } }, - port, ) .await; } diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 9b77ca840f..88343cfdcb 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -25,6 +25,7 @@ mod handle_child; mod job_logger; mod job_logger_ee; mod js_eval; +pub mod monitor; #[cfg(feature = "mysql")] mod mysql_executor; #[cfg(feature = "oracledb")] diff --git a/backend/windmill-worker/src/monitor.rs b/backend/windmill-worker/src/monitor.rs new file mode 100644 index 0000000000..5010d28c12 --- /dev/null +++ b/backend/windmill-worker/src/monitor.rs @@ -0,0 +1,450 @@ +use std::{ + env, + sync::{atomic::AtomicU16, Arc}, +}; + +use sqlx::{PgPool, Pool, Postgres}; +use tokio::sync::mpsc; +use uuid::Uuid; + +#[cfg(feature = "prometheus")] +use windmill_common::METRICS_ENABLED; +use windmill_common::{ + error, + flow_status::{FlowStatus, FlowStatusModule}, + jobs::QueuedJob, + utils::{now_from_db, report_critical_error}, + BASE_URL, +}; +use windmill_queue::cancel_job; + +use crate::{ + create_token_for_owner, handle_job_error, AuthedClient, SameWorkerPayload, SameWorkerSender, + SendResult, SCRIPT_TOKEN_EXPIRY, +}; + +lazy_static::lazy_static! { + static ref ZOMBIE_JOB_TIMEOUT: String = env::var("ZOMBIE_JOB_TIMEOUT") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or_else(|| "60".to_string()); + + static ref FLOW_ZOMBIE_TRANSITION_TIMEOUT: String = env::var("FLOW_ZOMBIE_TRANSITION_TIMEOUT") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or_else(|| "60".to_string()); + + pub static ref RESTART_ZOMBIE_JOBS: bool = env::var("RESTART_ZOMBIE_JOBS") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(true); +} + +#[cfg(feature = "prometheus")] +lazy_static::lazy_static! { + static ref QUEUE_ZOMBIE_RESTART_COUNT: prometheus::IntCounter = + prometheus::register_int_counter!( + "queue_zombie_restart_count", + "Total number of jobs restarted due to ping timeout." + ) + .unwrap(); + + static ref QUEUE_ZOMBIE_DELETE_COUNT: prometheus::IntCounter = + prometheus::register_int_counter!( + "queue_zombie_delete_count", + "Total number of jobs deleted due to their ping timing out in an unrecoverable state." + ) + .unwrap(); + + static ref QUEUE_COUNT: prometheus::IntGaugeVec = + prometheus::register_int_gauge_vec!( + "queue_count", + "Number of jobs in the queue", + &["tag"] + ).unwrap(); +} + +pub async fn monitor_once( + db: &PgPool, + base_internal_url: &str, + worker_name: &str, + timeout_seconds: Option<&str>, + restart_zombies: Option, + transition_timeout_seconds: Option<&str>, +) { + let timeout_seconds = timeout_seconds.unwrap_or_else(|| &*ZOMBIE_JOB_TIMEOUT); + let restart_zombies = restart_zombies.unwrap_or(*RESTART_ZOMBIE_JOBS); + handle_zombie_jobs( + &db, + base_internal_url, + worker_name, + timeout_seconds, + restart_zombies, + ) + .await; + let transition_timeout_seconds = + transition_timeout_seconds.unwrap_or_else(|| &*FLOW_ZOMBIE_TRANSITION_TIMEOUT); + let _ = handle_zombie_flows(&db, transition_timeout_seconds) + .await + .inspect_err(|err| tracing::error!("Error handling zombie flows: {:#}", err)); +} + +async fn handle_zombie_jobs( + db: &Pool, + base_internal_url: &str, + worker_name: &str, + timeout_seconds: &str, + restart_zombies: bool, +) { + if restart_zombies { + let restarted = sqlx::query!( + "WITH zombie_jobs AS ( + UPDATE v2_job_queue q SET running = false, started_at = null + FROM v2_job j, v2_job_runtime r + WHERE j.id = q.id AND j.id = r.id + AND ping < now() - ($1 || ' seconds')::interval + AND running = true + AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') + AND same_worker = false + RETURNING q.id, q.workspace_id, ping + ), + update_concurrency AS ( + UPDATE concurrency_counter cc + SET job_uuids = job_uuids - zj.id::text + FROM zombie_jobs zj + INNER JOIN concurrency_key ck ON ck.job_id = zj.id + WHERE cc.concurrency_id = ck.key + ) + SELECT id, workspace_id, ping FROM zombie_jobs", + timeout_seconds, + ) + .fetch_all(db) + .await + .ok() + .unwrap_or_else(|| vec![]); + + #[cfg(feature = "prometheus")] + if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { + QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); + } + + let base_url = BASE_URL.read().await.clone(); + for r in restarted { + let last_ping = r + .ping + .map(|x| format!("last ping at {x}")) + .unwrap_or_else(|| "no last ping".into()); + let url = format!("{}/run/{}?workspace={}", base_url, r.id, r.workspace_id,); + let error_message = format!( + "Zombie job {} on {} ({}) detected, restarting it, {}", + r.id, r.workspace_id, url, last_ping + ); + + let _ = sqlx::query!(" + INSERT INTO job_logs (job_id, logs) + VALUES ($1, 'Restarted job after not receiving job''s ping for too long the ' || now() || '\n\n') + ON CONFLICT (job_id) DO UPDATE SET logs = job_logs.logs || '\n' || EXCLUDED.logs + WHERE job_logs.job_id = $1", + r.id + ) + .execute(db) + .await; + tracing::error!(error_message); + report_critical_error(error_message, db.clone(), Some(&r.workspace_id), None).await; + } + } + + let same_worker_timeout_jobs = { + let long_same_worker_jobs = sqlx::query!( + "SELECT worker, array_agg(v2_job_queue.id) as ids + FROM v2_job_queue + JOIN v2_job USING (id) + LEFT JOIN v2_job_runtime USING (id) + WHERE v2_job_queue.created_at < now() - ('60 seconds')::interval + AND running = true AND ping IS NULL + AND same_worker = true + AND worker IS NOT NULL + GROUP BY worker", + ) + .fetch_all(db) + .await + .ok() + .unwrap_or_else(|| vec![]); + + let worker_ids = long_same_worker_jobs + .iter() + .map(|x| x.worker.clone().unwrap_or_default()) + .collect::>(); + + let long_dead_workers: std::collections::HashSet = sqlx::query_scalar!( + "WITH worker_ids AS (SELECT unnest($1::text[]) as worker) + SELECT worker_ids.worker FROM worker_ids + LEFT JOIN worker_ping ON worker_ids.worker = worker_ping.worker + WHERE worker_ping.worker IS NULL OR worker_ping.ping_at < now() - ('60 seconds')::interval", + &worker_ids[..] + ) + .fetch_all(db) + .await + .ok() + .unwrap_or_else(|| vec![]) + .into_iter() + .filter_map(|x| x) + .collect(); + + let mut timeouts: Vec = vec![]; + for worker in long_same_worker_jobs { + if worker.worker.is_some() && long_dead_workers.contains(&worker.worker.unwrap()) { + if let Some(ids) = worker.ids { + timeouts.extend(ids); + } + } + } + if !timeouts.is_empty() { + tracing::error!( + "Failing same worker zombie jobs: {:?}", + timeouts + .iter() + .map(|x| x.hyphenated().to_string()) + .collect::>() + .join(",") + ); + } + + let jobs = sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE id = ANY($1)") + .bind(&timeouts[..]) + .fetch_all(db) + .await + .map_err(|e| tracing::error!("Error fetching same worker jobs: {:?}", e)) + .unwrap_or_default(); + + jobs + }; + + let non_restartable_jobs = if restart_zombies { + vec![] + } else { + sqlx::query_as::<_, QueuedJob>( + "SELECT * FROM v2_as_queue + WHERE last_ping < now() - ($1 || ' seconds')::interval + AND running = true + AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') + AND same_worker = false", + ) + .bind(timeout_seconds) + .fetch_all(db) + .await + .ok() + .unwrap_or_else(|| vec![]) + }; + + let timeouts = non_restartable_jobs + .into_iter() + .chain(same_worker_timeout_jobs) + .collect::>(); + + #[cfg(feature = "prometheus")] + if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { + QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _); + } + + for job in timeouts { + // since the job is unrecoverable, the same worker queue should never be sent anything + let (same_worker_tx_never_used, _same_worker_rx_never_used) = + mpsc::channel::(1); + let same_worker_tx_never_used = + SameWorkerSender(same_worker_tx_never_used, Arc::new(AtomicU16::new(0))); + let (send_result_never_used, _send_result_rx_never_used) = mpsc::channel::(1); + + let label = if job.permissioned_as != format!("u/{}", job.created_by) + && job.permissioned_as != job.created_by + { + format!("ephemeral-script-end-user-{}", job.created_by) + } else { + "ephemeral-script".to_string() + }; + let token = create_token_for_owner( + &db, + &job.workspace_id, + &job.permissioned_as, + &label, + *SCRIPT_TOKEN_EXPIRY, + &job.email, + &job.id, + ) + .await + .expect("could not create job token"); + + let client = AuthedClient { + base_internal_url: base_internal_url.to_string(), + token, + workspace: job.workspace_id.to_string(), + force_client: None, + }; + + let last_ping = job.last_ping.clone(); + let _ = handle_job_error( + db, + &client, + &job, + 0, + None, + error::Error::ExecutionErr(format!( + "Job timed out after no ping from job since {} (timeout: {}, same_worker: {})", + last_ping + .map(|x| x.to_string()) + .unwrap_or_else(|| "no ping".to_string()), + timeout_seconds, + job.same_worker + )), + true, + same_worker_tx_never_used, + "", + worker_name, + send_result_never_used, + #[cfg(feature = "benchmark")] + &mut crate::bench::BenchmarkIter::new(), + ) + .await; + } +} + +async fn handle_zombie_flows(db: &PgPool, transition_timeout_seconds: &str) -> error::Result<()> { + let flows = sqlx::query!( + r#" + SELECT + id AS "id!", workspace_id AS "workspace_id!", parent_job, is_flow_step, + flow_status AS "flow_status: Box", last_ping, same_worker + FROM v2_as_queue + WHERE running = true AND suspend = 0 AND suspend_until IS null AND scheduled_for <= now() + AND (job_kind = 'flow' OR job_kind = 'flowpreview' OR job_kind = 'flownode') + AND last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval + AND canceled = false + "#, + transition_timeout_seconds + ) + .fetch_all(db) + .await?; + + for flow in flows { + let status = flow + .flow_status + .as_deref() + .and_then(|x| serde_json::from_str::(x).ok()); + if !flow.same_worker.unwrap_or(false) + && status.is_some_and(|s| { + s.modules + .get(0) + .is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. })) + }) + { + let error_message = format!( + "Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.", + flow.id, flow.workspace_id + ); + tracing::error!(error_message); + report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None).await; + // if the flow hasn't started and is a zombie, we can simply restart it + let mut tx = db.begin().await?; + + let concurrency_key = + sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", flow.id) + .fetch_optional(&mut *tx) + .await?; + + if let Some(key) = concurrency_key { + sqlx::query!( + "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", + key, + flow.id.hyphenated().to_string() + ) + .execute(&mut *tx) + .await?; + } + + sqlx::query!( + "UPDATE v2_job_queue SET running = false, started_at = null + WHERE id = $1 AND canceled_by IS NULL", + flow.id + ) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + } else { + let id = flow.id.clone(); + let last_ping = flow.last_ping.clone(); + let now = now_from_db(db).await?; + let reason = format!( + "{} was hanging in between 2 steps. Last ping: {last_ping:?} (now: {now})", + if flow.is_flow_step.unwrap_or(false) && flow.parent_job.is_some() { + format!("Flow was cancelled because subflow {id}") + } else { + format!("Flow {id} was cancelled because it") + } + ); + report_critical_error(reason.clone(), db.clone(), Some(&flow.workspace_id), None).await; + cancel_zombie_flow_job(db, flow.id, &flow.workspace_id, reason).await?; + } + } + + let flows2 = sqlx::query!( + r#" + DELETE + FROM parallel_monitor_lock + WHERE last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval + RETURNING parent_flow_id, job_id, last_ping, (SELECT workspace_id FROM v2_job_queue q + WHERE q.id = parent_flow_id AND q.running = true AND q.canceled_by IS NULL + ) AS workspace_id + "#, + transition_timeout_seconds + ) + .fetch_all(db) + .await?; + + for flow in flows2 { + if let Some(parent_flow_workspace_id) = flow.workspace_id { + tracing::error!( + "parallel Zombie flow detected: {} in workspace {}. Last ping was: {:?}.", + flow.parent_flow_id, + parent_flow_workspace_id, + flow.last_ping + ); + cancel_zombie_flow_job(db, flow.parent_flow_id, &parent_flow_workspace_id, format!( + "Flow {} cancelled as one of the parallel branch {} was unable to make the last transition ", + flow.parent_flow_id, + flow.job_id + )) + .await?; + } else { + tracing::info!("releasing lock for parallel flow: {}", flow.parent_flow_id); + } + } + Ok(()) +} + +async fn cancel_zombie_flow_job( + db: &Pool, + id: Uuid, + workspace_id: &str, + message: String, +) -> Result<(), error::Error> { + let mut tx = db.begin().await?; + tracing::error!( + "zombie flow detected: {} in workspace {}. Cancelling it.", + id, + workspace_id + ); + (tx, _) = cancel_job( + "monitor", + Some(message), + id, + workspace_id, + tx, + db, + true, + false, + ) + .await?; + tx.commit().await?; + Ok(()) +}