diff --git a/repo/blob/filesystem/filesystem_storage.go b/repo/blob/filesystem/filesystem_storage.go index 397409870..2c734f293 100644 --- a/repo/blob/filesystem/filesystem_storage.go +++ b/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. diff --git a/repo/blob/sharded/sharded_test.go b/repo/blob/sharded/sharded_test.go index 8e483518d..7f25d42c8 100644 --- a/repo/blob/sharded/sharded_test.go +++ b/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))) }) } }