test(raft-ha-core-safety): harness infrastructure + payload-loss smoke-RED (task 1.4)

New test_ha_core_safety.c with deterministic gate/probe seams (no sleeps,
drain-before-fini per bug-1224): payload-controllable WAL writer, on-disk
(term,pcrc,len) readback probe, walcompare observer, apply-gap + kill/
election injection for phases 2.3/4.1/5.1. Smoke-RED (Expected 64 Was 0)
proves the payload-blind (term,lsn)-only guard at persist.c:822 drops a
corrective full-payload write at a committed LSN. 5 tests/1 RED/4 seam
PASS, no ASAN errors.

.kiro/specs/raft-ha-core-safety/tasks.md

@@ -25,7 +25,7 @@
   - _Requirements: 7.1, 1.1_
   - _Boundary: Deterministic Chaotic Harness_
 
-- [ ] 1.4 Stand up the unit/integration harness infrastructure
+- [x] 1.4 Stand up the unit/integration harness infrastructure
   - Create `test_ha_core_safety.c` with the deterministic gate/probe seams (no sleeps; teardown drains pending delaymgr tasks to avoid the bug-1224 teardown-race UAF), the `(term, lsn)`/apply-gap/kill-injection fixtures, and register it in the controller unit-test build; include one smoke-RED case proving the harness can observe a divergence. The per-defect RED bodies are owned by their phases (2.3 ingest, 4.1 reconvergence, 5.1 election) — this task provides the shared seams they build on, not those cases
   - Observable completion: `test_ha_core_safety` builds and runs, the injection fixtures and probe seams are usable by later phases, and the smoke-RED case FAILs on the baseline with no ASAN teardown errors
   - _Requirements: 7.1, 7.4_

src/controller/meson.build

@@ -145,6 +145,7 @@ controller_unit_test_src = [
   'testsuite/test_commit_admission_pool.c',
   'testsuite/test_ha_vep_teardown.c',
   'testsuite/test_ha_dispatch_integration.c',
+  'testsuite/test_ha_core_safety.c',
   'testsuite/test_waldump.c',
   'testsuite/waldump.c',
   'testsuite/test_walcompare.c',

src/controller/testsuite/controller_test.c

@@ -71,6 +71,7 @@ extern void test_slowlane(void);
 extern void test_commit_admission_pool(void);
 extern void test_ha_vep_teardown(void);
 extern void test_ha_dispatch_integration(void);
+extern void test_ha_core_safety(void);
 
 void setUp(void) { return; }
 
@@ -211,6 +212,7 @@ int main(int argc, char* argv[])
     TEST(test_commit_admission_pool);
     TEST(test_ha_vep_teardown);
     TEST(test_ha_dispatch_integration);
+    TEST(test_ha_core_safety);
 
     return UNITY_END();
 }

src/controller/testsuite/test_ha_core_safety.c

@@ -0,0 +1,411 @@
+/*
+ * test_ha_core_safety.c — raft-ha-core-safety Phase 1, Task 1.4:
+ * the DETERMINISTIC CHAOTIC HARNESS infrastructure (shared seams only).
+ *
+ * This file stands up the reusable gate/probe/injection seams the later
+ * RED phases build their per-defect cases on.  It does NOT implement those
+ * defect cases — each is owned by its phase:
+ *
+ *   - 2.3 cross-term catch-up streamer payload-integrity  → uses the WAL
+ *         construction + readback-probe seams here.
+ *   - 4.1 standby reconvergence (apply-gap)               → uses the apply-gap
+ *         injection seam here.
+ *   - 5.1 election liveness / Election Safety             → uses the
+ *         kill/election injection seam here.
+ *
+ * Retargeted root cause (task 1.3, verified — see research.md §"Task 1.3"):
+ * the live committed divergence is CROSS-TERM CATCH-UP PAYLOAD LOSS — a
+ * non-empty entry persisted with PLEN=0 at a committed LSN whose (term, lsn)
+ * are CORRECT — NOT the refuted term-blind bypass.  So the seams here are
+ * built to construct and observe a payload-divergent entry at an existing
+ * (term, lsn): a payload-controllable WAL writer over the real persist guard
+ * (the very (term,lsn)-only idempotency guard the spec hardens in 2.2), an
+ * on-disk (term, pcrc, len) readback probe, and the walcompare (term,lsn,pcrc)
+ * divergence observer.
+ *
+ * Determinism: every seam is synchronous over a real persist_t WAL — NO
+ * sleeps on the correctness path.  Teardown DRAINS the persist background
+ * worker (persist_stop joins the fsync thread) BEFORE persist_fini, so no
+ * just-woken body races the teardown (the bug-1224 teardown-race UAF class).
+ * The seams here post NO delaymgr/pool work (ha pool is left NULL); a later
+ * phase that dispatches onto delaymgr/pool must add the same drain-before-fini
+ * discipline used by test_ha_dispatch_integration.c.
+ *
+ * Smoke-RED (the ONE case that FAILs on the baseline): the payload-blind
+ * persistence guard silently no-ops a corrective full-payload write over a
+ * previously-stored empty (PLEN=0) entry at the same committed LSN — so the
+ * WAL keeps the empty body.  This is the minimal observable form of the live
+ * payload-loss signature; it goes GREEN when task 2.2 makes the guard
+ * (term, lsn, payload-crc)-aware.
+ *
+ * Requirements: 7.1 (force divergence RED on baseline), 7.4 (unit harness).
+ */
+#define _GNU_SOURCE
+#include <stdatomic.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#include "ha.h"
+#include "persist.h"
+#include "wal_types.h"
+#include "walcompare.h"
+#include "waldump.h"
+
+/* ── Fixtures ──────────────────────────────────────────────────────────────
+ * A member is a real persist_t WAL on its own temp dir.  This is the exact
+ * receiver-side persistence surface the spec hardens, so a divergence
+ * constructed through it is the live divergence, not a synthetic stand-in. */
+
+typedef struct {
+    persist_t* p;
+    char       root[256];        /* mkdtemp root */
+    char       wal_dir[320];     /* <root>/ctrl/data/wal — what waldump reads */
+} hcs_member_t;
+
+/* Bring one member up.  max_wal_records is pinned high so no checkpoint is
+ * scheduled (keeps the harness off the delaymgr path — drain-clean teardown). */
+static int hcs_member_up(hcs_member_t* m, const char* suffix)
+{
+    memset(m, 0, sizeof(*m));
+    m->p = calloc(1, sizeof(persist_t));
+    if (!m->p)
+        return -1;
+
+    persist_config_t cfg = persist_config_defaults();
+    cfg.wal_sync_ms     = 5000;
+    cfg.sync_timeout_ms = 50;
+    cfg.max_wal_records = 1u << 30; /* never auto-checkpoint during a unit test */
+
+    snprintf(m->root, sizeof(m->root), "/tmp/xce_hcs_%s_XXXXXX", suffix);
+    if (!mkdtemp(m->root)) {
+        free(m->p);
+        m->p = NULL;
+        return -1;
+    }
+    snprintf(m->wal_dir, sizeof(m->wal_dir), "%s/ctrl/data/wal", m->root);
+
+    persist_set_workdir(m->p, m->root, "ctrl");
+    persist_init(m->p, &cfg, NULL);
+    persist_start(m->p);
+    return 0;
+}
+
+/* Drain + tear down: persist_stop joins the fsync worker BEFORE persist_fini,
+ * so no background body races the free (bug-1224 teardown-race discipline). */
+static void hcs_member_down(hcs_member_t* m)
+{
+    if (!m->p)
+        return;
+    persist_stop(m->p);   /* joins the background fsync thread (the drain) */
+    persist_fini(m->p);
+    free(m->p);
+    m->p = NULL;
+}
+
+/* ── Seam 1: payload-controllable WAL entry construction ────────────────────
+ * Write an entry with a caller-chosen (term, lsn, type, payload, len) through
+ * the real replicated-write guard (persist.c:802).  term is stamped via
+ * persist_set_term (what write_replicated copies into hdr.term).  A len==0 /
+ * payload==NULL write is the cross-term catch-up PLEN=0 signature 2.3 injects.
+ * Returns the guard's rc verbatim (0 = persisted or idempotent-skipped). */
+static int hcs_wal_put(hcs_member_t* m, uint64_t lsn, uint64_t term,
+                       uint32_t type, const void* payload, uint32_t len)
+{
+    persist_set_term(m->p, term);
+    return persist_wal_write_replicated(m->p, lsn, type, payload, len);
+}
+
+/* ── Seam 2: persisted (term, pcrc, len) readback probe ─────────────────────
+ * Reads back what is PHYSICALLY on disk at an LSN — the same (term, pcrc)
+ * fingerprint walcompare compares across members, plus the payload length that
+ * disambiguates an empty (PLEN=0) body from a full one.  On-disk truth, not
+ * in-memory state: this is how a phase asserts "the corrective write actually
+ * landed".  Returns 0 on a successful scan (set out->present), <0 on a hard
+ * scan error. */
+typedef struct {
+    int      present;
+    uint64_t term;   /* raw on-disk term (legacy 0 kept verbatim) */
+    uint32_t pcrc;   /* payload-only CRC32 (0 for an empty body) */
+    uint32_t len;    /* payload byte count (0 for an empty body) */
+} hcs_probe_t;
+
+typedef struct {
+    uint64_t     want_lsn;
+    hcs_probe_t* out;
+} hcs_probe_ctx_t;
+
+static int hcs_probe_visit(void* ctx, const waldump_entry_t* e)
+{
+    hcs_probe_ctx_t* pc = ctx;
+    if (e->lsn != pc->want_lsn)
+        return 0; /* keep scanning */
+    pc->out->present = 1;
+    pc->out->term    = e->term;
+    pc->out->pcrc    = e->pcrc;
+    pc->out->len     = e->payload_len;
+    return 1; /* found — stop the scan */
+}
+
+static int hcs_wal_probe(hcs_member_t* m, uint64_t lsn, hcs_probe_t* out)
+{
+    memset(out, 0, sizeof(*out));
+    hcs_probe_ctx_t pc = { .want_lsn = lsn, .out = out };
+    FILE*           sink = fopen("/dev/null", "w");
+    if (!sink)
+        return -1;
+    int rc = waldump_path_visit(m->wal_dir, sink, NULL, hcs_probe_visit, &pc);
+    fclose(sink);
+    return rc;
+}
+
+/* ── Seam 3: cross-member divergence observation (walcompare) ───────────────
+ * Run the (term, lsn, pcrc) committed-prefix detector over N members' WAL
+ * dirs and surface the verdict + first divergent identity.  This is how a
+ * phase asserts a constructed payload divergence is OBSERVABLE as a committed
+ * safety violation. */
+#define HCS_MAX_MEMBERS 8
+static int hcs_walcompare(hcs_member_t** members, int n, uint64_t committed_lsn,
+                          walcompare_divergence_t* div)
+{
+    walcompare_member_t wm[HCS_MAX_MEMBERS];
+    for (int i = 0; i < n; i++) {
+        wm[i].path          = members[i]->wal_dir;
+        wm[i].committed_lsn = committed_lsn;
+    }
+    return walcompare(wm, n, NULL, div);
+}
+
+/* ── Seam 4: apply-gap injection (for 4.1 reconvergence) ────────────────────
+ * Park a member with committed > applied so a phase can assert it must NOT be
+ * reported converged until applied catches the committed boundary.  Both are
+ * test-only atomic pokes of the canonical fields (ha->applied_lsn,
+ * persist->committed_lsn). */
+static void hcs_inject_apply_gap(ha_state_t* ha, persist_t* p,
+                                 uint64_t applied_lsn, uint64_t committed_lsn)
+{
+    atomic_store_explicit(&p->committed_lsn, committed_lsn, memory_order_release);
+    atomic_store_explicit(&ha->applied_lsn, applied_lsn, memory_order_release);
+}
+
+/* Committed-minus-applied gap (0 == fully applied / converged-eligible). */
+static uint64_t hcs_apply_gap(ha_state_t* ha, persist_t* p)
+{
+    uint64_t committed = atomic_load_explicit(&p->committed_lsn,
+                                              memory_order_acquire);
+    uint64_t applied   = atomic_load_explicit(&ha->applied_lsn,
+                                              memory_order_acquire);
+    return committed > applied ? committed - applied : 0;
+}
+
+/* ── Seam 5: kill/election injection (for 5.1 election liveness/safety) ──────
+ * Deterministically advance a member's (term, role) the way a completed
+ * election round would, WITHOUT spinning real election threads — so a phase
+ * can script rapid rounds and assert per-term primary counts (Election Safety)
+ * and a higher term each round (liveness) without sleeps. */
+static uint64_t hcs_inject_election(ha_state_t* ha, uint64_t new_term,
+                                    int32_t new_role)
+{
+    atomic_store_explicit(&ha->term, new_term, memory_order_release);
+    atomic_store_explicit(&ha->role, new_role, memory_order_release);
+    return atomic_load_explicit(&ha->term, memory_order_acquire);
+}
+
+/* A 64-byte non-empty body whose payload CRC is non-zero — stands in for a
+ * real JOB_SUBMIT body (the live divergence dropped a 3560-byte one). */
+#define HCS_FULL_LEN 64u
+static void hcs_fill_body(uint8_t* buf)
+{
+    for (uint32_t i = 0; i < HCS_FULL_LEN; i++)
+        buf[i] = (uint8_t)(0xA0 ^ i);
+}
+
+/* ═══════════════════════════════════════════════════════════════════════════
+ * SMOKE-RED — payload-blind persistence guard (FAILs on the baseline)
+ *
+ * Construct the live payload-loss signature at a single committed LSN: an
+ * empty (PLEN=0) cross-term catch-up body lands first, then the CORRECT full
+ * body is streamed for the SAME (term, lsn).  The baseline guard is
+ * LSN-only-idempotent (persist.c:822 `lsn+1 <= current_lsn → return 0`), so it
+ * silently no-ops the corrective write and the WAL keeps the empty body.
+ *
+ * The harness OBSERVES the divergence via the readback probe: the entry is
+ * present, term is correct, but the payload is gone (len==0, pcrc==0).  This
+ * FAILs now and goes GREEN when task 2.2 makes the guard
+ * (term, lsn, payload-crc)-aware (a differing payload-crc at an existing LSN
+ * is a conflict to repair, not an idempotent skip).
+ * ═══════════════════════════════════════════════════════════════════════════ */
+static void test_smoke_red_payload_blind_guard_drops_corrective_write(void)
+{
+    hcs_member_t m;
+    TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&m, "smoke"));
+
+    uint8_t full[HCS_FULL_LEN];
+    hcs_fill_body(full);
+
+    /* (1) cross-term catch-up ships an empty body for committed LSN 1. */
+    int rc_empty = hcs_wal_put(&m, /*lsn=*/1, /*term=*/3, WAL_JOB_SUBMIT,
+                               /*payload=*/NULL, /*len=*/0);
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_empty,
+        "the empty replicated write must persist (needs a real fs for the "
+        "inline fdatasync; if this is -EIO you are on overlayfs — bug-924)");
+
+    /* (2) the CORRECT full body is later streamed for the same (term, lsn). */
+    int rc_full = hcs_wal_put(&m, /*lsn=*/1, /*term=*/3, WAL_JOB_SUBMIT,
+                              full, HCS_FULL_LEN);
+    TEST_ASSERT_TRUE_MESSAGE(rc_full == 0,
+        "the corrective full-payload write returns 0 (baseline: an idempotent "
+        "no-op; fixed: an accepted repair) — either way rc==0");
+
+    /* (3) OBSERVE: on disk the entry must end with the FULL payload. */
+    hcs_probe_t pr;
+    TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&m, 1, &pr));
+    TEST_ASSERT_TRUE_MESSAGE(pr.present,
+        "committed LSN 1 must be present after the corrective write");
+    TEST_ASSERT_EQUAL_UINT32_MESSAGE(HCS_FULL_LEN, pr.len,
+        "RED: the payload-blind (term,lsn)-only guard silently no-ops the "
+        "corrective full-payload write — the WAL keeps the empty (PLEN=0) "
+        "body at a committed LSN (cross-term catch-up payload loss)");
+    TEST_ASSERT_NOT_EQUAL_MESSAGE(0u, pr.pcrc,
+        "RED: the committed entry must hold the real payload CRC, not 0");
+
+    hcs_member_down(&m);
+    printf("test_smoke_red_payload_blind_guard_drops_corrective_write: done\n");
+}
+
+/* ═══════════════════════════════════════════════════════════════════════════
+ * SCAFFOLDING (PASS) — prove each shared seam is usable by later phases.
+ * ═══════════════════════════════════════════════════════════════════════════ */
+
+/* Seam 1+2: a payload-controllable write is read back faithfully on disk. */
+static void test_seam_wal_construct_and_probe(void)
+{
+    hcs_member_t m;
+    TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&m, "probe"));
+
+    uint8_t full[HCS_FULL_LEN];
+    hcs_fill_body(full);
+    TEST_ASSERT_EQUAL_INT(0,
+        hcs_wal_put(&m, 1, /*term=*/7, WAL_JOB_SUBMIT, full, HCS_FULL_LEN));
+
+    hcs_probe_t pr;
+    TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&m, 1, &pr));
+    TEST_ASSERT_TRUE(pr.present);
+    TEST_ASSERT_EQUAL_UINT64(7, pr.term);
+    TEST_ASSERT_EQUAL_UINT32(HCS_FULL_LEN, pr.len);
+    TEST_ASSERT_NOT_EQUAL(0u, pr.pcrc);
+
+    /* An absent LSN reads back not-present (scan succeeds). */
+    hcs_probe_t pr2;
+    TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&m, 2, &pr2));
+    TEST_ASSERT_FALSE(pr2.present);
+
+    hcs_member_down(&m);
+    printf("test_seam_wal_construct_and_probe: PASS\n");
+}
+
+/* Seam 3: a constructed payload divergence at an identical (term, lsn) is
+ * OBSERVABLE as a committed-prefix safety violation by walcompare. */
+static void test_seam_divergence_is_observable(void)
+{
+    hcs_member_t maj, min_;
+    TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&maj, "maj"));
+    TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&min_, "min"));
+
+    uint8_t full[HCS_FULL_LEN];
+    hcs_fill_body(full);
+
+    /* Same (term, lsn) on both, but the minority holds an empty body — the
+     * exact live signature (term identical, only payload-CRC diverges). */
+    TEST_ASSERT_EQUAL_INT(0,
+        hcs_wal_put(&maj, 1, /*term=*/3, WAL_JOB_SUBMIT, full, HCS_FULL_LEN));
+    TEST_ASSERT_EQUAL_INT(0,
+        hcs_wal_put(&min_, 1, /*term=*/3, WAL_JOB_SUBMIT, NULL, 0));
+
+    hcs_member_t*           ms[2] = { &maj, &min_ };
+    walcompare_divergence_t div;
+    int verdict = hcs_walcompare(ms, 2, /*committed_lsn=*/1, &div);
+
+    TEST_ASSERT_EQUAL_INT_MESSAGE(WALCMP_DIVERGENT, verdict,
+        "the divergence-observation seam must flag the payload divergence");
+    TEST_ASSERT_EQUAL_UINT64(1, div.lsn);
+    TEST_ASSERT_EQUAL_INT(1, div.committed);
+    TEST_ASSERT_NOT_EQUAL_MESSAGE(div.member_pcrc[0], div.member_pcrc[1],
+        "the two members must disagree on payload CRC at the same committed LSN");
+
+    hcs_member_down(&maj);
+    hcs_member_down(&min_);
+    printf("test_seam_divergence_is_observable: PASS\n");
+}
+
+/* Shared ha_config for the ha-state seams (no pool → no delaymgr work). */
+static ha_config_t g_hcs_cfg = {
+    .heartbeat_interval_ms     = 3000,
+    .heartbeat_timeout_ms      = 6000,
+    .election_timeout_ms       = 500,
+    .probe_connect_timeout_ms  = 50,
+    .max_promotion_retries     = 3,
+    .election_jitter_max_ms    = 0,
+    .election_priority_step_ms = 500,
+};
+
+/* Seam 4: the apply-gap injection seam parks committed > applied and the gap
+ * query reflects it (used by 4.1 to gate "converged"). */
+static void test_seam_apply_gap_injection(void)
+{
+    hcs_member_t m;
+    TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&m, "gap"));
+    ha_state_t ha;
+    ha_init(&ha, &g_hcs_cfg, 0);
+
+    hcs_inject_apply_gap(&ha, m.p, /*applied=*/5, /*committed=*/9);
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, hcs_apply_gap(&ha, m.p),
+        "an injected apply gap must be observable (committed 9 - applied 5)");
+
+    /* Closing the gap makes the member converged-eligible (gap == 0). */
+    hcs_inject_apply_gap(&ha, m.p, /*applied=*/9, /*committed=*/9);
+    TEST_ASSERT_EQUAL_UINT64(0, hcs_apply_gap(&ha, m.p));
+
+    ha_fini(&ha);
+    hcs_member_down(&m);
+    printf("test_seam_apply_gap_injection: PASS\n");
+}
+
+/* Seam 5: the kill/election injection seam advances (term, role)
+ * deterministically (used by 5.1 to script rapid rounds, no sleeps). */
+static void test_seam_election_injection(void)
+{
+    ha_state_t ha;
+    ha_init(&ha, &g_hcs_cfg, 0);
+
+    uint64_t prev = atomic_load_explicit(&ha.term, memory_order_acquire);
+    for (int round = 1; round <= 3; round++) {
+        uint64_t t = hcs_inject_election(&ha, prev + 1, (int32_t)CTRL_PRIMARY);
+        TEST_ASSERT_EQUAL_UINT64_MESSAGE(prev + 1, t,
+            "each scripted round must yield a strictly higher term (liveness)");
+        TEST_ASSERT_EQUAL_INT32(CTRL_PRIMARY,
+            atomic_load_explicit(&ha.role, memory_order_acquire));
+        prev = t;
+    }
+
+    ha_fini(&ha);
+    printf("test_seam_election_injection: PASS\n");
+}
+
+/* ── Suite entry point ─────────────────────────────────────────────────────*/
+
+void test_ha_core_safety(void)
+{
+    UNITY_BEGIN();
+    /* The ONE smoke-RED: payload-blind guard drops the corrective write. */
+    RUN_TEST(test_smoke_red_payload_blind_guard_drops_corrective_write);
+    /* Scaffolding: each shared seam is usable by 2.3 / 4.1 / 5.1. */
+    RUN_TEST(test_seam_wal_construct_and_probe);
+    RUN_TEST(test_seam_divergence_is_observable);
+    RUN_TEST(test_seam_apply_gap_injection);
+    RUN_TEST(test_seam_election_injection);
+    UNITY_END();
+}