diff --git a/src/controller/meson.build b/src/controller/meson.build index 23ac5df0..26e35555 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -16,6 +16,7 @@ controller_src = [ 'worker.c', 'apiserver.c', 'myconf.c', + 'persist.c', ] incdir = include_directories('.', '../include', '../lib', @@ -24,13 +25,16 @@ incdir = include_directories('.', '../include', '../lib', # cc = meson.get_compiler('c') # math_dep = cc.find_library('m', required : false) +zlib_dep = dependency('zlib') + controller_deps = [ libxce_dep, librtti_dep, c_capnproto_dep, math_dep, tinylog_dep, - thread_dep + thread_dep, + zlib_dep, ] controller = executable('xce_controller', @@ -65,12 +69,16 @@ xce_test = executable('xce_test', controller_unit_test_src = [ 'testsuite/controller_test.c', 'testsuite/test_persist_types.c', + 'testsuite/test_persist_wal.c', + 'persist.c', ] controller_unit_test_deps = [ libxce_dep, unity_dep, tinylog_dep, + thread_dep, + zlib_dep, ] controller_unit_test = executable('controller_unit_test', diff --git a/src/controller/persist.c b/src/controller/persist.c new file mode 100644 index 00000000..e39aad58 --- /dev/null +++ b/src/controller/persist.c @@ -0,0 +1,309 @@ +#define _GNU_SOURCE +#include "persist.h" +#include "config.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* ------------------------------------------------------------------ */ +/* Constants */ +/* ------------------------------------------------------------------ */ + +#define WAL_SEGMENT_SIZE_BYTES (64UL * 1024 * 1024) /* 64 MB per segment */ +#define WAL_SEG_FMT "%s/seg-%016" PRIx64 ".wal" + +/* NFS_SUPER_MAGIC — reject WAL on NFS mounts (unreliable fsync) */ +#define NFS_SUPER_MAGIC 0x6969 + +/* ------------------------------------------------------------------ */ +/* Internal helpers */ +/* ------------------------------------------------------------------ */ + +static uint64_t now_us(void) +{ + struct timespec ts; + clock_gettime(CLOCK_REALTIME, &ts); + return (uint64_t)ts.tv_sec * 1000000ULL + (uint64_t)(ts.tv_nsec / 1000); +} + +/* Recursive mkdir — equivalent to mkdir -p */ +static int mkdir_p(const char* path, mode_t mode) +{ + char tmp[4096]; + char* p; + size_t len; + + snprintf(tmp, sizeof(tmp), "%s", path); + len = strlen(tmp); + if (len > 0 && tmp[len - 1] == '/') + tmp[len - 1] = '\0'; + + for (p = tmp + 1; *p; p++) { + if (*p == '/') { + *p = '\0'; + if (mkdir(tmp, mode) != 0 && errno != EEXIST) + return -1; + *p = '/'; + } + } + if (mkdir(tmp, mode) != 0 && errno != EEXIST) + return -1; + return 0; +} + +/* Open (and pre-allocate) a new WAL segment named after current_lsn */ +static int open_new_segment(persist_t* p) +{ + snprintf(p->seg_path, sizeof(p->seg_path), + WAL_SEG_FMT, p->wal_path, p->current_lsn); + + int fd = open(p->seg_path, O_RDWR | O_CREAT | O_TRUNC, 0600); + if (fd < 0) + return -errno; + + /* Pre-allocate segment space to avoid fragmentation and metadata updates + * on every write. posix_fallocate is best-effort — ignore errors. */ + posix_fallocate(fd, 0, (off_t)WAL_SEGMENT_SIZE_BYTES); + + p->wal_fd = fd; + p->seg_offset = 0; + return 0; +} + +/* Close current segment and open a fresh one */ +static int rotate_segment(persist_t* p) +{ + if (p->wal_fd >= 0) { + fdatasync(p->wal_fd); + close(p->wal_fd); + p->wal_fd = -1; + } + return open_new_segment(p); +} + +/* ------------------------------------------------------------------ */ +/* Dedicated fsync pthread (fires every wal_sync_ms milliseconds) */ +/* ------------------------------------------------------------------ */ + +static void* fsync_worker(void* arg) +{ + persist_t* p = arg; + + while (!atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) { + long ms = (long)p->config.wal_sync_ms; + struct timespec ts = { + .tv_sec = ms / 1000, + .tv_nsec = (ms % 1000) * 1000000L, + }; + nanosleep(&ts, NULL); + + pthread_mutex_lock(&p->lock); + if (p->wal_fd >= 0) + fdatasync(p->wal_fd); + pthread_mutex_unlock(&p->lock); + } + return NULL; +} + +/* ------------------------------------------------------------------ */ +/* Public API */ +/* ------------------------------------------------------------------ */ + +void persist_set_workdir(persist_t* p, const char* workdir, const char* name) +{ + char effective[4096]; + snprintf(effective, sizeof(effective), "%s/%s", workdir, name); + snprintf(p->wal_path, sizeof(p->wal_path), "%s/data/wal", effective); + snprintf(p->snap_path, sizeof(p->snap_path), "%s/data/snap", effective); +} + +int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) +{ + (void)c; + + p->config = cfg ? *cfg : persist_config_defaults(); + p->wal_fd = -1; + p->last_replicated_crc = WAL_CRC_NONE; + p->last_crc = WAL_CRC_NONE; + + /* Create WAL and snapshot directories */ + if (mkdir_p(p->wal_path, 0700) != 0) return -errno; + if (mkdir_p(p->snap_path, 0700) != 0) return -errno; + + /* Reject NFS mounts — fsync durability is not guaranteed over NFS */ + struct statfs sfs; + if (statfs(p->wal_path, &sfs) == 0 && + (unsigned long)sfs.f_type == (unsigned long)NFS_SUPER_MAGIC) { + fprintf(stderr, + "persist: WAL path %s is on NFS — unsafe for fsync durability\n", + p->wal_path); + return -EREMOTE; + } + + pthread_mutex_init(&p->lock, NULL); + pthread_mutex_init(&p->checkpoint_mutex, NULL); + pthread_cond_init(&p->checkpoint_done, NULL); + + fprintf(stderr, + "INFO: WAL/snapshot stored unencrypted in %s — " + "ensure filesystem is encrypted if job payloads contain sensitive data\n", + p->wal_path); + return 0; +} + +int persist_start(persist_t* p) +{ + atomic_store_explicit(&p->fsync_stop, 0, memory_order_release); + int rc = pthread_create(&p->fsync_thread, NULL, fsync_worker, p); + if (rc == 0) + atomic_store_explicit(&p->fsync_thread_running, 1, memory_order_release); + return rc; +} + +void persist_stop(persist_t* p) +{ + if (atomic_load_explicit(&p->fsync_thread_running, memory_order_acquire)) { + atomic_store_explicit(&p->fsync_stop, 1, memory_order_release); + pthread_join(p->fsync_thread, NULL); + atomic_store_explicit(&p->fsync_thread_running, 0, memory_order_release); + } + + pthread_mutex_lock(&p->lock); + if (p->wal_fd >= 0) { + fdatasync(p->wal_fd); + close(p->wal_fd); + p->wal_fd = -1; + } + pthread_mutex_unlock(&p->lock); +} + +void persist_fini(persist_t* p) +{ + pthread_mutex_destroy(&p->lock); + pthread_mutex_destroy(&p->checkpoint_mutex); + pthread_cond_destroy(&p->checkpoint_done); +} + +int persist_wal_async(persist_t* p, uint32_t type, + const void* payload, uint32_t len) +{ + /* WAL_CHECKPOINT is a local segment marker, never written via this path */ + if (type == WAL_CHECKPOINT) + return -EINVAL; + + pthread_mutex_lock(&p->lock); + + /* Open first segment lazily */ + if (p->wal_fd < 0) { + if (open_new_segment(p) != 0) { + pthread_mutex_unlock(&p->lock); + return -EIO; + } + } + + /* Build entry header */ + 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 = type; + hdr.payload_len = len; + hdr.prev_crc = (p->seg_offset == 0) ? WAL_CRC_NONE : p->last_crc; + + /* CRC32 over WAL_HDR_CRC_SIZE bytes of header (excludes crc + prev_crc + * fields themselves) then over the payload */ + uint32_t c = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE); + if (payload && len > 0) + c = crc32(c, (const Bytef*)payload, len); + hdr.crc = c; + + /* Write header */ + 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; + } + + /* Write payload (if any) */ + if (payload && len > 0) { + ssize_t pw = pwrite(p->wal_fd, payload, len, + (off_t)(p->seg_offset + sizeof(hdr))); + if (pw != (ssize_t)len) { + pthread_mutex_unlock(&p->lock); + return -EIO; + } + } + + uint32_t entry_size = (uint32_t)(sizeof(hdr) + len); + p->seg_offset += entry_size; + p->total_wal_bytes += entry_size; + p->last_crc = hdr.crc; + /* For single-controller P0 there is no replication; track same value */ + p->last_replicated_crc = hdr.crc; + + /* Rotate segment when full */ + if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) + rotate_segment(p); + + pthread_mutex_unlock(&p->lock); + return 0; +} + +int persist_flush(persist_t* p) +{ + pthread_mutex_lock(&p->lock); + int rc = (p->wal_fd >= 0) ? fdatasync(p->wal_fd) : 0; + pthread_mutex_unlock(&p->lock); + return rc; +} + +/* ------------------------------------------------------------------ */ +/* Stubs — wired in later tasks */ +/* ------------------------------------------------------------------ */ + +int persist_recover(persist_t* p) +{ + (void)p; + return 0; +} + +int persist_wal_sync_append(persist_t* p, uint32_t type, + const void* payload, uint32_t len, + int* sync_rc_out) +{ + int rc = persist_wal_async(p, type, payload, len); + if (sync_rc_out) + *sync_rc_out = rc; + 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, + "WAL entry: type=%u lsn=%lu term=%lu payload_len=%u crc=%08x\n", + hdr->type, + (unsigned long)hdr->lsn, + (unsigned long)hdr->term, + hdr->payload_len, + hdr->crc); +} diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index fd57952f..2d0ea1f1 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -6,6 +6,7 @@ #include extern void test_persist_types(void); +extern void test_persist_wal(void); void setUp(void) { return; } @@ -81,6 +82,7 @@ int main(int argc, char* argv[]) UNITY_BEGIN(); TEST(test_persist_types); + TEST(test_persist_wal); return UNITY_END(); } diff --git a/src/controller/testsuite/test_persist_wal.c b/src/controller/testsuite/test_persist_wal.c new file mode 100644 index 00000000..a235c44c --- /dev/null +++ b/src/controller/testsuite/test_persist_wal.c @@ -0,0 +1,140 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "config.h" +#include "persist.h" + +/* ------------------------------------------------------------------ */ +/* Test-local helpers */ +/* ------------------------------------------------------------------ */ + +static char g_tmpdir[256]; + +/* Count *.wal files in a directory */ +static int count_wal_files(const char* dir_path) +{ + DIR* dir = opendir(dir_path); + if (!dir) return 0; + + int count = 0; + struct dirent* ent; + while ((ent = readdir(dir)) != NULL) { + size_t len = strlen(ent->d_name); + if (len > 4 && strcmp(ent->d_name + len - 4, ".wal") == 0) + count++; + } + closedir(dir); + return count; +} + +/* Allocate and initialise a persist_t for testing */ +static persist_t* make_test_persist(void) +{ + strcpy(g_tmpdir, "/tmp/xce_test_XXXXXX"); + if (!mkdtemp(g_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; /* slow background fsync — don't want it firing during tests */ + + persist_set_workdir(p, g_tmpdir, "ctrl-test"); + + if (persist_init(p, &cfg, NULL) != 0) { + free(p); + return NULL; + } + return p; +} + +static void cleanup_test_persist(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); + /* tmpdir is left on disk — /tmp is ephemeral */ +} + +/* ------------------------------------------------------------------ */ +/* Tests */ +/* ------------------------------------------------------------------ */ + +static void test_persist_init_creates_directories(void) +{ + persist_t* p = make_test_persist(); + TEST_ASSERT_NOT_NULL(p); + + char wal_path[512], snap_path[512]; + snprintf(wal_path, sizeof(wal_path), "%s/ctrl-test/data/wal", g_tmpdir); + snprintf(snap_path, sizeof(snap_path), "%s/ctrl-test/data/snap", g_tmpdir); + + struct stat st; + TEST_ASSERT_EQUAL_INT(0, stat(wal_path, &st)); + TEST_ASSERT_TRUE(S_ISDIR(st.st_mode)); + TEST_ASSERT_EQUAL_INT(0, stat(snap_path, &st)); + TEST_ASSERT_TRUE(S_ISDIR(st.st_mode)); + + cleanup_test_persist(p); + printf("test_persist_init_creates_directories: PASS\n"); +} + +static void test_persist_wal_async_writes_entry(void) +{ + persist_t* p = make_test_persist(); + TEST_ASSERT_NOT_NULL(p); + TEST_ASSERT_EQUAL_INT(0, persist_start(p)); + + const char payload[] = "test_job_submit_payload"; + int rc = persist_wal_async(p, WAL_JOB_UPDATE, payload, + (uint32_t)strlen(payload)); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Force sync so WAL is on disk */ + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + /* WAL directory must contain at least one .wal segment file */ + char wal_path[512]; + snprintf(wal_path, sizeof(wal_path), "%s/ctrl-test/data/wal", g_tmpdir); + int found = count_wal_files(wal_path); + TEST_ASSERT_TRUE(found >= 1); + + cleanup_test_persist(p); + printf("test_persist_wal_async_writes_entry: PASS\n"); +} + +static void test_wal_checkpoint_type_rejected(void) +{ + persist_t* p = make_test_persist(); + TEST_ASSERT_NOT_NULL(p); + TEST_ASSERT_EQUAL_INT(0, persist_start(p)); + + /* WAL_CHECKPOINT must never be passed to persist_wal_async — + * it must return a non-zero error code (not abort/assert). */ + int rc = persist_wal_async(p, WAL_CHECKPOINT, NULL, 0); + TEST_ASSERT_TRUE(rc != 0); + + cleanup_test_persist(p); + printf("test_wal_checkpoint_type_rejected: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Suite entry point (called from controller_test.c) */ +/* ------------------------------------------------------------------ */ + +void test_persist_wal(void) +{ + printf("\n>> persist WAL writer test\n\n"); + + RUN_TEST(test_persist_init_creates_directories); + RUN_TEST(test_persist_wal_async_writes_entry); + RUN_TEST(test_wal_checkpoint_type_rejected); +} diff --git a/src/include/persist.h b/src/include/persist.h index b436c5dd..f139ad8d 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -81,10 +81,48 @@ typedef struct { } slot_key_t; /* ============================================================ - * persist_t — opaque; defined in persist.c (Task 7) + * persist_t — concrete definition (exposed for stack/calloc allocation) + * The opaque pattern can be restored in a later refactor once the + * full API surface is stable. * ============================================================ */ +struct persist_s { + /* WAL state */ + int wal_fd; + uint64_t current_lsn; + uint64_t last_crc; + uint64_t last_replicated_crc; + uint64_t term; + uint32_t seg_offset; + uint64_t checkpoint_lsn; + uint64_t total_wal_bytes; + char seg_path[4096]; + + /* Paths (derived from workdir + name via persist_set_workdir) */ + char wal_path[4096]; + char snap_path[4096]; + + /* Config */ + persist_config_t config; + + /* Concurrency */ + pthread_mutex_t lock; + atomic_int fsync_stop; + atomic_int fsync_thread_running; + pthread_t fsync_thread; + + /* Checkpoint (stubs for now — wired in Task 10) */ + atomic_int checkpoint_in_progress; + atomic_int checkpoint_requested; + pthread_cond_t checkpoint_done; + pthread_mutex_t checkpoint_mutex; + pthread_t snap_thread; +}; typedef struct persist_s persist_t; +/* Set workdir and controller name before persist_init(). + * Derives wal_path and snap_path as //data/{wal,snap}. */ +void persist_set_workdir(persist_t* p, const char* workdir, const char* name); + /* Lifecycle */ int persist_init(persist_t* p, persist_config_t* config, cluster_t* c); int persist_start(persist_t* p); @@ -99,6 +137,9 @@ int persist_wal_sync_append(persist_t* p, uint32_t type, const void* payload, uint32_t len, int* sync_rc_out); +/* Force WAL fdatasync (for tests and clean shutdown) */ +int persist_flush(persist_t* p); + /* Checkpoint */ int persist_checkpoint(persist_t* p);