vaultik/internal/vaultik/restore_skip_errors_test.go

package vaultik_test

import (
	"bytes"
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"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 lost: deleted from the store, or missing
// from the snapshot metadata.
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 lost blob.
	lost map[string]bool
}

// TestRestoreSkipErrorsSkipsFilesOfMissingBlob restores with SkipErrors
// after one blob of a two-blob snapshot was deleted from the store.
func TestRestoreSkipErrorsSkipsFilesOfMissingBlob(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	ctx := context.Background()
	assertRestoreSkipsLostFiles(ctx, t, backupThenDeleteOneBlob(ctx, t))
}

// TestRestoreSkipErrorsSkipsFilesWithChunkInNoBlob restores with
// SkipErrors a snapshot whose metadata lists chunks that no blob holds, as
// earlier backup defects left behind.
func TestRestoreSkipErrorsSkipsFilesWithChunkInNoBlob(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	ctx := context.Background()
	assertRestoreSkipsLostFiles(ctx, t, backupThenDropOneBlobFromMetadata(ctx, t))
}

// assertRestoreSkipsLostFiles restores backup with SkipErrors. Every file
// stored in the lost blob must be reported as failed and left absent,
// every other file must be restored intact, and Restore must still return
// an error.
func assertRestoreSkipsLostFiles(
	ctx context.Context, t *testing.T, backup *missingBlobBackup,
) {
	t.Helper()

	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 lost 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 lost 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")
}

// TestRestoreSkipErrorsVerifyReportsFailuresApart restores with
// SkipErrors and Verify after one blob of a two-blob snapshot was
// deleted, while one file that does restore is written one byte too long.
// The restore warning must list only the files of the deleted blob, the
// verification summary only the long file, each path once, and Restore
// must return an error.
func TestRestoreSkipErrorsVerifyReportsFailuresApart(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

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

	var long string

	for path := range backup.files {
		if !backup.lost[path] {
			long = path

			break
		}
	}

	var out bytes.Buffer

	fs := trailingByteFs{Fs: backup.fs, suffix: long}
	v := newReaderVaultik(ctx, backup.cfg, backup.storer, nil, 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 failed")

	_, afterWarning, found := strings.Cut(out.String(),
		fmt.Sprintf("%d file(s) failed to restore:", len(backup.lost)))
	require.Truef(t, found, "no restore warning in:\n%s", out.String())

	warning, summary, found := strings.Cut(afterWarning,
		"Verification failed: 1 files did not match")
	require.Truef(t, found, "no verification summary in:\n%s", out.String())

	for path := range backup.lost {
		assert.Equalf(t, 1, strings.Count(warning, path),
			"%s must be listed once as failed to restore", path)
		assert.NotContainsf(t, summary, path,
			"%s was not restored and must not fail verification", path)
	}

	assert.NotContainsf(t, warning, long,
		"%s was restored and must not be listed as failed to restore", long)
	assert.Equalf(t, 1, strings.Count(summary, long),
		"%s must be listed once as failing verification", long)
}

// trailingByteFs appends one byte, on close, to the file it opens whose
// path ends in suffix, so that file is restored one byte too long and
// fails verification.
type trailingByteFs struct {
	afero.Fs

	suffix string
}

//nolint:ireturn // afero.Fs.OpenFile is defined to return the interface
func (fs trailingByteFs) OpenFile(
	name string, flag int, perm os.FileMode,
) (afero.File, error) {
	f, err := fs.Fs.OpenFile(name, flag, perm)
	if err != nil || !strings.HasSuffix(name, fs.suffix) {
		return f, err
	}

	return trailingByteFile{File: f}, nil
}

type trailingByteFile struct {
	afero.File
}

func (f trailingByteFile) Close() error {
	_, _ = f.Write([]byte{'x'})

	return f.File.Close()
}

// 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)
}

// TestRestoreChunkInNoBlobAbortsWithoutSkipErrors checks that a chunk held
// by no blob still ends the restore, before any file is written, when
// SkipErrors is not set.
func TestRestoreChunkInNoBlobAbortsWithoutSkipErrors(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

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

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

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

	require.ErrorContains(t, err, "chunk missing from blob map")

	for path := range backup.files {
		assert.NoFileExists(t, filepath.Join(backup.restoreDir, path))
	}
}

// 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()

	backup, db, lostBlob := backupSingleChunkFiles(ctx, t)
	require.NoError(t, db.Close())

	require.NoError(t, backup.storer.Delete(ctx, fmt.Sprintf(
		"blobs/%s/%s/%s", lostBlob[:2], lostBlob[2:4], lostBlob)))

	return backup
}

// backupThenDropOneBlobFromMetadata backs up missingBlobFileCount
// single-chunk files, then removes the blob holding the first of them from
// the snapshot and exports the snapshot metadata again. The exported
// metadata then lists chunks that no blob holds.
func backupThenDropOneBlobFromMetadata(
	ctx context.Context, t *testing.T,
) *missingBlobBackup {
	t.Helper()

	backup, db, lostBlob := backupSingleChunkFiles(ctx, t)

	// The uploads row refers to the blob row, so it would keep the export
	// from deleting it.
	for _, query := range []string{
		"DELETE FROM snapshot_blobs WHERE blob_hash = ?",
		"DELETE FROM uploads WHERE blob_hash = ?",
	} {
		_, err := db.Conn().ExecContext(ctx, query, lostBlob)
		require.NoError(t, err)
	}

	sm := newFaultSnapshotManager(
		backup.fs, backup.storer, backup.cfg, database.NewRepositories(db))
	require.NoError(t,
		sm.ExportSnapshotMetadata(ctx, db.Path(), backup.snapshotID))
	require.NoError(t, db.Close())

	return backup
}

// backupSingleChunkFiles backs up missingBlobFileCount single-chunk files.
// It returns the backup, with lost naming the files stored in the same
// blob as the first of them, the still open local index, and the hash of
// that blob.
func backupSingleChunkFiles(
	ctx context.Context, t *testing.T,
) (*missingBlobBackup, *database.DB, string) {
	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)
	}

	lostBlob := blobOfFile[filepath.Join(dataDir, "file-00.bin")]
	lost := make(map[string]bool)

	for path, hash := range blobOfFile {
		if hash == lostBlob {
			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,
	}, db, lostBlob
}

// 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()
}