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

feat(config-upgrade-snapshot): restore host admin-state frame safely (create-if-missing)

Browse source 
Adds restore_host_state(cluster_t*, const char*, uint32_t admin_status)
to worker.c: OR's admin bits into an existing worker's status on hit;
on miss, creates a minimal XCE_WORKER_UNAVAIL placeholder via
makeWorker (mirroring the lost_and_found sentinel host precedent),
with a valid non-NULL hostattrs_t so the host's later real join routes
through rejoin_worker (not init_worker) and preserves the restored
admin status. Idempotent under retry (find first, create only if
missing). Standalone -- not yet wired into load_snapshot (task 4).

Requirements: 1.10, 1.11, 5.8
This commit is contained in:
Rongsong Shen 2026-07-24 12:48:28 +00:00
parent 387ea35c9f
commit 5b1600a8b2
6 changed files with 467 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/controller/include/worker.h
View file

@ -13,5 +13,6 @@ extern int init_worker(cluster_t*, endpoint_t*, const char*,
extern int rejoin_worker(cluster_t*, worker_t*, endpoint_t*,
hostattrs_t*, xce_hostconfig_t*);
extern int restore_worker(void*, xce_worker_t);
extern int restore_host_state(cluster_t*, const char*, uint32_t);
extern int worker_from_capnp(cluster_t*, const void*, size_t);
#endif

1
src/controller/meson.build
View file

@ -168,6 +168,7 @@ controller_unit_test_src = [
'testsuite/test_job_array_apply.c',
'testsuite/test_job_array_fence.c',
'testsuite/test_job_array_replay_fence.c',
'testsuite/test_upgrade_snapshot_restore.c',
'ctrlstatus.c',
'commit.c',
'job.c',

2
src/controller/testsuite/controller_test.c
View file

@ -88,6 +88,7 @@ extern void test_apply_pump_integration(void);
extern void test_job_array_apply(void);
extern void test_job_array_fence(void);
extern void test_job_array_replay_fence(void);
extern void test_upgrade_snapshot_restore(void);
void setUp(void) { return; }
@ -245,6 +246,7 @@ int main(int argc, char* argv[])
TEST(test_job_array_apply);
TEST(test_job_array_fence);
TEST(test_job_array_replay_fence);
TEST(test_upgrade_snapshot_restore);
return UNITY_END();
}

361
src/controller/testsuite/test_upgrade_snapshot_restore.c Normal file
View file

@ -0,0 +1,361 @@
/*
* test_upgrade_snapshot_restore.c — config-upgrade-snapshot task 3
*
* TDD tests for restore_host_state() (Requirements 1.10, 1.11, 5.8):
* restoring a host admin-state frame must work whether or not the target
* host has ever joined this member (create-if-missing, since hosts are
* join-provisioned, not config-provisioned -- Design Invariant 1), must be
* idempotent under a retried restore (Design Invariant 3), must never
* clobber a pre-existing worker's non-admin state on a hit, and the
* placeholder it creates on a miss must be safe for the host's later real
* join (routed through rejoin_worker by join.c's own existing
* key_find-based dispatch, unmodified) to adopt -- admin status set on the
* placeholder must still be set after that real join (Design Invariant 6).
*
* RED phase: restore_host_state does not exist yet, so this file fails to
* compile/link until Task 3 adds it to worker.c + worker.h.
*
* Fixture mirrors test_worker_join.c's bare in-memory cluster_t (workermgr +
* resmgr kvmgr instances, no disk/persist) -- restore_host_state only
* touches c->workermgr directly, and the placeholder-then-real-join test
* additionally needs the resmgr resource defs that apply_join_state's
* add_hostattrs_resvals()/set_host_slots() (invoked via rejoin_worker)
* require -- same resource-name list as test_worker_join.c.
*
* collect_confvars() (required for worker.c to link without pulling in the
* full dataex.c) is intentionally NOT redefined here: test_worker_join.c
* already provides it once for the whole controller_unit_test binary.
*/
#include <rtti.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "cluster.h"
#include "kvmgr.h"
#include "resobj.h"
#include "wobj.h"
#include "worker.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xstring.h"
#include "zarray.h"
/* ── Test fixture ────────────────────────────────────────────────────── */
/* Resource names/types matching add_hostattrs_resvals/set_host_slots in
* worker.c -- required so a real join (rejoin_worker) after a placeholder
* restore succeeds instead of failing resource lookups. */
#define N_RES 9
static const char* g_res_names[N_RES] = {
"type", "model", "cpuf", "maxcpus", "maxmem",
"maxswap", "maxtmp", "ndisks", "slots"
};
static const int g_res_valtypes[N_RES] = {
XCE_VAL_STRING, XCE_VAL_STRING,
XCE_VAL_NUMERIC, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC,
XCE_VAL_NUMERIC, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC,
XCE_VAL_NUMERIC
};
typedef struct {
cluster_t c;
var workermgr;
var resmgr;
} tur_cluster_t;
static tur_cluster_t g_tur;
static void tur_setup(void)
{
kvmgr_params_t str_params = {
.key_type = KVMGR_KEY_STRING,
.nslots = 64,
.delta = 16,
.datakiller = NULL,
};
/* Idempotent guard: the xstring intern table is process-global and
* shared across every test file in this binary (mirrors
* test_apply_seam.c / test_replay_apply_seam.c's guard, safer than
* test_worker_join.c's unconditional call since both files run in the
* same controller_unit_test process). */
if (makeString((char*)"tur-probe") == NULL)
xstring_table_init(139);
memset(&g_tur, 0, sizeof(g_tur));
g_tur.workermgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_tur.workermgr);
g_tur.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_tur.resmgr);
for (int i = 0; i < N_RES; i++) {
xce_resource_t rdef;
memset(&rdef, 0, sizeof(rdef));
rdef.name = (char*)g_res_names[i];
rdef.valtype = g_res_valtypes[i];
rdef.flags = 0;
rdef.idx = -1;
xce_resource_t* r = makeResource(g_tur.resmgr, &rdef);
TEST_ASSERT_NOT_NULL_MESSAGE(r, g_res_names[i]);
}
g_tur.c.workermgr = g_tur.workermgr;
g_tur.c.resmgr = g_tur.resmgr;
zarray_init(&g_tur.c.sharedinst, 64, NULL);
}
static void tur_teardown(void)
{
zarray_fini(&g_tur.c.sharedinst);
if (g_tur.resmgr) dereference(g_tur.resmgr);
if (g_tur.workermgr) dereference(g_tur.workermgr);
memset(&g_tur, 0, sizeof(g_tur));
}
static worker_t* tur_find(const char* name)
{
worker_t* w = NULL;
lock(g_tur.c.workermgr, RTTI_LOCK_READ);
key_find(g_tur.c.workermgr, (void*)name, (void**)&w);
unlock(g_tur.c.workermgr);
return w;
}
/* ── Miss: create-if-missing placeholder (Req 1.10) ────────────────────── */
static void test_restore_host_state_creates_placeholder_for_unknown_host(void)
{
tur_setup();
int rc = restore_host_state(&g_tur.c, "never-joined-host",
XCE_WORKER_CLOSED);
TEST_ASSERT_EQUAL_INT(0, rc);
worker_t* w = tur_find("never-joined-host");
TEST_ASSERT_NOT_NULL_MESSAGE(w,
"restore for a host with no local record must create exactly one placeholder");
TEST_ASSERT_TRUE_MESSAGE(w->host.status & XCE_WORKER_CLOSED,
"restored admin bit (CLOSED) must be set on the placeholder");
TEST_ASSERT_TRUE_MESSAGE(w->host.status & XCE_WORKER_UNAVAIL,
"placeholder must be unjoined/dead-until-connected");
TEST_ASSERT_NOT_NULL_MESSAGE(w->host.attrs,
"placeholder hostattrs must be non-NULL (rejoin_worker "
"dereferences it directly on the later real join)");
tur_teardown();
printf("test_restore_host_state_creates_placeholder_for_unknown_host: PASS\n");
}
/* ── Idempotent retry after partial failure (Req 5.8) ──────────────────── */
static void test_restore_host_state_retry_is_idempotent(void)
{
tur_setup();
int rc1 = restore_host_state(&g_tur.c, "retry-host", XCE_WORKER_CLOSED);
TEST_ASSERT_EQUAL_INT(0, rc1);
worker_t* w1 = tur_find("retry-host");
TEST_ASSERT_NOT_NULL(w1);
int idx1 = w1->idx;
/* Simulate a retried restore (e.g. after a partial earlier failure) --
* same name/admin bits, must find-and-continue, never a duplicate
* create or an -EEXIST-style conflict. */
int rc2 = restore_host_state(&g_tur.c, "retry-host", XCE_WORKER_CLOSED);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc2,
"retried restore must not error");
worker_t* w2 = tur_find("retry-host");
TEST_ASSERT_NOT_NULL(w2);
TEST_ASSERT_EQUAL_INT_MESSAGE(idx1, w2->idx,
"retry must reuse the same entry, never create a second one");
TEST_ASSERT_TRUE(w2->host.status & XCE_WORKER_CLOSED);
tur_teardown();
printf("test_restore_host_state_retry_is_idempotent: PASS\n");
}
/* ── Hit on a real, already-joined worker: OR bits only (Req 1.10) ──────── */
static void test_restore_host_state_on_real_worker_only_ors_admin_bits(void)
{
tur_setup();
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_tur.c, &ep, "real-host-1", &attrs, &hconfig);
TEST_ASSERT_GREATER_OR_EQUAL(0, idx);
worker_t* w = tur_find("real-host-1");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_FALSE(w->host.status & XCE_WORKER_UNAVAIL);
TEST_ASSERT_EQUAL_PTR(&ep, w->ep);
int rc = restore_host_state(&g_tur.c, "real-host-1", XCE_WORKER_CLOSED);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_TRUE_MESSAGE(w->host.status & XCE_WORKER_CLOSED,
"admin bit must be OR'd onto the real worker");
TEST_ASSERT_FALSE_MESSAGE(w->host.status & XCE_WORKER_UNAVAIL,
"restore on a real, already-joined worker must not mark it UNAVAIL");
TEST_ASSERT_EQUAL_STRING_MESSAGE("LINUX", w->host.attrs->hosttype,
"restore on a hit must not touch existing attrs");
TEST_ASSERT_EQUAL_PTR_MESSAGE(&ep, w->ep,
"restore on a hit must not touch ep");
tur_teardown();
printf("test_restore_host_state_on_real_worker_only_ors_admin_bits: PASS\n");
}
/* ── Observable (task 3's own line): placeholder -> later real join ────── */
static void
test_restore_host_state_placeholder_then_real_join_preserves_admin_status(void)
{
tur_setup();
/* Host admin-closed while down, then a configuration upgrade restores
* its admin-state frame on a member with no local record of the host's
* join -- must create a placeholder carrying the CLOSED bit. */
int rc = restore_host_state(&g_tur.c, "join-later-host",
XCE_WORKER_CLOSED);
TEST_ASSERT_EQUAL_INT(0, rc);
worker_t* w = tur_find("join-later-host");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_TRUE(w->host.status & XCE_WORKER_CLOSED);
TEST_ASSERT_TRUE(w->host.status & XCE_WORKER_UNAVAIL);
/* The host's real first join. join.c's own dispatch (join.c:90-103)
* does exactly: key_find() -> found w != NULL -> rejoin_worker() (never
* init_worker()). Reproduced directly here (rather than going through
* handle_worker_join) since that needs a live xce_request_t/endpoint_t/
* persist commit-barrier harness this unit test intentionally has none
* of -- the dispatch decision itself is not under test, join.c's is
* unmodified and already covered elsewhere; what's under test is that
* THIS placeholder is a valid, non-crashing input to rejoin_worker(). */
hostattrs_t new_attrs;
memset(&new_attrs, 0, sizeof(new_attrs));
new_attrs.hosttype = makeString("LINUX");
new_attrs.hostmodel = makeString("x86_64");
new_attrs.cpufactor = 1.0f;
new_attrs.maxcpus = 8;
new_attrs.maxmem = 32;
new_attrs.maxswap = 64;
new_attrs.maxtmp = 128;
new_attrs.ndisks = 2;
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));
worker_t* found = NULL;
int findrc = 0;
lock(g_tur.c.workermgr, RTTI_LOCK_READ);
findrc = key_find(g_tur.c.workermgr, (void*)"join-later-host",
(void**)&found);
unlock(g_tur.c.workermgr);
TEST_ASSERT_GREATER_THAN_MESSAGE(-1, findrc,
"join.c's dispatch must find the placeholder (hit, not miss)");
TEST_ASSERT_EQUAL_PTR(w, found);
int idx = rejoin_worker(&g_tur.c, found, &ep, &new_attrs, &hconfig);
TEST_ASSERT_GREATER_OR_EQUAL_MESSAGE(0, idx,
"rejoin_worker must not fail on the placeholder");
TEST_ASSERT_TRUE_MESSAGE(w->host.status & XCE_WORKER_CLOSED,
"admin status set via restore must still be set after the real join");
TEST_ASSERT_FALSE_MESSAGE(w->host.status & XCE_WORKER_UNAVAIL,
"the real join must clear UNAVAIL");
TEST_ASSERT_EQUAL_STRING("LINUX", w->host.attrs->hosttype);
dereference(new_attrs.hosttype);
dereference(new_attrs.hostmodel);
tur_teardown();
printf("test_restore_host_state_placeholder_then_real_join_preserves_admin_status: PASS\n");
}
/* ── Defense-in-depth: non-admin bits are masked out ────────────────────── */
static void test_restore_host_state_masks_non_admin_bits(void)
{
tur_setup();
int rc = restore_host_state(&g_tur.c, "masked-host",
XCE_WORKER_CLOSED | XCE_WORKER_BUSY);
TEST_ASSERT_EQUAL_INT(0, rc);
worker_t* w = tur_find("masked-host");
TEST_ASSERT_NOT_NULL(w);
TEST_ASSERT_TRUE_MESSAGE(w->host.status & XCE_WORKER_CLOSED,
"admin bit must be applied");
TEST_ASSERT_FALSE_MESSAGE(w->host.status & XCE_WORKER_BUSY,
"non-admin bit must be masked out even if a caller passes it");
tur_teardown();
printf("test_restore_host_state_masks_non_admin_bits: PASS\n");
}
/* ── Input validation at the boundary ───────────────────────────────────── */
static void test_restore_host_state_rejects_empty_name(void)
{
tur_setup();
TEST_ASSERT_EQUAL_INT(-EINVAL,
restore_host_state(&g_tur.c, NULL, XCE_WORKER_CLOSED));
TEST_ASSERT_EQUAL_INT(-EINVAL,
restore_host_state(&g_tur.c, "", XCE_WORKER_CLOSED));
tur_teardown();
printf("test_restore_host_state_rejects_empty_name: PASS\n");
}
/* ── Dispatch ───────────────────────────────────────────────────────── */
void test_upgrade_snapshot_restore(void)
{
UnitySetTestFile(__FILE__);
printf("\n>> config-upgrade-snapshot host restore tests (task 3)\n\n");
RUN_TEST(test_restore_host_state_creates_placeholder_for_unknown_host);
RUN_TEST(test_restore_host_state_retry_is_idempotent);
RUN_TEST(test_restore_host_state_on_real_worker_only_ors_admin_bits);
RUN_TEST(test_restore_host_state_placeholder_then_real_join_preserves_admin_status);
RUN_TEST(test_restore_host_state_masks_non_admin_bits);
RUN_TEST(test_restore_host_state_rejects_empty_name);
}

99
src/controller/worker.c
View file

@ -436,6 +436,105 @@ int restore_worker(void* cls, xce_worker_t host)
return idx;
}
/*
* restore_host_state — restore a host admin-state frame (config-upgrade-
* snapshot Requirements 1.10, 1.11, 5.8).
*
* Unlike restore_queue_state() (queues are config-provisioned, so an
* update-only lookup is always safe), a worker_t is join-provisioned: it
* may not exist yet on a member that hasn't locally replayed this host's
* original join record (e.g. one that received the upgrade snapshot via
* the out-of-band install path). So this restore path is create-if-missing:
*
* - Hit (host already registered -- a real joined worker, or an earlier
* placeholder left by a prior restore attempt): OR the restored admin
* bits into the existing status only. Nothing else is touched, so a
* real worker's live attrs/ep/resource values survive untouched.
* - Miss: create a minimal placeholder via makeWorker(), mirroring the
* synthetic lost_and_found sentinel host precedent exactly
* (config_repl.c:1114-1130) -- the same "makeWorker() without a join"
* mechanism, not a new one. The placeholder is marked
* XCE_WORKER_UNAVAIL (unjoined, dead-until-connected) and carries the
* restored admin bits. Its hostattrs_t is non-NULL with valid sentinel
* strings (copy_worker, wobj.c:53, requires non-NULL attrs; downstream
* apply_join_state()/add_hostattrs_resvals() treat hosttype/hostmodel
* as always-valid strings) so the host's later real join -- which
* join.c routes to rejoin_worker(), unmodified, because key_find() now
* finds this placeholder -- doesn't crash on rejoin_worker's direct
* attrs->hosttype dereference (worker.c:376-379).
*
* Idempotent under retry (5.8): find first via our own key_find, create
* only if truly missing. A second call for the same host (e.g. a retried
* restore after an earlier attempt failed partway through the snapshot)
* finds either the real worker or the already-created placeholder and just
* OR's the bits in again -- never a duplicate-create conflict.
*
* admin_status is masked to XCE_WORKER_ADMIN_MASK defensively (mirrors the
* collect side's own discipline, collect_host_frames only ever encoding
* admin-masked bits) so a caller passing an unmasked status word can never
* leak a non-admin bit (e.g. XCE_WORKER_BUSY) onto the restored host.
*/
int restore_host_state(cluster_t* c, const char* name, uint32_t admin_status)
{
worker_t* w = NULL;
int rc = 0;
if (name == NULL || name[0] == '\0')
return -EINVAL;
if (c == NULL || c->workermgr == NULL)
return -EINVAL;
admin_status &= XCE_WORKER_ADMIN_MASK;
/* No reference() pin across the workermgr-unlock/w-lock gap (unlike
* join.c's live-join path): this is called from checkpoint restore
* (ha_handle_snapshot_chunk -> persist_load_snapshot), which DOES run
* on the live vtask pool, not a quiesced phase, so a real first-time
* join for the same host CAN race this call. That race is still safe:
* a concurrent init_worker/makeWorker on a miss here would itself
* find-or-create via the same key_find/makeWorker machinery under
* workermgr's own lock, and either ordering converges to a worker
* with the admin bits OR'd in and UNAVAIL cleared by the real join's
* own apply_join_state() call -- the same "outer key_find, then
* unconditionally call into makeWorker" shape already used elsewhere
* in this codebase (worker_from_capnp/restore_worker,
* create_lost_and_found). Mirrors restore_queue_state's (queue.c:208)
* lock idiom, not a claim that this path is concurrency-free. */
lock(c->workermgr, RTTI_LOCK_READ);
rc = key_find(c->workermgr, (void*)name, (void**)&w);
unlock(c->workermgr);
if (rc < 0 && rc != -ENOENT)
return rc;
if (w != NULL) {
/* Hit: real joined worker or an earlier placeholder -- only OR the
* admin bits in, never touch anything else. */
lock(w, RTTI_LOCK_WRITE);
w->host.status |= admin_status;
unlock(w);
return 0;
}
/* Miss: create a minimal, valid placeholder -- mirrors the
* lost_and_found sentinel host precedent (config_repl.c:1114-1130). */
{
hostattrs_t placeholder_attrs = { .hosttype = "unknown",
.hostmodel = "unknown" };
xce_worker_t placeholder_host = { .name = (char*)name,
.attrs = &placeholder_attrs,
.status = XCE_WORKER_UNAVAIL
| admin_status };
w = makeWorker(c->workermgr, &placeholder_host);
if (w == NULL)
return -errno;
}
return 0;
}
struct worker_config_publish_args {
cluster_t* c;
endpoint_t* ep;