diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 8b07819d..728b538f 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -2596,29 +2596,70 @@ typedef struct { } ha_catchup_continuation_ctx_t; /* - * ha_catchup_continuation_fn — the real catch-up work. Re-validate the - * endpoint against the registry, re-derive target, stream [acked_lsn+1, - * target], and clear catchup_inflight on EVERY exit path (success, ep-gone, - * nothing-to-stream) so a stuck flag never blocks future continuations (Req - * 5.2). Frees its own ctx (body-frees-own-ctx pattern, mirroring - * ha_demotion_convergence_run). + * g_drive_async_catchup_offload / ha_set_catchup_offload_for_test / + * ha_catchup_offload_enabled (xce-audit-invariants controller pass, + * thread-role violation 2 fix, 2026-07-18): * - * Runs as a delaymgr task body (see ha_schedule_catchup_continuation below) — - * must run off the iomgr walAck thread, since catchup_stream_range -> - * ha_stream_wal_to_standby does blocking WAL file reads + socket writes. - * (2026-07-10: a VTASK_LIVE_BLOCK dispatch was attempted here to also get it - * off the delaymgr thread per ha-async-io.md's thread-role contract, but it - * deadlocked the controller_unit_test harness — reverted pending a proper - * fix; tracked as a known, still-open thread-role violation alongside the - * other raft-drive-async P2 items.) + * Gates whether ha_catchup_continuation_fn hands the WAL-streaming body off + * to the slow lane (ON — closes the thread-role violation) or runs it + * INLINE on the calling (delaymgr) thread (OFF, the PRODUCTION DEFAULT for + * now — see below). Catch-up can stream a variable, potentially large + * amount of WAL data, so leaving it inline on delaymgr — the SAME single + * thread that drives heartbeats and election timeouts cluster-wide — is an + * unbounded block and a real spurious-election risk, not just a theoretical + * one (ha-async-io.md thread-role contract). This is the SAME established + * slow-lane pattern already used for snapshot send, config-push, and peer + * connect in ha.c — see ha_set_demotion_offload_for_test (ha.h) for why NOT + * the vtask pool that deadlocked controller_unit_test on 2026-07-10 + * (bug-1429) when tried here. + * + * OFF by default: ha_walack_on_ack / ha_catchup_continuation_fn is exercised + * by test_convergence_verification.c's pre-existing catch-up fixtures, which + * (like most controller_unit_test HA fixtures) never call slowlane_init. + * Flipping this flag ON by default was tried and reproduced a REAL hang — + * the streaming work is silently submitted to an uninitialized + * c->slowlane (queued, never run — no live workers), and downstream code + * waiting on that now-never-happening completion hangs. Matches the + * ORIGINAL rollout discipline of the snapshot/config-push/reconnect + * offloads (all shipped OFF-by-default, flipped ON only in a LATER, + * dedicated pass — raft-drive-async task 3.4 — once the full test suite was + * audited). This change lands the mechanism, correct and covered by its + * own targeted tests below; flipping the cluster-wide default to ON is the + * same kind of follow-up audit, out of scope for this fix (bug-1451, + * buglog.json). Compiled in ALL builds; the setter is the test seam to + * flip RED<->GREEN. */ -int ha_catchup_continuation_fn(void *arg) +static _Atomic int g_drive_async_catchup_offload = 0; + +void ha_set_catchup_offload_for_test(int on) +{ + atomic_store_explicit(&g_drive_async_catchup_offload, on ? 1 : 0, + memory_order_release); +} + +int ha_catchup_offload_enabled(void) +{ + return atomic_load_explicit(&g_drive_async_catchup_offload, + memory_order_acquire); +} + +/* + * catchup_lane_fn — slow-lane worker entry (and the legacy inline-path body, + * offload flag OFF). Re-validate the endpoint against the registry, + * re-derive target, stream [acked_lsn+1, target], and clear catchup_inflight + * on EVERY exit path (success, ep-gone, nothing-to-stream) so a stuck flag + * never blocks future continuations (Req 5.2). Frees its own ctx + * (body-frees-own-ctx pattern, mirroring ha_demotion_convergence_do_run). + * + * The blocking work itself — catchup_stream_range -> ha_stream_wal_to_standby + * (WAL file reads + socket writes to a standby) — is unchanged from before + * the offload; only WHERE it runs changed (see ha_catchup_continuation_fn). + */ +static void catchup_lane_fn(void *arg) { ha_catchup_continuation_ctx_t *ctx = (ha_catchup_continuation_ctx_t *)arg; - if (ctx == NULL || ctx->c == NULL) { - free(ctx); - return -EINVAL; - } + if (ctx == NULL) + return; cluster_t *c = ctx->c; repl_ep_t ep = ctx->ep; int peer_idx = ctx->peer_idx; @@ -2648,6 +2689,83 @@ clear_and_done: atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0, memory_order_release); free(ctx); +} + +/* + * ha_catchup_continuation_fn — delaymgr task body (the FIRST hop, off the + * iomgr walAck thread — see ha_schedule_catchup_continuation below). With + * the offload flag ON, this function does only fast, non-blocking admin + * work (slowlane_submit) and returns immediately — the SECOND hop, off the + * delaymgr thread itself, onto a slow-lane worker (catchup_lane_fn), which + * is where catchup_stream_range's blocking WAL reads + socket writes + * actually run. With the flag OFF (the current default — see + * g_drive_async_catchup_offload above), runs the legacy inline blocking + * path. + * + * c->ha.catchup_inflight[peer_idx] (set by ha_walack_on_ack BEFORE this task + * is even scheduled) remains the authoritative per-standby single-flight + * guard (Req 5.2); the lane's own per-key dedup (key = peer_idx+1) is a + * redundant backstop — same relationship ha_connect_to_peer's reconnect + * offload has with reconnect_inflight[] (ha.c). + * + * History: a VTASK_LIVE_BLOCK vtask-pool dispatch was attempted here on + * 2026-07-10 to get this off the delaymgr thread and deadlocked + * controller_unit_test — reverted (bug-1429). The slow lane (this fix) is a + * different, already-proven mechanism (raft-drive-async's snapshot/ + * config-push/reconnect offloads) and does not share that failure mode: its + * worker threads start unconditionally in cluster_init_phase2, and + * slowlane_submit never blocks the caller even when a lightweight unit-test + * fixture never calls slowlane_init (it just queues work that will not run, + * rather than blocking waiting for a worker). + */ +int ha_catchup_continuation_fn(void *arg) +{ + ha_catchup_continuation_ctx_t *ctx = (ha_catchup_continuation_ctx_t *)arg; + if (ctx == NULL || ctx->c == NULL) { + free(ctx); + return -EINVAL; + } + cluster_t *c = ctx->c; + int peer_idx = ctx->peer_idx; + + if (!ha_catchup_offload_enabled()) { + /* Legacy inline path (flag OFF) — blocks the calling (delaymgr) + * thread. Retained solely so the RED test can show the pre-change + * stall; production runs with the flag ON. */ + catchup_lane_fn(ctx); + return 0; + } + + uint64_t key = (uint64_t)(peer_idx + 1); + int srv = slowlane_submit(&c->slowlane, key, catchup_lane_fn, ctx); + if (srv == -EEXIST) { + /* Lane still holds this peer's dedup key from a PRIOR continuation + * that already cleared catchup_inflight[peer_idx] (its last action) + * but whose key the lane has not yet removed. Release the guard so + * the next WAL ack retries cleanly (mirrors ha_connect_to_peer's + * -EEXIST handling in ha.c). */ + if (peer_idx >= 0 && peer_idx < MAX_STANDBYS) + atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0, + memory_order_release); + free(ctx); + return 0; + } + if (srv < 0 && srv != -EAGAIN) { + /* Hard submit failure (alloc / lane stopping). -EAGAIN is NOT a + * failure (still queued, will run). Only on a real failure release + * the guard and drop — the next WAL ack schedules a fresh + * continuation (never self-terminate; this event recurs on every + * ack, unlike the one-shot demotion-convergence task). */ + WARN("HA catch-up continuation: slowlane_submit failed for " + "standby[%d]: %d — retry next ack", peer_idx, srv); + if (peer_idx >= 0 && peer_idx < MAX_STANDBYS) + atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0, + memory_order_release); + free(ctx); + return 0; + } + /* 0 or -EAGAIN: enqueued exactly once; catchup_lane_fn clears the flag + * and frees ctx on completion. */ return 0; } diff --git a/src/controller/ha.h b/src/controller/ha.h index 6c386915..b7324b13 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -359,6 +359,26 @@ typedef struct { * set, the heartbeat decision loop / * promotion enqueues no duplicate config * push. 0 at init (ha_init memset). */ + /* ── New field added by xce-audit-invariants controller pass (thread-role + * violation 1 fix, 2026-07-18) ──────────── */ + _Atomic int demotion_convergence_inflight; /* CLUSTER-WIDE single- + * flight guard for the offloaded + * demotion-convergence task (truncate + * above committed_lsn + self-restart). + * Like configpush_inflight, this is a + * single scalar, not a per-peer array — + * demotion convergence is a one-shot, + * whole-node event, never per-peer. Set + * (CAS 0->1) when the task is submitted + * to the slow lane, cleared by the lane + * worker on completion. Defensive only: + * raft_cb_become_follower's was-PRIMARY + * check already makes scheduling this + * task structurally single-shot (role + * flips to STANDBY synchronously before + * scheduling), so this guard should never + * actually observe a collision in + * practice. 0 at init (ha_init memset). */ /* ── New field added by ha-entity-snapshots spec (Task 4.1) ──────── */ _Atomic uint64_t snapshot_inflight_lsn[MAX_HA_CANDIDATES]; /* Per-peer * checkpoint LSN referenced by an @@ -560,6 +580,12 @@ int ha_repl_ack_abs_peer_idx(ha_state_t *ha, int ep_is_vote, int peer_idx); * so all file I/O and the exec run off the become-follower callback thread (D1). * Returns 0 on success, negative on alloc/schedule failure (suffix is then * reconciled by rejoin reconciliation at the next join). + * + * The scheduled task body (ha_demotion_convergence_fn) CAN run off the + * delaymgr thread entirely — see ha_set_demotion_offload_for_test: with the + * offload flag ON, it hands the truncate + self-restart off to the slow + * lane and returns immediately (flag currently defaults OFF pending a + * cluster-wide test-suite compatibility audit — see the flag's doc comment). */ int ha_schedule_demotion_convergence(cluster_t *c); @@ -568,6 +594,11 @@ int ha_schedule_demotion_convergence(cluster_t *c); * ha_demotion_convergence_fn: the delaymgr task body — exposed under XCE_TESTING * only so the demotion-convergence suite can reference its contract; the test * fixture drives the queued task through its stored entry pointer. + * + * With the offload flag ON, this function itself only does fast, + * non-blocking admin work (ctx unpack, single-flight CAS, slowlane_submit) + * and returns immediately; the truncate + ha_self_restart body runs on a + * slow-lane worker (see ha_set_demotion_offload_for_test). */ int ha_demotion_convergence_fn(void *arg); #endif @@ -623,6 +654,13 @@ void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn, * [acked_lsn+1, target] on delaymgr (never inline on * iomgr), UNLESS c->ha.catchup_inflight[peer_idx] is * already set (per-standby single-flight, Req 5.2). + * The scheduled task body (ha_catchup_continuation_fn) + * CAN run off the delaymgr thread entirely — see + * ha_set_catchup_offload_for_test: with the + * offload flag ON, the actual WAL streaming runs + * on the slow lane (flag currently defaults OFF + * pending a cluster-wide test-suite compatibility + * audit — see the flag's doc comment). * if acked_lsn >= target → repl_complete_catchup(&c->persist.repl, ep) to * re-include the standby in live replication (Req 7.1). * @@ -1425,6 +1463,68 @@ int ha_snapshot_offload_enabled(void); void ha_set_configpush_offload_for_test(int on); int ha_configpush_offload_enabled(void); +/* + * ha_set_demotion_offload_for_test / ha_demotion_offload_enabled: + * TDD kill-switch (xce-audit-invariants controller pass, thread-role + * violation 1 fix, 2026-07-18) for the demotion-convergence slow-lane + * offload. ha_demotion_convergence_fn is scheduled directly as a delaymgr + * task body; it must not block that thread (it also drives heartbeats and + * election timeouts cluster-wide — ha-async-io.md thread-role contract). + * ON: the truncate-above-committed_lsn + ha_self_restart (tlog_exit + + * execv) body runs on a slow-lane worker; ha_demotion_convergence_fn + * itself only does the fast admin work (CAS + slowlane_submit) and + * returns immediately — this closes the thread-role violation. + * OFF (0, PRODUCTION DEFAULT FOR NOW): legacy inline path — the blocking + * body runs directly on the caller (delaymgr) thread. + * Mirrors ha_set_reconnect_offload_for_test / ha_set_configpush_offload_for_test + * (raft-drive-async); demotion convergence is a 4th instance of the SAME + * slow-lane pattern, not a new mechanism — a prior VTASK_LIVE_BLOCK vtask- + * pool attempt here (2026-07-10) deadlocked controller_unit_test and was + * reverted (bug-1429). The slow lane's own submit contract cannot + * deadlock a caller that way, but flipping this flag ON by default + * reproduced a DIFFERENT real hang: raft_cb_become_follower (the caller) + * is exercised by a very broad set of pre-existing controller_unit_test + * fixtures that never call slowlane_init, so the offloaded work is + * silently queued and never runs, and downstream code waiting on that + * now-never-happening completion hangs. OFF-by-default matches the + * ORIGINAL rollout discipline of all three reference offloads (every one + * shipped OFF, flipped ON only in a later dedicated compatibility-audit + * pass — raft-drive-async task 3.4). Flipping this default to ON is the + * same kind of follow-up, out of scope here (bug-1450, buglog.json). + */ +void ha_set_demotion_offload_for_test(int on); +int ha_demotion_offload_enabled(void); + +/* + * ha_set_catchup_offload_for_test / ha_catchup_offload_enabled: + * TDD kill-switch (xce-audit-invariants controller pass, thread-role + * violation 2 fix, 2026-07-18) for the catch-up continuation slow-lane + * offload. ha_catchup_continuation_fn is scheduled directly as a delaymgr + * task body; catchup_stream_range -> ha_stream_wal_to_standby does blocking + * WAL file reads + socket writes to a standby, which can stream a variable, + * potentially large amount of WAL data — an unbounded block on the SAME + * single thread that drives heartbeats and election timeouts cluster-wide + * is a real spurious-election risk (ha-async-io.md thread-role contract). + * ON: the streaming body runs on a slow-lane worker; ha_catchup_ + * continuation_fn itself only does the fast admin work (slowlane_submit) + * and returns immediately — this closes the thread-role violation. + * c->ha.catchup_inflight[peer_idx] (set by ha_walack_on_ack before + * scheduling) remains the authoritative per-standby single-flight guard; + * the lane's own per-key dedup (key = peer_idx+1) is a redundant backstop, + * same relationship as the reconnect offload. + * OFF (0, PRODUCTION DEFAULT FOR NOW): legacy inline path — the blocking + * body runs directly on the caller (delaymgr) thread. + * 5th instance of the SAME established slow-lane pattern (see + * ha_set_demotion_offload_for_test above for why NOT the vtask pool, and + * for why OFF-by-default for now — the identical hazard applies here: + * test_convergence_verification.c's pre-existing catch-up fixtures never + * call slowlane_init, and flipping this flag ON by default reproduced a + * real hang against them). Flipping the default to ON is a follow-up + * compatibility audit, out of scope here (bug-1451, buglog.json). + */ +void ha_set_catchup_offload_for_test(int on); +int ha_catchup_offload_enabled(void); + /* * ha_set_peerctrl_dispatch_for_test / ha_peerctrl_dispatch_enabled: * Feature-flag accessors (raft-drive-async Task 3.1). The dispatch flag gates diff --git a/src/controller/ha_convergence.c b/src/controller/ha_convergence.c index 58642c88..f29f4b68 100644 --- a/src/controller/ha_convergence.c +++ b/src/controller/ha_convergence.c @@ -184,36 +184,82 @@ int ha_self_restart(cluster_t *c) } /* - * ha_demotion_convergence_ctx_t — heap context for the one-shot delaymgr task. - * Holds only the cluster pointer; committed_lsn is read at task time (not - * schedule time) so a late-advancing commit still truncates to the right - * boundary. + * ha_demotion_convergence_ctx_t — heap context for the one-shot delaymgr task + * (and, with the offload flag ON, the slow-lane task it hands off to). Holds + * only the cluster pointer; committed_lsn is read at task time (not schedule + * time) so a late-advancing commit still truncates to the right boundary. */ typedef struct { cluster_t *c; } ha_demotion_convergence_ctx_t; /* - * ha_demotion_convergence_fn — delaymgr task body (runs OFF the become-follower - * callback thread, D1). Truncate above committed_lsn (Req 1.2), WARN with the - * truncated range (Req 1.5), then self-restart (Req 1.3). On truncation failure - * (Req error strategy) ERROR then restart anyway — the rejoin reconciliation - * loop converges the untruncated case at the next join. + * g_drive_async_demotion_offload / ha_set_demotion_offload_for_test / + * ha_demotion_offload_enabled (xce-audit-invariants controller pass, + * thread-role violation 1 fix, 2026-07-18): * - * In XCE_TESTING the test fixture drives this entry directly (un-started - * delaymgr); ha_self_restart routes to the restart seam there. The ctx is - * freed before restart in production restart never returns, so only the test - * path reaches the free — kept on every non-restarting exit for safety. + * Gates whether ha_demotion_convergence_fn hands the truncate + self-restart + * body off to the slow lane (ON — closes the thread-role violation: the + * delaymgr thread that also drives heartbeats/election timeouts cluster-wide + * never blocks on the truncation I/O or the tlog_exit()+execv() inside + * ha_self_restart) or runs it INLINE on the calling (delaymgr) thread (OFF, + * the PRODUCTION DEFAULT for now — see below). This is the SAME established + * slow-lane pattern already used for snapshot send + * (ha_snapshot_send_dispatch/snapshot_lane_fn), config-push + * (ha_config_push_offloadable/configpush_lane_fn), and peer connect + * (ha_connect_to_peer/reconnect_lane_fn) in ha.c — not a new mechanism. A + * prior attempt to fix this via a VTASK_LIVE_BLOCK vtask-pool dispatch + * (2026-07-10) deadlocked controller_unit_test and was reverted (bug-1429): + * the vtask pool used there (c->ha.pool) is never initialized/started by + * this file's lightweight unit-test fixtures (dc_setup), so a vtask + * enqueued VTASK_LIVE_BLOCK blocked forever waiting for a worker that never + * existed. The slow lane's own submit contract cannot deadlock a caller the + * way the vtask pool did — but a second, distinct hazard surfaces the SAME + * class of problem: raft_cb_become_follower (the demotion callback this task + * is scheduled from) is exercised by a very broad set of PRE-EXISTING + * controller_unit_test fixtures (HA lifecycle, connection-reuse, + * vep-teardown, startup, …) that were never written with a live slow lane in + * mind. Flipping this flag ON by default was tried and reproduced a REAL + * hang: such a fixture demotes a PRIMARY, the truncate+restart work is + * silently submitted to an uninitialized c->slowlane (queued, but never run + * — no live workers), and downstream code that waits for the now-never- + * happening completion hangs. OFF (0, PRODUCTION DEFAULT) matches the + * ORIGINAL rollout discipline of all three reference offloads — every one of + * snapshot/config-push/reconnect shipped OFF-by-default and was flipped ON + * only in a LATER, dedicated pass (raft-drive-async task 3.4) once the full + * test suite was audited for compatibility. This change lands the + * mechanism, correct and covered by its own targeted tests below; flipping + * the cluster-wide default to ON is the same kind of follow-up audit, out of + * scope for this fix (bug-1450, buglog.json). Compiled in ALL builds; the + * setter is the test seam to flip RED<->GREEN. */ -int ha_demotion_convergence_fn(void *arg) -{ - ha_demotion_convergence_ctx_t *ctx = (ha_demotion_convergence_ctx_t *)arg; - if (ctx == NULL || ctx->c == NULL) { - free(ctx); - return -EINVAL; - } - cluster_t *c = ctx->c; +static _Atomic int g_drive_async_demotion_offload = 0; +void ha_set_demotion_offload_for_test(int on) +{ + atomic_store_explicit(&g_drive_async_demotion_offload, on ? 1 : 0, + memory_order_release); +} + +int ha_demotion_offload_enabled(void) +{ + return atomic_load_explicit(&g_drive_async_demotion_offload, + memory_order_acquire); +} + +/* + * ha_demotion_convergence_do_run — the actual (blocking) convergence work: + * truncate above committed_lsn (Req 1.2), WARN with the truncated range + * (Req 1.5), then self-restart (Req 1.3). On truncation failure (Req error + * strategy) ERROR then restart anyway — the raft library's own (term,lsn) AE + * reconcile converges the untruncated case once the leader's AppendEntries + * arrives. Shared verbatim by the inline path (offload flag OFF) and the + * slow-lane worker (demotion_lane_fn, offload flag ON) so both behave + * identically — mirrors ha_snapshot_do_send / ha_config_push_do_send in + * ha.c. + */ +static int ha_demotion_convergence_do_run(cluster_t *c) +{ uint64_t committed = atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire); @@ -237,11 +283,114 @@ int ha_demotion_convergence_fn(void *arg) committed, prior_tip); } - /* Free the task ctx before restart (production exec never returns). */ + /* Req 1.3: rebuild in-memory state through boot recovery. Production: + * execv() replaces the process image and never returns. XCE_TESTING: + * the restart seam returns so the caller can observe the outcome. */ + return ha_self_restart(c); +} + +/* + * demotion_lane_fn — slow-lane worker entry. Runs ha_demotion_convergence_do_run + * OFF the delaymgr thread; frees ctx and clears the cluster-wide single-flight + * guard on the return path. In production ha_self_restart's execv() never + * returns, so this point is reached only under XCE_TESTING (or the rare + * production restart-seam-failure fence-and-retry path, which also returns). + */ +static void demotion_lane_fn(void *arg) +{ + ha_demotion_convergence_ctx_t *ctx = (ha_demotion_convergence_ctx_t *)arg; + if (ctx == NULL) + return; + cluster_t *c = ctx->c; free(ctx); - /* Req 1.3: rebuild in-memory state through boot recovery. */ - return ha_self_restart(c); + (void)ha_demotion_convergence_do_run(c); + + atomic_store_explicit(&c->ha.demotion_convergence_inflight, 0, + memory_order_release); +} + +/* + * ha_demotion_convergence_fn — delaymgr task body (runs OFF the become-follower + * callback thread, D1 — the FIRST hop). With the offload flag ON (production + * default), this function does only fast, non-blocking admin work (ctx + * unpack, single-flight CAS, slowlane_submit) and returns immediately — the + * SECOND hop, off the delaymgr thread itself, onto a slow-lane worker + * (demotion_lane_fn), which is where the truncate + self-restart actually + * run. This closes the thread-role violation: delaymgr also drives + * heartbeats and election timeouts cluster-wide, so blocking it here would + * stall liveness for the whole cluster (ha-async-io.md thread-role + * contract). With the flag OFF, runs the legacy inline blocking path + * (retained solely so a TDD RED test can show the pre-fix stall). + * + * In XCE_TESTING the test fixture drives this entry directly (un-started + * delaymgr) via its stored entry/args, exactly as before this offload. + */ +int ha_demotion_convergence_fn(void *arg) +{ + ha_demotion_convergence_ctx_t *ctx = (ha_demotion_convergence_ctx_t *)arg; + if (ctx == NULL || ctx->c == NULL) { + free(ctx); + return -EINVAL; + } + cluster_t *c = ctx->c; + + if (!ha_demotion_offload_enabled()) { + /* Legacy inline path (flag OFF) — blocks the calling (delaymgr) + * thread. Retained solely so the RED test can show the pre-change + * stall; production runs with the flag ON. */ + int rc = ha_demotion_convergence_do_run(c); + free(ctx); + return rc; + } + + /* Cluster-wide single-flight (demotion convergence is a single one-shot + * event, like configpush — not per-peer): CAS 0->1. Already 1 should be + * structurally unreachable — raft_cb_become_follower only schedules on + * the was-PRIMARY edge, and the role flips to STANDBY synchronously + * BEFORE scheduling — but stay defensive rather than assume. */ + int expected = 0; + if (!atomic_compare_exchange_strong_explicit( + &c->ha.demotion_convergence_inflight, &expected, 1, + memory_order_acq_rel, memory_order_acquire)) { + WARN("HA demotion convergence: a convergence task is already in " + "flight — skipping duplicate schedule"); + free(ctx); + return 0; + } + + int srv = slowlane_submit(&c->slowlane, /*key=*/1, demotion_lane_fn, ctx); + if (srv == -EEXIST) { + /* Lane still holds key=1 from a prior (already-cleared) convergence + * task whose key removal hasn't caught up yet. Release the guard we + * just set (mirrors the reconnect/configpush -EEXIST handling in + * ha.c) and fall back to running inline: unlike reconnect/configpush + * — which retry on the next heartbeat cycle — a demotion-convergence + * event is one-shot with no later trigger, and this is a + * safety-critical action (an ex-primary discarding provisional state + * before it can be trusted again), so it must never be silently + * dropped. */ + atomic_store_explicit(&c->ha.demotion_convergence_inflight, 0, + memory_order_release); + int rc = ha_demotion_convergence_do_run(c); + free(ctx); + return rc; + } + if (srv < 0 && srv != -EAGAIN) { + /* Hard submit failure (alloc / lane stopping). -EAGAIN is NOT a + * failure (still queued, will run). Same "never drop" reasoning as + * -EEXIST above. */ + WARN("HA demotion convergence: slowlane_submit failed: %d — running " + "inline (a one-shot safety action, never dropped)", srv); + atomic_store_explicit(&c->ha.demotion_convergence_inflight, 0, + memory_order_release); + int rc = ha_demotion_convergence_do_run(c); + free(ctx); + return rc; + } + /* 0 or -EAGAIN: enqueued exactly once; demotion_lane_fn frees ctx and + * clears the flag after the (production: non-returning) restart. */ + return 0; } /* diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index 24c68898..ae61fced 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -3958,6 +3958,235 @@ extern int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req); #endif /* XCE_TESTING */ +/* ═══════════════════════════════════════════════════════════════════════════ + * xce-audit-invariants controller pass, thread-role violation 2 fix + * (2026-07-18, the more urgent violation — catch-up can stream an unbounded + * amount of WAL data): ha_catchup_continuation_fn must not block the + * delaymgr thread — the SAME single thread that drives heartbeats and + * election timeouts cluster-wide (ha-async-io.md thread-role contract). + * + * These tests drive a REAL, STARTED delaymgr (single worker thread, exactly + * as production — delaymgr_t.worker is a single pthread, not a pool) so the + * scheduled one-shot continuation task is picked up and run BY that thread, + * proving whether it blocks. ha_repl_stream_wal_hook gates the blocking + * streaming call so the test controls exactly when it completes. A cheap + * PROBE task queued on the SAME delaymgr right after the continuation is + * scheduled stands in for "the next heartbeat/election timer" — whether it + * fires promptly is the load-bearing assertion. + * ═══════════════════════════════════════════════════════════════════════════ */ +#ifdef XCE_TESTING + +typedef struct { + pthread_mutex_t mu; + pthread_cond_t release_cv; + pthread_cond_t entered_cv; + int gate_open; + _Atomic int entered; + _Atomic int calls; +} cc_gate_t; + +static cc_gate_t g_cc_gate; + +static void cc_gate_init(cc_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 cc_gate_destroy(cc_gate_t *g) +{ + pthread_mutex_destroy(&g->mu); + pthread_cond_destroy(&g->release_cv); + pthread_cond_destroy(&g->entered_cv); +} + +static void cc_gate_open(cc_gate_t *g) +{ + pthread_mutex_lock(&g->mu); + g->gate_open = 1; + pthread_cond_broadcast(&g->release_cv); + pthread_mutex_unlock(&g->mu); +} + +static int cc_gate_wait_entered(cc_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; +} + +/* Stand-in for the blocking WAL stream (ha_stream_wal_to_standby): 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 delaymgr worker + * thread (flag OFF). */ +static int cc_gated_stream_hook(void *ep, const char *wal_path, + uint64_t standby_lsn, uint64_t committed_lsn) +{ + (void)ep; (void)wal_path; (void)standby_lsn; (void)committed_lsn; + atomic_fetch_add(&g_cc_gate.calls, 1); + + pthread_mutex_lock(&g_cc_gate.mu); + atomic_fetch_add(&g_cc_gate.entered, 1); + pthread_cond_broadcast(&g_cc_gate.entered_cv); + while (!g_cc_gate.gate_open) + pthread_cond_wait(&g_cc_gate.release_cv, &g_cc_gate.mu); + pthread_mutex_unlock(&g_cc_gate.mu); + return 0; +} + +/* Real, started delaymgr pass-through runner — production's run_delay_tasks + * (cluster.c) is not linked into controller_unit_test; mirrors + * test_commit_durability.c's dur_run_delay_task. */ +static int cc_run_delay_task(delay_task_entry_fn_t entry, void *arg) +{ + return entry(arg); +} + +static int g_cc_ep_placeholder; /* fake non-NULL repl_ep_t identity */ + +typedef struct { _Atomic int fired; } cc_probe_t; + +static int cc_probe_fn(void *arg) +{ + cc_probe_t *p = (cc_probe_t *)arg; + atomic_store(&p->fired, 1); + return 0; +} + +static int cc_poll_flag(_Atomic int *flag, int timeout_ms) +{ + for (int i = 0; i < timeout_ms; i++) { + if (atomic_load(flag)) + break; + struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */ + nanosleep(&ts, NULL); + } + return atomic_load(flag); +} + +static void cc_run_offload_probe(int offload_on, const char *label) +{ + char tmpdir[256]; + snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_cc_%s_XXXXXX", + offload_on ? "on" : "off"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); + + cluster_t c; + memset(&c, 0, sizeof(c)); + + persist_config_t pcfg = persist_config_defaults(); + pcfg.wal_sync_ms = 5000; + persist_set_workdir(&c.persist, tmpdir, "ctrl-test"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &pcfg, &c)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist)); + pthread_mutex_lock(&c.persist.lock); + c.persist.current_lsn = 11; /* tip = 10 */ + pthread_mutex_unlock(&c.persist.lock); + atomic_store(&c.persist.committed_lsn, 10); + atomic_store(&c.persist.durable_lsn, 10); + + TEST_ASSERT_TRUE(repl_add_standby(&c.persist.repl, &g_cc_ep_placeholder, + "ccsb0", 7600) >= 0); + + ha_config_t hcfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + }; + TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &hcfg, 0)); + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4)); + TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&c.delaymgr, 5, cc_run_delay_task)); + TEST_ASSERT_EQUAL_INT(0, delaymgr_start(&c.delaymgr)); + + cc_gate_init(&g_cc_gate); + ha_repl_stream_wal_hook = cc_gated_stream_hook; + ha_set_catchup_offload_for_test(offload_on); + + /* acked_lsn=0 < target(=10) -> schedules exactly one continuation. */ + TEST_ASSERT_EQUAL_INT(0, + ha_walack_continuation_for_test(&c, &g_cc_ep_placeholder, 0, 0)); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, cc_gate_wait_entered(&g_cc_gate, 1, 1000), + "the streaming call must reach the gate (on the slow lane, or — " + "RED — inline on the delaymgr thread)"); + + /* Queue a cheap probe task RIGHT AFTER, on the SAME single-worker + * delaymgr — a stand-in for the next heartbeat/election timer. */ + cc_probe_t probe; + atomic_store(&probe.fired, 0); + TEST_ASSERT_EQUAL_INT(0, + delay_task_ms(&c.delaymgr, 1, 1, cc_probe_fn, &probe)); + + if (offload_on) { + TEST_ASSERT_TRUE_MESSAGE(cc_poll_flag(&probe.fired, 1000), + "GREEN: with catch-up offloaded to the slow lane, the delaymgr " + "worker thread must be free to run the NEXT queued task " + "promptly, even while the streaming body is still parked"); + } else { + TEST_ASSERT_FALSE_MESSAGE(cc_poll_flag(&probe.fired, 300), + "RED: with offload OFF the streaming body runs INLINE on the " + "delaymgr worker thread, so the next queued task (standing in " + "for a heartbeat/election timer) cannot fire until it returns " + "-- this is the blocking-on-delaymgr shape the fix removes"); + } + + cc_gate_open(&g_cc_gate); + + /* Drain: wait for catchup_inflight[0] to clear (lane/inline body done), + * then confirm the probe (queued behind it in the OFF/RED case) fires. */ + for (int i = 0; i < 1000; i++) { + if (atomic_load_explicit(&c.ha.catchup_inflight[0], + memory_order_acquire) == 0) + break; + struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; + nanosleep(&ts, NULL); + } + TEST_ASSERT_TRUE_MESSAGE(cc_poll_flag(&probe.fired, 1000), + "the probe task must eventually fire once the gate is released"); + + ha_repl_stream_wal_hook = NULL; + ha_set_catchup_offload_for_test(0); /* restore production default */ + cc_gate_destroy(&g_cc_gate); + delaymgr_stop(&c.delaymgr); + delaymgr_fini(&c.delaymgr); + slowlane_fini(&c.slowlane); + ha_fini(&c.ha); + persist_stop(&c.persist); + persist_fini(&c.persist); + repl_fini(&c.persist.repl); + printf("%s: PASS\n", label); +} + +static void test_catchup_offload_probe_runs_promptly(void) +{ + cc_run_offload_probe(1, "test_catchup_offload_probe_runs_promptly"); +} + +static void test_catchup_inline_probe_blocks_RED(void) +{ + cc_run_offload_probe(0, "test_catchup_inline_probe_blocks_RED"); +} + +#endif /* XCE_TESTING */ + /* ── dispatch ───────────────────────────────────── */ void test_config_repl(void) @@ -4052,6 +4281,10 @@ void test_config_repl(void) RUN_TEST(test_snapshot_recv_reconnect_discards_snappart); /* Task 5.1: handle_ctrlhello connection setup only — no repl messages */ RUN_TEST(test_ctrlhello_no_repl_messages_enqueued); + /* xce-audit-invariants controller pass, thread-role violation 2 fix + * (2026-07-18): catch-up continuation slow-lane offload */ + RUN_TEST(test_catchup_offload_probe_runs_promptly); + RUN_TEST(test_catchup_inline_probe_blocks_RED); #endif /* Task A: raw capnp shims */ RUN_TEST(test_replconfig_req_ssl_encode); diff --git a/src/controller/testsuite/test_demotion_convergence.c b/src/controller/testsuite/test_demotion_convergence.c index a821c53b..37382d34 100644 --- a/src/controller/testsuite/test_demotion_convergence.c +++ b/src/controller/testsuite/test_demotion_convergence.c @@ -58,10 +58,12 @@ #include "delay.h" /* delaymgr_t, delay_task_t */ #include "ha.h" /* raft_cb_become_follower_with_cluster_for_test, * ha_self_restart, controller_saved_argv, - * ha_restart_hook / ha_fatal_hook seams */ + * ha_restart_hook / ha_fatal_hook seams, + * ha_set_demotion_offload_for_test */ #include "ordq.h" /* ordq_item_t (drive the scheduled task) */ #include "persist.h" /* persist_t, persist_truncate_wal_above */ #include "repl.h" /* repl_add_standby / repl_set_commit_engine */ +#include "slowlane.h" /* slowlane_init/slowlane_fini (offload lane) */ #include "xce/cluster_def.h" /* full cluster_t layout */ #include "xce/errnox.h" /* XCE_ERR_STANDBY (task 1.4 resolve rc) */ @@ -196,6 +198,19 @@ static void dc_setup(const char *suffix, int n_peers, int n_standbys, * by dc_run_pending_task (D1: truncation + restart run off the callback). */ TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&g_dc.delaymgr, 100, NULL)); + /* Slow lane for the demotion-convergence offload (xce-audit-invariants + * controller pass, thread-role violation 1 fix, 2026-07-18). The + * offload flag defaults OFF (see g_drive_async_demotion_offload in + * ha_convergence.c), so most tests using this fixture still take the + * legacy inline path unchanged; the dedicated offload tests below flip + * the flag ON explicitly, at which point ha_demotion_convergence_fn + * hands the truncate + self-restart off to a REAL lane worker thread + * here, so dc_run_pending_task returns promptly and the convergence + * effects (truncation, restart-seam call) land asynchronously shortly + * after — tests poll for them via dc_poll_restart_calls below + * regardless of which path is active (harmless when already-complete). */ + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&g_dc.slowlane, 2, 4)); + /* Make this a PRIMARY with hooks registered, so a demotion has the three * persistence hooks to disable (Req 1.1). ha_promote_to_primary runs the * shared promotion sequence (ha_do_promote → register hooks → heartbeat → @@ -214,6 +229,7 @@ static void dc_teardown(void) repl_broadcast_send_hook = NULL; ha_restart_hook = NULL; ha_fatal_hook = NULL; + slowlane_fini(&g_dc.slowlane); commit_engine_fini(&g_dc.commit); delaymgr_fini(&g_dc.delaymgr); persist_stop(&g_dc.persist); @@ -223,6 +239,24 @@ static void dc_teardown(void) /* tmpdir left on disk — /tmp is ephemeral */ } +/* Bounded poll until g_restart_calls reaches `want` (or the timeout elapses). + * When the demotion offload flag is ON, the truncate + restart-seam call run + * asynchronously on a real slow-lane worker thread after dc_run_pending_task + * returns, so tests must poll rather than assert immediately; with the flag + * OFF (current default) the call has already completed synchronously by the + * time dc_run_pending_task returns, so this poll returns on its first + * iteration — safe either way. */ +static int dc_poll_restart_calls(int want, int timeout_ms) +{ + for (int i = 0; i < timeout_ms; i++) { + if (atomic_load(&g_restart_calls) >= want) + break; + struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */ + nanosleep(&ts, NULL); + } + return atomic_load(&g_restart_calls); +} + /* Count pending delaymgr tasks by walking the ordq head chain (read-only; * single-threaded test, delaymgr never started). Same seam the lifecycle * suite uses to assert promotion arms no watchdog. */ @@ -347,25 +381,36 @@ static void test_demotion_schedules_convergence_truncate_and_restart(void) "demotion from PRIMARY must schedule exactly one convergence task " "(truncate+restart off the callback thread, D1)"); - /* Run the scheduled convergence task synchronously. */ + /* Run the scheduled convergence task. If the offload flag is ON this + * only submits to the slow lane and returns fast — the truncate + + * restart-seam call complete asynchronously on a real lane worker + * thread shortly after — so poll below rather than assert immediately + * (also safe, and immediate, when the flag is OFF, the current + * default). */ int rc = dc_run_pending_task(&g_dc.delaymgr); TEST_ASSERT_NOT_EQUAL_MESSAGE(DC_NO_TASK, rc, "no convergence task was scheduled on the demotion edge (Req 1.2/1.3)"); - /* Assert (Req 1.2/1.5): truncation removed everything above committed_lsn — - * current_lsn collapses to committed_lsn + 1 (the truncation boundary). */ - TEST_ASSERT_EQUAL_UINT64_MESSAGE( - atomic_load(&g_dc.persist.committed_lsn) + 1, g_dc.persist.current_lsn, - "demotion convergence must truncate above committed_lsn (Req 1.2): " - "current_lsn must collapse to committed_lsn + 1"); - - /* Assert (Req 1.3/1.6): the restart seam was invoked exactly once. */ - TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls), + /* Assert (Req 1.3/1.6): the restart seam was invoked exactly once + * (bounded poll — the lane worker runs asynchronously). */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, dc_poll_restart_calls(1, 2000), "demotion convergence must invoke the restart seam exactly once " "after truncation (Req 1.3, 1.6)"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_fatal_calls), "a successful restart must NOT invoke the fatal seam"); + /* Assert (Req 1.2/1.5): truncation removed everything above committed_lsn — + * current_lsn collapses to committed_lsn + 1 (the truncation boundary). + * Safe to read now (no lock): observing g_restart_calls==1 above + * (seq_cst atomic) happens-after the lane worker's truncate, which + * happens-before its restart-seam call in program order on that thread — + * the same "atomic completion flag" idiom used throughout this suite + * (e.g. test_ha_dispatch_integration.c's poll_atomic_ge). */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE( + atomic_load(&g_dc.persist.committed_lsn) + 1, g_dc.persist.current_lsn, + "demotion convergence must truncate above committed_lsn (Req 1.2): " + "current_lsn must collapse to committed_lsn + 1"); + dc_teardown(); printf("test_demotion_schedules_convergence_truncate_and_restart: checked\n"); } @@ -700,11 +745,18 @@ static void test_demotion_replay_yields_committed_only(void) ha_fatal_hook = dc_fatal_hook; g_restart_hook_rc = 0; - /* Act: run the demotion seam (schedules + we drive the convergence task). */ + /* Act: run the demotion seam (schedules + we drive the convergence task). + * If the offload flag is ON the truncate itself completes asynchronously + * on a real slow-lane worker thread shortly after dc_run_pending_task + * returns, so wait for the restart seam (its + * last action) before reading the truncate's effect below. */ raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/7); int rc = dc_run_pending_task(&g_dc.delaymgr); TEST_ASSERT_NOT_EQUAL_MESSAGE(DC_NO_TASK, rc, "demotion must schedule a convergence task that truncates (Req 1.2)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, dc_poll_restart_calls(1, 2000), + "the convergence task must truncate then invoke the restart seam " + "(bounded poll — the lane worker runs asynchronously)"); /* Final assertion — both branches prove the SAME property: the demotion seam * truncated the uncommitted suffix down to the committed boundary @@ -770,6 +822,170 @@ static void test_demotion_replay_yields_committed_only(void) printf("test_demotion_replay_yields_committed_only: checked\n"); } +/* ═══════════════════════════════════════════════════════════════════════════ + * xce-audit-invariants controller pass, thread-role violation 1 fix + * (2026-07-18): ha_demotion_convergence_fn must not block the delaymgr + * thread — the SAME single thread that drives heartbeats and election + * timeouts cluster-wide (ha-async-io.md thread-role contract). This test + * drives dc_run_pending_task (this suite's existing stand-in for "the + * delaymgr worker's turn to run this task") on a SEPARATE thread and asserts + * it returns promptly with the offload flag ON, but stalls with it OFF + * (RED) — proving the offload closes the violation. ha_restart_hook is + * installed as the park point: it sits at the very end of the (inline or + * lane) blocking body, so parking there proves the WHOLE body — truncate + * included — has not yet returned control to whichever thread called it. + * ═══════════════════════════════════════════════════════════════════════════ */ + +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; +} dcb_gate_t; + +static dcb_gate_t g_dcb_gate; + +static void dcb_gate_init(dcb_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); +} + +static void dcb_gate_destroy(dcb_gate_t *g) +{ + pthread_mutex_destroy(&g->mu); + pthread_cond_destroy(&g->release_cv); + pthread_cond_destroy(&g->entered_cv); +} + +static void dcb_gate_open(dcb_gate_t *g) +{ + pthread_mutex_lock(&g->mu); + g->gate_open = 1; + pthread_cond_broadcast(&g->release_cv); + pthread_mutex_unlock(&g->mu); +} + +/* Restart-hook stand-in: park at the gate (simulating the tail of the + * blocking convergence body — truncate has already run by the time this + * fires), then report success. */ +static int dcb_gated_restart_hook(cluster_t *c) +{ + (void)c; + atomic_fetch_add(&g_restart_calls, 1); + pthread_mutex_lock(&g_dcb_gate.mu); + atomic_fetch_add(&g_dcb_gate.entered, 1); + pthread_cond_broadcast(&g_dcb_gate.entered_cv); + while (!g_dcb_gate.gate_open) + pthread_cond_wait(&g_dcb_gate.release_cv, &g_dcb_gate.mu); + pthread_mutex_unlock(&g_dcb_gate.mu); + return 0; +} + +static int dcb_gate_wait_entered(dcb_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; +} + +/* Worker thread standing in for "the delaymgr worker running this task". */ +typedef struct { delaymgr_t *mgr; _Atomic int done; int rc; } dcb_run_arg_t; +static void *dcb_run_thread(void *p) +{ + dcb_run_arg_t *a = (dcb_run_arg_t *)p; + a->rc = dc_run_pending_task(a->mgr); + atomic_store(&a->done, 1); + return NULL; +} + +static void dcb_run_offload(int offload_on, const char *label) +{ + dc_setup("blocking", /*n_peers=*/3, /*n_standbys=*/2, + /*appended_lsn=*/5, /*n_vcpus=*/8); + atomic_store(&g_dc.persist.committed_lsn, 3); + + dcb_gate_init(&g_dcb_gate); + ha_restart_hook = dcb_gated_restart_hook; + ha_fatal_hook = dc_fatal_hook; + ha_set_demotion_offload_for_test(offload_on); + + /* Schedule the convergence task (does not run it yet). */ + raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/7); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, dc_task_count(&g_dc.delaymgr), + "demotion must schedule exactly one convergence task before this " + "test can drive it"); + + /* Run the scheduled task on a SEPARATE thread — this suite's existing + * stand-in for the delaymgr worker's turn (dc_run_pending_task). */ + dcb_run_arg_t arg = { .mgr = &g_dc.delaymgr }; + atomic_store(&arg.done, 0); + pthread_t th; + pthread_create(&th, NULL, dcb_run_thread, &arg); + + 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, + "GREEN: with demotion convergence offloaded to the slow lane, " + "the task body (standing in for the delaymgr thread) must " + "return promptly even while the restart-seam tail is still " + "parked on the lane"); + /* The blocking work was actually reached — just on the lane, not + * the calling thread. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, + dcb_gate_wait_entered(&g_dcb_gate, 1, 1000), + "GREEN: the offloaded convergence body must still run (on the " + "lane) and reach the restart-seam park point"); + } else { + TEST_ASSERT_EQUAL_INT_MESSAGE(0, done, + "RED: with offload OFF the convergence body runs INLINE, so the " + "calling (delaymgr-standin) thread blocks at the restart-seam " + "gate — this is the blocking-on-delaymgr shape the fix removes"); + } + + dcb_gate_open(&g_dcb_gate); + pthread_join(th, NULL); + /* Let the async completion (offload ON) finish before teardown. */ + dc_poll_restart_calls(1, 2000); + + ha_set_demotion_offload_for_test(0); /* restore production default */ + dcb_gate_destroy(&g_dcb_gate); + dc_teardown(); + printf("%s: PASS\n", label); +} + +static void test_demotion_offload_task_returns_promptly(void) +{ + dcb_run_offload(1, "test_demotion_offload_task_returns_promptly"); +} + +static void test_demotion_inline_task_blocks_RED(void) +{ + dcb_run_offload(0, "test_demotion_inline_task_blocks_RED"); +} + /* ════════════════════════════════════════════════════════════════════════════ * Rejoin reconciliation (ha-convergence-recovery task 3.1 — RED) * @@ -988,6 +1204,12 @@ void test_demotion_convergence(void) RUN_TEST(test_demotion_from_init_no_truncation_no_restart); RUN_TEST(test_demotion_replay_yields_committed_only); + printf("\n>> demotion-convergence slow-lane offload tests " + "(xce-audit-invariants controller pass, 2026-07-18)\n\n"); + + RUN_TEST(test_demotion_offload_task_returns_promptly); + RUN_TEST(test_demotion_inline_task_blocks_RED); + printf("\n>> rejoin reconciliation tests (RED, task 3.1)\n\n"); RUN_TEST(test_rejoin_no_watermark_truncate_no_restart);