From f73d6f46d2b98bb33b3a28399df9a2d64658d008 Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Tue, 22 Sep 2026 13:44:48 +0530 Subject: [PATCH v2] 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. 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 | 56 ++++++++++++++++++- 2 files changed, 59 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..1a0cd5852ff 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,36 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner) AclResult aclresult; bool run_as_owner = MySubscription->runasowner; Oid seqoid = seqinfo->localrelid; + XLogRecPtr statelsn; + Relation rel; + + /* + * Acquire the locks that UpdateSubscriptionRelState() requires before + * checking whether this sequence is still part of the subscription, and + * hold them until the state has been updated below. + * + * ALTER SUBSCRIPTION ... REFRESH PUBLICATION can remove the sequence's + * pg_subscription_rel row while we are synchronizing it. It holds both of + * these locks in exclusive mode until it commits, see AlterSubscription() + * and AlterSubscription_refresh(). Holding them here means the row cannot + * disappear between the check below and the update at the end of this + * function. + */ + LockSharedObject(SubscriptionRelationId, MySubscription->oid, 0, + AccessShareLock); + rel = table_open(SubscriptionRelRelationId, RowExclusiveLock); + + /* + * The sequence may no longer be part of the subscription. There is + * nothing left to synchronize, so leave the local sequence alone and let + * the caller skip it. + */ + if (GetSubscriptionRelState(MySubscription->oid, seqoid, + &statelsn) == SUBREL_STATE_UNKNOWN) + { + table_close(rel, NoLock); + return COPYSEQ_NOT_SUBSCRIBED; + } /* * If the user did not opt to run as the owner of the subscription @@ -418,6 +450,8 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner) if (!run_as_owner) RestoreUserContext(&ucxt); + table_close(rel, NoLock); + return COPYSEQ_SUBSCRIBER_INSUFFICIENT_PERM; } @@ -436,10 +470,13 @@ copy_sequence(LogicalRepSequenceInfo *seqinfo, Oid seqowner) /* * Record the remote sequence's LSN in pg_subscription_rel and mark the - * sequence as READY. + * sequence as READY. The locks taken above are the ones this needs, so + * tell it they are already held. */ UpdateSubscriptionRelState(MySubscription->oid, seqoid, SUBREL_STATE_READY, - seqinfo->page_lsn, false); + seqinfo->page_lsn, true); + + table_close(rel, NoLock); return COPYSEQ_SUCCESS; } @@ -655,6 +692,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