fix(lib): keep per-endpoint message order in the vtask dispatcher (bug-1864)

The dispatcher checked each queued task's endpoint busy flag one item at a
time, while the vcpu that finished the endpoint's previous message cleared
the flag without the taskq lock (a lock there deadlocked, bug-1282). Within
one scan, task N was skipped (endpoint still busy), the endpoint freed up,
and task N+1 of the SAME endpoint was dequeued first. Standby controllers
then handled WAL catch-up entry 258 before 257 and NACKed (log mismatch):
26 reorders -> ~576 NACKs in a 45-min production-like run, both standbys
at once at 21:39:24, primary lost its write quorum and answered
jobdispatch/join with E_XCE_NOT_READY, the scheduler client failed over and
rejoined, and the rejoin JOBS_PUSH carried lagging committed state
("refusing resync state ... existing job is RUN" WARNs).

dataq gains dataq_deq_ctx() (a per-scan matcher context). The scan is now
vtask_deq_eligible(taskq, is_blocked): it remembers every endpoint found
busy and keeps all later tasks of that endpoint ineligible for the rest of
the scan (past 64 distinct busy endpoints, every later IO task waits for
the next scan) -- per-endpoint FIFO without any new lock.

Test lib/testsuite/test_vtask_order.c injects is_blocked to free the
endpoint mid-scan: task N+1 was dequeued before the fix, none after; other
endpoints and FIFO/generic tasks unaffected. Suites: lib 1/1, controller
5/5, worker 10/10, scheduler 87/87, harness 5/5. Chaotic HA soak 2/2 PASS
with 0 WAL reorders (2 and 5 per run before), no ASAN.

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

src/include/dataq.h

@@ -9,6 +9,8 @@ typedef struct _data_item {
 } data_item_t;
 
 typedef int (*data_item_match_fn_t)(data_item_t*);
+/* like data_item_match_fn_t, plus a caller-owned per-scan context */
+typedef int (*data_item_match_ctx_fn_t)(data_item_t*, void*);
 typedef void (*data_item_killer_t)(data_item_t*);
 
 typedef struct {
@@ -22,6 +24,10 @@ typedef struct {
 
 extern int          dataq_init(dataq_t*, data_item_killer_t);
 extern data_item_t* dataq_deq(dataq_t*, data_item_match_fn_t);
+/* bug-1864: first item matching f, with ctx passed to every f call of this
+ * one scan (under the queue lock) -- lets a matcher remember what it saw
+ * earlier in the same scan */
+extern data_item_t* dataq_deq_ctx(dataq_t*, data_item_match_ctx_fn_t, void*);
 extern int          dataq_enq(dataq_t*, data_item_t*);
 extern void         dataq_fini(dataq_t*);
 

src/include/vtaskmgr.h

@@ -99,6 +99,13 @@ extern int      vtask_init(vtask_t*, vtask_func_t, void*);
 extern void     vtask_fini(vtask_t*);
 
 extern int vtask_enq(vtaskmgr_t*, vtask_t*);
+
+/* bug-1864: dequeue the first dispatchable task from a vtaskmgr task queue.
+ * is_blocked reports whether an endpoint already has a task running
+ * (production: endpoint_is_task_blocked); it is a parameter so a test can
+ * make an endpoint unblock part-way through one scan. */
+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);
 
 /* LivenessAdmission (raft-drive-async task 1.6) — pool-level reservation.

src/lib/dataq.c

@@ -62,6 +62,47 @@ static data_item_t* dataq_deq_first_match(dataq_t* q, data_item_match_fn_t f)
     return NULL;
 }
 
+static data_item_t* dataq_deq_first_match_ctx(dataq_t* q,
+                                              data_item_match_ctx_fn_t f,
+                                              void* ctx)
+{
+    data_item_t* prev = NULL;
+    data_item_t* curr = q->head;
+
+    while (curr != NULL) {
+        if ((*f)(curr, ctx)) {
+            if (prev == NULL) {
+                q->head = curr->next;
+                if (q->head == NULL) {
+                    q->tail = NULL;
+                }
+            } else {
+                prev->next = curr->next;
+                if (curr == q->tail) {
+                    q->tail = prev;
+                }
+            }
+            curr->next = NULL;
+            return curr;
+        }
+        prev = curr;
+        curr = curr->next;
+    }
+
+    return NULL;
+}
+
+data_item_t* dataq_deq_ctx(dataq_t* q, data_item_match_ctx_fn_t f, void* ctx)
+{
+    data_item_t* b;
+
+    mylock_lock(&(q->lock));
+    b = dataq_deq_first_match_ctx(q, f, ctx);
+    mylock_unlock(&(q->lock));
+
+    return b;
+}
+
 data_item_t* dataq_deq(dataq_t* q, data_item_match_fn_t f)
 {
     data_item_t* b = NULL;

src/lib/meson.build

@@ -226,6 +226,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_order.c',
                    'testsuite/test_vtaskmgr_teardown.c',
                    'testsuite/test_xcealloc.c',
                    'raft/raft_sim.c',

src/lib/testsuite/test.c

@@ -55,6 +55,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_order(void);
 extern void test_vtaskmgr_teardown(void);
 extern void test_xcealloc(void);
 
@@ -181,6 +182,7 @@ int main(int argc, char* argv[])
     TEST(test_ctrlstatus);
     TEST(test_vtask_admission);
     TEST(test_vtask_enq_race);
+    TEST(test_vtask_order);
     TEST(test_vtaskmgr_teardown);
     TEST(test_xcealloc);
 

src/lib/testsuite/test_vtask_order.c

@@ -0,0 +1,177 @@
+/*
+ * bug-1864: the vtaskmgr dispatcher checked each queued task's endpoint
+ * "busy" flag one item at a time while the vcpu finishing that endpoint's
+ * previous message cleared the flag without the queue lock. Within ONE
+ * scan, task N was skipped (endpoint still busy), the endpoint unblocked,
+ * and task N+1 for the SAME endpoint was then dequeued first -- a standby
+ * processed WAL entry seq 258 before 257 and NACKed (log-mismatch storms).
+ *
+ * vtask_deq_eligible() takes the busy check as a parameter so these tests
+ * can make an endpoint unblock part-way through a scan, deterministically.
+ */
+#include <stdlib.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#include "dataq.h"
+#include "netio.h"
+#include "vtaskmgr.h"
+
+static endpoint_t* g_ep1;
+static endpoint_t* g_ep2;
+static int         g_ep1_checks;
+
+static vtask_t* make_io_task(endpoint_t* ep)
+{
+    vtask_t*      t = (vtask_t*)calloc(1, sizeof(*t));
+    vtask_data_t* d = (vtask_data_t*)calloc(1, sizeof(*d));
+
+    TEST_ASSERT_NOT_NULL(t);
+    TEST_ASSERT_NOT_NULL(d);
+    d->vtype       = VTASK_IO;
+    d->u.iodata.ep = ep;
+    t->data        = d;
+    return t;
+}
+
+static vtask_t* make_generic_task(void)
+{
+    vtask_t*      t = (vtask_t*)calloc(1, sizeof(*t));
+    vtask_data_t* d = (vtask_data_t*)calloc(1, sizeof(*d));
+
+    TEST_ASSERT_NOT_NULL(t);
+    TEST_ASSERT_NOT_NULL(d);
+    d->vtype = VTASK_GENERIC;
+    t->data  = d;
+    return t;
+}
+
+static void free_task(vtask_t* t)
+{
+    if (t != NULL) {
+        free(t->data);
+        free(t);
+    }
+}
+
+/* ep1 is busy on the first check of this scan, then its running task
+ * finishes and it is free for every later check -- the live interleaving */
+static int ep1_unblocks_mid_scan(endpoint_t* ep)
+{
+    if (ep == g_ep1) {
+        return (g_ep1_checks++ == 0);
+    }
+    return 0;
+}
+
+static int ep1_busy(endpoint_t* ep) { return (ep == g_ep1); }
+static int none_busy(endpoint_t* ep) { (void)ep; return 0; }
+
+static void order_setup(dataq_t* q)
+{
+    dataq_init(q, NULL);
+    g_ep1        = (endpoint_t*)calloc(1, sizeof(endpoint_t));
+    g_ep2        = (endpoint_t*)calloc(1, sizeof(endpoint_t));
+    g_ep1_checks = 0;
+}
+
+static void order_teardown(dataq_t* q)
+{
+    data_item_t* it;
+
+    while ((it = dataq_deq(q, NULL)) != NULL) {
+        free_task(ITPTR_TO_VTASK(it));
+    }
+    free(g_ep1);
+    free(g_ep2);
+}
+
+/* the bug: a later message must never overtake an earlier one of the same
+ * endpoint, even if the endpoint frees up while the scan is between them */
+static void test_later_task_of_skipped_endpoint_is_not_dequeued(void)
+{
+    dataq_t  q;
+    vtask_t* first;
+    vtask_t* second;
+    vtask_t* got;
+
+    order_setup(&q);
+    first  = make_io_task(g_ep1);
+    second = make_io_task(g_ep1);
+    dataq_enq(&q, &(first->link));
+    dataq_enq(&q, &(second->link));
+
+    got = vtask_deq_eligible(&q, ep1_unblocks_mid_scan);
+    TEST_ASSERT_NULL_MESSAGE(got,
+        "task N was skipped as busy, so task N+1 of the same endpoint must "
+        "not be dequeued ahead of it in the same scan");
+
+    /* next scan: the endpoint is free, and N comes out first */
+    got = vtask_deq_eligible(&q, none_busy);
+    TEST_ASSERT_EQUAL_PTR(first, got);
+    free_task(got);
+    got = vtask_deq_eligible(&q, none_busy);
+    TEST_ASSERT_EQUAL_PTR(second, got);
+    free_task(got);
+
+    order_teardown(&q);
+}
+
+/* a busy endpoint must not hold up other endpoints' tasks */
+static void test_other_endpoint_still_dispatched(void)
+{
+    dataq_t  q;
+    vtask_t* a = NULL;
+    vtask_t* c = NULL;
+    vtask_t* got;
+
+    order_setup(&q);
+    a = make_io_task(g_ep1);
+    c = make_io_task(g_ep2);
+    dataq_enq(&q, &(a->link));
+    dataq_enq(&q, &(c->link));
+
+    got = vtask_deq_eligible(&q, ep1_busy);
+    TEST_ASSERT_EQUAL_PTR(c, got);
+    free_task(got);
+
+    order_teardown(&q);
+}
+
+/* FIFO when nothing is busy; non-IO tasks are always dispatchable */
+static void test_fifo_and_generic_tasks(void)
+{
+    dataq_t  q;
+    vtask_t* a;
+    vtask_t* b;
+    vtask_t* g;
+    vtask_t* got;
+
+    order_setup(&q);
+    a = make_io_task(g_ep1);
+    b = make_io_task(g_ep1);
+    dataq_enq(&q, &(a->link));
+    dataq_enq(&q, &(b->link));
+    got = vtask_deq_eligible(&q, none_busy);
+    TEST_ASSERT_EQUAL_PTR(a, got);
+    free_task(got);
+    order_teardown(&q);
+
+    order_setup(&q);
+    a = make_io_task(g_ep1);
+    g = make_generic_task();
+    dataq_enq(&q, &(a->link));
+    dataq_enq(&q, &(g->link));
+    got = vtask_deq_eligible(&q, ep1_busy);
+    TEST_ASSERT_EQUAL_PTR(g, got);
+    free_task(got);
+    order_teardown(&q);
+}
+
+void test_vtask_order(void)
+{
+    RUN_TEST(test_later_task_of_skipped_endpoint_is_not_dequeued);
+    RUN_TEST(test_other_endpoint_still_dispatched);
+    RUN_TEST(test_fifo_and_generic_tasks);
+}

src/lib/vtaskmgr.c

@@ -281,42 +281,79 @@ static vcpu_t* pick_idle_vcpu(vtaskmgr_t* vtaskmgr)
     return &(vtaskmgr->vcpus[i]);
 }
 
-static int vtask_eligible(data_item_t* data)
-{
-    vtask_t*      task     = ITPTR_TO_VTASK(data);
-    int           eligible = 0;
-    vtask_data_t* vdata;
+/* bug-1864: an endpoint's busy flag is cleared by the vcpu that finished its
+ * previous message WITHOUT the taskq lock (bug-1282: taking a lock there
+ * deadlocked), so it can flip from busy to free part-way through one scan.
+ * Checking each task independently then skips task N (busy) and dequeues
+ * task N+1 of the SAME endpoint (free by now) -- messages of one connection
+ * handled out of order (a standby NACKing WAL entry 258 before 257). The
+ * scan therefore remembers every endpoint it found busy: all later tasks of
+ * that endpoint stay ineligible for the rest of this scan, so per-endpoint
+ * FIFO holds. If more distinct busy endpoints turn up than fit, every later
+ * IO task is treated as ineligible (never reorders; picked up next scan). */
+#define VTASK_SCAN_MAX_SKIPPED 64
+
+typedef struct {
+    int (*is_blocked)(endpoint_t*);
+    endpoint_t* skipped[VTASK_SCAN_MAX_SKIPPED];
+    int         nskipped;
+    int         overflow;
+} vtask_scan_t;
+
+static int vtask_eligible_scan(data_item_t* data, void* arg)
+{
+    vtask_scan_t* scan  = (vtask_scan_t*)arg;
+    vtask_t*      task  = ITPTR_TO_VTASK(data);
+    vtask_data_t* vdata = task->data;
+    endpoint_t*   ep;
+    int           i;
 
-    vdata = task->data;
     if ((vdata == NULL) || (vdata->vtype != VTASK_IO)) {
         return 1;
     }
 
-    if (vdata->u.iodata.ep == NULL) {
+    ep = vdata->u.iodata.ep;
+    if (ep == NULL) {
         WARN("no endpoint for task 0x%p", task);
         return 0;
     }
 
-    eligible = !(endpoint_is_task_blocked(vdata->u.iodata.ep));
-    TRACE("%s vtask %p is %s", __func__, task,
-          eligible ? "eligible" : "not eligible");
+    if (scan->overflow) {
+        return 0;
+    }
 
-    return eligible;
+    for (i = 0; i < scan->nskipped; i++) {
+        if (scan->skipped[i] == ep) {
+            return 0;
+        }
+    }
+
+    if (!scan->is_blocked(ep)) {
+        return 1;
+    }
+
+    if (scan->nskipped < VTASK_SCAN_MAX_SKIPPED) {
+        scan->skipped[scan->nskipped++] = ep;
+    } else {
+        scan->overflow = 1;
+    }
+    return 0;
+}
+
+vtask_t* vtask_deq_eligible(dataq_t* taskq, int (*is_blocked)(endpoint_t*))
+{
+    vtask_scan_t scan = { .is_blocked = is_blocked, .nskipped = 0,
+                          .overflow = 0 };
+    data_item_t* data;
+
+    data = dataq_deq_ctx(taskq, vtask_eligible_scan, &scan);
+
+    return (data != NULL) ? ITPTR_TO_VTASK(data) : NULL;
 }
 
 static vtask_t* pick_eligible_task(vtaskmgr_t* vtaskmgr)
 {
-    data_item_t* data;
-    vtask_t*     task = NULL;
-
-    data = dataq_deq(&(vtaskmgr->taskq), vtask_eligible);
-    if (data == NULL) {
-        return NULL;
-    }
-
-    task = ITPTR_TO_VTASK(data);
-
-    return task;
+    return vtask_deq_eligible(&(vtaskmgr->taskq), endpoint_is_task_blocked);
 }
 
 static int bind_vtask(vtask_t* task, vcpu_t* vcpu)