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

feat(worker-rejoin-resync): task 6.1 - trigger reconciliation on rejoin

Browse source 
Arm a reconciliation round in apply_join_state() the moment a worker
host becomes connected (first join, reconnect, or failover-reconnect -
all routed through the same shared call site, Req 1.1-1.3), then send
the query from on_worker_join() strictly after the join reply and
existing config push (Req 2.1-2.2), verified via a structural test
that reads real wire order off the endpoint queue.

Also closes Req 3.7's config-delivery seam end-to-end: worker_resync_
window_ms now rides the existing Conf4Worker push (clustername
precedent, no new opcode) and the worker applies it to its retention
window on receipt.

Fixes bug-1752: myreq_handler had no OPCode_push dispatch case at all,
so every config push - including the pre-existing clustername field -
was silently discarded. Confirmed harmless until now (clustername was
never read anywhere in src/worker/); closed narrowly for
worker_resync_window_ms only.

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 19:43:18 +00:00
parent 48c6cfd58a
commit a3d12d37d9
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
8 changed files with 1045 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

645
src/controller/testsuite/test_worker_join.c
View file

@ -22,12 +22,26 @@
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h" /* worker-rejoin-resync task 6.1: OPCode_push/
* OPCode_jobStateQuery, ConfigData_data_*,
* Request_body_* constants */
#include "jobj.h" /* worker-rejoin-resync task 6.1: makeJob, for a real
* non-terminal job the arm-round wiring tests need */
#include "kvmgr.h"
#include "netio.h" /* worker-rejoin-resync task 6.1: endpoint_get/
* endpoint_deq/ITPTR_TO_PACKET, for real wire capture */
#include "request.h" /* worker-rejoin-resync task 6.1: xce_decode,
* xce_free_request */
#include "resobj.h"
#include "resync.h" /* worker-rejoin-resync task 6.1: worker_t.resync,
* XCE_WORKER_RECONCILING */
#include "wobj.h"
#include "worker.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xcealloc.h" /* worker-rejoin-resync task 6.1: make_job_alloc and
* friends, for a job assigned to a test worker's
* host */
#include "xstring.h"
#include "zarray.h"
@ -55,6 +69,19 @@ void collect_confvars(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
int apply_join_state(cluster_t* c, worker_t* w, endpoint_t* ep,
xce_hostconfig_t* hconfig);
/*
* worker-rejoin-resync task 6.1: worker_push_config and worker_send_resync_
* query are non-static in worker.c when XCE_TESTING is defined, mirroring
* apply_join_state's own convention above -- lets these tests drive the
* exact config-push-then-query wire sequence on_worker_join produces,
* without needing a live join flow or get_ep_worker()'s endpoint->worker
* resolution (a separate, pre-existing concern -- see
* test_on_worker_join_sends_config_push_even_when_worker_unresolvable
* below for why).
*/
int worker_push_config(cluster_t* c, endpoint_t* ep);
int worker_send_resync_query(cluster_t* c, worker_t* w, endpoint_t* ep);
/*
* init_worker and rejoin_worker are exported from worker.c.
*
@ -84,10 +111,26 @@ typedef struct {
cluster_t c;
var workermgr;
var resmgr;
var jobmgr; /* worker-rejoin-resync task 6.1: real jobmgr so the
* apply_join_state arm-round wiring tests below can
* seed a genuine non-terminal job assigned to a test
* worker's host (mirrors test_worker_resync.c's own
* g_twr.jobmgr fixture) */
} test_cluster_t;
static test_cluster_t g_tc;
/* worker-rejoin-resync task 6.1: jobj.c's job type table is a shared,
* process-wide resource across every test file in the controller_unit_test
* binary -- job_obj_init() is explicitly documented (jobj.c) as safe to
* call more than once per process without an intervening job_obj_fini()
* (Req 10.2, the exact "every controller unit-test file that constructs
* jobs calls job_obj_init() from its own setup" scenario this file is now
* one more instance of), so no cross-file coordination is needed -- guard
* only avoids redundant re-init within THIS file's own repeated setup/
* teardown cycles, mirroring test_worker_resync.c's g_twr_jobobj_ready. */
static int g_tc_jobobj_ready;
static void setup_test_cluster(void)
{
kvmgr_params_t str_params = {
@ -96,15 +139,28 @@ static void setup_test_cluster(void)
.delta = 16,
.datakiller = NULL,
};
kvmgr_params_t int_params = {
.key_type = KVMGR_KEY_INT,
.nslots = 32,
.delta = 16,
};
memset(&g_tc, 0, sizeof(g_tc));
if (!g_tc_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_tc_jobobj_ready = 1;
}
g_tc.workermgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_tc.workermgr);
g_tc.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_tc.resmgr);
g_tc.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(g_tc.jobmgr);
for (int i = 0; i < N_RES; i++) {
xce_resource_t rdef;
memset(&rdef, 0, sizeof(rdef));
@ -119,6 +175,18 @@ static void setup_test_cluster(void)
g_tc.c.workermgr = g_tc.workermgr;
g_tc.c.resmgr = g_tc.resmgr;
g_tc.c.jobmgr = g_tc.jobmgr;
/* worker-rejoin-resync task 6.1: c->myhost is the `from` peer identity
* worker_push_config()/worker_send_resync_query() pass to xce_
* pushconfig_req()/xce_job_state_query_req() -- both reject a NULL
* `from` outright, so any test exercising the real wire send needs
* this populated (mirrors production cluster.c's own myhost_info
* storage-then-point-at-it pattern, xce/cluster_def.h:101-102). */
g_tc.c.myhost_info.name = (char*)"test-controller";
g_tc.c.myhost_info.key = (char*)"test-key";
g_tc.c.myhost = &g_tc.c.myhost_info;
zarray_init(&g_tc.c.sharedinst, 64, NULL);
}
@ -127,9 +195,42 @@ static void teardown_test_cluster(void)
zarray_fini(&g_tc.c.sharedinst);
if (g_tc.resmgr) dereference(g_tc.resmgr);
if (g_tc.workermgr) dereference(g_tc.workermgr);
if (g_tc.jobmgr) dereference(g_tc.jobmgr);
memset(&g_tc, 0, sizeof(g_tc));
}
/* Seed a job directly in the fixture's jobmgr, bypassing dispatch/replay --
* mirrors test_worker_resync.c's twr_seed_job. */
static job_submit_data_t g_tc_seed_submit;
static job_t* seed_test_job(uint64_t jobid, uint32_t state)
{
xce_job_t jdef;
memset(&g_tc_seed_submit, 0, sizeof(g_tc_seed_submit));
g_tc_seed_submit.queue = (char*)"q";
g_tc_seed_submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = state;
jdef.submit = &g_tc_seed_submit;
return makeJob(g_tc.jobmgr, &jdef);
}
/* Build a single-host job allocation assigning `jobid` to `host` -- mirrors
* test_worker_resync.c's twr_make_test_alloc. Caller hands ownership to
* job->job.allocs. */
static xce_job_alloc_t* make_test_job_alloc(uint64_t jobid, const char* host)
{
return make_job_alloc(jobid, 1,
make_host_allocs(1,
make_host_alloc((char*)host, 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
}
/* Create a worker in the test workermgr; attrs strings must be makeString'd. */
static worker_t* make_test_worker(const char* name, hostattrs_t* attrs)
{
@ -691,6 +792,537 @@ static void test_rejoin_worker_preserves_admin_bits(void)
printf("test_rejoin_worker_preserves_admin_bits: PASS\n");
}
/* ── Task 6.1: apply_join_state arms a reconciliation round ──────────── */
/*
* Real (heap) endpoint_t via endpoint_get() -- NOT a hand-rolled/zeroed
* stack stand-in (bug-378: an under-sized/wrong-typed stand-in for
* endpoint_t caused a global-buffer-overflow when passed where a real
* endpoint_t* was expected). Needed only by the tests below that actually
* send something over the wire (worker_push_config/worker_send_resync_
* query); the pre-existing apply_join_state/init_worker/rejoin_worker
* tests above never send anything and so are unaffected, still using a
* plain zeroed stack endpoint_t. Mirrors src/lib/testsuite/
* test_job_state_wire.c's own established pattern for this exact
* situation.
*/
static endpoint_t* new_test_endpoint(void)
{
endpoint_t* ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
return ep;
}
/* Pop one queued page off ep->outq and decode it. Fails the test if the
* queue is empty. Caller must xce_free_request() the result. */
static void pop_and_decode(endpoint_t* ep, xce_request_t* out)
{
data_item_t* it = endpoint_deq(ep);
TEST_ASSERT_NOT_NULL(it);
io_packet_t* pkt = ITPTR_TO_PACKET(it);
memset(out, 0, sizeof(*out));
TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, out)); /* consumes/frees pkt */
}
static void assert_queue_empty(endpoint_t* ep)
{
TEST_ASSERT_NULL(endpoint_deq(ep));
}
/*
* rejoin_worker (via apply_join_state) must arm a reconciliation round --
* w->resync populated, XCE_WORKER_RECONCILING set, the query set matching
* exactly the host's non-terminal jobs -- when the controller currently
* holds one or more non-terminal jobs for that host. Req 1.1, 2.1, 2.2.
*/
static void test_rejoin_worker_arms_reconciliation_round_when_jobs_exist(void)
{
setup_test_cluster();
job_t* j = seed_test_job(96001, JOB_STAT_RUN);
TEST_ASSERT_NOT_NULL(j);
j->job.allocs = make_test_job_alloc(96001, "arm-host-1");
TEST_ASSERT_NOT_NULL(j->job.allocs);
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = makeString("LINUX");
attrs.hostmodel = makeString("x86");
attrs.cpufactor = 1.0f;
attrs.maxcpus = 4;
attrs.maxmem = 16;
attrs.maxswap = 32;
attrs.maxtmp = 64;
attrs.ndisks = 1;
worker_t* w = make_test_worker("arm-host-1", &attrs);
TEST_ASSERT_NOT_NULL(w);
hostattrs_t new_attrs;
memset(&new_attrs, 0, sizeof(new_attrs));
new_attrs.hosttype = makeString("LINUX");
new_attrs.hostmodel = makeString("x86");
new_attrs.cpufactor = 1.0f;
new_attrs.maxcpus = 4;
new_attrs.maxmem = 16;
new_attrs.maxswap = 32;
new_attrs.maxtmp = 64;
new_attrs.ndisks = 1;
xce_hostconfig_t hconfig;
memset(&hconfig, 0, sizeof(hconfig));
hconfig.hosttype = (char*)"!";
hconfig.hostmodel = (char*)"!";
hconfig.mxj = 4;
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
int idx = rejoin_worker(&g_tc.c, w, &ep, &new_attrs, &hconfig);
TEST_ASSERT_GREATER_OR_EQUAL(0, idx);
TEST_ASSERT_NOT_NULL_MESSAGE(w->resync,
"a worker rejoining with a known non-terminal job must have a "
"reconciliation round armed (Req 1.1)");
TEST_ASSERT_TRUE_MESSAGE(w->host.status & XCE_WORKER_RECONCILING,
"XCE_WORKER_RECONCILING must be set once a round is armed");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, w->resync->n_queried_jobids,
"the round's query set must match exactly the host's non-terminal "
"jobs (Req 2.1)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(96001, w->resync->queried_jobids[0],
"the queried job must be identified by its unique jobid (Req 2.2)");
dereference(new_attrs.hosttype);
dereference(new_attrs.hostmodel);
teardown_test_cluster();
printf("test_rejoin_worker_arms_reconciliation_round_when_jobs_exist: "
"PASS\n");
}
/*
* A worker rejoining with NO non-terminal jobs currently assigned to it
* must NOT have a round armed. Req 1.2.
*/
static void test_rejoin_worker_no_round_when_no_jobs(void)
{
setup_test_cluster();
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = makeString("LINUX");
attrs.hostmodel = makeString("x86");
attrs.cpufactor = 1.0f;
attrs.maxcpus = 4;
attrs.maxmem = 16;
attrs.maxswap = 32;
attrs.maxtmp = 64;
attrs.ndisks = 1;
worker_t* w = make_test_worker("arm-host-2", &attrs);
TEST_ASSERT_NOT_NULL(w);
hostattrs_t new_attrs = attrs;
new_attrs.hosttype = makeString("LINUX");
new_attrs.hostmodel = makeString("x86");
xce_hostconfig_t hconfig;
memset(&hconfig, 0, sizeof(hconfig));
hconfig.hosttype = (char*)"!";
hconfig.hostmodel = (char*)"!";
hconfig.mxj = 4;
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
rejoin_worker(&g_tc.c, w, &ep, &new_attrs, &hconfig);
TEST_ASSERT_NULL_MESSAGE(w->resync,
"a worker rejoining with no non-terminal jobs must not have a "
"reconciliation round armed (Req 1.2)");
TEST_ASSERT_FALSE_MESSAGE(w->host.status & XCE_WORKER_RECONCILING,
"XCE_WORKER_RECONCILING must not be set");
dereference(new_attrs.hosttype);
dereference(new_attrs.hostmodel);
teardown_test_cluster();
printf("test_rejoin_worker_no_round_when_no_jobs: PASS\n");
}
/*
* A terminal job (DONE) assigned to the host must not be queried -- only
* non-terminal jobs count toward the trigger (Req 1.1's own "non-terminal"
* qualifier).
*/
static void test_rejoin_worker_no_round_when_only_terminal_jobs(void)
{
setup_test_cluster();
job_t* j = seed_test_job(96002, JOB_STAT_DONE);
TEST_ASSERT_NOT_NULL(j);
j->job.allocs = make_test_job_alloc(96002, "arm-host-3");
TEST_ASSERT_NOT_NULL(j->job.allocs);
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = makeString("LINUX");
attrs.hostmodel = makeString("x86");
attrs.cpufactor = 1.0f;
attrs.maxcpus = 4;
attrs.maxmem = 16;
attrs.maxswap = 32;
attrs.maxtmp = 64;
attrs.ndisks = 1;
worker_t* w = make_test_worker("arm-host-3", &attrs);
TEST_ASSERT_NOT_NULL(w);
hostattrs_t new_attrs = attrs;
new_attrs.hosttype = makeString("LINUX");
new_attrs.hostmodel = makeString("x86");
xce_hostconfig_t hconfig;
memset(&hconfig, 0, sizeof(hconfig));
hconfig.hosttype = (char*)"!";
hconfig.hostmodel = (char*)"!";
hconfig.mxj = 4;
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
rejoin_worker(&g_tc.c, w, &ep, &new_attrs, &hconfig);
TEST_ASSERT_NULL_MESSAGE(w->resync,
"a terminal (DONE) job must not count toward the reconciliation "
"trigger");
dereference(new_attrs.hosttype);
dereference(new_attrs.hostmodel);
teardown_test_cluster();
printf("test_rejoin_worker_no_round_when_only_terminal_jobs: PASS\n");
}
/*
* Req 1.3: the SAME trigger condition -- no special-casing needed -- must
* apply to a host's first-ever join (init_worker), not just a rejoin.
*/
static void test_init_worker_arms_reconciliation_round_when_jobs_exist(void)
{
setup_test_cluster();
job_t* j = seed_test_job(96003, JOB_STAT_RUN);
TEST_ASSERT_NOT_NULL(j);
j->job.allocs = make_test_job_alloc(96003, "arm-host-4-first-join");
TEST_ASSERT_NOT_NULL(j->job.allocs);
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = makeString("LINUX");
attrs.hostmodel = makeString("x86");
attrs.cpufactor = 1.0f;
attrs.maxcpus = 4;
attrs.maxmem = 16;
attrs.maxswap = 32;
attrs.maxtmp = 64;
attrs.ndisks = 1;
xce_hostconfig_t hconfig;
memset(&hconfig, 0, sizeof(hconfig));
hconfig.hosttype = (char*)"!";
hconfig.hostmodel = (char*)"!";
hconfig.mxj = 4;
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
int idx = init_worker(&g_tc.c, &ep, "arm-host-4-first-join", &attrs,
&hconfig);
TEST_ASSERT_GREATER_OR_EQUAL(0, idx);
worker_t* w = NULL;
lock(g_tc.c.workermgr, RTTI_LOCK_READ);
key_find(g_tc.c.workermgr, (void*)"arm-host-4-first-join", (void**)&w);
unlock(g_tc.c.workermgr);
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_NOT_NULL_MESSAGE(w->resync,
"a host's first-ever join must arm a reconciliation round exactly "
"like a rejoin, with no special-casing needed (Req 1.3)");
TEST_ASSERT_EQUAL_INT(1, w->resync->n_queried_jobids);
TEST_ASSERT_EQUAL_UINT64(96003, w->resync->queried_jobids[0]);
teardown_test_cluster();
printf("test_init_worker_arms_reconciliation_round_when_jobs_exist: "
"PASS\n");
}
/* ── Task 6.1: worker_send_resync_query ───────────────────────────────── */
/*
* When a round is armed for `w`, worker_send_resync_query() must send
* exactly that round's job IDs as an OPCode_jobStateQuery. Req 2.1, 2.2.
*/
static void test_worker_send_resync_query_sends_armed_round(void)
{
setup_test_cluster();
job_t* j1 = seed_test_job(96010, JOB_STAT_RUN);
j1->job.allocs = make_test_job_alloc(96010, "query-host-1");
job_t* j2 = seed_test_job(96011, JOB_STAT_RUN);
j2->job.allocs = make_test_job_alloc(96011, "query-host-1");
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = makeString("LINUX");
attrs.hostmodel = makeString("x86");
attrs.cpufactor = 1.0f;
attrs.maxcpus = 4;
attrs.maxmem = 16;
attrs.maxswap = 32;
attrs.maxtmp = 64;
attrs.ndisks = 1;
worker_t* w = make_test_worker("query-host-1", &attrs);
TEST_ASSERT_NOT_NULL(w);
hostattrs_t new_attrs = attrs;
new_attrs.hosttype = makeString("LINUX");
new_attrs.hostmodel = makeString("x86");
xce_hostconfig_t hconfig;
memset(&hconfig, 0, sizeof(hconfig));
hconfig.hosttype = (char*)"!";
hconfig.hostmodel = (char*)"!";
hconfig.mxj = 4;
endpoint_t stack_ep;
memset(&stack_ep, 0, sizeof(stack_ep));
rejoin_worker(&g_tc.c, w, &stack_ep, &new_attrs, &hconfig);
TEST_ASSERT_NOT_NULL(w->resync);
TEST_ASSERT_EQUAL_INT(2, w->resync->n_queried_jobids);
endpoint_t* ep = new_test_endpoint();
TEST_ASSERT_EQUAL_INT(0, worker_send_resync_query(&g_tc.c, w, ep));
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateQuery, decoded.opcode);
TEST_ASSERT_EQUAL_INT(Request_body_jobstatequerymsg, decoded.body_which);
TEST_ASSERT_NOT_NULL(decoded.body.jobstatequerymsg);
TEST_ASSERT_EQUAL_INT(2, decoded.body.jobstatequerymsg->n_total);
int saw_96010 = 0, saw_96011 = 0;
for (int i = 0; i < decoded.body.jobstatequerymsg->count; i++) {
uint64_t id = decoded.body.jobstatequerymsg->jobids[i];
if (id == 96010) saw_96010 = 1;
if (id == 96011) saw_96011 = 1;
}
TEST_ASSERT_TRUE_MESSAGE(saw_96010, "queried set must include 96010");
TEST_ASSERT_TRUE_MESSAGE(saw_96011, "queried set must include 96011");
xce_free_request(&decoded);
assert_queue_empty(ep);
dereference(new_attrs.hosttype);
dereference(new_attrs.hostmodel);
endpoint_fini(ep);
free(ep);
teardown_test_cluster();
printf("test_worker_send_resync_query_sends_armed_round: PASS\n");
}
/*
* When no round is armed (w->resync == NULL), worker_send_resync_query()
* must send nothing at all -- the Req 1.2 no-round case's downstream
* completion criterion ("no round, no query").
*/
static void test_worker_send_resync_query_sends_nothing_when_no_round(void)
{
setup_test_cluster();
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = makeString("LINUX");
attrs.hostmodel = makeString("x86");
attrs.cpufactor = 1.0f;
attrs.maxcpus = 4;
attrs.maxmem = 16;
attrs.maxswap = 32;
attrs.maxtmp = 64;
attrs.ndisks = 1;
worker_t* w = make_test_worker("query-host-2", &attrs);
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_NULL(w->resync);
endpoint_t* ep = new_test_endpoint();
TEST_ASSERT_EQUAL_INT(0, worker_send_resync_query(&g_tc.c, w, ep));
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
teardown_test_cluster();
printf("test_worker_send_resync_query_sends_nothing_when_no_round: "
"PASS\n");
}
/* ── Task 6.1: config push, then query, in that fixed order ──────────── */
/*
* design.md's File Structure Plan for on_worker_join is explicit that this
* ordering is load-bearing ("config push, then query", "not as an
* extension of the join call itself"), and this spec's own Implementation
* Notes (task 4.3) warn that a placement-sensitive guarantee needs a
* STRUCTURAL assertion, not just an end-state one -- a behavioral test
* alone can pass even with the wrong order if it only checks that both
* packets eventually arrive. This test drives the exact two calls on_
* worker_join makes, in the exact order it makes them, and asserts the
* WIRE ORDER of the resulting packets directly.
*/
static void test_config_push_then_resync_query_wire_order(void)
{
setup_test_cluster();
cluster_config_t cc;
memset(&cc, 0, sizeof(cc));
cc.clustername = (char*)"test-cluster";
myconf_t myconf;
memset(&myconf, 0, sizeof(myconf));
myconf.cluster_config = &cc;
g_tc.c.myconf = &myconf;
job_t* j = seed_test_job(96020, JOB_STAT_RUN);
j->job.allocs = make_test_job_alloc(96020, "order-host-1");
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = makeString("LINUX");
attrs.hostmodel = makeString("x86");
attrs.cpufactor = 1.0f;
attrs.maxcpus = 4;
attrs.maxmem = 16;
attrs.maxswap = 32;
attrs.maxtmp = 64;
attrs.ndisks = 1;
worker_t* w = make_test_worker("order-host-1", &attrs);
TEST_ASSERT_NOT_NULL(w);
hostattrs_t new_attrs = attrs;
new_attrs.hosttype = makeString("LINUX");
new_attrs.hostmodel = makeString("x86");
xce_hostconfig_t hconfig;
memset(&hconfig, 0, sizeof(hconfig));
hconfig.hosttype = (char*)"!";
hconfig.hostmodel = (char*)"!";
hconfig.mxj = 4;
endpoint_t stack_ep;
memset(&stack_ep, 0, sizeof(stack_ep));
rejoin_worker(&g_tc.c, w, &stack_ep, &new_attrs, &hconfig);
TEST_ASSERT_NOT_NULL(w->resync);
endpoint_t* ep = new_test_endpoint();
/* The exact two calls on_worker_join() makes, in the exact order. */
worker_push_config(&g_tc.c, ep);
worker_send_resync_query(&g_tc.c, w, ep);
xce_request_t first;
pop_and_decode(ep, &first);
TEST_ASSERT_EQUAL_INT_MESSAGE(OPCode_push, first.opcode,
"the config push must be the FIRST packet on the wire");
xce_free_request(&first);
xce_request_t second;
pop_and_decode(ep, &second);
TEST_ASSERT_EQUAL_INT_MESSAGE(OPCode_jobStateQuery, second.opcode,
"the reconciliation query must be the SECOND packet on the wire, "
"strictly after the config push");
xce_free_request(&second);
assert_queue_empty(ep);
dereference(new_attrs.hosttype);
dereference(new_attrs.hostmodel);
endpoint_fini(ep);
free(ep);
teardown_test_cluster();
printf("test_config_push_then_resync_query_wire_order: PASS\n");
}
/*
* Req 3.7 (config-delivery gap): worker_push_config() must carry the
* REAL, currently-configured workerResyncWindowMs value -- not a
* hardcoded constant. Two distinct configured values must decode back to
* two distinct wire values.
*/
static void test_worker_push_config_carries_configured_resync_window(void)
{
setup_test_cluster();
cluster_config_t cc;
memset(&cc, 0, sizeof(cc));
cc.clustername = (char*)"test-cluster";
cc.worker_resync_window_ms = 424242;
myconf_t myconf;
memset(&myconf, 0, sizeof(myconf));
myconf.cluster_config = &cc;
g_tc.c.myconf = &myconf;
endpoint_t* ep = new_test_endpoint();
TEST_ASSERT_EQUAL_INT(0, worker_push_config(&g_tc.c, ep));
xce_request_t decoded;
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT(OPCode_push, decoded.opcode);
TEST_ASSERT_EQUAL_INT(Request_body_configpushmsg, decoded.body_which);
TEST_ASSERT_NOT_NULL(decoded.body.configpushmsg);
TEST_ASSERT_EQUAL_INT(ConfigData_data_workerconfig,
decoded.body.configpushmsg->data_which);
TEST_ASSERT_NOT_NULL(decoded.body.configpushmsg->data.conf4worker);
TEST_ASSERT_EQUAL_INT_MESSAGE(424242,
decoded.body.configpushmsg->data.conf4worker->worker_resync_window_ms,
"the pushed value must be the real configured ClusterConfig."
"workerResyncWindowMs, not a hardcoded constant");
xce_free_request(&decoded);
assert_queue_empty(ep);
endpoint_fini(ep);
free(ep);
/* A second, different configured value must decode back differently --
* confirms the field is genuinely read from config each call, not
* memoized/hardcoded from the first call above. */
cc.worker_resync_window_ms = 7;
ep = new_test_endpoint();
TEST_ASSERT_EQUAL_INT(0, worker_push_config(&g_tc.c, ep));
pop_and_decode(ep, &decoded);
TEST_ASSERT_EQUAL_INT_MESSAGE(7,
decoded.body.configpushmsg->data.conf4worker->worker_resync_window_ms,
"a second, different configured value must be reflected on the "
"wire, proving the field is read live, not cached");
xce_free_request(&decoded);
endpoint_fini(ep);
free(ep);
teardown_test_cluster();
printf("test_worker_push_config_carries_configured_resync_window: "
"PASS\n");
}
/* ── Dispatch ───────────────────────────────────────────────────────── */
void test_worker_join(void)
@ -716,4 +1348,17 @@ void test_worker_join(void)
/* Task 1.3: rejoin_worker (reconnect with admin-bit preservation) */
RUN_TEST(test_rejoin_worker_updates_attrs_and_ep);
RUN_TEST(test_rejoin_worker_preserves_admin_bits);
/* worker-rejoin-resync task 6.1: apply_join_state arms a
* reconciliation round; worker_send_resync_query sends it; config
* push then query in that fixed order; the workerResyncWindowMs
* config-delivery gap. */
RUN_TEST(test_rejoin_worker_arms_reconciliation_round_when_jobs_exist);
RUN_TEST(test_rejoin_worker_no_round_when_no_jobs);
RUN_TEST(test_rejoin_worker_no_round_when_only_terminal_jobs);
RUN_TEST(test_init_worker_arms_reconciliation_round_when_jobs_exist);
RUN_TEST(test_worker_send_resync_query_sends_armed_round);
RUN_TEST(test_worker_send_resync_query_sends_nothing_when_no_round);
RUN_TEST(test_config_push_then_resync_query_wire_order);
RUN_TEST(test_worker_push_config_carries_configured_resync_window);
}

132
src/controller/worker.c
View file

@ -7,6 +7,7 @@
#include "cluster.h"
#include "controller.capnp.h"
#include "persist.h"
#include "resync.h" /* worker-rejoin-resync task 6.1 */
#include "wobj.h"
#include "worker.h"
#include "xce/cluster_def.h"
@ -331,6 +332,24 @@ static int apply_join_state(cluster_t* c, worker_t* w, endpoint_t* ep,
w->host.status &= ~XCE_WORKER_UNAVAIL;
w->ep = ep;
/* worker-rejoin-resync task 6.1 (Req 1.1-1.3): the exact moment this
* host becomes "connected" again -- UNAVAIL cleared, ep set -- is the
* natural trigger point for job-state reconciliation (design.md
* Architecture: "rejoin_worker/apply_join_state ... already run under
* the per-worker write lock at the exact moment w->ep is set... this is
* the moment the host becomes connected again"). apply_join_state is
* shared by BOTH init_worker (first-ever join) and rejoin_worker
* (reconnect, whether a plain network reconnect or a rejoin to a new
* primary after controller failover) under the SAME already-held write
* lock, so this single call site satisfies Req 1.3 without needing to
* separately determine which of the three a given join is.
* resync_arm_round() is itself a documented no-op (Req 1.2) when the
* host has no non-terminal jobs currently assigned to it -- only ARMS
* the round here (w->resync + XCE_WORKER_RECONCILING); the actual query
* send happens later, from on_worker_join, strictly after the join
* reply and the config push (see worker_send_resync_query() below). */
resync_arm_round(c, w);
if (strcmp(hconfig->hosttype, "!") != 0) {
dereference(w->host.attrs->hosttype);
w->host.attrs->hosttype = makeString(hconfig->hosttype);
@ -586,7 +605,18 @@ static int check_conf4worker_(xce_config_t* c)
return 0;
}
/*
* worker_push_config is non-static in worker.c when XCE_TESTING is defined
* -- worker-rejoin-resync task 6.1, mirroring apply_join_state's own
* XCE_TESTING exposure convention above, so tests can drive the exact
* config-push-then-query wire ordering directly (see worker_send_resync_
* query() below) without needing a live join flow.
*/
#ifdef XCE_TESTING
int worker_push_config(cluster_t* c, endpoint_t* ep)
#else
static int worker_push_config(cluster_t* c, endpoint_t* ep)
#endif
{
xce_configdata_t configdata;
int rc = 0;
@ -610,6 +640,20 @@ static int worker_push_config(cluster_t* c, endpoint_t* ep)
configdata.data.conf4worker->clustername
= c->myconf->cluster_config->clustername;
/* worker-rejoin-resync task 6.1 (Req 3.7): mirrors the clustername
* field precedent immediately above exactly -- pulled directly from
* ClusterConfig rather than the generic confvars key/value list, no new
* opcode. Closes the gap task 4.1 discovered and documented in
* xcecd_finished_job_retention_set_window_ms()'s own doc comment
* (xcecd_worker_utils.h): "the worker daemon has no runtime delivery
* path for this ClusterConfig field today... this setter is the seam a
* later task wires the real operator-configured value into once that
* delivery path exists." This is that seam's controller-side half; the
* worker-side receive half is controller_connector.c's new
* handle_pushconfig() (task 6.1). */
configdata.data.conf4worker->worker_resync_window_ms
= c->myconf->cluster_config->worker_resync_window_ms;
rc = xce_pushconfig_req(ep, c->myhost, &configdata);
FREEUP(configdata.data.conf4worker);
@ -619,14 +663,102 @@ static int worker_push_config(cluster_t* c, endpoint_t* ep)
return rc;
}
/*
* worker_send_resync_query — worker-rejoin-resync task 6.1 (Req 1.1-1.3,
* 2.1, 2.2): sends the round's job-state query to a worker for which
* apply_join_state (above) already armed a reconciliation round earlier in
* this same join. A worker with no armed round (w->resync == NULL, the
* documented Req 1.2 no-op case) gets no query, silently -- this function
* simply has nothing to do in that case, not an error.
*
* Precondition: caller does NOT hold w's lock (on_worker_join's call site,
* below, runs after handle_worker_join's own lock scope has already ended)
* -- this function acquires w's write lock itself, narrowly, only to copy
* the round's queried-job-ID set, then releases it before the actual
* (non-blocking, fire-and-forget, no reply awaited -- matching the config
* push immediately above it) network send. The jobids array is copied
* rather than sent directly from w->resync->queried_jobids so the send can
* happen without holding w's lock across it: a concurrent rejoin of the
* SAME host on another connection could otherwise call resync_arm_round()
* again mid-send (resync_resolve_round() frees the prior round's array),
* which would make an unlocked read of the raw pointer a use-after-free.
*
* XCE_TESTING-exposed (mirrors apply_join_state's/worker_push_config's own
* convention above) so tests can drive this directly with a real,
* fixture-armed worker_t, independent of get_ep_worker()'s endpoint->worker
* resolution (on_worker_join's own concern, not this function's).
*/
#ifdef XCE_TESTING
int worker_send_resync_query(cluster_t* c, worker_t* w, endpoint_t* ep)
#else
static int worker_send_resync_query(cluster_t* c, worker_t* w, endpoint_t* ep)
#endif
{
xce_job_state_query_t query;
uint64_t* ids = NULL;
int n = 0;
if (c == NULL || w == NULL || ep == NULL)
return -EINVAL;
lock(w, RTTI_LOCK_WRITE);
if (w->resync != NULL) {
n = w->resync->n_queried_jobids;
ids = (uint64_t*)malloc((size_t)n * sizeof(uint64_t));
if (ids != NULL) {
memcpy(ids, w->resync->queried_jobids,
(size_t)n * sizeof(uint64_t));
} else {
ERROR("worker_send_resync_query: worker %s: failed to copy "
"%d queried jobid(s) -- query not sent this join (the "
"round's own Req 5.3 timeout still resolves it)",
(w->host.name != NULL) ? w->host.name : "?", n);
n = 0;
}
}
unlock(w);
if (n <= 0) {
free(ids);
return 0;
}
memset(&query, 0, sizeof(query));
query.n_total = n;
query.jobids = ids;
xce_job_state_query_req(ep, c->myhost, &query, NULL, NULL, 0);
free(ids);
return 0;
}
/* consider to push jobs on hosts to */
int on_worker_join(void* cls, endpoint_t* ep)
{
cluster_t* c = (cluster_t*)cls;
worker_t* w;
int rc = 0;
rc = worker_push_config(c, ep);
/* worker-rejoin-resync task 6.1 (Req 1.1-1.3, 2.1, 2.2): if
* apply_join_state (rejoin_worker/init_worker) armed a reconciliation
* round for this worker earlier in this same join, send its query now
* -- strictly AFTER the config push above (design.md File Structure
* Plan for worker.c: "config push, then query", both fire-and-forget,
* never an extension of the join call itself). get_ep_worker() is the
* same ep->worker_t resolution job.c's update_job_rusages() already
* relies on (misc.c) -- w == NULL (endpoint not resolvable to a known
* worker) degrades to "no query sent," matching worker_send_resync_
* query()'s own no-round no-op, not a crash. */
w = get_ep_worker(c, ep);
if (w != NULL) {
worker_send_resync_query(c, w, ep);
dereference(w);
}
return rc;
}

5
src/include/typemap.h
View file

@ -72,6 +72,11 @@ typedef struct {
char* clustername;
int n_confvars;
xce_config_t** confvars;
/* worker-rejoin-resync task 6.1 (Req 3.7): mirrors clustername above --
* pulled directly from c->myconf->cluster_config->worker_resync_window_ms
* (worker.c's worker_push_config()), not part of the generic confvars
* key/value list. */
int worker_resync_window_ms;
} xce_conf4worker_t;
typedef struct {

5
src/schema/controller.capnp
View file

@ -249,8 +249,9 @@ struct Conf4Sched $C.mapname("xce_conf4sched_t"){
}
struct Conf4Worker $C.mapname("xce_conf4worker_t") {
clustername @0: Text;
configvars @1: List(ConfigVar) $C.mapname("confvars") $C.maplistcount("n_confvars");
clustername @0: Text;
configvars @1: List(ConfigVar) $C.mapname("confvars") $C.maplistcount("n_confvars");
workerResyncWindowMs @2: UInt32 $C.mapname("worker_resync_window_ms");
}
struct ConfigData $C.mapname("xce_configdata_t") {

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

@ -1663,6 +1663,54 @@ static int handle_jobstatequery(xce_request_t* request)
return 0;
}
/**
* @brief Handles an incoming OPCode_push config-push message from the
* controller (worker-rejoin-resync task 6.1, Requirement 3.7).
*
* Confirmed by inspection: before this task, myreq_handler's dispatch
* switch below had NO case for OPCode_push at all -- a config push
* (worker_push_config()/on_worker_join(), controller/worker.c) fell through
* to the `default: break;` case and was silently discarded in its
* entirety, even for the pre-existing clustername field.
* xcecd_finished_job_retention_set_window_ms()'s own doc comment
* (xcecd_worker_utils.h, task 4.1) independently confirms this same gap:
* "the worker daemon has no runtime delivery path for this ClusterConfig
* field today... this setter is the seam a later task wires the real
* operator-configured value into once that delivery path exists." This
* function is that delivery path, scoped narrowly to the ONE field this
* task needs (workerResyncWindowMs) -- clustername/confvars remain
* unconsumed by the worker, unchanged and out of this task's boundary; a
* general config-receive rework is not attempted here.
*
* Fire-and-forget, matching worker_push_config()'s own controller-side
* send (no reply is expected or sent back).
*
* @param request Pointer to the xce_request_t structure containing the
* request data.
* @return int 0 always -- a malformed/unexpected payload shape (wrong
* body_which, wrong ConfigData union arm, NULL conf4worker) is
* simply ignored, not an error to report back.
*/
static int handle_pushconfig(xce_request_t* request)
{
xce_configdata_t* configdata;
if (request->body_which != Request_body_configpushmsg)
return 0;
configdata = request->body.configpushmsg;
if (configdata == NULL
|| configdata->data_which != ConfigData_data_workerconfig
|| configdata->data.conf4worker == NULL) {
return 0;
}
xcecd_finished_job_retention_set_window_ms(
(uint64_t)configdata->data.conf4worker->worker_resync_window_ms);
return 0;
}
/**
* @brief Handles incoming requests from the controller.
*
@ -1691,6 +1739,9 @@ static int myreq_handler(xce_request_t* request, void* args)
case OPCode_jobStateQuery:
ret = handle_jobstatequery(request);
break;
case OPCode_push:
ret = handle_pushconfig(request);
break;
default:
break;
}

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

@ -156,6 +156,23 @@ static const xce_job_state_query_t* g_resync_handle_query_page_last_page = NULL;
* touching it directly. */
static int g_resync_reset_call_count = 0;
/* worker-rejoin-resync task 6.1 (Requirement 3.7): link-satisfying spy stub
* for xcecd_worker_utils.c's real retention-window setter -- this file's
* white-box target is controller_connector.c only (does not #include
* xcecd_worker_utils.c), same rationale as the xcecd_resync.c stubs above.
* Lets the dispatch-wiring tests below observe that handle_pushconfig()
* (myreq_handler's new OPCode_push case) actually reaches this function
* with the decoded value, without needing a real xcecd_worker_utils.c
* retention store backing it. */
static int g_retention_set_call_count;
static uint64_t g_retention_window_ms_set;
void xcecd_finished_job_retention_set_window_ms(uint64_t window_ms)
{
g_retention_set_call_count++;
g_retention_window_ms_set = window_ms;
}
bool xcecd_resync_is_pending(void) { return g_test_resync_pending; }
void xcecd_resync_reset(void)
@ -201,6 +218,10 @@ void setUp(void)
g_resync_handle_query_page_call_count = 0;
g_resync_handle_query_page_last_page = NULL;
g_resync_reset_call_count = 0;
/* worker-rejoin-resync task 6.1: reset the retention-window setter spy. */
g_retention_set_call_count = 0;
g_retention_window_ms_set = 0;
}
void tearDown(void) { return; }
@ -902,6 +923,160 @@ test_myreq_handler_jobstatequery_without_connection_is_rejected(void)
"PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* worker-rejoin-resync task 6.1 (Requirement 3.7): myreq_handler's new
* OPCode_push case (handle_pushconfig) must apply a received
* workerResyncWindowMs value to the retention window before the push
* completes.
*
* Before this task, myreq_handler had NO case for OPCode_push at all --
* confirmed by inspection (this file's own white-box #include of the real,
* current controller_connector.c) -- so a config push fell through to
* `default: break;` and was silently discarded in its entirety, even for
* the pre-existing clustername field. xcecd_finished_job_retention_set_
* window_ms()'s own doc comment (xcecd_worker_utils.h, task 4.1)
* independently confirms the same gap. This is the first test coverage of
* that dispatch case existing at all.
* ════════════════════════════════════════════════════════════════════════ */
/* Builds a real, wire-encoded OPCode_push carrying a Conf4Worker payload
* with the given workerResyncWindowMs value, pops it back off the queue,
* and decodes it with the real xce_decode() -- exercising the actual
* capnp encode/decode round trip (controller.capnp.c, generated), not just
* a directly-constructed xce_request_t. Mirrors src/lib/testsuite/
* test_job_state_wire.c's own real-endpoint round-trip pattern. Caller
* must xce_free_request(out) and endpoint_fini()/free() the endpoint. */
static endpoint_t*
build_and_decode_pushconfig(uint32_t worker_resync_window_ms,
xce_request_t* out)
{
xce_conf4worker_t conf4worker = { 0 };
conf4worker.clustername = (char*)"test-cluster";
conf4worker.worker_resync_window_ms = worker_resync_window_ms;
xce_configdata_t configdata = { 0 };
configdata.data_which = ConfigData_data_workerconfig;
configdata.data.conf4worker = &conf4worker;
char nm[] = "test-controller";
char ky[] = "test-key";
xce_peerinfo_t from = { .name = nm, .key = ky };
endpoint_t* ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
TEST_ASSERT_EQUAL_INT(0, xce_pushconfig_req(ep, &from, &configdata));
data_item_t* it = endpoint_deq(ep);
TEST_ASSERT_NOT_NULL(it);
io_packet_t* pkt = ITPTR_TO_PACKET(it);
memset(out, 0, sizeof(*out));
TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, out));
return ep;
}
/* A config push carrying a non-default workerResyncWindowMs value must be
* applied to the retention window -- synchronously, within myreq_handler's
* own call, i.e. before the push "completes" from the caller's point of
* view (Requirement 3.7's own observable completion criterion). */
static void
test_myreq_handler_routes_pushconfig_workerconfig_to_retention_setter(void)
{
xce_request_t decoded;
endpoint_t* ep = build_and_decode_pushconfig(4242, &decoded);
int rc = myreq_handler(&decoded, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, g_retention_set_call_count,
"OPCode_push carrying a Conf4Worker payload must reach "
"xcecd_finished_job_retention_set_window_ms() exactly once -- the "
"dispatch case task 6.1 adds (no case existed for OPCode_push "
"before this task)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
4242, g_retention_window_ms_set,
"the applied value must be the real value decoded off the wire, "
"not a hardcoded/default one");
xce_free_request(&decoded);
endpoint_fini(ep);
free(ep);
printf("test_myreq_handler_routes_pushconfig_workerconfig_to_retention_"
"setter: PASS\n");
}
/* A second, different value must also be applied -- confirms the setter is
* called with the field genuinely read from each push, not a fixed
* constant from the first call. */
static void
test_myreq_handler_routes_pushconfig_second_value_also_applied(void)
{
xce_request_t decoded;
endpoint_t* ep = build_and_decode_pushconfig(7, &decoded);
int rc = myreq_handler(&decoded, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_retention_set_call_count);
TEST_ASSERT_EQUAL_UINT64(7, g_retention_window_ms_set);
xce_free_request(&decoded);
endpoint_fini(ep);
free(ep);
printf("test_myreq_handler_routes_pushconfig_second_value_also_applied: "
"PASS\n");
}
/* A config push for a DIFFERENT ConfigData union arm (e.g. a scheduler's
* Conf4Sched, or the raw ClusterConfig arm) must NOT be misread as a
* Conf4Worker payload -- the retention setter must not fire at all. */
static void test_myreq_handler_pushconfig_ignores_non_workerconfig_data(void)
{
cluster_config_t clusterconfig = { 0 };
clusterconfig.clustername = (char*)"test-cluster";
xce_configdata_t configdata = { 0 };
configdata.data_which = ConfigData_data_clusterconfig;
configdata.data.clusterconfig = &clusterconfig;
xce_request_t request = { 0 };
request.opcode = OPCode_push;
request.body_which = Request_body_configpushmsg;
request.body.configpushmsg = &configdata;
int rc = myreq_handler(&request, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_retention_set_call_count,
"a non-workerconfig ConfigData arm must never reach the retention "
"setter");
printf("test_myreq_handler_pushconfig_ignores_non_workerconfig_data: "
"PASS\n");
}
/* A NULL configpushmsg (defensive: should never happen for a genuinely
* decoded OPCode_push, but the handler must not crash) must also be a
* silent no-op. */
static void test_myreq_handler_pushconfig_ignores_null_payload(void)
{
xce_request_t request = { 0 };
request.opcode = OPCode_push;
request.body_which = Request_body_configpushmsg;
request.body.configpushmsg = NULL;
int rc = myreq_handler(&request, NULL);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, g_retention_set_call_count);
printf("test_myreq_handler_pushconfig_ignores_null_payload: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 2.3 — structural regression guard: xcecli_set_reconnect_handler()
* must be called AFTER xcecli_start() succeeds, never before.
@ -1252,6 +1427,11 @@ int main(void)
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_myreq_handler_routes_pushconfig_workerconfig_to_retention_setter);
RUN_TEST(test_myreq_handler_routes_pushconfig_second_value_also_applied);
RUN_TEST(test_myreq_handler_pushconfig_ignores_non_workerconfig_data);
RUN_TEST(test_myreq_handler_pushconfig_ignores_null_payload);
RUN_TEST(
test_reconnect_handler_registered_after_start_not_before_in_source);
RUN_TEST(