v18-0001-Maintain-the-oldest-non-removeable-tranasction-I.patch

From 4af7e9d503a96ce10da675ba9c05db4266a34ff6 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Thu, 26 Sep 2024 14:11:40 +0800
Subject: [PATCH v18 1/7] Maintain the oldest non removeable tranasction ID by
 apply worker

This set of patches aims to support the detection of update_deleted conflicts,
which occur when the apply worker cannot find the target tuple to be updated
(e.g., the tuple has been removed by a different origin).

To detect this conflict consistently and correctly, we must ensure that tuples
deleted by other origins are not prematurely removed by VACUUM before conflict
detection. If these tuples are removed too soon, a different conflict might
arise and be resolved incorrectly, causing data inconsistency between nodes.

To achieve this, we will retain the dead tuples on the subscriber for some
period.

The concept is that dead tuples are useful for detecting conflicts only during
the application of concurrent transactions from remote nodes. After applying
and flushing all remote transactions that occurred concurrently with the tuple
DELETE, any subsequent UPDATE from a remote node should have a later timestamp.
In such cases, it is acceptable to detect an update_missing scenario and
convert the UPDATE to an INSERT when applying it. But, for concurrent remote
transactions with earlier timestamps than the DELETE, detecting update_deleted
is necessary, as the UPDATEs in remote transactions should be ignored if their
timestamp is earlier than that of the dead tuples.

We assume that the appropriate resolution for update_deleted conflicts, to
achieve eventual consistency, is the last-update-win strategy. This means
that when detecting the update_deleted conflict, and the remote update has a
later timestamp, the resolution would be to convert the UPDATE to an INSERT.
Remote updates with earlier timestamps compared to the dead tuples will be
disregarded.

To implement this, an additional replication slot named pg_conflict_detection
will be created on the subscriber side and maintained by the launcher. This
slot will be used to retain dead tuples. Each apply worker will maintain its
own non-removable transaction ID, while the launcher collects these IDs to
determine whether to advance the xmin value of the replication slot.

The process of advancing the non-removable transaction ID in the apply worker
involves:

1) Call GetOldestActiveTransactionId() to take oldestRunningXid as the
candidate xid.
2) Send a message to the walsender requesting the publisher status, which
includes the latest WAL write position and information about transactions
that are in the commit phase.
3) Wait for the status from the walsender. After receiving the first status, do
not proceed if there are concurrent remote transactions that are still in the
commit phase. These transactions might have been assigned an earlier commit
timestamp but have not yet written the commit WAL record. Continue to request
the publisher status until all these transactions have completed.
4) Advance the non-removable transaction ID if the current flush location has
reached or surpassed the last received WAL position.

These steps are repeated at intervals defined by wal_receiver_status_interval
to minimize performance impact.

This mechanism ensures that dead tuples are not removed until all concurrent
transactions have been applied. It works for both bidirectional and
non-bidirectional replication scenarios.

Since the mechanism relies on a single replication slot, it not only
assists in retaining dead tuples but also preserves commit timestamps and
origin data. These information will be displayed in the additional logs
generated for logical replication conflicts.

Furthermore, the preserved commit timestamps and origin data are essential for
consistently detecting update_origin_differs conflicts.

This patch allows each apply worker to maintain the non-removable transaction ID
in the shared memory following the steps described above.

The actual replication slot management is implemented in the following patches.
---
 doc/src/sgml/protocol.sgml                 |  90 +++++
 src/backend/access/transam/xact.c          |   6 +-
 src/backend/replication/logical/launcher.c |   1 +
 src/backend/replication/logical/worker.c   | 417 ++++++++++++++++++++-
 src/backend/replication/walsender.c        |  54 +++
 src/backend/storage/ipc/procarray.c        |  59 +++
 src/include/replication/worker_internal.h  |  18 +
 src/include/storage/proc.h                 |   5 +
 src/include/storage/procarray.h            |   1 +
 src/tools/pgindent/typedefs.list           |   2 +
 10 files changed, 649 insertions(+), 4 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index fb5dec1172..49a4d3ab1b 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2450,6 +2450,69 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
          </variablelist>
         </listitem>
        </varlistentry>
+
+       <varlistentry id="protocol-replication-primary-status-update">
+        <term>Primary status update (B)</term>
+        <listitem>
+         <variablelist>
+          <varlistentry>
+           <term>Byte1('s')</term>
+           <listitem>
+            <para>
+             Identifies the message as a primary status update.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int64</term>
+           <listitem>
+            <para>
+             The latest WAL write position on the server.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int32</term>
+           <listitem>
+            <para>
+             The oldest transaction ID that is currently in the commit
+             phase on the server.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int32</term>
+           <listitem>
+            <para>
+             The next transaction ID to be assigned on the server.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int32</term>
+           <listitem>
+            <para>
+             The epoch of the next transaction ID to be assigned.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int64</term>
+           <listitem>
+            <para>
+             The server's system clock at the time of transmission, as
+             microseconds since midnight on 2000-01-01.
+            </para>
+           </listitem>
+          </varlistentry>
+         </variablelist>
+        </listitem>
+       </varlistentry>
       </variablelist>
 
       <para>
@@ -2594,6 +2657,33 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
          </variablelist>
         </listitem>
        </varlistentry>
+
+       <varlistentry id="protocol-replication-standby-wal-status-request">
+        <term>Request primary status update (F)</term>
+        <listitem>
+         <variablelist>
+          <varlistentry>
+           <term>Byte1('p')</term>
+           <listitem>
+            <para>
+             Identifies the message as a request for a primary status update.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int64</term>
+           <listitem>
+            <para>
+             The client's system clock at the time of transmission, as
+             microseconds since midnight on 2000-01-01.
+            </para>
+           </listitem>
+          </varlistentry>
+         </variablelist>
+        </listitem>
+       </varlistentry>
+
       </variablelist>
      </listitem>
     </varlistentry>
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 3ebd7c4041..611926c7d2 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -1431,9 +1431,9 @@ RecordTransactionCommit(void)
 		 * modifying it.  This makes checkpoint's determination of which xacts
 		 * are delaying the checkpoint a bit fuzzy, but it doesn't matter.
 		 */
-		Assert((MyProc->delayChkptFlags & DELAY_CHKPT_START) == 0);
+		Assert((MyProc->delayChkptFlags & DELAY_CHKPT_IN_COMMIT) == 0);
 		START_CRIT_SECTION();
-		MyProc->delayChkptFlags |= DELAY_CHKPT_START;
+		MyProc->delayChkptFlags |= DELAY_CHKPT_IN_COMMIT;
 
 		/*
 		 * Insert the commit XLOG record.
@@ -1536,7 +1536,7 @@ RecordTransactionCommit(void)
 	 */
 	if (markXidCommitted)
 	{
-		MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
+		MyProc->delayChkptFlags &= ~DELAY_CHKPT_IN_COMMIT;
 		END_CRIT_SECTION();
 	}
 
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 8b19642044..f8cf00c984 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -441,6 +441,7 @@ retry:
 	worker->stream_fileset = NULL;
 	worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
 	worker->parallel_apply = is_parallel_apply_worker;
+	worker->oldest_nonremovable_xid = InvalidFullTransactionId;
 	worker->last_lsn = InvalidXLogRecPtr;
 	TIMESTAMP_NOBEGIN(worker->last_send_time);
 	TIMESTAMP_NOBEGIN(worker->last_recv_time);
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 9e50c880f8..654078a726 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -173,12 +173,14 @@
 #include "replication/logicalrelation.h"
 #include "replication/logicalworker.h"
 #include "replication/origin.h"
+#include "replication/slot.h"
 #include "replication/walreceiver.h"
 #include "replication/worker_internal.h"
 #include "rewrite/rewriteHandler.h"
 #include "storage/buffile.h"
 #include "storage/ipc.h"
 #include "storage/lmgr.h"
+#include "storage/procarray.h"
 #include "tcop/tcopprot.h"
 #include "utils/acl.h"
 #include "utils/dynahash.h"
@@ -275,6 +277,44 @@ typedef enum
 	TRANS_PARALLEL_APPLY,
 } TransApplyAction;
 
+/*
+ * The phases involved in advancing the non-removable transaction ID.
+ *
+ * See maybe_advance_nonremovable_xid() for details of the transition
+ * between these phases.
+ */
+typedef enum
+{
+	RCI_GET_CANDIDATE_XID,
+	RCI_REQUEST_PUBLISHER_STATUS,
+	RCI_WAIT_FOR_PUBLISHER_STATUS,
+	RCI_WAIT_FOR_LOCAL_FLUSH
+} RetainConflictInfoPhase;
+
+/*
+ * Critical information for managing phase transitions within the
+ * RetainConflictInfoPhase.
+ */
+typedef struct RetainConflictInfoData
+{
+	RetainConflictInfoPhase phase;	/* current phase */
+	XLogRecPtr	remote_lsn;		/* WAL write position on the publisher */
+	TransactionId remote_oldestxid; /* oldest transaction ID that was in the
+									 * commit phase on the publisher */
+	TransactionId remote_nextxid;	/* next transaction ID to be assigned on
+									 * the publisher */
+	uint32		remote_epoch;	/* epoch of remote_nextxid */
+	FullTransactionId last_phase_at;	/* publisher transaction ID that must
+										 * be awaited to complete before
+										 * entering the final phase
+										 * (RCI_WAIT_FOR_LOCAL_FLUSH) */
+	FullTransactionId candidate_xid;	/* candidate for the non-removable
+										 * transaction ID */
+	TimestampTz reply_time;		/* when the publisher responds with status */
+	TimestampTz last_recv_time; /* when the last message was received */
+	TimestampTz candidate_xid_time; /* when the candidate_xid is decided */
+} RetainConflictInfoData;
+
 /* errcontext tracker */
 static ApplyErrorCallbackArg apply_error_callback_arg =
 {
@@ -339,6 +379,12 @@ static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr;
 /* BufFile handle of the current streaming file */
 static BufFile *stream_fd = NULL;
 
+/*
+ * The remote WAL position that has been applied and flushed locally. Refer to
+ * send_feedback() for details on its usage.
+ */
+static XLogRecPtr last_flushpos = InvalidXLogRecPtr;
+
 typedef struct SubXactInfo
 {
 	TransactionId xid;			/* XID of the subxact */
@@ -379,6 +425,15 @@ static void stream_close_file(void);
 
 static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
 
+static void maybe_advance_nonremovable_xid(RetainConflictInfoData *data,
+										   bool status_received);
+static void get_candidate_xid(RetainConflictInfoData *data);
+static void request_publisher_status(RetainConflictInfoData *data);
+static void wait_for_publisher_status(RetainConflictInfoData *data,
+									  bool status_received);
+static void wait_for_local_flush(RetainConflictInfoData *data);
+static inline bool can_advance_nonremovable_xid(RetainConflictInfoData *data);
+
 static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
 static void apply_handle_insert_internal(ApplyExecutionData *edata,
 										 ResultRelInfo *relinfo,
@@ -3573,6 +3628,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 	bool		ping_sent = false;
 	TimeLineID	tli;
 	ErrorContextCallback errcallback;
+	RetainConflictInfoData data = {0};
 
 	/*
 	 * Init the ApplyMessageContext which we clean up after each replication
@@ -3651,6 +3707,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 					last_recv_timestamp = GetCurrentTimestamp();
 					ping_sent = false;
 
+					data.last_recv_time = last_recv_timestamp;
+
 					/* Ensure we are reading the data into our memory context. */
 					MemoryContextSwitchTo(ApplyMessageContext);
 
@@ -3677,6 +3735,14 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 						UpdateWorkerStats(last_received, send_time, false);
 
 						apply_dispatch(&s);
+
+						/*
+						 * Attempt to advance the non-removable transaction ID
+						 * to avoid accumulating dead rows when the worker is
+						 * busy.
+						 */
+						if (can_advance_nonremovable_xid(&data))
+							maybe_advance_nonremovable_xid(&data, false);
 					}
 					else if (c == 'k')
 					{
@@ -3692,8 +3758,32 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 							last_received = end_lsn;
 
 						send_feedback(last_received, reply_requested, false);
+
+						maybe_advance_nonremovable_xid(&data, false);
+
 						UpdateWorkerStats(last_received, timestamp, true);
 					}
+					else if (c == 's')	/* Primary status update */
+					{
+						data.remote_lsn = pq_getmsgint64(&s);
+						data.remote_oldestxid = pq_getmsgint(&s, 4);
+						data.remote_nextxid = pq_getmsgint(&s, 4);
+						data.remote_epoch = pq_getmsgint(&s, 4);
+						data.reply_time = pq_getmsgint64(&s);
+
+						/*
+						 * This should never happen, see
+						 * ProcessStandbyPSRequestMessage. But if it happens
+						 * due to a bug, we don't want to proceed as it can
+						 * incorrectly advance oldest_nonremovable_xid.
+						 */
+						if (XLogRecPtrIsInvalid(data.remote_lsn))
+							elog(ERROR, "cannot get the latest WAL position from the publisher");
+
+						maybe_advance_nonremovable_xid(&data, true);
+
+						UpdateWorkerStats(last_received, data.reply_time, false);
+					}
 					/* other message types are purposefully ignored */
 
 					MemoryContextReset(ApplyMessageContext);
@@ -3706,6 +3796,11 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		/* confirm all writes so far */
 		send_feedback(last_received, false, false);
 
+		/* Reset the timestamp if no message was received */
+		data.last_recv_time = 0;
+
+		maybe_advance_nonremovable_xid(&data, false);
+
 		if (!in_remote_transaction && !in_streamed_transaction)
 		{
 			/*
@@ -3803,6 +3898,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 
 			send_feedback(last_received, requestReply, requestReply);
 
+			maybe_advance_nonremovable_xid(&data, false);
+
 			/*
 			 * Force reporting to ensure long idle periods don't lead to
 			 * arbitrarily delayed stats. Stats can only be reported outside
@@ -3838,7 +3935,6 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
 
 	static XLogRecPtr last_recvpos = InvalidXLogRecPtr;
 	static XLogRecPtr last_writepos = InvalidXLogRecPtr;
-	static XLogRecPtr last_flushpos = InvalidXLogRecPtr;
 
 	XLogRecPtr	writepos;
 	XLogRecPtr	flushpos;
@@ -3916,6 +4012,325 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
 		last_flushpos = flushpos;
 }
 
+/*
+ * Attempt to advance the non-removable transaction ID.
+ *
+ * The oldest_nonremovable_xid is maintained in shared memory to prevent dead
+ * rows from being removed prematurely when the apply worker still needs them
+ * to detect update_deleted conflicts.
+ *
+ * The non-removable transaction ID is advanced to the oldest running
+ * transaction ID once all concurrent transactions on the publisher have been
+ * applied and flushed locally. The process involves:
+ *
+ * - RCI_GET_CANDIDATE_XID:
+ *   Call GetOldestActiveTransactionId() to take oldestRunningXid as the
+ *   candidate xid.
+ *
+ * - RCI_REQUEST_PUBLISHER_STATUS:
+ *   Send a message to the walsender requesting the publisher status, which
+ *   includes the latest WAL write position and information about transactions
+ *   that are in the commit phase.
+ *
+ * - RCI_WAIT_FOR_PUBLISHER_STATUS:
+ *   Wait for the status from the walsender. After receiving the first status,
+ *   do not proceed if there are concurrent remote transactions that are still
+ *   in the commit phase. These transactions might have been assigned an
+ *   earlier commit timestamp but have not yet written the commit WAL record.
+ *   Continue to request the publisher status (RCI_REQUEST_PUBLISHER_STATUS)
+ *   until all these transactions have completed.
+ *
+ * - RCI_WAIT_FOR_LOCAL_FLUSH:
+ *   Advance the non-removable transaction ID if the current flush location has
+ *   reached or surpassed the last received WAL position.
+ *
+ * The overall state progression is: GET_CANDIDATE_XID ->
+ * REQUEST_PUBLISHER_STATUS -> WAIT_FOR_PUBLISHER_STATUS -> (loop to
+ * REQUEST_PUBLISHER_STATUS if concurrent remote transactions persist) ->
+ * WAIT_FOR_LOCAL_FLUSH -> loop back to GET_CANDIDATE_XID.
+ *
+ * Retaining the dead tuples for this period is sufficient for ensuring
+ * eventual consistency using last-update-wins strategy, as dead tuples are
+ * useful for detecting conflicts only during the application of concurrent
+ * transactions from remote nodes. After applying and flushing all remote
+ * transactions that occurred concurrently with the tuple DELETE, any
+ * subsequent UPDATE from a remote node should have a later timestamp. In such
+ * cases, it is acceptable to detect an update_missing scenario and convert the
+ * UPDATE to an INSERT when applying it. But, for concurrent remote
+ * transactions with earlier timestamps than the DELETE, detecting
+ * update_deleted is necessary, as the UPDATEs in remote transactions should be
+ * ignored if their timestamp is earlier than that of the dead tuples.
+ *
+ * Note that advancing the non-removable transaction ID is not supported if the
+ * publisher is also a physical standby. This is because the logical walsender
+ * on the standby can only get the WAL replay position but there may be more
+ * WALs that are being replicated from the primary and those WALs could have
+ * earlier commit timestamp.
+ *
+ * XXX It might seem feasible to track the latest commit timestamp on the
+ * publisher and send the WAL position once the timestamp exceeds that on the
+ * subscriber. However, commit timestamps can regress since a commit with a
+ * later LSN is not guaranteed to have a later timestamp than those with
+ * earlier LSNs.
+ */
+static void
+maybe_advance_nonremovable_xid(RetainConflictInfoData *data,
+							   bool status_received)
+{
+	/*
+	 * It is sufficient to manage non-removable transaction ID for a
+	 * subscription by the main apply worker to detect update_deleted conflict
+	 * even for table sync or parallel apply workers.
+	 */
+	if (!am_leader_apply_worker())
+		return;
+
+	switch (data->phase)
+	{
+		case RCI_GET_CANDIDATE_XID:
+			get_candidate_xid(data);
+			break;
+		case RCI_REQUEST_PUBLISHER_STATUS:
+			request_publisher_status(data);
+			break;
+		case RCI_WAIT_FOR_PUBLISHER_STATUS:
+			wait_for_publisher_status(data, status_received);
+			break;
+		case RCI_WAIT_FOR_LOCAL_FLUSH:
+			wait_for_local_flush(data);
+			break;
+	}
+}
+
+/*
+ * Workhorse for the RCI_GET_CANDIDATE_XID phase.
+ */
+static void
+get_candidate_xid(RetainConflictInfoData *data)
+{
+	TransactionId oldest_running_xid;
+	FullTransactionId next_full_xid;
+	FullTransactionId full_xid;
+	uint32		epoch;
+	TimestampTz now;
+
+	/*
+	 * Use last_recv_time when applying changes in the loop; otherwise, get
+	 * the latest timestamp.
+	 */
+	now = data->last_recv_time ? data->last_recv_time : GetCurrentTimestamp();
+
+	/*
+	 * Compute the candidate_xid and request the publisher status at most once
+	 * per wal_receiver_status_interval. This is to avoid using CPU and
+	 * network resources without making much progress.
+	 *
+	 * XXX The use of wal_receiver_status_interval is a bit arbitrary so we
+	 * can consider the other interval or a separate GUC if the need arises.
+	 */
+	if (!TimestampDifferenceExceeds(data->candidate_xid_time, now,
+									wal_receiver_status_interval * 1000))
+		return;
+
+	data->candidate_xid_time = now;
+
+	oldest_running_xid = GetOldestActiveTransactionId();
+	next_full_xid = ReadNextFullTransactionId();
+	epoch = EpochFromFullTransactionId(next_full_xid);
+
+	/* Compute the epoch of the oldest_running_xid */
+	if (oldest_running_xid > XidFromFullTransactionId(next_full_xid))
+		epoch--;
+
+	full_xid = FullTransactionIdFromEpochAndXid(epoch, oldest_running_xid);
+
+	/* Return if the oldest_nonremovable_xid cannot be advanced */
+	if (FullTransactionIdFollowsOrEquals(MyLogicalRepWorker->oldest_nonremovable_xid,
+										 full_xid))
+		return;
+
+	data->candidate_xid = full_xid;
+	data->phase = RCI_REQUEST_PUBLISHER_STATUS;
+
+	/* process the next phase */
+	maybe_advance_nonremovable_xid(data, false);
+}
+
+/*
+ * Workhorse for the RCI_REQUEST_PUBLISHER_STATUS phase.
+ */
+static void
+request_publisher_status(RetainConflictInfoData *data)
+{
+	static StringInfo request_message = NULL;
+
+	if (!request_message)
+	{
+		MemoryContext oldctx = MemoryContextSwitchTo(ApplyContext);
+
+		request_message = makeStringInfo();
+		MemoryContextSwitchTo(oldctx);
+	}
+	else
+		resetStringInfo(request_message);
+
+	/*
+	 * Send the current time to update the remote walsender's latest reply
+	 * message received time.
+	 */
+	pq_sendbyte(request_message, 'p');
+	pq_sendint64(request_message, GetCurrentTimestamp());
+
+	elog(DEBUG2, "sending publisher status request message");
+
+	/* Send a request for the publisher status */
+	walrcv_send(LogRepWorkerWalRcvConn,
+				request_message->data, request_message->len);
+
+	data->phase = RCI_WAIT_FOR_PUBLISHER_STATUS;
+
+	/*
+	 * Skip calling maybe_advance_nonremovable_xid() since further transition
+	 * is possible only once we receive the publisher status message.
+	 */
+}
+
+/*
+ * Workhorse for the RCI_WAIT_FOR_PUBLISHER_STATUS phase.
+ */
+static void
+wait_for_publisher_status(RetainConflictInfoData *data, bool status_received)
+{
+	FullTransactionId remote_full_xid;
+	uint32		remote_epoch = data->remote_epoch;
+
+	/*
+	 * Return if we have requested but not yet received the publisher status.
+	 */
+	if (!status_received)
+		return;
+
+	if (!FullTransactionIdIsValid(data->last_phase_at))
+		data->last_phase_at = FullTransactionIdFromEpochAndXid(data->remote_epoch,
+															   data->remote_nextxid);
+
+	/* Compute the epoch of the remote oldest running transaction ID */
+	if (data->remote_oldestxid > data->remote_nextxid)
+		remote_epoch--;
+
+	remote_full_xid = FullTransactionIdFromEpochAndXid(remote_epoch,
+													   data->remote_oldestxid);
+
+	/*
+	 * Check if all remote concurrent transactions that were active at the
+	 * first status request have now completed. If completed, proceed to the
+	 * next phase; otherwise, continue checking the publisher status until
+	 * these transactions finish.
+	 */
+	if (FullTransactionIdPrecedesOrEquals(data->last_phase_at,
+										  remote_full_xid))
+		data->phase = RCI_WAIT_FOR_LOCAL_FLUSH;
+	else
+		data->phase = RCI_REQUEST_PUBLISHER_STATUS;
+
+	/* process the next phase */
+	maybe_advance_nonremovable_xid(data, false);
+}
+
+/*
+ * Workhorse for the RCI_WAIT_FOR_LOCAL_FLUSH phase.
+ */
+static void
+wait_for_local_flush(RetainConflictInfoData *data)
+{
+	Assert(!XLogRecPtrIsInvalid(data->remote_lsn) &&
+		   FullTransactionIdIsValid(data->candidate_xid));
+
+	/*
+	 * We expect the publisher and subscriber clocks to be in sync using time
+	 * sync service like NTP. Otherwise, we will advance this worker's
+	 * oldest_nonremovable_xid prematurely, leading to the removal of rows
+	 * required to detect update_delete conflict.
+	 *
+	 * XXX Consider waiting for the publisher's clock to catch up with the
+	 * subscriber's before proceeding to the next phase.
+	 */
+	if (TimestampDifferenceExceeds(data->reply_time,
+								   data->candidate_xid_time, 0))
+		ereport(ERROR,
+				errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("oldest_nonremovable_xid transaction ID may be advanced prematurely"),
+				errdetail("The clock on the publisher is behind that of the subscriber."));
+
+	/*
+	 * Do not attempt to advance the non-removable transaction ID when table
+	 * sync is in progress. During this time, changes from a single
+	 * transaction may be applied by multiple table sync workers corresponding
+	 * to the target tables. In this case, confirming the apply and flush
+	 * progress across all table sync workers is complex and not worth the
+	 * effort.
+	 *
+	 * It is safe to add new tables with initial states to the subscription
+	 * after this check because WAL positions of changes from these new
+	 * tables, which will be applied, should be greater than remote_lsn and
+	 * are included in transactions with later commit timestamps. So, there is
+	 * no need to wait for these changes to be applied in this round of
+	 * advancement.
+	 */
+	if (!AllTablesyncsReady())
+		return;
+
+	/*
+	 * Return to wait for the changes to be applied.
+	 *
+	 * XXX The remote flush location (last_flushpos) is updated only when
+	 * feedback is sent to the server. So, the advancement of
+	 * oldest_nonremovable_xid may be delayed. We can always update
+	 * last_flushpos here if we notice such a delay.
+	 */
+	if (last_flushpos < data->remote_lsn)
+		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
+	 * flushed locally. So, now we can advance the non-removable transaction
+	 * ID.
+	 */
+	SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+	MyLogicalRepWorker->oldest_nonremovable_xid = data->candidate_xid;
+	SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+	elog(DEBUG2, "confirmed remote flush up to %X/%X: new oldest_nonremovable_xid %u",
+		 LSN_FORMAT_ARGS(data->remote_lsn),
+		 XidFromFullTransactionId(data->candidate_xid));
+
+	/*
+	 * Reset all data fields except those used to determine the timing for the
+	 * next round of transaction ID advancement.
+	 */
+	data->phase = RCI_GET_CANDIDATE_XID;
+	data->remote_lsn = InvalidXLogRecPtr;
+	data->remote_oldestxid = InvalidTransactionId;
+	data->remote_nextxid = InvalidTransactionId;
+	data->remote_epoch = 0;
+	data->last_phase_at = InvalidFullTransactionId;
+	data->candidate_xid = InvalidFullTransactionId;
+	data->reply_time = 0;
+
+	/* process the next phase */
+	maybe_advance_nonremovable_xid(data, false);
+}
+
+/*
+ * Determine if we can attempt to advance transaction ID.
+ */
+static inline bool
+can_advance_nonremovable_xid(RetainConflictInfoData *data)
+{
+	return data->phase == RCI_GET_CANDIDATE_XID;
+}
+
 /*
  * Exit routine for apply workers due to subscription parameter changes.
  */
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 371eef3ddd..68ca0c5712 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -83,6 +83,7 @@
 #include "storage/ipc.h"
 #include "storage/pmsignal.h"
 #include "storage/proc.h"
+#include "storage/procarray.h"
 #include "tcop/dest.h"
 #include "tcop/tcopprot.h"
 #include "utils/acl.h"
@@ -253,6 +254,7 @@ static void StartLogicalReplication(StartReplicationCmd *cmd);
 static void ProcessStandbyMessage(void);
 static void ProcessStandbyReplyMessage(void);
 static void ProcessStandbyHSFeedbackMessage(void);
+static void ProcessStandbyPSRequestMessage(void);
 static void ProcessRepliesIfAny(void);
 static void ProcessPendingWrites(void);
 static void WalSndKeepalive(bool requestReply, XLogRecPtr writePtr);
@@ -2314,6 +2316,10 @@ ProcessStandbyMessage(void)
 			ProcessStandbyHSFeedbackMessage();
 			break;
 
+		case 'p':
+			ProcessStandbyPSRequestMessage();
+			break;
+
 		default:
 			ereport(COMMERROR,
 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
@@ -2660,6 +2666,54 @@ ProcessStandbyHSFeedbackMessage(void)
 	}
 }
 
+/*
+ * Process the request for a primary status update message.
+ */
+static void
+ProcessStandbyPSRequestMessage(void)
+{
+	XLogRecPtr	lsn = InvalidXLogRecPtr;
+	TransactionId oldestXidInCommit;
+	FullTransactionId nextFullXid;
+	WalSnd	   *walsnd = MyWalSnd;
+	TimestampTz replyTime;
+
+	/*
+	 * This shouldn't happen because we don't support getting primary status
+	 * message from standby.
+	 */
+	if (RecoveryInProgress())
+		elog(ERROR, "the primary status is unavailable during recovery");
+
+	replyTime = pq_getmsgint64(&reply_message);
+
+	/*
+	 * Update shared state for this WalSender process based on reply data from
+	 * standby.
+	 */
+	SpinLockAcquire(&walsnd->mutex);
+	walsnd->replyTime = replyTime;
+	SpinLockRelease(&walsnd->mutex);
+
+	oldestXidInCommit = GetOldestTransactionIdInCommit();
+	nextFullXid = ReadNextFullTransactionId();
+	lsn = GetXLogWriteRecPtr();
+
+	elog(DEBUG2, "sending primary status");
+
+	/* construct the message... */
+	resetStringInfo(&output_message);
+	pq_sendbyte(&output_message, 's');
+	pq_sendint64(&output_message, lsn);
+	pq_sendint32(&output_message, oldestXidInCommit);
+	pq_sendint32(&output_message, XidFromFullTransactionId(nextFullXid));
+	pq_sendint32(&output_message, EpochFromFullTransactionId(nextFullXid));
+	pq_sendint64(&output_message, GetCurrentTimestamp());
+
+	/* ... and send it wrapped in CopyData */
+	pq_putmessage_noblock('d', output_message.data, output_message.len);
+}
+
 /*
  * Compute how long send/receive loops should sleep.
  *
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index c769b1aa3e..ae6da9fd91 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -2925,6 +2925,65 @@ GetOldestActiveTransactionId(void)
 	return oldestRunningXid;
 }
 
+
+/*
+ * GetOldestTransactionIdInCommit()
+ *
+ * Similar to GetOldestActiveTransactionId but returns the oldest transaction ID
+ * that is currently in the commit phase.
+ */
+TransactionId
+GetOldestTransactionIdInCommit(void)
+{
+	ProcArrayStruct *arrayP = procArray;
+	TransactionId *other_xids = ProcGlobal->xids;
+	TransactionId oldestXidInCommit;
+	int			index;
+
+	Assert(!RecoveryInProgress());
+
+	/*
+	 * Read nextXid, as the upper bound of what's still active.
+	 *
+	 * Reading a TransactionId is atomic, but we must grab the lock to make
+	 * sure that all XIDs < nextXid are already present in the proc array (or
+	 * have already completed), when we spin over it.
+	 */
+	LWLockAcquire(XidGenLock, LW_SHARED);
+	oldestXidInCommit = XidFromFullTransactionId(TransamVariables->nextXid);
+	LWLockRelease(XidGenLock);
+
+	/*
+	 * Spin over procArray collecting all xids and subxids.
+	 */
+	LWLockAcquire(ProcArrayLock, LW_SHARED);
+	for (index = 0; index < arrayP->numProcs; index++)
+	{
+		TransactionId xid;
+		int			pgprocno = arrayP->pgprocnos[index];
+		PGPROC	   *proc = &allProcs[pgprocno];
+
+		/* Fetch xid just once - see GetNewTransactionId */
+		xid = UINT32_ACCESS_ONCE(other_xids[index]);
+
+		if (!TransactionIdIsNormal(xid))
+			continue;
+
+		if ((proc->delayChkptFlags & DELAY_CHKPT_IN_COMMIT) != 0 &&
+			TransactionIdPrecedes(xid, oldestXidInCommit))
+			oldestXidInCommit = xid;
+
+		/*
+		 * Top-level XID of a transaction is always less than any of its
+		 * subxids, so we don't need to check if any of the subxids are
+		 * smaller than oldestXidInCommit
+		 */
+	}
+	LWLockRelease(ProcArrayLock);
+
+	return oldestXidInCommit;
+}
+
 /*
  * GetOldestSafeDecodingTransactionId -- lowest xid not affected by vacuum
  *
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 9646261d7e..1eab8a5e46 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -86,6 +86,24 @@ typedef struct LogicalRepWorker
 	/* Indicates whether apply can be performed in parallel. */
 	bool		parallel_apply;
 
+	/*
+	 * The changes made by this and later transactions are still non-removable
+	 * to allow for the detection of update_deleted conflicts when applying
+	 * changes in this logical replication worker.
+	 *
+	 * Note that this info cannot directly protect dead tuples from being
+	 * prematurely frozen or removed. The logical replication launcher
+	 * asynchronously collects this info to determine whether to advance the
+	 * xmin value of the replication slot.
+	 *
+	 * Therefore, FullTransactionId that includes both the transaction ID and
+	 * its epoch is used here instead of a single Transaction ID. This is
+	 * critical because without considering the epoch, the transaction ID
+	 * alone may appear as if it is in the future due to transaction ID
+	 * wraparound.
+	 */
+	FullTransactionId oldest_nonremovable_xid;
+
 	/* Stats. */
 	XLogRecPtr	last_lsn;
 	TimestampTz last_send_time;
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index 5a3dd5d2d4..7eca49e883 100644
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -121,9 +121,14 @@ extern PGDLLIMPORT int FastPathLockGroupsPerBackend;
  * the checkpoint are actually destroyed on disk. Replay can cope with a file
  * or block that doesn't exist, but not with a block that has the wrong
  * contents.
+ *
+ * Setting DELAY_CHKPT_IN_COMMIT is similar to setting DELAY_CHKPT_START, but
+ * it explicitly indicates that the reason for delaying the checkpoint is due
+ * to a transaction being within a critical commit section.
  */
 #define DELAY_CHKPT_START		(1<<0)
 #define DELAY_CHKPT_COMPLETE	(1<<1)
+#define DELAY_CHKPT_IN_COMMIT	(DELAY_CHKPT_START | 1<<2)
 
 typedef enum
 {
diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
index 56af0b40b3..c388eec6a5 100644
--- a/src/include/storage/procarray.h
+++ b/src/include/storage/procarray.h
@@ -57,6 +57,7 @@ extern bool TransactionIdIsActive(TransactionId xid);
 extern TransactionId GetOldestNonRemovableTransactionId(Relation rel);
 extern TransactionId GetOldestTransactionIdConsideredRunning(void);
 extern TransactionId GetOldestActiveTransactionId(void);
+extern TransactionId GetOldestTransactionIdInCommit(void);
 extern TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly);
 extern void GetReplicationHorizons(TransactionId *xmin, TransactionId *catalog_xmin);
 
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index e1c4f913f8..b5ea6bb5b8 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2474,6 +2474,8 @@ RestrictInfo
 Result
 ResultRelInfo
 ResultState
+RetainConflictInfoData
+RetainConflictInfoPhase
 ReturnSetInfo
 ReturnStmt
 RevmapContents
-- 
2.30.0.windows.2