diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 0b6af62b..902aedc5 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -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; diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index c5beefd6..1d85f5f1 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -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) { diff --git a/src/controller/ha.c b/src/controller/ha.c index 47e206fd..ccf3bbca 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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; diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index c86518be..db3f0430 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -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);