restic: Actually parallelize FindUsedBlobs

internal/restic/find.go

@@ -1,6 +1,11 @@
 package restic
 
-import "context"
+import (
+	"context"
+	"sync"
+
+	"golang.org/x/sync/errgroup"
+)
 
 // TreeLoader loads a tree from a repository.
 type TreeLoader interface {
@@ -10,30 +15,38 @@ type TreeLoader interface {
 // FindUsedBlobs traverses the tree ID and adds all seen blobs (trees and data
 // blobs) to the set blobs. Already seen tree blobs will not be visited again.
 func FindUsedBlobs(ctx context.Context, repo TreeLoader, treeID ID, blobs BlobSet) error {
-	h := BlobHandle{ID: treeID, Type: TreeBlob}
+	var lock sync.Mutex
-	if blobs.Has(h) {
-		return nil
-	}
-	blobs.Insert(h)
 
-	tree, err := repo.LoadTree(ctx, treeID)
+	wg, ctx := errgroup.WithContext(ctx)
-	if err != nil {
+	treeStream := StreamTrees(ctx, wg, repo, IDs{treeID}, func(treeID ID) bool {
-		return err
+		// locking is necessary the goroutine below concurrently adds data blobs
-	}
+		lock.Lock()
+		h := BlobHandle{ID: treeID, Type: TreeBlob}
+		blobReferenced := blobs.Has(h)
+		// noop if already referenced
+		blobs.Insert(h)
+		lock.Unlock()
+		return blobReferenced
+	})
 
-	for _, node := range tree.Nodes {
+	wg.Go(func() error {
-		switch node.Type {
+		for tree := range treeStream {
-		case "file":
+			if tree.Error != nil {
-			for _, blob := range node.Content {
+				return tree.Error
-				blobs.Insert(BlobHandle{ID: blob, Type: DataBlob})
 			}
-		case "dir":
+
-			err := FindUsedBlobs(ctx, repo, *node.Subtree, blobs)
+			lock.Lock()
-			if err != nil {
+			for _, node := range tree.Nodes {
-				return err
+				switch node.Type {
+				case "file":
+					for _, blob := range node.Content {
+						blobs.Insert(BlobHandle{ID: blob, Type: DataBlob})
+					}
+				}
 			}
+			lock.Unlock()
 		}
-	}
+		return nil
-
+	})
-	return nil
+	return wg.Wait()
 }