myxce/src/schema/controller.capnp
Rongsong Shen 026fe97b5a
fix(bug-1737): scheduler can never learn a job finished successfully
sched_job_exit_fun() (renamed sched_job_finish_fun, sched.c) subscribes
to HOOK_DETAIL_JOB_EXIT, which fires for EVERY job finish, both success
and failure -- record_job_finish_fact() (job.c) decides DONE vs EXIT
from finish->exit_code BEFORE firing this same hook detail either way
(mirrors persist.c's own HOOK_DETAIL_JOB_EXIT -> WAL_JOB_FINISH
mapping). This function discarded info->exit_code and sent only a bare
jobid, forcing sched_handle_notify() (scheduler/src/handler.c) to
unconditionally map EVERY finish -- including a clean, successful one
-- to JOB_STAT_EXIT. A subsequent HOOK_DETAIL_JOB_UPDATE notify usually
self-corrected this moments later via the full job state, but if that
second notify was dropped during a scheduler reconnect gap (the same
window bug-1731/bug-1734 already documented), the scheduler was left
permanently stuck showing EXIT for a job the controller correctly
recorded as DONE.

Live-reproduced on a real 3-node HA cluster + worker: job
0-8-1788143749-0[38] genuinely completed successfully (worker log:
"completed successfully", controller joblist: DONE) but the
scheduler's own replay showed EXIT.

Fix: carry {jobid, exit_code} across the wire via the existing
JobFinishInfo shape (the same struct apiserver_job_exit_fun already
reuses for this identical hook) instead of a bare UInt64, so
sched_handle_notify() can make the identical exit_code==0-vs-not
decision job.c itself makes. Renamed sched_job_exit_fun to
sched_job_finish_fun to reflect what it actually does.

Live-verified: a 51-job run (all exit 0) showed 100% DONE on the
scheduler side after the fix.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019QxaSzHRB7pmYFp4cVnwkA
2026-08-31 04:20:27 +00:00

1325 lines
56 KiB
Cap'n Proto
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

@0xdefcdf6377db2e8e;
using C = import "/c.capnp";
$C.fieldgetset;
$C.codecgen;
$C.extraheader("#include \"clusterconfig.h\"");
$C.extraheader("#include \"typemap.h\"");
enum OPCode {
nop @0;
reply @1;
hello @2;
join @3;
shutdown @4;
get @5;
push @6;
update @7;
jobdispatch @8;
startjob @9;
notify @10;
jobsub @11;
jobstatus @12;
jobrusage @13;
hostcontrol @14;
jobcontrol @15;
newschedjobs @16;
clusterconf @17;
queuecontrol @18;
runtasks @19;
setpreason @20;
clrpreason @21;
replEntry @22;
replAck @23;
redirect @24;
heartbeat @25;
steppingDown @26;
replSnapshot @27;
replCatchUpDone @28;
replConfig @29;
replConfigPush @30;
upgradeConfig @31;
requestVote @32;
probeConfigTag @33;
walAck @34;
upgradeConfigAck @35;
snapshotTransfer @36;
walCatchupEntry @37;
preVoteRequest @38;
preVoteReply @39;
ctrlstatus @40;
fetchfile @41;
schedAdmin @42;
}
struct PeerInfo $C.mapname("xce_peerinfo_t") {
host @0: Text $C.mapname("name");
key @1: Text;
}
struct HostBasedReason $C.mapname("xce_host_reason_t") {
host @0: Text;
reason @1: UInt32;
}
struct ResourceBasedReason $C.mapname("xce_res_reason_t") {
resource @0: Text;
host @1: Text;
reason @2: UInt32;
}
struct JobPendReason $C.mapname("xce_job_preason_t") {
mainReason @0: UInt32 $C.mapname("main_reason");
hostReasons @1: List(HostBasedReason) $C.mapname("host_based_reasons") $C.maplistcount("nhost_reasons");
resReasons @2: List(ResourceBasedReason) $C.mapname("res_based_reasons") $C.maplistcount("nres_reasons");
}
struct StrList $C.mapname("xce_strlist_t") {
data @0: List(Text) $C.mapname("data") $C.maplistcount("count");
}
struct TimeWindowTime $C.mapname("xce_timewindow_time_t") {
day @0: UInt32 $C.mapname("day");
hour @1: UInt32 $C.mapname("hour");
min @2: UInt32 $C.mapname("min");
}
struct TimeWindowSeg $C.mapname("xce_timewindow_seg_config_t") {
timezone @0: Text $C.mapname("tz");
open @1: TimeWindowTime $C.mapname("open");
close @2: TimeWindowTime $C.mapname("close");
}
struct TimeWindow $C.mapname("xce_timewindow_config_t") {
segs @0: List(TimeWindowSeg) $C.mapname("segs") $C.maplistcount("nsegs");
}
struct GroupConfig $C.mapname("xce_group_config_t") {
name @0: Text $C.mapname("name");
members @1: List(Text) $C.mapname("members.data") $C.maplistcount("members.count");
exmembers @2: List(Text) $C.mapname("ex_members.data") $C.maplistcount("ex_members.count");
}
# This data structure will not be transfer in wire
# directly, it will be assembled by multiple Data packet
struct Dataset $C.mapname("dataset_t") {
table @0: UInt32;
nrows @1: UInt32;
rows @2: Void; # do NOT change this, it's place holder
}
struct ResKey $C.mapname("xce_reskey_t") {
resname @0: Text;
hostname @1: Text;
}
struct MatchCondition $C.mapname("xce_match_t") {
condition :union $C.mapuniontag("type") {
matchall @0: Void;
matchjid @1: UInt64 $C.mapname("jobid");
matchwname @2: Text $C.mapname("wname");
matchreskey @3: ResKey $C.mapname("reskey");
}
}
struct HostAttrs $C.mapname("hostattrs_t") {
type @0: Text $C.mapname("hosttype");
model @1: Text $C.mapname("hostmodel");
cpufactor @2: Float64;
maxcpus @3: Int32;
maxmem @4: Float64;
maxswap @5: Float64;
maxtmp @6: Float64;
ndisks @7: Int32;
}
struct Worker $C.mapname("xce_worker_t"){
name @0: Text;
attrs @1: HostAttrs;
status @2: UInt32;
dispatchWindow @3: TimeWindow $C.mapname("dispatch_window_config");
thresholds @4: List(Threshold) $C.mapname("thresholds") $C.maplistcount("nthresholds");
resvals @5: List(NameValue) $C.mapname("resvals_to_sched") $C.maplistcount("nresvals_to_sched");
deadSince @6: UInt64 = 0 $C.mapname("dead_since"); # epoch seconds; set when worker becomes UNAVAILABLE
}
enum HostRole {
client @0;
worker @1;
scheduler @2;
controller @3;
apiserver @4;
}
struct JoinData $C.mapname("xce_joindata_t") {
role @0: HostRole;
data :union {
nodata @1: Void;
attrs @2: HostAttrs;
}
}
struct Value $C.mapname("xce_val_t"){
u :union $C.mapuniontag("which") {
ivalue32 @0: Int32;
uvalue32 @1: UInt32;
ivalue64 @2: Int64;
uvalue64 @3: UInt64;
svalue @4: Text;
fvalue @5: Float64;
bvalue @6: Bool;
}
}
struct LimitValue $C.mapname("xce_limit_val_t") {
resname @0: Text $C.mapname("resname");
percentage @1: Bool $C.mapname("percent");
value @2: Float64 $C.mapname("val");
}
struct ValueWithIdx $C.mapname("xce_val_with_idx_t") {
idx @0: UInt32;
val @1: Value $C.mapname("val");
}
struct NameValue $C.mapname("xce_name_val_t") {
name @0: Text $C.mapname("name");
val @1: Value $C.mapname("val");
}
struct ConfigVar $C.mapname("xce_config_t") {
name @0: Text;
value @1: Text;
}
struct ResourceDistribution $C.mapname("res_distribution_t") {
value @0: Value;
hosts @1: List(Text) $C.maplistcount("nhosts");
}
struct Resource $C.mapname("xce_resource_t"){
name @0: Text;
desc @1: Text;
idx @2: Int32;
valtype @3: Int32;
ordtype @4: Int32;
flags @5: Int32;
interval @6: Int32;
lastReport @7: UInt64 $C.mapname("last_report");
# distributions @8: List(ResourceDistribution) $C.maplistcount("ndistributions");
# hostBased @9: Value $C.mapname("hbased_defined");
}
struct Threshold $C.mapname("xce_res_threshold_t"){
name @0: Text;
sched @1: Float64;
stop @2: Float64;
}
struct QueueConf $C.mapname("xce_queue_t") {
name @0: Text $C.mapname("queue");
attrs @1: UInt32;
status @2: UInt32;
priority @3: Int32;
acceptIntervals @4: Int32 $C.mapname("accept_interval");
schedDelay @5: Int32 $C.mapname("sched_delay");
resreq @6: Text;
users @7: GroupConfig $C.mapname("users");
hosts @8: GroupConfig $C.mapname("hosts");
thresholds @9: List(Threshold) $C.maplistcount("nthresholds");
preExec @10: Text $C.mapname("pre_exec");
postExec @11: Text $C.mapname("post_exec");
runWindow @12: TimeWindow $C.mapname("runwindow_config");
dispatchWindow @13: TimeWindow $C.mapname("dispatchwindow_config");
}
struct Conf4Sched $C.mapname("xce_conf4sched_t"){
clustername @0: Text;
configvars @1: List(ConfigVar) $C.mapname("confvars") $C.maplistcount("n_confvars");
resources @2: List(Resource) $C.maplistcount("n_resources");
queues @3: List(QueueConf) $C.maplistcount("n_queues");
ugroups @4: List(GroupConfig) $C.maplistcount("n_ugroups");
hgroups @5: List(GroupConfig) $C.maplistcount("n_hgroups");
resvals @6: List(NameValue) $C.mapname("resvals") $C.maplistcount("nresvals");
hproviders @7: List(ScalerConfig) $C.mapname("hproviders") $C.maplistcount("nhproviders");
limits @8: List(LimitConfig) $C.mapname("limits") $C.maplistcount("nlimits");
}
struct Conf4Worker $C.mapname("xce_conf4worker_t") {
clustername @0: Text;
configvars @1: List(ConfigVar) $C.mapname("confvars") $C.maplistcount("n_confvars");
}
struct ConfigData $C.mapname("xce_configdata_t") {
data :union $C.mapuniontag("data_which") {
schedconfig @0: Conf4Sched $C.mapname("conf4sched");
workerconfig @1: Conf4Worker $C.mapname("conf4worker");
clusterconfig @2: ClusterConfig $C.mapname("clusterconfig");
}
}
struct JobPIDInfo $C.mapname("job_pidinfo_t"){
jobId @0: UInt64 $C.mapname("jobid");
pids @1: List(UInt32) $C.mapname("pids") $C.maplistcount("npids");
pgids @2: List(UInt32) $C.mapname("pgids") $C.maplistcount("npgids");
exdata @3: Text $C.mapname("exdata");
}
# bug-1690: xce_job_runusage_t (xce/types.h) previously had NO Cap'n Proto
# mapping at all -- unlike JobAlloc/JobPIDInfo above, which already existed
# as Job sub-fields the WAL_JOB_DISPATCH/WAL_JOB_START fixes (bug-1678/1687)
# could reuse standalone. WAL_JOB_RUSAGE needs one to carry the client-
# reported per-host CPU/mem usage in its fact payload so a restart/catch-up
# replay can reconstruct job.run_usage instead of leaving it permanently
# NULL. NOT the same type as JobRusages/SchedJobRusages below (xce_job_
# rusages_t) -- that is the client's WIRE UPDATE REQUEST shape
# ({jobid, pids[], pgids[], usages[]}); this is the CONTROLLER'S ACCUMULATED
# per-job/per-host usage shape, a different struct entirely.
struct HostRunusage $C.mapname("xce_host_runusage_t"){
hostName @0: Text $C.mapname("host_name");
cpuTime @1: Int32 $C.mapname("cpu_time");
memUsage @2: Float64 $C.mapname("mem_usage");
memSwap @3: Float64 $C.mapname("mem_swap");
tasksCurrent @4: Int32 $C.mapname("tasks_current");
}
struct JobRunusage $C.mapname("xce_job_runusage_t"){
maxMem @0: Float64 $C.mapname("max_mem");
avgMem @1: Float64 $C.mapname("avg_mem");
usedMem @2: Float64 $C.mapname("used_mem");
cpuTime @3: Int32 $C.mapname("cpu_time");
peakCpuTime @4: Int32 $C.mapname("peak_cpu_time");
peakCpuDuration @5: Int32 $C.mapname("peak_cpu_duration");
memEfficiency @6: Float32 $C.mapname("mem_efficiency");
avgCpuEfficiency @7: Float32 $C.mapname("avg_cpu_efficiency");
peakCpuEfficiency @8: Float32 $C.mapname("peak_cpu_efficiency");
hostRunusage @9: List(HostRunusage) $C.mapname("host_runusage") $C.maplistcount("num_hosts");
}
struct JobFail $C.mapname("job_failinfo_t"){
jobId @0: UInt64 $C.mapname("jobid");
code @1: Int32 $C.mapname("code");
desc @2: Text $C.mapname("detail");
stage @3: UInt32 $C.mapname("stage");
}
struct JobStartInfo $C.mapname("job_startinfo_t"){
exechome @0: Text;
execcwd @1: Text;
execuser @2: Text;
cmdline @3: Text;
exdata @4: Text;
launchtype @5: Text $C.mapname("launch_type");
qPreExec @6: Text $C.mapname("q_pre_exec");
qPostExec @7: Text $C.mapname("q_post_exec");
preExec @8: Text $C.mapname("job_pre_exec");
postExec @9: Text $C.mapname("job_post_exec");
}
struct Job $C.mapname("xce_job_t"){
jobid @0: UInt64;
submitData @1: JobSubmitData $C.mapname("submit");
subtime @2: UInt64 $C.mapname("sub_time");
queuetime @3: UInt64 $C.mapname("queue_time");
starttime @4: UInt64 $C.mapname("start_time");
finishtime @5: UInt64 $C.mapname("finish_time");
state @6: UInt32;
allocs @7: JobAlloc $C.mapname("allocs");
runtime @8: UInt32 $C.mapname("run_time");
pidinfo @9: JobPIDInfo;
pendreason @10: JobPendReason $C.mapname("pend_reasons");
transientSince @11: UInt64 = 0 $C.mapname("transient_since"); # epoch seconds; set when job enters JOB_STAT_TRANSIENT
}
struct SchedJob $C.mapname("xce_job_t"){
jobid @0: UInt64;
submitData @1: JobSubmitData $C.mapname("submit");
subtime @2: UInt64 $C.mapname("sub_time");
queuetime @3: UInt64 $C.mapname("queue_time");
starttime @4: UInt64 $C.mapname("start_time");
finishtime @5: UInt64 $C.mapname("finish_time");
state @6: UInt32;
allocs @7: JobAlloc $C.mapname("allocs");
runtime @8: UInt32 $C.mapname("run_time");
}
struct SchedJobRusages $C.mapname("xce_job_rusages_t") {
jobid @0: UInt64 $C.mapname("jobid");
usages @1: List(ResUsage) $C.mapname("usages") $C.maplistcount("count");
}
struct JobData $C.mapname("xce_jobs_data_t") {
ntotal @0: UInt32 $C.mapname("n_total");
pos @1: UInt32;
jobs @2: List(Job) $C.maplistcount("count");
}
struct SchedJobData $C.mapname("xce_jobs_data_t") {
ntotal @0: UInt32 $C.mapname("n_total");
pos @1: UInt32;
jobs @2: List(SchedJob) $C.maplistcount("count");
}
struct WorkerData $C.mapname("xce_workers_data_t"){
ntotal @0: UInt32 $C.mapname("n_total");
pos @1: UInt32;
workers @2: List(Worker) $C.maplistcount("count");
}
struct ResourceData $C.mapname("xce_resource_data_t") {
ntotal @0: UInt32 $C.mapname("n_total");
pos @1: UInt32;
resources @2: List(Resource) $C.maplistcount("count");
}
struct ResourceVal $C.mapname("xce_resval_t") {
name @0: Text $C.mapname("name");
host @1: Text $C.mapname("host");
val @2: Value $C.mapname("val");
}
struct ResourceVals $C.mapname("xce_resval_data_t") {
ntotal @0: UInt32 $C.mapname("n_total");
pos @1: UInt32;
vals @2: List(ResourceVal) $C.mapname("vals") $C.maplistcount("count");
}
struct XceData $C.mapname("xce_data_t"){
data :union $C.mapuniontag("table") {
schedjobs @0: SchedJobData $C.mapname("jobs");
schedworkers @1: WorkerData $C.mapname("workers");
jobs @2: JobData $C.mapname("jobs");
workers @3: WorkerData $C.mapname("workers");
resources @4: ResourceData $C.mapname("resources");
resvals @5: ResourceVals $C.mapname("resvals");
jobArray @6: JobSubmit $C.mapname("job_array");
}
}
struct Get $C.mapname("xce_get_t") {
table @0: UInt32;
condition @1: MatchCondition;
}
struct ResValList $C.mapname("xce_resval_list_t") {
vals @0: List(ResourceVal) $C.mapname("vals") $C.maplistcount("count");
}
struct ResUsage $C.mapname("xce_rusage_t") {
name @0: Text $C.mapname("resname");
value @1: Float64 $C.mapname("value");
}
struct JobRusages $C.mapname("xce_job_rusages_t") {
jobid @0: UInt64 $C.mapname("jobid");
pids @1: List(Int32) $C.mapname("pids") $C.maplistcount("npids");
pgids @2: List(Int32) $C.mapname("pgids") $C.maplistcount("npgids");
usages @3: List(ResUsage) $C.mapname("usages") $C.maplistcount("count");
}
struct Update $C.mapname("xce_update_data_t") {
data :union $C.mapuniontag("table") {
nodata @0: Void;
resvals @1: ResValList $C.mapname("resvals");
jobrusages @2: JobRusages $C.mapname("jobrusages");
}
}
struct ResAlloc $C.mapname("xce_alloc_t"){
name @0: Text $C.mapname("resname");
resvId @1: Text $C.mapname("reserve");
instid @2: UInt64;
value @3: Float64;
from @4: UInt64; # alloc start time
duration @5: Int64;
}
struct HostAlloc $C.mapname("xce_host_alloc_t"){
host @0: Text;
allocs @1: List(ResAlloc) $C.mapname("allocs") $C.maplistcount("nallocs");
}
struct JobAlloc $C.mapname("xce_job_alloc_t"){
jobid @0: UInt64;
allocs @1: List(HostAlloc) $C.mapname("allocs") $C.maplistcount("nhosts");
allocsShared @2: List(ResAlloc) $C.mapname("allocs_shared") $C.maplistcount("nres_shared");
}
struct JobDispatch $C.mapname("xce_job_dispatch_t"){
allocs @0: List(JobAlloc) $C.maplistcount("njobs");
}
struct HostLaunch $C.mapname("xce_host_launch_t") {
host @0: Text $C.mapname("host");
nranks @1: UInt32 $C.mapname("nranks");
}
struct JobLaunch $C.mapname("xce_launch_t") {
hosts @0: List(HostLaunch) $C.mapname("hosts") $C.maplistcount("nhosts");
}
struct JobExecInfo $C.mapname("xce_job_exec_t"){
jobid @0: UInt64;
startinfo @1: JobStartInfo $C.mapname("start_info");
launchinfo @2: JobLaunch $C.mapname("launchinfo");
}
struct JobFinishInfo $C.mapname("xce_jobfinish_t") {
jobid @0: UInt64;
code @1: Int32 $C.mapname("exit_code");
}
struct TaskPidInfo $C.mapname("xce_task_pidinfo_t") {
jobid @0: UInt64 $C.mapname("jobid");
host @1: Text $C.mapname("host");
taskidx @2: Int32 $C.mapname("taskidx");
pids @3: List(UInt32) $C.mapname("pids") $C.maplistcount("npids");
pgids @4: List(UInt32) $C.mapname("pgids") $C.maplistcount("npgids");
exdata @5: Text $C.mapname("exdata");
clientidx @6: UInt32 $C.mapname("clientidx");
}
struct TaskFailInfo $C.mapname("xce_task_failinfo_t") {
jobid @0: UInt64 $C.mapname("jobid");
host @1: Text $C.mapname("host");
taskidx @2: Int32 $C.mapname("taskidx");
code @3: Int32 $C.mapname("code");
detail @4: Text $C.mapname("detail");
clientidx @5: Int32 $C.mapname("clientidx");
}
struct TaskFinish $C.mapname("xce_task_finish_t") {
jobid @0: UInt64 $C.mapname("jobid");
host @1: Text $C.mapname("host");
taskidx @2: UInt32 $C.mapname("taskidx");
exitcode @3: Int32 $C.mapname("exit_code");
clientidx @4: Int32 $C.mapname("clientidx");
}
struct NotifyData $C.mapname("xce_notify_data_t"){
data :union $C.mapuniontag("event") $C.mapname("data"){
jobPidInfo @0: JobPIDInfo $C.mapname("pidinfo");
jobFailInfo @1: JobFail $C.mapname("failinfo");
jobDispatchFail @2: JobFail $C.mapname("dispatchfail");
jobFinishInfo @3: JobFinishInfo $C.mapname("finishinfo");
schedStartJob @4: UInt64 $C.mapname("sched_start_job");
# bug-1737: was a bare UInt64 jobid -- carried no exit_code, so
# sched_handle_notify() (scheduler/src/handler.c) could not distinguish
# a successful finish (DONE) from a failed one (EXIT) and unconditionally
# assumed EXIT. Reuses the existing JobFinishInfo shape (jobid + code)
# apiserver_job_exit_fun (apiserver.c) already sends for this identical
# HOOK_DETAIL_JOB_EXIT hook, via NotifyData_data_jobFinishInfo.
schedExitJob @5: JobFinishInfo $C.mapname("sched_exit_job");
schedFailJob @6: UInt64 $C.mapname("sched_fail_job");
schedUpdateJob @7: SchedJob $C.mapname("sched_update_job");
schedUpdateResvals @8: ResValList $C.mapname("sched_update_resvals");
stoppedJob @9: UInt64 $C.mapname("stopped_job");
resumedJob @10: UInt64 $C.mapname("resumed_job");
ckptinfo @11: CheckpointInfo $C.mapname("ckptinfo");
schedUpdateQueue @12: QueueConf $C.mapname("sched_update_queue");
taskPidInfo @13: TaskPidInfo $C.mapname("taskinfo");
taskFailInfo @14: TaskFailInfo $C.mapname("taskfailinfo");
taskFinish @15: TaskFinish $C.mapname("taskfinish");
postDoneJob @16: UInt64 $C.mapname("pdone_job");
schedJobRusages @17: SchedJobRusages $C.mapname("sched_job_rusages");
updateJob @18: Job $C.mapname("update_job");
}
}
struct CtrlPeer $C.mapname("xce_ctrlpeer_t") {
name @0: Text;
host @1: Text;
port @2: UInt32;
}
struct CtrlStatus $C.mapname("xce_ctrlstatus_t") {
clusterName @0: Text;
primaryName @1: Text;
responderName @2: Text;
responderRole @3: UInt8;
term @4: UInt64;
configVersion @5: UInt64;
configChecksum @6: Text;
committedConfigVersion @7: UInt64;
committedLsn @8: UInt64;
appliedLsn @9: UInt64;
fenced @10: Bool;
peers @11: List(CtrlPeer) $C.mapname("peers.list") $C.maplistcount("peers.count");
peerCount @12: UInt32;
responderTimeMs @13: UInt64 $C.mapname("responder_time_ms"); # responder's wall-clock at sample time (epoch milliseconds)
}
struct ClusterConf $C.mapname("xce_clusterconf_t") {
name @0: Text;
resources @1: List(Resource) $C.mapname("reslist.res") $C.maplistcount("reslist.count");
queues @2: List(QueueConf) $C.mapname("qlist.queues") $C.maplistcount("qlist.count");
workers @3: List(Worker) $C.mapname("wlist.workers") $C.maplistcount("wlist.count");
}
struct JobSubmitData $C.mapname("job_submit_data_t") {
queue @0: Text;
user @1: Text;
mintasks @2: UInt32;
maxtasks @3: UInt32;
jobname @4: Text $C.mapname("job_name");
project @5: Text $C.mapname("project_name");
ugroup @6: Text $C.mapname("user_group");
description @7: Text $C.mapname("job_description");
resreq @8: Text;
startinfo @9: JobStartInfo $C.mapname("start_info");
fromHost @10: Text $C.mapname("from_host");
}
struct JobSubmit $C.mapname("xce_job_submit_t") {
jobid @0: UInt64;
arraySize @1: UInt64 $C.mapname("array_size");
submit @2: JobSubmitData;
}
struct JobSignal $C.mapname("xce_signal_t") {
signo @0: UInt32 $C.mapname("signo");
force @1: Int32 $C.mapname("force");
}
struct CheckpointOptions $C.mapname("xce_ckpt_options_t"){
kill @0: UInt32 $C.mapname("kill");
method @1: Text $C.mapname("method");
}
struct CheckpointInfo $C.mapname("xce_ckptinfo_t") {
jobid @0: UInt64;
ckptinfo @1: Text;
}
struct JobControl $C.mapname("xce_jobctrl_t") {
jobid @0: UInt64 $C.mapname("jobid");
u :union $C.mapuniontag("action") $C.mapname("u") {
siginfo @1: JobSignal;
ckptoption @2: CheckpointOptions;
}
}
struct QueueControl $C.mapname("xce_queuectrl_t") {
action @0: UInt32 $C.mapname("action");
queue @1: Text $C.mapname("queue");
}
struct HostControl $C.mapname("xce_hostctrl_t") {
action @0: UInt32 $C.mapname("action");
hostname @1: Text $C.mapname("hostname");
}
struct RequestReply $C.mapname("xce_reply_t"){
seq @0: UInt64;
opcode @1: OPCode;
status @2: Int32;
data : union {
nodata @3: Void;
debugmsg @4: Text;
getdata @5: XceData;
jobid @6: UInt64;
clusterconf @7: ClusterConf;
replconfigreplymsg @8: ReplConfigReply;
replconfigpushreplymsg @9: ReplConfigPushReply;
upgradeconfigreplymsg @10: UpgradeConfigReply;
ctrlhelloreplymsg @13: CtrlHelloReply;
requestvotereplymsg @14: RequestVoteReply;
stateReportMsg @15: StateReport $C.mapname("state_report"); # heartbeat reply: standby state
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;
probeconfigtagreplymsg @20: ProbeConfigTagReply;
}
redirectHost @11: Text $C.mapname("redirect_host");
redirectPort @12: UInt16 $C.mapname("redirect_port");
}
struct TaskHost $C.mapname("xce_taskhost_t") {
host @0: Text;
ntasks @1: UInt32;
exechome @2: Text;
execcwd @3: Text;
execuser @4: Text;
cmdline @5: Text;
exdata @6: Text;
launchtype @7: Text $C.mapname("launch_type");
}
struct RunTask $C.mapname("xce_taskrun_t") {
clientidx @0: Int32 $C.mapname("clientidx");
hosts @1: List(TaskHost) $C.mapname("taskhosts") $C.maplistcount("nhosts");
jobid @2: UInt64 $C.mapname("jobid");
}
struct PendReason $C.mapname("xce_pendreason_t") {
jobid @0: UInt64 $C.mapname("jobid");
reasons @1: JobPendReason $C.mapname("reasons");
}
struct PendReasonSet $C.mapname("xce_preason_set_t") {
reason @0: List(PendReason) $C.mapname("reasons") $C.maplistcount("nreasons");
}
struct ClearPReason $C.mapname("xce_clear_reason_t") {
jobs @0: List(UInt64) $C.mapname("jobs") $C.maplistcount("njobs");
}
struct FetchFileMsg $C.mapname("xce_fetchfile_t") {
jobid @0: UInt64 $C.mapname("jobid");
filepath @1: Text $C.mapname("filepath");
offset @2: UInt64 $C.mapname("offset");
maxBytes @3: UInt32 $C.mapname("max_bytes");
}
struct FetchFileReply $C.mapname("xce_fetchfile_reply_t") {
error @0: Int32 $C.mapname("error");
content @1: List(UInt8) $C.mapname("content") $C.maplistcount("bytes_read");
fileSize @2: UInt64 $C.mapname("file_size");
bytesRead @3: UInt32 $C.mapname("bytes_read");
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;
from @2: PeerInfo;
body : union {
nomsg @3: Void;
replymsg @4: RequestReply;
hellomsg @5: Text;
joinmsg @6: JoinData;
configpushmsg @7: ConfigData;
pushmsg @8: XceData;
getmsg @9: Get;
datamsg @10: Dataset;
dispatchmsg @11: JobDispatch;
jobstartmsg @12: JobExecInfo;
notifymsg @13: NotifyData;
jobsubmsg @14: JobSubmit;
updatemsg @15: Update;
jobctrlmsg @16: JobControl;
queuectrlmsg @17: QueueControl;
runtaskmsg @18: RunTask;
preasonmsg @19: PendReasonSet;
clrpreasonmsg @20: ClearPReason;
hostctrlmsg @21: HostControl;
replconfigrequestmsg @22: ReplConfigRequest;
replconfigpushrequestmsg @23: ReplConfigPushRequest;
upgradeconfigrequestmsg @24: UpgradeConfigRequest;
heartbeatmsg @25: HeartbeatMsg;
ctrlhellomsg @26: CtrlHelloRequest;
requestvotemsg @27: RequestVoteRequest;
walcatchupentrymsg @28: WalCatchupEntry;
snapshottransfermsg @29: SnapshotTransfer;
fetchfilemsg @30: FetchFileMsg;
schedadminmsg @31: SchedAdminRequest;
probeconfigtagrequestmsg @32: ProbeConfigTagRequest;
}
}
struct HostConfig $C.mapname("xce_hostconfig_t") {
name @0: Text $C.mapname("name");
model @1: Text $C.mapname("hostmodel");
type @2: Text $C.mapname("hosttype");
mxj @3: UInt32 $C.mapname("mxj");
resources @4: List(ConfigVar) $C.mapname("resources") $C.maplistcount("nresources");
thresholds @5: List(Threshold) $C.mapname("thresholds") $C.maplistcount("nthresholds");
dispatchwindows @6: TimeWindow $C.mapname("dispatchwindow");
}
struct ScalerConfig $C.mapname("scaler_config_t") {
name @0: Text $C.mapname("name");
type @1: Text $C.mapname("prov_type");
confPath @2: Text $C.mapname("conf_path");
scriptPath @3: Text $C.mapname("script_path");
scriptOption @4: Text $C.mapname("script_option");
}
struct LimitConfig $C.mapname("xce_limit_config_t") {
name @0: Text $C.mapname("name");
users @1: GroupConfig $C.mapname("users");
queues @2: GroupConfig $C.mapname("queues");
projects @3: GroupConfig $C.mapname("projects");
hosts @4: GroupConfig $C.mapname("hosts");
val @5: LimitValue $C.mapname("val");
perUser @6: Bool $C.mapname("per_user");
perQueue @7: Bool $C.mapname("per_queue");
perProject @8: Bool $C.mapname("per_project");
perHost @9: Bool $C.mapname("per_host");
}
struct ClusterConfig $C.mapname("cluster_config_t") {
clusterName @0: Text $C.mapname("clustername");
heartbeatInterval @1: UInt32 $C.mapname("heartbeat_interval");
inactiveHeartbeatCount @2: UInt32 $C.mapname("inactive_heartbeat_count");
fetchJobInterval @3: UInt32 $C.mapname("fetch_job_interval");
netioTimeout @4: UInt32 $C.mapname("netio_timeout");
resources @5: List(Resource) $C.mapname("resources") $C.maplistcount("nres");
queues @6: List(QueueConf) $C.mapname("queues") $C.maplistcount("nqueues");
ugroups @7: List(GroupConfig) $C.mapname("ugroups") $C.maplistcount("nugroups");
hgroups @8: List(GroupConfig) $C.mapname("hgroups") $C.maplistcount("nhgroups");
hostConfigs @9: List(HostConfig) $C.mapname("hconfigs") $C.maplistcount("nhconfigs");
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");
}
# ── Persistence configuration (Part 2.6) ────────────────────────────
# $C.mapname uses capnp_persist_cfg_t (not persist_config_t) to avoid
# collision with the plain-C struct already defined in persist.h.
struct PersistConfig $C.mapname("capnp_persist_cfg_t") {
walSyncMs @0 :UInt32 = 10;
snapIntervalS @1 :UInt32 = 60;
snapWalSizeMb @2 :UInt32 = 64;
finishedRetainS @3 :UInt32 = 3600;
workerRetainS @4 :UInt32 = 86400;
syncTimeoutMs @5 :UInt32 = 50;
snapKeepN @6 :UInt32 = 2;
transientJobTimeoutS @7 :UInt32 = 300;
}
# ── Snapshot header (Part 2.5) ───────────────────────────────────────
struct SnapshotHeader $C.mapname("capnp_snap_hdr_t") {
lsn @0 :UInt64;
term @1 :UInt64;
timestamp @2 :UInt64;
version @3 :UInt32 = 1;
nJobs @4 :UInt32;
nQueues @5 :UInt32;
nWorkers @6 :UInt32;
checksum @7 :UInt32;
headerChecksum @8 :UInt32;
}
# ── Upgrade snapshot admin-state frames (config-upgrade-snapshot task 1,
# Req 1.9/5.7) ──────────────────────────────────────────────────────────
# Independently-extensible Cap'n Proto frames for the on-disk checkpoint's
# host/queue admin-state sections (SNAPSHOT_FORMAT_VERSION 2, persist.c).
# New admin-state fields (e.g. a lock reason or tag) get new ordinals under
# capnp's own schema-evolution rules -- no future SNAPSHOT_FORMAT_VERSION
# bump needed per field addition. Deliberately NOT mapname'd onto
# Worker/QueueConf (design.md Design Invariant 5): those live runtime
# structs bundle static attrs and dynamic state together, which a snapshot
# frame must never carry (Requirement 1.7 excludes dynamic fields). Each
# maps instead onto its own frame-only C counterpart in typemap.h, matching
# the existing capnp_snap_hdr_t/capnp_persist_cfg_t/capnp_repl_envelope_t
# convention already established below for schema-only wire concepts with
# no corresponding "live" C struct.
struct HostAdminState $C.mapname("capnp_host_admin_state_t") {
name @0: Text;
status @1: UInt32;
}
struct QueueAdminState $C.mapname("capnp_queue_admin_state_t") {
name @0: Text;
status @1: UInt32;
}
# ── HA heartbeat ────────────────────────────────────────────────────
struct HeartbeatMsg $C.mapname("heartbeat_msg_t") {
senderPriority @0: UInt32 $C.mapname("sender_priority");
term @1: UInt64 $C.mapname("term");
committedLsn @2: UInt64 $C.mapname("committed_lsn");
committedConfigTag @3: UInt64 $C.mapname("committed_config_tag");
}
# ── Config replication message structs ──────────────────────────────
struct ReplConfigRequest $C.mapname("repl_config_req_t") {
versionTag @0: UInt64 $C.mapname("version_tag");
}
struct ReplConfigReply $C.mapname("repl_config_reply_t") {
upToDate @0: Bool $C.mapname("up_to_date");
versionTag @1: UInt64 $C.mapname("version_tag");
sha256 @2: Data $C.mapname("sha256");
bytes @3: Data $C.mapname("bytes");
}
struct ReplConfigPushRequest $C.mapname("repl_config_push_req_t") {
versionTag @0: UInt64 $C.mapname("version_tag");
sha256 @1: Data $C.mapname("sha256");
bytes @2: Data $C.mapname("bytes");
}
struct ReplConfigPushReply $C.mapname("repl_config_push_reply_t") {
accepted @0: Bool $C.mapname("accepted");
versionTag @1: UInt64 $C.mapname("version_tag");
# raft-adapter-conformance Requirement 7.4: the replying member's current
# term, so a higher term can be discovered through this reply channel the
# same way it already is through heartbeat and WAL-NACK replies. Additive
# field (ordinal 2) — an old-code standby's reply decodes as term=0 on a
# new-code primary, which can never look like a higher term (terms start
# at 1), so no spurious adoption occurs under mixed-version rollout.
term @2: UInt64 $C.mapname("term");
}
struct UpgradeConfigRequest $C.mapname("upgrade_config_req_t") {
sha256 @0: Data $C.mapname("sha256");
bytes @1: Data $C.mapname("bytes");
}
struct UpgradeConfigReply $C.mapname("upgrade_config_reply_t") {
versionTag @0: UInt64 $C.mapname("version_tag");
error @1: Text $C.mapname("error");
ackedCount @2: UInt32 $C.mapname("acked_count");
}
# ── WAL replication envelope (Part 6 — NOT the on-disk format) ───────
# Named ReplEnvelope to avoid collision with the on-disk wal_entry_hdr_t.
struct ReplEnvelope $C.mapname("capnp_repl_envelope_t") {
term @0 :UInt64;
lsn @1 :UInt64;
entryType @2 :UInt32;
payload @3 :Data;
crc @4 :UInt32;
prevCrc @5 :UInt32;
}
# ── HA election and replication message structs (Task 1.1) ───────────
# Heartbeat reply state report — standby piggybacks its replication state
# onto every OPCode_nop reply so the primary can drive convergence decisions.
struct StateReport $C.mapname("state_report_t") {
appliedLsn @0 :UInt64 $C.mapname("applied_lsn"); # standby's last applied WAL LSN
configTag @1 :UInt64 $C.mapname("config_tag"); # standby's current config version tag
term @2 :UInt64 $C.mapname("term"); # standby's current raft term (bug-1421 term-discovery)
}
struct RequestVoteRequest $C.mapname("request_vote_req_t") {
term @0 :UInt64 $C.mapname("term");
candidateId @1 :Text $C.mapname("candidate_id");
configTag @2 :UInt64 $C.mapname("config_tag");
lastLogLsn @3 :UInt64 $C.mapname("last_log_lsn");
# Raft (lastLogTerm, lastLogIndex) freshness tuple (ha-term-safety task 1.2,
# Req 2.3/2.4/2.6). Distinct from the XCE freshness fields above: configTag /
# lastLogLsn drive the application-policy freshness gate (on_vote_eligible),
# while these two feed the library's logOk gate (raft.tla HandleRequestVoteRequest
# logOk). lastLogIndex == candidate WAL log_len; lastLogTerm == term of that entry.
lastLogTerm @4 :UInt64 $C.mapname("last_log_term");
lastLogIndex @5 :UInt64 $C.mapname("last_log_index");
}
struct RequestVoteReply $C.mapname("request_vote_reply_t") {
term @0 :UInt64 $C.mapname("term");
voteGranted @1 :Bool $C.mapname("vote_granted");
peerConfigTag @2 :UInt64 $C.mapname("peer_config_tag");
peerLastLogLsn @3 :UInt64 $C.mapname("peer_last_log_lsn");
}
struct ProbeConfigTagRequest $C.mapname("probe_config_tag_req_t") {
name @0 :Text $C.mapname("name");
}
struct ProbeConfigTagReply $C.mapname("probe_config_tag_reply_t") {
configTag @0 :UInt64 $C.mapname("config_tag");
}
struct WalAckRequest $C.mapname("wal_ack_req_t") {
term @0 :UInt64 $C.mapname("term");
lsn @1 :UInt64 $C.mapname("lsn");
}
struct WalAckReply $C.mapname("wal_ack_reply_t") {
term @0 :UInt64 $C.mapname("term");
lsn @1 :UInt64 $C.mapname("lsn");
success @2 :Bool $C.mapname("success");
}
struct UpgradeConfigAck $C.mapname("upgrade_config_ack_t") {
term @0 :UInt64 $C.mapname("term");
ackedTag @1 :UInt64 $C.mapname("acked_tag");
success @2 :Bool $C.mapname("success");
}
# ── Snapshot chunk transfer (Req 4.1–4.2) ───────────────────────────
# Used for OPCode_snapshotTransfer: full state snapshot transferred as
# a sequence of fixed-size chunks (64 KB max each). Restart-on-reconnect
# semantics — no resumption state needed. Fields @0 and @1 are preserved
# for wire-protocol continuity; @2–@5 carry the chunk envelope.
struct SnapshotTransfer $C.mapname("snapshot_transfer_t") {
snapshotLsn @0 :UInt64 $C.mapname("snapshot_lsn"); # checkpoint LSN this snapshot corresponds to
payload @1 :Data $C.mapname("payload"); # chunk payload (64 KB max)
chunkIndex @2 :UInt32 $C.mapname("chunk_index"); # 0-based chunk number
chunkOffset @3 :UInt64 $C.mapname("chunk_offset"); # byte offset in the full snapshot
totalSize @4 :UInt64 $C.mapname("total_size"); # total snapshot size in bytes
isLast @5 :Bool $C.mapname("is_last"); # true for the final chunk
# raft-ha-core-safety task 3.2 (Req 3.1/5.1): the leader's term for the
# checkpoint LSN (persist_entry_term(p, snapshotLsn) at send time). The
# receiver reconciles this into its own per-LSN term index after install so
# the snapshot boundary establishes a verifiable (term, lsn) pair and any
# divergent stale-term pre-snapshot entry at this LSN is authoritatively
# overwritten. Append-only field number (@6) — no collision with @0-@5
# above. 0 = unset (legacy sender that never stamped this field) — the
# receiver treats 0 as WAL_LEGACY_TERM, the same convention persist's
# existing legacy-0 term mapping already uses.
checkpointTerm @6 :UInt64 $C.mapname("checkpoint_term");
}
struct WalCatchupEntry $C.mapname("wal_catchup_entry_t") {
lsn @0 :UInt64 $C.mapname("lsn");
entryType @1 :UInt8 $C.mapname("entry_type");
# payload is List(UInt8), NOT Data (bug-1505 fix, same class as bug-1504 on
# SnapshotData.internalState -- see that struct's own comment for the full
# root cause: this codebase's capnpc-c backend silently no-ops every Data
# field's encoding in generated encode_XXX() helpers without zero-
# initializing the caller's local struct first, leaving a genuinely
# uninitialized capn_ptr written onto the wire -- a real, reproducible
# SIGSEGV in write_WalCatchupEntry, hit by
# test_wal_catchup_append_term_fields_round_trip even though that test
# never touches payload content at all (src.payload stays NULL -- the
# crash is from the OUTPUT struct's uninitialized field, independent of
# the input). List(UInt8) gets a real generated encode/decode from this
# same backend (proven precedent: FetchFileReply.content/bytesRead), so
# no hand-written accessor is needed. Reuses wirePayloadLen (@9, below)
# as its maplistcount companion -- already exactly "this sub-entry's
# payload byte count" for the bug-847 slicing use below, same value.
# NOTE: production (xce_wal_batch_encode_raw/wal_entry_to_wire, request.c)
# never goes through this generic encode_WalCatchupEntry() path at all --
# it memset()s its own struct and calls write_WalCatchupEntry() directly
# with every field hand-set (payload deliberately left at its post-memset
# zero value, per bug-847's own comment below) -- so this fix has zero
# production behavior impact, it only stops the generic codec (used by
# this one test) from crashing.
payload @2 :List(UInt8) $C.mapname("payload") $C.maplistcount("wire_payload_len");
entries @3 :List(WalCatchupEntry) $C.mapname("entries") $C.maplistcount("n_entries");
# Raft AppendEntries term fields (ha-term-safety task 1.1, Req 1.3/1.4).
# This envelope is the replication-carrying message (WAL catch-up); the four
# log callbacks (tasks 2.x) need prevLog{Index,Term} for the AppendEntries
# logOk / conflict-truncation decision and `term` to stamp the entry. lsn ==
# this entry's 1-based log index; prevLogIndex == lsn-1; prevLogTerm == term of
# the entry at prevLogIndex (0 if absent); term == this entry's term stamp.
term @4 :UInt64 $C.mapname("term");
prevLogIndex @5 :UInt64 $C.mapname("prev_log_index");
prevLogTerm @6 :UInt64 $C.mapname("prev_log_term");
# ha-term-safety task 3.2 (Req 2.5/4.1): the leader's commit index, carried so
# the standby intake passes it into raft_on_append_entries_request. Append-only
# field number (@7) — no collision with the @4/@5/@6 term fields above.
leaderCommit @7 :UInt64 $C.mapname("leader_commit");
# bug-1067: the streaming leader's CURRENT term — the Raft AppendEntries header
# term (`term` arg of raft_on_append_entries_request), DISTINCT from `term` @4
# which is the entry's STORED term. For a live append the two are equal; for
# catch-up of a HISTORICAL entry written by a prior leader, the entry's stored
# term is LOWER than the current leader's term. Using the stored term as the AE
# header term made the library stale-reject every historical entry
# (ae_reject_stale), wedging cross-term catch-up. The standby intake uses this
# as the AE header term (UpdateTerm/stale-check) and keeps `term` @4 as the
# per-entry term (logOk / on_persist_entry). 0 = unset (legacy sender) → the
# intake falls back to the entry's stored term (the prior, live-only behaviour).
leaderCurrentTerm @8 :UInt64 $C.mapname("leader_current_term");
# bug-847: a PER-ELEMENT `payload` (@2) Data blob loses its pointer when a
# multi-entry catch-up batch spans more than one capnp segment (the
# underlying capnp-c library's far-pointer handling for a List(UInt8) whose
# segment differs from its referencing struct's segment) -- the receiver
# decodes raw_payload_len=0 for entries at/after the boundary even though
# the sender held the real bytes, silently persisting empty WAL entries.
# A SINGLE top-level Data blob does not exhibit this loss (only scattered
# per-element Data does), so the batch encoder (xce_wal_batch_encode_raw)
# now carries every sub-entry's payload concatenated in ONE top-level
# batchPayloads blob, sliced back apart on decode using each sub-entry's
# own wirePayloadLen (a plain scalar -- already immune, decoded like
# lsn/entryType/term by the generic generated decode, no manual extraction
# needed). Populated for every xce_wal_batch_encode_raw send regardless of
# n_entries (0/1/N); sub-entries no longer carry their own @2 payload on
# this path (kept for any other/legacy sender; decode still handles it
# defensively as a fallback when batchPayloads is absent).
#
# NOTE (bug-1505 fix, same as `payload` above): this note's own "cross-
# segment loss" concern is about scattered PER-ELEMENT Data (or
# List(UInt8) -- identical wire representation either way in this
# codegen; the Data vs List(UInt8) schema keyword only changes whether
# THIS backend's generic encode_XXX() composite helper populates the
# field automatically, not the underlying wire type). batchPayloads
# itself is the single TOP-LEVEL blob this fix already routes around that
# exact problem with -- so switching ITS OWN keyword to List(UInt8) does
# not reintroduce bug-847's issue; it only stops the generic codec's
# SnapshotData.internalState-class no-op/uninitialized-capn_ptr crash
# (bug-1504) for this field too. batchPayloadsLen (@11, below) is its
# maplistcount companion, distinct from raw_batch_payloads_len
# (typemap.h) -- that field is populated by request.c's OWN hand-rolled
# xce_decode() extraction (a completely separate, already-correct code
# path this fix does not touch), not by this generic decode.
wirePayloadLen @9 :UInt32 $C.mapname("wire_payload_len");
batchPayloads @10 :List(UInt8) $C.mapname("batch_payloads") $C.maplistcount("batch_payloads_wire_len");
batchPayloadsLen @11 :UInt32 $C.mapname("batch_payloads_wire_len");
}
struct CtrlHelloRequest $C.mapname("ctrl_hello_req_t") {
name @0 :Text $C.mapname("name");
key @1 :Data $C.mapname("key");
appliedLsn @2 :UInt64 $C.mapname("applied_lsn");
configTag @3 :UInt64 $C.mapname("config_tag");
}
struct CtrlHelloReply $C.mapname("ctrl_hello_reply_t") {
committedLsn @0 :UInt64 $C.mapname("committed_lsn");
needsSnapshot @1 :Bool $C.mapname("needs_snapshot");
configTag @2 :UInt64 $C.mapname("config_tag");
}
# ── Replay stream envelope schema (scheduler-core Group 7 task 7.1,
# Req 5 AC 20) ────────────────────────────────────────────────────────────
# `.replay` file format: a single stream of consecutive capnp messages --
# ReplayHeader (msg 1) -> SnapshotEntry (msg 2, MUST be the first data
# message) -> ReplayEvent/ReplaySession cycles -> periodic SnapshotEntry
# checkpoints -> final ReplayEvent{endSession}. See requirements.md Req 5
# AC 20 for the full framing prose; this schema captures only the message
# *shapes* -- the file-level reader/writer that produces/consumes the
# stream is task 9+, entirely out of Group 7's (schema-only) scope.
#
# Struct-ownership split across Group 7 tasks, so a fresh implementer on
# 7.2/7.3/7.4 knows exactly what to add and where. Every later addition is a
# NEW ordinal on a struct THIS task already created -- nothing below is ever
# renumbered or restructured by a later task:
# - task 7.1 (this task): ReplayCluster, ReplayHeader, SnapshotData
# (zero-field placeholder), SnapshotEntry, ReplayEvent (9 members, all
# reusing pre-existing structs), SessionSummary (fully real),
# ReplayDecision (ts-only stub), ReplaySession.
# - task 7.2: adds the new JobStateChangeEvent + JobUpdateEvent structs,
# then adds two new union members (jobStateChange, jobUpdate) to
# ReplayEvent's `body` union at new ordinals (10, 11) -- purely additive.
# - task 7.3 (DONE): added SnapshotData's real 11 fields (Req 5 AC 21)
# starting at ordinal @0, replacing task 7.1's zero-field placeholder
# wholesale (both here and in typemap.h's sched_snapshot_data_t). Added
# DispatchDecision/PendDecision/PreemptVictim/PreemptDecision, then
# added `body :union { dispatch @1: DispatchDecision; pend @2:
# PendDecision; preempt @3: PreemptDecision; }` to ReplayDecision --
# purely additive new ordinals on the ts@0 struct task 7.1 created (see
# the JobControl/JoinData `:union` group precedent above: a union group
# name never consumes its own ordinal, only its member variants do).
# DEVIATION from design.md's literal text: SnapshotData also needed two
# new user/project wire types. design.md's literal spec names the
# mapname target `xce_sched_user_t`/`xce_sched_project_t` (the
# scheduler-core-internal C types in src/scheduler/include/sched/
# types.h) -- but that header is scheduler-private, unreachable from
# typemap.h, and already `#include`s typemap.h itself, so reusing that
# exact name here would be a guaranteed duplicate-typedef compile error
# across src/scheduler/. Used fresh mapnames `sched_snapshot_user_t`/
# `sched_snapshot_project_t` instead (see `struct SchedUser` comment
# below for the full rationale) -- field lists still match design.md's
# spec exactly, only the C type name differs.
struct ReplayCluster $C.mapname("replay_cluster_t") {
name @0: Text;
controllers @1: List(Text) $C.maplistcount("n_controllers"); # ["host1:9991", "host2:9992", "host3:9993"]
recordingHost @2: Text $C.mapname("recording_host"); # controller hostname that made the recording
}
struct ReplayHeader $C.mapname("replay_header_t") {
recorderVersion @0: Text $C.mapname("recorder_version"); # scheduler binary version that wrote this file
configHash @1: Data $C.mapname("config_hash"); # sha256 of xce_conf4sched_t
cluster @2: ReplayCluster;
recordingStartTs @3: UInt64 $C.mapname("recording_start_ts"); # logical time recording started (ns)
recordingStartWall @4: Text $C.mapname("recording_start_wall"); # wall-clock time recording started (ISO 8601)
schedPolicy @5: Text $C.mapname("sched_policy"); # active picker plugin name (e.g. "builtin_fcfs")
}
# task 7.3 (Req 5 AC 21): the two scheduler-only user/project wire types
# SnapshotData embeds. design.md's literal spec reuses the scheduler-core
# core-internal C type names xce_sched_user_t/xce_sched_project_t
# (sched/types.h) verbatim as the mapname target -- but those types are
# scheduler-PRIVATE (sched/types.h is under src/scheduler/include/, out of
# this schema-only task's edit scope, and is NOT reachable from typemap.h,
# which every OTHER binary linking controller.capnp -- controller, apiserver
# CGO, xceadmin, worker -- also compiles against). sched/types.h itself
# `#include`s typemap.h (confirmed: scheduler/src/core_events.c and
# schedobj.c already include both headers in the same translation unit), so
# adding a second, textually-distinct `xce_sched_user_t`/`xce_sched_project_t`
# typedef to typemap.h would be a guaranteed duplicate-typedef compile error
# across most of src/scheduler/ the moment both headers are included
# together -- not hypothetical, a certainty given the current include graph.
# Field lists match design.md's literal spec exactly (verified against
# sched/types.h: xce_sched_user_t has name only; xce_sched_project_t has
# name+priority) -- only the C TARGET TYPE NAME differs, to stay reachable
# from typemap.h without touching src/scheduler/. This mirrors the
# PreemptVictim/PreemptDecision precedent below: a fresh, wire-only mapname
# distinct from a similarly-shaped core-internal type of a different name.
struct SchedUser $C.mapname("sched_snapshot_user_t") {
name @0: Text;
}
struct SchedProject $C.mapname("sched_snapshot_project_t") {
name @0: Text;
priority @1: Int32;
}
# task 7.3 (Req 5 AC 21): the full snapshot message type, replacing task
# 7.1's zero-field placeholder wholesale. Every domain-object field reuses
# a pre-existing wire struct (Conf4Sched/SchedJobData/QueueConf/WorkerData/
# ResourceData/ResValList, all already used by ReplayEvent above) except
# SchedUser/SchedProject (fresh, see comment above) and the two
# scheduler-internal-only opaque fields. internalState is a versioned
# opaque blob (shared-resource tracking + unsatisfiable-resreq cache +
# session/iterator state) with no capnp-visible structure -- the core owns
# its byte layout; internalVersion is a format tag so a future incompatible
# internal layout is rejected rather than misinterpreted (this task's
# Observable).
#
# internalState is List(UInt8), NOT Data (task 12.1 review finding, fixed):
# every OTHER Data field in this schema (e.g. ReplayHeader.configHash) has
# its generated encode_/decode_ codec silently no-op'd by this codebase's
# capnpc-c setup -- the generated encode_SnapshotData_ptr()'s local `struct
# SnapshotData d;` is never zero-initialized, so a no-op'd Data field left
# capn_setp() writing an UNINITIALIZED stack capn_ptr onto the wire (real,
# reproducible SIGSEGV in write_SnapshotData, deterministic per call site's
# stack layout -- configHash happened to get lucky garbage, internalState
# did not). List(UInt8) sidesteps the whole no-op class: this same capnpc-c
# backend DOES correctly auto-generate real encode/decode for List(UInt8)
# (proven precedent: FetchFileReply.content/bytesRead above -- a genuine
# capn_new_list8()/capn_set8() loop, not a comment placeholder), so no
# hand-written low-level accessor code is needed here at all. Same
# maplistcount convention as content/bytesRead: internalStateLen (@11,
# below) carries the byte count as its own real wire scalar.
struct SnapshotData $C.mapname("sched_snapshot_data_t") {
config @0: Conf4Sched;
jobs @1: SchedJobData;
queues @2: List(QueueConf) $C.maplistcount("n_queues");
workers @3: WorkerData;
resources @4: ResourceData;
resvals @5: ResValList;
users @6: List(SchedUser) $C.maplistcount("n_users");
projects @7: List(SchedProject) $C.maplistcount("n_projects");
prngSeed @8: UInt64 $C.mapname("prng_seed");
internalState @9: List(UInt8) $C.mapname("internal_state") $C.maplistcount("internal_state_len");
internalVersion @10: UInt32 $C.mapname("internal_version");
internalStateLen @11: UInt32 $C.mapname("internal_state_len");
}
struct SnapshotEntry $C.mapname("replay_snapshot_entry_t") {
ts @0: UInt64; # logical time of snapshot
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, 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") {
configPush @1: Conf4Sched $C.mapname("config_push");
jobsPush @2: SchedJobData $C.mapname("jobs_push");
workersPush @3: WorkerData $C.mapname("workers_push");
resourcesPush @4: ResourceData $C.mapname("resources_push");
resvalsFull @5: ResValList $C.mapname("resvals_full");
resvalsDelta @6: ResValList $C.mapname("resvals_delta");
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");
}
}
struct SessionSummary $C.mapname("session_summary_t") {
durationNs @0: UInt64 $C.mapname("duration_ns");
nDecisions @1: UInt32 $C.mapname("n_decisions");
nDispatched @2: UInt32 $C.mapname("n_dispatched");
nPended @3: UInt32 $C.mapname("n_pended");
}
# task 7.3 (Req 5 AC 5, AC 20): the three decision payload types. dispatch
# and pend reuse pre-existing wire structs verbatim (JobAlloc, JobPendReason
# -- design review already confirmed JobAlloc.jobid is DispatchDecision's
# sole job-identity carrier, no redundant jobid field). PreemptVictim is
# fresh (jobid+workerid has no pre-existing wire counterpart). PreemptVictim/
# PreemptDecision are DELIBERATELY distinct wire-only types from decision.h's
# core-internal xce_preempt_victim_t/xce_preempt_t (task 4.4's in-process
# live-decision representation) -- the wire format and the core's live
# in-process decision representation are independent, never-aliased things
# that happen to model similar data.
struct DispatchDecision $C.mapname("sched_dispatch_decision_t") {
alloc @0: JobAlloc;
}
struct PendDecision $C.mapname("sched_pend_decision_t") {
jobid @0: UInt64;
reason @1: JobPendReason;
}
struct PreemptVictim $C.mapname("preempt_victim_t") {
jobid @0: UInt64;
workerid @1: Text;
}
struct PreemptDecision $C.mapname("sched_preempt_decision_t") {
jobid @0: UInt64;
victims @1: List(PreemptVictim) $C.maplistcount("n_victims");
reason @2: JobPendReason;
}
# task 7.3 (Req 5 AC 20): purely additive `body` union on top of task 7.1's
# ts@0-only stub -- new ordinals 1-3, ts@0 untouched (see the JobControl/
# JoinData precedent noted above: a union group name never consumes its own
# ordinal).
struct ReplayDecision $C.mapname("replay_decision_t") {
ts @0: UInt64;
body :union $C.mapuniontag("body_which") {
dispatch @1: DispatchDecision;
pend @2: PendDecision;
preempt @3: PreemptDecision;
}
}
struct ReplaySession $C.mapname("replay_session_t") {
beginRound @0: UInt64 $C.mapname("begin_round"); # logical time (ts value)
decisions @1: List(ReplayDecision) $C.maplistcount("n_decisions");
summary @2: SessionSummary; # optional diagnostic metadata
}