vaultik/internal/vaultik/snapshot_summary_test.go

package vaultik_test

import (
	"bytes"
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/spf13/afero"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"golang.org/x/sys/unix"
	"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/storage/faultstore"
	"sneak.berlin/go/vaultik/internal/ui"
	"sneak.berlin/go/vaultik/internal/vaultik"
)

// These tests cover https://git.eeqj.de/sneak/vaultik/issues/225: the
// summary printed after a backup, and the statistics stored in the
// snapshots table, count each file, byte and upload once, and a --cron
// run records its uploads.

// summaryUploadDelay slows every blob upload, so a run's upload time is
// at least this long per blob even on a local store.
const summaryUploadDelay = 20 * time.Millisecond

// summaryEnv is a backup setup whose user-facing output is kept in out.
type summaryEnv struct {
	v     *vaultik.Vaultik
	db    *database.DB
	repos *database.Repositories
	out   *bytes.Buffer

	// aPath is a.bin, whose content copy.bin repeats; aSize is its size
	// and totalSize the size of all three source files.
	aPath     string
	aSize     int64
	totalSize int64
}

// newSummaryEnv writes src/one/a.bin, src/one/small.txt and
// src/two/copy.bin, a copy of a.bin. Every chunk of copy.bin is therefore
// already stored by the time the backup reaches it.
//
// The snapshot names "first" and "second" back up src; "split" backs up
// src/one and src/two as two paths.
func newSummaryEnv(t *testing.T) *summaryEnv {
	t.Helper()

	fs := afero.NewOsFs()
	tempDir := t.TempDir()
	srcDir := filepath.Join(tempDir, "src")
	dirOne := filepath.Join(srcDir, "one")
	dirTwo := filepath.Join(srcDir, "two")
	dbPath := filepath.Join(tempDir, "index.sqlite")
	ctx := context.Background()

	aContent := bytesPattern("a-", int(3*faultChunkSize))
	smallContent := []byte("hello vaultik")
	files := map[string][]byte{
		filepath.Join(dirOne, "a.bin"):     aContent,
		filepath.Join(dirOne, "small.txt"): smallContent,
		filepath.Join(dirTwo, "copy.bin"):  aContent,
	}

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

	cfg := faultTestConfig()
	cfg.IndexPath = dbPath
	cfg.ChunkSize = config.Size(faultChunkSize)
	cfg.Snapshots = map[string]config.SnapshotConfig{
		"first":  {Paths: []string{srcDir}},
		"second": {Paths: []string{srcDir}},
		"split":  {Paths: []string{dirOne, dirTwo}},
	}

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

	store := faultstore.New(inner)
	store.OnPut = func(key string) faultstore.PutAction {
		if strings.HasPrefix(key, "blobs/") {
			time.Sleep(summaryUploadDelay)
		}

		return faultstore.PutNormal
	}

	db, err := database.New(ctx, dbPath)
	require.NoError(t, err)
	t.Cleanup(func() { _ = db.Close() })

	repos := database.NewRepositories(db)
	out := &bytes.Buffer{}
	v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
	v.UI = ui.NewWithColor(out, false)

	return &summaryEnv{
		v:         v,
		db:        db,
		repos:     repos,
		out:       out,
		aPath:     filepath.Join(dirOne, "a.bin"),
		aSize:     int64(len(aContent)),
		totalSize: int64(2*len(aContent) + len(smallContent)),
	}
}

// backUp runs a backup of the named snapshot and returns its output.
func (e *summaryEnv) backUp(t *testing.T, name string, cron bool) string {
	t.Helper()

	e.out.Reset()
	require.NoError(t, e.v.CreateSnapshot(&vaultik.SnapshotCreateOptions{
		Cron:      cron,
		Snapshots: []string{name},
	}))

	return e.out.String()
}

// snapshot returns the local snapshots row of the snapshot named name.
func (e *summaryEnv) snapshot(t *testing.T, name string) *database.Snapshot {
	t.Helper()

	ctx := context.Background()
	snap, err := e.repos.Snapshots.GetByID(ctx,
		localSnapshotID(ctx, t, e.repos, name))
	require.NoError(t, err)
	require.NotNil(t, snap)

	return snap
}

// uploads returns how many blobs the snapshot uploaded and their
// total size, as recorded in the uploads table.
func (e *summaryEnv) uploads(t *testing.T, snapshotID string) (int64, int64) {
	t.Helper()

	var count, size int64

	err := e.db.Conn().QueryRowContext(context.Background(), `
		SELECT COUNT(*), COALESCE(SUM(size), 0)
		FROM uploads WHERE snapshot_id = ?`, snapshotID).Scan(&count, &size)
	require.NoError(t, err)

	return count, size
}

// referencedBlobSizes returns the compressed and uncompressed sizes of
// all blobs the snapshot references.
func (e *summaryEnv) referencedBlobSizes(
	t *testing.T, snapshotID string,
) (int64, int64) {
	t.Helper()

	var compressed, uncompressed int64

	err := e.db.Conn().QueryRowContext(context.Background(), `
		SELECT COALESCE(SUM(b.compressed_size), 0),
			COALESCE(SUM(b.uncompressed_size), 0)
		FROM snapshot_blobs sb JOIN blobs b ON b.blob_hash = sb.blob_hash
		WHERE sb.snapshot_id = ?`, snapshotID).Scan(&compressed, &uncompressed)
	require.NoError(t, err)

	return compressed, uncompressed
}

// filesLine returns the summary's line of file counts.
func filesLine(examined, backedUp, unchanged int) string {
	return fmt.Sprintf("Files: %d examined, %d backed up, %d unchanged.",
		examined, backedUp, unchanged)
}

// dataLine returns the summary's line of byte counts.
func (e *summaryEnv) dataLine(total, backedUp int64) string {
	return fmt.Sprintf("Data: %s total (%s backed up).",
		e.v.UI.Size(total), e.v.UI.Size(backedUp))
}

// enumeratedLine returns the line printed when the walk of a source path
// ends.
func (e *summaryEnv) enumeratedLine(
	files int, total int64, toBackUp int, toBackUpSize int64,
) string {
	return fmt.Sprintf(
		"Enumerated %d snapshot source files (%s total), %d to back up (%s).",
		files, e.v.UI.Size(total), toBackUp, e.v.UI.Size(toBackUpSize))
}

// snapshotPaths returns the paths the snapshot lists, sorted.
func (e *summaryEnv) snapshotPaths(t *testing.T, snapshotID string) []string {
	t.Helper()

	rows, err := e.db.Conn().QueryContext(context.Background(), `
		SELECT f.path FROM snapshot_files sf JOIN files f ON f.id = sf.file_id
		WHERE sf.snapshot_id = ? ORDER BY f.path`, snapshotID)
	require.NoError(t, err)

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

	var paths []string

	for rows.Next() {
		var path string

		require.NoError(t, rows.Scan(&path))

		paths = append(paths, path)
	}

	require.NoError(t, rows.Err())

	return paths
}

// A first backup stores copy.bin's chunks while backing up a.bin, so
// copy.bin's chunks are deduplicated within the run. Each file and byte
// is still counted once.
func TestSnapshotSummaryFirstRun(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newSummaryEnv(t)

	summary := env.backUp(t, "first", false)

	assert.Contains(t, summary, filesLine(3, 3, 0))
	assert.Contains(t, summary, env.dataLine(env.totalSize, env.totalSize))

	snap := env.snapshot(t, "first")
	uploadCount, uploadBytes := env.uploads(t, snap.ID.String())
	require.Positive(t, uploadCount)

	assert.Contains(t, summary, fmt.Sprintf("Upload: %d blobs, %s in ",
		uploadCount, env.v.UI.Size(uploadBytes)))

	assert.Equal(t, int64(3), snap.FileCount)
	assert.Equal(t, env.totalSize, snap.TotalSize)
	assert.Equal(t, uploadCount, snap.BlobCount)
	assert.Equal(t, uploadBytes, snap.UploadBytes)
}

// An incremental backup where a.bin's mtime changed but its content did
// not: a.bin is backed up again and every one of its chunks is already
// stored.
func TestSnapshotSummaryIncrementalRunWithDeduplicatedChunks(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newSummaryEnv(t)

	env.backUp(t, "first", false)

	later := time.Now().Add(time.Hour)
	require.NoError(t, os.Chtimes(env.aPath, later, later))

	summary := env.backUp(t, "second", false)

	assert.Contains(t, summary, filesLine(3, 1, 2))
	assert.Contains(t, summary, env.dataLine(env.totalSize, env.aSize))
	assert.NotContains(t, summary, "Upload:")

	snap := env.snapshot(t, "second")
	compressed, uncompressed := env.referencedBlobSizes(t, snap.ID.String())
	require.Positive(t, compressed)

	assert.Equal(t, env.totalSize, snap.TotalSize)
	assert.Zero(t, snap.ChunkCount)
	assert.Zero(t, snap.BlobCount)
	assert.Zero(t, snap.UploadBytes)
	assert.Equal(t, compressed, snap.BlobSize,
		"blob_size must total the blobs the snapshot references")
	assert.Equal(t, uncompressed, snap.BlobUncompressedSize)
}

// A backup of a tree with directories and a symlink, none of which
// changed since the last backup, backs up nothing, and the snapshot
// still lists every one of them. Both sides of the line that ends the
// walk count the same entries, the files and the symlink, and no
// directory's size. See https://git.eeqj.de/sneak/vaultik/issues/305.
func TestSnapshotSummaryUnchangedRunWithDirectoriesAndSymlink(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newSummaryEnv(t)

	const linkTarget = "a.bin"

	require.NoError(t, os.Symlink(linkTarget,
		filepath.Join(filepath.Dir(env.aPath), "link")))

	// A symlink's size is the length of its target.
	total := env.totalSize + int64(len(linkTarget))

	first := env.backUp(t, "first", false)
	assert.Contains(t, first, env.enumeratedLine(4, total, 4, total))
	assert.Contains(t, first, filesLine(4, 4, 0))

	second := env.backUp(t, "second", false)
	assert.Contains(t, second, env.enumeratedLine(4, total, 0, 0))
	assert.Contains(t, second, filesLine(4, 0, 4))
	assert.Contains(t, second, env.dataLine(total, 0))

	assert.Equal(t,
		env.snapshotPaths(t, env.snapshot(t, "first").ID.String()),
		env.snapshotPaths(t, env.snapshot(t, "second").ID.String()))
}

// A directory whose mtime changed and a symlink whose target changed are
// recorded again by the next backup, which counts only the symlink as
// backed up.
func TestSnapshotSummaryChangedDirectoryAndSymlink(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newSummaryEnv(t)

	// The local index keeps paths with symlinks resolved.
	dir, err := filepath.EvalSymlinks(filepath.Dir(env.aPath))
	require.NoError(t, err)

	// The new link gets the old one's mtime, and a target, which need not
	// exist, of the old target's length, a symlink's size: only the
	// target differs.
	linkTime := unix.NsecToTimeval(
		time.Date(2029, time.March, 4, 5, 6, 7, 0, time.UTC).UnixNano())
	link := filepath.Join(dir, "link")
	require.NoError(t, os.Symlink("a.bin", link))
	require.NoError(t, unix.Lutimes(link, []unix.Timeval{linkTime, linkTime}))

	env.backUp(t, "first", false)

	require.NoError(t, os.Remove(link))
	require.NoError(t, os.Symlink("b.bin", link))
	require.NoError(t, unix.Lutimes(link, []unix.Timeval{linkTime, linkTime}))

	mtime := time.Date(2030, time.January, 2, 3, 4, 5, 0, time.UTC)
	require.NoError(t, os.Chtimes(dir, mtime, mtime))

	assert.Contains(t, env.backUp(t, "second", false), filesLine(4, 1, 3))

	ctx := context.Background()

	dirRow, err := env.repos.Files.GetByPath(ctx, dir)
	require.NoError(t, err)
	require.NotNil(t, dirRow)
	assert.Equal(t, mtime.Unix(), dirRow.MTime.Unix())

	linkRow, err := env.repos.Files.GetByPath(ctx, link)
	require.NoError(t, err)
	require.NotNil(t, linkRow)
	assert.Equal(t, "b.bin", linkRow.LinkTarget.String())
}

// Under --cron the progress reporter is off; the upload figures must
// still reach the summary and the snapshots row. The snapshot has two
// paths, each backed up by its own scan.
func TestSnapshotSummaryCronRunRecordsUploads(t *testing.T) {
	log.Initialize(log.Config{})
	t.Parallel()

	env := newSummaryEnv(t)

	summary := env.backUp(t, "split", true)

	snap := env.snapshot(t, "split")
	uploadCount, uploadBytes := env.uploads(t, snap.ID.String())
	require.Positive(t, uploadCount)

	assert.Contains(t, summary, filesLine(3, 3, 0))
	assert.Contains(t, summary, env.dataLine(env.totalSize, env.totalSize))
	assert.Contains(t, summary, fmt.Sprintf("Upload: %d blobs, %s in ",
		uploadCount, env.v.UI.Size(uploadBytes)))

	assert.Equal(t, env.totalSize, snap.TotalSize)
	assert.Equal(t, uploadCount, snap.BlobCount,
		"blob_count must count each blob once, however many paths the "+
			"snapshot has")
	assert.Equal(t, uploadBytes, snap.UploadBytes)
	assert.GreaterOrEqual(t, snap.UploadDurationMs,
		uploadCount*summaryUploadDelay.Milliseconds())

	compressed, uncompressed := env.referencedBlobSizes(t, snap.ID.String())
	require.Positive(t, uncompressed)

	assert.Equal(t, compressed, snap.BlobSize)
	assert.Equal(t, uncompressed, snap.BlobUncompressedSize)
	assert.InDelta(t, float64(compressed)/float64(uncompressed),
		snap.CompressionRatio, 1e-9)
}