feat(ha-term-safety): on_persist_entry callback body (task 2.1)

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).

src/controller/testsuite/test_term_safety_adapter.c

@@ -5,11 +5,15 @@
 #include <stdio.h>
 #include <unistd.h>
 
+#include <stdatomic.h>
+
 #include <unity/unity.h>
 
 #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);
 }