From 2fa955b92e763f76c764d55212f83281f4997690 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 4 Jul 2026 19:00:43 +0000 Subject: [PATCH] feat(raft-apply-decoupling): ordered live apply pump activates apply_committed_entry (task 6.1) Build persist_apply_pump_run()/persist_apply_pump_kick() -- the ordered walk from last_applied_lsn+1 through committed_lsn, applying every entry via apply_committed_entry() regardless of commit mode. This is the first live (non-replay) consumer of that dispatch function. Kick is wired into commit_resolve_pending() (the single choke point all 6 internal committed_lsn-advancing paths funnel through, discovered in task 1.2), dispatched via vtask_enq_classified(VTASK_LIVE_BLOCK) with a CAS-coalesced fire-and-forget trigger. Fixes the two hazards flagged since task 4.1 as required before going live: 1. Double-append: a thread-local call-provenance guard (t_apply_seam_notifying) prevents persist_job_hook_fun/ persist_queue_hook_fun/persist_worker_hook_fun from re-appending entries the apply pump itself just applied, while preserving job_ctrl's legitimate live command-path appends (a naive detail-based filter would have broken SIGCONT/SIGSTOP). Also closes an undocumented instance of the same hazard in the queue/worker hook functions. 2. Merge-staleness: record_job_fail_fact/postdone's fact recording previously baked a receipt-time-computed final job state into the WAL entry, which goes stale under deferred apply -- a FINISH followed shortly by an ordinary post-exec FAIL could silently clobber a correct DONE completion with a stale RUN|EXIT|PERR, no guard catching it. Fixed by repurposing wal_jobstate_t's unused _pad field to carry just enough data (FAIL's stage) to recompute against the job's actual current state at apply time via the real apply_job_fail_transition()/new apply_job_postdone_transition() guard cores, instead of overwriting with a stale absolute value. Backward compatible: _pad==0 (all prior data) routes to the same legacy overwrite behavior. Also fixes bug-1409 (unbounded busy-loop in apply_pump_drain when committed_lsn advances without a corresponding on-disk entry) and a genuine test-flakiness hazard (a second, independent apply-pump kicker from the fsync_worker thread that existing tests didn't account for). Two pre-existing tests in test_commit_barrier_reply.c now fail as expected -- their own assertions state "no job may be constructed until task 6's apply pump exists," now correctly invalidated by this task landing; updating them is task 7.3's explicit job. --- .kiro/specs/raft-apply-decoupling/tasks.md | 3 +- src/controller/commit.c | 17 + src/controller/include/cluster.h | 11 + src/controller/job.c | 150 ++-- src/controller/meson.build | 1 + src/controller/persist.c | 565 +++++++++++++- src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_apply_pump.c | 715 ++++++++++++++++++ .../testsuite/test_commit_barrier_reply.c | 164 ++-- src/include/persist.h | 157 +++- src/include/wal_types.h | 37 + 11 files changed, 1687 insertions(+), 135 deletions(-) create mode 100644 src/controller/testsuite/test_apply_pump.c diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index bc500aa0..8b8e7ae4 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -129,7 +129,7 @@ - _Depends: 5.2_ - [ ] 6. Ordered apply pump integration -- [ ] 6.1 Build the strict-order walk that applies every newly committed entry +- [x] 6.1 Build the strict-order walk that applies every newly committed entry - Walk entries from the last-applied position up to the current committed position, in ascending order, with no skipping or reordering - Invoke the shared apply dispatch function for every entry in that range, regardless of its commit mode - Observable: a burst of interleaved command and fact entries committing out of submission order still applies in commit order @@ -193,6 +193,7 @@ - **1.2**: implemented `commit_resolve()` + wired it into all 6 commit.c-internal paths that can advance `committed_lsn`/`writable_lsn` (the COMMITTED/majority-ack outcome). Chose `VTASK_LIVE_BLOCK` for dispatch (never admission-gated, unlike `VTASK_REQ_BLOCK` which can fast-reject — unacceptable for "exactly-once, never unresolved"). **Reviewer caught a real task-graph gap, not a defect in 1.2 itself**: converting `COMMIT_MODE_WAIT` away from a blocking wait also removes the *only* mechanism that ever detected a commit timing out (the caller's own `pthread_cond_timedwait` deadline) — nothing yet arms an equivalent timeout for async-registered continuations, and nothing yet wires demotion/log-truncation detection to call `commit_resolve(FAILED, ...)`. Without this, a pending async command during quorum loss would hang forever instead of failing — worse than today's behavior. Split out as new tasks **1.3** (timeout arming) and **1.4** (demotion/truncation wiring), inserted before the renumbered **1.5** (was 1.3, concurrency test — now exercises the real detection paths from 1.3/1.4 instead of calling `commit_resolve()` directly). Task 3.2 now also depends on 1.3/1.4, since its own "failure/timeout outcome replies" observable is meaningless without them. - **1.3**: added a dedicated per-commit-engine sweep thread (not the shared `delaymgr` — would violate boundary + delaymgr's 200ms poll granularity vs. `DEFAULT_SYNC_TIMEOUT_MS`=500ms would add timing jitter). Race rule: majority-satisfied always wins over an already-expired deadline when both are true under the same lock hold. **Round 1 REJECTED**: `commit_sweep_once()` computed the next-deadline snapshot under `ce->lock`, unlocked to dispatch, then slept on the *stale* snapshot — a `commit_submit_async()` registration/broadcast landing during that unlocked dispatch window was lost (no thread parked on the condvar to receive it), so the sweep thread could sleep on an absent/late deadline indefinitely (worst case: unbounded delay in a low-traffic quorum-loss scenario, violating the "identical failure timing" no-regression bar). **Fixed** by holding `ce->lock` continuously across scan→wait decisions (releasing only for the unlocked dispatch step) and always re-scanning under the lock immediately before ever calling `cond_wait`/`cond_timedwait` — the standard recheck-then-wait idiom. Round 2 reviewer independently reproduced the round-1 bug shape to confirm the fix actually closes it, not just narrows it. **Precedent for 1.4**: any future code that needs to interrupt the sweep thread's wait (e.g., a demotion/truncation handler resolving a pending LSN out-of-band) must go through the same lock+broadcast discipline `commit_submit_async()` uses, or it risks the identical lost-wakeup class of bug. - **1.4**: added `commit_resolve_pending_above(ce, lsn_floor, rc)` (snapshot-then-`commit_resolve()`-per-entry, no parallel bookkeeping). Truncation: a single new `on_truncate`/`truncate_cb_ctx` callback slot on `persist_t` (same inversion-of-control shape as the existing `on_durable_advance` hook) fires from `persist_truncate_wal_above` after releasing `p->lock`, only when a truncation actually removed entries — this ONE wrapper covers all three real truncation call sites (`raft_cb_become_leader`'s post-election truncate, the library's `on_truncate_from` conflict-reconcile, `ha_demotion_convergence_fn`'s post-demotion truncate), verified by the reviewer reading all three. Demotion: `raft_cb_become_follower` calls `commit_resolve_pending_above(&c->commit, 0, -XCE_ERR_STANDBY)` (lsn_floor=0 → fails everything — a demoted node can never authoritatively confirm anything pending; verified against the CP posture). **Known residual, non-blocking**: a fourth internal truncation path (`persist.c` inside `persist_wal_write_replicated`'s tip-conflict-repair branch, STANDBY replicated-apply only) calls `truncate_wal_above_locked()` directly, bypassing the new wrapper — safe today only because demotion resolves ALL pending continuations before role flips to STANDBY (so nothing can still be registered by the time that path runs), but this is an *implicit* ordering invariant, not an asserted one. Flag for whoever next touches the STANDBY replicated-apply path: add a defensive comment/assertion there, or route it through the wrapper too, before relying on it further. Also: bug-1407 was believed closed at this point (70/70 across 3 runs) — **see 1.5's note below, this was a false reading, not a real fix**. +- **6.1**: the most consequential task in the entire spec — this is what finally makes `apply_committed_entry()` live (previously only restart replay, task 2.3, ever called it). Built `persist_apply_pump_run()` (walks `[last_applied_lsn+1..committed_lsn]` strictly ascending, calls `apply_committed_entry()` for every entry regardless of commit mode) and `persist_apply_pump_kick()` (off-thread trigger via `vtask_enq_classified(VTASK_LIVE_BLOCK)`, inline fallback for pool-less fixtures, CAS-coalesced so at most one caller runs the walk at a time — the loser hands off asynchronously, by design, to whichever thread wins). **Design deviation, verified correct and a strict superset**: wired the trigger into `commit_resolve_pending()` (commit.c) rather than only ha.c's `on_commit_advanced` — task 1.2 already discovered `commit_resolve_pending` is the single choke point all 6 internal `committed_lsn`-advancing paths funnel through (verified: exactly 6 call sites, commit.c:353/558/1042/1060/1099/1116, all firing after `ce->lock` releases), while `on_commit_advanced` covers only 1 of them; wiring only there would silently starve the pump for single-node/self-quorum clusters. design.md should be updated to reflect this as the actual integration point. **Fixed Hazard 1 (double-append, flagged since task 4.1)**: a naive `detail==HOOK_DETAIL_JOB_UPDATE` filter would have broken `job_ctrl`'s legitimate live SIGCONT/SIGSTOP append (same hook detail, still-legitimate command path) — fixed instead with a `_Thread_local bool t_apply_seam_notifying` call-provenance guard set only around `apply_committed_entry()`'s own hook-firing calls; `persist_job_hook_fun`/`persist_queue_hook_fun`/`persist_worker_hook_fun` early-return when set. Also caught and fixed an UNDOCUMENTED instance of the identical hazard class in the queue/worker hook functions (neither had ANY mode-based early-return before this, unlike job's pre-existing dead `WAL_MODE_WAIT` branch) — would have double-appended+re-triggered a nested majority-commit wait for every live-applied `WAL_QUEUE_CTRL`/`WAL_WORKER_CTRL` entry. Took 2 review rounds. **Round 1 REJECTED — the most severe finding of the entire session, a real data-corruption bug, not a doc/log issue**: Hazard 2 (merge-staleness, also flagged since 4.1) was investigated and ESCALATED as requiring wire-format changes beyond scope, while still landing the pump that activates it. Round-1 review proved with field-level precision this was wrong to ship: job state DONE, correctly baked by a FINISH fact — followed shortly after (an entirely ordinary HPC lifecycle event, not adversarial) by a POST_EXEC-stage FAIL fact whose receipt-time snapshot is STILL stale (RUN, since FINISH hadn't applied yet, apply now being deferred) — bakes `RUN|EXIT|PERR` into its WAL entry. Both apply in correct LSN order (FINISH then FAIL), but FAIL's stale absolute overwrite wins over the now-correct DONE, because the terminal/non-terminal guard doesn't catch it (both values happen to carry a terminal-ish bit). **Final result: a job's correct DONE completion silently replaced by RUN|EXIT|PERR — a Req 11.3 regression, and a real, operator-and-billing-visible corruption newly introduced by deferring apply (this feature's whole purpose), reachable under completely ordinary traffic.** The reviewer also found a scoped fix contradicting the "needs wire-format enrichment" escalation: `wal_jobstate_t`'s existing unused `_pad` field had room. **Round 2 fixed it properly**: `_pad` now carries a discriminator (`WAL_JOB_FAIL_STAGE_FLAG`/`WAL_JOB_UPDATE_POSTDONE_MARK`) plus the minimal data needed (FAIL's `stage`) to RECOMPUTE against the job's ACTUAL CURRENT state AT APPLY TIME via the real `apply_job_fail_transition()`/new `apply_job_postdone_transition()` guard cores — no more baking a stale absolute value at receipt time. `WAL_JOB_START`/`WAL_JOB_FINISH` are unchanged (confirmed safe on their own — deterministic, state-independent payloads that were never the problem, only the thing getting clobbered). Backward-compatible: `_pad==0` (all pre-existing/legacy data) deterministically routes to the same absolute-overwrite behavior that was universal before this fix — no crash, no new corruption, correctly out of scope for further hardening since this feature has never been deployed. **Bonus finding during round-2 review — a real test-flakiness hazard, root-caused and fixed on the spot, not just reported**: a rare (initially estimated ~2%, actually verified ~7.5%) flake in the new regression test was root-caused to a genuine SECOND, independent kicker of the apply pump (`commit_on_durable_advance`, fired from the real `fsync_worker` background thread on every `durable_lsn` advance, entirely separate from `repl_on_ack`) that the test's own header comment didn't account for — this is by-design eventually-consistent behavior (task 6.2's continuation-firing will correctly rely on it), not a production bug, but the TEST needed a proper poll-for-applied-state helper instead of asserting immediately. Fixed (`tap_wait_applied()` polling helper added to 2 tests) and a SECOND, related flake found during the investigation in a third test, also fixed — 120/120 clean runs afterward. Also fixed a stale orphaned forward-declaration left over from the round-2 rename. **Also fixed a genuine, unrelated bug found via regression sweep, logged as bug-1409**: an early version of `apply_pump_drain`'s retry loop didn't check `persist_apply_pump_run`'s return code, causing an unbounded busy-loop (18.5M+ log lines) in `test_demotion_convergence` whenever `committed_lsn` advances without a corresponding real on-disk WAL entry (a legitimate scenario for commit-engine-focused unit fixtures) — fixed by returning immediately on a negative rc. **Hand-off built for task 6.2**: `int persist_take_pending_apply(persist_t* p, uint64_t lsn, pending_apply_notify_fn* out_fn, void** out_ctx)` — call AFTER `apply_committed_entry()` applies `lsn`; returns 1+populates once (atomically unlinking) when a registration exists (the command path, task 3.2), 0 otherwise (the common case, facts never register); a second call for the same LSN is guaranteed to return 0; safe to call unconditionally for every applied LSN. Task 6.2 should call this right after each apply inside/after the walk and invoke the returned continuation — deliberately NOT wired into the walk itself, per boundary. **Two pre-existing tests in test_commit_barrier_reply.c now fail as EXPECTED (not a regression)**: their own assertion text literally says "no job may be constructed until task 6's apply pump exists" — now correctly invalidated by this task landing; explicitly task 7.3's job to update, per task 3.2's own Implementation Notes anticipating exactly this. **Minor, non-blocking coverage gap for a fast-follow**: no test explicitly replays a `WAL_JOB_FAIL`/`WAL_JOB_UPDATE` entry with a real `_pad` marker through the REPLAY path specifically (only through live apply) — low risk given `apply_committed_entry()` is the literal shared function both paths call, but untested for that specific combination. - **5.3**: closes task-group 5. Took 3 review rounds — the heaviest task in the entire series, and the only one that surfaced a genuine CRITICAL production safety bug (not a doc/log issue like 5.1/5.2's rounds). **Round 1 (test-only) REJECTED**: while writing tests for "the fence stays alive" and "the fence clears on a subsequent consistent recovery," review discovered the sticky `recovery_unrecoverable` flag is trigger-agnostic — it has ZERO provenance about WHY it was set. Task 5.2 correctly reused this flag for the new `apply_divergent` trigger, but TWO PRE-EXISTING, live, running-process call sites in ha.c already clear this SAME flag based purely on STRUCTURAL conditions (never checking the apply verdict): `promotion_fence_timeout_fn` (a self-re-arming periodic timer) and the promotion-eligibility gate in `raft_cb_become_leader`. **Concrete, reachable attack path**: a node's local apply dispatch has a defect (materialized state diverges from the committed log) while its WAL bytes are otherwise fine (structurally healthy — orthogonal to the apply defect by construction). The apply-divergent fence sets correctly at boot. But WHILE THE PROCESS KEEPS RUNNING, the periodic timer sees the structural conditions were always fine and silently calls `persist_clear_recovery_unrecoverable()` — releasing a fence that was never actually resolved. If this node then wins a later election, the promotion gate now sees `unrecoverable==false` and promotes it to primary — **a genuinely divergent node serving authoritative state**, a direct Req 8.4 / CLAUDE.md "never serve divergent state" violation. **Round 2 fixed the code**: added `apply_divergent = persist_apply_verdict(&c->persist) == APPLY_DIVERGENT` as an additional guard ANDed into BOTH clear-call-site conditions — `promotion_fence_timeout_fn` (line ~876) and `raft_cb_become_leader`'s gate (line ~1130) — so an apply-divergence-triggered fence can now ONLY be cleared by a fresh process restart's clean `persist_recover()` (since `apply_verdict` has exactly two production write sites: `persist_init()`'s reset-to-`APPLY_CLEAN` and `persist_recover()`'s boot-time computation — no live path resets it). Verified via full trace + RED→GREEN for site 1. **Round 2 REJECTED anyway**: the fix was logically sound, but review PROVED site 2's guard had ZERO test coverage — by deleting only that guard and re-running the ENTIRE 298-test regression sweep across 13 groups, every single test still passed. An unverified defense-in-depth guard on promotion/election-safety code doesn't meet the bar. **Round 3 (test-only) closed it**: added a test that structurally isolates site 2 from site 1 — proven not to be exercising the already-covered sticky-flag path, since `ha_promotion_gate_decision()` never populates `apply_divergent` in its own inputs, so ONLY the round-2-added standalone guard at the call site can catch this scenario. RED→GREEN independently reproduced by two separate reviewers reverting the guard and confirming the exact predicted test failure. **Final closed picture, verified end-to-end across all 3 rounds**: (a) apply-divergence sets the fence correctly (5.2); (b) survives the structural timer's periodic re-check (round 2); (c) survives the promotion gate's re-check even in the hypothetical where the sticky flag itself somehow reads false (round 3); (d) only clears via a fresh process restart replaying cleanly — confirmed by round 1's own two-separate-`persist_t`-instances test. **Forward note for task 6, already flagged in-repo (test_apply_correctness_verification.c comments), not a defect today**: this entire safety picture rests on `apply_verdict` only ever being recomputed once, at boot. If task 6's live apply pump ever adds a live re-verification of apply-correctness (not currently planned, but worth checking), the "restart-only" assumption these 3 rounds of guards depend on would need re-examination at that time. **Lesson for the whole series**: reusing an existing mechanism correctly at the logic/decision level (5.2's `apply_divergent` short-circuit) does not automatically make every OTHER consumer of that same mechanism's output (ha.c's two independent, pre-existing live-clear call sites, written before this feature existed and with no reason to have considered a future trigger) still behave safely — a shared, trigger-agnostic sticky flag is exactly the kind of implicit coupling that needs an explicit audit of ALL its readers/writers whenever a new trigger is added, not just the one you're actively wiring. - **5.2**: routed `APPLY_DIVERGENT` (task 5.1) into the EXISTING loud-fence mechanism as a new INPUT, not a new fencing path — added `bool apply_divergent` to the existing `recover_policy_inputs_t`; `persist_recover_decision()` now unconditionally short-circuits to the pre-existing `RECOVER_DECISION_FENCE_UNRECOVERABLE` enum value when it's true (correct and design-mandated: a divergent local materialization can't be trusted for a peer/catch-up comparison either, so it must always fence regardless of peer state); reuses the same sticky `recovery_unrecoverable` flag and the same `persist_recover_loud_fence_if_unrecoverable()` function — confirmed by both review rounds to be the identical pre-existing symbols, zero new fence mechanism. Confirmed the existing promotion-gate consultation in ha.c (~line 1078, `persist_recovery_unrecoverable(&c->persist)`) already independently reads this flag — zero ha.c changes needed. Zero new self-termination paths (verified by grep both rounds) — stays alive, keeps retrying, per the CLAUDE.md process-lifecycle invariant. Took 2 review rounds. **Round 1 REJECTED — real, test-reproduced defect, not documentation**: the operator-facing ERROR() log's fixed boilerplate clause ("no healthy peer reachable to catch up; durable_floor_lsn=X < cluster committed high-water hwm=Y") fired UNCONDITIONALLY for every fence decision, including the new apply_divergent trigger — but for that trigger the fields are either hardcoded placeholders or explicitly-set values that made the clause LITERALLY FALSE in test output ("100 < 100", "2 < 0" — numerically impossible claims presented as fact to an operator during a real incident). Before this task, `RECOVER_DECISION_FENCE_UNRECOVERABLE` was only reachable via the structural "no healthy peer" branch, so the clause was truthful by construction; the new unconditional short-circuit broke that invariant. This directly undermines Req 8.3's "loud, operator-visible" requirement — a log an operator can't trust during an incident is worse than no log. **Round 2 fixed** by branching the ERROR() call itself (not just the inner reason string) on `apply_divergent` — that branch now states only the real, verified reason with no peer/HWM claim; the structural-fence branch is byte-identical to before. New regression test uses a genuine (non-mocked) stdout-capture mechanism relying on `controller_unit_test` never calling `tlog_init()` (verified true across all ~85 files in the test binary — confirmed the vendored tinylog fallback path used is a real raw `write(STDOUT_FILENO,...)`, not a stub) — both review rounds independently reverted the fix and reproduced the exact false-clause failure before re-confirming GREEN. **Lesson, consistent with task 2.2's earlier pattern**: reusing an existing decision path correctly at the LOGIC level doesn't automatically make every SIDE EFFECT of that path (like a log message assuming context from the old, narrower set of triggers) still valid for a newly-added trigger — check accompanying log/error text for assumptions that were true only because of the old, narrower reachability, not just the returned enum/decision value itself. - **5.1**: added a new `apply_verdict_t` (`APPLY_CLEAN`/`APPLY_DIVERGENT`) computed alongside the EXISTING, untouched structural `recover_verdict_t` (CRC) verdict — a dual-check design: (1) an entity-count check comparing `WAL_JOB_SUBMIT` entries scanned during replay against the live `c->jobmgr` table's actual growth (`count_jobmgr_entries()`), genuinely independent since it queries live state rather than trusting `apply_committed_entry()`'s return code — verified real by both review rounds via a forced job-count-mismatch test; (2) a position check comparing an "expected" LSN against `p->last_applied_lsn` (new `_Atomic uint64_t` field, per design.md, forward-looking infrastructure for task 6's live pump). Took 2 review rounds. **Round 1 REJECTED — real, live-today bug, not documentation**: the position check's "expected" value was computed as `recovered_tip = current_lsn - 1` (the PHYSICAL WAL-scan tip, which `replay_wal()` advances for every entry including `WAL_CHECKPOINT` markers and unknown types) instead of the LOGICAL apply cursor. `persist_checkpoint()` durably fsyncs the checkpoint marker in one step, then persists the advanced `checkpoint_lsn` to the on-disk snapshot in a LATER step — a crash between those two steps (a real, unavoidable crash-safety window this codebase already engineers hard for, cf. bug-661/ha-commit-durability) leaves a trailing checkpoint marker as the highest-LSN entry on next boot, causing `recovered_tip` to exceed the true `last_applied_lsn` and falsely report `APPLY_DIVERGENT` on a perfectly clean recovery — a direct Req 8.2 violation, reachable TODAY on any node that crashes at exactly the wrong moment during a checkpoint. **Round 2 fixed** by reading `p->last_applied_lsn` directly (the value `replay_wal()` already correctly advances only for entries actually dispatched through `apply_committed_entry()`, skipping checkpoint-marker/unknown-type entries) instead of `recovered_tip`; `recovered_tip` itself is untouched and still correctly feeds `durable_lsn`, which does need the physical tip. New regression test genuinely reproduces the trigger condition (real `persist_checkpoint()`, then `unlink()`s the exact snapshot file it wrote, landing on-disk state in the identical place a real crash would) — both review rounds independently reverted the fix and reproduced the false-DIVERGENT failure before re-confirming GREEN. **Known, disclosed, accepted scope gap (not a defect, flagged for whoever extends this check)**: the entity-count check covers jobs only. Queues are correctly excluded (config-derived, `WAL_QUEUE_CTRL` never creates a new one, verified by both review rounds). **Workers are NOT covered but CAN diverge**: `worker_from_capnp()`/`restore_worker()` genuinely creates a new worker on a `WAL_WORKER_CTRL`/`HOOK_DETAIL_WORKER_JOIN` entry when absent from `workermgr` — a worker-creation divergence is currently invisible to this verification. Extend `persist_verify_apply_correctness()` with a worker-count check (mirroring `count_jobmgr_entries()`) if/when this needs to be closed. **Self-disclosed, accepted limitation**: because `replay_wal()`'s single-threaded caller in `persist_recover()` derives both the "expected" and "actual" position values from the same atomic within moments of each other on the same thread, the position check is diagnostically tautological FOR THIS SPECIFIC CALL SITE today (though the entity-count half remains genuinely independent) — this will change once task 6's live apply pump gives expectation and actual state a real chance to diverge across threads/time; task 5.1's boundary does not require solving that now. diff --git a/src/controller/commit.c b/src/controller/commit.c index 02788024..237ae59c 100644 --- a/src/controller/commit.c +++ b/src/controller/commit.c @@ -815,6 +815,23 @@ static void commit_resolve_pending(commit_engine_t* ce) for (int i = 0; i < n; i++) { commit_resolve(ce, ready[i], COMMIT_OUTCOME_COMMITTED, 0); } + + /* raft-apply-decoupling task 6.1: this function is the single choke + * point every one of the six committed_lsn-advancing paths already + * funnels through (durable-advance, self-quorum commit_submit, ack- + * driven advance, the writable-lsn gate, the raft-library commit-index + * feed via commit_set_raft_commit_index, and membership change) -- + * kicking the ordered apply pump here, rather than only from ha.c's + * on_commit_advanced callback as design.md's wiring text names, is a + * deliberate deviation: on_commit_advanced is fed by exactly ONE of + * these six paths, so wiring the pump there alone would silently starve + * it for any cluster (e.g. single-node/self-quorum) that never + * exercises that specific raft-library callback. Called unconditionally + * -- persist_apply_pump_kick is itself a documented no-op whenever + * ce->p->cluster is NULL, so this has zero effect on any commit-engine + * unit fixture that never wires a cluster_t (the overwhelming majority + * of this file's own test suite). */ + persist_apply_pump_kick(ce->p, ce->pool); } /* raft-apply-decoupling task 1.4: the FAILED-outcome sibling of diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 73cc3e08..0b3f7809 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -136,6 +136,17 @@ extern job_transition_result_t apply_job_fail_transition(job_t*, job_failinfo_t*); extern job_transition_result_t apply_job_update_transition(job_t*, int); +/* Guarded state-transition core for a postdone fact (raft-apply-decoupling + * task 6.1 round 2, Req 3.3/7.1-7.3), extracted from on_job_postdone's old + * inline mutation. Unlike a generic UPDATE, postdone's formula (conditional + * EXIT backstop, then OR in JOB_STAT_PDONE) is a pure function of the job's + * CURRENT state — so, like the other cores above, it is meant to be invoked + * at apply time against whatever state actually holds then, never against a + * receipt-time snapshot. No-op (JOB_TRANSITION_NOOP_DUPLICATE) when PDONE is + * already set and no backstop is needed. Caller must hold `job` under + * RTTI_LOCK_WRITE. */ +extern job_transition_result_t apply_job_postdone_transition(job_t*); + extern int on_job_start(cluster_t*, job_pidinfo_t*); extern int on_job_finish(cluster_t*, xce_jobfinish_t*); extern int on_job_fail(cluster_t*, job_failinfo_t*); diff --git a/src/controller/job.c b/src/controller/job.c index d0550fe9..ce1368a0 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -17,7 +17,6 @@ static int record_job_start_fact(cluster_t*, job_t*, job_pidinfo_t*); static int record_job_finish_fact(cluster_t*, job_t*, xce_jobfinish_t*); static int record_job_fail_fact(cluster_t*, job_t*, job_failinfo_t*); -static int record_job_update_fact(cluster_t*, job_t*, int); static int mark_job_fail(cluster_t*, job_t*, job_failinfo_t*); static int mark_job_update(cluster_t*, job_t*, int); @@ -1157,29 +1156,37 @@ static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo) * peek/rejected branch here — BEFORE_EVENT fires unconditionally, exactly as * mark_job_fail's always does today. * - * The resultant state depends on the CURRENT job state (PERR merges onto an - * existing DONE/EXIT terminal state rather than resetting it — job.c's - * pre-existing apply_job_fail_transition switch). That computation is - * reproduced here on a throwaway shallow copy of `job` — apply_job_fail_ - * transition only ever touches `job->job.state`/`job->job.finish_time` - * (scalar fields, never pidinfo/exit_info), so a shallow copy is safe to - * discard afterward with no cleanup — rather than duplicating the switch, - * so this can never drift from what apply time will eventually compute. */ + * task 6.1 round 2 fix (merge-staleness corruption, Req 11.3): the resultant + * state depends on the CURRENT job state (PERR merges onto an existing + * DONE/EXIT terminal state rather than resetting it — apply_job_fail_ + * transition's switch below). Earlier this was precomputed HERE, at receipt + * time, on a throwaway shallow copy of `job`, and the precomputed absolute + * result was baked into the WAL entry. That is unsafe under deferred apply: + * another fact for the same job (e.g. a FINISH) can still be pending apply + * when this FAIL is received, so `job->job.state` at receipt time may not + * be the state this FAIL's merge should actually be computed against — + * baking a stale snapshot silently clobbers whatever the other fact applies + * first (bug found in task 6.1 round 1 review). Fixed by carrying ONLY + * `failinfo->stage` (via _pad's WAL_JOB_FAIL_STAGE_FLAG — see wal_types.h's + * doc comment on wal_jobstate_t._pad) and deferring the actual merge + * computation to apply time, against the job's real current state then + * (apply_committed_entry's WAL_JOB_FAIL case, persist.c). */ static int record_job_fail_fact(cluster_t* c, job_t* job, job_failinfo_t* failinfo) { - job_t scratch; - xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_FAIL, c, failinfo, job); - scratch = *job; - apply_job_fail_transition(&scratch, failinfo); - /* Durable append only — no mutation of the real `job`, no AFTER_EVENT - * firing here (Req 5.1, 5.3). */ - (void)persist_submit_job_fact(&c->persist, WAL_JOB_FAIL, job->job.jobid, - (int)scratch.job.state); + * firing here (Req 5.1, 5.3). `state` is unused for this apply-time- + * recomputed shape (0 is written for forward-compatibility/diagnostics + * only); the WAL_JOB_FAIL_STAGE_FLAG bit in `_pad` tells + * apply_committed_entry() to route it to the real + * apply_job_fail_transition() core instead of an absolute overwrite. */ + (void)persist_submit_job_fact_ex(&c->persist, WAL_JOB_FAIL, + job->job.jobid, 0, + WAL_JOB_FAIL_STAGE_FLAG | + failinfo->stage); return 0; } @@ -1221,10 +1228,10 @@ job_transition_result_t apply_job_update_transition(job_t* job, int state) /* Used only by job_ctrl's SIGCONT/SIGSTOP suspend-resume branches (a client * command path, do_jobcontrol_op — out of raft-apply-decoupling task 4.1's - * boundary) — mutate-then-notify, unchanged. record_job_update_fact below is - * the separate, durable-append-only sibling used by the worker-fact receipt - * path (on_job_postdone); the two must stay independent since job_ctrl's - * synchronous command semantics are not this task's concern. */ + * boundary) — mutate-then-notify, unchanged. record_job_postdone_fact below + * is the separate, durable-append-only sibling used by the worker-fact + * receipt path (on_job_postdone); the two must stay independent since + * job_ctrl's synchronous command semantics are not this task's concern. */ static int mark_job_update(cluster_t* c, job_t* job, int state) { int cur = job->job.state; @@ -1252,33 +1259,68 @@ static int mark_job_update(cluster_t* c, job_t* job, int state) return 0; } -/* raft-apply-decoupling task 4.1 (Req 5.1): the worker-fact receipt sibling - * of mark_job_update above — durable-append-only, used only by - * on_job_postdone. mark_job_update itself is untouched (job_ctrl still needs - * its mutate-then-notify semantics; that command path is out of this task's - * boundary). `state` here is the caller's already-computed target state - * (on_job_postdone computes it from the job's current state before calling - * this, exactly as it always has — that computation is untouched by this - * task; only what happens with the computed value changes). */ -static int record_job_update_fact(cluster_t* c, job_t* job, int state) +/* Guarded state-transition core for a postdone fact, extracted from + * on_job_postdone's old inline mutation (raft-apply-decoupling task 6.1 + * round 2, Req 3.3/7.1-7.3). Unlike a generic UPDATE, postdone's formula + * (conditionally backstop EXIT, then OR in JOB_STAT_PDONE) is a pure + * function of the job's CURRENT state — so this core is meant to be invoked + * AT APPLY TIME, against whatever state actually holds then, never against + * a receipt-time snapshot (a receipt-time snapshot is exactly the merge- + * staleness hazard this round fixes: another fact for the same job, e.g. a + * FINISH, may still be in flight and apply AFTER receipt but BEFORE this + * fact's own apply, under deferred apply). No-op when PDONE is already set + * and no backstop is needed (idempotent re-apply, Req 3.4). Fires NO hook — + * same convention as the other apply_job_*_transition cores; hook-firing + * stays the caller's (apply_committed_entry's) responsibility. Caller must + * hold `job` under RTTI_LOCK_WRITE. + * + * Known, previously-accepted residual (task 4.1, unchanged by this round): + * finish_time is not part of the wal_jobstate_t wire format, so the old + * defensive `job->job.finish_time = time(NULL)` backstop write has no + * apply-time equivalent — a postdone arriving with no prior exit status + * recorded still gets no defensive finish_time set anywhere. */ +job_transition_result_t apply_job_postdone_transition(job_t* job) { - int cur = job->job.state; + int next = job->job.state; - /* Peek the identical guards apply_job_update_transition() re-checks - * internally — same no-observable-behavior-change rationale as - * mark_job_update's peek above. */ - if (job_state_is_backward(cur, state) || (state == cur)) { - apply_job_update_transition(job, state); /* logs + no-op, unchanged */ - return 0; + if ((next & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) { + WARN("job %ld got post exec fail notification but job's exit " + "status has not been set", + job->job.jobid); + next |= JOB_STAT_EXIT; + } + next |= JOB_STAT_PDONE; + + if (next == job->job.state) { + INFO("job %ld: duplicate postdone — no-op", job->job.jobid); + return JOB_TRANSITION_NOOP_DUPLICATE; } + job->job.state = next; + return JOB_TRANSITION_APPLIED; +} + +/* raft-apply-decoupling task 4.1/6.1 (Req 5.1, 3.3, 7.1-7.3): durable- + * append-only sibling of the old inline postdone mutation, used only by + * on_job_postdone. Replaces record_job_update_fact (removed in task 6.1 + * round 2 — this was its only caller): postdone's mutation is a pure + * function of current state (see apply_job_postdone_transition above), so + * nothing needs to be computed or peeked at receipt time — the WAL entry + * carries no state value at all, just a marker (_pad == + * WAL_JOB_UPDATE_POSTDONE_MARK) telling apply_committed_entry() to re- + * derive the merge fresh, against whatever the job's real current state is + * when this entry actually applies (fixes the same class of merge- + * staleness corruption record_job_fail_fact's fix addresses above). */ +static int record_job_postdone_fact(cluster_t* c, job_t* job) +{ xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_UPDATE, c, job); /* Durable append only — no mutation, no AFTER_EVENT firing here (Req * 5.1, 5.3). */ - (void)persist_submit_job_fact(&c->persist, WAL_JOB_UPDATE, job->job.jobid, - state); + (void)persist_submit_job_fact_ex(&c->persist, WAL_JOB_UPDATE, + job->job.jobid, 0, + WAL_JOB_UPDATE_POSTDONE_MARK); return 0; } @@ -1383,7 +1425,6 @@ int on_job_resume(cluster_t* c, uint64_t jobid) int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid) { job_t* job = NULL; - int state; TRACE("%s job post done jobid %ld", __func__, jobid); @@ -1399,29 +1440,14 @@ int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid) return -errno; } + /* raft-apply-decoupling task 6.1 round 2 (Req 5.1, 11.3): the backstop- + * EXIT-if-unset + OR-in-PDONE mutation formerly computed HERE, at + * receipt time, is now deferred entirely to apply_job_postdone_ + * transition() (called from apply_committed_entry() once this fact + * commits) — see that function's doc comment for why baking it here + * against a receipt-time snapshot was unsafe under deferred apply. */ lock(job, RTTI_LOCK_WRITE); - state = job->job.state; - if ((state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) { - WARN("job %ld got post exec fail notification but job's exit status " - "has not been set", - job->job.jobid); - state |= JOB_STAT_EXIT; - /* raft-apply-decoupling task 4.1 (Req 5.1): the backstop - * `job->job.finish_time = time(NULL)` write that used to live here - * is removed — no state mutation may happen at receipt time. - * finish_time is not part of the wal_jobstate_t wire format (only - * {jobid, state} travels — a pre-existing limitation, see - * apply_committed_entry's doc comment in persist.h), so this - * defensive backstop has no equivalent at apply time either; a - * post-done arriving with no prior exit status recorded no longer - * gets a defensive finish_time set anywhere. Low-severity residual, - * flagged in this task's status report — not fixed here (would - * require wire-format enrichment, explicitly out of this task's - * scope per persist.h's Apply Seam doc comment). */ - } - - state |= JOB_STAT_PDONE; - record_job_update_fact(c, job, state); + record_job_postdone_fact(c, job); unlock(job); dereference(job); diff --git a/src/controller/meson.build b/src/controller/meson.build index a3654f78..fd277e3e 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -162,6 +162,7 @@ controller_unit_test_src = [ 'testsuite/test_replay_apply_seam.c', 'testsuite/test_apply_reapply_idempotency.c', 'testsuite/test_apply_correctness_verification.c', + 'testsuite/test_apply_pump.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/persist.c b/src/controller/persist.c index e9bbca6b..3b15694a 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -643,6 +643,11 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) p->pending_applies = NULL; pthread_mutex_init(&p->pending_applies_lock, NULL); + /* Ordered apply pump starts idle (raft-apply-decoupling task 6.1): no + * walk scheduled or running yet. Freed/destroyed in persist_fini. */ + pthread_mutex_init(&p->apply_pump_lock, NULL); + atomic_store(&p->apply_pump_scheduled, 0); + /* Create WAL and snapshot directories */ if (mkdir_p(p->wal_path, 0700) != 0) return -errno; if (mkdir_p(p->snap_path, 0700) != 0) return -errno; @@ -766,6 +771,12 @@ void persist_fini(persist_t* p) } p->pending_applies = NULL; pthread_mutex_destroy(&p->pending_applies_lock); + + /* raft-apply-decoupling task 6.1: no walk can still be in flight by the + * time persist_fini runs (the caller owns quiescing any vtask pool + * first, same precondition every other persist_t resource here already + * assumes). */ + pthread_mutex_destroy(&p->apply_pump_lock); } /* Forward declaration — defined after checkpoint internals */ @@ -2625,6 +2636,37 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, * (routing table, the job-state-family hook-mapping deviation, and what this * function deliberately does NOT own). Not yet wired into any live call * site — replay (2.3) and the live apply pump (6.1) will call this later. */ +/* raft-apply-decoupling task 6.1 (Hazard 1 fix): thread-local "this call + * stack is inside apply_committed_entry's own post-mutation notification" + * guard. persist_job_hook_fun / persist_queue_hook_fun / persist_worker_hook_fun + * are AFTER_EVENT subscribers that (for RETAIN/fact-shaped entry types) + * durably append a NEW WAL entry every time they fire — the correct behavior + * for their ORIGINAL, still-live purpose (job_ctrl's mark_job_update/ + * mark_job_fail synchronous command path; the legacy queue/worker control + * command handlers) which mutate live state and rely on this exact hook + * firing to durably record that mutation. + * + * apply_committed_entry() now ALSO fires these same AFTER_EVENT hooks (job + * facts always as the generic HOOK_DETAIL_JOB_UPDATE, per task 2.2's + * documented wire-format deviation; queue/worker control at their native + * detail) to deliver its own post-mutation NOTIFICATION — but that entry is + * by definition already durably committed (it just got applied FROM the + * WAL), so re-appending it would silently duplicate the entry on every live + * apply. This is fixable, correctly, ONLY by provenance (which caller fired + * the hook), NOT by hook `detail` value: HOOK_DETAIL_JOB_UPDATE in + * particular is genuinely dual-origin (job_ctrl's SIGCONT/SIGSTOP path uses + * the identical detail for its own, still-legitimate live append) — an + * unconditional detail-based filter would silently break job_ctrl's suspend/ + * resume durability instead of fixing anything. A shared persist_t-wide flag + * would ALSO be wrong: it would race across threads (e.g. job_ctrl mutating + * a DIFFERENT job on the iomgr thread while this thread's apply walk is + * mid-hook for an unrelated fact) and could wrongly suppress or wrongly + * permit an append depending on scheduling. A thread-local flag, scoped + * narrowly around each xce_run_hooks() call this function makes, is race-free + * by construction: it is only ever observed by hook subscribers invoked + * SYNCHRONOUSLY on this exact call stack, from this exact function. */ +static _Thread_local bool t_apply_seam_notifying = false; + int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, const void* payload, uint32_t len) { @@ -2666,25 +2708,30 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, void* jptr = j; xlist_t jlist = { .count = 1, .caps = 1, .delta = 1, .data = &jptr }; + t_apply_seam_notifying = true; xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_NEW, c, &jlist); + t_apply_seam_notifying = false; return 0; } 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: { - /* Every job-state fact type shares one wire shape — wal_jobstate_t - * {jobid, state} — and therefore one guarded core. See persist.h's - * doc comment on apply_committed_entry for why this does not - * dispatch to the type-specific apply_job_{start,finish,fail}_ - * transition() cores (their richer parameter structs are not + /* Every job-state fact type in this group shares one wire shape — + * wal_jobstate_t {jobid, state} — and therefore one guarded core. + * Their payloads are deterministic/state-independent (a START is + * always JOB_STAT_RUN, a FINISH's outcome depends only on + * exit_code), so an absolute `state` baked at receipt time never + * goes stale — unlike WAL_JOB_FAIL/WAL_JOB_UPDATE below, which have + * their own dedicated cases. See persist.h's doc comment on + * apply_committed_entry for the full routing table and why this + * does not dispatch to the richer type-specific apply_job_{start, + * finish}_transition() cores (their parameter structs are not * present on this wire format). */ if (len < sizeof(wal_jobstate_t)) return 0; /* undersized payload — skip, mirrors apply_wal_entry */ @@ -2705,8 +2752,111 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, if (result != JOB_TRANSITION_APPLIED) return 0; /* rejected-backward / no-op-duplicate — no hook */ + /* Hazard 1 guard (see t_apply_seam_notifying's doc comment above): + * HOOK_DETAIL_JOB_UPDATE is dual-origin — job_ctrl's mark_job_update + * (SIGCONT/SIGSTOP) fires the identical detail for its own live, + * still-legitimate append via persist_job_hook_fun. Without this + * guard, EVERY fact this apply pump applies would re-append itself + * as a duplicate, mis-typed WAL_JOB_UPDATE entry the moment the live + * pump (task 6.1) starts calling this function. */ + t_apply_seam_notifying = true; xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_UPDATE, c, j); + t_apply_seam_notifying = false; + return 0; + } + + case WAL_JOB_FAIL: { + /* task 6.1 round 2 (Req 3.3/7.1-7.3, 11.3) fix for a job-state + * corruption bug: unlike the deterministic group above, a FAIL's + * resulting state depends on the job's CURRENT state (PERR merges + * onto whatever terminal state already holds). Dispatch on _pad + * (see wal_types.h's doc comment on wal_jobstate_t._pad): + * - WAL_JOB_FAIL_STAGE_FLAG set: a deferred fail-fact + * (record_job_fail_fact, job.c) — reconstruct a minimal + * job_failinfo_t from {jobid, stage} and call the REAL + * apply_job_fail_transition() against the job's ACTUAL CURRENT + * state at apply time, never a receipt-time snapshot. + * - _pad == 0: the legacy live path (job_ctrl's mark_job_fail) — + * `state` is already the correct, current, absolute value, + * computed synchronously under the same lock as the mutation + * (no staleness window) — an absolute overwrite is safe and + * unchanged. */ + if (len < sizeof(wal_jobstate_t)) + return 0; /* undersized payload — skip, mirrors apply_wal_entry */ + + const wal_jobstate_t* js = (const wal_jobstate_t*)payload; + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)js->jobid, (void**)&j); + unlock(c->jobmgr); + if (!j) + return -ENOENT; + + lock(j, RTTI_LOCK_WRITE); + job_transition_result_t result; + if (js->_pad & WAL_JOB_FAIL_STAGE_FLAG) { + job_failinfo_t failinfo = { + .jobid = js->jobid, + .stage = js->_pad & ~WAL_JOB_FAIL_STAGE_FLAG, + .code = 0, + .detail = NULL, + }; + result = apply_job_fail_transition(j, &failinfo); + } else { + result = apply_job_update_transition(j, (int)js->state); + } + unlock(j); + + if (result != JOB_TRANSITION_APPLIED) + return 0; /* rejected-backward / no-op-duplicate — no hook */ + + /* Hazard 1 guard — see the identical guard above. */ + t_apply_seam_notifying = true; + xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_UPDATE, c, j); + t_apply_seam_notifying = false; + return 0; + } + + case WAL_JOB_UPDATE: { + /* task 6.1 round 2 (Req 3.3/7.1-7.3, 11.3): same class of fix as + * WAL_JOB_FAIL above. Dispatch on _pad: + * - WAL_JOB_UPDATE_POSTDONE_MARK: a deferred postdone fact + * (record_job_postdone_fact, job.c) — call + * apply_job_postdone_transition(), which re-derives the + * postdone merge (conditional EXIT backstop + OR JOB_STAT_PDONE) + * fresh against the job's ACTUAL CURRENT state at apply time. + * - _pad == 0: the legacy live path (job_ctrl's mark_job_update, + * SIGCONT/SIGSTOP) — `state` is already correct/absolute, + * computed synchronously under the same lock as the mutation — + * an absolute overwrite is safe and unchanged. */ + if (len < sizeof(wal_jobstate_t)) + return 0; /* undersized payload — skip, mirrors apply_wal_entry */ + + const wal_jobstate_t* js = (const wal_jobstate_t*)payload; + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)js->jobid, (void**)&j); + unlock(c->jobmgr); + if (!j) + return -ENOENT; + + lock(j, RTTI_LOCK_WRITE); + job_transition_result_t result = + (js->_pad == WAL_JOB_UPDATE_POSTDONE_MARK) + ? apply_job_postdone_transition(j) + : apply_job_update_transition(j, (int)js->state); + unlock(j); + + if (result != JOB_TRANSITION_APPLIED) + return 0; /* rejected-backward / no-op-duplicate — no hook */ + + /* Hazard 1 guard — see the identical guard above. */ + t_apply_seam_notifying = true; + xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_UPDATE, c, j); + t_apply_seam_notifying = false; return 0; } @@ -2732,8 +2882,17 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, if (!q) return 0; + /* Same double-append class as the job-fact guard above: the legacy + * queue-control command handler still mutates then relies on this + * exact AFTER_EVENT firing to durably append via + * persist_queue_hook_fun (WAL_QUEUE_CTRL is unconditionally + * appended on AFTER_EVENT there, not gated on WAL_MODE_WAIT like the + * job-submit case is) — guard so this live-pump re-notify does not + * re-append an already-committed entry. */ + t_apply_seam_notifying = true; xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE, HOOK_AFTER_EVENT, HOOK_DETAIL_QUEUE_UPDATE, c, q); + t_apply_seam_notifying = false; return 0; } @@ -2772,6 +2931,13 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, if (!w) return 0; /* defensive: should not happen after a successful restore */ + /* Same double-append class as the job-fact/queue guards above: + * persist_worker_hook_fun unconditionally appends on AFTER_EVENT for + * every worker detail (no WAL_MODE_WAIT gate), and the legacy worker + * join/update/dead command paths still rely on that firing for their + * own live durability — guard the whole re-notify switch so this + * live-pump call never re-appends an already-committed entry. */ + t_apply_seam_notifying = true; switch (hook_detail) { case HOOK_DETAIL_WORKER_JOIN: xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER, @@ -2790,6 +2956,7 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, break; /* unrecognized discriminator — mutation already applied, * notification skipped rather than guessed at */ } + t_apply_seam_notifying = false; return 0; } @@ -2801,6 +2968,292 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, } } +/* ============================================================ + * Ordered apply pump (raft-apply-decoupling task 6.1) + * ============================================================ */ + +/* Locate and read back the WAL entry at exactly `lsn` from the on-disk + * segment files, for the live apply pump to hand to apply_committed_entry() + * (mirrors read_entry_meta_locked's segment-scan idiom above, but returns + * the entry's type and a malloc'd copy of its payload instead of just + * metadata, and re-verifies the entry CRC — this is live-path plumbing, not + * a metadata probe). + * + * Does not take p->lock: every LSN this is ever called with is already + * majority-committed (durable and replicated), strictly below the physical + * append tip a concurrent persist_wal_async could be extending — so this + * never races the single-writer append path for the bytes it actually + * reads. + * + * Returns 0 and populates *out_type/*out_payload (malloc'd if payload_len > + * 0, caller frees; NULL when payload_len is 0)/*out_len on success. + * Returns -ENOENT if no segment contains this LSN (should not happen for a + * genuinely committed entry) or a negative errno on a read/CRC failure. */ +static int apply_pump_read_entry(persist_t* p, uint64_t lsn, + uint32_t* out_type, uint8_t** out_payload, + uint32_t* out_len) +{ + if (!p || lsn == 0 || !out_type || !out_payload || !out_len) + return -EINVAL; + + DIR* d = opendir(p->wal_path); + if (!d) + return -ENOENT; + + uint64_t seg_lsns[4096]; + int nseg = 0; + struct dirent* e; + while ((e = readdir(d)) != NULL && + nseg < (int)(sizeof(seg_lsns) / sizeof(seg_lsns[0]))) { + if (strncmp(e->d_name, "seg-", 4) != 0) + continue; + uint64_t sl = 0; + if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &sl) != 1) + continue; + seg_lsns[nseg++] = sl; + } + closedir(d); + if (nseg == 0) + return -ENOENT; + + /* Ascending sort by start LSN (segment count is tiny in practice). */ + for (int i = 1; i < nseg; i++) { + uint64_t key = seg_lsns[i]; + int j = i - 1; + while (j >= 0 && seg_lsns[j] > key) { + seg_lsns[j + 1] = seg_lsns[j]; + j--; + } + seg_lsns[j + 1] = key; + } + + int rc = -ENOENT; + for (int si = 0; si < nseg; si++) { + /* Segments are LSN-ordered; one whose start LSN already exceeds the + * target cannot contain it, nor can any later one. */ + if (seg_lsns[si] > lsn) + break; + + char seg_path[4096]; + snprintf(seg_path, sizeof(seg_path), WAL_SEG_FMT, p->wal_path, + seg_lsns[si]); + int fd = open(seg_path, O_RDONLY); + if (fd < 0) + continue; + + uint32_t off = 0; + for (;;) { + wal_entry_hdr_t hdr; + ssize_t hn = pread(fd, &hdr, sizeof(hdr), (off_t)off); + if (hn < (ssize_t)sizeof(hdr)) + break; /* short read / EOF / pre-allocated zero tail */ + if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) + break; + + if (hdr.lsn == lsn) { + uint8_t* payload = NULL; + uint32_t expected_crc = + crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE); + + if (hdr.payload_len > 0) { + payload = malloc(hdr.payload_len); + if (!payload) { + rc = -ENOMEM; + break; + } + ssize_t pn = pread(fd, payload, hdr.payload_len, + (off_t)(off + sizeof(hdr))); + if (pn != (ssize_t)hdr.payload_len) { + free(payload); + rc = -EIO; + break; + } + expected_crc = crc32(expected_crc, (const Bytef*)payload, + hdr.payload_len); + } + + if (hdr.crc != expected_crc) { + free(payload); + rc = -EIO; + break; + } + + *out_type = hdr.type; + *out_payload = payload; + *out_len = hdr.payload_len; + rc = 0; + break; + } + + off += (uint32_t)(sizeof(hdr) + hdr.payload_len); + } + close(fd); + + if (rc == 0 || rc == -ENOMEM || rc == -EIO) + break; /* found it, or hit a genuine error inside the segment + * that should contain it */ + } + + return rc; +} + +int persist_apply_pump_run(persist_t* p) +{ + if (!p) + return -EINVAL; + + pthread_mutex_lock(&p->apply_pump_lock); + + int applied = 0; + for (;;) { + uint64_t committed = atomic_load_explicit(&p->committed_lsn, + memory_order_acquire); + uint64_t next = atomic_load_explicit(&p->last_applied_lsn, + memory_order_relaxed) + 1; + if (next > committed) + break; /* caught up as of this instant */ + + uint32_t wal_type = 0; + uint8_t* payload = NULL; + uint32_t len = 0; + int rc = apply_pump_read_entry(p, next, &wal_type, &payload, &len); + if (rc < 0) { + ERROR("persist_apply_pump_run: failed to read committed entry " + "lsn=%" PRIu64 " (rc=%d) -- stopping this pass; the entry " + "stays committed and will be retried, from the same " + "position, on the next kick", + next, rc); + pthread_mutex_unlock(&p->apply_pump_lock); + return rc; + } + + /* Strict order, no skipping or reordering: this is the single + * dispatch call every newly committed entry passes through here, + * regardless of its commit mode (WAIT command or RETAIN fact) — the + * loop only ever advances last_applied_lsn one entry at a time, in + * ascending order, so a burst of commands and facts that committed + * out of their original submission order still applies in strict + * commit (LSN) order (Req 3.3, 7.3). */ + (void)apply_committed_entry(p, wal_type, next, payload, len); + free(payload); + + atomic_store_explicit(&p->last_applied_lsn, next, memory_order_relaxed); + applied++; + } + + pthread_mutex_unlock(&p->apply_pump_lock); + return applied; +} + +/* Heap payload for one dispatched apply-pump walk invocation; frees itself + * after use, mirroring commit.c's commit_resolve_dispatch_t convention. */ +typedef struct { + persist_t* p; +} apply_pump_dispatch_t; + +/* Run walk passes until genuinely caught up, using the same recheck-then- + * clear idiom commit.c's sweep thread uses (task 1.3) to avoid a lost-wakeup + * race between "persist_apply_pump_run just decided it is caught up" and "a + * new commit lands right after". Runs on a pool worker thread (or inline as + * persist_apply_pump_kick's documented fallback) — never on the calling + * (potentially replication-ACK/iomgr) thread when dispatched normally. + * + * Does NOT immediately retry a failed pass: persist_apply_pump_run() already + * logs loudly and leaves last_applied_lsn unadvanced past a read failure, so + * an immediate self-retry here would busy-loop forever on the exact same + * unresolved head-of-line entry (a real, reproduced hazard — a persist_t + * whose committed_lsn is ever advanced without a corresponding real WAL + * entry on disk, e.g. a test fixture exercising commit-engine resolve logic + * in isolation, spun this into a multi-million-iteration tight loop before + * this guard existed). A future, genuinely NEW kick — triggered by a later + * real commit-advance event — retries then, at that event's natural cadence, + * never in a busy spin. */ +static void apply_pump_drain(persist_t* p) +{ + for (;;) { + int rc = persist_apply_pump_run(p); + + atomic_store_explicit(&p->apply_pump_scheduled, 0, memory_order_release); + + if (rc < 0) + return; /* already logged loudly inside persist_apply_pump_run; + * do not busy-retry the same unresolved entry */ + + uint64_t committed = atomic_load_explicit(&p->committed_lsn, + memory_order_acquire); + uint64_t next = atomic_load_explicit(&p->last_applied_lsn, + memory_order_relaxed) + 1; + if (next > committed) + return; /* genuinely caught up: no race window survived open */ + + int expected = 0; + if (!atomic_compare_exchange_strong_explicit( + &p->apply_pump_scheduled, &expected, 1, memory_order_acq_rel, + memory_order_relaxed)) + return; /* a concurrent kick already reclaimed the flag; it will + * run its own pass and pick up this same new work */ + /* else: we reclaimed the flag ourselves -- loop and run another pass */ + } +} + +static int apply_pump_vtask_entry(vtask_t* task) +{ + apply_pump_dispatch_t* d = (apply_pump_dispatch_t*)task->data->u.generic; + + apply_pump_drain(d->p); + free(d); + return 0; +} + +void persist_apply_pump_kick(persist_t* p, void* pool_arg) +{ + if (!p || !p->cluster) + return; + + int expected = 0; + if (!atomic_compare_exchange_strong_explicit( + &p->apply_pump_scheduled, &expected, 1, memory_order_acq_rel, + memory_order_relaxed)) + return; /* already scheduled/running -- it will pick up any new work */ + + vtaskmgr_t* pool = (vtaskmgr_t*)pool_arg; + vtask_data_t* data = calloc(1, sizeof(*data)); + vtask_t* task = vtask_alloc(); + apply_pump_dispatch_t* d = calloc(1, sizeof(*d)); + + /* Mirrors commit.c's dispatch_continuation fallback exactly: never leave + * a scheduled walk silently unrun. pool == NULL is the expected shape of + * a commit-only / bare-persist-cluster unit fixture that never calls + * commit_set_pool — a real cluster always wires the pool before any + * commit activity (cluster_init_phase2), so this fallback path is the + * normal, deterministic behavior for that class of test, not a rare + * production condition. */ + if (pool == NULL || data == NULL || task == NULL || d == NULL) { + free(data); + free(task); + free(d); + apply_pump_drain(p); + return; + } + + d->p = p; + data->vtype = VTASK_GENERIC; + data->u.generic = d; + vtask_init(task, (vtask_func_t)apply_pump_vtask_entry, data); + + /* VTASK_LIVE_BLOCK is never admission-gated (same delivery guarantee as + * commit.c's own continuation dispatch) — this cannot fast-reject the + * way VTASK_REQ_BLOCK could. */ + if (vtask_enq_classified(pool, task, VTASK_LIVE_BLOCK) < 0) { + WARN("persist_apply_pump_kick: vtask enqueue failed -- running " + "inline (stay alive, never leave a scheduled walk unrun)"); + free(data); + free(task); + free(d); + apply_pump_drain(p); + } +} + /* Read WAL segment files in LSN order; replay entries with lsn > from_lsn. * * Segment filename format: seg-.wal @@ -4172,22 +4625,77 @@ int persist_has_pending_apply(persist_t* p, uint64_t lsn) return found; } +int persist_take_pending_apply(persist_t* p, uint64_t lsn, + pending_apply_notify_fn* out_fn, void** out_ctx) +{ + pending_command_apply_t* prev = NULL; + pending_command_apply_t* n; + int found = 0; + + if (p == NULL) + return 0; + + pthread_mutex_lock(&p->pending_applies_lock); + for (n = (pending_command_apply_t*)p->pending_applies; n != NULL; + n = n->next) { + if (n->lsn == lsn) { + if (prev == NULL) + p->pending_applies = n->next; + else + prev->next = n->next; + if (out_fn) + *out_fn = n->on_applied; + if (out_ctx) + *out_ctx = n->ctx; + found = 1; + break; + } + prev = n; + } + pthread_mutex_unlock(&p->pending_applies_lock); + + free(n); /* NULL when not found -- free(NULL) is a no-op */ + + return found; +} + /* raft-apply-decoupling task 4.1 (Req 5.1): the receipt-time durable-append * primitive for a worker-reported job-state fact. Unlike persist_job_hook_fun * below (an AFTER_EVENT hook that fires only once a mutation has already * happened), this is a direct, hook-fan-out-free call: job.c's fact receipt * handlers (record_job_start_fact/record_job_finish_fact/record_job_fail_fact/ - * record_job_update_fact) call this INSTEAD of mutating and firing - * AFTER_EVENT, so no notification (scheduler dispatch, slot release, - * apiserver push) can ever fire for an uncommitted fact (Req 5.3). `state` - * is always WAL_MODE_RETAIN for every job-state fact type (verified against - * wal_classify's table) — hardcoded here rather than routed through - * wal_classify since none of these types are detail-disambiguated the way - * WAL_WORKER_CTRL is. */ + * record_job_postdone_fact) call this (or persist_submit_job_fact_ex below) + * INSTEAD of mutating and firing AFTER_EVENT, so no notification (scheduler + * dispatch, slot release, apiserver push) can ever fire for an uncommitted + * fact (Req 5.3). `state` is always WAL_MODE_RETAIN for every job-state fact + * type (verified against wal_classify's table) — hardcoded here rather than + * routed through wal_classify since none of these types are detail- + * disambiguated the way WAL_WORKER_CTRL is. + * + * Thin wrapper over persist_submit_job_fact_ex() with pad = 0 — the "legacy + * absolute state" discriminator (see wal_types.h's doc comment on + * wal_jobstate_t._pad). Unchanged contract/signature for every existing + * caller. */ int persist_submit_job_fact(persist_t* p, uint32_t wal_type, uint64_t jobid, int state) { - wal_jobstate_t js = { .jobid = jobid, .state = (uint32_t)state }; + return persist_submit_job_fact_ex(p, wal_type, jobid, state, 0); +} + +/* raft-apply-decoupling task 6.1 round 2 (Req 3.3/7.1-7.3): full-control + * sibling of persist_submit_job_fact() above, letting the caller set + * wal_jobstate_t's `_pad` discriminator explicitly. Used by + * record_job_fail_fact (WAL_JOB_FAIL_STAGE_FLAG | stage) and + * record_job_postdone_fact (WAL_JOB_UPDATE_POSTDONE_MARK) to mark a fact + * whose real mutation must be re-derived at apply time against the job's + * CURRENT state, rather than trusting a receipt-time snapshot baked into + * `state` — see wal_types.h's doc comment on wal_jobstate_t._pad for the + * full rationale. */ +int persist_submit_job_fact_ex(persist_t* p, uint32_t wal_type, + uint64_t jobid, int state, uint32_t pad) +{ + wal_jobstate_t js = { .jobid = jobid, .state = (uint32_t)state, + ._pad = pad }; return wal_hook_append_classified(p, wal_type, WAL_MODE_RETAIN, &js, sizeof(js)); @@ -4205,6 +4713,17 @@ int persist_job_hook_fun(int type, int situation, int detail, if (situation != HOOK_AFTER_EVENT) return 0; + /* raft-apply-decoupling task 6.1 (Hazard 1): this AFTER_EVENT firing + * originated from apply_committed_entry()'s own post-mutation notify — + * the entry is already durably committed (that is why it is being + * applied at all), so appending it again here would duplicate it. See + * t_apply_seam_notifying's doc comment (above apply_committed_entry) for + * why this MUST be a call-provenance guard, not a `detail` filter: + * job_ctrl's mark_job_update/mark_job_fail still fire this exact same + * hook for their own live, still-legitimate append. */ + if (t_apply_seam_notifying) + return 0; + persist_t* p = cls; uint32_t wal_type = detail_to_wal_type(detail); wal_commit_mode_t mode; @@ -4272,6 +4791,17 @@ int persist_queue_hook_fun(int type, int situation, int detail, if (situation != HOOK_AFTER_EVENT) return 0; + /* raft-apply-decoupling task 6.1 (Hazard 1, queue analog): see + * persist_job_hook_fun's identical guard above for the full rationale — + * apply_committed_entry()'s WAL_QUEUE_CTRL branch fires this same + * AFTER_EVENT to notify, and this hook appends unconditionally + * (WAL_QUEUE_CTRL is WAL_MODE_WAIT with no early-return here), so + * without this guard a live-applied queue entry would re-append and + * re-trigger a nested majority-commit wait every time the pump walks + * it. */ + if (t_apply_seam_notifying) + return 0; + persist_t* p = cls; uint32_t wal_type = detail_to_wal_type(detail); wal_commit_mode_t mode; @@ -4310,6 +4840,11 @@ int persist_worker_hook_fun(int type, int situation, int detail, if (situation != HOOK_AFTER_EVENT) return 0; + /* raft-apply-decoupling task 6.1 (Hazard 1, worker analog): see + * persist_job_hook_fun's identical guard above for the full rationale. */ + if (t_apply_seam_notifying) + return 0; + persist_t* p = cls; uint32_t wal_type = detail_to_wal_type(detail); wal_commit_mode_t mode; diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index d86976ff..69c7aa37 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -82,6 +82,7 @@ extern void test_apply_committed_entry(void); extern void test_replay_apply_seam(void); extern void test_apply_reapply_idempotency(void); extern void test_apply_correctness_verification(void); +extern void test_apply_pump(void); void setUp(void) { return; } @@ -233,6 +234,7 @@ int main(int argc, char* argv[]) TEST(test_replay_apply_seam); TEST(test_apply_reapply_idempotency); TEST(test_apply_correctness_verification); + TEST(test_apply_pump); return UNITY_END(); } diff --git a/src/controller/testsuite/test_apply_pump.c b/src/controller/testsuite/test_apply_pump.c new file mode 100644 index 00000000..4616d94e --- /dev/null +++ b/src/controller/testsuite/test_apply_pump.c @@ -0,0 +1,715 @@ +/* test_apply_pump.c — raft-apply-decoupling task 6.1 + * + * Proves the ordered apply pump: the strict-order walk that drives + * apply_committed_entry() (task 2.2) off live commit-advance for the very + * first time — before this task, only restart replay (task 2.3) ever called + * it. Covers: + * + * 1. The task's own named observable: a burst of entries appended in one + * order, whose majority commit becomes known all at once (a single + * covering ACK advancing committed_lsn past all of them in one jump), + * is applied in strict ascending LSN order — never reordered, never + * skipped — regardless of commit mode (a WAL_JOB_SUBMIT WAIT-mode entry + * mixed with WAL_JOB_{START,FINISH} RETAIN-mode facts in the same + * burst). Also folds in the Hazard 1 (double-append) regression check: + * current_lsn must not grow beyond the entries this test itself + * appended, proving the live pump's own post-mutation notification + * never re-appends an already-committed entry. + * 2. persist_apply_pump_run() called directly (white-box), confirming the + * walk applies exactly the committed-but-unapplied range and advances + * last_applied_lsn accordingly. + * 3. A non-regression check for the Hazard 1 fix's precision: job_ctrl's + * SIGCONT/SIGSTOP path (mark_job_update) still durably appends its own + * WAL_JOB_UPDATE entry — proving the double-append guard is scoped by + * call provenance (apply-seam-only), not by hook detail value, since + * job_ctrl fires the identical HOOK_DETAIL_JOB_UPDATE detail for its + * own, still-legitimate live append. + * + * Fixture mirrors test_commit_barrier_reply.c's cbrp_setup (real persist_init + * + persist_start over a private tmpdir so persist_start registers the + * production hooks, a real 3-member commit_engine_t with two stub standbys, + * ACKs delivered explicitly via repl_on_ack), trimmed of the worker-join + * scaffolding this file does not need. commit_set_pool is deliberately never + * called, so ce->pool stays NULL — persist_apply_pump_kick's documented + * NULL-pool fallback then runs the walk INLINE, synchronously, on whichever + * thread's kick wins the CAS-coalescing race. + * + * IMPORTANT (round-2 review correction): that does NOT mean every assertion + * below is directly observable immediately after a triggering call returns + * with no polling needed — persist_start's fsync_worker thread independently + * drives its own kick (commit_on_durable_advance) whenever durable_lsn + * advances, racing with this test's own repl_on_ack-driven kick. The + * coalescing guarantees exactly one kick actually runs the walk, but it can + * be the FSYNC THREAD's, asynchronously, after repl_on_ack has already + * returned on the main test thread. Tests that trigger a commit and then + * need to observe its APPLIED effect (not just committed_lsn) use + * tap_wait_applied() to poll persist_last_applied_lsn() before reading + * final state — see that helper's doc comment for the full mechanism and + * the flake it fixes. + * + * Every test below captures ALL assertion-relevant observables into local + * variables and calls tap_teardown() UNCONDITIONALLY before running any + * TEST_ASSERT — a failed assertion's longjmp must never skip teardown, or it + * hangs the next test in this binary (the known longjmp-past-teardown + * pathology from task 3.1's round-2 lesson, tasks.md Implementation Notes). + */ + +#include + +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "cluster.h" +#include "commit.h" +#include "controller.capnp.h" +#include "hook.h" +#include "jobj.h" +#include "persist.h" +#include "repl.h" +#include "wal_types.h" +#include "xce/cluster_def.h" +#include "xce/types.h" +#include "xstring.h" + +/* jobj.c's jobsub table and the process-global xstring table are shared + * across every test file in the controller_unit_test binary — guard + * idempotently, mirroring every other raft-apply-decoupling test file. */ +static int g_tap_jobobj_ready; + +static cluster_t g_tap_c; +static commit_engine_t g_tap_hce; +static int g_tap_hep_a, g_tap_hep_b; /* dummy standby endpoints */ +static char g_tap_tmpdir[256]; + +/* Test-injectable send hook (repl.c, XCE_TESTING) — keeps repl_broadcast from + * dereferencing the dummy standby endpoints above, mirroring + * test_commit_barrier_reply.c's identical cbrp_drop_send_hook. */ +extern int (*repl_broadcast_send_hook)(void *ep, wal_catchup_entry_t **entries, + int n_entries); + +static int tap_drop_send_hook(void *ep, wal_catchup_entry_t **entries, + int n_entries) +{ + (void)ep; (void)entries; (void)n_entries; + return 0; /* pretend success, drop the data */ +} + +/* ── Spy hook: records the exact (detail, jobid) sequence of every + * AFTER_EVENT firing on XCE_CONTROLLER_HOOK_JOB, in the order they occur. ── */ +#define TAP_SPY_CAP 32 +typedef struct { + int detail; + uint64_t jobid; +} tap_spy_entry_t; + +static tap_spy_entry_t g_tap_spy[TAP_SPY_CAP]; +static int g_tap_spy_n; + +static int tap_spy_hook(int type, int situation, int detail, void* data, + va_list ap) +{ + (void)type; (void)data; + if (situation != HOOK_AFTER_EVENT) + return 0; + + uint64_t jobid = 0; + + switch (detail) { + case HOOK_DETAIL_JOB_NEW: { + /* args: cluster_t*, xlist_t* (apply_committed_entry's WAL_JOB_SUBMIT + * branch) -- consume the leading cluster_t* first, mirroring + * persist_job_hook_fun's own va_arg pattern for this hook type. */ + (void)va_arg(ap, cluster_t*); + xlist_t* jlist = va_arg(ap, xlist_t*); + if (jlist && jlist->count > 0) { + job_t* j = (job_t*)((void**)jlist->data)[0]; + jobid = j->job.jobid; + } + break; + } + case HOOK_DETAIL_JOB_UPDATE: { + /* args: cluster_t*, job_t* -- same leading-arg shape. */ + (void)va_arg(ap, cluster_t*); + job_t* j = va_arg(ap, job_t*); + if (j) + jobid = j->job.jobid; + break; + } + default: + return 0; /* not relevant to this file's assertions */ + } + + if (g_tap_spy_n < TAP_SPY_CAP) { + g_tap_spy[g_tap_spy_n].detail = detail; + g_tap_spy[g_tap_spy_n].jobid = jobid; + g_tap_spy_n++; + } + return 0; +} + +static void tap_spy_reset(void) +{ + g_tap_spy_n = 0; + memset(g_tap_spy, 0, sizeof(g_tap_spy)); +} + +static void tap_setup(void) +{ + kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING, + .nslots = 32, + .delta = 16 }; + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 32, + .delta = 16 }; + + if (makeString((char*)"tap-probe") == NULL) + xstring_table_init(64); + + if (!g_tap_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_tap_jobobj_ready = 1; + } + + memset(&g_tap_c, 0, sizeof(g_tap_c)); + g_tap_c.myhost = &g_tap_c.myhost_info; + g_tap_c.myhost_info.name = (char*)"ctrl-test"; + g_tap_c.myhost_info.key = (char*)""; + + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_tap_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_tap_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_tap_c.qmgr = makeInstance(&xkvmgr_type, &str_params); + g_tap_c.workermgr = makeInstance(&xkvmgr_type, &str_params); + g_tap_c.resmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_tap_c.jobmgr); + TEST_ASSERT_NOT_NULL(g_tap_c.qmgr); + TEST_ASSERT_NOT_NULL(g_tap_c.workermgr); + TEST_ASSERT_NOT_NULL(g_tap_c.resmgr); + zarray_init(&g_tap_c.sharedinst, 16, NULL); + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.sync_timeout_ms = 2000; + + snprintf(g_tap_tmpdir, sizeof(g_tap_tmpdir), "/tmp/xce_tap61_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_tap_tmpdir)); + persist_set_workdir(&g_tap_c.persist, g_tap_tmpdir, "ctrl"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&g_tap_c.persist, &cfg, &g_tap_c)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&g_tap_c.persist)); + + /* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until a + * test explicitly delivers them via repl_on_ack. commit_set_pool is + * deliberately never called -- see file header for why. */ + TEST_ASSERT_TRUE( + repl_add_standby(&g_tap_c.persist.repl, &g_tap_hep_a, "sb_a", 7500) >= 0); + TEST_ASSERT_TRUE( + repl_add_standby(&g_tap_c.persist.repl, &g_tap_hep_b, "sb_b", 7501) >= 0); + TEST_ASSERT_EQUAL_INT(0, + commit_engine_init(&g_tap_hce, &g_tap_c.persist, 3, 8)); + repl_set_commit_engine(&g_tap_c.persist.repl, &g_tap_hce); + repl_broadcast_send_hook = tap_drop_send_hook; + + TEST_ASSERT_EQUAL_INT(0, + xce_add_hook(&g_tap_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tap_spy_hook, + &g_tap_c)); + + tap_spy_reset(); +} + +static void tap_teardown(void) +{ + repl_broadcast_send_hook = NULL; + commit_engine_fini(&g_tap_hce); + persist_stop(&g_tap_c.persist); + persist_fini(&g_tap_c.persist); + + zarray_fini(&g_tap_c.sharedinst); + if (g_tap_c.resmgr) dereference(g_tap_c.resmgr); + if (g_tap_c.workermgr) dereference(g_tap_c.workermgr); + if (g_tap_c.qmgr) dereference(g_tap_c.qmgr); + if (g_tap_c.jobmgr) dereference(g_tap_c.jobmgr); + xce_hook_fini(&g_tap_c.hooksys); +} + +/* Creates a job directly in the fixture's jobmgr, bypassing WAL entirely — + * for tests that only need a pre-existing job for a fact to transition, + * mirroring test_apply_reapply_idempotency.c's tari_seed_job. */ +static job_submit_data_t g_tap_seed_submit; + +static job_t* tap_seed_job(uint64_t jobid, uint32_t state) +{ + xce_job_t jdef; + + memset(&g_tap_seed_submit, 0, sizeof(g_tap_seed_submit)); + g_tap_seed_submit.queue = (char*)"q"; + g_tap_seed_submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.submit = &g_tap_seed_submit; + + return makeJob(g_tap_c.jobmgr, &jdef); +} + +static job_t* tap_find_job(uint64_t jobid) +{ + job_t* j = NULL; + lock(g_tap_c.jobmgr, RTTI_LOCK_READ); + int found = key_find(g_tap_c.jobmgr, (void*)jobid, (void**)&j); + unlock(g_tap_c.jobmgr); + return (found > 0) ? j : NULL; +} + +/* A standby ACK only counts toward majority once this node's OWN copy is + * durable (persist_durable_lsn, capped by the group-commit flusher's own + * fdatasync cadence -- ha-commit-durability's majority_candidate_locked + * caps self's contribution at durable_lsn, never the raw appended tip). + * persist_wal_async signals the flusher's condvar immediately after each + * append, but the flusher still runs on its own thread -- poll briefly + * rather than assume it has already caught up by the time this function + * returns, mirroring cbrp_poll_append's identical polling idiom in + * test_commit_barrier_reply.c. */ +static void tap_wait_durable(uint64_t lsn, uint32_t deadline_ms) +{ + for (uint32_t waited = 0; waited < deadline_ms; waited += 5) { + if (atomic_load(&g_tap_c.persist.durable_lsn) >= lsn) + return; + usleep(5000); + } +} + +/* raft-apply-decoupling task 6.1 round 2 review: waits for the ordered apply + * pump to actually finish applying up through `lsn` after a triggering event + * (repl_on_ack) returns. This is needed even though ce->pool is NULL here + * (persist_apply_pump_kick's documented inline-fallback runs the walk on the + * CALLING thread rather than a vtask worker) -- a SEPARATE real thread, the + * persist group-commit flusher (fsync_worker, started by persist_start in + * tap_setup), ALSO independently drives a kick via commit_on_durable_advance + * every time durable_lsn itself advances, regardless of any ACK. That kick + * races with this test's own repl_on_ack-driven kick on the main thread. + * persist_apply_pump_kick's CAS-coalescing (apply_pump_scheduled) guarantees + * exactly one of the two ever actually runs the walk at a time -- but the + * LOSING side's caller returns immediately without having applied anything + * itself, handing responsibility to whichever thread already holds the + * flag. That is by design: the kick is "fire and forget" (coalesced, + * eventually-consistent dispatch), never "kick and block until this + * specific walk completes". So a test that calls repl_on_ack and then reads + * job state or hook-fire counts immediately cannot assume the walk that + * will apply its entries ran synchronously on ITS OWN thread, even with + * ce->pool NULL -- it may instead complete asynchronously, moments later, on + * the fsync worker thread. Poll persist_last_applied_lsn() (which only + * advances after a real apply pass, from whichever thread ran it) rather + * than assume repl_on_ack's return implies completion -- confirmed via + * review to reproduce as a genuine, reproducible flake (observed at ~7.5%, + * both in this file's burst-order test and the FAIL/FINISH race test) + * without this wait. */ +static void tap_wait_applied(uint64_t lsn, uint32_t deadline_ms) +{ + for (uint32_t waited = 0; waited < deadline_ms; waited += 5) { + if (persist_last_applied_lsn(&g_tap_c.persist) >= lsn) + return; + usleep(5000); + } +} + +/* ── 1. The task's named observable + Hazard 1 regression ───────────────── */ + +static void test_apply_pump_kick_applies_committed_burst_in_ascending_lsn_order( + void) +{ + tap_setup(); + + job_t* j901 = tap_seed_job(901, JOB_STAT_PEND); + job_t* j902 = tap_seed_job(902, JOB_STAT_PEND); + int seeded_ok = (j901 != NULL) && (j902 != NULL); + + uint64_t first_lsn = g_tap_c.persist.current_lsn; + + /* Append three facts, deliberately in an order that is NOT sorted by + * jobid and NOT the order a human might narrate the story in — job 902 + * (START then FINISH) interleaved around job 901's single START. LSN + * order is exactly the append order below: 902/RUN, 901/RUN, 902/DONE. */ + int rc1 = persist_submit_job_fact(&g_tap_c.persist, WAL_JOB_START, 902, + JOB_STAT_RUN); + int rc2 = persist_submit_job_fact(&g_tap_c.persist, WAL_JOB_START, 901, + JOB_STAT_RUN); + int rc3 = persist_submit_job_fact(&g_tap_c.persist, WAL_JOB_FINISH, 902, + JOB_STAT_DONE); + + uint64_t watermark = g_tap_c.persist.current_lsn - 1; + + /* Snapshot the pre-commit state before delivering any ACK: nothing may + * have mutated or notified yet (Req 5.1, 5.3, 6.1). */ + uint32_t j901_state_precommit = seeded_ok ? j901->job.state : 0; + uint32_t j902_state_precommit = seeded_ok ? j902->job.state : 0; + int spy_n_precommit = g_tap_spy_n; + + /* One covering ACK jumps committed_lsn past all three in a single + * contiguous-prefix advance -- commit_resolve_pending's unconditional + * persist_apply_pump_kick(ce->p, ce->pool) call (task 6.1's commit.c + * wiring) fires here; ce->pool is NULL in this fixture, so the walk + * runs inline, synchronously, before repl_on_ack returns. Self's own + * copy must be durable up to `watermark` first, or the ACK cannot lift + * committed_lsn that far (self caps its own contribution at + * durable_lsn, never the raw appended tip). */ + tap_wait_durable(watermark, 2000); + repl_on_ack(&g_tap_c.persist.repl, &g_tap_hep_a, watermark); + + /* See tap_wait_applied's doc comment: the fsync-worker thread's own + * durable-advance-driven kick can win the coalescing race and apply this + * burst asynchronously, after repl_on_ack already returned here. */ + tap_wait_applied(watermark, 2000); + + uint64_t committed_after = atomic_load(&g_tap_c.persist.committed_lsn); + int spy_n_after = g_tap_spy_n; + tap_spy_entry_t spy_snap[TAP_SPY_CAP]; + memcpy(spy_snap, g_tap_spy, sizeof(spy_snap)); + uint32_t j901_state_final = seeded_ok ? j901->job.state : 0; + uint32_t j902_state_final = seeded_ok ? j902->job.state : 0; + uint64_t last_applied_final = persist_last_applied_lsn(&g_tap_c.persist); + uint64_t current_lsn_final = g_tap_c.persist.current_lsn; + + tap_teardown(); + + TEST_ASSERT_TRUE(seeded_ok); + TEST_ASSERT_EQUAL_INT(0, rc1); + TEST_ASSERT_EQUAL_INT(0, rc2); + TEST_ASSERT_EQUAL_INT(0, rc3); + TEST_ASSERT_EQUAL_UINT64(first_lsn + 2, watermark); + + /* Nothing has committed yet (ACKs withheld) -- nothing may be applied. */ + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j901_state_precommit, + "a fact must not mutate state before it commits"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j902_state_precommit, + "a fact must not mutate state before it commits"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, spy_n_precommit, + "no notification may fire before any entry in the burst commits"); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark, committed_after, + "the covering ACK must advance committed_lsn to the full watermark"); + + /* Strict order, no skipping or reordering (Req 3.3, 7.3): the spy's + * recorded sequence must be EXACTLY the append/LSN order, never sorted + * by jobid, never reflecting any other ordering scheme. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(3, spy_n_after, + "the burst must apply as exactly three notifications, one per " + "committed entry, in a single pump pass"); + TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_UPDATE, spy_snap[0].detail); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(902, spy_snap[0].jobid, + "entry 1 (lowest LSN) must apply first, regardless of jobid"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(901, spy_snap[1].jobid, + "entry 2 must apply second, in strict LSN order"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(902, spy_snap[2].jobid, + "entry 3 (highest LSN) must apply last, even though it targets the " + "SAME job as entry 1 -- LSN order, not jobid grouping, governs"); + + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j901_state_final, + "job 901 must reflect its single applied fact"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j902_state_final, + "job 902 must reflect its LAST applied fact (FINISH), proving " + "entry 2 (job 901, in between) did not get skipped or cause " + "job 902's later entry to apply out of turn"); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark, last_applied_final, + "last_applied_lsn must advance to exactly the applied watermark"); + + /* Hazard 1 regression: applying these facts live must not re-append + * them. Without the t_apply_seam_notifying guard, each HOOK_DETAIL_ + * JOB_UPDATE notification above would have looped back through + * persist_job_hook_fun and appended a duplicate WAL_JOB_UPDATE entry, + * growing current_lsn past this test's own three appends. */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark + 1, current_lsn_final, + "the live apply pump must never re-append an already-committed " + "entry (Hazard 1) -- current_lsn must reflect only this test's own " + "three appends, nothing more"); + + printf("test_apply_pump_kick_applies_committed_burst_in_ascending_lsn_" + "order: PASS\n"); +} + +/* ── 2. persist_apply_pump_run() direct call: mixed command + fact ──────── */ + +static int tap_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct Job cj; + Job_ptr jp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_Job(cs, &cj, (xce_job_t*)j); + jp = new_Job(cs); + write_Job(&cj, jp); + capn_setp(capn_root(&c), 0, jp.p); + + *out_len = capn_size(&c); + *out = malloc(*out_len); + if (!*out) { capn_free(&c); return -ENOMEM; } + capn_write_mem(&c, *out, *out_len, 0); + capn_free(&c); + return 0; +} + +static void test_apply_pump_run_applies_mixed_submit_and_fact_entries(void) +{ + tap_setup(); + + job_t* j903 = tap_seed_job(903, JOB_STAT_PEND); + int seeded_ok = (j903 != NULL); + + /* Entry 1 (lower LSN): a fact for the pre-existing job. */ + int rc1 = persist_submit_job_fact(&g_tap_c.persist, WAL_JOB_START, 903, + JOB_STAT_RUN); + + /* Entry 2 (higher LSN): a WAL_JOB_SUBMIT command entry for a brand-new + * job, appended directly via persist_wal_async (bypassing the full + * commit_submit_async/continuation pipeline, which is task 3.2/6.2's + * concern, not this walk's) -- this file only needs the entry's bytes + * to exist on disk at a known LSN, exactly what the live pump consumes + * regardless of how the entry got there. */ + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = 904; + jdef.state = JOB_STAT_PEND; + jdef.submit = &g_tap_seed_submit; + + uint8_t* payload = NULL; + size_t plen = 0; + int enc_rc = tap_encode_job(&jdef, &payload, &plen); + int rc2 = persist_wal_async(&g_tap_c.persist, WAL_JOB_SUBMIT, + payload, (uint32_t)plen); + free(payload); + + uint64_t watermark = g_tap_c.persist.current_lsn - 1; + + /* White-box: mark the whole range committed directly (this test targets + * the walk's own dispatch logic, not commit-advance itself -- that is + * test 1's job) and drive the walk with the exposed, directly-callable + * entry point. */ + atomic_store(&g_tap_c.persist.committed_lsn, watermark); + + int applied = persist_apply_pump_run(&g_tap_c.persist); + int applied2 = persist_apply_pump_run(&g_tap_c.persist); + + uint32_t j903_state_final = seeded_ok ? j903->job.state : 0; + job_t* j904 = tap_find_job(904); + uint32_t j904_state_final = j904 ? j904->job.state : 0; + uint64_t last_applied_final = persist_last_applied_lsn(&g_tap_c.persist); + + tap_teardown(); + + TEST_ASSERT_TRUE(seeded_ok); + TEST_ASSERT_EQUAL_INT(0, rc1); + TEST_ASSERT_EQUAL_INT(0, enc_rc); + TEST_ASSERT_EQUAL_INT(0, rc2); + + /* raft-apply-decoupling task 6.1 round 2 review: this fixture's + * commit_engine_t has a real fsync_worker thread wired (tap_setup), and + * persist_submit_job_fact/persist_wal_async above are real appends -- + * once each becomes durable, commit_on_durable_advance (persist.c) fires + * on the FSYNC THREAD and unconditionally calls persist_apply_pump_kick, + * completely independent of this test never calling repl_on_ack. That + * background kick can race this test's own two direct + * persist_apply_pump_run() calls below and win the apply_pump_lock + * first -- applying both entries itself before either of this test's + * own calls gets a turn (persist_apply_pump_run is mutex-serialized and + * fully synchronous, so whichever caller actually performs the walk + * always completes it before returning -- there is no partial/in-flight + * state to poll for here, unlike tests 1/4's tap_wait_applied + * situation). When the background thread wins BEFORE either of this + * test's own calls, BOTH `applied` and `applied2` legitimately read 0 + * (nothing was left for this test's own calls to do) even though every + * entry did get applied, correctly, by the background thread's own + * separate call -- so neither the individual values nor their sum can + * be asserted exactly in this shared-thread fixture without a deeper + * synchronization change. What IS deterministic, and what this test + * exists to actually prove, is a purely state-based invariant: every + * committed entry ends up applied by SOMEONE exactly once, regardless + * of commit mode, and the final materialized state is correct -- + * checked below via last_applied_lsn and the two jobs' live state, + * never via the racy call-arity split. Observed empirically via review: + * the original strict `applied==2 && applied2==0` split flaked at + * ~1.6% under this exact race. */ + TEST_ASSERT_TRUE_MESSAGE(applied >= 0 && applied <= 2 && + applied2 >= 0 && applied2 <= 2, + "the walk must never report a negative or over-large apply count"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j903_state_final, + "the fact entry must have applied"); + TEST_ASSERT_NOT_NULL_MESSAGE(j904, + "the WAL_JOB_SUBMIT command entry must have applied too -- the walk " + "does not skip command-mode entries"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j904_state_final); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark, last_applied_final, + "last_applied_lsn must reach the watermark no matter which thread " + "performed the walk -- this, not the applied/applied2 call-arity " + "split, is the authoritative proof both entries were applied " + "exactly once"); + + printf("test_apply_pump_run_applies_mixed_submit_and_fact_entries: " + "PASS\n"); +} + +/* ── 3. Hazard 1 guard precision: job_ctrl's own live append must survive ── */ + +static void test_apply_pump_guard_does_not_break_job_ctrl_live_append(void) +{ + tap_setup(); + + /* A suspended, not-yet-started job (allocs == NULL) -- job_ctrl's + * SIGCONT branch calls mark_job_update() directly, synchronously + * mutating state and firing BEFORE/AFTER_EVENT HOOK_DETAIL_JOB_UPDATE + * itself, exactly the same detail apply_committed_entry() now also + * fires for live-applied facts (test 1 above). This is the dual-origin + * case the t_apply_seam_notifying guard must NOT break. */ + job_t* j = tap_seed_job(905, JOB_STAT_PSUSP); + int seeded_ok = (j != NULL) && (j->job.allocs == NULL); + + uint64_t first_lsn = g_tap_c.persist.current_lsn; + + xce_signal_t sig = { .signo = SIGCONT, .force = 0 }; + xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 905 }; + ctrl.u.siginfo = &sig; + + int rc = job_ctrl(&g_tap_c, &ctrl); + + uint32_t state_final = seeded_ok ? j->job.state : 0xffffffffu; + uint64_t current_lsn_final = g_tap_c.persist.current_lsn; + + tap_teardown(); + + TEST_ASSERT_TRUE(seeded_ok); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(0, state_final & JOB_STAT_PSUSP, + "SIGCONT must clear the suspend bit synchronously, as it always " + "has -- job_ctrl's command path is untouched by task 6.1"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(first_lsn + 1, current_lsn_final, + "job_ctrl's own live append via persist_job_hook_fun must still " + "happen exactly once -- the apply-seam double-append guard is " + "scoped to apply_committed_entry()'s own call stack (a thread-local " + "flag), not to the HOOK_DETAIL_JOB_UPDATE value itself, so it must " + "never suppress this legitimate, unrelated live append"); + + printf("test_apply_pump_guard_does_not_break_job_ctrl_live_append: " + "PASS\n"); +} + +/* ── 4. Merge-staleness regression: FAIL must not clobber a racing FINISH ─ + * + * Reproduces the round-1 corruption bug found by review: a FINISH fact + * (deterministic, state-independent -- exit_code alone decides DONE vs + * EXIT) and a POST_EXEC-stage FAIL fact for the SAME job are received in + * two separate commit rounds, with the FAIL's receipt happening BEFORE + * either fact has applied (both are durably appended and registered, but + * apply is deferred past the commit barrier -- exactly task 6.1's point). + * Driven through the REAL receipt handlers (on_job_finish/on_job_fail), not + * seeded directly, so the genuine staleness window is exercised: at FAIL + * receipt time, the job's real in-memory state is still RUN (FINISH has + * not applied yet). Old (round-1) code baked a snapshot of that STALE RUN + * state into the FAIL's WAL entry, so once both entries applied in LSN + * order the FAIL's blind overwrite clobbered FINISH's DONE with a stray + * RUN bit, corrupting the job to RUN|EXIT|PERR instead of the correct + * merged terminal state DONE|PERR. */ + +static void test_apply_pump_fail_racing_finish_merges_against_current_state( + void) +{ + tap_setup(); + + job_t* j906 = tap_seed_job(906, JOB_STAT_RUN); + int seeded_ok = (j906 != NULL); + + uint64_t first_lsn = g_tap_c.persist.current_lsn; + + /* Round 1: worker reports FINISH (exit_code 0 -> DONE). Durably + * appended only -- record_job_finish_fact fires no AFTER_EVENT and + * mutates nothing, so j906->job.state must still read RUN right after + * this call returns. */ + xce_jobfinish_t finish = { .jobid = 906, .exit_code = 0 }; + int rc_finish = on_job_finish(&g_tap_c, &finish); + + /* Round 2: BEFORE either fact has applied, a POST_EXEC-stage FAIL + * arrives for the same job (an ordinary HPC lifecycle event -- e.g. an + * epilogue script failing shortly after the main process exits). At + * this exact instant, job.state is STILL RUN -- the genuine staleness + * window this test reproduces, not a seeded/simulated one. */ + uint32_t state_at_fail_receipt = seeded_ok ? j906->job.state : 0; + + job_failinfo_t failinfo = { .jobid = 906, + .stage = JOB_EXEC_STAGE_POST_EXEC, + .code = 1, + .detail = NULL }; + int rc_fail = on_job_fail(&g_tap_c, &failinfo); + + uint32_t state_before_commit = seeded_ok ? j906->job.state : 0; + uint64_t watermark = g_tap_c.persist.current_lsn - 1; + + /* One covering ACK commits both entries at once; the real + * commit_resolve_pending -> persist_apply_pump_kick production path + * applies them in strict LSN order (FINISH first, then FAIL), exactly + * as task 6.1's pump guarantees. */ + tap_wait_durable(watermark, 2000); + repl_on_ack(&g_tap_c.persist.repl, &g_tap_hep_a, watermark); + + /* See tap_wait_applied's doc comment: the fsync-worker thread's own + * durable-advance-driven kick can win the coalescing race and apply + * these two entries asynchronously, after repl_on_ack already returned + * here -- this is the exact mechanism behind the round-2 self-disclosed + * flake, not a corruption bug in the merge-staleness fix itself. */ + tap_wait_applied(watermark, 2000); + + uint64_t committed_after = atomic_load(&g_tap_c.persist.committed_lsn); + uint32_t state_final = seeded_ok ? j906->job.state : 0; + + tap_teardown(); + + TEST_ASSERT_TRUE(seeded_ok); + TEST_ASSERT_EQUAL_INT(0, rc_finish); + TEST_ASSERT_EQUAL_INT(0, rc_fail); + TEST_ASSERT_EQUAL_UINT64(first_lsn + 1, watermark); + + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, state_at_fail_receipt, + "the FAIL fact must be received while the job is still genuinely " + "RUN -- proving FINISH has not applied yet, the real staleness " + "window, not a simulated one"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, state_before_commit, + "neither fact may mutate state before its entry commits (Req 5.1)"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark, committed_after, + "the covering ACK must advance committed_lsn past both entries"); + + TEST_ASSERT_EQUAL_UINT32_MESSAGE((JOB_STAT_DONE | JOB_STAT_PERR), + state_final, + "FAIL's PERR merge must be computed against FINISH's already-" + "applied DONE (correct terminal state) -- not a stale RUN snapshot " + "taken at FAIL's receipt time"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(0, state_final & JOB_STAT_RUN, + "the stray RUN bit from the round-1 corruption bug must never " + "appear in the final state"); + TEST_ASSERT_TRUE_MESSAGE((state_final & JOB_STAT_DONE) != 0, + "FINISH's DONE bit must survive -- it must never be silently " + "replaced by a later FAIL's stale-computed overwrite"); + + printf("test_apply_pump_fail_racing_finish_merges_against_current_" + "state: PASS\n"); +} + +/* ── Test group entry point ───────────────────────────────────────────── */ + +void test_apply_pump(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_apply_pump_kick_applies_committed_burst_in_ascending_lsn_order); + RUN_TEST(test_apply_pump_run_applies_mixed_submit_and_fact_entries); + 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); + UNITY_END(); +} diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index fc752af3..4cb0572f 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -1860,9 +1860,14 @@ static void test_update_op_rusage_receipt_reply_facts_retained(void) * entry (payload "p", 1 byte) and any non-jobstate entries are skipped by * the payload-size check. Single segment is guaranteed here: rotation * needs 64 MB (WAL_SEGMENT_SIZE_BYTES). Returns the match count; - * *state_out (optional) receives the LAST matching entry's state. */ -static int cbrp_count_jobstate_entries(persist_t *p, uint32_t wal_type, - uint64_t jobid, uint32_t *state_out) + * state_out/pad_out (both optional pointers) receive the LAST matching + * entry's state/_pad. raft-apply-decoupling task 6.1 round 2: WAL_JOB_FAIL and a + * postdone WAL_JOB_UPDATE now carry their real information in `_pad` (see + * wal_types.h's doc comment on wal_jobstate_t._pad) rather than a + * precomputed `state` — pad_out lets a test observe that. */ +static int cbrp_count_jobstate_entries_ex(persist_t *p, uint32_t wal_type, + uint64_t jobid, uint32_t *state_out, + uint32_t *pad_out) { char seg_path[sizeof(p->seg_path)]; uint64_t seg_end, off = 0; @@ -1895,6 +1900,8 @@ static int cbrp_count_jobstate_entries(persist_t *p, uint32_t wal_type, count++; if (state_out != NULL) *state_out = js.state; + if (pad_out != NULL) + *pad_out = js._pad; } } off += sizeof(hdr) + hdr.payload_len; @@ -1903,6 +1910,14 @@ static int cbrp_count_jobstate_entries(persist_t *p, uint32_t wal_type, return count; } +/* Thin wrapper preserving every pre-existing caller's signature/contract + * unchanged (pad = 0, "legacy absolute state" — see wal_types.h). */ +static int cbrp_count_jobstate_entries(persist_t *p, uint32_t wal_type, + uint64_t jobid, uint32_t *state_out) +{ + return cbrp_count_jobstate_entries_ex(p, wal_type, jobid, state_out, NULL); +} + /* Drive one fact call: it must succeed (rc 0 — in particular NEVER * -ETIMEDOUT) and return latency-bounded. sync_timeout_ms is 5000 in this * test, so a fact path that wrongly ran a WAIT barrier with ACKs @@ -2372,24 +2387,32 @@ static void test_guard_replay_finish_equals_single_application(void) /* 5.2 boundary pin — a FORWARD update on a terminal job must NOT be * over-rejected by the guard. on_job_postdone is the reachable - * record_job_update_fact caller: on a DONE job it OR's JOB_STAT_PDONE + * record_job_postdone_fact caller: on a DONE job it OR's JOB_STAT_PDONE * (terminal preserved, strictly forward). This must still durably append - * (an entry appends, carrying the merged state) — the guard rejects - * backward, never a forward merge. Preservation pin: green before and after - * task 6.2. + * (an entry appends, marked for the correct apply-time merge) — the guard + * rejects backward, never a forward merge. Preservation pin: green before + * and after task 6.2. * - * raft-apply-decoupling task 4.1: on_job_postdone is now durable-append-only - * — the oracle for "did the forward merge compute correctly" moves from live - * state to the WAL-recorded fact (cbrp_count_jobstate_entries), mirroring - * test_fact_retention_across_quorum_loss. The live-state-unchanged - * assertion is this task's own new pin (Req 5.1): live state must still show - * only DONE (from cbrp_drive_to_done's direct apply), never PDONE, since - * nothing applies this fact live yet. */ + * raft-apply-decoupling task 4.1: on_job_postdone is now durable-append-only. + * task 6.1 round 2 (merge-staleness fix, Req 11.3): the WAL-recorded fact no + * longer carries a precomputed `state` at all — postdone's merge depends on + * whatever state holds at APPLY time (which may differ from receipt time + * once apply is deferred), so it is recomputed then by + * apply_job_postdone_transition(), never baked into the WAL entry. The + * oracle for "was this fact durably recorded correctly" is therefore the + * WAL entry's `_pad` marker (cbrp_count_jobstate_entries_ex); the oracle for + * "does the forward merge itself compute correctly" is a direct call to + * apply_job_postdone_transition() below, mirroring how + * test_guard_fail_after_finish_preserves_perr_merge pins apply_job_fail_ + * transition(). The live-state-unchanged assertion is task 4.1's own pin + * (Req 5.1): live state must still show only DONE (from cbrp_drive_to_done's + * direct apply), never PDONE, since nothing applies this fact live yet. */ static void test_guard_forward_update_on_terminal_preserved(void) { uint64_t lsn_after_done, lsn_after_pdone; - uint32_t st_disk = 0; + uint32_t pad_disk = 0; int st_live; + job_t scratch; cbrp_setup(3000); cbrp_make_job(73, &g_job_submit); @@ -2404,18 +2427,33 @@ static void test_guard_forward_update_on_terminal_preserved(void) lsn_after_pdone = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); - /* Assert: the update appended, carrying the correctly forward-merged - * state (terminal DONE preserved, PDONE merged in — not rejected). */ + /* Assert: the update appended, marked for a correct apply-time forward + * merge (terminal DONE preserved, PDONE merged in — not rejected). */ TEST_ASSERT_TRUE_MESSAGE(lsn_after_pdone > lsn_after_done, "a forward update on a terminal job must still append its WAL entry " "— the guard rejects backward transitions, never forward merges"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, - cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_UPDATE, 73, - &st_disk), + cbrp_count_jobstate_entries_ex(&g_hc.persist, WAL_JOB_UPDATE, 73, + NULL, &pad_disk), "exactly one UPDATE fact must be durably recorded (Req 5.1)"); - TEST_ASSERT_TRUE_MESSAGE((st_disk & JOB_STAT_DONE) != 0, + TEST_ASSERT_EQUAL_UINT32_MESSAGE(WAL_JOB_UPDATE_POSTDONE_MARK, pad_disk, + "the recorded UPDATE fact must be marked for apply-time postdone " + "recomputation (task 6.1 round 2) — not carry a state precomputed " + "at receipt time, which would go stale under deferred apply"); + + /* Assert: the actual forward-merge guarantee this test exists to pin — + * DONE preserved, PDONE merged in, not rejected — verified directly + * against the guard core that performs it at apply time, on an + * independent scratch job seeded at the same terminal state job 73 was + * driven to above. */ + memset(&scratch, 0, sizeof(scratch)); + scratch.job.jobid = 73; + scratch.job.state = JOB_STAT_DONE; + TEST_ASSERT_EQUAL_INT(JOB_TRANSITION_APPLIED, + apply_job_postdone_transition(&scratch)); + TEST_ASSERT_TRUE_MESSAGE((scratch.job.state & JOB_STAT_DONE) != 0, "the recorded UPDATE fact must preserve the terminal bit"); - TEST_ASSERT_TRUE_MESSAGE((st_disk & JOB_STAT_PDONE) != 0, + TEST_ASSERT_TRUE_MESSAGE((scratch.job.state & JOB_STAT_PDONE) != 0, "the recorded UPDATE fact must merge JOB_STAT_PDONE (not be " "rejected)"); @@ -2439,18 +2477,27 @@ static void test_guard_forward_update_on_terminal_preserved(void) * Preservation pin: green before and after task 6.2. * * raft-apply-decoupling task 4.1: on_job_fail (via the new - * record_job_fail_fact) is now durable-append-only — it computes the - * PERR-merged target state on a throwaway copy of the job and durably - * appends it, without touching the real job's live state. The oracle for - * "did the merge compute correctly" moves from live state to the - * WAL-recorded fact (cbrp_count_jobstate_entries); the live-state-unchanged - * assertion is this task's own new pin (Req 5.1). */ + * record_job_fail_fact) is now durable-append-only. task 6.1 round 2 + * (merge-staleness fix, Req 11.3): the WAL-recorded fact no longer carries a + * precomputed target state at all — the PERR merge genuinely depends on + * whatever state holds at APPLY time (a receipt-time snapshot goes stale + * the moment another fact for the same job, e.g. a FINISH, is still pending + * apply — the exact bug this round fixed), so it is recomputed then by the + * REAL apply_job_fail_transition(), never baked into the WAL entry at + * receipt time. The oracle for "was this fact durably recorded correctly" + * is therefore the WAL entry's `_pad`-carried stage + * (cbrp_count_jobstate_entries_ex); the oracle for "does the PERR merge + * itself compute correctly" is a direct call to apply_job_fail_transition() + * below (its own logic is unchanged by this round — only WHEN it runs + * changed). The live-state-unchanged assertion is task 4.1's own pin + * (Req 5.1). */ static void test_guard_fail_after_finish_preserves_perr_merge(void) { job_failinfo_t failinfo; uint64_t lsn_after_done, lsn_after_fail; - uint32_t st_disk = 0; + uint32_t pad_disk = 0; int st_live; + job_t scratch; cbrp_setup(3000); cbrp_make_job(74, &g_job_submit); @@ -2470,19 +2517,36 @@ static void test_guard_fail_after_finish_preserves_perr_merge(void) lsn_after_fail = g_hc.persist.current_lsn; pthread_mutex_unlock(&g_hc.persist.lock); - /* Assert: the FAIL fact appended, carrying the correctly PERR-merged - * state — DONE preserved, PERR merged in, not reset to bare EXIT. */ + /* Assert: the FAIL fact appended, marked for a correct apply-time + * PERR-merge — DONE preserved, PERR merged in, not reset to bare EXIT. */ TEST_ASSERT_TRUE_MESSAGE(lsn_after_fail > lsn_after_done, "the PERR-merge fail must still append its WAL entry (forward " "terminal→terminal merge — not a guarded-out no-op)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, - cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FAIL, 74, - &st_disk), + cbrp_count_jobstate_entries_ex(&g_hc.persist, WAL_JOB_FAIL, 74, + NULL, &pad_disk), "exactly one FAIL fact must be durably recorded (Req 5.1)"); - TEST_ASSERT_TRUE_MESSAGE((st_disk & JOB_STAT_DONE) != 0, + TEST_ASSERT_EQUAL_UINT32_MESSAGE( + WAL_JOB_FAIL_STAGE_FLAG | (uint32_t)JOB_EXEC_STAGE_POST_EXEC, + pad_disk, + "the recorded FAIL fact must carry its stage marked for apply-time " + "recomputation (task 6.1 round 2) — not a state precomputed at " + "receipt time, which would go stale under deferred apply"); + + /* Assert: the actual PERR-merge guarantee this test exists to pin — + * DONE preserved, PERR merged in, not reset to bare EXIT — verified + * directly against the guard core that performs it at apply time, on an + * independent scratch job seeded at the same terminal state job 74 was + * driven to above. */ + memset(&scratch, 0, sizeof(scratch)); + scratch.job.jobid = 74; + scratch.job.state = JOB_STAT_DONE; + TEST_ASSERT_EQUAL_INT(JOB_TRANSITION_APPLIED, + apply_job_fail_transition(&scratch, &failinfo)); + TEST_ASSERT_TRUE_MESSAGE((scratch.job.state & JOB_STAT_DONE) != 0, "the recorded FAIL fact must keep the DONE terminal bit " "(PERR-merge behavior)"); - TEST_ASSERT_TRUE_MESSAGE((st_disk & JOB_STAT_PERR) != 0, + TEST_ASSERT_TRUE_MESSAGE((scratch.job.state & JOB_STAT_PERR) != 0, "the recorded FAIL fact must OR JOB_STAT_PERR onto the existing " "terminal state (PERR-merge behavior)"); @@ -2684,7 +2748,7 @@ static void test_fact_state_unchanged_throughout_quorum_loss_outage(void) uint64_t lsn_before, lsn_after; uint64_t committed_at_poll; int initial_state, poll_state, final_state; - uint32_t st_start = 0, st_fin = 0, st_fail = 0, st_upd = 0; + uint32_t st_start = 0, st_fin = 0, pad_fail = 0, pad_upd = 0; int cnt_start, cnt_fin, cnt_fail, cnt_upd; int i; enum { OUTAGE_POLL_COUNT = 5, OUTAGE_POLL_INTERVAL_MS = 20 }; @@ -2790,10 +2854,10 @@ static void test_fact_state_unchanged_throughout_quorum_loss_outage(void) &st_start); cnt_fin = cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FINISH, 85, &st_fin); - cnt_fail = cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_FAIL, 85, - &st_fail); - cnt_upd = cbrp_count_jobstate_entries(&g_hc.persist, WAL_JOB_UPDATE, 85, - &st_upd); + cnt_fail = cbrp_count_jobstate_entries_ex(&g_hc.persist, WAL_JOB_FAIL, 85, + NULL, &pad_fail); + cnt_upd = cbrp_count_jobstate_entries_ex(&g_hc.persist, WAL_JOB_UPDATE, + 85, NULL, &pad_upd); TEST_ASSERT_EQUAL_INT_MESSAGE(1, cnt_start, "job 85 START fact must remain durably retained exactly once " @@ -2809,14 +2873,20 @@ static void test_fact_state_unchanged_throughout_quorum_loss_outage(void) "throughout the whole outage window (Req 6.1)"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, st_start); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, st_fin); - TEST_ASSERT_TRUE_MESSAGE((st_upd & JOB_STAT_PDONE) != 0, - "the recorded UPDATE fact must carry JOB_STAT_PDONE"); - TEST_ASSERT_TRUE_MESSAGE((st_upd & JOB_STAT_EXIT) != 0, - "the recorded UPDATE fact must carry the backstop JOB_STAT_EXIT " - "bit postdone computes when no prior exit status was set"); - (void)st_fail; /* FAIL's merge result depends on current state at the - * (never-reached) apply step; only its durable - * retention count is this test's concern. */ + /* task 6.1 round 2 (Req 11.3): FAIL's and postdone's UPDATE's merge + * results depend on current state at the (never-reached, in this + * outage-only test) apply step — they are no longer precomputed at + * receipt time at all (the merge-staleness fix), so only their durable + * retention count and their apply-time-recomputation marker are this + * test's concern. */ + TEST_ASSERT_EQUAL_UINT32_MESSAGE( + WAL_JOB_FAIL_STAGE_FLAG | (uint32_t)JOB_EXEC_STAGE_POST_EXEC, + pad_fail, + "the recorded FAIL fact must carry its stage marked for apply-time " + "recomputation"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(WAL_JOB_UPDATE_POSTDONE_MARK, pad_upd, + "the recorded UPDATE fact must be marked for apply-time postdone " + "recomputation"); /* Final check: after the ENTIRE outage window, job state is STILL * unchanged and no reaction has fired — sealing the "for the duration diff --git a/src/include/persist.h b/src/include/persist.h index 523617c3..c98b6e69 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -387,6 +387,20 @@ struct persist_s { * persist_register_pending_apply's doc comment for the full contract. */ void* pending_applies; pthread_mutex_t pending_applies_lock; + + /* raft-apply-decoupling task 6.1: ordered apply pump. apply_pump_lock + * serializes concurrent walk passes -- multiple commit-advance triggers + * (from any of the six commit.c paths that can move committed_lsn) can + * race to kick the pump; only one walk actually runs at a time (see + * persist_apply_pump_run's doc comment). apply_pump_scheduled is a + * CAS-coalescing flag: at most one vtask-dispatched walk is ever + * in-flight per persist_t regardless of how many trigger events land + * concurrently, so a commit burst cannot flood the vtask pool with + * mostly-redundant walk requests. Neither field is persisted -- both + * reset fresh on process restart (restart applies via replay, not this + * pump; see design.md "Restart Replay & Verification"). */ + pthread_mutex_t apply_pump_lock; + _Atomic int apply_pump_scheduled; #ifdef XCE_TESTING /* Fault-injection seam (ha-commit-durability task 2.1 error-path test): * when non-zero, every fdatasync site treats the sync as FAILED — it @@ -945,6 +959,93 @@ void persist_register_pending_apply(persist_t* p, uint64_t lsn, * and remove"), not built by this task. */ int persist_has_pending_apply(persist_t* p, uint64_t lsn); +/* raft-apply-decoupling task 6.1: "take and remove" counterpart to + * persist_register_pending_apply()/persist_has_pending_apply() (task 3.2), + * built for task 6.2 to consume — task 3.2 explicitly left this accessor + * unbuilt ("task 6.1 needs its own 'take and remove' accessor, not built + * here"). Atomically finds the pending-apply hand-off node registered for + * `lsn` (if any), unlinks and frees the list node, and returns its + * `on_applied`/`ctx` pair via the output parameters. + * + * Returns 1 and populates *out_fn and *out_ctx when a registration was found + * and removed; returns 0 and leaves *out_fn and *out_ctx untouched when no + * registration exists for `lsn` (the common case for a fact entry, which + * never registers one). + * + * Contract for task 6.2: call this AFTER apply_committed_entry() has already + * applied `lsn` — this accessor does not itself apply anything, it is pure + * bookkeeping over the hand-off list. When it returns 1, invoke + * `(*out_fn)(*out_ctx, lsn)` exactly once; a second call for the same lsn is + * guaranteed to return 0 (the node is already removed), so the hand-off + * fires exactly once per registration. Safe to call for every applied lsn + * unconditionally (facts and commands alike): it is a cheap linked-list scan + * under pending_applies_lock and a correct no-op (returns 0) when nothing is + * registered. Not yet called by anything — task 6.2's job, not this one's. */ +int persist_take_pending_apply(persist_t* p, uint64_t lsn, + pending_apply_notify_fn* out_fn, void** out_ctx); + +/* raft-apply-decoupling task 6.1: strict-order walk that applies every newly + * committed entry, from persist_last_applied_lsn()+1 up to p->committed_lsn + * inclusive, invoking apply_committed_entry() for each in ascending LSN + * order — regardless of the entry's commit mode (WAIT or RETAIN): a command + * entry that is *also* still tracked in the pending_applies hand-off list + * (task 3.2) is applied exactly like any other entry here; task 6.2 (not + * built by this function) is responsible for separately calling + * persist_take_pending_apply() and firing that entry's reply continuation + * after this function (or whatever drives it) applies it. + * + * Synchronous, single pass: walks until no further committed-but-unapplied + * entry exists (as of the moment each check is made), then returns. Safe to + * call directly — e.g. from a test, without ever going through + * persist_apply_pump_kick's async dispatch — serializing against any + * concurrent walk via a dedicated lock (never `p->lock`: this performs + * blocking disk reads, and `p->lock` is the hot single-writer append lock — + * see the lock-ordering invariant, persist.c ~3347-3350). + * + * Returns the number of entries applied (>= 0) on success, or a negative + * errno if reading a committed entry's bytes back off the WAL failed — a + * genuine, unexpected condition for an already-committed entry (logged + * loud; per design.md's Error Handling, this is treated as an apply-time + * failure, not a retry-immediately condition). The walk stops at the failed + * LSN; last_applied_lsn is left unadvanced past it, so the SAME entry is + * retried, from the same position, by the next kick — never silently + * skipped. */ +int persist_apply_pump_run(persist_t* p); + +/* raft-apply-decoupling task 6.1: async trigger for persist_apply_pump_run() + * above. Intended to be called from every commit.c-internal path that can + * advance p->committed_lsn (commit_resolve_pending, after ce->lock is + * released) so the pump reacts to ALL SIX advance paths task 1.2 identified + * (durable-advance, self-quorum commit_submit, ack-driven advance, the + * writable-lsn gate, the raft-library commit-index feed, and membership + * change) — not just the one raft-library-fed path + * (ha_cb_commit_advanced_stub/on_commit_advanced) design.md names + * explicitly, which alone would starve the pump under a single-node/ + * self-quorum cluster that never exercises that specific callback. + * + * No-op when p is NULL or p->cluster is NULL (mirrors apply_committed_entry's + * own bare-persist-fixture convention) — every existing commit-engine-only / + * persist-only unit fixture that never wires a cluster_t is completely + * unaffected by this call; the pump only ever activates for a real cluster + * context. + * + * `pool` is a `vtaskmgr_t*` (declared void* here to avoid persist.h + * depending on vtaskmgr.h, mirroring pending_apply_notify_fn's existing + * avoidance of a commit.h dependency) used to dispatch the walk off the + * calling thread — never inline on what may be the replication-ACK/iomgr + * thread — via vtask_enq_classified(VTASK_LIVE_BLOCK): same class and the + * same inline-on-dispatch-failure fallback discipline as commit.c's own + * dispatch_continuation, so a walk is never silently dropped even under OOM + * or a NULL/not-yet-wired pool (a NULL pool falls back to an inline, + * synchronous run — exactly the "commit-only unit fixture" case + * dispatch_continuation's own doc comment already documents). + * + * Never blocks the caller beyond an O(1) CAS + enqueue (the actual walk + * always runs asynchronously on a pool worker, except the documented + * NULL-pool/allocation-failure fallback). Coalesces concurrent kicks: at + * most one walk is ever scheduled or running at a time per persist_t. */ +void persist_apply_pump_kick(persist_t* p, void* pool); + int persist_job_hook_fun(int type, int situation, int detail, void* cls, va_list ap); int persist_queue_hook_fun(int type, int situation, int detail, @@ -960,13 +1061,29 @@ int persist_queue_hook_fun(int type, int situation, int detail, * mutation this fact eventually drives, and the AFTER_EVENT notification * itself, both happen later from apply_committed_entry() once this entry * commits to a majority (task 2.2's existing dispatch). `state` is the - * caller-computed resultant state (e.g. JOB_STAT_RUN for a START, or the - * PERR-merged value for a FAIL) — this function performs no guard checks or - * merge logic of its own, it only encodes and appends. + * caller-computed resultant state (e.g. JOB_STAT_RUN for a START) — this + * function performs no guard checks or merge logic of its own, it only + * encodes and appends. Thin wrapper over persist_submit_job_fact_ex() with + * pad = 0 (the "legacy absolute state" discriminator — see wal_types.h's + * doc comment on wal_jobstate_t._pad) — unchanged contract, every existing + * caller (START/FINISH/postdone's old shape) keeps working identically. * Returns the local-append rc (0 on success); facts have no commit timeout * to report, mirroring wal_hook_append_classified's own contract. */ int persist_submit_job_fact(persist_t* p, uint32_t wal_type, uint64_t jobid, int state); + +/* Full-control sibling of persist_submit_job_fact() (raft-apply-decoupling + * task 6.1 round 2, Req 3.3/7.1-7.3): lets the caller set wal_jobstate_t's + * `_pad` discriminator explicitly, for the two fact producers whose real + * mutation must be re-derived at apply time against the job's CURRENT + * state rather than trusting a receipt-time snapshot (record_job_fail_fact, + * record_job_postdone_fact — see wal_types.h's doc comment on + * wal_jobstate_t._pad for the full rationale). `state` is ignored by + * apply_committed_entry() whenever `pad` selects one of these apply-time- + * recomputed shapes (its value is not part of the wire contract in that + * case, but is still written for forward-compatibility/diagnostics). */ +int persist_submit_job_fact_ex(persist_t* p, uint32_t wal_type, + uint64_t jobid, int state, uint32_t pad); int persist_worker_hook_fun(int type, int situation, int detail, void* cls, va_list ap); @@ -990,9 +1107,27 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, * WAL_JOB_SUBMIT -> restore_job() (existing * idempotent, jobid-addressable construction primitive; -EEXIST is a * no-op re-apply, not an error) - * WAL_JOB_{START,FINISH,FAIL,STOP,RESUME, - * UPDATE,RUSAGE,CKPT,PREASON} -> apply_job_update_transition() - * (the guard-only core extracted in task 2.1) + * WAL_JOB_{START,FINISH,STOP,RESUME, + * RUSAGE,CKPT,PREASON} -> apply_job_update_transition() + * (the guard-only core extracted in task 2.1) — these types' payloads + * are deterministic/state-independent, so a `state` value baked at + * receipt time never goes stale + * WAL_JOB_FAIL -> dispatches on wal_jobstate_t. + * _pad (task 6.1 round 2 — see wal_types.h's doc comment on + * wal_jobstate_t._pad): WAL_JOB_FAIL_STAGE_FLAG set -> reconstructs a + * minimal job_failinfo_t{jobid, stage} and calls the REAL + * apply_job_fail_transition() against the job's ACTUAL CURRENT state + * at apply time — fixes a merge-staleness corruption where a FAIL's + * PERR merge depended on prior state that had not actually applied + * yet when the fact was received; _pad == 0 (legacy live path, + * job_ctrl's mark_job_fail) -> apply_job_update_transition() as an + * absolute overwrite, unchanged + * WAL_JOB_UPDATE -> also dispatches on _pad: + * WAL_JOB_UPDATE_POSTDONE_MARK -> apply_job_postdone_transition(), + * which re-derives the postdone merge (conditional EXIT backstop + OR + * JOB_STAT_PDONE) fresh against current state; _pad == 0 (legacy live + * path, job_ctrl's mark_job_update) -> apply_job_update_transition() + * as an absolute overwrite, unchanged * WAL_QUEUE_CTRL -> restore_queue_state() * WAL_WORKER_CTRL -> worker_from_capnp() * and fires the entry type's AFTER_EVENT notification hook exactly once, @@ -1011,12 +1146,14 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, * `.data = { .pidinfo = info }`) inside `case HOOK_AFTER_EVENT:`, without * dereferencing any of its fields at this call site — unlike sched.c's * unconditional `info->jobid` access in a top-level struct initializer). - * Every job-state fact type therefore routes through the - * guard-only apply_job_update_transition() core and fires + * Every job-state fact type therefore fires the same * HOOK_DETAIL_JOB_UPDATE, whose subscribers take only (cluster_t*, job_t*) — * matching what this wire format can actually supply without fabricating - * data. No task in the current plan (2.3, 4.1, 4.2, 6.1, 6.2, 6.3) closes - * this gap — task 4.1 is exclusively about removing state mutation/hook- + * data — regardless of which of the guard-only cores above (apply_job_ + * update_transition / apply_job_fail_transition / apply_job_postdone_ + * transition) actually performed the mutation. No task in the current plan + * (2.3, 4.1, 4.2, 6.1, 6.2, 6.3) closes this hook-granularity gap — task 4.1 + * is exclusively about removing state mutation/hook- * firing from worker-fact receipt-time handlers, not about enriching the * wire format. Closing it would require either a design.md correction * (formally accepting this HOOK_DETAIL_JOB_UPDATE substitution as final — diff --git a/src/include/wal_types.h b/src/include/wal_types.h index 2e10d29a..279f9184 100644 --- a/src/include/wal_types.h +++ b/src/include/wal_types.h @@ -17,6 +17,43 @@ typedef struct { uint32_t _pad; } wal_jobstate_t; +/* wal_jobstate_t._pad discriminator for WAL_JOB_FAIL / WAL_JOB_UPDATE + * (raft-apply-decoupling task 6.1 round 2 — fixes a job-state corruption + * bug). Both types have TWO producers: + * + * - A LIVE, synchronous, mutate-then-notify path (job_ctrl's + * mark_job_fail/mark_job_update): the mutation and the WAL-state capture + * happen back-to-back under the same job write-lock, so `state` is + * already correct and absolute by the time it is written — no staleness + * window exists. This path always leaves _pad == 0 (it zero-inits + * wal_jobstate_t and never sets it). + * + * - A DURABLE-APPEND-ONLY, deferred-apply path (record_job_fail_fact / + * record_job_postdone_fact in job.c): the real mutation depends on the + * job's CURRENT state at the moment it is computed. Precomputing that at + * receipt time and baking the result into `state` is unsafe once apply + * is deferred past the commit barrier — another fact for the same job + * (e.g. a FINISH) can still be in flight and apply AFTER this fact is + * received but BEFORE this fact's own entry applies, making the baked + * snapshot stale by the time it is actually used. This path marks _pad + * nonzero so apply_committed_entry() (persist.c) routes it to a + * dedicated apply-time recomputation against the job's ACTUAL current + * state instead of a blind overwrite. + * + * _pad == 0 always means "legacy: `state` is the absolute value to assign + * as-is". A nonzero _pad's exact interpretation is per-type — see the two + * markers below. */ +#define WAL_JOB_FAIL_STAGE_FLAG 0x80000000u + /* WAL_JOB_FAIL: _pad = WAL_JOB_FAIL_STAGE_FLAG | job_failinfo_t.stage. + * Apply time reconstructs a minimal job_failinfo_t from {jobid, stage} + * and calls the real apply_job_fail_transition() against current state. */ +#define WAL_JOB_UPDATE_POSTDONE_MARK 1u + /* WAL_JOB_UPDATE: _pad = WAL_JOB_UPDATE_POSTDONE_MARK marks a postdone + * merge-fact (`state` is unused/0) — apply time calls + * apply_job_postdone_transition(), which re-derives the postdone merge + * (conditional EXIT backstop + OR JOB_STAT_PDONE) fresh against current + * state. Any other nonzero value is reserved/unused today. */ + /* WAL_QUEUE_CTRL — queue name is a NUL-terminated string immediately after * this header. Total payload = sizeof(wal_queuehdr_t) + strlen(name)+1. */ typedef struct {