replication: Don't use majority rule for old entries

Signed-off-by: Cole Miller <[email protected]>

src/convert.c

@@ -180,6 +180,12 @@ int convertToCandidate(struct raft *r, bool disrupt_leader)
     return 0;
 }
 
+void leaderBarrierCb(struct raft_barrier *req, int status)
+{
+    (void)status;
+    raft_free(req);
+}
+
 int convertToLeader(struct raft *r)
 {
     int rv;
@@ -211,9 +217,23 @@ int convertToLeader(struct raft *r)
      * we are the only voter around. */
     size_t n_voters = configurationVoterCount(&r->configuration);
     if (n_voters == 1 && (r->last_stored > r->commit_index)) {
-        tracef("Apply log entries after self election %llu %llu", r->last_stored, r->commit_index);
+        tracef("apply log entries after self election %llu %llu", r->last_stored, r->commit_index);
         r->commit_index = r->last_stored;
         rv = replicationApply(r);
+    } else {
+        /* Raft Dissertation, paragraph 6.4:
+         * The Leader Completeness Property guarantees that a leader has all committed
+         * entries, but at the start of its term, it may not know which those are.
+         * To find out, it needs to commit an entry from its term. Raft handles this by having
+         * each leader commit a blank no-op entry into the log at the start of its term */
+        struct raft_barrier *req = raft_malloc(sizeof(*req));
+        if (req == NULL) {
+            return RAFT_NOMEM;
+        }
+        rv = raft_barrier(r, req, leaderBarrierCb);
+        if (rv != 0) {
+            tracef("failed to send no-op barrier entry after leader conversion: %d", rv);
+        }
     }
 
     return rv;

src/replication.c

@@ -1620,16 +1620,33 @@ int replicationApply(struct raft *r)
     return rv;
 }
 
-void replicationQuorum(struct raft *r, const raft_index index)
+static bool matchIndicesAtLeast(struct raft *r, const raft_index index)
 {
     size_t votes = 0;
     size_t i;
+    for (i = 0; i < r->configuration.n; i++) {
+        struct raft_server *server = &r->configuration.servers[i];
+        tracef("server->id %llu server->role %d index %llu", server->id, server->role, index);
+        if (server->role != RAFT_VOTER) {
+            continue;
+        }
+        tracef("match_index %llu", r->leader_state.progress[i].match_index);
+        if (r->leader_state.progress[i].match_index >= index) {
+            votes++;
+        }
+    }
 
-    assert(r->state == RAFT_LEADER);
+    tracef("votes %ld", votes);
+    return votes > (configurationVoterCount(&r->configuration) / 2);
+}
 
-    if (index <= r->commit_index) {
-        return;
-    }
+/* Commit any entries up to the given index that have achieved majority
+ * replication. */
+void replicationQuorum(struct raft *r, const raft_index index)
+{
+    raft_index n;
+
+    assert(r->state == RAFT_LEADER);
 
     /* TODO: fuzzy-test --seed 0x8db5fccc replication/entries/partitioned
      * fails the assertion below. */
@@ -1639,19 +1656,21 @@ void replicationQuorum(struct raft *r, const raft_index index)
     // assert(logTermOf(&r->log, index) > 0);
     assert(logTermOf(&r->log, index) <= r->current_term);
 
-    for (i = 0; i < r->configuration.n; i++) {
-        struct raft_server *server = &r->configuration.servers[i];
-        if (server->role != RAFT_VOTER) {
-            continue;
-        }
-        if (r->leader_state.progress[i].match_index >= index) {
-            votes++;
+    /* Raft Paper fig. 2 - Rules for Servers
+     * If there exists an N such that N > commitIndex, a majority
+     * of matchIndex[i] >= N, and log[N].term == currentTerm:
+     * set commitIndex = N (5.3, 5.4).
+     * */
+    for (n = index; n > r->commit_index; n -= 1) {
+        if (logTermOf(&r->log, n) < r->current_term) {
+            break;
         }
-    }
 
-    if (votes > configurationVoterCount(&r->configuration) / 2) {
-        r->commit_index = index;
-        tracef("new commit index %llu", r->commit_index);
+        if (matchIndicesAtLeast(r, n)) {
+            r->commit_index = n;
+            tracef("new commit index %llu", n);
+            break;
+        }
     }
 
     return;