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

feat(raft-sim-conformance): bag push/deliver routes vote+AE RPCs through library handlers (task 2.1)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-24 12:10:11 +00:00
parent 631854303c
commit ae4495104a
4 changed files with 249 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/raft-sim-conformance/tasks.md
View file

@ -37,7 +37,7 @@
## 2. Message bag (mirrors raft.tla `messages`)
- [ ] 2.1 RED+GREEN: bag push/deliver routes vote and AppendEntries RPCs through the library handlers
- [x] 2.1 RED+GREEN: bag push/deliver routes vote and AppendEntries RPCs through the library handlers
- Test: pushing a vote-req and an AE-req then delivering them invokes `raft_on_vote_request` / `raft_on_append_entries_request`, and the handler reply is pushed back as a `*_RSP` message
- Implement `raft_sim_bag_push` and `raft_sim_bag_deliver(i)` (route by `kind`, honor `node.partition`, push replies)
- Observable completion: delivered requests produce matching responses in the bag; partitioned endpoints are not delivered

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

@ -522,6 +522,10 @@ typedef struct {
raft_sim_msg_kind_t kind;
int src, dst;
uint64_t term; /* RPC term (all kinds) */
/* vote request: candidate's advertised last-log tuple at SEND time, feeding the
* (lastLogTerm, lastLogIndex) freshness gate at delivery (mirrors
* raft_sim_rpc_t.last_log_*). Sourced from the candidate's own log at send. */
uint64_t last_log_term, last_log_index;
/* AE request */
uint64_t prev_log_index, prev_log_term, leader_commit;
uint64_t entry_term, entry_value; /* one entry per AE (raft.tla "<=1 entry") */
@ -668,6 +672,29 @@ int raft_sim_conformance_init(raft_sim_cluster_t *c, int n_nodes,
*/
uint64_t raft_sim_prng_next(raft_sim_cluster_t *c);
/*
* raft_sim_bag_push: append one message to the unified cluster bag[] (mirrors
* raft.tla `messages`). Fails loud (returns -1) on overflow rather than silently
* truncating; returns 0 on success. Exposed so tests (and later scheduler tasks)
* can seed the bag directly.
*/
int raft_sim_bag_push(raft_sim_cluster_t *c, const raft_sim_msg_t *msg);
/*
* raft_sim_bag_deliver: route the bag message at index `i` to its matching library
* handler (vote/AE request or response) and push any handler reply back onto the
* bag as a *_RSP message (src/dst swapped). Mirrors raft.tla Receive: delivering a
* request consumes it (swap-with-last removal, so bag indices stay compact for the
* Drop/Duplicate/Reorder actions in task 2.2) and produces a response member.
* UpdateTerm higher-term adoption and the stale-response drop live INSIDE the
* handlers (R2.4) — this only routes. Delivery honors node.partition: if either
* endpoint is partitioned the message is NOT delivered (no removal, no reply),
* mirroring a dropped network edge.
* Returns 0 on a delivered message, -1 on an out-of-range index, -2 when the
* endpoint is partitioned (distinct non-zero status, message left in the bag).
*/
int raft_sim_bag_deliver(raft_sim_cluster_t *c, int i);
/*
* raft_sim_dump_log: print trace logs for all nodes to fp.
* If RAFT_SIM_VERBOSE env var is set, always prints.

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

@ -312,7 +312,7 @@ static void sim_conf_on_commit_advanced(void *ctx, uint64_t commit_index)
* (returns -1) on overflow rather than silently truncating, so a bounds bug is
* visible (design Error Handling: bag overflow fails loud). Returns 0 on success.
*/
static int raft_sim_bag_push(raft_sim_cluster_t *c, const raft_sim_msg_t *msg)
int raft_sim_bag_push(raft_sim_cluster_t *c, const raft_sim_msg_t *msg)
{
if (c->bag_count >= RAFT_SIM_MAX_MSGS)
return -1; /* fail loud — do not silently drop */
@ -336,6 +336,15 @@ static int sim_conf_send_vote_request(void *ctx, int peer_idx, uint64_t term)
msg.src = node->idx;
msg.dst = peer_idx;
msg.term = term;
/* Capture the CANDIDATE's last-log tuple at SEND time from its own in-memory log
* (the library holds no entry bytes): last_log_index == log_count, last_log_term
* == the tail entry's term (0 when empty). This feeds the (lastLogTerm,
* lastLogIndex) freshness gate when raft_sim_bag_deliver routes the request —
* exactly as the legacy sim_send_vote_request does for the pending[] path. */
msg.last_log_index = (uint64_t)node->log_count;
msg.last_log_term = (node->log_count > 0)
? node->log[node->log_count - 1].term
: 0;
return raft_sim_bag_push(c, &msg);
}
@ -583,6 +592,112 @@ void raft_sim_deliver_append_entries(raft_sim_cluster_t *c)
}
}
/*
* raft_sim_bag_remove_at: swap-with-last removal of bag member `i`.
*
* Bag removal model (consumed on delivery, shared with task 2.2 Drop/Duplicate/
* Reorder): the message at `i` is overwritten by the last member and bag_count is
* decremented. This keeps the bag a compact dense array with no holes, so an index
* chosen by the PRNG always names a live member. Delivery removes the request
* FIRST, then pushes the reply, so the reply lands at a fresh tail slot and never
* collides with the just-vacated index.
*/
static void raft_sim_bag_remove_at(raft_sim_cluster_t *c, int i)
{
/* Caller guarantees 0 <= i < bag_count. */
int last = c->bag_count - 1;
if (i != last)
c->bag[i] = c->bag[last];
c->bag_count = last;
}
int raft_sim_bag_deliver(raft_sim_cluster_t *c, int i)
{
if (i < 0 || i >= c->bag_count)
return -1; /* out of range — nothing to deliver */
/* Snapshot the message by value before any removal/push reshuffles the bag. */
raft_sim_msg_t m = c->bag[i];
int src = m.src;
int dst = m.dst;
/* Honor partitions: a partitioned endpoint's message is not delivered, mirroring
* a dropped network edge. Leave it in the bag (a later heal can deliver it). */
if (c->nodes[src].partition || c->nodes[dst].partition)
return -2;
/* Consume the request/response FIRST (swap-with-last), then push any reply so
* indices stay sane and the reply occupies a fresh tail slot. */
raft_sim_bag_remove_at(c, i);
switch (m.kind) {
case RAFT_SIM_MSG_VOTE_REQ: {
/* Route to dst's vote handler, passing the candidate's send-time last-log
* tuple so dst evaluates the (lastLogTerm, lastLogIndex) freshness gate. */
uint64_t reply_term = 0;
int granted = raft_on_vote_request(&c->nodes[dst].state, src, m.term,
m.last_log_term, m.last_log_index,
&reply_term);
raft_sim_msg_t rsp;
memset(&rsp, 0, sizeof(rsp));
rsp.kind = RAFT_SIM_MSG_VOTE_RSP;
rsp.src = dst; /* voter replies */
rsp.dst = src; /* back to candidate */
rsp.term = reply_term;
rsp.granted = granted;
return raft_sim_bag_push(c, &rsp);
}
case RAFT_SIM_MSG_AE_REQ: {
/* For n_entries == 1 the single entry's bytes come from the carried value
* (the library holds none); re-encode the same little-endian uint64_t the
* SEND callback decoded, so the receiver's sim_persist_entry / decode round-
* trips it. Use the SEND-time leader term carried in the message (m.term),
* NOT the live source term. */
unsigned char entry_bytes[sizeof(uint64_t)];
const void *entry = NULL;
size_t entry_len = 0;
uint64_t entry_term = 0;
if (m.entries_len == 1) {
memcpy(entry_bytes, &m.entry_value, sizeof(entry_bytes));
entry = entry_bytes;
entry_len = sizeof(entry_bytes);
entry_term = m.entry_term;
}
raft_append_result_t result;
memset(&result, 0, sizeof(result));
raft_on_append_entries_request(&c->nodes[dst].state, src, m.term,
m.prev_log_index, m.prev_log_term,
m.leader_commit, entry, entry_len,
entry_term, m.entries_len, &result);
raft_sim_msg_t rsp;
memset(&rsp, 0, sizeof(rsp));
rsp.kind = RAFT_SIM_MSG_AE_RSP;
rsp.src = dst; /* follower replies */
rsp.dst = src; /* back to leader */
rsp.term = result.reply_term;
rsp.success = result.success;
rsp.match_index = result.match_index;
return raft_sim_bag_push(c, &rsp);
}
case RAFT_SIM_MSG_VOTE_RSP:
/* Responses are first-class bag members: route to the candidate's tally. */
raft_on_vote_reply(&c->nodes[dst].state, src, m.granted, m.term);
return 0;
case RAFT_SIM_MSG_AE_RSP:
raft_on_append_entries_reply(&c->nodes[dst].state, src, m.success,
m.match_index, m.term);
return 0;
}
return 0; /* unreachable — enum is exhaustive */
}
void raft_sim_start_all_elections(raft_sim_cluster_t *c)
{
for (int i = 0; i < c->n; i++)

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

@ -4542,6 +4542,108 @@ static void test_sim_conformance_init_append_readback(void)
TEST_ASSERT_EQUAL_UINT64(g1.prng, g2.prng);
}
/*
* test_sim_conformance_bag_deliver_routes_rpcs (raft-sim-conformance task 2.1; Req 2.2, 2.4)
*
* Observable completion for raft_sim_bag_push / raft_sim_bag_deliver(i):
* - A pushed VOTE_REQ delivered by index invokes raft_on_vote_request on dst and
* pushes a VOTE_RSP back onto the bag (src/dst swapped, granted=result). The
* request is consumed (swap-with-last removal); the reply is the surviving member.
* - A pushed AE_REQ delivered by index invokes raft_on_append_entries_request on
* dst (replicating the single entry) and pushes an AE_RSP back (src/dst swapped,
* success=result). The dst's in-memory log grows (handler ran).
* - A request to a PARTITIONED dst is NOT delivered: no reply is pushed, the dst
* state is unchanged, and the request stays in the bag (mirrors a dropped edge).
*/
static void test_sim_conformance_bag_deliver_routes_rpcs(void)
{
raft_sim_cluster_t c;
memset(&c, 0, sizeof(c));
/* 3-node conformance cluster; force node 0 to LEADER so it owns a real log
* tail and per-peer progress for a genuine AppendEntries. */
TEST_ASSERT_EQUAL_INT(0, raft_sim_conformance_init(&c, 3, 2000, 0xBA61ULL));
raft_start_election(&c.nodes[0].state); /* sends vote reqs to the bag */
/* Drain the election vote traffic via direct delivery so node 0 wins, leaving
* the bag empty for the targeted assertions below. */
for (int g = 0; g < 64 && c.bag_count > 0; g++)
raft_sim_bag_deliver(&c, 0); /* always deliver head */
TEST_ASSERT_EQUAL_INT(RAFT_LEADER, c.nodes[0].state.role);
/* ---- VOTE_REQ routing ---- */
c.bag_count = 0;
raft_sim_msg_t vreq;
memset(&vreq, 0, sizeof(vreq));
vreq.kind = RAFT_SIM_MSG_VOTE_REQ;
vreq.src = 1; /* candidate */
vreq.dst = 2; /* voter */
vreq.term = c.nodes[1].state.term + 5; /* fresh higher term -> grant */
vreq.last_log_term = 0; /* empty-log candidate vs empty voter */
vreq.last_log_index = 0;
TEST_ASSERT_EQUAL_INT(0, raft_sim_bag_push(&c, &vreq));
TEST_ASSERT_EQUAL_INT(1, c.bag_count);
TEST_ASSERT_EQUAL_INT(0, raft_sim_bag_deliver(&c, 0));
/* Request consumed, exactly one VOTE_RSP now in the bag, src/dst swapped. */
TEST_ASSERT_EQUAL_INT(1, c.bag_count);
TEST_ASSERT_EQUAL_INT(RAFT_SIM_MSG_VOTE_RSP, c.bag[0].kind);
TEST_ASSERT_EQUAL_INT(2, c.bag[0].src); /* voter replies */
TEST_ASSERT_EQUAL_INT(1, c.bag[0].dst); /* back to candidate */
TEST_ASSERT_EQUAL_INT(1, c.bag[0].granted);
/* Handler ran on dst: node 2 adopted the higher term and recorded the vote. */
TEST_ASSERT_EQUAL_UINT64(vreq.term, c.nodes[2].state.term);
TEST_ASSERT_EQUAL_INT(1, c.nodes[2].state.voted_for);
/* ---- AE_REQ routing ---- */
c.bag_count = 0;
/* Leader appends one client value, then sends AppendEntries to peer 1 — the
* conformance send callback pushes a real AE_REQ onto the bag. */
uint64_t value = 0xABCDEF12u;
uint64_t out_index = 0;
TEST_ASSERT_EQUAL_INT(RAFT_OK,
raft_log_append(&c.nodes[0].state, &value, sizeof(value), &out_index));
TEST_ASSERT_EQUAL_UINT64(1, out_index);
TEST_ASSERT_EQUAL_INT(RAFT_OK,
raft_send_append_entries(&c.nodes[0].state, 1, /*leader_commit=*/0));
TEST_ASSERT_EQUAL_INT(1, c.bag_count);
TEST_ASSERT_EQUAL_INT(RAFT_SIM_MSG_AE_REQ, c.bag[0].kind);
TEST_ASSERT_EQUAL_INT(0, raft_sim_bag_deliver(&c, 0));
/* AE request consumed, one AE_RSP remains, src/dst swapped, success. */
TEST_ASSERT_EQUAL_INT(1, c.bag_count);
TEST_ASSERT_EQUAL_INT(RAFT_SIM_MSG_AE_RSP, c.bag[0].kind);
TEST_ASSERT_EQUAL_INT(1, c.bag[0].src); /* follower replies */
TEST_ASSERT_EQUAL_INT(0, c.bag[0].dst); /* back to leader */
TEST_ASSERT_EQUAL_INT(1, c.bag[0].success);
/* Handler ran on dst: node 1 replicated the single entry. */
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 dst is NOT delivered ---- */
c.bag_count = 0;
raft_sim_partition(&c, 2, 1); /* isolate the voter */
int term2_before = (int)c.nodes[2].state.term;
int votedfor2_before = c.nodes[2].state.voted_for;
raft_sim_msg_t vreq2;
memset(&vreq2, 0, sizeof(vreq2));
vreq2.kind = RAFT_SIM_MSG_VOTE_REQ;
vreq2.src = 1;
vreq2.dst = 2; /* partitioned */
vreq2.term = c.nodes[2].state.term + 9;
TEST_ASSERT_EQUAL_INT(0, raft_sim_bag_push(&c, &vreq2));
TEST_ASSERT_EQUAL_INT(1, c.bag_count);
/* Delivery to a partitioned endpoint is a no-op: distinct non-zero status,
* the request stays in the bag, no reply pushed, dst state unchanged. */
TEST_ASSERT_NOT_EQUAL(0, raft_sim_bag_deliver(&c, 0));
TEST_ASSERT_EQUAL_INT(1, c.bag_count); /* request not consumed */
TEST_ASSERT_EQUAL_INT(RAFT_SIM_MSG_VOTE_REQ, c.bag[0].kind); /* still the request */
TEST_ASSERT_EQUAL_INT(term2_before, (int)c.nodes[2].state.term);
TEST_ASSERT_EQUAL_INT(votedfor2_before, c.nodes[2].state.voted_for);
}
/* ---------- Group A: Basic Election Correctness (task 5.1) ---------- */
/*
@ -7108,6 +7210,9 @@ void test_raft(void)
/* conformance init + in-memory adapter callbacks (task 1.2) */
RUN_TEST(test_sim_conformance_init_append_readback);
/* conformance bag push/deliver routing (task 2.1) */
RUN_TEST(test_sim_conformance_bag_deliver_routes_rpcs);
/* Group A: basic election correctness (task 5.1) */
RUN_TEST(test_proto_A1_single_node_self_elects);
RUN_TEST(test_proto_A2_3node_one_leader);