数据清理

src/main/java/com/adx/tencent/retention/LegacyRetentionJob.java

@@ -64,7 +64,7 @@ public class LegacyRetentionJob implements ApplicationRunner {
             LegacyRetentionStore.Table table = tables[next];
             next = (next + 1) % tables.length;
             try {
-                int deleted = store.deleteBatch(table);
+                int deleted = store.deleteBatch(table, stop::isStopped);
                 if (deleted > 0) log.info("[LegacyRetention] table={} deleted={} retention=30days", table.qualifiedName(), deleted);
                 else log.debug("[LegacyRetention] table={} no expired rows", table.qualifiedName());
             } catch (Exception e) {

src/main/java/com/adx/tencent/retention/LegacyRetentionStore.java

@@ -6,10 +6,19 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.EnumSet;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Collections;
+import java.sql.Timestamp;
+import java.math.BigDecimal;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** Deletes only the two explicitly requested legacy tables, never media-specific tables. */
 public class LegacyRetentionStore {
-    static final int BATCH_SIZE = 5000;
+    private static final Logger log = LoggerFactory.getLogger(LegacyRetentionStore.class);
+    static final int BATCH_SIZE = 50000;
+    static final int DELETE_CHUNK_SIZE = 5000;
     enum Table {
         BID("ad_bid_events"), TRACKING("tracking_reports");
         final String name;
@@ -22,21 +31,94 @@ public class LegacyRetentionStore {
 
     public LegacyRetentionStore(DataSource dataSource) { this.dataSource = dataSource; }
 
-    public synchronized int deleteBatch(Table table) throws SQLException {
+    public int deleteBatch(Table table) throws SQLException {
+        return deleteBatch(table, () -> false);
+    }
+
+    public synchronized int deleteBatch(Table table, java.util.function.BooleanSupplier stop) throws SQLException {
+        long started = System.nanoTime();
+        int selected = 0;
+        int deleted = 0;
+        int chunks = 0;
+        long selectMs = 0;
+        long deleteMs = 0;
+        boolean complete = false;
         try (Connection connection = dataSource.getConnection()) {
-            // One small auto-committed statement per batch, with no transaction spanning the wait.
             connection.setAutoCommit(true);
             if (!validated.contains(table)) {
                 validateCreatedAtIndex(connection, table);
+                validatePrimaryKey(connection, table);
                 validated.add(table);
             }
-            String sql = "DELETE FROM " + table.qualifiedName()
-                    + " WHERE `created_at` < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY)"
-                    + " ORDER BY `created_at` LIMIT ?";
-            try (PreparedStatement statement = connection.prepareStatement(sql)) {
+            if (stop.getAsBoolean()) return 0;
+            // Freeze the database-clock cutoff for both phases, including every delete chunk.
+            Timestamp cutoff;
+            try (PreparedStatement statement = connection.prepareStatement(
+                    "SELECT DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY)")) {
+                statement.setQueryTimeout(30);
+                try (ResultSet result = statement.executeQuery()) {
+                    if (!result.next() || (cutoff = result.getTimestamp(1)) == null) {
+                        throw new SQLException("Cannot determine retention cutoff");
+                    }
+                }
+            }
+            long selectStarted = System.nanoTime();
+            List<BigDecimal> ids = new ArrayList<>(BATCH_SIZE);
+            String selectSql = "SELECT `id` FROM " + table.qualifiedName()
+                    + " WHERE `created_at` < ? ORDER BY `created_at`, `id` LIMIT ?";
+            try (PreparedStatement statement = connection.prepareStatement(selectSql)) {
                 statement.setQueryTimeout(30);
-                statement.setInt(1, BATCH_SIZE);
-                return statement.executeUpdate();
+                statement.setTimestamp(1, cutoff);
+                statement.setInt(2, BATCH_SIZE);
+                try (ResultSet result = statement.executeQuery()) {
+                    while (result.next()) ids.add(result.getBigDecimal(1));
+                }
+            } finally {
+                selectMs = elapsedMs(selectStarted);
+                selected = ids.size();
+            }
+            // Up to 50,000 rows per scheduled batch, but small auto-committed primary-key deletes.
+            // No DELETE ORDER BY sort or one oversized 50,000-ID IN statement.
+            for (int offset = 0; offset < ids.size(); offset += DELETE_CHUNK_SIZE) {
+                if (stop.getAsBoolean()) return deleted;
+                List<BigDecimal> chunk = ids.subList(offset, Math.min(offset + DELETE_CHUNK_SIZE, ids.size()));
+                String sql = "DELETE FROM " + table.qualifiedName() + " WHERE `id` IN ("
+                        + String.join(",", Collections.nCopies(chunk.size(), "?")) + ") AND `created_at` < ?";
+                long deleteStarted = System.nanoTime();
+                try (PreparedStatement statement = connection.prepareStatement(sql)) {
+                    statement.setQueryTimeout(30);
+                    for (int i = 0; i < chunk.size(); i++) statement.setBigDecimal(i + 1, chunk.get(i));
+                    statement.setTimestamp(chunk.size() + 1, cutoff);
+                    deleted += statement.executeUpdate();
+                    chunks++;
+                } finally {
+                    deleteMs += elapsedMs(deleteStarted);
+                }
+            }
+            complete = true;
+            return deleted;
+        } finally {
+            log.info("[LegacyRetention] table={} selected={} deleted={} chunks={} selectMs={} deleteMs={} totalMs={} complete={}",
+                    table.qualifiedName(), selected, deleted, chunks, selectMs, deleteMs, elapsedMs(started), complete);
+        }
+    }
+
+    private static long elapsedMs(long started) { return (System.nanoTime() - started) / 1_000_000; }
+
+    private void validatePrimaryKey(Connection connection, Table table) throws SQLException {
+        String sql = "SELECT s.COLUMN_NAME, c.DATA_TYPE FROM information_schema.STATISTICS s"
+                + " JOIN information_schema.COLUMNS c ON c.TABLE_SCHEMA=s.TABLE_SCHEMA"
+                + " AND c.TABLE_NAME=s.TABLE_NAME AND c.COLUMN_NAME=s.COLUMN_NAME"
+                + " WHERE s.TABLE_SCHEMA='adx_proxy' AND s.TABLE_NAME=? AND s.INDEX_NAME='PRIMARY'";
+        try (PreparedStatement statement = connection.prepareStatement(sql)) {
+            statement.setQueryTimeout(30);
+            statement.setString(1, table.name);
+            try (ResultSet result = statement.executeQuery()) {
+                if (!result.next() || !"id".equals(result.getString(1))
+                        || !List.of("tinyint", "smallint", "mediumint", "int", "bigint").contains(result.getString(2))
+                        || result.next()) {
+                    throw new SQLException(table.qualifiedName() + " requires a single integer primary key id; cleanup skipped");
+                }
             }
         }
     }

src/main/resources/application-dev.yml

@@ -19,7 +19,7 @@ baidu:
     read-timeout-seconds: 30
 
 adx:
-  # Only adx_proxy.ad_bid_events / tracking_reports; 5000 rows per batch, 10s between batches.
+  # Only adx_proxy.ad_bid_events / tracking_reports; 50000 rows per round in 5000-ID chunks, 10s between rounds.
   legacy-retention-enabled: false
   # --- Baidu ADX ---
   baidu-adx-endpoint: "http://adsp.baidu.com/jbssp"

src/main/resources/application-prod.yml

@@ -19,7 +19,7 @@ baidu:
     read-timeout-seconds: 30
 
 adx:
-  # Only adx_proxy.ad_bid_events / tracking_reports; 5000 rows per batch, 10s between batches.
+  # Only adx_proxy.ad_bid_events / tracking_reports; 50000 rows per round in 5000-ID chunks, 10s between rounds.
   legacy-retention-enabled: ${ADX_LEGACY_RETENTION_ENABLED:true}
   # --- Baidu ADX ---
   baidu-adx-endpoint: "http://adsp.baidu.com/jbssp"

src/main/resources/application-test.yml

@@ -19,7 +19,7 @@ baidu:
     read-timeout-seconds: 30
 
 adx:
-  # Only adx_proxy.ad_bid_events / tracking_reports; 5000 rows per batch, 10s between batches.
+  # Only adx_proxy.ad_bid_events / tracking_reports; 50000 rows per round in 5000-ID chunks, 10s between rounds.
   legacy-retention-enabled: false
   # --- Baidu ADX ---
   baidu-adx-endpoint: "http://adsp.baidu.com/jbssp"

src/test/java/com/adx/tencent/retention/LegacyRetentionJobTest.java

@@ -15,11 +15,11 @@ class LegacyRetentionJobTest {
     @Test void alternatesTablesAndWaitsTenSecondsAfterEveryBatchIncludingFailures() throws Exception {
         LegacyRetentionStore store = mock(LegacyRetentionStore.class);
         List<String> events = new ArrayList<>();
-        when(store.deleteBatch(any())).thenAnswer(i -> {
+        when(store.deleteBatch(any(), any())).thenAnswer(i -> {
             LegacyRetentionStore.Table table = i.getArgument(0);
             events.add(table.name());
             if (table == LegacyRetentionStore.Table.BID) throw new SQLException("failed table");
-            return 5000;
+            return 50000;
         });
         AtomicBoolean stop = new AtomicBoolean();
         LegacyRetentionJob job = new LegacyRetentionJob(store, mock(LeaderElection.class), (millis, signal) -> {

src/test/java/com/adx/tencent/retention/LegacyRetentionMysqlIT.java

@@ -0,0 +1,118 @@
+package com.adx.tencent.retention;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DriverManagerDataSource;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.*;
+import static org.junit.jupiter.api.Assertions.*;
+
+/** Opt-in only: connects exclusively to a guarded, disposable loopback MySQL instance. */
+@EnabledIfSystemProperty(named = "retention.mysql.port", matches = "[0-9]+")
+class LegacyRetentionMysqlIT {
+    @Test void realMySqlDeletionSafetyAndBeforeAfterBenchmark() throws Exception {
+        String port = System.getProperty("retention.mysql.port");
+        assertNotEquals("3306", port, "Never use the normal local database port");
+        DriverManagerDataSource ds = new DriverManagerDataSource(
+                "jdbc:mysql://127.0.0.1:" + port + "/?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC", "root", "") {
+            @Override public Connection getConnection() throws SQLException {
+                Connection conn = super.getConnection();
+                try (Statement statement = conn.createStatement()) {
+                    statement.execute("SET time_zone='+00:00'");
+                    statement.execute("SET timestamp=1792065600"); // Freeze both phases and the 30-day boundary.
+                }
+                return conn;
+            }
+        };
+        JdbcTemplate jdbc = new JdbcTemplate(ds);
+        assertEquals(System.getProperty("retention.mysql.guard"),
+                jdbc.queryForObject("SELECT token FROM retention_test_guard.instance", String.class));
+        assertNotNull(System.getProperty("retention.mysql.guard"));
+        jdbc.execute("CREATE DATABASE adx_proxy");
+        jdbc.execute("CREATE TABLE adx_proxy.digits (n INT PRIMARY KEY)");
+        jdbc.update("INSERT INTO adx_proxy.digits VALUES (0),(1),(2),(3),(4),(5),(6),(7),(8),(9)");
+        Map<String, Object> report = new LinkedHashMap<>();
+        report.put("mysqlVersion", jdbc.queryForObject("SELECT VERSION()", String.class));
+        report.put("bufferPoolBytes", jdbc.queryForObject("SELECT @@innodb_buffer_pool_size", Long.class));
+        report.put("rowsPerTable", 240003);
+        report.put("description", "240000 synthetic rows with 1KB payload; 180000 expired, 60000 recent; plus boundary, just-recent, unsigned-ID expired sentinels. Three secondary indexes. Local MySQL only.");
+        List<Map<String, Object>> results = new ArrayList<>();
+        LegacyRetentionStore store = new LegacyRetentionStore(ds);
+        for (LegacyRetentionStore.Table table : LegacyRetentionStore.Table.values()) {
+            String target = table.qualifiedName();
+            String baseline = "adx_proxy.baseline_" + table.name;
+            jdbc.execute("CREATE TABLE " + target + " (id BIGINT UNSIGNED PRIMARY KEY, created_at TIMESTAMP(3) NOT NULL,"
+                    + " qk VARCHAR(64), kind VARCHAR(16), payload VARCHAR(2048), KEY idx_created(created_at),"
+                    + " KEY idx_qk(qk), KEY idx_kind_created(kind,created_at)) ENGINE=InnoDB");
+            jdbc.update("INSERT INTO " + target + " SELECT n+1, IF(MOD(n,4)=0, DATE_SUB(CURRENT_TIMESTAMP(3),INTERVAL 5 DAY),"
+                    + " DATE_SUB(CURRENT_TIMESTAMP(3),INTERVAL 40 DAY)+INTERVAL MOD(n,86400) SECOND),"
+                    + " CONCAT('qk-',n), IF(MOD(n,2)=0,'click','impression'), REPEAT('x',1024)"
+                    + " FROM (SELECT a.n+10*b.n+100*c.n+1000*d.n+10000*e.n+100000*f.n AS n"
+                    + " FROM adx_proxy.digits a CROSS JOIN adx_proxy.digits b CROSS JOIN adx_proxy.digits c"
+                    + " CROSS JOIN adx_proxy.digits d CROSS JOIN adx_proxy.digits e CROSS JOIN adx_proxy.digits f) numbers WHERE n<240000");
+            jdbc.update("INSERT INTO " + target + " VALUES"
+                    + " (900000001,DATE_SUB(CURRENT_TIMESTAMP(3),INTERVAL 30 DAY),'boundary','click','keep'),"
+                    + " (900000002,DATE_SUB(CURRENT_TIMESTAMP(3),INTERVAL 30 DAY)+INTERVAL 1000 MICROSECOND,'recent','click','keep'),"
+                    + " (9223372036854775810,DATE_SUB(CURRENT_TIMESTAMP(3),INTERVAL 31 DAY),'unsigned','click','delete')");
+            jdbc.execute("CREATE TABLE " + baseline + " LIKE " + target);
+            jdbc.update("INSERT INTO " + baseline + " SELECT * FROM " + target);
+            Map<String, Object> entry = new LinkedHashMap<>();
+            entry.put("table", table.name);
+            entry.put("selectPlan", jdbc.queryForList("EXPLAIN SELECT id FROM " + target
+                    + " WHERE created_at < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY) ORDER BY created_at,id LIMIT 50000"));
+            List<Double> baselineMs = new ArrayList<>(), optimizedMs = new ArrayList<>();
+            List<Integer> counts = new ArrayList<>();
+            for (int round = 0; round < 4; round++) {
+                long began;
+                int oldCount, newCount;
+                if (table == LegacyRetentionStore.Table.BID) {
+                    began = System.nanoTime();
+                    oldCount = jdbc.update("DELETE FROM " + baseline + " WHERE created_at < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY) ORDER BY created_at LIMIT 50000");
+                    baselineMs.add((System.nanoTime()-began)/1_000_000.0);
+                    began = System.nanoTime(); newCount = store.deleteBatch(table);
+                    optimizedMs.add((System.nanoTime()-began)/1_000_000.0);
+                } else {
+                    began = System.nanoTime(); newCount = store.deleteBatch(table);
+                    optimizedMs.add((System.nanoTime()-began)/1_000_000.0);
+                    began = System.nanoTime();
+                    oldCount = jdbc.update("DELETE FROM " + baseline + " WHERE created_at < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY) ORDER BY created_at LIMIT 50000");
+                    baselineMs.add((System.nanoTime()-began)/1_000_000.0);
+                }
+                assertEquals(oldCount, newCount); assertTrue(newCount <= 50000);
+                counts.add(newCount);
+                assertEquals(60002L, jdbc.queryForObject("SELECT COUNT(*) FROM " + target
+                        + " WHERE created_at >= DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY)", Long.class));
+            }
+            assertEquals(List.of(50000,50000,50000,30001), counts);
+            assertEquals(0, store.deleteBatch(table));
+            assertEquals(60002L, jdbc.queryForObject("SELECT COUNT(*) FROM " + target, Long.class));
+            assertEquals(0L, jdbc.queryForObject("SELECT COUNT(*) FROM " + target + " WHERE id=9223372036854775810", Long.class));
+            assertEquals(2L, jdbc.queryForObject("SELECT COUNT(*) FROM " + target + " WHERE id IN (900000001,900000002)", Long.class));
+            entry.put("deletedPerRound", counts); entry.put("baselineMs", baselineMs); entry.put("optimizedMs", optimizedMs);
+            entry.put("recentRowsPreserved", 60002); entry.put("remainingExpiredRows", 0);
+            // Simulate another connection changing a selected old row to a recent timestamp.
+            jdbc.update("INSERT INTO " + target + " VALUES (900000003,DATE_SUB(CURRENT_TIMESTAMP(3),INTERVAL 40 DAY),'updated-row','click','keep')");
+            java.util.concurrent.atomic.AtomicInteger checks = new java.util.concurrent.atomic.AtomicInteger();
+            assertEquals(0, store.deleteBatch(table, () -> {
+                if (checks.incrementAndGet() == 2) {
+                    jdbc.update("UPDATE " + target + " SET created_at=CURRENT_TIMESTAMP(3) WHERE id=900000003");
+                }
+                return false;
+            }));
+            assertEquals(1L, jdbc.queryForObject("SELECT COUNT(*) FROM " + target + " WHERE id=900000003", Long.class));
+            entry.put("rowUpdatedAfterSelectionPreserved", true);
+            results.add(entry);
+        }
+        report.put("results", results);
+        Path output = Path.of("target/retention-mysql-benchmark.json");
+        Files.writeString(output, new ObjectMapper().writerWithDefaultPrettyPrinter().writeValueAsString(report));
+        System.out.println("MYSQL_BENCHMARK_REPORT=" + output.toAbsolutePath());
+    }
+}

src/test/java/com/adx/tencent/retention/LegacyRetentionStoreTest.java

@@ -5,6 +5,9 @@ import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 import javax.sql.DataSource;
 import java.sql.*;
+import java.math.BigDecimal;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicBoolean;
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
@@ -12,49 +15,120 @@ import static org.mockito.Mockito.*;
 class LegacyRetentionStoreTest {
     final DataSource ds = mock(DataSource.class);
     final Connection connection = mock(Connection.class);
-    final PreparedStatement metadata = mock(PreparedStatement.class);
-    final PreparedStatement delete = mock(PreparedStatement.class);
-    final ResultSet result = mock(ResultSet.class);
+    final PreparedStatement metadata = mock(PreparedStatement.class), primary = mock(PreparedStatement.class),
+            clock = mock(PreparedStatement.class), select = mock(PreparedStatement.class), delete = mock(PreparedStatement.class);
+    final ResultSet indexResult = mock(ResultSet.class), primaryResult = mock(ResultSet.class),
+            clockResult = mock(ResultSet.class), ids = mock(ResultSet.class);
+    final Timestamp cutoff = Timestamp.valueOf("2026-09-15 12:00:00");
 
-    LegacyRetentionStore store(boolean indexExists) throws Exception {
+    LegacyRetentionStore store(int count) throws Exception {
         when(ds.getConnection()).thenReturn(connection);
         when(connection.prepareStatement(startsWith("SELECT COUNT(*)"))).thenReturn(metadata);
+        when(connection.prepareStatement(startsWith("SELECT s.COLUMN_NAME"))).thenReturn(primary);
+        when(connection.prepareStatement(startsWith("SELECT DATE_SUB"))).thenReturn(clock);
+        when(connection.prepareStatement(startsWith("SELECT `id`"))).thenReturn(select);
         when(connection.prepareStatement(startsWith("DELETE FROM"))).thenReturn(delete);
-        when(metadata.executeQuery()).thenReturn(result);
-        when(result.next()).thenReturn(true);
-        when(result.getLong(1)).thenReturn(indexExists ? 1L : 0L);
-        when(delete.executeUpdate()).thenReturn(5000);
+        when(metadata.executeQuery()).thenReturn(indexResult);
+        when(indexResult.next()).thenReturn(true);
+        when(indexResult.getLong(1)).thenReturn(1L);
+        when(primary.executeQuery()).thenReturn(primaryResult);
+        when(primaryResult.next()).thenReturn(true, false);
+        when(primaryResult.getString(1)).thenReturn("id");
+        when(primaryResult.getString(2)).thenReturn("bigint");
+        when(clock.executeQuery()).thenReturn(clockResult);
+        when(clockResult.next()).thenReturn(true);
+        when(clockResult.getTimestamp(1)).thenReturn(cutoff);
+        when(select.executeQuery()).thenAnswer(i -> {
+            AtomicInteger row = new AtomicInteger();
+            when(ids.next()).thenAnswer(x -> row.incrementAndGet() <= count);
+            // Noncontiguous unsigned IDs: preserve exact IDs without ranges or signed-long truncation.
+            when(ids.getBigDecimal(1)).thenAnswer(x -> new BigDecimal("9223372036854775808").add(BigDecimal.valueOf(row.get() * 2L)));
+            return ids;
+        });
+        when(delete.executeUpdate()).thenReturn(Math.min(count, 5000));
         return new LegacyRetentionStore(ds);
     }
 
     @ParameterizedTest @EnumSource(LegacyRetentionStore.Table.class)
-    void onlyDeletesExpiredRowsInExactTableWithBoundedAutoCommittedBatch(LegacyRetentionStore.Table table) throws Exception {
-        assertEquals(5000, store(true).deleteBatch(table));
-        verify(connection).prepareStatement("DELETE FROM `adx_proxy`.`" + table.name + "`"
-                + " WHERE `created_at` < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY)"
-                + " ORDER BY `created_at` LIMIT ?");
-        verify(metadata).setString(1, table.name);
-        verify(delete).setInt(1, 5000);
-        verify(delete).setQueryTimeout(30);
+    void selectsBoundedIdsAndDeletesOnlyThoseIdsWithSameDateGuard(LegacyRetentionStore.Table table) throws Exception {
+        assertEquals(2, store(2).deleteBatch(table));
+        verify(connection).prepareStatement("SELECT `id` FROM " + table.qualifiedName()
+                + " WHERE `created_at` < ? ORDER BY `created_at`, `id` LIMIT ?");
+        verify(connection).prepareStatement("SELECT DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY)");
+        verify(clock).executeQuery();
+        verify(select).setTimestamp(1, cutoff); verify(select).setInt(2, 50000);
+        verify(connection).prepareStatement("DELETE FROM " + table.qualifiedName() + " WHERE `id` IN (?,?) AND `created_at` < ?");
+        verify(delete).setBigDecimal(1, new BigDecimal("9223372036854775810"));
+        verify(delete).setBigDecimal(2, new BigDecimal("9223372036854775812"));
+        verify(delete).setTimestamp(3, cutoff); verify(delete).setQueryTimeout(30);
         verify(connection).setAutoCommit(true);
-        verify(delete).close(); verify(result).close(); verify(metadata).close(); verify(connection).close();
+        verify(ids).close(); verify(delete).close(); verify(connection).close();
     }
 
-    @Test void missingIndexBlocksDeletionAndIsRecheckedNextTime() throws Exception {
-        LegacyRetentionStore store = store(false);
+    @Test void largeRoundUsesSmallAutoCommittedDeletes() throws Exception {
+        assertEquals(50000, store(50000).deleteBatch(LegacyRetentionStore.Table.BID));
+        verify(delete, times(10)).executeUpdate();
+        verify(delete, times(10)).setTimestamp(5001, cutoff);
+        verify(select, times(1)).executeQuery();
+    }
+
+    @Test void partialLastChunkUsesOnlyRemainingIds() throws Exception {
+        LegacyRetentionStore store = store(5001);
+        when(delete.executeUpdate()).thenReturn(5000, 1);
+        assertEquals(5001, store.deleteBatch(LegacyRetentionStore.Table.BID));
+        verify(delete).setTimestamp(5001, cutoff); verify(delete).setTimestamp(2, cutoff);
+        verify(delete, times(2)).executeUpdate();
+    }
+
+    @Test void emptyCandidatesNeverIssueDelete() throws Exception {
+        assertEquals(0, store(0).deleteBatch(LegacyRetentionStore.Table.TRACKING));
+        verifyNoInteractions(delete);
+    }
+
+    @Test void missingIndexBlocksDeletionAndIsRechecked() throws Exception {
+        LegacyRetentionStore store = store(1);
+        when(indexResult.getLong(1)).thenReturn(0L, 1L);
         assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.BID));
-        verify(delete, never()).executeUpdate();
-        when(result.getLong(1)).thenReturn(1L);
-        assertEquals(5000, store.deleteBatch(LegacyRetentionStore.Table.BID));
+        verifyNoInteractions(select, delete);
+        assertEquals(1, store.deleteBatch(LegacyRetentionStore.Table.BID));
         verify(metadata, times(2)).executeQuery();
     }
 
-    @Test void failedDeleteClosesConnectionAndNextBatchCanRetry() throws Exception {
-        LegacyRetentionStore store = store(true);
-        when(delete.executeUpdate()).thenThrow(new SQLException("timeout")).thenReturn(0);
-        assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.TRACKING));
-        assertEquals(0, store.deleteBatch(LegacyRetentionStore.Table.TRACKING));
+    @Test void compositePrimaryKeyBlocksDeletion() throws Exception {
+        LegacyRetentionStore store = store(1);
+        when(primaryResult.next()).thenReturn(true, true);
+        assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.BID));
+        verifyNoInteractions(select, delete);
+    }
+
+    @Test void failedChunkClosesResourcesAndDoesNotRunRemainingChunks() throws Exception {
+        LegacyRetentionStore store = store(10001);
+        when(delete.executeUpdate()).thenReturn(5000).thenThrow(new SQLException("timeout"));
+        assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.BID));
+        verify(delete, times(2)).executeUpdate(); verify(delete, times(2)).close(); verify(connection).close();
+    }
+
+    @Test void nextRoundReselectsRemainingIdsAfterPartialFailure() throws Exception {
+        LegacyRetentionStore store = store(5001);
+        when(delete.executeUpdate()).thenReturn(5000).thenThrow(new SQLException("timeout"));
+        assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.BID));
+        ResultSet remaining = mock(ResultSet.class);
+        when(remaining.next()).thenReturn(true, false);
+        when(remaining.getBigDecimal(1)).thenReturn(new BigDecimal("9223372036854785810"));
+        when(select.executeQuery()).thenReturn(remaining);
+        doReturn(1).when(delete).executeUpdate();
+        assertEquals(1, store.deleteBatch(LegacyRetentionStore.Table.BID));
+        verify(select, times(2)).executeQuery();
+        verify(delete, times(3)).executeUpdate();
         verify(connection, times(2)).close();
-        verify(delete, times(2)).close();
+        verify(connection, never()).rollback(); // Successful chunks were already committed.
+    }
+
+    @Test void lostLeadershipStopsBetweenChunks() throws Exception {
+        LegacyRetentionStore store = store(10000);
+        AtomicBoolean stop = new AtomicBoolean();
+        when(delete.executeUpdate()).thenAnswer(i -> { stop.set(true); return 5000; });
+        assertEquals(5000, store.deleteBatch(LegacyRetentionStore.Table.BID, stop::get));
+        verify(delete, times(1)).executeUpdate();
     }
 }