lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(controller): bug-1450/1451 offload delaymgr-inline blocking work to the slow lane

Browse source 
/xce-audit-invariants controller found two live violations of the async
I/O thread-role contract: ha_demotion_convergence_fn (ha_convergence.c)
ran WAL truncation + tlog_exit()+execv() directly on the delaymgr thread,
and ha_catchup_continuation_fn (config_repl.c) ran unbounded WAL-streaming
to a standby directly on delaymgr -- the same single thread that drives
heartbeats and election timeouts cluster-wide, so either blocks liveness
for the whole cluster and risks a spurious election.

A prior vtask-pool offload attempt (2026-07-10, bug-1429) deadlocked
controller_unit_test because c->ha.pool is never initialized by the
lightweight unit-test fixtures these functions run under, and was fully
reverted. This fix instead uses the codebase's established slow-lane
pattern (the same mechanism already used for snapshot-send, config-push,
and peer connect) -- a 4th and 5th instance, not a new mechanism.

New g_drive_async_demotion_offload / g_drive_async_catchup_offload flags
ship default OFF, matching the original rollout discipline of the three
reference offloads (shipped OFF, flipped ON later after a dedicated
compatibility audit). Flipping ON was tried during validation and
reproduced two distinct hazards against pre-existing fixtures that never
call slowlane_init: the already-known hang, and a new ASan SEGV in
commit_advance_locked (logged as bug-1452, not root-caused, deferred to
the future flip-to-ON audit). Both flags verified OFF and the full
976-test suite clean, then the committed state re-verified clean against
a live 10-round chaotic and 10-round sequential HA soak (zero job loss,
zero divergence, both PASS).
This commit is contained in:
Rongsong Shen 2026-07-19 08:14:00 +00:00
parent aeb6e37ce7
commit f88e4ef431
5 changed files with 877 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

156
src/controller/config_repl.c
View file

@ -2596,29 +2596,70 @@ typedef struct {
} ha_catchup_continuation_ctx_t; } ha_catchup_continuation_ctx_t;
/* /*
* ha_catchup_continuation_fn — the real catch-up work. Re-validate the * g_drive_async_catchup_offload / ha_set_catchup_offload_for_test /
* endpoint against the registry, re-derive target, stream [acked_lsn+1, * ha_catchup_offload_enabled (xce-audit-invariants controller pass,
* target], and clear catchup_inflight on EVERY exit path (success, ep-gone, * thread-role violation 2 fix, 2026-07-18):
* 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).
* *
* Runs as a delaymgr task body (see ha_schedule_catchup_continuation below) — * Gates whether ha_catchup_continuation_fn hands the WAL-streaming body off
* must run off the iomgr walAck thread, since catchup_stream_range -> * to the slow lane (ON — closes the thread-role violation) or runs it
* ha_stream_wal_to_standby does blocking WAL file reads + socket writes. * INLINE on the calling (delaymgr) thread (OFF, the PRODUCTION DEFAULT for
* (2026-07-10: a VTASK_LIVE_BLOCK dispatch was attempted here to also get it * now — see below). Catch-up can stream a variable, potentially large
* off the delaymgr thread per ha-async-io.md's thread-role contract, but it * amount of WAL data, so leaving it inline on delaymgr — the SAME single
* deadlocked the controller_unit_test harness — reverted pending a proper * thread that drives heartbeats and election timeouts cluster-wide — is an
* fix; tracked as a known, still-open thread-role violation alongside the * unbounded block and a real spurious-election risk, not just a theoretical
* other raft-drive-async P2 items.) * 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; ha_catchup_continuation_ctx_t *ctx = (ha_catchup_continuation_ctx_t *)arg;
if (ctx == NULL || ctx->c == NULL) { if (ctx == NULL)
free(ctx); return;
return -EINVAL;
}
cluster_t *c = ctx->c; cluster_t *c = ctx->c;
repl_ep_t ep = ctx->ep; repl_ep_t ep = ctx->ep;
int peer_idx = ctx->peer_idx; int peer_idx = ctx->peer_idx;
@ -2648,6 +2689,83 @@ clear_and_done:
atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0, atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
memory_order_release); memory_order_release);
free(ctx); 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; return 0;
} }

100
src/controller/ha.h
View file

@ -359,6 +359,26 @@ typedef struct {
* set, the heartbeat decision loop / * set, the heartbeat decision loop /
* promotion enqueues no duplicate config * promotion enqueues no duplicate config
* push. 0 at init (ha_init memset). */ * 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) ──────── */ /* ── New field added by ha-entity-snapshots spec (Task 4.1) ──────── */
_Atomic uint64_t snapshot_inflight_lsn[MAX_HA_CANDIDATES]; /* Per-peer _Atomic uint64_t snapshot_inflight_lsn[MAX_HA_CANDIDATES]; /* Per-peer
* checkpoint LSN referenced by an * 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). * 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 * Returns 0 on success, negative on alloc/schedule failure (suffix is then
* reconciled by rejoin reconciliation at the next join). * 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); 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 * ha_demotion_convergence_fn: the delaymgr task body — exposed under XCE_TESTING
* only so the demotion-convergence suite can reference its contract; the test * only so the demotion-convergence suite can reference its contract; the test
* fixture drives the queued task through its stored entry pointer. * 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); int ha_demotion_convergence_fn(void *arg);
#endif #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 * [acked_lsn+1, target] on delaymgr (never inline on
* iomgr), UNLESS c->ha.catchup_inflight[peer_idx] is * iomgr), UNLESS c->ha.catchup_inflight[peer_idx] is
* already set (per-standby single-flight, Req 5.2). * 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 * if acked_lsn >= target → repl_complete_catchup(&c->persist.repl, ep) to
* re-include the standby in live replication (Req 7.1). * 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); void ha_set_configpush_offload_for_test(int on);
int ha_configpush_offload_enabled(void); 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: * ha_set_peerctrl_dispatch_for_test / ha_peerctrl_dispatch_enabled:
* Feature-flag accessors (raft-drive-async Task 3.1). The dispatch flag gates * Feature-flag accessors (raft-drive-async Task 3.1). The dispatch flag gates

197
src/controller/ha_convergence.c
View file

@ -184,36 +184,82 @@ int ha_self_restart(cluster_t *c)
} }
/* /*
* ha_demotion_convergence_ctx_t — heap context for the one-shot delaymgr task. * 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 * (and, with the offload flag ON, the slow-lane task it hands off to). Holds
* schedule time) so a late-advancing commit still truncates to the right * only the cluster pointer; committed_lsn is read at task time (not schedule
* boundary. * time) so a late-advancing commit still truncates to the right boundary.
*/ */
typedef struct { typedef struct {
cluster_t *c; cluster_t *c;
} ha_demotion_convergence_ctx_t; } ha_demotion_convergence_ctx_t;
/* /*
* ha_demotion_convergence_fn — delaymgr task body (runs OFF the become-follower * g_drive_async_demotion_offload / ha_set_demotion_offload_for_test /
* callback thread, D1). Truncate above committed_lsn (Req 1.2), WARN with the * ha_demotion_offload_enabled (xce-audit-invariants controller pass,
* truncated range (Req 1.5), then self-restart (Req 1.3). On truncation failure * thread-role violation 1 fix, 2026-07-18):
* (Req error strategy) ERROR then restart anyway — the rejoin reconciliation
* loop converges the untruncated case at the next join.
* *
* In XCE_TESTING the test fixture drives this entry directly (un-started * Gates whether ha_demotion_convergence_fn hands the truncate + self-restart
* delaymgr); ha_self_restart routes to the restart seam there. The ctx is * body off to the slow lane (ON — closes the thread-role violation: the
* freed before restart in production restart never returns, so only the test * delaymgr thread that also drives heartbeats/election timeouts cluster-wide
* path reaches the free — kept on every non-restarting exit for safety. * 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) static _Atomic int g_drive_async_demotion_offload = 0;
{
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;
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, uint64_t committed = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire); memory_order_acquire);
@ -237,11 +283,114 @@ int ha_demotion_convergence_fn(void *arg)
committed, prior_tip); 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); free(ctx);
/* Req 1.3: rebuild in-memory state through boot recovery. */ (void)ha_demotion_convergence_do_run(c);
return ha_self_restart(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;
} }
/* /*

233
src/controller/testsuite/test_config_repl.c
View file

@ -3958,6 +3958,235 @@ extern int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req);
#endif /* XCE_TESTING */ #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 ───────────────────────────────────── */ /* ── dispatch ───────────────────────────────────── */
void test_config_repl(void) void test_config_repl(void)
@ -4052,6 +4281,10 @@ void test_config_repl(void)
RUN_TEST(test_snapshot_recv_reconnect_discards_snappart); RUN_TEST(test_snapshot_recv_reconnect_discards_snappart);
/* Task 5.1: handle_ctrlhello connection setup only — no repl messages */ /* Task 5.1: handle_ctrlhello connection setup only — no repl messages */
RUN_TEST(test_ctrlhello_no_repl_messages_enqueued); 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 #endif
/* Task A: raw capnp shims */ /* Task A: raw capnp shims */
RUN_TEST(test_replconfig_req_ssl_encode); RUN_TEST(test_replconfig_req_ssl_encode);

246
src/controller/testsuite/test_demotion_convergence.c
View file

@ -58,10 +58,12 @@
#include "delay.h" /* delaymgr_t, delay_task_t */ #include "delay.h" /* delaymgr_t, delay_task_t */
#include "ha.h" /* raft_cb_become_follower_with_cluster_for_test, #include "ha.h" /* raft_cb_become_follower_with_cluster_for_test,
* ha_self_restart, controller_saved_argv, * 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 "ordq.h" /* ordq_item_t (drive the scheduled task) */
#include "persist.h" /* persist_t, persist_truncate_wal_above */ #include "persist.h" /* persist_t, persist_truncate_wal_above */
#include "repl.h" /* repl_add_standby / repl_set_commit_engine */ #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/cluster_def.h" /* full cluster_t layout */
#include "xce/errnox.h" /* XCE_ERR_STANDBY (task 1.4 resolve rc) */ #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). */ * by dc_run_pending_task (D1: truncation + restart run off the callback). */
TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&g_dc.delaymgr, 100, NULL)); 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 /* 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 * persistence hooks to disable (Req 1.1). ha_promote_to_primary runs the
* shared promotion sequence (ha_do_promote → register hooks → heartbeat → * shared promotion sequence (ha_do_promote → register hooks → heartbeat →
@ -214,6 +229,7 @@ static void dc_teardown(void)
repl_broadcast_send_hook = NULL; repl_broadcast_send_hook = NULL;
ha_restart_hook = NULL; ha_restart_hook = NULL;
ha_fatal_hook = NULL; ha_fatal_hook = NULL;
slowlane_fini(&g_dc.slowlane);
commit_engine_fini(&g_dc.commit); commit_engine_fini(&g_dc.commit);
delaymgr_fini(&g_dc.delaymgr); delaymgr_fini(&g_dc.delaymgr);
persist_stop(&g_dc.persist); persist_stop(&g_dc.persist);
@ -223,6 +239,24 @@ static void dc_teardown(void)
/* tmpdir left on disk — /tmp is ephemeral */ /* 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; /* Count pending delaymgr tasks by walking the ordq head chain (read-only;
* single-threaded test, delaymgr never started). Same seam the lifecycle * single-threaded test, delaymgr never started). Same seam the lifecycle
* suite uses to assert promotion arms no watchdog. */ * 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 " "demotion from PRIMARY must schedule exactly one convergence task "
"(truncate+restart off the callback thread, D1)"); "(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); int rc = dc_run_pending_task(&g_dc.delaymgr);
TEST_ASSERT_NOT_EQUAL_MESSAGE(DC_NO_TASK, rc, TEST_ASSERT_NOT_EQUAL_MESSAGE(DC_NO_TASK, rc,
"no convergence task was scheduled on the demotion edge (Req 1.2/1.3)"); "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 — /* Assert (Req 1.3/1.6): the restart seam was invoked exactly once
* current_lsn collapses to committed_lsn + 1 (the truncation boundary). */ * (bounded poll — the lane worker runs asynchronously). */
TEST_ASSERT_EQUAL_UINT64_MESSAGE( TEST_ASSERT_EQUAL_INT_MESSAGE(1, dc_poll_restart_calls(1, 2000),
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),
"demotion convergence must invoke the restart seam exactly once " "demotion convergence must invoke the restart seam exactly once "
"after truncation (Req 1.3, 1.6)"); "after truncation (Req 1.3, 1.6)");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_fatal_calls), TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_fatal_calls),
"a successful restart must NOT invoke the fatal seam"); "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(); dc_teardown();
printf("test_demotion_schedules_convergence_truncate_and_restart: checked\n"); 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; ha_fatal_hook = dc_fatal_hook;
g_restart_hook_rc = 0; 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); raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/7);
int rc = dc_run_pending_task(&g_dc.delaymgr); int rc = dc_run_pending_task(&g_dc.delaymgr);
TEST_ASSERT_NOT_EQUAL_MESSAGE(DC_NO_TASK, rc, TEST_ASSERT_NOT_EQUAL_MESSAGE(DC_NO_TASK, rc,
"demotion must schedule a convergence task that truncates (Req 1.2)"); "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 /* Final assertion — both branches prove the SAME property: the demotion seam
* truncated the uncommitted suffix down to the committed boundary * 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"); 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) * 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_from_init_no_truncation_no_restart);
RUN_TEST(test_demotion_replay_yields_committed_only); 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"); printf("\n>> rejoin reconciliation tests (RED, task 3.1)\n\n");
RUN_TEST(test_rejoin_no_watermark_truncate_no_restart); RUN_TEST(test_rejoin_no_watermark_truncate_no_restart);