Improve table locking behavior in the face of current DDL.

In the previous coding, callers were faced with an awkward choice:
look up the name, do permissions checks, and then lock the table; or
look up the name, lock the table, and then do permissions checks.
The first choice was wrong because the results of the name lookup
and permissions checks might be out-of-date by the time the table
lock was acquired, while the second allowed a user with no privileges
to interfere with access to a table by users who do have privileges
(e.g. if a malicious backend queues up for an AccessExclusiveLock on
a table on which AccessShareLock is already held, further attempts
to access the table will be blocked until the AccessExclusiveLock
is obtained and the malicious backend's transaction rolls back).

To fix, allow callers of RangeVarGetRelid() to pass a callback which
gets executed after performing the name lookup but before acquiring
the relation lock.  If the name lookup is retried (because
invalidation messages are received), the callback will be re-executed
as well, so we get the best of both worlds.  RangeVarGetRelid() is
renamed to RangeVarGetRelidExtended(); callers not wishing to supply
a callback can continue to invoke it as RangeVarGetRelid(), which is
now a macro.  Since the only one caller that uses nowait = true now
passes a callback anyway, the RangeVarGetRelid() macro defaults nowait
as well.  The callback can also be used for supplemental locking - for
example, REINDEX INDEX needs to acquire the table lock before the index
lock to reduce deadlock possibilities.

There's a lot more work to be done here to fix all the cases where this
can be a problem, but this commit provides the general infrastructure
and fixes the following specific cases: REINDEX INDEX, REINDEX TABLE,
LOCK TABLE, and and DROP TABLE/INDEX/SEQUENCE/VIEW/FOREIGN TABLE.

Per discussion with Noah Misch and Alvaro Herrera.

1 files changed, 100 insertions, 72 deletions

diff --git a/src/backend/commands/lockcmds.c b/src/backend/commands/lockcmds.c
index 875f868b969..91642ce03de 100644
--- a/src/backend/commands/lockcmds.c
+++ b/src/backend/commands/lockcmds.c
@@ -23,10 +23,12 @@
 #include "storage/lmgr.h"
 #include "utils/acl.h"
 #include "utils/lsyscache.h"
+#include "utils/syscache.h"
 
-static void LockTableRecurse(Relation rel, LOCKMODE lockmode, bool nowait,
-				 bool recurse);
-
+static void LockTableRecurse(Oid reloid, LOCKMODE lockmode, bool nowait);
+static AclResult LockTableAclCheck(Oid relid, LOCKMODE lockmode);
+static void RangeVarCallbackForLockTable(const RangeVar *rv, Oid relid,
+							 Oid oldrelid, void *arg);
 
 /*
  * LOCK TABLE
@@ -51,98 +53,124 @@ LockTableCommand(LockStmt *lockstmt)
 	foreach(p, lockstmt->relations)
 	{
 		RangeVar   *rv = (RangeVar *) lfirst(p);
-		Relation	rel;
 		bool		recurse = interpretInhOption(rv->inhOpt);
 		Oid			reloid;
 
-		reloid = RangeVarGetRelid(rv, lockstmt->mode, false, lockstmt->nowait);
-		rel = relation_open(reloid, NoLock);
+		reloid = RangeVarGetRelidExtended(rv, lockstmt->mode, false,
+										  lockstmt->nowait,
+										  RangeVarCallbackForLockTable,
+										  (void *) &lockstmt->mode);
 
-		LockTableRecurse(rel, lockstmt->mode, lockstmt->nowait, recurse);
+		if (recurse)
+			LockTableRecurse(reloid, lockstmt->mode, lockstmt->nowait);
 	}
 }
 
 /*
- * Apply LOCK TABLE recursively over an inheritance tree
+ * Before acquiring a table lock on the named table, check whether we have
+ * permission to do so.
  */
 static void
-LockTableRecurse(Relation rel, LOCKMODE lockmode, bool nowait, bool recurse)
+RangeVarCallbackForLockTable(const RangeVar *rv, Oid relid, Oid oldrelid,
+							 void *arg)
 {
+	LOCKMODE	lockmode = * (LOCKMODE *) arg;
+	char		relkind;
 	AclResult	aclresult;
-	Oid			reloid = RelationGetRelid(rel);
 
-	/* Verify adequate privilege */
-	if (lockmode == AccessShareLock)
-		aclresult = pg_class_aclcheck(reloid, GetUserId(),
-									  ACL_SELECT);
-	else
-		aclresult = pg_class_aclcheck(reloid, GetUserId(),
-									  ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE);
-	if (aclresult != ACLCHECK_OK)
-		aclcheck_error(aclresult, ACL_KIND_CLASS,
-					   RelationGetRelationName(rel));
+	if (!OidIsValid(relid))
+		return;			/* doesn't exist, so no permissions check */
+	relkind = get_rel_relkind(relid);
+	if (!relkind)
+		return;			/* woops, concurrently dropped; no permissions check */
 
 	/* Currently, we only allow plain tables to be locked */
-	if (rel->rd_rel->relkind != RELKIND_RELATION)
+	if (relkind != RELKIND_RELATION)
 		ereport(ERROR,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("\"%s\" is not a table",
-						RelationGetRelationName(rel))));
+						rv->relname)));
 
-	/*
-	 * If requested, recurse to children.  We use find_inheritance_children
-	 * not find_all_inheritors to avoid taking locks far in advance of
-	 * checking privileges.  This means we'll visit multiply-inheriting
-	 * children more than once, but that's no problem.
-	 */
-	if (recurse)
+	/* Check permissions. */
+	aclresult = LockTableAclCheck(relid, lockmode);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult, ACL_KIND_CLASS, rv->relname);
+}
+
+/*
+ * Apply LOCK TABLE recursively over an inheritance tree
+ *
+ * We use find_inheritance_children not find_all_inheritors to avoid taking
+ * locks far in advance of checking privileges.  This means we'll visit
+ * multiply-inheriting children more than once, but that's no problem.
+ */
+static void
+LockTableRecurse(Oid reloid, LOCKMODE lockmode, bool nowait)
+{
+	List	   *children;
+	ListCell   *lc;
+
+	children = find_inheritance_children(reloid, NoLock);
+
+	foreach(lc, children)
 	{
-		List	   *children = find_inheritance_children(reloid, NoLock);
-		ListCell   *lc;
-		Relation	childrel;
+		Oid			childreloid = lfirst_oid(lc);
+		AclResult	aclresult;
+
+		/* Check permissions before acquiring the lock. */
+		aclresult = LockTableAclCheck(childreloid, lockmode);
+		if (aclresult != ACLCHECK_OK)
+		{
+			char	   *relname = get_rel_name(childreloid);
+			if (!relname)
+				continue;	/* child concurrently dropped, just skip it */
+			aclcheck_error(aclresult, ACL_KIND_CLASS, relname);	
+		}
 
-		foreach(lc, children)
+		/* We have enough rights to lock the relation; do so. */
+		if (!nowait)
+			LockRelationOid(childreloid, lockmode);
+		else if (!ConditionalLockRelationOid(childreloid, lockmode))
 		{
-			Oid			childreloid = lfirst_oid(lc);
-
-			/*
-			 * Acquire the lock, to protect against concurrent drops.  Note
-			 * that a lock against an already-dropped relation's OID won't
-			 * fail.
-			 */
-			if (!nowait)
-				LockRelationOid(childreloid, lockmode);
-			else if (!ConditionalLockRelationOid(childreloid, lockmode))
-			{
-				/* try to throw error by name; relation could be deleted... */
-				char	   *relname = get_rel_name(childreloid);
-
-				if (relname)
-					ereport(ERROR,
-							(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-							 errmsg("could not obtain lock on relation \"%s\"",
-								relname)));
-				else
-					ereport(ERROR,
-							(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
-					 errmsg("could not obtain lock on relation with OID %u",
-								 reloid)));
-			}
-
-			/*
-			 * Now that we have the lock, check to see if the relation really
-			 * exists or not.
-			 */
-			childrel = try_relation_open(childreloid, NoLock);
-			if (!childrel)
-			{
-				/* Release useless lock */
-				UnlockRelationOid(childreloid, lockmode);
-			}
-
-			LockTableRecurse(childrel, lockmode, nowait, recurse);
+			/* try to throw error by name; relation could be deleted... */
+			char	   *relname = get_rel_name(childreloid);
+			if (!relname)
+				continue;	/* child concurrently dropped, just skip it */
+			ereport(ERROR,
+					(errcode(ERRCODE_LOCK_NOT_AVAILABLE),
+					 errmsg("could not obtain lock on relation \"%s\"",
+						relname)));
 		}
+
+		/*
+		 * Even if we got the lock, child might have been concurrently dropped.
+		 * If so, we can skip it.
+		 */
+		if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(childreloid)))
+		{
+			/* Release useless lock */
+			UnlockRelationOid(childreloid, lockmode);
+			continue;
+		}
+
+		LockTableRecurse(childreloid, lockmode, nowait);
 	}
+}
 
-	relation_close(rel, NoLock);	/* close rel, keep lock */
+/*
+ * Check whether the current user is permitted to lock this relation.
+ */
+static AclResult
+LockTableAclCheck(Oid reloid, LOCKMODE lockmode)
+{
+	AclResult	aclresult;
+
+	/* Verify adequate privilege */
+	if (lockmode == AccessShareLock)
+		aclresult = pg_class_aclcheck(reloid, GetUserId(),
+									  ACL_SELECT);
+	else
+		aclresult = pg_class_aclcheck(reloid, GetUserId(),
+									  ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE);
+	return aclresult;
 }