lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(worker-rejoin-resync): reconcile unrecognized job to pending (task 2.1)

Browse source 
Add record_job_reconcile_pending_fact(cluster_t*, job_t*) to job.c:
eagerly applies apply_job_update_transition(job, JOB_STAT_PEND) (real,
live mutation, not a deferred peek -- matching dispatch_a_job_'s
established pattern, not record_job_finish_fact's), releases
job->job.allocs on an applied (non-no-op) transition, then durably
appends via the generic persist_submit_job_fact(WAL_JOB_UPDATE). Caller
holds job's lock throughout and is responsible for calling
persist_notify_job_updated_already_persisted() after unlocking -- this
function deliberately does not call it itself.

Closes bug-1631 at this new call site: a RUN job with a real allocation
reconciled to PEND (Requirement 4.6, worker reports it unrecognized)
would otherwise carry a stale allocation into its next dispatch attempt,
exactly the gap bug-1631 documented as dormant until something re-admits
an already-dispatched job to the scheduler's PEND pool -- this feature
is that path. dispatch_a_job_ itself is unchanged (still correct for a
job's first-ever dispatch, the only case reachable there).

3 tests: allocs release + eager-mutation-before-unlock proof, no-op on
already-pending, and the Requirement 4.4 terminal-state-preservation
proof (rejects overwriting JOB_STAT_DONE).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
This commit is contained in:
Rongsong Shen 2026-09-01 10:13:08 +00:00
parent eda2eb1bf0
commit cfe66dde16
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
7 changed files with 433 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -183,6 +183,34 @@ extern job_transition_result_t apply_job_update_transition(job_t*, int);
* RTTI_LOCK_WRITE. */
extern job_transition_result_t apply_job_postdone_transition(job_t*);
/* worker-rejoin-resync task 2.1 (Req 4.6, 5.4, 7.1-7.3): returns a job the
* worker reports as unrecognized back to PEND, releasing any resource
* allocation it was carrying (bug-1631) so a later redispatch never
* inherits a stale allocation from the prior dispatch. Follows
* dispatch_a_job_()'s own eager-mutation-plus-explicit-notify pattern, NOT
* record_job_finish_fact's deferred one -- see design.md's "Controller:
* job.c" Components section for the full rationale.
*
* Precondition: caller holds `job` under RTTI_LOCK_WRITE on entry and does
* not release it until this call returns -- matching apply_job_update_
* transition's own calling convention (e.g. mark_job_update's callers).
* This function does not lock/unlock `job` itself.
*
* Returns the job_transition_result_t of the underlying apply_job_update_
* transition(job, JOB_STAT_PEND) call, cast to int, so the caller can tell
* JOB_TRANSITION_APPLIED apart from JOB_TRANSITION_REJECTED_BACKWARD/
* JOB_TRANSITION_NOOP_DUPLICATE exactly as that guard's other callers
* already do. Only on JOB_TRANSITION_APPLIED does this function release
* job->job.allocs (if non-NULL) and durably append the fact -- a rejected
* or no-op outcome touches neither (Req 4.3/4.4).
*
* Unlike record_job_finish_fact/record_job_fail_fact, the caller (not this
* function) must call persist_notify_job_updated_already_persisted(c, job)
* itself, AFTER releasing the write lock, exactly once, whenever this
* returns JOB_TRANSITION_APPLIED -- omitting it reproduces bug-1679 at this
* call site. */
extern int record_job_reconcile_pending_fact(cluster_t*, job_t*);
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*);

65
src/controller/job.c
View file

@ -2866,6 +2866,71 @@ static int mark_job_update(cluster_t* c, job_t* job, int state)
return 0;
}
/* worker-rejoin-resync task 2.1 (Req 4.1-4.6, 5.4, 7.1-7.3): returns a job
* the worker reports as unrecognized back to PEND, releasing any resource
* allocation it was carrying so a subsequent redispatch attempt never
* inherits a stale allocation from a prior dispatch (bug-1631, confirmed
* live and deliberately left unfixed until this feature provided the first
* real re-admission path -- see design.md's Error Handling section).
*
* Follows dispatch_a_job_()'s own established pattern (job.c, its `if (rc
* >= 0) { ... }` success block): eager, synchronous mutation under the
* write lock the caller already holds, immediately followed by a durable
* WAL append, immediately followed (by the CALLER, after unlocking) by an
* explicit persist_notify_job_updated_already_persisted() call -- NOT
* record_job_finish_fact's/record_job_fail_fact's purely-deferred pattern
* (mutation withheld until apply_committed_entry replays the committed
* fact). The fact recorded here is something the worker already confirmed
* (or, in this case, failed to confirm) really happened -- ground truth
* already changed outside this controller -- which is exactly dispatch_a_
* job_'s own justification for going eager rather than deferred (design.md,
* "Controller: job.c" Components section).
*
* Precondition: caller holds `job` under RTTI_LOCK_WRITE on entry and keeps
* holding it until this call returns -- this function does not lock/unlock
* `job` itself, matching apply_job_update_transition's own calling
* convention (its other caller, mark_job_update above, is called under the
* exact same already-held write lock).
*
* apply_job_update_transition()'s existing job_state_is_backward guard is
* the ONLY commit/idempotency guard exercised here -- reused unchanged,
* never duplicated: a terminal job's transition is rejected
* (JOB_TRANSITION_REJECTED_BACKWARD, Req 4.4) and an already-PEND job's
* transition is a no-op (JOB_TRANSITION_NOOP_DUPLICATE, Req 4.3) -- neither
* outcome touches job->job.allocs or appends a WAL fact. Only on
* JOB_TRANSITION_APPLIED does this release job->job.allocs (if non-NULL)
* and durably append the fact via the existing generic persist_submit_job_
* fact() -- the plain {jobid, state} shape is sufficient here, same as
* every other WAL_JOB_UPDATE caller (no extra fields, unlike DISPATCH/
* START/FINISH's richer siblings).
*
* Returns the job_transition_result_t of the underlying transition, cast to
* int, so the caller can tell APPLIED apart from REJECTED_BACKWARD/
* NOOP_DUPLICATE. Unlike record_job_finish_fact/record_job_fail_fact, this
* function does NOT call persist_notify_job_updated_already_persisted --
* the caller must do so itself, after releasing the write lock, exactly
* once, for every JOB_TRANSITION_APPLIED result (cluster.h's doc comment on
* the prototype carries this same contract). Omitting it reproduces
* bug-1679 at this call site. */
int record_job_reconcile_pending_fact(cluster_t* c, job_t* job)
{
job_transition_result_t result;
result = apply_job_update_transition(job, JOB_STAT_PEND);
if (result != JOB_TRANSITION_APPLIED)
return (int)result;
if (job->job.allocs != NULL) {
free_JobAlloc_ptr(&(job->job.allocs));
job->job.allocs = NULL;
}
(void)persist_submit_job_fact(&c->persist, WAL_JOB_UPDATE, job->job.jobid,
JOB_STAT_PEND);
return (int)result;
}
/* Guarded state-transition core for a postdone fact, extracted from
* on_job_postdone's old inline mutation (raft-apply-decoupling task 6.1
* round 2, Req 3.3/7.1-7.3). Unlike a generic UPDATE, postdone's formula

1
src/controller/meson.build
View file

@ -191,6 +191,7 @@ controller_unit_test_src = [
'testsuite/test_dispatch_fail_terminal.c',
'testsuite/test_dispatch_fail_retry.c',
'testsuite/test_jobdispatch_no_double_notify.c',
'testsuite/test_worker_resync.c',
'testsuite/test_upgrade_snapshot_restore.c',
'testsuite/test_upgrade_snapshot_apply.c',
'testsuite/test_bug857_regression.c',

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

@ -109,6 +109,7 @@ extern void test_dispatch_commit(void);
extern void test_dispatch_fail_terminal(void);
extern void test_dispatch_fail_retry(void);
extern void test_jobdispatch_no_double_notify(void);
extern void test_worker_resync(void);
void setUp(void) { return; }
@ -287,6 +288,7 @@ int main(int argc, char* argv[])
TEST(test_dispatch_fail_terminal);
TEST(test_dispatch_fail_retry);
TEST(test_jobdispatch_no_double_notify);
TEST(test_worker_resync);
return UNITY_END();
}

335
src/controller/testsuite/test_worker_resync.c Normal file
View file

@ -0,0 +1,335 @@
/* test_worker_resync.c — worker-rejoin-resync task 2.1
*
* Proves record_job_reconcile_pending_fact() (job.c) — the Requirement 4.6
* new fact-recording function that returns a worker-unrecognized job to
* PEND, releasing any resource allocation it was carrying (bug-1631's own
* deliberately-deferred trigger condition, closed here — see design.md's
* Error Handling section) and durably appending a plain WAL_JOB_UPDATE fact
* -- following dispatch_a_job_()'s eager-mutation-plus-explicit-notify
* pattern (design.md "Controller: job.c — record_job_reconcile_pending_
* fact, record_job_reconcile_assign_fact"), NOT record_job_finish_fact's
* deferred one.
*
* This task implements ONLY record_job_reconcile_pending_fact() itself --
* no live caller exists yet (dispatch_a_job_'s disconnected-worker split is
* task 2.3; resync.c's report-apply glue is a later task), so every test
* here directly exercises the exact sequence a real caller will perform:
* lock -> call record_job_reconcile_pending_fact() -> check the returned
* job_transition_result_t -> unlock -> (only when APPLIED) call
* persist_notify_job_updated_already_persisted() itself -- proving the full
* intended sequence end-to-end even though production wiring of that
* sequence is a later task's job (design.md's Service Interface
* Postconditions/Invariants: the caller, not this function, owns that
* post-unlock notify call).
*
* Fixture mirrors test_dispatch_commit.c's dc_setup/dc_teardown exactly: a
* bare cluster_t (jobmgr + hooksys) plus a real disk-backed persist_t
* initialized with cluster == NULL, so wal_hook_append_classified()
* (persist.c) takes the direct persist_wal_async() path for
* persist_submit_job_fact()'s plain RETAIN-mode WAL_JOB_UPDATE append -- no
* raft/commit-engine machinery needed. p->current_lsn before/after each
* call proves (or disproves) a durable append happened, mirroring
* test_dispatch_commit.c's test_dc_worker_not_found_leaves_job_pend_and_
* alloc_uncommitted's own LSN-delta idiom.
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <pthread.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "hook.h"
#include "jobj.h"
#include "kvmgr.h"
#include "persist.h"
#include "wal_types.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xstring.h"
/* jobj.c's job type table and the process-global xstring table are shared
* across every test file in the controller_unit_test binary -- guard
* idempotently, mirroring every sibling fixture in this suite
* (test_wal_job_dispatch_fact.c's g_wjd_jobobj_ready). */
static int g_twr_jobobj_ready;
static cluster_t g_twr;
static char g_twr_tmpdir[256];
static void twr_setup(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
persist_config_t cfg;
if (makeString((char*)"twr-probe") == NULL)
xstring_table_init(64);
if (!g_twr_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_twr_jobobj_ready = 1;
}
memset(&g_twr, 0, sizeof(g_twr));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_twr.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_twr.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(g_twr.jobmgr);
/* Real disk-backed persist, cluster == NULL: wal_hook_append_
* classified() (persist.c) takes the direct persist_wal_async() path --
* no raft/commit-engine machinery needed for a plain RETAIN-mode fact
* append (mirrors test_dispatch_commit.c's dc_setup exactly, which
* exercises dispatch_a_job_'s own real persist_submit_job_fact()-family
* call the same way). */
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
strcpy(g_twr_tmpdir, "/tmp/xce_twr21_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_twr_tmpdir));
persist_set_workdir(&g_twr.persist, g_twr_tmpdir, "ctrl-test");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_twr.persist, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_twr.persist));
}
static void twr_teardown(void)
{
char cmd[512];
persist_stop(&g_twr.persist);
persist_fini(&g_twr.persist);
if (g_twr.jobmgr)
dereference(g_twr.jobmgr);
xce_hook_fini(&g_twr.hooksys);
snprintf(cmd, sizeof(cmd), "rm -rf %s", g_twr_tmpdir);
if (system(cmd) != 0)
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", g_twr_tmpdir);
}
/* Creates a job directly in the fixture's jobmgr, bypassing dispatch/replay
* entirely -- for seeding fact-transition tests, mirroring
* test_wal_job_dispatch_fact.c's wjd_seed_job / test_dispatch_commit.c's
* dc_seed_ready. */
static job_submit_data_t g_twr_seed_submit;
static job_t* twr_seed_job(uint64_t jobid, uint32_t state)
{
xce_job_t jdef;
memset(&g_twr_seed_submit, 0, sizeof(g_twr_seed_submit));
g_twr_seed_submit.queue = (char*)"q";
g_twr_seed_submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = state;
jdef.submit = &g_twr_seed_submit;
return makeJob(g_twr.jobmgr, &jdef);
}
/* One host ("twr-worker"), one resource ("slots", value 1.0) -- the same
* minimal shape test_wal_job_dispatch_fact.c's/test_dispatch_commit.c's own
* dispatch-alloc tests use. Caller owns the result (release via
* free_JobAlloc_ptr, or hand ownership to job->job.allocs). */
static xce_job_alloc_t* twr_make_test_alloc(uint64_t jobid)
{
return make_job_alloc(jobid, 1,
make_host_allocs(1,
make_host_alloc((char*)"twr-worker", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
}
static uint64_t twr_current_lsn(void)
{
uint64_t lsn;
pthread_mutex_lock(&g_twr.persist.lock);
lsn = g_twr.persist.current_lsn;
pthread_mutex_unlock(&g_twr.persist.lock);
return lsn;
}
/* Spy hook: counts HOOK_AFTER_EVENT/HOOK_DETAIL_JOB_UPDATE fires -- the
* exact hook sched_job_update_fun/apiserver_job_update_fun subscribe to in
* production, mirroring test_dispatch_commit.c's dc_job_update_spy_hook. */
static int g_twr_update_hook_fires;
static int twr_job_update_spy_hook(int type, int situation, int detail,
void* data, va_list ap)
{
(void)type; (void)data; (void)ap;
if (situation == HOOK_AFTER_EVENT && detail == HOOK_DETAIL_JOB_UPDATE)
g_twr_update_hook_fires++;
return 0;
}
#define TWR_JOBID_APPLIED 81001ULL
#define TWR_JOBID_NOOP 81002ULL
#define TWR_JOBID_TERMINAL 81003ULL
/* THE core proof (Req 4.6, 7.1-7.3; bug-1631 regression): a job that was
* dispatched (real allocs committed, state RUN) and reconciles to
* unrecognized ends the call -- synchronously, BEFORE any commit round-trip,
* matching dispatch_a_job_'s own eager pattern, NOT record_job_finish_
* fact's deferred one -- with PEND state and no allocation, a durably
* appended fact, and (once the caller unlocks and notifies, exactly as a
* real caller must) the scheduler's notify path fired exactly once. */
static void test_reconcile_pending_applies_releases_allocs_and_notifies(void)
{
job_t* j;
xce_job_alloc_t* alloc;
int rc;
uint64_t lsn_before, lsn_after;
twr_setup();
g_twr_update_hook_fires = 0;
TEST_ASSERT_EQUAL_INT(0,
xce_add_hook(&g_twr.hooksys, XCE_CONTROLLER_HOOK_JOB,
twr_job_update_spy_hook, NULL));
j = twr_seed_job(TWR_JOBID_APPLIED, JOB_STAT_RUN);
TEST_ASSERT_NOT_NULL(j);
alloc = twr_make_test_alloc(TWR_JOBID_APPLIED);
TEST_ASSERT_NOT_NULL(alloc);
j->job.allocs = alloc; /* simulate dispatch_a_job_'s own committed alloc,
* ownership transferred to job->job.allocs. */
lsn_before = twr_current_lsn();
lock(j, RTTI_LOCK_WRITE);
rc = record_job_reconcile_pending_fact(&g_twr, j);
/* Observable completion (design.md): PEND + no allocation, asserted
* IMMEDIATELY after the call returns, still under the write lock,
* before any commit round-trip -- the mutation is eager, not
* deferred. */
TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_APPLIED, rc,
"an unrecognized RUN job must apply the PEND transition");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state,
"job must be PEND synchronously, before any WAL commit/apply "
"round trip");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"bug-1631: a job reconciled from RUN back to PEND must release its "
"resource allocation immediately, so a later redispatch never "
"inherits a stale allocation from the prior dispatch");
unlock(j);
lsn_after = twr_current_lsn();
TEST_ASSERT_TRUE_MESSAGE(lsn_after > lsn_before,
"record_job_reconcile_pending_fact must durably append a "
"WAL_JOB_UPDATE fact for an applied transition");
/* Caller's responsibility, AFTER unlocking (design.md Postconditions/
* Invariants): notify the scheduler through the existing notify path,
* exactly once, for this actual mutation (Req 7.1-7.3). */
persist_notify_job_updated_already_persisted(&g_twr, j);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_twr_update_hook_fires,
"the scheduler notify path must fire exactly once for this "
"reconciled pending transition (Req 7.2) when the caller notifies "
"after unlocking -- omitting it would reproduce bug-1679 at this "
"new call site");
xce_disable_hook(&g_twr.hooksys, XCE_CONTROLLER_HOOK_JOB,
twr_job_update_spy_hook);
twr_teardown();
printf("test_reconcile_pending_applies_releases_allocs_and_notifies: "
"PASS\n");
}
/* A job with no allocation at all (e.g. never dispatched, already PEND):
* reconciling it again must be a no-op per apply_job_update_transition's
* own duplicate-state guard (JOB_TRANSITION_NOOP_DUPLICATE) -- no mutation,
* no allocs touch (there is none to release), no durable fact, no crash
* (Req 4.3). */
static void test_reconcile_pending_noop_on_already_pending_job(void)
{
job_t* j;
int rc;
uint64_t lsn_before, lsn_after;
twr_setup();
j = twr_seed_job(TWR_JOBID_NOOP, JOB_STAT_PEND);
TEST_ASSERT_NOT_NULL(j);
TEST_ASSERT_NULL(j->job.allocs);
lsn_before = twr_current_lsn();
lock(j, RTTI_LOCK_WRITE);
rc = record_job_reconcile_pending_fact(&g_twr, j);
TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_NOOP_DUPLICATE, rc,
"reconciling an already-PEND job must be treated as a duplicate "
"no-op, matching apply_job_update_transition's own guard");
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state);
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"a job that never held an allocation must still have none after a "
"no-op reconcile");
unlock(j);
lsn_after = twr_current_lsn();
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
"a no-op reconcile must append no duplicate WAL fact (Req 4.3)");
twr_teardown();
printf("test_reconcile_pending_noop_on_already_pending_job: PASS\n");
}
/* A job the controller already knows is terminal (DONE) must never be
* clobbered back to PEND by a reconciliation report (Req 4.4) -- the
* existing job_state_is_backward guard inside apply_job_update_transition,
* reused unmodified, rejects it. */
static void test_reconcile_pending_rejects_terminal_job(void)
{
job_t* j;
int rc;
uint64_t lsn_before, lsn_after;
twr_setup();
j = twr_seed_job(TWR_JOBID_TERMINAL, JOB_STAT_DONE);
TEST_ASSERT_NOT_NULL(j);
lsn_before = twr_current_lsn();
lock(j, RTTI_LOCK_WRITE);
rc = record_job_reconcile_pending_fact(&g_twr, j);
TEST_ASSERT_EQUAL_INT_MESSAGE(JOB_TRANSITION_REJECTED_BACKWARD, rc,
"a terminal job must reject the backward PEND transition, never "
"clobbering more-final state (Req 4.4)");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state,
"the job's terminal state must be preserved exactly");
unlock(j);
lsn_after = twr_current_lsn();
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
"a rejected reconcile must append no WAL fact");
twr_teardown();
printf("test_reconcile_pending_rejects_terminal_job: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_worker_resync(void)
{
UnitySetTestFile(__FILE__);
RUN_TEST(test_reconcile_pending_applies_releases_allocs_and_notifies);
RUN_TEST(test_reconcile_pending_noop_on_already_pending_job);
RUN_TEST(test_reconcile_pending_rejects_terminal_job);
}