diff --git a/.kiro/specs/ha-commit-barrier-rollout/tasks.md b/.kiro/specs/ha-commit-barrier-rollout/tasks.md index c4eae15..be4a7a8 100644 --- a/.kiro/specs/ha-commit-barrier-rollout/tasks.md +++ b/.kiro/specs/ha-commit-barrier-rollout/tasks.md @@ -84,7 +84,7 @@ - Observable completion: new cases FAIL (red) against the unguarded `mark_job_start/finish/fail/update` (job.c:583/638/696/726 anchors) - _Requirements: 5.1, 5.2, 5.3_ -- [ ] 6.2 Implement the guards to green +- [x] 6.2 Implement the guards to green - Add transition guards at the top of each `mark_job_*` (under the caller-held job WRITE lock, before BEFORE_EVENT fires): duplicate → log + return no-op; terminal → non-terminal → WARN + reject preserving state; forward transitions unchanged - Audit all callers (`on_job_*`, `job_ctrl`, `update_job_rusages`) for the new return semantics - Observable completion: all task 6.1 tests PASS; existing job lifecycle tests green diff --git a/.wolf/memory.md b/.wolf/memory.md index 605a44e..51c354b 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -1018,3 +1018,9 @@ | 04:18 | task 6.1 review: RED tests for mark_job_* guards verified | test_commit_barrier_reply.c | APPROVED — 22/3 group, 508/3 census, boundary clean | ~38k | | 21:19 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | inline fix | ~13 | | 21:19 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | modified 2() | ~243 | +| 21:23 | Edited src/controller/job.c | modified job_state_is_terminal() | ~330 | +| 21:23 | Edited src/controller/job.c | added 1 condition(s) | ~224 | +| 21:24 | Edited src/controller/job.c | added 1 condition(s) | ~212 | +| 21:24 | Edited src/controller/job.c | added 2 condition(s) | ~294 | +| 04:37 | task 6.2: mark_job_* transition guards (dup no-op / backward reject / forward merge preserved) | src/controller/job.c | 22 ctrl tests 0 fail, 5 guard tests GREEN | ~14k | +| 21:54 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | inline fix | ~11 | diff --git a/src/controller/job.c b/src/controller/job.c index 5fe05cd..9a960c4 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -19,6 +19,28 @@ static int mark_job_finish(cluster_t*, job_t*, xce_jobfinish_t*); static int mark_job_fail(cluster_t*, job_t*, job_failinfo_t*); static int mark_job_update(cluster_t*, job_t*, int); +/* A job is terminal once it carries a DONE or EXIT bit. Terminal jobs reject + * backward (terminal -> non-terminal) transitions and treat re-applied + * terminal transitions as duplicates; forward bit-OR merges (PDONE, PERR, + * suspend bits) on a terminal job remain legal. */ +#define JOB_STATE_TERMINAL_BITS (JOB_STAT_DONE | JOB_STAT_EXIT) + +static inline bool job_state_is_terminal(int state) +{ + return (state & JOB_STATE_TERMINAL_BITS) != 0; +} + +/* Shared backward predicate for START and UPDATE (Req 5.2): a transition is + * backward when the job is already terminal but the requested next state drops + * every terminal bit it currently holds (e.g. START overwriting DONE with a + * bare JOB_STAT_RUN, or an UPDATE clearing the terminal bits). A forward merge + * that preserves at least one terminal bit (postdone OR'ing PDONE) is NOT + * backward. */ +static inline bool job_state_is_backward(int cur, int next) +{ + return job_state_is_terminal(cur) && !job_state_is_terminal(next); +} + static job_t* create_job(cluster_t* c, xce_job_t* j) { job_t* job = NULL; @@ -591,6 +613,20 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request) static int mark_job_start(cluster_t* c, 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. */ + 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; + } + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_START, c, info, job); /* update job information */ @@ -646,6 +682,19 @@ int on_job_start(cluster_t* c, job_pidinfo_t* info) static int mark_job_finish(cluster_t* c, 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. */ + if (job_state_is_terminal(cur)) { + INFO("job %ld: duplicate FINISH on terminal state 0x%x — no-op", + job->job.jobid, cur); + return 0; + } + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_EXIT, c, finish, job); /* update job information */ @@ -734,6 +783,26 @@ static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo) static int mark_job_update(cluster_t* c, 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. */ + 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; + } + if (state == cur) { + INFO("job %ld: duplicate UPDATE to state 0x%x — no-op", job->job.jobid, + cur); + return 0; + } + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_UPDATE, c, job); /* update job state */