myxce/src/controller/testsuite/test_sched_admin_dispatch.c

/*
 * test_sched_admin_dispatch.c — Unity tests for the OPCode_schedAdmin
 * controller dispatch case (scheduler-core spec, task 10.1, Req 5.10).
 *
 * Exercises myreq_handler's OPCode_schedAdmin case through the real
 * decode+dispatch entry point exposed for tests, myreq_handler_for_test
 * (cluster.c, #ifdef XCE_TESTING) — same precedent as test_config_repl.c's
 * config-push-reply term-adoption tests (myreq_handler_for_test's own doc
 * comment cites peer_ctrl_msg_handler_for_test, ha.c, as the original
 * precedent for this style of test seam).
 *
 * Covers this task's two literal Observables:
 *   - a request against a non-primary controller returns the not-primary
 *     error immediately, with NO forwarding attempt (c->sched.ep is never
 *     touched — proven by pointing it at an intentionally-invalid sentinel
 *     that would crash the process on any dereference, not just asserted on
 *     the returned status);
 *   - a request against a primary with no scheduler connected returns the
 *     not-connected error immediately, carrying an explicit
 *     SchedAdminReply{error="scheduler not connected"} payload.
 *
 * Also includes a basic successful forward/reply round trip (register ->
 * forward -> a second, independently-dispatched OPCode_reply carrying a
 * SchedAdminReply wakes the blocked waiter), since the test scaffolding
 * built for the two Observables above made it straightforward. The full
 * dispatch test matrix (malformed requests, concurrent/serialized
 * round-trips, etc.) belongs to task 10.2.
 *
 * ── Task 10.2 additions (Req 5.10) ──────────────────────────────────────
 *
 * Task 10.2's literal scenario list is "standby rejection, disconnected-
 * scheduler rejection, forward timeout, and a successful forward/reply
 * round trip against the real scheduler-side handler from 9.4". The first
 * two are the Observables above (already task 10.1's). This file adds:
 *
 *   - test_schedadmin_forward_timeout: the genuinely-missing scenario --
 *     no existing test drove the 5s deadline to actually expire. Uses
 *     schedadmin_set_timeout_ms_for_test (cluster.c, XCE_TESTING-only) to
 *     shrink the real deadline so pthread_cond_timedwait genuinely times
 *     out on a fast schedule, rather than mocking the wait away or eating
 *     a real 5s per test run (no precedent for that in this suite --  see
 *     the seam's own doc comment in cluster.c for the survey).
 *
 *   - test_schedadmin_round_trip_rpc_status_only /
 *     test_schedadmin_round_trip_business_error: task 10.1's round-trip
 *     test only exercises ONE of the three outcome branches in the
 *     schedAdmin case's post-wait dispatch (cluster.c: got_done+reply,
 *     got_done+no-reply, timeout) -- the got_done+reply "success" shape.
 *     These two cover the other real, previously-uncovered branch
 *     (got_done, no SchedAdminReply payload -- an RPC-level outcome) and a
 *     business-level error shape (ok=false, e.g. "already recording" --
 *     requirements.md's "Reply field semantics by action" error rows),
 *     proving reply.status stays 0 for a completed request that merely
 *     carries a domain error, never conflated with an RPC/timeout failure.
 *
 *   ON "against the real scheduler-side handler from 9.4": task 9.4's own
 *   scheduler_req_handler()/sched_handle_admin_request() are proven for
 *   real, end-to-end, including sched_record_status()'s actual field
 *   construction, by scheduler/testsuite/test_admin_request.c (already
 *   landed with task 9.4) -- that file white-box #includes handler.c,
 *   recording.c, and their scheduler-internal dependencies (core.c,
 *   driver.c, snapshot*.c) directly into the SCHEDULER's own test binary
 *   and spies on xcecli_reply_schedadmin() to capture the real reply. This
 *   controller test binary (controller_unit_test) is a wholly separate
 *   meson target/executable with its own main() and its own global state
 *   (cluster_t/ha_state_t fixtures, not xce_scheduler_t) -- pulling
 *   scheduler/src/handler.c + recording.c (which itself needs core.h and
 *   the rest of xce_scheduler_t's dependency graph) into THIS binary to
 *   invoke scheduler_req_handler() literally would require the same kind
 *   of careful selective-source-linking test_admin_request.c's own header
 *   comment documents doing for the scheduler side, replicated cross-
 *   directory inside an already-large, HA-critical controller test target
 *   -- real risk (symbol/global collisions between two "whole daemon"
 *   white-box bundles sharing one process image) for no correctness gain,
 *   since 9.4's own suite already covers that ground rigorously. A genuine
 *   two-process E2E (real controller + real scheduler binaries, a real
 *   connection between them) is an E2E/soak-test concern, not this task's
 *   scope. So: this file proves the CONTROLLER's forward/relay mechanism
 *   against the real wire contract (xce_encode/xce_decode, the same
 *   functions production uses) and the exact reply shapes 9.4's real
 *   sched_record_status()/sched_handle_admin_request() are proven (by
 *   their own suite) to produce -- not a second, redundant proof that the
 *   scheduler-side handler itself is correct.
 */
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include <unity/unity.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "ha.h"
#include "netio.h"
#include "request.h"
#include "typemap.h"
#include "xce/errnox.h"

/* myreq_handler exposed for tests (cluster.c). Packet ownership follows
 * myreq_handler's own contract — see its doc comment in cluster.c: `pkt` is
 * consumed (freed) by the call, never touch/free it afterward. */
extern int myreq_handler_for_test(endpoint_t* ep, io_packet_t* pkt);

/* Test-only override for the OPCode_schedAdmin case's forward-and-block
 * deadline (cluster.c, XCE_TESTING-only, task 10.2) -- see its own doc
 * comment in cluster.c for why this seam exists. */
extern void schedadmin_set_timeout_ms_for_test(long ms);

/* ── Fixture ──────────────────────────────────────────────────────────────
 *
 * The OPCode_schedAdmin dispatch case (and the ha_dispatch_gate primary/
 * standby gate ahead of it) only touch c->ha.role, c->ha.config_serve_fence /
 * c->ha.upgrade_materialization_fence (both NULL-safe no-ops when
 * c->ha.raft_c is left NULL — config_serve_fenced/upgrade_materialization_
 * fenced both guard on `if (!c) return 0;`), c->ha.n_peers (0 → single-node
 * quorum trivially satisfied, ha_has_write_quorum), c->sched.{lock,ep}, and
 * c->myhost (NULL-safe — xce_reply's encode path null-checks its `from`
 * argument, matching every other encode_*_ptr in controller.capnp.c). None
 * of this needs persist_init/persist_start (unlike test_config_repl.c's
 * t42_make_cluster, which exists for config-marker/WAL-touching tests) — a
 * bare calloc'd cluster_t is sufficient and keeps this file free of
 * temp-directory bookkeeping. */
static cluster_t* sa_make_cluster(void)
{
    cluster_t* c = calloc(1, sizeof(cluster_t));
    TEST_ASSERT_NOT_NULL(c);
    mylock_init(&(c->sched.lock));
    return c;
}

static void sa_teardown(cluster_t* c)
{
    if (c == NULL)
        return;
    mylock_fini(&(c->sched.lock));
    free(c);
}

/* Build a wire-encoded OPCode_schedAdmin request packet carrying a minimal
 * SchedAdminRequest — task 10.1 passes every field through untouched, so the
 * exact action/recordFile content is immaterial to these tests. */
static io_packet_t* sa_build_request_pkt(void)
{
    sched_admin_req_t areq;
    memset(&areq, 0, sizeof(areq));
    areq.action = SchedAdminAction_recordStatus;

    xce_request_t req;
    memset(&req, 0, sizeof(req));
    req.opcode             = OPCode_schedAdmin;
    req.body_which         = Request_body_schedadminmsg;
    req.body.schedadminmsg = &areq;

    io_packet_t* pkt = calloc(1, sizeof(*pkt));
    TEST_ASSERT_NOT_NULL(pkt);
    TEST_ASSERT_TRUE_MESSAGE(xce_encode(&req, pkt) >= 0,
                             "failed to encode SchedAdminRequest packet");
    return pkt;
}

/* Pop the single reply packet myreq_handler queued on ep->outq and decode it
 * into *out (caller must xce_free_request(out) when done with it). */
static void sa_decode_queued_reply(endpoint_t* ep, xce_request_t* out)
{
    data_item_t* it = endpoint_deq(ep);
    TEST_ASSERT_NOT_NULL_MESSAGE(it, "expected a reply packet on ep->outq");

    io_packet_t* pkt = ITPTR_TO_PACKET(it);
    memset(out, 0, sizeof(*out));
    int rc = xce_decode(pkt, out);
    TEST_ASSERT_EQUAL_INT(0, rc);
}

/* Monotonic milliseconds -- for measuring that the forward-timeout test
 * genuinely waited out its (shrunk) deadline rather than returning early.
 * Same construction as test_commit_resolve.c's mono_ms_r(). */
static uint64_t sa_mono_ms(void)
{
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return (uint64_t)ts.tv_sec * 1000u + (uint64_t)ts.tv_nsec / 1000000u;
}

/* ── Observable 1: non-primary rejects immediately, no forwarding attempt ──
 *
 * ha_dispatch_gate (ha.c) denies OPCode_schedAdmin for every non-primary
 * role — it is absent from the CTRL_STANDBY/CTRL_INIT/CTRL_ELECTING
 * allow-lists, exactly like OPCode_upgradeConfig — so the request is
 * rejected BEFORE myreq_handler's switch is ever reached; the
 * OPCode_schedAdmin case's own ha_is_primary() check is pure defensive
 * belt-and-suspenders for a role change racing between the gate and the
 * switch. c->sched.ep is set to a deliberately-invalid, non-NULL sentinel:
 * if a regression ever let this request reach the forwarding code (both
 * mylock_lock and the `sched.ep == NULL` check would pass on a garbage
 * non-NULL pointer), xce_schedadmin_req would dereference it and crash —
 * making this a self-verifying proof that no forwarding attempt was made,
 * not just an assertion on the returned status. */
static void test_schedadmin_rejects_when_not_primary(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_STANDBY);

    endpoint_t* poison = (endpoint_t*)(uintptr_t)0x1;
    c->sched.ep        = poison;

    endpoint_t ep;
    memset(&ep, 0, sizeof(ep));
    ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();
    int          rc  = myreq_handler_for_test(&ep, pkt);

    TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rc);
    TEST_ASSERT_TRUE_MESSAGE(
        c->sched.ep == poison,
        "c->sched.ep must be untouched when the request never reaches the "
        "dispatch case");

    xce_request_t decoded;
    sa_decode_queued_reply(&ep, &decoded);
    TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which);
    TEST_ASSERT_NOT_NULL(decoded.body.replymsg);
    TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, decoded.body.replymsg->status);
    TEST_ASSERT_EQUAL_INT(RequestReply_data_nodata,
                          decoded.body.replymsg->data_which);
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_rejects_when_not_primary: PASS\n");
}

/* ── bug-1700: a standby dispatch-gate rejection must redirect to the known
 * primary ──────────────────────────────────────────────────────────────────
 *
 * ha_dispatch_gate() rejects OPCode_schedAdmin on a CTRL_STANDBY node with
 * -XCE_ERR_STANDBY (ha.c's CTRL_STANDBY case default branch, ~line 4774) --
 * this node MAY know who the primary is (c->ha.primary_host/primary_port).
 * Before the fix, myreq_handler's generic ha_dispatch_gate rejection block
 * (cluster.c ~406-414) never called fill_redirect_reply() (the exact same
 * one-line helper OPCode_join's handler already uses, join.c:277), so
 * redirect_host/redirect_port were left at their zero-value defaults on
 * EVERY dispatch-gate rejection -- xceadmin's client-side, already fully
 * implemented and generic follow_redirects() had nothing to follow and
 * surfaced the bare -XCE_ERR_STANDBY status as-is instead of transparently
 * retrying against the primary. This is the exact scenario the user hit
 * running `xceadmin scheduler record status` against a standby. */
static void test_schedadmin_standby_with_known_primary_redirects(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_STANDBY);
    strncpy(c->ha.primary_host, "10.0.1.7", sizeof(c->ha.primary_host) - 1);
    c->ha.primary_port = 7001;

    /* Poison sentinel mirrors test_schedadmin_rejects_when_not_primary above
     * -- a gate rejection must never reach the forwarding code. */
    endpoint_t* poison = (endpoint_t*)(uintptr_t)0x1;
    c->sched.ep         = poison;

    endpoint_t ep;
    memset(&ep, 0, sizeof(ep));
    ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();
    int          rc  = myreq_handler_for_test(&ep, pkt);

    TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rc);
    TEST_ASSERT_TRUE_MESSAGE(
        c->sched.ep == poison,
        "c->sched.ep must be untouched -- a gate rejection never forwards");

    xce_request_t decoded;
    sa_decode_queued_reply(&ep, &decoded);
    TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rr->status);
    TEST_ASSERT_NOT_NULL_MESSAGE(
        rr->redirect_host,
        "bug-1700: a standby dispatch-gate rejection with a known primary "
        "must carry that primary's address so xceadmin's follow_redirects() "
        "can transparently retry there");
    TEST_ASSERT_EQUAL_STRING("10.0.1.7", rr->redirect_host);
    TEST_ASSERT_EQUAL_UINT16(7001, rr->redirect_port);
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_standby_with_known_primary_redirects: PASS\n");
}

/* Regression companion: a standby that does NOT yet know any live primary
 * (fresh ha_state_t straight out of calloc -- primary_host empty,
 * primary_port 0, e.g. very early boot / mid-election / primary just died
 * with no successor elected yet) must NOT crash and must produce a reply
 * the client correctly treats as non-redirecting: discovery.py's
 * `_redirect_target()` requires BOTH a non-empty host AND a non-zero port
 * to treat a reply as a redirect, so an empty host / zero port here is
 * harmless and falls back to today's bare-rejection behavior. Proves the
 * fix does not regress the genuinely-no-primary-yet case. */
static void test_schedadmin_standby_no_known_primary_no_redirect(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_STANDBY);
    /* c->ha.primary_host/primary_port left at their calloc'd zero value. */

    endpoint_t ep;
    memset(&ep, 0, sizeof(ep));
    ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();
    int          rc  = myreq_handler_for_test(&ep, pkt);

    TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rc);

    xce_request_t decoded;
    sa_decode_queued_reply(&ep, &decoded);
    TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT(-XCE_ERR_STANDBY, rr->status);
    TEST_ASSERT_TRUE_MESSAGE(
        rr->redirect_host == NULL || rr->redirect_host[0] == '\0',
        "no known primary: redirect_host must decode as empty/absent, "
        "never garbage -- the client's _redirect_target() treats an empty "
        "host as 'no redirect, use this reply as-is'");
    TEST_ASSERT_EQUAL_UINT16(0, rr->redirect_port);
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_standby_no_known_primary_no_redirect: PASS\n");
}

/* ── Observable 2: primary with no scheduler connected rejects immediately ──
 */
static void test_schedadmin_rejects_when_not_connected(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_PRIMARY);
    c->sched.ep = NULL;

    endpoint_t ep;
    memset(&ep, 0, sizeof(ep));
    ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();
    int          rc  = myreq_handler_for_test(&ep, pkt);

    TEST_ASSERT_EQUAL_INT(0, rc);

    xce_request_t decoded;
    sa_decode_queued_reply(&ep, &decoded);
    TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT(0, rr->status);
    TEST_ASSERT_EQUAL_INT(RequestReply_data_schedadminreplymsg, rr->data_which);
    TEST_ASSERT_NOT_NULL(rr->data.schedadminreplymsg);
    TEST_ASSERT_EQUAL_STRING("scheduler not connected",
                             rr->data.schedadminreplymsg->error);
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_rejects_when_not_connected: PASS\n");
}

/* ── Observable 2b (design review follow-up, 2026-08-19): admission-
 * exhausted requests fast-reject instead of blocking a vtask thread ────────
 *
 * ha-async-io.md Rule 3: bounded blocking on a vtask thread requires an
 * admission reservation. c->vtaskmgr starts gate-DISABLED by a bare
 * sa_make_cluster() (liveness_reserved defaults to 0 on a calloc'd
 * vtaskmgr_t -- vtaskmgr_blocking_acquire's own "Gate disabled: admit
 * uncounted (legacy: nothing reserved)" branch, vtaskmgr.c), matching
 * production behavior before vtaskmgr_set_liveness_reserved is called
 * (cluster.c's real init path) -- this is why every OTHER test in this
 * file, which never calls that setter, is completely unaffected by this
 * feature. This test explicitly enables the gate with a 1-slot cap
 * (vtaskmgr_set_liveness_reserved(..., 1) floors cap at 1 regardless of
 * n_vcpus, vtaskmgr_blocking_cap's own doc comment) and pre-saturates that
 * one slot -- simulating a concurrent commit-wait or another in-flight
 * schedAdmin request already holding it -- before dispatching. The poison
 * c->sched.ep sentinel (same technique test_schedadmin_rejects_when_not_
 * primary uses above) proves the fast-reject happens BEFORE
 * mylock_lock(&c->sched.lock)/any forwarding attempt, not merely that the
 * eventual reply looks right. */
static void test_schedadmin_fast_rejects_when_admission_exhausted(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_PRIMARY);

    endpoint_t* poison = (endpoint_t*)(uintptr_t)0x1;
    c->sched.ep        = poison;

    vtaskmgr_set_liveness_reserved(&c->vtaskmgr, 1);
    TEST_ASSERT_EQUAL_INT(0, vtaskmgr_blocking_acquire(&c->vtaskmgr));

    endpoint_t ep;
    memset(&ep, 0, sizeof(ep));
    ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();
    int          rc  = myreq_handler_for_test(&ep, pkt);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_TRUE_MESSAGE(
        c->sched.ep == poison,
        "c->sched.ep must be untouched -- admission-exhausted requests must "
        "fast-reject before ever locking c->sched.lock or forwarding");

    xce_request_t decoded;
    sa_decode_queued_reply(&ep, &decoded);
    TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT(0, rr->status);
    TEST_ASSERT_EQUAL_INT(RequestReply_data_schedadminreplymsg, rr->data_which);
    TEST_ASSERT_NOT_NULL(rr->data.schedadminreplymsg);
    TEST_ASSERT_EQUAL_STRING("server busy, try again",
                             rr->data.schedadminreplymsg->error);
    xce_free_request(&decoded);

    vtaskmgr_blocking_release(&c->vtaskmgr); /* release our pre-saturating slot */
    sa_teardown(c);
    printf("test_schedadmin_fast_rejects_when_admission_exhausted: PASS\n");
}

/* ── Basic round trip: forward -> scheduler reply wakes the waiter ──────────
 *
 * c->sched.ep points at a real (if unconnected) endpoint_t so the dispatch
 * case's forward+wait path is exercised for real: xce_schedadmin_req just
 * encodes+enqueues (like xce_reply — see request.c), so a zeroed,
 * non-socket-backed endpoint_t is sufficient — the same technique the two
 * tests above already rely on for the caller-side ep. A second thread plays
 * the scheduler: after a short delay it independently dispatches an
 * OPCode_reply/SchedAdminReply through myreq_handler_for_test on the SAME
 * cluster_t, exactly how the real reply arrives in production (a second
 * vtask handling c->sched.ep's inbound traffic) — this drives the real
 * schedadmin_signal()/demux path in the OPCode_reply case, not a shortcut. */
typedef struct {
    cluster_t* c;
} sa_reply_thread_arg_t;

static void* sa_deliver_scheduler_reply(void* argp)
{
    sa_reply_thread_arg_t* arg = (sa_reply_thread_arg_t*)argp;

    /* Give the main thread time to register the pending waiter and start
     * blocking on the condvar — comfortably inside the dispatch case's 5s
     * deadline. */
    usleep(200000);

    sched_admin_reply_t sreply;
    memset(&sreply, 0, sizeof(sreply));
    sreply.ok         = 1;
    sreply.recording  = 1;
    sreply.file       = "/tmp/xce_sa_roundtrip.replay";
    sreply.eventCount = 42;
    sreply.startedAt  = 1234;

    xce_reply_t rr;
    memset(&rr, 0, sizeof(rr));
    rr.opcode                  = OPCode_schedAdmin;
    rr.status                  = 0;
    rr.data_which              = RequestReply_data_schedadminreplymsg;
    rr.data.schedadminreplymsg = &sreply;

    xce_request_t req;
    memset(&req, 0, sizeof(req));
    req.opcode        = OPCode_reply;
    req.body_which    = Request_body_replymsg;
    req.body.replymsg = &rr;

    io_packet_t* pkt = calloc(1, sizeof(*pkt));
    TEST_ASSERT_NOT_NULL(pkt);
    TEST_ASSERT_TRUE(xce_encode(&req, pkt) >= 0);

    endpoint_t sched_reply_ep;
    memset(&sched_reply_ep, 0, sizeof(sched_reply_ep));
    sched_reply_ep.owner = arg->c;

    myreq_handler_for_test(&sched_reply_ep, pkt);
    return NULL;
}

static void test_schedadmin_round_trip_success(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_PRIMARY);

    endpoint_t sched_conn;
    memset(&sched_conn, 0, sizeof(sched_conn));
    c->sched.ep = &sched_conn;

    sa_reply_thread_arg_t arg = { .c = c };
    pthread_t             tid;
    TEST_ASSERT_EQUAL_INT(
        0, pthread_create(&tid, NULL, sa_deliver_scheduler_reply, &arg));

    endpoint_t caller_ep;
    memset(&caller_ep, 0, sizeof(caller_ep));
    caller_ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();
    int          rc  = myreq_handler_for_test(&caller_ep, pkt);

    pthread_join(tid, NULL);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc,
        "the round trip must complete (woken by the reply), not time out");

    xce_request_t decoded;
    sa_decode_queued_reply(&caller_ep, &decoded);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT(0, rr->status);
    TEST_ASSERT_EQUAL_INT(RequestReply_data_schedadminreplymsg, rr->data_which);
    sched_admin_reply_t* got = rr->data.schedadminreplymsg;
    TEST_ASSERT_NOT_NULL(got);
    TEST_ASSERT_EQUAL_INT(1, got->ok);
    TEST_ASSERT_EQUAL_INT(1, got->recording);
    TEST_ASSERT_EQUAL_UINT64(42, got->eventCount);
    TEST_ASSERT_EQUAL_UINT64(1234, got->startedAt);
    TEST_ASSERT_EQUAL_STRING("/tmp/xce_sa_roundtrip.replay", got->file);
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_round_trip_success: PASS\n");
}

/* ── Task 10.2: forward timeout ─────────────────────────────────────────────
 *
 * c->sched.ep points at a real (if unconnected) endpoint_t, exactly like the
 * round-trip test above, so xce_schedadmin_req really forwards and the
 * dispatch case really blocks on pthread_cond_timedwait -- but NO second
 * thread ever delivers a reply. schedadmin_set_timeout_ms_for_test shrinks
 * the deadline so the real wait-then-give-up path completes on a fast test
 * schedule; the override is restored to the production default IMMEDIATELY
 * after myreq_handler_for_test returns, before any TEST_ASSERT below can
 * longjmp out on failure (Unity's assertion macros unwind via longjmp) --
 * so a failing assertion in this test can never leave a later test in this
 * binary running against a shrunk deadline. */
static void test_schedadmin_forward_timeout(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_PRIMARY);

    endpoint_t sched_conn;
    memset(&sched_conn, 0, sizeof(sched_conn));
    c->sched.ep = &sched_conn;

    endpoint_t caller_ep;
    memset(&caller_ep, 0, sizeof(caller_ep));
    caller_ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();

    const long timeout_ms = 200;
    schedadmin_set_timeout_ms_for_test(timeout_ms);

    uint64_t t0   = sa_mono_ms();
    int      rc   = myreq_handler_for_test(&caller_ep, pkt);
    uint64_t took = sa_mono_ms() - t0;

    schedadmin_set_timeout_ms_for_test(5000); /* restore before any assert */

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc,
        "the dispatch case always replies -- it does not propagate "
        "-ETIMEDOUT as its own return, even on a genuine timeout");

    TEST_ASSERT_TRUE_MESSAGE(
        took + 20 >= (uint64_t)timeout_ms,
        "must not give up meaningfully before the (shrunk) deadline -- this "
        "is what proves the real pthread_cond_timedwait deadline fired, not "
        "a shortcut that skipped the wait");
    TEST_ASSERT_TRUE_MESSAGE(
        took < (uint64_t)timeout_ms + 2000,
        "must give up within a bounded window after the deadline, not hang");

    xce_request_t decoded;
    sa_decode_queued_reply(&caller_ep, &decoded);
    TEST_ASSERT_EQUAL_INT(Request_body_replymsg, decoded.body_which);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rr->status,
        "a timeout is relayed as an ok-RPC SchedAdminReply carrying "
        "error=\"request timeout\", per requirements.md AC 10 -- not a "
        "negative RPC status");
    TEST_ASSERT_EQUAL_INT(RequestReply_data_schedadminreplymsg, rr->data_which);
    sched_admin_reply_t* got = rr->data.schedadminreplymsg;
    TEST_ASSERT_NOT_NULL(got);
    TEST_ASSERT_EQUAL_INT(0, got->ok);
    TEST_ASSERT_EQUAL_STRING("request timeout", got->error);
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_forward_timeout: PASS\n");
}

/* ── Task 10.2: additional round-trip outcome branches ──────────────────────
 *
 * Shared "second thread plays the scheduler" helper, generalized from
 * test_schedadmin_round_trip_success's sa_deliver_scheduler_reply above so
 * both new tests below can drive either outcome (a SchedAdminReply payload,
 * or a bare RPC status with none) through the SAME real
 * schedadmin_signal()/demux path, without duplicating the thread/encode
 * boilerplate. Task 10.1's own sa_deliver_scheduler_reply is left untouched. */
typedef struct {
    cluster_t*           c;
    int                  status;
    sched_admin_reply_t* payload; /* NULL => RequestReply_data_nodata */
} sa_deliver_arg_t;

static void* sa_deliver_reply_generic(void* argp)
{
    sa_deliver_arg_t* arg = (sa_deliver_arg_t*)argp;

    usleep(200000); /* same margin as sa_deliver_scheduler_reply above */

    xce_reply_t rr;
    memset(&rr, 0, sizeof(rr));
    rr.opcode = OPCode_schedAdmin;
    rr.status = arg->status;
    if (arg->payload != NULL) {
        rr.data_which              = RequestReply_data_schedadminreplymsg;
        rr.data.schedadminreplymsg = arg->payload;
    } else {
        rr.data_which = RequestReply_data_nodata;
    }

    xce_request_t req;
    memset(&req, 0, sizeof(req));
    req.opcode        = OPCode_reply;
    req.body_which    = Request_body_replymsg;
    req.body.replymsg = &rr;

    io_packet_t* pkt = calloc(1, sizeof(*pkt));
    TEST_ASSERT_NOT_NULL(pkt);
    TEST_ASSERT_TRUE(xce_encode(&req, pkt) >= 0);

    endpoint_t sched_reply_ep;
    memset(&sched_reply_ep, 0, sizeof(sched_reply_ep));
    sched_reply_ep.owner = arg->c;

    myreq_handler_for_test(&sched_reply_ep, pkt);
    return NULL;
}

/* Exercises the got_done && got_reply==NULL branch (cluster.c, OPCode_reply
 * case): the scheduler's OPCode_reply can legitimately carry no
 * SchedAdminReply payload (an RPC-level outcome) -- the controller must
 * still reply promptly, relaying the bare status, not treat it as a
 * timeout. Never exercised by task 10.1's round-trip test, which always
 * supplies a payload. */
static void test_schedadmin_round_trip_rpc_status_only(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_PRIMARY);

    endpoint_t sched_conn;
    memset(&sched_conn, 0, sizeof(sched_conn));
    c->sched.ep = &sched_conn;

    sa_deliver_arg_t arg = { .c = c, .status = -EIO, .payload = NULL };
    pthread_t        tid;
    TEST_ASSERT_EQUAL_INT(
        0, pthread_create(&tid, NULL, sa_deliver_reply_generic, &arg));

    endpoint_t caller_ep;
    memset(&caller_ep, 0, sizeof(caller_ep));
    caller_ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();
    int          rc  = myreq_handler_for_test(&caller_ep, pkt);

    pthread_join(tid, NULL);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc,
        "the dispatch case always replies, even when the scheduler's "
        "OPCode_reply carries no SchedAdminReply payload");

    xce_request_t decoded;
    sa_decode_queued_reply(&caller_ep, &decoded);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        -EIO, rr->status,
        "a payload-less scheduler reply relays its bare RPC status verbatim");
    TEST_ASSERT_EQUAL_INT(RequestReply_data_nodata, rr->data_which);
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_round_trip_rpc_status_only: PASS\n");
}

/* A business-level error from the scheduler (ok=false, e.g. "already
 * recording" -- requirements.md's "Reply field semantics by action" error
 * rows) is a NORMAL completed round trip, not an RPC failure. Proves the
 * controller relays that shape verbatim with reply.status == 0 -- a
 * domain error is never conflated with an RPC/timeout error. */
static void test_schedadmin_round_trip_business_error(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_PRIMARY);

    endpoint_t sched_conn;
    memset(&sched_conn, 0, sizeof(sched_conn));
    c->sched.ep = &sched_conn;

    sched_admin_reply_t sreply;
    memset(&sreply, 0, sizeof(sreply));
    sreply.ok         = 0;
    sreply.error      = "already recording";
    sreply.recording  = 1;
    sreply.file       = "/var/log/xce/scheduler-replay-x.replay";
    sreply.eventCount = 7;
    sreply.startedAt  = 999;

    sa_deliver_arg_t arg = { .c = c, .status = 0, .payload = &sreply };
    pthread_t        tid;
    TEST_ASSERT_EQUAL_INT(
        0, pthread_create(&tid, NULL, sa_deliver_reply_generic, &arg));

    endpoint_t caller_ep;
    memset(&caller_ep, 0, sizeof(caller_ep));
    caller_ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();
    int          rc  = myreq_handler_for_test(&caller_ep, pkt);

    pthread_join(tid, NULL);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc,
        "a business-level error from the scheduler is still a completed "
        "round trip");

    xce_request_t decoded;
    sa_decode_queued_reply(&caller_ep, &decoded);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rr->status,
        "ok=false is a business error, not an RPC/timeout error -- "
        "reply.status must stay 0");
    TEST_ASSERT_EQUAL_INT(RequestReply_data_schedadminreplymsg, rr->data_which);
    sched_admin_reply_t* got = rr->data.schedadminreplymsg;
    TEST_ASSERT_NOT_NULL(got);
    TEST_ASSERT_EQUAL_INT(0, got->ok);
    TEST_ASSERT_EQUAL_STRING("already recording", got->error);
    TEST_ASSERT_EQUAL_INT(1, got->recording);
    TEST_ASSERT_EQUAL_STRING("/var/log/xce/scheduler-replay-x.replay",
                             got->file);
    TEST_ASSERT_EQUAL_UINT64(7, got->eventCount);
    TEST_ASSERT_EQUAL_UINT64(999, got->startedAt);
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_round_trip_business_error: PASS\n");
}

/* ── bug-1693 regression: scheduler disconnect wakes the waiter immediately
 * (Decision Log fidelity fix) ───────────────────────────────────────────────
 *
 * Before this fix, handle_scheduler_leave() (sched.c) only ever nulled
 * c->sched.ep on a scheduler disconnect -- it never signalled the pending
 * OPCode_schedAdmin waiter, so a disconnect mid-wait could only ever be
 * observed at the FULL SCHEDADMIN_TIMEOUT_MS deadline (5s in production).
 * That is exactly the Decision Log violation this task fixes: "No client
 * reply-wait timeout as a fix for a hung request -- fail loud at the real
 * transport/source instead of masking with a deadline backstop."
 *
 * c->sched.ep points at a real (if unconnected) endpoint_t, exactly like
 * test_schedadmin_round_trip_success above, so the dispatch case's
 * forward+wait path runs for real. A second thread plays the disconnect by
 * calling schedadmin_signal_disconnect() (cluster.c) directly -- the exact,
 * real, unmodified production function handle_scheduler_leave() (sched.c)
 * calls to wake the waiter, and the entire fix under test here.
 *
 * NOT calling handle_scheduler_leave() itself: this file's binary
 * (controller_unit_test) does not link the real sched.c at all (see
 * meson.build's comment on sched_push_unit_test/sched_worker_change_unit_
 * test for why -- sched.c's on_scheduler_join()/collect_confvars() collide
 * at link time with fake definitions several OTHER controller_unit_test_src
 * files already provide). test_config_repl.c (also in this binary) already
 * supplies its OWN no-op handle_scheduler_leave() stub for that reason,
 * predating this fix -- calling handle_scheduler_leave() here would
 * silently exercise THAT stub, not sched.c's real implementation, and prove
 * nothing. handle_scheduler_leave()'s own body is a thin, 2-statement
 * wrapper (call this function, then null c->sched.ep under c->sched.lock)
 * -- calling schedadmin_signal_disconnect() directly exercises the entire
 * real fix mechanism (the new function, and the real dispatch-side wait it
 * wakes) with zero loss of fidelity versus going through that wrapper. */
static void* sa_deliver_scheduler_disconnect(void* argp)
{
    (void)argp;

    /* Same margin as sa_deliver_scheduler_reply above -- comfortably inside
     * the 5s deadline, giving the main thread time to register the pending
     * waiter and start blocking on the condvar first. */
    usleep(200000);

    schedadmin_signal_disconnect();
    return NULL;
}

static void test_schedadmin_disconnect_wakes_immediately(void)
{
    cluster_t* c = sa_make_cluster();
    atomic_store(&c->ha.role, CTRL_PRIMARY);

    endpoint_t sched_conn;
    memset(&sched_conn, 0, sizeof(sched_conn));
    c->sched.ep = &sched_conn;

    pthread_t tid;
    TEST_ASSERT_EQUAL_INT(
        0, pthread_create(&tid, NULL, sa_deliver_scheduler_disconnect, c));

    endpoint_t caller_ep;
    memset(&caller_ep, 0, sizeof(caller_ep));
    caller_ep.owner = c;

    io_packet_t* pkt = sa_build_request_pkt();

    uint64_t t0   = sa_mono_ms();
    int      rc   = myreq_handler_for_test(&caller_ep, pkt);
    uint64_t took = sa_mono_ms() - t0;

    pthread_join(tid, NULL);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc, "the dispatch case always replies, even on a disconnect");

    TEST_ASSERT_TRUE_MESSAGE(
        took < 1000,
        "a scheduler disconnect mid-wait must wake the waiter immediately "
        "(well under the real, unshrunk 5s SCHEDADMIN_TIMEOUT_MS deadline) "
        "-- this is the regression this test locks: pre-fix, nothing ever "
        "signalled the waiter on disconnect, so this only ever completed at "
        "the full 5s deadline");

    xce_request_t decoded;
    sa_decode_queued_reply(&caller_ep, &decoded);
    xce_reply_t* rr = decoded.body.replymsg;
    TEST_ASSERT_NOT_NULL(rr);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rr->status,
        "a disconnect is relayed as an ok-RPC SchedAdminReply carrying a "
        "distinguishable error, not a negative RPC status");
    TEST_ASSERT_EQUAL_INT(RequestReply_data_schedadminreplymsg, rr->data_which);
    sched_admin_reply_t* got = rr->data.schedadminreplymsg;
    TEST_ASSERT_NOT_NULL(got);
    TEST_ASSERT_EQUAL_INT(0, got->ok);
    TEST_ASSERT_EQUAL_STRING_MESSAGE(
        "scheduler disconnected", got->error,
        "must be distinguishable from \"request timeout\" -- proves this "
        "woke via the disconnect signal, not the deadline");
    xce_free_request(&decoded);

    sa_teardown(c);
    printf("test_schedadmin_disconnect_wakes_immediately: PASS\n");
}

/* ── Suite entry point ──────────────────────────────────────────────────── */
void test_sched_admin_dispatch(void)
{
    UnitySetTestFile(__FILE__);
    RUN_TEST(test_schedadmin_rejects_when_not_primary);
    RUN_TEST(test_schedadmin_standby_with_known_primary_redirects);
    RUN_TEST(test_schedadmin_standby_no_known_primary_no_redirect);
    RUN_TEST(test_schedadmin_rejects_when_not_connected);
    RUN_TEST(test_schedadmin_fast_rejects_when_admission_exhausted);
    RUN_TEST(test_schedadmin_round_trip_success);
    RUN_TEST(test_schedadmin_forward_timeout);
    RUN_TEST(test_schedadmin_round_trip_rpc_status_only);
    RUN_TEST(test_schedadmin_round_trip_business_error);
    RUN_TEST(test_schedadmin_disconnect_wakes_immediately);
}