v3-0002-Maintain-the-replication-slot-in-logical-launcher.patch

From 7ed5f4fcda21ca338746fb8fac772aa0e4d0466d Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Thu, 26 Sep 2024 12:11:34 +0800
Subject: [PATCH v3 2/5] Maintain the replication slot in logical launcher to
 retain dead tuples

This patch enables the logical replication launcher to create and maintain a
replication slot named pg_conflict_detection.

The launcher periodically collects the oldest_nonremovable_xid from all apply
workers. It then computes the minimum transaction ID and advances the xmin
value of the replication slot if it precedes the computed value.

The interval for updating the slot (nap time) is dynamically adjusted based on
the activity of the apply workers. The launcher waits for a certain period
before performing the next update, with the duration varying depending on
whether the xmin value of the replication slot was updated during the last
cycle.
---
 src/backend/replication/logical/launcher.c | 209 +++++++++++++++++++++
 1 file changed, 209 insertions(+)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index e5fdca8bbf..7acdc1ebdd 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -46,6 +46,14 @@
 /* max sleep time between cycles (3min) */
 #define DEFAULT_NAPTIME_PER_CYCLE 180000L
 
+/*
+ * Min sleep time (200ms) between cycles to update the xmin value of the
+ * replication slot.
+ */
+#define MIN_NAPTIME_PER_SLOT_UPDATE 200
+
+#define CONFLICT_DETECTION_SLOT "pg_conflict_detection"
+
 /* GUC variables */
 int			max_logical_replication_workers = 4;
 int			max_sync_workers_per_subscription = 2;
@@ -100,6 +108,9 @@ 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 bool advance_conflict_slot_xmin(FullTransactionId new_xmin);
+static void compute_slot_update_naptime(bool slot_updated, long *sleep_time);
+static void drop_conflict_slot_if_exists(void);
 
 
 /*
@@ -1127,6 +1138,9 @@ ApplyLauncherWakeup(void)
 void
 ApplyLauncherMain(Datum main_arg)
 {
+	bool		slot_maybe_exist = true;
+	long		slot_update_wait_time = MIN_NAPTIME_PER_SLOT_UPDATE;
+
 	ereport(DEBUG1,
 			(errmsg_internal("logical replication launcher started")));
 
@@ -1155,6 +1169,8 @@ ApplyLauncherMain(Datum main_arg)
 		MemoryContext subctx;
 		MemoryContext oldctx;
 		long		wait_time = DEFAULT_NAPTIME_PER_CYCLE;
+		bool		updated;
+		FullTransactionId xmin = InvalidFullTransactionId;
 
 		CHECK_FOR_INTERRUPTS();
 
@@ -1179,6 +1195,21 @@ ApplyLauncherMain(Datum main_arg)
 
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 			w = logicalrep_worker_find(sub->oid, InvalidOid, false);
+
+			if (w != NULL)
+			{
+				FullTransactionId nonremovable_xid;
+
+				SpinLockAcquire(&w->relmutex);
+				nonremovable_xid = w->oldest_nonremovable_xid;
+				SpinLockRelease(&w->relmutex);
+
+				if (!FullTransactionIdIsValid(xmin) ||
+					!FullTransactionIdIsValid(nonremovable_xid) ||
+					FullTransactionIdPrecedes(nonremovable_xid, xmin))
+					xmin = nonremovable_xid;
+			}
+
 			LWLockRelease(LogicalRepWorkerLock);
 
 			if (w != NULL)
@@ -1215,6 +1246,26 @@ ApplyLauncherMain(Datum main_arg)
 			}
 		}
 
+		if (sublist)
+		{
+			/*
+			 * Maintain the xmin value of the replication slot for conflict
+			 * detection if needed, and update the sleep time before the next
+			 * attempt.
+			 */
+			updated = advance_conflict_slot_xmin(xmin);
+
+			compute_slot_update_naptime(updated, &slot_update_wait_time);
+			wait_time = Min(wait_time, slot_update_wait_time);
+
+			slot_maybe_exist = true;
+		}
+		else if (slot_maybe_exist)
+		{
+			drop_conflict_slot_if_exists();
+			slot_maybe_exist = false;
+		}
+
 		/* Switch back to original memory context. */
 		MemoryContextSwitchTo(oldctx);
 		/* Clean the temporary memory. */
@@ -1242,6 +1293,164 @@ ApplyLauncherMain(Datum main_arg)
 	/* Not reachable */
 }
 
+/*
+ * Attempt to advance the xmin value of the replication slot used to retain
+ * dead tuples for conflict detection.
+ */
+static bool
+advance_conflict_slot_xmin(FullTransactionId new_xmin)
+{
+	bool		updated_xmin = false;
+	TransactionId next_xid;
+	FullTransactionId next_full_xid;
+	FullTransactionId full_xmin;
+	uint32		xmin_epoch;
+
+	/*
+	 * Acquire the slot if it hasn't been acquired yet. If the slot is not yet
+	 * created, create it.
+	 */
+	if (!MyReplicationSlot)
+	{
+		if (SearchNamedReplicationSlot(CONFLICT_DETECTION_SLOT, true))
+		{
+			ReplicationSlotAcquire(CONFLICT_DETECTION_SLOT, true);
+		}
+		else
+		{
+			TransactionId xmin_horizon;
+
+			ReplicationSlotCreate(CONFLICT_DETECTION_SLOT, false,
+								  RS_PERSISTENT, false, false, false);
+
+			LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+
+			xmin_horizon = GetOldestSafeDecodingTransactionId(true);
+
+			SpinLockAcquire(&MyReplicationSlot->mutex);
+			MyReplicationSlot->effective_xmin = xmin_horizon;
+			MyReplicationSlot->data.xmin = xmin_horizon;
+			SpinLockRelease(&MyReplicationSlot->mutex);
+
+			ReplicationSlotsComputeRequiredXmin(true);
+
+			LWLockRelease(ProcArrayLock);
+		}
+	}
+
+	/* No need to update xmin if the slot has been invalidated */
+	if (MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
+	{
+		/*
+		 * Only a WARNING is reported here, which is intended to avoid
+		 * preventing the launcher from starting logical replication workers.
+		 */
+		ereport(WARNING,
+				errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("The replication slot \"%s\" has been invalidated",
+					   CONFLICT_DETECTION_SLOT),
+				errhint("Drop the replication slot \"%s\".",
+						CONFLICT_DETECTION_SLOT));
+
+		return false;
+	}
+
+	Assert(TransactionIdIsValid(MyReplicationSlot->data.xmin));
+
+	/*
+	 * Compute the epoch of the xmin value for the replication slot based on
+	 * the next full transaction ID and its epoch.
+	 */
+	next_full_xid = ReadNextFullTransactionId();
+	next_xid = XidFromFullTransactionId(next_full_xid);
+	xmin_epoch = EpochFromFullTransactionId(next_full_xid);
+
+	/*
+	 * Adjust the epoch if the next transaction ID is less than the current
+	 * xmin of the replication slot. This handles the case where transaction
+	 * ID wraparound has occurred.
+	 */
+	if (next_xid < MyReplicationSlot->data.xmin)
+		xmin_epoch--;
+
+	full_xmin = FullTransactionIdFromEpochAndXid(xmin_epoch,
+												 MyReplicationSlot->data.xmin);
+
+	SpinLockAcquire(&MyReplicationSlot->mutex);
+
+	if (FullTransactionIdIsValid(new_xmin) &&
+		FullTransactionIdPrecedes(full_xmin, new_xmin))
+	{
+		updated_xmin = true;
+		MyReplicationSlot->data.xmin = XidFromFullTransactionId(new_xmin);
+	}
+
+	SpinLockRelease(&MyReplicationSlot->mutex);
+
+	if (!updated_xmin)
+		return false;
+
+	/* first write new xmin to disk, so we know what's up after a crash */
+
+	ReplicationSlotMarkDirty();
+	ReplicationSlotSave();
+	elog(DEBUG1, "updated xmin: %u", MyReplicationSlot->data.xmin);
+
+	/*
+	 * Now the new xmin is safely on disk, we can let the global value
+	 * advance. We do not take ProcArrayLock or similar since we only advance
+	 * xmin here and there's not much harm done by a concurrent computation
+	 * missing that.
+	 */
+	SpinLockAcquire(&MyReplicationSlot->mutex);
+	MyReplicationSlot->effective_xmin = MyReplicationSlot->data.xmin;
+	SpinLockRelease(&MyReplicationSlot->mutex);
+
+	ReplicationSlotsComputeRequiredXmin(false);
+
+	return true;
+}
+
+/*
+ * Update the sleep time before the next slot update.
+ *
+ * If there is no slot activity, the wait time between sync cycles will double
+ * (up to a maximum of 3 minutes). If there is some slot activity, the wait
+ * time between sync cycles is reset to the minimum (200ms).
+ */
+static void
+compute_slot_update_naptime(bool slot_updated, long *sleep_time)
+{
+	if (!slot_updated)
+	{
+		/*
+		 * The slot was not updated, so double the sleep time, but not beyond
+		 * the maximum allowable value.
+		 */
+		*sleep_time = Min(*sleep_time * 2, DEFAULT_NAPTIME_PER_CYCLE);
+	}
+	else
+	{
+		/*
+		 * The slot was updated since the last sleep, so reset the sleep time.
+		 */
+		*sleep_time = MIN_NAPTIME_PER_SLOT_UPDATE;
+	}
+}
+
+/*
+ * Drop the replication slot used to retain dead tuples for conflict detection,
+ * if it exists.
+ */
+static void
+drop_conflict_slot_if_exists(void)
+{
+	if (MyReplicationSlot)
+		ReplicationSlotDropAcquired();
+	else if (SearchNamedReplicationSlot(CONFLICT_DETECTION_SLOT, true))
+		ReplicationSlotDrop(CONFLICT_DETECTION_SLOT, true);
+}
+
 /*
  * Is current process the logical replication launcher?
  */
-- 
2.30.0.windows.2