feat: implement job state, queue state, and worker recovery in WAL replay

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

@@ -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;

src/controller/persist.c

@@ -27,11 +27,14 @@
 #include <unistd.h>
 #include <zlib.h>
 
-/* 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;
+        /* Payload: wal_jobstate_t {jobid, state} */
-         * on standby recovery the submit entry already reconstructed the job so
+        if (hdr->payload_len >= sizeof(wal_jobstate_t)) {
-         * these transitions are replayed as no-ops (job state was snapshotted). */
+            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).
+    /* All other job events: payload = wal_jobstate_t {jobid, state}.
-     * The in-memory state is already updated; WAL records the transition. */
+     * State is read from the job_t after the AFTER_EVENT hook updates it. */
-    uint64_t jobid = 0;
+    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;
+    /* args: cluster_t*, queue_t* */
-    return persist_wal_async(p, wal_type, &payload, sizeof(payload));
+    (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,

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;
+}

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                                                         */
 /* ------------------------------------------------------------------ */

src/include/wal_types.h

@@ -6,6 +6,23 @@
 #include <stdint.h>
 #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)
  * ============================================================ */