fix(ha-repl-hardening): snapshot fallback off-by-one — empty standby must stream, not snapshot

After reserve-lsn-0 (oldest retained WAL entry = lsn 1 on a fresh cluster), a
freshly-joined empty standby (applied_lsn=0) hit the snapshot-fallback branch
'standby_lsn < oldest_wal_lsn' (0 < 1) and was sent a non-existent snapshot
instead of streaming — so the own-term NOOP (lsn 1) never replicated, committed_lsn
stayed 0, the own-term gate (writable_lsn=1) stayed closed, and standbys kept
empty WALs. Live evidence:
  ha_dispatch_repl_action_for_standby: standby[0] lsn=0 < oldest_wal_lsn=1 -> requesting snapshot transfer

The snapshot fallback should fire only when the standby's NEXT-needed entry
(standby_lsn+1) has been GC'd below the oldest retained entry. Fix the boundary
to 'standby_lsn + 1 < oldest_wal_lsn' in ha_evaluate_repl_action (and the parallel
dead repl_needs_snapshot predicate). An empty standby (next-needed=1, oldest=1)
now STREAMs lsn 1 -> acks -> committed_lsn 0->1 -> own-term gate opens -> standbys
receive the WAL. Snapshot still fires for genuinely GC'd-behind standbys
(standby_lsn+1 < oldest). No underflow (oldest=0 -> 1<0 false). Pre-reserve-lsn-0
this off-by-one was masked by oldest_wal_lsn=0 (0<0 false).

TDD: test_empty_standby_fresh_cluster_streams_not_snapshot (RED->GREEN);
snapshot + heartbeat pins strengthened to assert SEND_SNAPSHOT only for genuinely
GC'd standbys and STREAM_WAL at the empty/boundary case. Full controller suite
green under ASAN; commit engine / streamer / apply path / wire untouched.

src/controller/ha.c

@@ -1251,9 +1251,9 @@ void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf,
  *
  * Implements the decision tree from the ha-repl-hardening design:
  *
- *   if standby_tag < committed_tag       → HA_REPL_PUSH_CONFIG
+ *   if standby_tag < committed_tag           → HA_REPL_PUSH_CONFIG
- *   elif standby_lsn < oldest_wal_lsn   → HA_REPL_SEND_SNAPSHOT
+ *   elif standby_lsn + 1 < oldest_wal_lsn   → HA_REPL_SEND_SNAPSHOT
- *   elif standby_lsn < committed_lsn    → HA_REPL_STREAM_WAL
+ *   elif standby_lsn < committed_lsn        → HA_REPL_STREAM_WAL
  *   else (tag >= committed AND lsn >= committed) → HA_REPL_CONVERGED
  *
  * No side effects.  All callers must use the returned action to drive the
@@ -1271,7 +1271,14 @@ ha_repl_action_t ha_evaluate_repl_action(uint64_t standby_tag,
     if (standby_tag < committed_tag)
         return HA_REPL_PUSH_CONFIG;
 
-    if (standby_lsn < oldest_wal_lsn)
+    /* Snapshot only when the standby's NEXT-needed entry (standby_lsn + 1) has
+     * been garbage-collected below the oldest retained WAL entry — i.e. it can
+     * no longer be served by streaming.  An empty standby (standby_lsn=0) on a
+     * fresh cluster whose oldest retained entry IS lsn 1 (post reserve-lsn-0)
+     * needs lsn 1, which is retained: 0 + 1 < 1 is false → STREAM, no snapshot.
+     * Using standby_lsn + 1 (not standby_lsn - ... ) avoids unsigned underflow
+     * when oldest_wal_lsn == 0 (empty WAL): 0 + 1 < 0 is false → not snapshot. */
+    if (standby_lsn + 1 < oldest_wal_lsn)
         return HA_REPL_SEND_SNAPSHOT;
 
     if (standby_lsn < committed_lsn)

src/controller/repl.c

@@ -633,8 +633,13 @@ int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn,
     (void)r;
     /* No WAL at all: empty standby can start from LSN 0 without a snapshot */
     if (oldest_wal_lsn == 0) return 0;
-    /* Standby LSN predates oldest surviving WAL segment — full snapshot required */
+    /* Snapshot only when the standby's NEXT-needed entry (standby_lsn + 1) has
-    return (standby_lsn < oldest_wal_lsn) ? 1 : 0;
+     * been GC'd below the oldest surviving WAL segment — i.e. it can no longer be
+     * served by streaming.  An empty standby (standby_lsn=0) whose next-needed
+     * entry IS the oldest retained entry (e.g. lsn 1 post reserve-lsn-0) is
+     * streamable: 0 + 1 < 1 is false → no snapshot.  Mirrors the boundary in
+     * ha_evaluate_repl_action (ha.c). */
+    return (standby_lsn + 1 < oldest_wal_lsn) ? 1 : 0;
 }
 
 /* ============================================================

src/controller/testsuite/test_convergence_verification.c

@@ -484,16 +484,28 @@ static void test_heartbeat_decision_path_unchanged(void)
         "HA_REPL_STREAM_WAL — the heartbeat path stays the fallback catch-up "
         "initiator (Req 5.3)");
 
-    /* HA_REPL_SEND_SNAPSHOT: standby_lsn predates the oldest retained WAL. */
+    /* HA_REPL_SEND_SNAPSHOT: standby_lsn genuinely GC'd behind the oldest
+     * retained WAL (standby_lsn + 1 < oldest_wal_lsn → 2 + 1 = 3 < 5). */
     ha_repl_action_t a_snap = ha_evaluate_repl_action(
         /*standby_tag=*/5, /*standby_lsn=*/2,
         /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5);
     TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)a_snap,
-        "heartbeat decision: standby_lsn below oldest_wal_lsn must remain "
+        "heartbeat decision: standby GC'd behind oldest_wal_lsn must remain "
         "HA_REPL_SEND_SNAPSHOT — the heartbeat path remains the SOLE snapshot "
         "decision owner (Req 5.3); the ACK-driven continuation never decides "
         "snapshots");
 
+    /* Strict boundary (fix-snapshot-boundary): standby_lsn == oldest_wal_lsn - 1
+     * (next-needed entry retained) must STREAM, not snapshot — pins the
+     * standby_lsn + 1 < oldest_wal_lsn comparison on the heartbeat path. */
+    ha_repl_action_t a_edge = ha_evaluate_repl_action(
+        /*standby_tag=*/5, /*standby_lsn=*/4,
+        /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5);
+    TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_STREAM_WAL, (int)a_edge,
+        "heartbeat decision: standby_lsn (4) == oldest_wal_lsn (5) - 1 — the "
+        "next-needed entry (5) is retained, so the heartbeat path must STREAM, "
+        "not snapshot (strict standby_lsn + 1 < oldest_wal_lsn boundary)");
+
     printf("test_heartbeat_decision_path_unchanged: checked\n");
 }
 
@@ -805,28 +817,47 @@ static void test_pin_repl_broadcast_full_fanout_skips_empty_and_catchup(void)
  * decision-boundary pin.) */
 static void test_pin_evaluate_repl_action_snapshot_below_oldest_wal(void)
 {
-    /* Below the oldest retained WAL → SEND_SNAPSHOT. tags equal (no config push
+    /* fix-snapshot-boundary: the snapshot fallback fires iff the standby's
-     * shortcut); committed_lsn high so the only deciding factor is the snapshot
+     * NEXT-needed entry (standby_lsn + 1) is older than the oldest retained WAL
-     * boundary. */
+     * entry — i.e. standby_lsn + 1 < oldest_wal_lsn (genuinely GC'd behind).
-    ha_repl_action_t below = ha_evaluate_repl_action(
+     * tags equal (no config-push shortcut); committed_lsn high so the snapshot
+     * boundary is the only deciding factor. */
+
+    /* GENUINELY GC'd behind: standby_lsn=2, oldest_wal_lsn=5 → entries 3,4 were
+     * GC'd, the next-needed entry (3) is gone → SEND_SNAPSHOT.
+     * (2 + 1 = 3 < 5 → snapshot.) */
+    ha_repl_action_t gcd_behind = ha_evaluate_repl_action(
         /*standby_tag=*/5, /*standby_lsn=*/2,
         /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5);
-    TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)below,
+    TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)gcd_behind,
-        "standby_lsn (2) below oldest_wal_lsn (5) must decide SEND_SNAPSHOT "
+        "standby_lsn (2) is GC'd behind oldest_wal_lsn (5) — its next-needed "
-        "(umbrella 6.3 fallback boundary)");
+        "entry (3) is no longer retained, so it must decide SEND_SNAPSHOT "
+        "(snapshot still fires when genuinely needed)");
 
-    /* Exactly AT oldest_wal_lsn is inside the retained window → NOT snapshot
+    /* BOUNDARY (streamable edge): standby_lsn == oldest_wal_lsn - 1, i.e. the
-     * (streamable). The boundary is strict-less-than. */
+     * next-needed entry is EXACTLY the oldest retained entry → STREAM, not
-    ha_repl_action_t at_boundary = ha_evaluate_repl_action(
+     * snapshot.  standby_lsn=4, oldest_wal_lsn=5 → 4 + 1 = 5 < 5 is false. This
-        /*standby_tag=*/5, /*standby_lsn=*/5,
+     * is the strict boundary: a wrong comparison (standby_lsn < oldest_wal_lsn)
+     * would (incorrectly) snapshot here, failing this assertion. */
+    ha_repl_action_t boundary = ha_evaluate_repl_action(
+        /*standby_tag=*/5, /*standby_lsn=*/4,
         /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5);
-    TEST_ASSERT_NOT_EQUAL_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)at_boundary,
+    TEST_ASSERT_NOT_EQUAL_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)boundary,
-        "standby_lsn == oldest_wal_lsn is within the retained WAL window — the "
+        "standby_lsn (4) == oldest_wal_lsn (5) - 1: the next-needed entry (5) IS "
-        "snapshot boundary is strict-less-than, so this must NOT decide "
+        "the oldest retained entry — streamable, must NOT decide SEND_SNAPSHOT "
-        "SEND_SNAPSHOT (umbrella 6.3 boundary)");
+        "(strict standby_lsn + 1 < oldest_wal_lsn boundary)");
-    TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_STREAM_WAL, (int)at_boundary,
+    TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_STREAM_WAL, (int)boundary,
-        "at the snapshot boundary with lsn (5) below committed (9) the standby is "
+        "at the snapshot boundary with lsn (4) below committed (9) the standby is "
-        "streamable — HA_REPL_STREAM_WAL (umbrella 6.3 boundary)");
+        "streamable — HA_REPL_STREAM_WAL");
+
+    /* EMPTY standby on fresh cluster: standby_lsn=0, oldest_wal_lsn=1 → next
+     * entry (1) retained → STREAM. 0 + 1 = 1 < 1 false. The core fix case. */
+    ha_repl_action_t empty_fresh = ha_evaluate_repl_action(
+        /*standby_tag=*/5, /*standby_lsn=*/0,
+        /*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/1);
+    TEST_ASSERT_NOT_EQUAL_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)empty_fresh,
+        "empty standby (lsn=0), oldest_wal_lsn=1: next-needed entry (1) retained "
+        "→ must STREAM, not snapshot (strict boundary)");
 
     /* Above the oldest WAL and at committed → CONVERGED, never snapshot. */
     ha_repl_action_t converged = ha_evaluate_repl_action(
@@ -839,6 +870,38 @@ static void test_pin_evaluate_repl_action_snapshot_below_oldest_wal(void)
     printf("test_pin_evaluate_repl_action_snapshot_below_oldest_wal: checked\n");
 }
 
+/* fix-snapshot-boundary: an EMPTY standby on a FRESH cluster must STREAM, not
+ * demand a snapshot.
+ *
+ * After reserve-lsn-0 the first WAL entry is lsn 1, so a fresh cluster's oldest
+ * retained entry is lsn 1.  A freshly-joined empty standby reports applied_lsn=0.
+ * Its NEXT-needed entry is lsn 1, which IS retained — so streaming can serve it;
+ * no snapshot exists yet on a fresh cluster.  The snapshot fallback must fire
+ * only when the standby's next-needed entry (standby_lsn + 1) has been GC'd below
+ * the oldest retained entry, i.e. standby_lsn + 1 < oldest_wal_lsn.
+ *
+ * RED before the fix: the old boundary (standby_lsn < oldest_wal_lsn) gives
+ * 0 < 1 → SEND_SNAPSHOT for the empty standby.  GREEN after: 0+1=1 < 1 is
+ * false → STREAM_WAL streams lsn 1, the cold-start NOOP replicates + commits. */
+static void test_empty_standby_fresh_cluster_streams_not_snapshot(void)
+{
+    /* Empty standby (applied_lsn=0), fresh cluster (oldest_wal_lsn=1, NOOP at
+     * lsn 1), WAL tip > 0 (committed_lsn=1).  Tags equal so the config-push
+     * shortcut does not mask the snapshot/stream decision. */
+    ha_repl_action_t a = ha_evaluate_repl_action(
+        /*standby_tag=*/0, /*standby_lsn=*/0,
+        /*committed_tag=*/0, /*committed_lsn=*/1, /*oldest_wal_lsn=*/1);
+    TEST_ASSERT_NOT_EQUAL_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)a,
+        "empty standby (lsn=0) on a fresh cluster (oldest_wal_lsn=1) needs lsn 1, "
+        "which is retained — it must NOT demand a snapshot (there is none on a "
+        "fresh cluster); the next-needed entry (lsn 1) is streamable");
+    TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_STREAM_WAL, (int)a,
+        "empty standby (lsn=0 < committed_lsn=1) with next-needed entry retained "
+        "must STREAM_WAL so the cold-start NOOP (lsn 1) replicates and commits");
+
+    printf("test_empty_standby_fresh_cluster_streams_not_snapshot: checked\n");
+}
+
 /* ── Degraded-mode dispatch-gate fixture (task 6.1 (c)) ──────────────────────
  * Self-contained ha_state_t with role=CTRL_PRIMARY and write quorum ABSENT
  * (n_peers=3, zero connected standby endpoints → ha_has_write_quorum false via
@@ -1002,6 +1065,7 @@ void test_convergence_verification(void)
 
     RUN_TEST(test_pin_repl_broadcast_full_fanout_skips_empty_and_catchup);
     RUN_TEST(test_pin_evaluate_repl_action_snapshot_below_oldest_wal);
+    RUN_TEST(test_empty_standby_fresh_cluster_streams_not_snapshot);
     RUN_TEST(test_pin_dispatch_gate_degraded_allowlist);
     RUN_TEST(test_pin_repl_complete_catchup_reincludes_in_broadcast);
 #else

src/controller/testsuite/test_p2_integration.c

@@ -190,11 +190,26 @@ static void test_needs_snapshot_boundary(void)
     replicator_t r;
     repl_init(&r);
 
-    /* Exact boundary: standby_lsn == oldest_wal_lsn — no snapshot needed */
+    /* fix-snapshot-boundary: snapshot fires iff the standby's NEXT-needed entry
+     * (standby_lsn + 1) is older than the oldest retained WAL entry — i.e.
+     * standby_lsn + 1 < oldest_wal_lsn (genuinely GC'd behind). */
+
+    /* At/above oldest: standby_lsn == oldest_wal_lsn — no snapshot. */
     TEST_ASSERT_EQUAL_INT(0, repl_needs_snapshot(&r, 10, 10));
 
-    /* One behind: standby_lsn == oldest_wal_lsn - 1 — snapshot required */
+    /* Streamable edge: standby_lsn == oldest_wal_lsn - 1 — the next-needed entry
-    TEST_ASSERT_EQUAL_INT(1, repl_needs_snapshot(&r, 9, 10));
+     * (10) IS the oldest retained entry → streamable, NO snapshot.
+     * (9 + 1 = 10 < 10 is false.) */
+    TEST_ASSERT_EQUAL_INT(0, repl_needs_snapshot(&r, 9, 10));
+
+    /* Empty standby on a fresh cluster: next-needed entry (1) retained → no
+     * snapshot. (0 + 1 = 1 < 1 is false.) */
+    TEST_ASSERT_EQUAL_INT(0, repl_needs_snapshot(&r, 0, 1));
+
+    /* Genuinely GC'd behind: standby_lsn + 1 < oldest_wal_lsn — next-needed
+     * entry (9) was GC'd (oldest retained is 10) → snapshot required.
+     * (8 + 1 = 9 < 10 is true.) */
+    TEST_ASSERT_EQUAL_INT(1, repl_needs_snapshot(&r, 8, 10));
 
     repl_fini(&r);
     printf("test_needs_snapshot_boundary: PASS\n");