diff --git a/.kiro/specs/raft-ha-core-safety/tasks.md b/.kiro/specs/raft-ha-core-safety/tasks.md index 2dc12b93..f9830373 100644 --- a/.kiro/specs/raft-ha-core-safety/tasks.md +++ b/.kiro/specs/raft-ha-core-safety/tasks.md @@ -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`. diff --git a/src/controller/persist.c b/src/controller/persist.c index 6e2d3b70..b6071a28 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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; } diff --git a/src/controller/testsuite/test_ha_core_safety.c b/src/controller/testsuite/test_ha_core_safety.c index 7205565d..a31136a9 100644 --- a/src/controller/testsuite/test_ha_core_safety.c +++ b/src/controller/testsuite/test_ha_core_safety.c @@ -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);