list index: use helper in repository package

internal/repository/index_list.go

@@ -0,0 +1,41 @@
+package repository
+
+import (
+	"context"
+	"iter"
+
+	"github.com/restic/restic/internal/errors"
+	"github.com/restic/restic/internal/repository/index"
+	"github.com/restic/restic/internal/restic"
+)
+
+// IndexBlob is one blob handle from an on-disk index file, or an error from loading/decoding
+// that file.
+type IndexBlob struct {
+	Handle restic.BlobHandle
+	Error  error
+}
+
+// AllIndexBlobs streams blob handles from each index file without building a master index.
+func AllIndexBlobs(ctx context.Context, lister restic.Lister, loader restic.LoaderUnpacked) iter.Seq[IndexBlob] {
+	return func(yield func(IndexBlob) bool) {
+		stopIteration := errors.New("stop index blob iteration")
+		err := index.ForAllIndexes(ctx, lister, loader, func(_ restic.ID, idx *index.Index, err error) error {
+			if err != nil {
+				return err
+			}
+			for blob := range idx.Values() {
+				if ctx.Err() != nil {
+					return ctx.Err()
+				}
+				if !yield(IndexBlob{Handle: blob.BlobHandle}) {
+					return stopIteration
+				}
+			}
+			return nil
+		})
+		if err != nil && !errors.Is(err, stopIteration) {
+			yield(IndexBlob{Error: err})
+		}
+	}
+}

internal/repository/index_list_test.go

@@ -0,0 +1,62 @@
+package repository_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/restic/restic/internal/repository"
+	"github.com/restic/restic/internal/restic"
+	rtest "github.com/restic/restic/internal/test"
+)
+
+func TestAllIndexBlobs(t *testing.T) {
+	repo, _, _ := repository.TestRepositoryWithVersion(t, 0)
+
+	want := restic.NewBlobSet()
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaverWithAsync) error {
+		for i := range 5 {
+			data := []byte{byte('a' + i)}
+			id, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, data, restic.ID{}, false)
+			rtest.OK(t, err)
+			want.Insert(restic.BlobHandle{Type: restic.DataBlob, ID: id})
+		}
+		return nil
+	}))
+
+	rtest.OK(t, repo.LoadIndex(context.TODO(), nil))
+
+	fromMaster := restic.NewBlobSet()
+	rtest.OK(t, repo.ListBlobs(context.TODO(), func(pb restic.PackedBlob) {
+		fromMaster.Insert(pb.BlobHandle)
+	}))
+	rtest.Equals(t, want, fromMaster)
+
+	fromStream := restic.NewBlobSet()
+	for entry := range repository.AllIndexBlobs(context.TODO(), repo, repo) {
+		if entry.Error != nil {
+			t.Fatalf("unexpected error: %v", entry.Error)
+		}
+		fromStream.Insert(entry.Handle)
+	}
+	rtest.Equals(t, want, fromStream)
+}
+
+func TestAllIndexBlobsEarlyStop(t *testing.T) {
+	repo, _, _ := repository.TestRepositoryWithVersion(t, 0)
+
+	rtest.OK(t, repo.WithBlobUploader(context.TODO(), func(ctx context.Context, uploader restic.BlobSaverWithAsync) error {
+		for range 5 {
+			_, _, _, err := uploader.SaveBlob(ctx, restic.DataBlob, []byte("test"), restic.ID{}, false)
+			rtest.OK(t, err)
+		}
+		return nil
+	}))
+
+	var count int
+	for entry := range repository.AllIndexBlobs(context.TODO(), repo, repo) {
+		rtest.Assert(t, entry.Error == nil, "unexpected error after early stop: %v", entry.Error)
+		count++
+		break
+	}
+	rtest.Equals(t, 1, count)
+}