diff --git a/.kiro/specs/worker-rejoin-resync/tasks.md b/.kiro/specs/worker-rejoin-resync/tasks.md index 5d5ac1ae..e56a7af7 100644 --- a/.kiro/specs/worker-rejoin-resync/tasks.md +++ b/.kiro/specs/worker-rejoin-resync/tasks.md @@ -113,7 +113,7 @@ - _Boundary: scheduler candidate filter_ - [ ] 6. Integration: cross-daemon wiring -- [ ] 6.1 Trigger reconciliation on worker rejoin and send the query +- [x] 6.1 Trigger reconciliation on worker rejoin and send the query - At the moment a worker host becomes connected again (first join, network reconnect, or reconnect after controller failover, treated identically), check whether the controller currently holds non-terminal jobs for that host and, if so, arm a reconciliation round for it - Send the round's query to the worker as a follow-up exchange after the join reply and the existing configuration push have both already been sent, not as an extension of the join call itself - Also close a gap task 4.1 discovered and correctly left out of its own boundary: the worker daemon has no delivery path at all today for any `ClusterConfig` field (the existing `worker_push_config`/`on_worker_join` push carries a generic `confvars` key/value list plus a small number of directly-pulled fields — `clustername` is the existing precedent, `src/controller/worker.c` around the `collect_confvars`/`conf4worker` construction). Add `workerResyncWindowMs` to that same push, mirroring the `clustername` field precedent exactly (no new opcode needed — this reuses the existing `OPCode_push`/`WorkerConfig` wire struct this task already touches), and have the worker's existing config-receive handling call `xcecd_finished_job_retention_set_window_ms()` (task 4.1) with the received value so the retention window task 4.1 built is actually operator-tunable end-to-end, not stuck on its documented placeholder default @@ -166,3 +166,4 @@ - **Task 4.3 needed 2 review rounds — a genuine deadlock, not a style nitpick.** Round 1 placed the `resync_pending` gate check as `checking_connection()`'s entry guard (returning early before `xcecli_hello()`/the `needs_rejoin` check could run), so a worker whose connection dropped after a non-final page of a multi-page `OPCode_jobStateQuery` (no further page ever arriving to clear the gate) would have its ONLY connection-recovery mechanism permanently disabled — a direct Requirement 6.1 violation the existing suite didn't catch because no test drove the multi-page-then-disconnect sequence. Round 2 fixed it two ways: (a) narrowed the gate to guard only the `rejoin_call()` step inside the `needs_rejoin` branch, so `xcecli_hello()`/reconnect-detection always run every tick regardless of gate state; (b) added `xcecd_resync_reset()`, called from `xcecd_on_reconnect()` (the same signal that sets `needs_rejoin`), which clears the gate and discards any stalled partial multi-page accumulation the moment a real reconnect completes — bounding the deadlock window to at most one reconnect cycle instead of forever. Round 2's reviewer independently confirmed the fix by reintroducing the round-1 bug and rebuilding, and found the new *behavioral* regression test alone did NOT catch the reintroduced bug — only a new *structural* source-order test (asserting `xcecli_hello()` precedes the `needs_rejoin` check precedes the gate check, textually) actually distinguished correct from broken placement. **Lesson for future tasks in this codebase**: when a fix's correctness hinges on WHERE a check is placed relative to other statements (not just what value it reads), a pure behavioral/outcome test can pass even with the wrong placement if the test's specific stimulus happens not to exercise the broken ordering — add a structural/ordering assertion alongside the behavioral one for placement-sensitive gates. - Task 4.3 (advisory, non-blocking, not fixed in this pass): `resync_pending`/`query_accum` in `xcecd_resync.c` are plain unsynchronized statics touched from two different OS threads — the worker's dedicated connector thread (`checking_connection()`/`xcecd_on_reconnect()`) and a `vtaskmgr` pool thread (`xcecd_resync_handle_query_page()`, reached via the request-dispatch path). This mirrors the pre-existing, design-directed unsynchronized `needs_rejoin` pattern (same codebase convention, deliberately reused per design.md), but is riskier since `xcecd_resync_reset()` calls `free()` on the accumulator rather than just staling a bool read — a narrow, non-deterministic window exists where an independent reconnect could free the buffer while a pool thread is mid-read of the same connection's in-flight query page. Does not reproduce task 4.3's own violated requirement (Req 6.1) and isn't required by any Requirement/AC as written; this project's ASAN/UBSAN test harness has no TSAN, so it also isn't mechanically testable the way the deadlock fix was. Worth a small follow-up (a mutex around the gate+accumulator, or moving accumulation to the connector thread only) if this ever needs hardening beyond what `needs_rejoin`'s own accepted risk profile already tolerates. - **Task 5 found and fixed bug-1751**: design.md's cited candidate-filter site (`scheduler.c:1211`'s `sched_reset_worker_()`, reached only via `sched_begin_session()`) is dead code on both live drivers — `driver.c`/`sim_loop.c` actually call `xce_core_begin_session()` (`core_session.c`), which runs an independently-maintained mirror, `core_reset_workers()`, not listed in design.md's File Structure Plan. Fixed by applying the identical `IS_WORKER_UNAVAIL || IS_WORKER_RECONCILING` disjunct to both sites (parity/dead-code-hygiene on the design-cited one, correctness on the actually-live one) rather than patching only the literally-cited site and leaving the task's own Observable Completion Criterion false in production — same category as bug-1746/bug-1747's design-vs-reality corrections. design.md's File Structure Plan corrected. Also confirmed design.md's Testing Strategy referenced an "existing IS_WORKER_UNAVAIL candidate-filter test" that does not actually exist; new coverage added to `test_core_session.c` instead (white-box `#include`s `core_session.c` already; `scheduler.c`'s own dependency closure is too large to white-box per established precedent, `test_snapshot_load_allocator_regression.c`) — `scheduler.c`'s copy of the fix is therefore verified only by compilation + the identical, tested logic pattern in its mirror, not by a direct test; judged acceptable since it's confirmed dead code today, not a live gap. +- **Task 6.1 found and fixed bug-1752**: the worker daemon's `myreq_handler` had NO dispatch case for `OPCode_push` at all before this task — every prior config push (including the pre-existing `clustername` field, in production since before this spec started) was silently discarded via `default: break;`. Confirmed genuine (diffed `myreq_handler` pre/post, ruled out an alternate consumer — `xcectx.c`'s `vtask_entry` only special-cases the unrelated generic `pushmsg` union arm, not `configpushmsg`) and confirmed `clustername` has never been read anywhere in `src/worker/`, so this was harmless dead infrastructure until `workerResyncWindowMs` became its first real consumer. Fixed narrowly — `handle_pushconfig()` consumes only `worker_resync_window_ms`, leaving `clustername`/`confvars` still unconsumed and explicitly out of this task's boundary — rather than expanding scope to wire up every existing-but-never-consumed push field. Ordering (join reply → config push → conditional arm+query) and the live-configured-value plumbing were both verified via genuine structural tests (real packets popped off the wire, two distinct configured values round-tripped), not weaker behavioral proxies. diff --git a/.wolf/buglog.json b/.wolf/buglog.json index 6c4d1d35..5cb3d5bc 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -20495,6 +20495,33 @@ ], "occurrences": 1, "last_seen": "2026-09-01T09:00:00Z" + }, + { + "id": "bug-1752", + "timestamp": "2026-09-01T10:00:00Z", + "error_message": "worker-rejoin-resync task 6.1 (worker-side config-delivery gap): tasks.md's own task text says to have 'the worker's existing config-receive handling call xcecd_finished_job_retention_set_window_ms() with the received value' -- premised on some existing per-field config-receive dispatch already existing at the worker that just needed one more field wired in. Investigating before implementing (reading src/worker/src/controller_connector.c's myreq_handler dispatch switch and src/lib/xcectx.c's vtask_entry, which decodes every inbound request before myreq_handler ever sees it) found this premise false: myreq_handler's switch had NO case for OPCode_push at all (only OPCode_startjob/runtasks/jobcontrol/fetchfile/jobStateQuery existed) -- a config push (Request_body_configpushmsg, the body_which xce_pushconfig_req()/worker_push_config() send) fell straight through to `default: break;` and was silently discarded in its entirety, for EVERY field including the pre-existing clustername, not just the not-yet-added workerResyncWindowMs. xcecd_finished_job_retention_set_window_ms()'s own doc comment (xcecd_worker_utils.h, task 4.1) had already independently found and documented the same gap ('the worker daemon has no runtime delivery path for this ClusterConfig field today... worker_push_config() only ever pushes hosttype/hostmodel/resource confvars via collect_confvars()... this setter is the seam a later task wires the real operator-configured value into once that delivery path exists') -- task 6.1 is that later task, but its own task text still assumed the delivery path itself already existed when it did not.", + "file": "src/worker/src/controller_connector.c", + "root_cause": "No prior worker-rejoin-resync task (nor any task before this spec) ever needed the worker to consume ANY field off an OPCode_push config push, so myreq_handler's dispatch switch was never extended to handle that opcode at all -- xcectx.c's vtask_entry only special-cases the OTHER OPCode_push shape (Request_body_pushmsg, the generic job/worker/resource list-table push used for scheduler/apiserver connections) before handing off to the registered handler; a Request_body_configpushmsg-shaped OPCode_push passes through unmodified into myreq_handler's default: break; case. This predates worker-rejoin-resync entirely and was simply never exercised or noticed because nothing at the worker previously needed any pushed ClusterConfig field.", + "fix": "FIXED (2026-09-01, task 6.1, found live during implementation, same design-vs-reality-correction pattern as bug-1746/1747/1748/1751 earlier in this spec). Added a new case OPCode_push: to myreq_handler's dispatch switch, routing to a new handle_pushconfig() function that checks body_which == Request_body_configpushmsg and configdata->data_which == ConfigData_data_workerconfig before reading conf4worker->worker_resync_window_ms and calling xcecd_finished_job_retention_set_window_ms() with it -- scoped narrowly to closing the one seam task 6.1 needs (workerResyncWindowMs), not a general config-receive rework: clustername/confvars remain unconsumed by the worker, unchanged and explicitly out of this task's boundary. New tests in src/worker/testsuite/test_controller_connector.c (white-box, real wire encode via xce_pushconfig_req + real xce_decode + myreq_handler dispatch) cover: the real value reaching the retention setter exactly once, a second/different value also applied (proves live read, not memoized), a non-workerconfig ConfigData arm being ignored, and a NULL configpushmsg being a safe no-op.", + "tags": [ + "worker-rejoin-resync", + "task-6.1", + "controller_connector-c", + "myreq_handler", + "requirement-3.7", + "config-push", + "design-reality-mismatch", + "found-live-during-implementation", + "fixed" + ], + "related_bugs": [ + "bug-1746", + "bug-1747", + "bug-1748", + "bug-1751" + ], + "occurrences": 1, + "last_seen": "2026-09-01T10:00:00Z" } ] } diff --git a/src/controller/testsuite/test_worker_join.c b/src/controller/testsuite/test_worker_join.c index e7a4d620..4593cbdb 100644 --- a/src/controller/testsuite/test_worker_join.c +++ b/src/controller/testsuite/test_worker_join.c @@ -22,12 +22,26 @@ #include #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); } diff --git a/src/controller/worker.c b/src/controller/worker.c index 8ffa2b17..002c035c 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -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; } diff --git a/src/include/typemap.h b/src/include/typemap.h index 26e7fa2a..709fc024 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -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 { diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 73bfc108..2cd6846c 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -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") { diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index dfdd7acb..48784109 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -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; } diff --git a/src/worker/testsuite/test_controller_connector.c b/src/worker/testsuite/test_controller_connector.c index f3b5f8a2..bb1513b9 100644 --- a/src/worker/testsuite/test_controller_connector.c +++ b/src/worker/testsuite/test_controller_connector.c @@ -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(