> **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 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 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 #include 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 #include 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 #include #include #include #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 #include 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 #include #include 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 #include #include #include #include #include #include #include #include #include #include #include #include /* 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 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 #include 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 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-.tmp` → CRC32 → `rename()` to `snap-.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 = , HOOK_DETAIL_JOB_RESUME = , ``` - [ ] **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