replication: Try to reinstate entries that have been truncated away

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

src/log.c

@@ -480,6 +480,41 @@ static int ensureCapacity(struct raft_log *l)
     return 0;
 }
 
+int logReinstate(struct raft_log *l, raft_term term, unsigned short type)
+{
+    if (l->refs_size == 0) {
+        return 0;
+    }
+
+    raft_index index = logLastIndex(l) + 1;
+    size_t key = refsKey(index, l->refs_size);
+    struct raft_entry_ref *bucket = &l->refs[key];
+    struct raft_entry_ref *slot;
+    struct raft_entry *entry;
+
+    if (bucket->count == 0) {
+        return 0;
+    }
+
+    if (bucket->index != index) {
+        return 0;
+    }
+
+    for (slot = bucket; slot != NULL; slot = slot->next) {
+        if (slot->term == term) {
+            slot->count += (unsigned short)1;
+            l->back += 1;
+            l->back %= l->size;
+            entry = &l->entries[l->back];
+            assert(entry->term == term);
+            assert(entry->type == type);
+            return 1;
+        }
+    }
+
+    return 0;
+}
+
 int logAppend(struct raft_log *l,
               const raft_term term,
               const unsigned short type,

src/log.h

@@ -90,6 +90,8 @@ raft_index logSnapshotIndex(struct raft_log *l);
  * invoked. Return #NULL if there is no such entry. */
 const struct raft_entry *logGet(struct raft_log *l, const raft_index index);
 
+int logReinstate(struct raft_log *l, raft_term term, unsigned short type);
+
 /* Append a new entry to the log. */
 int logAppend(struct raft_log *l,
               raft_term term,

src/replication.c

@@ -1182,6 +1182,11 @@ int replicationAppend(struct raft *r,
             goto err_after_request_alloc;
         }
 
+        rv = logReinstate(r->log, copy.term, copy.type);
+        if (rv > 0) {
+            continue;
+        }
+
         rv = logAppend(r->log, copy.term, copy.type, &copy.buf, NULL);
         if (rv != 0) {
             goto err_after_request_alloc;