diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index 17107aa..6b29e71 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -55,7 +55,7 @@ - _Requirements: 2.2, 3.1, 3.2, 4.1, 4.2, 5.2_ - _Depends: 2.1_ -- [ ] 2.3 Point restart replay's per-entry application at the new dispatch function +- [x] 2.3 Point restart replay's per-entry application at the new dispatch function - Replace replay's direct calls to the individual restore primitives with calls to the dispatch function from 2.2 - Observable: restart replay of a log containing every classified entry type produces identical materialized state to before this change, verified by an existing or updated replay test - _Requirements: 3.1, 3.2_ @@ -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**. +- **2.3**: rewired `replay_wal()` (persist.c) to call `apply_committed_entry()` (task 2.2's seam) instead of `apply_wal_entry()` for restart replay's per-entry apply. `apply_wal_entry`/`apply_committed_entry` are siblings, not wrapper/wrappee — `apply_wal_entry` is NOT retired, it remains ha.c's live standby catch-up/replicated-apply dispatch (untouched, different call sites, out of this task's boundary). **Confirmed hazard, investigated to resolution by the reviewer (not left open)**: `replay_wal(p, p->checkpoint_lsn, ...)` replays the FULL WAL range since the last snapshot on every restart (`last_applied_lsn` is deliberately not persisted — recomputed at replay time per design.md), not just a narrow crash-gap tail. Since `apply_committed_entry` fires `AFTER_EVENT` hooks (unlike the old `apply_wal_entry`, which never did), this means every restart re-fires notification hooks for entries already applied+notified in a prior process life — a duplicate-notification hazard in principle. **Independently traced and confirmed inert today** for both existing hook consumers: scheduler hooks aren't registered until `on_scheduler_join()` (sched.c), which happens post-boot via network JOIN handshake, strictly after `persist_recover()` runs; apiserver hooks ARE registered before `persist_recover()` (`apiserver_init()` in `cluster_init_phase2()`), but every apiserver hook body gates its actual network notify on `apiservers[i].ep != NULL`, and the apiserver-join table starts empty each process life — a no-op except via a narrow reconnect race. **MUST be re-verified before tasks 4.1/6.1/6.2/6.3** make this path unambiguously live and broad (e.g. if a future task registers apiserver connections earlier, or scheduler hooks get registered pre-replay) — at that point this either needs a replay-mode flag suppressing hook-firing, or an explicit accepted-idempotent-duplicate-notification decision recorded in design.md. A loud in-line comment cross-referencing this note now sits at the `apply_committed_entry()` call site in `replay_wal()` (persist.c). Also confirmed: job-state facts now replay through 2.1's guarded `apply_job_update_transition()` instead of the previously-unguarded `restore_job_state()` — intentional (Req 3/7 online-replay divergence fix), verified no existing test depended on the old unguarded semantics. `apply_committed_entry`'s missing oversized-payload guard (vs. `apply_wal_entry`'s `MAX_WAL_PAYLOAD_BYTES`/`-EMSGSIZE` check) confirmed inert — `persist_wal_write_replicated()` already rejects oversized payloads at write time, before anything reaches disk. - **2.2**: implemented `apply_committed_entry(persist_t*, uint32_t wal_type, uint64_t lsn, const void* payload, uint32_t len)` in persist.c/persist.h, matching design.md's signature. Routes `WAL_JOB_SUBMIT` through `restore_job()` (idempotent on duplicate jobid — no-op, Req 7.1), `WAL_QUEUE_CTRL`/`WAL_WORKER_CTRL` through their existing restore primitives, and fires each type's post-mutation hook exactly once, after mutation. **Confirmed permanent, non-cosmetic deviation from design.md's Apply Seam wording**: ALL nine job-state fact WAL types (`WAL_JOB_{START,FINISH,FAIL,STOP,RESUME,UPDATE,RUSAGE,CKPT,PREASON}`) route through the single guard-only `apply_job_update_transition()` core from 2.1 — NOT the type-specific `apply_job_{start,finish,fail}_transition` cores — and fire one generic `HOOK_DETAIL_JOB_UPDATE` hook, not the type-specific `HOOK_DETAIL_JOB_START`/`_EXIT`/`_FAIL`. Root cause, independently verified by two review rounds: the on-disk wire format for every one of these types (`wal_jobstate_t {jobid, state}`) genuinely only carries the resulting state, not the richer `job_pidinfo_t`/`xce_jobfinish_t`/`job_failinfo_t` payload the type-specific cores/hooks require — this is a pre-existing wire-format reality (`apply_wal_entry` already collapsed these the same way before this task), not a shortcut invented here. **design.md's Apply Seam section and Fact Ingestion → Event Contract section currently describe per-type hook dispatch that this wire format cannot support as written** — this is a real, standing design/reality conflict, not yet reconciled in design.md itself. No task in the current plan (2.3, 4.1, 4.2, 6.1, 6.2, 6.3) enriches the wire format to close this gap — confirmed by reading all of them; task 4.1 is exclusively about removing receipt-time mutation/hook-firing, not payload enrichment. **Explicit blocking-style call-out for whoever implements 4.1/6.2/6.3**: wiring this dispatch function into a live fact-apply path will permanently change every worker-reported job start/finish/fail notification to a generic UPDATE detail instead of the type-specific one, unless the wire format is enriched first or design.md is corrected to formally accept this substitution — verify this is still the accepted behavior before wiring it live, don't assume it'll be fixed by then. Took 3 review rounds to land: round 1 rejected two doc-comment defects (a citation blaming the wrong crash site, and a dangling pointer to task 4.1 as the fix location); round 2's fix corrected the crash site (real one is `sched.c`'s `sched_job_start_fun`/`_exit_fun`/`_fail_fun`, which unconditionally dereference `info->jobid` before any NULL check) but fabricated a nonexistent function name (`encode_JobPIDInfo_ptr`) as supporting evidence; round 3 replaced it with a narrower, fully source-verified claim (apiserver.c's equivalent functions store the raw pointer into a notify union without dereferencing it at that call site — real, checked line-by-line). **Lesson**: when a reviewer catches a citation error, the safest fix is the narrowest claim you can personally verify by reading the actual code, not a broader-sounding replacement — round 2's attempt to explain "why the alternative is fine" reached for an unverified downstream mechanism and got caught fabricating it. - **2.1**: split `mark_job_start`/`mark_job_finish`/`mark_job_fail`/`mark_job_update` (job.c) into four hook-free cores (`apply_job_{start,finish,fail,update}_transition`, new `job_transition_result_t` enum in cluster.h) containing only the monotonic-rank guard + mutation, verified an unmodified lift line-by-line against pre-extraction code. Wrappers peek the same guard predicate to decide hook-firing, then delegate to the core — no unlock/relock gap between peek and delegated call, so no new race from the double-evaluation. `mark_job_fail`'s pre-existing lack of a guard is preserved, not added or removed. New `test_apply_seam.c` (8 tests) calls the cores directly, bypassing `on_job_*` receipt handlers entirely — this is the form task 2.2 (dispatch function) and task 6.1 (apply pump) will call into. **Non-blocking follow-up flagged by reviewer**: the new doc-comments on `apply_job_fail_transition`/`apply_job_update_transition` overclaim a universal "caller holds RTTI_LOCK_WRITE" precondition — `job_ctrl` (job.c:396, pre-existing, untouched) calls `mark_job_update`/`mark_job_fail` under only RTTI_LOCK_READ. No new race (extraction added no unlock/relock), but whoever writes task 2.2/6.1 should not trust that comment's blanket claim — verify the actual lock held at each call site instead. - **1.5**: pure test-only close-out of task-group 1 — 3 new genuine-multi-thread tests added across `test_commit_resolve.c`/`test_demotion_convergence.c`/`test_persist_truncate.c` (real pthreads, not sequential calls dressed as concurrent): an admission-cap burst with a concurrent monitor thread, a mixed majority-ack/timeout-sweep burst proving bounded dispatch-thread usage, and a genuine two-real-thread ack-vs-sweep-thread race (closing the "single-thread white-box only" gap both task-1.3 review rounds had flagged as a nice-to-have). No production code touched. **bug-1407 saga, definitively resolved**: it was NEVER a source bug. 1.4's "closed, 70/70" and 1.5's initial "reopened, 10/10 failing" were BOTH real, correct, *contradictory* measurements of the exact same source commit (9b4b805) — the difference was a stale/corrupted incremental-build artifact in the shared `build/` directory, which had been `ninja -C build`-rebuilt incrementally across all 5 tasks without ever a fresh `meson setup`. Proven by isolating each measurement in its own throwaway `git worktree add` + fresh `meson setup`/`ninja`: pre-feature commit and the exact failing commit both pass cleanly (0 failures) on a fresh build; only the main worktree's stale incremental `build/` produced the failure. Fixed with `rm -rf build && meson setup build --buildtype=debug -Db_sanitize=address && ninja -C build`; bug-1407 is now genuinely CLOSED with high confidence. **Lesson for every remaining task in this series**: distrust a surprising/contradictory `ninja -C build` incremental result before concluding a source regression — rebuild clean (or isolate via a throwaway worktree) first. Never use `git checkout -- .` to inspect an old commit in place — it silently reverts working-tree files against current HEAD; use `git worktree add ` instead. Both lessons are now in `.wolf/cerebrum.md` Do-Not-Repeat and `.wolf/buglog.json` bug-1407. diff --git a/src/controller/meson.build b/src/controller/meson.build index 1e91af2..004c2ef 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -159,6 +159,7 @@ controller_unit_test_src = [ 'testsuite/test_persist_snapshot_entities.c', 'testsuite/test_apply_seam.c', 'testsuite/test_apply_committed_entry.c', + 'testsuite/test_replay_apply_seam.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/persist.c b/src/controller/persist.c index c061750..aa8b54a 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -2961,10 +2961,34 @@ static int replay_wal(persist_t* p, uint64_t from_lsn, continue; } - /* Logical entity restore. The CRC chain + current_lsn are owned by - * this replay loop (current_lsn advanced below; last_crc at line - * ~1601), not by apply_wal_entry. */ - apply_wal_entry(p, hdr.type, hdr.lsn, payload, hdr.payload_len); + /* Logical entity restore via the shared apply seam + * (raft-apply-decoupling task 2.3, design.md "Apply Seam"): the + * SAME apply_committed_entry() dispatch the live commit-advance + * apply pump will use (task 6.1), so a node's materialized state + * is the deterministic result of its committed log regardless of + * whether it got there via restart replay or live apply + * (Req 3.1-3.4). apply_wal_entry() (this replay loop's previous + * per-entry dispatch) is NOT retired by this change — it remains + * the live standby catch-up/replicated-apply dispatch used + * elsewhere (ha.c); it is simply no longer this loop's call. The + * CRC chain + current_lsn remain owned by this replay loop + * (current_lsn advanced below; last_crc at line ~1601) — neither + * dispatch function ever touched them (bug-778). + * + * KNOWN HAZARD (tracked in tasks.md Implementation Notes, task + * 2.3): this replays the FULL post-checkpoint WAL range on every + * restart (last_applied_lsn is not persisted, deliberately + * recomputed here), so apply_committed_entry()'s AFTER_EVENT + * hooks fire again for entries already applied+notified in a + * prior process life. Confirmed inert today: scheduler hooks + * aren't registered until a post-boot network JOIN + * (on_scheduler_join, sched.c), and apiserver hooks — though + * registered before this runs — gate their notify on a + * connected endpoint that is empty at first-boot replay time. + * MUST be re-verified before tasks 4.1/6.1/6.2/6.3 make this + * path broadly live. */ + apply_committed_entry(p, hdr.type, hdr.lsn, payload, + hdr.payload_len); free(payload); /* Advance current_lsn to one past this entry */ diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 8ef26b7..9a19e27 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -79,6 +79,7 @@ extern void test_ha_partition(void); extern void test_persist_snapshot_entities(void); extern void test_apply_seam(void); extern void test_apply_committed_entry(void); +extern void test_replay_apply_seam(void); void setUp(void) { return; } @@ -227,6 +228,7 @@ int main(int argc, char* argv[]) TEST(test_persist_snapshot_entities); TEST(test_apply_seam); TEST(test_apply_committed_entry); + TEST(test_replay_apply_seam); return UNITY_END(); } diff --git a/src/controller/testsuite/test_replay_apply_seam.c b/src/controller/testsuite/test_replay_apply_seam.c new file mode 100644 index 0000000..332a832 --- /dev/null +++ b/src/controller/testsuite/test_replay_apply_seam.c @@ -0,0 +1,401 @@ +/* test_replay_apply_seam.c — raft-apply-decoupling task 2.3 + * + * Proves restart replay (persist_recover -> replay_wal) applies every + * classified WAL entry type through the SAME apply_committed_entry() seam + * task 2.2 built (routing job submit through restore_job, job-state facts + * through the 2.1 guarded core, queue control through restore_queue_state, + * worker control through worker_from_capnp) instead of the pre-2.3 replay + * path's direct calls inside apply_wal_entry(). + * + * Two things this proves: + * + * 1. Materialized-state parity (Req 3.1/3.2): a log containing one entry + * of every classified type — WAL_JOB_SUBMIT, a job-state fact + * (WAL_JOB_START), WAL_QUEUE_CTRL, WAL_WORKER_CTRL — replayed via + * persist_recover() on a fresh persist_t/cluster_t produces the + * correct end state. For this causally-ordered (submit-before-start), + * non-duplicate log, this is identical to what the pre-task-2.3 + * apply_wal_entry() dispatch already produced: neither switch body + * differs in materialized outcome for a forward, single-instance + * apply. This is the "pure refactor, existing behavior preserved" + * evidence for the call-site swap. + * + * 2. The call site genuinely switched, not just "happens to look the + * same": unlike apply_wal_entry() (which never fires a notification + * hook), apply_committed_entry() fires each type's AFTER_EVENT hook + * exactly once, strictly after its mutation. A spy hook counts these + * firings across the whole replay. Before task 2.3 (replay wired to + * apply_wal_entry) this count is 0; after (replay wired to + * apply_committed_entry) it is exactly 4 — one per entry. That + * flip is genuine RED (pre-change) / GREEN (post-change) evidence + * that replay's per-entry apply call site was actually rewired. + * + * Fixture mirrors test_apply_committed_entry.c's cluster_t construction + * (task 2.2), adapted to drive the real physical write path + * (persist_wal_async) and the real recovery path (persist_recover) instead + * of calling the dispatch function directly. + */ +#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 "qobj.h" +#include "wal_types.h" +#include "wobj.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 test_apply_seam.c / test_apply_committed_entry.c. */ +static int g_trr_jobobj_ready; + +static cluster_t g_trr_c; + +/* ── Spy hook: counts AFTER_EVENT firings across job/queue/worker hooks ─── */ +static int g_trr_hook_fires; + +static int trr_spy_hook(int type, int situation, int detail, void* data, + va_list ap) +{ + (void)type; (void)detail; (void)data; (void)ap; + if (situation == HOOK_AFTER_EVENT) + g_trr_hook_fires++; + return 0; +} + +static void trr_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*)"trr-probe") == NULL) + xstring_table_init(64); + + if (!g_trr_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_trr_jobobj_ready = 1; + } + + memset(&g_trr_c, 0, sizeof(g_trr_c)); + g_trr_c.myhost = &g_trr_c.myhost_info; + g_trr_c.myhost_info.name = (char*)"ctrl-test"; + g_trr_c.myhost_info.key = (char*)""; + + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_trr_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_trr_c.qmgr = makeInstance(&xkvmgr_type, &str_params); + g_trr_c.workermgr = makeInstance(&xkvmgr_type, &str_params); + g_trr_c.resmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_trr_c.jobmgr); + TEST_ASSERT_NOT_NULL(g_trr_c.qmgr); + TEST_ASSERT_NOT_NULL(g_trr_c.workermgr); + TEST_ASSERT_NOT_NULL(g_trr_c.resmgr); + zarray_init(&g_trr_c.sharedinst, 16, NULL); + + /* Queues come from config, not from WAL_QUEUE_CTRL (restore_queue_state + * -ENOENT's "queue not yet built from config" is a benign replay skip) + * — pre-create the queue the replayed WAL_QUEUE_CTRL entry targets, + * exactly as a real boot's config load would before replay runs. */ + xce_queue_t qdef; + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char*)"rq1"; + TEST_ASSERT_NOT_NULL(makeQueue(g_trr_c.qmgr, &qdef)); + + xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_JOB, trr_spy_hook, + NULL); + xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_QUEUE, trr_spy_hook, + NULL); + xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, trr_spy_hook, + NULL); + + g_trr_hook_fires = 0; +} + +static void trr_teardown_cluster(void) +{ + zarray_fini(&g_trr_c.sharedinst); + if (g_trr_c.resmgr) dereference(g_trr_c.resmgr); + if (g_trr_c.workermgr) dereference(g_trr_c.workermgr); + if (g_trr_c.qmgr) dereference(g_trr_c.qmgr); + if (g_trr_c.jobmgr) dereference(g_trr_c.jobmgr); + xce_hook_fini(&g_trr_c.hooksys); +} + +/* ── Encode helpers (mirror test_apply_committed_entry.c's tace_encode_*; + * capnp_encode_{job,worker} are file-static in persist.c). ──────────── */ + +static int trr_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 int trr_encode_worker(const xce_worker_t* w, uint8_t** out, + size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct Worker cw; + Worker_ptr wp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_Worker(cs, &cw, (xce_worker_t*)w); + wp = new_Worker(cs); + write_Worker(&cw, wp); + capn_setp(capn_root(&c), 0, wp.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; +} + +/* ── persist_t fixture (real disk-backed WAL, mirrors + * test_persist_recovery.c's make_fresh_persist). ─────────────────────── */ + +static persist_t* trr_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 trr_teardown_persist(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* Look up helpers, mirroring test_apply_committed_entry.c. */ +static job_t* trr_find_job(uint64_t jobid) +{ + job_t* j = NULL; + lock(g_trr_c.jobmgr, RTTI_LOCK_READ); + int found = key_find(g_trr_c.jobmgr, (void*)jobid, (void**)&j); + unlock(g_trr_c.jobmgr); + return (found > 0) ? j : NULL; +} + +static queue_t* trr_find_queue(const char* name) +{ + queue_t* q = NULL; + lock(g_trr_c.qmgr, RTTI_LOCK_READ); + key_find(g_trr_c.qmgr, (void*)name, (void**)&q); + unlock(g_trr_c.qmgr); + return q; +} + +static worker_t* trr_find_worker(const char* name) +{ + worker_t* w = NULL; + lock(g_trr_c.workermgr, RTTI_LOCK_READ); + key_find(g_trr_c.workermgr, (void*)name, (void**)&w); + unlock(g_trr_c.workermgr); + return w; +} + +/* + * Writes one physical WAL entry of every classified type (real + * persist_wal_async appends — no hand-rolled headers/CRCs) in causal order: + * submit before the state fact that mutates the same job. + */ +static void trr_write_classified_log(persist_t* p) +{ + /* WAL_JOB_SUBMIT: jobid 90001, PEND */ + job_submit_data_t submit; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"rq1"; + submit.user = (char*)"tester"; + + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = 90001; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + + uint8_t* job_payload = NULL; + size_t job_plen = 0; + TEST_ASSERT_EQUAL_INT(0, trr_encode_job(&jdef, &job_payload, &job_plen)); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_SUBMIT, job_payload, (uint32_t)job_plen)); + free(job_payload); + + /* WAL_JOB_START (job-state fact family): jobid 90001 -> RUN */ + wal_jobstate_t js = { .jobid = 90001, .state = JOB_STAT_RUN }; + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_START, &js, sizeof(js))); + + /* WAL_QUEUE_CTRL: close queue "rq1" */ + const char* qname = "rq1"; + size_t qname_len = strlen(qname) + 1; + size_t qtotal = sizeof(wal_queuehdr_t) + qname_len; + uint8_t* qpkt = malloc(qtotal); + TEST_ASSERT_NOT_NULL(qpkt); + wal_queuehdr_t qh = { .status = QUEUE_STAT_CLOSED }; + memcpy(qpkt, &qh, sizeof(qh)); + memcpy(qpkt + sizeof(qh), qname, qname_len); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_QUEUE_CTRL, qpkt, (uint32_t)qtotal)); + free(qpkt); + + /* WAL_WORKER_CTRL: join worker "rw1" */ + hostattrs_t attrs; + memset(&attrs, 0, sizeof(attrs)); + attrs.hosttype = (char*)"unknown"; + attrs.hostmodel = (char*)"unknown"; + + xce_worker_t wdef; + memset(&wdef, 0, sizeof(wdef)); + wdef.name = (char*)"rw1"; + wdef.attrs = &attrs; + + uint8_t* wbuf = NULL; + size_t wsize = 0; + TEST_ASSERT_EQUAL_INT(0, trr_encode_worker(&wdef, &wbuf, &wsize)); + + uint32_t disc = (uint32_t)HOOK_DETAIL_WORKER_JOIN; + size_t wtotal = sizeof(disc) + wsize; + uint8_t* wpkt = malloc(wtotal); + TEST_ASSERT_NOT_NULL(wpkt); + memcpy(wpkt, &disc, sizeof(disc)); + memcpy(wpkt + sizeof(disc), wbuf, wsize); + free(wbuf); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_WORKER_CTRL, wpkt, (uint32_t)wtotal)); + free(wpkt); + + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); +} + +/* + * Req 3.1/3.2 + genuine-rewire evidence: restart replay of a log containing + * every classified entry type reaches the correct end state, and does so by + * actually firing apply_committed_entry()'s post-mutation notification for + * each entry (proving replay is wired to the new seam, not merely producing + * a coincidentally-matching state). + */ +static void test_replay_applies_every_classified_type_via_apply_seam(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_replay_seam_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + /* Phase 1: write one entry of every classified type to a real WAL. No + * cluster needed — persist_wal_async is a physical-write-only path. */ + { + persist_t* p = trr_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + trr_write_classified_log(p); + trr_teardown_persist(p); + } + + /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk. */ + { + persist_t* p2 = trr_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p2); + + trr_build_cluster(); + p2->cluster = &g_trr_c; + + int rc = persist_recover(p2); + TEST_ASSERT_EQUAL_INT(0, rc); + + job_t* j = trr_find_job(90001); + TEST_ASSERT_NOT_NULL_MESSAGE(j, + "replay must construct the job via the apply seam's " + "WAL_JOB_SUBMIT route"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state, + "replay must apply the job-state fact via the apply seam's " + "guarded transition core, reaching the same end state as before"); + + queue_t* q = trr_find_queue("rq1"); + TEST_ASSERT_NOT_NULL(q); + TEST_ASSERT_TRUE_MESSAGE((q->queue.status & QUEUE_STAT_CLOSED) != 0, + "replay must apply WAL_QUEUE_CTRL via the apply seam"); + + worker_t* w = trr_find_worker("rw1"); + TEST_ASSERT_NOT_NULL_MESSAGE(w, + "replay must apply WAL_WORKER_CTRL via the apply seam"); + + /* One notification per entry: JOB_NEW, JOB_UPDATE, QUEUE_UPDATE, + * WORKER_JOIN. apply_wal_entry (the pre-2.3 replay dispatch) never + * fires hooks — this assertion is 0 there and 4 once replay is + * pointed at apply_committed_entry, proving the call site moved. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(4, g_trr_hook_fires, + "restart replay must drive every classified entry through " + "apply_committed_entry(), firing exactly one post-mutation " + "notification per entry"); + + trr_teardown_cluster(); + trr_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_replay_applies_every_classified_type_via_apply_seam: PASS\n"); +} + +void test_replay_apply_seam(void) +{ + RUN_TEST(test_replay_applies_every_classified_type_via_apply_seam); +}