forked from lei.chen/myxce
checkpoint: bug-851 root cause + raft TLA+ conformance spec family
Investigation & fixes (failover committed-job loss): - root-cause bug-851 on an owned 3-node cluster: chaotic failover causes silent committed-prefix divergence (WAL hdr.term always 0; two leaders per term; no (term,lsn) reconcile). Evidence in repro-851/ + INVESTIGATION. - live-validated fixes 848/849/850 (recovery-on-promotion, advertise real applied_lsn, persist-layer contiguity guard); add waldump WAL inspector. Specs (strict raft.tla conformance — library owns the log): - TLA-gap-analysis.md: impl vs raft/raft.tla, gaps G1-G6. - raft-log-replication, raft-commit-safety, raft-sim-conformance: extend src/lib/raft to own log/commit/verification via callbacks (specs generated, cross-spec review passed). - ha-term-safety: controller-side adapter + config marker (R1-R7); design done. - config-upgrade-snapshot: R8 descoped here (deps entity-bearing-snapshots, bug-857). - roadmap + buglog (851/854/857) + .wolf knowledge updated.
This commit is contained in:
parent
7d70820aac
commit
20e59d0b42
85 changed files with 10525 additions and 1993 deletions
|
|
@ -1356,6 +1356,20 @@ int cluster_init_phase2(cluster_t* c)
|
|||
return rc;
|
||||
}
|
||||
|
||||
/* bug-848: a (re)started member recovered its own WAL above — make it
|
||||
* ADVERTISE that recovered position for catch-up (sr->applied_lsn), not the
|
||||
* ha_init default of 0. Otherwise the primary streams the whole log and the
|
||||
* receiver overwrites already-consistent records instead of fetching only
|
||||
* the missing suffix (Raft incremental catch-up). */
|
||||
ha_sync_applied_lsn_from_recovery(&c->ha, &c->persist);
|
||||
INFO("HA: persist_recover done — recovered durable_floor=%" PRIu64
|
||||
" current_lsn=%" PRIu64 "; applied_lsn=%" PRIu64
|
||||
" advertised for catch-up",
|
||||
persist_durable_floor(&c->persist),
|
||||
(uint64_t)c->persist.current_lsn,
|
||||
(uint64_t)atomic_load_explicit(&c->ha.applied_lsn,
|
||||
memory_order_acquire));
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1570,6 +1570,16 @@ static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e)
|
|||
int prc = persist_wal_write_replicated(p, e->lsn, (uint32_t)e->entry_type,
|
||||
e->raw_payload,
|
||||
(uint32_t)e->raw_payload_len);
|
||||
if (prc == -EAGAIN) {
|
||||
/* Contiguity guard (bug-850): this entry is ahead of our log tail
|
||||
* (a forward gap) — e.g. a live broadcast that arrived before the
|
||||
* in-order catch-up of the prefix. NOT an error: drop it quietly and do
|
||||
* NOT ack, so the catch-up continuation re-streams from current_lsn and
|
||||
* the log fills in order. */
|
||||
TRACE("ha_apply_single_entry: lsn=%" PRIu64 " ahead of log tail — "
|
||||
"dropped (will arrive via in-order catch-up)", e->lsn);
|
||||
return prc;
|
||||
}
|
||||
if (prc != 0) {
|
||||
ERROR("ha_apply_single_entry: persist_wal_write_replicated failed for "
|
||||
"lsn=%" PRIu64 " type=%u rc=%d — not applying/acking",
|
||||
|
|
@ -1636,11 +1646,14 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
|
|||
|
||||
int rc = ha_apply_single_entry(p, sub);
|
||||
if (rc != 0) {
|
||||
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed "
|
||||
"for batch index %d lsn=%" PRIu64 " type=%u rc=%d "
|
||||
"— skipping entry",
|
||||
i, sub->lsn, (unsigned)sub->entry_type, rc);
|
||||
/* Task 3.2: no ACK for failed entries */
|
||||
/* -EAGAIN = contiguity gap (bug-850): entry ahead of our tail,
|
||||
* dropped quietly — not an error (catch-up will backfill). */
|
||||
if (rc != -EAGAIN)
|
||||
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed "
|
||||
"for batch index %d lsn=%" PRIu64 " type=%u rc=%d "
|
||||
"— skipping entry",
|
||||
i, sub->lsn, (unsigned)sub->entry_type, rc);
|
||||
/* Task 3.2: no ACK for failed/dropped entries */
|
||||
continue;
|
||||
}
|
||||
|
||||
|
|
@ -1662,9 +1675,12 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
|
|||
/* ── Single-entry mode (task 2.1 backward-compatible path) ────────────── */
|
||||
int rc = ha_apply_single_entry(p, entry);
|
||||
if (rc != 0) {
|
||||
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed for "
|
||||
"lsn=%" PRIu64 " type=%u rc=%d — skipping entry",
|
||||
entry->lsn, (unsigned)entry->entry_type, rc);
|
||||
/* -EAGAIN = contiguity gap (bug-850): dropped quietly, backfilled by
|
||||
* the in-order catch-up; any other rc is a real failure. */
|
||||
if (rc != -EAGAIN)
|
||||
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed for "
|
||||
"lsn=%" PRIu64 " type=%u rc=%d — skipping entry",
|
||||
entry->lsn, (unsigned)entry->entry_type, rc);
|
||||
/* Task 3.2: no ACK on apply failure */
|
||||
return rc;
|
||||
}
|
||||
|
|
@ -2600,6 +2616,23 @@ uint64_t ha_candidate_advertise_lsn(persist_t *p)
|
|||
return persist_durable_floor(p);
|
||||
}
|
||||
|
||||
/* bug-848: see ha.h. After local WAL recovery, advertise the recovered durable
|
||||
* tip for catch-up so the primary streams only the missing suffix instead of
|
||||
* re-streaming + overwriting consistent records. Monotonic: never lowers a
|
||||
* higher applied_lsn (e.g. if catch-up already advanced it). */
|
||||
void ha_sync_applied_lsn_from_recovery(ha_state_t *ha, persist_t *p)
|
||||
{
|
||||
if (!ha || !p)
|
||||
return;
|
||||
|
||||
uint64_t recovered = persist_durable_floor(p);
|
||||
uint64_t cur = atomic_load_explicit(&ha->applied_lsn,
|
||||
memory_order_acquire);
|
||||
if (recovered > cur)
|
||||
atomic_store_explicit(&ha->applied_lsn, recovered,
|
||||
memory_order_release);
|
||||
}
|
||||
|
||||
/* ── Vote Handler (Req 2.2–2.6, 6.1, 6.2, 6.4, 9.1) ────────────────────────
|
||||
*
|
||||
* handle_request_vote: process an incoming OP_requestVote RPC.
|
||||
|
|
|
|||
|
|
@ -793,6 +793,21 @@ void ha_vote_self_freshness(ha_state_t *ha,
|
|||
*/
|
||||
uint64_t ha_candidate_advertise_lsn(persist_t *p);
|
||||
|
||||
/* ha_sync_applied_lsn_from_recovery (bug-848) — after local WAL recovery on
|
||||
* (re)start, set ha->applied_lsn to the position the member actually recovered
|
||||
* (persist_durable_floor(p)), instead of the ha_init default of 0.
|
||||
*
|
||||
* Why: a restarted standby advertises ha->applied_lsn to the primary in its
|
||||
* state report; the primary then streams the catch-up range [applied_lsn+1 ..
|
||||
* committed]. If a node that already holds [1..N] on disk reports 0, the
|
||||
* primary re-streams the WHOLE log and the receiver O_TRUNC-overwrites already-
|
||||
* consistent records (instead of Raft incremental catch-up of only the missing
|
||||
* suffix). This mirrors the value the election/PreVote path already trusts
|
||||
* (ha_candidate_advertise_lsn == persist_durable_floor), so the catch-up and
|
||||
* vote-freshness advertisements finally agree. NULL-safe; never lowers an
|
||||
* already-higher applied_lsn (monotonic). */
|
||||
void ha_sync_applied_lsn_from_recovery(ha_state_t *ha, persist_t *p);
|
||||
|
||||
/*
|
||||
* ha_elect_evaluate_and_signal: evaluate accumulated ha_elect_ctx_t votes.
|
||||
* Sets ctx->result (1=primary, 0=standby) and posts ctx->done semaphore.
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ controller = executable('xce_controller',
|
|||
link_args: wrap_link_args,
|
||||
install: true)
|
||||
|
||||
xce_test_src = [ 'testsuite/test.c']
|
||||
xce_test_src = [ 'testsuite/test.c', 'testsuite/waldump.c']
|
||||
|
||||
readline_dep = dependency('readline')
|
||||
ncurses_dep = dependency('ncurses')
|
||||
|
|
@ -67,6 +67,8 @@ xce_test_deps = [
|
|||
history_dep,
|
||||
ncurses_dep,
|
||||
tinylog_dep,
|
||||
zlib_dep, # waldump: crc32 WAL validation
|
||||
c_capnproto_dep, # waldump: decode JOB_SUBMIT jobid
|
||||
]
|
||||
xce_test = executable('xce_test',
|
||||
xce_test_src,
|
||||
|
|
@ -136,6 +138,8 @@ controller_unit_test_src = [
|
|||
'testsuite/test_ctrlstatus.c',
|
||||
'testsuite/test_ha_heartbeat_committed.c',
|
||||
'testsuite/test_ha_vote_floor_heartbeat.c',
|
||||
'testsuite/test_waldump.c',
|
||||
'testsuite/waldump.c',
|
||||
'ctrlstatus.c',
|
||||
'commit.c',
|
||||
'job.c',
|
||||
|
|
|
|||
|
|
@ -317,6 +317,31 @@ skip_election:;
|
|||
if (role == CTRL_STANDBY || role == CTRL_INIT) {
|
||||
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
|
||||
memory_order_release);
|
||||
|
||||
/* bug-849: a follower that REJOINS an existing cluster skips its own
|
||||
* election (a primary already exists), so the election-result path
|
||||
* that calls cluster_init_phase2 — which builds resources, runs
|
||||
* persist_recover, and (bug-848) advertises the recovered
|
||||
* applied_lsn — NEVER runs on this path. The node then comes up on
|
||||
* an un-recovered WAL (current_lsn=1, applied_lsn=0), the primary
|
||||
* re-streams the whole log over its consistent records, and the
|
||||
* committed prefix is lost. Run phase2 HERE too (idempotent via
|
||||
* phase2_done, same once-per-process contract as the election
|
||||
* callers) so EVERY boot path recovers its own WAL and advertises
|
||||
* its real position BEFORE cluster_run starts accepting replication.
|
||||
* Recovery is decoupled from the primary's config push: a restarted
|
||||
* node already loaded its config from disk pre-election. */
|
||||
if (!c.ha.phase2_done) {
|
||||
int prc = cluster_init_phase2(&c);
|
||||
if (prc != 0) {
|
||||
WARN("myapp_run: cluster_init_phase2 on standby rejoin "
|
||||
"failed (rc=%d) — WAL not recovered; falling back to "
|
||||
"election-path recovery", prc);
|
||||
} else {
|
||||
c.ha.phase2_done = 1;
|
||||
}
|
||||
}
|
||||
|
||||
INFO("myapp_run: CTRL_STANDBY — waiting for primary to connect and push config");
|
||||
/* Arm heartbeat watchdog even when we skipped the election.
|
||||
* If the known primary never connects within heartbeat_timeout_ms,
|
||||
|
|
|
|||
|
|
@ -688,6 +688,23 @@ int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* Contiguity guard (bug-850): REFUSE a forward GAP. current_lsn is the
|
||||
* next-expected LSN, so a replicated entry with lsn > current_lsn would
|
||||
* leave [current_lsn .. lsn-1] missing AND advance current_lsn past that
|
||||
* hole — after which the idempotency guard above silently DROPS the later
|
||||
* in-order catch-up of the prefix (the failover prefix-loss mechanism).
|
||||
* Append only the contiguous next entry (lsn == current_lsn); reject a gap
|
||||
* without writing or acking so the ACK-driven catch-up continuation
|
||||
* re-streams from current_lsn and the log fills in order. This is Raft
|
||||
* log-matching: append entry N only when the log already holds N-1. A
|
||||
* behind standby that keeps receiving live broadcasts simply drops the
|
||||
* out-of-order ones until catch-up backfills the gap (no loss). -EAGAIN is
|
||||
* the "retry via catch-up" signal the apply path treats as a quiet skip. */
|
||||
if (lsn > p->current_lsn) {
|
||||
pthread_mutex_unlock(&p->lock);
|
||||
return -EAGAIN;
|
||||
}
|
||||
|
||||
/* Open first segment lazily. open_new_segment names the file from
|
||||
* current_lsn — for a fresh standby receiving lsn 1 first, current_lsn is
|
||||
* 1 (reserve-lsn-0), so the segment is seg-...001.wal. */
|
||||
|
|
|
|||
|
|
@ -63,6 +63,7 @@ extern void test_commit_durability(void);
|
|||
extern void test_ctrlstatus(void);
|
||||
extern void test_ha_heartbeat_committed(void);
|
||||
extern void test_ha_vote_floor_heartbeat(void);
|
||||
extern void test_waldump(void);
|
||||
|
||||
void setUp(void) { return; }
|
||||
|
||||
|
|
@ -195,6 +196,7 @@ int main(int argc, char* argv[])
|
|||
TEST(test_ctrlstatus);
|
||||
TEST(test_ha_heartbeat_committed);
|
||||
TEST(test_ha_vote_floor_heartbeat);
|
||||
TEST(test_waldump);
|
||||
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@
|
|||
#include "delay.h"
|
||||
#include "idgen.h"
|
||||
#include "preason.h"
|
||||
#include "waldump.h" /* offline WAL inspector (cmd_waldump) */
|
||||
#include "xce/cluster.h" /* xce_parse_peers, xce_cluster_config_t */
|
||||
#include "xce/types.h"
|
||||
|
||||
|
|
@ -122,6 +123,7 @@ extern int cmd_expect_jobs(int, char*[]);
|
|||
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*[]);
|
||||
static int worker_house_keeping(void*);
|
||||
|
||||
/* Session verdict / loop state (xce-test-e2e-primitives task 1.1).
|
||||
|
|
@ -179,6 +181,7 @@ static command_t top_commands[] = {
|
|||
{ "client", cmd_client, "In client context" },
|
||||
{ "set", cmd_set, "Show/Set settings" },
|
||||
{ "echo", cmd_echo, "Print a message" },
|
||||
{ "waldump", cmd_waldump, "Inspect WAL segments" },
|
||||
{ "quit", cmd_quit, "Quit xce test" },
|
||||
{ (char*)NULL, (cmd_func_t)NULL, (char*)NULL }
|
||||
};
|
||||
|
|
@ -216,6 +219,7 @@ static command_t client_commands[] = {
|
|||
{ "wait-primary", cmd_wait_primary,
|
||||
"wait until a primary serves [timeout_ms]" },
|
||||
{ "ctrlstatus", cmd_ctrlstatus, "show controller/cluster status" },
|
||||
{ "waldump", cmd_waldump, "Inspect WAL segments" },
|
||||
{ "shutdown", cmd_shutdown, "shutdown xce controller" },
|
||||
{ "jobrun", cmd_jobrun, "run a job" },
|
||||
{ "jobsub", cmd_jobsub, "submit a job" },
|
||||
|
|
@ -683,6 +687,35 @@ int cmd_echo(int argc, char* argv[])
|
|||
return 0;
|
||||
}
|
||||
|
||||
/* waldump <path> [<path> ...] — offline WAL inspector for failover/durability
|
||||
* triage. Each path may be a seg-*.wal file, a wal/ dir, or a node/data dir.
|
||||
* Read-only: lists every entry, decodes JOB_SUBMIT jobids, and flags LSN gaps
|
||||
* (a missing committed prefix is THE failover job-loss signal) plus the CRC
|
||||
* recovered tip. Returns nonzero if any path failed or any gap was found, so
|
||||
* it doubles as an assertion in scripted E2E runs. */
|
||||
int cmd_waldump(int argc, char* argv[])
|
||||
{
|
||||
if (argc < 2) {
|
||||
fprintf(stderr,
|
||||
"usage: waldump <wal_dir|data_dir|seg_file> [more...]\n");
|
||||
return -1;
|
||||
}
|
||||
|
||||
int rc = 0;
|
||||
for (int i = 1; i < argc; i++) {
|
||||
waldump_stats_t st;
|
||||
int r = waldump_path(argv[i], stdout, &st);
|
||||
if (r != 0) {
|
||||
fprintf(stderr, "waldump: %s: %s\n", argv[i], strerror(-r));
|
||||
rc = -1;
|
||||
continue;
|
||||
}
|
||||
if (st.n_gaps > 0 || st.crc_broken)
|
||||
rc = -1; /* surface the defect as a nonzero command verdict */
|
||||
}
|
||||
return rc;
|
||||
}
|
||||
|
||||
int cmd_up(int argc, char* argv[])
|
||||
{
|
||||
if (active_ctx->parent != NULL) {
|
||||
|
|
|
|||
|
|
@ -440,8 +440,23 @@ static void test_fact_hook_append_is_retained_and_nonblocking(void)
|
|||
"the uncommitted fact must be pending (lsn > committed_lsn), "
|
||||
"never confirmed early");
|
||||
|
||||
/* ...and a covering ACK later commits the retained entry — engine
|
||||
* accounting advances committed_lsn over it, no redelivery needed */
|
||||
/* ha-commit-durability 2.2: the primary's self-contribution to the commit
|
||||
* majority is now capped at durable_lsn, which the demand-driven
|
||||
* group-commit flusher advances asynchronously after fdatasync. Force the
|
||||
* fact durable here (synchronous fdatasync) so the covering ACK below
|
||||
* deterministically completes the majority (durable self + standby ACK);
|
||||
* a bare synchronous read would otherwise race the flusher and observe
|
||||
* durable_lsn == 0 → self-cap 0 → committed_lsn unchanged. This mirrors the
|
||||
* durable_lsn priming done for the in-memory stub fixtures in
|
||||
* barrier_setup_vcpus. */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, persist_flush(g_fp),
|
||||
"fact must fdatasync (durable) before the covering ACK");
|
||||
TEST_ASSERT_TRUE_MESSAGE(persist_durable_lsn(g_fp) >= fact_lsn,
|
||||
"durable_lsn must cover the fact after persist_flush");
|
||||
|
||||
/* ...and a covering ACK then commits the now-durable retained entry —
|
||||
* repl_on_ack drives commit_on_ack synchronously, advancing committed_lsn
|
||||
* over it, no redelivery needed */
|
||||
repl_on_ack(&g_fp->repl, &g_fep_a, fact_lsn);
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
|
||||
fact_lsn, atomic_load(&g_fp->committed_lsn),
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@
|
|||
#include <unity/unity.h>
|
||||
|
||||
#include "config.h"
|
||||
#include "ha.h" /* ha_state_t, ha_sync_applied_lsn_from_recovery (bug-848) */
|
||||
#include "persist.h"
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
|
@ -248,6 +249,66 @@ static void test_recover_picks_latest_snapshot(void)
|
|||
/* Test suite entry point */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* bug-848: a recovered member must advertise its REAL last LSN, not 0 */
|
||||
/* (else the primary re-streams the whole log and overwrites consistent */
|
||||
/* records instead of fetching only the missing suffix). */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
static void test_recovered_member_advertises_real_applied_lsn(void)
|
||||
{
|
||||
char tmp[256];
|
||||
strcpy(tmp, "/tmp/xce_rec848_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
|
||||
|
||||
/* Phase 1: write several WAL entries, flush, stop (simulate a node that
|
||||
* served writes, then was killed). */
|
||||
{
|
||||
persist_t* p = make_fresh_persist(tmp);
|
||||
TEST_ASSERT_NOT_NULL(p);
|
||||
for (int i = 0; i < 5; i++)
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
persist_wal_async(p, WAL_JOB_UPDATE, "z", 1));
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
|
||||
teardown_persist(p);
|
||||
}
|
||||
|
||||
/* Phase 2: restart — fresh persist recovers the WAL; a freshly ha-init'd
|
||||
* member (applied_lsn == 0) must be synced to the recovered durable tip. */
|
||||
{
|
||||
persist_t* p2 = make_fresh_persist(tmp);
|
||||
TEST_ASSERT_NOT_NULL(p2);
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
|
||||
|
||||
uint64_t recovered_floor = persist_durable_floor(p2);
|
||||
TEST_ASSERT_TRUE(recovered_floor > 0); /* WAL was non-empty */
|
||||
|
||||
ha_state_t ha;
|
||||
memset(&ha, 0, sizeof(ha));
|
||||
atomic_store_explicit(&ha.applied_lsn, 0ULL, memory_order_release);
|
||||
|
||||
ha_sync_applied_lsn_from_recovery(&ha, p2);
|
||||
|
||||
uint64_t advertised =
|
||||
atomic_load_explicit(&ha.applied_lsn, memory_order_acquire);
|
||||
/* The bug: this used to stay 0 → primary re-streams [1..committed]. */
|
||||
TEST_ASSERT_TRUE(advertised > 0);
|
||||
TEST_ASSERT_EQUAL_UINT64(recovered_floor, advertised);
|
||||
|
||||
/* Monotonic: never lowers an already-higher applied_lsn. */
|
||||
atomic_store_explicit(&ha.applied_lsn, recovered_floor + 100,
|
||||
memory_order_release);
|
||||
ha_sync_applied_lsn_from_recovery(&ha, p2);
|
||||
TEST_ASSERT_EQUAL_UINT64(
|
||||
recovered_floor + 100,
|
||||
atomic_load_explicit(&ha.applied_lsn, memory_order_acquire));
|
||||
|
||||
teardown_persist(p2);
|
||||
}
|
||||
|
||||
printf("test_recovered_member_advertises_real_applied_lsn: PASS\n");
|
||||
}
|
||||
|
||||
void test_persist_recovery(void)
|
||||
{
|
||||
RUN_TEST(test_recover_from_empty_state);
|
||||
|
|
@ -255,4 +316,5 @@ void test_persist_recovery(void)
|
|||
RUN_TEST(test_recover_replays_wal_after_checkpoint);
|
||||
RUN_TEST(test_recover_rejects_corrupt_snapshot);
|
||||
RUN_TEST(test_recover_picks_latest_snapshot);
|
||||
RUN_TEST(test_recovered_member_advertises_real_applied_lsn);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -395,6 +395,50 @@ static void test_repl_write_dedup_idempotent(void)
|
|||
printf("test_repl_write_dedup_idempotent: PASS\n");
|
||||
}
|
||||
|
||||
/* bug-850 contiguity guard: a replicated entry whose lsn is BEYOND the next-
|
||||
* expected current_lsn (a forward gap) must be REJECTED with -EAGAIN without
|
||||
* writing or advancing current_lsn — otherwise it would advance the watermark
|
||||
* past the hole and the later in-order catch-up of the prefix would be silently
|
||||
* dropped by the idempotency guard (the failover prefix-loss mechanism). The
|
||||
* entry becomes writable only once the gap is filled contiguously. */
|
||||
static void test_repl_write_rejects_forward_gap(void)
|
||||
{
|
||||
char tmpdir[256];
|
||||
persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir));
|
||||
|
||||
/* Contiguous first entry: lsn 1 == current_lsn → written. */
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_wal_write_replicated(
|
||||
p, 1, WAL_JOB_UPDATE, "a", 1));
|
||||
TEST_ASSERT_EQUAL_UINT64(2, p->current_lsn);
|
||||
|
||||
/* Forward gap: lsn 5 > current_lsn(2) — like a live broadcast arriving
|
||||
* ahead of catch-up. Must be refused (-EAGAIN) and leave current_lsn put. */
|
||||
int rc_gap = persist_wal_write_replicated(p, 5, WAL_JOB_UPDATE, "e", 1);
|
||||
TEST_ASSERT_EQUAL_INT(-EAGAIN, rc_gap);
|
||||
TEST_ASSERT_EQUAL_UINT64(2, p->current_lsn); /* did NOT jump past the gap */
|
||||
|
||||
/* Fill the gap in order: 2,3,4 contiguous → accepted. */
|
||||
for (uint64_t lsn = 2; lsn <= 4; lsn++)
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_wal_write_replicated(
|
||||
p, lsn, WAL_JOB_UPDATE, "x", 1));
|
||||
TEST_ASSERT_EQUAL_UINT64(5, p->current_lsn);
|
||||
|
||||
/* Now lsn 5 is contiguous (== current_lsn) → accepted. */
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_wal_write_replicated(
|
||||
p, 5, WAL_JOB_UPDATE, "e", 1));
|
||||
TEST_ASSERT_EQUAL_UINT64(6, p->current_lsn);
|
||||
|
||||
destroy_test_persist(p);
|
||||
|
||||
/* Recovery yields a clean contiguous chain [1..5] → current_lsn=6. */
|
||||
persist_t* p2 = reopen_test_persist(tmpdir);
|
||||
TEST_ASSERT_TRUE(persist_recover(p2) >= 0);
|
||||
TEST_ASSERT_EQUAL_UINT64(6, p2->current_lsn);
|
||||
destroy_test_persist(p2);
|
||||
|
||||
printf("test_repl_write_rejects_forward_gap: PASS\n");
|
||||
}
|
||||
|
||||
/* WAL_CHECKPOINT must never be persisted via the replicated-write path
|
||||
* (standbys never receive it; it must be rejected, not written). */
|
||||
static void test_repl_write_rejects_checkpoint(void)
|
||||
|
|
@ -428,5 +472,6 @@ void test_repl_apply(void)
|
|||
RUN_TEST(test_repl_write_persists_and_recovers);
|
||||
RUN_TEST(test_repl_write_crc_chain_recovers);
|
||||
RUN_TEST(test_repl_write_dedup_idempotent);
|
||||
RUN_TEST(test_repl_write_rejects_forward_gap);
|
||||
RUN_TEST(test_repl_write_rejects_checkpoint);
|
||||
}
|
||||
|
|
|
|||
295
src/controller/testsuite/test_waldump.c
Normal file
295
src/controller/testsuite/test_waldump.c
Normal file
|
|
@ -0,0 +1,295 @@
|
|||
#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 "controller.capnp.h" /* Job encode for the JOB_SUBMIT decode test */
|
||||
#include "wal_types.h"
|
||||
#include "waldump.h"
|
||||
|
||||
/* ------------------------------------------------------------------ *
|
||||
* Test harness: hand-write raw WAL segments so the test fully controls
|
||||
* LSNs (to inject gaps), the CRC chain (to inject a break), and a real
|
||||
* Cap'n Proto Job payload (to exercise the submit jobid decode). This
|
||||
* also pins the on-disk WAL format independently of persist.c.
|
||||
* ------------------------------------------------------------------ */
|
||||
|
||||
static char g_dir[256];
|
||||
|
||||
/* Append one entry to `fp`, chaining prev_crc from *last_crc and updating it.
|
||||
* Mirrors persist.c's append: CRC32 over WAL_HDR_CRC_SIZE header bytes + payload. */
|
||||
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);
|
||||
}
|
||||
|
||||
/* Encode a minimal Cap'n Proto Job carrying `jobid`+`subtime` (what a real
|
||||
* WAL_JOB_SUBMIT payload looks like). Returns malloc'd bytes; caller frees.
|
||||
* subtime lets a test make two payloads byte-identical (same subtime) or
|
||||
* distinct (different subtime) to exercise bug-845 dup detection. */
|
||||
static uint8_t* encode_job(uint64_t jobid, uint64_t subtime, size_t* out_len)
|
||||
{
|
||||
struct capn c;
|
||||
capn_init_malloc(&c);
|
||||
capn_ptr root = capn_root(&c);
|
||||
Job_ptr jp = new_Job(root.seg);
|
||||
struct Job cj;
|
||||
memset(&cj, 0, sizeof(cj));
|
||||
cj.jobid = jobid;
|
||||
cj.subtime = subtime;
|
||||
cj.state = 1;
|
||||
write_Job(&cj, jp);
|
||||
capn_setp(root, 0, jp.p);
|
||||
|
||||
size_t sz = capn_size(&c) + 64;
|
||||
uint8_t* buf = malloc(sz);
|
||||
int64_t n = capn_write_mem(&c, buf, sz, 0);
|
||||
capn_free(&c);
|
||||
*out_len = (size_t)n;
|
||||
return buf;
|
||||
}
|
||||
|
||||
static const char* seg_path(uint64_t start_lsn, char* out, size_t n)
|
||||
{
|
||||
snprintf(out, n, "%s/seg-%016lx.wal", g_dir, (unsigned long)start_lsn);
|
||||
return out;
|
||||
}
|
||||
|
||||
/* setUp/tearDown are owned by the shared controller_test runner, so each
|
||||
* test mints its own fresh temp WAL dir here instead. */
|
||||
static void fresh_dir(void)
|
||||
{
|
||||
strcpy(g_dir, "/tmp/xce_waldump_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(g_dir));
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* A healthy, contiguous log from LSN 1: no gaps, tip == last entry. */
|
||||
/* ------------------------------------------------------------------ */
|
||||
static void test_contiguous_log_has_no_gaps(void)
|
||||
{
|
||||
fresh_dir();
|
||||
char path[512];
|
||||
FILE* fp = fopen(seg_path(1, path, sizeof(path)), "wb");
|
||||
TEST_ASSERT_NOT_NULL(fp);
|
||||
|
||||
uint32_t last = WAL_CRC_NONE;
|
||||
put_entry(fp, 0, 1, WAL_NOOP, NULL, 0, &last, 1);
|
||||
size_t p2, p3;
|
||||
uint8_t* j2 = encode_job(1001, 100, &p2); /* distinct jobs → no dup */
|
||||
uint8_t* j3 = encode_job(1002, 100, &p3);
|
||||
put_entry(fp, 0, 2, WAL_JOB_SUBMIT, j2, (uint32_t)p2, &last, 0);
|
||||
put_entry(fp, 0, 3, WAL_JOB_SUBMIT, j3, (uint32_t)p3, &last, 0);
|
||||
free(j2);
|
||||
free(j3);
|
||||
fclose(fp);
|
||||
|
||||
waldump_stats_t st;
|
||||
int rc = waldump_path(g_dir, stdout, &st);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT(3, st.n_entries);
|
||||
TEST_ASSERT_EQUAL_UINT(2, st.n_submits);
|
||||
TEST_ASSERT_EQUAL_UINT(0, st.n_gaps);
|
||||
TEST_ASSERT_EQUAL_UINT(0, st.n_dup_submits); /* distinct jobids */
|
||||
TEST_ASSERT_EQUAL_UINT64(1, st.first_lsn);
|
||||
TEST_ASSERT_EQUAL_UINT64(3, st.last_lsn);
|
||||
TEST_ASSERT_EQUAL_UINT64(3, st.recovered_tip);
|
||||
TEST_ASSERT_EQUAL_INT(0, st.crc_broken);
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* The failover bug signature: log starts at a high LSN (empty-rejoin */
|
||||
/* caught up from the watermark) so the committed prefix is missing. */
|
||||
/* ------------------------------------------------------------------ */
|
||||
static void test_missing_prefix_flagged_as_gap(void)
|
||||
{
|
||||
fresh_dir();
|
||||
char path[512];
|
||||
/* seg file named from its first LSN, 13 (like ctrl-3 in the live repro) */
|
||||
FILE* fp = fopen(seg_path(13, path, sizeof(path)), "wb");
|
||||
TEST_ASSERT_NOT_NULL(fp);
|
||||
|
||||
uint32_t last = WAL_CRC_NONE;
|
||||
put_entry(fp, 0, 13, WAL_NOOP, NULL, 0, &last, 1);
|
||||
size_t p14, p15;
|
||||
uint8_t* j14 = encode_job(2001, 100, &p14);
|
||||
uint8_t* j15 = encode_job(2002, 100, &p15);
|
||||
put_entry(fp, 0, 14, WAL_JOB_SUBMIT, j14, (uint32_t)p14, &last, 0);
|
||||
put_entry(fp, 0, 15, WAL_JOB_SUBMIT, j15, (uint32_t)p15, &last, 0);
|
||||
free(j14);
|
||||
free(j15);
|
||||
fclose(fp);
|
||||
|
||||
waldump_stats_t st;
|
||||
int rc = waldump_path(g_dir, stdout, &st);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT(3, st.n_entries);
|
||||
TEST_ASSERT_EQUAL_UINT64(13, st.first_lsn);
|
||||
/* prefix [1..12] absent → exactly one gap */
|
||||
TEST_ASSERT_EQUAL_UINT(1, st.n_gaps);
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* bug-845: each job double-appended (same jobid at adjacent LSN). The */
|
||||
/* identical-payload case = a literal byte-for-byte double-write. */
|
||||
/* ------------------------------------------------------------------ */
|
||||
static void test_duplicate_submit_detection(void)
|
||||
{
|
||||
fresh_dir();
|
||||
char path[512];
|
||||
FILE* fp = fopen(seg_path(1, path, sizeof(path)), "wb");
|
||||
TEST_ASSERT_NOT_NULL(fp);
|
||||
|
||||
uint32_t last = WAL_CRC_NONE;
|
||||
put_entry(fp, 0, 1, WAL_NOOP, NULL, 0, &last, 1);
|
||||
/* Same jobid + same subtime → byte-identical payload (literal dup). */
|
||||
size_t pa, pb;
|
||||
uint8_t* ja = encode_job(7001, 555, &pa);
|
||||
uint8_t* jb = encode_job(7001, 555, &pb);
|
||||
put_entry(fp, 0, 2, WAL_JOB_SUBMIT, ja, (uint32_t)pa, &last, 0);
|
||||
put_entry(fp, 0, 3, WAL_JOB_SUBMIT, jb, (uint32_t)pb, &last, 0);
|
||||
/* A second job, also doubled but with a DIFFERING subtime per copy. */
|
||||
size_t pc, pd;
|
||||
uint8_t* jc = encode_job(7002, 600, &pc);
|
||||
uint8_t* jd = encode_job(7002, 601, &pd);
|
||||
put_entry(fp, 0, 4, WAL_JOB_SUBMIT, jc, (uint32_t)pc, &last, 0);
|
||||
put_entry(fp, 0, 5, WAL_JOB_SUBMIT, jd, (uint32_t)pd, &last, 0);
|
||||
free(ja); free(jb); free(jc); free(jd);
|
||||
fclose(fp);
|
||||
|
||||
waldump_stats_t st;
|
||||
int rc = waldump_path(g_dir, stdout, &st);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT(4, st.n_submits);
|
||||
TEST_ASSERT_EQUAL_UINT(2, st.n_dup_submits); /* both pairs flagged */
|
||||
TEST_ASSERT_EQUAL_UINT(1, st.n_dup_identical); /* only the first is byte-identical */
|
||||
TEST_ASSERT_EQUAL_UINT(0, st.n_gaps);
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* An interior hole (a skipped LSN between two present entries). */
|
||||
/* ------------------------------------------------------------------ */
|
||||
static void test_interior_gap_flagged(void)
|
||||
{
|
||||
fresh_dir();
|
||||
/* Two segments: [1..2] then [5..6]; 3,4 are missing. */
|
||||
char p1[512], p2[512];
|
||||
FILE* f1 = fopen(seg_path(1, p1, sizeof(p1)), "wb");
|
||||
uint32_t last = WAL_CRC_NONE;
|
||||
put_entry(f1, 0, 1, WAL_NOOP, NULL, 0, &last, 1);
|
||||
put_entry(f1, 0, 2, WAL_JOB_UPDATE, "x", 1, &last, 0);
|
||||
fclose(f1);
|
||||
|
||||
FILE* f2 = fopen(seg_path(5, p2, sizeof(p2)), "wb");
|
||||
uint32_t last2 = WAL_CRC_NONE;
|
||||
put_entry(f2, 0, 5, WAL_JOB_UPDATE, "y", 1, &last2, 1);
|
||||
put_entry(f2, 0, 6, WAL_JOB_UPDATE, "z", 1, &last2, 0);
|
||||
fclose(f2);
|
||||
|
||||
waldump_stats_t st;
|
||||
int rc = waldump_path(g_dir, stdout, &st);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT(4, st.n_entries);
|
||||
TEST_ASSERT_EQUAL_UINT(2, st.n_segments);
|
||||
TEST_ASSERT_EQUAL_UINT64(1, st.first_lsn);
|
||||
TEST_ASSERT_EQUAL_UINT64(6, st.last_lsn);
|
||||
TEST_ASSERT_EQUAL_UINT(1, st.n_gaps); /* the 3..4 hole */
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* A CRC break truncates the recovered tip below the last physical LSN. */
|
||||
/* ------------------------------------------------------------------ */
|
||||
static void test_crc_break_sets_recovered_tip(void)
|
||||
{
|
||||
fresh_dir();
|
||||
char path[512];
|
||||
FILE* fp = fopen(seg_path(1, path, sizeof(path)), "wb");
|
||||
uint32_t last = WAL_CRC_NONE;
|
||||
put_entry(fp, 0, 1, WAL_NOOP, NULL, 0, &last, 1);
|
||||
put_entry(fp, 0, 2, WAL_JOB_UPDATE, "ok", 2, &last, 0);
|
||||
fclose(fp);
|
||||
|
||||
/* Corrupt the second entry's stored CRC in place. */
|
||||
fp = fopen(path, "r+b");
|
||||
wal_entry_hdr_t h;
|
||||
long off2 = sizeof(wal_entry_hdr_t) + 0; /* entry 1 has 0 payload */
|
||||
fseek(fp, off2, SEEK_SET);
|
||||
fread(&h, sizeof(h), 1, fp);
|
||||
h.crc ^= 0xdeadbeefu; /* break it */
|
||||
fseek(fp, off2, SEEK_SET);
|
||||
fwrite(&h, sizeof(h), 1, fp);
|
||||
fclose(fp);
|
||||
|
||||
waldump_stats_t st;
|
||||
int rc = waldump_path(g_dir, stdout, &st);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT64(1, st.recovered_tip); /* stops before the bad LSN 2 */
|
||||
TEST_ASSERT_NOT_EQUAL(0, st.crc_broken);
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Missing path is a hard error. */
|
||||
/* ------------------------------------------------------------------ */
|
||||
static void test_missing_path_errors(void)
|
||||
{
|
||||
waldump_stats_t st;
|
||||
int rc = waldump_path("/tmp/xce_waldump_does_not_exist_zzz", stdout, &st);
|
||||
TEST_ASSERT_LESS_THAN_INT(0, rc);
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* A non-WAL file (e.g. a config blob) must ERROR, not report 0 entries.*/
|
||||
/* ------------------------------------------------------------------ */
|
||||
static void test_non_wal_file_errors(void)
|
||||
{
|
||||
fresh_dir();
|
||||
char path[512];
|
||||
snprintf(path, sizeof(path), "%s/cluster.bin", g_dir);
|
||||
FILE* fp = fopen(path, "wb");
|
||||
TEST_ASSERT_NOT_NULL(fp);
|
||||
const char junk[] = "not a wal segment — a config blob";
|
||||
fwrite(junk, sizeof(junk), 1, fp);
|
||||
fclose(fp);
|
||||
|
||||
/* Pointing at a non-seg-*.wal file must be a hard error (-EINVAL), not a
|
||||
* silent "0 entries" scan. */
|
||||
waldump_stats_t st;
|
||||
int rc = waldump_path(path, stdout, &st);
|
||||
TEST_ASSERT_LESS_THAN_INT(0, rc);
|
||||
}
|
||||
|
||||
void test_waldump(void)
|
||||
{
|
||||
RUN_TEST(test_contiguous_log_has_no_gaps);
|
||||
RUN_TEST(test_missing_prefix_flagged_as_gap);
|
||||
RUN_TEST(test_duplicate_submit_detection);
|
||||
RUN_TEST(test_interior_gap_flagged);
|
||||
RUN_TEST(test_crc_break_sets_recovered_tip);
|
||||
RUN_TEST(test_missing_path_errors);
|
||||
RUN_TEST(test_non_wal_file_errors);
|
||||
}
|
||||
444
src/controller/testsuite/waldump.c
Normal file
444
src/controller/testsuite/waldump.c
Normal file
|
|
@ -0,0 +1,444 @@
|
|||
#define _GNU_SOURCE
|
||||
#include "waldump.h"
|
||||
|
||||
#include <dirent.h>
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <inttypes.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <zlib.h>
|
||||
|
||||
#include "config.h" /* WAL_HDR_CRC_SIZE, WAL_CRC_NONE, MAX_WAL_PAYLOAD_BYTES */
|
||||
#include "controller.capnp.h" /* Job decode for WAL_JOB_SUBMIT */
|
||||
#include "wal_types.h"
|
||||
|
||||
#define WAL_SEG_FMT "%s/seg-%016" PRIx64 ".wal"
|
||||
#define MAX_SEGS 4096
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Small helpers */
|
||||
/* ------------------------------------------------------------------ */
|
||||
|
||||
static const char* wal_type_name(uint32_t t)
|
||||
{
|
||||
switch (t) {
|
||||
case WAL_JOB_SUBMIT: return "JOB_SUBMIT";
|
||||
case WAL_JOB_START: return "JOB_START";
|
||||
case WAL_JOB_FINISH: return "JOB_FINISH";
|
||||
case WAL_JOB_FAIL: return "JOB_FAIL";
|
||||
case WAL_JOB_STOP: return "JOB_STOP";
|
||||
case WAL_JOB_RESUME: return "JOB_RESUME";
|
||||
case WAL_JOB_UPDATE: return "JOB_UPDATE";
|
||||
case WAL_JOB_RUSAGE: return "JOB_RUSAGE";
|
||||
case WAL_JOB_CKPT: return "JOB_CKPT";
|
||||
case WAL_JOB_PREASON: return "JOB_PREASON";
|
||||
case WAL_QUEUE_CTRL: return "QUEUE_CTRL";
|
||||
case WAL_WORKER_CTRL: return "WORKER_CTRL";
|
||||
case WAL_NOOP: return "NOOP";
|
||||
case WAL_CHECKPOINT: return "CHECKPOINT";
|
||||
default: return "UNKNOWN";
|
||||
}
|
||||
}
|
||||
|
||||
/* Decoded fields of a WAL_JOB_SUBMIT Cap'n Proto payload (debug view). */
|
||||
typedef struct {
|
||||
uint64_t jobid;
|
||||
uint64_t subtime; /* job_submit stamps this with time(NULL) per submit */
|
||||
uint32_t state;
|
||||
} submit_info_t;
|
||||
|
||||
/* Decode a WAL_JOB_SUBMIT payload. Mirrors restore_job() (job.c) but reads a
|
||||
* few fields only, zero-alloc. Returns 0 on success, -1 if undecodable.
|
||||
* subtime is the decisive bug-845 signal: two entries of the SAME job written
|
||||
* in one submit share a subtime (a literal double-append); two separate
|
||||
* submits (e.g. a failover retry) would differ. */
|
||||
static int decode_submit_info(const uint8_t* payload, uint32_t len,
|
||||
submit_info_t* out)
|
||||
{
|
||||
if (!payload || len == 0)
|
||||
return -1;
|
||||
|
||||
struct capn cctx;
|
||||
if (capn_init_mem(&cctx, payload, len, 0) != 0)
|
||||
return -1;
|
||||
|
||||
Job_ptr jp;
|
||||
jp.p = capn_getp(capn_root(&cctx), 0, 1);
|
||||
if (jp.p.type == CAPN_NULL) {
|
||||
capn_free(&cctx);
|
||||
return -1;
|
||||
}
|
||||
|
||||
struct Job cj;
|
||||
memset(&cj, 0, sizeof(cj));
|
||||
read_Job(&cj, jp);
|
||||
out->jobid = cj.jobid;
|
||||
out->subtime = cj.subtime;
|
||||
out->state = cj.state;
|
||||
capn_free(&cctx);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* True if `dir` directly contains at least one seg-*.wal file. */
|
||||
static int dir_has_segments(const char* dir)
|
||||
{
|
||||
DIR* d = opendir(dir);
|
||||
if (!d)
|
||||
return 0;
|
||||
int found = 0;
|
||||
struct dirent* e;
|
||||
while ((e = readdir(d)) != NULL) {
|
||||
size_t n = strlen(e->d_name);
|
||||
if (n > 4 && strncmp(e->d_name, "seg-", 4) == 0 &&
|
||||
strcmp(e->d_name + n - 4, ".wal") == 0) {
|
||||
found = 1;
|
||||
break;
|
||||
}
|
||||
}
|
||||
closedir(d);
|
||||
return found;
|
||||
}
|
||||
|
||||
/* Resolve `path` (seg file / wal dir / node-or-data dir) to the wal dir that
|
||||
* holds the seg-*.wal files. Writes it to `out`. Returns 0, or -errno. When
|
||||
* `path` is itself a single seg file, sets *single_seg to that file path. */
|
||||
static int resolve_wal_dir(const char* path, char* out, size_t outsz,
|
||||
char* single_seg, size_t segsz)
|
||||
{
|
||||
struct stat st;
|
||||
if (stat(path, &st) != 0)
|
||||
return -errno;
|
||||
|
||||
single_seg[0] = '\0';
|
||||
|
||||
if (S_ISREG(st.st_mode)) {
|
||||
/* Only a seg-*.wal file is a WAL segment. Reject anything else (e.g. a
|
||||
* config blob like conf/cluster.bin) instead of parsing it as a WAL and
|
||||
* silently reporting 0 entries. */
|
||||
const char* base = strrchr(path, '/');
|
||||
base = base ? base + 1 : path;
|
||||
size_t bl = strlen(base);
|
||||
if (strncmp(base, "seg-", 4) != 0 || bl < 5 ||
|
||||
strcmp(base + bl - 4, ".wal") != 0)
|
||||
return -EINVAL;
|
||||
snprintf(single_seg, segsz, "%s", path);
|
||||
/* wal dir = dirname(path); only used for display */
|
||||
snprintf(out, outsz, "%s", path);
|
||||
char* slash = strrchr(out, '/');
|
||||
if (slash)
|
||||
*slash = '\0';
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (!S_ISDIR(st.st_mode))
|
||||
return -ENOTDIR;
|
||||
|
||||
/* A directory: try it directly, then data/wal, then wal/. */
|
||||
if (dir_has_segments(path)) {
|
||||
snprintf(out, outsz, "%s", path);
|
||||
return 0;
|
||||
}
|
||||
char cand[4096];
|
||||
snprintf(cand, sizeof(cand), "%s/data/wal", path);
|
||||
if (dir_has_segments(cand)) {
|
||||
snprintf(out, outsz, "%s", cand);
|
||||
return 0;
|
||||
}
|
||||
snprintf(cand, sizeof(cand), "%s/wal", path);
|
||||
if (dir_has_segments(cand)) {
|
||||
snprintf(out, outsz, "%s", cand);
|
||||
return 0;
|
||||
}
|
||||
return -ENOENT; /* no segments anywhere under path */
|
||||
}
|
||||
|
||||
/* Parse the start LSN out of a seg file basename ("seg-<hex16>.wal"). */
|
||||
static int parse_seg_start(const char* name, uint64_t* lsn)
|
||||
{
|
||||
return sscanf(name, "seg-%" SCNx64 ".wal", lsn) == 1 ? 0 : -1;
|
||||
}
|
||||
|
||||
/* Collect + sort segment start LSNs under wal_dir. Returns count, or -errno. */
|
||||
static int collect_segments(const char* wal_dir, uint64_t* seg_lsns, int max)
|
||||
{
|
||||
DIR* d = opendir(wal_dir);
|
||||
if (!d)
|
||||
return -errno;
|
||||
|
||||
int n = 0;
|
||||
struct dirent* e;
|
||||
while ((e = readdir(d)) != NULL && n < max) {
|
||||
uint64_t s;
|
||||
if (parse_seg_start(e->d_name, &s) == 0)
|
||||
seg_lsns[n++] = s;
|
||||
}
|
||||
closedir(d);
|
||||
|
||||
for (int i = 1; i < n; i++) { /* insertion sort — segment count is tiny */
|
||||
uint64_t key = seg_lsns[i];
|
||||
int j = i - 1;
|
||||
while (j >= 0 && seg_lsns[j] > key) {
|
||||
seg_lsns[j + 1] = seg_lsns[j];
|
||||
j--;
|
||||
}
|
||||
seg_lsns[j + 1] = key;
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Per-segment scan: prints entries, validates the CRC chain, detects */
|
||||
/* LSN gaps. Carries the chain CRC and prev LSN across segment calls. */
|
||||
/* ------------------------------------------------------------------ */
|
||||
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)
|
||||
{
|
||||
int fd = open(seg_path, O_RDONLY);
|
||||
if (fd < 0) {
|
||||
fprintf(out, " !! cannot open %s: %s\n", seg_path, strerror(errno));
|
||||
return;
|
||||
}
|
||||
|
||||
uint32_t off = 0;
|
||||
for (;;) {
|
||||
wal_entry_hdr_t hdr;
|
||||
ssize_t hn = pread(fd, &hdr, sizeof(hdr), (off_t)off);
|
||||
if (hn == 0)
|
||||
break; /* clean EOF */
|
||||
if (hn < (ssize_t)sizeof(hdr))
|
||||
break; /* truncated / pre-allocated zeros */
|
||||
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0)
|
||||
break; /* pre-allocated zero region = clean end */
|
||||
|
||||
/* Bound payload before trusting it (corruption guard). */
|
||||
if (hdr.payload_len > MAX_WAL_PAYLOAD_BYTES) {
|
||||
fprintf(out, " !! LSN %" PRIu64 " payload_len %u exceeds max — stop\n",
|
||||
hdr.lsn, hdr.payload_len);
|
||||
st->crc_broken = 1;
|
||||
break;
|
||||
}
|
||||
|
||||
/* CRC chain check (identical rule to persist recovery). */
|
||||
int chain_ok;
|
||||
if (off == 0)
|
||||
chain_ok = (hdr.prev_crc == WAL_CRC_NONE || *last_crc == WAL_CRC_NONE ||
|
||||
hdr.prev_crc == *last_crc);
|
||||
else
|
||||
chain_ok = (hdr.prev_crc == *last_crc);
|
||||
if (!chain_ok) {
|
||||
fprintf(out, " !! LSN %" PRIu64 " prev_crc chain break — stop\n",
|
||||
hdr.lsn);
|
||||
st->crc_broken = 1;
|
||||
break;
|
||||
}
|
||||
|
||||
/* Verify entry CRC over header + payload. */
|
||||
uint8_t* payload = NULL;
|
||||
uint32_t want_crc = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE);
|
||||
if (hdr.payload_len > 0) {
|
||||
payload = malloc(hdr.payload_len);
|
||||
if (!payload)
|
||||
break;
|
||||
ssize_t pn = pread(fd, payload, hdr.payload_len,
|
||||
(off_t)(off + sizeof(hdr)));
|
||||
if (pn != (ssize_t)hdr.payload_len) {
|
||||
free(payload);
|
||||
break; /* short read — clean end */
|
||||
}
|
||||
want_crc = crc32(want_crc, (const Bytef*)payload, hdr.payload_len);
|
||||
}
|
||||
if (hdr.crc != want_crc) {
|
||||
fprintf(out, " !! LSN %" PRIu64 " CRC mismatch — stop\n", hdr.lsn);
|
||||
st->crc_broken = 1;
|
||||
free(payload);
|
||||
break;
|
||||
}
|
||||
|
||||
/* Gap detection against the running LSN sequence. */
|
||||
if (*prev_lsn == 0) {
|
||||
if (hdr.lsn > 1) {
|
||||
fprintf(out, " *** GAP: missing prefix LSN 1..%" PRIu64
|
||||
" (%" PRIu64 " entries absent) ***\n",
|
||||
hdr.lsn - 1, hdr.lsn - 1);
|
||||
st->n_gaps++;
|
||||
}
|
||||
} else if (hdr.lsn > *prev_lsn + 1) {
|
||||
fprintf(out, " *** GAP: missing LSN %" PRIu64 "..%" PRIu64 " ***\n",
|
||||
*prev_lsn + 1, hdr.lsn - 1);
|
||||
st->n_gaps++;
|
||||
}
|
||||
|
||||
/* payload-only CRC — isolates the blob from the header (LSN/term/ts),
|
||||
* so two entries carrying the SAME job blob share a pcrc regardless of
|
||||
* their differing LSNs. Decisive for bug-845: equal pcrc on an adjacent
|
||||
* same-jobid pair ⇒ a literal byte-for-byte double-append. */
|
||||
uint32_t pcrc = (hdr.payload_len > 0)
|
||||
? crc32(0L, (const Bytef*)payload, hdr.payload_len)
|
||||
: 0;
|
||||
|
||||
/* Per-entry line, with jobid/subtime for submits. */
|
||||
char info[96] = "";
|
||||
int is_dup = 0;
|
||||
int dup_identical = 0;
|
||||
if (hdr.type == WAL_JOB_SUBMIT) {
|
||||
submit_info_t si;
|
||||
uint64_t jobid = 0;
|
||||
if (decode_submit_info(payload, hdr.payload_len, &si) == 0) {
|
||||
jobid = si.jobid;
|
||||
snprintf(info, sizeof(info),
|
||||
"jobid=%" PRIu64 " subtime=%" PRIu64 " state=%u",
|
||||
si.jobid, si.subtime, si.state);
|
||||
} else {
|
||||
snprintf(info, sizeof(info), "jobid=?");
|
||||
}
|
||||
st->n_submits++;
|
||||
|
||||
/* bug-845: adjacent duplicate submit — same jobid as the previous
|
||||
* entry at lsn = prev_lsn + 1. Capture identity BEFORE updating the
|
||||
* carried pcrc. */
|
||||
if (jobid != 0 && jobid == *prev_submit_jobid &&
|
||||
hdr.lsn == *prev_lsn + 1) {
|
||||
is_dup = 1;
|
||||
dup_identical = (pcrc == *prev_submit_pcrc);
|
||||
st->n_dup_submits++;
|
||||
if (dup_identical)
|
||||
st->n_dup_identical++;
|
||||
}
|
||||
*prev_submit_jobid = jobid;
|
||||
*prev_submit_pcrc = pcrc;
|
||||
} else {
|
||||
/* a non-submit entry breaks the adjacency for dup detection */
|
||||
*prev_submit_jobid = 0;
|
||||
}
|
||||
|
||||
fprintf(out, " %8" PRIu64 " | %5" PRIu64 " | %-11s | %5u | %08x | %s\n",
|
||||
hdr.lsn, hdr.term, wal_type_name(hdr.type), hdr.payload_len,
|
||||
pcrc, info);
|
||||
if (is_dup)
|
||||
fprintf(out, " ^^ DUP submit: payload %s the prior entry\n",
|
||||
dup_identical ? "BYTE-IDENTICAL to" : "DIFFERS from");
|
||||
|
||||
free(payload);
|
||||
|
||||
/* Advance watermarks. */
|
||||
if (st->first_lsn == 0 || hdr.lsn < st->first_lsn)
|
||||
st->first_lsn = hdr.lsn;
|
||||
if (hdr.lsn > st->last_lsn)
|
||||
st->last_lsn = hdr.lsn;
|
||||
if (hdr.lsn > st->recovered_tip)
|
||||
st->recovered_tip = hdr.lsn;
|
||||
st->n_entries++;
|
||||
*prev_lsn = hdr.lsn;
|
||||
*last_crc = hdr.crc;
|
||||
|
||||
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
|
||||
}
|
||||
|
||||
close(fd);
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ */
|
||||
/* Public entry point */
|
||||
/* ------------------------------------------------------------------ */
|
||||
int waldump_path(const char* path, FILE* out, waldump_stats_t* stats)
|
||||
{
|
||||
if (!path || !out)
|
||||
return -EINVAL;
|
||||
|
||||
char wal_dir[4096];
|
||||
char single_seg[4096];
|
||||
int rc = resolve_wal_dir(path, wal_dir, sizeof(wal_dir),
|
||||
single_seg, sizeof(single_seg));
|
||||
if (rc != 0) {
|
||||
/* Clear, actionable error instead of silently scanning a non-WAL path
|
||||
* (e.g. conf/cluster.bin) and reporting "0 entries". */
|
||||
fprintf(out,
|
||||
"waldump: no WAL segments at '%s' (%s)\n"
|
||||
" point at a node's WAL dir, e.g. "
|
||||
"<deploy>/controller/ctrl-N/data/wal\n"
|
||||
" (or a specific seg-*.wal file inside it)\n",
|
||||
path, strerror(-rc));
|
||||
return rc;
|
||||
}
|
||||
|
||||
waldump_stats_t st;
|
||||
memset(&st, 0, sizeof(st));
|
||||
|
||||
uint64_t seg_lsns[MAX_SEGS];
|
||||
int nseg;
|
||||
/* wider than wal_dir[4096] so the "/seg-<hex16>.wal" suffix never truncates */
|
||||
char seg_path[4096 + 64];
|
||||
|
||||
if (single_seg[0] != '\0') {
|
||||
/* A single seg file: parse its start for ordering display. */
|
||||
const char* base = strrchr(single_seg, '/');
|
||||
base = base ? base + 1 : single_seg;
|
||||
uint64_t start = 0;
|
||||
parse_seg_start(base, &start);
|
||||
seg_lsns[0] = start;
|
||||
nseg = 1;
|
||||
snprintf(seg_path, sizeof(seg_path), "%s", single_seg);
|
||||
|
||||
/* If the dir holds OTHER segments, this single-file dump shows only a
|
||||
* slice of the log. A node that recovered then caught up keeps its
|
||||
* prefix in seg-...001.wal and the suffix in seg-...<current_lsn>.wal,
|
||||
* so a single-file dump looks (wrongly) like missing data. Steer the
|
||||
* user to the whole dir. */
|
||||
uint64_t others[MAX_SEGS];
|
||||
int total = collect_segments(wal_dir, others, MAX_SEGS);
|
||||
if (total > 1)
|
||||
fprintf(out,
|
||||
"NOTE: %s holds %d segments — this dumps only %s.\n"
|
||||
" Run 'waldump %s' (the directory) for the full "
|
||||
"stitched 1..N log.\n",
|
||||
wal_dir, total, base, wal_dir);
|
||||
} else {
|
||||
nseg = collect_segments(wal_dir, seg_lsns, MAX_SEGS);
|
||||
if (nseg < 0)
|
||||
return nseg;
|
||||
if (nseg == 0)
|
||||
return -ENOENT;
|
||||
}
|
||||
|
||||
st.n_segments = (uint32_t)nseg;
|
||||
|
||||
fprintf(out, "== WAL: %s (%d segment%s) ==\n", wal_dir, nseg,
|
||||
nseg == 1 ? "" : "s");
|
||||
fprintf(out, " %8s | %5s | %-11s | %5s | %8s | %s\n",
|
||||
"LSN", "TERM", "TYPE", "PLEN", "PCRC", "INFO");
|
||||
|
||||
uint32_t last_crc = WAL_CRC_NONE;
|
||||
uint64_t prev_lsn = 0;
|
||||
uint64_t prev_submit_jobid = 0;
|
||||
uint32_t prev_submit_pcrc = 0;
|
||||
for (int i = 0; i < nseg; 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);
|
||||
}
|
||||
|
||||
fprintf(out, "-- summary --\n");
|
||||
fprintf(out, " segments: %u entries: %u submits: %u\n",
|
||||
st.n_segments, st.n_entries, st.n_submits);
|
||||
fprintf(out, " lsn range: %" PRIu64 "..%" PRIu64
|
||||
" recovered tip: %" PRIu64 "\n",
|
||||
st.first_lsn, st.last_lsn, st.recovered_tip);
|
||||
fprintf(out, " gaps: %u%s crc chain: %s\n", st.n_gaps,
|
||||
(st.n_gaps && st.first_lsn > 1) ? " (committed prefix missing!)" : "",
|
||||
st.crc_broken ? "BROKEN" : "intact");
|
||||
if (st.n_dup_submits > 0)
|
||||
fprintf(out, " duplicate submits: %u (byte-identical: %u) — bug-845\n",
|
||||
st.n_dup_submits, st.n_dup_identical);
|
||||
|
||||
if (stats)
|
||||
*stats = st;
|
||||
return 0;
|
||||
}
|
||||
42
src/controller/testsuite/waldump.h
Normal file
42
src/controller/testsuite/waldump.h
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
#pragma once
|
||||
/*
|
||||
* waldump.h — read-only WAL segment inspector for failover/durability triage.
|
||||
*
|
||||
* Decodes the on-disk WAL (`wal_entry_hdr_t` + payload, see wal_types.h),
|
||||
* lists every physical entry, decodes JOB_SUBMIT jobids via the Cap'n Proto
|
||||
* Job schema, and — the decisive signal for the committed-job-loss bug —
|
||||
* flags LSN gaps (a missing prefix, or any non-contiguous hole) and reports
|
||||
* the CRC-chain "recovered tip" a real recovery would stop at.
|
||||
*
|
||||
* Strictly read-only: opens segments O_RDONLY, applies no entries, mutates no
|
||||
* persist_t state. Safe to run against a live node's data dir.
|
||||
*/
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
|
||||
/* Aggregate result of scanning one path (a seg file, a wal/ dir, or a node
|
||||
* data dir). All counts are over the physically-present, CRC-valid entries. */
|
||||
typedef struct {
|
||||
uint64_t first_lsn; /* lowest LSN seen (0 = no entries) */
|
||||
uint64_t last_lsn; /* highest LSN seen */
|
||||
uint64_t recovered_tip; /* highest LSN reachable via an intact CRC chain */
|
||||
uint32_t n_entries; /* total entries scanned */
|
||||
uint32_t n_submits; /* WAL_JOB_SUBMIT entries */
|
||||
uint32_t n_gaps; /* LSN discontinuities (missing prefix counts as 1) */
|
||||
uint32_t n_segments; /* segment files scanned */
|
||||
uint32_t n_dup_submits; /* JOB_SUBMITs with same jobid as the immediately-
|
||||
* preceding entry at an adjacent LSN (bug-845) */
|
||||
uint32_t n_dup_identical; /* of n_dup_submits, those whose payload is
|
||||
* byte-identical to the prior entry */
|
||||
int crc_broken; /* nonzero if a CRC/chain break truncated a segment */
|
||||
} waldump_stats_t;
|
||||
|
||||
/* Dump every WAL entry under `path` to `out` (human-readable) and fill
|
||||
* `*stats` when non-NULL. `path` may be:
|
||||
* - a single seg-*.wal file,
|
||||
* - a directory holding seg-*.wal files (a wal/ dir), or
|
||||
* - a node/data dir whose wal/ subdir (data/wal or wal) holds the segments.
|
||||
* Returns 0 on success, or a negative errno on a hard I/O error (e.g. the
|
||||
* 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);
|
||||
Loading…
Add table
Add a link
Reference in a new issue