From 124f4e6a13f6023fb6cf31cb02fe7cadeb693d9b Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Fri, 21 Nov 2025 13:21:42 +0800
Subject: [PATCH v1 1/2] Fix race conditions causing invalidation of newly
 synced slots

When initially syncing a replication slot to the standby server, the remote
restart_lsn queried from the publisher is used as the initial restart_lsn for
newly synced slots. As slots are synced asynchronously, the synced restart_lsn
may lag behind the standby's redo pointer. Due to the lack of interlock between
WAL reservation and checkpoints, this creates a race conditions where
checkpoints might remove required WALs and invalidate the newly synced slots.

To address this issue, similar to commit 006dd4b, we first acquire the
ReplicationSlotAllocationLock when reserving WAL position for the newly synced
slot. This ensures that if WAL reservation occurs first, the checkpoint process
is forced to wait for slotsync to update the newly synced slot's restart_lsn
before calculating the minimum LSN.

However, unlike in ReplicationSlotReserveWal(), this lock cannot protect a newly
synced slot if a concurrent checkpoint has invoked CheckPointReplicationSlots()
before the the slotsync updates the restart_lsn. This is because in slotsync
case, the initial restart_lsn received from the remote server may be outdated,
preceding the redo pointer. Therefore, when selecting the initial restart_lsn,
this commit uses the redo pointer or the minimum slot LSN (if those values are
greater than the remote restart_lsn) instead of relying solely on the remote
value.
---
 src/backend/access/transam/xlog.c          |  3 +-
 src/backend/replication/logical/slotsync.c | 92 +++++++++++-----------
 src/include/access/xlog.h                  |  1 +
 3 files changed, 47 insertions(+), 49 deletions(-)

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 81dc86847c0..9f04125ef2a 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -669,7 +669,6 @@ static XLogRecPtr CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn,
 												  TimeLineID newTLI);
 static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags);
 static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo);
-static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void);
 
 static void AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli,
 								  bool opportunistic);
@@ -2679,7 +2678,7 @@ XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
  * Return the oldest LSN we must retain to satisfy the needs of some
  * replication slot.
  */
-static XLogRecPtr
+XLogRecPtr
 XLogGetReplicationSlotMinimumLSN(void)
 {
 	XLogRecPtr	retval;
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 73fc51ea53e..3217bd8929d 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -56,6 +56,7 @@
 
 #include <time.h>
 
+#include "access/xlog.h"
 #include "access/xlog_internal.h"
 #include "access/xlogrecovery.h"
 #include "catalog/pg_database.h"
@@ -536,70 +537,67 @@ drop_local_obsolete_slots(List *remote_slot_list)
  * Reserve WAL for the currently active local slot using the specified WAL
  * location (restart_lsn).
  *
- * If the given WAL location has been removed, reserve WAL using the oldest
- * existing WAL segment.
+ * If the given WAL location has been removed or is at risk of being removed,
+ * reserve WAL using the oldest segment that is non-removable.
  */
 static void
 reserve_wal_for_local_slot(XLogRecPtr restart_lsn)
 {
-	XLogSegNo	oldest_segno;
+	XLogRecPtr	slot_min_lsn;
+	XLogRecPtr	min_safe_lsn;
 	XLogSegNo	segno;
 	ReplicationSlot *slot = MyReplicationSlot;
 
 	Assert(slot != NULL);
 	Assert(!XLogRecPtrIsValid(slot->data.restart_lsn));
 
-	while (true)
-	{
-		SpinLockAcquire(&slot->mutex);
-		slot->data.restart_lsn = restart_lsn;
-		SpinLockRelease(&slot->mutex);
-
-		/* Prevent WAL removal as fast as possible */
-		ReplicationSlotsComputeRequiredLSN();
+	/*
+	 * Acquire an exclusive lock to prevent the checkpoint process from
+	 * concurrently calculating the minimum slot LSN (see
+	 * CheckPointReplicationSlots), ensuring that if WAL reservation occurs
+	 * first, the checkpoint must wait for the restart_lsn update before
+	 * calculating the minimum LSN.
+	 *
+	 * Note: Unlike in ReplicationSlotReserveWal(), this lock does not protect a
+	 * newly synced slot if a concurrent checkpoint has invoked
+	 * CheckPointReplicationSlots() before the WAL reservation here. This is
+	 * because the initial restart_lsn received from the remote server may be
+	 * outdated, preceding the redo pointer. Therefore, when selecting the
+	 * initial restart_lsn, we might use the redo pointer or the minimum slot
+	 * LSN (if those values are greater than the remote restart_lsn) instead of
+	 * relying solely on the remote value.
+	 */
+	LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
-		XLByteToSeg(slot->data.restart_lsn, segno, wal_segment_size);
+	/*
+	 * Determine the minimum non-removable LSN by comparing the redo pointer
+	 * with the minimum slot LSN.
+	 */
+	min_safe_lsn = GetRedoRecPtr();
+	slot_min_lsn = XLogGetReplicationSlotMinimumLSN();
 
-		/*
-		 * Find the oldest existing WAL segment file.
-		 *
-		 * Normally, we can determine it by using the last removed segment
-		 * number. However, if no WAL segment files have been removed by a
-		 * checkpoint since startup, we need to search for the oldest segment
-		 * file from the current timeline existing in XLOGDIR.
-		 *
-		 * XXX: Currently, we are searching for the oldest segment in the
-		 * current timeline as there is less chance of the slot's restart_lsn
-		 * from being some prior timeline, and even if it happens, in the
-		 * worst case, we will wait to sync till the slot's restart_lsn moved
-		 * to the current timeline.
-		 */
-		oldest_segno = XLogGetLastRemovedSegno() + 1;
+	if (XLogRecPtrIsValid(slot_min_lsn) && min_safe_lsn > slot_min_lsn)
+		min_safe_lsn = slot_min_lsn;
 
-		if (oldest_segno == 1)
-		{
-			TimeLineID	cur_timeline;
+	/*
+	 * If the minimum safe LSN is greater than the given restart_lsn, use it as
+	 * the initial restart_lsn for the newly synced slot.
+	 */
+	if (min_safe_lsn > restart_lsn)
+		restart_lsn = min_safe_lsn;
 
-			GetWalRcvFlushRecPtr(NULL, &cur_timeline);
-			oldest_segno = XLogGetOldestSegno(cur_timeline);
-		}
+	SpinLockAcquire(&slot->mutex);
+	slot->data.restart_lsn = restart_lsn;
+	SpinLockRelease(&slot->mutex);
 
-		elog(DEBUG1, "segno: " UINT64_FORMAT " of purposed restart_lsn for the synced slot, oldest_segno: " UINT64_FORMAT " available",
-			 segno, oldest_segno);
+	ReplicationSlotsComputeRequiredLSN();
 
-		/*
-		 * If all required WAL is still there, great, otherwise retry. The
-		 * slot should prevent further removal of WAL, unless there's a
-		 * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
-		 * the new restart_lsn above, so normally we should never need to loop
-		 * more than twice.
-		 */
-		if (segno >= oldest_segno)
-			break;
+	XLByteToSeg(slot->data.restart_lsn, segno, wal_segment_size);
+	if (XLogGetLastRemovedSegno() >= segno)
+		elog(ERROR, "WAL required by replication slot %s has been removed concurrently",
+			 NameStr(slot->data.name));
 
-		/* Retry using the location of the oldest wal segment */
-		XLogSegNoOffsetToRecPtr(oldest_segno, 0, wal_segment_size, restart_lsn);
-	}
+	LWLockRelease(ReplicationSlotAllocationLock);
 }
 
 /*
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 0591a885dd1..fdfb572467b 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -226,6 +226,7 @@ extern XLogSegNo XLogGetLastRemovedSegno(void);
 extern XLogSegNo XLogGetOldestSegno(TimeLineID tli);
 extern void XLogSetAsyncXactLSN(XLogRecPtr asyncXactLSN);
 extern void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn);
+extern XLogRecPtr XLogGetReplicationSlotMinimumLSN(void);
 
 extern void xlog_redo(struct XLogReaderState *record);
 extern void xlog_desc(StringInfo buf, struct XLogReaderState *record);
-- 
2.51.1.windows.1

