diff --git a/.kiro/specs/libxce-failover/design.md b/.kiro/specs/libxce-failover/design.md index d4ccb52..97b8054 100644 --- a/.kiro/specs/libxce-failover/design.md +++ b/.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). diff --git a/.kiro/specs/libxce-failover/tasks.md b/.kiro/specs/libxce-failover/tasks.md index 08f351e..4feb205 100644 --- a/.kiro/specs/libxce-failover/tasks.md +++ b/.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. diff --git a/src/lib/client.c b/src/lib/client.c index b0449ea..ed8713d 100644 --- a/src/lib/client.c +++ b/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; } diff --git a/src/lib/client_internal.h b/src/lib/client_internal.h index 50c5ff0..3b172cc 100644 --- a/src/lib/client_internal.h +++ b/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 diff --git a/src/lib/testsuite/test_libxce_retry.c b/src/lib/testsuite/test_libxce_retry.c index fe8ba51..a6f619f 100644 --- a/src/lib/testsuite/test_libxce_retry.c +++ b/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 = ∾ + + 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 = ∾ + + 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(); }