> **ARCHIVED — superseded.** This planned an LSF-style priority election (blocking `am_i_primary()` probes + dedicated watchdog pthread for self-promotion) on top of the pre-Kiro persistence-and-ha-design.md. What actually shipped is quorum-based Raft: `am_i_primary()` was deleted and replaced by `ha_start_election_async` (see closed specs raft-library, election-iomgr-refactor, raft-ha-core-safety, ha-prevote, raft-log-replication, raft-commit-safety, ha-term-safety, ha-single-leader-role), with real voting/term logic in `src/lib/raft/raft.c`. A few surface artifacts (`ha_init`, `ha_promote_to_primary`, `my_priority` as peer index, `E_XCE_REDIRECT`/`E_XCE_NOMASTER`, join.c redirect) survived in similar form but were rearchitected around Raft. # 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`):** ```bash 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: ```bash 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) ```bash # 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`: ```c #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:** ```bash 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`: ```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: ```c #if !defined(__HOOK_H__) #define __HOOK_H__ 1 #include #include #include /* 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: ```c 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: ```c 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: ```c 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:** ```bash 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`: ```c #include #include #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): ```c #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 #include #include #include #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): ```c #define _GNU_SOURCE #include "ha.h" #include #include #include #include #include #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:** ```bash 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`: ```c #include /* 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`: ```c #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`: ```c _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`: ```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()`: ```c 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:** ```bash 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`: ```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): ```c 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): ```c 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`**: ```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:** ```bash 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: ```bash 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): 1. `standby @5;` to `enum HostRole` (current max is `apiserver @4`) 2. `term @22: UInt64 = 0;` field to `struct Request` (after `from @2`) 3. `struct RedirectInfo` with `host` + `port` 4. `redirect @8: RedirectInfo` in `RequestReply.data` union ### RED — schema compile test ```bash # 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`: ```capnp 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): ```capnp 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: ```capnp 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`): ```capnp 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: ```bash nix develop --command bash -c "ninja -C build && ninja -C build test" ``` **Commit:** ```bash 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`: ```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: ```c 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: ```c /* 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: ```c 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: ```c 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:** ```bash 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`: ```c #include #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`: ```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: ```c 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): ```c #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:** ```bash 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 ```c #include #include #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`: ```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); ``` ```c /* 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:** ```bash 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 ```bash 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: ```bash 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 |