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

53 KiB

Raw Blame History

P1: HA Foundation — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development or superpowers:executing-plans to implement this plan task-by-task.

Goal: Standby controller can take over when primary fails.

Architecture: P1 builds LSF-style priority election on top of the existing iomgr/SSL/endpoint/vtask pipeline — no new TLS infrastructure is introduced because XCE runs on a trusted LAN. A dedicated watchdog pthread (never delaymgr) detects primary heartbeat timeout and triggers self-promotion. The hook system is extended with atomic disable-in-place semantics so persistence hooks can be cleanly registered on promotion and disabled on demotion without list growth across promote/demote cycles.

Tech Stack: C11, POSIX threads, _Atomic / stdatomic.h, nanosleep / CLOCK_MONOTONIC, iomgr/vtask/endpoint SSL pipeline, Cap'n Proto (capnpc-c), nix develop + meson + ninja

Design doc: src/docs/persistence-and-ha-design.md — Parts 5, 9, Appendices A.3–A.4

Build commands (inside nix develop):

meson setup build --wipe   # first time
ninja -C build             # incremental
ninja -C build test        # run all tests
# OR from src/frontend/:
make test                  # Go unit tests (unchanged by P1)

Dependencies: P0 complete — all tests green, persist_t persist wired into cluster_t, controller.capnp has HA opcodes (replEntry, replAck, redirect, heartbeat, steppingDown, replSnapshot, replCatchUpDone) already added in P0-9. Verify before starting:

grep -c 'heartbeat\|steppingDown\|replEntry' src/schema/controller.capnp  # must be > 0
ninja -C build test                                                          # must be green

P1-1: Verify iomgr/SSL pipeline reuse — no new TLS work needed

Files: src/controller/netio.c (read-only audit), src/controller/ha.c (will be created in P1-2 — just audit for now)

Goal: Confirm that create_ssl_context() in netio.c and the existing iomgr accept/connect path work unchanged for HA traffic. No modifications needed — this is an audit step that produces a commit containing only a comment block in the upcoming ha.h.

RED — audit test (compile-only, no new test file yet)

# Verify: iomgr/endpoint already handles SSL for controller-to-controller conns
grep -n 'create_ssl_context\|iomgr_set_ssl_ctx\|endpoint_init' src/controller/cluster.c
# Expected: ssl_ctx created in cluster_init(), set on iomgr — HA listeners will reuse it
grep -n 'SSL_CTX\|ssl_ctx' src/controller/netio.c | head -10

GREEN — document the decision in ha.h (stubbed header, creates the file)

Create src/controller/ha.h:

#ifndef HA_H
#define HA_H

/*
 * P1-1 audit result: iomgr/SSL pipeline reuse confirmed.
 *
 * The existing create_ssl_context() + iomgr_set_ssl_ctx() path (cluster.c:558-583)
 * handles all controller-to-controller SSL. HA listeners (election probe, heartbeat,
 * replication accept) will use the SAME ssl_ctx already installed on c->iomgr.
 *
 * No new TLS infrastructure is needed. XCE runs on a trusted LAN — peer identity is
 * validated by host:port match against the --peers candidate list, not mTLS.
 * See design doc Part 5.9 and memory note: Persistence & HA Trusted LAN Decision.
 */

#include "xce/cluster_def.h"  /* cluster_t forward dependency */
#include "persist.h"

/* Full struct and API declarations added in P1-2. */

#endif /* HA_H */

Commit:

git add src/controller/ha.h
git commit -m "feat: add ha.h stub documenting iomgr/SSL pipeline reuse decision (P1-1)"

P1-2a: Atomic hook disable — `xce_disable_hook()` + `_Atomic hook_fn_t`

Files to modify:

src/include/hook.h — change hook_item_t.fn to _Atomic hook_fn_t; add xce_disable_hook() declaration
src/lib/hook.c — update xce_add_hook(), xce_run_hooks(), add xce_disable_hook(); add NULL-slot reuse scan

Rationale: ha_demote_to_standby() (P1-3a) must atomically disable persistence hooks without removing them from the list, preventing unbounded list growth across promote→demote→re-promote cycles. xce_run_hooks() must skip NULL-fn slots. xce_add_hook() must reuse NULL slots of the same type.

RED — failing test

Add to src/lib/testsuite/test_hooks.c:

/* Test: xce_disable_hook nullifies fn atomically; xce_run_hooks skips it */
void test_disable_hook(void)
{
    hooksys_t hs;
    int call_count = 0;

    static int counter_fn(int type, int sit, int det, void* data, va_list ap) {
        (*(int*)data)++;
        return 0;
    }

    xce_hook_init(&hs, 4);
    TEST_ASSERT_EQUAL(0, xce_add_hook(&hs, 0, counter_fn, &call_count));

    /* Before disable: hook fires */
    xce_run_hooks(&hs, 0, HOOK_AFTER_EVENT, 0);
    TEST_ASSERT_EQUAL(1, call_count);

    /* Disable: xce_run_hooks must skip the now-NULL slot */
    TEST_ASSERT_EQUAL(0, xce_disable_hook(&hs, 0, counter_fn));
    xce_run_hooks(&hs, 0, HOOK_AFTER_EVENT, 0);
    TEST_ASSERT_EQUAL(1, call_count);  /* count unchanged — hook not called */

    xce_hook_fini(&hs);
}

/* Test: slot reuse — re-adding the same (type, fn) reuses the NULL slot, list stays bounded */
void test_add_hook_reuses_null_slot(void)
{
    hooksys_t hs;
    int call_count = 0;
    int initial_count;

    static int noop_fn(int type, int sit, int det, void* data, va_list ap) {
        (*(int*)data)++;
        return 0;
    }

    xce_hook_init(&hs, 4);
    xce_add_hook(&hs, 0, noop_fn, &call_count);
    initial_count = hs.hooks[0].count;  /* e.g. 1 */

    /* Disable then re-add — list must NOT grow */
    xce_disable_hook(&hs, 0, noop_fn);
    xce_add_hook(&hs, 0, noop_fn, &call_count);
    TEST_ASSERT_EQUAL(initial_count, hs.hooks[0].count);  /* same slot reused */

    xce_hook_fini(&hs);
}

Run ninja -C build test — must fail (function not declared yet).

GREEN — implementation

src/include/hook.h — change struct and add declaration:

#if !defined(__HOOK_H__)
#define __HOOK_H__ 1

#include <stdarg.h>
#include <stddef.h>
#include <stdatomic.h>   /* C11 — _Atomic */

#include "xlist.h"

#define HOOK_BEFORE_EVENT (0)
#define HOOK_AFTER_EVENT  (1)

typedef int (*hook_fn_t)(int, int, int, void*, va_list);

typedef struct {
    _Atomic hook_fn_t fn;   /* C11 atomic — NULL after xce_disable_hook() */
    void*             data;
} hook_item_t;

typedef struct {
    int      maxtypes;
    xlist_t* hooks;
} hooksys_t;

extern int  xce_hook_init(hooksys_t*, int);
extern int  xce_add_hook(hooksys_t*, int, hook_fn_t, void*);
extern int  xce_disable_hook(hooksys_t*, int, hook_fn_t);  /* NEW */
extern int  xce_run_hooks(hooksys_t*, int, int, int, ...);
extern void xce_hook_fini(hooksys_t*);

#endif

src/lib/hook.c — key changes to each function:

xce_add_hook() — scan for NULL slot before appending:

int xce_add_hook(hooksys_t* s, int typ, hook_fn_t fn, void* data)
{
    hook_item_t* it;
    int          i;

    if (s == NULL || fn == NULL) return -EINVAL;
    if ((typ < 0) || (typ >= s->maxtypes)) return -EINVAL;

    /* Reuse a disabled (NULL-fn) slot of the same type — prevents unbounded growth
     * on promote->demote->re-promote cycles (design doc Part 5.7). */
    xlist_t* l = &(s->hooks[typ]);
    for (i = 0; i < l->count; i++) {
        hook_item_t* slot = l->data[i];
        if (slot == NULL) continue;
        hook_fn_t cur = atomic_load_explicit(&slot->fn, memory_order_acquire);
        if (cur == NULL) {
            /* Reuse: set data first, then fn with release so readers see consistent pair */
            slot->data = data;
            atomic_store_explicit(&slot->fn, fn, memory_order_release);
            return 0;
        }
    }

    it = (hook_item_t*)calloc(1, sizeof(*it));
    if (it == NULL) return -ENOMEM;

    it->data = data;
    atomic_store_explicit(&it->fn, fn, memory_order_release);
    xlist_store(&(s->hooks[typ]), it);
    return 0;
}

xce_disable_hook() — new function:

int xce_disable_hook(hooksys_t* s, int typ, hook_fn_t fn)
{
    int i;

    if (s == NULL || fn == NULL) return -EINVAL;
    if ((typ < 0) || (typ >= s->maxtypes)) return -EINVAL;

    xlist_t* l = &(s->hooks[typ]);
    for (i = 0; i < l->count; i++) {
        hook_item_t* it = l->data[i];
        if (it == NULL) continue;
        hook_fn_t cur = atomic_load_explicit(&it->fn, memory_order_acquire);
        if (cur == fn) {
            atomic_store_explicit(&it->fn, (hook_fn_t)NULL, memory_order_release);
            return 0;
        }
    }
    return -ENOENT;
}

xce_run_hooks() — NULL-fn guard:

for (i = 0; i < l->count; i++) {
    hook_item_t* it = l->data[i];
    va_list      ap;

    if (it == NULL) continue;

    hook_fn_t local_fn = atomic_load_explicit(&it->fn, memory_order_acquire);
    if (!local_fn) continue;   /* disabled slot — skip */

    va_start(ap, detail);
    rc = local_fn(type, situation, detail, it->data, ap);
    va_end(ap);

    if ((situation == HOOK_BEFORE_EVENT) && (rc < 0)) break;
}

Run ninja -C build test — must pass.

Commit:

git add src/include/hook.h src/lib/hook.c src/lib/testsuite/test_hooks.c
git commit -m "feat: add xce_disable_hook() + atomic hook_fn_t + NULL-slot reuse guard (P1-2a)"

P1-2: `ha.c` + `ha.h` — role management, `am_i_primary()`, promotion/demotion stubs

Files:

src/controller/ha.h — full struct and API declarations (replaces stub from P1-1)
src/controller/ha.c — new file: ha_init(), ha_fini(), am_i_primary(), ha_promote_to_primary() (step-counted stub), ha_demote_to_standby() stub

Requires: P1-2a merged and green.

RED — failing tests

Add src/controller/testsuite/test_ha_role.c:

#include <unity/unity.h>
#include <stdatomic.h>
#include "ha.h"
#include "xce/cluster_def.h"

/* Minimal cluster_t stand-in for unit tests — avoids full cluster_init() */
static cluster_t g_cluster;

void setUp(void)   { memset(&g_cluster, 0, sizeof(g_cluster)); }
void tearDown(void) { }

/* ha_init() must set role = CTRL_STANDBY (default before am_i_primary()) */
void test_ha_init_default_standby(void)
{
    ha_config_t cfg = {
        .heartbeat_interval_ms    = 3000,
        .heartbeat_timeout_ms     = 10000,
        .election_timeout_ms      = 2000,
        .probe_connect_timeout_ms = 500,
        .max_promotion_retries    = 5,
    };
    TEST_ASSERT_EQUAL(0, ha_init(&g_cluster.ha, &cfg, 0 /* my_priority */));
    TEST_ASSERT_EQUAL(CTRL_STANDBY,
        (ctrl_role_t)atomic_load_explicit(&g_cluster.ha.role, memory_order_acquire));
    ha_fini(&g_cluster.ha);
}

/* ha_init() must reject heartbeat_interval_ms < 2000 */
void test_ha_init_rejects_fast_heartbeat(void)
{
    ha_config_t cfg = {
        .heartbeat_interval_ms = 1999,  /* below floor */
        .heartbeat_timeout_ms  = 10000,
        .election_timeout_ms   = 2000,
        .probe_connect_timeout_ms = 500,
        .max_promotion_retries = 5,
    };
    TEST_ASSERT_NOT_EQUAL(0, ha_init(&g_cluster.ha, &cfg, 0));
}

/* ha_init() must reject election_timeout_ms < 500 */
void test_ha_init_rejects_short_election_timeout(void)
{
    ha_config_t cfg = {
        .heartbeat_interval_ms = 3000,
        .heartbeat_timeout_ms  = 10000,
        .election_timeout_ms   = 499,   /* below floor */
        .probe_connect_timeout_ms = 500,
        .max_promotion_retries = 5,
    };
    TEST_ASSERT_NOT_EQUAL(0, ha_init(&g_cluster.ha, &cfg, 0));
}

Run ninja -C build test — must fail (ha.h and ha.c do not define ha_init yet).

GREEN — implementation

src/controller/ha.h (replace stub):

#ifndef HA_H
#define HA_H

/*
 * XCE Controller — HA role management.
 * Implements LSF-style priority election (design doc Parts 5, 9; Appendices A.3–A.4).
 *
 * iomgr/SSL pipeline reuse: all HA traffic (heartbeats, election probes, replication)
 * uses the existing create_ssl_context() + iomgr/endpoint pipeline. No new TLS
 * infrastructure is needed — XCE runs on a trusted LAN (P1-1 audit, design doc Part 5.9).
 */

#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <time.h>

#include "xce/cluster_def.h"  /* cluster_t, persist_t, endpoint_t */
#include "persist.h"

/* ── Role enum ───────────────────────────────────────────────────────── */
typedef enum { CTRL_PRIMARY = 0, CTRL_STANDBY = 1 } ctrl_role_t;

/* ── HA configuration (populated from ha_config_t in cluster config) ── */
typedef struct {
    uint32_t heartbeat_interval_ms;    /* Primary heartbeat send interval (min 2000) */
    uint32_t heartbeat_timeout_ms;     /* Standby timeout before promoting (min heartbeat_interval_ms) */
    uint32_t election_timeout_ms;      /* am_i_primary() shared deadline (min 500) */
    uint32_t probe_connect_timeout_ms; /* Per-probe poll() timeout in am_i_primary() */
    uint32_t max_promotion_retries;    /* Max consecutive failed promotion attempts */
} ha_config_t;

/* ── Replicator (P1 stub — P2 adds full fan-out send) ─────────────────
 * Forward declaration only — repl.h (P2-1) defines the full struct.
 * P1 includes the replicator field but uses it only for zero-init.     */
#ifndef REPL_H
typedef struct { int _placeholder; } replicator_t;
#endif

/* ── HA state (embedded in cluster_t as ha_state_t ha) ───────────────── */
typedef struct {
    _Atomic uint64_t term;              /* C11 atomic election fencing token */
    _Atomic int32_t  role;             /* _Atomic int32_t, cast to ctrl_role_t at access */
    int              my_priority;      /* Index in candidates[] (lower = higher priority) */
    replicator_t     repl;             /* Replicator — initialized in ha_init(), used in P2 */
    _Atomic uint64_t applied_lsn;      /* Standby: last applied WAL LSN (written by iomgr thread) */
    endpoint_t*      primary_ep;       /* Borrowed ref to primary SSL endpoint (standby only) */
    char             primary_host[XCE_MAXNAME_LEN];
    uint32_t         primary_port;
    struct timespec  last_heartbeat;   /* CLOCK_MONOTONIC timestamp of last primary heartbeat */
    _Atomic int      promotion_failed;
    _Atomic int      promotion_retry_count;
    _Atomic int      in_catch_up;      /* Standby: set while replaying historical entries */
    int              hooks_registered; /* Idempotency guard: non-zero after xce_add_hook() */
    atomic_int       failover_requested;   /* SIGUSR1 handler sets this; watchdog executes */
    uint32_t         last_received_crc;    /* Chained CRC of last replicated WAL entry (standby) */
    _Atomic int      ha_isolated;          /* Primary: set when no standby reachable for heartbeat */
    struct timespec  primary_isolated_since; /* When isolation began (CLOCK_MONOTONIC) */
    ha_config_t      config;
    pthread_mutex_t  lock;             /* Protects primary_ep + last_heartbeat; initialized once */
} ha_state_t;

/* ── Public API ───────────────────────────────────────────────────────── */

/* Initialize ha_state_t; validates config floors; must be called before am_i_primary().
 * Returns 0 on success, -EINVAL on bad config. */
int  ha_init(ha_state_t* ha, const ha_config_t* cfg, int my_priority);

/* Clean up: join watchdog thread, destroy mutex. */
void ha_fini(ha_state_t* ha);

/* Election: probe all higher-priority candidates; return 1 if I am primary, 0 if not. */
int  am_i_primary(cluster_t* c);

/* Promote self to primary (from standby). Called by watchdog on heartbeat timeout. */
int  ha_promote_to_primary(cluster_t* c);

/* Demote self from primary to standby (manual failover / term supersede). */
int  ha_demote_to_standby(cluster_t* c);

#endif /* HA_H */

src/controller/ha.c (initial skeleton — full promotion/demotion in P1-3a):

#define _GNU_SOURCE
#include "ha.h"

#include <errno.h>
#include <stdatomic.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>

#include "utils.h"   /* ERROR/WARN/INFO macros */

int ha_init(ha_state_t* ha, const ha_config_t* cfg, int my_priority)
{
    if (!ha || !cfg) return -EINVAL;

    /* Validate config floors (design doc Part 5.4, A5-021) */
    if (cfg->heartbeat_interval_ms < 2000) {
        ERROR("ha_init: heartbeat_interval_ms %u < 2000ms minimum",
              cfg->heartbeat_interval_ms);
        return -EINVAL;
    }
    if (cfg->heartbeat_timeout_ms < cfg->heartbeat_interval_ms) {
        ERROR("ha_init: heartbeat_timeout_ms %u < heartbeat_interval_ms %u",
              cfg->heartbeat_timeout_ms, cfg->heartbeat_interval_ms);
        return -EINVAL;
    }
    if (cfg->election_timeout_ms < 500) {
        ERROR("ha_init: election_timeout_ms %u < 500ms minimum",
              cfg->election_timeout_ms);
        return -EINVAL;
    }
    if (cfg->probe_connect_timeout_ms < 500 ||
        cfg->probe_connect_timeout_ms > cfg->election_timeout_ms) {
        ERROR("ha_init: probe_connect_timeout_ms %u out of range [500, %u]",
              cfg->probe_connect_timeout_ms, cfg->election_timeout_ms);
        return -EINVAL;
    }

    memset(ha, 0, sizeof(*ha));
    ha->config      = *cfg;
    ha->my_priority = my_priority;

    /* Default role is STANDBY — am_i_primary() will promote to PRIMARY if elected */
    atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);
    atomic_store_explicit(&ha->term, 0ULL, memory_order_release);
    atomic_store_explicit(&ha->promotion_failed,      0, memory_order_relaxed);
    atomic_store_explicit(&ha->promotion_retry_count, 0, memory_order_relaxed);
    atomic_store_explicit(&ha->in_catch_up,           0, memory_order_relaxed);
    atomic_store_explicit(&ha->failover_requested,    0, memory_order_relaxed);
    atomic_store_explicit(&ha->ha_isolated,           0, memory_order_relaxed);
    atomic_store_explicit(&ha->applied_lsn,           0ULL, memory_order_release);

    /* Initialize the mutex ONCE — never re-initialized across role changes */
    pthread_mutex_init(&ha->lock, NULL);

    return 0;
}

void ha_fini(ha_state_t* ha)
{
    if (!ha) return;
    /* Watchdog thread join added in P1-3 */
    pthread_mutex_destroy(&ha->lock);
}

int am_i_primary(cluster_t* c)
{
    /* Full implementation in P1-3 (election probe pthreads).
     * Stub: if my_priority == 0 (highest), assume primary for single-node tests. */
    ha_state_t* ha = &c->ha;
    return (ha->my_priority == 0) ? 1 : 0;
}

int ha_promote_to_primary(cluster_t* c)
{
    /* Full promotion sequence (design doc Part 5.6) implemented in P1-3a.
     * Stub: set role to PRIMARY. */
    ha_state_t* ha = &c->ha;
    atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release);
    return 0;
}

int ha_demote_to_standby(cluster_t* c)
{
    /* Full demotion sequence implemented in P1-3a.
     * Stub: disable hooks, set role to STANDBY. */
    ha_state_t* ha = &c->ha;
    atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);
    return 0;
}

Run ninja -C build test — must pass.

Commit:

git add src/controller/ha.h src/controller/ha.c src/controller/testsuite/test_ha_role.c
git commit -m "feat: add ha_state_t, ha_config_t, ha_init/fini stubs + role enum (P1-2)"

P1-3: Heartbeat watchdog pthread + `ha_check_heartbeat()`

Files:

src/controller/ha.c — add ha_watchdog_thread(), ha_check_heartbeat(); add _Atomic int ha_watchdog_stop and pthread_t watchdog_thread to ha_state_t
src/controller/ha.h — add new fields to ha_state_t struct

Key constraints from design doc Part 5.5:

Dedicated pthread with nanosleep(CLOCK_MONOTONIC) — NOT delaymgr (2s granularity too coarse)
CLOCK_MONOTONIC — immune to NTP jumps
ha_fini() sets stop flag + pthread_join() before returning — no use-after-free of ha_state_t
Never pthread_cancel() — not OpenSSL-safe

RED — failing test

Add to src/controller/testsuite/test_ha_role.c:

#include <time.h>

/* ha_check_heartbeat(): if no primary_ep, do nothing (no promotion triggered) */
void test_ha_check_heartbeat_no_primary(void)
{
    ha_config_t cfg = {
        .heartbeat_interval_ms    = 3000,
        .heartbeat_timeout_ms     = 10000,
        .election_timeout_ms      = 2000,
        .probe_connect_timeout_ms = 500,
        .max_promotion_retries    = 5,
    };
    ha_init(&g_cluster.ha, &cfg, 1 /* standby priority */);
    /* primary_ep is NULL — watchdog must return without promoting */
    ha_check_heartbeat_for_test(&g_cluster);
    /* Role must still be STANDBY */
    TEST_ASSERT_EQUAL(CTRL_STANDBY,
        (ctrl_role_t)atomic_load_explicit(&g_cluster.ha.role, memory_order_acquire));
    ha_fini(&g_cluster.ha);
}

/* ha_check_heartbeat(): if role is PRIMARY, function returns immediately */
void test_ha_check_heartbeat_primary_noop(void)
{
    ha_config_t cfg = {
        .heartbeat_interval_ms    = 3000,
        .heartbeat_timeout_ms     = 10000,
        .election_timeout_ms      = 2000,
        .probe_connect_timeout_ms = 500,
        .max_promotion_retries    = 5,
    };
    ha_init(&g_cluster.ha, &cfg, 0);
    atomic_store_explicit(&g_cluster.ha.role, (int32_t)CTRL_PRIMARY, memory_order_release);
    ha_check_heartbeat_for_test(&g_cluster);  /* must not crash or change role */
    TEST_ASSERT_EQUAL(CTRL_PRIMARY,
        (ctrl_role_t)atomic_load_explicit(&g_cluster.ha.role, memory_order_acquire));
    ha_fini(&g_cluster.ha);
}

Expose test-only entry point in ha.h:

#ifdef UNIT_TEST
void ha_check_heartbeat_for_test(cluster_t* c);
#endif

Run ninja -C build test — must fail (symbol not defined).

GREEN — implementation

Add to ha_state_t in ha.h:

    _Atomic int      ha_watchdog_stop;  /* Set by ha_fini() to signal watchdog to exit */
    pthread_t        watchdog_thread;   /* Joined in ha_fini() before returning */

Add to ha.c:

/* Exposed for unit tests only */
void ha_check_heartbeat_for_test(cluster_t* c) { ha_check_heartbeat_(c); }

static void ha_check_heartbeat_(cluster_t* c)
{
    ha_state_t* ha = &c->ha;

    /* Primary does not check its own heartbeat */
    if ((ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire) != CTRL_STANDBY)
        return;

    /* Snapshot primary_ep + last_heartbeat under ha->lock — iomgr clears primary_ep
     * (NULL-before-free) holding this same lock, so we see valid-or-NULL, never dangling. */
    pthread_mutex_lock(&ha->lock);
    int             no_primary = (ha->primary_ep == NULL);
    struct timespec last_hb    = ha->last_heartbeat;
    pthread_mutex_unlock(&ha->lock);

    if (no_primary)
        return;

    struct timespec now;
    clock_gettime(CLOCK_MONOTONIC, &now);
    int64_t elapsed_ms = (now.tv_sec  - last_hb.tv_sec)  * 1000LL
                       + (now.tv_nsec - last_hb.tv_nsec) / 1000000LL;

    if (elapsed_ms > (int64_t)ha->config.heartbeat_timeout_ms) {
        if (atomic_load_explicit(&ha->promotion_failed, memory_order_relaxed) &&
            atomic_load_explicit(&ha->promotion_retry_count, memory_order_relaxed)
                >= (int)ha->config.max_promotion_retries) {
            ERROR("ha: cannot promote after %d attempts — manual intervention required",
                  atomic_load_explicit(&ha->promotion_retry_count, memory_order_relaxed));
            return;
        }

        /* Also handle manual failover flag set by SIGUSR1 handler (P1-3a) */
        ha_promote_to_primary(c);
    }
}

static void* ha_watchdog_thread_(void* arg)
{
    cluster_t* c = (cluster_t*)arg;
    ha_state_t* ha = &c->ha;

    while (!atomic_load_explicit(&ha->ha_watchdog_stop, memory_order_acquire)) {
        struct timespec ts = {
            .tv_sec  = ha->config.heartbeat_interval_ms / 1000,
            .tv_nsec = (ha->config.heartbeat_interval_ms % 1000) * 1000000L,
        };
        nanosleep(&ts, NULL);

        if (atomic_load_explicit(&ha->ha_watchdog_stop, memory_order_acquire))
            break;

        /* Check manual failover request (set by SIGUSR1 handler — P1-3a) */
        if (atomic_load_explicit(&ha->failover_requested, memory_order_acquire)) {
            atomic_store_explicit(&ha->failover_requested, 0, memory_order_release);
            ha_demote_to_standby(c);
        }

        ha_check_heartbeat_(c);
    }
    return NULL;
}

/* Call from cluster_start() after ha_init() — starts the watchdog pthread */
int ha_start_watchdog(cluster_t* c)
{
    ha_state_t* ha = &c->ha;
    atomic_store_explicit(&ha->ha_watchdog_stop, 0, memory_order_release);
    return pthread_create(&ha->watchdog_thread, NULL, ha_watchdog_thread_, c);
}

Update ha_fini():

void ha_fini(ha_state_t* ha)
{
    if (!ha) return;
    atomic_store_explicit(&ha->ha_watchdog_stop, 1, memory_order_release);
    pthread_join(ha->watchdog_thread, NULL);  /* safe: watchdog checks stop flag */
    pthread_mutex_destroy(&ha->lock);
}

Run ninja -C build test — must pass.

Commit:

git add src/controller/ha.h src/controller/ha.c src/controller/testsuite/test_ha_role.c
git commit -m "feat: add ha_check_heartbeat + watchdog pthread with nanosleep/CLOCK_MONOTONIC (P1-3)"

P1-3a: Promotion sequence + SIGUSR1 handler

Files:

src/controller/ha.c — implement ha_promote_to_primary() (full 9 steps), ha_demote_to_standby() (full hook disable + role switch), SIGUSR1 signal handler
src/controller/ha.h — add ha_install_signal_handler() declaration

Requires: P1-2a (xce_disable_hook) and P1-3 (watchdog) merged.

Key design doc refs: Part 5.6 (promotion sequence), Part 5.7 (hook disable + memory ordering), Part 5.8 (SIGUSR1 manual failover)

RED — failing tests

Add to src/controller/testsuite/test_ha_role.c:

/* ha_promote_to_primary(): role becomes PRIMARY */
void test_promote_sets_primary_role(void)
{
    ha_config_t cfg = {
        .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
        .max_promotion_retries = 5,
    };
    ha_init(&g_cluster.ha, &cfg, 1);
    /* Starting as standby */
    TEST_ASSERT_EQUAL(CTRL_STANDBY,
        (ctrl_role_t)atomic_load_explicit(&g_cluster.ha.role, memory_order_acquire));

    ha_promote_to_primary(&g_cluster);

    TEST_ASSERT_EQUAL(CTRL_PRIMARY,
        (ctrl_role_t)atomic_load_explicit(&g_cluster.ha.role, memory_order_acquire));
    ha_fini(&g_cluster.ha);
}

/* ha_promote_to_primary(): term increments */
void test_promote_increments_term(void)
{
    ha_config_t cfg = {
        .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
        .max_promotion_retries = 5,
    };
    ha_init(&g_cluster.ha, &cfg, 1);
    uint64_t term_before = atomic_load_explicit(&g_cluster.ha.term, memory_order_acquire);
    ha_promote_to_primary(&g_cluster);
    uint64_t term_after  = atomic_load_explicit(&g_cluster.ha.term, memory_order_acquire);
    TEST_ASSERT_GREATER_THAN(term_before, term_after);
    ha_fini(&g_cluster.ha);
}

/* ha_demote_to_standby(): role reverts to STANDBY, hooks_registered cleared */
void test_demote_sets_standby_role(void)
{
    ha_config_t cfg = {
        .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
        .max_promotion_retries = 5,
    };
    ha_init(&g_cluster.ha, &cfg, 0);
    ha_promote_to_primary(&g_cluster);
    TEST_ASSERT_EQUAL(CTRL_PRIMARY,
        (ctrl_role_t)atomic_load_explicit(&g_cluster.ha.role, memory_order_acquire));

    ha_demote_to_standby(&g_cluster);

    TEST_ASSERT_EQUAL(CTRL_STANDBY,
        (ctrl_role_t)atomic_load_explicit(&g_cluster.ha.role, memory_order_acquire));
    TEST_ASSERT_EQUAL(0, g_cluster.ha.hooks_registered);
    ha_fini(&g_cluster.ha);
}

/* hooks_registered guard: double-promote must not add duplicate hooks */
void test_promote_idempotent_hook_registration(void)
{
    /* Minimal hooksys for test — real cluster_init is skipped */
    xce_hook_init(&g_cluster.hooksys, XCE_CONTROLLER_HOOK_MAX);
    ha_config_t cfg = {
        .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
        .max_promotion_retries = 5,
    };
    ha_init(&g_cluster.ha, &cfg, 0);

    ha_promote_to_primary(&g_cluster);
    int count_after_first = g_cluster.hooksys.hooks[XCE_CONTROLLER_HOOK_JOB].count;

    ha_promote_to_primary(&g_cluster);  /* second call — hooks_registered should gate */
    int count_after_second = g_cluster.hooksys.hooks[XCE_CONTROLLER_HOOK_JOB].count;

    TEST_ASSERT_EQUAL(count_after_first, count_after_second);
    ha_fini(&g_cluster.ha);
    xce_hook_fini(&g_cluster.hooksys);
}

Run ninja -C build test — must fail (promote/demote stubs don't implement these behaviors).

GREEN — full implementation

ha_promote_to_primary() in ha.c (steps follow design doc Part 5.6):

int ha_promote_to_primary(cluster_t* c)
{
    ha_state_t* ha = &c->ha;

    /* Step 2: increment term — fencing token for stale standbys */
    uint64_t new_term = atomic_fetch_add_explicit(&ha->term, 1ULL, memory_order_acq_rel) + 1;
    INFO("ha: promoting to primary, new term=%" PRIu64, new_term);

    /* Step 3: flush WAL + reset replication CRC chain.
     * persist_flush() is a no-op stub until P0 persist module is fully wired;
     * the CRC chain reset is critical for standbys to accept first broadcast. */
    persist_flush(&c->persist);
    c->persist.last_replicated_crc = WAL_CRC_NONE;  /* COH-5: chain restart on promotion */

    /* Step 4: switch role to PRIMARY — AFTER recovery succeeds.
     * Uses release ordering so vcpu threads see the change immediately. */
    atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release);

    /* Step 5: register persistence hooks (idempotent) */
    if (!ha->hooks_registered) {
        xce_install_hooks(c);   /* same call as cluster_init() — idempotent via guard */
        ha->hooks_registered = 1;
    }

    /* Reset retry state */
    atomic_store_explicit(&ha->promotion_failed,      0, memory_order_relaxed);
    atomic_store_explicit(&ha->promotion_retry_count, 0, memory_order_relaxed);

    return 0;
}

ha_demote_to_standby() in ha.c (design doc Part 5.7):

int ha_demote_to_standby(cluster_t* c)
{
    ha_state_t* ha = &c->ha;

    /* Disable persistence hooks BEFORE role switch so in-flight requests
     * see the role change and stop after the hook is already disabled.
     * Design doc Part 5.7: "Disabling must happen before the role switch." */
    if (ha->hooks_registered) {
        xce_uninstall_hooks(c);   /* calls xce_disable_hook() for each registered hook */
        ha->hooks_registered = 0;
    }

    /* Role switch with release ordering — pairs with acquire loads in vcpu threads */
    atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);

    INFO("ha: demoted to standby");
    return 0;
}

SIGUSR1 handler in ha.c:

/* Global pointer set during ha_install_signal_handler() — signal handler must be async-safe */
static cluster_t* g_sigusr1_cluster = NULL;
static atomic_int  g_sigusr1_enabled = ATOMIC_VAR_INIT(1);

static void sigusr1_handler(int sig)
{
    (void)sig;
    if (!atomic_load_explicit(&g_sigusr1_enabled, memory_order_relaxed))
        return;
    if (g_sigusr1_cluster) {
        /* Async-signal-safe: only atomic store — no malloc, mutex, or SSL */
        atomic_store_explicit(&g_sigusr1_cluster->ha.failover_requested, 1,
                              memory_order_release);
    }
}

void ha_install_signal_handler(cluster_t* c, int enable_sigusr1)
{
    g_sigusr1_cluster = c;
    atomic_store_explicit(&g_sigusr1_enabled, enable_sigusr1 ? 1 : 0,
                          memory_order_relaxed);

    struct sigaction sa;
    memset(&sa, 0, sizeof(sa));
    sa.sa_handler = sigusr1_handler;
    sigemptyset(&sa.sa_mask);
    sigaction(SIGUSR1, &sa, NULL);
}

Run ninja -C build test — must pass.

Commit:

git add src/controller/ha.h src/controller/ha.c src/controller/testsuite/test_ha_role.c
git commit -m "feat: full ha_promote_to_primary + ha_demote_to_standby + SIGUSR1 handler (P1-3a)"

P1-4: Cap'n Proto schema additions — `term` in Request, `standby` in HostRole, `RedirectInfo`

Files: src/schema/controller.capnp

Audit first: The HA opcodes (replEntry, replAck, redirect, heartbeat, steppingDown, replSnapshot, replCatchUpDone) were already added in P0-9. Verify:

grep -c 'heartbeat\|steppingDown\|replEntry\|replAck\|redirect\|replSnapshot\|replCatchUpDone' \
    src/schema/controller.capnp
# expected: 7 (one match each)

What P1-4 must still add (check each before adding):

standby @5; to enum HostRole (current max is apiserver @4)
term @22: UInt64 = 0; field to struct Request (after from @2)
struct RedirectInfo with host + port
redirect @8: RedirectInfo in RequestReply.data union

RED — schema compile test

# Expect compile failure for missing types before adding them:
grep 'RedirectInfo\|standby.*HostRole\|term.*Request' src/schema/controller.capnp
# should return empty — these are not yet present

GREEN — schema additions

Add standby to HostRole:

enum HostRole {
  client     @0;
  worker     @1;
  scheduler  @2;
  controller @3;
  apiserver  @4;
  standby    @5;   # P1-4: HA standby controller identity
}

Add term to Request (new field at next available index — check current max first):

struct Request $C.mapname("xce_request_t"){
  opcode  @0: OPCode;
  seq     @1: UInt64;
  from    @2: PeerInfo;
  term    @22: UInt64 = 0 $C.mapname("term");   # P1-4: election fencing token
  body      : union {
    # ... existing body fields unchanged ...
  }
}

NOTE: Cap'n Proto field numbers in a struct must not skip — use the next available number after the current highest union body field. Check @21: HostControl is the current last field and use @22 for term. If the body union already uses higher numbers internally (union fields share the struct's field number space in capnp-c), verify with capnpc --src-prefix=src/schema -oc src/schema/controller.capnp and fix the number if there is a collision.

Add RedirectInfo struct:

struct RedirectInfo $C.mapname("xce_redirect_t") {
  host @0 :Text $C.mapname("host");
  port @1 :UInt32 $C.mapname("port");
}

Add redirect to RequestReply.data union (next available index after clusterconf @7):

struct RequestReply $C.mapname("xce_reply_t"){
  seq    @0: UInt64;
  opcode @1: OPCode;
  status @2: Int32;

  data     : union {
    nodata       @3: Void;
    debugmsg     @4: Text;
    getdata      @5: Data;
    jobid        @6: UInt64;
    clusterconf  @7: ClusterConf;
    redirect     @8: RedirectInfo $C.mapname("redirect");   # P1-4
  }
}

Verify schema compiles:

nix develop --command bash -c "ninja -C build && ninja -C build test"

Commit:

git add src/schema/controller.capnp
git commit -m "feat: add standby HostRole, term field to Request, RedirectInfo struct (P1-4)"

P1-5: Wire `ha_state_t ha` into `cluster_t` + startup sequence

Files:

src/include/xce/cluster_def.h — add ha_state_t ha; field to cluster_t
src/controller/cluster.c — call ha_init(), am_i_primary(), ha_start_watchdog() in correct order; call ha_fini() in cluster_fini()

Design doc notes:

ha_init() must be called before am_i_primary() (mutex initialized in ha_init)
am_i_primary() runs before cluster_start() (before iomgr starts)
ha_start_watchdog() called from cluster_start() after iomgr is up
ha_fini() must be called from cluster_fini() before persist_fini() (watchdog may access c->persist on its last tick)

RED — failing test

Add to src/controller/testsuite/test_cluster_persist.c:

void test_cluster_ha_field_accessible(void)
{
    /* cluster_t must have an ha field after P1-5 */
    cluster_t c;
    memset(&c, 0, sizeof(c));
    /* Just verifying the field exists at compile time */
    TEST_ASSERT_NOT_NULL(&c.ha);
    /* role must be zero-initialized (CTRL_STANDBY = 0) */
    TEST_ASSERT_EQUAL(0, (int32_t)atomic_load_explicit(&c.ha.role, memory_order_acquire));
}

Run ninja -C build test — must fail (cluster_t has no ha field yet).

GREEN — implementation

src/include/xce/cluster_def.h — add field:

typedef struct cluster_s {
    myconf_t*           myconf;
    xce_peerinfo_t*     myhost;
    hooksys_t           hooksys;
    iomgr_t             iomgr;
    zarray_t            endpoints;
    endpoint_t          ep;
    zarray_t            configs;
    zarray_t            sharedinst;
    controller_t        controllers[N_CONTROLLER_MAX];
    cluster_apiserver_t apiservers[N_APISERVER_MAX];
    int                 myidx;
    int                 master;
    int                 ncontrollers;
    delaymgr_t          delaymgr;
    scheduler_t         sched;
    var                 jobmgr;
    var                 resmgr;
    var                 workermgr;
    var                 qmgr;
    var                 configmgr;
    vtaskmgr_t          vtaskmgr;
    cbuf_t              wbuf;
    int                 running;
    persist_t           persist;   /* Persistence module — P0-F */
    ha_state_t          ha;        /* HA role management — P1-5 */
} cluster_t;

Add #include "ha.h" to cluster_def.h (after the existing #include "persist.h").

src/controller/cluster.c — modify cluster_init() to add HA init after persist_recover:

    /* HA init: must happen BEFORE am_i_primary() so ha.lock is initialized */
    {
        ha_config_t ha_cfg = {
            .heartbeat_interval_ms    = 3000,  /* TODO: read from cluster_config */
            .heartbeat_timeout_ms     = 10000,
            .election_timeout_ms      = 2000,
            .probe_connect_timeout_ms = 500,
            .max_promotion_retries    = 5,
        };
        int my_priority = 0;  /* TODO: look up own position in candidates[] */
        rc = ha_init(&c->ha, &ha_cfg, my_priority);
        if (rc != 0) {
            ERROR("ha_init failed: %d", rc);
            persist_fini(&c->persist);
            return rc;
        }

        /* Restore election term from WAL/snapshot (design doc Part 4, step 6) */
        uint64_t recovered_term = c->persist.last_term; /* set by persist_recover() */
        if (recovered_term > 0) {
            atomic_store_explicit(&c->ha.term, recovered_term, memory_order_release);
        }

        /* Election: am I the primary? (runs pre-iomgr blocking probes) */
        if (am_i_primary(c)) {
            /* Primary: register persistence hooks NOW (standbys defer to promotion) */
            ha_promote_to_primary(c);
        }

        ha_install_signal_handler(c, 1 /* enable SIGUSR1 failover */);
    }

src/controller/cluster_start() — add watchdog start:

    rc = ha_start_watchdog(c);
    if (rc != 0) {
        ERROR("ha_start_watchdog failed: %d", rc);
        return rc;
    }

src/controller/cluster_fini() — ha_fini before persist_fini:

void cluster_fini(cluster_t* c)
{
    ha_fini(&c->ha);        /* joins watchdog thread — BEFORE persist_fini */
    persist_fini(&c->persist);
    /* ... rest of existing cleanup ... */
}

Run ninja -C build test — must pass.

Commit:

git add src/include/xce/cluster_def.h src/controller/cluster.c \
        src/controller/testsuite/test_cluster_persist.c
git commit -m "feat: wire ha_state_t into cluster_t; call ha_init/start_watchdog/fini in startup sequence (P1-5)"

P1-6: Redirect logic in `join.c`

Files: src/controller/join.c

Design doc ref: Part 9.4 — standby sends E_XCE_REDIRECT + RedirectInfo for worker/scheduler/apiserver join requests.

Edge case: if primary_host is empty (standby has not yet seen a heartbeat from any primary — cold start), return E_XCE_NOMASTER instead of a garbage redirect.

RED — failing test

Add src/controller/testsuite/test_ha_redirect.c:

#include <unity/unity.h>
#include "ha.h"
#include "xce/cluster_def.h"

/* Minimal mock: verify handle_join returns E_XCE_REDIRECT for worker when standby */
/* NOTE: This test is a compile-level integration check — full end-to-end redirect
 * requires a live endpoint (P1-7). Here we verify the path is reachable by calling
 * the testable inner function directly with a mock request. */

extern int handle_join_for_test(cluster_t* c, xce_request_t* req,
                                 int* out_status, char* out_host, uint32_t* out_port);

void test_redirect_worker_when_standby(void)
{
    cluster_t c;
    memset(&c, 0, sizeof(c));

    ha_config_t cfg = {
        .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
        .max_promotion_retries = 5,
    };
    ha_init(&c.ha, &cfg, 1);
    /* Standby with known primary */
    snprintf(c.ha.primary_host, sizeof(c.ha.primary_host), "ctrl-0.example.com");
    c.ha.primary_port = 7000;

    xce_request_t req = { 0 };
    xce_joindata_t join = { .role = HostRole_worker };
    req.body_which = Request_body_joinmsg;
    req.body.joinmsg = &join;

    int      status   = 0;
    char     host[XCE_MAXNAME_LEN] = {0};
    uint32_t port     = 0;

    handle_join_for_test(&c, &req, &status, host, &port);

    TEST_ASSERT_EQUAL(E_XCE_REDIRECT, status);
    TEST_ASSERT_EQUAL_STRING("ctrl-0.example.com", host);
    TEST_ASSERT_EQUAL(7000, port);

    ha_fini(&c.ha);
}

void test_redirect_nomaster_when_no_primary_known(void)
{
    cluster_t c;
    memset(&c, 0, sizeof(c));

    ha_config_t cfg = {
        .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
        .max_promotion_retries = 5,
    };
    ha_init(&c.ha, &cfg, 1);
    /* primary_host is empty — cold start, no heartbeat received yet */

    xce_request_t req = { 0 };
    xce_joindata_t join = { .role = HostRole_scheduler };
    req.body_which = Request_body_joinmsg;
    req.body.joinmsg = &join;

    int      status = 0;
    char     host[XCE_MAXNAME_LEN] = {0};
    uint32_t port   = 0;

    handle_join_for_test(&c, &req, &status, host, &port);

    TEST_ASSERT_EQUAL(E_XCE_NOMASTER, status);

    ha_fini(&c.ha);
}

Run ninja -C build test — must fail.

GREEN — implementation

Add #include "ha.h" to join.c.

Add a testable inner function in join.c:

/* Testable redirect check — separated from handle_join for unit testing.
 * Returns 1 if a redirect was needed (out_* filled), 0 if normal join proceeds. */
static int check_redirect(cluster_t* c, xce_request_t* req,
                           int* out_status, char* out_host, uint32_t* out_port)
{
    if ((ctrl_role_t)atomic_load_explicit(&c->ha.role, memory_order_acquire)
            != CTRL_STANDBY)
        return 0;

    xce_joindata_t* join = req->body.joinmsg;
    switch (join->role) {
    case HostRole_worker:
    case HostRole_scheduler:
    case HostRole_apiserver:
        /* Guard: no primary known yet */
        if (c->ha.primary_host[0] == '\0' || c->ha.primary_port == 0) {
            *out_status = E_XCE_NOMASTER;
            return 1;
        }
        *out_status = E_XCE_REDIRECT;
        strncpy(out_host, c->ha.primary_host, XCE_MAXNAME_LEN - 1);
        out_host[XCE_MAXNAME_LEN - 1] = '\0';
        *out_port = c->ha.primary_port;
        return 1;
    default:
        return 0;
    }
}

/* Test shim */
int handle_join_for_test(cluster_t* c, xce_request_t* req,
                          int* out_status, char* out_host, uint32_t* out_port)
{
    return check_redirect(c, req, out_status, out_host, out_port);
}

Modify handle_join() to call check_redirect() before normal processing:

int handle_join(cluster_t* c, endpoint_t* ep, xce_request_t* request)
{
    int      redirect_status = 0;
    char     redirect_host[XCE_MAXNAME_LEN] = {0};
    uint32_t redirect_port = 0;

    if (check_redirect(c, request, &redirect_status, redirect_host, &redirect_port)) {
        xce_reply_t reply;
        init_reply_by_req(&reply, request);
        reply.status = redirect_status;
        if (redirect_status == E_XCE_REDIRECT) {
            reply.data_which = RequestReply_data_redirect;
            reply.data.redirect.host = redirect_host;
            reply.data.redirect.port = redirect_port;
        }
        return xce_reply(ep, c->myhost, &reply);
    }

    return handle_join_(c, ep, request);
}

Add E_XCE_REDIRECT and E_XCE_NOMASTER to src/include/xce/types.h (or wherever XCE error codes live):

#define E_XCE_REDIRECT  (-100)   /* P1-6: standby redirects to primary */
#define E_XCE_NOMASTER  (-101)   /* P1-6: standby has no known primary yet */

Run ninja -C build test — must pass.

Commit:

git add src/controller/join.c src/include/xce/types.h \
        src/controller/testsuite/test_ha_redirect.c
git commit -m "feat: add standby redirect logic in handle_join(); E_XCE_REDIRECT + E_XCE_NOMASTER (P1-6)"

P1-7: Election, promotion, term fencing, and redirect tests

Files:

src/controller/testsuite/test_ha_election.c — new test file covering the scenarios below

Note: Full end-to-end election with live SSL requires integration test infra (P3). P1-7 covers unit-testable invariants using stubs and direct function calls.

Tests to implement

#include <unity/unity.h>
#include <stdatomic.h>
#include "ha.h"
#include "xce/cluster_def.h"

static cluster_t g_c;

void setUp(void)   { memset(&g_c, 0, sizeof(g_c)); }
void tearDown(void) { }

/* 1. Term fencing: standby ignores message with term < ha->term */
void test_stale_term_message_ignored(void)
{
    ha_config_t cfg = { .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
                        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
                        .max_promotion_retries = 5 };
    ha_init(&g_c.ha, &cfg, 1);
    atomic_store_explicit(&g_c.ha.term, 5ULL, memory_order_release);

    int stale = ha_is_stale_term(&g_c, 3ULL);  /* term 3 < current 5 */
    TEST_ASSERT_EQUAL(1, stale);

    int current = ha_is_stale_term(&g_c, 5ULL);  /* same term — not stale */
    TEST_ASSERT_EQUAL(0, current);

    ha_fini(&g_c.ha);
}

/* 2. Supersede: higher term received -> demote current primary */
void test_higher_term_triggers_demotion(void)
{
    ha_config_t cfg = { .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
                        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
                        .max_promotion_retries = 5 };
    ha_init(&g_c.ha, &cfg, 0);
    ha_promote_to_primary(&g_c);  /* become primary at term 1 */

    /* Simulate receiving a message with higher term (term 5) */
    ha_on_higher_term(&g_c, 5ULL);

    /* Must demote to standby and update term */
    TEST_ASSERT_EQUAL(CTRL_STANDBY,
        (ctrl_role_t)atomic_load_explicit(&g_c.ha.role, memory_order_acquire));
    TEST_ASSERT_EQUAL_UINT64(5ULL,
        atomic_load_explicit(&g_c.ha.term, memory_order_acquire));
    ha_fini(&g_c.ha);
}

/* 3. promotion_retry_count increments on repeated failed promotions */
void test_promotion_retry_count(void)
{
    ha_config_t cfg = { .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
                        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
                        .max_promotion_retries = 3 };
    ha_init(&g_c.ha, &cfg, 1);

    /* Simulate promotion failure */
    atomic_store_explicit(&g_c.ha.promotion_failed, 1, memory_order_relaxed);
    atomic_store_explicit(&g_c.ha.promotion_retry_count, 2, memory_order_relaxed);

    /* At max_promotion_retries=3, count=2 allows one more; count=3 blocks */
    atomic_store_explicit(&g_c.ha.promotion_retry_count, 3, memory_order_relaxed);

    /* ha_check_heartbeat_for_test with expired timeout: must NOT retry at max count */
    /* Inject a fake stale last_heartbeat */
    pthread_mutex_lock(&g_c.ha.lock);
    clock_gettime(CLOCK_MONOTONIC, &g_c.ha.last_heartbeat);
    g_c.ha.last_heartbeat.tv_sec -= 20;  /* 20s ago — well past 10s timeout */
    /* Inject a fake primary_ep != NULL so watchdog proceeds past no_primary check */
    g_c.ha.primary_ep = (endpoint_t*)0xDEADBEEF;  /* non-NULL sentinel */
    pthread_mutex_unlock(&g_c.ha.lock);

    ctrl_role_t before = (ctrl_role_t)atomic_load_explicit(&g_c.ha.role, memory_order_acquire);
    ha_check_heartbeat_for_test(&g_c);
    ctrl_role_t after  = (ctrl_role_t)atomic_load_explicit(&g_c.ha.role, memory_order_acquire);

    /* Role must NOT change — retry limit exceeded */
    TEST_ASSERT_EQUAL(before, after);

    ha_fini(&g_c.ha);
}

/* 4. OP_steppingDown handler: validates term and sender, then runs am_i_primary */
void test_stepping_down_handler_validates_term(void)
{
    ha_config_t cfg = { .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
                        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
                        .max_promotion_retries = 5 };
    ha_init(&g_c.ha, &cfg, 0);
    atomic_store_explicit(&g_c.ha.term, 3ULL, memory_order_release);
    snprintf(g_c.ha.primary_host, sizeof(g_c.ha.primary_host), "ctrl-1");
    g_c.ha.primary_port = 7001;

    /* Stale term (term=2 < our term=3) — must be ignored */
    int rc = ha_handle_stepping_down(&g_c, 2ULL, "ctrl-1", 7001);
    TEST_ASSERT_EQUAL(-EINVAL, rc);  /* stale term rejected */

    /* Correct term but wrong sender — must be ignored */
    rc = ha_handle_stepping_down(&g_c, 3ULL, "ctrl-99", 9999);
    TEST_ASSERT_EQUAL(-EINVAL, rc);

    ha_fini(&g_c.ha);
}

/* 5. Redirect: worker join on primary passes through (no redirect) */
void test_no_redirect_when_primary(void)
{
    ha_config_t cfg = { .heartbeat_interval_ms = 3000, .heartbeat_timeout_ms = 10000,
                        .election_timeout_ms = 2000, .probe_connect_timeout_ms = 500,
                        .max_promotion_retries = 5 };
    ha_init(&g_c.ha, &cfg, 0);
    ha_promote_to_primary(&g_c);

    xce_request_t req = { 0 };
    xce_joindata_t join = { .role = HostRole_worker };
    req.body_which = Request_body_joinmsg;
    req.body.joinmsg = &join;

    int      status = 0;
    char     host[XCE_MAXNAME_LEN] = {0};
    uint32_t port   = 0;

    int redirected = handle_join_for_test(&g_c, &req, &status, host, &port);
    TEST_ASSERT_EQUAL(0, redirected);  /* no redirect when primary */

    ha_fini(&g_c.ha);
}

Add ha_is_stale_term(), ha_on_higher_term(), ha_handle_stepping_down() to ha.h + ha.c:

/* ha.h additions */
int ha_is_stale_term(cluster_t* c, uint64_t incoming_term);  /* 1 if incoming < current */
void ha_on_higher_term(cluster_t* c, uint64_t higher_term);   /* demote + update term */
int  ha_handle_stepping_down(cluster_t* c, uint64_t msg_term,
                              const char* sender_host, uint32_t sender_port);

/* ha.c implementations */
int ha_is_stale_term(cluster_t* c, uint64_t incoming)
{
    return incoming < atomic_load_explicit(&c->ha.term, memory_order_acquire);
}

void ha_on_higher_term(cluster_t* c, uint64_t higher_term)
{
    atomic_store_explicit(&c->ha.term, higher_term, memory_order_release);
    ha_demote_to_standby(c);
}

int ha_handle_stepping_down(cluster_t* c, uint64_t msg_term,
                             const char* sender_host, uint32_t sender_port)
{
    ha_state_t* ha = &c->ha;

    /* Validate: msg_term must match our current term */
    if (msg_term != atomic_load_explicit(&ha->term, memory_order_acquire))
        return -EINVAL;

    /* Validate: sender must be the current primary */
    if (strncmp(sender_host, ha->primary_host, XCE_MAXNAME_LEN) != 0 ||
        sender_port != ha->primary_port)
        return -EINVAL;

    /* Dispatch am_i_primary() asynchronously (full impl: launch pthread — P3) */
    /* Stub: immediately check if we are priority 0 */
    if (am_i_primary(c)) {
        ha_promote_to_primary(c);
    }
    return 0;
}

Run ninja -C build test — must pass.

Commit:

git add src/controller/testsuite/test_ha_election.c \
        src/controller/ha.h src/controller/ha.c
git commit -m "feat: add term fencing, ha_on_higher_term, ha_handle_stepping_down + P1-7 tests (P1-7)"

Final P1 verification

nix develop --command bash -c "ninja -C build && ninja -C build test"
# Expected: all existing P0 tests (89) + new P1 tests pass
# New test count: ~25 additional tests across test_ha_role, test_ha_redirect, test_ha_election

Collect total test count:

ninja -C build test 2>&1 | grep -E 'Tests run|PASS|FAIL' | tail -5

Summary of files created/modified by P1

File	Action	Task
`src/controller/ha.h`	Create	P1-1 → P1-3a
`src/controller/ha.c`	Create	P1-2 → P1-7
`src/include/hook.h`	Modify: `_Atomic hook_fn_t` + `xce_disable_hook` decl	P1-2a
`src/lib/hook.c`	Modify: atomic fn, disable, NULL-slot reuse	P1-2a
`src/schema/controller.capnp`	Modify: `standby`, `term`, `RedirectInfo`	P1-4
`src/include/xce/cluster_def.h`	Modify: add `ha_state_t ha` to `cluster_t`	P1-5
`src/controller/cluster.c`	Modify: `ha_init/start/fini` in startup sequence	P1-5
`src/controller/join.c`	Modify: redirect check in `handle_join()`	P1-6
`src/include/xce/types.h`	Modify: `E_XCE_REDIRECT`, `E_XCE_NOMASTER`	P1-6
`src/lib/testsuite/test_hooks.c`	Modify: add disable/reuse tests	P1-2a
`src/controller/testsuite/test_ha_role.c`	Create	P1-2 → P1-3a
`src/controller/testsuite/test_ha_redirect.c`	Create	P1-6
`src/controller/testsuite/test_ha_election.c`	Create	P1-7

53 KiB Raw Blame History Unescape Escape

P1: HA Foundation — Implementation Plan

P1-1: Verify iomgr/SSL pipeline reuse — no new TLS work needed

RED — audit test (compile-only, no new test file yet)

GREEN — document the decision in ha.h (stubbed header, creates the file)

P1-2a: Atomic hook disable — xce_disable_hook() + _Atomic hook_fn_t

RED — failing test

GREEN — implementation

P1-2: ha.c + ha.h — role management, am_i_primary(), promotion/demotion stubs

RED — failing tests

GREEN — implementation

P1-3: Heartbeat watchdog pthread + ha_check_heartbeat()

RED — failing test

GREEN — implementation

P1-3a: Promotion sequence + SIGUSR1 handler

RED — failing tests

GREEN — full implementation

P1-4: Cap'n Proto schema additions — term in Request, standby in HostRole, RedirectInfo

RED — schema compile test

GREEN — schema additions

P1-5: Wire ha_state_t ha into cluster_t + startup sequence

RED — failing test

GREEN — implementation

P1-6: Redirect logic in join.c

RED — failing test

GREEN — implementation

P1-7: Election, promotion, term fencing, and redirect tests

Tests to implement

Final P1 verification

Summary of files created/modified by P1

53 KiB

Raw Blame History

P1-2a: Atomic hook disable — `xce_disable_hook()` + `_Atomic hook_fn_t`

P1-2: `ha.c` + `ha.h` — role management, `am_i_primary()`, promotion/demotion stubs

P1-3: Heartbeat watchdog pthread + `ha_check_heartbeat()`

P1-4: Cap'n Proto schema additions — `term` in Request, `standby` in HostRole, `RedirectInfo`

P1-5: Wire `ha_state_t ha` into `cluster_t` + startup sequence

P1-6: Redirect logic in `join.c`