feat(config-upgrade-snapshot): collect host admin-state frames into checkpoint

Adds collect_host_frames + capnp_encode_host_admin_state, walking
c->workermgr the same way collect_job_frames/collect_queue_frames do.
Every host currently in the table gets one HostAdminState frame
(status masked to XCE_WORKER_ADMIN_MASK only, no dynamic/liveness
fields), written into write_snapshot after the queue trailer. Task 2.1
of 19.

Requirements: 1.6, 1.7, 1.9
This commit is contained in:
Rongsong Shen 2026-07-24 09:17:46 +00:00
parent 9e68335aee
commit a71aa3f8a1
3 changed files with 594 additions and 5 deletions

View file

@ -69,6 +69,7 @@
#include "controller.capnp.h"
#include "jobj.h"
#include "qobj.h"
#include "wobj.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xstring.h"
@ -113,6 +114,39 @@ static uint32_t snapent_queue_status(int i)
return g_snapent_queue_statuses[i % (int)SNAPENT_N_QUEUE_STATUSES];
}
/* Host admin-status seed values (task 2.1, Req 1.6/1.7). Deliberately
* includes a ZERO entry (a host with no admin action ever taken -- the
* "collect every host, some just have status=0" case: collect_host_frames
* applies no retention predicate, mirroring collect_queue_frames) and a
* multi-bit combination, alongside single-bit CLOSED_BY_ADMIN/LOCKEDU
* cases. */
static const uint32_t g_snapent_host_admin_statuses[] = {
0u,
XCE_WORKER_CLOSED_BY_ADMIN,
XCE_WORKER_LOCKEDU,
(XCE_WORKER_CLOSED_BY_ADMIN | XCE_WORKER_HIDDEN),
};
#define SNAPENT_N_HOST_ADMIN_STATUSES \
(sizeof(g_snapent_host_admin_statuses) / sizeof(g_snapent_host_admin_statuses[0]))
/* Dynamic/non-admin bits ORed onto every seeded host's REAL status
* alongside its admin bits (Requirement 1.7): a seeded host's frame must
* exclude these even though they are set on the live worker_t, proving
* collect_host_frames actually masks with PERSIST_WORKER_ADMIN_MASK rather
* than merely happening to equal the admin bits because nothing else was
* ever set. */
#define SNAPENT_HOST_DYNAMIC_BITS (XCE_WORKER_UNAVAIL | XCE_WORKER_BUSY)
static void snapent_host_name(int i, char* buf, size_t buf_sz)
{
snprintf(buf, buf_sz, "snapent-host%d", i);
}
static uint32_t snapent_host_admin_status(int i)
{
return g_snapent_host_admin_statuses[i % (int)SNAPENT_N_HOST_ADMIN_STATUSES];
}
/* Deterministic, distinguishing submit-payload values per seeded job index
* (task 2.1 / Implementation Notes: richer than an all-zeroed submit, so a
* round-trip's field-by-field comparison is meaningful). Only the fields the
@ -210,8 +244,40 @@ static void snapent_seed_queues(cluster_t* c, int m)
}
}
/* Seeds n hosts directly into c->workermgr via makeWorker -- the same
* makeWorker()-without-join precedent create_lost_and_found (config_repl.c)
* uses, standing in for a real join the way seeded jobs/queues stand in for
* their own real insertion paths. Each host's REAL status carries its
* admin bits (snapent_host_admin_status) PLUS dynamic/liveness bits
* (SNAPENT_HOST_DYNAMIC_BITS) so a round-trip test can prove the collected
* frame excludes the latter (Requirement 1.7), not merely happens to match
* because nothing else was set. */
static void snapent_seed_hosts(cluster_t* c, int n)
{
for (int i = 0; i < n; i++) {
char name[32];
snapent_host_name(i, name, sizeof(name));
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
xce_worker_t hdef;
memset(&hdef, 0, sizeof(hdef));
hdef.name = name;
hdef.attrs = &attrs; /* makeWorker's copy_worker requires non-NULL attrs */
hdef.status = snapent_host_admin_status(i) | SNAPENT_HOST_DYNAMIC_BITS;
TEST_ASSERT_NOT_NULL_MESSAGE(makeWorker(c->workermgr, &hdef),
"seed host insertion must succeed");
}
}
/* make an entity-snapshot fixture: a real persist_t (private tmpdir) embedded
* in a minimal cluster_t whose jobmgr/qmgr are seeded with n_jobs/m_queues.
* workermgr is always created (empty) too -- tests that need hosts call
* snapent_seed_hosts explicitly afterward (task 2.1), mirroring how
* n_jobs/m_queues seed at construction time but keeping every existing
* 2-argument call site unchanged.
* persist_init registers the cluster back-pointer (p->cluster) so future
* write_snapshot/load_snapshot entity collection can walk c->jobmgr/c->qmgr —
* the same access path collect_jobs/collect_queues (dataex.c) use. */
@ -223,6 +289,9 @@ static snapent_fixture_t* snapent_make(int n_jobs, int m_queues)
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 64,
.delta = 16 };
kvmgr_params_t worker_str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 64,
.delta = 16 };
snapent_fixture_t* f = calloc(1, sizeof(*f));
if (f == NULL)
@ -241,10 +310,12 @@ static snapent_fixture_t* snapent_make(int n_jobs, int m_queues)
TEST_ASSERT_EQUAL_INT(
0, xce_hook_init(&f->c->hooksys, XCE_CONTROLLER_HOOK_MAX));
f->c->jobmgr = makeInstance(&xkvmgr_type, &int_params);
f->c->qmgr = makeInstance(&xkvmgr_type, &str_params);
f->c->jobmgr = makeInstance(&xkvmgr_type, &int_params);
f->c->qmgr = makeInstance(&xkvmgr_type, &str_params);
f->c->workermgr = makeInstance(&xkvmgr_type, &worker_str_params);
TEST_ASSERT_NOT_NULL(f->c->jobmgr);
TEST_ASSERT_NOT_NULL(f->c->qmgr);
TEST_ASSERT_NOT_NULL(f->c->workermgr);
zarray_init(&f->c->sharedinst, 64, NULL);
if (!g_snapent_jobobj_ready) {
@ -276,8 +347,9 @@ static void snapent_teardown(snapent_fixture_t* f)
persist_fini(&f->c->persist);
zarray_fini(&f->c->sharedinst);
if (f->c->qmgr) dereference(f->c->qmgr);
if (f->c->jobmgr) dereference(f->c->jobmgr);
if (f->c->qmgr) dereference(f->c->qmgr);
if (f->c->jobmgr) dereference(f->c->jobmgr);
if (f->c->workermgr) dereference(f->c->workermgr);
xce_hook_fini(&f->c->hooksys);
free(f->c);
@ -942,6 +1014,325 @@ static void test_snapent_checkpoint_queue_status_roundtrip(void)
printf("test_snapent_checkpoint_queue_status_roundtrip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test-local raw checkpoint reader for the host section (task 2.1). */
/* */
/* This task's boundary is write_snapshot ONLY -- load_snapshot's host */
/* restore is task 3/4's job (design.md "Host Restore"), mirroring the */
/* job/queue readers above. Reads the checkpoint's on-disk bytes */
/* directly per design.md's "Checkpoint File Layout": */
/* [57-byte legacy sentinel][uint32_t format_version] */
/* [uint32_t job_count][job_count * {payload_len; capnp Job bytes}] */
/* [uint32_t queue_count][queue_count * {name_len; name; status}] */
/* [uint32_t host_count][host_count * {payload_len; capnp */
/* HostAdminState bytes}] */
/* The queue section is still the pre-task-2.2 raw layout (this task */
/* does not touch collect_queue_frames), so it is skipped using the */
/* same raw shape snapent_read_checkpoint_queues above parses, not */
/* capnp-decoded here -- only the host section, this task's actual */
/* boundary, is decoded via read_HostAdminState/decode_HostAdminState, */
/* the same generated helpers restore_host_state (task 3) will use. */
/* ------------------------------------------------------------------ */
#define SNAPENT_MAX_HNAME_LEN 63
typedef struct {
char name[SNAPENT_MAX_HNAME_LEN + 1];
uint32_t status;
} snapent_decoded_host_t;
/* Reads every host frame in the checkpoint at `path` into `out`
* (caller-allocated array of `max_out` slots). Returns the number of
* frames decoded (<= host_count on disk, capped at max_out), or -1 on any
* structural read error (bad magic, bad/absent format_version, short
* read, oversized job/queue/host payload_len or queue name_len). */
static int snapent_read_checkpoint_hosts(const char* path,
snapent_decoded_host_t* out,
int max_out)
{
int fd = open(path, O_RDONLY);
if (fd < 0)
return -1;
char header[SNAPENT_LEGACY_HEADER_BYTES];
if (read(fd, header, sizeof(header)) != (ssize_t)sizeof(header) ||
memcmp(header, "XCESNAP\0", 8) != 0) {
close(fd);
return -1;
}
uint32_t format_version = 0;
if (read(fd, &format_version, sizeof(format_version)) !=
(ssize_t)sizeof(format_version) ||
format_version != SNAPENT_TEST_FORMAT_VERSION) {
close(fd);
return -1;
}
/* Skip the entire job section. */
uint32_t job_count = 0;
if (read(fd, &job_count, sizeof(job_count)) != (ssize_t)sizeof(job_count)) {
close(fd);
return -1;
}
for (uint32_t i = 0; i < job_count; i++) {
uint32_t payload_len = 0;
if (read(fd, &payload_len, sizeof(payload_len)) !=
(ssize_t)sizeof(payload_len)) {
close(fd);
return -1;
}
if (payload_len == 0 || payload_len > SNAPENT_MAX_DECODE_PAYLOAD) {
close(fd);
return -1;
}
uint8_t* discard = (uint8_t*)malloc(payload_len);
if (discard == NULL) {
close(fd);
return -1;
}
ssize_t got = read(fd, discard, payload_len);
free(discard);
if (got != (ssize_t)payload_len) {
close(fd);
return -1;
}
}
/* Skip the entire queue section -- still the pre-task-2.2 raw
* {name_len, name, status} layout (unchanged by this task). */
uint32_t queue_count = 0;
if (read(fd, &queue_count, sizeof(queue_count)) !=
(ssize_t)sizeof(queue_count)) {
close(fd);
return -1;
}
for (uint32_t i = 0; i < queue_count; i++) {
uint32_t name_len = 0;
if (read(fd, &name_len, sizeof(name_len)) != (ssize_t)sizeof(name_len)) {
close(fd);
return -1;
}
if (name_len > SNAPENT_MAX_QNAME_LEN) {
close(fd);
return -1;
}
char namebuf[SNAPENT_MAX_QNAME_LEN + 1];
if (name_len > 0 &&
read(fd, namebuf, name_len) != (ssize_t)name_len) {
close(fd);
return -1;
}
uint32_t status = 0;
if (read(fd, &status, sizeof(status)) != (ssize_t)sizeof(status)) {
close(fd);
return -1;
}
}
/* Host section (task 2.1's actual boundary): host_count + per-host
* [payload_len][capnp HostAdminState bytes]. */
uint32_t host_count = 0;
if (read(fd, &host_count, sizeof(host_count)) !=
(ssize_t)sizeof(host_count)) {
close(fd);
return -1;
}
int n = 0;
for (uint32_t i = 0; i < host_count; i++) {
uint32_t payload_len = 0;
if (read(fd, &payload_len, sizeof(payload_len)) !=
(ssize_t)sizeof(payload_len)) {
close(fd);
return -1;
}
if (payload_len == 0 || payload_len > SNAPENT_MAX_DECODE_PAYLOAD) {
close(fd);
return -1;
}
uint8_t* buf = (uint8_t*)malloc(payload_len);
if (buf == NULL) {
close(fd);
return -1;
}
if (read(fd, buf, payload_len) != (ssize_t)payload_len) {
free(buf);
close(fd);
return -1;
}
if (n < max_out) {
struct capn c_ctx;
capn_init_mem(&c_ctx, buf, payload_len, 0);
HostAdminState_ptr hp;
hp.p = capn_getp(capn_root(&c_ctx), 0, 1);
struct HostAdminState ch;
memset(&ch, 0, sizeof(ch));
read_HostAdminState(&ch, hp);
capnp_host_admin_state_t decoded;
memset(&decoded, 0, sizeof(decoded));
decode_HostAdminState(&decoded, &ch);
capn_free(&c_ctx);
snprintf(out[n].name, sizeof(out[n].name), "%s",
decoded.name ? decoded.name : "");
out[n].status = decoded.status;
free_HostAdminState(&decoded);
n++;
}
free(buf);
}
close(fd);
return n;
}
static snapent_decoded_host_t* snapent_find_decoded_host(
snapent_decoded_host_t* hosts, int count, const char* name)
{
for (int i = 0; i < count; i++) {
if (strcmp(hosts[i].name, name) == 0)
return &hosts[i];
}
return NULL;
}
/* ------------------------------------------------------------------ */
/* Test 6b (task 2.1): every host currently in the live workermgr table */
/* gets exactly one HostAdminState frame, whose status is EXACTLY the */
/* host's admin-issued bits (w->host.status & XCE_WORKER_ADMIN_MASK) -- */
/* never missing an admin bit, never carrying a dynamic/liveness bit */
/* (Requirements 1.6, 1.7). Covers a host with NO admin bits set too */
/* (status must decode to exactly 0, not be omitted) -- collect_host_ */
/* frames applies no retention predicate, mirroring collect_queue_frames.*/
/* ------------------------------------------------------------------ */
static void test_snapent_checkpoint_host_admin_status_roundtrip(void)
{
const int n_hosts = 4; /* covers every g_snapent_host_admin_statuses entry */
snapent_fixture_t* src = snapent_make(0, 0);
TEST_ASSERT_NOT_NULL(src);
snapent_seed_hosts(src->c, n_hosts);
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&src->c->persist));
char path[4096];
snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp",
src->c->persist.snap_path, src->c->persist.checkpoint_lsn);
snapent_decoded_host_t decoded[8];
int count = snapent_read_checkpoint_hosts(path, decoded, 8);
TEST_ASSERT_TRUE_MESSAGE(count >= 0,
"checkpoint file must parse as a well-formed entity-bearing "
"payload through the host section");
TEST_ASSERT_EQUAL_INT_MESSAGE(n_hosts, count,
"checkpoint must contain exactly one HostAdminState frame per host "
"currently in the live workermgr table -- collect_host_frames "
"applies no retention predicate, mirroring collect_queue_frames");
for (int i = 0; i < n_hosts; i++) {
char expect_name[32];
snapent_host_name(i, expect_name, sizeof(expect_name));
snapent_decoded_host_t* dh =
snapent_find_decoded_host(decoded, count, expect_name);
TEST_ASSERT_NOT_NULL_MESSAGE(dh,
"checkpoint's raw host frames must contain the seeded host");
uint32_t expect_admin = snapent_host_admin_status(i);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(expect_admin, dh->status,
"the frame's status must be EXACTLY the admin-issued bits -- "
"neither missing an admin bit nor carrying a dynamic bit "
"(Requirement 1.6, 1.7)");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(0,
dh->status & SNAPENT_HOST_DYNAMIC_BITS,
"the frame must never carry the seeded host's dynamic/liveness "
"bits, even though they are set on the live worker_t "
"(Requirement 1.7)");
}
snapent_teardown(src);
printf("test_snapent_checkpoint_host_admin_status_roundtrip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 6c (task 2.1): the task's own literal Observable criterion -- a */
/* checkpoint taken while exactly one host exists, administratively */
/* closed, contains exactly one HostAdminState frame for it, with ONLY */
/* the admin bit set (no dynamic fields). */
/* ------------------------------------------------------------------ */
static void test_snapent_checkpoint_closed_host_yields_exactly_one_admin_only_frame(void)
{
snapent_fixture_t* src = snapent_make(0, 0);
TEST_ASSERT_NOT_NULL(src);
const char* hostname = "snapent-closed-host";
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
xce_worker_t hdef;
memset(&hdef, 0, sizeof(hdef));
hdef.name = (char*)hostname;
hdef.attrs = &attrs;
hdef.status = XCE_WORKER_CLOSED_BY_ADMIN | SNAPENT_HOST_DYNAMIC_BITS;
TEST_ASSERT_NOT_NULL(makeWorker(src->c->workermgr, &hdef));
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&src->c->persist));
char path[4096];
snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp",
src->c->persist.snap_path, src->c->persist.checkpoint_lsn);
snapent_decoded_host_t decoded[4];
int count = snapent_read_checkpoint_hosts(path, decoded, 4);
TEST_ASSERT_TRUE(count >= 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, count,
"a checkpoint taken while exactly one host exists (administratively "
"closed) must contain exactly one HostAdminState frame for it");
TEST_ASSERT_EQUAL_STRING(hostname, decoded[0].name);
TEST_ASSERT_EQUAL_UINT32_MESSAGE((uint32_t)XCE_WORKER_CLOSED_BY_ADMIN,
decoded[0].status,
"the frame's status must contain ONLY the admin bit -- none of the "
"host's dynamic fields/bits");
snapent_teardown(src);
printf("test_snapent_checkpoint_closed_host_yields_exactly_one_admin_only_frame: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 6d (task 2.1): no hosts in the live table -> an empty, but */
/* well-formed, host section (not an error, not a phantom frame). */
/* ------------------------------------------------------------------ */
static void test_snapent_checkpoint_no_hosts_yields_empty_host_section(void)
{
snapent_fixture_t* src = snapent_make(0, 0);
TEST_ASSERT_NOT_NULL(src);
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&src->c->persist));
char path[4096];
snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp",
src->c->persist.snap_path, src->c->persist.checkpoint_lsn);
snapent_decoded_host_t decoded[4];
int count = snapent_read_checkpoint_hosts(path, decoded, 4);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, count,
"a checkpoint taken with no hosts in the live workermgr table must "
"have an empty host section, not an error and not a phantom frame");
snapent_teardown(src);
printf("test_snapent_checkpoint_no_hosts_yields_empty_host_section: PASS\n");
}
/* ───────────────────────────────────────────────────────────────────────
* ha-entity-snapshots Task 4.2 (Req 3.1-3.3): rotation-guard RED test.
*
@ -1307,6 +1698,9 @@ void test_persist_snapshot_entities(void)
RUN_TEST(test_snapent_checkpoint_write_failure_leaves_no_final_file);
RUN_TEST(test_snapent_checkpoint_mid_encode_failure_discards_whole_file);
RUN_TEST(test_snapent_checkpoint_queue_status_roundtrip);
RUN_TEST(test_snapent_checkpoint_host_admin_status_roundtrip);
RUN_TEST(test_snapent_checkpoint_closed_host_yields_exactly_one_admin_only_frame);
RUN_TEST(test_snapent_checkpoint_no_hosts_yields_empty_host_section);
RUN_TEST(test_snapent_rotation_preserves_in_flight_checkpoint);
RUN_TEST(test_snapent_load_snapshot_restores_entities_before_advancing_position);
RUN_TEST(test_snapent_load_snapshot_corrupted_entity_leaves_position_unchanged);