feat(libxce-failover): foundation — per-ctx failover state, redirect-hint fields, remove orphaned retry primitive (task 1.1)

This commit is contained in:
Rongsong Shen 2026-06-17 00:05:41 +00:00
parent 8c11655e53
commit 387ccfa8dc
17 changed files with 2122 additions and 562 deletions

View file

@ -176,10 +176,12 @@ extern int xce_upgrade_config_req_extract_raw(const uint8_t *pkt_data,
size_t *out_len,
uint8_t **out_sha256);
/* ── Task 10.2: Client-Side Transparent Retry (Req 7.1–7.6) ─────────────────
/* ── Client-Side Transparent Retry tunables (Req 6.1, 6.2) ──────────────────
*
* xce_retry_config_t: configurable retry parameters with safe production defaults.
* Pass NULL to xce_request_with_retry() to use defaults.
* xce_retry_config_t: configurable retry parameters with safe production
* defaults, consumed by the libxce client failover loop (client.c /
* client_failover.c). XCE_RETRY_DEFAULTS supplies the defaults applied when no
* override is configured.
*/
typedef struct {
int max_retries; /* default 20 */
@ -190,29 +192,4 @@ typedef struct {
extern const xce_retry_config_t XCE_RETRY_DEFAULTS;
/*
* xce_request_with_retry: transparent retry wrapper for transient controller errors.
*
* - XCE_ERR_NOT_READY → backoff, retry same endpoint (Req 7.1)
* - XCE_ERR_STANDBY with redirect_host/port → reconnect, retry (Req 7.2)
* - XCE_ERR_STANDBY without redirect → backoff, retry (Req 7.2 fallback)
* - ECONNREFUSED / ETIMEDOUT → reconnect, retry (Req 7.3)
* - EPERM / EINVAL / ENOENT / EACCES → surface immediately (Req 7.4)
* - Exhausted max_retries → surface last error (Req 7.5)
* - Backoff jitter via rand_r() to prevent thundering herd (Req 7.6)
*
* NULL cfg → uses XCE_RETRY_DEFAULTS.
*/
extern int xce_request_with_retry(endpoint_t *ep,
xce_request_t *req,
const xce_retry_config_t *cfg);
/*
* xce_retry_send_hook: test-only injection point for the send+receive round-trip.
* When non-NULL, xce_request_with_retry() calls this instead of the real network path.
* Returns 0 on success (reply_out populated), -errno on transport failure.
*/
extern int (*xce_retry_send_hook)(endpoint_t *ep, xce_request_t *req,
xce_reply_t *reply_out);
#endif

View file

@ -2,6 +2,9 @@
#define _XCECTX_H_ 1
#include <stdint.h>
#include "config.h" /* XCE_MAXHOSTNAME_LEN */
#include "netio.h"
#include "request.h"
#include "vtaskmgr.h"
@ -28,6 +31,13 @@ typedef struct {
cbuf_t wbuf;
dataset_t* dataset;
/* Redirect-hint address captured from the last redirect reply by
* handle_reply() (which only sees xce_ctx_t) and read by the failover
* rotator on the next retry. Empty host ("") means no hint. Advisory only;
* correctness never depends on it. */
char redirect_hint_host[XCE_MAXHOSTNAME_LEN];
uint16_t redirect_hint_port;
struct {
union {
char* debugmsg;

View file

@ -4,6 +4,7 @@
#include <sys/socket.h>
#include "client.h"
#include "client_internal.h"
#include "controller.capnp.h"
#include "mylock.h"
@ -13,19 +14,6 @@
#include "xcectx.h"
#include "xlist.h"
typedef enum {
API_CTX_INIT = 0,
API_CTX_RUN,
API_CTX_DOWN,
API_CTX_ERR
} api_ctx_state_t;
typedef struct {
xce_client_config_t* conf;
endpoint_t ep;
xce_ctx_t ctx;
} api_ctx_t;
int xcecli_ctx_init(xcecli_ctx_t* c, xce_client_config_t* cfg)
{
api_ctx_t* ctx;
@ -126,6 +114,20 @@ int xcecli_start(xcecli_ctx_t c)
return rc;
}
/* Initialize per-context failover state: sticky primary starts at
* masters[0] (the initial connect target above), retry tunables default
* to XCE_RETRY_DEFAULTS, and no rejoin hook is installed for pure clients.
*/
rc = mylock_init(&(ctx->failover.lock));
if (rc < 0) {
endpoint_fini(&(ctx->ep));
return rc;
}
ctx->failover.cur_master = 0;
ctx->failover.retry = XCE_RETRY_DEFAULTS;
ctx->failover.rejoin_cb = NULL;
ctx->failover.rejoin_arg = NULL;
rc = xce_ctx_start(&(ctx->ctx));
return rc;
@ -141,6 +143,7 @@ int xcecli_stop(xcecli_ctx_t c)
xce_ctx_stop(&(ctx->ctx));
endpoint_fini(&(ctx->ep));
mylock_fini(&(ctx->failover.lock));
return 0;
}

63
src/lib/client_internal.h Normal file
View file

@ -0,0 +1,63 @@
#if !defined(_CLIENT_INTERNAL_H_)
#define _CLIENT_INTERNAL_H_ 1
#include <stdint.h>
#include "client.h"
#include "mylock.h"
#include "netio.h"
#include "request.h"
#include "xcectx.h"
/*
* client_internal.h — private libxce client state shared between client.c and
* the forthcoming failover helper (client_failover.c, task 1.x).
*
* api_ctx_t was previously file-local to client.c. It is moved here so the
* failover helper can operate on it without exposing it through the public
* API in client.h.
*/
typedef enum {
API_CTX_INIT = 0,
API_CTX_RUN,
API_CTX_DOWN,
API_CTX_ERR
} 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.
*/
typedef int (*xcecli_reconnect_cb)(xcecli_ctx_t c, void* arg);
/*
* xce_failover_state_t: per-context failover state. Persists across calls so a
* sticky primary and the retry tunables survive between requests.
*
* lock — serializes xcecli_* per ctx (shared-ctx safety)
* cur_master — index into conf->masters[] (sticky primary)
* retry — defaults from XCE_RETRY_DEFAULTS unless overridden
* rejoin_cb — consumer rejoin hook; NULL for pure clients
* rejoin_arg — opaque argument passed to rejoin_cb
*/
typedef struct {
mylock_t lock;
int cur_master;
xce_retry_config_t retry;
xcecli_reconnect_cb rejoin_cb;
void* rejoin_arg;
} xce_failover_state_t;
typedef struct {
xce_client_config_t* conf;
endpoint_t ep;
xce_ctx_t ctx;
xce_failover_state_t failover;
} api_ctx_t;
#endif

View file

@ -1920,109 +1920,17 @@ int xce_clear_reason_req(endpoint_t* ep, xce_clear_reason_t* reason)
return xce_request(ep, &request);
}
/* ── Task 10.2: Client-Side Transparent Retry (Req 7.1–7.6) ────────────────
/* ── Client-Side Transparent Retry tunables (Req 6.1, 6.2) ─────────────────
*
* xce_retry_send_hook: test-injectable send+receive round-trip.
* NULL in production; set by unit tests to return controlled replies.
* XCE_RETRY_DEFAULTS provides safe production defaults for the failover retry
* loop that lives in the libxce client layer (client.c / client_failover.c).
* The orphaned no-op xce_request_with_retry() primitive and its
* xce_retry_send_hook test seam were removed; they are superseded by the
* xcecli_* failover loop.
*/
int (*xce_retry_send_hook)(endpoint_t *ep, xce_request_t *req,
xce_reply_t *reply_out) = NULL;
const xce_retry_config_t XCE_RETRY_DEFAULTS = {
.max_retries = 20,
.initial_retry_ms = 50,
.max_retry_ms = 5000,
.backoff_multiplier = 2.0,
};
/* is_non_retriable: errors that should surface immediately without retry */
static int is_non_retriable(int status)
{
return (status == -EPERM || status == -EACCES ||
status == -EINVAL || status == -ENOENT);
}
/* ms_sleep_jittered: sleep delay_ms + random jitter via rand_r (Req 7.6) */
static void ms_sleep_jittered(int delay_ms, unsigned int *seed)
{
if (delay_ms <= 0) return;
int jitter_ms = (int)((rand_r(seed) % (delay_ms / 4 + 1)));
long total_ns = ((long)delay_ms + jitter_ms) * 1000000L;
struct timespec ts = { .tv_sec = total_ns / 1000000000L,
.tv_nsec = total_ns % 1000000000L };
nanosleep(&ts, NULL);
}
int xce_request_with_retry(endpoint_t *ep,
xce_request_t *req,
const xce_retry_config_t *cfg)
{
if (cfg == NULL)
cfg = &XCE_RETRY_DEFAULTS;
unsigned int rand_seed = (unsigned int)(uintptr_t)ep ^ (unsigned int)time(NULL);
int delay_ms = cfg->initial_retry_ms;
int last_rc = -ETIMEDOUT;
for (int attempt = 0; attempt <= cfg->max_retries; attempt++) {
xce_reply_t reply;
memset(&reply, 0, sizeof(reply));
int rc;
if (xce_retry_send_hook) {
rc = xce_retry_send_hook(ep, req, &reply);
} else {
/* Production path: fire-and-forget send; reply handling is caller's
* responsibility via the event loop. Return send error directly. */
rc = xce_request(ep, req);
if (rc != 0)
return rc;
return 0; /* async: caller reads reply via event loop */
}
/* Transport-level failure (ECONNREFUSED, ETIMEDOUT) → retry (Req 7.3) */
if (rc == -ECONNREFUSED || rc == -ETIMEDOUT) {
last_rc = rc;
if (attempt < cfg->max_retries)
ms_sleep_jittered(delay_ms, &rand_seed);
goto next_backoff;
}
/* Any other transport error → surface immediately */
if (rc != 0)
return rc;
/* Application-layer status from reply */
int status = (int)reply.status;
if (status == 0)
return 0; /* success */
last_rc = status;
/* Non-retriable errors: surface immediately (Req 7.4) */
if (is_non_retriable(status))
return status;
/* XCE_ERR_STANDBY with redirect → reconnect (Req 7.2).
* In hook mode, the hook already updated ep/redirect state.
* Just retry; redirect handling is at the transport layer. */
if (status == -E_XCE_STANDBY && reply.redirect_host != NULL
&& reply.redirect_port != 0) {
/* no extra sleep for redirect — just retry immediately */
continue;
}
/* XCE_ERR_NOT_READY or XCE_ERR_STANDBY without redirect → backoff */
if (attempt < cfg->max_retries)
ms_sleep_jittered(delay_ms, &rand_seed);
next_backoff:
{
int next = (int)((double)delay_ms * cfg->backoff_multiplier);
delay_ms = next > cfg->max_retry_ms ? cfg->max_retry_ms : next;
}
}
return last_rc;
}

View file

@ -1,324 +1,35 @@
/*
* test_libxce_retry.c — RED-phase tests for xce_request_with_retry() (tasks 10.1/10.2).
* test_libxce_retry.c — placeholder for the libxce failover retry tests.
*
* Uses xce_retry_send_hook to inject mock round-trip outcomes without real
* network connections. Each mock call returns a configured status/redirect.
* The orphaned xce_request_with_retry() primitive and its xce_retry_send_hook
* test seam were removed (superseded by the xcecli_* failover loop). Task 1.2
* rewrites this suite against the new failover model (StatusClassifier, master
* rotation, redirect-hint, budget → cluster-not-ready).
*
* Requirements: 7.1, 7.2, 7.3, 7.4, 7.5, 7.6
* For now this is a minimal compiling placeholder that asserts the retained
* retry-tunables scaffolding (xce_retry_config_t + XCE_RETRY_DEFAULTS) is
* reachable, keeping the lib test suite GREEN between tasks 1.1 and 1.2.
*
* Requirements: 6.1, 6.2
*/
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
#include "request.h"
#include "xce/errnox.h"
/* ── Forward declarations for functions-under-test ───────────────── */
/* xce_request_with_retry is not yet implemented — linker will fail (RED). */
extern int xce_request_with_retry(endpoint_t *ep,
xce_request_t *req,
const xce_retry_config_t *cfg);
/*
* xce_retry_send_hook: test seam for synchronous request-reply.
* When set, xce_request_with_retry calls this instead of the real network path.
* The hook fills in *reply_out and returns 0 on success, -errno on failure.
*/
extern int (*xce_retry_send_hook)(endpoint_t *ep, xce_request_t *req,
xce_reply_t *reply_out);
/* ── Default retry config for tests (fast backoff) ──────────────── */
static const xce_retry_config_t g_fast_cfg = {
.max_retries = 5,
.initial_retry_ms = 1,
.max_retry_ms = 10,
.backoff_multiplier = 2.0,
};
/* ── Mock state ─────────────────────────────────────────────────── */
#define MAX_MOCK_REPLIES 32
struct mock_reply {
int32_t status; /* xce_reply_t.status to return */
int send_rc; /* return code from the hook (0 or -errno) */
char *redirect_host; /* non-NULL if status == -E_XCE_STANDBY */
uint16_t redirect_port;
};
static struct mock_reply g_replies[MAX_MOCK_REPLIES];
static int g_reply_count = 0;
static int g_reply_idx = 0;
static int g_call_count = 0;
static void reset_mock(void)
/* Req 6.1, 6.2: the retry-tunables struct + safe production defaults are
* retained and reachable for the failover loop. */
static void test_retry_defaults_are_reachable(void)
{
for (int i = 0; i < g_reply_count; i++) {
free(g_replies[i].redirect_host);
g_replies[i].redirect_host = NULL;
}
g_reply_count = 0;
g_reply_idx = 0;
g_call_count = 0;
memset(g_replies, 0, sizeof(g_replies));
}
xce_retry_config_t cfg = XCE_RETRY_DEFAULTS;
static void add_reply(int32_t status, int send_rc,
const char *rhost, uint16_t rport)
{
if (g_reply_count >= MAX_MOCK_REPLIES) return;
int i = g_reply_count++;
g_replies[i].status = status;
g_replies[i].send_rc = send_rc;
g_replies[i].redirect_host = rhost ? strdup(rhost) : NULL;
g_replies[i].redirect_port = rport;
}
TEST_ASSERT_EQUAL_INT(20, cfg.max_retries);
TEST_ASSERT_EQUAL_INT(50, cfg.initial_retry_ms);
TEST_ASSERT_EQUAL_INT(5000, cfg.max_retry_ms);
TEST_ASSERT_TRUE(cfg.backoff_multiplier > 1.0);
static int mock_send(endpoint_t *ep, xce_request_t *req, xce_reply_t *reply_out)
{
(void)ep; (void)req;
g_call_count++;
if (g_reply_idx >= g_reply_count) {
/* Default: return success after mocks exhausted */
memset(reply_out, 0, sizeof(*reply_out));
reply_out->status = 0;
return 0;
}
struct mock_reply *m = &g_replies[g_reply_idx++];
if (m->send_rc != 0)
return m->send_rc;
memset(reply_out, 0, sizeof(*reply_out));
reply_out->status = m->status;
reply_out->redirect_host = m->redirect_host;
reply_out->redirect_port = m->redirect_port;
return 0;
}
static void install_mock(void)
{
xce_retry_send_hook = mock_send;
}
/* ── Tests ──────────────────────────────────────────────────────── */
/* Req 7.1: E_XCE_NOT_READY → retry with backoff, eventually succeed */
static void test_not_ready_retries_then_succeeds(void)
{
install_mock();
reset_mock();
/* 3 NOT_READY responses, then success */
add_reply(-E_XCE_NOT_READY, 0, NULL, 0);
add_reply(-E_XCE_NOT_READY, 0, NULL, 0);
add_reply(-E_XCE_NOT_READY, 0, NULL, 0);
add_reply(0, 0, NULL, 0); /* success */
xce_request_t req;
memset(&req, 0, sizeof(req));
int rc = xce_request_with_retry((endpoint_t*)0x1, &req, &g_fast_cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(4, g_call_count);
printf("test_not_ready_retries_then_succeeds: PASS\n");
}
/* Req 7.2: E_XCE_STANDBY with redirect → reconnect to redirect, retry */
static void test_standby_with_redirect_reconnects_and_retries(void)
{
install_mock();
reset_mock();
/* Standby response with redirect */
add_reply(-E_XCE_STANDBY, 0, "primary.host", 9990);
add_reply(0, 0, NULL, 0); /* success on redirect target */
xce_request_t req;
memset(&req, 0, sizeof(req));
int rc = xce_request_with_retry((endpoint_t*)0x1, &req, &g_fast_cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, g_call_count);
printf("test_standby_with_redirect_reconnects_and_retries: PASS\n");
}
/* Req 7.2 fallback: E_XCE_STANDBY without redirect → backoff, retry same */
static void test_standby_without_redirect_falls_back_to_backoff(void)
{
install_mock();
reset_mock();
/* STANDBY without redirect, then success */
add_reply(-E_XCE_STANDBY, 0, NULL, 0);
add_reply(0, 0, NULL, 0);
xce_request_t req;
memset(&req, 0, sizeof(req));
int rc = xce_request_with_retry((endpoint_t*)0x1, &req, &g_fast_cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, g_call_count);
printf("test_standby_without_redirect_falls_back_to_backoff: PASS\n");
}
/* Req 7.4: non-retriable error (EPERM) → surface immediately, no retry */
static void test_non_retriable_error_surfaces_immediately(void)
{
install_mock();
reset_mock();
/* Auth/perm failure — must NOT retry */
add_reply(-EPERM, 0, NULL, 0);
/* Add a success after — it should never be reached */
add_reply(0, 0, NULL, 0);
xce_request_t req;
memset(&req, 0, sizeof(req));
int rc = xce_request_with_retry((endpoint_t*)0x1, &req, &g_fast_cfg);
TEST_ASSERT_EQUAL_INT(-EPERM, rc);
TEST_ASSERT_EQUAL_INT(1, g_call_count); /* only 1 attempt */
printf("test_non_retriable_error_surfaces_immediately: PASS\n");
}
/* Req 7.5: max_retries exhausted → surface last error */
static void test_max_retries_exhausted_returns_last_error(void)
{
install_mock();
reset_mock();
/* Fill with NOT_READY beyond max_retries */
for (int i = 0; i <= g_fast_cfg.max_retries + 1; i++)
add_reply(-E_XCE_NOT_READY, 0, NULL, 0);
xce_request_t req;
memset(&req, 0, sizeof(req));
int rc = xce_request_with_retry((endpoint_t*)0x1, &req, &g_fast_cfg);
/* Must return the last error, not 0 */
TEST_ASSERT_NOT_EQUAL(0, rc);
/* Must have tried max_retries+1 times total */
TEST_ASSERT_EQUAL_INT(g_fast_cfg.max_retries + 1, g_call_count);
printf("test_max_retries_exhausted_returns_last_error: PASS\n");
}
/* Req 7.3: ECONNREFUSED → reconnect, retry */
static void test_conn_refused_retries(void)
{
install_mock();
reset_mock();
add_reply(0, -ECONNREFUSED, NULL, 0);
add_reply(0, 0, NULL, 0);
xce_request_t req;
memset(&req, 0, sizeof(req));
int rc = xce_request_with_retry((endpoint_t*)0x1, &req, &g_fast_cfg);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(2, g_call_count);
printf("test_conn_refused_retries: PASS\n");
}
/* NULL cfg → uses defaults (max_retries=20, etc.) */
static void test_null_cfg_uses_defaults(void)
{
install_mock();
reset_mock();
add_reply(0, 0, NULL, 0); /* immediate success */
xce_request_t req;
memset(&req, 0, sizeof(req));
int rc = xce_request_with_retry((endpoint_t*)0x1, &req, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
printf("test_null_cfg_uses_defaults: PASS\n");
}
/* Req 7.6: 50 concurrent simulated clients all succeed (backoff prevents starvation) */
struct concurrent_client_args {
int id;
int rc;
atomic_int *success_count;
};
static void *concurrent_client_fn(void *arg)
{
struct concurrent_client_args *a = arg;
xce_request_t req;
memset(&req, 0, sizeof(req));
/* Each client sees NOT_READY twice, then success */
int rc = xce_request_with_retry((endpoint_t*)0x1, &req, &g_fast_cfg);
a->rc = rc;
if (rc == 0)
atomic_fetch_add(a->success_count, 1);
return NULL;
}
static int concurrent_mock_send(endpoint_t *ep, xce_request_t *req,
xce_reply_t *reply_out)
{
(void)ep; (void)req;
static atomic_int s_call_count = 0;
int n = atomic_fetch_add(&s_call_count, 1);
memset(reply_out, 0, sizeof(*reply_out));
/* First 3 calls per "client" return NOT_READY, then succeed.
* Since we can't track per-client state here, use a simple global
* threshold: first 100 calls NOT_READY, rest succeed. */
if (n < 100)
reply_out->status = -E_XCE_NOT_READY;
else
reply_out->status = 0;
return 0;
}
static void test_50_concurrent_clients_all_succeed(void)
{
xce_retry_send_hook = concurrent_mock_send;
const int N = 50;
pthread_t threads[50];
struct concurrent_client_args args[50];
atomic_int success_count;
atomic_init(&success_count, 0);
for (int i = 0; i < N; i++) {
args[i].id = i;
args[i].rc = -1;
args[i].success_count = &success_count;
pthread_create(&threads[i], NULL, concurrent_client_fn, &args[i]);
}
/* Wait for all clients (with 30s overall deadline from Unity timeout) */
for (int i = 0; i < N; i++)
pthread_join(threads[i], NULL);
TEST_ASSERT_EQUAL_INT(N, atomic_load(&success_count));
printf("test_50_concurrent_clients_all_succeed: PASS (%d/%d)\n",
atomic_load(&success_count), N);
printf("test_retry_defaults_are_reachable: PASS\n");
}
/* ── Suite entry point ──────────────────────────────────────────── */
@ -326,13 +37,6 @@ static void test_50_concurrent_clients_all_succeed(void)
void test_libxce_retry(void)
{
UNITY_BEGIN();
RUN_TEST(test_not_ready_retries_then_succeeds);
RUN_TEST(test_standby_with_redirect_reconnects_and_retries);
RUN_TEST(test_standby_without_redirect_falls_back_to_backoff);
RUN_TEST(test_non_retriable_error_surfaces_immediately);
RUN_TEST(test_max_retries_exhausted_returns_last_error);
RUN_TEST(test_conn_refused_retries);
RUN_TEST(test_null_cfg_uses_defaults);
RUN_TEST(test_50_concurrent_clients_all_succeed);
RUN_TEST(test_retry_defaults_are_reachable);
UNITY_END();
}