From 17b25efa8af1276922a7e4c9a8423c783019db80 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Tue, 24 Dec 2024 14:48:26 +0800
Subject: [PATCH v18 7/7] Wait for publisher's clock to catch up before
 proceeding

When the publisher's clock is behind the subscriber's, instead of repeatedly
reporting an ERROR, allow the apply worker to wait for the clock skew to
resolve and then send a new request to the publisher for the latest status.
This approach allows for the detection of update_deleted conflicts even if
clock skew exists.
---
 src/backend/replication/logical/worker.c | 71 ++++++++++++++++--------
 src/backend/replication/walsender.c      | 11 +++-
 2 files changed, 59 insertions(+), 23 deletions(-)

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 895aa4b9bc..f0b7653c35 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -313,6 +313,7 @@ typedef struct RetainConflictInfoData
 	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 */
+	TimestampTz next_request_time; /* when to send next request */
 	int			xid_advancement_interval;	/* how much time (ms) to wait
 											 * before attempting to advance
 											 * the non-removable transaction
@@ -4071,11 +4072,19 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
  *
  * - 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.
+ *   do not proceed if:
+ *   a) The publisher's clock lags behind that of the subscriber. 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.
+ *   b) 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.
+ *   until the publisher's clock catches up and all these transactions have
+ *   completed.
  *
  * - RCI_WAIT_FOR_LOCAL_FLUSH:
  *   Advance the non-removable transaction ID if the current flush location has
@@ -4083,8 +4092,9 @@ send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
  *
  * 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.
+ * REQUEST_PUBLISHER_STATUS if publisher's clock lags behind or 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
@@ -4179,6 +4189,15 @@ get_candidate_xid(RetainConflictInfoData *data)
 	next_full_xid = ReadNextFullTransactionId();
 	epoch = EpochFromFullTransactionId(next_full_xid);
 
+	/*
+	 * Get a new timestamp after obtaining the oldest running transaction ID.
+	 * This ensures that all transactions preceding this transaction ID were
+	 * assigned a commit timestamp earlier than the current timestamp. In
+	 * wait_for_local_flush(), this timestamp is used to prevent the premature
+	 * advancement of the oldest_nonremovable_xid due to clock skew.
+	 */
+	data->candidate_xid_time = GetCurrentTimestamp();
+
 	/* Compute the epoch of the oldest_running_xid */
 	if (oldest_running_xid > XidFromFullTransactionId(next_full_xid))
 		epoch--;
@@ -4244,6 +4263,7 @@ static void
 request_publisher_status(RetainConflictInfoData *data)
 {
 	static StringInfo request_message = NULL;
+	TimestampTz now;
 
 	if (!request_message)
 	{
@@ -4255,6 +4275,12 @@ request_publisher_status(RetainConflictInfoData *data)
 	else
 		resetStringInfo(request_message);
 
+	now = GetCurrentTimestamp();
+
+	/* Return if it is not yet time to send the next request */
+	if (now < data->next_request_time)
+		return;
+
 	/*
 	 * Send the current time to update the remote walsender's latest reply
 	 * message received time.
@@ -4284,6 +4310,7 @@ wait_for_publisher_status(RetainConflictInfoData *data, bool status_received)
 {
 	FullTransactionId remote_full_xid;
 	uint32		remote_epoch = data->remote_epoch;
+	TimestampTz skew;
 
 	/*
 	 * Return if we have requested but not yet received the publisher status.
@@ -4291,6 +4318,21 @@ wait_for_publisher_status(RetainConflictInfoData *data, bool status_received)
 	if (!status_received)
 		return;
 
+	skew = data->candidate_xid_time - data->reply_time;
+
+	/*
+	 * If the publisher's clock lags behind that of the subscriber, wait for
+	 * the publisher's clock to catch up before proceeding to the next phase.
+	 */
+	if (skew > 0)
+	{
+		data->phase = RCI_REQUEST_PUBLISHER_STATUS;
+		data->next_request_time = GetCurrentTimestamp() + skew;
+
+		maybe_advance_nonremovable_xid(data, false);
+		return;
+	}
+
 	if (!FullTransactionIdIsValid(data->last_phase_at))
 		data->last_phase_at = FullTransactionIdFromEpochAndXid(data->remote_epoch,
 															   data->remote_nextxid);
@@ -4327,22 +4369,6 @@ 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
@@ -4398,6 +4424,7 @@ wait_for_local_flush(RetainConflictInfoData *data)
 	data->last_phase_at = InvalidFullTransactionId;
 	data->candidate_xid = InvalidFullTransactionId;
 	data->reply_time = 0;
+	data->next_request_time = 0;
 
 	/* process the next phase */
 	maybe_advance_nonremovable_xid(data, false);
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 8ae2783440..598d3b7374 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -2679,6 +2679,7 @@ ProcessStandbyPSRequestMessage(void)
 	FullTransactionId nextFullXid;
 	WalSnd	   *walsnd = MyWalSnd;
 	TimestampTz replyTime;
+	TimestampTz now;
 
 	/*
 	 * This shouldn't happen because we don't support getting primary status
@@ -2697,6 +2698,14 @@ ProcessStandbyPSRequestMessage(void)
 	walsnd->replyTime = replyTime;
 	SpinLockRelease(&walsnd->mutex);
 
+	/*
+	 * Get a new timestamp at the beginning to ensure that any transactions
+	 * assigned a timestamp earlier than this point either complete before the
+	 * current WAL (Write-Ahead Logging) write position or are detected in
+	 * GetOldestTransactionIdInCommit().
+	 */
+	now = GetCurrentTimestamp();
+
 	oldestXidInCommit = GetOldestTransactionIdInCommit();
 	nextFullXid = ReadNextFullTransactionId();
 	lsn = GetXLogWriteRecPtr();
@@ -2710,7 +2719,7 @@ ProcessStandbyPSRequestMessage(void)
 	pq_sendint32(&output_message, oldestXidInCommit);
 	pq_sendint32(&output_message, XidFromFullTransactionId(nextFullXid));
 	pq_sendint32(&output_message, EpochFromFullTransactionId(nextFullXid));
-	pq_sendint64(&output_message, GetCurrentTimestamp());
+	pq_sendint64(&output_message, now);
 
 	/* ... and send it wrapped in CopyData */
 	pq_putmessage_noblock('d', output_message.data, output_message.len);
-- 
2.30.0.windows.2

