fix(raft-apply-decoupling): kill-mid-fact-stream soak + close bug-1416 boot-time TOCTOU (task 7.2)
Adds --factstream HA soak mode driving real jstart/jfinish/jfail worker-fact traffic against a live 3-node cluster, killing the primary mid-stream and asserting zero committed-state divergence and zero phantom fact reactions. Building this scenario surfaced bug-1416: persist_recover()'s boot-time apply-correctness verification can race the live apply pump and the standby live-replication apply path, both of which mutate the same last_applied_lsn/ job-table state the verification re-reads live -- producing a false APPLY_DIVERGENT that permanently fences an otherwise-healthy node. Fixed with a new persist_t.recovery_complete gate that defers only the in-memory apply (never the durability write or replication ACK) while a recovery pass is in flight, closed from the earliest safe point in myapp_run() so an election- retry candidate can't apply live traffic before persist_recover() ever runs. Round-2 adversarial review independently verified the ACK/durability paths stay ungated, reproduced the myapp.c fix as load-bearing via revert-and- reproduce, and confirmed real-divergence detection still fires. Live 3-node factstream soak: 2/2 rounds PASS, zero APPLY_DIVERGENT across 5 recover calls. controller_unit_test: zero new failures across 23 groups.
This commit is contained in:
parent
5f98d1b0ef
commit
a5e29a6ffa
17 changed files with 74793 additions and 156 deletions
|
|
@ -2229,6 +2229,26 @@ static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e)
|
|||
return prc;
|
||||
}
|
||||
|
||||
/* bug-1416 (raft-apply-decoupling task 7.2 round 2): defer the in-memory
|
||||
* entity apply while this node's OWN boot-time persist_recover() replay+
|
||||
* verification pass is still running. The entry is already durably
|
||||
* persisted above (persist_wal_write_replicated, fsync'd) — only the
|
||||
* ACK and applied-lsn advancement depend on that, per the persist-
|
||||
* before-apply-before-ACK contract this function's own header comment
|
||||
* documents, and bug-1043's precedent of decoupling durability/ACK from
|
||||
* entity materialization. Applying here too, concurrently with
|
||||
* replay_wal()'s single-threaded, unlocked bookkeeping (last_applied_lsn)
|
||||
* and the live job table it reads, can produce a false APPLY_DIVERGENT
|
||||
* verdict on a perfectly healthy boot (confirmed live). No entry is
|
||||
* lost: this node's own in-progress replay_wal() pass will pick it up
|
||||
* directly off disk if it lands within that pass's scan, and
|
||||
* persist_recover()'s own deferred catch-up kick (persist_apply_pump_
|
||||
* kick, issued right before it returns) covers it unconditionally
|
||||
* either way, since apply_pump_read_entry() does a fresh on-disk lookup
|
||||
* every call rather than trusting any earlier directory listing. */
|
||||
if (!persist_recovery_complete(p))
|
||||
return 0;
|
||||
|
||||
/* Logical entity restore only. persist_wal_write_replicated (above) already
|
||||
* wrote the entry and owns the on-disk CRC chain + current_lsn — so we pass
|
||||
* the entry fields directly rather than fabricating a header. (bug-778:
|
||||
|
|
@ -2319,9 +2339,21 @@ static int ha_apply_term_entry_via_library(ha_state_t *ha, persist_t *p,
|
|||
* find a committed job (it IS on disk; walcompare passes), looking like a
|
||||
* committed-job loss. Re-apply here, mirroring the legacy path. Idempotent
|
||||
* on a re-stream (restore_job upserts by jobid). WAL_NOOP / WAL_CONFIG_
|
||||
* MARKER carry no entity payload — apply_wal_entry no-ops on them. */
|
||||
apply_wal_entry(p, (uint32_t)e->entry_type, e->lsn,
|
||||
e->raw_payload, (uint32_t)e->raw_payload_len);
|
||||
* MARKER carry no entity payload — apply_wal_entry no-ops on them.
|
||||
*
|
||||
* bug-1416 (raft-apply-decoupling task 7.2 round 2): gated on
|
||||
* persist_recovery_complete() for the identical reason documented on
|
||||
* ha_apply_single_entry's own matching guard above — defer the
|
||||
* in-memory apply while this node's own boot-time persist_recover()
|
||||
* pass is in progress, never the ACK/applied_lsn advance below (the
|
||||
* entry is already durably persisted by on_persist_entry's
|
||||
* persist_wal_write_replicated, invoked via ha_consume_append_entries
|
||||
* above). This node's own in-progress replay, or the deferred
|
||||
* catch-up kick persist_recover() issues before returning, applies
|
||||
* it instead. */
|
||||
if (persist_recovery_complete(p))
|
||||
apply_wal_entry(p, (uint32_t)e->entry_type, e->lsn,
|
||||
e->raw_payload, (uint32_t)e->raw_payload_len);
|
||||
|
||||
ha_advance_applied_lsn(ha, e->lsn);
|
||||
ha_send_wal_ack_reply(ha, ep, c, e->lsn);
|
||||
|
|
|
|||
|
|
@ -200,6 +200,23 @@ int myapp_run(myconf_t *myconf)
|
|||
return rc;
|
||||
}
|
||||
|
||||
/* bug-1416 round 2: close the recovery_complete gate HERE, at the
|
||||
* earliest safe moment, not just around persist_recover()'s own call
|
||||
* later (cluster_init_phase2, well after election). This node's own
|
||||
* WAL/committed-state materialization is not truly "recovered" until
|
||||
* that call runs -- and this process can receive and durably APPLY
|
||||
* live-replicated entries (ha_apply_single_entry / ha_apply_term_
|
||||
* entry_via_library) before then, e.g. while stuck retrying its own
|
||||
* election as a candidate that keeps losing/timing out before
|
||||
* discovering a live primary and stepping down (confirmed live: a
|
||||
* node stuck ELECTING for tens of seconds while already accepting
|
||||
* AppendEntries from a genuine leader). Set BEFORE ha_start_election_
|
||||
* async/cluster_run() below can process any network traffic, so no
|
||||
* live-replication apply is ever reachable before this gate is
|
||||
* closed. persist_recover() clears it (and issues a catch-up kick)
|
||||
* once its own replay+verification completes. */
|
||||
persist_mark_recovery_pending(&c.persist);
|
||||
|
||||
/* ── Step 7: pre-load config_tag ─────────────────────────────────────── */
|
||||
|
||||
/* Pre-load config from disk if a local config file exists, so that this
|
||||
|
|
|
|||
|
|
@ -618,6 +618,14 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
|
|||
p->apply_verdict = APPLY_CLEAN;
|
||||
atomic_store(&p->last_applied_lsn, 0);
|
||||
|
||||
/* bug-1416: defaults to true (nothing to gate) so every fixture that
|
||||
* calls persist_init() but never persist_recover() -- most of this
|
||||
* spec's own test suite -- sees identical, ungated apply-pump/live-
|
||||
* replication behavior to before this flag existed. Only persist_
|
||||
* recover() ever sets this false, for its own single-threaded
|
||||
* replay+verify duration (see the field's doc comment in persist.h). */
|
||||
atomic_store(&p->recovery_complete, true);
|
||||
|
||||
/* Durable-advance wake hook starts unregistered (ha-commit-durability 2.2):
|
||||
* the commit engine registers it in commit_engine_init. Until then every
|
||||
* flusher durable advance is a no-op wake. */
|
||||
|
|
@ -3228,6 +3236,31 @@ void persist_apply_pump_kick(persist_t* p, void* pool_arg)
|
|||
if (!p || !p->cluster)
|
||||
return;
|
||||
|
||||
/* bug-1416 (raft-apply-decoupling task 7.2 round 2): defer while this
|
||||
* node's own boot-time persist_recover() replay+verification pass is
|
||||
* still running -- last_applied_lsn is being written UNLOCKED by
|
||||
* replay_wal() during that window (single-threaded-recovery
|
||||
* assumption), and persist_verify_apply_correctness() is about to
|
||||
* compare a live re-read of it (and of the live job table) against a
|
||||
* WAL-scan-derived expectation. Letting this walk run concurrently
|
||||
* would race BOTH: it could advance last_applied_lsn out from under
|
||||
* replay_wal's own plain (non-CAS) store, and it mutates the same job
|
||||
* table persist_verify_apply_correctness re-reads as "actual" --
|
||||
* producing a false APPLY_DIVERGENT verdict on a perfectly healthy
|
||||
* boot (confirmed live). No work is lost: persist_recover() sets this
|
||||
* flag back to true and issues exactly one catch-up kick right before
|
||||
* it returns, which will walk from last_applied_lsn+1 up through
|
||||
* whatever committed_lsn has grown to by then -- a full, ordered
|
||||
* drain, not a partial one. Intentionally checked with NO lock: a
|
||||
* stale `true` read racing the flag's transition to `false` is
|
||||
* harmless (the walk itself still only advances one entry at a time
|
||||
* under apply_pump_lock, so at worst this call proceeds a few
|
||||
* instructions into a window persist_recover() is about to close, and
|
||||
* the deferred re-kick after recover() still runs); a stale `false`
|
||||
* merely defers one no-op-cost kick to the next real trigger. */
|
||||
if (!atomic_load_explicit(&p->recovery_complete, memory_order_acquire))
|
||||
return;
|
||||
|
||||
int expected = 0;
|
||||
if (!atomic_compare_exchange_strong_explicit(
|
||||
&p->apply_pump_scheduled, &expected, 1, memory_order_acq_rel,
|
||||
|
|
@ -3635,12 +3668,47 @@ uint64_t persist_last_applied_lsn(persist_t* p)
|
|||
: 0;
|
||||
}
|
||||
|
||||
/* bug-1416 (raft-apply-decoupling task 7.2 round 2) -- see persist.h's
|
||||
* doc comment on the recovery_complete field for the full contract. */
|
||||
bool persist_recovery_complete(persist_t* p)
|
||||
{
|
||||
return !p || atomic_load_explicit(&p->recovery_complete,
|
||||
memory_order_acquire);
|
||||
}
|
||||
|
||||
/* bug-1416 round 2 -- see persist.h's doc comment on this function for the
|
||||
* full rationale (closes the gap between persist_init() and the FIRST
|
||||
* persist_recover() call, which persist_recover()'s own false/true bracket
|
||||
* cannot see). */
|
||||
void persist_mark_recovery_pending(persist_t* p)
|
||||
{
|
||||
if (!p)
|
||||
return;
|
||||
atomic_store_explicit(&p->recovery_complete, false, memory_order_release);
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Public: persist_recover */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
int persist_recover(persist_t* p)
|
||||
{
|
||||
/* bug-1416 (raft-apply-decoupling task 7.2 round 2): close the boot-time
|
||||
* race window for the FULL duration of this call, not just the position
|
||||
* check inside persist_verify_apply_correctness -- see the
|
||||
* recovery_complete field's doc comment (persist.h) for the complete
|
||||
* rationale. Held false until the natural (success) return at the
|
||||
* bottom of this function, which restores it and issues one deliberate
|
||||
* catch-up kick. The two hard-I/O-error early returns below
|
||||
* (load_snapshot / replay_wal failure) deliberately do NOT restore it:
|
||||
* those are pre-existing, rare disk-I/O-level failures the caller
|
||||
* already treats as fatal to this boot attempt (cluster_init_phase2
|
||||
* tears down persist_t on a nonzero return) -- leaving the gate closed
|
||||
* is the conservative, fail-static choice (never let the live apply
|
||||
* pump or live-replication path mutate state on top of a persist_t
|
||||
* whose own recovery never finished), not a new hazard. */
|
||||
atomic_store_explicit(&p->recovery_complete, false, memory_order_release);
|
||||
|
||||
/* --- Step 1: find latest snapshot (highest LSN filename) --- */
|
||||
char snap_file[4096] = "";
|
||||
uint64_t snap_lsn = 0;
|
||||
|
|
@ -3823,6 +3891,40 @@ int persist_recover(persist_t* p)
|
|||
|
||||
/* --- Step 4: retention pass (TTL check) — no-op in P0 (no entities) --- */
|
||||
|
||||
/* bug-1416: replay + verification is fully complete -- safe to let the
|
||||
* live apply pump and the standby live-replication apply path resume
|
||||
* mutating state (see the recovery_complete field's doc comment).
|
||||
* Restored unconditionally here regardless of the verdict computed
|
||||
* above: a genuine APPLY_DIVERGENT already fenced this member via
|
||||
* persist_recover_loud_fence_if_unrecoverable (non-promotable, stays
|
||||
* alive, keeps retrying) -- that sticky fence and its clearing logic
|
||||
* (task 5.3) are untouched by this flag, which governs ONLY whether
|
||||
* in-memory apply may proceed, not whether this member may lead. Issue
|
||||
* one deliberate catch-up kick for anything that committed locally (or
|
||||
* arrived via live replication and was durably persisted but had its
|
||||
* entity-apply deferred, see ha_apply_single_entry / ha_apply_term_
|
||||
* entry_via_library) while this pass ran -- persist_apply_pump_run's
|
||||
* own walk drains fully from last_applied_lsn+1 through the current
|
||||
* committed_lsn, so this is a complete, ordered catch-up, not a partial
|
||||
* one; a harmless no-op if nothing queued. */
|
||||
atomic_store_explicit(&p->recovery_complete, true, memory_order_release);
|
||||
/* n_vcpus==0 is this codebase's own established "not yet initialized"
|
||||
* sentinel for vtaskmgr_t (mirrors cluster_init_phase2's own
|
||||
* `if (c->vtaskmgr.n_vcpus == 0) vtaskmgr_init(...)` guard) -- a real
|
||||
* cluster always has n_vcpus > 0 by the time persist_recover() runs
|
||||
* (vtaskmgr_init happens earlier in myapp_run()/phase2), while a bare
|
||||
* recovery-focused unit-test cluster_t (none of which call
|
||||
* vtaskmgr_init, confirmed by inspection) leaves it zeroed. Passing a
|
||||
* pointer to an uninitialized vtaskmgr_t into vtask_enq_classified
|
||||
* would be unsafe, so fall back to persist_apply_pump_kick's own
|
||||
* existing, already-exercised NULL-pool inline path in that case --
|
||||
* identical to how every apply-pump test fixture in this spec already
|
||||
* exercises the pump. */
|
||||
void* pool = (p->cluster && p->cluster->vtaskmgr.n_vcpus > 0)
|
||||
? &p->cluster->vtaskmgr
|
||||
: NULL;
|
||||
persist_apply_pump_kick(p, pool);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -107,6 +107,8 @@ extern int cmd_shutdown(int, char*[]);
|
|||
extern int cmd_set(int, char*[]);
|
||||
extern int cmd_reslist(int, char*[]);
|
||||
extern int cmd_join_worker(int, char*[]);
|
||||
extern int cmd_jobstart(int, char*[]);
|
||||
extern int cmd_jobfail(int, char*[]);
|
||||
extern int cmd_join_sched(int, char*[]);
|
||||
extern int cmd_workerlist(int, char*[]);
|
||||
extern int cmd_joblist(int, char*[]);
|
||||
|
|
@ -206,7 +208,9 @@ static command_t worker_commands[] = {
|
|||
{ "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 }
|
||||
|
|
@ -1534,6 +1538,16 @@ static void print_jobinfo(xce_job_t* job)
|
|||
fmt_jobid(idstring, sizeof(idstring) - 1, job->jobid);
|
||||
printf("JobID: %s\n", idstring);
|
||||
printf("state: %s\n", job_state_str(job->state));
|
||||
/* raft-apply-decoupling task 7.2: job_state_str's if/else-if chain checks
|
||||
* JOB_STAT_RUN before JOB_STAT_EXIT, so a job that is simultaneously RUN
|
||||
* and EXIT|PERR (e.g. a worker-reported FAIL received without a prior
|
||||
* FINISH — apply_job_fail_transition ORs in EXIT|PERR without clearing
|
||||
* RUN) prints as "RUN", silently hiding the EXIT/PERR bits from the
|
||||
* pretty string. Print the raw bitmask too so a scriptable caller (the
|
||||
* xce-test-ha factstream regression scenario's phantom-fact check) can
|
||||
* test for a specific bit's presence unambiguously instead of relying on
|
||||
* the display string's priority ordering. */
|
||||
printf("state_raw: 0x%x\n", job->state);
|
||||
printf("Queue: %s\n", job->submit->queue);
|
||||
printf("User: %s\n", job->submit->user);
|
||||
printf("Slots: %d,%d\n", job->submit->mintasks, job->submit->maxtasks);
|
||||
|
|
@ -2443,6 +2457,69 @@ int cmd_join_worker(int argc, char* argv[])
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* jstart -j jobid [-p pid] — raft-apply-decoupling task 7.2: synthetic
|
||||
* worker-fact-stream primitive. Mirrors cmd_jobfinish's existing pattern
|
||||
* exactly (same getopt/parse_jobid/SET_ERROR shape) but drives the
|
||||
* xcecli_notify_jobstart wire call that cmd_jobfinish's sibling
|
||||
* xcecli_notify_jobfinish already exercises — libxce (client.c) already
|
||||
* implements the full worker-fact API (jobstart/jobfinish/jobfail/...); this
|
||||
* command was the missing REPL-level primitive for jobstart specifically.
|
||||
* Drives the SAME production receipt path (on_job_start -> record_job_start_fact
|
||||
* -> persist_submit_job_fact -> durable append -> commit -> apply pump ->
|
||||
* apply_committed_entry()) that a real worker's report would, without
|
||||
* requiring a real scheduler dispatch (on_job_start only requires the job to
|
||||
* already exist in jobmgr, i.e. have been submitted). */
|
||||
int cmd_jobstart(int argc, char* argv[])
|
||||
{
|
||||
int rc;
|
||||
int opt;
|
||||
uint64_t jobid = 0;
|
||||
unsigned int pid = 0;
|
||||
unsigned int pids[1];
|
||||
job_pidinfo_t info;
|
||||
|
||||
optind = 1;
|
||||
memset(&info, 0x0, sizeof(info));
|
||||
|
||||
while ((opt = getopt(argc, argv, "j:p:")) != -1) {
|
||||
switch (opt) {
|
||||
case 'j':
|
||||
parse_jobid(&jobid, optarg);
|
||||
break;
|
||||
case 'p':
|
||||
pid = (unsigned int)atoi(optarg);
|
||||
break;
|
||||
default:
|
||||
printf(" jstart -j jobid [-p pid]\n");
|
||||
return -EINVAL;
|
||||
}
|
||||
}
|
||||
|
||||
if (jobid == 0) {
|
||||
printf(" you need specify a positive jobid\n");
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
pids[0] = (pid != 0) ? pid : (unsigned int)getpid();
|
||||
info.jobid = jobid;
|
||||
info.npids = 1;
|
||||
info.pids = pids;
|
||||
info.npgids = 0;
|
||||
info.pgids = NULL;
|
||||
info.exdata = NULL;
|
||||
|
||||
rc = xcecli_notify_jobstart(active_ctx->ctx, &info);
|
||||
if (rc < 0) {
|
||||
SET_ERROR(-rc);
|
||||
perror("xcecli_jobstart");
|
||||
return rc;
|
||||
}
|
||||
|
||||
printf("ok\n");
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int cmd_jobfinish(int argc, char* argv[])
|
||||
{
|
||||
int rc;
|
||||
|
|
@ -2451,6 +2528,17 @@ int cmd_jobfinish(int argc, char* argv[])
|
|||
int code = 0;
|
||||
xce_jobfinish_t finish;
|
||||
|
||||
/* raft-apply-decoupling task 7.2: reset optind so repeated jfinish
|
||||
* invocations within one xce_test process (the fact-stream soak scenario
|
||||
* calls this many times in a single worker session) each re-parse their
|
||||
* own argv from the start. Without this, getopt's optind carries over
|
||||
* from the PRIOR command's parse position, silently skipping this
|
||||
* command's own "-j jobid" on the second and later calls (jobid stays 0
|
||||
* and the command errors out) — matches the reset convention most other
|
||||
* cmd_ handlers in this file already follow (e.g. cmd_join_worker just
|
||||
* above, cmd_jobsub, cmd_resupdate). */
|
||||
optind = 1;
|
||||
|
||||
while ((opt = getopt(argc, argv, "j:c:")) != -1) {
|
||||
switch (opt) {
|
||||
case 'j':
|
||||
|
|
@ -2485,6 +2573,64 @@ int cmd_jobfinish(int argc, char* argv[])
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* jfail -j jobid [-c code] [-s stage] — raft-apply-decoupling task 7.2:
|
||||
* synthetic worker-fact-stream primitive, sibling of jstart/jfinish above.
|
||||
* Drives xcecli_notify_jobfail -> on_job_fail -> record_job_fail_fact ->
|
||||
* persist_submit_job_fact_ex (the _pad-discriminated path, task 6.1 round 2)
|
||||
* -> apply_job_fail_transition() at apply time. Reuses the existing
|
||||
* job_failinfo_t shape already exercised internally by this file's own
|
||||
* notify_job_fail() helper (used by run_job() on a local exec failure). */
|
||||
int cmd_jobfail(int argc, char* argv[])
|
||||
{
|
||||
int rc;
|
||||
int opt;
|
||||
uint64_t jobid = 0;
|
||||
int code = 0;
|
||||
int stage = JOB_EXEC_STAGE_POST_EXEC;
|
||||
job_failinfo_t failinfo;
|
||||
|
||||
optind = 1;
|
||||
memset(&failinfo, 0x0, sizeof(failinfo));
|
||||
|
||||
while ((opt = getopt(argc, argv, "j:c:s:")) != -1) {
|
||||
switch (opt) {
|
||||
case 'j':
|
||||
parse_jobid(&jobid, optarg);
|
||||
break;
|
||||
case 'c':
|
||||
code = atoi(optarg);
|
||||
break;
|
||||
case 's':
|
||||
stage = atoi(optarg);
|
||||
break;
|
||||
default:
|
||||
printf(" jfail -j jobid [-c code] [-s stage]\n");
|
||||
return -EINVAL;
|
||||
}
|
||||
}
|
||||
|
||||
if (jobid == 0) {
|
||||
printf(" you need specify a positive jobid\n");
|
||||
return -EINVAL;
|
||||
}
|
||||
|
||||
failinfo.jobid = jobid;
|
||||
failinfo.stage = stage;
|
||||
failinfo.code = code;
|
||||
failinfo.detail = "synthetic fact-stream failure (xce_test jfail)";
|
||||
|
||||
rc = xcecli_notify_jobfail(active_ctx->ctx, &failinfo);
|
||||
if (rc < 0) {
|
||||
SET_ERROR(-rc);
|
||||
perror("xcecli_jobfail");
|
||||
return rc;
|
||||
}
|
||||
|
||||
printf("ok\n");
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int set_resval(xce_resval_t* vals, int idx, char* name, char* val)
|
||||
{
|
||||
static struct {
|
||||
|
|
|
|||
|
|
@ -84,6 +84,43 @@ static int decode_submit_info(const uint8_t* payload, uint32_t len,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* True for the WAL types whose payload is the flat, non-capnp
|
||||
* wal_jobstate_t {jobid, state, _pad} (wal_types.h) — every worker-reported
|
||||
* job-state fact (START/FINISH/FAIL/STOP/RESUME/UPDATE/RUSAGE/CKPT/PREASON).
|
||||
* Added for raft-apply-decoupling task 7.2: the kill-mid-fact-stream live
|
||||
* regression needs to verify, per jobid, which fact WAL entries actually
|
||||
* committed on each node — before this, waldump only decoded jobid for
|
||||
* WAL_JOB_SUBMIT, leaving every fact entry printed as "jobid=?". */
|
||||
static int wal_type_is_jobstate(uint32_t type)
|
||||
{
|
||||
switch (type) {
|
||||
case WAL_JOB_START:
|
||||
case WAL_JOB_FINISH:
|
||||
case WAL_JOB_FAIL:
|
||||
case WAL_JOB_STOP:
|
||||
case WAL_JOB_RESUME:
|
||||
case WAL_JOB_UPDATE:
|
||||
case WAL_JOB_RUSAGE:
|
||||
case WAL_JOB_CKPT:
|
||||
case WAL_JOB_PREASON:
|
||||
return 1;
|
||||
default:
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/* Decode a wal_jobstate_t payload. Native-endian flat struct (single-node WAL
|
||||
* — see wal_types.h), so this is a bounds-checked copy, not a parser. Returns
|
||||
* 0 on success, -1 if the payload is too short to hold the struct. */
|
||||
static int decode_jobstate_info(const uint8_t* payload, uint32_t len,
|
||||
wal_jobstate_t* out)
|
||||
{
|
||||
if (!payload || len < sizeof(*out))
|
||||
return -1;
|
||||
memcpy(out, payload, sizeof(*out));
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* True if `dir` directly contains at least one seg-*.wal file. */
|
||||
static int dir_has_segments(const char* dir)
|
||||
{
|
||||
|
|
@ -342,6 +379,25 @@ static void scan_segment(const char* seg_path, FILE* out,
|
|||
}
|
||||
*prev_submit_jobid = jobid;
|
||||
*prev_submit_pcrc = pcrc;
|
||||
} else if (wal_type_is_jobstate(hdr.type)) {
|
||||
/* Worker-reported job-state fact (task 7.2): decode the flat
|
||||
* {jobid, state, _pad} payload so a per-jobid committed-fact check
|
||||
* (e.g. "did WAL_JOB_FINISH for jobid X actually commit on this
|
||||
* node") can grep waldump output directly, the same way the
|
||||
* existing WAL_JOB_SUBMIT branch above already supports
|
||||
* expect-job/joblist cross-checks. */
|
||||
wal_jobstate_t js;
|
||||
if (decode_jobstate_info(payload, hdr.payload_len, &js) == 0) {
|
||||
char jidStr[128];
|
||||
if (fmt_jobid(jidStr, sizeof(jidStr), js.jobid) < 0)
|
||||
snprintf(jidStr, sizeof(jidStr), "%" PRIu64, js.jobid);
|
||||
snprintf(info, sizeof(info), "jobid=%s state=0x%x pad=0x%x",
|
||||
jidStr, js.state, js._pad);
|
||||
} else {
|
||||
snprintf(info, sizeof(info), "jobid=?");
|
||||
}
|
||||
/* a non-submit entry breaks the adjacency for dup detection */
|
||||
*prev_submit_jobid = 0;
|
||||
} else {
|
||||
/* a non-submit entry breaks the adjacency for dup detection */
|
||||
*prev_submit_jobid = 0;
|
||||
|
|
|
|||
|
|
@ -363,6 +363,38 @@ struct persist_s {
|
|||
_Atomic uint64_t last_applied_lsn;
|
||||
apply_verdict_t apply_verdict;
|
||||
|
||||
/* bug-1416 (raft-apply-decoupling task 7.2 round 2): closes a boot-time
|
||||
* TOCTOU between persist_recover()'s replay+verification pass above and
|
||||
* two concurrently-reachable live mutation paths that write the SAME
|
||||
* last_applied_lsn field / mutate the SAME live job table this pass
|
||||
* reads: the live apply pump (persist_apply_pump_run, driven by any of
|
||||
* the 6 commit.c committed_lsn-advance sites via persist_apply_pump_kick)
|
||||
* and the standby live-replication apply path (ha_apply_single_entry /
|
||||
* ha_apply_term_entry_via_library, ha.c). Both are reachable the moment
|
||||
* the iomgr event loop starts processing connections in cluster_run()
|
||||
* (myapp.c) -- which happens well BEFORE cluster_init_phase2() reaches
|
||||
* its persist_recover() call (an election round-trip and the whole of
|
||||
* phase2's own setup earlier in the same function separate the two).
|
||||
* Without this flag a fast-electing peer's replicated traffic can mutate
|
||||
* last_applied_lsn/the job table WHILE this node's own persist_recover()
|
||||
* is still computing its replay-derived expectation, producing a FALSE
|
||||
* APPLY_DIVERGENT verdict on a perfectly healthy boot (confirmed live,
|
||||
* reproduced on a fresh 3-node cluster's ordinary two-phase bootstrap --
|
||||
* committedLsn was identical across all nodes when it fired).
|
||||
*
|
||||
* Defaults to true (persist_init) so every existing unit-test fixture
|
||||
* that calls persist_init() directly and never calls persist_recover()
|
||||
* (the overwhelming majority of this spec's own test suite) sees
|
||||
* IDENTICAL, ungated behavior to before this field existed. persist_
|
||||
* recover() is the ONLY function that ever sets this false -- for the
|
||||
* exact, single-threaded duration of its own replay+verify pass -- and
|
||||
* always restores it to true (plus issues one deliberate catch-up kick)
|
||||
* before returning, regardless of the resulting verdict. This does NOT
|
||||
* touch the sticky recovery_unrecoverable fence or its clearing logic
|
||||
* (task 5.3) -- a genuinely DIVERGENT verdict still fences exactly as
|
||||
* before; this flag only prevents the comparison ITSELF from racing. */
|
||||
_Atomic bool recovery_complete;
|
||||
|
||||
/* Paths (derived from workdir + name via persist_set_workdir) */
|
||||
char wal_path[4096];
|
||||
char snap_path[4096];
|
||||
|
|
@ -706,6 +738,45 @@ persist_recover_loud_fence_if_unrecoverable(persist_t* p,
|
|||
bool persist_recovery_unrecoverable(persist_t* p);
|
||||
void persist_clear_recovery_unrecoverable(persist_t* p);
|
||||
|
||||
/* persist_recovery_complete (bug-1416, raft-apply-decoupling task 7.2 round
|
||||
* 2) -- true unless a persist_recover() call is CURRENTLY in progress on
|
||||
* this persist_t (see the recovery_complete field doc comment). Consulted
|
||||
* by the standby live-replication apply path (ha_apply_single_entry /
|
||||
* ha_apply_term_entry_via_library, ha.c) to defer in-memory entity mutation
|
||||
* while a boot-time replay+verification pass is running, so it never races
|
||||
* that pass's expectation-vs-actual comparison. Durability (WAL persist)
|
||||
* and the replication ACK are NEVER gated by this -- only the in-memory
|
||||
* apply; a deferred entry is picked up either by the in-progress replay
|
||||
* pass itself (if it observes the entry on disk) or by the single catch-up
|
||||
* kick persist_recover() issues right before returning. Safe on NULL p
|
||||
* (returns true -- nothing to defer against). */
|
||||
bool persist_recovery_complete(persist_t* p);
|
||||
|
||||
/* persist_mark_recovery_pending (bug-1416 round 2 -- live-cluster reproduction
|
||||
* confirmed a gap in the original fix). persist_recover() itself already
|
||||
* brackets recovery_complete false/true around its own call, but that window
|
||||
* is too NARROW for the real production boot sequence: a node can be
|
||||
* receiving and durably APPLYING live-replicated entries (ha_apply_single_
|
||||
* entry / ha_apply_term_entry_via_library -- e.g. while stuck retrying its
|
||||
* own election as a candidate that keeps losing/timing out before it
|
||||
* discovers a live primary and steps down) for an arbitrarily long stretch
|
||||
* BEFORE cluster_init_phase2() ever gets to call persist_recover() for the
|
||||
* first time. During that whole stretch recovery_complete still holds
|
||||
* persist_init()'s default (true), so the live-replication apply path runs
|
||||
* UNGATED and can mutate jobmgr/last_applied_lsn well before persist_
|
||||
* recover()'s own baseline capture, producing the exact same false
|
||||
* APPLY_DIVERGENT this flag exists to prevent, just from an earlier entry
|
||||
* point than persist_recover() alone can see.
|
||||
*
|
||||
* Call this ONCE, in the real boot path only (myapp_run, immediately after
|
||||
* persist_init()/persist_start() succeed and BEFORE election / cluster_run()
|
||||
* can process any network traffic), to close the gate from the earliest
|
||||
* safe moment instead of only from persist_recover()'s own entry. Safe on
|
||||
* NULL p (no-op). Never called by any test fixture -- every unit test that
|
||||
* constructs a persist_t directly (bypassing myapp_run entirely) is
|
||||
* unaffected and keeps persist_init()'s default (true, ungated). */
|
||||
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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue