fix(controller): refuse replicated WAL writes before this boot's recovery (bug-1920)

A restarted standby accepts the primary's connection during its startup
delay, before persist_recover runs. Catch-up entries written into the
un-recovered persist_t opened seg-...0001 with O_TRUNC and rewrote it,
and after recovery the stale fd kept filing new LSNs under that segment
name. The primary's filename-ordered streamer then skipped them, so a
standby needing that range never caught up and commits stalled.

persist_wal_write_replicated now returns -EBUSY until recovery completes
(the bug-1416 recovery_complete gate). ha_apply_wal_catchup_entry and
ha_handle_snapshot_chunk drop pre-recovery traffic without ACK or NACK.
Catch-up streams from min(NACK-derived join position, standby report),
and the streamer warns when a non-empty range yields no entries.

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

src/controller/config_repl.c

@@ -2511,6 +2511,7 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
     wal_catchup_entry_t *batch_entries[WAL_STREAM_BATCH_ENTRIES];
     int    batch_n     = 0;
     size_t batch_bytes = 0;
+    uint64_t n_queued  = 0; /* entries put in a batch (bug-1920 diagnostic) */
 
     /* ha-term-safety task 3.4: track the previous streamed entry's
      * (lsn, term) so each catch-up entry carries the AppendEntries prev tuple
@@ -2699,6 +2700,7 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
 
                 batch_entries[batch_n] = entry;
                 batch_n++;
+                n_queued++;
                 batch_bytes += sizeof(wal_entry_hdr_t) + hdr.payload_len;
 
                 /* Flush when batch is full (by count or accumulated size) */
@@ -2741,8 +2743,21 @@ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path,
         }
     }
 
+    /* bug-1920: a non-empty range that yields nothing means the on-disk
+     * segment layout does not match the filename-ordered scan above (an LSN
+     * filed under the wrong segment name) -- say so instead of logging a
+     * success the standby never sees. */
+    if (n_queued == 0) {
+        WARN("ha_stream_wal_to_standby: found NO entries for [%" PRIu64
+             "+1..%" PRIu64 "] in %d segment(s) under %s — nothing streamed "
+             "(WAL segment layout does not match LSN order?)",
+             standby_lsn, committed_lsn, nseg, wal_path);
+        return 0;
+    }
+
     INFO("ha_stream_wal_to_standby: streamed WAL [%" PRIu64 "+1..%" PRIu64
-         "] to standby", standby_lsn, committed_lsn);
+         "] to standby (%" PRIu64 " entries)", standby_lsn, committed_lsn,
+         n_queued);
     return 0;
 }
 

src/controller/ha.c

@@ -3658,6 +3658,21 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
     if (!ha || !p || !entry)
         return -EINVAL;
 
+    /* bug-1920: before this boot's persist_recover the local log is not
+     * loaded: persisting would corrupt the on-disk segment layout (see
+     * persist_wal_write_replicated), and a NACK would walk the primary's
+     * next_index back past entries this node really holds.  Drop silently --
+     * no persist, no ACK, no NACK; the first state report after recovery
+     * carries the real position and catch-up resumes from there. */
+    if (!persist_recovery_complete(p)) {
+        static _Atomic int s_logged;
+        if (!atomic_exchange_explicit(&s_logged, 1, memory_order_relaxed))
+            INFO("ha_apply_wal_catchup_entry: replication received before "
+                 "this boot's WAL recovery completed — dropping until "
+                 "recovered (bug-1920)");
+        return 0;
+    }
+
     /* ── ha-term-safety task 3.2: term-carrying AppendEntries → library route ──
      * A term-safe AppendEntries (one sent by repl_send_append_entries / the
      * library send path) carries a non-zero entry term stamp. Route it through
@@ -3965,6 +3980,12 @@ int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
     if (!ha || !p || !snap_dir || !chunk)
         return -EINVAL;
 
+    /* bug-1920: same pre-recovery drop as ha_apply_wal_catchup_entry -- a
+     * snapshot installed before persist_recover would be clobbered or mixed
+     * with the replay.  The transfer restarts from chunk 0 after recovery. */
+    if (!persist_recovery_complete(p))
+        return 0;
+
     char part_path[4096 + 16];
     snprintf(part_path, sizeof(part_path), "%s/snap.part", snap_dir);
 
@@ -4679,7 +4700,14 @@ int ha_dispatch_repl_action_for_standby(ha_state_t *ha,
      * corrected position at least once. */
     if (c) {
         uint64_t join_lsn;
-        if (repl_get_catchup_join_lsn(&c->persist.repl, ep, &join_lsn))
+        /* bug-1920: never above the standby's own report.  The NACK backoff
+         * moves next_index down one per reject, so join_lsn can sit far above
+         * where the standby's log really ends; streaming from there is
+         * rejected (one step per heartbeat) or finds nothing to send.  The
+         * report is an upper bound on what the standby holds; a lower
+         * join_lsn (a conflicted suffix the report cannot see) still wins. */
+        if (repl_get_catchup_join_lsn(&c->persist.repl, ep, &join_lsn)
+            && join_lsn < standby_lsn)
             standby_lsn = join_lsn;
     }
 

src/controller/persist.c

@@ -1809,6 +1809,18 @@ int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type,
     if (len > MAX_WAL_PAYLOAD_BYTES)
         return -EMSGSIZE;
 
+    /* bug-1920: refuse until this boot's persist_recover has run.  Before it,
+     * current_lsn/seg state are the persist_init defaults (current_lsn=1, no
+     * open segment), so a write here would open_new_segment(1) -- O_TRUNC'ing
+     * the existing seg-...0001 -- and keep that fd as the append segment after
+     * recovery moves current_lsn to the real tip, filing every later LSN under
+     * the wrong segment name.  The primary connects (and streams) during the
+     * standby's startup delay, so this window is live; the intake
+     * (ha_apply_wal_catchup_entry) drops such traffic first, this is the
+     * backstop for every other replicated-write caller. */
+    if (!persist_recovery_complete(p))
+        return -EBUSY;
+
     pthread_mutex_lock(&p->lock);
 
     /* Tip-repair bookkeeping (set only on the TIP-conflict repair branch below):

src/controller/testsuite/test_commit_durability.c

@@ -1289,6 +1289,83 @@ static int open_first_seg_rw(const char* tmpdir)
  * (via persist_recovery_truncated). The writer must already be torn down
  * so the segments are closed and on disk. */
 
+
+/* Count the WAL entries in one segment file (header walk up to the
+ * zero-filled pre-allocated tail). */
+static int dur_count_seg_entries(const char* seg_path)
+{
+    int fd = open(seg_path, O_RDONLY);
+    if (fd < 0) return -1;
+    int      n   = 0;
+    uint64_t off = 0;
+    for (;;) {
+        wal_entry_hdr_t hdr;
+        if (pread(fd, &hdr, sizeof(hdr), (off_t)off) < (ssize_t)sizeof(hdr))
+            break;
+        if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0)
+            break;
+        n++;
+        off += sizeof(hdr) + hdr.payload_len;
+    }
+    close(fd);
+    return n;
+}
+
+/* bug-1920: a replicated write that reaches a restarted standby BEFORE this
+ * boot's persist_recover (the primary connects during the startup delay) must
+ * be refused.  Before the fix the not-yet-recovered persist_t (current_lsn=1,
+ * no open segment) accepted it: open_new_segment(current_lsn=1) O_TRUNC'd the
+ * existing seg-...0001 and rewrote history into it, and once recovery moved
+ * current_lsn to the real tip the still-open fd kept appending NEW LSNs into
+ * that wrongly-named segment, which the primary's catch-up streamer (it orders
+ * segments by filename) then never streamed. */
+static void test_replicated_write_refused_before_recovery(void)
+{
+    char tmp[256];
+    strcpy(tmp, "/tmp/xce_prerecov_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
+    char seg1[512], seg4[512];
+    snprintf(seg1, sizeof(seg1), DUR_WAL_SEG_FMT, tmp, (uint64_t)1);
+    snprintf(seg4, sizeof(seg4), DUR_WAL_SEG_FMT, tmp, (uint64_t)4);
+
+    /* Previous life: three durable entries in seg-...0001. */
+    {
+        persist_t* p = make_durability_persist(tmp);
+        TEST_ASSERT_NOT_NULL(p);
+        TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1));
+        TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "b", 1));
+        TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "c", 1));
+        TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
+        teardown_durability_persist(p);
+    }
+    TEST_ASSERT_EQUAL_INT(3, dur_count_seg_entries(seg1));
+
+    /* Restart: the boot closes the recovery gate before any network traffic
+     * (myapp.c), then a replicated entry arrives before persist_recover. */
+    persist_t* p2 = make_durability_persist(tmp);
+    TEST_ASSERT_NOT_NULL(p2);
+    persist_mark_recovery_pending(p2);
+    int rc = persist_wal_write_replicated(p2, /*lsn=*/1, WAL_JOB_UPDATE, "x", 1);
+    TEST_ASSERT_EQUAL_INT_MESSAGE(
+        -EBUSY, rc,
+        "bug-1920: a replicated write before persist_recover must be refused");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(
+        3, dur_count_seg_entries(seg1),
+        "bug-1920: a refused pre-recovery write must not touch the on-disk log");
+
+    /* Recovery then appends at the real tip, into a segment named after it. */
+    TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
+    TEST_ASSERT_EQUAL_UINT64(4, p2->current_lsn);
+    TEST_ASSERT_EQUAL_INT(0,
+        persist_wal_write_replicated(p2, /*lsn=*/4, WAL_JOB_UPDATE, "d", 1));
+    TEST_ASSERT_EQUAL_INT(0, persist_flush(p2));
+    TEST_ASSERT_EQUAL_INT_MESSAGE(
+        1, dur_count_seg_entries(seg4),
+        "bug-1920: the first post-recovery append opens seg-<its LSN>");
+    TEST_ASSERT_EQUAL_INT(3, dur_count_seg_entries(seg1));
+    teardown_durability_persist(p2);
+    printf("test_replicated_write_refused_before_recovery: PASS\n");
+}
 /* ------------------------------------------------------------------ *
  * Test 3.1a: an intact log yields RECOVER_CLEAN (flag NOT set).       *
  * ------------------------------------------------------------------ */
@@ -2767,6 +2844,7 @@ void test_commit_durability(void)
     RUN_TEST(test_sole_durable_contributor_loss_survives);
     RUN_TEST(test_durability_work_off_iomgr_preserves_liveness);
     RUN_TEST(test_recover_verdict_clean_intact_log);
+    RUN_TEST(test_replicated_write_refused_before_recovery);
     RUN_TEST(test_recover_verdict_empty_first_boot_clean);
     RUN_TEST(test_recover_verdict_crc_break_first_entry_truncated);
     RUN_TEST(test_recover_verdict_snapshot_sentinel_clean);

src/controller/testsuite/test_convergence_verification.c

@@ -479,6 +479,47 @@ static void test_walack_at_parity_completes_catchup_and_clears_in_catch_up(void)
     printf("test_walack_at_parity_completes_catchup_and_clears_in_catch_up: checked\n");
 }
 
+
+/* bug-1920: the heartbeat-driven catch-up must never stream from ABOVE the
+ * standby's own reported position.  A log-mismatch NACK walks the primary's
+ * next_index back one per reject, so the catch-up join position can sit far
+ * above where the standby's log really ends (soak: join 340, standby at 323).
+ * Streaming [341..] from there is rejected again at best (one step per
+ * heartbeat) and silent at worst; the standby's own report is an upper bound
+ * on what it holds, so the start must be min(join position, report). */
+static void test_dispatch_streams_from_min_of_join_and_report(void)
+{
+    cv_setup("b1920", /*n_peers=*/3, /*n_standbys=*/1,
+             /*appended_lsn=*/342, /*n_vcpus=*/8);
+    atomic_store(&g_cv.persist.committed_lsn, 341);
+
+    repl_ep_t ep = &g_cv_eps[0];
+    repl_start_catchup(&g_cv.persist.repl, ep, /*join_applied_lsn=*/340);
+    g_cv.ha.standby_reported_lsn[0] = 323;
+    g_cv.ha.standby_reported_tag[0] = 0;
+
+    ha_dispatch_repl_action_for_standby(&g_cv.ha, &g_cv, (endpoint_t *)ep,
+                                        /*peer_idx=*/0, /*committed_tag=*/0,
+                                        /*committed_lsn=*/341,
+                                        /*oldest_wal_lsn=*/1);
+
+    TEST_ASSERT_EQUAL_INT(1, atomic_load(&g_stream_calls));
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
+        323, atomic_load(&g_stream_last_standby_lsn),
+        "bug-1920: stream from the standby's reported position (323), not the "
+        "higher NACK-derived join position (340)");
+
+    /* The join position still wins when it is the LOWER one (bug-1479: a
+     * conflicted suffix the standby's report cannot see). */
+    atomic_store(&g_stream_calls, 0);
+    repl_start_catchup(&g_cv.persist.repl, ep, /*join_applied_lsn=*/300);
+    ha_dispatch_repl_action_for_standby(&g_cv.ha, &g_cv, (endpoint_t *)ep,
+                                        0, 0, 341, 1);
+    TEST_ASSERT_EQUAL_UINT64(300, atomic_load(&g_stream_last_standby_lsn));
+
+    cv_teardown();
+    printf("test_dispatch_streams_from_min_of_join_and_report: checked\n");
+}
 /* 5.3 decision pin: the heartbeat decision function ha_evaluate_repl_action is
  * the fallback initiator of catch-up and the SOLE owner of the snapshot
  * decision; the ACK-driven continuation must not change its verdicts.  This pins
@@ -1457,6 +1498,7 @@ void test_convergence_verification(void)
     RUN_TEST(test_walack_single_flight_no_duplicate_continuation);
     RUN_TEST(test_walack_reschedules_after_continuation_completes);
     RUN_TEST(test_walack_at_parity_completes_catchup_and_clears_in_catch_up);
+    RUN_TEST(test_dispatch_streams_from_min_of_join_and_report);
     RUN_TEST(test_heartbeat_decision_path_unchanged);
 
     printf("\n>> Quorum transition logging tests (RED, task 5.1)\n\n");

src/controller/testsuite/test_term_safety_adapter.c

@@ -1424,6 +1424,70 @@ static void test_live_broadcast_receive_persists_leader_term(void)
     printf("test_live_broadcast_receive_persists_leader_term: PASS\n");
 }
 
+/* bug-1920: replication traffic that reaches a restarted standby before this
+ * boot's persist_recover must be dropped at intake: not persisted, not ACKed
+ * and NOT NACKed.  Persisting it corrupted the on-disk segment layout; a NACK
+ * would walk the primary's next_index back for every entry the standby
+ * actually holds on disk.  The standby reports its real position in its first
+ * state report after recovery and catch-up resumes from there. */
+static int      g_b1920_ack_calls;
+static uint64_t g_b1920_ack_lsn;
+static int      g_b1920_ack_success;
+static void b1920_ack_capture(uint64_t term, uint64_t lsn, int success)
+{
+    (void)term;
+    g_b1920_ack_calls++;
+    g_b1920_ack_lsn     = lsn;
+    g_b1920_ack_success = success;
+}
+
+static void test_intake_dropped_before_recovery(void)
+{
+    char tmp[256];
+    strcpy(tmp, "/tmp/xce_ts_prerecov_XXXXXX");
+    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
+
+    cluster_t* c = make_cluster(tmp);
+    TEST_ASSERT_NOT_NULL(c);
+    make_follower(c, /*term*/7, /*n_peers*/3);
+    persist_mark_recovery_pending(&c->persist);
+    uint64_t lsn_before = c->persist.current_lsn;
+
+    uint8_t pay[1] = { 'a' };
+    wal_catchup_entry_t sub;
+    memset(&sub, 0, sizeof(sub));
+    sub.lsn             = 1;
+    sub.entry_type      = WAL_JOB_SUBMIT;
+    sub.term            = 7;
+    sub.raw_payload     = pay;
+    sub.raw_payload_len = sizeof(pay);
+    wal_catchup_entry_t* sub_arr[1] = { &sub };
+    wal_catchup_entry_t batch;
+    memset(&batch, 0, sizeof(batch));
+    batch.n_entries = 1;
+    batch.entries   = sub_arr;
+    batch.term      = 7;
+
+    extern void (*ha_wal_ack_send_hook)(uint64_t, uint64_t, int);
+    g_b1920_ack_calls = 0;
+    ha_wal_ack_send_hook = b1920_ack_capture;
+    int rc = ha_apply_wal_catchup_entry(&c->ha, &c->persist, &batch,
+                                        /*ep*/NULL, c);
+    ha_wal_ack_send_hook = NULL;
+
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT_MESSAGE(
+        0, g_b1920_ack_calls,
+        "bug-1920: pre-recovery replication must be neither ACKed nor NACKed");
+    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
+        lsn_before, c->persist.current_lsn,
+        "bug-1920: pre-recovery replication must not be persisted");
+    TEST_ASSERT_EQUAL_UINT64(0, c->ha.raft.log_len);
+
+    teardown_cluster(c);
+    printf("test_intake_dropped_before_recovery: PASS\n");
+}
+
 /* ================================================================== */
 /* bug-991 — catch-up applied_lsn must be "last contiguously-held LSN" */
 /*           (NOT next-to-apply / lsn+1) so the leader's catch-up      */
@@ -2452,6 +2516,7 @@ void test_term_safety_adapter(void)
     RUN_TEST(test_intake_divergent_tail_truncated_and_reconverged);
     RUN_TEST(test_intake_logmismatch_rejected_no_false_convergence);
     RUN_TEST(test_live_broadcast_receive_persists_leader_term);
+    RUN_TEST(test_intake_dropped_before_recovery);
     RUN_TEST(test_catchup_applied_lsn_is_last_held_no_skip);
     RUN_TEST(test_intake_crossterm_catchup_uses_leader_current_term);
     RUN_TEST(test_intake_stale_leader_term_still_rejected);