fix(controller): WAL_JOB_START loses pidinfo/start_time on restart replay (bug-1687)
Same class as bug-1678: apply_job_start_transition() sets the real
job->job.pidinfo (worker-reported PIDs/PGIDs/exdata) and start_time, but
the WAL fact for WAL_JOB_START only carried the shared wal_jobstate_t
{jobid, state} header -- both apply seams (apply_committed_entry() for
primary boot replay, apply_wal_entry() for standby catch-up) restored
state only. pidinfo stayed permanently NULL after a restart; start_time
was fabricated with time(NULL) at replay instead of being carried,
drifting from the true value.
Mirrors bug-1678's exact fix shape: capnp_encode_pidinfo()/decode_pidinfo()
(persist.c, using the pre-existing JobPIDInfo capnp struct -- no schema
change needed) serialize job_pidinfo_t into the WAL payload, trailing a
new uint64_t start_time field appended after wal_jobstate_t.
persist_submit_job_start_fact() replaces the old state-only
persist_submit_job_fact() call for this type. Both apply seams updated;
a payload of exactly sizeof(wal_jobstate_t) (any entry written before
this fix) decodes with pidinfo left NULL and start_time fabricated via
time(NULL), exactly matching pre-fix behavior -- full backward
compatibility, verified by every pre-existing WAL_JOB_START test using
the old bare-struct shape.
Also fixed a genuine, unrelated deadlock this change's larger payload
exposed in test_commit_barrier_reply.c: cbrp_count_jobstate_entries_ex()
matched WAL entries by exact payload size (==), so once WAL_JOB_START
legitimately grew it silently undercounted to 0 and a downstream test
spun forever polling an uncounted target. Fixed the match to >= (jobid/
state always live at a fixed leading offset) -- bisected via /proc/*/wchan
futex inspection (no ptrace in this sandbox), confirmed as the actual
cause, not the teardown-ordering red herring first suspected.
Found by an /xce-audit-invariants controller pass (field-complete apply
parity dimension, added after bug-1678). 6 new tests in
test_wal_job_start_fact.c; RED/GREEN independently verified (4 tests
genuinely failed pre-fix with exact expected-vs-fabricated value
mismatches, 2 backward-compat tests correctly passed both before and
after). Full controller suite: 1146 tests, 2 failures -- both the
pre-existing, unrelated bug-1561 overlayfs sandbox flake.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg
This commit is contained in:
parent
bc3f76b42e
commit
955efecd19
9 changed files with 992 additions and 7 deletions
File diff suppressed because one or more lines are too long
|
|
@ -815,6 +815,42 @@ int restore_job_dispatch_state(cluster_t* c, uint64_t jobid, uint32_t state,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* bug-1687: WAL_JOB_START's own sibling of restore_job_dispatch_state()
|
||||
* above (bug-1678), for apply_wal_entry()'s standby catch-up / replicated-
|
||||
* apply seam -- takes ownership of an optional decoded job_pidinfo_t (NULL
|
||||
* for a state-only WAL_JOB_START entry, e.g. one written before this fix)
|
||||
* and a wire-carried start_time (0 meaning "not carried -- leave
|
||||
* job.start_time untouched", matching this seam's own pre-fix behavior of
|
||||
* never stamping it for WAL_JOB_START at all) so a promoted standby
|
||||
* recovers the worker-confirmed pids/pgids/exdata and the true start_time
|
||||
* the same way apply_committed_entry's WAL_JOB_START case already does for
|
||||
* the primary's own boot replay. */
|
||||
int restore_job_start_state(cluster_t* c, uint64_t jobid, uint32_t state,
|
||||
uint64_t start_time, job_pidinfo_t* pidinfo)
|
||||
{
|
||||
job_t* j = NULL;
|
||||
lock(c->jobmgr, RTTI_LOCK_READ);
|
||||
key_find(c->jobmgr, (void*)jobid, (void**)&j);
|
||||
unlock(c->jobmgr);
|
||||
if (!j) {
|
||||
if (pidinfo != NULL)
|
||||
free_JobPIDInfo_ptr(&pidinfo);
|
||||
return -ENOENT;
|
||||
}
|
||||
|
||||
lock(j, RTTI_LOCK_WRITE);
|
||||
j->job.state = (int)state;
|
||||
if (start_time != 0)
|
||||
j->job.start_time = (time_t)start_time;
|
||||
if (pidinfo != NULL) {
|
||||
if (j->job.pidinfo != NULL)
|
||||
free_JobPIDInfo_ptr(&(j->job.pidinfo));
|
||||
j->job.pidinfo = pidinfo;
|
||||
}
|
||||
unlock(j);
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int get_nslots_in_alloc(xce_host_alloc_t* alloc)
|
||||
{
|
||||
int i;
|
||||
|
|
@ -2190,9 +2226,25 @@ static int record_job_start_fact(cluster_t* c, job_t* job, job_pidinfo_t* info)
|
|||
* 5.1, 5.3). JOB_STAT_RUN is the resultant state a START always
|
||||
* produces once past the guard above; apply_committed_entry() will
|
||||
* re-validate the same guard against whatever state is current AT
|
||||
* APPLY TIME before setting it. */
|
||||
(void)persist_submit_job_fact(&c->persist, WAL_JOB_START, job->job.jobid,
|
||||
JOB_STAT_RUN);
|
||||
* APPLY TIME before setting it.
|
||||
*
|
||||
* bug-1687: persist_submit_job_start_fact() carries `info` (the real
|
||||
* worker-reported pids/pgids/exdata) and a receipt-time start_time
|
||||
* alongside {jobid, state=RUN} -- see that function's doc comment and
|
||||
* wal_types.h's WAL_JOB_START payload note for the full rationale
|
||||
* (mirrors bug-1678's WAL_JOB_DISPATCH fix). Before this fix,
|
||||
* persist_submit_job_fact()'s bare {jobid, state} payload silently
|
||||
* dropped `info` entirely, so job.pidinfo replayed permanently NULL and
|
||||
* job.start_time was re-fabricated (time(NULL)) fresh at every replay
|
||||
* instead of carrying the true worker-reported time. Capturing
|
||||
* start_time HERE, once, at receipt time (rather than re-deriving
|
||||
* time(NULL) independently at each of apply_committed_entry's/
|
||||
* apply_wal_entry's OWN apply-time moments) makes it the single
|
||||
* authoritative value both apply seams converge on. */
|
||||
time_t start_time = time(NULL);
|
||||
(void)persist_submit_job_start_fact(&c->persist, job->job.jobid,
|
||||
JOB_STAT_RUN, (uint64_t)start_time,
|
||||
info);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -170,6 +170,7 @@ controller_unit_test_src = [
|
|||
'testsuite/test_apply_committed_entry.c',
|
||||
'testsuite/test_replay_apply_seam.c',
|
||||
'testsuite/test_wal_job_dispatch_fact.c',
|
||||
'testsuite/test_wal_job_start_fact.c',
|
||||
'testsuite/test_job_dispatch_fail_count.c',
|
||||
'testsuite/test_apply_reapply_idempotency.c',
|
||||
'testsuite/test_apply_correctness_verification.c',
|
||||
|
|
|
|||
|
|
@ -42,6 +42,16 @@ extern int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state);
|
|||
* primary's own boot replay. */
|
||||
extern int restore_job_dispatch_state(cluster_t* c, uint64_t jobid,
|
||||
uint32_t state, xce_job_alloc_t* alloc);
|
||||
/* restore_job_start_state defined in job.c (bug-1687): WAL_JOB_START's own
|
||||
* sibling of restore_job_state() above, additionally taking ownership of an
|
||||
* optional decoded job_pidinfo_t and a wire-carried start_time (0 means
|
||||
* "not carried -- leave job.start_time untouched", the pre-fix payload's
|
||||
* behavior) so the worker-confirmed start survives standby catch-up /
|
||||
* replicated-apply the same way apply_committed_entry's WAL_JOB_START case
|
||||
* already restores it on the primary's own boot replay. */
|
||||
extern int restore_job_start_state(cluster_t* c, uint64_t jobid,
|
||||
uint32_t state, uint64_t start_time,
|
||||
job_pidinfo_t* pidinfo);
|
||||
extern int worker_from_capnp(cluster_t* c,
|
||||
const void* capnp_payload, size_t len);
|
||||
extern int restore_queue_state(cluster_t* c,
|
||||
|
|
@ -198,6 +208,62 @@ static xce_job_alloc_t* decode_joballoc(const uint8_t* payload, size_t len)
|
|||
return alloc;
|
||||
}
|
||||
|
||||
/* Serialize job_pidinfo_t to a malloc'd Cap'n Proto buffer (bug-1687):
|
||||
* lets WAL_JOB_START carry the worker-reported pids/pgids/exdata alongside
|
||||
* the {jobid, state} header, so a restart replay can reconstruct
|
||||
* job.pidinfo instead of leaving it permanently NULL. Mirrors
|
||||
* capnp_encode_joballoc above exactly, using the JobPIDInfo struct (already
|
||||
* in controller.capnp as the existing Job.pidinfo field -- no schema change
|
||||
* needed here). */
|
||||
static int capnp_encode_pidinfo(const job_pidinfo_t* info, uint8_t** out,
|
||||
size_t* out_len)
|
||||
{
|
||||
struct capn c;
|
||||
struct capn_segment* cs;
|
||||
struct JobPIDInfo cji;
|
||||
JobPIDInfo_ptr jip;
|
||||
|
||||
capn_init_malloc(&c);
|
||||
cs = capn_root(&c).seg;
|
||||
encode_JobPIDInfo(cs, &cji, (job_pidinfo_t*)info);
|
||||
jip = new_JobPIDInfo(cs);
|
||||
write_JobPIDInfo(&cji, jip);
|
||||
capn_setp(capn_root(&c), 0, jip.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;
|
||||
}
|
||||
|
||||
/* Decode a Cap'n Proto JobPIDInfo buffer (bug-1687) into a freshly heap-
|
||||
* allocated job_pidinfo_t*, mirroring decode_joballoc()'s pattern above.
|
||||
* Returns NULL on a NULL/empty payload; caller owns the result (release via
|
||||
* free_JobPIDInfo_ptr). */
|
||||
static job_pidinfo_t* decode_pidinfo(const uint8_t* payload, size_t len)
|
||||
{
|
||||
struct capn c_ctx;
|
||||
JobPIDInfo_ptr jip;
|
||||
struct JobPIDInfo cji;
|
||||
job_pidinfo_t* info;
|
||||
|
||||
if (!payload || len == 0)
|
||||
return NULL;
|
||||
|
||||
capn_init_mem(&c_ctx, payload, len, 0);
|
||||
jip.p = capn_getp(capn_root(&c_ctx), 0, 1);
|
||||
memset(&cji, 0, sizeof(cji));
|
||||
read_JobPIDInfo(&cji, jip);
|
||||
|
||||
info = calloc(1, sizeof(*info));
|
||||
if (!info) { capn_free(&c_ctx); return NULL; }
|
||||
decode_JobPIDInfo(info, &cji);
|
||||
capn_free(&c_ctx);
|
||||
return info;
|
||||
}
|
||||
|
||||
/* Serialize a host admin-state frame (config-upgrade-snapshot task 2.1, Req
|
||||
* 1.6/1.7/1.9) to a malloc'd Cap'n Proto buffer. Mirrors capnp_encode_job
|
||||
* above exactly -- capnp_host_admin_state_t (typemap.h) is a frame-only
|
||||
|
|
@ -3513,7 +3579,6 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
|
|||
restore_job_array(p->cluster, payload, payload_len, lsn, NULL);
|
||||
break;
|
||||
|
||||
case WAL_JOB_START:
|
||||
case WAL_JOB_FINISH:
|
||||
case WAL_JOB_FAIL:
|
||||
case WAL_JOB_STOP:
|
||||
|
|
@ -3529,6 +3594,43 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
|
|||
}
|
||||
break;
|
||||
|
||||
case WAL_JOB_START: {
|
||||
/* bug-1687: split out of the shared group above -- mirrors
|
||||
* WAL_JOB_DISPATCH's split just below (bug-1678). This is the one
|
||||
* fact type in that group whose payload can carry a fixed
|
||||
* start_time PLUS trailing Cap'n Proto JobPIDInfo bytes
|
||||
* (persist_submit_job_start_fact's shape) alongside the {jobid,
|
||||
* state} header, so a promoted standby's catch-up replay recovers
|
||||
* the worker-confirmed pids/pgids/exdata and the true start_time
|
||||
* too, not just state. A payload of exactly sizeof(wal_jobstate_t)
|
||||
* (no trailing bytes -- any WAL_JOB_START entry written before this
|
||||
* fix) decodes neither, matching the pre-fix state-only behavior
|
||||
* exactly. */
|
||||
if (payload_len >= sizeof(wal_jobstate_t)) {
|
||||
const wal_jobstate_t* js = (const wal_jobstate_t*)payload;
|
||||
uint64_t start_time = 0;
|
||||
job_pidinfo_t* pidinfo = NULL;
|
||||
uint32_t extra = payload_len - (uint32_t)sizeof(wal_jobstate_t);
|
||||
|
||||
if (extra >= sizeof(start_time)) {
|
||||
memcpy(&start_time,
|
||||
(const uint8_t*)payload + sizeof(wal_jobstate_t),
|
||||
sizeof(start_time));
|
||||
|
||||
uint32_t pidinfo_len = extra - (uint32_t)sizeof(start_time);
|
||||
if (pidinfo_len > 0) {
|
||||
pidinfo = decode_pidinfo(
|
||||
(const uint8_t*)payload + sizeof(wal_jobstate_t)
|
||||
+ sizeof(start_time),
|
||||
pidinfo_len);
|
||||
}
|
||||
}
|
||||
restore_job_start_state(p->cluster, js->jobid, js->state,
|
||||
start_time, pidinfo);
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case WAL_JOB_DISPATCH: {
|
||||
/* bug-1678: split out of the shared group above -- this is the one
|
||||
* fact type whose payload can carry a trailing Cap'n Proto JobAlloc
|
||||
|
|
@ -3835,7 +3937,41 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
|
|||
* cores do for their own (currently unreachable on this dispatch
|
||||
* path) callers. */
|
||||
if (wal_type == WAL_JOB_START) {
|
||||
j->job.start_time = time(NULL);
|
||||
/* bug-1687: a WAL_JOB_START payload can carry a fixed
|
||||
* start_time plus trailing Cap'n Proto JobPIDInfo bytes
|
||||
* beyond the wal_jobstate_t header (persist_submit_job_
|
||||
* start_fact's shape, mirroring bug-1678's WAL_JOB_DISPATCH
|
||||
* trailing-JobAlloc-bytes handling below). Decode and adopt
|
||||
* the wire-carried start_time/pidinfo when present; a
|
||||
* payload of exactly sizeof(wal_jobstate_t) (no trailing
|
||||
* bytes -- any WAL_JOB_START entry written before this fix)
|
||||
* intentionally falls through to the old fabricated-
|
||||
* time(NULL)/pidinfo-untouched behavior for full backward
|
||||
* compatibility. */
|
||||
if (len >= sizeof(wal_jobstate_t) + sizeof(uint64_t)) {
|
||||
uint64_t wire_start_time;
|
||||
memcpy(&wire_start_time,
|
||||
(const uint8_t*)payload + sizeof(wal_jobstate_t),
|
||||
sizeof(wire_start_time));
|
||||
j->job.start_time = (time_t)wire_start_time;
|
||||
|
||||
uint32_t pidinfo_len = len
|
||||
- (uint32_t)sizeof(wal_jobstate_t)
|
||||
- (uint32_t)sizeof(uint64_t);
|
||||
if (pidinfo_len > 0) {
|
||||
job_pidinfo_t* decoded = decode_pidinfo(
|
||||
(const uint8_t*)payload + sizeof(wal_jobstate_t)
|
||||
+ sizeof(uint64_t),
|
||||
pidinfo_len);
|
||||
if (decoded != NULL) {
|
||||
if (j->job.pidinfo != NULL)
|
||||
free_JobPIDInfo_ptr(&(j->job.pidinfo));
|
||||
j->job.pidinfo = decoded;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
j->job.start_time = time(NULL);
|
||||
}
|
||||
} else if (wal_type == WAL_JOB_FINISH) {
|
||||
j->job.finish_time = time(NULL);
|
||||
j->job.run_time = (int)(j->job.finish_time - j->job.start_time);
|
||||
|
|
@ -6855,6 +6991,49 @@ int persist_submit_job_dispatch_fact(persist_t* p, uint64_t jobid, int state,
|
|||
return rc;
|
||||
}
|
||||
|
||||
/* bug-1687: WAL_JOB_START's own richer sibling of persist_submit_job_fact()
|
||||
* -- mirrors persist_submit_job_dispatch_fact() above (bug-1678) exactly,
|
||||
* but for the worker-confirmed-start fact: the sole durable record of the
|
||||
* job's real job_pidinfo_t (pids/pgids/exdata a worker reported) and its
|
||||
* true start_time. Payload shape: wal_jobstate_t header, immediately
|
||||
* followed by a fixed uint64_t start_time (the receipt-time clock, captured
|
||||
* ONCE by the caller rather than re-derived at each apply seam's own
|
||||
* apply-time moment), immediately followed by the Cap'n Proto-encoded
|
||||
* JobPIDInfo bytes -- apply_committed_entry()'s and apply_wal_entry()'s
|
||||
* WAL_JOB_START cases decode exactly this shape, and treat a payload of
|
||||
* exactly sizeof(wal_jobstate_t) (no trailing bytes) as "state-only,
|
||||
* fabricate start_time and leave pidinfo untouched" for backward
|
||||
* compatibility with any WAL_JOB_START entry written before this fix. */
|
||||
int persist_submit_job_start_fact(persist_t* p, uint64_t jobid, int state,
|
||||
uint64_t start_time,
|
||||
const job_pidinfo_t* info)
|
||||
{
|
||||
wal_jobstate_t js = { .jobid = jobid, .state = (uint32_t)state,
|
||||
._pad = 0 };
|
||||
uint8_t* infobuf = NULL;
|
||||
size_t infolen = 0;
|
||||
uint8_t* payload;
|
||||
size_t total;
|
||||
int rc;
|
||||
|
||||
rc = capnp_encode_pidinfo(info, &infobuf, &infolen);
|
||||
if (rc != 0)
|
||||
return rc;
|
||||
|
||||
total = sizeof(js) + sizeof(start_time) + infolen;
|
||||
payload = malloc(total);
|
||||
if (!payload) { free(infobuf); return -ENOMEM; }
|
||||
memcpy(payload, &js, sizeof(js));
|
||||
memcpy(payload + sizeof(js), &start_time, sizeof(start_time));
|
||||
memcpy(payload + sizeof(js) + sizeof(start_time), infobuf, infolen);
|
||||
free(infobuf);
|
||||
|
||||
rc = wal_hook_append_classified(p, WAL_JOB_START, WAL_MODE_RETAIN,
|
||||
payload, (uint32_t)total);
|
||||
free(payload);
|
||||
return rc;
|
||||
}
|
||||
|
||||
/* bug-1679: see persist.h's doc comment for the full rationale. Mirrors
|
||||
* apply_committed_entry()'s own identical Hazard-1-guarded hook fire
|
||||
* (its shared WAL_JOB_START/.../DISPATCH case, right after its
|
||||
|
|
|
|||
|
|
@ -84,6 +84,7 @@ extern void test_apply_seam(void);
|
|||
extern void test_apply_committed_entry(void);
|
||||
extern void test_replay_apply_seam(void);
|
||||
extern void test_wal_job_dispatch_fact(void);
|
||||
extern void test_wal_job_start_fact(void);
|
||||
extern void test_job_dispatch_fail_count(void);
|
||||
extern void test_apply_reapply_idempotency(void);
|
||||
extern void test_apply_correctness_verification(void);
|
||||
|
|
@ -254,6 +255,7 @@ int main(int argc, char* argv[])
|
|||
TEST(test_apply_committed_entry);
|
||||
TEST(test_replay_apply_seam);
|
||||
TEST(test_wal_job_dispatch_fact);
|
||||
TEST(test_wal_job_start_fact);
|
||||
TEST(test_job_dispatch_fail_count);
|
||||
TEST(test_apply_reapply_idempotency);
|
||||
TEST(test_apply_correctness_verification);
|
||||
|
|
|
|||
|
|
@ -475,6 +475,22 @@ static void cbrp_teardown(void)
|
|||
* range in job.c's process-lifetime in-flight registry. Clear it here
|
||||
* so it never outlives this fixture. */
|
||||
inflight_submits_reset_for_test();
|
||||
/* bug-1687: persist_stop() must run BEFORE commit_engine_fini() --
|
||||
* exactly the ordering bug-1452 already fixed in test_persist_truncate.c
|
||||
* (that fix's own note flagged this file as a same-pattern, not-yet-
|
||||
* observed-to-crash follow-up). commit_engine_fini() tears down
|
||||
* g_hce.lock/g_hce.cond and, as its final step, nulls g_hce.p while the
|
||||
* fsync_worker background thread (started by persist_start(), stopped
|
||||
* only by persist_stop()'s pthread_join) can still be mid-flight inside
|
||||
* its unlocked p->on_durable_advance(p->durable_cb_ctx) callback
|
||||
* (persist.c's fsync_worker) -- a genuine TOCTOU data race with no
|
||||
* synchronization between the two threads. Calling persist_stop() here,
|
||||
* first, guarantees the flusher thread cannot be in-flight at all by the
|
||||
* time commit_engine_fini() runs. The later persist_stop() call below
|
||||
* becomes a safe, idempotent no-op on this second invocation (mirrors
|
||||
* bug-1452's fix exactly -- fsync_thread_running is already 0 and wal_fd
|
||||
* is already -1, so both of persist_stop()'s internal branches skip). */
|
||||
persist_stop(&g_hc.persist);
|
||||
commit_engine_fini(&g_hce);
|
||||
persist_stop(&g_hc.persist);
|
||||
persist_fini(&g_hc.persist);
|
||||
|
|
@ -2410,7 +2426,14 @@ static int cbrp_count_jobstate_entries_ex(persist_t *p, uint32_t wal_type,
|
|||
(int)pread(fd, &hdr, sizeof(hdr), (off_t)off),
|
||||
"WAL segment scan: short header read");
|
||||
if (hdr.type == wal_type
|
||||
&& hdr.payload_len == sizeof(wal_jobstate_t)) {
|
||||
&& hdr.payload_len >= sizeof(wal_jobstate_t)) {
|
||||
/* bug-1687: >= not == -- WAL_JOB_START (like bug-1678's
|
||||
* WAL_JOB_DISPATCH before it) can now carry trailing bytes
|
||||
* (start_time + Cap'n Proto JobPIDInfo) beyond the fixed
|
||||
* wal_jobstate_t header. jobid/state always live at the same
|
||||
* fixed leading offset regardless of trailing bytes, so reading
|
||||
* just sizeof(wal_jobstate_t) here remains correct for any
|
||||
* payload_len >= that size. */
|
||||
wal_jobstate_t js;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
|
|
|
|||
673
src/controller/testsuite/test_wal_job_start_fact.c
Normal file
673
src/controller/testsuite/test_wal_job_start_fact.c
Normal file
|
|
@ -0,0 +1,673 @@
|
|||
/* test_wal_job_start_fact.c — bug-1687 regression suite
|
||||
*
|
||||
* Proves WAL_JOB_START's payload -- previously bare wal_jobstate_t{jobid,
|
||||
* state}, exactly the same silent-field-drop bug class as bug-1678's
|
||||
* WAL_JOB_DISPATCH -- now carries the worker-reported job_pidinfo_t (real
|
||||
* pids/pgids/exdata) AND a receipt-time start_time through BOTH of
|
||||
* persist.c's apply seams:
|
||||
*
|
||||
* - apply_committed_entry() (live-commit apply / boot replay dispatch)
|
||||
* - apply_wal_entry() (standby catch-up / replicated-apply dispatch)
|
||||
*
|
||||
* and that a real, production on_job_start() call (the actual worker-START
|
||||
* receipt handler) survives a real controller restart (WAL write +
|
||||
* persist_recover() over a real on-disk WAL) with job.pidinfo and
|
||||
* job.start_time intact, not just job.state (Req: mirrors bug-1678's own
|
||||
* acceptance criterion, "after recover the job should get the same as it's
|
||||
* state before shutdown").
|
||||
*
|
||||
* Also proves backward compatibility: a WAL_JOB_START entry with the OLD
|
||||
* payload shape (exactly sizeof(wal_jobstate_t), no trailing bytes --
|
||||
* simulating an entry written before this fix) still decodes cleanly, with
|
||||
* pidinfo left NULL and start_time fabricated at apply time exactly as
|
||||
* before this fix (apply_committed_entry) / left untouched exactly as
|
||||
* before this fix (apply_wal_entry) -- no crash, no misread.
|
||||
*
|
||||
* Fixtures mirror test_wal_job_dispatch_fact.c's own conventions exactly:
|
||||
* - wjs_setup/wjs_teardown: bare cluster_t (jobmgr + hooksys) + in-memory
|
||||
* persist_t.
|
||||
* - wjs_make_fresh_persist/wjs_teardown_persist: real disk-backed
|
||||
* persist_t + persist_recover(), for restart-replay tests.
|
||||
*/
|
||||
#include <rtti.h>
|
||||
|
||||
#include <capnp_c.h>
|
||||
#include <errno.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 "persist.h"
|
||||
#include "wal_types.h"
|
||||
#include "xce/cluster_def.h"
|
||||
#include "xce/types.h"
|
||||
#include "xcealloc.h"
|
||||
#include "xstring.h"
|
||||
|
||||
/* jobj.c's jobsub 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. */
|
||||
static int g_wjs_jobobj_ready;
|
||||
|
||||
static cluster_t g_wjs_c;
|
||||
static persist_t g_wjs_p;
|
||||
|
||||
/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */
|
||||
static int g_wjs_hook_fires;
|
||||
|
||||
static int wjs_spy_hook(int type, int situation, int detail, void* data,
|
||||
va_list ap)
|
||||
{
|
||||
(void)type; (void)detail; (void)data; (void)ap;
|
||||
if (situation == HOOK_AFTER_EVENT)
|
||||
g_wjs_hook_fires++;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void wjs_setup(void)
|
||||
{
|
||||
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
|
||||
.nslots = 32,
|
||||
.delta = 16 };
|
||||
|
||||
if (makeString((char*)"wjs-probe") == NULL)
|
||||
xstring_table_init(64);
|
||||
|
||||
if (!g_wjs_jobobj_ready) {
|
||||
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
|
||||
g_wjs_jobobj_ready = 1;
|
||||
}
|
||||
|
||||
memset(&g_wjs_c, 0, sizeof(g_wjs_c));
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
xce_hook_init(&g_wjs_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
|
||||
|
||||
g_wjs_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
|
||||
TEST_ASSERT_NOT_NULL(g_wjs_c.jobmgr);
|
||||
|
||||
memset(&g_wjs_p, 0, sizeof(g_wjs_p));
|
||||
g_wjs_p.cluster = &g_wjs_c;
|
||||
|
||||
g_wjs_hook_fires = 0;
|
||||
}
|
||||
|
||||
static void wjs_teardown(void)
|
||||
{
|
||||
xce_hook_fini(&g_wjs_c.hooksys);
|
||||
if (g_wjs_c.jobmgr)
|
||||
dereference(g_wjs_c.jobmgr);
|
||||
}
|
||||
|
||||
/* Look up a job_t wrapper by jobid (NULL if absent). */
|
||||
static job_t* wjs_find_job(uint64_t jobid)
|
||||
{
|
||||
job_t* j = NULL;
|
||||
lock(g_wjs_c.jobmgr, RTTI_LOCK_READ);
|
||||
int found = key_find(g_wjs_c.jobmgr, (void*)jobid, (void**)&j);
|
||||
unlock(g_wjs_c.jobmgr);
|
||||
return (found > 0) ? j : NULL;
|
||||
}
|
||||
|
||||
/* 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. */
|
||||
static job_submit_data_t g_wjs_seed_submit;
|
||||
|
||||
static job_t* wjs_seed_job(uint64_t jobid, uint32_t state)
|
||||
{
|
||||
xce_job_t jdef;
|
||||
|
||||
memset(&g_wjs_seed_submit, 0, sizeof(g_wjs_seed_submit));
|
||||
g_wjs_seed_submit.queue = (char*)"q";
|
||||
g_wjs_seed_submit.user = (char*)"tester";
|
||||
|
||||
memset(&jdef, 0, sizeof(jdef));
|
||||
jdef.jobid = jobid;
|
||||
jdef.state = state;
|
||||
jdef.submit = &g_wjs_seed_submit;
|
||||
|
||||
return makeJob(g_wjs_c.jobmgr, &jdef);
|
||||
}
|
||||
|
||||
/* Build a wal_jobstate_t header + fixed uint64_t start_time + trailing
|
||||
* Cap'n Proto JobPIDInfo bytes -- persist_submit_job_start_fact()'s payload
|
||||
* shape -- mirroring wjd_build_dispatch_payload() (test_wal_job_dispatch_
|
||||
* fact.c) since job.c/persist.c's own static encode helpers aren't linkable
|
||||
* from a test TU. Caller owns *out. */
|
||||
static int wjs_build_start_payload(uint64_t jobid, uint32_t state,
|
||||
uint64_t start_time,
|
||||
job_pidinfo_t* pidinfo,
|
||||
uint8_t** out, size_t* out_len)
|
||||
{
|
||||
struct capn c;
|
||||
struct capn_segment* cs;
|
||||
struct JobPIDInfo cji;
|
||||
JobPIDInfo_ptr jip;
|
||||
uint8_t* infobuf;
|
||||
size_t infolen;
|
||||
wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 };
|
||||
|
||||
capn_init_malloc(&c);
|
||||
cs = capn_root(&c).seg;
|
||||
encode_JobPIDInfo(cs, &cji, pidinfo);
|
||||
jip = new_JobPIDInfo(cs);
|
||||
write_JobPIDInfo(&cji, jip);
|
||||
capn_setp(capn_root(&c), 0, jip.p);
|
||||
|
||||
infolen = capn_size(&c);
|
||||
infobuf = malloc(infolen);
|
||||
if (!infobuf) { capn_free(&c); return -ENOMEM; }
|
||||
capn_write_mem(&c, infobuf, infolen, 0);
|
||||
capn_free(&c);
|
||||
|
||||
*out_len = sizeof(js) + sizeof(start_time) + infolen;
|
||||
*out = malloc(*out_len);
|
||||
if (!*out) { free(infobuf); return -ENOMEM; }
|
||||
memcpy(*out, &js, sizeof(js));
|
||||
memcpy(*out + sizeof(js), &start_time, sizeof(start_time));
|
||||
memcpy(*out + sizeof(js) + sizeof(start_time), infobuf, infolen);
|
||||
free(infobuf);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Real pids/pgids/exdata -- mirrors what a worker actually reports on a
|
||||
* genuine job start. */
|
||||
static unsigned int g_wjs_test_pids[3] = { 41001, 41002, 41003 };
|
||||
static unsigned int g_wjs_test_pgids[2] = { 41001, 41500 };
|
||||
|
||||
static job_pidinfo_t wjs_make_test_pidinfo(uint64_t jobid)
|
||||
{
|
||||
job_pidinfo_t info;
|
||||
memset(&info, 0, sizeof(info));
|
||||
info.jobid = jobid;
|
||||
info.npids = 3;
|
||||
info.pids = g_wjs_test_pids;
|
||||
info.npgids = 2;
|
||||
info.pgids = g_wjs_test_pgids;
|
||||
info.exdata = (char*)"wjs-exdata-blob";
|
||||
return info;
|
||||
}
|
||||
|
||||
static void wjs_assert_pidinfo_matches(job_pidinfo_t* got, uint64_t jobid)
|
||||
{
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(got,
|
||||
"a WAL_JOB_START payload with trailing JobPIDInfo bytes must set "
|
||||
"job.pidinfo (bug-1687), not just state");
|
||||
TEST_ASSERT_EQUAL_UINT64(jobid, got->jobid);
|
||||
TEST_ASSERT_EQUAL_INT(3, got->npids);
|
||||
TEST_ASSERT_EQUAL_UINT32(41001, got->pids[0]);
|
||||
TEST_ASSERT_EQUAL_UINT32(41002, got->pids[1]);
|
||||
TEST_ASSERT_EQUAL_UINT32(41003, got->pids[2]);
|
||||
TEST_ASSERT_EQUAL_INT(2, got->npgids);
|
||||
TEST_ASSERT_EQUAL_UINT32(41001, got->pgids[0]);
|
||||
TEST_ASSERT_EQUAL_UINT32(41500, got->pgids[1]);
|
||||
TEST_ASSERT_EQUAL_STRING("wjs-exdata-blob", got->exdata);
|
||||
}
|
||||
|
||||
/* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */
|
||||
|
||||
static void test_wjs_apply_committed_entry_sets_pidinfo_and_start_time(void)
|
||||
{
|
||||
wjs_setup();
|
||||
|
||||
job_t* j = wjs_seed_job(72001, JOB_STAT_PEND);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
|
||||
job_pidinfo_t info = wjs_make_test_pidinfo(72001);
|
||||
uint64_t wire_start_time = 1700000001ULL;
|
||||
|
||||
uint8_t* payload = NULL;
|
||||
size_t plen = 0;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
wjs_build_start_payload(72001, JOB_STAT_RUN, wire_start_time, &info,
|
||||
&payload, &plen));
|
||||
|
||||
xce_add_hook(&g_wjs_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjs_spy_hook,
|
||||
NULL);
|
||||
|
||||
int rc = apply_committed_entry(&g_wjs_p, WAL_JOB_START, 200, 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,
|
||||
"WAL_JOB_START must adopt the wire-carried start_time (bug-1687), "
|
||||
"not fabricate time(NULL) when one is present");
|
||||
wjs_assert_pidinfo_matches(j->job.pidinfo, 72001);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wjs_hook_fires,
|
||||
"a forward WAL_JOB_START transition must still fire exactly one "
|
||||
"notification");
|
||||
|
||||
wjs_teardown();
|
||||
printf("test_wjs_apply_committed_entry_sets_pidinfo_and_start_time: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
static void
|
||||
test_wjs_apply_committed_entry_backward_compat_old_payload(void)
|
||||
{
|
||||
wjs_setup();
|
||||
|
||||
job_t* j = wjs_seed_job(72002, JOB_STAT_PEND);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
TEST_ASSERT_NULL_MESSAGE(j->job.pidinfo,
|
||||
"fixture must seed a job with pidinfo still unset");
|
||||
|
||||
wal_jobstate_t js = { .jobid = 72002, .state = JOB_STAT_RUN, ._pad = 0 };
|
||||
|
||||
time_t before = time(NULL);
|
||||
int rc = apply_committed_entry(&g_wjs_p, WAL_JOB_START, 201, &js,
|
||||
sizeof(js));
|
||||
time_t after = time(NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
(j->job.start_time >= before) && (j->job.start_time <= after),
|
||||
"an old-format (exactly sizeof(wal_jobstate_t)) WAL_JOB_START "
|
||||
"payload must still fabricate start_time at apply time, unchanged "
|
||||
"from before this fix");
|
||||
TEST_ASSERT_NULL_MESSAGE(j->job.pidinfo,
|
||||
"an old-format WAL_JOB_START payload must leave pidinfo NULL -- no "
|
||||
"crash, no misread of absent trailing bytes");
|
||||
|
||||
wjs_teardown();
|
||||
printf("test_wjs_apply_committed_entry_backward_compat_old_payload: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */
|
||||
|
||||
static void
|
||||
test_wjs_apply_wal_entry_sets_pidinfo_and_start_time_no_hook(void)
|
||||
{
|
||||
wjs_setup();
|
||||
|
||||
job_t* j = wjs_seed_job(72003, JOB_STAT_PEND);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
|
||||
job_pidinfo_t info = wjs_make_test_pidinfo(72003);
|
||||
uint64_t wire_start_time = 1700000003ULL;
|
||||
|
||||
uint8_t* payload = NULL;
|
||||
size_t plen = 0;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
wjs_build_start_payload(72003, JOB_STAT_RUN, wire_start_time, &info,
|
||||
&payload, &plen));
|
||||
|
||||
xce_add_hook(&g_wjs_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjs_spy_hook,
|
||||
NULL);
|
||||
|
||||
int rc = apply_wal_entry(&g_wjs_p, WAL_JOB_START, 202, 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,
|
||||
"apply_wal_entry's catch-up dispatch must also recover the "
|
||||
"wire-carried start_time (bug-1687), required for a promoted "
|
||||
"standby");
|
||||
wjs_assert_pidinfo_matches(j->job.pidinfo, 72003);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjs_hook_fires,
|
||||
"apply_wal_entry's catch-up dispatch must never fire a notification "
|
||||
"hook");
|
||||
|
||||
wjs_teardown();
|
||||
printf("test_wjs_apply_wal_entry_sets_pidinfo_and_start_time_no_hook: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
static void test_wjs_apply_wal_entry_backward_compat_old_payload(void)
|
||||
{
|
||||
wjs_setup();
|
||||
|
||||
job_t* j = wjs_seed_job(72004, JOB_STAT_PEND);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
j->job.start_time = 12345; /* sentinel -- must survive untouched */
|
||||
|
||||
wal_jobstate_t js = { .jobid = 72004, .state = JOB_STAT_RUN, ._pad = 0 };
|
||||
|
||||
int rc = apply_wal_entry(&g_wjs_p, WAL_JOB_START, 203, &js, sizeof(js));
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(12345, (uint64_t)j->job.start_time,
|
||||
"an old-format WAL_JOB_START payload must leave start_time "
|
||||
"untouched on this seam, unchanged from before this fix");
|
||||
TEST_ASSERT_NULL_MESSAGE(j->job.pidinfo,
|
||||
"an old-format WAL_JOB_START payload must leave pidinfo NULL");
|
||||
|
||||
wjs_teardown();
|
||||
printf("test_wjs_apply_wal_entry_backward_compat_old_payload: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Restart-replay Observables ──────────────────────────────────────── */
|
||||
|
||||
static persist_t* wjs_make_fresh_persist(const char* tmpdir)
|
||||
{
|
||||
persist_t* p = calloc(1, sizeof(persist_t));
|
||||
if (!p) return NULL;
|
||||
|
||||
persist_config_t cfg = persist_config_defaults();
|
||||
cfg.wal_sync_ms = 5000; /* slow fsync — no races in tests */
|
||||
|
||||
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 wjs_teardown_persist(persist_t* p)
|
||||
{
|
||||
persist_stop(p);
|
||||
persist_fini(p);
|
||||
free(p);
|
||||
}
|
||||
|
||||
/* Encode helper mirroring test_wal_job_dispatch_fact.c's wjd_encode_job
|
||||
* (capnp_encode_job is file-static in persist.c). */
|
||||
static int wjs_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;
|
||||
}
|
||||
|
||||
/* Direct-payload restart replay: appends WAL_JOB_SUBMIT + a new-format
|
||||
* WAL_JOB_START (real pidinfo + a fixed start_time) straight onto disk via
|
||||
* persist_wal_async, then replays via a fresh persist_t's persist_recover()
|
||||
* (a real restart). Isolates the WAL fact shape from the live caller,
|
||||
* mirroring test_wal_job_dispatch_fact.c's own restart-replay test. */
|
||||
static void test_wjs_restart_replay_recovers_pidinfo_and_start_time(void)
|
||||
{
|
||||
char tmp[256];
|
||||
strcpy(tmp, "/tmp/xce_wjs_replay_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
|
||||
|
||||
uint64_t wire_start_time = 1700009999ULL;
|
||||
|
||||
/* Phase 1: write the log to real disk. */
|
||||
{
|
||||
persist_t* p = wjs_make_fresh_persist(tmp);
|
||||
TEST_ASSERT_NOT_NULL(p);
|
||||
|
||||
job_submit_data_t submit;
|
||||
memset(&submit, 0, sizeof(submit));
|
||||
submit.queue = (char*)"q";
|
||||
submit.user = (char*)"tester";
|
||||
|
||||
xce_job_t jdef;
|
||||
memset(&jdef, 0, sizeof(jdef));
|
||||
jdef.jobid = 72005;
|
||||
jdef.state = JOB_STAT_PEND;
|
||||
jdef.submit = &submit;
|
||||
|
||||
uint8_t* job_payload = NULL;
|
||||
size_t job_plen = 0;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
wjs_encode_job(&jdef, &job_payload, &job_plen));
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
persist_wal_async(p, WAL_JOB_SUBMIT, job_payload,
|
||||
(uint32_t)job_plen));
|
||||
free(job_payload);
|
||||
|
||||
job_pidinfo_t info = wjs_make_test_pidinfo(72005);
|
||||
uint8_t* start_payload = NULL;
|
||||
size_t start_plen = 0;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
wjs_build_start_payload(72005, JOB_STAT_RUN, wire_start_time,
|
||||
&info, &start_payload, &start_plen));
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
persist_wal_async(p, WAL_JOB_START, start_payload,
|
||||
(uint32_t)start_plen));
|
||||
free(start_payload);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
|
||||
wjs_teardown_persist(p);
|
||||
}
|
||||
|
||||
/* Phase 2: fresh persist_t + fresh cluster_t, recover from disk —
|
||||
* simulates a controller restart. */
|
||||
{
|
||||
persist_t* p2 = wjs_make_fresh_persist(tmp);
|
||||
TEST_ASSERT_NOT_NULL(p2);
|
||||
|
||||
wjs_setup();
|
||||
p2->cluster = &g_wjs_c;
|
||||
|
||||
int rc = persist_recover(p2);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
job_t* j = wjs_find_job(72005);
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(j,
|
||||
"restart replay must construct the job via WAL_JOB_SUBMIT");
|
||||
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(wire_start_time,
|
||||
(uint64_t)j->job.start_time,
|
||||
"restart replay must recover the exact wire-carried start_time "
|
||||
"(bug-1687), not a freshly fabricated time(NULL)");
|
||||
wjs_assert_pidinfo_matches(j->job.pidinfo, 72005);
|
||||
|
||||
wjs_teardown();
|
||||
wjs_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_wjs_restart_replay_recovers_pidinfo_and_start_time: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* bug-1687's primary regression test: drives the REAL production
|
||||
* on_job_start() path (the actual worker-START receipt handler --
|
||||
* record_job_start_fact()'s only caller) with a real job_pidinfo_t (real
|
||||
* pids/pgids/exdata), through a real WAL-write + persist_recover() restart
|
||||
* cycle -- the exact technique test_dc_success_appends_wal_job_dispatch_
|
||||
* durably (test_dispatch_commit.c) already established for bug-1678.
|
||||
* Asserts the job's pidinfo AND start_time survive the restart intact, not
|
||||
* just state -- the audit's exact confirmed root cause. */
|
||||
static void test_wjs_real_on_job_start_survives_restart(void)
|
||||
{
|
||||
char tmp[256];
|
||||
strcpy(tmp, "/tmp/xce_wjs_onstart_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
|
||||
|
||||
const uint64_t jobid = 72006;
|
||||
time_t before_call, after_call;
|
||||
|
||||
/* Phase 1: real cluster + real disk persist. Seed WAL_JOB_SUBMIT
|
||||
* directly (so replay can reconstruct the job) AND a live in-memory job
|
||||
* object (so the real on_job_start() call site has something to find
|
||||
* and lock) -- mirrors test_dc_success_appends_wal_job_dispatch_durably's
|
||||
* dual-seed pattern exactly. Then drive an actual on_job_start() call
|
||||
* and let IT append WAL_JOB_START -- this is the production code path
|
||||
* under test, not a hand-built WAL entry. */
|
||||
{
|
||||
persist_config_t cfg;
|
||||
|
||||
if (makeString((char*)"wjs-onstart-probe") == NULL)
|
||||
xstring_table_init(64);
|
||||
if (!g_wjs_jobobj_ready) {
|
||||
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
|
||||
g_wjs_jobobj_ready = 1;
|
||||
}
|
||||
|
||||
cluster_t c;
|
||||
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
|
||||
.nslots = 32, .delta = 16 };
|
||||
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));
|
||||
|
||||
/* Durable WAL_JOB_SUBMIT seed. */
|
||||
{
|
||||
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*)"oq";
|
||||
wal_submit.user = (char*)"tester";
|
||||
|
||||
memset(&wal_jdef, 0, sizeof(wal_jdef));
|
||||
wal_jdef.jobid = jobid;
|
||||
wal_jdef.state = JOB_STAT_PEND;
|
||||
wal_jdef.submit = &wal_submit;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
wjs_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);
|
||||
}
|
||||
|
||||
/* Live in-memory job object for on_job_start() to find/lock. */
|
||||
{
|
||||
job_submit_data_t submit;
|
||||
xce_job_t jdef;
|
||||
|
||||
memset(&submit, 0, sizeof(submit));
|
||||
submit.queue = (char*)"oq";
|
||||
submit.user = (char*)"tester";
|
||||
|
||||
memset(&jdef, 0, sizeof(jdef));
|
||||
jdef.jobid = jobid;
|
||||
jdef.state = JOB_STAT_PEND;
|
||||
jdef.submit = &submit;
|
||||
|
||||
TEST_ASSERT_NOT_NULL(makeJob(c.jobmgr, &jdef));
|
||||
}
|
||||
|
||||
/* The real worker-reported pids/pgids/exdata. */
|
||||
job_pidinfo_t info = wjs_make_test_pidinfo(jobid);
|
||||
|
||||
before_call = time(NULL);
|
||||
int rc = on_job_start(&c, &info);
|
||||
after_call = time(NULL);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
|
||||
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 = wjs_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*)jobid, (void**)&j2);
|
||||
unlock(c2.jobmgr);
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(j2,
|
||||
"restart replay must reconstruct the job via WAL_JOB_SUBMIT");
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j2->job.state,
|
||||
"restart replay must recover the job as RUNNING from the "
|
||||
"WAL_JOB_START fact appended by the real on_job_start() call");
|
||||
|
||||
/* bug-1687: restart replay must recover not just STATE but the
|
||||
* worker-reported pidinfo AND the true start_time -- the audit's
|
||||
* exact confirmed root cause (mirrors bug-1678's own acceptance
|
||||
* criterion for WAL_JOB_DISPATCH/job.allocs). */
|
||||
wjs_assert_pidinfo_matches(j2->job.pidinfo, jobid);
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
(j2->job.start_time >= before_call)
|
||||
&& (j2->job.start_time <= after_call),
|
||||
"restart replay must recover the true, receipt-time start_time "
|
||||
"captured by the real on_job_start() call, not a fresh "
|
||||
"apply-time time(NULL) fabrication");
|
||||
|
||||
xce_hook_fini(&c2.hooksys);
|
||||
if (c2.jobmgr) dereference(c2.jobmgr);
|
||||
wjs_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_wjs_real_on_job_start_survives_restart: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test group entry point ───────────────────────────────────────────── */
|
||||
|
||||
void test_wal_job_start_fact(void)
|
||||
{
|
||||
UnitySetTestFile(__FILE__);
|
||||
RUN_TEST(test_wjs_apply_committed_entry_sets_pidinfo_and_start_time);
|
||||
RUN_TEST(test_wjs_apply_committed_entry_backward_compat_old_payload);
|
||||
RUN_TEST(test_wjs_apply_wal_entry_sets_pidinfo_and_start_time_no_hook);
|
||||
RUN_TEST(test_wjs_apply_wal_entry_backward_compat_old_payload);
|
||||
RUN_TEST(test_wjs_restart_replay_recovers_pidinfo_and_start_time);
|
||||
RUN_TEST(test_wjs_real_on_job_start_survives_restart);
|
||||
}
|
||||
|
|
@ -1421,6 +1421,22 @@ int persist_submit_job_fact_ex(persist_t* p, uint32_t wal_type,
|
|||
int persist_submit_job_dispatch_fact(persist_t* p, uint64_t jobid, int state,
|
||||
const xce_job_alloc_t* alloc);
|
||||
|
||||
/* bug-1687: WAL_JOB_START's own sibling of persist_submit_job_fact(),
|
||||
* mirroring persist_submit_job_dispatch_fact() (bug-1678) exactly but for
|
||||
* the worker-confirmed-start fact -- additionally encoding the
|
||||
* worker-reported job_pidinfo_t (real pids/pgids/exdata) AND a receipt-time
|
||||
* start_time into the fact's payload (header {jobid, state} + a fixed
|
||||
* uint64_t start_time + trailing Cap'n Proto JobPIDInfo bytes) so a
|
||||
* restart/catch-up replay can reconstruct job.pidinfo and the true
|
||||
* job.start_time instead of leaving pidinfo permanently NULL and
|
||||
* start_time fabricated fresh (time(NULL)) at every replay. `info` must be
|
||||
* non-NULL (the real call site, record_job_start_fact(), always has one by
|
||||
* construction -- it is the worker's START report itself). Returns the
|
||||
* local-append rc, same contract as persist_submit_job_fact(). */
|
||||
int persist_submit_job_start_fact(persist_t* p, uint64_t jobid, int state,
|
||||
uint64_t start_time,
|
||||
const job_pidinfo_t* info);
|
||||
|
||||
/* bug-1679: fire XCE_CONTROLLER_HOOK_JOB's HOOK_DETAIL_JOB_UPDATE AFTER_EVENT
|
||||
* for a job whose durable WAL fact the caller has ALREADY appended itself
|
||||
* (e.g. dispatch_a_job_()'s own persist_submit_job_dispatch_fact() call,
|
||||
|
|
|
|||
|
|
@ -101,7 +101,17 @@ _Static_assert(sizeof(wal_entry_hdr_t) == 40,
|
|||
* ============================================================ */
|
||||
typedef enum {
|
||||
WAL_JOB_SUBMIT = 1, /* SYNC replication */
|
||||
WAL_JOB_START = 2,
|
||||
WAL_JOB_START = 2, /* worker-confirmed job start. Payload is
|
||||
* wal_jobstate_t{jobid, state=RUN, _pad=0} for any
|
||||
* entry written before bug-1687; as of bug-1687 the
|
||||
* real producer (persist_submit_job_start_fact)
|
||||
* appends a fixed uint64_t start_time (receipt-time
|
||||
* clock, not fabricated at apply time) plus
|
||||
* trailing Cap'n Proto JobPIDInfo bytes (the
|
||||
* worker-reported pids/pgids/exdata) -- both apply
|
||||
* seams treat a payload of exactly
|
||||
* sizeof(wal_jobstate_t) as the legacy state-only
|
||||
* shape for full backward compatibility. */
|
||||
WAL_JOB_FINISH = 3,
|
||||
WAL_JOB_FAIL = 4,
|
||||
WAL_JOB_STOP = 5,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue