diff --git a/src/docs/persistence-and-ha-design.md b/src/docs/persistence-and-ha-design.md new file mode 100644 index 00000000..6f471046 --- /dev/null +++ b/src/docs/persistence-and-ha-design.md @@ -0,0 +1,2060 @@ +# XCE Controller: State Persistence & High Availability Design + +> **Status**: Draft +> **Last Updated**: 2026-04-20 +> **Target**: `src/controller/` (C daemon) + +--- + +## Table of Contents + +1. [Part 1: Unused Code Cleanup](#part-1-unused-code-cleanup) +2. [Part 2: Single-Controller Persistence](#part-2-single-controller-persistence) +3. [Part 3: Checkpoint — Dual zarray COW](#part-3-checkpoint--dual-zarray-cow) +4. [Part 4: Deferred Entity Cleanup at Checkpoint](#part-4-deferred-entity-cleanup-at-checkpoint) +5. [Part 5: HA Failover — LSF-Style Priority Election](#part-5-ha-failover--lsf-style-priority-election) +6. [Part 6: Fan-Out TCP Replication](#part-6-fan-out-tcp-replication) +7. [Part 7: Mixed Sync/Async Replication Protocol](#part-7-mixed-syncasync-replication-protocol) +8. [Part 8: Hook-Driven Architecture](#part-8-hook-driven-architecture) +9. [Part 9: Worker Connection Strategy](#part-9-worker-connection-strategy) +10. [Part 10: libxce Smart Discovery](#part-10-libxce-smart-discovery) +11. [Implementation Phases](#implementation-phases) +12. [Appendix: Data Structures & Signatures](#appendix-data-structures--signatures) + +--- + +## Part 1: Unused Code Cleanup + +The controller has no database code. It uses `kvmgr` (in-memory hash tables) for all state. The following are dead stubs and unused code that should be removed. + +### 1.1 Dead Code in `job.c` + +| Location | Code | Action | +|----------|------|--------| +| `job.c:18` | `#define NJOBS (100)` | Remove — unused constant | +| `job.c:20` | `uint64_t jobid = 100;` | Remove — unused global | +| `job.c:63-89` | `load_jobs()` body inside `#if 0` | Remove dead block, keep function signature (will be rewired in Part 2) | + +### 1.2 Unused Function in `worker.c` + +| Location | Code | Action | +|----------|------|--------| +| `worker.c:367-391` | `restore_worker()` | **Keep** — currently unused but will be the persistence restore entry point (Part 2) | + +### 1.3 Empty Function in `cluster.c` + +| Location | Code | Action | +|----------|------|--------| +| `cluster.c:400` | `cluster_build_groups()` — empty body (`return 0;`) | Keep for now (groups are a future feature), but add a `/* TODO: implement group building */` comment | + +### 1.4 Stale Frontend References + +The Go API server already removed PostgreSQL (`lib/pq` removed from `go.mod`). Remaining stale references: + +- `src/frontend/docs/` — Update any references to `lib/pq` or PostgreSQL +- `src/frontend/apiserver/store/audit_test.go` — Already has forbidden-pattern checks for `DBConnPool`, `DBHandle`, `InitDB`, `QueryFromDB`; keep these as regression guards + +--- + +## Part 2: Single-Controller Persistence + +### 2.1 Architecture: WAL + Periodic Checkpoint + +``` + ┌──────────────┐ + State mutations ──▶│ Hook System │ + └──────┬───────┘ + │ AFTER_EVENT + ┌──────▼───────┐ + │ persist.c │ + │ (WAL Writer) │ + └──────┬───────┘ + │ + ┌────────────┼────────────┐ + ▼ ▼ ▼ + ┌──────────┐ ┌──────────┐ ┌──────────────┐ + │ WAL File │ │ Checkpoint│ │ Replication │ + │ (append) │ │ (COW snap)│ │ (Part 6) │ + └──────────┘ └──────────┘ └──────────────┘ +``` + +The design draws from etcd's architecture, simplified for XCE's single-writer model: + +| etcd Concept | XCE Adaptation | Why Simplified | +|--------------|----------------|----------------| +| WAL (append-only log) | Adopted directly | Proven durability pattern | +| Chained CRC | Adopted (links each record to previous) | Detects torn/truncated writes | +| Pre-allocated segments | Adopted (64 MB segments, rotate) | Avoids filesystem allocation latency | +| 8-byte alignment | Adopted | Direct I/O compatibility | +| BoltDB snapshot | **Not adopted** — use dual zarray COW | Application-level COW avoids fork() pitfalls | +| MVCC key-value store | **Not adopted** | Single-writer, no concurrent readers needing versioning | +| Raft consensus | **Not adopted** — LSF-style priority (Part 5) | ~300 LOC vs ~5000 LOC | + +### 2.2 WAL Entry Format + +```c +/* WAL entry header — 40 bytes, 8-byte aligned */ +typedef struct { + uint64_t term; /* Election term (incremented on failover) */ + uint64_t lsn; /* Log sequence number within this term */ + uint64_t timestamp; /* Epoch microseconds */ + uint32_t type; /* wal_entry_type enum */ + uint32_t payload_len; /* Bytes following this header */ + uint32_t crc; /* CRC32 of (header + payload) */ + uint32_t prev_crc; /* Previous record's CRC (chained) */ +} wal_entry_hdr_t; +``` + +### 2.3 WAL Entry Types + +```c +enum wal_entry_type { + /* Job lifecycle — the high-frequency path */ + WAL_JOB_SUBMIT = 1, /* SYNC replication */ + WAL_JOB_START = 2, /* ASYNC */ + WAL_JOB_FINISH = 3, /* ASYNC */ + WAL_JOB_FAIL = 4, /* ASYNC */ + WAL_JOB_STOP = 5, /* ASYNC */ + WAL_JOB_RESUME = 6, /* ASYNC */ + WAL_JOB_UPDATE = 7, /* ASYNC */ + WAL_JOB_RUSAGE = 8, /* ASYNC */ + WAL_JOB_CKPT = 9, /* ASYNC */ + WAL_JOB_PREASON = 10, /* ASYNC */ + + /* Queue and host control — infrequent */ + WAL_QUEUE_CTRL = 20, /* ASYNC */ + WAL_WORKER_CTRL = 30, /* ASYNC */ + + /* Marker */ + WAL_CHECKPOINT = 100 /* Marker — no payload */ +}; +``` + +**Note**: There are no `WAL_JOB_DELETE`, `WAL_WORKER_DELETE`, or `WAL_QUEUE_DELETE` entry types. Entity cleanup is deferred to checkpoint (see Part 4). This means WAL never contains delete records — only creates and updates — simplifying replay logic. + +**Mapping to existing hook detail enums:** + +| Hook Detail (`cluster_def.h`) | WAL Entry Type | Replication | +|-------------------------------|----------------|-------------| +| `HOOK_DETAIL_JOB_NEW` | `WAL_JOB_SUBMIT` | **SYNC** | +| `HOOK_DETAIL_JOB_START` | `WAL_JOB_START` | ASYNC | +| `HOOK_DETAIL_JOB_EXIT` | `WAL_JOB_FINISH` | ASYNC | +| `HOOK_DETAIL_JOB_FAIL` | `WAL_JOB_FAIL` | ASYNC | +| `HOOK_DETAIL_JOB_UPDATE` | `WAL_JOB_UPDATE` | ASYNC | +| `HOOK_DETAIL_JOB_CKPT` | `WAL_JOB_CKPT` | ASYNC | +| `HOOK_DETAIL_JOB_RUSAGE_UPDATE` | `WAL_JOB_RUSAGE` | ASYNC | +| `HOOK_DETAIL_SETPREASON` | `WAL_JOB_PREASON` | ASYNC | +| `HOOK_DETAIL_QUEUE_UPDATE` | `WAL_QUEUE_CTRL` | ASYNC | +| `HOOK_DETAIL_WORKER_UPDATE` | `WAL_WORKER_CTRL` | ASYNC | + +### 2.4 WAL Segment Management + +``` +wal_path/ +├── seg-0000000000000000.wal # Pre-allocated 64 MB +├── seg-0000000000000064.wal # LSN offset in filename +└── ... +``` + +- Segments are pre-allocated to 64 MB (configurable) +- When a segment is full, rotate to the next +- `fsync()` at configurable interval (`wal_sync_ms`, default 10ms) — batched, not per-write +- Segments before the latest snapshot are deleted during recovery +- Chained CRC detects corruption: if `prev_crc` doesn't match previous record's CRC, stop replay at last valid entry + +### 2.5 Recovery Sequence + +On startup, `cluster_init()` calls `persist_recover()`: + +1. **Find latest snapshot**: Scan `snap_path/` for the snapshot with the highest LSN +2. **Load snapshot**: Deserialize Cap'n Proto data, rebuild kvmgr instances + - Jobs → `makeJob(c->jobmgr, ...)` (reuse existing `create_job()`) + - Workers → `restore_worker(c, host)` (already exists in `worker.c:367`) + - Queues → `create_queue(c, qconf)` (already exists) + - Resources → `makeResource(c->resmgr, ...)` (already exists) + - Configs → `add_config(c, key, val)` (already exists) +3. **Replay WAL**: Starting from the snapshot's LSN, read and apply WAL entries forward +4. **Trim WAL**: Delete WAL segments fully covered by the loaded snapshot + +### 2.6 Configuration + +```c +typedef struct { + char wal_path[XCE_MAXPATH_LEN]; /* Directory for WAL segments */ + char snap_path[XCE_MAXPATH_LEN]; /* Directory for snapshots */ + uint32_t wal_sync_ms; /* fsync interval (default 10) */ + uint32_t snap_interval_s; /* checkpoint interval (default 60) */ + uint32_t snap_wal_size_mb; /* checkpoint when WAL exceeds this (default 64) */ + uint32_t finished_retain_s;/* retain finished jobs for N sec (default 3600) */ + uint32_t worker_retain_s; /* retain unavailable workers for N sec (default 86400) */ + uint32_t sync_timeout_ms; /* sync replication ACK timeout (default 50-100) */ +} persist_config_t; +``` + +Added to `cluster_config_t` in the Cap'n Proto schema and `myconf.c` YAML parsing. + +### 2.7 Controller Connection Context + +The controller only handles connections from three peer types: + +1. **Workers** — compute node daemons reporting job status and receiving dispatch +2. **Scheduler** — job placement service +3. **API servers** — Go REST frontend (1–3 instances) + +End users never connect directly to the controller — the API server provides traffic control, rate limiting, and request validation. This has important implications: + +- **Sync replication**: Only triggered by job submit from API servers. At 1–3 API server connections, vcpu pool exhaustion from blocking `repl_wait_ack()` is not a concern — the wait is bounded by `sync_timeout_ms` (default 50ms) and only affects the vcpu handling that specific API server's request. +- **Connection count**: Workers (100s–1000s) + scheduler (1–2) + API servers (1–3). The scheduler and API server connections are long-lived and stable. +- **All connections use SSL**: The controller's iomgr enforces SSL on all connections (see `netio.c`). Replication channels must also use this pipeline, not raw TCP. + +--- + +## Part 3: Checkpoint — Dual zarray COW + +### 3.1 Why Not fork()? + +The original design considered `fork()` + COW for checkpoint snapshots. This is problematic for a multithreaded controller: + +- After `fork()`, only the calling thread survives. If another thread held a mutex (e.g., the WAL writer held `persist.lock`), the child inherits a **locked mutex with no owner** — instant deadlock. +- kvmgr's internal `my_rwlock_t` is not fork-safe. The child must not take any lock, but serializing kvmgr state likely requires traversal that touches locked structures. +- Linux memory overcommit: if the parent modifies many pages during the child's serialization, the OS may OOM-kill the child (or parent). + +### 3.2 Design: Application-Level COW via Dual zarray + +XCE's kvmgr uses a two-level structure: `kvtab` (hash table for key→index lookup) + `data` (xzarray — dense pointer array for iteration and index-based access). This structure makes application-level COW practical. + +**Key property**: The hash table maps keys to **stable indices** in the zarray. COW only needs to happen at the zarray level. The kvtab never needs to be duplicated. + +``` +kvmgr structure: + kvtab: key ──hash──▶ index (stable, never changes for a given key) + data: xzarray of { inuse, data_ptr } indexed by index + +COW operation: + 1. Swap: data → shadow (frozen, immutable) + 2. New data = empty array with same capacity, all inuse=0 + 3. During COW: reads check data first, fall through to shadow + 4. After serialization: merge shadow back into data, apply GC +``` + +### 3.3 xzarray COW Extensions + +Add two fields to `xzarray_t`: + +```c +typedef struct { + int count; + size_t caps; + size_t delta; + int pos; + xzarray_item_t* data; /* Active array (COW writes go here) */ + xzarray_item_t* shadow; /* Frozen snapshot (NULL when not in COW) */ + size_t shadow_caps; /* Shadow array capacity */ + destroy_fn_t killer; +} xzarray_t; +``` + +When `shadow != NULL`, the array is in COW mode. The existing `xzarray_item_t` is unchanged: + +```c +typedef struct { + unsigned int inuse : 1; + void* data; +} xzarray_item_t; +``` + +### 3.4 Modified zarray Operations During COW + +**Critical constraint**: During COW mode, `vclear` is never called (see Part 4). This means `inuse=0` in the active array always means "slot not yet touched" — there is no tombstone ambiguity. + +#### vget — Two-Level Lookup + +```c +static int xzarray_get(meta_obj_t* obj, int idx, void** data) { + xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj); + + /* COW mode: check active first, fall through to shadow */ + if (a->shadow != NULL) { + if (idx < a->caps && a->data[idx].inuse) { + *data = a->data[idx].data; + return 0; + } + if (idx < a->shadow_caps && a->shadow[idx].inuse) { + *data = a->shadow[idx].data; + return 0; + } + return -ENOENT; + } + + /* Normal path: unchanged */ + /* ... existing code ... */ +} +``` + +#### vput — Write-Through to Active at Same Index + +```c +/* vput during COW: write to active at the same index. + The kvtab still maps key→same index, so the next vget + finds the updated value in active first. */ +static int xzarray_put_vec(meta_obj_t* obj, int count, int* idxs, void** data) { + xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj); + + /* Ensure active has capacity */ + int maxid = 0; + for (int i = 0; i < count; i++) + if (idxs[i] > maxid) maxid = idxs[i]; + xzarray_ensure_items(a, maxid + 1); + + for (int i = 0; i < count; i++) { + int idx = idxs[i]; + if (a->data[idx].inuse && a->killer) + a->killer(a->data[idx].data); + a->data[idx].data = data[i]; + a->data[idx].inuse = 1; + } + return 0; +} +``` + +Since indices are stable (hash table maps key→same index always), a `vput(42, new_data)` writes to `active[42]`. The hash table still says key→42. Reads check `active[42]` first and find the updated value. The shadow's `shadow[42]` is silently masked. + +#### vstore — Always Goes to Active, Unchanged + +New entries (e.g., new job submit) get new indices that don't exist in shadow. `vstore` logic is unchanged. + +#### Iteration — Merge Walk of Both Arrays + +```c +static int xzarray_iter_next(meta_obj_t* obj, void** data) { + xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj); + + if (a->shadow != NULL) { + /* COW mode: walk both arrays, active takes precedence */ + size_t max_caps = (a->caps > a->shadow_caps) ? a->caps : a->shadow_caps; + for (int i = (int)(long)iter->data; i < max_caps; i++) { + if (i < a->caps && a->data[i].inuse) { + *data = a->data[i].data; /* active wins */ + iter->data = (void*)(long)(i + 1); + return RTTI_ITER_OK; + } + if (i < a->shadow_caps && a->shadow[i].inuse) { + *data = a->shadow[i].data; /* shadow (not in active) */ + iter->data = (void*)(long)(i + 1); + return RTTI_ITER_OK; + } + } + return RTTI_ITER_NO_DATA; + } + + /* Normal path: unchanged */ + /* ... existing code ... */ +} +``` + +### 3.5 Freeze — Start COW Phase + +```c +void xzarray_freeze(xzarray_t* a) { + /* Current data becomes shadow (frozen, immutable) */ + a->shadow = a->data; + a->shadow_caps = a->caps; + + /* Allocate new active with same capacity, all inuse=0 */ + a->data = calloc(a->caps, sizeof(xzarray_item_t)); + a->count = 0; + a->pos = 0; + /* caps stays the same */ +} +``` + +This is a **pointer swap** — O(1), no data copied. The write lock is held only for the duration of the swap (~nanoseconds). + +### 3.6 Thaw — End COW Phase with Merge and GC + +```c +void xzarray_thaw(xzarray_t* a, retain_fn_t should_retain, void* arg) { + for (int i = 0; i < a->shadow_caps; i++) { + if (!a->shadow[i].inuse) continue; /* empty slot, skip */ + + if (i < a->caps && a->data[i].inuse) { + /* Active was written during COW — free shadow's stale object */ + if (a->killer) a->killer(a->shadow[i].data); + } else if (should_retain != NULL + && !should_retain(arg, a->shadow[i].data)) { + /* Entity failed retention check — GC at checkpoint */ + if (a->killer) a->killer(a->shadow[i].data); + /* Caller must also remove kvtab entry for this index */ + } else { + /* Entity still alive — copy pointer to active */ + xzarray_ensure_items(a, i + 1); + a->data[i] = a->shadow[i]; /* struct copy (pointer, inuse) */ + /* Note: a->data[i].inuse is already 1 from shadow copy */ + } + } + + /* Free shadow */ + free(a->shadow); + a->shadow = NULL; + a->shadow_caps = 0; +} +``` + +### 3.7 Checkpoint Flow + +``` +persist_checkpoint() fires on timer thread + │ + ▼ + For each kvmgr (jobs, workers, queues, resources, configs): + rwlock_wlock() + xzarray_freeze() ← O(1) pointer swap + record checkpoint_lsn + rwlock_wunlock() + │ + ▼ + Serialize all shadows → Cap'n Proto → disk + (no locks held, controller serving requests normally, + reads go through COW two-level lookup transparently) + │ + ▼ + After serialization confirmed on disk: + For each kvmgr: + rwlock_wlock() + xzarray_thaw(should_retain): + - Merge shadow → active (pointer copies for untouched entries) + - Apply retention policy → free expired/hidden entities (GC) + - Free shadow + kvtab_gc(): ← remove hash entries for GC'd indices + rwlock_wunlock() + │ + ▼ + Delete WAL segments covered by this checkpoint +``` + +**Lock hold times:** +- Freeze: ~nanoseconds (pointer swap) +- Thaw + GC: O(shadow_caps) pointer copies + retention checks. For 10K entries, ~50-100 microseconds. + +### 3.8 Lock Hierarchy for COW + +XCE uses a two-lock hierarchy in kvmgr: + +``` +kvmgr rwlock (outer) — protects kvtab + serializes freeze/thaw + └─ xzarray mutex (inner) — protects individual vstore/vget/vput/vclear operations +``` + +- **kvmgr `my_rwlock_t`**: The outer lock. Taken as read-lock for normal lookups (`xkvmgr_find`). Taken as write-lock for additions (`xkvmgr_add`) and for COW freeze/thaw coordination. +- **xzarray `mylock_t`**: The inner lock. Automatically taken inside `xzarray_store`, `xzarray_get`, `xzarray_put_vec`, `xzarray_clear`. Not taken explicitly by COW code. + +**COW lock protocol:** + +``` +freeze: + rwlock_wlock(kvmgr) ← serialize: no concurrent lookups during swap + xzarray_freeze(data) ← O(1) pointer swap, no xzarray mutex needed + rwlock_wunlock(kvmgr) + +serialize shadows: + (no locks held — shadows are immutable) + +thaw + GC: + rwlock_wlock(kvmgr) ← serialize: block lookups during merge + xzarray_thaw(data, retain) ← merges shadow→active, frees GC'd entities + xzarray mutex taken internally per vput + kvtab_gc_keys(kvmgr) ← prune stale hash entries + rwlock_wunlock(kvmgr) +``` + +**Per-object locks are NOT nested inside kvmgr lock.** The kvmgr lock serializes container-level operations (key→index lookup, COW coordination). Per-object locks (`job_t.lock`, `worker_t.lock`) are taken after releasing the kvmgr lock — the lookup returns a referenced pointer, and the caller locks the object separately. + +This matches the existing pattern in `handle_jobsubmit()` (`job.c:455-525`): `lock(c->jobmgr, RTTI_LOCK_READ)` for lookup, release, then per-object lock for modification. + +### 3.9 Checkpoint Timer + +The checkpoint thread should use the existing `delaymgr` timer system (already used for heartbeat checks in `cluster.c`) rather than a raw `pthread_create`. This ensures the checkpoint timer integrates with the controller's timer infrastructure: + +```c +/* In persist_start() — register periodic checkpoint timer */ +delay_id_t ckpt_timer = delay_add(c->delaymgr, + persist_checkpoint_timer_fn, + (void*)c, + config->snap_interval_s * 1000, /* ms */ + true /* recurring */); +``` + +Benefits: +- No raw `pthread_create`/`pthread_join` management for the timer thread +- `delaymgr` already handles timer cancellation in `cluster_fini()` +- Consistent with how heartbeat timers are managed + +### 3.10 Snapshot File Format + +Serialized from shadow arrays while no locks are held: + +``` +┌─────────────────────────┐ +│ SnapshotHeader (Capnp) │ +│ - lsn: last WAL LSN │ +│ - timestamp │ +│ - term │ +│ - nJobs, nQueues, nWorkers │ +├─────────────────────────┤ +│ Job array (Capnp) │ +├─────────────────────────┤ +│ Queue array (Capnp) │ +├─────────────────────────┤ +│ Worker array (Capnp) │ +├─────────────────────────┤ +│ Resource array (Capnp) │ +├─────────────────────────┤ +│ Config array (Capnp) │ +└─────────────────────────┘ +``` + +### 3.11 Checkpoint Triggers + +- Time-based: every `snap_interval_s` (default 60s) +- Size-based: when WAL exceeds `snap_wal_size_mb` (default 64 MB) +- Shutdown: final checkpoint in `cluster_run()` before exiting the main loop + +--- + +## Part 4: Deferred Entity Cleanup at Checkpoint + +### 4.1 Design Philosophy + +**No entity is ever deleted during normal operation.** All entities follow the same lifecycle: + +``` +alive state ──▶ terminal state ──▶ [TTL/retention expires] ──▶ cleaned at next checkpoint +``` + +This eliminates `vclear` and `del_key` from the hot path, which: +1. Simplifies COW (no tombstones, no three-state flags — see Part 3) +2. Simplifies WAL (no delete entry types — replay is purely additive) +3. Makes cleanup predictable and batched — no random delete spikes + +### 4.2 Terminal States by Entity Type + +| Entity | Terminal States | Trigger | Cleanup Criteria | +|--------|----------------|---------|-------------------| +| Job | `JOB_DONE`, `JOB_EXITED`, `JOB_FAILED` | Job finishes naturally | No dependents AND `finish_time + finished_retain_s < now` | +| Worker | `WORKER_UNAVAILABLE` | Worker daemon dies / disconnects | `disconnect_time + worker_retain_s < now` | +| Worker | `WORKER_HIDDEN` | Admin explicit remove | Immediate cleanup at next checkpoint (no TTL) | +| Queue | `QUEUE_HIDDEN` | Admin deletes queue | Immediate cleanup at next checkpoint | +| Config | `CONFIG_HIDDEN` | Admin deletes config key | Immediate cleanup at next checkpoint | +| Resource | _(never deleted)_ | N/A | Resources persist permanently | + +### 4.3 How Terminal States Work + +Entities in terminal states remain `inuse=1` in the zarray and still have their kvtab entries. Business logic ignores them through **status-based filtering** that already exists: + +- Scheduler skips `JOB_DONE`/`JOB_EXITED`/`JOB_FAILED` jobs (already happens) +- Scheduler skips `WORKER_UNAVAILABLE` workers (already happens) +- API queries filter by status (already happens) + +The new states `WORKER_HIDDEN`, `QUEUE_HIDDEN`, `CONFIG_HIDDEN` follow the same pattern — business logic and queries skip them, but the data structure entries persist until the next checkpoint. + +### 4.4 New HostState / EntityState Values + +Add to the Cap'n Proto schema: + +```capnp +# In HostState enum — new terminal states for deferred cleanup +enum HostState { + # ... existing states ... + unavail @X; # Worker daemon died / disconnected (replaces immediate removal) + hidden @Y; # Admin explicitly removed — invisible but not yet cleaned up +} + +# QueueStatus — add hidden state +enum QueueStatus { + # ... existing states ... + hidden @X; # Admin deleted — invisible but not yet cleaned up +} +``` + +### 4.5 Retention Policy at Checkpoint + +The `should_retain` callback passed to `xzarray_thaw()` implements entity-specific retention: + +```c +/* Job retention: keep if it has dependents OR TTL not expired */ +static bool job_should_retain(void* arg, void* item) { + cluster_t* c = (cluster_t*)arg; + job_t* j = (job_t*)item; + + if (!job_is_terminal(j)) + return true; /* still active, keep */ + + if (job_has_dependents(c, j->jobid)) + return true; /* other jobs depend on it */ + + time_t now = time(NULL); + if (j->finish_time + c->persist.config.finished_retain_s > now) + return true; /* TTL not expired */ + + return false; /* safe to clean up */ +} + +/* Worker retention: keep if TTL not expired (or hidden = immediate) */ +static bool worker_should_retain(void* arg, void* item) { + cluster_t* c = (cluster_t*)arg; + worker_t* w = (worker_t*)item; + + if (w->state == WORKER_HIDDEN) + return false; /* admin removed, cleanup immediately */ + + if (w->state != WORKER_UNAVAILABLE) + return true; /* still connected, keep */ + + time_t now = time(NULL); + if (w->disconnect_time + c->persist.config.worker_retain_s > now) + return true; /* TTL not expired */ + + return false; /* safe to clean up */ +} +``` + +### 4.6 kvtab Cleanup at Checkpoint + +When `xzarray_thaw` frees an entity (returns it from `should_retain` as `false`), the kvtab still has a key→index mapping for it. After thaw completes, walk the kvtab and prune stale entries: + +```c +void kvmgr_gc_keys(meta_obj_t* obj) { + xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj); + + /* Walk all hash chains. For each entry, check if the + zarray slot at the mapped index is still inuse. + If inuse=0, the entity was GC'd — remove hash entry. */ + kvtab_gc_stale_indices(kvm->kvtab, kvm->data); +} +``` + +This runs while the write lock is still held (during thaw), so it's safe and bounded. For 10K entries, the hash chain scan is ~tens of microseconds. + +### 4.7 Index Reuse After Cleanup + +After checkpoint cleanup, freed zarray slots become `inuse=0` with their kvtab entries removed. The zarray's circular scan (`pos`) finds these freed slots, so new `vstore` calls reuse them. A new job can be assigned index 42 after the old job 42 was cleaned up — the kvtab gets a fresh key→42 entry for the new job. No confusion because the old key→42 entry was removed. + +### 4.8 Benefits of Deferred Cleanup + +| Aspect | Before (eager delete) | After (deferred to checkpoint) | +|--------|----------------------|-------------------------------| +| COW complexity | Requires tombstones or three-state flags | Simple — `inuse=0` always means "not touched" | +| WAL entry types | Need `WAL_*_DELETE` types | No delete types — WAL is purely additive | +| WAL replay | Must handle insert-delete-reinsert correctly | Insert or update — always correct | +| Delete spikes | Random (worker disconnect, admin action) | Bounded — only at checkpoint intervals | +| Memory overhead | Freed immediately | Entity retained for up to `snap_interval_s` (default 60s) | +| zarray fragmentation | Holes appear unpredictably | Compacted at checkpoint during merge | +| kvtab fragmentation | Hash chains shrink unpredictably | Pruned at checkpoint — predictable | + +### 4.9 Endpoint Deferred Cleanup + +Endpoints (the `endpoint_t` objects in the controller's `c->endpoints` zarray) also follow the deferred cleanup pattern, but with an important difference: the network resource (fd, SSL object) must be released immediately, while the slot cleanup is deferred. + +When a worker or scheduler disconnects: + +1. **Immediate**: Close `fd` and free `SSL*` — these are OS resources that cannot be retained +2. **Immediate**: Set `ep->busy = 0` (unblock any pending vtask queue entries) +3. **Deferred**: Set `ep->state = ENDPOINT_DISCONNECTED` but do **NOT** call `zarray_del(&(c->endpoints), ep->idx)` — this was the old eager cleanup in `handle_endpoint_error()` (`cluster.c`) + +The endpoint slot is cleaned up at checkpoint via the same `should_retain` mechanism: + +```c +static bool endpoint_should_retain(void* arg, void* item) { + endpoint_t* ep = (endpoint_t*)item; + /* Disconnected endpoints have fd=-1 and SSL=NULL (already freed). + The slot just needs to be reclaimed by the zarray. */ + return (ep->state != ENDPOINT_DISCONNECTED); +} +``` + +**Why not delete immediately?** The old code in `cluster.c` called `zarray_del()` in `handle_endpoint_error()`. This is the `xzarray_clear()` path — setting `inuse=0` and calling the killer. During COW mode, calling `vclear` would corrupt the active array (it would create a "hole" that doesn't exist in the shadow, causing `vget` two-level lookup to incorrectly fall through to stale shadow data). Deferring to checkpoint avoids this entirely. + +**Connection to vtaskmgr**: When `endpoint_block_tasks(ep)` sets `ep->busy = 1`, and later `endpoint_unblock_tasks(ep)` sets `ep->busy = 0`, this per-endpoint serialization is maintained regardless of whether the endpoint is in COW mode. The `busy` flag is on the endpoint object itself, not on the zarray slot. + +### 4.10 Existing Code Alignment: Worker Unavailable State + +The `mark_worker_unavail()` function already exists in `worker.c:516-577`. It currently: + +1. Sets the `XCE_WORKER_UNAVAIL` flag on the worker +2. Runs `AFTER_EVENT` hooks (which will trigger the WAL write for `WAL_WORKER_CTRL`) +3. Does **not** remove the worker from the workermgr — the worker remains in the zarray with `inuse=1` + +This matches the deferred cleanup design: after `mark_worker_unavail()`, the worker is in `WORKER_UNAVAILABLE` state and will be garbage-collected at the next checkpoint when `worker_retain_s` expires. + +The new additions needed: +1. Set `w->disconnect_time = time(NULL)` in `mark_worker_unavail()` so the TTL check works +2. Add `WORKER_HIDDEN` state for admin-initiated removal (sets state, never deletes from zarray) +3. The `handle_worker_leave()` path currently calls `mark_worker_unavail(c, w)` and sets `w->ep = NULL` — this is correct and needs no change + +--- + +## Part 5: HA Failover — LSF-Style Priority Election + +### 5.1 Design Philosophy + +Instead of Raft consensus (~5000 LOC), use LSF's simple priority-based model: + +- A fixed list of controller candidates is defined in cluster config, ordered by priority +- The first (highest-priority) candidate that is alive becomes primary +- No voting, no term negotiation, no split-brain risk +- "Am I the highest-priority alive candidate?" is the entire election algorithm + +**Trade-off**: Non-preemptive — if a higher-priority candidate recovers, it does NOT preempt the current primary. This avoids disruptive failback. Admin can trigger manual failover. + +### 5.2 Candidate Configuration + +The `MasterCandidate` struct already exists in `controller.capnp` with `name`, `host`, `port`. Priority is **list position** (index 0 = highest priority). + +```yaml +# cluster.yaml +controller: + candidates: + - name: ctrl-1 + host: 10.0.1.1 + port: 7000 + - name: ctrl-2 + host: 10.0.1.2 + port: 7000 + - name: ctrl-3 + host: 10.0.1.3 + port: 7000 +``` + +### 5.3 Role Management + +```c +typedef enum { CTRL_PRIMARY = 0, CTRL_STANDBY = 1 } ctrl_role_t; + +typedef struct { + ctrl_role_t role; + int my_priority; /* My index in candidates[] */ + replicator_t repl; /* Replicate to standbys (primary only) */ + uint64_t applied_lsn; /* Last applied WAL entry (standby only) */ + int primary_fd; /* Connection to current primary (standby only) */ + char primary_host[XCE_MAXNAME_LEN]; + uint32_t primary_port; + pthread_t monitor_tid; /* Monitors higher-priority candidates */ +} ha_state_t; +``` + +Added to `cluster_t` as `ha_state_t ha;`. + +### 5.4 Startup Sequence + +``` +Controller starts + │ + ▼ + Read candidates[] from config + Find my_priority (my position in list) + │ + ▼ + ┌─────────────────────────────┐ + │ am_i_primary()? │ + │ │ + │ For each candidate with │ + │ priority < my_priority: │ + │ Try TCP connect │ + │ If any connects: NO │ + │ │ + │ If none connect: YES │ + └──────┬──────────┬───────────┘ + │ │ + PRIMARY STANDBY + │ │ + ▼ ▼ + persist_init() connect to primary + persist_recover() receive WAL stream + cluster_init() replay WAL entries + persist_start() ha_monitor_thread() + cluster_start() standby_loop() +``` + +### 5.5 Monitor Thread (Standby) + +The standby's monitor thread watches all candidates with higher priority: + +```c +static void* ha_monitor_thread(void* arg) +{ + cluster_t* c = (cluster_t*)arg; + ha_state_t* ha = &c->ha; + + for (;;) { + sleep(HA_CHECK_INTERVAL); /* default 2-5 seconds */ + + /* Check if any higher-priority candidate is alive */ + for (int i = 0; i < ha->my_priority; i++) { + if (tcp_probe(candidates[i].host, candidates[i].port) == 0) { + /* Higher-priority candidate is alive — stay standby */ + goto next_check; + } + } + + /* No higher-priority candidate is alive — promote self */ + ha_promote_to_primary(c); + return NULL; + + next_check: + continue; + } +} +``` + +### 5.6 Promotion Sequence + +When a standby promotes to primary: + +1. **Stop WAL replay** from old primary +2. **Increment term** (`ha.term++`) — fencing token for stale standbys +3. **Switch role** to `CTRL_PRIMARY` +4. **Run `persist_recover()`** on locally accumulated WAL (replay any unapplied entries) +5. **Initialize replicator** — establish connections to remaining standbys +6. **Start normal primary operation** — `cluster_start()`, accept connections + +### 5.7 Term (Fencing Token) + +The `term` field prevents split-brain: + +- Incremented on every promotion +- Included in every WAL entry and replication message +- A standby receiving a message with a lower term than its own ignores it +- Workers/scheduler reconnect and verify term on join + +### 5.8 Non-Preemptive Failover + +``` +Time ─────────────────────────────────────────────▶ + +ctrl-1 (P=0): ████████░░░░░░████████████████████████ + PRIMARY DEAD PRIMARY (recovered, + │ does NOT reclaim) + │ +ctrl-2 (P=1): ░░░░░░░░██████████████████████████████ + STANDBY PRIMARY (promoted at T1) +``` + +When ctrl-1 recovers: +1. It probes for candidates with priority < 0 → none exist +2. It becomes primary? **No** — it probes ctrl-2 and finds it alive +3. ctrl-1 stays standby and connects to ctrl-2 as primary +4. Admin can issue `xce hostcontrol --failover` to manually move primary back to ctrl-1 + +### 5.9 New Opcodes and Cap'n Proto Additions + +```capnp +/* In HostRole enum: */ +standby @5; /* NEW — standby controller role */ + +/* In OPCode enum: */ +replEntry @23; /* Replicated WAL entry (primary → standby) */ +replAck @24; /* ACK for sync-replicated entry (standby → primary) */ +redirect @25; /* Redirect to current primary (standby → client) */ + +/* In Request struct: */ +term @22 :UInt64 $C.mapname("term"); /* NEW — fencing token */ +``` + +--- + +## Part 6: Fan-Out TCP Replication + +### 6.1 Topology Choice: Fan-Out vs Chain + +| Factor | Fan-Out | Chain | +|--------|---------|-------| +| Latency to all standbys | `max(RTT_1..N)` — single hop | `RTT_1 + RTT_2 + ... RTT_N` — cumulative | +| Primary load | N non-blocking sends | 1 send (lowest load) | +| Failure isolation | Standby failure only affects that standby | One slow standby blocks the chain | +| Implementation | Simple | Must handle chain reordering | +| At XCE scale (~3 standbys) | Primary sends 3 copies — trivial | Marginal load savings | + +**Decision**: Fan-out. At 3 standbys, the primary does 3 non-blocking `send()` calls per WAL entry — negligible overhead. The latency advantage (single hop) and failure isolation are decisive. + +### 6.2 Replicator Data Structure + +```c +#define MAX_STANDBYS (N_CONTROLLER_MAX - 1) /* 15 */ + +typedef struct { + int fd; /* TCP connection to this standby */ + uint64_t acked_lsn; /* Last LSN this standby has ACKed */ + int lagging; /* Flag: standby is behind threshold */ + char host[XCE_MAXNAME_LEN]; + uint32_t port; +} standby_conn_t; + +typedef struct { + standby_conn_t standbys[MAX_STANDBYS]; + int nstandbys; + mylock_t lock; /* Protects standbys array */ + uint64_t pending_lsn; /* LSN we're waiting ACK for (sync mode) */ + pthread_cond_t ack_cond; /* Signaled when ACK arrives */ + mylock_t ack_lock; /* Protects ack_cond + pending_lsn */ +} replicator_t; +``` + +### 6.3 Replication Flow + +``` +Primary Standby-1 Standby-2 + │ │ │ + │ WAL write + fsync │ │ + │──────────────────────────────────────│ │ + │ │ │ + │ repl_broadcast(entry) │ │ + │─── send(WAL_ENTRY) ─────────────────▶│ │ + │─── send(WAL_ENTRY) ─────────────────────────────────▶│ + │ │ │ + │ [for SYNC mode only] │ │ + │ repl_wait_ack(target_lsn, timeout) │ │ + │ │ │ + │◀── ACK(acked_lsn) ──────────────────│ │ + │◀── ACK(acked_lsn) ─────────────────────────────────│ + │ │ │ + │ Condition signal → resume │ │ + │ │ │ +``` + +### 6.4 Standby WAL Receiver + +The standby runs a simple receive loop: + +```c +/* On standby: receive WAL entries from primary */ +static void* standby_recv_thread(void* arg) +{ + cluster_t* c = (cluster_t*)arg; + ha_state_t* ha = &c->ha; + + for (;;) { + wal_entry_hdr_t hdr; + void* payload; + + /* Read WAL entry from primary */ + if (read(ha->primary_fd, &hdr, sizeof(hdr)) != sizeof(hdr)) + break; /* Primary connection lost */ + + payload = malloc(hdr.payload_len); + if (read(ha->primary_fd, payload, hdr.payload_len) != hdr.payload_len) + break; + + /* Validate term — reject stale entries */ + if (hdr.term < ha->term) { + free(payload); + continue; + } + + /* Write to local WAL */ + persist_wal_async(&c->persist, hdr.type, payload, hdr.payload_len); + + /* Apply to in-memory state */ + apply_wal_entry(c, &hdr, payload); + + ha->applied_lsn = hdr.lsn; + + /* Send ACK for sync-replicated entries */ + if (hdr.type == WAL_JOB_SUBMIT) { + send_repl_ack(ha->primary_fd, hdr.lsn); + } + + free(payload); + } + + /* Primary connection lost — start monitor to check for promotion */ + ha_monitor_thread(c); + return NULL; +} +``` + +### 6.5 Replication Uses the Endpoint/Iomgr/Vtask SSL Pipeline + +**All controller connections use SSL** — the iomgr (`netio.c`) enforces this. Replication between controllers is no exception. The raw `read/write` calls in the standby receiver pseudocode above are simplified; the actual implementation must use the full endpoint/iomgr/vtask pipeline: + +``` +Standby Primary + │ │ + │ iomgr_monitor(ep) │ + │ ─── endpoint established via SSL ──│ + │ │ + │ io_worker (epoll) │ + │ └─ read_fsm (SSL_read) │ + │ └─ handle_msg() │ + │ └─ vtask_handle_msg() │ + │ └─ OP_replEntry │ + │ └─ apply_wal_entry() │ + │ └─ persist_wal_async()│ + │ │ + │ For SYNC: vtask sends OP_replAck │ + │ └─ outq (per-endpoint serialized) │ + │ └─ write_fsm (SSL_write) │ + │ ──────────────────────────────▶│ +``` + +**Why the full pipeline?** +1. **SSL enforcement**: All connections in `netio.c` go through `SSL_new()`/`SSL_set_fd()`. Creating a raw TCP socket for replication would bypass this requirement. +2. **Per-endpoint serialization**: The vtaskmgr `endpoint_block_tasks()`/`endpoint_unblock_tasks()` mechanism ensures that messages on a given endpoint are processed in order. This guarantees that WAL entries from the primary are applied in order on the standby. +3. **Backpressure**: If the standby is slow (e.g., during checkpoint serialization), the endpoint's outq fills up, and `io_worker` naturally applies backpressure via the write FSM. Raw TCP would require a separate flow-control mechanism. +4. **Connection management**: Reconnection, heartbeat, and error handling all reuse the existing `handle_connection()`/`handle_endpoint_error()` code path. + +**`standby_conn_t` update**: Replace `int fd` with `endpoint_t* ep` — the standby connection is a full endpoint managed by iomgr, not a raw file descriptor. + +```c +typedef struct { + endpoint_t* ep; /* Full endpoint (SSL, FSM, outq) — not raw fd */ + uint64_t acked_lsn; /* Last LSN this standby has ACKed */ + int lagging; /* Flag: standby is behind threshold */ + char host[XCE_MAXNAME_LEN]; + uint32_t port; +} standby_conn_t; +``` + +The primary initiates the replication connection as a client (using iomgr's outgoing connection support), and the standby accepts it via its own iomgr. The `OP_replEntry` and `OP_replAck` opcodes flow through the same SSL/endpoint/vtask pipeline as all other controller messages. + +The standby_recv_thread pseudocode above would instead be handled by the iomgr event loop: the standby's `handle_msg()` dispatches `OP_replEntry` to `vtask_handle_msg()`, which applies the WAL entry and sends `OP_replAck` back via the endpoint's outq. + +--- + +## Part 7: Mixed Sync/Async Replication Protocol + +### 7.1 Design Rationale + +The user's mental model is: **"if the API server ACKed my job submit, the job survives a crash."** This requires sync replication for job submits only. + +All other events (job start, finish, fail, rusage, queue/worker control) are self-healing: +- If a job start is lost, the worker will re-notify on reconnect +- If a finish is lost, the job appears running until the worker re-reports +- Queue/host control is infrequent and can be re-sent by the admin + +**Sync replication** = WAL write + wait for at least one standby ACK before ACKing the client. +**Async replication** = WAL write + fire-and-forget send to standbys. + +### 7.2 Sync Mode: Job Submit Flow + +``` +API Server Controller (Primary) Standby + │ │ │ + │ job submit request │ │ + │──────────────────────────▶│ │ + │ │ BEFORE_EVENT hook │ + │ │ job_submit() │ + │ │ AFTER_EVENT hook ──┐ │ + │ │ │ │ + │ │ persist_wal_sync() │ │ + │ │ WAL write + fsync│ │ + │ │ send(entry) ─────────▶│ + │ │ wait_ack(timeout)│ │ + │ │◀─── ACK(ack_lsn) ───────│ + │ │ │ │ + │ │ ← hook returns │ │ + │ │ │ + │ reply(jobid) │ │ + │◀──────────────────────────│ │ + │ │ │ +``` + +**Key change**: `xce_reply()` is moved **after** `AFTER_EVENT` hooks in `handle_jobsubmit()`. This allows the persistence hook to complete sync replication before the client is ACKed. + +### 7.3 Async Mode: All Other Events + +``` +Worker Controller (Primary) Standby + │ │ │ + │ job start notify │ │ + │──────────────────────────▶│ │ + │ │ mark_job_start() │ + │ │ AFTER_EVENT hook ──┐ │ + │ │ │ │ + │ │ persist_wal_async()│ │ + │ │ WAL write │ │ + │ │ send(entry) ─────────▶│ + │ │ return immediately │ + │ │ │ │ + │ │ ← hook returns │ │ + │ │ │ + │ reply(OK) │ │ + │◀──────────────────────────│ │ +``` + +### 7.4 Degraded Mode + +When no standby ACKs within the timeout (`sync_timeout_ms`, default 50-100ms): + +1. **Do NOT block the client indefinitely** — timeout is hard +2. **ACK the client based on local WAL durability alone** — the primary's local WAL is already fsynced +3. **Log a warning** — admin should investigate standby health +4. **Self-heal later** — when the standby reconnects, it will catch up via WAL replay + +This trades strict consistency for availability. In practice, the primary's local WAL provides durability as long as the primary doesn't lose its disk. + +### 7.5 Sync Timeout Tuning + +| Setting | Latency Impact | Durability Guarantee | +|---------|----------------|---------------------| +| 10 ms | Low — client sees ~10ms extra | Weak — only one standby, might not have fsynced | +| 50 ms | Moderate | Good — likely one standby has persisted | +| 100 ms | Higher | Strong — high probability all standbys persisted | +| No timeout | Unbounded | Strongest — but blocks on standby failure | + +**Recommended default**: 50ms. At ~1000 submits/sec, average inter-submit gap is 1ms. A 50ms timeout adds 5% to average submit latency while providing strong durability. If all standbys are healthy (RTT < 5ms), ACKs arrive well within the timeout and there's no waiting. + +--- + +## Part 8: Hook-Driven Architecture + +### 8.1 Existing Hook Pattern + +XCE's hook system already integrates with the controller. Examples from `apiserver.c` and `sched.c`: + +```c +/* apiserver.c — existing pattern */ +static int apiserver_job_hook_fun(int type, int situation, int detail, + void* data, va_list ap) +{ + cluster_t* c = va_arg(ap, cluster_t*); + /* ... handle job events ... */ +} + +/* Registration in apiserver_init() */ +xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, + (hook_fn_t)apiserver_job_hook_fun, NULL); +``` + +### 8.2 Persistence Hook Functions + +Persistence follows the exact same pattern: + +```c +/* persist.c — new hook functions */ +int persist_job_hook_fun(int type, int situation, int detail, + void* data, va_list ap) +{ + cluster_t* c; + persist_t* p; + + /* Only act on AFTER_EVENT — state mutation is complete */ + if (situation != HOOK_AFTER_EVENT) + return 0; + + c = va_arg(ap, cluster_t*); + p = &c->persist; + + switch (detail) { + case HOOK_DETAIL_JOB_NEW: + /* SYNC: WAL write + replicate + wait ACK */ + return persist_wal_sync_append(p, WAL_JOB_SUBMIT, + data, compute_payload_len(data)); + case HOOK_DETAIL_JOB_START: + return persist_wal_async(p, WAL_JOB_START, data, ...); + case HOOK_DETAIL_JOB_EXIT: + return persist_wal_async(p, WAL_JOB_FINISH, data, ...); + case HOOK_DETAIL_JOB_FAIL: + return persist_wal_async(p, WAL_JOB_FAIL, data, ...); + /* ... etc ... */ + default: + return 0; + } +} + +int persist_queue_hook_fun(int type, int situation, int detail, + void* data, va_list ap) +{ + /* ... same pattern for queue events ... */ +} + +int persist_worker_hook_fun(int type, int situation, int detail, + void* data, va_list ap) +{ + /* ... same pattern for worker events ... */ +} +``` + +### 8.3 Critical Change: Reply After Hooks + +**Current code** in `handle_jobsubmit()` (`job.c:455-525`): + +```c +/* CURRENT — reply BEFORE hooks */ +xce_run_hooks(BEFORE_EVENT, HOOK_DETAIL_JOB_NEW, c, submit); +jobid = job_submit(c, submit); + +reply.status = 0; +reply.data_which = RequestReply_data_jobid; +reply.data.jobid = jobid; +xce_reply(ep, c->myhost, &reply); /* ← ACK before hooks */ + +/* Collect jobs into list... */ +xce_run_hooks(AFTER_EVENT, HOOK_DETAIL_JOB_NEW, c, &jlist); /* ← persistence happens here */ +``` + +**Modified code** — move `xce_reply()` after AFTER_EVENT: + +```c +/* MODIFIED — reply AFTER hooks */ +xce_run_hooks(BEFORE_EVENT, HOOK_DETAIL_JOB_NEW, c, submit); +jobid = job_submit(c, submit); + +if (jobid == 0) { + reply.status = -errno; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); /* Error reply is immediate */ + return -errno; +} + +/* Collect jobs into list... */ +xce_run_hooks(AFTER_EVENT, HOOK_DETAIL_JOB_NEW, c, &jlist); +/* ↑ SYNC replication happens inside this hook call */ + +/* Reply AFTER hooks — client ACK implies durability */ +reply.status = 0; +reply.data_which = RequestReply_data_jobid; +reply.data.jobid = jobid; +xce_reply(ep, c->myhost, &reply); +``` + +**Why this works**: The persistence hook runs inside `AFTER_EVENT`. For job submits, it does sync WAL + replication. Only after the hook returns (meaning the entry is durable) does the controller send the client ACK. This keeps all persistence logic inside hooks — no inline persistence code in `handle_jobsubmit()`. + +**Lock safety analysis**: The current `handle_jobsubmit()` acquires `lock(c->jobmgr, RTTI_LOCK_READ)` for job lookup, then operates on the per-job object. No kvmgr write lock is held during the hook call or `xce_reply()`. The change to move `xce_reply()` after hooks is therefore safe: + +1. `job_submit()` takes the jobmgr READ lock for lookup, then per-object lock for modification — no change here +2. `AFTER_EVENT` hooks (including `persist_wal_sync_append()`) acquire their own locks internally (`persist.lock` for WAL, `repl.lock` for replication) — no nesting with jobmgr +3. `xce_reply()` writes to the endpoint's outq — no kvmgr lock interaction +4. The only path that takes jobmgr WRITE lock during hooks is `handle_worker_join()` in `join.c:58-69` (via `add_worker()` + hook execution). This path runs on a separate vcpu from job submits and doesn't conflict with the reply ordering change + +**Replication scope**: Sync replication (`repl_wait_ack`) only triggers for `WAL_JOB_SUBMIT` entries, which originate from API server connections. The controller only handles connections from workers, scheduler, and API servers — never end users. At 1–3 API server connections, the vcpu pool is not at risk of exhaustion from blocking `repl_wait_ack()` calls. + +### 8.4 Registration in `cluster_init()` + +```c +/* In cluster_init(), after xce_install_hooks(): */ + +if (persist_config_enabled(&c->myconf->cluster_config->persist)) { + rc = persist_init(&c->persist, &c->myconf->cluster_config->persist, c); + if (rc < 0) return rc; + + rc = persist_recover(&c->persist, c); + if (rc < 0) return rc; + + /* Register persistence hooks */ + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, + (hook_fn_t)persist_job_hook_fun, NULL); + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_QUEUE, + (hook_fn_t)persist_queue_hook_fun, NULL); + xce_add_hook(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, + (hook_fn_t)persist_worker_hook_fun, NULL); +} +``` + +--- + +## Part 9: Worker Connection Strategy + +### 9.1 Current Design: Worker-to-Controller Long TCP + +Workers maintain a persistent TCP connection to the controller. This provides: + +1. **Fast failure detection**: If a worker dies, the controller gets an immediate TCP RST — no polling needed +2. **Easy push from controller**: The controller can push job dispatches, config updates, and heartbeats over the existing connection without initiating a new one +3. **Bidirectional heartbeat**: Both sides can detect liveness via the same connection + +The main concern is the **OS connection limit**: each TCP connection consumes a file descriptor and kernel socket buffer. However, for typical HPC cluster scales (100s–low 1000s of workers), this is a configuration issue, not an architectural one: + +```bash +# /etc/sysctl.conf — sufficient for a 2000-node cluster +net.core.somaxconn = 4096 +net.ipv4.tcp_max_syn_backlog = 8192 +fs.file-max = 200000 + +# In the controller process startup: +# ulimit -n 65535 +``` + +Linux handles tens of thousands of idle TCP connections with minimal overhead. The practical limit is ~65K connections per IP/port tuple, which is far beyond any single XCE cluster. + +### 9.2 Decision: Workers Connect Directly to Primary + +Three options were evaluated: + +**Option A — Direct-only**: Workers connect to primary only. + +**Option B — Standby as full proxy**: Workers connect to any candidate; standbys forward all traffic (both directions) to primary. + +**Option C — Standby as discovery + redirect**: Workers can connect to any candidate at startup; standbys send a redirect response pointing to the primary; worker reconnects to primary. + +| Aspect | A (Direct-only) | B (Full proxy) | C (Redirect) | +|--------|-----------------|----------------|--------------| +| Normal latency | 1 hop | 2 hops (doubles) | 1 hop | +| Complexity | Minimal | High — bidirectional relay, reply correlation, read-after-write consistency | Low — just a redirect opcode | +| Failover reconnection | Workers must discover new primary | Workers on promoting standby: zero cost; others: must reconnect | Workers reconnect with libxce auto-retry | +| Worker startup | Must know primary | Pick any candidate | Pick any candidate, get redirected | +| Read-after-write | No issue | **Critical issue** — standby forwards write to primary, but its own state is stale until replicated back | No issue — all writes go to primary directly | +| Connection load | Primary only | Distributed | Primary only (discovery is transient) | + +**Decision: Option C (Redirect).** It gives the "worker doesn't need to know the primary" benefit at startup, without paying the 2-hop latency and read-after-write consistency cost on every message. The redirect is a one-time cost during connection setup — amortized to zero over a long-lived connection. + +This pattern is used by etcd (proxy mode), Redis Sentinel, and MongoDB replica sets. + +### 9.3 Full Communication Topology + +``` + ┌─────────────────┐ + │ Controller │ + │ Candidates │ + │ (cluster.yaml) │ + └────────┬────────┘ + │ + ┌───────────────┼───────────────┐ + ▼ ▼ ▼ + ┌──────────┐ ┌──────────┐ ┌──────────┐ + │ ctrl-1 │ │ ctrl-2 │ │ ctrl-3 │ + │ PRIMARY │ │ STANDBY │ │ STANDBY │ + │ P=0 │ │ P=1 │ │ P=2 │ + └────┬─────┘ └────┬─────┘ └────┬─────┘ + │ │ │ + ┌────────┼──────┐ │ │ + │ │ │ │ │ + ▼ ▼ ▼ ▼ ▼ + Worker Worker Worker (accepts discovery + W1 W2 W3 connections only — + sends REDIRECT) + │ + New Worker W4 + startup flow: + 1. Try ctrl-2 (random) + 2. Get REDIRECT → ctrl-1 + 3. Connect to ctrl-1 +``` + +### 9.4 Standby Redirect Handling + +When a standby receives a join request from a non-controller peer (worker, scheduler, API server), it sends a redirect response instead of accepting the connection: + +```c +/* In handle_join() — modified for HA redirect */ +int handle_join(cluster_t* c, endpoint_t* ep, xce_request_t* request) +{ + xce_peerinfo_t* peer = request->body.joinmsg->peer; + int role = peer->role; + + /* If I'm a standby and the peer is a worker/scheduler/apiserver, + redirect them to the primary */ + if (c->ha.role == CTRL_STANDBY) { + switch (role) { + case HostRole_worker: + case HostRole_scheduler: + case HostRole_apiserver: { + xce_reply_t reply; + init_reply_by_req(&reply, request); + reply.status = E_XCE_REDIRECT; + reply.data_which = RequestReply_data_redirect; + reply.data.redirect.host = c->ha.primary_host; + reply.data.redirect.port = c->ha.primary_port; + xce_reply(ep, c->myhost, &reply); + return 0; + } + default: + break; + } + } + + /* Normal join processing (I am primary) */ + /* ... existing code ... */ +} +``` + +### 9.5 Thundering Herd Mitigation on Failover + +When the primary dies, all workers connected to it get TCP RST simultaneously. If they all reconnect to the new primary at once, the listen backlog and accept loop could be overwhelmed. + +**Mitigation**: Exponential backoff with random jitter in libxce (detailed in Part 10). This spreads worker reconnections over a time window instead of a single instant. + +For 2000 workers with 100ms base backoff + 50% jitter: +- First wave: ~0-150ms — ~500 workers reconnect +- Second wave: ~150-300ms — ~300 workers +- Third wave: ~300-600ms — remaining workers +- Total reconnection window: ~600ms — well within the new primary's capacity + +### 9.6 Worker Disconnect — Deferring to Terminal State + +When a worker disconnects, instead of immediately removing it from data structures, the controller transitions it to `WORKER_UNAVAILABLE`: + +```c +/* In handle_worker_disconnect() — modified for deferred cleanup */ +void handle_worker_disconnect(cluster_t* c, worker_t* w) { + /* Instead of: destroy(w); kvmgr_del(...); */ + w->state = WORKER_UNAVAILABLE; + w->disconnect_time = time(NULL); + + /* Worker remains in zarray (inuse=1) and kvtab. + Scheduler already ignores UNAVAILABLE workers. + Actual cleanup happens at next checkpoint if + worker_retain_s has expired. */ +} +``` + +### 9.7 Redirect Cap'n Proto Schema + +```capnp +/* New struct for redirect response */ +struct RedirectInfo $C.mapname("xce_redirect_t") { + host @0 :Text $C.mapname("host"); + port @1 :UInt32 $C.mapname("port"); +} + +/* In RequestReply union: */ +redirect @7 :RedirectInfo $C.mapname("redirect"); + +/* New error code in common/ */ +#define E_XCE_REDIRECT (-100) /* Not an error — redirect instruction */ +``` + +--- + +## Part 10: libxce Smart Discovery + +### 10.1 Problem + +Currently, workers and other components are configured with a single controller address. With HA, they need to: + +1. Discover which candidate is the current primary +2. Follow redirects from standbys +3. Reconnect automatically on failover with appropriate backoff +4. Verify they're talking to the primary after connecting + +### 10.2 Configuration API + +```c +/* libxce — xce_cluster.h */ + +typedef struct { + char host[XCE_MAXNAME_LEN]; + uint32_t port; +} xce_candidate_t; + +typedef struct { + xce_candidate_t candidates[N_CONTROLLER_MAX]; + int ncandidates; /* Number of candidates configured */ + uint32_t connect_timeout_ms; /* Per-candidate connect timeout (default 2000) */ + uint32_t retry_base_ms; /* Backoff base (default 100) */ + uint32_t retry_max_ms; /* Backoff cap (default 5000) */ + int max_retries; /* Max reconnection attempts (default 10) */ +} xce_cluster_config_t; + +/* Connect to the cluster — returns fd connected to primary */ +int xce_connect_cluster(xce_cluster_config_t* conf); + +/* Connect + join as specific role (worker, scheduler, apiserver) */ +int xce_join_cluster(xce_cluster_config_t* conf, xce_peerinfo_t* myinfo); + +/* Handle disconnection — auto-reconnect with backoff + redirect following */ +int xce_reconnect(int* fd, xce_cluster_config_t* conf); + +/* Load cluster config from YAML file */ +int xce_load_cluster_config(const char* yaml_path, xce_cluster_config_t* conf); +``` + +### 10.3 Connection State Machine + +``` + ┌─────────────┐ + │ INIT │ + └──────┬──────┘ + │ Load candidates[] from config + ▼ + ┌─────────────────┐ + ┌────▶│ CONNECTING │◀──────────────┐ + │ └────────┬────────┘ │ + │ timeout or │ │ + │ connection │ │ + │ failed │ connected │ + │ │ ▼ │ + │ │ ┌─────────────┐ │ + │ │ │ CONNECTED │ │ + │ │ └──────┬──────┘ │ + │ │ │ send join request │ + │ │ ▼ │ + │ │ ┌─────────────┐ │ + │ │ │ HANDSHAKE │ │ + │ │ └──────┬──────┘ │ + │ │ │ receive reply │ + │ │ ▼ │ + │ │ ┌──────────────┐ │ + │ │ │ CHECK REPLY │ │ + │ │ └───┬──────┬───┘ │ + │ │ │ │ │ + │ │ ACCEPTED REDIRECT │ + │ │ │ │ │ + │ │ ▼ ▼ │ + │ │ ┌───────┐ ┌────────────┐ │ + │ │ │ READY │ │ Follow │ │ + │ │ │(prime)│ │ redirect → │─────┘ + │ │ └───┬───┘ │ reconnect │ + │ │ │ │ to primary │ + │ │ │ └────────────┘ + │ │ │ + │ │ │ TCP RST / connection lost + │ │ ▼ + │ │ ┌────────────┐ + └────────┴──│ RECONNECT │ + backoff │ (backoff + │ + + jitter │ jitter) │ + └────────────┘ +``` + +### 10.4 Candidate Selection Strategy + +Two phases of selection: + +**Initial connection** — Try candidates in **priority order**: +- The highest-priority alive candidate is most likely the primary +- This minimizes redirects on first connection + +**Reconnection after failure** — Try candidates in **round-robin with priority bias**: +- Start from the candidate *after* the one we were connected to (it may be down) +- Round-robin distributes reconnection load across candidates after a failover +- Avoids all workers hammering the same candidate simultaneously + +```c +/* Priority-ordered for initial, round-robin for reconnection */ +static int select_candidate(xce_cluster_config_t* conf, int* attempt, + int last_candidate_idx) +{ + if (*attempt < conf->ncandidates) { + /* First pass: priority order (index 0 first) */ + return (*attempt)++; + } + + /* Subsequent passes: round-robin starting after last connected candidate */ + int start = (last_candidate_idx + 1) % conf->ncandidates; + return (start + (*attempt)++) % conf->ncandidates; +} +``` + +### 10.5 Backoff with Jitter + +```c +/* Exponential backoff with random jitter to spread reconnect storm */ +static void backoff_jitter(int attempt, xce_cluster_config_t* conf) +{ + int delay_ms = conf->retry_base_ms * (1 << attempt); + + /* Cap at max */ + if (delay_ms > conf->retry_max_ms) + delay_ms = conf->retry_max_ms; + + /* Add jitter: random value in [0, delay/2] */ + int jitter = rand() % (delay_ms / 2 + 1); + delay_ms += jitter; + + usleep(delay_ms * 1000); +} +``` + +**Why jitter matters**: Without jitter, all 2000 workers back off for exactly the same duration and reconnect simultaneously — a thundering herd. With 50% jitter, the reconnection window spreads across 1.5× the base backoff, allowing the new primary to accept connections gradually. + +### 10.6 Full Connection Flow + +```c +int xce_connect_cluster(xce_cluster_config_t* conf) +{ + int attempt = 0; + int last_candidate = -1; + + for (;;) { + int idx = select_candidate(conf, &attempt, last_candidate); + xce_candidate_t* cand = &conf->candidates[idx]; + + /* Try to connect */ + int fd = tcp_connect(cand->host, cand->port, conf->connect_timeout_ms); + if (fd < 0) { + backoff_jitter(attempt, conf); + continue; + } + + /* Send join request to discover role */ + xce_request_t req; + xce_reply_t reply; + init_join_request(&req, &conf->myinfo); + xce_send_request(fd, &req); + xce_recv_reply(fd, &reply); + + if (reply.status == 0) { + /* Accepted by primary */ + last_candidate = idx; + return fd; + } + + if (reply.status == E_XCE_REDIRECT) { + /* Standby redirected us — follow immediately, no backoff */ + close(fd); + fd = tcp_connect(reply.data.redirect.host, + reply.data.redirect.port, + conf->connect_timeout_ms); + if (fd >= 0) { + /* Verify this is the primary */ + init_join_request(&req, &conf->myinfo); + xce_send_request(fd, &req); + xce_recv_reply(fd, &reply); + + if (reply.status == 0) { + return fd; /* Connected to primary */ + } + /* Redirect pointed to another standby? Follow again */ + } + close(fd); + continue; + } + + /* Unexpected reply — try next candidate */ + close(fd); + backoff_jitter(attempt, conf); + } +} +``` + +### 10.7 Reconnection on Failover + +When the primary fails, workers detect it via TCP RST or heartbeat timeout. The reconnection flow: + +```c +int xce_reconnect(int* fd, xce_cluster_config_t* conf) +{ + close(*fd); + *fd = -1; + + /* xce_connect_cluster handles candidate selection, + redirect following, and backoff internally */ + *fd = xce_connect_cluster(conf); + return (*fd >= 0) ? 0 : -1; +} +``` + +The worker's main loop integrates reconnection naturally: + +```c +/* Worker main loop — simplified */ +for (;;) { + int rc = xce_process_message(fd); + + if (rc < 0) { + /* Connection lost */ + TRACE("connection to controller lost, reconnecting..."); + + rc = xce_reconnect(&fd, &conf); + if (rc < 0) { + ERROR("failed to reconnect to cluster"); + /* Could exit, or keep retrying depending on policy */ + continue; + } + + /* Reconnected — re-register with controller */ + on_worker_join(cluster_cls, fd_to_endpoint(fd)); + TRACE("reconnected to controller (primary)"); + } +} +``` + +### 10.8 Candidate List Distribution + +All components (controller, worker, scheduler, API server, libxce clients) read the same `cluster.yaml`: + +```yaml +# cluster.yaml — shared across all components +controller: + candidates: + - name: ctrl-1 + host: 10.0.1.1 + port: 7000 + - name: ctrl-2 + host: 10.0.1.2 + port: 7000 + - name: ctrl-3 + host: 10.0.1.3 + port: 7000 +``` + +This file is already read by the controller for its own candidate configuration (Part 5.2). Workers and libxce clients read the same file for discovery. Distribution is an ops concern — shared NFS mount, config management tool (Ansible, Salt), or deployment pipeline. + +### 10.9 libxce Changes Summary + +| File | Change | +|------|--------| +| `src/lib/include/xce/cluster.h` | Add `xce_cluster_config_t`, `xce_candidate_t` types | +| `src/lib/include/xce/cluster.h` | Add `xce_connect_cluster()`, `xce_join_cluster()`, `xce_reconnect()`, `xce_load_cluster_config()` API | +| `src/lib/cluster.c` (new) | Implement candidate selection, redirect following, backoff+jitter, reconnection | +| `src/worker/main.c` | Replace hardcoded controller address with `xce_cluster_config_t`; add reconnection loop; defer worker removal (set `WORKER_UNAVAILABLE` instead of eager delete) | +| `src/scheduler/` | Same pattern as worker — use `xce_cluster_config_t` | +| `src/frontend/apiserver/backend/clib.go` | Use `xce_cluster_config_t` for controller discovery in `RegisterToBackend()` | + +--- + +## Implementation Phases + +### P0: Foundation (Single-Controller Persistence) + +**Goal**: Controller survives restart with job/queue/worker state intact. + +| Step | File | Description | +|------|------|-------------| +| P0-1 | `src/controller/persist.c` + `include/persist.h` | WAL writer, segment management, fsync batching | +| P0-2 | `src/lib/zarray.c` + `include/zarray.h` | COW extensions: `shadow`, `shadow_caps`, `xzarray_freeze()`, `xzarray_thaw()` | +| P0-2a | `src/lib/zarray.c` | COW lock protocol: verify xzarray `mylock_t` mutex is taken correctly inside vget/vput during COW mode; kvmgr `my_rwlock_t` is the outer coordination lock (see Part 3.8) | +| P0-3 | `src/controller/persist.c` | Checkpoint via dual zarray freeze → serialize shadow → thaw + GC; use `delaymgr` for periodic checkpoint timer (see Part 3.9) | +| P0-4 | `src/controller/persist.c` | Recovery: load snapshot + replay WAL | +| P0-5 | `src/include/xce/cluster_def.h` | Add entity terminal states: `WORKER_UNAVAILABLE`, `WORKER_HIDDEN`, `QUEUE_HIDDEN`; add `persist_t persist;` to `cluster_t` | +| P0-6 | `src/controller/persist.c` | Hook functions: `persist_job_hook_fun`, `persist_queue_hook_fun`, `persist_worker_hook_fun` | +| P0-7 | `src/controller/job.c` | Move `xce_reply()` after `AFTER_EVENT` hooks in `handle_jobsubmit()`; verify no kvmgr lock held during hooks (see Part 8.3 lock safety analysis) | +| P0-8 | `src/controller/worker.c` | Extend existing `mark_worker_unavail()` with `disconnect_time`; add `WORKER_HIDDEN` state path; already deferred — no `zarray_del` removal needed (see Part 4.10) | +| P0-8a | `src/controller/cluster.c` | Replace `zarray_del(&(c->endpoints), ep->idx)` in `handle_endpoint_error()` with deferred endpoint cleanup: close fd/SSL immediately, set `ENDPOINT_DISCONNECTED` state, cleanup slot at checkpoint (see Part 4.9) | +| P0-9 | `controller.capnp` | Add `PersistConfig`, `WALEntry`, `SnapshotHeader`, `Snapshot` schemas; add `HostState.unavail`, `HostState.hidden` | +| P0-10 | `cluster_def.h` | Add `persist_t persist;` to `cluster_t` | +| P0-11 | `cluster.c` | Wire `persist_init/recover/start/stop/fini` into `cluster_init/run/fini` | +| P0-12 | `myconf.c` + YAML config | Parse persistence configuration | +| P0-13 | `worker.c` | Wire `restore_worker()` into recovery path | +| P0-14 | Tests | WAL write/read round-trip, COW freeze/serialize/thaw, checkpoint+recover, hook invocation, retention/GC, endpoint deferred cleanup | + +### P1: HA Foundation (LSF-Style Election) + +**Goal**: Standby controller can take over when primary fails. + +| Step | File | Description | +|------|------|-------------| +| P1-1 | `src/controller/ha.c` + `include/ha.h` | Role management, `am_i_primary()`, promotion/demotion | +| P1-2 | `src/controller/ha.c` | Monitor thread: probe higher-priority candidates | +| P1-3 | `src/controller/ha.c` | Promotion sequence: stop replay, increment term, recover, start primary | +| P1-4 | `controller.capnp` | Add `term` to Request, `standby` to HostRole, `replEntry/replAck/redirect` opcodes, `RedirectInfo` struct | +| P1-5 | `cluster_def.h` | Add `ha_state_t ha;` to `cluster_t` | +| P1-6 | `cluster.c` | Wire HA into startup sequence | +| P1-7 | `join.c` | Add redirect logic in `handle_join()` — standby sends REDIRECT for worker/scheduler/apiserver join requests | +| P1-8 | Tests | Election logic, promotion sequence, term fencing, redirect response | + +### P2: Replication (Fan-Out TCP) + +**Goal**: Primary replicates state changes to standbys in real time. + +| Step | File | Description | +|------|------|-------------| +| P2-1 | `src/controller/repl.c` + `include/repl.h` | Replicator: fan-out send via endpoint/iomgr SSL pipeline, ACK tracking; `standby_conn_t.ep` replaces `standby_conn_t.fd` (see Part 6.5) | +| P2-2 | `src/controller/repl.c` | Standby receiver: `OP_replEntry` dispatched via iomgr/vtask pipeline; `OP_replAck` sent via endpoint outq (not raw `send()`) | +| P2-3 | `src/controller/repl.c` | Sync replication: `repl_wait_ack()` with timeout; only triggered for `WAL_JOB_SUBMIT` from API server connections (1-3 connections, no pool exhaustion risk) | +| P2-4 | `src/controller/persist.c` | Wire sync/async replication into `persist_wal_sync_append()` / `persist_wal_async()` | +| P2-5 | `src/controller/repl.c` | Standby catch-up: when standby reconnects via endpoint, stream missing WAL entries | +| P2-5a | `src/controller/join.c` | Resolve `handle_worker_join()` WRITE lock during hooks — currently holds workermgr write lock during `add_worker()` + AFTER_EVENT hooks; if sync replication is needed for worker join, must release write lock before blocking (or accept that worker join is always async-replicated) | +| P2-6 | Tests | Replication round-trip (via SSL pipeline), sync ACK timeout, standby catch-up | + +### P2.5: libxce Smart Discovery + +**Goal**: Workers, schedulers, and API servers can discover the primary via any candidate. + +| Step | File | Description | +|------|------|-------------| +| P2.5-1 | `src/lib/include/xce/cluster.h` | Define `xce_cluster_config_t`, `xce_candidate_t`, API functions | +| P2.5-2 | `src/lib/cluster.c` (new) | Implement `xce_connect_cluster()`: candidate selection, redirect following, backoff+jitter | +| P2.5-3 | `src/lib/cluster.c` | Implement `xce_reconnect()`: auto-reconnect with state machine | +| P2.5-4 | `src/lib/cluster.c` | Implement `xce_load_cluster_config()`: parse `cluster.yaml` candidate list | +| P2.5-5 | `src/worker/main.c` | Replace hardcoded controller address with `xce_cluster_config_t`; add reconnection loop | +| P2.5-6 | `src/scheduler/` | Same pattern as worker — use `xce_cluster_config_t` | +| P2.5-7 | `src/frontend/apiserver/backend/clib.go` | Use `xce_cluster_config_t` for controller discovery in `RegisterToBackend()` | +| P2.5-8 | Tests | Candidate selection, redirect following, backoff+jitter, reconnection | + +### P3: Integration & Hardening + +**Goal**: Production-ready persistence + HA. + +| Step | Description | +|------|-------------| +| P3-1 | End-to-end failover test: kill primary, verify standby promotes, workers reconnect via libxce discovery | +| P3-2 | Redirect test: worker connects to standby, receives redirect, reconnects to primary | +| P3-3 | WAL corruption handling: test torn writes, partial segments, CRC mismatch | +| P3-4 | Performance: benchmark 1000 job submits/sec with sync replication | +| P3-5 | Checkpoint GC: verify expired jobs/workers/queues are cleaned; verify dependents block cleanup | +| P3-6 | Admin commands: `xce hostcontrol --failover` for manual primary switch | +| P3-7 | Metrics: WAL size, replication lag, checkpoint duration, standby health, entity GC stats | + +### P4: Cleanup & Documentation + +| Step | Description | +|------|-------------| +| P4-1 | Remove dead code (Part 1) | +| P4-2 | Update `src/frontend/docs/` to remove stale PostgreSQL references | +| P4-3 | Update CLAUDE.md files with persistence/HA architecture | +| P4-4 | Operational runbook: how to add/remove standby, manual failover, disaster recovery | + +--- + +## Appendix: Data Structures & Signatures + +### A.1 Persistence Module + +```c +/* persist.h */ + +typedef struct { + int wal_fd; /* Current WAL segment fd */ + uint64_t current_lsn; /* Next LSN to assign */ + uint64_t last_crc; /* Previous record's CRC (chained) */ + uint64_t term; /* Current election term */ + char seg_path[XCE_MAXPATH_LEN]; /* Current segment path */ + uint32_t seg_offset; /* Write offset in current segment */ + persist_config_t config; + cluster_t* cluster; /* Back-reference */ + pthread_t ckpt_tid; /* Checkpoint thread */ + int running; /* Flag */ + mylock_t lock; /* Protects WAL write + LSN */ +} persist_t; + +int persist_init(persist_t* p, persist_config_t* config, cluster_t* c); +int persist_start(persist_t* p); +void persist_stop(persist_t* p); +void persist_fini(persist_t* p); + +/* Async WAL write — returns immediately after writing + sending to standbys */ +int persist_wal_async(persist_t* p, uint32_t type, void* payload, uint32_t len); + +/* Sync WAL write — waits for at least one standby ACK before returning */ +int persist_wal_sync_append(persist_t* p, uint32_t type, void* payload, uint32_t len); + +/* Checkpoint — dual zarray freeze + serialize + thaw + GC */ +int persist_checkpoint(persist_t* p, cluster_t* c); + +/* Recovery — load latest snapshot + replay WAL */ +int persist_recover(persist_t* p, cluster_t* c); + +/* Hook functions — registered in cluster_init() */ +int persist_job_hook_fun(int type, int situation, int detail, void* data, va_list ap); +int persist_queue_hook_fun(int type, int situation, int detail, void* data, va_list ap); +int persist_worker_hook_fun(int type, int situation, int detail, void* data, va_list ap); +``` + +### A.2 zarray COW Extensions + +```c +/* zarray.h — additions for COW snapshot support */ + +typedef struct { + int count; + size_t caps; + size_t delta; + int pos; + xzarray_item_t* data; /* Active array */ + xzarray_item_t* shadow; /* Frozen snapshot (NULL = normal mode) */ + size_t shadow_caps; /* Shadow capacity when in COW mode */ + destroy_fn_t killer; +} xzarray_t; + +/* Freeze: swap data → shadow, create new empty active. + Caller must hold write lock. O(1) — pointer swap only. */ +void xzarray_freeze(xzarray_t* a); + +/* Thaw: merge shadow back into active, apply GC. + should_retain(arg, item) returns true if entity should survive. + Caller must hold write lock. + Frees shadow when done. */ +typedef bool (*retain_fn_t)(void* arg, void* item); +void xzarray_thaw(xzarray_t* a, retain_fn_t should_retain, void* arg); +``` + +### A.3 Replication Module + +```c +/* repl.h */ + +#define MAX_STANDBYS (N_CONTROLLER_MAX - 1) + +typedef struct { + endpoint_t* ep; /* Full endpoint (SSL, FSM, outq) — uses iomgr/vtask pipeline */ + uint64_t acked_lsn; + int lagging; + char host[XCE_MAXNAME_LEN]; + uint32_t port; +} standby_conn_t; + +typedef struct { + standby_conn_t standbys[MAX_STANDBYS]; + int nstandbys; + mylock_t lock; + uint64_t pending_lsn; + pthread_cond_t ack_cond; + mylock_t ack_lock; +} replicator_t; + +/* Fan-out send to all standbys */ +int repl_broadcast(replicator_t* r, wal_entry_hdr_t* hdr, void* payload); + +/* Wait for at least one standby to ACK the given LSN (sync mode) */ +int repl_wait_ack(replicator_t* r, uint64_t target_lsn, uint32_t timeout_ms); + +/* Called when ACK received from a standby */ +void repl_on_ack(replicator_t* r, uint64_t acked_lsn); + +/* Add/remove standby connections (via endpoint/iomgr SSL pipeline) */ +int repl_add_standby(replicator_t* r, endpoint_t* ep, const char* host, uint32_t port); +int repl_remove_standby(replicator_t* r, endpoint_t* ep); + +/* Check standby health and lag */ +int repl_health_check(replicator_t* r); +``` + +### A.4 HA Module + +```c +/* ha.h */ + +typedef enum { CTRL_PRIMARY = 0, CTRL_STANDBY = 1 } ctrl_role_t; + +typedef struct { + ctrl_role_t role; + int my_priority; /* My index in candidates[] */ + replicator_t repl; /* Replicate to standbys (primary) */ + uint64_t applied_lsn; /* Last applied WAL entry (standby) */ + endpoint_t* primary_ep; /* Connection to current primary via endpoint/iomgr SSL pipeline (standby) */ + char primary_host[XCE_MAXNAME_LEN]; + uint32_t primary_port; + pthread_t monitor_tid; /* Monitors higher-priority candidates */ +} ha_state_t; + +/* Check if this controller should be primary */ +static int am_i_primary(cluster_t* c); + +/* Monitor thread — runs on standby, watches higher-priority candidates */ +static void* ha_monitor_thread(void* arg); + +/* Promote self from standby to primary */ +int ha_promote_to_primary(cluster_t* c); + +/* Demote self from primary to standby (admin-triggered failover) */ +int ha_demote_to_standby(cluster_t* c); +``` + +### A.5 Cap'n Proto Schema Additions + +```capnp +# Persistence configuration — embedded in ClusterConfig +struct PersistConfig $C.mapname("persist_config_t") { + walPath @0 :Text $C.mapname("wal_path"); + snapPath @1 :Text $C.mapname("snap_path"); + walSyncMs @2 :UInt32 $C.mapname("wal_sync_ms") = 10; + snapIntervalS@3 :UInt32 $C.mapname("snap_interval_s") = 60; + snapWalSizeMb@4 :UInt32 $C.mapname("snap_wal_size_mb") = 64; + finishedRetainS @5 :UInt32 $C.mapname("finished_retain_s") = 3600; + workerRetainS @6 :UInt32 $C.mapname("worker_retain_s") = 86400; + syncTimeoutMs@7 :UInt32 $C.mapname("sync_timeout_ms") = 50; +} + +# WAL entry — wire format for replication and on-disk +struct WALEntry $C.mapname("wal_entry_t") { + term @0 :UInt64; + lsn @1 :UInt64; + timestamp @2 :UInt64; + type @3 :UInt32; + payload @4 :Data; + crc @5 :UInt32; + prevCrc @6 :UInt32; +} + +# Snapshot header — first record in a checkpoint file +struct SnapshotHeader $C.mapname("snapshot_hdr_t") { + lsn @0 :UInt64; + timestamp @1 :UInt64; + term @2 :UInt64; + nJobs @3 :UInt32; + nQueues @4 :UInt32; + nWorkers @5 :UInt32; +} + +# Full snapshot — checkpoint file contents +struct Snapshot $C.mapname("snapshot_t") { + header @0 :SnapshotHeader; + jobs @1 :List(Job); + queues @2 :List(QueueConf); + workers @3 :List(Worker); + resources @4 :List(Resource); + configs @5 :List(Config); +} + +# Redirect response — standby tells client where the primary is +struct RedirectInfo $C.mapname("xce_redirect_t") { + host @0 :Text $C.mapname("host"); + port @1 :UInt32 $C.mapname("port"); +} + +# In RequestReply union — add redirect variant: +# redirect @7 :RedirectInfo $C.mapname("redirect"); + +# New error code in common/ +# #define E_XCE_REDIRECT (-100) /* Not an error — redirect instruction */ + +# Entity terminal states for deferred cleanup +# Added to HostState enum: +# unavail — Worker daemon died / disconnected +# hidden — Admin explicitly removed entity +# Added to QueueStatus enum: +# hidden — Admin deleted queue +``` + +### A.6 libxce Discovery Module + +```c +/* xce_cluster.h — libxce cluster discovery API */ + +typedef struct { + char host[XCE_MAXNAME_LEN]; + uint32_t port; +} xce_candidate_t; + +typedef struct { + xce_candidate_t candidates[N_CONTROLLER_MAX]; + int ncandidates; /* Number of candidates configured */ + uint32_t connect_timeout_ms; /* Per-candidate connect timeout (default 2000) */ + uint32_t retry_base_ms; /* Backoff base (default 100) */ + uint32_t retry_max_ms; /* Backoff cap (default 5000) */ + int max_retries; /* Max reconnection attempts (default 10) */ +} xce_cluster_config_t; + +/* Connect to the cluster — returns fd connected to primary */ +int xce_connect_cluster(xce_cluster_config_t* conf); + +/* Connect + join as specific role */ +int xce_join_cluster(xce_cluster_config_t* conf, xce_peerinfo_t* myinfo); + +/* Handle disconnection — auto-reconnect with backoff + redirect following */ +int xce_reconnect(int* fd, xce_cluster_config_t* conf); + +/* Load cluster config from YAML file */ +int xce_load_cluster_config(const char* yaml_path, xce_cluster_config_t* conf); +``` + +### A.7 New Files Summary + +| File | Purpose | Approximate Size | +|------|---------|-----------------| +| `src/controller/persist.c` | WAL writer, checkpoint, recovery, hook functions | ~800 LOC | +| `src/controller/include/persist.h` | Persistence API and types | ~80 LOC | +| `src/controller/repl.c` | Fan-out replication via endpoint/iomgr SSL pipeline | ~500 LOC | +| `src/controller/include/repl.h` | Replication API and types | ~60 LOC | +| `src/controller/ha.c` | LSF-style priority election, monitor thread | ~300 LOC | +| `src/controller/include/ha.h` | HA API and types | ~40 LOC | +| `src/lib/cluster.c` | Candidate selection, redirect following, backoff+jitter, reconnection | ~400 LOC | +| `src/lib/include/xce/cluster.h` | Discovery API and types | ~50 LOC | +| **Total new code** | | **~2230 LOC** | + +### A.8 Modified Files Summary + +| File | Change | +|------|--------| +| `src/lib/zarray.c` + `include/zarray.h` | Add `shadow`, `shadow_caps` fields; add `xzarray_freeze()`, `xzarray_thaw()`; modify `vget`, `vput`, `iter_next` for COW mode | +| `src/controller/job.c` | Move `xce_reply()` after `AFTER_EVENT` in `handle_jobsubmit()`; remove dead `NJOBS`/`jobid`/`load_jobs` body | +| `src/controller/cluster.c` | Wire `persist_init/recover/start/stop/fini` and HA into lifecycle; wire `cluster_load_jobs()` to persistence loader; replace `zarray_del(&(c->endpoints), ep->idx)` in `handle_endpoint_error()` with deferred endpoint cleanup (close fd/SSL, set `ENDPOINT_DISCONNECTED`, GC at checkpoint); use `delaymgr` for checkpoint timer | +| `src/controller/join.c` | Add redirect logic in `handle_join()` — standby sends `E_XCE_REDIRECT` + `RedirectInfo` for worker/scheduler/apiserver join requests; resolve WRITE lock held during hooks — worker join should use async replication (never blocks on `repl_wait_ack` while holding workermgr write lock) | +| `src/controller/worker.c` | Extend existing `mark_worker_unavail()` with `disconnect_time`; add `WORKER_HIDDEN` state path; `handle_worker_leave()` already calls `mark_worker_unavail()` and sets `w->ep = NULL` — no structural change needed; wire `restore_worker()` into recovery path | +| `src/include/xce/cluster_def.h` | Add `persist_t persist;` and `ha_state_t ha;` to `cluster_t`; add `WORKER_UNAVAILABLE`, `WORKER_HIDDEN` entity states | +| `src/schema/controller.capnp` | Add `PersistConfig`, `WALEntry`, `SnapshotHeader`, `Snapshot`, `RedirectInfo` schemas; add `term` to Request; add `standby` to HostRole; add `replEntry/replAck/redirect` opcodes; add `redirect` variant to `RequestReply`; add `HostState.unavail`/`HostState.hidden` | +| `src/controller/myconf.c` | Parse persistence and HA configuration from YAML | +| `src/worker/main.c` | Replace hardcoded controller address with `xce_cluster_config_t`; add reconnection loop | +| `src/scheduler/` | Same pattern as worker — use `xce_cluster_config_t` | +| `src/frontend/apiserver/backend/clib.go` | Use `xce_cluster_config_t` for controller discovery in `RegisterToBackend()` | \ No newline at end of file