diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index cc5cd30b..78cab8dc 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -143,7 +143,7 @@ - _Requirements: 1.1, 2.1, 2.2, 2.3_ - _Depends: 6.1, 3.2_ -- [ ] 6.3 Integration test command and fact apply/notify timing through the full pump +- [x] 6.3 Integration test command and fact apply/notify timing through the full pump - Cover a successful job submission becoming visible and notified only after commit, exercised through the full pump (superseding the earlier synchronous-path check from task 3.1) - Cover that worker slot release, scheduler dispatch triggering, and client/apiserver notification for a fact do not occur until that fact's entry is committed - Observable: the test suite shows zero fact-reaction side effects during a delayed-commit window, and a submitted job's reply always follows its visibility in the job table @@ -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.3**: pure integration-test close-out, approved first round, zero production changes — breaks task-group 6's 2-for-2 real-bug streak (6.1, 6.2 each found genuine production-severity bugs; 6.3 confirmed the pump's existing behavior is correct rather than finding a new one). New `test_apply_pump_integration.c` (2 tests) exercises the FULL live pump (real commit engine, real apply dispatch, real hook subscribers — `pool=NULL` like every other fixture, does not overclaim bug-1415/6.2's dispatch-race coverage, which is that test's own job) for both a job submission and a fact. **Key design decision, independently verified true by the reviewer, not just asserted**: `apply_committed_entry()` fires ALL job-state facts via the generic `HOOK_DETAIL_JOB_UPDATE` only (task 2.2's documented deviation, confirmed UNCHANGED by 6.1's `_pad` fix — that fix changed which state-computation core runs, not which hook detail fires) — so `sched_job_start_fun`/`_exit_fun`/`_fail_fun`/apiserver.c's equivalents are structurally unreachable for any live-applied fact, meaning ONE reaction spy wired to `HOOK_DETAIL_JOB_NEW`+`HOOK_DETAIL_JOB_UPDATE` legitimately covers all three named reactions (worker slot release, scheduler dispatch trigger, client/apiserver notify) since they're all downstream of the identical single firing. **Strong independent confirmation this is load-bearing, not incidental**: the type-specific handlers are unconditionally `va_arg`-typed for `job_pidinfo_t*`/`xce_jobfinish_t*`/`job_failinfo_t*` — types `apply_committed_entry()` never passes (always `job_t*`) — so if a type-specific hook ever DID fire from the live pump, it would be an immediate type-confusion crash, not just a coverage gap. Repo-wide re-audit (not just job.c/persist.c) confirms exactly 2 live call sites ever fire these hook details: `apply_committed_entry()` (guarded) and `job_ctrl`'s `mark_job_update` (legitimate SIGCONT/SIGSTOP) — no orphaned third site. Both tests prove REAL ordering, not coincidental timing: reply-after-visibility is sampled synchronously inside the reaction-spy callback (not two separately-timed checks); Req 4.3's "no separate watermark re-check needed" is proven by construction (the walk only ever invokes `apply_committed_entry()` for LSNs already `<= committed_lsn`, so `committed_lsn_at_fire >= entry_lsn` holds structurally, not by scheduling luck) rather than merely asserted in prose. Delayed-commit and genuine quorum-loss windows are both polled repeatedly across real time (5×20ms), not checked once. RED phase independently reproduced by both implementer and reviewer via re-injecting the pre-4.1 eager-mutation bug class. **Task-group 6 running tally**: 3 tasks in, 2 real production-severity bugs found and fixed (6.1 data corruption, 6.2 client-hang race), 1 clean confirmation (6.3) — the pump's core mechanics and hook-routing design are now thoroughly adversarially tested from multiple independent angles. Remaining: 6.4 (quorum-loss catch-up ordering, mixing facts AND commands — 6.3's fact-only quorum-loss scenario is explicitly non-duplicative defense-in-depth per the reviewer, not a preemption of 6.4's charter). - **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. diff --git a/src/controller/meson.build b/src/controller/meson.build index fd277e3e..2037dab8 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -163,6 +163,7 @@ controller_unit_test_src = [ 'testsuite/test_apply_reapply_idempotency.c', 'testsuite/test_apply_correctness_verification.c', 'testsuite/test_apply_pump.c', + 'testsuite/test_apply_pump_integration.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 69c7aa37..0c04f9a8 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -83,6 +83,7 @@ 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); +extern void test_apply_pump_integration(void); void setUp(void) { return; } @@ -235,6 +236,7 @@ int main(int argc, char* argv[]) TEST(test_apply_reapply_idempotency); TEST(test_apply_correctness_verification); TEST(test_apply_pump); + TEST(test_apply_pump_integration); return UNITY_END(); } diff --git a/src/controller/testsuite/test_apply_pump_integration.c b/src/controller/testsuite/test_apply_pump_integration.c new file mode 100644 index 00000000..4092df42 --- /dev/null +++ b/src/controller/testsuite/test_apply_pump_integration.c @@ -0,0 +1,763 @@ +/* test_apply_pump_integration.c — raft-apply-decoupling task 6.3 + * + * Integration coverage for command AND fact apply/notify timing through the + * SAME full pump (task 6.1's ordered walk + task 6.2's continuation + * hand-off), superseding the earlier synchronous-path check from task 3.1 + * (which predated the pump and could only pin "no phantom job", not + * "notified only after commit"). + * + * Two things this file does NOT re-prove, because they are already covered + * elsewhere in this exact spec's own suite: + * - Pure command reply-after-visibility ordering for job submit — + * test_commit_barrier_reply.c's test_jobsubmit_success_reply_after_ + * apply_carries_jobid (task 6.2) already drives this through the real + * pump end to end. + * - Pure fact no-live-mutation-before-commit for a single fact type — + * test_apply_pump.c (task 6.1) and test_commit_barrier_reply.c's + * test_fact_state_unchanged_throughout_quorum_loss_outage (task 4.2) + * already cover this for state, across a genuine outage window. + * + * What is genuinely new ground, and this file's actual job: + * 1. A command (job submit) and a fact (job-state report) interleaved + * through the ONE SAME apply-pump pass, proving commit-order + * application and independent gating hold when they are mixed, not + * just individually (task boundary: "this needs BOTH command and fact + * coverage through the SAME full pump"). + * 2. The actual REACTION hooks a real sched.c/apiserver.c subscriber + * would receive for a fact — not merely "did live state change" — + * proving worker slot release / scheduler dispatch triggering / + * client-apiserver notification (all downstream of the identical + * AFTER_EVENT firing, see below) never fire during a delayed-commit + * window, then DO fire once the entry commits and applies. + * 3. That every such firing carries proof, inline, that the entry was + * ALREADY committed at fire time (committed_lsn_at_fire >= the + * entry's own lsn) — the structural reason a consumer of the + * notification never needs to separately re-check a commit watermark + * (Req 4.3): the notification's mere occurrence already implies it. + * + * Why ONE reaction spy stands in for THREE named reactions (worker slot + * release, scheduler dispatch triggering, client/apiserver notification): + * apply_committed_entry() (persist.c) fires every job-state fact as the + * SAME generic HOOK_DETAIL_JOB_UPDATE — a real, documented deviation from a + * literal per-type hook fan-out (persist.h's apply_committed_entry doc + * comment, task 2.2), because the on-disk fact payload (wal_jobstate_t + * {jobid,state}) cannot supply what the type-specific hooks + * (HOOK_DETAIL_JOB_START/_EXIT/_FAIL/_RUSAGE_UPDATE) require. Concretely, + * this means sched.c's sched_job_update_fun / apiserver.c's + * apiserver_job_update_fun (both HOOK_DETAIL_JOB_UPDATE subscribers) are + * the ONLY reachable reaction path for a live-applied fact — sched_job_ + * start_fun/_exit_fun/_fail_fun and apiserver's equivalents are, by this + * same documented deviation, structurally unreachable for a fact applied + * through the live pump. A spy that listened for the type-specific details + * instead would silently test something that can never happen in + * production and would never catch a real timing violation. This file + * deliberately mirrors that reality (HOOK_DETAIL_JOB_NEW for submit, + * HOOK_DETAIL_JOB_UPDATE for every fact type) rather than an idealized + * per-type routing table the wire format cannot support. The explicit + * per-task note (persist.h, tasks.md's task 2.2 Implementation Notes) + * states no task through 6.3 closes this hook-granularity gap — this file + * verifies TIMING correctness of whatever hook actually fires, which is + * this task's boundary, not the routing granularity itself. + * + * "Did we miss a second call site" audit (task 6.1 already found one + * undocumented double-append/double-notify hazard in queue/worker hooks — + * this file's own pass, via `grep -n xce_run_hooks job.c persist.c` + * cross-referenced against HOOK_DETAIL_JOB_NEW/_UPDATE): the only AFTER_ + * EVENT firers for these two details are (a) apply_committed_entry() + * itself (persist.c, guarded against double-appending via the thread-local + * t_apply_seam_notifying flag) and (b) job_ctrl's mark_job_update + * (job.c:1268, SIGCONT/SIGSTOP live command path, legitimately outside the + * apply seam and already covered by test_apply_pump.c's guard-precision + * test). No orphaned third call site was found for either detail. + * + * Fixture mirrors test_apply_pump.c's tap_setup (real persist_init + + * persist_start over a private tmpdir, a real 3-member commit_engine_t with + * two stub standbys, ACKs delivered explicitly via repl_on_ack, commit_ + * set_pool never called so the apply pump's NULL-pool inline fallback + * applies) plus test_commit_barrier_reply.c's reply-observation harness + * (a zeroed, no-iomgr requester endpoint; xce_reply's packets stay queued + * on ep->outq and are parsed with the raw capnp reader). + * + * Both tests below follow this series' established disciplines: capture + * every assertion-relevant observable into locals, then teardown + * unconditionally, THEN assert (a failed TEST_ASSERT's longjmp must never + * skip teardown); poll for eventual apply-pump consistency rather than + * assume synchronous completion (persist_start's fsync_worker thread + * independently kicks the pump on every durable_lsn advance, racing this + * test's own repl_on_ack-driven kick — see tap_wait_applied's lineage in + * test_apply_pump.c for the full mechanism); and poll ACROSS a real time + * window to prove an absence, not just check once. + */ + +#include + +#include +#include +#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 "netio.h" +#include "persist.h" +#include "repl.h" +#include "request.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_tpi_jobobj_ready; + +static cluster_t g_tpi_c; +static commit_engine_t g_tpi_hce; +static int g_tpi_hep_a, g_tpi_hep_b; /* dummy standby endpoints */ +static char g_tpi_tmpdir[256]; + +/* Test-injectable send hook (repl.c, XCE_TESTING) — keeps repl_broadcast + * from dereferencing the dummy standby endpoints above. */ +extern int (*repl_broadcast_send_hook)(void *ep, wal_catchup_entry_t **entries, + int n_entries); + +static int tpi_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 */ +} + +/* ── Requester endpoint + reply observation (mirrors test_commit_barrier_ + * reply.c's cbrp harness — see that file's header for the full rationale: + * xce_reply()'s packets stay queued on a zeroed, no-iomgr endpoint's outq, + * parsed with the raw capnp reader). ── */ +static endpoint_t g_tpi_req_ep; +static ep_private_t g_tpi_req_priv; + +typedef struct { + int n_replies; + int32_t last_status; + int32_t last_data_which; + uint64_t last_jobid; +} tpi_reply_obs_t; + +static void tpi_scan_replies(endpoint_t *ep, tpi_reply_obs_t *obs) +{ + data_item_t *it; + + while ((it = endpoint_deq(ep)) != NULL) { + io_packet_t *pkt = ITPTR_TO_PACKET(it); + struct capn c; + struct Request rq; + Request_ptr rp; + + capn_init_mem(&c, (const uint8_t *)pkt->data, pkt->sz, 0); + rp.p = capn_getp(capn_root(&c), 0, 1); + read_Request(&rq, rp); + if (rq.body_which == Request_body_replymsg) { + struct RequestReply rr; + 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; + } + capn_free(&c); + free_packet(pkt); + free(pkt); + } +} + +/* ── Reaction spy: stands in for sched.c's sched_job_update_fun / + * apiserver.c's apiserver_job_update_fun (the real, only-reachable-for- + * facts subscribers — see file header). Captures, at the EXACT moment the + * AFTER_EVENT fires: which detail, which job, the job's live state (for + * HOOK_DETAIL_JOB_UPDATE), committed_lsn sampled at that instant (the + * no-watermark-recheck-needed proof), and whether the job-submit reply was + * already enqueued (the reply-never-precedes-notification order proof, + * mirroring test_commit_barrier_reply.c's cbrp_jobsubmit_order_spy). ── */ +#define TPI_SPY_CAP 16 +typedef struct { + int detail; + uint64_t jobid; + uint32_t state_at_fire; + uint64_t committed_lsn_at_fire; + bool reply_already_enqueued; +} tpi_spy_entry_t; + +static tpi_spy_entry_t g_tpi_spy[TPI_SPY_CAP]; +static int g_tpi_spy_n; + +static int tpi_reaction_spy(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; + uint32_t state = 0; + + switch (detail) { + case HOOK_DETAIL_JOB_NEW: { + /* args: cluster_t*, xlist_t* (apply_committed_entry's WAL_JOB_SUBMIT + * branch) — mirrors test_apply_pump.c's tap_spy_hook decode. */ + (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; + state = j->job.state; + } + break; + } + case HOOK_DETAIL_JOB_UPDATE: { + /* args: cluster_t*, job_t* — the ONLY detail a live-applied fact of + * any type (START/FINISH/FAIL/UPDATE/...) ever fires, per the + * documented generic-routing deviation (see file header). */ + (void)va_arg(ap, cluster_t *); + job_t *j = va_arg(ap, job_t *); + if (j) { + jobid = j->job.jobid; + state = j->job.state; + } + break; + } + default: + /* HOOK_DETAIL_JOB_START/_EXIT/_FAIL/_RUSAGE_UPDATE: structurally + * unreachable for a live-applied fact (see file header) — not + * relevant to this file's assertions. */ + return 0; + } + + if (g_tpi_spy_n < TPI_SPY_CAP) { + g_tpi_spy[g_tpi_spy_n].detail = detail; + g_tpi_spy[g_tpi_spy_n].jobid = jobid; + g_tpi_spy[g_tpi_spy_n].state_at_fire = state; + g_tpi_spy[g_tpi_spy_n].committed_lsn_at_fire = + atomic_load(&g_tpi_c.persist.committed_lsn); + g_tpi_spy[g_tpi_spy_n].reply_already_enqueued = + !DATAQ_EMPTY(&g_tpi_req_ep.outq); + g_tpi_spy_n++; + } + return 0; +} + +static void tpi_spy_reset(void) +{ + g_tpi_spy_n = 0; + memset(g_tpi_spy, 0, sizeof(g_tpi_spy)); +} + +static void tpi_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 *)"tpi-probe") == NULL) + xstring_table_init(64); + + if (!g_tpi_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_tpi_jobobj_ready = 1; + } + + memset(&g_tpi_c, 0, sizeof(g_tpi_c)); + g_tpi_c.myhost = &g_tpi_c.myhost_info; + g_tpi_c.myhost_info.name = (char *)"ctrl-test"; + g_tpi_c.myhost_info.key = (char *)""; + + memset(&g_tpi_req_ep, 0, sizeof(g_tpi_req_ep)); + memset(&g_tpi_req_priv, 0, sizeof(g_tpi_req_priv)); + g_tpi_req_ep.fd = -1; + g_tpi_req_ep.data = &g_tpi_req_priv; + + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_tpi_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_tpi_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_tpi_c.qmgr = makeInstance(&xkvmgr_type, &str_params); + g_tpi_c.workermgr = makeInstance(&xkvmgr_type, &str_params); + g_tpi_c.resmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_tpi_c.jobmgr); + TEST_ASSERT_NOT_NULL(g_tpi_c.qmgr); + TEST_ASSERT_NOT_NULL(g_tpi_c.workermgr); + TEST_ASSERT_NOT_NULL(g_tpi_c.resmgr); + zarray_init(&g_tpi_c.sharedinst, 16, NULL); + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.sync_timeout_ms = 2000; + + snprintf(g_tpi_tmpdir, sizeof(g_tpi_tmpdir), "/tmp/xce_tpi63_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_tpi_tmpdir)); + persist_set_workdir(&g_tpi_c.persist, g_tpi_tmpdir, "ctrl"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&g_tpi_c.persist, &cfg, &g_tpi_c)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&g_tpi_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 -- ce->pool stays NULL, matching every + * other fixture in this spec (see test_apply_pump.c's file header). */ + TEST_ASSERT_TRUE( + repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a, "sb_a", 7700) >= 0); + TEST_ASSERT_TRUE( + repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b, "sb_b", 7701) >= 0); + TEST_ASSERT_EQUAL_INT(0, + commit_engine_init(&g_tpi_hce, &g_tpi_c.persist, 3, 8)); + repl_set_commit_engine(&g_tpi_c.persist.repl, &g_tpi_hce); + repl_broadcast_send_hook = tpi_drop_send_hook; + + TEST_ASSERT_EQUAL_INT(0, + xce_add_hook(&g_tpi_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tpi_reaction_spy, + &g_tpi_c)); + + tpi_spy_reset(); +} + +static void tpi_teardown(void) +{ + tpi_reply_obs_t drain = { 0 }; + + tpi_scan_replies(&g_tpi_req_ep, &drain); /* free any leftover packets */ + repl_broadcast_send_hook = NULL; + commit_engine_fini(&g_tpi_hce); + persist_stop(&g_tpi_c.persist); + persist_fini(&g_tpi_c.persist); + + zarray_fini(&g_tpi_c.sharedinst); + if (g_tpi_c.resmgr) dereference(g_tpi_c.resmgr); + if (g_tpi_c.workermgr) dereference(g_tpi_c.workermgr); + if (g_tpi_c.qmgr) dereference(g_tpi_c.qmgr); + if (g_tpi_c.jobmgr) dereference(g_tpi_c.jobmgr); + xce_hook_fini(&g_tpi_c.hooksys); +} + +/* Creates a job directly in the fixture's jobmgr, bypassing WAL entirely — + * for a pre-existing job that then receives a worker-reported fact. */ +static job_submit_data_t g_tpi_seed_submit; + +static job_t *tpi_seed_job(uint64_t jobid, uint32_t state) +{ + xce_job_t jdef; + + memset(&g_tpi_seed_submit, 0, sizeof(g_tpi_seed_submit)); + g_tpi_seed_submit.queue = (char *)"q"; + g_tpi_seed_submit.user = (char *)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.submit = &g_tpi_seed_submit; + + return makeJob(g_tpi_c.jobmgr, &jdef); +} + +static job_t *tpi_find_job(uint64_t jobid) +{ + job_t *j = NULL; + lock(g_tpi_c.jobmgr, RTTI_LOCK_READ); + int found = key_find(g_tpi_c.jobmgr, (void *)jobid, (void **)&j); + unlock(g_tpi_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) — poll briefly rather than assume the + * flusher thread has already caught up, mirroring test_apply_pump.c's + * tap_wait_durable. */ +static void tpi_wait_durable(uint64_t lsn, uint32_t deadline_ms) +{ + for (uint32_t waited = 0; waited < deadline_ms; waited += 5) { + if (atomic_load(&g_tpi_c.persist.durable_lsn) >= lsn) + return; + usleep(5000); + } +} + +/* Waits for the ordered apply pump to actually finish applying up through + * `lsn` — needed even with ce->pool == NULL, since persist_start's + * fsync_worker thread independently kicks the pump on every durable_lsn + * advance, racing this test's own repl_on_ack-driven kick (see test_apply_ + * pump.c's tap_wait_applied doc comment for the full mechanism and the + * flake it fixes — this is that exact same lesson, reapplied here). */ +static void tpi_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_tpi_c.persist) >= lsn) + return; + usleep(5000); + } +} + +static void tpi_sleep_ms(uint32_t ms) +{ + struct timespec ts; + ts.tv_sec = ms / 1000u; + ts.tv_nsec = (long)(ms % 1000u) * 1000000L; + nanosleep(&ts, NULL); +} + +/* ── Job submit request builder (mirrors test_commit_barrier_reply.c's + * cbrp_arrange_jobsubmit — single-element, non-array submission). ── */ +static job_startinfo_t g_tpi_jobsub_startinfo; +static job_submit_data_t g_tpi_jobsub_data; +static xce_job_submit_t g_tpi_jobsub_msg; +static xce_request_t g_tpi_jobsub_req; +static xce_peerinfo_t g_tpi_jobsub_from; + +static void tpi_arrange_jobsubmit(uint64_t jobid) +{ + memset(&g_tpi_jobsub_startinfo, 0, sizeof(g_tpi_jobsub_startinfo)); + g_tpi_jobsub_startinfo.cmdline = (char *)"/bin/true"; + + memset(&g_tpi_jobsub_data, 0, sizeof(g_tpi_jobsub_data)); + g_tpi_jobsub_data.queue = (char *)"normal"; + g_tpi_jobsub_data.user = (char *)"tester"; + g_tpi_jobsub_data.start_info = &g_tpi_jobsub_startinfo; + + memset(&g_tpi_jobsub_msg, 0, sizeof(g_tpi_jobsub_msg)); + g_tpi_jobsub_msg.jobid = jobid; + g_tpi_jobsub_msg.array_size = 1; + g_tpi_jobsub_msg.submit = &g_tpi_jobsub_data; + + g_tpi_jobsub_from.name = (char *)"requester"; + g_tpi_jobsub_from.key = (char *)""; + + memset(&g_tpi_jobsub_req, 0, sizeof(g_tpi_jobsub_req)); + g_tpi_jobsub_req.opcode = OPCode_jobsub; + g_tpi_jobsub_req.seq = 21; + g_tpi_jobsub_req.from = &g_tpi_jobsub_from; + g_tpi_jobsub_req.body_which = Request_body_jobsubmsg; + g_tpi_jobsub_req.body.jobsubmsg = &g_tpi_jobsub_msg; +} + +/* ═══════════════════════════════════════════════════════════════════════ + * Test 1 — command (job submit) and fact (job start) interleaved through + * the SAME apply-pump pass and the SAME covering commit. + * ═══════════════════════════════════════════════════════════════════════ + * + * Job X is submitted (command, task 3.2's continuation path); job Y + * pre-exists and receives a real worker-reported START fact (task 4.1's + * durable-append-only receipt path) — appended in that order, so X's + * submit entry has the LOWER lsn. A single covering ACK commits both at + * once; the pump must apply them in strict LSN order (X then Y) and gate + * BOTH reactions on that same commit — proving the mixed-type ordering and + * independent per-entry gating this task's boundary explicitly asks for, + * not just two separately-proven single-type cases. */ +static void test_pump_command_and_fact_interleave_through_same_commit(void) +{ + const uint64_t jobid_x = 9500; /* submitted via handle_jobsubmit */ + const uint64_t jobid_y = 9501; /* pre-existing, receives a START fact */ + + tpi_setup(); + + job_t *jy = tpi_seed_job(jobid_y, JOB_STAT_PEND); + int seeded_ok = (jy != NULL); + + pthread_mutex_lock(&g_tpi_c.persist.lock); + uint64_t lsn_before = g_tpi_c.persist.current_lsn; + pthread_mutex_unlock(&g_tpi_c.persist.lock); + + tpi_arrange_jobsubmit(jobid_x); + int submit_rc = handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req); + + static unsigned int y_pids[1] = { 7001 }; + job_pidinfo_t y_start = { .jobid = jobid_y, + .npids = 1, + .pids = &y_pids[0] }; + int start_rc = on_job_start(&g_tpi_c, &y_start); + + pthread_mutex_lock(&g_tpi_c.persist.lock); + uint64_t watermark = g_tpi_c.persist.current_lsn - 1; + pthread_mutex_unlock(&g_tpi_c.persist.lock); + uint64_t x_lsn = lsn_before; /* first append after lsn_before */ + uint64_t y_lsn = lsn_before + 1; /* second append */ + + /* ── Delayed-commit window: poll repeatedly across real time, proving + * an absence, not just checking once. Neither reaction may fire, no + * reply may be observable, job X must not exist, job Y's state must + * stay PEND. ── */ + enum { WINDOW_POLLS = 5, WINDOW_INTERVAL_MS = 20 }; + bool window_job_x_absent[WINDOW_POLLS]; + uint32_t window_jy_state[WINDOW_POLLS]; + int window_spy_n[WINDOW_POLLS]; + int window_replies[WINDOW_POLLS]; + + for (int i = 0; i < WINDOW_POLLS; i++) { + tpi_sleep_ms(WINDOW_INTERVAL_MS); + window_job_x_absent[i] = (tpi_find_job(jobid_x) == NULL); + window_jy_state[i] = seeded_ok ? jy->job.state : 0; + window_spy_n[i] = g_tpi_spy_n; + tpi_reply_obs_t obs = { 0 }; + tpi_scan_replies(&g_tpi_req_ep, &obs); + window_replies[i] = obs.n_replies; + } + + /* ── Deliver ONE covering ACK past both entries, then wait for the + * pump to actually finish applying (poll, per file header). ── */ + tpi_wait_durable(watermark, 2000); + repl_on_ack(&g_tpi_c.persist.repl, &g_tpi_hep_a, watermark); + tpi_wait_applied(watermark, 2000); + + tpi_reply_obs_t final_obs = { 0 }; + tpi_scan_replies(&g_tpi_req_ep, &final_obs); + + uint64_t committed_final = atomic_load(&g_tpi_c.persist.committed_lsn); + job_t *jx_final = tpi_find_job(jobid_x); + uint32_t jy_state_final = seeded_ok ? jy->job.state : 0; + int spy_n_final = g_tpi_spy_n; + tpi_spy_entry_t spy_snap[TPI_SPY_CAP]; + memcpy(spy_snap, g_tpi_spy, sizeof(spy_snap)); + + tpi_teardown(); + + /* Arrangement sanity */ + TEST_ASSERT_TRUE(seeded_ok); + TEST_ASSERT_EQUAL_INT(0, submit_rc); + TEST_ASSERT_EQUAL_INT(0, start_rc); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, watermark, + "exactly two entries (X's submit, Y's start) must append"); + + /* Delayed-commit window: zero side effects throughout (Req 5.3, 6.1) */ + for (int i = 0; i < WINDOW_POLLS; i++) { + TEST_ASSERT_TRUE_MESSAGE(window_job_x_absent[i], + "job X must not exist in the job table before its submit " + "entry commits, at every poll of the delayed-commit window " + "(Req 2.1)"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, window_jy_state[i], + "job Y's fact must not mutate live state before its entry " + "commits, at every poll of the delayed-commit window (Req 5.3)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, window_spy_n[i], + "no reaction (worker slot release / scheduler dispatch " + "trigger / client-apiserver notification) may fire for " + "either entry before its commit, at every poll of the " + "delayed-commit window (Req 5.3)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, window_replies[i], + "no client reply may be observable before job X's submit " + "entry commits (Req 2.2)"); + } + + /* After the covering ACK: both entries committed and applied */ + TEST_ASSERT_TRUE_MESSAGE(committed_final >= watermark, + "the covering ACK must advance committed_lsn over both entries"); + TEST_ASSERT_NOT_NULL_MESSAGE(jx_final, + "job X must be visible in the job table once its entry commits " + "and applies"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, jy_state_final, + "job Y must reflect its applied START fact"); + + /* Strict LSN/commit order across the MIXED command+fact pair, through + * the SAME pump pass — the actual new ground this test exists for. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(2, spy_n_final, + "exactly two reactions must fire: one for X's submit, one for " + "Y's fact"); + TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_NEW, spy_snap[0].detail, + "the LOWER-lsn entry (X's submit) must apply and notify FIRST, " + "even though it is a command and the second entry is a fact"); + TEST_ASSERT_EQUAL_UINT64(jobid_x, spy_snap[0].jobid); + TEST_ASSERT_EQUAL_INT_MESSAGE(HOOK_DETAIL_JOB_UPDATE, spy_snap[1].detail, + "the HIGHER-lsn entry (Y's fact) must apply and notify SECOND"); + TEST_ASSERT_EQUAL_UINT64(jobid_y, spy_snap[1].jobid); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, spy_snap[1].state_at_fire, + "Y's reaction must observe the ALREADY-applied RUN state — a " + "consumer never sees the notification before the mutation"); + + /* No-watermark-recheck-needed proof (Req 4.3): at the exact instant + * EACH reaction fired, committed_lsn already covered that entry. */ + TEST_ASSERT_TRUE_MESSAGE( + spy_snap[0].committed_lsn_at_fire >= x_lsn, + "X's reaction must fire only once committed_lsn already covers " + "X's own lsn -- no separate watermark check is ever needed"); + TEST_ASSERT_TRUE_MESSAGE( + spy_snap[1].committed_lsn_at_fire >= y_lsn, + "Y's reaction must fire only once committed_lsn already covers " + "Y's own lsn -- no separate watermark check is ever needed"); + + /* The client reply for X must never precede X's own reaction (the + * mutation+notify), mirroring test_commit_barrier_reply.c's task 6.2 + * ordering proof, now re-verified alongside an interleaved fact. */ + TEST_ASSERT_FALSE_MESSAGE(spy_snap[0].reply_already_enqueued, + "job X's client reply must be sent strictly AFTER X's own " + "post-mutation notification, never before or concurrently (Req " + "2.2)"); + TEST_ASSERT_TRUE_MESSAGE(final_obs.n_replies > 0, + "job X must eventually receive its success reply"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, final_obs.last_status, + "job X's reply must report success"); + TEST_ASSERT_EQUAL_INT_MESSAGE(RequestReply_data_jobid, + final_obs.last_data_which, + "job X's success reply must carry its job id"); + TEST_ASSERT_EQUAL_UINT64(jobid_x, final_obs.last_jobid); + + printf("test_pump_command_and_fact_interleave_through_same_commit: " + "PASS\n"); +} + +/* ═══════════════════════════════════════════════════════════════════════ + * Test 2 — multiple fact types, genuine quorum-loss-then-restore window: + * zero reactions throughout the outage, correct reactions after restore. + * ═══════════════════════════════════════════════════════════════════════ + * + * Extends test_commit_barrier_reply.c's test_fact_state_unchanged_ + * throughout_quorum_loss_outage (task 4.2), which proves no-mutation + * during a genuine outage but never restores quorum to check what happens + * next. This test drives the SAME kind of outage (repl_remove_standby, a + * structurally-cannot-commit window) through START then FINISH for one + * job, polls repeatedly for zero reactions during the outage, then + * reconnects, delivers the covering ACK, and confirms both reactions fire + * afterward, in order, each already-committed at fire time. */ +static void test_pump_fact_reactions_withheld_during_outage_then_fire_after_restore( + void) +{ + const uint64_t jobid_z = 9600; + + tpi_setup(); + + job_t *jz = tpi_seed_job(jobid_z, JOB_STAT_PEND); + int seeded_ok = (jz != NULL); + + /* Genuine quorum loss: self alone is 1 of 3, never a majority, so + * committed_lsn structurally cannot advance past this point (same + * mechanism as test_fact_state_unchanged_throughout_quorum_loss_ + * outage). */ + repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a); + repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b); + + pthread_mutex_lock(&g_tpi_c.persist.lock); + uint64_t lsn_before = g_tpi_c.persist.current_lsn; + pthread_mutex_unlock(&g_tpi_c.persist.lock); + + static unsigned int z_pids[1] = { 7101 }; + job_pidinfo_t z_start = { .jobid = jobid_z, + .npids = 1, + .pids = &z_pids[0] }; + int start_rc = on_job_start(&g_tpi_c, &z_start); + + xce_jobfinish_t z_finish = { .jobid = jobid_z, .exit_code = 0 }; + int finish_rc = on_job_finish(&g_tpi_c, &z_finish); + + pthread_mutex_lock(&g_tpi_c.persist.lock); + uint64_t watermark = g_tpi_c.persist.current_lsn - 1; + pthread_mutex_unlock(&g_tpi_c.persist.lock); + uint64_t start_lsn = lsn_before; + uint64_t finish_lsn = lsn_before + 1; + + /* ── Outage window: poll repeatedly across real time. Both facts are + * durably appended and pending, but structurally cannot commit while + * both standbys remain disconnected. ── */ + enum { OUTAGE_POLLS = 5, OUTAGE_INTERVAL_MS = 20 }; + uint32_t outage_state[OUTAGE_POLLS]; + int outage_spy_n[OUTAGE_POLLS]; + uint64_t outage_committed[OUTAGE_POLLS]; + + for (int i = 0; i < OUTAGE_POLLS; i++) { + tpi_sleep_ms(OUTAGE_INTERVAL_MS); + outage_state[i] = seeded_ok ? jz->job.state : 0; + outage_spy_n[i] = g_tpi_spy_n; + outage_committed[i] = atomic_load(&g_tpi_c.persist.committed_lsn); + } + + /* ── Restore quorum and deliver the covering ACK. ── */ + TEST_ASSERT_TRUE( + repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a, "sb_a", 7700) >= 0); + TEST_ASSERT_TRUE( + repl_add_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b, "sb_b", 7701) >= 0); + tpi_wait_durable(watermark, 2000); + repl_on_ack(&g_tpi_c.persist.repl, &g_tpi_hep_a, watermark); + tpi_wait_applied(watermark, 2000); + + uint64_t committed_final = atomic_load(&g_tpi_c.persist.committed_lsn); + uint32_t state_final = seeded_ok ? jz->job.state : 0; + int spy_n_final = g_tpi_spy_n; + tpi_spy_entry_t spy_snap[TPI_SPY_CAP]; + memcpy(spy_snap, g_tpi_spy, sizeof(spy_snap)); + + tpi_teardown(); + + TEST_ASSERT_TRUE(seeded_ok); + TEST_ASSERT_EQUAL_INT(0, start_rc); + TEST_ASSERT_EQUAL_INT(0, finish_rc); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, watermark, + "both facts (START, FINISH) must durably append during the " + "outage"); + + for (int i = 0; i < OUTAGE_POLLS; i++) { + TEST_ASSERT_TRUE_MESSAGE(outage_committed[i] < start_lsn, + "the outage must still be genuinely in effect at this poll " + "-- committed_lsn must not have advanced"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, outage_state[i], + "job Z's live state must remain unchanged at every point " + "during the outage (Req 6.1, 6.2) -- no local-apply escape " + "hatch exists"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, outage_spy_n[i], + "no reaction may fire for either fact while quorum is " + "unreachable, at every poll of the outage window (Req 5.3, " + "6.1, 6.2)"); + } + + TEST_ASSERT_TRUE_MESSAGE(committed_final >= watermark, + "a covering ACK after reconnect must advance committed_lsn over " + "the retained fact suffix"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, state_final, + "job Z must reflect both applied facts (START then FINISH) once " + "quorum is restored and the entries commit"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(2, spy_n_final, + "exactly two reactions must fire once quorum is restored -- one " + "per retained fact, none lost, none duplicated"); + TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_UPDATE, spy_snap[0].detail); + TEST_ASSERT_EQUAL_UINT64(jobid_z, spy_snap[0].jobid); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, spy_snap[0].state_at_fire, + "the START fact (lower lsn) must apply and notify FIRST, " + "reflecting RUN -- retained entries commit and apply in their " + "original log order (Req 6.3)"); + TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_UPDATE, spy_snap[1].detail); + TEST_ASSERT_EQUAL_UINT64(jobid_z, spy_snap[1].jobid); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, spy_snap[1].state_at_fire, + "the FINISH fact (higher lsn) must apply and notify SECOND, " + "reflecting the now-terminal DONE state"); + + TEST_ASSERT_TRUE_MESSAGE( + spy_snap[0].committed_lsn_at_fire >= start_lsn, + "the START reaction must fire only once committed_lsn already " + "covers its own lsn -- no separate watermark check is ever " + "needed (Req 4.3)"); + TEST_ASSERT_TRUE_MESSAGE( + spy_snap[1].committed_lsn_at_fire >= finish_lsn, + "the FINISH reaction must fire only once committed_lsn already " + "covers its own lsn -- no separate watermark check is ever " + "needed (Req 4.3)"); + + printf("test_pump_fact_reactions_withheld_during_outage_then_fire_" + "after_restore: PASS\n"); +} + +/* ── Test group entry point ───────────────────────────────────────────── */ + +void test_apply_pump_integration(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_pump_command_and_fact_interleave_through_same_commit); + RUN_TEST( + test_pump_fact_reactions_withheld_during_outage_then_fire_after_restore); + UNITY_END(); +}