From e76f76a88a410029b10a13d1779ce7b08e73298c Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 11:19:08 +0000 Subject: [PATCH] feat: implement persist_checkpoint with WAL marker, snapshot, trim, rotation (P0-3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - persist_write_checkpoint_marker(): writes WAL_CHECKPOINT header entry with CRC, increments current_lsn; never updates last_replicated_crc (checkpoint markers are local-only, never replicated) - write_snapshot(): atomic tmp→rename write of P0 sentinel snapshot ("XCESNAP\0" + checkpoint_lsn); full Cap'n Proto body in Task 11 - trim_wal_segments(): removes seg-*.wal files whose start LSN is strictly less than checkpoint_lsn - rotate_snapshots(): collects .capnp files, sorts by name (LSN order), deletes oldest beyond snap_keep_n - persist_checkpoint(): EBUSY guard via atomic_exchange, sequences marker→lsn record→snapshot→trim→rotate, clears flag on exit - 5 TDD tests in test_persist_checkpoint.c covering: snapshot file creation, checkpoint_lsn advance, WAL lsn advance, EBUSY guard, and rotation keeping at most snap_keep_n files - All 18 controller unit tests pass (0 failures) --- src/controller/meson.build | 1 + src/controller/persist.c | 196 ++++++++++++++++- src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_persist_checkpoint.c | 198 ++++++++++++++++++ 4 files changed, 391 insertions(+), 6 deletions(-) create mode 100644 src/controller/testsuite/test_persist_checkpoint.c diff --git a/src/controller/meson.build b/src/controller/meson.build index 56cb24b..feb75e8 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -71,6 +71,7 @@ controller_unit_test_src = [ 'testsuite/test_persist_types.c', 'testsuite/test_persist_wal.c', 'testsuite/test_persist_schema.c', + 'testsuite/test_persist_checkpoint.c', 'persist.c', ] diff --git a/src/controller/persist.c b/src/controller/persist.c index 76e3e64..65a017b 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -3,6 +3,7 @@ #include "config.h" #include +#include #include #include #include @@ -275,6 +276,195 @@ int persist_flush(persist_t* p) return rc; } +/* ------------------------------------------------------------------ */ +/* Checkpoint internals */ +/* ------------------------------------------------------------------ */ + +/* Write a WAL_CHECKPOINT marker to the current segment. + * Never calls repl_broadcast() — checkpoint markers are local only. */ +static int persist_write_checkpoint_marker(persist_t* p) +{ + pthread_mutex_lock(&p->lock); + + if (p->wal_fd < 0) { + if (open_new_segment(p) != 0) { + pthread_mutex_unlock(&p->lock); + return -EIO; + } + } + + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.term = p->term; + hdr.lsn = p->current_lsn++; + hdr.timestamp = now_us(); + hdr.type = WAL_CHECKPOINT; + hdr.payload_len = 0; + hdr.prev_crc = (p->seg_offset == 0) ? WAL_CRC_NONE : p->last_crc; + + uint32_t c = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE); + hdr.crc = c; + + ssize_t hw = pwrite(p->wal_fd, &hdr, sizeof(hdr), (off_t)p->seg_offset); + if (hw != (ssize_t)sizeof(hdr)) { + pthread_mutex_unlock(&p->lock); + return -EIO; + } + fdatasync(p->wal_fd); + + p->seg_offset += (uint32_t)sizeof(hdr); + p->total_wal_bytes += sizeof(hdr); + p->last_crc = hdr.crc; + /* NOTE: last_replicated_crc is NOT updated — WAL_CHECKPOINT is never replicated */ + + pthread_mutex_unlock(&p->lock); + return 0; +} + +/* Write a P0 sentinel snapshot: "XCESNAP\0" + checkpoint_lsn (16 bytes). + * Full Cap'n Proto serialization is Task 11 when recovery needs to read it back. */ +static int write_snapshot(persist_t* p) +{ + char tmp_path[4096]; + char final_path[4096]; + + snprintf(tmp_path, sizeof(tmp_path), + "%s/snap-%016" PRIx64 ".tmp", p->snap_path, p->checkpoint_lsn); + snprintf(final_path, sizeof(final_path), + "%s/snap-%016" PRIx64 ".capnp", p->snap_path, p->checkpoint_lsn); + + int fd = open(tmp_path, O_CREAT | O_WRONLY | O_TRUNC, 0600); + if (fd < 0) + return -errno; + + /* Sentinel header: magic (8 bytes) + checkpoint_lsn (8 bytes) */ + char header[16]; + memcpy(header, "XCESNAP\0", 8); + memcpy(header + 8, &p->checkpoint_lsn, 8); + + ssize_t nw = write(fd, header, sizeof(header)); + fdatasync(fd); + close(fd); + + if (nw != (ssize_t)sizeof(header)) { + unlink(tmp_path); + return -EIO; + } + + if (rename(tmp_path, final_path) != 0) { + int saved = errno; + unlink(tmp_path); + return -saved; + } + return 0; +} + +/* Delete WAL segment files whose start LSN is strictly less than trim_lsn. + * Conservative: only trims segments that started before the checkpoint. */ +static void trim_wal_segments(persist_t* p) +{ + DIR* d = opendir(p->wal_path); + if (!d) return; + + struct dirent* e; + while ((e = readdir(d)) != NULL) { + 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) { + char seg_path[4096]; + snprintf(seg_path, sizeof(seg_path), "%s/%s", p->wal_path, e->d_name); + unlink(seg_path); + } + } + closedir(d); +} + +/* Snapshot filename comparison helper for qsort — sorts ascending by name + * (which sorts ascending by LSN since filenames embed the LSN as hex). */ +static int snap_name_cmp(const void* a, const void* b) +{ + return strcmp(*(const char* const*)a, *(const char* const*)b); +} + +/* Keep only the snap_keep_n most recent snapshot files; delete the rest. */ +static void rotate_snapshots(persist_t* p) +{ + DIR* d = opendir(p->snap_path); + if (!d) return; + + /* Collect all .capnp snapshot filenames */ + char* names[1024]; + int n = 0; + struct dirent* e; + while ((e = readdir(d)) != NULL && n < (int)(sizeof(names) / sizeof(names[0]))) { + size_t len = strlen(e->d_name); + if (len <= 6 || strcmp(e->d_name + len - 6, ".capnp") != 0) continue; + names[n] = strdup(e->d_name); + if (names[n]) n++; + } + closedir(d); + + if (n == 0) return; + + /* Sort ascending — oldest first */ + qsort(names, (size_t)n, sizeof(names[0]), snap_name_cmp); + + /* Delete all but the snap_keep_n newest */ + int keep = (int)p->config.snap_keep_n; + int delete_up_to = n - keep; /* first index to keep */ + + for (int i = 0; i < delete_up_to && i < n; i++) { + char path[4096]; + snprintf(path, sizeof(path), "%s/%s", p->snap_path, names[i]); + unlink(path); + } + + for (int i = 0; i < n; i++) + free(names[i]); +} + +/* ------------------------------------------------------------------ */ +/* Public: persist_checkpoint */ +/* ------------------------------------------------------------------ */ + +int persist_checkpoint(persist_t* p) +{ + /* Guard against concurrent checkpoints */ + if (atomic_exchange(&p->checkpoint_in_progress, 1) != 0) + return -EBUSY; + + int rc = 0; + + /* 1. Write WAL_CHECKPOINT marker (local only — never replicated) */ + rc = persist_write_checkpoint_marker(p); + if (rc != 0) + goto done; + + /* 2. Record checkpoint_lsn = current position after the marker */ + pthread_mutex_lock(&p->lock); + p->checkpoint_lsn = p->current_lsn; + pthread_mutex_unlock(&p->lock); + + /* 3. Serialize in-memory state to a snapshot file + * (P0: no live kvmgrs — writes sentinel header only) */ + rc = write_snapshot(p); + if (rc != 0) + goto done; + + /* 4. Trim old WAL segments before the checkpoint */ + trim_wal_segments(p); + + /* 5. Rotate old snapshots — keep only snap_keep_n most recent */ + rotate_snapshots(p); + +done: + atomic_store(&p->checkpoint_in_progress, 0); + return rc; +} + /* ------------------------------------------------------------------ */ /* Stubs — wired in later tasks */ /* ------------------------------------------------------------------ */ @@ -295,12 +485,6 @@ int persist_wal_sync_append(persist_t* p, uint32_t type, return rc; } -int persist_checkpoint(persist_t* p) -{ - (void)p; - return 0; -} - void persist_log_entry_metadata(const wal_entry_hdr_t* hdr) { fprintf(stderr, diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 977a668..7b8610d 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -8,6 +8,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); void setUp(void) { return; } @@ -85,6 +86,7 @@ int main(int argc, char* argv[]) TEST(test_persist_types); TEST(test_persist_wal); TEST(test_persist_schema); + TEST(test_persist_checkpoint); return UNITY_END(); } diff --git a/src/controller/testsuite/test_persist_checkpoint.c b/src/controller/testsuite/test_persist_checkpoint.c new file mode 100644 index 0000000..f138694 --- /dev/null +++ b/src/controller/testsuite/test_persist_checkpoint.c @@ -0,0 +1,198 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "config.h" +#include "persist.h" + +/* ------------------------------------------------------------------ */ +/* Test-local helpers */ +/* ------------------------------------------------------------------ */ + +/* Count files with a given suffix in a directory */ +static int count_files_with_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; +} + +static persist_t* make_ckpt_persist(char* tmpdir_buf, size_t buf_sz) +{ + snprintf(tmpdir_buf, buf_sz, "/tmp/xce_ckpt_XXXXXX"); + if (!mkdtemp(tmpdir_buf)) 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; /* slow background fsync — prevent races */ + cfg.snap_keep_n = 2; + + persist_set_workdir(p, tmpdir_buf, "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 cleanup_ckpt_persist(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* ------------------------------------------------------------------ */ +/* Test 1: checkpoint creates at least one snapshot file */ +/* ------------------------------------------------------------------ */ + +static void test_checkpoint_creates_snapshot_file(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_persist(tmpdir, sizeof(tmpdir)); + TEST_ASSERT_NOT_NULL(p); + + /* Write a WAL entry so there is something to checkpoint */ + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "data", 4)); + + /* Trigger checkpoint */ + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + + /* Verify at least one .capnp snapshot file was created */ + char snap_path[512]; + snprintf(snap_path, sizeof(snap_path), "%s/ctrl-test/data/snap", tmpdir); + int n = count_files_with_suffix(snap_path, ".capnp"); + TEST_ASSERT_TRUE(n >= 1); + + cleanup_ckpt_persist(p); + printf("test_checkpoint_creates_snapshot_file: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 2: checkpoint advances checkpoint_lsn */ +/* ------------------------------------------------------------------ */ + +static void test_checkpoint_sets_checkpoint_lsn(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_persist(tmpdir, sizeof(tmpdir)); + TEST_ASSERT_NOT_NULL(p); + + /* Write 3 WAL entries */ + for (int i = 0; i < 3; i++) + persist_wal_async(p, WAL_JOB_UPDATE, "x", 1); + + uint64_t lsn_before = p->current_lsn; + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + + /* checkpoint_lsn must have advanced to at least lsn_before */ + TEST_ASSERT_TRUE(p->checkpoint_lsn >= lsn_before); + + cleanup_ckpt_persist(p); + printf("test_checkpoint_sets_checkpoint_lsn: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 3: checkpoint writes a WAL_CHECKPOINT marker (advances lsn) */ +/* ------------------------------------------------------------------ */ + +static void test_checkpoint_writes_marker_to_wal(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_persist(tmpdir, sizeof(tmpdir)); + TEST_ASSERT_NOT_NULL(p); + + uint64_t lsn_before = p->current_lsn; + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + + /* current_lsn must have advanced — WAL_CHECKPOINT marker was written */ + TEST_ASSERT_TRUE(p->current_lsn > lsn_before); + + cleanup_ckpt_persist(p); + printf("test_checkpoint_writes_marker_to_wal: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 4: concurrent checkpoint returns -EBUSY */ +/* ------------------------------------------------------------------ */ + +static void test_checkpoint_busy_guard(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_persist(tmpdir, sizeof(tmpdir)); + TEST_ASSERT_NOT_NULL(p); + + /* Simulate a checkpoint already in-progress */ + atomic_store(&p->checkpoint_in_progress, 1); + + int rc = persist_checkpoint(p); + TEST_ASSERT_EQUAL_INT(-EBUSY, rc); + + /* Clean up: allow stop to proceed */ + atomic_store(&p->checkpoint_in_progress, 0); + cleanup_ckpt_persist(p); + printf("test_checkpoint_busy_guard: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 5: snapshot rotation keeps at most snap_keep_n files */ +/* ------------------------------------------------------------------ */ + +static void test_checkpoint_rotation_keeps_snap_keep_n(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_persist(tmpdir, sizeof(tmpdir)); + TEST_ASSERT_NOT_NULL(p); + + /* snap_keep_n is 2; run 4 checkpoints */ + for (int i = 0; i < 4; i++) { + persist_wal_async(p, WAL_JOB_UPDATE, "y", 1); + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + } + + char snap_path[512]; + snprintf(snap_path, sizeof(snap_path), "%s/ctrl-test/data/snap", tmpdir); + int n = count_files_with_suffix(snap_path, ".capnp"); + TEST_ASSERT_TRUE(n <= (int)p->config.snap_keep_n); + + cleanup_ckpt_persist(p); + printf("test_checkpoint_rotation_keeps_snap_keep_n: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test suite entry point */ +/* ------------------------------------------------------------------ */ + +void test_persist_checkpoint(void) +{ + RUN_TEST(test_checkpoint_creates_snapshot_file); + RUN_TEST(test_checkpoint_sets_checkpoint_lsn); + RUN_TEST(test_checkpoint_writes_marker_to_wal); + RUN_TEST(test_checkpoint_busy_guard); + RUN_TEST(test_checkpoint_rotation_keeps_snap_keep_n); +}