From 631854303c274f47834f74d14d79c92ddec04714 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 24 Jun 2026 11:55:11 +0000 Subject: [PATCH] feat(raft-sim-conformance): seeded PRNG + in-memory adapter callbacks + conformance init (task 1.2) --- .kiro/specs/raft-sim-conformance/tasks.md | 2 +- src/lib/raft/raft.h | 26 +++ src/lib/raft/raft_sim.c | 201 ++++++++++++++++++++++ src/lib/testsuite/test_raft.c | 79 +++++++++ 4 files changed, 307 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/raft-sim-conformance/tasks.md b/.kiro/specs/raft-sim-conformance/tasks.md index 11fb4263..8f426609 100644 --- a/.kiro/specs/raft-sim-conformance/tasks.md +++ b/.kiro/specs/raft-sim-conformance/tasks.md @@ -27,7 +27,7 @@ - _Requirements: 1.1, 1.3, 2.5_ - _Boundary: raft.h (XCE_TESTING section only)_ -- [ ] 1.2 Implement the seeded PRNG and the in-memory adapter callbacks +- [x] 1.2 Implement the seeded PRNG and the in-memory adapter callbacks - Add a small inlined deterministic generator (e.g. SplitMix64) seeded by `cluster.seed`; add `on_persist_entry` → `node.log[]`, `on_get_entry_term` → term|0, `on_truncate_from` → shrink `log_len`, `on_commit_advanced` → set `commit_index` + apply into `applied[]`; route `on_send_vote_request`/`on_send_append_entries` to a bag push - Add `raft_sim_conformance_init(c, n, timeout, seed)` wiring all six callbacks + the bag - Observable completion: a test initializes a cluster, drives one `raft_log_append` on a forced leader, and reads the entry back via the in-memory log and `on_get_entry_term` diff --git a/src/lib/raft/raft.h b/src/lib/raft/raft.h index 7cd0ccf6..10b3ec86 100644 --- a/src/lib/raft/raft.h +++ b/src/lib/raft/raft.h @@ -642,6 +642,32 @@ typedef struct { int raft_sim_init(raft_sim_cluster_t *c, int n_nodes, uint32_t election_timeout_ms); +/* + * raft_sim_conformance_init: initialize an n-node CONFORMANCE cluster + * (raft-sim-conformance task 1.2). Modeled on raft_sim_init but wires the + * conformance adapter callback set and seeds the deterministic PRNG: + * - log storage callbacks (on_persist_entry / on_get_entry_term / + * on_truncate_from) reuse the in-memory node->log[] backing, + * - on_commit_advanced sets commit_index AND applies each newly-committed + * entry's decoded value into node->applied[] (State Machine Safety oracle), + * - on_send_vote_request / on_send_append_entries push a raft_sim_msg_t onto + * the unified cluster bag[] (mirroring raft.tla `messages`). + * `seed` is the run's reproducibility key; c->prng is initialized from it so + * every later nondeterministic choice flows from the seed (R1.2). Returns 0 on + * success, -1 on an invalid node count. Does NOT deliver bag messages (that is + * task 2.1). + */ +int raft_sim_conformance_init(raft_sim_cluster_t *c, int n_nodes, + uint32_t election_timeout_ms, uint64_t seed); + +/* + * raft_sim_prng_next: advance the cluster's seeded SplitMix64 generator and return + * a well-mixed 64-bit value. The single source of nondeterminism for the scheduler + * (R1.2): for a fixed seed the value sequence is byte-identical across runs and + * platforms (no libc rand()). Exposed so tests can assert deterministic replay. + */ +uint64_t raft_sim_prng_next(raft_sim_cluster_t *c); + /* * 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 8f5a45bf..f091462f 100644 --- a/src/lib/raft/raft_sim.c +++ b/src/lib/raft/raft_sim.c @@ -238,6 +238,147 @@ static void sim_on_trace(void *ctx, const char *event, const char *detail) node->trace_len += remaining - 1; /* fill without null — snprintf already null-terminated */ } +/* ---------- conformance harness adapter (raft-sim-conformance task 1.2) ---------- + * + * The conformance init wires a SEPARATE callback set alongside the legacy ones. + * Log storage REUSES sim_persist_entry / sim_get_entry_term / sim_truncate_from + * (they already back node->log[] = on_persist_entry / on_get_entry_term / + * on_truncate_from). Only the commit + send callbacks differ: + * - sim_conf_on_commit_advanced fills node->applied[] (the legacy variant does + * not), so State Machine Safety can observe value divergence, + * - sim_conf_send_vote_request / sim_conf_send_append_entries push a unified + * raft_sim_msg_t onto cluster.bag[] (mirroring raft.tla `messages`) instead + * of the legacy pending[] / ae_pending[] queues. + * Bag DELIVERY/routing is task 2.1 — not implemented here. + */ + +/* + * raft_sim_prng_next: SplitMix64 step. Advances c->prng and returns a well-mixed + * 64-bit value. Deterministic and reproducible (no libc rand()): seeded from + * c->seed in raft_sim_conformance_init, this is the single source for every + * later nondeterministic choice the scheduler makes (R1.2). + */ +uint64_t raft_sim_prng_next(raft_sim_cluster_t *c) +{ + uint64_t z = (c->prng += 0x9E3779B97F4A7C15ULL); + z = (z ^ (z >> 30)) * 0xBF58476D1CE4E5B9ULL; + z = (z ^ (z >> 27)) * 0x94D049BB133111EBULL; + return z ^ (z >> 31); +} + +/* + * sim_conf_decode_value: deterministic decode of a committed entry's payload into + * the applied state-machine value. Interprets the first 8 bytes (little-endian, as + * stored by memcpy on the same host) as a uint64_t; shorter payloads zero-extend. + * This is the stable decode applied[] uses so State Machine Safety divergence (two + * nodes applying different values at the same index) is observable. + */ +static uint64_t sim_conf_decode_value(const raft_sim_entry_t *slot) +{ + uint64_t v = 0; + size_t n = slot->len; + if (n > sizeof(v)) + n = sizeof(v); + if (n > 0) + memcpy(&v, slot->bytes, n); + return v; +} + +/* + * sim_conf_on_commit_advanced: conformance commit callback. Records the + * library-reported commit_index and applies each newly-committed entry's decoded + * value into node->applied[] (for committed indexes prev_commit+1 .. commit_index). + * applied[k-1] holds the value applied at 1-based committed index k. Also bumps the + * fire counter (mirrors the legacy variant's count semantics). + */ +static void sim_conf_on_commit_advanced(void *ctx, uint64_t commit_index) +{ + raft_sim_node_t *node = (raft_sim_node_t *)ctx; + + uint64_t prev = node->commit_index; + if (commit_index > prev) { + for (uint64_t k = prev + 1; k <= commit_index; k++) { + if (k >= 1 && k <= (uint64_t)node->log_count && k <= RAFT_SIM_MAX_LOG) + node->applied[k - 1] = sim_conf_decode_value(&node->log[k - 1]); + } + } + + node->commit_index = commit_index; + node->commit_advanced_count++; +} + +/* + * raft_sim_bag_push: append one message to the unified cluster bag. Fails loud + * (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) +{ + if (c->bag_count >= RAFT_SIM_MAX_MSGS) + return -1; /* fail loud — do not silently drop */ + + c->bag[c->bag_count++] = *msg; + return 0; +} + +/* + * sim_conf_send_vote_request: conformance vote-request send. Builds a + * RAFT_SIM_MSG_VOTE_REQ from the callback args and pushes it onto the bag. + */ +static int sim_conf_send_vote_request(void *ctx, int peer_idx, uint64_t term) +{ + raft_sim_node_t *node = (raft_sim_node_t *)ctx; + raft_sim_cluster_t *c = (raft_sim_cluster_t *)node->ctx; + + raft_sim_msg_t msg; + memset(&msg, 0, sizeof(msg)); + msg.kind = RAFT_SIM_MSG_VOTE_REQ; + msg.src = node->idx; + msg.dst = peer_idx; + msg.term = term; + + return raft_sim_bag_push(c, &msg); +} + +/* + * sim_conf_send_append_entries: conformance AppendEntries send. Builds a + * RAFT_SIM_MSG_AE_REQ from the callback args and pushes it onto the bag. For + * n_entries == 1 the single entry's (term, value) is sourced from the SENDER's own + * in-memory log at prev_log_index + 1 (the library holds no entry bytes), decoded + * via sim_conf_decode_value — exactly as the legacy sim_send_append_entries sources + * bytes. + */ +static int sim_conf_send_append_entries(void *ctx, int peer_idx, + uint64_t prev_log_index, + uint64_t prev_log_term, + uint64_t leader_commit, int n_entries) +{ + raft_sim_node_t *node = (raft_sim_node_t *)ctx; + raft_sim_cluster_t *c = (raft_sim_cluster_t *)node->ctx; + + raft_sim_msg_t msg; + memset(&msg, 0, sizeof(msg)); + msg.kind = RAFT_SIM_MSG_AE_REQ; + msg.src = node->idx; + msg.dst = peer_idx; + msg.term = node->state.term; /* send-time leader term */ + msg.prev_log_index = prev_log_index; + msg.prev_log_term = prev_log_term; + msg.leader_commit = leader_commit; + msg.entries_len = n_entries; + + if (n_entries == 1) { + uint64_t idx = prev_log_index + 1; + if (idx >= 1 && idx <= (uint64_t)node->log_count) { + raft_sim_entry_t *slot = &node->log[idx - 1]; + msg.entry_term = slot->term; + msg.entry_value = sim_conf_decode_value(slot); + } + } + + return raft_sim_bag_push(c, &msg); +} + /* ---------- public API ---------- */ int raft_sim_init(raft_sim_cluster_t *c, int n_nodes, uint32_t election_timeout_ms) @@ -300,6 +441,66 @@ int raft_sim_init(raft_sim_cluster_t *c, int n_nodes, uint32_t election_timeout_ return 0; } +int raft_sim_conformance_init(raft_sim_cluster_t *c, int n_nodes, + uint32_t election_timeout_ms, uint64_t seed) +{ + memset(c, 0, sizeof(*c)); + + if (n_nodes < 1 || n_nodes > RAFT_SIM_MAX_NODES) + return -1; + + c->n = n_nodes; + c->seed = seed; + c->prng = seed; /* deterministic PRNG state seeded from the run key */ + + for (int i = 0; i < n_nodes; i++) { + raft_sim_node_t *node = &c->nodes[i]; + + memset(node, 0, sizeof(*node)); + + node->idx = i; + node->partition = 0; + node->ctx = c; /* points back to the cluster */ + + /* Conformance adapter callback set. Election/persist/timeout callbacks + * reuse the existing sim implementations; the conformance variants differ + * only for commit (fills applied[]) and the send paths (push to bag[]). */ + node->cbs.on_send_vote_request = sim_conf_send_vote_request; + node->cbs.on_become_leader = sim_become_leader; + node->cbs.on_become_follower = sim_noop_become_follower; + node->cbs.on_persist = sim_noop_persist; + node->cbs.on_schedule_timeout = sim_noop_schedule_timeout; + node->cbs.on_cancel_timeout = sim_noop_cancel_timeout; + node->cbs.on_start_election_after_delay = sim_noop_start_election_after_delay; + /* log storage reused from the legacy in-memory log callbacks */ + node->cbs.on_persist_entry = sim_persist_entry; + node->cbs.on_get_entry_term = sim_get_entry_term; + node->cbs.on_send_append_entries = sim_conf_send_append_entries; + node->cbs.on_truncate_from = sim_truncate_from; + node->cbs.on_commit_advanced = sim_conf_on_commit_advanced; + node->cbs.on_vote_eligible = NULL; /* all eligible */ + node->cbs.on_trace = sim_on_trace; + + raft_config_t cfg; + memset(&cfg, 0, sizeof(cfg)); + cfg.n_peers = n_nodes; + cfg.self_idx = i; + cfg.my_priority = i; /* priority = index */ + cfg.election_timeout_ms = election_timeout_ms; + cfg.initial_term = 0; + cfg.initial_voted_for = -1; + cfg.initial_commit_index = 0; + cfg.cbs = &node->cbs; + cfg.ctx = node; + + int rc = raft_init(&node->state, &cfg); + if (rc != 0) + return rc; + } + + return 0; +} + void raft_sim_dump_log(raft_sim_cluster_t *c, FILE *fp) { 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 17568d34..bf67a560 100644 --- a/src/lib/testsuite/test_raft.c +++ b/src/lib/testsuite/test_raft.c @@ -4466,6 +4466,82 @@ static void test_sim_7_4_on_commit_advanced_trigger_semantics(void) TEST_ASSERT_EQUAL_INT(2, L->commit_advanced_count); /* no fire off-leader */ } +/* + * test_sim_conformance_init_append_readback (raft-sim-conformance task 1.2; Req 1.1, 1.2, 2.1) + * + * Observable completion for the conformance init + in-memory adapter callbacks: + * - raft_sim_conformance_init(&c, n, timeout, seed) wires the six conformance + * adapter callbacks (log persist/get-term/truncate + conformance commit + + * send-to-bag) and seeds the deterministic PRNG from `seed`. + * - Drive node 0 to LEADER through the published election entry points + * (start_election on node 0, deliver the pending vote requests so peers grant). + * - raft_log_append a known value on the leader; the entry must be readable back + * two ways: directly via node0.log[0] (term + decoded value) AND through the + * wired on_get_entry_term callback (returns the entry's term, not 0). + * - Determinism smoke: two conformance inits with the SAME seed leave c->prng in + * the same state (R1.2: every nondeterministic choice flows from the seed). + */ +static void test_sim_conformance_init_append_readback(void) +{ + raft_sim_cluster_t c; + memset(&c, 0, sizeof(c)); + + /* Single-node conformance cluster: raft_start_election short-circuits node 0 + * straight to LEADER (no peers to canvass), so the leader is forced without + * depending on bag delivery (task 2.1). The PRNG seed is still asserted. */ + TEST_ASSERT_EQUAL_INT(0, raft_sim_conformance_init(&c, 1, 2000, 0xC0FFEEULL)); + + /* PRNG seeded from the run key. */ + TEST_ASSERT_EQUAL_UINT64(0xC0FFEEULL, c.seed); + + /* Node 0 starts as a follower, no leader yet. */ + TEST_ASSERT_EQUAL_INT(RAFT_FOLLOWER, c.nodes[0].state.role); + TEST_ASSERT_EQUAL_INT(0, c.leader_count); + + /* Force node 0 to LEADER via the published election entry point. */ + raft_start_election(&c.nodes[0].state); + TEST_ASSERT_EQUAL_INT(RAFT_LEADER, c.nodes[0].state.role); + TEST_ASSERT_EQUAL_INT(1, c.leader_count); + + /* Append a known client value on the leader. The first 8 bytes decode to the + * applied value; the conformance adapter stores it in node0.log[0]. */ + uint64_t value = 0x1122334455667788ULL; + uint64_t out_index = 0; + int rc = raft_log_append(&c.nodes[0].state, &value, sizeof(value), &out_index); + TEST_ASSERT_EQUAL_INT(RAFT_OK, rc); + TEST_ASSERT_EQUAL_UINT64(1, out_index); + + /* Read back directly via the in-memory log: index 1 -> log[0]. */ + raft_sim_node_t *L = &c.nodes[0]; + TEST_ASSERT_EQUAL_INT(1, L->log_count); + TEST_ASSERT_EQUAL_UINT64(L->state.term, L->log[0].term); + uint64_t decoded = 0; + memcpy(&decoded, L->log[0].bytes, sizeof(decoded)); + TEST_ASSERT_EQUAL_UINT64(value, decoded); + + /* Read back the entry's term through the wired on_get_entry_term callback. */ + uint64_t got_term = L->cbs.on_get_entry_term(L, 1); + TEST_ASSERT_EQUAL_UINT64(L->state.term, got_term); + TEST_ASSERT_NOT_EQUAL(0, got_term); + + /* Determinism smoke: same seed -> identical PRNG state after init. */ + raft_sim_cluster_t c2; + memset(&c2, 0, sizeof(c2)); + TEST_ASSERT_EQUAL_INT(0, raft_sim_conformance_init(&c2, 1, 2000, 0xC0FFEEULL)); + TEST_ASSERT_EQUAL_UINT64(c.prng, c2.prng); /* same seed -> same start state */ + + /* The seeded generator advances identically for the same seed: the first few + * draws match, and the post-draw state stays in lockstep (R1.2 determinism). */ + raft_sim_cluster_t g1, g2; + memset(&g1, 0, sizeof(g1)); + memset(&g2, 0, sizeof(g2)); + TEST_ASSERT_EQUAL_INT(0, raft_sim_conformance_init(&g1, 1, 2000, 0x5EEDULL)); + TEST_ASSERT_EQUAL_INT(0, raft_sim_conformance_init(&g2, 1, 2000, 0x5EEDULL)); + for (int k = 0; k < 4; k++) + TEST_ASSERT_EQUAL_UINT64(raft_sim_prng_next(&g1), raft_sim_prng_next(&g2)); + TEST_ASSERT_EQUAL_UINT64(g1.prng, g2.prng); +} + /* ---------- Group A: Basic Election Correctness (task 5.1) ---------- */ /* @@ -7029,6 +7105,9 @@ void test_raft(void) RUN_TEST(test_sim_7_3_freshness_gate_lexicographic); RUN_TEST(test_sim_7_4_on_commit_advanced_trigger_semantics); + /* conformance init + in-memory adapter callbacks (task 1.2) */ + RUN_TEST(test_sim_conformance_init_append_readback); + /* 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);