From 41d36204d5312864514cc9f2375d0048691fa16c Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 3 Jul 2026 15:26:22 +0000 Subject: [PATCH] feat(raft-apply-decoupling): extract guarded fact transition logic into reusable cores (task 2.1) Split mark_job_start/finish/fail/update in job.c into four hook-free core functions (apply_job_*_transition) containing only the monotonic-rank guard and mutation. Wrappers now peek the same guard to decide hook-firing, then delegate to the core, preserving exact pre-extraction hook-firing timing. Prerequisite for task 2.2's apply_committed_entry() dispatch and the ordered apply pump (6.1). --- .kiro/specs/raft-apply-decoupling/tasks.md | 3 +- src/controller/include/cluster.h | 26 ++ src/controller/job.c | 152 ++++++++-- src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_apply_seam.c | 324 +++++++++++++++++++++ 6 files changed, 480 insertions(+), 28 deletions(-) create mode 100644 src/controller/testsuite/test_apply_seam.c diff --git a/.kiro/specs/raft-apply-decoupling/tasks.md b/.kiro/specs/raft-apply-decoupling/tasks.md index a5e4776..ea6e18b 100644 --- a/.kiro/specs/raft-apply-decoupling/tasks.md +++ b/.kiro/specs/raft-apply-decoupling/tasks.md @@ -39,7 +39,7 @@ - _Depends: 1.4_ - [ ] 2. Shared apply seam for committed entries -- [ ] 2.1 (P) Extract the guarded state-transition logic used by fact handling into a form usable outside the receipt-time call site +- [x] 2.1 (P) Extract the guarded state-transition logic used by fact handling into a form usable outside the receipt-time call site - Keep the existing monotonic-rank checks (reject backward transitions, no-op duplicate terminal transitions) intact and unchanged in behavior - Decouple the transition logic from firing its own hook inline, so hook-firing can be relocated by a later task without touching this logic again - Observable: the extracted transition logic can be invoked directly by a test without going through the worker-report receipt path, and produces the same guard behavior as before extraction @@ -193,4 +193,5 @@ - **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.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/include/cluster.h b/src/controller/include/cluster.h index fc881d5..73cc3e0 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -110,6 +110,32 @@ extern void unlock_collect_queues(xlist_t*); extern void unlock_collect_jobs(xlist_t*); extern void unlock_collect_workers(xlist_t*); +/* Outcome of a guarded job state-transition core (raft-apply-decoupling + * task 2.1, Req 3.3/7.1/7.2): APPLIED means the mutation happened; + * REJECTED_BACKWARD/NOOP_DUPLICATE mean the monotonic-rank guard held the + * transition back and no mutation occurred. These cores fire no hook + * themselves — hook-firing is the caller's responsibility, so it can be + * relocated (to the apply seam) without touching this guard logic again. */ +typedef enum { + JOB_TRANSITION_APPLIED = 0, + JOB_TRANSITION_REJECTED_BACKWARD, + JOB_TRANSITION_NOOP_DUPLICATE, +} job_transition_result_t; + +/* Guarded state-transition cores extracted from mark_job_start/finish/fail/ + * update (job.c). Each applies exactly one fact's mutation to an + * already-locked job_t, subject to the same monotonic-rank guard the + * receipt-time wrapper used to apply inline. Callable directly — e.g. by a + * test, or later by the shared apply seam — without going through the + * worker-report receipt path. */ +extern job_transition_result_t apply_job_start_transition(job_t*, + job_pidinfo_t*); +extern job_transition_result_t apply_job_finish_transition(job_t*, + xce_jobfinish_t*); +extern job_transition_result_t apply_job_fail_transition(job_t*, + job_failinfo_t*); +extern job_transition_result_t apply_job_update_transition(job_t*, int); + extern int on_job_start(cluster_t*, job_pidinfo_t*); extern int on_job_finish(cluster_t*, xce_jobfinish_t*); extern int on_job_fail(cluster_t*, job_failinfo_t*); diff --git a/src/controller/job.c b/src/controller/job.c index 3227dc0..2c8e7dc 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -661,24 +661,29 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) return 0; } -static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info) +/* Guarded state-transition core for a worker-reported START, extracted from + * mark_job_start (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains + * ONLY the monotonic-rank guard and the state mutation — it fires NO hook + * itself, so hook-firing can be relocated by a later task (the shared apply + * seam) without touching this guard logic again. Safe to invoke directly, + * bypassing the worker-report receipt path (on_job_start) entirely, as long + * as the caller already holds `job` under RTTI_LOCK_WRITE. */ +job_transition_result_t apply_job_start_transition(job_t* job, + job_pidinfo_t* info) { int cur = job->job.state; /* Transition guard (Req 5.2): START sets the non-terminal JOB_STAT_RUN. A * START on an already-terminal job (stale/duplicate worker report after - * the job finished) is a backward transition — reject it, preserve the - * terminal state, and fire NO hook (no WAL append). Return success-no-op - * so the caller (on_job_start) still reports success. */ + * the job finished) is a backward transition — reject it and preserve + * the terminal state. */ if (job_state_is_backward(cur, JOB_STAT_RUN)) { WARN("job %ld: rejecting backward START on terminal state 0x%x — " "preserving terminal state", job->job.jobid, cur); - return 0; + return JOB_TRANSITION_REJECTED_BACKWARD; } - xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, - HOOK_DETAIL_JOB_START, c, info, job); /* update job information */ job->job.state = JOB_STAT_RUN; job->job.start_time = time(NULL); @@ -696,6 +701,27 @@ static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info) job->job.pidinfo->exdata = NULL; } + return JOB_TRANSITION_APPLIED; +} + +static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info) +{ + /* Peek the identical guard apply_job_start_transition() re-checks + * internally, so this extraction changes NO observable behavior: today + * no hook fires on a rejected transition, and BEFORE_EVENT strictly + * precedes the mutation. The peek is a pure read of job->job.state, and + * every caller holds `job` under RTTI_LOCK_WRITE, so it cannot race with + * the core's own (redundant, defensive) re-check of the same guard. */ + if (job_state_is_backward(job->job.state, JOB_STAT_RUN)) { + apply_job_start_transition(job, info); /* logs + no-op, unchanged */ + return 0; + } + + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, + HOOK_DETAIL_JOB_START, c, info, job); + + apply_job_start_transition(job, info); + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_START, c, info, job); @@ -730,23 +756,28 @@ int on_job_start(cluster_t* c, job_pidinfo_t* info) return 0; } -static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish) +/* Guarded state-transition core for a worker-reported FINISH, extracted from + * mark_job_finish (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains + * ONLY the monotonic-rank guard and the state mutation — fires NO hook, so + * hook-firing can be relocated later without touching this guard logic + * again. Safe to invoke directly, bypassing the worker-report receipt path + * (on_job_finish) entirely, as long as the caller already holds `job` under + * RTTI_LOCK_WRITE. */ +job_transition_result_t apply_job_finish_transition(job_t* job, + xce_jobfinish_t* finish) { int cur = job->job.state; /* Transition guard (Req 5.1): FINISH drives the job to a terminal state * (DONE on exit_code 0, else EXIT). If the job is already terminal the - * FINISH has already been applied (retransmit / WAL replay) — log it and - * return a success-no-op WITHOUT firing the AFTER_EVENT hook, so no second - * WAL entry is appended and current_lsn does not advance. */ + * FINISH has already been applied (retransmit / WAL replay) — treat it + * as a no-op duplicate so no second WAL entry is appended. */ if (job_state_is_terminal(cur)) { INFO("job %ld: duplicate FINISH on terminal state 0x%x — no-op", job->job.jobid, cur); - return 0; + return JOB_TRANSITION_NOOP_DUPLICATE; } - xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, - HOOK_DETAIL_JOB_EXIT, c, finish, job); /* update job information */ if (finish->exit_code == 0) { job->job.state = JOB_STAT_DONE; @@ -767,6 +798,27 @@ static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish) job->job.run_time = job->job.finish_time - job->job.start_time; + return JOB_TRANSITION_APPLIED; +} + +static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish) +{ + /* Peek the identical guard apply_job_finish_transition() re-checks + * internally, so this extraction changes NO observable behavior: today + * no hook fires on a duplicate-terminal FINISH, and BEFORE_EVENT strictly + * precedes the mutation. The peek is a pure read of job->job.state, and + * every caller holds `job` under RTTI_LOCK_WRITE, so it cannot race with + * the core's own (redundant, defensive) re-check of the same guard. */ + if (job_state_is_terminal(job->job.state)) { + apply_job_finish_transition(job, finish); /* logs + no-op, unchanged */ + return 0; + } + + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, + HOOK_DETAIL_JOB_EXIT, c, finish, job); + + apply_job_finish_transition(job, finish); + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_EXIT, c, finish, job); @@ -801,12 +853,19 @@ int on_job_finish(cluster_t* c, xce_jobfinish_t* finish) return 0; } -static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo) +/* State-transition core for a worker-reported FAIL, extracted from + * mark_job_fail (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains + * ONLY the state mutation — fires NO hook, so hook-firing can be relocated + * later without touching this logic again. mark_job_fail has no monotonic- + * rank guard today (unlike START/FINISH/UPDATE): a FAIL always applies, + * merging JOB_STAT_PERR onto an existing DONE/EXIT terminal state rather + * than being rejected — that existing merge behavior is preserved unchanged + * by this extraction, not newly added. Safe to invoke directly, bypassing + * the worker-report receipt path (on_job_fail) entirely, as long as the + * caller already holds `job` under RTTI_LOCK_WRITE. */ +job_transition_result_t apply_job_fail_transition(job_t* job, + job_failinfo_t* failinfo) { - xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, - HOOK_DETAIL_JOB_FAIL, c, failinfo, job); - /* update job information */ - switch (failinfo->stage) { case JOB_EXEC_STAGE_Q_POST_EXEC: case JOB_EXEC_STAGE_POST_EXEC: @@ -825,38 +884,77 @@ static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo) job->job.finish_time = time(NULL); + return JOB_TRANSITION_APPLIED; +} + +static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo) +{ + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, + HOOK_DETAIL_JOB_FAIL, c, failinfo, job); + + apply_job_fail_transition(job, failinfo); + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_FAIL, c, failinfo, job); return 0; } -static int mark_job_update(cluster_t* c, job_t* job, int state) +/* Guarded state-transition core for an UPDATE, extracted from + * mark_job_update (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains + * ONLY the monotonic-rank guards and the state mutation — fires NO hook, so + * hook-firing can be relocated later without touching this guard logic + * again. Safe to invoke directly, bypassing the mark_job_update call sites + * entirely, as long as the caller already holds `job` under + * RTTI_LOCK_WRITE. */ +job_transition_result_t apply_job_update_transition(job_t* job, int state) { int cur = job->job.state; /* Transition guard (Req 5.1/5.2), sharing the backward predicate with * START: an UPDATE that clears every terminal bit a terminal job holds is - * backward — reject it, preserve the terminal state, fire no hook. A - * forward merge that keeps a terminal bit (postdone OR'ing PDONE) is not - * backward and proceeds. A no-op update (requested state already current) - * is logged and skipped so it appends no duplicate WAL entry. */ + * backward — reject it and preserve the terminal state. A forward merge + * that keeps a terminal bit (postdone OR'ing PDONE) is not backward and + * proceeds. A no-op update (requested state already current) is a + * duplicate — skip it so it appends no duplicate WAL entry. */ if (job_state_is_backward(cur, state)) { WARN("job %ld: rejecting backward UPDATE 0x%x -> 0x%x — preserving " "terminal state", job->job.jobid, cur, state); - return 0; + return JOB_TRANSITION_REJECTED_BACKWARD; } if (state == cur) { INFO("job %ld: duplicate UPDATE to state 0x%x — no-op", job->job.jobid, cur); + return JOB_TRANSITION_NOOP_DUPLICATE; + } + + /* update job state */ + job->job.state = state; + + return JOB_TRANSITION_APPLIED; +} + +static int mark_job_update(cluster_t* c, job_t* job, int state) +{ + int cur = job->job.state; + + /* Peek the identical guards apply_job_update_transition() re-checks + * internally, so this extraction changes NO observable behavior: today + * no hook fires on a rejected/duplicate transition, and BEFORE_EVENT + * strictly precedes the mutation. The peek is a pure read of + * job->job.state, and every caller holds `job` under RTTI_LOCK_WRITE, so + * it cannot race with the core's own (redundant, defensive) re-check of + * the same guards. */ + if (job_state_is_backward(cur, state) || (state == cur)) { + apply_job_update_transition(job, state); /* logs + no-op, unchanged */ return 0; } xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_UPDATE, c, job); - /* update job state */ - job->job.state = state; + + apply_job_update_transition(job, state); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_UPDATE, c, job); diff --git a/src/controller/meson.build b/src/controller/meson.build index 072dde1..8dd1b03 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -157,6 +157,7 @@ controller_unit_test_src = [ 'testsuite/test_snapdiff.c', 'testsuite/snapdiff.c', 'testsuite/test_persist_snapshot_entities.c', + 'testsuite/test_apply_seam.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index e2274b5..482db29 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -77,6 +77,7 @@ extern void test_ha_dispatch_integration(void); 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); void setUp(void) { return; } @@ -223,6 +224,7 @@ int main(int argc, char* argv[]) TEST(test_ha_core_safety); TEST(test_ha_partition); TEST(test_persist_snapshot_entities); + TEST(test_apply_seam); return UNITY_END(); } diff --git a/src/controller/testsuite/test_apply_seam.c b/src/controller/testsuite/test_apply_seam.c new file mode 100644 index 0000000..d6ddb17 --- /dev/null +++ b/src/controller/testsuite/test_apply_seam.c @@ -0,0 +1,324 @@ +/* test_apply_seam.c — raft-apply-decoupling task 2.1 + * + * Proves the guarded state-transition cores extracted from mark_job_start/ + * finish/fail/update (job.c) are directly callable WITHOUT going through the + * worker-report receipt path (on_job_start/on_job_finish/on_job_fail) and + * reproduce the exact same guard behavior mark_job_* had before extraction: + * - reject a backward transition (terminal -> non-terminal), state + * preserved (Req 5.2 guard, START/UPDATE) + * - no-op a duplicate terminal transition, state preserved (Req 5.1 guard, + * FINISH/UPDATE) + * - apply a forward transition normally, including a forward merge onto an + * already-terminal job (postdone-style OR, and the FAIL PERR-merge, + * neither of which the guard may over-reject) + * + * Fixture note: apply_job_*_transition() take only a job_t* plus the fact's + * payload struct — never a cluster_t* — because task 2.1 decouples hook- + * firing (the only thing that needed cluster_t->hooksys) out of these cores + * entirely. So the fixture here needs just a bare jobmgr + job_t, none of + * test_commit_barrier_reply.c's full cluster_t/persist/repl harness. + */ + +#include + +#include +#include +#include +#include + +#include + +#include "cluster.h" +#include "jobj.h" +#include "xce/types.h" +#include "xstring.h" + +/* jobj.c's jobsub/startinfo tables and the process-global xstring table are + * shared across every test file in the controller_unit_test binary — guard + * idempotently so this file runs standalone or alongside the others. */ +static int g_tas_jobobj_ready; + +static var g_tas_jobmgr; +static job_submit_data_t g_tas_submit; + +/* Bare jobmgr fixture: no cluster_t, no hooksys, no persist — the extracted + * apply_job_*_transition() cores need none of that. */ +static void tas_setup(void) +{ + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 16, + .delta = 16 }; + + if (makeString((char*)"tas-probe") == NULL) + xstring_table_init(64); + + if (!g_tas_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(16, 16)); + g_tas_jobobj_ready = 1; + } + + g_tas_jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(g_tas_jobmgr); +} + +static void tas_teardown(void) +{ + if (g_tas_jobmgr != NULL) { + dereference(g_tas_jobmgr); + g_tas_jobmgr = NULL; + } +} + +/* Creates a job directly in the bare jobmgr at the given initial state, + * bypassing handle_jobsubmit/create_job/on_job_* entirely. */ +static job_t* tas_make_job(uint64_t jobid, uint32_t state) +{ + xce_job_t jdef; + + memset(&g_tas_submit, 0, sizeof(g_tas_submit)); + g_tas_submit.queue = (char*)"q"; + g_tas_submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.submit = &g_tas_submit; + + return makeJob(g_tas_jobmgr, &jdef); +} + +/* ── apply_job_start_transition ──────────────────────────────────────── */ + +static void test_apply_start_transition_applies_forward(void) +{ + static unsigned int pids[2] = { 111, 112 }; + job_pidinfo_t info; + job_t* j; + job_transition_result_t rc; + + tas_setup(); + j = tas_make_job(1, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + + memset(&info, 0, sizeof(info)); + info.jobid = 1; + info.npids = 2; + info.pids = pids; + + /* Act directly on the job_t — no cluster_t, no on_job_start, no jobmgr + * lookup/lock dance. */ + rc = apply_job_start_transition(j, &info); + + TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_APPLIED, rc, + "a forward START on a non-terminal job must apply"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + TEST_ASSERT_NOT_NULL(j->job.pidinfo); + TEST_ASSERT_EQUAL_INT(2, j->job.pidinfo->npids); + + tas_teardown(); + printf("test_apply_start_transition_applies_forward: PASS\n"); +} + +static void test_apply_start_transition_rejects_backward_on_terminal(void) +{ + job_pidinfo_t info; + job_t* j; + job_transition_result_t rc; + + tas_setup(); + j = tas_make_job(2, JOB_STAT_DONE); + TEST_ASSERT_NOT_NULL(j); + + memset(&info, 0, sizeof(info)); + info.jobid = 2; + + rc = apply_job_start_transition(j, &info); + + TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_REJECTED_BACKWARD, rc, + "a START on an already-terminal job must be rejected as backward " + "(Req 5.2)"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state, + "a rejected backward START must preserve the terminal state"); + TEST_ASSERT_NULL_MESSAGE(j->job.pidinfo, + "a rejected backward START must perform no mutation at all"); + + tas_teardown(); + printf("test_apply_start_transition_rejects_backward_on_terminal: PASS\n"); +} + +/* ── apply_job_finish_transition ─────────────────────────────────────── */ + +static void test_apply_finish_transition_applies_forward(void) +{ + xce_jobfinish_t fin; + job_t* j; + job_transition_result_t rc; + + tas_setup(); + j = tas_make_job(3, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + j->job.start_time = time(NULL) - 5; + + memset(&fin, 0, sizeof(fin)); + fin.jobid = 3; + fin.exit_code = 0; + + rc = apply_job_finish_transition(j, &fin); + + TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_APPLIED, rc, + "a forward FINISH on a running job must apply"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, j->job.state); + TEST_ASSERT_NOT_NULL(j->job.exit_info); + TEST_ASSERT_EQUAL_INT(0, j->job.exit_info->exitcode); + + tas_teardown(); + printf("test_apply_finish_transition_applies_forward: PASS\n"); +} + +static void test_apply_finish_transition_noop_on_terminal_duplicate(void) +{ + xce_jobfinish_t fin; + job_t* j; + time_t finish_time_before; + job_transition_result_t rc; + + tas_setup(); + j = tas_make_job(4, JOB_STAT_DONE); + TEST_ASSERT_NOT_NULL(j); + j->job.finish_time = 12345; /* sentinel — must survive the no-op */ + finish_time_before = j->job.finish_time; + + memset(&fin, 0, sizeof(fin)); + fin.jobid = 4; + fin.exit_code = 0; + + rc = apply_job_finish_transition(j, &fin); + + TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_NOOP_DUPLICATE, rc, + "a duplicate FINISH on an already-terminal job must no-op (Req 5.1)"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state, + "a no-op duplicate FINISH must leave the job's state unchanged"); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)finish_time_before, + (int)j->job.finish_time, + "a no-op duplicate FINISH must perform no mutation at all"); + + tas_teardown(); + printf("test_apply_finish_transition_noop_on_terminal_duplicate: PASS\n"); +} + +/* ── apply_job_update_transition ─────────────────────────────────────── */ + +static void test_apply_update_transition_rejects_backward_on_terminal(void) +{ + job_t* j; + job_transition_result_t rc; + + tas_setup(); + j = tas_make_job(5, JOB_STAT_DONE); + TEST_ASSERT_NOT_NULL(j); + + /* An UPDATE that clears the terminal bit is backward (shared predicate + * with START, Req 5.2). */ + rc = apply_job_update_transition(j, JOB_STAT_RUN); + + TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_REJECTED_BACKWARD, rc, + "an UPDATE clearing a terminal bit must be rejected as backward"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state, + "a rejected backward UPDATE must preserve the terminal state"); + + tas_teardown(); + printf("test_apply_update_transition_rejects_backward_on_terminal: PASS\n"); +} + +static void test_apply_update_transition_noop_on_duplicate(void) +{ + job_t* j; + job_transition_result_t rc; + + tas_setup(); + j = tas_make_job(6, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + + rc = apply_job_update_transition(j, JOB_STAT_RUN); /* state == cur */ + + TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_NOOP_DUPLICATE, rc, + "an UPDATE requesting the already-current state must no-op " + "(Req 5.1)"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + + tas_teardown(); + printf("test_apply_update_transition_noop_on_duplicate: PASS\n"); +} + +static void test_apply_update_transition_forward_merge_on_terminal_applies( + void) +{ + job_t* j; + job_transition_result_t rc; + + tas_setup(); + j = tas_make_job(7, JOB_STAT_DONE); + TEST_ASSERT_NOT_NULL(j); + + /* A forward merge that keeps the terminal bit (postdone OR'ing PDONE) + * must NOT be over-rejected by the backward guard. */ + rc = apply_job_update_transition(j, JOB_STAT_DONE | JOB_STAT_PDONE); + + TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_APPLIED, rc, + "a forward merge preserving the terminal bit must apply, not be " + "rejected as backward"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE | JOB_STAT_PDONE, j->job.state); + + tas_teardown(); + printf("test_apply_update_transition_forward_merge_on_terminal_applies: " + "PASS\n"); +} + +/* ── apply_job_fail_transition (no monotonic-rank guard, pin) ────────── */ + +static void test_apply_fail_transition_merges_perr_on_terminal(void) +{ + job_failinfo_t failinfo; + job_t* j; + job_transition_result_t rc; + + tas_setup(); + j = tas_make_job(8, JOB_STAT_DONE); + TEST_ASSERT_NOT_NULL(j); + + memset(&failinfo, 0, sizeof(failinfo)); + failinfo.jobid = 8; + failinfo.stage = JOB_EXEC_STAGE_POST_EXEC; + failinfo.code = 1; + failinfo.detail = (char*)"post-exec fail after finish"; + + rc = apply_job_fail_transition(j, &failinfo); + + TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_APPLIED, rc, + "mark_job_fail has no monotonic-rank guard — a FAIL always applies"); + TEST_ASSERT_TRUE_MESSAGE((j->job.state & JOB_STAT_DONE) != 0, + "fail-after-finish must keep the DONE terminal bit (PERR-merge, " + "unchanged by this extraction)"); + TEST_ASSERT_TRUE_MESSAGE((j->job.state & JOB_STAT_PERR) != 0, + "fail-after-finish must OR JOB_STAT_PERR onto the existing terminal " + "state"); + + tas_teardown(); + printf("test_apply_fail_transition_merges_perr_on_terminal: PASS\n"); +} + +/* ── Test group entry point ──────────────────────────────────────────── */ + +void test_apply_seam(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_apply_start_transition_applies_forward); + RUN_TEST(test_apply_start_transition_rejects_backward_on_terminal); + RUN_TEST(test_apply_finish_transition_applies_forward); + RUN_TEST(test_apply_finish_transition_noop_on_terminal_duplicate); + RUN_TEST(test_apply_update_transition_rejects_backward_on_terminal); + RUN_TEST(test_apply_update_transition_noop_on_duplicate); + RUN_TEST(test_apply_update_transition_forward_merge_on_terminal_applies); + RUN_TEST(test_apply_fail_transition_merges_perr_on_terminal); + UNITY_END(); +}