From 762c6f1883573b31a10c53f742b5c32d3d21cb21 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 4 Oct 2026 12:02:37 +0000 Subject: [PATCH] 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) --- src/controller/config_repl.c | 17 +++- src/controller/ha.c | 30 ++++++- src/controller/persist.c | 12 +++ .../testsuite/test_commit_durability.c | 78 +++++++++++++++++++ .../testsuite/test_convergence_verification.c | 42 ++++++++++ .../testsuite/test_term_safety_adapter.c | 65 ++++++++++++++++ 6 files changed, 242 insertions(+), 2 deletions(-) diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index a05be211..f1538962 100644 --- a/src/controller/config_repl.c +++ b/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; } diff --git a/src/controller/ha.c b/src/controller/ha.c index a6dd1f13..9869c621 100644 --- a/src/controller/ha.c +++ b/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; } diff --git a/src/controller/persist.c b/src/controller/persist.c index 6cc6785e..23fe6f5a 100644 --- a/src/controller/persist.c +++ b/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): diff --git a/src/controller/testsuite/test_commit_durability.c b/src/controller/testsuite/test_commit_durability.c index 8640147e..eb488536 100644 --- a/src/controller/testsuite/test_commit_durability.c +++ b/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-"); + 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); diff --git a/src/controller/testsuite/test_convergence_verification.c b/src/controller/testsuite/test_convergence_verification.c index 5e7f49ef..f4718cc3 100644 --- a/src/controller/testsuite/test_convergence_verification.c +++ b/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"); diff --git a/src/controller/testsuite/test_term_safety_adapter.c b/src/controller/testsuite/test_term_safety_adapter.c index 972a2ff1..9564e935 100644 --- a/src/controller/testsuite/test_term_safety_adapter.c +++ b/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);