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

test: P0-G end-to-end smoke tests — WAL, checkpoint, hooks, GC, Cap'n Proto (P0-15a-e)

Browse source 
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).
This commit is contained in:
Rongsong Shen 2026-05-07 13:25:17 +00:00
parent b52fbef36d
commit a7094fc47f
7 changed files with 1002 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/controller/meson.build
View file

@ -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',
]

10
src/controller/testsuite/controller_test.c
View file

@ -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();
}

146
src/controller/testsuite/test_e2e_capnp.c Normal file
View file

@ -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 <stdint.h>
#include <stdio.h>
#include <string.h>
#include <unity/unity.h>
#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);
}

192
src/controller/testsuite/test_e2e_checkpoint.c Normal file
View file

@ -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 <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>
#include <unity/unity.h>
#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);
}

167
src/controller/testsuite/test_e2e_gc.c Normal file
View file

@ -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 <dirent.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
#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);
}

213
src/controller/testsuite/test_e2e_hooks.c Normal file
View file

@ -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 <dirent.h>
#include <fcntl.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>
#include <zlib.h>
#include <unity/unity.h>
#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);
}

269
src/controller/testsuite/test_e2e_wal.c Normal file
View file

@ -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 <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>
#include <zlib.h>
#include <unity/unity.h>
#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);
}