myxce/src/worker/testsuite/test_dbus_workers_update.c

/*
 * test_dbus_workers_update.c — regression test for bug-1589: xcecd_virtual's
 * (and real xcecd's) dbus_workers_update() (xcecd_dbus_manager.c) starved
 * every worker in xcecd_ctx.worker_list after the first one it visited.
 *
 * User-reported symptom: a 5-element job array (base_jobid=0-7-1787111880-0,
 * array_size=5) was fully dispatched by the controller (dispatch_a_job_
 * fired for [0]-[4]) and elected by the scheduler (limit_job_election fired
 * for [0]-[4]), but only element [0] EVER produced an on_job_start log line
 * -- elements [1]-[4] were dispatched but never reported as started, in
 * either log file, ever.
 *
 * Root cause, confirmed by tracing xcecd_virtual's real build (worker/
 * meson.build: xcecd_virtual == xcecd_srcs + main.c, -DVIRTUAL_CLUSTER
 * -DDBUS_MANAGER -- the SAME DBUS_MANAGER flag the real `xcecd` binary also
 * gets, so main.c's `#ifndef DBUS_MANAGER` call to process_worker_manager.c's
 * worker_process_update() is dead code for BOTH binaries; both actually call
 * dbus_workers_update() every ~10ms from main.c's run loop):
 *
 * dbus_workers_update() iterates xcecd_ctx.worker_list (one xcecd_worker_t
 * node per concurrently-dispatched job) with a single `while (worker) { ...
 * worker = worker->next; }` loop, but SEVERAL branches inside the loop body
 * used a bare `continue` (re-testing the SAME `worker` pointer forever
 * instead of advancing) or, worse, `return true` the moment ANY worker was
 * WORKER_DONE/WORKER_EXITED (aborting the scan of the REST of the list
 * entirely, not just skipping that one finished worker). Net effect: once
 * the function started making progress on the head of the list, every OTHER
 * worker queued behind it was starved of update_worker_status() (which is
 * what fires xcecd_reply_startjob() -- the on_job_start report) for as long
 * as the head worker stayed un-reaped -- exactly the observed "[0] started,
 * [1]-[4] never did" symptom for a 5-element array dispatched to one worker
 * within the SAME dbus_workers_update() call.
 *
 * Fix (xcecd_dbus_manager.c): capture `next = worker->next` once at the top
 * of every loop iteration and always advance to it -- every `continue`
 * becomes `worker = next; continue;`, and the `return true` early-abort
 * becomes `worker = next; continue;` (skip this already-finished worker,
 * keep scanning the rest of the list).
 *
 * White-box test: #include "../src/xcecd_dbus_manager.c" directly (same
 * technique as test_xcecd_startup_order.c/test_controller_connector.c) so
 * the REAL dbus_workers_update() traversal logic runs under test.
 * update_worker_status()/is_systemd_job_active()/get_systemd_job_exit_code()
 * /stop_systemd_job()/make_job_sd_unit_name() are stubbed to isolate the
 * loop's own control-flow bug from unrelated dependencies (real systemd/
 * dbus, real sqlite3, real IPC-to-controller reporting) -- the
 * update_worker_status() stub mirrors just enough of the real function's
 * state machine (JOB_STARTING+WORKER_RUNNING -> JOB_RUNNING on a
 * WORKER_RUNNING call; -> WORKER_DONE/JOB_DONE on a WORKER_DONE call) to
 * drive dbus_workers_update()'s own control flow identically to production,
 * while recording every call so the test can assert ALL workers were
 * visited, not just the first.
 *
 * Build flags: -DVIRTUAL_CLUSTER (matches the real xcecd_virtual executable,
 * worker/meson.build -- DBUS_MANAGER is irrelevant here since xcecd_dbus_
 * manager.c itself has no #ifdef DBUS_MANAGER branches, only VIRTUAL_CLUSTER
 * ones, which is exactly the code path this bug lives in).
 */
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include <unity/unity.h>

/* Pull the real dbus_workers_update() traversal logic under test into this
 * TU. */
#include "../src/xcecd_dbus_manager.c"

/* ════════════════════════════════════════════════════════════════════════
 * Link-satisfying stubs
 * ════════════════════════════════════════════════════════════════════════ */

#define MAX_CALLS 64
static int g_uws_pid[MAX_CALLS];
static int g_uws_worker_status[MAX_CALLS];
static int g_uws_ncalls;

/* Mirrors just the pieces of the REAL update_worker_status()
 * (xcecd_worker_utils.c) that affect dbus_workers_update()'s own control
 * flow: looks the worker up by pid in the full list (same as production),
 * transitions JOB_STARTING->JOB_RUNNING on a WORKER_RUNNING call and
 * ->WORKER_DONE/JOB_DONE on a WORKER_DONE call. Does NOT reimplement
 * xcecd_reply_startjob()/DB/IPC side effects -- those are outside this
 * loop-traversal bug's boundary and are covered by other tests. */
int update_worker_status(xcecd_worker_t* worker_list, int worker_pid,
                         int worker_status, xcecd_job_t* job, sqlite3* db)
{
    (void)db;

    if (g_uws_ncalls < MAX_CALLS) {
        g_uws_pid[g_uws_ncalls]           = worker_pid;
        g_uws_worker_status[g_uws_ncalls] = worker_status;
        g_uws_ncalls++;
    }

    xcecd_worker_t* w = worker_list;
    while (w && w->worker_pid != worker_pid) {
        w = w->next;
    }
    if (!w) {
        return XCECD_NOT_FOUND;
    }

    job->job_id = w->job_info.jobid;

    switch (worker_status) {
    case WORKER_RUNNING:
        if (w->job_status == JOB_STARTING && job->job_status == JOB_RUNNING) {
            w->job_status = JOB_RUNNING;
        }
        break;
    case WORKER_DONE:
        w->worker_status = WORKER_DONE;
        w->job_status     = JOB_DONE;
        break;
    case WORKER_EXITED:
        w->worker_status = WORKER_EXITED;
        w->job_status     = JOB_EXITED;
        break;
    default:
        break;
    }

    return XCECD_OK;
}

/* No real systemd/dbus unit ever exists for a VIRTUAL_CLUSTER fake-pid job
 * (xcecd.c's xcecd_process_command() never calls dbus_start_worker() when
 * VIRTUAL_CLUSTER is defined) -- matches both production virtual-cluster
 * reality and this sandbox (no dbus session). */
bool is_systemd_job_active(const char* unit_name)
{
    (void)unit_name;
    return false;
}

/* get_systemd_job_exit_code() (xcecd_cgroups.c) returns -1 on any sd_bus
 * failure, which is exactly what happens for a nonexistent unit -- verified
 * against its real implementation. */
int get_systemd_job_exit_code(const char* unit_name)
{
    (void)unit_name;
    return -1;
}

int stop_systemd_job(const char* unit_name)
{
    (void)unit_name;
    return 0;
}

int make_job_sd_unit_name(char* buf, size_t bufsize, const char* prefix,
                          uint64_t jobid)
{
    return snprintf(buf, bufsize, "%s-%llu.service", prefix,
                    (unsigned long long)jobid);
}

/* Link-satisfying stubs: xcecd_dbus_manager.c (#include'd whole) also
 * defines dbus_start_worker()/dbus_get_worker_pid() (never exercised by
 * this test -- only dbus_workers_update()'s traversal is under test), which
 * reference these two real-systemd helpers (normally defined in
 * xcecd_cgroups.c, not linked into this white-box test binary). */
char* start_systemd_job(const char* command, const res_limits_t* limits,
                        const char* unit_name_prefix, uint64_t job_id,
                        char* extra_args)
{
    (void)command;
    (void)limits;
    (void)unit_name_prefix;
    (void)job_id;
    (void)extra_args;
    return NULL;
}

pid_t get_systemd_job_pid(const char* unit_name)
{
    (void)unit_name;
    return -1;
}

/* Real check_worker_sleep_time_elapsed() (xcecd_worker_utils.c) is not
 * linked into this white-box test binary (its own file isn't #include'd,
 * to keep the stub surface for update_worker_status() etc. above simple
 * and hermetic) -- this reimplementation is byte-identical to the real
 * function's logic and is exactly what dbus_workers_update() needs to make
 * its own VIRTUAL_CLUSTER JOB_RUNNING->WORKER_DONE decision, which this
 * test's fixtures (sleep_time == 0) deliberately drive down the "already
 * elapsed" path for fast, deterministic execution. */
bool check_worker_sleep_time_elapsed(xcecd_worker_t* worker)
{
    if (!worker) {
        return true;
    }
    if (worker->sleep_time <= 0) {
        return true;
    }
    time_t current_time = time(NULL);
    if (current_time == (time_t)-1) {
        return true;
    }
    return difftime(current_time, worker->start_time) >= worker->sleep_time;
}

/* ════════════════════════════════════════════════════════════════════════
 * Fixtures
 * ════════════════════════════════════════════════════════════════════════ */

static xcecd_worker_t* make_starting_worker(int pid, uint64_t jobid)
{
    xcecd_worker_t* w = calloc(1, sizeof(*w));
    w->worker_pid      = pid;
    w->worker_status    = WORKER_RUNNING;
    w->job_status        = JOB_STARTING;
    w->job_info.jobid   = jobid;
    /* sleep_time == 0 -> check_worker_sleep_time_elapsed() returns true
     * immediately (matches a job whose command is not a literal "sleep N",
     * i.e. most real test payloads) -- keeps this test deterministic and
     * fast (no real elapsed-time dependency) while still exercising the
     * exact branches production hits. */
    w->sleep_time  = 0;
    w->start_time  = time(NULL);
    return w;
}

static void link_worker(xcecd_worker_t** list, xcecd_worker_t* w)
{
    if (!*list) {
        *list = w;
        return;
    }
    xcecd_worker_t* cur = *list;
    while (cur->next) {
        cur = cur->next;
    }
    cur->next = w;
    w->prev   = cur;
}

static void free_worker_list(xcecd_worker_t* list)
{
    while (list) {
        xcecd_worker_t* next = list->next;
        free(list);
        list = next;
    }
}

static bool started_call_seen(int pid)
{
    for (int i = 0; i < g_uws_ncalls; i++) {
        if (g_uws_pid[i] == pid && g_uws_worker_status[i] == WORKER_RUNNING) {
            return true;
        }
    }
    return false;
}

void setUp(void)
{
    g_uws_ncalls = 0;
}

void tearDown(void) { return; }

/* ════════════════════════════════════════════════════════════════════════
 * bug-1589 — every JOB_STARTING worker in the list must get its start
 * transition (the event that fires xcecd_reply_startjob()/on_job_start in
 * production) within one dbus_workers_update() call, not just the first.
 * ════════════════════════════════════════════════════════════════════════ */

static void test_all_array_elements_get_started_in_one_pass(void)
{
    /* Mirrors the user's live report: a 5-element job array dispatched to
     * one worker, all within one control loop tick. */
    xcecd_worker_t* list = NULL;
    xcecd_worker_t* workers[5];
    for (int i = 0; i < 5; i++) {
        workers[i] = make_starting_worker(1000 + i, 100 + (uint64_t)i);
        link_worker(&list, workers[i]);
    }

    xcecd_job_t job = { 0 };
    int         ret = dbus_workers_update(&list, &job, NULL);

    TEST_ASSERT_EQUAL_INT(XCECD_OK, ret);

    for (int i = 0; i < 5; i++) {
        char msg[160];
        snprintf(msg, sizeof(msg),
                "array element [%d] (pid %d) never received its "
                "WORKER_RUNNING/on_job_start transition -- starved by "
                "an earlier worker in the list (bug-1589)",
                i, 1000 + i);
        TEST_ASSERT_TRUE_MESSAGE(started_call_seen(1000 + i), msg);
    }

    free_worker_list(list);
}

/* Regression guard for the specific `return true` abort: a worker that is
 * ALREADY WORKER_DONE at the head of the list (e.g. a stale entry not yet
 * reaped by cleanup_worker_list()) must not stop the scan of the rest of
 * the list. */
static void test_done_head_worker_does_not_block_rest_of_list(void)
{
    xcecd_worker_t* list = NULL;

    xcecd_worker_t* done_head = calloc(1, sizeof(*done_head));
    done_head->worker_pid    = 2000;
    done_head->worker_status  = WORKER_DONE;
    done_head->job_status      = JOB_DONE;
    done_head->job_info.jobid = 200;
    link_worker(&list, done_head);

    xcecd_worker_t* starting = make_starting_worker(2001, 201);
    link_worker(&list, starting);

    xcecd_job_t job = { 0 };
    int         ret = dbus_workers_update(&list, &job, NULL);

    TEST_ASSERT_EQUAL_INT(XCECD_OK, ret);
    TEST_ASSERT_TRUE_MESSAGE(
        started_call_seen(2001),
        "a DONE worker at the head of the list must not abort the scan "
        "before the still-JOB_STARTING worker behind it is reached");

    free_worker_list(list);
}

/* ════════════════════════════════════════════════════════════════════════
 * main
 * ════════════════════════════════════════════════════════════════════════ */

int main(void)
{
    xce_init_xcecd_log("/tmp/test_dbus_workers_update.log", 1024 * 1024, 2);

    UNITY_BEGIN();

    RUN_TEST(test_all_array_elements_get_started_in_one_pass);
    RUN_TEST(test_done_head_worker_does_not_block_rest_of_list);

    return UNITY_END();
}