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 <noreply@anthropic.com>
This commit is contained in:
parent
3afe3d5549
commit
45400c3a28
SSH key fingerprint:
SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
18 changed files with 1499 additions and 195 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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 <cmd>` 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 [<cmd>]\n"
|
||||
" <cmd> 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 [<message...>]\n"
|
||||
" <message...> 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 <wal_dir|data_dir|seg_file> [more...]\n"
|
||||
" <path> 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 <N>] <memberA> <memberB> [more...]\n"
|
||||
" --report also emit a full per-member (lsn,term,pcrc,status)\n"
|
||||
" machine-checkable table, not just the verdict\n"
|
||||
" --committed <N> use LSN N as the shared committed boundary instead\n"
|
||||
" of deriving it as the minimum CRC-intact tip across\n"
|
||||
" members\n"
|
||||
" <memberX> 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 <snap_file> [<snap_file> ...]\n"
|
||||
" <snap_file> 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 <source_file> <received_file>\n"
|
||||
" <source_file> the primary's own locally generated checkpoint\n"
|
||||
" (snap-*.capnp)\n"
|
||||
" <received_file> 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 <member_dir> <member_dir> [<member_dir> ...]\n"
|
||||
" <member_dir> 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 <key> <value>\n"
|
||||
"set -e | +e | continue-on-error\n"
|
||||
" (no args) list all settings in the current context\n"
|
||||
" (current value + default)\n"
|
||||
" <key> <value> set setting <key> (e.g. masters, secret,\n"
|
||||
" myname) to <value>\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 <cpufactor>] [-c <maxcpus>] [-m <maxmem>] [-s <maxswap>]\n"
|
||||
" [-t <maxtmp>] [-d <ndisks>]\n"
|
||||
" -f <cpufactor> relative CPU speed factor (default 1.0)\n"
|
||||
" -c <maxcpus> max CPU count (default 4)\n"
|
||||
" -m <maxmem> max memory in bytes (default 64)\n"
|
||||
" -s <maxswap> max swap in bytes (default 1024)\n"
|
||||
" -t <maxtmp> max tmp space in bytes (default 2048)\n"
|
||||
" -d <ndisks> 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 <jobid> [-p <pid>]\n"
|
||||
" -j <jobid> jobid to report a synthetic job-start fact for\n"
|
||||
" (required)\n"
|
||||
" -p <pid> 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 <jobid> [-c <exitcode>]\n"
|
||||
" -j <jobid> jobid to report a synthetic job-finish fact for\n"
|
||||
" (required)\n"
|
||||
" -c <exitcode> exit code to report (default 0)\n"
|
||||
"example: jfinish -j 12345 -c 0\n";
|
||||
|
||||
static const char USAGE_JOBFAIL[] =
|
||||
"jfail -j <jobid> [-c <code>] [-s <stage>]\n"
|
||||
" -j <jobid> jobid to report a synthetic job-fail fact for\n"
|
||||
" (required)\n"
|
||||
" -c <code> failure code to report (default 0)\n"
|
||||
" -s <stage> 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 <resname> -v <value> [-n <resname> -v <value> ...]\n"
|
||||
" -n <resname> 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> 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 <count>] [-m <message>]\n"
|
||||
" -c <count> number of times to send the hello (default 1)\n"
|
||||
" -m <message> 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 <queue> [-a] [-i] [-o] [-c]\n"
|
||||
" -q <queue> 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 <hostname> (-o|-c)\n"
|
||||
" -h <hostname> 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 <cluster.bin>\n"
|
||||
" <cluster.bin> 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>\n"
|
||||
" <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 <jobid>\n"
|
||||
" <jobid> 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 [<timeout_ms>]\n"
|
||||
" <timeout_ms> 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 <jobid> [-m <host>[:<ntasks>]] [-m ...]\n"
|
||||
" -j <jobid> jobid to dispatch (required, must be > 0)\n"
|
||||
" -m <host>[:<n>] 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 <user> -q <queue> -c <cmdline> [-R <resreq>]\n"
|
||||
" [-n <min>[,<max>]] [-x <exdata>] [-a <arraysize>]\n"
|
||||
" -u <user> submitting user (required)\n"
|
||||
" -q <queue> target queue (required)\n"
|
||||
" -c <cmdline> command line to run (required)\n"
|
||||
" -R <resreq> resource requirement expression (default\n"
|
||||
" \"type == any\")\n"
|
||||
" -n <min>[,<max>] task count, or a min,max range (default 1)\n"
|
||||
" -x <exdata> raw exdata JSON string (overrides the built-in\n"
|
||||
" PATH env)\n"
|
||||
" -a <arraysize> 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 <resname> [-m <host>]\n"
|
||||
" -n <resname> resource name to query (required)\n"
|
||||
" -m <host> 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 <jobid> [-s <signo>]\n"
|
||||
" -j <jobid> jobid to signal (required)\n"
|
||||
" -s <signo> signal number to send (default SIGTERM)\n"
|
||||
"example: jobkill -j 12345 -s 9\n";
|
||||
|
||||
static const char USAGE_JOBCKPT[] =
|
||||
"jobckpt -j <jobid> [-k] [-m <method>]\n"
|
||||
" -j <jobid> jobid to checkpoint (required)\n"
|
||||
" -k kill the job immediately after checkpointing\n"
|
||||
" -m <method> checkpoint method name (default \"default\")\n"
|
||||
"example: jobckpt -j 12345 -m default\n";
|
||||
|
||||
static const char USAGE_JOBINFO[] =
|
||||
"jobinfo -j <jobid> [-j <jobid> ...]\n"
|
||||
" -j <jobid> 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 <jobid>] [-m <host>[:<n>]]... [-u <user>] [-c <cmdline>]\n"
|
||||
" [-x <exdata>] [-d <dir>] [-t <launch_type>]\n"
|
||||
" -j <jobid> jobid this task run belongs to\n"
|
||||
" -m <host>[:<n>] host to run on, optional :ntasks count\n"
|
||||
" (default 1); may be repeated up to 5 times\n"
|
||||
" -u <user> exec user, applied to every host\n"
|
||||
" -c <cmdline> command line, applied to every host\n"
|
||||
" -x <exdata> exdata string, applied to every host\n"
|
||||
" -d <dir> exec home/cwd directory, applied to every host\n"
|
||||
" -t <launch_type> 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 <jobid> -p <reason> [-p <reason> ...]\n"
|
||||
" -j <jobid> jobid to set the pend reason for (required)\n"
|
||||
" -p <reason> pend reason spec, may repeat: `<code>` sets the main\n"
|
||||
" reason; `<host>:<code>` adds a host-based reason;\n"
|
||||
" `<res>:<host>:<code>` 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 <cmd>` 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;
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -68,6 +68,7 @@
|
|||
#include <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <signal.h>
|
||||
#include <stdatomic.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -8,10 +8,15 @@
|
|||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <rtti.h>
|
||||
#include <unity/unity.h>
|
||||
|
||||
#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);
|
||||
|
|
|
|||
|
|
@ -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 —
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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 (<builddir>/controller/xce_controller ->
|
||||
# <builddir>/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: <builddir>/controller/xce_controller + <builddir>/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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue