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).
This commit is contained in:
Rongsong Shen 2026-05-07 13:07:47 +00:00
parent 3030012f51
commit b52fbef36d
9 changed files with 159 additions and 2 deletions

View file

@ -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));

View file

@ -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);

View file

@ -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',
]

View file

@ -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)

View file

@ -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();
}

View file

@ -0,0 +1,103 @@
#define _GNU_SOURCE
#include <stdio.h>
#include <string.h>
#include <unity/unity.h>
#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);
}

View file

@ -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*);

View file

@ -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;

View file

@ -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 */