From b52fbef36dc460bd6d6c899e58a6f1e957adf22e Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 13:07:47 +0000 Subject: [PATCH] feat: wire persist_init/recover/start/stop into cluster lifecycle (P0-F) - Add `persist_t persist` field to cluster_t (cluster_def.h includes persist.h) - Add `workdir` and `name` fields to myconf_t; main.c passes them through - cluster_init: call persist_set_workdir, persist_init, persist_recover after apiserver_init - cluster_start: call persist_start (launches fsync thread + registers WAL hooks) - cluster_fini: call persist_stop + persist_fini before delaymgr/vtaskmgr teardown - persist_init: store cluster back-pointer (replaces (void)c stub) - persist_start: register persist_job/queue/worker_hook_fun via xce_add_hook when cluster != NULL - Add test_cluster_persist.c: compile-time struct check + config defaults + workdir path derivation - Register test in controller_test.c and meson.build All controller tests pass (1/1 OK, 50s). --- src/controller/cluster.c | 33 ++++++ src/controller/main.c | 2 + src/controller/meson.build | 1 + src/controller/persist.c | 13 ++- src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_cluster_persist.c | 103 ++++++++++++++++++ src/include/myconf.h | 2 + src/include/persist.h | 3 + src/include/xce/cluster_def.h | 2 + 9 files changed, 159 insertions(+), 2 deletions(-) create mode 100644 src/controller/testsuite/test_cluster_persist.c diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 4f1ab6b8..2bc24c8c 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -610,6 +610,28 @@ int cluster_init(cluster_t* c) apiserver_init(c); + /* Persistence: derive paths, init, and recover from WAL/snapshot */ + { + const char* workdir = c->myconf->workdir ? c->myconf->workdir + : DEFAULT_WORKDIR; + const char* name = c->myconf->name ? c->myconf->name + : "controller"; + persist_config_t pcfg = persist_config_defaults(); + + persist_set_workdir(&c->persist, workdir, name); + rc = persist_init(&c->persist, &pcfg, c); + if (rc != 0) { + ERROR("persist_init failed: %d", rc); + return rc; + } + rc = persist_recover(&c->persist); + if (rc != 0) { + ERROR("persist_recover failed: %d", rc); + persist_fini(&c->persist); + return rc; + } + } + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT, HOOK_DETAIL_CLUSTER_INIT, c); @@ -758,6 +780,13 @@ int cluster_start(cluster_t* c) vtaskmgr_start(&(c->vtaskmgr)); delaymgr_start(&(c->delaymgr)); + /* Start persistence fsync thread and register WAL hooks */ + rc = persist_start(&c->persist); + if (rc != 0) { + ERROR("persist_start failed: %d", rc); + return rc; + } + delay_task(&(c->delaymgr), c->myconf->cluster_config->heartbeat_interval, -1, (delay_task_entry_fn_t)check_workers_timeout, (void*)c); @@ -790,6 +819,10 @@ int cluster_run(cluster_t* c) void cluster_fini(cluster_t* c) { + /* Stop persistence before tearing down the rest */ + persist_stop(&c->persist); + persist_fini(&c->persist); + delaymgr_fini(&(c->delaymgr)); vtaskmgr_fini(&(c->vtaskmgr)); endpoint_fini(&(c->ep)); diff --git a/src/controller/main.c b/src/controller/main.c index ba21bf34..bb6ed70d 100644 --- a/src/controller/main.c +++ b/src/controller/main.c @@ -109,6 +109,8 @@ int main(int argc, char* argv[]) myconf.conffile = conf_file; myconf.certfile = cert_file; myconf.keyfile = key_file; + myconf.workdir = arguments.workdir; + myconf.name = arguments.myname; myapp_run(&myconf); diff --git a/src/controller/meson.build b/src/controller/meson.build index b2e3ef27..440b2425 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -76,6 +76,7 @@ controller_unit_test_src = [ 'testsuite/test_hooks.c', 'testsuite/test_persist_hooks.c', 'testsuite/test_gc_retention.c', + 'testsuite/test_cluster_persist.c', 'persist.c', ] diff --git a/src/controller/persist.c b/src/controller/persist.c index 99768c50..6f4ae8cb 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -137,8 +137,7 @@ void persist_set_workdir(persist_t* p, const char* workdir, const char* name) int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) { - (void)c; - + p->cluster = c; p->config = cfg ? *cfg : persist_config_defaults(); p->wal_fd = -1; p->last_replicated_crc = WAL_CRC_NONE; @@ -171,6 +170,16 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) int persist_start(persist_t* p) { + /* Register AFTER_EVENT persistence hooks if we have a cluster back-pointer */ + if (p->cluster != NULL) { + xce_add_hook(&p->cluster->hooksys, XCE_CONTROLLER_HOOK_JOB, + (hook_fn_t)persist_job_hook_fun, p); + xce_add_hook(&p->cluster->hooksys, XCE_CONTROLLER_HOOK_QUEUE, + (hook_fn_t)persist_queue_hook_fun, p); + xce_add_hook(&p->cluster->hooksys, XCE_CONTROLLER_HOOK_WORKER, + (hook_fn_t)persist_worker_hook_fun, 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) diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 8186cd1f..fa0a6c5a 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -13,6 +13,7 @@ extern void test_persist_recovery(void); extern void test_hooks(void); extern void test_persist_hooks(void); extern void test_gc_retention(void); +extern void test_cluster_persist(void); void setUp(void) { return; } @@ -95,6 +96,7 @@ int main(int argc, char* argv[]) TEST(test_hooks); TEST(test_persist_hooks); TEST(test_gc_retention); + TEST(test_cluster_persist); return UNITY_END(); } diff --git a/src/controller/testsuite/test_cluster_persist.c b/src/controller/testsuite/test_cluster_persist.c new file mode 100644 index 00000000..801611cc --- /dev/null +++ b/src/controller/testsuite/test_cluster_persist.c @@ -0,0 +1,103 @@ +#define _GNU_SOURCE +#include +#include + +#include + +#include "config.h" +#include "persist.h" +#include "xce/cluster_def.h" + +/* ------------------------------------------------------------------ */ +/* Compile-time structural checks */ +/* ------------------------------------------------------------------ */ + +static void test_cluster_t_has_persist_field(void) +{ + /* Verify cluster_t has a persist field — compile-time structural check. + * If persist_t persist is not in cluster_t, this will fail to compile. */ + cluster_t c; + memset(&c, 0, sizeof(c)); + /* Access the persist field to confirm it exists at compile time */ + (void)c.persist.current_lsn; + TEST_PASS_MESSAGE("cluster_t has persist field"); +} + +/* ------------------------------------------------------------------ */ +/* persist_config_defaults() correctness */ +/* ------------------------------------------------------------------ */ + +static void test_persist_config_defaults_wal_sync_ms(void) +{ + persist_config_t cfg = persist_config_defaults(); + TEST_ASSERT_EQUAL_UINT32(DEFAULT_WAL_SYNC_MS, cfg.wal_sync_ms); +} + +static void test_persist_config_defaults_snap_interval_s(void) +{ + persist_config_t cfg = persist_config_defaults(); + TEST_ASSERT_EQUAL_UINT32(DEFAULT_SNAP_INTERVAL_S, cfg.snap_interval_s); +} + +static void test_persist_config_defaults_snap_wal_size_mb(void) +{ + persist_config_t cfg = persist_config_defaults(); + TEST_ASSERT_EQUAL_UINT32(DEFAULT_SNAP_WAL_SIZE_MB, cfg.snap_wal_size_mb); +} + +static void test_persist_config_defaults_snap_keep_n(void) +{ + persist_config_t cfg = persist_config_defaults(); + TEST_ASSERT_EQUAL_UINT32(DEFAULT_SNAP_KEEP_N, cfg.snap_keep_n); +} + +static void test_persist_config_defaults_finished_retain_s(void) +{ + persist_config_t cfg = persist_config_defaults(); + TEST_ASSERT_EQUAL_UINT32(DEFAULT_FINISHED_RETAIN_S, cfg.finished_retain_s); +} + +static void test_persist_config_defaults_sync_timeout_ms(void) +{ + persist_config_t cfg = persist_config_defaults(); + TEST_ASSERT_EQUAL_UINT32(DEFAULT_SYNC_TIMEOUT_MS, cfg.sync_timeout_ms); +} + +/* ------------------------------------------------------------------ */ +/* persist_set_workdir derives correct paths */ +/* ------------------------------------------------------------------ */ + +static void test_persist_set_workdir_derives_wal_path(void) +{ + persist_t p; + memset(&p, 0, sizeof(p)); + persist_set_workdir(&p, "/tmp/xce", "ctrl1"); + TEST_ASSERT_EQUAL_STRING("/tmp/xce/ctrl1/data/wal", p.wal_path); +} + +static void test_persist_set_workdir_derives_snap_path(void) +{ + persist_t p; + memset(&p, 0, sizeof(p)); + persist_set_workdir(&p, "/tmp/xce", "ctrl1"); + TEST_ASSERT_EQUAL_STRING("/tmp/xce/ctrl1/data/snap", p.snap_path); +} + +/* ------------------------------------------------------------------ */ +/* Suite entry point */ +/* ------------------------------------------------------------------ */ + +void test_cluster_persist(void) +{ + printf("\n>> cluster persist wiring tests (P0-F)\n\n"); + + RUN_TEST(test_cluster_t_has_persist_field); + RUN_TEST(test_persist_config_defaults_wal_sync_ms); + RUN_TEST(test_persist_config_defaults_snap_interval_s); + RUN_TEST(test_persist_config_defaults_snap_wal_size_mb); + RUN_TEST(test_persist_config_defaults_snap_keep_n); + RUN_TEST(test_persist_config_defaults_finished_retain_s); + RUN_TEST(test_persist_config_defaults_sync_timeout_ms); + RUN_TEST(test_persist_set_workdir_derives_wal_path); + RUN_TEST(test_persist_set_workdir_derives_snap_path); +} diff --git a/src/include/myconf.h b/src/include/myconf.h index 61b93b91..0b290e24 100644 --- a/src/include/myconf.h +++ b/src/include/myconf.h @@ -18,6 +18,8 @@ typedef struct { char* conffile; char* certfile; char* keyfile; + char* workdir; /* working directory for WAL/snapshot storage */ + char* name; /* controller name (hostname by default) */ } myconf_t; extern cluster_config_t* get_default_conf(void*); diff --git a/src/include/persist.h b/src/include/persist.h index 2d68f6f0..1c1e8206 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -87,6 +87,9 @@ typedef struct { * full API surface is stable. * ============================================================ */ struct persist_s { + /* Back-pointer to owning cluster (set by persist_init, used in persist_start) */ + cluster_t* cluster; + /* WAL state */ int wal_fd; uint64_t current_lsn; diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index f8dca84b..dc27d63e 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -14,6 +14,7 @@ #include "myconf.h" #include "mylock.h" #include "netio.h" +#include "persist.h" #include "vtaskmgr.h" #include "xce/types.h" #include "xlist.h" @@ -114,6 +115,7 @@ typedef struct cluster_s { vtaskmgr_t vtaskmgr; cbuf_t wbuf; int running; + persist_t persist; /* Persistence module — P0-F */ } cluster_t; #endif /* XCE_CLUSTER_DEF_H */