diff --git a/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md b/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md index 040b0d0a..125d51e6 100644 --- a/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md +++ b/.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. diff --git a/src/controller/cluster.c b/src/controller/cluster.c index a8e832b5..bc0fb5ae 100644 --- a/src/controller/cluster.c +++ b/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: diff --git a/src/controller/ha.c b/src/controller/ha.c index 6cf1dccd..47d948c7 100644 --- a/src/controller/ha.c +++ b/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; diff --git a/src/controller/ha.h b/src/controller/ha.h index 9392c4f1..19de6fa6 100644 --- a/src/controller/ha.h +++ b/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. */ diff --git a/src/controller/testsuite/test_ha_heartbeat_committed.c b/src/controller/testsuite/test_ha_heartbeat_committed.c index 99cb0863..d3db2e2a 100644 --- a/src/controller/testsuite/test_ha_heartbeat_committed.c +++ b/src/controller/testsuite/test_ha_heartbeat_committed.c @@ -18,6 +18,7 @@ #include #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); }