From 5e62922b2bec13734222a90d9516d0ea1bea98cf Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot Date: Mon, 6 Jul 2026 04:44:13 +0000 Subject: [PATCH v6 3/4] Add invalidation-based retry loop for Alter/Drop Subscription Following the approach of RangeVarGetRelidExtended() for relations, add a retry loop that includes name resolution, ownership check, and lock acquisition in AlterSubscription() and DropSubscription(). The loop records SharedInvalidMessageCounter, resolves the subscription name to an OID, checks ownership, then locks the subscription. If the invalidation counter changed (indicating concurrent DDL), we save the current OID and retry. On the next iteration, if the name still resolves to the same OID, we're done (already holding the correct lock). If it resolves to a different OID, we release the old lock and acquire the new one. This mirrors RangeVarGetRelidExtended()'s behavior: the lock is kept across retries to avoid a window where another session could have committed concurrent DDL modifying the ownership and/or the name resolution. Author: Bertrand Drouvot Reviewed-by: Dilip Kumar Reviewed-by: Hayato Kuroda (Fujitsu) Discussion: https://postgr.es/m/akZUpiDa1UfmzYxL%40bdtpg --- src/backend/commands/subscriptioncmds.c | 202 +++++++++++++++--------- 1 file changed, 129 insertions(+), 73 deletions(-) 100.0% src/backend/commands/ diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 11ef1822701..84bf7d873b6 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -51,6 +51,7 @@ #include "replication/worker_internal.h" #include "storage/lmgr.h" #include "storage/lock.h" +#include "storage/sinval.h" #include "utils/acl.h" #include "utils/builtins.h" #include "utils/guc.h" @@ -1666,23 +1667,67 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, rel = table_open(SubscriptionRelationId, RowExclusiveLock); - /* Fetch the existing tuple. */ - tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, ObjectIdGetDatum(MyDatabaseId), - CStringGetDatum(stmt->subname)); + /* + * Lock the subscription so nobody else can do anything with it. + * + * Like RangeVarGetRelidExtended() does for relations, we resolve the + * name, check ownership, and lock inside a loop. If invalidation messages + * arrive (indicating concurrent DDL), we retry. We keep the lock held + * across retries and only release it if the name resolves to a different + * OID on the next iteration. + */ + { + Oid oldSubId = InvalidOid; + bool retry = false; - if (!HeapTupleIsValid(tup)) - ereport(ERROR, - (errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("subscription \"%s\" does not exist", - stmt->subname))); + for (;;) + { + uint64 inval_count = SharedInvalidMessageCounter; - form = (Form_pg_subscription) GETSTRUCT(tup); - subid = form->oid; + tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, + ObjectIdGetDatum(MyDatabaseId), + CStringGetDatum(stmt->subname)); - /* must be owner */ - if (!object_ownercheck(SubscriptionRelationId, subid, GetUserId())) - aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION, - stmt->subname); + if (!HeapTupleIsValid(tup)) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("subscription \"%s\" does not exist", + stmt->subname))); + + form = (Form_pg_subscription) GETSTRUCT(tup); + subid = form->oid; + + if (!object_ownercheck(SubscriptionRelationId, subid, + GetUserId())) + aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION, + stmt->subname); + + /* + * If upon retry we get the same OID, the invalidation messages + * did not change the final answer. So we're done. If we got a + * different OID, unlock the old one and lock the new one below. + */ + if (retry) + { + if (subid == oldSubId) + break; + UnlockSharedObject(SubscriptionRelationId, oldSubId, 0, + AccessExclusiveLock); + } + + LockSharedObject(SubscriptionRelationId, subid, 0, + AccessExclusiveLock); + + /* If no invalidation messages, we're done. */ + if (inval_count == SharedInvalidMessageCounter) + break; + + /* Something may have changed, retry. */ + retry = true; + oldSubId = subid; + heap_freetuple(tup); + } + } /* parse and check options */ switch (stmt->kind) @@ -1730,30 +1775,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, if (supported_opts > 0) parse_subscription_options(pstate, stmt->options, supported_opts, &opts); - heap_freetuple(tup); - - /* Lock the subscription so nobody else can do anything with it. */ - LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock); - - /* - * Re-read the subscription tuple after acquiring the lock. A concurrent - * DROP or ALTER may have committed before we acquired the lock. - */ - tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid)); - - if (!HeapTupleIsValid(tup)) - ereport(ERROR, - (errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("subscription \"%s\" does not exist", - stmt->subname))); - - form = (Form_pg_subscription) GETSTRUCT(tup); - - /* must still be owner */ - if (!object_ownercheck(SubscriptionRelationId, subid, GetUserId())) - aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION, - stmt->subname); - /* * Determine in advance whether we need the original conninfo or not, so * that errors are generated consistently in cases where we do need it; @@ -2659,56 +2680,91 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) return; } - form = (Form_pg_subscription) GETSTRUCT(tup); - subid = form->oid; - - /* must be owner */ - if (!object_ownercheck(SubscriptionRelationId, subid, GetUserId())) - aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION, - stmt->subname); - ReleaseSysCache(tup); /* * Lock the subscription so nobody else can do anything with it (including * the replication workers). + * + * Like RangeVarGetRelidExtended() does for relations, we resolve the + * name, check ownership, and lock inside a loop. If invalidation messages + * arrive (indicating concurrent DDL), we retry. We keep the lock held + * across retries and only release it if the name resolves to a different + * OID on the next iteration. */ - LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock); + { + Oid oldSubId = InvalidOid; + bool retry = false; - /* - * Re-read the subscription tuple after acquiring the lock. A concurrent - * ALTER or DROP may have committed before we acquired the lock. - */ - tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid)); + for (;;) + { + uint64 inval_count = SharedInvalidMessageCounter; - if (!HeapTupleIsValid(tup)) - { - UnlockSharedObject(SubscriptionRelationId, subid, 0, - AccessExclusiveLock); - table_close(rel, NoLock); + tup = SearchSysCache2(SUBSCRIPTIONNAME, + ObjectIdGetDatum(MyDatabaseId), + CStringGetDatum(stmt->subname)); - if (!stmt->missing_ok) - ereport(ERROR, - (errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("subscription \"%s\" does not exist", - stmt->subname))); - else - ereport(NOTICE, - (errmsg("subscription \"%s\" does not exist, skipping", - stmt->subname))); + if (!HeapTupleIsValid(tup)) + { + if (retry) + UnlockSharedObject(SubscriptionRelationId, oldSubId, 0, + AccessExclusiveLock); + table_close(rel, NoLock); - return; - } + if (!stmt->missing_ok) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("subscription \"%s\" does not exist", + stmt->subname))); + else + ereport(NOTICE, + (errmsg("subscription \"%s\" does not exist, skipping", + stmt->subname))); - /* must still be owner */ - if (!object_ownercheck(SubscriptionRelationId, subid, GetUserId())) - aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION, - stmt->subname); + return; + } + + form = (Form_pg_subscription) GETSTRUCT(tup); + subid = form->oid; + + if (!object_ownercheck(SubscriptionRelationId, subid, + GetUserId())) + { + ReleaseSysCache(tup); + aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION, + stmt->subname); + } + + /* + * If upon retry we get the same OID, the invalidation messages + * did not change the final answer. So we're done. If we got a + * different OID, unlock the old one and lock the new one below. + */ + if (retry) + { + if (subid == oldSubId) + break; + UnlockSharedObject(SubscriptionRelationId, oldSubId, 0, + AccessExclusiveLock); + } + + LockSharedObject(SubscriptionRelationId, subid, 0, + AccessExclusiveLock); + + /* If no invalidation messages, we're done. */ + if (inval_count == SharedInvalidMessageCounter) + break; + + /* Something may have changed, retry. */ + retry = true; + oldSubId = subid; + ReleaseSysCache(tup); + } + } /* DROP hook for the subscription being removed */ InvokeObjectDropHook(SubscriptionRelationId, subid, 0); - form = (Form_pg_subscription) GETSTRUCT(tup); subowner = form->subowner; subserver = form->subserver; subconflictlogrelid = form->subconflictlogrelid; -- 2.34.1