fix(harness): xce_test fake worker answers the controller's job-state query (bug-1922)

The worker-rejoin-resync query sent when a worker with live jobs re-joins
had no handler in xce_test's worker context, so the controller resolved
every such round by its 15 s timeout, withholding dispatch meanwhile.
New fake_resync.{c,h} reassembles the paginated query and builds the
report like xcecd (queried jobs in order, running or unrecognized, then
unqueried running jobs); the handler sends it with a non-blocking request.

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

src/harness/fake_resync.c

@@ -0,0 +1,117 @@
+/*
+ * fake_resync.c -- xce_test fake worker's job-state query answer
+ * (bug-1922). See include/fake_resync.h.
+ */
+#include <errno.h>
+#include <stdbool.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "controller.capnp.h" /* JobStateStatus_* */
+#include "fake_resync.h"
+
+void fr_init(fr_accum_t* a)
+{
+    a->jobids   = NULL;
+    a->ntotal   = 0;
+    a->received = 0;
+}
+
+void fr_reset(fr_accum_t* a)
+{
+    free(a->jobids);
+    fr_init(a);
+}
+
+int fr_add_page(fr_accum_t* a, const xce_job_state_query_t* page)
+{
+    if (page == NULL)
+        return -EINVAL;
+
+    if (page->pos == 0) {
+        fr_reset(a);
+        if (page->n_total < 0)
+            return -EINVAL;
+        if (page->n_total > 0) {
+            a->jobids = calloc((size_t)page->n_total, sizeof(*a->jobids));
+            if (a->jobids == NULL)
+                return -ENOMEM;
+        }
+        a->ntotal = page->n_total;
+    }
+
+    if (page->count > 0) {
+        if (page->jobids == NULL || a->jobids == NULL || page->pos < 0
+            || page->pos + page->count > a->ntotal) {
+            fr_reset(a);
+            return -EINVAL;
+        }
+        memcpy(&a->jobids[page->pos], page->jobids,
+               (size_t)page->count * sizeof(*a->jobids));
+        a->received += page->count;
+    }
+
+    return (a->received >= a->ntotal) ? 1 : 0;
+}
+
+static bool in_list(const uint64_t* ids, int n, uint64_t id)
+{
+    for (int i = 0; i < n; i++) {
+        if (ids[i] == id)
+            return true;
+    }
+    return false;
+}
+
+int fr_build_report(const uint64_t* queried, int nqueried,
+                    const uint64_t* running, int nrunning,
+                    xce_job_state_entry_t*** out, int* nout)
+{
+    int extras = 0;
+
+    *out  = NULL;
+    *nout = 0;
+    for (int i = 0; i < nrunning; i++) {
+        if (!in_list(queried, nqueried, running[i]))
+            extras++;
+    }
+
+    int total = nqueried + extras;
+    if (total == 0)
+        return 0;
+
+    /* one block: the pointer array followed by the entries it points at */
+    xce_job_state_entry_t** ptrs = calloc(1, (size_t)total * (sizeof(*ptrs)
+                                              + sizeof(**ptrs)));
+    if (ptrs == NULL)
+        return -ENOMEM;
+    xce_job_state_entry_t* e = (xce_job_state_entry_t*)(ptrs + total);
+
+    int idx = 0;
+    for (int i = 0; i < nqueried; i++, idx++) {
+        e[idx].jobid   = queried[i];
+        e[idx].status  = in_list(running, nrunning, queried[i])
+                             ? JobStateStatus_running
+                             : JobStateStatus_unrecognized;
+        e[idx].queried = 1;
+        ptrs[idx]      = &e[idx];
+    }
+    for (int i = 0; i < nrunning; i++) {
+        if (in_list(queried, nqueried, running[i]))
+            continue;
+        e[idx].jobid   = running[i];
+        e[idx].status  = JobStateStatus_running;
+        e[idx].queried = 0;
+        ptrs[idx]      = &e[idx];
+        idx++;
+    }
+
+    *out  = ptrs;
+    *nout = total;
+    return 0;
+}
+
+void fr_free_report(xce_job_state_entry_t** entries)
+{
+    free(entries);
+}

src/harness/include/fake_resync.h

@@ -0,0 +1,36 @@
+#ifndef _XCE_HARNESS_FAKE_RESYNC_H
+#define _XCE_HARNESS_FAKE_RESYNC_H
+/*
+ * fake_resync -- xce_test fake worker's answer to the controller's
+ * worker-rejoin-resync job-state query (bug-1922), mirroring xcecd's
+ * xcecd_resync.c: reassemble a paginated OPCode_jobStateQuery, then build the
+ * OPCode_jobStateReport entries -- every queried job ID once, in query
+ * order (running when the fake worker runs it, otherwise unrecognized),
+ * followed by each running job that was not queried. The fake worker keeps
+ * no record of finished jobs (it notifies their finish instead), so a
+ * finished job is reported unrecognized.
+ */
+#include <stdint.h>
+
+#include "typemap.h" /* xce_job_state_query_t, xce_job_state_entry_t */
+
+typedef struct {
+    uint64_t* jobids;
+    int       ntotal;
+    int       received;
+} fr_accum_t;
+
+void fr_init(fr_accum_t* a);
+void fr_reset(fr_accum_t* a);
+/* Add one query page. Returns 1 when the query is complete (a->jobids,
+ * a->ntotal hold it), 0 when more pages are expected, -EINVAL/-ENOMEM on a
+ * bad page (the accumulator is reset). */
+int fr_add_page(fr_accum_t* a, const xce_job_state_query_t* page);
+/* Build the report entries. *out is an array of *nout pointers into one
+ * allocation; release with fr_free_report. Returns 0 or -ENOMEM. */
+int fr_build_report(const uint64_t* queried, int nqueried,
+                    const uint64_t* running, int nrunning,
+                    xce_job_state_entry_t*** out, int* nout);
+void fr_free_report(xce_job_state_entry_t** entries);
+
+#endif

src/harness/meson.build

@@ -21,6 +21,7 @@ xce_test_src = [
   'joblist_display.c',
   'join_worker_args.c',
   'worker_notify.c',
+  'fake_resync.c',
   'rcfile.c',
   'flagspec.c',
   'interruptcall.c',
@@ -94,6 +95,8 @@ harness_unit_test_src = [
   'testsuite/test_test_cli.c',
   'testsuite/test_worker_notify.c',
   'worker_notify.c',
+  'testsuite/test_fake_resync.c',
+  'fake_resync.c',
 ]
 
 harness_unit_test_deps = [

src/harness/test.c

@@ -79,6 +79,8 @@
 #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 "fake_resync.h"      /* job-state query answer (bug-1922) */
+#include "request.h"          /* xce_job_state_report_req (bug-1922) */
 #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) */
 
@@ -6757,6 +6759,55 @@ int run_job(xce_job_exec_t* startmsg)
     return rc;
 }
 
+/* bug-1922: answer the controller's worker-rejoin-resync job-state query
+ * (sent on every worker join; dispatch to this worker is withheld until the
+ * report arrives, or until a 15 s timeout). Pages are accumulated; on the
+ * last one the report -- queried jobs in order, then running jobs that were
+ * not queried -- is sent with a non-blocking request (no reply to wait for,
+ * so it is safe in this handler context). */
+static fr_accum_t g_resync_query;
+
+static int handle_jobstatequery(xce_request_t* request)
+{
+    uint64_t                running[MAX_RUN_JOBS];
+    int                     nrunning = 0;
+    xce_job_state_entry_t** entries  = NULL;
+    int                     nentries = 0;
+    int                     rc;
+
+    rc = fr_add_page(&g_resync_query, request->body.jobstatequerymsg);
+    if (rc <= 0) {
+        return rc; /* more pages expected, or a bad page */
+    }
+
+    for (int i = 0; i < MAX_RUN_JOBS; i++) {
+        if (run_jobs[i].jobid != 0 && run_jobs[i].pid > 0) {
+            running[nrunning++] = run_jobs[i].jobid;
+        }
+    }
+    rc = fr_build_report(g_resync_query.jobids, g_resync_query.ntotal, running,
+                         nrunning, &entries, &nentries);
+    fr_reset(&g_resync_query);
+    if (rc < 0) {
+        return rc;
+    }
+
+    xce_job_state_report_t report = { .n_total = nentries,
+                                      .pos     = 0,
+                                      .count   = nentries,
+                                      .entries = entries };
+    rc = xce_job_state_report_req(xcecli_get_my_endpoint(active_ctx->ctx),
+                                  xcecli_get_myinfo(active_ctx->ctx), &report,
+                                  NULL, NULL, 0);
+    fr_free_report(entries);
+    if (rc < 0) {
+        fprintf(stderr, "job state report send failed, rc %d\n", rc);
+        return rc;
+    }
+    printf("job state query answered: %d job(s)\n", nentries);
+    return 0;
+}
+
 static int handle_startjob(xce_request_t* request)
 {
     return run_job(request->body.jobstartmsg);
@@ -6962,6 +7013,9 @@ static int myreq_handler(xce_request_t* request, void* args)
     case OPCode_startjob:
         rc = handle_startjob(request);
         break;
+    case OPCode_jobStateQuery:
+        rc = handle_jobstatequery(request);
+        break;
     case OPCode_notify:
         rc = handle_notify(request);
         break;

src/harness/testsuite/harness_test.c

@@ -19,6 +19,7 @@ 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);
+extern void test_fake_resync(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
@@ -116,6 +117,7 @@ int main(int argc, char* argv[])
     TEST(test_xcetest_process_entry);
     TEST(test_test_cli);
     TEST(test_worker_notify);
+    TEST(test_fake_resync);
 
     return UNITY_END();
 }

src/harness/testsuite/test_fake_resync.c

@@ -0,0 +1,94 @@
+/*
+ * test_fake_resync.c -- bug-1922: xce_test's fake worker answers the
+ * controller's job-state query (worker-rejoin-resync) instead of letting
+ * every join wait out the 15 s reconciliation timeout.
+ */
+#include <errno.h>
+#include <stdint.h>
+#include <stdlib.h>
+
+#include <unity/unity.h>
+
+#include "controller.capnp.h" /* JobStateStatus_* */
+#include "fake_resync.h"
+
+static void test_single_page_query_completes(void)
+{
+    fr_accum_t a;
+    uint64_t   ids[] = { 11, 12 };
+    xce_job_state_query_t page = { .n_total = 2, .pos = 0, .count = 2,
+                                   .jobids = ids };
+    fr_init(&a);
+    TEST_ASSERT_EQUAL_INT(1, fr_add_page(&a, &page));
+    TEST_ASSERT_EQUAL_INT(2, a.ntotal);
+    TEST_ASSERT_EQUAL_UINT64(11, a.jobids[0]);
+    TEST_ASSERT_EQUAL_UINT64(12, a.jobids[1]);
+    fr_reset(&a);
+}
+
+static void test_multi_page_query_and_empty_query(void)
+{
+    fr_accum_t a;
+    uint64_t   p1[] = { 1, 2 }, p2[] = { 3 };
+    xce_job_state_query_t first  = { .n_total = 3, .pos = 0, .count = 2,
+                                     .jobids = p1 };
+    xce_job_state_query_t second = { .n_total = 3, .pos = 2, .count = 1,
+                                     .jobids = p2 };
+    fr_init(&a);
+    TEST_ASSERT_EQUAL_INT(0, fr_add_page(&a, &first));
+    TEST_ASSERT_EQUAL_INT(1, fr_add_page(&a, &second));
+    TEST_ASSERT_EQUAL_UINT64(3, a.jobids[2]);
+    fr_reset(&a);
+
+    /* an empty query is one complete page */
+    xce_job_state_query_t empty = { .n_total = 0, .pos = 0, .count = 0 };
+    TEST_ASSERT_EQUAL_INT(1, fr_add_page(&a, &empty));
+    TEST_ASSERT_EQUAL_INT(0, a.ntotal);
+    fr_reset(&a);
+}
+
+static void test_bad_page_rejected(void)
+{
+    fr_accum_t a;
+    uint64_t   ids[] = { 1, 2, 3 };
+    xce_job_state_query_t first = { .n_total = 2, .pos = 0, .count = 3,
+                                    .jobids = ids }; /* overflows n_total */
+    fr_init(&a);
+    TEST_ASSERT_EQUAL_INT(-EINVAL, fr_add_page(&a, &first));
+    TEST_ASSERT_NULL(a.jobids);
+}
+
+static void test_report_queried_in_order_then_running_extras(void)
+{
+    uint64_t queried[] = { 5, 6, 7 };
+    uint64_t running[] = { 7, 9, 5 };
+    xce_job_state_entry_t** e = NULL;
+    int n = 0;
+
+    TEST_ASSERT_EQUAL_INT(0, fr_build_report(queried, 3, running, 3, &e, &n));
+    TEST_ASSERT_EQUAL_INT(4, n);
+    TEST_ASSERT_EQUAL_UINT64(5, e[0]->jobid);
+    TEST_ASSERT_EQUAL_INT(JobStateStatus_running, e[0]->status);
+    TEST_ASSERT_EQUAL_INT(1, e[0]->queried);
+    TEST_ASSERT_EQUAL_UINT64(6, e[1]->jobid);
+    TEST_ASSERT_EQUAL_INT(JobStateStatus_unrecognized, e[1]->status);
+    TEST_ASSERT_EQUAL_INT(1, e[1]->queried);
+    TEST_ASSERT_EQUAL_UINT64(7, e[2]->jobid);
+    TEST_ASSERT_EQUAL_INT(JobStateStatus_running, e[2]->status);
+    TEST_ASSERT_EQUAL_UINT64(9, e[3]->jobid); /* running, not queried */
+    TEST_ASSERT_EQUAL_INT(JobStateStatus_running, e[3]->status);
+    TEST_ASSERT_EQUAL_INT(0, e[3]->queried);
+    fr_free_report(e);
+
+    TEST_ASSERT_EQUAL_INT(0, fr_build_report(NULL, 0, NULL, 0, &e, &n));
+    TEST_ASSERT_EQUAL_INT(0, n);
+    fr_free_report(e);
+}
+
+void test_fake_resync(void)
+{
+    RUN_TEST(test_single_page_query_completes);
+    RUN_TEST(test_multi_page_query_and_empty_query);
+    RUN_TEST(test_bad_page_rejected);
+    RUN_TEST(test_report_queried_in_order_then_running_extras);
+}