Phoenix-7920 Add replay-side handling for direct DEGRADED_STANDBY → STANDBY_TO_ACTIVE transition (ANISTS → AISTS)

phoenix-core-server/src/main/java/org/apache/phoenix/replication/ReplicationLogDiscovery.java

@@ -104,7 +104,7 @@ public abstract class ReplicationLogDiscovery {
   protected final ReplicationLogTracker replicationLogTracker;
   protected ScheduledExecutorService scheduler;
   protected volatile boolean isRunning = false;
-  protected ReplicationRound lastRoundProcessed;
+  protected volatile ReplicationRound lastRoundProcessed;
   protected MetricsReplicationLogDiscovery metrics;
   protected long roundTimeMills;
   protected long bufferMillis;
@@ -382,16 +382,30 @@ private Optional<Path> processOneRandomFile(final List<Path> files) throws IOExc
   /** Creates a new metrics source for monitoring operations. */
   protected abstract MetricsReplicationLogDiscovery createMetricsSource();
 
+  /**
+   * Initializes lastRoundProcessed, sampling the current time up front to use as the no-files
+   * fallback (see {@link #initializeLastRoundProcessed(long)}). Sampling here - before the file
+   * scans - keeps the fallback anchored to when initialization began.
+   * @throws IOException if there's an error reading file timestamps
+   */
+  protected void initializeLastRoundProcessed() throws IOException {
+    initializeLastRoundProcessed(EnvironmentEdgeManager.currentTime());
+  }
+
   /**
    * Initializes lastRoundProcessed based on minimum timestamp from 1. In-progress files (highest
    * priority) - indicates partially processed rounds 2. New files (medium priority) - indicates
-   * unprocessed rounds waiting to be replayed 3. Current time (fallback) - used when no files
-   * exist, starts from current time The minimum timestamp is converted to a replication round using
+   * unprocessed rounds waiting to be replayed 3. fallbackCurrentTime - used when no files exist,
+   * starts from the supplied time The minimum timestamp is converted to a replication round using
    * getReplicationRoundFromEndTime(), which rounds down to the nearest round boundary to ensure we
    * start from a complete round.
+   * @param fallbackCurrentTime the timestamp to start from when no files exist; sampled by the
+   *                            caller at the start of initialization rather than re-read here, so a
+   *                            slow init (or a state transition during init) cannot push the
+   *                            starting round forward
    * @throws IOException if there's an error reading file timestamps
    */
-  protected void initializeLastRoundProcessed() throws IOException {
+  protected void initializeLastRoundProcessed(long fallbackCurrentTime) throws IOException {
     Optional<Long> minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles();
     if (minTimestampFromInProgressFiles.isPresent()) {
       LOG.info(
@@ -412,11 +426,10 @@ protected void initializeLastRoundProcessed() throws IOException {
         this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager()
           .getReplicationRoundFromEndTime(minTimestampFromNewFiles.get());
       } else {
-        long currentTime = EnvironmentEdgeManager.currentTime();
         LOG.info("Initializing lastRoundProcessed for haGroup: {} from current time {}",
-          haGroupName, currentTime);
+          haGroupName, fallbackCurrentTime);
         this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager()
-          .getReplicationRoundFromEndTime(currentTime);
+          .getReplicationRoundFromEndTime(fallbackCurrentTime);
       }
     }
   }

phoenix-core-server/src/main/java/org/apache/phoenix/replication/reader/ReplicationLogDiscoveryReplay.java

@@ -127,6 +127,13 @@ public void init() throws IOException {
           clusterType == ClusterType.LOCAL
             && HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(toState)
         ) {
+          // Unconditional set (not compareAndSet): degradation is an authoritative, fail-closed
+          // signal. Whatever the current state (SYNC, SYNCED_RECOVERY, or even NOT_INITIALIZED if a
+          // degrade lands mid-init), replay must stop advancing the sync point. There is no single
+          // valid "from" state to CAS against, so a CAS would silently drop legitimate degrade
+          // signals arriving from the other states. Contrast triggerFailoverListner below, which is
+          // deliberately conditional (compareAndSet from DEGRADED only) so it cannot clobber a
+          // healthy SYNC or an already-pending SYNCED_RECOVERY.
           replicationReplayState.set(ReplicationReplayState.DEGRADED);
           LOG.info("Cluster degraded detected for {}. replicationReplayState={}", haGroupName,
             ReplicationReplayState.DEGRADED);
@@ -151,6 +158,13 @@ public void init() throws IOException {
           clusterType == ClusterType.LOCAL
             && HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(toState)
         ) {
+          // Direct DEGRADED_STANDBY -> STANDBY_TO_ACTIVE skips the STANDBY event that normally
+          // drives recovery. If we are DEGRADED, schedule the rewind so replay() re-syncs from
+          // lastRoundInSync before shouldTriggerFailover() (which gates on SYNC) can promote.
+          // compareAndSet, not set: a listener firing while already SYNC (healthy failover) or
+          // SYNCED_RECOVERY (rewind already pending) must not clobber a good state.
+          replicationReplayState.compareAndSet(ReplicationReplayState.DEGRADED,
+            ReplicationReplayState.SYNCED_RECOVERY);
           failoverPending.set(true);
           LOG.info(
             "Failover trigger detected for {}. replicationReplayState={}. "
@@ -202,50 +216,80 @@ protected void processFile(Path path) throws IOException {
   }
 
   /**
-   * Initializes lastRoundProcessed and lastRoundInSync based on HA group state. For DEGRADED states
-   * (DEGRADED_STANDBY, DEGRADED_STANDBY_FOR_WRITER): - Sets replicationReplayState to DEGRADED -
-   * Initializes lastRoundProcessed from minimum of: in-progress files, new files, or current time -
-   * Initializes lastRoundInSync from minimum of: lastSyncStateTimeInMs (from HA Store) or minimum
-   * timestamp from IN and IN PROGRESS files - This ensures lastRoundInSync represents the last
-   * known good sync point before degradation For SYNC states (STANDBY): - Sets
-   * replicationReplayState to SYNC - Calls parent's initializeLastRoundProcessed() to initialize
-   * lastRoundProcessed - Sets lastRoundInSync equal to lastRoundProcessed (both are in sync)
+   * Initializes lastRoundProcessed and lastRoundInSync based on the persisted HA group state.
+   * <ul>
+   * <li>DEGRADED_STANDBY: sets replicationReplayState to DEGRADED and initializes both rounds from
+   * the last known good sync point via {@link #initLastRoundsFromLastSyncPoint(HAGroupStoreRecord)}
+   * (lastRoundInSync from the minimum of lastSyncStateTimeInMs and the file frontier, so it
+   * represents the last consistent point before degradation).</li>
+   * <li>STANDBY_TO_ACTIVE: a restart while already mid-failover (e.g. after a direct
+   * DEGRADED_STANDBY -&gt; STANDBY_TO_ACTIVE transition). Initializes both rounds from the last
+   * known good sync point, then sets replicationReplayState to SYNCED_RECOVERY only when a rewind
+   * is actually pending (lastRoundInSync behind lastRoundProcessed) so the first replay() rewinds
+   * before failover can promote; when the rounds are already equal it initializes directly to SYNC
+   * so promotion is not delayed by a round window.</li>
+   * <li>Other states (e.g. STANDBY): sets replicationReplayState to SYNC, delegates to the parent
+   * to initialize lastRoundProcessed, and sets lastRoundInSync equal to lastRoundProcessed.</li>
+   * </ul>
+   * When the state is STANDBY_TO_ACTIVE, failoverPending is also armed.
    * @throws IOException if there's an error reading HA group state or file timestamps
    */
   @Override
   protected void initializeLastRoundProcessed() throws IOException {
     LOG.info("Initializing last round processed for haGroup: {}", haGroupName);
+    // Sample current time BEFORE reading the HA group state, so if the group transitions (e.g.
+    // SYNC -> DEGRADED_STANDBY) during init, the starting round still anchors to when init began
+    // rather than to a later point after the state read and file scans.
+    long frontierStartTime = EnvironmentEdgeManager.currentTime();
     HAGroupStoreRecord haGroupStoreRecord = getHAGroupRecord();
-    LOG.info("Found HA Group state during initialization as {} for haGroup: {}",
-      haGroupStoreRecord.getHAGroupState(), haGroupName);
-    if (
-      HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupStoreRecord.getHAGroupState())
-    ) {
+    HAGroupStoreRecord.HAGroupState haGroupState = haGroupStoreRecord.getHAGroupState();
+    LOG.info("Found HA Group state during initialization as {} for haGroup: {}", haGroupState,
+      haGroupName);
+    // Each branch below sets the initial state with compareAndSet(NOT_INITIALIZED, ...), not a
+    // plain set: the LOCAL state listeners are subscribed in init() before this runs, so a
+    // concurrent LOCAL transition may have already advanced the state off NOT_INITIALIZED. When
+    // that happens the CAS no-ops and we deliberately keep the listener's value -- a live
+    // transition is more recent (and thus more authoritative) than the record snapshot read above.
+    if (HAGroupStoreRecord.HAGroupState.DEGRADED_STANDBY.equals(haGroupState)) {
       replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED,
         ReplicationReplayState.DEGRADED);
-      long minimumTimestampFromFiles = EnvironmentEdgeManager.currentTime();
-      Optional<Long> minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles();
-      Optional<Long> minTimestampFromNewFiles = getMinTimestampFromNewFiles();
-      if (minTimestampFromInProgressFiles.isPresent()) {
-        LOG.info("Found minimum timestamp from IN PROGRESS files as {}",
-          minTimestampFromInProgressFiles.get());
-        minimumTimestampFromFiles =
-          Math.min(minimumTimestampFromFiles, minTimestampFromInProgressFiles.get());
+      initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime);
+    } else if (HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) {
+      // Restarted while already in STANDBY_TO_ACTIVE (e.g. an RS bounce after a direct
+      // DEGRADED_STANDBY -> STANDBY_TO_ACTIVE transition). No listener fires on a fresh process, so
+      // initialize as if recovering: lastRoundInSync at the last good sync point,
+      // lastRoundProcessed
+      // from the file frontier.
+      initLastRoundsFromLastSyncPoint(haGroupStoreRecord, frontierStartTime);
+      // Enter SYNCED_RECOVERY only when there is real rewind work (lastRoundInSync behind
+      // lastRoundProcessed) so the first replay() rewinds before shouldTriggerFailover() promotes.
+      // When the rounds are equal (the common "already fully replayed, RS bounced before
+      // setHAGroupStatusToSync() completed" case) there is nothing to rewind, so initialize
+      // directly to SYNC. Otherwise the SYNCED_RECOVERY -> SYNC CAS (which only fires inside the
+      // round-processing loop) would not run until a round becomes time-eligible, delaying
+      // promotion by ~one round window.
+      ReplicationReplayState initialState =
+        lastRoundInSync.getEndTime() < lastRoundProcessed.getEndTime()
+          ? ReplicationReplayState.SYNCED_RECOVERY
+          : ReplicationReplayState.SYNC;
+      // Distinct, greppable signal that this process restarted mid-failover, and which sub-case it
+      // hit, so a restart-driven promotion is diagnosable from logs (not inferred from the generic
+      // init summary below).
+      if (initialState == ReplicationReplayState.SYNCED_RECOVERY) {
+        LOG.info(
+          "Restart into STANDBY_TO_ACTIVE with rewind pending for haGroup {}: lastRoundInSync={} "
+            + "behind lastRoundProcessed={}; entering SYNCED_RECOVERY to re-sync before promotion.",
+          haGroupName, lastRoundInSync, lastRoundProcessed);
+      } else {
+        LOG.info("Restart into STANDBY_TO_ACTIVE with nothing to rewind for haGroup {}: "
+          + "lastRoundInSync == lastRoundProcessed ({}); entering SYNC, promotion eligible "
+          + "immediately.", haGroupName, lastRoundProcessed);
       }
-      if (minTimestampFromNewFiles.isPresent()) {
-        LOG.info("Found minimum timestamp from IN files as {}", minTimestampFromNewFiles.get());
-        minimumTimestampFromFiles =
-          Math.min(minimumTimestampFromFiles, minTimestampFromNewFiles.get());
-      }
-      this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager()
-        .getReplicationRoundFromEndTime(minimumTimestampFromFiles);
-      this.lastRoundInSync =
-        replicationLogTracker.getReplicationShardDirectoryManager().getReplicationRoundFromEndTime(
-          Math.min(haGroupStoreRecord.getLastSyncStateTimeInMs(), minimumTimestampFromFiles));
+      replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED, initialState);
     } else {
       replicationReplayState.compareAndSet(ReplicationReplayState.NOT_INITIALIZED,
         ReplicationReplayState.SYNC);
-      super.initializeLastRoundProcessed();
+      super.initializeLastRoundProcessed(frontierStartTime);
       this.lastRoundInSync =
         new ReplicationRound(lastRoundProcessed.getStartTime(), lastRoundProcessed.getEndTime());
     }
@@ -254,12 +298,56 @@ protected void initializeLastRoundProcessed() throws IOException {
         + "replication replay state as {}",
       lastRoundProcessed, lastRoundInSync, replicationReplayState);
 
-    // Update the failoverPending variable during initialization
-    if (
-      HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupStoreRecord.getHAGroupState())
-    ) {
-      failoverPending.compareAndSet(false, true);
+    // Arm failoverPending during initialization when restarting mid-failover. Plain set (not
+    // compareAndSet): it starts false and this is the only initializer, so there is no prior value
+    // to guard; this also matches triggerFailoverListner, which arms the same flag with set(true).
+    if (HAGroupStoreRecord.HAGroupState.STANDBY_TO_ACTIVE.equals(haGroupState)) {
+      failoverPending.set(true);
+    }
+  }
+
+  /**
+   * Initializes {@link #lastRoundProcessed} and {@link #lastRoundInSync} from the last known good
+   * sync point. Used by the DEGRADED_STANDBY and STANDBY_TO_ACTIVE branches, which both need a
+   * rewind to the last consistent point. lastRoundProcessed is derived from the minimum timestamp
+   * across IN-PROGRESS and IN files (or {@code frontierStartTime} when no files exist);
+   * lastRoundInSync is derived from the minimum of the record's lastSyncStateTimeInMs and that file
+   * frontier, so it never sits ahead of the last synced data. When lastSyncStateTimeInMs is 0 (no
+   * known sync point), it falls back to the file frontier so lastRoundInSync collapses onto
+   * lastRoundProcessed instead of rewinding to the epoch.
+   * @param haGroupStoreRecord the persisted HA group record supplying lastSyncStateTimeInMs
+   * @param frontierStartTime  current time sampled at the start of initialization; the file
+   *                           frontier upper bound, and the sole basis when no files exist
+   */
+  private void initLastRoundsFromLastSyncPoint(HAGroupStoreRecord haGroupStoreRecord,
+    long frontierStartTime) throws IOException {
+    long minimumTimestampFromFiles = frontierStartTime;
+    Optional<Long> minTimestampFromInProgressFiles = getMinTimestampFromInProgressFiles();
+    Optional<Long> minTimestampFromNewFiles = getMinTimestampFromNewFiles();
+    if (minTimestampFromInProgressFiles.isPresent()) {
+      LOG.info("Found minimum timestamp from IN PROGRESS files as {}",
+        minTimestampFromInProgressFiles.get());
+      minimumTimestampFromFiles =
+        Math.min(minimumTimestampFromFiles, minTimestampFromInProgressFiles.get());
+    }
+    if (minTimestampFromNewFiles.isPresent()) {
+      LOG.info("Found minimum timestamp from IN files as {}", minTimestampFromNewFiles.get());
+      minimumTimestampFromFiles =
+        Math.min(minimumTimestampFromFiles, minTimestampFromNewFiles.get());
     }
+    this.lastRoundProcessed = replicationLogTracker.getReplicationShardDirectoryManager()
+      .getReplicationRoundFromEndTime(minimumTimestampFromFiles);
+    // A lastSyncStateTimeInMs of 0 means "no known sync point" (never synced, or a record that
+    // predates the field). Do NOT feed 0 into the min: getReplicationRoundFromEndTime(0) returns
+    // ReplicationRound(0, 0), which would rewind SYNCED_RECOVERY to the epoch and drive the
+    // consistency point to 0 (retain-everything). Fall back to the file frontier so lastRoundInSync
+    // collapses onto lastRoundProcessed (no rewind) when there is no known sync point.
+    long lastSyncStateTimeInMs = haGroupStoreRecord.getLastSyncStateTimeInMs();
+    long lastSyncBasis = lastSyncStateTimeInMs > 0L
+      ? Math.min(lastSyncStateTimeInMs, minimumTimestampFromFiles)
+      : minimumTimestampFromFiles;
+    this.lastRoundInSync = replicationLogTracker.getReplicationShardDirectoryManager()
+      .getReplicationRoundFromEndTime(lastSyncBasis);
   }
 
   /**