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

feat: implement persist_recover with snapshot load and WAL replay (P0-4)

Browse source 
- Add load_snapshot(): reads P0 sentinel ("XCESNAP\0" + lsn), validates magic,
  extracts checkpoint_lsn
- Add replay_wal(): scans seg-*.wal files in LSN order, skips entries <= from_lsn,
  verifies CRC chain, stops on mismatch, applies entries via apply_wal_entry() stub
- Add persist_recover(): find latest .capnp snapshot, load it, replay post-checkpoint
  WAL, retention pass (no-op in P0)
- Fix trim_wal_segments(): compare against active segment start LSN (not checkpoint_lsn)
  to avoid deleting the currently-open segment and losing post-checkpoint WAL entries
- Add restore_job() stub in job.c (P0-12a placeholder)
- Add worker_from_capnp() stub in worker.c (P0-12 placeholder)
- Declare both new functions in cluster.h and worker.h
- Add 5 recovery tests: empty state, snapshot lsn loading, WAL replay, corrupt
  sentinel rejection, latest-snapshot selection; all 23 controller tests pass
This commit is contained in:
Rongsong Shen 2026-05-07 11:41:33 +00:00
parent e76f76a88a
commit f905f891a8
8 changed files with 584 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/controller/include/cluster.h
View file

@ -39,6 +39,7 @@ extern int add_resource(void*, char*, char*, int, xce_order_t, int, int);
extern int update_resvals(cluster_t*, worker_t*, int, xce_resval_t*);
extern int load_jobs(void*);
extern int restore_job(cluster_t*, uint64_t, const void*, size_t);
extern int update_job(void*, uint64_t, xce_kv_list_t*);
extern int do_job_started(void*, uint64_t, int, int*, int, int*);

1
src/controller/include/worker.h
View file

@ -9,4 +9,5 @@
extern int add_worker(void*, endpoint_t*, const char*, hostattrs_t*);
extern int restore_worker(void*, xce_worker_t);
extern int worker_from_capnp(cluster_t*, const void*, size_t);
#endif

12
src/controller/job.c
View file

@ -61,6 +61,18 @@ int load_jobs(void* c)
return 0;
}
/* P0 stub — full deserialization wired in Task 15 (cluster wiring).
* Inserts a job at its original jobid without allocating a new one.
* Returns -EEXIST if jobid already present (safety guard for replay). */
int restore_job(cluster_t* c, uint64_t jobid, const void* capnp_payload, size_t len)
{
(void)c;
(void)jobid;
(void)capnp_payload;
(void)len;
return 0;
}
static int get_nslots_in_alloc(xce_host_alloc_t* alloc)
{
int i;

1
src/controller/meson.build
View file

@ -72,6 +72,7 @@ controller_unit_test_src = [
'testsuite/test_persist_wal.c',
'testsuite/test_persist_schema.c',
'testsuite/test_persist_checkpoint.c',
'testsuite/test_persist_recovery.c',
'persist.c',
]

304
src/controller/persist.c
View file

@ -359,10 +359,26 @@ static int write_snapshot(persist_t* p)
return 0;
}
/* Delete WAL segment files whose start LSN is strictly less than trim_lsn.
* Conservative: only trims segments that started before the checkpoint. */
/* Delete WAL segment files whose start LSN is strictly less than the
* currently-open segment's start LSN. We parse the active segment
* filename (seg_path) to get its start LSN and never delete that file,
* even if its start_lsn < checkpoint_lsn. */
static void trim_wal_segments(persist_t* p)
{
/* Extract the start LSN of the currently-open segment from seg_path.
* seg_path has the form "<wal_path>/seg-<hex16>.wal".
* If no segment is open yet (wal_fd < 0), skip trimming. */
if (p->wal_fd < 0 || p->seg_path[0] == '\0')
return;
const char* seg_basename = strrchr(p->seg_path, '/');
if (!seg_basename) return;
seg_basename++; /* skip the '/' */
uint64_t active_seg_lsn = 0;
if (sscanf(seg_basename, "seg-%" SCNx64 ".wal", &active_seg_lsn) != 1)
return;
DIR* d = opendir(p->wal_path);
if (!d) return;
@ -371,9 +387,8 @@ static void trim_wal_segments(persist_t* p)
if (strncmp(e->d_name, "seg-", 4) != 0) continue;
uint64_t seg_start_lsn = 0;
if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &seg_start_lsn) != 1) continue;
/* Only trim segments that start strictly before the checkpoint LSN.
* The segment containing checkpoint_lsn is still the active segment. */
if (seg_start_lsn < p->checkpoint_lsn) {
/* Never delete the currently-open segment; only trim older ones. */
if (seg_start_lsn < active_seg_lsn) {
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path), "%s/%s", p->wal_path, e->d_name);
unlink(seg_path);
@ -466,12 +481,287 @@ done:
}
/* ------------------------------------------------------------------ */
/* Stubs — wired in later tasks */
/* Recovery internals */
/* ------------------------------------------------------------------ */
/* P0 sentinel format: "XCESNAP\0" (8 bytes) + checkpoint_lsn (8 bytes LE).
* Returns 0 on success and sets p->checkpoint_lsn / p->current_lsn.
* Returns -EINVAL on bad magic, -EIO on short read, -errno on open failure. */
static int load_snapshot(persist_t* p, const char* path, uint64_t lsn)
{
(void)lsn; /* lsn is encoded in the file itself — we re-read it */
int fd = open(path, O_RDONLY);
if (fd < 0)
return -errno;
char header[16];
ssize_t n = read(fd, header, sizeof(header));
close(fd);
if (n < 16)
return -EIO;
if (memcmp(header, "XCESNAP\0", 8) != 0)
return -EINVAL;
/* Extract checkpoint_lsn from bytes 8-15 (little-endian on this arch) */
uint64_t stored_lsn;
memcpy(&stored_lsn, header + 8, 8);
p->checkpoint_lsn = stored_lsn;
p->current_lsn = stored_lsn;
return 0;
}
/* P0 stub: no job/worker deserialization yet — wired in Task 12+/15. */
static int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr,
const void* payload)
{
(void)p;
(void)hdr;
(void)payload;
/* Full implementation in Task 12+ (hook wiring) and Task 15 (cluster wiring) */
return 0;
}
/* Read WAL segment files in LSN order; replay entries with lsn > from_lsn.
*
* Segment filename format: seg-<start_lsn_hex16>.wal
* Entry format: wal_entry_hdr_t (40 bytes) + payload (hdr.payload_len bytes).
*
* CRC chain: hdr.prev_crc must match last_crc (WAL_CRC_NONE for first entry
* in a segment). If prev_crc mismatches, stop at last good entry and log WARN.
*
* Returns number of entries replayed (>= 0) or negative errno on I/O error. */
static int replay_wal(persist_t* p, uint64_t from_lsn)
{
/* --- Collect and sort segment filenames by start LSN --- */
DIR* d = opendir(p->wal_path);
if (!d)
return 0; /* no WAL dir → first boot, nothing to replay */
/* Collect up to 4096 segment names */
uint64_t seg_lsns[4096];
int nseg = 0;
struct dirent* e;
while ((e = readdir(d)) != NULL && nseg < (int)(sizeof(seg_lsns)/sizeof(seg_lsns[0]))) {
if (strncmp(e->d_name, "seg-", 4) != 0) continue;
uint64_t seg_start = 0;
if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &seg_start) != 1) continue;
seg_lsns[nseg++] = seg_start;
}
closedir(d);
if (nseg == 0)
return 0;
/* Simple insertion sort — segment count is tiny in practice */
for (int i = 1; i < nseg; i++) {
uint64_t key = seg_lsns[i];
int j = i - 1;
while (j >= 0 && seg_lsns[j] > key) {
seg_lsns[j + 1] = seg_lsns[j];
j--;
}
seg_lsns[j + 1] = key;
}
/* --- Replay entries from each segment --- */
int replayed = 0;
uint32_t last_crc = WAL_CRC_NONE;
for (int si = 0; si < nseg; si++) {
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path),
WAL_SEG_FMT, p->wal_path, seg_lsns[si]);
int fd = open(seg_path, O_RDONLY);
if (fd < 0) {
fprintf(stderr,
"persist_recover: WARNING: cannot open segment %s: %s\n",
seg_path, strerror(errno));
continue;
}
uint32_t seg_off = 0;
for (;;) {
/* Read header */
wal_entry_hdr_t hdr;
ssize_t hn = pread(fd, &hdr, sizeof(hdr), (off_t)seg_off);
if (hn == 0)
break; /* clean end of segment */
if (hn < (ssize_t)sizeof(hdr)) {
if (hn < 0) {
fprintf(stderr,
"persist_recover: WARNING: read error in %s at offset %u: %s\n",
seg_path, seg_off, strerror(errno));
}
break; /* truncated or pre-allocated zeros */
}
/* Detect pre-allocated zero region (no valid entry starts with
* all-zero header: lsn == 0 && type == 0 && payload_len == 0) */
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0)
break;
/* CRC chain validation: prev_crc must match last_crc in segment */
if (seg_off == 0) {
/* First entry in this segment: prev_crc must be WAL_CRC_NONE
* (or we accept any value if last_crc is also WAL_CRC_NONE,
* e.g. across a segment boundary after checkpoint trim). */
if (hdr.prev_crc != WAL_CRC_NONE && last_crc != WAL_CRC_NONE &&
hdr.prev_crc != last_crc) {
fprintf(stderr,
"persist_recover: WARNING: CRC chain break at start of %s "
"(prev_crc=%08x expected=%08x); stopping replay\n",
seg_path, hdr.prev_crc, last_crc);
break;
}
} else {
if (hdr.prev_crc != last_crc) {
fprintf(stderr,
"persist_recover: WARNING: CRC chain break in %s at offset %u "
"(prev_crc=%08x expected=%08x); stopping replay at last good entry\n",
seg_path, seg_off, hdr.prev_crc, last_crc);
break;
}
}
/* Verify entry CRC */
uint32_t expected_crc = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE);
uint8_t* payload = NULL;
if (hdr.payload_len > 0) {
payload = malloc(hdr.payload_len);
if (!payload) {
close(fd);
return -ENOMEM;
}
ssize_t pn = pread(fd, payload, hdr.payload_len,
(off_t)(seg_off + sizeof(hdr)));
if (pn != (ssize_t)hdr.payload_len) {
fprintf(stderr,
"persist_recover: WARNING: short payload read in %s at offset %u\n",
seg_path, seg_off);
free(payload);
break;
}
expected_crc = crc32(expected_crc, (const Bytef*)payload, hdr.payload_len);
}
if (hdr.crc != expected_crc) {
fprintf(stderr,
"persist_recover: WARNING: CRC mismatch in %s at offset %u "
"(got=%08x expected=%08x); stopping replay\n",
seg_path, seg_off, hdr.crc, expected_crc);
free(payload);
break;
}
/* Advance segment-local CRC */
last_crc = hdr.crc;
/* Skip entries already covered by the snapshot */
if (hdr.lsn <= from_lsn) {
free(payload);
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
continue;
}
/* Skip WAL_CHECKPOINT markers (local bookkeeping only) */
if (hdr.type == WAL_CHECKPOINT) {
free(payload);
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
/* Advance current_lsn past the marker */
if (hdr.lsn + 1 > p->current_lsn)
p->current_lsn = hdr.lsn + 1;
continue;
}
/* Unknown future entry types: log and skip */
if (hdr.type > WAL_CHECKPOINT) {
fprintf(stderr,
"persist_recover: WARNING: unknown WAL entry type %u at LSN %lu; skipping\n",
hdr.type, (unsigned long)hdr.lsn);
free(payload);
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
continue;
}
/* Apply the entry (P0 stub: no-op) */
apply_wal_entry(p, &hdr, payload);
free(payload);
/* Advance current_lsn to one past this entry */
if (hdr.lsn + 1 > p->current_lsn)
p->current_lsn = hdr.lsn + 1;
replayed++;
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(fd);
}
/* Update last_crc so the next WAL write chains correctly */
pthread_mutex_lock(&p->lock);
p->last_crc = last_crc;
pthread_mutex_unlock(&p->lock);
return replayed;
}
/* ------------------------------------------------------------------ */
/* Public: persist_recover */
/* ------------------------------------------------------------------ */
int persist_recover(persist_t* p)
{
(void)p;
/* --- Step 1: find latest snapshot (highest LSN filename) --- */
char snap_file[4096] = "";
uint64_t snap_lsn = 0;
DIR* d = opendir(p->snap_path);
if (d) {
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)
continue;
uint64_t lsn = 0;
if (sscanf(e->d_name, "snap-%" SCNx64 ".capnp", &lsn) != 1)
continue;
if (lsn > snap_lsn) {
snap_lsn = lsn;
snprintf(snap_file, sizeof(snap_file),
"%s/%s", p->snap_path, e->d_name);
}
}
closedir(d);
}
/* --- Step 2: load snapshot (if any) --- */
if (snap_lsn > 0 && snap_file[0] != '\0') {
int rc = load_snapshot(p, snap_file, snap_lsn);
if (rc != 0) {
fprintf(stderr,
"persist_recover: failed to load snapshot %s: %s\n",
snap_file, strerror(-rc));
return rc;
}
/* checkpoint_lsn and current_lsn set by load_snapshot */
}
/* --- Step 3: replay WAL entries after checkpoint_lsn --- */
int replayed = replay_wal(p, p->checkpoint_lsn);
if (replayed < 0)
return replayed;
/* --- Step 4: retention pass (TTL check) — no-op in P0 (no entities) --- */
return 0;
}

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

@ -9,6 +9,7 @@ extern void test_persist_types(void);
extern void test_persist_wal(void);
extern void test_persist_schema(void);
extern void test_persist_checkpoint(void);
extern void test_persist_recovery(void);
void setUp(void) { return; }
@ -87,6 +88,7 @@ int main(int argc, char* argv[])
TEST(test_persist_wal);
TEST(test_persist_schema);
TEST(test_persist_checkpoint);
TEST(test_persist_recovery);
return UNITY_END();
}

258
src/controller/testsuite/test_persist_recovery.c Normal file
View file

@ -0,0 +1,258 @@
#define _GNU_SOURCE
#include <assert.h>
#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 persist_t* make_fresh_persist(const char* tmpdir)
{
persist_t* p = calloc(1, sizeof(persist_t));
if (!p) return NULL;
persist_config_t cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000; /* slow fsync — no races in tests */
cfg.snap_keep_n = 2;
persist_set_workdir(p, tmpdir, "ctrl-test");
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_persist(persist_t* p)
{
persist_stop(p);
persist_fini(p);
free(p);
}
/* ------------------------------------------------------------------ */
/* Test 1: recover from empty state (no snapshot, no WAL) */
/* ------------------------------------------------------------------ */
static void test_recover_from_empty_state(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_rec0_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
persist_t* p = make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
/* Stop so we can simulate a "fresh boot" re-init — actually
* persist_recover() is valid to call on a running instance too. */
int rc = persist_recover(p);
TEST_ASSERT_EQUAL_INT(0, rc);
/* No snapshot found — checkpoint_lsn stays at 0 */
TEST_ASSERT_EQUAL_UINT64(0, p->checkpoint_lsn);
teardown_persist(p);
printf("test_recover_from_empty_state: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 2: recover loads checkpoint_lsn from the snapshot sentinel */
/* ------------------------------------------------------------------ */
static void test_recover_loads_checkpoint_lsn(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_rec1_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Phase 1: write some WAL entries + take a checkpoint */
{
persist_t* p = make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "x", 1));
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "y", 1));
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p));
teardown_persist(p);
}
/* Phase 2: fresh persist_t, call recover — must find the snapshot */
{
persist_t* p2 = make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
int rc = persist_recover(p2);
TEST_ASSERT_EQUAL_INT(0, rc);
/* checkpoint_lsn must have been restored from the sentinel header */
TEST_ASSERT_TRUE(p2->checkpoint_lsn > 0);
teardown_persist(p2);
}
printf("test_recover_loads_checkpoint_lsn: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 3: recover replays WAL entries written after the checkpoint */
/* ------------------------------------------------------------------ */
static void test_recover_replays_wal_after_checkpoint(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_rec2_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Phase 1: checkpoint, then write 5 more WAL entries */
uint64_t last_lsn_before_stop = 0;
{
persist_t* p = make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); /* clean checkpoint */
for (int i = 0; i < 5; i++)
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_UPDATE, "z", 1));
last_lsn_before_stop = p->current_lsn;
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
teardown_persist(p);
}
/* Phase 2: recover — current_lsn must advance past the replayed entries */
{
persist_t* p2 = make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
int rc = persist_recover(p2);
TEST_ASSERT_EQUAL_INT(0, rc);
/* After replay, current_lsn should reach the last replayed LSN + 1
* (or at minimum equal last_lsn_before_stop). */
TEST_ASSERT_TRUE(p2->current_lsn >= last_lsn_before_stop);
teardown_persist(p2);
}
printf("test_recover_replays_wal_after_checkpoint: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 4: recover with corrupt snapshot sentinel returns error */
/* ------------------------------------------------------------------ */
static void test_recover_rejects_corrupt_snapshot(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_rec3_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Build paths (same layout as persist_set_workdir) */
char snap_dir[512];
snprintf(snap_dir, sizeof(snap_dir), "%s/ctrl-test/data/snap", tmp);
/* Create the snap dir manually (without starting persist) */
{
char mkpath[512];
snprintf(mkpath, sizeof(mkpath), "%s/ctrl-test/data/wal", tmp);
/* use system mkdir -p */
char cmd[1024];
snprintf(cmd, sizeof(cmd), "mkdir -p '%s' '%s'", snap_dir, mkpath);
system(cmd);
}
/* Write a corrupt snapshot file (wrong magic) */
char snap_file[512];
snprintf(snap_file, sizeof(snap_file),
"%s/snap-%016" PRIx64 ".capnp", snap_dir, (uint64_t)1);
{
int fd = open(snap_file, O_CREAT | O_WRONLY | O_TRUNC, 0600);
TEST_ASSERT_TRUE(fd >= 0);
const char bad[16] = "BADMAGIC\0\0\0\0\0\0\0\0";
write(fd, bad, sizeof(bad));
close(fd);
}
persist_t* p = make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
int rc = persist_recover(p);
/* Corrupt sentinel must cause a non-zero error return */
TEST_ASSERT_NOT_EQUAL(0, rc);
teardown_persist(p);
printf("test_recover_rejects_corrupt_snapshot: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 5: multiple checkpoints — recover picks the LATEST snapshot */
/* ------------------------------------------------------------------ */
static void test_recover_picks_latest_snapshot(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_rec4_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
uint64_t last_ckpt_lsn = 0;
/* Take 3 checkpoints; the last one should be loaded on recovery */
{
persist_t* p = make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
for (int i = 0; i < 3; i++) {
persist_wal_async(p, WAL_JOB_UPDATE, "w", 1);
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p));
}
last_ckpt_lsn = p->checkpoint_lsn;
teardown_persist(p);
}
/* Recover: checkpoint_lsn must equal the last checkpoint's lsn */
{
persist_t* p2 = make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
TEST_ASSERT_EQUAL_UINT64(last_ckpt_lsn, p2->checkpoint_lsn);
teardown_persist(p2);
}
printf("test_recover_picks_latest_snapshot: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test suite entry point */
/* ------------------------------------------------------------------ */
void test_persist_recovery(void)
{
RUN_TEST(test_recover_from_empty_state);
RUN_TEST(test_recover_loads_checkpoint_lsn);
RUN_TEST(test_recover_replays_wal_after_checkpoint);
RUN_TEST(test_recover_rejects_corrupt_snapshot);
RUN_TEST(test_recover_picks_latest_snapshot);
}

12
src/controller/worker.c
View file

@ -646,3 +646,15 @@ int do_hostcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xce_reply(ep, c->myhost, &reply);
return rc;
}
/* P0 stub — bridges Cap'n Proto deserialization to restore_worker().
* Full implementation in Task 15 (cluster wiring): deserializes the
* Cap'n Proto Worker object, deep-copies pointer fields, and calls
* restore_worker(cls, host). */
int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len)
{
(void)c;
(void)capnp_payload;
(void)len;
return 0;
}