lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(libxce-failover): reconnect/rejoin hook — public setter + post-reconnect invocation before resume (task 4.1)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-17 03:46:05 +00:00
parent 12284bc3f5
commit 8783ce5a9d
6 changed files with 264 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

14
src/include/client.h
View file

@ -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);

21
src/lib/client.c
View file

@ -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;

75
src/lib/client_failover.c
View file

@ -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;

27
src/lib/client_internal.h
View file

@ -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

150
src/lib/testsuite/test_libxce_retry.c
View file

@ -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();
}