Rongsong Shen 59597e105e feat: add --peers to controller for HA peer list and priority election

- myconf_t gains peers_cfg (xce_cluster_config_t) populated from --peers
- controller/main.c: parse --peers host:port,... via argp; ca_cert_path
  auto-derived from workdir/certs/xce.crt (no --ca-cert flag needed)
- cluster_init: derive my_priority from position in peers list (index 0 =
  primary); fall back to 0 when --peers not given (single-node mode)

2026-05-08 04:40:27 +00:00

52 KiB

Raw Blame History

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):

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:
```
#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:
```
/* TODO: implement group building */
```
Step 1.3 — Build and verify no regressions
```
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

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:

#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):

/* --- 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
```
meson test -C build -t 0 --suite lib
```
Expected: PASS.

Step 2.4 — Commit

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

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:

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:

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:

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:

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:

/* 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 */
};

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):

#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;
}

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

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

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:

#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);
}

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:

sleep(mgr->precision);

with:

struct timespec ts = { 0, (long)mgr->precision_ms * 1000000L };
nanosleep(&ts, NULL);

Update delaymgr_init() to store the parameter as precision_ms:

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:
```
delaymgr_init(&c->delaymgr, 200 /* precision_ms */, run_delay_tasks);
```
Step 4.5 — Audit all other delaymgr_init call sites
```
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

ninja -C build && meson test -C build -t 0 --suite lib

Expected: PASS.

Step 4.7 — Commit

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):

#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);
}

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

#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

ninja -C build && meson test -C build -t 0 --suite controller

Expected: PASS (types and size check).

Step 5.4 — Commit

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:

#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;
}

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:

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):
```
typedef struct {
    _Atomic _Bool inuse;
    void*         data;
} xzarray_item_t;
```
Run:
```
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:

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()

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

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:
```
.vfreeze = xzarray_freeze,
.vthaw   = xzarray_thaw,
```

Step 6.8 — Run tests

ninja -C build && meson test -C build -t 0 --suite lib

Expected: PASS.

Step 6.9 — Commit

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):

#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 ... */
}

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

#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():

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:

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

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

ninja -C build && meson test -C build -t 0 --suite controller

Expected: PASS.

Step 7.7 — Commit

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:

/* 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);
}

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:

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

ninja -C build && meson test -C build -t 0 --suite controller

Expected: PASS.

Step 8.4 — Commit

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:

#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);
}

meson test -C build -t 0 --suite lib

Expected: FAIL — kvmgr_capture_keys undefined.

Step 9.2 — Implement kvmgr_capture_keys in kvmgr.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:

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

ninja -C build && meson test -C build -t 0 --suite lib

Expected: PASS.

Step 9.4 — Commit

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:

#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():
```
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

ninja -C build && meson test -C build -t 0 --suite controller

Expected: PASS.

Step 10.5 — Commit

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:

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

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

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

ninja -C build && meson test -C build -t 0 --suite controller

Expected: PASS.

Step 11.6 — Commit

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

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:
```
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:
```
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:
```
grep -n 'mark_job_update\|HOOK_DETAIL_JOB' src/controller/job.c
```
Step 12.4 — Audit reply-after-hooks ordering across all handlers
```
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

ninja -C build && meson test -C build -t 0 --suite controller

Expected: PASS.

Step 12.6 — Commit

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

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

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()

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

ninja -C build && meson test -C build -t 0 --suite controller

Expected: PASS.

Step 13.6 — Commit

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

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):
```
#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:

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:
```
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:
```
/* TODO P0-8b: replace with deferred cleanup after P0-0a zarray merge */
```

Step 14.6 — Run tests

ninja -C build && meson test -C build -t 0 --suite controller

Expected: PASS.

Step 14.7 — Commit

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
```
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:

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():

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):

persist_start(&c->persist);

In cluster_fini():

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:

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:
```
"--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
```
ninja -C build && meson test -C build
```
Expected: PASS (all suites).

Step 15.7 — Commit

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

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

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

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

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

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
```
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

52 KiB Raw Blame History Unescape Escape

P0: Single-Controller Persistence — Implementation Plan

Sub-phase dependency order

Task 1 — P0-A: Dead Code Cleanup

Task 2 — P0-A: augment include/config.h with system constants

Task 3 — P0-A: rtti + zarray prerequisites (P0-0a + P0-0b)

Task 4 — P0-A: delaymgr precision enhancement (P0-2b)

Task 5 — P0-B: WAL types and entry format (headers)

Task 6 — P0-B: zarray COW extensions (P0-2)

Task 7 — P0-B: WAL writer + fsync thread (P0-1)

Task 8 — P0-C: Cap'n Proto schemas (P0-9)

Task 9 — P0-C: kvmgr key capture (P0-14 + P0-14a)

Task 10 — P0-C: checkpoint (P0-3)

Task 11 — P0-C: recovery (P0-4 + P0-12 + P0-12a)

Task 12 — P0-D: hook detail enums + reply ordering (P0-4a + P0-7)

Task 13 — P0-D: persistence hook functions (P0-6)

Task 14 — P0-E: entity terminal states + deferred GC (P0-5a + P0-8 + P0-13)

Task 15 — P0-F: cluster wiring (P0-5 + P0-10 + P0-11)

Task 16 — P0-G: end-to-end smoke tests

P0 Gate Checklist

What's next after P0

52 KiB

Raw Blame History

Task 2 — P0-A: augment `include/config.h` with system constants