From 4c4898ae00ef01b2e9794ff78236edf6d835106f Mon Sep 17 00:00:00 2001 From: Nisha Moond Date: Mon, 12 Oct 2026 14:37:26 +0530 Subject: [PATCH v80_2 3/3] Avoid stalling apply on the conflict log table with two_phase A prepared transaction that recorded a conflict holds a lock on the conflict log table until the apply worker applies COMMIT PREPARED or ROLLBACK PREPARED. A command waiting for a stronger lock on the table blocks the worker's next conflict insert, so the prepared transaction is never finished. This is not detected as a deadlock, so replication stalls. Fix by: - Refusing DROP SUBSCRIPTION while prepared transactions exist, if the subscription has a conflict log table. - Allowing conflict_log_destination to change from table/all to log only when the subscription is disabled, no worker is running and no prepared transactions exist, as for two_phase. - Making the apply worker error out instead of waiting when the lock on the conflict log table or its TOAST table is not available (e.g. TRUNCATE, VACUUM FULL), so that resolved conflicts are always recorded. --- doc/src/sgml/ref/alter_subscription.sgml | 9 ++ doc/src/sgml/ref/drop_subscription.sgml | 8 +- src/backend/commands/subscriptioncmds.c | 78 ++++++++++- src/backend/replication/logical/conflict.c | 73 +++++++++-- src/test/regress/expected/subscription.out | 6 + src/test/regress/sql/subscription.sql | 6 + src/test/subscription/t/035_conflicts.pl | 142 ++++++++++++++++++++- 7 files changed, 298 insertions(+), 24 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index d1c49c2c7c4..592c4f0c3b9 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -369,6 +369,15 @@ ALTER SUBSCRIPTION name RENAME TO < log, logging to the table stops and the conflict log table is automatically dropped. + + + Changing conflict_log_destination + to log from table or + all requires the subscription to be disabled and to + have no prepared transactions, as they may hold a lock on the conflict + log table. To resolve them, enable the subscription until they are + applied, then disable it and try again. + diff --git a/doc/src/sgml/ref/drop_subscription.sgml b/doc/src/sgml/ref/drop_subscription.sgml index e4c646fcf2a..2c19cf21901 100644 --- a/doc/src/sgml/ref/drop_subscription.sgml +++ b/doc/src/sgml/ref/drop_subscription.sgml @@ -122,7 +122,13 @@ DROP SUBSCRIPTION [ IF EXISTS ] name conflict_log_destination is set to table or all), DROP SUBSCRIPTION automatically drops - the associated conflict log table. + the associated conflict log table. In that case, the subscription cannot be + dropped while it has prepared transactions (see + two_phase), + because they may hold a lock on the conflict log table that is released only + when the apply worker applies the corresponding COMMIT + PREPARED or ROLLBACK PREPARED. Resolve these + transactions first (see pg_prepared_xacts). diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 36bb20a1f6e..0886810987b 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1504,13 +1504,16 @@ CheckAlterSubOption(Subscription *sub, const char *option, { Assert(strcmp(option, "failover") == 0 || strcmp(option, "two_phase") == 0 || - strcmp(option, "retain_dead_tuples") == 0); + strcmp(option, "retain_dead_tuples") == 0 || + strcmp(option, "conflict_log_destination") == 0); /* - * Altering the retain_dead_tuples option does not update the slot on the - * publisher. + * Altering the retain_dead_tuples or conflict_log_destination option does + * not update the slot on the publisher. */ - Assert(!slot_needs_update || strcmp(option, "retain_dead_tuples") != 0); + Assert(!slot_needs_update || + (strcmp(option, "retain_dead_tuples") != 0 && + strcmp(option, "conflict_log_destination") != 0)); /* * Do not allow changing the option if the subscription is enabled. This @@ -1555,6 +1558,13 @@ CheckAlterSubOption(Subscription *sub, const char *option, * avoid the race conditions described above, but we maintain the * restriction for both enable and disable operations for the sake of * consistency. + * + * Also, do not allow changing the conflict_log_destination option to + * 'log' when the subscription is enabled. This drops the conflict log + * table, which has to wait for any prepared transaction holding a lock on + * it. Only the apply worker can finish such a transaction, but the apply + * worker itself would be blocked if it tries to insert into the table + * while the drop is waiting, so neither could proceed. */ if (sub->enabled) ereport(ERROR, @@ -2103,6 +2113,52 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, bool update_relid; Oid relid = InvalidOid; + /* + * Changing to 'log' drops the conflict log table, + * which has to wait for any prepared transaction + * holding a lock on it. If the apply worker, which + * is the only one that can finish such a transaction, + * is blocked by that drop, neither can proceed. So + * allow it only when the subscription is disabled, no + * worker is running and there are no prepared + * transactions. + */ + if (CONFLICTS_LOGGED_TO_TABLE(old_dest) && + !CONFLICTS_LOGGED_TO_TABLE(opts.conflictlogdest)) + { + CheckAlterSubOption(sub, "conflict_log_destination", + false, isTopLevel); + + /* + * Note that workers may still survive even if the + * subscription has been disabled. + * + * Ensure workers have already been exited to + * avoid getting prepared transactions while we + * are dropping the conflict log table. + */ + if (logicalrep_workers_find(subid, true, true)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot alter option \"%s\" when logical replication worker is still running", + "conflict_log_destination"), + errhint("Try again after some time."))); + + /* + * The conflict log table cannot be dropped if + * there are any uncommitted prepared transactions + * present otherwise it can wait forever for a + * lock held by them. + */ + if (LookupGXactBySubid(subid)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot alter option \"%s\" when prepared transactions exist", + "conflict_log_destination"), + errdetail("Prepared transactions of the subscription may hold a lock on its conflict log table."), + errhint("Resolve these transactions and try again."))); + } + values[Anum_pg_subscription_subconflictlogdest - 1] = CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]); replaces[Anum_pg_subscription_subconflictlogdest - 1] = true; @@ -2742,6 +2798,20 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) } list_free(subworkers); + /* + * The subscription cannot be dropped if there are any uncommitted + * prepared transactions present, otherwise dropping the conflict log + * table can wait forever for a lock held by them, as the apply worker + * that would resolve them has been stopped above. + */ + if (OidIsValid(subconflictlogrelid) && LookupGXactBySubid(subid)) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot drop subscription \"%s\" when prepared transactions exist", + subname), + errdetail("Prepared transactions of the subscription may hold a lock on its conflict log table."), + errhint("Resolve these transactions and try again.")); + /* * Remove the no-longer-useful entry in the launcher's table of apply * worker start times. diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index 79394e8d1b6..90e618e2105 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -165,6 +165,7 @@ static void insert_conflict_log_tuple(EState *estate, Relation rel, ConflictType conflict_type, TupleTableSlot *searchslot, List *conflicttuples); +static void lock_conflictlog_relation(Oid relid); /* * Builds the TupleDesc for the conflict log table. @@ -354,11 +355,9 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel, log_dest_logfile = CONFLICTS_LOGGED_TO_LOG(dest); /* - * If a conflict log table was requested but it has been dropped - * concurrently (e.g. a concurrent ALTER SUBSCRIPTION changed - * conflict_log_destination), get_conflictlog_dest_and_table() - * returned NULL. Fall back to logging to the server log so that the - * conflict is not lost. + * If a conflict log table was requested but it no longer exists, + * get_conflictlog_dest_and_table() returned NULL. Fall back to + * logging to the server log so that the conflict is not lost. */ if (log_dest_table && conflictlogrel == NULL) { @@ -492,16 +491,18 @@ InitConflictIndexes(ResultRelInfo *relInfo) * destinations, opens and returns the relation handle for the conflict log * table. * - * The table is opened with try_table_open(), so NULL is returned if the - * conflict log table has been dropped concurrently (e.g. by an ALTER - * SUBSCRIPTION that changed conflict_log_destination). Callers must treat a - * NULL result for a table destination as "table unavailable" and fall back to - * server-log reporting rather than failing. + * Errors out if the lock on the conflict log table or its TOAST table is not + * immediately available, see lock_conflictlog_relation(). + * + * NULL is returned if the conflict log table no longer exists. Callers must + * treat a NULL result for a table destination as "table unavailable" and fall + * back to server-log reporting rather than failing. */ static Relation get_conflictlog_dest_and_table(ConflictLogDest *log_dest) { Oid conflictlogrelid; + Relation rel; /* * Convert the text log destination to the internal enum. MySubscription @@ -517,12 +518,56 @@ get_conflictlog_dest_and_table(ConflictLogDest *log_dest) Assert(OidIsValid(conflictlogrelid)); + lock_conflictlog_relation(conflictlogrelid); + /* - * Use try_table_open(): the table may have been dropped concurrently by - * an ALTER SUBSCRIPTION that changed conflict_log_destination. Returning - * NULL lets the caller fall back to the server log instead of failing. + * The table can't be dropped while the apply worker is running: ALTER + * SUBSCRIPTION ... SET (conflict_log_destination = 'log') requires no + * running worker, and DROP SUBSCRIPTION stops the workers first. Still, + * use try_table_open() to be safe, so that the caller falls back to the + * server log if the table is gone instead of failing. */ - return try_table_open(conflictlogrelid, RowExclusiveLock); + rel = try_table_open(conflictlogrelid, NoLock); + if (rel == NULL) + { + UnlockRelationOid(conflictlogrelid, RowExclusiveLock); + return NULL; + } + + /* + * Also lock the TOAST table now. Inserting a toasted value locks it + * separately, and a command run directly on it, such as REINDEX TABLE, is + * not noticed by the lock on the conflict log table above. + */ + if (OidIsValid(rel->rd_rel->reltoastrelid)) + lock_conflictlog_relation(rel->rd_rel->reltoastrelid); + + return rel; +} + +/* + * lock_conflictlog_relation + * + * Lock the conflict log table or its TOAST table in RowExclusiveLock mode, + * erroring out if the lock is not immediately available. + * + * Don't wait for the lock. A command that needs a stronger lock on the + * relation, such as TRUNCATE, may be waiting for a prepared transaction of + * this subscription that holds a lock on it. If we waited behind that + * command, we could never apply the COMMIT PREPARED or ROLLBACK PREPARED that + * releases the lock. Error out instead of falling back to the server log, so + * that a conflict is never resolved without being recorded in the table. + */ +static void +lock_conflictlog_relation(Oid relid) +{ + if (!ConditionalLockRelationOid(relid, RowExclusiveLock)) + ereport(ERROR, + errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on conflict log table of subscription \"%s\"", + MySubscription->name), + errdetail("Another session holds or is waiting for a conflicting lock on the conflict log table or its TOAST table."), + errhint("If that session is waiting for a prepared transaction of this subscription, cancel it, as only the apply worker can finish that transaction.")); } /* diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index b678b7fe007..565704d3df5 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -775,6 +775,12 @@ FROM pg_subscription WHERE subname = 'regress_conflict_test2'; table | t (1 row) +-- fail - transition from 'table' to 'log' requires a disabled subscription +BEGIN; +ALTER SUBSCRIPTION regress_conflict_test2 ENABLE; +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); +ERROR: cannot set option "conflict_log_destination" for enabled subscription +ROLLBACK; -- transition from 'table' to 'log' -- should drop the table and clear subconflictlogrelid ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index cfeebaf9302..39c2255d10d 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -586,6 +586,12 @@ ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'table SELECT subconflictlogdest, subconflictlogrelid = :old_relid AS relid_unchanged FROM pg_subscription WHERE subname = 'regress_conflict_test2'; +-- fail - transition from 'table' to 'log' requires a disabled subscription +BEGIN; +ALTER SUBSCRIPTION regress_conflict_test2 ENABLE; +ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); +ROLLBACK; + -- transition from 'table' to 'log' -- should drop the table and clear subconflictlogrelid ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log'); diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl index 9e7f56d3c08..4ab07e2ea3b 100644 --- a/src/test/subscription/t/035_conflicts.pl +++ b/src/test/subscription/t/035_conflicts.pl @@ -11,14 +11,19 @@ use Test::More; # Setup ############################### -# Create a publisher node +# Create a publisher node. Prepared transactions are needed by the two_phase +# tests at the end. my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); $node_publisher->init(allows_streaming => 'logical'); +$node_publisher->append_conf('postgresql.conf', + "max_prepared_transactions = 10"); $node_publisher->start; # Create a subscriber node my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); $node_subscriber->init(allows_streaming => 'logical'); +$node_subscriber->append_conf('postgresql.conf', + "max_prepared_transactions = 10"); $node_subscriber->start; # Create a table on publisher @@ -576,11 +581,11 @@ is($clt_check_temporal, 1, 'delete_missing on tab_temporal logged into CLT on Node B'); my $clt_row_temporal = $node_B->safe_psql('postgres', - "SELECT replica_identity::text FROM $clt_BA WHERE relname = 'tab_temporal';"); -is($clt_row_temporal, - '{"id":"[1,2)","valid_at":"{[2020-01-01,2021-01-01)}"}', - 'replica identity uses table column type rather than index opckeytype' + "SELECT replica_identity::text FROM $clt_BA WHERE relname = 'tab_temporal';" ); +is( $clt_row_temporal, + '{"id":"[1,2)","valid_at":"{[2020-01-01,2021-01-01)}"}', + 'replica identity uses table column type rather than index opckeytype'); # Restore tap_pub_A to publishing only 'tab', as the later tests expect. $node_A->safe_psql('postgres', @@ -1075,4 +1080,131 @@ $node_subscriber->safe_psql('clt_ts_test', "DROP SUBSCRIPTION sub_ts_test"); $node_subscriber->safe_psql('postgres', "DROP DATABASE clt_ts_test"); $node_subscriber->safe_psql('postgres', "DROP TABLESPACE backup_space"); +############################################################################### +# A prepared transaction that recorded a conflict holds a lock on the conflict +# log table until the apply worker applies COMMIT PREPARED or ROLLBACK +# PREPARED. Dropping the table would wait for it, while the apply worker can +# wait behind the drop, so neither could proceed. Check that the conflict log +# table can't be dropped by ALTER SUBSCRIPTION ... SET +# (conflict_log_destination = 'log') unless the subscription is disabled and +# has no prepared transactions, nor by DROP SUBSCRIPTION while prepared +# transactions exist. +############################################################################### +$node_publisher->safe_psql( + 'postgres', qq[ + CREATE TABLE tab_2pc (a int PRIMARY KEY); + CREATE PUBLICATION pub_2pc FOR TABLE tab_2pc; +]); +$node_subscriber->safe_psql('postgres', + "CREATE TABLE tab_2pc (a int PRIMARY KEY)"); + +my $appname_2pc = 'sub_2pc'; +$node_subscriber->safe_psql( + 'postgres', + "CREATE SUBSCRIPTION sub_2pc + CONNECTION '$publisher_connstr application_name=$appname_2pc' + PUBLICATION pub_2pc + WITH (two_phase = true, conflict_log_destination = 'table')"); +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname_2pc); +$node_subscriber->poll_query_until('postgres', + "SELECT subtwophasestate = 'e' FROM pg_subscription WHERE subname = 'sub_2pc'" +) or die "Timed out while waiting for two_phase to be enabled"; + +my $clt_2pc = $node_subscriber->safe_psql('postgres', + "SELECT subconflictlogrelid::regclass FROM pg_subscription WHERE subname = 'sub_2pc'" +); + +# Replicate a row, then remove it locally so that the remote DELETE of it +# raises a delete_missing conflict. +$node_publisher->safe_psql('postgres', "INSERT INTO tab_2pc VALUES (1)"); +$node_publisher->wait_for_catchup($appname_2pc); +$node_subscriber->safe_psql('postgres', "DELETE FROM tab_2pc WHERE a = 1"); + +$node_publisher->safe_psql( + 'postgres', qq[ + BEGIN; + DELETE FROM tab_2pc WHERE a = 1; + PREPARE TRANSACTION 'gx_2pc'; +]); +$node_publisher->wait_for_catchup($appname_2pc); + +# The prepared transaction on the subscriber holds the lock taken to record +# the conflict. +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_locks + WHERE relation = '$clt_2pc'::regclass AND pid IS NULL AND granted"), + '1', + 'prepared transaction holds a lock on the conflict log table'); + +# Fail - the subscription is enabled +($cmdret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION sub_2pc SET (conflict_log_destination = 'log')"); +like( + $stderr, + qr/ERROR: cannot set option "conflict_log_destination" for enabled subscription/, + "changing conflict_log_destination to 'log' is not allowed for enabled subscription" +); + +# Disable the subscription and wait for the apply worker to stop +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub_2pc DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'sub_2pc' AND pid IS NOT NULL" +) or die "Timed out while waiting for the apply worker to stop"; + +# Fail - prepared transactions exist +($cmdret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION sub_2pc SET (conflict_log_destination = 'log')"); +like( + $stderr, + qr/ERROR: cannot alter option "conflict_log_destination" when prepared transactions exist/, + "changing conflict_log_destination to 'log' is not allowed when prepared transactions exist" +); + +($cmdret, $stdout, $stderr) = + $node_subscriber->psql('postgres', "DROP SUBSCRIPTION sub_2pc"); +like( + $stderr, + qr/ERROR: cannot drop subscription "sub_2pc" when prepared transactions exist/, + "dropping subscription with a conflict log table is not allowed when prepared transactions exist" +); + +# Resolve the prepared transaction: commit it on the publisher and let the +# apply worker apply COMMIT PREPARED. +$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'gx_2pc'"); +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub_2pc ENABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_prepared_xacts") + or die + "Timed out while waiting for the prepared transaction to be resolved"; +is( $node_subscriber->safe_psql( + 'postgres', "SELECT conflict_type FROM $clt_2pc"), + 'delete_missing', + 'conflict recorded by the prepared transaction is committed'); + +# With the subscription disabled and no prepared transactions, the conflict +# log table can be dropped. +$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub_2pc DISABLE"); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_subscription WHERE subname = 'sub_2pc' AND pid IS NOT NULL" +) or die "Timed out while waiting for the apply worker to stop"; +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION sub_2pc SET (conflict_log_destination = 'log')"); +is( $node_subscriber->safe_psql( + 'postgres', + "SELECT subconflictlogrelid FROM pg_subscription WHERE subname = 'sub_2pc'" + ), + '0', + "conflict_log_destination is changed to 'log' for disabled subscription without prepared transactions" +); + +# Clean up +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub_2pc"); +$node_subscriber->safe_psql('postgres', "DROP TABLE tab_2pc"); +$node_publisher->safe_psql( + 'postgres', qq[ + DROP PUBLICATION pub_2pc; + DROP TABLE tab_2pc; +]); + done_testing(); -- 2.54.0 (Apple Git-157)