From ab881009fca98dd7e5c3d9c3ee21e618ffec6325 Mon Sep 17 00:00:00 2001
From: Nikolay Samokhvalov <nik@postgres.ai>
Date: Sat, 5 Sep 2026 09:51:24 +0500
Subject: [PATCH v2 1/2] Pause recovery before invalidating logical slots

An archive-fed standby cannot send hot_standby_feedback to the primary.
Catalog cleanup records can therefore invalidate a local logical
replication slot before its consumer has decoded the WAL already
available on the standby.

Add recovery_pause_on_logical_slot_conflict.  When enabled, recovery
pauses before such invalidation and resumes automatically after normal
logical decoding has advanced the slot's effective xmin horizons.  An
explicit replay-resume request retains the default invalidation
behavior.  Do not alter slot horizons in the startup process, since
confirmed_flush_lsn does not establish that older catalog rows are no
longer needed.

Use the normal slot-invalidation predicate for both xmin horizons and
for conflicts with shared catalogs.  Do not pause before Hot Standby is
active or for synchronized slots.  Distinguish pause requests so that
auto-resume cannot override a separate pause request.

Discussion: https://www.postgresql.org/message-id/flat/967DBEA9-E8B7-4705-AD36-447D839AACA9%40yandex-team.ru
---
 doc/src/sgml/config.sgml                      |  40 ++++++
 src/backend/access/transam/xlogrecovery.c     | 116 ++++++++++++++----
 src/backend/replication/slot.c                |  49 ++++++++
 src/backend/storage/ipc/standby.c             |  69 +++++++++++
 src/backend/utils/misc/guc_parameters.dat     |   8 ++
 src/backend/utils/misc/postgresql.conf.sample |   3 +
 src/include/access/xlogrecovery.h             |   9 +-
 src/include/replication/slot.h                |   2 +
 8 files changed, 269 insertions(+), 27 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 0165eb9ec02..95824cdcba4 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5680,6 +5680,46 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
         This parameter can only be set in the <filename>postgresql.conf</filename>
         file or on the server command line.
        </para>
+     </listitem>
+     </varlistentry>
+
+     <varlistentry id="guc-recovery-pause-on-logical-slot-conflict" xreflabel="recovery_pause_on_logical_slot_conflict">
+      <term><varname>recovery_pause_on_logical_slot_conflict</varname> (<type>boolean</type>)
+      <indexterm>
+       <primary><varname>recovery_pause_on_logical_slot_conflict</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        When enabled, recovery pauses before replay of a catalog cleanup
+        record would invalidate a local logical replication slot.  This gives
+        the slot owner an opportunity to consume already-replayed WAL so that
+        logical decoding can advance the slot's <structfield>xmin</structfield>
+        horizons.  Recovery resumes automatically once no local logical slot
+        still has a conflicting horizon.  It can also be resumed explicitly
+        with <function>pg_wal_replay_resume</function>, or by promoting the
+        standby.
+       </para>
+       <para>
+        When recovery resumes, each affected slot is checked again and is
+        invalidated if either its <structfield>xmin</structfield> or
+        <structfield>catalog_xmin</structfield> still conflicts with the
+        cleanup record.  Thus, resuming without allowing these horizons to
+        advance preserves the default invalidation behavior.  Synchronized
+        logical slots are not protected by this setting because they are
+        managed by the slot synchronization worker.
+       </para>
+       <para>
+        Pausing recovery does not guarantee that a slot can be preserved.
+        If logical decoding cannot derive a sufficiently recent safe horizon
+        from the WAL already replayed, the slot owner must either drop the
+        slot or resume recovery and allow the slot to be invalidated.
+       </para>
+       <para>
+        This parameter is disabled by default.  It can only be set in the
+        <filename>postgresql.conf</filename> file or on the server command
+        line.
+       </para>
       </listitem>
      </varlistentry>
 
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index acac97e89d3..dd20348fe5f 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -97,6 +97,9 @@ char	   *recoveryTargetName;
 XLogRecPtr	recoveryTargetLSN;
 int			recovery_min_apply_delay = 0;
 
+/* Pause recovery before invalidating local logical replication slots. */
+bool		recovery_pause_on_logical_slot_conflict = false;
+
 /* options formerly taken from recovery.conf for XLOG streaming */
 char	   *PrimaryConnInfo = NULL;
 char	   *PrimarySlotName = NULL;
@@ -1652,6 +1655,7 @@ PerformWalRecovery(void)
 	XLogRecoveryCtl->recoveryLastXTime = 0;
 	XLogRecoveryCtl->currentChunkStartTime = 0;
 	XLogRecoveryCtl->recoveryPauseState = RECOVERY_NOT_PAUSED;
+	XLogRecoveryCtl->recoveryPauseGeneration = 0;
 	SpinLockRelease(&XLogRecoveryCtl->info_lck);
 
 	/* Also ensure XLogReceiptTime has a sane value */
@@ -2932,28 +2936,7 @@ recoveryPausesHere(bool endOfRecovery)
 				(errmsg("recovery has paused"),
 				 errhint("Execute pg_wal_replay_resume() to continue.")));
 
-	/* loop until recoveryPauseState is set to RECOVERY_NOT_PAUSED */
-	while (GetRecoveryPauseState() != RECOVERY_NOT_PAUSED)
-	{
-		ProcessStartupProcInterrupts();
-		if (CheckForStandbyTrigger())
-			return;
-
-		/*
-		 * If recovery pause is requested then set it paused.  While we are in
-		 * the loop, user might resume and pause again so set this every time.
-		 */
-		ConfirmRecoveryPaused();
-
-		/*
-		 * We wait on a condition variable that will wake us as soon as the
-		 * pause ends, but we use a timeout so we can check the above exit
-		 * condition periodically too.
-		 */
-		ConditionVariableTimedSleep(&XLogRecoveryCtl->recoveryNotPausedCV, 1000,
-									WAIT_EVENT_RECOVERY_PAUSE);
-	}
-	ConditionVariableCancelSleep();
+	(void) WaitForRecoveryToResume(NULL, NULL);
 }
 
 /*
@@ -3070,27 +3053,68 @@ GetRecoveryPauseState(void)
 }
 
 /*
- * Set the recovery pause state.
+ * Set the recovery pause state.  A pause request returns a token if it
+ * introduced the pause, or zero if recovery was already paused.
  *
  * If recovery pause is requested then sets the recovery pause state to
  * 'pause requested' if it is not already 'paused'.  Otherwise, sets it
  * to 'not paused' to resume the recovery.  The recovery pause will be
  * confirmed by the ConfirmRecoveryPaused.
  */
-void
+uint64
 SetRecoveryPause(bool recoveryPause)
 {
+	uint64		pause_token = 0;
+
 	SpinLockAcquire(&XLogRecoveryCtl->info_lck);
 
 	if (!recoveryPause)
 		XLogRecoveryCtl->recoveryPauseState = RECOVERY_NOT_PAUSED;
-	else if (XLogRecoveryCtl->recoveryPauseState == RECOVERY_NOT_PAUSED)
-		XLogRecoveryCtl->recoveryPauseState = RECOVERY_PAUSE_REQUESTED;
+	else
+	{
+		XLogRecoveryCtl->recoveryPauseGeneration++;
+		if (XLogRecoveryCtl->recoveryPauseGeneration == 0)
+			XLogRecoveryCtl->recoveryPauseGeneration++;
+
+		if (XLogRecoveryCtl->recoveryPauseState == RECOVERY_NOT_PAUSED)
+		{
+			XLogRecoveryCtl->recoveryPauseState = RECOVERY_PAUSE_REQUESTED;
+			pause_token = XLogRecoveryCtl->recoveryPauseGeneration;
+		}
+	}
 
 	SpinLockRelease(&XLogRecoveryCtl->info_lck);
 
 	if (!recoveryPause)
 		ConditionVariableBroadcast(&XLogRecoveryCtl->recoveryNotPausedCV);
+
+	return pause_token;
+}
+
+/*
+ * Resume recovery if no later pause request superseded pause_token.  Return
+ * true if recovery was already unpaused or this call resumed it.
+ */
+bool
+ResumeRecoveryPauseIfUnchanged(uint64 pause_token)
+{
+	bool		resumed = false;
+
+	SpinLockAcquire(&XLogRecoveryCtl->info_lck);
+	if (XLogRecoveryCtl->recoveryPauseState == RECOVERY_NOT_PAUSED)
+		resumed = true;
+	else if (pause_token != 0 &&
+		XLogRecoveryCtl->recoveryPauseGeneration == pause_token)
+	{
+		XLogRecoveryCtl->recoveryPauseState = RECOVERY_NOT_PAUSED;
+		resumed = true;
+	}
+	SpinLockRelease(&XLogRecoveryCtl->info_lck);
+
+	if (resumed)
+		ConditionVariableBroadcast(&XLogRecoveryCtl->recoveryNotPausedCV);
+
+	return resumed;
 }
 
 /*
@@ -3107,6 +3131,46 @@ ConfirmRecoveryPaused(void)
 	SpinLockRelease(&XLogRecoveryCtl->info_lck);
 }
 
+/*
+ * Wait until shared recoveryPauseState is set to RECOVERY_NOT_PAUSED, a
+ * promotion is requested, or the optional predicate permits auto-resume.
+ * Return true only in the last case.
+ */
+bool
+WaitForRecoveryToResume(RecoveryPauseResumePredicate predicate, void *arg)
+{
+	bool		auto_resume = false;
+
+	Assert(AmStartupProcess() || !IsUnderPostmaster);
+
+	while (GetRecoveryPauseState() != RECOVERY_NOT_PAUSED)
+	{
+		ProcessStartupProcInterrupts();
+		if (CheckForStandbyTrigger())
+			break;
+		if (predicate != NULL && predicate(arg))
+		{
+			auto_resume = true;
+			break;
+		}
+
+		/*
+		 * While we are in the loop, a user might resume and pause again, so
+		 * confirm the pause every time.
+		 */
+		ConfirmRecoveryPaused();
+
+		/*
+		 * The timeout lets us check for interrupts and promotion requests.
+		 */
+		ConditionVariableTimedSleep(&XLogRecoveryCtl->recoveryNotPausedCV, 1000,
+									WAIT_EVENT_RECOVERY_PAUSE);
+	}
+	ConditionVariableCancelSleep();
+
+	return auto_resume;
+}
+
 
 /*
  * Attempt to read the next XLOG record.
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 63ce6d27885..5a859290c22 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1960,6 +1960,55 @@ DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s,
 	return RS_INVAL_NONE;
 }
 
+/*
+ * Return true if a local logical slot would be invalidated by the given
+ * snapshot conflict horizon.
+ *
+ * Synchronized slots are managed by the slot synchronization worker and
+ * cannot be consumed or dropped on the standby, so callers cannot resolve
+ * their conflicts locally.
+ */
+bool
+ReplicationSlotsHaveLogicalHorizonConflict(Oid dboid,
+										   TransactionId snapshotConflictHorizon)
+{
+	bool		conflict = false;
+	TimestampTz inactive_since = 0;
+
+	Assert(TransactionIdIsValid(snapshotConflictHorizon));
+
+	if (IsBinaryUpgrade)
+		return false;
+
+	if (max_replication_slots == 0 && max_repack_replication_slots == 0)
+		return false;
+
+	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+	for (int i = 0;
+		 i < max_replication_slots + max_repack_replication_slots;
+		 i++)
+	{
+		ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+		if (!s->in_use)
+			continue;
+
+		SpinLockAcquire(&s->mutex);
+		if (!s->data.synced && s->data.invalidated == RS_INVAL_NONE &&
+			DetermineSlotInvalidationCause(RS_INVAL_HORIZON, s, 0, dboid,
+										   snapshotConflictHorizon,
+										   &inactive_since, 0) == RS_INVAL_HORIZON)
+			conflict = true;
+		SpinLockRelease(&s->mutex);
+
+		if (conflict)
+			break;
+	}
+	LWLockRelease(ReplicationSlotControlLock);
+
+	return conflict;
+}
+
 /*
  * Helper for InvalidateObsoleteReplicationSlots
  *
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 7f011e04990..1998d914768 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -63,6 +63,12 @@ typedef struct RecoveryLockXidEntry
 	struct RecoveryLockEntry *head; /* chain head */
 } RecoveryLockXidEntry;
 
+typedef struct LogicalSlotConflict
+{
+	Oid			dboid;
+	TransactionId snapshotConflictHorizon;
+}			LogicalSlotConflict;
+
 static HTAB *RecoveryLockHash = NULL;
 static HTAB *RecoveryLockXidHash = NULL;
 
@@ -79,6 +85,9 @@ static void SendRecoveryConflictWithBufferPin(RecoveryConflictReason reason);
 static XLogRecPtr LogCurrentRunningXacts(RunningTransactions CurrRunningXacts);
 static void LogAccessExclusiveLocks(int nlocks, xl_standby_lock *locks);
 static const char *get_recovery_conflict_desc(RecoveryConflictReason reason);
+static void MaybePauseOnLogicalSlotConflict(Oid dboid,
+											TransactionId snapshotConflictHorizon);
+static bool LogicalSlotConflictHasResolved(void *arg);
 
 /*
  * InitRecoveryTransactionEnvironment
@@ -503,8 +512,68 @@ ResolveRecoveryConflictWithSnapshot(TransactionId snapshotConflictHorizon,
 	 * reached, e.g. due to using a physical replication slot.
 	 */
 	if (IsLogicalDecodingEnabled() && isCatalogRel)
+	{
+		MaybePauseOnLogicalSlotConflict(locator.dbOid, snapshotConflictHorizon);
 		InvalidateObsoleteReplicationSlots(RS_INVAL_HORIZON, 0, locator.dbOid,
 										   snapshotConflictHorizon);
+	}
+}
+
+/*
+ * Pause before replay invalidates a local logical replication slot.
+ *
+ * The slot owner can consume the WAL that has already been replayed.  Logical
+ * decoding will then advance the slot's effective xmin horizons if it is safe
+ * to do so.  On resume, InvalidateObsoleteReplicationSlots() rechecks the
+ * current horizons and either leaves the slot intact or invalidates it as
+ * usual.  We must not advance the horizons here: confirmed_flush alone does
+ * not prove that logical decoding no longer needs older catalog rows.
+ *
+ * Do not pause before Hot Standby is active, since no client could connect to
+ * resolve the conflict.  The normal invalidation behavior is retained then.
+ */
+static void
+MaybePauseOnLogicalSlotConflict(Oid dboid, TransactionId snapshotConflictHorizon)
+{
+	LogicalSlotConflict conflict;
+	bool		auto_resume;
+	uint64		pause_token;
+
+	if (!recovery_pause_on_logical_slot_conflict || !HotStandbyActive())
+		return;
+
+	if (!ReplicationSlotsHaveLogicalHorizonConflict(dboid,
+													snapshotConflictHorizon))
+		return;
+
+	ereport(LOG,
+			(errmsg("recovery has paused to avoid invalidating a logical replication slot"),
+			 errdetail("WAL replay at %X/%X has a snapshot conflict horizon of %u.",
+					   LSN_FORMAT_ARGS(GetXLogReplayRecPtr(NULL)),
+					   snapshotConflictHorizon),
+			 errhint("Allow the affected slot to advance its xmin horizons, drop it, or accept its invalidation, then execute pg_wal_replay_resume().")));
+
+	conflict.dboid = dboid;
+	conflict.snapshotConflictHorizon = snapshotConflictHorizon;
+
+	pause_token = SetRecoveryPause(true);
+	auto_resume = WaitForRecoveryToResume(LogicalSlotConflictHasResolved,
+										  &conflict);
+
+	/* Do not override an explicit pause requested before or during our wait. */
+	if (auto_resume && !ResumeRecoveryPauseIfUnchanged(pause_token))
+		ereport(LOG,
+				(errmsg("recovery remains paused due to a separate pause request")));
+}
+
+/* Return true once no locally consumable slot still blocks replay. */
+static bool
+LogicalSlotConflictHasResolved(void *arg)
+{
+	LogicalSlotConflict *conflict = arg;
+
+	return !ReplicationSlotsHaveLogicalHorizonConflict(conflict->dboid,
+													   conflict->snapshotConflictHorizon);
 }
 
 /*
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 3c5e16ad1e7..cb378a13004 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -2466,6 +2466,14 @@
   max => 'INT_MAX',
 },
 
+{ name => 'recovery_pause_on_logical_slot_conflict', type => 'bool',
+  context => 'PGC_SIGHUP', group => 'REPLICATION_STANDBY',
+  short_desc => 'Pauses recovery before invalidating a local logical replication slot.',
+  long_desc => 'Allows the slot to consume already-replayed WAL and advance its xmin horizons before recovery resumes automatically.',
+  variable => 'recovery_pause_on_logical_slot_conflict',
+  boot_val => 'false',
+},
+
 { name => 'recovery_prefetch', type => 'enum', context => 'PGC_SIGHUP', group => 'WAL_RECOVERY',
   short_desc => 'Prefetch referenced blocks during recovery.',
   long_desc => 'Look ahead in the WAL to find references to uncached data.',
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index e759f06b50f..2e06b7c17f9 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -406,6 +406,9 @@
 #wal_retrieve_retry_interval = 5s       # time to wait before retrying to
                                         # retrieve WAL after a failed attempt
 #recovery_min_apply_delay = 0           # minimum delay for applying changes during recovery
+#recovery_pause_on_logical_slot_conflict = off
+                                        # pause recovery before invalidating a
+                                        # local logical replication slot
 #sync_replication_slots = off           # enables slot synchronization on the physical standby from the primary
 
 # - Subscribers -
diff --git a/src/include/access/xlogrecovery.h b/src/include/access/xlogrecovery.h
index 8786b6d3a8e..beda1e1bc56 100644
--- a/src/include/access/xlogrecovery.h
+++ b/src/include/access/xlogrecovery.h
@@ -60,6 +60,8 @@ typedef enum RecoveryPauseState
 	RECOVERY_PAUSED,			/* recovery is paused */
 } RecoveryPauseState;
 
+typedef bool (*RecoveryPauseResumePredicate) (void *arg);
+
 /*
  * Shared-memory state for WAL recovery.
  */
@@ -124,6 +126,7 @@ typedef struct XLogRecoveryCtlData
 	TimestampTz currentChunkStartTime;
 	/* Recovery pause state */
 	RecoveryPauseState recoveryPauseState;
+	uint64		recoveryPauseGeneration;
 	ConditionVariable recoveryNotPausedCV;
 
 	slock_t		info_lck;		/* locks shared variables shown above */
@@ -135,6 +138,7 @@ extern PGDLLIMPORT XLogRecoveryCtlData *XLogRecoveryCtl;
 extern PGDLLIMPORT bool recoveryTargetInclusive;
 extern PGDLLIMPORT int recoveryTargetAction;
 extern PGDLLIMPORT int recovery_min_apply_delay;
+extern PGDLLIMPORT bool recovery_pause_on_logical_slot_conflict;
 extern PGDLLIMPORT char *PrimaryConnInfo;
 extern PGDLLIMPORT char *PrimarySlotName;
 extern PGDLLIMPORT char *recoveryRestoreCommand;
@@ -218,7 +222,10 @@ extern void RemovePromoteSignalFiles(void);
 extern bool HotStandbyActive(void);
 extern XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI);
 extern RecoveryPauseState GetRecoveryPauseState(void);
-extern void SetRecoveryPause(bool recoveryPause);
+extern uint64 SetRecoveryPause(bool recoveryPause);
+extern bool ResumeRecoveryPauseIfUnchanged(uint64 pause_token);
+extern bool WaitForRecoveryToResume(RecoveryPauseResumePredicate predicate,
+									void *arg);
 extern void GetXLogReceiptTime(TimestampTz *rtime, bool *fromStream);
 extern TimestampTz GetLatestXTime(void);
 extern TimestampTz GetCurrentChunkReplayStartTime(void);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 9b29444cbca..9fcfb9123c3 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -365,6 +365,8 @@ extern bool InvalidateObsoleteReplicationSlots(uint32 possible_causes,
 											   XLogSegNo oldestSegno,
 											   Oid dboid,
 											   TransactionId snapshotConflictHorizon);
+extern bool ReplicationSlotsHaveLogicalHorizonConflict(Oid dboid,
+													   TransactionId snapshotConflictHorizon);
 extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
 extern int	ReplicationSlotIndex(ReplicationSlot *slot);
 extern bool ReplicationSlotName(int index, Name name);
-- 
That's all, folks. May the source be with you.

