diff --git a/.kiro/specs/libxce-failover/tasks.md b/.kiro/specs/libxce-failover/tasks.md index 76f57f7..08f351e 100644 --- a/.kiro/specs/libxce-failover/tasks.md +++ b/.kiro/specs/libxce-failover/tasks.md @@ -56,7 +56,7 @@ - _Boundary: xcecli_retry_next_ - [ ] 4. Core: reconnect/rejoin hook -- [ ] 4.1 Add the reconnect/rejoin hook mechanism +- [x] 4.1 Add the reconnect/rejoin hook mechanism - Provide a client API to register a reconnect callback and argument, stored in the failover state - Invoke the callback after each successful reconnect to a new primary and before resuming the interrupted op; treat a failing callback as a failed attempt that rotates or backs off per its result - Done: with a registered hook a simulated failover invokes it before resume; with no hook the loop resumes directly @@ -106,3 +106,4 @@ - 2.3: RedirectCapture via non-static seam `xce_ctx_capture_redirect_hint()` called in handle_reply BEFORE the capnp free AND before the `status<0` early return (so negative E_XCE_REDIRECT still captures). Bounded strncpy into redirect_hint_host. ReplyWaitWake needed NO production change — existing `handle_disconnect` (err_handler, xcectx.c) already injects -ECONNRESET into wbuf + sets XCE_CTX_ERR; full err_handler path is covered by E2E (6.2). - 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. diff --git a/src/include/client.h b/src/include/client.h index 7d5e21c..6957319 100644 --- a/src/include/client.h +++ b/src/include/client.h @@ -11,6 +11,18 @@ typedef void* xcecli_ctx_t; +/* + * xcecli_reconnect_cb: consumer rejoin hook invoked once per transparent + * reconnect/failover, AFTER the reconnect/recovery and BEFORE the interrupted + * op resumes. The callback CONTENT (join handshake, worker job re-sync) belongs + * to the consumer; libxce owns only the mechanism + ordering. Runs on the + * calling thread. Return 0 to proceed with the interrupted op; <0 to treat the + * rejoin as failed (the new primary rejected the join) so the failover loop + * rotates/backs off again instead of resuming. Pure clients register NO hook + * (NULL) — the loop then resumes directly. + */ +typedef int (*xcecli_reconnect_cb)(xcecli_ctx_t ctx, void* arg); + #define N_MAX_MASTERS (3) typedef struct { @@ -26,6 +38,8 @@ typedef struct { extern int xcecli_ctx_init(xcecli_ctx_t*, xce_client_config_t*); extern void xcecli_ctx_fini(xcecli_ctx_t); extern int xcecli_set_handler(xcecli_ctx_t, xce_request_handler_t, void*); +extern int xcecli_set_reconnect_handler(xcecli_ctx_t, xcecli_reconnect_cb, + void*); extern int xcecli_start(xcecli_ctx_t); extern int xcecli_stop(xcecli_ctx_t); extern int xcecli_wakeup(xcecli_ctx_t, int); diff --git a/src/lib/client.c b/src/lib/client.c index 81dd81a..b0449ea 100644 --- a/src/lib/client.c +++ b/src/lib/client.c @@ -76,6 +76,27 @@ int xcecli_set_handler(xcecli_ctx_t c, xce_request_handler_t handler, return xce_ctx_set_handler(&(ctx->ctx), handler, args); } +/* + * xcecli_set_reconnect_handler (task 4.1) — register the consumer rejoin hook + * the failover loop invokes after a transparent reconnect, before resuming the + * interrupted op. A NULL cb clears any previously registered hook (back to the + * pure-client no-op path). Stores cb+arg on the per-ctx failover state. + */ +int xcecli_set_reconnect_handler(xcecli_ctx_t c, xcecli_reconnect_cb cb, + void* arg) +{ + api_ctx_t* ctx = (api_ctx_t*)c; + + if (c == NULL) { + return -EINVAL; + } + + ctx->failover.rejoin_cb = cb; + ctx->failover.rejoin_arg = arg; + + return 0; +} + int xcecli_wakeup(xcecli_ctx_t c, int status) { api_ctx_t* ctx = (api_ctx_t*)c; diff --git a/src/lib/client_failover.c b/src/lib/client_failover.c index ce2b13a..63c1f46 100644 --- a/src/lib/client_failover.c +++ b/src/lib/client_failover.c @@ -189,6 +189,25 @@ int xcecli_ctx_recover(api_ctx_t* ac) return 0; } +/* ════════════════════════════════════════════════════════════════════════ + * ReconnectRejoinHook — invoke the consumer rejoin callback (1.2) + * ════════════════════════════════════════════════════════════════════════ */ + +int xcecli_invoke_rejoin(api_ctx_t* ac) +{ + if (ac == NULL) + return -EINVAL; + + /* Pure client: no hook registered → nothing to do, resume directly. */ + if (ac->failover.rejoin_cb == NULL) + return 0; + + /* Stateful consumer: run its rejoin/state-exchange on the calling thread. + * Its rc is propagated verbatim — 0 to proceed, <0 to signal the join was + * rejected so the wrapper treats this attempt as failed (no resume). */ + return ac->failover.rejoin_cb((xcecli_ctx_t)ac, ac->failover.rejoin_arg); +} + /* ════════════════════════════════════════════════════════════════════════ * XcecliRetryLoop — decision core (pure) + side-effecting wrapper * ════════════════════════════════════════════════════════════════════════ */ @@ -318,26 +337,44 @@ bool xcecli_retry_next(api_ctx_t* ac, int* rc, bool is_write, int rrc = xce_failover_rotate(ac); if (rrc < 0) { WARN("xcecli failover: rotate failed (rc=%d); retrying anyway", rrc); - } else { - /* Reconnect succeeded: repair the ctx before the next xce_* attempt - * (CI2 ERR→RUN state reset, CI1 stale-reply drain) so the per-op guard - * passes and no prior-attempt reply is consumed (1.2, 4.4). */ - xcecli_ctx_recover(ac); + return true; + } - /* Log the rotation target (7.1). xce_failover_rotate consumes the - * redirect hint when present (so the post-rotate target is not always - * the sticky master); report the sticky master cursor as a best-effort - * endpoint label, which is exact for the configured-master rotation. */ - int cur = ac->failover.cur_master; - const char* host = - (ac->conf != NULL && cur >= 0 && cur < ac->conf->nmasters) - ? ac->conf->masters[cur] - : "(unknown)"; - int port = (ac->conf != NULL && cur >= 0 && cur < ac->conf->nmasters) - ? ac->conf->master_ports[cur] - : 0; - INFO("xcecli failover: attempt %d → %s:%d", st->attempt, - host ? host : "(null)", port); + /* Reconnect succeeded: repair the ctx before the next xce_* attempt + * (CI2 ERR→RUN state reset, CI1 stale-reply drain) so the per-op guard + * passes and no prior-attempt reply is consumed (1.2, 4.4). */ + xcecli_ctx_recover(ac); + + /* Log the rotation target (7.1). xce_failover_rotate consumes the redirect + * hint when present (so the post-rotate target is not always the sticky + * master); report the sticky master cursor as a best-effort endpoint label, + * which is exact for the configured-master rotation. */ + int cur = ac->failover.cur_master; + const char* host = + (ac->conf != NULL && cur >= 0 && cur < ac->conf->nmasters) + ? ac->conf->masters[cur] + : "(unknown)"; + int port = (ac->conf != NULL && cur >= 0 && cur < ac->conf->nmasters) + ? ac->conf->master_ports[cur] + : 0; + INFO("xcecli failover: attempt %d → %s:%d", st->attempt, + host ? host : "(null)", port); + + /* ReconnectRejoinHook (1.2): a stateful consumer re-establishes its session + * on the freshly-connected primary AFTER reconnect/recovery and BEFORE the + * interrupted op resumes. A pure client has no hook → no-op, resume now. */ + int hrc = xcecli_invoke_rejoin(ac); + if (hrc < 0) { + /* The new primary rejected the daemon's join: treat this attempt as + * FAILED — do NOT resume into a primary that doesn't know the daemon. + * Reclassify the rejoin failure as a transport-style failover cause so + * the loop rotates/backs off again on the next iteration (subject to the + * same retry budget). When the budget is exhausted decide() returns + * false and *rc carries the last cause. */ + WARN("xcecli failover: rejoin hook failed (rc=%d); rotating again", + hrc); + *rc = hrc; + return xcecli_retry_next(ac, rc, is_write, st); } return true; diff --git a/src/lib/client_internal.h b/src/lib/client_internal.h index aa3d252..50c5ff0 100644 --- a/src/lib/client_internal.h +++ b/src/lib/client_internal.h @@ -30,13 +30,10 @@ typedef enum { } api_ctx_state_t; /* - * xcecli_reconnect_cb: consumer rejoin hook invoked after a transparent - * reconnect/failover and before the in-flight request resumes. NULL for pure - * clients (e.g. xce_test). The public declaration and the - * xcecli_set_reconnect_handler() setter are introduced by a later task; this - * private typedef lets the failover state carry the hook now. + * xcecli_reconnect_cb (consumer rejoin hook) and its public setter + * xcecli_set_reconnect_handler() are declared in the public client.h (included + * above) as of task 4.1. The failover state below carries the registered hook. */ -typedef int (*xcecli_reconnect_cb)(xcecli_ctx_t c, void* arg); /* * xce_failover_state_t: per-context failover state. Persists across calls so a @@ -160,4 +157,22 @@ bool xcecli_retry_next(api_ctx_t* ac, int* rc, bool is_write, */ int xcecli_ctx_recover(api_ctx_t* ac); +/* + * xcecli_invoke_rejoin (task 4.1) — invoke the consumer rejoin hook once after a + * successful reconnect/recovery and BEFORE the interrupted op resumes. Socket- + * free (it only calls the stored callback), so it is unit-testable. The + * xcecli_retry_next wrapper calls it AFTER xce_failover_rotate + + * xcecli_ctx_recover and BEFORE re-sending the op. + * + * If no hook is registered (rejoin_cb == NULL, a pure client) it is a no-op and + * returns 0 so the loop resumes directly. Otherwise it returns the callback's + * rc: 0 to proceed, <0 to treat the rejoin as FAILED (the new primary rejected + * the daemon's join) — the wrapper then does NOT resume but rotates/backs off on + * the next iteration. + * + * Returns 0 on a successful (or absent) rejoin, the hook's <0 rc on a failed + * rejoin, or -EINVAL on a NULL ac. + */ +int xcecli_invoke_rejoin(api_ctx_t* ac); + #endif diff --git a/src/lib/testsuite/test_libxce_retry.c b/src/lib/testsuite/test_libxce_retry.c index 6de1614..fe8ba51 100644 --- a/src/lib/testsuite/test_libxce_retry.c +++ b/src/lib/testsuite/test_libxce_retry.c @@ -1127,6 +1127,151 @@ static void test_recover_null_ctx_safe(void) printf("test_recover_null_ctx_safe: PASS\n"); } +/* ════════════════════════════════════════════════════════════════════════ + * Task 4.1 — ReconnectRejoinHook (xcecli_set_reconnect_handler + + * xcecli_invoke_rejoin) RED→GREEN tests. + * + * libxce owns the hook MECHANISM + ORDERING (reconnect → hook → resume); the + * callback CONTENT is the consumer's. The failover machinery invokes the + * registered callback once per successful reconnect, AFTER reconnect/recovery + * and BEFORE the interrupted op resumes. A hook returning <0 means the new + * primary rejected the daemon's join → treat it as a FAILED attempt (do not + * resume; the loop rotates/backs off again). Pure clients register NO hook + * (NULL) → invoke is a no-op returning 0 and the loop resumes directly. + * + * xcecli_invoke_rejoin is socket-free (it only calls the stored cb), so it is + * unit-testable directly here without a live server. + * + * Requirements: 1.2 (re-establish session against new primary before resume). + * ════════════════════════════════════════════════════════════════════════ */ + +/* A test rejoin callback: records that it ran (and the ctx/arg it saw) and + * returns a caller-controlled rc so we can drive the success and failed-join + * paths. */ +typedef struct { + int called; /* number of times the hook fired */ + xcecli_ctx_t seen_ctx; /* the ctx the hook observed */ + void *seen_arg; /* the opaque arg the hook observed */ + int ret; /* rc the hook returns (0 ok, <0 failed rejoin) */ +} fo_rejoin_probe_t; + +static int fo_rejoin_cb(xcecli_ctx_t ctx, void *arg) +{ + fo_rejoin_probe_t *p = (fo_rejoin_probe_t *)arg; + p->called++; + p->seen_ctx = ctx; + p->seen_arg = arg; + return p->ret; +} + +/* Req 1.2: the setter stores cb+arg into the failover state; invoke then calls + * the registered cb exactly once, passing the opaque ctx and arg, and returns + * the cb's rc (0 on a successful rejoin). */ +static void test_invoke_rejoin_calls_registered_hook(void) +{ + xce_client_config_t conf; + api_ctx_t ac; + fo_make_client_masters(&conf); + fo_make_api_ctx(&ac, &conf); + + fo_rejoin_probe_t probe = { 0 }; + probe.ret = 0; + + /* setter stores cb+arg on the failover state */ + int src = xcecli_set_reconnect_handler((xcecli_ctx_t)&ac, fo_rejoin_cb, + &probe); + TEST_ASSERT_EQUAL_INT(0, src); + TEST_ASSERT_TRUE(ac.failover.rejoin_cb == fo_rejoin_cb); + TEST_ASSERT_TRUE(ac.failover.rejoin_arg == &probe); + + /* invoke fires the hook once, returns its rc (0), passes ctx + arg */ + int rc = xcecli_invoke_rejoin(&ac); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(1, probe.called); + TEST_ASSERT_TRUE(probe.seen_ctx == (xcecli_ctx_t)&ac); + TEST_ASSERT_TRUE(probe.seen_arg == &probe); + + printf("test_invoke_rejoin_calls_registered_hook: PASS\n"); +} + +/* Req 1.2: a hook that returns <0 (the new primary rejected the join) makes + * invoke return that <0 — the wrapper treats it as a failed attempt and does + * NOT resume the interrupted op. */ +static void test_invoke_rejoin_propagates_failure(void) +{ + xce_client_config_t conf; + api_ctx_t ac; + fo_make_client_masters(&conf); + fo_make_api_ctx(&ac, &conf); + + fo_rejoin_probe_t probe = { 0 }; + probe.ret = -ECONNREFUSED; /* join rejected by new primary */ + + TEST_ASSERT_EQUAL_INT( + 0, xcecli_set_reconnect_handler((xcecli_ctx_t)&ac, fo_rejoin_cb, + &probe)); + + int rc = xcecli_invoke_rejoin(&ac); + TEST_ASSERT_EQUAL_INT(-ECONNREFUSED, rc); /* failure surfaced to the loop */ + TEST_ASSERT_EQUAL_INT(1, probe.called); + + printf("test_invoke_rejoin_propagates_failure: PASS\n"); +} + +/* Req 1.2: a pure client registers NO hook (NULL) → invoke is a no-op that + * returns 0 and calls nothing, so the loop resumes directly. */ +static void test_invoke_rejoin_no_hook_is_noop(void) +{ + xce_client_config_t conf; + api_ctx_t ac; + fo_make_client_masters(&conf); + fo_make_api_ctx(&ac, &conf); /* leaves rejoin_cb == NULL */ + + TEST_ASSERT_NULL(ac.failover.rejoin_cb); + + int rc = xcecli_invoke_rejoin(&ac); + TEST_ASSERT_EQUAL_INT(0, rc); /* no hook → no-op success */ + + printf("test_invoke_rejoin_no_hook_is_noop: PASS\n"); +} + +/* Req 1.2: the setter is NULL-safe on a NULL ctx, and can clear a previously + * registered hook by passing NULL (back to the pure-client no-op path). */ +static void test_set_reconnect_handler_null_safe_and_clears(void) +{ + xce_client_config_t conf; + api_ctx_t ac; + fo_make_client_masters(&conf); + fo_make_api_ctx(&ac, &conf); + + /* NULL ctx rejected without crashing */ + TEST_ASSERT_EQUAL_INT(-EINVAL, + xcecli_set_reconnect_handler(NULL, fo_rejoin_cb, + NULL)); + + fo_rejoin_probe_t probe = { 0 }; + TEST_ASSERT_EQUAL_INT( + 0, xcecli_set_reconnect_handler((xcecli_ctx_t)&ac, fo_rejoin_cb, + &probe)); + TEST_ASSERT_TRUE(ac.failover.rejoin_cb == fo_rejoin_cb); + + /* clearing with a NULL cb restores the no-hook no-op path */ + TEST_ASSERT_EQUAL_INT( + 0, xcecli_set_reconnect_handler((xcecli_ctx_t)&ac, NULL, NULL)); + TEST_ASSERT_NULL(ac.failover.rejoin_cb); + TEST_ASSERT_EQUAL_INT(0, xcecli_invoke_rejoin(&ac)); + TEST_ASSERT_EQUAL_INT(0, probe.called); /* cleared hook never fired */ + + printf("test_set_reconnect_handler_null_safe_and_clears: PASS\n"); +} + +/* Guard: invoke on a NULL api_ctx is rejected without crashing. */ +static void test_invoke_rejoin_null_ctx_safe(void) +{ + TEST_ASSERT_EQUAL_INT(-EINVAL, xcecli_invoke_rejoin(NULL)); + printf("test_invoke_rejoin_null_ctx_safe: PASS\n"); +} + /* ── Suite entry point ──────────────────────────────────────────────── */ void test_libxce_retry(void) @@ -1165,5 +1310,10 @@ void test_libxce_retry(void) RUN_TEST(test_recover_leaves_down_state_untouched); RUN_TEST(test_recover_drains_stale_wbuf_entry); RUN_TEST(test_recover_null_ctx_safe); + RUN_TEST(test_invoke_rejoin_calls_registered_hook); + RUN_TEST(test_invoke_rejoin_propagates_failure); + 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); UNITY_END(); }