Add more LOG messages when starting and ending recovery from a backup

Three LOG messages are added in the recovery code paths, providing
information that can be useful to track corruption issues depending on
the state of the cluster, telling that:
- Recovery has started from a backup_label.
- Recovery is restarting from a backup start LSN, without a
backup_label.
- Recovery has completed from a backup.

This was originally applied on HEAD as of 1d35f705e191, and there is
consensus that this can be useful for older versions.  This applies
cleanly down to 15, so do it down to this version for now (older
versions have heavily refactored the WAL recovery paths, making the
change less straight-forward to do).

Author: Andres Freund
Reviewed-by: David Steele, Laurenz Albe, Michael Paquier
Discussion: https://postgr.es/m/20231117041811.vz4vgkthwjnwp2pp@awork3.anarazel.de
Backpatch-through: 15

src/backend/access/transam/xlogrecovery.c

@@ -623,6 +623,22 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,
 		if (StandbyModeRequested)
 			EnableStandbyMode();
 
+		/*
+		 * Omitting backup_label when creating a new replica, PITR node etc.
+		 * unfortunately is a common cause of corruption.  Logging that
+		 * backup_label was used makes it a bit easier to exclude that as the
+		 * cause of observed corruption.
+		 *
+		 * Do so before we try to read the checkpoint record (which can fail),
+		 * as otherwise it can be hard to understand why a checkpoint other
+		 * than ControlFile->checkPoint is used.
+		 */
+		ereport(LOG,
+				(errmsg("starting backup recovery with redo LSN %X/%X, checkpoint LSN %X/%X, on timeline ID %u",
+						LSN_FORMAT_ARGS(RedoStartLSN),
+						LSN_FORMAT_ARGS(CheckPointLoc),
+						CheckPointTLI)));
+
 		/*
 		 * When a backup_label file is present, we want to roll forward from
 		 * the checkpoint it identifies, rather than using pg_control.
@@ -761,6 +777,16 @@ InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,
 				EnableStandbyMode();
 		}
 
+		/*
+		 * For the same reason as when starting up with backup_label present,
+		 * emit a log message when we continue initializing from a base
+		 * backup.
+		 */
+		if (!XLogRecPtrIsInvalid(ControlFile->backupStartPoint))
+			ereport(LOG,
+					(errmsg("restarting backup recovery with redo LSN %X/%X",
+							LSN_FORMAT_ARGS(ControlFile->backupStartPoint))));
+
 		/* Get the last valid checkpoint record. */
 		CheckPointLoc = ControlFile->checkPoint;
 		CheckPointTLI = ControlFile->checkPointCopy.ThisTimeLineID;
@@ -2123,6 +2149,9 @@ CheckRecoveryConsistency(void)
 	if (!XLogRecPtrIsInvalid(backupEndPoint) &&
 		backupEndPoint <= lastReplayedEndRecPtr)
 	{
+		XLogRecPtr	saveBackupStartPoint = backupStartPoint;
+		XLogRecPtr	saveBackupEndPoint = backupEndPoint;
+
 		elog(DEBUG1, "end of backup reached");
 
 		/*
@@ -2133,6 +2162,11 @@ CheckRecoveryConsistency(void)
 		backupStartPoint = InvalidXLogRecPtr;
 		backupEndPoint = InvalidXLogRecPtr;
 		backupEndRequired = false;
+
+		ereport(LOG,
+				(errmsg("completed backup recovery with redo LSN %X/%X and end LSN %X/%X",
+						LSN_FORMAT_ARGS(saveBackupStartPoint),
+						LSN_FORMAT_ARGS(saveBackupEndPoint))));
 	}
 
 	/*

src/backend/commands/user.c

@@ -1093,7 +1093,7 @@ DropRole(DropRoleStmt *stmt)
 	Relation	pg_authid_rel,
 				pg_auth_members_rel;
 	ListCell   *item;
-	List	   *role_addresses = NIL;
+	List	   *role_oids = NIL;
 
 	if (!have_createrole_privilege())
 		ereport(ERROR,
@@ -1119,7 +1119,6 @@ DropRole(DropRoleStmt *stmt)
 		ScanKeyData scankey;
 		SysScanDesc sscan;
 		Oid			roleid;
-		ObjectAddress *role_address;
 
 		if (rolspec->roletype != ROLESPEC_CSTRING)
 			ereport(ERROR,
@@ -1260,21 +1259,16 @@ DropRole(DropRoleStmt *stmt)
 		 */
 		CommandCounterIncrement();
 
-		/* Looks tentatively OK, add it to the list. */
+		/* Looks tentatively OK, add it to the list if not there yet. */
-		role_address = palloc(sizeof(ObjectAddress));
+		role_oids = list_append_unique_oid(role_oids, roleid);
-		role_address->classId = AuthIdRelationId;
-		role_address->objectId = roleid;
-		role_address->objectSubId = 0;
-		role_addresses = lappend(role_addresses, role_address);
 	}
 
 	/*
 	 * Second pass over the roles to be removed.
 	 */
-	foreach(item, role_addresses)
+	foreach(item, role_oids)
 	{
-		ObjectAddress *role_address = lfirst(item);
+		Oid			roleid = lfirst_oid(item);
-		Oid			roleid = role_address->objectId;
 		HeapTuple	tuple;
 		Form_pg_authid roleform;
 		char	   *detail;

src/test/regress/expected/create_role.out

@@ -251,7 +251,8 @@ DROP INDEX tenant_idx;
 DROP TABLE tenant_table;
 DROP VIEW tenant_view;
 DROP SCHEMA regress_tenant2_schema;
-DROP ROLE regress_tenant;
+-- check for duplicated drop
+DROP ROLE regress_tenant, regress_tenant;
 DROP ROLE regress_tenant2;
 DROP ROLE regress_rolecreator;
 DROP ROLE regress_role_admin;

src/test/regress/sql/create_role.sql

@@ -206,7 +206,8 @@ DROP INDEX tenant_idx;
 DROP TABLE tenant_table;
 DROP VIEW tenant_view;
 DROP SCHEMA regress_tenant2_schema;
-DROP ROLE regress_tenant;
+-- check for duplicated drop
+DROP ROLE regress_tenant, regress_tenant;
 DROP ROLE regress_tenant2;
 DROP ROLE regress_rolecreator;
 DROP ROLE regress_role_admin;