fix(lib): detach the received packet before handing it to a vtask (bug-1916)

vtask_handle_msg published the VTASK_IO task while ep->fsm_in.pkt still
pointed at the packet; INVOKE_MSG_HANDLER clears the slot only after the
handler returns. A vtask could decode and free the packet (xce_decode)
and wake the client, whose failover reconnect then ran endpoint_fini and
freed the dangling slot again: ASAN heap-use-after-free in free_packet,
seen in the worker and scheduler during failovers under heavy CPU load.

vtask_handle_msg now clears ep->fsm_in.pkt before vtask_enq and marks
the task the packet's sole owner (owns_io_pkt), so never-run tasks free
it at vtaskmgr_fini and an enqueue failure frees it instead of leaking;
xcectx vtask_entry frees it on its early -EINVAL exits. Same take-and-
clear discipline the controller's peer-control path already used.

Tests: test_vtask_pkt_handoff.c (3), RED reproduces the soak's ASAN stack.
libxce 733/0; xce-build all green; loaded /xce-test-ha soak (8 rounds,
40 spinners) PASS with no ASAN reports (it crashed 2/2 before).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

src/include/vtaskmgr.h

@@ -51,12 +51,14 @@ typedef struct {
     /* raft-drive-async task 3.1: when non-zero AND data->vtype == VTASK_IO, the
      * carried io_packet_t (data->u.iodata.pkt) is OWNED by the task and must be
      * freed by taskq_item_killer if the task is destroyed while still queued
-     * (never executed) at vtaskmgr_fini.  Default 0 preserves the legacy
-     * endpoint-owned-packet model used by the libxce client path (xcectx.c's
-     * vtask_handle_msg does NOT clear ep->fsm_in.pkt, so the packet lifetime is
-     * tied to the endpoint, not the task — freeing it here would double-free).
-     * The controller's peer-control dispatch sets this to 1 because it clears
-     * ep->fsm_in.pkt and the task becomes the sole owner. */
+     * (never executed) at vtaskmgr_fini.  Both IO handoffs set it to 1 because
+     * both clear ep->fsm_in.pkt and the task becomes the sole owner: the
+     * controller's peer-control dispatch (ha.c) and, since bug-1916,
+     * vtask_handle_msg (libxce client xcectx.c + controller cluster.c), which
+     * used to leave ep->fsm_in.pkt naming a packet the task's xce_decode freed
+     * until INVOKE_MSG_HANDLER cleared it after the handler returned — a window
+     * in which endpoint_fini double-freed it.  0 only for tasks that do not
+     * own a packet. */
     int           owns_io_pkt;
 } vtask_t;
 

src/lib/meson.build

@@ -229,6 +229,7 @@ libxce_test_src = ['testsuite/test.c',
                    'testsuite/test_ctrlstatus.c',
                    'testsuite/test_vtask_admission.c',
                    'testsuite/test_vtask_enq_race.c',
+                   'testsuite/test_vtask_pkt_handoff.c',
                    'testsuite/test_vtask_order.c',
                    'testsuite/test_vtaskmgr_teardown.c',
                    'testsuite/test_xcealloc.c',

src/lib/testsuite/test.c

@@ -58,6 +58,7 @@ extern void test_wal_wire(void);
 extern void test_ctrlstatus(void);
 extern void test_vtask_admission(void);
 extern void test_vtask_enq_race(void);
+extern void test_vtask_pkt_handoff(void);
 extern void test_vtask_order(void);
 extern void test_vtaskmgr_teardown(void);
 extern void test_xcealloc(void);
@@ -188,6 +189,7 @@ int main(int argc, char* argv[])
     TEST(test_ctrlstatus);
     TEST(test_vtask_admission);
     TEST(test_vtask_enq_race);
+    TEST(test_vtask_pkt_handoff);
     TEST(test_vtask_order);
     TEST(test_vtaskmgr_teardown);
     TEST(test_xcealloc);

src/lib/testsuite/test_vtask_pkt_handoff.c

@@ -0,0 +1,197 @@
+/*
+ * test_vtask_pkt_handoff.c — bug-1916 regression: inbound-packet ownership
+ * handoff from the io_worker to a VTASK_IO vtask (vtask_handle_msg).
+ *
+ * Production sequence (libxce client, xcectx.c handle_msg):
+ *   io_worker: handle_input -> INVOKE_MSG_HANDLER(ep)
+ *                 -> ep->msg_handler(ep) == handle_msg
+ *                      -> vtask_handle_msg: task->pkt = ep->fsm_in.pkt; vtask_enq
+ *                 <- returns
+ *              (ep)->fsm_in.pkt = NULL   <-- the endpoint's slot is cleared
+ *                                            only HERE, after the handler
+ *   vcpu:      vtask_entry -> xce_decode(pkt) FREES pkt (free_packet+FREEUP)
+ *              -> xce_handle_reply -> xce_wakeup wakes the caller thread
+ *   caller:    xcecli_retry_next -> xce_failover_rotate -> xce_ctx_reconnect
+ *              -> endpoint_fini(ctx->ep) -> free_packet(ep->fsm_in.pkt)
+ *
+ * Between vtask_enq publishing the task and the io_worker's post-handler
+ * NULL store, ep->fsm_in.pkt still points at a packet the vtask owns and may
+ * already have freed.  Under CPU load the io_worker is preempted in exactly
+ * that window, the vtask decodes + frees the packet and wakes the caller,
+ * and the caller's endpoint_fini reads the dangling slot: the ASAN
+ * heap-use-after-free in free_packet <- endpoint_fini <- xce_ctx_reconnect
+ * (freed by xce_decode <- vtask_entry) seen in the loaded HA soaks.
+ *
+ * The test drives that interleaving deterministically: it calls
+ * vtask_handle_msg exactly as handle_msg does, then (standing in for the
+ * preempted io_worker, which has not yet run its NULL store) waits for the
+ * vtask to finish decoding — i.e. freeing — the packet, and then runs the
+ * caller's endpoint_fini.  Before the fix that endpoint_fini aborts the test
+ * binary with the bug-1916 ASAN report; after it, vtask_handle_msg has
+ * detached the packet from the endpoint before publishing the task, so the
+ * slot is NULL and endpoint_fini has nothing to free.
+ */
+#include <errno.h>
+#include <stdatomic.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <unity/unity.h>
+
+#include "netio.h"
+#include "request.h"
+#include "vtaskmgr.h"
+
+static atomic_int g_decoded;
+
+/* Mirrors xcectx.c vtask_entry's packet handling: xce_decode consumes (frees)
+ * the packet the task carries. */
+static int handoff_decode_entry(vtask_t* task)
+{
+    io_task_data_t* data = &(task->data->u.iodata);
+    xce_request_t   req;
+
+    memset(&req, 0, sizeof(req));
+    if (xce_decode(data->pkt, &req) == 0) {
+        xce_free_request(&req);
+    }
+    data->pkt = NULL;
+
+    atomic_store(&g_decoded, 1);
+    return 0;
+}
+
+static io_packet_t* encode_nop(void)
+{
+    xce_request_t request;
+    io_packet_t*  pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
+
+    TEST_ASSERT_NOT_NULL(pkt);
+    memset(&request, 0, sizeof(request));
+    request.opcode = OPCode_nop;
+    TEST_ASSERT_EQUAL_INT(1, xce_encode(&request, pkt));
+
+    return pkt;
+}
+
+/* Bounded poll for an atomic flag (liveness only, not a correctness sleep). */
+static int wait_flag(atomic_int* a, int want, int max_ms)
+{
+    for (int i = 0; i < max_ms * 10; i++) {
+        if (atomic_load(a) == want) {
+            return 1;
+        }
+        usleep(100);
+    }
+    return atomic_load(a) == want;
+}
+
+/* Arrange an endpoint exactly as handle_input leaves it when it invokes the
+ * msg handler: a completed inbound fsm holding the received packet. */
+static void arm_inbound_done(endpoint_t* ep, io_packet_t* pkt)
+{
+    ep->fsm_in.state = FSM_STATE_DONE;
+    ep->fsm_in.pkt   = pkt;
+}
+
+/* bug-1916 RED/GREEN: the vtask frees the packet before the io_worker clears
+ * the endpoint slot; endpoint_fini must not touch the freed packet. */
+static void test_endpoint_fini_after_vtask_freed_pkt(void)
+{
+    vtaskmgr_t mgr;
+    endpoint_t ep;
+
+    memset(&ep, 0, sizeof(ep));
+    TEST_ASSERT_TRUE(endpoint_init(&ep, AF_INET, SOCK_STREAM, 0, 0) >= 0);
+
+    TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 2));
+    TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&mgr));
+
+    atomic_store(&g_decoded, 0);
+    arm_inbound_done(&ep, encode_nop());
+
+    /* io_worker: msg handler hands the packet to a vtask ... */
+    TEST_ASSERT_TRUE(vtask_handle_msg(&mgr, (vtask_func_t)handoff_decode_entry,
+                                      &ep) >= 0);
+
+    /* ... and is preempted before INVOKE_MSG_HANDLER's NULL store.  The vtask
+     * decodes (frees) the packet and run_task drops its endpoint reference. */
+    TEST_ASSERT_TRUE_MESSAGE(wait_flag(&g_decoded, 1, 5000),
+                             "decode vtask never ran");
+    for (int i = 0; i < 50000 && atomic_load(&ep.refcnt) != 0; i++) {
+        usleep(100);
+    }
+    TEST_ASSERT_EQUAL_INT(0, atomic_load(&ep.refcnt));
+
+    /* caller thread: xce_ctx_reconnect -> endpoint_fini on the same ep.
+     * Before the fix ep.fsm_in.pkt still named the packet the vtask freed, and
+     * this aborted under ASAN (heap-use-after-free in free_packet <-
+     * endpoint_fini; freed by xce_decode on the vcpu thread) — bug-1916. */
+    endpoint_fini(&ep);
+    TEST_ASSERT_NULL(ep.fsm_in.pkt);
+
+    vtaskmgr_fini(&mgr);
+}
+
+/* The handoff is synchronous: once vtask_handle_msg returns, the endpoint no
+ * longer references the packet, even if the task has not run yet (a queued,
+ * never-executed task owns it and vtaskmgr_fini's killer frees it). */
+static void test_handoff_detaches_before_task_runs(void)
+{
+    vtaskmgr_t   mgr;
+    endpoint_t   ep;
+    io_packet_t* pkt;
+
+    memset(&ep, 0, sizeof(ep));
+    TEST_ASSERT_TRUE(endpoint_init(&ep, AF_INET, SOCK_STREAM, 0, 0) >= 0);
+
+    /* Not started: the task stays queued. */
+    TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 1));
+
+    pkt = encode_nop();
+    arm_inbound_done(&ep, pkt);
+
+    TEST_ASSERT_TRUE(vtask_handle_msg(&mgr, (vtask_func_t)handoff_decode_entry,
+                                      &ep) >= 0);
+    TEST_ASSERT_NULL(ep.fsm_in.pkt);
+
+    endpoint_fini(&ep);
+
+    /* Frees the queued task and (owns_io_pkt) its packet; drops its ep ref. */
+    vtaskmgr_fini(&mgr);
+    TEST_ASSERT_EQUAL_INT(0, atomic_load(&ep.refcnt));
+}
+
+/* Invalid handoffs leave the endpoint's packet where it was. */
+static void test_handoff_rejects_incomplete_fsm(void)
+{
+    vtaskmgr_t   mgr;
+    endpoint_t   ep;
+    io_packet_t* pkt;
+
+    memset(&ep, 0, sizeof(ep));
+    TEST_ASSERT_TRUE(endpoint_init(&ep, AF_INET, SOCK_STREAM, 0, 0) >= 0);
+    TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 1));
+
+    pkt              = encode_nop();
+    ep.fsm_in.state  = FSM_STATE_READ_PKT;
+    ep.fsm_in.pkt    = pkt;
+
+    TEST_ASSERT_EQUAL_INT(-EINVAL,
+                          vtask_handle_msg(&mgr,
+                                           (vtask_func_t)handoff_decode_entry,
+                                           &ep));
+    TEST_ASSERT_EQUAL_PTR(pkt, ep.fsm_in.pkt);
+
+    endpoint_fini(&ep); /* frees the still endpoint-owned packet */
+    vtaskmgr_fini(&mgr);
+}
+
+void test_vtask_pkt_handoff(void)
+{
+    RUN_TEST(test_endpoint_fini_after_vtask_freed_pkt);
+    RUN_TEST(test_handoff_detaches_before_task_runs);
+    RUN_TEST(test_handoff_rejects_incomplete_fsm);
+}

src/lib/vtaskmgr.c

@@ -27,7 +27,9 @@ int vtask_init(vtask_t* task, vtask_func_t entry, void* data)
     task->data  = data;
     task->owner = NULL;
     task->cls   = VTASK_LIVENESS; /* never-gated by default; legacy enq paths */
-    task->owns_io_pkt = 0;        /* legacy: endpoint owns the packet (task 3.1) */
+    /* Set to 1 by every IO handoff that detaches the packet
+     * (vtask_handle_msg, ha.c peer-control). */
+    task->owns_io_pkt = 0;
 
     return 0;
 }
@@ -208,11 +210,11 @@ static void taskq_item_killer(data_item_t* it)
         /* raft-drive-async task 3.1: a task destroyed while still queued (never
          * executed) at vtaskmgr_fini would otherwise leak its carried packet —
          * the ownership transfer (iod->pkt = NULL) only happens inside the
-         * executed entry (peer_ctrl_vtask_entry).  Free the packet here ONLY
-         * when the producer marked the task as the packet's sole owner
-         * (owns_io_pkt); the legacy endpoint-owned model (owns_io_pkt == 0,
-         * e.g. the libxce client path) is left untouched so we never double-free
-         * a packet whose lifetime is the endpoint's.  This runs only on
+         * executed entry (peer_ctrl_vtask_entry, xcectx vtask_entry,
+         * cluster.c myreq_handler).  Every IO handoff now detaches the packet
+         * from the endpoint and marks the task its sole owner (owns_io_pkt,
+         * bug-1916), so free it here; a task without owns_io_pkt carries no
+         * packet it owns and is left untouched.  This runs only on
          * un-executed tasks, so it cannot race the entry's free.  vtask_fini
          * frees only task->data (the wrapper struct), never the packet. */
         if (task->owns_io_pkt && task->data != NULL &&
@@ -608,6 +610,7 @@ int vtask_handle_msg(vtaskmgr_t* mgr, vtask_func_t entry, void* arg)
     endpoint_t*   ep = (endpoint_t*)arg;
     vtask_data_t* data;
     vtask_t*      task;
+    int           rc;
 
     if (ep == NULL) {
         return -EINVAL;
@@ -628,11 +631,39 @@ int vtask_handle_msg(vtaskmgr_t* mgr, vtask_func_t entry, void* arg)
         return -ENOMEM;
     }
 
+    /* bug-1916: hand the inbound packet to the task and DETACH it from the
+     * endpoint before vtask_enq publishes the task.  The task's entry (e.g.
+     * xcectx.c / cluster.c vtask_entry -> xce_decode) frees the packet, and a
+     * vcpu may run it the instant it is published.  INVOKE_MSG_HANDLER clears
+     * ep->fsm_in.pkt only after this handler returns, so leaving the slot set
+     * until then let a preempted io_worker expose a packet the vtask had
+     * already freed: a concurrent endpoint_fini (client failover's
+     * xce_ctx_reconnect, woken by that very vtask's reply) then freed it again
+     * (heap-use-after-free in free_packet <- endpoint_fini).  Ownership is now
+     * single: from here on only the task references the packet, and
+     * owns_io_pkt lets taskq_item_killer free it if the task never runs.
+     * Mirrors ha.c peer_ctrl_msg_handler's take-and-clear (raft-drive-async
+     * task 3.1). */
     data->vtype        = VTASK_IO;
     data->u.iodata.ep  = ep;
     data->u.iodata.pkt = ep->fsm_in.pkt;
+    ep->fsm_in.pkt     = NULL;
 
     vtask_init(task, entry, data);
+    task->owns_io_pkt = 1;
 
-    return vtask_enq(mgr, task);
+    rc = vtask_enq(mgr, task);
+    if (rc < 0) {
+        /* Never published: the packet would otherwise leak (the endpoint no
+         * longer references it).  vtask_fini frees only the data wrapper. */
+        io_packet_t* pkt = data->u.iodata.pkt;
+
+        free_packet(pkt);
+        free(pkt);
+        data->u.iodata.pkt = NULL;
+        vtask_fini(task);
+        free(task);
+    }
+
+    return rc;
 }

src/lib/xcectx.c

@@ -127,16 +127,23 @@ static int vtask_entry(vtask_t* task)
     int             rc   = 0;
     xce_request_t   request;
 
-    if (ep == NULL) {
-        return -EINVAL;
+    /* bug-1916: this task is the packet's sole owner (vtask_handle_msg detached
+     * it from ep->fsm_in.pkt); free it on the early-exit paths too.  An
+     * unowned ep is reachable: failover's xce_ctx_reconnect -> endpoint_fini
+     * clears ep->owner while a reply's task is still queued. */
+    if (ep != NULL) {
+        ctx = (xce_ctx_t*)ep->owner;
     }
-
-    ctx = (xce_ctx_t*)ep->owner;
     if (ctx == NULL) {
+        if (data->pkt != NULL) {
+            free_packet(data->pkt);
+            FREEUP(data->pkt);
+        }
         return -EINVAL;
     }
 
     rc = xce_decode(data->pkt, &request);
+    data->pkt = NULL; /* consumed by xce_decode */
     if (rc < 0) {
         return rc;
     }