From 6d510d8cf4ec12bff9838ffa46ac639e433a79ce Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 26 Jun 2026 17:37:26 +0000 Subject: [PATCH] fix(raft-drive-async): endpoint-quiescence for bug-1197 vote-ep teardown (task 3.4a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01S6fDCtXkkEgupu7rUah1y2 --- .kiro/specs/raft-drive-async/tasks.md | 9 +- .wolf/buglog.json | 189 +++++- .wolf/memory.md | 30 + src/controller/ha.c | 190 +++++- src/controller/ha.h | 21 + src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_ha_vep_teardown.c | 549 ++++++++++++++++++ src/include/netio.h | 13 + src/lib/netio.c | 43 ++ 10 files changed, 1037 insertions(+), 10 deletions(-) create mode 100644 src/controller/testsuite/test_ha_vep_teardown.c diff --git a/.kiro/specs/raft-drive-async/tasks.md b/.kiro/specs/raft-drive-async/tasks.md index 87607443..17532833 100644 --- a/.kiro/specs/raft-drive-async/tasks.md +++ b/.kiro/specs/raft-drive-async/tasks.md @@ -103,7 +103,8 @@ 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.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. - 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_ @@ -150,3 +151,9 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - **WHY DORMANT (flag-ON gate, owned by 3.4)**: `snapshot_lane_fn` captures a BORROWED standby endpoint (unlike reconnect, which owns its ep). With the flag ON the send runs on a lane thread while the io_worker may RDHUP-`endpoint_fini` that ep → cross-thread UAF (bug-1197 family). Flag OFF keeps send+teardown serialized on the io_worker. **3.4 is the single gate** that flips BOTH `g_drive_async_snapshot_offload` AND `g_drive_async_peerctrl_dispatch` ON, after establishing the endpoint-lifetime discipline (refcount or lane-serialized teardown) and proving liveness under pool saturation. Caught + fixed pre-commit: the offload flag was initially authored DEFAULT-ON (would have shipped the UAF) → corrected to OFF in both ha.c and the ha.h accessor doc (bug-1210). - **3.3 (landed DORMANT — checkbox `[~]` until 3.4 gate)**: requester-less config-push transport offload, flag `g_drive_async_configpush_offload` **DEFAULT 0/OFF** (toggle `ha_set_configpush_offload_for_test`). `ha_config_push_offloadable(c, bytes, blen, sha256, new_tag)` routes inline (flag OFF — verbatim `push_config_transport_only`, caller still owns+frees `bytes`) or to `slowlane_submit`+`configpush_lane_fn` (flag ON — DEEP-COPIES `bytes`+sha into a ctx the lane frees). Single cluster-wide single-flight via scalar `ha_state_t.configpush_inflight` CAS + lane key=1 (config push is a cluster-wide fan-out, not per-peer); flag cleared on all terminal paths (alloc-fail×2, -EEXIST, hard-error, lane completion). Send body factored into `ha_config_push_do_send` (production → `push_config_transport_only`; XCE_TESTING → `ha_configpush_probe` stub since cluster.c/transport isn't linked in test). THREE requester-less drive-thread call sites converted: `raft_cb_become_leader` (promotion, both bytes/NULL branches), the `HA_REPL_PUSH_CONFIG` heartbeat case, and the direct-CTRL_PRIMARY FSM promotion branch (~ha.c:4476). UNCHANGED: `push_config_to_standbys` admin-upgrade barrier (has a waiting requester) and `join.c`. Tests: `test_configpush_offload_runs_on_lane`, `test_configpush_inline_when_flag_off`, `test_configpush_offload_dedups` (test_ha_liveness.c). Same BORROWED-ENDPOINT/bug-1197 dormancy rationale as 3.2 (lane reads `c->ha.standby_endpoints`); **3.4 flips this ON together with snapshot + peerctrl flags**. cpp-review: no CRITICAL/HIGH/MEDIUM; controller suite EXIT=0. + +- **3.4a (bug-1197 endpoint-quiescence fix — committed; flags STILL default OFF)**: makes the flag-ON peer-control dispatch path UAF-free so `raft_cb_become_follower` (running as a pool vtask) can free peer vote-eps while other pool workers process peer-control on the same eps. Mechanism (no libxce-wide endpoint refcount; reuses the existing per-ep FIFO + busy serialization): + - **Deferred teardown-as-vtask**: the loser loop, when `ha_peerctrl_dispatch_enabled() && ha->pool`, does per vote-ep `endpoint_mark_tearing_down(ep)` → `iomgr_unmonitor(ep)` → enqueue a `VTASK_IO`/`VTASK_LIVENESS` teardown vtask bound to that ep (`vote_ep_teardown_vtask_entry`). Per-ep busy FIFO makes it run strictly AFTER any in-flight/queued same-ep peer-control vtask; it NULLs `iod->ep` before `endpoint_fini`+`free(ep)` to disarm `run_task`'s post-body `endpoint_unblock_tasks` (which would otherwise touch the freed `ep->lock`). OFF / pool-unwired path is byte-for-byte the legacy `endpoint_fini` (no free). + - **RCU-like drain closes the TOCTOU window**: a new `_Atomic int endpoint_t.dispatch_inflight` (+ `tearing_down:1` bit). The ON dispatch brackets its flag-check+enqueue: `fetch_add(seq_cst)` → if `tearing_down` drop pkt + `fetch_sub` → else enqueue → `fetch_sub`. `endpoint_mark_tearing_down` sets the flag under `ep->lock`, releases, then spins `while(load(dispatch_inflight,seq_cst)>0) sched_yield()`. become_follower marks (drains) BEFORE enqueuing the teardown vtask, so any dispatch that read the flag false is FIFO-before teardown; any later dispatch drops. Could NOT hold `ep->lock` across the enqueue — it inverts with the taskq lock (dispatcher takes taskq→ep in `vtask_eligible`); the counter is the lock-free handshake. Drain is bounded (the held window is just calloc+non-blocking enqueue on the single io_worker). + - Files: `src/include/netio.h` (`tearing_down:1`, `dispatch_inflight`, mark/is helpers), `src/lib/netio.c` (helpers + drain + `do_endpoint_init` zero-init), `src/controller/ha.c` (`vote_ep_teardown_*`, loser-loop async branch, dispatch bracket + late-drop, test seams), `src/controller/ha.h`. Tests: `test_ha_vep_teardown.c` (5 ASan tests: ordering, late-drop, disarm, OFF-path, deterministic drain-blocks). xcelib_test 282/0; controller EXIT=0; ASan-clean. Adversarial cpp-review: no CRITICAL/HIGH; handshake formally sound (SC total order on the counter + monotone `tearing_down` under `ep->lock`); flag-ON path proven race-free. + - **STILL OPEN for the prod flip (4.x)**: the borrowed-ep hazard in `snapshot_lane_fn`/`configpush_lane_fn` is a DISTINCT race (slowlane thread vs io_worker RDHUP teardown, not the vtask-pool FIFO) — needs its own per-standby-ep refcount/lane-serialized teardown before `g_drive_async_snapshot_offload`/`configpush_offload` go ON in production. 3.4b integration tests may enable the flags in-test (no mid-transfer ep teardown, so the lane race isn't exercised). diff --git a/.wolf/buglog.json b/.wolf/buglog.json index 5e83f49e..0a0df4dd 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -19800,15 +19800,20 @@ "error_message": "Wrong reference: 1100 should be 1197", "file": "src/controller/ha.c", "root_cause": "Used \"1100\" instead of \"1197\"", - "fix": "Changed 1100 \u2192 1197", + "fix": "FIXED 2026-06-26 (raft-drive-async 3.4a, behind flag default OFF): deferred teardown-as-vtask on the per-ep FIFO + RCU-like dispatch_inflight drain in endpoint_mark_tearing_down. become_follower marks+unmonitors+drains each vote-ep then enqueues a VTASK_IO teardown bound to that ep (runs strictly after any in-flight same-ep peer-ctrl vtask; NULLs iod->ep to disarm run_task post-body unblock). Closes the UAF for the peer-control dispatch path. NOTE: the snapshot/configpush lane borrowed-ep race is distinct and still open before the prod flag flip.", "tags": [ + "endpoint-quiescence", + "fixed", + "c", + "dispatch_inflight", + "task-3.4a", + "tearing_down", "auto-detected", - "wrong-reference", - "c" + "wrong-reference" ], "related_bugs": [], "occurrences": 2, - "last_seen": "2026-06-26T11:05:02.375Z" + "last_seen": "2026-06-26" }, { "id": "bug-1199", @@ -20043,6 +20048,182 @@ ], "occurrences": 1, "last_seen": "2026-06-26" + }, + { + "id": "bug-1211", + "timestamp": "2026-06-26T15:49:46.434Z", + "error_message": "Function not marked async", + "file": "src/controller/ha.h", + "root_cause": "Function uses await but wasn't declared async", + "fix": "Added async modifier", + "tags": [ + "auto-detected", + "async-fix", + "h" + ], + "related_bugs": [], + "occurrences": 2, + "last_seen": "2026-06-26T15:49:59.034Z" + }, + { + "id": "bug-1212", + "timestamp": "2026-06-26T15:50:27.084Z", + "error_message": "Function not marked async", + "file": "src/controller/ha.c", + "root_cause": "Function uses await but wasn't declared async", + "fix": "Added async modifier", + "tags": [ + "auto-detected", + "async-fix", + "c" + ], + "related_bugs": [], + "occurrences": 3, + "last_seen": "2026-06-26T15:52:41.021Z" + }, + { + "id": "bug-1213", + "timestamp": "2026-06-26T15:51:29.747Z", + "error_message": "Missing guard clause", + "file": "src/controller/ha.c", + "root_cause": "No early return/throw for edge case: ha_configpush_probe != NULL", + "fix": "Added guard clause: if (ha_configpush_probe != NULL)", + "tags": [ + "auto-detected", + "guard-clause", + "c" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T15:51:29.747Z" + }, + { + "id": "bug-1214", + "timestamp": "2026-06-26T15:54:05.463Z", + "error_message": "Function not marked async", + "file": "src/controller/testsuite/test_ha_liveness.c", + "root_cause": "Function uses await but wasn't declared async", + "fix": "Added async modifier", + "tags": [ + "auto-detected", + "async-fix", + "c" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T15:54:05.463Z" + }, + { + "id": "bug-1215", + "timestamp": "2026-06-26T15:57:27.700Z", + "error_message": "Significant refactor of ", + "file": "src/controller/ha.c", + "root_cause": "2 lines replaced/restructured", + "fix": "Rewrote 8\u219210 lines (2 removed)", + "tags": [ + "auto-detected", + "refactor", + "c" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T15:57:27.700Z" + }, + { + "id": "bug-1216", + "timestamp": "2026-06-26T16:54:57.384Z", + "error_message": "Wrong reference: ssl_handshake_done should be tearing_down", + "file": "src/include/netio.h", + "root_cause": "Used \"ssl_handshake_done\" instead of \"tearing_down\"", + "fix": "Changed ssl_handshake_done \u2192 tearing_down", + "tags": [ + "auto-detected", + "wrong-reference", + "h" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T16:54:57.384Z" + }, + { + "id": "bug-1217", + "timestamp": "2026-06-26T16:55:15.332Z", + "error_message": "Wrong reference: ssl_handshake_done should be tearing_down", + "file": "src/lib/netio.c", + "root_cause": "Used \"ssl_handshake_done\" instead of \"tearing_down\"", + "fix": "Changed ssl_handshake_done \u2192 tearing_down", + "tags": [ + "auto-detected", + "wrong-reference", + "c" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T16:55:15.332Z" + }, + { + "id": "bug-1218", + "timestamp": "2026-06-26T16:55:48.690Z", + "error_message": "Missing guard clause", + "file": "src/controller/ha.c", + "root_cause": "No early return/throw for edge case: ectx", + "fix": "Added guard clause: if (ectx)", + "tags": [ + "auto-detected", + "guard-clause", + "c" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T16:55:48.690Z" + }, + { + "id": "bug-1219", + "timestamp": "2026-06-26T17:07:46.423Z", + "error_message": "Significant refactor of ", + "file": "src/lib/netio.c", + "root_cause": "4 lines replaced/restructured", + "fix": "Rewrote 10\u219227 lines (4 removed)", + "tags": [ + "auto-detected", + "refactor", + "c" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T17:07:46.424Z" + }, + { + "id": "bug-1220", + "timestamp": "2026-06-26T17:08:14.641Z", + "error_message": "Significant refactor of ", + "file": "src/controller/ha.c", + "root_cause": "10 lines replaced/restructured", + "fix": "Rewrote 22\u219243 lines (10 removed)", + "tags": [ + "auto-detected", + "refactor", + "c" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T17:08:14.641Z" + }, + { + "id": "bug-1221", + "timestamp": "2026-06-26T17:12:06.544Z", + "error_message": "Missing error handling in struct", + "file": "src/controller/testsuite/test_ha_vep_teardown.c", + "root_cause": "Code path had no error handling \u2014 exceptions would propagate uncaught", + "fix": "Added try/catch block", + "tags": [ + "auto-detected", + "error-handling", + "c" + ], + "related_bugs": [], + "occurrences": 1, + "last_seen": "2026-06-26T17:12:06.544Z" } ] } \ No newline at end of file diff --git a/.wolf/memory.md b/.wolf/memory.md index b3099bc5..fc7f2f34 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -3486,3 +3486,33 @@ | 08:57 | Edited src/controller/ha.c | modified XCE_TESTING() | ~115 | | 09:05 | Edited .kiro/specs/raft-drive-async/tasks.md | inline fix | ~51 | | 09:05 | raft-drive-async 3.3: requester-less config-push SlowLane offload (DORMANT flag-OFF), 3 drive-thread sites + deep-copy bytes ownership | ha.c, ha.h, test_ha_liveness.c | 3 tests pass, ctrl EXIT=0, cpp-review GO no CRIT/HIGH/MED | ~58k | +| 09:54 | Edited src/include/netio.h | 2→3 lines | ~35 | +| 09:55 | Edited src/include/netio.h | 3→8 lines | ~137 | +| 09:55 | Edited src/lib/netio.c | 2→3 lines | ~26 | +| 09:55 | Edited src/lib/netio.c | modified endpoint_is_task_blocked() | ~251 | +| 09:55 | Edited src/controller/ha.c | 3→6 lines | ~99 | +| 09:55 | Edited src/controller/ha.c | added 1 condition(s) | ~308 | +| 09:56 | Edited src/controller/ha.c | added 6 condition(s) | ~1427 | +| 09:56 | Edited src/controller/ha.c | added 2 condition(s) | ~344 | +| 09:57 | Edited src/controller/ha.h | modified ha_vote_ep_teardown_for_test() | ~192 | +| 09:59 | Created src/controller/testsuite/test_ha_vep_teardown.c | — | ~3633 | +| 10:00 | Edited src/controller/testsuite/controller_test.c | 1→2 lines | ~23 | +| 10:00 | Edited src/controller/testsuite/controller_test.c | 3→4 lines | ~26 | +| 10:00 | Edited src/controller/meson.build | 1→2 lines | ~22 | +| 10:06 | Session end: 37 writes across 10 files (test_ha_liveness.c, ha.c, ha.h, persist.c, tasks.md) | 18 reads | ~251937 tok | +| 10:06 | Edited src/include/netio.h | 4→5 lines | ~30 | +| 10:06 | Edited src/include/netio.h | expanded (+6 lines) | ~152 | +| 10:07 | Edited src/lib/netio.c | 3→5 lines | ~27 | +| 10:07 | Edited src/lib/netio.c | 3→4 lines | ~34 | +| 10:07 | Edited src/lib/netio.c | modified endpoint_mark_tearing_down() | ~351 | +| 10:08 | Edited src/controller/ha.c | added 1 condition(s) | ~584 | +| 10:08 | Edited src/controller/ha.c | modified XCE_TESTING() | ~182 | +| 10:08 | Edited src/controller/ha.h | modified ha_peerctrl_predispatch_probe() | ~158 | +| 10:10 | Edited src/controller/testsuite/test_ha_vep_teardown.c | 5→10 lines | ~114 | +| 10:10 | Edited src/controller/testsuite/test_ha_vep_teardown.c | 5→2 lines | ~22 | +| 10:11 | Edited src/controller/testsuite/test_ha_vep_teardown.c | modified body_probe() | ~285 | +| 10:11 | Edited src/controller/testsuite/test_ha_vep_teardown.c | 3→5 lines | ~39 | +| 10:12 | Edited src/controller/testsuite/test_ha_vep_teardown.c | added error handling | ~1403 | +| 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 | diff --git a/src/controller/ha.c b/src/controller/ha.c index f5e0c252..b7434c30 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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 diff --git a/src/controller/ha.h b/src/controller/ha.h index 90d87a52..2d11a8e5 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -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 ────────────────────── */ /* diff --git a/src/controller/meson.build b/src/controller/meson.build index e0bd2e02..6590e6e4 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -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', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 38889f27..a5e660fa 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -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(); } diff --git a/src/controller/testsuite/test_ha_vep_teardown.c b/src/controller/testsuite/test_ha_vep_teardown.c new file mode 100644 index 00000000..8d288b10 --- /dev/null +++ b/src/controller/testsuite/test_ha_vep_teardown.c @@ -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 +#include +#include +#include +#include +#include +#include + +#include + +#include +#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); +} diff --git a/src/include/netio.h b/src/include/netio.h index 78ddab7f..11187479 100644 --- a/src/include/netio.h +++ b/src/include/netio.h @@ -4,6 +4,7 @@ #include #include +#include #include #include #include @@ -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); diff --git a/src/lib/netio.c b/src/lib/netio.c index 35d60f36..c756ca0c 100644 --- a/src/lib/netio.c +++ b/src/lib/netio.c @@ -5,7 +5,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -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));