/* * test_persist_snapshot_entities.c — entity-bearing checkpoint test suite * (ha-entity-snapshots). * * Task 1.1 (Requirements 1.1, 2.1): a reusable test fixture wrapping a real * persist_t embedded in a minimal test cluster_t whose jobmgr/qmgr tables are * seeded with N jobs (varying state) and M queues (varying status) — the * foundation later tasks in this file build on: * - 2.1/2.2: write_snapshot/load_snapshot job round-trip, per-entity * consistency, partial-write discard * - 3.1/3.2: queue admin-status round-trip * - 4.2: rotation-guard (in-flight checkpoint survives rotate_snapshots) * - 5.1/5.2: restore-before-advance, failure-safety, legacy-format handling * * Modeled on: * - test_persist_hooks.c's make_hook_test_persist/cleanup_hook_test_persist * (bare persist_t over a private tmpdir, no HA/commit-engine wiring) * - test_commit_barrier_reply.c's cbrp_setup / test_egress_filter.c's * ef_setup (persist_t embedded in a real cluster_t with jobmgr/qmgr wired * as xkvmgr instances, back-pointer registered via persist_init so * persist_start can register hooks) * * Deliberately OMITTED vs. cbrp_setup/ef_setup: commit engine, standby * registry, repl_broadcast_send_hook, and the LSN-0 priming append. None of * those are needed here — create_job/makeJob and makeQueue insert directly * into the cluster's tables with no WAL append or replication involved, and * write_snapshot/persist_checkpoint never call repl_broadcast (checkpoint * markers are local-only; see persist.c persist_write_checkpoint_marker). * Adding that machinery here would be unused scaffolding. * * job_t/queue_t insertion mirrors the established test idiom: * - ef_make_job (test_egress_filter.c): zero an xce_job_t, set * jobid/state/submit, call makeJob(jobmgr, &jdef) — makeJob deep-copies * the submit payload (jobj.c job_init -> makeJobsub), so a single reused * scratch job_submit_data_t is safe across a seeding loop. * - cbrp_arrange_queuectrl (test_commit_barrier_reply.c): zero an * xce_queue_t, set queue (name) + status, call makeQueue(qmgr, &qdef) — * queue_init deep-copies the name via makeString, and the underlying * string-keyed kvtab strdup's the key on insert (kv.c assign_string_key), * so a single reused scratch name buffer is likewise safe. * * NOTE (design.md / bug investigation 2026-06-30): queues are config- * provisioned at startup, not created from WAL/snapshot data — restore_queue_state * can only re-apply status to a queue that already exists in qmgr. Seeding * queues directly here (as makeQueue does) stands in for that config-load * step, matching what a real config-provisioned queue table looks like by * the time write_snapshot/load_snapshot ever touch it. */ #define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "cluster.h" #include "config_repl.h" #include "controller.capnp.h" #include "jobj.h" #include "myconf.h" #include "qobj.h" #include "wobj.h" #include "xce/cluster_def.h" #include "xce/types.h" #include "xstring.h" /* ------------------------------------------------------------------ */ /* Deterministic seed data — later tests predict seeded values from the */ /* loop index via these helpers rather than the fixture returning arrays */ /* ------------------------------------------------------------------ */ #define SNAPENT_JOBID_BASE (1000ULL) static const unsigned int g_snapent_job_states[] = { JOB_STAT_PEND, JOB_STAT_RUN, JOB_STAT_EXIT, JOB_STAT_DONE, JOB_STAT_PSUSP, }; #define SNAPENT_N_JOB_STATES \ (sizeof(g_snapent_job_states) / sizeof(g_snapent_job_states[0])) static const uint32_t g_snapent_queue_statuses[] = { QUEUE_STAT_OPEN, QUEUE_STAT_CLOSED, QUEUE_STAT_INACT_ADMIN, (QUEUE_STAT_CLOSED | QUEUE_STAT_INACT_ADMIN), }; #define SNAPENT_N_QUEUE_STATUSES \ (sizeof(g_snapent_queue_statuses) / sizeof(g_snapent_queue_statuses[0])) static uint64_t snapent_jobid(int i) { return SNAPENT_JOBID_BASE + (uint64_t)i; } static unsigned int snapent_job_state(int i) { return g_snapent_job_states[i % (int)SNAPENT_N_JOB_STATES]; } static void snapent_queue_name(int i, char* buf, size_t buf_sz) { snprintf(buf, buf_sz, "snapent-q%d", i); } 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) /* 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); } 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 * Job capnp schema actually carries (JobSubmitData: queue, user, ..., * JobStartInfo: cmdline, ...; src/schema/controller.capnp) are populated. */ static void snapent_job_queue_name(int i, char* buf, size_t buf_sz) { snprintf(buf, buf_sz, "snapent-jobqueue-%d", i); } static void snapent_job_user(int i, char* buf, size_t buf_sz) { snprintf(buf, buf_sz, "snapent-user-%d", i); } static void snapent_job_cmdline(int i, char* buf, size_t buf_sz) { snprintf(buf, buf_sz, "/bin/snapent-cmd-%d --arg=%d", i, i); } /* ------------------------------------------------------------------ */ /* Fixture: persist_t + cluster_t with seeded job/queue tables */ /* ------------------------------------------------------------------ */ typedef struct { cluster_t* c; char tmpdir[256]; /* Only populated/wired (f->c->myconf = &f->myconf) by tests that call * snapent_wire_myconf -- task 2.3's config-bytes-section tests. Every * other existing test leaves f->c->myconf NULL, exactly as before this * field was added (zero footprint on already-passing tests). */ myconf_t myconf; } snapent_fixture_t; /* jobj.c's jobsub/startinfo tables are process-global and idempotently * guarded — mirror the per-file static flag used by test_egress_filter.c / * test_commit_barrier_reply.c so this file can run standalone or alongside * them in the same controller_unit_test binary. */ static int g_snapent_jobobj_ready; static void snapent_seed_jobs(cluster_t* c, int n) { static job_submit_data_t submit_scratch; static job_startinfo_t startinfo_scratch; for (int i = 0; i < n; i++) { xce_job_t jdef; char queue_buf[48], user_buf[48], cmdline_buf[96]; memset(&jdef, 0, sizeof(jdef)); memset(&submit_scratch, 0, sizeof(submit_scratch)); memset(&startinfo_scratch, 0, sizeof(startinfo_scratch)); snapent_job_queue_name(i, queue_buf, sizeof(queue_buf)); snapent_job_user(i, user_buf, sizeof(user_buf)); snapent_job_cmdline(i, cmdline_buf, sizeof(cmdline_buf)); /* makeJob -> job_init -> makeJobsub deep-copies these via makeString * before this function returns, so the stack scratch buffers only * need to stay valid for the duration of the makeJob call below. */ startinfo_scratch.cmdline = cmdline_buf; submit_scratch.queue = queue_buf; submit_scratch.user = user_buf; submit_scratch.start_info = &startinfo_scratch; jdef.jobid = snapent_jobid(i); jdef.state = snapent_job_state(i); jdef.submit = &submit_scratch; /* job_should_retain (persist.c) excludes a TERMINAL job once its * finish_time + finished_retain_s has elapsed -- a never-set (0) * finish_time looks like "finished at the epoch", so a freshly * seeded terminal-state job (EXIT/DONE) would be silently excluded * from the checkpoint under the default finished_retain_s window. * Stamp a fresh finish_time so seeded terminal jobs are retained * exactly like a job that just finished -- matching what * job_should_retain actually gates on now that write_snapshot * applies it (ha-entity-snapshots task 2.2). */ if (jdef.state & (JOB_STAT_EXIT | JOB_STAT_DONE | JOB_STAT_PERR)) jdef.finish_time = time(NULL); TEST_ASSERT_NOT_NULL_MESSAGE(makeJob(c->jobmgr, &jdef), "seed job insertion must succeed"); } } static void snapent_seed_queues(cluster_t* c, int m) { for (int i = 0; i < m; i++) { char name[32]; snapent_queue_name(i, name, sizeof(name)); xce_queue_t qdef; memset(&qdef, 0, sizeof(qdef)); qdef.queue = name; /* deep-copied via makeString in queue_init */ qdef.status = snapent_queue_status(i); TEST_ASSERT_NOT_NULL_MESSAGE(makeQueue(c->qmgr, &qdef), "seed queue insertion must succeed"); } } /* 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. */ static snapent_fixture_t* snapent_make(int n_jobs, int m_queues) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 64, .delta = 16 }; 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) return NULL; f->c = calloc(1, sizeof(cluster_t)); if (f->c == NULL) { free(f); return NULL; } f->c->myhost = &f->c->myhost_info; f->c->myhost_info.name = (char*)"snapent-test"; f->c->myhost_info.key = (char*)""; 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->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) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(64, 64)); g_snapent_jobobj_ready = 1; } persist_config_t cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; /* slow background fsync — avoid races */ snprintf(f->tmpdir, sizeof(f->tmpdir), "/tmp/xce_snapent_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(f->tmpdir)); persist_set_workdir(&f->c->persist, f->tmpdir, "snapent"); TEST_ASSERT_EQUAL_INT(0, persist_init(&f->c->persist, &cfg, f->c)); TEST_ASSERT_EQUAL_INT(0, persist_start(&f->c->persist)); snapent_seed_jobs(f->c, n_jobs); snapent_seed_queues(f->c, m_queues); return f; } /* Wire a real myconf_t onto the fixture's cluster (config-upgrade-snapshot * task 2.3): f->myconf.workdir/name are set to the SAME (f->tmpdir, * "snapent") pair persist_set_workdir already used above -- exactly mirroring * production wiring, where a single controller instance's persist_set_workdir * and c->myconf->workdir/name are literally the same values (cluster.c:1408- * 1441). effective_dir = workdir/name is therefore shared between persist's * OWN data/{wal,snap} subtree and myconf's conf/ subtree with no collision * (different subdirectories under the same instance root), the same layout a * real controller process has on disk. * * Does NOT write any config bytes itself -- callers that need real bytes on * disk call conf_save_config() afterward (config_repl.h); a caller that * wires myconf but never saves anything is deliberately exercising the * "live cluster, no config pushed yet" -ENOENT path in * read_active_config_bytes (persist.c). */ static void snapent_wire_myconf(snapent_fixture_t* f) { f->myconf.workdir = f->tmpdir; f->myconf.name = (char*)"snapent"; f->myconf.conf_keep_n = 2; f->c->myconf = &f->myconf; } static void snapent_teardown(snapent_fixture_t* f) { if (f == NULL) return; persist_stop(&f->c->persist); 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->workermgr) dereference(f->c->workermgr); xce_hook_fini(&f->c->hooksys); /* free(NULL) is safe -- conffile is only heap-allocated (via * conf_save_config's asprintf) when a test actually wired+saved a real * config; every other test leaves it at its calloc'd NULL. */ free(f->myconf.conffile); free(f->c); free(f); } /* Lookup helpers — mirror ef_find_job (test_egress_filter.c). */ static job_t* snapent_find_job(cluster_t* c, uint64_t jobid) { job_t* j = NULL; int found = 0; lock(c->jobmgr, RTTI_LOCK_READ); found = key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); return (found > 0) ? j : NULL; } static queue_t* snapent_find_queue(cluster_t* c, const char* name) { queue_t* q = NULL; int found = 0; lock(c->qmgr, RTTI_LOCK_READ); found = key_find(c->qmgr, (void*)name, (void**)&q); unlock(c->qmgr); 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 */ /* ------------------------------------------------------------------ */ static void test_snapent_fixture_seeds_jobs_and_queues(void) { const int n_jobs = 5; const int m_queues = 4; snapent_fixture_t* f = snapent_make(n_jobs, m_queues); TEST_ASSERT_NOT_NULL(f); TEST_ASSERT_NOT_NULL(f->c); /* persist_init must have registered the cluster back-pointer — this is * what lets write_snapshot/load_snapshot reach c->jobmgr/c->qmgr. */ TEST_ASSERT_EQUAL_PTR(f->c, f->c->persist.cluster); for (int i = 0; i < n_jobs; i++) { job_t* j = snapent_find_job(f->c, snapent_jobid(i)); TEST_ASSERT_NOT_NULL_MESSAGE(j, "seeded job must be present in jobmgr"); TEST_ASSERT_EQUAL_UINT64(snapent_jobid(i), j->job.jobid); TEST_ASSERT_EQUAL_UINT32(snapent_job_state(i), j->job.state); } for (int i = 0; i < m_queues; i++) { char name[32]; snapent_queue_name(i, name, sizeof(name)); queue_t* q = snapent_find_queue(f->c, name); TEST_ASSERT_NOT_NULL_MESSAGE(q, "seeded queue must be present in qmgr"); TEST_ASSERT_EQUAL_UINT32(snapent_queue_status(i), q->queue.status); } /* Sanity: lookups for entities that were never seeded must miss. */ TEST_ASSERT_NULL(snapent_find_job(f->c, snapent_jobid(n_jobs) + 999)); TEST_ASSERT_NULL(snapent_find_queue(f->c, "snapent-does-not-exist")); snapent_teardown(f); printf("test_snapent_fixture_seeds_jobs_and_queues: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test-local raw checkpoint reader (task 2.2 remediation). */ /* */ /* This task's boundary is write_snapshot ONLY -- load_snapshot's entity */ /* restore is task 5.x's job (see design.md Requirement 2). The tests */ /* below must prove write_snapshot's on-disk output is field-identical */ /* WITHOUT routing through persist_load_snapshot/load_snapshot, so they */ /* parse the checkpoint's raw bytes directly here and decode each job */ /* frame with the SAME capnp helpers restore_job (job.c) uses */ /* (read_Job + decode_Job from controller.capnp.h) -- proving the */ /* WRITE side without any dependency on the READ side's future behavior. */ /* */ /* Byte layout mirrors design.md "Checkpoint File Layout (Physical)": */ /* [57-byte legacy sentinel: magic(8)+lsn(8)+cfgref tail(1+8+32)] */ /* [uint32_t format_version][uint32_t job_count] */ /* [job_count * { uint32_t payload_len; capnp Job bytes }] */ /* write_snapshot (persist.c) always emits the full 57-byte sentinel */ /* unconditionally (memset + full memcpy, no short-write branch for an */ /* absent cfgref), so skipping a fixed 57 bytes here is safe -- unlike */ /* load_snapshot's OWN reader, which must also tolerate a *shorter* */ /* 16-byte legacy sentinel from files predating the cfgref tail. */ /* ------------------------------------------------------------------ */ #define SNAPENT_LEGACY_HEADER_BYTES (16 + 1 + 8 + 32) #define SNAPENT_TEST_FORMAT_VERSION 2u /* must match persist.c's SNAPSHOT_FORMAT_VERSION * (config-upgrade-snapshot task 1 bumped 1->2) */ #define SNAPENT_MAX_DECODE_PAYLOAD (16u * 1024 * 1024) /* Decodes every job frame in the checkpoint at `path` into `out` * (caller-allocated array of `max_out` xce_job_t slots). Returns the * number of frames decoded (<= job_count on disk, capped at max_out), or * -1 on any structural read error (bad magic, bad/absent format_version, * short read, oversized payload_len). Each decoded slot in `out` up to * the returned count must be released by the caller via free_Job(). */ static int snapent_read_checkpoint_jobs(const char* path, xce_job_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; } uint32_t job_count = 0; if (read(fd, &job_count, sizeof(job_count)) != (ssize_t)sizeof(job_count)) { close(fd); return -1; } int n = 0; 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* 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); Job_ptr jp; jp.p = capn_getp(capn_root(&c_ctx), 0, 1); struct Job cj; memset(&out[n], 0, sizeof(out[n])); read_Job(&cj, jp); decode_Job(&out[n], &cj); capn_free(&c_ctx); n++; } free(buf); } close(fd); return n; } static xce_job_t* snapent_find_decoded_job(xce_job_t* jobs, int count, uint64_t jobid) { for (int i = 0; i < count; i++) { if (jobs[i].jobid == jobid) return &jobs[i]; } return NULL; } static void snapent_free_decoded_jobs(xce_job_t* jobs, int count) { for (int i = 0; i < count; i++) free_Job(&jobs[i]); } /* ------------------------------------------------------------------ */ /* Test 2 (task 2.1/2.2): checkpoint job content round-trips through */ /* write_snapshot's own on-disk bytes field-identical (Requirements 1.1, */ /* 1.2, 1.4). Parses the raw checkpoint file directly (see reader above)*/ /* rather than via persist_load_snapshot -- load_snapshot's entity */ /* restore is task 5.x's boundary, not this task's. */ /* ------------------------------------------------------------------ */ static void test_snapent_checkpoint_job_content_roundtrip(void) { const int n_jobs = 4; snapent_fixture_t* src = snapent_make(n_jobs, 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); xce_job_t decoded[8]; int count = snapent_read_checkpoint_jobs(path, decoded, 8); TEST_ASSERT_TRUE_MESSAGE(count >= 0, "checkpoint file must parse as a well-formed entity-bearing " "format_version=1 payload"); TEST_ASSERT_EQUAL_INT_MESSAGE(n_jobs, count, "checkpoint must contain exactly the retained seeded jobs"); for (int i = 0; i < n_jobs; i++) { char expect_queue[48], expect_user[48], expect_cmdline[96]; snapent_job_queue_name(i, expect_queue, sizeof(expect_queue)); snapent_job_user(i, expect_user, sizeof(expect_user)); snapent_job_cmdline(i, expect_cmdline, sizeof(expect_cmdline)); xce_job_t* dj = snapent_find_decoded_job(decoded, count, snapent_jobid(i)); TEST_ASSERT_NOT_NULL_MESSAGE(dj, "checkpoint's raw job frames must contain the seeded job"); TEST_ASSERT_EQUAL_UINT64(snapent_jobid(i), dj->jobid); TEST_ASSERT_EQUAL_UINT32(snapent_job_state(i), dj->state); TEST_ASSERT_NOT_NULL(dj->submit); TEST_ASSERT_EQUAL_STRING(expect_queue, dj->submit->queue); TEST_ASSERT_EQUAL_STRING(expect_user, dj->submit->user); TEST_ASSERT_NOT_NULL(dj->submit->start_info); TEST_ASSERT_EQUAL_STRING(expect_cmdline, dj->submit->start_info->cmdline); } snapent_free_decoded_jobs(decoded, count); snapent_teardown(src); printf("test_snapent_checkpoint_job_content_roundtrip: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 3 (task 2.1, RED): concurrent per-entity mutation during the */ /* checkpoint walk must never be captured as a torn mix of fields */ /* (Requirement 1.2). */ /* */ /* design.md "Existing Architecture Analysis": the table RTTI_LOCK_READ */ /* held by collect_jobs alone does NOT exclude a concurrent per-entity */ /* writer -- entity mutation takes a SEPARATE per-entity lock (mirrored */ /* here via job.c's own on_job_start / mark_job_start shape: state and */ /* start_time flip together under lock(j, RTTI_LOCK_WRITE)). Proving */ /* "never torn" requires a genuinely concurrent mutation, not a */ /* sequential one dressed up as concurrent. */ /* ------------------------------------------------------------------ */ #define SNAPENT_TORN_PRE_STATE JOB_STAT_PEND #define SNAPENT_TORN_POST_STATE JOB_STAT_RUN #define SNAPENT_TORN_POST_START_TIME ((time_t)987654321) typedef struct { cluster_t* c; uint64_t jobid; atomic_int done; } snapent_mutator_arg_t; /* Runs on a background thread concurrently with the main thread's * * checkpoint-and-reload loop below. Flips BOTH state and start_time * * together under the job's OWN per-entity write lock -- the same pairing * * mark_job_start (job.c) updates atomically on a real START transition -- * * so a "torn" restore would show up as one field from the pre-transition * * tuple mixed with one field from the post-transition tuple. */ static void* snapent_mutate_job_thread(void* arg) { snapent_mutator_arg_t* a = (snapent_mutator_arg_t*)arg; job_t* j = snapent_find_job(a->c, a->jobid); if (j != NULL) { /* Give the main thread's loop below a real chance to observe both * the pre- and post-transition tuple across its iterations. */ usleep(2000); lock(j, RTTI_LOCK_WRITE); j->job.state = SNAPENT_TORN_POST_STATE; j->job.start_time = SNAPENT_TORN_POST_START_TIME; unlock(j); } atomic_store(&a->done, 1); return NULL; } static void test_snapent_checkpoint_concurrent_mutation_not_torn(void) { const uint64_t jobid = snapent_jobid(0); snapent_fixture_t* src = snapent_make(1, 0); TEST_ASSERT_NOT_NULL(src); job_t* j0 = snapent_find_job(src->c, jobid); TEST_ASSERT_NOT_NULL(j0); TEST_ASSERT_EQUAL_UINT32((uint32_t)SNAPENT_TORN_PRE_STATE, j0->job.state); TEST_ASSERT_EQUAL_INT(0, (int)j0->job.start_time); snapent_mutator_arg_t marg; marg.c = src->c; marg.jobid = jobid; atomic_init(&marg.done, 0); pthread_t th; TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, snapent_mutate_job_thread, &marg)); bool found_any = false; bool found_torn = false; for (int iter = 0; iter < 100 && !atomic_load(&marg.done); iter++) { 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); /* Parse the raw checkpoint bytes directly (see reader above) -- * this proves write_snapshot's own encode of the job is never torn, * independent of load_snapshot's future restore behavior (task * 5.x's boundary, not this task's). */ xce_job_t decoded[4]; int count = snapent_read_checkpoint_jobs(path, decoded, 4); TEST_ASSERT_TRUE_MESSAGE(count >= 0, "checkpoint file must parse as a well-formed entity-bearing " "format_version=1 payload"); xce_job_t* dj = snapent_find_decoded_job(decoded, count, jobid); if (dj != NULL) { found_any = true; bool is_pre = (dj->state == (uint32_t)SNAPENT_TORN_PRE_STATE && dj->start_time == 0); bool is_post = (dj->state == (uint32_t)SNAPENT_TORN_POST_STATE && dj->start_time == SNAPENT_TORN_POST_START_TIME); if (!is_pre && !is_post) found_torn = true; } snapent_free_decoded_jobs(decoded, count); } pthread_join(th, NULL); TEST_ASSERT_FALSE_MESSAGE(found_torn, "checkpoint captured a torn mix of job state/start_time -- neither " "fully the pre-transition nor fully the post-transition tuple"); TEST_ASSERT_TRUE_MESSAGE(found_any, "a checkpoint must contain the seeded job's entity state at least " "once across the loop"); snapent_teardown(src); printf("test_snapent_checkpoint_concurrent_mutation_not_torn: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 4 (task 2.1): a real write-path I/O failure must leave NO file */ /* at the final checkpoint path (Requirement 1.3). */ /* */ /* NOTE ON RED-NESS: unlike the two tests above, this one is EXPECTED */ /* TO PASS already today. write_snapshot's existing open-failure and */ /* tmp-then-rename-failure paths already unlink/never-create the tmp */ /* file and never touch the final path on error (persist.c ~1487-1518). * * design.md's "Existing Architecture Analysis" explicitly notes this */ /* discard behavior ALREADY exists and is what lets the unchanged */ /* persist_checkpoint rc-gate (`if (rc != 0) goto done;`) satisfy */ /* Requirement 3 once entities are added -- so this test locks in that */ /* existing behavior as a regression guard BEFORE task 2.2 adds a */ /* multi-frame entity-encoding loop that could accidentally leave a */ /* partial file behind on a mid-loop failure. */ /* */ /* A real, portable fault is used (the snapshot directory is removed, so */ /* open(tmp_path, O_CREAT|...) fails with ENOENT) -- not a permission-bit */ /* trick, which would silently no-op under a root test runner. */ /* ------------------------------------------------------------------ */ static void test_snapent_checkpoint_write_failure_leaves_no_final_file(void) { const int n_jobs = 2; snapent_fixture_t* f = snapent_make(n_jobs, 0); TEST_ASSERT_NOT_NULL(f); char snap_path[4096]; snprintf(snap_path, sizeof(snap_path), "%s", f->c->persist.snap_path); /* Force a real I/O failure inside write_snapshot: remove the directory * tmp_path/final_path live under, so open(tmp_path, O_CREAT|...) fails * with ENOENT. */ TEST_ASSERT_EQUAL_INT(0, rmdir(snap_path)); int rc = persist_checkpoint(&f->c->persist); TEST_ASSERT_TRUE(rc != 0); /* Recreate the directory so teardown doesn't trip over a missing path. */ TEST_ASSERT_EQUAL_INT(0, mkdir(snap_path, 0700)); char final_path[4096], tmp_path[4096]; snprintf(final_path, sizeof(final_path), "%s/snap-%016" PRIx64 ".capnp", snap_path, f->c->persist.checkpoint_lsn); snprintf(tmp_path, sizeof(tmp_path), "%s/snap-%016" PRIx64 ".tmp", snap_path, f->c->persist.checkpoint_lsn); struct stat st; int final_rc = stat(final_path, &st); TEST_ASSERT_TRUE(final_rc != 0); TEST_ASSERT_EQUAL_INT(ENOENT, errno); int tmp_rc = stat(tmp_path, &st); TEST_ASSERT_TRUE(tmp_rc != 0); snapent_teardown(f); printf("test_snapent_checkpoint_write_failure_leaves_no_final_file: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 5 (task 2.2, GREEN): a genuine mid-encode-loop failure discards */ /* the WHOLE checkpoint, not just the one job whose encode failed */ /* (Requirement 1.3). */ /* */ /* Unlike test 4 above (which only exercises the pre-existing open() */ /* failure branch), this drives the NEW per-job encode loop itself via */ /* the persist_test_set_force_job_encode_fail_at seam -- forcing the */ /* (0-based) Nth retained job's encode to fail partway through a */ /* multi-job walk, proving write_snapshot doesn't leave a partial file */ /* (with only some of the jobs framed) behind on that failure. */ /* ------------------------------------------------------------------ */ static void test_snapent_checkpoint_mid_encode_failure_discards_whole_file(void) { const int n_jobs = 5; const int fail_at = 2; /* fail a job in the middle of the walk, not the first */ snapent_fixture_t* f = snapent_make(n_jobs, 0); TEST_ASSERT_NOT_NULL(f); persist_test_set_force_job_encode_fail_at(&f->c->persist, fail_at); int rc = persist_checkpoint(&f->c->persist); TEST_ASSERT_TRUE_MESSAGE(rc != 0, "a forced mid-loop job-encode failure must fail the checkpoint"); char final_path[4096], tmp_path[4096]; snprintf(final_path, sizeof(final_path), "%s/snap-%016" PRIx64 ".capnp", f->c->persist.snap_path, f->c->persist.checkpoint_lsn); snprintf(tmp_path, sizeof(tmp_path), "%s/snap-%016" PRIx64 ".tmp", f->c->persist.snap_path, f->c->persist.checkpoint_lsn); struct stat st; TEST_ASSERT_TRUE_MESSAGE(stat(final_path, &st) != 0, "a mid-encode failure must leave no file at the final checkpoint path"); TEST_ASSERT_TRUE_MESSAGE(stat(tmp_path, &st) != 0, "a mid-encode failure must not leave the tmp file behind either"); /* Disarm before teardown so no later checkpoint in this process forced * by teardown/other code paths trips over a stale forced failure. */ persist_test_set_force_job_encode_fail_at(&f->c->persist, -1); snapent_teardown(f); printf("test_snapent_checkpoint_mid_encode_failure_discards_whole_file: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test-local raw checkpoint reader for the queue section */ /* (config-upgrade-snapshot task 2.2). */ /* */ /* This task's boundary is write_snapshot ONLY -- load_snapshot's entity */ /* restore is task 4's job (design.md Requirement 5.7), mirroring the */ /* host-section reader below (task 2.1's own remediation note). Reads */ /* the checkpoint's on-disk bytes directly per design.md's "Logical Data */ /* Model" checkpoint file layout, SNAPSHOT_FORMAT_VERSION=2: */ /* [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 * {payload_len; capnp */ /* QueueAdminState bytes}] */ /* Migrated from the prior raw {name_len, name, status} layout (Req */ /* 1.9, 5.7) -- same [payload_len][capnp bytes] shape the job/host */ /* sections already use. Skips past the job section (whatever it */ /* contains) to reach the queue section, so this reader works */ /* regardless of how many jobs a fixture also seeded. */ /* ------------------------------------------------------------------ */ #define SNAPENT_MAX_QNAME_LEN 63 typedef struct { char name[SNAPENT_MAX_QNAME_LEN + 1]; uint32_t status; } snapent_decoded_queue_t; /* Reads every queue frame in the checkpoint at `path` into `out` * (caller-allocated array of `max_out` slots). Returns the number of * frames decoded (<= queue_count on disk, capped at max_out), or -1 on * any structural read error (bad magic, bad/absent format_version, short * read, oversized job payload_len, oversized queue name_len). */ static int snapent_read_checkpoint_queues(const char* path, snapent_decoded_queue_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 -- this reader only cares about what * follows it, but the on-disk layout is sequential so every job frame * must be walked past (by its own length prefix) to reach the queue * 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; } } /* Queue section (this task's actual boundary): queue_count + per-queue * [payload_len][capnp QueueAdminState bytes] -- migrated from the * prior raw {name_len, name, status} layout. Decoded via the same * generated read_QueueAdminState/decode_QueueAdminState helpers * load_snapshot's future restore step (task 4) will use, mirroring * how the host-section reader below decodes HostAdminState. */ uint32_t queue_count = 0; if (read(fd, &queue_count, sizeof(queue_count)) != (ssize_t)sizeof(queue_count)) { close(fd); return -1; } int n = 0; for (uint32_t i = 0; i < queue_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); 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); capnp_queue_admin_state_t decoded; memset(&decoded, 0, sizeof(decoded)); decode_QueueAdminState(&decoded, &cq); capn_free(&c_ctx); snprintf(out[n].name, sizeof(out[n].name), "%s", decoded.name ? decoded.name : ""); out[n].status = decoded.status; free_QueueAdminState(&decoded); n++; } free(buf); } close(fd); return n; } static snapent_decoded_queue_t* snapent_find_decoded_queue( snapent_decoded_queue_t* queues, int count, const char* name) { for (int i = 0; i < count; i++) { if (strcmp(queues[i].name, name) == 0) return &queues[i]; } return NULL; } /* ------------------------------------------------------------------ */ /* Test 6 (task 3.1, RED): a queue's (name, status) pair written into */ /* the checkpoint by write_snapshot is read back byte-identical */ /* (Requirement 1.1). */ /* */ /* write_snapshot currently hardcodes queue_count=0 (persist.c's task */ /* 3.2 placeholder -- see design.md "write_snapshot (extended)" and the */ /* Implementation Notes at the bottom of tasks.md) -- so this MUST fail */ /* today: the checkpoint's queue section is always empty regardless of */ /* how many queues are actually seeded in the fixture's qmgr. Task 3.2 */ /* makes this pass by collecting the live queue table into that section.*/ /* ------------------------------------------------------------------ */ static void test_snapent_checkpoint_queue_status_roundtrip(void) { const int m_queues = 4; snapent_fixture_t* src = snapent_make(0, m_queues); 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_queue_t decoded[8]; int count = snapent_read_checkpoint_queues(path, decoded, 8); TEST_ASSERT_TRUE_MESSAGE(count >= 0, "checkpoint file must parse as a well-formed entity-bearing " "format_version=1 payload through the queue section"); TEST_ASSERT_EQUAL_INT_MESSAGE(m_queues, count, "checkpoint must contain exactly the seeded queues -- write_snapshot " "currently hardcodes queue_count=0 (task 3.2 not yet implemented)"); for (int i = 0; i < m_queues; i++) { char expect_name[32]; snapent_queue_name(i, expect_name, sizeof(expect_name)); snapent_decoded_queue_t* dq = snapent_find_decoded_queue(decoded, count, expect_name); TEST_ASSERT_NOT_NULL_MESSAGE(dq, "checkpoint's raw queue frames must contain the seeded queue"); TEST_ASSERT_EQUAL_UINT32(snapent_queue_status(i), dq->status); } snapent_teardown(src); 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 -- migrated to capnp QueueAdminState * frames (task 2.2, Req 1.9/5.7); skip shape now matches the job * section's [payload_len][capnp bytes] framing exactly, no longer the * old raw {name_len, name, status} layout. */ 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 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; } } /* 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; } /* Reads the trailing config-bytes section a checkpoint file carries * (config-upgrade-snapshot task 2.3, Req 2.1, 2.3, 2.4, 5.1): skips past the * legacy header + format_version + job/queue/host sections exactly like * snapent_read_checkpoint_hosts above, then reads the final * [config_bytes_len][raw config bytes] blob -- the single flat section that * comes after every other section, never a [count][frame...] shape. * * On success returns 0 with *out_bytes (caller must free(), NULL when * out_len is 0) / *out_len set. Returns -1 on any parse/read failure * (malformed header, truncated section, etc.) -- mirrors every other * snapent_read_checkpoint_* reader's -1-on-failure convention in this file. */ static int snapent_read_checkpoint_config_bytes(const char* path, uint8_t** out_bytes, size_t* out_len) { 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 job, queue, and host sections -- all three share the same * [count][payload_len][bytes]* shape by this point (task 2.2 migrated * queue frames to capnp, matching job/host). */ for (int section = 0; section < 3; section++) { uint32_t count = 0; if (read(fd, &count, sizeof(count)) != (ssize_t)sizeof(count)) { close(fd); return -1; } for (uint32_t i = 0; i < 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; } } } /* Config-bytes section: config_bytes_len + raw bytes (no per-item * framing -- exactly one blob for the whole file). */ uint32_t cfg_len = 0; if (read(fd, &cfg_len, sizeof(cfg_len)) != (ssize_t)sizeof(cfg_len)) { close(fd); return -1; } if (cfg_len > SNAPENT_MAX_DECODE_PAYLOAD) { close(fd); return -1; } uint8_t* buf = NULL; if (cfg_len > 0) { buf = (uint8_t*)malloc(cfg_len); if (buf == NULL) { close(fd); return -1; } if (read(fd, buf, cfg_len) != (ssize_t)cfg_len) { free(buf); close(fd); return -1; } } close(fd); *out_bytes = buf; *out_len = (size_t)cfg_len; return 0; } 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"); } /* ─────────────────────────────────────────────────────────────────────── * config-upgrade-snapshot Task 2.3 (Req 2.1, 2.3, 2.4, 5.1): config-bytes * section content, absence, and determinism. * ─────────────────────────────────────────────────────────────────────── */ /* A config file test payload -- content is irrelevant to write_snapshot * (it reads raw bytes, never parses/interprets them), so any distinctive * byte string proves the round-trip without needing a real ClusterConfig * capnp blob. */ static const char SNAPENT_CONFIG_TEST_PAYLOAD[] = "snapent-config-upgrade-snapshot-task-2.3-test-payload"; /* Saves SNAPENT_CONFIG_TEST_PAYLOAD as f's currently-active config via the * real conf_save_config path (config_repl.c) -- the exact mechanism * read_active_config_bytes (persist.c) resolves against (conf/current -> * conf/cluster.bin.). Requires snapent_wire_myconf to have been called * on f first. */ static void snapent_save_test_config(snapent_fixture_t* f) { uint8_t sha[32]; SHA256((const unsigned char*)SNAPENT_CONFIG_TEST_PAYLOAD, sizeof(SNAPENT_CONFIG_TEST_PAYLOAD) - 1, sha); TEST_ASSERT_EQUAL_INT(0, conf_save_config(&f->myconf, (const uint8_t*)SNAPENT_CONFIG_TEST_PAYLOAD, sizeof(SNAPENT_CONFIG_TEST_PAYLOAD) - 1, sha)); } /* Task 2.3's own literal Observable: a checkpoint's config-bytes section * content hash matches the currently active configuration -- checked here * via an exact byte-for-byte memcmp against the source file's own bytes, * which subsumes a hash match (identical bytes trivially hash identically, * and is a strictly stronger, more direct assertion). */ static void test_snapent_checkpoint_config_bytes_section_matches_source_file(void) { snapent_fixture_t* f = snapent_make(0, 0); TEST_ASSERT_NOT_NULL(f); snapent_wire_myconf(f); snapent_save_test_config(f); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&f->c->persist)); char path[4096]; snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp", f->c->persist.snap_path, f->c->persist.checkpoint_lsn); uint8_t* section_bytes = NULL; size_t section_len = 0; TEST_ASSERT_EQUAL_INT_MESSAGE(0, snapent_read_checkpoint_config_bytes(path, §ion_bytes, §ion_len), "checkpoint file must parse as a well-formed entity-bearing payload " "through the trailing config-bytes section"); TEST_ASSERT_EQUAL_size_t_MESSAGE(sizeof(SNAPENT_CONFIG_TEST_PAYLOAD) - 1, section_len, "config-bytes section length must equal the source config file's " "exact byte length"); TEST_ASSERT_EQUAL_MEMORY_MESSAGE(SNAPENT_CONFIG_TEST_PAYLOAD, section_bytes, section_len, "config-bytes section content must be byte-for-byte identical to " "the currently active configuration on disk (Requirement 2.1) -- " "an exact match trivially implies the content hash also matches"); free(section_bytes); snapent_teardown(f); printf("test_snapent_checkpoint_config_bytes_section_matches_source_file: PASS\n"); } /* Requirement 2.4's "no config yet" edge: a live cluster (real myconf wired, * unlike every other test in this file which leaves f->c->myconf NULL) that * has never received a config push must still produce a valid checkpoint -- * a zero-length config-bytes section, not a failed checkpoint. Mirrors * test_snapent_checkpoint_no_hosts_yields_empty_host_section's "absence is * not failure" shape, one level up (persist.c's read_active_config_bytes * -ENOENT path). */ static void test_snapent_checkpoint_config_bytes_section_empty_when_no_local_config(void) { snapent_fixture_t* f = snapent_make(0, 0); TEST_ASSERT_NOT_NULL(f); snapent_wire_myconf(f); /* real myconf, but conf_save_config never called */ TEST_ASSERT_EQUAL_INT_MESSAGE(0, persist_checkpoint(&f->c->persist), "a checkpoint on a live cluster that has not yet received its first " "config push must still succeed -- absence of local config bytes is " "a benign early-boot state, never a checkpoint failure"); char path[4096]; snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp", f->c->persist.snap_path, f->c->persist.checkpoint_lsn); uint8_t* section_bytes = NULL; size_t section_len = 0; TEST_ASSERT_EQUAL_INT(0, snapent_read_checkpoint_config_bytes(path, §ion_bytes, §ion_len)); TEST_ASSERT_EQUAL_size_t_MESSAGE(0, section_len, "no local config bytes exist yet -- the section must be explicitly " "zero-length, not omitted or a phantom non-empty blob"); TEST_ASSERT_NULL(section_bytes); snapent_teardown(f); printf("test_snapent_checkpoint_config_bytes_section_empty_when_no_local_config: PASS\n"); } /* Requirement 5.1 determinism: two independent checkpoints taken with the * SAME active configuration unchanged on disk must produce byte-identical * config-bytes sections -- a pure function of current on-disk content, no * timestamp/ordering dependence. * * NOTE on why this does NOT also assert the two checkpoints land at * different LSNs/paths: this fixture is a bare persist_t with no live commit * engine (snapent_make's own doc comment -- "Deliberately OMITTED ... * commit engine"), so appending via persist_wal_async never advances * committed_lsn, and checkpoint_lsn (persist_checkpoint's * last_applied_lsn snapshot) never moves past its initial value here -- * both checkpoints legitimately land at the SAME path. That is fine for * what THIS test checks: write_snapshot's config-bytes-writing logic still * runs twice, as two genuinely separate persist_checkpoint calls (the * second's open/write/rename fully redoes the first's, it does not skip or * cache anything), which is exactly the "independent materialization" * property Req 5.1 requires -- proving it does not depend on the path * differing. */ static void test_snapent_checkpoint_config_bytes_section_deterministic_across_checkpoints(void) { snapent_fixture_t* f = snapent_make(0, 0); TEST_ASSERT_NOT_NULL(f); snapent_wire_myconf(f); snapent_save_test_config(f); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&f->c->persist)); char path1[4096]; snprintf(path1, sizeof(path1), "%s/snap-%016" PRIx64 ".capnp", f->c->persist.snap_path, f->c->persist.checkpoint_lsn); uint8_t* bytes1 = NULL; size_t len1 = 0; TEST_ASSERT_EQUAL_INT(0, snapent_read_checkpoint_config_bytes(path1, &bytes1, &len1)); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&f->c->persist)); char path2[4096]; snprintf(path2, sizeof(path2), "%s/snap-%016" PRIx64 ".capnp", f->c->persist.snap_path, f->c->persist.checkpoint_lsn); uint8_t* bytes2 = NULL; size_t len2 = 0; TEST_ASSERT_EQUAL_INT(0, snapent_read_checkpoint_config_bytes(path2, &bytes2, &len2)); TEST_ASSERT_EQUAL_size_t_MESSAGE(len1, len2, "independent materializations at the same unchanged configuration " "must yield equal-length config-bytes sections"); TEST_ASSERT_EQUAL_MEMORY_MESSAGE(bytes1, bytes2, len1, "independent materializations at the same unchanged configuration " "must yield byte-identical config-bytes sections (Requirement 5.1)"); TEST_ASSERT_EQUAL_size_t_MESSAGE(sizeof(SNAPENT_CONFIG_TEST_PAYLOAD) - 1, len1, "sanity: both materializations must actually carry the real config " "content, not two empty sections trivially matching each other"); free(bytes1); free(bytes2); snapent_teardown(f); printf("test_snapent_checkpoint_config_bytes_section_deterministic_across_checkpoints: PASS\n"); } /* ─────────────────────────────────────────────────────────────────────── * ha-entity-snapshots Task 4.2 (Req 3.1-3.3): rotation-guard RED test. * * A checkpoint file whose position matches an in-flight peer transfer * (task 4.1's snapshot_inflight_lsn[]) must survive rotate_snapshots even * when it would otherwise fall outside snap_keep_n. RED against today's * rotate_snapshots, which has no awareness of in-flight transfers at all. * ─────────────────────────────────────────────────────────────────────── */ static void test_snapent_rotation_preserves_in_flight_checkpoint(void) { snapent_fixture_t* f = snapent_make(0, 0); TEST_ASSERT_NOT_NULL(f); /* Force aggressive rotation so the protected file would otherwise be * deleted after just one more checkpoint cycle. */ f->c->persist.config.snap_keep_n = 1; /* Round 1: produce the checkpoint file we'll mark as in-flight. */ TEST_ASSERT_EQUAL_INT(0, persist_wal_async(&f->c->persist, WAL_JOB_UPDATE, "a", 1)); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&f->c->persist)); uint64_t protected_lsn = f->c->persist.checkpoint_lsn; char protected_path[4096]; snprintf(protected_path, sizeof(protected_path), "%s/snap-%016" PRIx64 ".capnp", f->c->persist.snap_path, protected_lsn); TEST_ASSERT_EQUAL_INT_MESSAGE(0, access(protected_path, F_OK), "the checkpoint file must exist right after it is written"); /* Mark this checkpoint's position as referenced by an in-flight peer * transfer -- the same field/lifecycle task 4.1 wired in ha.c. */ atomic_store_explicit(&f->c->ha.snapshot_inflight_lsn[0], protected_lsn, memory_order_release); /* Rounds 2-3: with snap_keep_n=1, each further checkpoint would normally * rotate out every older file, including the one marked in-flight. */ for (int i = 0; i < 2; i++) { TEST_ASSERT_EQUAL_INT(0, persist_wal_async(&f->c->persist, WAL_JOB_UPDATE, "b", 1)); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&f->c->persist)); } TEST_ASSERT_EQUAL_INT_MESSAGE(0, access(protected_path, F_OK), "Req 3.1-3.3: a checkpoint file referenced by an in-flight peer " "transfer must survive rotate_snapshots even when snap_keep_n " "would otherwise have deleted it -- rotate_snapshots does not yet " "consult snapshot_inflight_lsn[] (task 4.3 not yet implemented)"); atomic_store_explicit(&f->c->ha.snapshot_inflight_lsn[0], 0, memory_order_release); snapent_teardown(f); printf("test_snapent_rotation_preserves_in_flight_checkpoint: PASS\n"); } /* ─────────────────────────────────────────────────────────────────────── * ha-entity-snapshots Task 5.1 (Req 2.1, 2.2, 2.3, 2.4, 5.1, 5.2): * restore-before-advance, failure-safety, and legacy-format RED tests. * * Unlike every test above (which deliberately parses write_snapshot's raw * bytes to stay clear of load_snapshot's not-yet-implemented restore * behavior — task 2.2 remediation note), these three tests call * persist_load_snapshot DIRECTLY: it is the exact function under test for * this task (design.md "load_snapshot (extended)"). * * They are written against persist_load_snapshot's NEW 4-argument * signature (with the out_entities_restored parameter) even though * load_snapshot's internal restore behavior is still today's * watermark-only implementation — see persist.c's persist_load_snapshot * wrapper comment for exactly why that stopgap wrapper makes all three of * these genuinely fail (RED) for the right reason, not a compile error: * 1. entities are never actually restored into jobmgr/qmgr; * 2. a corrupted entity frame is never inspected, so failure is never * detected and the position is wrongly advanced anyway; * 3. a legacy (position-only) file is falsely reported as fully * restored (entities_restored=true) instead of false — the exact * Requirement 2.4 hazard this feature exists to close. * ─────────────────────────────────────────────────────────────────────── */ /* A status value outside every value in g_snapent_queue_statuses[] (which * only ever sets bits 0x00/0x01/0x04) -- so a post-restore match against the * checkpoint's real status can only be explained by a genuine * restore_queue_state call, never by coincidental pre-existing state. */ #define SNAPENT_PRE_RESTORE_QUEUE_STATUS 0xDEAD0000u static void snapent_force_queue_status(cluster_t* c, const char* name, uint32_t status) { queue_t* q = snapent_find_queue(c, name); TEST_ASSERT_NOT_NULL(q); lock(q, RTTI_LOCK_WRITE); q->queue.status = status; unlock(q); } /* ------------------------------------------------------------------ */ /* Test 7 (task 5.1, RED): a checkpoint with N jobs + M queues, loaded via */ /* persist_load_snapshot into a SEPARATE destination member, must restore */ /* every entity into the destination's live jobmgr/qmgr, report */ /* entities_restored=true, and only then reflect the checkpoint's position */ /* (Requirements 2.1, 2.2). */ /* ------------------------------------------------------------------ */ static void test_snapent_load_snapshot_restores_entities_before_advancing_position(void) { /* 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; /* Source: produces the entity-bearing checkpoint file (a real, already- * working write_snapshot, tasks 2.2/3.2). */ snapent_fixture_t* src = snapent_make(n_jobs, m_queues); TEST_ASSERT_NOT_NULL(src); 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: models a rejoining/far-behind member. jobmgr starts * EMPTY -- jobs are inserted FRESH by restore_job (job.c:96), never * pre-provisioned (unlike queues). qmgr has the SAME queue names * pre-provisioned (restore_queue_state, queue.c:208, returns -ENOENT * for a queue that doesn't already exist -- queues are config-owned, * per design.md), but seeded with a status that deliberately does NOT * match the checkpoint's real value, so a post-load match can only be * explained by a genuine restore_queue_state call. */ 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); } /* Sanity: destination truly starts without any of the source's jobs. */ for (int i = 0; i < n_jobs; i++) TEST_ASSERT_NULL(snapent_find_job(dst->c, snapent_jobid(i))); 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 well-formed entity-bearing checkpoint must succeed"); TEST_ASSERT_TRUE_MESSAGE(entities_restored, "a checkpoint carrying job/queue entities must report " "entities_restored=true (Requirement 2.1, 2.2)"); /* Requirement 2.1/2.2: every job restored into the live jobmgr, * field-identical to what was checkpointed. */ for (int i = 0; i < n_jobs; i++) { char expect_queue[48], expect_user[48], expect_cmdline[96]; snapent_job_queue_name(i, expect_queue, sizeof(expect_queue)); snapent_job_user(i, expect_user, sizeof(expect_user)); snapent_job_cmdline(i, expect_cmdline, sizeof(expect_cmdline)); job_t* j = snapent_find_job(dst->c, snapent_jobid(i)); TEST_ASSERT_NOT_NULL_MESSAGE(j, "every job in the checkpoint must be restored into the " "member's live jobmgr before the position advances"); TEST_ASSERT_EQUAL_UINT64(snapent_jobid(i), j->job.jobid); TEST_ASSERT_EQUAL_UINT32(snapent_job_state(i), j->job.state); TEST_ASSERT_NOT_NULL(j->job.submit); TEST_ASSERT_EQUAL_STRING(expect_queue, j->job.submit->queue); TEST_ASSERT_EQUAL_STRING(expect_user, j->job.submit->user); TEST_ASSERT_NOT_NULL(j->job.submit->start_info); TEST_ASSERT_EQUAL_STRING(expect_cmdline, j->job.submit->start_info->cmdline); } /* Requirement 1.1/2.1/2.2: every queue's admin status restored -- * matches the checkpoint's real value, not the pre-restore sentinel. */ 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, "restore_queue_state must have been called with the " "checkpoint's real status, not left at the pre-restore sentinel"); } /* Requirement 2.1/2.2: the position reflecting "caught up" must hold * TOGETHER with the entities actually being present -- both checked * above must be simultaneously true of the end state. */ TEST_ASSERT_EQUAL_UINT64(expected_lsn, dst->c->persist.checkpoint_lsn); TEST_ASSERT_EQUAL_UINT64(expected_lsn, dst->c->persist.current_lsn); snapent_teardown(dst); snapent_teardown(src); 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 -- */ /* not partially advanced, not advanced-then-reverted (Requirement 2.3). */ /* ------------------------------------------------------------------ */ /* Far beyond any small test checkpoint's actual remaining bytes -- * guarantees any correct decode implementation hits a short read / bounds * failure on this frame, regardless of its exact internal read strategy. */ #define SNAPENT_CORRUPT_JOB_PAYLOAD_LEN (10u * 1024u * 1024u) /* Byte offset of the FIRST job frame's payload_len field (design.md * "Checkpoint File Layout (Physical)"): legacy header, then format_version * (uint32_t), then job_count (uint32_t), then the first frame's own * payload_len (uint32_t) begins right after. */ #define SNAPENT_FIRST_JOB_PAYLOAD_LEN_OFFSET \ (SNAPENT_LEGACY_HEADER_BYTES + (long)sizeof(uint32_t) /* format_version */ \ + (long)sizeof(uint32_t) /* job_count */) static void snapent_corrupt_first_job_payload_len(const char* path) { int fd = open(path, O_RDWR); TEST_ASSERT_TRUE(fd >= 0); uint32_t bogus = SNAPENT_CORRUPT_JOB_PAYLOAD_LEN; ssize_t w = pwrite(fd, &bogus, sizeof(bogus), (off_t)SNAPENT_FIRST_JOB_PAYLOAD_LEN_OFFSET); TEST_ASSERT_EQUAL_INT((ssize_t)sizeof(bogus), w); close(fd); } static void test_snapent_load_snapshot_corrupted_entity_leaves_position_unchanged(void) { const int n_jobs = 2; snapent_fixture_t* src = snapent_make(n_jobs, 0); TEST_ASSERT_NOT_NULL(src); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&src->c->persist)); uint64_t checkpoint_lsn = src->c->persist.checkpoint_lsn; char path[4096]; snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp", src->c->persist.snap_path, checkpoint_lsn); snapent_corrupt_first_job_payload_len(path); snapent_fixture_t* dst = snapent_make(0, 0); TEST_ASSERT_NOT_NULL(dst); /* Sentinel pre-call position -- distinguishable from both 0 and the * checkpoint's real position, so "unchanged" is a meaningful assertion * rather than a 0==0 tautology. */ const uint64_t sentinel_current = 4242; const uint64_t sentinel_checkpoint = 4141; dst->c->persist.current_lsn = sentinel_current; dst->c->persist.checkpoint_lsn = sentinel_checkpoint; bool entities_restored = true; /* canary -- must not be trusted on rc!=0 */ int rc = persist_load_snapshot(&dst->c->persist, path, checkpoint_lsn, &entities_restored); TEST_ASSERT_TRUE_MESSAGE(rc != 0, "a checkpoint with a corrupted entity frame must fail to load, not " "silently succeed (Requirement 2.3)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(sentinel_current, dst->c->persist.current_lsn, "a failed restore must leave current_lsn COMPLETELY unchanged -- " "not partially advanced (Requirement 2.3)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(sentinel_checkpoint, dst->c->persist.checkpoint_lsn, "a failed restore must leave checkpoint_lsn COMPLETELY unchanged -- " "not partially advanced (Requirement 2.3)"); /* *out_entities_restored is unspecified on rc!=0 (design.md: "only set * when rc==0") -- deliberately NOT asserted here. */ snapent_teardown(dst); snapent_teardown(src); printf("test_snapent_load_snapshot_corrupted_entity_leaves_position_unchanged: PASS\n"); } /* ------------------------------------------------------------------ */ /* Test 9 (task 5.1, RED): a checkpoint file predating this feature */ /* (position marker only -- no format_version/job/queue sections, i.e. */ /* the file ends EXACTLY at the legacy header) must load without crashing, */ /* report entities_restored=false, and still advance current_lsn/ */ /* checkpoint_lsn to its recorded position -- the existing, PRESERVED */ /* legacy behavior (Requirements 2.4, 5.1, 5.2). */ /* ------------------------------------------------------------------ */ #define SNAPENT_LEGACY_CHECKPOINT_LSN 555555ULL /* Hand-builds a pre-ha-entity-snapshots checkpoint file: the exact 57-byte * legacy sentinel write_snapshot has always written (magic + lsn + optional * cfgref tail, cfgref_present=0 here) and NOTHING else -- no format_version, * no job/queue sections. Per design.md, "a file ending exactly at the * legacy header (no bytes remain) is unambiguously the old format." */ static void snapent_write_legacy_checkpoint(const char* path, uint64_t lsn) { int fd = open(path, O_CREAT | O_WRONLY | O_TRUNC, 0600); TEST_ASSERT_TRUE(fd >= 0); char header[SNAPENT_LEGACY_HEADER_BYTES]; memset(header, 0, sizeof(header)); memcpy(header, "XCESNAP\0", 8); memcpy(header + 8, &lsn, 8); /* header[16] (cfgref_present) left 0 -- no config-ref tail. */ TEST_ASSERT_EQUAL_INT((int)sizeof(header), (int)write(fd, header, sizeof(header))); close(fd); } static void test_snapent_load_snapshot_legacy_file_reports_position_only(void) { snapent_fixture_t* dst = snapent_make(0, 0); TEST_ASSERT_NOT_NULL(dst); char path[4096]; snprintf(path, sizeof(path), "%s/snap-legacy.capnp", dst->c->persist.snap_path); snapent_write_legacy_checkpoint(path, SNAPENT_LEGACY_CHECKPOINT_LSN); bool entities_restored = true; /* must come back false */ int rc = persist_load_snapshot(&dst->c->persist, path, SNAPENT_LEGACY_CHECKPOINT_LSN, &entities_restored); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "a legacy (pre-feature) checkpoint must load without error -- a " "short file is an expected shape, not corruption (Requirement 5.1, 5.2)"); TEST_ASSERT_FALSE_MESSAGE(entities_restored, "a legacy checkpoint carries no entity content -- it must never " "falsely claim full restoration (Requirement 2.4)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(SNAPENT_LEGACY_CHECKPOINT_LSN, dst->c->persist.checkpoint_lsn, "a legacy checkpoint still records the committed position it was " "taken at -- existing, preserved behavior"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(SNAPENT_LEGACY_CHECKPOINT_LSN, dst->c->persist.current_lsn, "a legacy checkpoint still advances current_lsn to its recorded " "position -- existing, preserved behavior"); snapent_teardown(dst); printf("test_snapent_load_snapshot_legacy_file_reports_position_only: PASS\n"); } /* ------------------------------------------------------------------ */ /* Suite entry point */ /* ------------------------------------------------------------------ */ void test_persist_snapshot_entities(void) { UnitySetTestFile(__FILE__); printf("\n>> entity-bearing snapshot fixture tests\n\n"); /* makeJob's deep-copy path (job_init -> makeJobsub -> makeString) needs * the global xstring table. In the full suite some other group may have * already initialised it (e.g. test_worker_join's unguarded call); * standalone (-t test_persist_snapshot_entities) it is uninitialised and * every makeString call silently returns NULL -- which job_init does not * check, so a seeded job's submit/start_info comes back NULL without any * visible failure at seed time. Probe instead of double-initialising * (mirrors test_commit_barrier_reply's cbrp-probe pattern). */ if (makeString((char*)"snapent-probe") == NULL) xstring_table_init(139); RUN_TEST(test_snapent_fixture_seeds_jobs_and_queues); RUN_TEST(test_snapent_checkpoint_job_content_roundtrip); RUN_TEST(test_snapent_checkpoint_concurrent_mutation_not_torn); 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_checkpoint_config_bytes_section_matches_source_file); RUN_TEST(test_snapent_checkpoint_config_bytes_section_empty_when_no_local_config); 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); }