From ea45e57d44cbe15d9166e157e7601e83184ee477 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Thu, 13 Feb 2025 14:08:11 +0800
Subject: [PATCH v28 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 |  37 +++---
 src/backend/replication/logical/worker.c   | 140 ++++++++-------------
 3 files changed, 77 insertions(+), 105 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 6decdc89eef..04067b0d437 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5239,7 +5239,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 af4cd864d97..4d92efd9e77 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -461,7 +461,8 @@ retry:
 	worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
 	worker->parallel_apply = is_parallel_apply_worker;
 	worker->oldest_nonremovable_xid = InvalidFullTransactionId;
-	worker->stop_conflict_retention = false;
+	worker->stop_conflict_retention = (MyReplicationSlot &&
+									   MyReplicationSlot->data.invalidated != RS_INVAL_NONE);
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -1300,27 +1301,20 @@ ApplyLauncherMain(Datum main_arg)
 			}
 		}
 
-		/*
-		 * Do nothing if the replication slot is invalidated due to conflict
-		 * retention duration.
-		 */
-		if (nretain_conflict_info &&
-			MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
-		{
-			Assert(MyReplicationSlot->data.invalidated == RS_INVAL_CONFLICT_RETENTION_DURATION);
-		}
-
 		/*
 		 * Invalidate the conflict slot if all workers with
 		 * retain_conflict_info enabled have stopped further conflict
 		 * retention.
 		 */
-		else if (nstop_retention && nretain_conflict_info == nstop_retention)
+		if (nstop_retention && nretain_conflict_info == nstop_retention)
 		{
-			ReplicationSlotRelease();
-			InvalidateObsoleteReplicationSlots(RS_INVAL_CONFLICT_RETENTION_DURATION,
-											   InvalidXLogRecPtr, InvalidOid,
-											   InvalidTransactionId);
+			if (MyReplicationSlot->data.invalidated == RS_INVAL_NONE)
+			{
+				ReplicationSlotRelease();
+				InvalidateObsoleteReplicationSlots(RS_INVAL_CONFLICT_RETENTION_DURATION,
+												   InvalidXLogRecPtr, InvalidOid,
+												   InvalidTransactionId);
+			}
 		}
 
 		/*
@@ -1332,6 +1326,17 @@ ApplyLauncherMain(Datum main_arg)
 		{
 			bool		updated = false;
 
+			/*
+			 * Re-create the replication slot if it was invalidated because
+			 * all workers stopped conflict retention, and an apply worker has
+			 * now resumed the process.
+			 */
+			if (MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
+			{
+				drop_conflict_slot_if_exists();
+				create_conflict_slot_if_not_exists();
+			}
+
 			if (can_advance_xmin)
 				updated = advance_conflict_slot_xmin(xmin);
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 6b1746e0e2b..548ebb37593 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -454,7 +454,6 @@ static void wait_for_publisher_status(RetainConflictInfoData *data,
 static void wait_for_local_flush(RetainConflictInfoData *data);
 static void adjust_xid_advance_interval(RetainConflictInfoData *data,
 										bool new_xid_found);
-static void update_conflict_retention_status(void);
 
 static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
 static void apply_handle_insert_internal(ApplyExecutionData *edata,
@@ -4126,10 +4125,6 @@ maybe_advance_nonremovable_xid(RetainConflictInfoData *data,
 	if (!am_leader_apply_worker())
 		return;
 
-	/* Exit early if we have already stopped retaining */
-	if (MyLogicalRepWorker->stop_conflict_retention)
-		return;
-
 	switch (data->phase)
 	{
 		case RCI_GET_CANDIDATE_XID:
@@ -4292,7 +4287,9 @@ wait_for_publisher_status(RetainConflictInfoData *data, bool status_received)
 static void
 wait_for_local_flush(RetainConflictInfoData *data)
 {
+	TimestampTz now;
 	bool		stop_conflict_retention = false;
+	bool		status_changed;
 
 	Assert(!XLogRecPtrIsInvalid(data->remote_lsn) &&
 		   FullTransactionIdIsValid(data->candidate_xid));
@@ -4355,31 +4352,36 @@ wait_for_local_flush(RetainConflictInfoData *data)
 		data->flushpos_update_time = data->last_recv_time;
 	}
 
-	/* Check if changes up to the remote_lsn have been applied and flushed */
-	if (last_flushpos < data->remote_lsn)
-	{
-		TimestampTz now;
+	/*
+	 * Use last_recv_time when applying changes in the loop to avoid
+	 * unnecessary system time retrieval. If last_recv_time is not available,
+	 * obtain the current timestamp.
+	 */
+	now = data->last_recv_time ? data->last_recv_time : GetCurrentTimestamp();
 
-		/*
-		 * Use last_recv_time when applying changes in the loop to avoid
-		 * unnecessary system time retrieval. If last_recv_time is not
-		 * available, obtain the current timestamp.
-		 */
-		now = data->last_recv_time ? data->last_recv_time : GetCurrentTimestamp();
+	/*
+	 * If the wait time has exceeded the maximum limit
+	 * (max_conflict_retention_duration), stop tracking the non-removable
+	 * transaction ID by this apply worker.
+	 */
+	if (max_conflict_retention_duration &&
+		TimestampDifferenceExceeds(data->candidate_xid_time, now,
+								   max_conflict_retention_duration))
+		stop_conflict_retention = true;
 
-		/*
-		 * If the wait time has not exceeded the maximum limit
-		 * (max_conflict_retention_duration), continue waiting for the changes
-		 * to be applied. Otherwise, stop tracking the non-removable
-		 * transaction ID by this apply worker.
-		 */
-		if (!max_conflict_retention_duration ||
-			!TimestampDifferenceExceeds(data->candidate_xid_time, now,
-										max_conflict_retention_duration))
-			return;
+	/*
+	 * Return if the apply worker is retaining conflict information, and the
+	 * changes up to the remote_lsn are not yet applied and flushed.
+	 */
+	if (!stop_conflict_retention && last_flushpos < data->remote_lsn)
+		return;
 
-		stop_conflict_retention = true;
-	}
+	/*
+	 * Determine whether the apply worker is about to stop or resume conflict
+	 * retention.
+	 */
+	if (MyLogicalRepWorker->stop_conflict_retention != stop_conflict_retention)
+		status_changed = true;
 
 	if (!stop_conflict_retention)
 	{
@@ -4391,13 +4393,25 @@ wait_for_local_flush(RetainConflictInfoData *data)
 		 */
 		SpinLockAcquire(&MyLogicalRepWorker->relmutex);
 		MyLogicalRepWorker->oldest_nonremovable_xid = data->candidate_xid;
+		MyLogicalRepWorker->stop_conflict_retention = false;
 		SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
+		if (status_changed)
+			ereport(LOG,
+					errmsg("logical replication worker for subscription \"%s\" will 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(data->remote_lsn),
+							  max_conflict_retention_duration));
+
 		elog(DEBUG2, "confirmed remote flush up to %X/%X: new oldest_nonremovable_xid %u",
 			 LSN_FORMAT_ARGS(data->remote_lsn),
 			 XidFromFullTransactionId(data->candidate_xid));
+
+		/* Notify launcher to update the xmin of the conflict slot */
+		ApplyLauncherWakeup();
 	}
-	else
+	else if (status_changed)
 	{
 		/*
 		 * Reaching here means the time spent applying changes up to the
@@ -4405,9 +4419,9 @@ wait_for_local_flush(RetainConflictInfoData *data)
 		 * (max_conflict_retention_duration). So, we will stop retaining
 		 * conflict information.
 		 *
-		 * Currently, the retention will not resume automatically unless user
-		 * manually disable retain_conflict_info and re-enable 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.
 		 */
 		SpinLockAcquire(&MyLogicalRepWorker->relmutex);
 		MyLogicalRepWorker->oldest_nonremovable_xid = InvalidFullTransactionId;
@@ -4420,6 +4434,13 @@ wait_for_local_flush(RetainConflictInfoData *data)
 				errdetail("The time spent applying changes up to LSN %X/%X has exceeded the maximum limit of %u ms.",
 						  LSN_FORMAT_ARGS(data->remote_lsn),
 						  max_conflict_retention_duration));
+
+		/*
+		 * Notify launcher to either update the xmin of the conflict slot or
+		 * invalidate the slot if no other workers are retaining conflict
+		 * information.
+		 */
+		ApplyLauncherWakeup();
 	}
 
 	/* Notify launcher to update the xmin of the conflict slot */
@@ -4484,51 +4505,6 @@ adjust_xid_advance_interval(RetainConflictInfoData *data, bool new_xid_found)
 	}
 }
 
-/*
- * Update the conflict retention status for the current apply worker. It checks
- * whether the worker should stop retaining conflict information due to
- * invalidation of the replication slot ("pg_conflict_detection").
- *
- * Currently, the replication slot is invalidated only if the duration for
- * retaining conflict information exceeds the allowed maximum.
- */
-static void
-update_conflict_retention_status(void)
-{
-	ReplicationSlotInvalidationCause cause = RS_INVAL_NONE;
-	ReplicationSlot *slot;
-
-	/* Exit early if retaining conflict information is not required */
-	if (!MySubscription->retainconflictinfo)
-		return;
-
-	/*
-	 * Only the leader apply worker manages conflict retention (see
-	 * maybe_advance_nonremovable_xid() for details).
-	 */
-	if (!am_leader_apply_worker())
-		return;
-
-	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
-
-	slot = SearchNamedReplicationSlot(CONFLICT_DETECTION_SLOT, false);
-
-	if (slot)
-	{
-		SpinLockAcquire(&slot->mutex);
-		cause = slot->data.invalidated;
-		SpinLockRelease(&slot->mutex);
-
-		Assert(cause == RS_INVAL_NONE || cause == RS_INVAL_CONFLICT_RETENTION_DURATION);
-	}
-
-	LWLockRelease(ReplicationSlotControlLock);
-
-	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
-	MyLogicalRepWorker->stop_conflict_retention = cause != RS_INVAL_NONE;
-	SpinLockRelease(&MyLogicalRepWorker->relmutex);
-}
-
 /*
  * Exit routine for apply workers due to subscription parameter changes.
  */
@@ -4700,16 +4676,6 @@ maybe_reread_subscription(void)
 		CommitTransactionCommand();
 
 	MySubscriptionValid = true;
-
-	/*
-	 * Update worker status to avoid unnecessary conflict retention if the
-	 * replication slot ("pg_conflict_detection") was invalidated prior to
-	 * enabling the retain_conflict_info option. This is also necessary to
-	 * restart conflict retention if the user has disabled and subsequently
-	 * re-enabled the retain_conflict_info option, resulting in the
-	 * replication slot being recreated.
-	 */
-	update_conflict_retention_status();
 }
 
 /*
@@ -5348,8 +5314,6 @@ InitializeLogRepWorker(void)
 						MySubscription->name)));
 
 	CommitTransactionCommand();
-
-	update_conflict_retention_status();
 }
 
 /*
-- 
2.30.0.windows.2