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

feat(component-failover-adoption): unify worker's dual connect paths into single pipeline

Browse source 
Replace worker's separate default/single-host and multi-candidate connect
branches with one path: synthesize-or-accept a candidate list, validate,
convert via xce_cluster_config_to_client_config(), connect once. Register
a flag-only rejoin callback (no I/O, no client-lib reentry) and remove the
liveness probe's bounded give-up so the library owns retry/recovery.
This commit is contained in:
Rongsong Shen 2026-08-11 05:42:48 +00:00
parent 86e442c9cd
commit ce9f734a38
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
6 changed files with 556 additions and 141 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/tools/xce-build.sh
View file

@ -90,9 +90,12 @@ case "$COMPONENT" in
;;
worker)
setup_build
echo "=== Building lib + worker ==="
meson compile -C "$BUILD_DIR" xcecd xcecd_worker xrun
echo "=== Worker built (no unit tests registered) ==="
echo "=== Building lib + worker + tests ==="
meson compile -C "$BUILD_DIR" xcecd xcecd_worker xrun controller_connector_test
if [ "$RUN_TESTS" = 1 ]; then
echo "=== Running worker tests ==="
meson test -C "$BUILD_DIR" --suite worker --print-errorlogs
fi
;;
apiserver)
setup_build

2
src/worker/meson.build
View file

@ -103,3 +103,5 @@ xcecd_virtual = executable('xcecd_virtual',
include_directories: [incdir],
dependencies:[xcecd_deps],
install:true)
subdir('testsuite')

330
src/worker/src/controller_connector.c
View file

@ -31,6 +31,23 @@ xcecli_ctx_t xcecli_ctx = { 0 };
xce_client_config_t xce_client_cfg = { 0 };
controller_connector_ctx_t* cc_ctx = NULL;
/* Persistent storage for the single candidate list the current connect/join
* attempt is driving (component-failover-adoption, task 2.1, Req 1 AC5).
* Must outlive the xcecli_ctx_t connection: xce_cluster_config_to_client_
* config() points xce_client_cfg.masters[] directly at this struct's
* hostname buffers (non-owning aliasing -- the library keeps a reference to
* xce_client_cfg for the life of the connection, so the source data behind
* it cannot be a stack-local temporary). Populated by build_cluster_config()
* below from either the operator-supplied --peers list or a synthesized
* single candidate -- never both, never a stack copy. */
static xce_cluster_config_t g_cluster_cfg = { 0 };
/* Set by the rejoin callback registered with the client library
* (xcecd_on_reconnect() below) when the library has transparently
* reconnected and this connector needs to re-issue xcecli_join() as a
* separate, later call. Consumed by checking_connection() (task 2.2). */
static bool needs_rejoin = false;
static int myreq_handler(xce_request_t*, void*);
/* make compiler happy */
@ -122,11 +139,17 @@ int start_controller_connector_thread(task_func_t function, void* arg)
}
/**
* @brief Initializes the client configuration for the controller connector.
* @brief Sets the caller-owned identity/auth fields (from, key) on the
* client config. Deliberately does NOT touch nmasters/masters[]/
* master_ports[] -- those are owned by the shared
* xce_cluster_config_to_client_config() helper (build_cluster_config()
* + the conversion call in xcecd_join_cluster() below), matching
* that helper's documented contract ("Does NOT set out->from /
* out->key -- caller sets those separately").
*
* @param config Pointer to the xce_client_config_t structure to be initialized.
* @param config Pointer to the xce_client_config_t structure to update.
*/
static void init_client_config(xce_client_config_t* config)
static void set_client_identity(xce_client_config_t* config)
{
int ret;
char local_host[MAXHOSTNAMELEN] = { 0 };
@ -141,9 +164,6 @@ static void init_client_config(xce_client_config_t* config)
strncpy(local_host, "localhost", sizeof(local_host) - 1);
}
config->nmasters = 1;
config->masters[0] = cc_ctx->remote_host;
config->master_ports[0] = cc_ctx->remote_port;
#ifdef VIRTUAL_CLUSTER
config->from = strdup(cc_ctx->virtual_name);
#else
@ -151,163 +171,197 @@ static void init_client_config(xce_client_config_t* config)
#endif
config->key = strdup("keepSecret");
XCECD_LOG(XCE_LOG_DEBUG, "Controller connector configuration initialized.");
XCECD_LOG(XCE_LOG_DEBUG,
"Join cluster configuration - Local host: %s, Remote host: %s, "
"Remote port: %d, Key: %s",
config->from, config->masters[0], config->master_ports[0],
config->key);
}
static void handle_disconnect()
{
XCECD_LOG(XCE_LOG_DEBUG, "Handling disconnect and shutdown XCECD...");
// Shutdown worker and waiting for restart
cc_ctx->running = false;
XCECD_LOG(XCE_LOG_DEBUG, "Disconnect handled.");
"Controller connector identity set: from=%s", config->from);
}
/**
* @brief Handles I/O events for the endpoint, focusing on error conditions
* and connection liveness checks using EPOLLIN/EPOLLOUT.
* @brief Build the single validated candidate list this connect/reconnect
* attempt will use (component-failover-adoption, task 2.1, Req 1
* AC5).
*
* @return int Always returns 0.
* A single controller is not a separately-maintained code path: when
* --peers configured a real candidate list (cc_ctx->cluster_cfg), it is
* used as-is; otherwise a one-candidate list is synthesized from the legacy
* remote_host/remote_port default. Either source is written into the same
* persistent g_cluster_cfg and then run through the identical
* xce_cluster_config_validate() call -- no branch downstream of this
* function distinguishes the two sources.
*
* @return 0 on a validated config, the validator's negative error code
* (or -EINVAL if cc_ctx is unset) otherwise.
*/
int checking_connection()
static int build_cluster_config(void)
{
const int MAX_RETRY_COUNT = 5;
const int RETRY_INTERVAL = 5; // seconds
int retry_count = 0;
int ret;
while (retry_count < MAX_RETRY_COUNT) {
ret = xcecli_hello(xcecli_ctx, "hello");
if (ret == 0) {
// Connection successful
if (retry_count > 0) {
XCECD_LOG(XCE_LOG_DEBUG,
"Connection to controller restored after %d retries",
retry_count);
}
return 0;
}
retry_count++;
XCECD_LOG(XCE_LOG_ERROR,
"Connection to controller is broken. Retry %d/%d failed",
retry_count, MAX_RETRY_COUNT);
if (0 == xcecd_join_cluster(&host_attrs)) {
XCECD_LOG(XCE_LOG_INFO,
"Rejoined cluster successfully after retry %d",
retry_count);
// api_ctx_t* api_ctx = (api_ctx_t*)xcecli_ctx;
// cc_ctx->xce_ctx = &(api_ctx->ctx);
return 0;
}
if (retry_count < MAX_RETRY_COUNT) {
XCECD_LOG(XCE_LOG_INFO, "Retrying connection in %d seconds...",
RETRY_INTERVAL);
sleep(RETRY_INTERVAL);
}
if (!cc_ctx) {
return -EINVAL;
}
// All retries exhausted
XCECD_LOG(XCE_LOG_ERROR,
"Connection to controller failed after %d retries. Handling "
"disconnect...",
MAX_RETRY_COUNT);
handle_disconnect();
if (cc_ctx->cluster_cfg.candidates_count > 0) {
g_cluster_cfg = cc_ctx->cluster_cfg;
} else {
memset(&g_cluster_cfg, 0, sizeof(g_cluster_cfg));
strncpy(g_cluster_cfg.candidates[0].hostname,
cc_ctx->remote_host ? cc_ctx->remote_host : "",
sizeof(g_cluster_cfg.candidates[0].hostname) - 1);
g_cluster_cfg.candidates[0].port = cc_ctx->remote_port;
g_cluster_cfg.candidates_count = 1;
strncpy(g_cluster_cfg.identity,
cc_ctx->virtual_name ? cc_ctx->virtual_name : "xcecd",
sizeof(g_cluster_cfg.identity) - 1);
}
return xce_cluster_config_validate(&g_cluster_cfg);
}
/**
* @brief Rejoin callback registered with the client library
* (component-failover-adoption, task 2.1, Req 2 AC1, Req 5 AC1-2).
*
* The library invokes this while still holding its own per-ctx lock
* (ac->failover.lock, a non-recursive mutex held for the whole
* XCECLI_FAILOVER_CALL retry loop) -- per design.md's "Rejoin callback
* constraint", this callback MUST NOT perform I/O or call any xcecli_*
* function: doing so would try to re-acquire the same already-held lock and
* self-deadlock the calling thread forever. It only sets a flag and returns
* 0 immediately so the interrupted operation resumes normally. Consuming
* this flag to issue the actual xcecli_join() as a separate, later call
* (once the triggering call has returned and the lock is free) is task
* 2.2's job.
*/
static int xcecd_on_reconnect(xcecli_ctx_t ctx, void* arg)
{
(void)ctx;
(void)arg;
needs_rejoin = true;
return 0;
}
/**
* @brief Joins the cluster using the provided host attributes.
* @brief Joins the cluster using the provided host attributes
* (component-failover-adoption, task 2.1, Req 1 AC1-3/AC5, Req 5
* AC1-2).
*
* In HA mode (cc_ctx->cluster_cfg.candidates_count > 0), iterates through
* all candidates until one accepts the join as primary. In single-host mode
* the original path is taken unchanged.
* Builds exactly one validated candidate list (build_cluster_config()),
* converts it via the shared libxce helper, and performs ONE
* init+start+join sequence. The client library's own failover mechanism
* (client_failover.c) rotates across candidates and follows standby
* redirects internally, within the single xcecli_join() call below --
* there is no per-candidate loop here anymore. A single configured
* candidate is simply the candidates_count=1 case of this exact sequence,
* not a separate fallback.
*
* @param attrs Pointer to the host attributes structure.
* @return int 0 on success, -1 on failure.
* @return int 0 on success, negative error code (or -1) on failure.
*/
int xcecd_join_cluster(hostattrs_t* attrs)
{
int rc;
if (!attrs) {
return -1;
}
int role = HostRole_worker;
/* HA mode: try each candidate until we find the primary */
if (cc_ctx != NULL && cc_ctx->cluster_cfg.candidates_count > 0) {
for (int i = 0; i < cc_ctx->cluster_cfg.candidates_count; i++) {
const char* host = cc_ctx->cluster_cfg.candidates[i].hostname;
int port = cc_ctx->cluster_cfg.candidates[i].port;
/* Point cc_ctx at this candidate so init_client_config() picks it up */
cc_ctx->remote_host = (char*)host;
cc_ctx->remote_port = port;
XCECD_LOG(XCE_LOG_INFO,
"xcecd_join_cluster: trying candidate %d: %s:%d",
i, host, port);
init_client_config(&xce_client_cfg);
xcecli_ctx_init(&xcecli_ctx, &xce_client_cfg);
xcecli_set_handler(xcecli_ctx, &myreq_handler, NULL);
int rc = xcecli_start(xcecli_ctx);
if (rc < 0) {
XCECD_LOG(XCE_LOG_WARN,
"xcecd_join_cluster: connect to %s:%d failed (rc=%d)",
host, port, rc);
xcecli_ctx_fini(xcecli_ctx); /* free context before next attempt */
continue;
}
rc = xcecli_join(xcecli_ctx, role, attrs);
if (rc == 0) {
XCECD_LOG(XCE_LOG_INFO,
"xcecd_join_cluster: joined primary at %s:%d",
host, port);
return 0;
}
/* Standby replied — try next candidate */
XCECD_LOG(XCE_LOG_INFO,
"xcecd_join_cluster: candidate %s:%d returned %d "
"(likely standby redirect), trying next",
host, port, rc);
xcecli_stop(xcecli_ctx);
xcecli_ctx_fini(xcecli_ctx); /* free context before next attempt */
}
XCECD_LOG(XCE_LOG_ERROR,
"xcecd_join_cluster: all %d candidates failed",
cc_ctx->cluster_cfg.candidates_count);
return -1;
}
/* Single-host fallback (original path) */
init_client_config(&xce_client_cfg);
xcecli_ctx_init(&xcecli_ctx, &xce_client_cfg);
xcecli_set_handler(xcecli_ctx, &myreq_handler, NULL);
xcecli_start(xcecli_ctx);
int rc = xcecli_join(xcecli_ctx, role, attrs);
rc = build_cluster_config();
if (rc < 0) {
errno = -rc;
XCECD_LOG(XCE_LOG_ERROR, "Call xcecli_join() failure: %s",
strerror(errno));
XCECD_LOG(XCE_LOG_ERROR,
"xcecd_join_cluster: invalid cluster configuration (rc=%d)",
rc);
return rc;
}
XCECD_LOG(XCE_LOG_INFO,
"xcecd_join_cluster: attempting %d candidate(s), first=%s:%d",
g_cluster_cfg.candidates_count,
g_cluster_cfg.candidates[0].hostname,
g_cluster_cfg.candidates[0].port);
rc = xce_cluster_config_to_client_config(&g_cluster_cfg, &xce_client_cfg);
if (rc < 0) {
XCECD_LOG(XCE_LOG_ERROR,
"xcecd_join_cluster: candidate list conversion failed "
"(rc=%d)",
rc);
return rc;
}
set_client_identity(&xce_client_cfg);
rc = xcecli_ctx_init(&xcecli_ctx, &xce_client_cfg);
if (rc < 0) {
XCECD_LOG(XCE_LOG_ERROR,
"xcecd_join_cluster: xcecli_ctx_init failed (rc=%d)", rc);
return rc;
}
xcecli_set_handler(xcecli_ctx, &myreq_handler, NULL);
xcecli_set_reconnect_handler(xcecli_ctx, xcecd_on_reconnect, NULL);
rc = xcecli_start(xcecli_ctx);
if (rc < 0) {
XCECD_LOG(XCE_LOG_WARN,
"xcecd_join_cluster: all %d candidate(s) unreachable "
"(rc=%d)",
g_cluster_cfg.candidates_count, rc);
xcecli_ctx_fini(xcecli_ctx);
xcecli_ctx = NULL;
return rc;
}
rc = xcecli_join(xcecli_ctx, HostRole_worker, attrs);
if (rc < 0) {
errno = -rc;
XCECD_LOG(XCE_LOG_ERROR,
"xcecd_join_cluster: xcecli_join() failed across %d "
"candidate(s): %s",
g_cluster_cfg.candidates_count, strerror(errno));
xcecli_stop(xcecli_ctx);
xcecli_ctx_fini(xcecli_ctx);
xcecli_ctx = NULL;
return rc;
}
XCECD_LOG(XCE_LOG_INFO,
"xcecd_join_cluster: joined cluster (from=%s, %d "
"candidate(s) configured)",
xce_client_cfg.from, g_cluster_cfg.candidates_count);
return 0;
}
/**
* @brief Periodic liveness probe on the connector thread
* (component-failover-adoption, task 2.1, Req 2 AC1-3, Req 3 AC1/AC3).
*
* A single xcecli_hello() call. xcecli_hello() is XCECLI_FAILOVER_CALL-
* wrapped: on a transport failure it already retries indefinitely across
* the full configured candidate list (rotate + backoff) and invokes the
* registered rejoin-flag callback on its own -- the client library now
* owns connection recovery. This probe therefore no longer runs its own
* bounded retry-then-give-up loop: an outage that outlasts the old
* MAX_RETRY_COUNT*RETRY_INTERVAL bound no longer tears the worker's
* connection down for good (cc_ctx->running is never forced false here);
* the connector thread's normal periodic cadence simply gives the library
* another chance on the next call.
*
* @return int Always returns 0.
*/
int checking_connection()
{
int ret = xcecli_hello(xcecli_ctx, "hello");
if (ret == 0) {
return 0;
}
XCECD_LOG(XCE_LOG_ERROR,
"Connection to controller not currently healthy (rc=%d); the "
"client library will keep retrying/rotating candidates in "
"the background",
ret);
return 0;
}
@ -1426,8 +1480,10 @@ bool controller_connector_start(void)
int retry_limit = 0;
static int error_count = 0;
init_client_config(&xce_client_cfg);
/* No standalone client-config construction here anymore -- the single
* synthesize-validate-convert-connect pipeline inside
* xcecd_join_cluster() below builds and sets xce_client_cfg itself on
* every attempt (component-failover-adoption, task 2.1, Req 1 AC5). */
do {
rc = xcecd_join_cluster(&host_attrs);
if (rc < 0) {

24
src/worker/testsuite/meson.build Normal file
View file

@ -0,0 +1,24 @@
# Worker unit test suite (component-failover-adoption, task 2.1). White-box
# tests #include the production .c file(s) under test directly (same pattern
# as scheduler/testsuite's test_handlers.c etc.), so they link the same
# dependency set the real xcecd/xcecd_worker executables use rather than a
# subset of xcecd_srcs -- see test_controller_connector.c's header comment
# for why controller_connector.c is not also listed in xcecd_srcs here (it
# would double-define every symbol in that file).
worker_test_incdir = include_directories('../include', '../../include', '.',
'../src', '../../lib')
worker_test_deps = [xcecd_worker_deps, unity_dep]
controller_connector_test = executable('controller_connector_test',
'test_controller_connector.c',
c_args: ['-DDBUS_MANAGER'],
include_directories: [worker_test_incdir],
dependencies: worker_test_deps,
install: false)
test('worker controller_connector test', controller_connector_test,
suite: 'worker',
timeout: 60,
env: ['ASAN_OPTIONS=detect_leaks=0'])

329
src/worker/testsuite/test_controller_connector.c Normal file
View file

@ -0,0 +1,329 @@
/*
* test_controller_connector.c — white-box tests for the worker's unified
* connect/rejoin pipeline (component-failover-adoption, task 2.1,
* boundary: WorkerControllerConnector).
*
* Pulls the real (mostly-static) connect logic in controller_connector.c
* into this TU (same white-box pattern as scheduler/testsuite/test_handlers.c
* #include-ing handler.c directly), so the tests exercise the actual
* production code, not a re-implementation of it. controller_connector.c's
* job-handling externs this file never exercises (thread_pool, uv_ipc,
* xcecd_worker_utils, xcecd_exdata) are link-satisfied by trivial stubs
* below, same rationale as test_handlers.c's stubs. The stub definitions
* are placed AFTER the #include so their parameter types (thread_pool_t,
* fork_task_t, xcecd_worker_t, res_limits_t) come from the real headers
* controller_connector.c already pulled in -- controller_connector.h's own
* `api_ctx_t` (an unused leftover type, only ever referenced from commented-
* out code) also means this file must NOT additionally include
* src/lib/client_internal.h: that header declares its OWN, differently-
* shaped `api_ctx_t`, and the two conflict in the same translation unit.
*
* Covers:
* - Req 1.5: no-candidate-list startup and a single-entry candidate list
* converge on the identical validated xce_cluster_config_t shape (same
* downstream connect log sequence, since those logs are driven entirely
* by that struct's contents) -- proving there is no separate single-host
* code path anymore.
* - Req 5.1/5.2: a real multi-candidate list passes through unmodified
* (no truncation/branch-specific mangling before the shared helper).
* - Req 5.1/5.2 (rejoin callback contract, design.md "Rejoin callback
* constraint"): the registered rejoin callback only sets a flag and
* returns immediately -- no I/O, no calling back into the client
* library.
* - Req 2.1-2.3: checking_connection()'s old bounded give-up
* (MAX_RETRY_COUNT=5 then handle_disconnect() tearing the worker down
* for good) is gone -- an outage far longer than that old bound must
* never permanently disable the worker, and a single probe must return
* fast (the library, not this probe, now owns retry/backoff).
*/
#include <errno.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
/* Pull the real connect/rejoin logic under test into this TU. */
#include "../src/controller_connector.c"
/* controller_connector.c declares `extern hostattrs_t host_attrs;` in its
* non-XRUN branch (this build); provide the definition normally supplied by
* xcecd.c, which this test does not link. */
hostattrs_t host_attrs = { 0 };
/* ── Link-satisfying stubs for externs controller_connector.c references
* but this file never exercises (job start/control/dispatch paths) ── */
int submit_fork_task(thread_pool_t* pool, fork_task_t* task)
{
(void)pool;
(void)task;
return 0;
}
xcecd_worker_t* find_worker_by_jobid(xcecd_worker_t* worker_list,
uint64_t jobid)
{
(void)worker_list;
(void)jobid;
return NULL;
}
bool parse_limits(const char* exdata, res_limits_t* limits)
{
(void)exdata;
(void)limits;
return true;
}
bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force)
{
(void)job_id;
(void)signal;
(void)force;
return true;
}
bool ipc_master_send_job_ckpt(uint64_t job_id, int kill, const char* method)
{
(void)job_id;
(void)kill;
(void)method;
return true;
}
void xcecd_process_command(void* arg) { (void)arg; }
/* ════════════════════════════════════════════════════════════════════════
* Fixtures
* ════════════════════════════════════════════════════════════════════════ */
void setUp(void)
{
memset(&g_cluster_cfg, 0, sizeof(g_cluster_cfg));
cc_ctx = NULL;
xcecli_ctx = NULL;
needs_rejoin = false;
}
void tearDown(void) { return; }
/* ════════════════════════════════════════════════════════════════════════
* Req 1.5 — unified candidate-list pipeline: no branch for "single host"
* ════════════════════════════════════════════════════════════════════════ */
static void test_build_cluster_config_no_peers_synthesizes_single_candidate(
void)
{
controller_connector_ctx_t fake_cc = { 0 };
fake_cc.remote_host = "localhost";
fake_cc.remote_port = 9990;
fake_cc.cluster_cfg.candidates_count = 0; /* no --peers given */
cc_ctx = &fake_cc;
int rc = build_cluster_config();
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_cluster_cfg.candidates_count);
TEST_ASSERT_EQUAL_STRING("localhost", g_cluster_cfg.candidates[0].hostname);
TEST_ASSERT_EQUAL_INT(9990, g_cluster_cfg.candidates[0].port);
TEST_ASSERT_NOT_EQUAL(0,
g_cluster_cfg.identity[0]); /* validate() requires this */
printf(
"test_build_cluster_config_no_peers_synthesizes_single_candidate: "
"PASS\n");
}
/* The core Req 1.5 assertion: starting with NO --peers and starting with a
* SINGLE-ENTRY --peers list pointed at the exact same host/port must
* produce the identical validated candidate-list shape. Every log line in
* xcecd_join_cluster() is derived from g_cluster_cfg's contents, so
* structural equality here is exactly what "identical connect log
* sequence" (the task's observable-completion criterion) reduces to. */
static void
test_build_cluster_config_single_entry_peers_matches_synthesized_shape(void)
{
controller_connector_ctx_t fake_cc_no_peers = { 0 };
fake_cc_no_peers.remote_host = "localhost";
fake_cc_no_peers.remote_port = 9990;
cc_ctx = &fake_cc_no_peers;
TEST_ASSERT_EQUAL_INT(0, build_cluster_config());
xce_cluster_config_t synthesized = g_cluster_cfg;
controller_connector_ctx_t fake_cc_one_peer = { 0 };
fake_cc_one_peer.remote_host = "unused";
fake_cc_one_peer.remote_port = 1;
fake_cc_one_peer.cluster_cfg.candidates_count = 1;
strncpy(fake_cc_one_peer.cluster_cfg.candidates[0].hostname, "localhost",
XCE_MAXHOSTNAME_LEN - 1);
fake_cc_one_peer.cluster_cfg.candidates[0].port = 9990;
strncpy(fake_cc_one_peer.cluster_cfg.identity, "workertest", 255);
cc_ctx = &fake_cc_one_peer;
TEST_ASSERT_EQUAL_INT(0, build_cluster_config());
xce_cluster_config_t explicit_single = g_cluster_cfg;
TEST_ASSERT_EQUAL_INT(explicit_single.candidates_count,
synthesized.candidates_count);
TEST_ASSERT_EQUAL_STRING(explicit_single.candidates[0].hostname,
synthesized.candidates[0].hostname);
TEST_ASSERT_EQUAL_INT(explicit_single.candidates[0].port,
synthesized.candidates[0].port);
printf(
"test_build_cluster_config_single_entry_peers_matches_synthesized_"
"shape: PASS\n");
}
static void test_build_cluster_config_multi_candidate_passthrough(void)
{
controller_connector_ctx_t fake_cc = { 0 };
fake_cc.cluster_cfg.candidates_count = 3;
strncpy(fake_cc.cluster_cfg.candidates[0].hostname, "ctrl-a",
XCE_MAXHOSTNAME_LEN - 1);
fake_cc.cluster_cfg.candidates[0].port = 9991;
strncpy(fake_cc.cluster_cfg.candidates[1].hostname, "ctrl-b",
XCE_MAXHOSTNAME_LEN - 1);
fake_cc.cluster_cfg.candidates[1].port = 9992;
strncpy(fake_cc.cluster_cfg.candidates[2].hostname, "ctrl-c",
XCE_MAXHOSTNAME_LEN - 1);
fake_cc.cluster_cfg.candidates[2].port = 9993;
strncpy(fake_cc.cluster_cfg.identity, "workertest", 255);
cc_ctx = &fake_cc;
int rc = build_cluster_config();
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(3, g_cluster_cfg.candidates_count);
TEST_ASSERT_EQUAL_STRING("ctrl-a", g_cluster_cfg.candidates[0].hostname);
TEST_ASSERT_EQUAL_INT(9991, g_cluster_cfg.candidates[0].port);
TEST_ASSERT_EQUAL_STRING("ctrl-b", g_cluster_cfg.candidates[1].hostname);
TEST_ASSERT_EQUAL_INT(9992, g_cluster_cfg.candidates[1].port);
TEST_ASSERT_EQUAL_STRING("ctrl-c", g_cluster_cfg.candidates[2].hostname);
TEST_ASSERT_EQUAL_INT(9993, g_cluster_cfg.candidates[2].port);
printf("test_build_cluster_config_multi_candidate_passthrough: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Rejoin callback contract (design.md "Rejoin callback constraint",
* Req 5.1/5.2): flag-only, no I/O, returns immediately.
* ════════════════════════════════════════════════════════════════════════ */
static long elapsed_us(struct timespec* t0, struct timespec* t1)
{
return (t1->tv_sec - t0->tv_sec) * 1000000L
+ (t1->tv_nsec - t0->tv_nsec) / 1000L;
}
static void test_rejoin_callback_sets_flag_and_returns_immediately(void)
{
struct timespec t0, t1;
TEST_ASSERT_FALSE(needs_rejoin);
clock_gettime(CLOCK_MONOTONIC, &t0);
int rc = xcecd_on_reconnect((xcecli_ctx_t)0x1, NULL);
clock_gettime(CLOCK_MONOTONIC, &t1);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_TRUE(needs_rejoin);
TEST_ASSERT_LESS_THAN_INT(5000, elapsed_us(&t0, &t1)); /* < 5ms: no I/O */
printf("test_rejoin_callback_sets_flag_and_returns_immediately: PASS\n");
}
static void test_rejoin_callback_registered_via_public_api(void)
{
/* Register through the real public entry point the production code
* calls (xcecli_set_reconnect_handler) against a REAL ctx allocated by
* xcecli_ctx_init() -- which does local struct setup only, no socket
* I/O -- so this proves the registration call in xcecd_join_cluster()
* is wired correctly end to end (right function, right signature),
* not just that xcecd_on_reconnect() behaves correctly in isolation. */
xce_client_config_t cfg = { 0 };
xcecli_ctx_t ctx = NULL;
cfg.nmasters = 1;
cfg.masters[0] = "127.0.0.1";
cfg.master_ports[0] = 1;
cfg.from = "test";
cfg.key = "test";
TEST_ASSERT_EQUAL_INT(0, xcecli_ctx_init(&ctx, &cfg));
int src = xcecli_set_reconnect_handler(ctx, xcecd_on_reconnect, NULL);
TEST_ASSERT_EQUAL_INT(0, src);
xcecli_ctx_fini(ctx);
printf("test_rejoin_callback_registered_via_public_api: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Give-up reconciliation (Req 2.1-2.3): checking_connection() must never
* permanently disable the worker; the library now owns retry/backoff.
* ════════════════════════════════════════════════════════════════════════ */
static void test_checking_connection_never_permanently_disables_worker(void)
{
controller_connector_ctx_t fake_cc = { 0 };
fake_cc.running = true;
cc_ctx = &fake_cc;
/* No connection at all -> xcecli_hello(NULL, ...) fails immediately
* with -EINVAL (checked before any I/O -- see client.c's xcecli_hello),
* so this loop is fast and deterministic without a real socket. */
xcecli_ctx = NULL;
struct timespec t0, t1;
clock_gettime(CLOCK_MONOTONIC, &t0);
/* Far more iterations than the OLD MAX_RETRY_COUNT (5) -- the old code
* tore the worker down for good (cc_ctx->running = false) once, inside
* a SINGLE checking_connection() call, after 5 consecutive failures. */
for (int i = 0; i < 25; i++) {
checking_connection();
TEST_ASSERT_TRUE_MESSAGE(
cc_ctx->running,
"checking_connection() must never permanently disable the "
"worker -- the client library now owns connection recovery");
}
clock_gettime(CLOCK_MONOTONIC, &t1);
long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000L
+ (t1.tv_nsec - t0.tv_nsec) / 1000000L;
/* The OLD code's single checking_connection() call, against a
* persistently failing hello, looped 5 times with a 5s sleep between
* each -- >=20s for ONE call alone. 25 calls completing well under a
* second proves the bounded retry-then-sleep loop is gone, not merely
* that the give-up flag was removed. */
TEST_ASSERT_LESS_THAN_INT(1000, elapsed_ms);
printf(
"test_checking_connection_never_permanently_disables_worker: "
"PASS (elapsed_ms=%ld)\n",
elapsed_ms);
}
/* ════════════════════════════════════════════════════════════════════════
* main
* ════════════════════════════════════════════════════════════════════════ */
int main(void)
{
xce_init_xcecd_log("/tmp/test_controller_connector.log", 1024 * 1024, 2);
UNITY_BEGIN();
RUN_TEST(test_build_cluster_config_no_peers_synthesizes_single_candidate);
RUN_TEST(
test_build_cluster_config_single_entry_peers_matches_synthesized_shape);
RUN_TEST(test_build_cluster_config_multi_candidate_passthrough);
RUN_TEST(test_rejoin_callback_sets_flag_and_returns_immediately);
RUN_TEST(test_rejoin_callback_registered_via_public_api);
RUN_TEST(test_checking_connection_never_permanently_disables_worker);
return UNITY_END();
}