Release diff: master → previous_release

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

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(