diff --git a/.kiro/specs/raft-drive-async/tasks.md b/.kiro/specs/raft-drive-async/tasks.md index 17532833..d523fcc9 100644 --- a/.kiro/specs/raft-drive-async/tasks.md +++ b/.kiro/specs/raft-drive-async/tasks.md @@ -103,8 +103,10 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - _Boundary: PeerCtrlDispatch, SlowLane_ - _Depends: 3.1_ -- [ ] 3.4 Receiver-side integration tests - - **3.4a (DONE)**: bug-1197 endpoint-quiescence fix landed so flag-ON peer-control dispatch is UAF-free. Production flags STILL DEFAULT OFF (per the "full fix, defer prod flip" decision). The prod flip happens only after 4.1 partition soak + 4.2 regression sweep pass. +- [x] 3.4 Receiver-side integration tests + - **3.4a (DONE, committed 6d510d8)**: bug-1197 endpoint-quiescence fix landed so flag-ON peer-control dispatch is UAF-free (deferred teardown-vtask + RCU dispatch_inflight drain). + - **3.4b (DONE)**: receiver-side liveness integration tests (`test_ha_dispatch_integration.c`) with the dispatch flags flipped ON in-test — P1 a parked snapshot/config-push lane transfer does not block another endpoint's peer-control on the pool; P2 a VTASK_LIVENESS ack runs under REQ_BLOCK pool saturation (reservation never starves it; one-over fast-rejects to prove the boundary); P3 cross-endpoint peer-control runs concurrently (per-ep FIFO does not serialize across eps). Fixed a flaky teardown-race hang (bug-1224: poll end-of-body completion counters before the cancel-based vtaskmgr_fini). 5/5 deterministic, full controller suite EXIT=0, ASan-clean. + - **Production flags STILL DEFAULT OFF** (peerctrl/snapshot/configpush). Per the "full fix, defer prod flip" decision, the production flip happens only after 4.1 partition soak + 4.2 regression sweep pass; the snapshot/configpush lane borrowed-ep refcount (distinct from 3.4a) is also required before flipping those two ON. - Assert that during a snapshot/config-push transfer the controller still processes other peers' heartbeats and votes; that a quorum of WAL-acks advances the commit position promptly even with the vtask pool near saturation; and that cross-endpoint acks are not serialized behind one another. - Observable: tests pass; commit advances after a quorum of acks under load; iomgr stays responsive during transfers. - _Requirements: 1.1, 1.3, 1.4, 4.3, 4.5_ diff --git a/.wolf/buglog.json b/.wolf/buglog.json index 0a0df4dd..c31f6504 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -20224,6 +20224,61 @@ "related_bugs": [], "occurrences": 1, "last_seen": "2026-06-26T17:12:06.544Z" + }, + { + "id": "bug-1222", + "timestamp": "2026-06-26T17:44:25.670Z", + "error_message": "Incorrect value in code", + "file": "src/controller/meson.build", + "root_cause": "Had 'testsuite/test_waldump.c'", + "fix": "Changed to 'testsuite/test_ha_dispatch_integration.c'", + "tags": [ + "auto-detected", + "wrong-value", + "build" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T17:44:25.670Z" + }, + { + "id": "bug-1223", + "timestamp": "2026-06-27T00:33:17.036Z", + "error_message": "Significant refactor of ", + "file": "src/controller/testsuite/test_ha_dispatch_integration.c", + "root_cause": "2 lines replaced/restructured", + "fix": "Rewrote 5\u21923 lines (2 removed) | Also: /* Release ep_A so its worker finishes before the ; TEST_ASSERT_EQUAL_INT_MESSAGE(2, poll_atomic_ge(&g", + "tags": [ + "auto-detected", + "refactor", + "c" + ], + "related_bugs": [], + "occurrences": 2, + "last_seen": "2026-06-27T00:34:00.682Z" + }, + { + "id": "bug-1224", + "timestamp": "2026-06-26", + "error_message": "raft-drive-async 3.4b integration tests (test_ha_dispatch_integration.c) hang intermittently \u2014 controller_unit_test never completes (timeout-killed)", + "file": "src/controller/testsuite/test_ha_dispatch_integration.c", + "root_cause": "Test teardown race: P1/P2/P3 released a parked vtask body (gate_open) then immediately called vtaskmgr_fini, which tears down workers via pthread_cancel. The tests polled a run-counter incremented MID-body, so fini raced the just-released body still unwinding (esp. P2 REQ_BLOCK bodies in vtaskmgr_blocking_release racing blk_lock teardown) \u2192 flaky deadlock. slowlane_fini is graceful (join) so the lane side was safe; vtaskmgr_fini (cancel) was the hazard.", + "fix": "Add end-of-body completion counters (g_pc_completed on both pc_probe paths incl. the park path; g_block_done at the end of block_task_entry) and poll them to the full expected count AFTER gate_open and BEFORE vtaskmgr_fini, so every pool body has fully returned (worker idle in cbuf_get) before the cancel-based fini. Verified deterministic 5/5 isolated runs + full suite EXIT=0, ASan-clean.", + "tags": [ + "raft-drive-async", + "task-3.4b", + "test", + "flaky", + "vtaskmgr_fini", + "pthread_cancel", + "teardown-race", + "deadlock" + ], + "related_bugs": [ + "bug-924" + ], + "occurrences": 1, + "last_seen": "2026-06-26" } ] } \ No newline at end of file diff --git a/.wolf/memory.md b/.wolf/memory.md index fc7f2f34..698a4bd6 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -3516,3 +3516,28 @@ | 10:35 | Edited src/controller/ha.c | modified NOTE() | ~147 | | 10:36 | Edited .kiro/specs/raft-drive-async/tasks.md | 1→2 lines | ~121 | | 10:38 | raft-drive-async 3.4a: bug-1197 endpoint-quiescence (deferred teardown-vtask + RCU dispatch_inflight drain); flags STILL OFF | netio.h, netio.c, ha.c, ha.h, test_ha_vep_teardown.c | xcelib 282/0, ctrl EXIT=0, 5 vep tests ASan-clean, cpp-review GO race-free | ~75k | +| 10:44 | Created src/controller/testsuite/test_ha_dispatch_integration.c | — | ~6396 | +| 10:44 | Edited src/controller/testsuite/controller_test.c | modified setUp() | ~32 | +| 10:44 | Edited src/controller/testsuite/controller_test.c | 3→4 lines | ~26 | +| 10:44 | Edited src/controller/meson.build | 2→3 lines | ~31 | + +## Session: 2026-06-27 17:10 + +| Time | Action | File(s) | Outcome | ~Tokens | +|------|--------|---------|---------|--------| + +## Session: 2026-06-27 17:10 + +| Time | Action | File(s) | Outcome | ~Tokens | +|------|--------|---------|---------|--------| +| 17:29 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | modified pc_probe() | ~62 | +| 17:29 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | modified snap_probe() | ~61 | +| 17:33 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | fprintf() → vtaskmgr_fini() | ~207 | +| 17:33 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | 5→3 lines | ~20 | +| 17:33 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | modified block_task_entry() | ~212 | +| 17:33 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | 2→3 lines | ~28 | +| 17:33 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | expanded (+7 lines) | ~193 | +| 17:34 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | 4→7 lines | ~123 | +| 17:34 | Edited src/controller/testsuite/test_ha_dispatch_integration.c | 5→9 lines | ~144 | +| 17:25 | raft-drive-async 3.4b: receiver-side liveness integration tests (P1 transfer-non-block, P2 ack-under-saturation, P3 cross-ep concurrency); fixed flaky teardown-race hang | test_ha_dispatch_integration.c, controller_test.c, meson.build | 4 P-tests + 5 vep PASS, full suite EXIT=0, 5/5 deterministic | ~120k | +| 17:36 | Edited .kiro/specs/raft-drive-async/tasks.md | 2→4 lines | ~366 | diff --git a/src/controller/meson.build b/src/controller/meson.build index 6590e6e4..f8249058 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -144,6 +144,7 @@ controller_unit_test_src = [ 'testsuite/test_slowlane.c', 'testsuite/test_commit_admission_pool.c', 'testsuite/test_ha_vep_teardown.c', + 'testsuite/test_ha_dispatch_integration.c', 'testsuite/test_waldump.c', 'testsuite/waldump.c', 'testsuite/test_walcompare.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index a5e660fa..b874878f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -70,6 +70,7 @@ extern void test_drive_async_config(void); extern void test_slowlane(void); extern void test_commit_admission_pool(void); extern void test_ha_vep_teardown(void); +extern void test_ha_dispatch_integration(void); void setUp(void) { return; } @@ -209,6 +210,7 @@ int main(int argc, char* argv[]) TEST(test_slowlane); TEST(test_commit_admission_pool); TEST(test_ha_vep_teardown); + TEST(test_ha_dispatch_integration); return UNITY_END(); } diff --git a/src/controller/testsuite/test_ha_dispatch_integration.c b/src/controller/testsuite/test_ha_dispatch_integration.c new file mode 100644 index 00000000..dbcf559b --- /dev/null +++ b/src/controller/testsuite/test_ha_dispatch_integration.c @@ -0,0 +1,606 @@ +/* + * test_ha_dispatch_integration.c — raft-drive-async Task 3.4b: + * receiver-side liveness integration tests for the new dispatch flags. + * + * With the dispatch flags ON, prove the three receiver-side liveness + * properties the spec requires: + * + * (P1) TRANSFER-DOES-NOT-BLOCK-CROSS-ENDPOINT — while a snapshot (or + * config-push) transfer is parked in flight on the SLOW LANE, the + * controller still processes ANOTHER peer's peer-control message on the + * vtask POOL promptly: the transfer (lane) and the drive path (pool) are + * separate executors, so the offloaded transfer never blocks peer + * control. + * + * (P2) ACK-ADVANCES-UNDER-SATURATION — a peer-control ack is classified + * VTASK_LIVENESS (peer_ctrl_classify returns VTASK_LIVENESS for ALL + * peer-control: heartbeat/vote/ack) and the LivenessAdmission + * reservation keeps it from being starved by *_BLOCK tasks: with the + * pool saturated to the reservation boundary (cap = n_vcpus − + * VTASK_LIVENESS_RESERVED REQ_BLOCK tasks parked, one more fast-rejected) + * the LIVENESS ack still dispatches and runs promptly. + * + * (P3) CROSS-ENDPOINT-NOT-SERIALIZED — per-endpoint FIFO serialization does + * NOT serialize across endpoints: while ep_A's peer-control body is + * parked (holding one pool worker + ep_A busy), ep_B's peer-control body + * still runs concurrently on a DIFFERENT pool worker. (Complement of + * test_peerctrl_preserves_per_endpoint_order, which proves SAME-ep + * ordering.) + * + * All tests are DETERMINISTIC (gates / cond-vars / bounded polls, never sleeps + * on the correctness path). Every feature flag is flipped ON in-test via the + * ha_set_*_for_test setters and restored to 0 (production default OFF) at the + * end of each test. No production feature-flag default is changed; the only + * seams used are the pre-existing XCE_TESTING test-entry points and hooks. + * + * Requirements: raft-drive-async Req 1.1, 1.4, 1.5, 3.1, 3.2, 4.1, 4.2, 8.1, 8.2. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include "ha.h" +#include "netio.h" +#include "slowlane.h" +#include "vtaskmgr.h" +#include "xce/cluster_def.h" +#include "xce/errnox.h" + +/* ── XCE_TESTING seams (those NOT already declared in ha.h / vtaskmgr.h) ───── */ +extern endpoint_t *standby_ep_ctx_build_for_test(endpoint_t *ep, ha_state_t *ha, + cluster_t *c, int peer_idx); +extern int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt); +extern io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted, + uint64_t peer_tag, + uint64_t peer_lsn); +/* config-push offload seam (Task 3.3): a void(void) probe + a direct router. */ +extern void (*ha_configpush_probe)(void); +extern int ha_config_push_offloadable_for_test(cluster_t *c, const uint8_t *bytes, + size_t blen, const uint8_t sha256[32], + uint64_t new_tag); + +/* ha.h already declares: ha_set_peerctrl_dispatch_for_test, + * ha_set_snapshot_offload_for_test, ha_set_configpush_offload_for_test, + * ha_peerctrl_body_probe, ha_dispatch_repl_action_for_standby, + * ha_repl_send_snapshot_hook. */ + +/* ── Fixture config ──────────────────────────────────────────────────────── */ + +static ha_config_t g_cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 6000, + .election_timeout_ms = 500, + .probe_connect_timeout_ms = 50, + .max_promotion_retries = 3, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 500, +}; + +/* ── Generic park gate (cond-var; deterministic, no sleeps) ──────────────── */ + +typedef struct { + pthread_mutex_t mu; + pthread_cond_t entered_cv; /* a body broadcasts when it has parked */ + pthread_cond_t release_cv; /* test broadcasts to release parked bodies */ + int open; + _Atomic int entered; /* # of bodies that reached the gate */ +} gate_t; + +static void gate_init(gate_t *g) +{ + pthread_mutex_init(&g->mu, NULL); + pthread_cond_init(&g->entered_cv, NULL); + pthread_cond_init(&g->release_cv, NULL); + g->open = 0; + atomic_store(&g->entered, 0); +} + +static void gate_destroy(gate_t *g) +{ + pthread_mutex_destroy(&g->mu); + pthread_cond_destroy(&g->entered_cv); + pthread_cond_destroy(&g->release_cv); +} + +static void gate_open(gate_t *g) +{ + pthread_mutex_lock(&g->mu); + g->open = 1; + pthread_cond_broadcast(&g->release_cv); + pthread_mutex_unlock(&g->mu); +} + +/* Body-side: record entry, broadcast, then park until the test opens the gate. */ +static void gate_park(gate_t *g) +{ + pthread_mutex_lock(&g->mu); + atomic_fetch_add(&g->entered, 1); + pthread_cond_broadcast(&g->entered_cv); + while (!g->open) + pthread_cond_wait(&g->release_cv, &g->mu); + pthread_mutex_unlock(&g->mu); +} + +/* Bounded wait until >= n bodies have parked; 0 on success, -1 on timeout. */ +static int gate_wait_entered(gate_t *g, int n, int timeout_ms) +{ + struct timespec d; + clock_gettime(CLOCK_REALTIME, &d); + d.tv_sec += timeout_ms / 1000; + d.tv_nsec += (long)(timeout_ms % 1000) * 1000000L; + if (d.tv_nsec >= 1000000000L) { d.tv_sec++; d.tv_nsec -= 1000000000L; } + + pthread_mutex_lock(&g->mu); + int rc = 0; + while (atomic_load(&g->entered) < n) { + rc = pthread_cond_timedwait(&g->entered_cv, &g->mu, &d); + if (rc != 0) break; + } + int entered = atomic_load(&g->entered); + pthread_mutex_unlock(&g->mu); + return entered >= n ? 0 : -1; +} + +/* Bounded poll until *a >= want; returns the observed value. */ +static int poll_atomic_ge(_Atomic int *a, int want, int timeout_ms) +{ + for (int i = 0; i < timeout_ms; i++) { + if (atomic_load(a) >= want) break; + struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */ + nanosleep(&ts, NULL); + } + return atomic_load(a); +} + +/* ── Peer-control body probe (selective park keyed on ep pointer) ────────── */ + +static _Atomic int g_pc_runs; /* # peer-control bodies that entered */ +static _Atomic int g_pc_completed; /* # NON-parked bodies that returned */ +static _Atomic pthread_t g_pc_other_thread;/* thread a non-parked body ran on */ +static _Atomic pthread_t g_pc_park_thread; /* thread the parked body ran on */ +static _Atomic(endpoint_t *) g_pc_park_ep; /* if body's ep == this → park */ +static gate_t g_pc_gate; + +static void pc_probe(endpoint_t *ep) +{ + atomic_fetch_add(&g_pc_runs, 1); + endpoint_t *park = atomic_load(&g_pc_park_ep); + if (park != NULL && ep == park) { + atomic_store_explicit(&g_pc_park_thread, pthread_self(), + memory_order_relaxed); + gate_park(&g_pc_gate); + /* Signal completion on the park path too, so teardown can wait for the + * released body to FULLY return before vtaskmgr_fini (which cancels + * workers) — otherwise fini races the just-woken body → flaky hang. */ + atomic_fetch_add(&g_pc_completed, 1); + return; + } + atomic_store_explicit(&g_pc_other_thread, pthread_self(), + memory_order_relaxed); + atomic_fetch_add(&g_pc_completed, 1); +} + +static void pc_probe_reset(void) +{ + atomic_store(&g_pc_runs, 0); + atomic_store(&g_pc_completed, 0); + atomic_store(&g_pc_park_ep, NULL); + gate_init(&g_pc_gate); +} + +/* ── Snapshot-send lane probe (parks on a gate, Task 3.2 hook) ───────────── */ + +static _Atomic int g_snap_runs; +static gate_t g_snap_gate; +static int g_snap_park; /* 1 → body parks at the gate */ + +static int snap_probe(int peer_idx) +{ + (void)peer_idx; + if (g_snap_park) + gate_park(&g_snap_gate); + atomic_fetch_add(&g_snap_runs, 1); + return 0; +} + +/* ── Config-push lane probe (parks on a gate, Task 3.3 hook) ─────────────── */ + +static _Atomic int g_cp_runs; +static gate_t g_cp_gate; +static int g_cp_park; /* 1 → body parks at the gate */ + +static void cp_probe(void) +{ + if (g_cp_park) + gate_park(&g_cp_gate); + atomic_fetch_add(&g_cp_runs, 1); +} + +/* Drive one SEND_SNAPSHOT decision (mirrors test_ha_liveness snap_dispatch_once): + * standby tag current but lsn behind the oldest WAL window → the action is + * HA_REPL_SEND_SNAPSHOT, which the offload path submits to the slow lane. */ +static int snap_dispatch_once(cluster_t *c, endpoint_t *ep, int peer_idx) +{ + c->ha.standby_reported_tag[peer_idx] = 5; + c->ha.standby_reported_lsn[peer_idx] = 10; /* < oldest_wal_lsn below */ + return ha_dispatch_repl_action_for_standby( + &c->ha, c, ep, peer_idx, + /*committed_tag=*/5, /*committed_lsn=*/100, /*oldest_wal_lsn=*/50); +} + +/* ═══════════════════════════════════════════════════════════════════════════ + * P1 — TRANSFER DOES NOT BLOCK CROSS-ENDPOINT + * + * Flags ON (peerctrl + snapshot OR peerctrl + configpush). Park a transfer in + * flight on the SLOW LANE; while it is parked, dispatch a peer-control message + * for a DIFFERENT endpoint and assert its body runs on a POOL worker (a + * different thread, within a bounded poll) WHILE the transfer is still parked. + * This proves the offloaded transfer (lane) does not block the drive path + * (pool). + * ═══════════════════════════════════════════════════════════════════════════ */ + +static void p1_run(int use_configpush, const char *label) +{ + vtaskmgr_t pool; + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&pool, 4)); /* >= 2 vcpus */ + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&pool)); + + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_init(&c.ha, &g_cfg, 0); + c.ha.pool = &pool; + atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release); + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4)); + + /* peer-control endpoint B (a DIFFERENT peer than the transfer target). */ + endpoint_t ep_b; + memset(&ep_b, 0, sizeof(ep_b)); + standby_ep_ctx_build_for_test(&ep_b, &c.ha, &c, 1); + + /* Probes: peer-control body just records thread + runs (never parks). */ + pc_probe_reset(); + ha_peerctrl_body_probe = pc_probe; + ha_set_peerctrl_dispatch_for_test(1); + + /* Endpoint used by the transfer decision (borrowed pointer only). */ + endpoint_t ep_a; + memset(&ep_a, 0, sizeof(ep_a)); + + pthread_t caller = pthread_self(); + + if (!use_configpush) { + /* ── snapshot variant ── */ + atomic_store(&g_snap_runs, 0); + gate_init(&g_snap_gate); + g_snap_park = 1; + ha_repl_send_snapshot_hook = snap_probe; + ha_set_snapshot_offload_for_test(1); + + /* Submit the snapshot to the lane; its body parks in flight. */ + TEST_ASSERT_EQUAL_INT(0, snap_dispatch_once(&c, &ep_a, 0)); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, gate_wait_entered(&g_snap_gate, 1, 2000), + "P1: the snapshot transfer must reach the lane and park in flight"); + + /* While the transfer is parked, dispatch peer-control on ep_B. */ + io_packet_t *pkt = vote_reply_build_pkt_for_test(1, 1, 0, 0); + TEST_ASSERT_NOT_NULL(pkt); + TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep_b, pkt)); + + int runs = poll_atomic_ge(&g_pc_runs, 1, 2000); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs, + "P1: ep_B's peer-control body must run on the pool while the " + "snapshot transfer is parked on the lane"); + /* The transfer is STILL parked — concurrency, not serialization. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_snap_runs), + "P1: the snapshot transfer must still be in flight (parked) while " + "ep_B's peer control is processed"); + TEST_ASSERT_FALSE_MESSAGE( + pthread_equal(atomic_load_explicit(&g_pc_other_thread, + memory_order_relaxed), caller), + "P1: ep_B's peer-control body must run on a POOL worker, not the " + "caller (io_worker/drive) thread"); + + gate_open(&g_snap_gate); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, poll_atomic_ge(&g_snap_runs, 1, 2000), + "P1: the released snapshot transfer must then complete on the lane"); + + ha_set_snapshot_offload_for_test(0); /* restore production default OFF */ + ha_repl_send_snapshot_hook = NULL; + gate_destroy(&g_snap_gate); + } else { + /* ── config-push variant ── */ + atomic_store(&g_cp_runs, 0); + gate_init(&g_cp_gate); + g_cp_park = 1; + ha_configpush_probe = cp_probe; + ha_set_configpush_offload_for_test(1); + + /* Submit the config push to the lane; its body parks in flight. */ + TEST_ASSERT_EQUAL_INT(0, + ha_config_push_offloadable_for_test(&c, NULL, 0, NULL, 7)); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, gate_wait_entered(&g_cp_gate, 1, 2000), + "P1: the config-push transfer must reach the lane and park in flight"); + + io_packet_t *pkt = vote_reply_build_pkt_for_test(1, 1, 0, 0); + TEST_ASSERT_NOT_NULL(pkt); + TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep_b, pkt)); + + int runs = poll_atomic_ge(&g_pc_runs, 1, 2000); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs, + "P1: ep_B's peer-control body must run on the pool while the " + "config-push transfer is parked on the lane"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_cp_runs), + "P1: the config-push transfer must still be in flight (parked) while " + "ep_B's peer control is processed"); + TEST_ASSERT_FALSE_MESSAGE( + pthread_equal(atomic_load_explicit(&g_pc_other_thread, + memory_order_relaxed), caller), + "P1: ep_B's peer-control body must run on a POOL worker, not the " + "caller (io_worker/drive) thread"); + + gate_open(&g_cp_gate); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, poll_atomic_ge(&g_cp_runs, 1, 2000), + "P1: the released config-push transfer must then complete on the lane"); + + ha_set_configpush_offload_for_test(0); /* restore production default OFF */ + ha_configpush_probe = NULL; + gate_destroy(&g_cp_gate); + } + + /* Teardown: ensure ep_B's pool body has FULLY returned (g_pc_completed, its + * last step) before vtaskmgr_fini cancels workers; then drain lane + pool. + * slowlane_fini is graceful (joins workers), so the (already-released) lane + * body is safe; vtaskmgr_fini cancels, so the pool body must be done. */ + poll_atomic_ge(&g_pc_completed, 1, 2000); + ha_set_peerctrl_dispatch_for_test(0); /* restore production default OFF */ + ha_peerctrl_body_probe = NULL; + slowlane_fini(&c.slowlane); + vtaskmgr_fini(&pool); + gate_destroy(&g_pc_gate); + free(ep_b.data); + ha_fini(&c.ha); + printf("%s: PASS\n", label); +} + +static void test_p1_snapshot_transfer_does_not_block_cross_endpoint(void) +{ + p1_run(0, "test_p1_snapshot_transfer_does_not_block_cross_endpoint"); +} + +static void test_p1_configpush_transfer_does_not_block_cross_endpoint(void) +{ + p1_run(1, "test_p1_configpush_transfer_does_not_block_cross_endpoint"); +} + +/* ═══════════════════════════════════════════════════════════════════════════ + * P2 — ACK ADVANCES UNDER POOL SATURATION + * + * Assertion used: LIVENESS-DISPATCH-UNDER-SATURATION (the narrowed assertion + * the task authorizes). We do NOT wire a full commit-engine repl_on_ack → + * commit-advance here — that is too heavy for the unit harness and commit- + * advance itself is already covered by the commit-engine tests. Instead we + * assert the load-bearing liveness property: a peer-control ack — classified + * VTASK_LIVENESS by peer_ctrl_classify (which returns VTASK_LIVENESS for ALL + * peer-control, including wal-acks) — is dispatched and RUNS promptly even + * while (n_vcpus − VTASK_LIVENESS_RESERVED) REQ_BLOCK tasks occupy the pool + * (the reservation boundary), with one more REQ_BLOCK fast-rejected to prove + * the pool really is saturated at that boundary. + * ═══════════════════════════════════════════════════════════════════════════ */ + +static gate_t g_block_gate; +static _Atomic int g_block_started; +static _Atomic int g_block_done; /* incremented as the LAST body step */ + +static int block_task_entry(void *p) +{ + (void)p; + atomic_fetch_add(&g_block_started, 1); + /* Park holding this vcpu until the test releases the gate. */ + pthread_mutex_lock(&g_block_gate.mu); + while (!g_block_gate.open) + pthread_cond_wait(&g_block_gate.release_cv, &g_block_gate.mu); + pthread_mutex_unlock(&g_block_gate.mu); + /* Signal full completion so teardown can wait for every released blocker + * to return BEFORE vtaskmgr_fini (which cancels workers) — otherwise fini + * races a body still in vtaskmgr_blocking_release → flaky hang. */ + atomic_fetch_add(&g_block_done, 1); + return 0; +} + +static vtask_t *make_block_task(void) +{ + vtask_data_t *d = calloc(1, sizeof(*d)); + vtask_t *t = vtask_alloc(); + TEST_ASSERT_NOT_NULL(d); + TEST_ASSERT_NOT_NULL(t); + d->vtype = VTASK_GENERIC; + d->u.generic = NULL; + vtask_init(t, block_task_entry, d); + return t; +} + +static void test_p2_ack_runs_under_pool_saturation(void) +{ + const int NVCPUS = 4; + const int RESERVED = 2; /* cap = n_vcpus - reserved = 2 */ + const int CAP = NVCPUS - RESERVED; + + vtaskmgr_t pool; + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&pool, NVCPUS)); + vtaskmgr_set_liveness_reserved(&pool, RESERVED); + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&pool)); + + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_init(&c.ha, &g_cfg, 0); + c.ha.pool = &pool; + atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release); + + /* Saturate the REQ_BLOCK admission cap: CAP blocking tasks, all parked, + * occupying CAP pool workers. The reservation keeps RESERVED workers free + * for LIVENESS — that is the property under test. */ + gate_init(&g_block_gate); + atomic_store(&g_block_started, 0); + atomic_store(&g_block_done, 0); + for (int i = 0; i < CAP; i++) { + vtask_t *t = make_block_task(); + TEST_ASSERT_EQUAL_INT(0, vtask_enq_classified(&pool, t, VTASK_REQ_BLOCK)); + } + TEST_ASSERT_EQUAL_INT_MESSAGE(CAP, + poll_atomic_ge(&g_block_started, CAP, 2000), + "P2: all REQ_BLOCK saturators must reach their body (occupy a vcpu)"); + + /* One more REQ_BLOCK over the cap fast-rejects — proves the pool is + * saturated exactly at the reservation boundary (not merely idle). */ + vtask_t *over = make_block_task(); + int over_rc = vtask_enq_classified(&pool, over, VTASK_REQ_BLOCK); + TEST_ASSERT_EQUAL_INT_MESSAGE(-E_XCE_NOT_READY, over_rc, + "P2: a REQ_BLOCK over the reservation cap must fast-reject " + "(pool saturated at the boundary)"); + vtask_fini(over); /* rejected → caller still owns it */ + free(over); + + /* Dispatch the ack as a peer-control message: peer_ctrl_classify makes it + * VTASK_LIVENESS, so admission never gates it. It must run promptly on one + * of the RESERVED workers despite the *_BLOCK saturation. */ + endpoint_t ep_ack; + memset(&ep_ack, 0, sizeof(ep_ack)); + standby_ep_ctx_build_for_test(&ep_ack, &c.ha, &c, 0); + + pc_probe_reset(); /* g_pc_park_ep = NULL → never parks */ + ha_peerctrl_body_probe = pc_probe; + ha_set_peerctrl_dispatch_for_test(1); + + io_packet_t *ack = vote_reply_build_pkt_for_test(1, 1, 0, 0); + TEST_ASSERT_NOT_NULL(ack); + TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep_ack, ack)); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, poll_atomic_ge(&g_pc_runs, 1, 2000), + "P2: the VTASK_LIVENESS ack must run promptly even while the pool is " + "saturated to the reservation boundary with *_BLOCK tasks " + "(never starved by the blocking admission)"); + + /* Release the parked blockers AND wait for every one to FULLY return + * before fini — vtaskmgr_fini cancels workers, so finishing here while a + * blocker is still in vtaskmgr_blocking_release races blk_lock teardown. */ + gate_open(&g_block_gate); + TEST_ASSERT_EQUAL_INT_MESSAGE(CAP, poll_atomic_ge(&g_block_done, CAP, 2000), + "P2: all released blockers must fully return before teardown"); + /* Also wait for the LIVENESS ack body to fully return (it increments + * g_pc_completed as its last step). */ + poll_atomic_ge(&g_pc_completed, 1, 2000); + + ha_set_peerctrl_dispatch_for_test(0); /* restore production default OFF */ + ha_peerctrl_body_probe = NULL; + vtaskmgr_fini(&pool); + gate_destroy(&g_block_gate); + gate_destroy(&g_pc_gate); /* allocated by pc_probe_reset */ + free(ep_ack.data); + ha_fini(&c.ha); + printf("test_p2_ack_runs_under_pool_saturation: PASS\n"); +} + +/* ═══════════════════════════════════════════════════════════════════════════ + * P3 — CROSS-ENDPOINT NOT SERIALIZED + * + * Flags ON (peerctrl). Park ep_A's peer-control body at the gate (holding one + * pool worker + ep_A busy via bind_vtask's per-endpoint serialization). While + * it is parked, dispatch a peer-control message on ep_B and assert ep_B's body + * runs to completion on a DIFFERENT pool worker WHILE ep_A's is still parked — + * proving per-endpoint FIFO does NOT serialize across endpoints. (Pool has + * >= 2 vcpus.) + * ═══════════════════════════════════════════════════════════════════════════ */ + +static void test_p3_cross_endpoint_not_serialized(void) +{ + vtaskmgr_t pool; + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&pool, 4)); /* >= 2 vcpus */ + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&pool)); + + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_init(&c.ha, &g_cfg, 0); + c.ha.pool = &pool; + atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release); + + endpoint_t ep_a, ep_b; + memset(&ep_a, 0, sizeof(ep_a)); + memset(&ep_b, 0, sizeof(ep_b)); + standby_ep_ctx_build_for_test(&ep_a, &c.ha, &c, 0); + standby_ep_ctx_build_for_test(&ep_b, &c.ha, &c, 1); + + pc_probe_reset(); + atomic_store(&g_pc_park_ep, &ep_a); /* only ep_A's body parks */ + ha_peerctrl_body_probe = pc_probe; + ha_set_peerctrl_dispatch_for_test(1); + + pthread_t caller = pthread_self(); + + /* Dispatch on ep_A — its body enters and parks (holds a pool worker). */ + io_packet_t *pkt_a = vote_reply_build_pkt_for_test(1, 1, 0, 0); + TEST_ASSERT_NOT_NULL(pkt_a); + TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep_a, pkt_a)); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, gate_wait_entered(&g_pc_gate, 1, 2000), + "P3: ep_A's peer-control body must reach the gate and park on a worker"); + + /* Dispatch on ep_B while ep_A is parked — must run concurrently. */ + io_packet_t *pkt_b = vote_reply_build_pkt_for_test(1, 1, 0, 0); + TEST_ASSERT_NOT_NULL(pkt_b); + TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep_b, pkt_b)); + + /* ep_B's body must ENTER and COMPLETE while ep_A's is still parked. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, poll_atomic_ge(&g_pc_completed, 1, 2000), + "P3: ep_B's peer-control body must run to completion while ep_A's body " + "is parked — per-endpoint FIFO must NOT serialize across endpoints"); + TEST_ASSERT_EQUAL_INT_MESSAGE(2, atomic_load(&g_pc_runs), + "P3: BOTH bodies must have entered concurrently (ep_A parked, ep_B ran)"); + + /* The two bodies ran on distinct pool workers, neither on the caller. */ + pthread_t tb = atomic_load_explicit(&g_pc_other_thread, memory_order_relaxed); + pthread_t ta = atomic_load_explicit(&g_pc_park_thread, memory_order_relaxed); + TEST_ASSERT_FALSE_MESSAGE(pthread_equal(tb, caller), + "P3: ep_B's body must run on a POOL worker, not the caller's thread"); + TEST_ASSERT_FALSE_MESSAGE(pthread_equal(tb, ta), + "P3: ep_A and ep_B bodies must run on DIFFERENT pool workers " + "(concurrent, not serialized)"); + + /* Release ep_A, then wait for BOTH bodies to FULLY return (g_pc_completed, + * incremented as the last body step) before fini — vtaskmgr_fini cancels + * workers, so finishing while the just-woken ep_A body is still unwinding + * races the cancel → flaky hang. */ + gate_open(&g_pc_gate); + TEST_ASSERT_EQUAL_INT_MESSAGE(2, poll_atomic_ge(&g_pc_completed, 2, 2000), + "P3: both bodies must fully complete after release before teardown"); + + ha_set_peerctrl_dispatch_for_test(0); /* restore production default OFF */ + ha_peerctrl_body_probe = NULL; + vtaskmgr_fini(&pool); + gate_destroy(&g_pc_gate); + free(ep_a.data); + free(ep_b.data); + ha_fini(&c.ha); + printf("test_p3_cross_endpoint_not_serialized: PASS\n"); +} + +/* ── Suite entry point ──────────────────────────────────────────────────── */ + +void test_ha_dispatch_integration(void) +{ + UNITY_BEGIN(); + /* P1 — transfer (lane) does not block cross-endpoint peer control (pool) */ + RUN_TEST(test_p1_snapshot_transfer_does_not_block_cross_endpoint); + RUN_TEST(test_p1_configpush_transfer_does_not_block_cross_endpoint); + /* P2 — VTASK_LIVENESS ack runs under *_BLOCK pool saturation */ + RUN_TEST(test_p2_ack_runs_under_pool_saturation); + /* P3 — cross-endpoint peer control is not serialized */ + RUN_TEST(test_p3_cross_endpoint_not_serialized); + UNITY_END(); +}