From 2b5c82677f8e40b9071180c78b708260fa084678 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Wed, 11 Jun 2025 15:48:09 +0800
Subject: [PATCH v41 3/6] 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 | 10 ++++
 src/backend/replication/logical/worker.c   | 64 +++++++++++++++-------
 3 files changed, 58 insertions(+), 21 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 4c870b5e806..504f7a01ef0 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5423,7 +5423,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 a988969636e..5261696aaa1 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -1391,6 +1391,16 @@ compute_min_nonremovable_xid(LogicalRepWorker *worker,
 	}
 	else
 	{
+		/*
+		 * Drop the invalidated slot and re-create it if the worker for a
+		 * subscription with retain_conflict_info enabled has stopped,
+		 * indicating it might restart to request retention of information
+		 * necessary for conflict detection.
+		 */
+		if (MyReplicationSlot &&
+			MyReplicationSlot->data.invalidated == RS_INVAL_CONFLICT_RETENTION_DURATION)
+			ReplicationSlotDropAcquired();
+
 		/*
 		 * Create a replication slot to retain information necessary for
 		 * conflict detection such as dead tuples, commit timestamps, and
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b80e028eb9d..a16c7d3a973 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -476,6 +476,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);
@@ -4139,10 +4141,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;
 }
 
@@ -4422,6 +4420,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
@@ -4429,6 +4446,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",
@@ -4470,9 +4488,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)
@@ -4501,19 +4518,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