myxce/src/controller/testsuite/test_ha_heartbeat_committed.c

/* 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 "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 ────────────────────────────────────────────────── */

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

static uint64_t hb_elected(const cluster_t* c)
{
    return atomic_load_explicit(&c->ha.last_known_elected_at_ms,
                                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");
}

/* ── 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");
}

/* ── controller-status delta (Req 12.2): learn the primary's election time
 * off a received heartbeat ─────────────────────────────────────────────────
 *
 * ha_consume_heartbeat_committed's SAME call site that feeds
 * ha_observe_heartbeat_committed(c, hb->committed_lsn, hb->committed_config_tag)
 * also carries hb->elected_at_ms — a genuinely-RECEIVED heartbeat (the
 * production caller is cluster.c's OP_nop dispatch) already gates on
 * non-primary role + a non-stale term before either call, so the
 * elected_at_ms store is a plain unconditional overwrite immediately
 * alongside the existing committed-position learn — no separate monotonic/
 * term compare (design.md Locked Decision 2). ha_observe_heartbeat_committed
 * itself is untouched; these tests exercise ONLY the new sibling store.
 */

/* (i) A term-valid heartbeat stores the carried election time as the
 * standby's own learned value (Req 12.2). */
static void test_consume_stores_elected_at_ms(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_elected_at_ms, 111,
                          memory_order_release);

    heartbeat_msg_t hb = {
        .sender_priority     = 0,
        .term                = 5, /* == standby term -> passes term-guard */
        .committed_lsn       = 9,
        .committed_config_tag = 0xAA,
        .elected_at_ms       = 222222,
    };
    ha_consume_heartbeat_committed(&c, &hb);

    TEST_ASSERT_EQUAL_UINT64(222222, hb_elected(&c)); /* learned (12.2) */

    hb_teardown_cluster(&c);
    printf("test_consume_stores_elected_at_ms: PASS\n");
}

/* (j) Reconcile-free: a heartbeat carrying ONLY a new election time (its
 * committed_lsn is 0, a liveness-only no-op for the committed observe) moves
 * last_known_elected_at_ms and NOTHING else -- committed position, role,
 * term, and applied position are all untouched. Mirrors
 * test_observe_no_reconcile_no_other_mutation's shape for the new field
 * (Req 12.2, 12.5). This is the RED test: it fails (elected_at_ms stays at
 * its seeded value) before the store exists in ha_consume_heartbeat_committed. */
static void test_consume_elected_at_ms_reconcile_free(void)
{
    cluster_t c;
    hb_setup_cluster(&c);

    /* Seed every learned/produced HA field the store must NOT touch. */
    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.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;
    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);
    atomic_store_explicit(&c.ha.last_known_elected_at_ms, 1111,
                          memory_order_release);

    /* Snapshot everything that must remain byte-for-byte unchanged. */
    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);
    uint64_t klsn0    = hb_lsn(&c);
    uint64_t ktag0    = hb_tag(&c);

    heartbeat_msg_t hb = {
        .sender_priority     = 0,
        .term                = 5,    /* == standby term -> passes term-guard */
        .committed_lsn       = 0,    /* liveness-only -> observe no-ops (4.2) */
        .committed_config_tag = 0xDEAD,
        .elected_at_ms       = 9999,
    };
    ha_consume_heartbeat_committed(&c, &hb);

    /* ONLY last_known_elected_at_ms changed. */
    TEST_ASSERT_EQUAL_UINT64(9999, hb_elected(&c));

    /* Committed position -- untouched (observe no-op on committed_lsn==0). */
    TEST_ASSERT_EQUAL_UINT64(klsn0, hb_lsn(&c));
    TEST_ASSERT_EQUAL_UINT64(ktag0, hb_tag(&c));

    /* No truncation / no produced-state mutation. */
    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_consume_elected_at_ms_reconcile_free: PASS\n");
}

/* (k) A stale-term heartbeat (hb->term < ha->term) must NOT store its carried
 * election time either -- the elected_at_ms store sits behind the SAME
 * term-guard early-return as the committed-position learn (3.2 shape). */
static void test_consume_stale_term_does_not_store_elected_at_ms(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_elected_at_ms, 4242,
                          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,
        .elected_at_ms       = 555555,
    };
    ha_consume_heartbeat_committed(&c, &hb);

    TEST_ASSERT_EQUAL_UINT64(4242, hb_elected(&c)); /* unchanged (3.2 shape) */

    hb_teardown_cluster(&c);
    printf("test_consume_stale_term_does_not_store_elected_at_ms: PASS\n");
}

/* ── suite entry point ─────────────────────────────────────── */

void test_ha_heartbeat_committed(void)
{
    UnitySetTestFile(__FILE__);
    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);
    /* 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);
    /* controller-status delta (task 7.3): heartbeat-carried election time */
    RUN_TEST(test_consume_stores_elected_at_ms);
    RUN_TEST(test_consume_elected_at_ms_reconcile_free);
    RUN_TEST(test_consume_stale_term_does_not_store_elected_at_ms);
}