mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-10 08:07:03 +00:00
perf(monitor): skip protected prefix in retention delete via cross-batch watermark (WIN-2088) (#9744)
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) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
29c67ced97
commit
e90b2be8fa
+30
@@ -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"
|
||||
}
|
||||
+31
@@ -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"
|
||||
}
|
||||
-24
@@ -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"
|
||||
}
|
||||
+83
-33
@@ -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<DateTime<Utc>> = 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<usize> {
|
||||
completed_at_floor: Option<DateTime<Utc>>,
|
||||
) -> error::Result<(usize, Option<DateTime<Utc>>)> {
|
||||
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<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) 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::<Vec<Uuid>>(), 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::<Vec<Uuid>>(), 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(
|
||||
|
||||
@@ -339,13 +339,15 @@ async fn cleanup_job_logs(
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let mut completed_at_floor: Option<DateTime<Utc>> = 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<ObjectPath> = 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<String>)> {
|
||||
completed_at_floor: Option<DateTime<Utc>>,
|
||||
) -> error::Result<(usize, Vec<String>, Option<DateTime<Utc>>)> {
|
||||
let mut tx = db.begin().await?;
|
||||
|
||||
let active_root_job_ids: Vec<Uuid> = 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<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) 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::<Vec<Uuid>>(), 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::<Vec<Uuid>>(), 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.
|
||||
|
||||
Reference in New Issue
Block a user