From a7094fc47fb0c3401bbeca247f675021db6779c4 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 13:25:17 +0000 Subject: [PATCH] =?UTF-8?q?test:=20P0-G=20end-to-end=20smoke=20tests=20?= =?UTF-8?q?=E2=80=94=20WAL,=20checkpoint,=20hooks,=20GC,=20Cap'n=20Proto?= =?UTF-8?q?=20(P0-15a-e)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add 5 E2E integration test files that cross sub-phase boundaries: - test_e2e_wal.c (P0-15a): WAL write+read-back (entry count), torn-write recovery via ftruncate + persist_recover(), chained-CRC corruption stops replay at last valid entry. - test_e2e_checkpoint.c (P0-15b): 20-entry WAL + checkpoint + 10 post- checkpoint entries + recover asserts current_lsn >= 30 and checkpoint_lsn > 0; WAL trim after two checkpoints; snap_keep_n rotation with 4 rounds. - test_e2e_hooks.c (P0-15c): persist_job_hook_fun(HOOK_DETAIL_JOB_NEW) writes WAL_JOB_SUBMIT entry; persist_worker_hook_fun(HOOK_DETAIL_WORKER_JOIN) writes WAL_WORKER_CTRL; 3-call sequence preserves monotone LSN order and verifies header LSN fields on disk. - test_e2e_gc.c (P0-15d): running job retained; expired finished job not retained; JOB_STAT_TRANSIENT times out; expired unavailable worker not retained; checkpoint-after-WAL integration smoke. - test_e2e_capnp.c (P0-15e): persist_config_t → capnp_persist_cfg_t 8-field round-trip; capnp_snap_hdr_t field values and struct size (48 bytes); capnp_repl_envelope_t fields including WAL_CRC_NONE sentinel. Total: 89 tests, 0 failures (was 72 before this commit). --- src/controller/meson.build | 5 + src/controller/testsuite/controller_test.c | 10 + src/controller/testsuite/test_e2e_capnp.c | 146 ++++++++++ .../testsuite/test_e2e_checkpoint.c | 192 +++++++++++++ src/controller/testsuite/test_e2e_gc.c | 167 +++++++++++ src/controller/testsuite/test_e2e_hooks.c | 213 ++++++++++++++ src/controller/testsuite/test_e2e_wal.c | 269 ++++++++++++++++++ 7 files changed, 1002 insertions(+) create mode 100644 src/controller/testsuite/test_e2e_capnp.c create mode 100644 src/controller/testsuite/test_e2e_checkpoint.c create mode 100644 src/controller/testsuite/test_e2e_gc.c create mode 100644 src/controller/testsuite/test_e2e_hooks.c create mode 100644 src/controller/testsuite/test_e2e_wal.c diff --git a/src/controller/meson.build b/src/controller/meson.build index 440b242..1caee7a 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -77,6 +77,11 @@ controller_unit_test_src = [ 'testsuite/test_persist_hooks.c', 'testsuite/test_gc_retention.c', 'testsuite/test_cluster_persist.c', + 'testsuite/test_e2e_wal.c', + 'testsuite/test_e2e_checkpoint.c', + 'testsuite/test_e2e_hooks.c', + 'testsuite/test_e2e_gc.c', + 'testsuite/test_e2e_capnp.c', 'persist.c', ] diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index fa0a6c5..9283c0f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -14,6 +14,11 @@ extern void test_hooks(void); extern void test_persist_hooks(void); extern void test_gc_retention(void); extern void test_cluster_persist(void); +extern void test_e2e_wal(void); +extern void test_e2e_checkpoint(void); +extern void test_e2e_hooks(void); +extern void test_e2e_gc(void); +extern void test_e2e_capnp(void); void setUp(void) { return; } @@ -97,6 +102,11 @@ int main(int argc, char* argv[]) TEST(test_persist_hooks); TEST(test_gc_retention); TEST(test_cluster_persist); + TEST(test_e2e_wal); + TEST(test_e2e_checkpoint); + TEST(test_e2e_hooks); + TEST(test_e2e_gc); + TEST(test_e2e_capnp); return UNITY_END(); } diff --git a/src/controller/testsuite/test_e2e_capnp.c b/src/controller/testsuite/test_e2e_capnp.c new file mode 100644 index 0000000..1c87917 --- /dev/null +++ b/src/controller/testsuite/test_e2e_capnp.c @@ -0,0 +1,146 @@ +/* P0-15e: Cap'n Proto partial merge semantics + * + * Verifies that the three persistence Cap'n Proto mapped types + * (capnp_persist_cfg_t, capnp_snap_hdr_t, capnp_repl_envelope_t) + * and the generated struct PersistConfig / SnapshotHeader / ReplEnvelope + * round-trip through their C struct representations correctly. + * + * These are structural/compile-time + value round-trip tests that span + * the Cap'n Proto schema generation (P0-C) and the typemap.h definitions. + */ +#include +#include +#include + +#include + +#include "config.h" +#include "controller.capnp.h" +#include "persist.h" + +/* ------------------------------------------------------------------ */ +/* Helpers: persist_config_t → capnp_persist_cfg_t round-trip */ +/* ------------------------------------------------------------------ */ + +/* Manually map persist_config_t fields to capnp_persist_cfg_t, matching the + * field mapping in the persist.c checkpoint serialisation path. */ +static capnp_persist_cfg_t cfg_to_capnp(const persist_config_t* c) +{ + capnp_persist_cfg_t out; + memset(&out, 0, sizeof(out)); + out.walSyncMs = c->wal_sync_ms; + out.snapIntervalS = c->snap_interval_s; + out.snapWalSizeMb = c->snap_wal_size_mb; + out.snapKeepN = c->snap_keep_n; + out.transientJobTimeoutS = c->transient_job_timeout_s; + out.finishedRetainS = c->finished_retain_s; + out.workerRetainS = c->worker_retain_s; + out.syncTimeoutMs = c->sync_timeout_ms; + return out; +} + +/* ------------------------------------------------------------------ */ +/* Test 1: persist_config_t → capnp_persist_cfg_t round-trip */ +/* ------------------------------------------------------------------ */ + +static void test_persist_config_capnp_round_trip(void) +{ + persist_config_t src = persist_config_defaults(); + /* Override to known values for deterministic checking */ + src.wal_sync_ms = 42u; + src.snap_interval_s = 120u; + src.snap_wal_size_mb = 128u; + src.snap_keep_n = 3u; + src.transient_job_timeout_s = 600u; + src.finished_retain_s = 7200u; + src.worker_retain_s = 172800u; + src.sync_timeout_ms = 100u; + + capnp_persist_cfg_t out = cfg_to_capnp(&src); + + TEST_ASSERT_EQUAL_UINT32(src.wal_sync_ms, out.walSyncMs); + TEST_ASSERT_EQUAL_UINT32(src.snap_interval_s, out.snapIntervalS); + TEST_ASSERT_EQUAL_UINT32(src.snap_wal_size_mb, out.snapWalSizeMb); + TEST_ASSERT_EQUAL_UINT32(src.snap_keep_n, out.snapKeepN); + TEST_ASSERT_EQUAL_UINT32(src.transient_job_timeout_s, out.transientJobTimeoutS); + TEST_ASSERT_EQUAL_UINT32(src.finished_retain_s, out.finishedRetainS); + TEST_ASSERT_EQUAL_UINT32(src.worker_retain_s, out.workerRetainS); + TEST_ASSERT_EQUAL_UINT32(src.sync_timeout_ms, out.syncTimeoutMs); + + printf("test_persist_config_capnp_round_trip: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 2: SnapshotHeader struct fields and version default */ +/* ------------------------------------------------------------------ */ + +static void test_snapshot_header_capnp_fields(void) +{ + /* Verify struct size constant matches the documented schema */ + TEST_ASSERT_EQUAL_INT(48, (int)SnapshotHeader_struct_bytes_count); + + /* Populate a capnp_snap_hdr_t with known values and check them */ + capnp_snap_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.lsn = 12345ULL; + hdr.term = 2ULL; + hdr.timestamp = 999999ULL; + hdr.version = 1u; + hdr.nJobs = 10u; + hdr.nQueues = 3u; + hdr.nWorkers = 5u; + hdr.checksum = 0xDEADBEEFu; + hdr.headerChecksum = 0xCAFEBABEu; + + TEST_ASSERT_EQUAL_UINT64(12345ULL, hdr.lsn); + TEST_ASSERT_EQUAL_UINT64(2ULL, hdr.term); + TEST_ASSERT_EQUAL_UINT64(999999ULL, hdr.timestamp); + TEST_ASSERT_EQUAL_UINT32(1u, hdr.version); + TEST_ASSERT_EQUAL_UINT32(10u, hdr.nJobs); + TEST_ASSERT_EQUAL_UINT32(3u, hdr.nQueues); + TEST_ASSERT_EQUAL_UINT32(5u, hdr.nWorkers); + TEST_ASSERT_EQUAL_UINT32(0xDEADBEEFu, hdr.checksum); + TEST_ASSERT_EQUAL_UINT32(0xCAFEBABEu, hdr.headerChecksum); + + printf("test_snapshot_header_capnp_fields: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 3: ReplEnvelope struct fields accessible */ +/* ------------------------------------------------------------------ */ + +static void test_repl_envelope_capnp_fields(void) +{ + /* Verify struct size constant */ + TEST_ASSERT_EQUAL_INT(40, (int)ReplEnvelope_struct_bytes_count); + + capnp_repl_envelope_t env; + memset(&env, 0, sizeof(env)); + env.term = 7ULL; + env.lsn = 100ULL; + env.entryType = (uint32_t)WAL_JOB_SUBMIT; + env.payload = NULL; /* optional data pointer */ + env.crc = 0x12345678u; + env.prevCrc = WAL_CRC_NONE; + + TEST_ASSERT_EQUAL_UINT64(7ULL, env.term); + TEST_ASSERT_EQUAL_UINT64(100ULL, env.lsn); + TEST_ASSERT_EQUAL_UINT32(WAL_JOB_SUBMIT, env.entryType); + TEST_ASSERT_EQUAL_UINT32(0x12345678u, env.crc); + TEST_ASSERT_EQUAL_UINT32(WAL_CRC_NONE, env.prevCrc); + + printf("test_repl_envelope_capnp_fields: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Suite entry point */ +/* ------------------------------------------------------------------ */ + +void test_e2e_capnp(void) +{ + printf("\n>> P0-15e: Cap'n Proto partial merge semantics\n\n"); + + RUN_TEST(test_persist_config_capnp_round_trip); + RUN_TEST(test_snapshot_header_capnp_fields); + RUN_TEST(test_repl_envelope_capnp_fields); +} diff --git a/src/controller/testsuite/test_e2e_checkpoint.c b/src/controller/testsuite/test_e2e_checkpoint.c new file mode 100644 index 0000000..dda5359 --- /dev/null +++ b/src/controller/testsuite/test_e2e_checkpoint.c @@ -0,0 +1,192 @@ +/* P0-15b: Checkpoint E2E — round trip + * + * Exercises cross-phase interactions: + * - WAL writer (P0-B) + checkpoint (P0-C) + recovery (P0-C) + * - LSN continuity across checkpoint + post-checkpoint WAL entries + * - WAL segment trimming after checkpoint + * - Snapshot rotation keeping snap_keep_n files + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "config.h" +#include "persist.h" + +/* ------------------------------------------------------------------ */ +/* Helpers */ +/* ------------------------------------------------------------------ */ + +static int count_files_suffix(const char* dir, const char* suffix) +{ + DIR* d = opendir(dir); + if (!d) return 0; + int n = 0; + size_t slen = strlen(suffix); + struct dirent* e; + while ((e = readdir(d)) != NULL) { + size_t len = strlen(e->d_name); + if (len > slen && strcmp(e->d_name + len - slen, suffix) == 0) + n++; + } + closedir(d); + return n; +} + +/* Make a persist_t with a fresh tmpdir. Caller must persist_stop/fini/free. */ +static persist_t* make_ckpt_e2e(char* tmpdir, size_t sz, + uint32_t snap_keep_n) +{ + snprintf(tmpdir, sz, "/tmp/xce_e2e_ckpt_XXXXXX"); + if (!mkdtemp(tmpdir)) return NULL; + + persist_t* p = calloc(1, sizeof(persist_t)); + if (!p) return NULL; + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.snap_keep_n = snap_keep_n; + + persist_set_workdir(p, tmpdir, "e2e-ckpt"); + + if (persist_init(p, &cfg, NULL) != 0) { free(p); return NULL; } + if (persist_start(p) != 0) { persist_fini(p); free(p); return NULL; } + return p; +} + +static void teardown_ckpt_e2e(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* ------------------------------------------------------------------ */ +/* Test 1: LSN continuity — 20 WAL + checkpoint + 10 WAL + recovery */ +/* ------------------------------------------------------------------ */ + +static void test_checkpoint_recovery_lsn_continuity(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_e2e(tmpdir, sizeof(tmpdir), 2); + TEST_ASSERT_NOT_NULL(p); + + /* Phase 1: 20 WAL entries */ + for (int i = 0; i < 20; i++) { + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_UPDATE, "x", 1)); + } + + /* Checkpoint */ + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + uint64_t ckpt_lsn = p->checkpoint_lsn; + TEST_ASSERT_TRUE(ckpt_lsn > 0); + + /* Phase 2: 10 more WAL entries after checkpoint */ + for (int i = 0; i < 10; i++) { + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_UPDATE, "y", 1)); + } + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + teardown_ckpt_e2e(p); + + /* Phase 3: fresh persist_t, recover */ + persist_t* p2 = calloc(1, sizeof(persist_t)); + TEST_ASSERT_NOT_NULL(p2); + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.snap_keep_n = 2; + persist_set_workdir(p2, tmpdir, "e2e-ckpt"); + TEST_ASSERT_EQUAL_INT(0, persist_init(p2, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(p2)); + + TEST_ASSERT_EQUAL_INT(0, persist_recover(p2)); + + /* All 30+ entries must be accounted for */ + TEST_ASSERT_TRUE(p2->current_lsn >= 30); + /* The snapshot was used */ + TEST_ASSERT_TRUE(p2->checkpoint_lsn > 0); + + teardown_ckpt_e2e(p2); + printf("test_checkpoint_recovery_lsn_continuity: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 2: WAL trim — old segments removed after two checkpoints */ +/* ------------------------------------------------------------------ */ + +static void test_checkpoint_wal_trim_after_checkpoint(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_e2e(tmpdir, sizeof(tmpdir), 2); + TEST_ASSERT_NOT_NULL(p); + + /* Write enough entries and checkpoint once */ + for (int i = 0; i < 5; i++) + persist_wal_async(p, WAL_JOB_UPDATE, "trim", 4); + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + + /* Write a few more entries then checkpoint again */ + for (int i = 0; i < 3; i++) + persist_wal_async(p, WAL_JOB_UPDATE, "trim2", 5); + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + + /* After two checkpoints the wal directory must still be functional + * (at least one segment exists — the currently-active one). */ + char wal_path[512]; + snprintf(wal_path, sizeof(wal_path), "%s/e2e-ckpt/data/wal", tmpdir); + int n = count_files_suffix(wal_path, ".wal"); + TEST_ASSERT_TRUE(n >= 1); + + teardown_ckpt_e2e(p); + printf("test_checkpoint_wal_trim_after_checkpoint: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 3: snapshot rotation keeps at most snap_keep_n files */ +/* ------------------------------------------------------------------ */ + +static void test_snapshot_rotation_keeps_snap_keep_n(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_e2e(tmpdir, sizeof(tmpdir), 2); + TEST_ASSERT_NOT_NULL(p); + + /* Run 4 checkpoints with snap_keep_n = 2 */ + for (int i = 0; i < 4; i++) { + persist_wal_async(p, WAL_JOB_UPDATE, "r", 1); + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + } + + char snap_path[512]; + snprintf(snap_path, sizeof(snap_path), "%s/e2e-ckpt/data/snap", tmpdir); + int n = count_files_suffix(snap_path, ".capnp"); + TEST_ASSERT_TRUE(n <= (int)p->config.snap_keep_n); + + teardown_ckpt_e2e(p); + printf("test_snapshot_rotation_keeps_snap_keep_n: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Suite entry point */ +/* ------------------------------------------------------------------ */ + +void test_e2e_checkpoint(void) +{ + printf("\n>> P0-15b: Checkpoint E2E — round trip\n\n"); + + RUN_TEST(test_checkpoint_recovery_lsn_continuity); + RUN_TEST(test_checkpoint_wal_trim_after_checkpoint); + RUN_TEST(test_snapshot_rotation_keeps_snap_keep_n); +} diff --git a/src/controller/testsuite/test_e2e_gc.c b/src/controller/testsuite/test_e2e_gc.c new file mode 100644 index 0000000..c89c039 --- /dev/null +++ b/src/controller/testsuite/test_e2e_gc.c @@ -0,0 +1,167 @@ +/* P0-15d: GC E2E — retention + transient job cleanup + * + * Exercises cross-phase interactions: + * - GC retention predicates (P0-E) applied to concrete job/worker fixtures + * - JOB_STAT_TRANSIENT timeout behaviour + * - Checkpoint succeeds after WAL writes (P0-C integration smoke) + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "config.h" +#include "persist.h" +#include "xce/types.h" + +/* ------------------------------------------------------------------ */ +/* Fixtures */ +/* ------------------------------------------------------------------ */ + +static job_t make_gc_job(unsigned int state, time_t finish_time, + uint64_t transient_since) +{ + job_t j; + memset(&j, 0, sizeof(j)); + j.job.state = state; + j.job.finish_time = finish_time; + j.job.transient_since = (uint64_t)transient_since; + return j; +} + +static xce_worker_t make_gc_worker(uint32_t status, time_t dead_since) +{ + xce_worker_t w; + memset(&w, 0, sizeof(w)); + w.status = status; + w.dead_since = dead_since; + return w; +} + +/* ------------------------------------------------------------------ */ +/* Test 1: running job must be retained */ +/* ------------------------------------------------------------------ */ + +static void test_job_gc_retain_running_job(void) +{ + persist_config_t cfg = persist_config_defaults(); + job_t j = make_gc_job(JOB_STAT_RUN, 0, 0); + TEST_ASSERT_TRUE(job_should_retain(&j, &cfg)); + printf("test_job_gc_retain_running_job: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 2: expired finished job must not be retained */ +/* ------------------------------------------------------------------ */ + +static void test_job_gc_cleanup_expired_finished_job(void) +{ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.finished_retain_s = 3600; + job_t j = make_gc_job(JOB_STAT_EXIT, now - 7200, 0); + TEST_ASSERT_FALSE(job_should_retain(&j, &cfg)); + printf("test_job_gc_cleanup_expired_finished_job: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 3: transient job times out */ +/* ------------------------------------------------------------------ */ + +static void test_transient_job_times_out(void) +{ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.transient_job_timeout_s = 300; + /* transient_since = now - 400 seconds — past the 300s window */ + job_t j = make_gc_job(JOB_STAT_TRANSIENT, 0, (uint64_t)(now - 400)); + TEST_ASSERT_FALSE(job_should_retain(&j, &cfg)); + printf("test_transient_job_times_out: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 4: expired unavailable worker must not be retained */ +/* ------------------------------------------------------------------ */ + +static void test_worker_gc_cleanup_expired_unavailable(void) +{ + time_t now = time(NULL); + persist_config_t cfg = persist_config_defaults(); + cfg.worker_retain_s = 86400; /* 24 h */ + xce_worker_t w = make_gc_worker(XCE_WORKER_UNAVAIL, now - 100000); + TEST_ASSERT_FALSE(worker_should_retain(&w, &cfg)); + printf("test_worker_gc_cleanup_expired_unavailable: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 5: checkpoint succeeds and creates snapshot (integration smoke)*/ +/* ------------------------------------------------------------------ */ + +static void test_checkpoint_gc_removes_expired_entity(void) +{ + char tmpdir[256]; + strcpy(tmpdir, "/tmp/xce_e2e_gc_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); + + persist_t* p = calloc(1, sizeof(persist_t)); + TEST_ASSERT_NOT_NULL(p); + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.snap_keep_n = 2; + + persist_set_workdir(p, tmpdir, "e2e-gc"); + TEST_ASSERT_EQUAL_INT(0, persist_init(p, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(p)); + + /* Write some WAL entries (representing state changes) */ + for (int i = 0; i < 5; i++) + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_UPDATE, "gc-test", 7)); + + /* Trigger checkpoint — must succeed and leave a snapshot */ + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + TEST_ASSERT_TRUE(p->checkpoint_lsn > 0); + + char snap_path[512]; + snprintf(snap_path, sizeof(snap_path), "%s/e2e-gc/data/snap", tmpdir); + + DIR* d = opendir(snap_path); + TEST_ASSERT_NOT_NULL(d); + int found = 0; + struct dirent* e; + while ((e = readdir(d)) != NULL) { + size_t len = strlen(e->d_name); + if (len > 6 && strcmp(e->d_name + len - 6, ".capnp") == 0) + found++; + } + closedir(d); + TEST_ASSERT_TRUE(found >= 1); + + persist_stop(p); + persist_fini(p); + free(p); + printf("test_checkpoint_gc_removes_expired_entity: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Suite entry point */ +/* ------------------------------------------------------------------ */ + +void test_e2e_gc(void) +{ + printf("\n>> P0-15d: GC E2E — retention + transient job cleanup\n\n"); + + RUN_TEST(test_job_gc_retain_running_job); + RUN_TEST(test_job_gc_cleanup_expired_finished_job); + RUN_TEST(test_transient_job_times_out); + RUN_TEST(test_worker_gc_cleanup_expired_unavailable); + RUN_TEST(test_checkpoint_gc_removes_expired_entity); +} diff --git a/src/controller/testsuite/test_e2e_hooks.c b/src/controller/testsuite/test_e2e_hooks.c new file mode 100644 index 0000000..94cb56c --- /dev/null +++ b/src/controller/testsuite/test_e2e_hooks.c @@ -0,0 +1,213 @@ +/* P0-15c: Hook E2E — ordering and WAL write + * + * Exercises cross-phase interactions: + * - Hook functions (P0-D) write WAL entries (P0-B) + * - LSN advances monotonically for each hook call + * - WAL segment contains entries with correct type field + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "config.h" +#include "persist.h" +#include "xce/cluster_def.h" + +/* ------------------------------------------------------------------ */ +/* Helpers */ +/* ------------------------------------------------------------------ */ + +static char g_e2e_hooks_tmpdir[256]; + +static persist_t* make_e2e_hook_persist(void) +{ + strcpy(g_e2e_hooks_tmpdir, "/tmp/xce_e2e_hook_XXXXXX"); + if (!mkdtemp(g_e2e_hooks_tmpdir)) return NULL; + + persist_t* p = calloc(1, sizeof(persist_t)); + if (!p) return NULL; + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 1; /* flush quickly */ + + persist_set_workdir(p, g_e2e_hooks_tmpdir, "e2e-hook"); + + if (persist_init(p, &cfg, NULL) != 0) { free(p); return NULL; } + if (persist_start(p) != 0) { persist_fini(p); free(p); return NULL; } + return p; +} + +static void teardown_e2e_hook(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* Find the first .wal segment in a WAL directory. */ +static int find_first_seg(const char* wal_dir, char* out, size_t outsz) +{ + DIR* d = opendir(wal_dir); + if (!d) return -1; + struct dirent* e; + while ((e = readdir(d)) != NULL) { + size_t len = strlen(e->d_name); + if (len > 4 && strcmp(e->d_name + len - 4, ".wal") == 0) { + snprintf(out, outsz, "%s/%s", wal_dir, e->d_name); + closedir(d); + return 0; + } + } + closedir(d); + return -1; +} + +/* Read the Nth (0-indexed) wal_entry_hdr_t from a segment file (sequential + * layout: hdr + payload). Returns 0 on success. */ +static int read_nth_hdr(const char* seg_path, int n, wal_entry_hdr_t* out) +{ + int fd = open(seg_path, O_RDONLY); + if (fd < 0) return -1; + + off_t offset = 0; + for (int i = 0; i <= n; i++) { + wal_entry_hdr_t hdr; + ssize_t rd = pread(fd, &hdr, sizeof(hdr), offset); + if (rd != (ssize_t)sizeof(hdr)) { close(fd); return -1; } + if (i == n) { *out = hdr; close(fd); return 0; } + offset += (off_t)(sizeof(hdr) + hdr.payload_len); + } + close(fd); + return -1; +} + +/* ------------------------------------------------------------------ */ +/* Test 1: job hook writes WAL before LSN advances */ +/* ------------------------------------------------------------------ */ + +static void test_job_hook_writes_wal_before_lsn_advances(void) +{ + persist_t* p = make_e2e_hook_persist(); + TEST_ASSERT_NOT_NULL(p); + + uint64_t lsn_before = p->current_lsn; + + int rc = persist_job_hook_fun(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_NEW, p, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* LSN must have advanced by exactly 1 */ + TEST_ASSERT_EQUAL_UINT64(lsn_before + 1, p->current_lsn); + + /* Flush and verify the entry type in the segment file */ + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + char seg_path[512]; + TEST_ASSERT_EQUAL_INT(0, find_first_seg(p->wal_path, seg_path, sizeof(seg_path))); + + wal_entry_hdr_t hdr; + TEST_ASSERT_EQUAL_INT(0, read_nth_hdr(seg_path, 0, &hdr)); + TEST_ASSERT_EQUAL_UINT32(WAL_JOB_SUBMIT, hdr.type); + + teardown_e2e_hook(p); + printf("test_job_hook_writes_wal_before_lsn_advances: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 2: worker hook writes WAL_WORKER_CTRL async */ +/* ------------------------------------------------------------------ */ + +static void test_worker_hook_writes_wal_async(void) +{ + persist_t* p = make_e2e_hook_persist(); + TEST_ASSERT_NOT_NULL(p); + + uint64_t lsn_before = p->current_lsn; + + int rc = persist_worker_hook_fun(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, + HOOK_DETAIL_WORKER_JOIN, p, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT64(lsn_before + 1, p->current_lsn); + + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + char seg_path[512]; + TEST_ASSERT_EQUAL_INT(0, find_first_seg(p->wal_path, seg_path, sizeof(seg_path))); + + wal_entry_hdr_t hdr; + TEST_ASSERT_EQUAL_INT(0, read_nth_hdr(seg_path, 0, &hdr)); + TEST_ASSERT_EQUAL_UINT32(WAL_WORKER_CTRL, hdr.type); + + teardown_e2e_hook(p); + printf("test_worker_hook_writes_wal_async: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 3: hook sequence preserves monotone LSN order */ +/* ------------------------------------------------------------------ */ + +static void test_hook_sequence_preserves_lsn_order(void) +{ + persist_t* p = make_e2e_hook_persist(); + TEST_ASSERT_NOT_NULL(p); + + uint64_t lsn0 = p->current_lsn; + + persist_job_hook_fun(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_NEW, p, NULL); + uint64_t lsn1 = p->current_lsn; + + persist_job_hook_fun(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_START, p, NULL); + uint64_t lsn2 = p->current_lsn; + + persist_job_hook_fun(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_EXIT, p, NULL); + uint64_t lsn3 = p->current_lsn; + + /* LSNs must be strictly increasing */ + TEST_ASSERT_TRUE(lsn1 > lsn0); + TEST_ASSERT_TRUE(lsn2 > lsn1); + TEST_ASSERT_TRUE(lsn3 > lsn2); + TEST_ASSERT_EQUAL_UINT64(lsn0 + 3, lsn3); + + /* Flush and verify three entries on disk in order */ + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + char seg_path[512]; + TEST_ASSERT_EQUAL_INT(0, find_first_seg(p->wal_path, seg_path, sizeof(seg_path))); + + wal_entry_hdr_t h0, h1, h2; + TEST_ASSERT_EQUAL_INT(0, read_nth_hdr(seg_path, 0, &h0)); + TEST_ASSERT_EQUAL_INT(0, read_nth_hdr(seg_path, 1, &h1)); + TEST_ASSERT_EQUAL_INT(0, read_nth_hdr(seg_path, 2, &h2)); + + /* LSNs in the headers must also be strictly increasing */ + TEST_ASSERT_TRUE(h1.lsn > h0.lsn); + TEST_ASSERT_TRUE(h2.lsn > h1.lsn); + + teardown_e2e_hook(p); + printf("test_hook_sequence_preserves_lsn_order: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Suite entry point */ +/* ------------------------------------------------------------------ */ + +void test_e2e_hooks(void) +{ + printf("\n>> P0-15c: Hook E2E — ordering and WAL write\n\n"); + + RUN_TEST(test_job_hook_writes_wal_before_lsn_advances); + RUN_TEST(test_worker_hook_writes_wal_async); + RUN_TEST(test_hook_sequence_preserves_lsn_order); +} diff --git a/src/controller/testsuite/test_e2e_wal.c b/src/controller/testsuite/test_e2e_wal.c new file mode 100644 index 0000000..2d71b1f --- /dev/null +++ b/src/controller/testsuite/test_e2e_wal.c @@ -0,0 +1,269 @@ +/* P0-15a: WAL integrity end-to-end smoke tests + * + * Exercises cross-phase interactions: + * - WAL writer (P0-B) produces segment files that can be read back + * - Torn-write simulation (ftruncate) + persist_recover() (P0-C) + * - Chained-CRC corruption detection during recovery replay + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "config.h" +#include "persist.h" + +/* ------------------------------------------------------------------ */ +/* Helpers */ +/* ------------------------------------------------------------------ */ + +static char g_e2e_wal_tmpdir[256]; + +static persist_t* make_e2e_wal_persist(void) +{ + strcpy(g_e2e_wal_tmpdir, "/tmp/xce_e2e_wal_XXXXXX"); + if (!mkdtemp(g_e2e_wal_tmpdir)) return NULL; + + persist_t* p = calloc(1, sizeof(persist_t)); + if (!p) return NULL; + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; /* suppress background fsync races */ + + persist_set_workdir(p, g_e2e_wal_tmpdir, "e2e-wal"); + + if (persist_init(p, &cfg, NULL) != 0) { free(p); return NULL; } + if (persist_start(p) != 0) { persist_fini(p); free(p); return NULL; } + return p; +} + +static void teardown_e2e_wal(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* Return the path to the first .wal segment in the WAL directory. */ +static int find_first_wal_segment(const char* wal_dir, char* out, size_t outsz) +{ + DIR* d = opendir(wal_dir); + if (!d) return -1; + + struct dirent* e; + while ((e = readdir(d)) != NULL) { + size_t len = strlen(e->d_name); + if (len > 4 && strcmp(e->d_name + len - 4, ".wal") == 0) { + snprintf(out, outsz, "%s/%s", wal_dir, e->d_name); + closedir(d); + return 0; + } + } + closedir(d); + return -1; +} + +/* Count valid wal_entry_hdr_t records in a segment file by reading + * sequentially and checking header CRC (first 32 bytes of header). */ +static int count_valid_wal_entries(const char* seg_path) +{ + int fd = open(seg_path, O_RDONLY); + if (fd < 0) return -1; + + int count = 0; + off_t offset = 0; + + for (;;) { + wal_entry_hdr_t hdr; + ssize_t n = pread(fd, &hdr, sizeof(hdr), offset); + if (n != (ssize_t)sizeof(hdr)) break; + + /* Verify header CRC (first WAL_HDR_CRC_SIZE bytes) */ + uint32_t expected_crc = (uint32_t)crc32(0L, + (const unsigned char*)&hdr, WAL_HDR_CRC_SIZE); + if (hdr.payload_len > 0) { + unsigned char* payload = malloc(hdr.payload_len); + if (!payload) break; + ssize_t pn = pread(fd, payload, hdr.payload_len, + offset + (off_t)sizeof(hdr)); + if (pn == (ssize_t)hdr.payload_len) + expected_crc = (uint32_t)crc32(expected_crc, + payload, hdr.payload_len); + free(payload); + } + + if (expected_crc != hdr.crc) break; + + count++; + offset += (off_t)(sizeof(hdr) + hdr.payload_len); + } + + close(fd); + return count; +} + +/* ------------------------------------------------------------------ */ +/* Test 1: write 10 WAL entries and verify they appear in segment file */ +/* ------------------------------------------------------------------ */ + +static void test_wal_write_and_verify_entry_count(void) +{ + persist_t* p = make_e2e_wal_persist(); + TEST_ASSERT_NOT_NULL(p); + + const char payload[] = "e2e-entry"; + for (int i = 0; i < 10; i++) { + int rc = persist_wal_async(p, WAL_JOB_UPDATE, payload, + (uint32_t)strlen(payload)); + TEST_ASSERT_EQUAL_INT(0, rc); + } + + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + char seg_path[512]; + TEST_ASSERT_EQUAL_INT(0, + find_first_wal_segment(p->wal_path, seg_path, sizeof(seg_path))); + + int count = count_valid_wal_entries(seg_path); + TEST_ASSERT_EQUAL_INT(10, count); + + teardown_e2e_wal(p); + printf("test_wal_write_and_verify_entry_count: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 2: torn-write recovery — truncate last entry, recover, check */ +/* ------------------------------------------------------------------ */ + +static void test_wal_torn_write_recovery(void) +{ + /* Phase 1: write 5 entries and flush */ + persist_t* p = make_e2e_wal_persist(); + TEST_ASSERT_NOT_NULL(p); + + const char payload[] = "torn"; + for (int i = 0; i < 5; i++) { + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_UPDATE, payload, + (uint32_t)strlen(payload))); + } + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + /* Locate segment file and determine size of 4 complete entries */ + char seg_path[512]; + TEST_ASSERT_EQUAL_INT(0, + find_first_wal_segment(p->wal_path, seg_path, sizeof(seg_path))); + + /* One entry = sizeof(wal_entry_hdr_t) + payload_len */ + off_t entry_size = (off_t)(sizeof(wal_entry_hdr_t) + strlen(payload)); + off_t trunc_offset = entry_size * 4; /* keep only 4 complete entries */ + + /* Truncate: simulate torn write on entry 5 */ + int fd = open(seg_path, O_RDWR); + TEST_ASSERT_TRUE(fd >= 0); + TEST_ASSERT_EQUAL_INT(0, ftruncate(fd, trunc_offset)); + close(fd); + + uint64_t lsn_before = p->current_lsn; + + teardown_e2e_wal(p); /* stop / destroy the first instance */ + + /* Phase 2: fresh persist_t pointing at the same directory */ + persist_t* p2 = calloc(1, sizeof(persist_t)); + TEST_ASSERT_NOT_NULL(p2); + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + persist_set_workdir(p2, g_e2e_wal_tmpdir, "e2e-wal"); + + TEST_ASSERT_EQUAL_INT(0, persist_init(p2, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(p2)); + + int rc = persist_recover(p2); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Recovery must have replayed at most 4 entries (not 5 — torn write) */ + TEST_ASSERT_TRUE(p2->current_lsn < lsn_before); + + teardown_e2e_wal(p2); + printf("test_wal_torn_write_recovery: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 3: chained CRC corruption stops replay at last valid entry */ +/* ------------------------------------------------------------------ */ + +static void test_wal_chained_crc_detects_corruption(void) +{ + /* Phase 1: write 5 entries and flush */ + persist_t* p = make_e2e_wal_persist(); + TEST_ASSERT_NOT_NULL(p); + + const char payload[] = "crc-chain"; + for (int i = 0; i < 5; i++) { + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_UPDATE, payload, + (uint32_t)strlen(payload))); + } + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + char seg_path[512]; + TEST_ASSERT_EQUAL_INT(0, + find_first_wal_segment(p->wal_path, seg_path, sizeof(seg_path))); + + /* Corrupt payload of entry 3 (zero-indexed: entries 0,1 valid; entry 2 + * corrupted; entries 3,4 unreachable via CRC chain). */ + off_t entry_size = (off_t)(sizeof(wal_entry_hdr_t) + strlen(payload)); + off_t corrupt_offset = entry_size * 2 + (off_t)sizeof(wal_entry_hdr_t); + /* Write garbage over the payload of entry 2 */ + int fd = open(seg_path, O_RDWR); + TEST_ASSERT_TRUE(fd >= 0); + const char garbage[] = "\xFF\xFF\xFF\xFF"; + pwrite(fd, garbage, sizeof(garbage) - 1, corrupt_offset); + close(fd); + + uint64_t lsn_before = p->current_lsn; + teardown_e2e_wal(p); + + /* Phase 2: fresh persist_t — recover must stop before the corrupt entry */ + persist_t* p2 = calloc(1, sizeof(persist_t)); + TEST_ASSERT_NOT_NULL(p2); + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + persist_set_workdir(p2, g_e2e_wal_tmpdir, "e2e-wal"); + + TEST_ASSERT_EQUAL_INT(0, persist_init(p2, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(p2)); + + int rc = persist_recover(p2); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Must have replayed fewer than 5 entries — stopped at corruption */ + TEST_ASSERT_TRUE(p2->current_lsn < lsn_before); + + teardown_e2e_wal(p2); + printf("test_wal_chained_crc_detects_corruption: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Suite entry point */ +/* ------------------------------------------------------------------ */ + +void test_e2e_wal(void) +{ + printf("\n>> P0-15a: WAL E2E — torn write + CRC corruption recovery\n\n"); + + RUN_TEST(test_wal_write_and_verify_entry_count); + RUN_TEST(test_wal_torn_write_recovery); + RUN_TEST(test_wal_chained_crc_detects_corruption); +}