fix(netio): bug-1447 same-batch endpoint use-after-free in io_worker dispatch

Raft-drive-async's reconnect-offload burst (Task 2.1/2.2) exposed a
latent race in io_worker's epoll dispatch loop: ha.c's
on_standby_connected() evicts a stale duplicate peer channel via
iomgr_unmonitor()+endpoint_unref(), but epoll_ctl(DEL) only stops
FUTURE epoll_wait() calls -- it does not invalidate an event for that
same endpoint already sitting in the CURRENT batch's events[] array.
A later event in the same batch then dereferences the freed
endpoint_t/SSL*, producing the ASAN SEGV observed deep inside
OpenSSL's ssl3_read_bytes during raft-drive-async's round-9 soak.

The original "concurrent SSL_do_handshake from two io_worker threads"
hypothesis was wrong and is retracted -- iomgr_start() creates exactly
one io_worker thread per iomgr_t.

Fix: extract io_worker_dispatch_batch(), which takes endpoint_ref() on
every endpoint in the batch before dispatching any of them and
endpoint_unref()s each only after the whole batch completes -- the
same lock-free ref/unref bracket already used for bug-1298 and
INVOKE_MSG_HANDLER, just never applied to io_worker's own loop. Pure
atomics, no locks, no blocking added to the hot path.

Adds test_netio_batch_dispatch_uaf.c, which drives the real dispatch
loop with a synthetic epoll_event[] batch (TDD RED confirmed via ASAN
heap-use-after-free before the fix, GREEN after). Verified further
with two independent clean 10-round live sequential HA soaks (33 and
36 jobs, zero loss) and full ASAN suite runs (xcelib_test 294/0,
controller_unit_test 972/0). raft-drive-async's kiro-validate-impl
re-run flips NO-GO to GO.

src/lib/testsuite/test_netio_batch_dispatch_uaf.c

@@ -0,0 +1,165 @@
+/*
+ * test_netio_batch_dispatch_uaf.c — regression test for bug-1447.
+ *
+ * bug-1447: an ASAN DEADLYSIGNAL SEGV (near-null READ) was observed inside
+ * OpenSSL's ssl3_read_bytes <- tls_get_message_header <- state_machine <-
+ * handle_ssl_handshake (netio.c) <- io_handler <- io_worker, during a
+ * raft-drive-async reconnect-offload burst right after a new primary
+ * election.
+ *
+ * Root cause: io_worker's epoll dispatch loop processes ALL ready events
+ * from ONE epoll_wait() batch in a plain for-loop calling io_handler() per
+ * event. A handler for an EARLIER event in the batch (in production: ha.c's
+ * on_standby_connected(), which evicts a stale duplicate outbound channel to
+ * the same peer via iomgr_unmonitor() + endpoint_unref()) can drop the LAST
+ * reference on a DIFFERENT endpoint that also has a ready event LATER in the
+ * SAME batch. iomgr_unmonitor()'s epoll_ctl(EPOLL_CTL_DEL) only stops a
+ * FUTURE epoll_wait() from returning events for that fd — it does not
+ * invalidate an event already sitting in the batch's stack-local
+ * `events[]` array. endpoint_unref() reaching refcnt 0 immediately
+ * SSL_free()s ep->ssl and free()s the endpoint_t itself, so the later
+ * io_handler() call in the SAME loop dereferences freed memory — a
+ * heap-use-after-free that manifests deep inside OpenSSL when the freed
+ * (or poisoned/reused) ep->ssl pointer is handed to SSL_do_handshake().
+ *
+ * This test reproduces the SAME-BATCH mechanism deterministically, without
+ * any real socket/SSL/epoll machinery, by driving
+ * io_worker_dispatch_batch_for_test() (the XCE_TESTING seam exposing
+ * io_worker's real batch-dispatch loop) with a synthetic 2-event batch:
+ *
+ *   events[0] -> ep_evictor: a conn_handler that endpoint_unref()s ep_victim,
+ *                simulating on_standby_connected's stale-duplicate eviction.
+ *   events[1] -> ep_victim:  refcnt==1 (its only ref is the one dropped by
+ *                ep_evictor's handler) — a conn_handler stub that just
+ *                records that it ran.
+ *
+ * RED (pre-fix): ep_evictor's handler drops ep_victim's refcount to 0,
+ * freeing it (SSL_free + free(ep)) DURING events[0]'s processing. Processing
+ * events[1] then dereferences the freed ep_victim inside io_handler() — under
+ * this project's mandatory ASAN build this aborts the process
+ * (heap-use-after-free), which is the expected RED failure mode for this
+ * exact bug class (the original bug was itself only caught by ASAN).
+ *
+ * GREEN (post-fix): io_worker_dispatch_batch takes a reference on every
+ * endpoint in the batch before dispatching any of them and releases it only
+ * after the whole batch is processed, so ep_evictor's endpoint_unref() drops
+ * ep_victim from 2 (owner + batch) to 1 (batch only) — not to 0. ep_victim
+ * survives to have events[1] safely dispatched, and is only freed by
+ * io_worker's own trailing unref after the whole batch completes.
+ */
+
+#define _GNU_SOURCE
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/epoll.h>
+
+#include <unity/unity.h>
+#include "netio.h"
+
+/* ── Static state for the injected conn_handlers ─────────────────────── */
+
+static endpoint_t *g_victim_ep       = NULL;
+static int         g_victim_ran      = 0;
+static int         g_evictor_ran     = 0;
+
+static void reset_state(void)
+{
+    g_victim_ep   = NULL;
+    g_victim_ran  = 0;
+    g_evictor_ran = 0;
+}
+
+/* Simulates ha.c's on_standby_connected: while handling ITS OWN event, tears
+ * down a DIFFERENT, already-batched endpoint (the stale duplicate). */
+static int evictor_conn_handler(struct epoll_event *p)
+{
+    (void)p;
+    g_evictor_ran = 1;
+    /* Drop the victim's only ref — pre-fix this frees it immediately. */
+    endpoint_unref(g_victim_ep);
+    return 0;
+}
+
+/* The "stale duplicate" endpoint's handler — if this runs on a still-valid
+ * endpoint, the fix held. If ep was already freed, this call itself is
+ * running on freed memory (the crash already happened by the time we get
+ * here, or ASAN has already aborted the process). */
+static int victim_conn_handler(struct epoll_event *p)
+{
+    (void)p;
+    g_victim_ran = 1;
+    return 0;
+}
+
+/*
+ * Test: processing an earlier batch event that evicts a later-batched
+ * endpoint's last reference must NOT free that endpoint before its own
+ * event is dispatched.
+ */
+static void test_batch_dispatch_survives_cross_endpoint_eviction(void)
+{
+    reset_state();
+
+    endpoint_t *ep_evictor = endpoint_get();
+    endpoint_t *ep_victim  = endpoint_get();
+    TEST_ASSERT_NOT_NULL(ep_evictor);
+    TEST_ASSERT_NOT_NULL(ep_victim);
+
+    /* Both endpoints are past handshake and dispatch through conn_handler
+     * (the msg_handler-less path in io_handler), exactly like an established
+     * outbound standby control channel. */
+    ep_evictor->ssl_handshake_done = 1;
+    ep_evictor->msg_handler        = NULL;
+    ep_evictor->conn_handler       = evictor_conn_handler;
+
+    ep_victim->ssl_handshake_done = 1;
+    ep_victim->msg_handler        = NULL;
+    ep_victim->conn_handler       = victim_conn_handler;
+
+    /* ep_victim starts with exactly ONE reference (its "owner" ref from
+     * endpoint_get()) — the same ref on_standby_connected's eviction path
+     * drops via endpoint_unref() in production (ha.c:8327). */
+    TEST_ASSERT_EQUAL_INT(1, atomic_load(&ep_victim->refcnt));
+
+    g_victim_ep = ep_victim;
+
+    struct epoll_event events[2];
+    memset(events, 0, sizeof(events));
+    events[0].events   = EPOLLIN;
+    events[0].data.ptr = ep_evictor;
+    events[1].events   = EPOLLIN;
+    events[1].data.ptr = ep_victim;
+
+    /* Drive the REAL io_worker batch-dispatch loop (via the XCE_TESTING
+     * seam) with this synthetic 2-event batch. */
+    io_worker_dispatch_batch_for_test(events, 2);
+
+    /* Both handlers must have run... */
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_evictor_ran,
+        "the evictor's own event must have been dispatched");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_victim_ran,
+        "bug-1447: the victim endpoint's event must be safely dispatched "
+        "even though an EARLIER event in the same batch dropped its last "
+        "reference — reaching this point without an ASAN heap-use-after-free "
+        "abort IS the regression proof");
+
+    /* ...and ep_victim must be gone now (unref'd to 0 exactly once the whole
+     * batch finished), not leaked and not double-freed. Nothing further to
+     * assert on it directly (it is freed memory), but a clean process exit
+     * (no ASAN report) for the whole test binary is the authoritative check,
+     * verified by the test harness's exit code. */
+
+    endpoint_unref(ep_evictor);
+
+    printf("test_batch_dispatch_survives_cross_endpoint_eviction: PASS\n");
+}
+
+/* ── Test suite entry point ───────────────────────────────────────── */
+
+void test_netio_batch_dispatch_uaf(void)
+{
+    printf("\n>> netio io_worker batch-dispatch UAF tests (bug-1447)\n\n");
+
+    RUN_TEST(test_batch_dispatch_survives_cross_endpoint_eviction);
+}