Make pg_replication_origen_drop safe against concurrent drops.

Amit Kapila · Amit Kapila · commit cd142e032ebd · 2021-02-10T07:17:09.000+05:30
Currently, we get the origen id from the name and then drop the origen by taking ExclusiveLock on ReplicationOriginRelationId. So, two concurrent sessions can get the id from the name at the same time and then when they try to drop the origen, one of the sessions will get the either "tuple concurrently deleted" or "cache lookup failed for replication origen ..". To prevent this race condition we do the entire operation under lock. This obviates the need for replorigen_drop() API and we have removed it so if any extension authors are using it they need to instead use replorigen_drop_by_name. See it's usage in pg_replication_origen_drop(). Author: Peter Smith Reviewed-by: Amit Kapila, Euler Taveira, Petr Jelinek, and Alvaro Herrera Discussion: https://www.postgresql.org/message-id/CAHut%2BPuW8DWV5fskkMWWMqzt-x7RPcNQOtJQBp6SdwyRghCk7A%40mail.gmail.com
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
@@ -926,7 +926,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	ListCell   *lc;
 	char		origenname[NAMEDATALEN];
 	char	   *err = NULL;
-	RepOriginId origenid;
 	WalReceiverConn *wrconn = NULL;
 	StringInfoData cmd;
 	Form_pg_subscription form;
@@ -1050,9 +1049,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 
 	/* Remove the origen tracking if exists. */
 	snprintf(origenname, sizeof(origenname), "pg_%u", subid);
-	origenid = replorigen_by_name(origenname, true);
-	if (origenid != InvalidRepOriginId)
-		replorigen_drop(origenid, false);
+	replorigen_drop_by_name(origenname, true, false);
 
 	/*
 	 * If there is no slot associated with the subscription, we can finish
diff --git a/src/backend/replication/logical/origen.c b/src/backend/replication/logical/origen.c
@@ -322,27 +322,15 @@ replorigen_create(char *roname)
 	return roident;
 }
 
-
 /*
- * Drop replication origen.
- *
- * Needs to be called in a transaction.
+ * Helper function to drop a replication origen.
  */
-void
-replorigen_drop(RepOriginId roident, bool nowait)
+static void
+replorigen_drop_guts(Relation rel, RepOriginId roident, bool nowait)
 {
 	HeapTuple	tuple;
-	Relation	rel;
 	int			i;
 
-	Assert(IsTransactionState());
-
-	/*
-	 * To interlock against concurrent drops, we hold ExclusiveLock on
-	 * pg_replication_origen throughout this function.
-	 */
-	rel = table_open(ReplicationOriginRelationId, ExclusiveLock);
-
 	/*
 	 * First, clean up the slot state info, if there is any matching slot.
 	 */
@@ -415,11 +403,40 @@ replorigen_drop(RepOriginId roident, bool nowait)
 	ReleaseSysCache(tuple);
 
 	CommandCounterIncrement();
-
-	/* now release lock again */
-	table_close(rel, ExclusiveLock);
 }
 
+/*
+ * Drop replication origen (by name).
+ *
+ * Needs to be called in a transaction.
+ */
+void
+replorigen_drop_by_name(char *name, bool missing_ok, bool nowait)
+{
+	RepOriginId roident;
+	Relation	rel;
+
+	Assert(IsTransactionState());
+
+	/*
+	 * To interlock against concurrent drops, we hold ExclusiveLock on
+	 * pg_replication_origen till xact commit.
+	 *
+	 * XXX We can optimize this by acquiring the lock on a specific origen by
+	 * using LockSharedObject if required. However, for that, we first to
+	 * acquire a lock on ReplicationOriginRelationId, get the origen_id, lock
+	 * the specific origen and then re-check if the origen still exists.
+	 */
+	rel = table_open(ReplicationOriginRelationId, ExclusiveLock);
+
+	roident = replorigen_by_name(name, missing_ok);
+
+	if (OidIsValid(roident))
+		replorigen_drop_guts(rel, roident, nowait);
+
+	/* We keep the lock on pg_replication_origen until commit */
+	table_close(rel, NoLock);
+}
 
 /*
  * Lookup replication origen via its oid and return the name.
@@ -1256,16 +1273,12 @@ Datum
 pg_replication_origen_drop(PG_FUNCTION_ARGS)
 {
 	char	   *name;
-	RepOriginId roident;
 
 	replorigen_check_prerequisites(false, false);
 
 	name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
 
-	roident = replorigen_by_name(name, false);
-	Assert(OidIsValid(roident));
-
-	replorigen_drop(roident, true);
+	replorigen_drop_by_name(name, false, true);
 
 	pfree(name);
 
diff --git a/src/include/replication/origen.h b/src/include/replication/origen.h
@@ -40,7 +40,7 @@ extern PGDLLIMPORT TimestampTz replorigen_session_origen_timestamp;
 /* API for querying & manipulating replication origens */
 extern RepOriginId replorigen_by_name(char *name, bool missing_ok);
 extern RepOriginId replorigen_create(char *name);
-extern void replorigen_drop(RepOriginId roident, bool nowait);
+extern void replorigen_drop_by_name(char *name, bool missing_ok, bool nowait);
 extern bool replorigen_by_oid(RepOriginId roident, bool missing_ok,
 							  char **roname);
 
