lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/docs/superpowers/plans/2026-05-07-p0-single-controller-persistence.md
Rongsong Shen d70bf6a402 update documents
2026-08-10 17:51:51 +08:00

1664 lines
53 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters

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

> **ARCHIVED — superseded.** This planned a single-controller WAL (40-byte header, fsync pthread) plus a dual-zarray copy-on-write freeze/thaw checkpoint mechanism, with HA/replication deferred to the companion P1–P3 plans (fan-out TCP, single-ACK). What actually shipped: the WAL/persist.c core survived and evolved, but checkpoint consistency uses table-level + per-entity RTTI_LOCK_READ/WRITE nesting, not the COW freeze/thaw path described here (confirmed by `.kiro/specs/ha-entity-snapshots`, which explicitly notes COW entity enumeration was a documented non-goal). Replication/HA is quorum-based Raft, not the fan-out/single-ACK design in the P1–P3 plans — see the closed specs `raft-library`, `raft-ha-core-safety`, `raft-log-replication`, `raft-commit-safety`, `ha-term-safety`, `ha-single-leader-role`, `ha-commit-engine`, `ha-commit-durability`, `ha-commit-barrier-rollout`.
# P0: Single-Controller Persistence — Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Controller survives restart with all job, queue, and worker state intact.
**Architecture:** Append-only WAL (64 MB pre-allocated segments, chained CRC32, 10 ms fsync via dedicated pthread) feeds a periodic checkpoint (dual-zarray COW freeze/thaw, Cap'n Proto snapshot). Recovery loads the latest snapshot then replays the bridging WAL. Hooks fire AFTER_EVENT mutations; `xce_reply()` follows hooks so the client gets a durable ACK.
**Tech Stack:** C11, Cap'n Proto c-capnproto, meson+ninja, nix develop shell
**Design doc:** `src/docs/persistence-and-ha-design.md` — Parts 1–4 + Appendices A.1–A.5 + A.10
**Build commands (all run inside `nix develop`):**
```bash
meson setup build # first time only
ninja -C build # incremental build
meson test -C build # all tests
meson test -C build -t 0 --suite controller # controller suite only
meson test -C build -t 0 --suite lib # lib suite only
```
---
## Sub-phase dependency order
```
P0-A (cleanup + prerequisites)
└─► P0-B (WAL + COW infrastructure)
└─► P0-C (checkpoint + recovery)
├─► P0-D (hooks + reply ordering)
└─► P0-E (deferred cleanup + GC)
└─► P0-F (schema + config + cluster wiring)
└─► P0-G (end-to-end smoke tests)
```
---
## Task 1 — P0-A: Dead Code Cleanup
**Reads:** Design doc Part 1.
**Files:**
- Modify: `src/controller/job.c`
- Modify: `src/controller/worker.c`
- Modify: `src/controller/cluster.c`
- [ ] **Step 1.1 — Remove dead `job.c` constants**
Open `src/controller/job.c`. Remove:
```c
#define NJOBS (100) // line ~18
uint64_t jobid = 100; // line ~20
```
Remove the `#if 0 ... #endif` block inside `load_jobs()` (~lines 63–89). Keep the `load_jobs()` function signature — it will be rewired in P0-C.
- [ ] **Step 1.2 — Add cluster_build_groups() comment**
In `src/controller/cluster.c`, find `cluster_build_groups()` (empty body returning 0). Add:
```c
/* TODO: implement group building */
```
- [ ] **Step 1.3 — Build and verify no regressions**
```bash
ninja -C build 2>&1 | tail -5
meson test -C build 2>&1 | tail -20
```
Expected: build green, test suite unchanged.
- [ ] **Step 1.4 — Commit**
```bash
git add src/controller/job.c src/controller/worker.c src/controller/cluster.c
git commit -m "chore: remove dead code from job.c/worker.c/cluster.c (Part 1)"
```
---
## Task 2 — P0-A: augment `include/config.h` with system constants
**Reads:** Design doc Appendix A.10.
**Files:**
- Modify: `src/include/config.h`
- [ ] **Step 2.1 — Write a failing test for the new constants**
In `src/lib/testsuite/test_config.c` (create if it doesn't exist), add:
```c
#include "config.h"
#include <assert.h>
void test_persistence_constants(void) {
assert(MAX_STANDBYS == 3);
assert(MAX_HA_CANDIDATES == 4);
assert(WAL_FORMAT_VERSION == 1);
assert(WAL_ENTRY_TYPE_MAX == 100);
assert(MAX_WAL_PAYLOAD_BYTES == (1 << 20));
assert(MAX_SNAPSHOT_BYTES == (256 * 1024 * 1024));
assert(DEFAULT_WAL_SYNC_MS == 10);
assert(DEFAULT_SNAP_INTERVAL_S == 60);
assert(DEFAULT_SNAP_WAL_SIZE_MB == 64);
assert(DEFAULT_SNAP_KEEP_N == 2);
assert(DEFAULT_FINISHED_RETAIN_S == 3600);
assert(DEFAULT_WORKER_RETAIN_S == 86400);
assert(DEFAULT_SYNC_TIMEOUT_MS == 50);
assert(DEFAULT_HB_INTERVAL_MS == 3000);
assert(WAL_HDR_CRC_SIZE == 32);
}
```
Run: `meson test -C build -t 0 --suite lib`
Expected: **FAIL** — constants not yet defined.
- [ ] **Step 2.2 — Add constants to `include/config.h`**
Append to `src/include/config.h` (after existing content):
```c
/* --- Persistence & HA tunable constants (Part 2.6, Appendix A.10) --- */
#define MAX_STANDBYS 3
#define MAX_HA_CANDIDATES 4
#define WAL_FORMAT_VERSION 1
#define WAL_ENTRY_TYPE_MAX 100 /* WAL_CHECKPOINT = 100, highest known type */
#define MAX_WAL_PAYLOAD_BYTES (1 << 20) /* 1 MB */
#define MAX_SNAPSHOT_BYTES (256 * 1024 * 1024) /* 256 MB */
#define WAL_HDR_CRC_SIZE 32 /* sizeof(wal_entry_hdr_t) - 2*sizeof(uint32_t) */
#define WAL_CRC_NONE 0xFFFFFFFFu
#define DEFAULT_WAL_SYNC_MS 10
#define DEFAULT_SNAP_INTERVAL_S 60
#define DEFAULT_SNAP_WAL_SIZE_MB 64
#define DEFAULT_SNAP_KEEP_N 2
#define DEFAULT_FINISHED_RETAIN_S 3600
#define DEFAULT_WORKER_RETAIN_S 86400
#define DEFAULT_SYNC_TIMEOUT_MS 50
#define DEFAULT_HB_INTERVAL_MS 3000
#define DEFAULT_TRANSIENT_JOB_TIMEOUT_S 300
```
- [ ] **Step 2.3 — Run test to verify it passes**
```bash
meson test -C build -t 0 --suite lib
```
Expected: **PASS**.
- [ ] **Step 2.4 — Commit**
```bash
git add src/include/config.h src/lib/testsuite/test_config.c
git commit -m "feat: add persistence/HA system constants to config.h (Appendix A.10)"
```
---
## Task 3 — P0-A: rtti + zarray prerequisites (P0-0a + P0-0b)
**Reads:** Design doc Part 3.4, Appendix A.2, P0-0a, P0-0b.
**IMPORTANT:** P0-0a and P0-0b must land on the same branch. Do NOT merge P0-0b before P0-0a.
**Files:**
- Modify: `src/lib/zarray.c`, `src/include/zarray.h`
- Modify: `rtti/typeinfo.h` (via nix flake source)
- Modify: `src/lib/kvmgr.c`, `src/lib/kv.c`
- [ ] **Step 3.1 — Check rtti version and existing freeze/thaw types**
```bash
grep vfreeze_fn_t $(nix eval --raw .#rtti 2>/dev/null)/include/rtti/typeinfo.h 2>/dev/null \
&& echo "ALREADY PRESENT" || echo "NEEDS ADDITION"
```
If **ALREADY PRESENT**: skip to Step 3.3.
If **NEEDS ADDITION**: continue with Step 3.2.
- [ ] **Step 3.2 — Add freeze/thaw types to rtti typeinfo.h**
Locate the rtti source via `nix flake show` or `flake.nix` inputs. In `rtti/typeinfo.h`, add after the existing varray function pointer typedefs:
```c
typedef bool (*retain_fn_t)(void* obj, void* arg);
typedef int (*vfreeze_fn_t)(meta_obj_t* obj);
typedef int (*vthaw_fn_t)(meta_obj_t* obj, retain_fn_t should_retain, void* arg);
```
In the `intf_varray_t` struct, add two new slots:
```c
typedef struct {
vstore_fn_t vstore;
vget_fn_t vget;
vget_vec_fn_t vget_vec;
vput_fn_t vput;
vput_vec_fn_t vput_vec;
vclear_fn_t vclear;
vfreeze_fn_t vfreeze; /* NEW */
vthaw_fn_t vthaw; /* NEW */
} intf_varray_t;
```
Add global wrapper declarations to `typeinfo.h`:
```c
int vfreeze(meta_obj_t* obj);
int vthaw(meta_obj_t* obj, retain_fn_t should_retain, void* arg);
```
Bump nix derivation version from `rtti-0.0.1` to `rtti-0.1.0` in `flake.nix`.
- [ ] **Step 3.3 — Convert all intf_varray_t initializers to designated style**
Find all initializer sites:
```bash
grep -rn 'intf_varray_t' src/ | grep -v '\.h:'
```
For each initializer in `zarray.c`, `kvmgr.c`, `kv.c`, convert from positional to designated style:
```c
/* BEFORE (positional — will break when struct grows): */
static intf_varray_t xzarray_iface = { xzarray_store, xzarray_get, ... };
/* AFTER (designated — safe to extend): */
static intf_varray_t xzarray_iface = {
.vstore = xzarray_store,
.vget = xzarray_get,
.vget_vec = xzarray_get_vec,
.vput = xzarray_put,
.vput_vec = xzarray_put_vec,
.vclear = xzarray_clear,
.vfreeze = NULL, /* filled in P0-2 */
.vthaw = NULL, /* filled in P0-2 */
};
```
```bash
ninja -C build 2>&1 | tail -10
```
Expected: build green (NULL slots are valid — not wired yet).
- [ ] **Step 3.4 — Write failing test for freeze/thaw interface presence**
In `src/lib/testsuite/test_zarray_cow.c` (new file):
```c
#include "zarray.h"
#include "rtti/typeinfo.h"
#include <assert.h>
extern intf_varray_t xzarray_type;
void test_freeze_thaw_slots_exist(void) {
/* After P0-2 these will be non-NULL; for now just verify the struct has the slots */
(void)xzarray_type.vfreeze;
(void)xzarray_type.vthaw;
}
```
```bash
meson test -C build -t 0 --suite lib
```
Expected: **PASS** (slots exist, values NULL — presence test only).
- [ ] **Step 3.5 — Full rebuild to catch ABI breaks**
```bash
rm -rf build && meson setup build && ninja -C build
meson test -C build 2>&1 | tail -20
```
Expected: clean build, all tests pass.
- [ ] **Step 3.6 — Commit P0-A prerequisites**
```bash
git add src/lib/zarray.c src/include/zarray.h src/lib/kvmgr.c src/lib/kv.c \
src/lib/testsuite/test_zarray_cow.c flake.nix
git commit -m "feat: add vfreeze/vthaw slots to intf_varray_t; convert initializers to designated style (P0-0b)"
```
---
## Task 4 — P0-A: delaymgr precision enhancement (P0-2b)
**Reads:** Design doc P0-2b, Part 3.10 note on delaymgr.
**Files:**
- Modify: `src/lib/delay.c`, `src/include/delay.h`
- Modify: `src/controller/cluster.c` (delaymgr_init call)
- [ ] **Step 4.1 — Write failing test for sub-second precision**
In `src/lib/testsuite/test_delay.c`, add:
```c
#include "delay.h"
#include <time.h>
#include <assert.h>
void test_delaymgr_precision_ms(void) {
delaymgr_t mgr;
/* precision_ms field must exist on the struct */
delaymgr_init(&mgr, 200 /* precision_ms */, NULL);
assert(mgr.precision_ms == 200);
delaymgr_fini(&mgr);
}
```
```bash
meson test -C build -t 0 --suite lib
```
Expected: **FAIL** — `precision_ms` field doesn't exist yet.
- [ ] **Step 4.2 — Rename `precision` → `precision_ms` in delay.h**
In `src/include/delay.h`, in `delaymgr_t`, rename `int precision` → `int precision_ms`.
- [ ] **Step 4.3 — Replace `sleep()` with `nanosleep()` in delay.c**
In `src/lib/delay.c`, in `delay_worker()`, replace:
```c
sleep(mgr->precision);
```
with:
```c
struct timespec ts = { 0, (long)mgr->precision_ms * 1000000L };
nanosleep(&ts, NULL);
```
Update `delaymgr_init()` to store the parameter as `precision_ms`:
```c
int delaymgr_init(delaymgr_t* mgr, int precision_ms, task_run_fn_t runner_fn) {
if ((mgr == NULL) || (precision_ms <= 0)) return -EINVAL;
...
mgr->precision_ms = precision_ms;
...
}
```
- [ ] **Step 4.4 — Update cluster.c delaymgr_init call**
Find `delaymgr_init(&c->delaymgr, 2, run_delay_tasks)` in `src/controller/cluster.c` and change to:
```c
delaymgr_init(&c->delaymgr, 200 /* precision_ms */, run_delay_tasks);
```
- [ ] **Step 4.5 — Audit all other delaymgr_init call sites**
```bash
grep -rn 'delaymgr_init' src/
```
For each: update second argument to `200` (milliseconds). The delay values passed to `delay_task()` remain in seconds — only the sleep granularity changes.
- [ ] **Step 4.6 — Run tests**
```bash
ninja -C build && meson test -C build -t 0 --suite lib
```
Expected: **PASS**.
- [ ] **Step 4.7 — Commit**
```bash
git add src/lib/delay.c src/include/delay.h src/controller/cluster.c
git commit -m "feat: rename delaymgr precision→precision_ms, replace sleep with nanosleep (P0-2b)"
```
---
## Task 5 — P0-B: WAL types and entry format (headers)
**Reads:** Design doc Parts 2.2, 2.3, 2.6, Appendix A.1.
**Files:**
- Create: `src/include/persist.h`
- Modify: `src/include/config.h` (WAL_HDR_CRC_SIZE already added in Task 2)
- [ ] **Step 5.1 — Write failing compile test for wal_entry_hdr_t size**
In `src/controller/testsuite/test_persist_types.c` (new file):
```c
#include "persist.h"
#include <stddef.h>
#include <assert.h>
void test_wal_header_size(void) {
static_assert(sizeof(wal_entry_hdr_t) == 40,
"wal_entry_hdr_t must be 40 bytes");
}
void test_wal_entry_types_defined(void) {
assert(WAL_JOB_SUBMIT == 1);
assert(WAL_JOB_START == 2);
assert(WAL_CHECKPOINT == 100);
}
void test_persist_config_defaults(void) {
persist_config_t cfg = persist_config_defaults();
assert(cfg.wal_sync_ms == DEFAULT_WAL_SYNC_MS);
assert(cfg.snap_interval_s == DEFAULT_SNAP_INTERVAL_S);
assert(cfg.snap_keep_n == DEFAULT_SNAP_KEEP_N);
assert(cfg.finished_retain_s == DEFAULT_FINISHED_RETAIN_S);
assert(cfg.sync_timeout_ms == DEFAULT_SYNC_TIMEOUT_MS);
}
```
```bash
ninja -C build 2>&1 | grep "persist.h: No such"
```
Expected: compile error — `persist.h` doesn't exist yet.
- [ ] **Step 5.2 — Create `src/include/persist.h`**
```c
#pragma once
#include <stdint.h>
#include <stddef.h>
#include <pthread.h>
#include <stdatomic.h>
#include "config.h"
#include "mylock.h"
/* ---- WAL entry header (40 bytes, 8-byte aligned, Part 2.2) ---- */
typedef struct {
uint64_t term; /* Election term */
uint64_t lsn; /* Log sequence number */
uint64_t timestamp; /* Epoch microseconds */
uint32_t type; /* wal_entry_type */
uint32_t payload_len;
uint32_t crc; /* CRC32(header[0..WAL_HDR_CRC_SIZE-1] + payload) */
uint32_t prev_crc; /* Previous record CRC; WAL_CRC_NONE for first in segment */
} wal_entry_hdr_t;
static_assert(sizeof(wal_entry_hdr_t) == 40, "wal_entry_hdr_t must be 40 bytes");
/* ---- WAL entry types (Part 2.3) ---- */
enum wal_entry_type {
WAL_JOB_SUBMIT = 1,
WAL_JOB_START = 2,
WAL_JOB_FINISH = 3,
WAL_JOB_FAIL = 4,
WAL_JOB_STOP = 5,
WAL_JOB_RESUME = 6,
WAL_JOB_UPDATE = 7,
WAL_JOB_RUSAGE = 8,
WAL_JOB_CKPT = 9,
WAL_JOB_PREASON = 10,
WAL_QUEUE_CTRL = 20,
WAL_WORKER_CTRL = 30,
WAL_CHECKPOINT = 100,
};
/* ---- Persistence configuration (Part 2.6) ---- */
typedef struct {
uint32_t wal_sync_ms;
uint32_t snap_interval_s;
uint32_t snap_wal_size_mb;
uint32_t snap_keep_n;
uint32_t transient_job_timeout_s;
uint32_t finished_retain_s;
uint32_t worker_retain_s;
uint32_t sync_timeout_ms;
} persist_config_t;
static inline persist_config_t persist_config_defaults(void) {
return (persist_config_t){
.wal_sync_ms = DEFAULT_WAL_SYNC_MS,
.snap_interval_s = DEFAULT_SNAP_INTERVAL_S,
.snap_wal_size_mb = DEFAULT_SNAP_WAL_SIZE_MB,
.snap_keep_n = DEFAULT_SNAP_KEEP_N,
.transient_job_timeout_s = DEFAULT_TRANSIENT_JOB_TIMEOUT_S,
.finished_retain_s = DEFAULT_FINISHED_RETAIN_S,
.worker_retain_s = DEFAULT_WORKER_RETAIN_S,
.sync_timeout_ms = DEFAULT_SYNC_TIMEOUT_MS,
};
}
/* ---- Key→index mapping for checkpoint serialization (Part 3.8, Appendix A.1) ---- */
typedef struct {
int index;
char key[XCE_MAXNAME_LEN]; /* owned deep copy */
size_t key_len;
} slot_key_t;
/* ---- Forward declarations (full persist_t defined in persist.c) ---- */
struct cluster_s;
typedef struct persist_s persist_t;
int persist_init(persist_t* p, persist_config_t* config, struct cluster_s* c);
int persist_start(persist_t* p);
void persist_stop(persist_t* p);
void persist_fini(persist_t* p);
int persist_recover(persist_t* p);
int persist_wal_async(persist_t* p, uint32_t type, const void* payload, uint32_t len);
int persist_wal_sync_append(persist_t* p, uint32_t type,
const void* payload, uint32_t len, int* sync_rc_out);
int persist_checkpoint(persist_t* p);
void persist_log_entry_metadata(const wal_entry_hdr_t* hdr);
```
- [ ] **Step 5.3 — Run test**
```bash
ninja -C build && meson test -C build -t 0 --suite controller
```
Expected: **PASS** (types and size check).
- [ ] **Step 5.4 — Commit**
```bash
git add src/include/persist.h src/controller/testsuite/test_persist_types.c
git commit -m "feat: add persist.h with WAL types, config struct, slot_key_t (P0-B)"
```
---
## Task 6 — P0-B: zarray COW extensions (P0-2)
**Reads:** Design doc Parts 3.3–3.6, Appendix A.2.
**Files:**
- Modify: `src/lib/zarray.c`, `src/include/zarray.h`
- [ ] **Step 6.1 — Write failing COW freeze/thaw test**
In `src/lib/testsuite/test_zarray_cow.c`, extend with:
```c
#include "zarray.h"
#include <assert.h>
#include <string.h>
void test_freeze_thaw_round_trip(void) {
/* Allocate an xzarray via RTTI (use existing makeInstance pattern) */
meta_obj_t* arr = makeInstance(&xzarray_type, NULL);
assert(arr != NULL);
/* Store 3 items */
int idx[3];
char* data[3] = { strdup("a"), strdup("b"), strdup("c") };
for (int i = 0; i < 3; i++) {
assert(vstore(arr, data[i]) >= 0);
}
/* Freeze */
assert(vfreeze(arr) == 0);
/* During COW: add one more item */
char* d = strdup("d");
assert(vstore(arr, d) >= 0);
/* Thaw (no GC — retain all) */
assert(vthaw(arr, NULL, NULL) == 0);
/* After thaw: all 4 items reachable */
int count = 0;
/* ... iterate and count ... */
freeInstance(arr);
}
void test_freeze_enomem_rollback(void) {
/* Simulate OOM at freeze: should restore original data pointer */
/* This is a compile-time structure test — real OOM injection deferred */
(void)0;
}
```
```bash
meson test -C build -t 0 --suite lib
```
Expected: **FAIL** — `vfreeze`/`vthaw` still NULL in xzarray_type.
- [ ] **Step 6.2 — Add `xzarray_private_t` fields for shadow**
In `src/lib/zarray.c`, in `xzarray_private_t`:
```c
typedef struct {
mylock_t lock;
xzarray_item_t* shadow; /* NULL when not in COW mode */
size_t shadow_caps;
} xzarray_private_t;
```
- [ ] **Step 6.3 — Change `xzarray_item_t.inuse` to `_Atomic _Bool`**
In `src/include/zarray.h` (or `zarray.c` if still file-private):
```c
typedef struct {
_Atomic _Bool inuse;
void* data;
} xzarray_item_t;
```
Run:
```bash
grep -n '\.inuse' src/lib/zarray.c
```
For every read: use `atomic_load_explicit(&item->inuse, memory_order_acquire)`.
For every write: use `atomic_store_explicit(&item->inuse, true/false, memory_order_release)`.
- [ ] **Step 6.4 — Implement `xzarray_freeze()`**
In `src/lib/zarray.c`:
```c
static int xzarray_freeze(meta_obj_t* obj) {
xzarray_private_t* priv = RTTI_OBJ_TO_PRIV(obj);
xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj);
assert(priv->shadow == NULL && "xzarray_freeze called while already in COW");
/* Pre-grow active array to 1.5x to avoid realloc during COW (A6-026) */
size_t new_caps = a->caps + (a->caps / 2) + 16;
xzarray_item_t* active = calloc(new_caps, sizeof(xzarray_item_t));
if (!active) return -ENOMEM;
priv->shadow = a->data;
priv->shadow_caps = a->caps;
a->data = active;
a->caps = new_caps;
a->count = 0;
a->pos = priv->shadow_caps; /* start past shadow to avoid index collisions */
return 0;
}
```
- [ ] **Step 6.5 — Implement `xzarray_thaw()`**
```c
static int xzarray_thaw(meta_obj_t* obj, retain_fn_t should_retain, void* arg) {
xzarray_private_t* priv = RTTI_OBJ_TO_PRIV(obj);
xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj);
assert(priv->shadow != NULL && "xzarray_thaw called without freeze");
/* Merge shadow into active: copy surviving shadow entries to active */
for (size_t i = 0; i < priv->shadow_caps; i++) {
if (!atomic_load_explicit(&priv->shadow[i].inuse, memory_order_relaxed))
continue;
void* item = priv->shadow[i].data;
if (should_retain && !should_retain(item, arg))
continue; /* GC: do not copy */
/* Write to active at the same index (kvtab maps to this index) */
if (i >= a->caps) {
/* Active was pre-grown — this shouldn't happen, but guard anyway */
continue;
}
if (!atomic_load_explicit(&a->data[i].inuse, memory_order_relaxed)) {
a->data[i].data = item;
atomic_store_explicit(&a->data[i].inuse, true, memory_order_relaxed);
a->count++;
}
/* If active[i].inuse==true, a mutation during COW already wrote new data — keep it */
}
free(priv->shadow);
priv->shadow = NULL;
priv->shadow_caps = 0;
a->pos = 0;
return 0;
}
```
- [ ] **Step 6.6 — Add `vclear` COW guard**
```c
static void xzarray_clear(meta_obj_t* obj) {
xzarray_private_t* priv = RTTI_OBJ_TO_PRIV(obj);
assert(priv->shadow == NULL && "xzarray_clear called during COW — use deferred cleanup");
/* ... existing clear logic ... */
}
```
- [ ] **Step 6.7 — Update `xzarray_type` registration with freeze/thaw**
In the `intf_varray_t xzarray_iface` designated initializer:
```c
.vfreeze = xzarray_freeze,
.vthaw = xzarray_thaw,
```
- [ ] **Step 6.8 — Run tests**
```bash
ninja -C build && meson test -C build -t 0 --suite lib
```
Expected: **PASS**.
- [ ] **Step 6.9 — Commit**
```bash
git add src/lib/zarray.c src/include/zarray.h src/lib/testsuite/test_zarray_cow.c
git commit -m "feat: implement xzarray COW freeze/thaw with shadow array (P0-2)"
```
---
## Task 7 — P0-B: WAL writer + fsync thread (P0-1)
**Reads:** Design doc Parts 2.2–2.4, Appendix A.1.
**Files:**
- Create: `src/controller/persist.c`
- Modify: `src/controller/meson.build`
- [ ] **Step 7.1 — Write failing WAL write test**
In `src/controller/testsuite/test_persist_wal.c` (new file):
```c
#include "persist.h"
#include <assert.h>
#include <string.h>
#include <unistd.h>
void test_wal_write_and_read_back(void) {
persist_t p;
persist_config_t cfg = persist_config_defaults();
/* Use a temp dir so no real cluster is needed */
char workdir[] = "/tmp/xce_test_XXXXXX";
assert(mkdtemp(workdir) != NULL);
/* persist_init with a stub cluster pointer */
assert(persist_init(&p, &cfg, NULL) == 0);
/* Write a WAL entry */
const char payload[] = "test_payload";
assert(persist_wal_async(&p, WAL_JOB_UPDATE, payload, sizeof(payload)) == 0);
/* WAL file should exist in workdir/data/wal/ */
char wal_path[256];
snprintf(wal_path, sizeof(wal_path), "%s/data/wal", workdir);
/* verify directory exists */
assert(access(wal_path, F_OK) == 0);
persist_stop(&p);
persist_fini(&p);
/* cleanup workdir ... */
}
```
```bash
ninja -C build 2>&1 | grep "persist.c"
```
Expected: link error — `persist.c` doesn't exist.
- [ ] **Step 7.2 — Create `src/controller/persist.c` with WAL writer skeleton**
```c
#include "persist.h"
#include "config.h"
#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <pthread.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> /* for crc32() */
struct persist_s {
int wal_fd;
uint64_t current_lsn;
uint64_t last_crc;
uint64_t term;
char seg_path[4096];
uint32_t seg_offset;
uint64_t checkpoint_lsn;
uint64_t total_wal_bytes;
uint32_t last_replicated_crc;
persist_config_t config;
mylock_t lock;
atomic_int checkpoint_in_progress;
atomic_int checkpoint_requested;
pthread_cond_t checkpoint_done;
pthread_mutex_t checkpoint_mutex;
pthread_t snap_thread;
pthread_t fsync_thread;
_Atomic int fsync_stop;
char wal_path[4096];
char snap_path[4096];
int wal_dirfd;
int snap_dirfd;
};
```
Add `persist_init()`:
```c
int persist_init(persist_t* p, persist_config_t* cfg, struct cluster_s* c) {
memset(p, 0, sizeof(*p));
p->config = cfg ? *cfg : persist_config_defaults();
p->wal_fd = -1;
/* Validate paths are not on NFS (Part 2.4 filesystem requirement) */
/* ... statfs() check for NFS_SUPER_MAGIC = 0x6969 ... */
/* Create wal/ and snap/ directories at mode 0700 */
/* For now: caller sets wal_path/snap_path before calling persist_init */
mylock_init(&p->lock);
pthread_mutex_init(&p->checkpoint_mutex, NULL);
pthread_cond_init(&p->checkpoint_done, NULL);
return 0;
}
```
Add `persist_wal_async()` stub:
```c
int persist_wal_async(persist_t* p, uint32_t type, const void* payload, uint32_t len) {
assert(type != WAL_CHECKPOINT && "persist_wal_async must not be called with WAL_CHECKPOINT");
/* TODO P0-1: write wal_entry_hdr_t + payload to WAL, update last_crc */
(void)p; (void)type; (void)payload; (void)len;
return 0;
}
```
Add remaining stubs (`persist_start`, `persist_stop`, `persist_fini`, `persist_recover`, `persist_wal_sync_append`, `persist_checkpoint`, `persist_log_entry_metadata`) that return 0/do nothing.
- [ ] **Step 7.3 — Add persist.c to meson.build**
In `src/controller/meson.build`, add `persist.c` to the controller source list.
- [ ] **Step 7.4 — Implement WAL segment init and entry write**
Full implementation of `persist_wal_async()`:
1. Acquire `p->lock`
2. If `p->wal_fd == -1`, open first segment (`seg-0000000000000000.wal`) with `O_CREAT|O_WRONLY|O_APPEND`, pre-allocate 64 MB with `posix_fallocate()`
3. Fill `wal_entry_hdr_t`: `term = p->term`, `lsn = p->current_lsn++`, `timestamp = now_us()`, `type`, `payload_len = len`
4. Compute `crc = crc32(0, header_bytes[0..WAL_HDR_CRC_SIZE-1]) then update with payload`
5. Set `prev_crc = (p->seg_offset == 0) ? WAL_CRC_NONE : p->last_crc`
6. `pwrite(p->wal_fd, &hdr, 40, p->seg_offset)` + `pwrite(payload, len, ...)`
7. Update `p->last_crc`, `p->seg_offset`, `p->total_wal_bytes`
8. Segment rotation when `seg_offset >= 64 MB`
9. Release lock
- [ ] **Step 7.5 — Implement dedicated fsync pthread**
```c
static void* fsync_worker(void* arg) {
persist_t* p = arg;
struct timespec ts = { 0, (long)p->config.wal_sync_ms * 1000000L };
while (!atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) {
nanosleep(&ts, NULL);
if (p->wal_fd >= 0)
fdatasync(p->wal_fd);
}
return NULL;
}
```
Start in `persist_start()`, join in `persist_stop()`.
- [ ] **Step 7.6 — Run tests**
```bash
ninja -C build && meson test -C build -t 0 --suite controller
```
Expected: **PASS**.
- [ ] **Step 7.7 — Commit**
```bash
git add src/controller/persist.c src/controller/meson.build \
src/controller/testsuite/test_persist_wal.c
git commit -m "feat: WAL writer with segment management and dedicated fsync pthread (P0-1)"
```
---
## Task 8 — P0-C: Cap'n Proto schemas (P0-9)
**Reads:** Design doc Part 2.2 (payload table), Appendix A.5.
**Files:**
- Modify: `src/schema/controller.capnp`
- [ ] **Step 8.1 — Write failing compile test for schema types**
In `src/controller/testsuite/test_persist_schema.c`:
```c
/* Verify Cap'n Proto generated types are accessible */
#include "controller.capnp.h"
#include <assert.h>
void test_persist_config_schema(void) {
/* PersistConfig must exist with correct field count */
struct PersistConfig cfg;
cfg.walSyncMs = 10;
cfg.snapIntervalS = 60;
cfg.snapWalSizeMb = 64;
cfg.snapKeepN = 2;
cfg.transientJobTimeoutS = 300;
assert(cfg.walSyncMs == 10);
}
```
```bash
ninja -C build 2>&1 | grep "PersistConfig"
```
Expected: compile error — `PersistConfig` not in schema yet.
- [ ] **Step 8.2 — Add PersistConfig to controller.capnp**
Append to `src/schema/controller.capnp`:
```capnp
struct PersistConfig $C.mapname("persist_config_t") {
walSyncMs @0 :UInt32 $C.mapname("wal_sync_ms") = 10;
snapIntervalS @1 :UInt32 $C.mapname("snap_interval_s") = 60;
snapWalSizeMb @2 :UInt32 $C.mapname("snap_wal_size_mb") = 64;
finishedRetainS @3 :UInt32 $C.mapname("finished_retain_s") = 3600;
workerRetainS @4 :UInt32 $C.mapname("worker_retain_s") = 86400;
syncTimeoutMs @5 :UInt32 $C.mapname("sync_timeout_ms") = 50;
snapKeepN @6 :UInt32 $C.mapname("snap_keep_n") = 2;
transientJobTimeoutS @7 :UInt32 $C.mapname("transient_job_timeout_s") = 300;
}
struct SnapshotHeader {
lsn @0 :UInt64;
term @1 :UInt64;
timestamp @2 :UInt64;
version @3 :UInt32 = 1;
nJobs @4 :UInt32;
nQueues @5 :UInt32;
nWorkers @6 :UInt32;
checksum @7 :UInt32; # CRC32 of Cap'n Proto payload
headerChecksum @8 :UInt32; # CRC32(lsn, term, timestamp, version, nJobs, nQueues, nWorkers)
}
struct WALEntry {
# Replication-only envelope (NOT on-disk format). On-disk uses raw wal_entry_hdr_t.
term @0 :UInt64;
lsn @1 :UInt64;
entryType @2 :UInt32;
payload @3 :Data;
crc @4 :UInt32;
prevCrc @5 :UInt32;
}
```
Also add `transientSince @N :UInt64` to the `Job` struct and `deadSince` to `Worker` struct, and `jobStatTransient = 0x400` comment.
- [ ] **Step 8.3 — Rebuild and run test**
```bash
ninja -C build && meson test -C build -t 0 --suite controller
```
Expected: **PASS**.
- [ ] **Step 8.4 — Commit**
```bash
git add src/schema/controller.capnp src/controller/testsuite/test_persist_schema.c
git commit -m "feat: add PersistConfig, SnapshotHeader, WALEntry Cap'n Proto schemas (P0-9)"
```
---
## Task 9 — P0-C: kvmgr key capture (P0-14 + P0-14a)
**Reads:** Design doc Part 3.8, Appendix A.1 `kvmgr_capture_keys`.
**Files:**
- Modify: `src/lib/kvmgr.c`, `src/lib/include/kvmgr.h`
- [ ] **Step 9.1 — Write failing test for kvmgr_capture_keys**
In `src/lib/testsuite/test_kvmgr_capture.c`:
```c
#include "kvmgr.h"
#include "persist.h"
#include <assert.h>
#include <string.h>
void test_capture_keys_round_trip(void) {
meta_obj_t* km = makeKvmgr(KVMGR_KEY_STRING, NULL);
assert(km != NULL);
/* Add 3 items */
kvmgr_set(km, "alpha", strdup("val_a"), NULL);
kvmgr_set(km, "beta", strdup("val_b"), NULL);
kvmgr_set(km, "gamma", strdup("val_c"), NULL);
slot_key_t* keys = NULL;
size_t n = 0;
/* Must hold write lock before calling */
assert(kvmgr_capture_keys(km, &keys, &n) == 0);
assert(n == 3);
/* All keys are owned copies */
int found = 0;
for (size_t i = 0; i < n; i++) {
if (strcmp(keys[i].key, "alpha") == 0 ||
strcmp(keys[i].key, "beta") == 0 ||
strcmp(keys[i].key, "gamma") == 0) found++;
}
assert(found == 3);
free(keys);
freeInstance(km);
}
```
```bash
meson test -C build -t 0 --suite lib
```
Expected: **FAIL** — `kvmgr_capture_keys` undefined.
- [ ] **Step 9.2 — Implement `kvmgr_capture_keys` in kvmgr.c**
```c
int kvmgr_capture_keys(meta_obj_t* kvmgr, slot_key_t** keys_out, size_t* n_keys_out) {
*keys_out = NULL;
*n_keys_out = 0;
/* Count live entries */
size_t n = kvmgr_count(kvmgr);
if (n == 0) return 0;
slot_key_t* keys = malloc(n * sizeof(slot_key_t));
if (!keys) return -ENOMEM;
size_t filled = 0;
/* Walk the kvtab: iterate hash chains, extract key and index for each live slot */
/* Use the htab walk primitive (P0-14 — implement kvtab_foreach if not present) */
kvtab_foreach(kvmgr, keys, &filled);
*keys_out = keys;
*n_keys_out = filled;
return 0;
}
```
Implement `kvtab_foreach()` in `src/lib/htab.c` or inline in `kvmgr.c` — walks hash chains and for each live entry calls:
```c
strncpy(dst->key, entry->key, XCE_MAXNAME_LEN - 1);
dst->key[XCE_MAXNAME_LEN - 1] = '\0';
dst->key_len = strlen(entry->key) + 1;
dst->index = entry->index;
```
- [ ] **Step 9.3 — Run test**
```bash
ninja -C build && meson test -C build -t 0 --suite lib
```
Expected: **PASS**.
- [ ] **Step 9.4 — Commit**
```bash
git add src/lib/kvmgr.c src/lib/htab.c src/lib/testsuite/test_kvmgr_capture.c
git commit -m "feat: implement kvmgr_capture_keys for lock-free checkpoint serialization (P0-14a)"
```
---
## Task 10 — P0-C: checkpoint (P0-3)
**Reads:** Design doc Parts 3.1–3.11, 2.5.
**Files:**
- Modify: `src/controller/persist.c`
- [ ] **Step 10.1 — Write failing checkpoint test**
In `src/controller/testsuite/test_persist_checkpoint.c`:
```c
#include "persist.h"
#include <assert.h>
void test_checkpoint_creates_snapshot_file(void) {
/* Setup persist_t with a temp dir, add some WAL entries, trigger checkpoint */
/* Verify snap/ directory contains a .capnp file */
/* ... setup boilerplate ... */
assert(persist_checkpoint(&p) == 0);
/* count files in snap_path */
/* assert at least one snapshot file exists */
}
void test_freeze_thaw_no_data_loss(void) {
/* Write 100 jobs to kvmgr, freeze, write 10 more during COW, thaw */
/* Verify all 110 jobs are reachable after thaw */
}
```
Expected: **FAIL**.
- [ ] **Step 10.2 — Implement `persist_checkpoint()` in persist.c**
Sequence (Part 3.8 + 3.10):
1. Set `checkpoint_in_progress = 1`
2. Acquire all kvmgr write locks (jobmgr, workermgr, queuemgr, resmgr)
3. `vfreeze()` on each data zarray
4. `kvmgr_capture_keys()` on each kvmgr (still under write lock)
5. Write `WAL_CHECKPOINT` marker via `persist_write_checkpoint_marker()` (local only, no `repl_broadcast`)
6. Release all write locks
7. Spawn serialization pthread: serialize shadow arrays to Cap'n Proto → write to `snap-<lsn>.tmp` → CRC32 → `rename()` to `snap-<lsn>.capnp`
8. Serialization pthread calls `vthaw()` on each zarray after `fsync(snap_fd)` completes
9. Signal `checkpoint_done` cond, set `checkpoint_in_progress = 0`
10. Trim old WAL segments up to `checkpoint_lsn` (subject to trim guard)
11. Rotate old snapshots — keep only `snap_keep_n` most recent
- [ ] **Step 10.3 — Register checkpoint timer via delaymgr**
In `persist_start()`:
```c
delay_task(&c->delaymgr, 1 /* seconds */, -1 /* repeat */, persist_checkpoint_timer_fn, p);
```
`persist_checkpoint_timer_fn()` checks:
1. If `checkpoint_in_progress != 0` → skip
2. If `checkpoint_requested != 0` → run checkpoint, clear flag
3. If periodic countdown reached zero → run checkpoint, reset countdown
- [ ] **Step 10.4 — Run tests**
```bash
ninja -C build && meson test -C build -t 0 --suite controller
```
Expected: **PASS**.
- [ ] **Step 10.5 — Commit**
```bash
git add src/controller/persist.c src/controller/testsuite/test_persist_checkpoint.c
git commit -m "feat: implement dual-zarray COW checkpoint with serialization pthread (P0-3)"
```
---
## Task 11 — P0-C: recovery (P0-4 + P0-12 + P0-12a)
**Reads:** Design doc Part 2.5, P0-4, P0-12, P0-12a.
**Files:**
- Modify: `src/controller/persist.c`
- Modify: `src/controller/worker.c`
- Modify: `src/controller/job.c`
- [ ] **Step 11.1 — Write failing recovery round-trip test**
In `src/controller/testsuite/test_persist_recovery.c`:
```c
void test_recovery_roundtrip(void) {
/* 1. Init persist, write 50 WAL_JOB_SUBMIT entries + 1 checkpoint */
/* 2. Call persist_fini (simulates shutdown) */
/* 3. Re-init persist, call persist_recover() */
/* 4. Verify jobmgr has 50 jobs */
}
void test_recovery_skips_expired_jobs(void) {
/* Write jobs with finish_time older than finished_retain_s */
/* After recovery + retention pass, they should be gone */
}
```
Expected: **FAIL**.
- [ ] **Step 11.2 — Implement `restore_job()` in job.c**
```c
int restore_job(cluster_t* c, uint64_t jobid, const void* capnp_payload, size_t len) {
/* Deserialize Job Cap'n Proto */
/* Insert at original jobid — bypass allocator */
/* Return -EEXIST if jobid already present (warn and skip) */
/* Do NOT fire BEFORE_EVENT or AFTER_EVENT hooks */
}
```
- [ ] **Step 11.3 — Implement `worker_from_capnp()` adapter in worker.c**
```c
int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len) {
xce_worker_t host = {0};
/* Deserialize Worker Cap'n Proto into xce_worker_t */
/* Deep-copy string fields (strdup name, copy resource arrays) */
return restore_worker(c, host);
}
```
- [ ] **Step 11.4 — Implement `persist_recover()` in persist.c**
Full sequence (Part 2.5):
1. Scan `snap_path` for snapshot files, pick highest LSN
2. Verify `headerChecksum` then `checksum` — try next-oldest on failure
3. Check `SnapshotHeader.version <= WAL_FORMAT_VERSION` — abort if higher
4. Deserialize: jobs → `restore_job()`, workers → `worker_from_capnp()`, queues → `create_queue()`
5. Open WAL segments from `snapshot.lsn + 1`, replay via `apply_wal_entry()`
- Skip unknown entry types with `WARN`, but abort if `type > WAL_ENTRY_TYPE_MAX`
6. Retention pass: run GC check on all entities (TTL expired during downtime)
7. Set `ha.term = max(snapshot_header.term, max_term_from_wal)`
8. Trim WAL segments fully covered by snapshot LSN
- [ ] **Step 11.5 — Run tests**
```bash
ninja -C build && meson test -C build -t 0 --suite controller
```
Expected: **PASS**.
- [ ] **Step 11.6 — Commit**
```bash
git add src/controller/persist.c src/controller/job.c src/controller/worker.c \
src/controller/testsuite/test_persist_recovery.c
git commit -m "feat: implement persist_recover() with snapshot+WAL replay and retention pass (P0-4)"
```
---
## Task 12 — P0-D: hook detail enums + reply ordering (P0-4a + P0-7)
**Reads:** Design doc Parts 2.3 footnotes 1–2, Part 8.3, P0-4a, P0-7.
**Files:**
- Modify: `src/include/xce/cluster_def.h`
- Modify: `src/controller/job.c`
- Audit: all handler files for `xce_run_hooks(AFTER_EVENT` + `xce_reply` ordering
- [ ] **Step 12.1 — Write failing test for new hook details**
```c
void test_job_stop_fires_correct_hook(void) {
/* Set up cluster with a mock hook that records detail */
/* Call on_job_stop() */
/* Assert HOOK_DETAIL_JOB_STOP was fired (not HOOK_DETAIL_JOB_UPDATE) */
}
```
Expected: **FAIL** — `HOOK_DETAIL_JOB_STOP` not defined.
- [ ] **Step 12.2 — Add hook detail enums**
In `src/include/xce/cluster_def.h`, add to the hook detail enum:
```c
HOOK_DETAIL_JOB_STOP = <next_value>,
HOOK_DETAIL_JOB_RESUME = <next_value + 1>,
```
- [ ] **Step 12.3 — Update `on_job_stop()` and `on_job_resume()` in job.c**
Find `on_job_stop()` — it currently calls `mark_job_update()`. Replace with direct hook call:
```c
xce_run_hooks(c->hooks, XCE_HOOK_TYPE_AFTER_EVENT, HOOK_DETAIL_JOB_STOP, j);
```
Same for `on_job_resume()` → `HOOK_DETAIL_JOB_RESUME`.
Verify with:
```bash
grep -n 'mark_job_update\|HOOK_DETAIL_JOB' src/controller/job.c
```
- [ ] **Step 12.4 — Audit reply-after-hooks ordering across all handlers**
```bash
grep -rn 'xce_run_hooks.*AFTER_EVENT\|xce_reply' src/controller/*.c
```
For each handler file: verify `xce_reply()` comes AFTER `xce_run_hooks(AFTER_EVENT)`. Fix any handler where reply precedes hooks.
- [ ] **Step 12.5 — Run tests**
```bash
ninja -C build && meson test -C build -t 0 --suite controller
```
Expected: **PASS**.
- [ ] **Step 12.6 — Commit**
```bash
git add src/include/xce/cluster_def.h src/controller/job.c
git commit -m "feat: add HOOK_DETAIL_JOB_STOP/RESUME; move xce_reply() after AFTER_EVENT hooks (P0-4a, P0-7)"
```
---
## Task 13 — P0-D: persistence hook functions (P0-6)
**Reads:** Design doc Parts 2.3, 8.1–8.3.
**Files:**
- Modify: `src/controller/persist.c`
- [ ] **Step 13.1 — Write failing hook invocation test**
```c
void test_job_submit_writes_wal_sync(void) {
/* Submit a job via handle_jobsubmit() with a wired persist_t */
/* Verify WAL contains a WAL_JOB_SUBMIT entry */
/* Verify xce_reply() received sync_rc == 0 */
}
```
Expected: **FAIL** — hook not registered yet.
- [ ] **Step 13.2 — Implement `persist_job_hook_fun` in persist.c**
```c
static int persist_job_hook_fun(void* arg, int detail, ...) {
persist_t* p = arg;
va_list ap;
va_start(ap, detail);
job_t* job = va_arg(ap, job_t*);
int* sync_rc_out = va_arg(ap, int*);
persist_batch_ctx_t* batch = va_arg(ap, persist_batch_ctx_t*);
uint32_t wal_type = detail_to_wal_type(detail);
if (wal_type == 0) { va_end(ap); return 0; }
/* Serialize job to Cap'n Proto payload (partial for non-submit types) */
/* ... capnp serialization ... */
int rc;
if (wal_type == WAL_JOB_SUBMIT) {
rc = persist_wal_sync_append(p, wal_type, payload_buf, payload_len, sync_rc_out);
} else {
rc = persist_wal_async(p, wal_type, payload_buf, payload_len);
if (sync_rc_out) *sync_rc_out = rc;
}
va_end(ap);
return 0; /* AFTER_EVENT hooks always return 0 — failures logged not propagated */
}
```
Implement `detail_to_wal_type()` mapping table (Part 2.3 table).
- [ ] **Step 13.3 — Implement persist_queue_hook_fun and persist_worker_hook_fun**
Same pattern: serialize queue/worker to Cap'n Proto, call `persist_wal_async()`.
- [ ] **Step 13.4 — Register hooks in `persist_start()`**
```c
xce_add_hook(c->hooks, XCE_HOOK_TYPE_AFTER_EVENT, HOOK_DETAIL_JOB_NEW, persist_job_hook_fun, p);
xce_add_hook(c->hooks, XCE_HOOK_TYPE_AFTER_EVENT, HOOK_DETAIL_JOB_START, persist_job_hook_fun, p);
/* ... all job detail types ... */
xce_add_hook(c->hooks, XCE_HOOK_TYPE_AFTER_EVENT, HOOK_DETAIL_QUEUE_UPDATE, persist_queue_hook_fun, p);
xce_add_hook(c->hooks, XCE_HOOK_TYPE_AFTER_EVENT, HOOK_DETAIL_WORKER_JOIN, persist_worker_hook_fun, p);
/* ... etc. */
```
- [ ] **Step 13.5 — Run tests**
```bash
ninja -C build && meson test -C build -t 0 --suite controller
```
Expected: **PASS**.
- [ ] **Step 13.6 — Commit**
```bash
git add src/controller/persist.c
git commit -m "feat: persistence hook functions wired to WAL write on AFTER_EVENT (P0-6)"
```
---
## Task 14 — P0-E: entity terminal states + deferred GC (P0-5a + P0-8 + P0-13)
**Reads:** Design doc Parts 4.1–4.5, P0-5a, P0-8, P0-13.
**Files:**
- Modify: `src/include/xce/cluster_def.h` or `types.h`
- Modify: `src/controller/worker.c`
- Modify: `src/controller/job.c`
- Modify: `src/lib/kvmgr.c`
- [ ] **Step 14.1 — Write failing test for JOB_STAT_TRANSIENT**
```c
void test_transient_job_auto_terminates(void) {
job_t* j = make_test_job(JOB_STAT_TRANSIENT);
j->transient_since = time(NULL) - 400; /* expired: > default 300s */
persist_config_t cfg = persist_config_defaults();
assert(!job_should_retain(j, &cfg));
}
void test_transient_job_retained_within_timeout(void) {
job_t* j = make_test_job(JOB_STAT_TRANSIENT);
j->transient_since = time(NULL) - 100; /* not expired */
persist_config_t cfg = persist_config_defaults();
assert(job_should_retain(j, &cfg));
}
```
Expected: **FAIL** — `JOB_STAT_TRANSIENT` and `transient_since` not defined.
- [ ] **Step 14.2 — Add terminal state bitmasks**
In `src/include/xce/types.h` (or `cluster_def.h`, wherever job status bits are defined):
```c
#define JOB_STAT_TRANSIENT 0x400 /* MUST NOT be 0x200 — that is JOB_STAT_WAIT */
```
Add `XCE_WORKER_HIDDEN` and `QUEUE_STAT_HIDDEN` to their respective enums.
Add `time_t transient_since` field to `job_t`.
- [ ] **Step 14.3 — Implement `job_is_terminal()` and `job_should_retain()`**
In `src/controller/job.c`:
```c
bool job_is_terminal(const job_t* j) {
return (j->stat & (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 if timeout elapsed */
if (j->stat & JOB_STAT_TRANSIENT) {
if (j->transient_since > 0 &&
now > j->transient_since + (time_t)cfg->transient_job_timeout_s) {
/* Will be auto-terminated by checkpoint GC */
return false;
}
return true; /* Not yet timed out */
}
if (!job_is_terminal(j)) return true;
if (job_has_dependents(j)) return true;
return (j->finish_time + (time_t)cfg->finished_retain_s >= now);
}
```
- [ ] **Step 14.4 — Extend `mark_worker_unavail()` to set `dead_since`**
In `src/controller/worker.c`, in `mark_worker_unavail()`, add:
```c
w->host.dead_since = time(NULL);
```
- [ ] **Step 14.5 — Add P0-8a TODO markers for endpoint cleanup**
In `src/controller/cluster.c`, find `handle_endpoint_error()` calls to `zarray_del()`/`zarray_clear()`. Add:
```c
/* TODO P0-8b: replace with deferred cleanup after P0-0a zarray merge */
```
- [ ] **Step 14.6 — Run tests**
```bash
ninja -C build && meson test -C build -t 0 --suite controller
```
Expected: **PASS**.
- [ ] **Step 14.7 — Commit**
```bash
git add src/include/xce/types.h src/controller/job.c src/controller/worker.c \
src/controller/cluster.c
git commit -m "feat: add JOB_STAT_TRANSIENT, job_should_retain, worker dead_since (P0-5a, P0-8, P0-13)"
```
---
## Task 15 — P0-F: cluster wiring (P0-5 + P0-10 + P0-11)
**Reads:** Design doc Parts 2.6–2.7, P0-5, P0-10, P0-11.
**Files:**
- Modify: `src/include/xce/cluster_def.h` (add `persist_t persist` to `cluster_t`)
- Modify: `src/controller/cluster.c` (`cluster_init/run/fini`)
- Modify: `src/controller/myconf.c` (YAML parsing)
- [ ] **Step 15.1 — Write pre-addition audit**
```bash
grep -rn 'sizeof.*cluster_t\|offsetof.*cluster\|cluster_t' src/ | grep -v '\.h' | head -20
```
Verify no code assumes fixed `sizeof(cluster_t)` before adding the large `persist_t` field.
- [ ] **Step 15.2 — Add `persist_t persist` to `cluster_t`**
In `src/include/xce/cluster_def.h`:
```c
typedef struct cluster_s {
/* ... existing fields ... */
persist_t persist; /* Persistence module — added P0-F */
} cluster_t;
```
- [ ] **Step 15.3 — Wire persist lifecycle in cluster.c**
In `cluster_init()`:
```c
persist_config_t pcfg = persist_config_from_myconf(&c->conf);
if (persist_init(&c->persist, &pcfg, c) != 0) return -1;
if (persist_recover(&c->persist) != 0) return -1;
```
In `cluster_run()` (after cluster is fully initialized):
```c
persist_start(&c->persist);
```
In `cluster_fini()`:
```c
persist_stop(&c->persist);
persist_fini(&c->persist);
```
- [ ] **Step 15.4 — Add persistence config YAML parsing in myconf.c**
In `myconf.c`, parse from `cluster.bin` config:
```c
cfg->persist.wal_sync_ms = yaml_get_uint(node, "wal_sync_ms", DEFAULT_WAL_SYNC_MS);
cfg->persist.snap_interval_s = yaml_get_uint(node, "snap_interval_s", DEFAULT_SNAP_INTERVAL_S);
cfg->persist.snap_wal_size_mb = yaml_get_uint(node, "snap_wal_size_mb", DEFAULT_SNAP_WAL_SIZE_MB);
cfg->persist.snap_keep_n = yaml_get_uint(node, "snap_keep_n", DEFAULT_SNAP_KEEP_N);
cfg->persist.finished_retain_s = yaml_get_uint(node, "finished_retain_s", DEFAULT_FINISHED_RETAIN_S);
cfg->persist.sync_timeout_ms = yaml_get_uint(node, "sync_timeout_ms", DEFAULT_SYNC_TIMEOUT_MS);
```
- [ ] **Step 15.5 — Add `--workdir` and `-n` CLI flags to main.c**
In `src/controller/main.c`, add:
```c
"--workdir", required_argument, ...
"-n", required_argument, ...
```
Pass `workdir` and `name` to `persist_init()` to derive `effective_dir`.
- [ ] **Step 15.6 — Build and run all tests**
```bash
ninja -C build && meson test -C build
```
Expected: **PASS** (all suites).
- [ ] **Step 15.7 — Commit**
```bash
git add src/include/xce/cluster_def.h src/controller/cluster.c \
src/controller/myconf.c src/controller/main.c
git commit -m "feat: wire persist_init/recover/start/stop into cluster lifecycle (P0-F)"
```
---
## Task 16 — P0-G: end-to-end smoke tests
**Reads:** Design doc P0-G (P0-15a through P0-15e).
**Files:**
- Create: `src/controller/testsuite/test_e2e_wal.c`
- Create: `src/controller/testsuite/test_e2e_checkpoint.c`
- Create: `src/controller/testsuite/test_e2e_hooks.c`
- Create: `src/controller/testsuite/test_e2e_gc.c`
- Create: `src/controller/testsuite/test_e2e_capnp_merge.c`
- [ ] **Step 16.1 — P0-15a: WAL E2E — write/crash/recover**
```c
void test_wal_torn_write_recovery(void) {
/* 1. Write 10 WAL entries */
/* 2. Truncate last entry mid-write (simulate crash) using ftruncate */
/* 3. Call persist_recover() */
/* 4. Verify only 9 entries replayed (last one discarded at CRC check) */
}
void test_wal_chained_crc_detects_corruption(void) {
/* 1. Write 5 WAL entries */
/* 2. Corrupt bytes in entry 3's payload */
/* 3. persist_recover() must stop replay at entry 2 */
}
```
Run, fix, pass, commit: `"test: WAL E2E — torn write and CRC corruption recovery (P0-15a)"`
- [ ] **Step 16.2 — P0-15b: Checkpoint E2E — round trip**
```c
void test_checkpoint_recovery_entity_counts(void) {
/* 1. Add 100 jobs, 5 queues, 10 workers to cluster */
/* 2. Trigger checkpoint (writes snapshot) */
/* 3. Add 20 more jobs to WAL (no checkpoint) */
/* 4. Destroy cluster, re-init, call persist_recover() */
/* 5. Assert: 120 jobs, 5 queues, 10 workers present */
}
```
Run, fix, pass, commit: `"test: checkpoint+recovery round-trip entity count (P0-15b)"`
- [ ] **Step 16.3 — P0-15c: Hook E2E — ordering**
```c
void test_reply_after_hook_ordering(void) {
/* Submit a job; intercept hook + reply calls */
/* Assert hook fires BEFORE reply is sent */
}
void test_sync_rc_propagates_on_wal_failure(void) {
/* Inject WAL write failure (close wal_fd) */
/* Submit job — assert client receives error, not success ACK */
}
```
Run, fix, pass, commit: `"test: hook invocation ordering and sync_rc propagation (P0-15c)"`
- [ ] **Step 16.4 — P0-15d: GC E2E — retention and transient jobs**
```c
void test_finished_jobs_cleaned_after_ttl(void) {
/* Create 5 finished jobs with finish_time = now - 7200 (> 3600s default) */
/* Trigger checkpoint GC (thaw with should_retain) */
/* Assert 0 finished jobs survive */
}
void test_transient_job_auto_terminated_at_checkpoint(void) {
/* Create 2 transient jobs: one < 300s ago, one > 300s ago */
/* Trigger checkpoint GC */
/* Assert: 1 job remains (not expired), 1 moved to JOB_STAT_EXIT */
/* Assert ctrl_transient_jobs_stuck counter incremented once */
}
```
Run, fix, pass, commit: `"test: GC retention and JOB_STAT_TRANSIENT auto-termination (P0-15d)"`
- [ ] **Step 16.5 — P0-15e: Cap'n Proto partial-merge semantics**
```c
void test_partial_wal_entry_does_not_zero_unset_fields(void) {
/* Create job with stat=RUN, exec_host="node01", rusage values set */
/* Apply a WAL_JOB_RUSAGE partial payload (only rusage fields set) */
/* Assert exec_host and stat are UNCHANGED after apply_wal_entry() */
}
```
Run, fix, pass, commit: `"test: Cap'n Proto partial WAL merge does not zero absent fields (P0-15e)"`
- [ ] **Step 16.6 — Final full test run**
```bash
ninja -C build && meson test -C build 2>&1
```
Expected: **ALL PASS**. No compiler warnings with `-Wall -Wextra`.
---
## P0 Gate Checklist
Before declaring P0 complete:
- [ ] `meson test -C build` — all tests pass
- [ ] `ninja -C build` — zero warnings with `-Wall -Wextra`
- [ ] Controller starts with `--workdir /tmp/xce -n ctrl-1` and creates `data/wal/` and `data/snap/` directories
- [ ] Controller survives `SIGKILL` and restarts with all jobs/queues/workers intact
- [ ] WAL segments are trimmed after checkpoint
- [ ] `JOB_STAT_TRANSIENT` jobs older than `transient_job_timeout_s` are moved to `JOB_STAT_EXIT` at checkpoint
- [ ] All P0-G smoke tests pass
- [ ] No `TODO P0-` comments remain except P0-8b (which requires P0-0a, tracked separately)
---
## What's next after P0
Create `2026-05-07-p1-ha-foundation.md` covering:
- `ha.c` / `ha.h` — role management, heartbeat watchdog
- `xce_disable_hook()` addition to `hook.c`
- LSF-style priority election (`am_i_primary()`)
- Promotion/demotion sequence
- Redirect logic in `join.c`
- Worker/scheduler reconnect after failover
Reference in a new issue View git blame Copy permalink