feat(ha-heartbeat-committed-propagation): standby OP_nop consume branch + term-guard (task 2.2)

Append a committed-consume branch to the OP_nop handler (existing liveness/reply
byte-unchanged): ha_consume_heartbeat_committed applies the non-primary gate
(3.4) + stale-term guard (accept hb->term >= ha->term, 3.2), then delegates to
the monotonic ha_observe_heartbeat_committed (1.2). Absent body / zero / failed
gate are no-ops (4.2). No vote-floor edit — safety is the term-guard. 4 tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.kiro/specs/ha-heartbeat-committed-propagation/tasks.md

@@ -24,7 +24,7 @@
   - _Requirements: 2.1, 2.2, 2.3, 3.1, 3.3_
   - _Boundary: ha_observe_heartbeat_committed_
   - _Depends: 1.1_
-- [ ] 2.2 Wire the standby `OP_nop` consume branch + term-guard
+- [x] 2.2 Wire the standby `OP_nop` consume branch + term-guard
   - In the `cluster.c` `OP_nop` handler, after the existing liveness reset + `StateReport` reply, if `request.body_which == Request_body_heartbeatmsg` and the responder is non-primary: apply the term-guard (`hb->term >= atomic_load(ha->term)`) and, only if it passes, call `ha_observe_heartbeat_committed(c, hb->committed_lsn, hb->committed_config_tag)`. A missing body, a failed term-guard, or a zero committed value updates nothing and leaves all existing `OP_nop` behavior intact.
   - TDD: tests confirm a term-valid higher-committed heartbeat advances `last_known_committed_*`; a stale-term heartbeat (`hb->term < ha->term`) carrying a higher committed leaves committed unchanged but STILL resets liveness; a bodyless / zero-committed nop is processed as liveness only; and a primary responder does not consume an inbound committed value.
   - Observable: the tests pass for the term-valid, stale-term, bodyless, and primary-responder cases, with liveness reset preserved in every case.

src/controller/cluster.c

@@ -315,6 +315,15 @@ static int myreq_handler(vtask_t* task)
             else if (role == CTRL_ELECTING)
                 ha_discover_primary(&c->ha, c);
         }
+        /* ha-heartbeat-committed-propagation task 2.2: after the unchanged
+         * liveness/reply path above, a heartbeat that carries a HeartbeatMsg
+         * body lets a standby learn the primary's committed position. The
+         * consume helper applies the non-primary gate (Req 3.4) + term-guard
+         * (Req 3.2) and a NULL/zero no-op (Req 4.2); a bodyless nop skips it
+         * entirely. Purely additive — no existing OP_nop behaviour changes
+         * (Req 4.1). */
+        if (request.body_which == Request_body_heartbeatmsg)
+            ha_consume_heartbeat_committed(c, request.body.heartbeatmsg);
         break;
     }
     case OPCode_join:

src/controller/ha.c

@@ -4959,6 +4959,38 @@ void ha_observe_heartbeat_committed(cluster_t *c, uint64_t committed_lsn,
          committed_lsn, committed_config_tag, cur);
 }
 
+/*
+ * ha_consume_heartbeat_committed: standby OP_nop consume branch — the
+ * non-primary gate (Req 3.4) + the stale-term guard (Req 3.2), then the
+ * monotonic observe (Req 1.2).  Reconcile-free; updates nothing on a NULL body,
+ * a failed gate, or a zero committed value (Req 4.2).  Pure additive helper —
+ * the caller runs the unchanged liveness/reply path first (Req 4.1).
+ */
+void ha_consume_heartbeat_committed(cluster_t *c, const heartbeat_msg_t *hb)
+{
+    if (!c || !hb)
+        return;
+
+    /* (3.4) A primary uses its own authoritative committed — never an inbound
+     * heartbeat value. Only non-primary nodes consume. */
+    ctrl_role_t role =
+        (ctrl_role_t)atomic_load_explicit(&c->ha.role, memory_order_acquire);
+    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. */
+    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). */
+    ha_observe_heartbeat_committed(c, hb->committed_lsn,
+                                   hb->committed_config_tag);
+}
+
 void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha)
 {
     (void)ha;

src/controller/ha.h

@@ -899,6 +899,24 @@ void ha_on_heartbeat_received(ha_state_t *ha);
 void ha_observe_heartbeat_committed(cluster_t *c, uint64_t committed_lsn,
                                     uint64_t committed_config_tag);
 
+/*
+ * ha_consume_heartbeat_committed: the standby-side OP_nop consume branch
+ * (ha-heartbeat-committed-propagation, Req 1.2, 3.2, 3.4, 4.2).
+ *
+ * Safe to call unconditionally from the OP_nop handler with the decoded
+ * heartbeatmsg body. It applies the two safety gates before delegating to the
+ * monotonic ha_observe_heartbeat_committed:
+ *   - non-primary gate (3.4): a primary uses its own authoritative committed and
+ *     never consumes an inbound value;
+ *   - term-guard (3.2): a heartbeat whose advertised term is behind this node's
+ *     term is rejected, fencing a stale/partitioned primary from inflating the
+ *     vote-freshness floor.
+ * NULL body, a failed gate, or a zero committed value updates nothing (4.2);
+ * the helper never touches liveness or any other OP_nop behaviour (4.1 — the
+ * caller invokes this AFTER the unchanged liveness/reply path).
+ */
+void ha_consume_heartbeat_committed(cluster_t *c, const heartbeat_msg_t *hb);
+
 /* 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. */

src/controller/testsuite/test_ha_heartbeat_committed.c

@@ -18,6 +18,7 @@
 #include <unity/unity.h>
 
 #include "ha.h"
+#include "typemap.h"         /* heartbeat_msg_t for the consume-branch tests */
 #include "xce/cluster_def.h" /* full cluster_t (embedded ha_state_t / persist_t) */
 
 /* ── helpers ────────────────────────────────────────────────── */
@@ -197,6 +198,140 @@ static void test_observe_no_reconcile_no_other_mutation(void)
     printf("test_observe_no_reconcile_no_other_mutation: PASS\n");
 }
 
+/* ── task 2.2: standby consume branch — term-guard + non-primary gate ─────────
+ *
+ * ha_consume_heartbeat_committed(cluster_t*, const heartbeat_msg_t*) is the
+ * safety-critical wrapper the OP_nop handler calls: it applies the non-primary
+ * gate (3.4) and the term-guard (3.2) before delegating to the monotonic
+ * ha_observe_heartbeat_committed (1.2). It is NULL-safe and zero-safe (4.2).
+ * Unit-testable without a socket/endpoint.
+ */
+
+/* (e) Term-valid heartbeat on a standby advances the learned committed (1.2). */
+static void test_consume_term_valid_advances(void)
+{
+    cluster_t c;
+    hb_setup_cluster(&c);
+
+    atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.term, 5, memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_lsn, 4,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x11,
+                          memory_order_release);
+
+    heartbeat_msg_t hb = {
+        .sender_priority     = 0,
+        .term                = 5, /* == standby term → passes term-guard */
+        .committed_lsn       = 9,
+        .committed_config_tag = 0xAA,
+    };
+    ha_consume_heartbeat_committed(&c, &hb);
+
+    TEST_ASSERT_EQUAL_UINT64(9, hb_lsn(&c));    /* advanced (1.2) */
+    TEST_ASSERT_EQUAL_UINT64(0xAA, hb_tag(&c)); /* tag moved with lsn */
+
+    hb_teardown_cluster(&c);
+    printf("test_consume_term_valid_advances: PASS\n");
+}
+
+/* (f) Stale-term heartbeat (hb->term < ha->term) carrying a higher committed
+ * leaves the learned committed UNCHANGED — the term-guard fences a stale primary
+ * from inflating the vote-freshness floor (3.2). */
+static void test_consume_stale_term_ignored(void)
+{
+    cluster_t c;
+    hb_setup_cluster(&c);
+
+    atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.term, 7, memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_lsn, 4,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x11,
+                          memory_order_release);
+
+    heartbeat_msg_t hb = {
+        .sender_priority     = 0,
+        .term                = 6, /* < standby term 7 → fails term-guard */
+        .committed_lsn       = 99,
+        .committed_config_tag = 0xBB,
+    };
+    ha_consume_heartbeat_committed(&c, &hb);
+
+    TEST_ASSERT_EQUAL_UINT64(4, hb_lsn(&c));    /* unchanged (3.2) */
+    TEST_ASSERT_EQUAL_UINT64(0x11, hb_tag(&c)); /* unchanged */
+
+    hb_teardown_cluster(&c);
+    printf("test_consume_stale_term_ignored: PASS\n");
+}
+
+/* (g) A primary responder does not consume an inbound committed value — the
+ * primary uses its own authoritative committed (3.4). */
+static void test_consume_primary_does_not_consume(void)
+{
+    cluster_t c;
+    hb_setup_cluster(&c);
+
+    atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.term, 5, memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_lsn, 4,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x11,
+                          memory_order_release);
+
+    heartbeat_msg_t hb = {
+        .sender_priority     = 0,
+        .term                = 9, /* fresh term, higher committed — still ignored */
+        .committed_lsn       = 50,
+        .committed_config_tag = 0xCC,
+    };
+    ha_consume_heartbeat_committed(&c, &hb);
+
+    TEST_ASSERT_EQUAL_UINT64(4, hb_lsn(&c));    /* unchanged (3.4) */
+    TEST_ASSERT_EQUAL_UINT64(0x11, hb_tag(&c)); /* unchanged */
+
+    hb_teardown_cluster(&c);
+    printf("test_consume_primary_does_not_consume: PASS\n");
+}
+
+/* (h) NULL body and zero committed are no-ops — an absent/liveness-only
+ * heartbeat updates nothing (4.2). */
+static void test_consume_null_or_zero_noop(void)
+{
+    cluster_t c;
+    hb_setup_cluster(&c);
+
+    atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.term, 5, memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_lsn, 6,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x66,
+                          memory_order_release);
+
+    /* NULL body → no-op (absent heartbeatmsg). */
+    ha_consume_heartbeat_committed(&c, NULL);
+    TEST_ASSERT_EQUAL_UINT64(6, hb_lsn(&c));
+    TEST_ASSERT_EQUAL_UINT64(0x66, hb_tag(&c));
+
+    /* zero committed (valid term, standby) → no-op. */
+    heartbeat_msg_t hb = {
+        .sender_priority     = 0,
+        .term                = 5,
+        .committed_lsn       = 0,
+        .committed_config_tag = 0xDD,
+    };
+    ha_consume_heartbeat_committed(&c, &hb);
+    TEST_ASSERT_EQUAL_UINT64(6, hb_lsn(&c));
+    TEST_ASSERT_EQUAL_UINT64(0x66, hb_tag(&c));
+
+    hb_teardown_cluster(&c);
+    printf("test_consume_null_or_zero_noop: PASS\n");
+}
+
 /* ── suite entry point ─────────────────────────────────────── */
 
 void test_ha_heartbeat_committed(void)
@@ -205,4 +340,9 @@ void test_ha_heartbeat_committed(void)
     RUN_TEST(test_observe_noop_on_equal_or_lower);
     RUN_TEST(test_observe_noop_on_zero);
     RUN_TEST(test_observe_no_reconcile_no_other_mutation);
+    /* task 2.2: consume branch — term-guard + non-primary gate */
+    RUN_TEST(test_consume_term_valid_advances);
+    RUN_TEST(test_consume_stale_term_ignored);
+    RUN_TEST(test_consume_primary_does_not_consume);
+    RUN_TEST(test_consume_null_or_zero_noop);
 }