From 6b402bd279231d68e51ac3801e6600da70503141 Mon Sep 17 00:00:00 2001
From: Zhijie Hou <houzj.fnst@fujitsu.com>
Date: Wed, 2 Sep 2026 17:24:26 +0800
Subject: [PATCH v3 1/2] Re-initialize conflict slot xmin when a database newly
 needs retention

Since the retain_dead_tuples feature maintains one replication slot for all
subscriptions, if two subscriptions are created at different times, the later
one may cause slot.xmin to move backwards.

For example, consider two subscriptions for databases A and B. The oldest XID
in database A is 700, while the oldest XID in database B is 500 (due to a
long-running transaction in that database). If subscription A is created first,
conflict_detection_slot.xmin advances to 700. When subscription B is later
created, it sees that the oldest XID in its own database is 500 -- which is older
than the current slot.xmin. This currently causes an Assert failure in the apply
worker. If the Assert is removed, it would allow slot.xmin to move backwards.

Moving slot.xmin backwards is actually correct behavior for subscription of
database B: the long-running transaction in database B is a candidate that could
generate dead tuples needed for update_deleted conflict detection, so slot.xmin
should not advance beyond that XID. However, instead of letting the worker
handle this (which could delay the slot.xmin update), we should have the
launcher detect the new subscription and adjust the slot directly.

The patch fix this by tracking the set of databases with actively-retaining
subscriptions in the launcher, and when a database newly appears in the set,
re-initialize the slot's xmin to the cluster-wide safe decoding horizon before
launching any workers.  The horizon accounts for all running transactions
cluster-wide, so it is a safe seed for every database.

Reported-by: Nisha Moond <nisha.moond412@gmail.com>
Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Reviewed-by: Nisha Moond <nisha.moond412@gmail.com>
Reviewed-by: shveta malik <shveta.malik@gmail.com>
Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Discussion: https://postgr.es/m/TY4PR01MB177182F547A62FC2666EC04EC94B72@TY4PR01MB17718.jpnprd01.prod.outlook.com
Backpatch-through: 19, where it was introduced
---
 src/backend/replication/logical/launcher.c | 173 ++++++++++++++-------
 src/test/subscription/t/035_conflicts.pl   |  73 +++++++++
 2 files changed, 191 insertions(+), 55 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 313e31ff2e3..e011789fe36 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -111,10 +111,12 @@ 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 bool collect_min_nonremovable_xid(Subscription *sub,
+										 LogicalRepWorker *worker,
+										 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);
+static void reset_conflict_slot_xmin_to_safe_horizon(void);
 
 
 /*
@@ -1204,6 +1206,8 @@ ApplyLauncherWakeup(void)
 void
 ApplyLauncherMain(Datum main_arg)
 {
+	List	   *retained_dbids;
+
 	ereport(DEBUG1,
 			(errmsg_internal("logical replication launcher started")));
 
@@ -1222,6 +1226,13 @@ ApplyLauncherMain(Datum main_arg)
 	 */
 	BackgroundWorkerInitializeConnection(NULL, NULL, 0);
 
+	/*
+	 * Databases with actively-retaining subscriptions as of the previous
+	 * cycle.  Local memory only, so every such database counts as new after a
+	 * launcher restart, resetting the slot's xmin once at startup.
+	 */
+	retained_dbids = NIL;
+
 	/*
 	 * Acquire the conflict detection slot at startup to ensure it can be
 	 * dropped if no longer needed after a restart.
@@ -1239,7 +1250,9 @@ ApplyLauncherMain(Datum main_arg)
 		long		wait_time = DEFAULT_NAPTIME_PER_CYCLE;
 		bool		can_update_xmin = true;
 		bool		retain_dead_tuples = false;
+		bool		reset_done = false;
 		TransactionId xmin = InvalidTransactionId;
+		List	   *current_dbids = NIL;
 
 		CHECK_FOR_INTERRUPTS();
 
@@ -1288,6 +1301,10 @@ ApplyLauncherMain(Datum main_arg)
 
 				if (sub->retentionactive)
 				{
+					/* Remember the retained databases for the next cycle. */
+					current_dbids = list_append_unique_oid(current_dbids,
+														   sub->dbid);
+
 					/*
 					 * Can't advance xmin of the slot unless all the
 					 * subscriptions actively retaining dead tuples are
@@ -1301,11 +1318,24 @@ ApplyLauncherMain(Datum main_arg)
 					can_update_xmin &= sub->enabled;
 
 					/*
-					 * Initialize the slot once the subscription activates
-					 * retention.
+					 * Reset the slot's xmin to the safe decoding horizon if
+					 * it is not valid, or if this database newly appears
+					 * among the retained databases: the xmin may be newer
+					 * than the new database's oldest active transaction ID,
+					 * violating the per-database invariant checked in
+					 * get_candidate_xid().  The set is keyed on
+					 * subretentionactive, so that a subscription resuming
+					 * retention is also treated as a new arrival.  One call
+					 * per cycle is enough, as the horizon is a safe seed for
+					 * every database.
 					 */
-					if (!TransactionIdIsValid(MyReplicationSlot->data.xmin))
-						init_conflict_slot_xmin();
+					if (!reset_done &&
+						(!TransactionIdIsValid(MyReplicationSlot->data.xmin) ||
+						 !list_member_oid(retained_dbids, sub->dbid)))
+					{
+						reset_conflict_slot_xmin_to_safe_horizon();
+						reset_done = true;
+					}
 				}
 			}
 
@@ -1316,37 +1346,20 @@ ApplyLauncherMain(Datum main_arg)
 			w = logicalrep_worker_find(WORKERTYPE_APPLY, sub->oid, InvalidOid,
 									   false);
 
-			if (w != NULL)
-			{
-				/*
-				 * Compute the minimum xmin required to protect dead tuples
-				 * required for conflict detection among all running apply
-				 * workers. This computation is performed while holding
-				 * LogicalRepWorkerLock to prevent accessing invalid worker
-				 * data, in scenarios where a worker might exit and reset its
-				 * state concurrently.
-				 */
-				if (sub->retaindeadtuples &&
-					sub->retentionactive &&
-					can_update_xmin)
-					compute_min_nonremovable_xid(w, &xmin);
-
-				LWLockRelease(LogicalRepWorkerLock);
-
-				/* worker is running already */
-				continue;
-			}
+			/*
+			 * Compute the minimum xmin required to protect dead tuples
+			 * required for conflict detection among all running apply
+			 * workers. One subscription that cannot contribute is enough to
+			 * prevent the slot's xmin from advancing, so stop collecting once
+			 * that happens.
+			 */
+			can_update_xmin = can_update_xmin &&
+				collect_min_nonremovable_xid(sub, w, &xmin);
 
 			LWLockRelease(LogicalRepWorkerLock);
 
-			/*
-			 * Can't advance xmin of the slot unless all the workers
-			 * corresponding to subscriptions actively retaining dead tuples
-			 * are running, disabling the further computation of the minimum
-			 * nonremovable xid.
-			 */
-			if (sub->retaindeadtuples && sub->retentionactive)
-				can_update_xmin = false;
+			if (w != NULL)
+				continue;		/* worker is running already */
 
 			/*
 			 * If the worker is eligible to start now, launch it.  Otherwise,
@@ -1414,6 +1427,14 @@ ApplyLauncherMain(Datum main_arg)
 
 		/* Switch back to original memory context. */
 		MemoryContextSwitchTo(oldctx);
+
+		/*
+		 * Remember the current set of retained databases for the next cycle,
+		 * in long-lived memory.
+		 */
+		list_free(retained_dbids);
+		retained_dbids = list_copy(current_dbids);
+
 		/* Clean the temporary memory. */
 		MemoryContextDelete(subctx);
 
@@ -1440,16 +1461,27 @@ 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.
+ * Fold this subscription's minimum non-removable transaction ID into *xmin,
+ * for subscriptions that are actively retaining conflict information.
+ *
+ * Returns false if the slot's xmin cannot be advanced in this cycle, which is
+ * the case when the apply worker is not running (worker is NULL) or does not
+ * yet hold a valid oldest_nonremovable_xid. Subscriptions that are not
+ * retaining place no constraint on the xmin and return true.
+ *
+ * The caller must hold LogicalRepWorkerLock, to prevent accessing invalid
+ * worker data in scenarios where a worker might exit and reset its state
+ * concurrently.
  */
-static void
-compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
+static bool
+collect_min_nonremovable_xid(Subscription *sub, LogicalRepWorker *worker,
+							 TransactionId *xmin)
 {
-	TransactionId nonremovable_xid;
+	TransactionId nonremovable_xid = InvalidTransactionId;
 
-	Assert(worker != NULL);
+	/* Skip subscriptions that are not actively retaining dead tuples */
+	if (!sub->retaindeadtuples || !sub->retentionactive)
+		return true;
 
 	/*
 	 * The replication slot for conflict detection must be created before the
@@ -1457,23 +1489,32 @@ compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
 	 */
 	Assert(MyReplicationSlot);
 
-	SpinLockAcquire(&worker->relmutex);
-	nonremovable_xid = worker->oldest_nonremovable_xid;
-	SpinLockRelease(&worker->relmutex);
+	if (worker)
+	{
+		SpinLockAcquire(&worker->relmutex);
+		nonremovable_xid = worker->oldest_nonremovable_xid;
+		SpinLockRelease(&worker->relmutex);
+	}
 
 	/*
-	 * Return if the apply worker has stopped retention concurrently.
+	 * Skip the xmin update if the apply worker's oldest_nonremovable_xid is
+	 * invalid, or if the worker is not running (worker is NULL).
 	 *
-	 * Although this function is invoked only when retentionactive is true,
-	 * the apply worker might stop retention after the launcher fetches the
-	 * retentionactive flag.
+	 * Although this function proceeds only when retentionactive is true,
+	 * worker's oldest_nonremovable_xid being invalid can still happen when
+	 * the worker has stopped retention after the launcher fetched
+	 * retentionactive, or when the worker recently enabled retention but
+	 * hasn't restarted yet and the XID is still invalid. In either case, it's
+	 * unnecessary to update the slot's xmin in this cycle.
 	 */
 	if (!TransactionIdIsValid(nonremovable_xid))
-		return;
+		return false;
 
 	if (!TransactionIdIsValid(*xmin) ||
 		TransactionIdPrecedes(nonremovable_xid, *xmin))
 		*xmin = nonremovable_xid;
+
+	return true;
 }
 
 /*
@@ -1530,27 +1571,49 @@ update_conflict_slot_xmin(TransactionId new_xmin)
 }
 
 /*
- * Initialize the xmin for the conflict detection slot.
+ * Reset the xmin of the conflict detection slot to the cluster-wide safe
+ * decoding horizon, which is a safe seed for an apply worker in any
+ * database.
+ *
+ * This is used both for the initial setup and when a database newly appears
+ * among the databases with actively-retaining subscriptions; see
+ * ApplyLauncherMain().  An already-valid xmin is only moved backwards, as
+ * advancing it here would bypass the advancement protocol; regressing is
+ * safe, and the workers will advance the xmin again in later cycles.
  */
 static void
-init_conflict_slot_xmin(void)
+reset_conflict_slot_xmin_to_safe_horizon(void)
 {
 	TransactionId xmin_horizon;
+	TransactionId old_xmin;
 
-	/* Replication slot must exist but shouldn't be initialized. */
-	Assert(MyReplicationSlot &&
-		   !TransactionIdIsValid(MyReplicationSlot->data.xmin));
+	Assert(MyReplicationSlot);
 
 	LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
 	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
 
 	xmin_horizon = GetOldestSafeDecodingTransactionId(false);
+	old_xmin = MyReplicationSlot->data.xmin;
+
+	/*
+	 * Nothing to do if the current xmin is valid and not newer than the
+	 * horizon.
+	 */
+	if (TransactionIdIsValid(old_xmin) &&
+		TransactionIdPrecedesOrEquals(old_xmin, xmin_horizon))
+	{
+		LWLockRelease(ProcArrayLock);
+		LWLockRelease(ReplicationSlotControlLock);
+		return;
+	}
 
 	SpinLockAcquire(&MyReplicationSlot->mutex);
 	MyReplicationSlot->effective_xmin = xmin_horizon;
 	MyReplicationSlot->data.xmin = xmin_horizon;
 	SpinLockRelease(&MyReplicationSlot->mutex);
 
+	elog(DEBUG1, "reset conflict detection slot's xmin to %u", xmin_horizon);
+
 	ReplicationSlotsComputeRequiredXmin(true);
 
 	LWLockRelease(ProcArrayLock);
@@ -1578,7 +1641,7 @@ CreateConflictDetectionSlot(void)
 	ReplicationSlotCreate(CONFLICT_DETECTION_SLOT, false, RS_PERSISTENT, false,
 						  false, false, false);
 
-	init_conflict_slot_xmin();
+	reset_conflict_slot_xmin_to_safe_horizon();
 }
 
 /*
diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl
index 3910a49c0aa..d66e71d1900 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -649,6 +649,79 @@ $result = $node_A->safe_psql('postgres',
 );
 is($result, qq(t), 'retention is active');
 
+###############################################################################
+# Check that the conflict detection slot's xmin is re-initialized when a
+# database newly appears among the databases with retain_dead_tuples
+# subscriptions.
+#
+# The slot's xmin is advanced according to the per-database horizons of the
+# databases seen so far.  Without re-initialization, a worker started in a
+# newly retaining database whose oldest active transaction ID is older would
+# be seeded with a value newer than its database's horizon.
+###############################################################################
+
+# Create a second database on node B, with the same table.
+$node_B->safe_psql('postgres', "CREATE DATABASE dbb");
+$node_B->safe_psql('dbb', "CREATE TABLE tab (a int PRIMARY KEY, b int)");
+
+# Hold a transaction with an assigned transaction ID open in dbb, pinning its
+# oldest active transaction ID.
+my $dbb_session = $node_B->background_psql('dbb');
+$dbb_session->query_until(
+	qr/starting_bg_psql/, q{
+	\echo starting_bg_psql
+	BEGIN;
+	SELECT txid_current();
+});
+
+# Push the transaction ID counter clearly past the pinned transaction ID and
+# wait for the slot's xmin to advance past it.  Only the apply worker in the
+# postgres database drives the slot's xmin here, and postgres has no old
+# transaction running.
+$next_xid = $node_B->safe_psql('postgres', "SELECT txid_current() + 1");
+ok( $node_B->poll_query_until(
+		'postgres',
+		"SELECT xmin::text::bigint >= $next_xid FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"slot xmin advanced past the transaction ID pinned in dbb");
+
+# Create the second retention subscription in dbb.  The launcher must
+# re-initialize the slot's xmin before launching dbb's apply worker.
+my $subname_BA2 = 'tap_sub_b_a2';
+$node_B->safe_psql('dbb',
+	"CREATE SUBSCRIPTION $subname_BA2
+	 CONNECTION '$node_A_connstr application_name=$subname_BA2'
+	 PUBLICATION tap_pub_A
+	 WITH (retain_dead_tuples = true, origin = none)");
+$node_B->wait_for_subscription_sync($node_A, $subname_BA2, 'dbb');
+
+# The slot's xmin must regress to the horizon pinned in dbb.
+ok( $node_B->poll_query_until(
+		'postgres',
+		"SELECT xmin::text::bigint < $next_xid FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"slot xmin regressed to the horizon pinned in dbb");
+
+# Once the pinned transaction commits, the xmin must be able to advance
+# again.
+$dbb_session->query_until(
+	qr/committed/, q{
+	COMMIT;
+	\echo committed
+});
+ok($dbb_session->quit, 'close pinned session');
+
+$next_xid = $node_B->safe_psql('postgres', "SELECT txid_current() + 1");
+ok( $node_B->poll_query_until(
+		'postgres',
+		"SELECT xmin::text::bigint >= $next_xid FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"slot xmin advances again after the pinned transaction commits");
+
+# Clean up the second database.
+$node_B->safe_psql('dbb', "DROP SUBSCRIPTION $subname_BA2");
+$node_B->safe_psql('postgres', "DROP DATABASE dbb");
+
 ###############################################################################
 # Check that the replication slot pg_conflict_detection is dropped after
 # removing all the subscriptions.
-- 
2.43.0

