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

feat(worker-rejoin-resync): task 4.3 - wire query receipt and post-join gate

Browse source 
Add OPCode_jobStateQuery dispatch to myreq_handler with multi-page query
reassembly, routing to task 4.2's xcecd_resync_answer_query() (Req
8.1, 8.3). Gate the worker's resumption of normal post-join operation
on a resync_pending flag until the full report has sent (Req 3.8).

Fixes bug-1749 (xcecd_resync.c was never added to the worker's
production source lists - task 4.2's build only ever exercised it via
the test binary's white-box include).

Round 1 review caught bug-1750: the gate sat as checking_connection()'s
entry guard, so a connection drop mid multi-page query left
resync_pending stuck forever, permanently disabling the worker's only
reconnect mechanism (Req 6.1 violation). Round 2 narrows the gate to
guard only the rejoin action, and resets it from xcecd_on_reconnect()
so a stalled query can never wedge it past one reconnect cycle.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
This commit is contained in:
Rongsong Shen 2026-09-01 18:01:03 +00:00
parent cfe58c3dcf
commit 5b69973677
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
8 changed files with 1161 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

103
src/worker/include/xcecd_resync.h
View file

@ -1,6 +1,7 @@
#ifndef XCECD_RESYNC_H
#define XCECD_RESYNC_H
#include <stdbool.h>
#include <stdint.h>
#include "netio.h"
@ -15,21 +16,19 @@
* additional job the worker is currently running or has recently finished
* that was NOT part of the query (Requirement 3.6's "extras").
*
* Boundary/scope: this module implements the pure classify/build/send
* logic only.
* - It does NOT wire OPCode_jobStateQuery's receipt into the worker's
* dispatch loop (controller_connector.c's myreq_handler switch) --
* that opcode-dispatch wiring is not claimed by any task in
* tasks.md's current text (task 6.1 is controller-side send +
* config-window delivery; task 6.2 is controller-side report
* ingestion). Whatever task performs that wiring must also
* reassemble a query that itself arrives across multiple incoming
* pages (Req 8.1) into one flat jobids array before calling
* xcecd_resync_answer_query() below -- this module takes an
* already-assembled list, it does not itself track query pages.
* - It does NOT implement the Requirement 3.8 "resync_pending" gate
* (task 4.3's boundary, same file) that delays the worker's own
* post-join operation until the report has fully sent.
* task 4.3 (Requirements 3.8, 8.1, 8.3) adds the actual wire-receipt entry
* point, xcecd_resync_handle_query_page() below: it is what
* controller_connector.c's myreq_handler dispatch case for
* OPCode_jobStateQuery calls, once per incoming query page, and it owns
* both the multi-page query reassembly (a query may itself arrive across
* several pages, Requirement 8.1/8.3) and the Requirement 3.8
* `resync_pending` gate (mirroring controller_connector.c's existing
* `needs_rejoin` static-bool pattern) that delays the worker's own
* resumption of normal post-join operation until the full report --
* queried and extra pages alike -- has finished sending.
* xcecd_resync_answer_query() itself remains the pure classify/build/send
* primitive task 4.2 left it as; it takes an already-assembled jobids
* array and knows nothing about the gate or multi-page query receipt.
*/
/* Classify one job ID against the worker's live running list, falling
@ -93,4 +92,78 @@ int xcecd_resync_answer_query(xcecd_worker_t* worker_list, endpoint_t* ep,
xce_peerinfo_t* from, const uint64_t* jobids,
int njobids, uint64_t now_ms, int pktlimit);
/*
* worker-rejoin-resync task 4.3 (Requirements 3.8, 8.1, 8.3): the real
* OPCode_jobStateQuery receive-side entry point. Called once per incoming
* query page from controller_connector.c's myreq_handler dispatch case,
* with `page` the just-decoded xce_job_state_query_t for that one page
* (its own {n_total, pos, count, jobids} slice, the same pagination idiom
* xce_job_state_query_req() -- the sender side, task 1.4 -- uses).
*
* Reassembles however many pages the query itself arrives in into one flat
* jobids array (a page with pos == 0 starts a fresh accumulation, discarding
* any prior incomplete one -- e.g. left over from a worker-side disconnect
* mid-query; only one reconciliation round is ever outstanding on a given
* worker's single long-lived controller connection at a time, per
* myreq_handler's own synchronous per-message dispatch). Once the last page
* has arrived, calls xcecd_resync_answer_query() above with the fully
* assembled list.
*
* Also owns the Requirement 3.8 `resync_pending` gate (mirroring
* controller_connector.c's existing `needs_rejoin` static-bool pattern,
* queried via xcecd_resync_is_pending() below): set the moment the first
* page of a query is received (before this page is even fully validated),
* held set across every remaining query page and every page of the report
* this function goes on to send, and cleared only once
* xcecd_resync_answer_query() has returned -- on success OR failure, so a
* transport error can never leave the worker's own periodic post-join
* activity gated forever (the controller's own bounded reconciliation
* timeout, Requirement 5.3, is what resolves an incomplete round from its
* side; this is purely a local, non-blocking gate on the worker's own
* follow-on behavior -- not a new wire-level ack, not an extension of
* xce_join()'s blocking wait, per design.md's Boundary Commitments).
*
* Returns 0 while still waiting for more query pages (report not sent
* yet), 1 once this query has been fully answered (report sent, whether
* or not the send itself succeeded), or a negative errno for a malformed
* page (NULL page/ep/from, or a page whose pos/count fall outside its own
* declared n_total) or allocation failure -- the gate and any in-progress
* accumulator are both cleared/discarded in that case too, never left
* stuck.
*/
int xcecd_resync_handle_query_page(xcecd_worker_t* worker_list, endpoint_t* ep,
xce_peerinfo_t* from,
const xce_job_state_query_t* page,
uint64_t now_ms, int pktlimit);
/* Requirement 3.8: true from the moment a job-state query's first page is
* received until the full report (queried and extra pages alike) has
* finished sending -- see xcecd_resync_handle_query_page() above. Checked
* by controller_connector.c's checking_connection() before it resumes the
* needs_rejoin-triggered rejoin action -- NOT before its own liveness probe
* (xcecli_hello()) or the needs_rejoin check itself, both of which run every
* tick unconditionally so the worker's own reconnect-detection can never be
* gated off (Requirement 6.1). */
bool xcecd_resync_is_pending(void);
/*
* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8,
* 6.1): unconditionally clears resync_pending and discards any in-progress
* multi-page query accumulation. xcecd_resync_handle_query_page() can only
* clear the gate itself by being called again with a query's LAST page --
* if the worker's connection drops after a non-final page arrived, no
* further page will ever come, and resync_pending would otherwise stay
* stuck true forever with nothing left to clear it.
*
* Call this the moment the worker itself detects it has reconnected --
* controller_connector.c's xcecd_on_reconnect() (the client library's
* transparent-reconnect callback, the same signal that sets needs_rejoin)
* is that moment: the connection any in-flight query arrived on is already
* gone, and the controller's own Requirement 6.1 Acceptance Criterion 1
* discards that incomplete round and re-initiates a fresh query on this
* worker's next rejoin, so nothing here is worth preserving. Safe to call
* from that exact site: it touches only this module's own statics (a bool
* and a small heap buffer), no I/O, no xcecli_* call. */
void xcecd_resync_reset(void);
#endif /* XCECD_RESYNC_H */

2
src/worker/meson.build
View file

@ -7,6 +7,7 @@ xcecd_worker_srcs = [
'src/static_array.c',
'src/xcecd_cgroups.c',
'src/xcecd_worker_utils.c',
'src/xcecd_resync.c',
'src/controller_connector.c',
'src/xcecd_dbus_manager.c',
'src/xcecd_db.c',
@ -84,6 +85,7 @@ xcecd_srcs = [
'src/controller_connector.c',
'src/resource_sync.c',
'src/xcecd_worker_utils.c',
'src/xcecd_resync.c',
'src/hardware_monitor.c',
'src/node_information.c',
'src/job_information.c'

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

@ -4,6 +4,7 @@
#include "uv_ipc.h"
#include "xcecd_cgroups.h"
#include "xcecd_exdata.h"
#include "xcecd_resync.h"
#include "xcecd_worker_utils.h"
#include "xcelog.h"
#include <errno.h>
@ -270,11 +271,26 @@ static void format_candidate_list(char* buf, size_t bufsz)
* 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.
* self-deadlock the calling thread forever. It only sets flags/statics and
* returns 0 immediately so the interrupted operation resumes normally.
* Consuming needs_rejoin 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.
*
* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8,
* 6.1): also resets xcecd_resync.c's resync_pending gate here. This is the
* moment the worker itself learns its connection to the controller was
* re-established -- any job-state query it was still assembling (or report
* it had not finished sending) belongs to the connection that just died and
* can never complete; the controller's own Requirement 6.1 Acceptance
* Criterion 1 already discards that incomplete round and re-initiates a
* fresh query on this worker's next rejoin. Without this reset,
* resync_pending can only be cleared by xcecd_resync_handle_query_page()
* seeing a query's LAST page -- which never arrives on a dead connection --
* leaving it stuck true forever (round 1's confirmed defect). Calling
* xcecd_resync_reset() here is safe under the constraint above: it only
* touches xcecd_resync.c's own statics (a bool and a small heap buffer), no
* I/O and no xcecli_* call.
*/
static int xcecd_on_reconnect(xcecli_ctx_t ctx, void* arg)
{
@ -282,6 +298,7 @@ static int xcecd_on_reconnect(xcecli_ctx_t ctx, void* arg)
(void)arg;
needs_rejoin = true;
xcecd_resync_reset();
return 0;
}
@ -464,6 +481,25 @@ static int (*rejoin_call)(void) = perform_pending_rejoin;
* index) but the old `rc == 0` check logged "rejoin ... failed" on every
* subsequent liveness probe.
*
* worker-rejoin-resync task 4.3 (Requirement 3.8), narrowed per round-2
* remediation: xcecd_resync_is_pending() gates ONLY the needs_rejoin-
* triggered rejoin_call() below -- the "resume normal post-join operation"
* step Requirement 3.8 actually names -- not this function's entry.
* xcecli_hello() and the needs_rejoin check itself both still run every
* tick unconditionally, so checking_connection() keeps the exact
* self-healing property needs_rejoin already had before this task: a fresh
* chance every cycle to detect and recover the worker's own connection
* (src/lib/xcectx.c's handle_disconnect() -- "so that checking_connection
* can detect the failure and attempt to rejoin"). A resync_pending stuck
* true (round 1's confirmed defect: gating this function's entry made the
* worker's sole connection-recovery mechanism permanently inert, breaking
* Requirement 6.1's "disconnect during an incomplete reconciliation round"
* case) can therefore never disable this probe. In practice resync_pending
* is also cleared the instant a reconnect is detected
* (xcecd_on_reconnect() -> xcecd_resync_reset(), above), so this check is a
* defensive backstop against the rejoin racing an in-flight report, not the
* mechanism that unsticks a stalled one.
*
* @return int Always returns 0.
*/
int checking_connection()
@ -479,18 +515,24 @@ int checking_connection()
}
if (needs_rejoin) {
int rc = rejoin_call();
if (rc >= 0) {
needs_rejoin = false;
XCECD_LOG(XCE_LOG_INFO,
"checking_connection: rejoin after transparent "
"reconnect succeeded");
if (xcecd_resync_is_pending()) {
XCECD_LOG(XCE_LOG_DEBUG,
"checking_connection: deferring rejoin -- job-state "
"resync report still being sent");
} else {
XCECD_LOG(XCE_LOG_WARN,
"checking_connection: rejoin after transparent "
"reconnect failed (rc=%d); will retry on the next "
"liveness probe",
rc);
int rc = rejoin_call();
if (rc >= 0) {
needs_rejoin = false;
XCECD_LOG(XCE_LOG_INFO,
"checking_connection: rejoin after transparent "
"reconnect succeeded");
} else {
XCECD_LOG(XCE_LOG_WARN,
"checking_connection: rejoin after transparent "
"reconnect failed (rc=%d); will retry on the next "
"liveness probe",
rc);
}
}
}
@ -1554,6 +1596,73 @@ static int handle_jobcontrol(xce_request_t* request)
return 0;
}
/* worker-rejoin-resync task 4.3 (Req 8.2/8.3 precedent): page-size cap for
* this worker's own OPCode_jobStateReport sends, matching the pktlimit
* value apiserver.c's own xce_push_data_req() call sites already use for
* the same {pos,count,n_total} pagination idiom. */
#define XCECD_RESYNC_REPORT_PKTLIMIT 100
/* Wall-clock (epoch) milliseconds -- same rationale and implementation as
* xcecd_worker_utils.c's private finished_job_wallclock_now_ms() (not
* reusable here: internal linkage, separate translation unit). Needed here
* because xcecd_resync.c's functions all take now_ms as a caller-captured
* parameter rather than sampling the clock themselves (task 4.1's own
* convention -- deterministic values in tests, real wall-clock in
* production), and this dispatch case is that production caller. */
static uint64_t resync_wallclock_now_ms(void)
{
struct timespec ts;
if (clock_gettime(CLOCK_REALTIME, &ts) != 0) {
return (uint64_t)time(NULL) * 1000ULL;
}
return (uint64_t)ts.tv_sec * 1000ULL + (uint64_t)ts.tv_nsec / 1000000ULL;
}
/**
* @brief Handles one incoming OPCode_jobStateQuery page from the controller
* (worker-rejoin-resync task 4.3, Requirements 3.8, 8.1, 8.3).
*
* Delegates to xcecd_resync_handle_query_page() (xcecd_resync.c), which
* reassembles a query that itself spans multiple incoming pages and, once
* the last page has arrived, answers it with a paginated
* OPCode_jobStateReport -- see that function's own header comment
* (xcecd_resync.h) for the resync_pending gate it also owns.
*
* @param request Pointer to the xce_request_t structure containing the
* request data.
* @return int 0 on success (whether or not this page completed the query),
* negative on error.
*/
static int handle_jobstatequery(xce_request_t* request)
{
endpoint_t* ep = xcecli_get_my_endpoint(xcecli_ctx);
xce_peerinfo_t* from = xcecli_get_myinfo(xcecli_ctx);
xcecd_worker_t* worker_list
= (cc_ctx != NULL) ? (xcecd_worker_t*)*cc_ctx->worker_list : NULL;
int rc;
if (ep == NULL || from == NULL) {
XCECD_LOG(XCE_LOG_ERROR,
"handle_jobstatequery: no active controller connection");
return -EINVAL;
}
rc = xcecd_resync_handle_query_page(
worker_list, ep, from, request->body.jobstatequerymsg,
resync_wallclock_now_ms(), XCECD_RESYNC_REPORT_PKTLIMIT);
if (rc < 0) {
XCECD_LOG(XCE_LOG_ERROR,
"handle_jobstatequery: failed to process query page "
"(rc=%d)",
rc);
return rc;
}
return 0;
}
/**
* @brief Handles incoming requests from the controller.
*
@ -1579,6 +1688,9 @@ static int myreq_handler(xce_request_t* request, void* args)
case OPCode_fetchfile:
ret = handle_fetchfile(request);
break;
case OPCode_jobStateQuery:
ret = handle_jobstatequery(request);
break;
default:
break;
}

131
src/worker/src/xcecd_resync.c
View file

@ -216,3 +216,134 @@ int xcecd_resync_answer_query(xcecd_worker_t* worker_list, endpoint_t* ep,
return rc;
}
/* worker-rejoin-resync task 4.3 (Requirements 3.8, 8.1, 8.3). See
* xcecd_resync.h for the full contract of both symbols below.
*
* `resync_pending` mirrors controller_connector.c's `needs_rejoin` exactly:
* a single file-scope static bool, set by the event it gates on, read
* through a small accessor, cleared once the gated work is done. */
static bool resync_pending = false;
/* Accumulates one incoming query's pages (Requirement 8.1/8.3) between
* calls to xcecd_resync_handle_query_page() -- reset whenever a page with
* pos == 0 arrives, so a stale/incomplete accumulation from an aborted
* prior query (e.g. the worker itself disconnected mid-query) is always
* discarded by the next query's first page rather than corrupting it. */
static uint64_t* query_accum = NULL;
static int query_accum_ntotal = 0;
static int query_accum_received = 0;
static void reset_query_accum(void)
{
free(query_accum);
query_accum = NULL;
query_accum_ntotal = 0;
query_accum_received = 0;
}
bool xcecd_resync_is_pending(void) { return resync_pending; }
/* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8/6.1):
* bounds resync_pending's worst case. xcecd_resync_handle_query_page() below
* can only clear the gate by being called again with a query's LAST page --
* if the worker's connection to the controller drops after a non-final page
* has already arrived, no further page will ever arrive on that now-dead
* connection, and resync_pending would otherwise stay stuck true forever
* with nothing left to clear it. That was round 1's confirmed defect: it
* permanently disabled checking_connection()'s entry-guard-gated body,
* including the worker's own liveness/reconnect-detection -- the exact
* mechanism src/lib/xcectx.c's handle_disconnect() documents as the sole
* recovery path ("so that checking_connection can detect the failure and
* attempt to rejoin").
*
* The worker's own reconnect-detection (xcecd_on_reconnect(),
* controller_connector.c -- fired by the client library the moment it
* transparently reconnects) is exactly the signal that makes any such
* in-flight query moot: the connection it arrived on is already gone, and
* the controller's own Requirement 6.1 Acceptance Criterion 1 already
* discards that incomplete reconciliation round and re-initiates a
* brand-new query (starting again at pos == 0) on this worker's next
* rejoin. There is nothing worth preserving from the old, abandoned
* accumulation -- discard it and clear the gate immediately, the same way
* a malformed page already does above. Called from controller_connector.c's
* xcecd_on_reconnect() (see that function's own comment for why this call
* site is safe under its "flag-only, no I/O, no xcecli_*" constraint: this
* function only touches this module's own statics -- a bool and a small
* heap buffer -- no socket/file I/O and no call back into the client
* library). */
void xcecd_resync_reset(void)
{
reset_query_accum();
resync_pending = false;
}
int xcecd_resync_handle_query_page(xcecd_worker_t* worker_list, endpoint_t* ep,
xce_peerinfo_t* from,
const xce_job_state_query_t* page,
uint64_t now_ms, int pktlimit)
{
int rc;
if (page == NULL || ep == NULL || from == NULL) {
return -EINVAL;
}
/* Requirement 3.8: set the moment a query is received -- the first
* page, even before this page (or any later one) is validated, so the
* worker's own periodic post-join activity never interleaves with any
* part of receiving/answering this exchange. Setting it again on every
* later page of the same query is harmless (already true). */
resync_pending = true;
if (page->pos == 0) {
reset_query_accum();
if (page->n_total > 0) {
query_accum = (uint64_t*)calloc((size_t)page->n_total,
sizeof(*query_accum));
if (query_accum == NULL) {
resync_pending = false;
return -ENOMEM;
}
}
query_accum_ntotal = page->n_total;
}
if (page->count > 0) {
if (page->jobids == NULL || query_accum == NULL || page->pos < 0
|| page->count < 0
|| (page->pos + page->count) > query_accum_ntotal) {
reset_query_accum();
resync_pending = false;
return -EINVAL;
}
memcpy(&query_accum[page->pos], page->jobids,
(size_t)page->count * sizeof(*query_accum));
query_accum_received += page->count;
}
if (query_accum_received < query_accum_ntotal) {
/* More pages of this query still expected -- gate stays set,
* nothing sent yet. */
return 0;
}
/* Last page: the query is fully assembled -- answer it for real. */
rc = xcecd_resync_answer_query(worker_list, ep, from, query_accum,
query_accum_ntotal, now_ms, pktlimit);
reset_query_accum();
/* Requirement 3.8: clear only once the full report -- queried and
* extra pages alike -- has finished sending, whether or not the send
* itself succeeded. A transport error must not leave the worker's own
* periodic activity gated forever; the controller's own bounded
* reconciliation timeout (Requirement 5.3) resolves an incomplete
* round from its side, and this gate is purely local/non-blocking. */
resync_pending = false;
return (rc < 0) ? rc : 1;
}

417
src/worker/testsuite/test_controller_connector.c
View file

@ -134,6 +134,53 @@ bool ipc_master_send_job_ckpt(uint64_t job_id, int kill, const char* method)
void xcecd_process_command(void* arg) { (void)arg; }
/* worker-rejoin-resync task 4.3 (Requirements 3.8, 8.1, 8.3): link-
* satisfying stubs for xcecd_resync.c's public API -- this file's white-box
* target is controller_connector.c only, it does not #include
* xcecd_resync.c (that module's own resync_pending lifecycle and
* multi-page query reassembly are covered against the real production
* code by test_xcecd_resync.c). g_test_resync_pending lets
* checking_connection()'s gate-check tests below control what the gate
* reads without a real xcecd_resync.c backing it; the call-count/last-page
* globals let the dispatch-wiring tests below observe that myreq_handler
* actually reaches this function, without needing a real report to be
* built and sent. */
static bool g_test_resync_pending = false;
static int g_resync_handle_query_page_call_count = 0;
static const xce_job_state_query_t* g_resync_handle_query_page_last_page = NULL;
/* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8,
* 6.1): lets the tests below observe that xcecd_on_reconnect() actually
* calls xcecd_resync_reset() (the round-2 fix that bounds a stalled
* multi-page query's worst case), and lets a deferred-rejoin test simulate
* the gate clearing on its own, independent of checking_connection() ever
* touching it directly. */
static int g_resync_reset_call_count = 0;
bool xcecd_resync_is_pending(void) { return g_test_resync_pending; }
void xcecd_resync_reset(void)
{
g_resync_reset_call_count++;
g_test_resync_pending = false;
}
int xcecd_resync_handle_query_page(xcecd_worker_t* worker_list, endpoint_t* ep,
xce_peerinfo_t* from,
const xce_job_state_query_t* page,
uint64_t now_ms, int pktlimit)
{
(void)worker_list;
(void)ep;
(void)from;
(void)now_ms;
(void)pktlimit;
g_resync_handle_query_page_call_count++;
g_resync_handle_query_page_last_page = page;
return (page != NULL) ? 1 : -EINVAL;
}
/* ════════════════════════════════════════════════════════════════════════
* Fixtures
* ════════════════════════════════════════════════════════════════════════ */
@ -147,6 +194,13 @@ void setUp(void)
/* Restore the real production call by default (task 2.2) -- individual
* rejoin-retry tests below override this to a controllable double. */
rejoin_call = perform_pending_rejoin;
/* worker-rejoin-resync task 4.3: default the resync gate to "not
* pending" and reset the dispatch-wiring spy globals. */
g_test_resync_pending = false;
g_resync_handle_query_page_call_count = 0;
g_resync_handle_query_page_last_page = NULL;
g_resync_reset_call_count = 0;
}
void tearDown(void) { return; }
@ -276,6 +330,35 @@ static void test_rejoin_callback_sets_flag_and_returns_immediately(void)
printf("test_rejoin_callback_sets_flag_and_returns_immediately: PASS\n");
}
/* worker-rejoin-resync task 4.3 remediation (round 2, Requirements 3.8,
* 6.1): the reconnect callback must also reset xcecd_resync.c's gate, not
* just set needs_rejoin -- this is what bounds a stalled multi-page
* query's worst case (see xcecd_resync_reset()'s own doc comment). Without
* this call, a query whose connection dropped mid-reassembly would leave
* resync_pending stuck true forever with nothing left to clear it. */
static void test_rejoin_callback_also_resets_resync_gate(void)
{
g_test_resync_pending = true;
g_resync_reset_call_count = 0;
int rc = xcecd_on_reconnect((xcecli_ctx_t)0x1, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_TRUE(needs_rejoin);
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, g_resync_reset_call_count,
"xcecd_on_reconnect() must call xcecd_resync_reset() exactly once "
"so a resync gate left stuck by a stalled multi-page query is "
"always cleared the instant the worker detects it has "
"reconnected");
TEST_ASSERT_FALSE_MESSAGE(
g_test_resync_pending,
"the resync gate must actually be cleared, not merely have its "
"reset function called");
printf("test_rejoin_callback_also_resets_resync_gate: PASS\n");
}
static void test_rejoin_callback_registered_via_public_api(void)
{
/* Register through the real public entry point the production code
@ -558,6 +641,267 @@ test_checking_connection_rejoin_retried_on_next_probe_after_failure(void)
"failure: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 4.3 (worker-rejoin-resync, Requirements 3.8, 8.1, 8.3) — gating the
* needs_rejoin-triggered rejoin action on xcecd_resync.c's resync_pending
* flag (the "resume normal post-join operation" step Requirement 3.8
* actually names), and wiring OPCode_jobStateQuery's real receipt.
*
* Round-2 remediation note: round 1 gated checking_connection()'s ENTRY on
* resync_pending, which also silently skipped xcecli_hello() and the
* needs_rejoin check itself -- confirmed to permanently disable the
* worker's own connection-recovery mechanism whenever resync_pending got
* stuck true (a stalled multi-page query, e.g.). The gate now sits INSIDE
* the needs_rejoin branch, guarding only rejoin_call(); see
* test_checking_connection_survives_permanently_stuck_resync_pending() and
* test_checking_connection_hello_and_needs_rejoin_precede_resync_gate_in_
* source() further below for the regression coverage of that distinction.
*
* What THIS file proves is the two things that are genuinely
* controller_connector.c's own: (1) checking_connection() actually
* consults the gate before attempting the rejoin -- the same way it
* already consults needs_rejoin -- and (2) myreq_handler's dispatch switch
* actually routes OPCode_jobStateQuery to the resync module at all, the
* exact gap task 4.2 left unclosed (confirmed by inspection: no case
* existed for this opcode before this task). xcecd_resync.c's own
* resync_pending set/clear lifecycle and multi-page query reassembly are
* covered end-to-end, against the real production code and real wire
* encode/decode, by test_xcecd_resync.c -- not duplicated here.
* ════════════════════════════════════════════════════════════════════════ */
static void
test_checking_connection_defers_rejoin_action_when_resync_pending(void)
{
controller_connector_ctx_t fake_cc = { 0 };
fake_cc.running = true;
cc_ctx = &fake_cc;
xcecli_ctx = NULL;
needs_rejoin = true; /* would normally trigger a rejoin attempt */
g_rejoin_call_count = 0;
rejoin_call = fake_rejoin_always_succeeds;
g_test_resync_pending = true;
checking_connection();
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_rejoin_call_count,
"checking_connection() must not attempt a rejoin while a job-state "
"resync report is still being sent (Req 3.8) -- the worker's "
"resumption of normal post-join operation must not begin or "
"interleave with it");
TEST_ASSERT_TRUE_MESSAGE(
needs_rejoin,
"a deferred tick must leave needs_rejoin untouched, not clear it "
"as if a rejoin had been attempted");
printf(
"test_checking_connection_defers_rejoin_action_when_resync_pending: "
"PASS\n");
}
static void test_checking_connection_resumes_once_resync_no_longer_pending(void)
{
controller_connector_ctx_t fake_cc = { 0 };
fake_cc.running = true;
cc_ctx = &fake_cc;
xcecli_ctx = NULL;
needs_rejoin = true;
g_rejoin_call_count = 0;
rejoin_call = fake_rejoin_always_succeeds;
g_test_resync_pending = false; /* report already finished sending */
checking_connection();
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, g_rejoin_call_count,
"once the gate clears, checking_connection() must resume its "
"normal periodic activity on its very next tick");
TEST_ASSERT_FALSE(needs_rejoin);
printf("test_checking_connection_resumes_once_resync_no_longer_pending: "
"PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Round-2 remediation regression guard (Requirements 3.8, 6.1): FINDING 1
* of the round-1 review -- gating checking_connection()'s ENTRY on
* resync_pending permanently disabled the worker's own connection-recovery
* mechanism whenever a multi-page job-state query's final page could never
* arrive (e.g. the worker's own connection to the controller dropped after
* a non-final page). resync_pending must now guard ONLY the
* needs_rejoin-triggered rejoin action -- xcecli_hello() (the worker's
* liveness probe / reconnect-detection trigger) and the needs_rejoin check
* itself must keep running every single tick regardless, extending
* test_checking_connection_never_permanently_disables_worker's exact
* pattern (many ticks, must never get permanently stuck) into this new
* adversarial condition: resync_pending genuinely stuck true (simulating a
* stalled multi-page query whose final page never arrives) for the whole
* loop, never cleared by anything inside checking_connection() itself.
* ════════════════════════════════════════════════════════════════════════ */
static void
test_checking_connection_survives_permanently_stuck_resync_pending(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, same fast/deterministic setup as
* test_checking_connection_never_permanently_disables_worker. */
xcecli_ctx = NULL;
needs_rejoin = true; /* a rejoin is genuinely pending throughout */
g_rejoin_call_count = 0;
rejoin_call = fake_rejoin_always_succeeds;
/* Simulates a multi-page job-state query that stalled after a
* non-final page (the worker's own connection dropped mid-query, so no
* further page -- and therefore no call into
* xcecd_resync_handle_query_page() that could ever clear the gate --
* will ever arrive). Left true for the entire loop below: nothing in
* this test ever clears it, matching the round-1 defect's worst case
* exactly. */
g_test_resync_pending = true;
struct timespec t0, t1;
clock_gettime(CLOCK_MONOTONIC, &t0);
/* Far more iterations than the old MAX_RETRY_COUNT (5) -- see
* test_checking_connection_never_permanently_disables_worker. Every
* single tick must still leave the worker able to detect/recover its
* own connection (cc_ctx->running never forced false), and the pending
* rejoin must never be silently dropped (needs_rejoin stays set, exactly
* because it is legitimately still deferred by the still-pending
* resync, not because the worker gave up). */
for (int i = 0; i < 25; i++) {
checking_connection();
TEST_ASSERT_TRUE_MESSAGE(
cc_ctx->running,
"a permanently-stuck resync_pending must never permanently "
"disable the worker -- checking_connection()'s liveness probe "
"and needs_rejoin check must keep running every tick "
"regardless (Requirement 6.1)");
TEST_ASSERT_TRUE_MESSAGE(
needs_rejoin,
"the rejoin action stays correctly deferred tick after tick "
"while resync_pending is genuinely stuck true -- it must "
"never be silently cleared without actually being attempted");
}
clock_gettime(CLOCK_MONOTONIC, &t1);
long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000L
+ (t1.tv_nsec - t0.tv_nsec) / 1000000L;
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_rejoin_call_count,
"the rejoin action itself must stay deferred for as long as "
"resync_pending is genuinely (if pathologically) stuck true -- "
"narrowing the gate must not mean it stops applying");
/* 25 ticks completing well under a second proves this loop never fell
* into any blocking wait -- same bound as the sibling invariant test. */
TEST_ASSERT_LESS_THAN_INT(1000, elapsed_ms);
printf(
"test_checking_connection_survives_permanently_stuck_resync_"
"pending: PASS (elapsed_ms=%ld)\n",
elapsed_ms);
}
/* A dummy backing buffer large enough for the real api_ctx_t
* (client_internal.h) that xcecli_get_my_endpoint()/xcecli_get_myinfo()
* (real functions, src/lib/client.c) cast xcecli_ctx to internally --
* this file must not include client_internal.h itself (its api_ctx_t
* conflicts with controller_connector.h's own unused same-named type, see
* this file's top-of-file comment), but neither accessor dereferences
* anything through the returned pointer in THIS translation unit, since
* xcecd_resync_handle_query_page() is a stub here that never dereferences
* its ep/from parameters either -- only their addresses need to be
* computable, which pure pointer arithmetic over a large enough zeroed
* buffer safely provides regardless of the real struct's exact layout. */
static uint8_t g_dummy_api_ctx[8192];
/* Req 8.1/8.3: myreq_handler's OPCode_jobStateQuery case must route an
* incoming query page to xcecd_resync_handle_query_page(). */
static void test_myreq_handler_routes_jobstatequery_to_resync_module(void)
{
controller_connector_ctx_t fake_cc = { 0 };
void* worker_list_head = NULL;
fake_cc.running = true;
fake_cc.worker_list = &worker_list_head;
cc_ctx = &fake_cc;
memset(g_dummy_api_ctx, 0, sizeof(g_dummy_api_ctx));
xcecli_ctx = (xcecli_ctx_t)g_dummy_api_ctx;
uint64_t jobids[2] = { 42, 43 };
xce_job_state_query_t page = { 0 };
page.n_total = 2;
page.pos = 0;
page.count = 2;
page.jobids = jobids;
xce_request_t request = { 0 };
request.opcode = OPCode_jobStateQuery;
request.body_which = Request_body_jobstatequerymsg;
request.body.jobstatequerymsg = &page;
int rc = myreq_handler(&request, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, g_resync_handle_query_page_call_count,
"OPCode_jobStateQuery must reach xcecd_resync_handle_query_page() "
"exactly once per incoming page -- this is the real dispatch "
"wiring task 4.2 left out (no case existed for this opcode before "
"this task)");
TEST_ASSERT_EQUAL_PTR_MESSAGE(
&page, g_resync_handle_query_page_last_page,
"the decoded query page must be passed through unchanged");
xcecli_ctx = NULL;
printf("test_myreq_handler_routes_jobstatequery_to_resync_module: PASS\n");
}
/* Without an active controller connection (xcecli_ctx == NULL), the
* dispatch case must fail fast rather than call into the resync module
* with a NULL endpoint/peerinfo. */
static void
test_myreq_handler_jobstatequery_without_connection_is_rejected(void)
{
controller_connector_ctx_t fake_cc = { 0 };
void* worker_list_head = NULL;
fake_cc.running = true;
fake_cc.worker_list = &worker_list_head;
cc_ctx = &fake_cc;
xcecli_ctx = NULL;
uint64_t jobids[1] = { 1 };
xce_job_state_query_t page = { 0 };
page.n_total = 1;
page.pos = 0;
page.count = 1;
page.jobids = jobids;
xce_request_t request = { 0 };
request.opcode = OPCode_jobStateQuery;
request.body_which = Request_body_jobstatequerymsg;
request.body.jobstatequerymsg = &page;
int rc = myreq_handler(&request, NULL);
TEST_ASSERT_TRUE(rc < 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_resync_handle_query_page_call_count,
"without an active controller connection, the resync module must "
"never be called");
printf("test_myreq_handler_jobstatequery_without_connection_is_rejected: "
"PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 2.3 — structural regression guard: xcecli_set_reconnect_handler()
* must be called AFTER xcecli_start() succeeds, never before.
@ -637,6 +981,69 @@ test_reconnect_handler_registered_after_start_not_before_in_source(void)
"source: PASS\n");
}
/* Structural companion to
* test_checking_connection_survives_permanently_stuck_resync_pending()
* above: proves WHY checking_connection() can never again become
* permanently inert the way round 1's entry-guard placement did. Same
* source-text-scan genre as the xcecli_start()-before-
* xcecli_set_reconnect_handler() guard just above, because the behavioral
* test cannot, by itself, distinguish "the gate is checked first and
* returns early" from "the gate is checked last and defers only the
* rejoin action" when needs_rejoin/resync_pending start out already true
* in both cases -- both produce the same rejoin_call_count/needs_rejoin
* observations. The distinction is textual: xcecli_hello() and the
* needs_rejoin check must appear in checking_connection()'s body strictly
* BEFORE any xcecd_resync_is_pending() call, never gated behind it. */
static void
test_checking_connection_hello_and_needs_rejoin_precede_resync_gate_in_source(
void)
{
char* src = read_whole_file(XCE_CONTROLLER_CONNECTOR_C_PATH);
TEST_ASSERT_NOT_NULL_MESSAGE(
src, "could not read controller_connector.c source for scan");
const char* fn_start = strstr(src, "int checking_connection()");
TEST_ASSERT_NOT_NULL_MESSAGE(fn_start,
"checking_connection() definition not found");
/* Search only from the function definition onward, so an unrelated,
* earlier mention of any of these tokens elsewhere in the file (e.g. in
* a doc comment) cannot produce a false pass. */
const char* hello_call = strstr(fn_start, "xcecli_hello(xcecli_ctx");
const char* needs_rejoin_check = strstr(fn_start, "if (needs_rejoin)");
const char* pending_check
= strstr(fn_start, "xcecd_resync_is_pending()");
TEST_ASSERT_NOT_NULL_MESSAGE(hello_call,
"xcecli_hello(xcecli_ctx, ...) call site "
"not found in checking_connection()");
TEST_ASSERT_NOT_NULL_MESSAGE(
needs_rejoin_check,
"if (needs_rejoin) check not found in checking_connection()");
TEST_ASSERT_NOT_NULL_MESSAGE(
pending_check,
"xcecd_resync_is_pending() call not found in checking_connection()");
TEST_ASSERT_TRUE_MESSAGE(
hello_call < needs_rejoin_check,
"xcecli_hello() must run unconditionally, before the needs_rejoin "
"check -- never gated behind resync_pending (Requirement 6.1's "
"self-healing liveness probe)");
TEST_ASSERT_TRUE_MESSAGE(
needs_rejoin_check < pending_check,
"the resync_pending gate must be checked INSIDE the needs_rejoin "
"branch (guarding only the rejoin action itself), not before it -- "
"an entry-guard placement (round 1's confirmed defect) would put "
"xcecd_resync_is_pending() before both xcecli_hello() and the "
"needs_rejoin check, permanently disabling the worker's own "
"connection-recovery mechanism whenever resync_pending gets stuck");
FREEUP(src);
printf(
"test_checking_connection_hello_and_needs_rejoin_precede_resync_"
"gate_in_source: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 2.3 — Req 1.6/5.2: all-candidates-unreachable failure identifies
* every attempted candidate, not a generic error.
@ -828,6 +1235,7 @@ int main(void)
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_also_resets_resync_gate);
RUN_TEST(test_rejoin_callback_registered_via_public_api);
RUN_TEST(test_checking_connection_never_permanently_disables_worker);
RUN_TEST(test_checking_connection_skips_rejoin_when_not_pending);
@ -837,8 +1245,17 @@ int main(void)
test_checking_connection_rejoin_success_with_nonzero_index_clears_flag);
RUN_TEST(
test_checking_connection_rejoin_retried_on_next_probe_after_failure);
RUN_TEST(
test_checking_connection_defers_rejoin_action_when_resync_pending);
RUN_TEST(test_checking_connection_resumes_once_resync_no_longer_pending);
RUN_TEST(
test_checking_connection_survives_permanently_stuck_resync_pending);
RUN_TEST(test_myreq_handler_routes_jobstatequery_to_resync_module);
RUN_TEST(test_myreq_handler_jobstatequery_without_connection_is_rejected);
RUN_TEST(
test_reconnect_handler_registered_after_start_not_before_in_source);
RUN_TEST(
test_checking_connection_hello_and_needs_rejoin_precede_resync_gate_in_source);
RUN_TEST(test_format_candidate_list_names_every_candidate_in_order);
RUN_TEST(test_xcecd_join_cluster_all_unreachable_reports_every_candidate);
RUN_TEST(

366
src/worker/testsuite/test_xcecd_resync.c
View file

@ -159,9 +159,28 @@ static void assert_queue_empty(endpoint_t* ep)
TEST_ASSERT_NULL(endpoint_deq(ep));
}
void setUp(void) { xcecd_finished_job_retention_reset(); }
/* Resets task 4.3's own file-scope statics (resync_pending, the query
* reassembly accumulator) between tests, same rationale as
* xcecd_finished_job_retention_reset() below -- this file white-box
* #includes xcecd_resync.c directly, so every test in this binary shares
* the same static state unless each test starts from a known-clean one. */
static void reset_resync_gate_state(void)
{
reset_query_accum();
resync_pending = false;
}
void tearDown(void) { xcecd_finished_job_retention_reset(); }
void setUp(void)
{
xcecd_finished_job_retention_reset();
reset_resync_gate_state();
}
void tearDown(void)
{
xcecd_finished_job_retention_reset();
reset_resync_gate_state();
}
/* ════════════════════════════════════════════════════════════════════════
* xcecd_resync_classify_job: per-ID classification (Requirements 3.1, 3.3)
@ -648,6 +667,339 @@ static void test_retention_snapshot_null_out_params_rejected(void)
-EINVAL, xcecd_finished_job_retention_snapshot(1000, &snap, NULL));
}
/* ════════════════════════════════════════════════════════════════════════
* xcecd_resync_handle_query_page / xcecd_resync_is_pending: the real
* OPCode_jobStateQuery receive-side entry point and its Requirement 3.8
* resync_pending gate (task 4.3).
* ════════════════════════════════════════════════════════════════════════ */
static xce_job_state_query_t make_query_page(int n_total, int pos, int count,
uint64_t* jobids)
{
xce_job_state_query_t page;
memset(&page, 0, sizeof(page));
page.n_total = n_total;
page.pos = pos;
page.count = count;
page.jobids = jobids;
return page;
}
/* A single-page query is fully answered by one call: the gate is false
* before and after, the call reports "fully answered" (rc == 1), and a
* real OPCode_jobStateReport -- decoded off the wire, not a stand-in --
* reaches the controller (Requirements 3.8, 8.2, 8.3). */
static void test_handle_query_page_single_page_answers_and_clears_gate(void)
{
xcecd_worker_t* list = NULL;
link_worker(&list, make_worker(1, 7001, WORKER_RUNNING));
uint64_t jobids[1] = { 7001 };
xce_job_state_query_t page = make_query_page(1, 0, 1, jobids);
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
int rc = xcecd_resync_handle_query_page(list, ep, &from, &page, 8000, 0);
TEST_ASSERT_EQUAL_INT(1, rc);
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode);
TEST_ASSERT_EQUAL_INT(1, decoded.body.jobstatereportmsg->n_total);
TEST_ASSERT_EQUAL_UINT64(7001,
decoded.body.jobstatereportmsg->entries[0]->jobid);
xce_free_request(&decoded);
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
free_worker_list(list);
}
/* A query that itself arrives across multiple pages (Requirement 8.1/8.3):
* the gate must be observably set and the report must NOT have been sent
* yet after the first (non-final) page, and only after the last page
* arrives does the gate clear and the full, correctly-reassembled report
* reach the wire -- covering every queried jobid, none dropped
* (Requirement 3.8's "never before the last page" completion criterion). */
static void test_handle_query_page_multipage_query_holds_gate_until_last(void)
{
xcecd_worker_t* list = NULL;
int i;
uint64_t jobids[4] = { 8001, 8002, 8003, 8004 };
for (i = 0; i < 4; i++) {
link_worker(&list, make_worker(300 + i, jobids[i], WORKER_RUNNING));
}
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
/* Page 1 of 2: jobids[0..2). */
xce_job_state_query_t page1 = make_query_page(4, 0, 2, &jobids[0]);
int rc1 = xcecd_resync_handle_query_page(list, ep, &from, &page1, 9000, 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc1,
"a non-final query page must not answer yet");
TEST_ASSERT_TRUE_MESSAGE(
xcecd_resync_is_pending(),
"the gate must stay set while more query pages are still expected");
assert_queue_empty(ep); /* nothing sent -- report not built yet */
/* Page 2 of 2 (final): jobids[2..4). */
xce_job_state_query_t page2 = make_query_page(4, 2, 2, &jobids[2]);
int rc2 = xcecd_resync_handle_query_page(list, ep, &from, &page2, 9000, 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc2,
"the final query page must trigger the "
"full answer");
TEST_ASSERT_FALSE_MESSAGE(
xcecd_resync_is_pending(),
"the gate must clear only after the full report has been sent");
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode);
TEST_ASSERT_EQUAL_INT(4, decoded.body.jobstatereportmsg->n_total);
TEST_ASSERT_EQUAL_INT(4, decoded.body.jobstatereportmsg->count);
/* Every queried jobid from the reassembled query must appear exactly
* once -- proving the two incoming query pages were correctly
* reassembled into one flat list before being answered. */
for (i = 0; i < 4; i++) {
TEST_ASSERT_NOT_NULL(
find_entry(decoded.body.jobstatereportmsg->entries, 4, jobids[i]));
}
xce_free_request(&decoded);
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
free_worker_list(list);
}
/* An empty query (n_total == 0) is still a complete, single-call exchange
* -- one empty report page sent, gate cleared. */
static void test_handle_query_page_empty_query_answers_immediately(void)
{
xce_job_state_query_t page = make_query_page(0, 0, 0, NULL);
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
int rc = xcecd_resync_handle_query_page(NULL, ep, &from, &page, 1000, 0);
TEST_ASSERT_EQUAL_INT(1, rc);
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode);
TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatereportmsg->n_total);
xce_free_request(&decoded);
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
}
/* A malformed page (count exceeding its own declared n_total) is rejected
* without ever sending a report, and must not leave the gate stuck set. */
static void test_handle_query_page_malformed_page_clears_gate(void)
{
uint64_t jobids[1] = { 1 };
xce_job_state_query_t page = make_query_page(1, 0, 5, jobids);
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
int rc = xcecd_resync_handle_query_page(NULL, ep, &from, &page, 1000, 0);
TEST_ASSERT_TRUE(rc < 0);
TEST_ASSERT_FALSE_MESSAGE(
xcecd_resync_is_pending(),
"a malformed page must not leave the gate stuck set");
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
}
/* NULL page/ep/from are rejected outright (-EINVAL), no gate side effect. */
static void test_handle_query_page_null_args_rejected(void)
{
uint64_t jobids[1] = { 1 };
xce_job_state_query_t page = make_query_page(1, 0, 1, jobids);
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
TEST_ASSERT_EQUAL_INT(-EINVAL, xcecd_resync_handle_query_page(
NULL, NULL, &from, &page, 1000, 0));
TEST_ASSERT_EQUAL_INT(-EINVAL, xcecd_resync_handle_query_page(
NULL, ep, NULL, &page, 1000, 0));
TEST_ASSERT_EQUAL_INT(-EINVAL, xcecd_resync_handle_query_page(
NULL, ep, &from, NULL, 1000, 0));
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
endpoint_fini(ep);
free(ep);
}
/* A stale, incomplete accumulation left by an aborted prior query (e.g. the
* worker disconnected mid-query) must be discarded, not merged into, the
* next query's own pages -- a fresh pos == 0 page always starts clean. */
static void test_handle_query_page_stale_accumulation_discarded(void)
{
uint64_t abandoned[2] = { 111, 222 };
xce_job_state_query_t abandoned_page1
= make_query_page(4, 0, 2, abandoned); /* never completed */
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
int rc1 = xcecd_resync_handle_query_page(NULL, ep, &from, &abandoned_page1,
1000, 0);
TEST_ASSERT_EQUAL_INT(0, rc1);
TEST_ASSERT_TRUE(xcecd_resync_is_pending());
assert_queue_empty(ep);
/* A brand-new, single-page, single-job query arrives instead of the
* abandoned one's second page. */
uint64_t fresh_jobids[1] = { 999 };
xce_job_state_query_t fresh_page = make_query_page(1, 0, 1, fresh_jobids);
int rc2
= xcecd_resync_handle_query_page(NULL, ep, &from, &fresh_page, 1000, 0);
TEST_ASSERT_EQUAL_INT(1, rc2);
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(1, decoded.body.jobstatereportmsg->n_total);
TEST_ASSERT_EQUAL_UINT64(999,
decoded.body.jobstatereportmsg->entries[0]->jobid);
xce_free_request(&decoded);
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
}
/* ════════════════════════════════════════════════════════════════════════
* xcecd_resync_reset: round-2 remediation for the confirmed round-1 defect
* (Requirements 3.8, 6.1) -- a stalled multi-page query (connection dropped
* after a non-final page, no further page ever arriving) must not leave
* resync_pending stuck true forever. controller_connector.c's
* xcecd_on_reconnect() calls this the moment the worker detects it has
* reconnected; here we exercise the function directly, the same white-box
* boundary this file already uses for everything else in xcecd_resync.c.
* ════════════════════════════════════════════════════════════════════════ */
/* The core regression scenario: a query stalls after its first (non-final)
* page -- the gate is left set and an accumulator is left in progress,
* exactly like a worker-side disconnect mid-query. No further page ever
* arrives (simulating the dead connection). xcecd_resync_reset() must clear
* the gate and discard the stale accumulation immediately, without needing
* another call to xcecd_resync_handle_query_page(). */
static void test_reset_clears_gate_stuck_by_a_stalled_multipage_query(void)
{
uint64_t stalled[4] = { 501, 502, 503, 504 };
xce_job_state_query_t page1 = make_query_page(4, 0, 2, &stalled[0]);
char nm[] = "worker1";
char ky[] = "key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = new_test_endpoint();
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
/* Page 1 of 2 arrives; page 2 never will (connection dropped). */
int rc1 = xcecd_resync_handle_query_page(NULL, ep, &from, &page1, 9500, 0);
TEST_ASSERT_EQUAL_INT(0, rc1);
TEST_ASSERT_TRUE_MESSAGE(
xcecd_resync_is_pending(),
"gate must be set after a non-final page, same as any other "
"in-progress multi-page query");
assert_queue_empty(ep);
/* The worker itself detects it has reconnected (the real trigger is
* controller_connector.c's xcecd_on_reconnect(); this test calls the
* resync module's own half of that directly, matching this file's
* white-box boundary). */
xcecd_resync_reset();
TEST_ASSERT_FALSE_MESSAGE(
xcecd_resync_is_pending(),
"xcecd_resync_reset() must clear a gate left stuck by a query "
"whose final page will never arrive -- round 1's confirmed defect "
"left this stuck true forever, permanently disabling "
"checking_connection()'s rejoin action");
/* No report must have been sent for the abandoned, half-received
* query -- discarding it is silent, not a partial/garbage answer. */
assert_queue_empty(ep);
/* A brand-new query arriving afterward must be answered cleanly from
* scratch, proving the stale accumulator was actually discarded, not
* just the gate flipped while corrupt state lingered underneath. */
uint64_t fresh_jobids[1] = { 999 };
xce_job_state_query_t fresh_page = make_query_page(1, 0, 1, fresh_jobids);
int rc2
= xcecd_resync_handle_query_page(NULL, ep, &from, &fresh_page, 9500, 0);
TEST_ASSERT_EQUAL_INT(1, rc2);
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(1, decoded.body.jobstatereportmsg->n_total);
TEST_ASSERT_EQUAL_UINT64(999,
decoded.body.jobstatereportmsg->entries[0]->jobid);
xce_free_request(&decoded);
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
}
/* xcecd_resync_reset() must also be a safe no-op when nothing was pending
* (e.g. a transparent reconnect that happens to land between queries). */
static void test_reset_is_a_safe_noop_when_nothing_pending(void)
{
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
xcecd_resync_reset();
TEST_ASSERT_FALSE(xcecd_resync_is_pending());
}
/* ════════════════════════════════════════════════════════════════════════
* main
* ════════════════════════════════════════════════════════════════════════ */
@ -682,5 +1034,15 @@ int main(void)
RUN_TEST(test_retention_snapshot_empty_store);
RUN_TEST(test_retention_snapshot_null_out_params_rejected);
RUN_TEST(test_handle_query_page_single_page_answers_and_clears_gate);
RUN_TEST(test_handle_query_page_multipage_query_holds_gate_until_last);
RUN_TEST(test_handle_query_page_empty_query_answers_immediately);
RUN_TEST(test_handle_query_page_malformed_page_clears_gate);
RUN_TEST(test_handle_query_page_null_args_rejected);
RUN_TEST(test_handle_query_page_stale_accumulation_discarded);
RUN_TEST(test_reset_clears_gate_stuck_by_a_stalled_multipage_query);
RUN_TEST(test_reset_is_a_safe_noop_when_nothing_pending);
return UNITY_END();
}