fix(controller): scan standby_endpoints by capacity, not count (bug-1474)
Root cause of the "spurious duplicate reconnect" (angle a): zarray_del()
NULLs the slot and decrements count without compacting, so a live entry
can sit at any index < caps. Every standby_endpoints loop bounded by
`count` missed it. Live evidence (--config-upgrade-wal-trim soak,
2026-09-26): ctrl-2 won, promoted vote eps [ctrl-1 (dead), ctrl-3]; the
dead one's removal left [NULL, ctrl-3] with count 1, so the primary
- stopped heartbeating ctrl-3 (primary_heartbeat_fn's send loop),
- re-dialed it ("peer[2] not connected -- initiating outbound connect"
9s later, for a peer that never dropped) and registered a second,
"late-joiner" channel that burned a MAX_STANDBYS repl slot,
- could not evict the stale channel (on_standby_connected's dedup scan),
- and skipped it in config pushes and the ctrlhello broadcast, where a
hole below count also handed endpoint_ref() a NULL.
The re-dial landed in the freed slot 0, which is what made ctrl-3 visible
again -- the "spurious" reconnect was masking the lost heartbeats.
All index scans of standby_endpoints (ha.c heartbeat, sweep, dedup, two
removals; cluster.c removal and ctrlhello; config_repl.c push collection)
now run to `caps` and skip NULL slots; the push collects exactly the
non-NULL entries. `count` stays where it means "how many connected".
New test_reconnect_dedups_channel_behind_deleted_slot (RED before: the
stale channel behind the hole was never evicted). Controller suite 1241/0.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
255b0faad1
commit
c3a17b19ab
4 changed files with 67 additions and 9 deletions
|
|
@ -88,7 +88,7 @@ static void handle_endpoint_error(endpoint_t* ep)
|
|||
if (c->ha.standby_endpoints) {
|
||||
pthread_mutex_lock(&c->ha.lock);
|
||||
for (int idx = 0;
|
||||
idx < c->ha.standby_endpoints->count; idx++) {
|
||||
idx < c->ha.standby_endpoints->caps; idx++) {
|
||||
endpoint_t *se = (endpoint_t *)zarray_get(
|
||||
c->ha.standby_endpoints, idx);
|
||||
if (se == ep) {
|
||||
|
|
@ -2424,7 +2424,8 @@ void ha_send_ctrlhello_to_all_standbys(cluster_t *c)
|
|||
char *my_name = c->myconf ? c->myconf->name : "";
|
||||
|
||||
pthread_mutex_lock(&ha->lock);
|
||||
int n = ha->standby_endpoints->count;
|
||||
/* bug-1474: zarray_del leaves holes, so scan caps, not count. */
|
||||
int n = ha->standby_endpoints->caps;
|
||||
for (int i = 0; i < n; i++) {
|
||||
endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
|
||||
if (!ep) continue;
|
||||
|
|
|
|||
|
|
@ -1558,8 +1558,16 @@ int push_config_transport_only(cluster_t *c, const uint8_t *bytes,
|
|||
" — skipping transport push", n);
|
||||
return 0;
|
||||
}
|
||||
for (int i = 0; i < n; i++) {
|
||||
ctxs[i].ep = (endpoint_t *)zarray_get(c->ha.standby_endpoints, i);
|
||||
/* bug-1474: zarray_del leaves holes without compacting, so live entries
|
||||
* can sit at any index < caps -- collect exactly the non-NULL ones (n,
|
||||
* the count, is how many there are). */
|
||||
int filled = 0;
|
||||
for (int k = 0; k < c->ha.standby_endpoints->caps && filled < n; k++) {
|
||||
endpoint_t *sep = (endpoint_t *)zarray_get(c->ha.standby_endpoints, k);
|
||||
if (sep == NULL)
|
||||
continue;
|
||||
int i = filled++;
|
||||
ctxs[i].ep = sep;
|
||||
ctxs[i].from = c->myhost;
|
||||
ctxs[i].new_tag = new_tag;
|
||||
ctxs[i].bytes = bytes;
|
||||
|
|
@ -1576,6 +1584,9 @@ int push_config_transport_only(cluster_t *c, const uint8_t *bytes,
|
|||
endpoint_ref(ctxs[i].ep);
|
||||
}
|
||||
pthread_mutex_unlock(&c->ha.lock);
|
||||
n = filled;
|
||||
if (n_out)
|
||||
*n_out = n;
|
||||
|
||||
pthread_t *threads = malloc((size_t)n * sizeof(pthread_t));
|
||||
if (!threads) {
|
||||
|
|
|
|||
|
|
@ -7775,7 +7775,7 @@ static void vote_ep_handle_promoted_disconnect(ha_state_t *ha, endpoint_t *ep,
|
|||
/* Promoted connection dropped — remove from standby_endpoints */
|
||||
if (ha->standby_endpoints) {
|
||||
pthread_mutex_lock(&ha->lock);
|
||||
for (int i = 0; i < ha->standby_endpoints->count; i++) {
|
||||
for (int i = 0; i < ha->standby_endpoints->caps; i++) { /* bug-1474 */
|
||||
if (zarray_get(ha->standby_endpoints, i) == ep) {
|
||||
zarray_del(ha->standby_endpoints, i);
|
||||
break;
|
||||
|
|
@ -9464,7 +9464,7 @@ static void on_standby_connected(endpoint_t *ep)
|
|||
int votes_idx_for_new = votes_idx_for_peer(ha, sctx->peer_idx);
|
||||
pthread_mutex_lock(&ha->lock);
|
||||
int already = 0;
|
||||
for (int i = 0; i < ha->standby_endpoints->count; i++) {
|
||||
for (int i = 0; i < ha->standby_endpoints->caps; i++) { /* bug-1474 */
|
||||
endpoint_t *cur = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
|
||||
if (cur == ep) { already = 1; continue; }
|
||||
/* Different ep, same peer → stale duplicate to evict. */
|
||||
|
|
@ -9599,7 +9599,7 @@ static int standby_ep_err_handler_primary(void *ep_raw)
|
|||
|
||||
if (ha->standby_endpoints) {
|
||||
pthread_mutex_lock(&ha->lock);
|
||||
for (int i = 0; i < ha->standby_endpoints->count; i++) {
|
||||
for (int i = 0; i < ha->standby_endpoints->caps; i++) { /* bug-1474 */
|
||||
if (zarray_get(ha->standby_endpoints, i) == ep) {
|
||||
zarray_del(ha->standby_endpoints, i);
|
||||
break;
|
||||
|
|
@ -9959,7 +9959,8 @@ static int primary_heartbeat_fn(void *arg)
|
|||
int n_evicted = 0;
|
||||
|
||||
pthread_mutex_lock(&ha->lock);
|
||||
int n = ha->standby_endpoints->count;
|
||||
/* bug-1474: zarray_del leaves holes, so scan caps, not count. */
|
||||
int n = ha->standby_endpoints->caps;
|
||||
for (int i = 0; i < n; i++) {
|
||||
endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
|
||||
if (!ep) continue;
|
||||
|
|
@ -10045,7 +10046,7 @@ static int primary_heartbeat_fn(void *arg)
|
|||
int found = 0;
|
||||
if (ha->standby_endpoints) {
|
||||
pthread_mutex_lock(&ha->lock);
|
||||
for (int j = 0; j < ha->standby_endpoints->count; j++) {
|
||||
for (int j = 0; j < ha->standby_endpoints->caps; j++) { /* bug-1474 */
|
||||
endpoint_t *sep = (endpoint_t *)zarray_get(
|
||||
ha->standby_endpoints, j);
|
||||
if (!sep || !sep->data) continue;
|
||||
|
|
|
|||
|
|
@ -1563,6 +1563,50 @@ static void test_reconnect_preserves_other_peer_channel(void)
|
|||
printf("test_reconnect_preserves_other_peer_channel: PASS\n");
|
||||
}
|
||||
|
||||
/* bug-1474 (angle a, root cause): zarray_del() leaves a NULL hole and
|
||||
* decrements count WITHOUT compacting, so a live entry can sit at an index
|
||||
* >= count. Every standby_endpoints loop bounded by `count` skipped it: after
|
||||
* peer 1's disconnect removed index 0, peer 2's channel at index 1 became
|
||||
* invisible -- the primary stopped heartbeating it, the reconnect sweep
|
||||
* re-dialed it (the "spurious duplicate reconnect"), and this dedup could not
|
||||
* evict the stale channel, leaving two live channels for one peer. */
|
||||
static void test_reconnect_dedups_channel_behind_deleted_slot(void)
|
||||
{
|
||||
cluster_t c;
|
||||
myconf_t conf;
|
||||
own_setup_primary(&c, &conf);
|
||||
|
||||
atomic_store(&ha_standby_evicted_count_for_test, 0);
|
||||
|
||||
endpoint_t *peer1 = own_make_standby_ep(&c.ha, &c, 1);
|
||||
zarray_store(c.ha.standby_endpoints, peer1); /* index 0 */
|
||||
endpoint_t *peer2_old = own_make_standby_ep(&c.ha, &c, 2);
|
||||
zarray_store(c.ha.standby_endpoints, peer2_old); /* index 1 */
|
||||
|
||||
/* peer 1 disconnects: its slot becomes a hole, count 2 -> 1 */
|
||||
for (int i = 0; i < c.ha.standby_endpoints->caps; i++) {
|
||||
if (zarray_get(c.ha.standby_endpoints, i) == peer1) {
|
||||
zarray_del(c.ha.standby_endpoints, i);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
endpoint_t *peer2_new = own_make_standby_ep(&c.ha, &c, 2);
|
||||
on_standby_connected_for_test(peer2_new);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
|
||||
atomic_load(&ha_standby_evicted_count_for_test),
|
||||
"bug-1474: the stale channel for peer 2 (behind the deleted slot) "
|
||||
"must be found and evicted");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.ha.standby_endpoints->count,
|
||||
"bug-1474: exactly one live channel per peer");
|
||||
|
||||
standby_ep_err_handler_primary_for_test(peer1);
|
||||
standby_ep_err_handler_primary_for_test(peer2_new);
|
||||
|
||||
fini_ha_primary_with_standby_eps(&c.ha);
|
||||
}
|
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════════
|
||||
* raft-drive-async Task 3.1 — PeerCtrlDispatch
|
||||
*
|
||||
|
|
@ -2419,6 +2463,7 @@ void test_ha_liveness(void)
|
|||
RUN_TEST(test_reconnect_resolves_dual_channel_same_peer);
|
||||
RUN_TEST(test_reconnect_resolves_dual_channel_promoted_vote_ep);
|
||||
RUN_TEST(test_reconnect_preserves_other_peer_channel);
|
||||
RUN_TEST(test_reconnect_dedups_channel_behind_deleted_slot);
|
||||
/* raft-drive-async Task 2.3 — reconnect-offload INTEGRATION (Req 2.1, 2.2, 2.5) */
|
||||
RUN_TEST(test_reconnect_offload_sustains_survivor_heartbeats);
|
||||
RUN_TEST(test_reconnect_inline_stalls_survivor_heartbeats_RED);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue