diff --git a/.kiro/specs/worker-rejoin-resync/tasks.md b/.kiro/specs/worker-rejoin-resync/tasks.md index 0ec7dd0b..08998123 100644 --- a/.kiro/specs/worker-rejoin-resync/tasks.md +++ b/.kiro/specs/worker-rejoin-resync/tasks.md @@ -1,7 +1,7 @@ # Implementation Plan - [ ] 1. Foundation: wire protocol schema and shared reconciliation types -- [ ] 1.1 Define the wire protocol for the reconciliation exchange +- [x] 1.1 Define the wire protocol for the reconciliation exchange - Add the controller→worker job-state query and worker→controller job-state report as new opcodes, each carrying a paginated list of per-job entries (job ID, status, exit code, worker-reported finish time, and whether the job was part of the original query) - Add the two new cluster-wide configuration fields: the recently-finished reporting window and the reconciliation timeout - Observable completion: the schema builds cleanly and the new message/opcode/config-field definitions are visible in the generated bindings diff --git a/src/include/clusterconfig.h b/src/include/clusterconfig.h index 2a6333fd..f2e12406 100644 --- a/src/include/clusterconfig.h +++ b/src/include/clusterconfig.h @@ -25,6 +25,12 @@ typedef struct { int nlimits; xce_limit_config_t** limits; uint64_t config_generation; /* monotonic audit counter from capnp schema */ + /* worker-rejoin-resync (Req 3.7, 5.3): shared, cluster-wide via + * operational-uniformity override (design.md Configuration Model) -- + * alongside heartbeat_interval/inactive_heartbeat_count above, same + * domain. */ + int worker_resync_window_ms; + int worker_resync_timeout_ms; } cluster_config_t; #endif diff --git a/src/include/typemap.h b/src/include/typemap.h index 5ba1bcc0..26e7fa2a 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -281,6 +281,42 @@ typedef struct { uint64_t responder_time_ms; /* @13: responder wall-clock at sample time (epoch ms) */ } xce_ctrlstatus_t; +/* ── Worker rejoin/resync job-state reconciliation (worker-rejoin-resync + * task 1.1, Req 2.1/2.2/3.5/3.7/5.3/8.1-8.3) ──────────────────────────── + * Frame-only wire types for the controller<->worker job-state + * reconciliation exchange (OPCode_jobStateQuery/OPCode_jobStateReport) -- + * same "no live C struct" convention as capnp_host_admin_state_t/ + * replay_cluster_t above (see controller.capnp's comment at `struct + * JobStateEntry`). An enum-typed capnp field (status) maps to a plain + * `int` here, not a separate named C enum type -- mirrors the existing + * HostRole/JoinData.role and SchedAdminAction/sched_admin_req_t.action + * convention (the generated `enum JobStateStatus` in controller.capnp.h + * already supplies the JobStateStatus_running/finishedOk/finishedFail/ + * unrecognized constants for callers that want them). This is schema-only: + * the C-side send/receive/encode/decode logic is later tasks (1.4 wire + * codec helpers; 6.1/6.2 controller wiring; 4.x worker wiring). */ +typedef struct { + uint64_t jobid; + int status; /* JobStateStatus_running/finishedOk/finishedFail/unrecognized */ + int32_t exit_code; + uint64_t finish_time_ms; + int queried; +} xce_job_state_entry_t; + +typedef struct { + int n_total; + int pos; + int count; + uint64_t* jobids; +} xce_job_state_query_t; + +typedef struct { + int n_total; + int pos; + int count; + xce_job_state_entry_t** entries; +} xce_job_state_report_t; + typedef struct { uint64_t seq; uint32_t opcode; @@ -343,6 +379,8 @@ typedef struct { xce_fetchfile_t* fetchfilemsg; sched_admin_req_t* schedadminmsg; probe_config_tag_req_t* probeconfigtagrequestmsg; + xce_job_state_query_t* jobstatequerymsg; + xce_job_state_report_t* jobstatereportmsg; } body; } xce_request_t; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index a8827ac6..73bfc108 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -51,6 +51,8 @@ enum OPCode { ctrlstatus @40; fetchfile @41; schedAdmin @42; + jobStateQuery @43; + jobStateReport @44; } struct PeerInfo $C.mapname("xce_peerinfo_t") { @@ -715,6 +717,48 @@ struct SchedAdminReply $C.mapname("sched_admin_reply_t") { startedAt @5: UInt64; # logical time recording started (nanoseconds since epoch) } +# ── Worker rejoin/resync job-state reconciliation (worker-rejoin-resync +# task 1.1, Req 2.1/2.2/3.5/3.7/5.3/8.1-8.3) ───────────────────────────── +# Wire protocol for the controller<->worker job-state reconciliation +# exchange sent on worker (re)join: OPCode_jobStateQuery (controller -> +# worker, lists the jobs the controller currently believes are non- +# terminal on that host) and OPCode_jobStateReport (worker -> controller, +# the worker's per-job answer, plus any extra running/recently-finished +# job outside the query, queried=false). Both are sent fire-and-forget +# (no reply arm -- same one-way pattern as e.g. OPCode_push/OPCode_notify) +# and are paginated via the same {ntotal, pos, count} shape already used +# by JobData/WorkerData/ResourceVals (see request.c's xce_reply_get/ +# xce_push_data_req). This is schema-only: the C-side encode/decode +# helpers (task 1.4), controller/worker send-receive wiring (tasks 3.1, +# 4.2, 6.1, 6.2), and reconciliation-apply logic (tasks 2.1-2.2, 3.2) are +# all later tasks, out of this task's scope. +enum JobStateStatus { + running @0; + finishedOk @1; + finishedFail @2; + unrecognized @3; +} + +struct JobStateEntry $C.mapname("xce_job_state_entry_t") { + jobid @0: UInt64; + status @1: JobStateStatus; + exitCode @2: Int32 $C.mapname("exit_code"); # meaningful only for finishedFail + finishTimeMs @3: UInt64 $C.mapname("finish_time_ms"); # meaningful only for finishedOk/finishedFail; worker-reported (Req 3.5), not report-send time + queried @4: Bool; # false = Req 3.6 extra, not part of the controller's original query +} + +struct JobStateQuery $C.mapname("xce_job_state_query_t") { + ntotal @0: UInt32 $C.mapname("n_total"); + pos @1: UInt32; + jobids @2: List(UInt64) $C.mapname("jobids") $C.maplistcount("count"); +} + +struct JobStateReport $C.mapname("xce_job_state_report_t") { + ntotal @0: UInt32 $C.mapname("n_total"); + pos @1: UInt32; + entries @2: List(JobStateEntry) $C.mapname("entries") $C.maplistcount("count"); +} + struct Request $C.mapname("xce_request_t"){ opcode @0: OPCode; seq @1: UInt64; @@ -750,6 +794,8 @@ struct Request $C.mapname("xce_request_t"){ fetchfilemsg @30: FetchFileMsg; schedadminmsg @31: SchedAdminRequest; probeconfigtagrequestmsg @32: ProbeConfigTagRequest; + jobstatequerymsg @33: JobStateQuery; + jobstatereportmsg @34: JobStateReport; } } @@ -798,6 +844,11 @@ struct ClusterConfig $C.mapname("cluster_config_t") { hproviders @10: List(ScalerConfig) $C.mapname("hproviders") $C.maplistcount("nhproviders"); limits @11: List(LimitConfig) $C.mapname("limits") $C.maplistcount("nlimits"); configGeneration @12: UInt64 $C.mapname("config_generation"); + # worker-rejoin-resync (Req 3.7, 5.3): shared, cluster-wide via + # operational-uniformity override (design.md Configuration Model) -- + # alongside heartbeatInterval/inactiveHeartbeatCount above, same domain. + workerResyncWindowMs @13: UInt32 $C.mapname("worker_resync_window_ms"); + workerResyncTimeoutMs @14: UInt32 $C.mapname("worker_resync_timeout_ms"); } # ── Persistence configuration (Part 2.6) ────────────────────────────