From 19c52d18345688f9ae179c7ccc21005dcfd23fca 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 v3 1/5] 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.

We assume that the appropriate resolution for update_deleted conflicts, to
achieve eventual consistency, is the latest_timestamp_wins 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.

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) Calling GetRunningTransactionData() to take oldestRunningXid as the
candidate xid and send a new message to request the remote WAL position from
the walsender.
2) It then waits (non-blocking) to receive the WAL position from the walsender.
3) After receiving the WAL position, the non-removable transaction ID is
advanced if the current flush location has reached or surpassed the received
WAL position.

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

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

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                |  52 ++++++
 src/backend/replication/logical/worker.c  | 185 +++++++++++++++++++++-
 src/backend/replication/walsender.c       |  23 +++
 src/include/replication/worker_internal.h |  18 +++
 src/tools/pgindent/typedefs.list          |   1 +
 5 files changed, 278 insertions(+), 1 deletion(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 2d2481bb8b..fd69b7b9cf 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2440,6 +2440,41 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
          </variablelist>
         </listitem>
        </varlistentry>
+
+       <varlistentry id="protocol-replication-primary-wal-status-update">
+        <term>Primary WAL status update (B)</term>
+        <listitem>
+         <variablelist>
+          <varlistentry>
+           <term>Byte1('s')</term>
+           <listitem>
+            <para>
+             Identifies the message as a primary WAL status update.
+            </para>
+           </listitem>
+          </varlistentry>
+
+          <varlistentry>
+           <term>Int64</term>
+           <listitem>
+            <para>
+             The latest WAL write position on the server.
+            </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>
@@ -2584,6 +2619,23 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
          </variablelist>
         </listitem>
        </varlistentry>
+
+       <varlistentry id="protocol-replication-standby-wal-status-request">
+        <term>Standby WAL status request (F)</term>
+        <listitem>
+         <variablelist>
+          <varlistentry>
+           <term>Byte1('x')</term>
+           <listitem>
+            <para>
+             Identifies the message as a request for the WAL status on the primary.
+            </para>
+           </listitem>
+          </varlistentry>
+         </variablelist>
+        </listitem>
+       </varlistentry>
+
       </variablelist>
      </listitem>
     </varlistentry>
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 925dff9cc4..a55ae55e1b 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,19 @@ typedef enum
 	TRANS_PARALLEL_APPLY,
 } TransApplyAction;
 
+/*
+ * The phases involved in advancing the non-removable transaction ID.
+ *
+ * Refer to maybe_advance_nonremovable_xid() for details on how the function
+ * transitions between these phases.
+ */
+typedef enum
+{
+	DTR_REQUEST_WALSENDER_WAL_POS,
+	DTR_WAIT_FOR_WALSENDER_WAL_POS,
+	DTR_WAIT_FOR_LOCAL_FLUSH
+} DeadTupleRetainPhase;
+
 /* errcontext tracker */
 static ApplyErrorCallbackArg apply_error_callback_arg =
 {
@@ -339,6 +354,8 @@ static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr;
 /* BufFile handle of the current streaming file */
 static BufFile *stream_fd = NULL;
 
+static XLogRecPtr last_flushpos = InvalidXLogRecPtr;
+
 typedef struct SubXactInfo
 {
 	TransactionId xid;			/* XID of the subxact */
@@ -378,6 +395,8 @@ static void stream_open_and_write_change(TransactionId xid, char action, StringI
 static void stream_close_file(void);
 
 static void send_feedback(XLogRecPtr recvpos, bool force, bool requestReply);
+static void maybe_advance_nonremovable_xid(XLogRecPtr *remote_wal_pos,
+										   DeadTupleRetainPhase *phase);
 
 static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
 static void apply_handle_insert_internal(ApplyExecutionData *edata,
@@ -3573,6 +3592,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 	bool		ping_sent = false;
 	TimeLineID	tli;
 	ErrorContextCallback errcallback;
+	XLogRecPtr	remote_lsn = InvalidXLogRecPtr;
+	DeadTupleRetainPhase phase = DTR_REQUEST_WALSENDER_WAL_POS;
 
 	/*
 	 * Init the ApplyMessageContext which we clean up after each replication
@@ -3694,6 +3715,16 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 						send_feedback(last_received, reply_requested, false);
 						UpdateWorkerStats(last_received, timestamp, true);
 					}
+					else if (c == 's')
+					{
+						TimestampTz timestamp;
+
+						remote_lsn = pq_getmsgint64(&s);
+						timestamp = pq_getmsgint64(&s);
+
+						maybe_advance_nonremovable_xid(&remote_lsn, &phase);
+						UpdateWorkerStats(last_received, timestamp, false);
+					}
 					/* other message types are purposefully ignored */
 
 					MemoryContextReset(ApplyMessageContext);
@@ -3706,6 +3737,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		/* confirm all writes so far */
 		send_feedback(last_received, false, false);
 
+		maybe_advance_nonremovable_xid(&remote_lsn, &phase);
+
 		if (!in_remote_transaction && !in_streamed_transaction)
 		{
 			/*
@@ -3838,7 +3871,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 +3948,157 @@ 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-delete 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:
+ *
+ * DTR_REQUEST_WALSENDER_WAL_POS - Call GetRunningTransactionData() to get the
+ * candidate xmin and send a message to request the remote WAL position from
+ * the walsender.
+ *
+ * DTR_WAIT_FOR_WALSENDER_WAL_POS - Wait for receiving the WAL position from
+ * the walsender.
+ *
+ * DTR_WAIT_FOR_LOCAL_FLUSH - Advance the non-removable transaction ID if the
+ * current flush location has reached or surpassed the received WAL position.
+ *
+ * Retaining the dead tuples for this period is sufficient because any
+ * subsequent transaction from the publisher will have a later timestamp.
+ * Therefore, it is acceptable if dead tuples are removed by vacuum and an
+ * update_missing conflict is detected, as the correct resolution for the
+ * last-update-wins strategy in this case is to convert the UPDATE to an INSERT
+ * and apply it anyway.
+ *
+ * The 'remote_wal_pos' will be reset after sending a new request to walsender.
+ */
+static void
+maybe_advance_nonremovable_xid(XLogRecPtr *remote_wal_pos,
+							   DeadTupleRetainPhase *phase)
+{
+	static TimestampTz xid_advance_attemp_time = 0;
+	static FullTransactionId candidate_xid;
+
+	Assert(remote_wal_pos);
+
+	/*
+	 * The non-removable transaction ID for a subscription is centrally
+	 * managed by the main apply worker.
+	 */
+	if (!am_leader_apply_worker())
+		return;
+
+	if (*phase == DTR_WAIT_FOR_WALSENDER_WAL_POS)
+	{
+		Assert(xid_advance_attemp_time);
+
+		/*
+		 * Return if we have requested but not yet received the remote wal
+		 * position.
+		 */
+		if (XLogRecPtrIsInvalid(*remote_wal_pos))
+			return;
+
+		*phase = DTR_WAIT_FOR_LOCAL_FLUSH;
+	}
+
+	if (*phase == DTR_WAIT_FOR_LOCAL_FLUSH)
+	{
+		Assert(!XLogRecPtrIsInvalid(*remote_wal_pos) &&
+			   FullTransactionIdIsValid(candidate_xid));
+
+		/*
+		 * 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.
+		 */
+		if (!AllTablesyncsReady())
+			return;
+
+		/* Return to wait for the changes to be applied */
+		if (last_flushpos < *remote_wal_pos)
+			return;
+
+		/*
+		 * Advance the non-removable transaction id if the remote wal position
+		 * has been received, and all transactions up to that position on the
+		 * publisher have been applied and flushed locally.
+		 */
+		SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+		MyLogicalRepWorker->oldest_nonremovable_xid = candidate_xid;
+		SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+		*phase = DTR_REQUEST_WALSENDER_WAL_POS;
+	}
+
+	if (*phase == DTR_REQUEST_WALSENDER_WAL_POS)
+	{
+		TimestampTz now;
+		RunningTransactions running_transactions;
+		TransactionId oldest_running_xid;
+		FullTransactionId full_xid;
+		uint32		epoch;
+
+		/*
+		 * Exit early if the user has disabled sending messages to the
+		 * publisher.
+		 */
+		if (wal_receiver_status_interval <= 0)
+			return;
+
+		now = GetCurrentTimestamp();
+
+		/*
+		 * Compute the candidate_xid and send a message at most once per
+		 * wal_receiver_status_interval.
+		 */
+		if (!TimestampDifferenceExceeds(xid_advance_attemp_time, now,
+										wal_receiver_status_interval * 1000))
+			return;
+
+		xid_advance_attemp_time = now;
+
+		running_transactions = GetRunningTransactionData();
+		oldest_running_xid = running_transactions->oldestRunningXid;
+		epoch = EpochFromFullTransactionId(TransamVariables->nextXid);
+
+		/* Release the locks acquired in GetRunningTransactionData() */
+		LWLockRelease(ProcArrayLock);
+		LWLockRelease(XidGenLock);
+
+		/* Compute the epoch of the oldestRunningXid */
+		if (oldest_running_xid > running_transactions->nextXid)
+			epoch--;
+
+		full_xid = FullTransactionIdFromEpochAndXid(epoch,
+													oldest_running_xid);
+
+		/* Return if the new transaction ID is unchanged */
+		if (FullTransactionIdFollowsOrEquals(MyLogicalRepWorker->oldest_nonremovable_xid,
+											 full_xid))
+			return;
+
+		candidate_xid = full_xid;
+
+		elog(DEBUG2, "sending wal position request message");
+
+		/* Send a wal position request message to the server */
+		walrcv_send(LogRepWorkerWalRcvConn, "x", sizeof(uint8));
+
+		*remote_wal_pos = InvalidXLogRecPtr;
+		*phase = DTR_WAIT_FOR_WALSENDER_WAL_POS;
+	}
+}
+
 /*
  * 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 c5f1009f37..5371711488 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -253,6 +253,7 @@ static void StartLogicalReplication(StartReplicationCmd *cmd);
 static void ProcessStandbyMessage(void);
 static void ProcessStandbyReplyMessage(void);
 static void ProcessStandbyHSFeedbackMessage(void);
+static void ProcessStandbyWalPosRequestMessage(void);
 static void ProcessRepliesIfAny(void);
 static void ProcessPendingWrites(void);
 static void WalSndKeepalive(bool requestReply, XLogRecPtr writePtr);
@@ -2342,6 +2343,10 @@ ProcessStandbyMessage(void)
 			ProcessStandbyHSFeedbackMessage();
 			break;
 
+		case 'x':
+			ProcessStandbyWalPosRequestMessage();
+			break;
+
 		default:
 			ereport(COMMERROR,
 					(errcode(ERRCODE_PROTOCOL_VIOLATION),
@@ -2688,6 +2693,24 @@ ProcessStandbyHSFeedbackMessage(void)
 	}
 }
 
+/*
+ * Process the standby message requesting the latest WAL write position.
+ */
+static void
+ProcessStandbyWalPosRequestMessage(void)
+{
+	elog(DEBUG2, "sending wal write position");
+
+	/* construct the message... */
+	resetStringInfo(&output_message);
+	pq_sendbyte(&output_message, 's');
+	pq_sendint64(&output_message, GetXLogWriteRecPtr());
+	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/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 9646261d7e..370c71c93e 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_delete 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/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 5fabb127d7..66d7f16a73 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -591,6 +591,7 @@ DbInfoArr
 DbLocaleInfo
 DeClonePtrType
 DeadLockState
+DeadTupleRetainPhase
 DeallocateStmt
 DeclareCursorStmt
 DecodedBkpBlock
-- 
2.30.0.windows.2

