feat: persist.c WAL writer with chained CRC32 and dedicated fsync pthread (P0-1)
- Expose persist_t struct concretely in persist.h (opaque pattern deferred) - persist_set_workdir() derives wal/snap paths from workdir+name - persist_init() creates WAL/snap directories, rejects NFS mounts - persist_wal_async() writes WAL entries with chained CRC32 via zlib - Dedicated fsync pthread fires every wal_sync_ms milliseconds - persist_flush() forces fdatasync for tests and clean shutdown - WAL_CHECKPOINT type rejected with -EINVAL (never written via async path) - Stubs: persist_recover, persist_wal_sync_append, persist_checkpoint - Add fsync_thread_running flag to guard persist_stop when thread not started - Add zlib_dep to controller and unit test meson targets - 3 new TDD tests: init_creates_directories, wal_async_writes_entry, checkpoint_type_rejected — all GREEN
This commit is contained in:
parent
d66a8a98b6
commit
b0a5fbb4e9
5 changed files with 502 additions and 2 deletions
|
|
@ -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',
|
||||
|
|
|
|||
309
src/controller/persist.c
Normal file
309
src/controller/persist.c
Normal file
|
|
@ -0,0 +1,309 @@
|
|||
#define _GNU_SOURCE
|
||||
#include "persist.h"
|
||||
#include "config.h"
|
||||
|
||||
#include <assert.h>
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <inttypes.h>
|
||||
#include <pthread.h>
|
||||
#include <stdatomic.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/stat.h>
|
||||
#include <sys/statfs.h>
|
||||
#include <time.h>
|
||||
#include <unistd.h>
|
||||
#include <zlib.h>
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* 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);
|
||||
}
|
||||
|
|
@ -6,6 +6,7 @@
|
|||
#include <unity/unity.h>
|
||||
|
||||
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();
|
||||
}
|
||||
|
|
|
|||
140
src/controller/testsuite/test_persist_wal.c
Normal file
140
src/controller/testsuite/test_persist_wal.c
Normal file
|
|
@ -0,0 +1,140 @@
|
|||
#define _GNU_SOURCE
|
||||
#include <assert.h>
|
||||
#include <dirent.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 */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
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);
|
||||
}
|
||||
|
|
@ -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 <workdir>/<name>/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);
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue