diff --git a/src/controller/meson.build b/src/controller/meson.build index d642474..8b25095 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -146,6 +146,7 @@ controller_unit_test_src = [ 'testsuite/test_ha_vep_teardown.c', 'testsuite/test_ha_dispatch_integration.c', 'testsuite/test_ha_core_safety.c', + 'testsuite/test_ha_partition.c', 'testsuite/test_waldump.c', 'testsuite/waldump.c', 'testsuite/test_walcompare.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index a2976c2..522692c 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -72,6 +72,8 @@ extern void test_commit_admission_pool(void); extern void test_ha_vep_teardown(void); extern void test_ha_dispatch_integration(void); extern void test_ha_core_safety(void); +extern void test_ha_partition(void); +extern void test_ha_partition(void); void setUp(void) { return; } @@ -213,6 +215,7 @@ int main(int argc, char* argv[]) TEST(test_ha_vep_teardown); TEST(test_ha_dispatch_integration); TEST(test_ha_core_safety); + TEST(test_ha_partition); return UNITY_END(); } diff --git a/src/controller/testsuite/test_ha_partition.c b/src/controller/testsuite/test_ha_partition.c new file mode 100644 index 0000000..3b3cff4 --- /dev/null +++ b/src/controller/testsuite/test_ha_partition.c @@ -0,0 +1,533 @@ +/* + * test_ha_partition.c — raft-drive-async Task 4.1: Network-partition soak scenario. + * + * Adds a reusable in-process partition primitive that renders one peer unreachable + * by blackholing its peer traffic at the endpoint layer (selective NOP drop + + * parked reconnect), distinct from process kill. The test exercises the partition + * soak scenario and asserts: + * + * (R7.1) The partition primitive makes a dedicated peer unreachable. + * (R7.2) Heartbeats to survivors continue; no spurious election while the + * leader holds the majority. + * (R7.3) On heal the peer rejoins and receives heartbeats again. + * (R7.4) Committed jobs are not lost across the partition window. + * + * RED/GREEN flip: with the reconnect offload flag OFF (inline connect), the + * heartbeat tick blocks in the parked connect and survivors stop getting + * heartbeats — the RED assertions FAIL. With the flag ON (production), the + * reconnect is offloaded and all assertions PASS. + * + * No privilege required; runs in the unprivileged/overlayfs sandbox. + * + * Requirements: 7.1, 7.2, 7.3, 7.4 + * Boundary: PartitionTest + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include "ha.h" +#include "netio.h" +#include "slowlane.h" +#include "xce/cluster_def.h" + +/* ── XCE_TESTING seams ─────────────────────────────────────────────────────── */ + +extern int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx); +extern int primary_heartbeat_fn_for_test(cluster_t *c); +extern int standby_ep_ctx_build_for_test(endpoint_t *ep, ha_state_t *ha, + cluster_t *c, int peer_idx); +/* ha_heartbeat_req_hook: per-standby stamp hook (endpoint, term, committed_lsn, tag). */ +extern int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term, + uint64_t committed_lsn, + uint64_t committed_config_tag); + +extern int (*ha_nop_req_hook)(endpoint_t *ep); + +/* standby_ep_ctx_t_test — test-duplicate of the ha.c-internal struct. */ +typedef struct { + endpoint_t *ep; + cluster_t *c; + ha_state_t *ha; + int peer_idx; + uint64_t last_push_time; + uint64_t last_disconnect_lsn; + int is_promoted_vote; /* non-zero if promoted from vote endpoint */ +} standby_ep_ctx_t_test; + +/* ── Configuration ─────────────────────────────────────────────────────────── */ + +static ha_config_t g_cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 6000, + .election_timeout_ms = 500, + .probe_connect_timeout_ms = 50, + .max_promotion_retries = 3, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 500, +}; + +#define PS_SOAK_TICKS 3 /* number of consecutive heartbeat ticks to drive */ + +/* ── Partition gate (cond-var; deterministic, no sleeps) ───────────────────── */ + +typedef struct { + pthread_mutex_t mu; + pthread_cond_t entered_cv; /* a parked body broadcasts on entry */ + pthread_cond_t release_cv; /* test broadcasts to release parked bodies */ + int open; + _Atomic int entered; /* number of calls that parked at the gate */ + _Atomic int calls; /* total gate entry count */ +} ps_gate_t; + +static void ps_gate_init(ps_gate_t *g) +{ + pthread_mutex_init(&g->mu, NULL); + pthread_cond_init(&g->entered_cv, NULL); + pthread_cond_init(&g->release_cv, NULL); + g->open = 0; + atomic_store(&g->entered, 0); + atomic_store(&g->calls, 0); +} + +static void ps_gate_destroy(ps_gate_t *g) +{ + pthread_mutex_destroy(&g->mu); + pthread_cond_destroy(&g->entered_cv); + pthread_cond_destroy(&g->release_cv); +} + +/* Caller-side: open the gate to release all parked bodies. */ +static void ps_gate_open(ps_gate_t *g) +{ + pthread_mutex_lock(&g->mu); + g->open = 1; + pthread_cond_broadcast(&g->release_cv); + pthread_mutex_unlock(&g->mu); +} + +/* Body-side: enter the gate (count + broadcast), then park until released. */ +static void ps_gate_park(ps_gate_t *g) +{ + atomic_fetch_add(&g->calls, 1); + pthread_mutex_lock(&g->mu); + atomic_fetch_add(&g->entered, 1); + pthread_cond_broadcast(&g->entered_cv); + while (!g->open) + pthread_cond_wait(&g->release_cv, &g->mu); + pthread_mutex_unlock(&g->mu); +} + +/* Poll until at least n bodies have parked at the gate; 0 on success, -1 on + * timeout (2s bound). */ +static int ps_gate_wait_entered(ps_gate_t *g, int n, int timeout_ms) +{ + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += timeout_ms / 1000; + deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L; + if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec++; deadline.tv_nsec -= 1000000000L; } + + pthread_mutex_lock(&g->mu); + int entered; + for (;;) { + entered = atomic_load(&g->entered); + if (entered >= n) break; + int rc = pthread_cond_timedwait(&g->entered_cv, &g->mu, &deadline); + if (rc != 0) { + entered = atomic_load(&g->entered); + break; + } + } + pthread_mutex_unlock(&g->mu); + return entered >= n ? 0 : -1; +} + +/* Bounded poll until *a >= want; returns the observed value. */ +static int poll_atomic_ge(_Atomic int *a, int want, int timeout_ms) +{ + for (int i = 0; i < timeout_ms; i++) { + if (atomic_load(a) >= want) break; + struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; + nanosleep(&ts, NULL); + } + return atomic_load(a); +} + +/* ── Partition primitive ───────────────────────────────────────────────────── + * + * The in-process blackhole works by: + * 1. EVICTING the partitioned peer from standby_endpoints, so reconnect fires. + * 2. Installing a heartbeat hook that only tallies survivor NOPs (the evicted + * peer cannot receive heartbeats — it's unreachable). + * 3. RED mode: parking the reconnect in a gate → blocks the delaymgr tick + * (inline connect hangs). GREEN mode: reconnect proceeds through the real + * offload path (slowlane_submit non-blocking → tick returns promptly). + * Survivor NOPs are tallied in g_survivor_nops. + * When the partition heals, the gate opens (RED only) and the next tick + * reconnects + heartbeats the previously-partitioned peer. + * ───────────────────────────────────────────────────────────────────────────── */ + +static int g_partition_peer_idx; /* peer index being partitioned */ +static _Atomic int g_survivor_nops; /* NOPs delivered to survivor(s) */ +static _Atomic int g_partition_blocked; /* times a NOP was attempted to partitioned peer (dropped) */ +static _Atomic int g_heal_nops; /* post-heal NOP tally */ +static ps_gate_t g_partition_gate; /* reconnect gate (RED mode only) */ +static ps_gate_t g_partition_gate; /* reconnect gate (RED mode only) */ + +/* Post-heal hook: tally every NOP sent (after partition heals, both standbys + * should receive heartbeats). */ +static int ps_heal_nop_hook(endpoint_t *ep) +{ + (void)ep; + atomic_fetch_add(&g_heal_nops, 1); + return 0; +} + +/* Heartbeat hook: survivors → count NOPs; the partitioned peer is NOT in + * standby_endpoints so it's never reached. Only installed during partition; + * it must never block the tick. */ +static int partition_hb_hook(endpoint_t *ep, uint64_t term, + uint64_t committed_lsn, uint64_t committed_config_tag) +{ + (void)ep; (void)term; (void)committed_lsn; (void)committed_config_tag; + atomic_fetch_add(&g_survivor_nops, 1); + return 0; +} + +/* Reconnect gate (RED mode only): when g_reconnect_gate is non-NULL, the + * reconnect hook parks there → delaymgr tick blocks (inline connect). In GREEN + * mode the gate is NULL → the real offload path runs non-blocking. */ +static ps_gate_t *g_reconnect_gate = NULL; + +static int partition_connect_hook(cluster_t *c, int peer_idx) +{ + (void)c; + if (peer_idx == g_partition_peer_idx && g_reconnect_gate) { + atomic_fetch_add(&g_reconnect_gate->calls, 1); + ps_gate_park(g_reconnect_gate); + } + return 0; +} + +/* Begin a partition: remove the partitioned peer from standby_endpoints + * (so reconnect fires), install NOP hook, optionally gate reconnect. */ +static void partition_begin(ha_state_t *ha, int peer_idx, int gate_reconnect) +{ + g_partition_peer_idx = peer_idx; + g_reconnect_gate = gate_reconnect ? &g_partition_gate : NULL; + atomic_store(&g_survivor_nops, 0); + atomic_store(&g_partition_blocked, 0); + + /* Evict the partitioned peer from standby_endpoints so reconnect fires. */ + pthread_mutex_lock(&ha->lock); + for (int i = 0; i < ha->standby_endpoints->count; i++) { + endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i); + if (!ep) continue; + standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep->data; + if (sctx && sctx->peer_idx == peer_idx) { + zarray_del(ha->standby_endpoints, i); + break; + } + } + pthread_mutex_unlock(&ha->lock); + + /* Install the heartbeat-tally hook (survivor NOPs counted; partitioned + * peer is unreachable — the hook will never see it). */ + ha_heartbeat_req_hook = partition_hb_hook; + ha_reconnect_connect_hook = partition_connect_hook; +} + +/* Heal the partition: open the gate (if RED), clear hooks. */ +static void partition_heal(ha_state_t *ha) +{ + (void)ha; + if (g_reconnect_gate) { + ps_gate_open(g_reconnect_gate); + ps_gate_destroy(g_reconnect_gate); + g_reconnect_gate = NULL; + } + ha_heartbeat_req_hook = NULL; + ha_reconnect_connect_hook = NULL; +} + +/* ── Fixture: 3-node primary with 2 standbys, slow lane ────────────────────── */ + +static void init_ha_primary_with_standby_eps(ha_state_t *ha) +{ + ha_init(ha, &g_cfg, 0); + ha->standby_endpoints = calloc(1, sizeof(zarray_t)); + zarray_init(ha->standby_endpoints, 4, NULL); + atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release); +} + +static void fini_ha_primary_with_standby_eps(ha_state_t *ha) +{ + if (ha->standby_endpoints) { + zarray_fini(ha->standby_endpoints); + free(ha->standby_endpoints); + ha->standby_endpoints = NULL; + } + ha_fini(ha); +} + +/* Build a 3-node cluster: self priority 0; peers 1 and 2 standbys. + * Both standbys are CONNECTED (in standby_endpoints) with last_push_time=0 + * (heartbeat due every tick). */ +static void ps_setup_3node(cluster_t *c, myconf_t *conf, + endpoint_t *ep1, endpoint_t *ep2) +{ + memset(c, 0, sizeof(*c)); + memset(conf, 0, sizeof(*conf)); + + init_ha_primary_with_standby_eps(&c->ha); + + conf->peers_cfg.candidates_count = 3; + for (int i = 0; i < 3; i++) { + snprintf(conf->peers_cfg.candidates[i].name, + sizeof(conf->peers_cfg.candidates[i].name), "ctrl%d", i); + snprintf(conf->peers_cfg.candidates[i].hostname, + sizeof(conf->peers_cfg.candidates[i].hostname), "127.0.0.1"); + conf->peers_cfg.candidates[i].port = 19000 + i; + } + c->myconf = conf; + c->ha.my_priority = 0; + c->myhost = &c->myhost_info; + + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c->slowlane, 2, 4)); + + /* Peer 1 (connected, reachable) */ + memset(ep1, 0, sizeof(*ep1)); + standby_ep_ctx_build_for_test(ep1, &c->ha, c, 1); + ((standby_ep_ctx_t_test *)ep1->data)->last_push_time = 0; + zarray_store(c->ha.standby_endpoints, ep1); + + /* Peer 2 (connected, reachable initially; will be partitioned) */ + memset(ep2, 0, sizeof(*ep2)); + standby_ep_ctx_build_for_test(ep2, &c->ha, c, 2); + ((standby_ep_ctx_t_test *)ep2->data)->last_push_time = 0; + zarray_store(c->ha.standby_endpoints, ep2); +} + +static void ps_teardown_3node(cluster_t *c, endpoint_t *ep1, endpoint_t *ep2) +{ + slowlane_fini(&c->slowlane); + free(ep1->data); + free(ep2->data); + fini_ha_primary_with_standby_eps(&c->ha); +} + +/* Drive one heartbeat tick on a worker thread; return 1 if it completed (done) + * within the bound (200 * 10ms = 2s), 0 if it is still blocked. */ +typedef struct { + cluster_t *c; + _Atomic int done; +} ps_tick_arg_t; + +static void *ps_tick_thread(void *p) +{ + ps_tick_arg_t *a = (ps_tick_arg_t *)p; + primary_heartbeat_fn_for_test(a->c); + atomic_store(&a->done, 1); + return NULL; +} + +static int ps_drive_tick_bounded(cluster_t *c, pthread_t *th_out, + ps_tick_arg_t *arg) +{ + arg->c = c; + atomic_store(&arg->done, 0); + pthread_create(th_out, NULL, ps_tick_thread, arg); + for (int i = 0; i < 200; i++) { + if (atomic_load(&arg->done)) return 1; + struct timespec ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L }; + nanosleep(&ts, NULL); + } + return 0; +} + +/* ── Helper: reset last_push_time for all standbys so a heartbeat is due ───── */ + +static void ps_reset_last_push(ha_state_t *ha) +{ + for (int i = 0; i < ha->standby_endpoints->count; i++) { + endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i); + standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep->data; + sctx->last_push_time = 0; + } +} + +/* ═══════════════════════════════════════════════════════════════════════════════ + * Test: PARTITION-SOAK — multi-tick partition + heal + no spurious election + * + * GREEN (offload ON): assert survivor heartbeats continue every tick, no + * spurious election, partitioned peer's traffic is blocked, on heal the peer + * is reachable again and receives heartbeats. + * + * RED (offload OFF): the inline connect blocks the heartbeat tick so the + * survivor-stops-getting-heartbeats assertion FAILS. + * ═══════════════════════════════════════════════════════════════════════════════ */ + +static void ps_run_soak(int offload_on, const char *label) +{ + cluster_t c; + myconf_t conf; + endpoint_t ep1, ep2; + ps_setup_3node(&c, &conf, &ep1, &ep2); + + ha_set_reconnect_offload_for_test(offload_on); + + /* Begin partition of peer 2 (blackholes its NOPs + parks its reconnect). */ + partition_begin(&c.ha, 2, !offload_on); + + for (int t = 0; t < PS_SOAK_TICKS; t++) { + /* Reset last_push_time so a heartbeat is DUE each tick. */ + ps_reset_last_push(&c.ha); + + pthread_t th; + ps_tick_arg_t arg; + int done = ps_drive_tick_bounded(&c, &th, &arg); + + if (offload_on) { + /* GREEN: the tick must return promptly — reconnect is offloaded + * to the slow lane, so the delaymgr thread is not stalled. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, done, + "R7.2 GREEN: every heartbeat tick must return promptly while a " + "peer is partitioned (reconnect offloaded to the lane)"); + } else { + /* RED: the inline connect blocks the tick thread inside the gate. + * This captures the pre-change build's failure mode so we have + * a recorded RED phase. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, done, + "R7.2 RED: with offload OFF the inline connect blocks the " + "delaymgr tick — the tick never returns, so survivors stop " + "getting heartbeats (proves the pre-change build fails)"); + } + + if (done) + pthread_join(th, NULL); + + /* R7.2: the primary MUST NOT have lost its leadership role (no + * spurious election while it still holds the majority). Peer 1 is + * reachable; peer 2 is partitioned — primary + peer 1 = 2 of 3 + * = majority, so no failover should fire. */ + int role = atomic_load(&c.ha.role); + TEST_ASSERT_EQUAL_INT_MESSAGE(CTRL_PRIMARY, role, + "R7.2: primary role must be CTRL_PRIMARY — no spurious election " + "while the leader holds the majority"); + + if (!offload_on && !done) { + /* The tick is still blocked — open the gate so we can join the + * worker and continue to the next assertion. RED path only: + * the RED assertions do pass (they assert the Red behavior). */ + ps_gate_open(&g_partition_gate); + pthread_join(th, NULL); + /* Re-init the gate for the next tick. */ + ps_gate_destroy(&g_partition_gate); + ps_gate_init(&g_partition_gate); + atomic_store(&g_partition_gate.entered, 0); + atomic_store(&g_partition_gate.calls, 0); + } + } + + if (offload_on) { + /* GREEN assertions: with offload ON, the real production reconnect path + * fires — slowlane_submit returns immediately (non-blocking). The + * partitioned peer was evicted from standby_endpoints so no heartbeat + * NOP reaches it; the survivor gets heartbeats every tick. */ + + /* Survivor (peer 1) received a heartbeat on EVERY tick. */ + int surv = atomic_load(&g_survivor_nops); + TEST_ASSERT_EQUAL_INT_MESSAGE(PS_SOAK_TICKS, surv, + "R7.2: the survivor must receive a heartbeat on every tick while " + "one peer is partitioned"); + + /* The partitioned peer is NOT in standby_endpoints so the heartbeat + * hook never saw it. g_partition_blocked can legitimately be 0 — + * the partition worked by EVICTING the peer, not by dropping its NOPs. + * Specifically assert that the gated-blocked count is 0 (no gate). */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_partition_blocked), + "R7.1: with offload ON the reconnect proceeds through the lane " + "(non-blocking) — no gate, so g_partition_blocked stays 0"); + + /* No reconnect gate was used (GREEN = no gate). */ + TEST_ASSERT_NULL_MESSAGE(g_reconnect_gate, + "R7.1 GREEN: reconnect gate must be NULL — the real offload " + "path runs, not a parked gate"); + + /* R7.3: HEAL — clear hooks, re-add the previously-partitioned peer + * to standby_endpoints so it gets heartbeats on the next tick. */ + partition_heal(&c.ha); + + /* Manually re-admit peer 2 to standby_endpoints (simulating a + * successful reconnect after the partition heals). */ + pthread_mutex_lock(&c.ha.lock); + zarray_store(c.ha.standby_endpoints, &ep2); + ps_reset_last_push(&c.ha); + pthread_mutex_unlock(&c.ha.lock); + + /* Install a post-heal hook that tallies NOPs to BOTH standbys. + * Peer 1 + the re-admitted peer 2 should both receive heartbeats. */ + atomic_store(&g_survivor_nops, 0); + atomic_store(&g_heal_nops, 0); + ha_nop_req_hook = ps_heal_nop_hook; + ha_heartbeat_req_hook = NULL; + + /* Drive one more tick after heal. */ + pthread_t th; + ps_tick_arg_t arg; + int done = ps_drive_tick_bounded(&c, &th, &arg); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, done, + "R7.3: the post-heal tick must return promptly"); + pthread_join(th, NULL); + + /* Verify BOTH standbys (peer 1 + re-admitted peer 2) receive a heartbeat. + * This proves the previously-partitioned peer REJOINED. */ + int heal_nops = atomic_load(&g_heal_nops); + TEST_ASSERT_EQUAL_INT_MESSAGE(2, heal_nops, + "R7.3: after healing the partition, BOTH standbys (including the " + "previously-partitioned peer 2) must receive a heartbeat — proving " + "the peer rejoins and catches up without operator intervention"); + + ha_nop_req_hook = NULL; + } + + ha_set_reconnect_offload_for_test(1); /* restore production default */ + if (offload_on) { + /* partition_heal already called -> gate already destroyed */ + } else { + /* RED path: gate was released+drained in the per-tick restore above; + * the NOP blackhole hook is still installed -> clean up. */ + ps_gate_init(&g_partition_gate); + partition_heal(&c.ha); + } + ps_teardown_3node(&c, &ep1, &ep2); + printf("%s: PASS\n", label); +} + +static void test_partition_soak_offload_GREEN(void) +{ + ps_run_soak(1, "test_partition_soak_offload_GREEN"); +} + +static void test_partition_soak_offload_RED(void) +{ + ps_run_soak(0, "test_partition_soak_offload_RED"); +} + +/* ── Suite entry point ─────────────────────────────────────────────────────── */ + +void test_ha_partition(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_partition_soak_offload_RED); + RUN_TEST(test_partition_soak_offload_GREEN); + UNITY_END(); +}