Rclone and testing improvements (#1202)

* sharded: added parallel iteration of blobs to improve performance

* retry: reduce first retry delay 1s->100ms

* testing: additional assertions for blob storage testing

* rclone: testing cleanup improvements, re-enabled OneDrive

* cli: added --list-parallelism parameter to fs,webdav,sftp and rclone

* sharded: added dedicated test

cli/storage_filesystem.go

@@ -36,6 +36,7 @@ func (c *storageFilesystemFlags) setup(_ storageProviderServices, cmd *kingpin.C
 	cmd.Flag("file-mode", "File mode for newly created files (0600)").PlaceHolder("MODE").StringVar(&c.connectFileMode)
 	cmd.Flag("dir-mode", "Mode of newly directory files (0700)").PlaceHolder("MODE").StringVar(&c.connectDirMode)
 	cmd.Flag("flat", "Use flat directory structure").BoolVar(&c.connectFlat)
+	cmd.Flag("list-parallelism", "Set list parallelism").Hidden().IntVar(&c.options.ListParallelism)
 }
 
 func (c *storageFilesystemFlags) connect(ctx context.Context, isNew bool) (blob.Storage, error) {

cli/storage_rclone.go

@@ -26,6 +26,7 @@ func (c *storageRcloneFlags) setup(_ storageProviderServices, cmd *kingpin.CmdCl
 	cmd.Flag("embed-rclone-config", "Embed the provider RClone config").ExistingFileVar(&c.embedRCloneConfigFile)
 	cmd.Flag("rclone-debug", "Log rclone output").Hidden().BoolVar(&c.opt.Debug)
 	cmd.Flag("rclone-nowait-for-transfers", "Don't wait for transfers when closing storage").Hidden().BoolVar(&c.opt.NoWaitForTransfers)
+	cmd.Flag("list-parallelism", "Set list parallelism").Hidden().IntVar(&c.opt.ListParallelism)
 }
 
 func (c *storageRcloneFlags) connect(ctx context.Context, isNew bool) (blob.Storage, error) {

cli/storage_sftp.go

@@ -35,6 +35,7 @@ func (c *storageSFTPFlags) setup(_ storageProviderServices, cmd *kingpin.CmdClau
 
 	cmd.Flag("flat", "Use flat directory structure").BoolVar(&c.connectFlat)
 	cmd.Flag("max-connections", "Maximum number of SFTP server connections to establish").Default("1").IntVar(&c.options.MaxConnections)
+	cmd.Flag("list-parallelism", "Set list parallelism").Hidden().IntVar(&c.options.ListParallelism)
 }
 
 func (c *storageSFTPFlags) getOptions() (*sftp.Options, error) {

cli/storage_webdav.go

@@ -20,6 +20,7 @@ func (c *storageWebDAVFlags) setup(_ storageProviderServices, cmd *kingpin.CmdCl
 	cmd.Flag("flat", "Use flat directory structure").BoolVar(&c.connectFlat)
 	cmd.Flag("webdav-username", "WebDAV username").Envar("KOPIA_WEBDAV_USERNAME").StringVar(&c.options.Username)
 	cmd.Flag("webdav-password", "WebDAV password").Envar("KOPIA_WEBDAV_PASSWORD").StringVar(&c.options.Password)
+	cmd.Flag("list-parallelism", "Set list parallelism").Hidden().IntVar(&c.options.ListParallelism)
 }
 
 func (c *storageWebDAVFlags) connect(ctx context.Context, isNew bool) (blob.Storage, error) {

internal/blobtesting/asserts.go

@@ -21,13 +21,10 @@ func AssertGetBlob(ctx context.Context, t *testing.T, s blob.Storage, blobID blo
 	t.Helper()
 
 	b, err := s.GetBlob(ctx, blobID, 0, -1)
-	if err != nil {
-		t.Errorf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected)
-		return
-	}
+	require.NoErrorf(t, err, "GetBlob(%v)", blobID)
 
 	if !bytes.Equal(b, expected) {
-		t.Errorf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected)
+		t.Fatalf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected)
 	}
 
 	half := int64(len(expected) / 2)
@@ -37,33 +34,33 @@ func AssertGetBlob(ctx context.Context, t *testing.T, s blob.Storage, blobID blo
 
 	b, err = s.GetBlob(ctx, blobID, 0, 0)
 	if err != nil {
-		t.Errorf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected)
+		t.Fatalf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected)
 		return
 	}
 
 	if len(b) != 0 {
-		t.Errorf("GetBlob(%v) returned non-zero length: %v", blobID, len(b))
+		t.Fatalf("GetBlob(%v) returned non-zero length: %v", blobID, len(b))
 		return
 	}
 
 	b, err = s.GetBlob(ctx, blobID, 0, half)
 	if err != nil {
-		t.Errorf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected)
+		t.Fatalf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected)
 		return
 	}
 
 	if !bytes.Equal(b, expected[0:half]) {
-		t.Errorf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected[0:half])
+		t.Fatalf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected[0:half])
 	}
 
 	b, err = s.GetBlob(ctx, blobID, half, int64(len(expected))-half)
 	if err != nil {
-		t.Errorf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected)
+		t.Fatalf("GetBlob(%v) returned error %v, expected data: %v", blobID, err, expected)
 		return
 	}
 
 	if !bytes.Equal(b, expected[len(expected)-int(half):]) {
-		t.Errorf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected[len(expected)-int(half):])
+		t.Fatalf("GetBlob(%v) returned %x, but expected %x", blobID, b, expected[len(expected)-int(half):])
 	}
 
 	AssertInvalidOffsetLength(ctx, t, s, blobID, -3, 1)
@@ -77,7 +74,7 @@ func AssertInvalidOffsetLength(ctx context.Context, t *testing.T, s blob.Storage
 	t.Helper()
 
 	if _, err := s.GetBlob(ctx, blobID, offset, length); err == nil {
-		t.Errorf("GetBlob(%v,%v,%v) did not return error for invalid offset/length", blobID, offset, length)
+		t.Fatalf("GetBlob(%v,%v,%v) did not return error for invalid offset/length", blobID, offset, length)
 	}
 }
 
@@ -87,7 +84,7 @@ func AssertGetBlobNotFound(ctx context.Context, t *testing.T, s blob.Storage, bl
 
 	b, err := s.GetBlob(ctx, blobID, 0, -1)
 	if !errors.Is(err, blob.ErrBlobNotFound) || b != nil {
-		t.Errorf("GetBlob(%v) returned %v, %v but expected ErrNotFound", blobID, b, err)
+		t.Fatalf("GetBlob(%v) returned %v, %v but expected ErrNotFound", blobID, b, err)
 	}
 }
 
@@ -97,7 +94,7 @@ func AssertGetMetadataNotFound(ctx context.Context, t *testing.T, s blob.Storage
 
 	_, err := s.GetMetadata(ctx, blobID)
 	if !errors.Is(err, blob.ErrBlobNotFound) {
-		t.Errorf("GetMetadata(%v) returned %v but expected ErrNotFound", blobID, err)
+		t.Fatalf("GetMetadata(%v) returned %v but expected ErrNotFound", blobID, err)
 	}
 }
 
@@ -112,15 +109,15 @@ func AssertListResults(ctx context.Context, t *testing.T, s blob.Storage, prefix
 
 		m2, err := s.GetMetadata(ctx, m.BlobID)
 		if err != nil {
-			t.Errorf("GetMetadata() failed: %v", err)
+			t.Fatalf("GetMetadata() failed: %v", err)
 		}
 
 		if got, want := m2.BlobID, m.BlobID; got != want {
-			t.Errorf("invalid blob ID on %v: %v, want %v", m.BlobID, got, want)
+			t.Fatalf("invalid blob ID on %v: %v, want %v", m.BlobID, got, want)
 		}
 
 		if got, want := m2.Length, m.Length; got != want {
-			t.Errorf("invalid length on %v: %v, want %v", m.BlobID, got, want)
+			t.Fatalf("invalid length on %v: %v, want %v", m.BlobID, got, want)
 		}
 
 		timeDiff := m2.Timestamp.Sub(m.Timestamp)
@@ -130,19 +127,19 @@ func AssertListResults(ctx context.Context, t *testing.T, s blob.Storage, prefix
 
 		// truncated time comparison, because some providers return different precision of time in list vs get
 		if timeDiff > maxTimeDiffBetweenGetAndList {
-			t.Errorf("invalid timestamp on %v: getmetadata returned %v, list returned %v", m.BlobID, m2.Timestamp, m.Timestamp)
+			t.Fatalf("invalid timestamp on %v: getmetadata returned %v, list returned %v", m.BlobID, m2.Timestamp, m.Timestamp)
 		}
 
 		return nil
 	}); err != nil {
-		t.Errorf("err: %v", err)
+		t.Fatalf("err: %v", err)
 	}
 
 	names = sorted(names)
 	want = sorted(want)
 
 	if !reflect.DeepEqual(names, want) {
-		t.Errorf("ListBlobs(%v) returned %v, but wanted %v", prefix, names, want)
+		t.Fatalf("ListBlobs(%v) returned %v, but wanted %v", prefix, names, want)
 	}
 }
 
@@ -156,7 +153,7 @@ func AssertListResultsIDs(ctx context.Context, t *testing.T, s blob.Storage, pre
 		names = append(names, m.BlobID)
 		return nil
 	}); err != nil {
-		t.Errorf("err: %v", err)
+		t.Fatalf("err: %v", err)
 	}
 
 	require.ElementsMatch(t, names, want)

internal/blobtesting/verify.go

@@ -5,11 +5,11 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"reflect"
 	"testing"
 	"time"
 
 	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
 
 	"github.com/kopia/kopia/internal/gather"
 	"github.com/kopia/kopia/repo/blob"
@@ -81,8 +81,30 @@ func VerifyStorage(ctx context.Context, t *testing.T, r blob.Storage) {
 		}
 	})
 
-	AssertListResults(ctx, t, r, "", blocks[0].blk, blocks[1].blk, blocks[2].blk, blocks[3].blk, blocks[4].blk)
-	AssertListResults(ctx, t, r, "ab", blocks[0].blk, blocks[2].blk, blocks[3].blk)
+	t.Run("ListBlobs", func(t *testing.T) {
+		errExpected := errors.New("expected error")
+
+		t.Run("ListErrorNoPrefix", func(t *testing.T) {
+			t.Parallel()
+			require.ErrorIs(t, r.ListBlobs(ctx, "", func(bm blob.Metadata) error {
+				return errExpected
+			}), errExpected)
+		})
+		t.Run("ListErrorWithPrefix", func(t *testing.T) {
+			t.Parallel()
+			require.ErrorIs(t, r.ListBlobs(ctx, "ab", func(bm blob.Metadata) error {
+				return errExpected
+			}), errExpected)
+		})
+		t.Run("ListNoPrefix", func(t *testing.T) {
+			t.Parallel()
+			AssertListResults(ctx, t, r, "", blocks[0].blk, blocks[1].blk, blocks[2].blk, blocks[3].blk, blocks[4].blk)
+		})
+		t.Run("ListWithPrefix", func(t *testing.T) {
+			t.Parallel()
+			AssertListResults(ctx, t, r, "ab", blocks[0].blk, blocks[2].blk, blocks[3].blk)
+		})
+	})
 
 	t.Run("OverwriteBlobs", func(t *testing.T) {
 		for _, b := range blocks {
@@ -91,10 +113,7 @@ func VerifyStorage(ctx context.Context, t *testing.T, r blob.Storage) {
 			t.Run(string(b.blk), func(t *testing.T) {
 				t.Parallel()
 
-				if err := r.PutBlob(ctx, b.blk, gather.FromSlice(b.contents)); err != nil {
-					t.Errorf("can't put blob: %v", err)
-				}
-
+				require.NoErrorf(t, r.PutBlob(ctx, b.blk, gather.FromSlice(b.contents)), "can't put blob: %v", b)
 				AssertGetBlob(ctx, t, r, b.blk, b.contents)
 			})
 		}
@@ -109,29 +128,25 @@ func VerifyStorage(ctx context.Context, t *testing.T, r blob.Storage) {
 			t.Run(string(b.blk), func(t *testing.T) {
 				t.Parallel()
 
-				if err := r.SetTime(ctx, b.blk, ts); errors.Is(err, blob.ErrSetTimeUnsupported) {
+				err := r.SetTime(ctx, b.blk, ts)
+				if errors.Is(err, blob.ErrSetTimeUnsupported) {
 					return
 				}
 
+				require.NoError(t, err)
+
 				md, err := r.GetMetadata(ctx, b.blk)
 				if err != nil {
 					t.Errorf("unable to get blob metadata")
 				}
 
-				if got, want := md.Timestamp, ts; !got.Equal(want) {
-					t.Errorf("invalid time after SetTme(): %vm want %v", got, want)
-				}
+				require.True(t, md.Timestamp.Equal(ts), "invalid time after SetTme(): %vm want %v", md.Timestamp, ts)
 			})
 		}
 	})
 
-	if err := r.DeleteBlob(ctx, blocks[0].blk); err != nil {
-		t.Errorf("unable to delete block: %v", err)
-	}
-
-	if err := r.DeleteBlob(ctx, blocks[0].blk); err != nil {
-		t.Errorf("invalid error when deleting deleted block: %v", err)
-	}
+	require.NoError(t, r.DeleteBlob(ctx, blocks[0].blk))
+	require.NoError(t, r.DeleteBlob(ctx, blocks[0].blk))
 
 	AssertListResults(ctx, t, r, "ab", blocks[2].blk, blocks[3].blk)
 	AssertListResults(ctx, t, r, "", blocks[1].blk, blocks[2].blk, blocks[3].blk, blocks[4].blk)
@@ -149,11 +164,7 @@ func AssertConnectionInfoRoundTrips(ctx context.Context, t *testing.T, s blob.St
 	}
 
 	ci2 := s2.ConnectionInfo()
-	if !reflect.DeepEqual(ci, ci2) {
-		t.Errorf("connection info does not round-trip: %v vs %v", ci, ci2)
-	}
+	require.Equal(t, ci, ci2)
 
-	if err := s2.Close(ctx); err != nil {
-		t.Errorf("unable to close storage: %v", err)
-	}
+	require.NoError(t, s2.Close(ctx))
 }

internal/retry/retry.go

@@ -14,7 +14,7 @@ var log = logging.GetContextLoggerFunc("retry")
 
 var (
 	maxAttempts             = 10
-	retryInitialSleepAmount = 1 * time.Second
+	retryInitialSleepAmount = 100 * time.Millisecond
 	retryMaxSleepAmount     = 32 * time.Second
 )
 

repo/blob/filesystem/filesystem_options.go

@@ -13,6 +13,8 @@ type Options struct {
 
 	FileUID *int `json:"uid,omitempty"`
 	FileGID *int `json:"gid,omitempty"`
+
+	ListParallelism int `json:"listParallelism,omitempty"`
 }
 
 func (fso *Options) fileMode() os.FileMode {

repo/blob/filesystem/filesystem_storage.go

@@ -292,10 +292,11 @@ func New(ctx context.Context, opts *Options) (blob.Storage, error) {
 
 	return &fsStorage{
 		sharded.Storage{
-			Impl:     &fsImpl{Options: *opts},
-			RootPath: opts.Path,
-			Suffix:   fsStorageChunkSuffix,
-			Shards:   opts.shards(),
+			Impl:            &fsImpl{Options: *opts},
+			RootPath:        opts.Path,
+			Suffix:          fsStorageChunkSuffix,
+			Shards:          opts.shards(),
+			ListParallelism: opts.ListParallelism,
 		},
 	}, nil
 }

repo/blob/rclone/rclone_options.go

@@ -11,4 +11,5 @@ type Options struct {
 	NoWaitForTransfers bool     `json:"noWaitForTransfers,omitempty"` // when set to true, don't wait for transfers to finish when closing
 	DirectoryShards    []int    `json:"dirShards"`
 	EmbeddedConfig     string   `json:"embeddedConfig,omitempty"`
+	ListParallelism    int      `json:"listParallelism,omitempty"`
 }

repo/blob/rclone/rclone_storage.go

@@ -53,12 +53,6 @@ func (r *rcloneStorage) PutBlob(ctx context.Context, b blob.ID, data blob.Bytes)
 	return errors.Wrap(r.Storage.PutBlob(ctx, b, data), "error writing blob using WebDAV")
 }
 
-func (r *rcloneStorage) DeleteBlob(ctx context.Context, b blob.ID) error {
-	atomic.StoreInt32(r.changeCount, 1)
-
-	return errors.Wrap(r.Storage.DeleteBlob(ctx, b), "error deleting blob using WebDAV")
-}
-
 func (r *rcloneStorage) ConnectionInfo() blob.ConnectionInfo {
 	return blob.ConnectionInfo{
 		Type:   rcloneStorageType,
@@ -290,6 +284,7 @@ func New(ctx context.Context, opt *Options) (blob.Storage, error) {
 		Username:                            webdavUsername,
 		Password:                            webdavPassword,
 		TrustedServerCertificateFingerprint: hex.EncodeToString(fingerprintBytes[:]),
+		ListParallelism:                     opt.ListParallelism,
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "error connecting to webdav storage")

repo/blob/rclone/rclone_storage_test.go

@@ -26,7 +26,7 @@ const defaultCleanupAge = time.Hour
 
 var rcloneExternalProviders = map[string]string{
 	"GoogleDrive": "gdrive:/kopia",
-	// "OneDrive":    "onedrive:/kopia", broken
+	"OneDrive":    "onedrive:/kopia",
 }
 
 func mustGetRcloneExeOrSkip(t *testing.T) string {
@@ -129,7 +129,6 @@ func TestRCloneStorageInvalidFlags(t *testing.T) {
 }
 
 func TestRCloneProviders(t *testing.T) {
-	t.Parallel()
 	testutil.ProviderTest(t)
 
 	var (
@@ -150,6 +149,11 @@ func TestRCloneProviders(t *testing.T) {
 		rcloneArgs = append(rcloneArgs, "--config="+cfg)
 	}
 
+	rcloneArgs = append(rcloneArgs,
+		"--vfs-cache-max-size=100M",
+		"--vfs-cache-mode=full",
+	)
+
 	if len(rcloneArgs)+len(embeddedConfig) == 0 {
 		t.Skipf("Either KOPIA_RCLONE_EMBEDDED_CONFIG_B64 or KOPIA_RCLONE_CONFIG_FILE must be provided")
 	}
@@ -160,18 +164,26 @@ func TestRCloneProviders(t *testing.T) {
 		rp := rp
 
 		opt := &rclone.Options{
-			RemotePath:     rp,
-			RCloneExe:      rcloneExe,
-			RCloneArgs:     rcloneArgs,
-			EmbeddedConfig: embeddedConfig,
+			RemotePath:      rp,
+			RCloneExe:       rcloneExe,
+			RCloneArgs:      rcloneArgs,
+			EmbeddedConfig:  embeddedConfig,
+			Debug:           true,
+			ListParallelism: 16,
 		}
 
-		t.Run(name, func(t *testing.T) {
+		t.Run("Cleanup-"+name, func(t *testing.T) {
 			t.Parallel()
 
 			ctx := testlogging.Context(t)
 
 			cleanupOldData(ctx, t, opt, defaultCleanupAge)
+		})
+
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			ctx := testlogging.Context(t)
 
 			// we are using shared storage, append a guid so that tests don't collide
 			opt.RemotePath += "/" + uuid.NewString()

repo/blob/sftp/sftp_options.go

@@ -23,6 +23,7 @@ type Options struct {
 	SSHArguments string `json:"sshArguments,omitempty"`
 
 	DirectoryShards []int `json:"dirShards"`
+	ListParallelism int   `json:"listParallelism,omitempty"`
 }
 
 func (sftpo *Options) shards() []int {

repo/blob/sftp/sftp_storage.go

@@ -591,10 +591,11 @@ func New(ctx context.Context, opts *Options) (blob.Storage, error) {
 
 	r := &sftpStorage{
 		sharded.Storage{
-			Impl:     impl,
-			RootPath: opts.Path,
-			Suffix:   fsStorageChunkSuffix,
-			Shards:   opts.shards(),
+			Impl:            impl,
+			RootPath:        opts.Path,
+			Suffix:          fsStorageChunkSuffix,
+			Shards:          opts.shards(),
+			ListParallelism: opts.ListParallelism,
 		},
 	}
 

repo/blob/sharded/sharded.go

@@ -9,7 +9,9 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 
+	"github.com/kopia/kopia/internal/parallelwork"
 	"github.com/kopia/kopia/repo/blob"
 )
 
@@ -29,9 +31,10 @@ type Impl interface {
 type Storage struct {
 	Impl Impl
 
-	RootPath string
-	Suffix   string
-	Shards   []int
+	RootPath        string
+	Suffix          string
+	Shards          []int
+	ListParallelism int
 }
 
 // GetBlob implements blob.Storage.
@@ -56,16 +59,29 @@ func (s Storage) makeFileName(blobID blob.ID) string {
 
 // ListBlobs implements blob.Storage.
 func (s Storage) ListBlobs(ctx context.Context, prefix blob.ID, callback func(blob.Metadata) error) error {
+	pw := parallelwork.NewQueue()
+
+	// channel to which pw will write blob.Metadata, some buf
+	result := make(chan blob.Metadata, 128) // nolint:gomnd
+
+	finished := make(chan struct{})
+	defer close(finished)
+
 	var walkDir func(string, string) error
 
 	walkDir = func(directory string, currentPrefix string) error {
+		select {
+		case <-finished: // already finished
+			return nil
+		default:
+		}
+
 		entries, err := s.Impl.ReadDir(ctx, directory)
 		if err != nil {
 			return errors.Wrap(err, "error reading directory")
 		}
 
 		for _, e := range entries {
-			// nolint:nestif
 			if e.IsDir() {
 				var match bool
 
@@ -77,27 +93,66 @@ func (s Storage) ListBlobs(ctx context.Context, prefix blob.ID, callback func(bl
 				}
 
 				if match {
-					if err := walkDir(directory+"/"+e.Name(), currentPrefix+e.Name()); err != nil {
-						return err
-					}
-				}
-			} else if fullID, ok := s.getBlobIDFromFileName(currentPrefix + e.Name()); ok {
-				if strings.HasPrefix(string(fullID), string(prefix)) {
-					if err := callback(blob.Metadata{
-						BlobID:    fullID,
-						Length:    e.Size(),
-						Timestamp: e.ModTime(),
-					}); err != nil {
-						return err
-					}
+					subdir := directory + "/" + e.Name()
+					subprefix := currentPrefix + e.Name()
+
+					pw.EnqueueFront(ctx, func() error {
+						return walkDir(subdir, subprefix)
+					})
 				}
+
+				continue
+			}
+
+			fullID, ok := s.getBlobIDFromFileName(currentPrefix + e.Name())
+			if !ok {
+				continue
+			}
+
+			if !strings.HasPrefix(string(fullID), string(prefix)) {
+				continue
+			}
+
+			select {
+			case result <- blob.Metadata{
+				BlobID:    fullID,
+				Length:    e.Size(),
+				Timestamp: e.ModTime(),
+			}:
+			case <-finished:
 			}
 		}
 
 		return nil
 	}
 
-	return walkDir(s.RootPath, "")
+	pw.EnqueueFront(ctx, func() error {
+		return walkDir(s.RootPath, "")
+	})
+
+	par := s.ListParallelism
+	if par == 0 {
+		par = 1
+	}
+
+	var eg errgroup.Group
+
+	// start populating the channel in parallel
+	eg.Go(func() error {
+		defer close(result)
+
+		return errors.Wrap(pw.Process(ctx, par), "error processing directory shards")
+	})
+
+	// invoke the callback on the current goroutine until it fails
+	for bm := range result {
+		if err := callback(bm); err != nil {
+			return err
+		}
+	}
+
+	// nolint:wrapcheck
+	return eg.Wait()
 }
 
 // GetMetadata implements blob.Storage.

repo/blob/sharded/sharded_test.go

@@ -0,0 +1,53 @@
+package sharded_test
+
+import (
+	"fmt"
+	"io/ioutil"
+	"path/filepath"
+	"testing"
+
+	"github.com/kopia/kopia/internal/blobtesting"
+	"github.com/kopia/kopia/internal/testlogging"
+	"github.com/kopia/kopia/internal/testutil"
+	"github.com/kopia/kopia/repo/blob/filesystem"
+)
+
+func TestShardedFileStorage(t *testing.T) {
+	t.Parallel()
+
+	ctx := testlogging.Context(t)
+
+	for _, parallel := range []int{0, 1, 4, 16} {
+		t.Run(fmt.Sprintf("parallel-%v", parallel), func(t *testing.T) {
+			for _, shardSpec := range [][]int{
+				{0},
+				{1},
+				{3, 3},
+				{2},
+				{1, 1},
+				{1, 2},
+				{2, 2, 2},
+			} {
+				path := testutil.TempDirectory(t)
+
+				r, err := filesystem.New(ctx, &filesystem.Options{
+					Path:            path,
+					DirectoryShards: shardSpec,
+				})
+
+				ioutil.WriteFile(filepath.Join(path, "foreign-file"), []byte{1, 2, 3}, 0600)
+
+				if r == nil || err != nil {
+					t.Errorf("unexpected result: %v %v", r, err)
+				}
+
+				blobtesting.VerifyStorage(ctx, t, r)
+				blobtesting.AssertConnectionInfoRoundTrips(ctx, t, r)
+
+				if err := r.Close(ctx); err != nil {
+					t.Fatalf("err: %v", err)
+				}
+			}
+		})
+	}
+}

repo/blob/webdav/webdav_options.go

@@ -7,6 +7,7 @@ type Options struct {
 	Username                            string `json:"username,omitempty"`
 	Password                            string `json:"password,omitempty" kopia:"sensitive"`
 	TrustedServerCertificateFingerprint string `json:"trustedServerCertificateFingerprint,omitempty"`
+	ListParallelism                     int    `json:"listParallelism,omitempty"`
 }
 
 func (fso *Options) shards() []int {

repo/blob/webdav/webdav_storage.go

@@ -214,9 +214,10 @@ func New(ctx context.Context, opts *Options) (blob.Storage, error) {
 				Options: *opts,
 				cli:     cli,
 			},
-			RootPath: "",
-			Suffix:   fsStorageChunkSuffix,
-			Shards:   opts.shards(),
+			RootPath:        "",
+			Suffix:          fsStorageChunkSuffix,
+			Shards:          opts.shards(),
+			ListParallelism: opts.ListParallelism,
 		},
 	})