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

feat(raft-apply-decoupling): post-replay apply-correctness verification (task 5.1)

Browse source 
Add apply_verdict_t (APPLY_CLEAN/APPLY_DIVERGENT) computed alongside
the existing structural (CRC) recover_verdict_t. Dual-check design:
an entity-count check (WAL_JOB_SUBMIT entries scanned vs live jobmgr
growth, genuinely independent of apply_committed_entry()'s return
code) and a position check against a new last_applied_lsn field
(forward-looking infrastructure for task 6's live pump).

Fixes a real bug found during review: the position check's
expectation was computed from the physical WAL-scan tip
(current_lsn - 1), which advances past checkpoint markers and
unknown entry types that never reach apply_committed_entry(). A
crash between persist_checkpoint()'s marker-fsync step and its
later snapshot-persist step leaves a trailing checkpoint marker on
next boot, which falsely reported APPLY_DIVERGENT on a perfectly
clean recovery -- reachable today, not gated behind a future task.
Fixed by reading last_applied_lsn directly, which only advances for
entries actually dispatched through the apply seam.

Known accepted gap: entity-count check covers jobs only, not
workers (which can genuinely be created via WAL_WORKER_CTRL) --
documented for future extension.
This commit is contained in:
Rongsong Shen 2026-07-04 11:32:21 +00:00
parent 665b29f5b0
commit c005d40741
6 changed files with 800 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -81,6 +81,7 @@ extern void test_apply_seam(void);
extern void test_apply_committed_entry(void);
extern void test_replay_apply_seam(void);
extern void test_apply_reapply_idempotency(void);
extern void test_apply_correctness_verification(void);
void setUp(void) { return; }
@ -231,6 +232,7 @@ int main(int argc, char* argv[])
TEST(test_apply_committed_entry);
TEST(test_replay_apply_seam);
TEST(test_apply_reapply_idempotency);
TEST(test_apply_correctness_verification);
return UNITY_END();
}

517
src/controller/testsuite/test_apply_correctness_verification.c Normal file
View file

@ -0,0 +1,517 @@
/* test_apply_correctness_verification.c — raft-apply-decoupling task 5.1
*
* Proves the post-replay apply-correctness verification added alongside the
* existing structural (CRC) recovery verdict (design.md "Restart Replay &
* Verification", Req 8.1/8.2):
*
* 1. A clean restart replay (real WAL, real cluster_t/jobmgr, no
* corruption, no divergence) produces an APPLY_CLEAN verdict from
* persist_apply_verdict() ALONGSIDE the existing RECOVER_CLEAN verdict
* from persist_recovery_verdict() — neither replaces the other.
*
* 2. The comparison logic itself — persist_verify_apply_correctness(), the
* pure function driving that verdict — can genuinely tell CLEAN apart
* from a mismatch: an expectation whose new-job-count disagrees with
* the live job table's actual growth is APPLY_DIVERGENT, and one whose
* last-applied position disagrees is APPLY_DIVERGENT, even when the
* other half of the expectation matches. This is exercised directly
* against the pure function (not through a real replay) — routing an
* APPLY_DIVERGENT verdict through the existing non-promotable fence
* mechanism is raft-apply-decoupling task 5.2, and a forced-DIVERGENT
* full-replay scenario proving the fence stays alive is task 5.3; both
* are out of this task's boundary.
*
* 3. persist_verify_apply_correctness() degrades safely to APPLY_CLEAN
* when there is no live cluster to query (a bare unit-test persist_t),
* mirroring apply_committed_entry()'s own "nothing to dispatch into"
* convention.
*
* Fixtures mirror test_replay_apply_seam.c (task 2.3, full cluster_t/WAL
* replay) and test_apply_seam.c (task 2.1, bare jobmgr, no cluster_t) —
* whichever is the lighter fit for each test.
*/
#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 <unistd.h>
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "hook.h"
#include "jobj.h"
#include "persist.h"
#include "wal_types.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 the other apply-seam test files. */
static int g_tacv_jobobj_ready;
/* ── Fixture A: full cluster_t + real disk-backed WAL, for the end-to-end
* clean-replay test (mirrors test_replay_apply_seam.c). ─────────────── */
static cluster_t g_tacv_c;
static void tacv_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*)"tacv-probe") == NULL)
xstring_table_init(64);
if (!g_tacv_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_tacv_jobobj_ready = 1;
}
memset(&g_tacv_c, 0, sizeof(g_tacv_c));
g_tacv_c.myhost = &g_tacv_c.myhost_info;
g_tacv_c.myhost_info.name = (char*)"ctrl-test";
g_tacv_c.myhost_info.key = (char*)"";
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_tacv_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_tacv_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_tacv_c.qmgr = makeInstance(&xkvmgr_type, &str_params);
g_tacv_c.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_tacv_c.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_tacv_c.jobmgr);
TEST_ASSERT_NOT_NULL(g_tacv_c.qmgr);
TEST_ASSERT_NOT_NULL(g_tacv_c.workermgr);
TEST_ASSERT_NOT_NULL(g_tacv_c.resmgr);
zarray_init(&g_tacv_c.sharedinst, 16, NULL);
}
static void tacv_teardown_cluster(void)
{
zarray_fini(&g_tacv_c.sharedinst);
if (g_tacv_c.resmgr) dereference(g_tacv_c.resmgr);
if (g_tacv_c.workermgr) dereference(g_tacv_c.workermgr);
if (g_tacv_c.qmgr) dereference(g_tacv_c.qmgr);
if (g_tacv_c.jobmgr) dereference(g_tacv_c.jobmgr);
xce_hook_fini(&g_tacv_c.hooksys);
}
static int tacv_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 persist_t* tacv_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 tacv_teardown_persist(persist_t* p)
{
persist_stop(p);
persist_fini(p);
free(p);
}
/*
* End-to-end: a clean restart replay of a log containing two WAL_JOB_SUBMIT
* entries (no corruption, no divergence) must produce APPLY_CLEAN alongside
* the existing RECOVER_CLEAN structural verdict (Req 8.1, 8.2).
*/
static void test_clean_replay_produces_apply_clean_alongside_crc_verdict(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_tacv_clean_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Phase 1: write two WAL_JOB_SUBMIT entries to a real WAL. */
{
persist_t* p = tacv_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
for (uint64_t jobid = 70001; jobid <= 70002; jobid++) {
xce_job_t jdef;
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_PEND;
job_submit_data_t submit;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"rq1";
submit.user = (char*)"tester";
jdef.submit = &submit;
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0, tacv_encode_job(&jdef, &payload, &plen));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_SUBMIT, payload, (uint32_t)plen));
free(payload);
}
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
tacv_teardown_persist(p);
}
/* Phase 2: fresh persist_t + fresh cluster_t, recover from disk. */
{
persist_t* p2 = tacv_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
tacv_build_cluster();
p2->cluster = &g_tacv_c;
TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
TEST_ASSERT_EQUAL_INT_MESSAGE(RECOVER_CLEAN,
persist_recovery_verdict(p2),
"existing structural (CRC) verdict must remain intact");
TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN,
persist_apply_verdict(p2),
"a clean replay with no divergence must report APPLY_CLEAN");
uint64_t expected_tip = (p2->current_lsn > 0) ? p2->current_lsn - 1 : 0;
TEST_ASSERT_EQUAL_UINT64_MESSAGE(expected_tip,
persist_last_applied_lsn(p2),
"the replay-derived apply cursor must reach the recovered "
"committed log tip");
job_t* j1 = NULL;
job_t* j2 = NULL;
lock(g_tacv_c.jobmgr, RTTI_LOCK_READ);
key_find(g_tacv_c.jobmgr, (void*)70001, (void**)&j1);
key_find(g_tacv_c.jobmgr, (void*)70002, (void**)&j2);
unlock(g_tacv_c.jobmgr);
TEST_ASSERT_NOT_NULL(j1);
TEST_ASSERT_NOT_NULL(j2);
tacv_teardown_cluster();
tacv_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_clean_replay_produces_apply_clean_alongside_crc_verdict: PASS\n");
}
/*
* raft-apply-decoupling task 5.1 round 2 (Req 8.2): a WAL whose tail (above
* the loaded snapshot's checkpoint_lsn) is a WAL_CHECKPOINT marker with no
* further logical entry after it must NOT be misreported as
* APPLY_DIVERGENT.
*
* This reproduces the persist_checkpoint() crash window directly:
* persist_checkpoint()'s step 1 (persist_write_checkpoint_marker) durably
* fsyncs the WAL_CHECKPOINT marker; step 3b (write_snapshot) only later
* persists the advanced checkpoint_lsn to a snapshot file. A crash between
* those two steps leaves the marker durable in the WAL with no snapshot
* reflecting it. There is no test seam to interrupt persist_checkpoint()
* mid-sequence, so this test runs a real (complete) checkpoint and then
* deletes the snapshot file it just wrote — landing the on-disk state in
* exactly the same place a genuine crash between steps 1 and 3b would:
* marker durable in the WAL, no snapshot recording the advanced
* checkpoint_lsn.
*
* On the next recover, persist_recover() finds no snapshot (checkpoint_lsn
* starts at 0) and replay_wal() processes the full WAL: two real
* WAL_JOB_SUBMIT entries (advancing the logical apply cursor,
* last_applied_lsn, to their LSNs) followed by the trailing WAL_CHECKPOINT
* marker, which only advances the PHYSICAL WAL-scan position (current_lsn)
* — replay_wal's WAL_CHECKPOINT skip branch never routes it through
* apply_committed_entry(). The physical tip (current_lsn - 1) therefore
* lands one LSN above the true logical cursor, and the fix under test must
* compare against the logical cursor, not the physical tip.
*/
static void test_checkpoint_tail_with_no_snapshot_is_apply_clean(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_tacv_ckpt_tail_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Phase 1: two real WAL_JOB_SUBMIT entries, then a checkpoint (which
* appends a WAL_CHECKPOINT marker as the new highest-LSN entry) — then
* delete the checkpoint's own snapshot file to simulate a crash between
* the marker's fsync (step 1) and the snapshot write (step 3b) that
* would have durably recorded the advanced checkpoint_lsn. */
uint64_t marker_checkpoint_lsn = 0;
{
persist_t* p = tacv_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
for (uint64_t jobid = 71001; jobid <= 71002; jobid++) {
xce_job_t jdef;
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_PEND;
job_submit_data_t submit;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"rq1";
submit.user = (char*)"tester";
jdef.submit = &submit;
uint8_t* payload = NULL;
size_t plen = 0;
TEST_ASSERT_EQUAL_INT(0, tacv_encode_job(&jdef, &payload, &plen));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_SUBMIT, payload, (uint32_t)plen));
free(payload);
}
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
/* Real, complete checkpoint — marker fsynced AND snapshot written.
* A genuine crash never reaches the snapshot write, but there is no
* seam to stop persist_checkpoint() midway, so the snapshot is
* deleted immediately below to reach the identical post-crash
* on-disk state. */
TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(p));
marker_checkpoint_lsn = p->checkpoint_lsn;
char snap_file[4096];
snprintf(snap_file, sizeof(snap_file), "%s/snap-%016" PRIx64 ".capnp",
p->snap_path, p->checkpoint_lsn);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, unlink(snap_file),
"fixture sanity: the checkpoint's own snapshot file must exist "
"before it is deleted to simulate the crash window");
tacv_teardown_persist(p);
}
TEST_ASSERT_TRUE(marker_checkpoint_lsn > 0);
/* Phase 2: fresh persist_t recovers from disk. No snapshot exists (it
* was deleted above), so replay starts from checkpoint_lsn=0 and
* replays the full WAL, ending on the trailing WAL_CHECKPOINT marker. */
{
persist_t* p2 = tacv_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
TEST_ASSERT_EQUAL_INT_MESSAGE(RECOVER_CLEAN,
persist_recovery_verdict(p2),
"a checkpoint-marker tail is not WAL corruption -- must stay "
"RECOVER_CLEAN");
/* Fixture sanity: confirm this test actually exercises the
* physical/logical cursor gap the round-2 bug conflated -- the
* physical WAL-scan tip must land strictly above the logical apply
* cursor (on the checkpoint marker's LSN). */
uint64_t physical_tip =
(p2->current_lsn > 0) ? p2->current_lsn - 1 : 0;
uint64_t logical_cursor = persist_last_applied_lsn(p2);
TEST_ASSERT_TRUE_MESSAGE(physical_tip > logical_cursor,
"fixture sanity: the checkpoint marker must sit above the last "
"logical entry for this test to exercise the round-2 bug");
TEST_ASSERT_EQUAL_UINT64(marker_checkpoint_lsn, physical_tip + 1);
TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN,
persist_apply_verdict(p2),
"a checkpoint-marker tail with no further logical entry must "
"NOT be misreported as APPLY_DIVERGENT (Req 8.2)");
tacv_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_checkpoint_tail_with_no_snapshot_is_apply_clean: PASS\n");
}
/* ── Fixture B: bare jobmgr + a minimal cluster_t wrapper (no hooksys, no
* persist/WAL), for driving persist_verify_apply_correctness() directly
* as a pure function — mirrors test_apply_seam.c's bare-jobmgr fixture,
* which needs none of Fixture A's full cluster machinery either. ──────── */
static var g_tacv_bare_jobmgr;
static void tacv_bare_setup(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 16,
.delta = 16 };
if (makeString((char*)"tacv-bare-probe") == NULL)
xstring_table_init(64);
if (!g_tacv_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(16, 16));
g_tacv_jobobj_ready = 1;
}
g_tacv_bare_jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(g_tacv_bare_jobmgr);
}
static void tacv_bare_teardown(void)
{
if (g_tacv_bare_jobmgr != NULL) {
dereference(g_tacv_bare_jobmgr);
g_tacv_bare_jobmgr = NULL;
}
}
static job_t* tacv_bare_make_job(uint64_t jobid)
{
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 = jobid;
jdef.state = JOB_STAT_PEND;
jdef.submit = &submit;
return makeJob(g_tacv_bare_jobmgr, &jdef);
}
/*
* The comparison logic must genuinely distinguish a matching expectation
* from one whose new-job-count disagrees with the live job table's actual
* growth (Req 8.1/8.2's comparison step, exercised directly — routing this
* outcome through the fence is task 5.2's job, not this test's).
*/
static void test_verify_apply_correctness_detects_job_count_mismatch(void)
{
tacv_bare_setup();
cluster_t c;
memset(&c, 0, sizeof(c));
c.jobmgr = g_tacv_bare_jobmgr;
persist_t p;
memset(&p, 0, sizeof(p));
p.cluster = &c;
uint64_t jobmgr_count_before = 0; /* empty table before "replay" */
/* Two jobs materialize — mimics what a real replay would have done via
* apply_committed_entry()'s WAL_JOB_SUBMIT route. */
TEST_ASSERT_NOT_NULL(tacv_bare_make_job(80001));
TEST_ASSERT_NOT_NULL(tacv_bare_make_job(80002));
atomic_store_explicit(&p.last_applied_lsn, 42, memory_order_relaxed);
/* Matching expectation: 2 new jobs, position 42 — CLEAN. */
apply_expectation_t exp_match = {
.expected_last_applied_lsn = 42,
.expected_new_job_count = 2,
};
TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN,
persist_verify_apply_correctness(&p, &exp_match, jobmgr_count_before),
"matching expectation must report APPLY_CLEAN");
/* Mismatched expectation: log says 3 jobs should exist, only 2 do. */
apply_expectation_t exp_count_mismatch = {
.expected_last_applied_lsn = 42,
.expected_new_job_count = 3,
};
TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_DIVERGENT,
persist_verify_apply_correctness(&p, &exp_count_mismatch,
jobmgr_count_before),
"a job-count mismatch must be detected as APPLY_DIVERGENT");
/* Mismatched expectation: position disagrees, count matches. */
apply_expectation_t exp_position_mismatch = {
.expected_last_applied_lsn = 99,
.expected_new_job_count = 2,
};
TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_DIVERGENT,
persist_verify_apply_correctness(&p, &exp_position_mismatch,
jobmgr_count_before),
"an apply-cursor position mismatch must be detected as "
"APPLY_DIVERGENT even when the job count matches");
tacv_bare_teardown();
printf("test_verify_apply_correctness_detects_job_count_mismatch: PASS\n");
}
/*
* Defensive: no live cluster to query (a bare unit-test persist_t, or any
* caller with nothing to dispatch into) must degrade to APPLY_CLEAN rather
* than false-positive DIVERGENT — mirrors apply_committed_entry()'s own
* "nothing to dispatch into" convention.
*/
static void test_verify_apply_correctness_clean_when_no_cluster(void)
{
persist_t p;
memset(&p, 0, sizeof(p));
p.cluster = NULL;
atomic_store_explicit(&p.last_applied_lsn, 7, memory_order_relaxed);
apply_expectation_t exp = {
.expected_last_applied_lsn = 7,
.expected_new_job_count = 5, /* irrelevant -- no cluster to check */
};
TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN,
persist_verify_apply_correctness(&p, &exp, 0),
"no live cluster to query must degrade to APPLY_CLEAN, not a false "
"DIVERGENT");
printf("test_verify_apply_correctness_clean_when_no_cluster: PASS\n");
}
void test_apply_correctness_verification(void)
{
RUN_TEST(test_clean_replay_produces_apply_clean_alongside_crc_verdict);
RUN_TEST(test_checkpoint_tail_with_no_snapshot_is_apply_clean);
RUN_TEST(test_verify_apply_correctness_detects_job_count_mismatch);
RUN_TEST(test_verify_apply_correctness_clean_when_no_cluster);
}