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

feat(ha-term-safety): wire five callbacks + library init restore (task 2.5)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-22 07:47:30 +00:00
parent 3470bdfb71
commit 3dd8493fff
3 changed files with 139 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

6
.kiro/specs/ha-term-safety/tasks.md
View file

@ -57,7 +57,7 @@
- _Requirements: 2.2, 4.1, 4.5, 5.4_
- _Depends: 1.3_
- [ ] 2.5 Wire the five callbacks into the controller callback set and the library init restore
- [x] 2.5 Wire the five callbacks into the controller callback set and the library init restore
- Add the five callbacks to the controller's `s_real_cbs`; on `raft_init` restore `initial_log_len`, `initial_commit_index`, and the recovered last-log term/index from the WAL so the library boots with the durable floor
- Observable completion: a controller node initializes the raft library with all required callbacks non-NULL and restored indices matching its recovered WAL; existing election callbacks remain wired
- _Requirements: 1.5, 2.1_
@ -164,3 +164,7 @@
- **FEATURE GATE (critical for 3.x):** ceiling is INERT while `raft_commit_index == 0` (no on_commit_advanced fired). Lowest real LSN is 1 (commit_submit rejects lsn==0), so `0` unambiguously means "nothing committed" — no 0-vs-unset window. Tasks 3.1–3.3 wiring `raft_advance_commit` on the ACK path will fire the FIRST `on_commit_advanced(commit_index>=1)` and engage the library source; until then the autonomous barrier is preserved (no steady-state regression, R5.4).
- **DEFERRED to 3.x (decommission the term-blind source):** `ha_compute_safe_committed_lsn` is no longer the commit source but is NOT yet removed — it still drives the promotion-time safe truncate (R5). 3.x must ensure no path still feeds it INTO the barrier as a commit source once `raft_advance_commit` is wired.
- For-test seam: `ha_cb_commit_advanced_for_test` (ha.h, XCE_TESTING) + `commit_set_raft_commit_index` are the hooks 5.1 integration tests should reuse.
- **2.5 done.** Init restore on the REAL post-recovery boot path `ha_init_raft` (ha.c ~4472, runs after `persist_recover` per myapp.c:237/242) — NOT the pre-recovery `ha_init` placeholder (~1756, raft_c==NULL, left at zero by design). `rcfg.initial_log_len = persist_durable_floor(&c->persist)`; `rcfg.initial_commit_index = committed_lsn` clamped to log_len (preserves library `commit_index<=log_len`, raft.h:61). There is NO `initial_last_log_term` field — the library sources last-log term via the wired `on_get_entry_term(s->log_len)` (raft.c:161). All five callbacks + election callbacks (on_become_leader/on_persist) verified non-NULL + real bodies in s_real_cbs. 2 tests; reviewer APPROVED (independently reverted restore→`Expected 3 Was 0` RED; investigated commit_index concern vs literal §1.5).
- **commit_index-at-recover = 0 is CORRECT-BY-DESIGN, not a gap:** `persist_recover` does NOT re-derive `committed_lsn` from disk (it stays 0 after persist_init); the committed position is re-established post-election from voter replies / via `on_commit_advanced` (tasks 3.x). §1.5/design Traceability row 1.5 mandate restoring `initial_log_len` + last-log-term ONLY. The clamp is just defensive wiring for any future path that sets committed_lsn pre-init. The 2.5 test sets committed_lsn=2 manually to exercise the clamp/pass-through (honestly commented); the load-bearing real-recover assertions are log_len==3 + last-log term==7.
- **PRE-EXISTING STALE COMMENT (cleanup, not 2.5 scope):** ha.c ~1326-1334 still calls the five callbacks "stubs"/"placeholders" that "perform NO persistence" — stale since 2.1–2.4 gave them real bodies. Also the callbacks are still NAMED `ha_cb_*_stub` (real bodies). A rename + comment refresh is a good 5.5 / cleanup-pass item; cosmetic only.

34
src/controller/ha.c
View file

@ -4450,6 +4450,32 @@ void ha_init_raft(ha_state_t *ha, cluster_t *c, int n_peers)
}
}
/* ha-term-safety task 2.5 (R1.5, R2.1): restore the durable log/commit floor
* from the recovered WAL so the library boots at the recovered tip — not an
* empty log at LSN 0. The library never reads storage; the caller restores
* initial_log_len / initial_commit_index here (same discipline as
* initial_term / initial_voted_for above).
*
* initial_log_len = recovered durable last-log index
* (persist_durable_floor: re-derived from the WAL tip
* by persist_recover; 0 on empty first boot).
* initial_commit_index = recovered committed LSN (committed_lsn). Recovery
* re-derives the durable tip but the post-election
* safe-commit point governs the committed position;
* we pass whatever committed prefix is established at
* init (0 on a fresh boot). Clamp to log_len to honor
* the library invariant commit_index <= log_len.
*
* The recovered last-log TERM is surfaced on demand via on_get_entry_term(
* log_len) (the library's logOk freshness gate reads it) — there is no
* separate initial last-log-term field. */
uint64_t initial_log_len = c ? persist_durable_floor(&c->persist) : 0;
uint64_t initial_commit_index = c
? atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire)
: 0;
if (initial_commit_index > initial_log_len)
initial_commit_index = initial_log_len;
raft_config_t rcfg;
memset(&rcfg, 0, sizeof(rcfg));
rcfg.n_peers = n_peers;
@ -4458,15 +4484,19 @@ void ha_init_raft(ha_state_t *ha, cluster_t *c, int n_peers)
rcfg.my_priority = ha->my_priority;
rcfg.initial_term = atomic_load_explicit(&ha->term, memory_order_acquire);
rcfg.initial_voted_for = restored_voted_for;
rcfg.initial_log_len = initial_log_len;
rcfg.initial_commit_index = initial_commit_index;
rcfg.cbs = &s_real_cbs;
rcfg.ctx = ha;
raft_init(&ha->raft, &rcfg);
ha->n_peers = n_peers;
ha->raft_c = c;
INFO("HA: raft initialized (n_peers=%d, self=%d, term=%" PRIu64 ")",
INFO("HA: raft initialized (n_peers=%d, self=%d, term=%" PRIu64
", log_len=%" PRIu64 ", commit_index=%" PRIu64 ")",
n_peers, ha->my_priority,
atomic_load_explicit(&ha->term, memory_order_acquire));
atomic_load_explicit(&ha->term, memory_order_acquire),
initial_log_len, initial_commit_index);
}
/* ── ha-prevote task 4.1: candidate-side PreVote orchestration ─────────────

102
src/controller/testsuite/test_term_safety_adapter.c
View file

@ -861,6 +861,106 @@ static void test_on_commit_advanced_wired_feeds_barrier(void)
printf("test_on_commit_advanced_wired_feeds_barrier: PASS\n");
}
/* ================================================================== */
/* Task 2.5 — library init restore from the recovered WAL floor */
/* ================================================================== */
/* Test: ha_init_raft restores the library log/commit floor from the
* recovered WAL instead of booting at LSN 0 (R1.5, R2.1).
*
* After a member recovers a WAL whose durable tip is LSN 3 (term 7) with a
* committed prefix at LSN 2, raft_init must boot with:
* raft.log_len == 3 (initial_log_len = persist_durable_floor)
* raft.commit_index == 2 (initial_commit_index = recovered committed_lsn)
* and the last-log term must be recoverable via the wired on_get_entry_term
* (== 7). Booting from an empty log (log_len/commit_index == 0) is the pre-fix
* behaviour this test fails on. */
static void test_init_restores_recovered_floor(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_ts_initrestore_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const uint64_t T = 7;
/* Phase 1: lay down a 3-entry WAL (term 7) and flush so the recovered tip
* reaches LSN 3. */
{
cluster_t* c = make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
atomic_store(&c->ha.role, CTRL_PRIMARY);
uint8_t blob[64];
for (uint64_t i = 1; i <= 3; i++) {
size_t n = pack_entry(blob, WAL_JOB_UPDATE, "x", 1);
TEST_ASSERT_EQUAL_INT(0,
ha_cb_persist_entry_for_test(&c->ha, i, T, blob, n));
}
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
teardown_cluster(c);
}
/* Phase 2: recover into a fresh cluster; durable floor == 3. Simulate a
* recovered committed prefix at LSN 2 (recovery re-derives the durable tip;
* the committed position is set by the surrounding boot path). */
{
cluster_t* c2 = make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c2);
TEST_ASSERT_EQUAL_INT(0, persist_recover(&c2->persist));
TEST_ASSERT_EQUAL_UINT64(3, persist_durable_floor(&c2->persist));
atomic_store(&c2->persist.committed_lsn, 2ULL);
/* Initialize the embedded raft as a single-node cluster. The restore
* must source the floor from the recovered persist, not LSN 0. */
ha_init_raft(&c2->ha, c2, /*n_peers*/1);
TEST_ASSERT_EQUAL_UINT64(3, c2->ha.raft.log_len);
TEST_ASSERT_EQUAL_UINT64(2, c2->ha.raft.commit_index);
/* The recovered last-log term is surfaced via the wired callback. */
TEST_ASSERT_EQUAL_UINT64(T,
ha_cb_get_entry_term_for_test(&c2->ha, c2->ha.raft.log_len));
/* All five term-safety callbacks remain wired (non-NULL). */
TEST_ASSERT_NOT_NULL(c2->ha.raft.cbs);
TEST_ASSERT_NOT_NULL(c2->ha.raft.cbs->on_persist_entry);
TEST_ASSERT_NOT_NULL(c2->ha.raft.cbs->on_get_entry_term);
TEST_ASSERT_NOT_NULL(c2->ha.raft.cbs->on_send_append_entries);
TEST_ASSERT_NOT_NULL(c2->ha.raft.cbs->on_truncate_from);
TEST_ASSERT_NOT_NULL(c2->ha.raft.cbs->on_commit_advanced);
/* Existing election callbacks remain wired. */
TEST_ASSERT_NOT_NULL(c2->ha.raft.cbs->on_become_leader);
TEST_ASSERT_NOT_NULL(c2->ha.raft.cbs->on_persist);
teardown_cluster(c2);
}
printf("test_init_restores_recovered_floor: PASS\n");
}
/* Test: an empty-WAL first boot restores a zero floor (no spurious non-zero
* indices) — the recovery-read-failure / empty-log edge of R1.5. */
static void test_init_empty_wal_zero_floor(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_ts_initempty_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t* c = make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
TEST_ASSERT_EQUAL_INT(0, persist_recover(&c->persist));
TEST_ASSERT_EQUAL_UINT64(0, persist_durable_floor(&c->persist));
ha_init_raft(&c->ha, c, /*n_peers*/1);
TEST_ASSERT_EQUAL_UINT64(0, c->ha.raft.log_len);
TEST_ASSERT_EQUAL_UINT64(0, c->ha.raft.commit_index);
teardown_cluster(c);
printf("test_init_empty_wal_zero_floor: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Suite entry point */
/* ------------------------------------------------------------------ */
@ -885,4 +985,6 @@ void test_term_safety_adapter(void)
RUN_TEST(test_commit_advanced_does_not_bypass_durable_cap);
RUN_TEST(test_own_term_gate_retained_no_deadlock);
RUN_TEST(test_on_commit_advanced_wired_feeds_barrier);
RUN_TEST(test_init_restores_recovered_floor);
RUN_TEST(test_init_empty_wal_zero_floor);
}