diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index 1f8d67fdd90..7d79e017d0e 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -395,19 +395,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; diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 52c6cc8c596..ecfee173c95 100644 --- a/src/backend/replication/logical/worker.c +++ b/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, @@ -406,7 +406,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. diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index bf856244788..2e2fb0309cd 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -2879,7 +2879,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");