From 0f344da89aa62c4cb5a9c56b6c9a8670414d10fa Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 1 Jul 2026 05:18:45 +0000 Subject: [PATCH] test(ha-entity-snapshots): RED tests for job checkpoint content (task 2.1) 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 Claude-Session: https://claude.ai/code/session_01JDYFbrgyDHbvNYcxApKhrX --- .kiro/specs/ha-entity-snapshots/tasks.md | 4 +- .../test_persist_snapshot_entities.c | 284 ++++++++++++++++++ 2 files changed, 287 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/ha-entity-snapshots/tasks.md b/.kiro/specs/ha-entity-snapshots/tasks.md index ae1579e..780b9b4 100644 --- a/.kiro/specs/ha-entity-snapshots/tasks.md +++ b/.kiro/specs/ha-entity-snapshots/tasks.md @@ -11,7 +11,7 @@ - _Requirements: 1.1, 2.1_ - [ ] 2. Core: entity-bearing checkpoint write path — jobs -- [ ] 2.1 RED: job checkpoint content, consistency, and failure-discard tests +- [x] 2.1 RED: job checkpoint content, consistency, and failure-discard tests - Assert a checkpoint containing N jobs round-trips through load with all N job records field-identical - Assert a job actively transitioning state on a concurrent thread during the checkpoint walk @@ -125,3 +125,5 @@ ## Implementation Notes - (1.1) Test fixture `snapent_make`/`snapent_teardown` lives in `test_persist_snapshot_entities.c`; seeded jobs currently carry only jobid+state+an all-zeroed submit. Task 2.1's "field-identical round-trip" assertion needs richer submit-payload seeding (e.g. non-zero cmdline/user/queue in `xce_job_submit_t`) to be a meaningful field-by-field comparison — extend `snapent_seed_jobs`, do not re-seed separately. +- (2.1) `write_snapshot`'s existing `fdatasync(fd)` call (persist.c:1505) does not check its return value at all — task 2.2 must fix this while adding entity encoding, since design.md's error-handling contract ("any failure at any step -- encode, write, sync -- leaves no file at the final path") requires it. +- (2.1) The write-failure-discard test (`test_snapent_checkpoint_write_failure_leaves_no_final_file`) only exercises the pre-existing `open()`-failure branch (via directory removal) and passes today without any entity-encode loop existing yet. Task 2.2 should add a genuine mid-encode-loop fault-injection test (failing to encode e.g. the Nth of N jobs) once the per-job encode loop exists — the current test cannot validate that path. diff --git a/src/controller/testsuite/test_persist_snapshot_entities.c b/src/controller/testsuite/test_persist_snapshot_entities.c index 93241d3..5db1921 100644 --- a/src/controller/testsuite/test_persist_snapshot_entities.c +++ b/src/controller/testsuite/test_persist_snapshot_entities.c @@ -50,10 +50,16 @@ #define _GNU_SOURCE #include +#include +#include +#include +#include #include #include #include #include +#include +#include #include #include @@ -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); }