From 3ba6f03a9e3fe7012a2748b34780b85ecba1dec1 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 21 Jun 2026 23:11:17 +0000 Subject: [PATCH] feat(raft-commit-safety): commit_index state + on_commit_advanced callback + init (tasks 1.1, 1.2, 2.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- .kiro/specs/raft-commit-safety/tasks.md | 6 +- src/lib/raft/raft.c | 9 +++ src/lib/raft/raft.h | 54 ++++++++++++++++- src/lib/raft/raft_sim.c | 27 ++++++++- src/lib/testsuite/test_raft.c | 81 +++++++++++++++++++++++++ 5 files changed, 170 insertions(+), 7 deletions(-) diff --git a/.kiro/specs/raft-commit-safety/tasks.md b/.kiro/specs/raft-commit-safety/tasks.md index 7b1d1a4b..66429df5 100644 --- a/.kiro/specs/raft-commit-safety/tasks.md +++ b/.kiro/specs/raft-commit-safety/tasks.md @@ -8,14 +8,14 @@ ## 1. Header contract: commit_index, on_commit_advanced, raft_advance_commit, freshness signatures -- [ ] 1.1 Extend `raft.h` with the commit-index state and restore field +- [x] 1.1 Extend `raft.h` with the commit-index state and restore field - Add `uint64_t commit_index;` to `raft_state_t` - Add `uint64_t initial_commit_index;` to `raft_config_t` (restored committed LSN at boot, default 0; library never reads storage — consistent with `initial_term`/`initial_log_len`) - Observable completion: `raft.h` compiles; `sizeof(raft_state_t)` grows by the new field; no entry-byte storage added - _Requirements: 1.1, 1.5_ - _Boundary: raft.h_ -- [ ] 1.2 Add the `on_commit_advanced` REQUIRED callback to `raft_callbacks_t` (verbatim contract) +- [x] 1.2 Add the `on_commit_advanced` REQUIRED callback to `raft_callbacks_t` (verbatim contract) - Declare `void on_commit_advanced(void *ctx, uint64_t commit_index)` with the exact signature from design.md / brief - Document it as REQUIRED, fired only on strict advance, and as a TRIGGER for the existing commit barrier (never performs durability itself) - Observable completion: the function-pointer member exists in `raft_callbacks_t` with the verbatim signature; header doc comment matches the Authoritative shared contract @@ -34,7 +34,7 @@ ## 2. Init validation and commit-index restore -- [ ] 2.1 RED+GREEN: `raft_init` requires `on_commit_advanced` and restores `initial_commit_index` +- [x] 2.1 RED+GREEN: `raft_init` requires `on_commit_advanced` and restores `initial_commit_index` - Test: `raft_init` returns `-EINVAL` when `on_commit_advanced` is NULL; returns 0 when present; `s->commit_index == cfg->initial_commit_index` after init - Implement the NULL check in `raft_init` (mirroring the existing required-callback checks); restore `s->commit_index = cfg->initial_commit_index` - Observable completion: new `-EINVAL` test case passes; restore assertion passes; existing election + log-replication init tests still green diff --git a/src/lib/raft/raft.c b/src/lib/raft/raft.c index c45ed4c5..a58b414e 100644 --- a/src/lib/raft/raft.c +++ b/src/lib/raft/raft.c @@ -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; diff --git a/src/lib/raft/raft.h b/src/lib/raft/raft.h index 8f7033e2..9908cfb4 100644 --- a/src/lib/raft/raft.h +++ b/src/lib/raft/raft.h @@ -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; diff --git a/src/lib/raft/raft_sim.c b/src/lib/raft/raft_sim.c index 1d8ef390..b050ec60 100644 --- a/src/lib/raft/raft_sim.c +++ b/src/lib/raft/raft_sim.c @@ -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 */ diff --git a/src/lib/testsuite/test_raft.c b/src/lib/testsuite/test_raft.c index 08a4f78f..efe52436 100644 --- a/src/lib/testsuite/test_raft.c +++ b/src/lib/testsuite/test_raft.c @@ -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);