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

feat(worker-rejoin-resync): reconcile unqueried job assignment (task 2.2)

Browse source 
Add record_job_reconcile_assign_fact(cluster_t*, job_t*, worker_t*,
xce_job_state_entry_t*) to job.c: eagerly sets job->job.state to match
the worker's reported outcome (mirroring dispatch_a_job_'s inline
mutation, minus allocation fields -- there are none to set, job->job.allocs
stays NULL per Requirement 4.8's documented scope). Calls the existing
persist_submit_job_dispatch_fact(..., alloc=NULL) directly, then leaves
the post-unlock persist_notify_job_updated_already_persisted() call to
the caller, same convention as task 2.1.

Fixes bug-1746 along the way: capnp_encode_joballoc() (persist.c, part
of bug-1678's original fix, predates this feature) unconditionally
dereferenced a NULL alloc -- never triggered before since
dispatch_a_job_ (the only prior caller) always passes non-NULL. Fixed
with a minimal guard producing a true zero-length buffer, matching
apply_committed_entry's alloc_len > 0 decode gate exactly. This
directly contradicted design.md's "tested, backward-compatible on both
seams" claim (true only for decode, never actually exercised for
encode) -- corrected in 4 places.

2 tests: applies-running-no-allocs-and-notifies, and a real restart-
replay round trip (write to real disk, fresh persist_recover(), confirm
jobid/host/state/NULL-allocs survive) -- proving this task's own
"durably survives a restart replay" completion criterion literally, not
just the in-memory mutation.

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 11:40:47 +00:00
parent b7ed6de2a1
commit ecfb4c942f
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
7 changed files with 550 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -13,6 +13,7 @@
#include "myconf.h"
#include "mylock.h"
#include "netio.h"
#include "typemap.h" /* xce_job_state_entry_t -- worker-rejoin-resync task 2.2 */
#include "vtaskmgr.h"
#include "worker.h"
#include "xce/cluster_def.h"
@ -211,6 +212,35 @@ extern job_transition_result_t apply_job_postdone_transition(job_t*);
* call site. */
extern int record_job_reconcile_pending_fact(cluster_t*, job_t*);
/* worker-rejoin-resync task 2.2 (Req 4.1, 4.3, 4.4, 4.7, 4.8, 7.1-7.3):
* records a job the worker reports running that the controller has no
* existing non-terminal record of ("unqueried extra", Req 3.6) as assigned
* to the reporting worker `w`, matching `report`'s reported status --
* 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, and job.c's own doc comment on this function
* for the full rationale (including why job->job.allocs is deliberately
* left NULL, and why there is no separate "host assignment" field for this
* function to set beyond job->job.state).
*
* Precondition: caller holds `job` under RTTI_LOCK_WRITE on entry and does
* not release it until this call returns -- this function does not
* lock/unlock `job` itself. Unlike record_job_reconcile_pending_fact above,
* this function performs no backward-transition guard of its own: the
* caller must only invoke it when the controller has no existing
* non-terminal record of this job (Req 4.8's own trigger condition), so
* there is nothing more-final to clobber.
*
* Always returns (int)JOB_TRANSITION_APPLIED -- this outcome has no
* reject/no-op case (unlike record_job_reconcile_pending_fact's
* REJECTED_BACKWARD/NOOP_DUPLICATE). Matching that sibling's contract, the
* caller (not this function) must call
* persist_notify_job_updated_already_persisted(c, job) itself, AFTER
* releasing the write lock, exactly once -- omitting it reproduces bug-1679
* at this call site. */
extern int record_job_reconcile_assign_fact(cluster_t*, job_t*, worker_t*,
xce_job_state_entry_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*);

119
src/controller/job.c
View file

@ -13,6 +13,7 @@
#include "persist.h"
#include "preason.h"
#include "request.h"
#include "typemap.h" /* xce_job_state_entry_t -- worker-rejoin-resync task 2.2 */
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xce/types.h"
@ -2931,6 +2932,124 @@ int record_job_reconcile_pending_fact(cluster_t* c, job_t* job)
return (int)result;
}
/* worker-rejoin-resync task 2.2 (Req 4.1, 4.3, 4.4, 4.7, 4.8, 7.1-7.3):
* records a job the worker reports as running that the controller has no
* existing non-terminal record of -- an "unqueried extra" (Req 3.6) the
* controller never had a chance to ask about -- as assigned to the
* reporting worker, through the same idempotent commit-then-apply path an
* ordinary dispatch uses.
*
* Follows dispatch_a_job_()'s own established pattern (job.c, its `if (rc
* >= 0) { ... }` success block, job.c:1686-1748) exactly as record_job_
* reconcile_pending_fact above does: 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.
* The fact recorded here is something the worker already confirmed really
* happened (it is running there right now) -- ground truth already changed
* outside this controller -- 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 record_job_reconcile_pending_fact's own calling
* convention above. Unlike that sibling, this function performs no
* backward-transition guard check of its own: design.md's Service Interface
* Invariants note that both new functions are "never called for a job the
* controller already knows is terminal or more-final" as a PRECONDITION the
* CALLER enforces, not something either function re-checks with its own
* guard -- and for this specific outcome the precondition holds by
* construction: Req 4.8's own trigger condition is precisely "the
* Controller has no existing non-terminal record of that job assigned
* elsewhere", so there is no terminal/more-final in-memory state this call
* could possibly clobber. This mirrors dispatch_a_job_'s own inline
* success-path mutation (job.c:1686-1717), which likewise sets
* job->job.state directly, with no apply_job_update_transition/
* job_state_is_backward call.
*
* "Host assignment": dispatch_a_job_'s own inline mutation (job.c:1686-
* 1717) carries "which host" ENTIRELY via job->job.allocs -- the host name
* lives inside xce_host_alloc_t.host, nested under xce_job_alloc_t.allocs.
* Neither job_t nor xce_job_t (xce/types.h) has any OTHER field
* representing a job's assigned worker/host independent of allocs
* (confirmed by direct inspection of both struct definitions -- not a
* guess). Since this function deliberately leaves job->job.allocs NULL (the
* controller genuinely does not know what the scheduler originally computed
* for this job -- design.md's Data Models "Documented scope decision"),
* there is structurally no separate field left for this function to set to
* represent "host assignment" beyond job->job.state itself. `w` is accepted
* only to match design.md's own cited service-interface signature -- it is
* otherwise inert here (see the (void)w; below): the still-running-outcome
* check just below operates on `report->status`, not on `w`, and nothing on
* job_t/xce_job_t is mutated from `w` either.
*
* Returns job_transition_result_t (cast to int), matching record_job_
* reconcile_pending_fact's return-type convention above -- always
* JOB_TRANSITION_APPLIED (this outcome has no reject/no-op case; see the
* Precondition note). 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 -- omitting it reproduces bug-1679 at this call site. */
int record_job_reconcile_assign_fact(cluster_t* c, job_t* job, worker_t* w,
xce_job_state_entry_t* report)
{
(void)w;
switch (report->status) {
case JobStateStatus_running:
break;
default:
/* This function's only defined caller usage (design.md's sequence
* flow -- task 3.2, not yet implemented) is the "unqueried extra
* job, still running" outcome: a finished extra routes through the
* existing record_job_finish_fact/record_job_fail_fact instead
* (Req 4.7), same as a queried finished job. Fail loud rather than
* silently mis-recording a finished job as RUN, mirroring
* dispatch_a_job_'s own "impossible per today's call convention"
* ERROR+assert(0)-in-XCE_TESTING-then-safe-fallback idiom used
* throughout this file. */
ERROR("record_job_reconcile_assign_fact: called for job %lu with "
"unexpected report status %d (expected JobStateStatus_running "
"-- a finished/unrecognized outcome must route through a "
"different fact function)",
(unsigned long)job->job.jobid, report->status);
#ifdef XCE_TESTING
assert(0
&& "record_job_reconcile_assign_fact called with a "
"non-running report status");
#endif
break;
}
job->job.state = JOB_STAT_RUN;
/* bug-1697: start_time is stamped HERE, synchronously alongside the RUN
* transition, mirroring dispatch_a_job_'s own identical stamp
* (job.c:1710) -- so that if this job's worker later also sends an
* ordinary WAL_JOB_START report for it, that report always finds
* state/start_time already set and applies as a harmless no-op, never
* the genuine first RUN transition. There is no better time source: the
* controller never observed this job's real dispatch time. */
job->job.start_time = time(NULL);
/* Req 4.8 / design.md "Documented scope decision": job->job.allocs is
* deliberately left untouched (NULL for a freshly-constructed job_t,
* per this file's own makeJob()/create-job convention) -- the
* controller has no way to learn what resource allocation the
* scheduler originally computed for a job it never held any record of.
* bug-1678's persist_submit_job_dispatch_fact() NULL-alloc case is
* reused verbatim below, exactly as design.md specifies, rather than
* fabricating one. */
(void)persist_submit_job_dispatch_fact(&c->persist, job->job.jobid,
JOB_STAT_RUN,
(uint64_t)job->job.start_time,
NULL);
return (int)JOB_TRANSITION_APPLIED;
}
/* 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

20
src/controller/persist.c
View file

@ -230,6 +230,26 @@ static int capnp_encode_joballoc(const xce_job_alloc_t* a, uint8_t** out,
struct JobAlloc cja;
JobAlloc_ptr jap;
/* worker-rejoin-resync task 2.2 (bug-1746): a NULL `a` means "no
* allocation to carry" (record_job_reconcile_assign_fact()'s Req 4.8
* case -- the controller genuinely never learned one). Produce a
* genuinely zero-length buffer, NOT a capnp-encoded null-pointer JobAlloc
* (what encode_JobAlloc_ptr()'s existing NULL branch would produce) --
* persist_submit_job_dispatch_fact()'s caller (both apply seams,
* apply_committed_entry()/apply_wal_entry()) discriminates "is there an
* allocation to decode" purely by `alloc_len > 0`, so any non-zero
* trailing byte count here would be misread as a real allocation to
* decode. encode_JobAlloc() below has no NULL guard of its own (it
* unconditionally dereferences `a` on its first line) -- this early
* return is the one and only guard needed, matching the sole documented
* call convention this function must now support (dispatch_a_job_()'s
* own call site continues to always pass non-NULL, unaffected). */
if (a == NULL) {
*out = NULL;
*out_len = 0;
return 0;
}
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_JobAlloc(cs, &cja, (xce_job_alloc_t*)a);

349
src/controller/testsuite/test_worker_resync.c
View file

@ -324,6 +324,353 @@ static void test_reconcile_pending_rejects_terminal_job(void)
printf("test_reconcile_pending_rejects_terminal_job: PASS\n");
}
/* ── record_job_reconcile_assign_fact() tests (task 2.2) ─────────────────
*
* Proves record_job_reconcile_assign_fact() (job.c) -- the Requirement 4.8
* new fact-recording function that records a job the worker reports running
* that the controller has no existing non-terminal record of ("unqueried
* extra", Req 3.6) as assigned to the reporting worker, through the same
* dispatch_a_job_()-style eager-mutation-plus-explicit-notify pattern as
* record_job_reconcile_pending_fact above, reusing bug-1678's NULL-alloc
* persist_submit_job_dispatch_fact() path (design.md "Controller: job.c --
* record_job_reconcile_pending_fact, record_job_reconcile_assign_fact").
*
* As with task 2.1, no live caller exists yet (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_assign_fact() -> check the result -> unlock -> call
* persist_notify_job_updated_already_persisted() itself. */
#define TWR_JOBID_ASSIGN 81004ULL
#define TWR_JOBID_ASSIGN_REPLAY 81005ULL
/* THE core proof (Req 4.1, 4.3, 4.4, 4.7, 4.8, 7.1-7.3): a job the
* controller never previously tracked, reported RUNNING by a worker, ends
* the call -- synchronously, before any commit round-trip -- recorded as
* running with no resource allocation (the controller genuinely does not
* know what the scheduler originally computed, design.md's "Documented
* scope decision"), 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_assign_applies_running_no_allocs_and_notifies(void)
{
job_t* j;
worker_t w;
xce_job_state_entry_t report;
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));
/* Seeded with no allocation, mirroring a job the caller (task 3.2, not
* yet implemented) constructed fresh for a jobid the controller had
* never seen before -- there is nothing to release here, only
* something new to record. */
j = twr_seed_job(TWR_JOBID_ASSIGN, JOB_STAT_PEND);
TEST_ASSERT_NOT_NULL(j);
TEST_ASSERT_NULL(j->job.allocs);
memset(&w, 0, sizeof(w));
w.host.name = (char*)"twr-worker";
report = (xce_job_state_entry_t){
.jobid = TWR_JOBID_ASSIGN,
.status = JobStateStatus_running,
.exit_code = 0,
.finish_time_ms = 0,
.queried = 0
};
lsn_before = twr_current_lsn();
lock(j, RTTI_LOCK_WRITE);
rc = record_job_reconcile_assign_fact(&g_twr, j, &w, &report);
/* Observable completion (design.md / this task): RUN + 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 unqueried running job must always apply (no reject/no-op case)");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
"job must be RUN synchronously, before any WAL commit/apply round "
"trip");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"Req 4.8: a reconciliation-assigned job must never carry a "
"fabricated resource allocation -- the controller genuinely does "
"not know what the scheduler originally computed for it");
unlock(j);
lsn_after = twr_current_lsn();
TEST_ASSERT_TRUE_MESSAGE(lsn_after > lsn_before,
"record_job_reconcile_assign_fact must durably append a "
"WAL_JOB_DISPATCH fact");
/* 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 assignment (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_assign_applies_running_no_allocs_and_notifies: "
"PASS\n");
}
/* Encode helper mirroring test_dispatch_commit.c's dc_encode_job /
* test_wal_job_dispatch_fact.c's wjd_encode_job (capnp_encode_job is
* file-static in persist.c, so a direct WAL_JOB_SUBMIT seed needs its own
* encoder built from the exported encode_Job/new_Job/write_Job codec
* chain). */
static int twr_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;
}
/* Fresh disk-backed persist_t pointed at an existing tmpdir -- mirrors
* test_dispatch_commit.c's dc_make_fresh_persist/dc_teardown_persist
* exactly, for this test's own Phase 2 (simulated restart). */
static persist_t* twr_make_fresh_persist(const char* tmpdir)
{
persist_t* p = calloc(1, sizeof(persist_t));
persist_config_t cfg;
if (!p) return NULL;
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
persist_set_workdir(p, tmpdir, "ctrl-test");
if (persist_init(p, &cfg, NULL) != 0) {
free(p);
return NULL;
}
if (persist_start(p) != 0) {
persist_fini(p);
free(p);
return NULL;
}
return p;
}
static void twr_teardown_persist(persist_t* p)
{
persist_stop(p);
persist_fini(p);
free(p);
}
/* Restart-replay round trip -- this task's own explicit completion
* criterion ("durably survives a restart replay with the same result"),
* mirroring test_dispatch_commit.c's test_dc_success_appends_wal_job_
* dispatch_durably's Phase 1/Phase 2 shape exactly, but driving Phase 1's
* WAL_JOB_DISPATCH fact via record_job_reconcile_assign_fact() instead of
* dispatch_a_job_() -- proving bug-1678's NULL-alloc decode branch (already
* tested on both apply seams by test_wal_job_dispatch_fact.c) round-trips
* this new call site's real fact correctly too. Uses its own local
* cluster_t/persist_t (not the shared g_twr fixture) so Phase 1's disk
* state survives past that phase's own teardown, exactly like dc's
* original. */
static void test_reconcile_assign_survives_restart_replay(void)
{
char tmp[256];
uint64_t captured_start_time = 0;
strcpy(tmp, "/tmp/xce_twr22_replay_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Phase 1: real cluster + real disk persist. Seed a durable
* WAL_JOB_SUBMIT fact directly (so replay can reconstruct the job --
* this job was never dispatched through any controller, so there is no
* other WAL_JOB_SUBMIT/DISPATCH in this log), seed the in-memory job_t
* the same way twr_seed_job() does (bypassing dispatch/replay --
* record_job_reconcile_assign_fact() operates on a job_t* the caller
* has already located/constructed, not this function's concern), then
* drive the real function under test. */
{
cluster_t c;
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
persist_config_t cfg;
job_t* j;
worker_t w;
xce_job_state_entry_t report;
int rc;
if (makeString((char*)"twr22-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(&c, 0, sizeof(c));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX));
c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(c.jobmgr);
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
persist_set_workdir(&c.persist, tmp, "ctrl-test");
TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist));
{
job_submit_data_t wal_submit;
xce_job_t wal_jdef;
uint8_t* buf = NULL;
size_t sz = 0;
memset(&wal_submit, 0, sizeof(wal_submit));
wal_submit.queue = (char*)"q";
wal_submit.user = (char*)"tester";
memset(&wal_jdef, 0, sizeof(wal_jdef));
wal_jdef.jobid = TWR_JOBID_ASSIGN_REPLAY;
wal_jdef.state = JOB_STAT_PEND;
wal_jdef.submit = &wal_submit;
TEST_ASSERT_EQUAL_INT(0, twr_encode_job(&wal_jdef, &buf, &sz));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(&c.persist, WAL_JOB_SUBMIT, buf,
(uint32_t)sz));
free(buf);
}
{
job_submit_data_t submit;
xce_job_t jdef;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"q";
submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = TWR_JOBID_ASSIGN_REPLAY;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
j = makeJob(c.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(j);
}
memset(&w, 0, sizeof(w));
w.host.name = (char*)"twr-worker-replay";
report = (xce_job_state_entry_t){
.jobid = TWR_JOBID_ASSIGN_REPLAY,
.status = JobStateStatus_running,
.exit_code = 0,
.finish_time_ms = 0,
.queried = 0
};
lock(j, RTTI_LOCK_WRITE);
rc = record_job_reconcile_assign_fact(&c, j, &w, &report);
TEST_ASSERT_EQUAL_INT(JOB_TRANSITION_APPLIED, rc);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
TEST_ASSERT_NULL(j->job.allocs);
captured_start_time = (uint64_t)j->job.start_time;
TEST_ASSERT_NOT_EQUAL(0, captured_start_time);
unlock(j);
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist));
persist_stop(&c.persist);
persist_fini(&c.persist);
dereference(c.jobmgr);
xce_hook_fini(&c.hooksys);
}
/* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only),
* recover from disk -- simulates a controller restart. */
{
persist_t* p2 = twr_make_fresh_persist(tmp);
cluster_t c2;
job_t* j2 = NULL;
int rc;
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
TEST_ASSERT_NOT_NULL(p2);
memset(&c2, 0, sizeof(c2));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&c2.hooksys, XCE_CONTROLLER_HOOK_MAX));
c2.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(c2.jobmgr);
p2->cluster = &c2;
rc = persist_recover(p2);
TEST_ASSERT_EQUAL_INT(0, rc);
lock(c2.jobmgr, RTTI_LOCK_READ);
key_find(c2.jobmgr, (void*)TWR_JOBID_ASSIGN_REPLAY, (void**)&j2);
unlock(c2.jobmgr);
TEST_ASSERT_NOT_NULL_MESSAGE(j2,
"restart replay must reconstruct the job via WAL_JOB_SUBMIT");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(TWR_JOBID_ASSIGN_REPLAY,
j2->job.jobid, "recovered job must be the same jobid");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j2->job.state,
"restart replay must recover the job as RUNNING from the "
"WAL_JOB_DISPATCH fact appended by record_job_reconcile_"
"assign_fact()'s real call (Req 4.8) -- no WAL_JOB_START "
"anywhere in this log");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(captured_start_time,
(uint64_t)j2->job.start_time,
"restart replay must recover the exact dispatch-time "
"start_time stamped by the real call in Phase 1");
TEST_ASSERT_NULL_MESSAGE(j2->job.allocs,
"Req 4.8 / design.md's documented scope decision: replay must "
"NOT fabricate a resource allocation for a reconciliation-"
"assigned job -- allocs stays NULL exactly as it was set "
"eagerly in Phase 1, exercising bug-1678's NULL-alloc decode "
"branch");
xce_hook_fini(&c2.hooksys);
if (c2.jobmgr) dereference(c2.jobmgr);
twr_teardown_persist(p2);
}
{
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp);
if (system(cmd) != 0)
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp);
}
printf("test_reconcile_assign_survives_restart_replay: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_worker_resync(void)
@ -332,4 +679,6 @@ void test_worker_resync(void)
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);
RUN_TEST(test_reconcile_assign_applies_running_no_allocs_and_notifies);
RUN_TEST(test_reconcile_assign_survives_restart_replay);
}