fix(controller): record standby state under its absolute peer index (bug-1891)

A promoted vote endpoint carries a peer index relative to the votes
array (which leaves out the primary); a late-joiner's heartbeat endpoint
carries the absolute index. repl_ack_decode_and_dispatch recorded state
reports under the raw index, so with ctrl-1 primary, ctrl-3's vote
endpoint (relative 1) and a rejoined ctrl-2's heartbeat endpoint
(absolute 1) shared standby_reported_*[1]. Each handler then acted on
the other's position over its own endpoint: the snapshot meant for the
empty, rejoining ctrl-2 went to the caught-up ctrl-3, and ctrl-2 never
caught up.

Resolve the absolute index with ha_repl_ack_abs_peer_idx (as the
WAL-ack branch already did) for the state report and for
ha_walack_on_ack, and size catchup_inflight by MAX_HA_CANDIDATES so
every absolute index fits.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

src/controller/config_repl.c

@@ -2929,7 +2929,7 @@ static void catchup_lane_fn(void *arg)
 clear_and_done:
     /* Req 5.2: release the single-flight slot so the next lagging ACK schedules
      * again.  release order so the decision path's acquire load observes it. */
-    if (peer_idx >= 0 && peer_idx < MAX_STANDBYS)
+    if (peer_idx >= 0 && peer_idx < MAX_HA_CANDIDATES)
         atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
                               memory_order_release);
     free(ctx);
@@ -2990,7 +2990,7 @@ int ha_catchup_continuation_fn(void *arg)
          * but whose key the lane has not yet removed.  Release the guard so
          * the next WAL ack retries cleanly (mirrors ha_connect_to_peer's
          * -EEXIST handling in ha.c). */
-        if (peer_idx >= 0 && peer_idx < MAX_STANDBYS)
+        if (peer_idx >= 0 && peer_idx < MAX_HA_CANDIDATES)
             atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
                                   memory_order_release);
         free(ctx);
@@ -3004,7 +3004,7 @@ int ha_catchup_continuation_fn(void *arg)
          * ack, unlike the one-shot demotion-convergence task). */
         WARN("HA catch-up continuation: slowlane_submit failed for "
              "standby[%d]: %d — retry next ack", peer_idx, srv);
-        if (peer_idx >= 0 && peer_idx < MAX_STANDBYS)
+        if (peer_idx >= 0 && peer_idx < MAX_HA_CANDIDATES)
             atomic_store_explicit(&c->ha.catchup_inflight[peer_idx], 0,
                                   memory_order_release);
         free(ctx);
@@ -3061,7 +3061,7 @@ int ha_walack_on_ack(cluster_t *c, void *ep, int peer_idx, uint64_t acked_lsn)
 {
     if (c == NULL || ep == NULL)
         return -EINVAL;
-    if (peer_idx < 0 || peer_idx >= MAX_STANDBYS)
+    if (peer_idx < 0 || peer_idx >= MAX_HA_CANDIDATES)
         return -EINVAL;
 
     uint64_t target = catchup_derive_target(c);

src/controller/ha.c

@@ -6870,8 +6870,14 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx,
             (void)raft_on_higher_term_observed(&ha->raft, sr.term);
             pthread_mutex_unlock(&ha->raft_lock);
 
-            /* peer_idx from standby_ep_ctx_t (layout-compatible with vote_ep_ctx_t) */
+            /* bug-1891: the ABSOLUTE peer index. A promoted vote ep carries a
-            int      peer_idx      = vctx->peer_idx;
+             * relative votes index (see ha_repl_ack_abs_peer_idx); recording
+             * its report under the raw index shared a slot with another
+             * standby's heartbeat ep, so one handler acted on the other's
+             * position -- e.g. sent a rejoining standby's snapshot to a peer
+             * that was already caught up. */
+            int      peer_idx      = ha_repl_ack_abs_peer_idx(
+                ha, /*ep_is_vote*/ vctx->ctx != NULL, vctx->peer_idx);
             uint64_t committed_lsn = 0;
             uint64_t committed_tag = 0;
             uint64_t last_written  = 0; /* latest WAL LSN — may be > committed */
@@ -6961,7 +6967,7 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx,
                  * decision the unit test drives via ha_walack_continuation_for_test.
                  * Scheduling is O(1) on iomgr; all WAL file I/O runs in the
                  * deferred continuation task. */
-                ha_walack_on_ack(c, ep, vctx->peer_idx, acked_lsn);
+                ha_walack_on_ack(c, ep, ack_abs_idx, acked_lsn); /* bug-1891 */
             } else if (!success) {
                 /* bug-1421: a WAL NACK can also carry a higher term (the
                  * standby rejected this entry as stale) — route it through
@@ -7043,6 +7049,26 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx,
     return 0;
 }
 
+#ifdef XCE_TESTING
+/* bug-1891: drive repl_ack_decode_and_dispatch for one inbound packet on a
+ * promoted vote endpoint (ep_is_vote=1, peer_idx relative to the votes
+ * array) or a heartbeat standby endpoint (ep_is_vote=0, peer_idx absolute).
+ * Takes ownership of pkt. */
+int ha_repl_ack_dispatch_for_test(ha_state_t *ha, endpoint_t *ep,
+                                  int ep_is_vote, int peer_idx,
+                                  io_packet_t *pkt)
+{
+    static ha_elect_ctx_t vote_ctx; /* only its non-NULL-ness is read */
+    vote_ep_ctx_t         vctx;
+
+    memset(&vctx, 0, sizeof(vctx));
+    vctx.ha       = ha;
+    vctx.ctx      = ep_is_vote ? &vote_ctx : NULL;
+    vctx.peer_idx = peer_idx;
+    return repl_ack_decode_and_dispatch(ep, &vctx, pkt);
+}
+#endif
+
 /*
  * peer_ctrl_msg_handler_inline: the state-aware peer-control routing body.
  * Routes by ha->role at dispatch time — no explicit handler reassignment needed.

src/controller/ha.h

@@ -279,7 +279,7 @@ typedef struct {
                                          * lifetime cannot re-fire the restart.
                                          * 0 at init (cleared by ha_init memset). */
     /* ── New field added by ha-convergence-recovery spec (Task 4.x) ──────── */
-    _Atomic int      catchup_inflight[MAX_STANDBYS]; /* Per-standby single-flight
+    _Atomic int      catchup_inflight[MAX_HA_CANDIDATES]; /* Per-standby (absolute peer index, bug-1891) single-flight
                                          * guard for ACK-driven catch-up
                                          * continuation (Req 5.2): set when a
                                          * continuation is scheduled for a
@@ -667,6 +667,16 @@ int ha_raft_on_repl_nack(cluster_t *c, int abs_peer_idx, uint64_t reply_term,
  * remap-vs-passthrough distinction cannot drift. */
 int ha_repl_ack_abs_peer_idx(ha_state_t *ha, int ep_is_vote, int peer_idx);
 
+#ifdef XCE_TESTING
+/* bug-1891: run one inbound packet through the primary's replication-reply
+ * dispatcher as if it arrived on a promoted vote endpoint (ep_is_vote=1,
+ * relative votes index) or a heartbeat standby endpoint (ep_is_vote=0,
+ * absolute index). Takes ownership of pkt. */
+int ha_repl_ack_dispatch_for_test(ha_state_t *ha, endpoint_t *ep,
+                                  int ep_is_vote, int peer_idx,
+                                  io_packet_t *pkt);
+#endif
+
 /*
  * ha_schedule_demotion_convergence: schedule the one-shot delaymgr task that
  * truncates above committed_lsn (Req 1.2), WARN-logs the discarded range
@@ -762,7 +772,7 @@ void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn,
  *   c        — cluster_t* (real persist + repl registry; never NULL).
  *   ep       — repl registry endpoint key for the standby (endpoint_t* /
  *              repl_ep_t); identifies the standby slot for streaming + parity.
- *   peer_idx — 0-based standby slot index (into c->ha.catchup_inflight[]).
+ *   peer_idx — absolute peer index (into c->ha.catchup_inflight[]).
  *   acked_lsn— the LSN the standby just acknowledged.
  *
  * Returns 0 on the parity / single-flight-suppressed paths and on a successfully
@@ -2293,7 +2303,7 @@ void raft_cb_become_leader_with_cluster_for_test(cluster_t *c, uint64_t term);
  *   c        — cluster_t* (real persist + repl registry; never NULL in tests)
  *   ep       — repl registry endpoint key for the standby (the same pointer
  *              passed to repl_add_standby); identifies the standby slot.
- *   peer_idx — 0-based standby slot index (into ha.catchup_inflight[]).
+ *   peer_idx — absolute peer index (into ha.catchup_inflight[]).
  *   acked_lsn— the LSN the standby just acknowledged.
  *
  * Returns 0 on the no-op / parity paths and on a successfully scheduled

src/controller/meson.build

@@ -175,6 +175,7 @@ controller_unit_test_src = [
   'testsuite/test_dispatch_fail_retry.c',
   'testsuite/test_jobdispatch_no_double_notify.c',
   'testsuite/test_worker_resync.c',
+  'testsuite/test_standby_state_peer_idx.c',
   'testsuite/test_upgrade_snapshot_restore.c',
   'testsuite/test_upgrade_snapshot_apply.c',
   'testsuite/test_bug857_regression.c',

src/controller/testsuite/controller_test.c

@@ -112,6 +112,7 @@ extern void test_dispatch_fail_terminal(void);
 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);
 
 void setUp(void) { return; }
 
@@ -301,6 +302,7 @@ int main(int argc, char* argv[])
     TEST(test_dispatch_fail_retry);
     TEST(test_jobdispatch_no_double_notify);
     TEST(test_worker_resync);
+    TEST(test_standby_state_peer_idx);
 
     return UNITY_END();
 }

src/controller/testsuite/test_standby_state_peer_idx.c

@@ -0,0 +1,139 @@
+/*
+ * test_standby_state_peer_idx.c — bug-1891: a standby's heartbeat state
+ * report must be recorded under that standby's own ABSOLUTE peer index.
+ *
+ * Two kinds of endpoint deliver state reports to the primary: a promoted vote
+ * endpoint (its vote_ep_ctx_t.peer_idx is RELATIVE to the votes array, which
+ * leaves out the primary itself) and a heartbeat standby endpoint (peer_idx
+ * already absolute). repl_ack_decode_and_dispatch recorded the report under
+ * the raw peer_idx. With ctrl-1 primary (self 0), ctrl-3's promoted vote
+ * endpoint (relative 1) and a rejoined ctrl-2's heartbeat endpoint
+ * (absolute 1) then shared ha->standby_reported_*[1]: one handler read the
+ * other's position and acted on its own endpoint, so the snapshot meant for
+ * the empty, rejoining ctrl-2 was sent to the caught-up ctrl-3 instead.
+ */
+
+#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 "request.h"
+
+#define SENTINEL_LSN 777
+#define CONFIG_TAG   42
+
+static ha_state_t g_sp_ha;
+
+static ha_config_t g_sp_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,
+};
+
+/* Primary at absolute index 0 of a 3-node cluster; every slot starts at a
+ * sentinel so a write to the wrong one shows. */
+static void sp_setup(void)
+{
+    int i;
+
+    TEST_ASSERT_EQUAL_INT(0, ha_init(&g_sp_ha, &g_sp_cfg, 0));
+    g_sp_ha.n_peers = 3;
+    atomic_store_explicit(&g_sp_ha.role, (int32_t)CTRL_PRIMARY,
+                          memory_order_release);
+    atomic_store_explicit(&g_sp_ha.committed_config_tag, CONFIG_TAG,
+                          memory_order_release);
+    atomic_store_explicit(&g_sp_ha.last_known_committed_lsn, 114,
+                          memory_order_release);
+    for (i = 0; i < MAX_HA_CANDIDATES; i++) {
+        g_sp_ha.standby_reported_lsn[i] = SENTINEL_LSN;
+        g_sp_ha.standby_reported_tag[i] = CONFIG_TAG;
+    }
+}
+
+/* A heartbeat NOP reply carrying a state report, as the standby sends it. */
+static io_packet_t* sp_state_report_pkt(uint64_t applied_lsn)
+{
+    state_report_t sr;
+    xce_reply_t    rep;
+    xce_request_t  req;
+    io_packet_t*   pkt;
+
+    memset(&sr, 0, sizeof(sr));
+    sr.applied_lsn = applied_lsn;
+    sr.config_tag  = CONFIG_TAG;
+
+    memset(&rep, 0, sizeof(rep));
+    rep.opcode            = OPCode_nop;
+    rep.data_which        = RequestReply_data_stateReportMsg;
+    rep.data.state_report = &sr;
+
+    memset(&req, 0, sizeof(req));
+    req.opcode        = OPCode_reply;
+    req.body_which    = Request_body_replymsg;
+    req.body.replymsg = &rep;
+
+    pkt = calloc(1, sizeof(*pkt));
+    TEST_ASSERT_NOT_NULL(pkt);
+    TEST_ASSERT_TRUE_MESSAGE(xce_encode(&req, pkt) >= 0,
+                             "failed to encode the state report");
+    return pkt;
+}
+
+/* Baseline: a heartbeat standby endpoint's index is already absolute. */
+static void test_heartbeat_ep_report_lands_on_its_absolute_index(void)
+{
+    endpoint_t ep;
+
+    sp_setup();
+    memset(&ep, 0, sizeof(ep));
+
+    ha_repl_ack_dispatch_for_test(&g_sp_ha, &ep, /*ep_is_vote=*/0,
+                                  /*peer_idx=*/1, sp_state_report_pkt(5));
+
+    TEST_ASSERT_EQUAL_UINT64(5, g_sp_ha.standby_reported_lsn[1]);
+    TEST_ASSERT_EQUAL_UINT64(SENTINEL_LSN, g_sp_ha.standby_reported_lsn[2]);
+
+    ha_fini(&g_sp_ha);
+    printf("test_heartbeat_ep_report_lands_on_its_absolute_index: PASS\n");
+}
+
+/* THE BUG: a promoted vote endpoint's relative index 1 is peer 2 (self is
+ * 0); its report must not land on peer 1's slot. */
+static void test_vote_ep_report_lands_on_its_absolute_index(void)
+{
+    endpoint_t ep;
+
+    sp_setup();
+    memset(&ep, 0, sizeof(ep));
+
+    ha_repl_ack_dispatch_for_test(&g_sp_ha, &ep, /*ep_is_vote=*/1,
+                                  /*peer_idx=*/1, sp_state_report_pkt(5));
+
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(5, g_sp_ha.standby_reported_lsn[2],
+        "bug-1891: peer 2's state report was not recorded under peer 2");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(SENTINEL_LSN,
+        g_sp_ha.standby_reported_lsn[1],
+        "bug-1891: peer 2's state report overwrote peer 1's slot");
+
+    ha_fini(&g_sp_ha);
+    printf("test_vote_ep_report_lands_on_its_absolute_index: PASS\n");
+}
+
+void test_standby_state_peer_idx(void)
+{
+    printf("\n>> standby state report peer index tests (bug-1891)\n\n");
+
+    RUN_TEST(test_heartbeat_ep_report_lands_on_its_absolute_index);
+    RUN_TEST(test_vote_ep_report_lands_on_its_absolute_index);
+}