lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/persist.c
Rongsong Shen 59d9df246c fix(ha-commit-durability)!: controller never self-terminates — loud non-promotable fence replaces recovery FATAL/abort 
Per design invariant (a controller exits only on crash or admin kill), replace the
self-abort recovery terminal (tasks 3.3/4.3/4.4) with a loud, permanent, non-promotable
fence that stays alive and keeps retrying catch-up (self-heals on peer return; no manual
admin recovery). Records the never-self-terminate invariant as project-wide guidance in
src/controller/CLAUDE.md + .kiro/steering/ha-async-io.md.
2026-06-18 11:10:19 +00:00

2587 lines
103 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#define _GNU_SOURCE
#include "persist.h"
#include "config.h"
#include "ctrl_metrics.h"
#include "hook.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xce/types.h"
#include "controller.capnp.h"
#include "xlist.h"
#include "cluster.h"
#include "commit.h" /* commit_submit RETAIN registration (barrier-rollout 1.2) */
#include <assert.h>
#include <stdbool.h>
#include <dirent.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>
/* restore_* functions defined in job.c / worker.c / queue.c */
extern int restore_job(cluster_t* c, uint64_t jobid, const void* capnp_payload,
size_t len, uint64_t entry_lsn);
extern int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state);
extern int worker_from_capnp(cluster_t* c,
const void* capnp_payload, size_t len);
extern int restore_queue_state(cluster_t* c,
const char* name, uint32_t status);
/* ---- Cap'n Proto serialization helpers (WAL payload encoding) ---------- */
/* Serialize xce_job_t to a malloc'd Cap'n Proto buffer.
* Caller must free(*out) on success. */
static int capnp_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len)
{
struct capn c;
struct capn_segment* cs;
struct Job cj;
Job_ptr jp;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_Job(cs, &cj, (xce_job_t*)j);
jp = new_Job(cs);
write_Job(&cj, jp);
capn_setp(capn_root(&c), 0, jp.p);
*out_len = capn_size(&c);
*out = malloc(*out_len);
if (!*out) { capn_free(&c); return -ENOMEM; }
capn_write_mem(&c, *out, *out_len, 0);
capn_free(&c);
return 0;
}
/* Serialize xce_worker_t to a malloc'd Cap'n Proto buffer. */
static int capnp_encode_worker(const xce_worker_t* w, uint8_t** out, size_t* out_len)
{
struct capn c;
struct capn_segment* cs;
struct Worker cw;
Worker_ptr wp;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_Worker(cs, &cw, (xce_worker_t*)w);
wp = new_Worker(cs);
write_Worker(&cw, wp);
capn_setp(capn_root(&c), 0, wp.p);
*out_len = capn_size(&c);
*out = malloc(*out_len);
if (!*out) { capn_free(&c); return -ENOMEM; }
capn_write_mem(&c, *out, *out_len, 0);
capn_free(&c);
return 0;
}
/* WAL_WORKER_CTRL payload: [uint32_t detail][capnp bytes...] or [uint32_t detail][name\0] */
#define WORKER_WAL_HDR_SIZE sizeof(uint32_t)
/* ------------------------------------------------------------------ */
/* 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);
}
/* ------------------------------------------------------------------ */
/* Demand-driven group-commit flusher (ha-commit-durability task 2.1) */
/* */
/* Repurposes the old fixed-interval fsync timer into a flusher driven */
/* by the append-vs-synced gap. On each pass it batches one fdatasync */
/* over all entries appended since the last sync, advances durable_lsn */
/* to the synced tip, and wakes the commit evaluation. --wal-sync-ms */
/* now bounds the MAX batching latency (a no-further-append entry is */
/* still synced within the cadence), not a durability gap. */
/* */
/* LOST-WAKEUP / CORRECTNESS REASONING (gating item): */
/* */
/* The "work pending" condition is LEVEL-triggered on two persistent */
/* variables, NOT an edge: pending ⇔ durable_lsn < current_lsn - 1 */
/* (flusher_work_pending below, evaluated under p->lock). Because it */
/* is a comparison of stored state — not a transient flag consumed on */
/* read — a signal that races the flusher's wait can never be lost: */
/* */
/* - persist_wal_async advances current_lsn under p->lock, THEN */
/* signals fsync_cond. If the flusher is mid-flush when the append */
/* lands, current_lsn has already moved past the tip the flusher */
/* captured, so after that flush durable_lsn < current_lsn - 1 */
/* STILL holds; the flusher re-checks the level under p->lock on */
/* its next loop iteration and flushes again (it does NOT wait */
/* first). The missed signal is therefore harmless — the level */
/* itself carries the pending work forward. */
/* - If the append's signal arrives while the flusher sits in */
/* pthread_cond_wait, it wakes it. If it arrives in the tiny window */
/* after the flusher decided "no work" but before it re-enters the */
/* wait, the flusher re-evaluates the level under p->lock before */
/* waiting (we re-check inside the fsync_mutex-held region by */
/* reading current_lsn/durable_lsn under p->lock first), so it */
/* will not block while work is pending. */
/* */
/* "NEVER durable before its fdatasync": the flusher snapshots */
/* tip = current_lsn - 1 AND calls fdatasync WHILE HOLDING p->lock. */
/* persist_wal_async also pwrites under p->lock, so no append can */
/* extend the segment between the tip snapshot and the fdatasync — by */
/* the time fdatasync returns, every byte for LSNs <= tip is on stable */
/* storage. durable_lsn is only then advanced to tip (monotonic max). */
/* Releasing p->lock and broadcasting commit-eval happens after the */
/* advance, so a WAIT caller that observes durable_lsn >= its LSN is */
/* guaranteed the bytes are synced. */
/* */
/* Lock discipline: p->lock and fsync_mutex are NEVER held together */
/* (fsync_mutex guards only the cond wait; the flush section takes */
/* p->lock alone). This keeps fdatasync off any lock the iomgr / */
/* heartbeat path needs and preserves liveness (Req 1.6). */
/* ------------------------------------------------------------------ */
/* Test-only fault-injection probe (ha-commit-durability task 2.1 error path).
* Returns true when the test has forced fdatasync to be treated as failed.
* Compiled to a constant `false` in production (no -DXCE_TESTING), so the
* fail-loud gate below collapses to the pure rc != 0 check with zero overhead
* and no dependency on the test-only struct field. */
static inline bool flusher_fsync_forced_fail(persist_t* p)
{
#ifdef XCE_TESTING
return atomic_load_explicit(&p->test_force_fsync_fail,
memory_order_acquire) != 0;
#else
(void)p;
return false;
#endif
}
/* Returns true iff there is un-synced appended data (level predicate).
* MUST be called with p->lock held. current_lsn - 1 is the appended tip
* (current_lsn is the NEXT lsn to assign); LSN 0 reserved, so a fresh
* store has current_lsn == 1 → tip 0 → no pending work. */
static bool flusher_work_pending_locked(persist_t* p)
{
uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire);
return dur < tip;
}
/* Perform one group-commit flush: snapshot the appended tip, fdatasync the
* live segment, and advance durable_lsn to the synced tip. Takes p->lock
* for the whole snapshot+sync+advance so no concurrent append can extend
* the segment between the tip read and the fdatasync. Returns the new
* durable_lsn (unchanged if there was nothing to sync). */
static uint64_t flusher_flush_once(persist_t* p)
{
pthread_mutex_lock(&p->lock);
uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire);
if (dur >= tip || p->wal_fd < 0) {
/* Nothing new to sync (or no segment yet). Still flush the fd for a
* clean-shutdown / fallback pass, but do not advance the watermark.
* The watermark is already <= tip here, so a failed fallback sync
* cannot make it claim more than is durable — but we still surface
* the error loudly so a degrading disk is never silent (Req 1.1). */
if (p->wal_fd >= 0) {
int rc = fdatasync(p->wal_fd);
if (rc != 0 || flusher_fsync_forced_fail(p)) {
int e = (rc != 0) ? errno : EIO;
ERROR("persist: flusher fallback fdatasync FAILED (errno=%d %s)"
" — durable_lsn held at %" PRIu64 "; WAL not advanced",
e, strerror(e), dur);
}
}
pthread_mutex_unlock(&p->lock);
return dur;
}
/* fdatasync under p->lock: appends also hold p->lock to pwrite, so the
* segment cannot grow between this tip snapshot and the sync. Every byte
* for LSNs <= tip is therefore on stable storage when fdatasync returns
* SUCCESSFULLY. */
int rc = fdatasync(p->wal_fd);
/* FAIL-LOUD GATE (Req 1.1/1.2): advance durable_lsn ONLY on a successful
* sync. On EIO/ENOSPC the bytes are NOT on stable storage, so leaving the
* watermark unchanged keeps the non-durable LSNs from being counted toward
* the commit majority. We do NOT retry into an advanced watermark — the
* level predicate stays pending and the next pass retries the sync. */
if (rc != 0 || flusher_fsync_forced_fail(p)) {
int e = (rc != 0) ? errno : EIO;
ERROR("persist: flusher fdatasync FAILED (errno=%d %s) — durable_lsn "
"held at %" PRIu64 " (tip=%" PRIu64 " NOT made durable); commit "
"majority will not count these LSNs",
e, strerror(e), dur, tip);
pthread_mutex_unlock(&p->lock);
return dur;
}
/* Monotonic advance to the synced tip — never past it (the invariant
* durable_lsn <= current_lsn - 1 is preserved because tip == current_lsn-1
* was read under this same lock). One fdatasync advances durable_lsn past
* however many LSNs were appended since the last sync (batching). */
if (tip > dur)
atomic_store_explicit(&p->durable_lsn, tip, memory_order_release);
#ifdef XCE_TESTING
/* Record which thread ran this successful fdatasync (task 5.1 / Req 1.6):
* a liveness test asserts this is the dedicated flusher thread, never the
* caller/iomgr/heartbeat thread. Written under p->lock; read after the
* flusher has caught up, so no extra synchronization is needed. */
p->test_last_fdatasync_thread = pthread_self();
atomic_store_explicit(&p->test_last_fdatasync_thread_set, 1,
memory_order_release);
#endif
pthread_mutex_unlock(&p->lock);
return tip;
}
static void* fsync_worker(void* arg)
{
persist_t* p = arg;
pthread_mutex_lock(&p->fsync_mutex);
while (!atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) {
/* Re-evaluate the level predicate BEFORE waiting so a signal that
* raced in just before we entered the wait is not needed for
* correctness — if work is already pending we skip the wait entirely
* and flush. (Reads current_lsn/durable_lsn under p->lock.) */
pthread_mutex_lock(&p->lock);
bool pending = flusher_work_pending_locked(p);
pthread_mutex_unlock(&p->lock);
if (!pending) {
/* No pending work: wait for an append-signal OR the cadence
* deadline (bounded fallback flush so a lone no-further-append
* entry is still synced within wal_sync_ms — Req 3.1). */
long ms = (long)p->config.wal_sync_ms;
struct timespec dl;
clock_gettime(CLOCK_REALTIME, &dl);
dl.tv_sec += ms / 1000;
dl.tv_nsec += (ms % 1000) * 1000000L;
if (dl.tv_nsec >= 1000000000L) {
dl.tv_sec += 1;
dl.tv_nsec -= 1000000000L;
}
pthread_cond_timedwait(&p->fsync_cond, &p->fsync_mutex, &dl);
if (atomic_load_explicit(&p->fsync_stop, memory_order_acquire))
break;
}
/* Flush section: never hold fsync_mutex across the fdatasync (the
* flush takes p->lock alone). Release fsync_mutex, flush, re-acquire.
* After the flush we loop and re-check the level — if a new append
* landed during the flush, the level is still pending and we flush
* again without needing its signal (lost-wakeup-safe). */
pthread_mutex_unlock(&p->fsync_mutex);
uint64_t before = atomic_load_explicit(&p->durable_lsn,
memory_order_acquire);
uint64_t after = flusher_flush_once(p);
/* COMMIT-EVAL WAKE (task 2.2 consumer wired): durable_lsn is advanced
* INSIDE flusher_flush_once (atomic store with release) BEFORE we
* return here, and flusher_flush_once has already RELEASED p->lock.
* Now — holding NO persist lock — invoke the registered durable-advance
* callback so the commit engine re-evaluates its majority and wakes any
* COMMIT_MODE_WAIT caller whose entry just became durable locally.
*
* LOST-WAKEUP SAFETY: the wake fires strictly AFTER the durable_lsn
* release-store, and the commit engine's predicate re-reads durable_lsn
* via an acquire load under ce->lock — so a WAIT caller either (a) sees
* the advanced watermark on its own re-check, or (b) is blocked in
* pthread_cond_wait and is woken by this broadcast; it can never both
* miss the watermark AND miss the wake. We only fire the callback when
* durable_lsn actually advanced (after > before): a no-op flush (or a
* failed fdatasync that held the watermark) needs no commit re-eval,
* avoiding spurious wakes.
*
* LOCK ORDERING (cycle-free): the flusher releases p->lock inside
* flusher_flush_once, THEN the callback takes ce->lock (commit side).
* The commit side only ever takes ce->lock and reads persist atomics —
* it never takes p->lock while holding ce->lock. So the only ordering
* is p->lock → ce->lock; there is no ce->lock → p->lock edge, hence no
* deadlock cycle. The callback explicitly MUST NOT take p->lock. */
if (after > before && p->on_durable_advance != NULL) {
p->on_durable_advance(p->durable_cb_ctx);
}
pthread_mutex_lock(&p->fsync_mutex);
}
pthread_mutex_unlock(&p->fsync_mutex);
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)
{
p->cluster = 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;
/* LSN 0 is RESERVED as the "none / empty" sentinel — it is NEVER assigned
* to a real WAL entry. A fresh store therefore starts current_lsn at 1, so
* the first-ever appended entry (e.g. the promotion WAL_NOOP) is lsn 1 and
* lsn 0 unambiguously means "nothing here":
* - committed_lsn == 0 → nothing committed (the lsn-1 NOOP is committable
* 0 → 1, vs. the old lsn-0 entry which could never advance committed_lsn);
* - writable_lsn == 0 → own-term gate off (the gate armed at lsn 1 is
* effective, vs. lsn 0 which read as "gate off" on a cold-start leader);
* - standby_lsn == 0 → empty standby (the streamer's `hdr.lsn > 0`
* filter streams the lsn-1 entry, closing the cold-start WAL hole that
* dropped lsn 0 on `0 > 0`).
* persist_recover() overwrites current_lsn from the durable snapshot/WAL tip
* (load_snapshot / replay_wal), so this fresh default only applies before any
* recovery — i.e. a brand-new store with no snapshot and no WAL. */
p->current_lsn = 1;
/* Enforce minimum value for max_wal_records to prevent runaway checkpointing */
if (p->config.max_wal_records < 100) {
ERROR("persist: max_wal_records=%u too low (minimum 100) — setting to 100",
p->config.max_wal_records);
p->config.max_wal_records = 100;
}
/* Initialise atomic checkpoint-trigger counters */
atomic_store(&p->wal_record_count, 0);
atomic_store(&p->checkpoint_in_progress, 0);
/* Durable watermark starts at 0 ("nothing fdatasync'd"): mirrors the
* LSN-0-reserved committed_lsn convention. It advances only on the
* flusher thread after a successful fdatasync (ha-commit-durability 2.1). */
atomic_store(&p->durable_lsn, 0);
/* Recovery-integrity verdict defaults to CLEAN (ha-commit-durability 3.1):
* a fresh store has no corruption. persist_recover overwrites this with the
* replay outcome (RECOVER_TRUNCATED only on a structural log corruption). */
p->recovery_verdict = RECOVER_CLEAN;
/* Sticky "unrecoverable, loudly fenced" flag starts clear (Req 2.4): a fresh
* store is promotable. Set by persist_recover_loud_fence_if_unrecoverable
* on an unrecoverable boot; cleared on catch-up (never self-exits). */
atomic_store(&p->recovery_unrecoverable, false);
/* Durable-advance wake hook starts unregistered (ha-commit-durability 2.2):
* the commit engine registers it in commit_engine_init. Until then every
* flusher durable advance is a no-op wake. */
p->on_durable_advance = NULL;
p->durable_cb_ctx = NULL;
/* 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) {
ERROR("persist: WAL path %s is on NFS — unsafe for fsync durability",
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);
pthread_mutex_init(&p->fsync_mutex, NULL);
pthread_cond_init(&p->fsync_cond, NULL);
repl_init(&p->repl); /* P2-4: fan-out replicator lifecycle */
WARN("WAL/snapshot stored unencrypted in %s — "
"ensure filesystem is encrypted if job payloads contain sensitive data",
p->wal_path);
return 0;
}
int persist_start(persist_t* p)
{
ctrl_metrics_init();
/* 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)
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)) {
/* Set the stop flag and wake the timer wait under fsync_mutex so the
* worker cannot miss the signal between its flag check and the wait. */
pthread_mutex_lock(&p->fsync_mutex);
atomic_store_explicit(&p->fsync_stop, 1, memory_order_release);
pthread_cond_signal(&p->fsync_cond);
pthread_mutex_unlock(&p->fsync_mutex);
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) {
/* Final synchronous flush-or-exit on stop: fdatasync everything
* appended so far and advance durable_lsn to the synced tip so a
* post-stop read reflects the durable state (the flusher thread is
* already joined and will not advance it again).
*
* FAIL-LOUD GATE (Req 1.1/1.2): advance the watermark ONLY when the
* final sync SUCCEEDS — on EIO/ENOSPC the bytes are not on stable
* storage, so durable_lsn must stay where it is and the failure is
* surfaced loudly. */
int rc = fdatasync(p->wal_fd);
if (rc == 0 && !flusher_fsync_forced_fail(p)) {
uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
uint64_t dur = atomic_load_explicit(&p->durable_lsn,
memory_order_acquire);
if (tip > dur)
atomic_store_explicit(&p->durable_lsn, tip,
memory_order_release);
} else {
int e = (rc != 0) ? errno : EIO;
ERROR("persist: final fdatasync on stop FAILED (errno=%d %s) — "
"durable_lsn left unchanged; appended-but-unsynced entries "
"are NOT durable",
e, strerror(e));
}
close(p->wal_fd);
p->wal_fd = -1;
}
pthread_mutex_unlock(&p->lock);
}
void persist_fini(persist_t* p)
{
repl_fini(&p->repl); /* P2-4: fan-out replicator lifecycle */
pthread_mutex_destroy(&p->lock);
pthread_mutex_destroy(&p->checkpoint_mutex);
pthread_cond_destroy(&p->checkpoint_done);
pthread_mutex_destroy(&p->fsync_mutex);
pthread_cond_destroy(&p->fsync_cond);
}
/* Forward declaration — defined after checkpoint internals */
static void schedule_checkpoint(persist_t* persist);
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;
/* Replicate to standbys — never replicate WAL_CHECKPOINT entries.
* last_replicated_crc is distinct from last_crc:
* last_crc includes checkpoint markers; last_replicated_crc skips them.
* This maintains a continuous CRC chain for standbys (design doc Part 2.3).
* Note: WAL_CHECKPOINT is blocked by the assert at function entry, so
* every entry that reaches here is a replicable entry. */
p->last_replicated_crc = hdr.crc;
repl_broadcast(&p->repl, &hdr, payload, len);
/* Rotate segment when full */
if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) {
if (rotate_segment(p) != 0) {
pthread_mutex_unlock(&p->lock);
return -EIO;
}
}
pthread_mutex_unlock(&p->lock);
/* Signal the group-commit flusher (ha-commit-durability 2.1): current_lsn
* advanced and the entry's bytes are pwritten, so durable_lsn < tip now
* holds (the level predicate). The signal is sent AFTER releasing p->lock
* and is level-triggered — if it races the flusher and is "lost", the
* flusher re-checks durable_lsn < current_lsn-1 under p->lock and flushes
* anyway (see fsync_worker). We do NOT fdatasync inline on this path; the
* flusher batches the sync across all pending appends. */
pthread_mutex_lock(&p->fsync_mutex);
pthread_cond_signal(&p->fsync_cond);
pthread_mutex_unlock(&p->fsync_mutex);
/* WAL record-count checkpoint trigger: schedule an async checkpoint once
* the record count reaches the configured threshold. The fetch_add is
* outside the lock — atomic_fetch_add is sufficient for the counter; the
* checkpoint itself is guarded by its own CAS inside schedule_checkpoint. */
if (atomic_fetch_add(&p->wal_record_count, 1) + 1 >=
p->config.max_wal_records) {
schedule_checkpoint(p);
}
return 0;
}
int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
const void* payload, uint32_t len)
{
if (!p)
return -EINVAL;
/* WAL_CHECKPOINT is a local-only marker, never streamed to standbys. */
if (type == WAL_CHECKPOINT)
return -EINVAL;
/* Reject oversized payloads BEFORE touching the segment. */
if (len > MAX_WAL_PAYLOAD_BYTES)
return -EMSGSIZE;
pthread_mutex_lock(&p->lock);
/* Idempotency / dedup guard: if this lsn is already persisted (current_lsn
* already advanced past it) treat the re-stream as a no-op. This keeps the
* write idempotent and avoids a duplicate entry / CRC-chain corruption when
* the primary re-sends an lsn the standby already wrote. */
if (lsn + 1 <= p->current_lsn) {
pthread_mutex_unlock(&p->lock);
return 0;
}
/* Open first segment lazily. open_new_segment names the file from
* current_lsn — for a fresh standby receiving lsn 1 first, current_lsn is
* 1 (reserve-lsn-0), so the segment is seg-...001.wal. */
if (p->wal_fd < 0) {
if (open_new_segment(p) != 0) {
pthread_mutex_unlock(&p->lock);
return -EIO;
}
}
/* Build entry header. Unlike persist_wal_async, lsn is GIVEN (not
* current_lsn++). term isn't streamed, so use our best-effort p->term.
* The CRC chain is SELF-computed (the streamed entry carries no crc):
* recovery validates an intra-segment chain (prev_crc == last_crc), so the
* standby's own writes must chain to each other — cross-node CRC equality
* is neither required nor possible. */
wal_entry_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.term = p->term;
hdr.lsn = 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;
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;
/* Advance current_lsn to one past the given lsn (max-idempotent — mirrors
* apply_wal_entry's monotonic advance). */
if (lsn + 1 > p->current_lsn)
p->current_lsn = lsn + 1;
/* Durability: fsync the segment BEFORE returning so the caller's ACK
* implies the entry is on stable storage (persist-before-ACK).
* persist_wal_async defers fsync to the background thread / persist_flush;
* here we sync inline because the entry feeds the primary's commit
* majority and must survive a standby restart. fdatasync is sufficient
* (entry bytes + segment metadata are pre-allocated by open_new_segment).
*
* Fail-loud rc-gate (ha-commit-durability Req 1.3, symmetric with the
* append-path gate at task 2.1): the ACK is sent by the caller ONLY when
* this returns success, and the ACK claims the entry is durable. So if
* the fdatasync FAILS (EIO/ENOSPC, or the forced-fail test seam), we must
* NOT return success — we log loudly and return -EIO so the caller's
* existing error path (ha_apply_single_entry) does NOT apply or ACK. */
int fsrc = fdatasync(p->wal_fd);
#ifdef XCE_TESTING
atomic_fetch_add_explicit(&p->test_replicated_fdatasync_count, 1,
memory_order_relaxed);
#endif
if (fsrc != 0 || flusher_fsync_forced_fail(p)) {
int e = (fsrc != 0) ? errno : EIO;
ERROR("persist: persist_wal_write_replicated fdatasync FAILED for "
"lsn=%" PRIu64 " (errno=%d %s) — entry NOT durable; returning "
"error so the standby does NOT ACK", lsn, e, strerror(e));
pthread_mutex_unlock(&p->lock);
return -EIO;
}
/* Rotate segment when full (same threshold as the primary append). */
if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) {
if (rotate_segment(p) != 0) {
pthread_mutex_unlock(&p->lock);
return -EIO;
}
}
/* NOTE: no repl_broadcast — standbys never re-replicate. */
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;
/* persist_flush is a synchronous durability point (tests, clean shutdown):
* it fdatasync'd everything appended so far, so advance durable_lsn to the
* synced tip too. Without this, an explicit-flush-only path (slow flusher
* cadence) would leave durable_lsn lagging even though the bytes are on
* disk. Monotonic max; preserves durable_lsn <= current_lsn - 1 (tip read
* under the same p->lock). Gated on rc == 0 (fail-loud, Req 1.1/1.2): a
* failed flush must not advance the watermark. */
if (rc == 0 && !flusher_fsync_forced_fail(p)) {
uint64_t tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
uint64_t dur = atomic_load_explicit(&p->durable_lsn,
memory_order_acquire);
if (tip > dur)
atomic_store_explicit(&p->durable_lsn, tip, memory_order_release);
} else if (rc != 0 || flusher_fsync_forced_fail(p)) {
int e = (rc != 0) ? errno : EIO;
ERROR("persist: persist_flush fdatasync FAILED (errno=%d %s) — "
"durable_lsn not advanced; flushed bytes are NOT durable",
e, strerror(e));
}
pthread_mutex_unlock(&p->lock);
return rc;
}
uint64_t persist_durable_lsn(persist_t* p)
{
return atomic_load_explicit(&p->durable_lsn, memory_order_acquire);
}
uint64_t persist_durable_floor(persist_t* p)
{
/* The durable freshness floor (durable last-log index) IS the durable
* watermark: persist_recover re-derives it from the recovered WAL tip
* (ha-commit-durability task 3.4), and the flusher advances it on every
* successful fdatasync — so a thin wrapper avoids a redundant field. NULL
* guard keeps the accessor safe for the election consumers (task 4.2). */
return p ? atomic_load_explicit(&p->durable_lsn, memory_order_acquire) : 0;
}
void persist_set_durable_cb(persist_t* p, void (*fn)(void* ctx), void* ctx)
{
if (!p)
return;
/* Set ctx before fn so the flusher (which checks fn != NULL) never sees a
* live callback with a stale/uninitialised ctx. Registration happens in
* commit_engine_init, before the engine is reachable by any ACK path. */
p->durable_cb_ctx = ctx;
p->on_durable_advance = fn;
}
#ifdef XCE_TESTING
void persist_test_set_force_fsync_fail(persist_t* p, int fail)
{
if (p)
atomic_store_explicit(&p->test_force_fsync_fail, fail ? 1 : 0,
memory_order_release);
}
unsigned persist_test_replicated_fdatasync_count(persist_t* p)
{
if (!p)
return 0;
return atomic_load_explicit(&p->test_replicated_fdatasync_count,
memory_order_relaxed);
}
pthread_t persist_test_last_fdatasync_thread(persist_t* p)
{
/* If no flusher fdatasync has advanced durable_lsn yet, return the
* caller's own id so an unprimed read never spuriously looks off-thread.
* After a flush, return the recorded flusher thread id (Req 1.6). */
if (!p ||
atomic_load_explicit(&p->test_last_fdatasync_thread_set,
memory_order_acquire) == 0)
return pthread_self();
return p->test_last_fdatasync_thread;
}
/* ------------------------------------------------------------------ *
* Durability-verification test seam (ha-commit-durability Req 3.2/3.3)*
* *
* Re-read + CRC-validate the WAL segments from stable storage and *
* return the highest fully-persisted LSN — independent of the *
* in-memory current_lsn / committed_lsn watermarks. Mirrors the *
* segment-scan + CRC-chain validation of replay_wal(), but is strictly *
* read-only: it applies no entries, fires no hooks, and mutates no *
* persist_t state, so it cannot change the externally observable *
* acknowledgment semantics (Req 3.3). *
* *
* Stops at the first corruption / short read / pre-allocated zero *
* region (exactly like recovery), so only entries genuinely on stable *
* storage are counted. Returns the highest such LSN, or 0 when nothing *
* is durable. *
* ------------------------------------------------------------------ */
uint64_t persist_test_durable_lsn_on_disk(persist_t* p)
{
if (!p)
return 0;
DIR* d = opendir(p->wal_path);
if (!d)
return 0; /* no WAL dir → nothing durable */
/* Collect segment start LSNs (same bound as replay_wal). */
uint64_t seg_lsns[4096];
int nseg = 0;
struct dirent* e;
while ((e = readdir(d)) != NULL &&
nseg < (int)(sizeof(seg_lsns) / sizeof(seg_lsns[0]))) {
if (strncmp(e->d_name, "seg-", 4) != 0)
continue;
uint64_t seg_start = 0;
if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &seg_start) != 1)
continue;
seg_lsns[nseg++] = seg_start;
}
closedir(d);
if (nseg == 0)
return 0;
/* Insertion sort by start LSN (segment count is tiny). */
for (int i = 1; i < nseg; i++) {
uint64_t key = seg_lsns[i];
int j = i - 1;
while (j >= 0 && seg_lsns[j] > key) {
seg_lsns[j + 1] = seg_lsns[j];
j--;
}
seg_lsns[j + 1] = key;
}
uint64_t highest_durable = 0;
uint32_t last_crc = WAL_CRC_NONE;
for (int si = 0; si < nseg; si++) {
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path),
WAL_SEG_FMT, p->wal_path, seg_lsns[si]);
int fd = open(seg_path, O_RDONLY);
if (fd < 0)
continue;
uint32_t seg_off = 0;
for (;;) {
wal_entry_hdr_t hdr;
ssize_t hn = pread(fd, &hdr, sizeof(hdr), (off_t)seg_off);
if (hn == 0)
break; /* clean end of segment */
if (hn < (ssize_t)sizeof(hdr))
break; /* truncated / pre-allocated zeros */
/* Pre-allocated zero region: no valid entry is all-zero. */
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0)
break;
/* CRC chain validation — identical rule to replay_wal. */
if (seg_off == 0) {
if (hdr.prev_crc != WAL_CRC_NONE && last_crc != WAL_CRC_NONE &&
hdr.prev_crc != last_crc)
break;
} else {
if (hdr.prev_crc != last_crc)
break;
}
/* Verify entry CRC over header + payload. */
uint32_t expected_crc =
crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE);
uint8_t* payload = NULL;
if (hdr.payload_len > 0) {
payload = malloc(hdr.payload_len);
if (!payload) {
close(fd);
return highest_durable;
}
ssize_t pn = pread(fd, payload, hdr.payload_len,
(off_t)(seg_off + sizeof(hdr)));
if (pn != (ssize_t)hdr.payload_len) {
free(payload);
break;
}
expected_crc =
crc32(expected_crc, (const Bytef*)payload, hdr.payload_len);
}
if (hdr.crc != expected_crc) {
free(payload);
break;
}
free(payload);
/* Entry is intact and durable on disk. */
last_crc = hdr.crc;
if (hdr.lsn > highest_durable)
highest_durable = hdr.lsn;
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(fd);
}
return highest_durable;
}
#endif /* XCE_TESTING */
/* ------------------------------------------------------------------ */
/* 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 the
* currently-open segment's start LSN. We parse the active segment
* filename (seg_path) to get its start LSN and never delete that file,
* even if its start_lsn < checkpoint_lsn. */
static void trim_wal_segments(persist_t* p)
{
/* Extract the start LSN of the currently-open segment from seg_path.
* seg_path has the form "<wal_path>/seg-<hex16>.wal".
* If no segment is open yet (wal_fd < 0), skip trimming. */
if (p->wal_fd < 0 || p->seg_path[0] == '\0')
return;
const char* seg_basename = strrchr(p->seg_path, '/');
if (!seg_basename) return;
seg_basename++; /* skip the '/' */
uint64_t active_seg_lsn = 0;
if (sscanf(seg_basename, "seg-%" SCNx64 ".wal", &active_seg_lsn) != 1)
return;
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;
/* Never delete the currently-open segment; only trim older ones. */
if (seg_start_lsn < active_seg_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]);
}
/* ------------------------------------------------------------------ */
/* Async checkpoint scheduling (schedule_checkpoint + task fn) */
/* ------------------------------------------------------------------ */
/* checkpoint_task_fn: executed by delaymgr worker thread.
*
* The CAS in schedule_checkpoint already holds the "in_progress" slot (=1).
* We must clear it before calling persist_checkpoint(), which has its own
* atomic_exchange guard. After persist_checkpoint() returns (it resets the
* flag to 0), we also reset wal_record_count so the counter starts fresh. */
static int checkpoint_task_fn(void* arg)
{
persist_t* persist = (persist_t*)arg;
/* Release the "scheduled" slot so persist_checkpoint's guard can acquire */
atomic_store(&persist->checkpoint_in_progress, 0);
persist_checkpoint(persist);
/* Reset the record counter so the next checkpoint window starts clean.
* Done after persist_checkpoint returns — the flag is already 0 by now. */
atomic_store(&persist->wal_record_count, 0);
return 0;
}
/* schedule_checkpoint: idempotent async checkpoint request via delaymgr.
* Safe to call from any thread. Uses atomic CAS to ensure at most one
* checkpoint is scheduled or running at a time.
*
* If cluster is NULL (unit tests without a live delaymgr) the function
* falls back to a synchronous checkpoint. */
static void schedule_checkpoint(persist_t* persist)
{
int expected = 0;
if (!atomic_compare_exchange_strong(&persist->checkpoint_in_progress,
&expected, 1))
return; /* already scheduled or running — idempotency guard */
if (persist->cluster == NULL) {
/* Unit-test / no-cluster path: run synchronously and reset. */
atomic_store(&persist->checkpoint_in_progress, 0);
persist_checkpoint(persist);
atomic_store(&persist->wal_record_count, 0);
return;
}
/* Schedule via delaymgr with 1 s coalescing delay, single execution */
delay_task(&persist->cluster->delaymgr, 1, 1,
checkpoint_task_fn, persist);
}
/* ------------------------------------------------------------------ */
/* 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;
}
/* ------------------------------------------------------------------ */
/* Recovery internals */
/* ------------------------------------------------------------------ */
/* P0 sentinel format: "XCESNAP\0" (8 bytes) + checkpoint_lsn (8 bytes LE).
* Returns 0 on success and sets p->checkpoint_lsn / p->current_lsn.
* Returns -EINVAL on bad magic, -EIO on short read, -errno on open failure. */
static int load_snapshot(persist_t* p, const char* path, uint64_t lsn)
{
(void)lsn; /* lsn is encoded in the file itself — we re-read it */
int fd = open(path, O_RDONLY);
if (fd < 0)
return -errno;
char header[16];
ssize_t n = read(fd, header, sizeof(header));
close(fd);
if (n < 16)
return -EIO;
if (memcmp(header, "XCESNAP\0", 8) != 0)
return -EINVAL;
/* Extract checkpoint_lsn from bytes 8-15 (little-endian on this arch) */
uint64_t stored_lsn;
memcpy(&stored_lsn, header + 8, 8);
p->checkpoint_lsn = stored_lsn;
p->current_lsn = stored_lsn;
return 0;
}
/* Public wrapper for load_snapshot — exposed via persist.h for ha.c (Task 3.5c). */
int persist_load_snapshot(persist_t *p, const char *path, uint64_t lsn)
{
return load_snapshot(p, path, lsn);
}
/* P2-2: Standby WAL replay — apply a single replicated entry to in-memory state.
*
* No hooks fire on standbys; the hook system is empty until promotion.
*
* Returns 0 on success (including skip for checkpoint / unknown types),
* negative errno on error (NULL args, oversized payload). */
int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payload)
{
if (!p || !hdr)
return -EINVAL;
/* Reject oversized payloads BEFORE any allocation */
if (hdr->payload_len > MAX_WAL_PAYLOAD_BYTES) {
ERROR("apply_wal_entry: payload_len %u exceeds MAX_WAL_PAYLOAD_BYTES %u",
hdr->payload_len, MAX_WAL_PAYLOAD_BYTES);
return -EMSGSIZE;
}
/* WAL_CHECKPOINT (100) is a local-only marker — standbys never receive it
* via replication. Skip silently; do NOT advance LSN. */
if (hdr->type == WAL_CHECKPOINT)
return 0;
/* Unknown types above WAL_ENTRY_TYPE_MAX: log WARN and skip for forward
* compatibility — an older controller can boot on WAL written by a newer
* one without hard-failing on unrecognised entry types. */
if (hdr->type > WAL_ENTRY_TYPE_MAX) {
WARN("apply_wal_entry: unknown type %u > WAL_ENTRY_TYPE_MAX %u"
" (lsn=%" PRIu64 ") — skipping for forward compatibility",
hdr->type, WAL_ENTRY_TYPE_MAX, hdr->lsn);
return 0;
}
/* Advance LSN chain: track the highest replicated LSN seen. */
pthread_mutex_lock(&p->lock);
if (hdr->lsn + 1 > p->current_lsn)
p->current_lsn = hdr->lsn + 1;
p->last_crc = hdr->crc;
pthread_mutex_unlock(&p->lock);
if (!p->cluster)
return 0; /* unit tests without a full cluster — skip entity restore */
switch (hdr->type) {
case WAL_JOB_SUBMIT:
/* Payload: Cap'n Proto serialized xce_job_t. Thread the entry LSN
* through so the restored job is tagged for the egress watermark
* (ha-convergence-recovery 3.3). */
restore_job(p->cluster, 0, payload, hdr->payload_len, hdr->lsn);
break;
case WAL_JOB_START:
case WAL_JOB_FINISH:
case WAL_JOB_FAIL:
case WAL_JOB_STOP:
case WAL_JOB_RESUME:
case WAL_JOB_UPDATE:
case WAL_JOB_CKPT:
case WAL_JOB_PREASON:
case WAL_JOB_RUSAGE:
/* Payload: wal_jobstate_t {jobid, state} */
if (hdr->payload_len >= sizeof(wal_jobstate_t)) {
const wal_jobstate_t* js = (const wal_jobstate_t*)payload;
restore_job_state(p->cluster, js->jobid, js->state);
}
break;
case WAL_WORKER_CTRL:
/* Payload: [uint32_t detail][capnp worker] */
if (hdr->payload_len > WORKER_WAL_HDR_SIZE)
worker_from_capnp(p->cluster, payload, hdr->payload_len);
break;
case WAL_QUEUE_CTRL:
/* Payload: [wal_queuehdr_t][name NUL-terminated] */
if (hdr->payload_len > sizeof(wal_queuehdr_t)) {
const wal_queuehdr_t* qh = (const wal_queuehdr_t*)payload;
const char* name = (const char*)(payload + sizeof(wal_queuehdr_t));
restore_queue_state(p->cluster, name, qh->status);
}
break;
case WAL_NOOP:
/* Internal promotion no-op: empty payload, no entity restore, no
* hooks. The common-path current_lsn advance (above) is its only
* effect — recognised explicitly here rather than via default. */
break;
default:
break;
}
return 0;
}
/* Read WAL segment files in LSN order; replay entries with lsn > from_lsn.
*
* Segment filename format: seg-<start_lsn_hex16>.wal
* Entry format: wal_entry_hdr_t (40 bytes) + payload (hdr.payload_len bytes).
*
* CRC chain: hdr.prev_crc must match last_crc (WAL_CRC_NONE for first entry
* in a segment). If prev_crc mismatches, stop at last good entry and log WARN.
*
* Recovery-integrity verdict (ha-commit-durability task 3.1, Req 2.1):
* *verdict_out (when non-NULL) is set to RECOVER_TRUNCATED iff replay stopped
* on a STRUCTURAL corruption — a CRC chain break, a CRC mismatch, or a
* short/partial entry read — and RECOVER_CLEAN otherwise (clean EOF,
* pre-allocated-zero tail, empty WAL, or all entries replayed). The verdict is
* about the STRUCTURE of the local log, not about being behind the cluster: a
* clean log that simply ends below the cluster high-water is still CLEAN.
*
* Returns number of entries replayed (>= 0) or negative errno on I/O error. */
static int replay_wal(persist_t* p, uint64_t from_lsn,
recover_verdict_t* verdict_out)
{
/* Default to CLEAN; only a structural corruption flips it to TRUNCATED. */
if (verdict_out)
*verdict_out = RECOVER_CLEAN;
/* --- Collect and sort segment filenames by start LSN --- */
DIR* d = opendir(p->wal_path);
if (!d)
return 0; /* no WAL dir → first boot, nothing to replay (CLEAN) */
/* Collect up to 4096 segment names */
uint64_t seg_lsns[4096];
int nseg = 0;
struct dirent* e;
while ((e = readdir(d)) != NULL && nseg < (int)(sizeof(seg_lsns)/sizeof(seg_lsns[0]))) {
if (strncmp(e->d_name, "seg-", 4) != 0) continue;
uint64_t seg_start = 0;
if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &seg_start) != 1) continue;
seg_lsns[nseg++] = seg_start;
}
closedir(d);
if (nseg == 0)
return 0;
/* Simple insertion sort — segment count is tiny in practice */
for (int i = 1; i < nseg; i++) {
uint64_t key = seg_lsns[i];
int j = i - 1;
while (j >= 0 && seg_lsns[j] > key) {
seg_lsns[j + 1] = seg_lsns[j];
j--;
}
seg_lsns[j + 1] = key;
}
/* --- Replay entries from each segment --- */
int replayed = 0;
uint32_t last_crc = WAL_CRC_NONE;
for (int si = 0; si < nseg; si++) {
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path),
WAL_SEG_FMT, p->wal_path, seg_lsns[si]);
int fd = open(seg_path, O_RDONLY);
if (fd < 0) {
WARN("persist_recover: cannot open segment %s: %s",
seg_path, strerror(errno));
continue;
}
uint32_t seg_off = 0;
for (;;) {
/* Read header */
wal_entry_hdr_t hdr;
ssize_t hn = pread(fd, &hdr, sizeof(hdr), (off_t)seg_off);
if (hn == 0)
break; /* clean end of segment (CLEAN) */
if (hn < (ssize_t)sizeof(hdr)) {
/* A partial header (or read error) is a SHORT READ — the log
* stops mid-entry, so it cannot be reconstructed contiguously
* (RECOVER_TRUNCATED, Req 2.1). Distinct from clean EOF above. */
if (hn < 0)
WARN("persist_recover: read error in %s at offset %u: %s",
seg_path, seg_off, strerror(errno));
else
WARN("persist_recover: short header read in %s at offset %u "
"(got %zd of %zu bytes); stopping replay",
seg_path, seg_off, hn, sizeof(hdr));
if (verdict_out)
*verdict_out = RECOVER_TRUNCATED;
break;
}
/* Detect pre-allocated zero region (no valid entry starts with
* all-zero header: lsn == 0 && type == 0 && payload_len == 0).
* This is the normal fallocate'd tail after the last valid entry —
* a clean stop, NOT corruption (stays RECOVER_CLEAN). */
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0)
break;
/* CRC chain validation: prev_crc must match last_crc in segment */
if (seg_off == 0) {
/* First entry in this segment: prev_crc must be WAL_CRC_NONE
* (or we accept any value if last_crc is also WAL_CRC_NONE,
* e.g. across a segment boundary after checkpoint trim). */
if (hdr.prev_crc != WAL_CRC_NONE && last_crc != WAL_CRC_NONE &&
hdr.prev_crc != last_crc) {
WARN("persist_recover: CRC chain break at start of %s "
"(prev_crc=%08x expected=%08x); stopping replay",
seg_path, hdr.prev_crc, last_crc);
if (verdict_out)
*verdict_out = RECOVER_TRUNCATED;
break;
}
} else {
if (hdr.prev_crc != last_crc) {
WARN("persist_recover: CRC chain break in %s at offset %u "
"(prev_crc=%08x expected=%08x); stopping replay at last good entry",
seg_path, seg_off, hdr.prev_crc, last_crc);
if (verdict_out)
*verdict_out = RECOVER_TRUNCATED;
break;
}
}
/* Verify entry CRC */
uint32_t expected_crc = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE);
uint8_t* payload = NULL;
if (hdr.payload_len > 0) {
payload = malloc(hdr.payload_len);
if (!payload) {
close(fd);
return -ENOMEM;
}
ssize_t pn = pread(fd, payload, hdr.payload_len,
(off_t)(seg_off + sizeof(hdr)));
if (pn != (ssize_t)hdr.payload_len) {
WARN("persist_recover: short payload read in %s at offset %u",
seg_path, seg_off);
if (verdict_out)
*verdict_out = RECOVER_TRUNCATED;
free(payload);
break;
}
expected_crc = crc32(expected_crc, (const Bytef*)payload, hdr.payload_len);
}
if (hdr.crc != expected_crc) {
WARN("persist_recover: CRC mismatch in %s at offset %u "
"(got=%08x expected=%08x); stopping replay",
seg_path, seg_off, hdr.crc, expected_crc);
if (verdict_out)
*verdict_out = RECOVER_TRUNCATED;
free(payload);
break;
}
/* Advance segment-local CRC */
last_crc = hdr.crc;
/* Skip entries already covered by the snapshot */
if (hdr.lsn <= from_lsn) {
free(payload);
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
continue;
}
/* Skip WAL_CHECKPOINT markers (local bookkeeping only) */
if (hdr.type == WAL_CHECKPOINT) {
free(payload);
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
/* Advance current_lsn past the marker */
if (hdr.lsn + 1 > p->current_lsn)
p->current_lsn = hdr.lsn + 1;
continue;
}
/* Unknown future entry types: log and skip */
if (hdr.type > WAL_CHECKPOINT) {
WARN("persist_recover: unknown WAL entry type %u at LSN %lu; skipping",
hdr.type, (unsigned long)hdr.lsn);
free(payload);
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
continue;
}
/* Apply the entry (P0 stub: no-op) */
apply_wal_entry(p, &hdr, payload);
free(payload);
/* Advance current_lsn to one past this entry */
if (hdr.lsn + 1 > p->current_lsn)
p->current_lsn = hdr.lsn + 1;
replayed++;
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(fd);
}
/* Update last_crc so the next WAL write chains correctly */
pthread_mutex_lock(&p->lock);
p->last_crc = last_crc;
pthread_mutex_unlock(&p->lock);
return replayed;
}
/* ------------------------------------------------------------------ */
/* Public: persist_recover */
/* ------------------------------------------------------------------ */
int persist_recover(persist_t* p)
{
/* --- Step 1: find latest snapshot (highest LSN filename) --- */
char snap_file[4096] = "";
uint64_t snap_lsn = 0;
DIR* d = opendir(p->snap_path);
if (d) {
struct dirent* e;
while ((e = readdir(d)) != NULL) {
size_t len = strlen(e->d_name);
if (len < 6 || strcmp(e->d_name + len - 6, ".capnp") != 0)
continue;
uint64_t lsn = 0;
if (sscanf(e->d_name, "snap-%" SCNx64 ".capnp", &lsn) != 1)
continue;
if (lsn > snap_lsn) {
snap_lsn = lsn;
snprintf(snap_file, sizeof(snap_file),
"%s/%s", p->snap_path, e->d_name);
}
}
closedir(d);
}
/* --- Step 2: load snapshot (if any) --- */
if (snap_lsn > 0 && snap_file[0] != '\0') {
int rc = load_snapshot(p, snap_file, snap_lsn);
if (rc != 0) {
ERROR("persist_recover: failed to load snapshot %s: %s",
snap_file, strerror(-rc));
return rc;
}
/* checkpoint_lsn and current_lsn set by load_snapshot */
}
/* --- Step 3: replay WAL entries after checkpoint_lsn --- */
/* Recovery-integrity verdict (ha-commit-durability task 3.1, Req 2.1):
* replay distinguishes a clean log end from a corruption-truncated end.
* We CLASSIFY + record the verdict here; we do NOT FATAL (task 3.3) and do
* NOT fence promotion (task 4.3 consumes p->recovery_verdict). A clean
* replay over a loaded snapshot-sentinel (no entries below the sentinel) is
* CLEAN — the sentinel established the committed position; replay only adds
* entries above it. */
recover_verdict_t verdict = RECOVER_CLEAN;
int replayed = replay_wal(p, p->checkpoint_lsn, &verdict);
if (replayed < 0)
return replayed;
p->recovery_verdict = verdict;
if (verdict == RECOVER_TRUNCATED)
WARN("persist_recover: RECOVER_TRUNCATED — local WAL is structurally "
"corrupt (CRC break / mismatch / short read); member marked "
"recovery-truncated (non-promotable until caught up)");
/* --- Step 3b: re-derive the durable floor from the recovered WAL tip
* (ha-commit-durability task 3.4, Req 2.3) ---
*
* Publish the member's durable floor (the durable last-log index consumed
* by the Raft freshness gate, task 4.2) as the highest INTACT replayed LSN.
* replay_wal advanced current_lsn to one past the last intact entry — incl.
* stopping at a corruption boundary on TRUNCATED — so current_lsn - 1 is
* exactly that intact WAL tip, and a snapshot-sentinel load leaves it at the
* sentinel/replayed tip. We store it into durable_lsn because the recovered
* on-disk entries are GENUINELY durable (each was CRC-validated and read
* from stable storage by replay_wal), so the freshness advertisement is now
* backed by durable state rather than a volatile in-memory position.
*
* Crux (Req 2.3 — a truncated member cannot over-advertise): this is an
* ASSIGN to the recovered tip, NOT a max() with whatever durable_lsn held.
* A fresh recover starts from durable_lsn=0 (persist_init), so on a
* truncating restart the floor lands at the LOWER intact tip — never a
* stale higher in-memory value and never the commit position
* (committed_lsn). The recovered tip is, by construction, <= current_lsn-1,
* preserving the durable_lsn <= current_lsn-1 invariant. The post-election
* safe-commit point continues to come from voter replies, not this floor
* (CLAUDE.md). */
uint64_t recovered_tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
atomic_store_explicit(&p->durable_lsn, recovered_tip, memory_order_release);
/* --- Step 4: retention pass (TTL check) — no-op in P0 (no entities) --- */
return 0;
}
/* Recovery-integrity accessors (ha-commit-durability task 3.1, Req 2.1). */
recover_verdict_t persist_recovery_verdict(persist_t* p)
{
return p ? p->recovery_verdict : RECOVER_CLEAN;
}
bool persist_recovery_truncated(persist_t* p)
{
return p && p->recovery_verdict == RECOVER_TRUNCATED;
}
/* ------------------------------------------------------------------ *
* Boot-policy loud-fence decision (ha-commit-durability task 3.3, Req 2.4;
* never-self-exit rework 2026-06-18).
*
* Implements the FENCE_UNRECOVERABLE terminal of the Boot Policy Decision Table
* (design.md). The decision is a PURE function of its inputs; the action (loud
* operator-visible log + set the sticky non-promotable flag) is separated into
* persist_recover_loud_fence_if_unrecoverable so the decision can be unit-tested
* in isolation. A reachable healthy peer NEVER yields FENCE_UNRECOVERABLE
* (rows 3, 5); it is reserved for a member that cannot recover locally AND
* cannot catch up (no healthy peer, rows 4, 4b).
*
* CRITICAL: this terminal is a LOUD, PERMANENT, NON-promotable fence — the
* process STAYS ALIVE and keeps retrying catch-up. There is NO abort()/exit():
* a controller MUST NEVER self-terminate as a designed response to an
* operational condition (user invariant, 2026-06-18). A self-exited controller
* needs manual admin recovery and degrades quorum meanwhile; one that stays
* alive and keeps retrying self-heals when a peer returns (Req 2.6).
* ------------------------------------------------------------------ */
recover_decision_t persist_recover_decision(const recover_policy_inputs_t* in)
{
if (in == NULL)
return RECOVER_DECISION_ELIGIBLE; /* defensive: nothing to fence */
bool truncated = (in->verdict == RECOVER_TRUNCATED);
/* Rows 1 & 6: a structurally CLEAN log that is NOT behind the cluster
* high-water mark is a normal promotable member — no fence. */
if (!truncated && !in->durable_below_hwm)
return RECOVER_DECISION_ELIGIBLE;
/* From here the member is truncated OR behind the HWM — it must NOT lead
* as-is. A truncated log is by definition not provably at/above the HWM
* (design.md note), so it is treated as behind for fence purposes. */
/* Rows 2, 3, 5: a healthy peer is reachable. Catch up from it and regain
* eligibility (task 4.3). A reachable peer NEVER yields the unrecoverable
* fence — even for a truncated log or a legitimate snapshot-sentinel. If
* catch-up has already succeeded the member is eligible; otherwise it stays
* fenced (4.3 drives the catch-up + release). */
if (in->healthy_peer_reachable) {
if (in->catch_up_succeeded)
return RECOVER_DECISION_ELIGIBLE;
return RECOVER_DECISION_FENCE;
}
/* No healthy peer. If catch-up did not reach a committed-consistent state,
* the member can neither recover locally nor catch up right now — rows 4 &
* 4b: a LOUD, PERMANENT, NON-promotable fence (NOT process exit). It stays
* alive and keeps retrying; the fence releases when a peer returns
* (Req 2.4/2.6). */
if (!in->catch_up_succeeded)
return RECOVER_DECISION_FENCE_UNRECOVERABLE;
/* No peer but catch-up is reported succeeded (e.g. already at/above after a
* local catch-up): eligible. */
return RECOVER_DECISION_ELIGIBLE;
}
bool persist_recovery_unrecoverable(persist_t* p)
{
return p && atomic_load_explicit(&p->recovery_unrecoverable,
memory_order_acquire);
}
void persist_clear_recovery_unrecoverable(persist_t* p)
{
if (!p)
return;
/* Clear only on a genuine transition so the loud "fence released" log fires
* once, not on every catch-up tick. */
if (atomic_exchange_explicit(&p->recovery_unrecoverable, false,
memory_order_acq_rel)) {
INFO("persist_recover: unrecoverable fence RELEASED — member reached a "
"committed-consistent state via catch-up; promotable again "
"(ha-commit-durability Req 2.6)");
}
}
recover_decision_t
persist_recover_loud_fence_if_unrecoverable(persist_t* p,
const recover_policy_inputs_t* in)
{
recover_decision_t decision = persist_recover_decision(in);
if (decision != RECOVER_DECISION_FENCE_UNRECOVERABLE)
return decision;
/* Loud, operator-visible alert: log the truncation boundary
* (durable_floor_lsn) and the cluster high-water mark so an operator sees
* exactly where the local log stops short of the cluster's committed state.
* The member STAYS ALIVE, stays non-promotable, and keeps retrying catch-up
* (Req 2.4) — it NEVER self-exits. */
const char* why = (in->verdict == RECOVER_TRUNCATED)
? "recovery-truncated log (structural corruption)"
: "log behind the cluster committed high-water mark";
ERROR("persist_recover: UNRECOVERABLE — %s and no healthy peer reachable to "
"catch up; truncation boundary durable_floor_lsn=%" PRIu64
" < cluster committed high-water hwm=%" PRIu64 ". Staying ALIVE and "
"PERMANENTLY non-promotable (refusing to serve divergent state) and "
"RETRYING catch-up indefinitely — will self-heal when a healthy peer "
"returns; operator attention may be needed "
"(ha-commit-durability Req 2.4)",
why, in->durable_floor_lsn, in->cluster_hwm_lsn);
/* Set the sticky non-promotable flag so the promotion gate refuses
* leadership as belt-and-suspenders until catch-up clears it (Req 2.4/2.6).
* p may be NULL for pure-classification callers that only want the loud log. */
if (p != NULL)
atomic_store_explicit(&p->recovery_unrecoverable, true,
memory_order_release);
return RECOVER_DECISION_FENCE_UNRECOVERABLE;
}
/* Advance committed_lsn to `written_lsn` on a majority ACK — but ONLY if that
* entry is already durable on local stable storage (ha-commit-durability task
* 2.4, Req 1.1/1.2). This is the legacy (engine-less / LSN-0) sibling of the
* commit engine's durable self-cap (commit.c majority_candidate_locked): a
* majority ACK no longer implies the local copy is fsync'd, because group-
* commit (task 2.1) deferred the append fsync to the demand-driven flusher. So
* advancing committed_lsn here on the ACK alone would reintroduce the bug-661
* ACK-without-durability hazard — committed could be durable on a single
* member and lost across failover. We gate the store on written_lsn <=
* durable_lsn so committed_lsn ⇒ durable-on-this-member on every path.
* committed_lsn is monotonic, so a max() store; the durable cap can only delay
* the advance until the flusher syncs the entry (the durable-advance commit
* wake re-drives the engine path; the legacy path is re-evaluated on the next
* sync_append). */
static void sync_append_legacy_commit_store(persist_t* p, uint64_t written_lsn)
{
uint64_t durable = atomic_load_explicit(&p->durable_lsn,
memory_order_acquire);
/* DURABILITY PRECONDITION: never advance committed past what is durable
* locally. The LSN-0 first-entry edge stays valid (0 <= 0). */
if (written_lsn > durable)
return;
uint64_t cur = atomic_load_explicit(&p->committed_lsn,
memory_order_acquire);
if (written_lsn <= cur)
return; /* monotonic: committed_lsn never regresses */
atomic_store_explicit(&p->committed_lsn, written_lsn, memory_order_release);
#ifdef XCE_TESTING
/* INVARIANT (debug guard): committed_lsn must never exceed durable_lsn on
* any commit advance. A future caller reintroducing ACK-without-durability
* trips this. */
assert(atomic_load_explicit(&p->committed_lsn, memory_order_acquire)
<= atomic_load_explicit(&p->durable_lsn, memory_order_acquire));
#endif
}
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 (rc != 0) {
if (sync_rc_out)
*sync_rc_out = rc;
return rc;
}
/* Majority-commit replication (Req 5.1–5.3; ha-commit-engine Req 1.1–1.3).
* persist_wal_async appends to the page cache and signals the demand-driven
* flusher; it no longer fsyncs inline (group-commit, task 2.1). Durability
* is therefore established by the flusher advancing durable_lsn, and a
* commit position advances only once the entry is durable locally — the
* engine path via the durable self-cap (commit.c), the legacy path below
* via sync_append_legacy_commit_store's written_lsn <= durable_lsn gate.
* We use current_lsn - 1 because persist_wal_async already incremented it. */
uint64_t written_lsn;
pthread_mutex_lock(&p->lock);
written_lsn = p->current_lsn - 1;
pthread_mutex_unlock(&p->lock);
int sync_err;
/* Engine-wired path (ha-commit-engine task 5.3): delegate the commit
* decision to the commit engine through the walAck adapter. The engine
* is the single committed_lsn writer (no store here), and its
* n_peers<=1 self-quorum replaces the old has_standbys → -ENODEV
* degraded special case (Req 6.3).
*
* LSN-0 edge (ACCEPTED interim behavior — spec Implementation Notes):
* written_lsn is passed unconditionally, but the FIRST-EVER WAL entry
* of a fresh cluster is LSN 0 and commit_submit rejects lsn==0, so the
* adapter's entry_lsn != 0 guard routes that one entry to the legacy
* branch below. This happens once per cluster lifetime;
* ha-single-leader-role's promotion no-op entry eliminates it in
* practice. */
if (p->repl.ce != NULL && written_lsn != 0) {
sync_err = repl_wait_majority_ack(&p->repl, written_lsn,
p->config.sync_timeout_ms,
(uint8_t)OPCode_walAck);
if (sync_err != 0)
CTRL_METRIC_INC(ctrl_repl_degraded_count);
} else {
/* Residual non-engine branch: engine-less (ce == NULL) unit-test
* contexts plus the LSN-0 first entry. The legacy connectivity-
* based counting was retired by ha-commit-barrier-rollout task 2.2:
* repl_wait_majority_ack now confirms LSN 0 immediately (documented
* once-per-lifetime edge) and surfaces engine-less walAck waits as
* -EINVAL via commit_submit, so the committed_lsn store below only
* fires for the LSN-0 case (a no-op store of 0). */
mylock_lock(&p->repl.lock);
int has_standbys = (p->repl.nstandbys > 0);
mylock_unlock(&p->repl.lock);
if (!has_standbys) {
/* No standbys registered → degraded mode (no quorum to wait for) */
sync_err = -ENODEV;
} else {
sync_err = repl_wait_majority_ack(&p->repl, written_lsn,
p->config.sync_timeout_ms,
(uint8_t)OPCode_walAck);
if (sync_err == 0)
/* Durable-gated commit advance (task 2.4): advance committed_lsn
* only if written_lsn is durable locally — never on the ACK
* alone (group-commit removed the inline append fsync). */
sync_append_legacy_commit_store(p, written_lsn);
else
CTRL_METRIC_INC(ctrl_repl_degraded_count);
}
}
if (sync_rc_out)
*sync_rc_out = sync_err;
/* sync_err < 0 — replication failed.
* -ENODEV: legacy degraded mode (no standbys, engine-less) — acceptable
* for forward progress.
* -ETIMEDOUT: no per-entry majority within the deadline — the write
* must be rejected to preserve the majority-consistency
* invariant.
* -XCE_ERR_NOT_READY: engine admission cap exhausted — retriable
* rejection.
* Timeout and admission rejection propagate to the job-submit error
* path (ha-commit-engine Req 1.1, 4.4). */
if (sync_err == -ETIMEDOUT || sync_err == -XCE_ERR_NOT_READY)
return sync_err;
return 0;
}
#ifdef XCE_TESTING
/* Test seam (task 2.4): invoke the EXACT durable-gated commit store that the
* legacy sync_append branch uses on a majority ACK, so a test can prove the
* ACK-without-durability hazard is closed without standing up the replication
* wire. Pre-fix this branch stored committed_lsn = written_lsn unconditionally
* (over-advancing past durable_lsn). */
void persist_test_sync_append_legacy_commit_store(persist_t* p,
uint64_t written_lsn)
{
sync_append_legacy_commit_store(p, written_lsn);
}
#endif /* XCE_TESTING */
void persist_log_entry_metadata(const wal_entry_hdr_t* hdr)
{
TRACE("WAL entry: type=%u lsn=%lu term=%lu payload_len=%u crc=%08x",
hdr->type,
(unsigned long)hdr->lsn,
(unsigned long)hdr->term,
hdr->payload_len,
hdr->crc);
}
/* ------------------------------------------------------------------ */
/* Hook functions — AFTER_EVENT handlers that write WAL entries */
/* ------------------------------------------------------------------ */
/* Map HOOK_DETAIL_* values to WAL entry types.
* Returns 0 for details that have no WAL representation (skip). */
static uint32_t detail_to_wal_type(int detail)
{
switch (detail) {
case HOOK_DETAIL_JOB_NEW: return WAL_JOB_SUBMIT;
case HOOK_DETAIL_JOB_START: return WAL_JOB_START;
case HOOK_DETAIL_JOB_EXIT: return WAL_JOB_FINISH;
case HOOK_DETAIL_JOB_FAIL: return WAL_JOB_FAIL;
case HOOK_DETAIL_JOB_STOP: return WAL_JOB_STOP;
case HOOK_DETAIL_JOB_RESUME: return WAL_JOB_RESUME;
case HOOK_DETAIL_JOB_UPDATE: return WAL_JOB_UPDATE;
case HOOK_DETAIL_JOB_CKPT: return WAL_JOB_CKPT;
case HOOK_DETAIL_JOB_RUSAGE_UPDATE: return WAL_JOB_RUSAGE;
case HOOK_DETAIL_SETPREASON: return WAL_JOB_PREASON;
case HOOK_DETAIL_CLRPREASON: return WAL_JOB_PREASON;
case HOOK_DETAIL_QUEUE_UPDATE: return WAL_QUEUE_CTRL;
case HOOK_DETAIL_WORKER_JOIN: return WAL_WORKER_CTRL;
case HOOK_DETAIL_WORKER_UPDATE: return WAL_WORKER_CTRL;
case HOOK_DETAIL_WORKER_DEAD: return WAL_WORKER_CTRL;
default: return 0; /* unknown — skip */
}
}
/* Write-path commit classification (ha-commit-barrier-rollout task 1.2).
* Pure lookup over the VERIFIED table (design.md §Write-Path
* Classification); `detail` disambiguates WAL_WORKER_CTRL only.
* Unclassified types fail fast under XCE_TESTING (abort) and fall back to
* RETAIN in production — never dropped, never falsely confirmed (Req 2.2). */
wal_commit_mode_t wal_classify(uint32_t wal_type, int detail)
{
switch (wal_type) {
case WAL_JOB_SUBMIT:
/* client job submit — request-shaped; keeps the in-hook sync path
* (reference implementation, handle_jobsubmit) */
return WAL_MODE_WAIT;
case WAL_QUEUE_CTRL:
/* client queue control — request-shaped; the handler barrier owes
* commit completion before the requester reply (tasks 2-3) */
return WAL_MODE_WAIT;
case WAL_WORKER_CTRL:
/* the only detail-disambiguated type: JOIN/UPDATE have a requester
* awaiting confirmation (WAIT); DEAD is a controller-detected fact.
* Unexpected details take the fact-safe RETAIN side. */
if (detail == HOOK_DETAIL_WORKER_JOIN ||
detail == HOOK_DETAIL_WORKER_UPDATE)
return WAL_MODE_WAIT;
return WAL_MODE_RETAIN;
case WAL_JOB_START:
case WAL_JOB_FINISH:
case WAL_JOB_FAIL:
case WAL_JOB_STOP:
case WAL_JOB_RESUME:
case WAL_JOB_UPDATE:
case WAL_JOB_RUSAGE:
case WAL_JOB_CKPT:
case WAL_JOB_PREASON:
/* worker/scheduler-reported facts — retained until committed.
* FAIL/UPDATE are dual-origin; their request-shaped side is
* realized by the handler barrier in do_jobcontrol_op
* (research.md §2 call-site verification). */
return WAL_MODE_RETAIN;
case WAL_NOOP:
/* internal promotion no-op — replicated like any committed entry,
* retained until committed (ha-commit-barrier-rollout pre-declared
* row: promotion | internal | RETAIN; guard forces classification) */
return WAL_MODE_RETAIN;
case WAL_CHECKPOINT:
/* local segment marker — never replicated, by design */
return WAL_MODE_LOCAL;
default:
ERROR("wal_classify: unclassified WAL type %u (detail %d) — "
"RETAIN fallback; add it to the classification table",
wal_type, detail);
#ifdef XCE_TESTING
assert(0 && "unclassified WAL type — add it to wal_classify");
#endif
return WAL_MODE_RETAIN;
}
}
/* Commit barrier helper (ha-commit-barrier-rollout task 2.2, design.md
* §Commit Barrier Helper; Req 1.1, 1.2, 1.3).
*
* Handler-callable WAIT completion on the appended watermark: snapshot the
* last-appended LSN, return 0 when the commit index already covers it,
* else block in the commit engine until coverage or deadline. The engine
* result is returned UNMODIFIED (0 / -ETIMEDOUT / -XCE_ERR_NOT_READY /
* -EINVAL) so callers map it to their own error replies.
*
* Lock discipline: p->lock is held only for the current_lsn snapshot and
* is released BEFORE commit_submit(WAIT) — the engine blocks on ce->lock
* + condvar, and p->lock must never be held across it (engine
* lock-ordering rule). */
int persist_commit_barrier(persist_t* p, uint32_t timeout_ms)
{
uint64_t watermark;
if (p == NULL)
return -EINVAL;
/* Snapshot the last-appended watermark under p->lock. current_lsn is
* the NEXT LSN to assign (persist.h invariant: committed_lsn <=
* current_lsn - 1), so the highest appended entry is current_lsn - 1
* (resolved contract, spec Implementation Notes 2.1). current_lsn ==
* 0 means nothing was ever appended — vacuously satisfied. */
pthread_mutex_lock(&p->lock);
if (p->current_lsn == 0) {
pthread_mutex_unlock(&p->lock);
return 0;
}
watermark = p->current_lsn - 1;
pthread_mutex_unlock(&p->lock);
/* Already covered — committed_lsn is monotone, so coverage observed
* here can never regress. This check also absorbs the LSN-0 first
* entry (watermark 0 vs committed_lsn 0): that once-per-lifetime entry
* has no quorum contract (ha-commit-engine Implementation Notes) and
* commit_submit would reject lsn == 0 anyway. */
if (atomic_load_explicit(&p->committed_lsn, memory_order_acquire) >=
watermark)
return 0;
/* Engine not wired: legacy/bootstrap contexts have no engine commit
* contract — there is nothing to wait on, so the barrier is a no-op.
* Production always wires the engine (cluster_init_phase2). */
if (p->repl.ce == NULL)
return 0;
/* timeout_ms == 0 borrows the configured sync deadline (Req 1.2). */
return commit_submit(p->repl.ce, watermark, COMMIT_MODE_WAIT,
timeout_ms != 0 ? timeout_ms
: p->config.sync_timeout_ms);
}
/* Classified append at the hook seam (barrier-rollout task 1.2).
*
* WAIT (queue/worker control): the append stays async here — the handler
* barrier (tasks 2-3) owes commit completion before the requester reply.
* WAL_JOB_SUBMIT never reaches this helper (it keeps the in-hook sync
* path in persist_job_hook_fun).
*
* RETAIN (facts): append + broadcast, then register the appended LSN with
* the commit engine. commit_submit(COMMIT_MODE_RETAIN) is registration-
* free bookkeeping (commit.c — returns 0 immediately; commit state is
* derived from registry watermarks at ACK time), so the call is contract
* documentation: the entry stays pending purely by lsn > committed_lsn
* and commits later via covering ACKs. Commit-progress results are NEVER
* propagated to hook callers — only the local append rc is (Req 3.1).
*
* Lock context: persist_wal_async takes/releases p->lock internally; the
* appended-LSN read below re-takes p->lock briefly (same discipline as
* persist_wal_sync_append) and releases it BEFORE commit_submit, so no
* p->lock is ever held across ce->lock (engine lock-ordering rule). */
static int wal_hook_append_classified(persist_t* p, uint32_t wal_type,
wal_commit_mode_t mode,
const void* payload, uint32_t len)
{
uint64_t appended_lsn;
int rc;
rc = persist_wal_async(p, wal_type, payload, len);
if (rc != 0 || mode != WAL_MODE_RETAIN)
return rc;
if (p->repl.ce == NULL)
return rc; /* engine not wired (bootstrap / engine-less contexts) */
/* current_lsn - 1 is the LSN persist_wal_async just assigned (it
* increments under p->lock). A concurrent append may bump it further;
* that is benign — contiguous advancement means the higher watermark
* still covers this entry. LSN-0 edge: commit_submit rejects lsn == 0
* (first-ever entry of a fresh cluster — accepted interim behavior,
* spec Implementation Notes). */
pthread_mutex_lock(&p->lock);
appended_lsn = p->current_lsn - 1;
pthread_mutex_unlock(&p->lock);
if (appended_lsn != 0)
(void)commit_submit(p->repl.ce, appended_lsn, COMMIT_MODE_RETAIN, 0);
return rc; /* local append rc only — facts have no commit timeout */
}
/* P0 payload: just the detail value (4 bytes).
* Full Cap'n Proto serialization is wired in Task 15. */
int persist_job_hook_fun(int type, int situation, int detail,
void* cls, va_list ap)
{
(void)type;
/* Write to WAL only after the state change is complete.
* BEFORE_EVENT has a different va_list layout and must be ignored. */
if (situation != HOOK_AFTER_EVENT)
return 0;
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
wal_commit_mode_t mode;
if (wal_type == 0)
return 0;
mode = wal_classify(wal_type, detail);
if (mode == WAL_MODE_WAIT) {
/* WAL_JOB_SUBMIT — the only WAIT type in the job hook; reference
* implementation keeps the existing in-hook sync path.
* AFTER_EVENT hook args: cluster_t* c, xlist_t* jlist */
(void)va_arg(ap, cluster_t*);
xlist_t* jlist = va_arg(ap, xlist_t*);
int rc = 0;
for (int i = 0; i < jlist->count && rc == 0; i++) {
job_t* j = (job_t*)xlist_get(jlist, i);
uint8_t* buf = NULL;
size_t sz = 0;
if (capnp_encode_job(&j->job, &buf, &sz) != 0) {
rc = -ENOMEM;
break;
}
/* persist_wal_sync_append returns 0 on successful local WAL write
* even when replication is degraded (no standbys → -ENODEV, timeout
* → -ETIMEDOUT). Only abort the loop on a real disk-write failure
* (non-zero return). Replication status is informational here. */
/* ha-convergence-recovery (3.3): snapshot the LSN this submit
* entry will receive BEFORE the append. persist_wal_async assigns
* hdr.lsn = current_lsn then increments (persist.c:318), so the
* pre-append current_lsn is exactly the written entry's LSN — the
* same value persist_wal_sync_append derives as current_lsn - 1. */
pthread_mutex_lock(&p->lock);
uint64_t submit_lsn = p->current_lsn;
pthread_mutex_unlock(&p->lock);
int sync_rc = 0;
rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, buf, (uint32_t)sz,
&sync_rc);
free(buf);
/* Tag the job with its submit entry's LSN whenever the LOCAL WAL
* write succeeded. persist_wal_sync_append returns the
* persist_wal_async error only on local-write failure; on a
* replication-degraded outcome (0 / -ETIMEDOUT / -ENODEV /
* -XCE_ERR_NOT_READY) the entry was appended locally and the LSN
* advanced, so the tag must be set (the egress predicate is
* independent of replication status). */
if (rc == 0 || rc == -ETIMEDOUT || rc == -XCE_ERR_NOT_READY) {
j->submit_lsn = submit_lsn;
}
}
return rc;
}
/* All other job events: payload = wal_jobstate_t {jobid, state}.
* State is read from the job_t after the AFTER_EVENT hook updates it. */
wal_jobstate_t js = {0};
switch (detail) {
case HOOK_DETAIL_JOB_START: {
/* args: cluster_t*, job_pidinfo_t*, job_t* */
(void)va_arg(ap, cluster_t*);
(void)va_arg(ap, void*);
job_t* j = va_arg(ap, job_t*);
js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state;
break;
}
case HOOK_DETAIL_JOB_EXIT: {
/* args: cluster_t*, xce_jobfinish_t*, job_t* */
(void)va_arg(ap, cluster_t*);
(void)va_arg(ap, void*);
job_t* j = va_arg(ap, job_t*);
js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state;
break;
}
case HOOK_DETAIL_JOB_FAIL: {
/* args: cluster_t*, job_failinfo_t*, job_t* */
(void)va_arg(ap, cluster_t*);
(void)va_arg(ap, void*);
job_t* j = va_arg(ap, job_t*);
js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state;
break;
}
default: {
/* STOP, RESUME, UPDATE, CKPT, PREASON, RUSAGE — args: cluster_t*, job_t* */
(void)va_arg(ap, cluster_t*);
job_t* j = va_arg(ap, job_t*);
js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state;
break;
}
}
/* mode == WAL_MODE_RETAIN — fact-shaped job state transition */
return wal_hook_append_classified(p, wal_type, mode, &js, sizeof(js));
}
int persist_queue_hook_fun(int type, int situation, int detail,
void* cls, va_list ap)
{
(void)type;
if (situation != HOOK_AFTER_EVENT)
return 0;
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
wal_commit_mode_t mode;
if (wal_type == 0)
return 0;
mode = wal_classify(wal_type, detail);
/* args: cluster_t*, queue_t* */
(void)va_arg(ap, cluster_t*);
queue_t* q = va_arg(ap, queue_t*);
/* Payload: [wal_queuehdr_t][queue name NUL-terminated] */
const char* name = q->queue.queue ? q->queue.queue : "";
size_t name_len = strlen(name) + 1;
size_t total = sizeof(wal_queuehdr_t) + name_len;
uint8_t* pkt = malloc(total);
if (!pkt) return -ENOMEM;
wal_queuehdr_t hdr = { .status = q->queue.status };
memcpy(pkt, &hdr, sizeof(hdr));
memcpy(pkt + sizeof(hdr), name, name_len);
/* mode == WAL_MODE_WAIT — the append stays async; the handler barrier
* owes commit completion before the requester reply (task 3) */
int rc = wal_hook_append_classified(p, wal_type, mode, pkt,
(uint32_t)total);
free(pkt);
return rc;
}
int persist_worker_hook_fun(int type, int situation, int detail,
void* cls, va_list ap)
{
(void)type;
if (situation != HOOK_AFTER_EVENT)
return 0;
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
wal_commit_mode_t mode;
if (wal_type == 0)
return 0;
mode = wal_classify(wal_type, detail);
/* WAL_WORKER_CTRL payload layout: [uint32_t detail][capnp worker OR name] */
uint32_t disc = (uint32_t)detail;
uint8_t* buf = NULL;
size_t capnp_sz = 0;
int rc = 0;
/* All worker detail types: args are cluster_t* then worker_t*.
* JOIN uses a unified signature with an extra char* name between them;
* DEAD and UPDATE use (cluster_t*, worker_t*) directly. */
(void)va_arg(ap, cluster_t*);
if (detail == HOOK_DETAIL_WORKER_JOIN)
(void)va_arg(ap, char*); /* skip name — present only for JOIN */
worker_t* w = va_arg(ap, worker_t*);
if (!w) return 0; /* guard: failed join produced no worker slot */
rc = capnp_encode_worker(&w->host, &buf, &capnp_sz);
if (rc != 0) return rc;
/* Build [disc | capnp bytes] in one allocation */
size_t total = WORKER_WAL_HDR_SIZE + capnp_sz;
uint8_t* pkt = malloc(total);
if (!pkt) { free(buf); return -ENOMEM; }
memcpy(pkt, &disc, WORKER_WAL_HDR_SIZE);
memcpy(pkt + WORKER_WAL_HDR_SIZE, buf, capnp_sz);
free(buf);
/* JOIN/UPDATE → WAIT (handler barrier owes completion, task 3);
* DEAD → RETAIN (controller-detected fact, registered with the engine) */
rc = wal_hook_append_classified(p, wal_type, mode, pkt, (uint32_t)total);
free(pkt);
return rc;
}
/* ============================================================
* GC retention predicates (P0-5a / P0-8 / P0-13)
*
* Placed in persist.c so the unit test binary (which links only
* persist.c from the controller sources) can access them without
* pulling in RTTI-dependent job.c / worker.c.
* ============================================================ */
bool job_is_terminal(const job_t* j)
{
return (j->job.state & (JOB_STAT_EXIT | JOB_STAT_DONE | JOB_STAT_PERR)) != 0;
}
bool job_should_retain(const job_t* j, const persist_config_t* cfg)
{
time_t now = time(NULL);
/* TRANSIENT jobs: auto-terminate when TTL elapses after failover.
* transient_since == 0 means the timestamp was never set — do not GC. */
if (j->job.state & JOB_STAT_TRANSIENT) {
if (j->job.transient_since > 0 &&
now > (time_t)j->job.transient_since + (time_t)cfg->transient_job_timeout_s) {
return false;
}
return true;
}
/* Non-terminal jobs are always retained */
if (!job_is_terminal(j)) return true;
/* Terminal jobs: retain until finished_retain_s expires */
return (j->job.finish_time + (time_t)cfg->finished_retain_s >= now);
}
bool worker_should_retain(const xce_worker_t* w, const persist_config_t* cfg)
{
time_t now = time(NULL);
/* HIDDEN workers: admin explicitly removed — clean up at next checkpoint */
if (w->status & XCE_WORKER_HIDDEN) return false;
/* UNAVAILABLE workers: retain until worker_retain_s expires.
* dead_since == 0 means not yet set — do not GC. */
if (w->status & XCE_WORKER_UNAVAIL) {
if (w->dead_since > 0 &&
now > w->dead_since + (time_t)cfg->worker_retain_s) {
return false;
}
}
return true;
}
/* ============================================================
* WAL truncation (task 7.5)
*
* Remove all WAL entries with LSN > cutoff_lsn.
* Idempotent: calling twice with same cutoff_lsn returns 0 both times.
* ============================================================ */
/* Clamp durable_lsn down to `tip` when the log tip recedes below it (truncate
* / divergent-suffix discard). Without this, durable_lsn could exceed
* current_lsn - 1 after a truncate, violating its invariant and letting the
* commit majority count self at an LSN whose bytes were just removed. MUST be
* called with p->lock held (mirrors the committed_lsn clamp in the truncate
* paths). durable_lsn only ever decreases here — the sole exception to its
* monotonic-non-decreasing rule (ha-commit-durability 2.1 State Management). */
static void clamp_durable_lsn_locked(persist_t* p, uint64_t tip)
{
uint64_t dur = atomic_load_explicit(&p->durable_lsn, memory_order_acquire);
if (dur > tip)
atomic_store_explicit(&p->durable_lsn, tip, memory_order_release);
}
int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
{
pthread_mutex_lock(&p->lock);
/* No-op: nothing written yet, or cutoff covers all written entries */
if (p->current_lsn == 0 || cutoff_lsn + 1 >= p->current_lsn) {
pthread_mutex_unlock(&p->lock);
return 0;
}
/* --- Collect segment filenames and sort by start LSN --- */
uint64_t seg_lsns[4096];
int nseg = 0;
DIR* d = opendir(p->wal_path);
if (d) {
struct dirent* e;
while ((e = readdir(d)) != NULL &&
nseg < (int)(sizeof(seg_lsns) / sizeof(seg_lsns[0]))) {
if (strncmp(e->d_name, "seg-", 4) != 0)
continue;
uint64_t sl = 0;
if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &sl) != 1)
continue;
seg_lsns[nseg++] = sl;
}
closedir(d);
}
if (nseg == 0) {
pthread_mutex_unlock(&p->lock);
return 0;
}
/* Insertion sort ascending by start LSN */
for (int i = 1; i < nseg; i++) {
uint64_t key = seg_lsns[i];
int j = i - 1;
while (j >= 0 && seg_lsns[j] > key) {
seg_lsns[j + 1] = seg_lsns[j];
j--;
}
seg_lsns[j + 1] = key;
}
/* --- Find segment containing cutoff_lsn --- */
int cutoff_seg_idx = -1;
for (int i = 0; i < nseg; i++) {
if (seg_lsns[i] <= cutoff_lsn)
cutoff_seg_idx = i;
else
break;
}
if (cutoff_seg_idx < 0) {
/* cutoff_lsn is below the oldest retained segment's start LSN, so EVERY
* on-disk entry has LSN > cutoff_lsn and must be removed (the function's
* contract: "remove all entries with LSN > cutoff_lsn"). This is the
* demotion-convergence case (ha-convergence-recovery Req 1.2): a demoted
* ex-primary whose committed boundary sits below its entire retained log
* must discard all of it and collapse the tip to cutoff_lsn + 1. Delete
* every segment, close the active fd, and reset the in-memory tip. */
for (int i = 0; i < nseg; i++) {
char del_path[4096];
snprintf(del_path, sizeof(del_path),
WAL_SEG_FMT, p->wal_path, seg_lsns[i]);
unlink(del_path);
}
if (p->wal_fd >= 0) {
fdatasync(p->wal_fd);
close(p->wal_fd);
p->wal_fd = -1;
}
p->seg_offset = 0;
p->seg_path[0] = '\0';
p->current_lsn = cutoff_lsn + 1;
p->last_crc = WAL_CRC_NONE;
atomic_store_explicit(&p->committed_lsn, cutoff_lsn,
memory_order_release);
clamp_durable_lsn_locked(p, cutoff_lsn);
pthread_mutex_unlock(&p->lock);
return 0;
}
/* --- Scan segment to find byte offset of first entry with lsn > cutoff --- */
char cutoff_seg_path[4096];
snprintf(cutoff_seg_path, sizeof(cutoff_seg_path),
WAL_SEG_FMT, p->wal_path, seg_lsns[cutoff_seg_idx]);
int scan_fd = open(cutoff_seg_path, O_RDONLY);
if (scan_fd < 0) {
pthread_mutex_unlock(&p->lock);
return -errno;
}
uint32_t scan_off = 0;
uint32_t trunc_off = 0;
uint32_t new_last_crc = WAL_CRC_NONE;
bool found_cut = false;
for (;;) {
wal_entry_hdr_t hdr;
ssize_t n = pread(scan_fd, &hdr, sizeof(hdr), (off_t)scan_off);
if (n < (ssize_t)sizeof(hdr))
break; /* short read or EOF */
/* Pre-allocated zero region: type==0 is not a valid WAL type (min=1) */
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0)
break;
if (hdr.lsn > cutoff_lsn) {
trunc_off = scan_off;
found_cut = true;
break;
}
new_last_crc = hdr.crc;
scan_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(scan_fd);
if (!found_cut)
trunc_off = scan_off; /* end of valid data in segment */
/* --- Close the current wal_fd --- */
if (p->wal_fd >= 0) {
fdatasync(p->wal_fd);
close(p->wal_fd);
p->wal_fd = -1;
}
/* --- Truncate the cutoff segment at the computed offset --- */
int new_fd = open(cutoff_seg_path, O_RDWR);
if (new_fd < 0) {
pthread_mutex_unlock(&p->lock);
return -errno;
}
if (ftruncate(new_fd, (off_t)trunc_off) != 0) {
int saved = errno;
close(new_fd);
pthread_mutex_unlock(&p->lock);
return -saved;
}
fdatasync(new_fd);
/* --- Delete all segments after the cutoff segment --- */
for (int i = cutoff_seg_idx + 1; i < nseg; i++) {
char del_path[4096];
snprintf(del_path, sizeof(del_path),
WAL_SEG_FMT, p->wal_path, seg_lsns[i]);
unlink(del_path);
}
/* --- Update persist_t state under lock --- */
p->wal_fd = new_fd;
p->seg_offset = trunc_off;
snprintf(p->seg_path, sizeof(p->seg_path),
WAL_SEG_FMT, p->wal_path, seg_lsns[cutoff_seg_idx]);
p->current_lsn = cutoff_lsn + 1;
p->last_crc = new_last_crc;
atomic_store_explicit(&p->committed_lsn, cutoff_lsn, memory_order_release);
clamp_durable_lsn_locked(p, cutoff_lsn);
pthread_mutex_unlock(&p->lock);
return 0;
}
Reference in a new issue View git blame Copy permalink