Merge pull request #5558 from MichaelEischer/simplify-blob-upload

repository: enforce correct usage of SaveBlob

internal/repository/fuzz_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/restic/restic/internal/restic"
-	"golang.org/x/sync/errgroup"
+	rtest "github.com/restic/restic/internal/test"
 )
 
 // Test saving a blob and loading it again, with varying buffer sizes.
@@ -20,17 +20,10 @@ func FuzzSaveLoadBlob(f *testing.F) {
 		id := restic.Hash(blob)
 		repo, _, _ := TestRepositoryWithVersion(t, 2)
 
-		var wg errgroup.Group
-		repo.StartPackUploader(context.TODO(), &wg)
-
-		_, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, blob, id, false)
-		if err != nil {
-			t.Fatal(err)
-		}
-		err = repo.Flush(context.TODO())
-		if err != nil {
-			t.Fatal(err)
-		}
+		rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+			_, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, blob, id, false)
+			return err
+		}))
 
 		buf, err := repo.LoadBlob(context.TODO(), restic.DataBlob, id, make([]byte, buflen))
 		if err != nil {

internal/repository/prune_internal_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/restic/restic/internal/restic"
 	rtest "github.com/restic/restic/internal/test"
 	"github.com/restic/restic/internal/ui/progress"
-	"golang.org/x/sync/errgroup"
 )
 
 // TestPruneMaxUnusedDuplicate checks that MaxUnused correctly accounts for duplicates.
@@ -48,16 +47,14 @@ func TestPruneMaxUnusedDuplicate(t *testing.T) {
 		{bufs[1], bufs[3]},
 		{bufs[2], bufs[3]},
 	} {
-		var wg errgroup.Group
-		repo.StartPackUploader(context.TODO(), &wg)
-
-		for _, blob := range blobs {
-			id, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, blob, restic.ID{}, true)
-			keep.Insert(restic.BlobHandle{Type: restic.DataBlob, ID: id})
-			rtest.OK(t, err)
-		}
-
-		rtest.OK(t, repo.Flush(context.Background()))
+		rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+			for _, blob := range blobs {
+				id, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, blob, restic.ID{}, true)
+				keep.Insert(restic.BlobHandle{Type: restic.DataBlob, ID: id})
+				rtest.OK(t, err)
+			}
+			return nil
+		}))
 	}
 
 	opts := PruneOptions{

internal/repository/prune_test.go

@@ -13,7 +13,6 @@ import (
 	"github.com/restic/restic/internal/restic"
 	rtest "github.com/restic/restic/internal/test"
 	"github.com/restic/restic/internal/ui/progress"
-	"golang.org/x/sync/errgroup"
 )
 
 func testPrune(t *testing.T, opts repository.PruneOptions, errOnUnused bool) {
@@ -26,16 +25,16 @@ func testPrune(t *testing.T, opts repository.PruneOptions, errOnUnused bool) {
 	createRandomBlobs(t, random, repo, 5, 0.5, true)
 	keep, _ := selectBlobs(t, random, repo, 0.5)
 
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
-	// duplicate a few blobs to exercise those code paths
-	for blob := range keep {
-		buf, err := repo.LoadBlob(context.TODO(), blob.Type, blob.ID, nil)
-		rtest.OK(t, err)
-		_, _, _, err = repo.SaveBlob(context.TODO(), blob.Type, buf, blob.ID, true)
-		rtest.OK(t, err)
-	}
-	rtest.OK(t, repo.Flush(context.TODO()))
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		// duplicate a few blobs to exercise those code paths
+		for blob := range keep {
+			buf, err := repo.LoadBlob(ctx, blob.Type, blob.ID, nil)
+			rtest.OK(t, err)
+			_, _, _, err = uploader.SaveBlob(ctx, blob.Type, buf, blob.ID, true)
+			rtest.OK(t, err)
+		}
+		return nil
+	}))
 
 	plan, err := repository.PlanPrune(context.TODO(), opts, repo, func(ctx context.Context, repo restic.Repository, usedBlobs restic.FindBlobSet) error {
 		for blob := range keep {
@@ -133,20 +132,19 @@ func TestPruneSmall(t *testing.T) {
 	const blobSize = 1000 * 1000
 	const numBlobsCreated = 55
 
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
 	keep := restic.NewBlobSet()
-	// we need a minum of 11 packfiles, each packfile will be about 5 Mb long
-	for i := 0; i < numBlobsCreated; i++ {
-		buf := make([]byte, blobSize)
-		random.Read(buf)
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		// we need a minum of 11 packfiles, each packfile will be about 5 Mb long
+		for i := 0; i < numBlobsCreated; i++ {
+			buf := make([]byte, blobSize)
+			random.Read(buf)
 
-		id, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, buf, restic.ID{}, false)
-		rtest.OK(t, err)
-		keep.Insert(restic.BlobHandle{Type: restic.DataBlob, ID: id})
-	}
-
-	rtest.OK(t, repo.Flush(context.Background()))
+			id, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, buf, restic.ID{}, false)
+			rtest.OK(t, err)
+			keep.Insert(restic.BlobHandle{Type: restic.DataBlob, ID: id})
+		}
+		return nil
+	}))
 
 	// gather number of packfiles
 	repoPacks, err := pack.Size(context.TODO(), repo, false)

internal/repository/repack.go

@@ -47,16 +47,12 @@ func Repack(
 		return nil, errors.New("repack step requires a backend connection limit of at least two")
 	}
 
-	wg, wgCtx := errgroup.WithContext(ctx)
-
-	dstRepo.StartPackUploader(wgCtx, wg)
-	wg.Go(func() error {
+	err = dstRepo.WithBlobUploader(ctx, func(ctx context.Context, uploader restic.BlobSaver) error {
 		var err error
-		obsoletePacks, err = repack(wgCtx, repo, dstRepo, packs, keepBlobs, p, logf)
+		obsoletePacks, err = repack(ctx, repo, dstRepo, uploader, packs, keepBlobs, p, logf)
 		return err
 	})
-
-	if err := wg.Wait(); err != nil {
+	if err != nil {
 		return nil, err
 	}
 	return obsoletePacks, nil
@@ -66,6 +62,7 @@ func repack(
 	ctx context.Context,
 	repo restic.Repository,
 	dstRepo restic.Repository,
+	uploader restic.BlobSaver,
 	packs restic.IDSet,
 	keepBlobs repackBlobSet,
 	p *progress.Counter,
@@ -132,7 +129,7 @@ func repack(
 				}
 
 				// We do want to save already saved blobs!
-				_, _, _, err = dstRepo.SaveBlob(wgCtx, blob.Type, buf, blob.ID, true)
+				_, _, _, err = uploader.SaveBlob(wgCtx, blob.Type, buf, blob.ID, true)
 				if err != nil {
 					return err
 				}
@@ -163,9 +160,5 @@ func repack(
 		return nil, err
 	}
 
-	if err := dstRepo.Flush(ctx); err != nil {
-		return nil, err
-	}
-
 	return packs, nil
 }

internal/repository/repack_test.go

@@ -11,7 +11,6 @@ import (
 	"github.com/restic/restic/internal/restic"
 	rtest "github.com/restic/restic/internal/test"
 	"github.com/restic/restic/internal/ui/progress"
-	"golang.org/x/sync/errgroup"
 )
 
 func randomSize(random *rand.Rand, min, max int) int {
@@ -19,50 +18,47 @@ func randomSize(random *rand.Rand, min, max int) int {
 }
 
 func createRandomBlobs(t testing.TB, random *rand.Rand, repo restic.Repository, blobs int, pData float32, smallBlobs bool) {
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
+	// two loops to allow creating multiple pack files
+	for blobs > 0 {
+		rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+			for blobs > 0 {
+				blobs--
+				var (
+					tpe    restic.BlobType
+					length int
+				)
 
-	for i := 0; i < blobs; i++ {
-		var (
-			tpe    restic.BlobType
-			length int
-		)
+				if random.Float32() < pData {
+					tpe = restic.DataBlob
+					if smallBlobs {
+						length = randomSize(random, 1*1024, 20*1024) // 1KiB to 20KiB of data
+					} else {
+						length = randomSize(random, 10*1024, 1024*1024) // 10KiB to 1MiB of data
+					}
+				} else {
+					tpe = restic.TreeBlob
+					length = randomSize(random, 1*1024, 20*1024) // 1KiB to 20KiB
+				}
 
-		if random.Float32() < pData {
-			tpe = restic.DataBlob
-			if smallBlobs {
-				length = randomSize(random, 1*1024, 20*1024) // 1KiB to 20KiB of data
-			} else {
-				length = randomSize(random, 10*1024, 1024*1024) // 10KiB to 1MiB of data
+				buf := make([]byte, length)
+				random.Read(buf)
+
+				id, exists, _, err := uploader.SaveBlob(ctx, tpe, buf, restic.ID{}, false)
+				if err != nil {
+					t.Fatalf("SaveFrom() error %v", err)
+				}
+
+				if exists {
+					t.Errorf("duplicate blob %v/%v ignored", id, restic.DataBlob)
+					continue
+				}
+
+				if rand.Float32() < 0.2 {
+					break
+				}
 			}
-		} else {
-			tpe = restic.TreeBlob
-			length = randomSize(random, 1*1024, 20*1024) // 1KiB to 20KiB
-		}
-
-		buf := make([]byte, length)
-		random.Read(buf)
-
-		id, exists, _, err := repo.SaveBlob(context.TODO(), tpe, buf, restic.ID{}, false)
-		if err != nil {
-			t.Fatalf("SaveFrom() error %v", err)
-		}
-
-		if exists {
-			t.Errorf("duplicate blob %v/%v ignored", id, restic.DataBlob)
-			continue
-		}
-
-		if rand.Float32() < 0.2 {
-			if err = repo.Flush(context.Background()); err != nil {
-				t.Fatalf("repo.Flush() returned error %v", err)
-			}
-			repo.StartPackUploader(context.TODO(), &wg)
-		}
-	}
-
-	if err := repo.Flush(context.Background()); err != nil {
-		t.Fatalf("repo.Flush() returned error %v", err)
+			return nil
+		}))
 	}
 }
 
@@ -74,16 +70,10 @@ func createRandomWrongBlob(t testing.TB, random *rand.Rand, repo restic.Reposito
 	// invert first data byte
 	buf[0] ^= 0xff
 
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
-	_, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, buf, id, false)
-	if err != nil {
-		t.Fatalf("SaveFrom() error %v", err)
-	}
-
-	if err := repo.Flush(context.Background()); err != nil {
-		t.Fatalf("repo.Flush() returned error %v", err)
-	}
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		_, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, buf, id, false)
+		return err
+	}))
 	return restic.BlobHandle{ID: id, Type: restic.DataBlob}
 }
 
@@ -349,24 +339,23 @@ func testRepackBlobFallback(t *testing.T, version uint) {
 	modbuf[0] ^= 0xff
 
 	// create pack with broken copy
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
-	_, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, modbuf, id, false)
-	rtest.OK(t, err)
-	rtest.OK(t, repo.Flush(context.Background()))
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		_, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, modbuf, id, false)
+		return err
+	}))
 
 	// find pack with damaged blob
 	keepBlobs := restic.NewBlobSet(restic.BlobHandle{Type: restic.DataBlob, ID: id})
 	rewritePacks := findPacksForBlobs(t, repo, keepBlobs)
 
 	// create pack with valid copy
-	repo.StartPackUploader(context.TODO(), &wg)
-	_, _, _, err = repo.SaveBlob(context.TODO(), restic.DataBlob, buf, id, true)
-	rtest.OK(t, err)
-	rtest.OK(t, repo.Flush(context.Background()))
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		_, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, buf, id, true)
+		return err
+	}))
 
 	// repack must fallback to valid copy
-	_, err = repository.Repack(context.TODO(), repo, repo, rewritePacks, keepBlobs, nil, nil)
+	_, err := repository.Repack(context.TODO(), repo, repo, rewritePacks, keepBlobs, nil, nil)
 	rtest.OK(t, err)
 
 	keepBlobs = restic.NewBlobSet(restic.BlobHandle{Type: restic.DataBlob, ID: id})

internal/repository/repair_index.go

@@ -93,10 +93,6 @@ func RepairIndex(ctx context.Context, repo *Repository, opts RepairIndexOptions,
 		}
 	}
 
-	if err := repo.Flush(ctx); err != nil {
-		return err
-	}
-
 	err = rewriteIndexFiles(ctx, repo, removePacks, oldIndexes, obsoleteIndexes, printer)
 	if err != nil {
 		return err

internal/repository/repair_pack.go

@@ -7,21 +7,17 @@ import (
 
 	"github.com/restic/restic/internal/restic"
 	"github.com/restic/restic/internal/ui/progress"
-	"golang.org/x/sync/errgroup"
 )
 
 func RepairPacks(ctx context.Context, repo *Repository, ids restic.IDSet, printer progress.Printer) error {
-	wg, wgCtx := errgroup.WithContext(ctx)
-	repo.StartPackUploader(wgCtx, wg)
-
 	printer.P("salvaging intact data from specified pack files")
 	bar := printer.NewCounter("pack files")
 	bar.SetMax(uint64(len(ids)))
 	defer bar.Done()
 
-	wg.Go(func() error {
+	err := repo.WithBlobUploader(ctx, func(ctx context.Context, uploader restic.BlobSaver) error {
 		// examine all data the indexes have for the pack file
-		for b := range repo.ListPacksFromIndex(wgCtx, ids) {
+		for b := range repo.ListPacksFromIndex(ctx, ids) {
 			blobs := b.Blobs
 			if len(blobs) == 0 {
 				printer.E("no blobs found for pack %v", b.PackID)
@@ -29,12 +25,12 @@ func RepairPacks(ctx context.Context, repo *Repository, ids restic.IDSet, printe
 				continue
 			}
 
-			err := repo.LoadBlobsFromPack(wgCtx, b.PackID, blobs, func(blob restic.BlobHandle, buf []byte, err error) error {
+			err := repo.LoadBlobsFromPack(ctx, b.PackID, blobs, func(blob restic.BlobHandle, buf []byte, err error) error {
 				if err != nil {
 					printer.E("failed to load blob %v: %v", blob.ID, err)
 					return nil
 				}
-				id, _, _, err := repo.SaveBlob(wgCtx, blob.Type, buf, restic.ID{}, true)
+				id, _, _, err := uploader.SaveBlob(ctx, blob.Type, buf, restic.ID{}, true)
 				if !id.Equal(blob.ID) {
 					panic("pack id mismatch during upload")
 				}
@@ -46,14 +42,12 @@ func RepairPacks(ctx context.Context, repo *Repository, ids restic.IDSet, printe
 			}
 			bar.Add(1)
 		}
-		return repo.Flush(wgCtx)
+		return nil
 	})
-
-	err := wg.Wait()
-	bar.Done()
 	if err != nil {
 		return err
 	}
+	bar.Done()
 
 	// remove salvaged packs from index
 	err = rewriteIndexFiles(ctx, repo, ids, nil, nil, printer)

internal/repository/repository.go

@@ -559,16 +559,30 @@ func (r *Repository) removeUnpacked(ctx context.Context, t restic.FileType, id r
 	return r.be.Remove(ctx, backend.Handle{Type: t, Name: id.String()})
 }
 
-// Flush saves all remaining packs and the index
-func (r *Repository) Flush(ctx context.Context) error {
-	if err := r.flushPacks(ctx); err != nil {
-		return err
-	}
-
-	return r.idx.Flush(ctx, &internalRepository{r})
+func (r *Repository) WithBlobUploader(ctx context.Context, fn func(ctx context.Context, uploader restic.BlobSaver) error) error {
+	wg, ctx := errgroup.WithContext(ctx)
+	r.startPackUploader(ctx, wg)
+	wg.Go(func() error {
+		if err := fn(ctx, &blobSaverRepo{repo: r}); err != nil {
+			return err
+		}
+		if err := r.flush(ctx); err != nil {
+			return fmt.Errorf("error flushing repository: %w", err)
+		}
+		return nil
+	})
+	return wg.Wait()
 }
 
-func (r *Repository) StartPackUploader(ctx context.Context, wg *errgroup.Group) {
+type blobSaverRepo struct {
+	repo *Repository
+}
+
+func (r *blobSaverRepo) SaveBlob(ctx context.Context, t restic.BlobType, buf []byte, id restic.ID, storeDuplicate bool) (newID restic.ID, known bool, size int, err error) {
+	return r.repo.saveBlob(ctx, t, buf, id, storeDuplicate)
+}
+
+func (r *Repository) startPackUploader(ctx context.Context, wg *errgroup.Group) {
 	if r.packerWg != nil {
 		panic("uploader already started")
 	}
@@ -584,6 +598,15 @@ func (r *Repository) StartPackUploader(ctx context.Context, wg *errgroup.Group)
 	})
 }
 
+// Flush saves all remaining packs and the index
+func (r *Repository) flush(ctx context.Context) error {
+	if err := r.flushPacks(ctx); err != nil {
+		return err
+	}
+
+	return r.idx.Flush(ctx, &internalRepository{r})
+}
+
 // FlushPacks saves all remaining packs.
 func (r *Repository) flushPacks(ctx context.Context) error {
 	if r.packerWg == nil {
@@ -697,7 +720,7 @@ func (r *Repository) createIndexFromPacks(ctx context.Context, packsize map[rest
 
 	// track spawned goroutines using wg, create a new context which is
 	// cancelled as soon as an error occurs.
-	wg, ctx := errgroup.WithContext(ctx)
+	wg, wgCtx := errgroup.WithContext(ctx)
 
 	type FileInfo struct {
 		restic.ID
@@ -710,8 +733,8 @@ func (r *Repository) createIndexFromPacks(ctx context.Context, packsize map[rest
 		defer close(ch)
 		for id, size := range packsize {
 			select {
-			case <-ctx.Done():
-				return ctx.Err()
+			case <-wgCtx.Done():
+				return wgCtx.Err()
 			case ch <- FileInfo{id, size}:
 			}
 		}
@@ -721,14 +744,14 @@ func (r *Repository) createIndexFromPacks(ctx context.Context, packsize map[rest
 	// a worker receives an pack ID from ch, reads the pack contents, and adds them to idx
 	worker := func() error {
 		for fi := range ch {
-			entries, _, err := r.ListPack(ctx, fi.ID, fi.Size)
+			entries, _, err := r.ListPack(wgCtx, fi.ID, fi.Size)
 			if err != nil {
 				debug.Log("unable to list pack file %v", fi.ID.Str())
 				m.Lock()
 				invalid = append(invalid, fi.ID)
 				m.Unlock()
 			}
-			if err := r.idx.StorePack(ctx, fi.ID, entries, &internalRepository{r}); err != nil {
+			if err := r.idx.StorePack(wgCtx, fi.ID, entries, &internalRepository{r}); err != nil {
 				return err
 			}
 			p.Add(1)
@@ -749,6 +772,12 @@ func (r *Repository) createIndexFromPacks(ctx context.Context, packsize map[rest
 		return invalid, err
 	}
 
+	// flush the index to the repository
+	err = r.flush(ctx)
+	if err != nil {
+		return invalid, err
+	}
+
 	return invalid, nil
 }
 
@@ -905,14 +934,14 @@ func (r *Repository) Close() error {
 	return r.be.Close()
 }
 
-// SaveBlob saves a blob of type t into the repository.
+// saveBlob saves a blob of type t into the repository.
 // It takes care that no duplicates are saved; this can be overwritten
 // by setting storeDuplicate to true.
 // If id is the null id, it will be computed and returned.
 // Also returns if the blob was already known before.
 // If the blob was not known before, it returns the number of bytes the blob
 // occupies in the repo (compressed or not, including encryption overhead).
-func (r *Repository) SaveBlob(ctx context.Context, t restic.BlobType, buf []byte, id restic.ID, storeDuplicate bool) (newID restic.ID, known bool, size int, err error) {
+func (r *Repository) saveBlob(ctx context.Context, t restic.BlobType, buf []byte, id restic.ID, storeDuplicate bool) (newID restic.ID, known bool, size int, err error) {
 
 	if int64(len(buf)) > math.MaxUint32 {
 		return restic.ID{}, false, 0, fmt.Errorf("blob is larger than 4GB")

internal/repository/repository_test.go

@@ -22,7 +22,6 @@ import (
 	"github.com/restic/restic/internal/repository/index"
 	"github.com/restic/restic/internal/restic"
 	rtest "github.com/restic/restic/internal/test"
-	"golang.org/x/sync/errgroup"
 )
 
 var testSizes = []int{5, 23, 2<<18 + 23, 1 << 20}
@@ -52,19 +51,17 @@ func testSave(t *testing.T, version uint, calculateID bool) {
 
 		id := restic.Hash(data)
 
-		var wg errgroup.Group
-		repo.StartPackUploader(context.TODO(), &wg)
-
-		// save
-		inputID := restic.ID{}
-		if !calculateID {
-			inputID = id
-		}
-		sid, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, data, inputID, false)
-		rtest.OK(t, err)
-		rtest.Equals(t, id, sid)
-
-		rtest.OK(t, repo.Flush(context.Background()))
+		rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+			// save
+			inputID := restic.ID{}
+			if !calculateID {
+				inputID = id
+			}
+			sid, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, data, inputID, false)
+			rtest.OK(t, err)
+			rtest.Equals(t, id, sid)
+			return nil
+		}))
 
 		// read back
 		buf, err := repo.LoadBlob(context.TODO(), restic.DataBlob, id, nil)
@@ -98,23 +95,22 @@ func testSavePackMerging(t *testing.T, targetPercentage int, expectedPacks int)
 		// minimum pack size to speed up test
 		PackSize: repository.MinPackSize,
 	})
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
 
 	var ids restic.IDs
-	// add blobs with size targetPercentage / 100 * repo.PackSize to the repository
-	blobSize := repository.MinPackSize / 100
-	for range targetPercentage {
-		data := make([]byte, blobSize)
-		_, err := io.ReadFull(rnd, data)
-		rtest.OK(t, err)
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		// add blobs with size targetPercentage / 100 * repo.PackSize to the repository
+		blobSize := repository.MinPackSize / 100
+		for range targetPercentage {
+			data := make([]byte, blobSize)
+			_, err := io.ReadFull(rnd, data)
+			rtest.OK(t, err)
 
-		sid, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, data, restic.ID{}, false)
-		rtest.OK(t, err)
-		ids = append(ids, sid)
-	}
-
-	rtest.OK(t, repo.Flush(context.Background()))
+			sid, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, data, restic.ID{}, false)
+			rtest.OK(t, err)
+			ids = append(ids, sid)
+		}
+		return nil
+	}))
 
 	// check that all blobs are readable
 	for _, id := range ids {
@@ -146,17 +142,18 @@ func benchmarkSaveAndEncrypt(t *testing.B, version uint) {
 	rtest.OK(t, err)
 
 	id := restic.ID(sha256.Sum256(data))
-	var wg errgroup.Group
-	repo.StartPackUploader(context.Background(), &wg)
 
 	t.ReportAllocs()
 	t.ResetTimer()
 	t.SetBytes(int64(size))
 
-	for i := 0; i < t.N; i++ {
-		_, _, _, err = repo.SaveBlob(context.TODO(), restic.DataBlob, data, id, true)
-		rtest.OK(t, err)
-	}
+	_ = repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		for i := 0; i < t.N; i++ {
+			_, _, _, err = uploader.SaveBlob(ctx, restic.DataBlob, data, id, true)
+			rtest.OK(t, err)
+		}
+		return nil
+	})
 }
 
 func TestLoadBlob(t *testing.T) {
@@ -170,12 +167,12 @@ func testLoadBlob(t *testing.T, version uint) {
 	_, err := io.ReadFull(rnd, buf)
 	rtest.OK(t, err)
 
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
-
-	id, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, buf, restic.ID{}, false)
-	rtest.OK(t, err)
-	rtest.OK(t, repo.Flush(context.Background()))
+	var id restic.ID
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		var err error
+		id, _, _, err = uploader.SaveBlob(ctx, restic.DataBlob, buf, restic.ID{}, false)
+		return err
+	}))
 
 	base := crypto.CiphertextLength(length)
 	for _, testlength := range []int{0, base - 20, base - 1, base, base + 7, base + 15, base + 1000} {
@@ -198,11 +195,12 @@ func TestLoadBlobBroken(t *testing.T) {
 	repo, _ := repository.TestRepositoryWithBackend(t, &damageOnceBackend{Backend: be}, restic.StableRepoVersion, repository.Options{})
 	buf := rtest.Random(42, 1000)
 
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
-	id, _, _, err := repo.SaveBlob(context.TODO(), restic.TreeBlob, buf, restic.ID{}, false)
-	rtest.OK(t, err)
-	rtest.OK(t, repo.Flush(context.Background()))
+	var id restic.ID
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		var err error
+		id, _, _, err = uploader.SaveBlob(ctx, restic.TreeBlob, buf, restic.ID{}, false)
+		return err
+	}))
 
 	// setup cache after saving the blob to make sure that the damageOnceBackend damages the cached data
 	c := cache.TestNewCache(t)
@@ -226,12 +224,12 @@ func benchmarkLoadBlob(b *testing.B, version uint) {
 	_, err := io.ReadFull(rnd, buf)
 	rtest.OK(b, err)
 
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
-
-	id, _, _, err := repo.SaveBlob(context.TODO(), restic.DataBlob, buf, restic.ID{}, false)
-	rtest.OK(b, err)
-	rtest.OK(b, repo.Flush(context.Background()))
+	var id restic.ID
+	rtest.OK(b, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		var err error
+		id, _, _, err = uploader.SaveBlob(ctx, restic.DataBlob, buf, restic.ID{}, false)
+		return err
+	}))
 
 	b.ResetTimer()
 	b.SetBytes(int64(length))
@@ -363,19 +361,19 @@ func TestRepositoryLoadUnpackedRetryBroken(t *testing.T) {
 
 // saveRandomDataBlobs generates random data blobs and saves them to the repository.
 func saveRandomDataBlobs(t testing.TB, repo restic.Repository, num int, sizeMax int) {
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		for i := 0; i < num; i++ {
+			size := rand.Int() % sizeMax
 
-	for i := 0; i < num; i++ {
-		size := rand.Int() % sizeMax
+			buf := make([]byte, size)
+			_, err := io.ReadFull(rnd, buf)
+			rtest.OK(t, err)
 
-		buf := make([]byte, size)
-		_, err := io.ReadFull(rnd, buf)
-		rtest.OK(t, err)
-
-		_, _, _, err = repo.SaveBlob(context.TODO(), restic.DataBlob, buf, restic.ID{}, false)
-		rtest.OK(t, err)
-	}
+			_, _, _, err = uploader.SaveBlob(ctx, restic.DataBlob, buf, restic.ID{}, false)
+			rtest.OK(t, err)
+		}
+		return nil
+	}))
 }
 
 func TestRepositoryIncrementalIndex(t *testing.T) {
@@ -389,14 +387,10 @@ func testRepositoryIncrementalIndex(t *testing.T, version uint) {
 
 	// add a few rounds of packs
 	for j := 0; j < 5; j++ {
-		// add some packs, write intermediate index
+		// add some packs and write index
 		saveRandomDataBlobs(t, repo, 20, 1<<15)
-		rtest.OK(t, repo.Flush(context.TODO()))
 	}
 
-	// save final index
-	rtest.OK(t, repo.Flush(context.TODO()))
-
 	packEntries := make(map[restic.ID]map[restic.ID]struct{})
 
 	err := repo.List(context.TODO(), restic.IndexFile, func(id restic.ID, size int64) error {
@@ -437,11 +431,12 @@ func TestListPack(t *testing.T) {
 	repo, _ := repository.TestRepositoryWithBackend(t, &damageOnceBackend{Backend: be}, restic.StableRepoVersion, repository.Options{})
 	buf := rtest.Random(42, 1000)
 
-	var wg errgroup.Group
-	repo.StartPackUploader(context.TODO(), &wg)
-	id, _, _, err := repo.SaveBlob(context.TODO(), restic.TreeBlob, buf, restic.ID{}, false)
-	rtest.OK(t, err)
-	rtest.OK(t, repo.Flush(context.Background()))
+	var id restic.ID
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaver) error {
+		var err error
+		id, _, _, err = uploader.SaveBlob(ctx, restic.TreeBlob, buf, restic.ID{}, false)
+		return err
+	}))
 
 	// setup cache after saving the blob to make sure that the damageOnceBackend damages the cached data
 	c := cache.TestNewCache(t)