feat(worker-rejoin-resync): define reconciliation wire protocol (task 1.1)

Add OPCode_jobStateQuery/OPCode_jobStateReport, JobStateEntry/
JobStateQuery/JobStateReport structs (paginated, mirroring JobData/
WorkerData/ResourceVals' {ntotal,pos,count} shape), and
workerResyncWindowMs/workerResyncTimeoutMs ClusterConfig fields to
controller.capnp, plus the companion typemap.h/clusterconfig.h C type
declarations this codebase's capnp $C.mapname codegen requires to
compile. Both new opcodes are fire-and-forget (Request.body union only,
no RequestReply arm), matching design.md and the push/notify precedent.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
This commit is contained in:
Rongsong Shen 2026-09-01 08:44:33 +00:00
parent 6c908d5017
commit 1ff57ef66e
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
4 changed files with 96 additions and 1 deletions

View file

@ -1,7 +1,7 @@
# Implementation Plan # Implementation Plan
- [ ] 1. Foundation: wire protocol schema and shared reconciliation types - [ ] 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 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 - 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 - Observable completion: the schema builds cleanly and the new message/opcode/config-field definitions are visible in the generated bindings

View file

@ -25,6 +25,12 @@ typedef struct {
int nlimits; int nlimits;
xce_limit_config_t** limits; xce_limit_config_t** limits;
uint64_t config_generation; /* monotonic audit counter from capnp schema */ 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; } cluster_config_t;
#endif #endif

View file

@ -281,6 +281,42 @@ typedef struct {
uint64_t responder_time_ms; /* @13: responder wall-clock at sample time (epoch ms) */ uint64_t responder_time_ms; /* @13: responder wall-clock at sample time (epoch ms) */
} xce_ctrlstatus_t; } 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 { typedef struct {
uint64_t seq; uint64_t seq;
uint32_t opcode; uint32_t opcode;
@ -343,6 +379,8 @@ typedef struct {
xce_fetchfile_t* fetchfilemsg; xce_fetchfile_t* fetchfilemsg;
sched_admin_req_t* schedadminmsg; sched_admin_req_t* schedadminmsg;
probe_config_tag_req_t* probeconfigtagrequestmsg; probe_config_tag_req_t* probeconfigtagrequestmsg;
xce_job_state_query_t* jobstatequerymsg;
xce_job_state_report_t* jobstatereportmsg;
} body; } body;
} xce_request_t; } xce_request_t;

View file

@ -51,6 +51,8 @@ enum OPCode {
ctrlstatus @40; ctrlstatus @40;
fetchfile @41; fetchfile @41;
schedAdmin @42; schedAdmin @42;
jobStateQuery @43;
jobStateReport @44;
} }
struct PeerInfo $C.mapname("xce_peerinfo_t") { 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) 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"){ struct Request $C.mapname("xce_request_t"){
opcode @0: OPCode; opcode @0: OPCode;
seq @1: UInt64; seq @1: UInt64;
@ -750,6 +794,8 @@ struct Request $C.mapname("xce_request_t"){
fetchfilemsg @30: FetchFileMsg; fetchfilemsg @30: FetchFileMsg;
schedadminmsg @31: SchedAdminRequest; schedadminmsg @31: SchedAdminRequest;
probeconfigtagrequestmsg @32: ProbeConfigTagRequest; 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"); hproviders @10: List(ScalerConfig) $C.mapname("hproviders") $C.maplistcount("nhproviders");
limits @11: List(LimitConfig) $C.mapname("limits") $C.maplistcount("nlimits"); limits @11: List(LimitConfig) $C.mapname("limits") $C.maplistcount("nlimits");
configGeneration @12: UInt64 $C.mapname("config_generation"); 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) ──────────────────────────── # ── Persistence configuration (Part 2.6) ────────────────────────────