From b5b67961bbbb15ad6a01185ee36361751fc4f5a5 Mon Sep 17 00:00:00 2001 From: Nisha Moond Date: Fri, 4 Sep 2026 12:37:08 +0530 Subject: [PATCH v2 2/2] Reproducer TAP test --- src/backend/replication/logical/worker.c | 11 ++ src/test/subscription/t/035_conflicts.pl | 156 +++++++++++++++++++++++ 2 files changed, 167 insertions(+) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 7781bb1c168..3175c7470ec 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -290,6 +290,7 @@ #include "tcop/tcopprot.h" #include "utils/acl.h" #include "utils/guc.h" +#include "utils/injection_point.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -4900,6 +4901,16 @@ resume_conflict_info_retention(RetainDeadTuplesData *rdt_data) ? errdetail("Retention is re-enabled because the apply process has caught up with the publisher within the configured max_retention_duration.") : errdetail("Retention is re-enabled because max_retention_duration has been set to unlimited.")); + /* + * The retentionactive flag has been committed and the launcher woken, but + * this worker has not yet cleared its shared memory entry. A launcher + * cycle running in this window sees a worker that is still present but has + * an invalid oldest_nonremovable_xid; this injection point lets + * 035_conflicts.pl hold the window open to verify that the launcher does + * not advance the conflict detection slot's xmin in that case. + */ + INJECTION_POINT("logical-rep-resume-retention-before-exit", NULL); + /* * Restart the worker to let the launcher initialize * oldest_nonremovable_xid at startup. diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl index d66e71d1900..2d91d1a1f26 100644 --- a/src/test/subscription/t/035_conflicts.pl +++ b/src/test/subscription/t/035_conflicts.pl @@ -722,6 +722,162 @@ ok( $node_B->poll_query_until( $node_B->safe_psql('dbb', "DROP SUBSCRIPTION $subname_BA2"); $node_B->safe_psql('postgres', "DROP DATABASE dbb"); +############################################################################### +# Check that the conflict detection slot's xmin is not advanced while a +# subscription is resuming retention. +# +# On resume, the apply worker commits subretentionactive = true and wakes the +# launcher before restarting itself. A launcher cycle running in that window +# finds the worker still present but with an invalid oldest_nonremovable_xid, +# so its database contributes nothing to the computed minimum. The slot's +# xmin must not advance on the strength of the remaining databases alone; +# otherwise the restarted worker would be seeded with a value newer than its +# own database's oldest active transaction ID, violating the invariant checked +# in get_candidate_xid(). +# +# The injection point holds the worker inside that window so that the launcher +# is guaranteed to run a full cycle against it. +############################################################################### + +if ($injection_points_supported != 0) +{ + # Create a third database on node B with the same table and a retaining + # subscription in it. tap_sub_b_a in the postgres database keeps driving + # the slot's xmin throughout; this new subscription is the one that stops + # and resumes retention. + my $subname_BA3 = 'tap_sub_b_a3'; + + $node_B->safe_psql('postgres', "CREATE DATABASE dbc"); + $node_B->safe_psql('dbc', "CREATE TABLE tab (a int PRIMARY KEY, b int)"); + $node_B->safe_psql('dbc', + "CREATE SUBSCRIPTION $subname_BA3 + CONNECTION '$node_A_connstr application_name=$subname_BA3' + PUBLICATION tap_pub_A + WITH (retain_dead_tuples = true, origin = none, failover = true)"); + $node_B->wait_for_subscription_sync($node_A, $subname_BA3, 'dbc'); + + # Stall the flush position for dbc's subscription so that its retention + # stops on the max_retention_duration timeout. Only this subscription has + # failover enabled, so tap_sub_b_a is unaffected and keeps advancing the + # slot's xmin. + $node_A->safe_psql('postgres', + "SELECT * FROM pg_create_physical_replication_slot('blocker3');"); + $node_A->append_conf('postgresql.conf', + "synchronized_standby_slots = 'blocker3'"); + $node_A->reload; + + # Give the worker something to flush and a new transaction ID to chase, so + # that it reaches the phase where the timeout is checked. + $node_A->safe_psql('postgres', "INSERT INTO tab VALUES (111, 111);"); + $node_B->safe_psql('dbc', "SELECT txid_current() + 1;"); + + $log_offset = -s $node_B->logfile; + + $node_B->safe_psql('dbc', + "ALTER SUBSCRIPTION $subname_BA3 SET (max_retention_duration = 1);"); + + $node_B->wait_for_log( + qr/logical replication worker for subscription "$subname_BA3" has stopped retaining the information for detecting conflicts/, + $log_offset); + + # Hold a transaction with an assigned transaction ID open in dbc. This has + # to happen after retention has stopped: while it is held, dbc's oldest + # active transaction ID never moves, so the worker would find no new + # candidate and would never reach the timeout check above. + my $dbc_session = $node_B->background_psql('dbc'); + $dbc_session->query_until( + qr/starting_bg_psql/, q{ + \echo starting_bg_psql + BEGIN; + SELECT txid_current(); + }); + + # Arrange for the worker to hold inside the resume window. + $node_B->safe_psql('postgres', + "SELECT injection_points_attach('logical-rep-resume-retention-before-exit', 'wait');" + ); + + $log_offset = -s $node_B->logfile; + + # Let retention resume. As above, clear max_retention_duration before + # lifting the stall so that resumption happens promptly. + $node_B->safe_psql('dbc', + "ALTER SUBSCRIPTION $subname_BA3 SET (max_retention_duration = 0);"); + $node_A->safe_psql('postgres', + "SELECT * FROM pg_drop_replication_slot('blocker3');"); + $node_A->adjust_conf('postgresql.conf', 'synchronized_standby_slots', "''"); + $node_A->reload; + + $node_B->wait_for_log( + qr/logical replication worker for subscription "$subname_BA3" will resume retaining the information for detecting conflicts/, + $log_offset); + + # The worker is now holding at the injection point, with retention marked + # active in the catalog but its oldest_nonremovable_xid still invalid. + ok( $node_B->poll_query_until( + 'postgres', + "SELECT count(*) = 1 FROM pg_stat_activity WHERE wait_event = 'logical-rep-resume-retention-before-exit'" + ), + "dbc's apply worker is holding in the resume window"); + + # Push the transaction ID counter well past the one pinned in dbc, giving + # the launcher cycles in which it could wrongly advance the slot's xmin + # from the postgres database alone. + $next_xid = $node_B->safe_psql('postgres', "SELECT txid_current() + 1"); + + $log_offset = -s $node_B->logfile; + + # Release the worker. The launcher then relaunches it, seeding + # oldest_nonremovable_xid from the slot's xmin. + $node_B->safe_psql( + 'postgres', qq[ + SELECT injection_points_wakeup('logical-rep-resume-retention-before-exit'); + SELECT injection_points_detach('logical-rep-resume-retention-before-exit'); + ]); + + $node_B->wait_for_log( + qr/logical replication apply worker for subscription "$subname_BA3" has started/, + $log_offset); + + # The slot's xmin must still be bounded by the transaction pinned in dbc, + # and the relaunched worker must survive its first advancement cycle. + ok( $node_B->poll_query_until( + 'postgres', + "SELECT xmin::text::bigint < $next_xid FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'" + ), + "slot xmin is still bounded by the transaction pinned in dbc"); + + $logfile = slurp_file($node_B->logfile(), $log_offset); + unlike($logfile, qr/TRAP: failed Assert/, + 'no assertion failure in the apply worker'); + unlike($logfile, qr/was terminated by signal/, 'no apply worker crash'); + + $result = $node_B->safe_psql('dbc', + "SELECT subretentionactive FROM pg_subscription WHERE subname='$subname_BA3';" + ); + is($result, qq(t), 'retention is active again in dbc'); + + # Once the pinned transaction commits, the xmin must be able to advance + # again. + $dbc_session->query_until( + qr/committed/, q{ + COMMIT; + \echo committed + }); + ok($dbc_session->quit, 'close the session pinned in dbc'); + + $next_xid = $node_B->safe_psql('postgres', "SELECT txid_current() + 1"); + ok( $node_B->poll_query_until( + 'postgres', + "SELECT xmin::text::bigint >= $next_xid FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'" + ), + "slot xmin advances again after the transaction pinned in dbc commits"); + + # Clean up the third database. + $node_B->safe_psql('dbc', "DROP SUBSCRIPTION $subname_BA3"); + $node_B->safe_psql('postgres', "DROP DATABASE dbc"); +} + ############################################################################### # Check that the replication slot pg_conflict_detection is dropped after # removing all the subscriptions. -- 2.50.1 (Apple Git-155)