Don't skip invalid databases when enabling data checksums

When enabling checksums cannot process a database, the launcher uses
DatabaseExists to tell a concurrent drop (benign) from a real failure.
Since 1df361e3d8 that check also treats a present, but-invalid, data-
base as non-existent.  An interrupted DROP DATABASE flush the invalid
marker before the row and files are removed, so a crash or ERROR can
leave an invalid row whose files remain on disk.

Report a database as existing whenever its catalog row is found to
ensure that checksums cannot be enabled if there are invalid databases.
The AccessShareLock in DatabaseExists already waits out an in-flight
drop, so an invalid-but-present row can only be an interrupted drop
leftover whose files still need checksums; enabling then aborts until
it is dropped.

Backpatch to v19 where online checksums were introduced.

Author: Ayush Tiwari <ayushtiwari.slg01@gmail.com>
Reviewed-by: Zsolt Parragi <zsolt.parragi@percona.com>
Reviewed-by: Daniel Gustafsson <daniel@yesql.se>
Discussion: https://postgr.es/m/CAN4CZFOGdqxtZ5-6gb4apqmvoH=Z+TNH8RKJ3mVtoR1HirKQWg@mail.gmail.com
Backpatch-through: 19

src/backend/commands/dbcommands.c

@@ -1883,6 +1883,8 @@ dropdb(const char *dbname, bool missing_ok, bool force)
 	systable_inplace_update_finish(inplace_state, tup);
 	XLogFlush(XactLastRecEnd);
 
+	INJECTION_POINT("dropdb-after-invalid-marker", NULL);
+
 	/*
 	 * Also delete the tuple - transactionally. If this transaction commits,
 	 * the row will be gone, but if we fail, dropdb() can be invoked again.

src/backend/postmaster/datachecksum_state.c

@@ -588,8 +588,10 @@ enable_data_checksums(PG_FUNCTION_ARGS)
 	 * An invalid database cannot be connected to, so the worker would fail to
 	 * process it, and unlike a dropped database its files stay around.  Error
 	 * out early with a hint rather than failing halfway through processing. A
-	 * database which turns invalid after this check is handled by the
-	 * launcher treating it as concurrently dropped.
+	 * database which turns invalid after this check, for example from an
+	 * interrupted DROP DATABASE, instead makes its worker fail; the launcher
+	 * then aborts and leaves checksums disabled, since the invalid database's
+	 * files would otherwise be left without valid checksums.
 	 */
 	ErrorOnInvalidDatabases();
 
@@ -996,11 +998,9 @@ ProcessDatabase(DataChecksumsWorkerDatabase *db)
 
 	/*
 	 * A worker which started but failed before reporting a result has most
-	 * likely FATALed in InitPostgres.  If the database was dropped, or was
-	 * invalidated by a DROP DATABASE which is bound to remove its files,
-	 * after we built the database list then that is the expected outcome and
-	 * not an error, so apply the same heuristic as when the worker failed to
-	 * start.
+	 * likely FATALed in InitPostgres.  If the database was dropped after we
+	 * built the database list then that is the expected outcome and not an
+	 * error, so apply the same heuristic as when the worker failed to start.
 	 */
 	if (result == DATACHECKSUMSWORKER_FAILED && !DatabaseExists(db->dboid))
 		result = DATACHECKSUMSWORKER_DROPDB;
@@ -1415,9 +1415,9 @@ DataChecksumsShmemRequest(void *arg)
  * DatabaseExists
  *
  * Scans the system catalog to check if a database with the given Oid exists
- * and returns true if it is found and valid, else false. Note, we cannot use
- * database_is_invalid_oid here as it will ERROR out, and we want to gracefully
- * handle errors.
+ * and returns true if it is found, even if it is marked invalid.  An invalid
+ * database still has files that need checksums, so only a missing catalog row
+ * proves that a concurrent DROP DATABASE completed.
  */
 static bool
 DatabaseExists(Oid dboid)
@@ -1427,7 +1427,6 @@ DatabaseExists(Oid dboid)
 	SysScanDesc scan;
 	bool		found;
 	HeapTuple	tuple;
-	Form_pg_database pg_database_tuple;
 
 	StartTransactionCommand();
 
@@ -1450,14 +1449,6 @@ DatabaseExists(Oid dboid)
 	tuple = systable_getnext(scan);
 	found = HeapTupleIsValid(tuple);
 
-	/* If the Oid exists, ensure that it's not partially dropped */
-	if (found)
-	{
-		pg_database_tuple = (Form_pg_database) GETSTRUCT(tuple);
-		if (database_is_invalid_form(pg_database_tuple))
-			found = false;
-	}
-
 	systable_endscan(scan);
 	table_close(rel, AccessShareLock);
 

src/test/modules/test_checksums/t/005_injection.pl

@@ -79,6 +79,69 @@ SKIP:
 	);
 }
 
+# ---------------------------------------------------------------------------
+# Test an interrupted DROP DATABASE while checksum enabling is in progress
+#
+
+disable_data_checksums($node, wait => 1);
+
+$node->safe_psql('postgres', 'CREATE DATABASE invalid_dropdb;');
+$node->safe_psql('invalid_dropdb',
+	"CREATE TABLE bad_t AS SELECT generate_series(1,1000) AS a;");
+
+# Hold the worker in "postgres", so invalid_dropdb is still waiting in the
+# launcher's database list when DROP DATABASE is interrupted.
+my $hold = $node->background_psql('postgres');
+$hold->query_safe('CREATE TEMP TABLE holdme (a int);');
+
+my $dropdb_log_offset = -s $node->logfile;
+enable_data_checksums($node);
+$node->poll_query_until(
+	'postgres', qq[
+	SELECT count(*) > 0 FROM pg_stat_activity
+	WHERE backend_type = 'datachecksums worker' AND datname = 'postgres'
+	  AND query LIKE 'Waiting for % temp tables to be removed']
+) or die "timed out waiting for worker to wait for temporary tables";
+
+my $dropdb_log =
+  PostgreSQL::Test::Utils::slurp_file($node->logfile, $dropdb_log_offset);
+unlike(
+	$dropdb_log,
+	qr/initiating data checksum processing in database "invalid_dropdb"/,
+	'invalid database has not been processed yet');
+
+# Leave the database durably marked invalid, but abort DROP DATABASE before
+# its catalog row and files are removed.
+$node->safe_psql('postgres',
+	"SELECT injection_points_attach('dropdb-after-invalid-marker','error');");
+my ($drop_ret, $drop_stdout, $drop_stderr) =
+  $node->psql('postgres', 'DROP DATABASE invalid_dropdb;');
+isnt($drop_ret, 0, 'DROP DATABASE was interrupted after invalidation');
+like(
+	$drop_stderr,
+	qr/dropdb-after-invalid-marker/,
+	'DROP DATABASE reached the invalid-marker injection point');
+$node->safe_psql('postgres',
+	"SELECT injection_points_detach('dropdb-after-invalid-marker');");
+
+my $invalid_state = $node->safe_psql('postgres',
+	"SELECT datconnlimit FROM pg_database WHERE datname = 'invalid_dropdb';");
+is($invalid_state, '-2', 'interrupted DROP left an invalid database row');
+
+# Let checksum processing continue.  The invalid database must be treated as
+# a processing failure, not as a successfully dropped database.
+$hold->query_safe('DROP TABLE holdme;');
+$hold->quit;
+$node->poll_query_until('postgres',
+		"SELECT count(*) = 0 "
+	  . "FROM pg_catalog.pg_stat_activity "
+	  . "WHERE backend_type = 'datachecksums launcher';")
+  or die "timed out waiting for datachecksums launcher to exit";
+test_checksum_state($node, 'off');
+
+# Remove the invalid database and continue with the remaining tests.
+$node->safe_psql('postgres', 'DROP DATABASE invalid_dropdb;');
+
 # ---------------------------------------------------------------------------
 # Test concurrent CREATE DATABASE which use the file_copy strategy
 #