lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(raft-sim-conformance): seeded PRNG + in-memory adapter callbacks + conformance init (task 1.2)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-24 11:55:11 +00:00
parent 35c012ddc2
commit 631854303c
4 changed files with 307 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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

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

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

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

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

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

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