feat: implement persist_checkpoint with WAL marker, snapshot, trim, rotation (P0-3)
- 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)
This commit is contained in:
parent
813a5a3359
commit
e76f76a88a
4 changed files with 391 additions and 6 deletions
|
|
@ -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',
|
||||
]
|
||||
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
#include "config.h"
|
||||
|
||||
#include <assert.h>
|
||||
#include <dirent.h>
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <inttypes.h>
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
|
|
|
|||
198
src/controller/testsuite/test_persist_checkpoint.c
Normal file
198
src/controller/testsuite/test_persist_checkpoint.c
Normal file
|
|
@ -0,0 +1,198 @@
|
|||
#define _GNU_SOURCE
|
||||
#include <assert.h>
|
||||
#include <dirent.h>
|
||||
#include <errno.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"
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* 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);
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue