Fix handling of shared statistics with dropped databases

Dropping a database while a connection is attempted on it was able to
lead to the presence of valid database entries in shared statistics.
The issue is that MyDatabaseId was getting set too early than it should,
as, if the connection attempted on the dropped database fails when
renamed or dropped, the shutdown callback of the shared statistics would
finish by re-inserting a correct entry related to the database already
dropped.

As analyzed by the bug reporters, this issue could lead to phantom
entries in the database list maintained by the autovacuum launcher
(in rebuild_database_list()) if the database dropped was part of the
database list when it was still valid.  After the database was dropped,
it would remain the highest on the list of databases to considered by
the autovacuum worker as things to process.  This would prevent
autovacuum jobs to happen on all the other databases still present.

The commit fixes this issue by delaying setting MyDatabaseId until the
database existence has been re-checked with the second scan on
pg_database after getting a shared lock on it, and by switching
pgstat_update_dbstats() so as nothing happens if MyDatabaseId is not
valid.

Issue introduced by 5891c7a8ed, so backpatch down to 15.

Reported-by: Will Mortensen, Jacob Speidel
Analyzed-by: Will Mortensen, Jacob Speidel
Author: Andres Freund
Discussion: https://postgr.es/m/17973-bca1f7d5c14f601e@postgresql.org
Backpatch-through: 15

src/backend/utils/activity/pgstat_database.c

@@ -271,6 +271,13 @@ pgstat_update_dbstats(TimestampTz ts)
 {
 	PgStat_StatDBEntry *dbentry;
 
+	/*
+	 * If not connected to a database yet, don't attribute time to "shared
+	 * state" (InvalidOid is used to track stats for shared relations, etc.).
+	 */
+	if (!OidIsValid(MyDatabaseId))
+		return;
+
 	dbentry = pgstat_prep_database_pending(MyDatabaseId);
 
 	/*
@@ -327,6 +334,12 @@ pgstat_prep_database_pending(Oid dboid)
 {
 	PgStat_EntryRef *entry_ref;
 
+	/*
+	 * This should not report stats on database objects before having
+	 * connected to a database.
+	 */
+	Assert(!OidIsValid(dboid) || OidIsValid(MyDatabaseId));
+
 	entry_ref = pgstat_prep_pending_entry(PGSTAT_KIND_DATABASE, dboid, InvalidOid,
 										  NULL);
 

src/backend/utils/init/postinit.c

@@ -1006,7 +1006,7 @@ InitPostgres(const char *in_dbname, Oid dboid,
 	 */
 	if (bootstrap)
 	{
-		MyDatabaseId = Template1DbOid;
+		dboid = Template1DbOid;
 		MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;
 	}
 	else if (in_dbname != NULL)
@@ -1020,32 +1020,9 @@ InitPostgres(const char *in_dbname, Oid dboid,
 					(errcode(ERRCODE_UNDEFINED_DATABASE),
 					 errmsg("database \"%s\" does not exist", in_dbname)));
 		dbform = (Form_pg_database) GETSTRUCT(tuple);
-		MyDatabaseId = dbform->oid;
+		dboid = dbform->oid;
-		MyDatabaseTableSpace = dbform->dattablespace;
-		/* take database name from the caller, just for paranoia */
-		strlcpy(dbname, in_dbname, sizeof(dbname));
 	}
-	else if (OidIsValid(dboid))
+	else if (!OidIsValid(dboid))
-	{
-		/* caller specified database by OID */
-		HeapTuple	tuple;
-		Form_pg_database dbform;
-
-		tuple = GetDatabaseTupleByOid(dboid);
-		if (!HeapTupleIsValid(tuple))
-			ereport(FATAL,
-					(errcode(ERRCODE_UNDEFINED_DATABASE),
-					 errmsg("database %u does not exist", dboid)));
-		dbform = (Form_pg_database) GETSTRUCT(tuple);
-		MyDatabaseId = dbform->oid;
-		MyDatabaseTableSpace = dbform->dattablespace;
-		Assert(MyDatabaseId == dboid);
-		strlcpy(dbname, NameStr(dbform->datname), sizeof(dbname));
-		/* pass the database name back to the caller */
-		if (out_dbname)
-			strcpy(out_dbname, dbname);
-	}
-	else
 	{
 		/*
 		 * If this is a background worker not bound to any particular
@@ -1083,30 +1060,7 @@ InitPostgres(const char *in_dbname, Oid dboid,
 	 * CREATE DATABASE.
 	 */
 	if (!bootstrap)
-		LockSharedObject(DatabaseRelationId, MyDatabaseId, 0,
+		LockSharedObject(DatabaseRelationId, dboid, 0, RowExclusiveLock);
-						 RowExclusiveLock);
-
-	/*
-	 * Now we can mark our PGPROC entry with the database ID.
-	 *
-	 * We assume this is an atomic store so no lock is needed; though actually
-	 * things would work fine even if it weren't atomic.  Anyone searching the
-	 * ProcArray for this database's ID should hold the database lock, so they
-	 * would not be executing concurrently with this store.  A process looking
-	 * for another database's ID could in theory see a chance match if it read
-	 * a partially-updated databaseId value; but as long as all such searches
-	 * wait and retry, as in CountOtherDBBackends(), they will certainly see
-	 * the correct value on their next try.
-	 */
-	MyProc->databaseId = MyDatabaseId;
-
-	/*
-	 * We established a catalog snapshot while reading pg_authid and/or
-	 * pg_database; but until we have set up MyDatabaseId, we won't react to
-	 * incoming sinval messages for unshared catalogs, so we won't realize it
-	 * if the snapshot has been invalidated.  Assume it's no good anymore.
-	 */
-	InvalidateCatalogSnapshot();
 
 	/*
 	 * Recheck pg_database to make sure the target database hasn't gone away.
@@ -1118,16 +1072,26 @@ InitPostgres(const char *in_dbname, Oid dboid,
 		HeapTuple	tuple;
 		Form_pg_database datform;
 
-		tuple = GetDatabaseTuple(dbname);
+		tuple = GetDatabaseTupleByOid(dboid);
+		if (HeapTupleIsValid(tuple))
+			datform = (Form_pg_database) GETSTRUCT(tuple);
+
 		if (!HeapTupleIsValid(tuple) ||
-			MyDatabaseId != ((Form_pg_database) GETSTRUCT(tuple))->oid ||
+			(in_dbname && namestrcmp(&datform->datname, in_dbname)))
-			MyDatabaseTableSpace != ((Form_pg_database) GETSTRUCT(tuple))->dattablespace)
+		{
+			if (in_dbname)
 				ereport(FATAL,
 						(errcode(ERRCODE_UNDEFINED_DATABASE),
-					 errmsg("database \"%s\" does not exist", dbname),
+						 errmsg("database \"%s\" does not exist", in_dbname),
 						 errdetail("It seems to have just been dropped or renamed.")));
+			else
+				ereport(FATAL,
+						(errcode(ERRCODE_UNDEFINED_DATABASE),
+						 errmsg("database %u does not exist", dboid)));
+		}
+
+		strlcpy(dbname, NameStr(datform->datname), sizeof(dbname));
 
-		datform = (Form_pg_database) GETSTRUCT(tuple);
 		if (database_is_invalid_form(datform))
 		{
 			ereport(FATAL,
@@ -1135,8 +1099,48 @@ InitPostgres(const char *in_dbname, Oid dboid,
 					errmsg("cannot connect to invalid database \"%s\"", dbname),
 					errhint("Use DROP DATABASE to drop invalid databases."));
 		}
+
+		MyDatabaseTableSpace = datform->dattablespace;
+		/* pass the database name back to the caller */
+		if (out_dbname)
+			strcpy(out_dbname, dbname);
 	}
 
+	/*
+	 * Now that we rechecked, we are certain to be connected to a database and
+	 * thus can set MyDatabaseId.
+	 *
+	 * It is important that MyDatabaseId only be set once we are sure that the
+	 * target database can no longer be concurrently dropped or renamed.  For
+	 * example, without this guarantee, pgstat_update_dbstats() could create
+	 * entries for databases that were just dropped in the pgstat shutdown
+	 * callback, which could confuse other code paths like the autovacuum
+	 * scheduler.
+	 */
+	MyDatabaseId = dboid;
+
+	/*
+	 * Now we can mark our PGPROC entry with the database ID.
+	 *
+	 * We assume this is an atomic store so no lock is needed; though actually
+	 * things would work fine even if it weren't atomic.  Anyone searching the
+	 * ProcArray for this database's ID should hold the database lock, so they
+	 * would not be executing concurrently with this store.  A process looking
+	 * for another database's ID could in theory see a chance match if it read
+	 * a partially-updated databaseId value; but as long as all such searches
+	 * wait and retry, as in CountOtherDBBackends(), they will certainly see
+	 * the correct value on their next try.
+	 */
+	MyProc->databaseId = MyDatabaseId;
+
+	/*
+	 * We established a catalog snapshot while reading pg_authid and/or
+	 * pg_database; but until we have set up MyDatabaseId, we won't react to
+	 * incoming sinval messages for unshared catalogs, so we won't realize it
+	 * if the snapshot has been invalidated.  Assume it's no good anymore.
+	 */
+	InvalidateCatalogSnapshot();
+
 	/*
 	 * Now we should be able to access the database directory safely. Verify
 	 * it's there and looks reasonable.