fix(controller): bug-1450/1451 offload delaymgr-inline blocking work to the slow lane
/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:
parent
aeb6e37ce7
commit
f88e4ef431
5 changed files with 877 additions and 55 deletions
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue