wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

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

6
.kiro/specs/ha-term-safety/tasks.md
View file

@ -71,7 +71,7 @@
- _Requirements: 1.1, 1.2, 2.1_
- _Depends: 2.5_
- [ ] 3.2 Route heartbeat/replication intake through `raft_on_append_entries_request` (replaces term-blind intake + committed-boundary reconcile)
- [x] 3.2 Route heartbeat/replication intake through `raft_on_append_entries_request` (replaces term-blind intake + committed-boundary reconcile)
- Replace the term-handling in `ha_consume_heartbeat_committed` and the committed-boundary truncate in `ha_rejoin_reconcile` with routing every incoming heartbeat/replication (0- or 1-entry AppendEntries) through the library so `UpdateTerm` (G5), logOk (G2), and `(term,lsn)` conflict-truncate (G3) happen via the callbacks; keep the loud non-promotable retry fence for an unreconcilable member
- Observable completion: a higher-term heartbeat makes the node adopt the term and step down (term recorded); a divergent same-LSN tail is truncated and reconverged; an unreconcilable member stays non-promotable and keeps retrying (no false convergence)
- _Requirements: 2.3, 2.5, 2.6, 3.1, 3.2, 3.3, 3.4, 3.5, 3.6_
@ -173,3 +173,7 @@
- **Anti-divergence invariant now LIVE + tested:** `p->current_lsn == s->log_len + 1` (next-LSN co-tracking) asserted across consecutive appends. Seeded at boot by 2.5 restore (s->log_len = durable_floor = current_lsn-1); preserved because raft_log_append (+1 to log_len) and its on_persist_entry→persist_wal_async (+1 to current_lsn) move in lockstep.
- **DEFERRED to 3.2:** heartbeat / 0-entry AppendEntries intake must route through `raft_on_append_entries_request` (UpdateTerm/logOk/(term,lsn) truncate via callbacks), replacing `ha_consume_heartbeat_committed` term handling + `ha_rejoin_reconcile` committed-boundary truncate; WalCatchupEntry still needs a `leaderCommit` capnp field + standby intake (deferred from 2.3).
- **PROCESS NOTE:** implementer first emitted `STATUS: COMPLETE` (non-parseable); corrected to READY_FOR_REVIEW. (Continue an existing subagent with SendMessage, not a fresh Agent call — a fresh Agent has no context.)
- **3.2 done.** Heartbeat/replication intake routes through the library. NEW `ha_consume_append_entries` (ha.c) = library-route intake; `ha_consume_heartbeat_committed` (~5691) now sends a 0-entry AE through `raft_on_append_entries_request` for UpdateTerm (G5) BEFORE the monotonic committed-learn — old term-blind `if(hb->term<myterm)return` DELETED; reachable on the LIVE heartbeat path (cluster.c:326). `ha_apply_wal_catchup_entry` (~2034) routes term-carrying single-entry AE through the library (logOk G2 + (term,lsn) truncate G3 via on_truncate_from/on_persist_entry). `ha_rejoin_reconcile` (ha_convergence.c) RETIRED to no-op; dead `ha_rejoin_reconcile_fn`/`_ctx_t` removed (the watermark committed-boundary truncate was the bug-851-unsafe heuristic). Wire gap CLOSED: `WalCatchupEntry.leaderCommit @7` (controller.capnp, append-only, no collision w/ term@4/prevLogIndex@5/prevLogTerm@6) encoded (request.c/repl.c) + decoded into typemap `leader_commit` + passed to the library (library currently `(void)`s it — commit-advance is 3.3). 3 new intake tests + 3 rewritten rejoin tests (test_demotion_convergence.c — preserve safety asserts, pin the retired contract, NOT weakened). 156 Tests 0 Failures; reviewer APPROVED (verified live-reachability, replace-not-parallel, leaderCommit end-to-end, baseline-stash proved bug-924 cluster pre-existing).
- **CRITICAL for 3.3 (the live-path gap):** live steady-state replication still uses TERM-0 `repl_broadcast` (legacy direct-apply in ha_apply_wal_catchup_entry, gated `entry->term != 0`). So the term-carrying single-entry AE RECEIVE route is UNIT-TEST-ONLY in production until 3.3 wires the library SEND loop (`raft_on_append_entries_reply` + `raft_advance_commit`) AND RETIRES the term-0 `repl_broadcast`. The higher-term-heartbeat UpdateTerm IS live (heartbeats are 0-entry → always go through the library route). Live divergent-tail convergence is validated in 5.2/5.4.
- **LOW cleanup (5.5 / cleanup pass):** (1) ha_convergence.c ~292-297 doc-comment on `cluster_apply_ctrlhello_reply` still describes the old "clamped truncate + restart" — stale. (2) `ha.h:197 rejoin_reconciled` atomic is now a dead struct field (once-guard unused). Both harmless/cosmetic. Add to the growing 5.5 cleanup list alongside the `ha_cb_*_stub` rename + stale stub-header comment.

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);
}

16
src/controller/ha.h
View file

@ -1012,6 +1012,22 @@ void ha_observe_heartbeat_committed(cluster_t *c, uint64_t committed_lsn,
*/
void ha_consume_heartbeat_committed(cluster_t *c, const heartbeat_msg_t *hb);
/*
* ha_consume_append_entries (ha-term-safety task 3.2): route ONE incoming
* heartbeat (entry->n_entries==0) or replication (entry->n_entries>0, one entry)
* AppendEntries through the raft library (raft_on_append_entries_request) so
* UpdateTerm (G5), logOk (G2), and the (term,lsn) conflict-truncate (G3) happen
* via the wired callbacks — replacing the term-blind term handling in
* ha_consume_heartbeat_committed and the committed-boundary truncate in
* ha_rejoin_reconcile. `entry` carries the AppendEntries metadata (prev tuple +
* leader_commit) on its top-level fields. `leader_idx` is the sending leader's
* raft peer index. Returns 0 on every path (never crashes the receive thread);
* the AE result (term adoption / append / truncate / reject) is reflected in the
* library + WAL state. The ACK reply that carries success/match_index is task 3.3.
*/
int ha_consume_append_entries(cluster_t *c, wal_catchup_entry_t *entry,
int leader_idx);
/* ha_heartbeat_is_fresh: 1 if a primary message arrived within
* heartbeat_timeout_ms (leader-stickiness predicate). Exposed for PreVote
* denial-reason logging in cluster.c; pure read, no state change. */

125
src/controller/ha_convergence.c
View file

@ -219,54 +219,13 @@ int ha_schedule_demotion_convergence(cluster_t *c)
}
/*
* ha_rejoin_reconcile_ctx_t — heap context for the one-shot rejoin delaymgr task.
* Carries the clamped cutoff computed at trigger time (under the same lock that
* read the tip) plus the prior tip for the WARN range, so the deferred body does
* not re-read state that may have advanced between schedule and run.
* ha-term-safety task 3.2: the rejoin delaymgr task body (ha_rejoin_reconcile_fn)
* and its heap context (ha_rejoin_reconcile_ctx_t) are REMOVED. They drove the
* watermark-clamped truncate + self-restart that is now retired — divergent-suffix
* reconciliation is owned by the raft library's (term,lsn) AppendEntries path
* (raft_on_append_entries_request → on_truncate_from). See ha_rejoin_reconcile
* below (now a no-op CtrlHelloReply hook).
*/
typedef struct {
cluster_t *c;
uint64_t cutoff; /* truncate above this LSN (clamped, Req 2.2) */
uint64_t prior_tip; /* local tip at trigger time (WARN range, Req 2.1) */
} ha_rejoin_reconcile_ctx_t;
/*
* ha_rejoin_reconcile_fn — delaymgr task body (runs OFF the CtrlHelloReply intake
* thread, which may be iomgr; D5/D1). Truncate above the clamped cutoff (Req 2.1,
* 2.2), WARN with the discarded range, then ha_self_restart (Req 2.1). On
* truncation failure ERROR then restart anyway — boot recovery + a subsequent
* rejoin converge the untruncated case (crash-safe by design, mirrors demotion).
*/
int ha_rejoin_reconcile_fn(void *arg)
{
ha_rejoin_reconcile_ctx_t *ctx = (ha_rejoin_reconcile_ctx_t *)arg;
if (ctx == NULL || ctx->c == NULL) {
free(ctx);
return -EINVAL;
}
cluster_t *c = ctx->c;
uint64_t cutoff = ctx->cutoff;
uint64_t prior_tip = ctx->prior_tip;
int trc = persist_truncate_wal_above(&c->persist, cutoff);
if (trc < 0) {
ERROR("HA rejoin reconcile: truncate above cutoff=%" PRIu64
" failed: %d — restarting anyway (boot recovery + next rejoin will "
"converge)", cutoff, trc);
} else {
/* Req 2.1: log the discarded range — (cutoff, prior tip]. */
WARN("HA rejoin reconcile: local tip=%" PRIu64 " exceeded the cluster "
"committed boundary; truncated WAL above cutoff=%" PRIu64
" (clamped to the locally committed prefix), restarting once to "
"rejoin from converged state", prior_tip, cutoff);
}
/* Free the task ctx before restart (production exec never returns). */
free(ctx);
/* Req 2.1: rebuild in-memory state through boot recovery. */
return ha_self_restart(c);
}
/*
* ha_rejoin_reconcile — converge a rejoining node whose local WAL log runs ahead
@ -297,58 +256,26 @@ int ha_rejoin_reconcile(cluster_t *c, uint64_t primary_committed_lsn)
if (c == NULL)
return -EINVAL;
/* Read the local tip + committed prefix together under the persist lock so
* the clamp decision is consistent. */
pthread_mutex_lock(&c->persist.lock);
uint64_t current_lsn = c->persist.current_lsn;
pthread_mutex_unlock(&c->persist.lock);
uint64_t local_tip = current_lsn > 0 ? current_lsn - 1 : 0;
/* Req 2.3: tip does not exceed the boundary → join with no truncation/restart. */
if (local_tip <= primary_committed_lsn)
return 0;
/* Req 2.1 (defensive once-guard): reconcile at most once per process lifetime.
* CAS 0→1 so a concurrent/duplicate CtrlHelloReply cannot schedule twice. */
int expected = 0;
if (!atomic_compare_exchange_strong_explicit(
&c->ha.rejoin_reconciled, &expected, 1,
memory_order_acq_rel, memory_order_acquire))
return 0; /* already reconciled — no second restart */
/* Req 2.2 (clamp): never truncate at/below the locally known committed prefix.
* cutoff = max(primary_committed_lsn, local committed_lsn). */
uint64_t local_committed = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
uint64_t cutoff = primary_committed_lsn > local_committed
? primary_committed_lsn
: local_committed;
ha_rejoin_reconcile_ctx_t *ctx = calloc(1, sizeof(*ctx));
if (ctx == NULL) {
ERROR("HA rejoin reconcile: ctx alloc failed — divergent suffix will be "
"reconciled at the next rejoin");
atomic_store_explicit(&c->ha.rejoin_reconciled, 0,
memory_order_release); /* roll back: allow retry */
return -ENOMEM;
}
ctx->c = c;
ctx->cutoff = cutoff;
ctx->prior_tip = local_tip;
/* D5/D1: defer truncation + restart off the (possibly iomgr) intake thread.
* One-shot, minimal delay — mirrors ha_schedule_demotion_convergence. Under
* XCE_TESTING the fixture drives the queued task synchronously. */
int rc = delay_task_ms(&c->delaymgr, 1, 1,
(delay_task_entry_fn_t)ha_rejoin_reconcile_fn, ctx);
if (rc < 0) {
ERROR("HA rejoin reconcile: delay_task_ms failed: %d", rc);
free(ctx);
atomic_store_explicit(&c->ha.rejoin_reconciled, 0,
memory_order_release); /* roll back: allow retry */
return rc;
}
/* ha-term-safety task 3.2: the committed-boundary (watermark) truncate is
* RETIRED. Divergent-suffix reconciliation is now owned by the raft library:
* an incoming AppendEntries from the leader is routed through
* raft_on_append_entries_request (ha_consume_append_entries), and the (term,
* lsn) log-matching there drives on_truncate_from to delete exactly the
* divergent suffix (G3) — never a committed-prefix entry (R3.3), and never by
* an LSN watermark that is term-blind. A node whose tip runs ahead of the
* cluster keeps its entries until the leader's AE proves the conflict by term,
* then the library truncates precisely; until reconciled the existing
* promotion-time fence keeps the member non-promotable (R3.5) without
* self-restarting.
*
* This function is retained as the CtrlHelloReply hook so the learned
* committed boundary is still recorded by the caller (cluster_apply_ctrlhello_
* reply), but it no longer performs a watermark truncate + self-restart. The
* local_tip vs boundary comparison and the clamped-cutoff heuristic that drove
* the old delaymgr truncate/restart task are removed: they are replaced by the
* library's term-aware truncation, not duplicated here.
*/
(void)primary_committed_lsn;
return 0;
}

12
src/controller/repl.c
View file

@ -379,11 +379,13 @@ int repl_send_append_entries(replicator_t *r, int peer_idx,
memset(&entry, 0, sizeof(entry));
entry.prev_log_index = prev_log_index;
entry.prev_log_term = prev_log_term;
/* NOTE: leader_commit is accepted as a parameter (and captured by the test
* seam) but is NOT yet wire-encoded — WalCatchupEntry has no leaderCommit
* field. Adding that capnp field and the standby-side intake of leader_commit
* is deferred to task 3.2; no standby acts on it until then.
* A heartbeat is marked by n_entries==0 (zero entries; lsn left 0). */
/* ha-term-safety task 3.2: leader_commit is now wire-encoded on the
* top-level container (WalCatchupEntry.leaderCommit @7) so the standby intake
* passes it into raft_on_append_entries_request. entry is entries[0] — the
* AppendEntries metadata holder; the encoder reads the prev tuple +
* leader_commit from it for both the heartbeat (n_entries==0) and the
* single-entry (n_entries==1) AE. A heartbeat carries zero entries. */
entry.leader_commit = leader_commit;
if (n_entries == 1) {
entry.lsn = entry_lsn;
entry.entry_type = entry_type;

107
src/controller/testsuite/test_demotion_convergence.c
View file

@ -592,46 +592,44 @@ static void rj_make_standby_with_seams(void)
g_restart_hook_rc = 0; /* model a successful restart */
}
/* 2.1/2.2: local tip ABOVE the primary boundary AND local committed_lsn ABOVE the
* boundary → reconcile truncates at the CLAMP max(boundary, local committed) (the
* higher local value, never the lower reported boundary) and self-restarts once.
* RED: ha_rejoin_reconcile is a no-op stub → no truncation, no restart. */
static void test_rejoin_clamp_truncates_at_local_committed_and_restarts_once(void)
/* ha-term-safety task 3.2: the committed-boundary (watermark) truncate + self-
* restart is RETIRED. Divergent-suffix reconciliation is now owned by the raft
* library's (term,lsn) log-matching (raft_on_append_entries_request →
* on_truncate_from), driven by the leader's AppendEntries — NOT by an LSN
* watermark on a CtrlHelloReply. ha_rejoin_reconcile therefore no longer
* truncates the local WAL or self-restarts; it is a no-op that returns 0 while
* the learned committed boundary is recorded by the caller and the precise
* truncation happens later via the library AE path. This test pins the RETIRED
* contract: even with a local tip ABOVE the reported boundary, NO truncation and
* NO restart occur here. */
static void test_rejoin_no_watermark_truncate_no_restart(void)
{
dc_setup("rjclamp", /*n_peers=*/3, /*n_standbys=*/0,
/*appended_lsn=*/9, /*n_vcpus=*/8);
rj_make_standby_with_seams();
/* Local tip = 9 (entries 1..9 on disk). Local committed_lsn = 6 — ABOVE the
* primary-reported boundary of 4. The clamp must protect [1..6] and truncate
* only above 6 → cutoff = max(4, 6) = 6 → current_lsn collapses to 7. */
/* Local tip = 9 (entries 1..9 on disk), local committed_lsn = 6, primary
* boundary = 4. Pre-3.2 this would have truncated above 6 + restarted; post-
* 3.2 it must leave the WAL and process untouched (the library reconciles by
* term when the leader's AE arrives). */
rj_set_local_state(/*tip_lsn=*/9, /*committed=*/6);
const uint64_t primary_boundary = 4;
const uint64_t local_committed = 6;
uint64_t tip_before = g_dc.persist.current_lsn;
/* Act. */
int rc = ha_rejoin_reconcile(&g_dc, primary_boundary);
/* Drive any deferred convergence task (the intake may defer to delaymgr,
* same rule as demotion); harmless if 3.2 truncates inline. */
int rc = ha_rejoin_reconcile(&g_dc, /*primary_committed_lsn=*/4);
(void)dc_run_pending_task(&g_dc.delaymgr);
/* Assert (Req 2.2 clamp): truncation cutoff is the LOCAL committed (6), not the
* lower reported boundary (4) — current_lsn collapses to local_committed + 1. */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(local_committed + 1, g_dc.persist.current_lsn,
"rejoin reconcile must clamp the cutoff to max(boundary, local committed) "
"(Req 2.2): with boundary=4 and local committed=6 it must truncate above 6, "
"NOT above the lower reported boundary 4");
/* Assert (Req 2.1): self-restart fired exactly once, fatal seam never. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls),
"rejoin reconcile with tip > boundary must self-restart exactly once "
"(Req 2.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"ha_rejoin_reconcile is retired to a no-op (task 3.2) and returns 0");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(tip_before, g_dc.persist.current_lsn,
"the retired watermark truncate must NOT shrink the local WAL — the "
"library's (term,lsn) AE reconcile owns truncation now (task 3.2)");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_restart_calls),
"the retired rejoin reconcile must NOT self-restart (task 3.2)");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_fatal_calls),
"a successful rejoin restart must NOT invoke the fatal seam");
(void)rc;
"the retired rejoin reconcile must NOT invoke the fatal seam");
dc_teardown();
printf("test_rejoin_clamp_truncates_at_local_committed_and_restarts_once: checked\n");
printf("test_rejoin_no_watermark_truncate_no_restart: checked\n");
}
/* 2.3: local tip AT/BELOW the boundary → no truncation, no restart, returns 0.
@ -664,38 +662,32 @@ static void test_rejoin_tip_at_or_below_boundary_no_action(void)
printf("test_rejoin_tip_at_or_below_boundary_no_action: checked\n");
}
/* 2.1 once-guard: a SECOND reconcile call in the same process lifetime must NOT
* self-restart again, even if the tip still exceeds the boundary (defensive guard;
* "once" is also structural post-restart).
* RED: the no-op stub never restarts even once, so the "exactly one restart total"
* assertion fails. */
static void test_rejoin_second_call_no_second_restart(void)
/* ha-term-safety task 3.2: with the watermark truncate retired, REPEATED rejoin
* reconcile calls remain side-effect-free — never truncating and never
* restarting, regardless of how far the local tip runs ahead. (Pre-3.2 this
* pinned a defensive once-guard around a single restart; post-3.2 there is no
* restart at all, so any number of calls is a no-op.) */
static void test_rejoin_repeated_calls_remain_no_op(void)
{
dc_setup("rjonce", /*n_peers=*/3, /*n_standbys=*/0,
/*appended_lsn=*/9, /*n_vcpus=*/8);
rj_make_standby_with_seams();
rj_set_local_state(/*tip_lsn=*/9, /*committed=*/4);
uint64_t tip_before = g_dc.persist.current_lsn;
/* Act: first reconcile (tip 9 > boundary 4 → truncate + restart once). */
(void)ha_rejoin_reconcile(&g_dc, /*primary_committed_lsn=*/4);
(void)dc_run_pending_task(&g_dc.delaymgr);
int restarts_after_first = atomic_load(&g_restart_calls);
/* Act: second reconcile in the SAME lifetime. The once-guard must suppress a
* second restart regardless of the reported boundary. */
(void)ha_rejoin_reconcile(&g_dc, /*primary_committed_lsn=*/4);
(void)dc_run_pending_task(&g_dc.delaymgr);
/* Assert (Req 2.1): exactly ONE restart across both calls. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, restarts_after_first,
"the FIRST rejoin reconcile with tip > boundary must self-restart once "
"(Req 2.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls),
"a SECOND rejoin reconcile in the same process lifetime must NOT restart "
"again — defensive once-guard (Req 2.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_restart_calls),
"the retired rejoin reconcile must NEVER self-restart (task 3.2)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(tip_before, g_dc.persist.current_lsn,
"repeated retired rejoin reconcile must leave the local WAL untouched "
"(task 3.2)");
dc_teardown();
printf("test_rejoin_second_call_no_second_restart: checked\n");
printf("test_rejoin_repeated_calls_remain_no_op: checked\n");
}
/* Intake integration: delivering a CtrlHelloReply through the REAL cluster.c
@ -727,12 +719,17 @@ static void test_ctrlhello_reply_intake_stores_and_reconciles(void)
atomic_load(&g_dc.ha.last_known_committed_config_tag),
"CtrlHelloReply intake must store the config tag (existing)");
/* Assert (intake integration): the intake also TRIGGERED reconciliation —
* tip 9 > boundary 4 → exactly one self-restart. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_restart_calls),
"CtrlHelloReply intake must trigger rejoin reconciliation when the local "
"tip exceeds the reported boundary — exactly one self-restart (task 3.2 "
"wires ha_rejoin_reconcile into the intake)");
/* Assert (task 3.2): the intake STILL stores the boundary (above) but the
* watermark truncate/restart is RETIRED — the intake must NOT self-restart
* even though the local tip (9) exceeds the reported boundary (4). The
* library's (term,lsn) AE reconcile owns the divergent-suffix truncation now;
* the member stays alive and converges when the leader's AppendEntries arrive. */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_restart_calls),
"CtrlHelloReply intake must NOT self-restart on a tip-ahead boundary — the "
"retired watermark reconcile no longer truncates/restarts (task 3.2)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(10, g_dc.persist.current_lsn,
"the intake must leave the local WAL untouched (tip 9 → current_lsn 10); "
"reconciliation is deferred to the library AE path (task 3.2)");
dc_teardown();
printf("test_ctrlhello_reply_intake_stores_and_reconciles: checked\n");
@ -757,9 +754,9 @@ void test_demotion_convergence(void)
printf("\n>> rejoin reconciliation tests (RED, task 3.1)\n\n");
RUN_TEST(test_rejoin_clamp_truncates_at_local_committed_and_restarts_once);
RUN_TEST(test_rejoin_no_watermark_truncate_no_restart);
RUN_TEST(test_rejoin_tip_at_or_below_boundary_no_action);
RUN_TEST(test_rejoin_second_call_no_second_restart);
RUN_TEST(test_rejoin_repeated_calls_remain_no_op);
RUN_TEST(test_ctrlhello_reply_intake_stores_and_reconciles);
#else
printf("test_demotion_convergence: SKIP (non-test build)\n");

177
src/controller/testsuite/test_term_safety_adapter.c
View file

@ -1102,6 +1102,180 @@ static void test_primary_log_append_non_leader_legacy_fallback(void)
printf("test_primary_log_append_non_leader_legacy_fallback: PASS\n");
}
/* ================================================================== */
/* Task 3.2 — route heartbeat/replication intake through the library */
/* raft_on_append_entries_request: UpdateTerm (G5) + logOk (G2) + */
/* (term,lsn) conflict-truncate (G3) via the wired callbacks. */
/* */
/* These drive the production intake helper ha_consume_append_entries */
/* (the in-process route that replaces the term-blind handling in */
/* ha_consume_heartbeat_committed + the committed-boundary truncate in */
/* ha_rejoin_reconcile). RED: that helper does not exist pre-3.2 */
/* (compile RED); once it exists the three observable behaviors hold. */
/* ================================================================== */
/* Make a node a STANDBY follower at a given term + raft log floor, with the
* raft library initialized so the intake can drive it. Seeds the library log
* to match the persisted WAL tip (the ha_init_raft floor restore). */
static void make_follower(cluster_t* c, uint64_t term, int n_peers)
{
atomic_store(&c->ha.role, CTRL_STANDBY);
atomic_store(&c->ha.term, term);
ha_init_raft(&c->ha, c, n_peers);
c->ha.raft.term = term;
c->ha.raft.role = RAFT_FOLLOWER;
}
/* Build a decoded single-entry WalCatchupEntry (n_entries==0 path == one entry
* carried in the top-level container, matching the on-wire single-entry mode).
* For a heartbeat use n_entries set so the helper sees zero entries. */
static void fill_ae_entry(wal_catchup_entry_t* e, int is_heartbeat,
uint64_t prev_log_index, uint64_t prev_log_term,
uint64_t leader_commit, uint64_t entry_lsn,
uint8_t entry_type, uint64_t entry_term,
uint8_t* payload, uint32_t plen)
{
memset(e, 0, sizeof(*e));
e->prev_log_index = prev_log_index;
e->prev_log_term = prev_log_term;
e->leader_commit = leader_commit;
if (is_heartbeat) {
e->n_entries = 0;
e->lsn = 0;
} else {
/* single-entry mode: top-level container carries the entry, n_entries==1
* with no nested sub-entries (one-entry AppendEntries). */
e->n_entries = 1;
e->lsn = entry_lsn;
e->entry_type = entry_type;
e->term = entry_term;
e->raw_payload = payload;
e->raw_payload_len = plen;
}
}
/* (a) A higher-term heartbeat (0-entry AppendEntries) routed through the library
* makes the follower ADOPT the higher term and step down — term recorded
* (R2.6 / G5). Pre-3.2 ha_consume_heartbeat_committed only DROPPED stale
* heartbeats and never adopted a higher term. */
static void test_intake_higher_term_heartbeat_adopts_term(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_ts_intake_ht_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t* c = make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
make_follower(c, /*term*/3, /*n_peers*/3);
/* The follower briefly believes itself a candidate/leader at term 3. */
c->ha.raft.role = RAFT_LEADER;
/* Heartbeat from a leader at term 7, empty log (prev=0), commit 0. The
* heartbeat's top-level container carries the leader's live term. */
wal_catchup_entry_t hb;
fill_ae_entry(&hb, /*heartbeat*/1, /*prev_idx*/0, /*prev_term*/0,
/*leader_commit*/0, 0, 0, 0, NULL, 0);
hb.term = 7;
int rc = ha_consume_append_entries(c, &hb, /*leader_idx*/1);
TEST_ASSERT_EQUAL_INT(0, rc);
/* The library adopted the higher term and stepped down to follower. */
TEST_ASSERT_EQUAL_UINT64(7, c->ha.raft.term);
TEST_ASSERT_NOT_EQUAL_INT(RAFT_LEADER, c->ha.raft.role);
teardown_cluster(c);
printf("test_intake_higher_term_heartbeat_adopts_term: PASS\n");
}
/* (b) A divergent same-LSN tail is truncated and the leader's entry accepted
* (R3.1/R3.2 / G3), and the committed prefix below the divergence is untouched
* (R3.3). The follower locally holds [LSN1@term2, LSN2@term2]; the leader sends
* an AppendEntries appending LSN2@term5 on top of LSN1@term2. The library
* detects the (term,lsn) conflict at LSN2, truncates LSN>=2, and re-persists
* the leader's LSN2@term5. LSN1 (the agreed prefix) stays intact. */
static void test_intake_divergent_tail_truncated_and_reconverged(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_ts_intake_div_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t* c = make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
make_follower(c, /*term*/2, /*n_peers*/3);
/* Seed the local log: LSN1@term2, LSN2@term2 (the divergent tail). */
uint8_t blob[64];
size_t n = pack_entry(blob, WAL_JOB_UPDATE, "a", 1);
TEST_ASSERT_EQUAL_INT(0, ha_cb_persist_entry_for_test(&c->ha, 1, 2, blob, n));
TEST_ASSERT_EQUAL_INT(0, ha_cb_persist_entry_for_test(&c->ha, 2, 2, blob, n));
c->ha.raft.log_len = 2;
TEST_ASSERT_EQUAL_UINT64(2, persist_entry_term(&c->persist, 1));
TEST_ASSERT_EQUAL_UINT64(2, persist_entry_term(&c->persist, 2));
/* Leader (term 5) sends AppendEntries: prev=(LSN1,term2), entry LSN2@term5.
* logOk holds at LSN1 (term2==term2); conflict at LSN2 (local term2 !=
* incoming term5) → truncate LSN>=2 then persist LSN2@term5. */
uint8_t lpay[1] = { 'b' };
wal_catchup_entry_t ae;
fill_ae_entry(&ae, /*heartbeat*/0, /*prev_idx*/1, /*prev_term*/2,
/*leader_commit*/1, /*entry_lsn*/2, WAL_JOB_UPDATE,
/*entry_term*/5, lpay, 1);
int rc = ha_consume_append_entries(c, &ae, /*leader_idx*/1);
TEST_ASSERT_EQUAL_INT(0, rc);
/* (term,lsn) reconverged: LSN2 now carries the leader's term 5. */
TEST_ASSERT_EQUAL_UINT64(5, persist_entry_term(&c->persist, 2));
/* The agreed committed prefix (LSN1@term2) is untouched (R3.3). */
TEST_ASSERT_EQUAL_UINT64(2, persist_entry_term(&c->persist, 1));
/* The higher leader term was adopted on the RPC (G5). */
TEST_ASSERT_EQUAL_UINT64(5, c->ha.raft.term);
teardown_cluster(c);
printf("test_intake_divergent_tail_truncated_and_reconverged: PASS\n");
}
/* (c) An unreconcilable member (logOk fails: the leader's prev tuple does not
* match the follower's log) is REJECTED — the follower neither persists the
* leader's entry nor reports false convergence, and stays available to retry
* (no crash, no committed advance). R3.5/R3.6: the member keeps its divergent
* state pending a later catch-up rather than spuriously accepting. */
static void test_intake_logmismatch_rejected_no_false_convergence(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_ts_intake_rej_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t* c = make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
make_follower(c, /*term*/4, /*n_peers*/3);
/* Local log: LSN1@term2. The leader claims prev=(LSN1,term3) — a term
* mismatch at LSN1 → logOk fails. */
uint8_t blob[64];
size_t n = pack_entry(blob, WAL_JOB_UPDATE, "a", 1);
TEST_ASSERT_EQUAL_INT(0, ha_cb_persist_entry_for_test(&c->ha, 1, 2, blob, n));
c->ha.raft.log_len = 1;
uint8_t lpay[1] = { 'z' };
wal_catchup_entry_t ae;
fill_ae_entry(&ae, /*heartbeat*/0, /*prev_idx*/1, /*prev_term*/3,
/*leader_commit*/0, /*entry_lsn*/2, WAL_JOB_UPDATE,
/*entry_term*/4, lpay, 1);
int rc = ha_consume_append_entries(c, &ae, /*leader_idx*/1);
/* The helper returns 0 (no crash) but the AE was NOT accepted: the leader's
* entry was not persisted at LSN2 and the local LSN1 entry is unchanged. */
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT64(0, persist_entry_term(&c->persist, 2)); /* absent */
TEST_ASSERT_EQUAL_UINT64(2, persist_entry_term(&c->persist, 1)); /* intact */
teardown_cluster(c);
printf("test_intake_logmismatch_rejected_no_false_convergence: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Suite entry point */
/* ------------------------------------------------------------------ */
@ -1131,4 +1305,7 @@ void test_term_safety_adapter(void)
RUN_TEST(test_primary_log_append_stamps_term_and_assigns_lsn);
RUN_TEST(test_own_term_noop_records_live_term_via_library);
RUN_TEST(test_primary_log_append_non_leader_legacy_fallback);
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);
}

4
src/include/typemap.h
View file

@ -360,6 +360,10 @@ struct wal_catchup_entry {
uint64_t term;
uint64_t prev_log_index;
uint64_t prev_log_term;
/* ha-term-safety task 3.2 (Req 2.5/4.1): the leader's commit index carried on
* the AppendEntries so the standby intake passes it into
* raft_on_append_entries_request. Wire field leaderCommit @7. */
uint64_t leader_commit;
};
/* ── Persistence Cap'n Proto mapped types (P0-9) ─────────────────────

31
src/lib/request.c
View file

@ -202,7 +202,15 @@ int xce_decode(io_packet_t* pkt, xce_request_t* req)
extract_data_bytes(WalCatchupEntry_get_payload(b_ptr),
&batch->raw_payload_len);
/* Batch sub-entries */
/* ha-term-safety task 3.2: the top-level container's AppendEntries
* metadata (term, prevLogIndex, prevLogTerm, leaderCommit) is decoded
* into batch->{term,prev_log_index,prev_log_term,leader_commit} by the
* generated decode_WalCatchupEntry (all scalar $C.mapname fields) — only
* the Data payload needs manual extraction (above/below). The standby
* intake (ha_consume_append_entries) reads those struct fields directly. */
/* Batch sub-entries: each carries its own lsn/entryType/term (scalar,
* decoded) + payload (Data, extracted here). */
WalCatchupEntry_list wlist = WalCatchupEntry_get_entries(b_ptr);
int wn = (wlist.p.type != CAPN_NULL) ? capn_len(wlist) : 0;
for (int i = 0; i < wn && i < batch->n_entries; i++) {
@ -467,6 +475,9 @@ static void wal_entry_to_wire(struct capn_segment *cs,
memset(w, 0, sizeof(*w));
w->lsn = e->lsn;
w->entryType = e->entry_type;
/* 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;
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);
@ -499,7 +510,10 @@ int xce_wal_batch_encode_raw(wal_catchup_entry_t **entries, int n_entries,
Request_ptr rp;
int rc;
if (pkt == NULL || n_entries <= 0 || entries == NULL)
/* n_entries == 0 is a valid AppendEntries HEARTBEAT (ha-term-safety task 3.2):
* entries[0] still carries the AppendEntries metadata (prev tuple +
* leader_commit) on the top-level container, with an empty entries list. */
if (pkt == NULL || n_entries < 0 || entries == NULL)
return -EINVAL;
capn_init_malloc(&c);
@ -508,7 +522,18 @@ int xce_wal_batch_encode_raw(wal_catchup_entry_t **entries, int n_entries,
memset(&rq, 0, sizeof(rq));
memset(&top, 0, sizeof(top));
/* Batch container: header fields zero, entries list carries the batch */
/* ha-term-safety task 3.2: the top-level container carries the AppendEntries
* metadata (prev tuple + leaderCommit) so a 0-entry heartbeat still conveys
* UpdateTerm/logOk inputs. entries[0] is the metadata holder for both the
* heartbeat (n_entries==0) and the single-entry (n_entries==1) AE built by
* repl_send_append_entries. Legacy catch-up batches leave these zero. */
top.term = entries[0]->term;
top.prevLogIndex = entries[0]->prev_log_index;
top.prevLogTerm = entries[0]->prev_log_term;
top.leaderCommit = entries[0]->leader_commit;
/* Batch container: entries list carries the per-entry payload(s) (empty for
* a heartbeat). Each sub-entry carries lsn/entryType/term/payload. */
WalCatchupEntry_list lst = new_WalCatchupEntry_list(cs, n_entries);
for (int i = 0; i < n_entries; i++) {
struct WalCatchupEntry w;

4
src/schema/controller.capnp
View file

@ -894,6 +894,10 @@ struct WalCatchupEntry $C.mapname("wal_catchup_entry_t") {
term @4 :UInt64 $C.mapname("term");
prevLogIndex @5 :UInt64 $C.mapname("prev_log_index");
prevLogTerm @6 :UInt64 $C.mapname("prev_log_term");
# ha-term-safety task 3.2 (Req 2.5/4.1): the leader's commit index, carried so
# the standby intake passes it into raft_on_append_entries_request. Append-only
# field number (@7) — no collision with the @4/@5/@6 term fields above.
leaderCommit @7 :UInt64 $C.mapname("leader_commit");
}
struct CtrlHelloRequest $C.mapname("ctrl_hello_req_t") {