public inbox for [email protected]
help / color / mirror / Atom feedFrom: Zhijie Hou (Fujitsu) <[email protected]>
To: PostgreSQL Hackers <[email protected]>
Subject: Fix slot synchronization with two_phase decoding enabled
Date: Tue, 25 Mar 2025 05:35:32 +0000
Message-ID: <TYAPR01MB5724CC7C288535BBCEEE65DA94A72@TYAPR01MB5724.jpnprd01.prod.outlook.com> (raw)
Hi,
When testing the slot synchronization with logical replication slots that
enabled two_phase decoding, I found that transactions prepared before two-phase
decoding is enabled may fail to replicate to the subscriber after being
committed on a promoted standby following a failover.
To reproduce this issue, please follow these steps (also detailed in the
attached TAP test, v1-0001):
1. sub: create a subscription with (two_phase = false)
2. primary (pub): prepare a txn A.
3. sub: alter subscription set (two_phase = true) and wait for the logical slot to
be synced to standby.
4. primary (pub): stop primary, promote the standby and let the subscriber use
the promoted standby as publisher.
5. promoted standby (pub): COMMIT PREPARED A;
6. sub: the apply worker will report the following ERROR because it didn't
receive the PREPARE.
ERROR: prepared transaction with identifier "pg_gid_16387_752" does not exist
I think the root cause of this issue is that the two_phase_at field of the
slot, which indicates the LSN from which two-phase decoding is enabled (used to
prevent duplicate data transmission for prepared transactions), is not
synchronized to the standby server.
In step 3, transaction A is not immediately replicated because it occurred
before enabling two-phase decoding. Thus, the prepared transaction should only
be replicated after decoding the final COMMIT PREPARED, as referenced in
ReorderBufferFinishPrepared(). However, due to the invalid two_phase_at on the
standby, the prepared transaction fails to send at that time.
This problem arises after the support for altering the two-phase option
(1462aad). Previously, two-phase was only enabled during slot creation, which
wait for all prepared transactions to finish (via ... -> SnapBuildWaitSnapshot)
before reaching a consistent state, so the bug didn't exist.
To address the issue, I propose synchronizing the two_phase_at field to the
standby server, as implemented in the attached patches. As mentioned
earlier,this bug exists only for PG18 so we do not need to back patch.
v1-0001: Tap test to reproduce the issue
I place this patch as the first one so that reviewers can run it
independently to reproduce the issue. Once the problem is thoroughly
understood and the fix proves stable, the patches can be integrated.
v1-0002: Display two_phase_at in the pg_replication_slots view
v1-0003: Sync the two_phase_at field of a replication slot to the standby
An alternative approach might be modifying ALTER_REPLICATION_SLOT to wait
for all prepared transactions to commit when enabling two-phase. However, this
appears inelegant and less user friendly.
Best Regards,
Hou zj
Attachments:
[application/octet-stream] v1-0001-Tap-test-to-reproduce-the-issue.patch (5.5K, ../TYAPR01MB5724CC7C288535BBCEEE65DA94A72@TYAPR01MB5724.jpnprd01.prod.outlook.com/2-v1-0001-Tap-test-to-reproduce-the-issue.patch)
download | inline diff:
From 81f47cfcf146dd2a99c3d6b2e1f74de3dcdc5b85 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Thu, 27 Feb 2025 10:49:32 +0800
Subject: [PATCH v1 1/3] Tap test to reproduce the issue
This patch set fixes a bug for slot synchronization with logical replication
slots that enabled two_phase decoding. As it stands, transactions prepared
before two-phase decoding is enabled may fail to replicate to the subscriber
after being committed on a promoted standby following a failover.
The issue arises because the two_phase_at field of a slot, which tracks the LSN
from which two-phase decoding starts, is not synchronized to standby servers.
Without this field, the logical decoding might incorrectly identify prepared
transaction as already replicated to the subscriber, causing them to be
skipped.
To address the issue, this patch set makes the two_phase_at field of the slot
visible in the pg_replication_slots view and enables the slot synchronization
to copy this value to the corresponding synced slot on the standby server.
---
.../t/040_standby_failover_slots_sync.pl | 68 ++++++++++++++++++-
1 file changed, 66 insertions(+), 2 deletions(-)
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index 50388a494d6..692b746ccd9 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -22,7 +22,11 @@ $publisher->init(
# Disable autovacuum to avoid generating xid during stats update as otherwise
# the new XID could then be replicated to standby at some random point making
# slots at primary lag behind standby during slot sync.
-$publisher->append_conf('postgresql.conf', 'autovacuum = off');
+$publisher->append_conf(
+ 'postgresql.conf', qq{
+autovacuum = off
+max_prepared_transactions = 1
+});
$publisher->start;
$publisher->safe_psql('postgres',
@@ -33,6 +37,7 @@ my $publisher_connstr = $publisher->connstr . ' dbname=postgres';
# Create a subscriber node, wait for sync to complete
my $subscriber1 = PostgreSQL::Test::Cluster->new('subscriber1');
$subscriber1->init;
+$subscriber1->append_conf('postgresql.conf', 'max_prepared_transactions = 1');
$subscriber1->start;
# Capture the time before the logical failover slot is created on the
@@ -830,13 +835,63 @@ $primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots',
"'sb1_slot'");
$primary->reload;
+##################################################
+# Test the synchronization of the two_phase setting for a subscription with the
+# standby. Additionally, prepare a transaction before enabling the two_phase
+# option; subsequent tests will verify if it can be correctly replicated to the
+# subscriber after committing it on the promoted standby.
+##################################################
+
+$standby1->start;
+
+# Prepare a transaction
+$primary->safe_psql(
+ 'postgres', qq[
+ BEGIN;
+ INSERT INTO tab_int values(0);
+ PREPARE TRANSACTION 'test_twophase_slotsync';
+]);
+
+$primary->wait_for_replay_catchup($standby1);
+$primary->wait_for_catchup('regress_mysub1');
+
+# Disable the subscription to allow changing the two_phase option.
+$subscriber1->safe_psql('postgres',
+ "ALTER SUBSCRIPTION regress_mysub1 DISABLE");
+
+# Wait for the replication slot to become inactive on the publisher
+$primary->poll_query_until(
+ 'postgres',
+ "SELECT COUNT(*) FROM pg_catalog.pg_replication_slots WHERE slot_name = 'lsub1_slot' AND active='f'",
+ 1);
+
+# Set two_phase to true and enable the subscription
+$subscriber1->safe_psql(
+ 'postgres', qq[
+ ALTER SUBSCRIPTION regress_mysub1 SET (two_phase = true);
+ ALTER SUBSCRIPTION regress_mysub1 ENABLE;
+]);
+
+# Confirm that two_phase setting of lsub1_slot slot is synced to the standby
+ok( $standby1->poll_query_until(
+ 'postgres',
+ "SELECT two_phase from pg_replication_slots WHERE slot_name = 'lsub1_slot' AND synced AND NOT temporary;"
+ ),
+ 'two_phase setting of slot lsub1_slot synced to standby');
+
+# Confirm that the prepared transaction is not yet replicated to the
+# subscriber.
+$result = $subscriber1->safe_psql('postgres',
+ "SELECT count(*) = 0 FROM pg_prepared_xacts;");
+is($result, 't',
+ "the prepared transaction is not replicated to the subscriber");
+
##################################################
# Promote the standby1 to primary. Confirm that:
# a) the slot 'lsub1_slot' and 'snap_test_slot' are retained on the new primary
# b) logical replication for regress_mysub1 is resumed successfully after failover
# c) changes can be consumed from the synced slot 'snap_test_slot'
##################################################
-$standby1->start;
$primary->wait_for_replay_catchup($standby1);
# Capture the time before the standby is promoted
@@ -885,6 +940,15 @@ $standby1->wait_for_catchup('regress_mysub1');
is($subscriber1->safe_psql('postgres', q{SELECT count(*) FROM tab_int;}),
"20", 'data replicated from the new primary');
+# Commit the prepared transaction
+$standby1->safe_psql('postgres',
+ "COMMIT PREPARED 'test_twophase_slotsync';");
+$standby1->wait_for_catchup('regress_mysub1');
+
+# Confirm that the prepared transaction is replicated to the subscriber
+is($subscriber1->safe_psql('postgres', q{SELECT count(*) FROM tab_int;}),
+ "21", 'prepared data replicated from the new primary');
+
# Consume the data from the snap_test_slot. The synced slot should reach a
# consistent point by restoring the snapshot at the restart_lsn serialized
# during slot synchronization.
--
2.30.0.windows.2
[application/octet-stream] v1-0003-Sync-the-two_phase_at-field-of-a-replication-slot.patch (4.4K, ../TYAPR01MB5724CC7C288535BBCEEE65DA94A72@TYAPR01MB5724.jpnprd01.prod.outlook.com/3-v1-0003-Sync-the-two_phase_at-field-of-a-replication-slot.patch)
download | inline diff:
From 4eb7cfcca3e93ba8caba37eb1c3415ef9a764797 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Thu, 27 Feb 2025 13:32:35 +0800
Subject: [PATCH v1 3/3] Sync the two_phase_at field of a replication slot to
the standby
This patch enables slot synchronization to copy this value of two_phase_at field
to the corresponding synced slot on the standby server.
---
src/backend/replication/logical/slotsync.c | 15 +++++++++++----
.../recovery/t/040_standby_failover_slots_sync.pl | 8 +++++++-
2 files changed, 18 insertions(+), 5 deletions(-)
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 2c0a7439be4..a2bf89d1ba6 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -139,6 +139,7 @@ typedef struct RemoteSlot
bool failover;
XLogRecPtr restart_lsn;
XLogRecPtr confirmed_lsn;
+ XLogRecPtr two_phase_at;
TransactionId catalog_xmin;
/* RS_INVAL_NONE if valid, or the reason of invalidation */
@@ -249,6 +250,7 @@ update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid,
SpinLockAcquire(&slot->mutex);
slot->data.restart_lsn = remote_slot->restart_lsn;
slot->data.confirmed_flush = remote_slot->confirmed_lsn;
+ slot->data.two_phase_at = remote_slot->two_phase_at;
slot->data.catalog_xmin = remote_slot->catalog_xmin;
SpinLockRelease(&slot->mutex);
@@ -276,7 +278,8 @@ update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid,
if (remote_dbid != slot->data.database ||
remote_slot->two_phase != slot->data.two_phase ||
remote_slot->failover != slot->data.failover ||
- strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) != 0)
+ strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) != 0 ||
+ remote_slot->two_phase_at != slot->data.two_phase_at)
{
NameData plugin_name;
@@ -287,6 +290,7 @@ update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid,
slot->data.plugin = plugin_name;
slot->data.database = remote_dbid;
slot->data.two_phase = remote_slot->two_phase;
+ slot->data.two_phase_at = remote_slot->two_phase_at;
slot->data.failover = remote_slot->failover;
SpinLockRelease(&slot->mutex);
@@ -788,9 +792,9 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
static bool
synchronize_slots(WalReceiverConn *wrconn)
{
-#define SLOTSYNC_COLUMN_COUNT 9
+#define SLOTSYNC_COLUMN_COUNT 10
Oid slotRow[SLOTSYNC_COLUMN_COUNT] = {TEXTOID, TEXTOID, LSNOID,
- LSNOID, XIDOID, BOOLOID, BOOLOID, TEXTOID, TEXTOID};
+ LSNOID, XIDOID, BOOLOID, LSNOID, BOOLOID, TEXTOID, TEXTOID};
WalRcvExecResult *res;
TupleTableSlot *tupslot;
@@ -798,7 +802,7 @@ synchronize_slots(WalReceiverConn *wrconn)
bool some_slot_updated = false;
bool started_tx = false;
const char *query = "SELECT slot_name, plugin, confirmed_flush_lsn,"
- " restart_lsn, catalog_xmin, two_phase, failover,"
+ " restart_lsn, catalog_xmin, two_phase, two_phase_at, failover,"
" database, invalidation_reason"
" FROM pg_catalog.pg_replication_slots"
" WHERE failover and NOT temporary";
@@ -853,6 +857,9 @@ synchronize_slots(WalReceiverConn *wrconn)
&isnull));
Assert(!isnull);
+ d = slot_getattr(tupslot, ++col, &isnull);
+ remote_slot->two_phase_at = isnull ? InvalidXLogRecPtr : DatumGetLSN(d);
+
remote_slot->failover = DatumGetBool(slot_getattr(tupslot, ++col,
&isnull));
Assert(!isnull);
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index 692b746ccd9..160e349b967 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -872,10 +872,16 @@ $subscriber1->safe_psql(
ALTER SUBSCRIPTION regress_mysub1 ENABLE;
]);
+$primary->wait_for_catchup('regress_mysub1');
+
+my $two_phase_at = $primary->safe_psql('postgres',
+ "SELECT two_phase_at from pg_replication_slots WHERE slot_name = 'lsub1_slot';"
+);
+
# Confirm that two_phase setting of lsub1_slot slot is synced to the standby
ok( $standby1->poll_query_until(
'postgres',
- "SELECT two_phase from pg_replication_slots WHERE slot_name = 'lsub1_slot' AND synced AND NOT temporary;"
+ "SELECT two_phase AND '$two_phase_at' = two_phase_at from pg_replication_slots WHERE slot_name = 'lsub1_slot' AND synced AND NOT temporary;"
),
'two_phase setting of slot lsub1_slot synced to standby');
--
2.30.0.windows.2
[application/octet-stream] v1-0002-Display-two_phase_at-in-the-pg_replication_slots-.patch (5.4K, ../TYAPR01MB5724CC7C288535BBCEEE65DA94A72@TYAPR01MB5724.jpnprd01.prod.outlook.com/4-v1-0002-Display-two_phase_at-in-the-pg_replication_slots-.patch)
download | inline diff:
From 0028f7ef8a731f423476b2e92e2723912a43f8ba Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Fri, 28 Feb 2025 13:09:36 +0800
Subject: [PATCH v1 2/3] Display two_phase_at in the pg_replication_slots view
This patch makes the two_phase_at field of the slot visible in the
pg_replication_slots view.
---
doc/src/sgml/system-views.sgml | 11 +++++++++++
src/backend/catalog/system_views.sql | 1 +
src/backend/replication/slotfuncs.c | 8 +++++++-
src/include/catalog/pg_proc.dat | 6 +++---
src/test/regress/expected/rules.out | 3 ++-
5 files changed, 24 insertions(+), 5 deletions(-)
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 3f5a306247e..141c140331d 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2560,6 +2560,17 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>two_phase_at</structfield> <type>pg_lsn</type>
+ </para>
+ <para>
+ The address (<literal>LSN</literal>) from which the decoding of prepared
+ transactions is enabled. Always <literal>NULL</literal> for physical
+ slots.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>inactive_since</structfield> <type>timestamptz</type>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index a4d2cfdcaf5..9cd6bac6e1d 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1025,6 +1025,7 @@ CREATE VIEW pg_replication_slots AS
L.wal_status,
L.safe_wal_size,
L.two_phase,
+ L.two_phase_at,
L.inactive_since,
L.conflicting,
L.invalidation_reason,
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 146eef5871e..8a314b5ff3b 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -235,7 +235,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_get_replication_slots(PG_FUNCTION_ARGS)
{
-#define PG_GET_REPLICATION_SLOTS_COLS 19
+#define PG_GET_REPLICATION_SLOTS_COLS 20
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
@@ -406,6 +406,12 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
values[i++] = BoolGetDatum(slot_contents.data.two_phase);
+ if (slot_contents.data.two_phase &&
+ !XLogRecPtrIsInvalid(slot_contents.data.two_phase_at))
+ values[i++] = LSNGetDatum(slot_contents.data.two_phase_at);
+ else
+ nulls[i++] = true;
+
if (slot_contents.inactive_since > 0)
values[i++] = TimestampTzGetDatum(slot_contents.inactive_since);
else
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index cd9422d0bac..3c483a067aa 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11366,9 +11366,9 @@
proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', prorettype => 'record',
proargtypes => '',
- proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,timestamptz,bool,text,bool,bool}',
- proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,inactive_since,conflicting,invalidation_reason,failover,synced}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,pg_lsn,timestamptz,bool,text,bool,bool}',
+ proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,two_phase_at,inactive_since,conflicting,invalidation_reason,failover,synced}',
prosrc => 'pg_get_replication_slots' },
{ oid => '3786', descr => 'set up a logical replication slot',
proname => 'pg_create_logical_replication_slot', provolatile => 'v',
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 62f69ac20b2..7803318bee5 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1474,12 +1474,13 @@ pg_replication_slots| SELECT l.slot_name,
l.wal_status,
l.safe_wal_size,
l.two_phase,
+ l.two_phase_at,
l.inactive_since,
l.conflicting,
l.invalidation_reason,
l.failover,
l.synced
- FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, inactive_since, conflicting, invalidation_reason, failover, synced)
+ FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, two_phase_at, inactive_since, conflicting, invalidation_reason, failover, synced)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
--
2.30.0.windows.2
view thread (69+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected]
Subject: Re: Fix slot synchronization with two_phase decoding enabled
In-Reply-To: <TYAPR01MB5724CC7C288535BBCEEE65DA94A72@TYAPR01MB5724.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox