wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

test(ha-entity-snapshots): RED tests for job checkpoint content (task 2.1)

Browse source 
Three tests against the current sentinel-only write_snapshot/load_snapshot:
job content round-trip and concurrent-mutation-not-torn genuinely FAIL
(RED) since no entity content is written today; write-failure-discard
already passes (write_snapshot's open()-failure path already discards
correctly) -- flagged transparently as a GREEN-from-day-one regression
guard rather than faked RED, per real fault injection (directory removal
-> genuine ENOENT).

Carries forward two notes for task 2.2: fdatasync's return value is
currently unchecked in write_snapshot (persist.c:1505), and a mid-encode-
loop fault-injection test should be added once the per-job encode loop
exists.

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 05:18:45 +00:00
parent 10f9bbfd9d
commit 0f344da89a
2 changed files with 287 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -50,10 +50,16 @@
#define _GNU_SOURCE
#include <rtti.h>
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
@ -104,6 +110,26 @@ static uint32_t snapent_queue_status(int i)
return g_snapent_queue_statuses[i % (int)SNAPENT_N_QUEUE_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 */
/* ------------------------------------------------------------------ */
@ -122,11 +148,27 @@ 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);
@ -294,6 +336,245 @@ static void test_snapent_fixture_seeds_jobs_and_queues(void)
printf("test_snapent_fixture_seeds_jobs_and_queues: PASS\n");
}
/* ------------------------------------------------------------------ */
/* 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). */
/* */
/* 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). */
/* ------------------------------------------------------------------ */
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);
/* 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);
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* dj = snapent_find_job(dst->c, 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");
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_STRING(expect_cmdline,
dj->job.submit->start_info->cmdline);
}
snapent_teardown(dst);
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_fixture_t* dst = snapent_make(0, 0);
TEST_ASSERT_NOT_NULL(dst);
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);
int rc = persist_load_snapshot(&dst->c->persist, path,
src->c->persist.checkpoint_lsn);
TEST_ASSERT_EQUAL_INT(0, rc);
job_t* dj = snapent_find_job(dst->c, 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);
if (!is_pre && !is_post)
found_torn = true;
}
}
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");
/* 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");
snapent_teardown(dst);
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");
}
/* ------------------------------------------------------------------ */
/* Suite entry point */
/* ------------------------------------------------------------------ */
@ -303,4 +584,7 @@ void test_persist_snapshot_entities(void)
printf("\n>> entity-bearing snapshot fixture tests\n\n");
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);
}