myxce/src/scheduler/testsuite/test_scheduler_backpressure.c

/*
 * test_scheduler_backpressure.c — worst-case decision/allocation
 * backpressure test (component-failover-adoption, task 3.3, boundary:
 * SchedulerControllerClient, Req 3 AC2).
 *
 * Scope note: this file proves Req 3 AC2 only -- the bounded, occupancy-
 * dependent buffering upstream of the publish call (picker/allocator
 * continue briefly, then genuinely block, then resume with nothing lost).
 * It does NOT exercise Req 7's retry-until-success-or-shutdown wrapper
 * around a `-E_XCE_NOT_READY` return from `xcecli_jobdispatch()`/
 * `xcecli_set_pend_reason()` (that wrapper, and its own test coverage, is
 * task 3.2's `sched_publish_dispatch()`/`sched_publish_pend_reason()` work,
 * see test_scheduler_decision_publish_retry.c). The wrapped
 * `xcecli_set_pend_reason()` double below BLOCKS on a condvar rather than
 * ever returning `-E_XCE_NOT_READY`, so it never enters that retry loop at
 * all -- it stands in for the call still being in-flight/stuck mid-attempt,
 * one layer below where Req 7 applies.
 *
 * Links the REAL production scheduler_srcs (same technique
 * test_scheduler_connect.c / test_scheduler_decision_publish_retry.c
 * already established for this boundary -- see meson.build) and calls the
 * REAL, public sched_init(), which -- unlike task 3.2's rejoin test, which
 * deliberately made its own setup join fail -- is made to SUCCEED here
 * (xcecli_start()/xcecli_join() wrapped to return success with no I/O),
 * so the real sched_job_picker() and sched_allocator() threads (scheduler.c)
 * actually start. xcecli_set_pend_reason() is wrapped to BLOCK on a
 * condition variable under test control, standing in for an ongoing
 * controller outage (an XCECLI_FAILOVER_CALL retry budget stuck retrying)
 * without any real network I/O or busy-spinning.
 *
 * ── Why the picker thread is never driven, and jobs are hand-built ──────
 *
 * sched_job_picker()/sched_allocator() are `static` (scheduler.c) and can
 * only be started via the real sched_init() -- both threads start together,
 * there is no way to run just one. sched_job_picker() is kept PERMANENTLY
 * DORMANT here by never calling sched_go(): it blocks forever on
 * cbuf_get(&core.wbuf) (scheduler.c), so it never touches core.jbuf itself
 * and never contends with this file's own direct cbuf_put()/cbuf_tryput()
 * calls on core.jbuf, which stand in for the picker's role of feeding one
 * jobid at a time -- exactly the "without also running the full
 * sched_job_picker thread" narrowing this task's own text pre-authorizes.
 *
 * sched_allocator() IS the real, unmodified production function under
 * test here (not a stand-in) -- proving its genuine cbuf_put()-blocks
 * behavior against a full `decisions` cbuf is this test's actual point.
 * Driving it through sched_make_allocation() with a fully realistic
 * job/queue/worker/resource model (parse_resreq(), the seqjob_allocator
 * plugin's "slots" resource lookup, sched_collect_workers() against a real
 * resmgr) would need substantial scaffolding unrelated to what this test
 * verifies. Instead, each test job's xce_sched_obj_t is built directly
 * with tag=XCE_RESREQ_COMPOUND and a COMPOUND resreq with an EMPTY
 * `primaries` list: sched_make_compound_allocation()'s for-loop
 * (scheduler.c) then has zero iterations, so it returns rc=0 ("pend")
 * without ever touching a worker, a resource, or the seqjob_allocator
 * plugin -- alloc_rc==0 is exactly the same "job is pend" branch
 * sched_allocator() takes for a real no-matching-worker pend outcome, so
 * this fixture exercises the identical decision-construction and
 * cbuf_put(&core.decisions, ...) code path a real pend decision would.
 * The pend reason itself is pre-set directly on the job's top-level
 * `schedobj->reason` field before sched_allocator() ever sees it;
 * sched_make_compound_allocation() only overwrites that field when its own
 * ctx.reason is non-NULL, which never happens with zero primaries, so the
 * pre-set reason survives untouched into make_pend_decision().
 *
 * ── Worst-case derivation (do not copy a number from design.md prose
 *    without re-deriving it against live cbuf state -- this task's own
 *    instruction) ──────────────────────────────────────────────────────
 *
 * Both cbuf capacities are read LIVE from the real, freshly-initialized
 * core (sched.core.decisions.ncap / sched.core.jbuf.ncap -- xce_sched_core_t
 * struct fields, mylock.h's cbuf_t) rather than hardcoded, and asserted
 * against the documented values (ndecisions=5, jbuf capacity=1, core.c) as
 * a sanity check only.
 *
 * The worst case (least slack) is: `decisions` is ALREADY FULL at the
 * instant the stuck decision is popped off it (design.md's "decisions cbuf
 * already at or near its capacity-5 occupancy at the moment the stall
 * begins"). This test constructs exactly that precondition (prefills
 * `decisions` to full capacity BEFORE starting the stall), then proves,
 * against the REAL sched_allocator() thread and REAL cbuf blocking
 * semantics (mylock.c, not mocked):
 *
 *   1. `decisions` has room for exactly ONE more decision once the stuck
 *      one is popped (occupancy ncap-1) -- job A's PEND decision lands.
 *   2. The NEXT decision (job B's) is fully built (not dropped, not
 *      errored -- Req 3 AC2's "no job picked or decided ... is ever lost")
 *      but then genuinely BLOCKS trying to enqueue, because `decisions` is
 *      full again.
 *   3. `jbuf` (capacity 1) absorbs exactly ONE more job (C) beyond that --
 *      the allocator had already drained job B out of jbuf before getting
 *      stuck on `decisions` -- and then a further job (D) is refused for a
 *      sustained bounded window, proving the picker itself would now
 *      block too (Req 3 AC2's "and then sched_job_picker genuinely
 *      block"), not silently drop work.
 *   4. Once the stall clears, every decision produced before AND during
 *      the outage (the prefilled backlog plus jobs A/B/C) is eventually
 *      drained and published -- none lost (Req 3 AC2's "no job picked or
 *      decided during that bounded window is ever lost").
 *
 * This matches requirements.md's "as few as 1-2" floor (job A enqueued +
 * job B built-but-blocked = 2 additional decisions) at the `decisions`
 * layer, plus the analogous one-slot-of-slack-then-genuine-block pattern
 * one layer down at `jbuf` -- derived here from the live cbuf mechanics,
 * not copied from prose.
 *
 * All "prove it's genuinely blocked" checks use cbuf_tryput() (non-blocking)
 * in a short bounded-retry loop, never a real blocking call -- this test
 * can never hang even if the backpressure guarantee were broken.
 */
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include <unity/unity.h>

#include "client.h"
#include "controller.capnp.h"
#include "core.h"
#include "decision.h"
#include "mylock.h"
#include "preason.h"
#include "resreq.h"
#include "sched/types.h"
#include "scheduler.h"
#include "xce/cluster.h"
#include "xlist.h"

extern int scheduler_req_handler(xce_request_t*, void*);

#define N_DUMMY_PREFILL   5
#define N_REAL_JOBS_FED   3                 /* A, B, C */
#define N_TOTAL_DECISIONS (N_DUMMY_PREFILL + N_REAL_JOBS_FED)
#define POLL_STEP_NS      (5 * 1000 * 1000) /* 5ms */
#define POLL_TIMEOUT_MS   2000

/* ════════════════════════════════════════════════════════════════════════
 * Test double: __wrap_xcecli_set_pend_reason -- blocks on a condvar until
 * the test releases it, standing in for a stuck outage retry (see file
 * header comment). __wrap_xcecli_start / __wrap_xcecli_join are trivial
 * always-succeed doubles (same rationale as
 * test_scheduler_decision_publish_retry.c's own __wrap_xcecli_start) so
 * this file's one real sched_init() call actually starts the picker/
 * allocator threads instead of returning early.
 * ════════════════════════════════════════════════════════════════════════ */

static pthread_mutex_t g_pend_mtx          = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  g_pend_release_cond = PTHREAD_COND_INITIALIZER;
static pthread_cond_t  g_pend_entered_cond = PTHREAD_COND_INITIALIZER;
static int             g_pend_release      = 0;
static int             g_pend_entered      = 0;

static void reset_pend_wrap(void)
{
    g_pend_release = 0;
    g_pend_entered = 0;
}

int __wrap_xcecli_start(xcecli_ctx_t c)
{
    (void)c;
    return 0;
}

int __wrap_xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs)
{
    (void)c;
    (void)role;
    (void)attrs;
    return 0;
}

int __wrap_xcecli_set_pend_reason(xcecli_ctx_t c, xce_preason_set_t* reasons)
{
    (void)c;
    (void)reasons;

    pthread_mutex_lock(&g_pend_mtx);
    g_pend_entered = 1;
    pthread_cond_broadcast(&g_pend_entered_cond);
    while (!g_pend_release) {
        pthread_cond_wait(&g_pend_release_cond, &g_pend_mtx);
    }
    pthread_mutex_unlock(&g_pend_mtx);

    return 0;
}

static void wait_for_pend_wrap_entered(void)
{
    pthread_mutex_lock(&g_pend_mtx);
    while (!g_pend_entered) {
        pthread_cond_wait(&g_pend_entered_cond, &g_pend_mtx);
    }
    pthread_mutex_unlock(&g_pend_mtx);
}

static void release_pend_wrap(void)
{
    pthread_mutex_lock(&g_pend_mtx);
    g_pend_release = 1;
    pthread_cond_broadcast(&g_pend_release_cond);
    pthread_mutex_unlock(&g_pend_mtx);
}

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

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

/* build_pend_only_job - see file header comment for why COMPOUND + zero
 * primaries is used: it drives sched_allocator()'s real "job is pend"
 * decision-construction path without needing any worker/resource/queue
 * scaffolding. */
static xce_sched_job_t* build_pend_only_job(uint64_t jobid)
{
    xce_sched_job_t* job      = (xce_sched_job_t*)calloc(1, sizeof(*job));
    xce_sched_obj_t* schedobj = (xce_sched_obj_t*)calloc(1, sizeof(*schedobj));
    xce_compound_schedobj_t* compound
        = (xce_compound_schedobj_t*)calloc(1, sizeof(*compound));
    xce_compound_resreq_t* resreq
        = (xce_compound_resreq_t*)calloc(1, sizeof(*resreq));

    xlist_init(&(resreq->primaries), 1);

    compound->owner  = schedobj;
    compound->resreq = resreq;
    compound->reason = NULL;

    schedobj->tag        = XCE_RESREQ_COMPOUND;
    schedobj->u.compound = compound;
    /* Top-level field, distinct from compound->reason above -- see file
     * header comment for why this survives sched_make_compound_allocation()
     * untouched. */
    schedobj->reason = make_pendreason(jobid);

    job->jobid    = jobid;
    job->state    = JOB_STAT_PEND;
    job->schedobj = schedobj;

    return job;
}

static xce_decision_t* build_dummy_pend_decision(uint64_t jobid)
{
    return make_pend_decision(make_pendreason(jobid));
}

/* ════════════════════════════════════════════════════════════════════════
 * Drain thread -- stands in for driver.c's decision-drain consumer
 * (xce_sched_driver_run_normal_mode_round()'s cbuf_get(&core.decisions)
 * loop). Pops exactly N_TOTAL_DECISIONS decisions, publishing each via the
 * REAL sched_publish_pend_reason() (scheduler.c). Its first iteration is
 * the one that blocks in the wrapped call (see file header comment).
 * ════════════════════════════════════════════════════════════════════════ */

typedef struct {
    xce_scheduler_t* sched;
    uint64_t         seen_jobids[N_TOTAL_DECISIONS];
    int              seen_rcs[N_TOTAL_DECISIONS];
    int              n_seen;
} drain_ctx_t;

static void* drain_thread_fn(void* arg)
{
    drain_ctx_t* dctx = (drain_ctx_t*)arg;
    int          i;

    for (i = 0; i < N_TOTAL_DECISIONS; i++) {
        xce_decision_t*   decision = NULL;
        xce_pendreason_t* reasons[1];
        xce_preason_set_t reason_set;
        int               rc;

        cbuf_get(&(dctx->sched->core.decisions), &decision);

        reasons[0]          = decision->u.reason;
        reason_set.nreasons = 1;
        reason_set.reasons  = reasons;

        rc = sched_publish_pend_reason(dctx->sched, &reason_set);

        dctx->seen_jobids[dctx->n_seen] = decision->u.reason->jobid;
        dctx->seen_rcs[dctx->n_seen]    = rc;
        dctx->n_seen++;
    }

    return NULL;
}

/* ════════════════════════════════════════════════════════════════════════
 * The test
 * ════════════════════════════════════════════════════════════════════════ */

static void test_worst_case_backpressure_blocks_then_recovers_no_loss(void)
{
    xce_client_config_t  config;
    xce_cluster_config_t cluster_cfg;
    xce_scheduler_t      sched;
    pthread_t            drain_tid;
    drain_ctx_t          dctx;
    int                  rc;
    int                  i;
    int                  ncap_decisions;
    int                  ncap_jbuf;

    memset(&config, 0, sizeof(config));
    config.nmasters        = 1;
    config.masters[0]      = "127.0.0.1";
    config.master_ports[0] = 19997;
    config.from            = "sched_backpressure_test";
    config.key             = "keepsecret$";

    memset(&cluster_cfg, 0, sizeof(cluster_cfg));
    cluster_cfg.candidates_count = 1;
    strncpy(cluster_cfg.candidates[0].hostname, "127.0.0.1",
            XCE_MAXHOSTNAME_LEN - 1);
    cluster_cfg.candidates[0].port = 19997;
    strncpy(cluster_cfg.identity, "sched_backpressure_test",
            sizeof(cluster_cfg.identity) - 1);

    memset(&sched, 0, sizeof(sched));
    rc = sched_init(&sched, &config, &cluster_cfg, scheduler_req_handler);
    TEST_ASSERT_TRUE_MESSAGE(
        rc >= 0, "fixture: sched_init() must succeed (xcecli_start()/"
                 "xcecli_join() wrapped to succeed with no I/O) so the "
                 "real picker/allocator threads actually start");

    ncap_decisions = sched.core.decisions.ncap;
    ncap_jbuf      = sched.core.jbuf.ncap;
    TEST_ASSERT_TRUE_MESSAGE(ncap_decisions >= 1,
                             "sanity: decisions cbuf must have positive "
                             "capacity");
    TEST_ASSERT_TRUE_MESSAGE(ncap_jbuf >= 1,
                             "sanity: jbuf cbuf must have positive "
                             "capacity");
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        N_DUMMY_PREFILL, ncap_decisions,
        "documented invariant (core.c: ndecisions=5) -- if this fails, "
        "N_DUMMY_PREFILL/N_TOTAL_DECISIONS above must be re-derived "
        "against the new live capacity, not just this assertion changed");
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        1, ncap_jbuf,
        "documented invariant (core.c: jbuf capacity fixed at 1, a "
        "scheduler-core determinism invariant this task does not "
        "adjust)");

    /* --- Step 1: prefill `decisions` to FULL capacity -- the worst-case
     * (least-slack) precondition. --- */
    for (i = 0; i < ncap_decisions; i++) {
        xce_decision_t* d = build_dummy_pend_decision(9000 + (uint64_t)i);

        TEST_ASSERT_NOT_NULL(d);
        rc = cbuf_tryput(&(sched.core.decisions), &d);
        TEST_ASSERT_TRUE_MESSAGE(
            rc >= 0, "fixture: prefilling decisions to capacity must not "
                     "itself block");
    }

    /* --- Step 2: start the drain thread. Its first cbuf_get() pops the
     * first prefilled decision and calls the REAL
     * sched_publish_pend_reason(), which blocks inside the wrapped
     * xcecli_set_pend_reason() -- simulating an ongoing outage. --- */
    dctx.sched  = &sched;
    dctx.n_seen = 0;
    rc          = pthread_create(&drain_tid, NULL, drain_thread_fn, &dctx);
    TEST_ASSERT_EQUAL_INT(0, rc);

    wait_for_pend_wrap_entered();

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        ncap_decisions - 1, sched.core.decisions.len,
        "the drain thread's single cbuf_get() must have removed exactly "
        "one decision, leaving decisions at capacity-1");

    /* --- Step 3: feed job A through the REAL sched_allocator() thread
     * (via jbuf, standing in for the picker). With one free slot in
     * `decisions`, job A's PEND decision must land successfully -- this is
     * the "as few as 1" floor. --- */
    {
        xce_sched_job_t* jobA   = build_pend_only_job(9101);
        uint64_t         jobidA = jobA->jobid;

        TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobA) >= 0);
        rc = cbuf_tryput(&(sched.core.jbuf), &jobidA);
        TEST_ASSERT_TRUE_MESSAGE(
            rc >= 0, "jbuf must have room for job A -- it starts empty "
                     "since the dormant picker thread never touches it "
                     "(sched_go() is never called in this test)");
    }

    {
        int waited_ms = 0;

        while ((sched.core.decisions.len < ncap_decisions)
               && (waited_ms < POLL_TIMEOUT_MS)) {
            struct timespec ts = { 0, POLL_STEP_NS };

            nanosleep(&ts, NULL);
            waited_ms += 5;
        }
        TEST_ASSERT_EQUAL_INT_MESSAGE(
            ncap_decisions, sched.core.decisions.len,
            "job A's decision must be the ONE additional decision the "
            "worst-case (already-full) occupancy has room for");
    }

    /* --- Step 4: feed job B. `decisions` is full again, so
     * sched_allocator() must build job B's decision (Req 3 AC2: deciding is
     * not lost) but then genuinely BLOCK trying to enqueue it -- the
     * second half of the "1-2" floor. --- */
    {
        xce_sched_job_t* jobB   = build_pend_only_job(9102);
        uint64_t         jobidB = jobB->jobid;

        TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobB) >= 0);
        rc = cbuf_tryput(&(sched.core.jbuf), &jobidB);
        TEST_ASSERT_TRUE_MESSAGE(
            rc >= 0, "jbuf must have room for job B -- the allocator "
                     "already drained job A's jobid out of it");
    }

    {
        /* Bounded window for the allocator to reach its blocking
         * cbuf_put() -- decisions occupancy must not exceed capacity once
         * it does. */
        struct timespec ts = { 0, 100 * 1000 * 1000 };

        nanosleep(&ts, NULL);
    }
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        ncap_decisions, sched.core.decisions.len,
        "job B's decision must be stuck trying to enqueue -- decisions "
        "occupancy must never exceed its own capacity");

    /* --- Step 5: prove the genuine block reaches jbuf too (Req 3 AC2:
     * "...and then sched_job_picker genuinely block"). jbuf absorbs
     * exactly one more job (C) -- the allocator already drained jbuf
     * before getting stuck on `decisions` -- and a further push (D) must
     * then fail for a sustained bounded window. --- */
    {
        xce_sched_job_t* jobC   = build_pend_only_job(9103);
        uint64_t         jobidC = jobC->jobid;

        TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobC) >= 0);
        rc = cbuf_tryput(&(sched.core.jbuf), &jobidC);
        TEST_ASSERT_TRUE_MESSAGE(
            rc >= 0, "jbuf's one slot of slack must still accept job C");
    }
    {
        uint64_t jobidD        = 9104;
        int      any_succeeded = 0;
        int      probe;

        for (probe = 0; probe < 10; probe++) {
            struct timespec ts = { 0, 20 * 1000 * 1000 };

            rc = cbuf_tryput(&(sched.core.jbuf), &jobidD);
            if (rc >= 0) {
                any_succeeded = 1;
                break;
            }
            nanosleep(&ts, NULL);
        }
        TEST_ASSERT_FALSE_MESSAGE(
            any_succeeded,
            "genuine block, Req 3 AC2: with sched_allocator stuck on a "
            "full decisions cbuf, jbuf's single slot is already occupied "
            "by job C -- a further job (D) must never be accepted over a "
            "sustained bounded window, proving the picker itself would "
            "now block too, not silently drop work");
    }

    /* --- Step 6: the outage clears. Req 3 AC2: the scheduler must resume
     * automatically and publish every decision produced during the bounded
     * window, with none lost. --- */
    release_pend_wrap();

    rc = pthread_join(drain_tid, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        N_TOTAL_DECISIONS, dctx.n_seen,
        "every decision produced before AND during the outage -- the "
        "prefilled backlog plus jobs A/B/C -- must eventually be drained "
        "and published, none silently lost (Req 3 AC2)");

    for (i = 0; i < dctx.n_seen; i++) {
        TEST_ASSERT_TRUE_MESSAGE(
            dctx.seen_rcs[i] >= 0,
            "every drained decision's publish call must eventually "
            "report success");
    }

    {
        int saw_a = 0, saw_b = 0, saw_c = 0;

        for (i = 0; i < dctx.n_seen; i++) {
            if (dctx.seen_jobids[i] == 9101) {
                saw_a = 1;
            }
            if (dctx.seen_jobids[i] == 9102) {
                saw_b = 1;
            }
            if (dctx.seen_jobids[i] == 9103) {
                saw_c = 1;
            }
        }
        TEST_ASSERT_TRUE_MESSAGE(
            saw_a && saw_b && saw_c,
            "jobs A, B, and C -- decided/picked during the bounded outage "
            "-- must all eventually be published, not just the "
            "pre-outage backlog");
    }

    printf("test_worst_case_backpressure_blocks_then_recovers_no_loss: "
           "PASS (decisions.ncap=%d jbuf.ncap=%d, worst-case additional "
           "slack before genuine block = 1 decision fully enqueued + 1 "
           "decision built-but-blocked + 1 job buffered in jbuf)\n",
           ncap_decisions, ncap_jbuf);

    /* --- Teardown: cancel + join the real threads before tearing down
     * core state, matching sched_fini()'s own ordering (scheduler.c) minus
     * the apictx cleanup calls (xcecli_wakeup/xcecli_stop/xcecli_ctx_fini)
     * this fixture's wrapped xcecli_start() never fully set up (it does no
     * real connect, so ctx->failover.lock was never mylock_init'd) --
     * deliberately skipped rather than also wrapped, consistent with this
     * test suite's existing detect_leaks=0 convention (meson.build). --- */
    pthread_cancel(sched.core.jobpicker);
    pthread_cancel(sched.core.allocator);
    {
        uint64_t wake = 0;

        cbuf_tryput(&(sched.core.jbuf), &wake);
    }
    pthread_join(sched.core.jobpicker, NULL);
    pthread_join(sched.core.allocator, NULL);
    xce_sched_core_fini(&(sched.core));
}

int main(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_worst_case_backpressure_blocks_then_recovers_no_loss);
    return UNITY_END();
}