Leader takes into account previous log when checking whether to trigger snapshot

Author: ivandika3

ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java

@@ -868,6 +868,10 @@ private TermIndex shouldNotifyToInstallSnapshot() {
       return firstAvailable;
     }
 
+    if (followerNextIndex == leaderStartIndex && getPrevious(followerNextIndex) == null) {
+      return firstAvailable;
+    }
+
     return null;
   }
 

ratis-server-api/src/main/java/org/apache/ratis/server/leader/LogAppender.java

@@ -134,6 +134,35 @@ default RaftPeerId getFollowerId() {
   /** @return an {@link Iterable} of {@link InstallSnapshotRequestProto} for sending the given snapshot. */
   Iterable<InstallSnapshotRequestProto> newInstallSnapshotRequests(String requestId, SnapshotInfo snapshot);
 
+  /**
+   * Get the previous {@link TermIndex} for the given next index.
+   * This is used to set the previous log entry in AppendEntries requests.
+   *
+   * @return the previous {@link TermIndex}, or null if unavailable
+   *         (e.g. the entry has been purged and the snapshot does not cover it).
+   */
+  default TermIndex getPrevious(long nextIndex) {
+    if (nextIndex == RaftLog.LEAST_VALID_LOG_INDEX) {
+      return null;
+    }
+
+    final long previousIndex = nextIndex - 1;
+    final TermIndex previous = getRaftLog().getTermIndex(previousIndex);
+    if (previous != null) {
+      return previous;
+    }
+
+    final SnapshotInfo snapshot = getServer().getStateMachine().getLatestSnapshot();
+    if (snapshot != null) {
+      final TermIndex snapshotTermIndex = snapshot.getTermIndex();
+      if (snapshotTermIndex.getIndex() == previousIndex) {
+        return snapshotTermIndex;
+      }
+    }
+
+    return null;
+  }
+
   /**
    * Should this {@link LogAppender} send a snapshot to the follower?
    *
@@ -164,6 +193,9 @@ default SnapshotInfo shouldInstallSnapshot() {
       if (followerNextIndex < logStartIndex || (logStartIndex == RaftLog.INVALID_LOG_INDEX && snapshot != null)) {
         return snapshot;
       }
+      if (followerNextIndex == logStartIndex && getPrevious(followerNextIndex) == null) {
+        return snapshot;
+      }
     }
     return null;
   }

ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderBase.java

@@ -165,27 +165,6 @@ public boolean hasPendingDataRequests() {
     return false;
   }
 
-  private TermIndex getPrevious(long nextIndex) {
-    if (nextIndex == RaftLog.LEAST_VALID_LOG_INDEX) {
-      return null;
-    }
-
-    final long previousIndex = nextIndex - 1;
-    final TermIndex previous = getRaftLog().getTermIndex(previousIndex);
-    if (previous != null) {
-      return previous;
-    }
-
-    final SnapshotInfo snapshot = server.getStateMachine().getLatestSnapshot();
-    if (snapshot != null) {
-      final TermIndex snapshotTermIndex = snapshot.getTermIndex();
-      if (snapshotTermIndex.getIndex() == previousIndex) {
-        return snapshotTermIndex;
-      }
-    }
-
-    return null;
-  }
 
   protected long getNextIndexForInconsistency(long requestFirstIndex, long replyNextIndex) {
     long next = replyNextIndex;

ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java

@@ -309,5 +309,9 @@ void runTestNewAppendEntriesRequestAfterPurge(CLUSTER cluster,
 
     Assertions.assertEquals(targetNextIndex, appender.getFollower().getNextIndex(),
         "Follower nextIndex should remain unchanged");
+
+    Assertions.assertNotNull(appender.shouldInstallSnapshot(),
+        "shouldInstallSnapshot should return non-null when followerNextIndex ("
+            + targetNextIndex + ") and previous entry has been purged");
   }
 }