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

From b3c112ff3dde08305eca415c48be990b9f16c1a7 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Tue, 3 Jun 2025 18:52:15 +0800
Subject: [PATCH v33 5/7] 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 | 41 ++++++++++-----
 src/backend/replication/logical/worker.c   | 60 +++++++++++++++-------
 3 files changed, 74 insertions(+), 32 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 8712894a791..2f8dfb10cc6 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5419,7 +5419,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 f5f3c87042b..d351d02d878 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -108,7 +108,7 @@ static int	logicalrep_pa_worker_count(Oid subid);
 static void logicalrep_launcher_attach_dshmem(void);
 static void ApplyLauncherSetWorkerStartTime(Oid subid, TimestampTz start_time);
 static TimestampTz ApplyLauncherGetWorkerStartTime(Oid subid);
-static void create_conflict_slot_if_not_exists(void);
+static void create_conflict_slot_if_not_exists(bool recreate_if_invalid);
 static void advance_conflict_slot_xmin(TransactionId new_xmin);
 static void drop_conflict_slot_if_exists(void);
 static void invalidate_conflict_slot(void);
@@ -322,7 +322,7 @@ logicalrep_worker_launch(LogicalRepWorkerType wtype,
 	bool		is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
 	bool		is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
 	bool		retaining_conflict_info = (MyReplicationSlot &&
-										   MyReplicationSlot->data.invalidated != RS_INVAL_NONE);
+										   MyReplicationSlot->data.invalidated == RS_INVAL_NONE);
 
 	/*----------
 	 * Sanity checks:
@@ -458,7 +458,7 @@ retry:
 	worker->oldest_nonremovable_xid = retaining_conflict_info
 									? MyReplicationSlot->data.xmin
 									: InvalidTransactionId;
-	worker->stop_conflict_info_retention = retaining_conflict_info;
+	worker->stop_conflict_info_retention = !retaining_conflict_info;
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -1206,7 +1206,7 @@ ApplyLauncherMain(Datum main_arg)
 				 * prevent it from unnecessarily maintaining its
 				 * oldest_nonremovable_xid.
 				 */
-				create_conflict_slot_if_not_exists();
+				create_conflict_slot_if_not_exists(!sub->enabled);
 			}
 
 			if (!sub->enabled)
@@ -1259,9 +1259,14 @@ ApplyLauncherMain(Datum main_arg)
 			/*
 			 * The worker has not yet started, so there is no valid
 			 * non-removable transaction ID available for advancement.
+			 * Additionally, recreate the slot to enable the new worker to
+			 * resume retaining conflict information.
 			 */
 			if (sub->retainconflictinfo)
+			{
 				can_advance_xmin = false;
+				create_conflict_slot_if_not_exists(true);
+			}
 
 			/*
 			 * If the worker is eligible to start now, launch it.  Otherwise,
@@ -1344,21 +1349,31 @@ ApplyLauncherMain(Datum main_arg)
  * conflict detection, if not yet.
  */
 static void
-create_conflict_slot_if_not_exists(void)
+create_conflict_slot_if_not_exists(bool recreate_if_invalid)
 {
 	TransactionId xmin_horizon;
 
-	/* Exit early if the replication slot is already created and acquired */
-	if (MyReplicationSlot)
-		return;
-
-	/* If the replication slot exists, acquire it and exit */
-	if (SearchNamedReplicationSlot(CONFLICT_DETECTION_SLOT, true))
-	{
+	/*
+	 * Acquire the replication slot if it exists and hasn't been acquired yet.
+	 */
+	if (!MyReplicationSlot &&
+		SearchNamedReplicationSlot(CONFLICT_DETECTION_SLOT, true))
 		ReplicationSlotAcquire(CONFLICT_DETECTION_SLOT, true, false);
+
+	/* Drop the slot if it's invalidated and recreate_if_invalid is true */
+	if (MyReplicationSlot &&
+		MyReplicationSlot->data.invalidated != RS_INVAL_NONE &&
+		recreate_if_invalid)
+		drop_conflict_slot_if_exists();
+
+	/* 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 c97c6937a02..a85858df977 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -482,6 +482,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);
@@ -4145,10 +4147,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;
 }
 
@@ -4428,6 +4426,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
@@ -4435,6 +4452,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",
@@ -4476,9 +4494,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)
@@ -4507,19 +4524,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));
+	/*
+	 * 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);
+		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();
+		/* Notify launcher to invalidate the conflict slot */
+		ApplyLauncherWakeup();
+	}
 
 	reset_conflict_info_fields(rci_data);
 
-- 
2.30.0.windows.2