Use WAL insert position for conflict detection.

The retain_dead_tuples feature waits for concurrent transactions to be
applied before advancing conflict_detection_slot.xmin. It gets the WAL
write position from the publisher and waits until the apply remote
position passes it. This ensures that dead tuples, commit timestamps, and
origins needed for conflict detection (update_deleted,
update_origin_differs, and delete_origin_differs) are retained while
those transactions are applied.

However, with asynchronous commit enabled on the publisher, a committed
transaction may not have updated the WAL write position yet. As a result,
the reported WAL position can lag behind the latest committed transaction.
This can cause conflict_detection_slot.xmin to advance prematurely,
allowing dead tuples needed for conflict detection to be removed before
subsequent asynchronously committed transactions are applied.

Report the end of the last inserted WAL record instead. A transaction
inserts its commit record before it is marked committed, so the insert
position cannot be behind a transaction that has already committed.
Transactions that have entered the commit phase but have not yet inserted
their commit record are still accounted for by the in-commit transaction
ID reported in the same message.

In addition to above, add missing trailing period to update_deleted
conflict detail messages which are added for the same feature.

Reported-by: Nisha Moond <nisha.moond412@gmail.com>
Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Discussion: https://postgr.es/m/TY4PR01MB177182F547A62FC2666EC04EC94B72@TY4PR01MB17718.jpnprd01.prod.outlook.com
Backpatch-through: 19, where it was introduced

src/backend/replication/logical/conflict.c

@@ -567,19 +567,19 @@ errdetail_apply_conflict(EState *estate, ResultRelInfo *relinfo,
 			if (localts)
 			{
 				if (localorigin == InvalidReplOriginId)
-					appendStringInfo(&err_detail, _("The row to be updated was deleted locally in transaction %u at %s"),
+					appendStringInfo(&err_detail, _("The row to be updated was deleted locally in transaction %u at %s."),
 									 localxmin, timestamptz_to_str(localts));
 				else if (replorigin_by_oid(localorigin, true, &origin_name))
-					appendStringInfo(&err_detail, _("The row to be updated was deleted by a different origin \"%s\" in transaction %u at %s"),
+					appendStringInfo(&err_detail, _("The row to be updated was deleted by a different origin \"%s\" in transaction %u at %s."),
 									 origin_name, localxmin, timestamptz_to_str(localts));
 
 				/* The origin that modified this row has been removed. */
 				else
-					appendStringInfo(&err_detail, _("The row to be updated was deleted by a non-existent origin in transaction %u at %s"),
+					appendStringInfo(&err_detail, _("The row to be updated was deleted by a non-existent origin in transaction %u at %s."),
 									 localxmin, timestamptz_to_str(localts));
 			}
 			else
-				appendStringInfoString(&err_detail, _("The row to be updated was deleted"));
+				appendStringInfoString(&err_detail, _("The row to be updated was deleted."));
 
 			break;
 

src/backend/replication/logical/worker.c

@@ -158,8 +158,8 @@
  *
  * - RDT_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.
+ *   includes the latest WAL insert position and information about
+ *   transactions that are in the commit phase.
  *
  * - RDT_WAIT_FOR_PUBLISHER_STATUS:
  *   Wait for the status from the walsender. After receiving the first status,
@@ -417,7 +417,7 @@ typedef enum
 typedef struct RetainDeadTuplesData
 {
 	RetainDeadTuplesPhase phase;	/* current phase */
-	XLogRecPtr	remote_lsn;		/* WAL write position on the publisher */
+	XLogRecPtr	remote_lsn;		/* WAL insert position on the publisher */
 
 	/*
 	 * Oldest transaction ID that was in the commit phase on the publisher.

src/backend/replication/walsender.c

@@ -2895,7 +2895,30 @@ ProcessStandbyPSRequestMessage(void)
 	nextFullXid = ReadNextFullTransactionId();
 	fullOldestXidInCommit = FullTransactionIdFromAllowableAt(nextFullXid,
 															 oldestXidInCommit);
-	lsn = GetXLogWriteRecPtr();
+
+	/*
+	 * Report the end of the last inserted WAL record rather than the WAL
+	 * write position.  A transaction that commits asynchronously (with
+	 * synchronous_commit = off) clears DELAY_CHKPT_IN_COMMIT without flushing
+	 * its commit record, so it is visible to neither the in-commit scan above
+	 * nor a write position that has not yet reached its commit record.  The
+	 * subscriber waits until it has applied and flushed up to the reported
+	 * position before advancing its non-removable transaction ID, so the
+	 * reported position must cover every transaction that has already
+	 * committed, as each of those already carries a commit timestamp earlier
+	 * than this reply.  A transaction's commit timestamp is written as part
+	 * of its commit record, so it cannot have committed without that record
+	 * already being at or before the insert position.  A transaction that has
+	 * determined its commit timestamp but not yet inserted the record is
+	 * still in the commit phase, and so is reported by the in-commit scan
+	 * above.
+	 *
+	 * GetXLogInsertEndRecPtr() is used rather than GetXLogInsertRecPtr()
+	 * because the latter can return a position past the page header when the
+	 * last record ends at a page boundary, which can never match a record end
+	 * and would needlessly stall the subscriber's wait.
+	 */
+	lsn = GetXLogInsertEndRecPtr();
 
 	elog(DEBUG2, "sending primary status");