diff --git a/.devcontainer/.vscode/launch.json b/.devcontainer/.vscode/launch.json
index 290f6573ab3..6de90ce09d5 100644
--- a/.devcontainer/.vscode/launch.json
+++ b/.devcontainer/.vscode/launch.json
@@ -16,5 +16,25 @@
                 }
             ],
         },
-    ]
+        {
+            "name": "Open core file",
+            "type": "cppdbg",
+            "request": "launch",
+            "program": "/home/citus/.pgenv/pgsql/bin/postgres",
+            "coreDumpPath": "${input:corefile}",
+            "cwd": "${workspaceFolder}",
+            "MIMode": "gdb",
+        }
+    ],
+    "inputs": [
+        {
+            "id": "corefile",
+            "type": "command",
+            "command": "extension.commandvariable.file.pickFile",
+            "args": {
+                "dialogTitle": "Select core file",
+                "include": "**/core*",
+            },
+        },
+    ],
 }
diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile
index 38055f367e0..13762e1e550 100644
--- a/.devcontainer/Dockerfile
+++ b/.devcontainer/Dockerfile
@@ -152,6 +152,7 @@ RUN sudo apt update \
     lsof \
     man \
     net-tools \
+    psmisc \
     pspg \
     tree \
     vim \
diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json
index 58c9e07a883..cddfcebf4c5 100644
--- a/.devcontainer/devcontainer.json
+++ b/.devcontainer/devcontainer.json
@@ -2,8 +2,11 @@
     "image": "ghcr.io/citusdata/citus-devcontainer:main",
     "runArgs": [
         "--cap-add=SYS_PTRACE",
+        "--ulimit=core=-1",
+    ],
+    "forwardPorts": [
+        9700
     ],
-    "forwardPorts": [9700],
     "customizations": {
         "vscode": {
             "extensions": [
@@ -14,6 +17,7 @@
                 "github.vscode-pull-request-github",
                 "ms-vscode.cpptools-extension-pack",
                 "ms-vsliveshare.vsliveshare",
+                "rioj7.command-variable",
             ],
             "settings": {
                 "files.exclude": {
@@ -30,3 +34,4 @@
     "updateContentCommand": "./configure",
     "postCreateCommand": "make -C .devcontainer/",
 }
+
diff --git a/DEVCONTAINER.md b/DEVCONTAINER.md
new file mode 100644
index 00000000000..d004e6f7122
--- /dev/null
+++ b/DEVCONTAINER.md
@@ -0,0 +1,43 @@
+# Devcontainer
+
+## Coredumps
+When postgres/citus crashes, there is the option to create a coredump. This is useful for debugging the issue. Coredumps are enabled in the devcontainer by default. However, not all environments are configured correctly out of the box. The most important configuration that is not standardized is the `core_pattern`. The configuration can be verified from the container, however, you cannot change this setting from inside the container as the filesystem containing this setting is in read only mode while inside the container.
+
+To verify if corefiles are written run the following command in a terminal. This shows the filename pattern with which the corefile will be written.
+```bash
+cat /proc/sys/kernel/core_pattern
+```
+
+This should be configured with a relative path or simply a simple filename, such as `core`. When your environment shows an absolute path you will need to change this setting. How to change this setting depends highly on the underlying system as the setting needs to be changed on the kernel of the host running the container.
+
+You can put any pattern in `/proc/sys/kernel/core_pattern` as you see fit. eg. You can add the PID to the core pattern in one of two ways;
+- You either include `%p` in the core_pattern. This gets substituted with the PID of the crashing process.
+- Alternatively you could set `/proc/sys/kernel/core_uses_pid` to `1` in the same way as you set `core_pattern`. This will append the PID to the corefile if `%p` is not explicitly contained in the core_pattern.
+
+When a coredump is written you can use the debug/launch configuration `Open core file` which is preconfigured in the devcontainer. This will open a fileprompt that lists all coredumps that are found in your workspace. When you want to debug coredumps from `citus_dev` that are run in your `/data` directory, you can add the data directory to your workspace. In the command pallet of vscode you can run `>Workspace: Add Folder to Workspace...` and select the `/data` directory. This will allow you to open the coredumps from the `/data` directory in the `Open core file` debug configuration.
+
+### Windows (docker desktop)
+When running in docker desktop on windows you will most likely need to change this setting. The linux guest in WSL2 that runs your container is the `docker-desktop` environment. The easiest way to get onto the host, where you can change this setting, is to open a powershell window and verify you have the docker-desktop environment listed.
+
+```powershell
+wsl --list
+```
+
+Among others this should list both `docker-desktop` and `docker-desktop-data`. You can then open a shell in the `docker-desktop` environment.
+
+```powershell
+wsl -d docker-desktop
+```
+
+Inside this shell you can verify that you have the right environment by running
+
+```bash
+cat /proc/sys/kernel/core_pattern
+```
+
+This should show the same configuration as the one you see inside the devcontainer. You can then change the setting by running the following command.
+This will change the setting for the current session. If you want to make the change permanent you will need to add this to a startup script.
+
+```bash
+echo "core" > /proc/sys/kernel/core_pattern
+```
diff --git a/src/backend/distributed/commands/database.c b/src/backend/distributed/commands/database.c
index 33223f41686..5479a59edcf 100644
--- a/src/backend/distributed/commands/database.c
+++ b/src/backend/distributed/commands/database.c
@@ -40,15 +40,38 @@
 #include "distributed/deparse_shard_query.h"
 #include "distributed/deparser.h"
 #include "distributed/listutils.h"
+#include "distributed/local_executor.h"
 #include "distributed/metadata/distobject.h"
 #include "distributed/metadata_sync.h"
 #include "distributed/metadata_utility.h"
 #include "distributed/multi_executor.h"
 #include "distributed/relation_access_tracking.h"
 #include "distributed/serialize_distributed_ddls.h"
+#include "distributed/shard_cleaner.h"
 #include "distributed/worker_protocol.h"
 #include "distributed/worker_transaction.h"
 
+
+/*
+ * Used to save original name of the database before it is replaced with a
+ * temporary name for failure handling purposes in PreprocessCreateDatabaseStmt().
+ */
+static char *CreateDatabaseCommandOriginalDbName = NULL;
+
+
+/*
+ * The format string used when creating a temporary databases for failure
+ * handling purposes.
+ *
+ * The fields are as follows to ensure using a unique name for each temporary
+ * database:
+ * - operationId: The operation id returned by RegisterOperationNeedingCleanup().
+ * - groupId:     The group id of the worker node where CREATE DATABASE command
+ *                is issued from.
+ */
+#define TEMP_DATABASE_NAME_FMT "citus_temp_database_%lu_%d"
+
+
 /*
  * DatabaseCollationInfo is used to store collation related information of a database.
  */
@@ -286,8 +309,9 @@ PreprocessAlterDatabaseStmt(Node *node, const char *queryString,
 		 * NontransactionalNodeDDLTask to run the command on the workers outside
 		 * the transaction block.
 		 */
-
-		return NontransactionalNodeDDLTaskList(NON_COORDINATOR_NODES, commands);
+		bool warnForPartialFailure = true;
+		return NontransactionalNodeDDLTaskList(NON_COORDINATOR_NODES, commands,
+											   warnForPartialFailure);
 	}
 	else
 	{
@@ -453,7 +477,12 @@ PreprocessAlterDatabaseSetStmt(Node *node, const char *queryString,
  *
  * In this stage, we perform validations that we want to ensure before delegating to
  * previous utility hooks because it might not be convenient to throw an error in an
- * implicit transaction that creates a database.
+ * implicit transaction that creates a database. Also in this stage, we save the original
+ * database name and replace dbname field with a temporary name for failure handling
+ * purposes. We let Postgres create the database with the temporary name, insert a cleanup
+ * record for the temporary database name on all nodes and let PostprocessCreateDatabaseStmt()
+ * to return the distributed DDL job that both creates the database with the temporary name
+ * and then renames it back to its original name.
  *
  * We also serialize database commands globally by acquiring a Citus specific advisory
  * lock based on OCLASS_DATABASE on the first primary worker node.
@@ -467,22 +496,56 @@ PreprocessCreateDatabaseStmt(Node *node, const char *queryString,
 		return NIL;
 	}
 
-	EnsurePropagationToCoordinator();
+	EnsureCoordinatorIsInMetadata();
 
 	CreatedbStmt *stmt = castNode(CreatedbStmt, node);
 	EnsureSupportedCreateDatabaseCommand(stmt);
 
 	SerializeDistributedDDLsOnObjectClass(OCLASS_DATABASE);
 
+	OperationId operationId = RegisterOperationNeedingCleanup();
+
+	char *tempDatabaseName = psprintf(TEMP_DATABASE_NAME_FMT,
+									  operationId, GetLocalGroupId());
+
+	List *remoteNodes = TargetWorkerSetNodeList(ALL_SHARD_NODES, RowShareLock);
+	WorkerNode *remoteNode = NULL;
+	foreach_ptr(remoteNode, remoteNodes)
+	{
+		InsertCleanupRecordOutsideTransaction(
+			CLEANUP_OBJECT_DATABASE,
+			pstrdup(quote_identifier(tempDatabaseName)),
+			remoteNode->groupId,
+			CLEANUP_ON_FAILURE
+			);
+	}
+
+	CreateDatabaseCommandOriginalDbName = stmt->dbname;
+	stmt->dbname = tempDatabaseName;
+
+	/*
+	 * Delete cleanup records in the same transaction so that if the current
+	 * transactions fails for some reason, then the cleanup records won't be
+	 * deleted. In the happy path, we will delete the cleanup records without
+	 * deferring them to the background worker.
+	 */
+	FinalizeOperationNeedingCleanupOnSuccess("create database");
+
 	return NIL;
 }
 
 
 /*
  * PostprocessCreateDatabaseStmt is executed after the statement is applied to the local
- * postgres instance. In this stage we prepare the commands that need to be run on
- * all workers to create the database.
+ * postgres instance.
  *
+ * In this stage, we first rename the temporary database back to its original name for
+ * local node and then return a list of distributed DDL jobs to create the database with
+ * the temporary name and then to rename it back to its original name. That way, if CREATE
+ * DATABASE fails on any of the nodes, the temporary database will be cleaned up by the
+ * cleanup records that we inserted in PreprocessCreateDatabaseStmt() and in case of a
+ * failure, we won't leak any databases called as the name that user intended to use for
+ * the database.
  */
 List *
 PostprocessCreateDatabaseStmt(Node *node, const char *queryString)
@@ -515,9 +578,55 @@ PostprocessCreateDatabaseStmt(Node *node, const char *queryString)
 	 * block, we need to use NontransactionalNodeDDLTaskList() to send the CREATE
 	 * DATABASE statement to the workers.
 	 */
+	bool warnForPartialFailure = false;
 	List *createDatabaseDDLJobList =
-		NontransactionalNodeDDLTaskList(REMOTE_NODES, createDatabaseCommands);
-	return createDatabaseDDLJobList;
+		NontransactionalNodeDDLTaskList(REMOTE_NODES, createDatabaseCommands,
+										warnForPartialFailure);
+
+	CreatedbStmt *stmt = castNode(CreatedbStmt, node);
+
+	char *renameDatabaseCommand =
+		psprintf("ALTER DATABASE %s RENAME TO %s",
+				 quote_identifier(stmt->dbname),
+				 quote_identifier(CreateDatabaseCommandOriginalDbName));
+
+	List *renameDatabaseCommands = list_make3(DISABLE_DDL_PROPAGATION,
+											  renameDatabaseCommand,
+											  ENABLE_DDL_PROPAGATION);
+
+	/*
+	 * We use NodeDDLTaskList() to send the RENAME DATABASE statement to the
+	 * workers because we want to execute it in a coordinated transaction.
+	 */
+	List *renameDatabaseDDLJobList =
+		NodeDDLTaskList(REMOTE_NODES, renameDatabaseCommands);
+
+	/*
+	 * Temporarily disable citus.enable_ddl_propagation before issuing
+	 * rename command locally because we don't want to execute it on remote
+	 * nodes yet. We will execute it on remote nodes by returning it as a
+	 * distributed DDL job.
+	 *
+	 * The reason why we don't want to execute it on remote nodes yet is that
+	 * the database is not created on remote nodes yet.
+	 */
+	int saveNestLevel = NewGUCNestLevel();
+	set_config_option("citus.enable_ddl_propagation", "off",
+					  (superuser() ? PGC_SUSET : PGC_USERSET), PGC_S_SESSION,
+					  GUC_ACTION_LOCAL, true, 0, false);
+
+	ExecuteUtilityCommand(renameDatabaseCommand);
+
+	AtEOXact_GUC(true, saveNestLevel);
+
+	/*
+	 * Restore the original database name because MarkObjectDistributed()
+	 * resolves oid of the object based on the database name and is called
+	 * after executing the distributed DDL job that renames temporary database.
+	 */
+	stmt->dbname = CreateDatabaseCommandOriginalDbName;
+
+	return list_concat(createDatabaseDDLJobList, renameDatabaseDDLJobList);
 }
 
 
@@ -571,8 +680,10 @@ PreprocessDropDatabaseStmt(Node *node, const char *queryString,
 	 * use NontransactionalNodeDDLTaskList() to send the DROP DATABASE statement
 	 * to the workers.
 	 */
+	bool warnForPartialFailure = true;
 	List *dropDatabaseDDLJobList =
-		NontransactionalNodeDDLTaskList(REMOTE_NODES, dropDatabaseCommands);
+		NontransactionalNodeDDLTaskList(REMOTE_NODES, dropDatabaseCommands,
+										warnForPartialFailure);
 	return dropDatabaseDDLJobList;
 }
 
diff --git a/src/backend/distributed/commands/index.c b/src/backend/distributed/commands/index.c
index c4113617666..e97312df271 100644
--- a/src/backend/distributed/commands/index.c
+++ b/src/backend/distributed/commands/index.c
@@ -493,6 +493,7 @@ GenerateCreateIndexDDLJob(IndexStmt *createIndexStatement, const char *createInd
 	ddlJob->startNewTransaction = createIndexStatement->concurrent;
 	ddlJob->metadataSyncCommand = createIndexCommand;
 	ddlJob->taskList = CreateIndexTaskList(createIndexStatement);
+	ddlJob->warnForPartialFailure = true;
 
 	return ddlJob;
 }
@@ -652,6 +653,7 @@ PreprocessReindexStmt(Node *node, const char *reindexCommand,
 																	"concurrently");
 			ddlJob->metadataSyncCommand = reindexCommand;
 			ddlJob->taskList = CreateReindexTaskList(relationId, reindexStatement);
+			ddlJob->warnForPartialFailure = true;
 
 			ddlJobs = list_make1(ddlJob);
 		}
@@ -780,6 +782,7 @@ PreprocessDropIndexStmt(Node *node, const char *dropIndexCommand,
 		ddlJob->metadataSyncCommand = dropIndexCommand;
 		ddlJob->taskList = DropIndexTaskList(distributedRelationId, distributedIndexId,
 											 dropIndexStatement);
+		ddlJob->warnForPartialFailure = true;
 
 		ddlJobs = list_make1(ddlJob);
 	}
diff --git a/src/backend/distributed/commands/utility_hook.c b/src/backend/distributed/commands/utility_hook.c
index a1a23331076..7dff9cbf6cb 100644
--- a/src/backend/distributed/commands/utility_hook.c
+++ b/src/backend/distributed/commands/utility_hook.c
@@ -1377,7 +1377,7 @@ ExecuteDistributedDDLJob(DDLJob *ddlJob)
 						 errhint("Use DROP INDEX CONCURRENTLY IF EXISTS to remove the "
 								 "invalid index, then retry the original command.")));
 			}
-			else
+			else if (ddlJob->warnForPartialFailure)
 			{
 				ereport(WARNING,
 						(errmsg(
@@ -1386,9 +1386,9 @@ ExecuteDistributedDDLJob(DDLJob *ddlJob)
 							 "state.\nIf the problematic command is a CREATE operation, "
 							 "consider using the 'IF EXISTS' syntax to drop the object,"
 							 "\nif applicable, and then re-attempt the original command.")));
-
-				PG_RE_THROW();
 			}
+
+			PG_RE_THROW();
 		}
 		PG_END_TRY();
 	}
@@ -1604,9 +1604,12 @@ DDLTaskList(Oid relationId, const char *commandString)
  * NontransactionalNodeDDLTaskList builds a list of tasks to execute a DDL command on a
  * given target set of nodes with cannotBeExecutedInTransaction is set to make sure
  * that task list is executed outside a transaction block.
+ *
+ * Also sets warnForPartialFailure for the returned DDLJobs.
  */
 List *
-NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands)
+NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands,
+								bool warnForPartialFailure)
 {
 	List *ddlJobs = NodeDDLTaskList(targets, commands);
 	DDLJob *ddlJob = NULL;
@@ -1617,6 +1620,8 @@ NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands)
 		{
 			task->cannotBeExecutedInTransaction = true;
 		}
+
+		ddlJob->warnForPartialFailure = warnForPartialFailure;
 	}
 	return ddlJobs;
 }
diff --git a/src/backend/distributed/operations/shard_cleaner.c b/src/backend/distributed/operations/shard_cleaner.c
index db1cad6bcae..2efce9a7b09 100644
--- a/src/backend/distributed/operations/shard_cleaner.c
+++ b/src/backend/distributed/operations/shard_cleaner.c
@@ -92,6 +92,8 @@ static bool TryDropReplicationSlotOutsideTransaction(char *replicationSlotName,
 													 char *nodeName,
 													 int nodePort);
 static bool TryDropUserOutsideTransaction(char *username, char *nodeName, int nodePort);
+static bool TryDropDatabaseOutsideTransaction(char *databaseName, char *nodeName,
+											  int nodePort);
 
 static CleanupRecord * GetCleanupRecordByNameAndType(char *objectName,
 													 CleanupObject type);
@@ -141,7 +143,6 @@ Datum
 citus_cleanup_orphaned_resources(PG_FUNCTION_ARGS)
 {
 	CheckCitusVersion(ERROR);
-	EnsureCoordinator();
 	PreventInTransactionBlock(true, "citus_cleanup_orphaned_resources");
 
 	int droppedCount = DropOrphanedResourcesForCleanup();
@@ -245,12 +246,6 @@ TryDropOrphanedResources()
 static int
 DropOrphanedResourcesForCleanup()
 {
-	/* Only runs on Coordinator */
-	if (!IsCoordinator())
-	{
-		return 0;
-	}
-
 	List *cleanupRecordList = ListCleanupRecords();
 
 	/*
@@ -608,6 +603,12 @@ TryDropResourceByCleanupRecordOutsideTransaction(CleanupRecord *record,
 			return TryDropUserOutsideTransaction(record->objectName, nodeName, nodePort);
 		}
 
+		case CLEANUP_OBJECT_DATABASE:
+		{
+			return TryDropDatabaseOutsideTransaction(record->objectName, nodeName,
+													 nodePort);
+		}
+
 		default:
 		{
 			ereport(WARNING, (errmsg(
@@ -888,6 +889,69 @@ TryDropUserOutsideTransaction(char *username,
 }
 
 
+/*
+ * TryDropDatabaseOutsideTransaction drops the database with the given name
+ * if it exists.
+ */
+static bool
+TryDropDatabaseOutsideTransaction(char *databaseName, char *nodeName, int nodePort)
+{
+	int connectionFlags = (OUTSIDE_TRANSACTION | FORCE_NEW_CONNECTION);
+	MultiConnection *connection = GetNodeUserDatabaseConnection(connectionFlags,
+																nodeName, nodePort,
+																CitusExtensionOwnerName(),
+																NULL);
+
+	if (PQstatus(connection->pgConn) != CONNECTION_OK)
+	{
+		return false;
+	}
+
+	/*
+	 * We want to disable DDL propagation and set lock_timeout before issuing
+	 * the DROP DATABASE command but we cannot do so in a way that's scoped
+	 * to the DROP DATABASE command. This is because, we cannot use a
+	 * transaction block for the DROP DATABASE command.
+	 *
+	 * For this reason, to avoid leaking the lock_timeout and DDL propagation
+	 * settings to future commands, we force the connection to close at the end
+	 * of the transaction.
+	 */
+	ForceConnectionCloseAtTransactionEnd(connection);
+
+	/*
+	 * The DROP DATABASE command should not propagate, so we disable DDL
+	 * propagation.
+	 */
+	List *commandList = list_make3(
+		"SET lock_timeout TO '1s'",
+		"SET citus.enable_ddl_propagation TO OFF;",
+		psprintf("DROP DATABASE IF EXISTS %s;", quote_identifier(databaseName))
+		);
+
+	bool executeCommand = true;
+
+	const char *commandString = NULL;
+	foreach_ptr(commandString, commandList)
+	{
+		/*
+		 * Cannot use SendOptionalCommandListToWorkerOutsideTransactionWithConnection()
+		 * because we don't want to open a transaction block on remote nodes as DROP
+		 * DATABASE commands cannot be run inside a transaction block.
+		 */
+		if (ExecuteOptionalRemoteCommand(connection, commandString, NULL) !=
+			RESPONSE_OKAY)
+		{
+			executeCommand = false;
+			break;
+		}
+	}
+
+	CloseConnection(connection);
+	return executeCommand;
+}
+
+
 /*
  * ErrorIfCleanupRecordForShardExists errors out if a cleanup record for the given
  * shard name exists.
diff --git a/src/include/distributed/commands/utility_hook.h b/src/include/distributed/commands/utility_hook.h
index 9046c73093e..52fcf70912c 100644
--- a/src/include/distributed/commands/utility_hook.h
+++ b/src/include/distributed/commands/utility_hook.h
@@ -75,6 +75,15 @@ typedef struct DDLJob
 	const char *metadataSyncCommand;
 
 	List *taskList;            /* worker DDL tasks to execute */
+
+	/*
+	 * Only applicable when any of the tasks cannot be executed in a
+	 * transaction block.
+	 *
+	 * Controls whether to emit a warning within the utility hook in case of a
+	 * failure.
+	 */
+	bool warnForPartialFailure;
 } DDLJob;
 
 extern ProcessUtility_hook_type PrevProcessUtility;
@@ -94,7 +103,8 @@ extern void ProcessUtilityParseTree(Node *node, const char *queryString,
 extern void MarkInvalidateForeignKeyGraph(void);
 extern void InvalidateForeignKeyGraphForDDL(void);
 extern List * DDLTaskList(Oid relationId, const char *commandString);
-extern List * NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands);
+extern List * NontransactionalNodeDDLTaskList(TargetWorkerSet targets, List *commands,
+											  bool warnForPartialFailure);
 extern List * NodeDDLTaskList(TargetWorkerSet targets, List *commands);
 extern bool AlterTableInProgress(void);
 extern bool DropSchemaOrDBInProgress(void);
diff --git a/src/include/distributed/shard_cleaner.h b/src/include/distributed/shard_cleaner.h
index 4967846b2ba..7609bd90024 100644
--- a/src/include/distributed/shard_cleaner.h
+++ b/src/include/distributed/shard_cleaner.h
@@ -41,7 +41,8 @@ typedef enum CleanupObject
 	CLEANUP_OBJECT_SUBSCRIPTION = 2,
 	CLEANUP_OBJECT_REPLICATION_SLOT = 3,
 	CLEANUP_OBJECT_PUBLICATION = 4,
-	CLEANUP_OBJECT_USER = 5
+	CLEANUP_OBJECT_USER = 5,
+	CLEANUP_OBJECT_DATABASE = 6
 } CleanupObject;
 
 /*
diff --git a/src/test/regress/expected/alter_database_propagation.out b/src/test/regress/expected/alter_database_propagation.out
index f01d39ab911..5c45a25e29c 100644
--- a/src/test/regress/expected/alter_database_propagation.out
+++ b/src/test/regress/expected/alter_database_propagation.out
@@ -140,7 +140,12 @@ DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 NOTICE:  issuing ALTER DATABASE regression RESET lock_timeout
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 set citus.enable_create_database_propagation=on;
+SET citus.next_operation_id TO 3000;
 create database "regression!'2";
+NOTICE:  issuing ALTER DATABASE citus_temp_database_3000_0 RENAME TO "regression!'2"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER DATABASE citus_temp_database_3000_0 RENAME TO "regression!'2"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 alter database "regression!'2" with CONNECTION LIMIT 100;
 NOTICE:  issuing ALTER DATABASE "regression!'2" WITH CONNECTION LIMIT 100;
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
@@ -189,7 +194,12 @@ alter DATABASE local_regression rename to local_regression2;
 drop database local_regression2;
 set citus.enable_create_database_propagation=on;
 drop database regression3;
+SET citus.next_operation_id TO 3100;
 create database "regression!'4";
+NOTICE:  issuing ALTER DATABASE citus_temp_database_3100_0 RENAME TO "regression!'4"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER DATABASE citus_temp_database_3100_0 RENAME TO "regression!'4"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 SELECT result FROM run_command_on_all_nodes(
   $$
   ALTER TABLESPACE alter_db_tablespace RENAME TO "ts-needs\!escape"
diff --git a/src/test/regress/expected/create_drop_database_propagation.out b/src/test/regress/expected/create_drop_database_propagation.out
index da4ec4eb711..4ddbaae3fe3 100644
--- a/src/test/regress/expected/create_drop_database_propagation.out
+++ b/src/test/regress/expected/create_drop_database_propagation.out
@@ -427,11 +427,16 @@ SELECT * FROM public.check_database_on_all_nodes('my_template_database') ORDER B
 --tests for special characters in database name
 set citus.enable_create_database_propagation=on;
 SET citus.log_remote_commands = true;
-set citus.grep_remote_commands = '%CREATE DATABASE%';
+set citus.grep_remote_commands = '%DATABASE%';
+SET citus.next_operation_id TO 2000;
 create database "mydatabase#1'2";
-NOTICE:  issuing CREATE DATABASE "mydatabase#1'2"
+NOTICE:  issuing CREATE DATABASE citus_temp_database_2000_0
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
-NOTICE:  issuing CREATE DATABASE "mydatabase#1'2"
+NOTICE:  issuing CREATE DATABASE citus_temp_database_2000_0
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER DATABASE citus_temp_database_2000_0 RENAME TO "mydatabase#1'2"
+DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
+NOTICE:  issuing ALTER DATABASE citus_temp_database_2000_0 RENAME TO "mydatabase#1'2"
 DETAIL:  on server postgres@localhost:xxxxx connectionId: xxxxxxx
 set citus.grep_remote_commands = '%DROP DATABASE%';
 drop database if exists "mydatabase#1'2";
@@ -1264,6 +1269,95 @@ SELECT 1 FROM run_command_on_all_nodes($$REVOKE ALL ON TABLESPACE pg_default FRO
 
 DROP DATABASE no_createdb;
 DROP USER no_createdb;
+-- Test a failure scenario by trying to create a distributed database that
+-- already exists on one of the nodes.
+\c - - - :worker_1_port
+CREATE DATABASE "test_\!failure";
+NOTICE:  Citus partially supports CREATE DATABASE for distributed databases
+DETAIL:  Citus does not propagate CREATE DATABASE command to other nodes
+HINT:  You can manually create a database and its extensions on other nodes.
+\c - - - :master_port
+SET citus.enable_create_database_propagation TO ON;
+CREATE DATABASE "test_\!failure";
+ERROR:  database "test_\!failure" already exists
+CONTEXT:  while executing command on localhost:xxxxx
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+ database_cleanedup_on_node
+---------------------------------------------------------------------
+ t
+ t
+ t
+(3 rows)
+
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure$$) ORDER BY node_type, result;
+      node_type       |                                                                                                                                                                                                                          result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": {"datacl": null, "datname": "test_\\!failure", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SET citus.enable_create_database_propagation TO OFF;
+CREATE DATABASE "test_\!failure1";
+NOTICE:  Citus partially supports CREATE DATABASE for distributed databases
+DETAIL:  Citus does not propagate CREATE DATABASE command to other nodes
+HINT:  You can manually create a database and its extensions on other nodes.
+\c - - - :worker_1_port
+DROP DATABASE "test_\!failure";
+SET citus.enable_create_database_propagation TO ON;
+CREATE DATABASE "test_\!failure1";
+ERROR:  database "test_\!failure1" already exists
+CONTEXT:  while executing command on localhost:xxxxx
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+ database_cleanedup_on_node
+---------------------------------------------------------------------
+ t
+ t
+ t
+(3 rows)
+
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure1$$) ORDER BY node_type, result;
+      node_type       |                                                                                                                                                                                                                           result
+---------------------------------------------------------------------
+ coordinator (remote) | {"database_properties": {"datacl": null, "datname": "test_\\!failure1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+\c - - - :master_port
+-- Before dropping local "test_\!failure1" database, test a failure scenario
+-- by trying to create a distributed database that already exists "on local
+-- node" this time.
+SET citus.enable_create_database_propagation TO ON;
+CREATE DATABASE "test_\!failure1";
+ERROR:  database "test_\!failure1" already exists
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+ database_cleanedup_on_node
+---------------------------------------------------------------------
+ t
+ t
+ t
+(3 rows)
+
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure1$$) ORDER BY node_type, result;
+      node_type       |                                                                                                                                                                                                                           result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": {"datacl": null, "datname": "test_\\!failure1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SET citus.enable_create_database_propagation TO OFF;
+DROP DATABASE "test_\!failure1";
 SET citus.enable_create_database_propagation TO ON;
 --clean up resources created by this test
 -- DROP TABLESPACE is not supported, so we need to drop it manually.
diff --git a/src/test/regress/expected/failure_create_database.out b/src/test/regress/expected/failure_create_database.out
new file mode 100644
index 00000000000..81fcd451965
--- /dev/null
+++ b/src/test/regress/expected/failure_create_database.out
@@ -0,0 +1,386 @@
+SET citus.enable_create_database_propagation TO ON;
+SET client_min_messages TO WARNING;
+SELECT 1 FROM citus_add_node('localhost', :master_port, 0);
+ ?column?
+---------------------------------------------------------------------
+        1
+(1 row)
+
+CREATE FUNCTION get_temp_databases_on_nodes()
+RETURNS TEXT AS $func$
+  SELECT array_agg(DISTINCT result ORDER BY result) AS temp_databases_on_nodes FROM run_command_on_all_nodes($$SELECT datname FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$) WHERE result != '';
+$func$
+LANGUAGE sql;
+CREATE FUNCTION count_db_cleanup_records()
+RETURNS TABLE(object_name TEXT, count INTEGER) AS $func$
+  SELECT object_name, COUNT(*) FROM pg_dist_cleanup WHERE object_name LIKE 'citus_temp_database_%' GROUP BY object_name;
+$func$
+LANGUAGE sql;
+CREATE FUNCTION ensure_no_temp_databases_on_any_nodes()
+RETURNS BOOLEAN AS $func$
+  SELECT bool_and(result::boolean) AS no_temp_databases_on_any_nodes FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+$func$
+LANGUAGE sql;
+-- cleanup any orphaned resources from previous runs
+CALL citus_cleanup_orphaned_resources();
+SET citus.next_operation_id TO 4000;
+ALTER SYSTEM SET citus.defer_shard_delete_interval TO -1;
+SELECT pg_reload_conf();
+ pg_reload_conf
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT pg_sleep(0.1);
+ pg_sleep
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT citus.mitmproxy('conn.kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection to the remote node postgres@localhost:xxxxx failed with the following error: connection not open
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^CREATE DATABASE").cancel(' || pg_backend_pid() || ')');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  canceling statement due to user request
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+ {citus_temp_database_4000_0}
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+        object_name         | count
+---------------------------------------------------------------------
+ citus_temp_database_4000_0 |     3
+(1 row)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^ALTER DATABASE").cancel(' || pg_backend_pid() || ')');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  canceling statement due to user request
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+ {citus_temp_database_4001_0}
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+        object_name         | count
+---------------------------------------------------------------------
+ citus_temp_database_4001_0 |     3
+(1 row)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection to the remote node postgres@localhost:xxxxx failed with the following error: connection not open
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^PREPARE TRANSACTION").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection not open
+CONTEXT:  while executing command on localhost:xxxxx
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+ {citus_temp_database_4002_0}
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+        object_name         | count
+---------------------------------------------------------------------
+ citus_temp_database_4002_0 |     3
+(1 row)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^COMMIT PREPARED").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+WARNING:  connection not open
+CONTEXT:  while executing command on localhost:xxxxx
+WARNING:  failed to commit transaction on localhost:xxxxx
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+-- not call citus_cleanup_orphaned_resources() but recover the prepared transactions this time
+SELECT 1 FROM recover_prepared_transactions();
+ ?column?
+---------------------------------------------------------------------
+        1
+(1 row)
+
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                                                                                                                                                                                    result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": {"datacl": null, "datname": "db1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": true, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": {"datacl": null, "datname": "db1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": true, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": {"datacl": null, "datname": "db1", "datctype": "C", "encoding": "UTF8", "datcollate": "C", "tablespace": "pg_default", "daticurules": null, "datallowconn": true, "datconnlimit": -1, "daticulocale": null, "datistemplate": false, "database_owner": "postgres", "datcollversion": null, "datlocprovider": "c"}, "pg_dist_object_record_for_db_exists": true, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+DROP DATABASE db1;
+-- after recovering the prepared transactions, cleanup records should also be removed
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+SELECT citus.mitmproxy('conn.onQuery(query="^SELECT citus_internal.acquire_citus_advisory_object_class_lock").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection to the remote node postgres@localhost:xxxxx failed with the following error: connection not open
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+SELECT citus.mitmproxy('conn.onParse(query="^WITH distributed_object_data").kill()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+CREATE DATABASE db1;
+ERROR:  connection not open
+CONTEXT:  while executing command on localhost:xxxxx
+SELECT citus.mitmproxy('conn.allow()');
+ mitmproxy
+---------------------------------------------------------------------
+
+(1 row)
+
+SELECT get_temp_databases_on_nodes();
+ get_temp_databases_on_nodes
+---------------------------------------------------------------------
+ {citus_temp_database_4004_0}
+(1 row)
+
+SELECT * FROM count_db_cleanup_records();
+        object_name         | count
+---------------------------------------------------------------------
+ citus_temp_database_4004_0 |     3
+(1 row)
+
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+ ensure_no_temp_databases_on_any_nodes
+---------------------------------------------------------------------
+ t
+(1 row)
+
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+      node_type       |                                                              result
+---------------------------------------------------------------------
+ coordinator (local)  | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+ worker node (remote) | {"database_properties": null, "pg_dist_object_record_for_db_exists": false, "stale_pg_dist_object_record_for_a_db_exists": false}
+(3 rows)
+
+CREATE DATABASE db1;
+-- show that a successful database creation doesn't leave any pg_dist_cleanup records behind
+SELECT * FROM count_db_cleanup_records();
+ object_name | count
+---------------------------------------------------------------------
+(0 rows)
+
+DROP DATABASE db1;
+DROP FUNCTION get_temp_databases_on_nodes();
+DROP FUNCTION ensure_no_temp_databases_on_any_nodes();
+DROP FUNCTION count_db_cleanup_records();
+SELECT 1 FROM citus_remove_node('localhost', :master_port);
+ ?column?
+---------------------------------------------------------------------
+        1
+(1 row)
+
diff --git a/src/test/regress/expected/isolation_database_cmd_from_any_node.out b/src/test/regress/expected/isolation_database_cmd_from_any_node.out
index 771c67fe88f..e952bb45728 100644
--- a/src/test/regress/expected/isolation_database_cmd_from_any_node.out
+++ b/src/test/regress/expected/isolation_database_cmd_from_any_node.out
@@ -1,6 +1,11 @@
 Parsed test spec with 2 sessions
 
 starting permutation: s1-begin s2-begin s1-acquire-citus-adv-oclass-lock s2-acquire-citus-adv-oclass-lock s1-commit s2-commit
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
 step s1-acquire-citus-adv-oclass-lock: SELECT citus_internal.acquire_citus_advisory_object_class_lock(value, NULL) FROM oclass_database;
@@ -18,8 +23,18 @@ acquire_citus_advisory_object_class_lock
 (1 row)
 
 step s2-commit: COMMIT;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s1-begin s2-begin s1-acquire-citus-adv-oclass-lock-with-oid-testdb1 s2-acquire-citus-adv-oclass-lock-with-oid-testdb1 s1-commit s2-commit s1-drop-testdb1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
@@ -39,8 +54,18 @@ acquire_citus_advisory_object_class_lock
 
 step s2-commit: COMMIT;
 step s1-drop-testdb1: DROP DATABASE testdb1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s2-create-testdb2 s1-begin s2-begin s1-acquire-citus-adv-oclass-lock-with-oid-testdb1 s2-acquire-citus-adv-oclass-lock-with-oid-testdb2 s1-commit s2-commit s1-drop-testdb1 s2-drop-testdb2
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-begin: BEGIN;
@@ -61,8 +86,18 @@ step s1-commit: COMMIT;
 step s2-commit: COMMIT;
 step s1-drop-testdb1: DROP DATABASE testdb1;
 step s2-drop-testdb2: DROP DATABASE testdb2;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s1-begin s2-begin s1-acquire-citus-adv-oclass-lock s2-acquire-citus-adv-oclass-lock-with-oid-testdb2 s1-commit s2-commit s2-drop-testdb2
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
@@ -81,8 +116,18 @@ acquire_citus_advisory_object_class_lock
 step s1-commit: COMMIT;
 step s2-commit: COMMIT;
 step s2-drop-testdb2: DROP DATABASE testdb2;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-set-lc_monetary s1-create-db1 s2-rollback s2-drop-testdb2 s1-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-set-lc_monetary: ALTER DATABASE testdb2 SET lc_monetary TO 'C';
@@ -90,8 +135,18 @@ step s1-create-db1: CREATE DATABASE db1;
 step s2-rollback: ROLLBACK;
 step s2-drop-testdb2: DROP DATABASE testdb2;
 step s1-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-set-lc_monetary s1-create-user-dbuser s1-grant-on-testdb2-to-dbuser s2-rollback s2-drop-testdb2 s1-drop-user-dbuser
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-set-lc_monetary: ALTER DATABASE testdb2 SET lc_monetary TO 'C';
@@ -100,8 +155,18 @@ step s1-grant-on-testdb2-to-dbuser: GRANT ALL ON DATABASE testdb2 TO dbuser;
 step s2-rollback: ROLLBACK;
 step s2-drop-testdb2: DROP DATABASE testdb2;
 step s1-drop-user-dbuser: DROP USER dbuser;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-set-lc_monetary s1-create-testdb1 s1-create-user-dbuser s1-grant-on-testdb1-to-dbuser s2-rollback s2-drop-testdb2 s1-drop-testdb1 s1-drop-user-dbuser
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-set-lc_monetary: ALTER DATABASE testdb2 SET lc_monetary TO 'C';
@@ -112,8 +177,18 @@ step s2-rollback: ROLLBACK;
 step s2-drop-testdb2: DROP DATABASE testdb2;
 step s1-drop-testdb1: DROP DATABASE testdb1;
 step s1-drop-user-dbuser: DROP USER dbuser;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s2-create-testdb2 s1-begin s2-begin s1-alter-testdb1-rename-to-db1 s2-alter-testdb2-rename-to-db1 s1-commit s2-rollback s1-drop-db1 s2-drop-testdb2
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-begin: BEGIN;
@@ -126,8 +201,18 @@ ERROR:  database "db1" already exists
 step s2-rollback: ROLLBACK;
 step s1-drop-db1: DROP DATABASE db1;
 step s2-drop-testdb2: DROP DATABASE testdb2;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s2-create-testdb2 s1-begin s2-begin s1-alter-testdb1-rename-to-db1 s2-alter-testdb2-rename-to-db1 s1-rollback s2-commit s1-drop-testdb1 s2-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-begin: BEGIN;
@@ -139,8 +224,18 @@ step s2-alter-testdb2-rename-to-db1: <... completed>
 step s2-commit: COMMIT;
 step s1-drop-testdb1: DROP DATABASE testdb1;
 step s2-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s1-begin s2-begin s1-alter-testdb1-rename-to-db1 s2-alter-testdb1-rename-to-db1 s1-commit s2-rollback s1-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
@@ -151,8 +246,18 @@ step s2-alter-testdb1-rename-to-db1: <... completed>
 ERROR:  database "testdb1" does not exist
 step s2-rollback: ROLLBACK;
 step s1-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s1-create-testdb1 s1-begin s2-begin s1-alter-testdb1-rename-to-db1 s2-alter-testdb1-rename-to-db1 s1-rollback s2-commit s2-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s1-create-testdb1: CREATE DATABASE testdb1;
 step s1-begin: BEGIN;
 step s2-begin: BEGIN;
@@ -162,8 +267,18 @@ step s1-rollback: ROLLBACK;
 step s2-alter-testdb1-rename-to-db1: <... completed>
 step s2-commit: COMMIT;
 step s2-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-rename-to-db1 s1-create-db1 s2-rollback s2-drop-testdb2 s1-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-rename-to-db1: ALTER DATABASE testdb2 RENAME TO db1;
@@ -172,8 +287,18 @@ step s2-rollback: ROLLBACK;
 step s1-create-db1: <... completed>
 step s2-drop-testdb2: DROP DATABASE testdb2;
 step s1-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-rename-to-db1 s1-create-db1 s2-commit s2-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-rename-to-db1: ALTER DATABASE testdb2 RENAME TO db1;
@@ -182,8 +307,18 @@ step s2-commit: COMMIT;
 step s1-create-db1: <... completed>
 ERROR:  database "db1" already exists
 step s2-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-rename-to-db2 s1-create-db1 s2-commit s2-drop-db2 s1-drop-db1
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-rename-to-db2: ALTER DATABASE testdb2 RENAME TO db2;
@@ -192,16 +327,36 @@ step s2-commit: COMMIT;
 step s1-create-db1: <... completed>
 step s2-drop-db2: DROP DATABASE db2;
 step s1-drop-db1: DROP DATABASE db1;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s2-begin s2-alter-testdb2-rename-to-db1 s1-drop-testdb2 s2-rollback
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s2-begin: BEGIN;
 step s2-alter-testdb2-rename-to-db1: ALTER DATABASE testdb2 RENAME TO db1;
 step s1-drop-testdb2: DROP DATABASE testdb2; <waiting ...>
 step s2-rollback: ROLLBACK;
 step s1-drop-testdb2: <... completed>
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 
 starting permutation: s2-create-testdb2 s1-create-db1 s2-begin s2-alter-testdb2-rename-to-db2 s1-drop-db1 s2-commit s2-drop-db2
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
 step s2-create-testdb2: CREATE DATABASE testdb2;
 step s1-create-db1: CREATE DATABASE db1;
 step s2-begin: BEGIN;
@@ -209,3 +364,8 @@ step s2-alter-testdb2-rename-to-db2: ALTER DATABASE testdb2 RENAME TO db2;
 step s1-drop-db1: DROP DATABASE db1;
 step s2-commit: COMMIT;
 step s2-drop-db2: DROP DATABASE db2;
+?column?
+---------------------------------------------------------------------
+       1
+(1 row)
+
diff --git a/src/test/regress/failure_schedule b/src/test/regress/failure_schedule
index e1ad362b518..8b992422ef2 100644
--- a/src/test/regress/failure_schedule
+++ b/src/test/regress/failure_schedule
@@ -35,6 +35,7 @@ test: failure_mx_metadata_sync
 test: failure_mx_metadata_sync_multi_trans
 test: failure_connection_establishment
 test: failure_non_main_db_2pc
+test: failure_create_database
 
 # this test syncs metadata to the workers
 test: failure_failover_to_local_execution
diff --git a/src/test/regress/spec/isolation_database_cmd_from_any_node.spec b/src/test/regress/spec/isolation_database_cmd_from_any_node.spec
index 1e004cb338f..8637a8942b6 100644
--- a/src/test/regress/spec/isolation_database_cmd_from_any_node.spec
+++ b/src/test/regress/spec/isolation_database_cmd_from_any_node.spec
@@ -2,11 +2,15 @@ setup
 {
     -- OCLASS for database changed in PG 16 from 25 to 26
     SELECT CASE WHEN substring(version(), '\d+')::integer < 16 THEN 25 ELSE 26 END AS value INTO oclass_database;
+
+    SELECT 1 FROM citus_add_node('localhost', 57636, 0);
 }
 
 teardown
 {
     DROP TABLE IF EXISTS oclass_database;
+
+    select 1 from citus_remove_node('localhost', 57636);
 }
 
 session "s1"
diff --git a/src/test/regress/sql/alter_database_propagation.sql b/src/test/regress/sql/alter_database_propagation.sql
index 4904919a6a6..9a8b1fab8af 100644
--- a/src/test/regress/sql/alter_database_propagation.sql
+++ b/src/test/regress/sql/alter_database_propagation.sql
@@ -49,6 +49,7 @@ alter database regression set lock_timeout to DEFAULT;
 alter database regression RESET lock_timeout;
 
 set citus.enable_create_database_propagation=on;
+SET citus.next_operation_id TO 3000;
 create database "regression!'2";
 alter database "regression!'2" with CONNECTION LIMIT 100;
 alter database "regression!'2" with IS_TEMPLATE true CONNECTION LIMIT 50;
@@ -90,6 +91,7 @@ set citus.enable_create_database_propagation=on;
 
 drop database regression3;
 
+SET citus.next_operation_id TO 3100;
 create database "regression!'4";
 
 
diff --git a/src/test/regress/sql/create_drop_database_propagation.sql b/src/test/regress/sql/create_drop_database_propagation.sql
index 329f4861203..de55258c3c5 100644
--- a/src/test/regress/sql/create_drop_database_propagation.sql
+++ b/src/test/regress/sql/create_drop_database_propagation.sql
@@ -218,7 +218,8 @@ SELECT * FROM public.check_database_on_all_nodes('my_template_database') ORDER B
 --tests for special characters in database name
 set citus.enable_create_database_propagation=on;
 SET citus.log_remote_commands = true;
-set citus.grep_remote_commands = '%CREATE DATABASE%';
+set citus.grep_remote_commands = '%DATABASE%';
+SET citus.next_operation_id TO 2000;
 
 create database "mydatabase#1'2";
 
@@ -746,6 +747,63 @@ SELECT 1 FROM run_command_on_all_nodes($$REVOKE ALL ON TABLESPACE pg_default FRO
 DROP DATABASE no_createdb;
 DROP USER no_createdb;
 
+-- Test a failure scenario by trying to create a distributed database that
+-- already exists on one of the nodes.
+
+\c - - - :worker_1_port
+CREATE DATABASE "test_\!failure";
+
+\c - - - :master_port
+
+SET citus.enable_create_database_propagation TO ON;
+
+CREATE DATABASE "test_\!failure";
+
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure$$) ORDER BY node_type, result;
+
+SET citus.enable_create_database_propagation TO OFF;
+CREATE DATABASE "test_\!failure1";
+
+\c - - - :worker_1_port
+DROP DATABASE "test_\!failure";
+
+SET citus.enable_create_database_propagation TO ON;
+
+CREATE DATABASE "test_\!failure1";
+
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure1$$) ORDER BY node_type, result;
+
+\c - - - :master_port
+
+-- Before dropping local "test_\!failure1" database, test a failure scenario
+-- by trying to create a distributed database that already exists "on local
+-- node" this time.
+
+SET citus.enable_create_database_propagation TO ON;
+
+CREATE DATABASE "test_\!failure1";
+
+SET client_min_messages TO WARNING;
+CALL citus_cleanup_orphaned_resources();
+RESET client_min_messages;
+
+SELECT result AS database_cleanedup_on_node FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+SELECT * FROM public.check_database_on_all_nodes($$test_\!failure1$$) ORDER BY node_type, result;
+
+SET citus.enable_create_database_propagation TO OFF;
+
+DROP DATABASE "test_\!failure1";
+
 SET citus.enable_create_database_propagation TO ON;
 
 --clean up resources created by this test
diff --git a/src/test/regress/sql/failure_create_database.sql b/src/test/regress/sql/failure_create_database.sql
new file mode 100644
index 00000000000..d117dc81192
--- /dev/null
+++ b/src/test/regress/sql/failure_create_database.sql
@@ -0,0 +1,128 @@
+SET citus.enable_create_database_propagation TO ON;
+SET client_min_messages TO WARNING;
+
+SELECT 1 FROM citus_add_node('localhost', :master_port, 0);
+
+CREATE FUNCTION get_temp_databases_on_nodes()
+RETURNS TEXT AS $func$
+  SELECT array_agg(DISTINCT result ORDER BY result) AS temp_databases_on_nodes FROM run_command_on_all_nodes($$SELECT datname FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$) WHERE result != '';
+$func$
+LANGUAGE sql;
+
+CREATE FUNCTION count_db_cleanup_records()
+RETURNS TABLE(object_name TEXT, count INTEGER) AS $func$
+  SELECT object_name, COUNT(*) FROM pg_dist_cleanup WHERE object_name LIKE 'citus_temp_database_%' GROUP BY object_name;
+$func$
+LANGUAGE sql;
+
+CREATE FUNCTION ensure_no_temp_databases_on_any_nodes()
+RETURNS BOOLEAN AS $func$
+  SELECT bool_and(result::boolean) AS no_temp_databases_on_any_nodes FROM run_command_on_all_nodes($$SELECT COUNT(*)=0 FROM pg_database WHERE datname LIKE 'citus_temp_database_%'$$);
+$func$
+LANGUAGE sql;
+
+-- cleanup any orphaned resources from previous runs
+CALL citus_cleanup_orphaned_resources();
+
+SET citus.next_operation_id TO 4000;
+
+ALTER SYSTEM SET citus.defer_shard_delete_interval TO -1;
+SELECT pg_reload_conf();
+SELECT pg_sleep(0.1);
+
+SELECT citus.mitmproxy('conn.kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^CREATE DATABASE").cancel(' || pg_backend_pid() || ')');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^ALTER DATABASE").cancel(' || pg_backend_pid() || ')');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^PREPARE TRANSACTION").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onQuery(query="^COMMIT PREPARED").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+-- not call citus_cleanup_orphaned_resources() but recover the prepared transactions this time
+SELECT 1 FROM recover_prepared_transactions();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+DROP DATABASE db1;
+
+-- after recovering the prepared transactions, cleanup records should also be removed
+SELECT * FROM count_db_cleanup_records();
+
+SELECT citus.mitmproxy('conn.onQuery(query="^SELECT citus_internal.acquire_citus_advisory_object_class_lock").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+SELECT citus.mitmproxy('conn.onParse(query="^WITH distributed_object_data").kill()');
+CREATE DATABASE db1;
+SELECT citus.mitmproxy('conn.allow()');
+
+SELECT get_temp_databases_on_nodes();
+SELECT * FROM count_db_cleanup_records();
+CALL citus_cleanup_orphaned_resources();
+SELECT ensure_no_temp_databases_on_any_nodes();
+SELECT * FROM public.check_database_on_all_nodes($$db1$$) ORDER BY node_type, result;
+
+CREATE DATABASE db1;
+
+-- show that a successful database creation doesn't leave any pg_dist_cleanup records behind
+SELECT * FROM count_db_cleanup_records();
+
+DROP DATABASE db1;
+
+DROP FUNCTION get_temp_databases_on_nodes();
+DROP FUNCTION ensure_no_temp_databases_on_any_nodes();
+DROP FUNCTION count_db_cleanup_records();
+
+SELECT 1 FROM citus_remove_node('localhost', :master_port);