lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/persist.c
Rongsong Shen 6429ef0bff fix: correct controller restart recovery — jobs, workers, and resource ordering 
Four bugs fixed:

1. persist.c: array job WAL loop aborted after index [0] when replication
   returned -ENODEV (no standbys). Loop guard now uses the return value of
   persist_wal_sync_append() which is 0 on local-disk success regardless of
   replication status.

2. join.c: WAL-recovered worker slot (status=UNAVAIL) caused add_worker() to
   return -EEXIST on reconnect. New EEXIST path in handle_worker_join() looks
   up the existing slot by name, updates its endpoint, and clears UNAVAIL so
   the normal hook/reply/on_worker_join flow proceeds.

3. xcectx.c: cbuf_get() in xce_hello() blocked indefinitely when the TCP
   connection dropped because no error handler was registered. Added
   handle_disconnect() which sets ctx->state=XCE_CTX_ERR and calls
   cbuf_tryput(-ECONNRESET) to unblock the waiter, allowing
   checking_connection() to detect the failure and rejoin.

4. cluster.c: persist_recover() was called before cluster_build_resources()
   populated resmgr, causing restore_worker() → add_hostattrs_resvals() to
   fail with EEXIST noise. Moved persist_recover() to after HOOK_AFTER_EVENT
   so resmgr is fully populated before WAL replay.
2026-05-09 03:02:39 +00:00

1194 lines
40 KiB
C
Raw Blame History

#define _GNU_SOURCE
#include "persist.h"
#include "config.h"
#include "ctrl_metrics.h"
#include "hook.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "controller.capnp.h"
#include "xlist.h"
#include "cluster.h"
#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);
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;
while (!atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) {
long ms = (long)p->config.wal_sync_ms;
struct timespec ts = {
.tv_sec = ms / 1000,
.tv_nsec = (ms % 1000) * 1000000L,
};
nanosleep(&ts, NULL);
pthread_mutex_lock(&p->lock);
if (p->wal_fd >= 0)
fdatasync(p->wal_fd);
pthread_mutex_unlock(&p->lock);
}
return NULL;
}
/* ------------------------------------------------------------------ */
/* Public API */
/* ------------------------------------------------------------------ */
void persist_set_workdir(persist_t* p, const char* workdir, const char* name)
{
char effective[4096];
snprintf(effective, sizeof(effective), "%s/%s", workdir, name);
snprintf(p->wal_path, sizeof(p->wal_path), "%s/data/wal", effective);
snprintf(p->snap_path, sizeof(p->snap_path), "%s/data/snap", effective);
}
int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
{
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;
/* 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);
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)) {
atomic_store_explicit(&p->fsync_stop, 1, memory_order_release);
pthread_join(p->fsync_thread, NULL);
atomic_store_explicit(&p->fsync_thread_running, 0, memory_order_release);
}
pthread_mutex_lock(&p->lock);
if (p->wal_fd >= 0) {
fdatasync(p->wal_fd);
close(p->wal_fd);
p->wal_fd = -1;
}
pthread_mutex_unlock(&p->lock);
}
void persist_fini(persist_t* p)
{
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);
}
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);
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;
}
/* ------------------------------------------------------------------ */
/* 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]);
}
/* ------------------------------------------------------------------ */
/* 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;
}
/* 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 */
restore_job(p->cluster, 0, payload, hdr->payload_len);
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;
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;
}
/* For sync replication (WAL_JOB_SUBMIT): wait for at least one standby ACK.
* 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 = repl_wait_ack(&p->repl, written_lsn,
p->config.sync_timeout_ms);
if (sync_err < 0)
CTRL_METRIC_INC(ctrl_repl_degraded_count);
if (sync_rc_out)
*sync_rc_out = sync_err;
/* -ETIMEDOUT = degraded mode (local-only durability); -ENODEV = no standbys.
* Both are acceptable — primary continues operating. */
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 */
}
}
/* 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);
if (wal_type == 0)
return 0;
if (wal_type == WAL_JOB_SUBMIT) {
/* 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. */
int sync_rc = 0;
rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, buf, (uint32_t)sz,
&sync_rc);
free(buf);
}
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;
}
}
return persist_wal_async(p, wal_type, &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);
if (wal_type == 0)
return 0;
/* 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);
int rc = persist_wal_async(p, wal_type, 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);
if (wal_type == 0)
return 0;
/* 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: cluster_t*, worker_t*.
* Caller holds the appropriate lock (workermgr write lock for JOIN,
* worker write lock for DEAD/UPDATE), so no extra lock needed here. */
(void)va_arg(ap, cluster_t*);
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);
rc = persist_wal_async(p, wal_type, 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;
}
Reference in a new issue View git blame Copy permalink