lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(raft-apply-decoupling): point restart replay at apply_committed_entry seam (task 2.3)

Browse source 
Rewire replay_wal()'s per-entry apply call from apply_wal_entry() to
apply_committed_entry() (task 2.2's shared dispatch seam), so restart
replay and the future live commit-advance pump (task 6.1) both
produce materialized state deterministically from the committed log.
apply_wal_entry() is not retired -- it remains ha.c's live standby
catch-up/replicated-apply dispatch, untouched.

Documents a real hazard: replay now fires AFTER_EVENT hooks for the
full post-checkpoint WAL range on every restart, including entries
already applied+notified in a prior process life. Traced and
confirmed inert today for both hook consumers (scheduler registers
post-boot, apiserver gates notify on a connected endpoint empty at
boot); flagged in tasks.md and in-line at the call site for
re-verification before tasks 4.1/6.1/6.2/6.3 make this path broadly
live.
This commit is contained in:
Rongsong Shen 2026-07-04 00:32:46 +00:00
parent 8c00f20cec
commit 50d016d905
5 changed files with 434 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/controller/meson.build
View file

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

32
src/controller/persist.c
View file

@ -2961,10 +2961,34 @@ static int replay_wal(persist_t* p, uint64_t from_lsn,
continue;
}
/* Logical entity restore. The CRC chain + current_lsn are owned by
* this replay loop (current_lsn advanced below; last_crc at line
* ~1601), not by apply_wal_entry. */
apply_wal_entry(p, hdr.type, hdr.lsn, payload, hdr.payload_len);
/* Logical entity restore via the shared apply seam
* (raft-apply-decoupling task 2.3, design.md "Apply Seam"): the
* SAME apply_committed_entry() dispatch the live commit-advance
* apply pump will use (task 6.1), so a node's materialized state
* is the deterministic result of its committed log regardless of
* whether it got there via restart replay or live apply
* (Req 3.1-3.4). apply_wal_entry() (this replay loop's previous
* per-entry dispatch) is NOT retired by this change — it remains
* the live standby catch-up/replicated-apply dispatch used
* elsewhere (ha.c); it is simply no longer this loop's call. The
* CRC chain + current_lsn remain owned by this replay loop
* (current_lsn advanced below; last_crc at line ~1601) — neither
* dispatch function ever touched them (bug-778).
*
* KNOWN HAZARD (tracked in tasks.md Implementation Notes, task
* 2.3): this replays the FULL post-checkpoint WAL range on every
* restart (last_applied_lsn is not persisted, deliberately
* recomputed here), so apply_committed_entry()'s AFTER_EVENT
* hooks fire again for entries already applied+notified in a
* prior process life. Confirmed inert today: scheduler hooks
* aren't registered until a post-boot network JOIN
* (on_scheduler_join, sched.c), and apiserver hooks — though
* registered before this runs — gate their notify on a
* connected endpoint that is empty at first-boot replay time.
* MUST be re-verified before tasks 4.1/6.1/6.2/6.3 make this
* path broadly live. */
apply_committed_entry(p, hdr.type, hdr.lsn, payload,
hdr.payload_len);
free(payload);
/* Advance current_lsn to one past this entry */

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

@ -79,6 +79,7 @@ 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);
extern void test_replay_apply_seam(void);
void setUp(void) { return; }
@ -227,6 +228,7 @@ int main(int argc, char* argv[])
TEST(test_persist_snapshot_entities);
TEST(test_apply_seam);
TEST(test_apply_committed_entry);
TEST(test_replay_apply_seam);
return UNITY_END();
}

401
src/controller/testsuite/test_replay_apply_seam.c Normal file
View file

@ -0,0 +1,401 @@
/* test_replay_apply_seam.c — raft-apply-decoupling task 2.3
*
* Proves restart replay (persist_recover -> replay_wal) applies every
* classified WAL entry type through the SAME apply_committed_entry() seam
* task 2.2 built (routing job submit through restore_job, job-state facts
* through the 2.1 guarded core, queue control through restore_queue_state,
* worker control through worker_from_capnp) instead of the pre-2.3 replay
* path's direct calls inside apply_wal_entry().
*
* Two things this proves:
*
* 1. Materialized-state parity (Req 3.1/3.2): a log containing one entry
* of every classified type — WAL_JOB_SUBMIT, a job-state fact
* (WAL_JOB_START), WAL_QUEUE_CTRL, WAL_WORKER_CTRL — replayed via
* persist_recover() on a fresh persist_t/cluster_t produces the
* correct end state. For this causally-ordered (submit-before-start),
* non-duplicate log, this is identical to what the pre-task-2.3
* apply_wal_entry() dispatch already produced: neither switch body
* differs in materialized outcome for a forward, single-instance
* apply. This is the "pure refactor, existing behavior preserved"
* evidence for the call-site swap.
*
* 2. The call site genuinely switched, not just "happens to look the
* same": unlike apply_wal_entry() (which never fires a notification
* hook), apply_committed_entry() fires each type's AFTER_EVENT hook
* exactly once, strictly after its mutation. A spy hook counts these
* firings across the whole replay. Before task 2.3 (replay wired to
* apply_wal_entry) this count is 0; after (replay wired to
* apply_committed_entry) it is exactly 4 — one per entry. That
* flip is genuine RED (pre-change) / GREEN (post-change) evidence
* that replay's per-entry apply call site was actually rewired.
*
* Fixture mirrors test_apply_committed_entry.c's cluster_t construction
* (task 2.2), adapted to drive the real physical write path
* (persist_wal_async) and the real recovery path (persist_recover) instead
* of calling the dispatch function directly.
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <inttypes.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.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 / test_apply_committed_entry.c. */
static int g_trr_jobobj_ready;
static cluster_t g_trr_c;
/* ── Spy hook: counts AFTER_EVENT firings across job/queue/worker hooks ─── */
static int g_trr_hook_fires;
static int trr_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_trr_hook_fires++;
return 0;
}
static void trr_build_cluster(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*)"trr-probe") == NULL)
xstring_table_init(64);
if (!g_trr_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_trr_jobobj_ready = 1;
}
memset(&g_trr_c, 0, sizeof(g_trr_c));
g_trr_c.myhost = &g_trr_c.myhost_info;
g_trr_c.myhost_info.name = (char*)"ctrl-test";
g_trr_c.myhost_info.key = (char*)"";
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_trr_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_trr_c.qmgr = makeInstance(&xkvmgr_type, &str_params);
g_trr_c.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_trr_c.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_trr_c.jobmgr);
TEST_ASSERT_NOT_NULL(g_trr_c.qmgr);
TEST_ASSERT_NOT_NULL(g_trr_c.workermgr);
TEST_ASSERT_NOT_NULL(g_trr_c.resmgr);
zarray_init(&g_trr_c.sharedinst, 16, NULL);
/* Queues come from config, not from WAL_QUEUE_CTRL (restore_queue_state
* -ENOENT's "queue not yet built from config" is a benign replay skip)
* — pre-create the queue the replayed WAL_QUEUE_CTRL entry targets,
* exactly as a real boot's config load would before replay runs. */
xce_queue_t qdef;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char*)"rq1";
TEST_ASSERT_NOT_NULL(makeQueue(g_trr_c.qmgr, &qdef));
xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_JOB, trr_spy_hook,
NULL);
xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_QUEUE, trr_spy_hook,
NULL);
xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, trr_spy_hook,
NULL);
g_trr_hook_fires = 0;
}
static void trr_teardown_cluster(void)
{
zarray_fini(&g_trr_c.sharedinst);
if (g_trr_c.resmgr) dereference(g_trr_c.resmgr);
if (g_trr_c.workermgr) dereference(g_trr_c.workermgr);
if (g_trr_c.qmgr) dereference(g_trr_c.qmgr);
if (g_trr_c.jobmgr) dereference(g_trr_c.jobmgr);
xce_hook_fini(&g_trr_c.hooksys);
}
/* ── Encode helpers (mirror test_apply_committed_entry.c's tace_encode_*;
* capnp_encode_{job,worker} are file-static in persist.c). ──────────── */
static int trr_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 trr_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;
}
/* ── persist_t fixture (real disk-backed WAL, mirrors
* test_persist_recovery.c's make_fresh_persist). ─────────────────────── */
static persist_t* trr_make_fresh_persist(const char* tmpdir)
{
persist_t* p = calloc(1, sizeof(persist_t));
if (!p) return NULL;
persist_config_t cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000; /* slow fsync — no races in tests */
cfg.snap_keep_n = 2;
persist_set_workdir(p, tmpdir, "ctrl-test");
if (persist_init(p, &cfg, NULL) != 0) {
free(p);
return NULL;
}
if (persist_start(p) != 0) {
persist_fini(p);
free(p);
return NULL;
}
return p;
}
static void trr_teardown_persist(persist_t* p)
{
persist_stop(p);
persist_fini(p);
free(p);
}
/* Look up helpers, mirroring test_apply_committed_entry.c. */
static job_t* trr_find_job(uint64_t jobid)
{
job_t* j = NULL;
lock(g_trr_c.jobmgr, RTTI_LOCK_READ);
int found = key_find(g_trr_c.jobmgr, (void*)jobid, (void**)&j);
unlock(g_trr_c.jobmgr);
return (found > 0) ? j : NULL;
}
static queue_t* trr_find_queue(const char* name)
{
queue_t* q = NULL;
lock(g_trr_c.qmgr, RTTI_LOCK_READ);
key_find(g_trr_c.qmgr, (void*)name, (void**)&q);
unlock(g_trr_c.qmgr);
return q;
}
static worker_t* trr_find_worker(const char* name)
{
worker_t* w = NULL;
lock(g_trr_c.workermgr, RTTI_LOCK_READ);
key_find(g_trr_c.workermgr, (void*)name, (void**)&w);
unlock(g_trr_c.workermgr);
return w;
}
/*
* Writes one physical WAL entry of every classified type (real
* persist_wal_async appends — no hand-rolled headers/CRCs) in causal order:
* submit before the state fact that mutates the same job.
*/
static void trr_write_classified_log(persist_t* p)
{
/* WAL_JOB_SUBMIT: jobid 90001, PEND */
job_submit_data_t submit;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"rq1";
submit.user = (char*)"tester";
xce_job_t jdef;
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = 90001;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
uint8_t* job_payload = NULL;
size_t job_plen = 0;
TEST_ASSERT_EQUAL_INT(0, trr_encode_job(&jdef, &job_payload, &job_plen));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_SUBMIT, job_payload, (uint32_t)job_plen));
free(job_payload);
/* WAL_JOB_START (job-state fact family): jobid 90001 -> RUN */
wal_jobstate_t js = { .jobid = 90001, .state = JOB_STAT_RUN };
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_START, &js, sizeof(js)));
/* WAL_QUEUE_CTRL: close queue "rq1" */
const char* qname = "rq1";
size_t qname_len = strlen(qname) + 1;
size_t qtotal = sizeof(wal_queuehdr_t) + qname_len;
uint8_t* qpkt = malloc(qtotal);
TEST_ASSERT_NOT_NULL(qpkt);
wal_queuehdr_t qh = { .status = QUEUE_STAT_CLOSED };
memcpy(qpkt, &qh, sizeof(qh));
memcpy(qpkt + sizeof(qh), qname, qname_len);
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_QUEUE_CTRL, qpkt, (uint32_t)qtotal));
free(qpkt);
/* WAL_WORKER_CTRL: join worker "rw1" */
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*)"rw1";
wdef.attrs = &attrs;
uint8_t* wbuf = NULL;
size_t wsize = 0;
TEST_ASSERT_EQUAL_INT(0, trr_encode_worker(&wdef, &wbuf, &wsize));
uint32_t disc = (uint32_t)HOOK_DETAIL_WORKER_JOIN;
size_t wtotal = sizeof(disc) + wsize;
uint8_t* wpkt = malloc(wtotal);
TEST_ASSERT_NOT_NULL(wpkt);
memcpy(wpkt, &disc, sizeof(disc));
memcpy(wpkt + sizeof(disc), wbuf, wsize);
free(wbuf);
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_WORKER_CTRL, wpkt, (uint32_t)wtotal));
free(wpkt);
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
}
/*
* Req 3.1/3.2 + genuine-rewire evidence: restart replay of a log containing
* every classified entry type reaches the correct end state, and does so by
* actually firing apply_committed_entry()'s post-mutation notification for
* each entry (proving replay is wired to the new seam, not merely producing
* a coincidentally-matching state).
*/
static void test_replay_applies_every_classified_type_via_apply_seam(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_replay_seam_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Phase 1: write one entry of every classified type to a real WAL. No
* cluster needed — persist_wal_async is a physical-write-only path. */
{
persist_t* p = trr_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
trr_write_classified_log(p);
trr_teardown_persist(p);
}
/* Phase 2: fresh persist_t + fresh cluster_t, recover from disk. */
{
persist_t* p2 = trr_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
trr_build_cluster();
p2->cluster = &g_trr_c;
int rc = persist_recover(p2);
TEST_ASSERT_EQUAL_INT(0, rc);
job_t* j = trr_find_job(90001);
TEST_ASSERT_NOT_NULL_MESSAGE(j,
"replay must construct the job via the apply seam's "
"WAL_JOB_SUBMIT route");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
"replay must apply the job-state fact via the apply seam's "
"guarded transition core, reaching the same end state as before");
queue_t* q = trr_find_queue("rq1");
TEST_ASSERT_NOT_NULL(q);
TEST_ASSERT_TRUE_MESSAGE((q->queue.status & QUEUE_STAT_CLOSED) != 0,
"replay must apply WAL_QUEUE_CTRL via the apply seam");
worker_t* w = trr_find_worker("rw1");
TEST_ASSERT_NOT_NULL_MESSAGE(w,
"replay must apply WAL_WORKER_CTRL via the apply seam");
/* One notification per entry: JOB_NEW, JOB_UPDATE, QUEUE_UPDATE,
* WORKER_JOIN. apply_wal_entry (the pre-2.3 replay dispatch) never
* fires hooks — this assertion is 0 there and 4 once replay is
* pointed at apply_committed_entry, proving the call site moved. */
TEST_ASSERT_EQUAL_INT_MESSAGE(4, g_trr_hook_fires,
"restart replay must drive every classified entry through "
"apply_committed_entry(), firing exactly one post-mutation "
"notification per entry");
trr_teardown_cluster();
trr_teardown_persist(p2);
}
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp);
if (system(cmd) != 0)
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp);
printf("test_replay_applies_every_classified_type_via_apply_seam: PASS\n");
}
void test_replay_apply_seam(void)
{
RUN_TEST(test_replay_applies_every_classified_type_via_apply_seam);
}