fix: add orphan adoption and atomic v2_job_completed deletion in job cleanup

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json

@@ -46,11 +46,11 @@
       ]
     },
     "nullable": [
-      false,
-      false,
-      false,
-      false,
-      false,
+      true,
+      true,
+      true,
+      true,
+      true,
       true,
       true
     ]

backend/.sqlx/query-086b2564495eb07d8ea362bbfa33c9571ab11fde8172aeea6b93ccfeca1c1418.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM v2_job_completed USING jobs_pending_deletion d\n         WHERE v2_job_completed.id = d.id AND d.marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "086b2564495eb07d8ea362bbfa33c9571ab11fde8172aeea6b93ccfeca1c1418"
+}

backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json

@@ -30,7 +30,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-0f7e01b613a94b29784aae6d7b17b23d6dcf2e5364852a5e85b3c41c417bace2.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json

@@ -122,7 +122,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-16d438374b03a9c515f4c2d638366f38ffe2f3a0958adea53e67757c6ac463ec.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json

@@ -40,7 +40,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json

@@ -34,7 +34,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }
@@ -67,7 +68,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-22e0e8a1aa48f8b21763452bd36fbe7db4887c4ac5295052c796bd78a7edc50b.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-23419adcd74c326d716527293eff518b42f4cdb33e034441015494bd26c172d2.json

@@ -40,7 +40,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-24aa22975e2bd9cd03bbfc289a14525d78eb8a4e7d88954994e1fdbb79247368.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM job_stats USING jobs_pending_deletion d\n         WHERE job_stats.job_id = d.id AND d.marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "24aa22975e2bd9cd03bbfc289a14525d78eb8a4e7d88954994e1fdbb79247368"
+}

backend/.sqlx/query-2b66cee1feee58537c949837e1b3ae8d2504cde2714b8bc40337d34e355ff3c5.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "DELETE FROM job_logs WHERE job_id = ANY($1) RETURNING log_file_index",
+  "query": "DELETE FROM job_logs USING jobs_pending_deletion d\n         WHERE job_logs.job_id = d.id AND d.marked_by = $1\n         RETURNING job_logs.log_file_index",
   "describe": {
     "columns": [
       {
@@ -11,12 +11,12 @@
     ],
     "parameters": {
       "Left": [
-        "UuidArray"
+        "Text"
       ]
     },
     "nullable": [
       true
     ]
   },
-  "hash": "288e99211bbd45a337fc9b79c43c5139ee535e23c8b3362c52eef49998349f15"
+  "hash": "2b66cee1feee58537c949837e1b3ae8d2504cde2714b8bc40337d34e355ff3c5"
 }

backend/.sqlx/query-2f2c8b6e3b75524f286a700b87a52697c28154b9ed551bcf9549275acd53625a.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO v2_job (id, tag, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email, kind, same_worker, visible_to_owner)\n                 VALUES ($1, $2, 'admins', now() - interval '30 minutes', $3, $3, 'bench@test.com', 'script', false, true)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Varchar",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "2f2c8b6e3b75524f286a700b87a52697c28154b9ed551bcf9549275acd53625a"
+}

backend/.sqlx/query-37937656035a487aa252bad3bdba28c129943e1c3ccd5bc3c8f652435d1a0021.json

@@ -1,20 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT COALESCE((SELECT value::bigint FROM global_settings WHERE name = 'retention_period_secs'), 2592000)",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "coalesce",
-        "type_info": "Int8"
-      }
-    ],
-    "parameters": {
-      "Left": []
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "37937656035a487aa252bad3bdba28c129943e1c3ccd5bc3c8f652435d1a0021"
-}

backend/.sqlx/query-3d71a98588d436f27ec6e597e856151e4bcddb638c3700f2f854e14856bc4d82.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM job_logs USING jobs_pending_deletion d\n             WHERE job_logs.job_id = d.id AND d.marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3d71a98588d436f27ec6e597e856151e4bcddb638c3700f2f854e14856bc4d82"
+}

backend/.sqlx/query-3ebf915f5a49f1de1003e790fec459ffec018847ba20dd30fdc7f8012d03bbd6.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT q.id FROM v2_job_queue q\n             JOIN v2_job j ON j.id = q.id\n             WHERE j.parent_job IS NULL\n               AND j.created_at <= now() - ($1::bigint::text || ' s')::interval",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "3ebf915f5a49f1de1003e790fec459ffec018847ba20dd30fdc7f8012d03bbd6"
+}

backend/.sqlx/query-408caf44233591446c21d7e180d91cec9ad768b06fc4cd816b3e066cbe71873c.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO jobs_pending_deletion (id, marked_by)\n         SELECT jc.id, $4 FROM v2_job_completed jc\n         JOIN v2_job j ON j.id = jc.id\n         WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval\n           AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($2)\n           AND j.tag = $3\n         ORDER BY jc.completed_at ASC\n         LIMIT 1000\n         ON CONFLICT DO NOTHING",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "UuidArray",
+        "Text",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "408caf44233591446c21d7e180d91cec9ad768b06fc4cd816b3e066cbe71873c"
+}

backend/.sqlx/query-42b4b73e9d60348e2d90fcade9dcad6d8995242dc20a4e14c1a8fae4fc6a9fd2.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-45024b932383199974616bba1fc2f7175cc6f2e02d9c565bb5159cae3e0b6835.json

@@ -30,7 +30,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-45997fcb4d9d62c7f7011966bf59bdb86e12ce1d0c8e925e738d2645121a5c1f.json

@@ -1,24 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM v2_job_completed\n         WHERE id IN (\n             SELECT jc.id FROM v2_job_completed jc\n             LEFT JOIN v2_job j ON j.id = jc.id\n             WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval\n               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)\n             ORDER BY jc.completed_at ASC\n             LIMIT $2\n             FOR UPDATE OF jc SKIP LOCKED\n         )\n         RETURNING id",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id",
-        "type_info": "Uuid"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Int8",
-        "Int8",
-        "UuidArray"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "45997fcb4d9d62c7f7011966bf59bdb86e12ce1d0c8e925e738d2645121a5c1f"
-}

backend/.sqlx/query-4a89f9f21876e7274ad8691ee2986918e22b12491d0977e320abaddcaf604c43.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM v2_job USING jobs_pending_deletion d\n             WHERE v2_job.id = d.id AND d.marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "4a89f9f21876e7274ad8691ee2986918e22b12491d0977e320abaddcaf604c43"
+}

backend/.sqlx/query-4d272cf4a77aab7007a5b35589e08532a1020cabaf5e22325a1e05f0491d785c.json

@@ -37,7 +37,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-4f547c0fd54f3bc57212ce87810e35adf640d44d607e62a1fb296e38ac3fdd36.json

@@ -32,7 +32,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }
@@ -70,7 +71,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-51e1d713dc57f57531dce518e564bd9440582ffd3e396a947f3c93f139e770e2.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM jobs_pending_deletion",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "51e1d713dc57f57531dce518e564bd9440582ffd3e396a947f3c93f139e770e2"
+}

backend/.sqlx/query-56740517a422c64d8e5b162cfde0fd2887738435313af441e4028ebe6efdf210.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COUNT(*) FROM jobs_pending_deletion",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "56740517a422c64d8e5b162cfde0fd2887738435313af441e4028ebe6efdf210"
+}

backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json

@@ -245,7 +245,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-66a0e51cf149ba532463e29dd361a803e1bced2f8e1a12f8933b7598ee85a147.json

@@ -35,7 +35,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-6967f6f7fef9a3d1364fd549974a8b77bd9bac9fbc3379ef9f5bd28841fdf66b.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO job_stats (job_id, workspace_id) VALUES ($1, 'admins')\n                 ON CONFLICT DO NOTHING",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6967f6f7fef9a3d1364fd549974a8b77bd9bac9fbc3379ef9f5bd28841fdf66b"
+}

backend/.sqlx/query-7065f23d04e26831664048f2cfc4f412c57af931f80621aee5012e9cb3535626.json

@@ -29,7 +29,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-73fdd01bad58b8be1a52f89faef8d92a983470adcd3cc850734960c905e61e83.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-7420b825e8e2ee74b7839dbffec5709b99faecb45044069bfa895262fb5ca940.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE jobs_pending_deletion SET marked_by = $1, marked_at = now()\n         WHERE marked_at < now() - interval '30 minutes'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "7420b825e8e2ee74b7839dbffec5709b99faecb45044069bfa895262fb5ca940"
+}

backend/.sqlx/query-74ac1a31ded8aed51a06a44aee9cd4b7ad3a17ed0a03c376957e819701726e4b.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM job_stats WHERE job_id IN (SELECT id FROM v2_job WHERE created_by = $1 AND tag = $2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "74ac1a31ded8aed51a06a44aee9cd4b7ad3a17ed0a03c376957e819701726e4b"
+}

backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json

@@ -40,7 +40,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-7d0546f5c70c8f9015791c8ee0717ad453e5d5e11253e90efc8518b03552e461.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval",
+  "query": "SELECT COUNT(*) FROM jobs_pending_deletion WHERE marked_by = $1",
   "describe": {
     "columns": [
       {
@@ -11,12 +11,12 @@
     ],
     "parameters": {
       "Left": [
-        "Int8"
+        "Text"
       ]
     },
     "nullable": [
       null
     ]
   },
-  "hash": "10cdcf3155d8e58350e7df9acdbc73893c57716663a579b8bd120b542cc186f7"
+  "hash": "7d0546f5c70c8f9015791c8ee0717ad453e5d5e11253e90efc8518b03552e461"
 }

backend/.sqlx/query-7fbf72d9059fcd77e4c1112fa4fa22e4276c1da653475628889ce17dc904fbaa.json

@@ -27,7 +27,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-836bac47d89113d90bd03a471446eb9016207975af1e37042d81df8cb6ae2c53.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-87564a196a1662f524407d853db506bf08c28efe82b68b3d44bafbd3d0e91c29.json

@@ -35,7 +35,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-940b6d78bab940a37a42492f030d2393e297043e4e58555d872b5c4dd89c196a.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-94778a1971be08e528c4fc87a15998b157a6ea80732c324bb422b1c5a0fad606.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM job_logs WHERE job_id IN (SELECT id FROM v2_job WHERE created_by = $1 AND tag = $2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "94778a1971be08e528c4fc87a15998b157a6ea80732c324bb422b1c5a0fad606"
+}

backend/.sqlx/query-9c50e3a136a8ee3ec56e083f26d3a960b89e02ec40b292f3b5198baf2a1d3dbf.json

@@ -32,7 +32,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json

@@ -30,7 +30,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-9fda5c99a4917fbb0ad38f3e1f57328353d49fa1b311c9d45d3d4e75d6d9e7d3.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM jobs_pending_deletion WHERE marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "9fda5c99a4917fbb0ad38f3e1f57328353d49fa1b311c9d45d3d4e75d6d9e7d3"
+}

backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json

@@ -155,7 +155,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json

@@ -185,7 +185,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-a9d80d005362d1748e354c6bb6f19b5205ecd4c92f16f2b756dddd3485b1b7d6.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH marked AS (\n                 INSERT INTO jobs_pending_deletion (id, marked_by)\n                 SELECT jc.id, $4 FROM v2_job_completed jc\n                 LEFT JOIN v2_job j ON j.id = jc.id\n                 WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval\n                   AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)\n                 ORDER BY jc.completed_at ASC\n                 LIMIT $2\n                 ON CONFLICT DO NOTHING\n                 RETURNING id\n             )\n             DELETE FROM v2_job_completed WHERE id IN (SELECT id FROM marked)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Int8",
+        "UuidArray",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "a9d80d005362d1748e354c6bb6f19b5205ecd4c92f16f2b756dddd3485b1b7d6"
+}

backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json

@@ -160,7 +160,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-b3f0595cacba194e08b9a3e244d9e637e9e156cd85b69126c87dfff89a47711d.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-b8205e1b6b9b75e1d1d445e23d158d6d0525c78f1d447d4abb7c5ef674bd6559.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM job_result_stream_v2 USING jobs_pending_deletion d\n             WHERE job_result_stream_v2.job_id = d.id AND d.marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "b8205e1b6b9b75e1d1d445e23d158d6d0525c78f1d447d4abb7c5ef674bd6559"
+}

backend/.sqlx/query-bc2755b3815b163b7199ed30118e3497f1a33b81b47e4877f0f443ed52a89755.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM v2_job WHERE created_by = $1 AND tag = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "bc2755b3815b163b7199ed30118e3497f1a33b81b47e4877f0f443ed52a89755"
+}

backend/.sqlx/query-bd768adb92ca06a0733278d01e016220bf72da0bf62f13816e161b789530b85a.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM job_stats USING jobs_pending_deletion d\n             WHERE job_stats.job_id = d.id AND d.marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "bd768adb92ca06a0733278d01e016220bf72da0bf62f13816e161b789530b85a"
+}

backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json

@@ -105,7 +105,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-c3b1152b554812d65eb27f95b1fd434f860922fbc021185beffb9827647feb8e.json

@@ -31,7 +31,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-c687c38563027e898cdaf2b3175d6f898c32e419cbb29707c54c92d8608ec762.json

@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM v2_job_completed\n             WHERE id IN (\n                 SELECT jc.id FROM v2_job_completed jc\n                 JOIN v2_job j ON j.id = jc.id\n                 WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval\n                   AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)\n                   AND j.tag = $4\n                 ORDER BY jc.completed_at ASC\n                 LIMIT $2\n                 FOR UPDATE OF jc SKIP LOCKED\n             )\n             RETURNING id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Int8",
+        "UuidArray",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "c687c38563027e898cdaf2b3175d6f898c32e419cbb29707c54c92d8608ec762"
+}

backend/.sqlx/query-c95643742fc5d3d9f16e3ba49bf67a3c88576981111b7141b61af500e1e15d6a.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM v2_job USING jobs_pending_deletion d\n         WHERE v2_job.id = d.id AND d.marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c95643742fc5d3d9f16e3ba49bf67a3c88576981111b7141b61af500e1e15d6a"
+}

backend/.sqlx/query-ca2d6133fb8c93f3073583514aba62f4eacac185ed98cce7cb06dba9ba74cb36.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO job_logs (job_id, workspace_id, logs) VALUES ($1, 'admins', 'test log output')\n                 ON CONFLICT DO NOTHING",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ca2d6133fb8c93f3073583514aba62f4eacac185ed98cce7cb06dba9ba74cb36"
+}

backend/.sqlx/query-ccef7a1bde5cac6c362c5fedb6c13f1f882b695f896f94e5cf91d205633355a1.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json

@@ -105,7 +105,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-d495c94b580fd34d5ae90615ef21a8a9cc35f362197c0766a5787436af141106.json

@@ -25,7 +25,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-db3b300319cd3a5dee5dd078dd9df4729e68170d85ab6988e19ac6c8fc753253.json

@@ -0,0 +1,18 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH marked AS (\n                 INSERT INTO jobs_pending_deletion (id, marked_by)\n                 SELECT jc.id, $5 FROM v2_job_completed jc\n                 JOIN v2_job j ON j.id = jc.id\n                 WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval\n                   AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)\n                   AND j.tag = $4\n                 ORDER BY jc.completed_at ASC\n                 LIMIT $2\n                 ON CONFLICT DO NOTHING\n                 RETURNING id\n             )\n             DELETE FROM v2_job_completed WHERE id IN (SELECT id FROM marked)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Int8",
+        "UuidArray",
+        "Text",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "db3b300319cd3a5dee5dd078dd9df4729e68170d85ab6988e19ac6c8fc753253"
+}

backend/.sqlx/query-de2acdcaee4f75bb04abd795354acb1d67493f89a79601e81e804154caa1fd19.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO v2_job_completed (id, workspace_id, duration_ms, deleted, status, completed_at, started_at)\n                 VALUES ($1, 'admins', 100, false, 'success', now() - interval '30 minutes', now() - interval '30 minutes')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "de2acdcaee4f75bb04abd795354acb1d67493f89a79601e81e804154caa1fd19"
+}

backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json

@@ -185,7 +185,8 @@
                 "sqs",
                 "gcp",
                 "mqtt",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-e80177f3ffd4c1f52cdb4757483f03f72ef81db302d727e18e63a307ac902022.json

@@ -31,7 +31,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-eac595e19e5c8e70f1514ef29dec35c7342ac9a814c73f6290e1d6ebd3a55423.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-ed8facbf29ebb670d05fe8aa34b50d6a6935420fbedc83aa3ad1e9be7465c8dd.json

@@ -24,7 +24,8 @@
                 "mqtt",
                 "gcp",
                 "default_email",
-                "nextcloud"
+                "nextcloud",
+                "google"
               ]
             }
           }

backend/.sqlx/query-f21c608aa332b1600d2236ceba2be64c825e8bf5a2cb6446a2d973d028d6311f.json

@@ -1,23 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM v2_job_completed\n                         WHERE id IN (\n                             SELECT id FROM v2_job_completed\n                             WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval\n                             ORDER BY completed_at ASC\n                             LIMIT $2\n                             FOR UPDATE SKIP LOCKED\n                         )\n                         RETURNING id",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id",
-        "type_info": "Uuid"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Int8",
-        "Int8"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "f21c608aa332b1600d2236ceba2be64c825e8bf5a2cb6446a2d973d028d6311f"
-}

backend/.sqlx/query-f36c2e5c4617e88c4fb7708842e6c6bab3194842be4a4306e3e7863eababdcf0.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM job_result_stream_v2 USING jobs_pending_deletion d\n         WHERE job_result_stream_v2.job_id = d.id AND d.marked_by = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f36c2e5c4617e88c4fb7708842e6c6bab3194842be4a4306e3e7863eababdcf0"
+}

backend/.sqlx/query-f4b86c8950739b4e8cbd447d9ac0814a1ae69067735721106af19c188c77377c.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM v2_job_completed WHERE id IN (SELECT id FROM v2_job WHERE created_by = $1 AND tag = $2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "f4b86c8950739b4e8cbd447d9ac0814a1ae69067735721106af19c188c77377c"
+}

backend/.sqlx/query-fa4b7688a4e83dee51d5c64b7660b824a7ec745ad62845d469a5be6219a2be29.json

@@ -1,23 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM v2_job_completed\n                 WHERE id IN (\n                     SELECT id FROM v2_job_completed\n                     WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval\n                     ORDER BY completed_at ASC\n                     LIMIT $2\n                     FOR UPDATE SKIP LOCKED\n                 )\n                 RETURNING id",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id",
-        "type_info": "Uuid"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Int8",
-        "Int8"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "fa4b7688a4e83dee51d5c64b7660b824a7ec745ad62845d469a5be6219a2be29"
-}

backend/migrations/20260215000000_jobs_pending_deletion.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS jobs_pending_deletion;

backend/migrations/20260215000000_jobs_pending_deletion.up.sql

@@ -0,0 +1,17 @@
+CREATE TABLE IF NOT EXISTS jobs_pending_deletion (
+    id UUID PRIMARY KEY,
+    marked_by VARCHAR(10) NOT NULL,
+    marked_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_jobs_pending_deletion_marked_by
+    ON jobs_pending_deletion (marked_by);
+
+-- Aggressive autovacuum: this table churns heavily (bulk insert + delete each cycle).
+-- Vacuum after every batch clear to prevent dead tuple bloat.
+ALTER TABLE jobs_pending_deletion SET (
+    autovacuum_vacuum_scale_factor = 0,
+    autovacuum_vacuum_threshold = 50,
+    autovacuum_analyze_scale_factor = 0,
+    autovacuum_analyze_threshold = 50
+);

backend/src/monitor.rs

@@ -79,9 +79,9 @@ use windmill_common::{
         WORKER_CONFIG, WORKER_GROUP,
     },
     KillpillSender, BASE_URL, CRITICAL_ALERTS_ON_DB_OVERSIZE, CRITICAL_ALERT_MUTE_UI_ENABLED,
-    CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS,
-    METRICS_DEBUG_ENABLED, METRICS_ENABLED, MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED,
-    OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED, SERVICE_LOG_RETENTION_SECS,
+    CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, INSTANCE_NAME,
+    JOB_RETENTION_SECS, METRICS_DEBUG_ENABLED, METRICS_ENABLED, MONITOR_LOGS_ON_OBJECT_STORE,
+    OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED, SERVICE_LOG_RETENTION_SECS,
 };
 use windmill_common::{client::AuthedClient, global_settings::APP_WORKSPACED_ROUTE_SETTING};
 use windmill_queue::{cancel_job, get_queued_job_v2, SameWorkerPayload};
@@ -1057,109 +1057,178 @@ pub async fn delete_expired_items(db: &DB) -> () {
     }
 }
 
-/// Delete a batch of expired jobs with LIMIT and SKIP LOCKED for high-scale environments.
-/// Uses a single transaction per batch to minimize lock duration.
-/// Returns the number of jobs deleted in this batch.
+/// Delete a batch of expired jobs using the `jobs_pending_deletion` staging table.
+///
+/// Each step runs in its own transaction so that crashes are recoverable: if the server
+/// restarts mid-cleanup, leftover rows in `jobs_pending_deletion` are picked up on the next run.
+/// No job UUIDs are held in application memory — all joins stay inside Postgres.
+///
+/// Rows are scoped by `marked_by` (the server's `INSTANCE_NAME`) so multiple servers
+/// can run cleanup concurrently without interfering with each other. Stale rows from dead
+/// servers are adopted via `marked_at` timestamp after 30 minutes.
+///
+/// Returns the number of jobs marked for deletion in this batch.
 async fn delete_expired_jobs_batch(
     db: &DB,
     job_retention_secs: i64,
     batch_size: i64,
 ) -> error::Result<usize> {
-    let mut tx = db.begin().await?;
+    let instance = &*INSTANCE_NAME;
 
-    // Fetch active ROOT job IDs that started before the retention period. We only care about
-    // these because their child jobs could be old enough to be deletion candidates.
-    // Jobs started after the retention period can't have children old enough to delete.
-    let active_root_job_ids: Vec<Uuid> = sqlx::query_scalar!(
-        "SELECT q.id FROM v2_job_queue q
-         JOIN v2_job j ON j.id = q.id
-         WHERE j.parent_job IS NULL
-           AND j.created_at <= now() - ($1::bigint::text || ' s')::interval",
-        job_retention_secs
+    // Step 0: Adopt orphaned rows from dead servers (stale for >30 min), then check
+    // for leftover rows from a previous incomplete run by this same instance.
+    sqlx::query!(
+        "UPDATE jobs_pending_deletion SET marked_by = $1, marked_at = now()
+         WHERE marked_at < now() - interval '30 minutes'",
+        instance
     )
-    .fetch_all(&mut *tx)
+    .execute(db)
     .await?;
 
-    // Use FOR UPDATE SKIP LOCKED to avoid contention between replicas
-    // ORDER BY completed_at ensures we delete oldest jobs first
-    let deleted_jobs: Vec<Uuid> = sqlx::query_scalar!(
-        "DELETE FROM v2_job_completed
-         WHERE id IN (
-             SELECT jc.id FROM v2_job_completed jc
-             LEFT JOIN v2_job j ON j.id = jc.id
-             WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval
-               AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)
-             ORDER BY jc.completed_at ASC
-             LIMIT $2
-             FOR UPDATE OF jc SKIP LOCKED
-         )
-         RETURNING id",
-        job_retention_secs,
-        batch_size,
-        &active_root_job_ids
+    let pending_count: i64 = sqlx::query_scalar!(
+        "SELECT COUNT(*) FROM jobs_pending_deletion WHERE marked_by = $1",
+        instance
     )
-    .fetch_all(&mut *tx)
-    .await?;
+    .fetch_one(db)
+    .await?
+    .unwrap_or(0);
 
-    let deleted_count = deleted_jobs.len();
-
-    if deleted_count > 0 {
-        tracing::debug!(
-            "Deleting batch of {} expired jobs (retention: {}s)",
-            deleted_count,
-            job_retention_secs
+    let marked_count = if pending_count > 0 {
+        tracing::info!(
+            "Job cleanup: resuming with {} leftover jobs from previous run (instance {})",
+            pending_count,
+            instance
         );
-
-        if let Err(e) = sqlx::query!(
-            "DELETE FROM job_stats WHERE job_id = ANY($1)",
-            &deleted_jobs
+        pending_count as usize
+    } else {
+        // Step 1: Mark expired jobs and atomically remove them from v2_job_completed.
+        let active_root_job_ids: Vec<Uuid> = sqlx::query_scalar!(
+            "SELECT q.id FROM v2_job_queue q
+             JOIN v2_job j ON j.id = q.id
+             WHERE j.parent_job IS NULL
+               AND j.created_at <= now() - ($1::bigint::text || ' s')::interval",
+            job_retention_secs
         )
-        .execute(&mut *tx)
-        .await
-        {
-            tracing::error!("Error deleting job stats: {:?}", e);
-        }
+        .fetch_all(db)
+        .await?;
 
-        match sqlx::query_scalar!(
-            "DELETE FROM job_logs WHERE job_id = ANY($1) RETURNING log_file_index",
-            &deleted_jobs
+        let result = sqlx::query!(
+            "WITH marked AS (
+                 INSERT INTO jobs_pending_deletion (id, marked_by)
+                 SELECT jc.id, $4 FROM v2_job_completed jc
+                 LEFT JOIN v2_job j ON j.id = jc.id
+                 WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval
+                   AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)
+                 ORDER BY jc.completed_at ASC
+                 LIMIT $2
+                 ON CONFLICT DO NOTHING
+                 RETURNING id
+             )
+             DELETE FROM v2_job_completed WHERE id IN (SELECT id FROM marked)",
+            job_retention_secs,
+            batch_size,
+            &active_root_job_ids,
+            instance
         )
-        .fetch_all(&mut *tx)
-        .await
-        {
-            Ok(log_file_index) => {
-                let paths = log_file_index
-                    .into_iter()
-                    .filter_map(|opt| opt)
-                    .flat_map(|inner_vec| inner_vec.into_iter())
-                    .collect();
-                delete_log_files_from_disk_and_store(paths, TMP_DIR, "").await;
-            }
-            Err(e) => tracing::error!("Error deleting job logs: {:?}", e),
-        }
+        .execute(db)
+        .await?;
 
-        if let Err(e) = sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted_jobs)
-            .execute(&mut *tx)
-            .await
-        {
-            tracing::error!("Error deleting job: {:?}", e);
-        }
+        result.rows_affected() as usize
+    };
 
-        // Should already be deleted but just in case
-        if let Err(e) = sqlx::query!(
-            "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)",
-            &deleted_jobs
-        )
-        .execute(&mut *tx)
-        .await
-        {
-            tracing::error!("Error deleting job result stream: {:?}", e);
-        }
+    if marked_count == 0 {
+        return Ok(0);
     }
 
-    tx.commit().await?;
+    tracing::debug!(
+        "Deleting batch of {} expired jobs (retention: {}s, instance: {})",
+        marked_count,
+        job_retention_secs,
+        instance
+    );
 
-    Ok(deleted_count)
+    // Step 2: Delete from job_stats
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM job_stats USING jobs_pending_deletion d
+         WHERE job_stats.job_id = d.id AND d.marked_by = $1",
+        instance
+    )
+    .execute(db)
+    .await
+    {
+        tracing::error!("Error deleting job stats: {:?}", e);
+    }
+
+    // Step 3: Delete from job_logs (collect log_file_index for disk cleanup)
+    match sqlx::query_scalar!(
+        "DELETE FROM job_logs USING jobs_pending_deletion d
+         WHERE job_logs.job_id = d.id AND d.marked_by = $1
+         RETURNING job_logs.log_file_index",
+        instance
+    )
+    .fetch_all(db)
+    .await
+    {
+        Ok(log_file_index) => {
+            let paths = log_file_index
+                .into_iter()
+                .filter_map(|opt| opt)
+                .flat_map(|inner_vec| inner_vec.into_iter())
+                .collect();
+            delete_log_files_from_disk_and_store(paths, TMP_DIR, "").await;
+        }
+        Err(e) => tracing::error!("Error deleting job logs: {:?}", e),
+    }
+
+    // Step 4: Delete from v2_job_completed (idempotent — already done for fresh batches
+    // via the CTE in step 1, but needed for crash-recovery and adopted orphan rows).
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM v2_job_completed USING jobs_pending_deletion d
+         WHERE v2_job_completed.id = d.id AND d.marked_by = $1",
+        instance
+    )
+    .execute(db)
+    .await
+    {
+        tracing::error!("Error deleting completed jobs: {:?}", e);
+    }
+
+    // Step 5: Delete from v2_job
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM v2_job USING jobs_pending_deletion d
+         WHERE v2_job.id = d.id AND d.marked_by = $1",
+        instance
+    )
+    .execute(db)
+    .await
+    {
+        tracing::error!("Error deleting jobs: {:?}", e);
+    }
+
+    // Step 6: Delete from job_result_stream_v2
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM job_result_stream_v2 USING jobs_pending_deletion d
+         WHERE job_result_stream_v2.job_id = d.id AND d.marked_by = $1",
+        instance
+    )
+    .execute(db)
+    .await
+    {
+        tracing::error!("Error deleting job result stream: {:?}", e);
+    }
+
+    // Step 7: Clear this instance's rows from the staging table
+    if let Err(e) = sqlx::query!(
+        "DELETE FROM jobs_pending_deletion WHERE marked_by = $1",
+        instance
+    )
+    .execute(db)
+    .await
+    {
+        tracing::error!("Error clearing jobs_pending_deletion: {:?}", e);
+    }
+
+    Ok(marked_count)
 }
 
 async fn delete_log_files_from_disk_and_store(

backend/tests/operator_db_sync.rs

@@ -7,7 +7,7 @@
  * - Protect certain internal settings from deletion
  */
 
-#[cfg(feature = "operator")]
+#[cfg(all(feature = "operator", feature = "private"))]
 mod tests {
     use std::collections::BTreeMap;
 
@@ -74,7 +74,7 @@ mod tests {
         );
         desired.insert("test_op_setting_b".to_string(), serde_json::json!(42));
 
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -97,7 +97,7 @@ mod tests {
         let mut desired = BTreeMap::new();
         desired.insert("test_op_existing".to_string(), serde_json::json!("new"));
 
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -117,7 +117,7 @@ mod tests {
         let mut desired = BTreeMap::new();
         desired.insert("test_op_keep".to_string(), serde_json::json!("keep"));
 
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -141,7 +141,7 @@ mod tests {
         // Sync with empty desired — protected key should survive
         let desired = BTreeMap::new();
 
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -165,7 +165,7 @@ mod tests {
 
         // Sync with empty desired
         let desired = BTreeMap::new();
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -183,7 +183,7 @@ mod tests {
         insert_global_setting(&db, "test_op_ephemeral", serde_json::json!("gone")).await;
 
         let desired = BTreeMap::new();
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -206,7 +206,7 @@ mod tests {
             }),
         );
 
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -230,7 +230,7 @@ mod tests {
             serde_json::json!({"dedicated_worker": false}),
         );
 
-        windmill_operator::db_sync::sync_worker_configs(&db, &desired)
+        windmill_operator::db_sync_ee::sync_worker_configs(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -256,7 +256,7 @@ mod tests {
             serde_json::json!({"new": true}),
         );
 
-        windmill_operator::db_sync::sync_worker_configs(&db, &desired)
+        windmill_operator::db_sync_ee::sync_worker_configs(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -290,7 +290,7 @@ mod tests {
             serde_json::json!({"keep": true}),
         );
 
-        windmill_operator::db_sync::sync_worker_configs(&db, &desired)
+        windmill_operator::db_sync_ee::sync_worker_configs(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -308,7 +308,7 @@ mod tests {
 
         // Sync with empty worker configs
         let desired = BTreeMap::new();
-        windmill_operator::db_sync::sync_worker_configs(&db, &desired)
+        windmill_operator::db_sync_ee::sync_worker_configs(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -334,7 +334,7 @@ mod tests {
             serde_json::json!({"init_bash": "echo native"}),
         );
 
-        windmill_operator::db_sync::sync_worker_configs(&db, &desired)
+        windmill_operator::db_sync_ee::sync_worker_configs(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -353,7 +353,7 @@ mod tests {
         .await;
 
         let desired = BTreeMap::new();
-        windmill_operator::db_sync::sync_worker_configs(&db, &desired)
+        windmill_operator::db_sync_ee::sync_worker_configs(&db, &desired)
             .await
             .expect("sync should succeed");
 
@@ -396,10 +396,10 @@ mod tests {
         );
 
         // Sync both
-        windmill_operator::db_sync::sync_global_settings(&db, &global_settings)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &global_settings)
             .await
             .expect("global sync should succeed");
-        windmill_operator::db_sync::sync_worker_configs(&db, &worker_configs)
+        windmill_operator::db_sync_ee::sync_worker_configs(&db, &worker_configs)
             .await
             .expect("worker sync should succeed");
 
@@ -433,10 +433,10 @@ mod tests {
         desired.insert("test_op_idempotent".to_string(), serde_json::json!("value"));
 
         // Run sync twice — should be idempotent
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("first sync should succeed");
-        windmill_operator::db_sync::sync_global_settings(&db, &desired)
+        windmill_operator::db_sync_ee::sync_global_settings(&db, &desired)
             .await
             .expect("second sync should succeed");
 

backend/windmill-queue/tests/job_cleanup_test.rs

@@ -1,299 +1,524 @@
-/// Integration test for batched job cleanup
+/// Integration benchmark comparing old (in-memory UUID) vs new (jobs_pending_deletion table)
+/// job cleanup approaches.
+///
+/// Both methods scope their candidate selection to a specific `tag` so that each run
+/// operates on an isolated set of jobs. This ensures fair comparison regardless of other
+/// expired jobs in the database.
 ///
 /// Prerequisites:
-///   - PostgreSQL running with windmill database
+///   - PostgreSQL running with windmill database (migrations applied)
 ///   - DATABASE_URL env var or default: postgres://postgres:changeme@localhost:5432/windmill
 ///
-/// Run all tests:
-///   cargo test --test job_cleanup_test -- --nocapture
+/// Run comparison benchmark:
+///   cargo test -p windmill-queue --test job_cleanup_test test_compare_cleanup_methods -- --nocapture --ignored
 ///
-/// Run specific test:
-///   cargo test --test job_cleanup_test test_batched_job_cleanup -- --nocapture
-///   cargo test --test job_cleanup_test test_skip_locked_prevents_contention -- --nocapture
-///
-/// Setup test data (run in psql first):
-///   -- Set short retention for testing
-///   UPDATE global_settings SET value = '3600' WHERE name = 'retention_period_secs';
-///
-///   -- Insert expired test jobs
-///   INSERT INTO v2_job (id, tag, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email, kind, same_worker, visible_to_owner)
-///   SELECT gen_random_uuid(), 'deno', 'admins', now() - interval '2 hours', 'test_user', 'test_user', 'test@example.com', 'script', false, true
-///   FROM generate_series(1, 10000);
-///
-///   INSERT INTO v2_job_completed (id, workspace_id, duration_ms, deleted, status, completed_at, started_at)
-///   SELECT j.id, j.workspace_id, 100, false, 'success', j.created_at + interval '1 second', j.created_at
-///   FROM v2_job j WHERE j.created_by = 'test_user';
+/// Run individual tests:
+///   cargo test -p windmill-queue --test job_cleanup_test test_old_method -- --nocapture --ignored
+///   cargo test -p windmill-queue --test job_cleanup_test test_new_method -- --nocapture --ignored
 use sqlx::postgres::PgPoolOptions;
+use sqlx::{Pool, Postgres};
 use std::time::Instant;
 use uuid::Uuid;
 
-#[tokio::test]
-#[ignore]
-async fn test_batched_job_cleanup() {
-    // Connect to database
-    let database_url = std::env::var("DATABASE_URL")
-        .unwrap_or("postgres://postgres:changeme@localhost:5432/windmill".to_string());
+const DEFAULT_DB_URL: &str = "postgres://postgres:changeme@localhost:5432/windmill";
+const DEFAULT_JOB_COUNT: usize = 50_000;
 
-    let db = PgPoolOptions::new()
+async fn connect() -> Pool<Postgres> {
+    let url = std::env::var("DATABASE_URL").unwrap_or(DEFAULT_DB_URL.to_string());
+    PgPoolOptions::new()
         .max_connections(5)
-        .connect(&database_url)
+        .connect(&url)
         .await
-        .expect("Failed to connect to database");
+        .expect("Failed to connect to database")
+}
 
-    // Get retention period from settings
-    let retention_secs: i64 = 2592000;
+async fn insert_test_jobs(db: &Pool<Postgres>, count: usize, tag: &str) {
+    for chunk in (0..count).collect::<Vec<_>>().chunks(1000) {
+        let chunk_ids: Vec<Uuid> = chunk.iter().map(|_| Uuid::new_v4()).collect();
 
-    println!(
-        "Retention period: {} seconds ({} days)",
-        retention_secs,
-        retention_secs / 86400
-    );
+        for id in &chunk_ids {
+            sqlx::query!(
+                "INSERT INTO v2_job (id, tag, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email, kind, same_worker, visible_to_owner)
+                 VALUES ($1, $2, 'admins', now() - interval '30 minutes', $3, $3, 'bench@test.com', 'script', false, true)",
+                id,
+                tag,
+                "bench_user"
+            )
+            .execute(db)
+            .await
+            .expect("Failed to insert v2_job");
+        }
 
-    // Count jobs eligible for deletion
-    let eligible_count: i64 = sqlx::query_scalar!(
-        "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval",
-        retention_secs
-    )
-    .fetch_one(&db)
-    .await
-    .expect("Failed to count eligible jobs")
-    .unwrap_or(0);
+        for id in &chunk_ids {
+            sqlx::query!(
+                "INSERT INTO v2_job_completed (id, workspace_id, duration_ms, deleted, status, completed_at, started_at)
+                 VALUES ($1, 'admins', 100, false, 'success', now() - interval '30 minutes', now() - interval '30 minutes')",
+                id
+            )
+            .execute(db)
+            .await
+            .expect("Failed to insert v2_job_completed");
+        }
 
-    println!("Jobs eligible for deletion: {}", eligible_count);
+        for id in &chunk_ids {
+            sqlx::query!(
+                "INSERT INTO job_stats (job_id, workspace_id) VALUES ($1, 'admins')
+                 ON CONFLICT DO NOTHING",
+                id,
+            )
+            .execute(db)
+            .await
+            .ok();
+        }
 
-    if eligible_count == 0 {
-        println!("No jobs to delete. To create test data, run:");
-        println!("  psql $DATABASE_URL -c \"UPDATE global_settings SET value = '3600' WHERE name = 'retention_period_secs';\"");
-        println!("  Then insert expired jobs (see test file header for SQL)");
-        return;
+        for id in &chunk_ids {
+            sqlx::query!(
+                "INSERT INTO job_logs (job_id, workspace_id, logs) VALUES ($1, 'admins', 'test log output')
+                 ON CONFLICT DO NOTHING",
+                id,
+            )
+            .execute(db)
+            .await
+            .ok();
+        }
     }
+}
 
-    // Test different batch sizes
-    let batch_sizes = [1000, 5000, 10000];
+async fn cleanup_test_data(db: &Pool<Postgres>, tag: &str) {
+    sqlx::query!("DELETE FROM jobs_pending_deletion")
+        .execute(db)
+        .await
+        .ok();
+    sqlx::query!(
+        "DELETE FROM job_stats WHERE job_id IN (SELECT id FROM v2_job WHERE created_by = $1 AND tag = $2)",
+        "bench_user",
+        tag
+    )
+    .execute(db)
+    .await
+    .ok();
+    sqlx::query!(
+        "DELETE FROM job_logs WHERE job_id IN (SELECT id FROM v2_job WHERE created_by = $1 AND tag = $2)",
+        "bench_user",
+        tag
+    )
+    .execute(db)
+    .await
+    .ok();
+    sqlx::query!(
+        "DELETE FROM v2_job_completed WHERE id IN (SELECT id FROM v2_job WHERE created_by = $1 AND tag = $2)",
+        "bench_user",
+        tag
+    )
+    .execute(db)
+    .await
+    .ok();
+    sqlx::query!(
+        "DELETE FROM v2_job WHERE created_by = $1 AND tag = $2",
+        "bench_user",
+        tag
+    )
+    .execute(db)
+    .await
+    .ok();
+}
 
-    for batch_size in batch_sizes {
-        let remaining: i64 = sqlx::query_scalar!(
-            "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval",
+/// Old method: single transaction, DELETE RETURNING into Vec<Uuid>, then ANY($1) cascades.
+/// Scoped to jobs with the given `tag` for isolation.
+async fn run_old_method(
+    db: &Pool<Postgres>,
+    retention_secs: i64,
+    batch_size: i64,
+    tag: &str,
+) -> (usize, std::time::Duration) {
+    let start = Instant::now();
+    let mut total = 0usize;
+
+    loop {
+        let batch_start = Instant::now();
+        let mut tx = db.begin().await.unwrap();
+
+        let active_roots: Vec<Uuid> = sqlx::query_scalar!(
+            "SELECT q.id FROM v2_job_queue q
+             JOIN v2_job j ON j.id = q.id
+             WHERE j.parent_job IS NULL
+               AND j.created_at <= now() - ($1::bigint::text || ' s')::interval",
             retention_secs
         )
-        .fetch_one(&db)
+        .fetch_all(&mut *tx)
         .await
-        .unwrap()
-        .unwrap_or(0);
+        .unwrap();
 
-        if remaining == 0 {
-            println!("All jobs deleted, stopping batch size tests");
+        let deleted_jobs: Vec<Uuid> = sqlx::query_scalar!(
+            "DELETE FROM v2_job_completed
+             WHERE id IN (
+                 SELECT jc.id FROM v2_job_completed jc
+                 JOIN v2_job j ON j.id = jc.id
+                 WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval
+                   AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)
+                   AND j.tag = $4
+                 ORDER BY jc.completed_at ASC
+                 LIMIT $2
+                 FOR UPDATE OF jc SKIP LOCKED
+             )
+             RETURNING id",
+            retention_secs,
+            batch_size,
+            &active_roots,
+            tag
+        )
+        .fetch_all(&mut *tx)
+        .await
+        .unwrap();
+
+        let count = deleted_jobs.len();
+        if count == 0 {
+            tx.commit().await.unwrap();
             break;
         }
 
+        sqlx::query!("DELETE FROM job_stats WHERE job_id = ANY($1)", &deleted_jobs)
+            .execute(&mut *tx)
+            .await
+            .ok();
+        sqlx::query!("DELETE FROM job_logs WHERE job_id = ANY($1)", &deleted_jobs)
+            .execute(&mut *tx)
+            .await
+            .ok();
+        sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted_jobs)
+            .execute(&mut *tx)
+            .await
+            .ok();
+        sqlx::query!(
+            "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)",
+            &deleted_jobs
+        )
+        .execute(&mut *tx)
+        .await
+        .ok();
+
+        tx.commit().await.unwrap();
+        total += count;
+
+        let rate = count as f64 / batch_start.elapsed().as_secs_f64();
         println!(
-            "\n--- Testing batch_size={} (remaining: {}) ---",
-            batch_size, remaining
+            "  [old] batch: {} jobs in {:?} ({:.0} jobs/sec)",
+            count,
+            batch_start.elapsed(),
+            rate
         );
+    }
 
-        let start = Instant::now();
-        let mut total_deleted = 0u64;
-        let mut batch_num = 0u32;
-        let max_batches = 5;
+    (total, start.elapsed())
+}
 
-        loop {
-            if batch_num >= max_batches {
-                println!("Reached max batches limit ({})", max_batches);
-                break;
-            }
+/// New method: INSERT INTO jobs_pending_deletion, then per-table DELETE USING joins.
+/// Scoped to jobs with the given `tag` for isolation.
+async fn run_new_method(
+    db: &Pool<Postgres>,
+    retention_secs: i64,
+    batch_size: i64,
+    tag: &str,
+) -> (usize, std::time::Duration) {
+    let start = Instant::now();
+    let mut total = 0usize;
+    let instance = "bench";
 
-            let batch_start = Instant::now();
+    loop {
+        let batch_start = Instant::now();
 
-            // Run batched deletion with LIMIT and SKIP LOCKED (same as monitor.rs)
-            let deleted_jobs: Vec<Uuid> = sqlx::query_scalar!(
-                "DELETE FROM v2_job_completed
-                 WHERE id IN (
-                     SELECT id FROM v2_job_completed
-                     WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval
-                     ORDER BY completed_at ASC
-                     LIMIT $2
-                     FOR UPDATE SKIP LOCKED
-                 )
-                 RETURNING id",
-                retention_secs,
-                batch_size as i64
-            )
-            .fetch_all(&db)
-            .await
-            .expect("Failed to delete batch");
+        let active_roots: Vec<Uuid> = sqlx::query_scalar!(
+            "SELECT q.id FROM v2_job_queue q
+             JOIN v2_job j ON j.id = q.id
+             WHERE j.parent_job IS NULL
+               AND j.created_at <= now() - ($1::bigint::text || ' s')::interval",
+            retention_secs
+        )
+        .fetch_all(db)
+        .await
+        .unwrap();
 
-            let batch_count = deleted_jobs.len();
-            let batch_elapsed = batch_start.elapsed();
+        let result = sqlx::query!(
+            "WITH marked AS (
+                 INSERT INTO jobs_pending_deletion (id, marked_by)
+                 SELECT jc.id, $5 FROM v2_job_completed jc
+                 JOIN v2_job j ON j.id = jc.id
+                 WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval
+                   AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($3)
+                   AND j.tag = $4
+                 ORDER BY jc.completed_at ASC
+                 LIMIT $2
+                 ON CONFLICT DO NOTHING
+                 RETURNING id
+             )
+             DELETE FROM v2_job_completed WHERE id IN (SELECT id FROM marked)",
+            retention_secs,
+            batch_size,
+            &active_roots,
+            tag,
+            instance
+        )
+        .execute(db)
+        .await
+        .unwrap();
 
-            if batch_count == 0 {
-                println!("No more jobs to delete");
-                break;
-            }
-
-            // Delete related records
-            sqlx::query!(
-                "DELETE FROM job_stats WHERE job_id = ANY($1)",
-                &deleted_jobs
-            )
-            .execute(&db)
-            .await
-            .ok();
-
-            sqlx::query!("DELETE FROM job_logs WHERE job_id = ANY($1)", &deleted_jobs)
-                .execute(&db)
-                .await
-                .ok();
-
-            sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted_jobs)
-                .execute(&db)
-                .await
-                .ok();
-
-            sqlx::query!(
-                "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)",
-                &deleted_jobs
-            )
-            .execute(&db)
-            .await
-            .ok();
-
-            total_deleted += batch_count as u64;
-            batch_num += 1;
-
-            let rate = batch_count as f64 / batch_elapsed.as_secs_f64();
-            println!(
-                "  Batch {}: deleted {} jobs in {:?} ({:.0} jobs/sec)",
-                batch_num, batch_count, batch_elapsed, rate
-            );
+        let count = result.rows_affected() as usize;
+        if count == 0 {
+            break;
         }
 
-        let total_elapsed = start.elapsed();
-        let overall_rate = if total_elapsed.as_secs_f64() > 0.0 {
-            total_deleted as f64 / total_elapsed.as_secs_f64()
+        sqlx::query!(
+            "DELETE FROM job_stats USING jobs_pending_deletion d
+             WHERE job_stats.job_id = d.id AND d.marked_by = $1",
+            instance
+        )
+        .execute(db)
+        .await
+        .ok();
+        sqlx::query!(
+            "DELETE FROM job_logs USING jobs_pending_deletion d
+             WHERE job_logs.job_id = d.id AND d.marked_by = $1",
+            instance
+        )
+        .execute(db)
+        .await
+        .ok();
+        sqlx::query!(
+            "DELETE FROM v2_job USING jobs_pending_deletion d
+             WHERE v2_job.id = d.id AND d.marked_by = $1",
+            instance
+        )
+        .execute(db)
+        .await
+        .ok();
+        sqlx::query!(
+            "DELETE FROM job_result_stream_v2 USING jobs_pending_deletion d
+             WHERE job_result_stream_v2.job_id = d.id AND d.marked_by = $1",
+            instance
+        )
+        .execute(db)
+        .await
+        .ok();
+        sqlx::query!(
+            "DELETE FROM jobs_pending_deletion WHERE marked_by = $1",
+            instance
+        )
+        .execute(db)
+        .await
+        .ok();
+
+        total += count;
+
+        let rate = count as f64 / batch_start.elapsed().as_secs_f64();
+        println!(
+            "  [new] batch: {} jobs in {:?} ({:.0} jobs/sec)",
+            count,
+            batch_start.elapsed(),
+            rate
+        );
+    }
+
+    (total, start.elapsed())
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_compare_cleanup_methods() {
+    let db = connect().await;
+    let job_count = std::env::var("BENCH_JOB_COUNT")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(DEFAULT_JOB_COUNT);
+
+    let retention_secs: i64 = 600;
+    let batch_sizes: &[i64] = &[1000, 5000, 10000, 20000];
+
+    for &batch_size in batch_sizes {
+        println!(
+            "\n========== batch_size={}, job_count={} ==========",
+            batch_size, job_count
+        );
+
+        let tag_old = format!("bench_old_{}", batch_size);
+        let tag_new = format!("bench_new_{}", batch_size);
+
+        // Clean slate for both tags
+        cleanup_test_data(&db, &tag_old).await;
+        cleanup_test_data(&db, &tag_new).await;
+
+        // --- Old method ---
+        println!("[old] Inserting {} test jobs...", job_count);
+        insert_test_jobs(&db, job_count, &tag_old).await;
+        println!("[old] Running old method (in-memory UUID, single tx)...");
+        let (old_total, old_dur) = run_old_method(&db, retention_secs, batch_size, &tag_old).await;
+        let old_rate = if old_dur.as_secs_f64() > 0.0 {
+            old_total as f64 / old_dur.as_secs_f64()
         } else {
             0.0
         };
         println!(
-            "batch_size={}: deleted {} jobs in {} batches, total time {:?} ({:.0} jobs/sec)",
-            batch_size, total_deleted, batch_num, total_elapsed, overall_rate
+            "[old] Deleted {} jobs in {:?} ({:.0} jobs/sec)",
+            old_total, old_dur, old_rate
         );
+        cleanup_test_data(&db, &tag_old).await;
+
+        // --- New method ---
+        println!("[new] Inserting {} test jobs...", job_count);
+        insert_test_jobs(&db, job_count, &tag_new).await;
+        println!("[new] Running new method (jobs_pending_deletion table)...");
+        let (new_total, new_dur) = run_new_method(&db, retention_secs, batch_size, &tag_new).await;
+        let new_rate = if new_dur.as_secs_f64() > 0.0 {
+            new_total as f64 / new_dur.as_secs_f64()
+        } else {
+            0.0
+        };
+        println!(
+            "[new] Deleted {} jobs in {:?} ({:.0} jobs/sec)",
+            new_total, new_dur, new_rate
+        );
+        cleanup_test_data(&db, &tag_new).await;
+
+        println!("\n--- Summary (batch_size={}) ---", batch_size);
+        println!(
+            "  Old: {} jobs in {:?} ({:.0} jobs/sec)",
+            old_total, old_dur, old_rate
+        );
+        println!(
+            "  New: {} jobs in {:?} ({:.0} jobs/sec)",
+            new_total, new_dur, new_rate
+        );
+        if old_total != new_total {
+            println!(
+                "  WARNING: job counts differ (old={}, new={})",
+                old_total, new_total
+            );
+        }
+        if old_dur.as_secs_f64() > 0.0 && new_dur.as_secs_f64() > 0.0 {
+            let speedup = old_dur.as_secs_f64() / new_dur.as_secs_f64();
+            println!("  Speedup: {:.2}x", speedup);
+        }
     }
-
-    let final_count: i64 = sqlx::query_scalar!(
-        "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval",
-        retention_secs
-    )
-    .fetch_one(&db)
-    .await
-    .unwrap()
-    .unwrap_or(0);
-
-    println!(
-        "\nFinal eligible count: {} (started with {})",
-        final_count, eligible_count
-    );
 }
 
 #[tokio::test]
 #[ignore]
-async fn test_skip_locked_prevents_contention() {
-    // This test verifies that SKIP LOCKED allows concurrent cleanup without deadlocks
-    let database_url = std::env::var("DATABASE_URL")
-        .unwrap_or("postgres://postgres:changeme@localhost:5432/windmill".to_string());
+async fn test_old_method() {
+    let db = connect().await;
+    let retention_secs: i64 = 600;
+    let batch_size: i64 = 10000;
+    let job_count = 10_000;
+    let tag = "bench_old_solo";
 
-    let db = PgPoolOptions::new()
-        .max_connections(10)
-        .connect(&database_url)
-        .await
-        .expect("Failed to connect to database");
+    cleanup_test_data(&db, tag).await;
+    println!("Inserting {} test jobs...", job_count);
+    insert_test_jobs(&db, job_count, tag).await;
+    println!("Running old method...");
+    let (total, dur) = run_old_method(&db, retention_secs, batch_size, tag).await;
+    println!(
+        "Deleted {} jobs in {:?} ({:.0} jobs/sec)",
+        total,
+        dur,
+        total as f64 / dur.as_secs_f64()
+    );
+    cleanup_test_data(&db, tag).await;
+}
 
-    let retention_secs: i64 = sqlx::query_scalar!(
-        "SELECT COALESCE((SELECT value::bigint FROM global_settings WHERE name = 'retention_period_secs'), 2592000)"
-    )
-    .fetch_one(&db)
-    .await
-    .expect("Failed to get retention period")
-    .unwrap_or(2592000);
+#[tokio::test]
+#[ignore]
+async fn test_new_method() {
+    let db = connect().await;
+    let retention_secs: i64 = 600;
+    let batch_size: i64 = 10000;
+    let job_count = 10_000;
+    let tag = "bench_new_solo";
 
-    let eligible: i64 = sqlx::query_scalar!(
-        "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval",
+    cleanup_test_data(&db, tag).await;
+    println!("Inserting {} test jobs...", job_count);
+    insert_test_jobs(&db, job_count, tag).await;
+    println!("Running new method...");
+    let (total, dur) = run_new_method(&db, retention_secs, batch_size, tag).await;
+    println!(
+        "Deleted {} jobs in {:?} ({:.0} jobs/sec)",
+        total,
+        dur,
+        total as f64 / dur.as_secs_f64()
+    );
+    cleanup_test_data(&db, tag).await;
+}
+
+#[tokio::test]
+#[ignore]
+async fn test_crash_recovery() {
+    let db = connect().await;
+    let retention_secs: i64 = 600;
+    let tag = "bench_crash_recovery";
+    let instance = "bench";
+
+    cleanup_test_data(&db, tag).await;
+    println!("Inserting 1000 test jobs...");
+    insert_test_jobs(&db, 1000, tag).await;
+
+    // Simulate: mark jobs but don't delete them (mimics crash after step 1)
+    let active_roots: Vec<Uuid> = sqlx::query_scalar!(
+        "SELECT q.id FROM v2_job_queue q
+         JOIN v2_job j ON j.id = q.id
+         WHERE j.parent_job IS NULL
+           AND j.created_at <= now() - ($1::bigint::text || ' s')::interval",
         retention_secs
     )
-    .fetch_one(&db)
+    .fetch_all(&db)
     .await
-    .unwrap()
-    .unwrap_or(0);
+    .unwrap();
 
-    println!("Retention period: {} seconds", retention_secs);
-    println!("Jobs eligible for deletion: {}", eligible);
+    let marked = sqlx::query!(
+        "INSERT INTO jobs_pending_deletion (id, marked_by)
+         SELECT jc.id, $4 FROM v2_job_completed jc
+         JOIN v2_job j ON j.id = jc.id
+         WHERE jc.completed_at <= now() - ($1::bigint::text || ' s')::interval
+           AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) != ALL($2)
+           AND j.tag = $3
+         ORDER BY jc.completed_at ASC
+         LIMIT 1000
+         ON CONFLICT DO NOTHING",
+        retention_secs,
+        &active_roots,
+        tag,
+        instance
+    )
+    .execute(&db)
+    .await
+    .unwrap();
 
-    if eligible == 0 {
-        println!("No jobs to delete, skipping concurrent test");
-        return;
-    }
+    let marked_count = marked.rows_affected();
+    println!(
+        "Marked {} jobs (simulating crash before deletion)",
+        marked_count
+    );
+    assert!(marked_count > 0, "Should have marked some jobs");
 
-    let batch_size: i64 = 1000;
-
-    // Run 3 concurrent cleanup tasks (simulating multiple server replicas)
-    let handles: Vec<_> = (0..3)
-        .map(|worker_id| {
-            let db = db.clone();
-            tokio::spawn(async move {
-                let start = Instant::now();
-                let mut total = 0u64;
-
-                for _batch in 0..3 {
-                    let deleted: Vec<Uuid> = sqlx::query_scalar!(
-                        "DELETE FROM v2_job_completed
-                         WHERE id IN (
-                             SELECT id FROM v2_job_completed
-                             WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval
-                             ORDER BY completed_at ASC
-                             LIMIT $2
-                             FOR UPDATE SKIP LOCKED
-                         )
-                         RETURNING id",
-                        retention_secs,
-                        batch_size
-                    )
-                    .fetch_all(&db)
-                    .await
-                    .unwrap_or_default();
-
-                    let count = deleted.len();
-                    total += count as u64;
-
-                    if count > 0 {
-                        sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted)
-                            .execute(&db)
-                            .await
-                            .ok();
-                    }
-
-                    if count == 0 {
-                        break;
-                    }
-                }
-
-                println!(
-                    "Worker {}: deleted {} jobs in {:?}",
-                    worker_id,
-                    total,
-                    start.elapsed()
-                );
-                total
-            })
-        })
-        .collect();
-
-    let results: Vec<u64> = futures::future::join_all(handles)
+    // Verify pending_deletion has rows
+    let pending: i64 = sqlx::query_scalar!("SELECT COUNT(*) FROM jobs_pending_deletion")
+        .fetch_one(&db)
         .await
-        .into_iter()
-        .map(|r| r.unwrap_or(0))
-        .collect();
+        .unwrap()
+        .unwrap_or(0);
+    assert!(pending > 0, "jobs_pending_deletion should have rows");
 
-    let total: u64 = results.iter().sum();
-    println!("\nTotal deleted by all workers: {}", total);
-    println!("Per worker: {:?}", results);
-    println!("\nWith SKIP LOCKED: no deadlocks, work distributed across workers");
+    // Now run the new method — it should detect leftovers and resume
+    println!("Running new method (should resume from crash)...");
+    let (total, dur) = run_new_method(&db, retention_secs, 1000, tag).await;
+    println!("Recovered and deleted {} jobs in {:?}", total, dur);
+
+    // Verify cleanup
+    let remaining: i64 = sqlx::query_scalar!("SELECT COUNT(*) FROM jobs_pending_deletion")
+        .fetch_one(&db)
+        .await
+        .unwrap()
+        .unwrap_or(0);
+    assert_eq!(
+        remaining, 0,
+        "jobs_pending_deletion should be empty after recovery"
+    );
+
+    cleanup_test_data(&db, tag).await;
 }