vaultik/internal/vaultik/restore_skip_errors_test.go

package vaultik_test

import (
	"bytes"
	"context"
	"fmt"
	"path/filepath"
	"testing"

	"github.com/spf13/afero"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"sneak.berlin/go/vaultik/internal/config"
	"sneak.berlin/go/vaultik/internal/database"
	"sneak.berlin/go/vaultik/internal/log"
	"sneak.berlin/go/vaultik/internal/storage"
	"sneak.berlin/go/vaultik/internal/ui"
	"sneak.berlin/go/vaultik/internal/vaultik"
)

// A file no larger than the chunker's minimum chunk size (a quarter of
// faultChunkSize) is stored as one chunk, so it lives in exactly one blob.
// More of them than fit in faultMaxBlobSize make the snapshot span two
// blobs.
const (
	missingBlobFileBytes = int(faultChunkSize / 4)
	missingBlobFileCount = 20
)

// missingBlobBackup is a snapshot of single-chunk files spread over two
// blobs, with one of those blobs deleted from the store.
type missingBlobBackup struct {
	fs         afero.Fs
	cfg        *config.Config
	storer     storage.Storer
	snapshotID string
	restoreDir string
	// files holds the original content by source path.
	files map[string][]byte
	// lost holds the source paths whose chunk is in the deleted blob.
	lost map[string]bool
}

// TestRestoreSkipErrorsSkipsFilesOfMissingBlob restores with SkipErrors
// after one blob of a two-blob snapshot was deleted. Every file stored in
// that blob must be reported as failed and left absent, every other file
// must be restored intact, and Restore must still return an error.
func TestRestoreSkipErrorsSkipsFilesOfMissingBlob(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	ctx := context.Background()
	backup := backupThenDeleteOneBlob(ctx, t)

	var out bytes.Buffer

	v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, backup.fs)
	v.UI = ui.NewWithColor(&out, false)

	err := v.Restore(&vaultik.RestoreOptions{
		SnapshotID: backup.snapshotID,
		TargetDir:  backup.restoreDir,
		SkipErrors: true,
	})

	require.Error(t, err, "restore must fail when files were skipped")
	assert.Contains(t, err.Error(),
		fmt.Sprintf("%d file(s) failed to restore", len(backup.lost)))

	for path, content := range backup.files {
		restored := filepath.Join(backup.restoreDir, path)

		if backup.lost[path] {
			assert.Containsf(t, out.String(), path,
				"%s needs the deleted blob and must be reported", path)
			assert.NoFileExists(t, restored)

			continue
		}

		got, err := afero.ReadFile(backup.fs, restored)
		require.NoErrorf(t, err, "%s does not need the deleted blob", path)
		assert.Equalf(t, content, got, "%s restored with wrong content", path)
	}
}

// TestRestoreSkipErrorsVerifyChecksOnlyRestoredFiles restores with
// SkipErrors and Verify after one blob of a two-blob snapshot was
// deleted. Verification must check only the files that were written, so
// a file of the deleted blob is reported as failed to restore and not
// also as failing verification, and Restore must still return an error.
func TestRestoreSkipErrorsVerifyChecksOnlyRestoredFiles(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	ctx := context.Background()
	backup := backupThenDeleteOneBlob(ctx, t)

	var out bytes.Buffer

	v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, backup.fs)
	v.UI = ui.NewWithColor(&out, false)

	err := v.Restore(&vaultik.RestoreOptions{
		SnapshotID: backup.snapshotID,
		TargetDir:  backup.restoreDir,
		Verify:     true,
		SkipErrors: true,
	})

	require.Error(t, err, "restore must fail when files were skipped")
	assert.Contains(t, err.Error(),
		fmt.Sprintf("%d file(s) failed to restore", len(backup.lost)))

	written := len(backup.files) - len(backup.lost)

	assert.Contains(t, out.String(), fmt.Sprintf("Verifying %d files", written))
	assert.Contains(t, out.String(), fmt.Sprintf("Verified %d files", written))
	assert.NotContains(t, out.String(), "Verification failed")
}

// TestRestoreMissingBlobAbortsWithoutSkipErrors checks that a deleted blob
// still ends the restore with an error when SkipErrors is not set.
func TestRestoreMissingBlobAbortsWithoutSkipErrors(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	ctx := context.Background()
	backup := backupThenDeleteOneBlob(ctx, t)

	v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, backup.fs)

	err := v.Restore(&vaultik.RestoreOptions{
		SnapshotID: backup.snapshotID,
		TargetDir:  backup.restoreDir,
	})

	require.ErrorIs(t, err, storage.ErrNotFound)
}

// backupThenDeleteOneBlob backs up missingBlobFileCount single-chunk
// files, then deletes from the store the blob holding the first of them.
func backupThenDeleteOneBlob(
	ctx context.Context, t *testing.T,
) *missingBlobBackup {
	t.Helper()

	fs := afero.NewOsFs()
	tempDir := t.TempDir()
	dataDir := filepath.Join(tempDir, "src")
	dbPath := filepath.Join(tempDir, "index.sqlite")
	cfg := faultTestConfig()

	require.NoError(t, fs.MkdirAll(dataDir, 0o755))

	files := make(map[string][]byte, missingBlobFileCount)

	for i := range missingBlobFileCount {
		path := filepath.Join(dataDir, fmt.Sprintf("file-%02d.bin", i))
		files[path] = bytesPattern(
			fmt.Sprintf("file-%02d-", i), missingBlobFileBytes)
		require.NoError(t, afero.WriteFile(fs, path, files[path], 0o644))
	}

	storer, err := storage.NewFileStorer(filepath.Join(tempDir, "remote"))
	require.NoError(t, err)

	db, err := database.New(ctx, dbPath)
	require.NoError(t, err)

	repos := database.NewRepositories(db)

	id := fullFaultBackup(
		ctx, t, fs, storer, cfg, repos, dataDir, dbPath, "missingblob")

	blobOfFile := make(map[string]string, len(files))
	for path := range files {
		blobOfFile[path] = blobHashOfSingleChunkFile(ctx, t, repos, path)
	}

	require.NoError(t, db.Close())

	deleted := blobOfFile[filepath.Join(dataDir, "file-00.bin")]
	require.NoError(t, storer.Delete(ctx, fmt.Sprintf(
		"blobs/%s/%s/%s", deleted[:2], deleted[2:4], deleted)))

	lost := make(map[string]bool)

	for path, hash := range blobOfFile {
		if hash == deleted {
			lost[path] = true
		}
	}

	require.Less(t, len(lost), len(files),
		"the snapshot must span more than one blob")

	return &missingBlobBackup{
		fs:         fs,
		cfg:        cfg,
		storer:     storer,
		snapshotID: id,
		restoreDir: filepath.Join(tempDir, "restored"),
		files:      files,
		lost:       lost,
	}
}

// blobHashOfSingleChunkFile returns the hash of the blob holding the one
// chunk of the file at path, as recorded in the local index.
func blobHashOfSingleChunkFile(
	ctx context.Context, t *testing.T,
	repos *database.Repositories, path string,
) string {
	t.Helper()

	chunks, err := repos.FileChunks.GetByPath(ctx, path)
	require.NoError(t, err)
	require.Lenf(t, chunks, 1, "%s must be a single chunk", path)

	blobChunk, err := repos.BlobChunks.GetByChunkHash(
		ctx, chunks[0].ChunkHash.String())
	require.NoError(t, err)
	require.NotNilf(t, blobChunk, "chunk of %s is in no blob", path)

	blob, err := repos.Blobs.GetByID(ctx, blobChunk.BlobID.String())
	require.NoError(t, err)

	return blob.Hash.String()
}