diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index f4368ade..ea064eb2 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -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 diff --git a/src/controller/testsuite/test_e2e_capnp.c b/src/controller/testsuite/test_e2e_capnp.c index 731d048f..82dac9bd 100644 --- a/src/controller/testsuite/test_e2e_capnp.c +++ b/src/controller/testsuite/test_e2e_capnp.c @@ -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); } diff --git a/src/include/typemap.h b/src/include/typemap.h index bbb301a7..d858afe1 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -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; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 5eb5f20e..273b7156 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -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"); } }