myxce/src/scheduler/testsuite/test_scheduler_decision_redelivery.c

/*
 * test_scheduler_decision_redelivery.c — queue-level test coverage for the
 * scheduler's decision-publish retry-until-success-or-shutdown wrapper,
 * exercised through the REAL decision-drain sequence rather than a single
 * isolated call (component-failover-adoption, task 3.4, boundary:
 * SchedulerControllerClient, Requirement 7).
 *
 * Task 3.2's test_scheduler_decision_publish_retry.c already proves the
 * single-decision-level contract: one sched_publish_dispatch()/
 * sched_publish_pend_reason() call retries past -E_XCE_NOT_READY until
 * success or shutdown (Req 7 AC1/AC2/AC4), composed with rejoin retry (Req
 * 2 AC5) -- NOT re-tested here. What THIS file proves is the SAME
 * guarantees at the QUEUE level: multiple decisions enqueued on
 * sched->core.decisions, drained one at a time exactly the way driver.c's
 * xce_sched_driver_run_normal_mode_round() does it (cbuf_get() then
 * sched->core.on_decision(...), i.e. xce_sched_driver_on_decision()) --
 * covering:
 *
 *  - Req 7 AC3: decisions queued behind a stuck one are never attempted out
 *    of order, and all eventually publish in original enqueue order.
 *  - Req 7 AC2 / design.md's "Shutdown interaction" paragraph, made
 *    concrete and deterministic: when shutdown fires mid-retry on the
 *    decision actively in flight, every decision still queued behind it
 *    sees core.stop already true on its own very FIRST
 *    sched_publish_retry_loop() check and is never attempted AT ALL (zero
 *    wrapped calls) -- not merely retried fewer times -- which is what
 *    keeps total shutdown latency bounded to roughly one retry cycle
 *    regardless of queue depth.
 *  - Req 7 AC5: scope confirmation that this fix stays within the
 *    scheduler's own publish direction (lightweight defensive net, not the
 *    primary proof -- see below).
 *  - Req 7 AC6: a scheduler restart mid-outage loses the same in-flight
 *    decisions it always would have, since no persistence for queued-but-
 *    unpublished decisions exists or is added by this requirement.
 *
 * This file drains the REAL production xce_sched_driver_on_decision()
 * (driver.c) in a loop that mirrors -- but does not reimplement -- the
 * shape of xce_sched_driver_run_normal_mode_round()'s own decision-drain
 * for(;;) loop (driver.c:762-793): cbuf_get() the next decision, then call
 * the real on_decision callback. Running the FULL normal-mode round
 * (session begin/go/end, plugin hooks, recording) would need scaffolding
 * unrelated to what this file verifies; draining the queue directly through
 * the real callback function is the simpler option this task's own text
 * pre-authorizes, and it still exercises the genuine
 * sched_publish_dispatch()/sched_publish_retry_loop() production code path,
 * not a reimplementation of it.
 *
 * Same real-scheduler_srcs link technique as
 * test_scheduler_decision_publish_retry.c / test_scheduler_backpressure.c
 * (see meson.build): the full production scheduler_srcs are linked in, and
 * only xcecli_jobdispatch() itself is intercepted via `-Wl,--wrap=` linker
 * symbol interposition -- the only libxce client call any scenario below
 * ever reaches. Unlike test_scheduler_decision_publish_retry.c's
 * rejoin-composition test, none of these fixtures call the real
 * sched_init() or touch the private sched_needs_rejoin flag, so
 * sched_publish_check_rejoin()'s `if (!sched_needs_rejoin) return;` guard
 * always short-circuits and xcecli_join() is never reached -- no need to
 * wrap it here.
 *
 * Fixtures call xce_sched_core_init() directly (not the full sched_init())
 * -- the same minimal-fixture technique test_sim_record_output.c's
 * build_fixture_scheduler() and test_core_events.c already established --
 * since every scenario here only needs a working core.decisions cbuf and a
 * core.stop flag, not a live (or faked) controller connection. None of
 * these fixtures call xce_sched_core_fini(): core->pend_updates[0]/[1] are
 * dataq_init()'d only by scheduler.c's sched_init(), never by
 * xce_sched_core_init() itself (core.c's own file header comment), so
 * calling xce_sched_core_fini() against a core built this way runs
 * dataq_fini() on still-NULL pointers and crashes. This is the exact same
 * "fixtures aren't torn down" convention test_core_events.c's own header
 * comment documents for this identical minimal-fixture technique;
 * ASAN_OPTIONS=detect_leaks=0 (meson.build) is this suite's accepted
 * tradeoff for it.
 *
 * Req 7 AC5's scope confirmation below is a lightweight defensive
 * regression net, not the primary proof: the real proof that this fix does
 * not touch worker job-state resync is that this task's diff is entirely
 * contained under src/scheduler/testsuite/ (test-only) and touches nothing
 * under src/worker/ -- a property of the diff itself, which a unit test
 * inside that same diff cannot meaningfully assert about itself. The scan
 * only guards against a *future* change silently blurring that boundary.
 */
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "client.h"
#include "core.h"
#include "decision.h"
#include "driver.h"
#include "mylock.h"
#include "scheduler.h"
#include "xce/errnox.h"

#ifndef XCE_SCHEDULER_C_PATH
#error                                                                         \
    "XCE_SCHEDULER_C_PATH must be defined by meson.build to scheduler.c's absolute path"
#endif
#ifndef XCE_DRIVER_C_PATH
#error                                                                         \
    "XCE_DRIVER_C_PATH must be defined by meson.build to driver.c's absolute path"
#endif

#define JOBID_1 990101ULL
#define JOBID_2 990102ULL
#define JOBID_3 990103ULL

#define MAX_LOG     16
#define MAX_RETURNS 4
#define MAX_SCRIPTS 4

/* ════════════════════════════════════════════════════════════════════════
 * Test double: __wrap_xcecli_jobdispatch -- a per-jobid scripted return
 * sequence, PLUS a global call log (which jobid was seen on each call, in
 * call order) so ordering ACROSS different decisions can be asserted, not
 * just per-jobid call counts.
 * ════════════════════════════════════════════════════════════════════════ */

typedef struct {
    uint64_t jobid;
    int      returns[MAX_RETURNS];
    int      nreturns;
    int      calls;
} jobid_script_t;

static jobid_script_t g_scripts[MAX_SCRIPTS];
static int            g_nscripts;

static uint64_t g_log_jobid[MAX_LOG];
static int      g_log_count;

/* When non-NULL, the wrap sets *g_stop_flag = 1 as a side effect of the
 * call whose 1-based GLOBAL call index equals g_stop_after_call --
 * simulating an operator-triggered shutdown request arriving concurrently
 * mid-retry. */
static int* g_stop_flag;
static int  g_stop_after_call;

static void reset_wrap_state(void)
{
    memset(g_scripts, 0, sizeof(g_scripts));
    g_nscripts = 0;
    memset(g_log_jobid, 0, sizeof(g_log_jobid));
    g_log_count       = 0;
    g_stop_flag       = NULL;
    g_stop_after_call = -1;
}

static void script_jobid(uint64_t jobid, const int* returns, int nreturns)
{
    g_scripts[g_nscripts].jobid = jobid;
    memcpy(g_scripts[g_nscripts].returns, returns,
           sizeof(int) * (size_t)nreturns);
    g_scripts[g_nscripts].nreturns = nreturns;
    g_scripts[g_nscripts].calls    = 0;
    g_nscripts++;
}

static jobid_script_t* find_script(uint64_t jobid)
{
    int i;

    for (i = 0; i < g_nscripts; i++) {
        if (g_scripts[i].jobid == jobid) {
            return &g_scripts[i];
        }
    }
    return NULL;
}

int __wrap_xcecli_jobdispatch(xcecli_ctx_t c, xce_job_dispatch_t* info)
{
    uint64_t        jobid;
    jobid_script_t* s;
    int             rc = 0;

    (void)c;

    jobid = info->allocs[0]->jobid;

    if (g_log_count < MAX_LOG) {
        g_log_jobid[g_log_count] = jobid;
    }
    g_log_count++;

    s = find_script(jobid);
    if (s != NULL) {
        int idx = (s->calls < s->nreturns) ? s->calls : (s->nreturns - 1);

        rc = s->returns[idx];
        s->calls++;
    }

    if ((g_stop_flag != NULL) && (g_log_count == g_stop_after_call)) {
        *g_stop_flag = 1;
    }

    return rc;
}

void setUp(void) { reset_wrap_state(); }
void tearDown(void) { return; }

/* ════════════════════════════════════════════════════════════════════════
 * Helpers
 * ════════════════════════════════════════════════════════════════════════ */

static xce_decision_t* make_dispatch_fixture_decision(uint64_t jobid)
{
    xce_job_alloc_t* alloc = (xce_job_alloc_t*)calloc(1, sizeof(*alloc));

    TEST_ASSERT_NOT_NULL(alloc);
    alloc->jobid = jobid;

    return make_dispatch_decision(alloc);
}

/*
 * drain_n_decisions - pops exactly n decisions off sched->core.decisions,
 * one at a time, calling the REAL xce_sched_driver_on_decision() for each --
 * the identical cbuf_get()-then-on_decision() sequence
 * xce_sched_driver_run_normal_mode_round() runs (driver.c), without
 * reimplementing any of sched_publish_dispatch()'s/
 * sched_publish_retry_loop()'s own retry logic (this task's instruction:
 * exercise the real production function, not a reimplementation of the
 * drain loop's logic).
 */
static void drain_n_decisions(xce_scheduler_t* sched, int n, int* rcs_out)
{
    int i;

    for (i = 0; i < n; i++) {
        xce_decision_t* decision = NULL;

        cbuf_get(&(sched->core.decisions), &decision);
        rcs_out[i] = xce_sched_driver_on_decision(sched, decision);
    }
}

static char* read_whole_file(const char* path)
{
    FILE*  f = fopen(path, "r");
    long   size;
    char*  buf;
    size_t nread;

    if (!f) {
        return NULL;
    }

    fseek(f, 0, SEEK_END);
    size = ftell(f);
    fseek(f, 0, SEEK_SET);
    if (size < 0) {
        fclose(f);
        return NULL;
    }

    buf = (char*)malloc((size_t)size + 1);
    if (!buf) {
        fclose(f);
        return NULL;
    }

    nread      = fread(buf, 1, (size_t)size, f);
    buf[nread] = '\0';
    fclose(f);
    return buf;
}

/* ════════════════════════════════════════════════════════════════════════
 * Req 7 AC3 — decisions queued behind a stuck one are never attempted out
 * of order, and all eventually publish in original enqueue order.
 * ════════════════════════════════════════════════════════════════════════ */

static void
test_ordering_stuck_decision_completes_before_next_is_attempted(void)
{
    xce_scheduler_t  sched;
    xce_decision_t*  d1 = make_dispatch_fixture_decision(JOBID_1);
    xce_decision_t*  d2 = make_dispatch_fixture_decision(JOBID_2);
    xce_decision_t*  d3 = make_dispatch_fixture_decision(JOBID_3);
    int              rcs[3];
    int              rc;
    static const int j1_returns[] = { -E_XCE_NOT_READY, -E_XCE_NOT_READY, 0 };
    static const int j2_returns[] = { 0 };
    static const int j3_returns[] = { 0 };

    memset(&sched, 0, sizeof(sched));
    rc = xce_sched_core_init(&(sched.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    script_jobid(JOBID_1, j1_returns, 3);
    script_jobid(JOBID_2, j2_returns, 1);
    script_jobid(JOBID_3, j3_returns, 1);

    rc = cbuf_put(&(sched.core.decisions), &d1);
    TEST_ASSERT_TRUE(rc >= 0);
    rc = cbuf_put(&(sched.core.decisions), &d2);
    TEST_ASSERT_TRUE(rc >= 0);
    rc = cbuf_put(&(sched.core.decisions), &d3);
    TEST_ASSERT_TRUE(rc >= 0);

    drain_n_decisions(&sched, 3, rcs);

    TEST_ASSERT_TRUE_MESSAGE(
        (rcs[0] >= 0) && (rcs[1] >= 0) && (rcs[2] >= 0),
        "all three decisions must eventually publish successfully");

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        5, g_log_count,
        "exactly 3 attempts for decision 1 (2 NOT_READY + 1 success) plus "
        "one each for decisions 2 and 3 -- 5 total wrapped calls");

    /* The load-bearing assertion: decision 1's THREE calls must all
     * precede decision 2's single call, which must precede decision 3's --
     * proving decision 2 is never attempted while decision 1 is still
     * stuck retrying (no out-of-order/concurrent publish), and that final
     * delivery order matches original enqueue order end to end. */
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(JOBID_1, g_log_jobid[0], "call 1");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(JOBID_1, g_log_jobid[1], "call 2");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(JOBID_1, g_log_jobid[2], "call 3");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
        JOBID_2, g_log_jobid[3],
        "decision 2 must not be attempted until decision 1's retry loop "
        "has fully resolved");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
        JOBID_3, g_log_jobid[4],
        "decision 3 must publish last, after decision 2, preserving FIFO "
        "order end to end");

    /* No xce_sched_core_fini() call: this fixture was built with
     * xce_sched_core_init() alone (not the full sched_init()), so
     * core->pend_updates[0]/[1] were never dataq_init()'d -- that setup is
     * scheduler.c's sched_init()'s job, not core.c's (see this file's
     * header comment). Calling xce_sched_core_fini() here would run
     * dataq_fini() on those still-NULL pointers and SEGV. Same "fixtures
     * aren't torn down" convention test_core_events.c/test_core_session.c
     * already established for this exact minimal-fixture technique;
     * ASAN_OPTIONS=detect_leaks=0 (meson.build) is this suite's accepted
     * tradeoff for it. */
    printf("test_ordering_stuck_decision_completes_before_next_is_attempted: "
           "PASS\n");
}

/* ════════════════════════════════════════════════════════════════════════
 * Req 7 AC2 / design.md "Shutdown interaction" — shutdown mid-retry bounds
 * total latency to roughly ONE retry cycle regardless of how many decisions
 * are queued behind the stuck one: every decision queued behind it sees
 * core.stop already true on its own very FIRST check and is never
 * attempted at all (zero wrapped calls), not merely retried fewer times.
 * ════════════════════════════════════════════════════════════════════════ */

static void
test_shutdown_mid_retry_bounds_latency_zero_calls_for_queued_decisions(void)
{
    xce_scheduler_t sched;
    xce_decision_t* d1 = make_dispatch_fixture_decision(JOBID_1);
    xce_decision_t* d2 = make_dispatch_fixture_decision(JOBID_2);
    xce_decision_t* d3 = make_dispatch_fixture_decision(JOBID_3);
    int             rcs[3];
    int             rc;
    /* Single-entry script, deliberately not "NOT_READY then success": if
     * core.stop were checked only at the BOTTOM of
     * sched_publish_retry_loop()'s loop (the bug this test guards against)
     * rather than the top, decision 1 would be re-attempted a second time
     * before the loop noticed shutdown -- with only one scripted return,
     * that second call would replay the same -E_XCE_NOT_READY forever
     * (find_script() clamps to the last entry), so a broken implementation
     * hangs this test rather than silently passing it. Fails loud, never
     * masked by a timeout -- this file's convention throughout. */
    static const int j1_returns[] = { -E_XCE_NOT_READY };

    memset(&sched, 0, sizeof(sched));
    rc = xce_sched_core_init(&(sched.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    script_jobid(JOBID_1, j1_returns, 1);
    g_stop_flag       = &(sched.core.stop);
    g_stop_after_call = 1; /* fires as a side effect of the very first
                            * wrapped call, simulating shutdown arriving
                            * concurrently mid-retry */

    rc = cbuf_put(&(sched.core.decisions), &d1);
    TEST_ASSERT_TRUE(rc >= 0);
    rc = cbuf_put(&(sched.core.decisions), &d2);
    TEST_ASSERT_TRUE(rc >= 0);
    rc = cbuf_put(&(sched.core.decisions), &d3);
    TEST_ASSERT_TRUE(rc >= 0);

    drain_n_decisions(&sched, 3, rcs);

    TEST_ASSERT_TRUE_MESSAGE(
        rcs[0] < 0,
        "decision 1, actively in-flight when shutdown fires, must be "
        "abandoned (not reported as success) -- it is the one decision "
        "allowed to cost up to one retry cycle");
    TEST_ASSERT_TRUE_MESSAGE(
        (rcs[1] < 0) && (rcs[2] < 0),
        "decisions 2 and 3, still queued behind the stuck one, must also "
        "be abandoned once core.stop is true, never retried to success");

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        1, g_log_count,
        "total wrapped-call count across ALL THREE decisions must be "
        "exactly 1 (decision 1's single attempt) -- decisions 2 and 3 "
        "must each see core.stop already true on their own very first "
        "sched_publish_retry_loop() check and return immediately WITHOUT "
        "ever invoking xcecli_jobdispatch(). This is what keeps total "
        "shutdown latency bounded to roughly one retry cycle rather than "
        "accumulating one cycle per queued decision, regardless of queue "
        "depth");

    /* No xce_sched_core_fini() call -- see the identical note in
     * test_ordering_stuck_decision_completes_before_next_is_attempted()
     * above. */
    printf("test_shutdown_mid_retry_bounds_latency_zero_calls_for_queued_"
           "decisions: PASS\n");
}

/* ════════════════════════════════════════════════════════════════════════
 * Req 7 AC5 — scope confirmation: this fix is confined to the scheduler's
 * own decision-publishing direction; it does not touch worker job-state
 * resync (owned by the separate worker-rejoin-resync backlog item). See
 * this file's header comment for why this is a defensive net, not the
 * primary proof.
 * ════════════════════════════════════════════════════════════════════════ */

static void test_no_worker_resync_symbols_in_scheduler_publish_path(void)
{
    char* sched_src  = read_whole_file(XCE_SCHEDULER_C_PATH);
    char* driver_src = read_whole_file(XCE_DRIVER_C_PATH);

    TEST_ASSERT_NOT_NULL_MESSAGE(sched_src,
                                 "could not read scheduler.c source for scan");
    TEST_ASSERT_NOT_NULL_MESSAGE(driver_src,
                                 "could not read driver.c source for scan");

    TEST_ASSERT_NULL_MESSAGE(
        strstr(sched_src, "resync"),
        "Req 7 AC5: the decision-publish retry fix in scheduler.c must not "
        "reach into worker job-state resync (owned by the separate "
        "worker-rejoin-resync backlog item) -- a 'resync' reference "
        "appearing here would mean this spec's scope silently grew beyond "
        "the scheduler's own decision-publishing direction");
    TEST_ASSERT_NULL_MESSAGE(strstr(driver_src, "resync"),
                             "same check for driver.c's decision-drain loop");

    free(sched_src);
    free(driver_src);
    printf("test_no_worker_resync_symbols_in_scheduler_publish_path: "
           "PASS\n");
}

/* ════════════════════════════════════════════════════════════════════════
 * Req 7 AC6 — restart-loses-in-flight-state confirmation: a freshly
 * xce_sched_core_init()'d core always starts with an empty decisions/jbuf
 * queue, regardless of what a PRIOR instance held queued. There is no
 * persistence mechanism for queued-but-unpublished decisions to survive a
 * scheduler process crash/restart, because no such state is ever written
 * anywhere a new instance could read it back from -- this is the concrete,
 * behaviorally-verifiable form of "a scheduler restart during an outage
 * loses the same in-flight decisions it would have lost before this
 * requirement".
 * ════════════════════════════════════════════════════════════════════════ */

static void
test_fresh_core_init_never_inherits_a_prior_instances_queued_decisions(void)
{
    xce_scheduler_t first_sched;
    xce_scheduler_t second_sched;
    xce_decision_t* d = make_dispatch_fixture_decision(JOBID_1);
    int             rc;

    memset(&first_sched, 0, sizeof(first_sched));
    rc = xce_sched_core_init(&(first_sched.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Simulate a decision still queued, unpublished, at the moment of an
     * abrupt process crash -- first_sched is deliberately never drained nor
     * cleanly torn down before "the crash". */
    rc = cbuf_put(&(first_sched.core.decisions), &d);
    TEST_ASSERT_TRUE(rc >= 0);
    TEST_ASSERT_EQUAL_INT(1, first_sched.core.decisions.len);

    /* "Restart": a brand-new core, from scratch -- nothing about
     * first_sched is passed to it, mirroring how a real process restart
     * gets a fresh xce_scheduler_t with no access to the dead process's
     * memory. */
    memset(&second_sched, 0, sizeof(second_sched));
    rc = xce_sched_core_init(&(second_sched.core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, second_sched.core.decisions.len,
        "Req 7 AC6: a freshly initialized core must start with an EMPTY "
        "decisions queue even though a prior instance still held one "
        "queued and unpublished -- no persistence/snapshot/WAL mechanism "
        "feeds queued-but-unpublished decisions back into a new instance, "
        "so a restart mid-outage loses them exactly as it always did "
        "before this requirement");
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, second_sched.core.jbuf.len,
        "same confirmation for jbuf (the picker->allocator handoff queue)");

    /* No xce_sched_core_fini() calls -- see the identical note in
     * test_ordering_stuck_decision_completes_before_next_is_attempted()
     * above; applies to both first_sched and second_sched here. */
    printf("test_fresh_core_init_never_inherits_a_prior_instances_queued_"
           "decisions: PASS\n");
}

int main(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_ordering_stuck_decision_completes_before_next_is_attempted);
    RUN_TEST(
        test_shutdown_mid_retry_bounds_latency_zero_calls_for_queued_decisions);
    RUN_TEST(test_no_worker_resync_symbols_in_scheduler_publish_path);
    RUN_TEST(
        test_fresh_core_init_never_inherits_a_prior_instances_queued_decisions);
    return UNITY_END();
}