fix(raft-ha-core-safety): (term,lsn,payload-crc)-aware persist guard (task 2.2)

Payload-aware repair: identical payload = idempotent no-op; non-empty
differing payload at the TIP = repair (truncate-to-lsn-1 + re-append,
committed_lsn snapshot/restored under lock); empty-over-stored = keep
(never degrade); BELOW-tip conflict = loud non-promotable fence
(recovery_unrecoverable), never truncate the tail or lower committed_lsn.
read_entry_meta_locked computes payload-crc identically to walcompare.
Turns 2.1 case (a) green; new below-tip regression test pins the
committed-boundary safety. Review round 1 caught + fixed a committed-data-
loss bug (tail-truncate on below-tip conflict). 8/8 target suite, sibling
persist/repl/catchup/durability suites green; persist-before-ACK intact.
This commit is contained in:
Rongsong Shen 2026-06-27 10:57:37 +00:00
parent 3b5eda1429
commit 056675c8b3
3 changed files with 335 additions and 14 deletions

View file

@ -46,7 +46,7 @@
- _Requirements: 1.1, 1.2, 2.1, 2.2, 3.1, 3.4_
- _Boundary: Persist Conflict Guard_
- [ ] 2.2 Make the persistence guard `(term, lsn, payload-crc)`-aware
- [x] 2.2 Make the persistence guard `(term, lsn, payload-crc)`-aware
- Extend the LSN-only idempotency check (persist.c:822): identical `(term, payload-crc)` at an LSN is an idempotent no-op; a differing payload-crc (or term) at an existing LSN is surfaced for repair, never a silent `return 0`; never accept a `PLEN=0` write as the durable form of a known-non-empty entry; preserve persist-before-ACK, the term index, and `WAL_LEGACY_TERM`
- Observable completion: the 2.1 cases pass; an empty-over-real write at an existing committed LSN no longer silently survives; identical re-writes stay idempotent
- _Requirements: 1.1, 1.2, 2.1, 2.2, 2.4, 3.1_
@ -139,3 +139,7 @@
- Observable completion: the soak divergence count is zero with closure justified by the live soak, OR a recorded new finding plus a follow-up task; `raft/raft.tla` invariants (Election Safety, Leader Completeness, State Machine Safety, Commit Ordering) reviewed against the final behavior
- _Requirements: 1.1, 1.4, 7.1, 8.1_
- _Depends: 7.2_
## Implementation Notes
- **Task 2.2 — payload repair must be TIP-ONLY (consensus-safety, caught in review round 1).** The `(term,lsn,payload-crc)`-aware guard repairs an empty-over-real entry by truncate-to-`lsn-1`+re-append, but this is safe ONLY when `lsn == current_lsn-1` (the conflict is at the tip with nothing committed above). For a BELOW-tip conflict (`lsn < current_lsn-1` — the actual live LSN-9 case with 10/11/12 above), truncating the tail would drop committed entries and roll `committed_lsn` backward (State Machine Safety violation). Below-tip is handled as a loud non-promotable fence (reuses `recovery_unrecoverable` sticky flag), never mutating any boundary; the real reconvergence comes from the 2.4 streamer fix + re-stream. The tip path snapshots/restores `committed_lsn` under `p->lock` so the boundary is never observably lowered. Any future in-WAL repair must preserve: never drop a committed entry above the conflict, never lower `committed_lsn`.

View file

@ -799,6 +799,113 @@ int persist_wal_async(persist_t* p, uint32_t type,
return 0;
}
/* Lock-held core of persist_truncate_wal_above (defined below): assumes the
* caller already holds p->lock and never touches the lock itself. The
* replicated-write conflict guard reuses it to repair a divergent entry while
* still under p->lock (the follower-AE-conflict truncate contract). */
static int truncate_wal_above_locked(persist_t* p, uint64_t cutoff_lsn);
/* read_entry_meta_locked: scan the on-disk WAL for the entry at `lsn` and
* return its (payload_len, payload-only CRC, raw stored term). The
* replicated-write conflict guard uses this to compare a re-streamed entry
* against the copy already on disk WITHOUT trusting the LSN alone. pcrc is
* computed as crc32(0, payload, len) — the SAME payload-only fingerprint
* walcompare/waldump compare across members (NOT the chained header CRC); an
* empty (PLEN=0) entry yields len=0, pcrc=0. Reads through the page cache so
* a written-but-not-yet-fdatasync'd entry is still visible. Caller MUST hold
* p->lock. Returns 0 if the entry is found, -1 if absent or unreadable. */
static int read_entry_meta_locked(persist_t* p, uint64_t lsn,
uint32_t* out_len, uint32_t* out_pcrc,
uint64_t* out_term)
{
if (!p || lsn == 0)
return -1;
DIR* d = opendir(p->wal_path);
if (!d)
return -1;
uint64_t seg_lsns[4096];
int nseg = 0;
struct dirent* e;
while ((e = readdir(d)) != NULL &&
nseg < (int)(sizeof(seg_lsns) / sizeof(seg_lsns[0]))) {
if (strncmp(e->d_name, "seg-", 4) != 0)
continue;
uint64_t sl = 0;
if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &sl) != 1)
continue;
seg_lsns[nseg++] = sl;
}
closedir(d);
if (nseg == 0)
return -1;
/* Ascending sort by start LSN (segment count is tiny). */
for (int i = 1; i < nseg; i++) {
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;
}
int rc = -1;
for (int si = 0; si < nseg && rc != 0; si++) {
/* Segments are LSN-ordered; one whose start LSN already exceeds the
* target cannot contain it (nor can any later one). */
if (seg_lsns[si] > lsn)
break;
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path),
WAL_SEG_FMT, p->wal_path, seg_lsns[si]);
int fd = open(seg_path, O_RDONLY);
if (fd < 0)
continue;
uint32_t off = 0;
for (;;) {
wal_entry_hdr_t hdr;
ssize_t hn = pread(fd, &hdr, sizeof(hdr), (off_t)off);
if (hn < (ssize_t)sizeof(hdr))
break; /* short read / EOF / pre-allocated zeros */
if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0)
break;
if (hdr.lsn == lsn) {
uint32_t pc = 0;
if (hdr.payload_len > 0) {
uint8_t* pl = malloc(hdr.payload_len);
if (!pl)
break; /* leave rc == -1: unreadable → conservative */
ssize_t pn = pread(fd, pl, hdr.payload_len,
(off_t)(off + sizeof(hdr)));
if (pn == (ssize_t)hdr.payload_len)
pc = crc32(0L, (const Bytef*)pl, hdr.payload_len);
free(pl);
if (pn != (ssize_t)hdr.payload_len)
break; /* unreadable body → conservative */
}
if (out_len)
*out_len = hdr.payload_len;
if (out_pcrc)
*out_pcrc = pc;
if (out_term)
*out_term = hdr.term;
rc = 0;
break;
}
off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}
close(fd);
}
return rc;
}
int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
const void* payload, uint32_t len)
{
@ -815,13 +922,117 @@ int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
pthread_mutex_lock(&p->lock);
/* Idempotency / dedup guard: if this lsn is already persisted (current_lsn
* already advanced past it) treat the re-stream as a no-op. This keeps the
* write idempotent and avoids a duplicate entry / CRC-chain corruption when
* the primary re-sends an lsn the standby already wrote. */
/* Tip-repair bookkeeping (set only on the TIP-conflict repair branch below):
* the truncate-and-reappend repair transiently lowers committed_lsn to
* lsn-1, so we snapshot the true committed boundary here and restore it
* EXACTLY after the re-append re-establishes the entry at `lsn` (never let a
* same-(term,lsn) payload repair regress the committed boundary — commit
* monotonicity / State Machine Safety). */
int repaired_tip = 0;
uint64_t repaired_committed_save = 0;
/* Payload-aware conflict guard (raft-ha-core-safety task 2.2): this lsn is
* already persisted (current_lsn advanced past it). The legacy guard was
* (lsn)-only — it no-op'd ANY re-stream, including a corrective full-payload
* write over a previously-stored empty (PLEN=0) body, which is exactly the
* cross-term catch-up payload-loss signature (a committed LSN that carries
* the right (term,lsn) but an empty payload). Compare (payload-crc, len)
* against the copy already on disk and act on the difference:
* (c) identical payload → idempotent no-op (keep).
* (b) incoming EMPTY over a stored entry → never degrade (keep stored);
* an empty body is never the durable form of a known entry.
* (a) incoming NON-EMPTY whose payload differs from the stored copy at an
* existing (term,lsn) → a genuine payload conflict whose handling
* depends on WHERE the conflict sits relative to the stored tip:
* · AT THE TIP (lsn == current_lsn - 1, nothing stored above) →
* REPAIR: truncate just this one entry and re-append the
* authoritative payload at the same (term,lsn). Because nothing
* is stored above it, no distinct committed entry is destroyed,
* and the committed boundary is restored verbatim afterwards.
* · BELOW THE TIP (lsn < current_lsn - 1, committed entries stored
* ABOVE the conflict) → MUST NOT truncate: dropping [lsn..tip]
* would delete committed entries above the conflict AND roll
* committed_lsn/durable_lsn backward — a State Machine Safety /
* commit-monotonicity violation. Instead keep the stored entry
* untouched, raise a loud non-promotable fence, and let catch-up
* reconverge. The real source fix (the streamer must stop
* shipping an empty body) is task 2.4; the guard's only job here
* is to NOT make things worse (never drop a committed entry,
* never regress a boundary) and to surface the corruption loudly.
* An unreadable stored entry stays a conservative no-op — never truncate a
* committed LSN we cannot read back. */
if (lsn + 1 <= p->current_lsn) {
pthread_mutex_unlock(&p->lock);
return 0;
uint32_t inc_pcrc =
(payload && len > 0) ? crc32(0L, (const Bytef*)payload, len) : 0;
uint32_t st_len = 0;
uint32_t st_pcrc = 0;
uint64_t st_term = 0;
int have = read_entry_meta_locked(p, lsn, &st_len, &st_pcrc,
&st_term);
int payload_matches =
(have == 0 && st_len == len && st_pcrc == inc_pcrc);
/* (c) identical payload, (b) empty incoming, or an unreadable stored
* entry → leave the stored entry untouched. */
if (payload_matches || len == 0 || have != 0) {
pthread_mutex_unlock(&p->lock);
return 0;
}
/* (a) non-empty incoming whose payload differs at an existing (term,lsn)
* — a genuine payload conflict. Branch on tip vs below-tip. */
uint64_t tip_lsn = (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
if (lsn != tip_lsn) {
/* BELOW-TIP conflict: committed entries are stored ABOVE `lsn`.
* Truncating here would delete them and roll committed_lsn/
* durable_lsn backward (State Machine Safety / commit-monotonicity
* violation — bug-class the round-1 review rejected). Do NOT
* truncate, do NOT lower any boundary, do NOT degrade the stored
* entry. Surface loudly and raise the sticky non-promotable fence
* (reuse the recovery_unrecoverable mechanism: loud, stays ALIVE,
* non-promotable, self-heals via persist_clear_recovery_unrecoverable
* once catch-up reaches a committed-consistent state — never
* self-exits, per the controller invariant). The source fix lives
* in task 2.4 (the streamer must stop shipping an empty payload). */
ERROR("persist: persist_wal_write_replicated payload CONFLICT BELOW "
"TIP at lsn=%" PRIu64 " (stored_term=%" PRIu64
" stored_len=%u stored_pcrc=%08x vs incoming_len=%u "
"incoming_pcrc=%08x); current_lsn=%" PRIu64 " committed_lsn=%"
PRIu64 " — refusing to truncate committed entries above the "
"conflict (would regress the committed boundary). Keeping the "
"stored entry, raising the non-promotable fence and staying "
"ALIVE; awaiting a correct re-stream (source fix: task 2.4)",
lsn, st_term, st_len, st_pcrc, len, inc_pcrc, p->current_lsn,
atomic_load_explicit(&p->committed_lsn, memory_order_acquire));
atomic_store_explicit(&p->recovery_unrecoverable, true,
memory_order_release);
pthread_mutex_unlock(&p->lock);
return 0; /* no-op the write: keep stored, never degrade/regress */
}
/* TIP conflict: `lsn` is the highest stored entry, so nothing is stored
* above it — truncate-and-reappend replaces EXACTLY this one divergent
* entry. truncate_wal_above_locked(lsn-1) drops it and sets
* current_lsn = lsn, so control falls through to the append below and
* re-persists the authoritative entry at this exact LSN. Snapshot the
* committed boundary first: the truncate transiently rewrites
* committed_lsn to lsn-1, and we restore it verbatim after the
* re-append so a same-(term,lsn) payload repair never regresses (nor
* spuriously advances) the committed boundary. */
repaired_committed_save =
atomic_load_explicit(&p->committed_lsn, memory_order_acquire);
int trc = truncate_wal_above_locked(p, lsn - 1);
if (trc != 0) {
ERROR("persist: persist_wal_write_replicated repair-truncate FAILED "
"for lsn=%" PRIu64 " (rc=%d) — divergent entry NOT repaired",
lsn, trc);
pthread_mutex_unlock(&p->lock);
return trc;
}
repaired_tip = 1;
/* fall through to the contiguity guard + append (lsn == current_lsn). */
}
/* Contiguity guard (bug-850): REFUSE a forward GAP. current_lsn is the
@ -952,6 +1163,23 @@ int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
atomic_store_explicit(&p->durable_lsn, tip, memory_order_release);
}
/* Tip-repair committed-boundary restore (raft-ha-core-safety task 2.2): a
* TIP-conflict repair above truncated [lsn..tip] (rewriting committed_lsn to
* lsn-1) and has now re-appended + fdatasync'd the authoritative entry at the
* SAME (term,lsn). Restore the committed boundary VERBATIM so the repair is
* commit-boundary-neutral: a same-(term,lsn) payload correction must never
* regress committed_lsn (State Machine Safety / commit monotonicity) nor
* spuriously advance it (truncate would otherwise have raised it to lsn-1).
* Reached only past the fail-loud fdatasync gate, with current_lsn re-advanced
* to lsn+1 and durable_lsn at the synced tip, so the invariant
* committed_lsn <= min(current_lsn - 1, durable_lsn) holds (saved value was
* itself <= the pre-repair tip == lsn). The whole repair is serialized under
* p->lock, so no external observer ever sees the transient lowered value. */
if (repaired_tip) {
atomic_store_explicit(&p->committed_lsn, repaired_committed_save,
memory_order_release);
}
/* Rotate segment when full (same threshold as the primary append). */
if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) {
if (rotate_segment(p) != 0) {
@ -2824,10 +3052,19 @@ static void clamp_durable_lsn_locked(persist_t* p, uint64_t tip)
int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
{
pthread_mutex_lock(&p->lock);
int rc = truncate_wal_above_locked(p, cutoff_lsn);
pthread_mutex_unlock(&p->lock);
return rc;
}
/* Lock-held core of persist_truncate_wal_above (see the forward declaration and
* the conflict-guard reuse near persist_wal_write_replicated). Caller MUST hold
* p->lock; this function never touches the lock so it is reusable from another
* already-locked path. */
static int truncate_wal_above_locked(persist_t* p, uint64_t cutoff_lsn)
{
/* No-op: nothing written yet, or cutoff covers all written entries */
if (p->current_lsn == 0 || cutoff_lsn + 1 >= p->current_lsn) {
pthread_mutex_unlock(&p->lock);
return 0;
}
@ -2851,7 +3088,6 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
}
if (nseg == 0) {
pthread_mutex_unlock(&p->lock);
return 0;
}
@ -2903,7 +3139,6 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
clamp_durable_lsn_locked(p, cutoff_lsn);
/* Keep the per-LSN term index consistent: drop terms for removed LSNs. */
truncate_term_index_locked(p, cutoff_lsn);
pthread_mutex_unlock(&p->lock);
return 0;
}
@ -2914,7 +3149,6 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
int scan_fd = open(cutoff_seg_path, O_RDONLY);
if (scan_fd < 0) {
pthread_mutex_unlock(&p->lock);
return -errno;
}
@ -2954,13 +3188,11 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
/* --- Truncate the cutoff segment at the computed offset --- */
int new_fd = open(cutoff_seg_path, O_RDWR);
if (new_fd < 0) {
pthread_mutex_unlock(&p->lock);
return -errno;
}
if (ftruncate(new_fd, (off_t)trunc_off) != 0) {
int saved = errno;
close(new_fd);
pthread_mutex_unlock(&p->lock);
return -saved;
}
fdatasync(new_fd);
@ -2985,6 +3217,5 @@ int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn)
/* Keep the per-LSN term index consistent: drop terms for removed LSNs. */
truncate_term_index_locked(p, cutoff_lsn);
pthread_mutex_unlock(&p->lock);
return 0;
}

View file

@ -388,6 +388,89 @@ static void test_2_1_identical_rewrite_is_idempotent_noop(void)
printf("test_2_1_identical_rewrite_is_idempotent_noop: PASS (pin)\n");
}
/* ═══════════════════════════════════════════════════════════════════════════
* TASK 2.2 REMEDIATION (round 1) — below-tip payload conflict must NOT drop
* committed entries above the conflict, and must NOT regress committed_lsn.
*
* The round-1 repair fired truncate-and-reappend for ANY lsn < current_lsn,
* which (when the conflict sits BELOW the stored tip — an empty body at a
* committed LSN with committed entries ABOVE it) deleted those committed
* entries and rolled committed_lsn backward — a State Machine Safety /
* commit-monotonicity violation on the still-live term-blind apply route.
*
* Construct exactly that shape: entries [1..N] (N=4) with an empty (PLEN=0)
* body at k=2 (k < N, so entries 3,4 sit ABOVE the conflict), committed_lsn=N.
* Stream the CORRECT full body at lsn=k (the corrective replicated write).
* The fix must NOT truncate the tail here: assert (a) entries k+1..N survive,
* and (b) committed_lsn is unchanged. This FAILs against the round-1 version
* (entries 3,4 dropped, committed_lsn rolled to k-1) and PASSes after the fix
* (below-tip conflict → loud non-promotable fence, stored entry kept).
* ═══════════════════════════════════════════════════════════════════════════ */
static void test_2_2_below_tip_conflict_keeps_committed_entries(void)
{
hcs_member_t m;
TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&m, "btip"));
enum { N = 4, K = 2 };
uint8_t full[HCS_FULL_LEN];
hcs_fill_body(full);
/* Build entries [1..N] contiguously through the real replicated guard, with
* the entry at K written EMPTY (PLEN=0) — the live cross-term catch-up
* payload-loss signature, with committed entries K+1..N stored ABOVE it. */
for (uint64_t lsn = 1; lsn <= N; lsn++) {
const void* body = (lsn == K) ? NULL : (const void*)full;
uint32_t blen = (lsn == K) ? 0u : HCS_FULL_LEN;
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
hcs_wal_put(&m, lsn, /*term=*/3, WAL_JOB_SUBMIT, body, blen),
"each contiguous build write must persist (real fs required — bug-924)");
}
/* The whole prefix [1..N] is committed. */
atomic_store_explicit(&m.p->committed_lsn, (uint64_t)N, memory_order_release);
/* Sanity: entries K+1..N are present with the full payload before the repair. */
for (uint64_t lsn = K + 1; lsn <= N; lsn++) {
hcs_probe_t pr;
TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&m, lsn, &pr));
TEST_ASSERT_TRUE_MESSAGE(pr.present, "above-conflict entry must exist pre-repair");
TEST_ASSERT_EQUAL_UINT32(HCS_FULL_LEN, pr.len);
}
/* The corrective FULL body is streamed for the SAME (term, lsn=K) — a
* non-empty payload that differs from the stored empty body, BELOW the tip. */
int rc = hcs_wal_put(&m, /*lsn=*/K, /*term=*/3, WAL_JOB_SUBMIT, full, HCS_FULL_LEN);
TEST_ASSERT_TRUE_MESSAGE(rc == 0,
"the below-tip conflict write must be a loud no-op (rc==0), never an error "
"that re-streams destructively");
/* (a) Entries K+1..N must STILL be present — the round-1 truncate dropped
* them; the fix must keep every committed entry above the conflict. */
for (uint64_t lsn = K + 1; lsn <= N; lsn++) {
hcs_probe_t pr;
TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&m, lsn, &pr));
TEST_ASSERT_TRUE_MESSAGE(pr.present,
"REGRESSION GUARD: a below-tip payload repair must NOT drop committed "
"entries stored ABOVE the conflict LSN (round-1 truncate deleted them)");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(HCS_FULL_LEN, pr.len,
"above-conflict committed entry must keep its full payload");
}
/* (b) committed_lsn must NOT be lowered — round-1 rolled it back to K-1. */
TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)N,
atomic_load_explicit(&m.p->committed_lsn, memory_order_acquire),
"REGRESSION GUARD: a below-tip payload repair must NOT regress committed_lsn "
"(commit monotonicity / State Machine Safety)");
/* The below-tip conflict raises the loud non-promotable fence (stays ALIVE,
* self-heals via catch-up) rather than corrupting committed state. */
TEST_ASSERT_TRUE_MESSAGE(persist_recovery_unrecoverable(m.p),
"a below-tip payload conflict must raise the non-promotable fence");
hcs_member_down(&m);
printf("test_2_2_below_tip_conflict_keeps_committed_entries: PASS\n");
}
/* ═══════════════════════════════════════════════════════════════════════════
* SCAFFOLDING (PASS) — prove each shared seam is usable by later phases.
* ═══════════════════════════════════════════════════════════════════════════ */
@ -519,6 +602,9 @@ void test_ha_core_safety(void)
RUN_TEST(test_smoke_red_payload_blind_guard_drops_corrective_write);
RUN_TEST(test_2_1_full_then_empty_must_not_degrade);
RUN_TEST(test_2_1_identical_rewrite_is_idempotent_noop);
/* Task 2.2 remediation: a below-tip payload conflict must not drop
* committed entries above it nor regress committed_lsn. */
RUN_TEST(test_2_2_below_tip_conflict_keeps_committed_entries);
/* Scaffolding: each shared seam is usable by 2.3 / 4.1 / 5.1. */
RUN_TEST(test_seam_wal_construct_and_probe);
RUN_TEST(test_seam_divergence_is_observable);