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

fix(raft-ha-core-safety): streamer fail-loud bounded-retry, never ship empty (task 2.4)

Browse source 
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.
This commit is contained in:
Rongsong Shen 2026-06-27 15:42:37 +00:00
parent ae8291ed56
commit 63b9c5a81a
3 changed files with 210 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

8
.kiro/specs/raft-ha-core-safety/tasks.md
View file

@ -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.

129
src/controller/config_repl.c
View file

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

99
src/controller/testsuite/test_ha_core_safety.c
View file

@ -41,6 +41,7 @@
* Requirements: 7.1 (force divergence RED on baseline), 7.4 (unit harness).
*/
#define _GNU_SOURCE
#include <errno.h>
#include <inttypes.h>
#include <stdatomic.h>
#include <stdint.h>
@ -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. */