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/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);