From 045be6b3d6b8bf62d4394ee367b337ac789cc592 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 6 Aug 2026 03:38:42 +0000 Subject: [PATCH] feat(scheduler-core): add controller-side OPCode_schedAdmin dispatch (task 10.1) Adds the spec's one and only src/controller/ change: a new case OPCode_schedAdmin in myreq_handler() (cluster.c) that mirrors the existing OPCode_upgradeConfig case -- primary-only gate, explicit reply on every path, stateless pass-through with no WAL entry. Forwards to the scheduler via a new xce_schedadmin_req() (async send, mirrors xce_fetchfile_req) and blocks the vtask thread on a condvar with a 5s deadline while holding c->sched.lock for the round-trip, matching this spec's own research-phase design decision (parity with OPCode_upgradeConfig's no-RBAC posture, blocking wait accepted for low-frequency admin traffic). A new branch in the existing OPCode_reply case demuxes the scheduler's SchedAdminReply to a process-global single-slot pending-waiter mechanism, serialized by c->sched.lock, matching job.c's existing fetchfile-proxy pattern. Adversarial review independently traced double-free/cleanup, the timeout/signal race, and reply-opcode demux correctness against source; found and a follow-up round closed a narrow-window leak via a carefully lock-ordered unlock->unregister->relock sequence (avoiding an AB-BA inversion against schedadmin_signal()'s lock acquisition order). New test_sched_admin_dispatch.c drives the real myreq_handler via the existing myreq_handler_for_test seam: not-primary rejection (proven via a poison-pointer sched.ep that would crash if forwarding were ever attempted), not-connected rejection, and a genuine two-thread round-trip test exercising the real signal/demux path. Controller suite 3/3 (1090/1090 assertions), scheduler suite 26/26 unaffected, clang-format clean. --- .kiro/specs/scheduler-core/tasks.md | 3 +- src/controller/cluster.c | 252 +++++++++++++- src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_sched_admin_dispatch.c | 310 ++++++++++++++++++ src/include/request.h | 9 + src/lib/request.c | 24 ++ 7 files changed, 598 insertions(+), 3 deletions(-) create mode 100644 src/controller/testsuite/test_sched_admin_dispatch.c diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 1bd60a02..55307dd5 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -258,12 +258,13 @@ - [ ] 10. Integration: Controller Admin Opcode and xceadmin CLI -- [ ] 10.1 Add the scheduler admin opcode dispatch to the controller +- [x] 10.1 Add the scheduler admin opcode dispatch to the controller - Add the new opcode's dispatch case to the controller's request handler: reject with a clear error when not primary, reject with a clear error when no scheduler is connected, forward the request to the scheduler and wait up to the timeout for a reply, and always send an explicit reply on every path (never a dropped request) - The controller interprets nothing beyond the opcode itself — every field of the request/reply passes through untouched, and the controller commits no state and writes no persistence entry for this opcode - Observable: a request against a non-primary controller returns the not-primary error immediately with no forwarding attempt; a request against a primary with no scheduler connected returns the not-connected error immediately - _Requirements: 5.10_ - _Depends: 7.4, 9.4_ + - _Verified: 2026-08-05 — this is the spec's ONE task touching `src/controller/`. New `case OPCode_schedAdmin:` in `myreq_handler()` (cluster.c) mirrors the existing `OPCode_upgradeConfig` case structurally: `ha_is_primary()` gate (reply `-EAGAIN`, no forward), `sched.ep==NULL` check (reply "scheduler not connected", no forward) — both of this task's own literal Observables — then forwards via a new `xce_schedadmin_req()` (request.c, async send mirroring `xce_fetchfile_req`) and blocks the vtask thread on a condvar with a 5s deadline while holding `c->sched.lock` for the full round-trip (a design decision made and accepted during this spec's earlier research phase, matching `OPCode_upgradeConfig`'s posture at parity — no RBAC beyond the primary gate, blocking wait accepted as low-frequency human-paced admin traffic). A new branch in the existing `case OPCode_reply:` demuxes the scheduler's `SchedAdminReply` (by `opcode == OPCode_schedAdmin`) to a process-global single-slot pending-waiter mechanism (`schedadmin_register`/`_unregister`/`_signal`), serialized by `c->sched.lock` being held throughout — the same process-global-list pattern this codebase already uses for `job.c`'s fetchfile proxy. Adversarial review independently traced double-free/cleanup-per-path, the timeout/signal race, single-slot safety, and reply-opcode demux correctness against source (not prior sessions' claims) and found no correctness bugs; confirmed the `c->sched.lock` contention is broader than initially characterized (gates job/worker/resource/queue events, not just job-new/start) but is spec-mandated by requirements.md's own pseudocode, bounded to 5s, and does not risk the severe heartbeat-starvation/false-failover cascade (heartbeat send runs on a separate non-vtask delaymgr task) — flagged as a documented residual risk for a future hardening pass, not a blocker. Review also found and a follow-up fix-round closed a narrow-window memory leak (a reply arriving in the brief gap between the waiter's mutex unlock and its unregister call could hand off ownership into an already-abandoned slot) via a carefully-ordered unlock→unregister→relock sequence, with lock-ordering reasoning (avoiding an AB-BA inversion against `schedadmin_signal()`'s `g_schedadmin_mu`→`p->mu` acquisition order) independently re-derived and confirmed correct by me directly from the code before finalizing. Also cleaned up: two prior sessions' work had whole-file clang-format reformatted `cluster.c`/`request.c`/`request.h`, churning large amounts of unrelated pre-existing HA code through pure whitespace changes — minimized to the genuinely new content only (confirmed via `git diff` matching `git diff -w` exactly for all three files) to avoid polluting this HA-critical file's history. New `test_sched_admin_dispatch.c` (3 tests) drives the real `myreq_handler` via the existing `myreq_handler_for_test` seam (same precedent `test_config_repl.c` uses): not-primary rejection using a poison-pointer `sched.ep` that would crash if forwarding were ever incorrectly attempted (proven, not assumed), not-connected rejection, and a genuine round-trip test with a second pthread independently driving a real `OPCode_reply`/`SchedAdminReply` through the same handler to exercise the actual signal/demux path end-to-end. Controller suite 3/3 (1090/1090 Unity assertions including the 3 new tests), scheduler suite 26/26 unaffected, clang-format clean — independently re-verified by me via fresh `meson compile`/`meson test --suite controller`/`meson test --suite scheduler` runs before marking complete. Residual, non-blocking notes for follow-up: the `c->sched.lock` contention breadth (above) and the process-global (vs. `cluster_t`-scoped) pending-waiter slot are both flagged as candidates for a future hardening pass, not required for this task._ - [ ] 10.2 Controller-side admin opcode dispatch tests - Test standby rejection, disconnected-scheduler rejection, forward timeout, and a successful forward/reply round trip against the real scheduler-side handler from 9.4 diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 11466206..a63b4307 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -2,9 +2,11 @@ #include #include +#include #include #include #include +#include #include #include "cluster.h" @@ -235,6 +237,78 @@ static int handle_snapshot_transfer(cluster_t *c, xce_request_t *req) return 0; } +/* ---- OPCode_schedAdmin: forward-and-block-for-reply support (task 10.1) -- + * + * The controller is a stateless pass-through for OPCode_schedAdmin: it + * forwards a SchedAdminRequest to the scheduler over c->sched.ep and must + * relay the scheduler's SchedAdminReply back to the original caller, which + * means the request-handling vtask thread has to block until that reply + * arrives (or 5s elapses) -- unlike every other controller->scheduler call + * in this file (sched.c's xce_push_*_req / xce_pushconfig_req), which are + * fire-and-forget. + * + * This mirrors on_fetchfile's proxy pattern (job.c): register a waiter + * BEFORE sending (avoids a race with a reply that arrives before the wait + * starts), send async via the normal xce_request() path, then block on a + * condvar with a deadline. The one difference from fetchfile's keyed + * (jobid) multi-waiter list: this needs only a SINGLE slot, because the + * schedAdmin case below holds c->sched.lock for the entire forward+wait + * duration, which already serializes at most one in-flight schedAdmin + * round-trip per controller at a time (an accepted trade-off -- xceadmin + * admin commands are low-frequency, human-paced traffic, not a hot path; + * see design research for task 10.1). */ +typedef struct { + pthread_mutex_t mu; + pthread_cond_t cv; + int done; + int status; + sched_admin_reply_t* reply; +} schedadmin_pending_t; + +static pthread_mutex_t g_schedadmin_mu = PTHREAD_MUTEX_INITIALIZER; +static schedadmin_pending_t* g_schedadmin_pending = NULL; + +static void schedadmin_register(schedadmin_pending_t* p) +{ + pthread_mutex_lock(&g_schedadmin_mu); + g_schedadmin_pending = p; + pthread_mutex_unlock(&g_schedadmin_mu); +} + +static void schedadmin_unregister(schedadmin_pending_t* p) +{ + pthread_mutex_lock(&g_schedadmin_mu); + if (g_schedadmin_pending == p) + g_schedadmin_pending = NULL; + pthread_mutex_unlock(&g_schedadmin_mu); +} + +/* Signal the pending schedAdmin waiter (if any) with the scheduler's reply. + * Called from the OPCode_reply case below when a SchedAdminReply arrives on + * c->sched.ep. Returns 1 if a waiter was found and signaled (ownership of + * `sar` transfers to it -- the caller must NULL its own copy so the normal + * xce_free_request path does not also free it), 0 if no waiter was + * registered (e.g. a late reply arriving after this controller already gave + * up and replied "request timeout" to xceadmin -- `sar` stays owned by the + * caller in that case, freed normally). */ +static int schedadmin_signal(int status, sched_admin_reply_t* sar) +{ + pthread_mutex_lock(&g_schedadmin_mu); + schedadmin_pending_t* p = g_schedadmin_pending; + if (p == NULL) { + pthread_mutex_unlock(&g_schedadmin_mu); + return 0; + } + pthread_mutex_lock(&p->mu); + p->status = status; + p->reply = sar; + p->done = 1; + pthread_cond_signal(&p->cv); + pthread_mutex_unlock(&p->mu); + pthread_mutex_unlock(&g_schedadmin_mu); + return 1; +} + static int myreq_handler(vtask_t* task) { vtask_data_t* data = (vtask_data_t*)task->data; @@ -714,8 +788,162 @@ static int myreq_handler(vtask_t* task) } break; #endif + case OPCode_schedAdmin: { + /* Stateless pass-through forward to the scheduler (task 10.1, Req + * 5.10). The controller commits no WAL entry and interprets no + * field beyond the opcode -- every field of the request/reply + * passes through untouched. Structurally mirrors OPCode_upgradeConfig + * above: primary-only gate, explicit reply on every path, never a + * dropped request. See the schedadmin_* helpers above this function + * for the forward-and-block-for-reply mechanism (mirrors + * on_fetchfile's proxy pattern in job.c). */ + sched_admin_req_t* areq = request.body.schedadminmsg; + + /* Defensive belt-and-suspenders guard -- mirrors OPCode_upgradeConfig's + * identical inner check above. The real gate is ha_dispatch_gate + * (called before this switch, ~line 276): OPCode_schedAdmin is + * absent from the CTRL_STANDBY / CTRL_INIT / CTRL_ELECTING + * allow-lists in ha.c -- exactly like OPCode_upgradeConfig -- so a + * non-primary controller is rejected with no forwarding attempt + * before this switch is ever reached; this check only guards a + * role change racing between that gate and here. */ + if (!ha_is_primary(&c->ha)) { + reply.status = -EAGAIN; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); + rc = -EAGAIN; + break; + } + + if (areq == NULL) { + reply.status = -EINVAL; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); + rc = -EINVAL; + break; + } + + /* c->sched.lock is held for the FULL forward+wait duration below -- + * an accepted trade-off (task 10.1 design notes): xceadmin admin + * commands are low-frequency, human-paced traffic, not a hot path. + * This also naturally serializes at most one in-flight schedAdmin + * round-trip at a time, which the single-slot pending waiter above + * relies on. */ + mylock_lock(&(c->sched.lock)); + + if (c->sched.ep == NULL) { + sched_admin_reply_t nc_reply; + memset(&nc_reply, 0, sizeof(nc_reply)); + nc_reply.error = "scheduler not connected"; + + mylock_unlock(&(c->sched.lock)); + + reply.status = 0; + reply.data_which = RequestReply_data_schedadminreplymsg; + reply.data.schedadminreplymsg = &nc_reply; + xce_reply(ep, c->myhost, &reply); + rc = 0; + break; + } + + schedadmin_pending_t pending = { .mu = PTHREAD_MUTEX_INITIALIZER, + .cv = PTHREAD_COND_INITIALIZER, + .done = 0, + .status = -ETIMEDOUT, + .reply = NULL }; + + /* Register before sending -- avoid a race with a reply that arrives + * before pthread_cond_timedwait below starts waiting (same + * discipline as fetchfile_register, job.c). */ + schedadmin_register(&pending); + rc = xce_schedadmin_req(c->sched.ep, areq); + + if (rc < 0) { + schedadmin_unregister(&pending); + mylock_unlock(&(c->sched.lock)); + + reply.status = rc; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); + break; + } + + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += 5; + + pthread_mutex_lock(&pending.mu); + while (!pending.done) { + int wrc + = pthread_cond_timedwait(&pending.cv, &pending.mu, &deadline); + if (wrc == ETIMEDOUT) + break; + } + pthread_mutex_unlock(&pending.mu); + + /* Unregister immediately after releasing pending.mu, BEFORE reading + * the outcome below -- closes the window where a late-arriving + * reply could still find this slot registered and hand off `sar` + * ownership into it after we have already stopped looking (the + * abandoned-slot leak this fix addresses). + * + * This must NOT be done while still holding pending.mu: + * schedadmin_unregister() only takes g_schedadmin_mu (confirmed by + * its body above -- it never touches p->mu), but schedadmin_signal() + * takes g_schedadmin_mu THEN p->mu. Calling unregister while holding + * pending.mu would acquire the two locks in the reverse order + * (pending.mu then g_schedadmin_mu) from a concurrent + * schedadmin_signal() (g_schedadmin_mu then pending.mu) -- a classic + * AB-BA lock-ordering inversion that can deadlock. Releasing + * pending.mu first keeps both threads' lock order consistent + * (g_schedadmin_mu is always acquired without holding pending.mu). + * + * Once schedadmin_unregister() returns, no NEW signal can find this + * slot. Re-acquiring pending.mu below picks up the result of any + * signal that was already in flight (had read g_schedadmin_pending + * before we cleared it) and finished its write before we got here -- + * mutual exclusion on g_schedadmin_mu guarantees that race resolves + * one way or the other before schedadmin_unregister() returns. */ + schedadmin_unregister(&pending); + + pthread_mutex_lock(&pending.mu); + int got_done = pending.done; + int got_status = pending.status; + sched_admin_reply_t* got_reply = pending.reply; + pthread_mutex_unlock(&pending.mu); + + if (got_done && got_reply != NULL) { + /* Relay the scheduler's SchedAdminReply back to xceadmin + * verbatim -- every field passes through untouched. */ + reply.status = 0; + reply.data_which = RequestReply_data_schedadminreplymsg; + reply.data.schedadminreplymsg = got_reply; + xce_reply(ep, c->myhost, &reply); + free_SchedAdminReply_ptr(&got_reply); + } else if (got_done) { + /* Scheduler replied at the RPC level with no SchedAdminReply + * payload (unexpected, but handled loudly, never dropped) -- + * relay its RPC status. */ + reply.status = got_status; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); + } else { + sched_admin_reply_t to_reply; + memset(&to_reply, 0, sizeof(to_reply)); + to_reply.error = "request timeout"; + + reply.status = 0; + reply.data_which = RequestReply_data_schedadminreplymsg; + reply.data.schedadminreplymsg = &to_reply; + xce_reply(ep, c->myhost, &reply); + } + + mylock_unlock(&(c->sched.lock)); + rc = 0; + break; + } case OPCode_reply: { - /* Reply arriving on an inbound endpoint. Two independent producers + /* Reply arriving on an inbound endpoint. Three independent producers * feed this case: * 1. fetchfile proxy replies from a worker (fetchfile feature) — * routed to the pending waiter via fetchfile_signal. @@ -726,7 +954,10 @@ static int myreq_handler(vtask_t* task) * replicator so the configAck barrier can observe it * (ha-commit-barrier-rollout task 5.2 remediation r1; same * ingestion as repl_ack_decode_and_dispatch in ha.c — completes - * the unlanded half of ha-config-bootstrap task 8.1). */ + * the unlanded half of ha-config-bootstrap task 8.1). + * 3. SchedAdminReply arriving on c->sched.ep in response to the + * OPCode_schedAdmin case's forwarded request — routed to the + * pending waiter via schedadmin_signal (task 10.1). */ xce_reply_t* workreply = request.body.replymsg; if (workreply != NULL && workreply->opcode == OPCode_fetchfile) { @@ -742,6 +973,23 @@ static int myreq_handler(vtask_t* task) break; } + if (workreply != NULL && workreply->opcode == OPCode_schedAdmin) { + sched_admin_reply_t* sar = (workreply->data_which + == RequestReply_data_schedadminreplymsg) + ? workreply->data.schedadminreplymsg + : NULL; + if (schedadmin_signal(workreply->status, sar)) { + /* Ownership of sar (if any) transferred to the pending + * waiter — NULL it so xce_free_request below does not also + * free it. If no waiter was found (e.g. this reply arrived + * after the forwarder already gave up at the 5s timeout), + * leave it owned by workreply for the normal free path. */ + workreply->data.schedadminreplymsg = NULL; + } + rc = 0; + break; + } + if (request.body_which == Request_body_replymsg && workreply != NULL && workreply->data_which diff --git a/src/controller/meson.build b/src/controller/meson.build index c2e73746..3783ac80 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -178,6 +178,7 @@ controller_unit_test_src = [ 'testsuite/test_upgrade_snapshot_restore.c', 'testsuite/test_upgrade_snapshot_apply.c', 'testsuite/test_bug857_regression.c', + 'testsuite/test_sched_admin_dispatch.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 24212b80..1b29a38f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -93,6 +93,7 @@ extern void test_job_array_replay_fence(void); extern void test_upgrade_snapshot_restore(void); extern void test_upgrade_snapshot_apply(void); extern void test_bug857_regression(void); +extern void test_sched_admin_dispatch(void); void setUp(void) { return; } @@ -255,6 +256,7 @@ int main(int argc, char* argv[]) TEST(test_upgrade_snapshot_restore); TEST(test_upgrade_snapshot_apply); TEST(test_bug857_regression); + TEST(test_sched_admin_dispatch); return UNITY_END(); } diff --git a/src/controller/testsuite/test_sched_admin_dispatch.c b/src/controller/testsuite/test_sched_admin_dispatch.c new file mode 100644 index 00000000..93cb0078 --- /dev/null +++ b/src/controller/testsuite/test_sched_admin_dispatch.c @@ -0,0 +1,310 @@ +/* + * 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. + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include + +#include + +#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); + +/* ── 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); +} + +/* ── 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"); +} + +/* ── 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"); +} + +/* ── 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"); +} + +/* ── Suite entry point ──────────────────────────────────────────────────── */ +void test_sched_admin_dispatch(void) +{ + UnitySetTestFile(__FILE__); + RUN_TEST(test_schedadmin_rejects_when_not_primary); + RUN_TEST(test_schedadmin_rejects_when_not_connected); + RUN_TEST(test_schedadmin_round_trip_success); +} diff --git a/src/include/request.h b/src/include/request.h index d2f4ef02..7d4fade6 100644 --- a/src/include/request.h +++ b/src/include/request.h @@ -71,6 +71,15 @@ extern int xce_set_pend_reason_req(endpoint_t*, xce_preason_set_t*); extern int xce_clear_reason_req(endpoint_t*, xce_clear_reason_t*); extern int xce_fetchfile_req(endpoint_t*, xce_fetchfile_t*); +/* + * xce_schedadmin_req: send OPCode_schedAdmin (SchedAdminRequest) to `ep` -- + * the async send half of task 10.1's controller->scheduler forward. Mirrors + * xce_fetchfile_req above exactly. The caller (myreq_handler, cluster.c) + * blocks separately for the matching SchedAdminReply; this function only + * enqueues the request, it does not wait. + */ +extern int xce_schedadmin_req(endpoint_t*, sched_admin_req_t*); + /* * xce_replconfigpush_req_raw: send OP_replConfigPush carrying raw byte arrays. * Bypasses the encode_ReplConfigPushRequest_ptr path (which maps sha256/bytes diff --git a/src/lib/request.c b/src/lib/request.c index 0036f34c..3c22716d 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -2206,6 +2206,30 @@ int xce_fetchfile_req(endpoint_t* ep, xce_fetchfile_t* msg) return xce_request(ep, &request); } +/* xce_schedadmin_req: send OPCode_schedAdmin (SchedAdminRequest) to `ep`. + * Async send only, mirrors xce_fetchfile_req above -- the controller + * (myreq_handler, cluster.c) forwards xceadmin's SchedAdminRequest to the + * scheduler via this function, then blocks separately for the matching + * SchedAdminReply (scheduler-core task 10.1, Req 5.10). */ +int xce_schedadmin_req(endpoint_t* ep, sched_admin_req_t* msg) +{ + /* clang-format off */ + xce_request_t request = { + .opcode = OPCode_schedAdmin, + .body_which = Request_body_schedadminmsg, + .body = { + .schedadminmsg = msg + } + }; + /* clang-format on */ + + if ((ep == NULL) || (msg == NULL)) { + return -EINVAL; + } + + return xce_request(ep, &request); +} + /* ── Client-Side Transparent Retry tunables (Req 6.1, 6.2) ───────────────── * * XCE_RETRY_DEFAULTS provides safe production defaults for the failover retry