From fc4a0d32f6bf41db8132555278b507d8a1e35fa2 Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Fri, 10 Jul 2026 18:03:10 +0530 Subject: [PATCH v1 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 emit the expected warning 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_warning.pl | 127 ++++++++++++++++++ 3 files changed, 133 insertions(+) create mode 100644 src/test/subscription/t/039_sequences_refresh_warning.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..79f7e5a6405 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_warning.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/039_sequences_refresh_warning.pl b/src/test/subscription/t/039_sequences_refresh_warning.pl new file mode 100644 index 00000000000..23508ef3780 --- /dev/null +++ b/src/test/subscription/t/039_sequences_refresh_warning.pl @@ -0,0 +1,127 @@ +# 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"); + +like( + $stderr, + qr/WARNING:.*sequence "public.regress_seq1" of subscription "regress_seq_sub" is already being synchronized, skipping/, + 'second REFRESH SEQUENCES warns that the sequence is already being synchronized' +); +like( + $stderr, + qr/HINT:.*Rerun ALTER SUBSCRIPTION \.\.\. REFRESH SEQUENCES 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)