fix(controller): extract job-submit commit barrier out of the hook fan-out — stop committed-job-loss (bug-1306)

raft-apply-decoupling item #1 (down payment). xce_run_hooks returns only the
LAST-registered hook's rc, so a later JOB AFTER_EVENT subscriber (sched/
apiserver) returning 0 clobbered persist_job_hook_fun's commit -ETIMEDOUT.
handle_jobsubmit then replied 'jobsubmit ok' for an entry that never reached
majority, and a new-term leader legally discarded it (committed-job-loss,
Commit Durability violation).

- persist.c/.h: new persist_submit_wal_and_commit(p, jlist) — the explicit raft
  step (WAL append + majority-commit barrier) returning the commit verdict
  directly. persist_job_hook_fun's WAL_MODE_WAIT branch is now a no-op.
- job.c handle_jobsubmit: call the explicit step, read commit_rc DIRECTLY,
  reject on <0 before firing the now notification-only AFTER hook
  (canonical commit -> notify -> reply).
- tests: +test_job_hook_new_is_notify_only regression pin; 4 hook tests
  relocated to drive the explicit step. persist_hooks 19/0, e2e_hooks 3/0,
  commit_barrier_reply 22/0, commit_durability 38/0.

Proven: 10-round chaotic soak committed_job_loss 5 -> 0 (same election code).
Residual soak FAILs (divergence catch-up lag + a round-4 election-escalation
tail) are NOT job loss — tracked as task 5.10.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JDYFbrgyDHbvNYcxApKhrX

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

@@ -165,14 +165,20 @@
   - _Boundary: Async-IO dispatch (endpoint lifetime)_
   - _Depends: 5.7_
 
-- [ ] 5.9 Committed job loss + prefix divergence under chaotic failover (NEW — the round-4..6 7.2 blocker AFTER 5.8; pre-existing, EXTERNAL spec)
+- [x] 5.9 Committed JOB LOSS under chaotic failover — FIXED (the durability half; divergence tracked separately in 5.10)
-  - _Surfaced by the FIRST soak ever to reach rounds 4-6 (every prior soak crashed at round 4 — bug-1282/1286/1298). With the crash gone, the 6-round chaotic soak FAILs the verdict: committed_job_loss=4, committed_divergence=7. Rounds 1-3 are CLEAN (all jobs present, zero divergence); loss begins exactly at round 4 — the first round whose overlapping kills elect a JUST-RESTARTED node. A committed job (e.g. 0-0-1782829181-1) is permanently absent on the reachable primary; the gap persists after settling. This is the documented bug-795/828/851 FAILOVER CATCH-UP GAP / committed-prefix divergence: a restarted node rejoins and is elected primary without backfilling the full committed prefix (empty-rejoin catch-up streams [watermark..committed], no snapshot backfill → permanent gap). NOT a 5.8 refcount regression: ASAN is clean (refcount governs ep lifetime, not WAL content; cannot selectively drop one committed job), and rounds 1-3 are loss-free._
+  - _ROOT CAUSE (sharpened 2026-06-30, waldump+log bisect — NOT the catch-up prefix-gap the earlier note guessed): `xce_run_hooks` (src/lib/hook.c:134-157) returns only the LAST-registered hook's rc. The durable-majority gate worked (a term-9 leader's job at LSN 30 timed out commit → `commit_submit` -ETIMEDOUT → `persist_job_hook_fun` returned -ETIMEDOUT), but a later JOB AFTER_EVENT hook (sched/apiserver) returned 0 and CLOBBERED it → `handle_jobsubmit` replied `jobsubmit ok` for an uncommitted entry → a term-10 leader legally overwrote LSN 30 → committed-job-loss. Violated the Commit Durability invariant (reply-success ⇒ majority-durable). All 3 WALs were gapless/CRC-intact/full-payload (rules out prefix-trim + bug-847); lost job only on ctrl-1@LSN30 term9 vs term-10 NOOP on the majority._
-  - **Ownership = EXTERNAL spec ha-commit-durability (+ raft term/log-safety).** Out of raft-ha-core-safety's election-liveness scope; route there. See `.kiro/specs/ha-commit-durability/INVESTIGATION-failover-job-loss.md` and the [[project_failover_job_loss]] / bug-851 memories.
+  - _FIX (raft-apply-decoupling item #1 down payment, commit pending): extracted the WAL append + majority-commit barrier out of `persist_job_hook_fun`'s WAIT branch into an explicit `persist_submit_wal_and_commit(p, jlist)` (persist.c/.h); `handle_jobsubmit` (job.c) calls it directly and reads the commit rc DIRECTLY — rejects on <0 BEFORE firing the now notification-only AFTER hook (canonical commit→notify→reply). The commit verdict is no longer a clobber-prone hook return. Regression pin `test_job_hook_new_is_notify_only` + 4 relocated hook tests; persist_hooks 19/0, e2e_hooks 3/0, commit_barrier_reply 22/0, commit_durability 38/0._
-  - Observable completion: the 6-round chaotic soak verdict is committed_job_loss=0 AND committed_divergence=0 — the green 7.2 gate
+  - _PROVEN: 10-round chaotic soak `committed_job_loss` 5 → **0** (same election code, A/B isolates the durability fix)._
-  - _Requirements: 6.1, 6.2, 6.3_
+  - _Boundary: Commit durability (raft-apply-decoupling item #1). Full canonical inversion (async keystone + apply_committed_entry seam + facts + restart validation) = the follow-on ha spec._
-  - _Boundary: Commit durability / catch-up backfill (EXTERNAL)_
   - _Depends: 5.8_
 
+- [ ] 5.10 Residual soak FAILs after 5.9 (NOT job loss) — divergence lag + election-escalation tail
+  - _(a) committed_divergence>0 from the just-restarted node's catch-up LAG (walcompare `present=2`: entry on the 2 live nodes, not yet backfilled on the restarted one). Transient, settles; NOT lost data. Closed by entity-bearing snapshots / the full raft-apply-decoupling apply seam (EXTERNAL — ha-commit-durability / ha-convergence-recovery). (b) Election-escalation TAIL: under an adversarial overlap (kill inside a split-vote window) a round can still stall + ladder terms (observed round-4 term 6→25 before electing) despite 5.7's jitter — liveness held (elected next round) but degraded. Harden: cap/dampen term escalation or widen the jitter under repeated same-round failovers._
+  - Observable completion: a 10-round chaotic soak verdict is committed_divergence=0 with no per-round election stall — the green 7.2 gate
+  - _Requirements: 4.1, 4.3, 6.1, 6.2, 6.3_
+  - _Boundary: catch-up backfill (EXTERNAL) + election-liveness tail (5.7 family)_
+  - _Depends: 5.9_
+
 - [ ] 6. Integration: single-seam invariant end-to-end
 - [ ] 6.1 Verify leader→standby replication carries (term, payload) faithfully end-to-end
   - Integration check that the live leader replication AND the cross-term catch-up streaming path transmit each entry's full `(term, payload)` to the receiver: no `term==0` on the wire from a live leader (defensive), and — the real concern — no non-empty entry ever arrives with `PLEN=0`; the walAck catch-up continuation feeds only the seam ingest path

src/controller/job.c

@@ -590,27 +590,43 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
         }
     }
 
-    int hook_rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB,
+    /* Explicit raft step (raft-apply-decoupling item #1 / bug-1306): WAL-append
+     * + majority-commit barrier, reading the commit verdict DIRECTLY.  This used
+     * to run inside the AFTER_EVENT hook (persist_job_hook_fun), but xce_run_hooks
+     * returns only the LAST hook's rc, so a later notify subscriber (sched/
+     * apiserver) returning 0 CLOBBERED the -ETIMEDOUT — the client was acked for
+     * an entry that never reached majority, which a new-term leader then legally
+     * discarded (committed-job-loss).  The client must not be acknowledged until
+     * the job is durable on a majority. */
+    int commit_rc = persist_submit_wal_and_commit(&c->persist, &jlist);
+    if (commit_rc < 0) {
+        WARN("handle_jobsubmit: job not durable (rc=%d) — rejecting submission",
+             commit_rc);
+        xlist_iter(&jlist, (xlist_iter_fn_t)xlist_deref_job_item_, NULL);
+        xlist_fini(&jlist);
+        reply.status     = commit_rc;
+        reply.data_which = RequestReply_data_nodata;
+
+        xce_reply(ep, c->myhost, &reply);
+        return commit_rc;
+    }
+
+    /* Committed on a majority — fire the AFTER_EVENT hook as NOTIFICATION only
+     * (scheduler push, apiserver SSE), observing already-committed state.  Its
+     * return no longer gates durability: a notify-hook failure cannot un-commit
+     * an already-durable job, so it is logged but never fails the client reply. */
+    int notify_rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB,
                                   HOOK_AFTER_EVENT,
                                   HOOK_DETAIL_JOB_NEW, c, &jlist);
+    if (notify_rc != 0)
+        WARN("handle_jobsubmit: post-commit notification hook rc=%d "
+             "(job already durable; not failing the reply)", notify_rc);
 
     xlist_iter(&jlist, (xlist_iter_fn_t)xlist_deref_job_item_, NULL);
 
     xlist_fini(&jlist);
 
-    /* If persistence failed (replication timeout, I/O error), reject the
+    /* Reply success only AFTER WAL write + majority commit succeeded. */
-     * submission so the client knows the job was not made durable. */
-    if (hook_rc != 0) {
-        WARN("handle_jobsubmit: persistence hook failed (rc=%d) — rejecting job",
-             hook_rc);
-        reply.status     = hook_rc;
-        reply.data_which = RequestReply_data_nodata;
-
-        xce_reply(ep, c->myhost, &reply);
-        return hook_rc;
-    }
-
-    /* Reply only AFTER WAL write + majority replication succeeded. */
     reply.status     = 0;
     reply.data_which = RequestReply_data_jobid;
     reply.data.jobid = jobid;

src/controller/persist.c

@@ -2782,6 +2782,60 @@ static int wal_hook_append_classified(persist_t* p, uint32_t wal_type,
     return rc; /* local append rc only — facts have no commit timeout */
 }
 
+/* Explicit controller-side raft step for a job submission (raft-apply-decoupling
+ * item #1 / bug-1306): WAL-append each job in `jlist` and block on the
+ * majority-commit barrier, returning the commit verdict DIRECTLY to the caller
+ * (handle_jobsubmit) instead of through the xce_run_hooks fan-out (which returns
+ * only the last hook's rc and so clobbered this -ETIMEDOUT with a later notify
+ * hook's 0 — the committed-job-loss root cause).
+ *
+ * Returns 0 when every job's entry committed on a majority; the FIRST negative
+ * rc otherwise (-ETIMEDOUT: no majority within the deadline; -XCE_ERR_NOT_READY:
+ * admission cap; -ENOMEM: encode failure; or a local-write error).  The caller
+ * MUST reject the submission on a negative rc (never reply success) to preserve
+ * the Commit Durability invariant.  On -ETIMEDOUT / -XCE_ERR_NOT_READY the entry
+ * WAS appended locally and the LSN advanced, so j->submit_lsn is still tagged —
+ * the entry remains a pending, uncommitted tail a new-term leader may discard. */
+int persist_submit_wal_and_commit(persist_t* p, xlist_t* jlist)
+{
+    int rc = 0;
+
+    if (p == NULL || jlist == NULL)
+        return -EINVAL;
+
+    for (int i = 0; i < jlist->count && rc == 0; i++) {
+        job_t*   j   = (job_t*)xlist_get(jlist, i);
+        uint8_t* buf = NULL;
+        size_t   sz  = 0;
+        if (capnp_encode_job(&j->job, &buf, &sz) != 0) {
+            rc = -ENOMEM;
+            break;
+        }
+        /* ha-convergence-recovery (3.3): snapshot the LSN this submit entry
+         * will receive BEFORE the append.  persist_wal_async assigns
+         * hdr.lsn = current_lsn then increments, so the pre-append current_lsn
+         * is exactly the written entry's LSN. */
+        pthread_mutex_lock(&p->lock);
+        uint64_t submit_lsn = p->current_lsn;
+        pthread_mutex_unlock(&p->lock);
+
+        int sync_rc = 0;
+        rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, buf, (uint32_t)sz,
+                                     &sync_rc);
+        free(buf);
+
+        /* Tag the job with its submit entry's LSN whenever the LOCAL WAL write
+         * succeeded.  persist_wal_sync_append returns the persist_wal_async
+         * error only on local-write failure; on a replication-degraded outcome
+         * (0 / -ETIMEDOUT / -ENODEV / -XCE_ERR_NOT_READY) the entry was appended
+         * locally and the LSN advanced, so the egress tag is set. */
+        if (rc == 0 || rc == -ETIMEDOUT || rc == -XCE_ERR_NOT_READY)
+            j->submit_lsn = submit_lsn;
+    }
+
+    return rc;
+}
+
 /* P0 payload: just the detail value (4 bytes).
  * Full Cap'n Proto serialization is wired in Task 15. */
 
@@ -2803,50 +2857,18 @@ int persist_job_hook_fun(int type, int situation, int detail,
     mode = wal_classify(wal_type, detail);
 
     if (mode == WAL_MODE_WAIT) {
-        /* WAL_JOB_SUBMIT — the only WAIT type in the job hook; reference
+        /* WAL_JOB_SUBMIT durability (WAL append + majority-commit barrier) was
-         * implementation keeps the existing in-hook sync path.
+         * MOVED OUT of this hook into the explicit raft step
-         * AFTER_EVENT hook args: cluster_t* c, xlist_t* jlist */
+         * persist_submit_wal_and_commit(), called directly by handle_jobsubmit
-        (void)va_arg(ap, cluster_t*);
+         * (raft-apply-decoupling item #1 / bug-1306).  Rationale: xce_run_hooks
-        xlist_t* jlist = va_arg(ap, xlist_t*);
+         * returns only the LAST-registered hook's rc, so a later JOB AFTER_EVENT
-        int rc = 0;
+         * subscriber (sched/apiserver, returning 0) CLOBBERED this branch's
-        for (int i = 0; i < jlist->count && rc == 0; i++) {
+         * -ETIMEDOUT — handle_jobsubmit then replied `jobsubmit ok` for an
-            job_t*    j   = (job_t*)xlist_get(jlist, i);
+         * entry that never reached majority, and a new-term leader legally
-            uint8_t*  buf = NULL;
+         * discarded it (committed-job-loss).  The commit verdict must be read
-            size_t    sz  = 0;
+         * from the protocol step directly, never as a hook return.  This branch
-            if (capnp_encode_job(&j->job, &buf, &sz) != 0) {
+         * is now a no-op; the AFTER_EVENT hook chain is notification-only. */
-                rc = -ENOMEM;
+        return 0;
-                break;
-            }
-            /* persist_wal_sync_append returns 0 on successful local WAL write
-             * even when replication is degraded (no standbys → -ENODEV, timeout
-             * → -ETIMEDOUT).  Only abort the loop on a real disk-write failure
-             * (non-zero return).  Replication status is informational here. */
-            /* ha-convergence-recovery (3.3): snapshot the LSN this submit
-             * entry will receive BEFORE the append.  persist_wal_async assigns
-             * hdr.lsn = current_lsn then increments (persist.c:318), so the
-             * pre-append current_lsn is exactly the written entry's LSN — the
-             * same value persist_wal_sync_append derives as current_lsn - 1. */
-            pthread_mutex_lock(&p->lock);
-            uint64_t submit_lsn = p->current_lsn;
-            pthread_mutex_unlock(&p->lock);
-
-            int sync_rc = 0;
-            rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, buf, (uint32_t)sz,
-                                         &sync_rc);
-            free(buf);
-
-            /* Tag the job with its submit entry's LSN whenever the LOCAL WAL
-             * write succeeded.  persist_wal_sync_append returns the
-             * persist_wal_async error only on local-write failure; on a
-             * replication-degraded outcome (0 / -ETIMEDOUT / -ENODEV /
-             * -XCE_ERR_NOT_READY) the entry was appended locally and the LSN
-             * advanced, so the tag must be set (the egress predicate is
-             * independent of replication status). */
-            if (rc == 0 || rc == -ETIMEDOUT || rc == -XCE_ERR_NOT_READY) {
-                j->submit_lsn = submit_lsn;
-            }
-        }
-        return rc;
     }
 
     /* All other job events: payload = wal_jobstate_t {jobid, state}.

src/controller/testsuite/test_e2e_hooks.c

@@ -130,11 +130,11 @@ static void test_job_hook_writes_wal_before_lsn_advances(void)
 
     uint64_t lsn_before = p->current_lsn;
 
-    /* JOB_NEW args: (cluster_t*, xlist_t* jlist) — list must have ≥1 job */
+    /* Submit durability moved out of the JOB_NEW hook into the explicit raft
+     * step persist_submit_wal_and_commit (raft-apply-decoupling item #1 /
+     * bug-1306); the hook is now notification-only. */
     STUB_JOB_XLIST(jlist);
-    int rc = call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT,
+    int rc = persist_submit_wal_and_commit(p, &jlist);
-                           HOOK_DETAIL_JOB_NEW, p,
-                           (cluster_t*)NULL, &jlist);
     TEST_ASSERT_EQUAL_INT(0, rc);
 
     /* LSN must have advanced by exactly 1 */
@@ -197,8 +197,7 @@ static void test_hook_sequence_preserves_lsn_order(void)
     uint64_t lsn0 = p->current_lsn;
 
     STUB_JOB_XLIST(jlist);
-    call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_NEW, p,
+    persist_submit_wal_and_commit(p, &jlist);  /* JOB_NEW via explicit raft step */
-                  (cluster_t*)NULL, &jlist);
     uint64_t lsn1 = p->current_lsn;
 
     call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_START, p,

src/controller/testsuite/test_persist_hooks.c

@@ -106,11 +106,12 @@ static void test_job_hook_new_writes_wal_entry(void)
 
     uint64_t lsn_before = p->current_lsn;
 
-    /* JOB_NEW args: (cluster_t*, xlist_t* jlist) — list must have ≥1 job */
+    /* Submit durability moved OUT of the JOB_NEW hook into the explicit raft
+     * step persist_submit_wal_and_commit (raft-apply-decoupling item #1 /
+     * bug-1306) — the hook is now notification-only, so the WAL-append is
+     * asserted against the explicit step. */
     STUB_JOB_XLIST(jlist);
-    int rc = call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT,
+    int rc = persist_submit_wal_and_commit(p, &jlist);
-                           HOOK_DETAIL_JOB_NEW, p,
-                           (cluster_t*)NULL, &jlist);
     TEST_ASSERT_EQUAL_INT(0, rc);
     TEST_ASSERT_GREATER_THAN(lsn_before, p->current_lsn);
 
@@ -118,6 +119,33 @@ static void test_job_hook_new_writes_wal_entry(void)
     printf("test_job_hook_new_writes_wal_entry: PASS\n");
 }
 
+/* Regression pin (bug-1306 / raft-apply-decoupling item #1): the JOB_NEW hook
+ * must be NOTIFICATION-ONLY — it must NOT perform the WAL append / commit
+ * barrier. If durability lived in the hook, xce_run_hooks (which returns only
+ * the last hook's rc) could let a later notify subscriber's 0 clobber a commit
+ * -ETIMEDOUT, acking the client for an uncommitted entry. Asserting the hook
+ * appends NOTHING structurally forecloses that clobber. */
+static void test_job_hook_new_is_notify_only(void)
+{
+    persist_t* p = make_hook_test_persist();
+    TEST_ASSERT_NOT_NULL(p);
+
+    uint64_t lsn_before = p->current_lsn;
+
+    STUB_JOB_XLIST(jlist);
+    int rc = call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT,
+                           HOOK_DETAIL_JOB_NEW, p,
+                           (cluster_t*)NULL, &jlist);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, p->current_lsn,
+        "JOB_NEW hook must be notify-only — durability moved to "
+        "persist_submit_wal_and_commit (bug-1306); a hook that still appends "
+        "re-opens the xce_run_hooks last-rc clobber");
+
+    cleanup_hook_test_persist(p);
+    printf("test_job_hook_new_is_notify_only: PASS\n");
+}
+
 static void test_job_hook_start_writes_wal_entry(void)
 {
     persist_t* p = make_hook_test_persist();
@@ -406,10 +434,10 @@ static void test_job_hook_new_writes_multiple_entries_sequentially(void)
 
     uint64_t lsn_before = p->current_lsn;
 
-    /* Fire three hook calls — each must advance LSN */
+    /* JOB_NEW submit durability goes through the explicit raft step; START/EXIT
+     * facts still flow through the hook (RETAIN). All three advance LSN. */
     STUB_JOB_XLIST(jlist);
-    call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_NEW, p,
+    persist_submit_wal_and_commit(p, &jlist);
-                  (cluster_t*)NULL, &jlist);
     call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_START, p,
                   (cluster_t*)NULL, (void*)NULL, &g_stub_job);
     call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_EXIT, p,
@@ -430,6 +458,7 @@ void test_persist_hooks(void)
     printf("\n>> persist hook function tests\n\n");
 
     RUN_TEST(test_job_hook_new_writes_wal_entry);
+    RUN_TEST(test_job_hook_new_is_notify_only);
     RUN_TEST(test_job_hook_start_writes_wal_entry);
     RUN_TEST(test_job_hook_exit_writes_wal_entry);
     RUN_TEST(test_job_hook_fail_writes_wal_entry);

src/include/persist.h

@@ -9,6 +9,7 @@
 #include "config.h"
 #include "wal_types.h"  /* wal_entry_hdr_t, wal_entry_type — P2-4: shared with repl.h */
 #include "repl.h"       /* replicator_t — P2-4: embedded in persist_t */
+#include "xlist.h"      /* xlist_t — persist_submit_wal_and_commit job batch */
 
 /* Forward declaration */
 typedef struct cluster_s cluster_t;
@@ -605,6 +606,14 @@ void persist_log_entry_metadata(const wal_entry_hdr_t* hdr);
  * P0: writes a minimal 4-byte payload (detail value) to the WAL.
  * Full Cap'n Proto serialization is wired in Task 15.
  * ============================================================ */
+/* Explicit controller-side raft step for job submission (raft-apply-decoupling
+ * item #1 / bug-1306): WAL-append every job in `jlist` and block on the
+ * majority-commit barrier, returning the commit verdict DIRECTLY (0 = all
+ * committed on a majority; first negative rc otherwise — -ETIMEDOUT / etc.).
+ * handle_jobsubmit calls this instead of reading the clobber-prone hook return;
+ * a negative rc MUST reject the submission (never reply success). */
+int persist_submit_wal_and_commit(persist_t* p, xlist_t* jlist);
+
 int persist_job_hook_fun(int type, int situation, int detail,
                          void* cls, va_list ap);
 int persist_queue_hook_fun(int type, int situation, int detail,