From ae8291ed567096cee3ed8cf1119b1b910ea9cce0 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 27 Jun 2026 11:24:16 +0000 Subject: [PATCH] test(raft-ha-core-safety): cross-term streamer payload-integrity RED (task 2.3) Minimal XCE_TESTING fault-injection seam in ha_stream_wal_to_standby (NULL/production-inert) forces a short body read on a cross-term entry. CORE RED proves the streamer ships PLEN=0 for a full (source_len=64) cross-term entry: instrumentation source_len=64 SEND_len=0 RECV_len=0 -> loss is SEND-side (answers task 1.3's open send-vs-transitive question). 2 honest pins (term-blind route guarded by 2.2; below-tip fence). 11 tests/1 RED; production xce_controller links clean without the seam. --- .kiro/specs/raft-ha-core-safety/tasks.md | 2 +- src/controller/config_repl.c | 27 ++ .../testsuite/test_ha_core_safety.c | 368 +++++++++++++++++- 3 files changed, 385 insertions(+), 12 deletions(-) diff --git a/.kiro/specs/raft-ha-core-safety/tasks.md b/.kiro/specs/raft-ha-core-safety/tasks.md index f983037..7b59a2b 100644 --- a/.kiro/specs/raft-ha-core-safety/tasks.md +++ b/.kiro/specs/raft-ha-core-safety/tasks.md @@ -52,7 +52,7 @@ - _Requirements: 1.1, 1.2, 2.1, 2.2, 2.4, 3.1_ - _Boundary: Persist Conflict Guard_ -- [ ] 2.3 RED: cross-term catch-up streamer payload-integrity tests +- [x] 2.3 RED: cross-term catch-up streamer payload-integrity tests - Add a case reproducing the live signature: a cross-term catch-up (stored term < leader current term, bug-1067 path) whose entry-body `pread` returns short / `malloc` fails must NOT result in a persisted `PLEN=0` entry at a committed LSN — instead the stream must error and retry. Also add the defensive case that a `term==0`-shaped catch-up entry routes through the reconcile seam (no direct-persist bypass), and the unrecoverable-fence case (a node that cannot obtain the majority-agreed committed prefix refuses promotion instead of serving divergent state). Instrument `payload_len` at send and at `on_persist_entry` to disambiguate send-side-empty vs transitive-empty propagation - Observable completion: the streamer case FAILs on the baseline because the short-read/malloc-fail path silently ships `PLEN=0`; the seam-routing and fence cases FAIL because those paths/fence do not exist yet - _Requirements: 1.1, 1.2, 2.1, 3.1, 3.2, 3.3, 7.1_ diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 88d18e3..e46b89d 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -1994,6 +1994,20 @@ int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req) /* Test-only seam: intercepts each batch xce_request() call. * Defined here (translation unit scope); declared extern in test file. */ int (*xce_wal_stream_request_hook)(endpoint_t *ep, xce_request_t *req) = NULL; + +/* raft-ha-core-safety task 2.3 — test-only fault-injection seam for the catch-up + * streamer's entry-body read (the "Catch-up Payload Integrity" boundary). When + * non-NULL AND it returns nonzero for an entry, ha_stream_wal_to_standby treats + * that entry's body pread as a SHORT READ — the same outcome as a malloc-fail — + * which reproduces the live cross-term catch-up payload-loss signature: a + * non-empty source entry is shipped with PLEN=0. NULL in production: behaviour + * is then byte-for-byte unchanged. This seam ONLY makes the defect + * deterministically observable for the task-2.3 RED test; the fix that stops the + * empty ship (fail loud + bounded retry) is task 2.4. Declared extern in the + * test file (same pattern as xce_wal_stream_request_hook). */ +int (*ha_stream_body_read_fault_hook)(uint64_t lsn, uint64_t stored_term, + uint64_t leader_term, + uint32_t payload_len) = NULL; #endif /* send_wal_batch: enqueue a single OPCode_walCatchupEntry batch message. @@ -2156,6 +2170,19 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, 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; diff --git a/src/controller/testsuite/test_ha_core_safety.c b/src/controller/testsuite/test_ha_core_safety.c index a31136a..50ce7ec 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 @@ -49,8 +50,11 @@ #include +#include "config_repl.h" /* ha_stream_wal_to_standby */ #include "ha.h" #include "persist.h" +#include "request.h" /* xce_request_t */ +#include "typemap.h" /* wal_catchup_entry_t */ #include "wal_types.h" #include "walcompare.h" #include "waldump.h" @@ -471,6 +475,349 @@ static void test_2_2_below_tip_conflict_keeps_committed_entries(void) printf("test_2_2_below_tip_conflict_keeps_committed_entries: PASS\n"); } +/* ═══════════════════════════════════════════════════════════════════════════ + * TASK 2.3 — cross-term catch-up streamer payload integrity (RED + PINs) + * + * The live LSN-9 divergence (research.md §"Task 1.3") is SEND-SIDE payload loss. + * The leader-side catch-up streamer ha_stream_wal_to_standby (config_repl.c + * ~2154-2166) reads each entry body with pread into a malloc'd buffer; on a + * SHORT read (or malloc-fail) it silently frees the buffer and ships the entry + * with raw_payload_len==0 — a non-empty COMMITTED entry transmitted as PLEN=0. + * The receiver faithfully persists the empty body, yielding a payload-CRC + * divergence at an identical (term, lsn) the (term,lsn)-only guard never sees. + * The live trigger is the cross-term catch-up (stored term < leader current + * term — the bug-1067 path), so these cases force that exact shape. + * + * (CORE RED) test_2_3_streamer_short_read_must_not_ship_empty: a non-empty + * cross-term entry whose body read returns short must NOT be shipped / + * persisted as PLEN=0 at a committed LSN. FAILS on the baseline (the + * short-read path ships PLEN=0); GREEN when task 2.4 makes that path fail + * loud + bounded-retry so the entry is shipped FULL or not at all. + * + * (DEFENSIVE PIN) test_2_3_termblind_route_payload_guarded: an empty (term==0) + * catch-up entry routed through the receiver intake's term-blind branch + * (ha_apply_wal_catchup_entry → ha_apply_single_entry direct-persist) must + * NOT degrade an existing committed payload — the task-2.2 payload-aware + * guard now protects even this legacy bypass route. Baseline-PASSING PIN + * (NOT a RED); the residual local-term-stamp hazard is retired by task + * 2.4's seam-routing and is not assertable at unit level with c==NULL. + * + * (FENCE PIN) test_2_3_unrepairable_catchup_raises_nonpromotable_fence: a + * below-tip payload conflict ARRIVING VIA the catch-up ingest route (a node + * that received divergent committed state it cannot repair) raises the + * sticky non-promotable fence (persist_recovery_unrecoverable) that the + * promotion gate (ha.c:1072) consults to REFUSE promotion — it keeps every + * committed entry, stays ALIVE, and self-heals. PIN against the task-2.2 + * fence mechanism, exercised through the ingest route (NOT a new RED). + * + * INSTRUMENTATION (the reviewer's open send-side-vs-transitive question): the + * CORE case prints source-WAL payload_len (probe), SEND payload_len (capture at + * the streamer), and RECEIVER payload_len (probe of the persisted body) for the + * target LSN — so a PLEN=0 is unambiguously attributable to the SEND side + * (source full, send empty) rather than a transitive-empty source. + * ═══════════════════════════════════════════════════════════════════════════ */ + +/* Shared ha_config for the ha-state seams (no pool → no delaymgr work). Used by + * the task-2.3 PIN tests and the apply-gap/election seam tests below. */ +static ha_config_t g_hcs_cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 6000, + .election_timeout_ms = 500, + .probe_connect_timeout_ms = 50, + .max_promotion_retries = 3, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 500, +}; + +/* test-only seams defined in config_repl.c (XCE_TESTING); extern'd here, the + * same way test_coldstart_lsn0.c references xce_wal_stream_request_hook. */ +extern int (*xce_wal_stream_request_hook)(endpoint_t *ep, xce_request_t *req); +extern int (*ha_stream_body_read_fault_hook)(uint64_t lsn, uint64_t stored_term, + uint64_t leader_term, + uint32_t payload_len); + +/* Fault target + capture state for the streamer RED. */ +static uint64_t g_fault_lsn; /* LSN to force a short read on */ +static int g_fault_saw_crossterm; /* hook saw stored_term < leader */ +static hcs_member_t* g_recv_member; /* receiver to persist sends into */ + +#define HCS_CAP_MAX 32 +typedef struct { + uint64_t lsn; + uint32_t send_len; /* raw_payload_len actually placed on the wire batch */ + uint64_t term; +} hcs_send_rec_t; +static hcs_send_rec_t g_sends[HCS_CAP_MAX]; +static int g_sends_n; + +static void hcs_2_3_reset(hcs_member_t* recv, uint64_t fault_lsn) +{ + memset(g_sends, 0, sizeof(g_sends)); + g_sends_n = 0; + g_fault_lsn = fault_lsn; + g_fault_saw_crossterm = 0; + g_recv_member = recv; +} + +/* Force a SHORT body read ONLY for the targeted cross-term LSN (stored term < + * leader current term — the bug-1067 catch-up path). */ +static int hcs_2_3_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; + return 1; /* force the short read at the target LSN */ +} + +/* Capture each streamed entry's SEND payload_len AND replay the receive→persist + * leg into g_recv_member. The receiver copies the wire body unconditionally and + * persists it (research.md §"Task 1.3", ha.c:2438), so the receiver WAL ends + * holding exactly what the wire carried — this is the on_persist_entry-side + * payload_len instrument that disambiguates send-side vs transitive empties. */ +static int hcs_2_3_capture_hook(endpoint_t* ep, xce_request_t* req) +{ + (void)ep; + if (!req || !req->body.walcatchupentrymsg) + return 0; + wal_catchup_entry_t* batch = + (wal_catchup_entry_t*)req->body.walcatchupentrymsg; + for (int i = 0; i < batch->n_entries; i++) { + wal_catchup_entry_t* e = batch->entries[i]; + if (!e) + continue; + if (g_sends_n < HCS_CAP_MAX) { + g_sends[g_sends_n].lsn = e->lsn; + g_sends[g_sends_n].send_len = (uint32_t)e->raw_payload_len; + g_sends[g_sends_n].term = e->term; + g_sends_n++; + } + if (g_recv_member && g_recv_member->p) { + persist_set_term(g_recv_member->p, e->term); + persist_wal_write_replicated(g_recv_member->p, e->lsn, + (uint32_t)e->entry_type, + e->raw_payload, + (uint32_t)e->raw_payload_len); + } + } + return 0; +} + +/* Was the target LSN shipped with an empty (PLEN=0) body? */ +static int hcs_2_3_target_shipped_empty(void) +{ + for (int i = 0; i < g_sends_n; i++) + if (g_sends[i].lsn == g_fault_lsn && g_sends[i].send_len == 0) + return 1; + return 0; +} + +/* 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) +{ + hcs_member_t src, rcv; + TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&src, "s23src")); + TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&rcv, "s23rcv")); + + enum { TGT = 1, STORED_TERM = 3, LEADER_TERM = 4 }; + uint8_t full[HCS_FULL_LEN]; + hcs_fill_body(full); + + /* Source (the streaming leader) holds a real non-empty JOB_SUBMIT body at the + * committed target LSN, stored under term 3 < leader current term 4 = the + * live bug-1067 cross-term path. */ + 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 for the inline " + "fdatasync; -EIO here means overlayfs — bug-924)"); + atomic_store_explicit(&src.p->committed_lsn, (uint64_t)TGT, + memory_order_release); + + /* INSTRUMENT (source): the streamed source genuinely holds a FULL body, so a + * PLEN=0 ship below is SEND-side, not a transitive-empty source. */ + hcs_probe_t src_pr; + TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&src, TGT, &src_pr)); + TEST_ASSERT_TRUE_MESSAGE(src_pr.present && src_pr.len == HCS_FULL_LEN, + "PIN: the streamed SOURCE entry must be non-empty (disambiguates " + "send-side-empty from transitive-empty propagation)"); + + /* Drive the REAL streamer with the body-read fault forced at the target. */ + hcs_2_3_reset(&rcv, /*fault_lsn=*/(uint64_t)TGT); + ha_stream_body_read_fault_hook = hcs_2_3_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; + + /* INSTRUMENT (send + receive): show exactly where the bytes were dropped. */ + uint32_t send_len = 0; + for (int i = 0; i < g_sends_n; i++) + 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)); + 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, + rcv_pr.present ? rcv_pr.len : 0u, rc, g_fault_saw_crossterm); + + TEST_ASSERT_TRUE_MESSAGE(g_fault_saw_crossterm, + "the injected fault must have fired on the cross-term (storedcommitted_lsn, 1, memory_order_release); + + /* A term==0-shaped catch-up entry (the legacy term-blind bypass) carrying an + * EMPTY body arrives for the same LSN. Routed through the public intake with + * c==NULL it takes the ha_apply_single_entry direct-persist branch — the + * exact bypass task 2.4 retires — and lands on the persist guard. */ + wal_catchup_entry_t sub; + memset(&sub, 0, sizeof(sub)); + sub.lsn = 1; + sub.term = 0; /* term-blind */ + sub.entry_type = WAL_JOB_SUBMIT; + sub.raw_payload = NULL; + sub.raw_payload_len = 0; + wal_catchup_entry_t* arr[1] = { &sub }; + wal_catchup_entry_t batch; + memset(&batch, 0, sizeof(batch)); + batch.n_entries = 1; + batch.entries = arr; + + int rc = ha_apply_wal_catchup_entry(&ha, m.p, &batch, + /*ep=*/NULL, /*c=*/NULL); + (void)rc; + + /* PIN: the stored committed entry keeps its FULL payload — the payload-aware + * guard no-ops an empty write even when it arrives via the term-blind route. */ + hcs_probe_t pr; + TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&m, 1, &pr)); + TEST_ASSERT_TRUE_MESSAGE(pr.present, "committed LSN 1 must remain present"); + TEST_ASSERT_EQUAL_UINT32_MESSAGE(HCS_FULL_LEN, pr.len, + "PIN: a term-blind empty catch-up must NOT degrade a committed entry to " + "empty (the task-2.2 guard protects the direct-persist bypass route too)"); + TEST_ASSERT_NOT_EQUAL_MESSAGE(0u, pr.pcrc, + "PIN: the committed entry keeps its real payload CRC, not 0"); + + ha_fini(&ha); + hcs_member_down(&m); + printf("test_2_3_termblind_route_payload_guarded: PASS (pin)\n"); +} + +/* FENCE PIN — an unrepairable below-tip divergence arriving via the catch-up + * ingest route raises the non-promotable fence the promotion gate consults. */ +static void test_2_3_unrepairable_catchup_raises_nonpromotable_fence(void) +{ + hcs_member_t m; + TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&m, "s23fence")); + ha_state_t ha; + ha_init(&ha, &g_hcs_cfg, 0); + + enum { N = 4, K = 2 }; + uint8_t full[HCS_FULL_LEN]; + hcs_fill_body(full); + + /* Build committed [1..N] with the entry at K stored EMPTY (the live payload- + * loss shape), so committed entries K+1..N sit ABOVE the conflict. */ + 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), + "contiguous build write must persist (real fs required — bug-924)"); + } + atomic_store_explicit(&m.p->committed_lsn, (uint64_t)N, memory_order_release); + TEST_ASSERT_FALSE_MESSAGE(persist_recovery_unrecoverable(m.p), + "precondition: the member is not fenced before the conflicting catch-up"); + + /* The CORRECT full body for LSN K arrives via the term-blind catch-up route + * (c==NULL → ha_apply_single_entry → persist guard). It conflicts with the + * stored empty body BELOW the tip; the guard must refuse to truncate the + * committed entries above it and instead raise the sticky fence. */ + wal_catchup_entry_t sub; + memset(&sub, 0, sizeof(sub)); + sub.lsn = (uint64_t)K; + sub.term = 0; + sub.entry_type = WAL_JOB_SUBMIT; + sub.raw_payload = full; + sub.raw_payload_len = HCS_FULL_LEN; + wal_catchup_entry_t* arr[1] = { &sub }; + wal_catchup_entry_t batch; + memset(&batch, 0, sizeof(batch)); + batch.n_entries = 1; + batch.entries = arr; + + int rc = ha_apply_wal_catchup_entry(&ha, m.p, &batch, NULL, NULL); + (void)rc; + + /* PIN (a): committed entries above the conflict survive. */ + 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, + "a below-tip catch-up conflict must NOT drop committed entries above it"); + TEST_ASSERT_EQUAL_UINT32(HCS_FULL_LEN, pr.len); + } + /* PIN (b): committed_lsn is never lowered. */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)N, + atomic_load_explicit(&m.p->committed_lsn, memory_order_acquire), + "a below-tip catch-up conflict must NOT regress committed_lsn"); + /* PIN (c): the non-promotable fence is raised — this is the flag the + * promotion gate (ha.c:1072 `!unrecoverable`) consults to REFUSE promotion, + * so the member never serves divergent committed state; it stays ALIVE and + * self-heals on a correct re-stream (the task-2.4 source fix). */ + TEST_ASSERT_TRUE_MESSAGE(persist_recovery_unrecoverable(m.p), + "PIN: an unrepairable catch-up divergence raises the non-promotable " + "fence (same mechanism as task 2.2, exercised through the ingest route)"); + + ha_fini(&ha); + hcs_member_down(&m); + printf("test_2_3_unrepairable_catchup_raises_nonpromotable_fence: PASS (pin)\n"); +} + /* ═══════════════════════════════════════════════════════════════════════════ * SCAFFOLDING (PASS) — prove each shared seam is usable by later phases. * ═══════════════════════════════════════════════════════════════════════════ */ @@ -536,17 +883,6 @@ static void test_seam_divergence_is_observable(void) printf("test_seam_divergence_is_observable: PASS\n"); } -/* Shared ha_config for the ha-state seams (no pool → no delaymgr work). */ -static ha_config_t g_hcs_cfg = { - .heartbeat_interval_ms = 3000, - .heartbeat_timeout_ms = 6000, - .election_timeout_ms = 500, - .probe_connect_timeout_ms = 50, - .max_promotion_retries = 3, - .election_jitter_max_ms = 0, - .election_priority_step_ms = 500, -}; - /* Seam 4: the apply-gap injection seam parks committed > applied and the gap * query reflects it (used by 4.1 to gate "converged"). */ static void test_seam_apply_gap_injection(void) @@ -605,6 +941,16 @@ void test_ha_core_safety(void) /* 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); + /* Task 2.3 cross-term catch-up streamer payload integrity: + * CORE RED — the streamer must never ship a non-empty cross-term entry + * as PLEN=0 on a short body read (FAILs on the baseline). + * PIN — the term-blind receive route cannot degrade a committed + * payload (task-2.2 guard protects the bypass; baseline-pass). + * 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); + 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. */ RUN_TEST(test_seam_wal_construct_and_probe); RUN_TEST(test_seam_divergence_is_observable);