myxce/src/scheduler/testsuite/test_handlers.c
Rongsong Shen a9feded88e
fix(scheduler): mid-round rejoin config push truncated live rounds (bug-1846 part 2)
sched_publish_retry_loop() calls sched_check_pending_rejoin() after every
successful publish -- inside the driver's decision-drain loop -- so after a
client-library reconnect the scheduler re-joins mid-round. The controller
answers every join (on_scheduler_join(), controller/sched.c) by re-pushing
config, workers and jobs. handler.c applied the config push synchronously
on the handler thread: sched_init_config() -> sched_init_queues() ->
builtin_fcfs on_queue_update(), which nulls the picker cursor when it
reinserts the queue the cursor rests on. next_job() then returned NULL and
the round ended with the rest of the queue never considered.

Evidence (scheduler-replay-20260923-085356.replay, temporary event-stream
diagnostics, reverted): 9 recorded rounds end after 11-12 decisions; every
one is followed by a workersPush + full 50-job jobsPush (the join's resync,
reporting just-dispatched jobs back as PEND), every full-length dispatching
round has none, and the config push never appears because it bypassed the
pending-update queue the recorder writes from -- which is also why replay
could not reproduce the truncation.

Fix (defer to the round edge, same direction as bug-1813): handler.c now
calls new sched_stash_config(), parking the Conf4Sched in new
xce_scheduler_t.pending_config under pending_config_lock (an older
unapplied push is freed). main.c's loop calls new
sched_apply_pending_config() before each round, running the unchanged
sched_init_config() strictly between sessions. sched_fini() frees a
never-applied push.

Tests:
- test_handlers.c: test_config_push_is_stashed_not_applied_on_handler_thread
  -- RED (sched_init_config called on the handler thread), GREEN after.
- new test_config_push_deferred.c (sched_config_push_deferred_test): real
  scheduler_req_handler() + real builtin_fcfs mid-session -- RED (next_job()
  returned NULL right after the push), GREEN (session walks the remaining
  jobs; configver unchanged until applied once between rounds).
- test_admin_request.c / test_job_array_expand.c / test_job_array_wire_e2e.c
  #include handler.c directly and get a link stub for sched_stash_config().
Full scheduler suite 87/87 (asan).

Not addressed: config pushes are still not written to a recording (a
replay spanning a config change won't see it); the ~100s reconnect churn
itself (see bug-1812/1813).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-24 06:30:34 +00:00

384 lines
14 KiB
C

/*
* test_handlers.c - white-box tests for the scheduler notify/dataset enqueue
* handlers (bug-765 G-6, tasks 9.2 + 9.3). Pulls the real (static) handlers in
* handler.c into this TU; enq_pend_update is local to handler.c, so we drive a
* real sched->pend_updates dataq and assert the correct op was enqueued (or
* skipped / -EINVAL) per each handler's contract.
*/
#define UNIT_TEST 1
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "dataq.h"
#include "mylock.h"
#include "scheduler.h"
/* Pull the real handlers under test into this TU. handler.c is self-contained
* for the enqueue path: enq_pend_update and sched_job_state_change are both
* static within it; the dispatchers call lib helpers (linked). The
* scheduler-internal externs handler.c references but this file does not
* exercise are stubbed below: sched_init_config (config-init request path)
* and sched_get_job (job-array-single-wal-record task 5.3's per-element
* dedup check, XceData_data_jobArray — not tested here, see
* test_job_array_expand.c for the real kvmgr-backed sched_get_job). */
static int g_init_config_calls = 0;
static int g_stash_config_calls = 0;
static xce_conf4sched_t* g_stashed_config = NULL;
int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
{
(void)s;
(void)c;
g_init_config_calls++;
return 0;
}
/* bug-1846: the config-push handler must hand the pushed config to the
* driver (applied between rounds) rather than calling sched_init_config()
* on the handler thread -- counted here so the test below can tell which
* path the real handler took. */
int sched_stash_config(xce_scheduler_t* s, xce_conf4sched_t* c)
{
(void)s;
g_stash_config_calls++;
g_stashed_config = c;
return 0;
}
/* sched_get_job relocated to core.h (scheduler-core spec, task 4.1) -- now
* operates on xce_sched_core_t*, not xce_scheduler_t*. */
xce_sched_job_t* sched_get_job(xce_sched_core_t* sched, uint64_t jobid)
{
(void)sched;
(void)jobid;
return NULL;
}
/* sched_record_status (recording.c, scheduler-core spec, task 9.4) --
* handler.c's OPCode_schedAdmin/recordStatus path references it; this file
* never exercises OPCode_schedAdmin (its own scope is notify/dataset
* enqueue), so a link-satisfying stub is enough, same rationale as
* sched_init_config above. */
int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply)
{
(void)sched;
(void)reply;
return -EINVAL;
}
#include "../src/handler.c"
void setUp(void) { return; }
void tearDown(void)
{
printf("\n-----------------------------------------------------\n");
return;
}
static void fixture_init(xce_scheduler_t* sched)
{
memset(sched, 0, sizeof(*sched));
mylock_init(&(sched->core.lock));
sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t));
dataq_init(sched->core.pend_updates[0], NULL);
sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
dataq_init(sched->core.pend_updates[1], NULL);
}
/* Dequeue one pending request (NULL if the queue is empty) and return its op
* via *op_out; frees the request shell (payloads are test-owned/leaked under
* the suite's leaks-off convention). Returns 1 if an item was present. */
static int pop_op(xce_scheduler_t* sched, xce_sched_op_t* op_out)
{
data_item_t* it = dataq_deq(sched->core.pend_updates[0], NULL);
xce_sched_update_request_t* req;
if (it == NULL) {
return 0;
}
req = ITPTR_TO_UPDATE_REQ(it);
if (op_out != NULL) {
*op_out = req->op;
}
free(req);
return 1;
}
/* Dequeue one pending request and assert it is an XCE_SCHED_OP_JSTATE_CHANGE,
* returning its jobid/state via jobid_out/state_out (both optional). Frees
* the request shell. Returns 1 if an item was present, 0 if the queue was
* empty (scheduler-dispatch-commit task 2.2: verifies the jobDispatchFail
* notify enqueues the same op the adjacent schedExitJob/schedFailJob cases
* already use). */
static int pop_jstate_change(xce_scheduler_t* sched, uint64_t* jobid_out,
uint32_t* state_out)
{
data_item_t* it = dataq_deq(sched->core.pend_updates[0], NULL);
xce_sched_update_request_t* req;
if (it == NULL) {
return 0;
}
req = ITPTR_TO_UPDATE_REQ(it);
TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_JSTATE_CHANGE, req->op);
if (jobid_out != NULL) {
*jobid_out = req->u.job_state_change.jobid;
}
if (state_out != NULL) {
*state_out = req->u.job_state_change.state;
}
free(req);
return 1;
}
/* ── 9.3: notify single-item handlers (NULL -> -EINVAL, valid -> enqueue) ── */
static void test_handle_update_job_enqueues_job_fields(void)
{
xce_scheduler_t sched;
xce_sched_op_t op = (xce_sched_op_t)-1;
fixture_init(&sched);
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_sched_update_job(&sched, NULL));
TEST_ASSERT_EQUAL_INT(0, pop_op(&sched, NULL)); /* NULL enqueued nothing */
xce_job_t* job = (xce_job_t*)calloc(1, sizeof(*job));
job->jobid = 42;
TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_job(&sched, job));
TEST_ASSERT_EQUAL_INT(1, pop_op(&sched, &op));
/* 5.6: notify job update takes the in-place UPDATE_JOB_FIELDS path, NOT
* the dataset bulk-create UPDATE_JOB path. */
TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_JOB_FIELDS, op);
free(job);
}
static void test_handle_update_queue_enqueues_queue(void)
{
xce_scheduler_t sched;
xce_sched_op_t op = (xce_sched_op_t)-1;
fixture_init(&sched);
TEST_ASSERT_EQUAL_INT(-EINVAL,
sched_handle_sched_update_queue(&sched, NULL));
xce_queue_t* q = (xce_queue_t*)calloc(1, sizeof(*q));
TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_queue(&sched, q));
TEST_ASSERT_EQUAL_INT(1, pop_op(&sched, &op));
TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_QUEUE, op);
free(q);
}
static void test_handle_job_rusage_enqueues_rusage(void)
{
xce_scheduler_t sched;
xce_sched_op_t op = (xce_sched_op_t)-1;
fixture_init(&sched);
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_sched_job_rusage(&sched, NULL));
xce_job_rusages_t* ru = (xce_job_rusages_t*)calloc(1, sizeof(*ru));
ru->jobid = 7;
TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_job_rusage(&sched, ru));
TEST_ASSERT_EQUAL_INT(1, pop_op(&sched, &op));
TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_JOB_RUSAGE, op);
free(ru);
}
static void test_handle_update_resvals_skips_empty_enqueues_nonempty(void)
{
xce_scheduler_t sched;
xce_sched_op_t op = (xce_sched_op_t)-1;
xce_resval_list_t empty = { .count = 0, .vals = NULL };
xce_resval_list_t full = { .count = 1, .vals = NULL };
fixture_init(&sched);
/* NULL or count==0 -> skip (return 0, nothing enqueued) */
TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, NULL));
TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, &empty));
TEST_ASSERT_EQUAL_INT(0, pop_op(&sched, NULL));
/* count>0 -> enqueue UPDATE_RESVALS */
TEST_ASSERT_EQUAL_INT(0, sched_handle_sched_update_resvals(&sched, &full));
TEST_ASSERT_EQUAL_INT(1, pop_op(&sched, &op));
TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_RESVALS, op);
}
/* ── 9.2: dataset handlers (skip on NULL/empty, enqueue on valid) ── */
static void test_handle_resources_skips_empty_enqueues_nonempty(void)
{
xce_scheduler_t sched;
xce_sched_op_t op = (xce_sched_op_t)-1;
xce_resource_t* one[1];
fixture_init(&sched);
TEST_ASSERT_EQUAL_INT(0, sched_handle_resources(&sched, 0, NULL));
TEST_ASSERT_EQUAL_INT(0, pop_op(&sched, NULL));
one[0] = (xce_resource_t*)calloc(1, sizeof(xce_resource_t));
TEST_ASSERT_EQUAL_INT(0, sched_handle_resources(&sched, 1, one));
TEST_ASSERT_EQUAL_INT(1, pop_op(&sched, &op));
TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_RESOURCES, op);
free(one[0]);
}
static void test_handle_resvals_full_skips_empty_enqueues_nonempty(void)
{
xce_scheduler_t sched;
xce_sched_op_t op = (xce_sched_op_t)-1;
xce_resval_t* one[1];
fixture_init(&sched);
TEST_ASSERT_EQUAL_INT(0, sched_handle_resvals(&sched, 0, NULL));
TEST_ASSERT_EQUAL_INT(0, pop_op(&sched, NULL));
one[0] = (xce_resval_t*)calloc(1, sizeof(xce_resval_t));
TEST_ASSERT_EQUAL_INT(0, sched_handle_resvals(&sched, 1, one));
TEST_ASSERT_EQUAL_INT(1, pop_op(&sched, &op));
TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_RESVALS_FULL, op);
free(one[0]);
}
/* ── scheduler-dispatch-commit task 2.2: sched_handle_notify's new
* jobDispatchFail case (bug-1625's actual fix) ── */
static void test_handle_notify_job_dispatch_fail_enqueues_pend(void)
{
xce_scheduler_t sched;
xce_notify_data_t notify;
job_failinfo_t info = { .jobid = 99, .stage = 0, .code = -5,
.detail = NULL };
uint64_t jobid = 0;
uint32_t state = (uint32_t)-1;
fixture_init(&sched);
memset(&notify, 0, sizeof(notify));
notify.event = NotifyData_data_jobDispatchFail;
notify.data.dispatchfail = &info;
/* Previously fell to `default: rc = -EINVAL;` and enqueued nothing --
* this is the actual observable fix for bug-1625: the notify is now
* processed instead of silently discarded. */
TEST_ASSERT_EQUAL_INT(0, sched_handle_notify(&sched, &notify));
TEST_ASSERT_EQUAL_INT(1, pop_jstate_change(&sched, &jobid, &state));
TEST_ASSERT_EQUAL_UINT64(99, jobid);
TEST_ASSERT_EQUAL_INT(JOB_STAT_PEND, state);
/* No classification logic here: the notify's meaning is exclusively
* "return this job to pending" regardless of the reported failure
* code -- verify a different `code` value still resolves to PEND. */
fixture_init(&sched);
info.jobid = 123;
info.code = 0;
jobid = 0;
state = (uint32_t)-1;
TEST_ASSERT_EQUAL_INT(0, sched_handle_notify(&sched, &notify));
TEST_ASSERT_EQUAL_INT(1, pop_jstate_change(&sched, &jobid, &state));
TEST_ASSERT_EQUAL_UINT64(123, jobid);
TEST_ASSERT_EQUAL_INT(JOB_STAT_PEND, state);
}
/* ── bug-1737: sched_handle_notify's schedExitJob case must branch on
* exit_code (mirroring job.c's own apply_job_finish_transition() decision)
* instead of unconditionally assuming every job finish is a failure ── */
static void test_handle_notify_job_exit_maps_exit_code_to_done_or_exit(void)
{
xce_scheduler_t sched;
xce_notify_data_t notify;
xce_jobfinish_t finish = { .jobid = 200, .exit_code = 0 };
uint64_t jobid = 0;
uint32_t state = (uint32_t)-1;
fixture_init(&sched);
memset(&notify, 0, sizeof(notify));
notify.event = NotifyData_data_schedExitJob;
notify.data.sched_exit_job = &finish;
/* exit_code == 0 -- clean finish -- must resolve to JOB_STAT_DONE, not
* the previous unconditional JOB_STAT_EXIT. */
TEST_ASSERT_EQUAL_INT(0, sched_handle_notify(&sched, &notify));
TEST_ASSERT_EQUAL_INT(1, pop_jstate_change(&sched, &jobid, &state));
TEST_ASSERT_EQUAL_UINT64(200, jobid);
TEST_ASSERT_EQUAL_INT(JOB_STAT_DONE, state);
/* exit_code != 0 -- a real failure -- still resolves to JOB_STAT_EXIT. */
fixture_init(&sched);
finish.jobid = 201;
finish.exit_code = 7;
jobid = 0;
state = (uint32_t)-1;
TEST_ASSERT_EQUAL_INT(0, sched_handle_notify(&sched, &notify));
TEST_ASSERT_EQUAL_INT(1, pop_jstate_change(&sched, &jobid, &state));
TEST_ASSERT_EQUAL_UINT64(201, jobid);
TEST_ASSERT_EQUAL_INT(JOB_STAT_EXIT, state);
/* NULL payload -- malformed notify -- must fail cleanly, not crash. */
fixture_init(&sched);
notify.data.sched_exit_job = NULL;
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_handle_notify(&sched, &notify));
TEST_ASSERT_EQUAL_INT(0, pop_op(&sched, NULL));
}
/* bug-1846: a config push arrives on the handler thread whenever the
* scheduler (re)joins -- including a rejoin triggered from inside a live
* round's decision publish. Applying it there (sched_init_config() ->
* sched_init_queues() -> picker on_queue_update()) reset the fcfs cursor
* mid-session and ended the round early. The handler must only stash it
* and take ownership (payload pointer nulled), never apply it. */
static void test_config_push_is_stashed_not_applied_on_handler_thread(void)
{
xce_scheduler_t sched;
xce_request_t request;
xce_configdata_t configdata;
xce_conf4sched_t* config = (xce_conf4sched_t*)calloc(1, sizeof(*config));
int rc;
fixture_init(&sched);
g_init_config_calls = 0;
g_stash_config_calls = 0;
g_stashed_config = NULL;
memset(&configdata, 0, sizeof(configdata));
configdata.data.conf4sched = config;
memset(&request, 0, sizeof(request));
request.opcode = OPCode_push;
request.body_which = Request_body_configpushmsg;
request.body.configpushmsg = &configdata;
rc = scheduler_req_handler(&request, &sched);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_init_config_calls,
"config push must not be applied synchronously on the handler "
"thread -- it can land mid-round and truncate the session");
TEST_ASSERT_EQUAL_INT(1, g_stash_config_calls);
TEST_ASSERT_EQUAL_PTR(config, g_stashed_config);
TEST_ASSERT_NULL_MESSAGE(configdata.data.conf4sched,
"handler must take ownership of the stashed "
"config so the request teardown won't free it");
free(config);
}
int main(int argc, char* argv[])
{
(void)argc;
(void)argv;
UNITY_BEGIN();
RUN_TEST(test_handle_update_job_enqueues_job_fields, 0);
RUN_TEST(test_handle_update_queue_enqueues_queue, 0);
RUN_TEST(test_handle_job_rusage_enqueues_rusage, 0);
RUN_TEST(test_handle_update_resvals_skips_empty_enqueues_nonempty, 0);
RUN_TEST(test_handle_resources_skips_empty_enqueues_nonempty, 0);
RUN_TEST(test_handle_resvals_full_skips_empty_enqueues_nonempty, 0);
RUN_TEST(test_handle_notify_job_dispatch_fail_enqueues_pend, 0);
RUN_TEST(test_handle_notify_job_exit_maps_exit_code_to_done_or_exit, 0);
RUN_TEST(test_config_push_is_stashed_not_applied_on_handler_thread, 0);
return UNITY_END();
}