diff --git a/.kiro/specs/ha-term-safety/tasks.md b/.kiro/specs/ha-term-safety/tasks.md index 922dc628..e187c44c 100644 --- a/.kiro/specs/ha-term-safety/tasks.md +++ b/.kiro/specs/ha-term-safety/tasks.md @@ -38,7 +38,7 @@ - _Requirements: 1.1, 1.2, 1.3, 1.4_ - _Depends: 1.3_ -- [ ] 2.2 Implement `on_get_entry_term` and `on_truncate_from` +- [x] 2.2 Implement `on_get_entry_term` and `on_truncate_from` - `on_get_entry_term` answers from the recovered/live per-LSN term index (legacy-0 rule from 1.3); `on_truncate_from` deletes entries at index..end via the existing LSN truncate, replacing the committed-boundary heuristic - Observable completion: the library reads back a stamped entry term via the callback; a truncate-from call removes the local suffix at and above the index and leaves the prefix intact - _Requirements: 1.5, 1.8, 3.1, 3.2_ @@ -153,3 +153,5 @@ - **DEFERRED to task 3.1 (review):** primary path uses persist_wal_async's `current_lsn`, NOT the library `index` — holds only while `s->log_len` and `p->current_lsn` stay co-tracked. Task 3.1 MUST retire/disable the hand-rolled primary append so no WAL append bypasses raft_log_append, else current_lsn diverges from s->log_len. Also: persist_set_term + write are two lock regions (safe today, no concurrent appender) — 3.1 should re-verify when real concurrent appends exist. - **DEFERRED to task 3.2 (review):** a heartbeat / 0-entry AppendEntries must NOT call on_persist_entry (entry==NULL,len==0 currently falls through to WAL_NOOP) — verify the library only calls on_persist_entry when there is an entry. - **3.1/3.2 producer format:** must pack `raft_entry_blob_t {type,payload_len}+payload` to match the on_persist_entry unpack (single shared def in wal_types.h). + +- **2.2 done.** on_get_entry_term → persist_entry_term (legacy-0 caveat documented per 1.3 review); on_truncate_from → persist_truncate_wal_above(p,index-1) (delete LSN>=index), returns rc, non-promotable fence preserved. NEW `truncate_term_index_locked` (persist.c) shrinks the in-memory term index on BOTH truncate return paths (under p->lock, cutoff-aligned) so on_get_entry_term returns 0 for truncated LSNs — no stale-term hazard. 3 tests. Reviewer APPROVED. diff --git a/src/controller/ha.c b/src/controller/ha.c index f1efb56f..eceb5616 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1414,10 +1414,27 @@ static int ha_cb_persist_entry_stub(void *ctx, uint64_t index, uint64_t term, return persist_wal_write_replicated(p, index, type, payload, plen); } +/* on_get_entry_term (ha-term-safety task 2.2, R1.5/R3.1): return the recorded + * Raft term for the entry at 1-based LSN `index`, sourced from the WAL term + * index via persist_entry_term. Returns 0 when the LSN is absent (no entry). + * The library calls this for its logOk / (term,lsn) reconcile gate. + * + * Legacy-0 caveat (from the task 1.3 review MEDIUM): persist_entry_term maps a + * present legacy-0 entry (hdr.term == 0, written before this feature) to + * WAL_LEGACY_TERM (== 1). In a mixed pre/post-feature cluster where a + * post-feature leader is genuinely in term 1, a (term,lsn) EQUALITY check + * cannot tell a legacy-0 entry (mapped to 1) apart from a real term-1 entry by + * term alone. This is safe because the Raft log-matching invariant (same term + * + same index ⇒ same entry, enforced across the prefix) plus the bounded + * one-time upgrade window (every entry written after upgrade carries a real + * term) mean the only collision is at the legacy/term-1 boundary, where the + * prefixes already agree. We note it explicitly rather than special-casing it. */ static uint64_t ha_cb_get_entry_term_stub(void *ctx, uint64_t index) { - (void)ctx; (void)index; - return 0; /* TODO task 2.2: 0 == absent (logOk reads local last-log term as 0) */ + ha_state_t *ha = (ha_state_t *)ctx; + if (!ha || !ha->raft_c) + return 0; /* test mode without a cluster: no log → absent */ + return persist_entry_term(&ha->raft_c->persist, index); } static int ha_cb_send_append_entries_stub(void *ctx, int peer_idx, @@ -1430,10 +1447,26 @@ static int ha_cb_send_append_entries_stub(void *ctx, int peer_idx, return 0; /* TODO task 2.3 */ } +/* on_truncate_from (ha-term-safety task 2.2, R1.8/R3.2): delete the local log + * entries at LSN `index..end` because they conflict with the leader's log at a + * (term,lsn) the library detected. persist_truncate_wal_above(p, cutoff) removes + * every entry with LSN > cutoff, so to drop LSN >= index we pass cutoff = + * index - 1. This REPLACES the committed-boundary truncate heuristic on the + * library-driven path: the library, not a watermark, decides the cut point. + * + * Returns the truncate rc. A non-zero rc propagates so the library/leader + * retries; the member stays non-promotable per R3.5 (that fence is existing + * behavior and is not touched here). persist_truncate_wal_above also shrinks + * the per-LSN term index for the removed LSNs, so a later on_get_entry_term on + * a truncated LSN reads back 0 (absent). */ static int ha_cb_truncate_from_stub(void *ctx, uint64_t index) { - (void)ctx; (void)index; - return 0; /* TODO task 2.4 */ + ha_state_t *ha = (ha_state_t *)ctx; + if (!ha || !ha->raft_c) + return -EINVAL; /* test mode without a cluster: cannot truncate */ + if (index == 0) + return -EINVAL; /* LSN 0 is reserved — never an entry to truncate */ + return persist_truncate_wal_above(&ha->raft_c->persist, index - 1); } static void ha_cb_commit_advanced_stub(void *ctx, uint64_t commit_index) @@ -1633,6 +1666,20 @@ int ha_cb_persist_entry_for_test(ha_state_t *ha, uint64_t index, uint64_t term, { return s_real_cbs.on_persist_entry(ha, index, term, entry, len); } + +/* ha-term-safety task 2.2: drive the WIRED on_get_entry_term callback so a unit + * test exercises the real entry-term accessor. Exposed only under XCE_TESTING. */ +uint64_t ha_cb_get_entry_term_for_test(ha_state_t *ha, uint64_t index) +{ + return s_real_cbs.on_get_entry_term(ha, index); +} + +/* ha-term-safety task 2.2: drive the WIRED on_truncate_from callback so a unit + * test exercises the real conflict truncation. Exposed only under XCE_TESTING. */ +int ha_cb_truncate_from_for_test(ha_state_t *ha, uint64_t index) +{ + return s_real_cbs.on_truncate_from(ha, index); +} #endif /* ha_check_heartbeat_nocluster() / ha_watchdog_thread() / ha_start_watchdog() / diff --git a/src/controller/ha.h b/src/controller/ha.h index 1fbc1cd8..7b7d45cd 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -771,6 +771,24 @@ int handle_request_vote(ha_state_t *ha, */ int ha_cb_persist_entry_for_test(ha_state_t *ha, uint64_t index, uint64_t term, const void *entry, size_t len); + +/* + * ha_cb_get_entry_term_for_test (ha-term-safety task 2.2): invoke the WIRED + * on_get_entry_term callback directly so a unit test exercises the real + * entry-term accessor (persist_entry_term at the given 1-based index == LSN). + * Returns the recorded term (legacy-0 mapped to WAL_LEGACY_TERM), or 0 when the + * LSN is absent. TEST-ONLY. + */ +uint64_t ha_cb_get_entry_term_for_test(ha_state_t *ha, uint64_t index); + +/* + * ha_cb_truncate_from_for_test (ha-term-safety task 2.2): invoke the WIRED + * on_truncate_from callback directly so a unit test exercises the real conflict + * truncation (delete local entries at LSN >= index). Returns the truncate rc + * (0 on success; non-zero propagates so the library/leader retries while the + * member stays non-promotable). TEST-ONLY. + */ +int ha_cb_truncate_from_for_test(ha_state_t *ha, uint64_t index); #endif /* diff --git a/src/controller/persist.c b/src/controller/persist.c index 818fd15a..164197bd 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -440,6 +440,21 @@ static void record_entry_term(persist_t* p, uint64_t lsn, uint64_t term) p->term_index_count = lsn; } +/* Shrink the in-memory per-LSN term index so that every LSN > `keep_lsn` reads + * back as absent (ha-term-safety task 2.2). Called from the truncate path after + * the on-disk entries above keep_lsn are removed, keeping the term index + * consistent with the WAL: a later persist_entry_term on a truncated LSN must + * return 0. The backing allocation is left as-is (only the live count moves); + * the zeroed-on-grow contract means a later re-append at those LSNs overwrites + * cleanly. Caller must hold p->lock. */ +static void truncate_term_index_locked(persist_t* p, uint64_t keep_lsn) +{ + if (!p) + return; + if (keep_lsn < p->term_index_count) + p->term_index_count = keep_lsn; +} + void persist_set_term(persist_t* p, uint64_t term) { if (!p) @@ -2659,6 +2674,8 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn) atomic_store_explicit(&p->committed_lsn, cutoff_lsn, memory_order_release); clamp_durable_lsn_locked(p, cutoff_lsn); + /* Keep the per-LSN term index consistent: drop terms for removed LSNs. */ + truncate_term_index_locked(p, cutoff_lsn); pthread_mutex_unlock(&p->lock); return 0; } @@ -2738,6 +2755,8 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn) p->last_crc = new_last_crc; atomic_store_explicit(&p->committed_lsn, cutoff_lsn, memory_order_release); clamp_durable_lsn_locked(p, cutoff_lsn); + /* Keep the per-LSN term index consistent: drop terms for removed LSNs. */ + truncate_term_index_locked(p, cutoff_lsn); pthread_mutex_unlock(&p->lock); return 0; diff --git a/src/controller/testsuite/test_term_safety_adapter.c b/src/controller/testsuite/test_term_safety_adapter.c index 1bd02b7c..2184f7ba 100644 --- a/src/controller/testsuite/test_term_safety_adapter.c +++ b/src/controller/testsuite/test_term_safety_adapter.c @@ -378,6 +378,122 @@ static void test_on_persist_entry_failure_propagates_no_ack(void) printf("test_on_persist_entry_failure_propagates_no_ack: PASS\n"); } +/* ================================================================== */ +/* Task 2.2 — on_get_entry_term + on_truncate_from callback bodies */ +/* ================================================================== */ + +/* Persist a couple of replicated entries (each at a 1-based index == LSN) + * through the WIRED on_persist_entry callback, stamping a leader term per + * entry. Returns the (started) cluster the caller tears down. */ +static cluster_t* persist_two_entries(const char* tmp, uint64_t t1, uint64_t t2) +{ + cluster_t* c = make_cluster(tmp); + if (!c) return NULL; + atomic_store(&c->ha.role, CTRL_STANDBY); + + uint8_t blob[64]; + size_t n1 = pack_entry(blob, WAL_JOB_UPDATE, "a", 1); + if (ha_cb_persist_entry_for_test(&c->ha, /*index*/1, t1, blob, n1) != 0) { + teardown_cluster(c); + return NULL; + } + size_t n2 = pack_entry(blob, WAL_JOB_UPDATE, "b", 1); + if (ha_cb_persist_entry_for_test(&c->ha, /*index*/2, t2, blob, n2) != 0) { + teardown_cluster(c); + return NULL; + } + return c; +} + +/* Test: on_get_entry_term returns the stamped per-LSN term for present entries, + * the lowest-known term for a legacy-0 entry, and 0 for an absent LSN + * (R1.5, R3.1). */ +static void test_on_get_entry_term_present_legacy_absent(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_ts_get_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + cluster_t* c = make_cluster(tmp); + TEST_ASSERT_NOT_NULL(c); + atomic_store(&c->ha.role, CTRL_STANDBY); + + uint8_t blob[64]; + + /* LSN 1: real term 6. */ + size_t n = pack_entry(blob, WAL_JOB_UPDATE, "r", 1); + TEST_ASSERT_EQUAL_INT(0, + ha_cb_persist_entry_for_test(&c->ha, /*index*/1, /*term*/6, blob, n)); + + /* LSN 2: legacy-0 entry (leader term 0) → reads back as WAL_LEGACY_TERM. */ + n = pack_entry(blob, WAL_JOB_UPDATE, "l", 1); + TEST_ASSERT_EQUAL_INT(0, + ha_cb_persist_entry_for_test(&c->ha, /*index*/2, /*term*/0, blob, n)); + + /* Present real-term entry returns its term verbatim. */ + TEST_ASSERT_EQUAL_UINT64(6, ha_cb_get_entry_term_for_test(&c->ha, 1)); + /* Present legacy-0 entry maps to the lowest-known real term. */ + TEST_ASSERT_EQUAL_UINT64(LEGACY_LOWEST_TERM, + ha_cb_get_entry_term_for_test(&c->ha, 2)); + /* Absent LSNs (0 reserved, 3 never appended) return 0. */ + TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 0)); + TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 3)); + + teardown_cluster(c); + printf("test_on_get_entry_term_present_legacy_absent: PASS\n"); +} + +/* Test: on_truncate_from(k) deletes local entries at LSN >= k; the prefix < k + * stays intact and on_get_entry_term returns 0 for the truncated LSNs while the + * term index shrinks consistently (R1.8, R3.1, R3.2). */ +static void test_on_truncate_from_drops_suffix(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_ts_trunc_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + /* Persist LSN 1 (term 3) and LSN 2 (term 5). */ + cluster_t* c = persist_two_entries(tmp, /*t1*/3, /*t2*/5); + TEST_ASSERT_NOT_NULL(c); + + /* Both present before truncation. */ + TEST_ASSERT_EQUAL_UINT64(3, ha_cb_get_entry_term_for_test(&c->ha, 1)); + TEST_ASSERT_EQUAL_UINT64(5, ha_cb_get_entry_term_for_test(&c->ha, 2)); + TEST_ASSERT_EQUAL_UINT64(3, c->persist.current_lsn); + + /* Truncate from LSN 2 → entries at LSN >= 2 are deleted. */ + TEST_ASSERT_EQUAL_INT(0, ha_cb_truncate_from_for_test(&c->ha, 2)); + + /* Prefix < 2 intact, suffix >= 2 gone (term index returns 0). */ + TEST_ASSERT_EQUAL_UINT64(3, ha_cb_get_entry_term_for_test(&c->ha, 1)); + TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 2)); + TEST_ASSERT_EQUAL_UINT64(2, c->persist.current_lsn); + + teardown_cluster(c); + printf("test_on_truncate_from_drops_suffix: PASS\n"); +} + +/* Test: on_truncate_from(1) deletes the whole log (LSN >= 1); every entry's + * term reads back absent and the tip collapses to LSN 1. */ +static void test_on_truncate_from_removes_all(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_ts_trunc_all_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + cluster_t* c = persist_two_entries(tmp, /*t1*/2, /*t2*/2); + TEST_ASSERT_NOT_NULL(c); + + TEST_ASSERT_EQUAL_INT(0, ha_cb_truncate_from_for_test(&c->ha, 1)); + + TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 1)); + TEST_ASSERT_EQUAL_UINT64(0, ha_cb_get_entry_term_for_test(&c->ha, 2)); + TEST_ASSERT_EQUAL_UINT64(1, c->persist.current_lsn); + + teardown_cluster(c); + printf("test_on_truncate_from_removes_all: PASS\n"); +} + /* ------------------------------------------------------------------ */ /* Suite entry point */ /* ------------------------------------------------------------------ */ @@ -392,4 +508,7 @@ void test_term_safety_adapter(void) RUN_TEST(test_on_persist_entry_stamps_term_standby); RUN_TEST(test_on_persist_entry_stamps_term_primary); RUN_TEST(test_on_persist_entry_failure_propagates_no_ack); + RUN_TEST(test_on_get_entry_term_present_legacy_absent); + RUN_TEST(test_on_truncate_from_drops_suffix); + RUN_TEST(test_on_truncate_from_removes_all); }