From 0e8c6ed05a435b818efdfca7d70c2784fd23450a Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 21 Jun 2026 16:40:11 +0000 Subject: [PATCH] feat(raft-log-replication): simulator AppendEntries delivery (task 8.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit raft_sim_deliver_append_entries drains the AE queue, routes each buffered AE to the (non-partitioned) destination via raft_on_append_entries_request, and routes the result back to the source via raft_on_append_entries_reply — mirroring the vote-RPC routing; partitioned endpoints are skipped. Review-applied (HIGH): raft_sim_ae_t now captures the leader's SEND-time term (leader_term) at buffer time and delivery uses it instead of the live source state.term — without this, reorder/churn delivery (9.3/9.4) would inject the wrong term and defeat UpdateTerm/stale-drop (G5). Smoke test asserts replication to a connected follower + isolation of a partitioned one. ASAN green. --- .kiro/specs/raft-log-replication/tasks.md | 3 +- src/lib/raft/raft.h | 14 ++++++ src/lib/raft/raft_sim.c | 43 ++++++++++++++++ src/lib/testsuite/test_raft.c | 60 +++++++++++++++++++++++ 4 files changed, 119 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/raft-log-replication/tasks.md b/.kiro/specs/raft-log-replication/tasks.md index 28f461d5..25be73d0 100644 --- a/.kiro/specs/raft-log-replication/tasks.md +++ b/.kiro/specs/raft-log-replication/tasks.md @@ -144,7 +144,8 @@ - _Boundary: raft_sim.c_ - _Depends: 6.3, 7.2_ -- [ ] 8.2 Add synchronous AppendEntries delivery honoring partitions +- [x] 8.2 Add synchronous AppendEntries delivery honoring partitions + - NOTE (impl 2026-06-21): `raft_sim_ae_t` now carries `leader_term` captured at SEND time (in `sim_send_append_entries`) and delivery uses `ae->leader_term`, NOT the live source `state.term`. Required so reorder (9.3) and stale-term churn (9.4) deliver the sending-epoch term — using the live term would bypass UpdateTerm/stale-drop and defeat the G5 regression. Caught in review. - Add `raft_sim_deliver_append_entries`: for each buffered AE, call `raft_on_append_entries_request` on the (non-partitioned) destination, then route the reply back via `raft_on_append_entries_reply`; skip partitioned nodes (mirrors the existing vote-RPC routing) - Observable completion: driving one append + delivery replicates the entry to a connected follower; partitioned follower receives nothing - _Requirements: 8.1_ diff --git a/src/lib/raft/raft.h b/src/lib/raft/raft.h index ca049ffe..f91cf414 100644 --- a/src/lib/raft/raft.h +++ b/src/lib/raft/raft.h @@ -402,6 +402,10 @@ typedef struct { typedef struct { int src; int dst; + uint64_t leader_term; /* leader's term captured at SEND time (not re-read + * at delivery) — required so reorder/churn tests + * (9.3/9.4) deliver the send-epoch term, preserving + * UpdateTerm/stale-drop (G5) semantics. */ uint64_t prev_log_index; uint64_t prev_log_term; uint64_t leader_commit; @@ -468,6 +472,16 @@ void raft_sim_dump_log(raft_sim_cluster_t *c, FILE *fp); */ void raft_sim_deliver_pending(raft_sim_cluster_t *c); +/* + * raft_sim_deliver_append_entries: route all buffered AppendEntries synchronously. + * For each buffered AE (src, dst, ...), calls raft_on_append_entries_request on dst + * then routes the reply back to src via raft_on_append_entries_reply, mirroring the + * vote-RPC routing in raft_sim_deliver_pending. Skips delivery if either endpoint + * has partition=1. Drains the queue (snapshots the count, since handling may buffer + * more AEs). The single entry (n_entries == 1) is sourced from the buffered AE. + */ +void raft_sim_deliver_append_entries(raft_sim_cluster_t *c); + /* * raft_sim_start_all_elections: call raft_start_election on each node 0..n-1. */ diff --git a/src/lib/raft/raft_sim.c b/src/lib/raft/raft_sim.c index ed44de21..8fa2db6b 100644 --- a/src/lib/raft/raft_sim.c +++ b/src/lib/raft/raft_sim.c @@ -170,6 +170,7 @@ static int sim_send_append_entries(void *ctx, int peer_idx, memset(ae, 0, sizeof(*ae)); ae->src = node->idx; ae->dst = peer_idx; + ae->leader_term = node->state.term; /* capture SEND-time term (see raft.h) */ ae->prev_log_index = prev_log_index; ae->prev_log_term = prev_log_term; ae->leader_commit = leader_commit; @@ -312,6 +313,48 @@ void raft_sim_deliver_pending(raft_sim_cluster_t *c) } } +void raft_sim_deliver_append_entries(raft_sim_cluster_t *c) +{ + /* Snapshot current count — handling may buffer more AEs (mirrors vote path) */ + int count = c->ae_pending_count; + c->ae_pending_count = 0; + + for (int i = 0; i < count; i++) { + raft_sim_ae_t *ae = &c->ae_pending[i]; + int src = ae->src; + int dst = ae->dst; + + /* Drop if either endpoint is partitioned */ + if (c->nodes[src].partition || c->nodes[dst].partition) + continue; + + /* The single entry's bytes/term were captured into the AE at buffer + * time (library holds no entry bytes); pass them through for n_entries==1. */ + const void *entry = (ae->n_entries == 1) ? ae->entry_bytes : NULL; + size_t entry_len = (ae->n_entries == 1) ? ae->entry_len : 0; + uint64_t entry_term = (ae->n_entries == 1) ? ae->entry_term : 0; + + /* Deliver AppendEntries request to dst using the SEND-time leader term + * captured into the AE buffer at sim_send_append_entries (NOT the live + * source term). This mirrors the vote path's buffered rpc->term and keeps + * reorder/churn delivery (9.3/9.4) honest about the sending epoch so the + * follower's UpdateTerm/stale-drop decisions match the real protocol. */ + uint64_t leader_term = ae->leader_term; + raft_append_result_t result; + memset(&result, 0, sizeof(result)); + raft_on_append_entries_request(&c->nodes[dst].state, + src, leader_term, + ae->prev_log_index, ae->prev_log_term, + ae->leader_commit, entry, entry_len, + entry_term, ae->n_entries, &result); + + /* Route reply back to src */ + raft_on_append_entries_reply(&c->nodes[src].state, dst, + result.success, result.match_index, + result.reply_term); + } +} + void raft_sim_start_all_elections(raft_sim_cluster_t *c) { for (int i = 0; i < c->n; i++) diff --git a/src/lib/testsuite/test_raft.c b/src/lib/testsuite/test_raft.c index e30b37e4..8c925b83 100644 --- a/src/lib/testsuite/test_raft.c +++ b/src/lib/testsuite/test_raft.c @@ -3041,6 +3041,65 @@ static void test_sim_fire_timeout_steps_down(void) TEST_ASSERT_EQUAL_INT(RAFT_FOLLOWER, c.nodes[0].state.role); } +/* + * test_sim_deliver_append_entries_replicates (task 8.2) + * + * Delivery smoke test for raft_sim_deliver_append_entries: a leader appends one + * entry and replicates it to a connected follower; a partitioned follower + * receives nothing. + * + * Steps: + * 1. raft_sim_init(c, 3, 2000); elect node 0 leader. + * 2. Partition node 2 (so it must not receive the entry). + * 3. Leader (node 0) raft_log_append one entry → lands in node 0's log[0]. + * 4. raft_send_append_entries to node 1 (connected) and node 2 (partitioned) + * → buffers two AEs. + * 5. raft_sim_deliver_append_entries(c) routes them. + * 6. Assert node 1's in-memory log now holds the entry (log_count >= 1, term set). + * 7. Assert node 2 (partitioned) received nothing (log_count == 0). + */ +static void test_sim_deliver_append_entries_replicates(void) +{ + raft_sim_cluster_t c; + memset(&c, 0, sizeof(c)); + + TEST_ASSERT_EQUAL_INT(0, raft_sim_init(&c, 3, 2000)); + + /* Elect node 0 leader. */ + raft_start_election(&c.nodes[0].state); + raft_sim_deliver_pending(&c); + TEST_ASSERT_EQUAL_INT(RAFT_LEADER, c.nodes[0].state.role); + + /* Isolate node 2 — it must receive nothing. */ + raft_sim_partition(&c, 2, 1); + + /* Leader appends one client entry; persisted into node 0's in-memory log. */ + unsigned char payload[4] = { 0xDE, 0xAD, 0xBE, 0xEF }; + uint64_t out_index = 0; + TEST_ASSERT_EQUAL_INT(RAFT_OK, + raft_log_append(&c.nodes[0].state, payload, sizeof(payload), &out_index)); + TEST_ASSERT_EQUAL_UINT64(1, out_index); + TEST_ASSERT_EQUAL_INT(1, c.nodes[0].log_count); + + /* Send AppendEntries to connected (1) and partitioned (2) followers. */ + TEST_ASSERT_EQUAL_INT(RAFT_OK, + raft_send_append_entries(&c.nodes[0].state, 1, /*leader_commit=*/0)); + TEST_ASSERT_EQUAL_INT(RAFT_OK, + raft_send_append_entries(&c.nodes[0].state, 2, /*leader_commit=*/0)); + + /* Deliver: node 1 receives + replies; node 2 is dropped (partitioned). */ + raft_sim_deliver_append_entries(&c); + + /* Connected follower replicated the entry — its term must equal the LEADER'S + * PERSISTED entry term (node 0's log[0]), the authoritative send-time term, + * not the leader's live state term (which only coincides here). */ + TEST_ASSERT_EQUAL_INT(1, c.nodes[1].log_count); + TEST_ASSERT_EQUAL_UINT64(c.nodes[0].log[0].term, c.nodes[1].log[0].term); + + /* Partitioned follower received nothing. */ + TEST_ASSERT_EQUAL_INT(0, c.nodes[2].log_count); +} + /* ---------- Group A: Basic Election Correctness (task 5.1) ---------- */ /* @@ -5288,6 +5347,7 @@ void test_raft(void) RUN_TEST(test_sim_start_all_elections_no_winner); RUN_TEST(test_sim_partition_no_quorum); RUN_TEST(test_sim_fire_timeout_steps_down); + RUN_TEST(test_sim_deliver_append_entries_replicates); /* Group A: basic election correctness (task 5.1) */ RUN_TEST(test_proto_A1_single_node_self_elects);