forked from lei.chen/myxce
Rongsong Shen
bd02064ebc
push_config_transport_only spawns one thread per standby then joins them with a CONF_PUSH_TIMEOUT_MS bounded wait — blocking the caller. On the requester-less drive-thread paths (promotion + heartbeat convergence) that wait stalls the io_worker. Route those through ha_config_push_offloadable, gated by g_drive_async_configpush_offload (default OFF → inline, byte-for-byte legacy; ON → slow lane, non-blocking). Flag ON deep-copies the config bytes + sha into a lane ctx (caller keeps owning and freeing its own buffer); a single cluster-wide single-flight guard (ha_state_t.configpush_inflight scalar CAS + lane key=1) bounds it to one config-push fan-out in flight; the in-flight flag is cleared on every terminal path (two alloc failures, -EEXIST, hard submit error, lane completion). Send body factored into ha_config_push_do_send. Three requester-less drive-thread call sites converted: raft_cb_become_leader (promotion), the HA_REPL_PUSH_CONFIG heartbeat case, and the direct-CTRL_PRIMARY FSM promotion branch. The admin-upgrade barrier (push_config_to_standbys, has a waiting requester) and join.c are untouched. Lands DORMANT like tasks 3.1/3.2: the lane reads c->ha.standby_endpoints and writes those borrowed standby endpoints, racing io_worker RDHUP teardown if run off-thread (bug-1197 family). Flag OFF keeps push+teardown serialized on the io_worker. Task 3.4 flips this ON together with the snapshot and peerctrl flags after establishing endpoint-lifetime discipline. Tests (controller suite EXIT=0): test_configpush_offload_runs_on_lane, test_configpush_inline_when_flag_off, test_configpush_offload_dedups. Adversarial cpp-review: no CRITICAL/HIGH/MEDIUM findings. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01S6fDCtXkkEgupu7rUah1y2
2222 lines
87 KiB
C
2222 lines
87 KiB
C
/*
|
||
* test_ha_liveness.c — Unit tests for task 2.1: liveness_poll_fn replaces
|
||
* hb_timeout_thread.
|
||
*
|
||
* Requirements: 1.2, 1.3, 1.4, 1.5, 1.6, 2.1, 4.2, 4.3, 5.4
|
||
*
|
||
* TDD approach: tests are written first (RED) to specify the exact behaviour
|
||
* expected from the new implementation. Each test invokes the poll fn
|
||
* directly via the XCE_TESTING test-entry-point exposed in ha.c:
|
||
*
|
||
* void liveness_poll_fn_test_entry(liveness_poll_ctx_t *ctx);
|
||
*
|
||
* This avoids needing a running delaymgr while still exercising the real
|
||
* poll logic.
|
||
*/
|
||
#define _GNU_SOURCE
|
||
#include <stdatomic.h>
|
||
#include <stdint.h>
|
||
#include <stdio.h>
|
||
#include <stdlib.h>
|
||
#include <string.h>
|
||
#include <time.h>
|
||
#include <unistd.h>
|
||
|
||
#include <pthread.h>
|
||
|
||
#include <unity/unity.h>
|
||
#include "ha.h"
|
||
#include "netio.h"
|
||
#include "slowlane.h"
|
||
#include "xce/cluster_def.h"
|
||
|
||
/* ── XCE_TESTING seam ────────────────────────────────────────────────────── */
|
||
|
||
/*
|
||
* liveness_poll_ctx_t: defined in ha.c; exposed here for direct construction
|
||
* in tests. The struct layout must match the one in ha.c exactly.
|
||
*/
|
||
typedef struct {
|
||
ha_state_t *ha;
|
||
cluster_t *c;
|
||
_Atomic int cancelled;
|
||
} liveness_poll_ctx_t;
|
||
|
||
/*
|
||
* liveness_poll_fn_test_entry: XCE_TESTING entry-point declared in ha.c so
|
||
* tests can invoke the poll function directly without a live delaymgr.
|
||
* Only compiled when XCE_TESTING is defined.
|
||
*/
|
||
extern void liveness_poll_fn_test_entry(liveness_poll_ctx_t *ctx);
|
||
|
||
/* ── Helpers ─────────────────────────────────────────────────────────────── */
|
||
|
||
static ha_config_t g_cfg = {
|
||
.heartbeat_interval_ms = 3000,
|
||
.heartbeat_timeout_ms = 6000, /* 2× interval; passes validate_heartbeat_config */
|
||
.election_timeout_ms = 500,
|
||
.probe_connect_timeout_ms = 50,
|
||
.max_promotion_retries = 3,
|
||
.election_jitter_max_ms = 0,
|
||
.election_priority_step_ms = 500,
|
||
};
|
||
|
||
/* Returns current CLOCK_MONOTONIC seconds. */
|
||
static time_t mono_now(void)
|
||
{
|
||
struct timespec ts;
|
||
clock_gettime(CLOCK_MONOTONIC, &ts);
|
||
return ts.tv_sec;
|
||
}
|
||
|
||
/* ── Test 1: poll fn fires failover when elapsed >= heartbeat_timeout_ms ─── */
|
||
|
||
/*
|
||
* Req 1.4, 2.1: set last_activity to a past timestamp beyond the timeout,
|
||
* call the poll fn, verify ha_on_primary_ctrl_fd_rdhup was triggered (role
|
||
* transitions to CTRL_ELECTING).
|
||
*/
|
||
static void test_liveness_poll_fires_failover_when_timeout(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha); /* role = CTRL_STANDBY */
|
||
|
||
/* Set last_activity to > heartbeat_timeout_ms ago */
|
||
time_t past = mono_now() - (time_t)(g_cfg.heartbeat_timeout_ms / 1000) - 10;
|
||
ha.last_activity = past;
|
||
|
||
liveness_poll_ctx_t ctx;
|
||
memset(&ctx, 0, sizeof(ctx));
|
||
ctx.ha = &ha;
|
||
ctx.c = NULL; /* NULL cluster — ha_on_primary_ctrl_fd_rdhup unit-test mode */
|
||
atomic_init(&ctx.cancelled, 0);
|
||
|
||
/* Call poll fn directly — should trigger failover (no re-arm) */
|
||
liveness_poll_fn_test_entry(&ctx);
|
||
|
||
/* Role must have transitioned to CTRL_ELECTING */
|
||
TEST_ASSERT_EQUAL_INT(CTRL_ELECTING, (int)atomic_load(&ha.role));
|
||
|
||
ha_fini(&ha);
|
||
printf("test_liveness_poll_fires_failover_when_timeout: PASS\n");
|
||
}
|
||
|
||
/* ── Test 2: poll fn re-arms when last_activity is fresh ─────────────────── */
|
||
|
||
/*
|
||
* Req 1.2: set last_activity to now (fresh), call poll fn, verify role stays
|
||
* CTRL_STANDBY (no failover) and that the ctx is still valid (not freed).
|
||
*
|
||
* We cannot easily verify that delay_task_ms was called without a running
|
||
* delaymgr, so we verify only the negative: role unchanged and no failover.
|
||
* The liveness_poll_fn_test_entry must NOT call ha_on_primary_ctrl_fd_rdhup.
|
||
*/
|
||
static void test_liveness_poll_rearms_when_fresh(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha);
|
||
|
||
/* last_activity = now → elapsed ≈ 0 < heartbeat_timeout_ms */
|
||
ha.last_activity = mono_now();
|
||
|
||
liveness_poll_ctx_t ctx;
|
||
memset(&ctx, 0, sizeof(ctx));
|
||
ctx.ha = &ha;
|
||
ctx.c = NULL;
|
||
atomic_init(&ctx.cancelled, 0);
|
||
|
||
liveness_poll_fn_test_entry(&ctx);
|
||
|
||
/* Role must still be CTRL_STANDBY — no failover triggered */
|
||
TEST_ASSERT_EQUAL_INT(CTRL_STANDBY, (int)atomic_load(&ha.role));
|
||
|
||
ha_fini(&ha);
|
||
printf("test_liveness_poll_rearms_when_fresh: PASS\n");
|
||
}
|
||
|
||
/* ── Test 3: cancelled poll fn is a no-op ───────────────────────────────── */
|
||
|
||
/*
|
||
* Req 1.5: set cancelled=1 before calling the poll fn; even if last_activity
|
||
* is in the far past the fn must not trigger failover.
|
||
*/
|
||
static void test_liveness_poll_cancelled_is_noop(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha);
|
||
|
||
/* Make it look like a timeout */
|
||
ha.last_activity = mono_now() - 9999;
|
||
|
||
liveness_poll_ctx_t ctx;
|
||
memset(&ctx, 0, sizeof(ctx));
|
||
ctx.ha = &ha;
|
||
ctx.c = NULL;
|
||
atomic_init(&ctx.cancelled, 1); /* cancelled before firing */
|
||
|
||
liveness_poll_fn_test_entry(&ctx);
|
||
|
||
/* Role must still be CTRL_STANDBY — cancelled flag suppresses failover */
|
||
TEST_ASSERT_EQUAL_INT(CTRL_STANDBY, (int)atomic_load(&ha.role));
|
||
|
||
ha_fini(&ha);
|
||
printf("test_liveness_poll_cancelled_is_noop: PASS\n");
|
||
}
|
||
|
||
/* ── Test 4: ha_on_heartbeat_received updates last_activity ─────────────── */
|
||
|
||
/*
|
||
* Req 1.3: ha_on_heartbeat_received must write a non-zero timestamp to
|
||
* ha->last_activity (CLOCK_MONOTONIC seconds).
|
||
*/
|
||
static void test_ha_on_heartbeat_received_updates_last_activity(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha);
|
||
|
||
/* last_activity starts at 0 (set by ha_init → memset + explicit zero) */
|
||
TEST_ASSERT_EQUAL_INT(0, (int)ha.last_activity);
|
||
|
||
ha_on_heartbeat_received(&ha);
|
||
|
||
/* last_activity must now be non-zero (current CLOCK_MONOTONIC seconds) */
|
||
TEST_ASSERT_NOT_EQUAL(0, (int)ha.last_activity);
|
||
|
||
ha_fini(&ha);
|
||
printf("test_ha_on_heartbeat_received_updates_last_activity: PASS\n");
|
||
}
|
||
|
||
/* ── Test 5: poll fn skips failover when role is not CTRL_STANDBY ────────── */
|
||
|
||
/*
|
||
* Req 5.4 (EPOLLRDHUP coexistence): if EPOLLRDHUP fires first and transitions
|
||
* role to CTRL_ELECTING, the poll fn must not trigger a second failover.
|
||
*/
|
||
static void test_liveness_poll_skips_if_not_standby(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
|
||
/* Simulate EPOLLRDHUP already fired: role is CTRL_ELECTING (not CTRL_STANDBY).
|
||
* ha_init sets CTRL_STANDBY; force it to CTRL_ELECTING here. */
|
||
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
|
||
TEST_ASSERT_EQUAL_INT(CTRL_ELECTING, (int)atomic_load(&ha.role));
|
||
|
||
/* last_activity in far past — would normally fire failover on a standby */
|
||
ha.last_activity = mono_now() - 9999;
|
||
|
||
liveness_poll_ctx_t ctx;
|
||
memset(&ctx, 0, sizeof(ctx));
|
||
ctx.ha = &ha;
|
||
ctx.c = NULL;
|
||
atomic_init(&ctx.cancelled, 0);
|
||
|
||
liveness_poll_fn_test_entry(&ctx);
|
||
|
||
/* Role must still be CTRL_ELECTING — poll fn must not call ha_on_primary_ctrl_fd_rdhup
|
||
* when role is already non-STANDBY (prevents double-failover, Req 5.4) */
|
||
TEST_ASSERT_EQUAL_INT(CTRL_ELECTING, (int)atomic_load(&ha.role));
|
||
|
||
ha_fini(&ha);
|
||
printf("test_liveness_poll_skips_if_not_standby: PASS\n");
|
||
}
|
||
|
||
/* ── Test 6: ha_arm_heartbeat_timeout_vtask does not spawn a thread ──────── */
|
||
|
||
/*
|
||
* Req 1.6: the new arm function must not use pthread_create. We verify this
|
||
* indirectly: call the function with a NULL cluster and verify no crash and
|
||
* that ha->last_activity was written (armed).
|
||
*
|
||
* Note: with c==NULL delaymgr is not running, so delay_task_ms will fail
|
||
* silently. The test only checks that no thread was spawned by looking at
|
||
* ha->last_activity being updated (arm path executes) and role unchanged.
|
||
*/
|
||
static void test_ha_arm_no_crash_no_thread(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha);
|
||
|
||
ha.last_activity = 0;
|
||
|
||
/* Should not crash or block */
|
||
ha_arm_heartbeat_timeout_vtask(&ha, NULL);
|
||
|
||
/* last_activity must have been initialised to current time by arm fn */
|
||
TEST_ASSERT_NOT_EQUAL(0, (int)ha.last_activity);
|
||
|
||
/* Clean up the allocated ctx to avoid leak */
|
||
ha_disarm_heartbeat_timeout_vtask(&ha);
|
||
|
||
ha_fini(&ha);
|
||
printf("test_ha_arm_no_crash_no_thread: PASS\n");
|
||
}
|
||
|
||
/* ── Test 7: ha_disarm sets cancelled=1 ──────────────────────────────────── */
|
||
|
||
/*
|
||
* Req 1.5: after disarm, a subsequent poll fn invocation must be a no-op.
|
||
* We arm, immediately disarm, then call the poll fn with the old ctx pointer
|
||
* — but since disarm frees ctx, we construct a fresh ctx with cancelled=1
|
||
* to verify the poll fn's own cancelled check works.
|
||
*
|
||
* The test indirectly verifies that ha_disarm_heartbeat_timeout_vtask sets
|
||
* cancelled to 1 before freeing: after disarm the static s_liveness_ctx is
|
||
* NULL, which means arm set cancelled=1.
|
||
*/
|
||
static void test_ha_disarm_makes_subsequent_poll_noop(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha);
|
||
|
||
ha_arm_heartbeat_timeout_vtask(&ha, NULL);
|
||
ha_disarm_heartbeat_timeout_vtask(&ha);
|
||
|
||
/* After disarm: construct a cancelled ctx manually and verify no failover */
|
||
ha.last_activity = mono_now() - 9999;
|
||
|
||
liveness_poll_ctx_t ctx;
|
||
memset(&ctx, 0, sizeof(ctx));
|
||
ctx.ha = &ha;
|
||
ctx.c = NULL;
|
||
atomic_init(&ctx.cancelled, 1);
|
||
|
||
liveness_poll_fn_test_entry(&ctx);
|
||
|
||
TEST_ASSERT_EQUAL_INT(CTRL_STANDBY, (int)atomic_load(&ha.role));
|
||
|
||
ha_fini(&ha);
|
||
printf("test_ha_disarm_makes_subsequent_poll_noop: PASS\n");
|
||
}
|
||
|
||
/* ── Task 2.2: primary idle-tracking tests ──────────────────────────────── */
|
||
|
||
/*
|
||
* standby_ep_ctx_t: internal ha.c struct; redeclared here for direct
|
||
* construction. Must match the layout in ha.c exactly.
|
||
*/
|
||
typedef struct {
|
||
ha_state_t *ha;
|
||
cluster_t *c;
|
||
void *ctx; /* ha_elect_ctx_t*, always NULL for standby eps */
|
||
int peer_idx;
|
||
time_t last_push_time;
|
||
} standby_ep_ctx_t_test; /* suffix _test to avoid redefinition; ha.c uses standby_ep_ctx_t */
|
||
|
||
/*
|
||
* ha_nop_req_hook: XCE_TESTING hook. When non-NULL, primary_heartbeat_fn
|
||
* calls this INSTEAD OF xce_nop_req so tests can count sends without a
|
||
* live endpoint. Declared in ha.c.
|
||
*/
|
||
extern int (*ha_nop_req_hook)(endpoint_t *ep);
|
||
|
||
/*
|
||
* ha_heartbeat_req_hook: XCE_TESTING hook (task 3.1). When non-NULL,
|
||
* primary_heartbeat_fn calls this with the (term, committed_lsn,
|
||
* committed_config_tag) it stamps onto the heartbeat, so a test can assert the
|
||
* heartbeat carries the primary's OWN current committed state. Declared in
|
||
* ha.h.
|
||
*/
|
||
extern int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term,
|
||
uint64_t committed_lsn,
|
||
uint64_t committed_config_tag);
|
||
|
||
/*
|
||
* primary_heartbeat_fn_for_test: test entry-point that exercises the primary
|
||
* heartbeat idle-tracking logic directly without a live delaymgr.
|
||
* Declared in ha.c under XCE_TESTING.
|
||
*/
|
||
extern int primary_heartbeat_fn_for_test(cluster_t *c);
|
||
|
||
/*
|
||
* standby_ep_ctx_build_for_test: allocate a standby_ep_ctx_t, wire ep->data.
|
||
* Declared in ha.c under XCE_TESTING.
|
||
*/
|
||
extern endpoint_t *standby_ep_ctx_build_for_test(endpoint_t *ep,
|
||
ha_state_t *ha,
|
||
cluster_t *c,
|
||
int peer_idx);
|
||
|
||
/* ── Helpers ─────────────────────────────────────────────────────────────── */
|
||
|
||
static int g_nop_calls = 0;
|
||
|
||
static int stub_nop_req(endpoint_t *ep)
|
||
{
|
||
(void)ep;
|
||
g_nop_calls++;
|
||
return 0;
|
||
}
|
||
|
||
/* Capture for ha_heartbeat_req_hook (task 3.1): the args the primary stamps. */
|
||
static int g_hb_calls = 0;
|
||
static uint64_t g_hb_term = 0;
|
||
static uint64_t g_hb_committed_lsn = 0;
|
||
static uint64_t g_hb_committed_config_tag = 0;
|
||
|
||
static int stub_heartbeat_req(endpoint_t *ep, uint64_t term,
|
||
uint64_t committed_lsn,
|
||
uint64_t committed_config_tag)
|
||
{
|
||
(void)ep;
|
||
g_hb_calls++;
|
||
g_hb_term = term;
|
||
g_hb_committed_lsn = committed_lsn;
|
||
g_hb_committed_config_tag = committed_config_tag;
|
||
return 0;
|
||
}
|
||
|
||
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);
|
||
/* Set role to PRIMARY so primary_heartbeat_fn does not exit early */
|
||
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);
|
||
}
|
||
|
||
/* ── Test 8: primary_heartbeat_fn sends NOP when last_push_time is 0 ────── */
|
||
|
||
/*
|
||
* Req 3.1, 3.2: standby ep with last_push_time=0 (long ago) — heartbeat fn
|
||
* must send NOP and update last_push_time to a recent timestamp.
|
||
*/
|
||
static void test_primary_hb_sends_nop_when_idle(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
init_ha_primary_with_standby_eps(&c.ha);
|
||
|
||
/* Build a mock standby endpoint with last_push_time=0 (idle since epoch) */
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||
sctx->last_push_time = 0; /* long ago — elapsed >> heartbeat_interval_ms */
|
||
|
||
zarray_store(c.ha.standby_endpoints, &ep);
|
||
|
||
/* Install NOP hook */
|
||
g_nop_calls = 0;
|
||
ha_nop_req_hook = stub_nop_req;
|
||
|
||
primary_heartbeat_fn_for_test(&c);
|
||
|
||
ha_nop_req_hook = NULL;
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_nop_calls,
|
||
"Req 3.2: primary_heartbeat_fn must send NOP when last_push_time is stale");
|
||
|
||
/* last_push_time must have been updated to a recent value */
|
||
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)sctx->last_push_time,
|
||
"Req 3.3: last_push_time must be updated after NOP send");
|
||
|
||
free(ep.data);
|
||
fini_ha_primary_with_standby_eps(&c.ha);
|
||
printf("test_primary_hb_sends_nop_when_idle: PASS\n");
|
||
}
|
||
|
||
/* ── Test 9: primary_heartbeat_fn skips NOP when last_push_time is fresh ── */
|
||
|
||
/*
|
||
* Req 3.4: standby ep with last_push_time=now — heartbeat fn must NOT send NOP.
|
||
*/
|
||
static void test_primary_hb_skips_nop_when_fresh(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
init_ha_primary_with_standby_eps(&c.ha);
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||
/* Set last_push_time = now so elapsed ≈ 0 < heartbeat_interval_ms/1000 */
|
||
sctx->last_push_time = mono_now();
|
||
|
||
zarray_store(c.ha.standby_endpoints, &ep);
|
||
|
||
g_nop_calls = 0;
|
||
ha_nop_req_hook = stub_nop_req;
|
||
|
||
primary_heartbeat_fn_for_test(&c);
|
||
|
||
ha_nop_req_hook = NULL;
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_nop_calls,
|
||
"Req 3.4: primary_heartbeat_fn must NOT send NOP when last_push_time is fresh");
|
||
|
||
free(ep.data);
|
||
fini_ha_primary_with_standby_eps(&c.ha);
|
||
printf("test_primary_hb_skips_nop_when_fresh: PASS\n");
|
||
}
|
||
|
||
/* ── Test 9b: primary evicts a standby that has gone silent (dead/restarted) ──
|
||
*
|
||
* Regression for the "kill+restart a controller → it loops elections forever"
|
||
* bug. When a peer is killed/restarted, its outbound standby endpoint on the
|
||
* primary can wedge (epoll re-arm fails) yet stay in standby_endpoints, so the
|
||
* reconnect scan treats the peer as still-connected and never re-establishes a
|
||
* fresh ctrlhello channel — the restarted node never hears from the primary and
|
||
* re-elects every heartbeat_timeout (term climbing unbounded).
|
||
*
|
||
* Fix: primary_heartbeat_fn evicts any standby endpoint that was responsive once
|
||
* (ts_last_in > 0) but has been silent longer than heartbeat_timeout, so the
|
||
* reconnect scan re-establishes the channel. Before the fix the ep persists.
|
||
*/
|
||
static void test_primary_hb_evicts_silent_standby(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
init_ha_primary_with_standby_eps(&c.ha);
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||
sctx->last_push_time = mono_now(); /* recently pushed — not the trigger */
|
||
|
||
/* Responsive once, then silent well beyond heartbeat_timeout (wall clock,
|
||
* matching endpoint_t.ts_last_in which handle_input stamps via time(NULL)). */
|
||
uint32_t timeout_s = g_cfg.heartbeat_timeout_ms / 1000;
|
||
ep.ts_last_in = time(NULL) - (time_t)(timeout_s + 5);
|
||
|
||
zarray_store(c.ha.standby_endpoints, &ep);
|
||
TEST_ASSERT_EQUAL_INT(1, c.ha.standby_endpoints->count);
|
||
|
||
g_nop_calls = 0;
|
||
ha_nop_req_hook = stub_nop_req;
|
||
primary_heartbeat_fn_for_test(&c);
|
||
ha_nop_req_hook = NULL;
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, c.ha.standby_endpoints->count,
|
||
"primary must evict a standby silent beyond heartbeat_timeout so it reconnects");
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_nop_calls,
|
||
"no NOP should be sent to an evicted dead standby");
|
||
|
||
free(ep.data);
|
||
fini_ha_primary_with_standby_eps(&c.ha);
|
||
printf("test_primary_hb_evicts_silent_standby: PASS\n");
|
||
}
|
||
|
||
/* ── Test 9c: a responsive standby is NOT evicted (no false positives) ─────── */
|
||
static void test_primary_hb_keeps_live_standby(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
init_ha_primary_with_standby_eps(&c.ha);
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||
sctx->last_push_time = mono_now();
|
||
|
||
/* Recent inbound activity — within the timeout window. */
|
||
ep.ts_last_in = time(NULL);
|
||
|
||
zarray_store(c.ha.standby_endpoints, &ep);
|
||
|
||
g_nop_calls = 0;
|
||
ha_nop_req_hook = stub_nop_req;
|
||
primary_heartbeat_fn_for_test(&c);
|
||
ha_nop_req_hook = NULL;
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.ha.standby_endpoints->count,
|
||
"a responsive standby (recent ts_last_in) must NOT be evicted");
|
||
|
||
free(ep.data);
|
||
fini_ha_primary_with_standby_eps(&c.ha);
|
||
printf("test_primary_hb_keeps_live_standby: PASS\n");
|
||
}
|
||
|
||
/* ── Test 9d: a never-responded standby is NOT evicted (grace for new joiners) ── */
|
||
static void test_primary_hb_keeps_unconfirmed_standby(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
init_ha_primary_with_standby_eps(&c.ha);
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||
sctx->last_push_time = mono_now();
|
||
|
||
/* Freshly connected, no reply yet: ts_last_in == 0. Must NOT be evicted
|
||
* (eviction keys on ts_last_in > 0 so a just-connected joiner gets grace). */
|
||
ep.ts_last_in = 0;
|
||
|
||
zarray_store(c.ha.standby_endpoints, &ep);
|
||
|
||
g_nop_calls = 0;
|
||
ha_nop_req_hook = stub_nop_req;
|
||
primary_heartbeat_fn_for_test(&c);
|
||
ha_nop_req_hook = NULL;
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.ha.standby_endpoints->count,
|
||
"a just-connected standby (ts_last_in==0) must NOT be evicted");
|
||
|
||
free(ep.data);
|
||
fini_ha_primary_with_standby_eps(&c.ha);
|
||
printf("test_primary_hb_keeps_unconfirmed_standby: PASS\n");
|
||
}
|
||
|
||
/* ── Test 9e: primary stamps its OWN current committed + term on heartbeat ──
|
||
*
|
||
* Task 3.1 / Req 1.1, 3.4: when an idle standby is due a heartbeat, the primary
|
||
* must send xce_ha_heartbeat_req carrying its CURRENT OWN state read atomically:
|
||
* term = ha->term
|
||
* committed_lsn = c->persist.committed_lsn
|
||
* committed_config_tag = ha->committed_config_tag
|
||
* NOT the standby-side last_known_* fields. The ha_heartbeat_req_hook captures
|
||
* those three args; we assert they equal the primary's own values.
|
||
*/
|
||
static void test_primary_hb_stamps_own_committed_and_term(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
init_ha_primary_with_standby_eps(&c.ha);
|
||
|
||
/* Primary's own authoritative state — distinct sentinel values so a wrong
|
||
* source (e.g. last_known_* or a hardcoded 0) is caught. */
|
||
const uint64_t PRIMARY_TERM = 7;
|
||
const uint64_t PRIMARY_COMMITTED_LSN = 42;
|
||
const uint64_t PRIMARY_COMMITTED_TAG = 13;
|
||
atomic_store_explicit(&c.ha.term, PRIMARY_TERM, memory_order_release);
|
||
atomic_store_explicit(&c.persist.committed_lsn, PRIMARY_COMMITTED_LSN,
|
||
memory_order_release);
|
||
atomic_store_explicit(&c.ha.committed_config_tag, PRIMARY_COMMITTED_TAG,
|
||
memory_order_release);
|
||
|
||
/* Poison the standby-side last_known_* fields with DIFFERENT values: a
|
||
* heartbeat that mistakenly read these (the 3.4 bug) would carry them. */
|
||
atomic_store_explicit(&c.ha.last_known_committed_lsn, 999,
|
||
memory_order_release);
|
||
atomic_store_explicit(&c.ha.last_known_committed_config_tag, 888,
|
||
memory_order_release);
|
||
|
||
/* Idle standby (last_push_time long ago) so the heartbeat is due. */
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||
sctx->last_push_time = 0;
|
||
zarray_store(c.ha.standby_endpoints, &ep);
|
||
|
||
g_hb_calls = 0;
|
||
g_hb_term = g_hb_committed_lsn = g_hb_committed_config_tag = 0;
|
||
ha_heartbeat_req_hook = stub_heartbeat_req;
|
||
|
||
primary_heartbeat_fn_for_test(&c);
|
||
|
||
ha_heartbeat_req_hook = NULL;
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_hb_calls,
|
||
"Req 1.1: primary must send a heartbeat (xce_ha_heartbeat_req) to an idle standby");
|
||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(PRIMARY_TERM, g_hb_term,
|
||
"Req 3.4: heartbeat must carry the primary's OWN current term");
|
||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(PRIMARY_COMMITTED_LSN, g_hb_committed_lsn,
|
||
"Req 1.1/3.4: heartbeat must carry the primary's OWN persist.committed_lsn");
|
||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(PRIMARY_COMMITTED_TAG, g_hb_committed_config_tag,
|
||
"Req 1.1/3.4: heartbeat must carry the primary's OWN committed_config_tag");
|
||
|
||
/* last_push_time still updated on success — bookkeeping unchanged. */
|
||
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)sctx->last_push_time,
|
||
"last_push_time must be updated after a successful heartbeat send");
|
||
|
||
free(ep.data);
|
||
fini_ha_primary_with_standby_eps(&c.ha);
|
||
printf("test_primary_hb_stamps_own_committed_and_term: PASS\n");
|
||
}
|
||
|
||
/* ── Test 10: on_standby_connected sets last_push_time ──────────────────── */
|
||
|
||
/*
|
||
* Req 3.3: on_standby_connected must write last_push_time so that the first
|
||
* heartbeat tick after connect does not send a redundant NOP.
|
||
*
|
||
* We use the standby_ep_ctx_build_for_test to construct a context with
|
||
* last_push_time=0, then verify that after a fresh connect (simulated by
|
||
* calling on_standby_connected_for_test), last_push_time is non-zero.
|
||
*/
|
||
extern int on_standby_connected_for_test(endpoint_t *ep);
|
||
|
||
static void test_on_standby_connected_sets_last_push_time(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_state_t *ha = &c.ha;
|
||
ha_init(ha, &g_cfg, 0);
|
||
ha->standby_endpoints = calloc(1, sizeof(zarray_t));
|
||
zarray_init(ha->standby_endpoints, 4, NULL);
|
||
/* Provide minimal myconf so on_standby_connected doesn't deref NULL */
|
||
myconf_t conf;
|
||
memset(&conf, 0, sizeof(conf));
|
||
c.myconf = &conf;
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, ha, &c, 0);
|
||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||
|
||
/* last_push_time is 0 from calloc */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, (int)sctx->last_push_time,
|
||
"last_push_time must start at 0 before on_standby_connected");
|
||
|
||
/* Call on_standby_connected directly via test entry (xce_request will fail
|
||
* on a null ep but last_push_time should be written before it) */
|
||
on_standby_connected_for_test(&ep);
|
||
|
||
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)sctx->last_push_time,
|
||
"Req 3.3: on_standby_connected must write last_push_time != 0");
|
||
|
||
free(ep.data);
|
||
zarray_fini(ha->standby_endpoints);
|
||
free(ha->standby_endpoints);
|
||
ha->standby_endpoints = NULL;
|
||
ha_fini(ha);
|
||
printf("test_on_standby_connected_sets_last_push_time: PASS\n");
|
||
}
|
||
|
||
/* ── Task 3.1: cluster.c dispatch guard tests ───────────────────────────── */
|
||
|
||
/*
|
||
* Simulates the cluster.c OPCode_nop dispatch guard:
|
||
* if role == CTRL_STANDBY → ha_on_heartbeat_received()
|
||
* Tests that the role-based guard (Req 1.3) correctly updates last_activity
|
||
* only when on a standby, and is a no-op when on a primary.
|
||
*/
|
||
|
||
/*
|
||
* Test 11: cluster.c nop dispatch guard — CTRL_STANDBY resets last_activity
|
||
*
|
||
* Req 1.3, 4.1: any primary message on a standby resets the liveness deadline.
|
||
* The cluster.c OPCode_nop handler must call ha_on_heartbeat_received when
|
||
* role == CTRL_STANDBY (not gated on ep->fd == primary_ctrl_fd).
|
||
*/
|
||
static void test_cluster_dispatch_nop_resets_activity_when_standby(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha); /* role = CTRL_STANDBY */
|
||
|
||
ha.last_activity = 0;
|
||
|
||
/* Simulate what cluster.c OPCode_nop handler now does:
|
||
* if (role == CTRL_STANDBY) ha_on_heartbeat_received(&ha); */
|
||
if ((ctrl_role_t)atomic_load_explicit(&ha.role, memory_order_acquire)
|
||
== CTRL_STANDBY)
|
||
ha_on_heartbeat_received(&ha);
|
||
|
||
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)ha.last_activity,
|
||
"Req 1.3: OPCode_nop on CTRL_STANDBY must reset last_activity");
|
||
|
||
ha_fini(&ha);
|
||
printf("test_cluster_dispatch_nop_resets_activity_when_standby: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Test 12: cluster.c nop dispatch guard — CTRL_PRIMARY does NOT reset last_activity
|
||
*
|
||
* Req 1.3: the guard must only call ha_on_heartbeat_received when CTRL_STANDBY.
|
||
* When role is CTRL_PRIMARY the call must be skipped so the primary does not
|
||
* accidentally update its own liveness clock via reflected nops.
|
||
*/
|
||
static void test_cluster_dispatch_nop_skips_when_primary(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 0));
|
||
/* Force role to CTRL_PRIMARY */
|
||
atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY, memory_order_release);
|
||
ha.last_activity = 0;
|
||
|
||
/* Simulate cluster.c OPCode_nop guard: skip when not CTRL_STANDBY */
|
||
if ((ctrl_role_t)atomic_load_explicit(&ha.role, memory_order_acquire)
|
||
== CTRL_STANDBY)
|
||
ha_on_heartbeat_received(&ha);
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, (int)ha.last_activity,
|
||
"Req 1.3: OPCode_nop on CTRL_PRIMARY must NOT update last_activity");
|
||
|
||
ha_fini(&ha);
|
||
printf("test_cluster_dispatch_nop_skips_when_primary: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Test 13: cluster.c ctrlhello dispatch — CTRL_STANDBY resets last_activity (Req 4.1)
|
||
*
|
||
* Req 4.1: ctrlhello counts as primary activity — the OPCode_hello ctrlhello
|
||
* path must call ha_on_heartbeat_received when role == CTRL_STANDBY.
|
||
*/
|
||
static void test_cluster_dispatch_ctrlhello_resets_activity_when_standby(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha); /* role = CTRL_STANDBY */
|
||
|
||
ha.last_activity = 0;
|
||
|
||
/* Simulate cluster.c OPCode_hello ctrlhello path:
|
||
* rc = handle_ctrlhello(c, ep, &request);
|
||
* if (role == CTRL_STANDBY) ha_on_heartbeat_received(&ha); */
|
||
if ((ctrl_role_t)atomic_load_explicit(&ha.role, memory_order_acquire)
|
||
== CTRL_STANDBY)
|
||
ha_on_heartbeat_received(&ha);
|
||
|
||
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)ha.last_activity,
|
||
"Req 4.1: ctrlhello on CTRL_STANDBY must reset last_activity");
|
||
|
||
ha_fini(&ha);
|
||
printf("test_cluster_dispatch_ctrlhello_resets_activity_when_standby: PASS\n");
|
||
}
|
||
|
||
/* ── Test 14: ha_arm_heartbeat_timeout_vtask does not increase thread count ── */
|
||
|
||
/*
|
||
* Task 3.2, Req 1.6: ha_arm_heartbeat_timeout_vtask must NOT spawn a pthread.
|
||
*
|
||
* We read the process thread count from /proc/self/status before and after
|
||
* calling the arm function and assert the count is unchanged. This is a
|
||
* stronger check than test_ha_arm_no_crash_no_thread (Test 6), which only
|
||
* verifies no crash and last_activity written.
|
||
*
|
||
* Note: with c==NULL the delaymgr is not running; delay_task_ms will fail
|
||
* silently (no actual delaymgr task is queued), but the arm path executes and
|
||
* must not create a thread as a side-effect.
|
||
*/
|
||
static int read_proc_thread_count(void)
|
||
{
|
||
FILE *f = fopen("/proc/self/status", "r");
|
||
if (!f) return -1;
|
||
char line[128];
|
||
int count = -1;
|
||
while (fgets(line, sizeof(line), f)) {
|
||
if (strncmp(line, "Threads:", 8) == 0) {
|
||
if (sscanf(line + 8, "%d", &count) != 1)
|
||
count = -1;
|
||
break;
|
||
}
|
||
}
|
||
fclose(f);
|
||
return count;
|
||
}
|
||
|
||
static void test_ha_arm_does_not_increase_thread_count(void)
|
||
{
|
||
ha_state_t ha;
|
||
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
|
||
ha_do_demote(&ha);
|
||
|
||
int threads_before = read_proc_thread_count();
|
||
TEST_ASSERT_NOT_EQUAL_MESSAGE(-1, threads_before,
|
||
"could not read /proc/self/status Threads");
|
||
|
||
/* Call arm — must not spawn a pthread (Req 1.6) */
|
||
ha_arm_heartbeat_timeout_vtask(&ha, NULL);
|
||
|
||
int threads_after = read_proc_thread_count();
|
||
TEST_ASSERT_NOT_EQUAL_MESSAGE(-1, threads_after,
|
||
"could not read /proc/self/status Threads after arm");
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(threads_before, threads_after,
|
||
"Req 1.6: ha_arm_heartbeat_timeout_vtask must not spawn a pthread");
|
||
|
||
/* Disarm to release the ctx allocation */
|
||
ha_disarm_heartbeat_timeout_vtask(&ha);
|
||
|
||
ha_fini(&ha);
|
||
printf("test_ha_arm_does_not_increase_thread_count: PASS\n");
|
||
}
|
||
|
||
/* ── Tests 15–17: raft-drive-async Task 2.1 — reconnect offload ──────────────
|
||
*
|
||
* Acceptance criteria (Req 2.1, 2.2, 2.5; design ReconnectOffload trigger):
|
||
* - the delaymgr heartbeat tick detects a disconnected peer and submits the
|
||
* reconnect to the slow lane, then returns immediately — it NEVER performs
|
||
* the blocking connect inline (flag OFF = old inline path, which stalls);
|
||
* - per-peer dedup: while a reconnect to peer X is in flight, a second tick
|
||
* enqueues no duplicate;
|
||
* - heartbeats to reachable peers keep flowing while one peer is unreachable.
|
||
*
|
||
* Determinism: the blocking connect half is replaced by ha_reconnect_connect_hook,
|
||
* which parks on a test-owned gate to simulate an unreachable peer. The test
|
||
* controls exactly when the "connect" completes, so no real socket/timing is
|
||
* involved. primary_heartbeat_fn_for_test is driven on a worker thread so the
|
||
* test can assert it returns promptly (offload) vs stalls (inline). */
|
||
|
||
extern int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx);
|
||
|
||
/* Gate that the reconnect hook parks on (simulates an unreachable peer). */
|
||
typedef struct {
|
||
pthread_mutex_t mu;
|
||
pthread_cond_t release_cv; /* test broadcasts to release the parked hook */
|
||
pthread_cond_t entered_cv; /* hook broadcasts when it has entered/parked */
|
||
int gate_open;
|
||
_Atomic int entered; /* hook invocations that reached the gate */
|
||
_Atomic int calls; /* total hook invocations (dedup oracle) */
|
||
} rc_gate_t;
|
||
|
||
static rc_gate_t g_rc_gate;
|
||
|
||
static void rc_gate_init(rc_gate_t *g)
|
||
{
|
||
pthread_mutex_init(&g->mu, NULL);
|
||
pthread_cond_init(&g->release_cv, NULL);
|
||
pthread_cond_init(&g->entered_cv, NULL);
|
||
g->gate_open = 0;
|
||
atomic_store(&g->entered, 0);
|
||
atomic_store(&g->calls, 0);
|
||
}
|
||
|
||
static void rc_gate_destroy(rc_gate_t *g)
|
||
{
|
||
pthread_mutex_destroy(&g->mu);
|
||
pthread_cond_destroy(&g->release_cv);
|
||
pthread_cond_destroy(&g->entered_cv);
|
||
}
|
||
|
||
/* The hook stand-in for the blocking connect: count, signal entry, park until
|
||
* the test opens the gate, then return success. Runs on a slow-lane worker
|
||
* (flag ON) or inline on the caller (flag OFF). */
|
||
static int rc_gated_connect_hook(cluster_t *c, int peer_idx)
|
||
{
|
||
(void)c; (void)peer_idx;
|
||
atomic_fetch_add(&g_rc_gate.calls, 1);
|
||
|
||
pthread_mutex_lock(&g_rc_gate.mu);
|
||
atomic_fetch_add(&g_rc_gate.entered, 1);
|
||
pthread_cond_broadcast(&g_rc_gate.entered_cv);
|
||
while (!g_rc_gate.gate_open)
|
||
pthread_cond_wait(&g_rc_gate.release_cv, &g_rc_gate.mu);
|
||
pthread_mutex_unlock(&g_rc_gate.mu);
|
||
return 0; /* simulate a successful connect */
|
||
}
|
||
|
||
static void rc_gate_open(rc_gate_t *g)
|
||
{
|
||
pthread_mutex_lock(&g->mu);
|
||
g->gate_open = 1;
|
||
pthread_cond_broadcast(&g->release_cv);
|
||
pthread_mutex_unlock(&g->mu);
|
||
}
|
||
|
||
/* Wait (bounded) until at least n hook invocations have parked at the gate. */
|
||
static int rc_gate_wait_entered(rc_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 rc = 0;
|
||
while (atomic_load(&g->entered) < n) {
|
||
rc = pthread_cond_timedwait(&g->entered_cv, &g->mu, &deadline);
|
||
if (rc != 0) break;
|
||
}
|
||
int entered = atomic_load(&g->entered);
|
||
pthread_mutex_unlock(&g->mu);
|
||
return entered >= n ? 0 : -1;
|
||
}
|
||
|
||
/* Build a primary cluster_t with: peers_cfg of 3 candidates (self at priority
|
||
* 0), one reachable standby already in standby_endpoints (peer 1), and peer 2
|
||
* absent (so the reconnect scan targets it). The slow lane is initialised. */
|
||
static void rc_setup_primary(cluster_t *c, myconf_t *conf, endpoint_t *standby_ep)
|
||
{
|
||
memset(c, 0, sizeof(*c));
|
||
memset(conf, 0, sizeof(*conf));
|
||
|
||
init_ha_primary_with_standby_eps(&c->ha);
|
||
|
||
/* 3-node cluster; self = priority 0. */
|
||
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;
|
||
|
||
/* Slow lane for the offload path. */
|
||
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c->slowlane, 2, 4));
|
||
|
||
/* Reachable standby for peer 1 (absolute index 1), already connected. */
|
||
memset(standby_ep, 0, sizeof(*standby_ep));
|
||
standby_ep_ctx_build_for_test(standby_ep, &c->ha, c, 1);
|
||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)standby_ep->data;
|
||
sctx->last_push_time = 0; /* stale → heartbeat will be sent */
|
||
zarray_store(c->ha.standby_endpoints, standby_ep);
|
||
}
|
||
|
||
static void rc_teardown_primary(cluster_t *c, endpoint_t *standby_ep)
|
||
{
|
||
slowlane_fini(&c->slowlane);
|
||
free(standby_ep->data);
|
||
fini_ha_primary_with_standby_eps(&c->ha);
|
||
}
|
||
|
||
/* Worker thread that drives one heartbeat tick; sets *done=1 when it returns. */
|
||
typedef struct { cluster_t *c; _Atomic int done; } rc_tick_arg_t;
|
||
static void *rc_tick_thread(void *p)
|
||
{
|
||
rc_tick_arg_t *a = (rc_tick_arg_t *)p;
|
||
primary_heartbeat_fn_for_test(a->c);
|
||
atomic_store(&a->done, 1);
|
||
return NULL;
|
||
}
|
||
|
||
/*
|
||
* Test 15 (Req 2.1, 2.2): with the offload flag ON, a heartbeat tick whose
|
||
* reconnect target is unreachable (the connect hook parks) RETURNS PROMPTLY —
|
||
* the blocking connect runs on the lane, not on the delaymgr thread — and the
|
||
* heartbeat to the reachable standby was still sent.
|
||
*
|
||
* With the flag OFF (RED) the inline connect runs on the tick thread and the
|
||
* tick BLOCKS in the gate, so the same assertion (tick done within the
|
||
* deadline) FAILS. The flag toggle is what flips RED↔GREEN.
|
||
*/
|
||
static void rc_run_offload_tick(int offload_on, const char *label)
|
||
{
|
||
cluster_t c;
|
||
myconf_t conf;
|
||
endpoint_t standby_ep;
|
||
rc_setup_primary(&c, &conf, &standby_ep);
|
||
|
||
rc_gate_init(&g_rc_gate);
|
||
ha_reconnect_connect_hook = rc_gated_connect_hook;
|
||
ha_set_reconnect_offload_for_test(offload_on);
|
||
|
||
g_nop_calls = 0;
|
||
ha_nop_req_hook = stub_nop_req;
|
||
|
||
rc_tick_arg_t arg = { .c = &c };
|
||
atomic_store(&arg.done, 0);
|
||
|
||
pthread_t th;
|
||
pthread_create(&th, NULL, rc_tick_thread, &arg);
|
||
|
||
/* The tick should return within a generous deadline IFF the connect was
|
||
* offloaded. Poll arg.done. */
|
||
int done = 0;
|
||
for (int i = 0; i < 200; i++) { /* up to ~2s */
|
||
if (atomic_load(&arg.done)) { done = 1; break; }
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
|
||
if (offload_on) {
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, done,
|
||
"Req 2.1/2.2: tick must return promptly with reconnect offloaded "
|
||
"(connect parked on the lane, delaymgr thread free)");
|
||
/* The reachable standby still got its heartbeat this tick. */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_nop_calls,
|
||
"Req 2.1: heartbeat to the reachable standby must be sent while a "
|
||
"peer is unreachable");
|
||
/* The reconnect connect was submitted to the lane (hook entered). */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_gate_wait_entered(&g_rc_gate, 1, 1000),
|
||
"Req 2.1: the offloaded reconnect connect must run on the lane");
|
||
} else {
|
||
/* RED: inline connect blocks the tick thread inside the gate. */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, done,
|
||
"RED: with offload OFF the inline connect blocks the delaymgr tick");
|
||
}
|
||
|
||
/* Release the parked hook so the worker/tick thread can finish, then join. */
|
||
rc_gate_open(&g_rc_gate);
|
||
pthread_join(th, NULL);
|
||
|
||
ha_nop_req_hook = NULL;
|
||
ha_reconnect_connect_hook = NULL;
|
||
ha_set_reconnect_offload_for_test(1); /* restore production default */
|
||
rc_gate_destroy(&g_rc_gate);
|
||
rc_teardown_primary(&c, &standby_ep);
|
||
printf("%s: PASS\n", label);
|
||
}
|
||
|
||
static void test_reconnect_offload_tick_returns_promptly(void)
|
||
{
|
||
rc_run_offload_tick(1, "test_reconnect_offload_tick_returns_promptly");
|
||
}
|
||
|
||
static void test_reconnect_inline_tick_blocks_RED(void)
|
||
{
|
||
rc_run_offload_tick(0, "test_reconnect_inline_tick_blocks_RED");
|
||
}
|
||
|
||
/*
|
||
* Test 17 (Req 2.5): per-peer dedup. While a reconnect to the unreachable
|
||
* peer is in flight (parked in the hook), a SECOND heartbeat tick enqueues NO
|
||
* duplicate reconnect for that peer — the connect hook is invoked exactly once
|
||
* across both ticks.
|
||
*/
|
||
static void test_reconnect_offload_dedups_per_peer(void)
|
||
{
|
||
cluster_t c;
|
||
myconf_t conf;
|
||
endpoint_t standby_ep;
|
||
rc_setup_primary(&c, &conf, &standby_ep);
|
||
|
||
rc_gate_init(&g_rc_gate);
|
||
ha_reconnect_connect_hook = rc_gated_connect_hook;
|
||
ha_set_reconnect_offload_for_test(1);
|
||
|
||
ha_nop_req_hook = stub_nop_req;
|
||
|
||
/* Tick 1: submits the reconnect; the lane worker enters the hook and parks. */
|
||
primary_heartbeat_fn_for_test(&c);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_gate_wait_entered(&g_rc_gate, 1, 1000),
|
||
"tick 1 must submit the reconnect (hook parks on the lane)");
|
||
|
||
/* Tick 2: peer still unreachable (in flight) — must NOT enqueue a duplicate. */
|
||
primary_heartbeat_fn_for_test(&c);
|
||
|
||
/* Give any (erroneously) duplicated task time to reach the gate. */
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 150 * 1000000L };
|
||
nanosleep(&ts, NULL);
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_rc_gate.calls),
|
||
"Req 2.5: a second tick must not enqueue a duplicate reconnect for an "
|
||
"already in-flight peer");
|
||
|
||
/* Release and let the single in-flight reconnect complete. */
|
||
rc_gate_open(&g_rc_gate);
|
||
|
||
/* Drain: wait for the in-flight flag to clear (lane worker done). */
|
||
for (int i = 0; i < 200; i++) {
|
||
if (atomic_load_explicit(&c.ha.reconnect_inflight[2],
|
||
memory_order_acquire) == 0) break;
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
|
||
ha_nop_req_hook = NULL;
|
||
ha_reconnect_connect_hook = NULL;
|
||
rc_gate_destroy(&g_rc_gate);
|
||
rc_teardown_primary(&c, &standby_ep);
|
||
printf("test_reconnect_offload_dedups_per_peer: PASS\n");
|
||
}
|
||
|
||
/* ── Tests 20–21: raft-drive-async Task 2.3 — reconnect-offload INTEGRATION ──
|
||
*
|
||
* Acceptance criteria (Req 2.1, 2.2, 2.5; design ReconnectOffload):
|
||
* Tests 15–17 prove the SINGLE-tick unit behaviour (one survivor, ≤2 ticks).
|
||
* Task 2.3 escalates to SUSTAINED liveness: while ONE peer stays unreachable
|
||
* (its reconnect parked in-flight on the lane for the WHOLE window), drive
|
||
* MULTIPLE consecutive heartbeat ticks and assert
|
||
* - every SURVIVING peer keeps receiving heartbeats on EVERY tick
|
||
* (fan-out is observable: a 4-node cluster has TWO survivors), and
|
||
* - each tick returns promptly — the delaymgr is NEVER stalled, and
|
||
* - the reconnect is deduped across the ENTIRE window (the connect hook is
|
||
* submitted exactly ONCE, not once-per-tick — Req 2.5).
|
||
*
|
||
* RED contrast: reverting the offload (flag OFF = inline connect) makes the
|
||
* FIRST tick block in the parked connect, so the tick never returns and
|
||
* survivors stop getting heartbeats. The flag toggle is the RED↔GREEN switch
|
||
* (production already has the offload; this is a test-only task).
|
||
*
|
||
* Determinism: same gate as tests 15–17 — the blocking connect is replaced by
|
||
* rc_gated_connect_hook, which parks on g_rc_gate, so the test controls exactly
|
||
* when the unreachable peer's "connect" completes (no real socket/timing). */
|
||
|
||
#define RC_SUSTAIN_TICKS 3
|
||
|
||
/* Build a 4-node primary (self = priority 0; peers 1,2,3). Peers 1 and 3 are
|
||
* reachable standbys already in standby_endpoints; peer 2 is absent, so the
|
||
* reconnect scan targets ONLY peer 2 (reconnect_inflight[2]). Two survivors
|
||
* make the heartbeat fan-out observable. Modeled on rc_setup_primary (3-node,
|
||
* one survivor). */
|
||
static void rc_setup_primary2(cluster_t *c, myconf_t *conf,
|
||
endpoint_t *standby_ep1, endpoint_t *standby_ep3)
|
||
{
|
||
memset(c, 0, sizeof(*c));
|
||
memset(conf, 0, sizeof(*conf));
|
||
|
||
init_ha_primary_with_standby_eps(&c->ha);
|
||
|
||
/* 4-node cluster; self = priority 0. */
|
||
conf->peers_cfg.candidates_count = 4;
|
||
for (int i = 0; i < 4; 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;
|
||
|
||
/* Slow lane for the offload path. */
|
||
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c->slowlane, 2, 4));
|
||
|
||
/* Reachable standby for peer 1 (absolute index 1), already connected. */
|
||
memset(standby_ep1, 0, sizeof(*standby_ep1));
|
||
standby_ep_ctx_build_for_test(standby_ep1, &c->ha, c, 1);
|
||
((standby_ep_ctx_t_test *)standby_ep1->data)->last_push_time = 0;
|
||
zarray_store(c->ha.standby_endpoints, standby_ep1);
|
||
|
||
/* Reachable standby for peer 3 (absolute index 3), already connected.
|
||
* Peer 2 is intentionally absent → it is the sole reconnect target. */
|
||
memset(standby_ep3, 0, sizeof(*standby_ep3));
|
||
standby_ep_ctx_build_for_test(standby_ep3, &c->ha, c, 3);
|
||
((standby_ep_ctx_t_test *)standby_ep3->data)->last_push_time = 0;
|
||
zarray_store(c->ha.standby_endpoints, standby_ep3);
|
||
}
|
||
|
||
static void rc_teardown_primary2(cluster_t *c, endpoint_t *standby_ep1,
|
||
endpoint_t *standby_ep3)
|
||
{
|
||
slowlane_fini(&c->slowlane);
|
||
free(standby_ep1->data);
|
||
free(standby_ep3->data);
|
||
fini_ha_primary_with_standby_eps(&c->ha);
|
||
}
|
||
|
||
/* Drive one heartbeat tick on a worker thread and return 1 iff it completed
|
||
* within the deadline (delaymgr not stalled), 0 if it is still blocked. */
|
||
static int rc_drive_tick_bounded(cluster_t *c, pthread_t *th_out,
|
||
rc_tick_arg_t *arg)
|
||
{
|
||
arg->c = c;
|
||
atomic_store(&arg->done, 0);
|
||
pthread_create(th_out, NULL, rc_tick_thread, arg);
|
||
for (int i = 0; i < 200; i++) { /* up to ~2s */
|
||
if (atomic_load(&arg->done)) return 1;
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
/*
|
||
* Test 20 / 21 (Req 2.1, 2.2, 2.5): parametrized GREEN/RED runner for the
|
||
* sustained reconnect-offload integration behaviour.
|
||
*/
|
||
static void rc_run_sustained(int offload_on, const char *label)
|
||
{
|
||
cluster_t c;
|
||
myconf_t conf;
|
||
endpoint_t standby_ep1, standby_ep3;
|
||
rc_setup_primary2(&c, &conf, &standby_ep1, &standby_ep3);
|
||
|
||
rc_gate_init(&g_rc_gate);
|
||
ha_reconnect_connect_hook = rc_gated_connect_hook;
|
||
ha_set_reconnect_offload_for_test(offload_on);
|
||
|
||
g_nop_calls = 0;
|
||
ha_nop_req_hook = stub_nop_req;
|
||
|
||
standby_ep_ctx_t_test *s1 = (standby_ep_ctx_t_test *)standby_ep1.data;
|
||
standby_ep_ctx_t_test *s3 = (standby_ep_ctx_t_test *)standby_ep3.data;
|
||
|
||
if (offload_on) {
|
||
/* GREEN: drive N consecutive ticks. Reset each survivor's
|
||
* last_push_time to 0 BEFORE each tick so a heartbeat is DUE this tick
|
||
* (production stamps last_push_time = now after a successful send and
|
||
* only re-sends once elapsed >= interval_s; without the reset, ticks 2+
|
||
* would send nothing). */
|
||
for (int t = 0; t < RC_SUSTAIN_TICKS; t++) {
|
||
s1->last_push_time = 0;
|
||
s3->last_push_time = 0;
|
||
|
||
pthread_t th;
|
||
rc_tick_arg_t arg;
|
||
int done = rc_drive_tick_bounded(&c, &th, &arg);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, done,
|
||
"Req 2.1/2.2: every heartbeat tick must return promptly while a "
|
||
"peer's reconnect is parked in-flight (delaymgr never stalled)");
|
||
pthread_join(th, NULL);
|
||
|
||
/* The unreachable peer's reconnect stays in-flight across the whole
|
||
* window (parked in the gate). */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
|
||
atomic_load_explicit(&c.ha.reconnect_inflight[2],
|
||
memory_order_acquire),
|
||
"Req 2.5: the unreachable peer's reconnect must remain in-flight "
|
||
"(parked) for the whole multi-tick window");
|
||
}
|
||
|
||
/* Every surviving peer (2) got a heartbeat on every tick (N) — survivors
|
||
* kept getting heartbeats across the WHOLE in-flight window (Req 2.1). */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(2 * RC_SUSTAIN_TICKS, g_nop_calls,
|
||
"Req 2.1: every surviving peer must keep receiving heartbeats across "
|
||
"ALL ticks while one peer is unreachable");
|
||
|
||
/* The offloaded reconnect ran on the lane (hook parked) ... */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_gate_wait_entered(&g_rc_gate, 1, 1000),
|
||
"Req 2.1: the offloaded reconnect connect must run on the lane");
|
||
/* ... exactly ONCE across all ticks — deduped, not once-per-tick. */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_rc_gate.calls),
|
||
"Req 2.5: the reconnect to an already-in-flight peer must be deduped "
|
||
"across ALL ticks (submitted exactly once, not once per tick)");
|
||
|
||
/* Release the parked reconnect; drain until the in-flight flag clears. */
|
||
rc_gate_open(&g_rc_gate);
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
|
||
for (int i = 0; i < 200; i++) {
|
||
if (atomic_load_explicit(&c.ha.reconnect_inflight[2],
|
||
memory_order_acquire) == 0) break;
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
} else {
|
||
/* RED: offload OFF — the FIRST tick runs the connect INLINE and blocks
|
||
* in the gate, so the tick never returns and survivors stop getting
|
||
* heartbeats. The flag toggle is what flips RED↔GREEN. */
|
||
s1->last_push_time = 0;
|
||
s3->last_push_time = 0;
|
||
|
||
pthread_t th;
|
||
rc_tick_arg_t arg;
|
||
int done = rc_drive_tick_bounded(&c, &th, &arg);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, done,
|
||
"RED: with offload OFF the inline connect blocks the delaymgr tick — "
|
||
"the tick never returns, so survivors stop getting heartbeats");
|
||
/* The stalled tick cannot sustain heartbeats to survivors across the
|
||
* window: it sent at most this single (blocked) tick's fan-out, far
|
||
* below the GREEN expectation of 2 survivors × N ticks. */
|
||
TEST_ASSERT_TRUE_MESSAGE(g_nop_calls < 2 * RC_SUSTAIN_TICKS,
|
||
"RED: a stalled tick cannot sustain heartbeats to survivors across "
|
||
"the multi-tick window");
|
||
|
||
/* Release the parked inline connect so the blocked tick thread finishes,
|
||
* then join it. */
|
||
rc_gate_open(&g_rc_gate);
|
||
pthread_join(th, NULL);
|
||
}
|
||
|
||
ha_nop_req_hook = NULL;
|
||
ha_reconnect_connect_hook = NULL;
|
||
ha_set_reconnect_offload_for_test(1); /* restore production default */
|
||
rc_gate_destroy(&g_rc_gate);
|
||
rc_teardown_primary2(&c, &standby_ep1, &standby_ep3);
|
||
printf("%s: PASS\n", label);
|
||
}
|
||
|
||
static void test_reconnect_offload_sustains_survivor_heartbeats(void)
|
||
{
|
||
rc_run_sustained(1, "test_reconnect_offload_sustains_survivor_heartbeats");
|
||
}
|
||
|
||
static void test_reconnect_inline_stalls_survivor_heartbeats_RED(void)
|
||
{
|
||
rc_run_sustained(0, "test_reconnect_inline_stalls_survivor_heartbeats_RED");
|
||
}
|
||
|
||
/* ── Tests 18–19: raft-drive-async Task 2.2 — inbound/outbound ownership ─────
|
||
*
|
||
* Acceptance criteria (Req 2.6, 4.4; design ReconnectOffload race resolution,
|
||
* ha-async-io.md connection ownership):
|
||
* - When a reconnect succeeds and registers a NEW outbound control channel
|
||
* for peer X, any PRE-EXISTING standby_endpoints entry for the SAME peer X
|
||
* (a promoted vote-ep or a stale prior channel that has not yet been torn
|
||
* down) must be resolved deterministically: exactly ONE live outbound
|
||
* channel per peer survives — no dual live connection, no orphaned/leaked
|
||
* endpoint, and the survivor is the freshly-connected one so no queued
|
||
* message is delivered down a dead duplicate (lost/reordered, Req 2.6).
|
||
*
|
||
* The collision: BOTH raft_cb_become_leader (promotes vote eps) and
|
||
* on_standby_connected (late-joiner reconnect) zarray_store into
|
||
* standby_endpoints keyed only by ep-POINTER identity — never by peer_idx.
|
||
* So two distinct eps for the same peer can both register → two state-push
|
||
* fan-out targets → exactly the bug-851 dual-channel hazard.
|
||
*
|
||
* Determinism: this is the ownership-DECISION logic (not a real socket-timing
|
||
* race); the real concurrent inbound-accept-vs-outbound-connect timing race is
|
||
* covered by the partition soak (task 4.1). Here we drive the registration
|
||
* point directly with two heap endpoints for the same peer_idx and assert the
|
||
* resolution. Heap eps + endpoint_init make the teardown production-faithful
|
||
* so LeakSanitizer would flag an orphaned/leaked evicted endpoint. */
|
||
|
||
extern int on_standby_connected_for_test(endpoint_t *ep);
|
||
|
||
/* Production teardown of an outbound standby channel (removes from
|
||
* standby_endpoints + repl, frees ep->data); exposed for test teardown. */
|
||
extern int standby_ep_err_handler_primary_for_test(endpoint_t *ep);
|
||
|
||
/* Test-observable counter: how many stale same-peer channels on_standby_connected
|
||
* evicted while resolving ownership (raft-drive-async Task 2.2). */
|
||
extern _Atomic int ha_standby_evicted_count_for_test;
|
||
|
||
/* Allocate + init a heap endpoint wired as a standby channel for peer_idx, the
|
||
* way ha_connect_prep_peer / a promoted vote-ep would leave it (production
|
||
* teardown via standby_ep_err_handler_primary frees ep->data + endpoint, so the
|
||
* evicted one must be reclaimed without a leak). */
|
||
static endpoint_t *own_make_standby_ep(ha_state_t *ha, cluster_t *c, int peer_idx)
|
||
{
|
||
endpoint_t *ep = endpoint_get();
|
||
TEST_ASSERT_NOT_NULL(ep);
|
||
endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
|
||
/* No live iomgr in this unit fixture: leave ep->iomgr NULL so the eviction
|
||
* path skips iomgr_unmonitor (it would epoll_ctl on a bogus fd). The
|
||
* ownership-resolution logic under test (array + replicator) is unaffected. */
|
||
ep->iomgr = NULL;
|
||
standby_ep_ctx_build_for_test(ep, ha, c, peer_idx);
|
||
return ep;
|
||
}
|
||
|
||
static void own_setup_primary(cluster_t *c, myconf_t *conf)
|
||
{
|
||
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 18 (Req 2.6/4.4): registering a NEW outbound channel for a peer that
|
||
* ALREADY has a live standby_endpoints entry (different ep, same peer_idx)
|
||
* resolves to exactly ONE live channel — the new one — and tears the stale one
|
||
* down (no dual connection, no leak).
|
||
*
|
||
* RED (before the resolution lands): on_standby_connected only de-dups by ep
|
||
* pointer, so both eps remain → count == 2 (dual live connection).
|
||
*/
|
||
static void test_reconnect_resolves_dual_channel_same_peer(void)
|
||
{
|
||
cluster_t c;
|
||
myconf_t conf;
|
||
own_setup_primary(&c, &conf);
|
||
|
||
atomic_store(&ha_standby_evicted_count_for_test, 0);
|
||
|
||
/* Stale prior channel for peer 2 already registered (e.g. a promoted
|
||
* vote-ep whose RDHUP has not yet been processed). */
|
||
endpoint_t *stale = own_make_standby_ep(&c.ha, &c, 2);
|
||
zarray_store(c.ha.standby_endpoints, stale);
|
||
TEST_ASSERT_EQUAL_INT(1, c.ha.standby_endpoints->count);
|
||
|
||
/* New reconnect to the SAME peer 2 completes and registers. */
|
||
endpoint_t *fresh = own_make_standby_ep(&c.ha, &c, 2);
|
||
on_standby_connected_for_test(fresh);
|
||
|
||
/* Exactly one live channel for peer 2 — the fresh one. */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.ha.standby_endpoints->count,
|
||
"Req 2.6/4.4: a reconnect for a peer with an existing channel must "
|
||
"leave exactly ONE live outbound channel (no dual connection)");
|
||
TEST_ASSERT_EQUAL_PTR_MESSAGE(fresh, zarray_get(c.ha.standby_endpoints, 0),
|
||
"Req 2.6: the survivor must be the freshly-connected channel so no "
|
||
"queued message is delivered down the dead duplicate");
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
|
||
atomic_load(&ha_standby_evicted_count_for_test),
|
||
"the stale same-peer channel must be evicted exactly once");
|
||
|
||
/* fresh is owned by standby_endpoints; tear it down the production way so
|
||
* LSan sees the heap ep + ctx reclaimed (the err handler frees both). */
|
||
standby_ep_err_handler_primary_for_test(fresh);
|
||
endpoint_fini(fresh);
|
||
free(fresh);
|
||
|
||
fini_ha_primary_with_standby_eps(&c.ha);
|
||
printf("test_reconnect_resolves_dual_channel_same_peer: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Test 19 (Req 2.6): registering a NEW channel for a DIFFERENT peer does NOT
|
||
* evict an existing channel for another peer — resolution is strictly
|
||
* per-peer, so unrelated peers' channels and their queued messages are
|
||
* untouched.
|
||
*/
|
||
static void test_reconnect_preserves_other_peer_channel(void)
|
||
{
|
||
cluster_t c;
|
||
myconf_t conf;
|
||
own_setup_primary(&c, &conf);
|
||
|
||
atomic_store(&ha_standby_evicted_count_for_test, 0);
|
||
|
||
endpoint_t *peer1 = own_make_standby_ep(&c.ha, &c, 1);
|
||
zarray_store(c.ha.standby_endpoints, peer1);
|
||
|
||
endpoint_t *peer2 = own_make_standby_ep(&c.ha, &c, 2);
|
||
on_standby_connected_for_test(peer2);
|
||
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(2, c.ha.standby_endpoints->count,
|
||
"Req 2.6: a reconnect to one peer must not evict another peer's "
|
||
"live channel");
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
|
||
atomic_load(&ha_standby_evicted_count_for_test),
|
||
"no eviction when the peer_idx differs");
|
||
|
||
standby_ep_err_handler_primary_for_test(peer1);
|
||
endpoint_fini(peer1);
|
||
free(peer1);
|
||
standby_ep_err_handler_primary_for_test(peer2);
|
||
endpoint_fini(peer2);
|
||
free(peer2);
|
||
|
||
fini_ha_primary_with_standby_eps(&c.ha);
|
||
printf("test_reconnect_preserves_other_peer_channel: PASS\n");
|
||
}
|
||
|
||
/* ═══════════════════════════════════════════════════════════════════════════
|
||
* raft-drive-async Task 3.1 — PeerCtrlDispatch
|
||
*
|
||
* The registered peer-control handler (peer_ctrl_msg_handler) must, when the
|
||
* dispatch flag is ON, hand the decoded packet to the vtask pool and return
|
||
* PROMPTLY (so peer-control no longer runs on the iomgr io_worker thread), while
|
||
* preserving per-connection arrival order; with the flag OFF it must run the
|
||
* routing body INLINE exactly as before (no behavior change — the production
|
||
* default for this commit). The flag toggle is the RED↔GREEN driver.
|
||
*
|
||
* Test 18 (Req 1.4, 4.5, 8.1, 8.2): flag ON → handler returns promptly and
|
||
* the routing body runs on a POOL worker (a different thread).
|
||
* Test 19 (Req 6.x no-regression): flag OFF → body runs INLINE, synchronously,
|
||
* on the calling thread (identical to the pre-task behavior).
|
||
* Test 20 (Req 4.1, 4.2): two messages on ONE endpoint are processed in
|
||
* arrival order — the second cannot run until the first completes
|
||
* (per-endpoint endpoint_block_tasks serialization).
|
||
* ═══════════════════════════════════════════════════════════════════════════ */
|
||
|
||
extern int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt);
|
||
extern io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted,
|
||
uint64_t peer_tag,
|
||
uint64_t peer_lsn);
|
||
|
||
/* Probe observation state (set via ha_peerctrl_body_probe). */
|
||
static _Atomic int g_pc_runs = 0; /* body invocation count */
|
||
/* _Atomic: written on the pool worker, read on main — avoid a formal C11 data
|
||
* race (raft-drive-async review finding-2). relaxed ordering is sufficient:
|
||
* g_pc_runs (acquire/release via fetch_add/load) already orders visibility. */
|
||
static _Atomic pthread_t g_pc_last_thread; /* thread the body ran on */
|
||
static _Atomic int g_pc_gate_on = 0; /* 1 → body parks at the gate */
|
||
static pthread_mutex_t g_pc_mu = PTHREAD_MUTEX_INITIALIZER;
|
||
static pthread_cond_t g_pc_entered_cv = PTHREAD_COND_INITIALIZER;
|
||
static pthread_cond_t g_pc_release_cv = PTHREAD_COND_INITIALIZER;
|
||
static int g_pc_parked = 0; /* a body is parked at the gate */
|
||
static int g_pc_release = 0; /* gate open → bodies may proceed */
|
||
|
||
static void pc_probe(endpoint_t *ep)
|
||
{
|
||
(void)ep;
|
||
atomic_store_explicit(&g_pc_last_thread, pthread_self(), memory_order_relaxed);
|
||
atomic_fetch_add(&g_pc_runs, 1);
|
||
|
||
if (atomic_load(&g_pc_gate_on)) {
|
||
pthread_mutex_lock(&g_pc_mu);
|
||
g_pc_parked = 1;
|
||
pthread_cond_broadcast(&g_pc_entered_cv);
|
||
while (!g_pc_release)
|
||
pthread_cond_wait(&g_pc_release_cv, &g_pc_mu);
|
||
g_pc_parked = 0;
|
||
pthread_mutex_unlock(&g_pc_mu);
|
||
}
|
||
}
|
||
|
||
/* Wait (bounded) until a body has parked at the gate. */
|
||
static int pc_wait_parked(int timeout_ms)
|
||
{
|
||
struct timespec d;
|
||
clock_gettime(CLOCK_REALTIME, &d);
|
||
d.tv_sec += timeout_ms / 1000;
|
||
d.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
|
||
if (d.tv_nsec >= 1000000000L) { d.tv_sec++; d.tv_nsec -= 1000000000L; }
|
||
pthread_mutex_lock(&g_pc_mu);
|
||
int rc = 0;
|
||
while (!g_pc_parked) {
|
||
rc = pthread_cond_timedwait(&g_pc_entered_cv, &g_pc_mu, &d);
|
||
if (rc != 0) break;
|
||
}
|
||
int parked = g_pc_parked;
|
||
pthread_mutex_unlock(&g_pc_mu);
|
||
return parked ? 0 : -1;
|
||
}
|
||
|
||
static void pc_release_gate(void)
|
||
{
|
||
pthread_mutex_lock(&g_pc_mu);
|
||
g_pc_release = 1;
|
||
pthread_cond_broadcast(&g_pc_release_cv);
|
||
pthread_mutex_unlock(&g_pc_mu);
|
||
}
|
||
|
||
/* Bounded poll until g_pc_runs >= want; returns the observed count. */
|
||
static int pc_wait_runs(int want, int timeout_ms)
|
||
{
|
||
for (int i = 0; i < timeout_ms; i++) {
|
||
if (atomic_load(&g_pc_runs) >= want) break;
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
return atomic_load(&g_pc_runs);
|
||
}
|
||
|
||
static void pc_reset_probe_state(void)
|
||
{
|
||
atomic_store(&g_pc_runs, 0);
|
||
atomic_store(&g_pc_gate_on, 0);
|
||
pthread_mutex_lock(&g_pc_mu);
|
||
g_pc_parked = 0;
|
||
g_pc_release = 0;
|
||
pthread_mutex_unlock(&g_pc_mu);
|
||
}
|
||
|
||
/*
|
||
* Test 18 (Req 1.4, 4.5, 8.1, 8.2): with the dispatch flag ON, the registered
|
||
* handler enqueues the decoded packet onto the vtask pool and RETURNS PROMPTLY;
|
||
* the routing body then runs on a POOL worker — a DIFFERENT thread from the
|
||
* caller (proving peer-control no longer runs on the iomgr io_worker thread).
|
||
* The body (role=CTRL_INIT → bounded default-drop path) frees the packet.
|
||
*/
|
||
static void test_peerctrl_dispatch_runs_on_pool(void)
|
||
{
|
||
vtaskmgr_t mgr;
|
||
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 4));
|
||
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&mgr));
|
||
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
c.ha.pool = &mgr;
|
||
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release);
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
|
||
pc_reset_probe_state();
|
||
ha_peerctrl_body_probe = pc_probe;
|
||
ha_set_peerctrl_dispatch_for_test(1);
|
||
|
||
pthread_t caller = pthread_self();
|
||
io_packet_t *pkt = vote_reply_build_pkt_for_test(1, 1, 0, 0);
|
||
TEST_ASSERT_NOT_NULL(pkt);
|
||
|
||
int rc = peer_ctrl_msg_handler_for_test(&ep, pkt);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
|
||
"Req 8.1/8.2: dispatch handler must return promptly (0) after enqueue");
|
||
|
||
int runs = pc_wait_runs(1, 2000);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
|
||
"Req 1.4/4.5: the routing body must run on the pool exactly once");
|
||
TEST_ASSERT_FALSE_MESSAGE(
|
||
pthread_equal(atomic_load_explicit(&g_pc_last_thread, memory_order_relaxed),
|
||
caller),
|
||
"Req 1.4: the body must run on a POOL worker, NOT the caller's "
|
||
"(io_worker) thread");
|
||
|
||
ha_set_peerctrl_dispatch_for_test(0); /* restore production default (OFF) */
|
||
ha_peerctrl_body_probe = NULL;
|
||
vtaskmgr_fini(&mgr);
|
||
free(ep.data);
|
||
ha_fini(&c.ha);
|
||
printf("test_peerctrl_dispatch_runs_on_pool: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Test 19 (Req 6.x no-regression): with the dispatch flag OFF (the production
|
||
* default), the handler runs the routing body INLINE on the calling thread,
|
||
* synchronously — byte-for-byte the pre-task behavior. This is the RED side of
|
||
* the flag toggle: the body did NOT move to a pool thread.
|
||
*/
|
||
static void test_peerctrl_inline_when_flag_off(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
c.ha.pool = NULL; /* OFF path must not need the pool */
|
||
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release);
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
|
||
pc_reset_probe_state();
|
||
ha_peerctrl_body_probe = pc_probe;
|
||
ha_set_peerctrl_dispatch_for_test(0); /* explicit OFF */
|
||
|
||
pthread_t caller = pthread_self();
|
||
io_packet_t *pkt = vote_reply_build_pkt_for_test(1, 1, 0, 0);
|
||
TEST_ASSERT_NOT_NULL(pkt);
|
||
|
||
int rc = peer_ctrl_msg_handler_for_test(&ep, pkt);
|
||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||
|
||
/* Synchronous: the body already ran on THIS thread before the handler
|
||
* returned — no pool, no wait. */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_pc_runs),
|
||
"Req 6.x: with the flag OFF the body must run inline (synchronously)");
|
||
TEST_ASSERT_TRUE_MESSAGE(
|
||
pthread_equal(atomic_load_explicit(&g_pc_last_thread, memory_order_relaxed),
|
||
caller),
|
||
"Req 6.x: with the flag OFF the body must run on the caller's thread "
|
||
"(inline), unchanged from before the task");
|
||
|
||
ha_peerctrl_body_probe = NULL;
|
||
free(ep.data);
|
||
ha_fini(&c.ha);
|
||
printf("test_peerctrl_inline_when_flag_off: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Test 20 (Req 4.1, 4.2): two peer-control messages arriving on the SAME
|
||
* endpoint are processed in arrival order — the second body cannot start until
|
||
* the first completes. Proof: with the flag ON, the first body parks at the
|
||
* gate (holding its pool worker; the endpoint is blocked by bind_vtask's
|
||
* endpoint_block_tasks). While it is parked, dispatching a second message on
|
||
* the same endpoint does NOT run its body (run count stays 1). Releasing the
|
||
* first lets it complete (endpoint_unblock_tasks) and only THEN does the second
|
||
* run (count → 2).
|
||
*/
|
||
static void test_peerctrl_preserves_per_endpoint_order(void)
|
||
{
|
||
vtaskmgr_t mgr;
|
||
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 4));
|
||
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&mgr));
|
||
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
c.ha.pool = &mgr;
|
||
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release);
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||
|
||
pc_reset_probe_state();
|
||
atomic_store(&g_pc_gate_on, 1); /* bodies park at the gate */
|
||
ha_peerctrl_body_probe = pc_probe;
|
||
ha_set_peerctrl_dispatch_for_test(1);
|
||
|
||
/* Message 1 — dispatched; its body parks at the gate. */
|
||
io_packet_t *pkt1 = vote_reply_build_pkt_for_test(1, 1, 0, 0);
|
||
TEST_ASSERT_NOT_NULL(pkt1);
|
||
TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep, pkt1));
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, pc_wait_parked(2000),
|
||
"the first body must reach the gate (running on the pool)");
|
||
|
||
/* Message 2 — dispatched on the SAME endpoint while body 1 is parked. */
|
||
io_packet_t *pkt2 = vote_reply_build_pkt_for_test(1, 1, 0, 0);
|
||
TEST_ASSERT_NOT_NULL(pkt2);
|
||
TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep, pkt2));
|
||
|
||
/* Give a generous window: body 2 must NOT run while body 1 holds the
|
||
* endpoint (per-connection serialization). */
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 200 * 1000000L }; /* 200 ms */
|
||
nanosleep(&ts, NULL);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_pc_runs),
|
||
"Req 4.1/4.2: the second message must NOT be processed while the first "
|
||
"is in flight on the same connection");
|
||
|
||
/* Release body 1; body 2 then becomes eligible and runs. */
|
||
pc_release_gate();
|
||
int runs = pc_wait_runs(2, 2000);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(2, runs,
|
||
"Req 4.1/4.2: the second message runs only AFTER the first completes "
|
||
"(arrival order preserved)");
|
||
|
||
ha_set_peerctrl_dispatch_for_test(0); /* restore production default (OFF) */
|
||
ha_peerctrl_body_probe = NULL;
|
||
vtaskmgr_fini(&mgr);
|
||
free(ep.data);
|
||
ha_fini(&c.ha);
|
||
printf("test_peerctrl_preserves_per_endpoint_order: PASS\n");
|
||
}
|
||
|
||
/* ───────────────────────────────────────────────────────────────────────
|
||
* Task 3.2 — SnapshotSend slow-lane offload (Req 1.1, 1.5; bug-1198).
|
||
*
|
||
* Probe: a snapshot-send hook (ha_repl_send_snapshot_hook) that records the
|
||
* thread it ran on and supports an optional gate so a test can hold the lane
|
||
* worker in flight while it submits a second snapshot for the same peer
|
||
* (per-peer dedup proof, Req 1.5). Mirrors the peerctrl pc_probe gate.
|
||
* ─────────────────────────────────────────────────────────────────────── */
|
||
extern int (*ha_repl_send_snapshot_hook)(int peer_idx);
|
||
|
||
static _Atomic int g_snap_runs = 0;
|
||
static _Atomic pthread_t g_snap_last_thread;
|
||
static _Atomic int g_snap_gate_on = 0;
|
||
static pthread_mutex_t g_snap_mu = PTHREAD_MUTEX_INITIALIZER;
|
||
static pthread_cond_t g_snap_parked_cv = PTHREAD_COND_INITIALIZER;
|
||
static pthread_cond_t g_snap_release_cv = PTHREAD_COND_INITIALIZER;
|
||
static int g_snap_parked = 0;
|
||
static int g_snap_release = 0;
|
||
|
||
static int snap_probe(int peer_idx)
|
||
{
|
||
(void)peer_idx;
|
||
atomic_store_explicit(&g_snap_last_thread, pthread_self(),
|
||
memory_order_relaxed);
|
||
if (atomic_load(&g_snap_gate_on)) {
|
||
pthread_mutex_lock(&g_snap_mu);
|
||
g_snap_parked = 1;
|
||
pthread_cond_broadcast(&g_snap_parked_cv);
|
||
while (!g_snap_release)
|
||
pthread_cond_wait(&g_snap_release_cv, &g_snap_mu);
|
||
pthread_mutex_unlock(&g_snap_mu);
|
||
}
|
||
atomic_fetch_add(&g_snap_runs, 1);
|
||
return 0;
|
||
}
|
||
|
||
/* Bounded poll until the gated body parks; 0 on park, -1 on timeout. */
|
||
static int snap_wait_parked(int timeout_ms)
|
||
{
|
||
pthread_mutex_lock(&g_snap_mu);
|
||
int parked = 0;
|
||
for (int i = 0; i < timeout_ms; i++) {
|
||
if (g_snap_parked) { parked = 1; break; }
|
||
pthread_mutex_unlock(&g_snap_mu);
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
|
||
nanosleep(&ts, NULL);
|
||
pthread_mutex_lock(&g_snap_mu);
|
||
}
|
||
pthread_mutex_unlock(&g_snap_mu);
|
||
return parked ? 0 : -1;
|
||
}
|
||
|
||
static int snap_wait_runs(int want, int timeout_ms)
|
||
{
|
||
for (int i = 0; i < timeout_ms; i++) {
|
||
if (atomic_load(&g_snap_runs) >= want) break;
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
return atomic_load(&g_snap_runs);
|
||
}
|
||
|
||
static void snap_release_gate(void)
|
||
{
|
||
pthread_mutex_lock(&g_snap_mu);
|
||
g_snap_release = 1;
|
||
pthread_cond_broadcast(&g_snap_release_cv);
|
||
pthread_mutex_unlock(&g_snap_mu);
|
||
}
|
||
|
||
static void snap_reset_probe(void)
|
||
{
|
||
atomic_store(&g_snap_runs, 0);
|
||
atomic_store(&g_snap_gate_on, 0);
|
||
pthread_mutex_lock(&g_snap_mu);
|
||
g_snap_parked = 0;
|
||
g_snap_release = 0;
|
||
pthread_mutex_unlock(&g_snap_mu);
|
||
}
|
||
|
||
/* Drive one SEND_SNAPSHOT decision: standby tag current but lsn behind the
|
||
* oldest WAL window → ha_evaluate_repl_action yields HA_REPL_SEND_SNAPSHOT. */
|
||
static int snap_dispatch_once(cluster_t *c, endpoint_t *ep, int peer_idx)
|
||
{
|
||
c->ha.standby_reported_tag[peer_idx] = 5;
|
||
c->ha.standby_reported_lsn[peer_idx] = 10; /* < oldest_wal_lsn below */
|
||
return ha_dispatch_repl_action_for_standby(
|
||
&c->ha, c, ep, peer_idx,
|
||
/*committed_tag=*/5, /*committed_lsn=*/100, /*oldest_wal_lsn=*/50);
|
||
}
|
||
|
||
/*
|
||
* Task 3.2 GREEN (Req 1.1, 8.1/8.2): with the offload flag ON, SEND_SNAPSHOT
|
||
* is submitted to the slow lane — the send body runs on a LANE worker (a
|
||
* DIFFERENT thread from the caller) and the dispatch returns promptly. This
|
||
* proves the snapshot transfer no longer blocks the io_worker/drive thread.
|
||
*
|
||
* RED before Task 3.2 lands: the snapshot ran inline on the caller thread, so
|
||
* the different-thread assertion fails.
|
||
*/
|
||
static void test_snapshot_offload_runs_on_lane(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4));
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
|
||
snap_reset_probe();
|
||
ha_repl_send_snapshot_hook = snap_probe;
|
||
ha_set_snapshot_offload_for_test(1);
|
||
|
||
pthread_t caller = pthread_self();
|
||
int rc = snap_dispatch_once(&c, &ep, 0);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
|
||
"Req 8.1/8.2: dispatch must return promptly (0) after lane submit");
|
||
|
||
int runs = snap_wait_runs(1, 2000);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
|
||
"Req 1.1: the snapshot send body must run on the lane exactly once");
|
||
TEST_ASSERT_FALSE_MESSAGE(
|
||
pthread_equal(atomic_load_explicit(&g_snap_last_thread,
|
||
memory_order_relaxed), caller),
|
||
"Req 1.1: the snapshot send must run on a LANE worker, NOT the "
|
||
"caller's (io_worker/drive) thread");
|
||
|
||
ha_set_snapshot_offload_for_test(0); /* restore production default (OFF) */
|
||
ha_repl_send_snapshot_hook = NULL;
|
||
slowlane_fini(&c.slowlane);
|
||
ha_fini(&c.ha);
|
||
printf("test_snapshot_offload_runs_on_lane: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Task 3.2 RED side / no-regression (Req 6.x): with the offload flag OFF (the
|
||
* production default until Task 3.4 resolves the borrowed-endpoint lifetime),
|
||
* SEND_SNAPSHOT runs INLINE on the calling thread — byte-for-byte the
|
||
* pre-task behavior.
|
||
*/
|
||
static void test_snapshot_inline_when_flag_off(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
|
||
snap_reset_probe();
|
||
ha_repl_send_snapshot_hook = snap_probe;
|
||
ha_set_snapshot_offload_for_test(0); /* explicit OFF */
|
||
|
||
pthread_t caller = pthread_self();
|
||
int rc = snap_dispatch_once(&c, &ep, 0);
|
||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||
|
||
/* Synchronous: the body already ran on THIS thread before returning. */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_snap_runs),
|
||
"Req 6.x: with the flag OFF the snapshot send must run inline");
|
||
TEST_ASSERT_TRUE_MESSAGE(
|
||
pthread_equal(atomic_load_explicit(&g_snap_last_thread,
|
||
memory_order_relaxed), caller),
|
||
"Req 6.x: with the flag OFF the send must run on the caller's thread");
|
||
|
||
ha_repl_send_snapshot_hook = NULL;
|
||
ha_fini(&c.ha);
|
||
printf("test_snapshot_inline_when_flag_off: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Task 3.2 (Req 1.5): per-peer single-flight. While a snapshot to peer 0 is
|
||
* in flight on the lane (parked at the gate), a second SEND_SNAPSHOT decision
|
||
* for the SAME peer enqueues nothing — the CAS on snapshot_inflight[0] fails.
|
||
* Proof: with the first body parked, the run count stays at 0; after release,
|
||
* exactly ONE body has run (the duplicate was deduped, not queued).
|
||
*/
|
||
static void test_snapshot_offload_dedups_per_peer(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4));
|
||
|
||
endpoint_t ep;
|
||
memset(&ep, 0, sizeof(ep));
|
||
|
||
snap_reset_probe();
|
||
atomic_store(&g_snap_gate_on, 1); /* lane body parks at the gate */
|
||
ha_repl_send_snapshot_hook = snap_probe;
|
||
ha_set_snapshot_offload_for_test(1);
|
||
|
||
/* First snapshot — submitted; its body parks in the lane. */
|
||
TEST_ASSERT_EQUAL_INT(0, snap_dispatch_once(&c, &ep, 0));
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, snap_wait_parked(2000),
|
||
"the first snapshot body must reach the gate (running on the lane)");
|
||
|
||
/* Second snapshot for the SAME peer while the first is in flight. */
|
||
TEST_ASSERT_EQUAL_INT(0, snap_dispatch_once(&c, &ep, 0));
|
||
|
||
/* Give a generous window: the duplicate must NOT run. */
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 200 * 1000000L }; /* 200 ms */
|
||
nanosleep(&ts, NULL);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_snap_runs),
|
||
"Req 1.5: the duplicate snapshot must be deduped while one is in flight");
|
||
|
||
/* Release; exactly the first body completes. */
|
||
snap_release_gate();
|
||
int runs = snap_wait_runs(1, 2000);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
|
||
"Req 1.5: exactly one snapshot ran for the peer (duplicate deduped)");
|
||
/* And no second body sneaks in after release. */
|
||
nanosleep(&ts, NULL);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_snap_runs),
|
||
"Req 1.5: the deduped duplicate was dropped, not queued behind the first");
|
||
|
||
ha_set_snapshot_offload_for_test(0); /* restore production default (OFF) */
|
||
ha_repl_send_snapshot_hook = NULL;
|
||
slowlane_fini(&c.slowlane);
|
||
ha_fini(&c.ha);
|
||
printf("test_snapshot_offload_dedups_per_peer: PASS\n");
|
||
}
|
||
|
||
/* ───────────────────────────────────────────────────────────────────────
|
||
* Task 3.3 — requester-less ConfigPush slow-lane offload (Req 1.1, 1.5; bug-1198).
|
||
*
|
||
* Probe: ha_configpush_probe records the thread the push body ran on and
|
||
* supports an optional gate so a test can hold the lane worker in flight while
|
||
* it submits a second config push (cluster-wide dedup proof, Req 1.5). The two
|
||
* production call sites are #ifndef XCE_TESTING, so the tests drive the router
|
||
* directly via ha_config_push_offloadable_for_test(). Mirrors the snapshot
|
||
* offload probe/gate above.
|
||
* ─────────────────────────────────────────────────────────────────────── */
|
||
extern void (*ha_configpush_probe)(void);
|
||
extern int ha_config_push_offloadable_for_test(cluster_t *c, const uint8_t *bytes,
|
||
size_t blen, const uint8_t sha256[32],
|
||
uint64_t new_tag);
|
||
|
||
static _Atomic int g_cp_runs = 0;
|
||
static _Atomic pthread_t g_cp_last_thread;
|
||
static _Atomic int g_cp_gate_on = 0;
|
||
static pthread_mutex_t g_cp_mu = PTHREAD_MUTEX_INITIALIZER;
|
||
static pthread_cond_t g_cp_release_cv = PTHREAD_COND_INITIALIZER;
|
||
static int g_cp_parked = 0;
|
||
static int g_cp_release = 0;
|
||
|
||
static void cp_probe(void)
|
||
{
|
||
atomic_store_explicit(&g_cp_last_thread, pthread_self(),
|
||
memory_order_relaxed);
|
||
if (atomic_load(&g_cp_gate_on)) {
|
||
pthread_mutex_lock(&g_cp_mu);
|
||
g_cp_parked = 1;
|
||
while (!g_cp_release)
|
||
pthread_cond_wait(&g_cp_release_cv, &g_cp_mu);
|
||
pthread_mutex_unlock(&g_cp_mu);
|
||
}
|
||
atomic_fetch_add(&g_cp_runs, 1);
|
||
}
|
||
|
||
/* Bounded poll until the gated body parks; 0 on park, -1 on timeout. */
|
||
static int cp_wait_parked(int timeout_ms)
|
||
{
|
||
pthread_mutex_lock(&g_cp_mu);
|
||
int parked = 0;
|
||
for (int i = 0; i < timeout_ms; i++) {
|
||
if (g_cp_parked) { parked = 1; break; }
|
||
pthread_mutex_unlock(&g_cp_mu);
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
|
||
nanosleep(&ts, NULL);
|
||
pthread_mutex_lock(&g_cp_mu);
|
||
}
|
||
pthread_mutex_unlock(&g_cp_mu);
|
||
return parked ? 0 : -1;
|
||
}
|
||
|
||
static int cp_wait_runs(int want, int timeout_ms)
|
||
{
|
||
for (int i = 0; i < timeout_ms; i++) {
|
||
if (atomic_load(&g_cp_runs) >= want) break;
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
return atomic_load(&g_cp_runs);
|
||
}
|
||
|
||
static void cp_release_gate(void)
|
||
{
|
||
pthread_mutex_lock(&g_cp_mu);
|
||
g_cp_release = 1;
|
||
pthread_cond_broadcast(&g_cp_release_cv);
|
||
pthread_mutex_unlock(&g_cp_mu);
|
||
}
|
||
|
||
static void cp_reset_probe(void)
|
||
{
|
||
atomic_store(&g_cp_runs, 0);
|
||
atomic_store(&g_cp_gate_on, 0);
|
||
pthread_mutex_lock(&g_cp_mu);
|
||
g_cp_parked = 0;
|
||
g_cp_release = 0;
|
||
pthread_mutex_unlock(&g_cp_mu);
|
||
}
|
||
|
||
/*
|
||
* Task 3.3 GREEN (Req 1.1, 8.1/8.2): with the offload flag ON, a requester-less
|
||
* config push is submitted to the slow lane — the push body runs on a LANE
|
||
* worker (a DIFFERENT thread from the caller) and the dispatch returns promptly.
|
||
* This proves the config push no longer blocks the io_worker/drive thread.
|
||
*
|
||
* RED before Task 3.3 lands: the push ran inline on the caller thread, so the
|
||
* different-thread assertion fails.
|
||
*/
|
||
static void test_configpush_offload_runs_on_lane(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4));
|
||
|
||
cp_reset_probe();
|
||
ha_configpush_probe = cp_probe;
|
||
ha_set_configpush_offload_for_test(1);
|
||
|
||
pthread_t caller = pthread_self();
|
||
int rc = ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
|
||
"Req 8.1/8.2: dispatch must return promptly (0) after lane submit");
|
||
|
||
int runs = cp_wait_runs(1, 2000);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
|
||
"Req 1.1: the config push body must run on the lane exactly once");
|
||
TEST_ASSERT_FALSE_MESSAGE(
|
||
pthread_equal(atomic_load_explicit(&g_cp_last_thread,
|
||
memory_order_relaxed), caller),
|
||
"Req 1.1: the config push must run on a LANE worker, NOT the "
|
||
"caller's (io_worker/drive) thread");
|
||
|
||
ha_set_configpush_offload_for_test(0); /* restore production default (OFF) */
|
||
ha_configpush_probe = NULL;
|
||
slowlane_fini(&c.slowlane);
|
||
ha_fini(&c.ha);
|
||
printf("test_configpush_offload_runs_on_lane: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Task 3.3 no-regression (Req 6.x): with the offload flag OFF (the production
|
||
* default until Task 3.4 resolves the borrowed-endpoint lifetime), the config
|
||
* push runs INLINE on the calling thread — byte-for-byte the pre-task behavior.
|
||
*/
|
||
static void test_configpush_inline_when_flag_off(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
|
||
cp_reset_probe();
|
||
ha_configpush_probe = cp_probe;
|
||
ha_set_configpush_offload_for_test(0); /* explicit OFF */
|
||
|
||
pthread_t caller = pthread_self();
|
||
int rc = ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7);
|
||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||
|
||
/* Synchronous: the body already ran on THIS thread before returning. */
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_cp_runs),
|
||
"Req 6.x: with the flag OFF the config push must run inline");
|
||
TEST_ASSERT_TRUE_MESSAGE(
|
||
pthread_equal(atomic_load_explicit(&g_cp_last_thread,
|
||
memory_order_relaxed), caller),
|
||
"Req 6.x: with the flag OFF the push must run on the caller's thread");
|
||
|
||
ha_configpush_probe = NULL;
|
||
ha_fini(&c.ha);
|
||
printf("test_configpush_inline_when_flag_off: PASS\n");
|
||
}
|
||
|
||
/*
|
||
* Task 3.3 (Req 1.5): cluster-wide single-flight. While one config push is in
|
||
* flight on the lane (parked at the gate), a second push enqueues nothing — the
|
||
* CAS on configpush_inflight fails. Proof: with the first body parked, the run
|
||
* count stays at 0; after release, exactly ONE body has run (the duplicate was
|
||
* deduped, not queued).
|
||
*/
|
||
static void test_configpush_offload_dedups(void)
|
||
{
|
||
cluster_t c;
|
||
memset(&c, 0, sizeof(c));
|
||
ha_init(&c.ha, &g_cfg, 0);
|
||
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4));
|
||
|
||
cp_reset_probe();
|
||
atomic_store(&g_cp_gate_on, 1); /* lane body parks at the gate */
|
||
ha_configpush_probe = cp_probe;
|
||
ha_set_configpush_offload_for_test(1);
|
||
|
||
/* First config push — submitted; its body parks in the lane. */
|
||
TEST_ASSERT_EQUAL_INT(0, ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7));
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, cp_wait_parked(2000),
|
||
"the first config push body must reach the gate (running on the lane)");
|
||
|
||
/* Second config push while the first is in flight. */
|
||
TEST_ASSERT_EQUAL_INT(0, ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7));
|
||
|
||
/* Give a generous window: the duplicate must NOT run. */
|
||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 200 * 1000000L }; /* 200 ms */
|
||
nanosleep(&ts, NULL);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_cp_runs),
|
||
"Req 1.5: the duplicate config push must be deduped while one is in flight");
|
||
|
||
/* Release; exactly the first body completes. */
|
||
cp_release_gate();
|
||
int runs = cp_wait_runs(1, 2000);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
|
||
"Req 1.5: exactly one config push ran (duplicate deduped)");
|
||
/* And no second body sneaks in after release. */
|
||
nanosleep(&ts, NULL);
|
||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_cp_runs),
|
||
"Req 1.5: the deduped duplicate was dropped, not queued behind the first");
|
||
|
||
ha_set_configpush_offload_for_test(0); /* restore production default (OFF) */
|
||
ha_configpush_probe = NULL;
|
||
slowlane_fini(&c.slowlane);
|
||
ha_fini(&c.ha);
|
||
printf("test_configpush_offload_dedups: PASS\n");
|
||
}
|
||
|
||
/* ── Suite entry point ──────────────────────────────────────────────────── */
|
||
|
||
void test_ha_liveness(void)
|
||
{
|
||
UNITY_BEGIN();
|
||
RUN_TEST(test_liveness_poll_fires_failover_when_timeout);
|
||
RUN_TEST(test_liveness_poll_rearms_when_fresh);
|
||
RUN_TEST(test_liveness_poll_cancelled_is_noop);
|
||
RUN_TEST(test_ha_on_heartbeat_received_updates_last_activity);
|
||
RUN_TEST(test_liveness_poll_skips_if_not_standby);
|
||
RUN_TEST(test_ha_arm_no_crash_no_thread);
|
||
RUN_TEST(test_ha_disarm_makes_subsequent_poll_noop);
|
||
RUN_TEST(test_primary_hb_sends_nop_when_idle);
|
||
RUN_TEST(test_primary_hb_skips_nop_when_fresh);
|
||
RUN_TEST(test_primary_hb_evicts_silent_standby);
|
||
RUN_TEST(test_primary_hb_keeps_live_standby);
|
||
RUN_TEST(test_primary_hb_keeps_unconfirmed_standby);
|
||
RUN_TEST(test_primary_hb_stamps_own_committed_and_term);
|
||
RUN_TEST(test_on_standby_connected_sets_last_push_time);
|
||
RUN_TEST(test_cluster_dispatch_nop_resets_activity_when_standby);
|
||
RUN_TEST(test_cluster_dispatch_nop_skips_when_primary);
|
||
RUN_TEST(test_cluster_dispatch_ctrlhello_resets_activity_when_standby);
|
||
RUN_TEST(test_ha_arm_does_not_increase_thread_count);
|
||
/* raft-drive-async Task 2.1 — reconnect offload (Req 2.1, 2.2, 2.5) */
|
||
RUN_TEST(test_reconnect_offload_tick_returns_promptly);
|
||
RUN_TEST(test_reconnect_inline_tick_blocks_RED);
|
||
RUN_TEST(test_reconnect_offload_dedups_per_peer);
|
||
/* raft-drive-async Task 2.2 — inbound/outbound ownership (Req 2.6, 4.4) */
|
||
RUN_TEST(test_reconnect_resolves_dual_channel_same_peer);
|
||
RUN_TEST(test_reconnect_preserves_other_peer_channel);
|
||
/* raft-drive-async Task 2.3 — reconnect-offload INTEGRATION (Req 2.1, 2.2, 2.5) */
|
||
RUN_TEST(test_reconnect_offload_sustains_survivor_heartbeats);
|
||
RUN_TEST(test_reconnect_inline_stalls_survivor_heartbeats_RED);
|
||
/* raft-drive-async Task 3.1 — PeerCtrlDispatch (Req 1.4, 4.1, 4.2, 4.5, 6.x, 8.1, 8.2) */
|
||
RUN_TEST(test_peerctrl_dispatch_runs_on_pool);
|
||
RUN_TEST(test_peerctrl_inline_when_flag_off);
|
||
RUN_TEST(test_peerctrl_preserves_per_endpoint_order);
|
||
/* raft-drive-async Task 3.2 — SnapshotSend slow-lane offload (Req 1.1, 1.5, 6.x) */
|
||
RUN_TEST(test_snapshot_offload_runs_on_lane);
|
||
RUN_TEST(test_snapshot_inline_when_flag_off);
|
||
RUN_TEST(test_snapshot_offload_dedups_per_peer);
|
||
/* raft-drive-async Task 3.3 — requester-less ConfigPush slow-lane offload (Req 1.1, 1.5, 6.x) */
|
||
RUN_TEST(test_configpush_offload_runs_on_lane);
|
||
RUN_TEST(test_configpush_inline_when_flag_off);
|
||
RUN_TEST(test_configpush_offload_dedups);
|
||
UNITY_END();
|
||
}
|