Fix consensus subscription WAL retention and replay cursor

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java

@@ -2404,7 +2404,7 @@
         && LOGGER.isWarnEnabled()) {
       LOGGER.warn(
           DataNodePipeMessages
               .LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98,
           TimeUnit.NANOSECONDS.toMillis(stalledNanos),
           this,
           getReplayStallDiagnostics());
@@ -2466,7 +2466,7 @@
     return result;
   }
 
   private List<Map<String, Object>> sampleWriterDiagnostics(
       final Set<Integer> writerNodeIds, final Map<Integer, SafeFrontier> safeFrontiers) {
     final List<Map<String, Object>> samples = new ArrayList<>();
     final Set<Integer> sampledWriterNodeIds = new LinkedHashSet<>();
@@ -4791,14 +4791,19 @@
         commitManager.getCommittedRegionProgress(consumerGroupId, topicName, consensusGroupId);
 
     synchronized (committedRetentionLock) {
+      // Unchanged ACK progress can reuse a live boundary, but a roll must recheck the sealed file.
       if (Objects.equals(lastCommittedProgressForRetention, committedRegionProgress)
+          && consensusReqReader instanceof WALNode
+          && committedRetainedMinVersionId
+              == ((WALNode) consensusReqReader).getCurrentWALFileVersion()
           && Objects.nonNull(lastRetainedWalFileForRetention)
           && lastRetainedWalFileForRetention.exists()) {
         return false;
       }
 
       // The previous boundary still applies while its sealed requirement is uncovered, or while
-      // it is the current live WAL file. Both cases require monotonically advancing progress.
+      // it is the current live WAL file. Recheck even unchanged progress: a roll may have sealed
+      // the formerly live file after its last ACK, making that file reclaimable without new writes.
       if (canReuseCommittedWalRetentionBound(committedRegionProgress)) {
         lastCommittedProgressForRetention = committedRegionProgress;
         return false;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java

@@ -321,7 +321,10 @@ public void advanceTo(
       return;
     }
 
-    final int firstFileToSkip = Math.max(0, currentFileIndex);
+    // Without a reader (including a deferred one), currentFileIndex is the predecessor already
+    // passed by replay or a previous fast-forward. It may have been reclaimed in the meantime.
+    final int firstFileToSkip =
+        Math.max(0, currentReaderVersionId < 0 ? currentFileIndex + 1 : currentFileIndex);
     try {
       for (int fileIndex = firstFileToSkip; fileIndex < targetFileIndex; fileIndex++) {
         if (!isWalFileCovered(fileIndex, writerProgressCoverage)) {
@@ -351,10 +354,14 @@ private void refreshForNewerLiveWalFile() {
   }
 
   private int locateTargetFile(final long targetSearchIndex) {
-    if (walFiles.length == 0
-        || (liveWalNode != null && targetSearchIndex > liveWalNode.getCurrentSearchIndex())) {
+    if (walFiles.length == 0) {
       return -1;
     }
+    // Pending delivery may have advanced one past the live tail. Still realign to its file after
+    // verifying coverage of the sealed prefix, before committed history can be reclaimed.
+    if (liveWalNode != null && targetSearchIndex > liveWalNode.getCurrentSearchIndex()) {
+      return walFiles.length - 1;
+    }
     return WALFileUtils.binarySearchFileBySearchIndex(walFiles, targetSearchIndex);
   }
 

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java

@@ -973,9 +973,8 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException {
           new File(
               walDirectory,
               WALFileUtils.getLogFileName(3L, 4L, WALFileStatus.CONTAINS_SEARCH_INDEX));
-      try (WALWriter ignored = new WALWriter(nextLiveWal, WALFileVersion.V3)) {
-        // Rolling v2 makes its previously live metadata a sealed retention requirement.
-      }
+      writeWalMetadata(nextLiveWal, 4L, 8L, 1008L, 7);
+      // Rolling v2 makes its previously live metadata a sealed retention requirement.
       currentWalVersion.set(3L);
       committedProgress.set(
           new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(1006L, 6L))));
@@ -990,9 +989,27 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException {
       assertEquals(3L, committedBound.getLong(queue));
 
       committedProgress.set(
-          new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(1001L, 1L))));
+          new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(1008L, 8L))));
+      assertFalse((Boolean) refreshRetentionBound.invoke(queue));
+      assertEquals(3, progressWalReaderOpenCount.get());
+      final File finalLiveWal =
+          new File(
+              walDirectory,
+              WALFileUtils.getLogFileName(4L, 5L, WALFileStatus.CONTAINS_SEARCH_INDEX));
+      try (WALWriter ignored = new WALWriter(finalLiveWal, WALFileVersion.V3)) {
+        // No new ACK follows this roll: the same committed progress must release sealed v3.
+      }
+      currentWalVersion.set(4L);
       assertTrue((Boolean) refreshRetentionBound.invoke(queue));
       assertEquals(4, progressWalReaderOpenCount.get());
+      assertEquals(4L, committedBound.getLong(queue));
+      assertFalse((Boolean) refreshRetentionBound.invoke(queue));
+      assertEquals(4, progressWalReaderOpenCount.get());
+
+      committedProgress.set(
+          new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(1001L, 1L))));
+      assertTrue((Boolean) refreshRetentionBound.invoke(queue));
+      assertEquals(5, progressWalReaderOpenCount.get());
       assertEquals(0L, committedBound.getLong(queue));
       assertEquals(0L, retentionSupplier.get().getAsLong());
     } finally {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java

@@ -26,6 +26,7 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode;
 import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
 import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader;
 import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion;
 import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData;
 import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter;
@@ -1250,6 +1251,33 @@ public void testAdvanceToSkipsOnlyFullyCoveredWalFiles() throws Exception {
         assertTrue(iterator.hasNext());
         assertEquals(6L, iterator.next().getSearchIndex());
       }
+
+      final WALNode walNode = mock(WALNode.class);
+      when(walNode.getLogDirectory()).thenReturn(dir.toFile());
+      when(walNode.getCurrentWALFileVersion()).thenReturn(2L);
+      when(walNode.getCurrentSearchIndex()).thenReturn(7L);
+      try (ProgressWALReader reader = new ProgressWALReader(targetWal)) {
+        when(walNode.getWALMetaDataSnapshot(2L)).thenReturn(reader.getMetaData());
+      }
+      // A pending-path cursor can advance one past the live tail. Local search indexes still
+      // cannot justify skipping an uncovered follower with no local search index.
+      try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) {
+        iterator.advanceTo(8L, (physicalTime, nodeId, localSeq) -> nodeId == 7);
+        assertTrue(iterator.hasNext());
+        assertEquals(8, iterator.next().getNodeId());
+      }
+      try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) {
+        iterator.advanceTo(8L, (physicalTime, nodeId, localSeq) -> true);
+        // Committed/materialized history may now be reclaimed before the next WAL probe. The
+        // iterator must already be positioned beyond those files, without reporting data loss.
+        Files.delete(localWal.toPath());
+        Files.delete(followerWal.toPath());
+        for (int probe = 0; probe < 3; probe++) {
+          assertFalse(iterator.hasNext());
+        }
+        assertEquals(0, iterator.getSkippedBrokenWalFileCount());
+        assertFalse(iterator.isWaitingForReadableWal());
+      }
     } finally {
       Files.deleteIfExists(localWal.toPath());
       Files.deleteIfExists(followerWal.toPath());
@@ -1258,6 +1286,49 @@ public void testAdvanceToSkipsOnlyFullyCoveredWalFiles() throws Exception {
     }
   }
 
+  @Test
+  public void testRepeatedAdvanceDoesNotRecheckReclaimedPrefix() throws Exception {
+    final Path dir = Files.createTempDirectory("progress-wal-iterator-reclaimed-prefix");
+    final File historyWal =
+        dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX))
+            .toFile();
+    final File middleWal =
+        dir.resolve(WALFileUtils.getLogFileName(1, 2, WALFileStatus.CONTAINS_SEARCH_INDEX))
+            .toFile();
+    final File targetWal =
+        dir.resolve(WALFileUtils.getLogFileName(2, 4, WALFileStatus.CONTAINS_SEARCH_INDEX))
+            .toFile();
+    try {
+      try (WALWriter writer = new WALWriter(historyWal, WALFileVersion.V3)) {
+        writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L));
+        writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L));
+      }
+      try (WALWriter writer = new WALWriter(middleWal, WALFileVersion.V3)) {
+        writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L));
+        writer.write(searchableEntry(4L), singleEntryMeta(19, 4L, 1L, 400L, 7, 4L));
+      }
+      try (WALWriter writer = new WALWriter(targetWal, WALFileVersion.V3)) {
+        writer.write(searchableEntry(5L), singleEntryMeta(19, 5L, 1L, 500L, 7, 5L));
+      }
+      try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) {
+        iterator.advanceTo(3L, (physicalTime, nodeId, localSeq) -> true);
+        Files.delete(historyWal.toPath());
+        // No new WAL roll refreshes the file list between these pending-path advances.
+        iterator.advanceTo(5L, (physicalTime, nodeId, localSeq) -> true);
+        Files.delete(middleWal.toPath());
+        assertTrue(iterator.hasNext());
+        assertEquals(5L, iterator.next().getSearchIndex());
+        assertFalse(iterator.hasSkippedBrokenWalFiles());
+        assertFalse(iterator.isWaitingForReadableWal());
+      }
+    } finally {
+      Files.deleteIfExists(historyWal.toPath());
+      Files.deleteIfExists(middleWal.toPath());
+      Files.deleteIfExists(targetWal.toPath());
+      Files.deleteIfExists(dir);
+    }
+  }
+
   @Test
   public void testAdvanceToKeepsIncompleteWriterMetadata() throws Exception {
     final Path dir = Files.createTempDirectory("progress-wal-iterator-incomplete-progress");