From 7fd75a690bc4e7f181769ed22dd09dbb001a4057 Mon Sep 17 00:00:00 2001 From: Nisha Moond Date: Fri, 16 Oct 2026 18:06:48 +0530 Subject: [PATCH v83 2/4] 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. - When two_phase is enabled, 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. - When two_phase is enabled and the lock on the conflict log table or its TOAST table is not immediately available (e.g. TRUNCATE, VACUUM FULL), erroring out immediately if prepared transactions exist for the subscription. Otherwise, wait for the lock with lock_timeout set to deadlock_timeout + 5s (unless lock_timeout is already set) so that transient lock waits and autovacuum cancellations can succeed while still breaking out if a parallel apply worker prepares a transaction during the wait. --- doc/src/sgml/ref/alter_subscription.sgml | 10 + doc/src/sgml/ref/drop_subscription.sgml | 16 +- src/backend/commands/subscriptioncmds.c | 80 +++++++- src/backend/replication/logical/conflict.c | 92 ++++++++- src/test/regress/expected/subscription.out | 7 + src/test/regress/sql/subscription.sql | 7 + src/test/subscription/t/035_conflicts.pl | 206 ++++++++++++++++++++- 7 files changed, 406 insertions(+), 12 deletions(-) diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml index d1c49c2c7c4..4bdaa8d1387 100644 --- a/doc/src/sgml/ref/alter_subscription.sgml +++ b/doc/src/sgml/ref/alter_subscription.sgml @@ -369,6 +369,16 @@ ALTER SUBSCRIPTION name RENAME TO < log, logging to the table stops and the conflict log table is automatically dropped. + + + When two_phase + is enabled, 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..2c33a5de06e 100644 --- a/doc/src/sgml/ref/drop_subscription.sgml +++ b/doc/src/sgml/ref/drop_subscription.sgml @@ -118,11 +118,19 @@ DROP SUBSCRIPTION [ IF EXISTS ] name - If a conflict log table exists for the subscription (that is, when - + If conflict_log_destination is set to table - or all), DROP SUBSCRIPTION automatically drops - the associated conflict log table. + or all, the subscription has an associated conflict log + table, which DROP SUBSCRIPTION drops along with the + subscription. However, DROP SUBSCRIPTION cannot be + executed while the subscription has prepared transactions (see + two_phase), + because such a transaction may hold a lock on the conflict log table that is + released only when the apply worker processes the corresponding + COMMIT PREPARED or ROLLBACK PREPARED. + Resolve these transactions first (see + pg_prepared_xacts) + before dropping the subscription. diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 36bb20a1f6e..384ebd1541a 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,10 @@ 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, when two_phase is enabled, do not allow changing the + * conflict_log_destination option to 'log' when the subscription is + * enabled; see AlterSubscription() and DropSubscription() for details. */ if (sub->enabled) ereport(ERROR, @@ -2103,6 +2110,50 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, bool update_relid; Oid relid = InvalidOid; + /* + * Similar to DropSubscription(), when two_phase is + * enabled, changing to 'log' drops the conflict log + * table, which can be blocked forever if a prepared + * transaction of the subscription holds a lock on it. + * 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) && + sub->twophasestate != LOGICALREP_TWOPHASE_STATE_DISABLED) + { + 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."))); + + /* + * Disallow the operation when a prepared + * transaction exists, as explained above. + */ + 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; @@ -2613,6 +2664,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) Oid subowner; Oid subserver; Oid subconflictlogrelid; + char subtwophasestate; char *subconninfo = NULL; Datum datum; bool isnull; @@ -2666,6 +2718,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) subowner = form->subowner; subserver = form->subserver; subconflictlogrelid = form->subconflictlogrelid; + subtwophasestate = form->subtwophasestate; must_use_password = !superuser_arg(subowner) && form->subpasswordrequired; /* must be owner */ @@ -2742,6 +2795,25 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) } list_free(subworkers); + /* + * Disallow dropping the subscription when there is a deadlock risk where + * the conflict log table is locked by a prepared transaction of the + * subscription. In that case, the subsequent attempt to drop the conflict + * log table would be blocked forever because we have stopped all the + * workers above and disallowed starting new ones, closing the possibility + * of applying COMMIT PREPARED or ROLLBACK PREPARED for that prepared + * transaction. + */ + if (OidIsValid(subconflictlogrelid) && + subtwophasestate != LOGICALREP_TWOPHASE_STATE_DISABLED && + 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 efc398286de..6b4f0a4da24 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -18,6 +18,7 @@ #include "access/genam.h" #include "access/heapam.h" #include "access/tableam.h" +#include "access/twophase.h" #include "catalog/heap.h" #include "catalog/pg_am.h" #include "catalog/pg_namespace.h" @@ -28,8 +29,10 @@ #include "replication/conflict.h" #include "replication/worker_internal.h" #include "storage/lmgr.h" +#include "storage/proc.h" #include "utils/array.h" #include "utils/builtins.h" +#include "utils/guc.h" #include "utils/json.h" #include "utils/lsyscache.h" #include "utils/pg_lsn.h" @@ -164,6 +167,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. @@ -449,6 +453,9 @@ InitConflictIndexes(ResultRelInfo *relInfo) * continues) are recorded in the conflict log table. ERROR-level conflicts * halt replication and abort the transaction, so they are always reported * exclusively to the server log. + * + * With two_phase enabled, errors out if the lock on the conflict log table or + * its TOAST table cannot be obtained, see lock_conflictlog_relation(). */ static Relation get_conflictlog_dest_and_table(int elevel, ConflictLogDest *log_dest) @@ -476,20 +483,92 @@ get_conflictlog_dest_and_table(int elevel, ConflictLogDest *log_dest) Assert(OidIsValid(conflictlogrelid)); + lock_conflictlog_relation(conflictlogrelid); + /* * Use try_table_open() because the table may have been dropped * concurrently by an ALTER SUBSCRIPTION that changed * conflict_log_destination. In that case, fall back to the server log * instead of failing. */ - conflictlogrel = try_table_open(conflictlogrelid, RowExclusiveLock); - - if (!conflictlogrel) + conflictlogrel = try_table_open(conflictlogrelid, NoLock); + if (conflictlogrel == NULL) + { + UnlockRelationOid(conflictlogrelid, RowExclusiveLock); *log_dest = CONFLICT_LOG_DEST_LOG; + 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(conflictlogrel->rd_rel->reltoastrelid)) + lock_conflictlog_relation(conflictlogrel->rd_rel->reltoastrelid); return conflictlogrel; } +/* + * Lock the conflict log table or its TOAST table in RowExclusiveLock mode. + * + * With two_phase enabled, a prepared transaction of this subscription that + * recorded a conflict holds a lock on the relation until we apply its COMMIT + * PREPARED or ROLLBACK PREPARED. A command that needs a stronger lock, such + * as TRUNCATE, waits for that transaction, and if we waited indefinitely + * behind the command we could never apply the COMMIT PREPARED or ROLLBACK + * PREPARED that releases the lock. + * + * Therefore, if the lock is not immediately available with two_phase enabled + * (and lock_timeout is not already configured), error out if any prepared + * transactions exist for this subscription. Otherwise, wait for the lock with + * lock_timeout set to DeadlockTimeout + 5s so that ProcSleep() can run + * DeadLockCheck() and interrupt a blocking autovacuum truncation, while still + * breaking out if a parallel apply worker of this subscription prepares a + * transaction while we wait. + * + * Without two_phase, no prepared transaction of this subscription can exist + * or appear while we run, as two_phase only takes effect after the workers + * restart, so wait for the lock as usual. + */ +static void +lock_conflictlog_relation(Oid relid) +{ + int save_nestlevel; + char buf[32]; + + if (ConditionalLockRelationOid(relid, RowExclusiveLock)) + return; + + if (MySubscription->twophasestate != LOGICALREP_TWOPHASE_STATE_ENABLED || + LockTimeout > 0) + { + LockRelationOid(relid, RowExclusiveLock); + return; + } + + if (LookupGXactBySubid(MySubscription->oid)) + 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("The apply worker will retry. If that session is waiting for a prepared transaction of this subscription, cancel it, as only the apply worker can finish that transaction.")); + + save_nestlevel = NewGUCNestLevel(); + + snprintf(buf, sizeof(buf), "%d", + Min(DeadlockTimeout, INT_MAX - 5000) + 5000); + (void) set_config_option("lock_timeout", buf, + PGC_USERSET, PGC_S_SESSION, + GUC_ACTION_SAVE, true, 0, false); + + LockRelationOid(relid, RowExclusiveLock); + + AtEOXact_GUC(true, save_nestlevel); +} + /* * Add SQLSTATE error code to the current conflict report. */ @@ -1015,6 +1094,7 @@ build_index_key_json(EState *estate, Relation localrel, StringInfoData result; int indnkeyatts; Datum datum; + int save_nestlevel; Assert(slot != NULL); @@ -1026,6 +1106,11 @@ build_index_key_json(EState *estate, Relation localrel, build_index_datums_from_slot(estate, localrel, slot, indexDesc, values, isnull); + save_nestlevel = NewGUCNestLevel(); + (void) set_config_option("datestyle", "ISO", + PGC_USERSET, PGC_S_SESSION, + GUC_ACTION_SAVE, true, 0, false); + /* * Build the JSON object here rather than handing the values to * row_to_json(), and render each of them with its type's output function. @@ -1099,6 +1184,7 @@ build_index_key_json(EState *estate, Relation localrel, appendStringInfoChar(&result, '}'); + AtEOXact_GUC(true, save_nestlevel); index_close(indexDesc, NoLock); datum = PointerGetDatum(cstring_to_text_with_len(result.data, result.len)); diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index b678b7fe007..68525c41698 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -775,6 +775,13 @@ FROM pg_subscription WHERE subname = 'regress_conflict_test2'; table | t (1 row) +-- fail - transition from 'table' to 'log' with two_phase requires a disabled subscription +BEGIN; +ALTER SUBSCRIPTION regress_conflict_test2 SET (two_phase = true); +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..89c9a32af06 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -586,6 +586,13 @@ 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' with two_phase requires a disabled subscription +BEGIN; +ALTER SUBSCRIPTION regress_conflict_test2 SET (two_phase = true); +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 970238c855e..69206eb3299 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 @@ -1183,4 +1188,203 @@ $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, and that the apply worker doesn't wait for the conflict +# log table lock behind such a command. +############################################################################### +$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)"); + +$node_subscriber->append_conf('postgresql.conf', "deadlock_timeout = 10ms"); +$node_subscriber->reload; + +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 rows, then remove them locally so that the remote DELETEs of +# them raise delete_missing conflicts. +$node_publisher->safe_psql('postgres', + "INSERT INTO tab_2pc VALUES (0), (1), (2)"); +$node_publisher->wait_for_catchup($appname_2pc); +$node_subscriber->safe_psql('postgres', "DELETE FROM tab_2pc"); + +# When no prepared transaction exists, the apply worker waits for a conflicting +# lock on the conflict log table past deadlock_timeout without erroring out, +# and records the conflict once the lock is released. +my $log_offset_2pc = -s $node_subscriber->logfile; +my $lock_session = $node_subscriber->background_psql('postgres'); +$lock_session->query_safe( + qq[ + BEGIN; + LOCK TABLE $clt_2pc IN ACCESS EXCLUSIVE MODE; +]); + +$node_publisher->safe_psql('postgres', "DELETE FROM tab_2pc WHERE a = 0"); +$node_subscriber->wait_for_log( + qr/still waiting for RowExclusiveLock on relation/, + $log_offset_2pc); +unlike( + slurp_file($node_subscriber->logfile, $log_offset_2pc), + qr/ERROR: could not obtain lock on conflict log table of subscription "sub_2pc"/, + 'apply worker does not error out on conflict log table lock when no prepared transactions exist' +); + +$lock_session->query_safe("ROLLBACK;"); +$lock_session->quit; +$node_publisher->wait_for_catchup($appname_2pc); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM $clt_2pc"), + '1', + 'conflict is recorded after conflict log table lock is released'); +$node_subscriber->safe_psql('postgres', "TRUNCATE $clt_2pc"); + +$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'); + +# A command needing a stronger lock on the conflict log table, such as +# TRUNCATE, waits for the prepared transaction. Check that the apply worker +# errors out on its next conflict instead of waiting behind the command, and +# applies the change once the command is cancelled. +$log_offset_2pc = -s $node_subscriber->logfile; +my $truncate_session = + $node_subscriber->background_psql('postgres', on_error_stop => 0); +my $truncate_pid = $truncate_session->query_safe("SELECT pg_backend_pid()"); +$truncate_session->query_until( + qr/truncate_started/, qq[ + \\echo truncate_started + TRUNCATE $clt_2pc; +]); +$node_subscriber->poll_query_until( + 'postgres', + "SELECT count(*) = 1 FROM pg_locks + WHERE relation = '$clt_2pc'::regclass + AND mode = 'AccessExclusiveLock' AND NOT granted" +) or die "Timed out while waiting for TRUNCATE to wait for the lock"; + +$node_publisher->safe_psql('postgres', "DELETE FROM tab_2pc WHERE a = 2"); +$node_subscriber->wait_for_log( + qr/ERROR: could not obtain lock on conflict log table of subscription "sub_2pc"/, + $log_offset_2pc); +pass( + 'apply worker errors out when the conflict log table lock is not available with two_phase enabled' +); + +$node_subscriber->safe_psql('postgres', + "SELECT pg_cancel_backend($truncate_pid)"); +$truncate_session->quit; +$node_publisher->wait_for_catchup($appname_2pc); +is( $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM $clt_2pc"), + '1', + 'conflict is recorded after the waiting command is cancelled'); + +# 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 count(*) FROM $clt_2pc WHERE conflict_type = 'delete_missing'" + ), + '2', + '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)