PGPRO-1248: new status ORPHAN for backup with corrupted parent

Author: gsmolk

src/catalog.c

@@ -556,6 +556,8 @@ readBackupControlFile(const char *path)
 			backup->status = BACKUP_STATUS_DELETED;
 		else if (strcmp(status, "DONE") == 0)
 			backup->status = BACKUP_STATUS_DONE;
+		else if (strcmp(status, "ORPHAN") == 0)
+			backup->status = BACKUP_STATUS_ORPHAN;
 		else if (strcmp(status, "CORRUPT") == 0)
 			backup->status = BACKUP_STATUS_CORRUPT;
 		else

src/parsexlog.c

@@ -258,7 +258,9 @@ validate_backup_wal_from_start_to_stop(pgBackup *backup,
 		 */
 			backup->status = BACKUP_STATUS_CORRUPT;
 			pgBackupWriteBackupControlFile(backup);
-			elog(ERROR, "there are not enough WAL records to restore from %X/%X to %X/%X",
+			elog(WARNING, "There are not enough WAL records to consistenly restore "
+				"backup %s from START LSN: %X/%X to STOP LSN: %X/%X",
+				 base36enc(backup->start_time),
 				 (uint32) (backup->start_lsn >> 32),
 				 (uint32) (backup->start_lsn),
 				 (uint32) (backup->stop_lsn >> 32),
@@ -329,14 +331,19 @@ validate_wal(pgBackup *backup,
 
 	free(backup_id);
 
+	if (backup->status == BACKUP_STATUS_CORRUPT)
+	{
+		elog(WARNING, "Backup %s WAL segments are corrupted", base36enc(backup->start_time));
+		return;
+	}
 	/*
 	 * If recovery target is provided check that we can restore backup to a
-	 * recoverty target time or xid.
+	 * recovery target time or xid.
 	 */
 	if (!TransactionIdIsValid(target_xid) && target_time == 0)
 	{
-		/* Recoverty target is not given so exit */
-		elog(INFO, "backup validation completed successfully");
+		/* Recovery target is not given so exit */
+		elog(INFO, "Backup %s WAL segments are valid", base36enc(backup->start_time));
 		return;
 	}
 

src/pg_probackup.h

@@ -118,6 +118,7 @@ typedef enum BackupStatus
 	BACKUP_STATUS_DELETING,		/* data files are being deleted */
 	BACKUP_STATUS_DELETED,		/* data files have been deleted */
 	BACKUP_STATUS_DONE,			/* completed but not validated yet */
+	BACKUP_STATUS_ORPHAN,		/* backup validity is unknown but at least one parent backup is corrupted */
 	BACKUP_STATUS_CORRUPT		/* files are corrupted, not available */
 } BackupStatus;
 

src/restore.c

@@ -90,8 +90,10 @@ do_restore_or_validate(time_t target_backup_id,
 	pgBackup   *current_backup = NULL;
 	pgBackup   *dest_backup = NULL;
 	pgBackup   *base_full_backup = NULL;
+	pgBackup   *corrupted_backup = NULL;
 	int			dest_backup_index = 0;
 	int			base_full_backup_index = 0;
+	int			corrupted_backup_index = 0;
 	char 	   *action = is_restore ? "Restore":"Validate";
 
 	if (is_restore)
@@ -205,7 +207,7 @@ do_restore_or_validate(time_t target_backup_id,
 				}
 			}
 			else
-				/* Skip differential backups are ok */
+				/* It`s ok to skip incremental backup */
 				continue;
 		}
 	}
@@ -220,36 +222,77 @@ do_restore_or_validate(time_t target_backup_id,
 	if (is_restore)
 		check_tablespace_mapping(dest_backup);
 
+	if (dest_backup->backup_mode != BACKUP_MODE_FULL)
+		elog(INFO, "Validating parents for backup %s", base36enc(dest_backup->start_time));
+
 	/*
 	 * Validate backups from base_full_backup to dest_backup.
 	 */
 	for (i = base_full_backup_index; i >= dest_backup_index; i--)
 	{
 		pgBackup   *backup = (pgBackup *) parray_get(backups, i);
 		pgBackupValidate(backup);
+		if (backup->status == BACKUP_STATUS_CORRUPT)
+		{
+			corrupted_backup = backup;
+			corrupted_backup_index = i;
+			break;
+		}
+	}
+	/* There is no point in wal validation
+	 * if there is corrupted backup between base_backup and dest_backup
+	 */
+	if (!corrupted_backup)
+		/*
+		 * Validate corresponding WAL files.
+		 * We pass base_full_backup timeline as last argument to this function,
+		 * because it's needed to form the name of xlog file.
+		 */
+		validate_wal(dest_backup, arclog_path, rt->recovery_target_time,
+					rt->recovery_target_xid, base_full_backup->tli);
+
+	/* Set every incremental backup between corrupted backup and nearest FULL backup as orphans */
+	if (corrupted_backup)
+	{
+		for (i = corrupted_backup_index - 1; i >= 0; i--)
+		{
+			pgBackup   *backup = (pgBackup *) parray_get(backups, i);
+			/* Mark incremental OK backup as orphan */
+			if (backup->backup_mode == BACKUP_MODE_FULL)
+				break;
+			if (backup->status != BACKUP_STATUS_OK)
+				continue;
+			else
+			{
+				backup->status = BACKUP_STATUS_ORPHAN;
+				pgBackupWriteBackupControlFile(backup);
+				elog(WARNING, "Backup %s is orphaned because his parent %s is corrupted",
+						base36enc(backup->start_time), base36enc(corrupted_backup->start_time));
+			}
+		}
 	}
 
 	/*
-	 * Validate corresponding WAL files.
-	 * We pass base_full_backup timeline as last argument to this function,
-	 * because it's needed to form the name of xlog file.
+	 * If dest backup is corrupted or was orphaned in previous check
+	 * produce corresponding error message
 	 */
-	validate_wal(dest_backup, arclog_path, rt->recovery_target_time,
-						 rt->recovery_target_xid, base_full_backup->tli);
-
+	if (dest_backup->status == BACKUP_STATUS_OK)
+		elog(INFO, "Backup %s is valid.", base36enc(dest_backup->start_time));
+	else if (dest_backup->status == BACKUP_STATUS_CORRUPT)
+		elog(ERROR, "Backup %s is corrupt.", base36enc(dest_backup->start_time));
+	else if (dest_backup->status == BACKUP_STATUS_ORPHAN)
+		elog(ERROR, "Backup %s is orphan.", base36enc(dest_backup->start_time));
+	else
+		elog(ERROR, "Backup %s has status: %s",
+				base36enc(dest_backup->start_time), status2str(dest_backup->status));
 
 	/* We ensured that all backups are valid, now restore if required */
 	if (is_restore)
 	{
-		pgBackup   *backup;
 		for (i = base_full_backup_index; i >= dest_backup_index; i--)
 		{
-			backup = (pgBackup *) parray_get(backups, i);
-			if (backup->status == BACKUP_STATUS_OK)
-				restore_backup(backup);
-			else
-				elog(ERROR, "backup %s is not valid",
-					 base36enc(backup->start_time));
+			pgBackup   *backup = (pgBackup *) parray_get(backups, i);
+			restore_backup(backup);
 		}
 
 		/*

src/util.c

@@ -230,9 +230,9 @@ status2str(BackupStatus status)
 		"DELETING",
 		"DELETED",
 		"DONE",
+		"ORPHAN",
 		"CORRUPT"
 	};
-
 	if (status < BACKUP_STATUS_INVALID || BACKUP_STATUS_CORRUPT < status)
 		return "UNKNOWN";
 

src/validate.c

@@ -97,9 +97,9 @@ pgBackupValidate(pgBackup *backup)
 	pgBackupWriteBackupControlFile(backup);
 
 	if (corrupted)
-		elog(WARNING, "Backup %s is corrupted", backup_id_string);
+		elog(WARNING, "Backup %s data files are corrupted", backup_id_string);
 	else
-		elog(INFO, "Backup %s is valid", backup_id_string);
+		elog(INFO, "Backup %s data files are valid", backup_id_string);
 	free(backup_id_string);
 }
 
@@ -269,8 +269,6 @@ do_validate_instance(void)
 		current_backup = (pgBackup *) parray_get(backups, i);
 		backup_id = base36enc(current_backup->start_time);
 
-		elog(INFO, "Validate backup %s", backup_id);
-
 		if (current_backup->backup_mode != BACKUP_MODE_FULL)
 		{
 			int			j;
@@ -301,9 +299,27 @@ do_validate_instance(void)
 			validate_wal(current_backup, arclog_path, 0,
 						0, base_full_backup->tli);
 		}
-
+		/* Set every incremental backup between corrupted backup and nearest FULL backup as orphans */
 		if (current_backup->status != BACKUP_STATUS_OK)
+		{
+			int			j;
 			corrupted_backup_found = true;
+			for (j = i - 1; j >= 0; j--)
+			{
+				pgBackup   *backup = (pgBackup *) parray_get(backups, j);
+				if (backup->backup_mode == BACKUP_MODE_FULL)
+					break;
+				if (backup->status != BACKUP_STATUS_OK)
+					continue;
+				else
+				{
+					backup->status = BACKUP_STATUS_ORPHAN;
+					pgBackupWriteBackupControlFile(backup);
+					elog(WARNING, "Backup %s is orphaned because his parent %s is corrupted",
+							base36enc(backup->start_time), base36enc(current_backup->start_time));
+				}
+			}
+		}
 
 		free(backup_id);
 	}