From 8c00f20cec4d64fca7b2d0725d02c2ff71d17b85 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 3 Jul 2026 16:51:47 +0000 Subject: [PATCH] feat(raft-apply-decoupling): build apply_committed_entry single dispatch seam (task 2.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add apply_committed_entry(persist_t*, wal_type, lsn, payload, len) in persist.c/persist.h as the single seam that turns a committed WAL entry into an in-memory mutation plus its post-mutation notification hook, covering job submit, job-state facts, queue control, and worker control. Job-state facts route through 2.1's guard-only apply_job_update_transition() core and fire a generic HOOK_DETAIL_JOB_UPDATE, not per-type hooks — the wire format (wal_jobstate_t {jobid, state}) cannot supply what the type-specific cores/hooks require. This is a real, documented deviation from design.md's per-type dispatch wording; tasks.md's Implementation Notes flags it as unresolved and blocking for whoever wires this into a live path (4.1/6.2/6.3). Not yet wired into any live call site (replay wiring is 2.3). --- .kiro/specs/raft-apply-decoupling/tasks.md | 3 +- src/controller/meson.build | 1 + src/controller/persist.c | 180 +++++++ src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_apply_committed_entry.c | 452 ++++++++++++++++++ src/include/persist.h | 52 ++ 6 files changed, 689 insertions(+), 1 deletion(-) create mode 100644 src/controller/testsuite/test_apply_committed_entry.c diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index ea6e18b4..17107aa3 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -46,7 +46,7 @@ - _Requirements: 3.3, 7.1, 7.2_ - _Boundary: Apply Seam — persist.c / job.c apply-dispatch and replay call sites only, no dependency on the commit-engine change in task 1_ -- [ ] 2.2 Build the single dispatch function that applies one committed entry to in-memory state, covering every classified entry type +- [x] 2.2 Build the single dispatch function that applies one committed entry to in-memory state, covering every classified entry type - Route job submission through the existing idempotent, jobid-addressable construction primitive instead of a raw uncoordinated insert - Route job state transitions through the extracted guarded logic from 2.1 - Route queue control and worker control entries through their existing restore primitives @@ -193,5 +193,6 @@ - **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.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 8dd1b033..1e91af2e 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -158,6 +158,7 @@ controller_unit_test_src = [ 'testsuite/snapdiff.c', 'testsuite/test_persist_snapshot_entities.c', 'testsuite/test_apply_seam.c', + 'testsuite/test_apply_committed_entry.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/persist.c b/src/controller/persist.c index 4ea9ddad..c0617500 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -2578,6 +2578,186 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, return 0; } +/* raft-apply-decoupling task 2.2 — see persist.h for the full contract doc + * (routing table, the job-state-family hook-mapping deviation, and what this + * function deliberately does NOT own). Not yet wired into any live call + * site — replay (2.3) and the live apply pump (6.1) will call this later. */ +int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, + const void* payload, uint32_t len) +{ + if (!p) + return -EINVAL; + + if (!p->cluster) + return 0; /* unit tests without a full cluster — nothing to dispatch into */ + + cluster_t* c = p->cluster; + + switch (wal_type) { + case WAL_JOB_SUBMIT: { + int rc = restore_job(c, 0, payload, len, lsn); + if (rc == -EEXIST) + return 0; /* idempotent re-apply (replay / resync) — no hook */ + if (rc < 0) + return rc; + + /* Recover the jobid via a cheap raw capnp field read (read_Job only + * reads the fixed uint64 field — no allocation, unlike decode_Job) + * so the just-constructed job can be looked up for the notification + * hook without changing restore_job's own contract. */ + struct capn jc_ctx; + capn_init_mem(&jc_ctx, (const uint8_t*)payload, len, 0); + Job_ptr jp; + jp.p = capn_getp(capn_root(&jc_ctx), 0, 1); + struct Job cj; + read_Job(&cj, jp); + uint64_t jobid = cj.jobid; + capn_free(&jc_ctx); + + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)jobid, (void**)&j); + unlock(c->jobmgr); + if (!j) + return 0; /* defensive: should not happen after a successful restore */ + + void* jptr = j; + xlist_t jlist = { .count = 1, .caps = 1, .delta = 1, .data = &jptr }; + xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_NEW, c, &jlist); + return 0; + } + + case WAL_JOB_START: + case WAL_JOB_FINISH: + case WAL_JOB_FAIL: + case WAL_JOB_STOP: + case WAL_JOB_RESUME: + case WAL_JOB_UPDATE: + case WAL_JOB_RUSAGE: + case WAL_JOB_CKPT: + case WAL_JOB_PREASON: { + /* Every job-state fact type shares one wire shape — wal_jobstate_t + * {jobid, state} — and therefore one guarded core. See persist.h's + * doc comment on apply_committed_entry for why this does not + * dispatch to the type-specific apply_job_{start,finish,fail}_ + * transition() cores (their richer parameter structs are not + * present on this wire format). */ + if (len < sizeof(wal_jobstate_t)) + return 0; /* undersized payload — skip, mirrors apply_wal_entry */ + + const wal_jobstate_t* js = (const wal_jobstate_t*)payload; + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)js->jobid, (void**)&j); + unlock(c->jobmgr); + if (!j) + return -ENOENT; + + lock(j, RTTI_LOCK_WRITE); + job_transition_result_t result = + apply_job_update_transition(j, (int)js->state); + unlock(j); + + if (result != JOB_TRANSITION_APPLIED) + return 0; /* rejected-backward / no-op-duplicate — no hook */ + + xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_UPDATE, c, j); + return 0; + } + + case WAL_QUEUE_CTRL: { + if (len <= sizeof(wal_queuehdr_t)) + return 0; /* undersized payload — skip, mirrors apply_wal_entry */ + + const wal_queuehdr_t* qh = (const wal_queuehdr_t*)payload; + const char* name = + (const char*)((const uint8_t*)payload + sizeof(wal_queuehdr_t)); + + /* restore_queue_state's -ENOENT ("queue not yet built from config") + * and -EINVAL (bad name) are both benign skips here, matching + * apply_wal_entry's existing behavior of never fencing on this + * entry type's restore outcome. */ + if (restore_queue_state(c, name, qh->status) < 0) + return 0; + + queue_t* q = NULL; + lock(c->qmgr, RTTI_LOCK_READ); + key_find(c->qmgr, (void*)name, (void**)&q); + unlock(c->qmgr); + if (!q) + return 0; + + xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE, HOOK_AFTER_EVENT, + HOOK_DETAIL_QUEUE_UPDATE, c, q); + return 0; + } + + case WAL_WORKER_CTRL: { + if (len <= WORKER_WAL_HDR_SIZE) + return 0; /* undersized payload — skip, mirrors apply_wal_entry */ + + uint32_t hook_detail; + memcpy(&hook_detail, payload, sizeof(hook_detail)); + + int rc = worker_from_capnp(c, payload, len); + if (rc == -EEXIST) + return 0; /* idempotent re-apply (replay / resync) — no hook */ + if (rc < 0) + return rc; + + /* Recover the worker's name via a cheap raw capnp field read + * (read_Worker's capn_get_text is a borrowed, NUL-terminated pointer + * into the decode buffer — no allocation, unlike decode_Worker) so + * the just-restored worker can be looked up for the notification + * hook without changing worker_from_capnp's own contract. */ + const uint8_t* wpayload = (const uint8_t*)payload + WORKER_WAL_HDR_SIZE; + size_t wlen = len - WORKER_WAL_HDR_SIZE; + struct capn wc_ctx; + capn_init_mem(&wc_ctx, wpayload, wlen, 0); + Worker_ptr wp; + wp.p = capn_getp(capn_root(&wc_ctx), 0, 1); + struct Worker cw; + read_Worker(&cw, wp); + + worker_t* w = NULL; + lock(c->workermgr, RTTI_LOCK_READ); + key_find(c->workermgr, (void*)cw.name.str, (void**)&w); + unlock(c->workermgr); + capn_free(&wc_ctx); + if (!w) + return 0; /* defensive: should not happen after a successful restore */ + + switch (hook_detail) { + case HOOK_DETAIL_WORKER_JOIN: + xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER, + HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_JOIN, c, + w->host.name, w); + break; + case HOOK_DETAIL_WORKER_UPDATE: + xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER, + HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w); + break; + case HOOK_DETAIL_WORKER_DEAD: + xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER, + HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_DEAD, c, w); + break; + default: + break; /* unrecognized discriminator — mutation already applied, + * notification skipped rather than guessed at */ + } + return 0; + } + + default: + /* WAL_NOOP / WAL_CONFIG_MARKER / WAL_CHECKPOINT / unknown — no + * entity effect at this seam, no hook (unchanged from + * apply_wal_entry's existing treatment of these types). */ + return 0; + } +} + /* Read WAL segment files in LSN order; replay entries with lsn > from_lsn. * * Segment filename format: seg-.wal diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 482db297..8ef26b70 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -78,6 +78,7 @@ extern void test_ha_core_safety(void); 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); void setUp(void) { return; } @@ -225,6 +226,7 @@ int main(int argc, char* argv[]) TEST(test_ha_partition); TEST(test_persist_snapshot_entities); TEST(test_apply_seam); + TEST(test_apply_committed_entry); return UNITY_END(); } diff --git a/src/controller/testsuite/test_apply_committed_entry.c b/src/controller/testsuite/test_apply_committed_entry.c new file mode 100644 index 00000000..652fb36d --- /dev/null +++ b/src/controller/testsuite/test_apply_committed_entry.c @@ -0,0 +1,452 @@ +/* test_apply_committed_entry.c — raft-apply-decoupling task 2.2 + * + * Proves apply_committed_entry() (persist.c) is the single dispatch seam + * that turns ONE committed WAL entry into an in-memory mutation for every + * classified entry type, and fires that entry type's post-mutation + * notification hook exactly once, strictly after the mutation completes: + * + * - WAL_JOB_SUBMIT -> restore_job() + HOOK_DETAIL_JOB_NEW + * - WAL_JOB_START/... -> apply_job_update_transition() (2.1's guarded + * core) + HOOK_DETAIL_JOB_UPDATE + * - WAL_QUEUE_CTRL -> restore_queue_state() + HOOK_DETAIL_QUEUE_UPDATE + * - WAL_WORKER_CTRL -> worker_from_capnp() + HOOK_DETAIL_WORKER_JOIN + * + * and that a guard-rejected or idempotent-duplicate application performs no + * mutation and fires NO notification (Req 7.1/7.2). + * + * This task does NOT wire apply_committed_entry into any live call site + * (replay / apply pump are later tasks 2.3/6.1) — every test here calls the + * dispatch function directly. + * + * Fixture: a minimal cluster_t (hooksys + jobmgr/qmgr/workermgr) and a bare + * persist_t whose ONLY field apply_committed_entry reads is ->cluster — no + * real WAL/disk/repl/commit-engine setup needed (mirrors test_apply_seam.c's + * "bare fixture, no more than the function under test needs" philosophy). + */ + +#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. */ +static int g_tace_jobobj_ready; + +static cluster_t g_tace_c; +static persist_t g_tace_p; + +/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */ +static int g_tace_hook_fires; + +static int tace_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_tace_hook_fires++; + return 0; +} + +static void tace_setup(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*)"tace-probe") == NULL) + xstring_table_init(64); + + if (!g_tace_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_tace_jobobj_ready = 1; + } + + memset(&g_tace_c, 0, sizeof(g_tace_c)); + g_tace_c.myhost = &g_tace_c.myhost_info; + g_tace_c.myhost_info.name = (char*)"ctrl-test"; + g_tace_c.myhost_info.key = (char*)""; + + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_tace_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + g_tace_c.qmgr = makeInstance(&xkvmgr_type, &str_params); + g_tace_c.workermgr = makeInstance(&xkvmgr_type, &str_params); + g_tace_c.resmgr = makeInstance(&xkvmgr_type, &str_params); + TEST_ASSERT_NOT_NULL(g_tace_c.jobmgr); + TEST_ASSERT_NOT_NULL(g_tace_c.qmgr); + TEST_ASSERT_NOT_NULL(g_tace_c.workermgr); + TEST_ASSERT_NOT_NULL(g_tace_c.resmgr); + zarray_init(&g_tace_c.sharedinst, 16, NULL); + + memset(&g_tace_p, 0, sizeof(g_tace_p)); + g_tace_p.cluster = &g_tace_c; + + g_tace_hook_fires = 0; +} + +static void tace_teardown(void) +{ + zarray_fini(&g_tace_c.sharedinst); + if (g_tace_c.resmgr) dereference(g_tace_c.resmgr); + if (g_tace_c.workermgr) dereference(g_tace_c.workermgr); + if (g_tace_c.qmgr) dereference(g_tace_c.qmgr); + if (g_tace_c.jobmgr) dereference(g_tace_c.jobmgr); + xce_hook_fini(&g_tace_c.hooksys); +} + +/* ── Encode helpers (mirror the static capnp_encode_{job,worker} in + * persist.c — those are file-static, so tests keep local copies, exactly + * as test_egress_filter.c's ef_encode_job already does). ────────────── */ + +static int tace_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 tace_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; +} + +/* Look up a job_t wrapper by jobid (NULL if absent). */ +static job_t* tace_find_job(uint64_t jobid) +{ + job_t* j = NULL; + lock(g_tace_c.jobmgr, RTTI_LOCK_READ); + int found = key_find(g_tace_c.jobmgr, (void*)jobid, (void**)&j); + unlock(g_tace_c.jobmgr); + return (found > 0) ? j : NULL; +} + +/* Creates a job directly in the fixture's jobmgr, bypassing + * handle_jobsubmit/restore_job entirely — for seeding fact-transition tests. */ +static job_submit_data_t g_tace_seed_submit; + +static job_t* tace_seed_job(uint64_t jobid, uint32_t state) +{ + xce_job_t jdef; + + memset(&g_tace_seed_submit, 0, sizeof(g_tace_seed_submit)); + g_tace_seed_submit.queue = (char*)"q"; + g_tace_seed_submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.submit = &g_tace_seed_submit; + + return makeJob(g_tace_c.jobmgr, &jdef); +} + +/* ── WAL_JOB_SUBMIT ───────────────────────────────────────────────────── */ + +static void test_ace_job_submit_constructs_and_notifies(void) +{ + tace_setup(); + + job_submit_data_t submit; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"q"; + submit.user = (char*)"tester"; + + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = 501; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, tace_encode_job(&jdef, &payload, &plen)); + + TEST_ASSERT_EQUAL_INT(0, + xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, + tace_spy_hook, NULL)); + + int rc = apply_committed_entry(&g_tace_p, WAL_JOB_SUBMIT, 10, payload, + (uint32_t)plen); + TEST_ASSERT_EQUAL_INT(0, rc); + + job_t* j = tace_find_job(501); + TEST_ASSERT_NOT_NULL_MESSAGE(j, + "WAL_JOB_SUBMIT must construct the job via restore_job"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires, + "a fresh job submit must fire exactly one post-mutation notification"); + + free(payload); + tace_teardown(); + printf("test_ace_job_submit_constructs_and_notifies: PASS\n"); +} + +static void test_ace_job_submit_duplicate_is_noop_no_extra_notify(void) +{ + tace_setup(); + + 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 = 502; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, tace_encode_job(&jdef, &payload, &plen)); + + xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook, + NULL); + + TEST_ASSERT_EQUAL_INT(0, apply_committed_entry(&g_tace_p, WAL_JOB_SUBMIT, + 11, payload, + (uint32_t)plen)); + TEST_ASSERT_EQUAL_INT(1, g_tace_hook_fires); + + /* Re-apply the SAME entry (replay / resync redelivery) — must be a + * no-op: no error, no second job, no second notification (Req 7.1). */ + TEST_ASSERT_EQUAL_INT(0, apply_committed_entry(&g_tace_p, WAL_JOB_SUBMIT, + 11, payload, + (uint32_t)plen)); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires, + "an idempotent duplicate submit must not fire a second notification"); + + free(payload); + tace_teardown(); + printf("test_ace_job_submit_duplicate_is_noop_no_extra_notify: PASS\n"); +} + +/* ── WAL_JOB_START / job-state fact family ───────────────────────────── */ + +static void test_ace_job_state_transition_applies_forward_and_notifies(void) +{ + tace_setup(); + + job_t* j = tace_seed_job(601, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + + wal_jobstate_t js = { .jobid = 601, .state = JOB_STAT_RUN }; + + xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_tace_p, WAL_JOB_START, 20, &js, + sizeof(js)); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state, + "a forward job-state fact must apply via the 2.1 guarded core"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires, + "a forward job-state transition must fire exactly one notification"); + + tace_teardown(); + printf("test_ace_job_state_transition_applies_forward_and_notifies: " + "PASS\n"); +} + +static void +test_ace_job_state_transition_rejects_backward_suppresses_notify(void) +{ + tace_setup(); + + /* Job already terminal (as if FINISH already applied); a stale, + * previously-unapplied START must not regress it (Req 7.1/7.2). */ + job_t* j = tace_seed_job(602, JOB_STAT_DONE); + TEST_ASSERT_NOT_NULL(j); + + wal_jobstate_t js = { .jobid = 602, .state = JOB_STAT_RUN }; + + xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_tace_p, WAL_JOB_START, 21, &js, + sizeof(js)); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "a guard-rejected backward transition is a success no-op, not an " + "apply failure"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state, + "a rejected backward transition must preserve the terminal state"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_tace_hook_fires, + "a guard-rejected transition must fire NO notification"); + + tace_teardown(); + printf("test_ace_job_state_transition_rejects_backward_suppresses_" + "notify: PASS\n"); +} + +/* ── WAL_QUEUE_CTRL ───────────────────────────────────────────────────── */ + +static void test_ace_queue_ctrl_updates_and_notifies(void) +{ + tace_setup(); + + xce_queue_t qdef; + memset(&qdef, 0, sizeof(qdef)); + qdef.queue = (char*)"q1"; + TEST_ASSERT_NOT_NULL(makeQueue(g_tace_c.qmgr, &qdef)); + + const char* name = "q1"; + size_t name_len = strlen(name) + 1; + size_t total = sizeof(wal_queuehdr_t) + name_len; + uint8_t* pkt = malloc(total); + TEST_ASSERT_NOT_NULL(pkt); + wal_queuehdr_t qh = { .status = QUEUE_STAT_CLOSED }; + memcpy(pkt, &qh, sizeof(qh)); + memcpy(pkt + sizeof(qh), name, name_len); + + xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_QUEUE, tace_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_tace_p, WAL_QUEUE_CTRL, 30, pkt, + (uint32_t)total); + TEST_ASSERT_EQUAL_INT(0, rc); + + queue_t* q = NULL; + lock(g_tace_c.qmgr, RTTI_LOCK_READ); + key_find(g_tace_c.qmgr, (void*)name, (void**)&q); + unlock(g_tace_c.qmgr); + TEST_ASSERT_NOT_NULL(q); + TEST_ASSERT_TRUE_MESSAGE((q->queue.status & QUEUE_STAT_CLOSED) != 0, + "WAL_QUEUE_CTRL must apply via restore_queue_state"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires, + "a queue control apply must fire exactly one notification"); + + free(pkt); + tace_teardown(); + printf("test_ace_queue_ctrl_updates_and_notifies: PASS\n"); +} + +/* ── WAL_WORKER_CTRL ──────────────────────────────────────────────────── */ + +static void test_ace_worker_ctrl_join_restores_and_notifies(void) +{ + tace_setup(); + + /* copy_worker (wobj.c) rejects a NULL src->attrs with -EINVAL — a real + * WAL_WORKER_CTRL JOIN payload always carries real host attributes, so + * the fixture must too. */ + 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*)"w1"; + wdef.attrs = &attrs; + + uint8_t* wbuf = NULL; + size_t wsize = 0; + TEST_ASSERT_EQUAL_INT(0, tace_encode_worker(&wdef, &wbuf, &wsize)); + + uint32_t disc = (uint32_t)HOOK_DETAIL_WORKER_JOIN; + size_t total = sizeof(disc) + wsize; + uint8_t* pkt = malloc(total); + TEST_ASSERT_NOT_NULL(pkt); + memcpy(pkt, &disc, sizeof(disc)); + memcpy(pkt + sizeof(disc), wbuf, wsize); + free(wbuf); + + xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, tace_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_tace_p, WAL_WORKER_CTRL, 40, pkt, + (uint32_t)total); + TEST_ASSERT_EQUAL_INT(0, rc); + + worker_t* w = NULL; + lock(g_tace_c.workermgr, RTTI_LOCK_READ); + key_find(g_tace_c.workermgr, (void*)"w1", (void**)&w); + unlock(g_tace_c.workermgr); + TEST_ASSERT_NOT_NULL_MESSAGE(w, + "WAL_WORKER_CTRL must restore the worker via worker_from_capnp"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires, + "a worker control apply must fire exactly one notification"); + + free(pkt); + tace_teardown(); + printf("test_ace_worker_ctrl_join_restores_and_notifies: PASS\n"); +} + +/* ── Defensive / boundary ─────────────────────────────────────────────── */ + +static void test_ace_null_persist_returns_error(void) +{ + int rc = apply_committed_entry(NULL, WAL_JOB_UPDATE, 1, NULL, 0); + TEST_ASSERT_NOT_EQUAL(0, rc); + printf("test_ace_null_persist_returns_error: PASS\n"); +} + +/* ── Test group entry point ───────────────────────────────────────────── */ + +void test_apply_committed_entry(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_ace_job_submit_constructs_and_notifies); + RUN_TEST(test_ace_job_submit_duplicate_is_noop_no_extra_notify); + RUN_TEST(test_ace_job_state_transition_applies_forward_and_notifies); + RUN_TEST(test_ace_job_state_transition_rejects_backward_suppresses_notify); + RUN_TEST(test_ace_queue_ctrl_updates_and_notifies); + RUN_TEST(test_ace_worker_ctrl_join_restores_and_notifies); + RUN_TEST(test_ace_null_persist_returns_error); + UNITY_END(); +} diff --git a/src/include/persist.h b/src/include/persist.h index 784bd53e..fdb63361 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -739,6 +739,58 @@ int persist_worker_hook_fun(int type, int situation, int detail, int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, const void* payload, uint32_t payload_len); +/* apply_committed_entry — raft-apply-decoupling task 2.2, design.md + * "Apply Seam". The single dispatch function that mutates in-memory state + * for ONE committed WAL entry, shared (once tasks 2.3/6.1 wire their call + * sites) by restart replay and the live commit-advance apply pump. Routes: + * WAL_JOB_SUBMIT -> restore_job() (existing + * idempotent, jobid-addressable construction primitive; -EEXIST is a + * no-op re-apply, not an error) + * WAL_JOB_{START,FINISH,FAIL,STOP,RESUME, + * UPDATE,RUSAGE,CKPT,PREASON} -> apply_job_update_transition() + * (the guard-only core extracted in task 2.1) + * WAL_QUEUE_CTRL -> restore_queue_state() + * WAL_WORKER_CTRL -> worker_from_capnp() + * and fires the entry type's AFTER_EVENT notification hook exactly once, + * strictly after the mutation completes, and ONLY when a genuine (non- + * duplicate, non-guard-rejected) mutation actually occurred. + * + * Deviation from a literal per-type hook fan-out: the on-disk fact payload + * (wal_jobstate_t {jobid, state}) carries only the job's resulting state, not + * the original job_pidinfo_t / xce_jobfinish_t / job_failinfo_t a type- + * specific mark_job_* hook (HOOK_DETAIL_JOB_START/EXIT/FAIL) expects its + * subscribers to dereference. The real risk is sched.c's sched_job_start_fun/ + * sched_job_exit_fun/sched_job_fail_fun, which dereference info->jobid + * directly in a struct initializer — before even their `situation` switch, + * with no NULL guard (apiserver.c's equivalent subscribers are NOT at risk: + * they store the `info` pointer itself into the notify union (e.g. + * `.data = { .pidinfo = info }`) inside `case HOOK_AFTER_EVENT:`, without + * dereferencing any of its fields at this call site — unlike sched.c's + * unconditional `info->jobid` access in a top-level struct initializer). + * Every job-state fact type therefore routes through the + * guard-only apply_job_update_transition() core and fires + * HOOK_DETAIL_JOB_UPDATE, whose subscribers take only (cluster_t*, job_t*) — + * matching what this wire format can actually supply without fabricating + * data. No task in the current plan (2.3, 4.1, 4.2, 6.1, 6.2, 6.3) closes + * this gap — task 4.1 is exclusively about removing state mutation/hook- + * firing from worker-fact receipt-time handlers, not about enriching the + * wire format. Closing it would require either a design.md correction + * (formally accepting this HOOK_DETAIL_JOB_UPDATE substitution as final — + * design.md's Apply Seam section and Fact Ingestion Event Contract + * currently describe per-type hook dispatch this wire format cannot + * support) or a new task to enrich the WAL wire format for job-state facts. + * + * NOT this function's concern (owned by whichever caller drives it — the + * replay segment reader today, the live apply pump added by task 6.1): + * advancing p->current_lsn, oversized-payload rejection, WAL_CHECKPOINT + * skip. See apply_wal_entry's doc comment above for why those stay + * physical-layer-owned. + * + * Returns 0 on success (including a no-op idempotent re-apply or a guard- + * rejected/duplicate transition), negative on a genuine apply failure. */ +int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, + const void* payload, uint32_t len); + /* Load a snapshot file from path, set p->checkpoint_lsn and p->current_lsn. * lsn hint is informational — the actual LSN is read from the file header. *