From 7547a8f14b45d43d76828113097b31fcd229cca5 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 26 Jul 2026 04:06:42 +0000 Subject: [PATCH] fix(worker): replicate admin host-control status updates to standbys (bug-1472) 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. --- .../testsuite/test_apply_committed_entry.c | 199 ++++++++++++++++++ src/controller/worker.c | 45 +++- 2 files changed, 241 insertions(+), 3 deletions(-) diff --git a/src/controller/testsuite/test_apply_committed_entry.c b/src/controller/testsuite/test_apply_committed_entry.c index ce88720f..853be5fb 100644 --- a/src/controller/testsuite/test_apply_committed_entry.c +++ b/src/controller/testsuite/test_apply_committed_entry.c @@ -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 diff --git a/src/controller/worker.c b/src/controller/worker.c index 4df8f963..e0c4e3d6 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -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); }