feat(scheduler-core): define new event payload types (task 7.2)

Adds JobStateChangeEvent (jobid/oldState/newState/exitCode) and
JobUpdateEvent (jobid/changedFields/job: Job) to controller.capnp,
matching design.md's spec verbatim. JobUpdateEvent.job reuses the
existing full 12-field Job/xce_job_t record, verified distinct from
the narrower SchedJob struct sharing the same mapname.

Extends ReplayEvent.body's union (task 7.1) with jobStateChange@10/
jobUpdate@11 -- the only two valid next ordinals, since capnp requires
strictly sequential ordinals with no gaps across a struct and task
7.1 deliberately stopped at @9. Existing @0-@9 fields and every other
struct in the schema are byte-identical.

Deep round-trip tests in test_e2e_capnp.c exercise every field with
non-default values, including the 3-level nested Job substructs
(submitData/allocs/pidinfo/pendreason). JOB_F_* bitmask bit values
deliberately left undefined -- a later task's job, not this
schema-only one's.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1
This commit is contained in:
Rongsong Shen 2026-08-04 10:17:57 +00:00
parent 68a6196517
commit 755230705c
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
4 changed files with 295 additions and 4 deletions

View file

@ -165,12 +165,13 @@
- _Verified: 2026-08-04 — ReplayCluster/ReplayHeader/SnapshotEntry/ReplayEvent/SessionSummary/ReplaySession added fully; SnapshotData (empty {}) and ReplayDecision ({ts only}) added as compiler-mandated placeholders task 7.3 fills in (real fields for SnapshotData wholesale-replaced, a body union appended to ReplayDecision at new ordinals) since capnp's $C.codecgen has no per-struct opt-out from requiring the mapname'd C type to exist. ReplayEvent's union deliberately holds only the 9 members with pre-existing backing types (configPush/jobsPush/workersPush/resourcesPush/resvalsFull/resvalsDelta/queueUpdate/jobRusage/endSession); jobStateChange/jobUpdate deferred to task 7.2 as new ordinals — proven compiler-mandatory (not stylistic) by review, which independently confirmed capnp rejects both the AC 20 pseudocode's literal union syntax and any ordinal gap. Round-trip test (test_e2e_capnp.c) proven real via full clean rebuild (deleted+regenerated controller.capnp.{c,h}) both by implementer and reviewer. Full project build clean (controller/apiserver-CGO/scheduler/worker/xceadmin), 1075/1075 controller_unit_test passing, review APPROVED first pass with reviewer going beyond implementer's own verification (real capn_resolve()-omission and placeholder-removal breakage reproduced firsthand). Zero existing structs/fields touched._
- _Depends: 6_
- [ ] 7.2 Define new event payload types
- [x] 7.2 Define new event payload types
- Add the two genuinely new event payload message types (job state-change delta, job update delta with embedded full record) to the wire schema; every other event payload type reuses an existing wire message already used elsewhere in the schema
- Observable: encoding a job state-change event and a job update event and decoding them back yields field-for-field equality with the original values
- _Requirements: 5.20_
- _Depends: 7.1_
- _Boundary: ReplayFormat — not (P): 7.2/7.3/7.4 all edit the same schema file (controller.capnp); run sequentially to avoid concurrent edits to one file, even though each adds a logically disjoint set of structs_
- _Verified: 2026-08-04 — JobStateChangeEvent/JobUpdateEvent added to controller.capnp matching design.md's spec verbatim; JobUpdateEvent.job correctly reuses the full 12-field Job/xce_job_t record (not the narrower SchedJob sharing the same mapname — verified by field list). ReplayEvent.body's union extended with jobStateChange@10/jobUpdate@11, the only two valid next ordinals per task 7.1's baseline (capnp requires no gaps) — confirmed via a from-scratch capnp regeneration. Existing @0-@9 fields byte-identical. Deep round-trip tests (non-default values throughout, including 3-level nested Job substructs) prove the Observable; JOB_F_* bitmask values deliberately left undefined (later task's job). 1077/1077 controller_unit_test, full project build clean, review APPROVED first pass._
- [ ] 7.3 Define decision payload types and the snapshot schema
- Add the dispatch, pend, and preempt decision payload message types, reusing existing allocation and pend-reason wire messages where they already carry the needed fields

View file

@ -545,6 +545,248 @@ static void test_session_summary_round_trip(void)
printf("test_session_summary_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 9: JobStateChangeEvent round-trip (scheduler-core task 7.2, */
/* Req 5 AC 20) -- the first of the two genuinely new ReplayEvent.body */
/* payload types. Uses the encode_X/decode_X codec bridge (Test 6's */
/* pattern) since job_state_change_event_t is a real, fully-scalar C */
/* struct with no Data fields -- unlike the frame-only replay-envelope */
/* types in Tests 7-8, which have no live C struct yet. */
/* ------------------------------------------------------------------ */
static void test_job_state_change_event_round_trip(void)
{
job_state_change_event_t src;
memset(&src, 0, sizeof(src));
src.jobid = 42424ULL;
src.old_state = 2u;
src.new_state = 8u;
src.exit_code = -7;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
JobStateChangeEvent_ptr p;
encode_JobStateChangeEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
int64_t sz = capn_size(&c);
TEST_ASSERT_TRUE(sz > 0);
uint8_t *buf = (uint8_t *)malloc((size_t)sz);
TEST_ASSERT_NOT_NULL(buf);
capn_write_mem(&c, buf, (size_t)sz, 0);
capn_free(&c);
struct capn d;
TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0));
JobStateChangeEvent_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct JobStateChangeEvent wire;
read_JobStateChangeEvent(&wire, dp);
job_state_change_event_t dst;
memset(&dst, 0, sizeof(dst));
decode_JobStateChangeEvent(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(42424ULL, dst.jobid);
TEST_ASSERT_EQUAL_UINT32(2u, dst.old_state);
TEST_ASSERT_EQUAL_UINT32(8u, dst.new_state);
TEST_ASSERT_EQUAL_INT32(-7, dst.exit_code);
capn_free(&d);
free(buf);
printf("test_job_state_change_event_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Test 10: JobUpdateEvent round-trip (scheduler-core task 7.2, */
/* Req 5 AC 20) -- the second genuinely new ReplayEvent.body payload */
/* type. `job` embeds the FULL updated record by reusing the */
/* pre-existing Job/xce_job_t struct; populated non-trivially, including*/
/* nested submit/allocs/pidinfo/pend_reasons (incl. the HostAlloc/ */
/* ResAlloc list-of-pointers two levels down), to prove the whole tree */
/* round-trips through encode_Job_ptr/decode_Job_ptr -- the same nested-*/
/* pointer codec chain already proven for xce_job_submit_t->submit in */
/* Test 6, now exercised for Job's full field set. */
/* ------------------------------------------------------------------ */
static void test_job_update_event_round_trip(void)
{
/* Innermost: one HostAlloc with one ResAlloc. */
xce_alloc_t res_alloc;
memset(&res_alloc, 0, sizeof(res_alloc));
res_alloc.resname = "cpu";
res_alloc.reserve = "resv1";
res_alloc.instid = 42ULL;
res_alloc.value = 4.0;
res_alloc.from = 500ULL;
res_alloc.duration = 3600LL;
xce_alloc_t *res_allocs[1] = { &res_alloc };
xce_host_alloc_t host_alloc;
memset(&host_alloc, 0, sizeof(host_alloc));
host_alloc.host = "hostA";
host_alloc.nallocs = 1;
host_alloc.allocs = res_allocs;
xce_host_alloc_t *host_allocs[1] = { &host_alloc };
xce_job_alloc_t job_alloc;
memset(&job_alloc, 0, sizeof(job_alloc));
job_alloc.jobid = 8001ULL;
job_alloc.nhosts = 1;
job_alloc.allocs = host_allocs;
job_alloc.nres_shared = 0;
job_alloc.allocs_shared = NULL;
job_submit_data_t submit;
memset(&submit, 0, sizeof(submit));
submit.queue = "normal";
submit.user = "alice";
submit.mintasks = 2u;
submit.maxtasks = 8u;
submit.start_info = NULL;
submit.job_name = "update-test-job";
submit.project_name = "proj-x";
submit.user_group = "grp-x";
submit.job_description = "job update event round-trip test";
submit.resreq = "select=2";
unsigned int pids[2] = { 111u, 222u };
unsigned int pgids[1] = { 333u };
job_pidinfo_t pidinfo;
memset(&pidinfo, 0, sizeof(pidinfo));
pidinfo.jobid = 8001ULL;
pidinfo.npids = 2;
pidinfo.pids = pids;
pidinfo.npgids = 1;
pidinfo.pgids = pgids;
pidinfo.exdata = "pid-exdata";
xce_job_preason_t pend_reasons;
memset(&pend_reasons, 0, sizeof(pend_reasons));
pend_reasons.main_reason = 7;
pend_reasons.nhost_reasons = 0;
pend_reasons.host_based_reasons = NULL;
pend_reasons.nres_reasons = 0;
pend_reasons.res_based_reasons = NULL;
xce_job_t job;
memset(&job, 0, sizeof(job));
job.jobid = 8001ULL;
job.submit = &submit;
job.sub_time = 1000;
job.queue_time = 2000;
job.start_time = 3000;
job.finish_time = 4000;
job.state = 5u;
job.allocs = &job_alloc;
job.run_time = 120;
job.pidinfo = &pidinfo;
job.pend_reasons = &pend_reasons;
job.transient_since = 9999ULL;
job_update_event_t src;
memset(&src, 0, sizeof(src));
src.jobid = 8001ULL;
src.changed_fields = 0x5u;
src.job = &job;
struct capn c;
capn_init_malloc(&c);
struct capn_segment *cs = capn_root(&c).seg;
JobUpdateEvent_ptr p;
encode_JobUpdateEvent_ptr(cs, &p, &src);
capn_setp(capn_root(&c), 0, p.p);
int64_t sz = capn_size(&c);
TEST_ASSERT_TRUE(sz > 0);
uint8_t *buf = (uint8_t *)malloc((size_t)sz);
TEST_ASSERT_NOT_NULL(buf);
capn_write_mem(&c, buf, (size_t)sz, 0);
capn_free(&c);
struct capn d;
TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0));
JobUpdateEvent_ptr dp;
dp.p = capn_getp(capn_root(&d), 0, 1);
struct JobUpdateEvent wire;
read_JobUpdateEvent(&wire, dp);
job_update_event_t dst;
memset(&dst, 0, sizeof(dst));
decode_JobUpdateEvent(&dst, &wire);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.jobid);
TEST_ASSERT_EQUAL_UINT32(0x5u, dst.changed_fields);
TEST_ASSERT_NOT_NULL(dst.job);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.job->jobid);
TEST_ASSERT_EQUAL_UINT64(1000ULL, (uint64_t)dst.job->sub_time);
TEST_ASSERT_EQUAL_UINT64(2000ULL, (uint64_t)dst.job->queue_time);
TEST_ASSERT_EQUAL_UINT64(3000ULL, (uint64_t)dst.job->start_time);
TEST_ASSERT_EQUAL_UINT64(4000ULL, (uint64_t)dst.job->finish_time);
TEST_ASSERT_EQUAL_UINT32(5u, dst.job->state);
TEST_ASSERT_EQUAL_INT(120, dst.job->run_time);
TEST_ASSERT_EQUAL_UINT64(9999ULL, dst.job->transient_since);
TEST_ASSERT_NOT_NULL(dst.job->submit);
TEST_ASSERT_EQUAL_STRING("normal", dst.job->submit->queue);
TEST_ASSERT_EQUAL_STRING("alice", dst.job->submit->user);
TEST_ASSERT_EQUAL_UINT32(2u, dst.job->submit->mintasks);
TEST_ASSERT_EQUAL_UINT32(8u, dst.job->submit->maxtasks);
TEST_ASSERT_EQUAL_STRING("update-test-job", dst.job->submit->job_name);
TEST_ASSERT_EQUAL_STRING("proj-x", dst.job->submit->project_name);
TEST_ASSERT_EQUAL_STRING("grp-x", dst.job->submit->user_group);
TEST_ASSERT_EQUAL_STRING("job update event round-trip test",
dst.job->submit->job_description);
TEST_ASSERT_EQUAL_STRING("select=2", dst.job->submit->resreq);
TEST_ASSERT_NOT_NULL(dst.job->allocs);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.job->allocs->jobid);
TEST_ASSERT_EQUAL_INT(1, dst.job->allocs->nhosts);
TEST_ASSERT_NOT_NULL(dst.job->allocs->allocs);
TEST_ASSERT_EQUAL_STRING("hostA", dst.job->allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_INT(1, dst.job->allocs->allocs[0]->nallocs);
TEST_ASSERT_NOT_NULL(dst.job->allocs->allocs[0]->allocs);
TEST_ASSERT_EQUAL_STRING("cpu",
dst.job->allocs->allocs[0]->allocs[0]->resname);
TEST_ASSERT_EQUAL_STRING("resv1",
dst.job->allocs->allocs[0]->allocs[0]->reserve);
TEST_ASSERT_EQUAL_UINT64(42ULL,
dst.job->allocs->allocs[0]->allocs[0]->instid);
/* Unity is built without UNITY_INCLUDE_DOUBLE; compare via float cast
* (same pattern as scheduler/testsuite/test.c). */
TEST_ASSERT_EQUAL_FLOAT(4.0f,
(float)dst.job->allocs->allocs[0]->allocs[0]->value);
TEST_ASSERT_EQUAL_UINT64(500ULL,
dst.job->allocs->allocs[0]->allocs[0]->from);
TEST_ASSERT_EQUAL_INT64(3600LL,
dst.job->allocs->allocs[0]->allocs[0]->duration);
TEST_ASSERT_EQUAL_INT(0, dst.job->allocs->nres_shared);
TEST_ASSERT_NULL(dst.job->allocs->allocs_shared);
TEST_ASSERT_NOT_NULL(dst.job->pidinfo);
TEST_ASSERT_EQUAL_UINT64(8001ULL, dst.job->pidinfo->jobid);
TEST_ASSERT_EQUAL_INT(2, dst.job->pidinfo->npids);
TEST_ASSERT_EQUAL_UINT32(111u, dst.job->pidinfo->pids[0]);
TEST_ASSERT_EQUAL_UINT32(222u, dst.job->pidinfo->pids[1]);
TEST_ASSERT_EQUAL_INT(1, dst.job->pidinfo->npgids);
TEST_ASSERT_EQUAL_UINT32(333u, dst.job->pidinfo->pgids[0]);
TEST_ASSERT_EQUAL_STRING("pid-exdata", dst.job->pidinfo->exdata);
TEST_ASSERT_NOT_NULL(dst.job->pend_reasons);
TEST_ASSERT_EQUAL_INT(7, dst.job->pend_reasons->main_reason);
TEST_ASSERT_EQUAL_INT(0, dst.job->pend_reasons->nhost_reasons);
TEST_ASSERT_NULL(dst.job->pend_reasons->host_based_reasons);
TEST_ASSERT_EQUAL_INT(0, dst.job->pend_reasons->nres_reasons);
TEST_ASSERT_NULL(dst.job->pend_reasons->res_based_reasons);
free_JobUpdateEvent(&dst);
capn_free(&d);
free(buf);
printf("test_job_update_event_round_trip: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Suite entry point */
/* ------------------------------------------------------------------ */
@ -562,4 +804,6 @@ void test_e2e_capnp(void)
RUN_TEST(test_data_job_array_variant_round_trip);
RUN_TEST(test_replay_header_snapshot_entry_round_trip);
RUN_TEST(test_session_summary_round_trip);
RUN_TEST(test_job_state_change_event_round_trip);
RUN_TEST(test_job_update_event_round_trip);
}

View file

@ -589,6 +589,28 @@ typedef struct {
sched_snapshot_data_t *data;
} replay_snapshot_entry_t;
/* task 7.2 (Req 5 AC 20): the two genuinely new event payload types -- every
* other replay_event_t.body variant above reuses a pre-existing wire type
* (see controller.capnp's comment at `struct JobStateChangeEvent`). */
typedef struct {
uint64_t jobid;
uint32_t old_state;
uint32_t new_state;
int32_t exit_code;
} job_state_change_event_t;
typedef struct {
uint64_t jobid;
/* changedFields is a UInt32 bitmask (JOB_F_PRIORITY, JOB_F_QUEUE,
* JOB_F_RESREQ per design.md) -- no JOB_F_* constants exist anywhere in
* the codebase yet (task 5.1's grep); concrete bit assignments are
* task 9+ event-translation/recording code's job, not this schema-only
* task's. */
uint32_t changed_fields;
xce_job_t *job; /* full updated record -- reuses xce_job_t (Job in
* controller.capnp) */
} job_update_event_t;
typedef struct {
uint64_t ts;
int body_which;
@ -602,6 +624,8 @@ typedef struct {
xce_queue_t *queue_update;
xce_job_rusages_t *job_rusage;
/* endSession: Void, no storage */
job_state_change_event_t *job_state_change;
job_update_event_t *job_update;
} body;
} replay_event_t;

View file

@ -1041,12 +1041,32 @@ struct SnapshotEntry $C.mapname("replay_snapshot_entry_t") {
data @1: SnapshotData; # full core state
}
# task 7.2 (Req 5 AC 20): the two genuinely new event payload types --
# every other ReplayEvent.body variant reuses a pre-existing wire struct
# (see ReplayEvent's comment block below). jobStateChange/jobUpdate are
# added as ReplayEvent.body union members at ordinals 10/11 below.
struct JobStateChangeEvent $C.mapname("job_state_change_event_t") {
jobid @0: UInt64;
oldState @1: UInt32 $C.mapname("old_state");
newState @2: UInt32 $C.mapname("new_state");
exitCode @3: Int32 $C.mapname("exit_code");
}
struct JobUpdateEvent $C.mapname("job_update_event_t") {
jobid @0: UInt64;
# bitmask: JOB_F_PRIORITY, JOB_F_QUEUE, JOB_F_RESREQ -- no JOB_F_* constants
# exist anywhere in the codebase yet (confirmed by task 5.1's grep); concrete
# bit assignments are task 9+ event-translation/recording code's job, not
# this schema-only task's.
changedFields @1: UInt32 $C.mapname("changed_fields");
job @2: Job; # full updated record; reuses existing Job struct
}
# ReplayEvent.body reuses 7 pre-existing wire structs verbatim (Conf4Sched,
# SchedJobData, WorkerData, ResourceData, ResValList x2, QueueConf,
# SchedJobRusages -- verified present in controller.capnp, research.md) plus
# a Void end-of-recording marker. jobStateChange/jobUpdate (task 7.2) are
# added later as new union members at ordinals 10/11, once
# JobStateChangeEvent/JobUpdateEvent exist.
# a Void end-of-recording marker, plus the two new JobStateChangeEvent/
# JobUpdateEvent structs (task 7.2) above.
struct ReplayEvent $C.mapname("replay_event_t") {
ts @0: UInt64; # logical time
body :union $C.mapuniontag("body_which") {
@ -1059,6 +1079,8 @@ struct ReplayEvent $C.mapname("replay_event_t") {
queueUpdate @7: QueueConf $C.mapname("queue_update");
jobRusage @8: SchedJobRusages $C.mapname("job_rusage");
endSession @9: Void; # recording stop marker
jobStateChange @10: JobStateChangeEvent $C.mapname("job_state_change");
jobUpdate @11: JobUpdateEvent $C.mapname("job_update");
}
}