wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

test(scheduler): G-8 dequeue routing — UPDATE_JOB_FIELDS → in-place, bulk → no damage

Browse source 
Closes the msg-gap G-8 test gap the research flagged ("handler→dequeue ROUTING
is verified by code review, NOT an automated integration test"). The
remediation (UPDATE_JOB_FIELDS op routed to sched_update_existing_job) is
already in the code (93124c5); these tests lock the routing:

- test_update_fields_routes_to_inplace_path: driving the real sched_update()
  dequeue loop, an UPDATE_JOB_FIELDS request reaches sched_update_existing_job
  and mutates the existing job in place (same object, runtime state kept).
- test_bulk_update_existing_job_no_damage: a re-pushed dataset row for an
  existing jobid is EEXIST-dropped by key_add (via the real sched_add_job) —
  the live job keeps identity/state/runtime state.

10/10 sched_redispatch_test PASS; full scheduler suite 6/6 OK. Live cluster:
job PEND→RUN→DONE via notify updates, no WARN, no duplicate instance.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-08-14 02:08:49 +00:00
parent 6f520a677b
commit 200303d8fe
1 changed files with 93 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

94
src/scheduler/testsuite/test_redispatch.c
View file

@ -24,7 +24,9 @@
#include <unity/unity.h>
#include "dataq.h"
#include "kvmgr.h"
#include "mylock.h"
#include "scheduler.h"
/* Pull the real implementation under test. */
@ -74,9 +76,22 @@ int sched_apply_job_rusage(xce_scheduler_t* s, xce_job_rusages_t* r)
{
(void)s; (void)r; return -ENOENT;
}
/* Recorded for the routing test below; mimics the real schedobj.c semantics
* (in-place state update of the existing job, -ENOENT when unknown). */
static xce_job_t* g_last_update_fields_job = NULL;
int sched_update_existing_job(xce_scheduler_t* s, xce_job_t* j)
{
(void)s; (void)j; return -ENOENT;
xce_sched_job_t* job;
if (j == NULL) {
return -EINVAL;
}
g_last_update_fields_job = j;
job = sched_get_job(s, j->jobid);
if (job == NULL) {
return -ENOENT;
}
job->state = j->state;
return 0;
}
int sched_update_existing_queue(xce_scheduler_t* s, xce_queue_t* q)
{
@ -129,6 +144,13 @@ static void fixture_init(xce_scheduler_t* sched)
xstring_table_init(1024 * 16);
sched->jobs = makeInstance(&xkvmgr_type, &jobs_params);
/* pend_updates double-buffer + lock for driving sched_update() */
mylock_init(&(sched->lock));
sched->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t));
dataq_init(sched->pend_updates[0], NULL);
sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
dataq_init(sched->pend_updates[1], NULL);
}
static xce_sched_job_t* seed_job(xce_scheduler_t* sched, uint64_t jobid,
@ -224,6 +246,74 @@ static void test_dispatch_commit_null_alloc_unchanged(void)
TEST_ASSERT_EQUAL_UINT(JOB_STAT_PEND, job->state);
}
/* ── G-8 dequeue routing (bug-765 remediation; research.md follow-up) ────── */
/* The notify single-job update path: the handler enqueues
* XCE_SCHED_OP_UPDATE_JOB_FIELDS, and sched_update() must dispatch it to
* sched_update_existing_job() (in-place), NOT to the bulk CREATE path
* (sched_update_jobs → make_sched_job → key_add EEXIST-drop). We drive the
* real dequeue loop with a call-recording stub for sched_update_existing_job
* and assert it is reached and the existing job is mutated in place. */
static void test_update_fields_routes_to_inplace_path(void)
{
xce_scheduler_t sched;
fixture_init(&sched);
xce_sched_job_t* job = seed_job(&sched, 42, JOB_STAT_PEND);
xce_sched_obj_t* so = (xce_sched_obj_t*)calloc(1, sizeof(*so));
job->schedobj = so; /* live runtime state must survive */
g_last_update_fields_job = NULL;
xce_sched_update_request_t* req =
(xce_sched_update_request_t*)calloc(1, sizeof(*req));
xce_job_t* upd = (xce_job_t*)calloc(1, sizeof(*upd));
upd->jobid = 42;
upd->state = JOB_STAT_RUN;
req->op = XCE_SCHED_OP_UPDATE_JOB_FIELDS;
req->u.jobfieldsupdate.job = upd;
dataq_enq(sched.pend_updates[0], &(req->link));
TEST_ASSERT_EQUAL_INT(0, sched_update(&sched));
/* routed to sched_update_existing_job (not the bulk CREATE path) */
TEST_ASSERT_EQUAL_PTR(upd, g_last_update_fields_job);
/* existing job mutated in place, runtime state kept */
TEST_ASSERT_EQUAL_PTR(job, sched_get_job(&sched, 42));
TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state);
TEST_ASSERT_EQUAL_PTR(so, job->schedobj);
}
/* The bulk dataset path (XCE_SCHED_OP_UPDATE_JOB) routes to sched_update_jobs
* (CREATE) → sched_add_job → key_add. For an ALREADY-existing jobid, key_add
* must EEXIST-drop the fresh object rather than replacing the live one — the
* existing job keeps its identity and runtime state (no corruption from a
* re-pushed dataset row). We drive the REAL sched_add_job (the exact call the
* bulk path makes after make_sched_job) with a fresh duplicate object. */
static void test_bulk_update_existing_job_no_damage(void)
{
xce_scheduler_t sched;
fixture_init(&sched);
xce_sched_job_t* job = seed_job(&sched, 7, JOB_STAT_RUN);
xce_sched_obj_t* so = (xce_sched_obj_t*)calloc(1, sizeof(*so));
job->schedobj = so;
/* A fresh object for the same jobid carrying STALE state — what
* make_sched_job would build from a re-pushed dataset row. */
xce_sched_job_t* fresh = (xce_sched_job_t*)calloc(1, sizeof(*fresh));
fresh->jobid = 7;
fresh->state = JOB_STAT_PEND;
fresh->schedobj = (xce_sched_obj_t*)calloc(1, sizeof(xce_sched_obj_t));
int rc = sched_add_job(&sched, fresh);
TEST_ASSERT(rc != 0); /* key_add EEXIST — the stale duplicate is dropped */
/* the live job is untouched: identity, state, runtime state all preserved */
TEST_ASSERT_EQUAL_PTR(job, sched_get_job(&sched, 7));
TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state);
TEST_ASSERT_EQUAL_PTR(so, job->schedobj);
}
/* ── Runner ──────────────────────────────────────────────────────────────── */
int main(int argc, char* argv[])
@ -238,5 +328,7 @@ int main(int argc, char* argv[])
RUN_TEST(test_ready_suspended_not_ready, 0);
RUN_TEST(test_dispatch_commit_marks_job_run, 0);
RUN_TEST(test_dispatch_commit_null_alloc_unchanged, 0);
RUN_TEST(test_update_fields_routes_to_inplace_path, 0);
RUN_TEST(test_bulk_update_existing_job_no_damage, 0);
return UNITY_END();
}