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(