fix(ha): bug-1421 close term-desync AE-reject blindspot in HA election

A primary never learns of a follower's higher term via AppendEntries-reject
responses: ha_raft_on_repl_ack only routes AE replies through
raft_on_append_entries_reply for success=1 ACKs, and that function itself
drops any reply whose term != s->term (raft.c stale-term drop) before it
could adopt a higher term anyway. Net effect: once a follower outpaces a
stale primary's term, every AppendEntries the primary sends is rejected
forever and it can never regain write-quorum — a permanent election livelock
(surfaced as "no primary elected at term>8" in HA soak testing).

Adds two fallback term-discovery vectors that fire regardless of AE-reject
outcome:
- StateReport heartbeat-reply now carries the standby's current term
  (controller.capnp, typemap.h) so the primary learns a higher term on every
  convergence cycle.
- New raft_on_higher_term_observed() mirrors the UpdateTerm step (strictly
  greater-than only; touches term/voted_for/role; persists before notifying
  on_become_follower) and is wired into both the heartbeat state-report path
  and the WAL-NACK path in repl_ack_decode_and_dispatch.

Validated: 265 controller + 287 lib unit tests pass; 10-round sequential and
chaotic HA soaks both PASS with zero job loss and full leadership coverage.

src/controller/ha.c

@@ -2818,6 +2818,7 @@ void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf,
     sr->applied_lsn = atomic_load_explicit(&ha->applied_lsn,
                                            memory_order_acquire);
     sr->config_tag  = myconf ? myconf->config_version_tag : 0;
+    sr->term        = atomic_load_explicit(&ha->term, memory_order_acquire);
     reply_out->data_which        = RequestReply_data_stateReportMsg;
     reply_out->data.state_report = sr;
 }
@@ -5487,8 +5488,21 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx,
             state_report_t sr;
             sr.applied_lsn = (uint64_t)sr_raw.appliedLsn;
             sr.config_tag  = (uint64_t)sr_raw.configTag;
+            sr.term        = (uint64_t)sr_raw.term;
             capn_free(&capn_ctx);
 
+            /* bug-1421: a primary never learns of a follower's higher term via
+             * the AE-response path (an AE reject is dropped before it reaches
+             * raft_on_append_entries_reply — see ha_raft_on_repl_ack). The
+             * heartbeat-reply state report is the fallback term-discovery
+             * vector: it fires on every convergence cycle regardless of
+             * replication outcome, so a stale-term primary steps down here
+             * instead of retrying AppendEntries against a follower that will
+             * reject every entry forever. raft_on_higher_term_observed only
+             * touches (term, voted_for, role) — never peer-indexed
+             * bookkeeping — so it is safe to call before peer_idx is resolved. */
+            (void)raft_on_higher_term_observed(&ha->raft, sr.term);
+
             /* peer_idx from standby_ep_ctx_t (layout-compatible with vote_ep_ctx_t) */
             int      peer_idx      = vctx->peer_idx;
             uint64_t committed_lsn = 0;
@@ -5580,6 +5594,11 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx,
                  * deferred continuation task. */
                 ha_walack_on_ack(c, ep, vctx->peer_idx, acked_lsn);
             } else if (!success) {
+                /* bug-1421: a WAL NACK can also carry a higher term (the
+                 * standby rejected this entry as stale) — route it through
+                 * the same term-discovery path as the heartbeat state report
+                 * so the primary steps down instead of retrying forever. */
+                (void)raft_on_higher_term_observed(&ha->raft, reply_term);
                 WARN("repl_ack_decode_and_dispatch: WAL NACK from standby"
                      " lsn=%lu success=0 — ignoring", (unsigned long)acked_lsn);
             }

src/include/typemap.h

@@ -193,10 +193,14 @@ typedef struct {
     uint64_t peer_last_log_lsn;
 } request_vote_reply_t;
 
-/* Heartbeat reply state report — carries standby's applied_lsn and config_tag */
+/* Heartbeat reply state report — carries standby's applied_lsn and config_tag,
+ * plus its raft term so the primary can discover a higher term it lost track
+ * of (bug-1421 term-discovery: an AE reject response never reaches the raft
+ * library, so a heartbeat-reply term stamp is the fallback path). */
 typedef struct {
     uint64_t applied_lsn;
     uint64_t config_tag;
+    uint64_t term;
 } state_report_t;
 
 /* WAL apply ACK — sent by standby to primary after applying a WAL catch-up

src/lib/raft/raft.c

@@ -851,3 +851,24 @@ int raft_on_append_entries_reply(raft_state_t *s, int peer_idx, int success,
 
     return RAFT_OK;
 }
+
+int raft_on_higher_term_observed(raft_state_t *s, uint64_t observed_term)
+{
+    if (observed_term <= s->term)
+        return RAFT_OK;
+
+    int was_leader = (s->role == RAFT_LEADER);
+    s->term           = observed_term;
+    s->voted_for      = -1;
+    s->voted_for_term = 0;
+    s->role           = RAFT_FOLLOWER;
+    /* Persist the adopted term before notifying (mirrors UpdateTerm ordering
+     * in raft_on_append_entries_request, Req 6.2). */
+    s->cbs->on_persist(s->ctx, s->term, -1);
+    if (was_leader)
+        s->cbs->on_become_follower(s->ctx, s->term, -1);
+    if (s->cbs->on_trace)
+        s->cbs->on_trace(s->ctx, "term_adopt_observed", NULL);
+
+    return RAFT_BECAME_FOLLOWER;
+}

src/lib/raft/raft.h

@@ -403,6 +403,22 @@ int raft_on_append_entries_request(raft_state_t *s, int leader_idx,
 int raft_on_append_entries_reply(raft_state_t *s, int peer_idx, int success,
                                  uint64_t match_index, uint64_t reply_term);
 
+/*
+ * raft_on_higher_term_observed: adopt a higher term learned from ANY
+ * peer-carried term stamp that is not itself a full RequestVote/AppendEntries
+ * exchange (e.g. a heartbeat-reply state report) — closes bug-1421, where a
+ * primary never learns of a follower's higher term because AE reject
+ * responses are not routed to raft_on_append_entries_reply (which drops any
+ * reply whose term != s->term, so it cannot be reused here without corrupting
+ * match_index/next_index on the common equal-term case). Mirrors the
+ * UpdateTerm step shared by raft_on_append_entries_request (raft.tla 406):
+ * strictly-greater-than only; touches ONLY (term, voted_for, role) — never
+ * peer-indexed bookkeeping — so it is safe to call with an unresolved/unknown
+ * peer_idx. No-op (RAFT_OK) when observed_term <= s->term. Returns
+ * RAFT_BECAME_FOLLOWER when it steps a leader down.
+ */
+int raft_on_higher_term_observed(raft_state_t *s, uint64_t observed_term);
+
 /*
  * raft_advance_commit: leader recomputes its commit index (raft.tla
  * AdvanceCommitIndex, 259). A SEPARATE action from raft_on_append_entries_reply

src/schema/controller.capnp

@@ -818,6 +818,7 @@ struct ReplEnvelope $C.mapname("capnp_repl_envelope_t") {
 struct StateReport $C.mapname("state_report_t") {
   appliedLsn @0 :UInt64 $C.mapname("applied_lsn");   # standby's last applied WAL LSN
   configTag  @1 :UInt64 $C.mapname("config_tag");    # standby's current config version tag
+  term       @2 :UInt64 $C.mapname("term");          # standby's current raft term (bug-1421 term-discovery)
 }
 
 struct RequestVoteRequest $C.mapname("request_vote_req_t") {