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

feat(ha-term-safety): route AppendEntries intake through library + leaderCommit wire (task 3.2)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-22 08:42:35 +00:00
parent 6a519a928b
commit 754e586f8f
10 changed files with 508 additions and 168 deletions
Download patch file Download diff file Expand all files Collapse all files

194
src/controller/ha.c
View file

@ -2021,6 +2021,40 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
if (!ha || !p || !entry)
return -EINVAL;
/* ── ha-term-safety task 3.2: term-carrying AppendEntries → library route ──
* A term-safe AppendEntries (one sent by repl_send_append_entries / the
* library send path) carries a non-zero entry term stamp. Route it through
* raft_on_append_entries_request so the LIBRARY performs UpdateTerm (G5),
* logOk (G2), and the (term,lsn) conflict-truncate (G3) — its on_persist_entry
* callback does the WAL write (persist-before-ACK + bug-850 guard beneath).
* This REPLACES the direct persist_wal_write_replicated apply for term-safe
* entries; the legacy term-blind broadcast (entry term 0, no prev tuple) keeps
* the direct apply below until the live send path is retired (task 3.3).
*
* Single-entry term-safe AE only (n_entries==1 with a term). The library
* 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);
}
}
return 0;
}
/* ── Batch mode: entry->n_entries > 0 ─────────────────────────────────── */
if (entry->n_entries > 0) {
/* Apply each sub-entry in LSN order. Track the last successfully
@ -2088,6 +2122,129 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
return 0;
}
/* ── ha-term-safety task 3.2: library-routed AppendEntries intake ───────────
*
* ha_consume_append_entries: route ONE incoming heartbeat (0-entry) or
* replication (1-entry) AppendEntries through the raft library
* (raft_on_append_entries_request) so the LIBRARY — not hand-rolled term-blind
* code — performs:
* - UpdateTerm first on EVERY RPC incl. heartbeat (G5, R2.6): a higher-term
* AE makes the node adopt the term + step down (via on_become_follower).
* - logOk (G2, R3.1): (prevLogIndex,prevLogTerm) match against the local log
* via on_get_entry_term — a mismatch is REJECTED (no false convergence).
* - (term,lsn) conflict-truncate (G3, R3.2/R3.3): a divergent same-LSN tail is
* truncated (on_truncate_from) then the leader's entry re-persisted
* (on_persist_entry), never touching the agreed prefix below the divergence.
*
* This REPLACES the term-blind term handling in ha_consume_heartbeat_committed
* and the committed-boundary truncate in ha_rejoin_reconcile: the library's
* (term,lsn) decision, not a watermark, drives truncation.
*
* `entry` is a decoded wal_catchup_entry_t whose TOP-LEVEL fields carry the AE
* metadata (prev_log_index, prev_log_term, leader_commit) and, for a single
* entry, the entry itself (lsn, entry_type, term, raw_payload). `leader_idx` is
* the raft peer index of the sending leader.
*
* The library callbacks own all persistence/truncation; on a REJECT (stale term
* or logOk fail) the library mutates nothing and the member keeps its current
* state to retry on a later catch-up (R3.5/R3.6 — never report false
* convergence). The loud non-promotable fence for an unreconcilable member is
* the EXISTING ha-commit-durability promotion-time fence; this intake never
* advertises convergence it did not achieve.
*
* Single-receiver-thread model (controller request handling), matching the
* raft_on_vote_request intake: no ha->lock is taken (the raft state is serial
* per the receive thread). Returns 0 (no crash) on every path; the AE result is
* reflected in the library/WAL state, and a future ACK reply (task 3.3) carries
* the success/match_index to the leader.
*/
int ha_consume_append_entries(cluster_t *c, wal_catchup_entry_t *entry,
int leader_idx)
{
if (!c || !entry)
return -EINVAL;
int n_entries = (entry->n_entries > 0) ? 1 : 0; /* 0 = heartbeat, 1 = entry */
/* AppendEntries metadata from the top-level container. */
uint64_t prev_log_index = entry->prev_log_index;
uint64_t prev_log_term = entry->prev_log_term;
uint64_t leader_commit = entry->leader_commit;
/* The incoming AE's term: for a single entry, the entry's term stamp; for a
* heartbeat, the top-level container's term (set by the sender to its live
* term). raft_on_append_entries_request uses this for UpdateTerm (G5). */
uint64_t ae_term;
const void *blob = NULL;
size_t blob_len = 0;
uint64_t entry_term = 0;
/* For a single entry, repack the [raft_entry_blob_t{type,len}][payload]
* envelope that on_persist_entry unpacks (the shared wal_types.h format), so
* the library-driven persist writes the same bytes the leader sent. */
uint8_t stackbuf[512];
uint8_t *heapbuf = NULL;
if (n_entries == 1) {
/* Single-entry mode: the entry lives in entries[0] (batch) or on the
* top-level container (single). Prefer the sub-entry when present. */
wal_catchup_entry_t *e =
(entry->entries && entry->entries[0]) ? entry->entries[0] : entry;
uint32_t type = e->entry_type;
const void *payload = e->raw_payload;
uint32_t plen = (uint32_t)e->raw_payload_len;
entry_term = e->term;
/* ae_term for UpdateTerm is the entry's term (the leader appends at its
* current term, so entry_term == leader term for a live append). */
ae_term = entry_term;
blob_len = sizeof(raft_entry_blob_t) + plen;
uint8_t *buf = (blob_len <= sizeof(stackbuf)) ? stackbuf
: (heapbuf = malloc(blob_len));
if (!buf)
return -ENOMEM;
raft_entry_blob_t *bh = (raft_entry_blob_t *)buf;
bh->type = type;
bh->payload_len = plen;
if (payload && plen > 0)
memcpy(buf + sizeof(*bh), payload, plen);
blob = buf;
} else {
/* Heartbeat: the top-level container's term carries the leader's live
* term (the sender stamps it). No entry to persist. */
ae_term = entry->term;
}
raft_append_result_t res;
memset(&res, 0, sizeof(res));
int rc = raft_on_append_entries_request(&c->ha.raft, leader_idx, ae_term,
prev_log_index, prev_log_term,
leader_commit, blob, blob_len,
entry_term, n_entries, &res);
if (heapbuf)
free(heapbuf);
if (rc != 0) {
/* Library-internal error (not a protocol reject — those return RAFT_OK
* with res.success==0). Log loud; the member stays available to retry. */
WARN("ha_consume_append_entries: raft_on_append_entries_request rc=%d "
"(leader_idx=%d term=%" PRIu64 " prevIdx=%" PRIu64 ")",
rc, leader_idx, ae_term, prev_log_index);
return 0; /* never crash the receive thread */
}
TRACE("ha_consume_append_entries: leader_idx=%d term=%" PRIu64
" prevIdx=%" PRIu64 " n=%d -> success=%d match=%" PRIu64,
leader_idx, ae_term, prev_log_index, n_entries,
res.success, res.match_index);
/* res.success / res.match_index are consumed by the ACK reply path (task
* 3.3). On a reject (res.success==0) the library left the log untouched —
* R3.5/R3.6: no false convergence, the member reconciles on a later AE. */
return 0;
}
/* ha_handle_snapshot_chunk — standby snapshot receive handler (Task 3.5c).
*
* Each call appends payload_bytes to snap_dir/snap.part.
@ -5534,15 +5691,42 @@ void ha_consume_heartbeat_committed(cluster_t *c, const heartbeat_msg_t *hb)
if (role == CTRL_PRIMARY)
return;
/* (3.2) Term-guard: a heartbeat from a primary whose term is behind ours is
* stale/partitioned — reject its committed so it cannot inflate the
* vote-freshness floor. Equal-or-newer term passes. */
/* ha-term-safety task 3.2 (R2.6 / G5): route the heartbeat through the raft
* library as a 0-entry AppendEntries so the LIBRARY performs UpdateTerm — a
* heartbeat at a strictly-higher term makes this node adopt the term and step
* down. This REPLACES the prior hand-rolled term-guard that only DROPPED a
* stale heartbeat (hb->term < myterm) and never adopted a higher term.
*
* The heartbeat carries no entries and no prev tuple, so we present the
* node's own log tip as (prevLogIndex,prevLogTerm) — logOk holds (already-
* have at the tip) for an equal/higher-term heartbeat, and UpdateTerm runs
* first regardless. raft_init must have wired the raft state (production
* boot); guard for the pre-init placeholder where raft_c is unset. */
if (c->ha.raft_c == c) {
uint64_t my_log_len = c->ha.raft.log_len;
uint64_t my_tip_term = (my_log_len > 0)
? persist_entry_term(&c->persist, my_log_len) : 0;
wal_catchup_entry_t hbae;
memset(&hbae, 0, sizeof(hbae));
hbae.n_entries = 0; /* heartbeat: zero entries */
hbae.term = hb->term; /* leader's live term (UpdateTerm) */
hbae.prev_log_index = my_log_len; /* already-have at our tip */
hbae.prev_log_term = my_tip_term;
hbae.leader_commit = hb->committed_lsn;
(void)ha_consume_append_entries(c, &hbae, /*leader_idx*/-1);
}
/* (3.2) Stale-term guard for the committed LEARN below: after UpdateTerm a
* heartbeat from a primary whose term is behind ours is stale/partitioned —
* do not let its committed inflate the vote-freshness floor. Re-read our term
* (it may have advanced via the library UpdateTerm above). */
uint64_t myterm = atomic_load_explicit(&c->ha.term, memory_order_acquire);
if (hb->term < myterm)
return;
/* (1.2) Monotonic, reconcile-free learn. Zero/non-advancing LSN no-ops
* inside the observe (Req 4.2). */
/* (1.2) Monotonic, reconcile-free learn of the committed position. Zero/
* non-advancing LSN no-ops inside the observe (Req 4.2). */
ha_observe_heartbeat_committed(c, hb->committed_lsn,
hb->committed_config_tag);
}