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.

Author: Zhijie Hou

doc/src/sgml/config.sgml

@@ -5403,7 +5403,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

src/backend/replication/logical/launcher.c

@@ -449,7 +449,8 @@ logicalrep_worker_launch(LogicalRepWorkerType wtype,
 	worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
 	worker->parallel_apply = is_parallel_apply_worker;
 	worker->oldest_nonremovable_xid = InvalidFullTransactionId;
-	worker->stop_conflict_info_retention = false;
+	worker->stop_conflict_info_retention = (MyReplicationSlot &&
+									   MyReplicationSlot->data.invalidated != RS_INVAL_NONE);
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -1287,27 +1288,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);
+			}
 		}
 
 		/*
@@ -1316,6 +1310,17 @@ ApplyLauncherMain(Datum main_arg)
 		 */
 		else if (nretain_conflict_info)
 		{
+			/*
+			 * 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)
 				advance_conflict_slot_xmin(xmin);
 

src/backend/replication/logical/worker.c

@@ -456,7 +456,6 @@ static void reset_conflict_info_fields(RetainConflictInfoData *data);
 static bool should_stop_conflict_info_retention(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_info_retention)
-		return;
-
 	switch (data->phase)
 	{
 		case RCI_GET_CANDIDATE_XID:
@@ -4374,6 +4369,19 @@ wait_for_local_flush(RetainConflictInfoData *data)
 	if (last_flushpos < 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.
+	 */
+	if (MyLogicalRepWorker->stop_conflict_info_retention)
+		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));
+
 	/*
 	 * Reaching here means the remote WAL position has been received, and
 	 * all transactions up to that position on the publisher have been
@@ -4382,6 +4390,7 @@ wait_for_local_flush(RetainConflictInfoData *data)
 	 */
 	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
 	MyLogicalRepWorker->oldest_nonremovable_xid = data->candidate_xid;
+	MyLogicalRepWorker->stop_conflict_info_retention = false;
 	SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
 	elog(DEBUG2, "confirmed remote flush up to %X/%X: new oldest_nonremovable_xid %u",
@@ -4423,9 +4432,8 @@ reset_conflict_info_fields(RetainConflictInfoData *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 *data)
@@ -4450,19 +4458,26 @@ should_stop_conflict_info_retention(RetainConflictInfoData *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 = InvalidFullTransactionId;
-	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 = InvalidFullTransactionId;
+		MyLogicalRepWorker->stop_conflict_info_retention = true;
+		SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+		/* Notify launcher to invalidate the conflict slot */
+		ApplyLauncherWakeup();
+	}
 
 	reset_conflict_info_fields(data);
 
@@ -4510,51 +4525,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_info_retention = cause != RS_INVAL_NONE;
-	SpinLockRelease(&MyLogicalRepWorker->relmutex);
-}
-
 /*
  * Exit routine for apply workers due to subscription parameter changes.
  */
@@ -4726,16 +4696,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();
 }
 
 /*
@@ -5382,8 +5342,6 @@ InitializeLogRepWorker(void)
 						MySubscription->name)));
 
 	CommitTransactionCommand();
-
-	update_conflict_retention_status();
 }
 
 /*