copy: Implement by reusing repack

The repack operation copies all selected blobs from a set of pack files
into new pack files. For prune the source and destination repositories
are identical. To implement copy, just use a different source and
destination repository.

cmd/restic/cmd_copy.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/restic/restic/internal/debug"
+	"github.com/restic/restic/internal/repository"
 	"github.com/restic/restic/internal/restic"
 	"golang.org/x/sync/errgroup"
 
@@ -131,17 +132,11 @@ func runCopy(opts CopyOptions, gopts GlobalOptions, args []string) error {
 			}
 		}
 		Verbosef("  copy started, this may take a while...\n")
-
-		if err := copyTree(ctx, srcRepo, dstRepo, visitedTrees, *sn.Tree); err != nil {
+		if err := copyTree(ctx, srcRepo, dstRepo, visitedTrees, *sn.Tree, gopts.Quiet); err != nil {
 			return err
 		}
 		debug.Log("tree copied")
 
-		if err = dstRepo.Flush(ctx); err != nil {
-			return err
-		}
-		debug.Log("flushed packs and saved index")
-
 		// save snapshot
 		sn.Parent = nil // Parent does not have relevance in the new repo.
 		// Use Original as a persistent snapshot ID
@@ -176,82 +171,61 @@ func similarSnapshots(sna *restic.Snapshot, snb *restic.Snapshot) bool {
 	return true
 }
 
-const numCopyWorkers = 8
-
 func copyTree(ctx context.Context, srcRepo restic.Repository, dstRepo restic.Repository,
-	visitedTrees restic.IDSet, rootTreeID restic.ID) error {
+	visitedTrees restic.IDSet, rootTreeID restic.ID, quiet bool) error {
 
-	idChan := make(chan restic.ID)
-	wg, ctx := errgroup.WithContext(ctx)
+	wg, wgCtx := errgroup.WithContext(ctx)
 
-	treeStream := restic.StreamTrees(ctx, wg, srcRepo, restic.IDs{rootTreeID}, func(treeID restic.ID) bool {
+	treeStream := restic.StreamTrees(wgCtx, wg, srcRepo, restic.IDs{rootTreeID}, func(treeID restic.ID) bool {
 		visited := visitedTrees.Has(treeID)
 		visitedTrees.Insert(treeID)
 		return visited
 	}, nil)
 
+	copyBlobs := restic.NewBlobSet()
+	packList := restic.NewIDSet()
+
+	enqueue := func(h restic.BlobHandle) {
+		pb := srcRepo.Index().Lookup(h)
+		copyBlobs.Insert(h)
+		for _, p := range pb {
+			packList.Insert(p.PackID)
+		}
+	}
+
 	wg.Go(func() error {
-		defer close(idChan)
-		// reused buffer
-		var buf []byte
 		for tree := range treeStream {
 			if tree.Error != nil {
 				return fmt.Errorf("LoadTree(%v) returned error %v", tree.ID.Str(), tree.Error)
 			}
 
 			// Do we already have this tree blob?
-			if !dstRepo.Index().Has(restic.BlobHandle{ID: tree.ID, Type: restic.TreeBlob}) {
+			treeHandle := restic.BlobHandle{ID: tree.ID, Type: restic.TreeBlob}
+			if !dstRepo.Index().Has(treeHandle) {
 				// copy raw tree bytes to avoid problems if the serialization changes
-				var err error
-				buf, err = srcRepo.LoadBlob(ctx, restic.TreeBlob, tree.ID, buf)
-				if err != nil {
-					return fmt.Errorf("LoadBlob(%v) for tree returned error %v", tree.ID, err)
-				}
-
-				_, _, err = dstRepo.SaveBlob(ctx, restic.TreeBlob, buf, tree.ID, false)
-				if err != nil {
-					return fmt.Errorf("SaveBlob(%v) for tree returned error %v", tree.ID.Str(), err)
-				}
+				enqueue(treeHandle)
 			}
 
 			for _, entry := range tree.Nodes {
 				// Recursion into directories is handled by StreamTrees
 				// Copy the blobs for this file.
 				for _, blobID := range entry.Content {
-					select {
-					case idChan <- blobID:
-					case <-ctx.Done():
-						return ctx.Err()
+					h := restic.BlobHandle{Type: restic.DataBlob, ID: blobID}
+					if !dstRepo.Index().Has(h) {
+						enqueue(h)
 					}
 				}
 			}
 		}
 		return nil
 	})
-
-	for i := 0; i < numCopyWorkers; i++ {
-		wg.Go(func() error {
-			// reused buffer
-			var buf []byte
-			for blobID := range idChan {
-				// Do we already have this data blob?
-				if dstRepo.Index().Has(restic.BlobHandle{ID: blobID, Type: restic.DataBlob}) {
-					continue
-				}
-				debug.Log("Copying blob %s\n", blobID.Str())
-				var err error
-				buf, err = srcRepo.LoadBlob(ctx, restic.DataBlob, blobID, buf)
-				if err != nil {
-					return fmt.Errorf("LoadBlob(%v) returned error %v", blobID, err)
-				}
-
-				_, _, err = dstRepo.SaveBlob(ctx, restic.DataBlob, buf, blobID, false)
-				if err != nil {
-					return fmt.Errorf("SaveBlob(%v) returned error %v", blobID, err)
-				}
-			}
-			return nil
-		})
+	err := wg.Wait()
+	if err != nil {
+		return err
 	}
-	return wg.Wait()
+
+	bar := newProgressMax(!quiet, uint64(len(packList)), "packs copied")
+	_, err = repository.Repack(ctx, srcRepo, dstRepo, packList, copyBlobs, bar)
+	bar.Done()
+	return err
 }

cmd/restic/cmd_prune.go

@@ -500,7 +500,7 @@ func prune(opts PruneOptions, gopts GlobalOptions, repo restic.Repository, usedB
 	if len(repackPacks) != 0 {
 		Verbosef("repacking packs\n")
 		bar := newProgressMax(!gopts.Quiet, uint64(len(repackPacks)), "packs repacked")
-		_, err := repository.Repack(ctx, repo, repackPacks, keepBlobs, bar)
+		_, err := repository.Repack(ctx, repo, repo, repackPacks, keepBlobs, bar)
 		bar.Done()
 		if err != nil {
 			return errors.Fatalf("%s", err)