diff --git a/src/controller/job.c b/src/controller/job.c index b52bc64..325ce42 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -1,9 +1,11 @@ #include #include #include +#include #include #include "cluster.h" +#include "controller.capnp.h" #include "jobj.h" #include "persist.h" #include "preason.h" @@ -63,15 +65,40 @@ int load_jobs(void* c) return 0; } -/* P0 stub — full deserialization wired in Task 15 (cluster wiring). - * Inserts a job at its original jobid without allocating a new one. - * Returns -EEXIST if jobid already present (safety guard for replay). */ -int restore_job(cluster_t* c, uint64_t jobid, const void* capnp_payload, size_t len) +/* Reconstruct a job from a WAL_JOB_SUBMIT Cap'n Proto payload. + * Uses the jobid embedded in the serialized xce_job_t (not the caller's arg). + * Returns -EEXIST if the job is already present (idempotent replay guard). */ +int restore_job(cluster_t* c, uint64_t jobid_unused, + const void* capnp_payload, size_t len) { - (void)c; - (void)jobid; - (void)capnp_payload; - (void)len; + (void)jobid_unused; + + if (!capnp_payload || len == 0) + return -EINVAL; + + struct capn c_ctx; + capn_init_mem(&c_ctx, (const uint8_t*)capnp_payload, len, 0); + Job_ptr jp; + jp.p = capn_getp(capn_root(&c_ctx), 0, 1); + struct Job cj; + xce_job_t j; + memset(&j, 0, sizeof(j)); + read_Job(&cj, jp); + decode_Job(&j, &cj); + capn_free(&c_ctx); + + /* Safety guard: skip if already present (WAL replay may see duplicates + * when recovery restarts from a segment boundary). */ + job_t* existing = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + int found = key_find(c->jobmgr, (void*)j.jobid, (void**)&existing); + unlock(c->jobmgr); + if (found > 0 && existing != NULL) + return -EEXIST; + + job_t* job = create_job(c, &j); + if (!job) + return -errno; return 0; } diff --git a/src/controller/join.c b/src/controller/join.c index ffd9528..2db110e 100644 --- a/src/controller/join.c +++ b/src/controller/join.c @@ -64,8 +64,12 @@ static int handle_worker_join(cluster_t* c, endpoint_t* ep, reply.status = idx; reply.data_which = RequestReply_data_nodata; + /* Look up the newly added worker so the persist hook can serialize it */ + worker_t* w = NULL; + if (idx >= 0) + vget(c->workermgr, idx, (void**)&w); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_WORKER, HOOK_AFTER_EVENT, - HOOK_DETAIL_WORKER_JOIN, c, from->name, idx); + HOOK_DETAIL_WORKER_JOIN, c, w); xce_reply(ep, c->myhost, &reply); unlock(c->workermgr); if (idx < 0) { diff --git a/src/controller/persist.c b/src/controller/persist.c index 1fd6b54..751af80 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -5,6 +5,9 @@ #include "hook.h" #include "xce/cluster_def.h" #include "xce/types.h" +#include "controller.capnp.h" +#include "xlist.h" +#include "cluster.h" #include #include @@ -24,14 +27,64 @@ #include #include -/* Forward declarations for P0 restore stubs (defined in job.c / worker.c). - * Avoids pulling in the heavy cluster.h / worker.h header chains here. - * Full Cap'n Proto wiring happens in P2-4+. */ +/* restore_job / worker_from_capnp are defined in job.c / worker.c */ extern int restore_job(cluster_t* c, uint64_t jobid, const void* capnp_payload, size_t len); extern int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len); +/* ---- Cap'n Proto serialization helpers (WAL payload encoding) ---------- */ + +/* Serialize xce_job_t to a malloc'd Cap'n Proto buffer. + * Caller must free(*out) on success. */ +static int capnp_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; +} + +/* Serialize xce_worker_t to a malloc'd Cap'n Proto buffer. */ +static int capnp_encode_worker(const xce_worker_t* w, uint8_t** out, size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct Worker cw; + Worker_ptr wp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_Worker(cs, &cw, (xce_worker_t*)w); + wp = new_Worker(cs); + write_Worker(&cw, wp); + capn_setp(capn_root(&c), 0, wp.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; +} + +/* WAL_WORKER_CTRL payload: [uint32_t detail][capnp bytes...] or [uint32_t detail][name\0] */ +#define WORKER_WAL_HDR_SIZE sizeof(uint32_t) + /* ------------------------------------------------------------------ */ /* Constants */ /* ------------------------------------------------------------------ */ @@ -585,23 +638,40 @@ int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payloa p->last_crc = hdr->crc; pthread_mutex_unlock(&p->lock); - /* Entity-creating entries: delegate to existing restore stubs. - * Gate on p->cluster != NULL so unit tests without a full cluster work. */ + if (!p->cluster) + return 0; /* unit tests without a full cluster — skip entity restore */ + switch (hdr->type) { case WAL_JOB_SUBMIT: - /* restore_job() is a P0 stub — full Cap'n Proto decode in P2-4+. */ - if (p->cluster) - restore_job(p->cluster, 0 /* jobid decoded from payload in P2-4 */, - payload, hdr->payload_len); + /* Payload: Cap'n Proto serialized xce_job_t (jobid embedded in struct) */ + restore_job(p->cluster, 0, payload, hdr->payload_len); break; + + case WAL_JOB_START: + case WAL_JOB_FINISH: + case WAL_JOB_FAIL: + case WAL_JOB_STOP: + case WAL_JOB_RESUME: + case WAL_JOB_UPDATE: + 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). */ + break; + case WAL_WORKER_CTRL: - /* worker_from_capnp() is a P0 stub — full decode in P2-4+. */ - if (p->cluster) + /* Payload: [uint32_t detail][capnp worker OR name string] */ + 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 */ + break; + default: - /* All other known types: LSN/CRC already advanced above. - * Full per-type handling wired in P2-4 (Cap'n Proto decode). */ break; } @@ -914,22 +984,71 @@ int persist_job_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; /* unknown detail — skip */ - - uint32_t payload = (uint32_t)detail; + return 0; if (wal_type == WAL_JOB_SUBMIT) { - /* Job submission uses sync replication path */ - int sync_rc = 0; - persist_wal_sync_append(p, wal_type, &payload, sizeof(payload), &sync_rc); - return sync_rc; + /* AFTER_EVENT hook args: cluster_t* c, xlist_t* jlist */ + (void)va_arg(ap, cluster_t*); + xlist_t* jlist = va_arg(ap, xlist_t*); + int rc = 0; + for (int i = 0; i < jlist->count && rc == 0; i++) { + job_t* j = (job_t*)xlist_get(jlist, i); + uint8_t* buf = NULL; + size_t sz = 0; + if (capnp_encode_job(&j->job, &buf, &sz) != 0) { + rc = -ENOMEM; + break; + } + int sync_rc = 0; + persist_wal_sync_append(p, WAL_JOB_SUBMIT, buf, (uint32_t)sz, &sync_rc); + free(buf); + if (sync_rc != 0) rc = sync_rc; + } + return rc; } - return persist_wal_async(p, wal_type, &payload, sizeof(payload)); + + /* 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; + 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* */ + job_t* j = va_arg(ap, job_t*); + jobid = j->job.jobid; + 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* */ + job_t* j = va_arg(ap, job_t*); + jobid = j->job.jobid; + 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* */ + job_t* j = va_arg(ap, job_t*); + jobid = j->job.jobid; + 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; + break; + } + } + + return persist_wal_async(p, wal_type, &jobid, sizeof(jobid)); } int persist_queue_hook_fun(int type, int situation, int detail, @@ -953,15 +1072,38 @@ int persist_worker_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)); + /* WAL_WORKER_CTRL payload layout: [uint32_t detail][capnp worker OR name] */ + uint32_t disc = (uint32_t)detail; + uint8_t* buf = NULL; + size_t capnp_sz = 0; + int rc = 0; + + /* All worker detail types — args: cluster_t*, worker_t*. + * Caller holds the appropriate lock (workermgr write lock for JOIN, + * worker write lock for DEAD/UPDATE), so no extra lock needed here. */ + (void)va_arg(ap, cluster_t*); + worker_t* w = va_arg(ap, worker_t*); + if (!w) return 0; /* guard: failed join produced no worker slot */ + rc = capnp_encode_worker(&w->host, &buf, &capnp_sz); + if (rc != 0) return rc; + + /* Build [disc | capnp bytes] in one allocation */ + size_t total = WORKER_WAL_HDR_SIZE + capnp_sz; + uint8_t* pkt = malloc(total); + if (!pkt) { free(buf); return -ENOMEM; } + memcpy(pkt, &disc, WORKER_WAL_HDR_SIZE); + memcpy(pkt + WORKER_WAL_HDR_SIZE, buf, capnp_sz); + free(buf); + + rc = persist_wal_async(p, wal_type, pkt, (uint32_t)total); + free(pkt); + return rc; } /* ============================================================ diff --git a/src/controller/worker.c b/src/controller/worker.c index eb90707..431fd8e 100644 --- a/src/controller/worker.c +++ b/src/controller/worker.c @@ -1,9 +1,11 @@ #include #include #include +#include #include #include "cluster.h" +#include "controller.capnp.h" #include "persist.h" #include "wobj.h" #include "worker.h" @@ -652,16 +654,38 @@ int do_hostcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request) return rc; } -/* P0 stub — bridges Cap'n Proto deserialization to restore_worker(). - * Full implementation in Task 15 (cluster wiring): deserializes the - * Cap'n Proto Worker object, deep-copies pointer fields, and calls - * restore_worker(cls, host). */ +/* Reconstruct a worker from a WAL_WORKER_CTRL Cap'n Proto payload. + * Payload layout: [uint32_t detail][Cap'n Proto xce_worker_t]. + * Returns -EEXIST if the worker is already registered (idempotent replay). */ int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len) { - (void)c; - (void)capnp_payload; - (void)len; - return 0; + if (!capnp_payload || len <= sizeof(uint32_t)) + return -EINVAL; + + /* Skip the 4-byte detail discriminator written by persist_worker_hook_fun */ + const uint8_t* data = (const uint8_t*)capnp_payload + sizeof(uint32_t); + size_t data_len = len - sizeof(uint32_t); + + struct capn c_ctx; + capn_init_mem(&c_ctx, data, data_len, 0); + Worker_ptr wp; + wp.p = capn_getp(capn_root(&c_ctx), 0, 1); + struct Worker cw; + xce_worker_t host; + memset(&host, 0, sizeof(host)); + read_Worker(&cw, wp); + decode_Worker(&host, &cw); + capn_free(&c_ctx); + + /* Idempotency guard: skip if already present */ + worker_t* existing = NULL; + lock(c->workermgr, RTTI_LOCK_READ); + int found = key_find(c->workermgr, (void*)host.name, (void**)&existing); + unlock(c->workermgr); + if (found > 0 && existing != NULL) + return -EEXIST; + + return restore_worker(c, host); } /* worker_should_retain lives in persist.c —