wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(raft-apply-decoupling): extract guarded fact transition logic into reusable cores (task 2.1)

Browse source 
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).
This commit is contained in:
Rongsong Shen 2026-07-03 15:26:22 +00:00
parent 2d9f1e947a
commit 41d36204d5
6 changed files with 480 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.kiro/specs/raft-apply-decoupling/tasks.md
View file

@ -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 <sha> -- .` to inspect an old commit in place — it silently reverts working-tree files against current HEAD; use `git worktree add <path> <sha>` instead. Both lessons are now in `.wolf/cerebrum.md` Do-Not-Repeat and `.wolf/buglog.json` bug-1407.

26
src/controller/include/cluster.h
View file

@ -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*);

152
src/controller/job.c
View file

@ -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);

1
src/controller/meson.build
View file

@ -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',

2
src/controller/testsuite/controller_test.c
View file

@ -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();
}

324
src/controller/testsuite/test_apply_seam.c Normal file
View file

@ -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 <rtti.h>
#include <errno.h>
#include <stdbool.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#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();
}