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

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

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