#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 #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include /* 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; 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; /* 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); 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); } /* 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_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 "/seg-.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-.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; }