refactored storage.ListBlocks() to be callback based which simplifies usage significantly

block/block_cache.go

@@ -105,21 +105,18 @@ func (c *blockCache) sweepDirectory(ctx context.Context) (err error) {
 	t0 := time.Now()
 	log.Debug().Msg("sweeping cache")
 
-	ctx, cancel := context.WithCancel(ctx)
-	ch := c.cacheStorage.ListBlocks(ctx, "")
-	defer cancel()
-
 	var items []storage.BlockMetadata
-
-	for it := range ch {
-		if it.Error != nil {
-			return fmt.Errorf("error listing cache: %v", it.Error)
-		}
+	err = c.cacheStorage.ListBlocks(ctx, "", func(it storage.BlockMetadata) error {
 		items = append(items, it)
+		return nil
+	})
+
+	if err != nil {
+		return fmt.Errorf("error listing cache: %v", err)
 	}
 
 	sort.Slice(items, func(i, j int) bool {
-		return items[i].TimeStamp.After(items[j].TimeStamp)
+		return items[i].Timestamp.After(items[j].Timestamp)
 	})
 
 	var totalRetainedSize int64

block/block_manager.go

@@ -871,22 +871,21 @@ func (bm *Manager) FindUnreferencedStorageFiles(ctx context.Context) ([]storage.
 	}
 
 	usedPackBlocks := findPackBlocksInUse(infos)
-	ch := bm.st.ListBlocks(ctx, PackBlockPrefix)
-
 	var unused []storage.BlockMetadata
-	for bi := range ch {
-		if bi.Error != nil {
-			return nil, fmt.Errorf("error listing storage blocks: %v", bi.Error)
-		}
-
+	err = bm.st.ListBlocks(ctx, PackBlockPrefix, func(bi storage.BlockMetadata) error {
 		u := usedPackBlocks[bi.BlockID]
 		if u > 0 {
 			log.Printf("pack %v, in use by %v blocks", bi.BlockID, u)
-			continue
+			return nil
 		}
 
 		unused = append(unused, bi)
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error listing storage blocks: %v", err)
 	}
+
 	return unused, nil
 }
 
@@ -1059,31 +1058,18 @@ type cachedList struct {
 // if 'full' is false, the function returns only blocks from the last 2 compactions.
 // The list of blocks is not guaranteed to be sorted.
 func listIndexBlocksFromStorage(ctx context.Context, st storage.Storage) ([]IndexInfo, error) {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	ch := st.ListBlocks(ctx, newIndexBlockPrefix)
-
 	var results []IndexInfo
-	for it := range ch {
-		if it.Error != nil {
-			return nil, it.Error
-		}
-
+	err := st.ListBlocks(ctx, newIndexBlockPrefix, func(it storage.BlockMetadata) error {
 		ii := IndexInfo{
 			FileName:  it.BlockID,
-			Timestamp: it.TimeStamp,
+			Timestamp: it.Timestamp,
 			Length:    it.Length,
 		}
 		results = append(results, ii)
-	}
+		return nil
+	})
 
-	select {
-	case <-ctx.Done():
-		return nil, ctx.Err()
-	default:
-		return results, nil
-	}
+	return results, err
 }
 
 // NewManager creates new block manager with given packing options and a formatter.

cli/command_repository_create.go

@@ -2,6 +2,7 @@ package cli
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 
@@ -44,22 +45,17 @@ func newRepositoryOptionsFromFlags() *repo.NewRepositoryOptions {
 }
 
 func ensureEmpty(ctx context.Context, s storage.Storage) error {
-	ctx, cancel := context.WithCancel(ctx)
-	ch := s.ListBlocks(ctx, "")
-	d, hasData := <-ch
-	cancel()
-
-	if hasData {
-		if d.Error != nil {
-			return d.Error
-		}
-
+	hasDataError := errors.New("has data")
+	err := s.ListBlocks(ctx, "", func(cb storage.BlockMetadata) error {
+		return hasDataError
+	})
+	if err == hasDataError {
 		if !*createOverwrite {
 			return fmt.Errorf("found existing data in storage, specify --overwrite to use anyway")
 		}
 	}
 
-	return nil
+	return err
 }
 
 func runCreateCommandWithStorage(ctx context.Context, st storage.Storage) error {

cli/command_storage_list.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/kopia/kopia/repo"
+	"github.com/kopia/kopia/storage"
 )
 
 var (
@@ -15,27 +16,18 @@ var (
 )
 
 func runListStorageBlocks(ctx context.Context, rep *repo.Repository) error {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	ch := rep.Storage.ListBlocks(ctx, *storageListPrefix)
-
-	for b := range ch {
-		if b.Error != nil {
-			return b.Error
-		}
-
+	return rep.Storage.ListBlocks(ctx, *storageListPrefix, func(b storage.BlockMetadata) error {
 		if *storageListMaxSize != 0 && b.Length > *storageListMaxSize {
-			continue
+			return nil
 		}
 
 		if *storageListMinSize != 0 && b.Length < *storageListMinSize {
-			continue
+			return nil
 		}
 
-		fmt.Printf("%-70v %10v %v\n", b.BlockID, b.Length, b.TimeStamp.Local().Format(timeFormat))
-	}
-
-	return nil
+		fmt.Printf("%-70v %10v %v\n", b.BlockID, b.Length, b.Timestamp.Local().Format(timeFormat))
+		return nil
+	})
 }
 
 func init() {

internal/storagetesting/asserts.go

@@ -37,11 +37,11 @@ func AssertGetBlockNotFound(ctx context.Context, t *testing.T, s storage.Storage
 func AssertListResults(ctx context.Context, t *testing.T, s storage.Storage, prefix string, expected ...string) {
 	var names []string
 
-	ctx, cancel := context.WithCancel(ctx)
-	blocks := s.ListBlocks(ctx, prefix)
-	defer cancel()
-	for e := range blocks {
+	if err := s.ListBlocks(ctx, prefix, func(e storage.BlockMetadata) error {
 		names = append(names, e.BlockID)
+		return nil
+	}); err != nil {
+		t.Fatalf("err: %v", err)
 	}
 
 	if !reflect.DeepEqual(names, expected) {

internal/storagetesting/map.go

@@ -59,36 +59,30 @@ func (s *mapStorage) DeleteBlock(ctx context.Context, id string) error {
 	return nil
 }
 
-func (s *mapStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata {
-	ch := make(chan storage.BlockMetadata)
-	go func() {
-		defer close(ch)
-		s.mutex.RLock()
-		defer s.mutex.RUnlock()
+func (s *mapStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
 
-		keys := []string{}
-		for k := range s.data {
-			if strings.HasPrefix(k, prefix) {
-				keys = append(keys, k)
-			}
+	keys := []string{}
+	for k := range s.data {
+		if strings.HasPrefix(k, prefix) {
+			keys = append(keys, k)
 		}
+	}
 
-		sort.Strings(keys)
+	sort.Strings(keys)
 
-		for _, k := range keys {
-			v := s.data[k]
-			select {
-			case <-ctx.Done():
-				return
-			case ch <- storage.BlockMetadata{
-				BlockID:   k,
-				Length:    int64(len(v)),
-				TimeStamp: s.keyTime[k],
-			}:
-			}
+	for _, k := range keys {
+		v := s.data[k]
+		if err := callback(storage.BlockMetadata{
+			BlockID:   k,
+			Length:    int64(len(v)),
+			Timestamp: s.keyTime[k],
+		}); err != nil {
+			return err
 		}
-	}()
-	return ch
+	}
+	return nil
 }
 
 func (s *mapStorage) Close(ctx context.Context) error {

internal/storagetesting/verify.go

@@ -34,27 +34,5 @@ func VerifyStorage(ctx context.Context, t *testing.T, r storage.Storage) {
 		AssertGetBlock(ctx, t, r, b.blk, b.contents)
 	}
 
-	// List
-	ctx, cancel := context.WithCancel(ctx)
-	ch := r.ListBlocks(ctx, "ab")
-	defer cancel()
-	e1 := verifyNextBlock(t, blocks[0].blk, ch)
-	e2 := verifyNextBlock(t, blocks[2].blk, ch)
-	e3 := verifyNextBlock(t, blocks[3].blk, ch)
-	e4, ok := <-ch
-	if ok {
-		t.Errorf("unexpected item: %v", e4)
-	}
-
-	if e1.TimeStamp.After(e2.TimeStamp) || e2.TimeStamp.After(e3.TimeStamp) {
-		t.Errorf("timings are not sorted: %v %v %v", e1.TimeStamp, e2.TimeStamp, e3.TimeStamp)
-	}
-}
-
-func verifyNextBlock(t *testing.T, expectedBlockID string, ch <-chan storage.BlockMetadata) storage.BlockMetadata {
-	bm, ok := <-ch
-	if !ok || bm.BlockID != expectedBlockID {
-		t.Errorf("missing result: %v", expectedBlockID)
-	}
-	return bm
+	AssertListResults(ctx, t, r, "ab", blocks[0].blk, blocks[2].blk, blocks[3].blk)
 }

storage/filesystem/filesystem_storage.go

@@ -65,53 +65,48 @@ func makeFileName(blockID string) string {
 	return blockID + fsStorageChunkSuffix
 }
 
-func (fs *fsStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata {
-	result := make(chan storage.BlockMetadata)
+func (fs *fsStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error {
+	var walkDir func(string, string) error
 
-	var walkDir func(string, string)
+	walkDir = func(directory string, currentPrefix string) error {
+		entries, err := ioutil.ReadDir(directory)
+		if err != nil {
+			return err
+		}
 
-	walkDir = func(directory string, currentPrefix string) {
-		if entries, err := ioutil.ReadDir(directory); err == nil {
-			//log.Println("Walking", directory, "looking for", prefix)
+		for _, e := range entries {
+			if e.IsDir() {
+				newPrefix := currentPrefix + e.Name()
+				var match bool
 
-			for _, e := range entries {
-				if e.IsDir() {
-					newPrefix := currentPrefix + e.Name()
-					var match bool
+				if len(prefix) > len(newPrefix) {
+					match = strings.HasPrefix(prefix, newPrefix)
+				} else {
+					match = strings.HasPrefix(newPrefix, prefix)
+				}
 
-					if len(prefix) > len(newPrefix) {
-						match = strings.HasPrefix(prefix, newPrefix)
-					} else {
-						match = strings.HasPrefix(newPrefix, prefix)
+				if match {
+					if err := walkDir(directory+"/"+e.Name(), currentPrefix+e.Name()); err != nil {
+						return err
 					}
-
-					if match {
-						walkDir(directory+"/"+e.Name(), currentPrefix+e.Name())
-					}
-				} else if fullID, ok := getstringFromFileName(currentPrefix + e.Name()); ok {
-					if strings.HasPrefix(fullID, prefix) {
-						select {
-						case <-ctx.Done():
-							return
-						case result <- storage.BlockMetadata{
-							BlockID:   fullID,
-							Length:    e.Size(),
-							TimeStamp: e.ModTime(),
-						}:
-						}
+				}
+			} else if fullID, ok := getstringFromFileName(currentPrefix + e.Name()); ok {
+				if strings.HasPrefix(fullID, prefix) {
+					if err := callback(storage.BlockMetadata{
+						BlockID:   fullID,
+						Length:    e.Size(),
+						Timestamp: e.ModTime(),
+					}); err != nil {
+						return err
 					}
 				}
 			}
 		}
+
+		return nil
 	}
 
-	walkDirAndClose := func(directory string) {
-		walkDir(directory, "")
-		close(result)
-	}
-
-	go walkDirAndClose(fs.Path)
-	return result
+	return walkDir(fs.Path, "")
 }
 
 func (fs *fsStorage) PutBlock(ctx context.Context, blockID string, data []byte) error {

storage/gcs/gcs_storage.go

@@ -126,42 +126,28 @@ func (gcs *gcsStorage) getObjectNameString(blockID string) string {
 	return gcs.Prefix + blockID
 }
 
-func (gcs *gcsStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata {
-	ch := make(chan storage.BlockMetadata, 100)
+func (gcs *gcsStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error {
+	lst := gcs.bucket.Objects(gcs.ctx, &gcsclient.Query{
+		Prefix: gcs.getObjectNameString(prefix),
+	})
 
-	go func() {
-		defer close(ch)
-
-		lst := gcs.bucket.Objects(gcs.ctx, &gcsclient.Query{
-			Prefix: gcs.getObjectNameString(prefix),
-		})
-
-		oa, err := lst.Next()
-		for err == nil {
-			bm := storage.BlockMetadata{
-				BlockID:   oa.Name[len(gcs.Prefix):],
-				Length:    oa.Size,
-				TimeStamp: oa.Created,
-			}
-			select {
-			case ch <- bm:
-			case <-ctx.Done():
-				return
-			}
-			oa, err = lst.Next()
+	oa, err := lst.Next()
+	for err == nil {
+		if err = callback(storage.BlockMetadata{
+			BlockID:   oa.Name[len(gcs.Prefix):],
+			Length:    oa.Size,
+			Timestamp: oa.Created,
+		}); err != nil {
+			return err
 		}
+		oa, err = lst.Next()
+	}
 
-		if err != iterator.Done {
-			select {
-			case ch <- storage.BlockMetadata{Error: translateError(err)}:
-				return
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
+	if err != iterator.Done {
+		return err
+	}
 
-	return ch
+	return nil
 }
 
 func (gcs *gcsStorage) ConnectionInfo() storage.ConnectionInfo {

storage/logging/logging_storage.go

@@ -43,11 +43,15 @@ func (s *loggingStorage) DeleteBlock(ctx context.Context, id string) error {
 	return err
 }
 
-func (s *loggingStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata {
+func (s *loggingStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error {
 	t0 := time.Now()
-	ch := s.base.ListBlocks(ctx, prefix)
-	s.printf(s.prefix+"ListBlocks(%q) took %v", prefix, time.Since(t0))
-	return ch
+	cnt := 0
+	err := s.base.ListBlocks(ctx, prefix, func(bi storage.BlockMetadata) error {
+		cnt++
+		return callback(bi)
+	})
+	s.printf(s.prefix+"ListBlocks(%q)=%v returned %v items and took %v", prefix, err, cnt, time.Since(t0))
+	return err
 }
 
 func (s *loggingStorage) Close(ctx context.Context) error {

storage/s3/s3_storage.go

@@ -120,38 +120,25 @@ func (s *s3Storage) getObjectNameString(b string) string {
 	return s.Prefix + b
 }
 
-func (s *s3Storage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata {
-	ch := make(chan storage.BlockMetadata, 100)
-
-	go func() {
-		defer close(ch)
-
-		oi := s.cli.ListObjects(s.BucketName, s.Prefix+prefix, false, ctx.Done())
-		for o := range oi {
-			if err := o.Err; err != nil {
-				select {
-				case ch <- storage.BlockMetadata{Error: translateError(err)}:
-					return
-				case <-ctx.Done():
-					return
-				}
-			}
-
-			bm := storage.BlockMetadata{
-				BlockID:   o.Key[len(s.Prefix):],
-				Length:    o.Size,
-				TimeStamp: o.LastModified,
-			}
-
-			select {
-			case ch <- bm:
-			case <-ctx.Done():
-				return
-			}
+func (s *s3Storage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error {
+	oi := s.cli.ListObjects(s.BucketName, s.Prefix+prefix, false, ctx.Done())
+	for o := range oi {
+		if err := o.Err; err != nil {
+			return err
 		}
-	}()
 
-	return ch
+		bm := storage.BlockMetadata{
+			BlockID:   o.Key[len(s.Prefix):],
+			Length:    o.Size,
+			Timestamp: o.LastModified,
+		}
+
+		if err := callback(bm); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func (s *s3Storage) ConnectionInfo() storage.ConnectionInfo {

storage/s3/s3_storage_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/kopia/kopia/internal/storagetesting"
+	"github.com/kopia/kopia/storage"
 	"github.com/minio/minio-go"
 )
 
@@ -85,16 +86,8 @@ func cleanupOldData(ctx context.Context, t *testing.T) {
 		t.Fatalf("err: %v", err)
 	}
 
-	ctx, cancel := context.WithCancel(ctx)
-	items := st.ListBlocks(ctx, "")
-	defer cancel()
-	for it := range items {
-		if it.Error != nil {
-			t.Errorf("can't cleanup: %v", it.Error)
-			return
-		}
-
-		age := time.Since(it.TimeStamp)
+	st.ListBlocks(ctx, "", func(it storage.BlockMetadata) error {
+		age := time.Since(it.Timestamp)
 		if age > cleanupAge {
 			if err := st.DeleteBlock(ctx, it.BlockID); err != nil {
 				t.Errorf("warning: unable to delete %q: %v", it.BlockID, err)
@@ -102,5 +95,6 @@ func cleanupOldData(ctx context.Context, t *testing.T) {
 		} else {
 			log.Printf("keeping %v", it.BlockID)
 		}
-	}
+		return nil
+	})
 }

storage/storage.go

@@ -35,7 +35,7 @@ type Storage interface {
 
 	// ListBlocks returns a channel of BlockMetadata that describes storage blocks with existing name prefixes.
 	// Iteration continues until all blocks have been listed or until client code invokes the returned cancellation function.
-	ListBlocks(ctx context.Context, prefix string) <-chan BlockMetadata
+	ListBlocks(ctx context.Context, prefix string, cb func(bm BlockMetadata) error) error
 
 	// ConnectionInfo returns JSON-serializable data structure containing information required to
 	// connect to storage.
@@ -46,12 +46,10 @@ type Storage interface {
 }
 
 // BlockMetadata represents metadata about a single block in a storage.
-// If Error field is set, no other field values should be assumed to be correct.
 type BlockMetadata struct {
 	BlockID   string
 	Length    int64
-	TimeStamp time.Time
-	Error     error
+	Timestamp time.Time
 }
 
 // ErrBlockNotFound is returned when a block cannot be found in storage.

storage/webdav/webdav_storage.go

@@ -74,51 +74,48 @@ func makeFileName(blockID string) string {
 	return blockID + fsStorageChunkSuffix
 }
 
-func (d *davStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata {
-	result := make(chan storage.BlockMetadata)
+func (d *davStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error {
+	var walkDir func(string, string) error
 
-	var walkDir func(string, string)
+	walkDir = func(urlStr string, currentPrefix string) error {
+		entries, err := d.propFindChildren(urlStr)
+		if err != nil {
+			return err
+		}
 
-	walkDir = func(urlStr string, currentPrefix string) {
-		if entries, err := d.propFindChildren(urlStr); err == nil {
-			for _, e := range entries {
-				if e.isCollection {
-					newPrefix := currentPrefix + e.name
-					var match bool
+		for _, e := range entries {
+			if e.isCollection {
+				newPrefix := currentPrefix + e.name
+				var match bool
 
-					if len(prefix) > len(newPrefix) {
-						match = strings.HasPrefix(prefix, newPrefix)
-					} else {
-						match = strings.HasPrefix(newPrefix, prefix)
+				if len(prefix) > len(newPrefix) {
+					match = strings.HasPrefix(prefix, newPrefix)
+				} else {
+					match = strings.HasPrefix(newPrefix, prefix)
+				}
+
+				if match {
+					if err := walkDir(urlStr+"/"+e.name, currentPrefix+e.name); err != nil {
+						return err
 					}
-
-					if match {
-						walkDir(urlStr+"/"+e.name, currentPrefix+e.name)
-					}
-				} else if fullID, ok := getBlockIDFromFileName(currentPrefix + e.name); ok {
-					if strings.HasPrefix(fullID, prefix) {
-						select {
-						case <-ctx.Done():
-							return
-						case result <- storage.BlockMetadata{
-							BlockID:   fullID,
-							Length:    e.length,
-							TimeStamp: e.modTime,
-						}:
-						}
+				}
+			} else if fullID, ok := getBlockIDFromFileName(currentPrefix + e.name); ok {
+				if strings.HasPrefix(fullID, prefix) {
+					if err := callback(storage.BlockMetadata{
+						BlockID:   fullID,
+						Length:    e.length,
+						Timestamp: e.modTime,
+					}); err != nil {
+						return err
 					}
 				}
 			}
 		}
+
+		return nil
 	}
 
-	walkDirAndClose := func(urlStr string) {
-		walkDir(urlStr, "")
-		close(result)
-	}
-
-	go walkDirAndClose(d.URL)
-	return result
+	return walkDir(d.URL, "")
 }
 
 func (d *davStorage) makeCollectionAll(urlStr string) error {