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

fix(worker): replicate admin host-control status updates to standbys (bug-1472)

Browse source 
worker_from_capnp()'s idempotency guard returned -EEXIST unconditionally
whenever the target worker already existed, discarding the decoded WAL
payload entirely. Correct for a duplicate WAL_WORKER_CTRL/JOIN replay,
but wrong for HOOK_DETAIL_WORKER_UPDATE (hostctrl -c/-o admin close/open):
the target existing is the normal case there, yet the status change was
silently dropped on every member other than the one that processed the
RPC directly -- a permanent, unconditional, cluster-wide replication gap
for host admin-status, confirmed via byte-precise checkpoint divergence
on a live 3-node soak.

Now decodes the entry's hook_detail before applying the existing-worker
guard: JOIN keeps its no-op short-circuit; UPDATE replaces (not ORs,
since HOST_OPEN must clear closed bits) the existing worker's
XCE_WORKER_ADMIN_MASK bits with the decoded payload's, in place, under
the worker's own write lock. HOOK_DETAIL_WORKER_DEAD is deliberately
left unfixed -- its replication would need fields (resvals) that never
round-trip through the Worker capnp wire schema at all, and this
codebase already treats liveness state as locally-derived rather than
WAL-replicated elsewhere -- filed separately as bug-1472-remainder.

Independently reviewed: RED->GREEN cycle re-reproduced from scratch,
replace-vs-OR semantics and lock discipline verified directly, DEAD
deferral corroborated against an independent design-invariant comment
this fix didn't originally cite. Suite: 1030 -> 1033 tests, 0 failures.
This commit is contained in:
Rongsong Shen 2026-07-26 04:06:42 +00:00
parent 8b3e2ab03c
commit 7547a8f14b
2 changed files with 241 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

199
src/controller/testsuite/test_apply_committed_entry.c
View file

@ -496,6 +496,202 @@ static void test_ace_worker_ctrl_join_restores_and_notifies(void)
printf("test_ace_worker_ctrl_join_restores_and_notifies: PASS\n");
}
/* Builds a [uint32 detail][capnp Worker] WAL_WORKER_CTRL payload for a
* worker named `name` carrying `status`. Mirrors the JOIN test's inline
* construction above, factored out so the UPDATE-case tests below (bug-1472)
* can build a payload against an arbitrary pre-existing worker name/status.
* Caller frees *pkt_out. */
static void tace_build_worker_ctrl_pkt(const char* name, uint32_t status,
uint32_t detail, uint8_t** pkt_out,
size_t* total_out)
{
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = (char*)"unknown";
attrs.hostmodel = (char*)"unknown";
xce_worker_t wdef;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char*)name;
wdef.attrs = &attrs;
wdef.status = status;
uint8_t* wbuf = NULL;
size_t wsize = 0;
TEST_ASSERT_EQUAL_INT(0, tace_encode_worker(&wdef, &wbuf, &wsize));
size_t total = sizeof(detail) + wsize;
uint8_t* pkt = malloc(total);
TEST_ASSERT_NOT_NULL(pkt);
memcpy(pkt, &detail, sizeof(detail));
memcpy(pkt + sizeof(detail), wbuf, wsize);
free(wbuf);
*pkt_out = pkt;
*total_out = total;
}
/* bug-1472: worker_ctrl() (HOST_OPEN/HOST_CLOSE, worker.c:732-794) mutates
* only the admin-mutable status bits on the member that received the RPC,
* then fires the SAME WAL_WORKER_CTRL/HOOK_DETAIL_WORKER_UPDATE apply path
* on every OTHER member (standby, or this member's own WAL replay). Before
* the fix, worker_from_capnp's existing-worker guard returned -EEXIST
* unconditionally, silently discarding the decoded status change -- an
* admin hostctrl -c/-o against an already-joined host never propagated.
* This proves an UPDATE entry against an existing worker actually mutates
* it and fires the notification, matching worker_ctrl's own encode (which
* always carries the worker's post-mutation w->host.status). */
static void test_ace_worker_ctrl_update_applies_to_existing_worker(void)
{
tace_setup();
/* Pre-existing worker (e.g. from an earlier JOIN), currently open. */
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = (char*)"unknown";
attrs.hostmodel = (char*)"unknown";
xce_worker_t wdef;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char*)"w-update";
wdef.attrs = &attrs;
wdef.status = 0;
TEST_ASSERT_NOT_NULL(makeWorker(g_tace_c.workermgr, &wdef));
uint8_t* pkt = NULL;
size_t total = 0;
tace_build_worker_ctrl_pkt("w-update",
XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN,
(uint32_t)HOOK_DETAIL_WORKER_UPDATE, &pkt,
&total);
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, tace_spy_hook,
NULL);
int rc = apply_committed_entry(&g_tace_p, WAL_WORKER_CTRL, 41, pkt,
(uint32_t)total);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"an UPDATE against an already-existing worker must succeed (0), not "
"-EEXIST -- bug-1472: -EEXIST silently discarded the status change");
worker_t* w = NULL;
lock(g_tace_c.workermgr, RTTI_LOCK_READ);
key_find(g_tace_c.workermgr, (void*)"w-update", (void**)&w);
unlock(g_tace_c.workermgr);
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_TRUE_MESSAGE(
(w->host.status & (XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN))
== (XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN),
"the existing worker's admin status must be replaced with the "
"decoded UPDATE payload's, not silently discarded");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"a genuinely-applied UPDATE must fire exactly one notification");
free(pkt);
tace_teardown();
printf("test_ace_worker_ctrl_update_applies_to_existing_worker: PASS\n");
}
/* worker_ctrl's HOST_OPEN clears the closed bits via assignment-style
* &= ~(...) (worker.c:773), not a pure OR -- so the fix must REPLACE the
* existing worker's admin-masked bits with the decoded payload's, not OR
* them in (an OR-based fix would leave a re-opened worker stuck closed
* forever). This guards that exact semantics: apply an UPDATE encoding
* status=0 (no admin bits, i.e. a HOST_OPEN) against a worker that starts
* admin-closed and confirms the closed bits are actually cleared. */
static void test_ace_worker_ctrl_update_replaces_not_ors_admin_bits(void)
{
tace_setup();
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = (char*)"unknown";
attrs.hostmodel = (char*)"unknown";
xce_worker_t wdef;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char*)"w-reopen";
wdef.attrs = &attrs;
wdef.status = XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN;
TEST_ASSERT_NOT_NULL(makeWorker(g_tace_c.workermgr, &wdef));
uint8_t* pkt = NULL;
size_t total = 0;
tace_build_worker_ctrl_pkt("w-reopen", 0,
(uint32_t)HOOK_DETAIL_WORKER_UPDATE, &pkt,
&total);
int rc = apply_committed_entry(&g_tace_p, WAL_WORKER_CTRL, 42, pkt,
(uint32_t)total);
TEST_ASSERT_EQUAL_INT(0, rc);
worker_t* w = NULL;
lock(g_tace_c.workermgr, RTTI_LOCK_READ);
key_find(g_tace_c.workermgr, (void*)"w-reopen", (void**)&w);
unlock(g_tace_c.workermgr);
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
w->host.status & (XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN),
"a HOST_OPEN UPDATE must CLEAR the closed bits (replace semantics) "
"-- an OR-based fix would incorrectly leave the worker closed");
free(pkt);
tace_teardown();
printf("test_ace_worker_ctrl_update_replaces_not_ors_admin_bits: PASS\n");
}
/* Regression guard (bug-1472 fix must NOT change JOIN's behavior): a
* duplicate JOIN replay against an already-registered worker (e.g. WAL
* replay / resync redelivery) must remain a genuine no-op -- -EEXIST
* translated to 0 by apply_committed_entry, no status mutation, no
* notification. */
static void test_ace_worker_ctrl_join_duplicate_existing_is_noop(void)
{
tace_setup();
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = (char*)"unknown";
attrs.hostmodel = (char*)"unknown";
xce_worker_t wdef;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char*)"w-dupjoin";
wdef.attrs = &attrs;
wdef.status = 0;
TEST_ASSERT_NOT_NULL(makeWorker(g_tace_c.workermgr, &wdef));
uint8_t* pkt = NULL;
size_t total = 0;
tace_build_worker_ctrl_pkt("w-dupjoin",
XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN,
(uint32_t)HOOK_DETAIL_WORKER_JOIN, &pkt,
&total);
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, tace_spy_hook,
NULL);
int rc = apply_committed_entry(&g_tace_p, WAL_WORKER_CTRL, 43, pkt,
(uint32_t)total);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"apply_committed_entry translates worker_from_capnp's -EEXIST to 0 "
"(idempotent no-op)");
worker_t* w = NULL;
lock(g_tace_c.workermgr, RTTI_LOCK_READ);
key_find(g_tace_c.workermgr, (void*)"w-dupjoin", (void**)&w);
unlock(g_tace_c.workermgr);
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
w->host.status & (XCE_WORKER_CLOSED | XCE_WORKER_CLOSED_BY_ADMIN),
"a duplicate JOIN replay must NOT mutate the existing worker's "
"status -- this pre-existing JOIN behavior is intentionally "
"unchanged by the bug-1472 UPDATE fix");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_tace_hook_fires,
"a duplicate JOIN no-op must not fire a notification");
free(pkt);
tace_teardown();
printf("test_ace_worker_ctrl_join_duplicate_existing_is_noop: PASS\n");
}
/* ── WAL_JOB_ARRAY_SUBMIT (job-array-single-wal-record task 3.2) ────────── */
static job_submit_data_t tace_array_make_spec(void)
@ -748,6 +944,9 @@ void test_apply_committed_entry(void)
RUN_TEST(test_ace_job_state_transition_rejects_backward_suppresses_notify);
RUN_TEST(test_ace_queue_ctrl_updates_and_notifies);
RUN_TEST(test_ace_worker_ctrl_join_restores_and_notifies);
RUN_TEST(test_ace_worker_ctrl_update_applies_to_existing_worker);
RUN_TEST(test_ace_worker_ctrl_update_replaces_not_ors_admin_bits);
RUN_TEST(test_ace_worker_ctrl_join_duplicate_existing_is_noop);
RUN_TEST(test_ace_job_array_submit_constructs_all_and_notifies_once);
RUN_TEST(test_ace_job_array_submit_idempotent_replay_no_extra_notify);
#ifdef XCE_TESTING

45
src/controller/worker.c
View file

@ -823,12 +823,37 @@ int do_hostcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
/* Reconstruct a worker from a WAL_WORKER_CTRL Cap'n Proto payload.
* Payload layout: [uint32_t detail][Cap'n Proto xce_worker_t].
* Returns -EEXIST if the worker is already registered (idempotent replay). */
*
* If the worker is already registered:
* - HOOK_DETAIL_WORKER_JOIN: a duplicate JOIN replay is genuinely a
* no-op -- returns -EEXIST, no mutation (unchanged, pre-bug-1472
* behavior).
* - HOOK_DETAIL_WORKER_UPDATE: the target already existing is the
* NORMAL case (worker_ctrl()'s HOST_OPEN/HOST_CLOSE, worker.c:732-794,
* only ever runs against an already-joined host). bug-1472: replace
* (not OR -- worker_ctrl uses assignment-style clear/set, so a
* HOST_OPEN must be able to CLEAR the closed bits) the existing
* worker's admin-masked status bits with the decoded payload's, then
* return 0 so the caller (persist.c's WAL_WORKER_CTRL apply case)
* falls through to fire the HOOK_DETAIL_WORKER_UPDATE notification
* instead of silently discarding the change.
* - HOOK_DETAIL_WORKER_DEAD: left as -EEXIST (unchanged) -- see
* buglog.json bug-1472 for why DEAD's replication needs its own,
* separately-tracked fix: its mutation isn't confined to
* XCE_WORKER_ADMIN_MASK (XCE_WORKER_UNAVAIL + dead_since + hostattrs
* reset), and the live per-host resource-value array
* (xce_worker_t.resvals, a zarray_t) that mark_worker_unavail()
* zeroes via set_host_slots() never round-trips through the Worker
* capnp schema at all (only resvals_to_sched does) -- a naive
* status-replace here would be an incomplete fix. */
int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len)
{
if (!capnp_payload || len <= sizeof(uint32_t))
return -EINVAL;
uint32_t hook_detail;
memcpy(&hook_detail, capnp_payload, sizeof(hook_detail));
/* Skip the 4-byte detail discriminator written by persist_worker_hook_fun */
const uint8_t* data = (const uint8_t*)capnp_payload + sizeof(uint32_t);
size_t data_len = len - sizeof(uint32_t);
@ -844,13 +869,27 @@ int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len)
decode_Worker(&host, &cw);
capn_free(&c_ctx);
/* Idempotency guard: skip if already present */
/* Idempotency guard: an existing worker means either a duplicate JOIN
* replay (no-op) or an UPDATE targeting an already-joined host (the
* normal case -- must be applied, see bug-1472). */
worker_t* existing = NULL;
lock(c->workermgr, RTTI_LOCK_READ);
int found = key_find(c->workermgr, (void*)host.name, (void**)&existing);
unlock(c->workermgr);
if (found > 0 && existing != NULL)
if (found > 0 && existing != NULL) {
if (hook_detail == HOOK_DETAIL_WORKER_UPDATE) {
lock(existing, RTTI_LOCK_WRITE);
existing->host.status
= (existing->host.status & ~XCE_WORKER_ADMIN_MASK)
| (host.status & XCE_WORKER_ADMIN_MASK);
unlock(existing);
free_Worker(&host);
return 0;
}
return -EEXIST;
}
return restore_worker(c, host);
}