diff --git a/src/controller/job.c b/src/controller/job.c index 41b3988c..98c052ad 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -785,6 +785,36 @@ int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state) return 0; } +/* bug-1678: WAL_JOB_DISPATCH's own sibling of restore_job_state() above, for + * apply_wal_entry()'s standby catch-up / replicated-apply seam -- takes + * ownership of an optional decoded resource allocation (NULL for a + * state-only WAL_JOB_DISPATCH entry, e.g. one written before this fix) so a + * promoted standby recovers EXEC_HOST the same way apply_committed_entry's + * WAL_JOB_DISPATCH case already does for the primary's own boot replay. */ +int restore_job_dispatch_state(cluster_t* c, uint64_t jobid, uint32_t state, + xce_job_alloc_t* alloc) +{ + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)jobid, (void**)&j); + unlock(c->jobmgr); + if (!j) { + if (alloc != NULL) + free_JobAlloc_ptr(&alloc); + return -ENOENT; + } + + lock(j, RTTI_LOCK_WRITE); + j->job.state = (int)state; + if (alloc != NULL) { + if (j->job.allocs != NULL) + free_JobAlloc_ptr(&(j->job.allocs)); + j->job.allocs = alloc; + } + unlock(j); + return 0; +} + static int get_nslots_in_alloc(xce_host_alloc_t* alloc) { int i; @@ -1362,12 +1392,16 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) j->dispatch_fail_count = 0; /* Durable record of the dispatch-time RUN transition (task 1.2), - * mirroring record_job_start_fact()'s persist_submit_job_fact() - * call shape exactly -- _pad stays 0 (the "legacy: state is - * absolute value" convention, matching every other job-state fact - * committed synchronously here rather than via deferred apply). */ - (void)persist_submit_job_fact(&c->persist, WAL_JOB_DISPATCH, - alloc->jobid, JOB_STAT_RUN); + * now including the resource allocation itself (bug-1678): the + * plain {jobid, state} shape persist_submit_job_fact() writes was + * not enough to reconstruct job.allocs on a restart replay, leaving + * EXEC_HOST permanently blank for a job that was RUNNING before + * shutdown. j->job.allocs is read here, still under this same + * write lock, as the freshly-constructed canonical struct just + * stored above -- not the (now partially moved-from) local + * `alloc`/`allocs` variables. */ + (void)persist_submit_job_dispatch_fact(&c->persist, alloc->jobid, + JOB_STAT_RUN, j->job.allocs); unlock(j); } else { diff --git a/src/controller/persist.c b/src/controller/persist.c index 21bd2e32..6e39c76e 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -34,6 +34,14 @@ extern int restore_job(cluster_t* c, uint64_t jobid, const void* capnp_payload, size_t len, uint64_t entry_lsn); extern int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state); +/* restore_job_dispatch_state defined in job.c (bug-1678): WAL_JOB_DISPATCH's + * own sibling of restore_job_state() above, additionally taking ownership of + * an optional decoded xce_job_alloc_t so the dispatch-time resource + * allocation survives standby catch-up / replicated-apply the same way + * apply_committed_entry's WAL_JOB_DISPATCH case already restores it on the + * 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); extern int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len); extern int restore_queue_state(cluster_t* c, @@ -134,6 +142,62 @@ static int capnp_encode_worker(const xce_worker_t* w, uint8_t** out, size_t* out return 0; } +/* Serialize xce_job_alloc_t to a malloc'd Cap'n Proto buffer (bug-1678): + * lets WAL_JOB_DISPATCH carry the scheduler's resource allocation alongside + * the {jobid, state} header, so a restart replay can reconstruct it instead + * of leaving job.allocs permanently NULL. Mirrors capnp_encode_worker above + * exactly, using the JobAlloc struct (already in controller.capnp as the + * existing Job.allocs field -- no schema change needed here). */ +static int capnp_encode_joballoc(const xce_job_alloc_t* a, uint8_t** out, + size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct JobAlloc cja; + JobAlloc_ptr jap; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_JobAlloc(cs, &cja, (xce_job_alloc_t*)a); + jap = new_JobAlloc(cs); + write_JobAlloc(&cja, jap); + capn_setp(capn_root(&c), 0, jap.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 JobAlloc buffer (bug-1678) into a freshly heap- + * allocated xce_job_alloc_t*, mirroring restore_job()'s Job decode pattern + * (job.c) but returning the struct directly instead of constructing a job. + * Returns NULL on a NULL/empty payload; caller owns the result (release via + * free_JobAlloc_ptr). */ +static xce_job_alloc_t* decode_joballoc(const uint8_t* payload, size_t len) +{ + struct capn c_ctx; + JobAlloc_ptr jap; + struct JobAlloc cja; + xce_job_alloc_t* alloc; + + if (!payload || len == 0) + return NULL; + + capn_init_mem(&c_ctx, payload, len, 0); + jap.p = capn_getp(capn_root(&c_ctx), 0, 1); + memset(&cja, 0, sizeof(cja)); + read_JobAlloc(&cja, jap); + + alloc = calloc(1, sizeof(*alloc)); + if (!alloc) { capn_free(&c_ctx); return NULL; } + decode_JobAlloc(alloc, &cja); + capn_free(&c_ctx); + return alloc; +} + /* 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 @@ -3458,7 +3522,6 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, case WAL_JOB_CKPT: case WAL_JOB_PREASON: case WAL_JOB_RUSAGE: - case WAL_JOB_DISPATCH: /* Payload: wal_jobstate_t {jobid, state} */ if (payload_len >= sizeof(wal_jobstate_t)) { const wal_jobstate_t* js = (const wal_jobstate_t*)payload; @@ -3466,6 +3529,30 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, } 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 + * (persist_submit_job_dispatch_fact's shape) alongside the + * {jobid, state} header, so a promoted standby's catch-up replay + * recovers the dispatch-time resource allocation too, not just the + * state. A payload of exactly sizeof(wal_jobstate_t) (no trailing + * bytes -- any WAL_JOB_DISPATCH entry written before this fix) + * decodes no allocation, 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; + xce_job_alloc_t* alloc = NULL; + if (payload_len > sizeof(wal_jobstate_t)) { + alloc = decode_joballoc( + (const uint8_t*)payload + sizeof(wal_jobstate_t), + payload_len - sizeof(wal_jobstate_t)); + } + restore_job_dispatch_state(p->cluster, js->jobid, js->state, + alloc); + } + break; + } + case WAL_WORKER_CTRL: /* Payload: [uint32_t detail][capnp worker] */ if (payload_len > WORKER_WAL_HDR_SIZE) @@ -3752,6 +3839,24 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, } 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); + } else if (wal_type == WAL_JOB_DISPATCH + && len > sizeof(wal_jobstate_t)) { + /* bug-1678: trailing bytes beyond the header carry the + * scheduler's resource allocation (persist_submit_job_ + * dispatch_fact's payload shape) -- decode and adopt it so + * a restart replay recovers EXEC_HOST, not just STATUS. A + * payload of exactly sizeof(wal_jobstate_t) (no trailing + * bytes -- e.g. any WAL_JOB_DISPATCH entry written before + * this fix) intentionally falls through to the same + * state-only behavior every other type in this group gets. */ + xce_job_alloc_t* decoded = decode_joballoc( + (const uint8_t*)payload + sizeof(wal_jobstate_t), + len - sizeof(wal_jobstate_t)); + if (decoded != NULL) { + if (j->job.allocs != NULL) + free_JobAlloc_ptr(&(j->job.allocs)); + j->job.allocs = decoded; + } } } unlock(j); @@ -6712,6 +6817,44 @@ int persist_submit_job_fact_ex(persist_t* p, uint32_t wal_type, sizeof(js)); } +/* bug-1678: WAL_JOB_DISPATCH's own richer sibling of persist_submit_job_fact() + * -- the ONLY job-state fact type that needs to carry a resource allocation + * alongside {jobid, state}, since it is the sole durable record of the + * dispatch-time PEND->RUN transition (design.md "WAL Dispatch Fact"). Payload + * shape: wal_jobstate_t header immediately followed by the Cap'n Proto- + * encoded JobAlloc bytes -- apply_committed_entry()'s and apply_wal_entry()'s + * WAL_JOB_DISPATCH cases decode exactly this shape, and treat a payload of + * exactly sizeof(wal_jobstate_t) (no trailing bytes) as "state-only, leave + * allocs untouched" for backward compatibility with any WAL_JOB_DISPATCH + * entry written before this fix. */ +int persist_submit_job_dispatch_fact(persist_t* p, uint64_t jobid, int state, + const xce_job_alloc_t* alloc) +{ + wal_jobstate_t js = { .jobid = jobid, .state = (uint32_t)state, + ._pad = 0 }; + uint8_t* allocbuf = NULL; + size_t alloclen = 0; + uint8_t* payload; + size_t total; + int rc; + + rc = capnp_encode_joballoc(alloc, &allocbuf, &alloclen); + if (rc != 0) + return rc; + + total = sizeof(js) + alloclen; + payload = malloc(total); + if (!payload) { free(allocbuf); return -ENOMEM; } + memcpy(payload, &js, sizeof(js)); + memcpy(payload + sizeof(js), allocbuf, alloclen); + free(allocbuf); + + rc = wal_hook_append_classified(p, WAL_JOB_DISPATCH, WAL_MODE_RETAIN, + payload, (uint32_t)total); + free(payload); + return rc; +} + /* P0 payload: just the detail value (4 bytes). * Full Cap'n Proto serialization is wired in Task 15. */ diff --git a/src/controller/testsuite/test_repl_apply.c b/src/controller/testsuite/test_repl_apply.c index db7792b3..29fd3a77 100644 --- a/src/controller/testsuite/test_repl_apply.c +++ b/src/controller/testsuite/test_repl_apply.c @@ -50,6 +50,14 @@ int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state) return 0; } +__attribute__((weak)) +int restore_job_dispatch_state(cluster_t* c, uint64_t jobid, uint32_t state, + xce_job_alloc_t* alloc) +{ + (void)c; (void)jobid; (void)state; (void)alloc; + return 0; +} + __attribute__((weak)) int restore_queue_state(cluster_t* c, const char* name, uint32_t status) { diff --git a/src/include/persist.h b/src/include/persist.h index ec8d7d62..45dcc6d9 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -1408,6 +1408,19 @@ int persist_submit_job_fact(persist_t* p, uint32_t wal_type, uint64_t jobid, * case, but is still written for forward-compatibility/diagnostics). */ int persist_submit_job_fact_ex(persist_t* p, uint32_t wal_type, uint64_t jobid, int state, uint32_t pad); + +/* bug-1678: WAL_JOB_DISPATCH's own sibling of persist_submit_job_fact(), + * additionally encoding the scheduler's resource allocation into the fact's + * payload (header {jobid, state} + trailing Cap'n Proto JobAlloc bytes) so a + * restart/catch-up replay can reconstruct job.allocs instead of leaving it + * permanently NULL -- the WAL_JOB_DISPATCH-only reason persist_submit_job_fact + * itself was never sufficient for this one fact type. `alloc` must be + * non-NULL (the real call site, dispatch_a_job_(), always has one by + * construction). Returns the local-append rc, same contract as + * persist_submit_job_fact(). */ +int persist_submit_job_dispatch_fact(persist_t* p, uint64_t jobid, int state, + const xce_job_alloc_t* alloc); + int persist_worker_hook_fun(int type, int situation, int detail, void* cls, va_list ap);