Slightly simplify buildPayload method in RaftReplicator

CHANGELOG.md

@@ -3,8 +3,8 @@
 ## Unreleased
 
 ### Bug fixes
-- Under complicated conditions, replication towards a particular follower could become stuck.
-(to workaround, restart leader node)
+- Under complicated conditions (follower is very far behind leader + network instabilities),
+replication towards a particular follower could become stuck. (to workaround, restart leader node)
 - Running ``DEL`` on a lease key would cause all nodes in a cluster to crash
 with an assertion. ``DEL`` will now simply release the given lease, as if
 ``lease-release`` had been called.
@@ -21,8 +21,9 @@ checkpoint for long-term storage.
 - Attempt to detect potential ``MANIFEST`` corruption early by measuring mtime lag
 compared to newest SST file.
 
-Many thanks to Pete Eby (ORNL) for finding and reporting the bug causing replication
-to become stuck.
+Many thanks to Crystal Chua (AARNet) for the bug report and all support offered
+related to RocksDB's ``MANIFEST`` corruption issue, as well as to Pete Eby (ORNL)
+for finding and reporting the bug causing replication to become stuck.
 
 ## 0.4.1 (2020-01-17)
 

src/raft/RaftReplicator.cc

@@ -91,9 +91,9 @@ RaftReplicaTracker::~RaftReplicaTracker() {
 }
 
 bool RaftReplicaTracker::buildPayload(LogIndex nextIndex, int64_t payloadLimit,
-  std::vector<std::string> &entries, int64_t &payloadSize, RaftTerm &lastEntryTerm) {
+  std::vector<std::string> &entries, RaftTerm &lastEntryTerm) {
 
-  payloadSize = std::min(payloadLimit, journal.getLogSize() - nextIndex);
+  int64_t payloadSize = std::min(payloadLimit, journal.getLogSize() - nextIndex);
   entries.resize(payloadSize);
 
   RaftJournal::Iterator iterator = journal.getIterator(nextIndex, true);
@@ -322,11 +322,13 @@ bool RaftReplicaTracker::sendPayload(RaftTalker &talker, LogIndex nextIndex, int
   LogIndex commitIndexForTarget = journal.getCommitIndex();
 
   std::vector<RaftSerializedEntry> entries;
-  if(!buildPayload(nextIndex, payloadLimit, entries, payloadSize, lastEntryTerm)) {
+  if(!buildPayload(nextIndex, payloadLimit, entries, lastEntryTerm)) {
     state.observed(snapshot->term+1, {});
     return false;
   }
 
+  payloadSize = entries.size();
+
   contact = std::chrono::steady_clock::now();
   reply = talker.appendEntries(
     snapshot->term,

src/raft/RaftReplicator.hh

@@ -82,7 +82,7 @@ private:
 
   void triggerResilvering();
   bool buildPayload(LogIndex nextIndex, int64_t payloadLimit, std::vector<RaftSerializedEntry> &entries,
-    int64_t &payloadSize, RaftTerm &lastEntryTerm);
+    RaftTerm &lastEntryTerm);
 
   bool sendPayload(RaftTalker &talker, LogIndex nextIndex, int64_t payloadLimit,
     std::future<redisReplyPtr> &reply, std::chrono::steady_clock::time_point &contact, int64_t &payloadSize,