From a166ecb45b1a49441f9183ad5de625484a0f2fd4 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 22 Jun 2026 13:05:23 +0000 Subject: [PATCH] test(ha-term-safety): cross-member walcompare divergence tool (task 5.3) Diffs member WALs by (lsn -> term, pcrc) over the committed prefix, exits non-zero on any committed-prefix divergence with the diverging LSN + per-member entry identities. Reuses the canonical waldump decoder via a new per-entry visitor seam (no second WAL parser). Detects same-LSN/diff-term and same-LSN/diff-content divergences (the bug-851 class) that an LSN-range check would mask. --- .kiro/specs/ha-term-safety/tasks.md | 5 +- src/controller/meson.build | 4 +- src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test.c | 68 ++++ src/controller/testsuite/test_walcompare.c | 370 +++++++++++++++++++++ src/controller/testsuite/walcompare.c | 272 +++++++++++++++ src/controller/testsuite/walcompare.h | 64 ++++ src/controller/testsuite/waldump.c | 35 +- src/controller/testsuite/waldump.h | 27 ++ 9 files changed, 842 insertions(+), 5 deletions(-) create mode 100644 src/controller/testsuite/test_walcompare.c create mode 100644 src/controller/testsuite/walcompare.c create mode 100644 src/controller/testsuite/walcompare.h diff --git a/.kiro/specs/ha-term-safety/tasks.md b/.kiro/specs/ha-term-safety/tasks.md index d239490b..95ed0f5b 100644 --- a/.kiro/specs/ha-term-safety/tasks.md +++ b/.kiro/specs/ha-term-safety/tasks.md @@ -118,7 +118,7 @@ - _Requirements: 5.1, 5.2, 5.3, 5.4_ - _Depends: 3.3_ -- [ ] 5.3 Cross-member `walcompare` divergence tool +- [x] 5.3 Cross-member `walcompare` divergence tool - Build a tool that diffs member WALs by `(lsn → term, pcrc)` and exits non-zero on any committed-prefix divergence, reporting the diverging LSN and per-member entry identities; reuse the existing WAL decoder - Observable completion: the tool returns success on identical member WALs and fails (with the diverging LSN) on a hand-divergent pair - _Requirements: 6.1, 6.2, 6.3_ @@ -199,3 +199,6 @@ - **5.1 done.** Coverage audit found all 5 scenarios + 5 callbacks already covered by per-task tests; the genuine gap was the task's literal observable ("exercises every callback AND the marker path"). Added ONE integration test `test_every_callback_and_marker_path_end_to_end` (test_term_safety_adapter.c, test-only, empty production diff) driving a leader-lifecycle flow through the REAL seams: ha_primary_log_append→raft_log_append→on_persist_entry (job@1, marker@2, job@3 term-stamped); on_get_entry_term read-back; on_send_append_entries captures the marker entry (lsn2/WAL_CONFIG_MARKER/payload40) via repl_send_ae_capture_hook; on_commit_advanced ceiling caps then real ha_raft_on_repl_ack→raft_advance_commit releases; on_truncate_from drops LSN3, committed job+marker prefix + derived config intact. RED proven by neuter-and-revert. 33 Tests 0 Failures under DEFAULT ASAN (leak-detect ON), no new leak. Reviewer APPROVED (independently rebuilt the coverage map; confirmed real seams + load-bearing assertions). - **NEXT (5.x live env):** 5.2 (clean single-failover integration) _Depends 3.3_ and 5.4 (live chaotic-multi-kill regression) _Depends 5.2,5.3,4.3_ require a real 3-node deploy (/xce-deploy-ha) on a real-fs (bug-924: overlayfs fdatasync degrades meson + would corrupt timing-sensitive live runs). 5.3 (walcompare tool) is build-only (no live env). The live-send retirement (term-0 repl_broadcast → library send loop) deferred from 3.2/3.3 lands/validates in 5.2/5.4. + +- **5.3 done.** `walcompare` tool: `xce_test walcompare [--committed N] [...]` + reusable `walcompare()` (walcompare.c/h). Diffs members by `(lsn → term, pcrc)` over the committed prefix; pcrc = payload-only crc32 (NOT header) so same-LSN/diff-content divergences surface; legacy-0→WAL_LEGACY_TERM. Exit 0 identical / WALCMP_DIVERGENT(1) on committed-prefix divergence (reports diverging LSN + per-member term/pcrc/present-or-MISSING for ALL N members) / negative errno on load error. REUSES the canonical waldump decoder via a NEW `waldump_path_visit` per-entry visitor seam (`waldump_path` is now a thin wrapper — existing waldump behavior + tests unchanged, NO 5th WAL parser). 11 tests; reviewer APPROVED (RED empirically reproduced by neutering compare; CLI hand-verified). + - **ADVISORY for 5.4 (reviewer MEDIUM):** the DEFAULT committed boundary = `min(recovered_tip)` across members. A member that diverged then has a CRC-broken/short tail carries a LOWER recovered_tip → lowers the shared boundary → could MASK a divergence above it. 5.4 MUST pass an EXPLICIT `--committed N` (highest acked LSN), and pair walcompare with `expect-job $jobid` (R6.4) which independently catches a lost committed job. diff --git a/src/controller/meson.build b/src/controller/meson.build index 8bbfe4b4..2f1409f2 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -53,7 +53,7 @@ controller = executable('xce_controller', link_args: wrap_link_args, install: true) -xce_test_src = [ 'testsuite/test.c', 'testsuite/waldump.c'] +xce_test_src = [ 'testsuite/test.c', 'testsuite/waldump.c', 'testsuite/walcompare.c'] readline_dep = dependency('readline') ncurses_dep = dependency('ncurses') @@ -141,6 +141,8 @@ controller_unit_test_src = [ 'testsuite/test_term_safety_adapter.c', 'testsuite/test_waldump.c', 'testsuite/waldump.c', + 'testsuite/test_walcompare.c', + 'testsuite/walcompare.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 0e0bf809..c3e2cb65 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -65,6 +65,7 @@ extern void test_ha_heartbeat_committed(void); extern void test_ha_vote_floor_heartbeat(void); extern void test_term_safety_adapter(void); extern void test_waldump(void); +extern void test_walcompare(void); void setUp(void) { return; } @@ -199,6 +200,7 @@ int main(int argc, char* argv[]) TEST(test_ha_vote_floor_heartbeat); TEST(test_term_safety_adapter); TEST(test_waldump); + TEST(test_walcompare); return UNITY_END(); } diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index 2ac6c8e1..6681ad6e 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -1,6 +1,7 @@ #include #include #include +#include #include #include #include @@ -22,6 +23,7 @@ #include "delay.h" #include "idgen.h" #include "preason.h" +#include "walcompare.h" /* cross-member committed-prefix divergence (cmd_walcompare) */ #include "waldump.h" /* offline WAL inspector (cmd_waldump) */ #include "xce/cluster.h" /* xce_parse_peers, xce_cluster_config_t */ #include "xce/types.h" @@ -124,6 +126,7 @@ extern int cmd_expect_job(int, char*[]); extern int cmd_wait_primary(int, char*[]); extern int cmd_ctrlstatus(int, char*[]); extern int cmd_waldump(int, char*[]); +extern int cmd_walcompare(int, char*[]); static int worker_house_keeping(void*); /* Session verdict / loop state (xce-test-e2e-primitives task 1.1). @@ -182,6 +185,7 @@ static command_t top_commands[] = { { "set", cmd_set, "Show/Set settings" }, { "echo", cmd_echo, "Print a message" }, { "waldump", cmd_waldump, "Inspect WAL segments" }, + { "walcompare", cmd_walcompare, "Diff member committed WAL prefixes" }, { "quit", cmd_quit, "Quit xce test" }, { (char*)NULL, (cmd_func_t)NULL, (char*)NULL } }; @@ -220,6 +224,7 @@ static command_t client_commands[] = { "wait until a primary serves [timeout_ms]" }, { "ctrlstatus", cmd_ctrlstatus, "show controller/cluster status" }, { "waldump", cmd_waldump, "Inspect WAL segments" }, + { "walcompare", cmd_walcompare, "Diff member committed WAL prefixes" }, { "shutdown", cmd_shutdown, "shutdown xce controller" }, { "jobrun", cmd_jobrun, "run a job" }, { "jobsub", cmd_jobsub, "submit a job" }, @@ -716,6 +721,69 @@ int cmd_waldump(int argc, char* argv[]) return rc; } +/* walcompare [--committed N] [more...] — diff member WAL + * committed prefixes by (lsn -> term, pcrc) and FAIL (nonzero) on any + * committed-prefix divergence (ha-term-safety R6.1/R6.2/R6.3), reusing the + * waldump decoder. Each member path may be a seg-*.wal file, a wal/ dir, or a + * node/data dir (same resolution as waldump). With no --committed, the shared + * committed boundary = the minimum CRC-intact tip across members. The diverging + * LSN and per-member (term, pcrc) identities are reported. Exit 0 = identical, + * nonzero = divergence or load error — so it doubles as a scripted assertion. */ +int cmd_walcompare(int argc, char* argv[]) +{ + uint64_t committed = 0; /* 0 = derive the shared boundary */ + int ai = 1; + + /* Optional global committed boundary applied to every member. */ + if (argc > ai && strcmp(argv[ai], "--committed") == 0) { + if (argc <= ai + 1) { + fprintf(stderr, "walcompare: --committed needs an LSN\n"); + return -1; + } + char* end = NULL; + unsigned long v = strtoul(argv[ai + 1], &end, 10); + if (end == argv[ai + 1] || *end != '\0') { + fprintf(stderr, "walcompare: bad --committed LSN '%s'\n", + argv[ai + 1]); + return -1; + } + committed = (uint64_t)v; + ai += 2; + } + + int n = argc - ai; + if (n < 2) { + fprintf(stderr, + "usage: walcompare [--committed N] " + "[more...]\n"); + return -1; + } + if (n > 16) { + fprintf(stderr, "walcompare: at most 16 members\n"); + return -1; + } + + walcompare_member_t members[16]; + for (int i = 0; i < n; i++) { + members[i].path = argv[ai + i]; + members[i].committed_lsn = committed; /* 0 => derive shared boundary */ + } + + walcompare_divergence_t div; + int rc = walcompare(members, n, stdout, &div); + if (rc == WALCMP_OK) + return 0; + if (rc == WALCMP_DIVERGENT) { + fprintf(stderr, + "walcompare: FAIL — committed-prefix divergence at LSN %" PRIu64 + "\n", + div.lsn); + return -1; /* nonzero session verdict */ + } + fprintf(stderr, "walcompare: error: %s\n", strerror(-rc)); + return -1; +} + int cmd_up(int argc, char* argv[]) { if (active_ctx->parent != NULL) { diff --git a/src/controller/testsuite/test_walcompare.c b/src/controller/testsuite/test_walcompare.c new file mode 100644 index 00000000..25e8e3b3 --- /dev/null +++ b/src/controller/testsuite/test_walcompare.c @@ -0,0 +1,370 @@ +#define _GNU_SOURCE +#include +#include +#include +#include + +#include + +#include + +#include "config.h" /* WAL_HDR_CRC_SIZE, WAL_CRC_NONE */ +#include "persist.h" /* WAL_LEGACY_TERM */ +#include "walcompare.h" +#include "wal_types.h" + +/* ------------------------------------------------------------------ * + * Harness: hand-write raw WAL segments (LSN, term, payload all under + * test control) into per-member temp dirs, then drive walcompare over + * them. Mirrors test_waldump's writer so the on-disk format is pinned + * independently of persist.c — and exercises the REAL waldump decoder + * that walcompare reuses (no second parser). + * ------------------------------------------------------------------ */ + +/* Append one entry to `fp`, chaining prev_crc from *last_crc. CRC32 over + * WAL_HDR_CRC_SIZE header bytes + payload (identical to persist.c append). */ +static void put_entry(FILE* fp, uint64_t term, uint64_t lsn, uint32_t type, + const void* payload, uint32_t plen, uint32_t* last_crc, + int first_in_seg) +{ + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.term = term; + hdr.lsn = lsn; + hdr.timestamp = 0; + hdr.type = type; + hdr.payload_len = plen; + hdr.prev_crc = first_in_seg ? WAL_CRC_NONE : *last_crc; + + uint32_t c = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE); + if (plen > 0) + c = crc32(c, (const Bytef*)payload, plen); + hdr.crc = c; + *last_crc = c; + + fwrite(&hdr, sizeof(hdr), 1, fp); + if (plen > 0) + fwrite(payload, plen, 1, fp); +} + +/* A single hand-built WAL entry spec for a member. */ +typedef struct { + uint64_t term; + uint32_t type; + const char* payload; /* NUL-terminated; length is strlen (0 => no payload) */ +} wal_spec_t; + +/* Mint a fresh temp dir and write a single seg-0001.wal holding `specs` + * (LSNs 1..n contiguous). Returns the dir path in `dir_out`. */ +static void build_member(char* dir_out, size_t dirsz, const wal_spec_t* specs, + int n) +{ + snprintf(dir_out, dirsz, "/tmp/xce_walcmp_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(dir_out)); + + char path[512]; + snprintf(path, sizeof(path), "%s/seg-%016lx.wal", dir_out, 1UL); + FILE* fp = fopen(path, "wb"); + TEST_ASSERT_NOT_NULL(fp); + + uint32_t last = WAL_CRC_NONE; + for (int i = 0; i < n; i++) { + uint32_t plen = + specs[i].payload ? (uint32_t)strlen(specs[i].payload) : 0; + put_entry(fp, specs[i].term, (uint64_t)(i + 1), specs[i].type, + specs[i].payload, plen, &last, i == 0); + } + fclose(fp); +} + +/* ------------------------------------------------------------------ */ +/* Identical member WALs → success (exit 0). */ +/* ------------------------------------------------------------------ */ +static void test_identical_members_ok(void) +{ + wal_spec_t specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, + { 2, WAL_JOB_SUBMIT, "jobB" }, + }; + char a[256], b[256]; + build_member(a, sizeof(a), specs, 3); + build_member(b, sizeof(b), specs, 3); + + walcompare_member_t m[2] = { + { a, 0 }, + { b, 0 }, + }; + walcompare_divergence_t div; + int rc = walcompare(m, 2, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc); +} + +/* ------------------------------------------------------------------ */ +/* Hand-divergent pair: SAME LSN, DIFFERENT term within the committed */ +/* prefix → FAIL with the diverging LSN (R6.2/R6.3). */ +/* ------------------------------------------------------------------ */ +static void test_same_lsn_diff_term_divergent(void) +{ + wal_spec_t a_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, + { 2, WAL_JOB_SUBMIT, "jobB" }, /* LSN 3, term 2 */ + }; + wal_spec_t b_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, + { 3, WAL_JOB_SUBMIT, "jobB" }, /* LSN 3, term 3 — divergent */ + }; + char a[256], b[256]; + build_member(a, sizeof(a), a_specs, 3); + build_member(b, sizeof(b), b_specs, 3); + + walcompare_member_t m[2] = { { a, 0 }, { b, 0 } }; + walcompare_divergence_t div; + int rc = walcompare(m, 2, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc); + TEST_ASSERT_EQUAL_UINT64(3, div.lsn); /* the diverging LSN */ + TEST_ASSERT_EQUAL_UINT64(2, div.member_term[0]); + TEST_ASSERT_EQUAL_UINT64(3, div.member_term[1]); +} + +/* ------------------------------------------------------------------ */ +/* SAME LSN, SAME term, DIFFERENT payload (pcrc) → FAIL (R6.2: a content */ +/* divergence is not masked by equal term + LSN range). */ +/* ------------------------------------------------------------------ */ +static void test_same_lsn_diff_payload_divergent(void) +{ + wal_spec_t a_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, /* LSN 2 */ + }; + wal_spec_t b_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobZ" }, /* LSN 2, same term, different bytes */ + }; + char a[256], b[256]; + build_member(a, sizeof(a), a_specs, 2); + build_member(b, sizeof(b), b_specs, 2); + + walcompare_member_t m[2] = { { a, 0 }, { b, 0 } }; + walcompare_divergence_t div; + int rc = walcompare(m, 2, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc); + TEST_ASSERT_EQUAL_UINT64(2, div.lsn); + TEST_ASSERT_EQUAL_UINT64(1, div.member_term[0]); + TEST_ASSERT_EQUAL_UINT64(1, div.member_term[1]); + TEST_ASSERT_NOT_EQUAL(div.member_pcrc[0], div.member_pcrc[1]); +} + +/* ------------------------------------------------------------------ */ +/* A member that is AHEAD (longer log) with a DIFFERENT uncommitted tail */ +/* is NOT a divergence — the shared committed prefix matches (§6.1). */ +/* ------------------------------------------------------------------ */ +static void test_uncommitted_tail_not_flagged(void) +{ + wal_spec_t a_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, /* shared committed prefix: LSN 1..2 */ + }; + wal_spec_t b_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, + { 2, WAL_JOB_SUBMIT, "tailX" }, /* uncommitted tail — beyond shared tip */ + }; + char a[256], b[256]; + build_member(a, sizeof(a), a_specs, 2); /* tip 2 */ + build_member(b, sizeof(b), b_specs, 3); /* tip 3 */ + + /* Shared boundary = min recovered tip = 2; the LSN-3 tail is not compared. */ + walcompare_member_t m[2] = { { a, 0 }, { b, 0 } }; + walcompare_divergence_t div; + int rc = walcompare(m, 2, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc); +} + +/* ------------------------------------------------------------------ */ +/* An explicit committed boundary SHORTER than a member's log: a */ +/* divergence ABOVE the boundary must NOT be flagged (§6.1 committed */ +/* prefix only). */ +/* ------------------------------------------------------------------ */ +static void test_divergence_above_committed_boundary_not_flagged(void) +{ + wal_spec_t a_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, /* LSN 2 committed */ + { 2, WAL_JOB_SUBMIT, "uncA" }, /* LSN 3 uncommitted, divergent */ + }; + wal_spec_t b_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, + { 9, WAL_JOB_SUBMIT, "uncB" }, /* LSN 3, different term+bytes */ + }; + char a[256], b[256]; + build_member(a, sizeof(a), a_specs, 3); + build_member(b, sizeof(b), b_specs, 3); + + /* committed boundary pinned at 2 → the LSN-3 divergence is below scope. */ + walcompare_member_t m[2] = { { a, 2 }, { b, 2 } }; + walcompare_divergence_t div; + int rc = walcompare(m, 2, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc); + + /* Same data, but raise the boundary to 3 → now it IS a divergence. */ + walcompare_member_t m3[2] = { { a, 3 }, { b, 3 } }; + rc = walcompare(m3, 2, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc); + TEST_ASSERT_EQUAL_UINT64(3, div.lsn); +} + +/* ------------------------------------------------------------------ */ +/* A member MISSING a committed entry (a hole inside the committed */ +/* prefix the boundary covers) is a divergence (R4.2/R6.4). */ +/* ------------------------------------------------------------------ */ +static void test_missing_committed_entry_divergent(void) +{ + /* Member A holds 1..3; member B holds only 1..2 but is told its + * committed boundary is 3 — it is MISSING the committed entry at LSN 3. */ + wal_spec_t a_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, + { 2, WAL_JOB_SUBMIT, "jobB" }, + }; + wal_spec_t b_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, + }; + char a[256], b[256]; + build_member(a, sizeof(a), a_specs, 3); + build_member(b, sizeof(b), b_specs, 2); + + walcompare_member_t m[2] = { { a, 3 }, { b, 3 } }; + walcompare_divergence_t div; + int rc = walcompare(m, 2, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc); + TEST_ASSERT_EQUAL_UINT64(3, div.lsn); + TEST_ASSERT_EQUAL_INT(1, div.member_present[0]); /* A holds it */ + TEST_ASSERT_EQUAL_INT(0, div.member_present[1]); /* B is missing it */ +} + +/* ------------------------------------------------------------------ */ +/* Two legacy-0-term entries at the same LSN are NOT divergent: legacy 0 */ +/* maps to WAL_LEGACY_TERM for both (R1.8). */ +/* ------------------------------------------------------------------ */ +static void test_legacy_zero_term_not_divergent(void) +{ + wal_spec_t specs[] = { + { 0, WAL_NOOP, NULL }, /* legacy-0 term */ + { 0, WAL_JOB_SUBMIT, "jobA" }, /* legacy-0 term, same payload */ + }; + char a[256], b[256]; + build_member(a, sizeof(a), specs, 2); + build_member(b, sizeof(b), specs, 2); + + walcompare_member_t m[2] = { { a, 0 }, { b, 0 } }; + walcompare_divergence_t div; + int rc = walcompare(m, 2, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc); +} + +/* ------------------------------------------------------------------ */ +/* Three members, all identical → success (the live regression shape). */ +/* ------------------------------------------------------------------ */ +static void test_three_members_identical_ok(void) +{ + wal_spec_t specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "j1" }, + { 1, WAL_CONFIG_MARKER, "cfgref" }, /* a config marker in the log */ + { 2, WAL_JOB_SUBMIT, "j2" }, + }; + char a[256], b[256], c[256]; + build_member(a, sizeof(a), specs, 4); + build_member(b, sizeof(b), specs, 4); + build_member(c, sizeof(c), specs, 4); + + walcompare_member_t m[3] = { { a, 0 }, { b, 0 }, { c, 0 } }; + walcompare_divergence_t div; + int rc = walcompare(m, 3, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_OK, rc); +} + +/* ------------------------------------------------------------------ */ +/* Three members, one with a divergent committed config marker → FAIL */ +/* (R6.6: config markers reconcile by (term,lsn) like job entries). */ +/* ------------------------------------------------------------------ */ +static void test_three_members_divergent_config_marker(void) +{ + wal_spec_t ok_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "j1" }, + { 1, WAL_CONFIG_MARKER, "cfgX" }, /* LSN 3 config ref */ + }; + wal_spec_t bad_specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "j1" }, + { 1, WAL_CONFIG_MARKER, "cfgY" }, /* LSN 3, different config ref */ + }; + char a[256], b[256], c[256]; + build_member(a, sizeof(a), ok_specs, 3); + build_member(b, sizeof(b), ok_specs, 3); + build_member(c, sizeof(c), bad_specs, 3); + + walcompare_member_t m[3] = { { a, 0 }, { b, 0 }, { c, 0 } }; + walcompare_divergence_t div; + int rc = walcompare(m, 3, stdout, &div); + TEST_ASSERT_EQUAL_INT(WALCMP_DIVERGENT, rc); + TEST_ASSERT_EQUAL_UINT64(3, div.lsn); + /* A and B agree; C diverges. */ + TEST_ASSERT_EQUAL_UINT32(div.member_pcrc[0], div.member_pcrc[1]); + TEST_ASSERT_NOT_EQUAL(div.member_pcrc[0], div.member_pcrc[2]); +} + +/* ------------------------------------------------------------------ */ +/* A missing WAL source is a hard (negative) error, not a false OK. */ +/* ------------------------------------------------------------------ */ +static void test_missing_member_errors(void) +{ + wal_spec_t specs[] = { + { 1, WAL_NOOP, NULL }, + { 1, WAL_JOB_SUBMIT, "jobA" }, + }; + char a[256]; + build_member(a, sizeof(a), specs, 2); + + walcompare_member_t m[2] = { + { a, 0 }, + { "/tmp/xce_walcmp_does_not_exist_zzz", 0 }, + }; + walcompare_divergence_t div; + int rc = walcompare(m, 2, stdout, &div); + TEST_ASSERT_LESS_THAN_INT(0, rc); /* negative errno, not OK/DIVERGENT */ +} + +/* ------------------------------------------------------------------ */ +/* Fewer than two members is a usage error. */ +/* ------------------------------------------------------------------ */ +static void test_too_few_members_errors(void) +{ + wal_spec_t specs[] = { { 1, WAL_NOOP, NULL } }; + char a[256]; + build_member(a, sizeof(a), specs, 1); + walcompare_member_t m[1] = { { a, 0 } }; + int rc = walcompare(m, 1, stdout, NULL); + TEST_ASSERT_LESS_THAN_INT(0, rc); +} + +void test_walcompare(void) +{ + RUN_TEST(test_identical_members_ok); + RUN_TEST(test_same_lsn_diff_term_divergent); + RUN_TEST(test_same_lsn_diff_payload_divergent); + RUN_TEST(test_uncommitted_tail_not_flagged); + RUN_TEST(test_divergence_above_committed_boundary_not_flagged); + RUN_TEST(test_missing_committed_entry_divergent); + RUN_TEST(test_legacy_zero_term_not_divergent); + RUN_TEST(test_three_members_identical_ok); + RUN_TEST(test_three_members_divergent_config_marker); + RUN_TEST(test_missing_member_errors); + RUN_TEST(test_too_few_members_errors); +} diff --git a/src/controller/testsuite/walcompare.c b/src/controller/testsuite/walcompare.c new file mode 100644 index 00000000..c541091e --- /dev/null +++ b/src/controller/testsuite/walcompare.c @@ -0,0 +1,272 @@ +#define _GNU_SOURCE +#include "walcompare.h" + +#include +#include +#include +#include + +#include "persist.h" /* WAL_LEGACY_TERM */ +#include "waldump.h" + +#define WALCMP_MAX_MEMBERS 16 + +/* One member's collected log, indexed by LSN. `slot[lsn]` is valid when + * `present[lsn]` is set. LSNs are 1-based; index 0 is unused. */ +typedef struct { + uint64_t recovered_tip; /* highest CRC-intact LSN held (waldump tip) */ + uint64_t committed_lsn; /* effective committed boundary for this member */ + uint64_t cap; /* allocated index size (LSN range [0..cap-1]) */ + uint64_t* term; /* raw on-disk term per LSN (0 = legacy / absent) */ + uint32_t* pcrc; /* payload CRC per LSN */ + uint8_t* present; /* 1 if this LSN is held */ +} member_log_t; + +/* Visitor context: collect entries into a member_log_t. */ +typedef struct { + member_log_t* log; + int oom; /* set on an allocation failure mid-scan */ +} collect_ctx_t; + +/* Map a raw on-disk term to its matching term: legacy 0 -> WAL_LEGACY_TERM + * (R1.8) so two pre-feature entries are not spuriously divergent. */ +static uint64_t match_term(uint64_t raw) +{ + return raw == 0 ? WAL_LEGACY_TERM : raw; +} + +/* Grow the per-LSN index so `lsn` is addressable. Returns 0 or -ENOMEM. */ +static int member_log_ensure(member_log_t* m, uint64_t lsn) +{ + if (lsn < m->cap) + return 0; + uint64_t newcap = m->cap ? m->cap : 64; + while (newcap <= lsn) + newcap *= 2; + + uint64_t* nt = realloc(m->term, newcap * sizeof(*nt)); + uint32_t* np = realloc(m->pcrc, newcap * sizeof(*np)); + uint8_t* ne = realloc(m->present, newcap * sizeof(*ne)); + if (!nt || !np || !ne) { + /* realloc may have partially succeeded; keep whatever grew so the + * eventual free() is safe, but report OOM. */ + if (nt) + m->term = nt; + if (np) + m->pcrc = np; + if (ne) + m->present = ne; + return -ENOMEM; + } + /* Zero the newly added tail. */ + memset(nt + m->cap, 0, (newcap - m->cap) * sizeof(*nt)); + memset(np + m->cap, 0, (newcap - m->cap) * sizeof(*np)); + memset(ne + m->cap, 0, (newcap - m->cap) * sizeof(*ne)); + m->term = nt; + m->pcrc = np; + m->present = ne; + m->cap = newcap; + return 0; +} + +static int collect_visit(void* ctx, const waldump_entry_t* e) +{ + collect_ctx_t* cc = ctx; + member_log_t* m = cc->log; + if (e->lsn == 0) + return 0; /* LSN 0 is never a real committed entry */ + if (member_log_ensure(m, e->lsn) != 0) { + cc->oom = 1; + return 1; /* stop the scan */ + } + m->term[e->lsn] = e->term; + m->pcrc[e->lsn] = e->pcrc; + m->present[e->lsn] = 1; + return 0; +} + +static void member_log_free(member_log_t* m) +{ + free(m->term); + free(m->pcrc); + free(m->present); + m->term = NULL; + m->pcrc = NULL; + m->present = NULL; + m->cap = 0; +} + +/* Load one member's WAL into `*m`. Returns 0 or a negative errno. */ +static int load_member(const walcompare_member_t* src, member_log_t* m, FILE* out) +{ + memset(m, 0, sizeof(*m)); + + waldump_stats_t st; + memset(&st, 0, sizeof(st)); + collect_ctx_t cc = { .log = m, .oom = 0 }; + + /* Send the human dump to the caller's `out` (or a discard sink). The + * visitor harvests the (lsn -> term, pcrc) index from the same decode. */ + FILE* dump_sink = out ? out : fopen("/dev/null", "w"); + int own_sink = (out == NULL); + if (!dump_sink) + return -errno; + + int rc = waldump_path_visit(src->path, dump_sink, &st, collect_visit, &cc); + if (own_sink) + fclose(dump_sink); + + if (rc != 0) { + member_log_free(m); + return rc; /* hard load error (missing path / no segments) */ + } + if (cc.oom) { + member_log_free(m); + return -ENOMEM; + } + + m->recovered_tip = st.recovered_tip; + /* An explicit committed_lsn wins; 0 means "derive from the shared + * boundary" (filled by the caller after all members are loaded). */ + m->committed_lsn = src->committed_lsn; + return 0; +} + +int walcompare(const walcompare_member_t* members, int n, FILE* out, + walcompare_divergence_t* div) +{ + if (!members || n < 2) + return -EINVAL; + if (n > WALCMP_MAX_MEMBERS) + return -E2BIG; + + member_log_t logs[WALCMP_MAX_MEMBERS]; + int loaded = 0; + int rc = 0; + + for (int i = 0; i < n; i++) { + rc = load_member(&members[i], &logs[i], out); + if (rc != 0) { + if (out) + fprintf(out, "walcompare: cannot load member %d ('%s'): %s\n", i, + members[i].path ? members[i].path : "(null)", + strerror(-rc)); + goto done; + } + loaded = i + 1; + } + + /* Shared committed boundary = the minimum recovered tip across members: + * the largest committed prefix every member can be expected to hold. A + * member with an explicit committed_lsn keeps its own. */ + uint64_t shared = logs[0].recovered_tip; + for (int i = 1; i < n; i++) + if (logs[i].recovered_tip < shared) + shared = logs[i].recovered_tip; + + uint64_t boundary = 0; /* highest committed LSN compared across the cluster */ + for (int i = 0; i < n; i++) { + if (logs[i].committed_lsn == 0) + logs[i].committed_lsn = shared; + if (logs[i].committed_lsn > boundary) + boundary = logs[i].committed_lsn; + } + + if (out) + fprintf(out, + "== walcompare: %d members, committed boundary up to LSN %" PRIu64 + " ==\n", + n, boundary); + + /* Walk every committed LSN. At each LSN, compare the (term, pcrc) of the + * members whose own committed boundary covers it and that hold the entry. + * A member that does not hold a committed LSN it SHOULD (boundary covers it + * but the entry is absent) is itself a committed-prefix divergence: a + * committed entry is missing (R4.2 / R6.4). */ + for (uint64_t lsn = 1; lsn <= boundary; lsn++) { + int have = 0; /* members whose boundary covers this LSN */ + int present = 0; /* of those, how many actually hold it */ + uint64_t ref_term = 0; + uint32_t ref_pcrc = 0; + int have_ref = 0; + int divergent = 0; + + for (int i = 0; i < n; i++) { + if (lsn > logs[i].committed_lsn) + continue; /* this LSN is beyond member i's committed prefix */ + have++; + + int holds = (lsn < logs[i].cap) && logs[i].present[lsn]; + if (!holds) { + /* member i should hold this committed LSN but does not */ + divergent = 1; + continue; + } + present++; + uint64_t mt = match_term(logs[i].term[lsn]); + uint32_t mp = logs[i].pcrc[lsn]; + if (!have_ref) { + ref_term = mt; + ref_pcrc = mp; + have_ref = 1; + } else if (mt != ref_term || mp != ref_pcrc) { + divergent = 1; + } + } + + if (have < 2) + continue; /* nothing to compare at this LSN */ + if (!divergent) + continue; + + /* First divergent committed LSN — report and fail. */ + if (div) { + memset(div, 0, sizeof(*div)); + div->lsn = lsn; + div->n_members = n; + for (int i = 0; i < n && i < 16; i++) { + int holds = (lsn <= logs[i].committed_lsn) && + (lsn < logs[i].cap) && logs[i].present[lsn]; + div->member_present[i] = holds ? 1 : 0; + div->member_term[i] = holds ? match_term(logs[i].term[lsn]) : 0; + div->member_pcrc[i] = holds ? logs[i].pcrc[lsn] : 0; + } + } + if (out) { + fprintf(out, + "*** DIVERGENCE at committed LSN %" PRIu64 + " (compared %d, present %d) ***\n", + lsn, have, present); + for (int i = 0; i < n; i++) { + if (lsn > logs[i].committed_lsn) { + fprintf(out, + " member %d ('%s'): LSN beyond its committed prefix" + " (committed=%" PRIu64 ")\n", + i, members[i].path, logs[i].committed_lsn); + continue; + } + int holds = (lsn < logs[i].cap) && logs[i].present[lsn]; + if (holds) + fprintf(out, + " member %d ('%s'): term=%" PRIu64 " pcrc=%08x\n", i, + members[i].path, match_term(logs[i].term[lsn]), + logs[i].pcrc[lsn]); + else + fprintf(out, + " member %d ('%s'): MISSING committed entry\n", i, + members[i].path); + } + } + rc = WALCMP_DIVERGENT; + goto done; + } + + if (out) + fprintf(out, "walcompare: OK — committed prefixes identical\n"); + rc = WALCMP_OK; + +done: + for (int i = 0; i < loaded; i++) + member_log_free(&logs[i]); + return rc; +} diff --git a/src/controller/testsuite/walcompare.h b/src/controller/testsuite/walcompare.h new file mode 100644 index 00000000..4f368c74 --- /dev/null +++ b/src/controller/testsuite/walcompare.h @@ -0,0 +1,64 @@ +#pragma once +/* + * walcompare.h — cross-member WAL committed-prefix divergence detector + * (ha-term-safety R6.1/R6.2/R6.3). + * + * Reuses the canonical waldump.c WAL decoder (via waldump_path_visit) to index + * each member's log by `(lsn -> term, pcrc)`, then asserts that every member's + * COMMITTED prefix is identical by `(term, pcrc)` — not merely by LSN range. + * + * The decisive signal for bug-851: two members can report the same committed + * LSN over a contiguous, CRC-intact WAL yet hold DIFFERENT entries at the same + * LSN (same lsn, different term and/or payload). Comparing LSN ranges alone + * masks that; comparing `(term, pcrc)` per LSN surfaces it (R6.2). + * + * Strictly read-only — opens member WAL dirs O_RDONLY, mutates nothing. Safe to + * run against a live cluster's data dirs. + */ +#include +#include + +/* A single member's WAL source and its committed boundary. */ +typedef struct { + const char* path; /* WAL dir / data dir / seg-*.wal (waldump rules) */ + uint64_t committed_lsn; /* highest committed LSN for this member; 0 = + * "derive" (use the shared boundary, see below) */ +} walcompare_member_t; + +/* Identity of the first divergent entry, filled on a WALCMP_DIVERGENT result. */ +typedef struct { + uint64_t lsn; /* the diverging LSN within the committed prefix */ + int n_members; /* members compared at this LSN */ + uint64_t member_term[16]; /* per-member matched term at lsn */ + uint32_t member_pcrc[16]; /* per-member payload CRC at lsn */ + int member_present[16]; /* 1 if the member holds this LSN, else 0 */ +} walcompare_divergence_t; + +/* Result codes. */ +enum { + WALCMP_OK = 0, /* all members' committed prefixes are identical */ + WALCMP_DIVERGENT = 1, /* a committed-prefix divergence was found */ + /* negative values are -errno (a load/usage error: missing WAL, bad args) */ +}; + +/* Compare `n` members' committed WAL prefixes by `(lsn -> term, pcrc)`. + * + * Committed boundary: each member with `committed_lsn == 0` adopts the SHARED + * boundary = the minimum recovered tip (highest CRC-intact LSN) across all + * members, i.e. the largest prefix every member can be expected to hold. A + * member with an explicit `committed_lsn` uses that. An LSN in `[1..boundary]` + * is compared only against the members whose own boundary covers it; a member + * that is legitimately AHEAD (longer log) or behind the shared boundary is not + * flagged for the uncommitted tail it doesn't share (R6.1 is the committed + * prefix, not the full log). + * + * Legacy-zero term (R1.8) is mapped to WAL_LEGACY_TERM for matching, so two + * pre-feature entries are not spuriously divergent. + * + * On the FIRST divergent LSN, fills `*div` (when non-NULL) and prints a report + * to `out`, then returns WALCMP_DIVERGENT. Returns WALCMP_OK on identical + * committed prefixes, or a negative errno on a load/usage error. `out` may be + * NULL to suppress the report. + */ +int walcompare(const walcompare_member_t* members, int n, FILE* out, + walcompare_divergence_t* div); diff --git a/src/controller/testsuite/waldump.c b/src/controller/testsuite/waldump.c index 441d94ba..ce0fcd6b 100644 --- a/src/controller/testsuite/waldump.c +++ b/src/controller/testsuite/waldump.c @@ -197,7 +197,9 @@ static int collect_segments(const char* wal_dir, uint64_t* seg_lsns, int max) static void scan_segment(const char* seg_path, FILE* out, uint32_t* last_crc, uint64_t* prev_lsn, uint64_t* prev_submit_jobid, uint32_t* prev_submit_pcrc, - waldump_stats_t* st) + waldump_stats_t* st, + waldump_visit_fn visit, void* visit_ctx, + int* visit_stop) { int fd = open(seg_path, O_RDONLY); if (fd < 0) { @@ -282,6 +284,25 @@ static void scan_segment(const char* seg_path, FILE* out, ? crc32(0L, (const Bytef*)payload, hdr.payload_len) : 0; + /* Surface the CRC-valid entry to the visitor so a caller can index the + * log by (lsn -> term, pcrc) — the single canonical decoder, no second + * parser. */ + if (visit) { + waldump_entry_t e = { + .lsn = hdr.lsn, + .term = hdr.term, + .type = hdr.type, + .payload_len = hdr.payload_len, + .pcrc = pcrc, + }; + if (visit(visit_ctx, &e) != 0) { + if (visit_stop) + *visit_stop = 1; + free(payload); + break; /* visitor asked to stop */ + } + } + /* Per-entry line, with jobid/subtime for submits. */ char info[96] = ""; int is_dup = 0; @@ -347,6 +368,12 @@ static void scan_segment(const char* seg_path, FILE* out, /* Public entry point */ /* ------------------------------------------------------------------ */ int waldump_path(const char* path, FILE* out, waldump_stats_t* stats) +{ + return waldump_path_visit(path, out, stats, NULL, NULL); +} + +int waldump_path_visit(const char* path, FILE* out, waldump_stats_t* stats, + waldump_visit_fn visit, void* ctx) { if (!path || !out) return -EINVAL; @@ -417,12 +444,14 @@ int waldump_path(const char* path, FILE* out, waldump_stats_t* stats) uint64_t prev_lsn = 0; uint64_t prev_submit_jobid = 0; uint32_t prev_submit_pcrc = 0; - for (int i = 0; i < nseg; i++) { + int visit_stop = 0; + for (int i = 0; i < nseg && !visit_stop; i++) { if (single_seg[0] == '\0') snprintf(seg_path, sizeof(seg_path), WAL_SEG_FMT, wal_dir, seg_lsns[i]); scan_segment(seg_path, out, &last_crc, &prev_lsn, - &prev_submit_jobid, &prev_submit_pcrc, &st); + &prev_submit_jobid, &prev_submit_pcrc, &st, + visit, ctx, &visit_stop); } fprintf(out, "-- summary --\n"); diff --git a/src/controller/testsuite/waldump.h b/src/controller/testsuite/waldump.h index 16aaa046..2a34b37b 100644 --- a/src/controller/testsuite/waldump.h +++ b/src/controller/testsuite/waldump.h @@ -31,6 +31,24 @@ typedef struct { int crc_broken; /* nonzero if a CRC/chain break truncated a segment */ } waldump_stats_t; +/* One physically-present, CRC-valid WAL entry, surfaced to a visitor so a + * caller (e.g. walcompare) can index the log by `(lsn -> term, pcrc)` without + * re-implementing the segment/CRC parse. `pcrc` is the payload-only CRC32 (0 + * for an empty payload) — the same fingerprint waldump prints, isolating the + * blob from the header so a same-LSN content divergence is detectable. */ +typedef struct { + uint64_t lsn; + uint64_t term; /* raw on-disk term (legacy 0 kept verbatim) */ + uint32_t type; /* WAL entry type (wal_types.h) */ + uint32_t payload_len; /* payload byte count */ + uint32_t pcrc; /* CRC32 of payload only (0 if payload_len == 0) */ +} waldump_entry_t; + +/* Per-entry visitor. Invoked once per CRC-valid entry in ascending scan order + * (before the gap line is printed). Return 0 to continue, non-zero to stop the + * scan early. `ctx` is the caller's opaque pointer. */ +typedef int (*waldump_visit_fn)(void* ctx, const waldump_entry_t* e); + /* Dump every WAL entry under `path` to `out` (human-readable) and fill * `*stats` when non-NULL. `path` may be: * - a single seg-*.wal file, @@ -40,3 +58,12 @@ typedef struct { * path does not exist / no segments found). Gaps and CRC breaks are reported * through `out`/`stats`, NOT via the return code. */ int waldump_path(const char* path, FILE* out, waldump_stats_t* stats); + +/* Same as waldump_path, but also invokes `visit(ctx, &entry)` for every + * CRC-valid entry so a caller can collect the `(lsn -> term, pcrc)` index. + * `visit` may be NULL (then identical to waldump_path). `out` may be a sink + * the caller does not care about (e.g. a /dev/null FILE) — the human dump is + * still produced. The single canonical WAL decoder is reused; no second + * segment parser. */ +int waldump_path_visit(const char* path, FILE* out, waldump_stats_t* stats, + waldump_visit_fn visit, void* ctx);