[Subscription] Preserve writer order and back off on blocked safe frontiers

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTracker.java

@@ -107,6 +107,39 @@ public synchronized Map<Integer, Long> snapshotEffectiveSafePts() {
     return Collections.unmodifiableMap(snapshot);
   }
 
+  /**
+   * Returns each safe time together with the applied sequence it covers. A subscription may use the
+   * safe time only after consuming that sequence; applying records to the state machine does not
+   * imply that an independent WAL reader has consumed them.
+   */
+  public synchronized Map<Integer, SafeFrontier> snapshotSafeFrontiers() {
+    final Map<Integer, SafeFrontier> snapshot = new HashMap<>();
+    for (final Map.Entry<Integer, WriterFrontierState> entry : states.entrySet()) {
+      snapshot.put(
+          entry.getKey(),
+          new SafeFrontier(entry.getValue().effectiveSafePt, entry.getValue().appliedLocalSeq));
+    }
+    return Collections.unmodifiableMap(snapshot);
+  }
+
+  public static final class SafeFrontier {
+    private final long physicalTime;
+    private final long localSeq;
+
+    private SafeFrontier(final long physicalTime, final long localSeq) {
+      this.physicalTime = physicalTime;
+      this.localSeq = localSeq;
+    }
+
+    public long getPhysicalTime() {
+      return physicalTime;
+    }
+
+    public long getLocalSeq() {
+      return localSeq;
+    }
+  }
+
   private void promotePendingIfReady(final WriterFrontierState state) {
     if (!state.hasPendingSafeTime()) {
       return;

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTrackerTest.java

@@ -21,10 +21,30 @@
 
 import org.junit.Test;
 
+import java.util.Map;
+
 import static org.junit.Assert.assertEquals;
 
 public class WriterSafeFrontierTrackerTest {
 
+  @Test
+  public void testSafeFrontierSnapshotCarriesAppliedSequenceAndIsDetached() {
+    final WriterSafeFrontierTracker tracker = new WriterSafeFrontierTracker();
+    tracker.recordAppliedProgress(100L, 7, 10L);
+    tracker.observePendingWriterSafeTimeBarrier(130L, 7, 20L);
+    final Map<Integer, WriterSafeFrontierTracker.SafeFrontier> before =
+        tracker.snapshotSafeFrontiers();
+    assertEquals(100L, before.get(7).getPhysicalTime());
+    assertEquals(10L, before.get(7).getLocalSeq());
+
+    tracker.recordAppliedProgress(125L, 7, 20L);
+    final WriterSafeFrontierTracker.SafeFrontier after = tracker.snapshotSafeFrontiers().get(7);
+    assertEquals(130L, after.getPhysicalTime());
+    assertEquals(20L, after.getLocalSeq());
+    assertEquals(100L, before.get(7).getPhysicalTime());
+    assertEquals(10L, before.get(7).getLocalSeq());
+  }
+
   @Test
   public void testPendingWriterSafeTimeBarrierPromotesWhenBarrierIsApplied() {
     final WriterSafeFrontierTracker tracker = new WriterSafeFrontierTracker();