From 8f3d3ce8f43b5db92099f41f332852babae5be34 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 22 Jun 2026 04:53:18 +0000 Subject: [PATCH] feat(ha-term-safety): on_persist_entry callback body (task 2.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the iter-1 stub: persist_set_term(p, term) stamps the leader/library term into hdr.term, then role-routes the write — primary via persist_wal_async, standby via persist_wal_write_replicated(p, index, ...). Persist failure returns non-zero (no false ACK; durable_lsn unchanged); bug-850 contiguity guard + persist-before-ACK preserved beneath. In-flight raft_entry_blob_t {type,payload_len} envelope added to wal_types.h (never enters the 40-byte header). 3 tests, reviewer APPROVED. Deferred (recorded in tasks.md): task 3.1 must retire the hand-rolled primary append so current_lsn tracks the library index; task 3.2 must not call on_persist_entry on heartbeats. Per-group 0 failures (pre-existing test_commit_barrier_reply leak aside). --- .kiro/specs/ha-term-safety/tasks.md | 7 +- src/controller/ha.c | 79 ++++++++- src/controller/ha.h | 15 ++ .../testsuite/test_term_safety_adapter.c | 166 ++++++++++++++++++ src/include/wal_types.h | 21 +++ 5 files changed, 284 insertions(+), 4 deletions(-) diff --git a/.kiro/specs/ha-term-safety/tasks.md b/.kiro/specs/ha-term-safety/tasks.md index a4db120..922dc62 100644 --- a/.kiro/specs/ha-term-safety/tasks.md +++ b/.kiro/specs/ha-term-safety/tasks.md @@ -32,7 +32,7 @@ ## 2. Core: the five callback adapter bodies -- [ ] 2.1 Implement `on_persist_entry` over the WAL write path +- [x] 2.1 Implement `on_persist_entry` over the WAL write path - Stamp `hdr.term` from the library-supplied term (via `persist_set_term`) at the given 1-based index == LSN, through the existing async/replicated WAL write; preserve persist-before-ACK and the bug-850 contiguity guard beneath it - Observable completion: a library append/accept persists an entry whose recovered `hdr.term` equals the library term; persist failure propagates a non-zero rc and does not ACK - _Requirements: 1.1, 1.2, 1.3, 1.4_ @@ -148,3 +148,8 @@ - **1.3 done.** persist_set_term + per-LSN in-memory term index (built in replay_wal, freed in persist_fini) + persist_entry_term accessor; legacy-0 -> WAL_LEGACY_TERM=1 at read-back only (raw 0 kept so present-legacy-0=1 vs absent=0). 40-byte header preserved. 5 new tests in test_term_safety_adapter.c. - **Task 2.2 MUST document (review MEDIUM):** in a mixed pre/post-feature cluster where the post-feature leader is in term 1, the (term,lsn) check cannot distinguish a legacy-0 entry from a real term-1 entry by term alone; the Raft log-matching invariant + the bounded upgrade window make this safe, but note it in the on_get_entry_term/conflict-truncate comment. - **Validation env (bug-924):** controller suite creates many /tmp/xce_* WAL dirs; overlayfs fdatasync degrades as they accumulate -> non-deterministic meson timeouts. Reliable gate: `rm -rf /tmp/xce_*` then run the `controller_unit_test` binary directly (all groups 0 failures). Pre-existing ASAN leak in test_commit_barrier_reply.c (not gated by meson detect_leaks=0). + +- **2.1 done.** on_persist_entry body: persist_set_term(p,term) then role-routed write (primary persist_wal_async / standby persist_wal_write_replicated(p,index,...)); failure propagates non-zero, durable_lsn unchanged, persist-before-ACK + bug-850 guard preserved. Added in-flight `raft_entry_blob_t {type,payload_len}` envelope in wal_types.h (NEVER in the 40-byte header). 3 tests. + - **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). diff --git a/src/controller/ha.c b/src/controller/ha.c index 53bca78..f1efb56 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1332,7 +1332,7 @@ static void raft_cb_trace(void *ctx, const char *event, const char *detail) * These placeholders exist ONLY to restore the controller build (tasks 1.1/1.2); * they perform NO persistence, NO replication, and NO truncation. See CONCERNS. * - * on_persist_entry → task 2.1 (durable WAL append/overwrite) + * on_persist_entry → task 2.1 (durable WAL append/overwrite) — IMPLEMENTED * on_get_entry_term → task 2.2 (WAL entry-term lookup; persist_entry_term, 1.3) * on_send_append_entries→ task 2.3 (AppendEntries transmit) * on_truncate_from → task 2.4 (conflict truncation) @@ -1343,11 +1343,75 @@ static void raft_cb_trace(void *ctx, const char *event, const char *detail) * grants on lastLogTerm >= 0 — i.e. it does NOT yet tighten the existing * config_tag/last_log_lsn freshness gate already enforced in cluster.c. The real * accessor (task 2.2/1.3) makes the gate effective. */ +/* on_persist_entry (ha-term-safety task 2.1, R1.1–R1.4): durably append/overwrite + * the log entry at 1-based `index` (== WAL LSN) stamping the LEADER-supplied + * `term` into hdr.term, then return the persist rc. + * + * Persist-before-act: this is called BEFORE the library reports the append/accept + * (raft_log_append / raft_on_append_entries_request). On a non-zero (failure) rc + * the library does NOT advance its log and a standby does NOT ACK (the existing + * persist-before-ACK + bug-850 contiguity guard sit beneath the WAL write and are + * preserved — we do not bypass them). We never false-ACK on failure. + * + * `entry` is the adapter envelope (raft_entry_blob_t prefix + payload) packed by + * the append/replication driver (tasks 3.1/3.2); the library holds no entry bytes + * and treats it as opaque. We decode (type, payload, payload_len) from it for the + * existing persist write path. + * + * Path selection by role: + * - PRIMARY own-append: persist_wal_async — assigns the contiguous next LSN + * (== index here, since the library's index == log_len+1 == current_lsn) and + * defers fsync to the group-commit flusher (the steady-state append path). + * - STANDBY replicated apply: persist_wal_write_replicated — writes at the GIVEN + * lsn=index with the contiguity guard + inline persist-before-ACK fsync. + * + * NOTE (deferred to 2.3/3.2): the leader's per-entry term is carried on the repl + * envelope by task 2.3/3.2; until then the standby path stamps the term we set + * here via persist_set_term — which IS the supplied `term` argument — so a test + * that drives this callback already records the leader term. The envelope wiring + * makes the live replication path source the same value. */ static int ha_cb_persist_entry_stub(void *ctx, uint64_t index, uint64_t term, const void *entry, size_t len) { - (void)ctx; (void)index; (void)term; (void)entry; (void)len; - return 0; /* TODO task 2.1 */ + ha_state_t *ha = (ha_state_t *)ctx; + if (!ha || !ha->raft_c) + return -EINVAL; /* test mode without a cluster: cannot persist */ + + persist_t *p = &ha->raft_c->persist; + + /* Decode the adapter entry envelope: a fixed raft_entry_blob_t prefix + * carrying the WAL type + payload_len, followed by the payload bytes. */ + uint32_t type = WAL_NOOP; + const void *payload = NULL; + uint32_t plen = 0; + if (entry && len >= sizeof(raft_entry_blob_t)) { + const raft_entry_blob_t *blob = (const raft_entry_blob_t *)entry; + type = blob->type; + plen = blob->payload_len; + /* Reject a malformed envelope whose declared payload exceeds the buffer + * — fail loud rather than read past the blob. */ + if ((size_t)plen + sizeof(*blob) > len) + return -EINVAL; + if (plen > 0) + payload = (const uint8_t *)entry + sizeof(*blob); + } else if (entry && len > 0) { + /* Defensive: an un-enveloped blob is treated as malformed. */ + return -EINVAL; + } + + /* Stamp the live election term so the WAL header records the LEADER's term + * for this entry (R1.1/R1.2/R1.4). */ + persist_set_term(p, term); + + /* Persist at LSN=index through the existing write path (persist-before-ACK + * and the bug-850 contiguity guard are preserved beneath these). */ + if (ha_is_primary(ha)) { + /* Own-append: persist_wal_async assigns the contiguous next LSN, which + * equals `index` for the leader (library index == current_lsn). */ + return persist_wal_async(p, type, payload, plen); + } + + return persist_wal_write_replicated(p, index, type, payload, plen); } static uint64_t ha_cb_get_entry_term_stub(void *ctx, uint64_t index) @@ -1560,6 +1624,15 @@ uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha) ha_do_promote(ha); return atomic_load_explicit(&ha->term, memory_order_acquire); } + +/* ha-term-safety task 2.1: drive the WIRED on_persist_entry callback so a unit + * test exercises the real adapter body (term stamp + WAL write) without + * raft_log_append (task 3.1). Exposed only under XCE_TESTING. */ +int ha_cb_persist_entry_for_test(ha_state_t *ha, uint64_t index, uint64_t term, + const void *entry, size_t len) +{ + return s_real_cbs.on_persist_entry(ha, index, term, entry, len); +} #endif /* ha_check_heartbeat_nocluster() / ha_watchdog_thread() / ha_start_watchdog() / diff --git a/src/controller/ha.h b/src/controller/ha.h index b68a513..1fbc1cd 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -758,6 +758,21 @@ int handle_request_vote(ha_state_t *ha, int in_catch_up_flag); #endif +#ifdef XCE_TESTING +/* + * ha_cb_persist_entry_for_test (ha-term-safety task 2.1): invoke the WIRED + * on_persist_entry callback (s_real_cbs.on_persist_entry) directly so a unit + * test exercises the real adapter body — term stamping + WAL write at the given + * 1-based index == LSN — without driving raft_log_append (task 3.1). `ha` must + * have raft_c set to a cluster whose persist module is started. The `entry` + * blob is the adapter envelope (raft_entry_blob_t prefix + payload). Returns + * the callback rc (0 on durable persist; non-zero on persist failure, in which + * case nothing is ACKed). TEST-ONLY. + */ +int ha_cb_persist_entry_for_test(ha_state_t *ha, uint64_t index, uint64_t term, + const void *entry, size_t len); +#endif + /* * ha_vote_self_freshness: compute the voter's effective freshness floor. * diff --git a/src/controller/testsuite/test_term_safety_adapter.c b/src/controller/testsuite/test_term_safety_adapter.c index 315abd0..1bd02b7 100644 --- a/src/controller/testsuite/test_term_safety_adapter.c +++ b/src/controller/testsuite/test_term_safety_adapter.c @@ -5,11 +5,15 @@ #include #include +#include + #include #include "config.h" #include "persist.h" #include "wal_types.h" +#include "xce/cluster_def.h" +#include "ha.h" /* ha-term-safety task 1.3 — WAL term seam + recovery read-back + legacy-0 rule. * @@ -215,6 +219,165 @@ static void test_entry_term_null_safe(void) printf("test_entry_term_null_safe: PASS\n"); } +/* ================================================================== */ +/* Task 2.1 — on_persist_entry callback body */ +/* ================================================================== */ + +/* Build a cluster_t whose embedded persist module is initialised + started in + * `tmpdir`, with ha.raft_c self-referencing the cluster (the convention every + * raft callback uses to recover persist/role from ctx == &c->ha). Returns a + * calloc'd cluster_t the caller must teardown via teardown_cluster. */ +static cluster_t* make_cluster(const char* tmpdir) +{ + cluster_t* c = calloc(1, sizeof(cluster_t)); + if (!c) return NULL; + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.snap_keep_n = 2; + + persist_set_workdir(&c->persist, tmpdir, "ctrl-test"); + if (persist_init(&c->persist, &cfg, c) != 0) { free(c); return NULL; } + if (persist_start(&c->persist) != 0) { + persist_fini(&c->persist); + free(c); + return NULL; + } + + /* The callback recovers persist/role from ctx == &c->ha → c->ha.raft_c. */ + c->ha.raft_c = c; + return c; +} + +static void teardown_cluster(cluster_t* c) +{ + if (!c) return; + persist_stop(&c->persist); + persist_fini(&c->persist); + free(c); +} + +/* Pack one adapter entry blob: [raft_entry_blob_t{type,len}][payload]. */ +static size_t pack_entry(uint8_t* buf, uint32_t type, + const void* payload, uint32_t plen) +{ + raft_entry_blob_t* h = (raft_entry_blob_t*)buf; + h->type = type; + h->payload_len = plen; + if (payload && plen > 0) + memcpy(buf + sizeof(*h), payload, plen); + return sizeof(*h) + plen; +} + +/* Test: the WIRED on_persist_entry stamps hdr.term from the library term at the + * supplied 1-based index == LSN, persisting through the existing WAL path; the + * recovered hdr.term equals the supplied term (R1.1–R1.4). Drives the standby + * (replicated) path so persist-before-ACK is exercised. */ +static void test_on_persist_entry_stamps_term_standby(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_ts_op_sb_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + const uint64_t T = 9; + + /* Phase 1: standby applies a replicated entry at LSN 1 with leader term T. */ + { + cluster_t* c = make_cluster(tmp); + TEST_ASSERT_NOT_NULL(c); + atomic_store(&c->ha.role, CTRL_STANDBY); + + uint8_t blob[64]; + size_t n = pack_entry(blob, WAL_JOB_UPDATE, "x", 1); + + TEST_ASSERT_EQUAL_INT(0, + ha_cb_persist_entry_for_test(&c->ha, /*index*/1, T, blob, n)); + + /* Persisted at LSN 1 with hdr.term == T (live read-back). */ + TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1)); + TEST_ASSERT_EQUAL_UINT64(2, c->persist.current_lsn); + + teardown_cluster(c); + } + + /* Phase 2: fresh recover reads the stamped term back from disk (R1.5). */ + { + cluster_t* c2 = make_cluster(tmp); + TEST_ASSERT_NOT_NULL(c2); + TEST_ASSERT_EQUAL_INT(0, persist_recover(&c2->persist)); + TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c2->persist, 1)); + teardown_cluster(c2); + } + + printf("test_on_persist_entry_stamps_term_standby: PASS\n"); +} + +/* Test: the primary's own-append path through on_persist_entry stamps the term + * and persists at the contiguous next LSN. */ +static void test_on_persist_entry_stamps_term_primary(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_ts_op_pr_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + const uint64_t T = 4; + + cluster_t* c = make_cluster(tmp); + TEST_ASSERT_NOT_NULL(c); + atomic_store(&c->ha.role, CTRL_PRIMARY); + + uint8_t blob[64]; + size_t n = pack_entry(blob, WAL_NOOP, NULL, 0); /* own-term promotion no-op */ + + TEST_ASSERT_EQUAL_INT(0, + ha_cb_persist_entry_for_test(&c->ha, /*index*/1, T, blob, n)); + TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); + + TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1)); + TEST_ASSERT_EQUAL_UINT64(2, c->persist.current_lsn); + + teardown_cluster(c); + printf("test_on_persist_entry_stamps_term_primary: PASS\n"); +} + +/* Test: a forced persist failure (fdatasync fail-loud seam on the replicated + * write) makes on_persist_entry return non-zero and DOES NOT advance the log / + * ACK (persist-before-ACK preserved). */ +static void test_on_persist_entry_failure_propagates_no_ack(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_ts_op_fail_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); + + uint64_t durable_before = + atomic_load(&c->persist.durable_lsn); + + /* Force the persist-before-ACK fdatasync to be treated as failed. */ + atomic_store(&c->persist.test_force_fsync_fail, 1); + + uint8_t blob[64]; + size_t n = pack_entry(blob, WAL_JOB_UPDATE, "y", 1); + + int rc = ha_cb_persist_entry_for_test(&c->ha, /*index*/1, 7, blob, n); + TEST_ASSERT_NOT_EQUAL_INT(0, rc); /* failure propagates → caller does NOT ACK */ + + /* Persist-before-ACK durability contract: the entry is NOT durable, so the + * durable watermark must NOT advance (the standby would ACK only on a + * successful, durable write — the non-zero rc above is exactly the signal + * that suppresses the ACK). durable_lsn (not the write cursor current_lsn) + * is the durability authority; it stays put on fdatasync failure. */ + TEST_ASSERT_EQUAL_UINT64(durable_before, + atomic_load(&c->persist.durable_lsn)); + + atomic_store(&c->persist.test_force_fsync_fail, 0); + teardown_cluster(c); + printf("test_on_persist_entry_failure_propagates_no_ack: PASS\n"); +} + /* ------------------------------------------------------------------ */ /* Suite entry point */ /* ------------------------------------------------------------------ */ @@ -226,4 +389,7 @@ void test_term_safety_adapter(void) RUN_TEST(test_recover_mixed_real_and_legacy_terms); RUN_TEST(test_entry_term_absent_lsn_is_zero); RUN_TEST(test_entry_term_null_safe); + 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); } diff --git a/src/include/wal_types.h b/src/include/wal_types.h index 8b17c7c..1e18948 100644 --- a/src/include/wal_types.h +++ b/src/include/wal_types.h @@ -23,6 +23,27 @@ typedef struct { uint32_t status; /* xce_queue status flags after this transition */ } wal_queuehdr_t; +/* ── Raft adapter entry-blob envelope (ha-term-safety task 2.1) ───────────── + * The raft library (src/lib/raft/raft.c) holds NO entry bytes: it carries one + * opaque (entry, len) blob through raft_log_append / on_persist_entry, leaving + * its encoding entirely to the controller adapter. This struct IS that + * encoding: a fixed 8-byte prefix recording the WAL entry `type` and the + * `payload_len` of the bytes that immediately follow it, so on_persist_entry + * can recover (type, payload, len) for the existing persist write path. + * + * This is purely an in-flight wire/blob format between the library callback + * and persist.c — it is NEVER written to the WAL header (that stays the 40-byte + * wal_entry_hdr_t, R1.7); only the decoded `type` + `payload` reach the WAL. + * + * The append/replication driver (tasks 3.1/3.2) packs this same prefix before + * the payload when calling raft_log_append / routing an AppendEntries, so the + * encoder and decoder share one definition (no drift). */ +typedef struct { + uint32_t type; /* wal_entry_type enum value for this entry */ + uint32_t payload_len; /* bytes of payload immediately following this header */ + /* uint8_t payload[payload_len] follows */ +} raft_entry_blob_t; + /* ============================================================ * WAL entry header — 40 bytes, 8-byte aligned (Part 2.2) * ============================================================ */