wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity
myxce/src/lib/raft/raft_sim.c

1521 lines
60 KiB
C
Raw Blame History

#ifdef XCE_TESTING
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "raft.h"
/* ---------- no-op callbacks ---------- */
static void sim_noop_become_follower(void *ctx, uint64_t term, int leader_idx)
{
(void)ctx; (void)term; (void)leader_idx;
}
static int sim_noop_persist(void *ctx, uint64_t term, int voted_for)
{
(void)ctx; (void)term; (void)voted_for;
return 0;
}
static void sim_noop_schedule_timeout(void *ctx, uint32_t delay_ms)
{
(void)ctx; (void)delay_ms;
}
static void sim_noop_cancel_timeout(void *ctx)
{
(void)ctx;
}
static void sim_noop_start_election_after_delay(void *ctx, uint32_t delay_ms)
{
(void)ctx; (void)delay_ms;
}
/* ---------- wired callbacks ---------- */
/*
* sim_send_vote_request: buffer a pending RPC into the cluster queue.
*
* ctx is raft_sim_node_t * (the source node).
* The node's ctx field points to raft_sim_cluster_t *.
*/
static int sim_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;
if (c->pending_count >= 64)
return -1; /* queue full — drop */
raft_sim_rpc_t *rpc = &c->pending[c->pending_count++];
rpc->src = node->idx;
rpc->dst = peer_idx;
rpc->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 == term of the tail entry (0 when the log is empty). This
* feeds the (lastLogTerm, lastLogIndex) freshness gate at delivery. */
rpc->last_log_index = (uint64_t)node->log_count;
rpc->last_log_term = (node->log_count > 0)
? node->log[node->log_count - 1].term
: 0;
return 0;
}
/*
* sim_become_leader: increment cluster-level leader_count.
*
* ctx is raft_sim_node_t *.
*/
static void sim_become_leader(void *ctx, uint64_t term)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
raft_sim_cluster_t *c = (raft_sim_cluster_t *)node->ctx;
(void)term;
c->leader_count++;
}
/* ---------- log-replication callbacks (raft-log-replication) ----------
*
* These back the four required log callbacks with the node's in-memory log.
* ctx is raft_sim_node_t * (same convention as the wired election callbacks).
* The library holds no entry bytes, so the simulator owns storage here:
* - on_persist_entry: append/overwrite at 1-based index (term + bytes)
* - on_get_entry_term: read the stored term (0 = absent)
* - on_truncate_from: drop entries index..end
* - on_send_append_entries: buffer one AE (sourcing the single entry from the
* sender's own log at prev_log_index+1 when n_entries == 1)
* No AppendEntries delivery happens here — that is raft_sim_deliver_append_entries
* (task 8.2). These callbacks only do storage/term-read/truncate/buffer.
*/
/*
* sim_persist_entry: durably append or overwrite the entry at 1-based `index`.
* Sets the slot's term and copies up to RAFT_SIM_ENTRY_LEN bytes; extends
* log_count when appending past the current tail.
*/
static int sim_persist_entry(void *ctx, uint64_t index, uint64_t term,
const void *entry, size_t len)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
if (index < 1 || index > RAFT_SIM_MAX_LOG)
return -1; /* out of fixed sim bounds */
raft_sim_entry_t *slot = &node->log[index - 1];
slot->term = term;
size_t n = len;
if (n > RAFT_SIM_ENTRY_LEN)
n = RAFT_SIM_ENTRY_LEN; /* bounded copy */
if (entry && n > 0)
memcpy(slot->bytes, entry, n);
slot->len = n;
if ((int)index > node->log_count)
node->log_count = (int)index;
return 0;
}
/*
* sim_get_entry_term: return the stored term at 1-based `index`, or 0 if absent
* (0-means-absent convention).
*/
static uint64_t sim_get_entry_term(void *ctx, uint64_t index)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
if (index < 1 || index > (uint64_t)node->log_count)
return 0;
return node->log[index - 1].term;
}
/*
* sim_truncate_from: drop entries at `index`..end (clear term so they read as
* absent) and shrink log_count.
*/
static int sim_truncate_from(void *ctx, uint64_t index)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
if (index < 1)
return -1;
for (uint64_t i = index; i <= (uint64_t)node->log_count; i++) {
if (i <= RAFT_SIM_MAX_LOG) {
raft_sim_entry_t *slot = &node->log[i - 1];
memset(slot, 0, sizeof(*slot)); /* term=0 => absent */
}
}
if (node->log_count >= (int)index)
node->log_count = (int)index - 1;
return 0;
}
/*
* sim_send_append_entries: buffer one AppendEntries into the cluster AE queue.
* For n_entries == 1 the single entry's index/term/bytes are sourced from the
* sender's own in-memory log at prev_log_index + 1 (the library holds no bytes).
*/
static int sim_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;
if (c->ae_pending_count >= RAFT_SIM_MAX_AE)
return -1; /* queue full — drop */
raft_sim_ae_t *ae = &c->ae_pending[c->ae_pending_count++];
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;
ae->n_entries = n_entries;
if (n_entries == 1) {
uint64_t idx = prev_log_index + 1;
ae->entry_index = idx;
if (idx >= 1 && idx <= (uint64_t)node->log_count) {
raft_sim_entry_t *slot = &node->log[idx - 1];
ae->entry_term = slot->term;
ae->entry_len = slot->len;
if (slot->len > 0)
memcpy(ae->entry_bytes, slot->bytes, slot->len);
}
}
return 0;
}
/*
* sim_on_commit_advanced: record the library-reported commit_index and bump the
* fire counter so tests can assert the strict-advance / fire-once semantics. The
* library fires this ONLY on a strict commit_index advance (never on a held
* index), so the count == number of strict advances. No durability here — the
* sim has no commit barrier; this is the trigger mirror only.
*/
static void sim_on_commit_advanced(void *ctx, uint64_t commit_index)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
node->commit_index = commit_index;
node->commit_advanced_count++;
}
/*
* sim_on_trace: append the event string to the node's trace_buf.
*
* ctx is raft_sim_node_t *.
*/
static void sim_on_trace(void *ctx, const char *event, const char *detail)
{
raft_sim_node_t *node = (raft_sim_node_t *)ctx;
(void)detail;
if (!event)
return;
int remaining = (int)sizeof(node->trace_buf) - node->trace_len - 1;
if (remaining <= 0)
return;
int written = snprintf(node->trace_buf + node->trace_len, (size_t)remaining,
"%s\n", event);
if (written > 0 && written < remaining)
node->trace_len += written;
else if (written >= remaining)
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.
*/
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;
/* 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);
}
/*
* 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)
{
memset(c, 0, sizeof(*c));
if (n_nodes < 1 || n_nodes > RAFT_SIM_MAX_NODES)
return -1;
c->n = n_nodes;
c->leader_count = 0;
c->pending_count = 0;
for (int i = 0; i < n_nodes; i++) {
raft_sim_node_t *node = &c->nodes[i];
/* Zero the node struct */
memset(node, 0, sizeof(*node));
node->idx = i;
node->partition = 0;
node->ctx = c; /* points back to the cluster */
/* Wire callbacks */
node->cbs.on_send_vote_request = sim_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-replication callbacks backed by the node's in-memory log */
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_send_append_entries;
node->cbs.on_truncate_from = sim_truncate_from;
node->cbs.on_commit_advanced = sim_on_commit_advanced;
node->cbs.on_vote_eligible = NULL; /* optional — treat all as eligible */
node->cbs.on_trace = sim_on_trace;
/* Initialize raft state — pass node as ctx so callbacks can find both
* node->idx (source) and node->ctx (cluster). */
raft_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = n_nodes;
cfg.self_idx = i;
cfg.my_priority = i; /* priority = index: node 0 has highest priority */
cfg.election_timeout_ms = election_timeout_ms;
cfg.initial_term = 0;
cfg.initial_voted_for = -1;
cfg.initial_commit_index = 0; /* first boot — library never reads storage */
cfg.cbs = &node->cbs;
cfg.ctx = node; /* node* as ctx — callbacks cast back to raft_sim_node_t* */
int rc = raft_init(&node->state, &cfg);
if (rc != 0)
return rc;
}
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++) {
raft_sim_node_t *node = &c->nodes[i];
if (node->trace_len > 0) {
fprintf(fp, "[node%d trace]\n%.*s\n",
i, node->trace_len, node->trace_buf);
}
}
}
void raft_sim_deliver_pending(raft_sim_cluster_t *c)
{
/* Snapshot current count — callbacks may add more entries */
int count = c->pending_count;
c->pending_count = 0;
for (int i = 0; i < count; i++) {
raft_sim_rpc_t *rpc = &c->pending[i];
int src = rpc->src;
int dst = rpc->dst;
uint64_t term = rpc->term;
/* Drop if either endpoint is partitioned */
if (c->nodes[src].partition || c->nodes[dst].partition)
continue;
/* Deliver vote request to dst, 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, term,
rpc->last_log_term,
rpc->last_log_index,
&reply_term);
/* Route reply back to src */
raft_on_vote_reply(&c->nodes[src].state, dst, granted, reply_term);
}
}
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);
}
}
/*
* 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 */
}
/*
* raft_sim_bag_drop: DropMessage (raft.tla:448). Discard the bag member at `i`
* (swap-with-last removal, bag_count -= 1) without touching ANY node. Mirrors the
* TLA+ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>: this is a
* pure bag mutation. Bounds-checked; returns -1 (fail loud, bag untouched) on an
* out-of-range index, 0 on a discarded member.
*/
int raft_sim_bag_drop(raft_sim_cluster_t *c, int i)
{
if (i < 0 || i >= c->bag_count)
return -1; /* fail loud — no silent no-op */
raft_sim_bag_remove_at(c, i); /* swap-with-last; decrements bag_count */
return 0;
}
/*
* raft_sim_bag_duplicate: DuplicateMessage (raft.tla:443). Re-enqueue a COPY of the
* bag member at `i` onto the tail (multiplicity += 1) without touching ANY node —
* mirrors the TLA+ UNCHANGED of all non-message vars. Honors RAFT_SIM_MAX_MSGS via
* raft_sim_bag_push (fail loud on overflow). The original at `i` is left intact.
* Returns -1 on an out-of-range index, the push status otherwise (0 on success,
* -1 on bag overflow).
*/
int raft_sim_bag_duplicate(raft_sim_cluster_t *c, int i)
{
if (i < 0 || i >= c->bag_count)
return -1; /* fail loud — no silent no-op */
/* Copy by value FIRST: raft_sim_bag_push may grow the array, but the source is
* already an independent local, so a tail push can never alias the source. */
raft_sim_msg_t copy = c->bag[i];
return raft_sim_bag_push(c, &copy);
}
/*
* raft_sim_bag_pick: Reorder source. Return a PRNG-chosen in-flight bag index in
* [0, bag_count) so the scheduler delivers in a deterministic, seed-replayable
* order rather than FIFO. Pure read of the bag: advances only the cluster PRNG, never
* any node's (term, role, votedFor, log). Returns -1 when the bag is empty (nothing
* to deliver). The subsequent raft_sim_bag_deliver does the node-side work.
*/
int raft_sim_bag_pick(raft_sim_cluster_t *c)
{
if (c->bag_count <= 0)
return -1; /* empty bag — nothing to pick */
return (int)(raft_sim_prng_next(c) % (uint64_t)c->bag_count);
}
void raft_sim_start_all_elections(raft_sim_cluster_t *c)
{
for (int i = 0; i < c->n; i++)
raft_start_election(&c->nodes[i].state);
}
void raft_sim_fire_timeout(raft_sim_cluster_t *c, int node_idx)
{
if (node_idx >= 0 && node_idx < c->n)
raft_on_election_timeout(&c->nodes[node_idx].state);
}
void raft_sim_partition(raft_sim_cluster_t *c, int node_idx, int isolated)
{
if (node_idx >= 0 && node_idx < c->n)
c->nodes[node_idx].partition = isolated;
}
void raft_sim_advance_commit(raft_sim_cluster_t *c, int node_idx)
{
if (node_idx < 0 || node_idx >= c->n)
return;
/* Library decides whether the commit index advanced; on a strict advance the
* node's on_commit_advanced (sim_on_commit_advanced) updates commit_index /
* commit_advanced_count. The -EPERM off-leader return is intentionally
* discarded — tests read the per-node mirror, not this status. */
(void)raft_advance_commit(&c->nodes[node_idx].state);
}
/* Two log entries are identical iff their term, length, and payload bytes
* (over the recorded length) all match. */
static int sim_entry_identical(const raft_sim_entry_t *a,
const raft_sim_entry_t *b)
{
if (a->term != b->term || a->len != b->len)
return 0;
return memcmp(a->bytes, b->bytes, a->len) == 0;
}
int raft_sim_check_log_matching(const raft_sim_cluster_t *c)
{
/* For every ordered-pair-free pair (a < b) and every index i present in
* both logs: if the entries at i carry the same term, the prefixes 1..i
* must be entry-identical. Pairs are symmetric, so a < b suffices. */
for (int a = 0; a < c->n; a++) {
const raft_sim_node_t *na = &c->nodes[a];
for (int b = a + 1; b < c->n; b++) {
const raft_sim_node_t *nb = &c->nodes[b];
int shared = na->log_count < nb->log_count
? na->log_count : nb->log_count;
for (int i = 1; i <= shared; i++) {
const raft_sim_entry_t *ea = &na->log[i - 1];
const raft_sim_entry_t *eb = &nb->log[i - 1];
/* Same (index, term) ⇒ prefixes through i must be identical. */
if (ea->term != eb->term)
continue;
for (int k = 1; k <= i; k++) {
if (!sim_entry_identical(&na->log[k - 1], &nb->log[k - 1]))
return 0; /* Log Matching violated */
}
}
}
}
return 1; /* holds */
}
int raft_sim_check_election_safety(const raft_sim_cluster_t *c)
{
/* For every distinct pair (a < b): fail if BOTH are LEADER at the SAME term.
* Pairs are symmetric, so a < b suffices. Role/term are read from each node's
* observable snapshot (Req 3.6) — no internal protocol bookkeeping. */
for (int a = 0; a < c->n; a++) {
raft_snapshot_t sa;
raft_snapshot(&c->nodes[a].state, &sa);
if (sa.role != RAFT_LEADER)
continue;
for (int b = a + 1; b < c->n; b++) {
raft_snapshot_t sb;
raft_snapshot(&c->nodes[b].state, &sb);
if (sb.role == RAFT_LEADER && sb.term == sa.term)
return 0; /* two LEADERs at the same term — Election Safety violated */
}
}
return 1; /* holds */
}
int raft_sim_check_leader_completeness(const raft_sim_cluster_t *c)
{
/* Leader Completeness: an entry committed in term t must be present (same
* index, same term) in the log of every leader of a strictly-higher term.
*
* Pure per-snapshot observer (Req 3.6): the committed set is every index k in
* 1..commit_index over ALL nodes (carrying that node's log[k-1].term); the
* current leaders are the role==RAFT_LEADER nodes. For each leader L at term
* T_L and each committed (k, t) with t < T_L, L's log must hold index k at
* term t — else the property is violated. (index, term) presence suffices;
* Log Matching covers payload identity. */
for (int L = 0; L < c->n; L++) {
raft_snapshot_t sl;
raft_snapshot(&c->nodes[L].state, &sl);
if (sl.role != RAFT_LEADER)
continue;
const raft_sim_node_t *leader = &c->nodes[L];
/* Scan every node's committed prefix for entries committed under a term
* strictly lower than this leader's term. */
for (int d = 0; d < c->n; d++) {
const raft_sim_node_t *src = &c->nodes[d];
int committed = (int)src->commit_index;
if (committed > src->log_count)
committed = src->log_count;
for (int k = 1; k <= committed; k++) {
uint64_t t = src->log[k - 1].term;
if (t >= sl.term)
continue; /* only lower-term committed entries bind */
/* Leader L must contain index k at term t. */
if (k > leader->log_count)
return 0; /* leader missing the committed index */
if (leader->log[k - 1].term != t)
return 0; /* leader has a different term at index k */
}
}
}
return 1; /* holds */
}
int raft_sim_check_state_machine_safety(const raft_sim_cluster_t *c)
{
/* State Machine Safety: no two nodes have applied a different value at the same
* (1-based) log index. Observable-only (Req 3.6): each node's applied extent is
* 1..commit_index (on_commit_advanced fills applied[k-1] for committed index k).
* For every pair (a < b) and every index k in the shared applied prefix
* 1..min(commit_a, commit_b), fail if applied[k-1] differs. Pairs are symmetric,
* so a < b suffices. */
for (int a = 0; a < c->n; a++) {
const raft_sim_node_t *na = &c->nodes[a];
for (int b = a + 1; b < c->n; b++) {
const raft_sim_node_t *nb = &c->nodes[b];
uint64_t shared = na->commit_index < nb->commit_index
? na->commit_index : nb->commit_index;
if (shared > RAFT_SIM_MAX_LOG)
shared = RAFT_SIM_MAX_LOG;
for (uint64_t k = 1; k <= shared; k++) {
if (na->applied[k - 1] != nb->applied[k - 1])
return 0; /* divergent applied value — State Machine Safety violated */
}
}
}
return 1; /* holds */
}
raft_sim_invariant_t raft_sim_check_all(const raft_sim_cluster_t *c)
{
/* Evaluate the four invariants in the FIXED order required by Req 3.5 and
* short-circuit on the FIRST violation, so the reported failing invariant is
* deterministic. Each checker returns 1=holds / 0=violated. */
if (!raft_sim_check_election_safety(c))
return RAFT_SIM_INV_ELECTION_SAFETY;
if (!raft_sim_check_log_matching(c))
return RAFT_SIM_INV_LOG_MATCHING;
if (!raft_sim_check_leader_completeness(c))
return RAFT_SIM_INV_LEADER_COMPLETENESS;
if (!raft_sim_check_state_machine_safety(c))
return RAFT_SIM_INV_STATE_MACHINE_SAFETY;
return RAFT_SIM_INV_NONE; /* all four hold */
}
const char *raft_sim_invariant_name(raft_sim_invariant_t inv)
{
switch (inv) {
case RAFT_SIM_INV_NONE: return "none";
case RAFT_SIM_INV_ELECTION_SAFETY: return "election-safety";
case RAFT_SIM_INV_LOG_MATCHING: return "log-matching";
case RAFT_SIM_INV_LEADER_COMPLETENESS: return "leader-completeness";
case RAFT_SIM_INV_STATE_MACHINE_SAFETY: return "state-machine-safety";
}
return "unknown";
}
const char *raft_sim_action_kind_name(raft_sim_action_kind_t kind)
{
switch (kind) {
case RAFT_SIM_ACT_RESTART: return "restart";
case RAFT_SIM_ACT_TIMEOUT: return "timeout";
case RAFT_SIM_ACT_REQUEST_VOTE: return "request-vote";
case RAFT_SIM_ACT_BECOME_LEADER: return "become-leader";
case RAFT_SIM_ACT_CLIENT_REQUEST: return "client-request";
case RAFT_SIM_ACT_APPEND_ENTRIES: return "append-entries";
case RAFT_SIM_ACT_ADVANCE_COMMIT: return "advance-commit";
case RAFT_SIM_ACT_RECEIVE: return "receive";
case RAFT_SIM_ACT_DUPLICATE_MESSAGE: return "duplicate-message";
case RAFT_SIM_ACT_DROP_MESSAGE: return "drop-message";
}
return "unknown";
}
/*
* raft_sim_trace_record: append the chosen action to c->trace[] at trace_len and
* advance trace_len (Req 1.4 ordering — the scheduler calls this BEFORE applying the
* action). RAFT_SIM_MAX_STEPS bounds the buffer: an overflow is a harness bug (the
* step budget is fixed), so fail loud via assert rather than writing out of bounds
* or silently dropping the action.
*/
void raft_sim_trace_record(raft_sim_cluster_t *c, raft_sim_action_t action)
{
/* Fail loud on overflow — a full trace means the run exceeded its fixed step
* budget, which is a harness bug, not a silently-droppable condition. */
assert(c->trace_len < RAFT_SIM_MAX_STEPS);
if (c->trace_len >= RAFT_SIM_MAX_STEPS)
return; /* belt-and-suspenders: never write OOB if assert is compiled out */
c->trace[c->trace_len++] = action;
}
void raft_sim_dump_trace(const raft_sim_cluster_t *c, FILE *fp,
raft_sim_invariant_t failing)
{
/* Quiet-on-success (Req 5.4): a clean run dumps nothing unless verbose tracing
* was explicitly requested. A real failure always dumps. */
if (failing == RAFT_SIM_INV_NONE && getenv("RAFT_SIM_VERBOSE") == NULL)
return;
fprintf(fp, "raft-sim trace: seed=%llu n=%d failing-invariant=%s steps=%d\n",
(unsigned long long)c->seed, c->n,
raft_sim_invariant_name(failing), c->trace_len);
for (int i = 0; i < c->trace_len; i++) {
const raft_sim_action_t *a = &c->trace[i];
/* One human-readable line per action (Req 5.2): kind name, actor,
* msg_index, term, index. */
fprintf(fp,
" [%d] %s actor=%d msg_index=%d term=%llu index=%llu\n",
i, raft_sim_action_kind_name(a->kind), a->actor, a->msg_index,
(unsigned long long)a->term, (unsigned long long)a->index);
}
}
int raft_sim_restart_node(raft_sim_cluster_t *c, int node_idx)
{
if (node_idx < 0 || node_idx >= c->n)
return -1;
raft_sim_node_t *node = &c->nodes[node_idx];
/* Snapshot the DURABLE state that survives a process restart: the persisted
* (term, voted_for) and the log length (the in-memory log[] models the WAL on
* disk and is left untouched). Volatile consensus state is discarded by the
* re-init below. */
uint64_t restored_term = node->state.term;
int restored_voted_for = node->state.voted_for;
uint64_t restored_log_len = (uint64_t)node->log_count;
/* The committed LSN is durable too: on recovery the controller restores it
* into initial_commit_index. Model that here so a restarted node retains its
* committed position (never regresses). */
uint64_t restored_commit_index = node->state.commit_index;
raft_config_t cfg;
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = node->state.n_peers;
cfg.self_idx = node->idx;
cfg.my_priority = node->idx;
cfg.election_timeout_ms = node->state.election_timeout_ms;
cfg.initial_term = restored_term;
cfg.initial_voted_for = restored_voted_for;
cfg.initial_log_len = restored_log_len;
cfg.initial_commit_index = restored_commit_index;
cfg.cbs = &node->cbs;
cfg.ctx = node; /* same ctx convention as raft_sim_init */
/* raft_init returns the node to FOLLOWER and clears votes/election/leader
* progress while restoring (term, voted_for, log_len). The log[] array and
* the partition flag are preserved. */
return raft_init(&node->state, &cfg);
}
/* ---------- scheduler step (raft-sim-conformance task 4.2) ---------- */
/*
* sim_find_leader: return the index of a node currently in role RAFT_LEADER (read
* via raft_snapshot — observable state only), or -1 if none. When more than one
* node reports LEADER (only possible AFTER an invariant violation, which the
* post-step check then catches), the lowest index is returned deterministically.
*/
static int sim_find_leader(const raft_sim_cluster_t *c)
{
for (int i = 0; i < c->n; i++) {
raft_snapshot_t snap;
raft_snapshot(&c->nodes[i].state, &snap);
if (snap.role == RAFT_LEADER)
return i;
}
return -1;
}
int raft_sim_step(raft_sim_cluster_t *c)
{
/* ---- 1. Compute the ENABLED action set (weighted) ----
* Build the eligible-action multiset for the CURRENT state. Only enabled
* actions are eligible (design "Action scheduler"); weighting is achieved by
* how many slots each kind occupies in `enabled[]`, while every enabled kind
* keeps a NON-ZERO weight (so the whole action space stays reachable). The
* weights bias the run toward LIVENESS — delivery and leader progress dominate,
* faults (drop/duplicate) and restart remain occasional perturbations — so a
* fresh cluster reliably elects a leader, replicates, and ADVANCES COMMIT within
* a bounded run, while still exercising every fault kind. RESTART and TIMEOUT are
* always enabled (n >= 1) so the cluster stays live; leader-gated actions
* (CLIENT_REQUEST, APPEND_ENTRIES, ADVANCE_COMMIT) require a LEADER; bag-gated
* actions (RECEIVE, DROP, DUPLICATE) require a non-empty bag. */
enum { SIM_W_RESTART = 1, SIM_W_TIMEOUT = 2, SIM_W_CLIENT = 3,
SIM_W_ADVANCE = 3, SIM_W_APPEND = 4, SIM_W_RECEIVE = 6,
SIM_W_FAULT = 1 };
/* Upper bound: sum of all weights above (every kind enabled at once). */
raft_sim_action_kind_t enabled[SIM_W_RESTART + SIM_W_TIMEOUT + SIM_W_CLIENT +
SIM_W_ADVANCE + SIM_W_APPEND + SIM_W_RECEIVE +
2 * SIM_W_FAULT];
int n_enabled = 0;
for (int w = 0; w < SIM_W_RESTART; w++)
enabled[n_enabled++] = RAFT_SIM_ACT_RESTART; /* occasional process restart */
for (int w = 0; w < SIM_W_TIMEOUT; w++)
enabled[n_enabled++] = RAFT_SIM_ACT_TIMEOUT; /* (re)start an election */
int leader = sim_find_leader(c);
int has_leader = (leader >= 0);
/* A leader needs at least one OTHER node to replicate to for AppendEntries. */
int has_peer = (c->n > 1);
if (has_leader) {
for (int w = 0; w < SIM_W_CLIENT; w++)
enabled[n_enabled++] = RAFT_SIM_ACT_CLIENT_REQUEST;
for (int w = 0; w < SIM_W_ADVANCE; w++)
enabled[n_enabled++] = RAFT_SIM_ACT_ADVANCE_COMMIT;
if (has_peer)
for (int w = 0; w < SIM_W_APPEND; w++)
enabled[n_enabled++] = RAFT_SIM_ACT_APPEND_ENTRIES;
}
if (c->bag_count > 0) {
for (int w = 0; w < SIM_W_RECEIVE; w++)
enabled[n_enabled++] = RAFT_SIM_ACT_RECEIVE;
for (int w = 0; w < SIM_W_FAULT; w++)
enabled[n_enabled++] = RAFT_SIM_ACT_DROP_MESSAGE;
for (int w = 0; w < SIM_W_FAULT; w++)
enabled[n_enabled++] = RAFT_SIM_ACT_DUPLICATE_MESSAGE;
}
/* ---- 2. PRNG-pick one enabled action ---- */
raft_sim_action_kind_t kind =
enabled[raft_sim_prng_next(c) % (uint64_t)n_enabled];
/* The action's operands (actor / message index / referenced term+index) are
* filled per-kind below, recorded BEFORE applying, then applied. */
raft_sim_action_t action;
memset(&action, 0, sizeof(action));
action.kind = kind;
action.actor = -1;
action.msg_index = -1;
action.term = 0;
action.index = 0;
switch (kind) {
case RAFT_SIM_ACT_RESTART: {
int idx = (int)(raft_sim_prng_next(c) % (uint64_t)c->n);
action.actor = idx;
action.term = c->nodes[idx].state.term;
raft_sim_trace_record(c, action);
(void)raft_sim_restart_node(c, idx);
break;
}
case RAFT_SIM_ACT_TIMEOUT: {
int idx = (int)(raft_sim_prng_next(c) % (uint64_t)c->n);
action.actor = idx;
action.term = c->nodes[idx].state.term;
raft_sim_trace_record(c, action);
/* TLA+ Timeout: a node becomes candidate and campaigns. Drive the published
* raft_start_election entry point — the library performs the term bump,
* self-vote, persist, and emits vote requests onto the bag (R2.1). (The
* separate raft_on_election_timeout is the give-up/step-down path, not the
* campaign-start action.) */
(void)raft_start_election(&c->nodes[idx].state);
break;
}
case RAFT_SIM_ACT_CLIENT_REQUEST: {
/* Append a synthetic monotone client value on the leader so State Machine
* Safety can observe divergence. Derive a distinguishable value from the
* PRNG (R2.1 ClientRequest via raft_log_append). */
uint64_t value = raft_sim_prng_next(c);
uint64_t out_index = 0;
action.actor = leader;
/* Record the leader's NEXT index as the referenced operand BEFORE applying
* (R1.4): the entry will land at the current log tail + 1. */
action.index = (uint64_t)c->nodes[leader].log_count + 1;
raft_sim_trace_record(c, action);
(void)raft_log_append(&c->nodes[leader].state, &value, sizeof(value),
&out_index);
break;
}
case RAFT_SIM_ACT_APPEND_ENTRIES: {
/* Pick a peer != leader and emit one AppendEntries. raft_send_append_entries
* computes prev_log_index/prev_log_term from the library's per-peer progress
* (next_index[peer]); the harness only supplies leader_commit (R2.2). */
int peer = (int)(raft_sim_prng_next(c) % (uint64_t)c->n);
if (peer == leader)
peer = (peer + 1) % c->n;
action.actor = leader;
action.msg_index = peer; /* record the target peer in the operand slot */
raft_sim_trace_record(c, action);
(void)raft_send_append_entries(&c->nodes[leader].state, peer,
c->nodes[leader].state.commit_index);
break;
}
case RAFT_SIM_ACT_ADVANCE_COMMIT: {
/* The library applies the Figure-8 current-term commit rule; on a strict
* advance the leader's on_commit_advanced applies committed values into
* applied[] (R2.3). The harness only triggers the recompute. */
action.actor = leader;
raft_sim_trace_record(c, action);
(void)raft_advance_commit(&c->nodes[leader].state);
break;
}
case RAFT_SIM_ACT_RECEIVE: {
/* Deliver a PRNG-picked bag message through its library handler. This
* organically covers RequestVote propagation, BecomeLeader (a quorum-
* completing vote-rsp makes raft_on_vote_reply elect the candidate), AE
* request/response, and UpdateTerm/stale-drop — all INSIDE the handlers,
* never pre-filtered (R2.4/R2.6). */
int i = raft_sim_bag_pick(c);
action.msg_index = i;
if (i >= 0 && i < c->bag_count) {
action.actor = c->bag[i].dst;
action.term = c->bag[i].term;
}
raft_sim_trace_record(c, action);
if (i >= 0)
(void)raft_sim_bag_deliver(c, i);
break;
}
case RAFT_SIM_ACT_DUPLICATE_MESSAGE: {
/* Bag-only fault: re-enqueue a copy of a PRNG-picked member (R2.5). The bag
* is treated uniformly (raft.tla DuplicateMessage operates on the whole
* `messages` bag); the harness never pre-filters by kind. */
int i = raft_sim_bag_pick(c);
action.msg_index = i;
raft_sim_trace_record(c, action);
if (i >= 0)
(void)raft_sim_bag_duplicate(c, i);
break;
}
case RAFT_SIM_ACT_DROP_MESSAGE: {
/* Bag-only fault: discard a PRNG-picked member (R2.5). */
int i = raft_sim_bag_pick(c);
action.msg_index = i;
raft_sim_trace_record(c, action);
if (i >= 0)
(void)raft_sim_bag_drop(c, i);
break;
}
case RAFT_SIM_ACT_REQUEST_VOTE:
case RAFT_SIM_ACT_BECOME_LEADER:
/* Never SELECTED as standalone scheduler actions: RequestVote and
* BecomeLeader are realized organically by the RECEIVE path (delivering a
* vote-request / the quorum-completing vote-response through the library
* handlers — design "Action scheduler"). Listed here only so the dispatch is
* exhaustive over raft_sim_action_kind_t; reached only if a future caller
* forges one, in which case it is a recorded no-op. */
raft_sim_trace_record(c, action);
break;
}
/* ---- 5. Check all four invariants AFTER applying (Req 3.5) ----
* On a violation, dump the replayable trace and return < 0 (the negated failing
* invariant). A clean step returns 0. */
raft_sim_invariant_t failing = raft_sim_check_all(c);
if (failing != RAFT_SIM_INV_NONE) {
raft_sim_dump_trace(c, stderr, failing);
return -(int)failing;
}
return 0;
}
/* ---------- multi-step run driver (raft-sim-conformance task 4.4) ---------- */
int raft_sim_run(raft_sim_cluster_t *c, int max_steps)
{
/* Bounded loop of single steps. raft_sim_step is the SOLE consumer of c->prng,
* so a run is a pure deterministic function of (n, seed) and the step count —
* two same-keyed clusters produce identical traces and final state (R1.5/1.6/
* 5.3). Halt on the FIRST violation: raft_sim_step returns < 0 and has already
* dumped the replayable trace, so propagate that value unchanged (R3.5). A clean
* exhaustion of the step budget returns 0. */
for (int step = 0; step < max_steps; step++) {
int rc = raft_sim_step(c);
if (rc < 0)
return rc; /* first invariant violation — stop and surface it */
}
return 0;
}
/* ---------- partition/heal scenario (raft-sim-conformance task 5.1) ---------- */
/*
* sim_scen_check: run all four invariants over `c`. On a violation, dump the
* replayable trace to stderr and return the negated failing invariant (matching the
* raft_sim_step / raft_sim_run contract); 0 when all hold. The scenario calls this
* after EVERY action it applies so a violation is surfaced the moment it appears
* (Req 4.1 — invariants asserted THROUGHOUT, not only at the end).
*/
static int sim_scen_check(raft_sim_cluster_t *c)
{
raft_sim_invariant_t failing = raft_sim_check_all(c);
if (failing != RAFT_SIM_INV_NONE) {
raft_sim_dump_trace(c, stderr, failing);
return -(int)failing;
}
return 0;
}
/*
* sim_scen_drain_bag: deliver every in-flight bag message (PRNG-picked order, so a
* partition leaves partitioned-endpoint messages parked rather than delivered),
* running the invariant check after EACH delivery. `cap` bounds the work so a
* duplicate/heal storm can never spin unbounded. Stops early and propagates < 0 on
* the first violation. Returns 0 when the bag drains (or only undeliverable,
* partitioned messages remain) with no violation.
*/
static int sim_scen_drain_bag(raft_sim_cluster_t *c, int cap)
{
for (int g = 0; g < cap && c->bag_count > 0; g++) {
int i = raft_sim_bag_pick(c);
if (i < 0)
break;
int rc = raft_sim_bag_deliver(c, i);
/* rc == -2 means the picked message targets a partitioned endpoint and was
* left in the bag (a heal can deliver it later). To avoid re-picking the same
* parked message forever, sweep for ANY deliverable message; if none remain,
* the bag holds only partitioned traffic and we are done draining for now. */
if (rc == -2) {
int progressed = 0;
for (int j = 0; j < c->bag_count; j++) {
int s = c->bag[j].src, d = c->bag[j].dst;
if (!c->nodes[s].partition && !c->nodes[d].partition) {
if (raft_sim_bag_deliver(c, j) >= 0)
progressed = 1;
break;
}
}
if (!progressed)
break; /* only partitioned traffic left */
}
int chk = sim_scen_check(c);
if (chk < 0)
return chk;
}
return 0;
}
/*
* sim_scen_replicate_round: on the current leader, append one client value, send
* AppendEntries to every other node, drain the bag, then recompute commit — checking
* invariants after every action. Non-leaders and partitioned peers are handled by the
* library / bag delivery (a partitioned peer's AE simply never delivers). Returns 0 on
* a clean round, < 0 on the first invariant violation.
*/
static int sim_scen_replicate_round(raft_sim_cluster_t *c, int leader, uint64_t value)
{
uint64_t out_index = 0;
(void)raft_log_append(&c->nodes[leader].state, &value, sizeof(value), &out_index);
int chk = sim_scen_check(c);
if (chk < 0)
return chk;
for (int p = 0; p < c->n; p++) {
if (p == leader)
continue;
(void)raft_send_append_entries(&c->nodes[leader].state, p,
c->nodes[leader].state.commit_index);
chk = sim_scen_check(c);
if (chk < 0)
return chk;
}
chk = sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 4);
if (chk < 0)
return chk;
(void)raft_advance_commit(&c->nodes[leader].state);
chk = sim_scen_check(c);
if (chk < 0)
return chk;
/* Commit advance can release a higher leader_commit; replicate it out so followers
* apply the newly-committed entries, then re-check. */
for (int p = 0; p < c->n; p++) {
if (p == leader)
continue;
(void)raft_send_append_entries(&c->nodes[leader].state, p,
c->nodes[leader].state.commit_index);
}
return sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 4);
}
int raft_sim_scenario_partition_heal(raft_sim_cluster_t *c, int n, uint64_t seed)
{
/* A minority partition with a surviving majority needs n >= 3 (floor((n-1)/2) >= 1
* isolated, with a strict majority still connected). */
if (n < 3 || n > RAFT_SIM_MAX_NODES)
return -1;
if (raft_sim_conformance_init(c, n, 2000, seed) != 0)
return -1;
int chk;
/* ---- 1. Elect a leader: node 0 campaigns; drain the vote traffic. ---- */
(void)raft_start_election(&c->nodes[0].state);
chk = sim_scen_check(c);
if (chk < 0)
return chk;
chk = sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 4);
if (chk < 0)
return chk;
int leader = sim_find_leader(c);
if (leader < 0) {
/* No leader emerged — retry a couple of campaigns (still all via the library).
* If the cluster genuinely cannot elect, that is observable, not a violation;
* the post-scenario single-leader assertion in the test will catch it. */
for (int retry = 0; retry < 3 && leader < 0; retry++) {
(void)raft_start_election(&c->nodes[retry % n].state);
chk = sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 4);
if (chk < 0)
return chk;
leader = sim_find_leader(c);
}
if (leader < 0)
return 0; /* no violation occurred; nothing more to drive */
}
/* ---- 2. Replicate + commit a few entries with the cluster fully connected. ---- */
for (int r = 0; r < 3; r++) {
chk = sim_scen_replicate_round(c, leader, 0x100ULL + (uint64_t)r);
if (chk < 0)
return chk;
leader = sim_find_leader(c);
if (leader < 0)
return 0;
}
/* ---- 3. Isolate a MINORITY (highest indices); majority keeps progressing. ---- */
int minority = (n - 1) / 2; /* < n/2, so the remaining nodes are a strict majority */
for (int m = 0; m < minority; m++)
raft_sim_partition(c, n - 1 - m, 1);
/* If the isolated set happened to include the leader, re-elect within the majority
* so the partitioned phase still makes progress on the connected side. */
{
int leader_isolated = c->nodes[leader].partition;
if (leader_isolated) {
leader = -1;
for (int cand = 0; cand < n - minority && leader < 0; cand++) {
if (c->nodes[cand].partition)
continue;
(void)raft_start_election(&c->nodes[cand].state);
chk = sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 4);
if (chk < 0)
return chk;
leader = sim_find_leader(c);
/* Only accept a CONNECTED leader (a stale isolated leader does not count). */
if (leader >= 0 && c->nodes[leader].partition)
leader = -1;
}
if (leader < 0)
return 0;
}
}
/* Drive a phase of client appends/replication while partitioned. The majority must
* keep committing; AE to isolated peers simply never delivers. A minority node also
* times out and campaigns — it must NOT be able to win a higher-term election (the
* library enforces quorum; the harness just drives + checks). */
for (int r = 0; r < 4; r++) {
chk = sim_scen_replicate_round(c, leader, 0x200ULL + (uint64_t)r);
if (chk < 0)
return chk;
/* A minority node campaigns mid-partition (cannot reach a majority of votes). */
if (minority > 0) {
(void)raft_start_election(&c->nodes[n - 1].state);
chk = sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 4);
if (chk < 0)
return chk;
}
leader = sim_find_leader(c);
if (leader < 0 || c->nodes[leader].partition)
return 0;
}
/* ---- 4. HEAL the partition and drain everything so the minority catches up. ---- */
for (int m = 0; m < minority; m++)
raft_sim_partition(c, n - 1 - m, 0);
chk = sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 8);
if (chk < 0)
return chk;
/* Draining the parked minority traffic (those nodes campaigned at higher terms while
* isolated) can step the connected leader down via UpdateTerm — correct library
* behavior. Re-establish a single leader over the now-fully-connected cluster before
* catch-up: campaign one node at a time, draining each round, until exactly one
* leader holds. This is the natural Raft recovery, driven only via library entry
* points. */
leader = sim_find_leader(c);
for (int attempt = 0; attempt < 2 * n && leader < 0; attempt++) {
(void)raft_start_election(&c->nodes[attempt % n].state);
chk = sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 8);
if (chk < 0)
return chk;
leader = sim_find_leader(c);
}
if (leader < 0)
return 0; /* no leader emerged — observable, not a violation; test asserts it */
/* Catch-up rounds: re-replicate to ALL nodes (now reachable) until the formerly
* isolated followers converge to the leader's committed prefix. Re-establish the
* leader inside the loop if a late vote-request steps it down again. Bounded. */
for (int r = 0; r < n + 4; r++) {
chk = sim_scen_replicate_round(c, leader, 0x300ULL + (uint64_t)r);
if (chk < 0)
return chk;
leader = sim_find_leader(c);
for (int attempt = 0; attempt < 2 * n && leader < 0; attempt++) {
(void)raft_start_election(&c->nodes[attempt % n].state);
chk = sim_scen_drain_bag(c, RAFT_SIM_MAX_MSGS * 8);
if (chk < 0)
return chk;
leader = sim_find_leader(c);
}
if (leader < 0)
return 0;
}
/* ---- 5. Final convergence check (also covered after every step above). ---- */
return sim_scen_check(c);
}
#endif /* XCE_TESTING */
View git blame Copy permalink