From c9ceb242e3e3cb8913c0778fcea45b2a9a48deb8 Mon Sep 17 00:00:00 2001 From: Amit Kapila Date: Fri, 17 Jul 2026 15:07:56 +0530 Subject: [PATCH v1] Handle concurrent sequence refreshes. 'ALTER SUBSCRIPTION ... REFRESH SEQUENCES' can race with a running sequence synchronization worker: if the worker has fetched a sequence's value from the publisher but not yet marked it READY, a concurrent refresh that resets the sequence to INIT can be overwritten by the worker's stale value, silently losing the refresh request. Handle this by stopping any running sequence sync worker and then resetting the sequences to INIT. To keep it race-free, lock pg_subscription_rel in AccessExclusiveLock mode across the stop and the reset: this blocks the apply worker from re-launching a worker and any re-launched worker from proceeding, since both read pg_subscription_rel in AccessShareLock mode. Stopping the worker is also required, because a worker merely blocked on the lock would otherwise resume after commit and overwrite the reset with the stale value it had already fetched. --- src/backend/commands/subscriptioncmds.c | 88 ++++++++++++------------ src/test/subscription/t/036_sequences.pl | 4 -- 2 files changed, 44 insertions(+), 48 deletions(-) diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 3de358f9374..78c6a22e493 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1362,33 +1362,8 @@ AlterSubscription_refresh_seq(Subscription *sub) char *err = NULL; WalReceiverConn *wrconn; bool must_use_password; - - /* - * Disallow a concurrent REFRESH SEQUENCES while a sequence sync worker - * for this subscription is still running. This avoids a race where the - * publisher's sequence advances after the current worker has fetched its - * value but before it marks the sequence READY. A user may then issue - * another REFRESH SEQUENCES to synchronize the updated value. Since the - * affected sequences are already in the INIT state, the running worker - * has no indication that a new synchronization has been requested. It - * would then apply the stale value it already fetched and mark the - * sequence READY, causing the new synchronization request to be lost and - * preventing the updated publisher values from being synchronized. - */ - LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - if (logicalrep_worker_find(WORKERTYPE_SEQUENCESYNC, sub->oid, InvalidOid, - true)) - { - LWLockRelease(LogicalRepWorkerLock); - ereport(ERROR, - errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - /* translator: %s is an SQL ALTER command */ - errmsg("cannot execute %s while a sequence synchronization worker is running", - "ALTER SUBSCRIPTION ... REFRESH SEQUENCES"), - errhint("Try again after the current synchronization completes.")); - } - - LWLockRelease(LogicalRepWorkerLock); + Relation rel; + List *subrel_states; /* Load the library providing us libpq calls. */ load_file("libpqwalreceiver", false); @@ -1403,33 +1378,58 @@ AlterSubscription_refresh_seq(Subscription *sub) errmsg("subscription \"%s\" could not connect to the publisher: %s", sub->name, err)); + /* + * Validate the publisher side while holding no lock on pg_subscription_rel, + * so the network round-trips do not stall other subscriptions' + * synchronization bookkeeping. + */ PG_TRY(); { - List *subrel_states; - check_publications_origin_sequences(wrconn, sub->publications, true, sub->origin, NULL, 0, sub->name); - - /* Get local sequence list. */ - subrel_states = GetSubscriptionRelations(sub->oid, false, true, false); - foreach_ptr(SubscriptionRelState, subrel, subrel_states) - { - Oid relid = subrel->relid; - - UpdateSubscriptionRelState(sub->oid, relid, SUBREL_STATE_INIT, - InvalidXLogRecPtr, false); - ereport(DEBUG1, - errmsg_internal("sequence \"%s.%s\" of subscription \"%s\" set to INIT state", - get_namespace_name(get_rel_namespace(relid)), - get_rel_name(relid), - sub->name)); - } } PG_FINALLY(); { walrcv_disconnect(wrconn); } PG_END_TRY(); + + /* + * Reset the sequences to INIT so they get re-synchronized with the latest + * publisher values. + * + * Lock pg_subscription_rel with AccessExclusiveLock to prevent a race with + * the apply worker re-launching a sequence sync worker while we reset the + * states. Even if a new worker starts, it can't progress: it opens + * pg_subscription_rel in AccessShareLock mode in LogicalRepSyncSequences() + * and will block until we commit, by which time the sequences are INIT and + * it will sync the latest values. + * + * The lock alone is not enough, though: a worker already running may have + * fetched a sequence's value but not yet marked it READY, and would resume + * after we commit and overwrite our reset with that stale value. So we also + * stop any running sequence sync worker while holding the lock. + */ + rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock); + + logicalrep_worker_stop(WORKERTYPE_SEQUENCESYNC, sub->oid, InvalidOid); + + /* Reset every local sequence of this subscription to INIT. */ + subrel_states = GetSubscriptionRelations(sub->oid, false, true, false); + foreach_ptr(SubscriptionRelState, subrel, subrel_states) + { + Oid relid = subrel->relid; + + UpdateSubscriptionRelState(sub->oid, relid, SUBREL_STATE_INIT, + InvalidXLogRecPtr, false); + ereport(DEBUG1, + errmsg_internal("sequence \"%s.%s\" of subscription \"%s\" set to INIT state", + get_namespace_name(get_rel_namespace(relid)), + get_rel_name(relid), + sub->name)); + } + + table_close(rel, NoLock); } /* diff --git a/src/test/subscription/t/036_sequences.pl b/src/test/subscription/t/036_sequences.pl index b3b3b20f82b..77ac9386cd8 100644 --- a/src/test/subscription/t/036_sequences.pl +++ b/src/test/subscription/t/036_sequences.pl @@ -286,10 +286,6 @@ $node_publisher->safe_psql( CREATE SEQUENCE regress_s4 START 10 INCREMENT 2; )); -# Wait for the missing sequence added to be synced -$node_subscriber->poll_query_until('postgres', $synced_query) - or die "Timed out while waiting for subscriber to synchronize data"; - ########## # Ensure that insufficient privileges on the publisher for a sequence # are reported correctly as a permission issue, not as a missing sequence. -- 2.54.0