From b7ce39fc7248b39bd482ed610928658e59a3a1d4 Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Fri, 10 Jul 2026 18:03:10 +0530 Subject: [PATCH 2/2] Add a test for concurrent REFRESH SEQUENCES Add a TAP test covering the case where 'ALTER SUBSCRIPTION ... REFRESH SEQUENCES' is executed while a sequence synchronization is already in progress. The test uses an injection point to pause the sequence synchronization worker immediately before it applies the fetched sequence value to the subscriber. While the worker is paused, a second 'ALTER SUBSCRIPTION ... REFRESH SEQUENCES' command is issued and verified to thrown an error indicating that the sequence is already being synchronized. This test exercises the race condition and verifies that concurrent refresh requests are handled safely without resetting the synchronization state of sequences that are already being processed. --- .../replication/logical/sequencesync.c | 5 + src/test/subscription/meson.build | 1 + .../t/039_sequences_refresh_error.pl | 129 ++++++++++++++++++ 3 files changed, 135 insertions(+) create mode 100644 src/test/subscription/t/039_sequences_refresh_error.pl diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c index 770fa5de10b..631b9c4bf88 100644 --- a/src/backend/replication/logical/sequencesync.c +++ b/src/backend/replication/logical/sequencesync.c @@ -68,6 +68,7 @@ #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" +#include "utils/injection_point.h" #include "utils/pg_lsn.h" #include "utils/syscache.h" #include "utils/usercontext.h" @@ -552,8 +553,12 @@ copy_sequences(WalReceiverConn *conn) sync_status = get_and_validate_seq_info(slot, &sequence_rel, &seqinfo, &seqidx); if (sync_status == COPYSEQ_SUCCESS) + { + INJECTION_POINT("sequencesync-before-copy-sequence", NULL); + sync_status = copy_sequence(seqinfo, sequence_rel->rd_rel->relowner); + } switch (sync_status) { diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index e71e95c6297..42a6706e531 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_sequences_refresh_error.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/039_sequences_refresh_error.pl b/src/test/subscription/t/039_sequences_refresh_error.pl new file mode 100644 index 00000000000..21011786f93 --- /dev/null +++ b/src/test/subscription/t/039_sequences_refresh_error.pl @@ -0,0 +1,129 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test that ALTER SUBSCRIPTION ... REFRESH SEQUENCES emits a WARNING and +# skips resetting a sequence that is already in INIT state, i.e. one whose +# previous REFRESH SEQUENCES has not yet been picked up by the +# sequencesync worker. +# +# This is exercised deterministically using an injection point placed just +# before copy_sequence() is called in the sequencesync worker's main loop: +# +# 1. Run a first REFRESH SEQUENCES, which puts the sequence into INIT state +# and lets the sequencesync worker start. +# 2. The worker blocks at the injection point after fetching the sequence's +# remote values but before applying them / marking the row READY. +# 3. Run a second REFRESH SEQUENCES while the worker is still blocked. Since +# the sequence is still in INIT state, the command must emit a WARNING +# (with a HINT to retry later) and must not disturb the row. +# 4. Release the injection point and let the worker finish; the sequence +# should reach READY normally. + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init; +$node_subscriber->start; + +# This test depends on an injection point to block the sequencesync worker +# right before it applies a sequence's fetched values. +my $injection_points_supported = + $node_subscriber->check_extension('injection_points'); + +if (!$injection_points_supported) +{ + plan skip_all => 'Server does not support injection points.'; +} + +$node_subscriber->append_conf('postgresql.conf', + "shared_preload_libraries = 'injection_points'"); +$node_subscriber->restart; + +$node_subscriber->safe_psql('postgres', "CREATE EXTENSION injection_points;"); + +# Identical sequence definitions on both nodes. +my $ddl = qq(CREATE SEQUENCE regress_seq1;); +$node_publisher->safe_psql('postgres', $ddl); +$node_subscriber->safe_psql('postgres', $ddl); + +$node_publisher->safe_psql('postgres', + "CREATE PUBLICATION regress_seq_pub FOR ALL SEQUENCES"); + +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION regress_seq_sub CONNECTION '$publisher_connstr' PUBLICATION regress_seq_pub" +); + +# Wait for the initial sync to complete before testing REFRESH SEQUENCES. +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_subscription_rel WHERE srsubstate <> 'r'") + or die "timed out waiting for initial sequence sync"; + +# Attach the injection point that will block the sequencesync worker just +# before it applies a sequence's fetched values. +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('sequencesync-before-copy-sequence', 'wait');" +); + +# First REFRESH SEQUENCES: puts the sequence into INIT and starts the +# sequencesync worker. +$node_subscriber->safe_psql('postgres', + "ALTER SUBSCRIPTION regress_seq_sub REFRESH SEQUENCES"); + +# Wait until the worker is blocked at the injection point, i.e. after +# fetching the remote values but before applying them. +$node_subscriber->wait_for_event('logical replication sequencesync worker', + 'sequencesync-before-copy-sequence'); + +my $result = $node_subscriber->safe_psql('postgres', + "SELECT srsubstate FROM pg_subscription_rel WHERE srrelid = 'regress_seq1'::regclass" +); +is($result, 'i', + 'sequence is still in INIT state while the worker is blocked'); + +# Second REFRESH SEQUENCES while the sequence is still in INIT: it must +# not disturb the row, and must emit a WARNING with a HINT instead. +my ($cmdret, $stdout, $stderr) = $node_subscriber->psql('postgres', + "ALTER SUBSCRIPTION regress_seq_sub REFRESH SEQUENCES"); + +isnt($cmdret, 0, + 'second REFRESH SEQUENCES fails while a sequencesync worker is running' +); +like( + $stderr, + qr/ERROR:.*cannot execute ALTER SUBSCRIPTION \.\.\. REFRESH SEQUENCES while a sequence synchronization worker for subscription "regress_seq_sub" is still running/, + 'second REFRESH SEQUENCES reports the sequencesync worker is still running' +); +like( + $stderr, + qr/HINT:.*Try again after the current synchronization completes\./, + 'second REFRESH SEQUENCES hints to retry later'); + +$result = $node_subscriber->safe_psql('postgres', + "SELECT srsubstate FROM pg_subscription_rel WHERE srrelid = 'regress_seq1'::regclass" +); +is($result, 'i', + 'sequence remains in INIT state after the second REFRESH SEQUENCES'); + +# Release the worker and let it finish normally. +$node_subscriber->safe_psql('postgres', + "SELECT injection_points_wakeup('sequencesync-before-copy-sequence'); + SELECT injection_points_detach('sequencesync-before-copy-sequence');" +); + +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_subscription_rel WHERE srsubstate <> 'r'") + or die "timed out waiting for sequence to reach READY state"; + +$result = $node_subscriber->safe_psql('postgres', + "SELECT last_value, is_called FROM regress_seq1"); +is($result, '1|f', 'sequence reaches READY with expected values'); + +done_testing(); -- 2.50.1 (Apple Git-155)