diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 9467f58a23d..9f2f1abb586 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1575,7 +1575,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) * Lock pg_subscription with AccessExclusiveLock to ensure that the * launcher doesn't restart new worker during dropping the subscription */ - rel = table_open(SubscriptionRelationId, AccessExclusiveLock); + rel = table_open(SubscriptionRelationId, AccessShareLock); tup = SearchSysCache2(SUBSCRIPTIONNAME, MyDatabaseId, CStringGetDatum(stmt->subname)); diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 294d0d74d8c..bdce4467553 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -35,6 +35,7 @@ #include "replication/walreceiver.h" #include "replication/worker_internal.h" #include "storage/ipc.h" +#include "storage/lmgr.h" #include "storage/proc.h" #include "storage/procarray.h" #include "tcop/tcopprot.h" @@ -171,6 +172,43 @@ get_subscription_list(void) return res; } +/* + * Check subscription exists or not. + */ +static bool +is_subscription_exist(Oid subid) +{ + Relation rel; + TableScanDesc scan; + HeapTuple tup; + bool found = false; + + /* Start a transaction, for detail see comments in get_subscription_list */ + StartTransactionCommand(); + (void) GetTransactionSnapshot(); + + rel = table_open(SubscriptionRelationId, AccessShareLock); + scan = table_beginscan_catalog(rel, 0, NULL); + + while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection))) + { + Form_pg_subscription subform = (Form_pg_subscription) GETSTRUCT(tup); + + if (subform->oid == subid) + { + found = true; + break; + } + } + + table_endscan(scan); + table_close(rel, AccessShareLock); + + CommitTransactionCommand(); + + return found; +} + /* * Wait for a background worker to start up and attach to the shmem context. * @@ -1209,6 +1247,19 @@ ApplyLauncherMain(Datum main_arg) if (last_start == 0 || (elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval) { + /* + * Lock the subscription to prevent it from being concurrently dropped, + * then re-verify its existence. + */ + LockSharedObject(SubscriptionRelationId, sub->oid, 0, + AccessShareLock); + if (!is_subscription_exist(sub->oid)) + { + UnlockSharedObject(SubscriptionRelationId, sub->oid, 0, + AccessShareLock); + continue; + } + ApplyLauncherSetWorkerStartTime(sub->oid, now); if (!logicalrep_worker_launch(WORKERTYPE_APPLY, sub->dbid, sub->oid, sub->name, @@ -1225,6 +1276,8 @@ ApplyLauncherMain(Datum main_arg) wait_time = Min(wait_time, wal_retrieve_retry_interval); } + UnlockSharedObject(SubscriptionRelationId, sub->oid, 0, + AccessShareLock); } else {