Persist local_last_revision in storeLog even while shutting down (#1313)

Motivation:
- After a cluster reboot, a replica entered read-only mode while replaying the latest ZooKeeper log: the mirror PushAsIsCommand replayed at a revision whose effect was already in the local data, so it failed with a RedundantChangeException ("changes did not change anything ...").
- Root cause: ZooKeeperCommandExecutor.doStop() sets listenerInfo to null at the very beginning, before draining the in-flight blockingExecute via shutdown(executor). storeLog() persisted local_last_revision only when listenerInfo != null. When a command was being originated at shutdown time, its change had already been applied to local data and the ZooKeeper log node was still created, but the local_last_revision update was silently skipped. This left the local data ahead of both progress files (git HEAD at the new revision while last_revision and local_last_revision lagged by one).
- On the next start-up, replay re-executed that already-applied revision. Because the mirror command is an unnormalized PushAsIsCommand whose base revision is Revision.HEAD, re-applying it against the already-advanced head produced an empty diff and threw RedundantChangeException, tripping read-only mode. This was the gap introduced together with local_last_revision in #1303.

Modifications:
- Decouple the local_last_revision disk write from the listenerInfo guard in storeLog(): always call updateLocalLastAppliedRevision(revision), and gate only the in-memory ListenerInfo.localLastAppliedRevision update on listenerInfo.

Result:
- A command being originated while the replica is shutting down now durably records its progress in local_last_revision, so the data and the progress files stay consistent and the replica no longer re-applies an already-applied revision (and no longer enters read-only mode) on the next start-up.

Author: minwoox

server/src/main/java/com/linecorp/centraldogma/server/internal/replication/ZooKeeperCommandExecutor.java

@@ -376,6 +376,7 @@ private static final class ListenerInfo {
         }
     }
 
+    @Nullable
     private volatile ListenerInfo listenerInfo;
 
     public ZooKeeperCommandExecutor(ZooKeeperReplicationConfig cfg,
@@ -1096,9 +1097,13 @@ private long storeLog(ReplicationLog<?> log) {
                 revision = revisionFromPath(logPath);
 
                 final ListenerInfo info = listenerInfo;
-                if (info != null && revision > info.localLastAppliedRevision) {
+                if (info != null) {
+                    if (revision > info.localLastAppliedRevision) {
+                        updateLocalLastAppliedRevision(revision);
+                        info.localLastAppliedRevision = revision;
+                    }
+                } else if (revision > getLocalLastAppliedRevision()) {
                     updateLocalLastAppliedRevision(revision);
-                    info.localLastAppliedRevision = revision;
                 }
             }
 

server/src/test/java/com/linecorp/centraldogma/server/internal/replication/ZooKeeperCommandExecutorTest.java

@@ -36,6 +36,7 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -704,6 +705,79 @@ void testLocalLastRevisionParseFailureFailsStartup() throws Exception {
         }
     }
 
+    /**
+     * Regression test for the shutdown race that left {@code local_last_revision} behind the local data
+     * (the bug introduced by #1303 and observed as a {@code RedundantChangeException} replay failure).
+     *
+     * <p>{@code doStop()} sets {@code listenerInfo} to {@code null} before draining the in-flight
+     * {@code blockingExecute}. If a command is being originated at that moment, its storage side effect has
+     * already been applied and {@code storeLog} still creates the ZooKeeper log node — but the
+     * {@code local_last_revision} update used to be gated on {@code listenerInfo != null} and was therefore
+     * silently skipped. On the next start-up the replica re-applied that already-applied revision and, for a
+     * mirror {@link PushAsIsCommand} (whose base revision is {@code HEAD}), failed with a
+     * {@code RedundantChangeException} and entered read-only mode.
+     *
+     * <p>This test parks an originating command inside the delegate, stops the executor so that
+     * {@code listenerInfo} becomes {@code null}, then lets the command reach {@code storeLog} and asserts
+     * that {@code local_last_revision} is still persisted.
+     */
+    @Test
+    void localLastRevisionPersistedWhenStoreLogRacesShutdown() throws Exception {
+        final CountDownLatch executeEntered = new CountDownLatch(1);
+        final CountDownLatch proceed = new CountDownLatch(1);
+
+        final Supplier<Function<Command<?>, CompletableFuture<?>>> delegateSupplier = () -> {
+            final Function<Command<?>, CompletableFuture<?>> base = newMockDelegate();
+            return command -> {
+                if (command != null && command.type() == CommandType.CREATE_REPOSITORY) {
+                    // Park the origination inside the delegate so that we can stop the executor while this
+                    // command is in flight (after delegate.execute() but before storeLog()).
+                    executeEntered.countDown();
+                    return CompletableFuture.supplyAsync(() -> {
+                        try {
+                            proceed.await();
+                        } catch (InterruptedException e) {
+                            throw new RuntimeException(e);
+                        }
+                        return null;
+                    }, CommonPools.blockingTaskExecutor());
+                }
+                return base.apply(command);
+            };
+        };
+
+        try (Cluster cluster = Cluster.of(delegateSupplier)) {
+            final Replica self = cluster.get(0); // replicaId=1, the originator we will stop mid-store.
+
+            // Warm up with two self-originated commands so local_last_revision advances to 1.
+            self.commandExecutor().execute(Command.createProject(Author.SYSTEM, "p1")).join();
+            self.commandExecutor().execute(Command.createProject(Author.SYSTEM, "p2")).join();
+            await().untilAsserted(() -> assertThat(self.localLastAppliedRevision()).isEqualTo(1L));
+
+            // Originate a third command (ZK revision 2). The delegate parks it before storeLog runs.
+            final CompletableFuture<Void> blocked =
+                    self.commandExecutor().execute(Command.createRepository(Author.SYSTEM, "p1", "r-block"));
+            assertThat(executeEntered.await(10, TimeUnit.SECONDS)).isTrue();
+
+            // Stop the executor. doStop() nulls listenerInfo first, then blocks in shutdown(executor)
+            // waiting for the parked command. The gauge reads 0 once listenerInfo is null, which is our
+            // sync point for "the store is now racing a shutdown".
+            final CompletableFuture<Void> stopFuture = self.commandExecutor().stop();
+            await().untilAsserted(() -> assertThat(MoreMeters.measureAll(self.meterRegistry()))
+                    .containsEntry("replica.last.local.applied.revision#value", 0.0));
+
+            // Let the parked command reach storeLog while listenerInfo is null.
+            proceed.countDown();
+            blocked.join();
+            stopFuture.join();
+
+            // storeLog must have persisted local_last_revision to the new revision (2) even though
+            // listenerInfo was null during the store. Before the fix it stayed at 1, so on the next
+            // start-up the replica would re-apply revision 2 and enter read-only mode.
+            assertThat(self.localLastAppliedRevision()).isEqualTo(2L);
+        }
+    }
+
     @Test
     void testLogMetaSerde() throws JsonProcessingException {
         final LogMeta logMeta = new LogMeta(1, 1L, 10, false, false);