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..edd9b24ed3f 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)