myxce/src/controller/testsuite/test_late_joiner_catchup_kick.c

/*
 * test_late_joiner_catchup_kick.c — bug-1913: a standby that registers AFTER
 * the new primary appended its own-term WAL_NOOP must be caught up promptly,
 * not two heartbeat cycles later.
 *
 * In a "late-joiner" election no vote endpoint is promoted to a standby
 * channel (raft_cb_become_leader), so ha_arm_own_term_gate's WAL_NOOP is
 * appended while no standby is registered (repl_broadcast: "no standbys
 * registered ... not replicated").  Both standbys then register through
 * on_standby_connected, which used to send only a ctrlhello: the replication
 * decision loop (ha_dispatch_repl_action_for_standby) runs only on a
 * heartbeat REPLY's state report, so the first action waited for the next
 * heartbeat tick (~3 s), and — one action per state report, config before
 * WAL — the WAL stream carrying the no-op waited for the tick after that
 * (~6 s).  Every client write in that window was NACKed and failed -110.
 *
 * The fix keeps the decision loop and its pacing (one action per state
 * report) and only elicits the state report sooner, through the same
 * non-blocking heartbeat send the periodic tick uses:
 *   1. on_standby_connected sends a heartbeat probe right after registering
 *      the standby (its reply drives the first decision at ~RTT);
 *   2. the standby's ACCEPTED config-push ACK sends ONE more probe (armed
 *      once per registration), so the WAL stream follows the config push as
 *      soon as the standby confirms it, instead of on the next tick.
 *
 * The one-shot arming bounds the ACK-driven probe: a standby whose reported
 * tag stays behind (e.g. activation deferred) cannot drive a
 * probe → push → ACK → probe loop; it falls back to the periodic heartbeat.
 */

#define _GNU_SOURCE
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "ha.h"
#include "netio.h"
#include "repl.h"
#include "request.h"

/* ha.c XCE_TESTING seams */
extern int         on_standby_connected_for_test(endpoint_t *ep);
extern void        raft_cb_become_follower_for_test(ha_state_t *ha, uint64_t term);
extern endpoint_t *standby_ep_ctx_build_for_test(endpoint_t *ep,
                                                  ha_state_t *ha,
                                                  cluster_t  *c,
                                                  int         peer_idx);
extern int         peer_ctrl_msg_handler_for_test(endpoint_t *ep,
                                                  io_packet_t *pkt);

#define LJ_PEER_IDX   1          /* absolute index of the late-joining standby */
#define LJ_TERM       5
#define LJ_COMMITTED  0          /* no-op (lsn 1) not committed yet */
#define LJ_CFG_TAG    0x5EEDULL

static ha_config_t g_lj_cfg = {
    .heartbeat_interval_ms     = 3000,
    .heartbeat_timeout_ms      = 10000,
    .election_timeout_ms       = 2000,
    .probe_connect_timeout_ms  = 500,
    .max_promotion_retries     = 5,
    .election_jitter_max_ms    = 0,
    .election_priority_step_ms = 500,
};

/* statics survive Unity longjmp */
static cluster_t  g_lj_c;
static myconf_t   g_lj_conf;
static endpoint_t g_lj_ep;

/* Heartbeat-send capture (ha_heartbeat_req_hook). */
static int         g_lj_hb_calls;
static endpoint_t *g_lj_hb_ep;
static uint64_t    g_lj_hb_term;
static uint64_t    g_lj_hb_committed;
static uint64_t    g_lj_hb_tag;

static int lj_hb_hook(endpoint_t *ep, uint64_t term, uint64_t committed_lsn,
                      uint64_t committed_config_tag, uint64_t elected_at_ms)
{
    (void)elected_at_ms;
    g_lj_hb_calls++;
    g_lj_hb_ep        = ep;
    g_lj_hb_term      = term;
    g_lj_hb_committed = committed_lsn;
    g_lj_hb_tag       = committed_config_tag;
    return 0;
}

static void lj_reset_capture(void)
{
    g_lj_hb_calls     = 0;
    g_lj_hb_ep        = NULL;
    g_lj_hb_term      = 0;
    g_lj_hb_committed = 0;
    g_lj_hb_tag       = 0;
}

/* A 3-node primary (self = absolute 0) with no registered standby — the
 * state right after a late-joiner election appended its own-term no-op. */
static void lj_setup(ctrl_role_t role)
{
    /* A failed assertion in an earlier test longjmps past lj_teardown; never
     * inherit its hook. */
    ha_heartbeat_req_hook = NULL;
    memset(&g_lj_c, 0, sizeof(g_lj_c));
    memset(&g_lj_conf, 0, sizeof(g_lj_conf));
    memset(&g_lj_ep, 0, sizeof(g_lj_ep));
    g_lj_ep.fd = -1;

    TEST_ASSERT_EQUAL_INT(0, ha_init(&g_lj_c.ha, &g_lj_cfg, 0));
    g_lj_c.ha.n_peers = 3;
    g_lj_c.ha.standby_endpoints = calloc(1, sizeof(zarray_t));
    TEST_ASSERT_NOT_NULL(g_lj_c.ha.standby_endpoints);
    zarray_init(g_lj_c.ha.standby_endpoints, 4, NULL);
    TEST_ASSERT_EQUAL_INT(0, repl_init(&g_lj_c.persist.repl));

    g_lj_conf.peers_cfg.candidates_count = 3;
    for (int i = 0; i < 3; i++) {
        snprintf(g_lj_conf.peers_cfg.candidates[i].name,
                 sizeof(g_lj_conf.peers_cfg.candidates[i].name), "ctrl-%d", i + 1);
        snprintf(g_lj_conf.peers_cfg.candidates[i].hostname,
                 sizeof(g_lj_conf.peers_cfg.candidates[i].hostname), "localhost");
        g_lj_conf.peers_cfg.candidates[i].port = 9991 + i;
    }
    g_lj_c.myconf = &g_lj_conf;

    atomic_store_explicit(&g_lj_c.ha.term, LJ_TERM, memory_order_release);
    atomic_store_explicit(&g_lj_c.persist.committed_lsn, LJ_COMMITTED,
                          memory_order_release);
    atomic_store_explicit(&g_lj_c.ha.committed_config_tag, LJ_CFG_TAG,
                          memory_order_release);
    atomic_store_explicit(&g_lj_c.ha.role, (int32_t)role,
                          memory_order_release);

    TEST_ASSERT_NOT_NULL(standby_ep_ctx_build_for_test(&g_lj_ep, &g_lj_c.ha,
                                                       &g_lj_c, LJ_PEER_IDX));
    lj_reset_capture();
    ha_heartbeat_req_hook = lj_hb_hook;
}

static void lj_teardown(void)
{
    ha_heartbeat_req_hook = NULL;
    repl_remove_standby(&g_lj_c.persist.repl, &g_lj_ep);
    free(g_lj_ep.data);
    g_lj_ep.data = NULL;
    zarray_fini(g_lj_c.ha.standby_endpoints);
    free(g_lj_c.ha.standby_endpoints);
    g_lj_c.ha.standby_endpoints = NULL;
    repl_fini(&g_lj_c.persist.repl);
    ha_fini(&g_lj_c.ha);
}

/* The standby's OP_replConfigPush reply exactly as it travels on the wire
 * (config_repl.c send_config_push_reply): Request{opcode=reply,
 * body.replymsg=RequestReply{data.replconfigpushreplymsg=
 * ReplConfigPushReply{accepted, versionTag}}}.  The handler takes ownership. */
static io_packet_t *lj_config_ack_pkt(int accepted, uint64_t tag)
{
    struct capn                 cs;
    struct capn_segment        *seg;
    struct ReplConfigPushReply  ack_capn;
    ReplConfigPushReply_ptr     ack_ptr;
    struct RequestReply         rr_capn;
    RequestReply_ptr            rr_ptr;
    struct Request              rq_capn;
    Request_ptr                 rq_ptr;
    io_packet_t                *pkt;

    capn_init_malloc(&cs);
    seg = capn_root(&cs).seg;

    memset(&ack_capn, 0, sizeof(ack_capn));
    ack_capn.accepted   = accepted ? 1 : 0;
    ack_capn.versionTag = tag;
    ack_ptr = new_ReplConfigPushReply(seg);
    write_ReplConfigPushReply(&ack_capn, ack_ptr);

    memset(&rr_capn, 0, sizeof(rr_capn));
    rr_capn.seq        = 1;
    rr_capn.opcode     = OPCode_replConfigPush;
    rr_capn.status     = 0;
    rr_capn.data_which = RequestReply_data_replconfigpushreplymsg;
    rr_capn.data.replconfigpushreplymsg = ack_ptr;
    rr_ptr = new_RequestReply(seg);
    write_RequestReply(&rr_capn, rr_ptr);

    memset(&rq_capn, 0, sizeof(rq_capn));
    rq_capn.opcode        = OPCode_reply;
    rq_capn.seq           = 1;
    rq_capn.body_which    = Request_body_replymsg;
    rq_capn.body.replymsg = rr_ptr;
    rq_ptr = new_Request(seg);
    write_Request(&rq_capn, rq_ptr);
    capn_setp(capn_root(&cs), 0, rq_ptr.p);

    pkt = calloc(1, sizeof(*pkt));
    TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "config ACK pkt alloc");
    pkt->sz = capn_size(&cs);
    TEST_ASSERT_TRUE_MESSAGE(alloc_packet(pkt) >= 0, "config ACK pkt data alloc");
    capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0);
    capn_free(&cs);
    return pkt;
}

/* RED: registering a late-joining standby sent only a ctrlhello; the state
 * report that drives its catch-up was not requested until the next periodic
 * heartbeat tick (and last_push_time = now pushed that to the tick after). */
static void test_late_joiner_registration_probes_standby_state(void)
{
    lj_setup(CTRL_PRIMARY);

    on_standby_connected_for_test(&g_lj_ep);

    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_lj_hb_calls,
        "bug-1913: registering a late-joining standby must immediately request "
        "its state report (heartbeat probe) so catch-up of the own-term no-op "
        "starts at ~RTT, not on the next heartbeat tick");
    TEST_ASSERT_EQUAL_PTR(&g_lj_ep, g_lj_hb_ep);
    /* The probe is an ordinary heartbeat: the primary's own term/commit/tag. */
    TEST_ASSERT_EQUAL_UINT64(LJ_TERM, g_lj_hb_term);
    TEST_ASSERT_EQUAL_UINT64(LJ_COMMITTED, g_lj_hb_committed);
    TEST_ASSERT_EQUAL_UINT64(LJ_CFG_TAG, g_lj_hb_tag);

    lj_teardown();
    printf("test_late_joiner_registration_probes_standby_state: PASS\n");
}

/* RED: after the config push the decision loop's next action (WAL stream of
 * the no-op) waited for the next heartbeat tick.  The standby's accepted
 * config ACK must re-request its state ONCE; a second ACK must not (bounded:
 * no probe/push/ACK loop). */
static void test_config_ack_after_registration_reprobes_once(void)
{
    lj_setup(CTRL_PRIMARY);
    on_standby_connected_for_test(&g_lj_ep);
    lj_reset_capture();

    peer_ctrl_msg_handler_for_test(&g_lj_ep, lj_config_ack_pkt(1, LJ_CFG_TAG));
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_lj_hb_calls,
        "bug-1913: the late joiner's accepted config-push ACK must request its "
        "state report so the WAL stream follows without waiting a heartbeat "
        "cycle");
    TEST_ASSERT_EQUAL_PTR(&g_lj_ep, g_lj_hb_ep);

    peer_ctrl_msg_handler_for_test(&g_lj_ep, lj_config_ack_pkt(1, LJ_CFG_TAG));
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_lj_hb_calls,
        "bug-1913: the ACK-driven probe is one-shot per registration — a "
        "repeated config ACK must not re-probe (no probe/push/ACK loop)");

    lj_teardown();
    printf("test_config_ack_after_registration_reprobes_once: PASS\n");
}

/* A rejected config push must not trigger the probe (and must not consume
 * the arming: a later accepted ACK still re-probes). */
static void test_rejected_config_ack_does_not_probe(void)
{
    lj_setup(CTRL_PRIMARY);
    on_standby_connected_for_test(&g_lj_ep);
    lj_reset_capture();

    peer_ctrl_msg_handler_for_test(&g_lj_ep, lj_config_ack_pkt(0, LJ_CFG_TAG));
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_lj_hb_calls,
        "a rejected config push must not trigger the catch-up probe");

    peer_ctrl_msg_handler_for_test(&g_lj_ep, lj_config_ack_pkt(1, LJ_CFG_TAG));
    TEST_ASSERT_EQUAL_INT(1, g_lj_hb_calls);

    lj_teardown();
    printf("test_rejected_config_ack_does_not_probe: PASS\n");
}

/* Steady state is unchanged: a config ACK from a standby that was NOT just
 * registered (e.g. an ordinary config upgrade) sends no extra heartbeat. */
static void test_config_ack_without_registration_does_not_probe(void)
{
    lj_setup(CTRL_PRIMARY);
    TEST_ASSERT_TRUE(repl_add_standby(&g_lj_c.persist.repl, &g_lj_ep,
                                      "localhost", 9992) >= 0);

    peer_ctrl_msg_handler_for_test(&g_lj_ep, lj_config_ack_pkt(1, LJ_CFG_TAG));
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_lj_hb_calls,
        "a config ACK outside a fresh registration must not send a probe");

    lj_teardown();
    printf("test_config_ack_without_registration_does_not_probe: PASS\n");
}

/* A node that is no longer primary when the outbound connect completes must
 * not send a heartbeat (only a primary heartbeats its standbys). */
static void test_registration_probe_requires_primary_role(void)
{
    lj_setup(CTRL_STANDBY);

    on_standby_connected_for_test(&g_lj_ep);
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_lj_hb_calls,
        "a demoted node must not heartbeat a freshly connected peer");

    lj_teardown();
    printf("test_registration_probe_requires_primary_role: PASS\n");
}

/* An arm belongs to one registration under one leadership: stepping down
 * must clear it, so it cannot fire an extra probe on a later, unrelated
 * config-push ACK after this node leads again. */
static void test_step_down_clears_catchup_arm(void)
{
    lj_setup(CTRL_PRIMARY);
    on_standby_connected_for_test(&g_lj_ep);
    TEST_ASSERT_EQUAL_INT(1, atomic_load_explicit(
        &g_lj_c.ha.catchup_kick_armed[LJ_PEER_IDX], memory_order_acquire));

    raft_cb_become_follower_for_test(&g_lj_c.ha, LJ_TERM + 1);
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load_explicit(
        &g_lj_c.ha.catchup_kick_armed[LJ_PEER_IDX], memory_order_acquire),
        "bug-1913: a role/term change must clear the catch-up arm");

    lj_teardown();
    printf("test_step_down_clears_catchup_arm: PASS\n");
}

void test_late_joiner_catchup_kick(void)
{
    UnitySetTestFile(__FILE__);
    printf("\n>> late-joiner catch-up kick tests (bug-1913)\n\n");

    RUN_TEST(test_late_joiner_registration_probes_standby_state);
    RUN_TEST(test_config_ack_after_registration_reprobes_once);
    RUN_TEST(test_rejected_config_ack_does_not_probe);
    RUN_TEST(test_config_ack_without_registration_does_not_probe);
    RUN_TEST(test_registration_probe_requires_primary_role);
    RUN_TEST(test_step_down_clears_catchup_arm);
    ha_heartbeat_req_hook = NULL;
}