From e90b2be8fade1eb78cd685890291f5a4553a6a10 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 23 Jun 2026 22:49:55 +0200 Subject: [PATCH] perf(monitor): skip protected prefix in retention delete via cross-batch watermark (WIN-2088) (#9744) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The expired-job retention loop re-scanned the same oldest rows on every batch. When the oldest completed jobs are undeletable (children of a still-active root flow), the ORDER BY completed_at ASC scan walked that protected prefix on each of the up-to-20 batches, doing a v2_job PK lookup per row — quadratic in prefix size (measured ~9s/batch, ~180s/cleanup-cycle on a 1.5M-row prefix). Carry a completed_at watermark (max deleted) across batches and re-apply it as completed_at >= floor so each batch resumes past the already-processed prefix. Also skip the v2_job join entirely when no old root flow is active (the common case), since nothing is protected then. Measured: subsequent batches 9000ms -> 159ms; empty-set path 154 -> 36ms. The watermark only ever skips rows the current run already deleted, was protecting, or skip-locked — all deferred to the next run, identical to the unbounded scan's row set (verified: union of batched deletes == single delete, 0 diff). Mirrored in windmill-api-settings log_cleanup. Co-authored-by: Claude Opus 4.8 (1M context) --- ...41a9e9f2d5a56e12a9d6805ca58ff40f61614.json | 30 +++++ ...fa7cfd861f1dcad88c7ea75a0a8b014ec1f75.json | 31 +++++ ...6b2bd117ceb6bc152d2abcd45850ff6aecff9.json | 24 ---- backend/src/monitor.rs | 116 +++++++++++++----- .../windmill-api-settings/src/log_cleanup.rs | 85 +++++++++---- 5 files changed, 202 insertions(+), 84 deletions(-) create mode 100644 backend/.sqlx/query-a2d4a8aedb15e9faf0a2512fa4241a9e9f2d5a56e12a9d6805ca58ff40f61614.json create mode 100644 backend/.sqlx/query-c033a690fde04da79745e850b72fa7cfd861f1dcad88c7ea75a0a8b014ec1f75.json delete mode 100644 backend/.sqlx/query-fbe3a876efd1253d2ef086b03366b2bd117ceb6bc152d2abcd45850ff6aecff9.json diff --git a/backend/.sqlx/query-a2d4a8aedb15e9faf0a2512fa4241a9e9f2d5a56e12a9d6805ca58ff40f61614.json b/backend/.sqlx/query-a2d4a8aedb15e9faf0a2512fa4241a9e9f2d5a56e12a9d6805ca58ff40f61614.json new file mode 100644 index 0000000000..24e387a783 --- /dev/null +++ b/backend/.sqlx/query-a2d4a8aedb15e9faf0a2512fa4241a9e9f2d5a56e12a9d6805ca58ff40f61614.json @@ -0,0 +1,30 @@ +{ + "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 AND ($3::timestamptz IS NULL OR completed_at >= $3)\n ORDER BY completed_at ASC\n LIMIT $2\n FOR UPDATE SKIP LOCKED\n )\n RETURNING id, completed_at", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "completed_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8", + "Timestamptz" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "a2d4a8aedb15e9faf0a2512fa4241a9e9f2d5a56e12a9d6805ca58ff40f61614" +} diff --git a/backend/.sqlx/query-c033a690fde04da79745e850b72fa7cfd861f1dcad88c7ea75a0a8b014ec1f75.json b/backend/.sqlx/query-c033a690fde04da79745e850b72fa7cfd861f1dcad88c7ea75a0a8b014ec1f75.json new file mode 100644 index 0000000000..b2e218e511 --- /dev/null +++ b/backend/.sqlx/query-c033a690fde04da79745e850b72fa7cfd861f1dcad88c7ea75a0a8b014ec1f75.json @@ -0,0 +1,31 @@ +{ + "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 ($4::timestamptz IS NULL OR jc.completed_at >= $4)\n AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) NOT IN (\n SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL\n )\n ORDER BY jc.completed_at ASC\n LIMIT $2\n FOR UPDATE OF jc SKIP LOCKED\n )\n RETURNING id, completed_at", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "completed_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8", + "UuidArray", + "Timestamptz" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "c033a690fde04da79745e850b72fa7cfd861f1dcad88c7ea75a0a8b014ec1f75" +} diff --git a/backend/.sqlx/query-fbe3a876efd1253d2ef086b03366b2bd117ceb6bc152d2abcd45850ff6aecff9.json b/backend/.sqlx/query-fbe3a876efd1253d2ef086b03366b2bd117ceb6bc152d2abcd45850ff6aecff9.json deleted file mode 100644 index 62ec17f0f8..0000000000 --- a/backend/.sqlx/query-fbe3a876efd1253d2ef086b03366b2bd117ceb6bc152d2abcd45850ff6aecff9.json +++ /dev/null @@ -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) NOT IN (\n SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL\n )\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": "fbe3a876efd1253d2ef086b03366b2bd117ceb6bc152d2abcd45850ff6aecff9" -} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index f04646120e..fe3e3e4601 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1324,6 +1324,9 @@ pub async fn delete_expired_items(db: &DB) -> () { let cleanup_start = Instant::now(); let mut total_deleted = 0u64; let mut batch_num = 0i32; + // Watermark carried across batches so each one resumes after the rows the previous batch + // already processed instead of re-scanning the (potentially undeletable) oldest prefix. + let mut completed_at_floor: Option> = None; // Process batches until no more expired jobs or max batches reached loop { @@ -1336,14 +1339,17 @@ pub async fn delete_expired_items(db: &DB) -> () { } // Each batch runs in its own transaction to avoid long-running locks - let batch_result = delete_expired_jobs_batch(db, job_retention_secs, batch_size).await; + let batch_result = + delete_expired_jobs_batch(db, job_retention_secs, batch_size, completed_at_floor) + .await; match batch_result { - Ok(deleted_count) => { + Ok((deleted_count, max_completed_at)) => { if deleted_count == 0 { // No more expired jobs to delete break; } + completed_at_floor = max_completed_at.or(completed_at_floor); total_deleted += deleted_count as u64; batch_num += 1; } @@ -1510,12 +1516,20 @@ pub async fn check_expiring_tokens(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. +/// +/// `completed_at_floor` is the watermark from the previous batch in the same cleanup run (the +/// max `completed_at` it deleted); pass `None` for the first batch. It is re-applied as +/// `completed_at >= floor` so the scan resumes past the rows already processed instead of +/// re-walking them (see the inline comment on the DELETE for why this matters). +/// +/// Returns `(jobs deleted in this batch, max completed_at deleted)`. The caller feeds the +/// returned watermark back in as `completed_at_floor` for the next batch. async fn delete_expired_jobs_batch( db: &DB, job_retention_secs: i64, batch_size: i64, -) -> error::Result { + completed_at_floor: Option>, +) -> error::Result<(usize, Option>)> { let mut tx = db.begin().await?; // Fetch active ROOT job IDs that started before the retention period. We only care about @@ -1531,34 +1545,70 @@ async fn delete_expired_jobs_batch( .fetch_all(&mut *tx) .await?; - // Use FOR UPDATE SKIP LOCKED to avoid contention between replicas - // ORDER BY completed_at ensures we delete oldest jobs first. - // Active-root exclusion uses `NOT IN (SELECT ... unnest($3))` rather than - // `!= ALL($3)`: the subquery form lets the planner build a one-time hashed - // SubPlan and apply it as a filter on the ordered index scan, giving O(1) - // membership per candidate instead of a per-row linear array scan (which - // degrades sharply when many root jobs are active). The `u IS NOT NULL` guard - // sidesteps NOT IN's null-trap semantics ($3 holds non-null PK ids). - let deleted_jobs: Vec = 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) NOT IN ( - SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL - ) - 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 - ) - .fetch_all(&mut *tx) - .await?; + // `completed_at_floor` is a watermark carried across batches within a cleanup run: it is the + // max(completed_at) deleted by the previous batch. Re-applying it as `completed_at >= floor` + // lets each batch resume after the rows the previous batch already processed instead of + // re-scanning them. This matters when the oldest rows are undeletable (children of a + // still-active root flow): without the floor the `ORDER BY completed_at ASC` scan walks that + // same protected prefix on every batch, turning a cleanup run quadratic in prefix size. + // Floor only ever skips rows the current run already deleted, was protecting, or skip-locked — + // all correctly deferred to the next run, identical to the unbounded scan's semantics. + // + // Use FOR UPDATE SKIP LOCKED to avoid contention between replicas; ORDER BY completed_at + // deletes oldest jobs first. + let (deleted_jobs, max_completed_at) = if active_root_job_ids.is_empty() { + // Common case: no old root flow is still running, so nothing is protected and the + // v2_job join (a PK lookup per candidate) is pure overhead — skip it entirely. + let rows = sqlx::query!( + "DELETE FROM v2_job_completed + WHERE id IN ( + SELECT id FROM v2_job_completed + WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval + AND ($3::timestamptz IS NULL OR completed_at >= $3) + ORDER BY completed_at ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + RETURNING id, completed_at", + job_retention_secs, + batch_size, + completed_at_floor, + ) + .fetch_all(&mut *tx) + .await?; + let max = rows.iter().map(|r| r.completed_at).max(); + (rows.into_iter().map(|r| r.id).collect::>(), max) + } else { + // Active-root exclusion uses `NOT IN (SELECT ... unnest($3))` rather than `!= ALL($3)`: + // the subquery form lets the planner build a one-time hashed SubPlan and apply it as a + // filter on the ordered index scan, giving O(1) membership per candidate instead of a + // per-row linear array scan (which degrades sharply when many root jobs are active). The + // `u IS NOT NULL` guard sidesteps NOT IN's null-trap semantics ($3 holds non-null PK ids). + let rows = sqlx::query!( + "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 ($4::timestamptz IS NULL OR jc.completed_at >= $4) + AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) NOT IN ( + SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL + ) + ORDER BY jc.completed_at ASC + LIMIT $2 + FOR UPDATE OF jc SKIP LOCKED + ) + RETURNING id, completed_at", + job_retention_secs, + batch_size, + &active_root_job_ids, + completed_at_floor, + ) + .fetch_all(&mut *tx) + .await?; + let max = rows.iter().map(|r| r.completed_at).max(); + (rows.into_iter().map(|r| r.id).collect::>(), max) + }; let deleted_count = deleted_jobs.len(); @@ -1618,7 +1668,7 @@ async fn delete_expired_jobs_batch( tx.commit().await?; - Ok(deleted_count) + Ok((deleted_count, max_completed_at)) } async fn delete_log_files_from_disk_and_store( diff --git a/backend/windmill-api-settings/src/log_cleanup.rs b/backend/windmill-api-settings/src/log_cleanup.rs index 2264e27037..1e71029874 100644 --- a/backend/windmill-api-settings/src/log_cleanup.rs +++ b/backend/windmill-api-settings/src/log_cleanup.rs @@ -339,13 +339,15 @@ async fn cleanup_job_logs( return Ok(()); } + let mut completed_at_floor: Option> = None; loop { - let (deleted_count, rel_paths) = - delete_expired_jobs_batch(db, retention_secs, JOB_BATCH).await?; + let (deleted_count, rel_paths, max_completed_at) = + delete_expired_jobs_batch(db, retention_secs, JOB_BATCH, completed_at_floor).await?; if deleted_count == 0 { break; } + completed_at_floor = max_completed_at.or(completed_at_floor); let s3_paths: Vec = rel_paths .iter() @@ -382,7 +384,8 @@ async fn delete_expired_jobs_batch( db: &DB, job_retention_secs: i64, batch_size: i64, -) -> error::Result<(usize, Vec)> { + completed_at_floor: Option>, +) -> error::Result<(usize, Vec, Option>)> { let mut tx = db.begin().await?; let active_root_job_ids: Vec = sqlx::query_scalar!( @@ -395,33 +398,61 @@ async fn delete_expired_jobs_batch( .fetch_all(&mut *tx) .await?; - // Active-root exclusion via NOT IN (hashed SubPlan) instead of `!= ALL($3)`; - // see backend/src/monitor.rs::delete_expired_jobs_batch for the rationale. - let deleted_jobs: Vec = 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) NOT IN ( - SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL - ) - 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 - ) - .fetch_all(&mut *tx) - .await?; + // `completed_at_floor` carries a watermark across batches so each one resumes after the rows + // the previous batch processed instead of re-scanning the (potentially undeletable) oldest + // prefix; the empty-active-roots branch skips the v2_job join entirely. See + // backend/src/monitor.rs::delete_expired_jobs_batch for the full rationale. + let (deleted_jobs, max_completed_at) = if active_root_job_ids.is_empty() { + let rows = sqlx::query!( + "DELETE FROM v2_job_completed + WHERE id IN ( + SELECT id FROM v2_job_completed + WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval + AND ($3::timestamptz IS NULL OR completed_at >= $3) + ORDER BY completed_at ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + RETURNING id, completed_at", + job_retention_secs, + batch_size, + completed_at_floor, + ) + .fetch_all(&mut *tx) + .await?; + let max = rows.iter().map(|r| r.completed_at).max(); + (rows.into_iter().map(|r| r.id).collect::>(), max) + } else { + let rows = sqlx::query!( + "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 ($4::timestamptz IS NULL OR jc.completed_at >= $4) + AND COALESCE(j.root_job, j.flow_innermost_root_job, jc.id) NOT IN ( + SELECT u FROM unnest($3::uuid[]) AS u WHERE u IS NOT NULL + ) + ORDER BY jc.completed_at ASC + LIMIT $2 + FOR UPDATE OF jc SKIP LOCKED + ) + RETURNING id, completed_at", + job_retention_secs, + batch_size, + &active_root_job_ids, + completed_at_floor, + ) + .fetch_all(&mut *tx) + .await?; + let max = rows.iter().map(|r| r.completed_at).max(); + (rows.into_iter().map(|r| r.id).collect::>(), max) + }; let deleted_count = deleted_jobs.len(); if deleted_count == 0 { tx.commit().await?; - return Ok((0, Vec::new())); + return Ok((0, Vec::new(), max_completed_at)); } if let Err(e) = sqlx::query!( @@ -471,7 +502,7 @@ async fn delete_expired_jobs_batch( tx.commit().await?; - Ok((deleted_count, log_paths)) + Ok((deleted_count, log_paths, max_completed_at)) } /// Scan S3 under the `logs/` prefix for orphan log files and delete them.