fix(controller): probe a late-joining standby at once so its catch-up starts without waiting a heartbeat (bug-1913)
After an election where the vote connections are not promoted, both standbys register later via on_standby_connected; the own-term no-op was appended before any standby existed, and catch-up waited for heartbeat- driven state reports (config push one 3 s cycle, WAL stream the next), so the no-op committed ~6 s late and every client write in that window failed -110. ha_send_catchup_probe() sends the ordinary heartbeat request (same fields as primary_heartbeat_fn; encode + enqueue only; primary only) at registration and once after the standby's first accepted config-push ACK, so the decision loop's next action follows at ~RTT. The decision loop, its one-action-per-report pacing, commit/majority rules and the dispatch gate are unchanged; WAL streaming still runs only on the vtask. The one-shot arm is cleared on disconnect and on every become_leader / become_follower. Tests: test_late_joiner_catchup_kick.c (6). Controller suite 1285/0. Live: catch-up starts 0-6 ms after a late registration (was up to ~6 s); /xce-test-ha sequential 20 rounds, chaotic 6 rounds and sequential 8 rounds under 40-spinner load all PASS with zero job loss. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
6d9ef3d024
commit
2aefa939db
GPG key ID:
80533536B0EB651C
5 changed files with 485 additions and 0 deletions
|
|
@ -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) */
|
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 int primary_heartbeat_fn(void *arg);
|
||||||
static void on_standby_connected(endpoint_t *ep);
|
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 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
|
/* ── ha-elect-ctx-lifecycle task 2.1: election carrier reference-counted
|
||||||
* lifetime ──────────────────────────────────────────────────────────────
|
* 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) */
|
/* Sync ha->term from raft term (raft owns term increments) */
|
||||||
atomic_store_explicit(&ha->term, term, memory_order_release);
|
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 cluster committed high-water mark for this promotion — derived from
|
||||||
* the voter replies (NOT a stale ctrlhello). The task-4.3 pre-promotion
|
* 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 */
|
/* Sync ha->term */
|
||||||
atomic_store_explicit(&ha->term, term, memory_order_release);
|
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;
|
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);
|
repl_on_config_ack(&c->persist.repl, ep, acked_tag);
|
||||||
TRACE("repl_ack_decode_and_dispatch: config ACK from standby"
|
TRACE("repl_ack_decode_and_dispatch: config ACK from standby"
|
||||||
" tag=%016" PRIx64 " accepted=1", acked_tag);
|
" 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) {
|
} else if (!accepted) {
|
||||||
WARN("repl_ack_decode_and_dispatch: config push rejected by"
|
WARN("repl_ack_decode_and_dispatch: config push rejected by"
|
||||||
" standby (tag=%016" PRIx64 ") — ignoring", acked_tag);
|
" 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 */
|
/* Set state-aware dispatcher for subsequent messages on this endpoint */
|
||||||
ep->msg_handler = peer_ctrl_msg_handler;
|
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)
|
if (c)
|
||||||
repl_remove_standby(&c->persist.repl, ep);
|
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
|
/* 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
|
* peer-control vtask for this same ep may already be queued or in
|
||||||
* flight (peer_ctrl_msg_handler_inline reads ep->data), same hazard
|
* flight (peer_ctrl_msg_handler_inline reads ep->data), same hazard
|
||||||
|
|
|
||||||
|
|
@ -289,6 +289,25 @@ typedef struct {
|
||||||
* Slot identity (the repl registry slot
|
* Slot identity (the repl registry slot
|
||||||
* index) is stable across schedule/run.
|
* index) is stable across schedule/run.
|
||||||
* 0 at init (cleared by ha_init memset). */
|
* 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) ──────── */
|
/* ── New field added by ha-convergence-recovery spec (Task 5.x) ──────── */
|
||||||
_Atomic int quorum_lost; /* Quorum-transition edge detector (Req
|
_Atomic int quorum_lost; /* Quorum-transition edge detector (Req
|
||||||
* 8.1-8.3): 0 = quorum held (or unknown
|
* 8.1-8.3): 0 = quorum held (or unknown
|
||||||
|
|
|
||||||
|
|
@ -177,6 +177,7 @@ controller_unit_test_src = [
|
||||||
'testsuite/test_jobdispatch_no_double_notify.c',
|
'testsuite/test_jobdispatch_no_double_notify.c',
|
||||||
'testsuite/test_worker_resync.c',
|
'testsuite/test_worker_resync.c',
|
||||||
'testsuite/test_standby_state_peer_idx.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_restore.c',
|
||||||
'testsuite/test_upgrade_snapshot_apply.c',
|
'testsuite/test_upgrade_snapshot_apply.c',
|
||||||
'testsuite/test_bug857_regression.c',
|
'testsuite/test_bug857_regression.c',
|
||||||
|
|
|
||||||
|
|
@ -113,6 +113,7 @@ extern void test_dispatch_fail_retry(void);
|
||||||
extern void test_jobdispatch_no_double_notify(void);
|
extern void test_jobdispatch_no_double_notify(void);
|
||||||
extern void test_worker_resync(void);
|
extern void test_worker_resync(void);
|
||||||
extern void test_standby_state_peer_idx(void);
|
extern void test_standby_state_peer_idx(void);
|
||||||
|
extern void test_late_joiner_catchup_kick(void);
|
||||||
extern void test_jobsubmit_validate(void);
|
extern void test_jobsubmit_validate(void);
|
||||||
|
|
||||||
void setUp(void) { return; }
|
void setUp(void) { return; }
|
||||||
|
|
@ -304,6 +305,7 @@ int main(int argc, char* argv[])
|
||||||
TEST(test_jobdispatch_no_double_notify);
|
TEST(test_jobdispatch_no_double_notify);
|
||||||
TEST(test_worker_resync);
|
TEST(test_worker_resync);
|
||||||
TEST(test_standby_state_peer_idx);
|
TEST(test_standby_state_peer_idx);
|
||||||
|
TEST(test_late_joiner_catchup_kick);
|
||||||
TEST(test_jobsubmit_validate);
|
TEST(test_jobsubmit_validate);
|
||||||
|
|
||||||
return UNITY_END();
|
return UNITY_END();
|
||||||
|
|
|
||||||
340
src/controller/testsuite/test_late_joiner_catchup_kick.c
Normal file
340
src/controller/testsuite/test_late_joiner_catchup_kick.c
Normal file
|
|
@ -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 <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;
|
||||||
|
}
|
||||||
Loading…
Add table
Add a link
Reference in a new issue