v35-0005-Re-create-the-replication-slot-if-the-conflict-r.patch

From c4ca4ed73de7c257f6fc5ed929ab4ddbc5609aea Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Mon, 9 Jun 2025 19:13:55 +0800
Subject: [PATCH v35 5/8] Re-create the replication slot if the conflict
 retention duration reduced

The patch allows the launcher to drop and re-create the invalidated slot, if at
least one apply worker has confirmed that the retention duration is now within
the max_conflict_retention_duration.
---
 doc/src/sgml/config.sgml                   |  5 +-
 src/backend/replication/logical/launcher.c |  8 +++
 src/backend/replication/logical/worker.c   | 64 +++++++++++++++-------
 3 files changed, 56 insertions(+), 21 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 355a4ac2977..d97eef1216d 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5421,7 +5421,10 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
         <literal>max_conflict_retention_duration</literal>. If the replication
         slot is invalidated, you can disable
         <literal>retain_conflict_info</literal> and re-enable it after
-        confirming this replication slot has been dropped.
+        confirming this replication slot has been dropped. Alternatively, the
+        invalidated slot will be automatically dropped and re-created once the
+        apply worker confirms that the retention duration is within the
+        specified limit.
        </para>
        <para>
         This option is effective only if a subscription with
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 611f58b237d..d1ecb6fb780 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -1401,6 +1401,14 @@ create_conflict_slot_if_not_exists(void)
 		return;
 	}
 
+	/* Return if a valid replication slot is already created and acquired */
+	if (MyReplicationSlot)
+		return;
+
+	/*
+	 * Create a new replication slot if none exists or the previous one was
+	 * dropped.
+	 */
 	ReplicationSlotCreate(CONFLICT_DETECTION_SLOT, false,
 						  RS_PERSISTENT, false, false, false);
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ebf0f007e5a..1c968b13d2a 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -483,6 +483,8 @@ static void apply_handle_tuple_routing(ApplyExecutionData *edata,
 									   LogicalRepTupleData *newtup,
 									   CmdType operation);
 
+static void apply_worker_exit(void);
+
 /* Functions for skipping changes */
 static void maybe_start_skipping_changes(XLogRecPtr finish_lsn);
 static void stop_skipping_changes(void);
@@ -4146,10 +4148,6 @@ can_advance_nonremovable_xid(RetainConflictInfoData *rci_data)
 	if (!MySubscription->retainconflictinfo)
 		return false;
 
-	/* No need to advance if we have already stopped retaining */
-	if (MyLogicalRepWorker->stop_conflict_info_retention)
-		return false;
-
 	return true;
 }
 
@@ -4429,6 +4427,25 @@ wait_for_local_flush(RetainConflictInfoData *rci_data)
 	if (last_flushpos < rci_data->remote_lsn)
 		return;
 
+	/*
+	 * If conflict info retention was previously stopped due to a timeout, and
+	 * the time required to advance the non-removable transaction ID has now
+	 * decreased to within acceptable limits, log a message and exit. This
+	 * allows the launcher to recreate the replication slot prior to
+	 * restarting the worker.
+	 */
+	if (MyLogicalRepWorker->stop_conflict_info_retention)
+	{
+		ereport(LOG,
+				errmsg("logical replication worker for subscription \"%s\" will restart to resume retaining conflict information",
+					   MySubscription->name),
+				errdetail("The time spent applying changes up to LSN %X/%X is now within the maximum limit of %u ms.",
+						  LSN_FORMAT_ARGS(rci_data->remote_lsn),
+						  max_conflict_retention_duration));
+
+		apply_worker_exit();
+	}
+
 	/*
 	 * Reaching here means the remote WAL position has been received, and all
 	 * transactions up to that position on the publisher have been applied and
@@ -4436,6 +4453,7 @@ wait_for_local_flush(RetainConflictInfoData *rci_data)
 	 */
 	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
 	MyLogicalRepWorker->oldest_nonremovable_xid = rci_data->candidate_xid;
+	MyLogicalRepWorker->stop_conflict_info_retention = false;
 	SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
 	elog(DEBUG2, "confirmed flush up to remote lsn %X/%X: new oldest_nonremovable_xid %u",
@@ -4477,9 +4495,8 @@ reset_conflict_info_fields(RetainConflictInfoData *rci_data)
  * LogicalRepWorker->stop_conflict_info_retention to true, notify the launcher to
  * invalidate the slot, and return true. Return false otherwise.
  *
- * Currently, the retention will not resume automatically unless user manually
- * disables retain_conflict_info and re-enables it after confirming that the
- * replication slot has been dropped.
+ * The retention will resume automatically if the worker has confirmed that the
+ * retention duration is now within the max_conflict_retention_duration.
  */
 static bool
 should_stop_conflict_info_retention(RetainConflictInfoData *rci_data)
@@ -4508,19 +4525,26 @@ should_stop_conflict_info_retention(RetainConflictInfoData *rci_data)
 									max_conflict_retention_duration))
 		return false;
 
-	ereport(LOG,
-			errmsg("logical replication worker for subscription \"%s\" will stop retaining conflict information",
-				   MySubscription->name),
-			errdetail("The time spent advancing the non-removable transaction ID has exceeded the maximum limit of %u ms.",
-					  max_conflict_retention_duration));
-
-	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
-	MyLogicalRepWorker->oldest_nonremovable_xid = InvalidTransactionId;
-	MyLogicalRepWorker->stop_conflict_info_retention = true;
-	SpinLockRelease(&MyLogicalRepWorker->relmutex);
-
-	/* Notify launcher to invalidate the conflict slot */
-	ApplyLauncherWakeup();
+	/*
+	 * Log a message and reset relevant data when the worker is about to stop
+	 * retaining conflict information.
+	 */
+	if (!MyLogicalRepWorker->stop_conflict_info_retention)
+	{
+		ereport(LOG,
+				errmsg("logical replication worker for subscription \"%s\" will stop retaining conflict information",
+					   MySubscription->name),
+				errdetail("The time spent advancing the non-removable transaction ID has exceeded the maximum limit of %u ms.",
+						  max_conflict_retention_duration));
+
+		SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+		MyLogicalRepWorker->oldest_nonremovable_xid = InvalidTransactionId;
+		MyLogicalRepWorker->stop_conflict_info_retention = true;
+		SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+		/* Notify launcher to invalidate the conflict slot */
+		ApplyLauncherWakeup();
+	}
 
 	reset_conflict_info_fields(rci_data);
 
-- 
2.30.0.windows.2