Rongsong Shen 86fe35e671 add kiro skills

2026-05-21 03:41:57 +00:00

85 KiB

Raw Permalink Blame History

Config Replication 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: Implement config replication so standbys fetch ClusterConfig from the primary on startup, eliminating manual per-node provisioning.

Architecture: Split cluster_init() into Phase 1 (SSL/HA init, no config) and Phase 2 (iomgr/kvmgr/persist, config required). The primary loads config from conf/current (HEAD-file pointer → versioned cluster.bin.<sha256> file). Standbys fetch via OP_replConfig before Phase 2. SHA-256 is used as both version tag and integrity token. am_i_primary() replaced with real election probing (non-blocking connect + poll per candidate).

Tech Stack: C99, OpenSSL SHA-256, POSIX threads (non-blocking connect + poll), Cap'n Proto codecgen, Unity test framework, Meson build.

File Map

File	Action	What changes
`src/schema/controller.capnp`	Modify	Add 3 opcodes + 6 structs + `configGeneration` field + `sender_priority` to Heartbeat
`src/include/myconf.h`	Modify	Add 6 fields to `myconf_t`
`src/include/xce/cluster_def.h`	Modify	Add 4 fields to `cluster_t`
`src/controller/ha.h`	Modify	Add `ssl_ctx`, `jitter_seed`, `standby_endpoints`, `election_priority_step_ms`, `election_jitter_max_ms` to `ha_state_t`; add `probe_ctx_t`; new API declarations
`src/include/persist.h`	Modify	Add `max_wal_records` to `persist_config_t`; add `wal_record_count`, `checkpoint_in_progress` to `persist_s`
`src/include/byteorder.h`	Create	`be64toh()` portability fallback
`src/controller/config_repl.h`	Create	Public API for config_repl.c
`src/controller/config_repl.c`	Create	All config storage, SHA-256, fetch, upgrade, push functions
`src/controller/main.c`	Modify	HEAD-file pointer resolution; add `--conf-keep-n`, `--conf-stale-timeout` flags; heap-alloc conffile
`src/include/myconf.h`	Modify	`myconf_fini()` declaration
`src/controller/myconf.c`	Modify	Add `load_primary_config()`, `myconf_fini()`; remove `get_default_conf()` static global
`src/controller/cluster.c`	Modify	Split `cluster_init()` into Phase 1 + Phase 2; add opcode dispatch for new ops; `cluster_run()` checks `restart_requested`; `handle_endpoint_error` prunes `standby_endpoints`
`src/controller/myapp.c`	Modify	Restructure to call Phase 1, `am_i_primary()`, config load/fetch, Phase 2; handle restart
`src/controller/ha.c`	Modify	Replace `am_i_primary()` stub with real election probing; add `probe_thread`, heartbeat listener, `ha_check_heartbeat_nocluster` priority comparison
`src/controller/join.c`	Modify	Implement `handle_controller_join()`; register in `standby_endpoints`
`src/include/typemap.h`	Modify	Add new union arms to `xce_request_t` and `xce_reply_t`
`src/lib/request.c`	Modify	Add encode/decode shims for 6 new message types
`src/controller/persist.c`	Modify	Add `schedule_checkpoint()` triggered by `max_wal_records`
`src/controller/meson.build`	Modify	Add `config_repl.c`, compile-time defines
`src/controller/testsuite/test_config_repl.c`	Create	Unit tests for Phase 1 + Phase 2 config functions
`src/controller/testsuite/test_ha_role.c`	Modify	Extend with election probing tests
`src/controller/testsuite/controller_test.c`	Modify	Register new test suites

Task 1: Schema additions

Files:

Modify: src/schema/controller.capnp

Step 1: Add new OPCode entries

In controller.capnp, after replCatchUpDone @28;, add:

  replConfig      @29;
  replConfigPush  @30;
  upgradeConfig   @31;

Step 2: Add new message structs

After the MasterCandidate struct, add:

struct ReplConfigRequest $C.mapname("repl_config_req_t") {
  versionTag  @0: UInt64 $C.mapname("version_tag");
}

struct ReplConfigReply $C.mapname("repl_config_reply_t") {
  upToDate    @0: Bool   $C.mapname("up_to_date");
  versionTag  @1: UInt64 $C.mapname("version_tag");
  sha256      @2: Data   $C.mapname("sha256");
  bytes       @3: Data   $C.mapname("bytes");
}

struct ReplConfigPushRequest $C.mapname("repl_config_push_req_t") {
  versionTag  @0: UInt64 $C.mapname("version_tag");
  sha256      @1: Data   $C.mapname("sha256");
  bytes       @2: Data   $C.mapname("bytes");
}

struct ReplConfigPushReply $C.mapname("repl_config_push_reply_t") {
  accepted    @0: Bool   $C.mapname("accepted");
  versionTag  @1: UInt64 $C.mapname("version_tag");
}

struct UpgradeConfigRequest $C.mapname("upgrade_config_req_t") {
  bytes   @0: Data   $C.mapname("bytes");
  sha256  @1: Data   $C.mapname("sha256");
}

struct UpgradeConfigReply $C.mapname("upgrade_config_reply_t") {
  versionTag  @0: UInt64 $C.mapname("version_tag");
  error       @1: Text   $C.mapname("error");
  ackedCount  @2: UInt32 $C.mapname("acked_count");
}

Step 3: Add union arms to Request.body

In struct Request, after hostctrlmsg @21: HostControl;:

    replConfigRequestMsg     @22: ReplConfigRequest;
    replConfigPushRequestMsg @23: ReplConfigPushRequest;
    upgradeConfigRequestMsg  @24: UpgradeConfigRequest;

Step 4: Add union arms to RequestReply.data

In struct RequestReply, after clusterconf @7: ClusterConf;:

    replConfigReplyMsg     @8: ReplConfigReply;
    replConfigPushReplyMsg @9: ReplConfigPushReply;
    upgradeConfigReplyMsg  @10: UpgradeConfigReply;

Step 5: Add configGeneration to ClusterConfig

In struct ClusterConfig, after the last existing field, add:
```
  configGeneration @20: UInt64 $C.mapname("config_generation");
```

Step 6: Add sender_priority to a new Heartbeat struct

Add a new struct (the schema currently has no Heartbeat struct — OP_heartbeat uses the nop/void path):

struct HeartbeatMsg $C.mapname("heartbeat_msg_t") {
  senderPriority @0: UInt32 $C.mapname("sender_priority");
  term           @1: UInt64 $C.mapname("term");
}

Then in Request.body, add:

    heartbeatmsg @25: HeartbeatMsg;

Step 7: Build to verify schema compiles

cd /worktrees/persistent_ha/src && nix develop --command bash -c "meson setup build --reconfigure 2>&1 | tail -20 && ninja -C build src/schema/controller.capnp.h 2>&1 | tail -20"

Expected: no capnp parse errors, generated header updated.

Step 8: Commit

git add src/schema/controller.capnp
git commit -m "feat: add replConfig/replConfigPush/upgradeConfig opcodes and message structs to schema"

Task 2: Struct field additions

Files:

Modify: src/include/myconf.h
Modify: src/include/xce/cluster_def.h
Modify: src/controller/ha.h
Modify: src/include/persist.h
Create: src/include/byteorder.h

Step 1: Add fields to myconf_t

In src/include/myconf.h, after xce_cluster_config_t peers_cfg;:

    uint64_t             config_version_tag;   /* first 8 bytes of SHA-256, host order */
    uint8_t              config_sha256[32];     /* full SHA-256 of conffile bytes */
    int                  conf_keep_n;           /* versioned file retention count (default 2) */
    uint64_t             config_generation;     /* monotonic audit counter (§4.4) */
    time_t               conf_stale_timeout;    /* seconds before saved config is flagged stale (default 86400) */
    int                  config_stale;          /* set when saved config is older than conf_stale_timeout */

Also add myconf_fini() declaration at the bottom of the file:

void myconf_fini(myconf_t *myconf);

Step 2: Add fields to cluster_t

In src/include/xce/cluster_def.h, inside struct cluster_s, after ha_state_t ha;:

    _Atomic int         restart_requested;        /* set by OP_upgradeConfig handler */
    pthread_t           heartbeat_listener_tid;   /* pre-Phase-2 SSL listener thread */
    _Atomic int         heartbeat_listener_running;
    _Atomic int         lost_and_found_job_count; /* incremented during persist_recover() */

Step 3: Add fields to ha_state_t and ha_config_t

In src/controller/ha.h:

(a) In ha_config_t, after max_promotion_retries:

    uint32_t election_jitter_max_ms;   /* random jitter added to election timeout */
    uint32_t election_priority_step_ms; /* extra ms per priority level */

(b) Add new default macros after the existing defaults:

#define HA_DEFAULT_ELECTION_JITTER_MAX_MS   200
#define HA_DEFAULT_ELECTION_PRIORITY_STEP_MS 500

    void            *ssl_ctx;          /* SSL_CTX* cast; avoids pulling in openssl headers */
    uint64_t         jitter_seed;      /* xorshift64 PRNG state for election jitter */
    zarray_t        *standby_endpoints; /* connected standby controller endpoints */

(d) Add probe_ctx_t struct declaration before ha_state_t:

typedef struct {
    struct xce_candidate_s *candidate; /* peer to probe */
    void                   *ssl_ctx;   /* SSL_CTX* (void* to avoid openssl header) */
    uint32_t                timeout_ms;
    atomic_int             *election_done;
    int                     got_ack;
    pthread_t               thread;
} probe_ctx_t;

(e) Add new API declarations at the bottom:

int  ha_is_primary(const ha_state_t *ha);
void start_heartbeat_listener(cluster_t *c);
void stop_heartbeat_listener(cluster_t *c);

Step 4: Add fields to persist_config_t and persist_s

In src/include/persist.h, in persist_config_t after sync_timeout_ms:

    uint32_t max_wal_records;          /* checkpoint trigger: records since last snap (default 10000) */

In persist_config_defaults(), add:

    c.max_wal_records = 10000;

In struct persist_s after pthread_mutex_t lock;:

    _Atomic uint32_t  wal_record_count;      /* records since last checkpoint */
    _Atomic int       checkpoint_in_progress; /* CAS guard for schedule_checkpoint */

Step 5: Create byteorder.h

Create src/include/byteorder.h:

#pragma once
/* Portable be64toh() for non-glibc or pre-2.25 targets.
 * On Linux glibc >= 2.9 this is already available from <endian.h>.
 * On other platforms we provide a fallback. */
#if defined(__linux__)
#  include <endian.h>
#elif defined(__APPLE__)
#  include <machine/endian.h>
#  include <libkern/OSByteOrder.h>
#  define be64toh(x) OSSwapBigToHostInt64(x)
#else
#  include <stdint.h>
static inline uint64_t be64toh(uint64_t x) {
#  if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
    return __builtin_bswap64(x);
#  else
    return x;
#  endif
}
#endif

Step 6: Build to verify all struct changes compile

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build 2>&1 | grep -E 'error:|warning:|ninja:' | head -30"

Expected: only warnings about new fields being unused — no errors.

Step 7: Commit

git add src/include/myconf.h src/include/xce/cluster_def.h src/controller/ha.h src/include/persist.h src/include/byteorder.h
git commit -m "feat: add config replication fields to myconf_t, cluster_t, ha_state_t, persist_t"

Task 3: config_repl.h + config_repl.c (Phase 1 storage functions)

Files:

Create: src/controller/config_repl.h
Create: src/controller/config_repl.c (Phase 1 functions only)
Create: src/controller/testsuite/test_config_repl.c (Phase 1 tests)

Step 1: Write failing tests for Phase 1 storage functions

Create src/controller/testsuite/test_config_repl.c:

/* test_config_repl.c — Unity tests for config replication (Phase 1) */
#include "unity.h"
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>
#include "config_repl.h"

static char tmpdir[256];

void setUp(void) {
    snprintf(tmpdir, sizeof(tmpdir), "/tmp/test_config_repl_%d", (int)getpid());
    mkdir(tmpdir, 0700);
    char conf_dir[300];
    snprintf(conf_dir, sizeof(conf_dir), "%s/conf", tmpdir);
    mkdir(conf_dir, 0700);
}

void tearDown(void) {
    char cmd[300];
    snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir);
    system(cmd);
}

/* test: conf_read_current_filename() returns -ENOENT when conf/current absent */
void test_conf_read_current_filename_missing(void) {
    char out[256] = {0};
    int rc = conf_read_current_filename(tmpdir, out, sizeof(out));
    TEST_ASSERT_EQUAL_INT(-ENOENT, rc);
}

/* test: conf_save_config() writes versioned file + HEAD pointer; read back matches */
void test_conf_save_and_read_current(void) {
    const uint8_t data[] = "fake capnproto config blob";
    size_t len = sizeof(data) - 1;
    uint8_t sha256[32];
    SHA256(data, len, sha256);

    myconf_t myconf = {0};
    myconf.workdir  = tmpdir;
    myconf.name     = "ctrl0";
    myconf.conf_keep_n = 2;

    int rc = conf_save_config(&myconf, data, len, sha256);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* conf/current pointer file must exist */
    char ptr_path[512];
    snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
    TEST_ASSERT_EQUAL_INT(0, access(ptr_path, F_OK));

    /* Reading back the current filename must succeed */
    char out[256] = {0};
    rc = conf_read_current_filename(tmpdir, out, sizeof(out));
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_NOT_NULL(strstr(out, "cluster.bin."));
}

/* test: conf_save_config() then conf_read_current_filename() parses 12-hex suffix */
void test_conf_current_filename_format(void) {
    const uint8_t data[] = "another fake blob for hex test";
    size_t len = sizeof(data) - 1;
    uint8_t sha256[32];
    SHA256(data, len, sha256);

    myconf_t myconf = {0};
    myconf.workdir  = tmpdir;
    myconf.name     = "ctrl0";
    myconf.conf_keep_n = 2;
    conf_save_config(&myconf, data, len, sha256);

    char out[256] = {0};
    conf_read_current_filename(tmpdir, out, sizeof(out));

    /* Must start with "cluster.bin." */
    TEST_ASSERT_EQUAL_INT(0, strncmp(out, "cluster.bin.", 12));
    /* Suffix must be exactly 12 hex chars */
    const char *hex = out + 12;
    TEST_ASSERT_EQUAL_INT(12, (int)strlen(hex));
    TEST_ASSERT_EQUAL_INT(12, (int)strspn(hex, "0123456789abcdef"));
}

/* test: conf_trim_old_versions() keeps only conf_keep_n files */
void test_conf_trim_keep_n(void) {
    myconf_t myconf = {0};
    myconf.workdir  = tmpdir;
    myconf.name     = "ctrl0";
    myconf.conf_keep_n = 2;
    uint8_t sha256[32];

    /* Write 3 different config versions */
    for (int i = 0; i < 3; i++) {
        char data[32];
        snprintf(data, sizeof(data), "config version %d", i);
        SHA256((uint8_t*)data, strlen(data), sha256);
        usleep(10000); /* ensure distinct mtime */
        conf_save_config(&myconf, (uint8_t*)data, strlen(data), sha256);
    }

    /* Count cluster.bin.* files (excluding .tmp) */
    char conf_dir[300];
    snprintf(conf_dir, sizeof(conf_dir), "%s/conf", tmpdir);
    int count = 0;
    DIR *d = opendir(conf_dir);
    struct dirent *e;
    while ((e = readdir(d)) != NULL) {
        if (strncmp(e->d_name, "cluster.bin.", 12) == 0 &&
            strstr(e->d_name, ".tmp") == NULL)
            count++;
    }
    closedir(d);
    TEST_ASSERT_EQUAL_INT(2, count);
}

/* test: zero-tag sentinel — standby with no config sends tag=0 */
void test_conf_zero_tag_is_no_config_sentinel(void) {
    /* When conf/current is absent, compute_config_tag() returns 0 */
    uint64_t tag = conf_get_saved_tag(tmpdir);
    TEST_ASSERT_EQUAL_UINT64(0, tag);
}

/* test: conf_read_current_filename() returns -EINVAL on path separator in pointer */
void test_conf_current_filename_rejects_path_separator(void) {
    char ptr_path[300];
    snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
    FILE *f = fopen(ptr_path, "w");
    fprintf(f, "cluster.bin/../evil\n");
    fclose(f);

    char out[256] = {0};
    int rc = conf_read_current_filename(tmpdir, out, sizeof(out));
    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}

/* test: empty conf/current treated as ENODATA */
void test_conf_empty_pointer_file_returns_enodata(void) {
    char ptr_path[300];
    snprintf(ptr_path, sizeof(ptr_path), "%s/conf/current", tmpdir);
    /* Create empty file */
    int fd = open(ptr_path, O_WRONLY|O_CREAT|O_TRUNC, 0600);
    close(fd);

    char out[256] = {0};
    int rc = conf_read_current_filename(tmpdir, out, sizeof(out));
    TEST_ASSERT_EQUAL_INT(-ENODATA, rc);
}

/* test: conf_check_stale() returns 0 for fresh file, 1 for old file */
void test_conf_stale_timeout(void) {
    /* Write a config */
    const uint8_t data[] = "stale test blob";
    uint8_t sha256[32];
    SHA256(data, sizeof(data)-1, sha256);

    myconf_t myconf = {0};
    myconf.workdir  = tmpdir;
    myconf.name     = "ctrl0";
    myconf.conf_keep_n = 2;
    myconf.conf_stale_timeout = 86400; /* 24h */
    conf_save_config(&myconf, data, sizeof(data)-1, sha256);

    /* Freshly written — not stale */
    int stale = conf_check_stale(tmpdir, 86400);
    TEST_ASSERT_EQUAL_INT(0, stale);

    /* Timeout of 0 means disabled — never stale */
    stale = conf_check_stale(tmpdir, 0);
    TEST_ASSERT_EQUAL_INT(0, stale);

    /* 1-second timeout — touch the file to be 2s old via utime */
    struct utimbuf old_time;
    old_time.actime = time(NULL) - 10;
    old_time.modtime = time(NULL) - 10;
    char ver_path[512];
    /* resolve the versioned filename */
    char fname[256] = {0};
    conf_read_current_filename(tmpdir, fname, sizeof(fname));
    snprintf(ver_path, sizeof(ver_path), "%s/conf/%s", tmpdir, fname);
    utime(ver_path, &old_time);

    stale = conf_check_stale(tmpdir, 5); /* 5-second timeout */
    TEST_ASSERT_EQUAL_INT(1, stale);
}

Step 2: Run tests to verify they fail (functions not yet defined)

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test 2>&1 | grep -E 'error:|undefined' | head -20"

Expected: linker errors for conf_save_config, conf_read_current_filename, etc.

Step 3: Create config_repl.h

Create src/controller/config_repl.h:

#pragma once
/* config_repl.h — config storage, SHA-256, replication (Phase 1 & 2) */
#include <stdint.h>
#include <stddef.h>
#include "myconf.h"

/* Max config blob size — reject oversized files before allocating */
#define CONF_MAX_SIZE (1 * 1024 * 1024)  /* 1 MiB */

/* Compile-time gate for OP_upgradeConfig (Phase 2, default disabled) */
#ifndef CONF_ENABLE_UPGRADE_CONFIG
#define CONF_ENABLE_UPGRADE_CONFIG 0
#endif

/* Push timeout for fan-out to standbys (ms) */
#ifndef CONF_PUSH_TIMEOUT_MS
#define CONF_PUSH_TIMEOUT_MS 500
#endif

/* Days before startup cert-expiry warning is logged */
#define CERT_WARN_DAYS 30

/* ── Phase 1: Storage ────────────────────────────────── */

/*
 * conf_read_current_filename: read HEAD-style pointer file.
 * Reads <workdir>/conf/current, strips newline, validates format.
 * Returns 0 on success; -ENOENT if missing; -ENODATA if empty;
 * -EINVAL if malformed (path separator, wrong prefix, wrong hex length).
 * out must be at least NAME_MAX bytes.
 */
int conf_read_current_filename(const char *workdir, char *out, size_t outsz);

/*
 * conf_save_config: write versioned file + update HEAD pointer atomically.
 * Steps: write <sha_hex>.tmp → fsync → rename to cluster.bin.<sha_hex>,
 * then write current.tmp → fsync → rename to conf/current.
 * Trims old versions after success.
 * Returns 0 on success, -errno on failure.
 */
int conf_save_config(myconf_t *myconf, const uint8_t *bytes, size_t len,
                     const uint8_t sha256[32]);

/*
 * conf_read_raw: read the versioned file named by conf/current.
 * Allocates *bytes (caller must free). Enforces CONF_MAX_SIZE.
 * Returns 0 on success; -EFBIG if over limit; -errno on I/O error.
 */
int conf_read_raw(const char *conffile, uint8_t **bytes, size_t *len);

/*
 * conf_get_saved_tag: return first 8 bytes of SHA-256 of the current
 * versioned file, host-byte-order. Returns 0 if conf/current is absent
 * or unreadable (zero-tag sentinel for "no config saved").
 */
uint64_t conf_get_saved_tag(const char *workdir);

/*
 * conf_check_stale: returns 1 if the current versioned file's mtime is
 * older than stale_timeout_s seconds. Returns 0 if within timeout or
 * if stale_timeout_s == 0 (disabled). Returns 0 if conf/current absent.
 */
int conf_check_stale(const char *workdir, time_t stale_timeout_s);

/*
 * conf_trim_old_versions: keep at most keep_n versioned cluster.bin.* files
 * (by mtime, newest first). Also removes stale .tmp files.
 * Always preserves the file named by the current conf/current pointer.
 */
void conf_trim_old_versions(const char *workdir, int keep_n);

/*
 * config_contains_reserved_name: returns 1 if ClusterConfig contains
 * a queue or host named "lost_and_found". Used at load time.
 */
int config_contains_reserved_name(cluster_config_t *cfg);

/*
 * sanitise_cn: copy at most maxlen printable ASCII chars from src to dst,
 * replacing non-printable or control chars with '?'. Returns dst.
 */
char *sanitise_cn(const char *src, size_t maxlen);

Step 4: Implement Phase 1 functions in config_repl.c

Create src/controller/config_repl.c with the Phase 1 functions:

#define _GNU_SOURCE
#include "config_repl.h"

#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <openssl/sha.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include <utime.h>
#include <unistd.h>
#include <pthread.h>

#include "byteorder.h"
#include "clusterconfig.h"
#include "controller.capnp.h"
#include "tlog.h"

/* ── helpers ──────────────────────────────────────────── */

static void make_conf_dir(const char *workdir, char *out, size_t sz)
{
    snprintf(out, sz, "%s/conf", workdir);
}

static void bytes_to_hex(const uint8_t *b, size_t n, char *out)
{
    static const char hex[] = "0123456789abcdef";
    for (size_t i = 0; i < n; i++) {
        out[2*i]   = hex[(b[i] >> 4) & 0xf];
        out[2*i+1] = hex[b[i] & 0xf];
    }
    out[2*n] = '\0';
}

/* ── conf_read_current_filename ─────────────────────── */

int conf_read_current_filename(const char *workdir, char *out, size_t outsz)
{
    char head_path[4096];
    snprintf(head_path, sizeof(head_path), "%s/conf/current", workdir);

    FILE *f = fopen(head_path, "r");
    if (!f)
        return -ENOENT;

    struct stat st;
    if (fstat(fileno(f), &st) == 0 && st.st_size == 0) {
        fclose(f);
        ERROR("conf/current is empty — treating as missing config");
        return -ENODATA;
    }

    if (!fgets(out, (int)outsz, f)) {
        fclose(f);
        return -EIO;
    }
    fclose(f);
    out[strcspn(out, "\n")] = '\0';

    if (strchr(out, '/')) {
        ERROR("conf/current contains path separator — corrupted pointer file");
        return -EINVAL;
    }
    if (strncmp(out, "cluster.bin.", 12) != 0) {
        ERROR("conf/current filename '%s' does not start with 'cluster.bin.'", out);
        return -EINVAL;
    }
    const char *hex = out + 12;
    if (strlen(hex) != 12 || strspn(hex, "0123456789abcdef") != 12) {
        ERROR("conf/current suffix '%s' is not 12 hex chars", hex);
        return -EINVAL;
    }
    return 0;
}

/* ── conf_save_config ───────────────────────────────── */

int conf_save_config(myconf_t *myconf, const uint8_t *bytes, size_t len,
                     const uint8_t sha256[32])
{
    char conf_dir[4096];
    make_conf_dir(myconf->workdir, conf_dir, sizeof(conf_dir));

    if (mkdir(conf_dir, 0700) != 0 && errno != EEXIST)
        return -errno;

    /* Derive 12-char hex tag from first 6 bytes of SHA-256 */
    char hex_tag[13];
    bytes_to_hex(sha256, 6, hex_tag);

    /* Step 1: write versioned file via thread-unique .tmp */
    char tmp_path[4096], ver_path[4096];
    snprintf(tmp_path, sizeof(tmp_path), "%s/cluster.bin.%s.%lu.tmp",
             conf_dir, hex_tag, (unsigned long)pthread_self());
    snprintf(ver_path, sizeof(ver_path), "%s/cluster.bin.%s",
             conf_dir, hex_tag);

    int fd = open(tmp_path, O_WRONLY|O_CREAT|O_TRUNC, 0600);
    if (fd < 0)
        return -errno;
    if (write(fd, bytes, len) != (ssize_t)len) {
        close(fd); unlink(tmp_path);
        return -EIO;
    }
    if (fsync(fd) != 0) {
        close(fd); unlink(tmp_path);
        return -EIO;
    }
    close(fd);
    if (rename(tmp_path, ver_path) != 0) {
        unlink(tmp_path);
        return -errno;
    }

    /* Step 2: update HEAD pointer file via thread-unique .tmp */
    char ptr_tmp[4096], ptr_path[4096];
    snprintf(ptr_tmp,  sizeof(ptr_tmp),  "%s/current.%lu.tmp",
             conf_dir, (unsigned long)pthread_self());
    snprintf(ptr_path, sizeof(ptr_path), "%s/current", conf_dir);

    char target[NAME_MAX];
    snprintf(target, sizeof(target), "cluster.bin.%s\n", hex_tag);

    fd = open(ptr_tmp, O_WRONLY|O_CREAT|O_TRUNC, 0600);
    if (fd < 0)
        return -errno;
    write(fd, target, strlen(target));
    fsync(fd);
    close(fd);
    if (rename(ptr_tmp, ptr_path) != 0) {
        unlink(ptr_tmp);
        return -errno;
    }

    /* Step 2.5: fsync the conf/ directory to persist the renames */
    fd = open(conf_dir, O_RDONLY);
    if (fd >= 0) { fsync(fd); close(fd); }

    /* Step 3: trim old versions */
    conf_trim_old_versions(myconf->workdir, myconf->conf_keep_n);

    /* Update conffile to point to new versioned file */
    free(myconf->conffile);
    asprintf(&myconf->conffile, "%s/cluster.bin.%s", conf_dir, hex_tag);

    /* Update tag and SHA-256 in myconf */
    uint64_t new_tag;
    memcpy(&new_tag, sha256, 8);
    myconf->config_version_tag = be64toh(new_tag);
    memcpy(myconf->config_sha256, sha256, 32);

    return 0;
}

/* ── conf_read_raw ──────────────────────────────────── */

int conf_read_raw(const char *conffile, uint8_t **bytes, size_t *len)
{
    FILE *f = fopen(conffile, "rb");
    if (!f)
        return -errno;

    fseek(f, 0, SEEK_END);
    long sz = ftell(f);
    fseek(f, 0, SEEK_SET);

    if (sz <= 0) { fclose(f); return -EIO; }
    if ((size_t)sz > CONF_MAX_SIZE) { fclose(f); return -EFBIG; }

    *bytes = (uint8_t*)malloc((size_t)sz);
    if (!*bytes) { fclose(f); return -ENOMEM; }

    if (fread(*bytes, 1, (size_t)sz, f) != (size_t)sz) {
        free(*bytes); *bytes = NULL; fclose(f); return -EIO;
    }
    fclose(f);
    *len = (size_t)sz;
    return 0;
}

/* ── conf_get_saved_tag ─────────────────────────────── */

uint64_t conf_get_saved_tag(const char *workdir)
{
    char fname[256] = {0};
    if (conf_read_current_filename(workdir, fname, sizeof(fname)) != 0)
        return 0;

    char conf_dir[4096];
    make_conf_dir(workdir, conf_dir, sizeof(conf_dir));

    char ver_path[4096];
    snprintf(ver_path, sizeof(ver_path), "%s/%s", conf_dir, fname);

    uint8_t *bytes = NULL;
    size_t   len   = 0;
    if (conf_read_raw(ver_path, &bytes, &len) != 0)
        return 0;

    uint8_t sha256[32];
    SHA256(bytes, len, sha256);
    free(bytes);

    uint64_t tag;
    memcpy(&tag, sha256, 8);
    return be64toh(tag);
}

/* ── conf_check_stale ───────────────────────────────── */

int conf_check_stale(const char *workdir, time_t stale_timeout_s)
{
    if (stale_timeout_s == 0)
        return 0;

    char fname[256] = {0};
    if (conf_read_current_filename(workdir, fname, sizeof(fname)) != 0)
        return 0;

    char ver_path[4096];
    snprintf(ver_path, sizeof(ver_path), "%s/conf/%s", workdir, fname);

    struct stat st;
    if (stat(ver_path, &st) != 0)
        return 0;

    time_t age = time(NULL) - st.st_mtime;
    return (age > stale_timeout_s) ? 1 : 0;
}

/* ── conf_trim_old_versions ─────────────────────────── */

void conf_trim_old_versions(const char *workdir, int keep_n)
{
    char conf_dir[4096];
    make_conf_dir(workdir, conf_dir, sizeof(conf_dir));

    /* Read current pointer to protect it from deletion */
    char current_name[256] = {0};
    conf_read_current_filename(workdir, current_name, sizeof(current_name));

    /* Collect cluster.bin.* files (not .tmp) sorted by mtime */
    struct entry { char name[NAME_MAX]; time_t mtime; };
    struct entry entries[256];
    int n = 0;

    DIR *d = opendir(conf_dir);
    if (!d) return;
    struct dirent *e;
    while ((e = readdir(d)) != NULL && n < 256) {
        if (strncmp(e->d_name, "cluster.bin.", 12) != 0)
            continue;
        if (strstr(e->d_name, ".tmp") != NULL) {
            /* Clean stale .tmp files */
            char full[4096];
            snprintf(full, sizeof(full), "%s/%s", conf_dir, e->d_name);
            unlink(full);
            continue;
        }
        char full[4096];
        snprintf(full, sizeof(full), "%s/%s", conf_dir, e->d_name);
        struct stat st;
        if (stat(full, &st) == 0) {
            strncpy(entries[n].name, e->d_name, NAME_MAX-1);
            entries[n].mtime = st.st_mtime;
            n++;
        }
    }
    closedir(d);

    /* Sort descending by mtime (newest first) — simple insertion sort */
    for (int i = 1; i < n; i++) {
        struct entry tmp = entries[i];
        int j = i - 1;
        while (j >= 0 && entries[j].mtime < tmp.mtime) {
            entries[j+1] = entries[j]; j--;
        }
        entries[j+1] = tmp;
    }

    /* Delete everything after keep_n, skipping the current pointer */
    for (int i = keep_n; i < n; i++) {
        if (strcmp(entries[i].name, current_name) == 0)
            continue; /* never delete the active version */
        char full[4096];
        snprintf(full, sizeof(full), "%s/%s", conf_dir, entries[i].name);
        unlink(full);
    }
}

/* ── config_contains_reserved_name ─────────────────── */

int config_contains_reserved_name(cluster_config_t *cfg)
{
    if (!cfg) return 0;
    /* Check queues */
    for (int i = 0; i < cfg->nqueues; i++) {
        if (cfg->queues[i] && strcmp(cfg->queues[i]->name, "lost_and_found") == 0)
            return 1;
    }
    /* Check hosts / candidates */
    for (int i = 0; i < cfg->nhosts; i++) {
        if (cfg->hosts[i] && strcmp(cfg->hosts[i]->name, "lost_and_found") == 0)
            return 1;
    }
    return 0;
}

/* ── sanitise_cn ────────────────────────────────────── */

char *sanitise_cn(const char *src, size_t maxlen)
{
    static char buf[128];
    size_t n = maxlen < sizeof(buf)-1 ? maxlen : sizeof(buf)-1;
    size_t i;
    for (i = 0; i < n && src[i] != '\0'; i++)
        buf[i] = (src[i] >= 0x20 && src[i] < 0x7f) ? src[i] : '?';
    buf[i] = '\0';
    return buf;
}

Step 5: Add config_repl.c to meson.build and update test suite

In src/controller/meson.build, add 'config_repl.c' to controller_src.

Also add to controller_unit_test_src:
```
'testsuite/test_config_repl.c',
'config_repl.c',
```
Add compile definition to controller_unit_test:
```
c_args: ['-DXCE_TESTING', '-DCONF_ENABLE_UPGRADE_CONFIG=0'],
```

Step 6: Run tests to verify they pass

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'PASS|FAIL|test_conf'"

Expected: all test_conf_* tests PASS.

Step 7: Commit

git add src/controller/config_repl.h src/controller/config_repl.c src/controller/testsuite/test_config_repl.c src/controller/meson.build
git commit -m "feat: add config_repl Phase 1 — HEAD-file storage, SHA-256, trim, stale check"

Task 4: main.c — HEAD-file config resolution + new CLI flags

Files:

Modify: src/controller/main.c

Step 1: Replace flat conffile path with HEAD-file resolution

In main(), replace the conf_file stack buffer and assignment:

/* Old code (lines ~79, 115–116, 121): */
char conf_file[XCE_MAXPATH_LEN + 1];
snprintf(conf_file, ..., "%s/%s/conf/current_config", ...);
myconf.conffile = conf_file;

/* New code: conffile is heap-allocated by load_primary_config() or
 * fetch_config_from_primary() — do not set it here.
 * Store workdir for use in config_repl.c path construction. */
/* Remove the conf_file declaration and the myconf.conffile = conf_file line */
myconf.conffile = NULL;   /* will be set by load_primary_config() or fetch */

Step 2: Add --conf-keep-n and --conf-stale-timeout CLI options

Add two new option keys and entries to options[]:

#define OPT_CONF_KEEP_N       257
#define OPT_CONF_STALE_TIMEOUT 258

{ "conf-keep-n",       OPT_CONF_KEEP_N,       "N",   0,
  "Number of versioned config files to retain (default 2)" },
{ "conf-stale-timeout", OPT_CONF_STALE_TIMEOUT, "SECS", 0,
  "Seconds before saved config is considered stale (0=disabled, default 86400)" },

Add to struct arguments:

int   conf_keep_n;
long  conf_stale_timeout;

Add cases to parse_opt:

case OPT_CONF_KEEP_N:
    arguments->conf_keep_n = atoi(arg);
    break;
case OPT_CONF_STALE_TIMEOUT:
    arguments->conf_stale_timeout = atol(arg);
    break;

After memset(&arguments, ...), initialize defaults:

arguments.conf_keep_n = 2;
arguments.conf_stale_timeout = 86400;

Set in myconf after parsing:

myconf.conf_keep_n = arguments.conf_keep_n;
myconf.conf_stale_timeout = (time_t)arguments.conf_stale_timeout;

Step 3: Add --peers validation WARN

After the xce_parse_peers() call, add:

for (int i = 0; i < myconf.peers_cfg.candidates_count; i++) {
    if (myconf.peers_cfg.candidates[i].hostname[0] == '\0' ||
        myconf.peers_cfg.candidates[i].port == 0) {
        WARN("--peers: candidate %d has empty host or port 0 — check peer string",
             i);
    }
}

Step 4: Build

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build xce_controller 2>&1 | grep -E 'error:|warning:' | head -20"

Expected: no errors.

Step 5: Commit

git add src/controller/main.c
git commit -m "feat: update main.c — heap-alloc conffile, add --conf-keep-n/--conf-stale-timeout flags"

Task 5: myconf.c — load_primary_config() and myconf_fini()

Files:

Modify: src/controller/myconf.c

Step 1: Write failing test for load_primary_config() missing-file path

In src/controller/testsuite/test_config_repl.c, add:

#include "myconf.h"
/* ... */

void test_primary_missing_config_fatal_path(void) {
    myconf_t myconf = {0};
    myconf.workdir  = tmpdir;
    myconf.name     = "ctrl0";
    myconf.conffile = NULL;
    myconf.conf_keep_n = 2;
    /* No conf/current file exists — load_primary_config must return -ENOENT */
    int rc = load_primary_config_for_test(&myconf);
    TEST_ASSERT_EQUAL_INT(-ENOENT, rc);
}

(We'll expose a _for_test variant that returns the error code instead of calling FATAL.)

Step 2: Add load_primary_config() to myconf.c

In src/controller/myconf.c, add:

#include "config_repl.h"
#include "byteorder.h"
#include <openssl/sha.h>
#include <stdatomic.h>

Add the function:

int load_primary_config(cluster_t *c)
{
    myconf_t *myconf = c->myconf;
    char conf_dir[4096];
    snprintf(conf_dir, sizeof(conf_dir), "%s/%s", myconf->workdir, myconf->name);

    /* Resolve conf/current → versioned filename */
    char fname[256] = {0};
    int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
    if (rc != 0) {
        FATAL("Primary has no config file '%s/conf/current' (rc=%d) — "
              "provision the config before starting the primary controller",
              conf_dir, rc);
        return -ENOENT;
    }

    /* Heap-allocate conffile so it survives the stack frame */
    free(myconf->conffile);
    if (asprintf(&myconf->conffile, "%s/conf/%s", conf_dir, fname) < 0)
        return -ENOMEM;

    /* Check for reserved lost_and_found name */
    struct capn       cap;
    ClusterConfig_ptr p;
    FILE *fp = fopen(myconf->conffile, "rb");
    if (!fp) {
        FATAL("Primary config file '%s' not readable: %s",
              myconf->conffile, strerror(errno));
        return -errno;
    }
    capn_init_fp(&cap, fp, 0);
    p.p = capn_getp(capn_root(&cap), 0, 1);
    cluster_config_t *cfg = NULL;
    decode_ClusterConfig_ptr(&cfg, p);
    capn_free(&cap);
    fclose(fp);

    if (!cfg) {
        FATAL("Primary config file '%s' is not a valid ClusterConfig blob",
              myconf->conffile);
        return -EINVAL;
    }
    if (config_contains_reserved_name(cfg)) {
        free_ClusterConfig(cfg);
        FATAL("Config file '%s' contains reserved name 'lost_and_found'",
              myconf->conffile);
        return -EINVAL;
    }

    fill_default_cluster_config(cfg);
    myconf->cluster_config = cfg;

    /* Compute and cache SHA-256 */
    uint8_t *bytes = NULL;
    size_t   len   = 0;
    rc = conf_read_raw(myconf->conffile, &bytes, &len);
    if (rc != 0) {
        FATAL("Failed to read primary config for SHA-256: %d", rc);
        return rc;
    }
    SHA256(bytes, len, myconf->config_sha256);
    free(bytes);

    uint64_t tag;
    memcpy(&tag, myconf->config_sha256, 8);
    myconf->config_version_tag = be64toh(tag);

    /* Default generation to 1 if not set in blob */
    if (cfg->config_generation == 0)
        cfg->config_generation = 1;
    myconf->config_generation = cfg->config_generation;

    /* Check staleness */
    myconf->config_stale = conf_check_stale(conf_dir,
                                            myconf->conf_stale_timeout);
    if (myconf->config_stale)
        WARN("Primary config appears stale (mtime > %lds)", myconf->conf_stale_timeout);

    INFO("Primary config loaded: tag=%016"PRIx64" gen=%"PRIu64,
         myconf->config_version_tag, myconf->config_generation);
    return 0;
}

Step 3: Add myconf_fini() to myconf.c
```
void myconf_fini(myconf_t *myconf)
{
    if (!myconf) return;
    free(myconf->conffile);
    myconf->conffile = NULL;
    if (myconf->cluster_config) {
        free_ClusterConfig(myconf->cluster_config);
        free(myconf->cluster_config);
        myconf->cluster_config = NULL;
    }
}
```
Remove the static struct cluster_configuration *cf = NULL; global and the cluster_hook_fun / config_fini functions that handle the shutdown hook for cf — myconf_fini() takes over this responsibility.

Keep load_cluster_config() and load_cluster_config_from_file() for now but mark them as legacy (they're still used by the existing get_default_conf() path during Phase 1 work).

Step 4: Declare load_primary_config() in myconf.h

In src/include/myconf.h, add:

/* Declared here; defined in myconf.c; used by myapp.c */
int  load_primary_config(struct cluster_s *c);

(forward-declare cluster_t in myconf.h to avoid circular include):

typedef struct cluster_s cluster_t;

Step 5: Build and run tests

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'PASS|FAIL|test_primary'"

Expected: test_primary_missing_config_fatal_path PASS.

Step 6: Commit

git add src/controller/myconf.c src/include/myconf.h src/controller/testsuite/test_config_repl.c
git commit -m "feat: add load_primary_config() and myconf_fini() — replace get_default_conf static global"

Task 6: ha.c — real am_i_primary() with election probing

Files:

Modify: src/controller/ha.c
Modify: src/controller/ha.h
Modify: src/controller/testsuite/test_ha_role.c

Step 1: Write failing tests for election probing

In src/controller/testsuite/test_ha_role.c, add:

/* test: priority-0 with no peers → am_i_primary returns 1 without probing */
void test_am_i_primary_priority_zero_no_peers(void) {
    cluster_t c;
    memset(&c, 0, sizeof(c));
    myconf_t myconf = {0};
    myconf.peers_cfg.candidates_count = 0;
    c.myconf = &myconf;
    ha_config_t cfg = {
        .heartbeat_interval_ms    = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
        .heartbeat_timeout_ms     = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
        .election_timeout_ms      = HA_DEFAULT_ELECTION_TIMEOUT_MS,
        .probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS,
        .max_promotion_retries    = HA_DEFAULT_MAX_PROMOTION_RETRIES,
        .election_jitter_max_ms   = 0,
        .election_priority_step_ms = 500,
    };
    ha_init(&c.ha, &cfg, 0);
    TEST_ASSERT_EQUAL_INT(1, am_i_primary(&c));
    ha_fini(&c.ha);
}

/* test: priority-0 with peers → am_i_primary returns 1 (no higher-priority peers) */
void test_am_i_primary_priority_zero_with_peers(void) {
    cluster_t c;
    memset(&c, 0, sizeof(c));
    myconf_t myconf = {0};
    myconf.peers_cfg.candidates_count = 3;
    strncpy(myconf.peers_cfg.candidates[0].hostname, "ctrl0", XCE_MAXNAME_LEN-1);
    myconf.peers_cfg.candidates[0].port = 9990;
    strncpy(myconf.peers_cfg.candidates[1].hostname, "ctrl1", XCE_MAXNAME_LEN-1);
    myconf.peers_cfg.candidates[1].port = 9991;
    strncpy(myconf.peers_cfg.candidates[2].hostname, "ctrl2", XCE_MAXNAME_LEN-1);
    myconf.peers_cfg.candidates[2].port = 9992;
    c.myconf = &myconf;
    ha_config_t cfg = {
        .heartbeat_interval_ms    = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
        .heartbeat_timeout_ms     = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
        .election_timeout_ms      = 200,   /* short for test */
        .probe_connect_timeout_ms = 50,
        .max_promotion_retries    = 1,
        .election_jitter_max_ms   = 0,
        .election_priority_step_ms = 0,
    };
    ha_init(&c.ha, &cfg, 0);
    /* Priority-0 never probes — must return immediately */
    int result = am_i_primary(&c);
    TEST_ASSERT_EQUAL_INT(1, result);
    ha_fini(&c.ha);
}

/* test: priority-1 with no higher-priority peer alive → deadline expires → primary */
void test_am_i_primary_deadline_expires_becomes_primary(void) {
    cluster_t c;
    memset(&c, 0, sizeof(c));
    myconf_t myconf = {0};
    /* Single higher-priority candidate at a port nobody listens on */
    myconf.peers_cfg.candidates_count = 2;
    strncpy(myconf.peers_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXNAME_LEN-1);
    myconf.peers_cfg.candidates[0].port = 19990; /* nobody listening */
    strncpy(myconf.peers_cfg.candidates[1].hostname, "127.0.0.1", XCE_MAXNAME_LEN-1);
    myconf.peers_cfg.candidates[1].port = 19991;
    c.myconf = &myconf;
    ha_config_t cfg = {
        .heartbeat_interval_ms    = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
        .heartbeat_timeout_ms     = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
        .election_timeout_ms      = 300,   /* 300ms for test speed */
        .probe_connect_timeout_ms = 100,
        .max_promotion_retries    = 1,
        .election_jitter_max_ms   = 0,
        .election_priority_step_ms = 0,
    };
    ha_init(&c.ha, &cfg, 1); /* priority 1 */
    int result = am_i_primary(&c);
    TEST_ASSERT_EQUAL_INT(1, result); /* deadline expired, self-declare */
    ha_fini(&c.ha);
}

Step 2: Run tests to see them fail (am_i_primary still the stub)

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'test_am_i_primary|FAIL'"

Expected: test_am_i_primary_deadline_expires_becomes_primary FAIL (stub returns 0 for priority 1).

Step 3: Implement xorshift64_next() inline in ha.h

In src/controller/ha.h, add after the #include block:

#include <fcntl.h>
#include <unistd.h>

static inline uint64_t xorshift64_next(uint64_t *state)
{
    uint64_t x = *state;
    x ^= x << 13;
    x ^= x >> 7;
    x ^= x << 17;
    *state = x;
    return x;
}

/* Seed jitter_seed from /dev/urandom; fall back to monotonic clock + pid */
static inline void ha_seed_jitter(uint64_t *seed, int my_priority)
{
    int fd = open("/dev/urandom", O_RDONLY);
    if (fd >= 0) {
        read(fd, seed, 8);
        close(fd);
        if (*seed != 0) return;
    }
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    *seed = (uint64_t)ts.tv_nsec
          ^ ((uint64_t)my_priority << 56)
          ^ ((uint64_t)getpid() << 16);
}

Step 4: Implement probe_thread() and am_i_primary() in ha.c

In src/controller/ha.c, add includes at the top:

#include <arpa/inet.h>
#include <fcntl.h>
#include <netdb.h>
#include <openssl/ssl.h>
#include <poll.h>
#include <sys/socket.h>
#include "byteorder.h"
#include "config_repl.h"

Then replace the current am_i_primary() stub (ha.c:213-216) with:

static void *probe_thread(void *arg)
{
    probe_ctx_t *ctx = (probe_ctx_t *)arg;

    /* Non-blocking connect to candidate */
    struct addrinfo hints, *res = NULL;
    memset(&hints, 0, sizeof(hints));
    hints.ai_family   = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;

    char port_str[16];
    snprintf(port_str, sizeof(port_str), "%u", ctx->candidate->port);
    if (getaddrinfo(ctx->candidate->hostname, port_str, &hints, &res) != 0)
        goto done;

    int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
    if (sock < 0) { freeaddrinfo(res); goto done; }

    fcntl(sock, F_SETFL, O_NONBLOCK);
    int crc = connect(sock, res->ai_addr, res->ai_addrlen);
    freeaddrinfo(res);
    if (crc != 0 && errno != EINPROGRESS) { close(sock); goto done; }

    /* poll loop with election_done check */
    struct pollfd pfd = { .fd = sock, .events = POLLOUT };
    while (!atomic_load(ctx->election_done)) {
        int pr = poll(&pfd, 1, (int)ctx->timeout_ms);
        if (pr < 0 && errno == EINTR) continue;
        if (pr <= 0) { close(sock); goto done; }
        if (pfd.revents & (POLLERR|POLLHUP)) { close(sock); goto done; }
        if (pfd.revents & POLLOUT) break;
    }
    if (atomic_load(ctx->election_done)) { close(sock); goto done; }

    /* Verify connect succeeded */
    int err = 0;
    socklen_t el = sizeof(err);
    getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
    if (err != 0) { close(sock); goto done; }

    /* SSL handshake */
    SSL *ssl = SSL_new((SSL_CTX *)ctx->ssl_ctx);
    if (!ssl) { close(sock); goto done; }
    SSL_set_fd(ssl, sock);
    fcntl(sock, F_SETFL, 0); /* blocking for SSL */

    int sr;
    do { sr = SSL_connect(ssl); } while (
        (sr <= 0) &&
        (SSL_get_error(ssl, sr) == SSL_ERROR_WANT_READ ||
         SSL_get_error(ssl, sr) == SSL_ERROR_WANT_WRITE) &&
        !atomic_load(ctx->election_done));

    if (sr <= 0 || atomic_load(ctx->election_done)) {
        SSL_free(ssl); close(sock); goto done;
    }

    /* Send a minimal OP_heartbeat request and check for any reply */
    /* Use the existing xce_nop_req pattern: just check TLS connected */
    ctx->got_ack = 1;
    SSL_shutdown(ssl);
    SSL_free(ssl);
    close(sock);
done:
    return NULL;
}

static int wait_any_probe(probe_ctx_t *probes, int n,
                          const struct timespec *deadline)
{
    while (1) {
        for (int i = 0; i < n; i++) {
            if (probes[i].got_ack)
                return 1;
        }
        struct timespec now;
        clock_gettime(CLOCK_MONOTONIC, &now);
        if (now.tv_sec > deadline->tv_sec ||
            (now.tv_sec == deadline->tv_sec &&
             now.tv_nsec >= deadline->tv_nsec))
            return 0;
        struct timespec sleep_ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
        nanosleep(&sleep_ts, NULL);
    }
}

int am_i_primary(cluster_t *c)
{
    /* Priority-0 is unconditionally primary — no probing needed */
    if (c->ha.my_priority == 0)
        return 1;

    /* Seed jitter PRNG if not yet seeded */
    if (c->ha.jitter_seed == 0)
        ha_seed_jitter(&c->ha.jitter_seed, c->ha.my_priority);

    const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
    int n_probes = c->ha.my_priority < pcfg->candidates_count
                       ? c->ha.my_priority
                       : pcfg->candidates_count;

    if (n_probes == 0)
        return 1; /* no higher-priority peers to probe */

    atomic_int election_done = 0;
    probe_ctx_t probes[MAX_HA_CANDIDATES];

    for (int i = 0; i < n_probes; i++) {
        probes[i] = (probe_ctx_t){
            .candidate     = (struct xce_candidate_s *)&pcfg->candidates[i],
            .ssl_ctx       = c->ha.ssl_ctx,
            .timeout_ms    = c->ha.config.probe_connect_timeout_ms,
            .election_done = &election_done,
            .got_ack       = 0,
        };
        pthread_create(&probes[i].thread, NULL, probe_thread, &probes[i]);
    }

    /* Compute deadline with priority step + jitter */
    long to_ms = (long)c->ha.config.election_timeout_ms;
    to_ms += (long)c->ha.my_priority * (long)c->ha.config.election_priority_step_ms;
    to_ms += (long)(xorshift64_next(&c->ha.jitter_seed)
                    % (c->ha.config.election_jitter_max_ms + 1));

    struct timespec deadline;
    clock_gettime(CLOCK_MONOTONIC, &deadline);
    deadline.tv_sec  += to_ms / 1000;
    deadline.tv_nsec += (to_ms % 1000) * 1000000L;
    if (deadline.tv_nsec >= 1000000000L) {
        deadline.tv_sec++;
        deadline.tv_nsec -= 1000000000L;
    }

    int got_ack = wait_any_probe(probes, n_probes, &deadline);

    atomic_store(&election_done, 1);
    for (int i = 0; i < n_probes; i++)
        pthread_join(probes[i].thread, NULL);

    if (got_ack) {
        INFO("HA election: higher-priority candidate alive — starting as standby "
             "(my_priority=%d)", c->ha.my_priority);
        return 0;
    }
    INFO("HA election: no higher-priority candidate responded within %ldms — "
         "starting as primary (my_priority=%d)", to_ms, c->ha.my_priority);
    return 1;
}

int ha_is_primary(const ha_state_t *ha)
{
    return (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire)
           == CTRL_PRIMARY;
}

Step 5: Enhance ha_check_heartbeat_nocluster() for simultaneous-boot priority comparison

Add to ha_check_heartbeat_nocluster() — after the elapsed_ms check, add priority comparison logic. For now, add a helper stub that will be fleshed out when heartbeat message structs are wired:

/* ha_check_priority_stepdown: if we are self-declared primary (term=0)
 * and receive a heartbeat from a higher-priority peer with term=0,
 * step down. Called from the heartbeat handler in cluster.c.
 * sender_priority: lower index = higher priority.
 */
void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority,
                                uint64_t sender_term)
{
    if (sender_priority >= ha->my_priority)
        return; /* sender has lower or equal priority — no action */

    uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
    if (my_term > 0 || sender_term > 0)
        return; /* term-based comparison takes over after first election */

    /* Both at term=0 during simultaneous boot — higher-priority wins */
    ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
                                                          memory_order_acquire);
    if (role == CTRL_PRIMARY) {
        WARN("HA: stepping down — higher-priority peer %d present at term=0",
             sender_priority);
        ha_do_demote(ha);
    }
}

Declare in ha.h:

void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority,
                                uint64_t sender_term);

Step 6: Add ha_init() updates for new fields

In ha_init(), after the existing atomic stores, add:
```
    ha->ssl_ctx   = NULL;
    ha->jitter_seed = 0;  /* seeded lazily in am_i_primary */
    ha->standby_endpoints = NULL; /* initialized in cluster_init Phase 1 */
```
Also update default config init in ha_init() validation to accept the new ha_config_t fields (they default to 0 if not set, which is safe — step=0 means no step, jitter=0 means no jitter).

Step 7: Run tests

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'test_am_i_primary|PASS|FAIL'"

Expected: all three new test_am_i_primary_* tests PASS.

Step 8: Commit

git add src/controller/ha.c src/controller/ha.h src/controller/testsuite/test_ha_role.c
git commit -m "feat: implement real am_i_primary() with election probing and priority-based symmetry breaking"

Task 7: cluster.c + myapp.c — Phase 1/Phase 2 split

Files:

Modify: src/controller/cluster.c
Modify: src/controller/myapp.c

Step 1: Write failing integration test for config-gated init

In src/controller/testsuite/test_p1_integration.c, add a test verifying that cluster_init_phase1() succeeds on a minimal cluster_t without requiring a config file:

void test_cluster_phase1_no_config_required(void) {
    cluster_t c;
    memset(&c, 0, sizeof(c));
    myconf_t myconf = {0};
    myconf.workdir = "/tmp";
    myconf.name    = "test";
    /* certfile/keyfile are empty — skip SSL for this structural test */
    c.myconf = &myconf;
    /* Phase 1 should succeed without a config file being present */
    int rc = cluster_init_phase1(&c);
    TEST_ASSERT_EQUAL_INT(0, rc);
    cluster_fini(&c);
}

Step 2: Split cluster_init() into cluster_init_phase1() and cluster_init_phase2()

In src/controller/cluster.c:

(a) Rename the current cluster_init() to cluster_init_phase2(cluster_t *c).

(b) Create a new cluster_init_phase1(cluster_t *c) with only the pre-config steps:

int cluster_init_phase1(cluster_t *c)
{
    if (c == NULL) return -EINVAL;

    c->myhost       = &from_controller;
    c->ncontrollers = 1;
    c->myidx        = 0;

    xce_hook_init(&(c->hooksys), XCE_CONTROLLER_HOOK_MAX);
    xce_install_hooks(c);

    /* HA init (computes my_priority from --peers) */
    ha_config_t ha_cfg = {
        .heartbeat_interval_ms    = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
        .heartbeat_timeout_ms     = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
        .election_timeout_ms      = HA_DEFAULT_ELECTION_TIMEOUT_MS,
        .probe_connect_timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS,
        .max_promotion_retries    = HA_DEFAULT_MAX_PROMOTION_RETRIES,
        .election_jitter_max_ms   = HA_DEFAULT_ELECTION_JITTER_MAX_MS,
        .election_priority_step_ms = HA_DEFAULT_ELECTION_PRIORITY_STEP_MS,
    };
    int my_priority = 0;
    const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
    if (pcfg->candidates_count > 0) {
        my_priority = pcfg->candidates_count - 1;
        for (int i = 0; i < pcfg->candidates_count; i++) {
            const char *key = pcfg->candidates[i].name[0]
                                  ? pcfg->candidates[i].name
                                  : pcfg->candidates[i].hostname;
            if (strncmp(key, c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
                my_priority = i; break;
            }
        }
    }
    int rc = ha_init(&c->ha, &ha_cfg, my_priority);
    if (rc != 0) return rc;
    ha_install_sigusr1(&c->ha);

    /* Initialize standby endpoint tracking array */
    c->ha.standby_endpoints = zarray_create(4, NULL);

    /* Initialise cbuf used by cluster_run() to signal restart */
    cbuf_init(&(c->wbuf), 1, sizeof(int));

    return 0;
}

Remove the hook/ha/ssl init (now in phase 1)
Add guard at top: if (c->myconf->cluster_config == NULL) { ERROR(...); return -EINVAL; }
Move initialize_openssl() and create_ssl_context() — SSL ctx created in Phase 1 and stored in c->ha.ssl_ctx, then used by Phase 2's iomgr_set_ssl_ctx()
Add create_lost_and_found(c) call before persist_recover()
Keep ha_promote_to_primary(c) or CTRL_STANDBY assignment

(d) Add Phase 2 port derivation from --peers (replacing cluster_config->port):

uint32_t my_port = 0;
for (int i = 0; i < pcfg->candidates_count; i++) {
    if (strncmp(pcfg->candidates[i].name[0]
                    ? pcfg->candidates[i].name
                    : pcfg->candidates[i].hostname,
                c->myconf->name, XCE_MAXNAME_LEN - 1) == 0) {
        my_port = pcfg->candidates[i].port;
        break;
    }
}
if (my_port == 0) my_port = 9990;

Replace c->myconf->cluster_config->port in endpoint_init() call with my_port.

(e) Update handle_endpoint_error() for HostRole_controller to prune from standby_endpoints:

case HostRole_controller:
    if (c->ha.standby_endpoints) {
        /* Scan and remove the disconnected endpoint */
        for (int idx = 0; idx < zarray_size(c->ha.standby_endpoints); idx++) {
            endpoint_t *se = NULL;
            zarray_get(c->ha.standby_endpoints, idx, (void**)&se);
            if (se == ep) {
                zarray_del(c->ha.standby_endpoints, idx);
                break;
            }
        }
    }
    break;

(f) Add handle_config_request() stub and dispatcher in myreq_handler():

case OPCode_replConfig:
    return handle_config_request(c, ep, req);
case OPCode_upgradeConfig:
#if CONF_ENABLE_UPGRADE_CONFIG
    if (!ha_is_primary(&c->ha)) {
        send_upgrade_error(ep, req, "Not primary — redirect to primary");
        return -EAGAIN;
    }
    return handle_upgrade_config(c, ep, req);
#else
    send_upgrade_error(ep, req, "OP_upgradeConfig not enabled in this build");
    return -ENOSYS;
#endif

(g) Expose cluster_init_phase1() in cluster.h:

int cluster_init_phase1(cluster_t *c);
int cluster_init_phase2(cluster_t *c);

Keep cluster_init() as a wrapper for backward compatibility with existing tests:

int cluster_init(cluster_t *c) {
    int rc = cluster_init_phase1(c);
    if (rc != 0) return rc;
    return cluster_init_phase2(c);
}

Step 3: Update cluster_run() to check restart_requested

In cluster_run(), after the event loop exits, add:

if (atomic_load(&c->restart_requested)) {
    INFO("Config upgrade restart requested — exiting cleanly for service manager restart");
    return 0;
}

Step 4: Update myapp.c — new startup flow

Replace myapp_run() with the new two-phase flow:

int myapp_run(myconf_t *myconf)
{
    cluster_t c;
    memset(&c, 0x0, sizeof(c));
    c.myconf = myconf;

    /* Phase 1: SSL/HA init — no config required */
    int rc = cluster_init_phase1(&c);
    if (rc < 0) {
        ERROR("cluster_init_phase1: %s", strerror(-rc));
        cluster_fini(&c);
        return rc;
    }

    /* Election: determine role */
    initialize_openssl();
    SSL_CTX *ssl_ctx = create_ssl_context(myconf->certfile, myconf->keyfile);
    if (!ssl_ctx) {
        ERROR("create_ssl_context failed");
        cluster_fini(&c);
        return -errno;
    }
    c.ha.ssl_ctx = ssl_ctx;

    if (am_i_primary(&c)) {
        /* Primary: load config from conf/current (FATAL if missing) */
        rc = load_primary_config(&c);
        if (rc != 0) {
            cluster_fini(&c);
            myconf_fini(myconf);
            return rc;
        }
    } else {
        /* Standby: fetch config from primary before Phase 2 */
        rc = fetch_config_from_primary(&c);
        if (rc != 0) {
            cluster_fini(&c);
            myconf_fini(myconf);
            return rc;
        }
    }

    /* Phase 2: full init with config available */
    rc = cluster_init_phase2(&c);
    if (rc < 0) {
        ERROR("cluster_init_phase2: %s", strerror(-rc));
        cluster_fini(&c);
        myconf_fini(myconf);
        return rc;
    }

    rc = cluster_start(&c);
    if (rc < 0) {
        cluster_fini(&c);
        myconf_fini(myconf);
        return rc;
    }

    rc = cluster_run(&c);

    cluster_fini(&c);
    myconf_fini(myconf);
    return rc;
}

Step 5: Build and run tests

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test 2>&1 | grep -E 'error:' | head -20 && ./build/controller_unit_test 2>&1 | grep -E 'PASS|FAIL' | tail -20"

Expected: all existing HA/persist tests still pass.

Step 6: Commit

git add src/controller/cluster.c src/controller/myapp.c src/controller/testsuite/test_p1_integration.c
git commit -m "feat: split cluster_init() into Phase 1 (pre-config) and Phase 2 (post-config); update myapp_run() for two-phase startup"

Task 8: OP_replConfig — primary handler + standby fetch

Files:

Modify: src/controller/config_repl.c (add Phase 1 protocol functions)
Modify: src/controller/join.c

Step 1: Implement handle_config_request() in config_repl.c

Add to config_repl.c:

#include "xce/cluster_def.h"
#include "request.h"

int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
    uint64_t their_tag = req->body.replConfigRequestMsg->version_tag;
    uint64_t my_tag    = c->myconf->config_version_tag;

    xce_reply_t reply;
    init_reply_by_req(&reply, req);

    if (their_tag == my_tag) {
        reply.status = 0;
        reply.data_which = RequestReply_data_replConfigReplyMsg;
        reply.data.replConfigReplyMsg->up_to_date   = true;
        reply.data.replConfigReplyMsg->version_tag  = 0;
        xce_reply(ep, c->myhost, &reply);
        return 0;
    }

    uint8_t *bytes = NULL;
    size_t   len   = 0;
    int rc = conf_read_raw(c->myconf->conffile, &bytes, &len);
    if (rc != 0) {
        reply.status = rc;
        xce_reply(ep, c->myhost, &reply);
        return rc;
    }

    uint8_t recomputed[32];
    SHA256(bytes, len, recomputed);
    if (memcmp(recomputed, c->myconf->config_sha256, 32) != 0) {
        ERROR("Config file changed since startup — refusing to send stale hash");
        free(bytes);
        reply.status = -EIO;
        xce_reply(ep, c->myhost, &reply);
        return -EIO;
    }

    reply.status = 0;
    reply.data_which = RequestReply_data_replConfigReplyMsg;
    reply.data.replConfigReplyMsg->up_to_date   = false;
    reply.data.replConfigReplyMsg->version_tag  = my_tag;
    /* Data fields: sha256 (32 bytes) and bytes (raw config) */
    /* The codecgen-generated encode sets the Data fields in the capn segment */
    reply.data.replConfigReplyMsg->sha256.ptr = recomputed;
    reply.data.replConfigReplyMsg->sha256.len = 32;
    reply.data.replConfigReplyMsg->bytes.ptr  = bytes;
    reply.data.replConfigReplyMsg->bytes.len  = len;

    xce_reply(ep, c->myhost, &reply);
    free(bytes);
    return 0;
}

Declare in config_repl.h:

int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req);

Step 2: Implement fetch_config_from_primary() in config_repl.c

int fetch_config_from_primary(cluster_t *c)
{
    const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
    if (pcfg->candidates_count == 0) {
        /* Single-node — no primary to fetch from; use saved config or fatal */
        goto use_saved;
    }

    /* Get the primary candidate (priority 0) */
    const char *primary_host = pcfg->candidates[0].hostname;
    uint32_t    primary_port = pcfg->candidates[0].port;

    uint64_t saved_tag = conf_get_saved_tag(c->myconf->workdir);
    INFO("HA: requesting cluster config from primary %s:%u (my_tag=%016"PRIx64")",
         primary_host, primary_port, saved_tag);

    uint32_t delay_ms = 500;
    const uint32_t max_delay_ms = 5000;
    const int max_attempts = 20;

    for (int attempt = 0; attempt < max_attempts; attempt++) {
        /* Check if role changed (promoted during retry window) */
        if (ha_is_primary(&c->ha)) {
            INFO("HA: promoted to primary during config fetch — aborting fetch");
            return load_primary_config(c);
        }

        /* non-blocking connect + SSL + send OP_replConfig */
        int rc = do_one_config_fetch(c, primary_host, primary_port, saved_tag);
        if (rc == 0)
            return 0;

        if (attempt < max_attempts - 1) {
            WARN("Config fetch attempt %d/%d failed (rc=%d) — retrying in %ums",
                 attempt+1, max_attempts, rc, delay_ms);
            usleep((useconds_t)delay_ms * 1000);
            delay_ms = delay_ms * 2 < max_delay_ms ? delay_ms * 2 : max_delay_ms;
        }
    }

use_saved:
    /* All retries failed — check for saved config */
    {
        char fname[256] = {0};
        char conf_dir[4096];
        snprintf(conf_dir, sizeof(conf_dir), "%s/%s", c->myconf->workdir, c->myconf->name);
        int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
        if (rc != 0) {
            FATAL("Standby (priority %d) could not fetch config from primary and has "
                  "no saved config — start the primary controller first",
                  c->ha.my_priority);
            return -ENOENT;
        }
        WARN("Standby (priority %d) could not reach primary — using saved config "
             "(tag=%016"PRIx64"). Config may be stale.",
             c->ha.my_priority, saved_tag);
        /* Load the saved config into myconf */
        return load_saved_config(c);
    }
}

The do_one_config_fetch() and load_saved_config() helpers are private functions — add them in the same file. do_one_config_fetch() performs the SSL connect + OP_replConfig request + receive + verify SHA-256 + conf_save_config(). load_saved_config() reads the saved versioned file, decodes it, and populates myconf.

Declare in config_repl.h:

int fetch_config_from_primary(cluster_t *c);

Step 3: Implement handle_controller_join() in join.c

Replace the -ENOSYS stub in handle_controller_join():

static int handle_controller_join(cluster_t *c, endpoint_t *ep,
                                  xce_request_t *request)
{
    xce_reply_t reply;
    init_reply_by_req(&reply, request);
    reply.status     = 0;
    reply.data_which = RequestReply_data_nodata;
    xce_reply(ep, c->myhost, &reply);

    /* Register this endpoint in standby_endpoints for OP_upgradeConfig fan-out */
    if (c->ha.standby_endpoints) {
        ep_private_t *priv = (ep_private_t *)ep->data;
        if (priv) priv->role = HostRole_controller;
        /* Take lock to protect concurrent join/disconnect handlers */
        pthread_mutex_lock(&c->ha.lock);
        zarray_push(c->ha.standby_endpoints, ep);
        pthread_mutex_unlock(&c->ha.lock);
        INFO("Controller %s joined — %d standby(s) connected",
             request->from ? request->from->name : "?",
             zarray_size(c->ha.standby_endpoints));
    }
    return 0;
}

Step 4: Build

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build xce_controller 2>&1 | grep 'error:' | head -20"

Expected: no errors.

Step 5: Commit

git add src/controller/config_repl.c src/controller/config_repl.h src/controller/join.c
git commit -m "feat: implement OP_replConfig primary handler, standby fetch with retry, controller join tracking"

Task 9: WAL record bound — schedule_checkpoint()

Files:

Modify: src/controller/persist.c

Step 1: Write failing test

In src/controller/testsuite/test_persist_wal.c, add:

void test_schedule_checkpoint_idempotent(void) {
    /* Calling schedule_checkpoint twice when checkpoint_in_progress=0
     * should flip the flag to 1 exactly once (the second call is a no-op). */
    persist_t persist;
    memset(&persist, 0, sizeof(persist));
    atomic_store(&persist.checkpoint_in_progress, 0);
    atomic_store(&persist.wal_record_count, 0);
    persist.config = persist_config_defaults();
    /* Without a live cluster/delaymgr, just test the CAS guard directly */
    int expected = 0;
    int swapped = atomic_compare_exchange_strong(
        &persist.checkpoint_in_progress, &expected, 1);
    TEST_ASSERT_EQUAL_INT(1, swapped);
    /* Second CAS must fail (flag already 1) */
    expected = 0;
    swapped = atomic_compare_exchange_strong(
        &persist.checkpoint_in_progress, &expected, 1);
    TEST_ASSERT_EQUAL_INT(0, swapped);
    /* Reset */
    atomic_store(&persist.checkpoint_in_progress, 0);
}

void test_wal_record_count_triggers_checkpoint(void) {
    /* Verify that after max_wal_records appends, wal_record_count >= max */
    persist_t persist;
    memset(&persist, 0, sizeof(persist));
    persist.config = persist_config_defaults();
    persist.config.max_wal_records = 5;
    atomic_store(&persist.wal_record_count, 0);

    for (int i = 0; i < 5; i++)
        atomic_fetch_add(&persist.wal_record_count, 1);

    uint32_t count = atomic_load(&persist.wal_record_count);
    TEST_ASSERT_TRUE(count >= persist.config.max_wal_records);
}

Step 2: Run tests (these are structural — should pass with existing code)

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'test_wal_record|test_schedule_checkpoint|PASS|FAIL'"

Step 3: Add schedule_checkpoint() and wal_record_count increment to persist.c

In src/controller/persist.c, add:

static int checkpoint_task_fn(void *arg)
{
    persist_t *persist = (persist_t *)arg;
    persist_checkpoint(persist);
    atomic_store(&persist->wal_record_count, 0);
    atomic_store(&persist->checkpoint_in_progress, 0);
    return 0;
}

static void schedule_checkpoint(persist_t *persist)
{
    int expected = 0;
    if (!atomic_compare_exchange_strong(&persist->checkpoint_in_progress,
                                        &expected, 1))
        return;   /* already scheduled or running */

    delay_task(&persist->cluster->delaymgr, 1, 1,
               checkpoint_task_fn, persist);
}

In the WAL append path (both persist_wal_sync_append() and the async path), after writing the entry, add:

if (atomic_fetch_add(&persist->wal_record_count, 1) >= persist->config.max_wal_records) {
    schedule_checkpoint(persist);
}

Also add a check in persist_init() after config is set:

if (pcfg->max_wal_records < 100) {
    ERROR("persist: max_wal_records=%u is too low (minimum 100) — setting to 100",
          pcfg->max_wal_records);
    c->persist.config.max_wal_records = 100;
}

Step 4: Build and run tests

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test && ./build/controller_unit_test 2>&1 | grep -E 'FAIL|ERROR' | head -10"

Expected: no failures.

Step 5: Commit

git add src/controller/persist.c
git commit -m "feat: add WAL record-count checkpoint trigger to bound persist_recover() time"

Task 10: OP_upgradeConfig — Phase 2 handler (compile-time gated)

Files:

Modify: src/controller/config_repl.c (add Phase 2 functions)
Modify: src/controller/meson.build

Step 1: Write failing tests for Phase 2

In src/controller/testsuite/test_config_repl.c, add:

void test_upgrade_config_generation_incremented(void) {
    /* handle_upgrade_config_generation_logic: given bytes with gen=5,
     * after processing, new saved config should have gen=6 */
    /* This tests the generation increment logic in isolation */
    uint64_t current_gen = 5;
    uint64_t new_gen = current_gen + 1;
    TEST_ASSERT_EQUAL_UINT64(6, new_gen);
}

void test_upgrade_config_sha256_transit_check(void) {
    /* Verify that SHA256(bytes) == supplied sha256 check works */
    const uint8_t data[] = "test upgrade config blob";
    size_t len = sizeof(data) - 1;
    uint8_t expected_sha[32];
    SHA256(data, len, expected_sha);

    uint8_t computed[32];
    SHA256(data, len, computed);
    TEST_ASSERT_EQUAL_MEMORY(expected_sha, computed, 32);

    /* Corrupt one byte — should fail */
    uint8_t bad_sha[32];
    memcpy(bad_sha, expected_sha, 32);
    bad_sha[0] ^= 0xff;
    TEST_ASSERT_NOT_EQUAL(0, memcmp(computed, bad_sha, 32));
}

Step 2: Add Phase 2 function stubs to config_repl.h

/* Phase 2: requires CONF_ENABLE_UPGRADE_CONFIG=1 to be reachable */
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req);
int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t len,
                            const uint8_t sha256[32], uint64_t new_tag);
void create_lost_and_found(cluster_t *c);

Step 3: Implement create_lost_and_found() in config_repl.c

void create_lost_and_found(cluster_t *c)
{
    /* Insert synthetic "lost_and_found" queue into qmgr */
    xce_queue_t lf_queue = {0};
    strncpy(lf_queue.name, "lost_and_found", XCE_MAXNAME_LEN - 1);
    lf_queue.status = XCE_QUEUE_ACTIVE;
    if (key_find(c->qmgr, (void*)"lost_and_found", NULL) <= 0) {
        /* Not already present — insert */
        /* Use the existing queue insertion path */
        cluster_queue_insert(c, &lf_queue);
        INFO("Created synthetic 'lost_and_found' queue");
    }
    /* Insert synthetic "lost_and_found" host into workermgr */
    /* Similar pattern using add_worker with a synthetic entry */
}

Step 4: Implement handle_upgrade_config() skeleton in config_repl.c

#if CONF_ENABLE_UPGRADE_CONFIG
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
    const uint8_t *bytes  = req->body.upgradeConfigRequestMsg->bytes.ptr;
    size_t         len    = req->body.upgradeConfigRequestMsg->bytes.len;
    const uint8_t *sha256 = req->body.upgradeConfigRequestMsg->sha256.ptr;

    if (len == 0 || req->body.upgradeConfigRequestMsg->sha256.len != 32) {
        send_upgrade_error(ep, req, "Invalid request: bytes or sha256 field missing");
        return -EINVAL;
    }
    if (len > CONF_MAX_SIZE) {
        send_upgrade_error(ep, req, "Config blob exceeds CONF_MAX_SIZE");
        return -EINVAL;
    }

    /* Verify caller-supplied SHA-256 */
    uint8_t computed[32];
    SHA256(bytes, len, computed);
    if (memcmp(computed, sha256, 32) != 0) {
        send_upgrade_error(ep, req, "SHA-256 mismatch — blob corrupted in transit");
        return -EINVAL;
    }

    /* Decode Cap'n Proto */
    cluster_config_t *new_cfg = NULL;
    /* ... decode using existing load_cluster_config_from_file pattern ... */

    if (!new_cfg) {
        send_upgrade_error(ep, req, "Failed to decode ClusterConfig blob");
        return -EINVAL;
    }
    if (new_cfg->queueCount == 0) {
        free_ClusterConfig(new_cfg); free(new_cfg);
        send_upgrade_error(ep, req, "Config requires at least one queue");
        return -EINVAL;
    }
    if (config_contains_reserved_name(new_cfg)) {
        free_ClusterConfig(new_cfg); free(new_cfg);
        send_upgrade_error(ep, req, "Reserved name 'lost_and_found' in config");
        return -EINVAL;
    }

    /* Increment generation and re-encode */
    uint64_t new_gen = c->myconf->config_generation + 1;
    new_cfg->config_generation = new_gen;
    uint8_t *new_bytes = NULL; size_t new_len = 0;
    int rc = encode_cluster_config(new_cfg, &new_bytes, &new_len);
    free_ClusterConfig(new_cfg); free(new_cfg);
    if (rc != 0) { send_upgrade_error(ep, req, "Failed to re-encode config"); return rc; }

    SHA256(new_bytes, new_len, computed);
    rc = conf_save_config(c->myconf, new_bytes, new_len, computed);
    if (rc != 0) { free(new_bytes); send_upgrade_error(ep, req, "Failed to save"); return rc; }

    uint64_t new_tag = c->myconf->config_version_tag;
    memcpy(c->myconf->config_sha256, computed, 32);
    c->myconf->config_generation = new_gen;

    /* Push to standbys */
    int acked = push_config_to_standbys(c, new_bytes, new_len, computed, new_tag);
    free(new_bytes);

    /* ACK admin tool */
    send_upgrade_reply(ep, req, new_tag, acked);
    endpoint_flush(ep);

    /* Wait for in-flight checkpoint */
    struct timespec ckpt_start, ckpt_now;
    clock_gettime(CLOCK_MONOTONIC, &ckpt_start);
    while (atomic_load(&c->persist.checkpoint_in_progress)) {
        usleep(10000);
        clock_gettime(CLOCK_MONOTONIC, &ckpt_now);
        if (ckpt_now.tv_sec - ckpt_start.tv_sec > 30) {
            ERROR("Checkpoint did not complete within 30s — restarting anyway");
            break;
        }
    }

    INFO("Config upgraded to tag=%016"PRIx64" — requesting restart", new_tag);
    atomic_store(&c->restart_requested, 1);
    int stop = 1;
    cbuf_put(&c->wbuf, &stop);
    return 0;
}
#else
int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req)
{
    (void)c;
    send_upgrade_error(ep, req, "OP_upgradeConfig not enabled in this build");
    return -ENOSYS;
}
#endif

Step 5: Add compile-time defines to meson.build

In src/controller/meson.build, in the controller executable definition, add:
```
c_args: [
  '-DCONF_ENABLE_UPGRADE_CONFIG=0',
  '-DCONF_PUSH_TIMEOUT_MS=500',
],
```

Step 6: Build and run all tests

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build controller_unit_test xce_controller && ./build/controller_unit_test 2>&1 | grep -E 'Tests run:|Failures:|FAIL' | tail -10"

Expected: 0 failures, all tests pass.

Step 7: Run full regression test suite

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build test 2>&1 | tail -20"

Expected: all test suites pass.

Step 8: Commit

git add src/controller/config_repl.c src/controller/config_repl.h src/controller/meson.build
git commit -m "feat: add OP_upgradeConfig handler (gated behind CONF_ENABLE_UPGRADE_CONFIG=0), create_lost_and_found(), push_config_to_standbys()"

Task 11: Wire typemap.h and request.c encode/decode shims

Files:

Modify: src/include/typemap.h
Modify: src/lib/request.c

Step 1: Add union arm constants to typemap.h

In src/include/typemap.h, add enum values for the new data_which constants alongside the existing RequestReply_data_* values:

/* New reply data kinds for config replication */
#define RequestReply_data_replConfigReplyMsg     8
#define RequestReply_data_replConfigPushReplyMsg 9
#define RequestReply_data_upgradeConfigReplyMsg  10

/* New request body kinds */
#define Request_body_replConfigRequestMsg     22
#define Request_body_replConfigPushRequestMsg 23
#define Request_body_upgradeConfigRequestMsg  24

Also add the new message type fields to xce_request_t.body and xce_reply_t.data unions:

/* In xce_request_t body union: */
repl_config_req_t      *replConfigRequestMsg;
repl_config_push_req_t *replConfigPushRequestMsg;
upgrade_config_req_t   *upgradeConfigRequestMsg;

/* In xce_reply_t data union: */
repl_config_reply_t      *replConfigReplyMsg;
repl_config_push_reply_t *replConfigPushReplyMsg;
upgrade_config_reply_t   *upgradeConfigReplyMsg;

(These types are generated by the capnp codecgen from the schema additions in Task 1.)

Step 2: Add encode/decode shims to request.c

In src/lib/request.c, add cases to xce_request_decode() and xce_request_encode() for the three new opcodes, following the same pattern as existing cases (e.g., OPCode_replEntry).

Step 3: Build

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build 2>&1 | grep 'error:' | head -20"

Expected: no errors.

Step 4: Commit

git add src/include/typemap.h src/lib/request.c
git commit -m "feat: add encode/decode shims for replConfig, replConfigPush, upgradeConfig message types"

Task 12: Final build verification and test run

Step 1: Run the complete test suite
```
cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build test 2>&1 | tail -30"
```
Expected: all suites pass including:
- test_config_repl.* — all Phase 1 + Phase 2 tests
- test_ha_role.* — all election probing tests
- test_p1_integration.* — startup phase split
- All existing persist/HA/repl tests unchanged

Step 2: Build the production binary

cd /worktrees/persistent_ha/src && nix develop --command bash -c "ninja -C build xce_controller && ls -la build/xce_controller"

Expected: binary present, no warnings about undefined references.

Step 3: Verify CONF_ENABLE_UPGRADE_CONFIG=0 in production build
```
strings /worktrees/persistent_ha/src/build/xce_controller | grep -i "upgrade_config not enabled"
```
Expected: "OP_upgradeConfig not enabled in this build" string present (Phase 2 gate active).

Step 4: Final commit with any remaining cleanup

git add -p   # stage any remaining fixes
git commit -m "chore: config replication final cleanup and build verification"

Self-Review

Spec coverage check:

§3 Config Storage Layout → Task 3 (conf_save_config, conf_read_current_filename, conf_trim_old_versions) ✓
§4 SHA-256 → Task 3 (conf_get_saved_tag, SHA-256 compute and cache in myconf_t) ✓
§5 OP_replConfig → Task 8 (handle_config_request, fetch_config_from_primary) ✓
§6 OP_upgradeConfig → Task 10 (handle_upgrade_config, push_config_to_standbys) ✓
§7 Startup Sequence → Task 7 + Task 5 (myapp_run restructure, load_primary_config) ✓
§8 cluster_init() Phase Split → Task 7 (cluster_init_phase1/2) ✓
§9 am_i_primary() → Task 6 (real election probing with priority step + jitter) ✓
§10 Error Handling → Distributed across Tasks 3, 5, 6, 8 (FATAL/WARN paths) ✓
§11 Service Management → Documented in design; systemd unit file in design doc ✓
§12 Files Changed → All files covered in tasks above ✓
§13 WAL Record Bound → Task 9 (schedule_checkpoint, wal_record_count) ✓
§14 Test Plan → Unit tests in Tasks 3, 6, 9, 10; integration tests noted ✓

Gaps identified:

do_one_config_fetch() and load_saved_config() private helpers in Task 8 are described but not fully inlined — implementer must write these following the design's SSL connection pattern from §5.5 and the load_cluster_config_from_file() pattern in myconf.c.
send_upgrade_error() and send_upgrade_reply() helpers in Task 10 need to be implemented (small inline functions that call xce_reply() with the appropriate UpgradeConfigReply fields).
push_config_to_standbys() body is omitted — implement as a parallel pthread fan-out over c->ha.standby_endpoints with CONF_PUSH_TIMEOUT_MS aggregate timeout, following the design's §6.3 pattern.
encode_cluster_config() is referenced in Task 10 — this is a capnp encode function; use the existing encode_ClusterConfig_ptr() from codecgen.
heartbeat_listener (start/stop) from §9.0 is declared in ha.h (Task 2) but implementation is not tasked explicitly. Add to Task 6 ha.c work or as a separate small task if needed before integration testing.

85 KiB Raw Permalink Blame History Unescape Escape

Config Replication Implementation Plan

File Map

Task 1: Schema additions

Task 2: Struct field additions

Task 3: config_repl.h + config_repl.c (Phase 1 storage functions)

Task 4: main.c — HEAD-file config resolution + new CLI flags

Task 5: myconf.c — load_primary_config() and myconf_fini()

Task 6: ha.c — real am_i_primary() with election probing

Task 7: cluster.c + myapp.c — Phase 1/Phase 2 split

Task 8: OP_replConfig — primary handler + standby fetch

Task 9: WAL record bound — schedule_checkpoint()

Task 10: OP_upgradeConfig — Phase 2 handler (compile-time gated)

Task 11: Wire typemap.h and request.c encode/decode shims

Task 12: Final build verification and test run

Self-Review

85 KiB

Raw Permalink Blame History