From ce9f734a380dd35d1930b5fc332e5b3deac585df Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 11 Aug 2026 05:42:48 +0000 Subject: [PATCH] feat(component-failover-adoption): unify worker's dual connect paths into single pipeline 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. --- .../component-failover-adoption/tasks.md | 3 +- src/tools/xce-build.sh | 9 +- src/worker/meson.build | 2 + src/worker/src/controller_connector.c | 330 ++++++++++-------- src/worker/testsuite/meson.build | 24 ++ .../testsuite/test_controller_connector.c | 329 +++++++++++++++++ 6 files changed, 556 insertions(+), 141 deletions(-) create mode 100644 src/worker/testsuite/meson.build create mode 100644 src/worker/testsuite/test_controller_connector.c diff --git a/.kiro/specs/component-failover-adoption/tasks.md b/.kiro/specs/component-failover-adoption/tasks.md index d0ceb746..64a03a22 100644 --- a/.kiro/specs/component-failover-adoption/tasks.md +++ b/.kiro/specs/component-failover-adoption/tasks.md @@ -24,7 +24,7 @@ - _Depends: 1.2_ - [ ] 2. Worker: unified candidate-list connect and transparent reconnect -- [ ] 2.1 (P) Replace worker's dual connect mechanisms with the single unified pipeline +- [x] 2.1 (P) Replace worker's dual connect mechanisms with the single unified pipeline - Delete the standalone client-config construction used for worker's default connection and the separate single-host branch used when no candidate list is configured — both bypass the shared candidate-list validation and conversion helper entirely today - Build exactly one validated candidate list before connecting, whether it came from an operator-supplied peer list or was synthesized as a single candidate from the existing default when none was given, then run it through the shared helper and connect once - Register a rejoin callback that only sets a flag and returns immediately — it must not call back into the client library itself, since the library invokes it while still holding its own lock @@ -164,3 +164,4 @@ - Task 1.1 follow-up resolved in 1.2: the `>3`→`>MAX_HA_CANDIDATES` rename initially proposed for `xce_cluster_config_validate()`'s WARN check turned out to be dead code — after task 1.1's constant reconciliation, that check tests the identical field against the identical constant as an earlier `-EINVAL` return in the same function, so it can never fire. Deleted rather than renamed (with the matching stale doc-comment line also removed), caught by task 1.2's review round 1. - Task 1.2: its 8 new tests in `test_cluster.c` substantively cover task 1.3's stated checklist already (candidate order, one-candidate-no-branch, truncation-with-warning, NULL/empty rejection) — task 1.3 should scope down to just the constant-equality assertion (`N_MAX_MASTERS == MAX_HA_CANDIDATES`) rather than duplicating this coverage. - Task 1.3: closed with zero new code. Verified fresh (source read + a clean `xce-build.sh lib asan` run) that every item in its checklist was already satisfied by prior tasks: the `_Static_assert(N_MAX_MASTERS == MAX_HA_CANDIDATES, ...)` landed in task 1.1 (`test_libxce_retry.c:44`), and the four behavioral cases (order/no-branch/truncation/NULL-empty) landed in task 1.2's 8 tests (`test_cluster.c`). No implementer/reviewer cycle was dispatched since there was no diff to review — dispatching one would have either duplicated existing test names (build conflict) or been a no-op review of an empty diff. +- Task 2.1: `xce_cluster_config_to_client_config()` (task 1.2) aliases `out->masters[i]` directly to `cfg->candidates[i].hostname` — no `strdup`. Any caller building the source `xce_cluster_config_t` on the stack and passing it through the helper then using the resulting client config beyond that stack frame's lifetime has a use-after-scope bug. Worker's fix keeps the source config in a file-scope `static` for this reason; tasks 3.1 (scheduler) and 4.2 (apiserver) hit the identical hazard and must do the same (or otherwise guarantee the source config outlives the client config's use). diff --git a/src/tools/xce-build.sh b/src/tools/xce-build.sh index 53e44ff1..cd964900 100755 --- a/src/tools/xce-build.sh +++ b/src/tools/xce-build.sh @@ -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 diff --git a/src/worker/meson.build b/src/worker/meson.build index 2af1b1af..f090e9e2 100644 --- a/src/worker/meson.build +++ b/src/worker/meson.build @@ -103,3 +103,5 @@ xcecd_virtual = executable('xcecd_virtual', include_directories: [incdir], dependencies:[xcecd_deps], install:true) + +subdir('testsuite') diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 96a606b9..ac0365c5 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -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) { diff --git a/src/worker/testsuite/meson.build b/src/worker/testsuite/meson.build new file mode 100644 index 00000000..9322317e --- /dev/null +++ b/src/worker/testsuite/meson.build @@ -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']) diff --git a/src/worker/testsuite/test_controller_connector.c b/src/worker/testsuite/test_controller_connector.c new file mode 100644 index 00000000..7ef424ad --- /dev/null +++ b/src/worker/testsuite/test_controller_connector.c @@ -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 +#include +#include +#include +#include +#include + +#include + +/* 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(); +}