v1: Cleanup memory ownership of RAFT_RECEIVE event

src/legacy.c

@@ -193,7 +193,8 @@ static int legacyHandleUpdateEntries(struct raft *r,
 
     req = raft_malloc(sizeof *req);
     if (req == NULL) {
-        return RAFT_NOMEM;
+        rv = RAFT_NOMEM;
+        goto err;
     }
     req->r = r;
     req->index = index;
@@ -208,27 +209,28 @@ static int legacyHandleUpdateEntries(struct raft *r,
         struct raft_entry entry;
         rv = entryCopy(&entries[i], &entry);
         if (rv != 0) {
-            goto err;
+            goto err_after_req_alloc;
         }
         rv = logAppend(r->legacy.log, entry.term, entry.type, &entry.buf, NULL);
         if (rv != 0) {
-            goto err;
+            goto err_after_req_alloc;
         }
     }
 
     assert(n > 0);
     assert(entries[0].batch != NULL);
     raft_free(entries[0].batch);
+    raft_free(entries);
 
     rv = r->io->truncate(r->io, index);
     if (rv != 0) {
-        goto err;
+        goto err_after_req_alloc;
     }
 
     rv = logAcquire(r->legacy.log, index, &acquired, &n_acquired);
     assert(n_acquired == n);
     if (rv != 0) {
-        goto err;
+        goto err_after_req_alloc;
     }
 
     req->entries = acquired;
@@ -243,10 +245,12 @@ static int legacyHandleUpdateEntries(struct raft *r,
 
 err_after_acquired:
     logRelease(r->legacy.log, index, acquired, n_acquired);
-err:
+err_after_req_alloc:
     logDiscard(r->legacy.log, index);
     raft_free(req);
     ErrMsgTransferf(r->io->errmsg, r->errmsg, "append %u entries", n);
+err:
+    assert(rv != 0);
     return rv;
 }
 
@@ -1657,6 +1661,12 @@ static void recvCb(struct raft_io *io, struct raft_message *message)
                 raft_free(message->append_entries.entries);
             }
             break;
+        case RAFT_INSTALL_SNAPSHOT:
+            if (rv != 0) {
+                raft_free(message->install_snapshot.data.base);
+            }
+            raft_configuration_close(&message->install_snapshot.conf);
+            break;
         default:
             break;
     }

src/recv_append_entries.c

@@ -30,6 +30,15 @@ int recvAppendEntries(struct raft *r,
     assert(args != NULL);
     assert(address != NULL);
 
+    /* Sanity check that all entries belong to the same batch */
+    if (args->n_entries > 0) {
+        unsigned i;
+        assert(args->entries[0].batch != NULL);
+        for (i = 1; i < args->n_entries; i++) {
+            assert(args->entries[i].batch == args->entries[i - 1].batch);
+        }
+    }
+
     result->rejected = args->prev_log_index;
     result->version = MESSAGE__APPEND_ENTRIES_RESULT_VERSION;
     result->features = MESSAGE__FEATURE_CAPACITY;
@@ -107,11 +116,7 @@ int recvAppendEntries(struct raft *r,
      * should be in charge of serializing everything. */
     if (r->snapshot.installing && args->n_entries > 0) {
         infof("snapshot install in progress -> ignore");
-        if (args->n_entries > 0) {
-            assert(args->entries[0].batch != NULL);
-            raft_free(args->entries[0].batch);
-        }
-        return 0;
+        goto out;
     }
 
     rv = replicationAppend(r, args, &result->rejected, &async);
@@ -156,12 +161,6 @@ int recvAppendEntries(struct raft *r,
 reply:
     result->term = r->current_term;
 
-    /* Free the entries batch, if any. */
-    if (args->n_entries > 0) {
-        assert(args->entries[0].batch != NULL);
-        raft_free(args->entries[0].batch);
-    }
-
     result->capacity = r->capacity;
 
     message.type = RAFT_APPEND_ENTRIES_RESULT;
@@ -173,14 +172,17 @@ int recvAppendEntries(struct raft *r,
         goto err;
     }
 
-    return 0;
-
-err:
-    assert(rv != 0);
+out:
+    /* Free the entries batch, if any. */
     if (args->n_entries > 0) {
         assert(args->entries[0].batch != NULL);
         raft_free(args->entries[0].batch);
     }
+
+    return 0;
+
+err:
+    assert(rv != 0);
     return rv;
 }
 

src/recv_install_snapshot.c

@@ -75,10 +75,6 @@ int recvInstallSnapshot(struct raft *r,
     result->term = r->current_term;
     result->rejected = 0;
 
-    /* Free the snapshot data. */
-    raft_configuration_close(&args->conf);
-    raft_free(args->data.base);
-
     result->capacity = r->capacity;
 
     message.type = RAFT_APPEND_ENTRIES_RESULT;
@@ -90,6 +86,9 @@ int recvInstallSnapshot(struct raft *r,
         return rv;
     }
 
+    /* Free the snapshot data. */
+    raft_free(args->data.base);
+
     return 0;
 }
 

src/replication.c

@@ -363,31 +363,52 @@ int replicationPersistEntriesDone(struct raft *r, raft_index index)
     return 0;
 }
 
-static void persistEntries(struct raft *r,
-                           raft_index index,
-                           struct raft_entry entries[],
-                           unsigned n)
+static int persistEntries(struct raft *r,
+                          raft_index index,
+                          struct raft_entry entries[],
+                          unsigned n)
 {
+    struct raft_entry *batch;
+    unsigned i;
+
     assert(n > 0);
     assert(entries != NULL);
 
+    batch = raft_malloc(sizeof *batch * n);
+    if (batch == NULL) {
+        return RAFT_NOMEM;
+    }
+
+    for (i = 0; i < n; i++) {
+        batch[i] = entries[i];
+    }
+
     /* This must be the first time during this raft_step() call where we set new
      * entries to be persisted. */
     assert(!(r->update->flags & RAFT_UPDATE_ENTRIES));
 
     r->update->flags |= RAFT_UPDATE_ENTRIES;
 
     r->update->entries.index = index;
-    r->update->entries.batch = entries;
+    r->update->entries.batch = batch;
     r->update->entries.n = n;
+
+    return 0;
 }
 
 int replicationTrigger(struct raft *r,
                        raft_index index,
                        struct raft_entry *entries,
                        unsigned n)
 {
-    persistEntries(r, index, entries, n);
+    int rv;
+
+    rv = persistEntries(r, index, entries, n);
+    if (rv != 0) {
+        assert(rv == RAFT_NOMEM);
+        return rv;
+    }
+
     return triggerAll(r);
 }
 
@@ -751,14 +772,18 @@ int replicationAppend(struct raft *r,
     match = checkLogMatchingProperty(r, args);
     if (match != 0) {
         assert(match == 1 || match == -1);
-        return match == 1 ? 0 : RAFT_SHUTDOWN;
+        if (match == 1) {
+            return 0;
+        }
+        rv = RAFT_SHUTDOWN;
+        goto err;
     }
 
     /* Check for conflicting entries. */
     rv = checkConflictingEntries(r, args, &i, &truncate);
     if (rv != 0) {
         assert(rv == RAFT_SHUTDOWN);
-        return rv;
+        goto err;
     }
 
     /* From Figure 3.1:
@@ -772,7 +797,7 @@ int replicationAppend(struct raft *r,
         rv = deleteConflictingEntries(r, truncate);
         if (rv != 0) {
             assert(rv == RAFT_NOMEM);
-            return rv;
+            goto err;
         }
     }
 
@@ -788,7 +813,7 @@ int replicationAppend(struct raft *r,
         struct raft_entry *entry = &args->entries[i + j];
         rv = TrailAppend(&r->trail, entry->term);
         if (rv != 0) {
-            goto err;
+            goto err_after_trail_append;
         }
     }
 
@@ -850,23 +875,28 @@ int replicationAppend(struct raft *r,
         if (entry->type == RAFT_CHANGE) {
             rv = membershipUncommittedChange(r, index, entry);
             if (rv != 0) {
-                goto err;
+                goto err_after_trail_append;
             }
         }
     }
 
-    persistEntries(r, index, entries, n_entries);
+    rv = persistEntries(r, index, entries, n_entries);
+    if (rv != 0) {
+        assert(rv == RAFT_NOMEM);
+        goto err_after_trail_append;
+    }
 
     return 0;
 
-err:
+err_after_trail_append:
     /* Release all entries added to the in-memory log, making
      * sure the in-memory log and disk don't diverge, leading
      * to future log entries not being persisted to disk. */
     if (j != 0) {
         TrailTruncate(&r->trail, index);
     }
 
+err:
     assert(rv != 0);
     return rv;
 }
@@ -920,6 +950,7 @@ int replicationInstallSnapshot(struct raft *r,
 {
     struct raft_snapshot_metadata metadata;
     raft_term local_term;
+    int rv;
 
     assert(r->state == RAFT_FOLLOWER);
 
@@ -937,6 +968,7 @@ int replicationInstallSnapshot(struct raft *r,
      *
      * TODO: we should do something smarter. */
     if (r->snapshot.installing) {
+        raft_free(args->data.base);
         *async = true;
         infof("already taking or installing snapshot");
         return 0;
@@ -957,20 +989,24 @@ int replicationInstallSnapshot(struct raft *r,
 
     *async = true;
 
+    metadata.index = args->last_index;
+    metadata.term = args->last_term;
+    metadata.index = args->last_index;
+    metadata.configuration_index = args->conf_index;
+    rv = configurationCopy(&args->conf, &metadata.configuration);
+    if (rv != 0) {
+        assert(rv == RAFT_NOMEM);
+        return rv;
+    }
+
     /* Preemptively update our in-memory state. */
-    TrailRestore(&r->trail, args->last_index, args->last_term);
+    TrailRestore(&r->trail, metadata.index, metadata.term);
 
     r->last_stored = 0;
 
     assert(!r->snapshot.installing);
     r->snapshot.installing = true;
 
-    metadata.index = args->last_index;
-    metadata.term = args->last_term;
-    metadata.index = args->last_index;
-    metadata.configuration_index = args->conf_index;
-    metadata.configuration = args->conf;
-
     assert(!(r->update->flags & RAFT_UPDATE_SNAPSHOT));
 
     infof("start persisting snapshot (%llu^%llu)", metadata.index,

test/lib/cluster.c

@@ -635,6 +635,7 @@ static void serverProcessEntries(struct test_server *s,
     if (n > 0) {
         munit_assert_ptr_not_null(entries[0].batch);
         raft_free(entries[0].batch);
+        raft_free(entries);
     }
 
     step->id = s->raft.id;
@@ -1063,8 +1064,16 @@ static void serverCompleteReceive(struct test_server *s, struct step *step)
     rv = serverStep(s, event);
     munit_assert_int(rv, ==, 0);
 
-    if (event->receive.message->type == RAFT_APPEND_ENTRIES) {
-        raft_free(event->receive.message->append_entries.entries);
+    switch (event->receive.message->type) {
+        case RAFT_APPEND_ENTRIES:
+            raft_free(event->receive.message->append_entries.entries);
+            break;
+        case RAFT_INSTALL_SNAPSHOT:
+            raft_configuration_close(
+                &event->receive.message->install_snapshot.conf);
+            break;
+        default:
+            break;
     }
 
     free(event->receive.message);