fix(worker): lock the worker list against vtask-thread readers (bug-1862)

The main loop adds and reaps worker-list entries (add_worker,
cleanup_worker_list -> remove_worker) while controller request handlers on
vtask threads walk the same list with no lock. A rejoin resync overlapping
a job reap read a freed node (ASAN heap-use-after-free in
find_worker_by_jobid), killing the worker mid chaotic HA soak.

- xcecd_worker_list_lock/unlock: add_worker links a fully built entry under
  it; remove_worker unlinks under it and frees after.
- handle_jobstatequery holds it from reading the head through the query
  page (the report send is a non-blocking xce_request); job control and
  remote task start copy what they need under it instead of keeping the
  entry pointer.
- The finished-job retention store (appended by the main loop, read by
  resync) gets its own mutex around its public entry points.

Test: test_build_report_safe_against_concurrent_list_mutation -- UAF 3/3
with no-op lock stubs, clean 5/5 with the fix; worker suite 10/10; chaotic
HA soak 2/2 PASS, 6/6 resync jobs reconciled, no ASAN reports.

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

src/worker/include/xcecd_worker_utils.h

@@ -133,6 +133,14 @@ uint64_t xcecd_finished_job_retention_get_window_ms(void);
 void xcecd_finished_job_retention_note_reconnect(uint64_t now_ms);
 void xcecd_finished_job_retention_set_connection_healthy(bool healthy);
 
+/* bug-1862: the worker list is mutated only by the main loop, but read by
+ * controller request handlers on vtask threads. add_worker/remove_worker
+ * take this lock around their link/unlink; a reader holds it for its whole
+ * lookup and must copy out what it needs before releasing it. Never hold it
+ * across a blocking call. */
+void xcecd_worker_list_lock(void);
+void xcecd_worker_list_unlock(void);
+
 /* worker-rejoin-resync task 4.2 (Requirement 3.6): copies every currently
  * retained, non-expired (as of now_ms) finished-job entry into a freshly
  * heap-allocated array of xcecd_finished_job_t VALUES (not linked -- each

src/worker/src/controller_connector.c

@@ -1527,12 +1527,18 @@ static int handle_jobcontrol(xce_request_t* request)
         task_ctx->signal = d->u.siginfo->signo;
         task_ctx->force  = d->u.siginfo->force;
 
+        /* bug-1862: vtask thread vs the main loop reaping entries -- read
+         * what we need under the list lock, never keep the pointer */
+        xcecd_worker_list_lock();
         xcecd_worker_t* worker = find_worker_by_jobid(
             (xcecd_worker_t*)*cc_ctx->worker_list, d->jobid);
+        int worker_found      = (worker != NULL);
+        int worker_job_status = worker ? worker->job_status : 0;
+        xcecd_worker_list_unlock();
 
-        if (worker) {
+        if (worker_found) {
-            if (worker->job_status == JOB_PENDING
+            if (worker_job_status == JOB_PENDING
-                || worker->job_status == JOB_STARTING) {
+                || worker_job_status == JOB_STARTING) {
                 XCECD_LOG(XCE_LOG_ERROR,
                           "Job is pending or starting, controller should not "
                           "pass the signal to %s.",
@@ -1573,7 +1579,13 @@ static int handle_jobcontrol(xce_request_t* request)
                   "Job control task handled directly for job ID: %s",
                   jobidStr);
 
-        worker->killed_signal = d->u.siginfo->signo; // Mark job as killed
+        xcecd_worker_list_lock();
+        worker = find_worker_by_jobid((xcecd_worker_t*)*cc_ctx->worker_list,
+                                      d->jobid);
+        if (worker) {
+            worker->killed_signal = d->u.siginfo->signo; // Mark job as killed
+        }
+        xcecd_worker_list_unlock();
 #endif
     } else if (d->action == JobControl_u_ckptoption) {
 #if !defined(VIRTUAL_CLUSTER)
@@ -1649,8 +1661,7 @@ static int handle_jobstatequery(xce_request_t* request)
 {
     endpoint_t*     ep   = xcecli_get_my_endpoint(xcecli_ctx);
     xce_peerinfo_t* from = xcecli_get_myinfo(xcecli_ctx);
-    xcecd_worker_t* worker_list
+    xcecd_worker_t* worker_list;
-        = (cc_ctx != NULL) ? (xcecd_worker_t*)*cc_ctx->worker_list : NULL;
     int rc;
 
     if (ep == NULL || from == NULL) {
@@ -1659,9 +1670,17 @@ static int handle_jobstatequery(xce_request_t* request)
         return -EINVAL;
     }
 
+    /* bug-1862: this runs on a vtask thread while the main loop adds and
+     * reaps list entries -- hold the list lock from reading the head
+     * through the whole page (the report build walks the list several
+     * times; the report send is a non-blocking xce_request). */
+    xcecd_worker_list_lock();
+    worker_list
+        = (cc_ctx != NULL) ? (xcecd_worker_t*)*cc_ctx->worker_list : NULL;
     rc = xcecd_resync_handle_query_page(
         worker_list, ep, from, request->body.jobstatequerymsg,
         resync_wallclock_now_ms(), XCECD_RESYNC_REPORT_PKTLIMIT);
+    xcecd_worker_list_unlock();
     if (rc < 0) {
         XCECD_LOG(XCE_LOG_ERROR,
                   "handle_jobstatequery: failed to process query page "
@@ -1972,8 +1991,26 @@ int xcecd_job_remote_start(uint64_t jobid,
         return -1;
     }
     
+    /* bug-1862: copy the fields under the list lock -- the main loop may
+     * reap this entry while xcecli_runtask below is still in flight */
+    char* exechome    = NULL;
+    char* execcwd     = NULL;
+    char* exdata      = NULL;
+    char* execuser    = NULL;
+    char* launch_type = NULL;
+
+    xcecd_worker_list_lock();
     xcecd_worker_t* worker = find_worker_by_jobid((xcecd_worker_t *)*cc_ctx->worker_list, 
                                                   jobid);
+    if (worker) {
+        exechome    = worker->exechome ? strdup(worker->exechome) : NULL;
+        execcwd     = worker->execcwd ? strdup(worker->execcwd) : NULL;
+        exdata      = worker->job_exdata ? strdup(worker->job_exdata) : NULL;
+        execuser    = worker->execuser ? strdup(worker->execuser) : NULL;
+        launch_type = worker->launch_type ? strdup(worker->launch_type) : NULL;
+    }
+    xcecd_worker_list_unlock();
+
     if (!worker) {
         XCECD_LOG(XCE_LOG_ERROR,
                   "Worker not found for job ID %s", jobidStr);
@@ -1981,12 +2018,12 @@ int xcecd_job_remote_start(uint64_t jobid,
     }
 
     xce_taskhost_t host[] = {
-        {.exechome    = worker->exechome,
+        {.exechome    = exechome,
-        .execcwd     = worker->execcwd,
+        .execcwd     = execcwd,
         .cmdline     = strdup(command),
-        .exdata      = worker->job_exdata,
+        .exdata      = exdata,
-        .execuser    = worker->execuser,
+        .execuser    = execuser,
-        .launch_type = worker->launch_type,
+        .launch_type = launch_type,
         .host        = strdup(hostname),
         .ntasks      = 1},
         {0}
@@ -2002,6 +2039,11 @@ int xcecd_job_remote_start(uint64_t jobid,
     taskrun.taskhosts[0] = &host[0];
 
     rc = xcecli_runtask(xcecli_ctx, &taskrun);
+    FREEUP(exechome);
+    FREEUP(execcwd);
+    FREEUP(exdata);
+    FREEUP(execuser);
+    FREEUP(launch_type);
     if (rc < 0) {
         XCECD_LOG(XCE_LOG_ERROR,
                   "Failed to start job %s on remote host %s, rc: %d",

src/worker/src/xcecd_worker_utils.c

@@ -4,6 +4,7 @@
 #include "xcecd_utils.h"
 #include "xcelog.h"
 #include <errno.h>
+#include <pthread.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -24,6 +25,10 @@
 #define XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS 60000ULL
 
 static xcecd_finished_job_t* g_finished_job_retention = NULL;
+/* bug-1862: the store is appended to by the main loop (job exit) and read
+ * by the resync query handler on a vtask thread; every public entry point
+ * below takes this lock around a *_locked_ body. */
+static pthread_mutex_t g_finished_job_retention_lock = PTHREAD_MUTEX_INITIALIZER;
 static uint64_t              g_finished_job_retention_window_ms
     = XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS;
 
@@ -244,6 +249,8 @@ void add_worker(xcecd_worker_t** worker_list, int worker_pid, int worker_status,
 
 add_worker_no_startinfo:
 
+    /* bug-1862: fully built before it becomes visible to readers */
+    xcecd_worker_list_lock();
     if (*worker_list == NULL) {
         *worker_list = worker;
     } else {
@@ -254,6 +261,7 @@ add_worker_no_startinfo:
         current->next = worker;
         worker->prev  = current;
     }
+    xcecd_worker_list_unlock();
 }
 
 void remove_worker(xcecd_worker_t** worker_list, int worker_pid)
@@ -261,6 +269,9 @@ void remove_worker(xcecd_worker_t** worker_list, int worker_pid)
 
     XCECD_LOG(XCE_LOG_DEBUG, "Removing worker %d from worker list", worker_pid);
 
+    /* bug-1862: unlink under the lock so no reader can still be holding
+     * this entry; free it after, once it is unreachable */
+    xcecd_worker_list_lock();
     xcecd_worker_t* worker = *worker_list;
 
     while (worker) {
@@ -273,18 +284,21 @@ void remove_worker(xcecd_worker_t** worker_list, int worker_pid)
             if (worker->next) {
                 worker->next->prev = worker->prev;
             }
-
-            FREEUP(worker->exechome);
-            FREEUP(worker->execcwd);
-            FREEUP(worker->execuser);
-            FREEUP(worker->job_cmd);
-            FREEUP(worker->job_info.pidinfo.pids);
-            FREEUP(worker->job_exdata);
-            FREEUP(worker);
             break;
         }
         worker = worker->next;
     }
+    xcecd_worker_list_unlock();
+
+    if (worker) {
+        FREEUP(worker->exechome);
+        FREEUP(worker->execcwd);
+        FREEUP(worker->execuser);
+        FREEUP(worker->job_cmd);
+        FREEUP(worker->job_info.pidinfo.pids);
+        FREEUP(worker->job_exdata);
+        FREEUP(worker);
+    }
 }
 
 int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
@@ -517,8 +531,8 @@ void cleanup_worker_list(xcecd_worker_t** worker_list, sqlite3* db)
     }
 }
 
-void xcecd_retain_finished_job(uint64_t jobid, int exit_status, int exit_code,
+static void retain_finished_job_locked_(uint64_t jobid, int exit_status,
-                               uint64_t finish_time_ms)
+                                        int exit_code, uint64_t finish_time_ms)
 {
     char                    jobidStr[64];
     xcecd_finished_job_t**  cur   = &g_finished_job_retention;
@@ -568,8 +582,8 @@ void xcecd_retain_finished_job(uint64_t jobid, int exit_status, int exit_code,
               (unsigned long long)finish_time_ms);
 }
 
-bool xcecd_finished_job_lookup(uint64_t jobid, uint64_t now_ms,
+static bool finished_job_lookup_locked_(uint64_t jobid, uint64_t now_ms,
-                               xcecd_finished_job_t* out)
+                                        xcecd_finished_job_t* out)
 {
     xcecd_finished_job_t* e;
 
@@ -613,9 +627,20 @@ void xcecd_finished_job_retention_set_connection_healthy(bool healthy)
     g_finished_job_retention_conn_healthy = healthy;
 }
 
-int xcecd_finished_job_retention_snapshot(uint64_t               now_ms,
+static pthread_mutex_t g_worker_list_lock = PTHREAD_MUTEX_INITIALIZER;
-                                          xcecd_finished_job_t** out,
+
-                                          int*                   out_n)
+void xcecd_worker_list_lock(void)
+{
+    pthread_mutex_lock(&g_worker_list_lock);
+}
+
+void xcecd_worker_list_unlock(void)
+{
+    pthread_mutex_unlock(&g_worker_list_lock);
+}
+
+static int retention_snapshot_locked_(uint64_t now_ms, xcecd_finished_job_t** out,
+                                      int* out_n)
 {
     xcecd_finished_job_t* e;
     xcecd_finished_job_t* copy;
@@ -656,7 +681,7 @@ int xcecd_finished_job_retention_snapshot(uint64_t               now_ms,
     return 0;
 }
 
-void xcecd_finished_job_retention_reset(void)
+static void retention_reset_locked_(void)
 {
     xcecd_finished_job_t* e = g_finished_job_retention;
 
@@ -672,6 +697,44 @@ void xcecd_finished_job_retention_reset(void)
     g_finished_job_retention_conn_healthy      = true;
 }
 
+void xcecd_retain_finished_job(uint64_t jobid, int exit_status, int exit_code,
+                               uint64_t finish_time_ms)
+{
+    pthread_mutex_lock(&g_finished_job_retention_lock);
+    retain_finished_job_locked_(jobid, exit_status, exit_code, finish_time_ms);
+    pthread_mutex_unlock(&g_finished_job_retention_lock);
+}
+
+bool xcecd_finished_job_lookup(uint64_t jobid, uint64_t now_ms,
+                               xcecd_finished_job_t* out)
+{
+    bool found;
+
+    pthread_mutex_lock(&g_finished_job_retention_lock);
+    found = finished_job_lookup_locked_(jobid, now_ms, out);
+    pthread_mutex_unlock(&g_finished_job_retention_lock);
+    return found;
+}
+
+int xcecd_finished_job_retention_snapshot(uint64_t               now_ms,
+                                          xcecd_finished_job_t** out,
+                                          int*                   out_n)
+{
+    int rc;
+
+    pthread_mutex_lock(&g_finished_job_retention_lock);
+    rc = retention_snapshot_locked_(now_ms, out, out_n);
+    pthread_mutex_unlock(&g_finished_job_retention_lock);
+    return rc;
+}
+
+void xcecd_finished_job_retention_reset(void)
+{
+    pthread_mutex_lock(&g_finished_job_retention_lock);
+    retention_reset_locked_();
+    pthread_mutex_unlock(&g_finished_job_retention_lock);
+}
+
 xcecd_worker_t* find_worker_by_pid(void* worker_list, int pid)
 {
     XCECD_LOG(XCE_LOG_DEBUG, "Finding worker %d in worker list", pid);

src/worker/testsuite/test_controller_connector.c

@@ -184,6 +184,10 @@ void xcecd_finished_job_retention_set_connection_healthy(bool healthy)
     (void)healthy;
 }
 
+/* bug-1862 worker-list lock (xcecd_worker_utils.c is not linked here) */
+void xcecd_worker_list_lock(void) {}
+void xcecd_worker_list_unlock(void) {}
+
 bool xcecd_resync_is_pending(void) { return g_test_resync_pending; }
 
 void xcecd_resync_reset(void)

src/worker/testsuite/test_worker_launch_type_dispatch.c

@@ -262,6 +262,10 @@ void xcecd_finished_job_retention_set_connection_healthy(bool healthy)
     (void)healthy;
 }
 
+/* bug-1862 worker-list lock (xcecd_worker_utils.c is not linked here) */
+void xcecd_worker_list_lock(void) {}
+void xcecd_worker_list_unlock(void) {}
+
 bool xcecd_resync_is_pending(void) { return false; }
 
 void xcecd_resync_reset(void) { }

src/worker/testsuite/test_xcecd_resync.c

@@ -26,6 +26,8 @@
  * packet.
  */
 
+#include <pthread.h>
+#include <stdatomic.h>
 #include <stdbool.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -347,6 +349,63 @@ static void test_build_report_answers_every_queried_id_exactly_once(void)
     free_worker_list(list);
 }
 
+/* bug-1862: the main loop adds/reaps worker-list entries (add_worker,
+ * remove_worker via cleanup_worker_list) and retains finished outcomes
+ * while a controller job-state query is answered on a vtask thread that
+ * walks the same list and retention store. Under ASAN this interleaving
+ * was a heap-use-after-free in find_worker_by_jobid. The reader holds the
+ * worker-list lock for the whole report, exactly as handle_jobstatequery
+ * does; add_worker/remove_worker and the retention store lock internally. */
+static xcecd_worker_t* g_stress_list;
+static atomic_int      g_stress_stop;
+
+static void* stress_mutator(void* arg)
+{
+    int pid = 1000;
+
+    (void)arg;
+    while (!atomic_load(&g_stress_stop)) {
+        add_worker(&g_stress_list, pid, WORKER_RUNNING,
+                   (uint64_t)(5000 + pid % 8), 1, NULL);
+        if (pid >= 1004) {
+            remove_worker(&g_stress_list, pid - 4);
+        }
+        xcecd_retain_finished_job((uint64_t)(7000 + pid % 16), WORKER_DONE,
+                                  0, 1000);
+        pid++;
+    }
+    return NULL;
+}
+
+static void test_build_report_safe_against_concurrent_list_mutation(void)
+{
+    pthread_t th;
+    uint64_t  jobids[4] = { 5001, 5002, 5003, 9999 };
+    int       i;
+
+    g_stress_list = NULL;
+    atomic_store(&g_stress_stop, 0);
+    xcecd_finished_job_retention_set_window_ms(1000000);
+    TEST_ASSERT_EQUAL_INT(0,
+        pthread_create(&th, NULL, stress_mutator, NULL));
+
+    for (i = 0; i < 20000; i++) {
+        xce_job_state_entry_t** entries = NULL;
+        int                     n       = 0;
+
+        xcecd_worker_list_lock();
+        TEST_ASSERT_EQUAL_INT(0, xcecd_resync_build_report(
+            g_stress_list, jobids, 4, 2000, &entries, &n));
+        xcecd_worker_list_unlock();
+        xcecd_resync_free_entries(entries, n);
+    }
+
+    atomic_store(&g_stress_stop, 1);
+    pthread_join(th, NULL);
+    free_worker_list(g_stress_list);
+    g_stress_list = NULL;
+}
+
 /* Requirement 3.6: a job the worker is running that was NOT part of the
  * query is reported as an extra (queried=0). */
 static void test_build_report_includes_running_extra(void)
@@ -1043,6 +1102,7 @@ int main(void)
 
     RUN_TEST(test_reset_clears_gate_stuck_by_a_stalled_multipage_query);
     RUN_TEST(test_reset_is_a_safe_noop_when_nothing_pending);
+    RUN_TEST(test_build_report_safe_against_concurrent_list_mutation);
 
     return UNITY_END();
 }