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

fix(raft-drive-async): endpoint-quiescence for bug-1197 vote-ep teardown (task 3.4a)

Browse source 
When the (still-default-OFF) peer-control dispatch flag is ON,
raft_cb_become_follower runs as a vtask pool worker and frees every peer
vote-ep while other pool workers may still hold those eps' data — a UAF
(bug-1197). This makes that path quiescence-safe so the flag can later be
flipped ON (deferred to the 4.x prod-flip gate; production stays OFF here).

Mechanism (no libxce-wide endpoint refcount; reuses the existing per-ep FIFO +
busy serialization):

- Deferred teardown-as-vtask: the loser loop, when dispatch is ON and the pool
  is wired, does per vote-ep endpoint_mark_tearing_down -> iomgr_unmonitor ->
  enqueue a VTASK_IO/VTASK_LIVENESS teardown vtask bound to that ep. Per-ep busy
  FIFO makes it run strictly after any in-flight/queued same-ep peer-control
  vtask; the body NULLs iod->ep before endpoint_fini+free(ep) to disarm
  run_task's post-body endpoint_unblock_tasks (which would touch the freed
  ep->lock). OFF / pool-unwired path is byte-for-byte the legacy endpoint_fini.

- RCU-like drain closes the dispatch/teardown TOCTOU window: new _Atomic int
  endpoint_t.dispatch_inflight (+ tearing_down:1 bit). The ON dispatch brackets
  its flag-check+enqueue with fetch_add/fetch_sub (seq_cst);
  endpoint_mark_tearing_down sets the flag under ep->lock, releases, then spins
  until dispatch_inflight drains to 0. So every dispatch that read the flag
  false finishes its enqueue before mark returns (FIFO-before the teardown
  vtask); any later dispatch reads the flag set and drops the packet. ep->lock
  is deliberately NOT held across the enqueue (it inverts with the taskq lock in
  vtask_eligible); the counter is the lock-free handshake. Drain is bounded: the
  held window is just a calloc + non-blocking enqueue on the single io_worker.

Still open before the production flag flip (4.x): the borrowed-ep race in
snapshot_lane_fn/configpush_lane_fn is distinct (slowlane thread vs io_worker
RDHUP teardown, not the vtask-pool FIFO) and needs its own per-standby-ep
discipline.

Tests: test_ha_vep_teardown.c — 5 ASan tests (ordering, late-event drop, run_task
unblock disarm, OFF-path unchanged, deterministic mark-drains-inflight-dispatch).
xcelib_test 282/0; controller_unit_test EXIT=0; ASan-clean. Adversarial
cpp-review: no CRITICAL/HIGH; handshake formally sound, flag-ON path race-free.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01S6fDCtXkkEgupu7rUah1y2
This commit is contained in:
Rongsong Shen 2026-06-26 17:37:26 +00:00
parent bd02064ebc
commit 6d510d8cf4
10 changed files with 1037 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

190
src/controller/ha.c
View file

@ -108,6 +108,9 @@ extern int config_schedule_activate_retry(cluster_t *c);
static void on_vote_connected(endpoint_t *ep);
static void on_pre_vote_connected(endpoint_t *ep);
static int vote_ep_err_handler(void *ep_raw);
/* bug-1197: defer a vote-ep teardown onto the per-ep FIFO of the vtask pool so
* it runs strictly after the last live use of the ep (flag-ON loser path). */
static void vote_ep_teardown_async(ha_state_t *ha, endpoint_t *ep);
/* ha-single-leader-role task 2.2: shared promotion seam — every road to
* PRIMARY registers the persist hooks and arms the own-term gate. */
static void ha_register_persist_hooks(cluster_t *c);
@ -1285,11 +1288,24 @@ static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx)
&ha->elect_ctx, memory_order_acquire);
if (ectx) {
for (int i = 0; i < ectx->n_peers; i++) {
if (ectx->votes[i].ep != NULL) {
INFO("HA: closing vote ep[%d] (loser path)", i);
endpoint_fini(ectx->votes[i].ep);
ectx->votes[i].ep = NULL;
endpoint_t *ep = ectx->votes[i].ep;
if (ep == NULL)
continue;
INFO("HA: closing vote ep[%d] (loser path)", i);
if (ha_peerctrl_dispatch_enabled() && ha->pool != NULL) {
/* bug-1197: this callback runs on a pool worker; another
* worker may be mid-vtask (or have a queued vtask) on this
* same ep reading ep->data. Defer teardown onto the ep's
* per-ep FIFO so endpoint_fini+free runs strictly after the
* last live use, and gate late dispatch (tearing_down). */
vote_ep_teardown_async(ha, ep);
} else {
/* Flag OFF / pool unwired: UNCHANGED legacy behavior —
* endpoint_fini only, no free(ep) (the struct is leaked per
* lost election, exactly as before this commit). */
endpoint_fini(ep);
}
ectx->votes[i].ep = NULL;
}
}
}
@ -5365,6 +5381,127 @@ static vtask_class_t peer_ctrl_classify(ha_state_t *ha, io_packet_t *pkt)
return VTASK_LIVENESS;
}
/* ── bug-1197: deferred vote-ep teardown ──────────────────────────────────────
*
* When g_drive_async_peerctrl_dispatch_enabled is ON, raft_cb_become_follower
* runs on a pool worker and must endpoint_fini() every loser vote-ep. But
* another pool worker can be mid-vtask (or have a queued vtask) on one of those
* same eps reading ep->data (a vote_ep_ctx_t) — freeing ep->data underneath it
* is a UAF. Instead of freeing inline we enqueue a teardown vtask BOUND to the
* ep (VTASK_IO): the pool's per-ep busy serialization guarantees it runs after
* any in-flight or earlier-queued same-ep vtask completes. We also mark the ep
* tearing_down + iomgr_unmonitor it BEFORE enqueue so the io_worker drops any
* late same-batch event instead of dispatching a new vtask onto a doomed ep. */
#ifdef XCE_TESTING
/* test-only ordering probe: fired (with the real ep) at the top of the teardown
* vtask body, before endpoint_fini, so a test can assert teardown ran strictly
* after the last live same-ep use. NULL in production. */
void (*ha_vep_teardown_probe)(endpoint_t *) = NULL;
/* test-only drain probe: fired in peer_ctrl_msg_handler AFTER the tearing_down
* check passes but BEFORE the pool enqueue (inside the dispatch_inflight
* bracket), so a test can park a dispatch in the drained window and prove
* endpoint_mark_tearing_down blocks until the enqueue completes. NULL in
* production. */
void (*ha_peerctrl_predispatch_probe)(endpoint_t *) = NULL;
#endif
/* Runs on a pool worker, bound VTASK_IO to ep, strictly AFTER any in-flight or
* queued same-ep peer-control vtask (per-ep busy FIFO). On the loser path
* ectx->votes[i].ep is the sole reference to ep, so this owns the free. */
static int vote_ep_teardown_vtask_entry(vtask_t *task)
{
io_task_data_t *iod = &task->data->u.iodata;
endpoint_t *ep = iod->ep;
/* Disarm run_task's post-body endpoint_unblock_tasks: it would touch
* ep->lock AFTER we free ep below. NULLing the task's ep makes run_task
* skip the unblock (run_task: `if (ep != NULL) endpoint_unblock_tasks`). */
iod->ep = NULL;
if (ep != NULL) {
#ifdef XCE_TESTING
if (ha_vep_teardown_probe)
ha_vep_teardown_probe(ep);
#endif
endpoint_fini(ep);
free(ep);
}
return 0;
}
/* Build + enqueue the teardown vtask. Caller has ALREADY
* endpoint_mark_tearing_down(ep) + iomgr_unmonitor(ep). Controllers never
* self-terminate: on any alloc/enqueue failure we fail loud (WARN) and tear the
* ep down inline — with the pool unwired or the task rejected there is no other
* worker that could be racing this ep, so an inline endpoint_fini+free is safe. */
static void vote_ep_teardown_enqueue(ha_state_t *ha, endpoint_t *ep)
{
vtaskmgr_t *pool = (vtaskmgr_t *)ha->pool;
vtask_data_t *data = calloc(1, sizeof(*data));
vtask_t *task = calloc(1, sizeof(*task));
if (pool == NULL || data == NULL || task == NULL) {
WARN("bug-1197: vote-ep teardown enqueue unavailable (pool=%p data=%p "
"task=%p) — tearing down inline (stay alive)",
(void *)pool, (void *)data, (void *)task);
free(data);
free(task);
endpoint_fini(ep);
free(ep);
return;
}
data->vtype = VTASK_IO;
data->u.iodata.ep = ep; /* drives per-ep serialization in bind_vtask */
data->u.iodata.pkt = NULL;
vtask_init(task, (vtask_func_t)vote_ep_teardown_vtask_entry, data);
/* VTASK_LIVENESS is never admission-gated, so the teardown can always be
* enqueued even when the pool is saturated by blocking tasks. */
if (vtask_enq_classified(pool, task, VTASK_LIVENESS) < 0) {
WARN("bug-1197: vote-ep teardown enqueue failed — tearing down inline "
"(stay alive)");
free(data);
free(task);
endpoint_fini(ep);
free(ep);
}
}
/* The flag-ON loser-path teardown for one vote-ep: mark tearing_down, stop epoll
* wakeups, then defer the free onto the ep's per-ep FIFO. ORDERING (bug-1197):
* - endpoint_mark_tearing_down sets the flag under ep->lock; the io_worker's
* late-event gate (peer_ctrl_msg_handler, ON branch) reads it under ep->lock
* via endpoint_is_tearing_down, so any dispatch DECISION taken after the
* mark observes it and drops — nothing new is enqueued for this ep.
* - iomgr_unmonitor stops FUTURE epoll events (an event already in the current
* batch is handled by the tearing_down gate above).
* - the teardown vtask is itself VTASK_IO bound to ep, so the pool's per-ep
* busy serialization runs it strictly after any in-flight / earlier-queued
* same-ep vtask completes.
* NOTE: we cannot hold ep->lock ACROSS the dispatch enqueue (vtask_enq_classified
* → dataq_enq takes the taskq lock, while the pool dispatcher takes ep->lock
* UNDER the taskq lock in vtask_eligible → endpoint_is_task_blocked — an AB/BA
* inversion). The check is therefore done under ep->lock and the enqueue
* outside it; see the io_worker gate for the residual-window note. */
static void vote_ep_teardown_async(ha_state_t *ha, endpoint_t *ep)
{
endpoint_mark_tearing_down(ep);
if (ep->iomgr != NULL)
iomgr_unmonitor(ep->iomgr, ep);
vote_ep_teardown_enqueue(ha, ep);
}
#ifdef XCE_TESTING
/* ha_vote_ep_teardown_for_test: drive the flag-ON per-ep async teardown branch
* for a single ep without staging a full election/become_follower. */
void ha_vote_ep_teardown_for_test(ha_state_t *ha, endpoint_t *ep)
{
vote_ep_teardown_async(ha, ep);
}
#endif
/*
* peer_ctrl_dispatch_to_pool (raft-drive-async task 3.1): hand the decoded
* peer-control packet to the vtask pool and return promptly. Per-endpoint
@ -5466,7 +5603,50 @@ static int peer_ctrl_msg_handler(void *ep_raw)
if (!ha_peerctrl_dispatch_enabled())
return peer_ctrl_msg_handler_inline(ep, pkt);
return peer_ctrl_dispatch_to_pool(ep, pkt);
/* bug-1197 late-event gate (ON path only): if this ep is being torn down,
* a same-batch event may still be delivered by the io_worker after the
* teardown thread called iomgr_unmonitor. Dropping it here keeps the doomed
* ep off the vtask pool.
*
* The window between the tearing_down check and the pool enqueue is closed
* by ep->dispatch_inflight: we increment it (seq_cst) BEFORE reading the
* flag and decrement it AFTER the enqueue completes. endpoint_mark_tearing_
* down sets the flag (under ep->lock) and then DRAINS this counter to 0
* before returning. So once the teardown caller's mark returns, every
* dispatch that read the flag as false has already finished its enqueue;
* because the teardown vtask is enqueued only AFTER mark returns, that
* peer-control vtask is strictly FIFO-before the teardown vtask (and gates
* on the same ep busy bit). Any dispatch starting after the flag is set
* reads it true here and drops. No same-ep vtask is ever enqueued after the
* teardown vtask — the window is fully closed. We must NOT hold ep->lock
* across the enqueue (it would invert with the taskq lock, see
* vote_ep_teardown_async); the counter is the lock-free handshake instead. */
atomic_fetch_add_explicit(&ep->dispatch_inflight, 1, memory_order_seq_cst);
if (endpoint_is_tearing_down(ep)) {
if (pkt) { free_packet(pkt); free(pkt); }
atomic_fetch_sub_explicit(&ep->dispatch_inflight, 1,
memory_order_seq_cst);
return 0;
}
#ifdef XCE_TESTING
/* test-only seam: fires AFTER the flag-check passed but BEFORE the enqueue,
* so a test can park a dispatch in exactly the drained window and prove
* endpoint_mark_tearing_down blocks until the enqueue completes. */
if (ha_peerctrl_predispatch_probe)
ha_peerctrl_predispatch_probe(ep);
#endif
/* NOTE (drain liveness): this call runs while dispatch_inflight is held, so
* a concurrent endpoint_mark_tearing_down on this ep spins until it returns.
* peer_ctrl_dispatch_to_pool only ever enqueues (non-blocking) or, on OOM,
* runs peer_ctrl_msg_handler_inline — which must also stay non-blocking
* (the vtask-handler no-block invariant). Keep it bounded or the drain is. */
int rc = peer_ctrl_dispatch_to_pool(ep, pkt);
atomic_fetch_sub_explicit(&ep->dispatch_inflight, 1, memory_order_seq_cst);
return rc;
}
#ifdef XCE_TESTING

21
src/controller/ha.h
View file

@ -1313,6 +1313,27 @@ int ha_peerctrl_dispatch_enabled(void);
* production (zero cost) — a plain global like ha_nop_req_hook. */
extern void (*ha_peerctrl_body_probe)(endpoint_t *ep);
/*
* ha_vep_teardown_probe / ha_vote_ep_teardown_for_test (bug-1197): test seams
* for the deferred vote-ep teardown. ha_vep_teardown_probe (NULL in production)
* fires with the real ep at the top of the teardown vtask body, before
* endpoint_fini, so a test can assert teardown ran strictly after the last live
* same-ep use. ha_vote_ep_teardown_for_test drives the flag-ON per-ep async
* teardown branch (mark + iomgr_unmonitor + enqueue) for a single ep without
* staging a full election. Both compiled only under XCE_TESTING.
*/
extern void (*ha_vep_teardown_probe)(endpoint_t *ep);
void ha_vote_ep_teardown_for_test(ha_state_t *ha, endpoint_t *ep);
/*
* ha_peerctrl_predispatch_probe (bug-1197): test seam fired inside the dispatch
* path AFTER the tearing_down check passes but BEFORE the pool enqueue (within
* the ep->dispatch_inflight bracket). A test installs it to park a dispatch in
* exactly the drained window and prove endpoint_mark_tearing_down blocks until
* the enqueue completes. NULL in production; compiled only under XCE_TESTING.
*/
extern void (*ha_peerctrl_predispatch_probe)(endpoint_t *ep);
/* ── Task 2.4: Non-blocking async election initiator ────────────────────── */
/*

1
src/controller/meson.build
View file

@ -143,6 +143,7 @@ controller_unit_test_src = [
'testsuite/test_drive_async_config.c',
'testsuite/test_slowlane.c',
'testsuite/test_commit_admission_pool.c',
'testsuite/test_ha_vep_teardown.c',
'testsuite/test_waldump.c',
'testsuite/waldump.c',
'testsuite/test_walcompare.c',

2
src/controller/testsuite/controller_test.c
View file

@ -69,6 +69,7 @@ extern void test_walcompare(void);
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);
void setUp(void) { return; }
@ -207,6 +208,7 @@ int main(int argc, char* argv[])
TEST(test_drive_async_config);
TEST(test_slowlane);
TEST(test_commit_admission_pool);
TEST(test_ha_vep_teardown);
return UNITY_END();
}

549
src/controller/testsuite/test_ha_vep_teardown.c Normal file
View file

@ -0,0 +1,549 @@
/*
* test_ha_vep_teardown.c — bug-1197: deferred vote-ep teardown.
*
* When g_drive_async_peerctrl_dispatch_enabled is ON, raft_cb_become_follower
* runs on a vtask pool worker and must endpoint_fini() every loser vote-ep while
* another pool worker may be mid-vtask (or have a queued vtask) on one of those
* same eps reading ep->data → use-after-free. The fix defers each vote-ep
* teardown onto the SAME per-endpoint FIFO the pool already enforces (so the
* free runs strictly after the last live use) and gates late dispatch on a
* tearing_down flag.
*
* Tests (all run with ASAN; the runner uses detect_leaks=0 for the pre-existing
* suite-wide leak, but these tests are use-after-free / double-free clean):
* (a) ORDERING — teardown runs strictly after an in-flight same-ep vtask.
* (b) LATE-DROP — a late same-batch event on a tearing-down ep is dropped,
* its packet freed, no new task enqueued.
* (c) DISARM — a teardown vtask run through the real pool is ASAN-clean
* (iod->ep = NULL prevents run_task touching the freed lock).
* (d) OFF-PATH — with the flag OFF the loser loop uses plain endpoint_fini
* and never touches the pool (behavior unchanged).
*/
#define _GNU_SOURCE
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <pthread.h>
#include <unity/unity.h>
#include "ha.h"
#include "netio.h"
#include "vtaskmgr.h"
/* ── XCE_TESTING seams declared in ha.c ──────────────────────────────────── */
extern int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt);
extern io_packet_t *repl_catchup_done_pkt_for_test(void);
extern void raft_cb_become_follower_for_test(ha_state_t *ha, uint64_t term);
/* ── Fixtures ────────────────────────────────────────────────────────────── */
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,
};
/* Sequence oracle: the in-flight vtask body and the teardown probe each grab a
* monotonically increasing ticket; teardown's ticket must be the larger. */
static _Atomic int g_seq = 0;
static _Atomic int g_inflight_seq = -1;
static _Atomic int g_teardown_seq = -1;
static _Atomic int g_teardown_done = 0;
static _Atomic int g_body_calls = 0;
static _Atomic int g_body_seq = -1;
static _Atomic int g_body_ran = 0;
/* Gate the in-flight vtask body parks on (so the test controls when it ends). */
static struct {
pthread_mutex_t mu;
pthread_cond_t entered_cv;
pthread_cond_t release_cv;
int open;
_Atomic int entered;
} g_gate;
static void gate_init(void)
{
pthread_mutex_init(&g_gate.mu, NULL);
pthread_cond_init(&g_gate.entered_cv, NULL);
pthread_cond_init(&g_gate.release_cv, NULL);
g_gate.open = 0;
atomic_store(&g_gate.entered, 0);
}
static void gate_destroy(void)
{
pthread_mutex_destroy(&g_gate.mu);
pthread_cond_destroy(&g_gate.entered_cv);
pthread_cond_destroy(&g_gate.release_cv);
}
static void gate_open(void)
{
pthread_mutex_lock(&g_gate.mu);
g_gate.open = 1;
pthread_cond_broadcast(&g_gate.release_cv);
pthread_mutex_unlock(&g_gate.mu);
}
static int gate_wait_entered(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_gate.mu);
int rc = 0;
while (atomic_load(&g_gate.entered) < 1) {
rc = pthread_cond_timedwait(&g_gate.entered_cv, &g_gate.mu, &d);
if (rc != 0) break;
}
int entered = atomic_load(&g_gate.entered);
pthread_mutex_unlock(&g_gate.mu);
return entered >= 1 ? 0 : -1;
}
/* In-flight peer-control vtask body: record a ticket, signal entry, park. */
static int inflight_vtask_body(vtask_t *task)
{
(void)task;
atomic_store(&g_inflight_seq, atomic_fetch_add(&g_seq, 1));
pthread_mutex_lock(&g_gate.mu);
atomic_fetch_add(&g_gate.entered, 1);
pthread_cond_broadcast(&g_gate.entered_cv);
while (!g_gate.open)
pthread_cond_wait(&g_gate.release_cv, &g_gate.mu);
pthread_mutex_unlock(&g_gate.mu);
return 0;
}
/* Teardown probe (fires at the top of the teardown vtask body). */
static void teardown_probe(endpoint_t *ep)
{
(void)ep;
atomic_store(&g_teardown_seq, atomic_fetch_add(&g_seq, 1));
atomic_store(&g_teardown_done, 1);
}
/* Body probe (fires at the top of the inline routing body). */
static void body_probe(endpoint_t *ep)
{
(void)ep;
atomic_fetch_add(&g_body_calls, 1);
}
/* Body probe variant that also records a sequence ticket (drain test). */
static void body_seq_probe(endpoint_t *ep)
{
(void)ep;
atomic_store(&g_body_seq, atomic_fetch_add(&g_seq, 1));
atomic_store(&g_body_ran, 1);
atomic_fetch_add(&g_body_calls, 1);
}
/* Predispatch probe (drain test): fires INSIDE the dispatch_inflight bracket,
* after the tearing_down check passed but before the pool enqueue. Signals
* entry, then parks on the gate so the dispatch stays in-flight (counter == 1,
* not yet enqueued) while the test drives endpoint_mark_tearing_down's drain. */
static void predispatch_park_probe(endpoint_t *ep)
{
(void)ep;
pthread_mutex_lock(&g_gate.mu);
atomic_fetch_add(&g_gate.entered, 1);
pthread_cond_broadcast(&g_gate.entered_cv);
while (!g_gate.open)
pthread_cond_wait(&g_gate.release_cv, &g_gate.mu);
pthread_mutex_unlock(&g_gate.mu);
}
/* Heap ep with a heap ep->data (so endpoint_fini's FREEUP has something to free,
* matching the production vote_ep_ctx_t allocation). */
static endpoint_t *make_ep(iomgr_t *iomgr)
{
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
ep->iomgr = iomgr;
ep->fd = -1;
ep->data = calloc(1, 128); /* stand-in for vote_ep_ctx_t */
TEST_ASSERT_NOT_NULL(ep->data);
return ep;
}
static void reset_oracles(void)
{
atomic_store(&g_seq, 0);
atomic_store(&g_inflight_seq, -1);
atomic_store(&g_teardown_seq, -1);
atomic_store(&g_teardown_done, 0);
atomic_store(&g_body_calls, 0);
atomic_store(&g_body_seq, -1);
atomic_store(&g_body_ran, 0);
}
/* Enqueue an arbitrary VTASK_IO body bound to ep (drives per-ep serialization). */
static void enqueue_io_vtask(vtaskmgr_t *pool, endpoint_t *ep, vtask_func_t fn)
{
vtask_data_t *data = calloc(1, sizeof(*data));
vtask_t *task = calloc(1, sizeof(*task));
TEST_ASSERT_NOT_NULL(data);
TEST_ASSERT_NOT_NULL(task);
data->vtype = VTASK_IO;
data->u.iodata.ep = ep;
data->u.iodata.pkt = NULL;
vtask_init(task, fn, data);
TEST_ASSERT_EQUAL_INT(0, vtask_enq_classified(pool, task, VTASK_LIVENESS));
}
static int poll_until(_Atomic int *flag, int want, int timeout_ms)
{
for (int i = 0; i < timeout_ms / 5; i++) {
if (atomic_load(flag) == want)
return 0;
struct timespec ts = { .tv_sec = 0, .tv_nsec = 5 * 1000000L };
nanosleep(&ts, NULL);
}
return atomic_load(flag) == want ? 0 : -1;
}
/* ── Test (a): ORDERING ──────────────────────────────────────────────────── */
static void test_vep_teardown_runs_after_inflight(void)
{
reset_oracles();
gate_init();
ha_state_t ha;
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
vtaskmgr_t pool;
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&pool, 2));
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&pool));
ha.pool = &pool;
iomgr_t iomgr;
TEST_ASSERT_EQUAL_INT(0, iomgr_init(&iomgr));
endpoint_t *ep = make_ep(&iomgr);
ha_set_peerctrl_dispatch_for_test(1);
ha_vep_teardown_probe = teardown_probe;
/* Park an in-flight peer-control vtask on this ep (busy=1 while gated). */
enqueue_io_vtask(&pool, ep, (vtask_func_t)inflight_vtask_body);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, gate_wait_entered(2000),
"in-flight vtask must run and park (busy=1)");
/* Now tear the SAME ep down: marks tearing_down + enqueues the teardown
* vtask, which is busy-blocked behind the parked in-flight vtask. */
ha_vote_ep_teardown_for_test(&ha, ep);
/* While the in-flight vtask is still parked, teardown must NOT have run. */
TEST_ASSERT_TRUE_MESSAGE(endpoint_is_tearing_down(ep),
"ep must be marked tearing_down once teardown is initiated");
struct timespec ts = { .tv_sec = 0, .tv_nsec = 80 * 1000000L };
nanosleep(&ts, NULL);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_teardown_done),
"teardown must NOT run while an in-flight same-ep vtask is parked");
/* Release the in-flight vtask; teardown becomes eligible and runs. */
gate_open();
TEST_ASSERT_EQUAL_INT_MESSAGE(0, poll_until(&g_teardown_done, 1, 2000),
"teardown must run once the in-flight vtask completes");
TEST_ASSERT_GREATER_THAN_INT_MESSAGE(atomic_load(&g_inflight_seq),
atomic_load(&g_teardown_seq),
"teardown_seq must be strictly after inflight_seq");
/* Join the pool (workers idle; ep already freed by the teardown body). */
vtaskmgr_fini(&pool);
iomgr_fini(&iomgr);
ha_vep_teardown_probe = NULL;
ha_set_peerctrl_dispatch_for_test(0);
ha.pool = NULL;
ha_fini(&ha);
gate_destroy();
printf("test_vep_teardown_runs_after_inflight: PASS\n");
}
/* ── Test (b): LATE-DROP ─────────────────────────────────────────────────── */
static void test_vep_late_event_dropped(void)
{
reset_oracles();
ha_state_t ha;
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
vtaskmgr_t pool;
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&pool, 2));
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&pool));
ha.pool = &pool;
iomgr_t iomgr;
TEST_ASSERT_EQUAL_INT(0, iomgr_init(&iomgr));
endpoint_t *ep = make_ep(&iomgr);
ha_set_peerctrl_dispatch_for_test(1);
ha_peerctrl_body_probe = body_probe;
/* mark + unmonitor (the teardown caller's first two steps) — but do NOT
* enqueue a teardown vtask, so the ep stays alive for this call. */
endpoint_mark_tearing_down(ep);
iomgr_unmonitor(ep->iomgr, ep);
io_packet_t *pkt = repl_catchup_done_pkt_for_test();
TEST_ASSERT_NOT_NULL(pkt);
int rc = peer_ctrl_msg_handler_for_test(ep, pkt); /* takes ownership of pkt */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"late event on a tearing-down ep must return 0 (dropped)");
/* Give any (erroneously) enqueued task time to run the body. */
struct timespec ts = { .tv_sec = 0, .tv_nsec = 100 * 1000000L };
nanosleep(&ts, NULL);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_body_calls),
"a dropped late event must NOT enqueue/run the routing body");
ha_peerctrl_body_probe = NULL;
ha_set_peerctrl_dispatch_for_test(0);
/* ep was not torn down by a vtask here — free it ourselves (pkt was freed
* by the drop path). */
endpoint_fini(ep);
free(ep);
vtaskmgr_fini(&pool);
iomgr_fini(&iomgr);
ha.pool = NULL;
ha_fini(&ha);
printf("test_vep_late_event_dropped: PASS\n");
}
/* ── Test (c): DISARM ────────────────────────────────────────────────────── */
static void test_vep_teardown_vtask_disarms_unblock(void)
{
reset_oracles();
ha_state_t ha;
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
vtaskmgr_t pool;
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&pool, 2));
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&pool));
ha.pool = &pool;
iomgr_t iomgr;
TEST_ASSERT_EQUAL_INT(0, iomgr_init(&iomgr));
endpoint_t *ep = make_ep(&iomgr);
ha_set_peerctrl_dispatch_for_test(1);
ha_vep_teardown_probe = teardown_probe;
/* Run a teardown vtask straight through the pool (no in-flight task). If the
* iod->ep = NULL disarm were missing, run_task would call
* endpoint_unblock_tasks on the freed ep->lock — ASAN would flag a UAF. */
ha_vote_ep_teardown_for_test(&ha, ep);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, poll_until(&g_teardown_done, 1, 2000),
"teardown vtask must run through the pool");
vtaskmgr_fini(&pool);
iomgr_fini(&iomgr);
ha_vep_teardown_probe = NULL;
ha_set_peerctrl_dispatch_for_test(0);
ha.pool = NULL;
ha_fini(&ha);
printf("test_vep_teardown_vtask_disarms_unblock: PASS\n");
}
/* ── Test (d): OFF-PATH ──────────────────────────────────────────────────── */
static void test_vep_off_path_uses_plain_fini(void)
{
reset_oracles();
ha_state_t ha;
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
ha.pool = NULL; /* off path must not need the pool */
ha_set_peerctrl_dispatch_for_test(0); /* flag OFF */
ha_vep_teardown_probe = teardown_probe;
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
ep->iomgr = NULL; /* OFF path: endpoint_fini only, no unmonitor */
ep->fd = -1;
ep->data = calloc(1, 128);
TEST_ASSERT_NOT_NULL(ep->data);
/* One-peer election context with this ep as the loser vote-ep. */
ha_elect_ctx_t *ectx = calloc(1, sizeof(*ectx) + sizeof(ha_vote_entry_t));
TEST_ASSERT_NOT_NULL(ectx);
ectx->n_peers = 1;
ectx->votes[0].ep = ep;
atomic_store_explicit(&ha.elect_ctx, ectx, memory_order_release);
raft_cb_become_follower_for_test(&ha, 5);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_teardown_done),
"OFF path must NOT run the async teardown vtask");
TEST_ASSERT_NULL_MESSAGE(ectx->votes[0].ep,
"loser loop must clear votes[i].ep after endpoint_fini");
/* OFF path called endpoint_fini(ep) (freed ep->data) but NOT free(ep) — the
* legacy per-lost-election leak. Free the struct here so the test is clean. */
free(ep);
atomic_store_explicit(&ha.elect_ctx, NULL, memory_order_release);
free(ectx);
/* become_follower armed the heartbeat watchdog (c=NULL) — release its ctx. */
ha_disarm_heartbeat_timeout_vtask(&ha);
ha_vep_teardown_probe = NULL;
ha_fini(&ha);
printf("test_vep_off_path_uses_plain_fini: PASS\n");
}
/* ── Test (e): DRAIN — mark blocks until an in-flight dispatch finishes ───────
*
* Closes the TOCTOU window: a dispatch that read tearing_down as false but is
* preempted before its enqueue must NOT be able to enqueue after the teardown
* vtask. endpoint_mark_tearing_down drains ep->dispatch_inflight to 0 (after
* setting the flag) before returning, so the teardown caller enqueues the
* teardown vtask only after every such in-flight dispatch has finished enqueuing
* → the peer-control vtask is strictly FIFO-before the teardown vtask.
*
* Deterministic: a predispatch probe parks thread A inside the dispatch_inflight
* bracket (counter == 1, not yet enqueued). Thread B runs the teardown seam; we
* assert its mark/drain does NOT return while A is parked, then release A and
* assert mark returns and the peer-control body ran before the teardown probe. */
typedef struct { endpoint_t *ep; _Atomic int done; } dispatch_arg_t;
typedef struct { ha_state_t *ha; endpoint_t *ep; _Atomic int done; } teardown_arg_t;
/* Layout-compatible stand-in for vote_ep_ctx_t: the inline routing body reads
* only the first member (ha) before the default-drop path frees the packet. */
struct mini_vctx { ha_state_t *ha; char pad[160]; };
static void *drain_dispatch_thread(void *p)
{
dispatch_arg_t *a = (dispatch_arg_t *)p;
io_packet_t *pkt = repl_catchup_done_pkt_for_test();
peer_ctrl_msg_handler_for_test(a->ep, pkt); /* parks in predispatch probe */
atomic_store(&a->done, 1);
return NULL;
}
static void *drain_teardown_thread(void *p)
{
teardown_arg_t *b = (teardown_arg_t *)p;
ha_vote_ep_teardown_for_test(b->ha, b->ep); /* mark drains, then enqueues */
atomic_store(&b->done, 1);
return NULL;
}
static void test_vep_mark_drains_inflight_dispatch(void)
{
reset_oracles();
gate_init();
ha_state_t ha;
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
/* CTRL_INIT → the inline body hits the default case (drop + free pkt) so it
* never decodes, keeping the stand-in vote_ep_ctx_t sufficient. */
atomic_store_explicit(&ha.role, (int32_t)CTRL_INIT, memory_order_release);
vtaskmgr_t pool;
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&pool, 2));
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&pool));
ha.pool = &pool;
iomgr_t iomgr;
TEST_ASSERT_EQUAL_INT(0, iomgr_init(&iomgr));
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
ep->iomgr = &iomgr;
ep->fd = -1;
struct mini_vctx *v = calloc(1, sizeof(*v));
TEST_ASSERT_NOT_NULL(v);
v->ha = &ha;
ep->data = v;
ha_set_peerctrl_dispatch_for_test(1);
ha_peerctrl_predispatch_probe = predispatch_park_probe;
ha_peerctrl_body_probe = body_seq_probe;
ha_vep_teardown_probe = teardown_probe;
/* Thread A: dispatch parks in the drained window (inflight == 1). */
dispatch_arg_t aarg = { .ep = ep };
atomic_store(&aarg.done, 0);
pthread_t ta;
TEST_ASSERT_EQUAL_INT(0, pthread_create(&ta, NULL, drain_dispatch_thread, &aarg));
TEST_ASSERT_EQUAL_INT_MESSAGE(0, gate_wait_entered(2000),
"dispatch must reach the predispatch window and park (inflight=1)");
/* Thread B: teardown mark must BLOCK draining on A. */
teardown_arg_t barg = { .ha = &ha, .ep = ep };
atomic_store(&barg.done, 0);
pthread_t tb;
TEST_ASSERT_EQUAL_INT(0, pthread_create(&tb, NULL, drain_teardown_thread, &barg));
TEST_ASSERT_EQUAL_INT_MESSAGE(-1, poll_until(&barg.done, 1, 150),
"endpoint_mark_tearing_down must block while a dispatch is in-flight");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_teardown_done),
"teardown vtask must not be enqueued/run while mark is still draining");
/* Release A → it completes its enqueue + decrements → B's drain ends. */
gate_open();
TEST_ASSERT_EQUAL_INT_MESSAGE(0, poll_until(&barg.done, 1, 2000),
"mark must return once the in-flight dispatch finishes its enqueue");
pthread_join(ta, NULL);
pthread_join(tb, NULL);
/* Both vtasks run on the pool; peer-control is FIFO-before teardown. */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, poll_until(&g_teardown_done, 1, 2000),
"teardown vtask must run");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_body_ran),
"the in-flight peer-control body must have run");
TEST_ASSERT_GREATER_THAN_INT_MESSAGE(atomic_load(&g_body_seq),
atomic_load(&g_teardown_seq),
"peer-control vtask must be FIFO-before the teardown vtask");
vtaskmgr_fini(&pool);
iomgr_fini(&iomgr);
ha_peerctrl_predispatch_probe = NULL;
ha_peerctrl_body_probe = NULL;
ha_vep_teardown_probe = NULL;
ha_set_peerctrl_dispatch_for_test(0);
ha.pool = NULL;
ha_fini(&ha);
gate_destroy();
printf("test_vep_mark_drains_inflight_dispatch: PASS\n");
}
/* ── Entry point ─────────────────────────────────────────────────────────── */
void test_ha_vep_teardown(void)
{
RUN_TEST(test_vep_teardown_runs_after_inflight);
RUN_TEST(test_vep_late_event_dropped);
RUN_TEST(test_vep_teardown_vtask_disarms_unblock);
RUN_TEST(test_vep_off_path_uses_plain_fini);
RUN_TEST(test_vep_mark_drains_inflight_dispatch);
}

13
src/include/netio.h
View file

@ -4,6 +4,7 @@
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <sys/epoll.h>
@ -72,7 +73,14 @@ typedef struct endpoint_t {
int idx;
int flags;
unsigned int busy : 1;
unsigned int tearing_down : 1;
unsigned int ssl_handshake_done : 1;
/* bug-1197: count of peer-control dispatch attempts currently between the
* tearing_down flag-read and the completion of their pool enqueue. Bitfields
* cannot be _Atomic, so this is a full atomic int. endpoint_mark_tearing_down
* drains this to 0 (after setting the flag) so no dispatch can enqueue a
* same-ep vtask after teardown begins. */
_Atomic int dispatch_inflight;
dataq_t outq;
uint64_t seq;
conn_handler_t conn_handler;
@ -130,6 +138,11 @@ extern uint64_t endpoint_getseq(endpoint_t*);
extern void endpoint_block_tasks(endpoint_t*);
extern void endpoint_unblock_tasks(endpoint_t*);
extern int endpoint_is_task_blocked(endpoint_t*);
/* bug-1197: mark an endpoint as undergoing teardown so a late same-batch event
* delivered by the io_worker after iomgr_unmonitor is dropped instead of
* dispatched onto the vtask pool (which could outlive the ep). */
extern void endpoint_mark_tearing_down(endpoint_t*);
extern int endpoint_is_tearing_down(endpoint_t*);
extern int endpoint_set_owner(endpoint_t*, void*);
extern int endpoint_set_msg_handler(endpoint_t*, msg_handler_t);
extern int endpoint_set_err_handler(endpoint_t*, err_handler_t);

43
src/lib/netio.c
View file

@ -5,7 +5,9 @@
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <poll.h>
#include <sched.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -725,6 +727,8 @@ static int do_endpoint_init(endpoint_t* ep)
ep->iomgr = NULL;
ep->ssl = NULL;
ep->busy = 0;
ep->tearing_down = 0;
ep->dispatch_inflight = 0;
ep->ssl_handshake_done = 0;
ep->msg_handler = NULL;
ep->err_handler = NULL;
@ -1008,6 +1012,45 @@ int endpoint_is_task_blocked(endpoint_t* ep)
return (busy != 0);
}
/* bug-1197: set the tearing-down flag, then DRAIN in-flight dispatches.
*
* Once set, tearing_down is never cleared (the ep is about to be endpoint_fini'd
* + freed). The flag is set under ep->lock so any late-event dispatch decision
* (which reads it under ep->lock via endpoint_is_tearing_down) observes it.
*
* The drain closes the TOCTOU window: a dispatch increments dispatch_inflight
* (seq_cst) BEFORE reading the flag and decrements it AFTER its pool enqueue
* completes. We release ep->lock before spinning (the dispatch's flag-read
* takes ep->lock, so holding it here would deadlock the very dispatch we wait
* on), then spin until the counter is 0. When this returns, every dispatch
* that read the flag as false has finished its enqueue; since the teardown
* caller enqueues the teardown vtask only AFTER this returns, no same-ep
* peer-control vtask can be enqueued after the teardown vtask. The drained
* window is just a calloc + non-blocking enqueue on the single io_worker, so
* the spin is bounded and short. */
void endpoint_mark_tearing_down(endpoint_t* ep)
{
mylock_lock(&(ep->lock));
ep->tearing_down = 1;
mylock_unlock(&(ep->lock));
while (atomic_load_explicit(&ep->dispatch_inflight, memory_order_seq_cst)
> 0) {
sched_yield();
}
}
int endpoint_is_tearing_down(endpoint_t* ep)
{
int tearing;
mylock_lock(&(ep->lock));
tearing = ep->tearing_down;
mylock_unlock(&(ep->lock));
return (tearing != 0);
}
void endpoint_fini(endpoint_t* ep)
{
dataq_fini(&(ep->outq));