From c005d40741a9852a07ff1fa468ebfd643d6caf54 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 4 Jul 2026 11:32:21 +0000 Subject: [PATCH] feat(raft-apply-decoupling): post-replay apply-correctness verification (task 5.1) Add apply_verdict_t (APPLY_CLEAN/APPLY_DIVERGENT) computed alongside the existing structural (CRC) recover_verdict_t. Dual-check design: an entity-count check (WAL_JOB_SUBMIT entries scanned vs live jobmgr growth, genuinely independent of apply_committed_entry()'s return code) and a position check against a new last_applied_lsn field (forward-looking infrastructure for task 6's live pump). Fixes a real bug found during review: the position check's expectation was computed from the physical WAL-scan tip (current_lsn - 1), which advances past checkpoint markers and unknown entry types that never reach apply_committed_entry(). A crash between persist_checkpoint()'s marker-fsync step and its later snapshot-persist step leaves a trailing checkpoint marker on next boot, which falsely reported APPLY_DIVERGENT on a perfectly clean recovery -- reachable today, not gated behind a future task. Fixed by reading last_applied_lsn directly, which only advances for entries actually dispatched through the apply seam. Known accepted gap: entity-count check covers jobs only, not workers (which can genuinely be created via WAL_WORKER_CTRL) -- documented for future extension. --- .kiro/specs/raft-apply-decoupling/tasks.md | 3 +- src/controller/meson.build | 1 + src/controller/persist.c | 181 +++++- src/controller/testsuite/controller_test.c | 2 + .../test_apply_correctness_verification.c | 517 ++++++++++++++++++ src/include/persist.h | 99 ++++ 6 files changed, 800 insertions(+), 3 deletions(-) create mode 100644 src/controller/testsuite/test_apply_correctness_verification.c diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index 99c5dee..2c390a2 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -107,7 +107,7 @@ - _Depends: 4.1_ - [ ] 5. Restart apply-correctness verification -- [ ] 5.1 (P) Compute a post-replay expectation from the committed log tip and compare it against materialized state +- [x] 5.1 (P) Compute a post-replay expectation from the committed log tip and compare it against materialized state - Derive a lightweight expectation (e.g., last-applied position and a count or checksum of materialized entities) from the recovered log - Compare the expectation against the actual state produced by replay through the shared apply dispatch function - Observable: a clean replay produces a CLEAN verdict alongside the existing structural (CRC) verdict @@ -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**. +- **5.1**: added a new `apply_verdict_t` (`APPLY_CLEAN`/`APPLY_DIVERGENT`) computed alongside the EXISTING, untouched structural `recover_verdict_t` (CRC) verdict — a dual-check design: (1) an entity-count check comparing `WAL_JOB_SUBMIT` entries scanned during replay against the live `c->jobmgr` table's actual growth (`count_jobmgr_entries()`), genuinely independent since it queries live state rather than trusting `apply_committed_entry()`'s return code — verified real by both review rounds via a forced job-count-mismatch test; (2) a position check comparing an "expected" LSN against `p->last_applied_lsn` (new `_Atomic uint64_t` field, per design.md, forward-looking infrastructure for task 6's live pump). Took 2 review rounds. **Round 1 REJECTED — real, live-today bug, not documentation**: the position check's "expected" value was computed as `recovered_tip = current_lsn - 1` (the PHYSICAL WAL-scan tip, which `replay_wal()` advances for every entry including `WAL_CHECKPOINT` markers and unknown types) instead of the LOGICAL apply cursor. `persist_checkpoint()` durably fsyncs the checkpoint marker in one step, then persists the advanced `checkpoint_lsn` to the on-disk snapshot in a LATER step — a crash between those two steps (a real, unavoidable crash-safety window this codebase already engineers hard for, cf. bug-661/ha-commit-durability) leaves a trailing checkpoint marker as the highest-LSN entry on next boot, causing `recovered_tip` to exceed the true `last_applied_lsn` and falsely report `APPLY_DIVERGENT` on a perfectly clean recovery — a direct Req 8.2 violation, reachable TODAY on any node that crashes at exactly the wrong moment during a checkpoint. **Round 2 fixed** by reading `p->last_applied_lsn` directly (the value `replay_wal()` already correctly advances only for entries actually dispatched through `apply_committed_entry()`, skipping checkpoint-marker/unknown-type entries) instead of `recovered_tip`; `recovered_tip` itself is untouched and still correctly feeds `durable_lsn`, which does need the physical tip. New regression test genuinely reproduces the trigger condition (real `persist_checkpoint()`, then `unlink()`s the exact snapshot file it wrote, landing on-disk state in the identical place a real crash would) — both review rounds independently reverted the fix and reproduced the false-DIVERGENT failure before re-confirming GREEN. **Known, disclosed, accepted scope gap (not a defect, flagged for whoever extends this check)**: the entity-count check covers jobs only. Queues are correctly excluded (config-derived, `WAL_QUEUE_CTRL` never creates a new one, verified by both review rounds). **Workers are NOT covered but CAN diverge**: `worker_from_capnp()`/`restore_worker()` genuinely creates a new worker on a `WAL_WORKER_CTRL`/`HOOK_DETAIL_WORKER_JOIN` entry when absent from `workermgr` — a worker-creation divergence is currently invisible to this verification. Extend `persist_verify_apply_correctness()` with a worker-count check (mirroring `count_jobmgr_entries()`) if/when this needs to be closed. **Self-disclosed, accepted limitation**: because `replay_wal()`'s single-threaded caller in `persist_recover()` derives both the "expected" and "actual" position values from the same atomic within moments of each other on the same thread, the position check is diagnostically tautological FOR THIS SPECIFIC CALL SITE today (though the entity-count half remains genuinely independent) — this will change once task 6's live apply pump gives expectation and actual state a real chance to diverge across threads/time; task 5.1's boundary does not require solving that now. - **4.2**: pure test-only close-out of task-group 4 — one new test `test_fact_state_unchanged_throughout_quorum_loss_outage` in `test_commit_barrier_reply.c` (30→31), approved first round, zero production changes. Drives a genuine `repl_remove_standby`-based quorum-loss window through all 4 real fact-receipt handlers, polls live job state + `committed_lsn` at 5 time-spread points (real 20ms sleeps, ~100ms total) rather than once, asserts a real hook-spy never sees `AFTER_EVENT` fire, and confirms durable WAL presence via direct on-disk segment read (`cbrp_count_jobstate_entries`) — combining and strengthening two pre-existing tests that each covered only half (4.1's own single-point-in-time check under normal conditions; an older pre-4.1 test that proved WAL durability but never inspected live state). **Static/inspection half of Req 6.2 confirmed independently by the reviewer, re-reading 4.1's landed diff**: none of the four `record_*_fact` functions call `apply_committed_entry`/`restore_job`/mutate the real job object — `record_job_fail_fact` specifically operates on a throwaway `job_t scratch = *job` shallow copy. **Architecturally significant finding for task 6 planning**: facts have NO commit-timeout/failure concept at the receipt layer, unlike commands (task 3.2's `commit_submit_async`). `persist_submit_job_fact()` → `wal_hook_append_classified(WAL_MODE_RETAIN)` → `commit_submit(..., COMMIT_MODE_RETAIN, 0)`, and `COMMIT_MODE_RETAIN`'s branch in commit.c is an immediate, unconditional return with NO continuation registered (explicit comment: "RETAIN never honors the timeout") — `commit_resolve()` never fires for a retained fact. A fact's outage is therefore open-ended by construction; its eventual commit is discovered only when `commit_advance_locked()`/`commit_on_ack` notices `committed_lsn` can advance past it via the ACK-registry watermarks, purely. **Task 6's apply pump must rely entirely on `committed_lsn` advancement to discover newly-eligible retained facts — there is no per-fact signal to wait on.** Confirmed neither of 4.1's two flagged hazards (double-append, merge-staleness) is triggered by this test, since it never drives apply. **Task-group 4 (fact apply-after-commit split) is now complete**: 4.1 (receipt converted to durable-append-only, two critical hazards discovered and documented for task 6) → 4.2 (retention proven under sustained quorum loss, fact-vs-command architectural asymmetry documented). **Both task-groups 1-4 are now done; task 6 (ordered apply pump) is the critical next milestone that must resolve, before going live: the double-append hazard, the merge-staleness hazard, AND design around facts' lack of a per-entry commit signal (commit_advance/committed_lsn-driven discovery only).** Groups 5 (restart verification) and 6 (apply pump) remain; 5.1 depends only on 2.3 (already done) so could run in parallel with 6, per tasks.md's dependency graph — continuing in numeric order (5 next) unless reprioritized. - **4.1**: converted the four worker-fact receipt handlers (`on_job_start`, `on_job_finish`, `on_job_fail`, `on_job_postdone` — NOT literally named `on_job_update`, task text's naming was approximate) to validate-and-durably-append-only: BEFORE_EVENT validation hook unchanged, AFTER_EVENT (post-mutation) hook no longer fires at receipt, live state mutation removed. **Key discovery**: the durable WAL append was NOT a separate call — it lived INSIDE the very `persist_job_hook_fun` AFTER_EVENT subscriber this task needed to silence, so a new `persist_submit_job_fact()` (persist.c) was added that calls the same unmodified `wal_hook_append_classified(WAL_MODE_RETAIN, wal_jobstate_t{jobid,state}, ...)` directly — verified wire-format-identical to before, `apply_committed_entry()`'s existing switch (task 2.2) routes it unchanged. `mark_job_start`/`mark_job_finish` (single caller each, verified via diff against 41d3620) renamed in place to `record_job_start_fact`/`record_job_finish_fact`; `mark_job_fail`/`mark_job_update` (shared with `job_ctrl`'s OUT-OF-BOUNDARY client-command path, 2 and 3 additional call sites) left untouched, with new durable-append-only siblings `record_job_fail_fact`/`record_job_update_fact` added instead — `job_ctrl`/`do_jobcontrol_op` confirmed byte-identical to pre-task by the reviewer. Approved first round. **Real, permanent regression accepted, not fixed (Req 5.1 required the removal)**: `on_job_postdone`'s `finish_time = time(NULL)` anomaly backstop is gone with no wire-format-level equivalent — concretely, a job hitting this rare no-prior-exit-status path now has `finish_time` stuck at 0 forever, which `job_should_retain()` (persist.c) evaluates as immediately GC-eligible instead of retained for the configured window. Narrow, rare-path, but real. **TWO CRITICAL HAZARDS DISCOVERED BY REVIEW, MUST BE RESOLVED BEFORE TASK 6 WIRES apply_committed_entry() INTO A LIVE PATH — currently inert only because replay runs before persist_start() registers hooks (cluster.c: persist_recover() at line ~1451 vs persist_start() at ~1694)**: 1. **Latent double-append**: `persist_job_hook_fun` is STILL registered on `(XCTL_HOOK_JOB, HOOK_AFTER_EVENT)` with no detail filter for `HOOK_DETAIL_JOB_UPDATE`, and does NOT early-return for that detail (it only early-returns for the WAIT-mode submit path). The moment task 6 wires `apply_committed_entry()` into live commit-advance, EVERY fact `apply_committed_entry()` applies will fire `HOOK_DETAIL_JOB_UPDATE`, which `persist_job_hook_fun` will catch and append AGAIN — a duplicate, mis-typed WAL entry per live-applied fact. **Task 6.1/6.2 must retire or detail-filter `persist_job_hook_fun`'s fact-handling branch before going live**, since no live receipt path fires AFTER_EVENT for facts anymore (this task removed the only other source). diff --git a/src/controller/meson.build b/src/controller/meson.build index 395795c..a3654f7 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -161,6 +161,7 @@ controller_unit_test_src = [ 'testsuite/test_apply_committed_entry.c', 'testsuite/test_replay_apply_seam.c', 'testsuite/test_apply_reapply_idempotency.c', + 'testsuite/test_apply_correctness_verification.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/persist.c b/src/controller/persist.c index 30a60a7..77c8052 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -612,6 +612,12 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) * on an unrecoverable boot; cleared on catch-up (never self-exits). */ atomic_store(&p->recovery_unrecoverable, false); + /* Post-replay apply-correctness verdict defaults to CLEAN (raft-apply- + * decoupling task 5.1, Req 8.2): a fresh store has nothing to diverge + * from. persist_recover overwrites both after every replay. */ + p->apply_verdict = APPLY_CLEAN; + atomic_store(&p->last_applied_lsn, 0); + /* Durable-advance wake hook starts unregistered (ha-commit-durability 2.2): * the commit engine registers it in commit_engine_init. Until then every * flusher durable advance is a no-op wake. */ @@ -2811,13 +2817,24 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, * about the STRUCTURE of the local log, not about being behind the cluster: a * clean log that simply ends below the cluster high-water is still CLEAN. * + * new_job_submits_out (raft-apply-decoupling task 5.1, when non-NULL) is set + * to the number of WAL_JOB_SUBMIT entries this call actually dispatched via + * apply_committed_entry() — i.e. entries above `from_lsn` that reached + * dispatch, not merely present in the log. This is the log-derived half of + * the post-replay apply-correctness expectation (see apply_expectation_t); + * it is a pure scan tally, independent of apply_committed_entry()'s own + * return code. + * * Returns number of entries replayed (>= 0) or negative errno on I/O error. */ static int replay_wal(persist_t* p, uint64_t from_lsn, - recover_verdict_t* verdict_out) + recover_verdict_t* verdict_out, + uint64_t* new_job_submits_out) { /* Default to CLEAN; only a structural corruption flips it to TRUNCATED. */ if (verdict_out) *verdict_out = RECOVER_CLEAN; + if (new_job_submits_out) + *new_job_submits_out = 0; /* --- Collect and sort segment filenames by start LSN --- */ DIR* d = opendir(p->wal_path); @@ -3028,6 +3045,27 @@ static int replay_wal(persist_t* p, uint64_t from_lsn, hdr.payload_len); free(payload); + /* raft-apply-decoupling task 5.1: track the replay-derived + * LOGICAL apply cursor separately from current_lsn (the physical + * WAL position owned by this loop, bug-778) — last_applied_lsn is + * the "highest entry actually dispatched via + * apply_committed_entry()" position the post-replay verification + * compares against. Segments are scanned in ascending LSN order + * (sorted above) and entries within a segment are sequential, so + * a plain store (no monotonic max needed) is correct here. */ + atomic_store_explicit(&p->last_applied_lsn, hdr.lsn, + memory_order_relaxed); + + /* Log-derived half of the post-replay apply-correctness + * expectation (Req 8.1): every WAL_JOB_SUBMIT that reaches this + * point is, by construction, above the loaded snapshot's + * checkpoint_lsn (the from_lsn skip above already filtered out + * anything the snapshot covers) — so it names a job that must now + * exist in the live job table that did not exist before this + * replay pass began. */ + if (hdr.type == WAL_JOB_SUBMIT && new_job_submits_out) + (*new_job_submits_out)++; + /* Advance current_lsn to one past this entry */ if (hdr.lsn + 1 > p->current_lsn) p->current_lsn = hdr.lsn + 1; @@ -3047,6 +3085,85 @@ static int replay_wal(persist_t* p, uint64_t from_lsn, return replayed; } +/* raft-apply-decoupling task 5.1: cheap read-lock count of c->jobmgr's live + * entries, mirroring collect_job_frames' iteration idiom (this file, above) + * without collecting or filtering anything. Used only to derive the + * post-replay materialized-entity expectation (never on a hot path). Safe on + * a NULL cluster or jobmgr (a bare unit-test persist_t) — returns 0. */ +static uint64_t count_jobmgr_entries(cluster_t* c) +{ + if (!c || !c->jobmgr) + return 0; + + lock(c->jobmgr, RTTI_LOCK_READ); + + iterator_t it; + int irc = iter_make(&it, c->jobmgr); + if (irc < 0) { + unlock(c->jobmgr); + return 0; /* empty table */ + } + + uint64_t n = 0; + for (;;) { + job_t* j = NULL; + int nrc = iter_next(&it, (void**)&j); + if (nrc < 0 || nrc == RTTI_ITER_NO_DATA) + break; + if (j == NULL) + continue; + n++; + } + + iter_destroy(&it); + unlock(c->jobmgr); + return n; +} + +/* persist_verify_apply_correctness — see persist.h for the full contract + * doc (raft-apply-decoupling task 5.1, Req 8.1/8.2). Pure comparison: no + * logging, no fence side effects. */ +apply_verdict_t persist_verify_apply_correctness( + persist_t* p, const apply_expectation_t* exp, + uint64_t jobmgr_count_before_replay) +{ + if (!p || !exp) + return APPLY_CLEAN; /* defensive: nothing to verify */ + + uint64_t actual_last_applied = + atomic_load_explicit(&p->last_applied_lsn, memory_order_acquire); + if (actual_last_applied != exp->expected_last_applied_lsn) + return APPLY_DIVERGENT; + + /* Entity-materialization half of the check: skip gracefully when there + * is no live cluster to query — mirrors apply_committed_entry()'s own + * "nothing to dispatch into" convention for a bare unit-test persist_t. + * Only the position check above applies in that case. */ + if (!p->cluster || !p->cluster->jobmgr) + return APPLY_CLEAN; + + uint64_t jobmgr_count_after = count_jobmgr_entries(p->cluster); + if (jobmgr_count_after < jobmgr_count_before_replay) + return APPLY_DIVERGENT; /* table shrank — impossible on a clean replay */ + + uint64_t actual_new_jobs = jobmgr_count_after - jobmgr_count_before_replay; + if (actual_new_jobs != exp->expected_new_job_count) + return APPLY_DIVERGENT; + + return APPLY_CLEAN; +} + +apply_verdict_t persist_apply_verdict(persist_t* p) +{ + return p ? p->apply_verdict : APPLY_CLEAN; +} + +uint64_t persist_last_applied_lsn(persist_t* p) +{ + return p ? atomic_load_explicit(&p->last_applied_lsn, memory_order_acquire) + : 0; +} + /* ------------------------------------------------------------------ */ /* Public: persist_recover */ /* ------------------------------------------------------------------ */ @@ -3104,6 +3221,20 @@ int persist_recover(persist_t* p) } } + /* raft-apply-decoupling task 5.1 (Req 8.1): seed the replay-derived + * apply cursor from the loaded baseline (checkpoint_lsn stays 0 when no + * snapshot was found) so that, on a CLEAN replay with no new entries + * beyond the snapshot, the cursor already sits at the recovered tip — + * mirroring how current_lsn is seeded by load_snapshot. Step 3's + * replay_wal advances it further for every entry actually dispatched + * above this baseline. Also capture the live job table's size BEFORE + * replay runs: the post-replay verification below needs this baseline to + * isolate the jobs replay itself materializes from anything the snapshot + * already restored. */ + atomic_store_explicit(&p->last_applied_lsn, p->checkpoint_lsn, + memory_order_release); + uint64_t jobmgr_count_before_replay = count_jobmgr_entries(p->cluster); + /* --- Step 3: replay WAL entries after checkpoint_lsn --- */ /* Recovery-integrity verdict (ha-commit-durability task 3.1, Req 2.1): * replay distinguishes a clean log end from a corruption-truncated end. @@ -3113,7 +3244,8 @@ int persist_recover(persist_t* p) * CLEAN — the sentinel established the committed position; replay only adds * entries above it. */ recover_verdict_t verdict = RECOVER_CLEAN; - int replayed = replay_wal(p, p->checkpoint_lsn, &verdict); + uint64_t new_job_submits = 0; + int replayed = replay_wal(p, p->checkpoint_lsn, &verdict, &new_job_submits); if (replayed < 0) return replayed; @@ -3148,6 +3280,51 @@ int persist_recover(persist_t* p) uint64_t recovered_tip = (p->current_lsn > 0) ? p->current_lsn - 1 : 0; atomic_store_explicit(&p->durable_lsn, recovered_tip, memory_order_release); + /* raft-apply-decoupling task 5.1 (Req 8.1/8.2): post-replay apply- + * correctness verification, reported ALONGSIDE the structural + * recovery_verdict set above — neither implies the other. See + * persist_verify_apply_correctness's doc comment for exactly what is + * compared. Routing an APPLY_DIVERGENT verdict into the existing + * non-promotable loud-fence mechanism is task 5.2; this only classifies + * and logs. + * + * Round-2 fix: the expected position MUST be the LOGICAL apply cursor + * replay_wal() already maintains in p->last_applied_lsn — NOT + * recovered_tip (current_lsn - 1, the PHYSICAL WAL-scan tip used above + * for durable_lsn). current_lsn advances for every WAL entry replay_wal + * scans, including WAL_CHECKPOINT markers and unknown future types, + * neither of which is ever routed through apply_committed_entry() (see + * the skip branches in replay_wal). last_applied_lsn only advances when + * a real dispatch happens. If the highest-LSN entry above checkpoint_lsn + * is a trailing WAL_CHECKPOINT marker — the exact state left behind by a + * crash between persist_write_checkpoint_marker() (step 1, durably + * fsynced) and write_snapshot() (step 3b, persists the advanced + * checkpoint_lsn) — recovered_tip over-counts past it while + * last_applied_lsn correctly does not, and comparing against + * recovered_tip previously produced a false APPLY_DIVERGENT for a + * perfectly clean recovery (violates Req 8.2). */ + uint64_t apply_cursor = + atomic_load_explicit(&p->last_applied_lsn, memory_order_acquire); + apply_expectation_t apply_exp = { + .expected_last_applied_lsn = apply_cursor, + .expected_new_job_count = new_job_submits, + }; + p->apply_verdict = persist_verify_apply_correctness( + p, &apply_exp, jobmgr_count_before_replay); + if (p->apply_verdict == APPLY_DIVERGENT) + ERROR("persist_recover: APPLY_DIVERGENT — materialized state does " + "not match the committed log tip (expected last_applied_lsn=%" + PRIu64 " new_job_count=%" PRIu64 "); a hidden apply defect is " + "present (ha-commit-durability recover_verdict=%s)", + apply_exp.expected_last_applied_lsn, + apply_exp.expected_new_job_count, + (verdict == RECOVER_TRUNCATED) ? "RECOVER_TRUNCATED" + : "RECOVER_CLEAN"); + else + INFO("persist_recover: APPLY_CLEAN — materialized state matches the " + "committed log tip (last_applied_lsn=%" PRIu64 ")", + apply_cursor); + /* --- Step 4: retention pass (TTL check) — no-op in P0 (no entities) --- */ return 0; diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 8ccafcd..d86976f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -81,6 +81,7 @@ extern void test_apply_seam(void); extern void test_apply_committed_entry(void); extern void test_replay_apply_seam(void); extern void test_apply_reapply_idempotency(void); +extern void test_apply_correctness_verification(void); void setUp(void) { return; } @@ -231,6 +232,7 @@ int main(int argc, char* argv[]) TEST(test_apply_committed_entry); TEST(test_replay_apply_seam); TEST(test_apply_reapply_idempotency); + TEST(test_apply_correctness_verification); return UNITY_END(); } diff --git a/src/controller/testsuite/test_apply_correctness_verification.c b/src/controller/testsuite/test_apply_correctness_verification.c new file mode 100644 index 0000000..a32aa84 --- /dev/null +++ b/src/controller/testsuite/test_apply_correctness_verification.c @@ -0,0 +1,517 @@ +/* test_apply_correctness_verification.c — raft-apply-decoupling task 5.1 + * + * Proves the post-replay apply-correctness verification added alongside the + * existing structural (CRC) recovery verdict (design.md "Restart Replay & + * Verification", Req 8.1/8.2): + * + * 1. A clean restart replay (real WAL, real cluster_t/jobmgr, no + * corruption, no divergence) produces an APPLY_CLEAN verdict from + * persist_apply_verdict() ALONGSIDE the existing RECOVER_CLEAN verdict + * from persist_recovery_verdict() — neither replaces the other. + * + * 2. The comparison logic itself — persist_verify_apply_correctness(), the + * pure function driving that verdict — can genuinely tell CLEAN apart + * from a mismatch: an expectation whose new-job-count disagrees with + * the live job table's actual growth is APPLY_DIVERGENT, and one whose + * last-applied position disagrees is APPLY_DIVERGENT, even when the + * other half of the expectation matches. This is exercised directly + * against the pure function (not through a real replay) — routing an + * APPLY_DIVERGENT verdict through the existing non-promotable fence + * mechanism is raft-apply-decoupling task 5.2, and a forced-DIVERGENT + * full-replay scenario proving the fence stays alive is task 5.3; both + * are out of this task's boundary. + * + * 3. persist_verify_apply_correctness() degrades safely to APPLY_CLEAN + * when there is no live cluster to query (a bare unit-test persist_t), + * mirroring apply_committed_entry()'s own "nothing to dispatch into" + * convention. + * + * Fixtures mirror test_replay_apply_seam.c (task 2.3, full cluster_t/WAL + * replay) and test_apply_seam.c (task 2.1, bare jobmgr, no cluster_t) — + * whichever is the lighter fit for each test. + */ +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "cluster.h" +#include "controller.capnp.h" +#include "hook.h" +#include "jobj.h" +#include "persist.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 the other apply-seam test files. */ +static int g_tacv_jobobj_ready; + +/* ── Fixture A: full cluster_t + real disk-backed WAL, for the end-to-end + * clean-replay test (mirrors test_replay_apply_seam.c). ─────────────── */ + +static cluster_t g_tacv_c; + +static void tacv_build_cluster(void) +{ + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 32, + .delta = 16 }; + kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING, + .nslots = 32, + .delta = 16 }; + + if (makeString((char*)"tacv-probe") == NULL) + xstring_table_init(64); + + if (!g_tacv_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_tacv_jobobj_ready = 1; + } + + memset(&g_tacv_c, 0, sizeof(g_tacv_c)); + g_tacv_c.myhost = &g_tacv_c.myhost_info; + g_tacv_c.myhost_info.name = (char*)"ctrl-test"; + g_tacv_c.myhost_info.key = (char*)""; + + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_tacv_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_tacv_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_tacv_c.qmgr = makeInstance(&xkvmgr_type, &str_params); + g_tacv_c.workermgr = makeInstance(&xkvmgr_type, &str_params); + g_tacv_c.resmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_tacv_c.jobmgr); + TEST_ASSERT_NOT_NULL(g_tacv_c.qmgr); + TEST_ASSERT_NOT_NULL(g_tacv_c.workermgr); + TEST_ASSERT_NOT_NULL(g_tacv_c.resmgr); + zarray_init(&g_tacv_c.sharedinst, 16, NULL); +} + +static void tacv_teardown_cluster(void) +{ + zarray_fini(&g_tacv_c.sharedinst); + if (g_tacv_c.resmgr) dereference(g_tacv_c.resmgr); + if (g_tacv_c.workermgr) dereference(g_tacv_c.workermgr); + if (g_tacv_c.qmgr) dereference(g_tacv_c.qmgr); + if (g_tacv_c.jobmgr) dereference(g_tacv_c.jobmgr); + xce_hook_fini(&g_tacv_c.hooksys); +} + +static int tacv_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct Job cj; + Job_ptr jp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_Job(cs, &cj, (xce_job_t*)j); + jp = new_Job(cs); + write_Job(&cj, jp); + capn_setp(capn_root(&c), 0, jp.p); + + *out_len = capn_size(&c); + *out = malloc(*out_len); + if (!*out) { capn_free(&c); return -ENOMEM; } + capn_write_mem(&c, *out, *out_len, 0); + capn_free(&c); + return 0; +} + +static persist_t* tacv_make_fresh_persist(const char* tmpdir) +{ + persist_t* p = calloc(1, sizeof(persist_t)); + if (!p) return NULL; + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; /* slow fsync — no races in tests */ + cfg.snap_keep_n = 2; + + persist_set_workdir(p, tmpdir, "ctrl-test"); + + if (persist_init(p, &cfg, NULL) != 0) { + free(p); + return NULL; + } + if (persist_start(p) != 0) { + persist_fini(p); + free(p); + return NULL; + } + return p; +} + +static void tacv_teardown_persist(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* + * End-to-end: a clean restart replay of a log containing two WAL_JOB_SUBMIT + * entries (no corruption, no divergence) must produce APPLY_CLEAN alongside + * the existing RECOVER_CLEAN structural verdict (Req 8.1, 8.2). + */ +static void test_clean_replay_produces_apply_clean_alongside_crc_verdict(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_tacv_clean_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + /* Phase 1: write two WAL_JOB_SUBMIT entries to a real WAL. */ + { + persist_t* p = tacv_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + for (uint64_t jobid = 70001; jobid <= 70002; jobid++) { + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = JOB_STAT_PEND; + + job_submit_data_t submit; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"rq1"; + submit.user = (char*)"tester"; + jdef.submit = &submit; + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, tacv_encode_job(&jdef, &payload, &plen)); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_SUBMIT, payload, (uint32_t)plen)); + free(payload); + } + + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + tacv_teardown_persist(p); + } + + /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk. */ + { + persist_t* p2 = tacv_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p2); + + tacv_build_cluster(); + p2->cluster = &g_tacv_c; + + TEST_ASSERT_EQUAL_INT(0, persist_recover(p2)); + + TEST_ASSERT_EQUAL_INT_MESSAGE(RECOVER_CLEAN, + persist_recovery_verdict(p2), + "existing structural (CRC) verdict must remain intact"); + TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN, + persist_apply_verdict(p2), + "a clean replay with no divergence must report APPLY_CLEAN"); + + uint64_t expected_tip = (p2->current_lsn > 0) ? p2->current_lsn - 1 : 0; + TEST_ASSERT_EQUAL_UINT64_MESSAGE(expected_tip, + persist_last_applied_lsn(p2), + "the replay-derived apply cursor must reach the recovered " + "committed log tip"); + + job_t* j1 = NULL; + job_t* j2 = NULL; + lock(g_tacv_c.jobmgr, RTTI_LOCK_READ); + key_find(g_tacv_c.jobmgr, (void*)70001, (void**)&j1); + key_find(g_tacv_c.jobmgr, (void*)70002, (void**)&j2); + unlock(g_tacv_c.jobmgr); + TEST_ASSERT_NOT_NULL(j1); + TEST_ASSERT_NOT_NULL(j2); + + tacv_teardown_cluster(); + tacv_teardown_persist(p2); + } + + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp); + if (system(cmd) != 0) + fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp); + + printf("test_clean_replay_produces_apply_clean_alongside_crc_verdict: PASS\n"); +} + +/* + * raft-apply-decoupling task 5.1 round 2 (Req 8.2): a WAL whose tail (above + * the loaded snapshot's checkpoint_lsn) is a WAL_CHECKPOINT marker with no + * further logical entry after it must NOT be misreported as + * APPLY_DIVERGENT. + * + * This reproduces the persist_checkpoint() crash window directly: + * persist_checkpoint()'s step 1 (persist_write_checkpoint_marker) durably + * fsyncs the WAL_CHECKPOINT marker; step 3b (write_snapshot) only later + * persists the advanced checkpoint_lsn to a snapshot file. A crash between + * those two steps leaves the marker durable in the WAL with no snapshot + * reflecting it. There is no test seam to interrupt persist_checkpoint() + * mid-sequence, so this test runs a real (complete) checkpoint and then + * deletes the snapshot file it just wrote — landing the on-disk state in + * exactly the same place a genuine crash between steps 1 and 3b would: + * marker durable in the WAL, no snapshot recording the advanced + * checkpoint_lsn. + * + * On the next recover, persist_recover() finds no snapshot (checkpoint_lsn + * starts at 0) and replay_wal() processes the full WAL: two real + * WAL_JOB_SUBMIT entries (advancing the logical apply cursor, + * last_applied_lsn, to their LSNs) followed by the trailing WAL_CHECKPOINT + * marker, which only advances the PHYSICAL WAL-scan position (current_lsn) + * — replay_wal's WAL_CHECKPOINT skip branch never routes it through + * apply_committed_entry(). The physical tip (current_lsn - 1) therefore + * lands one LSN above the true logical cursor, and the fix under test must + * compare against the logical cursor, not the physical tip. + */ +static void test_checkpoint_tail_with_no_snapshot_is_apply_clean(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_tacv_ckpt_tail_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + /* Phase 1: two real WAL_JOB_SUBMIT entries, then a checkpoint (which + * appends a WAL_CHECKPOINT marker as the new highest-LSN entry) — then + * delete the checkpoint's own snapshot file to simulate a crash between + * the marker's fsync (step 1) and the snapshot write (step 3b) that + * would have durably recorded the advanced checkpoint_lsn. */ + uint64_t marker_checkpoint_lsn = 0; + { + persist_t* p = tacv_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + for (uint64_t jobid = 71001; jobid <= 71002; jobid++) { + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = JOB_STAT_PEND; + + job_submit_data_t submit; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"rq1"; + submit.user = (char*)"tester"; + jdef.submit = &submit; + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, tacv_encode_job(&jdef, &payload, &plen)); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_SUBMIT, payload, (uint32_t)plen)); + free(payload); + } + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + + /* Real, complete checkpoint — marker fsynced AND snapshot written. + * A genuine crash never reaches the snapshot write, but there is no + * seam to stop persist_checkpoint() midway, so the snapshot is + * deleted immediately below to reach the identical post-crash + * on-disk state. */ + TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p)); + marker_checkpoint_lsn = p->checkpoint_lsn; + + char snap_file[4096]; + snprintf(snap_file, sizeof(snap_file), "%s/snap-%016" PRIx64 ".capnp", + p->snap_path, p->checkpoint_lsn); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, unlink(snap_file), + "fixture sanity: the checkpoint's own snapshot file must exist " + "before it is deleted to simulate the crash window"); + + tacv_teardown_persist(p); + } + TEST_ASSERT_TRUE(marker_checkpoint_lsn > 0); + + /* Phase 2: fresh persist_t recovers from disk. No snapshot exists (it + * was deleted above), so replay starts from checkpoint_lsn=0 and + * replays the full WAL, ending on the trailing WAL_CHECKPOINT marker. */ + { + persist_t* p2 = tacv_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p2); + + TEST_ASSERT_EQUAL_INT(0, persist_recover(p2)); + + TEST_ASSERT_EQUAL_INT_MESSAGE(RECOVER_CLEAN, + persist_recovery_verdict(p2), + "a checkpoint-marker tail is not WAL corruption -- must stay " + "RECOVER_CLEAN"); + + /* Fixture sanity: confirm this test actually exercises the + * physical/logical cursor gap the round-2 bug conflated -- the + * physical WAL-scan tip must land strictly above the logical apply + * cursor (on the checkpoint marker's LSN). */ + uint64_t physical_tip = + (p2->current_lsn > 0) ? p2->current_lsn - 1 : 0; + uint64_t logical_cursor = persist_last_applied_lsn(p2); + TEST_ASSERT_TRUE_MESSAGE(physical_tip > logical_cursor, + "fixture sanity: the checkpoint marker must sit above the last " + "logical entry for this test to exercise the round-2 bug"); + TEST_ASSERT_EQUAL_UINT64(marker_checkpoint_lsn, physical_tip + 1); + + TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN, + persist_apply_verdict(p2), + "a checkpoint-marker tail with no further logical entry must " + "NOT be misreported as APPLY_DIVERGENT (Req 8.2)"); + + tacv_teardown_persist(p2); + } + + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp); + if (system(cmd) != 0) + fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp); + + printf("test_checkpoint_tail_with_no_snapshot_is_apply_clean: PASS\n"); +} + +/* ── Fixture B: bare jobmgr + a minimal cluster_t wrapper (no hooksys, no + * persist/WAL), for driving persist_verify_apply_correctness() directly + * as a pure function — mirrors test_apply_seam.c's bare-jobmgr fixture, + * which needs none of Fixture A's full cluster machinery either. ──────── */ + +static var g_tacv_bare_jobmgr; + +static void tacv_bare_setup(void) +{ + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 16, + .delta = 16 }; + + if (makeString((char*)"tacv-bare-probe") == NULL) + xstring_table_init(64); + + if (!g_tacv_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(16, 16)); + g_tacv_jobobj_ready = 1; + } + + g_tacv_bare_jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(g_tacv_bare_jobmgr); +} + +static void tacv_bare_teardown(void) +{ + if (g_tacv_bare_jobmgr != NULL) { + dereference(g_tacv_bare_jobmgr); + g_tacv_bare_jobmgr = NULL; + } +} + +static job_t* tacv_bare_make_job(uint64_t jobid) +{ + xce_job_t jdef; + job_submit_data_t submit; + + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"q"; + submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + + return makeJob(g_tacv_bare_jobmgr, &jdef); +} + +/* + * The comparison logic must genuinely distinguish a matching expectation + * from one whose new-job-count disagrees with the live job table's actual + * growth (Req 8.1/8.2's comparison step, exercised directly — routing this + * outcome through the fence is task 5.2's job, not this test's). + */ +static void test_verify_apply_correctness_detects_job_count_mismatch(void) +{ + tacv_bare_setup(); + + cluster_t c; + memset(&c, 0, sizeof(c)); + c.jobmgr = g_tacv_bare_jobmgr; + + persist_t p; + memset(&p, 0, sizeof(p)); + p.cluster = &c; + + uint64_t jobmgr_count_before = 0; /* empty table before "replay" */ + + /* Two jobs materialize — mimics what a real replay would have done via + * apply_committed_entry()'s WAL_JOB_SUBMIT route. */ + TEST_ASSERT_NOT_NULL(tacv_bare_make_job(80001)); + TEST_ASSERT_NOT_NULL(tacv_bare_make_job(80002)); + + atomic_store_explicit(&p.last_applied_lsn, 42, memory_order_relaxed); + + /* Matching expectation: 2 new jobs, position 42 — CLEAN. */ + apply_expectation_t exp_match = { + .expected_last_applied_lsn = 42, + .expected_new_job_count = 2, + }; + TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN, + persist_verify_apply_correctness(&p, &exp_match, jobmgr_count_before), + "matching expectation must report APPLY_CLEAN"); + + /* Mismatched expectation: log says 3 jobs should exist, only 2 do. */ + apply_expectation_t exp_count_mismatch = { + .expected_last_applied_lsn = 42, + .expected_new_job_count = 3, + }; + TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_DIVERGENT, + persist_verify_apply_correctness(&p, &exp_count_mismatch, + jobmgr_count_before), + "a job-count mismatch must be detected as APPLY_DIVERGENT"); + + /* Mismatched expectation: position disagrees, count matches. */ + apply_expectation_t exp_position_mismatch = { + .expected_last_applied_lsn = 99, + .expected_new_job_count = 2, + }; + TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_DIVERGENT, + persist_verify_apply_correctness(&p, &exp_position_mismatch, + jobmgr_count_before), + "an apply-cursor position mismatch must be detected as " + "APPLY_DIVERGENT even when the job count matches"); + + tacv_bare_teardown(); + printf("test_verify_apply_correctness_detects_job_count_mismatch: PASS\n"); +} + +/* + * Defensive: no live cluster to query (a bare unit-test persist_t, or any + * caller with nothing to dispatch into) must degrade to APPLY_CLEAN rather + * than false-positive DIVERGENT — mirrors apply_committed_entry()'s own + * "nothing to dispatch into" convention. + */ +static void test_verify_apply_correctness_clean_when_no_cluster(void) +{ + persist_t p; + memset(&p, 0, sizeof(p)); + p.cluster = NULL; + + atomic_store_explicit(&p.last_applied_lsn, 7, memory_order_relaxed); + + apply_expectation_t exp = { + .expected_last_applied_lsn = 7, + .expected_new_job_count = 5, /* irrelevant -- no cluster to check */ + }; + TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN, + persist_verify_apply_correctness(&p, &exp, 0), + "no live cluster to query must degrade to APPLY_CLEAN, not a false " + "DIVERGENT"); + + printf("test_verify_apply_correctness_clean_when_no_cluster: PASS\n"); +} + +void test_apply_correctness_verification(void) +{ + RUN_TEST(test_clean_replay_produces_apply_clean_alongside_crc_verdict); + RUN_TEST(test_checkpoint_tail_with_no_snapshot_is_apply_clean); + RUN_TEST(test_verify_apply_correctness_detects_job_count_mismatch); + RUN_TEST(test_verify_apply_correctness_clean_when_no_cluster); +} diff --git a/src/include/persist.h b/src/include/persist.h index a0099cd..7366da7 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -87,6 +87,52 @@ typedef struct { uint64_t cluster_hwm_lsn; /* cluster committed high-water mark, for logging */ } recover_policy_inputs_t; +/* ============================================================ + * Post-replay apply-correctness verdict (raft-apply-decoupling task 5.1, + * design.md "Restart Replay & Verification", Req 8.1/8.2) + * + * DISTINCT from recover_verdict_t above, and NOT a replacement for it: + * RECOVER_CLEAN/RECOVER_TRUNCATED is a structural verdict over the WAL's + * bytes (CRC chain intact or not). apply_verdict_t is a SEMANTIC verdict + * over the result of replaying those bytes through apply_committed_entry() — + * it proves the materialized state (e.g. the live job table) actually + * matches a lightweight expectation derived from the committed log tip, not + * merely that the bytes were readable. This catches a class of hidden apply + * defect (e.g. a dispatch call that reports success without the entity + * actually landing in the live table) that a byte-level CRC check cannot + * see. Both verdicts are reported side by side on every restart (Req 8.1); + * neither one implies the other. + * ============================================================ */ +typedef enum { + APPLY_CLEAN = 0, /* materialized state matches the log-derived + * expectation (Req 8.2). */ + APPLY_DIVERGENT = 1, /* materialized state does NOT match the log-derived + * expectation (Req 8.3) — a hidden apply defect. + * Routing this into the existing non-promotable + * fence mechanism is raft-apply-decoupling task 5.2; + * this type only classifies, it does not fence. */ +} apply_verdict_t; + +/* Lightweight post-replay expectation, derived purely from scanning the + * committed log during replay (task 5.1, Req 8.1) — never from trusting + * apply_committed_entry()'s own return code, so the comparison is a genuine + * second, independent check on the materialized result. + * + * expected_last_applied_lsn — the highest LSN of any logical (non- + * checkpoint, non-unknown-type) entry replay actually dispatched via + * apply_committed_entry() in this replay pass: the "committed log tip" + * position component of the expectation. + * expected_new_job_count — the number of WAL_JOB_SUBMIT entries replay + * dispatched in this pass. Every such entry is, by construction, above + * the loaded snapshot's checkpoint_lsn (entries at/below it are skipped + * before ever reaching dispatch — see replay_wal), so each one names a + * job that must now exist in the live job table that did NOT exist + * before this replay pass began. */ +typedef struct { + uint64_t expected_last_applied_lsn; + uint64_t expected_new_job_count; +} apply_expectation_t; + /* ============================================================ * Persistence configuration (Part 2.6) * ============================================================ */ @@ -278,6 +324,27 @@ struct persist_s { * threads. Defaults to false before recovery runs. */ _Atomic bool recovery_unrecoverable; + /* Post-replay apply-correctness verdict + cursor (raft-apply-decoupling + * task 5.1, design.md "Restart Replay & Verification", Req 8.1/8.2). + * last_applied_lsn is the replay-derived LOGICAL apply cursor — the + * highest LSN actually dispatched via apply_committed_entry() — seeded + * from checkpoint_lsn before replay begins and advanced by replay_wal + * per entry; it is DISTINCT from current_lsn (the physical WAL position + * owned by the physical layer, bug-778) even though the two coincide on + * a clean replay. Never persisted — recomputed fresh every + * persist_recover() (design.md State Management), so it cannot itself + * drift from disk state. The live apply pump (task 6.1, not yet built) + * takes over its advancement once replay completes. apply_verdict is the + * sticky APPLY_CLEAN/APPLY_DIVERGENT result of comparing this cursor (and + * the live job table's growth) against the log-derived expectation; + * defaults to APPLY_CLEAN before recovery runs, mirroring + * recovery_verdict's default-clean convention. Written single-threaded + * during recovery; read by persist_apply_verdict/persist_last_applied_lsn + * thereafter (task 5.2 will additionally consume apply_verdict to route a + * DIVERGENT result into the existing loud-fence mechanism). */ + _Atomic uint64_t last_applied_lsn; + apply_verdict_t apply_verdict; + /* Paths (derived from workdir + name via persist_set_workdir) */ char wal_path[4096]; char snap_path[4096]; @@ -607,6 +674,38 @@ persist_recover_loud_fence_if_unrecoverable(persist_t* p, bool persist_recovery_unrecoverable(persist_t* p); void persist_clear_recovery_unrecoverable(persist_t* p); +/* persist_verify_apply_correctness (raft-apply-decoupling task 5.1, Req + * 8.1/8.2). Pure comparison — mirrors persist_recover_decision's separation + * of decision from action: no logging, no fence side effects, no mutation of + * `p` beyond the read of last_applied_lsn. Compares `exp` (derived by + * scanning the committed log during replay — see apply_expectation_t) against + * the actual materialized state: p's replay-derived apply cursor, and — when + * p->cluster is non-NULL — the live job table's growth since + * `jobmgr_count_before_replay` (the job count captured just before replay + * began). Returns APPLY_CLEAN on a match. Returns APPLY_DIVERGENT when either + * the apply cursor or the live job count disagrees with `exp`. + * + * Safe to call with p->cluster == NULL (a bare unit-test persist_t, or any + * caller with nothing to dispatch into): the job-count half of the check is + * skipped — only the position check applies — mirroring + * apply_committed_entry()'s own "nothing to dispatch into" convention. Safe + * on NULL p or NULL exp (returns APPLY_CLEAN — nothing to verify). */ +apply_verdict_t persist_verify_apply_correctness( + persist_t* p, const apply_expectation_t* exp, + uint64_t jobmgr_count_before_replay); + +/* Accessor for the sticky verdict computed by the last persist_recover() + * (task 5.1). APPLY_CLEAN before recovery has run (fresh store default, + * mirrors persist_recovery_verdict's RECOVER_CLEAN default). Safe on NULL p. */ +apply_verdict_t persist_apply_verdict(persist_t* p); + +/* Accessor for the replay-derived logical apply cursor (task 5.1, + * design.md "Restart Replay & Verification" state model) — see the + * last_applied_lsn field doc comment for its precise semantics and how it + * differs from current_lsn/persist_durable_floor. Safe on NULL p (returns + * 0). Lock-free atomic read. */ +uint64_t persist_last_applied_lsn(persist_t* p); + /* Register a durable-advance wake callback (ha-commit-durability task 2.2). * The group-commit flusher calls fn(ctx) after it advances durable_lsn and * after releasing p->lock, so a consumer (the commit engine) can re-evaluate