From 45400c3a284fe007eeb0b52654e420fa3c684f2e Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 16 Aug 2026 10:20:02 +0000 Subject: [PATCH] fix(controller): close 6 data-safety races + apply-pump O(N^2) regression + boot TOCTOU Bundles the fixes validated this session across ha.c/persist.c/worker.c: - raft-apply-decoupling Task 7.4: apply-pump O(N^2) throughput regression fixed via a segment-read cache. - bug-1416: boot-time TOCTOU race in persist_verify_apply_correctness() closed with a recovery-complete gate that closes at process-lifecycle start, not function entry. - bug-451: HOOK_DETAIL_JOB_RUSAGE_UPDATE now has its own persist hook arm instead of silently falling through the default job-state case. - bug-1426: trim_wal_segments() also caps its trim floor at the local apply pump's own progress, not just remote standby-acked positions. - bug-1427: persist_verify_apply_correctness() extended with a worker-count check mirroring the existing job-count check (count_jobmgr_entries and count_workermgr_entries now share one count_manager_entries(var) helper). - bug-1428: persist_register_pending_apply's critical section widened to close the actual race instead of patching a symptom. - bug-1472 (remainder): WORKER_DEAD replay in worker_from_capnp() now applies mark_worker_unavail()'s mutations via a shared apply_worker_dead_mutation() helper instead of duplicating them. - bug-1474 (angle-a): on_standby_connected's stale-duplicate detection fixed for promoted vote endpoints via a shared votes_idx_for_peer() index-scale converter. Each fix independently verified via TDD (RED before fix, GREEN after). Full ASAN rebuild + controller_unit_test: 1101 tests, 0 failures. Live single-node HA deployment validated Task 7.4 and bug-1416 across 5 restart/recovery cycles (45/45 job submissions, zero APPLY_DIVERGENT fences). Co-Authored-By: Claude Sonnet 5 --- src/configs/resources.yaml | 65 ++- src/controller/ha.c | 41 +- src/controller/persist.c | 204 ++++++- src/controller/testsuite/test.c | 534 +++++++++++++++--- .../testsuite/test_apply_committed_entry.c | 102 +++- .../test_apply_correctness_verification.c | 101 +++- src/controller/testsuite/test_apply_pump.c | 112 ++++ .../testsuite/test_commit_barrier_reply.c | 52 +- .../testsuite/test_e2e_checkpoint.c | 89 ++- src/controller/testsuite/test_ha_liveness.c | 71 +++ src/controller/testsuite/test_persist_hooks.c | 73 ++- src/controller/worker.c | 72 ++- src/frontend/Makefile | 29 +- src/include/persist.h | 42 +- src/tools/deploy-ha.sh | 63 ++- src/tools/xce-build.sh | 7 +- src/worker/src/node_information.c | 29 +- src/worker/src/xcecd.c | 8 +- 18 files changed, 1499 insertions(+), 195 deletions(-) diff --git a/src/configs/resources.yaml b/src/configs/resources.yaml index 1b2b00a5..360d7a8b 100644 --- a/src/configs/resources.yaml +++ b/src/configs/resources.yaml @@ -11,9 +11,9 @@ resources: - name: mem type: numeric interval: 60 - consumable: false + consumable: true order: decrease - description: "available memory in MB (built-in dynamic load index)" + description: "available memory in bytes (built-in dynamic load index; CLI display converts to a human-readable unit)" # built-in: worker load indices registered by # worker/src/resource_sync.c init_builtin_load_indices() @@ -21,22 +21,71 @@ resources: type: numeric interval: 60 consumable: false - order: decrease + order: increase description: "1-minute load average (built-in dynamic load index)" + - name: r15m + type: numeric + interval: 60 + consumable: false + order: increase + description: "15-minute load average (built-in dynamic load index)" + - name: r15s type: numeric interval: 60 consumable: false - order: decrease + order: increase description: "15-second load average (built-in dynamic load index)" - - name: tmp + - name: ut type: numeric interval: 60 consumable: false + order: increase + description: "cpu utilization" + + - name: pg + type: numeric + interval: 60 + consumable: false + order: increase + description: "page ratio" + + - name: ls + type: numeric + interval: 60 + consumable: false + order: increase + description: "login count" + + - name: it + type: numeric + interval: 60 + consumable: false + order: increase + description: "idle time" + + - name: swp + type: numeric + interval: 60 + consumable: true order: decrease - description: "available tmp disk space in MB (built-in dynamic load index)" + description: "swap size" + + - name: io + type: numeric + interval: 60 + consumable: false + order: increase + description: "io rate" + + - name: tmp + type: numeric + interval: 60 + consumable: true + order: decrease + description: "available tmp disk space in bytes (built-in dynamic load index; CLI display converts to a human-readable unit)" # built-in host attributes, registered per-worker by # controller/worker.c add_hostattrs_resvals() @@ -62,11 +111,11 @@ resources: - name: maxswap type: numeric - description: "max swap in MB (built-in host attribute)" + description: "max swap in bytes (built-in host attribute; CLI display converts to a human-readable unit)" - name: maxtmp type: numeric - description: "max tmp disk space in MB (built-in host attribute)" + description: "max tmp disk space in bytes (built-in host attribute; CLI display converts to a human-readable unit)" - name: ndisks type: numeric diff --git a/src/controller/ha.c b/src/controller/ha.c index 61937f0b..092c8d3f 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -9209,6 +9209,19 @@ int standby_ep_err_handler_primary_for_test(endpoint_t *ep) } #endif /* XCE_TESTING */ +/* + * votes_idx_for_peer: convert an ABSOLUTE candidate index to the RELATIVE + * votes[] index (self excluded) used by a promoted vote ep's peer_idx -- see + * raft_cb_send_vote_request. Shared by on_standby_connected's stale-duplicate + * check and primary_heartbeat_fn's reconnect-sweep check, which both need to + * compare a standby_ep_ctx_t's peer_idx (absolute) against a promoted vote + * ep's peer_idx (relative) on the same scale. + */ +static int votes_idx_for_peer(ha_state_t *ha, int abs_idx) +{ + return (abs_idx < ha->my_priority) ? abs_idx : abs_idx - 1; +} + /* * on_standby_connected: iomgr on_connected callback for outbound standby * endpoints created by the heartbeat task. @@ -9248,6 +9261,25 @@ static void on_standby_connected(endpoint_t *ep) */ endpoint_t *stale_same_peer = NULL; if (ha->standby_endpoints) { + /* bug-1474 (angle-a): sctx->peer_idx is ALWAYS the ABSOLUTE candidate + * index (standby_ep_ctx_t's own contract, doc comment above). But a + * DIFFERENT entry already in standby_endpoints may be a PROMOTED VOTE + * EP (raft_cb_become_leader), whose peer_idx is the RELATIVE votes[] + * index (self excluded) -- see raft_cb_send_vote_request. Comparing + * the two peer_idx values directly, as this loop previously did, only + * happens to work when cur is itself a standby ep (both absolute); + * against a promoted vote ep for the SAME physical peer, the raw + * comparison is comparing two different index scales and can never + * match -- the stale/duplicate promoted-vote channel then survives + * alongside this fresh one, exactly the "two live outbound state-push + * channels to one peer" hazard this function's own doc comment above + * describes and this loop exists to prevent. Convert sctx's absolute + * index to the relative scale before comparing against a promoted + * vote ep -- same votes_idx_for_peer() primary_heartbeat_fn's own + * reconnect-sweep check already uses for the identical discrimination + * (see its "Two endpoint types can be in standby_endpoints" comment, + * the established precedent this mirrors). */ + int votes_idx_for_new = votes_idx_for_peer(ha, sctx->peer_idx); pthread_mutex_lock(&ha->lock); int already = 0; for (int i = 0; i < ha->standby_endpoints->count; i++) { @@ -9256,7 +9288,12 @@ static void on_standby_connected(endpoint_t *ep) /* Different ep, same peer → stale duplicate to evict. */ standby_ep_ctx_t *cur_sctx = cur ? (standby_ep_ctx_t *)cur->data : NULL; - if (cur_sctx && cur_sctx->peer_idx == sctx->peer_idx) { + if (!cur_sctx) continue; + int cur_is_promoted_vote_ep = (cur_sctx->ctx != NULL); + int match = cur_is_promoted_vote_ep + ? (cur_sctx->peer_idx == votes_idx_for_new) + : (cur_sctx->peer_idx == sctx->peer_idx); + if (match) { stale_same_peer = cur; zarray_del(ha->standby_endpoints, i); /* count shrank by one; the surviving scan continues but we @@ -9793,7 +9830,7 @@ static int primary_heartbeat_fn(void *arg) * - Heartbeat standby eps: ep->data = standby_ep_ctx_t (ctx==NULL), * peer_idx is absolute candidate index. Compare directly with i. */ - int votes_idx_for_i = (i < ha->my_priority) ? i : i - 1; + int votes_idx_for_i = votes_idx_for_peer(ha, i); int found = 0; if (ha->standby_endpoints) { pthread_mutex_lock(&ha->lock); diff --git a/src/controller/persist.c b/src/controller/persist.c index 112e1fbe..e65ca8f2 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -7,6 +7,7 @@ #include "xce/errnox.h" #include "xce/types.h" #include "controller.capnp.h" +#include "typemap.h" /* xce_job_rusages_t -- bug-451 RUSAGE hook arm */ #include "xlist.h" #include "cluster.h" #include "commit.h" /* commit_submit RETAIN registration (barrier-rollout 1.2) */ @@ -2622,6 +2623,40 @@ static void trim_wal_segments(persist_t* p) uint64_t floor = repl_trim_lsn(&p->repl, active_seg_lsn, p->config.snap_interval_s); + /* bug-1426: also cap the floor at the LOCAL apply pump's own progress + * (last_applied_lsn), not just remote standby-acked positions. + * repl_trim_lsn's floor only reflects what OTHER members have + * durably received -- it has no idea whether this member's own live + * apply pump has actually caught up locally. If the pump is lagging + * (e.g. mid-recovery, or a slow apply burst) when a trim fires, a + * segment below the standby-floor but AT OR ABOVE the pump's own + * still-unread position could be deleted while apply_pump_read_entry's + * cache holds an open fd into it -- POSIX unlink-keeps-fd-alive lets + * the pump keep reading (no immediate corruption, no wrong data), but + * a FRESH read of the same range (a restart's persist_recover() replay, + * or the pump's own cache falling back to its slow path after a miss) + * would find the segment already gone -- a genuine, if narrow, apply- + * correctness gap: what the live process quietly caught up to and what + * a fresh read of disk state can see would diverge. Mirrors the exact + * same "floor = min(...)" pattern already used above for the standby/ + * disconnect-guard case -- same imprecision (segment-granularity, not + * exact-entry-granularity), same established acceptable tradeoff. + * + * +1, not the raw value: last_applied_lsn == N means the entry AT lsn N + * has been fully applied -- unlike repl_trim_lsn's floor (an acked_lsn + * where equality means "the standby might still need this segment, + * preserve it" -- see the seg_start_lsn < floor check below, which + * already treats equality as preserve for that input by construction), + * the LOCAL apply cursor's equality case means the opposite: this + * process is DONE with that entry and will never re-read it. Feeding + * last_applied_lsn+1 into the same shared "< floor" check gets the + * correct delete-on-equal semantics for this term specifically, + * without needing a second, differently-shaped comparison. */ + uint64_t apply_floor = + atomic_load_explicit(&p->last_applied_lsn, memory_order_acquire) + 1; + if (apply_floor < floor) + floor = apply_floor; + DIR* d = opendir(p->wal_path); if (!d) return; @@ -3845,7 +3880,14 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, * notification skipped rather than guessed at */ } t_apply_seam_notifying = false; - return 0; + /* bug-1427: propagate rc (1 = genuinely created a new worker via + * worker_from_capnp()'s restore_worker() path, 0 = an UPDATE/DEAD + * applied to an already-existing worker) instead of the previous + * hardcoded 0 — replay_wal()'s new-worker tally below relies on + * this distinction, exactly mirroring WAL_JOB_SUBMIT's apply_rc. + * Every existing caller only checks sign / -EEXIST, never the + * exact non-negative value, so this is a safe, additive change. */ + return rc; } case WAL_CONFIG_MARKER: { @@ -4620,16 +4662,26 @@ void persist_apply_pump_run_sync(persist_t* p) * instead would tally that duplicate as a second job and spuriously fail the * post-replay comparison against jobmgr's actual growth (APPLY_DIVERGENT). * + * new_worker_joins_out (bug-1427, when non-NULL) is the worker analog of + * new_job_submits_out — the number of WAL_WORKER_CTRL entries this call + * dispatched that genuinely created a new worker (apply_committed_entry()'s + * return value == 1, per worker_from_capnp()'s bug-1427 contract), not raw + * WAL_WORKER_CTRL occurrences (which also include UPDATE/DEAD entries and + * idempotent JOIN replays, neither of which grows the worker table). + * * Returns number of entries replayed (>= 0) or negative errno on I/O error. */ static int replay_wal(persist_t* p, uint64_t from_lsn, recover_verdict_t* verdict_out, - uint64_t* new_job_submits_out) + uint64_t* new_job_submits_out, + uint64_t* new_worker_joins_out) { /* Default to CLEAN; only a structural corruption flips it to TRUNCATED. */ if (verdict_out) *verdict_out = RECOVER_CLEAN; if (new_job_submits_out) *new_job_submits_out = 0; + if (new_worker_joins_out) + *new_worker_joins_out = 0; /* --- Collect and sort segment filenames by start LSN --- */ DIR* d = opendir(p->wal_path); @@ -4997,6 +5049,19 @@ static int replay_wal(persist_t* p, uint64_t from_lsn, apply_rc > 0) (*new_job_submits_out) += (uint64_t)apply_rc; + /* bug-1427: worker analog of the WAL_JOB_SUBMIT tally above + * — apply_rc == 1 means apply_committed_entry()'s + * WAL_WORKER_CTRL case genuinely created a new worker (a + * JOIN against a jobmgr slot that didn't exist yet); + * apply_rc == 0 means an UPDATE/DEAD applied to an + * already-existing worker (or a no-hook idempotent case + * already excluded above via the apply_rc < 0 branch and + * worker_from_capnp's -EEXIST-maps-to-0 path) — neither + * grows the worker table, so only == 1 is tallied. */ + if (hdr.type == WAL_WORKER_CTRL && new_worker_joins_out && + apply_rc == 1) + (*new_worker_joins_out)++; + replayed++; } @@ -5035,47 +5100,62 @@ static int replay_wal(persist_t* p, uint64_t from_lsn, return replayed; } -/* raft-apply-decoupling task 5.1: cheap read-lock count of c->jobmgr's live - * entries, mirroring collect_job_frames' iteration idiom (this file, above) - * without collecting or filtering anything. Used only to derive the - * post-replay materialized-entity expectation (never on a hot path). Safe on - * a NULL cluster or jobmgr (a bare unit-test persist_t) — returns 0. */ -static uint64_t count_jobmgr_entries(cluster_t* c) +/* raft-apply-decoupling task 5.1 / bug-1427: cheap read-lock count of a + * manager's live entries, mirroring collect_job_frames' iteration idiom + * (this file, above) without collecting or filtering anything. `mgr` is an + * rtti `var` handle (c->jobmgr or c->workermgr) — lock/iter_* are generic + * over the stored element type, so one counter serves both tables. Used only + * to derive the post-replay materialized-entity expectation (never on a hot + * path). Safe on a NULL mgr — returns 0. */ +static uint64_t count_manager_entries(var mgr) { - if (!c || !c->jobmgr) + if (!mgr) return 0; - lock(c->jobmgr, RTTI_LOCK_READ); + lock(mgr, RTTI_LOCK_READ); iterator_t it; - int irc = iter_make(&it, c->jobmgr); + int irc = iter_make(&it, mgr); if (irc < 0) { - unlock(c->jobmgr); + unlock(mgr); return 0; /* empty table */ } uint64_t n = 0; for (;;) { - job_t* j = NULL; - int nrc = iter_next(&it, (void**)&j); + void* e = NULL; + int nrc = iter_next(&it, &e); if (nrc < 0 || nrc == RTTI_ITER_NO_DATA) break; - if (j == NULL) + if (e == NULL) continue; n++; } iter_destroy(&it); - unlock(c->jobmgr); + unlock(mgr); return n; } +static uint64_t count_jobmgr_entries(cluster_t* c) +{ + return (!c) ? 0 : count_manager_entries(c->jobmgr); +} + +/* bug-1427: worker analog of count_jobmgr_entries above — same NULL-safety + * contract, over c->workermgr instead of c->jobmgr. */ +static uint64_t count_workermgr_entries(cluster_t* c) +{ + return (!c) ? 0 : count_manager_entries(c->workermgr); +} + /* persist_verify_apply_correctness — see persist.h for the full contract - * doc (raft-apply-decoupling task 5.1, Req 8.1/8.2). Pure comparison: no - * logging, no fence side effects. */ + * doc (raft-apply-decoupling task 5.1, Req 8.1/8.2; worker half bug-1427). + * Pure comparison: no logging, no fence side effects. */ apply_verdict_t persist_verify_apply_correctness( persist_t* p, const apply_expectation_t* exp, - uint64_t jobmgr_count_before_replay) + uint64_t jobmgr_count_before_replay, + uint64_t workermgr_count_before_replay) { if (!p || !exp) return APPLY_CLEAN; /* defensive: nothing to verify */ @@ -5100,6 +5180,22 @@ apply_verdict_t persist_verify_apply_correctness( if (actual_new_jobs != exp->expected_new_job_count) return APPLY_DIVERGENT; + /* bug-1427: worker-count check, mirroring the job-count check above + * exactly. Independently skipped (not just via the shared p->cluster + * check above) when workermgr itself is unset, mirroring the jobmgr + * NULL-safety convention. */ + if (!p->cluster->workermgr) + return APPLY_CLEAN; + + uint64_t workermgr_count_after = count_workermgr_entries(p->cluster); + if (workermgr_count_after < workermgr_count_before_replay) + return APPLY_DIVERGENT; /* table shrank — impossible on a clean replay */ + + uint64_t actual_new_workers = + workermgr_count_after - workermgr_count_before_replay; + if (actual_new_workers != exp->expected_new_worker_count) + return APPLY_DIVERGENT; + return APPLY_CLEAN; } @@ -5242,6 +5338,8 @@ int persist_recover(persist_t* p) atomic_store_explicit(&p->last_applied_lsn, p->checkpoint_lsn, memory_order_release); uint64_t jobmgr_count_before_replay = count_jobmgr_entries(p->cluster); + uint64_t workermgr_count_before_replay = + count_workermgr_entries(p->cluster); /* bug-1427 */ /* --- Step 3: replay WAL entries after checkpoint_lsn --- */ /* Recovery-integrity verdict (ha-commit-durability task 3.1, Req 2.1): @@ -5253,7 +5351,9 @@ int persist_recover(persist_t* p) * entries above it. */ recover_verdict_t verdict = RECOVER_CLEAN; uint64_t new_job_submits = 0; - int replayed = replay_wal(p, p->checkpoint_lsn, &verdict, &new_job_submits); + uint64_t new_worker_joins = 0; /* bug-1427 */ + int replayed = replay_wal(p, p->checkpoint_lsn, &verdict, &new_job_submits, + &new_worker_joins); if (replayed < 0) return replayed; @@ -5316,16 +5416,19 @@ int persist_recover(persist_t* p) apply_expectation_t apply_exp = { .expected_last_applied_lsn = apply_cursor, .expected_new_job_count = new_job_submits, + .expected_new_worker_count = new_worker_joins, /* bug-1427 */ }; p->apply_verdict = persist_verify_apply_correctness( - p, &apply_exp, jobmgr_count_before_replay); + p, &apply_exp, jobmgr_count_before_replay, workermgr_count_before_replay); if (p->apply_verdict == APPLY_DIVERGENT) { ERROR("persist_recover: APPLY_DIVERGENT — materialized state does " "not match the committed log tip (expected last_applied_lsn=%" - PRIu64 " new_job_count=%" PRIu64 "); a hidden apply defect is " - "present (ha-commit-durability recover_verdict=%s)", + PRIu64 " new_job_count=%" PRIu64 " new_worker_count=%" PRIu64 + "); a hidden apply defect is present (ha-commit-durability " + "recover_verdict=%s)", apply_exp.expected_last_applied_lsn, apply_exp.expected_new_job_count, + apply_exp.expected_new_worker_count, (verdict == RECOVER_TRUNCATED) ? "RECOVER_TRUNCATED" : "RECOVER_CLEAN"); @@ -6381,14 +6484,37 @@ void persist_register_pending_apply(persist_t* p, uint64_t lsn, * Fire the hand-off immediately, synchronously, on THIS thread instead * -- never leave a continuation unresolved (controllers never * self-terminate; the same "fail loud and complete the work" posture - * commit.c's dispatch_continuation fallback uses). */ + * commit.c's dispatch_continuation fallback uses). + * + * bug-1428: this self-heal check MUST run under pending_applies_lock, + * not before acquiring it. The window this closes: this thread reads + * last_applied_lsn (sees "not yet applied"), then -- BEFORE it acquires + * the lock below to insert -- the apply pump's thread advances + * last_applied_lsn past `lsn` AND completes its own + * persist_take_pending_apply(lsn) call (finding nothing, since this + * thread hasn't inserted yet). This thread then inserts a node the walk + * will never revisit -- a permanently orphaned hand-off, the exact + * class of hang the round-2 self-heal above was meant to eliminate, + * reintroduced by the gap between the unlocked check and the locked + * insert. Holding pending_applies_lock across BOTH the check and the + * insert makes the two mutually exclusive with + * persist_take_pending_apply's own locked scan: either this thread's + * check runs before the pump's take (sees "not yet applied", inserts, + * and the pump's later take -- now blocked on this same lock until this + * critical section ends -- finds the freshly-inserted node), or it runs + * after (sees "already applied", self-heals here instead of inserting, + * and the pump's take correctly finds nothing) -- never the interleaved + * gap that orphans it. */ + pthread_mutex_lock(&p->pending_applies_lock); if (lsn <= persist_last_applied_lsn(p)) { + pthread_mutex_unlock(&p->pending_applies_lock); on_applied(ctx, lsn); return; } node = calloc(1, sizeof(*node)); if (node == NULL) { + pthread_mutex_unlock(&p->pending_applies_lock); ERROR("persist_register_pending_apply: OOM registering the apply " "hand-off for lsn=%lu -- the entry is still durably committed " "and will still be applied by a future restart replay, but no " @@ -6400,7 +6526,6 @@ void persist_register_pending_apply(persist_t* p, uint64_t lsn, node->on_applied = on_applied; node->ctx = ctx; - pthread_mutex_lock(&p->pending_applies_lock); node->next = (pending_command_apply_t*)p->pending_applies; p->pending_applies = node; pthread_mutex_unlock(&p->pending_applies_lock); @@ -6573,8 +6698,35 @@ int persist_job_hook_fun(int type, int situation, int detail, js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state; break; } + case HOOK_DETAIL_JOB_RUSAGE_UPDATE: { + /* args: cluster_t*, xce_job_rusages_t* -- bug-451: a DIFFERENT + * va_list shape than every other case below. job.c's + * update_job_rusages() fires this detail with the raw rusages + * payload it received (not the job_t* it already holds + * write-locked), because this hook's other two listeners + * (sched_job_rusage_update_fun / apiserver_job_rusage_update_fun) + * genuinely need the full payload (pids/pgids/usages), not just a + * job_t*. Reading the second vararg as job_t* here (as the + * default arm below does) would misinterpret an + * xce_job_rusages_t* as a job_t* and read state from a byte + * offset past the smaller struct's end. Look the job up by + * rusages->jobid via the cluster_t* this call already carries -- + * the same jobmgr lookup job.c's own caller already did moments + * earlier -- to recover the real state for this generic + * wal_jobstate_t fact. */ + cluster_t* c = va_arg(ap, cluster_t*); + xce_job_rusages_t* rusages = va_arg(ap, xce_job_rusages_t*); + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)rusages->jobid, (void**)&j); + unlock(c->jobmgr); + if (j == NULL) + return 0; /* job gone by the time this fired -- nothing to log */ + js.jobid = rusages->jobid; js.state = (uint32_t)j->job.state; + break; + } default: { - /* STOP, RESUME, UPDATE, CKPT, PREASON, RUSAGE — args: cluster_t*, job_t* */ + /* STOP, RESUME, UPDATE, CKPT, PREASON — args: cluster_t*, job_t* */ (void)va_arg(ap, cluster_t*); job_t* j = va_arg(ap, job_t*); js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state; diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index a1b1052a..ce850d49 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -36,9 +36,10 @@ typedef int (*cmd_func_t)(int, char*[]); typedef struct { - char* name; - cmd_func_t func; - char* doc; + char* name; + cmd_func_t func; + char* doc; + const char* usage; } command_t; typedef struct { @@ -189,20 +190,370 @@ static setting_t settings[] = { { (char*)NULL, (char*)NULL, (char*)NULL } }; +/* Detailed per-command `help ` text (design: keep these OUT of the + * command_t table initializers below so the tables stay readable). Each + * constant's shape: first line = command name + flag/positional syntax, + * then one indented line per flag/positional explaining what it does and + * whether it's required, then (where useful) an `example:` line. Derived + * from actually reading each cmd_* handler's getopt() string and switch + * cases (or its positional argv[] handling) -- not guessed from the name. + * + * Commands that are identical (same func, same semantics) across more than + * one of the 4 command tables share ONE constant here, referenced from every + * table, so the text can't drift out of sync between contexts. */ + +static const char USAGE_HELP[] = + "help []\n" + " optional command name; prints its detailed usage instead of\n" + " the one-line summary\n" + " With no argument, lists every command in the current context with its\n" + " one-line summary (unaffected by this detailed-usage text).\n" + "example: help resval\n"; + +static const char USAGE_QUIT[] = + "quit\n" + " (no arguments)\n" + " Leaves the current context (if any) and exits xce_test.\n"; + +static const char USAGE_ECHO[] = + "echo []\n" + " zero or more words, printed space-joined followed by a\n" + " newline (always succeeds)\n" + " Useful to print the $jobid / $jobcount / $? auto-variables in scripts.\n" + "example: echo job is $jobid\n"; + +static const char USAGE_UP[] = + "..\n" + " (no arguments)\n" + " Leaves the current context and returns to its parent, closing the\n" + " underlying client connection for this context.\n"; + +static const char USAGE_WALDUMP[] = + "waldump [more...]\n" + " one or more paths: a seg-*.wal file, a wal/ dir, or a\n" + " node/data dir\n" + " Offline, read-only WAL inspector: lists every entry, decodes\n" + " JOB_SUBMIT jobids, and flags LSN gaps and a CRC-broken tail.\n" + " Returns nonzero if any path fails to load or any gap/CRC break is\n" + " found.\n" + "example: waldump /var/xce/ctrl-1/wal\n"; + +static const char USAGE_WALCOMPARE[] = + "walcompare [--report] [--committed ] [more...]\n" + " --report also emit a full per-member (lsn,term,pcrc,status)\n" + " machine-checkable table, not just the verdict\n" + " --committed use LSN N as the shared committed boundary instead\n" + " of deriving it as the minimum CRC-intact tip across\n" + " members\n" + " 2-16 member paths (seg-*.wal file, wal/ dir, or\n" + " node/data dir)\n" + " Returns 0 if all members' committed WAL prefixes are identical,\n" + " nonzero on divergence or load error.\n" + "example: walcompare --report /var/xce/ctrl-1/wal /var/xce/ctrl-2/wal\n"; + +static const char USAGE_SNAPVERIFY[] = + "snapverify [ ...]\n" + " one or more snap-*.capnp or snap-*.bin checkpoint files\n" + " Offline structural validator: walks the same layout\n" + " persist_load_snapshot() parses, without restoring anything. Reports\n" + " the byte offset of the first structural defect and how many frames\n" + " parsed cleanly before it.\n" + " Returns nonzero if any file fails to verify.\n" + "example: snapverify /var/xce/ctrl-1/snap/snap-000123.capnp\n"; + +static const char USAGE_SNAPDIFF[] = + "snapdiff \n" + " the primary's own locally generated checkpoint\n" + " (snap-*.capnp)\n" + " the copy a standby assembled from the wire\n" + " (snap-*.bin)\n" + " Byte-level comparator; reports the first differing offset and\n" + " whether it lands on an OPCode_snapshotTransfer chunk boundary.\n" + " Returns 0 if identical, nonzero otherwise.\n" + "example: snapdiff snap-000123.capnp snap-000123.bin\n"; + +static const char USAGE_CONFIGCOMPARE[] = + "configcompare [ ...]\n" + " a member's workdir/name conf-root directory (NOT a raw\n" + " config file path) -- the dir under which conf/current\n" + " resolves\n" + " Recomputes each member's active-config sha256 fresh from its own\n" + " on-disk bytes and reports whether all members match.\n" + " Returns 0 if all members are present and identical, nonzero if any\n" + " member is unreadable or divergent.\n" + "example: configcompare /var/xce/ctrl-1 /var/xce/ctrl-2 /var/xce/ctrl-3\n"; + +static const char USAGE_WORKER[] = + "worker\n" + " (no arguments)\n" + " Enters the worker context: connects to the controller cluster as a\n" + " worker client and starts the local delay-manager housekeeping loop.\n"; + +static const char USAGE_SCHEDULER[] = + "scheduler\n" + " (no arguments)\n" + " Enters the scheduler context: connects to the controller cluster as\n" + " a scheduler client.\n"; + +static const char USAGE_CLIENT[] = + "client\n" + " (no arguments)\n" + " Enters the client context: connects to the controller cluster as a\n" + " regular API client (job submission, resource/worker queries, etc.).\n"; + +static const char USAGE_SET[] = + "set\n" + "set \n" + "set -e | +e | continue-on-error\n" + " (no args) list all settings in the current context\n" + " (current value + default)\n" + " set setting (e.g. masters, secret,\n" + " myname) to \n" + " -e abort the read loop on the first failing\n" + " command (errexit)\n" + " +e | continue-on-error keep the read loop running after a\n" + " failing command\n" + "example: set masters ctrl1=localhost:9991,ctrl2=localhost:9992\n"; + +static const char USAGE_JOIN_WORKER[] = + "join [-f ] [-c ] [-m ] [-s ]\n" + " [-t ] [-d ]\n" + " -f relative CPU speed factor (default 1.0)\n" + " -c max CPU count (default 4)\n" + " -m max memory in bytes (default 64)\n" + " -s max swap in bytes (default 1024)\n" + " -t max tmp space in bytes (default 2048)\n" + " -d number of local disks (default 1)\n" + " Joins the cluster as a worker (hosttype LINUX64, hostmodel Intel64).\n" + "example: join -c 8 -m 16384\n"; + +static const char USAGE_JOBSTART[] = + "jstart -j [-p ]\n" + " -j jobid to report a synthetic job-start fact for\n" + " (required)\n" + " -p pid to report (default: this xce_test process's own\n" + " pid)\n" + "example: jstart -j 12345 -p 4242\n"; + +static const char USAGE_JOBFINISH[] = + "jfinish -j [-c ]\n" + " -j jobid to report a synthetic job-finish fact for\n" + " (required)\n" + " -c exit code to report (default 0)\n" + "example: jfinish -j 12345 -c 0\n"; + +static const char USAGE_JOBFAIL[] = + "jfail -j [-c ] [-s ]\n" + " -j jobid to report a synthetic job-fail fact for\n" + " (required)\n" + " -c failure code to report (default 0)\n" + " -s exec-stage code to report (default\n" + " JOB_EXEC_STAGE_POST_EXEC)\n" + "example: jfail -j 12345 -c 1\n"; + +static const char USAGE_RESUPDATE[] = + "resupdate -n -v [-n -v ...]\n" + " -n resource name to update (e.g. slots, mem, cpuf,\n" + " maxswap, maxtmp, maxcpus, ndisks, type, model,\n" + " dynres1, dynres2, r1m, r15s); may repeat for\n" + " multiple resources\n" + " -v value for the immediately preceding -n (numeric for\n" + " most resources, string for type/model)\n" + " Order matters: pair each -n with its -v right after it.\n" + "example: resupdate -n slots -v 4 -n mem -v 8192\n"; + +static const char USAGE_HELLO[] = + "hello [-c ] [-m ]\n" + " -c number of times to send the hello (default 1)\n" + " -m greeting text to send (default \"Greeting from xce\n" + " client\")\n" + "example: hello -c 3 -m \"ping\"\n"; + +static const char USAGE_RESLIST[] = + "reslist\n" + " (no arguments)\n" + " Prints all cluster-wide resources (name, flags, interval,\n" + " description).\n"; + +static const char USAGE_WORKERLIST[] = + "workerlist\n" + " (no arguments)\n" + " Prints all joined workers (host, status, host type/model, cpu\n" + " factor, max cpu/mem/swap/tmp, ndisks).\n"; + +static const char USAGE_QCTRL[] = + "qctrl -q [-a] [-i] [-o] [-c]\n" + " -q queue name to control (required)\n" + " -a set queue ACTIVE (accepts dispatch)\n" + " -i set queue INACTIVE (rejects dispatch)\n" + " -o set queue OPEN (accepts new submissions)\n" + " -c set queue CLOSE (rejects new submissions)\n" + "example: qctrl -q normal -i\n"; + +static const char USAGE_HOSTCTRL[] = + "hostctrl -h (-o|-c)\n" + " -h host to administratively control (required)\n" + " -o open the host (allow scheduling)\n" + " -c close the host (admin-close; shows CLOSED in\n" + " workerlist)\n" + "example: hostctrl -h node01 -c\n"; + +static const char USAGE_UPGRADECONFIG[] = + "upgradeconfig \n" + " path to a capnp-encoded ClusterConfig blob (same\n" + " format `xceadmin config build` produces and\n" + " deploy-ha.sh installs)\n" + " Computes its SHA-256 and submits it as a configuration upgrade via\n" + " OP_upgradeConfig.\n" + "example: upgradeconfig /var/xce/cluster.bin\n"; + +static const char USAGE_JOBLIST[] = + "joblist\n" + " (no arguments)\n" + " Prints all jobs in the cluster (JobId, user, queue, state, times,\n" + " ntasks, command) and captures the count into $jobcount.\n"; + +static const char USAGE_EXPECT_JOBS[] = + "expect-jobs \n" + " non-negative integer; asserts the current job count equals N\n" + " Fails (nonzero) on a malformed N, a failed joblist query, or a count\n" + " mismatch.\n" + "example: expect-jobs 3\n"; + +static const char USAGE_EXPECT_JOB[] = + "expect-job \n" + " a JobId string (as printed by joblist / captured into\n" + " $jobid); asserts a job with this id is present in the\n" + " current job list\n" + " Fails (nonzero) on a missing argument, a failed joblist query, or if\n" + " the job is not found.\n" + "example: expect-job $jobid\n"; + +static const char USAGE_WAIT_PRIMARY[] = + "wait-primary []\n" + " optional max wait in milliseconds (default 30000);\n" + " must be a non-negative integer\n" + " Blocks, polling every 200 ms, until the AUTHORITATIVE controller\n" + " status probe reaches the serving primary (a standby/hint answer does\n" + " not count), or the timeout elapses.\n" + " Returns 0 once a primary answers, nonzero (ETIMEDOUT) on timeout.\n" + "example: wait-primary 10000\n"; + +static const char USAGE_CTRLSTATUS[] = + "ctrlstatus\n" + " (no arguments)\n" + " Prints the responding controller's identity/role, cluster name,\n" + " peer topology, config version/checksum, and HA health (term,\n" + " committedLsn, appliedLsn, fenced) from a single any-node status\n" + " read.\n"; + +static const char USAGE_SHUTDOWN[] = + "shutdown\n" + " (no arguments)\n" + " Shuts down the xce controller and, on success, leaves the client\n" + " context (same as running `..`).\n"; + +static const char USAGE_JOBRUN[] = + "jobrun -j [-m [:]] [-m ...]\n" + " -j jobid to dispatch (required, must be > 0)\n" + " -m [:] host to run tasks on, optional :ntasks count\n" + " (default 1); may be repeated up to 5 times\n" + "example: jobrun -j 12345 -m node01:2 -m node02\n"; + +static const char USAGE_JOBSUB[] = + "jobsub -u -q -c [-R ]\n" + " [-n [,]] [-x ] [-a ]\n" + " -u submitting user (required)\n" + " -q target queue (required)\n" + " -c command line to run (required)\n" + " -R resource requirement expression (default\n" + " \"type == any\")\n" + " -n [,] task count, or a min,max range (default 1)\n" + " -x raw exdata JSON string (overrides the built-in\n" + " PATH env)\n" + " -a job array size (default 1)\n" + " On success, captures the submitted JobId into the $jobid\n" + " auto-variable.\n" + "example: jobsub -u alice -q normal -c \"sleep 10\" -n 1,4\n"; + +static const char USAGE_RESVAL[] = + "resval -n [-m ]\n" + " -n resource name to query (required)\n" + " -m host to query (required unless the resource is a\n" + " shared/cluster-wide resource, not per-host)\n" + "example: resval -n slots -m myhost\n"; + +static const char USAGE_JOBKILL[] = + "jobkill -j [-s ]\n" + " -j jobid to signal (required)\n" + " -s signal number to send (default SIGTERM)\n" + "example: jobkill -j 12345 -s 9\n"; + +static const char USAGE_JOBCKPT[] = + "jobckpt -j [-k] [-m ]\n" + " -j jobid to checkpoint (required)\n" + " -k kill the job immediately after checkpointing\n" + " -m checkpoint method name (default \"default\")\n" + "example: jobckpt -j 12345 -m default\n"; + +static const char USAGE_JOBINFO[] = + "jobinfo -j [-j ...]\n" + " -j jobid to show info for; may be repeated to show\n" + " several jobs in one call\n" + "example: jobinfo -j 12345\n"; + +static const char USAGE_TASKRUN[] = + "taskrun [-j ] [-m [:]]... [-u ] [-c ]\n" + " [-x ] [-d ] [-t ]\n" + " -j jobid this task run belongs to\n" + " -m [:] host to run on, optional :ntasks count\n" + " (default 1); may be repeated up to 5 times\n" + " -u exec user, applied to every host\n" + " -c command line, applied to every host\n" + " -x exdata string, applied to every host\n" + " -d exec home/cwd directory, applied to every host\n" + " -t launch type (default \"mpi\" if the total task\n" + " count across all -m hosts is > 1, else a\n" + " single-task default)\n" + "example: taskrun -j 12345 -m node01:2 -c \"hostname\" -u alice\n"; + +static const char USAGE_JOIN_SCHED[] = + "join\n" + " (no arguments)\n" + " Joins the cluster as a scheduler.\n"; + +static const char USAGE_GETCONF[] = + "getconf\n" + " (no arguments)\n" + " Prints the full cluster configuration (name, resources, queues,\n" + " workers, user/host groups, configs) known to the scheduler.\n"; + +static const char USAGE_SETPEND[] = + "setpend -j -p [-p ...]\n" + " -j jobid to set the pend reason for (required)\n" + " -p pend reason spec, may repeat: `` sets the main\n" + " reason; `:` adds a host-based reason;\n" + " `::` adds a resource-based reason\n" + "example: setpend -j 12345 -p 5\n"; + static command_t top_commands[] = { - { "help", cmd_help, "Display this text" }, - { "worker", cmd_worker, "In worker context" }, - { "scheduler", cmd_scheduler, "In scheduler context" }, - { "client", cmd_client, "In client context" }, - { "set", cmd_set, "Show/Set settings" }, - { "echo", cmd_echo, "Print a message" }, - { "waldump", cmd_waldump, "Inspect WAL segments" }, - { "walcompare", cmd_walcompare, "Diff member committed WAL prefixes" }, - { "snapverify", cmd_snapverify, "Verify checkpoint file structure" }, - { "snapdiff", cmd_snapdiff, "Byte-diff two checkpoint files" }, - { "configcompare", cmd_configcompare, "Compare active config hash across members" }, - { "quit", cmd_quit, "Quit xce test" }, - { (char*)NULL, (cmd_func_t)NULL, (char*)NULL } + { "help", cmd_help, "Display this text", USAGE_HELP }, + { "worker", cmd_worker, "In worker context", USAGE_WORKER }, + { "scheduler", cmd_scheduler, "In scheduler context", USAGE_SCHEDULER }, + { "client", cmd_client, "In client context", USAGE_CLIENT }, + { "set", cmd_set, "Show/Set settings", USAGE_SET }, + { "echo", cmd_echo, "Print a message", USAGE_ECHO }, + { "waldump", cmd_waldump, "Inspect WAL segments", USAGE_WALDUMP }, + { "walcompare", cmd_walcompare, "Diff member committed WAL prefixes", + USAGE_WALCOMPARE }, + { "snapverify", cmd_snapverify, "Verify checkpoint file structure", + USAGE_SNAPVERIFY }, + { "snapdiff", cmd_snapdiff, "Byte-diff two checkpoint files", + USAGE_SNAPDIFF }, + { "configcompare", cmd_configcompare, "Compare active config hash across members", + USAGE_CONFIGCOMPARE }, + { "quit", cmd_quit, "Quit xce test", USAGE_QUIT }, + { (char*)NULL, (cmd_func_t)NULL, (char*)NULL, (const char*)NULL } }; static command_ctx_t topctx = { .parent = NULL, @@ -211,16 +562,16 @@ static command_ctx_t topctx = { .parent = NULL, .commands = top_commands }; static command_t worker_commands[] = { - { "..", cmd_up, "Goto upper context" }, - { "help", cmd_help, "Display this text" }, - { "quit", cmd_quit, "Quit xce test" }, - { "join", cmd_join_worker, "Join the cluster" }, - { "jstart", cmd_jobstart, "tell controller jobstart" }, - { "jfinish", cmd_jobfinish, "tell controller jobfinish" }, - { "jfail", cmd_jobfail, "tell controller jobfail" }, - { "resupdate", cmd_resupdate, "update res vals" }, - { "echo", cmd_echo, "Print a message" }, - { (char*)NULL, (cmd_func_t)NULL, (char*)NULL } + { "..", cmd_up, "Goto upper context", USAGE_UP }, + { "help", cmd_help, "Display this text", USAGE_HELP }, + { "quit", cmd_quit, "Quit xce test", USAGE_QUIT }, + { "join", cmd_join_worker, "Join the cluster", USAGE_JOIN_WORKER }, + { "jstart", cmd_jobstart, "tell controller jobstart", USAGE_JOBSTART }, + { "jfinish", cmd_jobfinish, "tell controller jobfinish", USAGE_JOBFINISH }, + { "jfail", cmd_jobfail, "tell controller jobfail", USAGE_JOBFAIL }, + { "resupdate", cmd_resupdate, "update res vals", USAGE_RESUPDATE }, + { "echo", cmd_echo, "Print a message", USAGE_ECHO }, + { (char*)NULL, (cmd_func_t)NULL, (char*)NULL, (const char*)NULL } }; static command_ctx_t workerctx = { .parent = &topctx, @@ -229,37 +580,41 @@ static command_ctx_t workerctx = { .parent = &topctx, .commands = worker_commands }; static command_t client_commands[] = { - { "..", cmd_up, "Goto upper context" }, - { "hello", cmd_hello, "Say hello to xce controller" }, - { "reslist", cmd_reslist, "Display cluster resources" }, - { "workerlist", cmd_workerlist, "list workers" }, - { "qctrl", cmd_qctrl, "queue control" }, - { "hostctrl", cmd_hostctrl, "host control" }, + { "..", cmd_up, "Goto upper context", USAGE_UP }, + { "hello", cmd_hello, "Say hello to xce controller", USAGE_HELLO }, + { "reslist", cmd_reslist, "Display cluster resources", USAGE_RESLIST }, + { "workerlist", cmd_workerlist, "list workers", USAGE_WORKERLIST }, + { "qctrl", cmd_qctrl, "queue control", USAGE_QCTRL }, + { "hostctrl", cmd_hostctrl, "host control", USAGE_HOSTCTRL }, { "upgradeconfig", cmd_upgradeconfig, - "submit a config upgrade (cluster.bin file)" }, - { "joblist", cmd_joblist, "list jobs" }, - { "expect-jobs", cmd_expect_jobs, "assert current job count" }, - { "expect-job", cmd_expect_job, "assert a JobId is present" }, + "submit a config upgrade (cluster.bin file)", USAGE_UPGRADECONFIG }, + { "joblist", cmd_joblist, "list jobs", USAGE_JOBLIST }, + { "expect-jobs", cmd_expect_jobs, "assert current job count", USAGE_EXPECT_JOBS }, + { "expect-job", cmd_expect_job, "assert a JobId is present", USAGE_EXPECT_JOB }, { "wait-primary", cmd_wait_primary, - "wait until a primary serves [timeout_ms]" }, - { "ctrlstatus", cmd_ctrlstatus, "show controller/cluster status" }, - { "waldump", cmd_waldump, "Inspect WAL segments" }, - { "walcompare", cmd_walcompare, "Diff member committed WAL prefixes" }, - { "snapverify", cmd_snapverify, "Verify checkpoint file structure" }, - { "snapdiff", cmd_snapdiff, "Byte-diff two checkpoint files" }, - { "configcompare", cmd_configcompare, "Compare active config hash across members" }, - { "shutdown", cmd_shutdown, "shutdown xce controller" }, - { "jobrun", cmd_jobrun, "run a job" }, - { "jobsub", cmd_jobsub, "submit a job" }, - { "jobkill", cmd_jobkill, "signal a job" }, - { "jobckpt", cmd_jobckpt, "checkpoint a job" }, - { "jobinfo", cmd_jobinfo, "job information" }, - { "resval", cmd_resval, "get value of res" }, - { "taskrun", cmd_runtasks, "run tasks" }, - { "help", cmd_help, "Display this text" }, - { "echo", cmd_echo, "Print a message" }, - { "quit", cmd_quit, "Quit xce test" }, - { (char*)NULL, (cmd_func_t)NULL, (char*)NULL } + "wait until a primary serves [timeout_ms]", USAGE_WAIT_PRIMARY }, + { "ctrlstatus", cmd_ctrlstatus, "show controller/cluster status", USAGE_CTRLSTATUS }, + { "waldump", cmd_waldump, "Inspect WAL segments", USAGE_WALDUMP }, + { "walcompare", cmd_walcompare, "Diff member committed WAL prefixes", + USAGE_WALCOMPARE }, + { "snapverify", cmd_snapverify, "Verify checkpoint file structure", + USAGE_SNAPVERIFY }, + { "snapdiff", cmd_snapdiff, "Byte-diff two checkpoint files", + USAGE_SNAPDIFF }, + { "configcompare", cmd_configcompare, "Compare active config hash across members", + USAGE_CONFIGCOMPARE }, + { "shutdown", cmd_shutdown, "shutdown xce controller", USAGE_SHUTDOWN }, + { "jobrun", cmd_jobrun, "run a job", USAGE_JOBRUN }, + { "jobsub", cmd_jobsub, "submit a job", USAGE_JOBSUB }, + { "jobkill", cmd_jobkill, "signal a job", USAGE_JOBKILL }, + { "jobckpt", cmd_jobckpt, "checkpoint a job", USAGE_JOBCKPT }, + { "jobinfo", cmd_jobinfo, "job information", USAGE_JOBINFO }, + { "resval", cmd_resval, "get value of res", USAGE_RESVAL }, + { "taskrun", cmd_runtasks, "run tasks", USAGE_TASKRUN }, + { "help", cmd_help, "Display this text", USAGE_HELP }, + { "echo", cmd_echo, "Print a message", USAGE_ECHO }, + { "quit", cmd_quit, "Quit xce test", USAGE_QUIT }, + { (char*)NULL, (cmd_func_t)NULL, (char*)NULL, (const char*)NULL } }; static command_ctx_t clientctx = { .parent = &topctx, @@ -268,14 +623,14 @@ static command_ctx_t clientctx = { .parent = &topctx, .commands = client_commands }; static command_t scheduler_commands[] = { - { "..", cmd_up, "Goto upper context" }, - { "help", cmd_help, "Display this text" }, - { "quit", cmd_quit, "Quit xce test" }, - { "join", cmd_join_sched, "Join the cluster" }, - { "getconf", cmd_getconf, "Get conf of cluster" }, - { "setpend", cmd_setpend, "set job pend reason" }, - { "echo", cmd_echo, "Print a message" }, - { (char*)NULL, (cmd_func_t)NULL, (char*)NULL } + { "..", cmd_up, "Goto upper context", USAGE_UP }, + { "help", cmd_help, "Display this text", USAGE_HELP }, + { "quit", cmd_quit, "Quit xce test", USAGE_QUIT }, + { "join", cmd_join_sched, "Join the cluster", USAGE_JOIN_SCHED }, + { "getconf", cmd_getconf, "Get conf of cluster", USAGE_GETCONF }, + { "setpend", cmd_setpend, "set job pend reason", USAGE_SETPEND }, + { "echo", cmd_echo, "Print a message", USAGE_ECHO }, + { (char*)NULL, (cmd_func_t)NULL, (char*)NULL, (const char*)NULL } }; static command_ctx_t schedulerctx = { .parent = &topctx, @@ -678,8 +1033,18 @@ int cmd_help(int argc, char* argv[]) for (i = 0; active_ctx->commands[i].name != NULL; i++) { if (strcmp(argv[1], active_ctx->commands[i].name) == 0) { - printf("%s\t\t%s.\n", active_ctx->commands[i].name, - active_ctx->commands[i].doc); + /* `help ` prints the detailed per-command usage (flags, + * positionals, example) -- distinct from the compact one-line + * `doc` the bare `help` listing below still uses unchanged. + * Every table entry is expected to carry a non-NULL `usage` + * now; the `doc` fallback stays only as a defensive guard + * against a future entry that forgets to set it. */ + if (active_ctx->commands[i].usage != NULL) { + fputs(active_ctx->commands[i].usage, stdout); + } else { + printf("%s\t\t%s.\n", active_ctx->commands[i].name, + active_ctx->commands[i].doc); + } return 0; } } @@ -972,7 +1337,10 @@ int cmd_hello(int argc, char* argv[]) msg = optarg; break; default: - fprintf(stderr, " hello [-c count] [-w wait_usec] [-m message]\n"); + /* -w was never implemented (getopt string above is "c:m:" only) + * -- this hint previously advertised a nonexistent -w wait_usec + * flag; corrected to match the actual accepted options. */ + fprintf(stderr, " hello [-c count] [-m message]\n"); return -1; break; } @@ -1071,20 +1439,26 @@ int cmd_workerlist(int argc, char* argv[]) "MaxSwap", "MaxTmp", "NDisks"); for (i = 0; i < rc; i++) { - /* MaxMem is stored/wired as raw bytes (xce_resource_t.unit="bytes" - * for the maxmem resource definition) -- humanize for display only, - * every other column stays as-is. */ + /* MaxMem/MaxSwap/MaxTmp are stored/wired as raw bytes (see + * is_bytes_valued_resource) -- humanize for display only, every + * other column stays as-is. */ char maxmem_buf[64]; + char maxswap_buf[64]; + char maxtmp_buf[64]; format_bytes_human(workers[i]->attrs->maxmem, maxmem_buf, sizeof(maxmem_buf)); + format_bytes_human(workers[i]->attrs->maxswap, maxswap_buf, + sizeof(maxswap_buf)); + format_bytes_human(workers[i]->attrs->maxtmp, maxtmp_buf, + sizeof(maxtmp_buf)); - printf("%-20s%-8s%-25s%-25s%12f%12d%12s%12f%12f%12d\n", + printf("%-20s%-8s%-25s%-25s%12f%12d%12s%12s%12s%12d\n", workers[i]->name, worker_status_str(workers[i]->status), workers[i]->attrs->hosttype, workers[i]->attrs->hostmodel, workers[i]->attrs->cpufactor, workers[i]->attrs->maxcpus, - maxmem_buf, workers[i]->attrs->maxswap, - workers[i]->attrs->maxtmp, workers[i]->attrs->ndisks); + maxmem_buf, maxswap_buf, maxtmp_buf, + workers[i]->attrs->ndisks); } free_Worker_list(rc, workers); @@ -2038,15 +2412,19 @@ cleanup: /* * Resource names whose canonical stored/wire value is raw bytes (not a * generic schema-driven unit system -- deliberately kept as a small, - * explicit, client-side list rather than a wire field, since exactly one - * resource needs this today and a wire-level unit hint would have grown - * every Resource/ResourceVal message regardless of whether it's used). - * CLI display picks the best-fit human unit adaptively (format_bytes_human); - * the stored value is never converted. + * explicit, client-side list rather than a wire field, since only a + * handful of resources need this and a wire-level unit hint would have + * grown every Resource/ResourceVal message regardless of whether it's + * used). CLI display picks the best-fit human unit adaptively + * (format_bytes_human); the stored value is never converted. Covers every + * memory/disk-scale built-in host attribute and load index (maxmem, mem, + * tmp, maxswap, maxtmp) -- deliberately NOT maxcpus/ndisks (plain counts) + * or cpuf/r1m/r15s (unitless factors/averages). */ static int is_bytes_valued_resource(const char* name) { - static const char* const BYTES_VALUED_RESOURCES[] = { "maxmem" }; + static const char* const BYTES_VALUED_RESOURCES[] + = { "maxmem", "mem", "tmp", "maxswap", "maxtmp" }; const int nres = (int)(sizeof(BYTES_VALUED_RESOURCES) / sizeof(BYTES_VALUED_RESOURCES[0])); int i; diff --git a/src/controller/testsuite/test_apply_committed_entry.c b/src/controller/testsuite/test_apply_committed_entry.c index 853be5fb..6970340c 100644 --- a/src/controller/testsuite/test_apply_committed_entry.c +++ b/src/controller/testsuite/test_apply_committed_entry.c @@ -480,7 +480,10 @@ static void test_ace_worker_ctrl_join_restores_and_notifies(void) int rc = apply_committed_entry(&g_tace_p, WAL_WORKER_CTRL, 40, pkt, (uint32_t)total); - TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc, + "bug-1427: a genuine new-worker JOIN must return 1 (not the old " + "hardcoded 0), mirroring WAL_JOB_SUBMIT's apply_rc==1 creation " + "signal — replay_wal()'s new-worker tally relies on this"); worker_t* w = NULL; lock(g_tace_c.workermgr, RTTI_LOCK_READ); @@ -692,6 +695,102 @@ static void test_ace_worker_ctrl_join_duplicate_existing_is_noop(void) printf("test_ace_worker_ctrl_join_duplicate_existing_is_noop: PASS\n"); } +/* bug-1472-remainder: a HOOK_DETAIL_WORKER_DEAD entry against an already- + * existing worker previously returned -EEXIST unconditionally (a no-op, + * same as JOIN's genuine idempotency case) -- so a dead-worker transition + * detected by one member (mark_worker_unavail(), worker.c) never replicated + * to standbys/this member's own WAL replay. Proves the fix replays + * mark_worker_unavail()'s own local mutations: XCE_WORKER_UNAVAIL set, + * dead_since taken from the decoded payload (the PRIMARY's own death + * timestamp), and hostattrs reset to the "unknown"/0 sentinels -- mirroring + * exactly what the primary itself did before firing this same WAL entry. */ +static void test_ace_worker_ctrl_dead_applies_to_existing_worker(void) +{ + tace_setup(); + + /* Pre-existing worker with live (non-"unknown") attrs, as if genuinely + * joined and reporting real hardware facts. */ + hostattrs_t attrs; + memset(&attrs, 0, sizeof(attrs)); + attrs.hosttype = (char*)"LINUX"; + attrs.hostmodel = (char*)"x86_64"; + attrs.cpufactor = 1.0f; + attrs.maxcpus = 8; + attrs.maxmem = 16384; + xce_worker_t wdef; + memset(&wdef, 0, sizeof(wdef)); + wdef.name = (char*)"w-dead"; + wdef.attrs = &attrs; + wdef.status = 0; + TEST_ASSERT_NOT_NULL(makeWorker(g_tace_c.workermgr, &wdef)); + + /* Encode a DEAD entry the way mark_worker_unavail()'s own hook fire + * would -- attrs already reset to "unknown"/0 (capnp_encode_worker + * captures whatever the primary's worker_t held AT hook-fire time, + * which is always post-reset for a real DEAD transition), a specific + * non-zero deadSince (the primary's own death timestamp, not this + * member's clock). */ + hostattrs_t dead_attrs; + memset(&dead_attrs, 0, sizeof(dead_attrs)); + dead_attrs.hosttype = (char*)"unknown"; + dead_attrs.hostmodel = (char*)"unknown"; + xce_worker_t dead_wdef; + memset(&dead_wdef, 0, sizeof(dead_wdef)); + dead_wdef.name = (char*)"w-dead"; + dead_wdef.attrs = &dead_attrs; + dead_wdef.status = XCE_WORKER_UNAVAIL; + dead_wdef.dead_since = 1755300000; /* arbitrary fixed epoch time */ + + uint8_t* wbuf = NULL; + size_t wsize = 0; + TEST_ASSERT_EQUAL_INT(0, tace_encode_worker(&dead_wdef, &wbuf, &wsize)); + + uint32_t disc = (uint32_t)HOOK_DETAIL_WORKER_DEAD; + size_t total = sizeof(disc) + wsize; + uint8_t* pkt = malloc(total); + TEST_ASSERT_NOT_NULL(pkt); + memcpy(pkt, &disc, sizeof(disc)); + memcpy(pkt + sizeof(disc), wbuf, wsize); + free(wbuf); + + xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, tace_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_tace_p, WAL_WORKER_CTRL, 44, pkt, + (uint32_t)total); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "bug-1472-remainder: a DEAD entry against an already-existing " + "worker must succeed (0), not -EEXIST -- -EEXIST silently " + "discarded the dead-worker transition entirely"); + + worker_t* w = NULL; + lock(g_tace_c.workermgr, RTTI_LOCK_READ); + key_find(g_tace_c.workermgr, (void*)"w-dead", (void**)&w); + unlock(g_tace_c.workermgr); + TEST_ASSERT_NOT_NULL(w); + TEST_ASSERT_TRUE_MESSAGE( + (w->host.status & XCE_WORKER_UNAVAIL) != 0, + "the existing worker must be marked XCE_WORKER_UNAVAIL"); + TEST_ASSERT_EQUAL_INT64_MESSAGE((int64_t)1755300000, (int64_t)w->host.dead_since, + "dead_since must come from the decoded payload (the primary's own " + "death timestamp), not be silently discarded or re-derived locally"); + TEST_ASSERT_EQUAL_STRING_MESSAGE("unknown", w->host.attrs->hosttype, + "hostattrs must be reset to the \"unknown\" sentinel, mirroring " + "mark_worker_unavail()'s own local effect on the primary"); + TEST_ASSERT_EQUAL_STRING_MESSAGE("unknown", w->host.attrs->hostmodel, + "hostattrs must be reset to the \"unknown\" sentinel"); + TEST_ASSERT_EQUAL_FLOAT_MESSAGE(0, w->host.attrs->maxmem, + "numeric hostattrs (maxmem and friends) must be zeroed, not left " + "at their last-known-live values"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires, + "a genuinely-applied DEAD transition must fire exactly one " + "notification"); + + free(pkt); + tace_teardown(); + printf("test_ace_worker_ctrl_dead_applies_to_existing_worker: PASS\n"); +} + /* ── WAL_JOB_ARRAY_SUBMIT (job-array-single-wal-record task 3.2) ────────── */ static job_submit_data_t tace_array_make_spec(void) @@ -947,6 +1046,7 @@ void test_apply_committed_entry(void) RUN_TEST(test_ace_worker_ctrl_update_applies_to_existing_worker); RUN_TEST(test_ace_worker_ctrl_update_replaces_not_ors_admin_bits); RUN_TEST(test_ace_worker_ctrl_join_duplicate_existing_is_noop); + RUN_TEST(test_ace_worker_ctrl_dead_applies_to_existing_worker); RUN_TEST(test_ace_job_array_submit_constructs_all_and_notifies_once); RUN_TEST(test_ace_job_array_submit_idempotent_replay_no_extra_notify); #ifdef XCE_TESTING diff --git a/src/controller/testsuite/test_apply_correctness_verification.c b/src/controller/testsuite/test_apply_correctness_verification.c index 6b9cdfd0..7eef76fc 100644 --- a/src/controller/testsuite/test_apply_correctness_verification.c +++ b/src/controller/testsuite/test_apply_correctness_verification.c @@ -49,6 +49,7 @@ #include "jobj.h" #include "persist.h" #include "wal_types.h" +#include "wobj.h" #include "xce/cluster_def.h" #include "xce/types.h" #include "xstring.h" @@ -436,12 +437,16 @@ static void test_checkpoint_tail_with_no_snapshot_is_apply_clean(void) * which needs none of Fixture A's full cluster machinery either. ──────── */ static var g_tacv_bare_jobmgr; +static var g_tacv_bare_workermgr; /* bug-1427 */ static void tacv_bare_setup(void) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 16, .delta = 16 }; + kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING, /* bug-1427 */ + .nslots = 16, + .delta = 16 }; if (makeString((char*)"tacv-bare-probe") == NULL) xstring_table_init(64); @@ -453,6 +458,8 @@ static void tacv_bare_setup(void) g_tacv_bare_jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(g_tacv_bare_jobmgr); + g_tacv_bare_workermgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_tacv_bare_workermgr); } static void tacv_bare_teardown(void) @@ -461,6 +468,28 @@ static void tacv_bare_teardown(void) dereference(g_tacv_bare_jobmgr); g_tacv_bare_jobmgr = NULL; } + if (g_tacv_bare_workermgr != NULL) { + dereference(g_tacv_bare_workermgr); + g_tacv_bare_workermgr = NULL; + } +} + +/* bug-1427: worker analog of tacv_bare_make_job below. */ +static worker_t* tacv_bare_make_worker(const char* name) +{ + xce_worker_t wdef; + hostattrs_t* attrs = calloc(1, sizeof(*attrs)); + TEST_ASSERT_NOT_NULL(attrs); + attrs->hosttype = makeString((char*)"LINUX"); + attrs->hostmodel = makeString((char*)"x86"); + attrs->cpufactor = 1.0f; + attrs->maxcpus = 4; + + memset(&wdef, 0, sizeof(wdef)); + wdef.name = (char*)name; + wdef.attrs = attrs; + + return makeWorker(g_tacv_bare_workermgr, &wdef); } static job_t* tacv_bare_make_job(uint64_t jobid) @@ -513,7 +542,7 @@ static void test_verify_apply_correctness_detects_job_count_mismatch(void) .expected_new_job_count = 2, }; TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN, - persist_verify_apply_correctness(&p, &exp_match, jobmgr_count_before), + persist_verify_apply_correctness(&p, &exp_match, jobmgr_count_before, 0), "matching expectation must report APPLY_CLEAN"); /* Mismatched expectation: log says 3 jobs should exist, only 2 do. */ @@ -523,7 +552,7 @@ static void test_verify_apply_correctness_detects_job_count_mismatch(void) }; TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_DIVERGENT, persist_verify_apply_correctness(&p, &exp_count_mismatch, - jobmgr_count_before), + jobmgr_count_before, 0), "a job-count mismatch must be detected as APPLY_DIVERGENT"); /* Mismatched expectation: position disagrees, count matches. */ @@ -533,7 +562,7 @@ static void test_verify_apply_correctness_detects_job_count_mismatch(void) }; TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_DIVERGENT, persist_verify_apply_correctness(&p, &exp_position_mismatch, - jobmgr_count_before), + jobmgr_count_before, 0), "an apply-cursor position mismatch must be detected as " "APPLY_DIVERGENT even when the job count matches"); @@ -541,6 +570,69 @@ static void test_verify_apply_correctness_detects_job_count_mismatch(void) printf("test_verify_apply_correctness_detects_job_count_mismatch: PASS\n"); } +/* + * bug-1427: worker analog of the job-count test above. Before this fix, + * persist_verify_apply_correctness() had no worker-count check at all — a + * worker-creation divergence between the log-derived expectation and the + * live workermgr table went completely undetected. Also proves the two + * checks are independent: a job-count-only expectation still passes even + * with workers present (worker check only engages when workermgr is set + * AND expected_new_worker_count disagrees). + */ +static void test_verify_apply_correctness_detects_worker_count_mismatch(void) +{ + tacv_bare_setup(); + + cluster_t c; + memset(&c, 0, sizeof(c)); + c.jobmgr = g_tacv_bare_jobmgr; + c.workermgr = g_tacv_bare_workermgr; + + persist_t p; + memset(&p, 0, sizeof(p)); + p.cluster = &c; + + uint64_t jobmgr_count_before = 0; + uint64_t workermgr_count_before = 0; /* empty table before "replay" */ + + /* Two workers materialize — mimics what a real replay would have done + * via apply_committed_entry()'s WAL_WORKER_CTRL/HOOK_DETAIL_WORKER_JOIN + * route (worker_from_capnp() returning 1 for each genuine creation). */ + TEST_ASSERT_NOT_NULL(tacv_bare_make_worker("wkr-a")); + TEST_ASSERT_NOT_NULL(tacv_bare_make_worker("wkr-b")); + + atomic_store_explicit(&p.last_applied_lsn, 42, memory_order_relaxed); + + /* Matching expectation: 0 new jobs, 2 new workers, position 42 — CLEAN. */ + apply_expectation_t exp_match = { + .expected_last_applied_lsn = 42, + .expected_new_job_count = 0, + .expected_new_worker_count = 2, + }; + TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN, + persist_verify_apply_correctness(&p, &exp_match, jobmgr_count_before, + workermgr_count_before), + "matching expectation must report APPLY_CLEAN"); + + /* Mismatched expectation: log says 3 workers should exist, only 2 do — + * the exact bug-1427 gap: a worker-creation divergence must now be + * caught, not silently invisible to this check. */ + apply_expectation_t exp_count_mismatch = { + .expected_last_applied_lsn = 42, + .expected_new_job_count = 0, + .expected_new_worker_count = 3, + }; + TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_DIVERGENT, + persist_verify_apply_correctness(&p, &exp_count_mismatch, + jobmgr_count_before, + workermgr_count_before), + "a worker-count mismatch must be detected as APPLY_DIVERGENT " + "(bug-1427)"); + + tacv_bare_teardown(); + printf("test_verify_apply_correctness_detects_worker_count_mismatch: PASS\n"); +} + /* * Defensive: no live cluster to query (a bare unit-test persist_t, or any * caller with nothing to dispatch into) must degrade to APPLY_CLEAN rather @@ -560,7 +652,7 @@ static void test_verify_apply_correctness_clean_when_no_cluster(void) .expected_new_job_count = 5, /* irrelevant -- no cluster to check */ }; TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN, - persist_verify_apply_correctness(&p, &exp, 0), + persist_verify_apply_correctness(&p, &exp, 0, 0), "no live cluster to query must degrade to APPLY_CLEAN, not a false " "DIVERGENT"); @@ -1500,6 +1592,7 @@ void test_apply_correctness_verification(void) RUN_TEST(test_clean_replay_produces_apply_clean_alongside_crc_verdict); RUN_TEST(test_checkpoint_tail_with_no_snapshot_is_apply_clean); RUN_TEST(test_verify_apply_correctness_detects_job_count_mismatch); + RUN_TEST(test_verify_apply_correctness_detects_worker_count_mismatch); RUN_TEST(test_verify_apply_correctness_clean_when_no_cluster); RUN_TEST(test_apply_divergent_decision_forces_unconditional_fence); RUN_TEST(test_apply_divergent_loud_fence_sets_sticky_flag_and_stays_alive); diff --git a/src/controller/testsuite/test_apply_pump.c b/src/controller/testsuite/test_apply_pump.c index db910b67..c8b3c57a 100644 --- a/src/controller/testsuite/test_apply_pump.c +++ b/src/controller/testsuite/test_apply_pump.c @@ -68,6 +68,7 @@ #include #include #include +#include #include #include #include @@ -1406,6 +1407,116 @@ static void test_apply_pump_late_pending_apply_registration_self_heals(void) "PASS\n"); } +/* bug-1428: a narrower TOCTOU residual of the SAME self-heal mechanism the + * test above proves. There, the register call arrives after the walk has + * ALREADY finished applying `lsn` -- the well-known bug-1415 case, closed + * by the fast-path self-heal check. This test targets the tighter, + * previously-unclosed window: the register call's check runs BEFORE the + * walk applies `lsn` (sees "not yet applied") but the walk's own + * advance-then-take sequence completes entirely BEFORE the register call + * reaches its insert -- so the walk's persist_take_pending_apply() finds + * nothing (register hasn't inserted yet), and the register call then + * inserts a node the walk will never revisit. Permanently orphaned, + * identical failure mode to bug-1415, different trigger window. + * + * Reproduced deterministically (no real-time race dependent on timing) + * by driving the exact bad ordering through controlled state, mirroring + * how the register thread's old code observed state before its lock and + * how the walk observed/mutated it independently: + * 1. Main thread holds pending_applies_lock (standing in for "the walk's + * advance-then-take sequence is in flight, about to complete"). + * 2. A second thread calls the real persist_register_pending_apply() for + * an lsn NOT YET applied (last_applied_lsn is still below it) -- under + * the pre-fix code this thread's unlocked check would already have + * run and concluded "not yet applied" by this point; under the fixed + * code it blocks entirely on the lock, check included. + * 3. Main thread (still holding the lock) advances last_applied_lsn past + * lsn -- simulating the walk finishing its apply for this lsn while + * the register call is blocked/in-flight. + * 4. Main thread releases the lock. + * 5. The second thread's persist_register_pending_apply() call proceeds. + * + * Post-fix: the register call's self-heal check runs AFTER acquiring the + * lock in step 5, observes the now-current last_applied_lsn from step 3, + * and fires on_applied directly -- no insert, no orphan. Pre-fix, the + * check would have already run (unlocked) back in step 2 against the + * STALE last_applied_lsn, concluded "not yet applied", and inserted + * unconditionally once the lock freed in step 5 -- orphaning it forever + * (on_applied never fires; the node stays linked on pending_applies). + */ +typedef struct { + persist_t* p; + uint64_t lsn; + int fired; + uint64_t fired_lsn; +} trpa_ctx_t; + +static void trpa_on_applied(void* ctx, uint64_t lsn) +{ + trpa_ctx_t* c = (trpa_ctx_t*)ctx; + c->fired = 1; + c->fired_lsn = lsn; +} + +static void* trpa_register_thread(void* arg) +{ + trpa_ctx_t* c = (trpa_ctx_t*)arg; + persist_register_pending_apply(c->p, c->lsn, trpa_on_applied, c); + return NULL; +} + +static void test_register_pending_apply_race_against_concurrent_apply_self_heals(void) +{ + persist_t p; + memset(&p, 0, sizeof(p)); + TEST_ASSERT_EQUAL_INT(0, pthread_mutex_init(&p.pending_applies_lock, NULL)); + atomic_store_explicit(&p.last_applied_lsn, 40, memory_order_relaxed); + + const uint64_t target_lsn = 42; /* not yet applied at the moment the + * register thread's call begins */ + + trpa_ctx_t rctx = { .p = &p, .lsn = target_lsn, .fired = 0, .fired_lsn = 0 }; + + pthread_mutex_lock(&p.pending_applies_lock); + + pthread_t th; + TEST_ASSERT_EQUAL_INT(0, + pthread_create(&th, NULL, trpa_register_thread, &rctx)); + + /* Give the register thread every opportunity to run past an unlocked + * check (pre-fix) before the walk's simulated advance below — a bounded + * sleep is fine here since the correctness assertion is deterministic + * regardless of exactly how far it got; this only maximizes pre-fix + * failure detection sensitivity. */ + usleep(20000); + + /* Simulate the walk finishing its apply of target_lsn while the + * register call is blocked on (or, pre-fix, already past) the check. */ + atomic_store_explicit(&p.last_applied_lsn, target_lsn, memory_order_relaxed); + + pthread_mutex_unlock(&p.pending_applies_lock); + + struct timespec ts = { .tv_sec = 0, .tv_nsec = 200000000L }; + nanosleep(&ts, NULL); + TEST_ASSERT_EQUAL_INT(0, pthread_join(th, NULL)); + + int has_pending = persist_has_pending_apply(&p, target_lsn); + + pthread_mutex_destroy(&p.pending_applies_lock); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, rctx.fired, + "bug-1428: a registration racing a concurrent apply of the same lsn " + "must still fire its hand-off (self-heal) exactly once -- 0 means " + "the TOCTOU orphaned it permanently"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(target_lsn, rctx.fired_lsn, + "the hand-off must fire for the correct lsn"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, has_pending, + "no orphaned registration may remain linked on pending_applies"); + + printf("test_register_pending_apply_race_against_concurrent_apply_self_heals: " + "PASS\n"); +} + /* ── 6. Standby-ingest commit-then-apply (2026-07-10 fix) ──────────────── * * Regression coverage for ha.c's standby WAL-replication ingest path. @@ -1648,6 +1759,7 @@ void test_apply_pump(void) RUN_TEST(test_apply_pump_guard_does_not_break_job_ctrl_live_append); RUN_TEST(test_apply_pump_fail_racing_finish_merges_against_current_state); RUN_TEST(test_apply_pump_late_pending_apply_registration_self_heals); + RUN_TEST(test_register_pending_apply_race_against_concurrent_apply_self_heals); RUN_TEST(test_standby_apply_deferred_until_committed_then_materializes); RUN_TEST(test_standby_truncate_before_commit_leaves_no_phantom_job); } diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index 478f49b2..c4916452 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -2136,17 +2136,22 @@ static void test_clr_preason_receipt_reply_fact_retained(void) /* ── do_update_op rusage path fixture ────────────────────────────────── */ -/* KNOWN PRE-EXISTING MISMATCH (discovered while pinning, reported in the - * task evidence — NOT fixed here, production is out of boundary): - * update_job_rusages fires the AFTER_EVENT HOOK_DETAIL_JOB_RUSAGE_UPDATE - * hook with (cluster_t*, xce_job_rusages_t*) [job.c:1015-1016], but - * persist_job_hook_fun's default arm reads the second vararg as a job_t* - * [persist.c:1354-1359] and dereferences j->job.state at offset 56 — - * exactly sizeof(xce_job_rusages_t) — an out-of-bounds read on an - * exactly-sized allocation (the WAL_JOB_RUSAGE payload's state field is - * garbage). The padded box below keeps THIS pin's read in-bounds under - * ASAN without masking the report; the jobid field (offset 0 in both - * types) lines up by accident. */ +/* bug-451, FIXED: update_job_rusages fires the AFTER_EVENT + * HOOK_DETAIL_JOB_RUSAGE_UPDATE hook with (cluster_t*, xce_job_rusages_t*) + * — a different va_list shape than every other job hook detail — because + * sched_job_rusage_update_fun/apiserver_job_rusage_update_fun (this hook's + * other two listeners) genuinely need the full rusages payload + * (pids/pgids/usages), not just a job_t*. persist_job_hook_fun previously + * had no dedicated arm for this detail and fell through to the default + * arm, which misread the xce_job_rusages_t* as a job_t* and dereferenced + * j->job.state at offset 56 — exactly sizeof(xce_job_rusages_t) — an + * out-of-bounds read on an exactly-sized allocation. Now fixed with a + * dedicated arm that reads the real xce_job_rusages_t* and looks the job + * up by jobid via the cluster_t* the hook already carries. The padded box + * below predates the fix (kept so this box's own read stays in-bounds + * under ASAN regardless); the jobid field (offset 0 in both types) lined + * up by accident even under the old bug — see the new assertion below for + * proof the state field (which did NOT line up) is now correct too. */ static union { xce_job_rusages_t r; unsigned char pad[sizeof(xce_job_rusages_t) + 64]; @@ -2198,12 +2203,35 @@ static void cbrp_arrange_update_rusage(uint64_t jobid) * WAL_JOB_UPDATE, HOOK_DETAIL_JOB_RUSAGE_UPDATE → WAL_JOB_RUSAGE, both * RETAIN per wal_classify) stay pending until ACK-covered. Umbrella * Reply-Ordering Contract: receipt-only ACK for fact-shaped delivery. */ +/* Forward-declared: defined further below, alongside its _ex sibling and + * every other caller (this test is the only one that needs it this early + * in the file). */ +static int cbrp_count_jobstate_entries(persist_t *p, uint32_t wal_type, + uint64_t jobid, uint32_t *state_out); + static void test_update_op_rusage_receipt_reply_facts_retained(void) { + uint32_t rusage_state = 0xdeadbeef; + int rusage_count; + cbrp_setup(3000); - cbrp_arrange_jobctrl(72, 1); /* job 72 exists; jobctrl req unused */ + cbrp_arrange_jobctrl(72, 1); /* job 72 exists, state = JOB_STAT_PEND */ cbrp_arrange_update_rusage(72); cbrp_drive_receipt_fact(do_update_op, &g_update_req, 2); + + /* bug-451 regression pin: the WAL_JOB_RUSAGE fact's jobid AND state + * must both be correct — under the pre-fix bug, jobid happened to + * line up by accident (offset 0 in both misread types) but state + * (offset 56, past xce_job_rusages_t's end) read back as 0 from the + * zeroed g_rusages_box padding, not the job's real JOB_STAT_PEND. */ + rusage_count = cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_RUSAGE, + 72, &rusage_state); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, rusage_count, + "WAL_JOB_RUSAGE fact for job 72 must be found by jobid"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE((uint32_t)JOB_STAT_PEND, rusage_state, + "bug-451: WAL_JOB_RUSAGE's state must be the job's real state, not " + "a type-confused OOB read past xce_job_rusages_t's end"); + g_stub_ep_worker = NULL; cbrp_teardown(); printf("test_update_op_rusage_receipt_reply_facts_retained: PASS\n"); diff --git a/src/controller/testsuite/test_e2e_checkpoint.c b/src/controller/testsuite/test_e2e_checkpoint.c index 692efb5f..735dd655 100644 --- a/src/controller/testsuite/test_e2e_checkpoint.c +++ b/src/controller/testsuite/test_e2e_checkpoint.c @@ -158,14 +158,23 @@ static void test_checkpoint_wal_trim_after_checkpoint(void) persist_t* p = make_ckpt_e2e(tmpdir, sizeof(tmpdir), 2); TEST_ASSERT_NOT_NULL(p); - /* Write enough entries and checkpoint once */ + /* Write enough entries and checkpoint once. No live apply pump runs in + * this cluster-less fixture, so last_applied_lsn stays 0 unless + * simulated -- bug-1426's trim floor now also caps at last_applied_lsn + * (the local apply cursor), so it must be advanced here to reflect the + * pump having caught up, or trim_wal_segments would correctly refuse + * to delete anything this test expects trimmed. */ for (int i = 0; i < 5; i++) persist_wal_async(p, WAL_JOB_UPDATE, "trim", 4); + atomic_store_explicit(&p->last_applied_lsn, p->current_lsn - 1, + memory_order_relaxed); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); /* Write a few more entries then checkpoint again */ for (int i = 0; i < 3; i++) persist_wal_async(p, WAL_JOB_UPDATE, "trim2", 5); + atomic_store_explicit(&p->last_applied_lsn, p->current_lsn - 1, + memory_order_relaxed); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); /* After two checkpoints the wal directory must still be functional @@ -259,6 +268,14 @@ static void test_trim_guard_preserves_standby_segments(void) TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "y", 1)); TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p)); + /* bug-1426: simulate the local apply pump having caught up to the + * active segment tip -- trim_wal_segments now also caps its floor at + * last_applied_lsn (this test targets the standby-floor guard + * specifically, not the apply-cursor one, so the pump must not be the + * limiting factor here). */ + atomic_store_explicit(&p->last_applied_lsn, p->current_lsn - 1, + memory_order_relaxed); + /* Trigger checkpoint → trim_wal_segments runs */ TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); @@ -307,6 +324,10 @@ static void test_trim_without_standbys_proceeds_normally(void) TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p)); TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "b", 1)); TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p)); + /* bug-1426: simulate the local apply pump having caught up (see + * test_trim_guard_preserves_standby_segments's identical note above). */ + atomic_store_explicit(&p->last_applied_lsn, p->current_lsn - 1, + memory_order_relaxed); TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); char wal_path[512]; @@ -331,6 +352,71 @@ static void test_trim_without_standbys_proceeds_normally(void) printf("test_trim_without_standbys_proceeds_normally: PASS\n"); } +/* ------------------------------------------------------------------ */ +/* Test 6: local apply-cursor trim guard — bug-1426. With no standbys */ +/* registered (so the standby-floor guard alone would allow trimming */ +/* everything below the active segment), a LAGGING local apply pump */ +/* (last_applied_lsn below the active segment) must still block */ +/* trimming of segments it hasn't caught up to yet. */ +/* */ +/* RED proof: revert the last_applied_lsn floor cap in */ +/* trim_wal_segments -> seg-at-2 (which the lagging pump hasn't read */ +/* yet) is deleted anyway, because the standby-only floor (= active_seg */ +/* _lsn=3, no standbys) never considers local apply progress -> a */ +/* concurrent live apply-pump reader relying on an already-open cached */ +/* fd into that segment (POSIX unlink-keeps-fd-alive) would silently */ +/* keep succeeding, while any FRESH read of the same range (a restart's */ +/* persist_recover() replay) would find it already gone -- the apply- */ +/* correctness gap this bug describes. */ +/* ------------------------------------------------------------------ */ + +static void test_trim_guard_preserves_unapplied_local_segments(void) +{ + char tmpdir[256]; + persist_t* p = make_ckpt_e2e(tmpdir, sizeof(tmpdir), 2); + TEST_ASSERT_NOT_NULL(p); + + /* No standbys registered — the pre-bug-1426 floor would be + * active_seg_lsn=3 alone, same as test_trim_without_standbys_proceeds_ + * normally above. This time, deliberately advance last_applied_lsn only + * through the entry in seg-at-1 (LSN 1), NOT the one in seg-at-2 (LSN + * 2) — simulating a live apply pump that is still lagging (e.g. mid + * catch-up burst) when the checkpoint's trim fires. seg-at-1 is fully + * consumed and safe to delete; seg-at-2 is not — the pump hasn't + * reached its entry yet. */ + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1)); + TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p)); + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "b", 1)); + TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(p)); + atomic_store_explicit(&p->last_applied_lsn, 1, memory_order_relaxed); + + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + + char wal_path[512]; + snprintf(wal_path, sizeof(wal_path), "%s/e2e-ckpt/data/wal", tmpdir); + + /* bug-1426 floor = min(active_seg_lsn=3, last_applied_lsn+1=2) = 2. + * seg_at_1 (start=1) < 2 → deleted: its one entry (LSN 1) is fully + * applied, safe to reclaim. seg_at_2 (start=2) < 2 is FALSE → + * preserved: its entry (LSN 2) has NOT been applied yet — deleting it + * would strand data the local apply pump still needs. */ + TEST_ASSERT_FALSE_MESSAGE( + seg_file_exists(wal_path, 1), + "seg-at-1: fully applied (last_applied_lsn=1), safe to trim even " + "though no standby needs it either"); + TEST_ASSERT_TRUE_MESSAGE( + seg_file_exists(wal_path, 2), + "seg-at-2: NOT yet applied by the lagging local pump " + "(last_applied_lsn=1 < 2), MUST be preserved (bug-1426 guard) even " + "though no standby needs it"); + TEST_ASSERT_TRUE_MESSAGE( + seg_file_exists(wal_path, 3), + "seg-at-3: active segment, MUST always be preserved"); + + teardown_ckpt_e2e(p); + printf("test_trim_guard_preserves_unapplied_local_segments: PASS\n"); +} + /* ------------------------------------------------------------------ */ /* Suite entry point */ /* ------------------------------------------------------------------ */ @@ -345,4 +431,5 @@ void test_e2e_checkpoint(void) RUN_TEST(test_snapshot_rotation_keeps_snap_keep_n); RUN_TEST(test_trim_guard_preserves_standby_segments); RUN_TEST(test_trim_without_standbys_proceeds_normally); + RUN_TEST(test_trim_guard_preserves_unapplied_local_segments); } diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index 68d051d4..c2447d58 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -1359,6 +1359,17 @@ extern int standby_ep_err_handler_primary_for_test(endpoint_t *ep); * evicted while resolving ownership (raft-drive-async Task 2.2). */ extern _Atomic int ha_standby_evicted_count_for_test; +/* vote_ep_ctx_build_for_test (bug-1474): build a promoted-vote-ep test + * double, mirroring test_ha_connection_reuse.c's declaration. */ +extern endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep, + ha_state_t *ha, + ha_elect_ctx_t *ctx, + int peer_idx, + uint64_t my_term, + uint64_t my_tag, + uint64_t my_lsn, + const char *my_name); + /* Allocate + init a heap endpoint wired as a standby channel for peer_idx, the * way ha_connect_prep_peer / a promoted vote-ep would leave it (production * teardown via standby_ep_err_handler_primary frees ep->data + endpoint, so the @@ -1442,6 +1453,65 @@ static void test_reconnect_resolves_dual_channel_same_peer(void) printf("test_reconnect_resolves_dual_channel_same_peer: PASS\n"); } +/* + * bug-1474 (angle-a): the SAME dual-channel resolution as the test above, but + * for the mixed case that test never exercised — the stale entry is a + * PROMOTED VOTE EP (raft_cb_become_leader), not another plain standby ep. + * A promoted vote ep's peer_idx is the RELATIVE votes[] index (self + * excluded); a fresh standby_ep_ctx_t's peer_idx is always the ABSOLUTE + * candidate index. Comparing them directly (the pre-fix code) can never + * match for the SAME physical peer, so the stale promoted-vote channel + * survives alongside the fresh one — two live outbound channels to one + * peer, exactly the hazard on_standby_connected's own doc comment + * describes. own_setup_primary sets my_priority=0 (self is candidate 0); + * peer 2's relative votes[] index is (2 < 0) ? 2 : 2-1 = 1. + * + * RED (before the fix): count == 2 (both channels survive, no eviction). + * GREEN (after the fix): count == 1 (fresh survives, stale evicted). + */ +static void test_reconnect_resolves_dual_channel_promoted_vote_ep(void) +{ + cluster_t c; + myconf_t conf; + own_setup_primary(&c, &conf); + + atomic_store(&ha_standby_evicted_count_for_test, 0); + + /* Stale entry: a promoted vote ep for peer 2, relative index 1. */ + ha_elect_ctx_t *dummy_ectx = calloc(1, sizeof(*dummy_ectx)); + TEST_ASSERT_NOT_NULL(dummy_ectx); + endpoint_t *stale = endpoint_get(); + TEST_ASSERT_NOT_NULL(stale); + endpoint_init(stale, AF_INET, SOCK_STREAM, 0, 0); + stale->iomgr = NULL; + vote_ep_ctx_build_for_test(stale, &c.ha, dummy_ectx, /*peer_idx=*/1, + /*my_term=*/1, /*my_tag=*/0, /*my_lsn=*/0, + "test"); + zarray_store(c.ha.standby_endpoints, stale); + TEST_ASSERT_EQUAL_INT(1, c.ha.standby_endpoints->count); + + /* Fresh reconnect to the SAME physical peer (absolute index 2). */ + endpoint_t *fresh = own_make_standby_ep(&c.ha, &c, 2); + on_standby_connected_for_test(fresh); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.ha.standby_endpoints->count, + "bug-1474: a reconnect to a peer whose ONLY existing channel is a " + "promoted vote ep must still resolve to exactly ONE live channel"); + TEST_ASSERT_EQUAL_PTR_MESSAGE(fresh, zarray_get(c.ha.standby_endpoints, 0), + "the survivor must be the freshly-connected channel"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, + atomic_load(&ha_standby_evicted_count_for_test), + "the stale promoted-vote-ep channel must be evicted exactly once"); + + standby_ep_err_handler_primary_for_test(fresh); + endpoint_fini(fresh); + free(fresh); + free(dummy_ectx); + + fini_ha_primary_with_standby_eps(&c.ha); + printf("test_reconnect_resolves_dual_channel_promoted_vote_ep: PASS\n"); +} + /* * Test 19 (Req 2.6): registering a NEW channel for a DIFFERENT peer does NOT * evict an existing channel for another peer — resolution is strictly @@ -2334,6 +2404,7 @@ void test_ha_liveness(void) RUN_TEST(test_reconnect_offload_dedups_per_peer); /* raft-drive-async Task 2.2 — inbound/outbound ownership (Req 2.6, 4.4) */ RUN_TEST(test_reconnect_resolves_dual_channel_same_peer); + RUN_TEST(test_reconnect_resolves_dual_channel_promoted_vote_ep); RUN_TEST(test_reconnect_preserves_other_peer_channel); /* raft-drive-async Task 2.3 — reconnect-offload INTEGRATION (Req 2.1, 2.2, 2.5) */ RUN_TEST(test_reconnect_offload_sustains_survivor_heartbeats); diff --git a/src/controller/testsuite/test_persist_hooks.c b/src/controller/testsuite/test_persist_hooks.c index dad9bf5d..061ca1ec 100644 --- a/src/controller/testsuite/test_persist_hooks.c +++ b/src/controller/testsuite/test_persist_hooks.c @@ -8,10 +8,15 @@ #include #include +#include #include +#include "cluster.h" #include "config.h" +#include "jobj.h" +#include "kvmgr.h" #include "persist.h" +#include "typemap.h" #include "xce/cluster_def.h" #include "xlist.h" @@ -62,6 +67,41 @@ static worker_t g_stub_worker; static job_t g_stub_job; static queue_t g_stub_queue; +/* bug-451: HOOK_DETAIL_JOB_RUSAGE_UPDATE is the only job hook detail whose + * persist.c arm actually dereferences the cluster_t* it's called with + * (every other case discards it via `(void)va_arg(ap, cluster_t*)`), since + * it must look the job up by jobid via c->jobmgr rather than receiving a + * job_t* directly. Minimal real (non-NULL) cluster+jobmgr+job, built once + * on first use — every other test in this file keeps passing + * (cluster_t*)NULL safely, this is the one exception. */ +static cluster_t g_hook_cluster; +static int g_hook_jobobj_ready; + +static void ensure_hook_cluster_with_job(uint64_t jobid, unsigned int state) +{ + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 64, .delta = 16 }; + xce_job_t jdef; + job_submit_data_t submit; + + if (!g_hook_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(64, 64)); + g_hook_jobobj_ready = 1; + } + if (g_hook_cluster.jobmgr != NULL) + dereference(g_hook_cluster.jobmgr); + memset(&g_hook_cluster, 0, sizeof(g_hook_cluster)); + g_hook_cluster.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(g_hook_cluster.jobmgr); + + memset(&submit, 0, sizeof(submit)); + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.submit = &submit; + TEST_ASSERT_NOT_NULL(makeJob(g_hook_cluster.jobmgr, &jdef)); +} + /* Variadic wrappers so tests can pass typed args through a real va_list */ static int call_worker_hook(int type, int sit, int detail, persist_t* p, ...) { @@ -276,9 +316,16 @@ static void test_job_hook_rusage_writes_wal_entry(void) uint64_t lsn_before = p->current_lsn; + /* bug-451: RUSAGE_UPDATE args are (cluster_t*, xce_job_rusages_t*) — + * NOT (cluster_t*, job_t*) like STOP/RESUME/UPDATE/CKPT/PREASON below. + * A real cluster+jobmgr+job is required (see ensure_hook_cluster_with_ + * job) since persist.c's arm looks the job up by jobid, it doesn't + * receive a job_t* directly. */ + ensure_hook_cluster_with_job(42, JOB_STAT_RUN); + xce_job_rusages_t rusages = { .jobid = 42 }; int rc = call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_RUSAGE_UPDATE, p, - (cluster_t*)NULL, &g_stub_job); + &g_hook_cluster, &rusages); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_GREATER_THAN(lsn_before, p->current_lsn); @@ -286,6 +333,29 @@ static void test_job_hook_rusage_writes_wal_entry(void) printf("test_job_hook_rusage_writes_wal_entry: PASS\n"); } +static void test_job_hook_rusage_unknown_job_is_silently_skipped(void) +{ + persist_t* p = make_hook_test_persist(); + TEST_ASSERT_NOT_NULL(p); + + uint64_t lsn_before = p->current_lsn; + + /* jobid 999 was never inserted into g_hook_cluster's jobmgr — the job + * may have already been reaped by the time this hook fires. Must + * return 0 (no error) and append nothing, never crash on a NULL + * job_t*. */ + ensure_hook_cluster_with_job(42, JOB_STAT_RUN); + xce_job_rusages_t rusages = { .jobid = 999 }; + int rc = call_job_hook(HOOK_AFTER_EVENT, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_RUSAGE_UPDATE, p, + &g_hook_cluster, &rusages); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT64(lsn_before, p->current_lsn); + + cleanup_hook_test_persist(p); + printf("test_job_hook_rusage_unknown_job_is_silently_skipped: PASS\n"); +} + static void test_job_hook_setpreason_writes_wal_entry(void) { persist_t* p = make_hook_test_persist(); @@ -468,6 +538,7 @@ void test_persist_hooks(void) RUN_TEST(test_job_hook_update_writes_wal_entry); RUN_TEST(test_job_hook_ckpt_writes_wal_entry); RUN_TEST(test_job_hook_rusage_writes_wal_entry); + RUN_TEST(test_job_hook_rusage_unknown_job_is_silently_skipped); RUN_TEST(test_job_hook_setpreason_writes_wal_entry); RUN_TEST(test_job_hook_clrpreason_writes_wal_entry); RUN_TEST(test_queue_hook_update_writes_wal_entry); diff --git a/src/controller/worker.c b/src/controller/worker.c index 069c88f2..4ace6314 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -676,20 +676,32 @@ int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val) return 1; } -int mark_worker_unavail(cluster_t* c, worker_t* w) +/* Local mutation body of "mark a worker dead" -- status flag, dead_since + * (only if not already set), and the derived hostattrs/resvals reset. No + * hook side effects and no locking of its own: callers hold w's lock (or, + * for mark_worker_unavail, are about to fire the surrounding hooks) and + * decide where dead_since comes from (wall clock live, or the WAL payload's + * recorded timestamp on replay -- bug-1472-remainder). */ +static void apply_worker_dead_mutation(cluster_t* c, worker_t* w, + time_t dead_since) { - xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT, - HOOK_DETAIL_WORKER_DEAD, c, w); - w->host.status |= XCE_WORKER_UNAVAIL; if (w->host.dead_since == 0) - w->host.dead_since = time(NULL); + w->host.dead_since = dead_since; if (w->host.attrs != NULL) { reset_hostattrs_unknown(w->host.attrs); add_hostattrs_resvals(c, w); set_host_slots(c, w, 0); } +} + +int mark_worker_unavail(cluster_t* c, worker_t* w) +{ + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_BEFORE_EVENT, + HOOK_DETAIL_WORKER_DEAD, c, w); + + apply_worker_dead_mutation(c, w, time(NULL)); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_DEAD, c, w); @@ -855,15 +867,26 @@ int do_hostcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request) * return 0 so the caller (persist.c's WAL_WORKER_CTRL apply case) * falls through to fire the HOOK_DETAIL_WORKER_UPDATE notification * instead of silently discarding the change. - * - HOOK_DETAIL_WORKER_DEAD: left as -EEXIST (unchanged) -- see - * buglog.json bug-1472 for why DEAD's replication needs its own, - * separately-tracked fix: its mutation isn't confined to - * XCE_WORKER_ADMIN_MASK (XCE_WORKER_UNAVAIL + dead_since + hostattrs - * reset), and the live per-host resource-value array - * (xce_worker_t.resvals, a zarray_t) that mark_worker_unavail() - * zeroes via set_host_slots() never round-trips through the Worker - * capnp schema at all (only resvals_to_sched does) -- a naive - * status-replace here would be an incomplete fix. */ + * - HOOK_DETAIL_WORKER_DEAD (bug-1472-remainder): replays + * mark_worker_unavail()'s own local mutations on the existing worker -- + * XCE_WORKER_UNAVAIL, dead_since (from the decoded payload's deadSince + * field, the PRIMARY's own death timestamp, not this member's clock), + * hostattrs reset to "unknown" sentinels, and the derived resvals + * re-computed from that reset state (add_hostattrs_resvals + slots=0 + * via set_host_slots). This does NOT need a new wire field despite + * the original investigation's design-decision framing: the live + * xce_worker_t.resvals zarray_t the earlier analysis worried about + * never needs to travel on the wire at all, because + * reset-to-"unknown"-and-zero is a fixed, data-independent operation + * -- add_hostattrs_resvals/set_host_slots derive their values FROM + * w->host.attrs's (already-reset) fields, not from anything the + * primary computed. capnp_encode_worker (persist_worker_hook_fun) + * already serializes the FULL xce_worker_t.host on every WAL_WORKER_ + * CTRL entry including DEAD, and mark_worker_unavail() calls + * reset_hostattrs_unknown() BEFORE firing the hook that triggers that + * encode -- so the Worker schema's existing `attrs` field was never + * actually the gap; this function's own DEAD branch simply never + * existed to apply anything. */ int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len) { if (!capnp_payload || len <= sizeof(uint32_t)) @@ -906,10 +929,29 @@ int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len) return 0; } + if (hook_detail == HOOK_DETAIL_WORKER_DEAD) { + lock(existing, RTTI_LOCK_WRITE); + apply_worker_dead_mutation(c, existing, host.dead_since); + unlock(existing); + + free_Worker(&host); + return 0; + } + return -EEXIST; } - return restore_worker(c, host); + /* bug-1427: distinguish "genuinely created a new worker" from a + * negative errno, discarding restore_worker()'s raw idx (its own + * return value, w->idx, is NOT a creation flag -- it can legitimately + * be 0 for the first worker, which would collide with the + * update-applied return above). Neither of this function's two + * callers (persist.c) inspects the exact non-negative value, only + * sign and -EEXIST, so this is a safe, backward-compatible + * repurposing -- see apply_committed_entry()'s WAL_WORKER_CTRL case + * and persist_verify_apply_correctness()'s new worker-count check. */ + int rc = restore_worker(c, host); + return (rc < 0) ? rc : 1; } /* worker_should_retain lives in persist.c — diff --git a/src/frontend/Makefile b/src/frontend/Makefile index 32c6913f..7bbb45fb 100644 --- a/src/frontend/Makefile +++ b/src/frontend/Makefile @@ -9,6 +9,21 @@ XCE_BIN ?= /tmp/xce_static SRV_URL ?= http://localhost:8080 GOFLAGS := CGO_ENABLED=0 +# bug-1558: test-backend's CGO_CFLAGS/CGO_LDFLAGS, self-derived via +# pkg-config so `make test-backend` works standalone instead of only inside +# meson's custom_target env (apiserver/meson.build is the source of truth +# this mirrors -- librtti/c-capnproto/openssl/tinylog resolve via +# pkg-config there; math/rt are cc.find_library() results with no +# pkg-config package, added directly as -lm/-lrt to match). +# BUILD_DIR defaults to the tree tools/xce-build.sh produces (profile asan); +# override for a non-ASAN build, e.g. +# make test-backend BUILD_DIR=$(CURDIR)/../build.dbg FRONTEND_CGO_SANITIZE= +BUILD_DIR ?= $(CURDIR)/../xce.build +FRONTEND_CGO_PKGS := librtti c-capnproto openssl tinylog +FRONTEND_CGO_SANITIZE ?= -fsanitize=address -g +FRONTEND_CGO_CFLAGS := $(FRONTEND_CGO_SANITIZE) $(shell pkg-config --cflags $(FRONTEND_CGO_PKGS)) -I$(CURDIR)/../include -I$(BUILD_DIR)/lib +FRONTEND_CGO_LDFLAGS := $(FRONTEND_CGO_SANITIZE) $(shell pkg-config --libs $(FRONTEND_CGO_PKGS)) -lm -lrt -L$(BUILD_DIR)/lib -lxce + .PHONY: build-cli build-server test test-backend test-integration test-all build-cli: @@ -25,12 +40,18 @@ test: @echo "=== Unit tests: xcecli ===" cd xcecli && $(GOFLAGS) go test ./... -count=1 -## Run backend CGO conversion tests. -## Requires CGO_ENABLED=1 and CGO_CFLAGS/CGO_LDFLAGS pointing at the build tree. -## Run from within the nix develop shell after a full meson build. +## Run backend CGO conversion tests. Standalone-runnable: CGO_CFLAGS/ +## CGO_LDFLAGS are self-derived above via pkg-config (bug-1558) -- no manual +## env setup needed. Still requires the nix develop shell (for pkg-config's +## .pc files) and a prior `xce-build.sh lib` (or any component that builds +## libxce.a) so BUILD_DIR/lib/libxce.a and BUILD_DIR/lib's generated headers +## exist. test-backend: @echo "=== Backend CGO tests: apiserver/backend ===" - cd apiserver && CGO_ENABLED=1 go test ./backend/... -count=1 + cd apiserver && CGO_ENABLED=1 \ + CGO_CFLAGS="$(FRONTEND_CGO_CFLAGS)" \ + CGO_LDFLAGS="$(FRONTEND_CGO_LDFLAGS)" \ + go test ./backend/... -count=1 ## Run regression suite against a live server (requires XCE_BIN and SRV_URL) test-integration: build-cli diff --git a/src/include/persist.h b/src/include/persist.h index 103b02b1..38581342 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -151,10 +151,18 @@ typedef enum { * the loaded snapshot's checkpoint_lsn (entries at/below it are skipped * before ever reaching dispatch — see replay_wal), so each one names a * job that must now exist in the live job table that did NOT exist - * before this replay pass began. */ + * before this replay pass began. + * expected_new_worker_count — bug-1427: the worker analog of + * expected_new_job_count. The number of WAL_WORKER_CTRL/ + * HOOK_DETAIL_WORKER_JOIN entries replay dispatched in this pass that + * genuinely created a new worker (worker_from_capnp() returns 1 for + * that case, 0 for an UPDATE/DEAD applied to an already-existing + * worker or -EEXIST for a duplicate JOIN — only the 1 case is tallied, + * mirroring WAL_JOB_SUBMIT's apply_rc == 1 convention). */ typedef struct { uint64_t expected_last_applied_lsn; uint64_t expected_new_job_count; + uint64_t expected_new_worker_count; } apply_expectation_t; /* ============================================================ @@ -915,24 +923,28 @@ bool persist_recovery_complete(persist_t* p); void persist_mark_recovery_pending(persist_t* p); /* persist_verify_apply_correctness (raft-apply-decoupling task 5.1, Req - * 8.1/8.2). Pure comparison — mirrors persist_recover_decision's separation - * of decision from action: no logging, no fence side effects, no mutation of - * `p` beyond the read of last_applied_lsn. Compares `exp` (derived by - * scanning the committed log during replay — see apply_expectation_t) against - * the actual materialized state: p's replay-derived apply cursor, and — when - * p->cluster is non-NULL — the live job table's growth since - * `jobmgr_count_before_replay` (the job count captured just before replay - * began). Returns APPLY_CLEAN on a match. Returns APPLY_DIVERGENT when either - * the apply cursor or the live job count disagrees with `exp`. + * 8.1/8.2; worker half added bug-1427). Pure comparison — mirrors + * persist_recover_decision's separation of decision from action: no + * logging, no fence side effects, no mutation of `p` beyond the read of + * last_applied_lsn. Compares `exp` (derived by scanning the committed log + * during replay — see apply_expectation_t) against the actual materialized + * state: p's replay-derived apply cursor, and — when p->cluster is + * non-NULL — the live job table's growth since `jobmgr_count_before_replay` + * and the live worker table's growth since `workermgr_count_before_replay` + * (both counts captured just before replay began). Returns APPLY_CLEAN on a + * match. Returns APPLY_DIVERGENT when the apply cursor, the live job count, + * or the live worker count disagrees with `exp`. * * Safe to call with p->cluster == NULL (a bare unit-test persist_t, or any - * caller with nothing to dispatch into): the job-count half of the check is - * skipped — only the position check applies — mirroring - * apply_committed_entry()'s own "nothing to dispatch into" convention. Safe - * on NULL p or NULL exp (returns APPLY_CLEAN — nothing to verify). */ + * caller with nothing to dispatch into): the job-count and worker-count + * halves of the check are both skipped — only the position check applies — + * mirroring apply_committed_entry()'s own "nothing to dispatch into" + * convention. Safe on NULL p or NULL exp (returns APPLY_CLEAN — nothing to + * verify). */ apply_verdict_t persist_verify_apply_correctness( persist_t* p, const apply_expectation_t* exp, - uint64_t jobmgr_count_before_replay); + uint64_t jobmgr_count_before_replay, + uint64_t workermgr_count_before_replay); /* Accessor for the sticky verdict computed by the last persist_recover() * (task 5.1). APPLY_CLEAN before recovery has run (fresh store default, diff --git a/src/tools/deploy-ha.sh b/src/tools/deploy-ha.sh index a4637107..cbc3347e 100755 --- a/src/tools/deploy-ha.sh +++ b/src/tools/deploy-ha.sh @@ -21,20 +21,59 @@ BINARY=$(realpath "$BINARY") # xceadmin is meson-built now (src/xceadmin/meson.build), not a plain # `pip install -e .` venv checked out under src/xceadmin/ -- its build # output lives under whichever meson build dir xce-build.sh's profile -# picked, or under a nix-installed prefix's bin/. Try each in turn. +# picked, or under a nix-installed prefix's bin/, and each candidate is a +# STATIC SNAPSHOT (meson-build.sh copies src/xceadmin out-of-tree and does +# a plain, non-editable `pip install`) taken whenever that profile's +# `xceadmin` target was last built -- editing src/xceadmin/ source does +# NOT retroactively update an already-built snapshot. +# +# Correctness, not just freshness, matters here: xceadmin builds the +# cluster.bin that $BINARY (the controller daemon being deployed) will +# decode, so the two must come from the SAME build generation -- e.g. if +# the capnp wire schema changed between builds, an xceadmin from one +# generation and a controller from another can silently mis-encode/ +# mis-decode a field (this is exactly how a stale-xceadmin deploy bug +# went undiagnosed: a fresher-but-DIFFERENT-build xceadmin could just as +# easily mismatch $BINARY as a staler one). Prefer the xceadmin living in +# $BINARY's OWN build dir (/controller/xce_controller -> +# /xceadmin/.venv/bin/xceadmin) -- guaranteed same-generation. +# Only if that specific one doesn't exist (e.g. xceadmin was never built +# in that profile) fall back to whichever other candidate is newest, with +# a loud warning that same-generation consistency isn't guaranteed. +BINARY_BUILD_DIR=$(dirname "$(dirname "$BINARY")") +BINARY_DIR=$(dirname "$BINARY") XCEADMIN="" -for candidate in \ - "$SCRIPT_DIR/../xce.build/xceadmin/.venv/bin/xceadmin" \ - "$SCRIPT_DIR/../build.release/xceadmin/.venv/bin/xceadmin" \ - "$SCRIPT_DIR/../build.dbg/xceadmin/.venv/bin/xceadmin" \ - "$SCRIPT_DIR/../result/bin/xceadmin" \ - "$(command -v xceadmin 2>/dev/null || true)" -do - if [ -n "$candidate" ] && [ -x "$candidate" ]; then - XCEADMIN="$candidate" - break +if [ -x "$BINARY_BUILD_DIR/xceadmin/.venv/bin/xceadmin" ]; then + # meson layout: /controller/xce_controller + /xceadmin/.venv/... + XCEADMIN="$BINARY_BUILD_DIR/xceadmin/.venv/bin/xceadmin" + echo "Using xceadmin: $XCEADMIN (same build dir as $BINARY)" +elif [ -x "$BINARY_DIR/xceadmin" ]; then + # nix result layout: result/bin/xce_controller + result/bin/xceadmin (flat, no + # per-component subdir) -- BINARY_BUILD_DIR's two-levels-up guess doesn't apply. + XCEADMIN="$BINARY_DIR/xceadmin" + echo "Using xceadmin: $XCEADMIN (same directory as $BINARY)" +else + echo "Warning: no xceadmin found alongside $BINARY (checked $BINARY_BUILD_DIR/xceadmin/.venv/bin/xceadmin and $BINARY_DIR/xceadmin) -- falling back to the newest xceadmin found elsewhere, which is NOT guaranteed to be the same build generation as $BINARY." + XCEADMIN_MTIME=0 + for candidate in \ + "$SCRIPT_DIR/../xce.build/xceadmin/.venv/bin/xceadmin" \ + "$SCRIPT_DIR/../build.release/xceadmin/.venv/bin/xceadmin" \ + "$SCRIPT_DIR/../build.dbg/xceadmin/.venv/bin/xceadmin" \ + "$SCRIPT_DIR/../result/bin/xceadmin" \ + "$(command -v xceadmin 2>/dev/null || true)" + do + if [ -n "$candidate" ] && [ -x "$candidate" ]; then + candidate_mtime=$(stat -c %Y "$candidate" 2>/dev/null || stat -f %m "$candidate" 2>/dev/null || echo 0) + if [ "$candidate_mtime" -gt "$XCEADMIN_MTIME" ]; then + XCEADMIN="$candidate" + XCEADMIN_MTIME="$candidate_mtime" + fi + fi + done + if [ -n "$XCEADMIN" ]; then + echo "Using xceadmin: $XCEADMIN (built $(date -d "@$XCEADMIN_MTIME" 2>/dev/null || date -r "$XCEADMIN_MTIME" 2>/dev/null))" fi -done +fi if [ -z "$XCEADMIN" ]; then echo "Error: xceadmin binary not found -- run 'xce-build.sh xceadmin' or 'nix build .#xcecore' first" exit 1 diff --git a/src/tools/xce-build.sh b/src/tools/xce-build.sh index cd964900..899558a0 100755 --- a/src/tools/xce-build.sh +++ b/src/tools/xce-build.sh @@ -85,7 +85,12 @@ case "$COMPONENT" in meson compile -C "$BUILD_DIR" xce_scheduler sched_test if [ "$RUN_TESTS" = 1 ]; then echo "=== Running scheduler tests ===" - "./$BUILD_DIR/scheduler/testsuite/sched_test" + # bug-1559: run the whole 'scheduler' meson suite (meson test + # rebuilds by default), not just the sched_test binary directly + # by path -- that bypassed every other scheduler test target + # registered under the same suite tag (component-failover- + # adoption's 4 task-3.x binaries among them). + meson test -C "$BUILD_DIR" --suite scheduler --print-errorlogs fi ;; worker) diff --git a/src/worker/src/node_information.c b/src/worker/src/node_information.c index 7a69adb2..c3cb7bba 100644 --- a/src/worker/src/node_information.c +++ b/src/worker/src/node_information.c @@ -799,8 +799,12 @@ static int read_memory_info(double* mem_total, double* mem_free, } /** - * @brief The mem index gives the currently available physical memory in MB. - * @return A string representing the available physical memory in megabytes. + * @brief The mem index gives the currently available physical memory in + * bytes (canonical internal/wire unit -- same convention as the + * maxmem host attribute; CLI display converts to a human-readable + * unit, config-time thresholds may be written with a unit suffix + * like "500MB" and are converted to bytes when parsed). + * @return A string representing the available physical memory in bytes. */ char* node_info_get_mem(void) { @@ -813,14 +817,14 @@ char* node_info_get_mem(void) return NULL; } - // Read memory information + // Read memory information (read_memory_info reports MB) if (read_memory_info(&mem_total, &mem_free, &mem_used) != 0) { free(result); return NULL; } - // Format the available memory as a string (in MB) - int ret = snprintf(result, 64, "%.2f", mem_free); + // Format the available memory as a string (in bytes) + int ret = snprintf(result, 64, "%.2f", mem_free * 1024.0 * 1024.0); if (ret < 0 || ret >= 64) { XCECD_LOG(XCE_LOG_ERROR, "Failed to format memory information string"); free(result); @@ -831,9 +835,11 @@ char* node_info_get_mem(void) } /** - * @brief The tmp index is the space available in MB or in units set - * on the file system that contains the temporary directory. - * @return A string representing the available space in temporary directory. + * @brief The tmp index is the space available in bytes (canonical internal/ + * wire unit -- see node_info_get_mem) on the file system that + * contains the temporary directory. + * @return A string representing the available space in the temporary + * directory, in bytes. */ char* node_info_get_tmp(void) { @@ -859,11 +865,8 @@ char* node_info_get_tmp(void) unsigned long long free_bytes = (unsigned long long)fs_stat.f_bavail * fs_stat.f_frsize; - // Convert to MB - double tmp_free_mb = (double)free_bytes / (1024.0 * 1024.0); - - // Format the available tmp space as a string (in MB) - int ret = snprintf(result, 64, "%.2f", tmp_free_mb); + // Format the available tmp space as a string (in bytes) + int ret = snprintf(result, 64, "%llu", free_bytes); if (ret < 0 || ret >= 64) { XCECD_LOG(XCE_LOG_ERROR, "Failed to format tmp space string"); free(result); diff --git a/src/worker/src/xcecd.c b/src/worker/src/xcecd.c index d20cc80a..89796b46 100644 --- a/src/worker/src/xcecd.c +++ b/src/worker/src/xcecd.c @@ -348,8 +348,12 @@ static void init_host_attrs(void) = hardware_monitor_get_total_memory(xcecd_ctx.hardware_monitor); } - host_attrs.maxswap = 1024; - host_attrs.maxtmp = 2048; + /* Placeholder stub values (bug-1554, real swap/tmp-disk detection not + * yet implemented) -- kept as bytes to match every other memory-scale + * host attribute's canonical unit (maxmem, mem, tmp). Magnitude chosen + * to match the pre-existing MB-scale placeholders (1024MB/2048MB). */ + host_attrs.maxswap = 1024.0 * 1024.0 * 1024.0; + host_attrs.maxtmp = 2048.0 * 1024.0 * 1024.0; host_attrs.ndisks = 1; host_attrs.hostmodel = strdup("Intel64");