v56-0002-Introduce-a-new-GUC-max_conflict_retention_durat.patch

From 7bdfd7c8c97b87e23d3d1d9cbf80a901eb9e96c3 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Wed, 23 Jul 2025 13:38:12 +0800
Subject: [PATCH v56 2/3] Introduce a new GUC 'max_conflict_retention_duration'

This commit introduces a GUC option max_conflict_retention_duration, designed
to prevent excessive accumulation of dead tuples when subscription with
retain_dead_tuples enabled is present and the the apply worker cannot catch
up with the publisher's workload.

If the time spent advancing non-removable transaction ID surpasses the
max_conflict_retention_duration threshold, the apply worker would stop
retaining information for conflict detection. The replication slot
pg_conflict_detection will be invalidated if all apply workers associated with
the subscription, where retain_dead_tuples is enabled, confirm that the
retention duration exceeded the max_conflict_retention_duration.

In this patch, a replication slot will not be automatically re-created if it
becomes invalidated. Users can disable retain_dead_tuples and re-enable it
after confirming that the replication slot has been dropped. An upcoming patch
will include support for automatic slot recreation once at least one apply
worker confirms that the retention duration is within the
max_conflict_retention_duration limit.

To monitor worker's conflict retention status, this patch also introduces a new
column 'retain_dead_tuples' in the pg_stat_subscription view. This column
indicates whether the apply worker is effectively retaining conflict
information. The value is set to true only if retain_dead_tuples is enabled
for the associated subscription, and the retention duration for conflict
detection by the apply worker has not exceeded max_conflict_retention_duration.
---
 doc/src/sgml/config.sgml                      |  43 ++++++
 doc/src/sgml/monitoring.sgml                  |  13 ++
 doc/src/sgml/system-views.sgml                |  11 ++
 src/backend/catalog/system_views.sql          |   3 +-
 src/backend/replication/logical/launcher.c    | 120 +++++++++++++++--
 src/backend/replication/logical/worker.c      | 126 ++++++++++++++++--
 src/backend/replication/slot.c                |  13 ++
 src/backend/utils/misc/guc_tables.c           |  13 ++
 src/backend/utils/misc/postgresql.conf.sample |   1 +
 src/include/catalog/pg_proc.dat               |   6 +-
 src/include/replication/logicallauncher.h     |   1 +
 src/include/replication/slot.h                |   4 +-
 src/include/replication/worker_internal.h     |   6 +
 src/test/regress/expected/rules.out           |   5 +-
 src/test/subscription/t/035_conflicts.pl      |  10 +-
 15 files changed, 348 insertions(+), 27 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 20ccb2d6b54..0d6616857e7 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5399,6 +5399,49 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-max-conflict-retention-duration" xreflabel="max_conflict_retention_duration">
+      <term><varname>max_conflict_retention_duration</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>max_conflict_retention_duration</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Maximum duration for which each apply worker can request to retain the
+        information useful for conflict detection when
+        <literal>retain_dead_tuples</literal> is enabled for the associated
+        subscriptions. The default value is <literal>0</literal>, indicating
+        that conflict information is retained until it is no longer needed for
+        detection purposes. If this value is specified without units, it is
+        taken as milliseconds.
+       </para>
+       <para>
+        The replication slot
+        <quote><literal>pg_conflict_detection</literal></quote> that used to
+        retain conflict information will be invalidated if all apply workers
+        associated with the subscriptions, where
+        <literal>retain_dead_tuples</literal> is enabled, confirm that the
+        retention duration exceeded the
+        <literal>max_conflict_retention_duration</literal>. If the replication
+        slot is invalidated, you can disable
+        <literal>retain_dead_tuples</literal> and re-enable it after
+        confirming this replication slot has been dropped.
+       </para>
+       <para>
+        This option is effective only if a subscription with
+        <literal>retain_dead_tuples</literal> enabled is present, and the
+        associated apply worker is active.
+       </para>
+       <warning>
+        <para>
+         Note that setting a non-zero value for this option could lead to
+         conflict information being removed prematurely, potentially missing
+         some conflict detections.
+        </para>
+       </warning>
+      </listitem>
+     </varlistentry>
+
      </variablelist>
     </sect2>
 
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index fa78031ccbb..9a0fd556c5f 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -2114,6 +2114,19 @@ description | Waiting for a newly initialized WAL file to reach durable storage
        sender; NULL for parallel apply workers
       </para></entry>
      </row>
+
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>retain_dead_tuples</structfield> <type>boolean</type>
+      </para>
+      <para>
+       True if <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
+       is enabled and the duration for which conflict information is
+       retained for conflict detection by this apply worker does not exceed
+       <link linkend="guc-max-conflict-retention-duration"><literal>max_conflict_retention_duration</literal></link>; NULL for
+       parallel apply workers and table synchronization workers.
+      </para></entry>
+     </row>
     </tbody>
    </tgroup>
   </table>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 4187191ea74..27e55bb0f35 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -3007,6 +3007,17 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
           <xref linkend="guc-idle-replication-slot-timeout"/> duration.
          </para>
         </listitem>
+        <listitem>
+         <para>
+          <literal>conflict_retention_exceeds_max_duration</literal> means that
+          the duration for retaining conflict information, which is used
+          in logical replication conflict detection, has exceeded the maximum
+          allowable limit. It is set only for the slot
+          <literal>pg_conflict_detection</literal>, which is created when
+          <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
+          is enabled.
+         </para>
+        </listitem>
        </itemizedlist>
       </para></entry>
      </row>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 77c693f630e..235b6b4f323 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -996,7 +996,8 @@ CREATE VIEW pg_stat_subscription AS
             st.last_msg_send_time,
             st.last_msg_receipt_time,
             st.latest_end_lsn,
-            st.latest_end_time
+            st.latest_end_time,
+            st.retain_dead_tuples
     FROM pg_subscription su
             LEFT JOIN pg_stat_get_subscription(NULL) st
                       ON (st.subid = su.oid);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 742d9ba68e9..c0e785b1a8c 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -23,6 +23,7 @@
 #include "access/tableam.h"
 #include "access/xact.h"
 #include "catalog/pg_subscription.h"
+#include "catalog/pg_subscription_d.h"
 #include "catalog/pg_subscription_rel.h"
 #include "funcapi.h"
 #include "lib/dshash.h"
@@ -43,6 +44,7 @@
 #include "utils/memutils.h"
 #include "utils/pg_lsn.h"
 #include "utils/snapmgr.h"
+#include "utils/syscache.h"
 
 /* max sleep time between cycles (3min) */
 #define DEFAULT_NAPTIME_PER_CYCLE 180000L
@@ -51,6 +53,7 @@
 int			max_logical_replication_workers = 4;
 int			max_sync_workers_per_subscription = 2;
 int			max_parallel_apply_workers_per_subscription = 2;
+int			max_conflict_retention_duration = 0;
 
 LogicalRepWorker *MyLogicalRepWorker = NULL;
 
@@ -100,9 +103,11 @@ 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 compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin);
+static void compute_min_nonremovable_xid(LogicalRepWorker *worker, bool can_advance_xmin,
+										 TransactionId *xmin, bool *stop_retention);
 static bool acquire_conflict_slot_if_exists(void);
 static void advance_conflict_slot_xmin(TransactionId new_xmin);
+static void invalidate_conflict_slot(void);
 
 
 /*
@@ -465,6 +470,8 @@ retry:
 	worker->oldest_nonremovable_xid = retain_dead_tuples
 		? MyReplicationSlot->data.xmin
 		: InvalidTransactionId;
+	worker->stop_conflict_info_retention = (retain_dead_tuples &&
+											MyReplicationSlot->data.invalidated != RS_INVAL_NONE);
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -1181,6 +1188,7 @@ ApplyLauncherMain(Datum main_arg)
 		long		wait_time = DEFAULT_NAPTIME_PER_CYCLE;
 		bool		can_advance_xmin = true;
 		bool		retain_dead_tuples = false;
+		bool		stop_retention = true;
 		TransactionId xmin = InvalidTransactionId;
 
 		CHECK_FOR_INTERRUPTS();
@@ -1223,6 +1231,13 @@ ApplyLauncherMain(Datum main_arg)
 				 */
 				can_advance_xmin &= sub->enabled;
 
+				/*
+				 * Stop the conflict information retention only if all workers
+				 * for subscriptions with retain_dead_tuples enabled have
+				 * requested it.
+				 */
+				stop_retention &= sub->enabled;
+
 				/*
 				 * Create a replication slot to retain information necessary
 				 * for conflict detection such as dead tuples, commit
@@ -1254,8 +1269,9 @@ ApplyLauncherMain(Datum main_arg)
 				 * required for conflict detection among all running apply
 				 * workers that enables retain_dead_tuples.
 				 */
-				if (sub->retaindeadtuples && can_advance_xmin)
-					compute_min_nonremovable_xid(w, &xmin);
+				if (sub->retaindeadtuples)
+					compute_min_nonremovable_xid(w, can_advance_xmin, &xmin,
+												 &stop_retention);
 
 				/* worker is running already */
 				continue;
@@ -1265,10 +1281,15 @@ ApplyLauncherMain(Datum main_arg)
 			 * Can't advance xmin of the slot unless all the workers
 			 * corresponding to subscriptions with retain_dead_tuples are
 			 * running, disabling the further computation of the minimum
-			 * nonremovable xid.
+			 * nonremovable xid. Similarly, stop the conflict information
+			 * retention only if all workers for subscriptions with
+			 * retain_dead_tuples enabled have requested it.
 			 */
 			if (sub->retaindeadtuples)
+			{
 				can_advance_xmin = false;
+				stop_retention = false;
+			}
 
 			/*
 			 * If the worker is eligible to start now, launch it.  Otherwise,
@@ -1314,13 +1335,16 @@ ApplyLauncherMain(Datum main_arg)
 		}
 
 		/*
-		 * Drop the CONFLICT_DETECTION_SLOT slot if there is no subscription
-		 * that requires us to retain dead tuples. Otherwise, if required,
-		 * advance the slot's xmin to protect dead tuples required for the
-		 * conflict detection.
+		 * Invalidate the slot if requested by all apply workers. Or, drop the
+		 * CONFLICT_DETECTION_SLOT slot if there is no subscription that
+		 * requires us to retain dead tuples. Otherwise, if required, advance
+		 * the slot's xmin to protect dead tuples required for the conflict
+		 * detection.
 		 */
 		if (MyReplicationSlot)
 		{
+			if (retain_dead_tuples && stop_retention)
+				invalidate_conflict_slot();
 			if (!retain_dead_tuples)
 				ReplicationSlotDropAcquired();
 			else if (can_advance_xmin)
@@ -1358,11 +1382,16 @@ ApplyLauncherMain(Datum main_arg)
  * Determine the minimum non-removable transaction ID across all apply workers
  * for subscriptions that have retain_dead_tuples enabled. Store the result
  * in *xmin.
+ *
+ * if any worker continues retaining conflict information, *stop_retention is
+ * set to false.
  */
 static void
-compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
+compute_min_nonremovable_xid(LogicalRepWorker *worker, bool can_advance_xmin,
+							 TransactionId *xmin, bool *stop_retention)
 {
 	TransactionId nonremovable_xid;
+	bool		stop_conflict_info_retention;
 
 	Assert(worker != NULL);
 
@@ -1374,8 +1403,22 @@ compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
 
 	SpinLockAcquire(&worker->relmutex);
 	nonremovable_xid = worker->oldest_nonremovable_xid;
+	stop_conflict_info_retention = worker->stop_conflict_info_retention;
 	SpinLockRelease(&worker->relmutex);
 
+	/*
+	 * Stop the conflict information retention only if all workers for
+	 * subscriptions with retain_dead_tuples enabled have requested it.
+	 */
+	*stop_retention &= stop_conflict_info_retention;
+
+	/*
+	 * Skip collecting oldest_nonremovable_xid for workers that have stopped
+	 * conflict retention or if advancement is not possible.
+	 */
+	if (stop_conflict_info_retention || !can_advance_xmin)
+		return;
+
 	Assert(TransactionIdIsValid(nonremovable_xid));
 
 	if (!TransactionIdIsValid(*xmin) ||
@@ -1436,6 +1479,37 @@ advance_conflict_slot_xmin(TransactionId new_xmin)
 	return;
 }
 
+/*
+ * Invalidate the replication slot used to retain information for conflict
+ * detection.
+ */
+static void
+invalidate_conflict_slot(void)
+{
+	Assert(MyReplicationSlot);
+
+	/*
+	 * Do nothing if the replication slot has already been invalidated due to
+	 * conflict retention duration.
+	 */
+	if (MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
+	{
+		Assert(MyReplicationSlot->data.invalidated == RS_INVAL_CONFLICT_RETENTION_DURATION);
+		return;
+	}
+
+	ReplicationSlotRelease();
+	InvalidateObsoleteReplicationSlots(RS_INVAL_CONFLICT_RETENTION_DURATION,
+									   InvalidXLogRecPtr, InvalidOid,
+									   InvalidTransactionId);
+
+	/*
+	 * Acquire the invalidated slot to allow the launcher to drop it in the
+	 * next cycle if no subscriptions have retain_dead_tuples enabled.
+	 */
+	ReplicationSlotAcquire(CONFLICT_DETECTION_SLOT, true, false);
+}
+
 /*
  * Create and acquire the replication slot used to retain information for
  * conflict detection, if not yet.
@@ -1516,7 +1590,7 @@ GetLeaderApplyWorkerPid(pid_t pid)
 Datum
 pg_stat_get_subscription(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_SUBSCRIPTION_COLS	10
+#define PG_STAT_GET_SUBSCRIPTION_COLS	11
 	Oid			subid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0);
 	int			i;
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
@@ -1593,6 +1667,32 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
 				elog(ERROR, "unknown worker type");
 		}
 
+		/*
+		 * Only the leader apply worker manages conflict retention (see
+		 * maybe_advance_nonremovable_xid() for details).
+		 */
+		if (!isParallelApplyWorker(&worker) && !isTablesyncWorker(&worker))
+		{
+			HeapTuple	tup;
+			Form_pg_subscription subform;
+
+			tup = SearchSysCache1(SUBSCRIPTIONOID,
+								  ObjectIdGetDatum(worker.subid));
+
+			if (!HeapTupleIsValid(tup))
+				elog(ERROR, "cache lookup failed for subscription %u",
+					 worker.subid);
+
+			subform = (Form_pg_subscription) GETSTRUCT(tup);
+
+			values[10] = subform->subretaindeadtuples &&
+				!worker.stop_conflict_info_retention;
+
+			ReleaseSysCache(tup);
+		}
+		else
+			nulls[10] = true;
+
 		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
 							 values, nulls);
 
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index f0b37a32ac1..d8d929c252b 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -415,6 +415,9 @@ typedef struct RetainDeadTuplesData
 										 * updated in final phase
 										 * (RDT_WAIT_FOR_LOCAL_FLUSH) */
 
+	long		table_sync_wait_time;	/* time spent waiting for table sync
+										 * to finish */
+
 	/*
 	 * The following fields are used to determine the timing for the next
 	 * round of transaction ID advancement.
@@ -555,6 +558,8 @@ static void request_publisher_status(RetainDeadTuplesData *rdt_data);
 static void wait_for_publisher_status(RetainDeadTuplesData *rdt_data,
 									  bool status_received);
 static void wait_for_local_flush(RetainDeadTuplesData *rdt_data);
+static void reset_retention_data_fields(RetainDeadTuplesData *rdt_data);
+static bool should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data);
 static void adjust_xid_advance_interval(RetainDeadTuplesData *rdt_data,
 										bool new_xid_found);
 
@@ -4100,7 +4105,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		 * Ensure to wake up when it's possible to advance the non-removable
 		 * transaction ID.
 		 */
-		if (rdt_data.phase == RDT_GET_CANDIDATE_XID &&
+		if (!MyLogicalRepWorker->stop_conflict_info_retention &&
+			rdt_data.phase == RDT_GET_CANDIDATE_XID &&
 			rdt_data.xid_advance_interval)
 			wait_time = Min(wait_time, rdt_data.xid_advance_interval);
 
@@ -4315,6 +4321,10 @@ can_advance_nonremovable_xid(RetainDeadTuplesData *rdt_data)
 	if (!MySubscription->retaindeadtuples)
 		return false;
 
+	/* No need to advance if we have already stopped retaining */
+	if (MyLogicalRepWorker->stop_conflict_info_retention)
+		return false;
+
 	return true;
 }
 
@@ -4458,6 +4468,13 @@ wait_for_publisher_status(RetainDeadTuplesData *rdt_data,
 	if (!status_received)
 		return;
 
+	/*
+	 * Stop retaining conflict information if required (See
+	 * should_stop_conflict_info_retention() for details).
+	 */
+	if (should_stop_conflict_info_retention(rdt_data))
+		return;
+
 	if (!FullTransactionIdIsValid(rdt_data->remote_wait_for))
 		rdt_data->remote_wait_for = rdt_data->remote_nextxid;
 
@@ -4539,6 +4556,27 @@ wait_for_local_flush(RetainDeadTuplesData *rdt_data)
 	 * have a WAL position greater than the rdt_data->remote_lsn.
 	 */
 	if (!AllTablesyncsReady())
+	{
+		TimestampTz now;
+
+		now = rdt_data->last_recv_time
+			? rdt_data->last_recv_time : GetCurrentTimestamp();
+
+		/*
+		 * Record the time spent waiting for table sync, it is needed for the
+		 * timeout check in should_stop_conflict_info_retention().
+		 */
+		rdt_data->table_sync_wait_time =
+			TimestampDifferenceMilliseconds(rdt_data->candidate_xid_time, now);
+
+		return;
+	}
+
+	/*
+	 * Stop retaining conflict information if required (See
+	 * should_stop_conflict_info_retention() for details).
+	 */
+	if (should_stop_conflict_info_retention(rdt_data))
 		return;
 
 	/*
@@ -4584,12 +4622,21 @@ wait_for_local_flush(RetainDeadTuplesData *rdt_data)
 	/* Notify launcher to update the xmin of the conflict slot */
 	ApplyLauncherWakeup();
 
-	/*
-	 * Reset all data fields except those used to determine the timing for the
-	 * next round of transaction ID advancement. We can even use
-	 * flushpos_update_time in the next round to decide whether to get the
-	 * latest flush position.
-	 */
+	reset_retention_data_fields(rdt_data);
+
+	/* process the next phase */
+	process_rdt_phase_transition(rdt_data, false);
+}
+
+/*
+ * Reset all data fields of RetainDeadTuplesData except those used to
+ * determine the timing for the next round of transaction ID advancement. We
+ * can even use flushpos_update_time in the next round to decide whether to get
+ * the latest flush position.
+ */
+static void
+reset_retention_data_fields(RetainDeadTuplesData *rdt_data)
+{
 	rdt_data->phase = RDT_GET_CANDIDATE_XID;
 	rdt_data->remote_lsn = InvalidXLogRecPtr;
 	rdt_data->remote_oldestxid = InvalidFullTransactionId;
@@ -4597,9 +4644,66 @@ wait_for_local_flush(RetainDeadTuplesData *rdt_data)
 	rdt_data->reply_time = 0;
 	rdt_data->remote_wait_for = InvalidFullTransactionId;
 	rdt_data->candidate_xid = InvalidTransactionId;
+	rdt_data->table_sync_wait_time = 0;
+}
 
-	/* process the next phase */
-	process_rdt_phase_transition(rdt_data, false);
+/*
+ * Check whether conflict information retention should be stopped because the
+ * wait time has exceeded the maximum limit (max_conflict_retention_duration).
+ *
+ * If retention should be stopped, set
+ * 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_dead_tuples and re-enables it after confirming that the
+ * replication slot has been dropped.
+ */
+static bool
+should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data)
+{
+	TimestampTz now;
+
+	Assert(TransactionIdIsValid(rdt_data->candidate_xid));
+	Assert(rdt_data->phase == RDT_WAIT_FOR_PUBLISHER_STATUS ||
+		   rdt_data->phase == RDT_WAIT_FOR_LOCAL_FLUSH);
+
+	if (!max_conflict_retention_duration)
+		return false;
+
+	/*
+	 * 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 = rdt_data->last_recv_time ? rdt_data->last_recv_time : GetCurrentTimestamp();
+
+	/*
+	 * Return if the wait time has not exceeded the maximum limit
+	 * (max_conflict_retention_duration).
+	 */
+	if (!TimestampDifferenceExceeds(rdt_data->candidate_xid_time, now,
+									max_conflict_retention_duration +
+									rdt_data->table_sync_wait_time))
+		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 = InvalidTransactionId;
+	MyLogicalRepWorker->stop_conflict_info_retention = true;
+	SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+	/* Notify launcher to invalidate the conflict slot */
+	ApplyLauncherWakeup();
+
+	reset_retention_data_fields(rdt_data);
+
+	return true;
 }
 
 /*
@@ -4632,6 +4736,10 @@ adjust_xid_advance_interval(RetainDeadTuplesData *rdt_data, bool new_xid_found)
 		 */
 		rdt_data->xid_advance_interval = Min(rdt_data->xid_advance_interval * 2,
 											 max_interval);
+
+		/* Ensure the wait time remains within the maximum limit */
+		rdt_data->xid_advance_interval = Min(rdt_data->xid_advance_interval,
+											 max_conflict_retention_duration);
 	}
 	else
 	{
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 8605776ad86..6a7d2076385 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -116,6 +116,7 @@ static const SlotInvalidationCauseMap SlotInvalidationCauses[] = {
 	{RS_INVAL_HORIZON, "rows_removed"},
 	{RS_INVAL_WAL_LEVEL, "wal_level_insufficient"},
 	{RS_INVAL_IDLE_TIMEOUT, "idle_timeout"},
+	{RS_INVAL_CONFLICT_RETENTION_DURATION, "conflict_retention_exceeds_max_duration"},
 };
 
 /*
@@ -1665,6 +1666,11 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
 								 "idle_replication_slot_timeout");
 				break;
 			}
+		case RS_INVAL_CONFLICT_RETENTION_DURATION:
+			appendStringInfo(&err_detail,
+							 _("The duration for retaining conflict information exceeds the configured \"%s\" limit of %d milliseconds"),
+							 "max_conflict_retention_duration", max_conflict_retention_duration);
+			break;
 		case RS_INVAL_NONE:
 			pg_unreachable();
 	}
@@ -1783,6 +1789,12 @@ DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s,
 		}
 	}
 
+	if (possible_causes & RS_INVAL_CONFLICT_RETENTION_DURATION)
+	{
+		if (IsSlotForConflictCheck(NameStr(s->data.name)))
+			return RS_INVAL_CONFLICT_RETENTION_DURATION;
+	}
+
 	return RS_INVAL_NONE;
 }
 
@@ -2044,6 +2056,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
  * - RS_INVAL_WAL_LEVEL: is logical and wal_level is insufficient
  * - RS_INVAL_IDLE_TIMEOUT: has been idle longer than the configured
  *   "idle_replication_slot_timeout" duration.
+ * - RS_INVAL_CONFLICT_RETENTION_DURATION: is "pg_conflict_detection"
  *
  * Note: This function attempts to invalidate the slot for multiple possible
  * causes in a single pass, minimizing redundant iterations. The "cause"
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index d14b1678e7f..580762c6a00 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -3388,6 +3388,19 @@ struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"max_conflict_retention_duration",
+			PGC_SIGHUP,
+			REPLICATION_SUBSCRIBERS,
+			gettext_noop("Maximum duration for retaining information used in logical replication conflict detection."),
+			NULL,
+			GUC_UNIT_MS
+		},
+		&max_conflict_retention_duration,
+		0, 0, INT_MAX,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"log_rotation_age", PGC_SIGHUP, LOGGING_WHERE,
 			gettext_noop("Sets the amount of time to wait before forcing "
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index a9d8293474a..45c9e0a16a0 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -396,6 +396,7 @@
 					# (change requires restart)
 #max_sync_workers_per_subscription = 2	# taken from max_logical_replication_workers
 #max_parallel_apply_workers_per_subscription = 2	# taken from max_logical_replication_workers
+#max_conflict_retention_duration = 0	# in milliseconds; 0 disables
 
 
 #------------------------------------------------------------------------------
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 118d6da1ace..c284b4d2c6d 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5696,9 +5696,9 @@
   proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => 'oid',
-  proallargtypes => '{oid,oid,oid,int4,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz,text}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{subid,subid,relid,pid,leader_pid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time,worker_type}',
+  proallargtypes => '{oid,oid,oid,int4,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz,text,bool}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{subid,subid,relid,pid,leader_pid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time,worker_type,retain_dead_tuples}',
   prosrc => 'pg_stat_get_subscription' },
 { oid => '2026', descr => 'statistics: current backend PID',
   proname => 'pg_backend_pid', provolatile => 's', proparallel => 'r',
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index b29453e8e4f..6e3007db5f0 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -15,6 +15,7 @@
 extern PGDLLIMPORT int max_logical_replication_workers;
 extern PGDLLIMPORT int max_sync_workers_per_subscription;
 extern PGDLLIMPORT int max_parallel_apply_workers_per_subscription;
+extern PGDLLIMPORT int max_conflict_retention_duration;
 
 extern void ApplyLauncherRegister(void);
 extern void ApplyLauncherMain(Datum main_arg);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index e8fc342d1a9..2f459404c9e 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -66,10 +66,12 @@ typedef enum ReplicationSlotInvalidationCause
 	RS_INVAL_WAL_LEVEL = (1 << 2),
 	/* idle slot timeout has occurred */
 	RS_INVAL_IDLE_TIMEOUT = (1 << 3),
+	/* duration of conflict info retention exceeds the maximum limit */
+	RS_INVAL_CONFLICT_RETENTION_DURATION = (1 << 4),
 } ReplicationSlotInvalidationCause;
 
 /* Maximum number of invalidation causes */
-#define	RS_INVAL_MAX_CAUSES 4
+#define	RS_INVAL_MAX_CAUSES 5
 
 /*
  * On-Disk data of a replication slot, preserved across restarts.
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 7c0204dd6f4..5bcd5c84844 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -97,6 +97,12 @@ typedef struct LogicalRepWorker
 	 */
 	TransactionId oldest_nonremovable_xid;
 
+	/*
+	 * Indicates whether the apply worker has stopped retaining conflict
+	 * information. This is used only when retain_dead_tuples is enabled.
+	 */
+	bool		stop_conflict_info_retention;
+
 	/* Stats. */
 	XLogRecPtr	last_lsn;
 	TimestampTz last_send_time;
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 6509fda77a9..30dafcfe10c 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2169,9 +2169,10 @@ pg_stat_subscription| SELECT su.oid AS subid,
     st.last_msg_send_time,
     st.last_msg_receipt_time,
     st.latest_end_lsn,
-    st.latest_end_time
+    st.latest_end_time,
+    st.retain_dead_tuples
    FROM (pg_subscription su
-     LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, leader_pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time, worker_type) ON ((st.subid = su.oid)));
+     LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, leader_pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time, worker_type, retain_dead_tuples) ON ((st.subid = su.oid)));
 pg_stat_subscription_stats| SELECT ss.subid,
     s.subname,
     ss.apply_error_count,
diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl
index 36aeb14c563..aa6ef19dd70 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -214,6 +214,10 @@ ok( $node_B->poll_query_until(
 	),
 	"the xmin value of slot 'pg_conflict_detection' is valid on Node B");
 
+my $result = $node_B->safe_psql('postgres',
+	"SELECT retain_dead_tuples FROM pg_stat_subscription WHERE subname='$subname_BA';");
+is($result, qq(t), 'worker on node B retains conflict information');
+
 ##################################################
 # Check that the retain_dead_tuples option can be enabled only for disabled
 # subscriptions. Validate the NOTICE message during the subscription DDL, and
@@ -254,6 +258,10 @@ ok( $node_A->poll_query_until(
 	),
 	"the xmin value of slot 'pg_conflict_detection' is valid on Node A");
 
+$result = $node_A->safe_psql('postgres',
+	"SELECT retain_dead_tuples FROM pg_stat_subscription WHERE subname='$subname_AB';");
+is($result, qq(t), 'worker on node A retains conflict information');
+
 ##################################################
 # Check the WARNING when changing the origin to ANY, if retain_dead_tuples is
 # enabled. This warns of the possibility of receiving changes from origins
@@ -281,7 +289,7 @@ $node_A->psql('postgres',
 $node_A->safe_psql('postgres', "INSERT INTO tab VALUES (1, 1), (2, 2);");
 $node_A->wait_for_catchup($subname_BA);
 
-my $result = $node_B->safe_psql('postgres', "SELECT * FROM tab;");
+$result = $node_B->safe_psql('postgres', "SELECT * FROM tab;");
 is($result, qq(1|1
 2|2), 'check replicated insert on node B');
 
-- 
2.50.1.windows.1