fix(raft): reset log always on receiving a snapshot

There were some cases where the log is not reset and leads to scenarios where a follower is not able to replicate new events. This case is explained in #10183 (comment)

atomix/cluster/src/main/java/io/atomix/raft/roles/PassiveRole.java

@@ -68,10 +68,7 @@ public PassiveRole(final RaftContext context) {
   public CompletableFuture<RaftRole> start() {
     snapshotListener = createSnapshotListener();
 
-    return super.start()
-        .thenRun(this::truncateUncommittedEntries)
-        .thenRun(this::addSnapshotListener)
-        .thenApply(v -> this);
+    return super.start().thenRun(this::truncateUncommittedEntries).thenApply(v -> this);
   }
 
   @Override
@@ -100,14 +97,6 @@ private void truncateUncommittedEntries() {
       raft.getLog().flush();
       raft.setLastWrittenIndex(raft.getCommitIndex());
     }
-
-    // to fix the edge case where we might have been stopped
-    // between persisting snapshot and truncating log we need to call on restart snapshot listener
-    // again, such that we truncate the log when necessary
-    final var latestSnapshot = raft.getCurrentSnapshot();
-    if (latestSnapshot != null && snapshotListener != null) {
-      snapshotListener.onNewSnapshot(latestSnapshot);
-    }
   }
 
   /**
@@ -268,11 +257,12 @@ public CompletableFuture<InstallResponse> onInstall(final InstallRequest request
       final long elapsed = System.currentTimeMillis() - pendingSnapshotStartTimestamp;
       log.debug("Committing snapshot {}", pendingSnapshot);
       try {
+        // Reset before committing to prevent the edge case where the system crashes after
+        // committing the snapshot, and restart with a snapshot and invalid log.
+        resetLogOnReceivingSnapshot(pendingSnapshot.index());
+
         final var snapshot = pendingSnapshot.persist().join();
         log.info("Committed snapshot {}", snapshot);
-        // Must be executed immediately before any other operation on this threadcontext. Hence
-        // don't wait for the listener to be notified by the snapshot store.
-        snapshotListener.onNewSnapshot(snapshot);
       } catch (final Exception e) {
         log.error("Failed to commit pending snapshot {}, rolling back", pendingSnapshot, e);
         abortPendingSnapshots();
@@ -745,6 +735,17 @@ protected boolean completeAppend(
     return succeeded;
   }
 
+  private void resetLogOnReceivingSnapshot(final long snapshotIndex) {
+    final var raftLog = raft.getLog();
+
+    log.info(
+        "Delete existing log (lastIndex '{}') and replace with received snapshot (index '{}'). First entry in the log will be at index {}",
+        raftLog.getLastIndex(),
+        snapshotIndex,
+        snapshotIndex + 1);
+    raftLog.reset(snapshotIndex + 1);
+  }
+
   private static final class ResetWriterSnapshotListener implements PersistedSnapshotListener {
 
     private final ThreadContext threadContext;