Fix pg_control check for WAL-E backups (#169)

* Fix pg_control check for WAL-E backups

Currently, WAL-E backup restoration was broken since preprelease 0.2.1+
We were checking for pg_control before understanding that it is not necessary. This commit fixes it.

internal/backup.go

@@ -63,7 +63,7 @@ func (backup *Backup) GetTarNames() ([]string, error) {
 }
 
 // TODO : unit tests
-func (backup *Backup) fetchSentinel() (BackupSentinelDto, error) {
+func (backup *Backup) FetchSentinel() (BackupSentinelDto, error) {
 	sentinelDto := BackupSentinelDto{}
 	backupReaderMaker := NewStorageReaderMaker(backup.BaseBackupFolder, backup.getStopSentinelPath())
 	backupReader, err := backupReaderMaker.Reader()
@@ -120,14 +120,20 @@ func (backup *Backup) unwrap(dbDataDirectory string, sentinelDto BackupSentinelD
 	if err != nil {
 		return err
 	}
+
+	// Check name for backwards compatibility. Will check for `pg_control` if WALG version of backup.
+	needPgControl := IsPgControlRequired(backup, sentinelDto)
+
+	if pgControlKey == "" && needPgControl {
+		return NewPgControlNotFoundError()
+	}
+
 	err = ExtractAll(tarInterpreter, tarsToExtract)
 	if err != nil {
 		return err
 	}
-	// Check name for backwards compatibility. Will check for `pg_control` if WALG version of backup.
-	re := regexp.MustCompile(`^([^_]+._{1}[^_]+._{1})`)
-	match := re.FindString(backup.Name)
-	if match == "" || sentinelDto.isIncremental() {
+
+	if needPgControl {
 		err = ExtractAll(tarInterpreter, []ReaderMaker{NewStorageReaderMaker(backup.getTarPartitionFolder(), pgControlKey)})
 		if err != nil {
 			return errors.Wrap(err, "failed to extract pg_control")
@@ -138,6 +144,13 @@ func (backup *Backup) unwrap(dbDataDirectory string, sentinelDto BackupSentinelD
 	return nil
 }
 
+func IsPgControlRequired(backup *Backup, sentinelDto BackupSentinelDto) bool {
+	re := regexp.MustCompile(`^([^_]+._{1}[^_]+._{1})`)
+	walgBasebackupName := re.FindString(backup.Name) == ""
+	needPgControl := walgBasebackupName || sentinelDto.isIncremental()
+	return needPgControl
+}
+
 // TODO : unit tests
 func IsDirectoryEmpty(directoryPath string) (bool, error) {
 	var isEmpty = true
@@ -178,8 +191,5 @@ func (backup *Backup) getTarsToExtract() (tarsToExtract []ReaderMaker, pgControl
 		tarToExtract := NewStorageReaderMaker(backup.getTarPartitionFolder(), tarName)
 		tarsToExtract = append(tarsToExtract, tarToExtract)
 	}
-	if pgControlKey == "" {
-		return nil, "", NewPgControlNotFoundError()
-	}
 	return
 }

internal/backup_fetch_handler.go

@@ -111,7 +111,7 @@ func deltaFetchRecursion(backupName string, folder StorageFolder, dbDataDirector
 	if err != nil {
 		return err
 	}
-	sentinelDto, err := backup.fetchSentinel()
+	sentinelDto, err := backup.FetchSentinel()
 	if err != nil {
 		return err
 	}

internal/backup_push_handler.go

@@ -54,7 +54,7 @@ func HandleBackupPush(archiveDirectory string, uploader *Uploader) {
 			}
 		} else {
 			previousBackup := NewBackup(basebackupFolder, previousBackupName)
-			previousBackupSentinelDto, err = previousBackup.fetchSentinel()
+			previousBackupSentinelDto, err = previousBackup.FetchSentinel()
 			if err != nil {
 				tracelog.ErrorLogger.FatalError(err)
 			}
@@ -72,7 +72,7 @@ func HandleBackupPush(archiveDirectory string, uploader *Uploader) {
 					tracelog.InfoLogger.Println("Delta will be made from full backup.")
 					previousBackupName = *previousBackupSentinelDto.IncrementFullName
 					previousBackup := NewBackup(basebackupFolder, previousBackupName)
-					previousBackupSentinelDto, err = previousBackup.fetchSentinel()
+					previousBackupSentinelDto, err = previousBackup.FetchSentinel()
 					if err != nil {
 						tracelog.ErrorLogger.FatalError(err)
 					}

internal/delete_handler.go

@@ -29,7 +29,7 @@ type DeleteCommandArguments struct {
 
 // TODO : unit tests
 func adjustDeleteTarget(target *Backup, findFull bool) *Backup {
-	sentinelDto, err := target.fetchSentinel()
+	sentinelDto, err := target.FetchSentinel()
 	if err != nil {
 		tracelog.ErrorLogger.FatalError(err)
 	}
@@ -87,7 +87,7 @@ func HandleDelete(folder StorageFolder, args []string) {
 					return
 				}
 				backup := NewBackup(baseBackupFolder, b.BackupName)
-				dto, err := backup.fetchSentinel()
+				dto, err := backup.FetchSentinel()
 				if err != nil {
 					tracelog.ErrorLogger.FatalError(err)
 				}

test/backup_fetch_test.go

@@ -5,14 +5,15 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/wal-g/wal-g/internal"
 	"github.com/wal-g/wal-g/testtools"
+	"strings"
 	"testing"
 )
 
 func createMockStorageFolder() internal.StorageFolder {
 	var folder = testtools.MakeDefaultInMemoryStorageFolder()
 	subFolder := folder.GetSubFolder(internal.BaseBackupPath)
 	subFolder.PutObject("base_123_backup_stop_sentinel.json", &bytes.Buffer{})
-	subFolder.PutObject("base_456_backup_stop_sentinel.json", &bytes.Buffer{})
+	subFolder.PutObject("base_456_backup_stop_sentinel.json", strings.NewReader("{}"))
 	subFolder.PutObject("base_000_backup_stop_sentinel.json", &bytes.Buffer{}) // last put
 	subFolder.PutObject("base_123312", &bytes.Buffer{})                        // not a sentinel
 	subFolder.PutObject("base_321/nop", &bytes.Buffer{})

test/backup_test.go

@@ -29,3 +29,17 @@ func TestGetTarNames(t *testing.T) {
 	assert.NoError(t, err)
 	assert.ElementsMatch(t, []string{"1", "2", "3"}, tarNames)
 }
+
+func TestIsPgControlRequired(t *testing.T) {
+	folder := createMockStorageFolder()
+	backup := internal.NewBackup(folder.GetSubFolder(internal.BaseBackupPath), "base_456")
+	dto, err := backup.FetchSentinel()
+	assert.NoError(t, err)
+	assert.True(t, internal.IsPgControlRequired(backup, dto))
+}
+
+func TestIsPgControlNotRequiredForWALEBackups(t *testing.T) {
+	folder := createMockStorageFolder()
+	backup := internal.NewBackup(folder.GetSubFolder(internal.BaseBackupPath), "base_000000010000DD170000000C_00743784")
+	assert.False(t, internal.IsPgControlRequired(backup, internal.BackupSentinelDto{}))
+}