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_conflict_info 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_conflict_info 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_conflict_info 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_conflict_info' 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_conflict_info is enabled
for the associated subscription, and the retention duration for conflict
detection by the apply worker has not exceeded max_conflict_retention_duration.

Author: Zhijie Hou

doc/src/sgml/config.sgml

@@ -5379,6 +5379,47 @@ 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 (in milliseconds) for which conflict
+        information can be retained for conflict detection by the apply worker.
+        The default value is <literal>0</literal>, indicating that conflict
+        information is retained until it is no longer needed for detection
+        purposes.
+       </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_conflict_info</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_conflict_info</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_conflict_info</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>
 

doc/src/sgml/monitoring.sgml

@@ -2109,6 +2109,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_conflict_info</structfield> <type>boolean</type>
+      </para>
+      <para>
+       True if <link linkend="sql-createsubscription-params-with-retain-conflict-info"><literal>retain_conflict_info</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>

doc/src/sgml/system-views.sgml

@@ -2936,6 +2936,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-conflict-info"><literal>retain_conflict_info</literal></link>
+          is enabled.
+         </para>
+        </listitem>
        </itemizedlist>
       </para></entry>
      </row>

src/backend/catalog/system_views.sql

@@ -993,7 +993,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_conflict_info
     FROM pg_subscription su
             LEFT JOIN pg_stat_get_subscription(NULL) st
                       ON (st.subid = su.oid);

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;
 
@@ -446,6 +449,7 @@ 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->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -1157,7 +1161,8 @@ ApplyLauncherMain(Datum main_arg)
 		MemoryContext oldctx;
 		long		wait_time = DEFAULT_NAPTIME_PER_CYCLE;
 		bool		can_advance_xmin = true;
-		bool		retain_conflict_info = false;
+		int			nretain_conflict_info = 0;
+		int			nstop_retention = 0;
 		FullTransactionId xmin = InvalidFullTransactionId;
 
 		CHECK_FOR_INTERRUPTS();
@@ -1186,7 +1191,7 @@ ApplyLauncherMain(Datum main_arg)
 			 */
 			if (sub->retainconflictinfo)
 			{
-				retain_conflict_info = true;
+				nretain_conflict_info++;
 				can_advance_xmin &= sub->enabled;
 
 				/*
@@ -1212,22 +1217,32 @@ ApplyLauncherMain(Datum main_arg)
 				 * the new xmin for advancing the replication slot used in
 				 * conflict detection.
 				 */
-				if (sub->retainconflictinfo && can_advance_xmin)
+				if (sub->retainconflictinfo)
 				{
 					FullTransactionId nonremovable_xid;
+					bool		stop_conflict_info_retention;
 
 					SpinLockAcquire(&w->relmutex);
 					nonremovable_xid = w->oldest_nonremovable_xid;
+					stop_conflict_info_retention = w->stop_conflict_info_retention;
 					SpinLockRelease(&w->relmutex);
 
+					/*
+					 * Skip collecting oldest_nonremovable_xid for workers
+					 * that have stopped conflict retention.
+					 */
+					if (stop_conflict_info_retention)
+						nstop_retention++;
+
 					/*
 					 * Stop advancing xmin if an invalid non-removable
 					 * transaction ID is found, otherwise update xmin.
 					 */
-					if (!FullTransactionIdIsValid(nonremovable_xid))
+					else if (!FullTransactionIdIsValid(nonremovable_xid))
 						can_advance_xmin = false;
-					else if (!FullTransactionIdIsValid(xmin) ||
-							 FullTransactionIdPrecedes(nonremovable_xid, xmin))
+					else if (can_advance_xmin &&
+							 (!FullTransactionIdIsValid(xmin) ||
+							  FullTransactionIdPrecedes(nonremovable_xid, xmin)))
 						xmin = nonremovable_xid;
 				}
 
@@ -1272,11 +1287,34 @@ 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)
+		{
+			ReplicationSlotRelease();
+			InvalidateObsoleteReplicationSlots(RS_INVAL_CONFLICT_RETENTION_DURATION,
+											   InvalidXLogRecPtr, InvalidOid,
+											   InvalidTransactionId);
+		}
+
 		/*
 		 * Maintain the xmin value of the replication slot for conflict
 		 * detection if needed.
 		 */
-		if (retain_conflict_info)
+		else if (nretain_conflict_info)
 		{
 			if (can_advance_xmin)
 				advance_conflict_slot_xmin(xmin);
@@ -1470,7 +1508,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;
@@ -1547,6 +1585,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->subretainconflictinfo &&
+				!worker.stop_conflict_info_retention;
+
+			ReleaseSysCache(tup);
+		}
+		else
+			nulls[10] = true;
+
 		tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
 							 values, nulls);