test(raft-adapter-conformance): batch payload persist-failure propagation regression lock

Adds test_batch_payload_persist_failure_propagates_to_whole_batch,
proving persist_submit_job_payloads_async resolves every payload in a
3-entry batch with the same first-failure rc when the first payload's
append fails, rather than letting later payloads independently
succeed (Req 5.3).

Fault injection: chmod the WAL leaf directory to 0500 to force
open_new_segment's EACCES (surfaced as -EIO), restored the instant the
first failure is observed -- a transient fault, not a permanent one,
so the test genuinely distinguishes propagation from independent
per-payload failure. No existing fault hook was reachable from this
code path; this is a real, previously-uncovered gap.

1066 -> 1067 tests, 0 failures. Task 5 (regression-lock tests for
already-shipped/already-conformant behavior) now complete -- Core
phase of raft-adapter-conformance finished (tasks 1-5).

.kiro/specs/raft-adapter-conformance/tasks.md

@@ -70,7 +70,7 @@
   - _Requirements: 2.1_
   - _Boundary: test_ha_core_safety.c_
 
-- [ ] 5.2 (P) Test that one failing payload in a submitted batch propagates to the whole batch
+- [x] 5.2 (P) Test that one failing payload in a submitted batch propagates to the whole batch
   - Add a test asserting `persist_submit_job_payloads_async` resolves every payload in a batch with the first failure's outcome when one payload fails to persist, rather than completing a subset
   - Observable: new test in `test_commit_barrier_reply.c` fails if a future change lets a batch partially succeed
   - _Requirements: 5.3_

src/controller/testsuite/test_commit_barrier_reply.c

@@ -66,6 +66,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/stat.h>
 #include <time.h>
 #include <unistd.h>
 
@@ -3288,6 +3289,179 @@ static void test_fact_state_unchanged_throughout_quorum_loss_outage(void)
           "PASS\n");
 }
 
+/* ── Batch failure propagation (raft-adapter-conformance task 5.2, Req 5.3)
+ * ──────────────────────────────────────────────────────────────────────
+ *
+ * persist_submit_job_payloads_async (persist.c:6135-6209) walks an xlist_t
+ * of xce_job_t* payloads and tracks a single `first_rc` across the loop:
+ * the FIRST payload whose append_job_wal_entry_local call fails sets
+ * first_rc and resolves that payload's completion callback (fn) with the
+ * failing rc (persist.c:6161-6166); every payload processed AFTER that
+ * point takes the `if (first_rc != 0)` branch and resolves with the SAME
+ * rc WITHOUT attempting its own append (persist.c:6150-6158) -- so no
+ * payload in a batch that has already failed can silently complete on
+ * its own.
+ *
+ * No existing fixture in this file or in test_commit_barrier_rollout.c
+ * engineers append_job_wal_entry_local/persist_wal_async itself to fail
+ * (the two existing XCE_TESTING fault hooks -- test_force_fsync_fail and
+ * test_force_job_encode_fail_at -- only guard the flusher's fdatasync and
+ * the SEPARATE collect_job_frames snapshot-encode loop, neither of which
+ * this call path reaches). This test instead forces a real filesystem
+ * failure: persist_init() has already mkdir_p'd persist_t's wal_path
+ * directory by the time it returns, so revoking write permission on that
+ * directory BEFORE the first append makes open_new_segment's
+ * open(..., O_CREAT) fail with EACCES on the very first payload --
+ * persist_wal_async surfaces this as -EIO (persist.c:988-992) before
+ * current_lsn is ever incremented. Making the FIRST payload the failing
+ * one exercises both branches above: payload 0 hits the original-failure
+ * branch, payloads 1 and 2 hit the propagation branch.
+ *
+ * A bare (cluster == NULL) persist_t is used, with no commit engine wired
+ * (p->repl.ce stays NULL) and persist_start() never called -- every fn()
+ * resolution in persist_submit_job_payloads_async's success path is then
+ * synchronous too (persist.c:6178-6181), so the whole batch resolves
+ * within the single call under test, matching the same bare-persist_t,
+ * no-commit-engine technique test_commit_durability.c's dur_flag_persist
+ * already uses for a persist_init-only fixture (persist_fini alone tears
+ * it down -- persist_start/persist_stop were never paired).
+ *
+ * The injected fault is made deliberately TRANSIENT, not permanent: the
+ * notify callback restores write permission on the WAL directory the
+ * instant it observes payload 0's failure, before payload 1 is ever
+ * processed. This is what makes the test actually exercise the
+ * first_rc-propagation branch rather than merely re-observing the same
+ * permanently-broken directory three times over: if a future change
+ * deleted the `if (first_rc != 0) { fn(...); continue; }` short-circuit,
+ * payload 1 would attempt its OWN append against a directory that is
+ * writable again by then, and that attempt would SUCCEED -- turning this
+ * test's per-payload rc assertions below into an observable failure
+ * instead of a vacuously-identical result. */
+
+typedef struct {
+    int      count;
+    uint64_t lsns[4];
+    int      rcs[4];
+} cbrp_bf_notify_obs_t;
+
+/* Set by the test right after persist_init(), before the fault-injecting
+ * chmod(); read back by cbrp_bf_notify_fn to restore write permission the
+ * moment the first (and, on correct code, only genuinely-attempted)
+ * payload's failure is observed. */
+static char g_cbrp_bf_wal_path[512];
+
+static void cbrp_bf_notify_fn(void *ctx, uint64_t lsn, int rc)
+{
+    cbrp_bf_notify_obs_t *obs = (cbrp_bf_notify_obs_t *)ctx;
+    if (obs->count == 0)
+        chmod(g_cbrp_bf_wal_path, 0700);
+    if (obs->count < 4) {
+        obs->lsns[obs->count] = lsn;
+        obs->rcs[obs->count]  = rc;
+    }
+    obs->count++;
+}
+
+static job_submit_data_t g_cbrp_bf_submit;
+
+static void test_batch_payload_persist_failure_propagates_to_whole_batch(void)
+{
+    persist_t             p;
+    persist_config_t      cfg;
+    char                  tmpdir[256];
+    xlist_t               plist;
+    uint64_t              out_lsns[3] = { 0, 0, 0 };
+    cbrp_bf_notify_obs_t  obs;
+    int                   rc;
+    int                   i;
+
+    memset(&obs, 0, sizeof(obs));
+    memset(&g_cbrp_bf_submit, 0, sizeof(g_cbrp_bf_submit));
+    g_cbrp_bf_submit.queue      = (char *)"normal";
+    g_cbrp_bf_submit.user       = (char *)"tester";
+    g_cbrp_bf_submit.mintasks   = 1;
+    g_cbrp_bf_submit.maxtasks   = 1;
+    g_cbrp_bf_submit.start_info = NULL;
+    g_cbrp_bf_submit.job_name   = (char *)"batch-fail-job";
+
+    memset(&p, 0, sizeof(p));
+    cfg = persist_config_defaults();
+    cfg.wal_sync_ms = 5000;
+    snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_cbrpbf_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
+    persist_set_workdir(&p, tmpdir, "ctrl-cbrpbf");
+    /* cluster == NULL: append_job_wal_entry_local routes straight through
+     * persist_wal_async (no raft library, no commit engine involved). */
+    TEST_ASSERT_EQUAL_INT(0, persist_init(&p, &cfg, NULL));
+
+    /* persist_init already created p.wal_path. Revoke write permission on
+     * it now, before any append is attempted, so the first payload's
+     * open_new_segment() call fails deterministically. cbrp_bf_notify_fn
+     * restores it the instant payload 0's failure is observed (see the
+     * comment above), making the fault transient rather than permanent. */
+    snprintf(g_cbrp_bf_wal_path, sizeof(g_cbrp_bf_wal_path), "%s", p.wal_path);
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, chmod(p.wal_path, 0500),
+        "test precondition: must be able to revoke write permission on "
+        "the WAL directory");
+
+    TEST_ASSERT_EQUAL_INT(0, xlist_init(&plist, 3));
+    for (i = 0; i < 3; i++) {
+        xce_job_t *jp = calloc(1, sizeof(*jp));
+        TEST_ASSERT_NOT_NULL(jp);
+        jp->submit     = &g_cbrp_bf_submit;
+        jp->jobid      = 70000 + (uint64_t)i;
+        jp->sub_time   = time(NULL);
+        jp->queue_time = jp->sub_time;
+        jp->state      = JOB_STAT_PEND;
+        TEST_ASSERT_TRUE(xlist_store(&plist, jp) >= 0);
+    }
+
+    rc = persist_submit_job_payloads_async(&p, &plist, out_lsns,
+                                           cbrp_bf_notify_fn, &obs);
+
+    /* Defensive fallback: cbrp_bf_notify_fn already restores write
+     * permission once the first payload's failure is observed (the
+     * normal case), but make sure the directory is writable before any
+     * further filesystem access (including this test's own cleanup)
+     * regardless of how fn was actually invoked. */
+    chmod(p.wal_path, 0700);
+
+    TEST_ASSERT_EQUAL_INT_MESSAGE(-EIO, rc,
+        "the forced WAL-directory permission failure must surface as the "
+        "same -EIO persist_wal_async returns on an open_new_segment "
+        "failure (Req 5.1)");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(3, obs.count,
+        "every payload in the batch must resolve its completion callback "
+        "exactly once, even though none of them ever durably persisted");
+    for (i = 0; i < 3; i++) {
+        TEST_ASSERT_EQUAL_INT_MESSAGE(rc, obs.rcs[i],
+            "every payload -- not just the one whose append actually "
+            "failed -- must resolve with the SAME first-failure rc (Req "
+            "5.3): a future change that lets a later payload silently "
+            "succeed instead must fail this assertion");
+        TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, obs.lsns[i],
+            "a failed payload must never be resolved with a real "
+            "(non-zero) lsn");
+    }
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, p.current_lsn,
+        "current_lsn must not advance for any payload in a batch whose "
+        "very first append already failed (Req 5.2)");
+
+    for (i = 0; i < plist.count; i++) {
+        xce_job_t *jp = (xce_job_t *)xlist_get(&plist, i);
+        free(jp);
+    }
+    xlist_fini(&plist);
+    persist_fini(&p);
+    {
+        char cmd[300];
+        snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
+        system(cmd);
+    }
+    printf("test_batch_payload_persist_failure_propagates_to_whole_batch: "
+          "PASS\n");
+}
+
 /* ── Test group entry point ──────────────────────────────────────────── */
 
 void test_commit_barrier_reply(void)
@@ -3356,4 +3530,7 @@ void test_commit_barrier_reply(void)
     /* Durable retention held for the outage's sustained duration under
      * genuine quorum loss (task 4.2, Req 6.1, 6.2) */
     RUN_TEST(test_fact_state_unchanged_throughout_quorum_loss_outage);
+    /* Batch failure propagation regression-lock (raft-adapter-conformance
+     * task 5.2, Req 5.3) */
+    RUN_TEST(test_batch_payload_persist_failure_propagates_to_whole_batch);
 }