forked from lei.chen/myxce
test(raft-ha-core-safety): end-to-end (term,payload) fidelity across replication + catch-up (task 6.1)
Integration test proves the guarantee tasks 2.3/2.4 hardened in isolation (synthetic fault injection on the streamer) also holds end-to-end over two real persist_t/WAL instances through the real production ingest seam (ha_apply_wal_catchup_entry). Covers live in-term replication and cross-term catch-up: no term==0 sourced for a live entry, no non-empty entry ever lands as PLEN=0, standby persists the entry's stored term (not the AE header's leader-current-term, the bug-1067 split), and walcompare reports a clean committed prefix. test_ha_core_safety 26/0, no regression, no ASAN reports.
This commit is contained in:
parent
a53b9a7b6c
commit
4e1e3e07fb
2 changed files with 262 additions and 1 deletions
|
|
@ -181,7 +181,7 @@
|
|||
- _Depends: 5.9_
|
||||
|
||||
- [ ] 6. Integration: single-seam invariant end-to-end
|
||||
- [ ] 6.1 Verify leader→standby replication carries (term, payload) faithfully end-to-end
|
||||
- [x] 6.1 Verify leader→standby replication carries (term, payload) faithfully end-to-end
|
||||
- Integration check that the live leader replication AND the cross-term catch-up streaming path transmit each entry's full `(term, payload)` to the receiver: no `term==0` on the wire from a live leader (defensive), and — the real concern — no non-empty entry ever arrives with `PLEN=0`; the walAck catch-up continuation feeds only the seam ingest path
|
||||
- Observable completion: an end-to-end test (or instrumented soak round) shows leader and standby hold identical `(term, payload-crc)` for every replicated and cross-term-caught-up LSN; `walcompare --report` clean across members
|
||||
- _Requirements: 2.1, 2.3, 3.2_
|
||||
|
|
|
|||
|
|
@ -2121,6 +2121,259 @@ static void test_seam_election_injection(void)
|
|||
printf("test_seam_election_injection: PASS\n");
|
||||
}
|
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════════
|
||||
* TASK 6.1 — Integration: leader→standby replication carries (term, payload)
|
||||
* faithfully end-to-end, across BOTH live in-term replication and cross-term
|
||||
* catch-up streaming.
|
||||
*
|
||||
* design.md "Catch-up Payload Integrity" + "Unified Entry Ingest" (Req 2.1,
|
||||
* 2.3, 3.2): tasks 2.3/2.4 hardened the streamer's short-read/malloc-fail path
|
||||
* in ISOLATION (synthetic fault injection). This is the end-to-end proof that
|
||||
* the SAME (term, payload) fidelity guarantee holds on the HAPPY PATH across
|
||||
* two REAL persist_t/WAL instances, through the REAL production ingest seam
|
||||
* (ha_apply_wal_catchup_entry -> ha_consume_append_entries ->
|
||||
* raft_on_append_entries_request), for both:
|
||||
*
|
||||
* (1) LIVE in-term replication: N entries appended by a "leader" persist_t
|
||||
* at its current term, relayed one-at-a-time (the live single-entry AE
|
||||
* shape) to a real standby cluster_t. The entry's term is sourced via
|
||||
* persist_entry_term -- the SAME authoritative call
|
||||
* ha_cb_send_append_entries_stub (the real live-send callback, ha.c
|
||||
* ~1689-1731) uses -- so the DEFENSIVE claim "no term==0 on the wire
|
||||
* from a live leader" is grounded in the real sourcing function, not a
|
||||
* fabricated test value. (A real network send is not exercised here --
|
||||
* repl_send_append_entries needs a live endpoint/transport this unit
|
||||
* test has none of -- so, per this task's latitude to drive
|
||||
* ha_consume_append_entries directly, the relay sources its term the
|
||||
* same way the real send callback does and delivers into the same real
|
||||
* receive seam.)
|
||||
*
|
||||
* (2) CROSS-TERM catch-up: the leader's log holds entries stored at an
|
||||
* OLDER term than its current session term (the bug-1067/2.3 shape);
|
||||
* the REAL ha_stream_wal_to_standby streams them (no fault injection --
|
||||
* this is the happy-path proof, not another regression pin for the
|
||||
* already-fixed short-read bug 2.3/2.4 own) through a capture hook that
|
||||
* feeds the decoded batch to the SAME production seam,
|
||||
* ha_apply_wal_catchup_entry, used by (1) -- proving "the walAck
|
||||
* catch-up continuation feeds only the seam ingest path".
|
||||
*
|
||||
* Observable completion (tasks.md 6.1): after BOTH phases, the leader and
|
||||
* standby hold identical (term, payload-crc) for every replicated/caught-up
|
||||
* LSN -- verified by direct (term, pcrc) probes AND by walcompare reporting
|
||||
* OK (the "walcompare --report clean" observable) -- and no non-empty entry
|
||||
* ever lands on the standby as PLEN=0.
|
||||
*
|
||||
* Requirements: 2.1, 2.3, 3.2. Boundary: Catch-up Payload Integrity, Unified
|
||||
* Entry Ingest.
|
||||
* ═══════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
/* Receiver cluster for the 6.1 catch-up capture hook (mirrors g_recv_member's
|
||||
* role for 2.3, but routes through the REAL seam ha_apply_wal_catchup_entry
|
||||
* instead of a raw persist_wal_write_replicated bypass -- so the catch-up leg
|
||||
* of 6.1 also exercises Unified Entry Ingest, not just Catch-up Payload
|
||||
* Integrity). */
|
||||
static cluster_t* g_hcs61_recv_cluster;
|
||||
|
||||
/* Feed each streamed batch to the REAL production ingest seam
|
||||
* (ha_apply_wal_catchup_entry) -- the same function every incoming
|
||||
* OPCode_walCatchupEntry request is routed to, for both live single-entry AEs
|
||||
* and multi-entry catch-up batches (design.md "Unified Entry Ingest"). */
|
||||
static int hcs_6_1_seam_capture_hook(endpoint_t* ep, xce_request_t* req)
|
||||
{
|
||||
if (!req || !req->body.walcatchupentrymsg || !g_hcs61_recv_cluster)
|
||||
return 0;
|
||||
wal_catchup_entry_t* batch =
|
||||
(wal_catchup_entry_t*)req->body.walcatchupentrymsg;
|
||||
ha_apply_wal_catchup_entry(&g_hcs61_recv_cluster->ha,
|
||||
&g_hcs61_recv_cluster->persist, batch, ep,
|
||||
g_hcs61_recv_cluster);
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void test_6_1_replication_and_catchup_preserve_term_payload_e2e(void)
|
||||
{
|
||||
hcs_member_t leader;
|
||||
TEST_ASSERT_EQUAL_INT(0, hcs_member_up(&leader, "61ldr"));
|
||||
cluster_t* standby = hcs4_cluster_up("61std");
|
||||
TEST_ASSERT_NOT_NULL(standby);
|
||||
hcs4_make_follower(standby, /*term=*/0, /*n_peers=*/3);
|
||||
|
||||
uint8_t full[HCS_FULL_LEN];
|
||||
hcs_fill_body(full);
|
||||
|
||||
enum { LIVE_TERM = 6, N_LIVE = 3, CATCHUP_LEADER_TERM = 9, N_CATCHUP = 2 };
|
||||
|
||||
/* ── Phase 1: LIVE in-term replication ───────────────────────────────────
|
||||
* The leader appends N_LIVE entries at its current term; each is relayed
|
||||
* one-at-a-time -- the live single-entry AE shape -- straight into the
|
||||
* standby's REAL ingest seam (ha_apply_wal_catchup_entry). */
|
||||
uint64_t prev_lsn = 0, prev_term = 0;
|
||||
for (uint64_t lsn = 1; lsn <= (uint64_t)N_LIVE; lsn++) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
|
||||
hcs_wal_put(&leader, lsn, (uint64_t)LIVE_TERM, WAL_JOB_SUBMIT, full,
|
||||
HCS_FULL_LEN),
|
||||
"leader live-append write must persist (real fs required — bug-924)");
|
||||
atomic_store_explicit(&leader.p->committed_lsn, lsn, memory_order_release);
|
||||
|
||||
/* Source the entry's term the SAME way the real live-send callback
|
||||
* (ha_cb_send_append_entries_stub) does. */
|
||||
uint64_t entry_term = persist_entry_term(leader.p, lsn);
|
||||
TEST_ASSERT_NOT_EQUAL_MESSAGE(0u, entry_term,
|
||||
"DEFENSIVE: a live leader must never source term==0 for a real "
|
||||
"entry — persist_entry_term (the same call the real live-send "
|
||||
"callback uses) must return the assigned term");
|
||||
TEST_ASSERT_EQUAL_UINT64((uint64_t)LIVE_TERM, entry_term);
|
||||
|
||||
wal_catchup_entry_t sub;
|
||||
memset(&sub, 0, sizeof(sub));
|
||||
sub.lsn = lsn;
|
||||
sub.entry_type = WAL_JOB_SUBMIT;
|
||||
sub.raw_payload = full;
|
||||
sub.raw_payload_len = HCS_FULL_LEN;
|
||||
sub.term = entry_term;
|
||||
sub.prev_log_index = prev_lsn;
|
||||
sub.prev_log_term = prev_term;
|
||||
sub.leader_commit = lsn;
|
||||
/* Live append: the AE header term (leader_current_term) equals the
|
||||
* entry's own term -- the leader appends at its current term. */
|
||||
sub.leader_current_term = entry_term;
|
||||
|
||||
wal_catchup_entry_t* sub_arr[1] = { &sub };
|
||||
wal_catchup_entry_t top;
|
||||
memset(&top, 0, sizeof(top));
|
||||
top.n_entries = 1;
|
||||
top.entries = sub_arr;
|
||||
|
||||
/* The REAL production ingest seam -- the exact function every
|
||||
* incoming WAL replication message (live or catch-up) is routed to. */
|
||||
ha_apply_wal_catchup_entry(&standby->ha, &standby->persist, &top, NULL,
|
||||
standby);
|
||||
|
||||
/* OBSERVE: the standby's on-disk (term, payload) match the leader's,
|
||||
* via the REAL production ingest seam -- no synthetic stand-in. */
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(entry_term,
|
||||
persist_entry_term(&standby->persist, lsn),
|
||||
"the standby must record the SAME term the live leader sourced "
|
||||
"for this entry (no term loss/substitution end-to-end)");
|
||||
|
||||
hcs_probe_t pr;
|
||||
TEST_ASSERT_EQUAL_INT(0, hcs4_probe(standby, lsn, &pr));
|
||||
TEST_ASSERT_TRUE_MESSAGE(pr.present,
|
||||
"a live-replicated entry must be present on the standby");
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(HCS_FULL_LEN, pr.len,
|
||||
"a non-empty live entry must NEVER arrive on the standby as PLEN=0");
|
||||
TEST_ASSERT_NOT_EQUAL_MESSAGE(0u, pr.pcrc,
|
||||
"the live-replicated entry must carry a real payload CRC, not 0");
|
||||
|
||||
prev_lsn = lsn;
|
||||
prev_term = entry_term;
|
||||
}
|
||||
|
||||
/* ── Phase 2: CROSS-TERM catch-up ────────────────────────────────────────
|
||||
* The leader's log holds MORE entries stored at the SAME (older) term as
|
||||
* phase 1, while its session is streamed with a HIGHER current term --
|
||||
* exactly the bug-1067/2.3 "stored term < leader current term" shape.
|
||||
* Stream them via the REAL ha_stream_wal_to_standby (no fault injection --
|
||||
* this proves the happy path; 2.3/2.4 already pin the fault path) through
|
||||
* a capture hook that feeds the REAL production ingest seam. */
|
||||
uint64_t catchup_first = (uint64_t)N_LIVE + 1;
|
||||
uint64_t catchup_last = (uint64_t)N_LIVE + (uint64_t)N_CATCHUP;
|
||||
for (uint64_t lsn = catchup_first; lsn <= catchup_last; lsn++) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
|
||||
hcs_wal_put(&leader, lsn, (uint64_t)LIVE_TERM, WAL_JOB_SUBMIT, full,
|
||||
HCS_FULL_LEN),
|
||||
"leader catch-up-source write must persist (real fs required — "
|
||||
"bug-924)");
|
||||
}
|
||||
atomic_store_explicit(&leader.p->committed_lsn, catchup_last,
|
||||
memory_order_release);
|
||||
|
||||
/* PIN (source-side): the streamed source entries are genuinely non-empty
|
||||
* at the OLDER stored term (disambiguates send-side vs a transitive-empty
|
||||
* source, the same instrument 2.3 uses). */
|
||||
for (uint64_t lsn = catchup_first; lsn <= catchup_last; lsn++) {
|
||||
hcs_probe_t src_pr;
|
||||
TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&leader, lsn, &src_pr));
|
||||
TEST_ASSERT_TRUE_MESSAGE(src_pr.present && src_pr.len == HCS_FULL_LEN,
|
||||
"PIN: the streamed SOURCE entry must be non-empty");
|
||||
TEST_ASSERT_EQUAL_UINT64((uint64_t)LIVE_TERM, src_pr.term);
|
||||
}
|
||||
|
||||
g_hcs61_recv_cluster = standby;
|
||||
xce_wal_stream_request_hook = hcs_6_1_seam_capture_hook;
|
||||
int rc = ha_stream_wal_to_standby((endpoint_t*)NULL, leader.wal_dir, leader.p,
|
||||
/*leader_term=*/(uint64_t)CATCHUP_LEADER_TERM,
|
||||
/*standby_lsn=*/(uint64_t)N_LIVE,
|
||||
/*committed_lsn=*/catchup_last);
|
||||
xce_wal_stream_request_hook = NULL;
|
||||
g_hcs61_recv_cluster = NULL;
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
|
||||
"the happy-path cross-term catch-up stream must complete cleanly");
|
||||
|
||||
for (uint64_t lsn = catchup_first; lsn <= catchup_last; lsn++) {
|
||||
hcs_probe_t pr;
|
||||
TEST_ASSERT_EQUAL_INT(0, hcs4_probe(standby, lsn, &pr));
|
||||
TEST_ASSERT_TRUE_MESSAGE(pr.present,
|
||||
"a cross-term caught-up entry must be present on the standby");
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(HCS_FULL_LEN, pr.len,
|
||||
"a non-empty cross-term entry must NEVER arrive on the standby as "
|
||||
"PLEN=0 (the live LSN-9 signature 2.3/2.4 fixed at the streamer — "
|
||||
"this is the end-to-end happy-path proof it stays fixed)");
|
||||
TEST_ASSERT_NOT_EQUAL_MESSAGE(0u, pr.pcrc,
|
||||
"the caught-up entry must carry a real payload CRC, not 0");
|
||||
|
||||
/* THE cross-term distinction: the standby must record the entry's
|
||||
* STORED term (LIVE_TERM), NOT the AE header's leader-CURRENT term
|
||||
* (CATCHUP_LEADER_TERM) -- persist_entry_term is the STORED-term
|
||||
* accessor, mirroring the leader_current_term-vs-stored-term split
|
||||
* bug-1067 fixed. */
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)LIVE_TERM,
|
||||
persist_entry_term(&standby->persist, lsn),
|
||||
"the standby must persist the entry's STORED term, not the AE "
|
||||
"header's leader-current-term field");
|
||||
}
|
||||
|
||||
/* The standby's SESSION adopts the leader's CURRENT term (UpdateTerm/G5
|
||||
* from the AE header) even though the individual entries keep their
|
||||
* OLDER stored term -- the header-term vs stored-term split holds at
|
||||
* both the session level and the per-entry level. */
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)CATCHUP_LEADER_TERM,
|
||||
standby->ha.raft.term,
|
||||
"the standby must adopt the leader's CURRENT term from the AE header");
|
||||
|
||||
/* ── Cross-member (term, payload-crc) parity over EVERY replicated/caught-
|
||||
* up LSN, both by direct probe and by walcompare (the "walcompare --report
|
||||
* clean" observable). ─────────────────────────────────────────────────── */
|
||||
for (uint64_t lsn = 1; lsn <= catchup_last; lsn++) {
|
||||
hcs_probe_t src_pr, dst_pr;
|
||||
TEST_ASSERT_EQUAL_INT(0, hcs_wal_probe(&leader, lsn, &src_pr));
|
||||
TEST_ASSERT_EQUAL_INT(0, hcs4_probe(standby, lsn, &dst_pr));
|
||||
TEST_ASSERT_TRUE_MESSAGE(src_pr.present && dst_pr.present,
|
||||
"every replicated/caught-up LSN must be present on BOTH members");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(src_pr.term, dst_pr.term,
|
||||
"leader and standby must agree on TERM at every LSN");
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(src_pr.pcrc, dst_pr.pcrc,
|
||||
"leader and standby must agree on PAYLOAD-CRC at every LSN");
|
||||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(HCS_FULL_LEN, dst_pr.len,
|
||||
"no LSN may ever land on the standby as an empty (PLEN=0) body");
|
||||
}
|
||||
|
||||
walcompare_member_t wm[2] = {
|
||||
{ leader.wal_dir, catchup_last },
|
||||
{ standby->persist.wal_path, catchup_last },
|
||||
};
|
||||
walcompare_divergence_t div;
|
||||
memset(&div, 0, sizeof(div));
|
||||
int verdict = walcompare(wm, 2, NULL, &div);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(WALCMP_OK, verdict,
|
||||
"walcompare must report a CLEAN committed prefix across leader and "
|
||||
"standby after live replication + cross-term catch-up");
|
||||
|
||||
hcs4_cluster_down(standby);
|
||||
hcs_member_down(&leader);
|
||||
printf("test_6_1_replication_and_catchup_preserve_term_payload_e2e: done\n");
|
||||
}
|
||||
|
||||
/* ── Suite entry point ─────────────────────────────────────────────────────*/
|
||||
|
||||
void test_ha_core_safety(void)
|
||||
|
|
@ -2198,5 +2451,13 @@ void test_ha_core_safety(void)
|
|||
RUN_TEST(test_seam_divergence_is_observable);
|
||||
RUN_TEST(test_seam_apply_gap_injection);
|
||||
RUN_TEST(test_seam_election_injection);
|
||||
/* Task 6.1 integration: leader->standby replication carries (term,
|
||||
* payload) faithfully end-to-end across BOTH live in-term replication and
|
||||
* cross-term catch-up streaming, through the REAL production ingest seam
|
||||
* (ha_apply_wal_catchup_entry). Proves the happy-path invariant 2.3/2.4
|
||||
* hardened in isolation also holds end-to-end over two real persist_t/WAL
|
||||
* instances: no term==0 on the wire from a live leader, no non-empty
|
||||
* entry ever lands as PLEN=0, and walcompare reports a clean prefix. */
|
||||
RUN_TEST(test_6_1_replication_and_catchup_preserve_term_payload_e2e);
|
||||
UNITY_END();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue