feat(job-array-single-wal-record): make apply pump respect apply failures

persist_apply_pump_run() no longer discards apply_committed_entry()'s
return value. On a negative rc for any WAL type (not just the new
array type), the walk halts at that LSN — no last_applied_lsn advance,
no pending-apply hand-off — logs a loud error, and retries from the
same position on the next kick, mirroring the existing
apply_pump_read_entry failure branch's shape.

Investigated whether this newly turns the pre-existing job-fact
-ENOENT path into a permanent-wedge hazard: it does not, since
c->jobmgr never deletes entries and facts are only ever applied after
their causally-prior submit already succeeded, so a live -ENOENT there
can only mean genuine divergence — halting is the correct behavior.

Task 4.1 of job-array-single-wal-record (Requirement 9.4).

.kiro/specs/job-array-single-wal-record/tasks.md

@@ -82,7 +82,7 @@
   - _Boundary: Array Apply Expansion_
 
 - [ ] 4. Live Apply Resource Fence
-- [ ] 4.1 Make the apply pump respect apply failures
+- [x] 4.1 Make the apply pump respect apply failures
   - Stop discarding the array-expansion (and every other WAL type's) apply outcome; on a failure, halt the apply walk at that position without advancing past it, and log a loud, operator-visible error naming the stuck position
   - Observable: a simulated apply failure at a given position causes the walk to stop exactly there and not proceed to later entries
   - _Requirements: 9.4_
@@ -175,3 +175,4 @@
 - **Task 1.3**: `makeJobsub()`/`makeJobstartinfo()` now return `-ENODEV` (not a generic `-EINVAL`) when the interning table isn't initialized — if a later task's error handling branches on `job_init()`'s return value, `-ENODEV` is a distinct, checkable signal for "table not ready" vs. other construction failures.
 - **Task 3.3**: `apply_wal_entry()` (persist.c:2574) is dead code in production — confirmed via exhaustive grep (zero call sites outside its own declaration/definition/comments) and via `ha.c:8024`'s `ha_standby_advance_committed_and_apply` (dated 2026-07-10), whose own doc comment explicitly says standby entity materialization now goes through the SAME `apply_committed_entry()`/`persist_apply_pump_kick` pump the primary uses, not `apply_wal_entry()`. The persist.c:3722 comment claiming otherwise predates that refactor and was never updated (design.md corrected accordingly). Requirement 3.1/3.2/4.1 are fully delivered by task 3.2 alone. Task 3.3's code was kept (not discarded) as defensive parity with `WAL_JOB_SUBMIT`'s existing, also-dead case in the same switch — it is tested and correct, just not currently reachable in production. If `apply_wal_entry()` is ever revived or removed entirely in a future spec, this task's case should move or be deleted with it.
 - **Task 3.1**: `restore_job_array()`'s atomicity has one narrow, documented exception (design.md updated to match): the final bulk-insert-into-`c->jobmgr` pass can itself fail on allocation/growth exhaustion (`c->jobmgr` grows via `realloc()`; no capacity-reservation primitive exists anywhere in `kv.c`/`kvmgr.c`/`zarray.c` to pre-check this before mutating). On this specific failure, the function never silently claims success — it returns a negative errno — but an already-inserted prefix `[0, inserted)` can be left genuinely, permanently visible in `c->jobmgr` (no `key_del` exists to undo it), and that prefix is NEVER included in `out_new_jobs`, so `HOOK_DETAIL_JOB_NEW` never fires for it, not on this attempt or any later successful retry of the same LSN. **Tasks 3.2/4.x must account for this**: those elements are individually addressable (status/cancel/completion work normally) but permanently un-pushed to the scheduler/API server unless something later re-derives and re-notifies them — this is a real gap in the current design, not yet solved, and task 4.x's Live Apply Resource Fence design should explicitly decide whether/how to handle it (e.g. detecting this residual state and re-notifying on self-heal) rather than only handling the construction-phase failure case.
+- **Task 4.1**: `persist_apply_pump_run()` now halts the walk (no `last_applied_lsn` advance) on ANY negative return from `apply_committed_entry()`, not just the new array type — this is a general fix. Investigated whether this newly exposes a hazard for the pre-existing job-state fact entries (`WAL_JOB_START/FINISH/FAIL/UPDATE`), which have their own real `-ENOENT` branch when the target jobid isn't in `c->jobmgr`: confirmed NOT a hazard, because `c->jobmgr` (`xkvmgr_type`) has no `key_del` (jobs are never removed for the process lifetime, corroborated by `job.c:428,455,645`), and this task's own strict sequential-no-skip apply means a fact's LSN is only reached after its causally-prior submit already applied. So a live `-ENOENT` on a fact can now only mean genuine corruption/divergence — halting is correct, not a new footgun. Independent reviewer confirmed this via `test_term_safety_adapter.c:2039-2043`'s synthetic (non-production-representative) trigger. design.md's "Existing Architecture Analysis" table row overclaims ("every existing WAL type... returns 0 or positive in practice") — technically inaccurate re: the fact-ENOENT branch — left as-is since the persist.c code comment at the site already documents the exception correctly; low-severity, non-blocking.

src/controller/persist.c

@@ -3343,7 +3343,43 @@ int persist_apply_pump_run(persist_t* p)
          * ascending order, so a burst of commands and facts that committed
          * out of their original submission order still applies in strict
          * commit (LSN) order (Req 3.3, 7.3). */
-        (void)apply_committed_entry(p, wal_type, next, payload, len);
+        int apply_rc = apply_committed_entry(p, wal_type, next, payload, len);
+        if (apply_rc < 0) {
+            /* job-array-single-wal-record task 4.1 (design.md "Live Apply
+             * Resource Fence"): this return value was previously discarded
+             * -- (void)apply_committed_entry(...) -- with last_applied_lsn
+             * advancing unconditionally, silently and permanently losing
+             * any apply failure. A negative rc here is never a content-
+             * derived failure (Req 9.1-9.3's pre-commit validation already
+             * made those impossible post-commit); it is a node-local
+             * resource-exhaustion failure (e.g. OOM mid array expansion,
+             * Req 9.4) that must not be skipped. Mirror the
+             * apply_pump_read_entry failure branch immediately above: stop
+             * the walk exactly at this LSN, do not advance
+             * last_applied_lsn, and do not take this LSN's pending-apply
+             * hand-off (persist_take_pending_apply/on_applied) -- the entry
+             * never actually applied, so that hand-off must not fire for
+             * it. The entry stays committed and will be retried, from the
+             * same position, on the next kick. This is a GENERAL fix: every
+             * WAL type's apply outcome is now respected, not just
+             * WAL_JOB_ARRAY_SUBMIT's. Under NORMAL operation every other
+             * existing type's apply case returns 0 or positive, so this
+             * branch is a behavior no-op for them in practice; the
+             * fact-type cases' pre-existing "-ENOENT: referenced job not
+             * found" branch (persist.h's dispatch table) is one other real,
+             * pre-existing source of a negative return here, and now also
+             * correctly halts the walk here instead of being silently
+             * discarded, exactly as this fix intends. */
+            ERROR("persist_apply_pump_run: apply_committed_entry failed for "
+                  "committed entry lsn=%" PRIu64 " wal_type=%" PRIu32
+                  " (rc=%d) -- stopping this pass at this position; the "
+                  "entry stays committed and will be retried, from the same "
+                  "position, on the next kick",
+                  next, wal_type, apply_rc);
+            free(payload);
+            pthread_mutex_unlock(&p->apply_pump_lock);
+            return apply_rc;
+        }
         free(payload);
 
         atomic_store_explicit(&p->last_applied_lsn, next, memory_order_relaxed);

src/controller/testsuite/test_apply_pump.c

@@ -571,6 +571,202 @@ static void test_apply_pump_run_applies_mixed_submit_and_fact_entries(void)
           "PASS\n");
 }
 
+/* ── 2b. Task 4.1 (job-array-single-wal-record, design.md "Live Apply
+ * Resource Fence"): the walk must respect a negative apply outcome ─────── */
+
+/* Encodes a (base_jobid, array_size, spec) JobSubmit payload -- the exact
+ * WAL_JOB_ARRAY_SUBMIT payload shape, mirroring test_job_array_apply.c's
+ * taj_encode_job_submit. Caller frees *out. */
+static int tap_encode_job_submit(uint64_t base_jobid, uint64_t array_size,
+                                 job_submit_data_t* spec, uint8_t** out,
+                                 size_t* out_len)
+{
+    struct capn          c;
+    struct capn_segment* cs;
+    struct JobSubmit     cjs;
+    JobSubmit_ptr        jsp;
+    xce_job_submit_t     s;
+
+    memset(&s, 0, sizeof(s));
+    s.jobid      = base_jobid;
+    s.array_size = array_size;
+    s.submit     = spec;
+
+    capn_init_malloc(&c);
+    cs = capn_root(&c).seg;
+    encode_JobSubmit(cs, &cjs, &s);
+    jsp = new_JobSubmit(cs);
+    write_JobSubmit(&cjs, jsp);
+    capn_setp(capn_root(&c), 0, jsp.p);
+
+    *out_len = capn_size(&c);
+    *out     = malloc(*out_len);
+    if (!*out) {
+        capn_free(&c);
+        return -ENOMEM;
+    }
+    capn_write_mem(&c, *out, *out_len, 0);
+    capn_free(&c);
+    return 0;
+}
+
+/* Before this fix, persist_apply_pump_run() discarded
+ * apply_committed_entry()'s return value entirely
+ * ((void)apply_committed_entry(...)) and unconditionally advanced
+ * last_applied_lsn -- silently, permanently losing any apply failure
+ * (design.md "Live Apply Resource Fence", Req 9.4). This test forces a REAL
+ * negative return through the REAL WAL_JOB_ARRAY_SUBMIT dispatch path
+ * (task 3.1's scratch-build-phase fault-injection seam), rather than
+ * mocking apply_committed_entry() itself, and proves the walk now:
+ *   (a) returns negative -- not silent 0/success;
+ *   (b) does not advance last_applied_lsn past the failed position;
+ *   (c) genuinely halts -- a LATER, already-committed entry in the same
+ *       batch is NOT applied on this pass, proving this is a stop, not a
+ *       skip-and-continue;
+ *   (d) reaches the loud, operator-visible ERROR log branch: this is the
+ *       ONLY source of a negative return from persist_apply_pump_run()'s
+ *       WAL_JOB_ARRAY_SUBMIT dispatch in this scenario, and the ERROR()
+ *       call is unconditionally on that exact return path (see persist.c) --
+ *       confirmed structurally rather than by parsing log output, per this
+ *       suite's convention (no log-capture mechanism exists in this
+ *       fixture).
+ * A subsequent retry, with the fault cleared, then proves the walk resumes
+ * from the SAME stuck position (not a saved-forward offset) and completes
+ * both entries once unstuck -- mirroring the "entry stays committed and
+ * will be retried, from the same position, on the next kick" contract the
+ * pre-existing apply_pump_read_entry failure branch already established
+ * immediately above this one in the same function. */
+static void test_apply_pump_run_halts_at_failed_lsn_without_advancing(void)
+{
+    tap_setup();
+
+#ifdef XCE_TESTING
+    job_array_apply_test_clear_force_fail();
+#endif
+
+    const uint64_t     base_jobid = 93000;
+    const uint64_t     array_size = 3;
+    job_submit_data_t  spec;
+    memset(&spec, 0, sizeof(spec));
+    spec.queue    = (char*)"q";
+    spec.user     = (char*)"tester";
+    spec.mintasks = 1;
+    spec.maxtasks = 1;
+    spec.job_name = (char*)"array-fail-test";
+
+    uint8_t* array_payload = NULL;
+    size_t   array_len     = 0;
+    int      enc_rc = tap_encode_job_submit(base_jobid, array_size, &spec,
+                                            &array_payload, &array_len);
+
+    /* Fail at scratch-build index 1 (of 3) -- set BEFORE either entry below
+     * is appended, so no background fsync-driven kick (see file header) can
+     * ever observe an unfailed attempt at this LSN: the fault is
+     * deterministic and sticky until cleared, so every caller that reaches
+     * this LSN -- this test's own call below, or a racing background one --
+     * gets the identical negative rc without advancing anything. */
+#ifdef XCE_TESTING
+    job_array_apply_test_set_force_fail_at(1);
+#endif
+
+    int rc_array = persist_wal_async(&g_tap_c.persist, WAL_JOB_ARRAY_SUBMIT,
+                                     array_payload, (uint32_t)array_len);
+    free(array_payload);
+    uint64_t array_lsn = g_tap_c.persist.current_lsn - 1;
+
+    /* A second, later entry -- an ordinary single-job submit -- committed in
+     * the SAME batch as the array entry above. If the walk merely skipped
+     * the failed array record instead of halting at it, this job would end
+     * up materialized; a genuine halt must leave it untouched. */
+    memset(&g_tap_seed_submit, 0, sizeof(g_tap_seed_submit));
+    g_tap_seed_submit.queue = (char*)"q";
+    g_tap_seed_submit.user  = (char*)"tester";
+
+    xce_job_t jdef;
+    memset(&jdef, 0, sizeof(jdef));
+    jdef.jobid  = 94000;
+    jdef.state  = JOB_STAT_PEND;
+    jdef.submit = &g_tap_seed_submit;
+
+    uint8_t* later_payload = NULL;
+    size_t   later_len     = 0;
+    int      enc_rc2  = tap_encode_job(&jdef, &later_payload, &later_len);
+    int      rc_later = persist_wal_async(&g_tap_c.persist, WAL_JOB_SUBMIT,
+                                          later_payload, (uint32_t)later_len);
+    free(later_payload);
+    uint64_t later_lsn = g_tap_c.persist.current_lsn - 1;
+
+    uint64_t baseline_applied = persist_last_applied_lsn(&g_tap_c.persist);
+
+    /* White-box: mark both entries committed directly (this test targets
+     * the walk's own return-value handling, not commit-advance itself --
+     * that is test 1's job). */
+    atomic_store(&g_tap_c.persist.committed_lsn, later_lsn);
+
+    int walk_rc = persist_apply_pump_run(&g_tap_c.persist);
+
+    uint64_t last_applied_after_failure =
+        persist_last_applied_lsn(&g_tap_c.persist);
+    job_t* array_job0_after_failure = tap_find_job(base_jobid + 0);
+    job_t* later_job_after_failure  = tap_find_job(94000);
+
+    /* Clear the fault and retry: the walk must resume from the SAME stuck
+     * LSN (not have skipped it) and, this time, apply both entries. */
+#ifdef XCE_TESTING
+    job_array_apply_test_clear_force_fail();
+#endif
+    int walk_rc_retry = persist_apply_pump_run(&g_tap_c.persist);
+
+    uint64_t last_applied_after_retry =
+        persist_last_applied_lsn(&g_tap_c.persist);
+    job_t* array_job0_after_retry = tap_find_job(base_jobid + 0);
+    job_t* array_job1_after_retry = tap_find_job(base_jobid + 1);
+    job_t* array_job2_after_retry = tap_find_job(base_jobid + 2);
+    job_t* later_job_after_retry  = tap_find_job(94000);
+
+    tap_teardown();
+
+    TEST_ASSERT_EQUAL_INT(0, enc_rc);
+    TEST_ASSERT_EQUAL_INT(0, rc_array);
+    TEST_ASSERT_EQUAL_INT(0, enc_rc2);
+    TEST_ASSERT_EQUAL_INT(0, rc_later);
+    TEST_ASSERT_EQUAL_UINT64(array_lsn + 1, later_lsn);
+
+    TEST_ASSERT_TRUE_MESSAGE(walk_rc < 0,
+        "the walk must return a negative value -- not silent 0/success -- "
+        "when apply_committed_entry fails");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(baseline_applied,
+        last_applied_after_failure,
+        "last_applied_lsn must NOT advance past the failed position");
+    TEST_ASSERT_NULL_MESSAGE(array_job0_after_failure,
+        "no element of the failed array attempt may be visible -- "
+        "restore_job_array's own scratch-build atomicity plus this walk's "
+        "halt must together leave nothing materialized");
+    TEST_ASSERT_NULL_MESSAGE(later_job_after_failure,
+        "REGRESSION GUARD: the later, already-committed entry must NOT be "
+        "applied on this same pass -- the walk genuinely halted at the "
+        "failed position, it did not skip past it and continue");
+
+    TEST_ASSERT_TRUE_MESSAGE(walk_rc_retry >= 0,
+        "once the fault clears, the retry from the SAME stuck position "
+        "must succeed");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(later_lsn, last_applied_after_retry,
+        "the retry must resume from the same stuck LSN and reach the "
+        "watermark -- proving the earlier failure did not skip the entry");
+    TEST_ASSERT_NOT_NULL_MESSAGE(array_job0_after_retry,
+        "array element 0 must exist once the walk is unstuck");
+    TEST_ASSERT_NOT_NULL_MESSAGE(array_job1_after_retry,
+        "array element 1 (the previously-forced index) must exist once the "
+        "walk is unstuck");
+    TEST_ASSERT_NOT_NULL_MESSAGE(array_job2_after_retry,
+        "array element 2 must exist once the walk is unstuck");
+    TEST_ASSERT_NOT_NULL_MESSAGE(later_job_after_retry,
+        "the later entry must apply too, once the walk is unstuck");
+
+    printf("test_apply_pump_run_halts_at_failed_lsn_without_advancing: "
+          "PASS\n");
+}
+
 /* ── 3. Hazard 1 guard precision: job_ctrl's own live append must survive ── */
 
 static void test_apply_pump_guard_does_not_break_job_ctrl_live_append(void)
@@ -1295,6 +1491,7 @@ void test_apply_pump(void)
     UNITY_BEGIN();
     RUN_TEST(test_apply_pump_kick_applies_committed_burst_in_ascending_lsn_order);
     RUN_TEST(test_apply_pump_run_applies_mixed_submit_and_fact_entries);
+    RUN_TEST(test_apply_pump_run_halts_at_failed_lsn_without_advancing);
     RUN_TEST(test_apply_pump_guard_does_not_break_job_ctrl_live_append);
     RUN_TEST(test_apply_pump_fail_racing_finish_merges_against_current_state);
     RUN_TEST(test_apply_pump_late_pending_apply_registration_self_heals);