From 19dcaac9c5981563215ecdde3a71e6b2ea3edeea Mon Sep 17 00:00:00 2001 From: Free Ekanayaka Date: Wed, 3 Dec 2025 10:35:33 +0000 Subject: [PATCH 1/4] recv: Assert that all entries are in the same batch When receiving an AppendEntries RPC it's expected that all entries belong to the same batch. Signed-off-by: Free Ekanayaka --- src/recv_append_entries.c | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/recv_append_entries.c b/src/recv_append_entries.c index 0517a122..d13e0810 100644 --- a/src/recv_append_entries.c +++ b/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; From d9ad91a99600a065cb698c78e55d299ee7f78d81 Mon Sep 17 00:00:00 2001 From: Free Ekanayaka Date: Wed, 3 Dec 2025 10:48:23 +0000 Subject: [PATCH 2/4] recv: Don't destroy received entries in case of error If an error occurs ownership of the entries is returned to the caller. Signed-off-by: Free Ekanayaka --- src/recv_append_entries.c | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/src/recv_append_entries.c b/src/recv_append_entries.c index d13e0810..f7ee89f2 100644 --- a/src/recv_append_entries.c +++ b/src/recv_append_entries.c @@ -165,12 +165,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; @@ -182,14 +176,16 @@ int recvAppendEntries(struct raft *r, goto err; } - return 0; - -err: - assert(rv != 0); + /* 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; } From bdb3167c2d6b8857b300ce4abc44afbc5156ded8 Mon Sep 17 00:00:00 2001 From: Free Ekanayaka Date: Wed, 3 Dec 2025 11:57:32 +0000 Subject: [PATCH 3/4] v1: Cleanup RAFT_RECEIVE memory ownership Now the ownership of the entries array is left entirely to the user. Signed-off-by: Free Ekanayaka --- src/legacy.c | 16 ++++++----- src/recv_append_entries.c | 7 ++--- src/replication.c | 56 ++++++++++++++++++++++++++++++--------- test/lib/cluster.c | 1 + 4 files changed, 56 insertions(+), 24 deletions(-) diff --git a/src/legacy.c b/src/legacy.c index 5506e63f..0984ac1f 100644 --- a/src/legacy.c +++ b/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; } diff --git a/src/recv_append_entries.c b/src/recv_append_entries.c index f7ee89f2..0161a85d 100644 --- a/src/recv_append_entries.c +++ b/src/recv_append_entries.c @@ -116,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); @@ -176,6 +172,7 @@ int recvAppendEntries(struct raft *r, goto err; } +out: /* Free the entries batch, if any. */ if (args->n_entries > 0) { assert(args->entries[0].batch != NULL); diff --git a/src/replication.c b/src/replication.c index a8d3606c..bdbacc84 100644 --- a/src/replication.c +++ b/src/replication.c @@ -363,14 +363,26 @@ 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)); @@ -378,8 +390,10 @@ static void persistEntries(struct raft *r, 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, @@ -387,7 +401,14 @@ int replicationTrigger(struct raft *r, 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,16 +875,20 @@ 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. */ @@ -867,6 +896,7 @@ int replicationAppend(struct raft *r, TrailTruncate(&r->trail, index); } +err: assert(rv != 0); return rv; } diff --git a/test/lib/cluster.c b/test/lib/cluster.c index b63017e6..2a3b2ead 100644 --- a/test/lib/cluster.c +++ b/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; From 2ee8f4d405722ac35bd870f62e614bacbca2ed5c Mon Sep 17 00:00:00 2001 From: Free Ekanayaka Date: Wed, 3 Dec 2025 12:30:02 +0000 Subject: [PATCH 4/4] v1: Memory ownership of RAFT_INSTALL_SNAPSHOT messages Make a copy of the provided snapshot configuration, and don't discard data upon error. Signed-off-by: Free Ekanayaka --- src/legacy.c | 6 ++++++ src/recv_install_snapshot.c | 7 +++---- src/replication.c | 20 +++++++++++++------- test/lib/cluster.c | 12 ++++++++++-- 4 files changed, 32 insertions(+), 13 deletions(-) diff --git a/src/legacy.c b/src/legacy.c index 0984ac1f..b941c6bd 100644 --- a/src/legacy.c +++ b/src/legacy.c @@ -1661,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; } diff --git a/src/recv_install_snapshot.c b/src/recv_install_snapshot.c index 14471356..d1f5c6e5 100644 --- a/src/recv_install_snapshot.c +++ b/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; } diff --git a/src/replication.c b/src/replication.c index bdbacc84..50ace906 100644 --- a/src/replication.c +++ b/src/replication.c @@ -950,6 +950,7 @@ int replicationInstallSnapshot(struct raft *r, { struct raft_snapshot_metadata metadata; raft_term local_term; + int rv; assert(r->state == RAFT_FOLLOWER); @@ -967,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; @@ -987,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, diff --git a/test/lib/cluster.c b/test/lib/cluster.c index 2a3b2ead..1da3611c 100644 --- a/test/lib/cluster.c +++ b/test/lib/cluster.c @@ -1064,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);