refactor(snapshots): extracted snapshotfs.Verifier component (#1921)

* refactor(snapshots): extracted snapshotfs.Verifier component

* refactor(repository): added tests for snapshotfs.Verifier, misc cleanups

* fixed data race

* fixed atomic alignment

* nit

cli/command_content_verify.go

@@ -37,26 +37,6 @@ func (c *commandContentVerify) setup(svc appServices, parent commandParent) {
 	cmd.Action(svc.directRepositoryReadAction(c.run))
 }
 
-func readBlobMap(ctx context.Context, br blob.Reader) (map[blob.ID]blob.Metadata, error) {
-	blobMap := map[blob.ID]blob.Metadata{}
-
-	log(ctx).Infof("Listing blobs...")
-
-	if err := br.ListBlobs(ctx, "", func(bm blob.Metadata) error {
-		blobMap[bm.BlobID] = bm
-		if len(blobMap)%10000 == 0 {
-			log(ctx).Infof("  %v blobs...", len(blobMap))
-		}
-		return nil
-	}); err != nil {
-		return nil, errors.Wrap(err, "unable to list blobs")
-	}
-
-	log(ctx).Infof("Listed %v blobs.", len(blobMap))
-
-	return blobMap, nil
-}
-
 func (c *commandContentVerify) run(ctx context.Context, rep repo.DirectRepository) error {
 	blobMap := map[blob.ID]blob.Metadata{}
 	downloadPercent := c.contentVerifyPercent
@@ -65,9 +45,9 @@ func (c *commandContentVerify) run(ctx context.Context, rep repo.DirectRepositor
 		downloadPercent = 100.0
 	}
 
-	blobMap, err := readBlobMap(ctx, rep.BlobReader())
+	blobMap, err := blob.ReadBlobMap(ctx, rep.BlobReader())
 	if err != nil {
-		return err
+		return errors.Wrap(err, "unable to read blob map")
 	}
 
 	verifiedCount := new(int32)

cli/command_snapshot_verify.go

@@ -3,31 +3,17 @@ package cli
 import (
 	"context"
 	"fmt"
-	"io"
-	"math/rand"
 	"runtime"
-	"sync"
-	"sync/atomic"
-	"time"
 
 	"github.com/pkg/errors"
 
-	"github.com/kopia/kopia/fs"
-	"github.com/kopia/kopia/internal/iocopy"
-	"github.com/kopia/kopia/internal/timetrack"
 	"github.com/kopia/kopia/repo"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/manifest"
-	"github.com/kopia/kopia/repo/object"
 	"github.com/kopia/kopia/snapshot"
 	"github.com/kopia/kopia/snapshot/snapshotfs"
 )
 
-type verifyFileWorkItem struct {
-	oid       object.ID
-	entryPath string
-}
-
 type commandSnapshotVerify struct {
 	verifyCommandErrorThreshold int
 	verifyCommandDirObjectIDs   []string
@@ -57,92 +43,6 @@ func (c *commandSnapshotVerify) setup(svc appServices, parent commandParent) {
 	cmd.Action(svc.repositoryReaderAction(c.run))
 }
 
-type verifier struct {
-	throttle timetrack.Throttle
-	// +checkatomic
-	queued int32
-	// +checkatomic
-	processed int32
-
-	fileWorkItems chan verifyFileWorkItem
-
-	rep repo.Repository
-
-	blobMap map[blob.ID]blob.Metadata
-
-	downloadFilesPercent float64
-}
-
-func (v *verifier) showStats(ctx context.Context) {
-	processed := atomic.LoadInt32(&v.processed)
-
-	log(ctx).Infof("Processed %v objects.", processed)
-}
-
-func (v *verifier) verifyFile(ctx context.Context, oid object.ID, entryPath string) error {
-	log(ctx).Debugf("verifying object %v", oid)
-
-	defer func() {
-		atomic.AddInt32(&v.processed, 1)
-	}()
-
-	contentIDs, err := v.rep.VerifyObject(ctx, oid)
-	if err != nil {
-		return errors.Wrap(err, "verify object")
-	}
-
-	if v.blobMap != nil {
-		for _, cid := range contentIDs {
-			ci, err := v.rep.ContentInfo(ctx, cid)
-			if err != nil {
-				return errors.Wrapf(err, "error verifying content %v", cid)
-			}
-
-			if _, ok := v.blobMap[ci.GetPackBlobID()]; !ok {
-				return errors.Errorf("object %v is backed by missing blob %v", oid, ci.GetPackBlobID())
-			}
-		}
-	}
-
-	//nolint:gosec
-	if 100*rand.Float64() < v.downloadFilesPercent {
-		if err := v.readEntireObject(ctx, oid, entryPath); err != nil {
-			return errors.Wrapf(err, "error reading object %v", oid)
-		}
-	}
-
-	return nil
-}
-
-func (v *verifier) doVerifyObject(ctx context.Context, e fs.Entry, oid object.ID, entryPath string) error {
-	if v.throttle.ShouldOutput(time.Second) {
-		v.showStats(ctx)
-	}
-
-	if !e.IsDir() {
-		v.fileWorkItems <- verifyFileWorkItem{oid, entryPath}
-		atomic.AddInt32(&v.queued, 1)
-	} else {
-		atomic.AddInt32(&v.queued, 1)
-		atomic.AddInt32(&v.processed, 1)
-	}
-
-	return nil
-}
-
-func (v *verifier) readEntireObject(ctx context.Context, oid object.ID, path string) error {
-	log(ctx).Debugf("reading object %v %v", oid, path)
-
-	// also read the entire file
-	r, err := v.rep.OpenObject(ctx, oid)
-	if err != nil {
-		return errors.Wrapf(err, "unable to open object %v", oid)
-	}
-	defer r.Close() //nolint:errcheck
-
-	return errors.Wrap(iocopy.JustCopy(io.Discard, r), "unable to read data")
-}
-
 func (c *commandSnapshotVerify) run(ctx context.Context, rep repo.Repository) error {
 	if c.verifyCommandAllSources {
 		log(ctx).Errorf("DEPRECATED: --all-sources flag has no effect and is the default when no sources are provided.")
@@ -152,113 +52,72 @@ func (c *commandSnapshotVerify) run(ctx context.Context, rep repo.Repository) er
 		dr.DisableIndexRefresh()
 	}
 
-	v := &verifier{
-		rep:                  rep,
-		downloadFilesPercent: c.verifyCommandFilesPercent,
-		fileWorkItems:        make(chan verifyFileWorkItem, c.fileQueueLength),
+	opts := snapshotfs.VerifierOptions{
+		VerifyFilesPercent: c.verifyCommandFilesPercent,
+		FileQueueLength:    c.fileQueueLength,
+		Parallelism:        c.fileParallelism,
+		MaxErrors:          c.verifyCommandErrorThreshold,
 	}
 
-	tw, twerr := snapshotfs.NewTreeWalker(snapshotfs.TreeWalkerOptions{
-		Parallelism:   c.verifyCommandParallel,
-		EntryCallback: v.doVerifyObject,
-		MaxErrors:     c.verifyCommandErrorThreshold,
-	})
-	if twerr != nil {
-		return errors.Wrap(twerr, "unable to initialize tree walker")
-	}
-
-	defer tw.Close()
-
 	if dr, ok := rep.(repo.DirectRepository); ok {
-		blobMap, err := readBlobMap(ctx, dr.BlobReader())
+		blobMap, err := blob.ReadBlobMap(ctx, dr.BlobReader())
+		if err != nil {
+			return errors.Wrap(err, "unable to read blob map")
+		}
+
+		opts.BlobMap = blobMap
+	}
+
+	v := snapshotfs.NewVerifier(ctx, rep, opts)
+
+	// nolint:wrapcheck
+	return v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+		manifests, err := c.loadSourceManifests(ctx, rep, c.verifyCommandSources)
 		if err != nil {
 			return err
 		}
 
-		v.blobMap = blobMap
-	}
+		for _, man := range manifests {
+			rootPath := fmt.Sprintf("%v@%v", man.Source, formatTimestamp(man.StartTime))
 
-	var vwg sync.WaitGroup
-
-	for i := 0; i < c.fileParallelism; i++ {
-		vwg.Add(1)
-
-		go func() {
-			defer vwg.Done()
-
-			for wi := range v.fileWorkItems {
-				if tw.TooManyErrors() {
-					continue
-				}
-
-				if err := v.verifyFile(ctx, wi.oid, wi.entryPath); err != nil {
-					tw.ReportError(ctx, wi.entryPath, err)
-				}
+			if man.RootEntry == nil {
+				continue
 			}
-		}()
-	}
 
-	err := c.processRoots(ctx, tw, rep)
+			root, err := snapshotfs.SnapshotRoot(rep, man)
+			if err != nil {
+				return errors.Wrapf(err, "unable to get snapshot root: %q", rootPath)
+			}
 
-	close(v.fileWorkItems)
-	vwg.Wait()
-
-	if err != nil {
-		return err
-	}
-
-	v.showStats(ctx)
-
-	// nolint:wrapcheck
-	return tw.Err()
-}
-
-func (c *commandSnapshotVerify) processRoots(ctx context.Context, tw *snapshotfs.TreeWalker, rep repo.Repository) error {
-	manifests, err := c.loadSourceManifests(ctx, rep, c.verifyCommandSources)
-	if err != nil {
-		return err
-	}
-
-	for _, man := range manifests {
-		rootPath := fmt.Sprintf("%v@%v", man.Source, formatTimestamp(man.StartTime))
-
-		if man.RootEntry == nil {
-			continue
+			// ignore error now, return aggregate error at a higher level.
+			// nolint:errcheck
+			tw.Process(ctx, root, rootPath)
 		}
 
-		root, err := snapshotfs.SnapshotRoot(rep, man)
-		if err != nil {
-			return errors.Wrapf(err, "unable to get snapshot root: %q", rootPath)
+		for _, oidStr := range c.verifyCommandDirObjectIDs {
+			oid, err := snapshotfs.ParseObjectIDWithPath(ctx, rep, oidStr)
+			if err != nil {
+				return errors.Wrapf(err, "unable to parse: %q", oidStr)
+			}
+
+			// ignore error now, return aggregate error at a higher level.
+			// nolint:errcheck
+			tw.Process(ctx, snapshotfs.DirectoryEntry(rep, oid, nil), oidStr)
 		}
 
-		// ignore error now, return aggregate error at a higher level.
-		// nolint:errcheck
-		tw.Process(ctx, root, rootPath)
-	}
+		for _, oidStr := range c.verifyCommandFileObjectIDs {
+			oid, err := snapshotfs.ParseObjectIDWithPath(ctx, rep, oidStr)
+			if err != nil {
+				return errors.Wrapf(err, "unable to parse %q", oidStr)
+			}
 
-	for _, oidStr := range c.verifyCommandDirObjectIDs {
-		oid, err := snapshotfs.ParseObjectIDWithPath(ctx, rep, oidStr)
-		if err != nil {
-			return errors.Wrapf(err, "unable to parse: %q", oidStr)
+			// ignore error now, return aggregate error at a higher level.
+			// nolint:errcheck
+			tw.Process(ctx, snapshotfs.AutoDetectEntryFromObjectID(ctx, rep, oid, oidStr), oidStr)
 		}
 
-		// ignore error now, return aggregate error at a higher level.
-		// nolint:errcheck
-		tw.Process(ctx, snapshotfs.DirectoryEntry(rep, oid, nil), oidStr)
-	}
-
-	for _, oidStr := range c.verifyCommandFileObjectIDs {
-		oid, err := snapshotfs.ParseObjectIDWithPath(ctx, rep, oidStr)
-		if err != nil {
-			return errors.Wrapf(err, "unable to parse %q", oidStr)
-		}
-
-		// ignore error now, return aggregate error at a higher level.
-		// nolint:errcheck
-		tw.Process(ctx, snapshotfs.AutoDetectEntryFromObjectID(ctx, rep, oid, oidStr), oidStr)
-	}
-
-	return nil
+		return nil
+	})
 }
 
 func (c *commandSnapshotVerify) loadSourceManifests(ctx context.Context, rep repo.Repository, sources []string) ([]*snapshot.Manifest, error) {

internal/repotesting/repotesting.go

@@ -17,6 +17,7 @@ import (
 	"github.com/kopia/kopia/repo/content"
 	"github.com/kopia/kopia/repo/encryption"
 	"github.com/kopia/kopia/repo/object"
+	"github.com/kopia/kopia/snapshot"
 )
 
 const defaultPassword = "foobarbazfoobarbaz"
@@ -241,6 +242,15 @@ func (e *Environment) VerifyBlobCount(tb testing.TB, want int) {
 	}
 }
 
+// LocalPathSourceInfo is a convenience method that returns SourceInfo for the local user and path.
+func (e *Environment) LocalPathSourceInfo(path string) snapshot.SourceInfo {
+	return snapshot.SourceInfo{
+		UserName: e.Repository.ClientOptions().Username,
+		Host:     e.Repository.ClientOptions().Hostname,
+		Path:     path,
+	}
+}
+
 func repoOptions(openOpts []func(*repo.Options)) *repo.Options {
 	openOpt := &repo.Options{}
 

internal/server/api_paths_test.go

@@ -33,5 +33,5 @@ func TestPathsAPI(t *testing.T) {
 	resp := &serverapi.ResolvePathResponse{}
 	require.NoError(t, cli.Post(ctx, "paths/resolve", req, resp))
 
-	require.Equal(t, localSource(env, dir0), resp.SourceInfo)
+	require.Equal(t, env.LocalPathSourceInfo(dir0), resp.SourceInfo)
 }

internal/server/api_policies_test.go

@@ -32,19 +32,19 @@ func TestPolicies(t *testing.T) {
 	require.NoError(t, cli.FetchCSRFTokenForTesting(ctx))
 
 	dir0 := testutil.TempDirectory(t)
-	si0 := localSource(env, dir0)
+	si0 := env.LocalPathSourceInfo(dir0)
 
 	dir1 := testutil.TempDirectory(t)
-	si1 := localSource(env, dir1)
+	si1 := env.LocalPathSourceInfo(dir1)
 
 	dir2 := filepath.Join(dir1, "subdir1")
-	si2 := localSource(env, dir2)
+	si2 := env.LocalPathSourceInfo(dir2)
 
 	dir3 := filepath.Join(dir2, "subdir2")
-	si3 := localSource(env, dir3)
+	si3 := env.LocalPathSourceInfo(dir3)
 
 	dir4 := filepath.Join(dir3, "subdir3")
-	si4 := localSource(env, dir4)
+	si4 := env.LocalPathSourceInfo(dir4)
 
 	mustSetPolicy(t, cli, si1, &policy.Policy{
 		CompressionPolicy: policy.CompressionPolicy{

internal/server/api_snapshots_test.go

@@ -19,8 +19,8 @@ import (
 func TestListAndDeleteSnapshots(t *testing.T) {
 	ctx, env := repotesting.NewEnvironment(t, repotesting.FormatNotImportant)
 
-	si1 := localSource(env, "/dummy/path")
-	si2 := localSource(env, "/another/path")
+	si1 := env.LocalPathSourceInfo("/dummy/path")
+	si2 := env.LocalPathSourceInfo("/another/path")
 
 	var id11, id12, id13, id14, id21 manifest.ID
 
@@ -181,7 +181,7 @@ func TestListAndDeleteSnapshots(t *testing.T) {
 func TestEditSnapshots(t *testing.T) {
 	ctx, env := repotesting.NewEnvironment(t, repotesting.FormatNotImportant)
 
-	si1 := localSource(env, "/dummy/path")
+	si1 := env.LocalPathSourceInfo("/dummy/path")
 
 	var id11 manifest.ID
 

internal/server/api_sources_test.go

@@ -33,7 +33,7 @@ func TestSnapshotCounters(t *testing.T) {
 	require.NoError(t, cli.FetchCSRFTokenForTesting(ctx))
 
 	dir := testutil.TempDirectory(t)
-	si := localSource(env, dir)
+	si := env.LocalPathSourceInfo(dir)
 
 	mustCreateSource(t, cli, dir, &policy.Policy{})
 	require.Len(t, mustListSources(t, cli, &snapshot.SourceInfo{}), 1)
@@ -113,7 +113,7 @@ func TestSourceRefreshesAfterPolicy(t *testing.T) {
 	require.NoError(t, cli.FetchCSRFTokenForTesting(ctx))
 
 	dir := testutil.TempDirectory(t)
-	si := localSource(env, dir)
+	si := env.LocalPathSourceInfo(dir)
 
 	currentHour := clock.Now().Hour()
 

internal/server/util_test.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/kopia/kopia/internal/apiclient"
 	"github.com/kopia/kopia/internal/clock"
-	"github.com/kopia/kopia/internal/repotesting"
 	"github.com/kopia/kopia/internal/serverapi"
 	"github.com/kopia/kopia/internal/testlogging"
 	"github.com/kopia/kopia/internal/uitask"
@@ -88,11 +87,3 @@ func waitForTask(t *testing.T, cli *apiclient.KopiaAPIClient, taskID string, tim
 
 	return lastInfo
 }
-
-func localSource(env *repotesting.Environment, path string) snapshot.SourceInfo {
-	return snapshot.SourceInfo{
-		UserName: env.Repository.ClientOptions().Username,
-		Host:     env.Repository.ClientOptions().Hostname,
-		Path:     path,
-	}
-}

repo/blob/storage.go

@@ -9,8 +9,12 @@ import (
 
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
+
+	"github.com/kopia/kopia/repo/logging"
 )
 
+var log = logging.Module("blob")
+
 // ErrSetTimeUnsupported is returned by implementations of Storage that don't support SetTime.
 var ErrSetTimeUnsupported = errors.Errorf("SetTime is not supported")
 
@@ -182,7 +186,7 @@ func (m *Metadata) String() string {
 var ErrBlobNotFound = errors.New("BLOB not found")
 
 // ListAllBlobs returns Metadata for all blobs in a given storage that have the provided name prefix.
-func ListAllBlobs(ctx context.Context, st Storage, prefix ID) ([]Metadata, error) {
+func ListAllBlobs(ctx context.Context, st Reader, prefix ID) ([]Metadata, error) {
 	var result []Metadata
 
 	err := st.ListBlobs(ctx, prefix, func(bm Metadata) error {
@@ -337,3 +341,24 @@ func PutBlobAndGetMetadata(ctx context.Context, st Storage, blobID ID, data Byte
 		Timestamp: *opts.GetModTime,
 	}, err // nolint:wrapcheck
 }
+
+// ReadBlobMap reads the map of all the blobs indexed by ID.
+func ReadBlobMap(ctx context.Context, br Reader) (map[ID]Metadata, error) {
+	blobMap := map[ID]Metadata{}
+
+	log(ctx).Infof("Listing blobs...")
+
+	if err := br.ListBlobs(ctx, "", func(bm Metadata) error {
+		blobMap[bm.BlobID] = bm
+		if len(blobMap)%10000 == 0 {
+			log(ctx).Infof("  %v blobs...", len(blobMap))
+		}
+		return nil
+	}); err != nil {
+		return nil, errors.Wrap(err, "unable to list blobs")
+	}
+
+	log(ctx).Infof("Listed %v blobs.", len(blobMap))
+
+	return blobMap, nil
+}

snapshot/snapshotfs/snapshot_storage_stats.go

@@ -26,7 +26,7 @@ func CalculateStorageStats(ctx context.Context, rep repo.Repository, manifests [
 
 	var uniqueContents sync.Map
 
-	tw, err := NewTreeWalker(TreeWalkerOptions{
+	tw := NewTreeWalker(TreeWalkerOptions{
 		EntryCallback: func(ctx context.Context, entry fs.Entry, oid object.ID, entryPath string) error {
 			if !entry.IsDir() {
 				atomic.AddInt32(&unique.FileObjectCount, 1)
@@ -71,9 +71,6 @@ func CalculateStorageStats(ctx context.Context, rep repo.Repository, manifests [
 			return nil
 		},
 	})
-	if err != nil {
-		return errors.Wrap(err, "unable to create tree walker")
-	}
 	defer tw.Close()
 
 	src := manifests[0].Source

snapshot/snapshotfs/snapshot_tree_walker.go

@@ -164,7 +164,7 @@ type TreeWalkerOptions struct {
 }
 
 // NewTreeWalker creates new tree walker.
-func NewTreeWalker(options TreeWalkerOptions) (*TreeWalker, error) {
+func NewTreeWalker(options TreeWalkerOptions) *TreeWalker {
 	if options.Parallelism <= 0 {
 		options.Parallelism = runtime.NumCPU() * walkersPerCPU
 	}
@@ -176,5 +176,5 @@ func NewTreeWalker(options TreeWalkerOptions) (*TreeWalker, error) {
 	return &TreeWalker{
 		options: options,
 		wp:      workshare.NewPool(options.Parallelism - 1),
-	}, nil
+	}
 }

snapshot/snapshotfs/snapshot_tree_walker_test.go

@@ -19,15 +19,13 @@ import (
 func TestSnapshotTreeWalker(t *testing.T) {
 	callbackCounter := new(int32)
 
-	w, err := snapshotfs.NewTreeWalker(
+	w := snapshotfs.NewTreeWalker(
 		snapshotfs.TreeWalkerOptions{
 			EntryCallback: func(ctx context.Context, entry fs.Entry, oid object.ID, entryPath string) error {
 				atomic.AddInt32(callbackCounter, 1)
 				return nil
 			},
 		})
-	require.NoError(t, err)
-
 	defer w.Close()
 
 	ctx, env := repotesting.NewEnvironment(t, repotesting.FormatNotImportant)
@@ -81,7 +79,7 @@ func TestSnapshotTreeWalker(t *testing.T) {
 func TestSnapshotTreeWalker_Errors(t *testing.T) {
 	someErr1 := errors.Errorf("some error")
 
-	w, err := snapshotfs.NewTreeWalker(
+	w := snapshotfs.NewTreeWalker(
 		snapshotfs.TreeWalkerOptions{
 			Parallelism: 1,
 			EntryCallback: func(ctx context.Context, entry fs.Entry, oid object.ID, entryPath string) error {
@@ -92,8 +90,6 @@ func TestSnapshotTreeWalker_Errors(t *testing.T) {
 				return nil
 			},
 		})
-	require.NoError(t, err)
-
 	defer w.Close()
 
 	ctx, env := repotesting.NewEnvironment(t, repotesting.FormatNotImportant)
@@ -122,7 +118,7 @@ func TestSnapshotTreeWalker_Errors(t *testing.T) {
 func TestSnapshotTreeWalker_MultipleErrors(t *testing.T) {
 	someErr1 := errors.Errorf("some error")
 
-	w, err := snapshotfs.NewTreeWalker(
+	w := snapshotfs.NewTreeWalker(
 		snapshotfs.TreeWalkerOptions{
 			Parallelism: 1,
 			MaxErrors:   -1,
@@ -138,8 +134,6 @@ func TestSnapshotTreeWalker_MultipleErrors(t *testing.T) {
 				return nil
 			},
 		})
-	require.NoError(t, err)
-
 	defer w.Close()
 
 	ctx, env := repotesting.NewEnvironment(t, repotesting.FormatNotImportant)

snapshot/snapshotfs/snapshot_verifier.go

@@ -0,0 +1,187 @@
+package snapshotfs
+
+import (
+	"context"
+	"io"
+	"math/rand"
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/kopia/kopia/fs"
+	"github.com/kopia/kopia/internal/iocopy"
+	"github.com/kopia/kopia/internal/timetrack"
+	"github.com/kopia/kopia/repo"
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/logging"
+	"github.com/kopia/kopia/repo/object"
+)
+
+var verifierLog = logging.Module("verifier")
+
+type verifyFileWorkItem struct {
+	oid       object.ID
+	entryPath string
+}
+
+// Verifier allows efficient verification of large amounts of filesystem entries in parallel.
+type Verifier struct {
+	throttle timetrack.Throttle
+
+	// +checkatomic
+	queued int32
+	// +checkatomic
+	processed int32
+
+	fileWorkQueue chan verifyFileWorkItem
+	rep           repo.Repository
+	opts          VerifierOptions
+	workersWG     sync.WaitGroup
+
+	blobMap map[blob.ID]blob.Metadata // when != nil, will check that each backing blob exists
+}
+
+// ShowStats logs verification statistics.
+func (v *Verifier) ShowStats(ctx context.Context) {
+	processed := atomic.LoadInt32(&v.processed)
+
+	verifierLog(ctx).Infof("Processed %v objects.", processed)
+}
+
+// VerifyFile verifies a single file object (using content check, blob map check or full read).
+func (v *Verifier) VerifyFile(ctx context.Context, oid object.ID, entryPath string) error {
+	verifierLog(ctx).Debugf("verifying object %v", oid)
+
+	defer func() {
+		atomic.AddInt32(&v.processed, 1)
+	}()
+
+	contentIDs, err := v.rep.VerifyObject(ctx, oid)
+	if err != nil {
+		return errors.Wrap(err, "verify object")
+	}
+
+	if v.blobMap != nil {
+		for _, cid := range contentIDs {
+			ci, err := v.rep.ContentInfo(ctx, cid)
+			if err != nil {
+				return errors.Wrapf(err, "error verifying content %v", cid)
+			}
+
+			if _, ok := v.blobMap[ci.GetPackBlobID()]; !ok {
+				return errors.Errorf("object %v is backed by missing blob %v", oid, ci.GetPackBlobID())
+			}
+		}
+	}
+
+	//nolint:gosec
+	if 100*rand.Float64() < v.opts.VerifyFilesPercent {
+		if err := v.readEntireObject(ctx, oid, entryPath); err != nil {
+			return errors.Wrapf(err, "error reading object %v", oid)
+		}
+	}
+
+	return nil
+}
+
+// verifyObject enqueues a single object for verification.
+func (v *Verifier) verifyObject(ctx context.Context, e fs.Entry, oid object.ID, entryPath string) error {
+	if v.throttle.ShouldOutput(time.Second) {
+		v.ShowStats(ctx)
+	}
+
+	if !e.IsDir() {
+		v.fileWorkQueue <- verifyFileWorkItem{oid, entryPath}
+		atomic.AddInt32(&v.queued, 1)
+	} else {
+		atomic.AddInt32(&v.queued, 1)
+		atomic.AddInt32(&v.processed, 1)
+	}
+
+	return nil
+}
+
+func (v *Verifier) readEntireObject(ctx context.Context, oid object.ID, path string) error {
+	verifierLog(ctx).Debugf("reading object %v %v", oid, path)
+
+	// read the entire file
+	r, err := v.rep.OpenObject(ctx, oid)
+	if err != nil {
+		return errors.Wrapf(err, "unable to open object %v", oid)
+	}
+	defer r.Close() //nolint:errcheck
+
+	return errors.Wrap(iocopy.JustCopy(io.Discard, r), "unable to read data")
+}
+
+// VerifierOptions provides options for the verifier.
+type VerifierOptions struct {
+	VerifyFilesPercent float64
+	FileQueueLength    int
+	Parallelism        int
+	MaxErrors          int
+	BlobMap            map[blob.ID]blob.Metadata
+}
+
+// InParallel starts parallel verification and invokes the provided function which can call
+// call Process() on in the provided TreeWalker.
+func (v *Verifier) InParallel(ctx context.Context, enqueue func(tw *TreeWalker) error) error {
+	tw := NewTreeWalker(TreeWalkerOptions{
+		Parallelism:   v.opts.Parallelism,
+		EntryCallback: v.verifyObject,
+		MaxErrors:     v.opts.MaxErrors,
+	})
+	defer tw.Close()
+
+	v.fileWorkQueue = make(chan verifyFileWorkItem, v.opts.FileQueueLength)
+
+	for i := 0; i < v.opts.Parallelism; i++ {
+		v.workersWG.Add(1)
+
+		go func() {
+			defer v.workersWG.Done()
+
+			for wi := range v.fileWorkQueue {
+				if tw.TooManyErrors() {
+					continue
+				}
+
+				if err := v.VerifyFile(ctx, wi.oid, wi.entryPath); err != nil {
+					tw.ReportError(ctx, wi.entryPath, err)
+				}
+			}
+		}()
+	}
+
+	err := enqueue(tw)
+
+	close(v.fileWorkQueue)
+	v.workersWG.Wait()
+	v.fileWorkQueue = nil
+
+	if err != nil {
+		return err
+	}
+
+	return tw.Err()
+}
+
+// NewVerifier creates a verifier.
+func NewVerifier(ctx context.Context, rep repo.Repository, opts VerifierOptions) *Verifier {
+	if opts.Parallelism == 0 {
+		opts.Parallelism = runtime.NumCPU()
+	}
+
+	if opts.FileQueueLength == 0 {
+		opts.FileQueueLength = 20000
+	}
+
+	return &Verifier{
+		opts:    opts,
+		rep:     rep,
+		blobMap: opts.BlobMap,
+	}
+}

snapshot/snapshotfs/snapshot_verifier_test.go

@@ -0,0 +1,168 @@
+package snapshotfs_test
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+
+	"github.com/kopia/kopia/internal/mockfs"
+	"github.com/kopia/kopia/internal/repotesting"
+	"github.com/kopia/kopia/repo"
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/object"
+	"github.com/kopia/kopia/snapshot/snapshotfs"
+)
+
+func TestSnapshotVerifier(t *testing.T) {
+	ctx, te := repotesting.NewEnvironment(t, repotesting.FormatNotImportant)
+
+	u := snapshotfs.NewUploader(te.RepositoryWriter)
+	dir1 := mockfs.NewDirectory()
+
+	si1 := te.LocalPathSourceInfo("/dummy/path")
+
+	dir1.AddFile("file1", []byte{1, 2, 3}, 0o644)
+	dir1.AddFile("file2", []byte{1, 2, 4}, 0o644)
+	dir1.AddFile("file3", []byte{1, 2, 5}, 0o644)
+
+	var obj1 object.ID
+
+	require.NoError(t, repo.WriteSession(ctx, te.Repository, repo.WriteSessionOptions{}, func(ctx context.Context, w repo.RepositoryWriter) error {
+		snap1, err := u.Upload(ctx, dir1, nil, si1)
+		require.NoError(t, err)
+		obj1 = snap1.RootObjectID()
+		return nil
+	}))
+
+	require.NoError(t, te.RepositoryWriter.Flush(ctx))
+
+	te2 := te.MustOpenAnother(t)
+
+	t.Run("PositiveWithBlobMap", func(t *testing.T) {
+		opts := snapshotfs.VerifierOptions{
+			VerifyFilesPercent: 0,
+			Parallelism:        1,
+			MaxErrors:          3,
+			FileQueueLength:    4,
+		}
+
+		bm, err := blob.ReadBlobMap(ctx, te.RepositoryWriter.BlobReader())
+		require.NoError(t, err)
+
+		opts.BlobMap = bm
+
+		v := snapshotfs.NewVerifier(ctx, te2, opts)
+
+		someErr := errors.Errorf("some error")
+
+		require.ErrorIs(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			return someErr
+		}), someErr)
+
+		require.ErrorIs(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			return someErr
+		}), someErr)
+
+		require.NoError(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			return nil
+		}))
+
+		require.NoError(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			tw.Process(ctx, snapshotfs.DirectoryEntry(te.Repository, obj1, nil), ".")
+			return nil
+		}))
+	})
+
+	t.Run("FullFileReadsAndBlobMap", func(t *testing.T) {
+		// full verification with file reads
+		opts := snapshotfs.VerifierOptions{
+			VerifyFilesPercent: 100,
+			MaxErrors:          30,
+		}
+
+		bm, err := blob.ReadBlobMap(ctx, te.RepositoryWriter.BlobReader())
+		require.NoError(t, err)
+
+		opts.BlobMap = bm
+
+		v := snapshotfs.NewVerifier(ctx, te2, opts)
+
+		require.NoError(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			tw.Process(ctx, snapshotfs.DirectoryEntry(te.Repository, obj1, nil), ".")
+			return nil
+		}))
+
+		// now remove all 'p' blobs from the blob map
+		for k := range opts.BlobMap {
+			if strings.HasPrefix(string(k), "p") {
+				delete(opts.BlobMap, k)
+			}
+		}
+
+		require.ErrorContains(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			tw.Process(ctx, snapshotfs.DirectoryEntry(te.Repository, obj1, nil), ".")
+			return nil
+		}), "encountered 3 errors")
+	})
+
+	t.Run("MaxErrors", func(t *testing.T) {
+		// now set max errors to 1 where we have 3
+		opts := snapshotfs.VerifierOptions{
+			MaxErrors:   1,
+			Parallelism: 1,
+		}
+
+		bm, err := blob.ReadBlobMap(ctx, te.RepositoryWriter.BlobReader())
+		require.NoError(t, err)
+
+		opts.BlobMap = bm
+
+		v := snapshotfs.NewVerifier(ctx, te2, opts)
+
+		require.NoError(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			tw.Process(ctx, snapshotfs.DirectoryEntry(te.Repository, obj1, nil), ".")
+			return nil
+		}))
+
+		// now remove all 'p' blobs from the blob map
+		for k := range opts.BlobMap {
+			if strings.HasPrefix(string(k), "p") {
+				delete(opts.BlobMap, k)
+			}
+		}
+
+		// we have 3 errors but max==1
+		require.ErrorContains(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			tw.Process(ctx, snapshotfs.DirectoryEntry(te.Repository, obj1, nil), ".")
+			return nil
+		}), "is backed by missing blob")
+	})
+
+	t.Run("FullFileReadsNoBlobMap", func(t *testing.T) {
+		opts := snapshotfs.VerifierOptions{
+			VerifyFilesPercent: 100,
+			MaxErrors:          30,
+		}
+		v := snapshotfs.NewVerifier(ctx, te2, opts)
+
+		require.NoError(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			tw.Process(ctx, snapshotfs.DirectoryEntry(te.Repository, obj1, nil), ".")
+			return nil
+		}))
+
+		blobs, err := blob.ListAllBlobs(ctx, te.RepositoryWriter.BlobReader(), "p")
+		require.NoError(t, err)
+
+		for _, bm := range blobs {
+			require.NoError(t, te.RepositoryWriter.BlobStorage().DeleteBlob(ctx, bm.BlobID))
+		}
+
+		require.ErrorContains(t, v.InParallel(ctx, func(tw *snapshotfs.TreeWalker) error {
+			tw.Process(ctx, snapshotfs.DirectoryEntry(te.Repository, obj1, nil), ".")
+			return nil
+		}), "encountered 3 errors")
+	})
+}

snapshot/snapshotgc/gc.go

@@ -33,7 +33,7 @@ func findInUseContentIDs(ctx context.Context, rep repo.Repository, used *sync.Ma
 		return errors.Wrap(err, "unable to load manifest IDs")
 	}
 
-	w, twerr := snapshotfs.NewTreeWalker(snapshotfs.TreeWalkerOptions{
+	w := snapshotfs.NewTreeWalker(snapshotfs.TreeWalkerOptions{
 		EntryCallback: func(ctx context.Context, entry fs.Entry, oid object.ID, entryPath string) error {
 			contentIDs, err := rep.VerifyObject(ctx, oid)
 			if err != nil {
@@ -47,10 +47,6 @@ func findInUseContentIDs(ctx context.Context, rep repo.Repository, used *sync.Ma
 			return nil
 		},
 	})
-	if twerr != nil {
-		return errors.Wrap(twerr, "unable to initialize tree walker")
-	}
-
 	defer w.Close()
 
 	log(ctx).Infof("Looking for active contents...")