wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(raft-log-replication): simulator AppendEntries delivery (task 8.2)

Browse source 
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.
This commit is contained in:
Rongsong Shen 2026-06-21 16:40:11 +00:00
parent eeb95591ae
commit 0e8c6ed05a
4 changed files with 119 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.kiro/specs/raft-log-replication/tasks.md
View file

@ -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_

14
src/lib/raft/raft.h
View file

@ -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.
*/

43
src/lib/raft/raft_sim.c
View file

@ -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++)

60
src/lib/testsuite/test_raft.c
View file

@ -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);