fix(harness): xce_test fake worker defers handler notifications and re-joins after a reconnect (bug-1921)

Worker-context handlers (startjob, runtasks, jobcontrol) run as tasks of
the worker's own connection, so their synchronous notify RPCs waited for
a reply queued behind themselves, timed out (-110) and failed the client
over; the session never joined again, so the controller rejected every
later worker request ("is not a worker") and job usage stayed zero.

Handlers now queue the notification (new worker_notify.{c,h} FIFO) and a
one-shot delay-manager task sends it; housekeeping retries anything left.
A recorded worker join is repeated with the saved attributes once the
connection's reconnect_gen changes.

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

src/harness/include/worker_notify.h

@@ -0,0 +1,68 @@
+#ifndef _XCE_HARNESS_WORKER_NOTIFY_H
+#define _XCE_HARNESS_WORKER_NOTIFY_H
+/*
+ * worker_notify -- deferred worker notifications for xce_test's fake worker
+ * (bug-1921).
+ *
+ * xce_test's worker-context request handlers (startjob, runtasks,
+ * jobcontrol) run as tasks of the worker's own connection.  A notify RPC
+ * waits for its reply (notifyAcked, bug-1741), and that reply is dispatched
+ * as another task of the same connection -- queued behind the handler that
+ * waits for it -- so the call can only time out (-110) and fail the client
+ * over to a new connection.  Handlers therefore queue the notification with
+ * wn_defer() and return; the worker's delay-manager thread sends the queue
+ * with wn_drain().
+ *
+ * wn_need_rejoin() is the companion check for a reconnect: the controller
+ * knows only the joined connection as a worker, so after
+ * xce_ctx_reconnect() (endpoint reconnect_gen bumped) a joined worker joins
+ * again before its next worker request.
+ */
+#include <pthread.h>
+#include <stdbool.h>
+#include <stdint.h>
+
+typedef enum {
+    WN_JOB_START = 1, /* jobid, pid, pgid */
+    WN_JOB_FAIL,      /* jobid, code */
+    WN_TASK_START,    /* jobid, clientidx, taskidx, pid, pgid */
+    WN_TASK_FAIL,     /* jobid, clientidx, taskidx, code */
+    WN_JOB_CHKPT,     /* jobid, info */
+} wn_kind_t;
+
+typedef struct wn_item {
+    wn_kind_t       kind;
+    uint64_t        jobid;
+    int             clientidx;
+    int             taskidx;
+    int             pid;
+    int             pgid;
+    int             code;
+    char*           info; /* owned copy inside the queue */
+    struct wn_item* next;
+} wn_item_t;
+
+typedef struct {
+    pthread_mutex_t lock;
+    wn_item_t*      head;
+    wn_item_t*      tail;
+    int             count;
+} wn_queue_t;
+
+typedef int (*wn_send_fn)(const wn_item_t* item, void* arg);
+
+void wn_init(wn_queue_t* q);
+/* Frees every notification still queued. */
+void wn_fini(wn_queue_t* q);
+/* Queue a copy of *proto (info is duplicated). Never sends. 0 or -ENOMEM. */
+int  wn_defer(wn_queue_t* q, const wn_item_t* proto);
+/* Send queued notifications in order. Stops at the first send that fails,
+ * keeping that one and the rest queued for the next drain. Returns the
+ * number sent. */
+int  wn_drain(wn_queue_t* q, wn_send_fn send, void* arg);
+int  wn_pending(wn_queue_t* q);
+/* True when a worker that joined at connection generation joined_gen must
+ * join again because the connection is now at cur_gen. */
+bool wn_need_rejoin(bool joined, uint64_t joined_gen, uint64_t cur_gen);
+
+#endif

src/harness/meson.build

@@ -20,6 +20,7 @@ xce_test_src = [
   'resval_typeresolve.c',
   'joblist_display.c',
   'join_worker_args.c',
+  'worker_notify.c',
   'rcfile.c',
   'flagspec.c',
   'interruptcall.c',
@@ -91,6 +92,8 @@ harness_unit_test_src = [
   'listfmt_json.c',
   'testsuite/test_xcetest_process_entry.c',
   'testsuite/test_test_cli.c',
+  'testsuite/test_worker_notify.c',
+  'worker_notify.c',
 ]
 
 harness_unit_test_deps = [

src/harness/test.c

@@ -79,6 +79,7 @@
 #include "resval_typeresolve.h" /* resval_resolve_type (resource-value-verification task 2.1) */
 #include "xce/cluster.h" /* xce_parse_peers, xce_cluster_config_t */
 #include "xce/types.h"
+#include "worker_notify.h"    /* deferred worker notifications (bug-1921) */
 #include "xcetest_interrupt.h" /* xcetest_install_sigint, xcetest_interrupted (xce-test-cli-ux task 1.3) */
 
 typedef int (*cmd_func_t)(int, char*[]);
@@ -217,6 +218,9 @@ extern int cmd_snapdiff(int, char*[]);
 extern int cmd_configcompare(int, char*[]);
 extern int cmd_source(int, char*[]);
 static int worker_house_keeping(void*);
+static void worker_record_join(const hostattrs_t* attrs); /* bug-1921 */
+static wn_queue_t g_worker_notify;        /* bug-1921 */
+static int        g_worker_notify_inited; /* bug-1921 */
 
 /* Session verdict / loop state (xce-test-e2e-primitives task 1.1).
  *
@@ -1914,6 +1918,11 @@ int cmd_worker(int argc, char* argv[])
 
     active_ctx = &workerctx;
 
+    if (!g_worker_notify_inited) {
+        wn_init(&g_worker_notify);
+        g_worker_notify_inited = 1;
+    }
+
     delaymgr_init(&delaymgr, 2000 /* precision_ms */, NULL);
     delaymgr_start(&delaymgr);
     delay_task(&delaymgr, 5, -1, (delay_task_entry_fn_t)worker_house_keeping,
@@ -5377,6 +5386,7 @@ int cmd_join_worker(int argc, char* argv[])
         return rc;
     }
 
+    worker_record_join(&attrs);
     printf("host join ok\n");
 
     return 0;
@@ -6397,73 +6407,175 @@ static int sched_on_workers(int nworkers, xce_worker_t** workers)
     return 0;
 }
 
+/* bug-1921: worker-context handlers (startjob, runtasks, jobcontrol) run as
+ * tasks of the worker's own connection, so a notify RPC made there waits for
+ * a reply queued behind itself and times out, failing the client over to a
+ * connection the controller does not know as a worker.  They queue the
+ * notification instead (worker_notify.h); the worker's delay-manager thread
+ * sends it, re-joining first if the connection was re-established. */
+static bool        g_worker_joined;
+static uint64_t    g_worker_joined_gen;
+static hostattrs_t g_worker_join_attrs;
+
+static uint64_t worker_conn_gen(command_ctx_t* ctx)
+{
+    endpoint_t* ep = xcecli_get_my_endpoint(ctx->ctx);
+
+    return (ep == NULL) ? 0
+                        : atomic_load_explicit(&ep->reconnect_gen,
+                                               memory_order_acquire);
+}
+
+static void worker_record_join(const hostattrs_t* attrs)
+{
+    if (active_ctx != &workerctx) {
+        return;
+    }
+    free(g_worker_join_attrs.hosttype);
+    free(g_worker_join_attrs.hostmodel);
+    g_worker_join_attrs           = *attrs;
+    g_worker_join_attrs.hosttype  = attrs->hosttype ? strdup(attrs->hosttype)
+                                                    : NULL;
+    g_worker_join_attrs.hostmodel = attrs->hostmodel ? strdup(attrs->hostmodel)
+                                                     : NULL;
+    g_worker_joined_gen = worker_conn_gen(&workerctx);
+    g_worker_joined     = true;
+}
+
+/* Re-join after xce_ctx_reconnect: the controller only treats the joined
+ * connection as a worker. Runs on the delay-manager thread. */
+static void worker_ensure_joined(command_ctx_t* ctx)
+{
+    uint64_t gen = worker_conn_gen(ctx);
+    int      rc;
+
+    if (!wn_need_rejoin(g_worker_joined, g_worker_joined_gen, gen)) {
+        return;
+    }
+    rc = xcecli_join(ctx->ctx, HostRole_worker, &g_worker_join_attrs);
+    if (rc < 0) {
+        cmdutil_rpc_error("xce_join (re-join after reconnect)", rc);
+        return;
+    }
+    g_worker_joined_gen = gen;
+    printf("host re-join ok (connection re-established)\n");
+}
+
+static int worker_notify_send(const wn_item_t* it, void* arg)
+{
+    command_ctx_t* ctx = (command_ctx_t*)arg;
+
+    switch (it->kind) {
+    case WN_JOB_START: {
+        job_pidinfo_t pidinfo = {
+            .jobid  = it->jobid,
+            .npids  = 1,
+            .pids   = (unsigned int[]) { (unsigned int)it->pid },
+            .npgids = 0,
+            .pgids  = (unsigned int[]) { (unsigned int)it->pgid },
+            .exdata = NULL,
+        };
+        return xcecli_notify_jobstart(ctx->ctx, &pidinfo);
+    }
+    case WN_JOB_FAIL: {
+        job_failinfo_t failinfo = {
+            .jobid  = it->jobid,
+            .stage  = JOB_EXEC_STAGE_MAIN,
+            .code   = it->code,
+            .detail = "no detail information",
+        };
+        return xcecli_notify_jobfail(ctx->ctx, &failinfo);
+    }
+    case WN_TASK_START: {
+        xce_task_pidinfo_t pidinfo = {
+            .jobid     = it->jobid,
+            .host      = get_setting("myname"),
+            .clientidx = it->clientidx,
+            .taskidx   = it->taskidx,
+            .npids     = 1,
+            .pids      = (unsigned int[]) { (unsigned int)it->pid },
+            .npgids    = 1,
+            .pgids     = (unsigned int[]) { (unsigned int)it->pgid },
+        };
+        return xcecli_notify_taskstart(ctx->ctx, &pidinfo);
+    }
+    case WN_TASK_FAIL: {
+        xce_task_failinfo_t failinfo = {
+            .jobid     = it->jobid,
+            .host      = get_setting("myname"),
+            .clientidx = it->clientidx,
+            .taskidx   = it->taskidx,
+            .code      = it->code,
+            .detail    = NULL,
+        };
+        return xcecli_notify_taskfail(ctx->ctx, &failinfo);
+    }
+    case WN_JOB_CHKPT:
+        return xcecli_notify_jobchkpt(ctx->ctx, it->jobid, it->info);
+    }
+    return 0;
+}
+
+static void worker_notify_drain(command_ctx_t* ctx)
+{
+    xce_use(ctx);
+    worker_ensure_joined(ctx);
+    (void)wn_drain(&g_worker_notify, worker_notify_send, ctx);
+}
+
+static int worker_notify_flush(void* arg)
+{
+    worker_notify_drain((command_ctx_t*)arg);
+    return 0;
+}
+
+/* Queue a worker notification and have the delay-manager thread send it on
+ * its next tick; a send that fails is retried by worker_house_keeping. */
+static void worker_notify_defer(const wn_item_t* it)
+{
+    if (wn_defer(&g_worker_notify, it) < 0) {
+        fprintf(stderr, "worker notify: out of memory, notification lost\n");
+        return;
+    }
+    (void)delay_task_ms(&delaymgr, 0, 1,
+                        (delay_task_entry_fn_t)worker_notify_flush, &workerctx);
+}
+
 static void notify_job_fail(uint64_t jobid, int code)
 {
-    /* clang-format off */
+    wn_item_t it = { .kind = WN_JOB_FAIL, .jobid = jobid, .code = code };
-    job_failinfo_t failinfo = {
+
-        .jobid = jobid,
+    worker_notify_defer(&it);
-        .stage = JOB_EXEC_STAGE_MAIN,
-        .code  = code,
-        .detail  = "no detail information"
-    };
-    /*clang-format on*/
-    
-    xcecli_notify_jobfail(active_ctx->ctx, &failinfo);
-    return;
 }
 
 static void notify_task_fail(uint64_t jobid,
                              int clientidx,
                              int taskidx,
                              int code) {
-    xce_task_failinfo_t failinfo = {
+    wn_item_t it = { .kind = WN_TASK_FAIL, .jobid = jobid,
-        .jobid = jobid,
+                     .clientidx = clientidx, .taskidx = taskidx,
-        .host = get_setting("myname"),
+                     .code = code };
-        .clientidx = clientidx,
-        .taskidx = taskidx,
-        .code = code,
-        .detail = NULL
-    };
 
-    xcecli_notify_taskfail(active_ctx->ctx, &failinfo);
+    worker_notify_defer(&it);
 }
 
 static void notify_jobstart(uint64_t jobid,
                             int pid,
                             int pgid) {
-    /* clang-format off */
+    wn_item_t it = { .kind = WN_JOB_START, .jobid = jobid, .pid = pid,
-    job_pidinfo_t pidinfo = {
+                     .pgid = pgid };
-        .jobid = jobid,
-        .npids = 1,
-        .pids  = (unsigned int []){(unsigned int)pid},
-        .npgids = 0,
-        .pgids = (unsigned int []){(unsigned int)pgid},
-        .exdata = NULL
-    };
-    /* clang-format on */
 
-    xcecli_notify_jobstart(active_ctx->ctx, &pidinfo);
+    worker_notify_defer(&it);
-
-    return;
 }
 
 static void notify_task_start(uint64_t jobid, int clientidx, int taskidx,
                               int pid, int pgid)
 {
-    xce_task_pidinfo_t pidinfo
+    wn_item_t it = { .kind = WN_TASK_START, .jobid = jobid,
-        = { .jobid     = jobid,
+                     .clientidx = clientidx, .taskidx = taskidx,
-            .host      = get_setting("myname"),
+                     .pid = pid, .pgid = pgid };
-            .clientidx = clientidx,
-            .taskidx   = taskidx,
-            .npids     = 1,
-            .pids      = (unsigned int[]) { (unsigned int)pid },
-            .npgids    = 1,
-            .pgids     = (unsigned int[]) { (unsigned int)pgid },
-            .exdata    = NULL };
 
-    xcecli_notify_taskstart(active_ctx->ctx, &pidinfo);
+    worker_notify_defer(&it);
-
-    return;
 }
 
 static int run_task_remote(uint64_t jobid, job_startinfo_t* startinfo,
@@ -6714,7 +6826,11 @@ static int job_chkpnt(uint64_t jobid, char* method, int kill)
             "{\nhost_select=\"criu\"; \nrestart_command=\"cd %s; criu restore "
             "-vvv\"}\n",
             buf);
-    xcecli_notify_jobchkpt(active_ctx->ctx, jobid, info);
+    {
+        wn_item_t it = { .kind = WN_JOB_CHKPT, .jobid = jobid, .info = info };
+
+        worker_notify_defer(&it); /* bug-1921: handler context */
+    }
 
     if (kill) {
         job_kill(jobid, SIGTERM);
@@ -7002,6 +7118,10 @@ static int worker_house_keeping(void* arg)
 
     xce_use(ctx);
 
+    /* bug-1921: re-join after a reconnect, and retry notifications a
+     * handler queued that could not be sent yet. */
+    worker_notify_drain(ctx);
+
     check_run_jobs();
     check_run_tasks();
 

src/harness/testsuite/harness_test.c

@@ -18,6 +18,7 @@ extern void test_confirm(void);
 extern void test_listfmt_json(void);
 extern void test_xcetest_process_entry(void);
 extern void test_test_cli(void);
+extern void test_worker_notify(void);
 
 /* test_test_cli.c's leak-safety net (see its own comment): reaps and
  * closes any clitest session left open by a test function that aborted
@@ -114,6 +115,7 @@ int main(int argc, char* argv[])
     TEST(test_listfmt_json);
     TEST(test_xcetest_process_entry);
     TEST(test_test_cli);
+    TEST(test_worker_notify);
 
     return UNITY_END();
 }

src/harness/testsuite/test_worker_notify.c

@@ -0,0 +1,146 @@
+/*
+ * test_worker_notify.c -- bug-1921: xce_test's fake worker must not make a
+ * synchronous notify RPC from a handler running as a task of its own
+ * connection (the reply is queued behind the handler, so the call times out
+ * and the client fails over).  Handlers queue worker notifications instead;
+ * the worker's delay-manager thread drains the queue.  After a reconnect the
+ * worker re-joins before sending worker requests.
+ */
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#include "worker_notify.h"
+
+#define MAX_SENT 8
+static wn_item_t g_sent[MAX_SENT];
+static char      g_sent_info[MAX_SENT][32];
+static int       g_nsent;
+static int       g_fail_at; /* 1-based send call that fails; 0 = never */
+static int       g_calls;
+
+static int record_send(const wn_item_t* it, void* arg)
+{
+    (void)arg;
+    g_calls++;
+    if (g_fail_at != 0 && g_calls == g_fail_at)
+        return -110;
+    g_sent[g_nsent] = *it;
+    g_sent_info[g_nsent][0] = '\0';
+    if (it->info != NULL)
+        snprintf(g_sent_info[g_nsent], sizeof(g_sent_info[0]), "%s", it->info);
+    g_nsent++;
+    return 0;
+}
+
+static void reset_sends(int fail_at)
+{
+    memset(g_sent, 0, sizeof(g_sent));
+    g_nsent   = 0;
+    g_calls   = 0;
+    g_fail_at = fail_at;
+}
+
+static void test_defer_does_not_send(void)
+{
+    wn_queue_t q;
+    wn_init(&q);
+    reset_sends(0);
+
+    wn_item_t start = { .kind = WN_JOB_START, .jobid = 7, .pid = 100, .pgid = 100 };
+    TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &start));
+    TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_calls,
+        "bug-1921: queuing a notification must not send it");
+    TEST_ASSERT_EQUAL_INT(1, wn_pending(&q));
+
+    wn_fini(&q);
+}
+
+static void test_drain_sends_fifo_with_copied_payload(void)
+{
+    wn_queue_t q;
+    wn_init(&q);
+    reset_sends(0);
+
+    char info[16];
+    strcpy(info, "ckpt-a");
+    wn_item_t a = { .kind = WN_JOB_START, .jobid = 1, .pid = 11, .pgid = 11 };
+    wn_item_t b = { .kind = WN_TASK_FAIL, .jobid = 2, .clientidx = 3,
+                    .taskidx = 4, .code = -5 };
+    wn_item_t c = { .kind = WN_JOB_CHKPT, .jobid = 3, .info = info };
+    TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &a));
+    TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &b));
+    TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &c));
+    strcpy(info, "CLOBBERED"); /* the queue keeps its own copy */
+
+    TEST_ASSERT_EQUAL_INT(3, wn_drain(&q, record_send, NULL));
+    TEST_ASSERT_EQUAL_INT(3, g_nsent);
+    TEST_ASSERT_EQUAL_INT(WN_JOB_START, g_sent[0].kind);
+    TEST_ASSERT_EQUAL_UINT64(1, g_sent[0].jobid);
+    TEST_ASSERT_EQUAL_INT(11, g_sent[0].pid);
+    TEST_ASSERT_EQUAL_INT(WN_TASK_FAIL, g_sent[1].kind);
+    TEST_ASSERT_EQUAL_INT(3, g_sent[1].clientidx);
+    TEST_ASSERT_EQUAL_INT(4, g_sent[1].taskidx);
+    TEST_ASSERT_EQUAL_INT(-5, g_sent[1].code);
+    TEST_ASSERT_EQUAL_INT(WN_JOB_CHKPT, g_sent[2].kind);
+    TEST_ASSERT_EQUAL_STRING("ckpt-a", g_sent_info[2]);
+    TEST_ASSERT_EQUAL_INT(0, wn_pending(&q));
+
+    wn_fini(&q);
+}
+
+/* A failed send (e.g. not connected) keeps that notification and every later
+ * one queued, in order, for the next drain: a job start is never lost. */
+static void test_drain_stops_at_failure_and_resumes(void)
+{
+    wn_queue_t q;
+    wn_init(&q);
+    reset_sends(2);
+
+    for (int i = 1; i <= 3; i++) {
+        wn_item_t it = { .kind = WN_JOB_START, .jobid = (uint64_t)i };
+        TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &it));
+    }
+    TEST_ASSERT_EQUAL_INT(1, wn_drain(&q, record_send, NULL));
+    TEST_ASSERT_EQUAL_INT(2, wn_pending(&q));
+
+    reset_sends(0);
+    TEST_ASSERT_EQUAL_INT(2, wn_drain(&q, record_send, NULL));
+    TEST_ASSERT_EQUAL_UINT64(2, g_sent[0].jobid);
+    TEST_ASSERT_EQUAL_UINT64(3, g_sent[1].jobid);
+    TEST_ASSERT_EQUAL_INT(0, wn_pending(&q));
+
+    wn_fini(&q);
+}
+
+static void test_fini_frees_undrained(void)
+{
+    wn_queue_t q;
+    wn_init(&q);
+    wn_item_t c = { .kind = WN_JOB_CHKPT, .jobid = 9, .info = "x" };
+    TEST_ASSERT_EQUAL_INT(0, wn_defer(&q, &c));
+    wn_fini(&q); /* LeakSanitizer checks the copy is released */
+    TEST_ASSERT_EQUAL_INT(0, wn_pending(&q));
+}
+
+/* A worker that joined must join again once its connection was re-established
+ * (the controller only knows the joined connection as a worker). */
+static void test_need_rejoin(void)
+{
+    TEST_ASSERT_FALSE(wn_need_rejoin(false, 0, 0));
+    TEST_ASSERT_FALSE(wn_need_rejoin(false, 0, 3));
+    TEST_ASSERT_FALSE(wn_need_rejoin(true, 2, 2));
+    TEST_ASSERT_TRUE(wn_need_rejoin(true, 2, 3));
+}
+
+void test_worker_notify(void)
+{
+    RUN_TEST(test_defer_does_not_send);
+    RUN_TEST(test_drain_sends_fifo_with_copied_payload);
+    RUN_TEST(test_drain_stops_at_failure_and_resumes);
+    RUN_TEST(test_fini_frees_undrained);
+    RUN_TEST(test_need_rejoin);
+}

src/harness/worker_notify.c

@@ -0,0 +1,108 @@
+/*
+ * worker_notify.c -- deferred worker notifications for xce_test's fake
+ * worker (bug-1921). See include/worker_notify.h.
+ */
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "worker_notify.h"
+
+void wn_init(wn_queue_t* q)
+{
+    pthread_mutex_init(&q->lock, NULL);
+    q->head  = NULL;
+    q->tail  = NULL;
+    q->count = 0;
+}
+
+static void wn_free_item(wn_item_t* it)
+{
+    free(it->info);
+    free(it);
+}
+
+void wn_fini(wn_queue_t* q)
+{
+    pthread_mutex_lock(&q->lock);
+    wn_item_t* it = q->head;
+    while (it != NULL) {
+        wn_item_t* next = it->next;
+        wn_free_item(it);
+        it = next;
+    }
+    q->head  = NULL;
+    q->tail  = NULL;
+    q->count = 0;
+    pthread_mutex_unlock(&q->lock);
+    pthread_mutex_destroy(&q->lock);
+}
+
+int wn_defer(wn_queue_t* q, const wn_item_t* proto)
+{
+    wn_item_t* it = malloc(sizeof(*it));
+    if (it == NULL)
+        return -ENOMEM;
+    *it      = *proto;
+    it->next = NULL;
+    it->info = NULL;
+    if (proto->info != NULL) {
+        it->info = strdup(proto->info);
+        if (it->info == NULL) {
+            free(it);
+            return -ENOMEM;
+        }
+    }
+
+    pthread_mutex_lock(&q->lock);
+    if (q->tail != NULL)
+        q->tail->next = it;
+    else
+        q->head = it;
+    q->tail = it;
+    q->count++;
+    pthread_mutex_unlock(&q->lock);
+    return 0;
+}
+
+int wn_drain(wn_queue_t* q, wn_send_fn send, void* arg)
+{
+    int sent = 0;
+
+    for (;;) {
+        /* Peek under the lock, send without it (a send blocks on the RPC),
+         * then pop: only the drain thread removes items, so the head is
+         * still this item; wn_defer only ever appends. */
+        pthread_mutex_lock(&q->lock);
+        wn_item_t* it = q->head;
+        pthread_mutex_unlock(&q->lock);
+        if (it == NULL)
+            break;
+
+        if (send(it, arg) < 0)
+            break; /* keep it (and the rest) for the next drain */
+
+        pthread_mutex_lock(&q->lock);
+        q->head = it->next;
+        if (q->head == NULL)
+            q->tail = NULL;
+        q->count--;
+        pthread_mutex_unlock(&q->lock);
+        wn_free_item(it);
+        sent++;
+    }
+    return sent;
+}
+
+int wn_pending(wn_queue_t* q)
+{
+    pthread_mutex_lock(&q->lock);
+    int n = q->count;
+    pthread_mutex_unlock(&q->lock);
+    return n;
+}
+
+bool wn_need_rejoin(bool joined, uint64_t joined_gen, uint64_t cur_gen)
+{
+    return joined && cur_gen != joined_gen;
+}