From 4ebf288619402b85bebbc01671b8804e595029df Mon Sep 17 00:00:00 2001 From: Thomas Marchand Date: Fri, 3 Jul 2026 07:45:49 +0100 Subject: [PATCH] fix: batch batch_index TTL deletes and index received_at The cleanup task deleted all expired batch_index rows in a single statement. With a backlog (first enable, or downtime) that becomes one giant transaction cascading millions of module_dispatches rows. Delete in 10k chunks instead, and add an index on received_at so the expiry scan doesn't seq-scan the table. --- schema.sql | 3 +++ src/object_store_cleanup.rs | 23 +++++++++++++++++++++-- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/schema.sql b/schema.sql index 076c27c..8ba6aab 100644 --- a/schema.sql +++ b/schema.sql @@ -146,6 +146,9 @@ create index if not exists idx_batch_index_server_time on public.batch_index (server_id, received_at desc); create index if not exists idx_batch_index_session on public.batch_index (session_id, received_at desc); +-- Used by the TTL cleanup task (object_store_cleanup.rs) to find expired rows. +create index if not exists idx_batch_index_received_at + on public.batch_index (received_at); -------------------------------------------------------------------------------- -- FINDINGS: detections/alerts produced by processors diff --git a/src/object_store_cleanup.rs b/src/object_store_cleanup.rs index 44d4eef..6d1a780 100644 --- a/src/object_store_cleanup.rs +++ b/src/object_store_cleanup.rs @@ -94,11 +94,30 @@ async fn cleanup_batch_index( return Ok(count.max(0) as u64); } - let res = sqlx::query("delete from public.batch_index where received_at < $1") + // Delete in chunks so a large backlog (e.g. first enable, or the service + // being down for a while) never turns into one giant transaction that + // locks batch_index and cascades millions of module_dispatches rows. + const CHUNK: i64 = 10_000; + let mut total: u64 = 0; + loop { + let res = sqlx::query( + "delete from public.batch_index where id in ( + select id from public.batch_index + where received_at < $1 + order by received_at + limit $2)", + ) .bind(cutoff) + .bind(CHUNK) .execute(&state.db) .await?; - Ok(res.rows_affected()) + let deleted = res.rows_affected(); + total += deleted; + if deleted < CHUNK as u64 { + break; + } + } + Ok(total) } async fn cleanup_local_store(