Merge #29291

29291: sql: resolve deadlock in waitForCacheState r=vivekmenezes a=vivekmenezes

change #25313 introduced getDatabaseID as the callback
used in waitForCacheState. However, waitForCacheState holds
on to a lock while calling its callback and can get stuck
on an intent written by another transaction. The other
transaction having written an intent can be trying to acquire
the databaseCacheHolder lock, resulting in the system getting
stuck in a deadlock.

Originally, the deadlock was thought to be cause by another
bug which got fixed through #28381, but now we know of
an actual legitimate situation in which the deadlock
can happen.

The fix is to use another callback that will fix #25313
while not going to the store.

fixes #29090

Release note: None

Co-authored-by: Vivek Menezes <vivek@cockroachlabs.com>

pkg/sql/database.go

@@ -253,6 +253,24 @@ func (dc *databaseCache) getDatabaseID(
 	return desc.ID, nil
 }
 
+// getCachedDatabaseID returns the ID of a database given its name
+// from the cache. This method never goes to the store to resolve
+// the name to id mapping. Returns 0 if the name to id mapping or
+// the database descriptor are not in the cache.
+func (dc *databaseCache) getCachedDatabaseID(ctx context.Context, name string) (sqlbase.ID, error) {
+	if id := dc.getID(name); id != 0 {
+		return id, nil
+	}
+
+	desc, err := dc.getCachedDatabaseDesc(name, false /*required*/)
+	if err != nil || desc == nil {
+		// desc can be nil if required == false and the database was not found.
+		return 0, err
+	}
+
+	return desc.ID, nil
+}
+
 // renameDatabase implements the DatabaseDescEditor interface.
 func (p *planner) renameDatabase(
 	ctx context.Context, oldDesc *sqlbase.DatabaseDescriptor, newName string,

pkg/sql/table.go

@@ -400,8 +400,7 @@ func (tc *TableCollection) waitForCacheToDropDatabases(ctx context.Context) {
 		tc.dbCacheSubscriber.waitForCacheState(
 			func(dc *databaseCache) bool {
 				// Resolve the database name from the database cache.
-				dbID, err := dc.getDatabaseID(ctx,
-					tc.leaseMgr.execCfg.DB.Txn, uc.name, false /*required*/)
+				dbID, err := dc.getCachedDatabaseID(ctx, uc.name)
 				if err != nil || dbID == 0 {
 					// dbID can still be 0 if required is false and
 					// the database is not found. Swallowing error here