From 6df3bd5906d3e1cc845b26b0113276a88bf8bd8f Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 8 May 2026 08:19:20 +0000 Subject: [PATCH] feat: implement job state, queue state, and worker recovery in WAL replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WAL payloads: - Job state transitions (START/FINISH/FAIL/STOP/RESUME/UPDATE etc): wal_jobstate_t {uint64_t jobid, uint32_t state} — restores exact state flags - Queue state (QUEUE_CTRL): wal_queuehdr_t {uint32_t status} + name string — re-applies admin open/close/active/inactive after config queues are built - Worker CTRL: existing [uint32_t detail][capnp xce_worker_t] unchanged New recovery functions: - restore_job_state(c, jobid, state): looks up job, overwrites state flags - restore_queue_state(c, name, status): looks up queue, overwrites status apply_wal_entry: dispatches all WAL types including job state and queue test_repl_apply.c: weak stubs for restore_job_state/restore_queue_state so unit test binary links without job.c/queue.c --- src/controller/job.c | 18 ++++++ src/controller/persist.c | 67 +++++++++++++++------- src/controller/queue.c | 22 +++++++ src/controller/testsuite/test_repl_apply.c | 14 +++++ src/include/wal_types.h | 17 ++++++ 5 files changed, 117 insertions(+), 21 deletions(-) diff --git a/src/controller/job.c b/src/controller/job.c index d9082ea7..cccd6dfc 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -102,6 +102,24 @@ int restore_job(cluster_t* c, uint64_t jobid_unused, return 0; } +/* Apply a recorded job state transition during WAL recovery. + * Looks up the job (which must already exist from WAL_JOB_SUBMIT replay) + * and overwrites its state with the value stored in the WAL entry. */ +int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state) +{ + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)jobid, (void**)&j); + unlock(c->jobmgr); + if (!j) + return -ENOENT; + + lock(j, RTTI_LOCK_WRITE); + j->job.state = (int)state; + unlock(j); + return 0; +} + static int get_nslots_in_alloc(xce_host_alloc_t* alloc) { int i; diff --git a/src/controller/persist.c b/src/controller/persist.c index 751af801..85f50615 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -27,11 +27,14 @@ #include #include -/* restore_job / worker_from_capnp are defined in job.c / worker.c */ +/* restore_* functions defined in job.c / worker.c / queue.c */ extern int restore_job(cluster_t* c, uint64_t jobid, const void* capnp_payload, size_t len); +extern int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state); extern int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len); +extern int restore_queue_state(cluster_t* c, + const char* name, uint32_t status); /* ---- Cap'n Proto serialization helpers (WAL payload encoding) ---------- */ @@ -643,7 +646,7 @@ int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payloa switch (hdr->type) { case WAL_JOB_SUBMIT: - /* Payload: Cap'n Proto serialized xce_job_t (jobid embedded in struct) */ + /* Payload: Cap'n Proto serialized xce_job_t */ restore_job(p->cluster, 0, payload, hdr->payload_len); break; @@ -656,19 +659,26 @@ int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payloa case WAL_JOB_CKPT: case WAL_JOB_PREASON: case WAL_JOB_RUSAGE: - /* Payload: uint64_t jobid — state already reflected in memory on primary; - * on standby recovery the submit entry already reconstructed the job so - * these transitions are replayed as no-ops (job state was snapshotted). */ + /* Payload: wal_jobstate_t {jobid, state} */ + if (hdr->payload_len >= sizeof(wal_jobstate_t)) { + const wal_jobstate_t* js = (const wal_jobstate_t*)payload; + restore_job_state(p->cluster, js->jobid, js->state); + } break; case WAL_WORKER_CTRL: - /* Payload: [uint32_t detail][capnp worker OR name string] */ + /* Payload: [uint32_t detail][capnp worker] */ if (hdr->payload_len > WORKER_WAL_HDR_SIZE) worker_from_capnp(p->cluster, payload, hdr->payload_len); break; case WAL_QUEUE_CTRL: - /* Queue state changes — no entity reconstruction needed for now */ + /* Payload: [wal_queuehdr_t][name NUL-terminated] */ + if (hdr->payload_len > sizeof(wal_queuehdr_t)) { + const wal_queuehdr_t* qh = (const wal_queuehdr_t*)payload; + const char* name = (const char*)(payload + sizeof(wal_queuehdr_t)); + restore_queue_state(p->cluster, name, qh->status); + } break; default: @@ -1011,44 +1021,44 @@ int persist_job_hook_fun(int type, int situation, int detail, return rc; } - /* All other job events: payload is just the jobid (uint64_t). - * The in-memory state is already updated; WAL records the transition. */ - uint64_t jobid = 0; + /* All other job events: payload = wal_jobstate_t {jobid, state}. + * State is read from the job_t after the AFTER_EVENT hook updates it. */ + wal_jobstate_t js = {0}; switch (detail) { case HOOK_DETAIL_JOB_START: { /* args: cluster_t*, job_pidinfo_t*, job_t* */ (void)va_arg(ap, cluster_t*); - (void)va_arg(ap, void*); /* job_pidinfo_t* */ + (void)va_arg(ap, void*); job_t* j = va_arg(ap, job_t*); - jobid = j->job.jobid; + js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state; break; } case HOOK_DETAIL_JOB_EXIT: { /* args: cluster_t*, xce_jobfinish_t*, job_t* */ (void)va_arg(ap, cluster_t*); - (void)va_arg(ap, void*); /* xce_jobfinish_t* */ + (void)va_arg(ap, void*); job_t* j = va_arg(ap, job_t*); - jobid = j->job.jobid; + js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state; break; } case HOOK_DETAIL_JOB_FAIL: { /* args: cluster_t*, job_failinfo_t*, job_t* */ (void)va_arg(ap, cluster_t*); - (void)va_arg(ap, void*); /* job_failinfo_t* */ + (void)va_arg(ap, void*); job_t* j = va_arg(ap, job_t*); - jobid = j->job.jobid; + js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state; break; } default: { /* STOP, RESUME, UPDATE, CKPT, PREASON, RUSAGE — args: cluster_t*, job_t* */ (void)va_arg(ap, cluster_t*); job_t* j = va_arg(ap, job_t*); - jobid = j->job.jobid; + js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state; break; } } - return persist_wal_async(p, wal_type, &jobid, sizeof(jobid)); + return persist_wal_async(p, wal_type, &js, sizeof(js)); } int persist_queue_hook_fun(int type, int situation, int detail, @@ -1056,15 +1066,30 @@ int persist_queue_hook_fun(int type, int situation, int detail, { (void)type; (void)situation; - (void)ap; persist_t* p = cls; uint32_t wal_type = detail_to_wal_type(detail); if (wal_type == 0) return 0; - uint32_t payload = (uint32_t)detail; - return persist_wal_async(p, wal_type, &payload, sizeof(payload)); + /* args: cluster_t*, queue_t* */ + (void)va_arg(ap, cluster_t*); + queue_t* q = va_arg(ap, queue_t*); + + /* Payload: [wal_queuehdr_t][queue name NUL-terminated] */ + const char* name = q->queue.queue ? q->queue.queue : ""; + size_t name_len = strlen(name) + 1; + size_t total = sizeof(wal_queuehdr_t) + name_len; + uint8_t* pkt = malloc(total); + if (!pkt) return -ENOMEM; + + wal_queuehdr_t hdr = { .status = q->queue.status }; + memcpy(pkt, &hdr, sizeof(hdr)); + memcpy(pkt + sizeof(hdr), name, name_len); + + int rc = persist_wal_async(p, wal_type, pkt, (uint32_t)total); + free(pkt); + return rc; } int persist_worker_hook_fun(int type, int situation, int detail, diff --git a/src/controller/queue.c b/src/controller/queue.c index 77953cfa..37172f4f 100644 --- a/src/controller/queue.c +++ b/src/controller/queue.c @@ -7,6 +7,7 @@ #include "xce/cluster_def.h" #include "xce/types.h" #include "xstring.h" +#include "wal_types.h" void queue_fini(queue_t* q) { @@ -178,3 +179,24 @@ int get_queue_exec(cluster_t* c, char* queue, char** pre_exec, char** post_exec) return 0; } + +/* Restore queue status flags from a WAL_QUEUE_CTRL entry during recovery. + * Queues are built from config at startup; this re-applies admin state changes + * (open/close/active/inactive) that were recorded in the WAL. */ +int restore_queue_state(cluster_t* c, const char* name, uint32_t status) +{ + if (!name || name[0] == '\0') + return -EINVAL; + + queue_t* q = NULL; + lock(c->qmgr, RTTI_LOCK_READ); + key_find(c->qmgr, (void*)name, (void**)&q); + unlock(c->qmgr); + if (!q) + return -ENOENT; /* queue not yet built from config — skip */ + + lock(q, RTTI_LOCK_WRITE); + q->queue.status = status; + unlock(q); + return 0; +} diff --git a/src/controller/testsuite/test_repl_apply.c b/src/controller/testsuite/test_repl_apply.c index 69899d34..aaa0047b 100644 --- a/src/controller/testsuite/test_repl_apply.c +++ b/src/controller/testsuite/test_repl_apply.c @@ -41,6 +41,20 @@ int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len) return 0; } +__attribute__((weak)) +int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state) +{ + (void)c; (void)jobid; (void)state; + return 0; +} + +__attribute__((weak)) +int restore_queue_state(cluster_t* c, const char* name, uint32_t status) +{ + (void)c; (void)name; (void)status; + return 0; +} + /* ------------------------------------------------------------------ */ /* Test helpers */ /* ------------------------------------------------------------------ */ diff --git a/src/include/wal_types.h b/src/include/wal_types.h index 2597f708..c4d89e9b 100644 --- a/src/include/wal_types.h +++ b/src/include/wal_types.h @@ -6,6 +6,23 @@ #include #include "config.h" +/* ── WAL payload helper structs ──────────────────────────────────────────── + * These are the binary formats embedded after wal_entry_hdr_t for each type. + * All fields are native-endian (single-node WAL; replication uses capnp). */ + +/* WAL_JOB_START / FINISH / FAIL / STOP / RESUME / UPDATE / CKPT / PREASON / RUSAGE */ +typedef struct { + uint64_t jobid; + uint32_t state; /* xce_job state flags after this transition */ + uint32_t _pad; +} wal_jobstate_t; + +/* WAL_QUEUE_CTRL — queue name is a NUL-terminated string immediately after + * this header. Total payload = sizeof(wal_queuehdr_t) + strlen(name)+1. */ +typedef struct { + uint32_t status; /* xce_queue status flags after this transition */ +} wal_queuehdr_t; + /* ============================================================ * WAL entry header — 40 bytes, 8-byte aligned (Part 2.2) * ============================================================ */