数据清理

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

@@ -6,11 +6,6 @@ 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;
 
@@ -18,7 +13,6 @@ import org.slf4j.LoggerFactory;
 public class LegacyRetentionStore {
     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;
@@ -37,92 +31,74 @@ public class LegacyRetentionStore {
 
     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 validationMs = 0;
         long deleteMs = 0;
         boolean complete = false;
-        try (Connection connection = dataSource.getConnection()) {
-            connection.setAutoCommit(true);
-            if (!validated.contains(table)) {
-                validateCreatedAtIndex(connection, table);
-                validatePrimaryKey(connection, table);
-                validated.add(table);
-            }
+        String phase = "connection";
+        String reason = "stopped";
+        String sqlState = "-";
+        int errorCode = 0;
+        String error = "-";
+        try {
             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");
+            try (Connection connection = dataSource.getConnection()) {
+                connection.setAutoCommit(true);
+                phase = "validation";
+                long validationStarted = System.nanoTime();
+                try {
+                    if (!validated.contains(table)) {
+                        validateCreatedAtIndex(connection, table);
+                        validated.add(table);
                     }
+                } finally {
+                    validationMs = elapsedMs(validationStarted);
                 }
-            }
-            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.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` < ?";
+                if (stop.getAsBoolean()) return 0;
+                phase = "delete";
+                // One bounded, auto-committed DELETE. No ID query, IN-list or repeated commits.
+                String sql = "DELETE FROM " + table.qualifiedName()
+                        + " WHERE `created_at` < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY)"
+                        + " ORDER BY `created_at` LIMIT ?";
                 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++;
+                    statement.setInt(1, BATCH_SIZE);
+                    deleted = statement.executeUpdate();
                 } finally {
-                    deleteMs += elapsedMs(deleteStarted);
+                    deleteMs = elapsedMs(deleteStarted);
                 }
+                phase = "close_connection";
             }
             complete = true;
+            phase = "done";
+            reason = "ok";
             return deleted;
+        } catch (SQLException e) {
+            reason = "sql_error";
+            sqlState = e.getSQLState() == null ? "-" : e.getSQLState();
+            errorCode = e.getErrorCode();
+            error = oneLine(e.getMessage());
+            throw e;
+        } catch (RuntimeException e) {
+            reason = "runtime_error";
+            error = oneLine(e.getMessage());
+            throw e;
         } finally {
-            log.info("[LegacyRetention] table={} selected={} deleted={} chunks={} selectMs={} deleteMs={} totalMs={} complete={}",
-                    table.qualifiedName(), selected, deleted, chunks, selectMs, deleteMs, elapsedMs(started), complete);
+            String summary = "[LegacyRetention] mode=direct table={} batchSize={} retentionDays=30 deleted={} validationMs={} deleteMs={} totalMs={} complete={} reason={} phase={} sqlState={} errorCode={} error={}";
+            Object[] values = {table.qualifiedName(), BATCH_SIZE, deleted, validationMs, deleteMs,
+                    elapsedMs(started), complete, reason, phase, sqlState, errorCode, error};
+            if (reason.endsWith("error")) log.error(summary, values);
+            else log.info(summary, values);
         }
     }
 
-    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");
-                }
-            }
-        }
+    private static String oneLine(String message) {
+        return message == null ? "-" : message.replace('\n', ' ').replace('\r', ' ');
     }
 
+    private static long elapsedMs(long started) { return (System.nanoTime() - started) / 1_000_000; }
+
     private void validateCreatedAtIndex(Connection connection, Table table) throws SQLException {
         // Avoid repeatedly scanning a very large table when the required leading index is absent.
         String sql = "SELECT COUNT(*) FROM information_schema.STATISTICS s"

src/main/resources/application-dev.yml

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

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

@@ -14,10 +14,10 @@ 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. */
+/** Opt-in only: exclusively uses a guarded, disposable loopback MySQL instance. */
 @EnabledIfSystemProperty(named = "retention.mysql.port", matches = "[0-9]+")
 class LegacyRetentionMysqlIT {
-    @Test void realMySqlDeletionSafetyAndBeforeAfterBenchmark() throws Exception {
+    @Test void realMySqlDirectDeleteKeepsRecentAndBoundaryRows() throws Exception {
         String port = System.getProperty("retention.mysql.port");
         assertNotEquals("3306", port, "Never use the normal local database port");
         DriverManagerDataSource ds = new DriverManagerDataSource(
@@ -26,29 +26,30 @@ class LegacyRetentionMysqlIT {
                 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.
+                    statement.execute("SET timestamp=1792065600");
                 }
                 return conn;
             }
         };
         JdbcTemplate jdbc = new JdbcTemplate(ds);
+        assertNotNull(System.getProperty("retention.mysql.guard"));
         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("mode", "direct");
         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.");
+        report.put("description", "Local synthetic rows, 1KB payload, three secondary indexes; not a production benchmark.");
         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,"
+            // Prove tracking cleanup has no dependency on an id-named primary key.
+            String key = table == LegacyRetentionStore.Table.BID ? "id" : "record_key";
+            jdbc.execute("CREATE TABLE " + target + " (" + key + " 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),"
@@ -61,58 +62,25 @@ class LegacyRetentionMysqlIT {
                     + " (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<Double> directMs = 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);
+                long began = System.nanoTime();
+                counts.add(store.deleteBatch(table));
+                directMs.add((System.nanoTime()-began)/1_000_000.0);
                 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);
+            assertEquals(2L, jdbc.queryForObject("SELECT COUNT(*) FROM " + target + " WHERE " + key + " IN (900000001,900000002)", Long.class));
+            results.add(Map.of("table", table.name, "primaryKey", key, "deletedPerRound", counts,
+                    "directMs", directMs, "recentRowsPreserved", 60002, "remainingExpiredRows", 0));
         }
         report.put("results", results);
-        Path output = Path.of("target/retention-mysql-benchmark.json");
+        Path output = Path.of("target/retention-mysql-direct.json");
         Files.writeString(output, new ObjectMapper().writerWithDefaultPrettyPrinter().writeValueAsString(report));
-        System.out.println("MYSQL_BENCHMARK_REPORT=" + output.toAbsolutePath());
+        System.out.println("MYSQL_DIRECT_REPORT=" + output.toAbsolutePath());
     }
 }

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

@@ -1,13 +1,15 @@
 package com.adx.tencent.retention;
 
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import org.slf4j.LoggerFactory;
 import org.junit.jupiter.api.Test;
 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.*;
@@ -15,120 +17,78 @@ 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), 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");
+    final PreparedStatement metadata = mock(PreparedStatement.class), delete = mock(PreparedStatement.class);
+    final ResultSet result = mock(ResultSet.class);
 
-    LegacyRetentionStore store(int count) throws Exception {
+    LegacyRetentionStore store() 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(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));
+        when(metadata.executeQuery()).thenReturn(result);
+        when(result.next()).thenReturn(true);
+        when(result.getLong(1)).thenReturn(1L);
+        when(delete.executeUpdate()).thenReturn(50000);
         return new LegacyRetentionStore(ds);
     }
 
     @ParameterizedTest @EnumSource(LegacyRetentionStore.Table.class)
-    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);
+    void oneDeletePerBatchPreservesThirtyDayGuardAndFiftyThousandLimit(LegacyRetentionStore.Table table) throws Exception {
+        assertEquals(50000, store().deleteBatch(table));
+        verify(connection).prepareStatement("DELETE FROM " + table.qualifiedName()
+                + " WHERE `created_at` < DATE_SUB(CURRENT_TIMESTAMP(3), INTERVAL 30 DAY) ORDER BY `created_at` LIMIT ?");
+        verify(delete).setInt(1, 50000); verify(delete).setQueryTimeout(30);
+        verify(delete, times(1)).executeUpdate();
         verify(connection).setAutoCommit(true);
-        verify(ids).close(); verify(delete).close(); verify(connection).close();
-    }
-
-    @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);
+        verify(connection, times(2)).prepareStatement(anyString()); // Metadata + DELETE, no ID lookup.
+        verify(result).close(); verify(delete).close(); verify(connection).close();
     }
 
-    @Test void missingIndexBlocksDeletionAndIsRechecked() throws Exception {
-        LegacyRetentionStore store = store(1);
-        when(indexResult.getLong(1)).thenReturn(0L, 1L);
-        assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.BID));
-        verifyNoInteractions(select, delete);
-        assertEquals(1, store.deleteBatch(LegacyRetentionStore.Table.BID));
-        verify(metadata, times(2)).executeQuery();
+    @Test void emptyTableReturnsZeroWithoutExtraQueryOrPrimaryKeyRequirement() throws Exception {
+        LegacyRetentionStore store = store();
+        when(delete.executeUpdate()).thenReturn(0);
+        assertEquals(0, store.deleteBatch(LegacyRetentionStore.Table.TRACKING));
+        verify(connection, times(2)).prepareStatement(anyString());
     }
 
-    @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 missingIndexIsLoggedWithExplicitValidationFailureAndRetriedLater() throws Exception {
+        LegacyRetentionStore store = store();
+        when(result.getLong(1)).thenReturn(0L, 1L);
+        Logger logger = (Logger) LoggerFactory.getLogger(LegacyRetentionStore.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>(); appender.start(); logger.addAppender(appender);
+        try {
+            assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.TRACKING));
+            verifyNoInteractions(delete);
+            String message = appender.list.get(0).getFormattedMessage();
+            assertTrue(message.contains("mode=direct"));
+            assertTrue(message.contains("complete=false reason=sql_error phase=validation"));
+            assertTrue(message.contains("requires a date/time created_at column"));
+            assertEquals(50000, store.deleteBatch(LegacyRetentionStore.Table.TRACKING));
+        } finally { logger.detachAppender(appender); appender.stop(); }
     }
 
-    @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 deleteFailureReportsSqlStateAndVendorCodeAndCanRetry() throws Exception {
+        LegacyRetentionStore store = store();
+        when(delete.executeUpdate()).thenThrow(new SQLException("lock wait timeout", "HY000", 1205)).thenReturn(50000);
+        Logger logger = (Logger) LoggerFactory.getLogger(LegacyRetentionStore.class);
+        ListAppender<ILoggingEvent> appender = new ListAppender<>(); appender.start(); logger.addAppender(appender);
+        try {
+            assertThrows(SQLException.class, () -> store.deleteBatch(LegacyRetentionStore.Table.BID));
+            String message = appender.list.get(0).getFormattedMessage();
+            assertTrue(message.contains("reason=sql_error phase=delete"));
+            assertTrue(message.contains("sqlState=HY000 errorCode=1205 error=lock wait timeout"));
+            assertEquals(50000, store.deleteBatch(LegacyRetentionStore.Table.BID));
+            verify(connection, times(2)).close(); verify(delete, times(2)).close();
+        } finally { logger.detachAppender(appender); appender.stop(); }
     }
 
-    @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(connection, never()).rollback(); // Successful chunks were already committed.
+    @Test void lossOfLeadershipBeforeDeleteDoesNotIssueDelete() throws Exception {
+        LegacyRetentionStore store = store(); AtomicInteger checks = new AtomicInteger();
+        assertEquals(0, store.deleteBatch(LegacyRetentionStore.Table.BID, () -> checks.incrementAndGet() > 1));
+        verifyNoInteractions(delete); verify(connection).close();
     }
 
-    @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();
+    @Test void alreadyStoppedDoesNotBorrowConnection() throws Exception {
+        assertEquals(0, new LegacyRetentionStore(ds).deleteBatch(LegacyRetentionStore.Table.BID, () -> true));
+        verifyNoInteractions(ds);
     }
 }