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

feat(scheduler-core): implement periodic snapshot checkpointing (task 9.3)

Browse source 
Periodic checkpointing hooks into sched_record_write_events(): once per
call, when record_snapshot_interval > 0 and record_event_count crosses a
multiple of it, writes a fresh SnapshotEntry via the same
write_snapshot_entry() helper sched_record_start() already uses (made
non-static, zero duplication). Safe from the session_active precondition
because this call site is only ever reached from
xce_sched_driver_translate_pending_events(), which always runs strictly
between sessions.

Confirmed the write-failure error-handling half (AC 10) was already
complete from task 9.2's sched_record_abort_on_error() -- added the
missing end-to-end test using a genuine closed-fd write failure rather
than a mocked return code, proving the driver keeps processing sessions
afterward.

25/25 scheduler suite, clang-format clean. Two non-blocking follow-ups
logged in tasks.md: --record-snapshot-interval has no CLI wiring yet
(AC 11's own struct-field mechanism is fully tested, but nothing sets it
in production), and test_record_lifecycle.c has grown past this suite's
prior size precedent with a natural future split point identified.
This commit is contained in:
Rongsong Shen 2026-08-05 10:02:16 +00:00
parent ff8d4f416e
commit 7494542a60
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
6 changed files with 531 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

412
src/scheduler/testsuite/test_record_lifecycle.c
View file

@ -1,9 +1,10 @@
/*
* test_record_lifecycle.c - contract test for the record start/stop
* lifecycle and the driver-loop recording-write hooks (scheduler-core
* spec, task 9.2, Req 5 AC 10).
* lifecycle, the driver-loop recording-write hooks, periodic snapshot
* checkpointing, and recording error handling (scheduler-core spec, tasks
* 9.2/9.3, Req 5 AC 10/AC 11).
*
* Observable under test (this task's own text):
* Observable under test (task 9.2's own text):
* - start -> the file begins with a decodable ReplayHeader then a
* decodable SnapshotEntry.
* - start -> submit an event batch -> complete one session -> stop
@ -14,6 +15,27 @@
* - a TOCTOU-style symlink/hardlink swap between path validation and the
* eventual open() is rejected, not silently followed.
*
* Observable under test (task 9.3's own text, added this revision):
* - recording through a configured `record_snapshot_interval` produces
* the expected NUMBER of extra periodic SnapshotEntry messages in the
* file (interval=3, 7 events across 7 calls -> 2 extra checkpoints;
* interval=0 -> zero extra checkpoints regardless of event count; one
* large single-call batch crossing two interval multiples -> exactly
* ONE checkpoint, proving/documenting this implementation's "once per
* call" boundary choice -- see recording_events.c's own comment on
* sched_record_write_events() for the full rationale).
* - simulating a write failure leaves the scheduler still processing
* sessions afterward with recording cleanly disabled: a genuine fd-level
* write failure (not a simulated return code -- see the fflush()+
* setvbuf(_IONBF)+close(fd) sequence in
* test_write_failure_disables_recording_and_driver_keeps_running(), and
* .wolf/cerebrum.md's matching Key Learning for why plain fd-closing
* alone is not sufficient against a buffered stdio stream) triggers
* sched_record_abort_on_error()'s cleanup (recording=false,
* record_file/record_file_path both NULL), then a full real
* normal-mode round still completes with no crash/hang and no further
* recording I/O attempted.
*
* White-box #include of ../src/core.c, ../src/core_clock.c,
* ../src/core_events.c, ../src/core_session.c, ../src/driver.c,
* ../src/plugin.c (same file set/rationale as
@ -809,6 +831,383 @@ static void test_toctou_hardlink_planted_after_validation_rejected(void)
TEST_ASSERT_EQUAL_STRING(victim_content, readback);
}
/* ---------------------------------------------------------------------- */
/* Test 6: periodic snapshot checkpointing (task 9.3, Req 5 AC 11) -- */
/* interval N produces the expected NUMBER of extra SnapshotEntry */
/* messages, decoded and counted (not just file-size-checked) from the */
/* real generated capnp decoders, same convention as the tests above. */
/* ---------------------------------------------------------------------- */
/* Builds and submits one synthetic JOB_STATE_CHANGE event via
* sched_record_write_events() -- the direct entry point recording_events.c
* exposes, bypassing the need to drive a real session (xce_sched_core_t's
* `session_active` stays 0 throughout, which is exactly what
* write_snapshot_entry()'s xce_sched_core_save_snapshot() call requires --
* core.h's own documented precondition -- so calling this hook directly,
* without ever running a session, is a faithful, simpler stand-in for the
* real driver.c call site). `jobid` varies per call purely so decoded
* events are individually distinguishable in assertions below. */
static void submit_one_jobstate_event(xce_scheduler_t* sched, uint64_t jobid)
{
xce_core_event_t ev;
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
ev.u.job_state_change.jobid = jobid;
TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(sched, &ev, 1));
}
/* Reads exactly one capnp message off `f` and decodes it as a ReplayEvent,
* asserting it is a jobStateChange event for `expected_jobid`. */
static void expect_next_msg_is_jobstate_event(FILE* f, uint64_t expected_jobid)
{
struct capn c;
ReplayEvent_ptr p;
struct ReplayEvent wire;
replay_event_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayEvent(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayEvent(&dst, &wire);
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_jobStateChange,
(int)dst.body_which);
TEST_ASSERT_NOT_NULL(dst.body.job_state_change);
TEST_ASSERT_EQUAL_UINT64(expected_jobid, dst.body.job_state_change->jobid);
free_ReplayEvent(&dst);
capn_free(&c);
}
/* Reads exactly one capnp message off `f` and decodes it as a
* SnapshotEntry, asserting its `data` field is present (same structural
* check test 1 above already establishes for the initial SnapshotEntry). */
static void expect_next_msg_is_snapshot_entry(FILE* f)
{
struct capn c;
SnapshotEntry_ptr snap;
SnapshotData_ptr sdata;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
snap.p = capn_getp(capn_root(&c), 0, 1);
sdata = SnapshotEntry_get_data(snap);
TEST_ASSERT_FALSE(sdata.p.type == CAPN_NULL);
capn_free(&c);
}
/* Reads exactly one capnp message off `f` and decodes it as the
* end-of-recording marker, ReplayEvent{endSession}. */
static void expect_next_msg_is_end_session_marker(FILE* f)
{
struct capn c;
ReplayEvent_ptr p;
struct ReplayEvent wire;
replay_event_t dst;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
p.p = capn_getp(capn_root(&c), 0, 1);
read_ReplayEvent(&wire, p);
memset(&dst, 0, sizeof(dst));
decode_ReplayEvent(&dst, &wire);
TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_endSession,
(int)dst.body_which);
free_ReplayEvent(&dst);
capn_free(&c);
}
/* Skips the ReplayHeader + initial SnapshotEntry that sched_record_start()
* always writes first (Req 5 AC 20) -- every test below only cares about
* what comes AFTER those two, so this consumes them structurally (same
* "structural check only" pattern test 2 above already uses for its own
* ReplaySession message). */
static void skip_header_and_initial_snapshot(FILE* f)
{
struct capn c;
TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0)); /* ReplayHeader */
capn_free(&c);
TEST_ASSERT_EQUAL_INT(0,
capn_init_fp(&c, f, 0)); /* initial SnapshotEntry */
capn_free(&c);
}
/* interval=3, 7 events submitted ONE PER CALL (7 separate
* sched_record_write_events() calls) -- record_event_count runs
* 1,2,3,4,5,6,7, landing exactly on the interval's multiples at 3 and 6.
* Expect exactly 2 extra periodic SnapshotEntry messages, positioned
* immediately after the 3rd and 6th events respectively (this task's own
* Observable, using this task's own example numbers). */
static void
test_periodic_checkpoint_interval_produces_expected_snapshot_count(void)
{
char record_dir[PATH_MAX];
xce_scheduler_t sched;
char path_copy[PATH_MAX];
FILE* f;
struct capn c;
int i;
make_tmp_dir(record_dir, sizeof(record_dir), "_periodic");
fixture_init(&sched, record_dir);
sched.record_snapshot_interval = 3ULL;
TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, "periodic.replay"));
strcpy(path_copy, sched.record_file_path);
for (i = 0; i < 7; i++) {
submit_one_jobstate_event(&sched, (uint64_t)(100 + i));
}
TEST_ASSERT_EQUAL_UINT64(7ULL, sched.record_event_count);
TEST_ASSERT_TRUE(sched.recording); /* no abort fired along the way */
TEST_ASSERT_EQUAL_INT(0, sched_record_stop(&sched));
f = fopen(path_copy, "r");
TEST_ASSERT_NOT_NULL(f);
skip_header_and_initial_snapshot(f);
expect_next_msg_is_jobstate_event(f, 100ULL);
expect_next_msg_is_jobstate_event(f, 101ULL);
expect_next_msg_is_jobstate_event(f, 102ULL);
expect_next_msg_is_snapshot_entry(f); /* count==3, extra checkpoint #1 */
expect_next_msg_is_jobstate_event(f, 103ULL);
expect_next_msg_is_jobstate_event(f, 104ULL);
expect_next_msg_is_jobstate_event(f, 105ULL);
expect_next_msg_is_snapshot_entry(f); /* count==6, extra checkpoint #2 */
expect_next_msg_is_jobstate_event(f, 106ULL);
expect_next_msg_is_end_session_marker(f);
/* Nothing more to read -- exactly 2 extra SnapshotEntry messages, no
* more, no fewer. */
TEST_ASSERT_NOT_EQUAL(0, capn_init_fp(&c, f, 0));
fclose(f);
printf("test_periodic_checkpoint_interval_produces_expected_snapshot_count:"
" PASS\n");
}
/* record_snapshot_interval == 0 (the default/disabled case) -- many events
* recorded, zero extra SnapshotEntry messages produced regardless. */
static void test_periodic_checkpoint_interval_zero_disables_checkpointing(void)
{
char record_dir[PATH_MAX];
xce_scheduler_t sched;
char path_copy[PATH_MAX];
FILE* f;
struct capn c;
int i;
make_tmp_dir(record_dir, sizeof(record_dir), "_nointerval");
fixture_init(&sched, record_dir);
TEST_ASSERT_EQUAL_UINT64(0ULL,
sched.record_snapshot_interval); /* default */
TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, "nointerval.replay"));
strcpy(path_copy, sched.record_file_path);
for (i = 0; i < 10; i++) {
submit_one_jobstate_event(&sched, (uint64_t)(200 + i));
}
TEST_ASSERT_EQUAL_UINT64(10ULL, sched.record_event_count);
TEST_ASSERT_EQUAL_INT(0, sched_record_stop(&sched));
f = fopen(path_copy, "r");
TEST_ASSERT_NOT_NULL(f);
skip_header_and_initial_snapshot(f);
for (i = 0; i < 10; i++) {
expect_next_msg_is_jobstate_event(f, (uint64_t)(200 + i));
}
expect_next_msg_is_end_session_marker(f);
/* No SnapshotEntry anywhere in that stream -- exactly the 10 events
* plus the end marker, nothing more. */
TEST_ASSERT_NOT_EQUAL(0, capn_init_fp(&c, f, 0));
fclose(f);
printf("test_periodic_checkpoint_interval_zero_disables_checkpointing:"
" PASS\n");
}
/* interval=3, but all 6 events are submitted in ONE single
* sched_record_write_events() call (one batch) -- record_event_count jumps
* straight from 0 to 6 in that single call, crossing BOTH the count==3 and
* count==6 multiples within it. Documents/proves this task's chosen
* "once per call" semantics: only ONE checkpoint is written for the call,
* not one per crossed boundary (see recording_events.c's own comment on
* sched_record_write_events() for the full rationale) -- this is the exact
* scenario where "once per crossed boundary" and "once per call" would
* observably differ, and this test proves which one this implementation
* chose. */
static void
test_periodic_checkpoint_single_batch_crossing_two_boundaries_checkpoints_once(
void)
{
char record_dir[PATH_MAX];
xce_scheduler_t sched;
char path_copy[PATH_MAX];
xce_core_event_t events[6];
FILE* f;
struct capn c;
int i;
make_tmp_dir(record_dir, sizeof(record_dir), "_batchboundary");
fixture_init(&sched, record_dir);
sched.record_snapshot_interval = 3ULL;
TEST_ASSERT_EQUAL_INT(0,
sched_record_start(&sched, "batchboundary.replay"));
strcpy(path_copy, sched.record_file_path);
memset(events, 0, sizeof(events));
for (i = 0; i < 6; i++) {
events[i].type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
events[i].u.job_state_change.jobid = (uint64_t)(300 + i);
}
TEST_ASSERT_EQUAL_INT(0, sched_record_write_events(&sched, events, 6));
TEST_ASSERT_EQUAL_UINT64(6ULL, sched.record_event_count);
TEST_ASSERT_TRUE(sched.recording);
TEST_ASSERT_EQUAL_INT(0, sched_record_stop(&sched));
f = fopen(path_copy, "r");
TEST_ASSERT_NOT_NULL(f);
skip_header_and_initial_snapshot(f);
for (i = 0; i < 6; i++) {
expect_next_msg_is_jobstate_event(f, (uint64_t)(300 + i));
}
expect_next_msg_is_snapshot_entry(f); /* exactly ONE, not two */
expect_next_msg_is_end_session_marker(f);
TEST_ASSERT_NOT_EQUAL(0, capn_init_fp(&c, f, 0));
fclose(f);
printf("test_periodic_checkpoint_single_batch_crossing_two_boundaries_"
"checkpoints_once: PASS\n");
}
/* A periodic checkpoint's OWN write can fail independently of the
* per-event writes that precede it in the same call -- proven here by
* forcing xce_sched_core_save_snapshot() (which write_snapshot_entry()
* calls first) to fail via core.h's own documented `session_active != 0`
* precondition, entirely independent of file-descriptor health. Proves
* task 9.3's "route through the SAME sched_record_abort_on_error() path,
* do not special-case it" requirement for the periodic-checkpoint branch
* specifically (the general write-failure test below already covers the
* per-event-write branch). */
static void
test_periodic_checkpoint_write_failure_disables_recording_cleanly(void)
{
char record_dir[PATH_MAX];
xce_scheduler_t sched;
xce_core_event_t ev;
make_tmp_dir(record_dir, sizeof(record_dir), "_ckptfail");
fixture_init(&sched, record_dir);
sched.record_snapshot_interval = 1ULL; /* every single event checkpoints */
TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, "ckptfail.replay"));
TEST_ASSERT_TRUE(sched.recording);
/* Force xce_sched_core_save_snapshot()'s precondition check to fail
* (core.h: "-EBUSY ... checked BEFORE any mutation") -- no real session
* is ever driven; this only fakes the flag write_snapshot_entry()'s
* save_snapshot() call reads. */
sched.core.session_active = 1;
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
ev.u.job_state_change.jobid = 400ULL;
/* The event itself writes fine; the periodic checkpoint that fires
* immediately after it (interval=1) is what fails and aborts. */
TEST_ASSERT_TRUE(sched_record_write_events(&sched, &ev, 1) < 0);
TEST_ASSERT_FALSE(sched.recording);
TEST_ASSERT_NULL(sched.record_file);
TEST_ASSERT_NULL(sched.record_file_path);
sched.core.session_active = 0; /* restore -- avoid tripping other state */
}
/* ---------------------------------------------------------------------- */
/* Test 7: write-failure end-to-end (task 9.3's own Observable) -- */
/* a genuine write failure (the underlying fd closed out from under the */
/* buffered FILE*, then forced unbuffered so the very next fwrite() hits */
/* the syscall directly instead of silently buffering past the break) */
/* cleanly disables recording via sched_record_abort_on_error(), AND the */
/* driver loop keeps working afterward -- a full real session still */
/* completes with no crash/hang, recording simply staying inactive. */
/* ---------------------------------------------------------------------- */
static void test_write_failure_disables_recording_and_driver_keeps_running(void)
{
char record_dir[PATH_MAX];
xce_scheduler_t sched;
xce_core_event_t ev;
int fd;
uint64_t event_count_before_failure;
make_tmp_dir(record_dir, sizeof(record_dir), "_writefail");
fixture_init(&sched, record_dir);
TEST_ASSERT_EQUAL_INT(0, sched_record_start(&sched, "writefail.replay"));
TEST_ASSERT_TRUE(sched.recording);
TEST_ASSERT_NOT_NULL(sched.record_file);
/* Commit the header+initial-snapshot bytes sched_record_start() already
* buffered to the still-healthy fd, then force the stream unbuffered so
* every subsequent fwrite() hits the underlying fd directly (no stdio
* buffering to mask a failure), THEN close that fd out from under the
* FILE* -- "close the underlying fd out from under the FILE* before a
* write", per this task's own suggested technique. */
TEST_ASSERT_EQUAL_INT(0, fflush(sched.record_file));
TEST_ASSERT_EQUAL_INT(0, setvbuf(sched.record_file, NULL, _IONBF, 0));
fd = fileno(sched.record_file);
TEST_ASSERT_TRUE(fd >= 0);
TEST_ASSERT_EQUAL_INT(0, close(fd));
memset(&ev, 0, sizeof(ev));
ev.type = XCE_CORE_EVENT_JOB_STATE_CHANGE;
ev.u.job_state_change.jobid = 500ULL;
/* The write genuinely fails now (EBADF on the closed fd, surfaced
* immediately since the stream is unbuffered) -- proving a REAL I/O
* failure, not a simulated return code. */
TEST_ASSERT_TRUE(sched_record_write_events(&sched, &ev, 1) < 0);
/* sched_record_abort_on_error()'s cleanup fired: recording cleanly
* disabled, file handle closed+NULLed, path freed+NULLed -- this
* task's own Observable, "recording cleanly disabled". */
TEST_ASSERT_FALSE(sched.recording);
TEST_ASSERT_NULL(sched.record_file);
TEST_ASSERT_NULL(sched.record_file_path);
event_count_before_failure = sched.record_event_count;
/* "the scheduler still processing sessions afterward" -- drive one full
* real normal-mode round (real picker thread + stand-in allocator
* thread, same helper the earlier full-cycle test uses) through the
* now-recording-inactive scheduler. No crash, no hang, and no further
* recording I/O is attempted (recording_inactive_hooks test above
* already proves the zero-I/O contract for `recording == false`; this
* proves it holds true specifically AFTER a real abort, not just from a
* fixture that never started recording). */
run_one_real_session(&sched);
TEST_ASSERT_FALSE(sched.recording);
TEST_ASSERT_NULL(sched.record_file);
TEST_ASSERT_NULL(sched.record_file_path);
TEST_ASSERT_EQUAL_UINT64(event_count_before_failure,
sched.record_event_count);
printf("test_write_failure_disables_recording_and_driver_keeps_running:"
" PASS\n");
}
int main(void)
{
UNITY_BEGIN();
@ -821,6 +1220,13 @@ int main(void)
RUN_TEST(test_null_sched_returns_einval);
RUN_TEST(test_toctou_symlink_planted_after_validation_rejected);
RUN_TEST(test_toctou_hardlink_planted_after_validation_rejected);
RUN_TEST(
test_periodic_checkpoint_interval_produces_expected_snapshot_count);
RUN_TEST(test_periodic_checkpoint_interval_zero_disables_checkpointing);
RUN_TEST(
test_periodic_checkpoint_single_batch_crossing_two_boundaries_checkpoints_once);
RUN_TEST(test_periodic_checkpoint_write_failure_disables_recording_cleanly);
RUN_TEST(test_write_failure_disables_recording_and_driver_keeps_running);
return UNITY_END();
}