feat(ha-term-safety): drive raft_log_append on primary append path (task 3.1)

.kiro/specs/ha-term-safety/tasks.md

@@ -65,7 +65,7 @@
 
 ## 3. Integration: drive the library entry points on the controller paths
 
-- [ ] 3.1 Drive `raft_log_append` on the primary append path
+- [x] 3.1 Drive `raft_log_append` on the primary append path
   - Route the primary's job-state append (and the own-term promotion no-op) through `raft_log_append` so the library stamps the term and assigns the LSN, firing `on_persist_entry`
   - Observable completion: a primary `jobsub`-driven append goes through the library and lands in the WAL with the live term stamped; the promotion no-op records the newly established term
   - _Requirements: 1.1, 1.2, 2.1_
@@ -168,3 +168,8 @@
 - **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.
+
+- **3.1 done.** `raft_log_append` is now the SINGLE primary append entry point. NEW `ha_primary_log_append(c,type,payload,len)` (ha.c) packs the `raft_entry_blob_t{type,payload_len}+payload` envelope → `raft_log_append` → on_persist_entry → persist_wal_async (leaf). Hand-rolled bypass RETIRED: persist.c `wal_hook_append_classified` (~2332) + `persist_wal_sync_append` (~2047) route through `ha_primary_log_append` when `p->cluster != NULL` (standalone persist keeps direct path). `ha_arm_own_term_gate` NOOP also routes through it (stale current_lsn-1 LSN derivation deleted). Surviving direct persist_wal_async callers are ONLY: the leaf writer, on_persist_entry's library sink, and the `-EPERM` legacy fallback (stamps real `ha->term` via persist_set_term — never term 0, only the non-raft/engineless road). 3 tests; reviewer APPROVED (verified retire-not-parallel, no recursion [persist_wal_async is leaf], envelope parity, persist-before-act, stash-baseline proved bug-924 7-fail cluster pre-existing).
+  - **Anti-divergence invariant now LIVE + tested:** `p->current_lsn == s->log_len + 1` (next-LSN co-tracking) asserted across consecutive appends. Seeded at boot by 2.5 restore (s->log_len = durable_floor = current_lsn-1); preserved because raft_log_append (+1 to log_len) and its on_persist_entry→persist_wal_async (+1 to current_lsn) move in lockstep.
+  - **DEFERRED to 3.2:** heartbeat / 0-entry AppendEntries intake must route through `raft_on_append_entries_request` (UpdateTerm/logOk/(term,lsn) truncate via callbacks), replacing `ha_consume_heartbeat_committed` term handling + `ha_rejoin_reconcile` committed-boundary truncate; WalCatchupEntry still needs a `leaderCommit` capnp field + standby intake (deferred from 2.3).
+  - **PROCESS NOTE:** implementer first emitted `STATUS: COMPLETE` (non-parseable); corrected to READY_FOR_REVIEW. (Continue an existing subagent with SendMessage, not a fresh Agent call — a fresh Agent has no context.)

src/controller/ha.c

@@ -1866,6 +1866,15 @@ void ha_cb_commit_advanced_for_test(ha_state_t *ha, uint64_t commit_index)
 {
     s_real_cbs.on_commit_advanced(ha, commit_index);
 }
+
+/* ha-term-safety task 3.1: drive the (static) own-term gate arm — the WAL_NOOP
+ * append now routed through ha_primary_log_append — so a unit test can assert
+ * the promotion no-op records the live term via the library. Exposed only under
+ * XCE_TESTING. */
+void ha_arm_own_term_gate_for_test(cluster_t *c)
+{
+    ha_arm_own_term_gate(c);
+}
 #endif
 
 /* ha_check_heartbeat_nocluster() / ha_watchdog_thread() / ha_start_watchdog() /
@@ -2645,6 +2654,89 @@ static void ha_register_persist_hooks(cluster_t *c)
     c->ha.hooks_registered = 1;
 }
 
+/*
+ * ha_primary_log_append (ha-term-safety task 3.1, R1.1/R1.2/R2.1): the SINGLE
+ * primary append entry point. Routes a local primary append (job-state entry or
+ * the own-term promotion no-op) through the raft library's raft_log_append, so
+ * the LIBRARY stamps the live election term (s->term) and assigns the LSN
+ * (s->log_len+1), then drives on_persist_entry — which (primary role) performs
+ * the actual WAL write via persist_wal_async. This makes the library index the
+ * authority for both the per-entry term (R1) and the LSN, retiring the
+ * hand-rolled bypass that assigned the LSN directly through persist_wal_async.
+ *
+ * Co-tracking invariant (closes the task 2.1 deferral): because raft_log_append
+ * is the only primary append path, every primary append advances BOTH the
+ * library index (s->log_len) and the persist write cursor (p->current_lsn) by
+ * exactly one, in lockstep — so p->current_lsn == s->log_len + 1 always holds and
+ * the LSN the library assigns equals the LSN on_persist_entry writes. raft_init
+ * (task 2.5) seeds s->log_len from the recovered durable floor, which equals
+ * p->current_lsn - 1 at boot, establishing the invariant before the first append.
+ *
+ * Persist-before-act: raft_log_append calls on_persist_entry BEFORE advancing
+ * s->log_len and propagates a non-zero persist rc without mutating the log (no
+ * phantom entry) — preserving persist-before-ACK and the bug-850 contiguity
+ * guard beneath the WAL write.
+ *
+ * Legacy fallback (non-raft road): when the raft library is not in RAFT_LEADER
+ * state (raft_log_append returns -EPERM) — the init-time non-raft promotion road
+ * (ha_promote_to_primary) and engine-less unit-test contexts — the library is
+ * not tracking the log, so there is no co-tracking hazard. We append locally
+ * through the legacy persist_wal_async path (which still records the live term
+ * via the persist_set_term seam set on the prior promotion) so those roads keep
+ * working unchanged. Once raft drives leadership in production, the -EPERM
+ * branch is never taken and the library is the sole LSN/term source.
+ *
+ * out_lsn (optional) receives the assigned 1-based LSN on success.
+ * Returns 0 on success; propagates a non-zero persist rc on append failure.
+ */
+int ha_primary_log_append(cluster_t *c, uint32_t type,
+                          const void *payload, uint32_t len, uint64_t *out_lsn)
+{
+    if (!c)
+        return -EINVAL;
+
+    /* Pack the in-flight adapter envelope [raft_entry_blob_t{type,len}][payload]
+     * — the SAME format on_persist_entry (ha_cb_persist_entry_stub) unpacks, so
+     * the encoder and decoder share the single wal_types.h definition. */
+    size_t   blob_len = sizeof(raft_entry_blob_t) + len;
+    uint8_t  stackbuf[256];
+    uint8_t *blob = (blob_len <= sizeof(stackbuf)) ? stackbuf : malloc(blob_len);
+    if (!blob)
+        return -ENOMEM;
+
+    raft_entry_blob_t *bh = (raft_entry_blob_t *)blob;
+    bh->type        = type;
+    bh->payload_len = len;
+    if (payload && len > 0)
+        memcpy(blob + sizeof(*bh), payload, len);
+
+    uint64_t idx = 0;
+    int rc = raft_log_append(&c->ha.raft, blob, blob_len, &idx);
+
+    if (rc == -EPERM) {
+        /* Non-raft road: the library is not LEADER and is not tracking the log.
+         * Append locally through the legacy path (no library index to co-track).
+         * Stamp the live election term (ha->term) so the entry still records the
+         * leader's term (R1.1) even off the raft road — matching on_persist_entry's
+         * persist_set_term on the raft path. */
+        persist_set_term(&c->persist,
+                         atomic_load_explicit(&c->ha.term, memory_order_acquire));
+        rc = persist_wal_async(&c->persist, type, payload, len);
+        if (rc == 0) {
+            pthread_mutex_lock(&c->persist.lock);
+            idx = c->persist.current_lsn - 1; /* the LSN just assigned */
+            pthread_mutex_unlock(&c->persist.lock);
+        }
+    }
+
+    if (blob != stackbuf)
+        free(blob);
+
+    if (rc == 0 && out_lsn)
+        *out_lsn = idx;
+    return rc;
+}
+
 /*
  * ha_arm_own_term_gate: append the own-term WAL_NOOP and arm the commit
  * engine's writable_lsn gate (ha-single-leader-role task 2.2, Req 1.4/2.1/2.2).
@@ -2670,7 +2762,13 @@ static void ha_register_persist_hooks(cluster_t *c)
  */
 static void ha_arm_own_term_gate(cluster_t *c)
 {
-    int rc = persist_wal_async(&c->persist, WAL_NOOP, NULL, 0);
+    /* ha-term-safety task 3.1: route the own-term promotion no-op through the
+     * library (ha_primary_log_append → raft_log_append) so the WAL_NOOP records
+     * the newly established term (R1.2) and the library assigns its LSN. On the
+     * non-raft promotion road this falls back to the legacy append; either way
+     * the returned LSN is the no-op's WAL LSN. */
+    uint64_t noop_lsn = 0;
+    int rc = ha_primary_log_append(c, WAL_NOOP, NULL, 0, &noop_lsn);
     if (rc < 0) {
         ERROR("HA: own-term WAL_NOOP append failed: %d", rc);
         WARN("HA: own-term gate left UNSET (degraded-but-safe); first "
@@ -2678,13 +2776,6 @@ static void ha_arm_own_term_gate(cluster_t *c)
         return;
     }
 
-    /* Derive the no-op's LSN: the append post-incremented current_lsn, so the
-     * entry just written took current_lsn - 1 (read under p->lock). */
-    uint64_t noop_lsn;
-    pthread_mutex_lock(&c->persist.lock);
-    noop_lsn = c->persist.current_lsn - 1;
-    pthread_mutex_unlock(&c->persist.lock);
-
     /* Observability-only bookkeeping (lock-free, non-blocking, records
      * nothing per entry — commit state derives from registry watermarks). */
     (void)commit_submit(&c->commit, noop_lsn, COMMIT_MODE_RETAIN, 0);

src/controller/ha.h

@@ -317,6 +317,22 @@ char **controller_saved_argv(void);
  */
 int ha_self_restart(cluster_t *c);
 
+/*
+ * ha_primary_log_append (ha-term-safety task 3.1, R1.1/R1.2/R2.1): the single
+ * primary append entry point. Routes a local primary append (job-state entry or
+ * the own-term promotion no-op) through the raft library (raft_log_append) so
+ * the library stamps the live election term and assigns the LSN, firing
+ * on_persist_entry for the actual WAL write — making the library the sole
+ * LSN/term source and keeping p->current_lsn co-tracked with s->log_len. When
+ * the raft library is not RAFT_LEADER (the non-raft promotion road / engine-less
+ * tests), falls back to a legacy local persist_wal_async append (the library is
+ * not tracking the log, so no co-tracking hazard). `type` is a wal_entry_type;
+ * `payload`/`len` may be NULL/0 (e.g. WAL_NOOP). out_lsn (optional) receives the
+ * assigned 1-based LSN. Returns 0 on success; propagates a non-zero persist rc.
+ */
+int ha_primary_log_append(cluster_t *c, uint32_t type,
+                          const void *payload, uint32_t len, uint64_t *out_lsn);
+
 /*
  * ha_schedule_demotion_convergence: schedule the one-shot delaymgr task that
  * truncates above committed_lsn (Req 1.2), WARN-logs the discarded range
@@ -812,6 +828,14 @@ int ha_cb_send_append_entries_for_test(ha_state_t *ha, int peer_idx,
  * source. TEST-ONLY.
  */
 void ha_cb_commit_advanced_for_test(ha_state_t *ha, uint64_t commit_index);
+
+/*
+ * ha_arm_own_term_gate_for_test (ha-term-safety task 3.1): drive the static
+ * own-term gate arm (the promotion WAL_NOOP, now routed through
+ * ha_primary_log_append) so a unit test can assert the no-op records the live
+ * term via the library. TEST-ONLY.
+ */
+void ha_arm_own_term_gate_for_test(cluster_t *c);
 #endif
 
 /*

src/controller/persist.c

@@ -2032,7 +2032,19 @@ int persist_wal_sync_append(persist_t* p, uint32_t type,
                             const void* payload, uint32_t len,
                             int* sync_rc_out)
 {
-    int rc = persist_wal_async(p, type, payload, len);
+    /* ha-term-safety task 3.1: the primary's WAL_JOB_SUBMIT append is routed
+     * through the raft library (ha_primary_log_append → raft_log_append) so the
+     * library stamps the term + assigns the LSN (firing on_persist_entry → the
+     * WAL write), keeping p->current_lsn co-tracked with s->log_len; the existing
+     * majority-commit barrier below (repl_wait_majority_ack) is UNCHANGED and runs
+     * after the append. Non-raft road / engine-less tests (library not LEADER)
+     * fall back inside ha_primary_log_append to persist_wal_async. Standalone
+     * persist (p->cluster == NULL, lib-level tests) keeps the direct append. */
+    int rc;
+    if (p->cluster != NULL)
+        rc = ha_primary_log_append(p->cluster, type, payload, len, NULL);
+    else
+        rc = persist_wal_async(p, type, payload, len);
     if (rc != 0) {
         if (sync_rc_out)
             *sync_rc_out = rc;
@@ -2305,6 +2317,18 @@ static int wal_hook_append_classified(persist_t* p, uint32_t wal_type,
     uint64_t appended_lsn;
     int      rc;
 
+    /* ha-term-safety task 3.1: route the primary's local append through the raft
+     * library (ha_primary_log_append → raft_log_append) so the LIBRARY stamps the
+     * live election term and assigns the LSN (firing on_persist_entry → the WAL
+     * write), keeping p->current_lsn co-tracked with s->log_len. The library is
+     * the SINGLE primary append entry — no append bypasses it on the raft road.
+     * On the non-raft road (library not LEADER) ha_primary_log_append falls back
+     * to persist_wal_async, so engine-less / init-time contexts are unchanged.
+     * When p->cluster is NULL (persist used standalone in lib tests) keep the
+     * direct append (no cluster → no raft state to drive). */
+    if (p->cluster != NULL)
+        rc = ha_primary_log_append(p->cluster, wal_type, payload, len, NULL);
+    else
         rc = persist_wal_async(p, wal_type, payload, len);
     if (rc != 0 || mode != WAL_MODE_RETAIN)
         return rc;

src/controller/testsuite/test_term_safety_adapter.c

@@ -961,6 +961,147 @@ static void test_init_empty_wal_zero_floor(void)
     printf("test_init_empty_wal_zero_floor: PASS\n");
 }
 
+/* ================================================================== */
+/* Task 3.1 — drive raft_log_append on the primary append path         */
+/* ================================================================== */
+
+/* Bring a make_cluster()'d node up as the raft LEADER so raft_log_append is
+ * enabled (it returns -EPERM off the leader). We init the library from the
+ * recovered floor (task 2.5) and set the live election term + raft role
+ * directly — modelling a node that has already won an election at `term`
+ * (the cerebrum raft-sim convention: set state.term after a clean election;
+ * raft_log_append stamps entries from s->term). */
+static void make_leader(cluster_t* c, uint64_t term, int n_peers)
+{
+    atomic_store(&c->ha.role, CTRL_PRIMARY);
+    atomic_store(&c->ha.term, term);
+    ha_init_raft(&c->ha, c, n_peers);
+    c->ha.raft.term = term;
+    c->ha.raft.role = RAFT_LEADER;
+}
+
+/* Test: a primary job-state append driven through ha_primary_log_append goes
+ * through the library — the library assigns the LSN (== s->log_len post-append),
+ * stamps the live election term into the WAL header, and keeps the persist
+ * write cursor co-tracked with the library index (current_lsn == log_len + 1).
+ * Against the pre-3.1 hand-rolled path this entry point does not exist
+ * (compile RED) — and once it exists the library MUST be the LSN/term source
+ * (R1.1, R1.2, R2.1). */
+static void test_primary_log_append_stamps_term_and_assigns_lsn(void)
+{
+    char tmp[256];
+    strcpy(tmp, "/tmp/xce_ts_plap_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
+
+    const uint64_t T = 7;
+
+    cluster_t* c = make_cluster(tmp);
+    TEST_ASSERT_NOT_NULL(c);
+    make_leader(c, T, /*n_peers*/1);
+
+    /* Fresh leader, empty log: first append takes library index 1. */
+    TEST_ASSERT_EQUAL_UINT64(0, c->ha.raft.log_len);
+
+    wal_jobstate_t js = { .jobid = 42, .state = 3 };
+    uint64_t lsn = 0;
+    TEST_ASSERT_EQUAL_INT(0,
+        ha_primary_log_append(c, WAL_JOB_UPDATE, &js, sizeof(js), &lsn));
+
+    /* (a) The LIBRARY assigned the LSN and grew its log to it. */
+    TEST_ASSERT_EQUAL_UINT64(1, lsn);
+    TEST_ASSERT_EQUAL_UINT64(1, c->ha.raft.log_len);
+
+    /* (b) The WAL header recorded the live election term for that LSN. */
+    TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1));
+
+    /* (d) Co-tracking invariant: the persist write cursor advanced exactly in
+     * lockstep with the library index — current_lsn == log_len + 1 (next LSN). */
+    TEST_ASSERT_EQUAL_UINT64(c->ha.raft.log_len + 1, c->persist.current_lsn);
+
+    /* A second append keeps the two counters co-tracked. */
+    js.jobid = 43;
+    TEST_ASSERT_EQUAL_INT(0,
+        ha_primary_log_append(c, WAL_JOB_UPDATE, &js, sizeof(js), &lsn));
+    TEST_ASSERT_EQUAL_UINT64(2, lsn);
+    TEST_ASSERT_EQUAL_UINT64(2, c->ha.raft.log_len);
+    TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 2));
+    TEST_ASSERT_EQUAL_UINT64(c->ha.raft.log_len + 1, c->persist.current_lsn);
+
+    teardown_cluster(c);
+    printf("test_primary_log_append_stamps_term_and_assigns_lsn: PASS\n");
+}
+
+/* Test: the own-term promotion no-op (ha_arm_own_term_gate) is routed through
+ * the library, so the WAL_NOOP entry records the newly established term (not 0)
+ * and the library index tracks it (R1.2). Pre-3.1 the noop went through a
+ * hand-rolled persist_wal_async with a best-effort p->term and did NOT advance
+ * s->log_len. */
+static void test_own_term_noop_records_live_term_via_library(void)
+{
+    char tmp[256];
+    strcpy(tmp, "/tmp/xce_ts_noop_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
+
+    const uint64_t T = 5;
+
+    cluster_t* c = make_cluster(tmp);
+    TEST_ASSERT_NOT_NULL(c);
+    make_leader(c, T, /*n_peers*/1);
+
+    /* Register the persist hooks then arm the own-term gate exactly as the
+     * promotion tail does (ha_promote_to_primary / raft_cb_become_leader). */
+    ha_arm_own_term_gate_for_test(c);
+    TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist));
+
+    /* The no-op landed at library index 1 with the live term recorded. */
+    TEST_ASSERT_EQUAL_UINT64(1, c->ha.raft.log_len);
+    TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1));
+    TEST_ASSERT_EQUAL_UINT64(c->ha.raft.log_len + 1, c->persist.current_lsn);
+
+    teardown_cluster(c);
+    printf("test_own_term_noop_records_live_term_via_library: PASS\n");
+}
+
+/* Test: when raft leadership is NOT active (the library is not RAFT_LEADER —
+ * the init-time non-raft promotion road and engine-less test contexts),
+ * ha_primary_log_append still appends locally through the legacy path so the
+ * non-raft promotion tail keeps working, WITHOUT growing the library log (the
+ * library is not tracking the log on that road, so no co-tracking hazard).
+ * The entry still records the live election term via persist_set_term. */
+static void test_primary_log_append_non_leader_legacy_fallback(void)
+{
+    char tmp[256];
+    strcpy(tmp, "/tmp/xce_ts_plapfb_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
+
+    const uint64_t T = 6;
+
+    cluster_t* c = make_cluster(tmp);
+    TEST_ASSERT_NOT_NULL(c);
+    /* Non-raft promotion road: controller role PRIMARY + live term, but the
+     * raft library was never driven to LEADER (raft.role stays FOLLOWER). */
+    atomic_store(&c->ha.role, CTRL_PRIMARY);
+    atomic_store(&c->ha.term, T);
+    ha_init_raft(&c->ha, c, /*n_peers*/1);
+    TEST_ASSERT_NOT_EQUAL_INT(RAFT_LEADER, c->ha.raft.role);
+
+    wal_jobstate_t js = { .jobid = 7, .state = 1 };
+    uint64_t lsn = 0;
+    TEST_ASSERT_EQUAL_INT(0,
+        ha_primary_log_append(c, WAL_JOB_UPDATE, &js, sizeof(js), &lsn));
+
+    /* Appended at the contiguous next LSN (current_lsn started at 1). */
+    TEST_ASSERT_EQUAL_UINT64(1, lsn);
+    TEST_ASSERT_EQUAL_UINT64(2, c->persist.current_lsn);
+    /* Live term still recorded for the entry. */
+    TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1));
+    /* The library log was NOT grown by the legacy fallback (no raft leadership). */
+    TEST_ASSERT_EQUAL_UINT64(0, c->ha.raft.log_len);
+
+    teardown_cluster(c);
+    printf("test_primary_log_append_non_leader_legacy_fallback: PASS\n");
+}
+
 /* ------------------------------------------------------------------ */
 /* Suite entry point                                                   */
 /* ------------------------------------------------------------------ */
@@ -987,4 +1128,7 @@ void test_term_safety_adapter(void)
     RUN_TEST(test_on_commit_advanced_wired_feeds_barrier);
     RUN_TEST(test_init_restores_recovered_floor);
     RUN_TEST(test_init_empty_wal_zero_floor);
+    RUN_TEST(test_primary_log_append_stamps_term_and_assigns_lsn);
+    RUN_TEST(test_own_term_noop_records_live_term_via_library);
+    RUN_TEST(test_primary_log_append_non_leader_legacy_fallback);
 }