Fix a possibility of logical replication slot's restart_lsn going backwards.

MasahikoSawada · MasahikoSawada · commit 26c4e8968690 · 2024-11-15T17:06:00.000-08:00
Previously LogicalIncreaseRestartDecodingForSlot() accidentally accepted any LSN as the candidate_lsn and candidate_valid after the restart_lsn of the replication slot was updated, so it potentially caused the restart_lsn to move backwards. A scenario where this could happen in logical replication is: after a logical replication restart, based on previous candidate_lsn and candidate_valid values in memory, the restart_lsn advances upon receiving a subscriber acknowledgment. Then, logical decoding restarts from an older point, setting candidate_lsn and candidate_valid based on an old RUNNING_XACTS record. Subsequent subscriber acknowledgments then update the restart_lsn to an LSN older than the current value. In the reported case, after WAL files were removed by a checkpoint, the retreated restart_lsn prevented logical replication from restarting due to missing WAL segments. This change essentially modifies the 'if' condition to 'else if' condition within the function. The previous code had an asymmetry in this regard compared to LogicalIncreaseXminForSlot(), which does almost the same thing for different fields. The WAL removal issue was reported by Hubert Depesz Lubaczewski. Backpatch to all supported versions, since the bug exists since 9.4 where logical decoding was introduced. Reviewed-by: Tomas Vondra, Ashutosh Bapat, Amit Kapila Discussion: https://postgr.es/m/Yz2hivgyjS1RfMKs%40depesz.com Discussion: https://postgr.es/m/85fff40e-148b-4e86-b921-b4b846289132%40vondra.me Backpatch-through: 13
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
@@ -1645,6 +1645,7 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart
 	/* don't overwrite if have a newer restart lsn */
 	if (restart_lsn <= slot->data.restart_lsn)
 	{
+		SpinLockRelease(&slot->mutex);
 	}
 
 	/*
@@ -1655,6 +1656,7 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart
 	{
 		slot->candidate_restart_valid = current_lsn;
 		slot->candidate_restart_lsn = restart_lsn;
+		SpinLockRelease(&slot->mutex);
 
 		/* our candidate can directly be used */
 		updated_lsn = true;
@@ -1665,7 +1667,7 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart
 	 * might never end up updating if the receiver acks too slowly. A missed
 	 * value here will just cause some extra effort after reconnecting.
 	 */
-	if (slot->candidate_restart_valid == InvalidXLogRecPtr)
+	else if (slot->candidate_restart_valid == InvalidXLogRecPtr)
 	{
 		slot->candidate_restart_valid = current_lsn;
 		slot->candidate_restart_lsn = restart_lsn;

Original file line number	Diff line number	Diff line change
`@@ -1645,6 +1645,7 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart`
`1645`	`1645`	`/* don't overwrite if have a newer restart lsn */`
`1646`	`1646`	`if (restart_lsn <= slot->data.restart_lsn)`
`1647`	`1647`	`{`
	`1648`	`+ SpinLockRelease(&slot->mutex);`
`1648`	`1649`	`}`
`1649`	`1650`
`1650`	`1651`	`/*`
`@@ -1655,6 +1656,7 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart`
`1655`	`1656`	`{`
`1656`	`1657`	`slot->candidate_restart_valid = current_lsn;`
`1657`	`1658`	`slot->candidate_restart_lsn = restart_lsn;`
	`1659`	`+ SpinLockRelease(&slot->mutex);`
`1658`	`1660`
`1659`	`1661`	`/* our candidate can directly be used */`
`1660`	`1662`	`updated_lsn = true;`
`@@ -1665,7 +1667,7 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart`
`1665`	`1667`	`* might never end up updating if the receiver acks too slowly. A missed`
`1666`	`1668`	`* value here will just cause some extra effort after reconnecting.`
`1667`	`1669`	`*/`
`1668`		`- if (slot->candidate_restart_valid == InvalidXLogRecPtr)`
	`1670`	`+ else if (slot->candidate_restart_valid == InvalidXLogRecPtr)`
`1669`	`1671`	`{`
`1670`	`1672`	`slot->candidate_restart_valid = current_lsn;`
`1671`	`1673`	`slot->candidate_restart_lsn = restart_lsn;`