diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index 26c4ab05..755f00f6 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -182,12 +182,13 @@ - _Verified: 2026-08-04 — DispatchDecision/PendDecision/PreemptVictim/PreemptDecision added, reusing JobAlloc/JobPendReason per spec. SnapshotData's task-7.1 empty placeholder replaced wholesale with the real 11 fields (config/jobs/queues/workers/resources/resvals/users/projects/prngSeed/internalState/internalVersion), both in the capnp schema and its typemap.h C mirror. ReplayDecision's ts-only placeholder gained a purely additive body union (dispatch@1/pend@2/preempt@3). One deviation from design.md's literal text: SchedUser/SchedProject use fresh mapnames (sched_snapshot_user_t/sched_snapshot_project_t) instead of design.md's literal xce_sched_user_t/xce_sched_project_t, since those names are already real C structs in sched/types.h that share a translation unit with typemap.h in core_events.c/schedobj.c — reusing them would be a genuine duplicate-typedef compile error, not a style choice. Field lists (name-only; name+priority) verified unchanged, deviation rigorously re-derived and confirmed necessary by independent review, documented in both files. Found and fixed a real bug mid-implementation (a doc comment's literal */ substring prematurely closing a C block comment) logged as bug-1493. ReplayEvent's union (tasks 7.1/7.2) and every other existing struct byte-identical. Five deep round-trip tests (all 11 SnapshotData fields populated, version-mismatch demonstrated). 1082/1082 controller_unit_test, full clean project build, confirmed independently three times (implementer, orchestrator, reviewer, all matching exactly), review APPROVED first pass._ - _Boundary: ReplayFormat — not (P), same file as 7.2/7.4 (see 7.2)_ -- [ ] 7.4 Add the scheduler admin opcode schema +- [x] 7.4 Add the scheduler admin opcode schema - Add the new opcode value, the admin action enum (record start/stop/status), and the paired request/reply message types (reply carrying ok/error/recording-state/file/event-count/started-at) to the wire schema, plus their entries in the existing request/reply message unions - Observable: encoding each of the three admin actions and their corresponding replies and decoding them back yields field-for-field equality - _Requirements: 5.10_ - _Depends: 7.1_ - _Boundary: ReplayFormat — not (P), same file as 7.2/7.3 (see 7.2)_ + - _Verified: 2026-08-04 — SchedAdminAction enum + SchedAdminRequest/SchedAdminReply structs added matching requirements.md Req 5 AC 10 verbatim. schedAdmin@42 (OPCode), schedadminmsg@31 (Request.body), schedadminreplymsg@19 (RequestReply.data) — all three the exact next-available ordinals, verified live before dispatch and re-verified byte-identical below by review. Deliberately schema-only per this task's own text — the controller dispatch logic (cluster.c's case OPCode_schedAdmin:) is task 10.1's job, not this one; confirmed cluster.c untouched. Round-trip tests cover all 3 SchedAdminAction values and both ok=true/ok=false SchedAdminReply cases with an asymmetric ok/recording combination, ruling out a Bool-field bit-packing mixup. 1084/1084 controller_unit_test, full clean project build, confirmed independently three times, review APPROVED first pass. Closes out Group 7 (Foundation: Replay Wire Format) entirely — tasks 7.1-7.4 all complete._ - [ ] 7.5 Replay format round-trip test suite - Encode and decode every new message type introduced in this group, asserting field-for-field equality after round-trip diff --git a/src/controller/testsuite/test_e2e_capnp.c b/src/controller/testsuite/test_e2e_capnp.c index 2b05641f..e867c8ea 100644 --- a/src/controller/testsuite/test_e2e_capnp.c +++ b/src/controller/testsuite/test_e2e_capnp.c @@ -1344,6 +1344,157 @@ static void test_snapshot_data_round_trip(void) printf("test_snapshot_data_round_trip: PASS\n"); } +/* ------------------------------------------------------------------ */ +/* Test 16: SchedAdminRequest round-trip (scheduler-core task 7.4, */ +/* Req 5 AC 10) -- encodes and decodes a SchedAdminRequest for each of */ +/* the three SchedAdminAction enum values (recordStart/recordStop/ */ +/* recordStatus), proving the plain enum-typed `action` field (mapped */ +/* to `int` in sched_admin_req_t, same convention as HostRole/ */ +/* JoinData.role) and the `recordFile` Text field round-trip field-for- */ +/* field. recordStart uses a non-empty recordFile; recordStop/ */ +/* recordStatus use an empty one (auto-generate / pure query), covering */ +/* both the populated and empty-string Text cases. Same encode_X_ptr/ */ +/* decode_X codec-bridge pattern as Test 9 (job_state_change_event_t) */ +/* since sched_admin_req_t is a fully-scalar struct plus one Text */ +/* field, no Data fields. */ +/* ------------------------------------------------------------------ */ + +static void sched_admin_request_round_trip_case(int action, const char *record_file) +{ + sched_admin_req_t src; + memset(&src, 0, sizeof(src)); + src.action = action; + src.recordFile = (char *)record_file; + + struct capn c; + capn_init_malloc(&c); + struct capn_segment *cs = capn_root(&c).seg; + + SchedAdminRequest_ptr p; + encode_SchedAdminRequest_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)); + SchedAdminRequest_ptr dp; + dp.p = capn_getp(capn_root(&d), 0, 1); + struct SchedAdminRequest wire; + read_SchedAdminRequest(&wire, dp); + sched_admin_req_t dst; + memset(&dst, 0, sizeof(dst)); + decode_SchedAdminRequest(&dst, &wire); + + TEST_ASSERT_EQUAL_INT(action, (int)dst.action); + TEST_ASSERT_EQUAL_STRING(record_file, dst.recordFile); + + free_SchedAdminRequest(&dst); + capn_free(&d); + free(buf); +} + +static void test_sched_admin_request_round_trip(void) +{ + /* recordStart with a non-empty recordFile. */ + sched_admin_request_round_trip_case(SchedAdminAction_recordStart, + "manual-session.replay"); + /* recordStop with an empty recordFile (stop takes no filename). */ + sched_admin_request_round_trip_case(SchedAdminAction_recordStop, ""); + /* recordStatus with an empty recordFile (status is a pure query). */ + sched_admin_request_round_trip_case(SchedAdminAction_recordStatus, ""); + + printf("test_sched_admin_request_round_trip: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 17: SchedAdminReply round-trip (scheduler-core task 7.4, */ +/* Req 5 AC 10) -- encodes and decodes a SchedAdminReply for both an */ +/* ok=true success case and an ok=false error case, each populating all */ +/* six fields (ok, error, recording, file, eventCount, startedAt) with */ +/* non-default values, proving every field -- including the two Bool */ +/* fields (ok, recording) packed into the same wire byte -- round-trips */ +/* independently regardless of the other's value. `error` is modelled */ +/* as a general diagnostic-message field (populated with a non-fatal */ +/* advisory even on success, an outright failure reason on error) so */ +/* both cases can legitimately carry a non-empty value. Same codec- */ +/* bridge pattern as Test 9/16. */ +/* ------------------------------------------------------------------ */ + +static void sched_admin_reply_round_trip_case(int ok, const char *error, + int recording, const char *file, + uint64_t event_count, + uint64_t started_at) +{ + sched_admin_reply_t src; + memset(&src, 0, sizeof(src)); + src.ok = ok; + src.error = (char *)error; + src.recording = recording; + src.file = (char *)file; + src.eventCount = event_count; + src.startedAt = started_at; + + struct capn c; + capn_init_malloc(&c); + struct capn_segment *cs = capn_root(&c).seg; + + SchedAdminReply_ptr p; + encode_SchedAdminReply_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)); + SchedAdminReply_ptr dp; + dp.p = capn_getp(capn_root(&d), 0, 1); + struct SchedAdminReply wire; + read_SchedAdminReply(&wire, dp); + sched_admin_reply_t dst; + memset(&dst, 0, sizeof(dst)); + decode_SchedAdminReply(&dst, &wire); + + TEST_ASSERT_EQUAL_INT(ok ? 1 : 0, dst.ok ? 1 : 0); + TEST_ASSERT_EQUAL_STRING(error, dst.error); + TEST_ASSERT_EQUAL_INT(recording ? 1 : 0, dst.recording ? 1 : 0); + TEST_ASSERT_EQUAL_STRING(file, dst.file); + TEST_ASSERT_EQUAL_UINT64(event_count, dst.eventCount); + TEST_ASSERT_EQUAL_UINT64(started_at, dst.startedAt); + + free_SchedAdminReply(&dst); + capn_free(&d); + free(buf); +} + +static void test_sched_admin_reply_round_trip(void) +{ + /* ok=true success case (e.g. a recordStatus reply): all six fields + * non-default. */ + sched_admin_reply_round_trip_case(1, "warning: recording directory nearing disk quota", + 1, "session-0007.replay", + 12345ULL, 1700000000000000000ULL); + + /* ok=false error case (e.g. a failed recordStart): all six fields + * still non-default, proving the wire carries every field regardless + * of `ok`. */ + sched_admin_reply_round_trip_case(0, "recording directory not writable", + 1, "session-0006.replay", + 999ULL, 1699999999000000000ULL); + + printf("test_sched_admin_reply_round_trip: PASS\n"); +} + /* ------------------------------------------------------------------ */ /* Suite entry point */ /* ------------------------------------------------------------------ */ @@ -1368,4 +1519,6 @@ void test_e2e_capnp(void) RUN_TEST(test_preempt_decision_round_trip); RUN_TEST(test_replay_decision_dispatch_union_round_trip); RUN_TEST(test_snapshot_data_round_trip); + RUN_TEST(test_sched_admin_request_round_trip); + RUN_TEST(test_sched_admin_reply_round_trip); } diff --git a/src/include/typemap.h b/src/include/typemap.h index fd5be3b1..459e6a49 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -216,6 +216,31 @@ typedef struct { int success; } wal_ack_reply_t; +/* Scheduler admin opcode request/reply (scheduler-core task 7.4, Req 5 AC 10). + * OPCode_schedAdmin: start/stop/query a scheduler-side `.replay` recording + * session. No field-level $C.mapname on either capnp struct (per the exact + * verbatim schema spec), so -- same convention as xce_ctrlstatus_t below -- + * the generated accessors and this mapped struct's field names match the + * capnp schema's camelCase names exactly (no snake_case conversion). action + * mirrors the existing HostRole/JoinData.role convention: an enum-typed + * capnp field maps to a plain `int` here, not a separate named C enum type + * (the generated `enum SchedAdminAction` in controller.capnp.h already + * supplies the SchedAdminAction_recordStart/recordStop/recordStatus + * constants for callers that want them). */ +typedef struct { + int action; + char *recordFile; +} sched_admin_req_t; + +typedef struct { + int ok; + char *error; + int recording; + char *file; + uint64_t eventCount; + uint64_t startedAt; +} sched_admin_reply_t; + /* ctrlstatus reply payload (controller-status spec): one configured * controller (CtrlPeer) and the lean identity/health snapshot (CtrlStatus). * The peers list maps via $C.mapname("peers.list")/$C.maplistcount("peers.count") @@ -265,6 +290,7 @@ typedef struct { wal_ack_reply_t* walackreplymsg; xce_ctrlstatus_t* ctrlstatus; xce_fetchfile_reply_t* fetchfilereply; + sched_admin_reply_t* schedadminreplymsg; } data; /* Req 7: standby redirect fields — populated when status == XCE_ERR_STANDBY */ char *redirect_host; @@ -304,6 +330,7 @@ typedef struct { wal_catchup_entry_t* walcatchupentrymsg; snapshot_transfer_t* snapshottransfermsg; xce_fetchfile_t* fetchfilemsg; + sched_admin_req_t* schedadminmsg; } body; } xce_request_t; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 7cb889b3..76b46e67 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -50,6 +50,7 @@ enum OPCode { preVoteReply @39; ctrlstatus @40; fetchfile @41; + schedAdmin @42; } struct PeerInfo $C.mapname("xce_peerinfo_t") { @@ -591,6 +592,7 @@ struct RequestReply $C.mapname("xce_reply_t"){ walackreplymsg @16: WalAckReply $C.mapname("walackreplymsg"); # WAL apply ACK from standby ctrlstatus @17: CtrlStatus; # ctrlstatus reply: identity/health snapshot fetchfilereply @18: FetchFileReply; + schedadminreplymsg @19: SchedAdminReply; } redirectHost @11: Text $C.mapname("redirect_host"); @@ -642,6 +644,38 @@ struct FetchFileReply $C.mapname("xce_fetchfile_reply_t") { jobid @4: UInt64 $C.mapname("jobid"); } +# ── Scheduler admin opcode (scheduler-core task 7.4, Req 5 AC 10) ──── +# OPCode_schedAdmin request/reply pair: start/stop/query a scheduler-side +# `.replay` recording session. The actual controller-side dispatch +# (cluster.c's myreq_handler forwarding this opcode to the scheduler) is a +# LATER task, out of this schema-only task's scope -- this struct trio is +# purely the wire contract. +enum SchedAdminAction { + recordStart @0; + recordStop @1; + recordStatus @2; +} + +struct SchedAdminRequest $C.mapname("sched_admin_req_t") { + action @0: SchedAdminAction; + recordFile @1: Text; # filename (not a path) for recordStart; empty = auto-generate. + # See AC 10 "Record file path validation" — the scheduler + # resolves this beneath a configured recording directory; + # it is never used as a caller-controlled absolute path. +} + +struct SchedAdminReply $C.mapname("sched_admin_reply_t") { + ok @0: Bool; + error @1: Text; # human-readable error if !ok + + # Fields populated for recordStatus (also returned by recordStart/recordStop + # for confirmation — caller can verify state after each action): + recording @2: Bool; # true between start and stop + file @3: Text; # record file path (auto-generated name returned by start) + eventCount @4: UInt64; # events written since recording started + startedAt @5: UInt64; # logical time recording started (nanoseconds since epoch) +} + struct Request $C.mapname("xce_request_t"){ opcode @0: OPCode; seq @1: UInt64; @@ -675,6 +709,7 @@ struct Request $C.mapname("xce_request_t"){ walcatchupentrymsg @28: WalCatchupEntry; snapshottransfermsg @29: SnapshotTransfer; fetchfilemsg @30: FetchFileMsg; + schedadminmsg @31: SchedAdminRequest; } }