From 63b9c5a81ac2819060bba6e89274fc283a49d968 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 27 Jun 2026 15:42:37 +0000 Subject: [PATCH] fix(raft-ha-core-safety): streamer fail-loud bounded-retry, never ship empty (task 2.4) Root-cause source fix: ha_stream_wal_to_standby reads the entry body via stream_read_entry_body (bounded retry max 4, no sleep); on persistent short pread/malloc-fail it fails loud + ABORTS the stream (flush valid prefix, drop unreadable entry, -EIO) instead of shipping PLEN=0. Incomplete stream re-driven by ACK-driven catch-up. With the 2.2 guard this closes the live cross-term payload-loss divergence. 2.3 CORE RED -> GREEN (fail-loud: no empty persisted; + retry-recover test ships full 64). Term-blind structural retirement deferred to 6.2 (empty-over-full safety enforced by 2.2 guard + pinned). 12/12 core suite; siblings green; no regression introduced. Phase 2 complete. --- .kiro/specs/raft-ha-core-safety/tasks.md | 8 +- src/controller/config_repl.c | 129 +++++++++++++++--- .../testsuite/test_ha_core_safety.c | 99 +++++++++++++- 3 files changed, 210 insertions(+), 26 deletions(-) diff --git a/.kiro/specs/raft-ha-core-safety/tasks.md b/.kiro/specs/raft-ha-core-safety/tasks.md index 7b59a2bb..7dd953e9 100644 --- a/.kiro/specs/raft-ha-core-safety/tasks.md +++ b/.kiro/specs/raft-ha-core-safety/tasks.md @@ -59,7 +59,7 @@ - _Boundary: Catch-up Payload Integrity, Unified Entry Ingest_ - _Depends: 2.2_ -- [ ] 2.4 Fix the streamer payload integrity + defensive seam routing + fence +- [x] 2.4 Fix the streamer payload integrity + defensive seam routing + fence - Core: in `ha_stream_wal_to_standby` (config_repl.c ~2154-2163), fail loud on a short `pread`/`malloc`-fail with a bounded retry instead of freeing the buffer and shipping `PLEN=0`; a non-empty entry is never transmitted empty. Disambiguate and handle send-side-empty vs transitive-empty propagation per the 2.3 instrumentation. Defensive: retire the term-blind `ha_apply_single_entry` direct-persist branch so all ingest routes through the seam; surface genuinely-unrecoverable divergence as a loud non-promotable fence (never self-terminate); defer file I/O to delaymgr (never the callback/iomgr thread) - Observable completion: 2.3 passes (streamer never ships empty; seam-routing and fence cases green); on the 1.3 reproduction path (real-fs release chaotic soak) `walcompare` reports zero committed divergence at the previously-divergent LSNs - _Requirements: 1.1, 1.2, 1.3, 2.1, 2.3, 3.1, 3.2, 3.3, 3.4_ @@ -116,8 +116,8 @@ - _Depends: 2.4_ - [ ] 6.2 Verify the single-seam invariant and restart identity - - Assert no ingest path (live replication, catch-up batch, snapshot post-install, rejoin) writes the WAL outside the reconcile seam, and that each entry's `(term, lsn)` identity survives a restart/recovery - - Observable completion: an integration test confirms every ingest path reconciles through the seam and that entry terms are preserved across restart; no direct-persist bypass remains + - Assert no ingest path (live replication, catch-up batch, snapshot post-install, rejoin) writes the WAL outside the reconcile seam, and that each entry's `(term, lsn)` identity survives a restart/recovery. **Carry-forward from 2.4:** the term-blind `ha_apply_single_entry` direct-persist branch was NOT structurally retired in 2.4 (its empty-over-full safety is enforced by the 2.2 payload guard and pinned by `test_2_3_termblind_route_payload_guarded`/`test_2_3_unrepairable_catchup_raises_nonpromotable_fence`). This task must either structurally retire that bypass (route term-0 catch-up through the seam, with its own RED first) OR formally document it as safe-by-persist-guard and adjust the invariant wording accordingly + - Observable completion: an integration test confirms every ingest path reconciles through the seam and that entry terms are preserved across restart; the term-blind bypass is either removed (no direct-persist path remains) or documented safe-by-guard with the pin tests cited - _Requirements: 1.1, 1.2, 2.4, 3.1_ - _Depends: 2.4, 3.2, 4.2_ @@ -143,3 +143,5 @@ ## 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`. + +- **Task 2.4 — root-cause source fix + a deferral.** `ha_stream_wal_to_standby` now reads the entry body via `stream_read_entry_body` (bounded retry, max 4 no-progress attempts, partial-read-accumulating, no sleep); on persistent short `pread`/`malloc`-fail it emits a loud ERROR and ABORTS the stream (flushes the valid contiguous prefix, drops the unreadable entry, returns `-EIO`) instead of shipping `PLEN=0`. The incomplete stream is re-driven by the ACK-driven catch-up continuation. Combined with the 2.2 self-healing guard, this closes the live cross-term payload-loss divergence. **Deferred:** the structural retirement of the term-blind `ha_apply_single_entry` bypass — its empty-over-full safety is enforced by the 2.2 guard + pinned by two passing tests, so it cannot corrupt committed state; the structural removal is tracked to task 6.2. Live ≥6-round soak proof is task 7.2. diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index e46b89d1..c64532c6 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -2043,6 +2043,73 @@ static int send_wal_batch(endpoint_t *ep, return 0; } +/* Maximum bounded re-attempts when an entry body read makes no forward progress + * (a zero/EOF/EINTR short read or an allocation failure). Small and finite so the + * loop can never busy-spin; a regular-file body completes in one or two reads, so + * this only ever spends attempts on a genuinely failing read. No sleep/backoff — + * a transient cause (EINTR, partial read) recovers on the next immediate attempt; + * a persistent cause exhausts the budget and is surfaced loud (never shipped + * empty). raft-ha-core-safety task 2.4. */ +#define WAL_STREAM_BODY_READ_MAX_ATTEMPTS 4 + +/* stream_read_entry_body: read a WAL entry body of `payload_len` bytes at + * `body_off` into a freshly malloc'd buffer, accumulating partial reads and + * bounded-retrying a no-progress read. On success returns 0 with *out_buf set to + * a heap buffer the caller owns (free()); on persistent failure returns negative + * (-ENOMEM / -EIO) with *out_buf == NULL. + * + * raft-ha-core-safety task 2.4 — the SEND-side payload-integrity fix. The + * baseline freed the buffer and shipped the entry with PLEN=0 on a short + * pread/malloc-fail, silently transmitting a non-empty COMMITTED entry as empty + * (the live LSN-9 divergence). This helper instead fails LOUD so the caller can + * abort the stream rather than degrade the entry to empty. A short read is not + * inherently fatal: pread may legitimately return fewer bytes, so we re-read from + * the advanced offset (forward progress never spends an attempt); only a read + * that returns <= 0 consumes the bounded retry budget. */ +static int stream_read_entry_body(int fd, uint32_t body_off, + uint32_t payload_len, uint64_t lsn, + uint64_t stored_term, uint64_t leader_term, + uint8_t **out_buf) +{ + *out_buf = NULL; + + uint8_t *pbuf = NULL; + for (int a = 0; a < WAL_STREAM_BODY_READ_MAX_ATTEMPTS && !pbuf; a++) + pbuf = malloc(payload_len); + if (!pbuf) + return -ENOMEM; + + uint32_t got = 0; + int attempts = 0; + while (got < payload_len) { + ssize_t pr = pread(fd, pbuf + got, payload_len - got, + (off_t)(body_off + got)); +#ifdef XCE_TESTING + /* raft-ha-core-safety task 2.3/2.4 fault-injection seam: a test may force + * this read to look SHORT (pr=0) to exercise the fail-loud + bounded-retry + * path. NULL in production — no behaviour change. The hook fires on every + * attempt, so a permanent fault exhausts the budget (fail loud) while a + * transient one (hook returns 0 after N) recovers. */ + if (ha_stream_body_read_fault_hook != NULL && + ha_stream_body_read_fault_hook(lsn, stored_term, leader_term, + payload_len)) + pr = 0; +#endif + if (pr > 0) { + got += (uint32_t)pr; /* forward progress — does not spend an attempt */ + continue; + } + /* pr <= 0: EOF/zero/EINTR/error — bounded retry, no busy-spin. */ + if (++attempts >= WAL_STREAM_BODY_READ_MAX_ATTEMPTS) { + free(pbuf); + return -EIO; + } + } + + *out_buf = pbuf; + return 0; +} + int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, const persist_t *p, uint64_t leader_term, uint64_t standby_lsn, uint64_t committed_lsn) @@ -2166,30 +2233,48 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, entry->leader_current_term = leader_term; if (hdr.payload_len > 0) { - uint8_t *pbuf = malloc(hdr.payload_len); - if (pbuf) { - ssize_t pr = pread(fd, pbuf, hdr.payload_len, - (off_t)(off + sizeof(hdr))); -#ifdef XCE_TESTING - /* raft-ha-core-safety task 2.3 fault-injection seam: a - * test may force this body read to look SHORT (the same - * outcome as the malloc-fail tail above) to prove the - * baseline ships a non-empty entry as PLEN=0. NULL in - * production — no behaviour change. Task 2.4 owns the - * fix (fail loud + bounded retry); this seam ships no fix. */ - if (ha_stream_body_read_fault_hook != NULL && - ha_stream_body_read_fault_hook(hdr.lsn, entry->term, - leader_term, - hdr.payload_len)) - pr = 0; /* simulate a short read */ -#endif - if (pr == (ssize_t)hdr.payload_len) { - entry->raw_payload = pbuf; - entry->raw_payload_len = hdr.payload_len; - } else { - free(pbuf); + /* raft-ha-core-safety task 2.4: read the body with bounded + * retry; a non-empty COMMITTED entry must NEVER be shipped + * empty (the live LSN-9 send-side payload loss). */ + uint8_t *pbuf = NULL; + int rrc = stream_read_entry_body( + fd, off + (uint32_t)sizeof(hdr), hdr.payload_len, + hdr.lsn, entry->term, leader_term, &pbuf); + if (rrc != 0) { + /* FAIL LOUD: the body could not be read in full after the + * bounded retry. Do NOT degrade the entry to PLEN=0 — + * abort the stream here. Flush the valid prefix already + * batched (entries strictly below this LSN remain correct + * and contiguous), drop THIS entry, and return an error so + * the ACK-driven catch-up continuation re-streams from this + * LSN on the next cycle. The standby simply does not + * advance past the gap — it never receives a committed + * entry as empty. The controller stays alive and + * self-heals (CLAUDE.md process-lifecycle invariant: + * fail-loud = refuse to ship divergent state, not exit). */ + ERROR("ha_stream_wal_to_standby: entry body read FAILED " + "for lsn=%" PRIu64 " payload_len=%u stored_term=%" + PRIu64 " leader_term=%" PRIu64 " rc=%d after %d " + "attempts — ABORTING stream, NOT shipping empty (a " + "non-empty committed entry must never be transmitted " + "as PLEN=0; catch-up will re-stream from this LSN)", + hdr.lsn, hdr.payload_len, entry->term, leader_term, + rrc, WAL_STREAM_BODY_READ_MAX_ATTEMPTS); + free(entry); + if (batch_n > 0) { + send_wal_batch(ep, batch_entries, batch_n); + for (int bi = 0; bi < batch_n; bi++) { + free(batch_entries[bi]->raw_payload); + free(batch_entries[bi]); + } + batch_n = 0; + batch_bytes = 0; } + close(fd); + return -EIO; } + entry->raw_payload = pbuf; + entry->raw_payload_len = hdr.payload_len; } batch_entries[batch_n] = entry; diff --git a/src/controller/testsuite/test_ha_core_safety.c b/src/controller/testsuite/test_ha_core_safety.c index 50ce7ecb..314ecea7 100644 --- a/src/controller/testsuite/test_ha_core_safety.c +++ b/src/controller/testsuite/test_ha_core_safety.c @@ -41,6 +41,7 @@ * Requirements: 7.1 (force divergence RED on baseline), 7.4 (unit harness). */ #define _GNU_SOURCE +#include #include #include #include @@ -614,6 +615,90 @@ static int hcs_2_3_target_shipped_empty(void) return 0; } +/* The send_len captured for the target LSN (0 if it was never shipped). */ +static uint32_t hcs_2_3_target_send_len(void) +{ + for (int i = 0; i < g_sends_n; i++) + if (g_sends[i].lsn == g_fault_lsn) + return g_sends[i].send_len; + return 0; +} + +/* task 2.4: a TRANSIENT body-read fault — fails the first g_fault_transient_left + * attempts, then lets the read succeed. Proves the bounded retry RECOVERS the + * full payload (the streamer ships SEND_len==source_len), the counterpart to the + * permanent-fault fail-loud path above. */ +static int g_fault_transient_left; +static int hcs_2_4_transient_fault_hook(uint64_t lsn, uint64_t stored_term, + uint64_t leader_term, uint32_t payload_len) +{ + (void)payload_len; + if (lsn != g_fault_lsn) + return 0; + if (stored_term < leader_term) + g_fault_saw_crossterm = 1; + if (g_fault_transient_left > 0) { + g_fault_transient_left--; + return 1; /* force a short read on this attempt */ + } + return 0; /* recovered — let the real read complete */ +} + +/* GREEN (task 2.4) — a transient short read is recovered by the bounded retry: + * the entry is shipped FULL (SEND_len == source_len), never empty. */ +static void test_2_4_streamer_short_read_retry_recovers(void) +{ + hcs_member_t src, rcv; + TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&src, "s24src")); + TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&rcv, "s24rcv")); + + enum { TGT = 1, STORED_TERM = 3, LEADER_TERM = 4 }; + uint8_t full[HCS_FULL_LEN]; + hcs_fill_body(full); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, + hcs_wal_put(&src, TGT, STORED_TERM, WAL_JOB_SUBMIT, full, HCS_FULL_LEN), + "source full write must persist (real fs required — bug-924)"); + atomic_store_explicit(&src.p->committed_lsn, (uint64_t)TGT, + memory_order_release); + + /* Fail the first 2 read attempts (< the bounded budget) then recover. */ + hcs_2_3_reset(&rcv, /*fault_lsn=*/(uint64_t)TGT); + g_fault_transient_left = 2; + ha_stream_body_read_fault_hook = hcs_2_4_transient_fault_hook; + xce_wal_stream_request_hook = hcs_2_3_capture_hook; + int rc = ha_stream_wal_to_standby((endpoint_t*)NULL, src.wal_dir, src.p, + /*leader_term=*/(uint64_t)LEADER_TERM, + /*standby_lsn=*/0, + /*committed_lsn=*/(uint64_t)TGT); + ha_stream_body_read_fault_hook = NULL; + xce_wal_stream_request_hook = NULL; + + uint32_t send_len = hcs_2_3_target_send_len(); + hcs_probe_t rcv_pr; + int rcv_rc = hcs_wal_probe(&rcv, (uint64_t)TGT, &rcv_pr); + printf("test_2_4 retry-recover LSN %d: source_len=%u SEND_len=%u RECV_len=%u " + "(rc=%d, crossterm_seen=%d)\n", (int)TGT, HCS_FULL_LEN, send_len, + rcv_pr.present ? rcv_pr.len : 0u, rc, g_fault_saw_crossterm); + + TEST_ASSERT_TRUE_MESSAGE(g_fault_saw_crossterm, + "the transient fault must fire on the cross-term entry"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "a transient short read recovered within the retry budget — the stream " + "completes cleanly (rc==0), not aborted"); + /* The recovered entry is shipped FULL — SEND_len equals the source length. */ + TEST_ASSERT_EQUAL_UINT32_MESSAGE(HCS_FULL_LEN, send_len, + "GREEN: the bounded retry recovered the body; the entry is shipped with " + "its FULL payload (SEND_len == source_len), never empty"); + TEST_ASSERT_EQUAL_INT(0, rcv_rc); + TEST_ASSERT_TRUE_MESSAGE(rcv_pr.present && rcv_pr.len == HCS_FULL_LEN, + "GREEN: the receiver persisted the FULL recovered body at the target LSN"); + + hcs_member_down(&src); + hcs_member_down(&rcv); + printf("test_2_4_streamer_short_read_retry_recovers: done\n"); +} + /* CORE RED — the streamer must never ship a non-empty cross-term entry empty. */ static void test_2_3_streamer_short_read_must_not_ship_empty(void) { @@ -660,7 +745,16 @@ static void test_2_3_streamer_short_read_must_not_ship_empty(void) if (g_sends[i].lsn == (uint64_t)TGT) send_len = g_sends[i].send_len; hcs_probe_t rcv_pr; - TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&rcv, (uint64_t)TGT, &rcv_pr)); + /* task 2.4: when the fix correctly ABORTS the stream rather than ship empty, + * the receiver gets NOTHING, so its WAL dir has no segment and the probe + * returns -ENOENT (absent) — which equally satisfies the contract "the + * receiver must never hold an empty committed entry". A clean scan (rc==0, + * not-present, e.g. the retry-recovers path persisting the FULL body) is also + * valid; only a present-with-empty body below is a failure. */ + int rcv_rc = hcs_wal_probe(&rcv, (uint64_t)TGT, &rcv_pr); + TEST_ASSERT_TRUE_MESSAGE(rcv_rc == 0 || rcv_rc == -ENOENT, + "receiver WAL probe must scan cleanly (rc==0) or find no segment at all " + "(rc==-ENOENT, the aborted-stream outcome) — never a probe error"); printf("test_2_3 cross-term LSN %d: stored_term=%d leader_term=%d " "source_len=%u SEND_len=%u RECV_len=%u (rc=%d, crossterm_seen=%d)\n", (int)TGT, (int)STORED_TERM, (int)LEADER_TERM, src_pr.len, send_len, @@ -949,6 +1043,9 @@ void test_ha_core_safety(void) * PIN — an unrepairable below-tip catch-up divergence raises the * non-promotable fence (task-2.2 mechanism via the route). */ RUN_TEST(test_2_3_streamer_short_read_must_not_ship_empty); + /* task 2.4 GREEN: the counterpart recover path — a transient short read is + * recovered by the bounded retry and the entry is shipped FULL. */ + RUN_TEST(test_2_4_streamer_short_read_retry_recovers); RUN_TEST(test_2_3_termblind_route_payload_guarded); RUN_TEST(test_2_3_unrepairable_catchup_raises_nonpromotable_fence); /* Scaffolding: each shared seam is usable by 2.3 / 4.1 / 5.1. */