lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add PersistConfig, SnapshotHeader, ReplEnvelope Cap'n Proto schemas (P0-9)

Browse source 
Extends src/schema/controller.capnp with persistence structures needed
for WAL replication and snapshots:

- PersistConfig (8 fields, @0-@7): capnp-serialisable copy of the
  persistence configuration; mapped to capnp_persist_cfg_t to avoid
  collision with the plain-C persist_config_t in persist.h
- SnapshotHeader (9 fields, @0-@8): snapshot manifest with LSN, term,
  timestamp, record counts and checksums; mapped to capnp_snap_hdr_t
- ReplEnvelope (6 fields, @0-@5): replication wire envelope (NOT the
  on-disk format); named to avoid collision with wal_entry_hdr_t and
  the WALEntry_ internal symbol c-capnproto generates; mapped to
  capnp_repl_envelope_t

Also extends existing structs:
- Job.transientSince @11 (UInt64=0, mapname transient_since): set when
  job enters JOB_STAT_TRANSIENT; also adds uint64_t transient_since to
  xce_job_t in types.h
- Worker.deadSince @6 (UInt64=0, mapname dead_since): mirrors existing
  time_t dead_since field in xce_worker_t

OPCode enum extended with 7 replication opcodes @22-@28:
  replEntry, replAck, redirect, heartbeat, steppingDown,
  replSnapshot, replCatchUpDone

New C typedefs (capnp_persist_cfg_t, capnp_snap_hdr_t,
capnp_repl_envelope_t) added to typemap.h for the encode/decode shims.

TDD: test_persist_schema.c added (6 tests, all GREEN). Controller test
suite remains 1/1 PASS.
This commit is contained in:
Rongsong Shen 2026-05-07 10:19:33 +00:00
parent 1fe79ff934
commit 4903c83f28
6 changed files with 214 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

79
src/schema/controller.capnp
View file

@ -30,6 +30,13 @@ enum OPCode {
runtasks @19;
setpreason @20;
clrpreason @21;
replEntry @22;
replAck @23;
redirect @24;
heartbeat @25;
steppingDown @26;
replSnapshot @27;
replCatchUpDone @28;
}
struct PeerInfo $C.mapname("xce_peerinfo_t") {
@ -114,12 +121,13 @@ struct HostAttrs $C.mapname("hostattrs_t") {
}
struct Worker $C.mapname("xce_worker_t"){
name @0: Text;
attrs @1: HostAttrs;
status @2: UInt32;
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(ValueWithIdx) $C.mapname("resvals_to_sched") $C.maplistcount("nresvals_to_sched");
thresholds @4: List(Threshold) $C.mapname("thresholds") $C.maplistcount("nthresholds");
resvals @5: List(ValueWithIdx) $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 {
@ -265,17 +273,18 @@ struct JobStartInfo $C.mapname("job_startinfo_t"){
}
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");
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"){
@ -666,3 +675,41 @@ struct ClusterConfig $C.mapname("cluster_config_t") {
hproviders @18: List(ScalerConfig) $C.mapname("hproviders") $C.maplistcount("nhproviders");
limits @19: List(LimitConfig) $C.mapname("limits") $C.maplistcount("nlimits");
}
# ── 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;
}
# ── 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;
}