check links to parent backup in validate

Author: alubennikova

src/catalog.c

@@ -714,3 +714,48 @@ pgBackupGetPath2(const pgBackup *backup, char *path, size_t len,
 
 	make_native_path(path);
 }
+
+/* Find parent base FULL backup for current backup using parent_backup_link,
+ * return NULL if not found
+ */
+pgBackup*
+find_parent_backup(pgBackup *current_backup)
+{
+	pgBackup   *base_full_backup = NULL;
+	base_full_backup = current_backup;
+
+	while (base_full_backup->backup_mode != BACKUP_MODE_FULL)
+	{
+		/*
+		 * If we haven't found parent for incremental backup,
+		 * mark it and all depending backups as orphaned
+		 */
+		if (base_full_backup->parent_backup_link == NULL
+			|| (base_full_backup->status != BACKUP_STATUS_OK
+				&& base_full_backup->status != BACKUP_STATUS_DONE))
+		{
+			pgBackup   *orphaned_backup = current_backup;
+
+			while (orphaned_backup != NULL)
+			{
+				orphaned_backup->status = BACKUP_STATUS_ORPHAN;
+				pgBackupWriteBackupControlFile(orphaned_backup);
+				if (base_full_backup->parent_backup_link == NULL)
+					elog(WARNING, "Backup %s is orphaned because its parent backup is not found",
+							base36enc(orphaned_backup->start_time));
+				else
+					elog(WARNING, "Backup %s is orphaned because its parent backup is corrupted",
+							base36enc(orphaned_backup->start_time));
+
+				orphaned_backup = orphaned_backup->parent_backup_link;
+			}
+
+			base_full_backup = NULL;
+			break;
+		}
+
+		base_full_backup = base_full_backup->parent_backup_link;
+	}
+
+	return base_full_backup;
+}

src/pg_probackup.h

@@ -443,6 +443,8 @@ extern void pgBackupFree(void *backup);
 extern int pgBackupCompareId(const void *f1, const void *f2);
 extern int pgBackupCompareIdDesc(const void *f1, const void *f2);
 
+extern pgBackup* find_parent_backup(pgBackup *current_backup);
+
 /* in dir.c */
 extern void dir_list_file(parray *files, const char *root, bool exclude,
 						  bool omit_symlink, bool add_root);

src/restore.c

@@ -202,41 +202,40 @@ do_restore_or_validate(time_t target_backup_id,
 			dest_backup_index = i;
 		}
 
-		/* If we already found dest_backup, look for full backup. */
-		/* TODO Now, as we have all backups linked, we can probably get rid of that?"*/
-		if (dest_backup)
+	}
+
+	/* If we already found dest_backup, look for full backup. */
+	if (dest_backup)
+	{
+		base_full_backup = current_backup;
+
+		if (current_backup->backup_mode != BACKUP_MODE_FULL)
+		{
+			base_full_backup = find_parent_backup(current_backup);
+
+			if (base_full_backup == NULL)
+				elog(ERROR, "Valid full backup for backup %s is not found.",
+					base36enc(current_backup->start_time));
+		}
+
+		/*
+		 * We have found full backup by link,
+		 * now we need to walk the list to find its index.
+		 *
+		 * TODO I think we should rewrite it someday to use double linked list
+		 * and avoid relying on sort order anymore.
+		 */
+		for (i = dest_backup_index; i < parray_num(backups); i++)
 		{
-			if (current_backup->backup_mode == BACKUP_MODE_FULL)
+			pgBackup * temp_backup = (pgBackup *) parray_get(backups, i);
+			if (temp_backup->start_time == base_full_backup->start_time)
 			{
-				if (current_backup->status != BACKUP_STATUS_OK)
-				{
-					/* Full backup revalidation can be done only for DONE and CORRUPT */
-					if (current_backup->status == BACKUP_STATUS_DONE ||
-							current_backup->status == BACKUP_STATUS_CORRUPT)
-						elog(WARNING, "base backup %s for given backup %s is in %s status, trying to revalidate",
-							base36enc_dup(current_backup->start_time),
-							base36enc_dup(dest_backup->start_time),
-							status2str(current_backup->status));
-					else
-						elog(ERROR, "base backup %s for given backup %s is in %s status",
-							base36enc_dup(current_backup->start_time),
-							base36enc_dup(dest_backup->start_time),
-							status2str(current_backup->status));
-				}
-				/* We found both dest and base backups. */
-				base_full_backup = current_backup;
 				base_full_backup_index = i;
 				break;
 			}
-			else
-				/* It`s ok to skip incremental backup */
-				continue;
 		}
 	}
 
-	if (base_full_backup == NULL)
-		elog(ERROR, "Full backup satisfying target options is not found.");
-
 	/*
 	 * Ensure that directories provided in tablespace mapping are valid
 	 * i.e. empty or not exist.

src/validate.c

@@ -282,44 +282,33 @@ do_validate_instance(void)
 	if (backups == NULL)
 		elog(ERROR, "Failed to get backup list.");
 
-	/* Valiate each backup along with its xlog files. */
-	/* TODO Maybe use parent_backup_link instead of looking for backups in the list */
+	/* Examine backups one by one and validate them */
 	for (i = 0; i < parray_num(backups); i++)
 	{
-		pgBackup   *base_full_backup = NULL;
-
 		current_backup = (pgBackup *) parray_get(backups, i);
 
-		if (current_backup->backup_mode != BACKUP_MODE_FULL)
+		/* Valiate each backup along with its xlog files. */
+		pgBackupValidate(current_backup);
+
+		/* Ensure that the backup has valid list of parent backups */
+		if (current_backup->status == BACKUP_STATUS_OK)
 		{
-			int			j;
+			pgBackup   *base_full_backup = current_backup;
 
-			for (j = i + 1; j < parray_num(backups); j++)
+			if (current_backup->backup_mode != BACKUP_MODE_FULL)
 			{
-				pgBackup	   *backup = (pgBackup *) parray_get(backups, j);
+				base_full_backup = find_parent_backup(current_backup);
 
-				if (backup->backup_mode == BACKUP_MODE_FULL)
-				{
-					base_full_backup = backup;
-					break;
-				}
+				if (base_full_backup == NULL)
+					elog(ERROR, "Valid full backup for backup %s is not found.",
+						 base36enc(current_backup->start_time));
 			}
-		}
-		else
-			base_full_backup = current_backup;
-
-		pgBackupValidate(current_backup);
 
-		/* There is no point in wal validation for corrupted backup */
-		if (current_backup->status == BACKUP_STATUS_OK)
-		{
-			if (base_full_backup == NULL)
-				elog(ERROR, "Valid full backup for backup %s is not found.",
-					 base36enc(current_backup->start_time));
 			/* Validate corresponding WAL files */
 			validate_wal(current_backup, arclog_path, 0,
-						0, base_full_backup->tli);
+							0, base_full_backup->tli);
 		}
+
 		/* Mark every incremental backup between corrupted backup and nearest FULL backup as orphans */
 		if (current_backup->status == BACKUP_STATUS_CORRUPT)
 		{