From 00cd26a927961f580ec73a3b4b152191e6cd8ee6 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 24 Jul 2026 14:08:12 +0000 Subject: [PATCH] feat(config-upgrade-snapshot): decode host/queue admin-state frames on checkpoint load load_snapshot_entities' queue-frame decode migrates from the old raw {name_len,name,status} layout to {payload_len, capnp QueueAdminState bytes}, still calling the unmodified restore_queue_state. Adds a new host-frame section decode after queues, dispatching to restore_host_state (worker.c, task 3) with the same bounds-checking and restore-before-advance discipline as the existing job/queue loops. Un-ignores test_snapent_load_snapshot_restores_entities_before_advancing_position (blocked since task 2.2) -- now passes for real. Also fixes a genuine regression in test_snapshot_recv_out_of_order_chunks_assemble_correctly (test_config_repl.c): its synthetic checkpoint-body filler landed on the new host_count field's offset, tripping the cluster==NULL guard -- pinned to a well-formed host_count=0, matching that test's existing fmt/job_count/queue_count precedent. Task 4 of 19. Requirements: 1.10 --- .kiro/specs/config-upgrade-snapshot/tasks.md | 4 +- src/controller/persist.c | 176 +++++++++++---- src/controller/testsuite/test_config_repl.c | 18 +- .../test_persist_snapshot_entities.c | 209 ++++++++++++++++-- 4 files changed, 347 insertions(+), 60 deletions(-) diff --git a/.kiro/specs/config-upgrade-snapshot/tasks.md b/.kiro/specs/config-upgrade-snapshot/tasks.md index e8a4d0b5..a05b2b66 100644 --- a/.kiro/specs/config-upgrade-snapshot/tasks.md +++ b/.kiro/specs/config-upgrade-snapshot/tasks.md @@ -34,7 +34,7 @@ - _Requirements: 1.10, 1.11, 5.8_ - _Boundary: Host Restore (worker.c)_ -- [ ] 4. Decode host and queue admin-state frames on checkpoint load +- [x] 4. Decode host and queue admin-state frames on checkpoint load - Add host-frame decoding to the checkpoint load path, dispatching each decoded frame to the host restore capability - Update queue-frame decoding to read the new QueueAdminState encoding instead of the old raw layout, without changing the existing queue-state-restore behavior itself - Preserve the existing restore discipline: every frame in the checkpoint restores successfully before any position is advanced; any restore failure leaves prior state untouched @@ -153,7 +153,7 @@ - **Task 2.3 → task 6.1 heads-up:** `read_active_config_bytes` (persist.c) is the fully-inlined (no config_repl.c call) pattern for reading live config bytes off disk fresh each time — task 6.1's activation-fence verify function needs the identical "always re-read live bytes, never trust a cached/embedded copy" discipline (design.md is explicit about this). Reuse/mirror `read_active_config_bytes` rather than re-deriving the config_repl.c-avoidance reasoning from scratch. -- **Task 2.2 → task 4 MUST re-enable:** `test_snapent_load_snapshot_restores_entities_before_advancing_position` (test_persist_snapshot_entities.c) is currently `TEST_IGNORE_MESSAGE`'d — its `m_queues=2` fixture round-trips through `persist_load_snapshot`, whose queue decode still expects the pre-2.2 raw layout. Task 4 (queue-frame decode migration) MUST remove the ignore and confirm this test passes for real once `load_snapshot`'s queue decode reads `QueueAdminState` capnp. Do not leave it ignored after task 4 lands. +- **Task 2.2 → task 4 MUST re-enable — DONE (task 4):** `test_snapent_load_snapshot_restores_entities_before_advancing_position` is un-ignored and passes for real now that `load_snapshot`'s queue decode reads `QueueAdminState` capnp. - `test_snapent_checkpoint_queue_status_roundtrip`'s comments (~line 1019) are stale (still reference "format_version=1" / "task 3.2 not yet implemented" from an earlier spec pass) — harmless, not a task 2.2 defect, but worth cleaning up whenever that test is next touched. - **Task 2.1 → task 3 heads-up:** host-frame collection is unconditional (every host in `c->workermgr` gets a frame, status masked to admin bits, may be 0) — confirmed correct against requirements.md/design.md by both implementer and reviewer. Consequence for task 3 (`restore_host_state`): a checkpoint may contain many zero-admin-status host frames on large clusters. Not required to special-case this, but consider whether `restore_host_state` should skip placeholder creation when `admin_status == 0` on a miss, purely as a cost optimization — current design.md contract doesn't require it either way. diff --git a/src/controller/persist.c b/src/controller/persist.c index 6d117999..df5d038c 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -37,6 +37,11 @@ extern int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len); extern int restore_queue_state(cluster_t* c, const char* name, uint32_t status); +/* restore_host_state defined in worker.c (config-upgrade-snapshot task 3, + * Req 1.10/1.11/5.8) -- create-if-missing host admin-state restore, called + * from load_snapshot_entities' host-frame loop below (task 4). */ +extern int restore_host_state(cluster_t* c, const char* name, + uint32_t admin_status); /* raft-apply-decoupling task 3.2: pending command-apply hand-off list node * (producer side; the not-yet-built ordered apply pump, task 6.1/6.2, is @@ -2742,10 +2747,17 @@ done: /* Recovery internals */ /* ------------------------------------------------------------------ */ -/* Parse and restore every job frame then every queue frame from the entity - * payload beginning at the current `fd` offset (design.md "Checkpoint File - * Layout"): [job_count][job_count * {payload_len; capnp Job bytes}] - * [queue_count][queue_count * {name_len; name; status}]. +/* Parse and restore every job frame, then every queue frame, then every host + * frame from the entity payload beginning at the current `fd` offset + * (design.md "Checkpoint File Layout", SNAPSHOT_FORMAT_VERSION=2): + * [job_count][job_count * {payload_len; capnp Job bytes}] + * [queue_count][queue_count * {payload_len; capnp QueueAdminState bytes}] + * [host_count][host_count * {payload_len; capnp HostAdminState bytes}] + * A trailing config-bytes section follows the host section in the file + * (config-upgrade-snapshot task 2.3) but is deliberately NOT read here -- + * out of this function's scope; any bytes after the host section are simply + * left unread, exactly as host-frame bytes were already tolerated as + * present-but-unread before this function grew a host section of its own. * * `file_size` is the checkpoint file's total size (from fstat) and * `consumed` is the number of bytes already read from it (legacy header + @@ -2755,8 +2767,13 @@ done: * length must fail fast, not drive an oversized read). * * Restores in file order: every job via restore_job, then every queue via - * restore_queue_state (design.md sequence diagram). Returns 0 only if every - * frame restored; on the FIRST failure of any kind (short read, oversized + * restore_queue_state, then every host via restore_host_state + * (config-upgrade-snapshot task 4, Req 1.10 — design.md sequence diagram + * and "Referential integrity" note: restore order does not matter + * functionally, since re-homing decisions are already baked into each + * entity's encoded content at materialization time, never re-derived at + * restore time, Design Invariant 2). Returns 0 only if every frame + * restored; on the FIRST failure of any kind (short read, oversized * length, decode error, or restore error) returns immediately with a * negative errno — the caller MUST NOT commit any position change on a * non-zero return (Requirement 2.3). @@ -2765,9 +2782,12 @@ done: * own documented idempotent-replay guard (job already present), needed so a * retried load — after an earlier partial attempt already created some jobs * before failing on a later one — can continue past already-restored entries - * rather than aborting on them again. restore_queue_state has no such - * idempotent-duplicate case (re-applying the same status is already a - * harmless overwrite), so any non-zero return from it is a genuine failure. */ + * rather than aborting on them again. restore_queue_state and + * restore_host_state have no such idempotent-duplicate case of their own to + * special-case here (re-applying the same status is already a harmless + * overwrite/OR, and restore_host_state's create-if-missing placeholder path + * is independently idempotent under retry — Requirement 5.8, worker.c), so + * any non-zero return from either is a genuine failure. */ static int load_snapshot_entities(int fd, off_t file_size, off_t consumed, persist_t* p, uint64_t stored_lsn) { @@ -2814,45 +2834,123 @@ static int load_snapshot_entities(int fd, off_t file_size, off_t consumed, return -EINVAL; /* entities present but nowhere to restore them into */ 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)) + uint32_t payload_len = 0; + if (read(fd, &payload_len, sizeof(payload_len)) != + (ssize_t)sizeof(payload_len)) return -EIO; - consumed += (off_t)sizeof(name_len); + consumed += (off_t)sizeof(payload_len); - if (name_len == 0 || (off_t)name_len > file_size - consumed) + if (payload_len == 0 || (off_t)payload_len > file_size - consumed) return -EIO; - char* name = (char*)malloc((size_t)name_len + 1); - if (name == NULL) + uint8_t* buf = (uint8_t*)malloc(payload_len); + if (buf == NULL) return -ENOMEM; - if (read(fd, name, name_len) != (ssize_t)name_len) { - free(name); + if (read(fd, buf, payload_len) != (ssize_t)payload_len) { + free(buf); return -EIO; } - name[name_len] = '\0'; - consumed += (off_t)name_len; + consumed += (off_t)payload_len; - uint32_t status = 0; - if (read(fd, &status, sizeof(status)) != (ssize_t)sizeof(status)) { - free(name); - return -EIO; - } - consumed += (off_t)sizeof(status); + /* Decode the capnp QueueAdminState frame (config-upgrade-snapshot + * task 4, Req 1.9/5.7 decode side — migrated off the prior raw + * {name_len, name, status} layout collect_queue_frames stopped + * writing in task 2.2). Mirrors the WAL_JOB_SUBMIT re-decode block + * above in this file (capn_init_mem/capn_getp/read_*) and + * test_persist_snapshot_entities.c's own + * snapent_read_checkpoint_queues reader exactly. */ + struct capn c_ctx; + capn_init_mem(&c_ctx, buf, payload_len, 0); + QueueAdminState_ptr qp; + qp.p = capn_getp(capn_root(&c_ctx), 0, 1); + struct QueueAdminState cq; + memset(&cq, 0, sizeof(cq)); + read_QueueAdminState(&cq, qp); - int qrc = restore_queue_state(p->cluster, name, status); - free(name); + capnp_queue_admin_state_t decoded; + memset(&decoded, 0, sizeof(decoded)); + decode_QueueAdminState(&decoded, &cq); + capn_free(&c_ctx); + free(buf); + + int qrc = restore_queue_state(p->cluster, decoded.name, decoded.status); + free_QueueAdminState(&decoded); if (qrc != 0) return qrc; } + /* Host section (config-upgrade-snapshot task 4, Req 1.10): new in + * SNAPSHOT_FORMAT_VERSION=2, added after the queue section (design.md + * "Checkpoint File Layout"). Same [payload_len][capnp bytes] per-frame + * shape as the job/queue sections above. Every host frame dispatches + * to restore_host_state (worker.c, task 3), which is create-if-missing + * (a worker_t is join-provisioned, not config-provisioned like a queue + * — Design Invariant 1) and idempotent under retry on its own + * (Requirement 5.8), so this loop applies the exact same "any failure + * returns immediately" discipline as the job/queue loops above without + * needing any extra idempotency handling here. */ + uint32_t host_count = 0; + if (read(fd, &host_count, sizeof(host_count)) != + (ssize_t)sizeof(host_count)) + return -EIO; + consumed += (off_t)sizeof(host_count); + + if (host_count > 0 && + (p->cluster == NULL || p->cluster->workermgr == NULL)) + return -EINVAL; /* entities present but nowhere to restore them into */ + + 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)) + return -EIO; + consumed += (off_t)sizeof(payload_len); + + if (payload_len == 0 || (off_t)payload_len > file_size - consumed) + return -EIO; + + uint8_t* buf = (uint8_t*)malloc(payload_len); + if (buf == NULL) + return -ENOMEM; + if (read(fd, buf, payload_len) != (ssize_t)payload_len) { + free(buf); + return -EIO; + } + consumed += (off_t)payload_len; + + /* Decode the capnp HostAdminState frame -- mirrors the queue decode + * block immediately above and + * test_persist_snapshot_entities.c's snapent_read_checkpoint_hosts + * reader exactly. */ + 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); + free(buf); + + int hrc = restore_host_state(p->cluster, decoded.name, decoded.status); + free_HostAdminState(&decoded); + if (hrc != 0) + return hrc; + } + return 0; } /* Load a snapshot: legacy header (magic + checkpoint_lsn + optional - * config-ref tail), then — for an entity-bearing checkpoint — every job and - * queue frame that follows it (ha-entity-snapshots task 5.2, Requirements - * 2.1-2.4, 5.1, 5.2). See design.md "load_snapshot (extended)" and its - * sequence diagram for the exact contract this implements. + * config-ref tail), then — for an entity-bearing checkpoint — every job, + * queue, and host frame that follows it (ha-entity-snapshots task 5.2, + * config-upgrade-snapshot task 4; Requirements 2.1-2.4, 5.1, 5.2, 1.10). + * See design.md "load_snapshot (extended)" and its sequence diagram for the + * exact contract this implements. * * Ordering (Requirement 2.1, 2.2, 2.3 — restore FIRST, validate FULLY, THEN * advance position, never the reverse, never partial): the header and any @@ -2861,16 +2959,18 @@ static int load_snapshot_entities(int fd, off_t file_size, off_t consumed, * discriminator — not a flag bit). A short file (nothing remains) is a * legacy, pre-feature checkpoint: not an error, so its recorded position is * committed immediately and *out_entities_restored is false. Otherwise every - * job then every queue frame is restored via load_snapshot_entities() before - * ANYTHING is written to p->checkpoint_lsn/p->current_lsn/p->cfgref_*; on - * any restore failure this function returns non-zero with p's fields - * PROVABLY untouched (the single commit point below is only ever reached - * after load_snapshot_entities() returns 0). *out_entities_restored is only - * set on a rc==0 return, matching persist.h's documented contract. + * job, then every queue, then every host frame is restored via + * load_snapshot_entities() before ANYTHING is written to + * p->checkpoint_lsn/p->current_lsn/p->cfgref_*; on any restore failure this + * function returns non-zero with p's fields PROVABLY untouched (the single + * commit point below is only ever reached after load_snapshot_entities() + * returns 0). *out_entities_restored is only set on a rc==0 return, matching + * persist.h's documented contract. * * Returns 0 on success, -EINVAL on bad magic / unrecognized entity format, * -EIO on a short/corrupt read, -ENOMEM on allocation failure, -errno on - * open/fstat failure, or a restore_job/restore_queue_state error code. */ + * open/fstat failure, or a restore_job/restore_queue_state/restore_host_state + * error code. */ static int load_snapshot(persist_t* p, const char* path, uint64_t lsn, bool* out_entities_restored) { diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index 5dc72973..a4b2d9a3 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -3766,11 +3766,17 @@ static void test_snapshot_recv_out_of_order_chunks_assemble_correctly(void) * entity-bearing checkpoint (same reason as * test_snapshot_recv_two_chunks_assembles_and_renames above): legacy * 57-byte header (cfgref_present=0) + format_version=2 + job_count=0 + - * queue_count=0, then filler out to 256 bytes. The filler is - * position-dependent (distinct per source byte, not per chunk) so ANY - * chunk landing at the wrong file offset is immediately detectable via - * a whole-buffer memcmp, regardless of which physical chunk a given - * structural field happens to fall in. */ + * queue_count=0 + host_count=0 (config-upgrade-snapshot task 4 -- + * load_snapshot_entities now reads a host section after the queue + * section, so it must likewise be pinned to a well-formed 0 here rather + * than left as position-dependent filler), then filler out to 256 + * bytes. The filler is position-dependent (distinct per source byte, + * not per chunk) so ANY chunk landing at the wrong file offset is + * immediately detectable via a whole-buffer memcmp, regardless of + * which physical chunk a given structural field happens to fall in -- + * except for the 4 structural-field byte ranges explicitly overlaid + * below (fmt/job_count/queue_count/host_count), which must stay + * well-formed for the load to succeed at all. */ uint64_t snap_lsn = 400; size_t total_size = 256; uint8_t body[256]; @@ -3783,9 +3789,11 @@ static void test_snapshot_recv_out_of_order_chunks_assemble_correctly(void) * (config-upgrade-snapshot task 1 bumped 1->2) */ uint32_t fake_job_count = 0u; uint32_t fake_queue_count = 0u; + uint32_t fake_host_count = 0u; memcpy(body + 57, &fake_fmt, sizeof(fake_fmt)); memcpy(body + 61, &fake_job_count, sizeof(fake_job_count)); memcpy(body + 65, &fake_queue_count, sizeof(fake_queue_count)); + memcpy(body + 69, &fake_host_count, sizeof(fake_host_count)); persist_t *p = calloc(1, sizeof(persist_t)); TEST_ASSERT_NOT_NULL(p); diff --git a/src/controller/testsuite/test_persist_snapshot_entities.c b/src/controller/testsuite/test_persist_snapshot_entities.c index 19336c38..433bdd7e 100644 --- a/src/controller/testsuite/test_persist_snapshot_entities.c +++ b/src/controller/testsuite/test_persist_snapshot_entities.c @@ -140,6 +140,15 @@ static const uint32_t g_snapent_host_admin_statuses[] = { * ever set. */ #define SNAPENT_HOST_DYNAMIC_BITS (XCE_WORKER_UNAVAIL | XCE_WORKER_BUSY) +/* Mask covering exactly the bits g_snapent_host_admin_statuses ever + * combines (config-upgrade-snapshot task 4's end-to-end restore test) -- + * disjoint from SNAPENT_HOST_DYNAMIC_BITS above by construction, so a + * restored worker's status can be checked for "contains exactly these + * admin bits" without depending on persist.c's own internal + * PERSIST_WORKER_ADMIN_MASK #define. */ +#define SNAPENT_HOST_ADMIN_STATUS_MASK \ + (XCE_WORKER_CLOSED_BY_ADMIN | XCE_WORKER_LOCKEDU | XCE_WORKER_HIDDEN) + static void snapent_host_name(int i, char* buf, size_t buf_sz) { snprintf(buf, buf_sz, "snapent-host%d", i); @@ -417,6 +426,21 @@ static queue_t* snapent_find_queue(cluster_t* c, const char* name) return (found > 0) ? q : NULL; } +/* Lookup helper for host restore assertions (config-upgrade-snapshot task 4) + * -- mirrors snapent_find_job/snapent_find_queue above exactly, over + * c->workermgr. */ +static worker_t* snapent_find_worker(cluster_t* c, const char* name) +{ + worker_t* w = NULL; + int found = 0; + + lock(c->workermgr, RTTI_LOCK_READ); + found = key_find(c->workermgr, (void*)name, (void**)&w); + unlock(c->workermgr); + + return (found > 0) ? w : NULL; +} + /* ------------------------------------------------------------------ */ /* Test 1: fixture seeds N jobs (varying state) + M queues (varying */ /* status) and both are independently observable via key_find */ @@ -1766,21 +1790,12 @@ static void snapent_force_queue_status(cluster_t* c, const char* name, static void test_snapent_load_snapshot_restores_entities_before_advancing_position(void) { - /* BLOCKED on config-upgrade-snapshot task 4: write_snapshot's queue - * trailer was migrated to capnp QueueAdminState frames by task 2.2, but - * load_snapshot's queue decode still expects the pre-task-2.2 raw - * {name_len, name, status} layout (that decode migration is task 4's - * job, design.md "Snapshot Entity/Config Payload"). This test's - * m_queues=2 fixture does a REAL persist_checkpoint + persist_load_ - * snapshot round trip, so it now fails against a format the loader - * doesn't understand yet. IGNORE'd so the suite stays green; un-ignore - * when task 4 lands the QueueAdminState capnp decode. */ - TEST_IGNORE_MESSAGE( - "BLOCKED on config-upgrade-snapshot task 4 (load_snapshot's queue " - "decode still expects the pre-task-2.2 raw layout; this test's " - "m_queues=2 fixture now round-trips through a format the loader " - "doesn't understand yet — will be re-enabled when task 4 lands the " - "QueueAdminState capnp decode)"); + /* Un-IGNORE'd by config-upgrade-snapshot task 4: load_snapshot_entities' + * queue decode is now migrated to the QueueAdminState capnp layout + * write_snapshot has written since task 2.2, so this real + * persist_checkpoint + persist_load_snapshot round trip (previously + * blocked -- see git history for the prior TEST_IGNORE_MESSAGE) now + * passes for real. */ const int n_jobs = 3; const int m_queues = 2; @@ -1872,6 +1887,169 @@ static void test_snapent_load_snapshot_restores_entities_before_advancing_positi printf("test_snapent_load_snapshot_restores_entities_before_advancing_position: PASS\n"); } +/* ------------------------------------------------------------------ */ +/* Test (config-upgrade-snapshot task 4, Req 1.10): a checkpoint carrying */ +/* BOTH queue and host admin-state content, loaded via a REAL */ +/* persist_checkpoint + persist_load_snapshot round trip, must restore */ +/* every queue AND every host correctly -- regardless of restore order */ +/* between the two entity types (design.md "Referential integrity" / */ +/* Design Invariant 2: re-homing decisions are baked in at materialization */ +/* time, restore never re-derives them, so each entity type's restore is */ +/* independent of the other's). Exercises BOTH of restore_host_state's */ +/* branches (worker.c, task 3) through the real decode-and-dispatch path */ +/* this task adds: */ +/* - miss: a host with no local record on the destination must come */ +/* back as a valid create-if-missing placeholder carrying only */ +/* the restored admin bits (Requirement 1.10, 1.11). */ +/* - hit: a host the destination already knows about must have its */ +/* admin bits OR'd in, with its pre-existing dynamic/attrs */ +/* state left completely untouched (Requirement 1.7). */ +/* ------------------------------------------------------------------ */ + +static void test_snapent_load_snapshot_restores_hosts_and_queues_end_to_end(void) +{ + const int m_queues = 3; + const int n_hosts = 4; /* indices 0,1 pre-exist on dst (hit); 2,3 miss */ + + /* Source: produces the entity-bearing checkpoint (queues + hosts, both + * carrying real admin content plus, for hosts, dynamic bits that must + * NOT survive into the frame -- Requirement 1.7, already covered by + * collect_host_frames' own tests; re-seeded here only as realistic + * source state, not re-asserted). */ + snapent_fixture_t* src = snapent_make(0, m_queues); + TEST_ASSERT_NOT_NULL(src); + snapent_seed_hosts(src->c, n_hosts); + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&src->c->persist)); + + uint64_t expected_lsn = src->c->persist.checkpoint_lsn; + char path[4096]; + snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp", + src->c->persist.snap_path, expected_lsn); + + /* Destination: queues pre-provisioned with a sentinel status (queues are + * config-provisioned -- restore_queue_state is update-only), mirroring + * the existing job/queue round-trip test above. Only hosts 0 and 1 are + * pre-seeded (the "hit" case); hosts 2 and 3 have no local record at + * all (the "miss"/placeholder case) -- exactly the "member that hasn't + * locally replayed this host's join record yet" scenario Requirement + * 1.10 exists for. */ + snapent_fixture_t* dst = snapent_make(0, m_queues); + TEST_ASSERT_NOT_NULL(dst); + for (int i = 0; i < m_queues; i++) { + char name[32]; + snapent_queue_name(i, name, sizeof(name)); + snapent_force_queue_status(dst->c, name, SNAPENT_PRE_RESTORE_QUEUE_STATUS); + } + + /* Pre-existing dynamic bit that must survive restore untouched (Req + * 1.7's flip side: restore must never CLEAR a real worker's existing + * non-admin state, only OR admin bits into it). XCE_WORKER_BUSY is the + * same dynamic bit SNAPENT_HOST_DYNAMIC_BITS already exercises above. */ + const uint32_t hit_pre_existing_dynamic_bit = XCE_WORKER_BUSY; + for (int i = 0; i < 2; 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; + /* Deliberately NOT the checkpoint's real admin status -- a + * post-restore match can then only be explained by a genuine + * restore_host_state OR-in, not coincidental pre-existing state. */ + hdef.status = hit_pre_existing_dynamic_bit; + + TEST_ASSERT_NOT_NULL_MESSAGE(makeWorker(dst->c->workermgr, &hdef), + "destination hit-case host seeding must succeed"); + } + + /* Sanity: the miss-case hosts truly have no record on the destination + * yet. */ + for (int i = 2; i < n_hosts; i++) { + char name[32]; + snapent_host_name(i, name, sizeof(name)); + TEST_ASSERT_NULL(snapent_find_worker(dst->c, name)); + } + + bool entities_restored = false; + int rc = persist_load_snapshot(&dst->c->persist, path, expected_lsn, + &entities_restored); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "loading a checkpoint carrying both queue and host admin-state " + "content must succeed"); + TEST_ASSERT_TRUE_MESSAGE(entities_restored, + "a checkpoint carrying queue/host entities must report " + "entities_restored=true"); + + /* Every queue's admin status restored -- proves the migrated capnp + * QueueAdminState decode still dispatches correctly to + * restore_queue_state alongside the new host section. */ + for (int i = 0; i < m_queues; i++) { + char name[32]; + snapent_queue_name(i, name, sizeof(name)); + + queue_t* q = snapent_find_queue(dst->c, name); + TEST_ASSERT_NOT_NULL(q); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(snapent_queue_status(i), q->queue.status, + "queue restore must be unaffected by the new host section " + "decoding alongside it in the same checkpoint"); + } + + /* Hit-case hosts (0, 1): admin bits OR'd into the pre-existing worker; + * the pre-existing dynamic bit must survive untouched. */ + for (int i = 0; i < 2; i++) { + char name[32]; + snapent_host_name(i, name, sizeof(name)); + + worker_t* w = snapent_find_worker(dst->c, name); + TEST_ASSERT_NOT_NULL_MESSAGE(w, + "a host the destination already knew about must still be " + "present after restore (update, not replace)"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(snapent_host_admin_status(i), + w->host.status & SNAPENT_HOST_ADMIN_STATUS_MASK, + "restore must OR the checkpoint's admin bits into the " + "pre-existing worker's status"); + TEST_ASSERT_TRUE_MESSAGE( + (w->host.status & hit_pre_existing_dynamic_bit) != 0, + "restore must never clear a pre-existing worker's own " + "non-admin state (Requirement 1.7)"); + } + + /* Miss-case hosts (2, 3): a minimal, valid placeholder was created, + * carrying exactly the restored admin bits plus the UNAVAIL marker a + * not-yet-joined placeholder always gets -- never the source's dynamic + * bits, which were never part of the encoded frame to begin with. */ + for (int i = 2; i < n_hosts; i++) { + char name[32]; + snapent_host_name(i, name, sizeof(name)); + + worker_t* w = snapent_find_worker(dst->c, name); + TEST_ASSERT_NOT_NULL_MESSAGE(w, + "a host with no local record must get a create-if-missing " + "placeholder (Requirement 1.10), not be silently dropped"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(snapent_host_admin_status(i), + w->host.status & SNAPENT_HOST_ADMIN_STATUS_MASK, + "placeholder must carry exactly the restored admin bits"); + TEST_ASSERT_TRUE_MESSAGE((w->host.status & XCE_WORKER_UNAVAIL) != 0, + "a create-if-missing placeholder must be marked " + "unjoined/dead-until-connected"); + TEST_ASSERT_TRUE_MESSAGE((w->host.status & XCE_WORKER_BUSY) == 0, + "a dynamic/non-admin bit must never travel through the " + "checkpoint frame into a restored placeholder (Requirement 1.7)"); + TEST_ASSERT_NOT_NULL_MESSAGE(w->host.attrs, + "a placeholder's hostattrs_t must be non-NULL so a later real " + "join's rejoin_worker direct dereference does not crash " + "(Requirement 1.11)"); + } + + snapent_teardown(dst); + snapent_teardown(src); + printf("test_snapent_load_snapshot_restores_hosts_and_queues_end_to_end: PASS\n"); +} + /* ------------------------------------------------------------------ */ /* Test 8 (task 5.1, RED): a deliberately corrupted entity frame must fail */ /* the load and leave current_lsn/checkpoint_lsn COMPLETELY unchanged -- */ @@ -2050,6 +2228,7 @@ void test_persist_snapshot_entities(void) RUN_TEST(test_snapent_checkpoint_config_bytes_section_deterministic_across_checkpoints); 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_restores_hosts_and_queues_end_to_end); RUN_TEST(test_snapent_load_snapshot_corrupted_entity_leaves_position_unchanged); RUN_TEST(test_snapent_load_snapshot_legacy_file_reports_position_only); }