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) <noreply@anthropic.com>

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

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

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

backend/.sqlx/query-d1dcc7fc8a1e1bc4dad263ec5163a94fca9dd95cc3b26b33611eab9d2a261141.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT args as \"args: sqlx::types::Json<Box<RawValue>>\"\n                        FROM v2_job\n                        WHERE id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "args: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "d1dcc7fc8a1e1bc4dad263ec5163a94fca9dd95cc3b26b33611eab9d2a261141"
+}

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

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)