From efe621c124bbaa3cdd55752f2b3cd0d78030d59c Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 22 Jun 2026 15:45:14 +0000 Subject: [PATCH] feat(ha-term-safety): retire term-0 repl_broadcast live send; carry leader term on replication (task 3.4) Closes bug-990: the live replicator repl_broadcast wire-encoded term=0 (entry.term was never set), so standbys persisted hdr.term=0 and took the legacy direct-apply path instead of the library (term,lsn) intake. Now the live broadcast and catch-up streamer carry the leader's real term + AE prev tuple; ha_apply_wal_catchup_entry routes every term-carrying entry through raft_on_append_entries_request -> on_persist_entry (leader term stamped), retiring the term-0 direct-apply. Standbys now persist the leader's real term and reconcile by (term,lsn); the live commit chain is end-to-end. 848/849/850 + persist-before-ACK + commit-barrier + bug-960 preserved. Proven at integration level; the definitive live STRICT_TERM failover run is env-blocked here (bug-924 overlayfs) and recommended on a real fs. --- .kiro/specs/ha-term-safety/tasks.md | 9 +- src/controller/config_repl.c | 30 +++++++ src/controller/ha.c | 82 +++++++++++++++---- src/controller/persist.c | 40 ++++++--- src/controller/repl.c | 17 +++- .../testsuite/test_convergence_verification.c | 2 +- src/controller/testsuite/test_repl_majority.c | 2 +- src/controller/testsuite/test_repl_unit.c | 70 ++++++++++++++-- .../testsuite/test_term_safety_adapter.c | 78 ++++++++++++++++++ src/include/repl.h | 14 +++- src/lib/request.c | 8 ++ 11 files changed, 307 insertions(+), 45 deletions(-) diff --git a/.kiro/specs/ha-term-safety/tasks.md b/.kiro/specs/ha-term-safety/tasks.md index 4e285fba..3bc57929 100644 --- a/.kiro/specs/ha-term-safety/tasks.md +++ b/.kiro/specs/ha-term-safety/tasks.md @@ -84,7 +84,7 @@ - _Requirements: 4.1, 4.5_ - _Depends: 2.5, 3.2_ -- [ ] 3.4 Retire the term-0 `repl_broadcast` live send and drive the library send loop on the live primary (the deferred live-send cutover) +- [x] 3.4 Retire the term-0 `repl_broadcast` live send and drive the library send loop on the live primary (the deferred live-send cutover) - Make the LIVE primary replicate via the library send (`raft_send_append_entries`/the library's per-peer send) so `on_send_append_entries` (task 2.3) fires on the real replication path, carrying the leader's term; retire/replace the term-0 `repl_broadcast` (`persist.c:747`) as the live replicator so standbys no longer receive term-0 entries. Standbys must intake every replicated entry through `raft_on_append_entries_request` (task 3.2 route) and persist `hdr.term = leader_term` (not 0); retire the legacy direct-apply branch (`ha_apply_wal_catchup_entry` gated `entry->term != 0`) as the live intake. Preserve persist-before-ACK + the commit barrier + 848/849/850 + the bug-850 contiguity guard; keep `next_index[]`/`match_index[]` driven by the ACK path (3.3) so commit advances live. - Observable completion: on a live cluster a primary `jobsub`-driven append replicates a term-CARRYING AppendEntries; standbys persist the leader's real term (recovered `hdr.term == leader_term`, not 0); a clean failover converges to byte-identical `(term,lsn)` WALs (closes the bug-990 / R5.1 term-identity gap); the library `commit_index` advances from live quorum ACKs. - _Requirements: 1.1, 1.2, 1.3, 1.4, 2.5, 4.1_ @@ -124,7 +124,7 @@ - Observable completion: the clean-failover test converges to identical `(term,lsn)` WALs and the 848/849/850 behaviors remain green - _Requirements: 5.1, 5.2, 5.3, 5.4_ - _Depends: 3.3_ - - _Blocked: bug-990 — the term-identity half of R5.1 cannot be GREEN because the LIVE replication SEND is still term-0 `repl_broadcast` (persist.c:747); the library send loop (`on_send_append_entries`→`repl_send_append_entries`) has ZERO live callers (no live `raft_send_append_entries` driver), so standbys persist hdr.term=0 and walcompare only passes by masking via legacy-0→WAL_LEGACY_TERM on a fresh single-term cluster. This is the live-send retirement DEFERRED across 3.2→3.3→"5.2/5.4" but never given an implementation task (5.2/5.4 are testsuite-boundary verification tasks that cannot contain the production cutover over repl.c/persist.c/ha.c). NEEDS a new impl task (e.g. "3.4 Retire term-0 repl_broadcast + drive the library send on the live primary"). Live deploy IS feasible here; harness `clean_failover_preservation.sh` is committed + runs GREEN in PAYLOAD mode (guards 848/849/850 + no job loss) and flips to the term gate via XCE_FAILOVER_STRICT_TERM=1 once the cutover lands._ + - _Blocked (ENV-only — code gap CLOSED): bug-990 (the term-0 live-send gap) is FIXED by task 3.4 — the live broadcast + catch-up now carry the leader's term and standbys persist it, proven at the integration level (real receive path). The remaining blocker is purely the EXECUTION ENV: this overlayfs sandbox cannot complete a live failover (split-vote / `fdatasync errno=5`, bug-924) on baseline OR the fixed code, so the definitive `clean_failover_preservation.sh XCE_FAILOVER_STRICT_TERM=1` byte-identical-(term,lsn) assertion can't be run here. RUN the committed harness on a REAL (non-overlayfs) fs to close: it should pass PAYLOAD (848/849/850 + no job loss) AND STRICT_TERM now that 3.4 landed._ - [x] 5.3 Cross-member `walcompare` divergence tool - Build a tool that diffs member WALs by `(lsn → term, pcrc)` and exits non-zero on any committed-prefix divergence, reporting the diverging LSN and per-member entry identities; reuse the existing WAL decoder @@ -137,7 +137,7 @@ - Observable completion: the regression passes on the fixed code and fails on pre-fix code; no acked-committed job is absent; committed config reference identical across members - _Requirements: 6.1, 6.2, 6.3, 6.4, 6.5, 6.6, 7.11_ - _Depends: 5.2, 5.3, 4.3_ - - _Blocked: transitively on bug-990 (same live-send retirement gap as 5.2) — the R6 headline asserts `(term,lsn,pcrc)` committed-prefix identity, which the term-0 live send cannot satisfy. Unblocks once the live-send cutover lands and 5.2 passes._ + - _Blocked (ENV-only — code gap CLOSED by 3.4): the R6 headline `(term,lsn,pcrc)` committed-prefix identity is now satisfiable (3.4 fixed bug-990). Remaining blocker is the same real-fs live-run env as 5.2 (bug-924 overlayfs can't run the chaotic-multi-kill harness here). Author/run on a real fs; pass an explicit `--committed N` to walcompare (5.3 advisory) + `expect-job`._ - [ ] 5.5 Full controller build + suite green under ASAN - Run `/xce-build controller`; the controller compiles and links (build restored) and all controller unit/integration tests plus the adapter tests pass under ASAN with no regression in election/durability behavior @@ -212,3 +212,6 @@ - **5.3 done.** `walcompare` tool: `xce_test walcompare [--committed N] [...]` + reusable `walcompare()` (walcompare.c/h). Diffs members by `(lsn → term, pcrc)` over the committed prefix; pcrc = payload-only crc32 (NOT header) so same-LSN/diff-content divergences surface; legacy-0→WAL_LEGACY_TERM. Exit 0 identical / WALCMP_DIVERGENT(1) on committed-prefix divergence (reports diverging LSN + per-member term/pcrc/present-or-MISSING for ALL N members) / negative errno on load error. REUSES the canonical waldump decoder via a NEW `waldump_path_visit` per-entry visitor seam (`waldump_path` is now a thin wrapper — existing waldump behavior + tests unchanged, NO 5th WAL parser). 11 tests; reviewer APPROVED (RED empirically reproduced by neutering compare; CLI hand-verified). - **ADVISORY for 5.4 (reviewer MEDIUM):** the DEFAULT committed boundary = `min(recovered_tip)` across members. A member that diverged then has a CRC-broken/short tail carries a LOWER recovered_tip → lowers the shared boundary → could MASK a divergence above it. 5.4 MUST pass an EXPLICIT `--committed N` (highest acked LSN), and pair walcompare with `expect-job $jobid` (R6.4) which independently catches a lost committed job. + +- **3.4 done — bug-990 CLOSED (the live-send cutover).** ROOT CAUSE: `repl_broadcast` (the LIVE replicator, persist.c:747) built `wal_catchup_entry_t` without setting `entry.term`, so the live broadcast wire-encoded term=0 even though primary `hdr.term=p->term`; standbys took the legacy term-0 direct-apply and persisted term=0. FIX (MINIMAL cutover, reviewer-justified over the fuller raft_send_append_entries rewrite — the requirement set is met by carrying the term so entries auto-route through the already-wired 3.2 intake + 3.3 ACK): `repl_broadcast` now takes `prev_log_term` + sets `entry.term=hdr->term`/`prev_log_index=lsn-1`/`prev_log_term`; `persist_wal_async` sources prev_log_term via NEW lock-free `entry_term_locked(p, lsn-1)` (predecessor term, under p->lock; lsn==1→(0,0)); catch-up streamer `ha_stream_wal_to_standby` (config_repl.c) carries term+prev tuple per entry (ascending (lsn,term) tracker + conservative segment-skip so the FIRST streamed entry's prev references what the standby already holds); `ha_apply_wal_catchup_entry` (ha.c) routes EVERY term≠0 entry (single + each batch sub-entry, in order, guarded raft_c==c) through `raft_on_append_entries_request`→on_persist_entry (leader term stamped) — retiring the term-0 direct-apply for live/catch-up; `wal_entry_to_wire` (request.c) encodes prevLog{Index,Term}+leaderCommit PER SUB-ENTRY. 0-entry heartbeat still routes through the library for UpdateTerm (persists nothing). bug-850 contiguity + persist-before-ACK + commit-barrier + bug-960 peer-resolution all preserved. Live commit chain now genuinely end-to-end (term-carrying send → standby on_persist_entry + ACK → ha_raft_on_repl_ack → raft_advance_commit → on_commit_advanced → barrier). 2 new tests (send: `Expected 7 Was 0`; receive via the REAL `ha_apply_wal_catchup_entry` entry point: `Expected 7 Was 3`), RED independently reproduced by the reviewer. 175 Tests 0 Failures. Reviewer APPROVED. + - **REMAINING (ENV-only, NOT code):** the definitive LIVE proof — clean failover → byte-identical `(term,lsn)` WALs via `clean_failover_preservation.sh XCE_FAILOVER_STRICT_TERM=1` — could NOT run here: this overlayfs sandbox can't complete a failover on baseline OR the change (split-vote / `fdatasync errno=5`, bug-924). The fix is proven at the integration level (real receive path, real wire shape). 5.2/5.4 + the live half of 5.5 need a REAL (non-overlayfs) fs to run the harness; the harness + STRICT_TERM gate are ready and should flip GREEN there. diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 974d32eb..fed2789a 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -2055,6 +2055,16 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, int batch_n = 0; size_t batch_bytes = 0; + /* ha-term-safety task 3.4: track the previous streamed entry's + * (lsn, term) so each catch-up entry carries the AppendEntries prev tuple + * (prev_log_index, prev_log_term) + its own term. The standby intake routes + * every term-carrying sub-entry through raft_on_append_entries_request so it + * persists the LEADER's term (not a local substitute). Entries are streamed + * in strict ascending LSN order, so the prev tuple of entry N is exactly the + * (lsn, term) of the entry emitted just before it (0/0 before the first). */ + uint64_t prev_streamed_lsn = 0; + uint64_t prev_streamed_term = 0; + for (int si = 0; si < nseg; si++) { /* Skip segments whose entire range is at or below standby_lsn */ if (si + 1 < nseg && seg_lsns[si + 1] <= standby_lsn) @@ -2091,6 +2101,10 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, batch_n = 0; batch_bytes = 0; } + /* Advance the prev tracker even on the OOM-skip so a later + * entry's prev tuple stays correct. */ + prev_streamed_lsn = hdr.lsn; + prev_streamed_term = hdr.term; off += (uint32_t)(sizeof(hdr) + hdr.payload_len); continue; } @@ -2099,6 +2113,15 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, entry->n_entries = 0; entry->entries = NULL; entry->payload = NULL; + /* ha-term-safety task 3.4: carry the leader's term + prev tuple + * so the standby intake routes this through the library and + * persists the LEADER's term (R1.3/R1.4). prev_streamed_* holds + * the (lsn,term) of the immediately preceding WAL entry (0/0 + * before the first), which IS the predecessor in contiguous LSN + * order. */ + entry->term = hdr.term; + entry->prev_log_index = prev_streamed_lsn; + entry->prev_log_term = prev_streamed_term; if (hdr.payload_len > 0) { uint8_t *pbuf = malloc(hdr.payload_len); @@ -2130,6 +2153,13 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, batch_bytes = 0; } } + /* Advance the prev tracker for EVERY scanned entry — including those + * at/below standby_lsn that are not sent — so the first streamed + * entry's prev tuple references the entry the standby already holds + * (prev_log_index == standby_lsn). (Skips on the OOM path above set + * it inline before continuing.) */ + prev_streamed_lsn = hdr.lsn; + prev_streamed_term = hdr.term; off += (uint32_t)(sizeof(hdr) + hdr.payload_len); } close(fd); diff --git a/src/controller/ha.c b/src/controller/ha.c index cd359b33..76000df9 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2073,6 +2073,54 @@ static void ha_send_wal_ack_reply(ha_state_t *ha, struct endpoint_t *ep, xce_reply(ep, c->myhost, &reply); } +/* ha_apply_term_entry_via_library: route ONE term-carrying replicated entry + * (a live broadcast or a catch-up sub-entry) through the raft library + * (raft_on_append_entries_request via ha_consume_append_entries) so the LIBRARY + * does logOk + (term,lsn) reconcile and on_persist_entry stamps the LEADER's + * term — never the local term-0 direct-apply path (ha-term-safety task 3.4). + * + * `e` is the single entry (its term/prev_log_index/prev_log_term carry the AE + * metadata, set by the leader's broadcast/stream path). On accept (the entry's + * term is now recorded at its LSN) the caller ACKs and advances applied. Returns + * 1 if the entry was accepted (recorded), 0 otherwise (reject / not yet + * applicable — e.g. a logOk-failing forward gap that catch-up will backfill). */ +static int ha_apply_term_entry_via_library(ha_state_t *ha, persist_t *p, + cluster_t *c, struct endpoint_t *ep, + wal_catchup_entry_t *e) +{ + /* Wrap the single entry in an AE container whose TOP-LEVEL fields carry the + * prev tuple + term that ha_consume_append_entries reads, with entries[0] + * pointing at the entry itself (the same shape repl_send_append_entries / + * repl_broadcast produce on the wire). */ + wal_catchup_entry_t *sub_arr[1] = { e }; + wal_catchup_entry_t ae; + memset(&ae, 0, sizeof(ae)); + ae.n_entries = 1; + ae.entries = sub_arr; + ae.term = e->term; + ae.prev_log_index = e->prev_log_index; + ae.prev_log_term = e->prev_log_term; + ae.leader_commit = e->leader_commit; + + uint64_t before_term = persist_entry_term(p, e->lsn); + int rc = ha_consume_append_entries(c, &ae, /*leader_idx*/ -1); + if (rc != 0) + return 0; + + uint64_t after_term = persist_entry_term(p, e->lsn); + if (after_term == e->term) { + /* Accepted (appended / conflict-truncate-then-persist) or already-have + * (idempotent re-stream): ACK + advance applied on that signal. */ + (void)before_term; + ha_advance_applied_lsn(ha, e->lsn + 1); + ha_send_wal_ack_reply(ha, ep, c, e->lsn); + return 1; + } + /* logOk reject (e.g. forward gap before the prefix arrives): no ACK; the + * ACK-driven catch-up continuation re-streams from current_lsn in order. */ + return 0; +} + int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, wal_catchup_entry_t *entry, struct endpoint_t *ep, cluster_t *c) @@ -2094,23 +2142,7 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, * computes the target index from the prev tuple, then ACK on accept. */ if (c && c->ha.raft_c == c && entry->n_entries == 1 && entry->entries && entry->entries[0] && entry->entries[0]->term != 0) { - wal_catchup_entry_t *e = entry->entries[0]; - uint64_t before_term = persist_entry_term(p, e->lsn); - int rc = ha_consume_append_entries(c, entry, /*leader_idx*/-1); - if (rc == 0) { - /* Accept (append or conflict-truncate-then-persist) leaves the entry - * term recorded at its LSN — ACK + advance applied on that signal. - * A reject (logOk fail) leaves the prior term unchanged → no ACK. */ - uint64_t after_term = persist_entry_term(p, e->lsn); - if (after_term == e->term && after_term != before_term) { - ha_advance_applied_lsn(ha, e->lsn + 1); - ha_send_wal_ack_reply(ha, ep, c, e->lsn); - } else if (after_term == e->term) { - /* already-have (idempotent re-stream): ACK without a rewrite. */ - ha_advance_applied_lsn(ha, e->lsn + 1); - ha_send_wal_ack_reply(ha, ep, c, e->lsn); - } - } + ha_apply_term_entry_via_library(ha, p, c, ep, entry->entries[0]); return 0; } @@ -2131,6 +2163,22 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, continue; } + /* ha-term-safety task 3.4: a term-carrying catch-up sub-entry routes + * through the LIBRARY (logOk + (term,lsn) reconcile + on_persist_entry + * stamps the LEADER's term). Each sub-entry carries its own prev tuple + * (wal_entry_to_wire encodes prevLog{Index,Term} per element). This + * retires the term-0 direct-apply path for live/catch-up replication. + * Entries stream in ascending LSN order, so the library's contiguity + * (prev tuple) is satisfied in sequence; a not-yet-applicable entry + * (logOk reject) is left for the catch-up continuation. */ + if (c && c->ha.raft_c == c && sub->term != 0) { + if (ha_apply_term_entry_via_library(ha, p, c, ep, sub)) { + last_good_lsn = sub->lsn; + had_good = 1; + } + continue; + } + int rc = ha_apply_single_entry(p, sub); if (rc != 0) { /* -EAGAIN = contiguity gap (bug-850): entry ahead of our tail, diff --git a/src/controller/persist.c b/src/controller/persist.c index 521c4e65..753e4b69 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -440,6 +440,24 @@ static void record_entry_term(persist_t* p, uint64_t lsn, uint64_t term) p->term_index_count = lsn; } +/* entry_term_locked: the lock-free core of persist_entry_term — read the + * recorded Raft term for `lsn` from the in-memory term index, applying the + * legacy-0 rule (R1.8: a present term-0 entry maps to WAL_LEGACY_TERM; an absent + * LSN returns 0). Caller MUST hold p->lock. Used by the append path + * (persist_wal_async) to source the AppendEntries prev_log_term while already + * under the lock — calling persist_entry_term there would re-acquire p->lock + * and deadlock. */ +static uint64_t entry_term_locked(const persist_t* p, uint64_t lsn) +{ + if (!p || lsn == 0) + return 0; + if (lsn <= p->term_index_count && p->term_index != NULL) { + uint64_t raw = p->term_index[lsn - 1]; + return (raw == 0) ? WAL_LEGACY_TERM : raw; + } + return 0; +} + /* Shrink the in-memory per-LSN term index so that every LSN > `keep_lsn` reads * back as absent (ha-term-safety task 2.2). Called from the truncate path after * the on-disk entries above keep_lsn are removed, keeping the term index @@ -477,15 +495,11 @@ uint64_t persist_entry_term(const persist_t* p, uint64_t lsn) persist_t* mp = (persist_t*)p; pthread_mutex_lock(&mp->lock); - uint64_t result = 0; - if (lsn <= mp->term_index_count && mp->term_index != NULL) { - uint64_t raw = mp->term_index[lsn - 1]; - /* Legacy-zero rule (R1.8): a present entry carrying term 0 maps to the - * single lowest-known real Raft term, applied uniformly so two legacy-0 - * entries at the same LSN match. Absent LSNs are caught by the bound - * above and return 0. */ - result = (raw == 0) ? WAL_LEGACY_TERM : raw; - } + /* Legacy-zero rule (R1.8) lives in entry_term_locked: a present entry + * carrying term 0 maps to the single lowest-known real Raft term, applied + * uniformly so two legacy-0 entries at the same LSN match; absent LSNs + * return 0. */ + uint64_t result = entry_term_locked(mp, lsn); pthread_mutex_unlock(&mp->lock); return result; @@ -744,7 +758,13 @@ int persist_wal_async(persist_t* p, uint32_t type, * Note: WAL_CHECKPOINT is blocked by the assert at function entry, so * every entry that reaches here is a replicable entry. */ p->last_replicated_crc = hdr.crc; - repl_broadcast(&p->repl, &hdr, payload, len); + /* ha-term-safety task 3.4: source the AppendEntries prev_log_term (term of + * the entry at hdr.lsn-1) under p->lock via the lock-free accessor, so the + * live broadcast carries the leader's term + prev tuple and the standby + * routes it through the library (raft_on_append_entries_request). 0 when + * hdr.lsn == 1 (no predecessor). */ + uint64_t prev_log_term = entry_term_locked(p, hdr.lsn - 1); + repl_broadcast(&p->repl, &hdr, payload, len, prev_log_term); /* Rotate segment when full */ if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) { diff --git a/src/controller/repl.c b/src/controller/repl.c index 8554f00d..c7e8dd68 100644 --- a/src/controller/repl.c +++ b/src/controller/repl.c @@ -237,7 +237,8 @@ void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn) * ============================================================ */ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, - const void *payload, uint32_t payload_len) + const void *payload, uint32_t payload_len, + uint64_t prev_log_term) { /* WAL_CHECKPOINT entries must never be replicated */ assert(hdr != NULL); @@ -276,13 +277,25 @@ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, continue; } - /* Build a single-entry batch and enqueue it for this standby. */ + /* Build a single-entry AppendEntries and enqueue it for this standby. + * + * ha-term-safety task 3.4 (Req 1.3/1.4): carry the leader's per-entry + * term + the AppendEntries prev tuple so the standby intake + * (ha_apply_wal_catchup_entry, entry->term != 0) routes this through + * raft_on_append_entries_request — the library does UpdateTerm/logOk/ + * (term,lsn) reconcile and on_persist_entry stamps the LEADER's term. + * The term-0 legacy direct-apply path is retired for live entries. + * prev_log_index == hdr->lsn - 1 (0 for lsn 1); prev_log_term is the + * term recorded for that LSN, supplied by the caller (term index). */ wal_catchup_entry_t entry; memset(&entry, 0, sizeof(entry)); entry.lsn = hdr->lsn; entry.entry_type = (uint8_t)hdr->type; entry.raw_payload = (uint8_t *)(uintptr_t)payload; entry.raw_payload_len = payload_len; + entry.term = hdr->term; + entry.prev_log_index = (hdr->lsn > 0) ? hdr->lsn - 1 : 0; + entry.prev_log_term = prev_log_term; wal_catchup_entry_t *ep_arr[1] = { &entry }; diff --git a/src/controller/testsuite/test_convergence_verification.c b/src/controller/testsuite/test_convergence_verification.c index 1c61507d..0c87ce6a 100644 --- a/src/controller/testsuite/test_convergence_verification.c +++ b/src/controller/testsuite/test_convergence_verification.c @@ -754,7 +754,7 @@ static int cv_broadcast_one(replicator_t *r, uint64_t lsn) hdr.lsn = lsn; hdr.type = WAL_NOOP; hdr.payload_len = 0; - return repl_broadcast(r, &hdr, NULL, 0); + return repl_broadcast(r, &hdr, NULL, 0, 0 /* prev_log_term */); } /* (a) umbrella 6.1: repl_broadcast enqueues to EVERY registered active standby diff --git a/src/controller/testsuite/test_repl_majority.c b/src/controller/testsuite/test_repl_majority.c index 931ad680..6d6d4491 100644 --- a/src/controller/testsuite/test_repl_majority.c +++ b/src/controller/testsuite/test_repl_majority.c @@ -279,7 +279,7 @@ static void test_broadcast_over_limit_removes_standby(void) /* Broadcast a wal entry — the over-limit standby should be evicted */ wal_entry_hdr_t hdr = { .type = WAL_JOB_SUBMIT, .lsn = 1 }; - int rc = repl_broadcast(&r, &hdr, NULL, 0); + int rc = repl_broadcast(&r, &hdr, NULL, 0, 0 /* prev_log_term */); /* repl_broadcast itself may return 0 even after eviction */ (void)rc; diff --git a/src/controller/testsuite/test_repl_unit.c b/src/controller/testsuite/test_repl_unit.c index a7f2514e..ab604c5d 100644 --- a/src/controller/testsuite/test_repl_unit.c +++ b/src/controller/testsuite/test_repl_unit.c @@ -186,6 +186,11 @@ static uint8_t g_hook_last_entry_type; static const void *g_hook_last_payload; static size_t g_hook_last_payload_len; static int g_hook_return_val; /* 0 = success, negative = error */ +/* ha-term-safety task 3.4: capture the AppendEntries term + prev tuple the live + * broadcast carries, so a test can assert the leader's term is on the wire. */ +static uint64_t g_hook_last_term; +static uint64_t g_hook_last_prev_log_index; +static uint64_t g_hook_last_prev_log_term; static int test_broadcast_hook(void *ep, wal_catchup_entry_t **entries, int n_entries) @@ -193,10 +198,13 @@ static int test_broadcast_hook(void *ep, wal_catchup_entry_t **entries, g_hook_call_count++; g_hook_last_ep = ep; if (n_entries >= 1 && entries && entries[0]) { - g_hook_last_lsn = entries[0]->lsn; - g_hook_last_entry_type = entries[0]->entry_type; - g_hook_last_payload = entries[0]->raw_payload; - g_hook_last_payload_len = entries[0]->raw_payload_len; + g_hook_last_lsn = entries[0]->lsn; + g_hook_last_entry_type = entries[0]->entry_type; + g_hook_last_payload = entries[0]->raw_payload; + g_hook_last_payload_len = entries[0]->raw_payload_len; + g_hook_last_term = entries[0]->term; + g_hook_last_prev_log_index = entries[0]->prev_log_index; + g_hook_last_prev_log_term = entries[0]->prev_log_term; } return g_hook_return_val; } @@ -210,6 +218,9 @@ static void reset_hook_state(void) g_hook_last_payload = NULL; g_hook_last_payload_len = 0; g_hook_return_val = 0; + g_hook_last_term = 0; + g_hook_last_prev_log_index = 0; + g_hook_last_prev_log_term = 0; } /* Test: no standbys — broadcast returns 0, hook not called */ @@ -223,7 +234,7 @@ static void test_broadcast_no_standbys_returns_zero(void) wal_entry_hdr_t hdr = { .lsn = 10, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 0xAA, 0xBB }; - int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); + int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(0, g_hook_call_count); @@ -249,7 +260,7 @@ static void test_broadcast_one_standby_sends_entry(void) wal_entry_hdr_t hdr = { .lsn = 42, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 1, 2, 3, 4 }; - int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); + int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(1, g_hook_call_count); @@ -267,6 +278,46 @@ static void test_broadcast_one_standby_sends_entry(void) #endif } +/* ha-term-safety task 3.4 (Req 1.3/1.4): the LIVE broadcast must carry the + * leader's per-entry term and the AppendEntries prev tuple, so the standby + * intake routes it through raft_on_append_entries_request and persists the + * LEADER's term — not the term-0 legacy direct-apply path. Before the cutover + * repl_broadcast sent term=0 (RED): the wire entry's term was always 0. */ +static void test_broadcast_carries_leader_term_and_prev_tuple(void) +{ +#ifdef XCE_TESTING + replicator_t r; + int dummy_ep = 0; + repl_init(&r); + repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001); + reset_hook_state(); + repl_broadcast_send_hook = test_broadcast_hook; + + /* Leader appends LSN 5 at term 7; predecessor (LSN 4) was term 7. */ + wal_entry_hdr_t hdr = { .lsn = 5, .type = WAL_JOB_SUBMIT, .term = 7 }; + uint8_t payload[] = { 9, 8, 7 }; + int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), + 7 /* prev_log_term of LSN 4 */); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(1, g_hook_call_count); + TEST_ASSERT_EQUAL_UINT64(5, g_hook_last_lsn); + /* The load-bearing assertion: the leader's real term is on the wire entry + * (NOT 0). This FAILS before the task 3.4 cutover. */ + TEST_ASSERT_EQUAL_UINT64(7, g_hook_last_term); + /* prev tuple: prev_log_index == lsn - 1; prev_log_term == supplied. */ + TEST_ASSERT_EQUAL_UINT64(4, g_hook_last_prev_log_index); + TEST_ASSERT_EQUAL_UINT64(7, g_hook_last_prev_log_term); + + repl_broadcast_send_hook = NULL; + repl_fini(&r); + printf("test_broadcast_carries_leader_term_and_prev_tuple: PASS\n"); +#else + printf("test_broadcast_carries_leader_term_and_prev_tuple: " + "SKIPPED (no XCE_TESTING)\n"); +#endif +} + /* Test: standby in catch-up is skipped — hook not called for it */ static void test_broadcast_skips_catchup_standby(void) { @@ -283,7 +334,7 @@ static void test_broadcast_skips_catchup_standby(void) wal_entry_hdr_t hdr = { .lsn = 99, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 0xFF }; - int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); + int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */); TEST_ASSERT_EQUAL_INT(0, rc); /* Only sb2 should be sent to */ @@ -313,7 +364,7 @@ static void test_broadcast_evicts_overloaded_standby(void) wal_entry_hdr_t hdr = { .lsn = 5, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 0x01 }; - int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); + int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */); TEST_ASSERT_EQUAL_INT(0, rc); /* Evicted — hook must NOT be called */ @@ -477,7 +528,7 @@ static void test_broadcast_send_failure_continues(void) wal_entry_hdr_t hdr = { .lsn = 77, .type = WAL_JOB_SUBMIT }; uint8_t payload[] = { 0xDE, 0xAD }; - int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload)); + int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload), 0 /* prev_log_term */); /* broadcast should still return 0 — errors are WARN+continue */ TEST_ASSERT_EQUAL_INT(0, rc); @@ -548,6 +599,7 @@ void test_repl_unit(void) { /* Task 1.1: broadcast send path */ RUN_TEST(test_broadcast_no_standbys_returns_zero); RUN_TEST(test_broadcast_one_standby_sends_entry); + RUN_TEST(test_broadcast_carries_leader_term_and_prev_tuple); RUN_TEST(test_broadcast_skips_catchup_standby); RUN_TEST(test_broadcast_evicts_overloaded_standby); RUN_TEST(test_broadcast_send_failure_continues); diff --git a/src/controller/testsuite/test_term_safety_adapter.c b/src/controller/testsuite/test_term_safety_adapter.c index fb7f4a26..9039df91 100644 --- a/src/controller/testsuite/test_term_safety_adapter.c +++ b/src/controller/testsuite/test_term_safety_adapter.c @@ -1278,6 +1278,83 @@ static void test_intake_logmismatch_rejected_no_false_convergence(void) printf("test_intake_logmismatch_rejected_no_false_convergence: PASS\n"); } +/* ================================================================== */ +/* Task 3.4 — live broadcast/catch-up receive cutover (closes bug-990) */ +/* ================================================================== */ + +/* The live cutover's observable: a standby applying a LIVE-broadcast-shaped + * AppendEntries (a single entry in the `entries` LIST carrying the leader's term + * + prev tuple — exactly what repl_broadcast → xce_wal_batch_req_raw produces and + * xce_decode yields) through the PRODUCTION receive entry point + * ha_apply_wal_catchup_entry persists the LEADER's term (hdr.term == leader_term), + * NOT 0 (the retired term-0 legacy direct-apply path) and NOT its local term. + * + * Before the cutover repl_broadcast sent entry->term == 0, so this entry took the + * term-0 legacy direct-apply branch and persisted hdr.term = local p->term — the + * bug-990 term-identity gap. With the cutover the term-carrying entry routes + * through raft_on_append_entries_request → on_persist_entry → leader term stamped. */ +static void test_live_broadcast_receive_persists_leader_term(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_ts_live_bcast_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + cluster_t* c = make_cluster(tmp); + TEST_ASSERT_NOT_NULL(c); + /* The standby is a follower at a LOW local term (3) so we can prove the + * persisted term is the LEADER's (7), not the local term. */ + make_follower(c, /*term*/3, /*n_peers*/3); + + /* Seed LSN 1 at the leader's term 7 so the broadcast of LSN 2 satisfies the + * library's logOk (prev=(1,7)). This mirrors the contiguous live stream. */ + uint8_t blob[64]; + size_t n = pack_entry(blob, WAL_JOB_SUBMIT, "a", 1); + TEST_ASSERT_EQUAL_INT(0, ha_cb_persist_entry_for_test(&c->ha, 1, 7, blob, n)); + c->ha.raft.log_len = 1; + TEST_ASSERT_EQUAL_UINT64(7, persist_entry_term(&c->persist, 1)); + + /* Build the LIVE receive shape: a single entry in the `entries` list, each + * sub-entry carrying its term + prev tuple (what wal_entry_to_wire now encodes + * per element and xce_decode reproduces). repl_broadcast at term 7 of LSN 2 + * with prev=(1,7). */ + uint8_t lpay[2] = { 'b', 'c' }; + wal_catchup_entry_t sub; + memset(&sub, 0, sizeof(sub)); + sub.lsn = 2; + sub.entry_type = WAL_JOB_SUBMIT; + sub.term = 7; /* the LEADER's term on the wire */ + sub.prev_log_index = 1; + sub.prev_log_term = 7; + sub.raw_payload = lpay; + sub.raw_payload_len = sizeof(lpay); + + wal_catchup_entry_t* sub_arr[1] = { &sub }; + wal_catchup_entry_t batch; + memset(&batch, 0, sizeof(batch)); + batch.n_entries = 1; + batch.entries = sub_arr; + /* Top-level container also carries the AE metadata (the encoder mirrors it). */ + batch.term = 7; + batch.prev_log_index = 1; + batch.prev_log_term = 7; + + int rc = ha_apply_wal_catchup_entry(&c->ha, &c->persist, &batch, + /*ep*/NULL, c); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* The load-bearing assertion (bug-990): LSN 2 persists the LEADER's term 7, + * NOT 0 and NOT the local term 3. Before the cutover this read back as the + * standby's local term via the term-0 direct-apply path. */ + TEST_ASSERT_EQUAL_UINT64(7, persist_entry_term(&c->persist, 2)); + /* The agreed prefix (LSN 1@term7) is untouched. */ + TEST_ASSERT_EQUAL_UINT64(7, persist_entry_term(&c->persist, 1)); + /* The library adopted the leader's term on the RPC (G5). */ + TEST_ASSERT_EQUAL_UINT64(7, c->ha.raft.term); + + teardown_cluster(c); + printf("test_live_broadcast_receive_persists_leader_term: PASS\n"); +} + /* ================================================================== */ /* Task 3.3 — drive raft_on_append_entries_reply + raft_advance_commit */ /* on the replication ACK path */ @@ -1800,6 +1877,7 @@ void test_term_safety_adapter(void) RUN_TEST(test_intake_higher_term_heartbeat_adopts_term); RUN_TEST(test_intake_divergent_tail_truncated_and_reconverged); RUN_TEST(test_intake_logmismatch_rejected_no_false_convergence); + RUN_TEST(test_live_broadcast_receive_persists_leader_term); RUN_TEST(test_repl_ack_currentterm_quorum_advances_commit); RUN_TEST(test_repl_ack_priorterm_quorum_holds_commit); RUN_TEST(test_repl_ack_staleterm_reply_dropped); diff --git a/src/include/repl.h b/src/include/repl.h index e1f085bb..937c32a9 100644 --- a/src/include/repl.h +++ b/src/include/repl.h @@ -67,9 +67,19 @@ void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_i * NULL registry counts zero. Consumed by ha_has_write_quorum (ha.c). */ int repl_active_standby_count(replicator_t *r); -/* Replication send (stub in P2-1; wired in P2-4) */ +/* Replication send (stub in P2-1; wired in P2-4). + * + * ha-term-safety task 3.4 (Req 1.3/1.4): the LIVE primary append replicator. + * Carries the leader's per-entry Raft term (`hdr->term`) and the AppendEntries + * prev tuple (`prev_log_index = hdr->lsn - 1`, `prev_log_term` supplied by the + * caller from its term index) on every broadcast, so the standby intake routes + * the entry through `raft_on_append_entries_request` and persists the LEADER's + * term (not a local substitute or 0). `prev_log_term` is the term recorded for + * `hdr->lsn - 1` (0 when `hdr->lsn == 1`, i.e. no predecessor). The term-0 + * legacy direct-apply receive path is thereby retired for live entries. */ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr, - const void *payload, uint32_t payload_len); + const void *payload, uint32_t payload_len, + uint64_t prev_log_term); /* AppendEntries send (ha-term-safety task 2.3, Req 1.3/3.1). * diff --git a/src/lib/request.c b/src/lib/request.c index 29ce10d2..34d9a09f 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -478,6 +478,14 @@ static void wal_entry_to_wire(struct capn_segment *cs, /* ha-term-safety task 3.2: carry the entry's term stamp so the standby * intake persists the LEADER's recorded term (R1.3/R1.4), not a substitute. */ w->term = e->term; + /* ha-term-safety task 3.4: carry the AppendEntries prev tuple + leaderCommit + * PER SUB-ENTRY so a multi-entry catch-up batch is routable through the + * library one entry at a time (each sub-entry's prev tuple differs; the + * top-level container can hold only one). For a single-entry broadcast this + * is redundant with the top-level metadata but harmless. */ + w->prevLogIndex = e->prev_log_index; + w->prevLogTerm = e->prev_log_term; + w->leaderCommit = e->leader_commit; if (e->raw_payload && e->raw_payload_len > 0) { capn_list8 pl = capn_new_list8(cs, (int)e->raw_payload_len); capn_setv8(pl, 0, e->raw_payload, (int)e->raw_payload_len);