From 8f542c28ac658c4a05b7b7cbe943b79d73f3e4c6 Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Tue, 22 Sep 2026 13:44:48 +0530 Subject: [PATCH v3] Skip sequences removed by a concurrent subscription refresh A sequence synchronization worker can capture a sequence before a concurrent REFRESH PUBLICATION removes it from pg_subscription_rel. Previously, the worker could update the local sequence and then fail when trying to mark the sequence as READY, aborting the batch. With disable_on_error the subscription was disabled, tables included. Check that the sequence is still part of the subscription before updating it and skip it if it has been removed. This avoids updating sequences that are no longer subscribed and prevents the failure from affecting other sequences in the batch. --- src/backend/commands/subscriptioncmds.c | 6 +++ .../replication/logical/sequencesync.c | 50 +++++++++++++++++-- 2 files changed, 53 insertions(+), 3 deletions(-) diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 22a61dca65d..36bb20a1f6e 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1341,6 +1341,12 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, RemoveSubscriptionRel(sub->oid, relid); + /* + * A sequence sync worker may already be running with this + * sequence in its to-do list. It does not have to be stopped. + * It notices that the sequence is no longer part of the + * subscription and skips it, see copy_sequence(). + */ ereport(DEBUG1, errmsg_internal("sequence \"%s.%s\" removed from subscription \"%s\"", get_namespace_name(get_rel_namespace(relid)), diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c index 6d551d45791..64c29d7b437 100644 --- a/src/backend/replication/logical/sequencesync.c +++ b/src/backend/replication/logical/sequencesync.c @@ -60,6 +60,7 @@ #include "postmaster/interrupt.h" #include "replication/logicalworker.h" #include "replication/worker_internal.h" +#include "storage/lmgr.h" #include "storage/lwlock.h" #include "utils/acl.h" #include "utils/builtins.h" @@ -80,7 +81,8 @@ typedef enum CopySeqResult COPYSEQ_MISMATCH, COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM, COPYSEQ_PUBLISHER_INSUFFICIENT_PERM, - COPYSEQ_SKIPPED + COPYSEQ_SKIPPED, + COPYSEQ_NOT_SUBSCRIBED } CopySeqResult; static List *seqinfos = NIL; @@ -403,6 +405,29 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner) AclResult aclresult; bool run_as_owner = MySubscription->runasowner; Oid seqoid = seqinfo->localrelid; + Relation rel; + + /* + * Take the subscription object lock before checking whether this sequence + * is still part of the subscription. The lock is held until the end of + * the transaction, so the check and the state update below are protected + * from a concurrent ALTER SUBSCRIPTION ... REFRESH PUBLICATION. + * + * AlterSubscription() takes this lock in AccessExclusiveLock mode while + * removing pg_subscription_rel rows, so the row cannot be removed between + * the check and the state update. + */ + LockSharedObject(SubscriptionRelationId, MySubscription->oid, 0, + AccessShareLock); + + /* + * The sequence may no longer be part of the subscription, in which case + * there is nothing to synchronize and the caller just skips it. + */ + if (!SearchSysCacheExists2(SUBSCRIPTIONRELMAP, + ObjectIdGetDatum(seqoid), + ObjectIdGetDatum(MySubscription->oid))) + return COPYSEQ_NOT_SUBSCRIBED; /* * If the user did not opt to run as the owner of the subscription @@ -434,12 +459,18 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner) if (!run_as_owner) RestoreUserContext(&ucxt); + rel = table_open(SubscriptionRelRelationId, RowExclusiveLock); + /* * Record the remote sequence's LSN in pg_subscription_rel and mark the - * sequence as READY. + * sequence as READY. Both locks it needs are held already, the object + * lock from further up and the relation lock just taken, so say so rather + * than have it take and release them again. */ UpdateSubscriptionRelState(MySubscription->oid, seqoid, SUBREL_STATE_READY, - seqinfo->page_lsn, false); + seqinfo->page_lsn, true); + + table_close(rel, NoLock); return COPYSEQ_SUCCESS; } @@ -655,6 +686,19 @@ copy_sequences(WalReceiverConn *conn) batch_skipped_count++; } break; + case COPYSEQ_NOT_SUBSCRIBED: + + /* + * A concurrent refresh removed this sequence from the + * subscription. Skipping it is the only sensible action, + * and it must not be treated as an error. + */ + ereport(LOG, + errmsg("skip synchronization of sequence \"%s.%s\" because it is no longer part of subscription \"%s\"", + seqinfo->nspname, seqinfo->seqname, + MySubscription->name)); + batch_skipped_count++; + break; } if (sequence_rel) -- 2.55.0