From 42ce4354f51a856c9151b475866683aae6ace9c3 Mon Sep 17 00:00:00 2001 From: Ayush Tiwari Date: Tue, 29 Sep 2026 11:18:20 +0530 Subject: [PATCH v3 2/2] Test table sync after a concurrent refresh Add an injection point just before the apply worker locks the subscription to mark a synchronized table READY. Use it to remove the table with ALTER SUBSCRIPTION ... REFRESH PUBLICATION while the apply worker waits, and check that the subscription stays enabled with disable_on_error. --- src/backend/replication/logical/tablesync.c | 2 + src/test/subscription/meson.build | 1 + .../subscription/t/039_tablesync_refresh.pl | 86 +++++++++++++++++++ 3 files changed, 89 insertions(+) create mode 100644 src/test/subscription/t/039_tablesync_refresh.pl diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 6c211ec0e62..77a4e13a94d 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -117,6 +117,7 @@ #include "utils/acl.h" #include "utils/array.h" #include "utils/builtins.h" +#include "utils/injection_point.h" #include "utils/lsyscache.h" #include "utils/rls.h" #include "utils/snapmgr.h" @@ -451,6 +452,7 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) * are doing during DDL commands to avoid deadlocks. See * AlterSubscription_refresh. */ + INJECTION_POINT("tablesync-before-mark-ready", NULL); LockSharedObject(SubscriptionRelationId, MyLogicalRepWorker->subid, 0, AccessShareLock); diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index e71e95c6297..32b2a5b3eef 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -48,6 +48,7 @@ tests += { 't/036_sequences.pl', 't/037_except.pl', 't/038_walsnd_shutdown_timeout.pl', + 't/039_tablesync_refresh.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/039_tablesync_refresh.pl b/src/test/subscription/t/039_tablesync_refresh.pl new file mode 100644 index 00000000000..8c2331b5de4 --- /dev/null +++ b/src/test/subscription/t/039_tablesync_refresh.pl @@ -0,0 +1,86 @@ + +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test that ALTER SUBSCRIPTION ... REFRESH PUBLICATION can remove a table +# while the apply worker is about to mark it READY. +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +if ($ENV{enable_injection_points} ne 'yes') +{ + plan skip_all => 'Injection points not supported by this build'; +} + +my $publisher = PostgreSQL::Test::Cluster->new('publisher'); +$publisher->init(allows_streaming => 'logical'); +$publisher->start; + +my $subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$subscriber->init; +$subscriber->start; +$subscriber->safe_psql('postgres', 'CREATE EXTENSION injection_points'); + +# tab_other stays in the publication throughout; tab_sync is added and +# then removed again. +foreach my $node ($publisher, $subscriber) +{ + $node->safe_psql('postgres', + 'CREATE TABLE tab_other (a int); CREATE TABLE tab_sync (a int);'); +} +$publisher->safe_psql('postgres', + 'CREATE PUBLICATION pub FOR TABLE tab_other'); + +# With copy_data = false, tab_other starts out READY, so only tab_sync goes +# through table synchronization below. With disable_on_error, an error in +# the apply worker disables the whole subscription. +my $connstr = $publisher->connstr . ' dbname=postgres'; +$subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION sub CONNECTION '$connstr' PUBLICATION pub WITH (copy_data = false, disable_on_error = true)" +); + +# Make the apply worker wait just before it marks a table READY. +$subscriber->safe_psql('postgres', + "SELECT injection_points_attach('tablesync-before-mark-ready', 'wait')"); + +# Add tab_sync and wait for its initial sync to finish (SYNCDONE). +$publisher->safe_psql('postgres', 'ALTER PUBLICATION pub ADD TABLE tab_sync'); +$subscriber->safe_psql('postgres', + 'ALTER SUBSCRIPTION sub REFRESH PUBLICATION'); +$subscriber->poll_query_until('postgres', + "SELECT srsubstate = 's' FROM pg_subscription_rel WHERE srrelid = 'tab_sync'::regclass" +) or die "timed out waiting for tab_sync to reach SYNCDONE"; + +# Replicate a change, so the apply worker moves past the sync position and +# stops at the injection point. +$publisher->safe_psql('postgres', 'INSERT INTO tab_other VALUES (1)'); +$subscriber->wait_for_event('logical replication apply worker', + 'tablesync-before-mark-ready'); + +# While the apply worker waits, remove tab_sync from the subscription. +$publisher->safe_psql('postgres', + 'ALTER PUBLICATION pub DROP TABLE tab_sync'); +$subscriber->safe_psql('postgres', + 'ALTER SUBSCRIPTION sub REFRESH PUBLICATION'); + +# Let the apply worker continue. It must skip the removed table rather +# than fail, since an error would disable the subscription. +$subscriber->safe_psql( + 'postgres', q{ +SELECT injection_points_detach('tablesync-before-mark-ready'); +SELECT injection_points_wakeup('tablesync-before-mark-ready'); +}); + +# Replication of tab_other should carry on. +$publisher->safe_psql('postgres', 'INSERT INTO tab_other VALUES (2)'); +$subscriber->poll_query_until('postgres', + "SELECT (SELECT count(*) FROM tab_other) = 2 OR NOT subenabled FROM pg_subscription" +) or die "timed out waiting for the apply worker"; +is( $subscriber->safe_psql( + 'postgres', 'SELECT subenabled FROM pg_subscription'), + 't', + 'subscription is still enabled'); + +done_testing(); -- 2.34.1