From e07667ff9b6f9d7fc2e8d81f290e8fe1b21fa587 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 26 Jun 2026 08:08:31 +0000 Subject: [PATCH] feat(raft-drive-async): offload delaymgr peer reconnect to slow lane (task 2.1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Heartbeat timer detects a disconnected peer, CAS-acquires a per-peer single-flight flag, slowlane_submits the bounded connect, and returns — the delaymgr never blocks on connect. Lane worker (reconnect_lane_fn) does the bounded endpoint_connect + lock-guarded iomgr_monitor; flag cleared on all terminal paths (incl. -EEXIST race fix). Default-ON kill-switch retains the inline path for TDD-RED/fallback. Never self-terminates on failure. 2.2 adds the inbound/outbound race + failure hardening (documented seam). 196/0 green. --- .kiro/specs/raft-drive-async/tasks.md | 3 +- src/controller/ha.c | 261 ++++++++++++++++-- src/controller/ha.h | 42 +++ src/controller/testsuite/test_ha_liveness.c | 280 ++++++++++++++++++++ 4 files changed, 569 insertions(+), 17 deletions(-) diff --git a/.kiro/specs/raft-drive-async/tasks.md b/.kiro/specs/raft-drive-async/tasks.md index c36840f2..35b6e731 100644 --- a/.kiro/specs/raft-drive-async/tasks.md +++ b/.kiro/specs/raft-drive-async/tasks.md @@ -55,7 +55,7 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - _Depends: 1.6_ - [ ] 2. Phase 2 — Reconnect offload (sender-side liveness) -- [ ] 2.1 Offload the delaymgr peer reconnect to the slow lane +- [x] 2.1 Offload the delaymgr peer reconnect to the slow lane - The heartbeat timer detects a disconnected peer and submits the reconnect to the lane (per-peer dedup key), then returns immediately; the timer never performs the connect inline. - Refcount the endpoint across the in-flight window. - Observable: while one peer is unreachable, the delaymgr timer keeps firing and heartbeats to reachable peers continue within the interval; only one reconnect attempt per peer is in flight at a time. @@ -137,4 +137,5 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - **Build/test env**: builds MUST run in the nix dev shell (`nix develop --command meson compile -C src/xce.build `). The IDE clang reports false errors (missing openssl/unity headers, unknown `-fvar-tracking`) because it isn't in the nix shell — ignore those. Controller tests: `rm -rf /tmp/xce_*` first (bug-924 overlayfs), run `./src/xce.build/controller/controller_unit_test`, set `ASAN_OPTIONS=detect_leaks=0`. - **Pre-existing leak (not ours)**: the default-options full controller suite exits non-zero due to a LeakSanitizer leak in `test_commit_barrier_reply.c` / `test_worker_join.c` / `test_egress_filter.c` (~174KB). Unrelated to this spec. Run with `detect_leaks=0` (or a `-t` filter) so it doesn't mask raft-drive-async regressions. - **1.1**: `drive_async_liveness_reserved(n_peers, n_vcpus) = max(1, min(n_peers−1, n_vcpus−1))` in `ha.c` is the single source of truth for `VTASK_LIVENESS_RESERVED` (floor applied LAST). Consumed by 1.6. `myconf_t` CLI override fields for the new `DEFAULT_*` constants were deferred to the consumer tasks (1.2 connect-timeout, 1.3/1.6) per YAGNI — add each field with its reader. +- **2.1**: reconnect offload is gated by a default-ON kill-switch `g_drive_async_reconnect_offload` (legacy inline path retained for the TDD-RED proof + operational fallback). Per-peer single-flight via `ha_state_t.reconnect_inflight[]` CAS + slowlane key; flag cleared on all 4 terminal paths (prep-fail, lane-fn success/-EAGAIN, -EEXIST, hard-error). Lane fn = `reconnect_lane_fn` → `ha_connect_finish_peer` (happy path: bounded connect + lock-guarded iomgr_monitor; 2.2 adds inbound/outbound race + failure hardening via the documented seam). **FOLLOW-UP (test hygiene, LOW)**: the GREEN reconnect tests in test_ha_liveness.c leak the prepped endpoint on the simulated-success hook path (production-clean; masked by detect_leaks=0) — free the un-adopted ep in teardown when convenient. - **1.6**: pool reservation lives in vtaskmgr (`vtask_class_t` + `blocking_inflight`/`liveness_reserved` + `vtaskmgr_blocking_acquire/release`); commit folds in via opt-in `commit_set_pool` (kept `commit_engine_init(n_vcpus)` signature — the `n_vcpus` local cap is the genuine pool==NULL fallback for unit fixtures). **Production wiring is in `cluster.c` phase2** (once-guarded, right after `commit_engine_init`): `vtaskmgr_set_liveness_reserved(R)` + `commit_set_pool` — DO NOT remove; without it the reservation is dead. Lock order `p->lock → ce->lock → blk_lock` (cycle-free). New-cap = `n_vcpus−R` (tighter than old `n_vcpus−2` for 5/7-node — intended). LOW deferred: `vtaskmgr_blocking_cap()` reads `liveness_reserved` without `blk_lock` (init-time set-once, benign). When 3.1 dispatches peer_ctrl, REQ_BLOCK slot is released in `run_task` — ensure no double-count vs the commit path. diff --git a/src/controller/ha.c b/src/controller/ha.c index 44b946ac..ef61e529 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -3767,6 +3767,33 @@ int handle_request_vote(ha_state_t *ha, } #endif /* XCE_TESTING */ +/* ── raft-drive-async Task 2.1: reconnect-offload feature flag (ALL builds) ── + * + * drive_async_reconnect_offload gates the peer reconnect path: + * ON (production default) → primary_heartbeat_fn submits the peer reconnect + * to the slow lane and returns immediately; the delaymgr thread NEVER + * performs the blocking connect inline (ha-async-io.md thread-role + * contract — a blocking connect on the heartbeat/election-timeout thread + * stalls liveness to healthy peers and triggers spurious elections). + * OFF → legacy inline blocking connect on the delaymgr thread; retained only + * so the TDD RED test can demonstrate the pre-change stall. + * + * Compiled in ALL builds (production reads it on every heartbeat tick); the + * setter is the test seam to flip RED↔GREEN. */ +static _Atomic int g_drive_async_reconnect_offload = 1; + +void ha_set_reconnect_offload_for_test(int on) +{ + atomic_store_explicit(&g_drive_async_reconnect_offload, on ? 1 : 0, + memory_order_release); +} + +int ha_reconnect_offload_enabled(void) +{ + return atomic_load_explicit(&g_drive_async_reconnect_offload, + memory_order_acquire); +} + /* ── Test-only hook shims (XCE_TESTING only) ────────────────────────────── * Type definitions (election_timeout_ctx_t), extern declarations * (cluster_init_phase2, push_config_transport_only, @@ -3816,6 +3843,16 @@ int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term, uint64_t committed_lsn, uint64_t committed_config_tag) = NULL; +/* + * ha_reconnect_connect_hook: test-only seam for the BLOCKING half of the + * offloaded peer reconnect (raft-drive-async Task 2.1). When non-NULL, the + * slow-lane reconnect worker calls this INSTEAD OF the real bounded + * endpoint_connect + iomgr_monitor, so a test can drive the reconnect path + * deterministically (e.g. park the worker to simulate an unreachable peer). + * Returns 0 to simulate success, <0 to simulate failure. Req 2.1, 2.2, 2.5. + */ +int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx) = NULL; + /* * ha_repl_push_config_for_standby_hook: test-only seam for the per-standby * config push triggered by the heartbeat-reply decision loop (Task 3.3). @@ -6406,24 +6443,43 @@ static int standby_ep_err_handler_primary(void *ep_raw) } /* - * ha_connect_to_peer: open a new outbound TLS connection to a peer candidate - * and register it as a standby channel. Called from primary_heartbeat_fn - * for peers that are absent from standby_endpoints. + * reconnect_ctx_t: owns the prepared endpoint across the in-flight reconnect + * window (raft-drive-async Task 2.1). Built on the delaymgr tick by the + * non-blocking prep (ha_connect_prep_peer), consumed once on a slow-lane + * worker by reconnect_lane_fn, then freed. The ctx keeps `ep`/`sctx` valid + * for the whole window so the lane worker never touches freed memory while the + * delaymgr thread races ahead to the next tick (Req 2.3 lifetime). */ -static void ha_connect_to_peer(cluster_t *c, int peer_idx) +typedef struct { + cluster_t *c; + int peer_idx; /* ABSOLUTE candidate index */ + endpoint_t *ep; /* prepared, not yet connected */ + standby_ep_ctx_t *sctx; /* == ep->data */ + int should_log; +} reconnect_ctx_t; + +/* + * ha_connect_prep_peer: the NON-BLOCKING half of the outbound peer connect. + * Allocates the endpoint context, initialises the endpoint, sets the SSL + * client state and the on-connected / err handlers — all of which are + * non-blocking and safe to run on the delaymgr tick. Does NOT perform the + * blocking connect (that is ha_connect_finish_peer, run on the slow lane). + * + * Returns a heap reconnect_ctx_t* (caller owns) on success, or NULL on an + * allocation failure (nothing leaked). + */ +static reconnect_ctx_t *ha_connect_prep_peer(cluster_t *c, int peer_idx) { - ha_state_t *ha = &c->ha; - const xce_candidate_t *peer = - &c->myconf->peers_cfg.candidates[peer_idx]; + ha_state_t *ha = &c->ha; standby_ep_ctx_t *sctx = calloc(1, sizeof(*sctx)); - if (!sctx) return; + if (!sctx) return NULL; sctx->ha = ha; sctx->c = c; sctx->peer_idx = peer_idx; endpoint_t *ep = endpoint_get(); - if (!ep) { free(sctx); return; } + if (!ep) { free(sctx); return NULL; } endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0); endpoint_set_peer(ep, c->myhost); @@ -6443,14 +6499,74 @@ static void ha_connect_to_peer(cluster_t *c, int peer_idx) int n = ++ha->peer_connect_fail_count[peer_idx]; int should_log = (n == 1) || (n % HA_CONNECT_LOG_THROTTLE == 0); - if (should_log) - INFO("HA: peer[%d] (%s) not connected — initiating outbound connect (attempt %d)", - peer_idx, peer->hostname, n); + reconnect_ctx_t *rc_ctx = calloc(1, sizeof(*rc_ctx)); + if (!rc_ctx) { + ep->data = NULL; + endpoint_fini(ep); + free(sctx); + return NULL; + } + rc_ctx->c = c; + rc_ctx->peer_idx = peer_idx; + rc_ctx->ep = ep; + rc_ctx->sctx = sctx; + rc_ctx->should_log = should_log; + return rc_ctx; +} + +/* + * ha_connect_finish_peer: the BLOCKING half of the outbound peer connect — + * the bounded connect + lock-guarded iomgr registration. MUST run on a + * slow-lane worker, NEVER on the delaymgr tick (Req 2.1, 2.2; ha-async-io.md + * thread-role contract). + * + * On success the endpoint is connecting and registered with iomgr; on_connected + * (on_standby_connected) fires later from the iomgr loop. On failure the + * endpoint is torn down here and the controller simply retries on the next + * heartbeat tick (controllers never self-terminate). + * + * Returns 0 on a successful connect+register, <0 otherwise. + * + * NOTE (Task 2.2 seam): this implements the happy-path connect+register + + * teardown-on-failure. The inbound-vs-outbound connection-ownership race + * resolution, the no-lost/no-reordered-message guarantee on success, and the + * richer failure/retry hardening are owned by Task 2.2 — they layer on top of + * this happy path without changing this function's contract (return 0/<0, + * caller clears the per-peer in-flight flag). + */ +static int ha_connect_finish_peer(reconnect_ctx_t *rc_ctx) +{ + cluster_t *c = rc_ctx->c; + endpoint_t *ep = rc_ctx->ep; + standby_ep_ctx_t *sctx = rc_ctx->sctx; + int peer_idx = rc_ctx->peer_idx; + const xce_candidate_t *peer = &c->myconf->peers_cfg.candidates[peer_idx]; + +#ifdef XCE_TESTING + if (ha_reconnect_connect_hook) { + /* Deterministic test path: the hook stands in for the real bounded + * connect + register so a test can park the worker (unreachable peer) + * or return a synthetic success/failure without a live socket. On + * the test path we do NOT touch the real endpoint fd. */ + int hrc = ha_reconnect_connect_hook(c, peer_idx); + if (hrc < 0) { + ep->data = NULL; + endpoint_fini(ep); + free(sctx); + } + return hrc; + } +#endif + + if (rc_ctx->should_log) + INFO("HA: peer[%d] (%s) not connected — initiating outbound connect " + "(attempt %d)", peer_idx, peer->hostname, + c->ha.peer_connect_fail_count[peer_idx]); int rc = endpoint_connect(ep, peer->hostname, peer->port, DEFAULT_CONNECT_TIMEOUT_MS); if (rc < 0) { - if (should_log) + if (rc_ctx->should_log) WARN("HA: connect to peer[%d] (%s:%u) failed: %d", peer_idx, peer->hostname, peer->port, rc); /* endpoint_fini calls FREEUP(ep->data) which frees sctx — do not @@ -6458,17 +6574,130 @@ static void ha_connect_to_peer(cluster_t *c, int peer_idx) ep->data = NULL; endpoint_fini(ep); free(sctx); - return; + return rc; } - rc = iomgr_monitor(ep); + rc = iomgr_monitor(ep); /* lock-guarded epoll registration (Req 4.4) */ if (rc < 0) { ep->data = NULL; endpoint_fini(ep); free(sctx); - return; + return rc; } INFO("HA: connecting outbound to peer[%d] (%s:%u)", peer_idx, peer->hostname, peer->port); + return 0; +} + +/* + * reconnect_lane_fn: the slow-lane worker entry (raft-drive-async Task 2.1). + * Runs the bounded blocking connect off the delaymgr thread, then ALWAYS + * clears the per-peer in-flight flag (success or failure) so the dedup never + * wedges and the next heartbeat tick can retry an unreachable peer (Req 2.5, + * 2.4 — never self-terminates). Frees the reconnect ctx. + */ +static void reconnect_lane_fn(void *arg) +{ + reconnect_ctx_t *rc_ctx = (reconnect_ctx_t *)arg; + if (!rc_ctx) return; + + cluster_t *c = rc_ctx->c; + int peer_idx = rc_ctx->peer_idx; + + (void)ha_connect_finish_peer(rc_ctx); + + /* Clear in-flight LAST, after the endpoint is fully connected/registered + * or torn down — so a concurrent heartbeat tick cannot enqueue a second + * reconnect that races this one's endpoint setup (Req 2.5). */ + atomic_store_explicit(&c->ha.reconnect_inflight[peer_idx], 0, + memory_order_release); + free(rc_ctx); +} + +/* + * ha_connect_to_peer: open a new outbound TLS connection to a peer candidate + * and register it as a standby channel. Called from primary_heartbeat_fn for + * peers that are absent from standby_endpoints. + * + * Offload flag ON (default, Task 2.1): prepares the endpoint (non-blocking) + * and submits the bounded connect to the slow lane with a per-peer dedup key, + * then returns immediately — the delaymgr thread NEVER blocks on connect. + * + * Offload flag OFF (legacy / TDD RED only): runs the blocking connect inline + * on the calling (delaymgr) thread. Retained solely so the RED test can show + * the pre-change stall; production runs with the flag ON. + */ +static void ha_connect_to_peer(cluster_t *c, int peer_idx) +{ + ha_state_t *ha = &c->ha; + + if (!ha_reconnect_offload_enabled()) { + /* ── Legacy inline path (flag OFF) — blocks the caller. ── */ + reconnect_ctx_t *rc_ctx = ha_connect_prep_peer(c, peer_idx); + if (!rc_ctx) return; + (void)ha_connect_finish_peer(rc_ctx); + free(rc_ctx); + return; + } + + /* ── Offload path (flag ON) — non-blocking on the delaymgr thread. ── + * + * Per-peer single-flight: CAS the in-flight flag 0->1. If it was already + * 1, a reconnect to this peer is already in flight on the lane — enqueue + * nothing (Req 2.5). This guard is the authoritative dedup; the lane's + * own per-key dedup is a redundant backstop. */ + int expected = 0; + if (!atomic_compare_exchange_strong_explicit( + &ha->reconnect_inflight[peer_idx], &expected, 1, + memory_order_acq_rel, memory_order_acquire)) { + return; /* already reconnecting this peer */ + } + + reconnect_ctx_t *rc_ctx = ha_connect_prep_peer(c, peer_idx); + if (!rc_ctx) { + /* prep failed (alloc) — release the in-flight guard so the next tick + * retries; controller stays alive (Req 2.4). */ + atomic_store_explicit(&ha->reconnect_inflight[peer_idx], 0, + memory_order_release); + return; + } + + /* Submit the blocking connect to the slow lane. Dedup key is the + * per-peer absolute index (+1 so it is non-zero; key 0 disables dedup). */ + uint64_t key = (uint64_t)(peer_idx + 1); + int srv = slowlane_submit(&c->slowlane, key, reconnect_lane_fn, rc_ctx); + if (srv == -EEXIST) { + /* Lane still holds this peer's dedup key from a PRIOR reconnect task. + * This is the narrow window where that task already cleared our + * reconnect_inflight flag (its last action) but the lane has not yet + * removed its key. Our CAS therefore succeeded (flag was 0) even + * though the lane refuses the submit. We MUST release the flag we just + * set — otherwise no task owns its clear and dedup wedges permanently. + * Tear down our duplicate prep; the next tick retries cleanly. */ + rc_ctx->ep->data = NULL; + endpoint_fini(rc_ctx->ep); + free(rc_ctx->sctx); + free(rc_ctx); + atomic_store_explicit(&ha->reconnect_inflight[peer_idx], 0, + memory_order_release); + return; + } + if (srv < 0 && srv != -EAGAIN) { + /* Hard submit failure (alloc / lane stopping). -EAGAIN is NOT a + * failure: the task is still queued and will run, so we keep the flag + * set and let reconnect_lane_fn clear it. Only on a true failure do + * we tear down and release the guard so the next tick retries. */ + WARN("HA: slowlane_submit reconnect peer[%d] failed: %d — retry next tick", + peer_idx, srv); + rc_ctx->ep->data = NULL; + endpoint_fini(rc_ctx->ep); + free(rc_ctx->sctx); + free(rc_ctx); + atomic_store_explicit(&ha->reconnect_inflight[peer_idx], 0, + memory_order_release); + return; + } + /* 0 or -EAGAIN: the task is enqueued and will run exactly once on a lane + * worker; reconnect_lane_fn clears the in-flight flag on completion. */ } /* diff --git a/src/controller/ha.h b/src/controller/ha.h index 8b02bbd8..65f60eea 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -254,6 +254,19 @@ typedef struct { * cleared and the config is activated. * NEVER self-terminates (loud non-serving * fence only). 0 at init (ha_init memset). */ + /* ── New field added by raft-drive-async spec (Task 2.1) ──────── */ + _Atomic int reconnect_inflight[MAX_HA_CANDIDATES]; /* Per-peer + * single-flight guard for the offloaded + * delaymgr reconnect (Req 2.5): set (CAS + * 0->1) when a reconnect to peer[i] is + * submitted to the slow lane, cleared by + * the lane worker on completion (success + * or failure) so the next heartbeat tick + * can retry. While set, the heartbeat + * reconnect scan enqueues no duplicate for + * that peer. Indexed by ABSOLUTE candidate + * index (peers_cfg.candidates[]). 0 at + * init (ha_init memset). */ } ha_state_t; /* ── Public API ────────────────────────────────────────────── */ @@ -1196,6 +1209,18 @@ void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha); */ void ha_start_primary_heartbeat(cluster_t *c); +/* + * ha_set_reconnect_offload_for_test / ha_reconnect_offload_enabled: + * Feature-flag accessors (raft-drive-async Task 2.1). The offload flag gates + * whether primary_heartbeat_fn submits the peer reconnect to the slow lane (ON, + * the production default — the delaymgr thread never blocks on connect) or runs + * the legacy inline blocking connect on the delaymgr thread (OFF, retained only + * so the TDD RED test can demonstrate the pre-change stall). These accessors + * are compiled in ALL builds; the setter is used by tests to flip the flag. + */ +void ha_set_reconnect_offload_for_test(int on); +int ha_reconnect_offload_enabled(void); + /* ── Task 2.4: Non-blocking async election initiator ────────────────────── */ /* @@ -1353,6 +1378,23 @@ extern int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term, uint64_t committed_lsn, uint64_t committed_config_tag); +/* + * ha_reconnect_connect_hook: test-only seam for the BLOCKING half of the + * offloaded peer reconnect (raft-drive-async Task 2.1). The reconnect splits + * into a non-blocking prep (alloc endpoint, set callbacks) on the delaymgr + * tick and a blocking connect+monitor that runs on a slow-lane worker. When + * non-NULL, the lane worker calls this INSTEAD OF the real bounded + * endpoint_connect + iomgr_monitor, so a test can drive the reconnect path + * deterministically (e.g. park the worker to simulate an unreachable peer) + * without a real socket. + * + * Signature: int hook(cluster_t *c, int peer_idx) + * returns 0 to simulate a successful connect, <0 to simulate failure. + * The lane worker clears reconnect_inflight[peer_idx] after this returns + * regardless of outcome, so dedup never wedges (Req 2.5). + */ +extern int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx); + /* * ha_repl_push_config_for_standby_hook: test-only seam for the per-standby * config push triggered by the heartbeat-reply decision loop (Task 3.3). diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index ee2df4eb..dffa47fc 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -22,9 +22,12 @@ #include #include +#include + #include #include "ha.h" #include "netio.h" +#include "slowlane.h" #include "xce/cluster_def.h" /* ── XCE_TESTING seam ────────────────────────────────────────────────────── */ @@ -842,6 +845,279 @@ static void test_ha_arm_does_not_increase_thread_count(void) 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"); +} + /* ── Suite entry point ──────────────────────────────────────────────────── */ void test_ha_liveness(void) @@ -865,5 +1141,9 @@ void test_ha_liveness(void) 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); UNITY_END(); }