v70-0002-Resume-retaining-the-information-for-conflict-de.patch

From 44162d3da40b3ba7f295da79efe8ab5507e632e1 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Mon, 1 Sep 2025 12:40:36 +0800
Subject: [PATCH v70 2/3] Resume retaining the information for conflict
 detection

The patch allows the launcher to re-initialized invalidated slot, if at
least one apply worker has confirmed that the retention duration is now within
the max_retention_duration.
---
 doc/src/sgml/ref/create_subscription.sgml  |   9 +-
 src/backend/commands/subscriptioncmds.c    |   9 +-
 src/backend/replication/logical/launcher.c |  48 ++++-
 src/backend/replication/logical/worker.c   | 206 ++++++++++++++++++---
 src/include/replication/worker_internal.h  |   6 +
 5 files changed, 240 insertions(+), 38 deletions(-)

diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index fc314437311..ee22d0fb3ca 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -538,10 +538,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
           <literal>retain_dead_tuples</literal> is enabled, confirm that the
           retention duration has exceeded the
           <literal>max_retention_duration</literal> set within the corresponding
-          subscription. The retention will not be automatically resumed unless a
-          new subscription is created with <literal>retain_dead_tuples =
-          true</literal>, or the user manually re-enables
-          <literal>retain_dead_tuples</literal>.
+          subscription. The retention will be automatically resumed once at least
+          one apply worker confirms that the retention duration is within the
+          specified limit, or a new subscription is created with
+          <literal>retain_dead_tuples = true</literal>, or the user manually
+          re-enables <literal>retain_dead_tuples</literal>.
          </para>
          <para>
           Note that overall retention will not stop if other subscriptions that
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 82cf65fae73..cc622fce58f 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -854,7 +854,14 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 
 	pgstat_create_subscription(subid);
 
-	if (opts.enabled)
+	/*
+	 * If the subscription is enabled, notify the launcher to start the apply
+	 * worker.
+	 *
+	 * If the subscription has retain_dead_tuples enabled, notify the launcher
+	 * to create or resume the conflict detection slot.
+	 */
+	if (opts.enabled || opts.retaindeadtuples)
 		ApplyLauncherWakeupAtCommit();
 
 	ObjectAddressSet(myself, SubscriptionRelationId, subid);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index add2e2e066c..226421e3dfa 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -101,7 +101,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 void compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin);
+static void compute_min_nonremovable_xid(LogicalRepWorker *worker,
+										 bool can_update_xmin,
+										 TransactionId *xmin);
 static bool acquire_conflict_slot_if_exists(void);
 static void update_conflict_slot_xmin(TransactionId new_xmin);
 static void init_conflict_slot_xmin(void);
@@ -468,6 +470,7 @@ retry:
 	worker->oldest_nonremovable_xid = retain_dead_tuples
 		? MyReplicationSlot->data.xmin
 		: InvalidTransactionId;
+	worker->wait_for_initial_xid = false;
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -1270,10 +1273,8 @@ ApplyLauncherMain(Datum main_arg)
 				 * required for conflict detection among all running apply
 				 * workers.
 				 */
-				if (sub->retaindeadtuples &&
-					sub->retentionactive &&
-					can_update_xmin)
-					compute_min_nonremovable_xid(w, &xmin);
+				if (sub->retaindeadtuples && sub->retentionactive)
+					compute_min_nonremovable_xid(w, can_update_xmin, &xmin);
 
 				/* worker is running already */
 				continue;
@@ -1382,11 +1383,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.
+ *
+ * Additionally, if an apply worker has an invalid XID and is requesting to
+ * resume retention, assign the slot's xmin value to it.
  */
 static void
-compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
+compute_min_nonremovable_xid(LogicalRepWorker *worker, bool can_update_xmin,
+							 TransactionId *xmin)
 {
 	TransactionId nonremovable_xid;
+	bool		wait_for_xid;
 
 	Assert(worker != NULL);
 
@@ -1398,16 +1404,42 @@ compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
 
 	SpinLockAcquire(&worker->relmutex);
 	nonremovable_xid = worker->oldest_nonremovable_xid;
+	wait_for_xid = worker->wait_for_initial_xid;
 	SpinLockRelease(&worker->relmutex);
 
 	/*
-	 * Return if the apply worker has stopped retention concurrently.
+	 * Assign slot.xmin to the apply worker's oldest_nonremovable_xid if
+	 * requested. This ensures the apply worker continues to maintain the
+	 * oldest_nonremovable_xid (see resume_conflict_info_retention).
+	 */
+	if (wait_for_xid)
+	{
+		nonremovable_xid = MyReplicationSlot->data.xmin;
+
+		Assert(TransactionIdIsValid(nonremovable_xid));
+
+		SpinLockAcquire(&worker->relmutex);
+		worker->oldest_nonremovable_xid = nonremovable_xid;
+		SpinLockRelease(&worker->relmutex);
+
+		/* Notify the apply worker to start the next cycle of management */
+		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+		logicalrep_worker_wakeup_ptr(worker);
+		LWLockRelease(LogicalRepWorkerLock);
+	}
+
+	/*
+	 * Return if the apply worker has stopped retention concurrently and has not
+	 * yet resumed.
 	 *
 	 * Although this function is invoked only when retentionactive is true,
 	 * the apply worker might stop retention after the launcher fetches the
 	 * retentionactive flag.
 	 */
-	if (!TransactionIdIsValid(nonremovable_xid))
+	else if (!TransactionIdIsValid(nonremovable_xid))
+		return;
+
+	if (!can_update_xmin)
 		return;
 
 	if (!TransactionIdIsValid(*xmin) ||
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index f5f9183bdaf..0f014f44db4 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -181,6 +181,19 @@
  *   pg_subscription.subretentionactive is updated to false within a new
  *   transaction, and oldest_nonremovable_xid is set to InvalidTransactionId.
  *
+ * - RDT_RESUME_CONFLICT_INFO_RETENTION:
+ *   This phase is required only when max_retention_duration is defined. We
+ *   enter this phase if the retention was previously stopped, and the time
+ *   required to advance the non-removable transaction ID in the
+ *   RDT_WAIT_FOR_LOCAL_FLUSH phase has decreased to within acceptable limits
+ *   (or if max_retention_duration is set to 0). During this phase,
+ *   pg_subscription.subretentionactive is updated to true within a new
+ *   transaction, and we wait for the launcher to initialize the
+ *   oldest_nonremovable_xid before proceeding to RDT_GET_CANDIDATE_XID phase.
+ *   Note that the state could transition to RDT_RESUME_CONFLICT_INFO_RETENTION
+ *   at any phase if the retention has been stopped, but max_retention_duration
+ *   is now set to 0.
+ *
  * The overall state progression is: GET_CANDIDATE_XID ->
  * REQUEST_PUBLISHER_STATUS -> WAIT_FOR_PUBLISHER_STATUS -> (loop to
  * REQUEST_PUBLISHER_STATUS till concurrent remote transactions end) ->
@@ -382,7 +395,8 @@ typedef enum
 	RDT_REQUEST_PUBLISHER_STATUS,
 	RDT_WAIT_FOR_PUBLISHER_STATUS,
 	RDT_WAIT_FOR_LOCAL_FLUSH,
-	RDT_STOP_CONFLICT_INFO_RETENTION
+	RDT_STOP_CONFLICT_INFO_RETENTION,
+	RDT_RESUME_CONFLICT_INFO_RETENTION,
 } RetainDeadTuplesPhase;
 
 /*
@@ -569,6 +583,10 @@ static void wait_for_publisher_status(RetainDeadTuplesData *rdt_data,
 static void wait_for_local_flush(RetainDeadTuplesData *rdt_data);
 static bool should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data);
 static void stop_conflict_info_retention(RetainDeadTuplesData *rdt_data);
+static bool should_resume_retention_immediately(RetainDeadTuplesData *rdt_data,
+												bool status_received);
+static void resume_conflict_info_retention(RetainDeadTuplesData *rdt_data);
+static void update_retention_status(bool active);
 static void reset_retention_data_fields(RetainDeadTuplesData *rdt_data);
 static void adjust_xid_advance_interval(RetainDeadTuplesData *rdt_data,
 										bool new_xid_found);
@@ -4339,6 +4357,13 @@ maybe_advance_nonremovable_xid(RetainDeadTuplesData *rdt_data,
 	if (!can_advance_nonremovable_xid(rdt_data))
 		return;
 
+	/*
+	 * Resume retention immediately if required. (See
+	 * should_resume_retention_immediately() for details).
+	 */
+	if (should_resume_retention_immediately(rdt_data, status_received))
+		rdt_data->phase = RDT_RESUME_CONFLICT_INFO_RETENTION;
+
 	process_rdt_phase_transition(rdt_data, status_received);
 }
 
@@ -4361,10 +4386,6 @@ can_advance_nonremovable_xid(RetainDeadTuplesData *rdt_data)
 	if (!MySubscription->retaindeadtuples)
 		return false;
 
-	/* No need to advance if we have already stopped retaining */
-	if (!MySubscription->retentionactive)
-		return false;
-
 	return true;
 }
 
@@ -4393,6 +4414,9 @@ process_rdt_phase_transition(RetainDeadTuplesData *rdt_data,
 		case RDT_STOP_CONFLICT_INFO_RETENTION:
 			stop_conflict_info_retention(rdt_data);
 			break;
+		case RDT_RESUME_CONFLICT_INFO_RETENTION:
+			resume_conflict_info_retention(rdt_data);
+			break;
 	}
 }
 
@@ -4649,6 +4673,18 @@ wait_for_local_flush(RetainDeadTuplesData *rdt_data)
 	if (last_flushpos < rdt_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, resume the rentention.
+	 */
+	if (!MySubscription->retentionactive)
+	{
+		rdt_data->phase = RDT_RESUME_CONFLICT_INFO_RETENTION;
+		process_rdt_phase_transition(rdt_data, false);
+		return;
+	}
+
 	/*
 	 * Reaching here means the remote WAL position has been received, and all
 	 * transactions up to that position on the publisher have been applied and
@@ -4679,9 +4715,8 @@ wait_for_local_flush(RetainDeadTuplesData *rdt_data)
  * RDT_STOP_CONFLICT_INFO_RETENTION phase and return true. Otherwise, return
  * false.
  *
- * Note: Retention won't be resumed automatically. The user must manually
- * disable retain_dead_tuples and re-enable it after confirming that the
- * replication slot maintained by the launcher has been dropped.
+ * The retention will resume automatically if the worker has confirmed that the
+ * retention duration is now within the max_retention_duration.
  */
 static bool
 should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data)
@@ -4712,10 +4747,16 @@ should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data)
 									rdt_data->table_sync_wait_time))
 		return false;
 
-	rdt_data->phase = RDT_STOP_CONFLICT_INFO_RETENTION;
+	/* Stop retention if not yet */
+	if (MySubscription->retentionactive)
+	{
+		rdt_data->phase = RDT_STOP_CONFLICT_INFO_RETENTION;
 
-	/* process the next phase */
-	process_rdt_phase_transition(rdt_data, false);
+		/* process the next phase */
+		process_rdt_phase_transition(rdt_data, false);
+	}
+
+	reset_retention_data_fields(rdt_data);
 
 	return true;
 }
@@ -4725,6 +4766,131 @@ should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data)
  */
 static void
 stop_conflict_info_retention(RetainDeadTuplesData *rdt_data)
+{
+	update_retention_status(false);
+
+	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+	MyLogicalRepWorker->oldest_nonremovable_xid = InvalidTransactionId;
+	SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+	ereport(LOG,
+			errmsg("logical replication worker for subscription \"%s\" has stopped retaining the information for detecting conflicts",
+				   MySubscription->name),
+			errdetail("Retention of information used for conflict detection has exceeded max_retention_duration of %u ms.",
+					  MySubscription->maxretention));
+
+	reset_retention_data_fields(rdt_data);
+
+	/* process the next phase */
+	process_rdt_phase_transition(rdt_data, false);
+}
+
+/*
+ * Check whether retention should be resumed immediately if it has been
+ * previously stopped, but max_retention_duration is now set to 0.
+ */
+static bool
+should_resume_retention_immediately(RetainDeadTuplesData *rdt_data, bool status_received)
+{
+	/* Return false if retention is already being resumed */
+	if (rdt_data->phase == RDT_RESUME_CONFLICT_INFO_RETENTION)
+		return false;
+
+	/* Return false if max_retention_duration is not 0 */
+	if (MySubscription->maxretention)
+		return false;
+
+	/*
+	 * Do not resume when waiting for publisher status, as doing so may result
+	 * in the message being processed after the data and phase have been
+	 * reset, potentially causing it to be mistakenly identified as a new
+	 * message. This could lead to the premature advancement of
+	 * oldest_nonremovable_xid.
+	 */
+	if (rdt_data->phase == RDT_WAIT_FOR_PUBLISHER_STATUS &&
+		!status_received)
+		return false;
+
+	/*
+	 * Resume retention if we are in the process of stopping or have already
+	 * stopped retention.
+	 */
+	return rdt_data->phase == RDT_STOP_CONFLICT_INFO_RETENTION ||
+		!MySubscription->retentionactive;
+}
+
+/*
+ * Workhorse for the RDT_RESUME_CONFLICT_INFO_RETENTION phase.
+ */
+static void
+resume_conflict_info_retention(RetainDeadTuplesData *rdt_data)
+{
+	TransactionId nonremovable_xid;
+
+	/* Update the pg_subscription.retentionactive if not yet */
+	if (!MySubscription->retentionactive)
+	{
+		update_retention_status(true);
+
+		SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+		MyLogicalRepWorker->wait_for_initial_xid = true;
+		SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+		ereport(LOG,
+				errmsg("logical replication worker for subscription \"%s\" will resume retaining the information for detecting conflicts",
+					   MySubscription->name),
+				MySubscription->maxretention
+				? errdetail("Retention of information used for conflict detection is now within the max_retention_duration of %u ms.",
+							MySubscription->maxretention)
+				: errdetail("Retention of information used for conflict detection is now indefinite."));
+	}
+
+	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+	nonremovable_xid = MyLogicalRepWorker->oldest_nonremovable_xid;
+	SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+	/*
+	 * Return if the launcher has not initialized oldest_nonremovable_xid.
+	 *
+	 * It might seem feasible to directly check the conflict detection
+	 * slot.xmin instead of relying on the launcher to assign the worker's
+	 * oldest_nonremovable_xid; however, that could lead to a race condition
+	 * where slot.xmin is set to InvalidTransactionId immediately after the
+	 * check. In such cases, oldest_nonremovable_xid would no longer be
+	 * protected by a replication slot and could become unreliable if a
+	 * wraparound occurs.
+	 */
+	if (!TransactionIdIsValid(nonremovable_xid))
+		return;
+
+	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+	MyLogicalRepWorker->wait_for_initial_xid = false;
+	SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+	/*
+	 * Proceed to the next phase if either the launcher has initialized
+	 * slot.xmin and assigned it to oldest_nonremovable_xid, or retention has
+	 * not been stopped yet. The latter situation arises when transitioning
+	 * from the RDT_STOP_CONFLICT_INFO_RETENTION phase but subretentionactive
+	 * has not been updated due to the inability to start a new transaction
+	 * (see stop_conflict_info_retention).
+	 */
+	Assert(MySubscription->retentionactive);
+
+	reset_retention_data_fields(rdt_data);
+
+	/* process the next phase */
+	process_rdt_phase_transition(rdt_data, false);
+}
+
+/*
+ * Update pg_subscription.subretentionactive to the given value within a new
+ * transaction.
+ *
+ * Skip the update if currently within an existing transaction.
+ */
+static void
+update_retention_status(bool active)
 {
 	/*
 	 * Do not update the catalog during an active transaction. The transaction
@@ -4742,26 +4908,16 @@ stop_conflict_info_retention(RetainDeadTuplesData *rdt_data)
 	 */
 	PushActiveSnapshot(GetTransactionSnapshot());
 
-	/* Set pg_subscription.subretentionactive to false */
-	UpdateDeadTupleRetentionStatus(MySubscription->oid, false);
+	/* Update pg_subscription.subretentionactive */
+	UpdateDeadTupleRetentionStatus(MySubscription->oid, active);
 
 	PopActiveSnapshot();
 	CommitTransactionCommand();
 
-	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
-	MyLogicalRepWorker->oldest_nonremovable_xid = InvalidTransactionId;
-	SpinLockRelease(&MyLogicalRepWorker->relmutex);
-
-	ereport(LOG,
-			errmsg("logical replication worker for subscription \"%s\" has stopped retaining the information for detecting conflicts",
-				   MySubscription->name),
-			errdetail("Retention of information used for conflict detection has exceeded max_retention_duration of %u ms.",
-					  MySubscription->maxretention));
-
-	/* Notify launcher to update the conflict slot */
+	/* Notify launcher to update the xmin of the conflict slot */
 	ApplyLauncherWakeup();
 
-	reset_retention_data_fields(rdt_data);
+	MySubscription->retentionactive = active;
 }
 
 /*
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 62ea1a00580..58f2dcc7f0b 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -100,6 +100,12 @@ typedef struct LogicalRepWorker
 	 */
 	TransactionId oldest_nonremovable_xid;
 
+	/*
+	 * Indicates whether the apply worker is resuming retention and is waiting
+	 * for the launcher to initialize oldest_nonremovable_xid.
+	 */
+	bool	wait_for_initial_xid;
+
 	/* Stats. */
 	XLogRecPtr	last_lsn;
 	TimestampTz last_send_time;
-- 
2.51.0.windows.1