wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(raft-apply-decoupling): build apply_committed_entry single dispatch seam (task 2.2)

Browse source 
Add apply_committed_entry(persist_t*, wal_type, lsn, payload, len) in
persist.c/persist.h as the single seam that turns a committed WAL
entry into an in-memory mutation plus its post-mutation notification
hook, covering job submit, job-state facts, queue control, and
worker control. Job-state facts route through 2.1's guard-only
apply_job_update_transition() core and fire a generic
HOOK_DETAIL_JOB_UPDATE, not per-type hooks — the wire format
(wal_jobstate_t {jobid, state}) cannot supply what the type-specific
cores/hooks require. This is a real, documented deviation from
design.md's per-type dispatch wording; tasks.md's Implementation
Notes flags it as unresolved and blocking for whoever wires this
into a live path (4.1/6.2/6.3).

Not yet wired into any live call site (replay wiring is 2.3).
This commit is contained in:
Rongsong Shen 2026-07-03 16:51:47 +00:00
parent 41d36204d5
commit 8c00f20cec
6 changed files with 689 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/controller/meson.build
View file

@ -158,6 +158,7 @@ controller_unit_test_src = [
'testsuite/snapdiff.c',
'testsuite/test_persist_snapshot_entities.c',
'testsuite/test_apply_seam.c',
'testsuite/test_apply_committed_entry.c',
'ctrlstatus.c',
'commit.c',
'job.c',

180
src/controller/persist.c
View file

@ -2578,6 +2578,186 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
return 0;
}
/* raft-apply-decoupling task 2.2 — see persist.h for the full contract doc
* (routing table, the job-state-family hook-mapping deviation, and what this
* function deliberately does NOT own). Not yet wired into any live call
* site — replay (2.3) and the live apply pump (6.1) will call this later. */
int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
const void* payload, uint32_t len)
{
if (!p)
return -EINVAL;
if (!p->cluster)
return 0; /* unit tests without a full cluster — nothing to dispatch into */
cluster_t* c = p->cluster;
switch (wal_type) {
case WAL_JOB_SUBMIT: {
int rc = restore_job(c, 0, payload, len, lsn);
if (rc == -EEXIST)
return 0; /* idempotent re-apply (replay / resync) — no hook */
if (rc < 0)
return rc;
/* Recover the jobid via a cheap raw capnp field read (read_Job only
* reads the fixed uint64 field — no allocation, unlike decode_Job)
* so the just-constructed job can be looked up for the notification
* hook without changing restore_job's own contract. */
struct capn jc_ctx;
capn_init_mem(&jc_ctx, (const uint8_t*)payload, len, 0);
Job_ptr jp;
jp.p = capn_getp(capn_root(&jc_ctx), 0, 1);
struct Job cj;
read_Job(&cj, jp);
uint64_t jobid = cj.jobid;
capn_free(&jc_ctx);
job_t* j = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
unlock(c->jobmgr);
if (!j)
return 0; /* defensive: should not happen after a successful restore */
void* jptr = j;
xlist_t jlist = { .count = 1, .caps = 1, .delta = 1, .data = &jptr };
xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_NEW, c, &jlist);
return 0;
}
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_RUSAGE:
case WAL_JOB_CKPT:
case WAL_JOB_PREASON: {
/* Every job-state fact type shares one wire shape — wal_jobstate_t
* {jobid, state} — and therefore one guarded core. See persist.h's
* doc comment on apply_committed_entry for why this does not
* dispatch to the type-specific apply_job_{start,finish,fail}_
* transition() cores (their richer parameter structs are not
* present on this wire format). */
if (len < sizeof(wal_jobstate_t))
return 0; /* undersized payload — skip, mirrors apply_wal_entry */
const wal_jobstate_t* js = (const wal_jobstate_t*)payload;
job_t* j = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)js->jobid, (void**)&j);
unlock(c->jobmgr);
if (!j)
return -ENOENT;
lock(j, RTTI_LOCK_WRITE);
job_transition_result_t result =
apply_job_update_transition(j, (int)js->state);
unlock(j);
if (result != JOB_TRANSITION_APPLIED)
return 0; /* rejected-backward / no-op-duplicate — no hook */
xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, j);
return 0;
}
case WAL_QUEUE_CTRL: {
if (len <= sizeof(wal_queuehdr_t))
return 0; /* undersized payload — skip, mirrors apply_wal_entry */
const wal_queuehdr_t* qh = (const wal_queuehdr_t*)payload;
const char* name =
(const char*)((const uint8_t*)payload + sizeof(wal_queuehdr_t));
/* restore_queue_state's -ENOENT ("queue not yet built from config")
* and -EINVAL (bad name) are both benign skips here, matching
* apply_wal_entry's existing behavior of never fencing on this
* entry type's restore outcome. */
if (restore_queue_state(c, name, qh->status) < 0)
return 0;
queue_t* q = NULL;
lock(c->qmgr, RTTI_LOCK_READ);
key_find(c->qmgr, (void*)name, (void**)&q);
unlock(c->qmgr);
if (!q)
return 0;
xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE, HOOK_AFTER_EVENT,
HOOK_DETAIL_QUEUE_UPDATE, c, q);
return 0;
}
case WAL_WORKER_CTRL: {
if (len <= WORKER_WAL_HDR_SIZE)
return 0; /* undersized payload — skip, mirrors apply_wal_entry */
uint32_t hook_detail;
memcpy(&hook_detail, payload, sizeof(hook_detail));
int rc = worker_from_capnp(c, payload, len);
if (rc == -EEXIST)
return 0; /* idempotent re-apply (replay / resync) — no hook */
if (rc < 0)
return rc;
/* Recover the worker's name via a cheap raw capnp field read
* (read_Worker's capn_get_text is a borrowed, NUL-terminated pointer
* into the decode buffer — no allocation, unlike decode_Worker) so
* the just-restored worker can be looked up for the notification
* hook without changing worker_from_capnp's own contract. */
const uint8_t* wpayload = (const uint8_t*)payload + WORKER_WAL_HDR_SIZE;
size_t wlen = len - WORKER_WAL_HDR_SIZE;
struct capn wc_ctx;
capn_init_mem(&wc_ctx, wpayload, wlen, 0);
Worker_ptr wp;
wp.p = capn_getp(capn_root(&wc_ctx), 0, 1);
struct Worker cw;
read_Worker(&cw, wp);
worker_t* w = NULL;
lock(c->workermgr, RTTI_LOCK_READ);
key_find(c->workermgr, (void*)cw.name.str, (void**)&w);
unlock(c->workermgr);
capn_free(&wc_ctx);
if (!w)
return 0; /* defensive: should not happen after a successful restore */
switch (hook_detail) {
case HOOK_DETAIL_WORKER_JOIN:
xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER,
HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_JOIN, c,
w->host.name, w);
break;
case HOOK_DETAIL_WORKER_UPDATE:
xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER,
HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_UPDATE, c, w);
break;
case HOOK_DETAIL_WORKER_DEAD:
xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER,
HOOK_AFTER_EVENT, HOOK_DETAIL_WORKER_DEAD, c, w);
break;
default:
break; /* unrecognized discriminator — mutation already applied,
* notification skipped rather than guessed at */
}
return 0;
}
default:
/* WAL_NOOP / WAL_CONFIG_MARKER / WAL_CHECKPOINT / unknown — no
* entity effect at this seam, no hook (unchanged from
* apply_wal_entry's existing treatment of these types). */
return 0;
}
}
/* Read WAL segment files in LSN order; replay entries with lsn > from_lsn.
*
* Segment filename format: seg-<start_lsn_hex16>.wal

2
src/controller/testsuite/controller_test.c
View file

@ -78,6 +78,7 @@ extern void test_ha_core_safety(void);
extern void test_ha_partition(void);
extern void test_persist_snapshot_entities(void);
extern void test_apply_seam(void);
extern void test_apply_committed_entry(void);
void setUp(void) { return; }
@ -225,6 +226,7 @@ int main(int argc, char* argv[])
TEST(test_ha_partition);
TEST(test_persist_snapshot_entities);
TEST(test_apply_seam);
TEST(test_apply_committed_entry);
return UNITY_END();
}

452
src/controller/testsuite/test_apply_committed_entry.c Normal file
View file

@ -0,0 +1,452 @@
/* test_apply_committed_entry.c — raft-apply-decoupling task 2.2
*
* Proves apply_committed_entry() (persist.c) is the single dispatch seam
* that turns ONE committed WAL entry into an in-memory mutation for every
* classified entry type, and fires that entry type's post-mutation
* notification hook exactly once, strictly after the mutation completes:
*
* - WAL_JOB_SUBMIT -> restore_job() + HOOK_DETAIL_JOB_NEW
* - WAL_JOB_START/... -> apply_job_update_transition() (2.1's guarded
* core) + HOOK_DETAIL_JOB_UPDATE
* - WAL_QUEUE_CTRL -> restore_queue_state() + HOOK_DETAIL_QUEUE_UPDATE
* - WAL_WORKER_CTRL -> worker_from_capnp() + HOOK_DETAIL_WORKER_JOIN
*
* and that a guard-rejected or idempotent-duplicate application performs no
* mutation and fires NO notification (Req 7.1/7.2).
*
* This task does NOT wire apply_committed_entry into any live call site
* (replay / apply pump are later tasks 2.3/6.1) — every test here calls the
* dispatch function directly.
*
* Fixture: a minimal cluster_t (hooksys + jobmgr/qmgr/workermgr) and a bare
* persist_t whose ONLY field apply_committed_entry reads is ->cluster — no
* real WAL/disk/repl/commit-engine setup needed (mirrors test_apply_seam.c's
* "bare fixture, no more than the function under test needs" philosophy).
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <stdarg.h>
#include <stdbool.h>
#include <string.h>
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "hook.h"
#include "jobj.h"
#include "persist.h"
#include "qobj.h"
#include "wal_types.h"
#include "wobj.h"
#include "xce/cluster_def.h"
#include "xce/types.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 test_apply_seam.c. */
static int g_tace_jobobj_ready;
static cluster_t g_tace_c;
static persist_t g_tace_p;
/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */
static int g_tace_hook_fires;
static int tace_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_tace_hook_fires++;
return 0;
}
static void tace_setup(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32,
.delta = 16 };
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 32,
.delta = 16 };
if (makeString((char*)"tace-probe") == NULL)
xstring_table_init(64);
if (!g_tace_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_tace_jobobj_ready = 1;
}
memset(&g_tace_c, 0, sizeof(g_tace_c));
g_tace_c.myhost = &g_tace_c.myhost_info;
g_tace_c.myhost_info.name = (char*)"ctrl-test";
g_tace_c.myhost_info.key = (char*)"";
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_tace_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_tace_c.qmgr = makeInstance(&xkvmgr_type, &str_params);
g_tace_c.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_tace_c.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_tace_c.jobmgr);
TEST_ASSERT_NOT_NULL(g_tace_c.qmgr);
TEST_ASSERT_NOT_NULL(g_tace_c.workermgr);
TEST_ASSERT_NOT_NULL(g_tace_c.resmgr);
zarray_init(&g_tace_c.sharedinst, 16, NULL);
memset(&g_tace_p, 0, sizeof(g_tace_p));
g_tace_p.cluster = &g_tace_c;
g_tace_hook_fires = 0;
}
static void tace_teardown(void)
{
zarray_fini(&g_tace_c.sharedinst);
if (g_tace_c.resmgr) dereference(g_tace_c.resmgr);
if (g_tace_c.workermgr) dereference(g_tace_c.workermgr);
if (g_tace_c.qmgr) dereference(g_tace_c.qmgr);
if (g_tace_c.jobmgr) dereference(g_tace_c.jobmgr);
xce_hook_fini(&g_tace_c.hooksys);
}
/* ── Encode helpers (mirror the static capnp_encode_{job,worker} in
* persist.c — those are file-static, so tests keep local copies, exactly
* as test_egress_filter.c's ef_encode_job already does). ────────────── */
static int tace_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;
}
static int tace_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;
}
/* Look up a job_t wrapper by jobid (NULL if absent). */
static job_t* tace_find_job(uint64_t jobid)
{
job_t* j = NULL;
lock(g_tace_c.jobmgr, RTTI_LOCK_READ);
int found = key_find(g_tace_c.jobmgr, (void*)jobid, (void**)&j);
unlock(g_tace_c.jobmgr);
return (found > 0) ? j : NULL;
}
/* Creates a job directly in the fixture's jobmgr, bypassing
* handle_jobsubmit/restore_job entirely — for seeding fact-transition tests. */
static job_submit_data_t g_tace_seed_submit;
static job_t* tace_seed_job(uint64_t jobid, uint32_t state)
{
xce_job_t jdef;
memset(&g_tace_seed_submit, 0, sizeof(g_tace_seed_submit));
g_tace_seed_submit.queue = (char*)"q";
g_tace_seed_submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = state;
jdef.submit = &g_tace_seed_submit;
return makeJob(g_tace_c.jobmgr, &jdef);
}
/* ── WAL_JOB_SUBMIT ───────────────────────────────────────────────────── */
static void test_ace_job_submit_constructs_and_notifies(void)
{
tace_setup();
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 = 501;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0, tace_encode_job(&jdef, &payload, &plen));
TEST_ASSERT_EQUAL_INT(0,
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB,
tace_spy_hook, NULL));
int rc = apply_committed_entry(&g_tace_p, WAL_JOB_SUBMIT, 10, payload,
(uint32_t)plen);
TEST_ASSERT_EQUAL_INT(0, rc);
job_t* j = tace_find_job(501);
TEST_ASSERT_NOT_NULL_MESSAGE(j,
"WAL_JOB_SUBMIT must construct the job via restore_job");
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"a fresh job submit must fire exactly one post-mutation notification");
free(payload);
tace_teardown();
printf("test_ace_job_submit_constructs_and_notifies: PASS\n");
}
static void test_ace_job_submit_duplicate_is_noop_no_extra_notify(void)
{
tace_setup();
xce_job_t jdef;
job_submit_data_t submit;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"q";
submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = 502;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0, tace_encode_job(&jdef, &payload, &plen));
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook,
NULL);
TEST_ASSERT_EQUAL_INT(0, apply_committed_entry(&g_tace_p, WAL_JOB_SUBMIT,
11, payload,
(uint32_t)plen));
TEST_ASSERT_EQUAL_INT(1, g_tace_hook_fires);
/* Re-apply the SAME entry (replay / resync redelivery) — must be a
* no-op: no error, no second job, no second notification (Req 7.1). */
TEST_ASSERT_EQUAL_INT(0, apply_committed_entry(&g_tace_p, WAL_JOB_SUBMIT,
11, payload,
(uint32_t)plen));
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"an idempotent duplicate submit must not fire a second notification");
free(payload);
tace_teardown();
printf("test_ace_job_submit_duplicate_is_noop_no_extra_notify: PASS\n");
}
/* ── WAL_JOB_START / job-state fact family ───────────────────────────── */
static void test_ace_job_state_transition_applies_forward_and_notifies(void)
{
tace_setup();
job_t* j = tace_seed_job(601, JOB_STAT_PEND);
TEST_ASSERT_NOT_NULL(j);
wal_jobstate_t js = { .jobid = 601, .state = JOB_STAT_RUN };
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook,
NULL);
int rc = apply_committed_entry(&g_tace_p, WAL_JOB_START, 20, &js,
sizeof(js));
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
"a forward job-state fact must apply via the 2.1 guarded core");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"a forward job-state transition must fire exactly one notification");
tace_teardown();
printf("test_ace_job_state_transition_applies_forward_and_notifies: "
"PASS\n");
}
static void
test_ace_job_state_transition_rejects_backward_suppresses_notify(void)
{
tace_setup();
/* Job already terminal (as if FINISH already applied); a stale,
* previously-unapplied START must not regress it (Req 7.1/7.2). */
job_t* j = tace_seed_job(602, JOB_STAT_DONE);
TEST_ASSERT_NOT_NULL(j);
wal_jobstate_t js = { .jobid = 602, .state = JOB_STAT_RUN };
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook,
NULL);
int rc = apply_committed_entry(&g_tace_p, WAL_JOB_START, 21, &js,
sizeof(js));
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"a guard-rejected backward transition is a success no-op, not an "
"apply failure");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state,
"a rejected backward transition must preserve the terminal state");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_tace_hook_fires,
"a guard-rejected transition must fire NO notification");
tace_teardown();
printf("test_ace_job_state_transition_rejects_backward_suppresses_"
"notify: PASS\n");
}
/* ── WAL_QUEUE_CTRL ───────────────────────────────────────────────────── */
static void test_ace_queue_ctrl_updates_and_notifies(void)
{
tace_setup();
xce_queue_t qdef;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char*)"q1";
TEST_ASSERT_NOT_NULL(makeQueue(g_tace_c.qmgr, &qdef));
const char* name = "q1";
size_t name_len = strlen(name) + 1;
size_t total = sizeof(wal_queuehdr_t) + name_len;
uint8_t* pkt = malloc(total);
TEST_ASSERT_NOT_NULL(pkt);
wal_queuehdr_t qh = { .status = QUEUE_STAT_CLOSED };
memcpy(pkt, &qh, sizeof(qh));
memcpy(pkt + sizeof(qh), name, name_len);
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_QUEUE, tace_spy_hook,
NULL);
int rc = apply_committed_entry(&g_tace_p, WAL_QUEUE_CTRL, 30, pkt,
(uint32_t)total);
TEST_ASSERT_EQUAL_INT(0, rc);
queue_t* q = NULL;
lock(g_tace_c.qmgr, RTTI_LOCK_READ);
key_find(g_tace_c.qmgr, (void*)name, (void**)&q);
unlock(g_tace_c.qmgr);
TEST_ASSERT_NOT_NULL(q);
TEST_ASSERT_TRUE_MESSAGE((q->queue.status & QUEUE_STAT_CLOSED) != 0,
"WAL_QUEUE_CTRL must apply via restore_queue_state");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"a queue control apply must fire exactly one notification");
free(pkt);
tace_teardown();
printf("test_ace_queue_ctrl_updates_and_notifies: PASS\n");
}
/* ── WAL_WORKER_CTRL ──────────────────────────────────────────────────── */
static void test_ace_worker_ctrl_join_restores_and_notifies(void)
{
tace_setup();
/* copy_worker (wobj.c) rejects a NULL src->attrs with -EINVAL — a real
* WAL_WORKER_CTRL JOIN payload always carries real host attributes, so
* the fixture must too. */
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = (char*)"unknown";
attrs.hostmodel = (char*)"unknown";
xce_worker_t wdef;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char*)"w1";
wdef.attrs = &attrs;
uint8_t* wbuf = NULL;
size_t wsize = 0;
TEST_ASSERT_EQUAL_INT(0, tace_encode_worker(&wdef, &wbuf, &wsize));
uint32_t disc = (uint32_t)HOOK_DETAIL_WORKER_JOIN;
size_t total = sizeof(disc) + wsize;
uint8_t* pkt = malloc(total);
TEST_ASSERT_NOT_NULL(pkt);
memcpy(pkt, &disc, sizeof(disc));
memcpy(pkt + sizeof(disc), wbuf, wsize);
free(wbuf);
xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, tace_spy_hook,
NULL);
int rc = apply_committed_entry(&g_tace_p, WAL_WORKER_CTRL, 40, pkt,
(uint32_t)total);
TEST_ASSERT_EQUAL_INT(0, rc);
worker_t* w = NULL;
lock(g_tace_c.workermgr, RTTI_LOCK_READ);
key_find(g_tace_c.workermgr, (void*)"w1", (void**)&w);
unlock(g_tace_c.workermgr);
TEST_ASSERT_NOT_NULL_MESSAGE(w,
"WAL_WORKER_CTRL must restore the worker via worker_from_capnp");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
"a worker control apply must fire exactly one notification");
free(pkt);
tace_teardown();
printf("test_ace_worker_ctrl_join_restores_and_notifies: PASS\n");
}
/* ── Defensive / boundary ─────────────────────────────────────────────── */
static void test_ace_null_persist_returns_error(void)
{
int rc = apply_committed_entry(NULL, WAL_JOB_UPDATE, 1, NULL, 0);
TEST_ASSERT_NOT_EQUAL(0, rc);
printf("test_ace_null_persist_returns_error: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_apply_committed_entry(void)
{
UNITY_BEGIN();
RUN_TEST(test_ace_job_submit_constructs_and_notifies);
RUN_TEST(test_ace_job_submit_duplicate_is_noop_no_extra_notify);
RUN_TEST(test_ace_job_state_transition_applies_forward_and_notifies);
RUN_TEST(test_ace_job_state_transition_rejects_backward_suppresses_notify);
RUN_TEST(test_ace_queue_ctrl_updates_and_notifies);
RUN_TEST(test_ace_worker_ctrl_join_restores_and_notifies);
RUN_TEST(test_ace_null_persist_returns_error);
UNITY_END();
}

52
src/include/persist.h
View file

@ -739,6 +739,58 @@ int persist_worker_hook_fun(int type, int situation, int detail,
int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
const void* payload, uint32_t payload_len);
/* apply_committed_entry — raft-apply-decoupling task 2.2, design.md
* "Apply Seam". The single dispatch function that mutates in-memory state
* for ONE committed WAL entry, shared (once tasks 2.3/6.1 wire their call
* sites) by restart replay and the live commit-advance apply pump. Routes:
* WAL_JOB_SUBMIT -> restore_job() (existing
* idempotent, jobid-addressable construction primitive; -EEXIST is a
* no-op re-apply, not an error)
* WAL_JOB_{START,FINISH,FAIL,STOP,RESUME,
* UPDATE,RUSAGE,CKPT,PREASON} -> apply_job_update_transition()
* (the guard-only core extracted in task 2.1)
* WAL_QUEUE_CTRL -> restore_queue_state()
* WAL_WORKER_CTRL -> worker_from_capnp()
* and fires the entry type's AFTER_EVENT notification hook exactly once,
* strictly after the mutation completes, and ONLY when a genuine (non-
* duplicate, non-guard-rejected) mutation actually occurred.
*
* Deviation from a literal per-type hook fan-out: the on-disk fact payload
* (wal_jobstate_t {jobid, state}) carries only the job's resulting state, not
* the original job_pidinfo_t / xce_jobfinish_t / job_failinfo_t a type-
* specific mark_job_* hook (HOOK_DETAIL_JOB_START/EXIT/FAIL) expects its
* subscribers to dereference. The real risk is sched.c's sched_job_start_fun/
* sched_job_exit_fun/sched_job_fail_fun, which dereference info->jobid
* directly in a struct initializer — before even their `situation` switch,
* with no NULL guard (apiserver.c's equivalent subscribers are NOT at risk:
* they store the `info` pointer itself into the notify union (e.g.
* `.data = { .pidinfo = info }`) inside `case HOOK_AFTER_EVENT:`, without
* dereferencing any of its fields at this call site — unlike sched.c's
* unconditional `info->jobid` access in a top-level struct initializer).
* Every job-state fact type therefore routes through the
* guard-only apply_job_update_transition() core and fires
* HOOK_DETAIL_JOB_UPDATE, whose subscribers take only (cluster_t*, job_t*) —
* matching what this wire format can actually supply without fabricating
* data. No task in the current plan (2.3, 4.1, 4.2, 6.1, 6.2, 6.3) closes
* this gap — task 4.1 is exclusively about removing state mutation/hook-
* firing from worker-fact receipt-time handlers, not about enriching the
* wire format. Closing it would require either a design.md correction
* (formally accepting this HOOK_DETAIL_JOB_UPDATE substitution as final —
* design.md's Apply Seam section and Fact Ingestion Event Contract
* currently describe per-type hook dispatch this wire format cannot
* support) or a new task to enrich the WAL wire format for job-state facts.
*
* NOT this function's concern (owned by whichever caller drives it — the
* replay segment reader today, the live apply pump added by task 6.1):
* advancing p->current_lsn, oversized-payload rejection, WAL_CHECKPOINT
* skip. See apply_wal_entry's doc comment above for why those stay
* physical-layer-owned.
*
* Returns 0 on success (including a no-op idempotent re-apply or a guard-
* rejected/duplicate transition), negative on a genuine apply failure. */
int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
const void* payload, uint32_t len);
/* Load a snapshot file from path, set p->checkpoint_lsn and p->current_lsn.
* lsn hint is informational — the actual LSN is read from the file header.
*