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

feat(ha-entity-snapshots): entity-bearing checkpoint write path for jobs (task 2.2)

Browse source 
write_snapshot now Cap'n-Proto-encodes every retained job under nested
table+per-entity RTTI_LOCK_READ (table lock alone does not exclude a
concurrent per-entity writer -- design.md's central correctness finding),
appends length-prefixed job frames after a new format_version marker, and
fixes a previously-unchecked fdatasync return. Any failure anywhere in the
path (encode, write, sync) discards the whole checkpoint -- no partial
file at the final path, proven by a real mid-encode-loop fault-injection
test (forces one specific job's encode to fail among N, asserts nothing
survives).

Also fixed two pre-existing bugs in task 1.1's test fixture: terminal-state
seeded jobs need a non-zero finish_time or job_should_retain's retention
window silently excludes them from any checkpoint; the fixture needs
xstring_table_init or every seeded job's submit/start_info silently
decodes as NULL.

Round 1 of this task also touched load_snapshot/persist.h to make the
round-trip tests pass end-to-end, but review caught a live hazard: it
advanced current_lsn/checkpoint_lsn BEFORE the entity-decode loop ran, so
a decode failure still left the position advanced -- reachable today
through the unmodified ha_handle_snapshot_chunk production path, not
hypothetical. Remediated by fully reverting load_snapshot/persist.h to
byte-identical pre-task-2.2 state and rewriting the round-trip tests to
parse write_snapshot's raw on-disk bytes directly (read_Job/decode_Job)
instead of round-tripping through persist_load_snapshot -- proves the
write side without depending on load_snapshot, which stays task 5.2's
scope entirely, not started here.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JDYFbrgyDHbvNYcxApKhrX
This commit is contained in:
Rongsong Shen 2026-07-01 06:30:28 +00:00
parent 0f344da89a
commit 5929af1078
4 changed files with 453 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

278
src/controller/testsuite/test_persist_snapshot_entities.c
View file

@ -51,6 +51,7 @@
#include <rtti.h>
#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdbool.h>
@ -65,10 +66,12 @@
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "jobj.h"
#include "qobj.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xstring.h"
/* ------------------------------------------------------------------ */
/* Deterministic seed data — later tests predict seeded values from the */
@ -174,6 +177,18 @@ static void snapent_seed_jobs(cluster_t* c, int n)
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");
}
@ -337,14 +352,131 @@ static void test_snapent_fixture_seeds_jobs_and_queues(void)
}
/* ------------------------------------------------------------------ */
/* Test 2 (task 2.1, RED): checkpoint job content round-trips through */
/* load_snapshot field-identical (Requirements 1.1, 1.2, 2.1, 2.2). */
/* Test-local raw checkpoint reader (task 2.2 remediation). */
/* */
/* Today's write_snapshot (persist.c ~1477) writes ONLY the legacy */
/* position sentinel (magic + checkpoint_lsn + optional cfgref tail) -- */
/* zero job/queue entities. A restored checkpoint therefore never */
/* contains the seeded job, so this test genuinely FAILS against */
/* current code (RED). */
/* 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 1u
#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)
@ -360,15 +492,13 @@ static void test_snapent_checkpoint_job_content_roundtrip(void)
snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp",
src->c->persist.snap_path, src->c->persist.checkpoint_lsn);
/* A second, empty fixture stands in for a rejoining/empty member --
* load_snapshot must reconstruct the source's entities into THIS
* cluster's jobmgr, not merely advance a position (Requirement 2.1). */
snapent_fixture_t* dst = snapent_make(0, 0);
TEST_ASSERT_NOT_NULL(dst);
int rc = persist_load_snapshot(&dst->c->persist, path,
src->c->persist.checkpoint_lsn);
TEST_ASSERT_EQUAL_INT(0, rc);
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];
@ -376,22 +506,22 @@ static void test_snapent_checkpoint_job_content_roundtrip(void)
snapent_job_user(i, expect_user, sizeof(expect_user));
snapent_job_cmdline(i, expect_cmdline, sizeof(expect_cmdline));
job_t* dj = snapent_find_job(dst->c, snapent_jobid(i));
xce_job_t* dj = snapent_find_decoded_job(decoded, count,
snapent_jobid(i));
TEST_ASSERT_NOT_NULL_MESSAGE(dj,
"RED: today's write_snapshot writes no entity content, so a "
"restored checkpoint never contains the seeded job");
"checkpoint's raw job frames must contain the seeded job");
TEST_ASSERT_EQUAL_UINT64(snapent_jobid(i), dj->job.jobid);
TEST_ASSERT_EQUAL_UINT32(snapent_job_state(i), dj->job.state);
TEST_ASSERT_NOT_NULL(dj->job.submit);
TEST_ASSERT_EQUAL_STRING(expect_queue, dj->job.submit->queue);
TEST_ASSERT_EQUAL_STRING(expect_user, dj->job.submit->user);
TEST_ASSERT_NOT_NULL(dj->job.submit->start_info);
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->job.submit->start_info->cmdline);
dj->submit->start_info->cmdline);
}
snapent_teardown(dst);
snapent_free_decoded_jobs(decoded, count);
snapent_teardown(src);
printf("test_snapent_checkpoint_job_content_roundtrip: PASS\n");
}
@ -458,9 +588,6 @@ static void test_snapent_checkpoint_concurrent_mutation_not_torn(void)
TEST_ASSERT_EQUAL_UINT32((uint32_t)SNAPENT_TORN_PRE_STATE, j0->job.state);
TEST_ASSERT_EQUAL_INT(0, (int)j0->job.start_time);
snapent_fixture_t* dst = snapent_make(0, 0);
TEST_ASSERT_NOT_NULL(dst);
snapent_mutator_arg_t marg;
marg.c = src->c;
marg.jobid = jobid;
@ -480,21 +607,29 @@ static void test_snapent_checkpoint_concurrent_mutation_not_torn(void)
snprintf(path, sizeof(path), "%s/snap-%016" PRIx64 ".capnp",
src->c->persist.snap_path, src->c->persist.checkpoint_lsn);
int rc = persist_load_snapshot(&dst->c->persist, path,
src->c->persist.checkpoint_lsn);
TEST_ASSERT_EQUAL_INT(0, rc);
/* 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");
job_t* dj = snapent_find_job(dst->c, jobid);
xce_job_t* dj = snapent_find_decoded_job(decoded, count, jobid);
if (dj != NULL) {
found_any = true;
bool is_pre = (dj->job.state == (uint32_t)SNAPENT_TORN_PRE_STATE
&& dj->job.start_time == 0);
bool is_post = (dj->job.state == (uint32_t)SNAPENT_TORN_POST_STATE
&& dj->job.start_time == SNAPENT_TORN_POST_START_TIME);
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);
@ -503,14 +638,10 @@ static void test_snapent_checkpoint_concurrent_mutation_not_torn(void)
"checkpoint captured a torn mix of job state/start_time -- neither "
"fully the pre-transition nor fully the post-transition tuple");
/* RED: today's write_snapshot writes no entity content at all, so the
* job is never found in any restored checkpoint across the whole loop. */
TEST_ASSERT_TRUE_MESSAGE(found_any,
"a restored checkpoint must contain the seeded job's entity state "
"at least once across the loop -- today's sentinel-only "
"write_snapshot never restores it");
"a checkpoint must contain the seeded job's entity state at least "
"once across the loop");
snapent_teardown(dst);
snapent_teardown(src);
printf("test_snapent_checkpoint_concurrent_mutation_not_torn: PASS\n");
}
@ -575,6 +706,53 @@ static void test_snapent_checkpoint_write_failure_leaves_no_final_file(void)
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");
}
/* ------------------------------------------------------------------ */
/* Suite entry point */
/* ------------------------------------------------------------------ */
@ -583,8 +761,20 @@ void test_persist_snapshot_entities(void)
{
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);
}