myxce/src/controller/testsuite/test_repl_majority.c

/*
 * test_repl_majority.c — majority-commit replication tests.
 *
 * Tests for:
 *   repl_wait_majority_ack(replicator_t *, uint64_t entry_lsn,
 *                          uint32_t timeout_ms, uint8_t ack_op)
 *   repl_broadcast majority semantics (pending_ack_count, over-limit eviction)
 *
 * ha-commit-engine task 5.1: the walAck cases are wired to a commit engine
 * (stub persist_t fixture + repl_set_commit_engine, pattern from
 * test_commit_engine.c) and assert PER-LSN semantics — entry N is confirmed
 * only when self + a majority of active standbys have acked_lsn >= N, and
 * confirmation advances committed_lsn through the engine. They are RED
 * against the unwired legacy counting (`acked_lsn > 0`, n_active
 * denominator, zero-active success short-circuit) and go green with tasks
 * 5.2/5.3.
 *
 * ha-commit-barrier-rollout task 5.1: the configAck case now follows the
 * per-push arithmetic (design §Config Barrier): the call passes the config
 * PUSH TAG as entry_lsn (no longer a vacuous 0 — the hardened branch counts
 * standbys with acked_config_lsn >= push_tag, majority 1 (self) + count >
 * n_peers / 2 over configured members). The expectation below (-ETIMEDOUT
 * with zero covering acks) holds under BOTH the legacy counting and the
 * per-push arithmetic; the RED carriers for the arithmetic change live in
 * test_commit_barrier_rollout.c (task 5.1 section).
 *
 * Requirements: ha-commit-engine 1.1, 1.2, 2.1, 4.2, 4.4, 6.3
 * (original repl coverage: 5.1, 5.4, 5.5, 5.8, 8.2)
 */
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#include "commit.h"
#include "config.h"
#include "controller.capnp.h"  /* OPCode_walAck, OPCode_upgradeConfigAck */
#include "persist.h"
#include "repl.h"

/* MAX_PENDING_ACKS: defined in repl.h after task 7.2; fallback for RED phase */
#ifndef MAX_PENDING_ACKS
#define MAX_PENDING_ACKS 16
#endif

/* ── Helpers ───────────────────────────────────────────────────────────────── */

/* Dummy endpoints — repl.c only compares pointers */
static int ep0, ep1, ep2;

/* ── Engine-wired fixture (ha-commit-engine task 5.1) ──────────────────────
 * Stub persist_t whose live parts are committed_lsn, current_lsn, and the
 * embedded replicator registry (test_commit_engine.c pattern). Statics
 * survive Unity's longjmp on assertion failure. */
static persist_t       g_mp;
static commit_engine_t g_meng;

static int *g_meps[3] = { &ep0, &ep1, &ep2 };

/* Arrange a fresh wired fixture: n_peers cluster members (self counted),
 * n_standbys connected standbys, entries 1..appended_lsn appended locally
 * (persist invariant: current_lsn = appended_lsn + 1). */
static void majority_setup(int n_peers, int n_standbys, uint64_t appended_lsn)
{
    int  i;
    char name[16];

    memset(&g_mp, 0, sizeof(g_mp));
    TEST_ASSERT_EQUAL_INT(0, repl_init(&g_mp.repl));
    for (i = 0; i < n_standbys; i++) {
        snprintf(name, sizeof(name), "sb%d", i);
        TEST_ASSERT_EQUAL_INT(
            i, repl_add_standby(&g_mp.repl, g_meps[i], name,
                                7000 + (uint32_t)i));
    }
    g_mp.current_lsn = appended_lsn + 1;
    atomic_store(&g_mp.committed_lsn, 0);

    TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_meng, &g_mp, n_peers, 8));
    repl_set_commit_engine(&g_mp.repl, &g_meng);
}

static void majority_teardown(void)
{
    commit_engine_fini(&g_meng);
    repl_fini(&g_mp.repl);
}

static uint64_t m_committed(void)
{
    return atomic_load(&g_mp.committed_lsn);
}

/* ACK a standby from another thread (production feed: repl_on_ack) */
struct ack_args {
    replicator_t *r;
    void         *ep;
    uint64_t      lsn;
    uint32_t      delay_ms;
};

static void *ack_thread(void *arg)
{
    struct ack_args *a = arg;
    if (a->delay_ms > 0) {
        struct timespec ts = { .tv_sec = 0,
                               .tv_nsec = (long)a->delay_ms * 1000000L };
        nanosleep(&ts, NULL);
    }
    repl_on_ack(a->r, a->ep, a->lsn);
    return NULL;
}

/* ── Test 1: per-entry majority — 2/3 standbys ACK the entry → 0 ──────────
 * 4-member cluster (self + 3): entry 1 confirmed when 2 standbys' ACK
 * watermarks cover LSN 1; confirmation advances committed_lsn through the
 * wired engine. (ha-commit-engine Req 1.1, 1.2) */

/* Static so ack_args survive Unity longjmp if assert fires before join */
static struct ack_args g_a1_0, g_a1_1;

static void test_wait_majority_ack_two_of_three_returns_zero(void)
{
    majority_setup(4, 3, 1);

    /* 2 of 3 standbys ACK watermark 1 — self + 2 = 3 >= majority(3) of 4 */
    pthread_t t0, t1;
    g_a1_0 = (struct ack_args){ &g_mp.repl, &ep0, 1, 10 };
    g_a1_1 = (struct ack_args){ &g_mp.repl, &ep1, 1, 20 };
    pthread_create(&t0, NULL, ack_thread, &g_a1_0);
    pthread_create(&t1, NULL, ack_thread, &g_a1_1);

    int rc = repl_wait_majority_ack(&g_mp.repl, 1, 2000, (uint8_t)OPCode_walAck);

    /* Join threads before asserting — prevents use-after-return if assert longjmps */
    pthread_join(t0, NULL);
    pthread_join(t1, NULL);

    TEST_ASSERT_EQUAL_INT(0, rc);
    /* Per-LSN semantics: confirmation means the ENGINE committed entry 1 */
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, m_committed(),
        "majority confirmation must advance committed_lsn to the entry");

    majority_teardown();
    printf("test_wait_majority_ack_two_of_three_returns_zero: PASS\n");
}

/* ── Test 2: stale earlier-entry ACKs → -ETIMEDOUT (per-LSN regression) ───
 * Standbys' watermarks cover only entry 1; waiting on entry 2 must time
 * out — `acked_lsn > 0` ("ACKed anything ever") counting would dishonestly
 * confirm it. committed_lsn never moves past the unacked entry.
 * (ha-commit-engine Req 1.2, 1.3, 2.1, 4.4) */
static void test_wait_majority_ack_stale_acks_time_out(void)
{
    majority_setup(4, 3, 2);

    /* Both ACKs cover only entry 1 — stale relative to entry 2 */
    repl_on_ack(&g_mp.repl, &ep0, 1);
    repl_on_ack(&g_mp.repl, &ep1, 1);

    int rc = repl_wait_majority_ack(&g_mp.repl, 2, 100, (uint8_t)OPCode_walAck);
    TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc,
        "earlier-entry ACK watermarks must never confirm a later entry");
    TEST_ASSERT_TRUE_MESSAGE(m_committed() < 2,
        "committed_lsn must stay unmoved past the unacked entry");

    majority_teardown();
    printf("test_wait_majority_ack_stale_acks_time_out: PASS\n");
}

/* ── Test 3: single-member cluster → 0 immediately, committed advances ────
 * n_peers=1 self-quorum: local confirmation through the engine advances
 * committed_lsn to the entry. (ha-commit-engine Req 6.3) */
static void test_wait_majority_ack_no_standbys_returns_zero(void)
{
    majority_setup(1, 0, 1);

    int rc = repl_wait_majority_ack(&g_mp.repl, 1, 100, (uint8_t)OPCode_walAck);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, m_committed(),
        "single-member self-quorum must advance committed_lsn to the entry");

    majority_teardown();
    printf("test_wait_majority_ack_no_standbys_returns_zero: PASS\n");
}

/* ── Test 4: in_catch_up standby's ACK never counts toward the entry ──────
 * 3-member cluster: the catch-up standby's covering watermark is excluded;
 * only the active standby's ACK (self + 1 = majority of 3) confirms.
 * (ha-commit-engine Req 1.2, 4.2) */
static struct ack_args g_a4_0;

static void test_wait_majority_ack_catchup_excluded_from_quorum(void)
{
    majority_setup(3, 2, 1);

    /* sb1 enters catch-up, then ACKs a covering watermark — excluded */
    repl_start_catchup(&g_mp.repl, &ep1, 0);
    repl_on_ack(&g_mp.repl, &ep1, 1);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, m_committed(),
        "catch-up standby's ACK must not commit the entry");

    /* sb0 (active) ACKs — self + sb0 reach majority */
    pthread_t t0;
    g_a4_0 = (struct ack_args){ &g_mp.repl, &ep0, 1, 10 };
    pthread_create(&t0, NULL, ack_thread, &g_a4_0);

    int rc = repl_wait_majority_ack(&g_mp.repl, 1, 2000, (uint8_t)OPCode_walAck);

    pthread_join(t0, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, m_committed(),
        "active-standby majority must advance committed_lsn to the entry");

    majority_teardown();
    printf("test_wait_majority_ack_catchup_excluded_from_quorum: PASS\n");
}

/* ── Test 5: OP_walAck vs OP_upgradeConfigAck dispatched independently ─────
 * walAck half is engine-wired per-LSN; the configAck half passes a real
 * push tag as entry_lsn (per-push arithmetic, barrier-rollout task 5.1):
 * n_peers=2, one active standby with NO covering config ack → timeout under
 * both the legacy counting (acked_config_lsn 0, not > 0) and the hardened
 * arithmetic (count(acked_config_lsn >= tag) = 0; 1 + 0 > 2/2 is false). */
static struct ack_args g_a5_0;

static void test_wait_majority_ack_ack_op_discriminates(void)
{
    majority_setup(2, 1, 1);

    pthread_t t0;
    g_a5_0 = (struct ack_args){ &g_mp.repl, &ep0, 1, 10 };
    pthread_create(&t0, NULL, ack_thread, &g_a5_0);

    int rc_wal = repl_wait_majority_ack(&g_mp.repl, 1, 2000,
                                        (uint8_t)OPCode_walAck);

    pthread_join(t0, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc_wal);
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, m_committed(),
        "walAck confirmation must advance committed_lsn per-LSN");

    /* upgradeConfigAck for push tag 0xC0FFEE with no covering config ACKs
     * → timeout (entry_lsn carries the push tag — per-push arithmetic) */
    int rc_cfg = repl_wait_majority_ack(&g_mp.repl, 0xC0FFEE, 50,
                                        (uint8_t)OPCode_upgradeConfigAck);
    TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc_cfg);

    majority_teardown();
    printf("test_wait_majority_ack_ack_op_discriminates: PASS\n");
}

/* ── Test 6: pending_ack_count > max removes standby from active set ──────── */

static void test_broadcast_over_limit_removes_standby(void)
{
    replicator_t r;
    repl_init(&r);
    repl_add_standby(&r, &ep0, "sb0", 7000);

    /* Artificially inflate pending_ack_count past the limit.
     * After overflow the standby should be evicted (in_catch_up=1 or removed). */
    for (int i = 0; i < r.standbys[0].ep != NULL ? 1 : 0; i++) {
        /* Access the pending_ack_count field — this will fail to compile if
         * the field does not exist on standby_conn_t */
        r.standbys[0].pending_ack_count = MAX_PENDING_ACKS + 1;
    }
    /* Also verify the field exists via a direct write */
    r.standbys[0].pending_ack_count = MAX_PENDING_ACKS + 1;

    /* Broadcast a wal entry — the over-limit standby should be evicted */
    wal_entry_hdr_t hdr = { .type = WAL_JOB_SUBMIT, .lsn = 1 };
    int rc = repl_broadcast(&r, &hdr, NULL, 0);
    /* repl_broadcast itself may return 0 even after eviction */
    (void)rc;

    /* After broadcast with pending_ack_count > max, standby should be in catch-up */
    int evicted = 0;
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r.standbys[i].ep == &ep0) {
            if (atomic_load_explicit(&r.standbys[i].in_catch_up,
                                     memory_order_acquire))
                evicted = 1;
        }
    }
    /* Alternatively the standby may have been fully removed */
    if (!evicted) {
        evicted = (r.nstandbys == 0);
    }
    TEST_ASSERT_TRUE(evicted);

    repl_fini(&r);
    printf("test_broadcast_over_limit_removes_standby: PASS\n");
}

/* ── Test group entry point ────────────────────────────────────────────────── */

void test_repl_majority(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_wait_majority_ack_two_of_three_returns_zero);
    RUN_TEST(test_wait_majority_ack_stale_acks_time_out);
    RUN_TEST(test_wait_majority_ack_no_standbys_returns_zero);
    RUN_TEST(test_wait_majority_ack_catchup_excluded_from_quorum);
    RUN_TEST(test_wait_majority_ack_ack_op_discriminates);
    RUN_TEST(test_broadcast_over_limit_removes_standby);
    UNITY_END();
}