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

fix(controller): job.start_time never set -- dispatch/WAL_JOB_START RUN-state collision (bug-1697)

Browse source 
dispatch_a_job_() synchronously sets state=JOB_STAT_RUN at dispatch time
(scheduler-dispatch-commit task 3.1's deliberate no-worker-ack-wait
semantic). By the time the worker's later WAL_JOB_START fact commits,
apply_job_update_transition() sees state already == RUN and returns
NOOP_DUPLICATE instead of APPLIED -- so bug-1687's start_time/pidinfo
decode/adopt block, gated on result==APPLIED, could never run for any
job. start_time (and consequently run_time) stayed 0 forever. Same bug
class already fixed for WAL_JOB_RUSAGE/CKPT/PREASON in this identical
apply function; WAL_JOB_START was missed because bug-1687 predated the
dispatch-time RUN semantic change.

Moves start_time to dispatch-owned: dispatch_a_job_() now stamps
job.start_time alongside state=RUN, under the same write lock.
WAL_JOB_DISPATCH gains a flag-gated (WAL_JOB_DISPATCH_HAS_START_TIME_FLAG)
start_time field so it survives replay/restart; both apply seams
(apply_committed_entry, apply_wal_entry) and restore_job_dispatch_state
decode it when present, falling through to the old alloc-only shape
otherwise. WAL_JOB_START's own now-permanently-dead start_time write is
removed; pidinfo adoption made unconditional to match.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg
This commit is contained in:
Rongsong Shen 2026-08-27 08:25:40 +00:00
parent faedc6fc49
commit ac1e312e2f
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
9 changed files with 859 additions and 184 deletions
Download patch file Download diff file Expand all files Collapse all files

149
src/controller/testsuite/test_wal_job_dispatch_fact.c
View file

@ -162,13 +162,21 @@ static void test_wjd_apply_committed_entry_transitions_pend_to_run(void)
printf("test_wjd_apply_committed_entry_transitions_pend_to_run: PASS\n");
}
/* bug-1678: build a wal_jobstate_t header + trailing Cap'n Proto JobAlloc
* bytes -- persist_submit_job_dispatch_fact()'s payload shape -- mirroring
/* bug-1678 (extended bug-1697): build a wal_jobstate_t header + optional
* fixed uint64_t start_time + trailing Cap'n Proto JobAlloc bytes --
* persist_submit_job_dispatch_fact()'s payload shape -- mirroring
* capnp_encode_joballoc (persist.c, file-static) since job.c/persist.c's own
* static helpers aren't linkable from a test TU. Caller owns *out. */
static int wjd_build_dispatch_payload(uint64_t jobid, uint32_t state,
xce_job_alloc_t* alloc,
uint8_t** out, size_t* out_len)
* static helpers aren't linkable from a test TU. Caller owns *out.
* `has_start_time` sets WAL_JOB_DISPATCH_HAS_START_TIME_FLAG in _pad and
* inserts `start_time` between the header and the JobAlloc bytes (the
* post-bug-1697 shape); when false, `start_time` is ignored and the payload
* is exactly the pre-bug-1697 shape (header immediately followed by
* JobAlloc bytes, _pad=0). */
static int wjd_build_dispatch_payload_ex(uint64_t jobid, uint32_t state,
bool has_start_time,
uint64_t start_time,
xce_job_alloc_t* alloc,
uint8_t** out, size_t* out_len)
{
struct capn c;
struct capn_segment* cs;
@ -176,7 +184,12 @@ static int wjd_build_dispatch_payload(uint64_t jobid, uint32_t state,
JobAlloc_ptr jap;
uint8_t* allocbuf;
size_t alloclen;
wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 };
size_t start_field = has_start_time ? sizeof(start_time)
: 0;
wal_jobstate_t js = { .jobid = jobid, .state = state,
._pad = has_start_time
? WAL_JOB_DISPATCH_HAS_START_TIME_FLAG
: 0 };
capn_init_malloc(&c);
cs = capn_root(&c).seg;
@ -191,15 +204,27 @@ static int wjd_build_dispatch_payload(uint64_t jobid, uint32_t state,
capn_write_mem(&c, allocbuf, alloclen, 0);
capn_free(&c);
*out_len = sizeof(js) + alloclen;
*out_len = sizeof(js) + start_field + alloclen;
*out = malloc(*out_len);
if (!*out) { free(allocbuf); return -ENOMEM; }
memcpy(*out, &js, sizeof(js));
memcpy(*out + sizeof(js), allocbuf, alloclen);
if (has_start_time)
memcpy(*out + sizeof(js), &start_time, sizeof(start_time));
memcpy(*out + sizeof(js) + start_field, allocbuf, alloclen);
free(allocbuf);
return 0;
}
/* Pre-bug-1697 shape wrapper -- every call site that predates the start_time
* field keeps working unchanged. */
static int wjd_build_dispatch_payload(uint64_t jobid, uint32_t state,
xce_job_alloc_t* alloc,
uint8_t** out, size_t* out_len)
{
return wjd_build_dispatch_payload_ex(jobid, state, false, 0, alloc, out,
out_len);
}
/* One host ("wjd-worker"), one resource ("slots", value 1.0) -- the same
* minimal shape test_dispatch_commit.c's own dispatch tests use. Caller owns
* the result (release via free_JobAlloc_ptr). */
@ -242,12 +267,60 @@ static void test_wjd_apply_committed_entry_transitions_and_sets_allocs(void)
TEST_ASSERT_EQUAL_STRING("wjd-worker", j->job.allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_STRING("slots",
j->job.allocs->allocs[0]->allocs[0]->resname);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, (uint64_t)j->job.start_time,
"bug-1697 backward compat: a pre-bug-1697 WAL_JOB_DISPATCH payload "
"(no WAL_JOB_DISPATCH_HAS_START_TIME_FLAG, header+allocs only) must "
"leave job.start_time untouched -- no crash, no misread of absent "
"start_time bytes");
wjd_teardown();
printf("test_wjd_apply_committed_entry_transitions_and_sets_allocs: "
"PASS\n");
}
/* bug-1697: WAL_JOB_DISPATCH_HAS_START_TIME_FLAG set -- the post-fix shape
* carrying a fixed uint64_t start_time between the header and the trailing
* JobAlloc bytes. Proves apply_committed_entry() decodes/adopts it. */
static void test_wjd_apply_committed_entry_sets_start_time_and_allocs(void)
{
wjd_setup();
job_t* j = wjd_seed_job(71007, JOB_STAT_PEND);
TEST_ASSERT_NOT_NULL(j);
xce_job_alloc_t* alloc = wjd_make_test_alloc(71007);
TEST_ASSERT_NOT_NULL(alloc);
uint64_t wire_start_time = 1700000123ULL;
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0,
wjd_build_dispatch_payload_ex(71007, JOB_STAT_RUN, true,
wire_start_time, alloc, &payload,
&plen));
free_JobAlloc_ptr(&alloc);
int rc = apply_committed_entry(&g_wjd_p, WAL_JOB_DISPATCH, 104, payload,
(uint32_t)plen);
free(payload);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(wire_start_time,
(uint64_t)j->job.start_time,
"bug-1697: a WAL_JOB_DISPATCH payload with "
"WAL_JOB_DISPATCH_HAS_START_TIME_FLAG set must adopt the "
"wire-carried start_time, not leave it at 0");
TEST_ASSERT_NOT_NULL_MESSAGE(j->job.allocs,
"job.allocs must still be recovered alongside start_time");
TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts);
TEST_ASSERT_EQUAL_STRING("wjd-worker", j->job.allocs->allocs[0]->host);
wjd_teardown();
printf("test_wjd_apply_committed_entry_sets_start_time_and_allocs: "
"PASS\n");
}
/* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */
static void
@ -309,11 +382,54 @@ static void test_wjd_apply_wal_entry_transitions_and_sets_allocs(void)
"primary's own boot replay");
TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts);
TEST_ASSERT_EQUAL_STRING("wjd-worker", j->job.allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, (uint64_t)j->job.start_time,
"bug-1697 backward compat: a pre-bug-1697 WAL_JOB_DISPATCH payload "
"must leave job.start_time untouched on this seam too");
wjd_teardown();
printf("test_wjd_apply_wal_entry_transitions_and_sets_allocs: PASS\n");
}
/* bug-1697: sibling of test_wjd_apply_committed_entry_sets_start_time_and_
* allocs, for apply_wal_entry's standby catch-up / replicated-apply seam. */
static void test_wjd_apply_wal_entry_sets_start_time_and_allocs(void)
{
wjd_setup();
job_t* j = wjd_seed_job(71008, JOB_STAT_PEND);
TEST_ASSERT_NOT_NULL(j);
xce_job_alloc_t* alloc = wjd_make_test_alloc(71008);
TEST_ASSERT_NOT_NULL(alloc);
uint64_t wire_start_time = 1700000456ULL;
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0,
wjd_build_dispatch_payload_ex(71008, JOB_STAT_RUN, true,
wire_start_time, alloc, &payload,
&plen));
free_JobAlloc_ptr(&alloc);
int rc = apply_wal_entry(&g_wjd_p, WAL_JOB_DISPATCH, 105, payload,
(uint32_t)plen);
free(payload);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(wire_start_time,
(uint64_t)j->job.start_time,
"bug-1697: apply_wal_entry's catch-up dispatch must also recover "
"the wire-carried start_time when "
"WAL_JOB_DISPATCH_HAS_START_TIME_FLAG is set, required for a "
"promoted standby");
TEST_ASSERT_NOT_NULL(j->job.allocs);
TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts);
wjd_teardown();
printf("test_wjd_apply_wal_entry_sets_start_time_and_allocs: PASS\n");
}
/* ── Restart-replay Observable (this task's completion criterion) ───────── */
static persist_t* wjd_make_fresh_persist(const char* tmpdir)
@ -498,11 +614,15 @@ test_wjd_restart_replay_recovers_running_and_allocs_with_dispatch_fact(void)
xce_job_alloc_t* alloc = wjd_make_test_alloc(71006);
TEST_ASSERT_NOT_NULL(alloc);
/* bug-1697: new-format payload (WAL_JOB_DISPATCH_HAS_START_TIME_
* FLAG set), so this restart-replay test also proves start_time
* -- not just allocs -- survives a real restart. */
uint8_t* dispatch_payload = NULL;
size_t dispatch_plen = 0;
TEST_ASSERT_EQUAL_INT(0,
wjd_build_dispatch_payload(71006, JOB_STAT_RUN, alloc,
&dispatch_payload, &dispatch_plen));
wjd_build_dispatch_payload_ex(71006, JOB_STAT_RUN, true,
71006000ULL, alloc,
&dispatch_payload, &dispatch_plen));
free_JobAlloc_ptr(&alloc);
TEST_ASSERT_EQUAL_INT(0,
@ -537,6 +657,11 @@ test_wjd_restart_replay_recovers_running_and_allocs_with_dispatch_fact(void)
TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts);
TEST_ASSERT_EQUAL_STRING("wjd-worker",
j->job.allocs->allocs[0]->host);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(71006000ULL,
(uint64_t)j->job.start_time,
"bug-1697: restart replay must also recover the dispatch-time "
"start_time from the WAL_JOB_DISPATCH fact's wire-carried "
"field, not leave it at 0");
wjd_teardown();
wjd_teardown_persist(p2);
@ -558,8 +683,10 @@ void test_wal_job_dispatch_fact(void)
UnitySetTestFile(__FILE__);
RUN_TEST(test_wjd_apply_committed_entry_transitions_pend_to_run);
RUN_TEST(test_wjd_apply_committed_entry_transitions_and_sets_allocs);
RUN_TEST(test_wjd_apply_committed_entry_sets_start_time_and_allocs);
RUN_TEST(test_wjd_apply_wal_entry_transitions_pend_to_run_no_hook);
RUN_TEST(test_wjd_apply_wal_entry_transitions_and_sets_allocs);
RUN_TEST(test_wjd_apply_wal_entry_sets_start_time_and_allocs);
RUN_TEST(test_wjd_restart_replay_recovers_running_without_start_fact);
RUN_TEST(
test_wjd_restart_replay_recovers_running_and_allocs_with_dispatch_fact);