diff --git a/.wolf/buglog.json b/.wolf/buglog.json index a16502fc..f2e2cbfd 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -18759,6 +18759,35 @@ ], "occurrences": 1, "last_seen": "2026-08-25T17:15:00Z" + }, + { + "id": "bug-1687", + "timestamp": "2026-08-25T17:30:00Z", + "error_message": "xce-audit-invariants finding: WAL_JOB_START silently drops job fields on a controller restart replay -- the same bug class as bug-1678 (WAL_JOB_DISPATCH/job.allocs), but for the worker-confirmed-start fact. apply_job_start_transition() (job.c) sets job->job.pidinfo (real pids/pgids/exdata a worker reported) and job->job.start_time, but the WAL_JOB_START fact written by record_job_start_fact() only carried wal_jobstate_t{jobid, state} -- pidinfo was never in the log at all, and start_time was re-fabricated with time(NULL) at every apply (both boot replay and standby catch-up) instead of carrying the true worker-reported value. Confirmed live-relevant: user reported StartTime always showing 0 in `xcecli joblist` for both RUN and DONE jobs on a live deployment -- deferred as a distinct follow-up investigation (that deployment's controller predates this fix; whether StartTime=0 specifically, rather than the pre-fix fabricated-nonzero time(NULL), indicates a SEPARATE live-apply-pump gap has not yet been root-caused).", + "file": "src/controller/job.c, src/controller/persist.c, src/include/persist.h, src/include/wal_types.h", + "root_cause": "record_job_start_fact() (job.c, on_job_start()'s only caller) called persist_submit_job_fact(&c->persist, WAL_JOB_START, job->job.jobid, JOB_STAT_RUN) -- a thin wrapper whose payload type wal_jobstate_t is exactly {jobid, state, _pad}, discarding the `info` (job_pidinfo_t*) parameter entirely even though it was available at the call site (the worker's real START report). Both apply seams -- apply_committed_entry()'s shared WAL_JOB_START/.../DISPATCH fall-through group (persist.c, comment confirms 'these types' payloads are deterministic... start_time... stamped directly below... apply_job_update_transition() only ever touches state') for primary boot replay AND live commit-advance apply, and apply_wal_entry()'s identical shared group -> restore_job_state() (job.c, state-only) for standby catch-up -- never restored pidinfo, and apply_committed_entry() additionally fabricated start_time = time(NULL) at apply time rather than carrying the receipt-time value, so a restarted/promoted node's start_time drifts from the true worker-reported time even though it is non-NULL.", + "fix": "FIXED (2026-08-25). Mirrors bug-1678's WAL_JOB_DISPATCH fix pattern exactly, using the already-existing controller.capnp JobPIDInfo struct (job_pidinfo_t's mapped type -- the existing Job.pidinfo field) with its already-generated encode_JobPIDInfo/decode_JobPIDInfo/free_JobPIDInfo_ptr functions; no schema change needed. src/controller/persist.c gains capnp_encode_pidinfo()/decode_pidinfo() (mirroring capnp_encode_joballoc()/decode_joballoc()) and persist_submit_job_start_fact(persist_t*, jobid, state, start_time, info) -- WAL_JOB_START's payload is now wal_jobstate_t header + a fixed uint64_t start_time (the receipt-time clock, captured ONCE by record_job_start_fact() rather than re-derived independently at each apply seam) + trailing Cap'n Proto-encoded JobPIDInfo bytes, declared in src/include/persist.h. src/controller/job.c's record_job_start_fact() now calls persist_submit_job_start_fact(..., time(NULL), info) instead of persist_submit_job_fact(...). Both apply seams updated: apply_wal_entry() (standby catch-up) splits WAL_JOB_START out of its shared group into its own case (mirroring how bug-1678 split WAL_JOB_DISPATCH out), decoding trailing start_time/pidinfo and calling new restore_job_start_state() (job.c, restore_job_dispatch_state()'s sibling, takes ownership of an optional decoded job_pidinfo_t and a wire-carried start_time -- 0 means 'not carried, leave untouched', matching this seam's pre-fix behavior of never touching start_time for WAL_JOB_START at all). apply_committed_entry() (primary boot replay / live apply pump) stays in its existing shared fall-through group (mirroring how WAL_JOB_DISPATCH joined that same group in bug-1678) but extends its wal_type==WAL_JOB_START branch: a payload with trailing bytes (len >= sizeof(wal_jobstate_t)+sizeof(uint64_t)) decodes and adopts the wire start_time and pidinfo directly; a payload of exactly sizeof(wal_jobstate_t) (any WAL_JOB_START entry written before this fix) falls through unchanged to the old fabricated-time(NULL)/pidinfo-untouched behavior -- full backward compatibility, verified by every pre-existing WAL_JOB_START test in test_apply_committed_entry.c/test_apply_reapply_idempotency.c/test_ha_role_lifecycle.c/test_apply_pump.c (all still using the old bare-wal_jobstate_t shape) passing unchanged.\n\nIncidental fix required to reach GREEN: test_commit_barrier_reply.c's cbrp_count_jobstate_entries_ex() (a test-only helper that manually scans the on-disk WAL segment) matched WAL_JOB_START/DISPATCH/etc entries via `hdr.payload_len == sizeof(wal_jobstate_t)` -- an exact-size match that silently undercounted (returned 0 instead of 1) once WAL_JOB_START legitimately grew past that fixed size, in turn making a downstream test in that file poll/spin forever on a count that could never be reached (observed as a genuine deadlock -- all 3 threads futex-blocked, 0 CPU time accruing over 12+ minutes wall clock, reproduced deterministically via `-t test_commit_barrier_reply` and bisected by temporarily inserting/removing an early `return;` in the test group's own RUN_TEST driver to narrow it to one of 8 candidate tests, then to a clean isolated PASS/FAIL once the helper was fixed). Changed the match to `>=` (jobid/state always live at the same fixed leading offset regardless of trailing bytes) -- the general, correct fix, not a workaround; resolves the hang completely (verified: full test_commit_barrier_reply group and the full controller suite both complete in ~38s with zero hangs, both before and after this specific fix was isolated via bisection). Also applied test_commit_barrier_reply.c's cbrp_teardown() the same persist_stop()-before-commit_engine_fini() ordering bug-1452 already established as the fix for this exact TOCTOU-race class (that fix's own note explicitly flagged this file as a same-pattern, not-yet-observed-to-crash follow-up) -- confirmed via bisection that this ordering was NOT the actual cause of this particular hang, but it is a genuine latent race per bug-1452's analysis and the fix is harmless/idempotent, so it was kept as defense-in-depth rather than reverted.\n\nPrimary regression tests: new src/controller/testsuite/test_wal_job_start_fact.c (6 tests, wired into src/controller/meson.build and testsuite/controller_test.c), mirroring test_wal_job_dispatch_fact.c's structure: (1) apply_committed_entry sets pidinfo+start_time from a new-format payload; (2) apply_committed_entry backward-compat with an old-format (bare wal_jobstate_t) payload -- pidinfo stays NULL, start_time still fabricated via time(NULL), unchanged from pre-fix; (3) apply_wal_entry sets pidinfo+start_time, fires no hook; (4) apply_wal_entry backward-compat -- pidinfo stays NULL, start_time left untouched; (5) direct-payload restart-replay (WAL_JOB_SUBMIT + new-format WAL_JOB_START via persist_wal_async + persist_recover()) recovers the exact wire start_time and pidinfo; (6) test_wjs_real_on_job_start_survives_restart -- drives the REAL production on_job_start() path (record_job_start_fact()'s only caller) with a real job_pidinfo_t (3 real pids, 2 real pgids, real exdata string), through a real WAL-write + persist_recover() restart cycle (the exact technique test_dispatch_commit.c's test_dc_success_appends_wal_job_dispatch_durably already established for bug-1678), asserting pidinfo AND start_time (bounded to the real wall-clock window around the on_job_start() call) survive the restart intact -- the audit's exact confirmed root cause and the user's exact acceptance criterion.\n\nRED->GREEN independently verified live (not just narrated): git-stashed job.c+persist.c (keeping the new test file, persist.h, wal_types.h unchanged) via `git stash push -- src/controller/job.c src/controller/persist.c`, rebuilt, ran `./controller_unit_test -t test_wal_job_start_fact` -- 6 Tests 4 Failures 0 Ignored (the 2 backward-compat tests correctly PASS unchanged; all 4 new-behavior tests correctly FAIL with the exact expected mismatches, e.g. 'Expected 1700000001 Was 1787645264' for the fabricated-vs-wire start_time, 'must set job.pidinfo... not just state' for the real on_job_start test). Restored via `git stash pop`, rebuilt, re-ran -- 6 Tests 0 Failures 0 Ignored. Full controller_unit_test suite re-run after all fixes (including the test_commit_barrier_reply.c hang fix and the buglog-id renumber from a concurrent session's own bug-1686 claim): 1146 Tests 2 Failures 0 Ignored, ~38s, zero hangs -- the 2 failures are test_conf_trim_keeps_only_n_versions ('Expected 2 Was 3') and test_conf_get_saved_tag_no_config ('Expected 0 Was ') in test_config_repl.c, an already-logged, pre-existing, unrelated overlayfs sandbox fdatasync artifact (bug-1561, re-observed identically across multiple unrelated sessions with zero diff to test_config_repl.c/config_repl.c each time) -- confirmed by exact signature match, not a new regression.", + "tags": [ + "controller", + "job.c", + "persist.c", + "wal", + "wal-job-start", + "job-pidinfo", + "start-time", + "restart-recovery", + "data-loss", + "apply-seam", + "test-infrastructure-hang", + "cbrp-teardown-race" + ], + "related_bugs": [ + "bug-1678", + "bug-1452", + "bug-1561" + ], + "occurrences": 1, + "last_seen": "2026-08-25T17:30:00Z" } ] } diff --git a/src/controller/job.c b/src/controller/job.c index 5134dbee..4ec30809 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -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; } diff --git a/src/controller/meson.build b/src/controller/meson.build index f1a76c0f..d11e2cc2 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -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', diff --git a/src/controller/persist.c b/src/controller/persist.c index b4ae59a0..5770ba28 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.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 diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 3b5148e9..1b010e2f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -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); diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index b01a7a07..e8f7d8b7 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -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( diff --git a/src/controller/testsuite/test_wal_job_start_fact.c b/src/controller/testsuite/test_wal_job_start_fact.c new file mode 100644 index 00000000..799b9a86 --- /dev/null +++ b/src/controller/testsuite/test_wal_job_start_fact.c @@ -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 + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#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); +} diff --git a/src/include/persist.h b/src/include/persist.h index 40fe0d01..72b58227 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -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, diff --git a/src/include/wal_types.h b/src/include/wal_types.h index a8503ed2..0515affb 100644 --- a/src/include/wal_types.h +++ b/src/include/wal_types.h @@ -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,