vaultik/internal/vaultik/fault_injection_test.go

package vaultik_test

import (
	"context"
	"errors"
	"io"
	"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/snapshot"
	"sneak.berlin/go/vaultik/internal/storage"
	"sneak.berlin/go/vaultik/internal/storage/faultstore"
	"sneak.berlin/go/vaultik/internal/ui"
	"sneak.berlin/go/vaultik/internal/vaultik"
)

// These tests cover the failure modes a backup tool must survive:
// interrupted uploads, an interrupted metadata export, corrupt and
// truncated reads, a full restore disk, and a backend that reports
// success while storing nothing. Faults are injected through the
// storage.Storer seam (internal/storage/faultstore), never by patching
// production code. Each test asserts on the observable end state — what
// is in the index, what is at the destination, what the user is told —
// not merely that an error was returned. See
// https://git.eeqj.de/sneak/vaultik/issues/72.
//
// Object-level write atomicity (no partial blob object left behind) is
// covered by the file:// backend's atomic-write work
// (https://git.eeqj.de/sneak/vaultik/issues/130) and is not re-tested
// here; these tests target the layers above the backend.
//
// The tests run serially, not with t.Parallel: each calls
// log.Initialize, which replaces the package-global logger, and a
// backup or restore running concurrently reads that same logger. Under
// -race the two collide. Running one at a time is the same choice
// prune_count_test.go already makes for the same reason.

const (
	faultChunkSize   = int64(64 * 1024)
	faultMaxBlobSize = int64(256 * 1024)
)

// faultTestConfig returns the config shared by the fault-injection
// tests: a real recipient/secret keypair so blobs are genuinely
// encrypted, and a blob size limit the restore sweeper can divide.
func faultTestConfig() *config.Config {
	return &config.Config{
		AgeRecipients:    []string{testAgePublicKey},
		AgeSecretKey:     testAgeSecretKey,
		CompressionLevel: 3,
		Hostname:         testHostname,
		BlobSizeLimit:    config.Size(faultMaxBlobSize),
	}
}

// writeFaultSourceTree writes a spread of file sizes that forces several
// chunks across more than one blob, so a fault landing on a single blob
// still leaves other data intact. Returns the expected content by path.
func writeFaultSourceTree(
	t *testing.T, fs afero.Fs, dataDir string,
) map[string][]byte {
	t.Helper()

	files := map[string][]byte{
		filepath.Join(dataDir, "small.txt"):    []byte("hello vaultik"),
		filepath.Join(dataDir, "a.bin"):        bytesPattern("a-", int(faultChunkSize*3)),
		filepath.Join(dataDir, "sub", "b.bin"): bytesPattern("b-", int(faultChunkSize*3)),
		filepath.Join(dataDir, "sub", "c.bin"): bytesPattern("c-", int(faultChunkSize*2)),
	}

	for path, content := range files {
		require.NoError(t, fs.MkdirAll(filepath.Dir(path), 0o755))
		require.NoError(t, afero.WriteFile(fs, path, content, 0o644))
	}

	return files
}

// newFaultScanner builds a scanner writing through the given storer.
func newFaultScanner(
	fs afero.Fs, storer storage.Storer,
	cfg *config.Config, repos *database.Repositories,
) *snapshot.Scanner {
	return snapshot.NewScanner(snapshot.ScannerConfig{
		FS:               fs,
		Storage:          storer,
		ChunkSize:        faultChunkSize,
		MaxBlobSize:      faultMaxBlobSize,
		CompressionLevel: cfg.CompressionLevel,
		AgeRecipients:    cfg.AgeRecipients,
		Repositories:     repos,
	})
}

// newFaultSnapshotManager builds a snapshot manager writing through the
// given storer.
func newFaultSnapshotManager(
	fs afero.Fs, storer storage.Storer,
	cfg *config.Config, repos *database.Repositories,
) *snapshot.SnapshotManager {
	sm := snapshot.NewSnapshotManager(snapshot.SnapshotManagerParams{
		Repos:   repos,
		Storage: storer,
		Config:  cfg,
	})
	sm.SetFilesystem(fs)

	return sm
}

// fullFaultBackup runs a complete backup (create, scan, complete,
// export) through storer and returns the snapshot ID.
func fullFaultBackup(
	ctx context.Context, t *testing.T, fs afero.Fs, storer storage.Storer,
	cfg *config.Config, repos *database.Repositories,
	dataDir, dbPath, name string,
) string {
	t.Helper()

	sm := newFaultSnapshotManager(fs, storer, cfg, repos)
	scanner := newFaultScanner(fs, storer, cfg, repos)

	id, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, name, "v", "g")
	require.NoError(t, err)

	_, err = scanner.Scan(ctx, dataDir, id)
	require.NoError(t, err)

	require.NoError(t, sm.CompleteSnapshot(ctx, id))
	require.NoError(t, sm.ExportSnapshotMetadata(ctx, dbPath, id))

	return id
}

// newReaderVaultik builds a Vaultik that reads (restore/verify) through
// storer, with the given repositories (nil is fine for restore/verify,
// which read metadata from storage).
func newReaderVaultik(
	ctx context.Context, cfg *config.Config, storer storage.Storer,
	repos *database.Repositories, fs afero.Fs,
) *vaultik.Vaultik {
	v := &vaultik.Vaultik{
		Config:       cfg,
		Storage:      storer,
		Repositories: repos,
		Fs:           fs,
		Stdout:       io.Discard,
		Stderr:       io.Discard,
		UI:           ui.NewWithColor(io.Discard, false),
	}
	v.SetContext(ctx)

	return v
}

// Scenario 3: a stored blob's bytes are flipped before restore reads
// them. Restore must fail loudly, and no file must be left on the
// restore target holding corrupt content.
//
//nolint:paralleltest // installs the global logger via log.Initialize
func TestRestoreRejectsCorruptBlob(t *testing.T) {
	assertRestoreRejectsDamagedBlob(t, faultstore.GetCorrupt, "corrupt")
}

// Scenario 4: a stored blob is truncated before restore reads it. Same
// contract as the corrupt case.
//
//nolint:paralleltest // installs the global logger via log.Initialize
func TestRestoreRejectsTruncatedBlob(t *testing.T) {
	assertRestoreRejectsDamagedBlob(t, faultstore.GetTruncate, "truncated")
}

// assertRestoreRejectsDamagedBlob backs up the source tree, then restores
// through a store that damages every blob read with the given fault, and
// asserts restore fails naming a blob and leaves no file on the target
// holding wrong bytes. Metadata reads are returned intact so the failure
// is isolated to the blob.
func assertRestoreRejectsDamagedBlob(
	t *testing.T, fault faultstore.GetFault, name string,
) {
	t.Helper()
	log.Initialize(log.Config{})

	fs := afero.NewOsFs()
	tempDir := t.TempDir()
	dataDir := filepath.Join(tempDir, "src")
	storeDir := filepath.Join(tempDir, "remote")
	restoreDir := filepath.Join(tempDir, "restored")
	dbPath := filepath.Join(tempDir, "index.sqlite")

	ctx := context.Background()
	cfg := faultTestConfig()
	testFiles := writeFaultSourceTree(t, fs, dataDir)

	inner, err := storage.NewFileStorer(storeDir)
	require.NoError(t, err)

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

	repos := database.NewRepositories(db)

	id := fullFaultBackup(ctx, t, fs, inner, cfg, repos, dataDir, dbPath, name)
	require.NoError(t, db.Close())

	faultStore := faultstore.New(inner)
	faultStore.OnGet = func(key string) faultstore.GetFault {
		if strings.HasPrefix(key, "blobs/") {
			return fault
		}

		return faultstore.GetNormal
	}

	v := newReaderVaultik(ctx, cfg, faultStore, nil, fs)
	err = v.Restore(&vaultik.RestoreOptions{SnapshotID: id, TargetDir: restoreDir})

	require.Error(t, err, "restore must fail on a damaged blob")
	assert.Contains(t, err.Error(), "blob",
		"error should name the blob that failed")
	assertNoCorruptFiles(t, fs, restoreDir, testFiles)
}

// Scenario 6: the backend accepts blob uploads and reports success but
// stores nothing. verify --deep must catch it.
//
//nolint:paralleltest // installs the global logger via log.Initialize
func TestDeepVerifyCatchesLyingBackend(t *testing.T) {
	log.Initialize(log.Config{})

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

	ctx := context.Background()
	cfg := faultTestConfig()

	writeFaultSourceTree(t, fs, dataDir)

	inner, err := storage.NewFileStorer(storeDir)
	require.NoError(t, err)

	// Blob uploads are swallowed; metadata uploads land, so verify can
	// download the manifest and database and then discover the blobs are
	// absent.
	lying := faultstore.New(inner)
	lying.OnPut = func(key string) faultstore.PutAction {
		if strings.HasPrefix(key, "blobs/") {
			return faultstore.PutSwallow
		}

		return faultstore.PutNormal
	}

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

	repos := database.NewRepositories(db)

	id := fullFaultBackup(ctx, t, fs, lying, cfg, repos, dataDir, dbPath, "lying")
	require.NoError(t, db.Close())

	// No blob objects were actually written.
	blobKeys, err := inner.List(ctx, "blobs/")
	require.NoError(t, err)
	assert.Empty(t, blobKeys, "lying backend should have stored no blobs")

	// Read back through the honest underlying store.
	v := newReaderVaultik(ctx, cfg, inner, nil, fs)
	err = v.VerifySnapshotWithOptions(id, &vaultik.VerifyOptions{Deep: true})
	require.Error(t, err, "deep verify must catch a backend that stored nothing")
}

// Scenario 1a: a blob upload fails partway through. The interrupted run
// must not record the blob as uploaded, must not reference it from the
// snapshot, and must leave no blob object at the destination.
//
//nolint:paralleltest // installs the global logger via log.Initialize
func TestInterruptedBlobUploadRecordsNoUploadedBlob(t *testing.T) {
	log.Initialize(log.Config{})

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

	ctx := context.Background()
	cfg := faultTestConfig()

	writeFaultSourceTree(t, fs, dataDir)

	inner, err := storage.NewFileStorer(storeDir)
	require.NoError(t, err)

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

	defer func() { _ = db.Close() }()

	repos := database.NewRepositories(db)

	// Every blob upload fails partway through. The scan must surface it.
	fault := faultstore.New(inner)
	fault.OnPut = func(key string) faultstore.PutAction {
		if strings.HasPrefix(key, "blobs/") {
			return faultstore.PutFail
		}

		return faultstore.PutNormal
	}

	sm := newFaultSnapshotManager(fs, fault, cfg, repos)
	scanner := newFaultScanner(fs, fault, cfg, repos)

	id, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, "interrupted", "v", "g")
	require.NoError(t, err)

	_, err = scanner.Scan(ctx, dataDir, id)
	require.Error(t, err, "scan must fail when a blob upload fails")

	// No blob may claim to be uploaded.
	blobs, err := repos.Blobs.GetAll(ctx)
	require.NoError(t, err)

	for _, b := range blobs {
		assert.Nilf(t, b.UploadedTS,
			"blob %s marked uploaded after a failed upload", b.Hash)
	}

	// The snapshot may reference no blobs, and the destination holds none.
	hashes, err := repos.Snapshots.GetBlobHashes(ctx, id)
	require.NoError(t, err)
	assert.Empty(t, hashes, "interrupted snapshot must reference no blobs")

	blobKeys, err := inner.List(ctx, "blobs/")
	require.NoError(t, err)
	assert.Empty(t, blobKeys, "no blob object may survive at the destination")
}

// Scenario 1b: after an interrupted upload, a retry on the same local
// index must produce a restorable snapshot. The interrupted run leaves
// the blob's chunk rows in the index; the fix for
// https://git.eeqj.de/sneak/vaultik/issues/148 discards those un-uploaded
// blob rows at the start of the next scan and deduplicates only against
// chunks in a blob that was actually uploaded, so the retry re-chunks and
// re-uploads the affected data instead of silently referencing data that
// never reached storage.
//
//nolint:paralleltest // installs the global logger via log.Initialize
func TestBackupRetryAfterInterruptedUploadIsRestorable(t *testing.T) {
	log.Initialize(log.Config{})

	fs := afero.NewOsFs()
	tempDir := t.TempDir()
	dataDir := filepath.Join(tempDir, "src")
	storeDir := filepath.Join(tempDir, "remote")
	restoreDir := filepath.Join(tempDir, "restored")
	dbPath := filepath.Join(tempDir, "index.sqlite")

	ctx := context.Background()
	cfg := faultTestConfig()
	testFiles := writeFaultSourceTree(t, fs, dataDir)

	inner, err := storage.NewFileStorer(storeDir)
	require.NoError(t, err)

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

	repos := database.NewRepositories(db)

	// Attempt 1: every blob upload fails.
	fault := faultstore.New(inner)
	fault.OnPut = func(key string) faultstore.PutAction {
		if strings.HasPrefix(key, "blobs/") {
			return faultstore.PutFail
		}

		return faultstore.PutNormal
	}

	sm := newFaultSnapshotManager(fs, fault, cfg, repos)
	scanner := newFaultScanner(fs, fault, cfg, repos)

	id1, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, "interrupted", "v", "g")
	require.NoError(t, err)

	_, err = scanner.Scan(ctx, dataDir, id1)
	require.Error(t, err)

	// Retry on the same local index with a working backend.
	id2 := fullFaultBackup(ctx, t, fs, inner, cfg, repos, dataDir, dbPath, "retry")
	require.NoError(t, db.Close())

	v := newReaderVaultik(ctx, cfg, inner, nil, fs)
	require.NoError(t, v.Restore(&vaultik.RestoreOptions{
		SnapshotID: id2,
		TargetDir:  restoreDir,
		Verify:     true,
	}), "retry after an interrupted upload must produce a restorable snapshot")

	assertRestoredTree(t, fs, restoreDir, testFiles)
}

// Scenario 2: the process dies during the metadata export, after the
// database is uploaded but before the manifest. The destination is left
// with blobs and a database but no manifest. verify and snapshot list
// must report the damage honestly rather than crashing or passing.
// Automatic detection and repair of this partial state on the next run
// is tracked in https://git.eeqj.de/sneak/vaultik/issues/177 and is not
// asserted here.
//
//nolint:paralleltest // installs the global logger via log.Initialize
func TestBackupSurvivesMetadataExportInterruption(t *testing.T) {
	log.Initialize(log.Config{})

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

	ctx := context.Background()
	cfg := faultTestConfig()

	writeFaultSourceTree(t, fs, dataDir)

	inner, err := storage.NewFileStorer(storeDir)
	require.NoError(t, err)

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

	repos := database.NewRepositories(db)

	// Back up and complete with a working backend.
	sm := newFaultSnapshotManager(fs, inner, cfg, repos)
	scanner := newFaultScanner(fs, inner, cfg, repos)

	id, err := sm.CreateSnapshotWithName(ctx, cfg.Hostname, "export", "v", "g")
	require.NoError(t, err)

	_, err = scanner.Scan(ctx, dataDir, id)
	require.NoError(t, err)
	require.NoError(t, sm.CompleteSnapshot(ctx, id))

	// Export through a backend that fails only the manifest upload. The
	// database uploads first and lands; the manifest does not.
	fault := faultstore.New(inner)
	fault.OnPut = func(key string) faultstore.PutAction {
		if strings.HasSuffix(key, "manifest.json.zst") {
			return faultstore.PutFail
		}

		return faultstore.PutNormal
	}

	smFault := newFaultSnapshotManager(fs, fault, cfg, repos)

	err = smFault.ExportSnapshotMetadata(ctx, dbPath, id)
	require.Error(t, err, "export must fail when the manifest upload fails")

	// The destination is in the partial state the scenario describes.
	key := snapshot.RemoteSnapshotKey(id)

	_, err = inner.Stat(ctx, "metadata/"+key+"/db.zst.age")
	require.NoError(t, err, "database should have been uploaded before the manifest")

	_, err = inner.Stat(ctx, "metadata/"+key+"/manifest.json.zst")
	require.ErrorIs(t, err, storage.ErrNotFound, "manifest upload should not have landed")

	// verify must fail loudly for this snapshot, in both modes.
	reader := newReaderVaultik(ctx, cfg, inner, repos, fs)

	deepOpts := &vaultik.VerifyOptions{Deep: true}
	require.Error(t, reader.VerifySnapshotWithOptions(id, deepOpts),
		"deep verify must report the missing manifest")

	shallowOpts := &vaultik.VerifyOptions{Deep: false}
	require.Error(t, reader.VerifySnapshotWithOptions(id, shallowOpts),
		"shallow verify must report the missing manifest")

	// snapshot list must not crash on the partial snapshot.
	require.NoError(t, reader.ListSnapshots(false),
		"snapshot list must tolerate a partially-exported snapshot")
}

// Scenario 5: the restore target runs out of space mid-file. Restore
// must fail with an out-of-space error, and must not leave a truncated
// file at the target path presenting as a complete restore. Restore
// today writes each file straight to its final path and does not remove
// it when a write fails, so the truncated file survives; deleting it is
// tracked by https://git.eeqj.de/sneak/vaultik/issues/163. Skipped until
// that lands, so the destination assertion below is recorded rather than
// dropped.
//
//nolint:paralleltest // installs the global logger via log.Initialize
func TestRestoreReportsDiskFull(t *testing.T) {
	t.Skip("blocked on https://git.eeqj.de/sneak/vaultik/issues/163: " +
		"a disk-full write leaves a truncated file at the target path " +
		"instead of removing it")
	log.Initialize(log.Config{})

	osFS := afero.NewOsFs()
	tempDir := t.TempDir()
	dataDir := filepath.Join(tempDir, "src")
	storeDir := filepath.Join(tempDir, "remote")
	restoreDir := filepath.Join(tempDir, "restored")
	dbPath := filepath.Join(tempDir, "index.sqlite")

	ctx := context.Background()
	cfg := faultTestConfig()

	testFiles := writeFaultSourceTree(t, osFS, dataDir)

	inner, err := storage.NewFileStorer(storeDir)
	require.NoError(t, err)

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

	repos := database.NewRepositories(db)

	id := fullFaultBackup(ctx, t, osFS, inner, cfg, repos, dataDir, dbPath, "diskfull")
	require.NoError(t, db.Close())

	// Restore onto a filesystem that allows only a few bytes of file
	// content: enough to create files, far too little to hold them.
	budget := int64(8)
	quota := &quotaFS{Fs: osFS, remaining: &budget}

	v := newReaderVaultik(ctx, cfg, inner, nil, quota)
	err = v.Restore(&vaultik.RestoreOptions{SnapshotID: id, TargetDir: restoreDir})

	require.Error(t, err, "restore must fail when the target disk is full")
	assert.Contains(t, err.Error(), errNoSpace.Error(),
		"restore error should surface the out-of-space cause")

	// The failure must not leave a truncated file behind presenting as a
	// complete restore: any file at the target must hold the original
	// bytes, or be absent.
	assertNoCorruptFiles(t, osFS, restoreDir, testFiles)
}

// assertRestoredTree byte-compares every restored file against the
// original.
func assertRestoredTree(
	t *testing.T, fs afero.Fs, restoreDir string, testFiles map[string][]byte,
) {
	t.Helper()

	for origPath, expected := range testFiles {
		restoredPath := filepath.Join(restoreDir, origPath)
		got, err := afero.ReadFile(fs, restoredPath)
		require.NoErrorf(t, err, "restored file missing: %s", origPath)
		require.Equalf(t, expected, got, "restored content mismatch for %s", origPath)
	}
}

// errNoSpace is the out-of-space error quotaFS returns once its byte
// budget is exhausted, mirroring a real ENOSPC.
var errNoSpace = errors.New("no space left on device")

// quotaFS is an afero.Fs whose files may write only a fixed total number
// of content bytes before failing, simulating a full restore target. It
// wraps the interface so every method except Create delegates to the
// real filesystem; only file writes are capped.
type quotaFS struct {
	afero.Fs

	remaining *int64
}

//nolint:ireturn // afero.Fs.Create's signature requires returning afero.File.
func (q *quotaFS) Create(name string) (afero.File, error) {
	f, err := q.Fs.Create(name)
	if err != nil {
		return nil, err
	}

	return &quotaFile{File: f, remaining: q.remaining}, nil
}

// quotaFile fails writes once the shared byte budget is exhausted.
type quotaFile struct {
	afero.File

	remaining *int64
}

func (q *quotaFile) Write(p []byte) (int, error) {
	if *q.remaining <= 0 {
		return 0, errNoSpace
	}

	allowed := min(int64(len(p)), *q.remaining)

	n, err := q.File.Write(p[:allowed])
	*q.remaining -= int64(n)

	if err != nil {
		return n, err
	}

	if int64(n) < int64(len(p)) {
		return n, errNoSpace
	}

	return n, nil
}

// assertNoCorruptFiles fails if any file that made it to the restore
// target holds content that differs from the original: a failed restore
// may leave a file absent, but must never leave wrong bytes presenting
// as the real file.
func assertNoCorruptFiles(
	t *testing.T, fs afero.Fs, restoreDir string, testFiles map[string][]byte,
) {
	t.Helper()

	for origPath, expected := range testFiles {
		restoredPath := filepath.Join(restoreDir, origPath)

		got, err := afero.ReadFile(fs, restoredPath)
		if err != nil {
			if os.IsNotExist(err) {
				continue
			}

			require.NoError(t, err)
		}

		assert.Equalf(t, expected, got,
			"restored file %s holds corrupt content", origPath)
	}
}