/* * 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(); }