From 012726ea1e65a914e9f32d129bbb18d87b0ebe00 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 4 Apr 2026 13:13:31 +0000 Subject: [PATCH] fix: detect zombie flows with NULL pings and cap suspend timeout at retention period Flows waiting for a suspended child (e.g. approval step) have running=true but ping=NULL by design. The zombie flow handler required ping IS NOT NULL, so these flows were invisible to detection and could stay in the queue indefinitely, blocking retention cleanup of their completed children. Two fixes: 1. monitor.rs: Add detection for root flows with NULL pings older than the retention period. These are cancelled as zombies with a critical error. 2. worker_flow.rs: Cap suspend_until at the retention period to prevent flows from being parked longer than retention allows. Co-Authored-By: Claude Opus 4.6 (1M context) --- ...8d48ce3ede5fdd7f8d64840f8c62cf89f99e7.json | 64 +++++++++++++++++++ ...b7c0417161bf69e596e2e34fc8d9384824776.json | 64 +++++++++++++++++++ ...ca43f00456a27fda32583846497116f92f96c.json | 38 +++++++++++ ...3a94fca9dd95cc3b26b33611eab9d2a261141.json | 22 +++++++ backend/src/monitor.rs | 43 +++++++++++++ backend/windmill-worker/src/worker_flow.rs | 16 ++++- 6 files changed, 244 insertions(+), 3 deletions(-) create mode 100644 backend/.sqlx/query-19478c2a3c05b2ed17b110f8ff58d48ce3ede5fdd7f8d64840f8c62cf89f99e7.json create mode 100644 backend/.sqlx/query-464895f217536bf713ec381b6c3b7c0417161bf69e596e2e34fc8d9384824776.json create mode 100644 backend/.sqlx/query-cb3862634f18160207ee2621ddfca43f00456a27fda32583846497116f92f96c.json create mode 100644 backend/.sqlx/query-d1dcc7fc8a1e1bc4dad263ec5163a94fca9dd95cc3b26b33611eab9d2a261141.json diff --git a/backend/.sqlx/query-19478c2a3c05b2ed17b110f8ff58d48ce3ede5fdd7f8d64840f8c62cf89f99e7.json b/backend/.sqlx/query-19478c2a3c05b2ed17b110f8ff58d48ce3ede5fdd7f8d64840f8c62cf89f99e7.json new file mode 100644 index 0000000000..d3a6b60613 --- /dev/null +++ b/backend/.sqlx/query-19478c2a3c05b2ed17b110f8ff58d48ce3ede5fdd7f8d64840f8c62cf89f99e7.json @@ -0,0 +1,64 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n script_lang as \"language: _\",\n COUNT(*) as \"count!\",\n SUM(duration_ms)::BIGINT as \"total_duration!\"\n FROM\n v2_job_completed\n JOIN\n v2_job\n ON\n v2_job.id = v2_job_completed.id\n WHERE\n created_at >= CURRENT_DATE - INTERVAL '1 day'\n AND created_at < CURRENT_DATE\n GROUP BY\n script_lang\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "language: _", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby", + "rlang" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "count!", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "total_duration!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + true, + null, + null + ] + }, + "hash": "19478c2a3c05b2ed17b110f8ff58d48ce3ede5fdd7f8d64840f8c62cf89f99e7" +} diff --git a/backend/.sqlx/query-464895f217536bf713ec381b6c3b7c0417161bf69e596e2e34fc8d9384824776.json b/backend/.sqlx/query-464895f217536bf713ec381b6c3b7c0417161bf69e596e2e34fc8d9384824776.json new file mode 100644 index 0000000000..7b61cc3cd1 --- /dev/null +++ b/backend/.sqlx/query-464895f217536bf713ec381b6c3b7c0417161bf69e596e2e34fc8d9384824776.json @@ -0,0 +1,64 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n script_lang as \"language: _\",\n COUNT(*) as \"count!\",\n SUM(duration_ms)::BIGINT as \"total_duration!\"\n FROM\n v2_job_completed\n JOIN\n v2_job\n ON\n v2_job.id = v2_job_completed.id\n WHERE\n created_at > NOW() - INTERVAL '48 hours'\n GROUP BY\n script_lang\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "language: _", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby", + "rlang" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "count!", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "total_duration!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + true, + null, + null + ] + }, + "hash": "464895f217536bf713ec381b6c3b7c0417161bf69e596e2e34fc8d9384824776" +} diff --git a/backend/.sqlx/query-cb3862634f18160207ee2621ddfca43f00456a27fda32583846497116f92f96c.json b/backend/.sqlx/query-cb3862634f18160207ee2621ddfca43f00456a27fda32583846497116f92f96c.json new file mode 100644 index 0000000000..8ccf623719 --- /dev/null +++ b/backend/.sqlx/query-cb3862634f18160207ee2621ddfca43f00456a27fda32583846497116f92f96c.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH active_users as (SELECT distinct username as email FROM audit WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n active_authors as (SELECT distinct email FROM usr WHERE usr.operator IS false AND email IN (SELECT email FROM active_users)),\n active_authors_agg as (SELECT array_agg(email) as authors FROM active_authors),\n active_ops_agg as (SELECT array_agg(email) as operators from active_users WHERE email NOT IN (SELECT email FROM active_authors))\n SELECT active_authors_agg.authors, active_ops_agg.operators, array_length(active_authors_agg.authors, 1) as author_count, array_length(active_ops_agg.operators, 1) as operator_count FROM active_authors_agg, active_ops_agg", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "authors", + "type_info": "VarcharArray" + }, + { + "ordinal": 1, + "name": "operators", + "type_info": "VarcharArray" + }, + { + "ordinal": 2, + "name": "author_count", + "type_info": "Int4" + }, + { + "ordinal": 3, + "name": "operator_count", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null, + null, + null, + null + ] + }, + "hash": "cb3862634f18160207ee2621ddfca43f00456a27fda32583846497116f92f96c" +} diff --git a/backend/.sqlx/query-d1dcc7fc8a1e1bc4dad263ec5163a94fca9dd95cc3b26b33611eab9d2a261141.json b/backend/.sqlx/query-d1dcc7fc8a1e1bc4dad263ec5163a94fca9dd95cc3b26b33611eab9d2a261141.json new file mode 100644 index 0000000000..4a0592701c --- /dev/null +++ b/backend/.sqlx/query-d1dcc7fc8a1e1bc4dad263ec5163a94fca9dd95cc3b26b33611eab9d2a261141.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT args as \"args: sqlx::types::Json>\"\n FROM v2_job\n WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "args: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + true + ] + }, + "hash": "d1dcc7fc8a1e1bc4dad263ec5163a94fca9dd95cc3b26b33611eab9d2a261141" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 5e87f3d161..b2765fc00b 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -3305,6 +3305,49 @@ Please check your worker logs for more details and feel free to report it to the } } + // Detect flows with NULL pings that have been running longer than the retention period. + // This catches parent flows/flownodes waiting for a suspended child (e.g. approval step) + // where no worker is actively processing them. The normal zombie detection above requires + // ping IS NOT NULL so these flows would otherwise stay in the queue indefinitely. + let job_retention_secs = *JOB_RETENTION_SECS.read().await; + if job_retention_secs > 0 { + let stale_null_ping_flows = sqlx::query!( + r#" + SELECT j.id AS "id!", j.workspace_id AS "workspace_id!" + FROM v2_job_queue q + JOIN v2_job j USING (id) + LEFT JOIN v2_job_runtime r USING (id) + WHERE q.running = true + AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode') + AND r.ping IS NULL + AND j.parent_job IS NULL + AND j.created_at < NOW() - ($1::bigint::text || ' s')::interval + AND q.canceled_by IS NULL + "#, + job_retention_secs + ) + .fetch_all(db) + .await?; + + for flow in &stale_null_ping_flows { + let base_url = BASE_URL.read().await; + let reason = format!( + "Flow {} ({}/run/{}?workspace={}) has been running with no active worker (NULL ping) for longer than the retention period ({}s). \ + This typically happens when a sub-flow is suspended (e.g. approval step) and the parent flow's worker was interrupted before recording a ping.", + flow.id, base_url, flow.id, flow.workspace_id, job_retention_secs + ); + 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?; + } + + if !stale_null_ping_flows.is_empty() { + tracing::info!( + "Cancelled {} stale flows with NULL pings older than retention period", + stale_null_ping_flows.len() + ); + } + } + let flows2 = sqlx::query!( r#" DELETE diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index baadad65c4..f4c2bda757 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -2911,6 +2911,18 @@ async fn push_next_flow_job( FlowStatusModule::WaitingForPriorSteps { .. } ) && is_disapproved.is_none() { + let suspend_timeout_secs: u64 = + suspend.timeout.map(|t| t.into()).unwrap_or(30 * 60); + // Cap suspend timeout at the retention period to prevent flows from being + // parked indefinitely (which blocks retention cleanup of completed children). + let job_retention_secs = + *windmill_common::JOB_RETENTION_SECS.read().await; + let suspend_timeout_secs = if job_retention_secs > 0 { + suspend_timeout_secs.min(job_retention_secs as u64) + } else { + suspend_timeout_secs + }; + sqlx::query!( "WITH suspend AS ( UPDATE v2_job_queue SET suspend = $2, suspend_until = now() + $3 @@ -2927,9 +2939,7 @@ async fn push_next_flow_job( job: last }), (required_events - resume_messages.len() as u16) as i32, - Duration::from_secs( - suspend.timeout.map(|t| t.into()).unwrap_or_else(|| 30 * 60) - ) as Duration, + Duration::from_secs(suspend_timeout_secs) as Duration, flow_job.id, ) .execute(&mut *tx)