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

fix(controller): close 6 data-safety races + apply-pump O(N^2) regression + boot TOCTOU

Browse source 
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:
Rongsong Shen 2026-08-16 10:20:02 +00:00
parent 3afe3d5549
commit 45400c3a28
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
18 changed files with 1499 additions and 195 deletions
Download patch file Download diff file Expand all files Collapse all files

204
src/controller/persist.c
View file

@ -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;