database/postgres
database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-05-31 03:21:24 +03:00
Code

Don't retreat slot's confirmed_flush LSN.

Browse Source 
Prevent moving the confirmed_flush backwards, as this could lead to data
duplication issues caused by replicating already replicated changes.

This can happen when a client acknowledges an LSN it doesn't have to do
anything for, and thus didn't store persistently. After a restart, the
client can send the prior LSN that it stored persistently as an
acknowledgement, but we need to ignore such an LSN to avoid retreating
confirm_flush LSN.

Diagnosed-by: Zhijie Hou <houzj.fnst@fujitsu.com>
Author: shveta malik <shveta.malik@gmail.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Reviewed-by: Dilip Kumar <dilipbalaut@gmail.com>
Tested-by: Nisha Moond <nisha.moond412@gmail.com>
Backpatch-through: 13
Discussion: https://postgr.es/m/CAJpy0uDZ29P=BYB1JDWMCh-6wXaNqMwG1u1mB4=10Ly0x7HhwQ@mail.gmail.com
Discussion: https://postgr.es/m/OS0PR01MB57164AB5716AF2E477D53F6F9489A@OS0PR01MB5716.jpnprd01.prod.outlook.com
This commit is contained in:
Amit Kapila 2025-05-19 12:13:06 +05:30
parent f8db5c7a3f
commit ad5eaf390c
1 changed files with 21 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
src/backend/replication/logical/logical.c
View File

@ -1828,7 +1828,19 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
SpinLockAcquire(&MyReplicationSlot->mutex); SpinLockAcquire(&MyReplicationSlot->mutex);
MyReplicationSlot->data.confirmed_flush = lsn; /*
* Prevent moving the confirmed_flush backwards, as this could lead to
* data duplication issues caused by replicating already replicated
* changes.
*
* This can happen when a client acknowledges an LSN it doesn't have
* to do anything for, and thus didn't store persistently. After a
* restart, the client can send the prior LSN that it stored
* persistently as an acknowledgement, but we need to ignore such an
* LSN. See similar case handling in CreateDecodingContext.
*/
if (lsn > MyReplicationSlot->data.confirmed_flush)
MyReplicationSlot->data.confirmed_flush = lsn;
/* if we're past the location required for bumping xmin, do so */ /* if we're past the location required for bumping xmin, do so */
if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr && if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr &&
@ -1893,7 +1905,14 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
else else
{ {
SpinLockAcquire(&MyReplicationSlot->mutex); SpinLockAcquire(&MyReplicationSlot->mutex);
MyReplicationSlot->data.confirmed_flush = lsn;
/*
* Prevent moving the confirmed_flush backwards. See comments above
* for the details.
*/
if (lsn > MyReplicationSlot->data.confirmed_flush)
MyReplicationSlot->data.confirmed_flush = lsn;
SpinLockRelease(&MyReplicationSlot->mutex); SpinLockRelease(&MyReplicationSlot->mutex);
} }
} }