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

feat(raft-commit-safety): commit_index state + on_commit_advanced callback + init (tasks 1.1, 1.2, 2.1)

Browse source 
1.1: raft_state_t.commit_index (monotonic, ==WAL LSN, <=log_len, advanced only via
raft_advance_commit, untouched by AE reply) + raft_config_t.initial_commit_index.
1.2: on_commit_advanced(ctx, commit_index) REQUIRED callback — fired only on strict
advance, a TRIGGER for the controller commit barrier (never does durability itself).
2.1: raft_init requires it (-EINVAL if NULL) + restores commit_index from
initial_commit_index. raft_advance_commit prototype declared (impl in 3.2).

Coupled callback wiring to keep the required-callback ripple green: sim wires
sim_on_commit_advanced (+per-node commit_index mirror & fire-count), all test cbs
builders set it, raft_sim_restart_node restores the durable commit position. Vote
signatures unchanged; no commit-advance logic yet. ASAN green (224 tests).
This commit is contained in:
Rongsong Shen 2026-06-21 23:11:17 +00:00
parent 3d6dc9cc56
commit 3ba6f03a9e
5 changed files with 170 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/lib/raft/raft.c
View file

@ -52,6 +52,11 @@ int raft_init(raft_state_t *s, const raft_config_t *cfg)
return -EINVAL;
if (!cfg->cbs->on_truncate_from)
return -EINVAL;
/* Commit-advance trigger (raft-commit-safety Req 2.5): required — the library
* decides whether commit_index advanced; this callback triggers the
* controller's existing commit barrier. */
if (!cfg->cbs->on_commit_advanced)
return -EINVAL;
s->term = cfg->initial_term;
s->role = RAFT_FOLLOWER;
@ -73,6 +78,10 @@ int raft_init(raft_state_t *s, const raft_config_t *cfg)
* first boot. (raft-log-replication Req 1.1; design "log_len restored at
* init from initial_log_len".) */
s->log_len = cfg->initial_log_len;
/* Commit index restored from the adapter's recovered committed LSN; the
* library never reads storage (same discipline as initial_term/log_len). 0 on
* first boot. (raft-commit-safety Req 1.1.) */
s->commit_index = cfg->initial_commit_index;
s->cbs = cfg->cbs;
s->ctx = cfg->ctx;

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

@ -53,6 +53,13 @@ typedef struct {
uint64_t log_len; /* highest local log index (0 = empty) */
uint64_t next_index[RAFT_MAX_PEERS]; /* leader: next index to send each peer */
uint64_t match_index[RAFT_MAX_PEERS]; /* leader: highest index known replicated to each peer */
/* ---- Commit index (raft-commit-safety, raft.tla commitIndex) ----
* Highest 1-based index (== WAL LSN) the library considers safely committed.
* Monotonic non-decreasing. Restored at init from initial_commit_index (the
* library never reads storage — same discipline as initial_term/log_len).
* Advanced ONLY through raft_advance_commit (the current-term Figure-8 rule);
* raft_on_append_entries_reply leaves it untouched. commit_index <= log_len. */
uint64_t commit_index;
/* set by raft_init — needed by the state machine to fire callbacks */
const raft_callbacks_t *cbs;
void *ctx;
@ -79,7 +86,8 @@ typedef struct {
* on_send_vote_request, on_become_leader, on_become_follower,
* on_persist, on_schedule_timeout, on_cancel_timeout,
* on_start_election_after_delay,
* on_persist_entry, on_get_entry_term, on_send_append_entries, on_truncate_from
* on_persist_entry, on_get_entry_term, on_send_append_entries, on_truncate_from,
* on_commit_advanced
*
* OPTIONAL callbacks (may be NULL):
* on_vote_eligible — if NULL, all Raft-eligible candidates are granted
@ -147,6 +155,19 @@ typedef struct raft_callbacks_t {
/* REQUIRED: delete local entries at `index`..end. */
int (*on_truncate_from)(void *ctx, uint64_t index);
/* ---- Commit-advance trigger (raft-commit-safety) ----
* REQUIRED: fired AFTER commit_index strictly advances. The controller adapter
* feeds this into the EXISTING synchronous commit barrier / persist-before-ACK
* durability — this callback TRIGGERS commit-eval; it does NOT replace the
* barrier.
*
* Contract: the library decides WHETHER the commit index advanced; the callback
* only triggers the adapter's commit work. Fired exactly once per strict advance,
* NEVER on a held commit index (current-term Figure-8 rule), and NEVER from
* raft_on_append_entries_reply. NULL-checked / required in raft_init (-EINVAL if
* NULL). It NEVER performs durability itself. */
void (*on_commit_advanced)(void *ctx, uint64_t commit_index);
/* OPTIONAL: Trace callback for observability.
* event: short string literal, e.g. "vote_granted", "became_leader"
* detail: formatted human-readable context string
@ -175,6 +196,7 @@ typedef struct {
uint64_t initial_term; /* term restored from disk; 0 on first boot */
int initial_voted_for; /* voted_for restored from disk; -1 if none */
uint64_t initial_log_len; /* WAL length restored from disk; 0 on first boot (library never reads storage) */
uint64_t initial_commit_index; /* committed LSN restored from disk; 0 on first boot (library never reads storage) */
const raft_callbacks_t *cbs; /* all callbacks; required callbacks must be non-NULL */
void *ctx; /* opaque caller context passed to every callback */
} raft_config_t;
@ -359,6 +381,28 @@ int raft_on_append_entries_request(raft_state_t *s, int leader_idx,
int raft_on_append_entries_reply(raft_state_t *s, int peer_idx, int success,
uint64_t match_index, uint64_t reply_term);
/*
* raft_advance_commit: leader recomputes its commit index (raft.tla
* AdvanceCommitIndex, 259). A SEPARATE action from raft_on_append_entries_reply
* (which only updates match_index[]) — mirroring raft.tla, where
* AdvanceCommitIndex is distinct from HandleAppendEntriesResponse; this lets a
* single-node leader commit ({self} in agreeIndexes).
*
* agreeIndexes = { index in 1..log_len : {self} U {peer : match_index[peer] >=
* index} is a quorum }. When agreeIndexes is non-empty AND
* on_get_entry_term(Max(agreeIndexes)) == s->term, sets commit_index =
* Max(agreeIndexes) (the Figure-8 current-term rule, line 270) and, on a STRICT
* advance, fires on_commit_advanced(ctx, commit_index) exactly once. Otherwise
* leaves commit_index unchanged and does NOT fire on_commit_advanced.
*
* Monotonic: commit_index never decreases. Reads entry terms only via
* on_get_entry_term; performs NO storage I/O. on_commit_advanced only TRIGGERS
* the controller's existing synchronous commit barrier — the library performs
* no durability work here. Returns RAFT_OK; -EPERM if role != RAFT_LEADER
* (commit_index left unchanged, no callback).
*/
int raft_advance_commit(raft_state_t *s);
/* Fills snap with current state. Caller must hold lock. */
void raft_snapshot(const raft_state_t *s, raft_snapshot_t *snap);
@ -427,6 +471,14 @@ typedef struct {
* log[i-1] holds entry i; log_count is the highest occupied index). */
raft_sim_entry_t log[RAFT_SIM_MAX_LOG];
int log_count;
/* ---- commit-advance callback mirror (raft-commit-safety) ----
* Updated by the node's on_commit_advanced callback: commit_index records the
* last value the library reported (the strict-advance commit position), and
* commit_advanced_count counts how many times on_commit_advanced fired (so
* tests can assert fire-once / never-on-hold trigger semantics). The
* commit-advance DRIVER helper is a later task; this is only the callback half. */
uint64_t commit_index;
int commit_advanced_count;
/* per-node trace event log for test assertions */
char trace_buf[4096];
int trace_len;

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

@ -191,6 +191,20 @@ static int sim_send_append_entries(void *ctx, int peer_idx,
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.
*
@ -252,6 +266,7 @@ int raft_sim_init(raft_sim_cluster_t *c, int n_nodes, uint32_t election_timeout_
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;
@ -265,6 +280,7 @@ int raft_sim_init(raft_sim_cluster_t *c, int n_nodes, uint32_t election_timeout_
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* */
@ -424,9 +440,13 @@ int raft_sim_restart_node(raft_sim_cluster_t *c, int node_idx)
* (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;
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));
@ -437,6 +457,7 @@ int raft_sim_restart_node(raft_sim_cluster_t *c, int node_idx)
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 */

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

@ -77,6 +77,13 @@ static int stub_truncate_from(void *ctx, uint64_t index)
return 0;
}
/* ---- commit-advance callback stub (raft-commit-safety) ---- */
static void stub_commit_advanced(void *ctx, uint64_t commit_index)
{
(void)ctx; (void)commit_index;
}
/* Build a fully-populated callbacks struct */
static raft_callbacks_t make_full_cbs(void)
{
@ -93,6 +100,7 @@ static raft_callbacks_t make_full_cbs(void)
cbs.on_get_entry_term = stub_get_entry_term;
cbs.on_send_append_entries = stub_send_append_entries;
cbs.on_truncate_from = stub_truncate_from;
cbs.on_commit_advanced = stub_commit_advanced;
/* on_vote_eligible and on_trace left NULL (optional) */
return cbs;
}
@ -263,6 +271,7 @@ static raft_callbacks_t make_mock_cbs(void)
cbs.on_get_entry_term = stub_get_entry_term;
cbs.on_send_append_entries = stub_send_append_entries;
cbs.on_truncate_from = stub_truncate_from;
cbs.on_commit_advanced = stub_commit_advanced;
/* on_vote_eligible and on_trace left NULL — tests that need them set explicitly */
return cbs;
}
@ -482,6 +491,74 @@ static void test_raft_init_log_callbacks_present_succeeds(void)
TEST_ASSERT_EQUAL_UINT64(5, s.log_len);
}
/* ---- commit-advance required callback + commit_index restore (raft-commit-safety
* task 2.1, Req 2.5, 1.1) ---- */
/* on_commit_advanced NULL → -EINVAL (Req 2.5), like the other required callbacks */
static void test_raft_init_null_commit_advanced_returns_einval(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0, sizeof(s));
cbs.on_commit_advanced = NULL; /* knock out the new required callback */
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3; cfg.self_idx = 0; cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(-EINVAL, raft_init(&s, &cfg));
}
/* on_commit_advanced present → succeeds (Req 2.5) */
static void test_raft_init_commit_advanced_present_succeeds(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0, sizeof(s));
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3; cfg.self_idx = 0; cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(0, raft_init(&s, &cfg));
}
/* commit_index restored from initial_commit_index at init (Req 1.1) */
static void test_raft_init_restores_commit_index(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0xAB, sizeof(s)); /* poison: prove init writes the field */
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3;
cfg.self_idx = 0;
cfg.initial_commit_index = 7; /* committed LSN restored from disk */
cfg.cbs = &cbs;
TEST_ASSERT_EQUAL_INT(0, raft_init(&s, &cfg));
TEST_ASSERT_EQUAL_UINT64(7, s.commit_index);
}
/* default commit_index is 0 on first boot (Req 1.1, default 0) */
static void test_raft_init_commit_index_default_zero(void)
{
raft_state_t s;
raft_callbacks_t cbs = make_full_cbs();
raft_config_t cfg;
memset(&s, 0xAB, sizeof(s)); /* poison */
memset(&cfg, 0, sizeof(cfg));
cfg.n_peers = 3;
cfg.self_idx = 0;
cfg.cbs = &cbs; /* initial_commit_index left 0 */
TEST_ASSERT_EQUAL_INT(0, raft_init(&s, &cfg));
TEST_ASSERT_EQUAL_UINT64(0, s.commit_index);
}
/* n_peers even (>1) → -EINVAL: even-sized clusters have no fault-tolerance
* advantage over the next-smaller odd cluster; reject to prevent misconfiguration. */
static void test_raft_init_even_peers_returns_einval(void)
@ -5808,6 +5885,10 @@ void test_raft(void)
RUN_TEST(test_raft_init_null_send_append_entries_returns_einval);
RUN_TEST(test_raft_init_null_truncate_from_returns_einval);
RUN_TEST(test_raft_init_log_callbacks_present_succeeds);
RUN_TEST(test_raft_init_null_commit_advanced_returns_einval);
RUN_TEST(test_raft_init_commit_advanced_present_succeeds);
RUN_TEST(test_raft_init_restores_commit_index);
RUN_TEST(test_raft_init_commit_index_default_zero);
RUN_TEST(test_raft_init_even_peers_returns_einval);
RUN_TEST(test_raft_init_oversize_peers_returns_einval);
RUN_TEST(test_raft_init_zero_peers_returns_einval);