v21-0002-Maintain-the-replication-slot-in-logical-launche.patch

From a3fc09aa1450078a08e8f1ae6e0c697ac9e8aac6 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 v21 2/7] 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.
---
 doc/src/sgml/config.sgml                      |   2 +
 doc/src/sgml/func.sgml                        |  14 +-
 doc/src/sgml/protocol.sgml                    |   2 +
 doc/src/sgml/ref/create_subscription.sgml     |   4 +-
 src/backend/access/transam/xlogrecovery.c     |   2 +-
 src/backend/commands/subscriptioncmds.c       |   2 +-
 src/backend/replication/logical/launcher.c    | 224 ++++++++++++++++++
 .../replication/logical/reorderbuffer.c       |   2 +-
 src/backend/replication/slot.c                |  34 ++-
 src/include/replication/slot.h                |  11 +-
 10 files changed, 287 insertions(+), 10 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index f1ab614575..e613e4ced1 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4879,6 +4879,8 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
           new setting.
           This setting has no effect if <varname>primary_conninfo</varname> is not
           set or the server is not in standby mode.
+          The name cannot be <literal>pg_conflict_detection</literal>, as it is
+          reserved for logical replication conflict detection.
          </para>
         </listitem>
        </varlistentry>
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index 47370e581a..feb026d0c5 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -29213,7 +29213,9 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
        </para>
        <para>
         Creates a new physical replication slot named
-        <parameter>slot_name</parameter>. The optional second parameter,
+        <parameter>slot_name</parameter>. The name cannot be
+        <literal>pg_conflict_detection</literal>, as it is reserved for
+        logical replication conflict detection. The optional second parameter,
         when <literal>true</literal>, specifies that the <acronym>LSN</acronym> for this
         replication slot be reserved immediately; otherwise
         the <acronym>LSN</acronym> is reserved on first connection from a streaming
@@ -29258,7 +29260,9 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
        <para>
         Creates a new logical (decoding) replication slot named
         <parameter>slot_name</parameter> using the output plugin
-        <parameter>plugin</parameter>. The optional third
+        <parameter>plugin</parameter>. The name cannot be
+        <literal>pg_conflict_detection</literal>, as it is reserved for
+        logical replication conflict detection. The optional third
         parameter, <parameter>temporary</parameter>, when set to true, specifies that
         the slot should not be permanently stored to disk and is only meant
         for use by the current session. Temporary slots are also
@@ -29288,6 +29292,8 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
        <para>
         Copies an existing physical replication slot named <parameter>src_slot_name</parameter>
         to a physical replication slot named <parameter>dst_slot_name</parameter>.
+        The new slot name cannot be <literal>pg_conflict_detection</literal>,
+        as it is reserved for logical replication conflict detection.
         The copied physical slot starts to reserve WAL from the same <acronym>LSN</acronym> as the
         source slot.
         <parameter>temporary</parameter> is optional. If <parameter>temporary</parameter>
@@ -29309,7 +29315,9 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
         Copies an existing logical replication slot
         named <parameter>src_slot_name</parameter> to a logical replication
         slot named <parameter>dst_slot_name</parameter>, optionally changing
-        the output plugin and persistence.  The copied logical slot starts
+        the output plugin and persistence.  The name cannot be
+        <literal>pg_conflict_detection</literal>, as it is reserved for
+        logical replication conflict detection.  The copied logical slot starts
         from the same <acronym>LSN</acronym> as the source logical slot.  Both
         <parameter>temporary</parameter> and <parameter>plugin</parameter> are
         optional; if they are omitted, the values of the source slot are used.
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 49a4d3ab1b..e20b48a49a 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2032,6 +2032,8 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
          <para>
           The name of the slot to create. Must be a valid replication slot
           name (see <xref linkend="streaming-replication-slots-manipulation"/>).
+          The name cannot be <literal>pg_conflict_detection</literal>, as it
+          is reserved for logical replication conflict detection.
          </para>
         </listitem>
        </varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 6cf7d4f9a1..e95e0cd271 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -169,7 +169,9 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
         <listitem>
          <para>
           Name of the publisher's replication slot to use.  The default is
-          to use the name of the subscription for the slot name.
+          to use the name of the subscription for the slot name. The name cannot
+          be <literal>pg_conflict_detection</literal>, as it is reserved for
+          logical replication conflict detection.
          </para>
 
          <para>
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index 0bbe2eea20..d49ba78b13 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -4739,7 +4739,7 @@ bool
 check_primary_slot_name(char **newval, void **extra, GucSource source)
 {
 	if (*newval && strcmp(*newval, "") != 0 &&
-		!ReplicationSlotValidateName(*newval, WARNING))
+		!ReplicationSlotValidateName(*newval, false, WARNING))
 		return false;
 
 	return true;
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 2d8a71ca1e..3cf1e05539 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -210,7 +210,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 			if (strcmp(opts->slot_name, "none") == 0)
 				opts->slot_name = NULL;
 			else
-				ReplicationSlotValidateName(opts->slot_name, ERROR);
+				ReplicationSlotValidateName(opts->slot_name, false, ERROR);
 		}
 		else if (IsSet(supported_opts, SUBOPT_COPY_DATA) &&
 				 strcmp(defel->defname, "copy_data") == 0)
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 31ebef831c..350c86540f 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -46,6 +46,12 @@
 /* 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
+
 /* GUC variables */
 int			max_logical_replication_workers = 4;
 int			max_sync_workers_per_subscription = 2;
@@ -100,6 +106,10 @@ 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 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);
 
 
 /*
@@ -1119,6 +1129,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")));
 
@@ -1147,6 +1160,8 @@ ApplyLauncherMain(Datum main_arg)
 		MemoryContext subctx;
 		MemoryContext oldctx;
 		long		wait_time = DEFAULT_NAPTIME_PER_CYCLE;
+		bool		can_advance_xmin = true;
+		FullTransactionId xmin = InvalidFullTransactionId;
 
 		CHECK_FOR_INTERRUPTS();
 
@@ -1167,14 +1182,49 @@ ApplyLauncherMain(Datum main_arg)
 			long		elapsed;
 
 			if (!sub->enabled)
+			{
+				can_advance_xmin = false;
 				continue;
+			}
 
 			LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 			w = logicalrep_worker_find(sub->oid, InvalidOid, false);
 			LWLockRelease(LogicalRepWorkerLock);
 
 			if (w != NULL)
+			{
+				/*
+				 * Collect non-removable transaction IDs from all apply
+				 * workers to determine the xmin for advancing the replication
+				 * slot used in conflict detection.
+				 */
+				if (can_advance_xmin)
+				{
+					FullTransactionId nonremovable_xid;
+
+					SpinLockAcquire(&w->relmutex);
+					nonremovable_xid = w->oldest_nonremovable_xid;
+					SpinLockRelease(&w->relmutex);
+
+					/*
+					 * Stop advancing xmin if an invalid non-removable
+					 * transaction ID is found, otherwise update xmin.
+					 */
+					if (!FullTransactionIdIsValid(nonremovable_xid))
+						can_advance_xmin = false;
+					else if (!FullTransactionIdIsValid(xmin) ||
+							 FullTransactionIdPrecedes(nonremovable_xid, xmin))
+						xmin = nonremovable_xid;
+				}
+
 				continue;		/* worker is running already */
+			}
+
+			/*
+			 * Create the conflict slot before starting the worker to prevent
+			 * it from unnecessarily maintaining its oldest_nonremovable_xid.
+			 */
+			create_conflict_slot_if_not_exists();
 
 			/*
 			 * If the worker is eligible to start now, launch it.  Otherwise,
@@ -1207,6 +1257,33 @@ ApplyLauncherMain(Datum main_arg)
 			}
 		}
 
+		/*
+		 * Maintain the xmin value of the replication slot for conflict
+		 * detection if needed, and update the sleep time before the next
+		 * attempt.
+		 */
+		if (sublist)
+		{
+			bool		updated = false;
+
+			if (can_advance_xmin)
+				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;
+		}
+
+		/*
+		 * Drop the slot if we're no longer retaining dead tuples.
+		 */
+		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. */
@@ -1234,6 +1311,153 @@ ApplyLauncherMain(Datum main_arg)
 	/* Not reachable */
 }
 
+/*
+ * Create and acquire the replication slot used to retain dead tuples for
+ * conflict detection, if not yet.
+ */
+static void
+create_conflict_slot_if_not_exists(void)
+{
+	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))
+	{
+		ReplicationSlotAcquire(CONFLICT_DETECTION_SLOT, true);
+		return;
+	}
+
+	/*
+	 * Initially create persistent slot as ephemeral - that allows us to
+	 * nicely handle errors during initialization because it'll get dropped if
+	 * this transaction fails. We'll make it persistent after safely
+	 * initializing the xmin of the slot.
+	 */
+	ReplicationSlotCreate(CONFLICT_DETECTION_SLOT, false,
+						  RS_EPHEMERAL, false, false, false);
+
+	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+
+	xmin_horizon = GetOldestSafeDecodingTransactionId(false);
+
+	SpinLockAcquire(&MyReplicationSlot->mutex);
+	MyReplicationSlot->effective_xmin = xmin_horizon;
+	MyReplicationSlot->data.xmin = xmin_horizon;
+	SpinLockRelease(&MyReplicationSlot->mutex);
+
+	ReplicationSlotsComputeRequiredXmin(true);
+
+	LWLockRelease(ProcArrayLock);
+
+	ReplicationSlotPersist();
+}
+
+/*
+ * 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)
+{
+	TransactionId next_xid;
+	FullTransactionId next_full_xid;
+	FullTransactionId full_xmin;
+	uint32		xmin_epoch;
+
+	Assert(MyReplicationSlot);
+	Assert(FullTransactionIdIsValid(new_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);
+
+	if (FullTransactionIdPrecedesOrEquals(new_xmin, full_xmin))
+		return false;
+
+	SpinLockAcquire(&MyReplicationSlot->mutex);
+	MyReplicationSlot->data.xmin = XidFromFullTransactionId(new_xmin);
+	SpinLockRelease(&MyReplicationSlot->mutex);
+
+	/* 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?
  */
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 79b60df7cf..64b43346a9 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -4665,7 +4665,7 @@ StartupReorderBuffer(void)
 			continue;
 
 		/* if it cannot be a slot, skip the directory */
-		if (!ReplicationSlotValidateName(logical_de->d_name, DEBUG2))
+		if (!ReplicationSlotValidateName(logical_de->d_name, true, DEBUG2))
 			continue;
 
 		/*
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index b30e0473e1..3a3884b382 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -47,6 +47,7 @@
 #include "miscadmin.h"
 #include "pgstat.h"
 #include "postmaster/interrupt.h"
+#include "replication/logicallauncher.h"
 #include "replication/slotsync.h"
 #include "replication/slot.h"
 #include "replication/walsender_private.h"
@@ -157,6 +158,7 @@ static SyncStandbySlotsConfigData *synchronized_standby_slots_config;
 static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr;
 
 static void ReplicationSlotShmemExit(int code, Datum arg);
+static bool IsReservedSlotName(const char *name);
 static void ReplicationSlotDropPtr(ReplicationSlot *slot);
 
 /* internal persistency functions */
@@ -243,13 +245,17 @@ ReplicationSlotShmemExit(int code, Datum arg)
 /*
  * Check whether the passed slot name is valid and report errors at elevel.
  *
+ * An error will be reported for a reserved replication slot name if
+ * allow_reserved_name is set to false.
+ *
  * Slot names may consist out of [a-z0-9_]{1,NAMEDATALEN-1} which should allow
  * the name to be used as a directory name on every supported OS.
  *
  * Returns whether the directory name is valid or not if elevel < ERROR.
  */
 bool
-ReplicationSlotValidateName(const char *name, int elevel)
+ReplicationSlotValidateName(const char *name, bool allow_reserved_name,
+							int elevel)
 {
 	const char *cp;
 
@@ -285,9 +291,29 @@ ReplicationSlotValidateName(const char *name, int elevel)
 			return false;
 		}
 	}
+
+	if (!allow_reserved_name && IsReservedSlotName(name))
+	{
+		ereport(elevel,
+				errcode(ERRCODE_RESERVED_NAME),
+				errmsg("replication slot name \"%s\" is reserved",
+					   name));
+
+		return false;
+	}
+
 	return true;
 }
 
+/*
+ * Return true if the replication slot name is "pg_conflict_detection".
+ */
+static bool
+IsReservedSlotName(const char *name)
+{
+	return (strcmp(name, CONFLICT_DETECTION_SLOT) == 0);
+}
+
 /*
  * Create a new replication slot and mark it as used by this backend.
  *
@@ -315,7 +341,11 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 
 	Assert(MyReplicationSlot == NULL);
 
-	ReplicationSlotValidateName(name, ERROR);
+	/*
+	 * The logical launcher might be creating an internal slot, so using a
+	 * reserved name is allowed in this case.
+	 */
+	ReplicationSlotValidateName(name, IsLogicalLauncher(), ERROR);
 
 	if (failover)
 	{
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index bf62b36ad0..ea0485efa2 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -20,6 +20,13 @@
 /* directory to store replication slot data in */
 #define PG_REPLSLOT_DIR     "pg_replslot"
 
+/*
+ * The reserved name for a replication slot used to retain dead tuples for
+ * conflict detection in logical replication. See
+ * maybe_advance_nonremovable_xid() for detail.
+ */
+#define CONFLICT_DETECTION_SLOT "pg_conflict_detection"
+
 /*
  * Behaviour of replication slots, upon release or crash.
  *
@@ -261,7 +268,9 @@ extern void ReplicationSlotMarkDirty(void);
 
 /* misc stuff */
 extern void ReplicationSlotInitialize(void);
-extern bool ReplicationSlotValidateName(const char *name, int elevel);
+extern bool ReplicationSlotValidateName(const char *name,
+										bool allow_reserved_name,
+										int elevel);
 extern void ReplicationSlotReserveWal(void);
 extern void ReplicationSlotsComputeRequiredXmin(bool already_locked);
 extern void ReplicationSlotsComputeRequiredLSN(void);
-- 
2.43.0