From 88577f4658041e02f0e3669545a0c269292a507d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 5 Jul 2026 00:38:02 +0000 Subject: [PATCH] feat(raft-apply-decoupling): fire command continuation post-apply (task 6.2) Wire persist_apply_pump_run() to call persist_take_pending_apply() immediately after each entry's apply_committed_entry() call, invoking the registered continuation strictly after the mutation. Adds jobsubmit_reply_success_once() (job.c) so a successful job submission finally sends a client reply once every array element applies -- closes the standing gap from task 3.2 where committed submissions never got a response. Fixes bug-1415: commit_resolve_pending() dispatches the commit continuation and the apply-pump kick as two independent VTASK_LIVE_BLOCK vtasks with no completion-order guarantee. On any real multi-vcpu pool (the production default, n_vcpus=8), if the apply pump won the race and advanced last_applied_lsn past an LSN before its continuation registered, the registration was permanently orphaned -- the job would be constructed and visible, but the client would hang forever. Untested by every existing fixture in this spec (all use pool=NULL, which trivially serializes and can't reproduce the race). Fixed with a self-healing check in persist_register_pending_apply(): if the LSN is already <= last_applied_lsn, invoke the callback immediately instead of registering for later pickup. Verified by the spec's first genuine multi-vcpu regression test (real vtaskmgr_t, n_vcpus=4), independently RED->GREEN reproduced by two review rounds. --- .kiro/specs/raft-apply-decoupling/tasks.md | 3 +- src/controller/job.c | 81 ++-- src/controller/persist.c | 35 ++ src/controller/testsuite/test_apply_pump.c | 357 ++++++++++++++++++ .../testsuite/test_commit_barrier_reply.c | 245 ++++++++++++ src/include/persist.h | 23 ++ 6 files changed, 719 insertions(+), 25 deletions(-) diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index 8b8e7ae..cc5cd30 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -136,7 +136,7 @@ - _Requirements: 3.3, 7.3_ - _Depends: 1.2, 2.2_ -- [ ] 6.2 Fire a command's registered continuation immediately after its entry applies +- [x] 6.2 Fire a command's registered continuation immediately after its entry applies - After the walk in 6.1 applies an entry that has a registered commit continuation (from task 3.2), invoke that continuation now, after the mutation, not before - Leave fact entries with no further action beyond what the apply dispatch function itself performs - Observable: a successful job submission's client reply carries the job id only after the job is visible in the in-memory job table; the reply always occurs strictly after the mutation @@ -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.2**: closes the standing operational gap flagged since task 3.2 — successful job submissions finally get a client reply again. Wired `persist_apply_pump_run()` (task 6.1) to call `persist_take_pending_apply(p, lsn, &out_fn, &out_ctx)` immediately after each entry's `apply_committed_entry()` call, invoking the continuation when found — strictly after the mutation, per the task's named observable. Facts need no special-casing: `persist_take_pending_apply()` naturally returns 0 for them since the fact-receipt path (task 4.1) never calls `persist_register_pending_apply()`. Added `jobsubmit_reply_success_once()` (job.c, sibling of the pre-existing `jobsubmit_reply_failure_once()`) — `jobsubmit_apply_complete()` now calls it once every array element resolves, sending `RequestReply_data_jobid` to the client. Took 2 review rounds. **Round 1 REJECTED — a second real, production-severity bug found this task-group, on top of task 6.1's data-corruption bug**: self-disclosed as bug-1415, confirmed by the reviewer via direct code trace: `commit_resolve_pending()` (commit.c) enqueues the WAIT-mode continuation's dispatch and `persist_apply_pump_kick()`'s walk as TWO INDEPENDENT `VTASK_LIVE_BLOCK` vtasks with NO completion-order guarantee — `vdispatcher_entry()` binds each ready task to a different idle vcpu and does not wait for a prior dispatch to complete. Confirmed reachable on EVERY default production deployment (`cluster_init_phase2()` always wires a real `vtaskmgr_t`, default `n_vcpus=8`, never NULL) — and confirmed completely untested by every fixture in this entire spec, all of which use `pool=NULL` (which trivially serializes the two dispatches in enqueue order by falling back to synchronous inline execution, and so can never exercise the race). **Consequence**: if the apply-pump task wins, it applies the entry and advances `last_applied_lsn` past the LSN before the continuation task ever calls `persist_register_pending_apply()` — when that registration finally arrives for an already-passed LSN, nothing will ever come back to consume it. **The job is correctly constructed and visible, but the client hangs forever, plus a leaked `jobsubmit_async_ctx_t`/endpoint ref** — a coin-flip-or-worse failure mode on the feature's own primary success path, for every default multi-vcpu production controller. **Round 2 fixed it** with a self-healing check at the top of `persist_register_pending_apply()`: `if (lsn <= persist_last_applied_lsn(p)) { on_applied(ctx, lsn); return; }` — using the existing `persist_last_applied_lsn()` accessor, no signature change. Verified race-safe in both interleavings: non-racing case is a no-op (confirmed unchanged behavior for every existing test, all of which use `pool=NULL` and so never trigger the check); racing case fires the callback exactly once, immediately, on the registering thread — and can never double-fire because a self-healed registration is never linked onto `pending_applies`, so the walk's own `persist_take_pending_apply()` call for that LSN (which already ran and found nothing, which is precisely how we know the registration is late) has nothing left to find. **First and only test in this entire spec to wire a REAL multi-vcpu `vtaskmgr_t`** (`n_vcpus=4`, via `commit_set_pool()`) — every other fixture in the spec deliberately uses `pool=NULL` and structurally cannot reproduce this class of bug; the new test's poll-barrier ties directly to the real race predicate (`persist_last_applied_lsn(p) < lsn`), not a timing guess, and self-checks its own precondition so it fails loudly rather than silently degrading if the forced interleaving doesn't occur. RED→GREEN independently reproduced by both rounds' reviewers. **One narrow, non-blocking residual finding for a fast-follow, not required to close bug-1415**: the self-heal check (an unlocked atomic read) and the subsequent list-link (under `pending_applies_lock`) are not one atomic unit — a sufficiently long preemption between the check and the lock could theoretically still reproduce a vastly narrower version of the original race. Recommended fix: perform the self-heal check while already holding `pending_applies_lock` (the same lock `persist_take_pending_apply()` uses), making check-and-link atomic with respect to the walk's take. Not blocking since the reported, reproduced, production-relevant (dispatch-level) race is genuinely and completely closed. **Two pre-existing tests in test_commit_barrier_reply.c still fail as EXPECTED** (unchanged from task 6.1 — same root cause, task 7.3's job). **Task-group-6 running tally: 2 of 2 tasks landed so far have each required a full REJECTED→fixed round for a genuine, independently-verified production-severity bug** (6.1: job-state corruption from stale receipt-time fact computation; 6.2: a client-hang race from unordered async dispatch) — this task group is where deferred-apply's real concurrency implications are surfacing, exactly as expected for the task that finally activates the live path. Remaining: 6.3/6.4 (integration tests through the full pump) should be reviewed with the same level of scrutiny for further concurrency surprises. - **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. diff --git a/src/controller/job.c b/src/controller/job.c index ce1368a..a2cb0ee 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -623,6 +623,11 @@ typedef struct { * once, in jobsubmit_async_ctx_release() */ uint64_t seq; /* xce_request_t.seq, copied for the reply */ uint32_t opcode; /* xce_request_t.opcode, copied for the reply */ + uint64_t jobid; /* base jobid of the (possibly array) submission, + * echoed back on the success reply (task 6.2) -- + * matches pre-3.1 semantics (git history commit + * 6f1fa97), which always reported the BASE id, + * never a per-element id */ pthread_mutex_t lock; /* guards pending/replied below */ int pending; /* entries not yet FULLY resolved (see above) */ int replied; /* 1 once a reply has been sent for this batch */ @@ -663,24 +668,55 @@ static void jobsubmit_reply_failure_once(jobsubmit_async_ctx_t* actx, int rc) xce_reply(actx->ep, actx->c->myhost, &reply); } -/* raft-apply-decoupling task 6 hand-off target (task 6 itself is NOT built - * by this task): invoked by the future ordered apply pump immediately - * after it applies this entry via apply_committed_entry() — task 6.2's own - * contract, "fire a command's registered continuation immediately after - * its entry applies". Until task 6 exists, nothing ever calls this - * function; see jobsubmit_commit_continuation's COMMITTED branch below, - * which registers it as the hand-off target instead of applying or - * replying itself. +/* Sends exactly one success reply for the whole batch, carrying the base + * jobid — the sibling of jobsubmit_reply_failure_once, gated on the SAME + * actx->replied flag (Req 9.3) so a batch that already sent a failure reply + * for a sibling element never follows it with a contradictory success reply, + * and vice versa. Only ever reached from jobsubmit_apply_complete once every + * element's continuation has resolved (raft-apply-decoupling task 6.2). */ +static void jobsubmit_reply_success_once(jobsubmit_async_ctx_t* actx) +{ + xce_reply_t reply; + int send = 0; + + pthread_mutex_lock(&actx->lock); + if (!actx->replied) { + actx->replied = 1; + send = 1; + } + pthread_mutex_unlock(&actx->lock); + + if (!send) + return; + + memset(&reply, 0, sizeof(reply)); + reply.seq = actx->seq; + reply.opcode = actx->opcode; + reply.status = 0; + reply.data_which = RequestReply_data_jobid; + reply.data.jobid = actx->jobid; + + xce_reply(actx->ep, actx->c->myhost, &reply); +} + +/* raft-apply-decoupling task 6 hand-off target: invoked by the ordered apply + * pump (persist_apply_pump_run(), task 6.1/6.2) immediately after it applies + * this entry via apply_committed_entry() — task 6.2's own contract, "fire a + * command's registered continuation immediately after its entry applies". + * See jobsubmit_commit_continuation's COMMITTED branch below, which + * registers this as the hand-off target instead of applying or replying + * itself. * - * NOTE for whoever builds task 6.2: apply_committed_entry()'s WAL_JOB_SUBMIT - * case (persist.c) already constructs the job (via restore_job()) AND fires - * the HOOK_DETAIL_JOB_NEW AFTER_EVENT notification, independently per entry - * — so by the time this function is invoked, the job is very likely already - * visible and already notified. This function most likely only needs to - * send the batch's success reply (echoing the jobid) once every entry has - * been applied, NOT reconstruct or re-notify anything — see the removed-code - * note above handle_jobsubmit's old (pre-3.2) construct/notify block for the - * full reasoning (git history, commit 6f1fa97, task 3.1). */ + * apply_committed_entry()'s WAL_JOB_SUBMIT case (persist.c) already + * constructs the job (via restore_job()) AND fires the HOOK_DETAIL_JOB_NEW + * AFTER_EVENT notification, independently per entry, strictly BEFORE the + * apply pump reaches the persist_take_pending_apply()/invoke step that calls + * this function — so by the time this function runs, the job is already + * visible in c->jobmgr and already notified. This function therefore only + * sends the batch's success reply (echoing the base jobid) once every entry + * has been applied; it never reconstructs or re-notifies anything — see the + * removed-code note above handle_jobsubmit's old (pre-3.2) construct/notify + * block for the full reasoning (git history, commit 6f1fa97, task 3.1). */ static void jobsubmit_apply_complete(void* ctx, uint64_t lsn) { jobsubmit_async_ctx_t* actx = (jobsubmit_async_ctx_t*)ctx; @@ -693,14 +729,10 @@ static void jobsubmit_apply_complete(void* ctx, uint64_t lsn) done = (actx->pending <= 0); pthread_mutex_unlock(&actx->lock); - /* Task 6.2's success-reply belongs here (RequestReply_data_jobid, - * gated on actx->replied exactly like jobsubmit_reply_failure_once, so - * a partially-failed batch's earlier failure reply is never followed - * by a contradictory success reply) — not built by this task, which - * must not reply from the committed-outcome path at all (see the - * struct doc comment above). */ - if (done) + if (done) { + jobsubmit_reply_success_once(actx); jobsubmit_async_ctx_release(actx); + } } /* raft-apply-decoupling task 3.2: per-entry outcome of one array element's @@ -847,6 +879,7 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) actx->ep = ep; actx->seq = request->seq; actx->opcode = request->opcode; + actx->jobid = jobid; actx->pending = plist.count; actx->replied = 0; pthread_mutex_init(&actx->lock, NULL); diff --git a/src/controller/persist.c b/src/controller/persist.c index 3b15694..084122f 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -3139,6 +3139,24 @@ int persist_apply_pump_run(persist_t* p) atomic_store_explicit(&p->last_applied_lsn, next, memory_order_relaxed); applied++; + + /* raft-apply-decoupling task 6.2 (Req 1.1, 2.1-2.3): consume this + * LSN's command hand-off, if any (task 3.2's + * persist_register_pending_apply(), e.g. handle_jobsubmit's + * commit-outcome continuation) -- strictly AFTER the mutation above + * and after last_applied_lsn already reflects it, never before or + * during. A fact entry never registers a hand-off, so + * persist_take_pending_apply() simply returns 0 for it and this is a + * correct no-op with no per-entry-type branching needed here. + * persist_take_pending_apply() unlinks the registration before + * returning it, so re-applying the same lsn (a future retry/replay + * landing on this exact walk position again) can never fire the + * continuation a second time. */ + pending_apply_notify_fn on_applied = NULL; + void* apply_ctx = NULL; + + if (persist_take_pending_apply(p, next, &on_applied, &apply_ctx)) + on_applied(apply_ctx, next); } pthread_mutex_unlock(&p->apply_pump_lock); @@ -4585,6 +4603,23 @@ void persist_register_pending_apply(persist_t* p, uint64_t lsn, if (p == NULL || on_applied == NULL) return; + /* bug-1415 self-healing (round 2): commit.c's continuation dispatch and + * the apply pump's walk (persist_apply_pump_run) are two independent + * vtasks with no completion-order guarantee -- see this function's doc + * comment in persist.h for the full race. If the walk has already + * applied `lsn` (last_applied_lsn already at or past it) by the time we + * get here, linking this registration onto pending_applies would orphan + * it forever: the walk only ever visits each lsn once, and it already + * called persist_take_pending_apply() for this lsn and found nothing. + * Fire the hand-off immediately, synchronously, on THIS thread instead + * -- never leave a continuation unresolved (controllers never + * self-terminate; the same "fail loud and complete the work" posture + * commit.c's dispatch_continuation fallback uses). */ + if (lsn <= persist_last_applied_lsn(p)) { + on_applied(ctx, lsn); + return; + } + node = calloc(1, sizeof(*node)); if (node == NULL) { ERROR("persist_register_pending_apply: OOM registering the apply " diff --git a/src/controller/testsuite/test_apply_pump.c b/src/controller/testsuite/test_apply_pump.c index 4616d94..f49fb2f 100644 --- a/src/controller/testsuite/test_apply_pump.c +++ b/src/controller/testsuite/test_apply_pump.c @@ -52,16 +52,26 @@ * 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). + * + * Test 5 (bug-1415, task 6.2 round 2) is a deliberate exception to the + * "commit_set_pool is never called" note above: it is the one test in this + * file that wires a REAL multi-vcpu vtaskmgr_t pool, in its own dedicated + * fixture (tpr_setup/tpr_teardown, disjoint globals from tap_setup's) that + * leaves every other test in this file untouched. See that test's own + * header comment for why a real pool is required to reproduce bug-1415 at + * all. */ #include #include #include +#include #include #include #include #include +#include #include #include @@ -73,6 +83,7 @@ #include "jobj.h" #include "persist.h" #include "repl.h" +#include "vtaskmgr.h" #include "wal_types.h" #include "xce/cluster_def.h" #include "xce/types.h" @@ -702,6 +713,351 @@ static void test_apply_pump_fail_racing_finish_merges_against_current_state( "state: PASS\n"); } +/* ── 5. bug-1415 regression: late pending-apply registration self-heals ── + * + * Every test above (and every other test in this spec's test files) + * deliberately keeps ce->pool == NULL, so both commit.c's + * dispatch_continuation and persist_apply_pump_kick fall back to their + * documented inline/synchronous behavior — the continuation-registration + * path and the apply-pump walk can never actually run on two different + * threads in that mode, so the race bug-1415 describes cannot manifest + * there by construction (this is exactly why round 1's review found it + * untested by every fixture in this spec). + * + * This test wires a REAL vtaskmgr_t (n_vcpus > 1) via commit_set_pool, in + * its own dedicated fixture (tpr_setup/tpr_teardown, disjoint globals from + * tap_setup's) — commit_resolve_pending()'s two independent dispatches, the + * commit continuation (commit.c's dispatch_continuation) and the apply-pump + * kick (persist_apply_pump_kick), now run as REAL vtasks bound to separate + * idle vcpu worker threads by vtaskmgr's own dispatcher (vdispatcher_entry), + * exactly as production does (n_vcpus=8 by default, cluster_init_phase2). + * vtaskmgr.c has no completion-order guarantee between the two, so this + * test forces the specific bad interleaving deterministically instead of + * relying on natural timing luck: the test's own commit-continuation + * callback (tpr_commit_continuation, standing in for + * jobsubmit_commit_continuation — this test drives commit.c/persist.c + * directly, the actual layer bug-1415 lives in, rather than routing through + * job.c) blocks, polling persist_last_applied_lsn(), until the apply pump + * has ALREADY advanced past this entry's lsn — i.e. until + * persist_take_pending_apply() has already run (and found nothing, since + * nothing is registered yet) for this exact lsn — before ever calling + * persist_register_pending_apply(). This is a genuine barrier tied to the + * actual race condition, not a timing guess: as long as the pump eventually + * applies a durably committed entry (guaranteed here), the poll always + * terminates with the pump strictly ahead, reproducing the exact + * interleaving bug-1415 describes on every run, not just occasionally. + * + * Against the pre-fix persist_register_pending_apply(), this reproduces the + * permanent hang byte for byte: the registration links onto pending_applies + * after the walk has already passed this lsn for good (the walk visits each + * lsn exactly once), so nothing ever consumes it — the hand-off (this + * test's stand-in for jobsubmit_apply_complete / the client reply) never + * fires, and the stand-in resource (this test's stand-in for the leaked + * jobsubmit_async_ctx_t / endpoint ref) is never released. Against the + * self-healing fix, persist_register_pending_apply() detects the + * already-applied lsn and fires the hand-off immediately and synchronously + * instead of linking it, so the test passes: the hand-off fires exactly + * once, the resource is released exactly once, and no registration is left + * dangling on pending_applies. */ + +#define TPR_BARRIER_TIMEOUT_MS 3000 +#define TPR_WAIT_TIMEOUT_MS 8000 + +typedef struct { + persist_t* p; + pthread_mutex_t lock; + pthread_cond_t cond; + int saw_failed_outcome; + int pump_won_race; /* 1 once confirmed the walk + * applied lsn before registration + * ran */ + int apply_complete_fires; /* must end up == 1 */ + uint64_t apply_complete_lsn; + void* fake_resource; /* stands in for + * jobsubmit_async_ctx_t / the + * endpoint ref -- released exactly + * once by the hand-off firing */ + int resource_released; /* must end up == 1 */ +} tpr_race_ctx_t; + +static cluster_t g_tpr_c; +static commit_engine_t g_tpr_hce; +static vtaskmgr_t g_tpr_pool; +static int g_tpr_hep_a, g_tpr_hep_b; +static char g_tpr_tmpdir[256]; +static job_submit_data_t g_tpr_seed_submit; + +static job_t* tpr_seed_job(uint64_t jobid, uint32_t state) +{ + xce_job_t jdef; + + memset(&g_tpr_seed_submit, 0, sizeof(g_tpr_seed_submit)); + g_tpr_seed_submit.queue = (char*)"q"; + g_tpr_seed_submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.submit = &g_tpr_seed_submit; + + return makeJob(g_tpr_c.jobmgr, &jdef); +} + +static void tpr_setup(int n_vcpus) +{ + 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*)"tpr-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_tpr_c, 0, sizeof(g_tpr_c)); + g_tpr_c.myhost = &g_tpr_c.myhost_info; + g_tpr_c.myhost_info.name = (char*)"ctrl-test-race"; + g_tpr_c.myhost_info.key = (char*)""; + + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_tpr_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_tpr_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_tpr_c.qmgr = makeInstance(&xkvmgr_type, &str_params); + g_tpr_c.workermgr = makeInstance(&xkvmgr_type, &str_params); + g_tpr_c.resmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_tpr_c.jobmgr); + TEST_ASSERT_NOT_NULL(g_tpr_c.qmgr); + TEST_ASSERT_NOT_NULL(g_tpr_c.workermgr); + TEST_ASSERT_NOT_NULL(g_tpr_c.resmgr); + zarray_init(&g_tpr_c.sharedinst, 16, NULL); + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.sync_timeout_ms = 2000; + + snprintf(g_tpr_tmpdir, sizeof(g_tpr_tmpdir), "/tmp/xce_tpr62_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_tpr_tmpdir)); + persist_set_workdir(&g_tpr_c.persist, g_tpr_tmpdir, "ctrl"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&g_tpr_c.persist, &cfg, &g_tpr_c)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&g_tpr_c.persist)); + + /* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until + * this test explicitly delivers one via repl_on_ack — mirrors + * tap_setup exactly, up to this point. */ + TEST_ASSERT_TRUE( + repl_add_standby(&g_tpr_c.persist.repl, &g_tpr_hep_a, "sb_a", 7600) >= 0); + TEST_ASSERT_TRUE( + repl_add_standby(&g_tpr_c.persist.repl, &g_tpr_hep_b, "sb_b", 7601) >= 0); + TEST_ASSERT_EQUAL_INT(0, + commit_engine_init(&g_tpr_hce, &g_tpr_c.persist, 3, n_vcpus)); + repl_set_commit_engine(&g_tpr_c.persist.repl, &g_tpr_hce); + repl_broadcast_send_hook = tap_drop_send_hook; + + /* The one deliberate divergence from every other fixture in this file: + * a REAL multi-vcpu pool, started (vtaskmgr_start -- unlike + * test_commit_admission_pool.c's admission-only fixture, this test + * needs actual vtask execution on real worker threads), wired via + * commit_set_pool so commit_resolve_pending()'s two dispatches run as + * genuine concurrent vtasks instead of the inline fallback. */ + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_tpr_pool, n_vcpus)); + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&g_tpr_pool)); + commit_set_pool(&g_tpr_hce, &g_tpr_pool); +} + +static void tpr_teardown(void) +{ + repl_broadcast_send_hook = NULL; + commit_engine_fini(&g_tpr_hce); + vtaskmgr_fini(&g_tpr_pool); + persist_stop(&g_tpr_c.persist); + persist_fini(&g_tpr_c.persist); + + zarray_fini(&g_tpr_c.sharedinst); + if (g_tpr_c.resmgr) dereference(g_tpr_c.resmgr); + if (g_tpr_c.workermgr) dereference(g_tpr_c.workermgr); + if (g_tpr_c.qmgr) dereference(g_tpr_c.qmgr); + if (g_tpr_c.jobmgr) dereference(g_tpr_c.jobmgr); + xce_hook_fini(&g_tpr_c.hooksys); +} + +static void tpr_wait_durable(uint64_t lsn, uint32_t deadline_ms) +{ + for (uint32_t waited = 0; waited < deadline_ms; waited += 5) { + if (atomic_load(&g_tpr_c.persist.durable_lsn) >= lsn) + return; + usleep(5000); + } +} + +/* Task B's hand-off target: this test's stand-in for jobsubmit_apply_complete. + * Fires exactly once when persist_register_pending_apply()'s self-healing + * check invokes it inline (the fix), or would fire never (pre-fix hang) if + * the walk had already orphaned the registration. */ +static void tpr_apply_complete(void* ctx, uint64_t lsn) +{ + tpr_race_ctx_t* rctx = (tpr_race_ctx_t*)ctx; + + pthread_mutex_lock(&rctx->lock); + rctx->apply_complete_fires++; + rctx->apply_complete_lsn = lsn; + if (rctx->fake_resource != NULL) { + free(rctx->fake_resource); + rctx->fake_resource = NULL; + rctx->resource_released++; + } + pthread_cond_broadcast(&rctx->cond); + pthread_mutex_unlock(&rctx->lock); +} + +/* Task A stand-in for jobsubmit_commit_continuation: deterministically + * forces the exact bad interleaving bug-1415 describes (see this test's + * header comment) by blocking, on this vtask's OWN worker thread, until the + * apply pump (Task B, running concurrently on a separate vcpu) has already + * applied `lsn` -- THEN registering, reproducing a genuinely late + * persist_register_pending_apply() call. */ +static void tpr_commit_continuation(void* ctx, uint64_t lsn, + commit_outcome_t outcome, int rc) +{ + tpr_race_ctx_t* rctx = (tpr_race_ctx_t*)ctx; + + (void)rc; + + if (outcome != COMMIT_OUTCOME_COMMITTED) { + pthread_mutex_lock(&rctx->lock); + rctx->saw_failed_outcome = 1; + pthread_cond_broadcast(&rctx->cond); + pthread_mutex_unlock(&rctx->lock); + return; + } + + uint32_t waited_ms = 0; + while (persist_last_applied_lsn(rctx->p) < lsn && + waited_ms < TPR_BARRIER_TIMEOUT_MS) { + usleep(2000); + waited_ms += 2; + } + + pthread_mutex_lock(&rctx->lock); + rctx->pump_won_race = (persist_last_applied_lsn(rctx->p) >= lsn); + pthread_mutex_unlock(&rctx->lock); + + /* The exact call bug-1415 describes as arriving too late -- the apply + * pump has already visited (and will never revisit) this lsn. */ + persist_register_pending_apply(rctx->p, lsn, tpr_apply_complete, ctx); +} + +static void test_apply_pump_late_pending_apply_registration_self_heals(void) +{ + const int N_VCPUS = 4; /* > 1 is the whole point -- see header comment */ + + tpr_setup(N_VCPUS); + + job_t* j = tpr_seed_job(950, JOB_STAT_PEND); + int seeded_ok = (j != NULL); + + tpr_race_ctx_t rctx; + memset(&rctx, 0, sizeof(rctx)); + pthread_mutex_init(&rctx.lock, NULL); + pthread_cond_init(&rctx.cond, NULL); + rctx.p = &g_tpr_c.persist; + rctx.fake_resource = malloc(64); + int resource_alloc_ok = (rctx.fake_resource != NULL); + + int rc1 = persist_submit_job_fact(&g_tpr_c.persist, WAL_JOB_START, 950, + JOB_STAT_RUN); + uint64_t lsn = g_tpr_c.persist.current_lsn - 1; + + /* Register the commit-outcome continuation directly against the commit + * engine -- the same mechanism jobsubmit_commit_continuation uses + * (commit_submit_async), standing in for it at this layer since this + * test's boundary is commit.c/persist.c, not job.c. */ + int async_rc = commit_submit_async(&g_tpr_hce, lsn, tpr_commit_continuation, + &rctx); + + tpr_wait_durable(lsn, 2000); + /* Majority ACK (self + 1 of 2 standbys = 2 of 3 peers) drives + * commit_on_ack -> commit_advance_locked -> commit_resolve_pending, + * which dispatches BOTH the continuation (Task A) and the apply-pump + * kick (Task B) as real vtasks onto the pool wired above. */ + repl_on_ack(&g_tpr_c.persist.repl, &g_tpr_hep_a, lsn); + + /* Bounded poll for the hand-off -- the key assertion: the client-facing + * hand-off must still fire even though the walk (Task B) won the race + * and applied lsn before the continuation (Task A) ever registered. */ + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += (TPR_WAIT_TIMEOUT_MS / 1000); + deadline.tv_nsec += (long)(TPR_WAIT_TIMEOUT_MS % 1000) * 1000000L; + if (deadline.tv_nsec >= 1000000000L) { + deadline.tv_sec += 1; + deadline.tv_nsec -= 1000000000L; + } + + pthread_mutex_lock(&rctx.lock); + while (rctx.apply_complete_fires == 0) { + if (pthread_cond_timedwait(&rctx.cond, &rctx.lock, &deadline) != 0) + break; + } + int fires = rctx.apply_complete_fires; + int released = rctx.resource_released; + int pump_won_race = rctx.pump_won_race; + int saw_failed = rctx.saw_failed_outcome; + uint64_t fired_lsn = rctx.apply_complete_lsn; + pthread_mutex_unlock(&rctx.lock); + + int has_pending_after = persist_has_pending_apply(&g_tpr_c.persist, lsn); + + if (rctx.fake_resource != NULL) { + /* Only reachable pre-fix (orphaned forever, never released by + * tpr_apply_complete) -- free here so THIS test never leaks + * regardless of which way the assertions below come out. */ + free(rctx.fake_resource); + rctx.fake_resource = NULL; + } + + pthread_mutex_destroy(&rctx.lock); + pthread_cond_destroy(&rctx.cond); + + tpr_teardown(); + + TEST_ASSERT_TRUE(seeded_ok); + TEST_ASSERT_TRUE(resource_alloc_ok); + TEST_ASSERT_EQUAL_INT(0, rc1); + TEST_ASSERT_EQUAL_INT(0, async_rc); + TEST_ASSERT_FALSE_MESSAGE(saw_failed, + "the entry must genuinely commit -- a FAILED outcome would mean " + "this test never exercises the race it targets"); + TEST_ASSERT_TRUE_MESSAGE(pump_won_race, + "the forced barrier must confirm the apply pump actually applied " + "this lsn BEFORE the continuation registered -- otherwise this test " + "is not proving what it claims"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, fires, + "bug-1415: a persist_register_pending_apply() call that arrives " + "after the apply pump has already applied its lsn must still fire " + "its hand-off exactly once (self-healing) -- 0 means a permanent " + "hang (the orphaned-registration bug), >1 means a double-fire"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn, fired_lsn, + "the hand-off must fire for the correct lsn"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, released, + "the hand-off's resource (standing in for jobsubmit_async_ctx_t / " + "the endpoint ref) must be released exactly once -- no leak, no " + "double-free"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, has_pending_after, + "no orphaned registration may remain linked on pending_applies " + "after the hand-off fires"); + + printf("test_apply_pump_late_pending_apply_registration_self_heals: " + "PASS\n"); +} + /* ── Test group entry point ───────────────────────────────────────────── */ void test_apply_pump(void) @@ -711,5 +1067,6 @@ void test_apply_pump(void) 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); + RUN_TEST(test_apply_pump_late_pending_apply_registration_self_heals); UNITY_END(); } diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index 4cb0572..c2ff292 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -200,6 +200,13 @@ typedef struct { int n_replies; /* replymsg packets seen by this scan */ int32_t last_status; /* status of the last replymsg observed */ uint64_t committed_at_obs; /* committed_lsn sampled at observation time */ + int32_t last_data_which; /* RequestReply_data_which of the last + * replymsg observed (raft-apply-decoupling + * task 6.2: distinguishes a jobid-bearing + * success reply from a nodata one) */ + uint64_t last_jobid; /* rr.data.jobid of the last replymsg observed + * -- meaningful only when last_data_which == + * RequestReply_data_jobid */ } cbrp_reply_obs_t; static persist_t *g_obs_persist; /* sampled into committed_at_obs */ @@ -224,6 +231,9 @@ static void cbrp_scan_replies(endpoint_t *ep, cbrp_reply_obs_t *obs) read_RequestReply(&rr, rq.body.replymsg); obs->n_replies += 1; obs->last_status = rr.status; + obs->last_data_which = rr.data_which; + if (rr.data_which == RequestReply_data_jobid) + obs->last_jobid = rr.data.jobid; obs->committed_at_obs = atomic_load(&g_obs_persist->committed_lsn); } capn_free(&c); @@ -1389,6 +1399,237 @@ static void test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_element " PASS\n"); } +/* ── handle_jobsubmit success path via the ordered apply pump + * (raft-apply-decoupling task 6.2, Req 1.1, 2.1-2.3) ──────────────────── + * + * Everything above this point pins the COMMITTED-outcome path's gap left by + * task 3.2: a committed job submission is handed off to + * persist_register_pending_apply() and then has NO further observable + * effect (no reply, no job in c->jobmgr) — by design, until task 6 exists. + * Task 6.1 built the ordered apply pump that finally calls + * apply_committed_entry() live; task 6.2 (this section) wires the pump to + * consume that hand-off via persist_take_pending_apply() and fire the + * registered continuation (jobsubmit_apply_complete -> the client's success + * reply), strictly after the entry's mutation, never before or during it. + * + * Ordering proof mechanism: apply_committed_entry()'s WAL_JOB_SUBMIT case + * (persist.c) fires HOOK_DETAIL_JOB_NEW as an AFTER_EVENT, synchronously, + * immediately after inserting the job into c->jobmgr — and, by + * persist_apply_pump_run()'s own strict sequencing (task 6.2), strictly + * BEFORE the pump reaches its persist_take_pending_apply()/invoke step for + * that same lsn. A spy hook on this exact (situation, detail) pair can + * therefore observe, synchronously, whether a reply has already been + * enqueued on the requester's outq at the moment the mutation completes — + * mirroring this file's own pre-existing g_notify_reply_already_enqueued + * idiom (see notify() capture above) for the identical class of ordering + * assertion. */ + +static bool g_jobsubmit_order_spy_fired; +static bool g_jobsubmit_order_spy_reply_seen_early; +static uint64_t g_jobsubmit_order_spy_last_jobid; +static int g_jobsubmit_order_spy_fire_count; + +static int cbrp_jobsubmit_order_spy(int type, int situation, int detail, + void *data, va_list ap) +{ + (void)type; (void)data; + if (situation != HOOK_AFTER_EVENT || detail != HOOK_DETAIL_JOB_NEW) + return 0; + + /* apply_committed_entry's WAL_JOB_SUBMIT branch fires this hook with + * (cluster_t*, xlist_t*) — mirroring test_apply_pump.c's tap_spy_hook, + * which decodes the identical detail the same way. */ + (void)va_arg(ap, cluster_t *); + xlist_t *jlist = va_arg(ap, xlist_t *); + + g_jobsubmit_order_spy_fired = true; + g_jobsubmit_order_spy_fire_count++; + if (jlist && jlist->count > 0) { + job_t *j = (job_t *)((void **)jlist->data)[0]; + g_jobsubmit_order_spy_last_jobid = j->job.jobid; + } + /* THE ordering proof, latched sticky (OR'd) across every firing: once + * true, a reply was observably enqueued no later than one of these + * post-mutation notifications — a real ordering violation, regardless + * of which array element it happened on. */ + if (!DATAQ_EMPTY(&g_req_ep.outq)) + g_jobsubmit_order_spy_reply_seen_early = true; + return 0; +} + +static void cbrp_jobsubmit_order_spy_reset(void) +{ + g_jobsubmit_order_spy_fired = false; + g_jobsubmit_order_spy_reply_seen_early = false; + g_jobsubmit_order_spy_last_jobid = 0; + g_jobsubmit_order_spy_fire_count = 0; +} + +/* Single-element submission: the success reply must carry the job id, and + * must be observable only after the job is visible in c->jobmgr. */ +static void test_jobsubmit_success_reply_after_apply_carries_jobid(void) +{ + const uint64_t jobid = 9200; + uint64_t lsn_before, watermark; + cbrp_reply_obs_t early = { 0 }; + cbrp_reply_obs_t final = { 0 }; + bool job_exists_before_ack, job_exists_at_reply; + + cbrp_setup(2000); + TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_hc.hooksys, XCE_CONTROLLER_HOOK_JOB, + cbrp_jobsubmit_order_spy, &g_hc)); + cbrp_jobsubmit_order_spy_reset(); + cbrp_arrange_jobsubmit(jobid, 1); + + pthread_mutex_lock(&g_hc.persist.lock); + lsn_before = g_hc.persist.current_lsn; + pthread_mutex_unlock(&g_hc.persist.lock); + + g_op_arg.fn = handle_jobsubmit; + g_op_arg.req = &g_jobsub_req; + g_op_arg.rc = -9999; + TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, + cbrp_op_thread_fn, &g_op_arg)); + TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); + + watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000); + + /* Before any covering ACK: no reply, no job -- the pre-existing + * task-3.2 contract this task's own change must not disturb. */ + cbrp_sleep_ms(150); + cbrp_scan_replies(&g_req_ep, &early); + job_exists_before_ack = cbrp_job_exists(jobid); + + repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark); + cbrp_poll_reply(&g_req_ep, 2000, &final); + + job_exists_at_reply = cbrp_job_exists(jobid); + + cbrp_teardown(); + + TEST_ASSERT_TRUE_MESSAGE(watermark > 0, + "job submission must append its WAL entry"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, early.n_replies, + "no reply may be observable before the entry commits (Req 1.1)"); + TEST_ASSERT_FALSE_MESSAGE(job_exists_before_ack, + "the job must not exist in the in-memory table before commit"); + + TEST_ASSERT_TRUE_MESSAGE(final.n_replies > 0, + "a successful job submission must reply once the ordered apply " + "pump consumes the commit hand-off (raft-apply-decoupling task " + "6.2 -- closes the standing gap left open since task 3.2)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, final.last_status, + "a committed, applied job submission must reply success"); + TEST_ASSERT_EQUAL_INT_MESSAGE(RequestReply_data_jobid, + final.last_data_which, + "the success reply must carry the job id (RequestReply_data_jobid)"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(jobid, final.last_jobid, + "the success reply's job id must match the submitted job"); + + TEST_ASSERT_TRUE_MESSAGE(job_exists_at_reply, + "the job must be visible in the in-memory job table by the time " + "the success reply is observed"); + + TEST_ASSERT_TRUE_MESSAGE(g_jobsubmit_order_spy_fired, + "apply_committed_entry's post-mutation HOOK_DETAIL_JOB_NEW " + "notification must fire for this job submission"); + TEST_ASSERT_EQUAL_UINT64(jobid, g_jobsubmit_order_spy_last_jobid); + TEST_ASSERT_FALSE_MESSAGE(g_jobsubmit_order_spy_reply_seen_early, + "the client reply must be sent strictly AFTER the job's mutation " + "(apply_committed_entry's own post-mutation hook), never before or " + "concurrently with it -- task 6.2's own named observable " + "(Req 1.1, 2.1-2.3)"); + + printf("test_jobsubmit_success_reply_after_apply_carries_jobid: PASS\n"); +} + +/* Array submission (3 elements, base jobid 9210): a single covering ACK + * commits all three entries at once (mirroring test_apply_pump.c's + * burst-order coverage), and the pump applies + hands off each one + * independently, in ascending LSN order. Confirms task 3.2's per-element + * independent registration is correctly drained by task 6.2's per-element + * consumption: every element's own post-mutation notification fires before + * the single batch-level reply is ever observable, and every element is + * visible in c->jobmgr by the time that reply arrives. */ +static void test_jobsubmit_array_success_all_elements_apply_before_single_reply( + void) +{ + const uint64_t base_jobid = 9210; + const uint64_t array_size = 3; + uint64_t lsn_before, watermark; + cbrp_reply_obs_t final = { 0 }; + bool job_exists[3]; + + cbrp_setup(2000); + TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&g_hc.hooksys, XCE_CONTROLLER_HOOK_JOB, + cbrp_jobsubmit_order_spy, &g_hc)); + cbrp_jobsubmit_order_spy_reset(); + cbrp_arrange_jobsubmit(base_jobid, array_size); + + pthread_mutex_lock(&g_hc.persist.lock); + lsn_before = g_hc.persist.current_lsn; + pthread_mutex_unlock(&g_hc.persist.lock); + + g_op_arg.fn = handle_jobsubmit; + g_op_arg.req = &g_jobsub_req; + g_op_arg.rc = -9999; + TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, + cbrp_op_thread_fn, &g_op_arg)); + TEST_ASSERT_EQUAL_INT(0, pthread_join(g_op_tid, NULL)); + + /* All three elements append back-to-back before handle_jobsubmit + * returns (task 3.2) -- one covering ACK past the last of them commits + * the whole batch in a single jump. */ + watermark = cbrp_poll_append(&g_hc.persist, lsn_before + (array_size - 1), + 2000); + repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark); + cbrp_poll_reply(&g_req_ep, 2000, &final); + + job_exists[0] = cbrp_job_exists(base_jobid); + job_exists[1] = cbrp_job_exists(base_jobid + 1); + job_exists[2] = cbrp_job_exists(base_jobid + 2); + + cbrp_teardown(); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + (array_size - 1), watermark, + "all three array elements must append before this poll returns"); + + TEST_ASSERT_TRUE_MESSAGE(final.n_replies > 0, + "the batch must eventually reply once every element applies"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, final.n_replies, + "an all-succeeding array batch must produce exactly ONE reply for " + "the whole request, not one per element (Req 9.3)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, final.last_status, + "the batch reply must report success"); + TEST_ASSERT_EQUAL_INT_MESSAGE(RequestReply_data_jobid, + final.last_data_which, + "the batch success reply must carry a job id"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(base_jobid, final.last_jobid, + "the batch success reply must echo the BASE job id, matching " + "pre-task-3.1 semantics (git history commit 6f1fa97), never a " + "per-element id"); + + TEST_ASSERT_TRUE_MESSAGE(job_exists[0], + "array element 0 must be visible in c->jobmgr by reply time"); + TEST_ASSERT_TRUE_MESSAGE(job_exists[1], + "array element 1 must be visible in c->jobmgr by reply time"); + TEST_ASSERT_TRUE_MESSAGE(job_exists[2], + "array element 2 must be visible in c->jobmgr by reply time"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(3, g_jobsubmit_order_spy_fire_count, + "each of the three array elements must independently fire its own " + "post-mutation notification (task 3.2's per-element registration, " + "drained one at a time by task 6.2)"); + TEST_ASSERT_FALSE_MESSAGE(g_jobsubmit_order_spy_reply_seen_early, + "no array element's post-mutation notification may ever observe " + "the batch's single reply already enqueued -- the reply can only " + "follow the LAST element's own mutation, and every earlier " + "element's mutation must equally precede it"); + + printf("test_jobsubmit_array_success_all_elements_apply_before_single_" + "reply: PASS\n"); +} + /* REMOVED (raft-apply-decoupling task 3.2): the array-partial-construction- * failure remediation test (task 3.1 round-2 fix, bug-1408) that lived here * drove handle_jobsubmit's OLD sequential, per-element BLOCKING append loop @@ -2942,6 +3183,10 @@ void test_commit_barrier_reply(void) RUN_TEST(test_jobsubmit_repeated_timeout_failures_no_phantom_jobs); RUN_TEST(test_jobsubmit_repeated_truncation_forced_failures_no_phantom_jobs); RUN_TEST(test_jobsubmit_array_mixed_outcome_no_phantom_job_for_failed_elements); + /* handle_jobsubmit success path via the ordered apply pump + * (raft-apply-decoupling task 6.2, Req 1.1, 2.1-2.3) */ + RUN_TEST(test_jobsubmit_success_reply_after_apply_carries_jobid); + RUN_TEST(test_jobsubmit_array_success_all_elements_apply_before_single_reply); /* Receipt-ACK verdict pins (ha-commit-barrier-rollout task 3.3, Req 2.3) * — green on landing */ RUN_TEST(test_jobdispatch_receipt_ack_no_wal_append); diff --git a/src/include/persist.h b/src/include/persist.h index c98b6e6..f65d885 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -939,6 +939,29 @@ int persist_submit_job_payloads_async(persist_t* p, xlist_t* payloads, * during quorum loss (Req 6.1), not a leak — task 6.1 is expected to drain * and free each node as it walks committed LSNs. * + * bug-1415 self-healing (round 2): the caller (e.g. jobsubmit_commit_ + * continuation, dispatched off-thread via commit.c's dispatch_continuation) + * and the ordered apply pump's walk (persist_apply_pump_run(), dispatched + * off-thread via persist_apply_pump_kick()) are two INDEPENDENT vtasks + * enqueued back-to-back onto the same pool with no completion-order + * guarantee between them — with n_vcpus > 1 (the production default), the + * walk can apply `lsn` and advance last_applied_lsn PAST it before this + * function ever runs for that same `lsn`. A registration arriving that late + * would otherwise be permanently orphaned: the walk visits each lsn exactly + * once, so nothing would ever call persist_take_pending_apply() for it + * again, silently hanging the caller forever. This function guards against + * that by checking `lsn <= persist_last_applied_lsn(p)` BEFORE linking the + * node onto the list; if the walk has already passed `lsn`, `on_applied` is + * invoked immediately and synchronously, right here, on the CALLER's thread + * (never the apply-pump thread) instead of being linked for a walk pass + * that will never revisit it. This is race-safe in both interleavings: + * when the walk has NOT reached `lsn` yet, the check is false and + * registration proceeds exactly as before (unchanged behavior for the + * common, non-racing case); when the walk already passed `lsn`, the walk's + * own persist_take_pending_apply() call for that lsn already ran and found + * nothing (that is precisely how we know this registration is late), so + * invoking `on_applied` here can never double-fire alongside it. + * * NOTE for whoever builds task 6.1: apply_committed_entry()'s WAL_JOB_SUBMIT * case (persist.c) already constructs the job (via restore_job()) AND fires * the HOOK_DETAIL_JOB_NEW AFTER_EVENT notification, per entry, independently