diff --git a/.kiro/specs/raft-sim-conformance/tasks.md b/.kiro/specs/raft-sim-conformance/tasks.md index 8f42660..cd0861a 100644 --- a/.kiro/specs/raft-sim-conformance/tasks.md +++ b/.kiro/specs/raft-sim-conformance/tasks.md @@ -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 diff --git a/src/lib/raft/raft.h b/src/lib/raft/raft.h index 10b3ec8..f7477ca 100644 --- a/src/lib/raft/raft.h +++ b/src/lib/raft/raft.h @@ -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. diff --git a/src/lib/raft/raft_sim.c b/src/lib/raft/raft_sim.c index f091462..59cf819 100644 --- a/src/lib/raft/raft_sim.c +++ b/src/lib/raft/raft_sim.c @@ -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++) diff --git a/src/lib/testsuite/test_raft.c b/src/lib/testsuite/test_raft.c index bf67a56..ee1b851 100644 --- a/src/lib/testsuite/test_raft.c +++ b/src/lib/testsuite/test_raft.c @@ -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);