1986 lines
72 KiB
C
1986 lines
72 KiB
C
#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);
|
||
}
|
||
|
||
/* ------------------------------------------------------------------ */
|
||
/* Dedicated fsync pthread (fires every wal_sync_ms milliseconds) */
|
||
/* ------------------------------------------------------------------ */
|
||
|
||
static void* fsync_worker(void* arg)
|
||
{
|
||
persist_t* p = arg;
|
||
|
||
/* Sleep the configured interval via a timed condvar wait rather than an
|
||
* uninterruptible nanosleep, so persist_stop wakes us immediately instead
|
||
* of blocking its pthread_join for up to a full wal_sync_ms (5s in tests
|
||
* that set a slow interval — was the controller-suite timeout cause). */
|
||
pthread_mutex_lock(&p->fsync_mutex);
|
||
while (!atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) {
|
||
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);
|
||
|
||
/* Stop requested during the wait: exit before the next fdatasync. */
|
||
if (atomic_load_explicit(&p->fsync_stop, memory_order_acquire))
|
||
break;
|
||
|
||
/* fdatasync under p->lock — never held together with fsync_mutex. */
|
||
pthread_mutex_unlock(&p->fsync_mutex);
|
||
pthread_mutex_lock(&p->lock);
|
||
if (p->wal_fd >= 0)
|
||
fdatasync(p->wal_fd);
|
||
pthread_mutex_unlock(&p->lock);
|
||
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);
|
||
|
||
/* 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) {
|
||
fdatasync(p->wal_fd);
|
||
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);
|
||
|
||
/* 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). */
|
||
fdatasync(p->wal_fd);
|
||
|
||
/* 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;
|
||
pthread_mutex_unlock(&p->lock);
|
||
return rc;
|
||
}
|
||
|
||
#ifdef XCE_TESTING
|
||
/* ------------------------------------------------------------------ *
|
||
* 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.
|
||
*
|
||
* Returns number of entries replayed (>= 0) or negative errno on I/O error. */
|
||
static int replay_wal(persist_t* p, uint64_t from_lsn)
|
||
{
|
||
/* --- 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 */
|
||
|
||
/* 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 */
|
||
if (hn < (ssize_t)sizeof(hdr)) {
|
||
if (hn < 0)
|
||
WARN("persist_recover: read error in %s at offset %u: %s",
|
||
seg_path, seg_off, strerror(errno));
|
||
break; /* truncated or pre-allocated zeros */
|
||
}
|
||
|
||
/* Detect pre-allocated zero region (no valid entry starts with
|
||
* all-zero header: lsn == 0 && type == 0 && payload_len == 0) */
|
||
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);
|
||
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);
|
||
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);
|
||
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);
|
||
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 --- */
|
||
int replayed = replay_wal(p, p->checkpoint_lsn);
|
||
if (replayed < 0)
|
||
return replayed;
|
||
|
||
/* --- Step 4: retention pass (TTL check) — no-op in P0 (no entities) --- */
|
||
|
||
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 (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).
|
||
* After local fsync via persist_wal_async, confirm per-entry majority.
|
||
* 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)
|
||
atomic_store_explicit(&p->committed_lsn, written_lsn,
|
||
memory_order_release);
|
||
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;
|
||
}
|
||
|
||
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.
|
||
* ============================================================ */
|
||
|
||
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);
|
||
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);
|
||
|
||
pthread_mutex_unlock(&p->lock);
|
||
return 0;
|
||
}
|