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

test(ha-term-safety): cross-member walcompare divergence tool (task 5.3)

Browse source 
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.
This commit is contained in:
Rongsong Shen 2026-06-22 13:05:23 +00:00
parent d39500ba48
commit a166ecb45b
9 changed files with 842 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.kiro/specs/ha-term-safety/tasks.md
View file

@ -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] <memberA> <memberB> [...]` + 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.

4
src/controller/meson.build
View file

@ -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',

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

@ -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();
}

68
src/controller/testsuite/test.c
View file

@ -1,6 +1,7 @@
#include <argp.h>
#include <errno.h>
#include <getopt.h>
#include <inttypes.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
@ -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] <memberA> <memberB> [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] <memberA> <memberB> "
"[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) {

370
src/controller/testsuite/test_walcompare.c Normal file
View file

@ -0,0 +1,370 @@
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <zlib.h>
#include <unity/unity.h>
#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);
}

272
src/controller/testsuite/walcompare.c Normal file
View file

@ -0,0 +1,272 @@
#define _GNU_SOURCE
#include "walcompare.h"
#include <errno.h>
#include <inttypes.h>
#include <stdlib.h>
#include <string.h>
#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;
}

64
src/controller/testsuite/walcompare.h Normal file
View file

@ -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 <stdint.h>
#include <stdio.h>
/* 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);

35
src/controller/testsuite/waldump.c
View file

@ -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");

27
src/controller/testsuite/waldump.h
View file

@ -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);