diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 6bf0cedb..0919a72a 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -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*); diff --git a/src/controller/include/worker.h b/src/controller/include/worker.h index 77834ee2..6c9691b5 100644 --- a/src/controller/include/worker.h +++ b/src/controller/include/worker.h @@ -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 diff --git a/src/controller/job.c b/src/controller/job.c index 36572a96..829437bb 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -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; diff --git a/src/controller/meson.build b/src/controller/meson.build index feb75e82..07659d11 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -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', ] diff --git a/src/controller/persist.c b/src/controller/persist.c index 65a017b7..1c3e138b 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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 "/seg-.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-.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; } diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 7b8610d8..5ab71c9f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -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(); } diff --git a/src/controller/testsuite/test_persist_recovery.c b/src/controller/testsuite/test_persist_recovery.c new file mode 100644 index 00000000..b888680e --- /dev/null +++ b/src/controller/testsuite/test_persist_recovery.c @@ -0,0 +1,258 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#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); +} diff --git a/src/controller/worker.c b/src/controller/worker.c index f9b08ada..bb099f7d 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -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; +}