refactor(providers): conditionally retry failed blob renames (#5717)

- refactor(storage): conditionally retry failed blob renames
- only applies to FS storage provider
- when the env var is set
- nit: remove no-longer needed linter annotations
- retry rename in sharded file storage test
- run sub-tests in parallel, instead of parent test
- leverage require in test

- Avoid running the same test multiple times.
The 'parallel' test parameter was not used.
The same test cases were being run 4 times each.
Their total execution time was >12s, after the
change their execution time is ~3s

repo/blob/filesystem/filesystem_storage.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/internal/dirutil"
+	"github.com/kopia/kopia/internal/envflag"
 	"github.com/kopia/kopia/internal/iocopy"
 	"github.com/kopia/kopia/internal/retry"
 	"github.com/kopia/kopia/repo/blob"
@@ -47,31 +48,35 @@ type fsImpl struct {
 var errRetriableInvalidLength = errors.New("invalid length (retriable)")
 
 func (fs *fsImpl) isRetriable(err error) bool {
+	return isRetriable(fs.osi, err)
+}
+
+func isRetriable(osi osInterface, err error) bool {
 	if err == nil {
 		return false
 	}
 
 	err = errors.Cause(err)
 
-	if fs.osi.IsStale(err) {
+	if osi.IsStale(err) {
 		// errors indicative of stale resource handle or invalid
 		// descriptors should not be retried
 		return false
 	}
 
-	if fs.osi.IsNotExist(err) {
+	if osi.IsNotExist(err) {
 		return false
 	}
 
-	if fs.osi.IsExist(err) {
+	if osi.IsExist(err) {
 		return false
 	}
 
-	if fs.osi.IsPathError(err) {
+	if osi.IsPathError(err) {
 		return true
 	}
 
-	if fs.osi.IsLinkError(err) {
+	if osi.IsLinkError(err) {
 		return true
 	}
 
@@ -155,7 +160,6 @@ func (fs *fsImpl) GetMetadataFromPath(ctx context.Context, dirPath, path string)
 	}, fs.isRetriable)
 }
 
-//nolint:wrapcheck
 func (fs *fsImpl) PutBlobInPath(ctx context.Context, dirPath, path string, data blob.Bytes, opts blob.PutOptions) error {
 	_ = dirPath
 
@@ -174,13 +178,12 @@ func (fs *fsImpl) PutBlobInPath(ctx context.Context, dirPath, path string, data
 			return err
 		}
 
-		err = fs.osi.Rename(tempFile, path)
+		err = renameBlob(ctx, fs.osi, tempFile, path)
 		if err != nil {
 			if removeErr := fs.osi.Remove(tempFile); removeErr != nil {
 				log(ctx).Errorf("can't remove temp file: %v", removeErr)
 			}
 
-			//nolint:wrapcheck
 			return err
 		}
 
@@ -211,6 +214,25 @@ func (fs *fsImpl) PutBlobInPath(ctx context.Context, dirPath, path string, data
 	return err
 }
 
+func renameBlob(ctx context.Context, osi osInterface, oldPath, newPath string) error {
+	var attemptDelay time.Duration // zero delay when not retrying
+
+	count := 1
+
+	if envflag.Bool("KOPIA_FS_RETRY_BLOB_RENAME") {
+		count = 10
+		attemptDelay = time.Second
+	}
+
+	return retry.PeriodicallyNoValue(ctx, attemptDelay, count, "rename blob",
+		func() error {
+			return osi.Rename(oldPath, newPath)
+		},
+		func(err error) bool {
+			return isRetriable(osi, err)
+		})
+}
+
 // createTempFileWithData creates a temporary file, writes data to it, syncs and closes it.
 // Returns the name of the temporary file and an error.
 // If there is an error writing, syncing, or closing the file, the temporary file is removed.

repo/blob/sharded/sharded_test.go

@@ -50,41 +50,39 @@ func TestShardedOpenLatestFileStorage(t *testing.T) {
 }
 
 func TestShardedFileStorage(t *testing.T) {
-	t.Parallel()
+	t.Setenv("KOPIA_FS_RETRY_BLOB_RENAME", "true")
 
 	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)
+	shardSpecs := [][]int{
+		{0},
+		{1},
+		{3, 3},
+		{2},
+		{1, 1},
+		{1, 2},
+		{2, 2, 2},
+	}
+	for _, shardSpec := range shardSpecs {
+		t.Run(fmt.Sprintf("shardSpec:%v", shardSpec), func(t *testing.T) {
+			t.Parallel()
 
-				r, err := filesystem.New(ctx, &filesystem.Options{
-					Path:            path,
-					DirectoryShards: shardSpec,
-				}, true)
+			path := testutil.TempDirectory(t)
 
-				os.WriteFile(filepath.Join(path, "foreign-file"), []byte{1, 2, 3}, 0o600)
+			r, err := filesystem.New(ctx, &filesystem.Options{
+				Path:            path,
+				DirectoryShards: shardSpec,
+			}, true)
 
-				if r == nil || err != nil {
-					t.Errorf("unexpected result: %v %v", r, err)
-				}
+			os.WriteFile(filepath.Join(path, "foreign-file"), []byte{1, 2, 3}, 0o600)
 
-				blobtesting.VerifyStorage(ctx, t, r, blob.PutOptions{})
-				blobtesting.AssertConnectionInfoRoundTrips(ctx, t, r)
+			require.NoError(t, err)
+			require.NotNil(t, r)
 
-				if err := r.Close(ctx); err != nil {
-					t.Fatalf("err: %v", err)
-				}
-			}
+			blobtesting.VerifyStorage(ctx, t, r, blob.PutOptions{})
+			blobtesting.AssertConnectionInfoRoundTrips(ctx, t, r)
+
+			require.NoError(t, r.Close(testlogging.ContextForCleanup(t)))
 		})
 	}
 }