feat(ha-heartbeat-committed-propagation): add reconcile-free ha_observe_heartbeat_committed (task 2.1)

Strict-monotonic updater for a standby's learned committed position: advance
last_known_committed_lsn only when strictly greater (zero/lower no-op), set the
config tag together (tag-before-lsn release ordering for reader safety). No
ha_rejoin_reconcile, no truncation, no other mutation (Req 2.1-2.3, 3.1, 3.3).
4 revert-detecting unit tests; controller suite 0 failures.

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

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

@@ -17,7 +17,7 @@
   - _Boundary: libxce sender_
 
 - [ ] 2. Standby consume (reconcile-free, term-guarded, monotonic)
-- [ ] 2.1 Add the reconcile-free monotonic updater `ha_observe_heartbeat_committed`
+- [x] 2.1 Add the reconcile-free monotonic updater `ha_observe_heartbeat_committed`
   - Add `ha_observe_heartbeat_committed(cluster_t* c, uint64_t committed_lsn, uint64_t committed_config_tag)` (`ha.c`, declared in `ha.h`): advance `last_known_committed_lsn` only when the carried LSN is strictly greater (monotonic; atomic), advancing `last_known_committed_config_tag` together with it; no-op when `<=` current or zero. It must NOT call `ha_rejoin_reconcile`, truncate the WAL, or modify any produced/persisted committed state.
   - TDD: unit tests assert advance-on-higher (LSN + config tag updated), no-op on equal/lower/zero, config tag never regresses independently, and — critically — that the call performs no WAL truncation / no reconcile and mutates nothing beyond `last_known_committed_*`.
   - Observable: the unit tests go red → green; the no-truncation / no-mutation assertions hold.

src/controller/ha.c

@@ -4919,6 +4919,46 @@ void ha_on_heartbeat_received(ha_state_t *ha)
     }
 }
 
+/*
+ * ha_observe_heartbeat_committed: reconcile-free monotonic learn of the
+ * primary's committed position from a heartbeat (ha-heartbeat-committed-
+ * propagation, Req 2.1, 2.2, 2.3, 3.1, 3.3).
+ *
+ * Single-receiver-thread path (controller request handling), so a relaxed
+ * compare-then-store is sufficient. The config tag is written BEFORE the LSN
+ * (both release-ordered) so a concurrent acquire-load reader that observes the
+ * advanced LSN never pairs it with a stale tag.
+ *
+ * Advisory only: NO ha_rejoin_reconcile / NO WAL truncation / NO change to any
+ * produced/persisted committed, role, term, or applied state. This is the
+ * explicit distinction from cluster_apply_ctrlhello_reply (which truncates).
+ */
+void ha_observe_heartbeat_committed(cluster_t *c, uint64_t committed_lsn,
+                                    uint64_t committed_config_tag)
+{
+    if (!c) return;
+    ha_state_t *ha = &c->ha;
+
+    uint64_t cur = atomic_load_explicit(&ha->last_known_committed_lsn,
+                                        memory_order_acquire);
+    /* Monotonic no-op: a zero (absent/liveness-only) or non-advancing LSN must
+     * leave the learned position — and its config tag — untouched (Req 2.2). */
+    if (committed_lsn == 0 || committed_lsn <= cur)
+        return;
+
+    /* Tag moves with the LSN, never independently regressed (Req 2.3): publish
+     * the tag first so the LSN's release store carries the matching tag. */
+    atomic_store_explicit(&ha->last_known_committed_config_tag,
+                          committed_config_tag, memory_order_release);
+    atomic_store_explicit(&ha->last_known_committed_lsn, committed_lsn,
+                          memory_order_release);
+    /* Reconcile-free: nothing else is touched (Req 3.1, 3.3). */
+    tlog(TLOG_DEBUG3,
+         "HA: learned committed_lsn=%" PRIu64 " config_tag=%" PRIu64
+         " from heartbeat (was %" PRIu64 ")",
+         committed_lsn, committed_config_tag, cur);
+}
+
 void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha)
 {
     (void)ha;

src/controller/ha.h

@@ -878,6 +878,27 @@ void ha_discover_primary(ha_state_t *ha, cluster_t *c);
 /* ha_on_heartbeat_received: reset the heartbeat deadline (Req 6.3). */
 void ha_on_heartbeat_received(ha_state_t *ha);
 
+/*
+ * ha_observe_heartbeat_committed: reconcile-free monotonic update of a standby's
+ * learned committed position from a primary heartbeat
+ * (ha-heartbeat-committed-propagation, Req 2.1, 2.2, 2.3, 3.1, 3.3).
+ *
+ * Advances ha->last_known_committed_lsn to committed_lsn ONLY when it is
+ * strictly greater than the current value (monotonic); when it advances, the
+ * carried committed_config_tag is stored together with it (they move as a pair —
+ * the tag never regresses independently). No-op when committed_lsn is 0 or
+ * <= the current learned LSN.
+ *
+ * This is the reconcile-free sibling of cluster_apply_ctrlhello_reply: it MUST
+ * NOT call ha_rejoin_reconcile, truncate the WAL, or modify ANY other state
+ * (produced/persisted committed, role, term, applied, etc.). The value is
+ * advisory — for observability and vote-floor freshness only. The caller (the
+ * OP_nop receive path) applies the term-guard before invoking this; the helper
+ * itself is term-agnostic and single-receiver-thread safe.
+ */
+void ha_observe_heartbeat_committed(cluster_t *c, uint64_t committed_lsn,
+                                    uint64_t committed_config_tag);
+
 /* 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/meson.build

@@ -134,6 +134,7 @@ controller_unit_test_src = [
   'testsuite/test_coldstart_lsn0.c',
   'testsuite/test_commit_durability.c',
   'testsuite/test_ctrlstatus.c',
+  'testsuite/test_ha_heartbeat_committed.c',
   'ctrlstatus.c',
   'commit.c',
   'job.c',

src/controller/testsuite/controller_test.c

@@ -61,6 +61,7 @@ extern void test_convergence_verification(void);
 extern void test_coldstart_lsn0(void);
 extern void test_commit_durability(void);
 extern void test_ctrlstatus(void);
+extern void test_ha_heartbeat_committed(void);
 
 void setUp(void) { return; }
 
@@ -191,6 +192,7 @@ int main(int argc, char* argv[])
     TEST(test_coldstart_lsn0);
     TEST(test_commit_durability);
     TEST(test_ctrlstatus);
+    TEST(test_ha_heartbeat_committed);
 
     return UNITY_END();
 }

src/controller/testsuite/test_ha_heartbeat_committed.c

@@ -0,0 +1,208 @@
+/* test_ha_heartbeat_committed.c — unit tests for the reconcile-free monotonic
+ * heartbeat-borne committed updater (ha-heartbeat-committed-propagation,
+ * task 2.1).
+ *
+ * Exercises ha_observe_heartbeat_committed(cluster_t*, committed_lsn,
+ * committed_config_tag): the standby-side consumer that monotonically advances
+ * ha->last_known_committed_lsn / last_known_committed_config_tag from a primary
+ * heartbeat WITHOUT any reconciliation or WAL truncation. This is the explicit
+ * reconcile-free sibling of cluster_apply_ctrlhello_reply (which DOES truncate).
+ *
+ * Covers Requirements 2.1, 2.2, 2.3, 3.1, 3.3.
+ */
+#include <stdatomic.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#include "ha.h"
+#include "xce/cluster_def.h" /* full cluster_t (embedded ha_state_t / persist_t) */
+
+/* ── helpers ────────────────────────────────────────────────── */
+
+static ha_config_t hb_valid_cfg(void)
+{
+    ha_config_t cfg = {
+        .heartbeat_interval_ms    = 3000,
+        .heartbeat_timeout_ms     = 10000,
+        .election_timeout_ms      = 2000,
+        .probe_connect_timeout_ms = 500,
+        .max_promotion_retries    = 5,
+    };
+    return cfg;
+}
+
+/* Build a minimal cluster_t whose ha state the updater touches. Only the fields
+ * ha_observe_heartbeat_committed() reads/writes are initialised by ha_init. */
+static void hb_setup_cluster(cluster_t* c)
+{
+    memset(c, 0, sizeof(*c));
+    ha_config_t cfg = hb_valid_cfg();
+    TEST_ASSERT_EQUAL_INT(0, ha_init(&c->ha, &cfg, 0));
+}
+
+static void hb_teardown_cluster(cluster_t* c)
+{
+    ha_fini(&c->ha);
+}
+
+static uint64_t hb_lsn(const cluster_t* c)
+{
+    return atomic_load_explicit(&c->ha.last_known_committed_lsn,
+                                memory_order_acquire);
+}
+
+static uint64_t hb_tag(const cluster_t* c)
+{
+    return atomic_load_explicit(&c->ha.last_known_committed_config_tag,
+                                memory_order_acquire);
+}
+
+/* ── tests ──────────────────────────────────────────────────── */
+
+/* (a) Advance on a strictly higher carried LSN: both the LSN and the config tag
+ * move together to the carried values. (Req 2.1, 2.3) */
+static void test_observe_advances_on_higher(void)
+{
+    cluster_t c;
+    hb_setup_cluster(&c);
+
+    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);
+
+    ha_observe_heartbeat_committed(&c, 9, 0xAA);
+
+    TEST_ASSERT_EQUAL_UINT64(9, hb_lsn(&c));     /* advanced (2.1) */
+    TEST_ASSERT_EQUAL_UINT64(0xAA, hb_tag(&c));  /* tag moved with lsn (2.3) */
+
+    hb_teardown_cluster(&c);
+    printf("test_observe_advances_on_higher: PASS\n");
+}
+
+/* (b) No-op on equal or lower carried LSN: neither the LSN nor the tag changes,
+ * proving the config tag never regresses independently of the LSN. (Req 2.2, 2.3) */
+static void test_observe_noop_on_equal_or_lower(void)
+{
+    cluster_t c;
+    hb_setup_cluster(&c);
+
+    atomic_store_explicit(&c.ha.last_known_committed_lsn, 9,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0xAA,
+                          memory_order_release);
+
+    /* equal LSN, different tag → no change (2.2): tag must NOT take 0xBB */
+    ha_observe_heartbeat_committed(&c, 9, 0xBB);
+    TEST_ASSERT_EQUAL_UINT64(9, hb_lsn(&c));
+    TEST_ASSERT_EQUAL_UINT64(0xAA, hb_tag(&c));
+
+    /* lower LSN (out-of-order / lagging primary), different tag → no change */
+    ha_observe_heartbeat_committed(&c, 3, 0xCC);
+    TEST_ASSERT_EQUAL_UINT64(9, hb_lsn(&c));     /* unchanged (2.2) */
+    TEST_ASSERT_EQUAL_UINT64(0xAA, hb_tag(&c));  /* no independent regress (2.3) */
+
+    hb_teardown_cluster(&c);
+    printf("test_observe_noop_on_equal_or_lower: PASS\n");
+}
+
+/* (c) No-op on a zero carried LSN: an absent/liveness-only committed value must
+ * leave the known position untouched. (Req 2.2 / liveness no-op) */
+static void test_observe_noop_on_zero(void)
+{
+    cluster_t c;
+    hb_setup_cluster(&c);
+
+    atomic_store_explicit(&c.ha.last_known_committed_lsn, 5,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x55,
+                          memory_order_release);
+
+    ha_observe_heartbeat_committed(&c, 0, 0xDD);
+
+    TEST_ASSERT_EQUAL_UINT64(5, hb_lsn(&c));     /* unchanged */
+    TEST_ASSERT_EQUAL_UINT64(0x55, hb_tag(&c));  /* unchanged */
+
+    hb_teardown_cluster(&c);
+    printf("test_observe_noop_on_zero: PASS\n");
+}
+
+/* (d) Reconcile-free: an advancing observe must mutate ONLY last_known_committed_*
+ * and nothing else — no WAL truncation, no produced/persisted committed change,
+ * no role/term/applied change. (Req 3.1, 3.3) */
+static void test_observe_no_reconcile_no_other_mutation(void)
+{
+    cluster_t c;
+    hb_setup_cluster(&c);
+
+    /* Seed the authoritative / produced state the helper must NOT touch. */
+    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.applied_lsn, 100, memory_order_release);
+    atomic_store_explicit(&c.persist.committed_lsn, 80, memory_order_release);
+    atomic_store_explicit(&c.persist.durable_lsn, 80, memory_order_release);
+    c.persist.current_lsn = 81; /* WAL tip (non-atomic, under p->lock in prod) */
+    atomic_store_explicit(&c.ha.committed_config_tag, 0xC0FFEE,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_lsn, 60,
+                          memory_order_release);
+    atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x6060,
+                          memory_order_release);
+
+    /* Snapshot the produced/persisted state that MUST remain untouched. */
+    int32_t  role0    = atomic_load_explicit(&c.ha.role, memory_order_acquire);
+    uint64_t term0    = atomic_load_explicit(&c.ha.term, memory_order_acquire);
+    uint64_t applied0 = atomic_load_explicit(&c.ha.applied_lsn,
+                                             memory_order_acquire);
+    uint64_t pcommit0 = atomic_load_explicit(&c.persist.committed_lsn,
+                                             memory_order_acquire);
+    uint64_t pdur0    = atomic_load_explicit(&c.persist.durable_lsn,
+                                             memory_order_acquire);
+    uint64_t ptip0    = c.persist.current_lsn;
+    uint64_t ctag0    = atomic_load_explicit(&c.ha.committed_config_tag,
+                                             memory_order_acquire);
+
+    /* Advance the learned committed via the heartbeat path. */
+    ha_observe_heartbeat_committed(&c, 75, 0x7575);
+
+    /* ONLY last_known_committed_* changed. */
+    TEST_ASSERT_EQUAL_UINT64(75, hb_lsn(&c));
+    TEST_ASSERT_EQUAL_UINT64(0x7575, hb_tag(&c));
+
+    /* No truncation: the persisted/produced committed tip is intact (3.1, 3.3). */
+    TEST_ASSERT_EQUAL_UINT64(pcommit0,
+                             atomic_load_explicit(&c.persist.committed_lsn,
+                                                  memory_order_acquire));
+    TEST_ASSERT_EQUAL_UINT64(pdur0,
+                             atomic_load_explicit(&c.persist.durable_lsn,
+                                                  memory_order_acquire));
+    TEST_ASSERT_EQUAL_UINT64(ptip0, c.persist.current_lsn);
+    /* No other HA state moved. */
+    TEST_ASSERT_EQUAL_INT(role0,
+                          atomic_load_explicit(&c.ha.role, memory_order_acquire));
+    TEST_ASSERT_EQUAL_UINT64(term0,
+                             atomic_load_explicit(&c.ha.term, memory_order_acquire));
+    TEST_ASSERT_EQUAL_UINT64(applied0,
+                             atomic_load_explicit(&c.ha.applied_lsn,
+                                                  memory_order_acquire));
+    TEST_ASSERT_EQUAL_UINT64(ctag0,
+                             atomic_load_explicit(&c.ha.committed_config_tag,
+                                                  memory_order_acquire));
+
+    hb_teardown_cluster(&c);
+    printf("test_observe_no_reconcile_no_other_mutation: PASS\n");
+}
+
+/* ── suite entry point ─────────────────────────────────────── */
+
+void test_ha_heartbeat_committed(void)
+{
+    RUN_TEST(test_observe_advances_on_higher);
+    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);
+}