fix(lib): quiesce the old endpoint before xce_ctx_reconnect re-inits it (bug-1917)

Work dispatched for the old connection (the io_worker's epoll batch,
queued VTASK_IO reply tasks) could still run while the reconnect tore
down and re-armed ctx->ep; a queued reply of the old connection then
reached the new session. The reconnect now unmonitors the old fd,
clears ep->owner (stale tasks are dropped) and waits for the endpoint's
references to drain, bounded at 2 s so a caller that is itself one of
those tasks cannot deadlock. vtask_entry reads ep->owner under ep->lock.

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

src/lib/meson.build

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

src/lib/testsuite/test.c

@@ -59,6 +59,7 @@ 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_ctx_reconnect_quiesce(void);
 extern void test_vtask_order(void);
 extern void test_vtaskmgr_teardown(void);
 extern void test_xcealloc(void);
@@ -190,6 +191,7 @@ int main(int argc, char* argv[])
     TEST(test_vtask_admission);
     TEST(test_vtask_enq_race);
     TEST(test_vtask_pkt_handoff);
+    TEST(test_ctx_reconnect_quiesce);
     TEST(test_vtask_order);
     TEST(test_vtaskmgr_teardown);
     TEST(test_xcealloc);

src/lib/testsuite/test_ctx_reconnect_quiesce.c

@@ -0,0 +1,132 @@
+/*
+ * test_ctx_reconnect_quiesce.c — bug-1917 regression: xce_ctx_reconnect must
+ * quiesce the old endpoint before re-initialising it.
+ *
+ * xce_ctx_reconnect re-inits ctx->ep (an endpoint embedded in the client
+ * context) on the caller's thread.  Work already dispatched for the OLD
+ * connection — the io_worker's in-flight batch, and VTASK_IO reply tasks still
+ * queued on the context's vtask pool — holds endpoint references and reads
+ * ep->owner.  Before the fix nothing waited for that work: a reply task of the
+ * old connection that ran after the reconnect found ep->owner re-armed to the
+ * context and was delivered to the NEW session (and the io_worker could still
+ * be inside SSL_read / handle_output on the fd and SSL being freed).
+ *
+ * The test queues an inbound packet for the old connection exactly as
+ * handle_input does (INVOKE_MSG_HANDLER -> handle_msg -> vtask_handle_msg)
+ * while the vtask pool is stopped, starts the pool 100 ms into a second
+ * reconnect, and asserts the stale task never reaches the context's request
+ * handler.
+ */
+#include <arpa/inet.h>
+#include <errno.h>
+#include <netinet/in.h>
+#include <pthread.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"
+#include "xcectx.h"
+
+static atomic_int g_handler_calls;
+
+static int counting_handler(xce_request_t* req, void* args)
+{
+    (void)req;
+    (void)args;
+    atomic_fetch_add(&g_handler_calls, 1);
+    return 0;
+}
+
+static int listen_local(int* port_out)
+{
+    struct sockaddr_in sa;
+    socklen_t          len = sizeof(sa);
+    int                fd  = socket(AF_INET, SOCK_STREAM, 0);
+
+    TEST_ASSERT_TRUE(fd >= 0);
+    memset(&sa, 0, sizeof(sa));
+    sa.sin_family      = AF_INET;
+    sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    sa.sin_port        = 0;
+    TEST_ASSERT_EQUAL_INT(0, bind(fd, (struct sockaddr*)&sa, sizeof(sa)));
+    TEST_ASSERT_EQUAL_INT(0, listen(fd, 8));
+    TEST_ASSERT_EQUAL_INT(0, getsockname(fd, (struct sockaddr*)&sa, &len));
+    *port_out = ntohs(sa.sin_port);
+    return fd;
+}
+
+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;
+}
+
+static void* start_pool_later(void* arg)
+{
+    usleep(100 * 1000);
+    vtaskmgr_start((vtaskmgr_t*)arg);
+    return NULL;
+}
+
+static void test_reconnect_drains_old_connection_tasks(void)
+{
+    xce_ctx_t  ctx;
+    endpoint_t ep;
+    pthread_t  th;
+    int        port;
+    int        lfd = listen_local(&port);
+
+    atomic_store(&g_handler_calls, 0);
+    TEST_ASSERT_EQUAL_INT(0, xce_ctx_init(&ctx, "t1917", "key"));
+    memset(&ep, 0, sizeof(ep));
+    TEST_ASSERT_TRUE(endpoint_init(&ep, AF_INET, SOCK_STREAM, 0, 0) >= 0);
+    TEST_ASSERT_EQUAL_INT(0, xce_ctx_set_endpoint(&ctx, &ep));
+    TEST_ASSERT_EQUAL_INT(0, xce_ctx_set_handler(&ctx, counting_handler, NULL));
+
+    /* First connection: arms owner + handle_msg on the endpoint. */
+    TEST_ASSERT_EQUAL_INT(0, xce_ctx_reconnect(&ctx, "127.0.0.1", port));
+
+    /* A message of the OLD connection is dispatched (io_worker path) while
+     * the pool is busy (here: not started), so its task is still queued. */
+    ep.fsm_in.state = FSM_STATE_DONE;
+    ep.fsm_in.pkt   = encode_nop();
+    INVOKE_MSG_HANDLER(&ep);
+    TEST_ASSERT_NULL(ep.fsm_in.pkt);
+    TEST_ASSERT_EQUAL_INT(1, atomic_load(&ep.refcnt));
+
+    /* Failover reconnect; the pool gets to run 100 ms into it. */
+    TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, start_pool_later,
+                                            &ctx.vtaskmgr));
+    TEST_ASSERT_EQUAL_INT(0, xce_ctx_reconnect(&ctx, "127.0.0.1", port));
+    pthread_join(th, NULL);
+
+    usleep(300 * 1000); /* let a still-queued stale task run (pre-fix) */
+    TEST_ASSERT_EQUAL_INT_MESSAGE(
+        0, atomic_load(&g_handler_calls),
+        "bug-1917: a task of the old connection must not be delivered to the "
+        "reconnected session");
+    TEST_ASSERT_EQUAL_INT(0, atomic_load(&ep.refcnt));
+    TEST_ASSERT_EQUAL_PTR(&ctx, ep.owner);
+
+    xce_ctx_fini(&ctx);
+    endpoint_fini(&ep);
+    close(lfd);
+}
+
+void test_ctx_reconnect_quiesce(void)
+{
+    RUN_TEST(test_reconnect_drains_old_connection_tasks);
+}

src/lib/xcectx.c

@@ -1,6 +1,8 @@
 #include <errno.h>
+#include <stdatomic.h>
 #include <string.h>
 #include <sys/socket.h>
+#include <unistd.h>
 
 #include "config.h" /* DEFAULT_CONNECT_TIMEOUT_MS */
 #include "dataq.h"
@@ -132,7 +134,11 @@ static int vtask_entry(vtask_t* task)
      * 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) {
+        /* Under ep->lock: xce_ctx_reconnect clears/re-arms the owner while a
+         * task of the old connection may still run (bug-1917). */
+        mylock_lock(&(ep->lock));
         ctx = (xce_ctx_t*)ep->owner;
+        mylock_unlock(&(ep->lock));
     }
     if (ctx == NULL) {
         if (data->pkt != NULL) {
@@ -360,6 +366,47 @@ int xce_ctx_start(xce_ctx_t* ctx)
  * (crash at netio.c endpoint_connect); without re-arming the handlers + monitor
  * the reconnected endpoint would never route replies to ctx->wbuf.
  */
+/* Upper bound on waiting for the old connection's in-flight work.  The wait
+ * normally ends within one vtask run; the bound only matters if the caller is
+ * itself a task of this endpoint (its own reference never drains), where the
+ * reconnect proceeds as it did before bug-1917 rather than deadlocking. */
+#define XCE_RECONNECT_QUIESCE_MS 2000
+
+/* xce_ctx_quiesce_ep (bug-1917): stop new work on ctx->ep and wait out the
+ * work already dispatched for the old connection before the caller
+ * re-initialises the endpoint.  Unmonitor the old fd so the io_worker gets no
+ * further events for it; clear the owner so reply tasks of the old connection
+ * still queued on the vtask pool are dropped (vtask_entry frees their packet)
+ * instead of reaching the reconnected session; then wait until the endpoint's
+ * reference count is back at its base value -- the io_worker holds a reference
+ * for the whole epoll batch it is dispatching, and every queued task holds
+ * one -- so nothing still touches the old fd, SSL or packets when
+ * endpoint_fini runs. */
+static void xce_ctx_quiesce_ep(xce_ctx_t* ctx)
+{
+    endpoint_t* ep   = ctx->ep;
+    int         base = ep->heap_alloc ? 1 : 0;
+    int         waited;
+
+    if (ep->fd >= 0) {
+        (void)iomgr_unmonitor(&(ctx->iomgr), ep);
+    }
+
+    mylock_lock(&(ep->lock));
+    ep->owner = NULL;
+    mylock_unlock(&(ep->lock));
+
+    for (waited = 0; waited < XCE_RECONNECT_QUIESCE_MS; waited++) {
+        if (atomic_load_explicit(&ep->refcnt, memory_order_acquire) <= base) {
+            return;
+        }
+        usleep(1000);
+    }
+    WARN("xce_ctx_reconnect: old connection still has %d in-flight "
+         "reference(s) after %d ms; reconnecting anyway",
+         atomic_load(&ep->refcnt) - base, XCE_RECONNECT_QUIESCE_MS);
+}
+
 int xce_ctx_reconnect(xce_ctx_t* ctx, char* host, int port)
 {
     int rc;
@@ -368,6 +415,10 @@ int xce_ctx_reconnect(xce_ctx_t* ctx, char* host, int port)
         return -EINVAL;
     }
 
+    /* bug-1917: nothing may still be dispatching the old connection when the
+     * endpoint struct is torn down and re-initialised below. */
+    xce_ctx_quiesce_ep(ctx);
+
     /* Tear down the dead connection (closes fd, frees SSL, clears handlers and
      * resets the endpoint struct — including ep->iomgr). */
     endpoint_fini(ctx->ep);