diff --git a/src/controller/ha.c b/src/controller/ha.c index 56056d7a..c9aac532 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -541,7 +541,20 @@ static void ha_arm_own_term_gate(cluster_t *c); int ha_heartbeat_is_fresh(ha_state_t *ha); /* exposed for PreVote denial-reason logging (cluster.c) */ static int primary_heartbeat_fn(void *arg); static void on_standby_connected(endpoint_t *ep); + +/* bug-1913: drop every armed catch-up probe. An arm belongs to one standby + * registration under one leadership; a role or term change, or the + * standby's disconnect, ends that registration, so a stale arm must not + * fire an extra probe on some later, unrelated config-push ACK. */ +static void ha_clear_catchup_kicks(ha_state_t *ha) +{ + for (int i = 0; i < MAX_HA_CANDIDATES; i++) + atomic_store_explicit(&ha->catchup_kick_armed[i], 0, + memory_order_release); +} static int standby_ep_err_handler_primary(void *ep_raw); +static int ha_send_catchup_probe(cluster_t *c, endpoint_t *ep, int peer_idx, + const char *why); /* ── ha-elect-ctx-lifecycle task 2.1: election carrier reference-counted * lifetime ────────────────────────────────────────────────────────────── @@ -1631,6 +1644,7 @@ static void raft_cb_become_leader(void *ctx, uint64_t term) /* Sync ha->term from raft term (raft owns term increments) */ atomic_store_explicit(&ha->term, term, memory_order_release); + ha_clear_catchup_kicks(ha); /* bug-1913: arms are per leadership */ /* The cluster committed high-water mark for this promotion — derived from * the voter replies (NOT a stale ctrlhello). The task-4.3 pre-promotion @@ -2180,6 +2194,7 @@ static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx) /* Sync ha->term */ atomic_store_explicit(&ha->term, term, memory_order_release); + ha_clear_catchup_kicks(ha); /* bug-1913: arms are per leadership */ int is_first_boot = !ha->phase2_done; @@ -7031,6 +7046,28 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx, repl_on_config_ack(&c->persist.repl, ep, acked_tag); TRACE("repl_ack_decode_and_dispatch: config ACK from standby" " tag=%016" PRIx64 " accepted=1", acked_tag); + + /* bug-1913: completion-driven catch-up for a freshly + * registered standby. The decision loop takes one action + * per state report (config before WAL); the next report used + * to come only from the next heartbeat tick, so the WAL + * stream carrying the own-term no-op trailed the config push + * by a full heartbeat cycle. Now that the standby confirmed + * the config, request its state report once. One-shot per + * registration (catchup_kick_armed), so a standby whose + * reported tag stays behind cannot loop probe → push → ACK; + * it falls back to the periodic heartbeat. Runs on the + * peer-control vtask; the probe is a non-blocking enqueue. */ + int kick_idx = ha_repl_ack_abs_peer_idx( + ha, /*ep_is_vote*/ vctx->ctx != NULL, vctx->peer_idx); + if (kick_idx >= 0 && kick_idx < MAX_HA_CANDIDATES) { + int armed = 1; + if (atomic_compare_exchange_strong_explicit( + &ha->catchup_kick_armed[kick_idx], &armed, 0, + memory_order_acq_rel, memory_order_acquire)) + (void)ha_send_catchup_probe(c, ep, kick_idx, + "config push acked"); + } } else if (!accepted) { WARN("repl_ack_decode_and_dispatch: config push rejected by" " standby (tag=%016" PRIx64 ") — ignoring", acked_tag); @@ -9666,6 +9703,86 @@ static void on_standby_connected(endpoint_t *ep) /* Set state-aware dispatcher for subsequent messages on this endpoint */ ep->msg_handler = peer_ctrl_msg_handler; + + /* bug-1913: request the new standby's state report NOW instead of on the + * next heartbeat tick. A standby registering here (late-joiner election: + * no vote ep was promoted; or a reconnect) missed every entry broadcast + * while it was unregistered — in particular the own-term WAL_NOOP, which + * repl_broadcast dropped ("no standbys registered") and repl_add_standby + * does not resend. Only the decision loop (heartbeat reply → state + * report → ha_dispatch_repl_action_for_standby) catches it up, and the + * first report used to wait for the next tick (last_push_time = now above + * can push it to the tick after), keeping committed_lsn below the no-op + * — and every client write NACKed — for up to two heartbeat cycles. + * Same non-blocking heartbeat send the tick uses (enqueue only — allowed + * on this iomgr callback, like the ctrlhello above); sent after + * msg_handler is set so the reply is routed to the decision loop. Also + * arm the one-shot ACK-driven probe (repl_ack_decode_and_dispatch's + * config-ACK branch) so the WAL stream follows a config push at ~RTT. */ + if (sctx->peer_idx >= 0 && sctx->peer_idx < MAX_HA_CANDIDATES) + atomic_store_explicit(&ha->catchup_kick_armed[sctx->peer_idx], 1, + memory_order_release); + (void)ha_send_catchup_probe(c, ep, sctx->peer_idx, "standby registered"); +} + +/* + * ha_send_catchup_probe — bug-1913: send one out-of-cycle heartbeat to a + * standby so its reply (RequestReply.stateReport) drives the replication + * decision loop (repl_ack_decode_and_dispatch → ha_dispatch_repl_action_for_ + * standby) now rather than on the next periodic tick. + * + * It is exactly the heartbeat primary_heartbeat_fn sends — the primary's own + * current term / committed_lsn / committed_config_tag / elected_at_ms, read + * atomically — so the standby handles it like any heartbeat; no new message, + * opcode or decision path. Non-blocking: xce_ha_heartbeat_req only encodes + * and enqueues on the endpoint's send queue (safe on iomgr and vtask + * threads). Only a CTRL_PRIMARY heartbeats its standbys; a node demoted + * before this runs sends nothing. Returns the send rc, -EPERM when not + * primary, -EINVAL on bad args. Failure is logged and otherwise harmless: + * the periodic heartbeat remains the fallback driver. + */ +static int ha_send_catchup_probe(cluster_t *c, endpoint_t *ep, int peer_idx, + const char *why) +{ + if (!c || !ep) + return -EINVAL; + + ha_state_t *ha = &c->ha; + if ((ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire) + != CTRL_PRIMARY) + return -EPERM; + + uint64_t term = atomic_load_explicit(&ha->term, memory_order_acquire); + uint64_t committed_lsn = + atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire); + uint64_t committed_tag = + atomic_load_explicit(&ha->committed_config_tag, memory_order_acquire); + uint64_t elected_at_ms = + atomic_load_explicit(&ha->last_known_elected_at_ms, + memory_order_acquire); + + int rc; +#ifdef XCE_TESTING + if (ha_heartbeat_req_hook) + rc = ha_heartbeat_req_hook(ep, term, committed_lsn, committed_tag, + elected_at_ms); + else if (ha_nop_req_hook) + rc = ha_nop_req_hook(ep); + else + rc = xce_ha_heartbeat_req(ep, term, committed_lsn, committed_tag, + elected_at_ms); +#else + rc = xce_ha_heartbeat_req(ep, term, committed_lsn, committed_tag, + elected_at_ms); +#endif + if (rc < 0) + WARN("HA: catch-up probe (%s) to standby peer[%d] failed: %d — " + "periodic heartbeat remains the fallback", why, peer_idx, rc); + else + INFO("HA: catch-up probe (%s) sent to standby peer[%d] " + "(term=%" PRIu64 " committed_lsn=%" PRIu64 ")", + why, peer_idx, term, committed_lsn); + return rc; } /* @@ -9710,6 +9827,12 @@ static int standby_ep_err_handler_primary(void *ep_raw) if (c) repl_remove_standby(&c->persist.repl, ep); + /* bug-1913: this registration is over; its catch-up arm must not fire + * on a later, unrelated config-push ACK. */ + if (sctx->peer_idx >= 0 && sctx->peer_idx < MAX_HA_CANDIDATES) + atomic_store_explicit(&ha->catchup_kick_armed[sctx->peer_idx], 0, + memory_order_release); + /* bug-1839: do not free sctx / null ep->data synchronously here — a * peer-control vtask for this same ep may already be queued or in * flight (peer_ctrl_msg_handler_inline reads ep->data), same hazard diff --git a/src/controller/ha.h b/src/controller/ha.h index c1b3ca63..0ce71ca7 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -289,6 +289,25 @@ typedef struct { * Slot identity (the repl registry slot * index) is stable across schedule/run. * 0 at init (cleared by ha_init memset). */ + _Atomic int catchup_kick_armed[MAX_HA_CANDIDATES]; /* bug-1913: per-standby + * (absolute peer index) one-shot arm for + * the catch-up probe. Set by + * on_standby_connected when a standby + * registers; consumed (1->0 CAS) by that + * standby's first ACCEPTED config-push + * ACK, which then sends one heartbeat + * probe so the decision loop's next + * action (WAL stream) follows the config + * push at ~RTT instead of on the next + * heartbeat tick. One-shot so a standby + * whose reported tag stays behind cannot + * drive a probe/push/ACK loop. + * Cleared on the standby's disconnect + * (standby_ep_err_handler_primary) and + * on every become_leader/become_follower, + * so an arm never outlives its + * registration or leadership. + * 0 at init (cleared by ha_init memset). */ /* ── New field added by ha-convergence-recovery spec (Task 5.x) ──────── */ _Atomic int quorum_lost; /* Quorum-transition edge detector (Req * 8.1-8.3): 0 = quorum held (or unknown diff --git a/src/controller/meson.build b/src/controller/meson.build index bf8f9e41..6f8ab48d 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -177,6 +177,7 @@ controller_unit_test_src = [ 'testsuite/test_jobdispatch_no_double_notify.c', 'testsuite/test_worker_resync.c', 'testsuite/test_standby_state_peer_idx.c', + 'testsuite/test_late_joiner_catchup_kick.c', 'testsuite/test_upgrade_snapshot_restore.c', 'testsuite/test_upgrade_snapshot_apply.c', 'testsuite/test_bug857_regression.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 908e8f5f..b18cbee1 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -113,6 +113,7 @@ extern void test_dispatch_fail_retry(void); extern void test_jobdispatch_no_double_notify(void); extern void test_worker_resync(void); extern void test_standby_state_peer_idx(void); +extern void test_late_joiner_catchup_kick(void); extern void test_jobsubmit_validate(void); void setUp(void) { return; } @@ -304,6 +305,7 @@ int main(int argc, char* argv[]) TEST(test_jobdispatch_no_double_notify); TEST(test_worker_resync); TEST(test_standby_state_peer_idx); + TEST(test_late_joiner_catchup_kick); TEST(test_jobsubmit_validate); return UNITY_END(); diff --git a/src/controller/testsuite/test_late_joiner_catchup_kick.c b/src/controller/testsuite/test_late_joiner_catchup_kick.c new file mode 100644 index 00000000..f86cce12 --- /dev/null +++ b/src/controller/testsuite/test_late_joiner_catchup_kick.c @@ -0,0 +1,340 @@ +/* + * 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 +#include +#include +#include +#include + +#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; +}