merge: sync Rongsong's scheduler-framework (122 commits, 070c7869..f01ae353)

Introduces WAL replay field-completeness fixes (bug-1687..1697, 1731), scheduler snapshot job-allocation restore, heap UAF fixes (vtask_enq, xcealloc, builtin_fcfs, limit), HA fixes (bug-1694/1695/1696), xceadmin retry, worker signal-exit severity, and their test suites.

Conflicts resolved keeping both sides (drop-audit passed):
- worker.c: mark_worker_jobs_exit (bug-1509 B1) + fetchfile_signal_worker_disconnect (bug-1693)
- handler.go: DefaultBroker.Publish (EventJobFinished) + bug-1731 logging
- config_repl.c / test_config_repl.c: our ns-resolution comment / unique-tmpdir
- test_dispatch_commit.c: both B3-UNKWN and bug-1697 start_time assertions

Excluded Rongsong's bookkeeping per policy: .wolf/, .kiro/, CHANGES.md, reports/;
.claude/ + CLAUDE.md kept ours (his settings.json has machine-absolute paths).
This commit is contained in:
lei.chen 2026-08-31 14:24:17 +00:00
commit 3b6f53bb65
168 changed files with 20744 additions and 568495 deletions

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)
@ -502,11 +618,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,
@ -546,6 +666,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);
@ -567,8 +692,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);