Preserve subscription progress during consensus leader activation

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

@@ -602,6 +602,7 @@
     private final long targetSearchIndex;
     private final RegionProgress committedRegionProgress;
     private final String seekReason;
+    private final boolean explicitSeek;
     private final boolean previousPrefetchInitialized;
     private final long previousSeekGeneration;
     private final long targetSeekGeneration;
@@ -613,12 +614,14 @@
         final long targetSearchIndex,
         final RegionProgress committedRegionProgress,
         final String seekReason,
+        final boolean explicitSeek,
         final boolean previousPrefetchInitialized,
         final long previousSeekGeneration,
         final long targetSeekGeneration) {
       this.targetSearchIndex = targetSearchIndex;
       this.committedRegionProgress = committedRegionProgress;
       this.seekReason = seekReason;
+      this.explicitSeek = explicitSeek;
       this.previousPrefetchInitialized = previousPrefetchInitialized;
       this.previousSeekGeneration = previousSeekGeneration;
       this.targetSeekGeneration = targetSeekGeneration;
@@ -1715,7 +1718,7 @@
   private static final long REPLAY_STALL_WARNING_THRESHOLD_NANOS = TimeUnit.SECONDS.toNanos(60L);
   private static final long REPLAY_STALL_WARNING_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(30L);
   private static final int MAX_REPLAY_DIAGNOSTIC_WRITERS = 4;
 
   private static final long SLOW_PREFETCH_ROUND_THRESHOLD_MS = 10_000L;
 
   private String getPrefetchDebugId() {
@@ -4384,15 +4387,32 @@
    * denotes an empty WAL. If old files have been reclaimed, replay may start at a later index.
    */
   public void seekToBeginning() {
+    seekToBeginning(true, -1L);
+  }
+
+  private void seekToBeginning(final boolean explicitSeek, final long expectedSeekGeneration) {
     seekToResolvedPosition(
-        FIRST_CONSENSUS_SEARCH_INDEX, new RegionProgress(Collections.emptyMap()), "beginning");
+        FIRST_CONSENSUS_SEARCH_INDEX,
+        new RegionProgress(Collections.emptyMap()),
+        "beginning",
+        explicitSeek,
+        expectedSeekGeneration);
   }
 
   /**
    * Seeks to the current WAL write position. After this, only newly written data will be consumed.
    */
   public void seekToEnd() {
-    seekToResolvedPosition(nextSearchIndexAfterCurrent(), computeTailRegionProgress(), "end");
+    seekToEnd(true, -1L);
+  }
+
+  private void seekToEnd(final boolean explicitSeek, final long expectedSeekGeneration) {
+    seekToResolvedPosition(
+        nextSearchIndexAfterCurrent(),
+        computeTailRegionProgress(),
+        "end",
+        explicitSeek,
+        expectedSeekGeneration);
   }
 
   public void seekToRegionProgress(final RegionProgress regionProgress) {
@@ -4422,7 +4442,9 @@
         seekToResolvedPosition(
             replayTarget.getStartSearchIndex(),
             replayTarget.getRecoveryRegionProgress(),
-            "regionProgress");
+            "regionProgress",
+            true,
+            -1L);
         return;
       case RETRY_LATER:
       case LOCATE_MISS:
@@ -4438,12 +4460,19 @@
   }
 
   public void seekAfterRegionProgress(final RegionProgress regionProgress) {
+    seekAfterRegionProgress(regionProgress, true, -1L);
+  }
+
+  private void seekAfterRegionProgress(
+      final RegionProgress regionProgress,
+      final boolean explicitSeek,
+      final long expectedSeekGeneration) {
     if (!(consensusReqReader instanceof WALNode)) {
       LOGGER.warn(
           DataNodePipeMessages
               .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_SEEKAFTERREGIONPROGRESS_NOT_SUPPORTED_55F36BE8,
           this);
-      seekToEnd();
+      seekToEnd(explicitSeek, expectedSeekGeneration);
       return;
     }
     final WALNode walNode = (WALNode) consensusReqReader;
@@ -4464,7 +4493,9 @@
         seekToResolvedPosition(
             replayTarget.getStartSearchIndex(),
             replayTarget.getRecoveryRegionProgress(),
-            "regionProgressAfter");
+            "regionProgressAfter",
+            explicitSeek,
+            expectedSeekGeneration);
         return;
       case RETRY_LATER:
       case LOCATE_MISS:
@@ -4482,14 +4513,21 @@
   private synchronized void seekToResolvedPosition(
       final long targetSearchIndex,
       final RegionProgress committedRegionProgress,
-      final String seekReason) {
+      final String seekReason,
+      final boolean explicitSeek,
+      final long expectedSeekGeneration) {
     final PendingSeekRequest request;
 
     acquireWriteLock();
     try {
       if (isClosed || closeRequested) {
         return;
       }
+      // A user seek completed after activation captured its frontier. Its cursor and recovery
+      // progress supersede this older lookup, even if that lookup resolved to a higher checkpoint.
+      if (!explicitSeek && seekGeneration.get() != expectedSeekGeneration) {
+        return;
+      }
       // Fence old commit contexts immediately. The grouped reset itself is applied later by the
       // prefetch worker so WAL state and queue state still move under the queue's serial context.
       final boolean previousPrefetchInitialized = prefetchInitialized;
@@ -4500,6 +4538,7 @@
               targetSearchIndex,
               committedRegionProgress,
               seekReason,
+              explicitSeek,
               previousPrefetchInitialized,
               previousSeekGeneration,
               targetSeekGeneration);
@@ -4650,10 +4689,18 @@
     writerStates.clear();
     clearRecoveryWriterProgress();
     materializedProgressByWriter.clear();
-    if (Objects.nonNull(request.committedRegionProgress)
-        && !request.committedRegionProgress.getWriterPositions().isEmpty()) {
-      installRecoveryWriterProgress(request.committedRegionProgress);
-    }
+    final RegionProgress committedRegionProgress =
+        commitManager.resetState(
+            consumerGroupId,
+            topicName,
+            consensusGroupId,
+            request.committedRegionProgress,
+            request.explicitSeek,
+            progress -> {
+              if (!progress.getWriterPositions().isEmpty()) {
+                installRecoveryWriterProgress(progress);
+              }
+            });
 
     // 4. Reset WAL read position. The target has been normalized by replay lookup: if the
     // requested progress points before the first retained replayable WAL entry, it is moved to that
@@ -4668,10 +4715,7 @@
     walGapWaitStartTimeMs = 0L;
     lastWalGapWaitLogTimeMs = 0L;
 
-    // 5. Reset commit state to the writer progress immediately before the first re-delivered
-    // entry so seek/rebind resumes from the intended frontier.
-    commitManager.resetState(
-        consumerGroupId, topicName, consensusGroupId, request.committedRegionProgress);
+    // 5. Refresh WAL retention from the installed committed frontier.
     refreshCommittedWalRetentionBoundAndNotify();
 
     LOGGER.info(
@@ -4680,9 +4724,7 @@
         this,
         request.seekReason,
         request.targetSearchIndex,
-        Objects.nonNull(request.committedRegionProgress)
-            ? request.committedRegionProgress.getWriterPositions().size()
-            : 0,
+        committedRegionProgress.getWriterPositions().size(),
         seekGeneration.get());
   }
 
@@ -5864,18 +5906,26 @@
       if (Objects.isNull(pendingActivationRuntimeState)) {
         return true;
       }
-      final ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult queryResult =
-          commitManager.refreshAuthoritativeProgress(consumerGroupId, topicName, consensusGroupId);
+      final long activationSeekGeneration;
+      final ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult queryResult;
+      synchronized (this) {
+        // Wait for an already submitted user seek to finish before capturing the activation
+        // snapshot. A later user seek is detected again when the resolved lookup is submitted.
+        activationSeekGeneration = seekGeneration.get();
+        queryResult =
+            commitManager.refreshAuthoritativeProgress(
+                consumerGroupId, topicName, consensusGroupId);
+      }
       if (!queryResult.isAvailable()) {
         return false;
       }
 
       final RegionProgress authoritativeProgress = queryResult.getRegionProgress();
       if (prefetchInitialized) {
         if (authoritativeProgress.getWriterPositions().isEmpty()) {
-          seekToBeginning();
+          seekToBeginning(false, activationSeekGeneration);
         } else {
-          seekAfterRegionProgress(authoritativeProgress);
+          seekAfterRegionProgress(authoritativeProgress, false, activationSeekGeneration);
         }
       }
       pendingActivationRuntimeState = null;