feat(libxce-failover): wrap 18 reply-bearing xcecli_* ops in failover loop under per-ctx lock; correct nop OpCategory (task 5.1)

.kiro/specs/libxce-failover/design.md

@@ -372,9 +372,9 @@ Three categories drive the wrapper generated for each of the ~33 ops:
 
 | Category | Ops (examples) | `xce_*` shape | Failover behavior |
 |----------|----------------|---------------|-------------------|
-| Reply-read | `joblist`, `reslist`, `workerlist`, `clusterconf`, `resval`, `hello`, `nop` | `*_req` + `cbuf_get` → status | Full failover; `is_write=false` |
+| Reply-read | `joblist`, `reslist`, `workerlist`, `clusterconf`, `resval`, `hello`, `jobinfo` | `*_req` + `cbuf_get` → status | Full failover; `is_write=false` |
 | Reply-write | `jobsubmit`, `jobctrl`, `queuectrl`, `hostctrl`, `update_resvals`, `runtask`, `join` | `*_req` + `cbuf_get` → status | Full failover; `is_write=true` (DUP-replay→success) |
-| Fire-and-forget | `notify_*` (jobstart/finish/fail/stop/chkpt), `heartbeat`, `wakeup` | `*_req` only, **no status** | Best-effort: reconnect on a known-dead connection only; **no** standby/redirect detection, **no** no-primary backoff. Redelivery deferred to `worker-rejoin-resync` |
+| Fire-and-forget | `notify_*` (jobstart/finish/fail/stop/chkpt), `heartbeat`, `wakeup`, `nop` | `*_req` only, **no status** | Best-effort: reconnect on a known-dead connection only; **no** standby/redirect detection, **no** no-primary backoff. Redelivery deferred to `worker-rejoin-resync` |
 
 - The exact per-op assignment is enumerated during task generation; `join` is reply-write and is the
   op that triggers `E_XCE_REDIRECT` at reconnect time (it carries the daemon's role/attrs).

.kiro/specs/libxce-failover/tasks.md

@@ -64,7 +64,7 @@
   - _Depends: 3.1, 3.2_
 
 - [ ] 5. Integration: wrap the client ops; controller redirect address
-- [ ] 5.1 Wrap reply-bearing client ops in the failover loop under the per-context lock
+- [x] 5.1 Wrap reply-bearing client ops in the failover loop under the per-context lock
   - Wrap each reply-bearing read and write op so it runs its single-shot call inside the decision loop, holding the per-context lock for the loop's duration
   - Tag write ops so the idempotent-replay rule applies; preserve every public signature and success/error convention
   - Done: existing callers compile unchanged; a reply-bearing op transparently completes against a new primary after a simulated failover; concurrent use of one context is serialized
@@ -107,3 +107,4 @@
 - 3.1: pure `xcecli_retry_decide` (no I/O — classify+budget+jittered-backoff calc, D6 exhaustion: no-primary→-E_XCE_NOT_READY, transport→keep last rc) + thin `xcecli_retry_next` (decide + RETRY_BACKOFF nanosleep + xce_failover_rotate + tlog_info/tlog_warn). Logger = utils.h tlog_* macros. State-reset/drain deferred to 3.2; rejoin hook to 4.1.
 - 3.2: `xcecli_ctx_recover` (socket-free): flips ac->ctx.state ERR→RUN ONLY, drains stale wbuf via non-blocking cbuf_tryget (-EAGAIN on empty); wired into xcecli_retry_next after successful rotate, before return true. THIS is the -EINVAL bug fix. Added missing prototype `extern int cbuf_tryget(cbuf_t*,void*)` to mylock.h (function already defined mylock.c:166, just undeclared).
 - 4.1: public `xcecli_reconnect_cb` typedef + `xcecli_set_reconnect_handler` (client.h/client.c, stores cb/arg in failover); socket-free `xcecli_invoke_rejoin` wired into xcecli_retry_next AFTER rotate+recover, BEFORE resume. hook<0 = failed attempt → bounded tail-recursion (consumes attempt budget, capped at max_retries). typedef MOVED from client_internal.h to client.h. API additive.
+- 5.1: `XCECLI_FAILOVER_CALL(ac,is_write,EXPR)` macro (client_internal.h) = lock(failover.lock) → retry init → do{rc=EXPR}while(xcecli_retry_next) → unlock → yield rc. Wrapped 18 reply-bearing ops; writes=true for jobsubmit/jobdispatch/jobctrl/queuectrl/hostctrl/update_resvals/update_jusage/runtask/join/set_pend_reason/clear_reason, reads=false otherwise. R1.3: zero signature changes. DESIGN CORRECTION: `xce_nop` is fire-and-forget (no cbuf_get, xceapi.c:30) — moved nop reply-read→fire-and-forget in design OpCategory; nop wraps in 5.2 not 5.1.

src/lib/client.c

@@ -199,7 +199,8 @@ int xcecli_hello(xcecli_ctx_t c, char* msg)
         return -EINVAL;
     }
 
-    return xce_hello(&(ctx->ctx), msg);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
+                                xce_hello(&(ctx->ctx), msg));
 }
 
 int xcecli_nop(xcecli_ctx_t c)
@@ -243,7 +244,8 @@ int xcecli_reslist(xcecli_ctx_t c, xce_resource_t*** res)
         return -EINVAL;
     }
 
-    return xce_reslist(&(ctx->ctx), res);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
+                                xce_reslist(&(ctx->ctx), res));
 }
 
 int xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs)
@@ -253,7 +255,8 @@ int xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs)
         return -EINVAL;
     }
 
-    return xce_join(&(ctx->ctx), role, attrs);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                                xce_join(&(ctx->ctx), role, attrs));
 }
 
 int xcecli_workerlist(xcecli_ctx_t c, xce_worker_t*** workers)
@@ -263,7 +266,8 @@ int xcecli_workerlist(xcecli_ctx_t c, xce_worker_t*** workers)
         return -EINVAL;
     }
 
-    return xce_workerlist(&(ctx->ctx), workers);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
+                                xce_workerlist(&(ctx->ctx), workers));
 }
 
 int xcecli_jobdispatch(xcecli_ctx_t c, xce_job_dispatch_t* info)
@@ -274,7 +278,8 @@ int xcecli_jobdispatch(xcecli_ctx_t c, xce_job_dispatch_t* info)
         return -EINVAL;
     }
 
-    return xce_jobdispatch(&(ctx->ctx), info);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                                xce_jobdispatch(&(ctx->ctx), info));
 }
 
 int xcecli_jobsubmit(xcecli_ctx_t c, xce_job_submit_t* submit, uint64_t* jobid)
@@ -285,7 +290,8 @@ int xcecli_jobsubmit(xcecli_ctx_t c, xce_job_submit_t* submit, uint64_t* jobid)
         return -EINVAL;
     }
 
-    return xce_jobsubmit(&(ctx->ctx), submit, jobid);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                                xce_jobsubmit(&(ctx->ctx), submit, jobid));
 }
 
 int xcecli_joblist(xcecli_ctx_t c, xce_job_t*** jobs)
@@ -296,7 +302,8 @@ int xcecli_joblist(xcecli_ctx_t c, xce_job_t*** jobs)
         return -EINVAL;
     }
 
-    return xce_joblist(&(ctx->ctx), jobs);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
+                                xce_joblist(&(ctx->ctx), jobs));
 }
 
 int xcecli_clusterconf(xcecli_ctx_t c, xce_clusterconf_t** conf)
@@ -307,7 +314,8 @@ int xcecli_clusterconf(xcecli_ctx_t c, xce_clusterconf_t** conf)
         return -EINVAL;
     }
 
-    return xce_clusterconf(&(ctx->ctx), conf);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
+                                xce_clusterconf(&(ctx->ctx), conf));
 }
 
 int xcecli_notify_jobfinish(xcecli_ctx_t c, xce_jobfinish_t* jfinish)
@@ -342,7 +350,8 @@ int xcecli_update_resvals(xcecli_ctx_t c, int nres, xce_resval_t** vals)
         return -EINVAL;
     }
 
-    rc = xce_update_resvals(&(ctx->ctx), nres, vals);
+    rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                              xce_update_resvals(&(ctx->ctx), nres, vals));
 
     return rc;
 }
@@ -358,8 +367,10 @@ int xcecli_update_jusage(xcecli_ctx_t c, uint64_t jobid, int npids, int* pids,
         return -EINVAL;
     }
 
-    rc = xce_update_jusage(&(ctx->ctx), jobid, npids, pids, npgids, pgids,
-                           nusage, usages);
+    rc = XCECLI_FAILOVER_CALL(
+        ctx, /*is_write=*/true,
+        xce_update_jusage(&(ctx->ctx), jobid, npids, pids, npgids, pgids, nusage,
+                          usages));
 
     return rc;
 }
@@ -376,7 +387,8 @@ int xcecli_resval(xcecli_ctx_t c, char* host, char* res, xce_resval_t*** val)
         return -EINVAL;
     }
 
-    return xce_resval(&(ctx->ctx), host, res, val);
+    return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
+                                xce_resval(&(ctx->ctx), host, res, val));
 }
 
 int xcecli_notify(xcecli_ctx_t c, xce_notify_data_t* data)
@@ -441,7 +453,8 @@ int xcecli_jobctrl(xcecli_ctx_t c, xce_jobctrl_t* ctrlinfo)
         return -EINVAL;
     }
 
-    rc = xce_jobctrl(&(ctx->ctx), ctrlinfo);
+    rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                              xce_jobctrl(&(ctx->ctx), ctrlinfo));
 
     return rc;
 }
@@ -535,7 +548,8 @@ int xcecli_queuectrl(xcecli_ctx_t c, xce_queuectrl_t* ctrlinfo)
         return -EINVAL;
     }
 
-    rc = xce_queuectrl(&(ctx->ctx), ctrlinfo);
+    rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                              xce_queuectrl(&(ctx->ctx), ctrlinfo));
 
     return rc;
 }
@@ -549,7 +563,8 @@ int xcecli_hostctrl(xcecli_ctx_t c, xce_hostctrl_t* ctrlinfo)
         return -EINVAL;
     }
 
-    rc = xce_hostctrl(&(ctx->ctx), ctrlinfo);
+    rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                              xce_hostctrl(&(ctx->ctx), ctrlinfo));
     return rc;
 }
 
@@ -566,7 +581,8 @@ int xcecli_runtask(xcecli_ctx_t c, xce_taskrun_t* info)
                            * controller to trace back
                            * client
                            */
-    rc = xce_runtask(&(ctx->ctx), info);
+    rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                              xce_runtask(&(ctx->ctx), info));
 
     return rc;
 }
@@ -658,7 +674,8 @@ int xcecli_set_pend_reason(xcecli_ctx_t c, xce_preason_set_t* reasons)
         return -EINVAL;
     }
 
-    rc = xce_set_pend_reason(&(ctx->ctx), reasons);
+    rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                              xce_set_pend_reason(&(ctx->ctx), reasons));
     return rc;
 }
 
@@ -681,7 +698,8 @@ int xcecli_clear_reason(xcecli_ctx_t c, int njobs, uint64_t* jobs)
         return 0;
     }
 
-    rc = xce_clear_reason(&(ctx->ctx), &reason);
+    rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
+                              xce_clear_reason(&(ctx->ctx), &reason));
     return rc;
 }
 
@@ -694,6 +712,7 @@ int xcecli_jobinfo(xcecli_ctx_t c, uint64_t jobid, xce_job_t** job)
         return -EINVAL;
     }
 
-    rc = xce_jobinfo(&(ctx->ctx), jobid, job);
+    rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
+                              xce_jobinfo(&(ctx->ctx), jobid, job));
     return rc;
 }

src/lib/client_internal.h

@@ -136,6 +136,41 @@ bool xcecli_retry_decide(api_ctx_t* ac, int* rc, bool is_write,
 bool xcecli_retry_next(api_ctx_t* ac, int* rc, bool is_write,
                        xcecli_retry_t* st);
 
+/*
+ * XCECLI_FAILOVER_CALL(ac, is_write, EXPR) — the shared XcecliRetryLoop wrapper
+ * (task 5.1). Turns a one-shot reply-bearing xce_* call into a transparent
+ * failover loop, holding the per-ctx lock (ac->failover.lock) for the FULL loop
+ * duration so concurrent ops on a shared ctx are serialised (CI2): the
+ * single-slot wbuf already implies one in-flight request per ctx; the lock makes
+ * that explicit and covers the reconnect/rotate window.
+ *
+ *   ac       — api_ctx_t* (already NULL-checked by the caller's guard)
+ *   is_write — true for mutating ops (DUP-on-replay → success, 5.2), false reads
+ *   EXPR     — the unchanged one-shot xce_* call, evaluated each iteration; its
+ *              rc is what the loop classifies and ultimately returns
+ *
+ * Evaluates to the final rc (>=0 success; -E_XCE_NOT_READY on no-primary budget
+ * exhaustion; terminal errors surfaced as-is). Used as the RHS of `return`.
+ *
+ * Implemented as a GCC/clang statement-expression so it yields a value and keeps
+ * each public xcecli_* signature/return convention intact (R1.3). The lock is
+ * released on every exit path because the do/while always falls through to the
+ * unlock before the result is produced.
+ */
+#define XCECLI_FAILOVER_CALL(ac, is_write, EXPR)                               \
+    __extension__({                                                            \
+        api_ctx_t*     _fo_ac = (ac);                                          \
+        xcecli_retry_t _fo_st = XCECLI_RETRY_INIT(_fo_ac);                     \
+        int            _fo_rc;                                                 \
+        mylock_lock(&_fo_ac->failover.lock);                                   \
+        do {                                                                   \
+            _fo_rc = (EXPR);                                                   \
+        } while (                                                              \
+            xcecli_retry_next(_fo_ac, &_fo_rc, (is_write), &_fo_st));          \
+        mylock_unlock(&_fo_ac->failover.lock);                                 \
+        _fo_rc;                                                                \
+    })
+
 /*
  * xcecli_ctx_recover (task 3.2) — repair the inner xce_ctx_t after a successful
  * reconnect so the next xce_* attempt can proceed. Socket-free (no I/O, no

src/lib/testsuite/test_libxce_retry.c

@@ -32,6 +32,7 @@
 
 #include "client_internal.h"  /* api_ctx_t, xce_failover_state_t */
 #include "failover_harness.h"
+#include "mylock.h"
 #include "request.h"
 #include "xce/client_failover.h"
 #include "xce/errnox.h"
@@ -1272,6 +1273,128 @@ static void test_invoke_rejoin_null_ctx_safe(void)
     printf("test_invoke_rejoin_null_ctx_safe: PASS\n");
 }
 
+/* ════════════════════════════════════════════════════════════════════════
+ * Task 5.1 — XcecliRetryLoop wrapping + per-ctx lock (CI2 concurrency).
+ *
+ * Each reply-bearing xcecli_* op wraps its one-shot xce_* call in the failover
+ * do/while loop AND holds ac->failover.lock for the loop's full duration so
+ * concurrent ops on ONE shared ctx are serialised (the single-slot wbuf already
+ * implies one in-flight request per ctx; the lock makes that explicit and covers
+ * the reconnect/rotate window). Full failover behaviour against a live primary
+ * is validated by E2E task 6.2; here we prove the serialisation contract.
+ *
+ * The XCECLI_FAILOVER_CALL macro (client_internal.h) is the shared wrapper that
+ * every op body uses: lock → init retry carrier → do/while(xcecli_retry_next) →
+ * unlock → return rc. We drive that exact macro here with a trivial EXPR that
+ * mutates shared state under the held lock, so N threads contending on one ctx's
+ * failover.lock must not lose updates (mutual exclusion), and a non-failing EXPR
+ * runs exactly once (the success path returns immediately).
+ *
+ * Requirements: 1.1, 1.3, 2.4, 5.1, 5.3, and design "FailoverState → Concurrency
+ * (CI2)".
+ * ════════════════════════════════════════════════════════════════════════ */
+
+#define FO_LOCK_NTHREADS 32
+#define FO_LOCK_ITERS    2000
+
+typedef struct {
+    api_ctx_t *ac;          /* the ONE shared ctx all threads contend on        */
+    long       counter;     /* mutated under failover.lock via the wrapper macro */
+    int        max_inflight;/* peak concurrent in-section (must stay 1)          */
+    int        inflight;    /* current in-section count                          */
+} fo_lock_ctx_t;
+
+/* A trivial reply-bearing EXPR that always "succeeds" (rc==0) but, as a side
+ * effect, performs a non-atomic read-modify-write on the shared counter and
+ * tracks the peak in-section concurrency. If the wrapper macro did NOT hold
+ * ac->failover.lock for the section, concurrent threads would interleave the
+ * RMW (lost updates) and/or observe inflight > 1. */
+static int fo_lock_body(fo_lock_ctx_t *c)
+{
+    c->inflight++;
+    if (c->inflight > c->max_inflight)
+        c->max_inflight = c->inflight;
+
+    /* non-atomic RMW: a data race here would lose increments */
+    long v = c->counter;
+    for (volatile int s = 0; s < 50; s++) {
+    } /* widen the race window */
+    c->counter = v + 1;
+
+    c->inflight--;
+    return 0; /* success → wrapper returns immediately, no retry */
+}
+
+/* The per-thread worker drives FO_LOCK_ITERS wrapped "ops" on the SHARED ctx.
+ * Each op uses the real XCECLI_FAILOVER_CALL wrapper, so it acquires/releases
+ * ac->failover.lock exactly as a production xcecli_* op does. */
+static void fo_lock_worker(void *ctx, int idx)
+{
+    (void)idx;
+    fo_lock_ctx_t *c  = (fo_lock_ctx_t *)ctx;
+    api_ctx_t     *ac = c->ac;
+
+    for (int i = 0; i < FO_LOCK_ITERS; i++) {
+        int rc = XCECLI_FAILOVER_CALL(ac, /*is_write=*/false, fo_lock_body(c));
+        TEST_ASSERT_EQUAL_INT(0, rc);
+    }
+}
+
+/* Req 5.1, 2.4 (CI2): N threads sharing ONE ctx are serialised by the per-ctx
+ * failover lock the wrapper holds — no lost updates and peak in-section
+ * concurrency stays 1. The total counter equals N * ITERS exactly. */
+static void test_failover_call_serialises_shared_ctx(void)
+{
+    xce_client_config_t conf;
+    api_ctx_t           ac;
+    fo_make_client_masters(&conf);
+    fo_make_api_ctx(&ac, &conf);
+
+    /* the wrapper takes ac->failover.lock; init it (xcecli_start would in prod) */
+    TEST_ASSERT_EQUAL_INT(0, mylock_init(&ac.failover.lock));
+    ac.failover.retry = XCE_RETRY_DEFAULTS;
+
+    fo_lock_ctx_t lc;
+    memset(&lc, 0, sizeof(lc));
+    lc.ac = &ac;
+
+    int rc = fo_run_threads(FO_LOCK_NTHREADS, fo_lock_worker, &lc);
+    mylock_fini(&ac.failover.lock);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    /* mutual exclusion held: every increment landed, none lost */
+    TEST_ASSERT_EQUAL_INT((long)FO_LOCK_NTHREADS * FO_LOCK_ITERS, lc.counter);
+    /* the critical section was never entered by two threads at once */
+    TEST_ASSERT_EQUAL_INT(1, lc.max_inflight);
+
+    printf("test_failover_call_serialises_shared_ctx: PASS\n");
+}
+
+/* Req 1.1, 5.3: the wrapper returns the op's success rc on the first try (no
+ * spurious retry) for a status-0 EXPR, and the per-op retry carrier starts at
+ * attempt 0 — i.e. a success path does not consume the retry budget. */
+static void test_failover_call_success_runs_once(void)
+{
+    xce_client_config_t conf;
+    api_ctx_t           ac;
+    fo_make_client_masters(&conf);
+    fo_make_api_ctx(&ac, &conf);
+    TEST_ASSERT_EQUAL_INT(0, mylock_init(&ac.failover.lock));
+    ac.failover.retry = XCE_RETRY_DEFAULTS;
+
+    fo_lock_ctx_t lc;
+    memset(&lc, 0, sizeof(lc));
+    lc.ac = &ac;
+
+    int rc = XCECLI_FAILOVER_CALL(&ac, /*is_write=*/false, fo_lock_body(&lc));
+    mylock_fini(&ac.failover.lock);
+
+    TEST_ASSERT_EQUAL_INT(0, rc);          /* success surfaced verbatim         */
+    TEST_ASSERT_EQUAL_INT(1, lc.counter);  /* body ran exactly once, no replay  */
+
+    printf("test_failover_call_success_runs_once: PASS\n");
+}
+
 /* ── Suite entry point ──────────────────────────────────────────────── */
 
 void test_libxce_retry(void)
@@ -1315,5 +1438,7 @@ void test_libxce_retry(void)
     RUN_TEST(test_invoke_rejoin_no_hook_is_noop);
     RUN_TEST(test_set_reconnect_handler_null_safe_and_clears);
     RUN_TEST(test_invoke_rejoin_null_ctx_safe);
+    RUN_TEST(test_failover_call_serialises_shared_ctx);
+    RUN_TEST(test_failover_call_success_runs_once);
     UNITY_END();
 }