fix(lib): reconnect from an endpoint's own task neither waits nor leaks old replies (bug-1917)

A reconnect made from inside a task of the endpoint it re-establishes (a
request handler that failed over) waited out the full 2 s quiesce bound:
its own reference cannot drain, and the old connection's queued tasks are
serialised behind it -- and those then ran after the owner was re-armed,
reaching the new session.

Each VTASK_IO task now records the endpoint's reconnect_gen when queued and
vtask_entry drops a task whose generation is stale; the vtask manager
tracks the endpoint whose task the current thread runs
(vtask_current_io_ep), and xce_ctx_quiesce_ep skips the wait in that case.

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

src/include/netio.h

@@ -170,6 +170,10 @@ typedef struct endpoint_t {
 typedef struct {
     endpoint_t*  ep;
     io_packet_t* pkt;
+    /* ep->reconnect_gen when the message was received (bug-1917): a task of
+     * a connection that has since been re-established is stale and is
+     * dropped instead of being delivered to the new session. */
+    uint64_t     gen;
 } io_task_data_t;
 
 extern int  alloc_packet(io_packet_t*);

src/include/vtaskmgr.h

@@ -109,6 +109,9 @@ extern int vtask_enq(vtaskmgr_t*, vtask_t*);
 extern vtask_t* vtask_deq_eligible(dataq_t* taskq,
                                    int (*is_blocked)(endpoint_t*));
 extern int vtask_handle_msg(vtaskmgr_t*, vtask_func_t, void* arg);
+/* bug-1917: the endpoint whose VTASK_IO task the calling thread is running,
+ * or NULL when it is not running one. */
+extern endpoint_t* vtask_current_io_ep(void);
 
 /* LivenessAdmission (raft-drive-async task 1.6) — pool-level reservation.
  *

src/lib/testsuite/test_ctx_reconnect_quiesce.c

@@ -25,6 +25,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <sys/socket.h>
+#include <time.h>
 #include <unistd.h>
 
 #include <unity/unity.h>
@@ -126,7 +127,88 @@ static void test_reconnect_drains_old_connection_tasks(void)
     close(lfd);
 }
 
+/* bug-1917 follow-up: a reconnect made from inside a task of the endpoint it
+ * re-establishes (a request handler that failed over) cannot drain: its own
+ * task holds a reference, and the old connection's queued tasks are
+ * serialised behind it.  It must not wait out the 2 s bound for nothing, and
+ * those queued tasks -- which run only after it returns, when the owner is
+ * re-armed -- must still never reach the new session. */
+static xce_ctx_t* g_own_ctx;
+static int        g_own_port;
+static atomic_int g_own_rc;
+static atomic_int g_own_ms;
+static atomic_int g_own_done;
+
+static int reconnect_from_own_task(vtask_t* task)
+{
+    io_task_data_t* data = &(task->data->u.iodata);
+    struct timespec t0, t1;
+
+    free_packet(data->pkt);
+    free(data->pkt);
+    data->pkt = NULL;
+
+    /* a stale message of the old connection arrives meanwhile: it is queued
+     * behind this task (same endpoint) */
+    endpoint_t* ep  = g_own_ctx->ep;
+    ep->fsm_in.state = FSM_STATE_DONE;
+    ep->fsm_in.pkt   = encode_nop();
+    INVOKE_MSG_HANDLER(ep);
+
+    clock_gettime(CLOCK_MONOTONIC, &t0);
+    atomic_store(&g_own_rc, xce_ctx_reconnect(g_own_ctx, "127.0.0.1", g_own_port));
+    clock_gettime(CLOCK_MONOTONIC, &t1);
+    atomic_store(&g_own_ms, (int)((t1.tv_sec - t0.tv_sec) * 1000
+                                  + (t1.tv_nsec - t0.tv_nsec) / 1000000));
+    atomic_store(&g_own_done, 1);
+    return 0;
+}
+
+static void test_reconnect_from_own_task_does_not_wait_or_leak(void)
+{
+    xce_ctx_t  ctx;
+    endpoint_t ep;
+    int        lfd = listen_local(&g_own_port);
+
+    atomic_store(&g_handler_calls, 0);
+    atomic_store(&g_own_done, 0);
+    TEST_ASSERT_EQUAL_INT(0, xce_ctx_init(&ctx, "t1917b", "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));
+    TEST_ASSERT_EQUAL_INT(0, xce_ctx_reconnect(&ctx, "127.0.0.1", g_own_port));
+    g_own_ctx = &ctx;
+    TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&ctx.vtaskmgr));
+
+    /* a request handler of this connection that fails over */
+    ep.fsm_in.state = FSM_STATE_DONE;
+    ep.fsm_in.pkt   = encode_nop();
+    TEST_ASSERT_TRUE(vtask_handle_msg(&ctx.vtaskmgr,
+                                      (vtask_func_t)reconnect_from_own_task,
+                                      &ep) >= 0);
+
+    for (int i = 0; i < 50000 && !atomic_load(&g_own_done); i++) {
+        usleep(100);
+    }
+    TEST_ASSERT_TRUE(atomic_load(&g_own_done));
+    usleep(300 * 1000); /* let the queued stale task run */
+
+    TEST_ASSERT_EQUAL_INT(0, atomic_load(&g_own_rc));
+    TEST_ASSERT_TRUE_MESSAGE(atomic_load(&g_own_ms) < 500,
+        "bug-1917: a reconnect from the endpoint's own task must not wait out "
+        "the quiesce bound");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_handler_calls),
+        "bug-1917: an old-connection task queued behind the reconnect must "
+        "not reach the reconnected session");
+
+    xce_ctx_fini(&ctx);
+    endpoint_fini(&ep);
+    close(lfd);
+}
+
 void test_ctx_reconnect_quiesce(void)
 {
     RUN_TEST(test_reconnect_drains_old_connection_tasks);
+    RUN_TEST(test_reconnect_from_own_task_does_not_wait_or_leak);
 }

src/lib/vtaskmgr.c

@@ -73,13 +73,27 @@ void vcpu_idle(vcpu_t* vcpu)
     return;
 }
 
+/* bug-1917: the endpoint of the VTASK_IO task this thread is running. */
+static __thread endpoint_t* t_current_io_ep;
+
+endpoint_t* vtask_current_io_ep(void)
+{
+    return t_current_io_ep;
+}
+
 void run_task(vtask_t* task)
 {
     vtaskmgr_t*   vtaskmgr;
     vtask_data_t* data;
 
     if (task->entry != NULL) {
+        endpoint_t* prev = t_current_io_ep;
+
+        if (task->data != NULL && task->data->vtype == VTASK_IO) {
+            t_current_io_ep = task->data->u.iodata.ep;
+        }
         (task->entry)(task);
+        t_current_io_ep = prev;
     }
 
     data = task->data;
@@ -647,6 +661,8 @@ int vtask_handle_msg(vtaskmgr_t* mgr, vtask_func_t entry, void* arg)
     data->vtype        = VTASK_IO;
     data->u.iodata.ep  = ep;
     data->u.iodata.pkt = ep->fsm_in.pkt;
+    data->u.iodata.gen = atomic_load_explicit(&ep->reconnect_gen,
+                                              memory_order_acquire);
     ep->fsm_in.pkt     = NULL;
 
     vtask_init(task, entry, data);

src/lib/xcectx.c

@@ -139,6 +139,14 @@ static int vtask_entry(vtask_t* task)
         mylock_lock(&(ep->lock));
         ctx = (xce_ctx_t*)ep->owner;
         mylock_unlock(&(ep->lock));
+        /* bug-1917: a message of a connection that has since been
+         * re-established (the task was queued behind a reconnect made from
+         * this endpoint's own task, so the reconnect could not drain it) must
+         * not reach the new session. */
+        if (data->gen != atomic_load_explicit(&ep->reconnect_gen,
+                                              memory_order_acquire)) {
+            ctx = NULL;
+        }
     }
     if (ctx == NULL) {
         if (data->pkt != NULL) {
@@ -396,6 +404,15 @@ static void xce_ctx_quiesce_ep(xce_ctx_t* ctx)
     ep->owner = NULL;
     mylock_unlock(&(ep->lock));
 
+    /* Called from a task of this very endpoint (a request handler that
+     * failed over): its own reference cannot drain until it returns, and the
+     * old connection's queued tasks are serialised behind it, so waiting is
+     * pointless.  Those queued tasks are dropped by their stale
+     * reconnect_gen in vtask_entry instead. */
+    if (vtask_current_io_ep() == ep) {
+        return;
+    }
+
     for (waited = 0; waited < XCE_RECONNECT_QUIESCE_MS; waited++) {
         if (atomic_load_explicit(&ep->refcnt, memory_order_acquire) <= base) {
             return;