mirror of
https://github.com/pgsty/minio.git
synced 2026-11-09 17:21:51 +03:00
Keeps the contributor commit from PR #286 unchanged and adds the same cleanup on the xl.meta commit-failure path.
This commit is contained in:
@@ -28,10 +28,15 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"slices"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/dustin/go-humanize"
|
||||
"github.com/minio/madmin-go/v3"
|
||||
"github.com/minio/minio/internal/config/storageclass"
|
||||
)
|
||||
|
||||
@@ -1302,3 +1307,247 @@ func TestJoinErrs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type stagedDataGoneDisk struct {
|
||||
StorageAPI
|
||||
}
|
||||
|
||||
func (d stagedDataGoneDisk) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string, opts RenameOptions) (RenameDataResp, error) {
|
||||
if err := d.Delete(ctx, srcVolume, srcPath, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
|
||||
return RenameDataResp{}, err
|
||||
}
|
||||
return d.StorageAPI.RenameData(ctx, srcVolume, srcPath, fi, dstVolume, dstPath, opts)
|
||||
}
|
||||
|
||||
// stagedDataGoneDisks makes every drive of the only set lose the staged data
|
||||
// of a write right before RenameData, as when the caller has already given up
|
||||
// and removed it. It returns a function that restores the drives.
|
||||
func stagedDataGoneDisks(z *erasureServerPools) func() {
|
||||
// Swap the drives behind the set's getDisks under the mutex that
|
||||
// getDisks takes, so that background readers do not race the swap.
|
||||
sets := z.serverPools[0]
|
||||
sets.erasureDisksMu.Lock()
|
||||
disks := sets.erasureDisks[0]
|
||||
original := slices.Clone(disks)
|
||||
for i, disk := range disks {
|
||||
disks[i] = stagedDataGoneDisk{StorageAPI: disk}
|
||||
}
|
||||
sets.erasureDisksMu.Unlock()
|
||||
return func() {
|
||||
sets.erasureDisksMu.Lock()
|
||||
copy(sets.erasureDisks[0], original)
|
||||
sets.erasureDisksMu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// A write whose staged data is gone before the drives rename it fails. The
|
||||
// object directory the drives create for it must not stay behind: it is
|
||||
// invisible while its parent holds other objects, but once they are deleted
|
||||
// the parent can no longer be removed and is listed as an empty prefix.
|
||||
func TestRenameDataStagedDataGoneLeavesNoPrefix(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
obj, dirs, err := prepareErasure16(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer obj.Shutdown(context.Background())
|
||||
defer removeRoots(dirs)
|
||||
z := obj.(*erasureServerPools)
|
||||
const bucket = "bucket"
|
||||
const sibling, putObject, mpuObject = "day=1/b=9/sibling", "day=1/b=9/put", "day=1/b=9/mpu"
|
||||
if err := z.MakeBucket(ctx, bucket, MakeBucketOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body := bytes.Repeat([]byte{'a'}, 8<<20) // above the inline limit: data goes to a data dir
|
||||
if _, err := z.PutObject(ctx, bucket, sibling, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
res, err := z.NewMultipartUpload(ctx, bucket, mpuObject, ObjectOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pi, err := z.PutObjectPart(ctx, bucket, mpuObject, res.UploadID, 1, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
restore := stagedDataGoneDisks(z)
|
||||
if _, err := z.PutObject(ctx, bucket, putObject, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err == nil {
|
||||
t.Error("PutObject succeeded without its staged data")
|
||||
}
|
||||
if _, err := z.CompleteMultipartUpload(ctx, bucket, mpuObject, res.UploadID, []CompletePart{{PartNumber: 1, ETag: pi.ETag}}, ObjectOptions{}); err == nil {
|
||||
t.Error("CompleteMultipartUpload succeeded without its staged data")
|
||||
}
|
||||
restore()
|
||||
|
||||
if _, err := z.DeleteObject(ctx, bucket, sibling, ObjectOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, prefix := range []string{"", "day=1/", "day=1/b=9/"} {
|
||||
res, err := z.ListObjectsV2(ctx, bucket, prefix, "", "/", 1000, false, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(res.Objects) != 0 || len(res.Prefixes) != 0 {
|
||||
t.Errorf("ListObjectsV2 %q: %d objects, prefixes %v, want none", prefix, len(res.Objects), res.Prefixes)
|
||||
}
|
||||
}
|
||||
for _, dir := range dirs {
|
||||
if _, err := os.Stat(filepath.Join(dir, bucket, "day=1")); !os.IsNotExist(err) {
|
||||
t.Errorf("%s: day=1 is still on the drive (%v)", dir, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A failed overwrite whose staged data is gone must keep the existing object.
|
||||
func TestRenameDataStagedDataGoneKeepsExistingObject(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
obj, dirs, err := prepareErasure16(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer obj.Shutdown(context.Background())
|
||||
defer removeRoots(dirs)
|
||||
z := obj.(*erasureServerPools)
|
||||
const bucket, object = "bucket", "day=1/b=9/object"
|
||||
if err := z.MakeBucket(ctx, bucket, MakeBucketOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body := bytes.Repeat([]byte{'a'}, 8<<20) // above the inline limit: data goes to a data dir
|
||||
if _, err := z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
restore := stagedDataGoneDisks(z)
|
||||
other := bytes.Repeat([]byte{'b'}, 8<<20)
|
||||
if _, err := z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(other), int64(len(other)), "", ""), ObjectOptions{}); err == nil {
|
||||
t.Error("overwrite succeeded without its staged data")
|
||||
}
|
||||
restore()
|
||||
|
||||
var buf bytes.Buffer
|
||||
if err := GetObject(ctx, z, bucket, object, 0, int64(len(body)), &buf, "", ObjectOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(buf.Bytes(), body) {
|
||||
t.Error("existing object content changed")
|
||||
}
|
||||
}
|
||||
|
||||
// A failed RenameData that runs late removes its object directory while it is
|
||||
// empty. When it does so on a drive between the directory creation and the
|
||||
// rename of another write of the same key, that write must create the
|
||||
// directory again and succeed instead of failing on the drive.
|
||||
func TestRenameDataSurvivesConcurrentEmptyDirCleanup(t *testing.T) {
|
||||
if runtime.GOOS == globalWindowsOSName {
|
||||
t.Skip("directory creation is not traced on Windows")
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
size int
|
||||
multipart bool
|
||||
}{
|
||||
{"put", 8 << 20, false},
|
||||
{"inline_put", 1 << 10, false},
|
||||
{"multipart", 8 << 20, true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
obj, dirs, err := prepareErasure16(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer obj.Shutdown(context.Background())
|
||||
defer removeRoots(dirs)
|
||||
z := obj.(*erasureServerPools)
|
||||
const bucket, object = "bucket", "day=1/b=9/object"
|
||||
if err := z.MakeBucket(ctx, bucket, MakeBucketOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body := bytes.Repeat([]byte{'a'}, tc.size)
|
||||
var uploadID, etag string
|
||||
if tc.multipart {
|
||||
res, err := z.NewMultipartUpload(ctx, bucket, object, ObjectOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pi, err := z.PutObjectPart(ctx, bucket, object, res.UploadID, 1, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
uploadID, etag = res.UploadID, pi.ETag
|
||||
}
|
||||
|
||||
drives := make(map[string]*xlStorage)
|
||||
for _, disk := range z.serverPools[0].sets[0].getDisks() {
|
||||
s := disk.(*xlStorageDiskIDCheck).storage
|
||||
drives[filepath.Join(s.drivePath, bucket, filepath.FromSlash(object))] = s
|
||||
}
|
||||
var mu sync.Mutex
|
||||
hit := make(map[string]bool)
|
||||
var hits atomic.Int32
|
||||
var lateErrs []error
|
||||
traceOSCalls(t, func(tr madmin.TraceInfo) {
|
||||
if tr.FuncName != "os.Mkdir" || tr.Error != "" {
|
||||
return
|
||||
}
|
||||
s := drives[tr.Path]
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
mu.Lock()
|
||||
first := !hit[tr.Path]
|
||||
hit[tr.Path] = true
|
||||
mu.Unlock()
|
||||
if !first {
|
||||
return
|
||||
}
|
||||
// The write has just created the object directory on this
|
||||
// drive. A late RenameData of the same key, whose staged data
|
||||
// is gone, fails now and removes the empty directory.
|
||||
fi := newFileInfo(object, 8, 8)
|
||||
fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
|
||||
fi.Erasure.Index = 1
|
||||
fi.ModTime = time.Now().UTC()
|
||||
fi.Size = 1
|
||||
fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
|
||||
_, err := s.RenameData(context.Background(), minioMetaTmpBucket, mustGetUUID(), fi, bucket, object, RenameOptions{})
|
||||
if _, serr := os.Stat(tr.Path); !os.IsNotExist(serr) {
|
||||
err = fmt.Errorf("late RenameData left %s (%v, rename error %v)", tr.Path, serr, err)
|
||||
} else if errors.Is(err, errFileNotFound) {
|
||||
err = nil
|
||||
}
|
||||
mu.Lock()
|
||||
if err != nil {
|
||||
lateErrs = append(lateErrs, err)
|
||||
}
|
||||
mu.Unlock()
|
||||
hits.Add(1)
|
||||
})
|
||||
|
||||
if tc.multipart {
|
||||
_, err = z.CompleteMultipartUpload(ctx, bucket, object, uploadID, []CompletePart{{PartNumber: 1, ETag: etag}}, ObjectOptions{})
|
||||
} else {
|
||||
_, err = z.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{})
|
||||
}
|
||||
if n := hits.Load(); n != int32(len(dirs)) {
|
||||
t.Fatalf("late cleanup ran on %d drives, want %d", n, len(dirs))
|
||||
}
|
||||
mu.Lock()
|
||||
for _, lateErr := range lateErrs {
|
||||
t.Error(lateErr)
|
||||
}
|
||||
mu.Unlock()
|
||||
if err != nil {
|
||||
t.Fatalf("write failed after a concurrent cleanup: %v", err)
|
||||
}
|
||||
var buf bytes.Buffer
|
||||
if err := GetObject(ctx, z, bucket, object, 0, int64(len(body)), &buf, "", ObjectOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(buf.Bytes(), body) {
|
||||
t.Fatal("object content differs")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+30
-25
@@ -142,32 +142,37 @@ func renameAll(srcFilePath, dstFilePath, baseDir string) (err error) {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = reliableRename(srcFilePath, dstFilePath, baseDir); err != nil {
|
||||
switch {
|
||||
case isSysErrNotDir(err) && !osIsNotExist(err):
|
||||
// Windows can have both isSysErrNotDir(err) and osIsNotExist(err) returning
|
||||
// true if the source file path contains an non-existent directory. In that case,
|
||||
// we want to return errFileNotFound instead, which will honored in subsequent
|
||||
// switch cases
|
||||
return errFileAccessDenied
|
||||
case isSysErrPathNotFound(err):
|
||||
// This is a special case should be handled only for
|
||||
// windows, because windows API does not return "not a
|
||||
// directory" error message. Handle this specifically here.
|
||||
return errFileAccessDenied
|
||||
case isSysErrCrossDevice(err):
|
||||
return fmt.Errorf("%w (%s)->(%s)", errCrossDeviceLink, srcFilePath, dstFilePath)
|
||||
case osIsNotExist(err):
|
||||
return errFileNotFound
|
||||
case osIsExist(err):
|
||||
// This is returned only when destination is a directory and we
|
||||
// are attempting a rename from file to directory.
|
||||
return errIsNotRegular
|
||||
default:
|
||||
return err
|
||||
}
|
||||
return renameAllErr(reliableRename(srcFilePath, dstFilePath, baseDir), srcFilePath, dstFilePath)
|
||||
}
|
||||
|
||||
// renameAllErr converts an error of reliableRename into the error that
|
||||
// renameAll returns.
|
||||
func renameAllErr(err error, srcFilePath, dstFilePath string) error {
|
||||
switch {
|
||||
case err == nil:
|
||||
return nil
|
||||
case isSysErrNotDir(err) && !osIsNotExist(err):
|
||||
// Windows can have both isSysErrNotDir(err) and osIsNotExist(err) returning
|
||||
// true if the source file path contains an non-existent directory. In that case,
|
||||
// we want to return errFileNotFound instead, which will honored in subsequent
|
||||
// switch cases
|
||||
return errFileAccessDenied
|
||||
case isSysErrPathNotFound(err):
|
||||
// This is a special case should be handled only for
|
||||
// windows, because windows API does not return "not a
|
||||
// directory" error message. Handle this specifically here.
|
||||
return errFileAccessDenied
|
||||
case isSysErrCrossDevice(err):
|
||||
return fmt.Errorf("%w (%s)->(%s)", errCrossDeviceLink, srcFilePath, dstFilePath)
|
||||
case osIsNotExist(err):
|
||||
return errFileNotFound
|
||||
case osIsExist(err):
|
||||
// This is returned only when destination is a directory and we
|
||||
// are attempting a rename from file to directory.
|
||||
return errIsNotRegular
|
||||
default:
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Reliably retries os.RenameAll if for some reason os.RenameAll returns
|
||||
|
||||
@@ -18,6 +18,7 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
@@ -25,9 +26,12 @@ import (
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/minio/madmin-go/v3"
|
||||
)
|
||||
|
||||
func TestSysErrors(t *testing.T) {
|
||||
@@ -179,3 +183,346 @@ func TestXLStorageRenameDataCommitMetaMissing(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A failed commit must not leave behind the object directory that renameAll()
|
||||
// created for it when nothing was moved into it: once the other objects under
|
||||
// its parent are gone, it keeps the parent and every ancestor listed as an
|
||||
// empty prefix. A data directory that was already moved keeps the directory.
|
||||
func TestXLStorageRenameDataCommitMetaMissingRemovesEmptyDirs(t *testing.T) {
|
||||
const object = "day=1/b=9/object"
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
data string // "" (zero bytes), "inline" or "dir"
|
||||
sibling bool // another object under day=1/b=9/
|
||||
}{
|
||||
{"empty_object", "", false},
|
||||
{"inline_object_with_sibling", "inline", true},
|
||||
{"data_directory", "dir", false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
disk, root, err := newXLStorageTestSetup(t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer disk.Close()
|
||||
s := disk.storage
|
||||
if err := s.MakeVol(t.Context(), "bucket"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bucketDir := filepath.Join(root, "bucket")
|
||||
objDir := filepath.Join(bucketDir, filepath.FromSlash(object))
|
||||
siblingMeta := filepath.Join(bucketDir, "day=1", "b=9", "sibling", xlStorageFormatFile)
|
||||
if tc.sibling {
|
||||
if err := os.MkdirAll(filepath.Dir(siblingMeta), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(siblingMeta, []byte("sibling"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
fi := newFileInfo(object, 2, 2)
|
||||
fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
|
||||
fi.Erasure.Index = 1
|
||||
fi.ModTime = time.Now().UTC()
|
||||
fi.Fresh = true
|
||||
ctx := &missingCommitMetaContext{
|
||||
Context: t.Context(),
|
||||
meta: filepath.Join(root, minioMetaTmpBucket, "staged", xlStorageFormatFile),
|
||||
}
|
||||
switch tc.data {
|
||||
case "inline":
|
||||
fi.Size = 1
|
||||
fi.Data = []byte("a")
|
||||
case "dir":
|
||||
fi.Size = 1
|
||||
fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
|
||||
part := filepath.Join(root, minioMetaTmpBucket, "staged", fi.DataDir, "part.1")
|
||||
if err := os.MkdirAll(filepath.Dir(part), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(part, []byte("a"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx.dataDest = filepath.Join(objDir, fi.DataDir)
|
||||
}
|
||||
_, err = s.RenameData(ctx, minioMetaTmpBucket, "staged", fi, "bucket", object, RenameOptions{})
|
||||
if !ctx.removed {
|
||||
t.Fatalf("metadata removal did not execute: removal error = %v, RenameData error = %v", ctx.removeErr, err)
|
||||
}
|
||||
if !errors.Is(err, errCommitMetaMissing) {
|
||||
t.Fatalf("expected commit metadata missing, got %v", err)
|
||||
}
|
||||
|
||||
if tc.data == "dir" {
|
||||
// The moved data directory is not deleted, so the object
|
||||
// directory stays and holds only it.
|
||||
entries, err := os.ReadDir(objDir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(entries) != 1 || entries[0].Name() != fi.DataDir {
|
||||
t.Fatalf("object directory holds %v, want only data directory %s", entries, fi.DataDir)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(objDir, fi.DataDir, "part.1")); err != nil {
|
||||
t.Fatalf("moved data must be kept: %v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if _, err := os.Stat(objDir); !os.IsNotExist(err) {
|
||||
t.Fatalf("empty object directory left after a failed commit (%v)", err)
|
||||
}
|
||||
if tc.sibling {
|
||||
if _, err := os.Stat(siblingMeta); err != nil {
|
||||
t.Fatalf("sibling object must be kept: %v", err)
|
||||
}
|
||||
} else if _, err := os.Stat(filepath.Join(bucketDir, "day=1")); !os.IsNotExist(err) {
|
||||
t.Fatalf("empty parent day=1 left after a failed commit (%v)", err)
|
||||
}
|
||||
if _, err := os.Stat(bucketDir); err != nil {
|
||||
t.Fatalf("bucket directory must be kept: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// traceOSCalls calls hook synchronously after every instrumented OS call
|
||||
// until the test ends, so that a test can run other storage calls at an
|
||||
// exact point inside a storage call without changing production code.
|
||||
func traceOSCalls(t *testing.T, hook func(madmin.TraceInfo)) {
|
||||
t.Helper()
|
||||
before := globalTrace.Subscribers()
|
||||
done := make(chan struct{})
|
||||
if err := globalTrace.Subscribe(madmin.TraceOS, make(chan madmin.TraceInfo, 1), done, func(tr madmin.TraceInfo) bool {
|
||||
hook(tr)
|
||||
return false
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
close(done)
|
||||
// Wait for the unsubscribe, which takes the pubsub write lock, so
|
||||
// that it cannot block a nested trace of a later test.
|
||||
for globalTrace.Subscribers() > before {
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The walk up from a removed path removes only empty directories. Another
|
||||
// walk, for example the cleanup of a second failed write of the same key,
|
||||
// can remove the rest of the chain while this one is between two steps, and
|
||||
// a writer can then commit an object whose xl.meta is the path this walk
|
||||
// removes next. That object must stay.
|
||||
func TestXLStorageWalkUpKeepsCommittedObject(t *testing.T) {
|
||||
const nested, parent = "parent/xl.meta/child", "parent"
|
||||
for _, tc := range []string{"delete_version", "data_rename", "commit"} {
|
||||
t.Run(tc, func(t *testing.T) {
|
||||
disk, root, err := newXLStorageTestSetup(t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer disk.Close()
|
||||
s := disk.storage
|
||||
ctx := t.Context()
|
||||
if err := s.MakeVol(ctx, "bucket"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
newFI := func(object string) FileInfo {
|
||||
fi := newFileInfo(object, 2, 2)
|
||||
fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
|
||||
fi.Erasure.Index = 1
|
||||
fi.ModTime = time.Now().UTC()
|
||||
fi.Fresh = true
|
||||
return fi
|
||||
}
|
||||
inline := func(object string) FileInfo {
|
||||
fi := newFI(object)
|
||||
fi.Size, fi.Data = 1, []byte("x")
|
||||
fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
|
||||
fi.SetInlineData()
|
||||
return fi
|
||||
}
|
||||
childDir := filepath.Join(root, "bucket", filepath.FromSlash(nested))
|
||||
metaDir := filepath.Dir(childDir)
|
||||
parentDir := filepath.Dir(metaDir)
|
||||
|
||||
var nestedFI FileInfo
|
||||
if tc == "delete_version" {
|
||||
nestedFI = inline(nested)
|
||||
if _, err := s.RenameData(ctx, minioMetaTmpBucket, "nested", nestedFI, "bucket", nested, RenameOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
parentFI := inline(parent)
|
||||
var fired atomic.Bool
|
||||
var hookErr error
|
||||
traceOSCalls(t, func(tr madmin.TraceInfo) {
|
||||
if tr.FuncName != "os.Remove" || tr.Error != "" || filepath.Clean(tr.Path) != childDir || !fired.CompareAndSwap(false, true) {
|
||||
return
|
||||
}
|
||||
// This walk has removed child/. Another one removes the rest
|
||||
// of the chain, then a writer commits the object parent,
|
||||
// whose metadata file is parent/xl.meta.
|
||||
for _, dir := range []string{metaDir, parentDir} {
|
||||
if err := os.Remove(dir); err != nil {
|
||||
hookErr = err
|
||||
return
|
||||
}
|
||||
}
|
||||
if _, err := s.RenameData(context.Background(), minioMetaTmpBucket, "writer", parentFI, "bucket", parent, RenameOptions{}); err != nil {
|
||||
hookErr = err
|
||||
}
|
||||
})
|
||||
|
||||
switch tc {
|
||||
case "delete_version":
|
||||
err = s.DeleteVersion(ctx, "bucket", nested, nestedFI, false, DeleteOptions{})
|
||||
case "data_rename":
|
||||
// The staged data directory is gone.
|
||||
fi := newFI(nested)
|
||||
fi.Size = 1
|
||||
fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
|
||||
_, err = s.RenameData(ctx, minioMetaTmpBucket, "late", fi, "bucket", nested, RenameOptions{})
|
||||
case "commit":
|
||||
mctx := &missingCommitMetaContext{
|
||||
Context: ctx,
|
||||
meta: filepath.Join(root, minioMetaTmpBucket, "late", xlStorageFormatFile),
|
||||
}
|
||||
_, err = s.RenameData(mctx, minioMetaTmpBucket, "late", newFI(nested), "bucket", nested, RenameOptions{})
|
||||
}
|
||||
if tc == "delete_version" && err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tc != "delete_version" && err == nil {
|
||||
t.Fatal("RenameData succeeded without its staged data")
|
||||
}
|
||||
if !fired.Load() {
|
||||
t.Fatalf("%s did not remove the empty directory %s", tc, nested)
|
||||
}
|
||||
if hookErr != nil {
|
||||
t.Fatal(hookErr)
|
||||
}
|
||||
got, err := s.ReadVersion(ctx, "", "bucket", parent, parentFI.VersionID, ReadOptions{ReadData: true})
|
||||
if err != nil {
|
||||
t.Fatalf("committed object %s lost: %v", parent, err)
|
||||
}
|
||||
if !bytes.Equal(got.Data, parentFI.Data) {
|
||||
t.Fatalf("committed object %s has data %q, want %q", parent, got.Data, parentFI.Data)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A cleanup that walks up removes each empty directory of a write's
|
||||
// destination path once. When it removes them one by one while the write
|
||||
// creates them, so that every directory the write has just made or found is
|
||||
// gone before it uses it, the write must keep retrying until the walk has
|
||||
// passed and then succeed. That holds also when another write recreates the
|
||||
// directories after a failed attempt, which must not hide the failure.
|
||||
func TestXLStorageRenameDataRetriesWhileCleanupWalksUp(t *testing.T) {
|
||||
if runtime.GOOS == globalWindowsOSName {
|
||||
t.Skip("directory creation is not traced on Windows")
|
||||
}
|
||||
const object = "a/b/object"
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
inline, nested bool
|
||||
}{
|
||||
{"data_dir", false, false},
|
||||
{"inline", true, false},
|
||||
{"data_dir_nested_writer", false, true},
|
||||
} {
|
||||
inline := tc.inline
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
disk, root, err := newXLStorageTestSetup(t)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer disk.Close()
|
||||
s := disk.storage
|
||||
ctx := t.Context()
|
||||
if err := s.MakeVol(ctx, "bucket"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fi := newFileInfo(object, 2, 2)
|
||||
fi.DataDir, fi.VersionID = mustGetUUID(), mustGetUUID()
|
||||
fi.Erasure.Index = 1
|
||||
fi.ModTime = time.Now().UTC()
|
||||
fi.Fresh = true
|
||||
fi.Size = 1
|
||||
fi.AddObjectPart(1, "", 1, 1, fi.ModTime, nil, nil)
|
||||
if inline {
|
||||
fi.Data = []byte("x")
|
||||
fi.SetInlineData()
|
||||
} else {
|
||||
part := filepath.Join(root, minioMetaTmpBucket, "staged", fi.DataDir, "part.1")
|
||||
if err := os.MkdirAll(filepath.Dir(part), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(part, []byte("x"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
objDir := filepath.Join(root, "bucket", filepath.FromSlash(object))
|
||||
bDir := filepath.Dir(objDir)
|
||||
aDir := filepath.Dir(bDir)
|
||||
// The walk removes object/, then b/, then a/. Each step happens
|
||||
// right after the write's next directory creation reached the
|
||||
// directory, whether it created it or found it in place.
|
||||
remove := func(dir string) func() error { return func() error { return os.Remove(dir) } }
|
||||
steps := []struct {
|
||||
after string
|
||||
do func() error
|
||||
}{
|
||||
{objDir, remove(objDir)},
|
||||
{bDir, remove(bDir)},
|
||||
{aDir, remove(aDir)},
|
||||
}
|
||||
if tc.nested {
|
||||
// The next creation of b/ fails. A write of the key
|
||||
// a/b/object/child then creates the directories again.
|
||||
steps = append(steps, struct {
|
||||
after string
|
||||
do func() error
|
||||
}{bDir, func() error { return os.MkdirAll(filepath.Join(objDir, "child"), 0o755) }})
|
||||
}
|
||||
var mu sync.Mutex
|
||||
step := 0
|
||||
var hookErr error
|
||||
traceOSCalls(t, func(tr madmin.TraceInfo) {
|
||||
if tr.FuncName != "os.Mkdir" {
|
||||
return
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if step == len(steps) || filepath.Clean(tr.Path) != steps[step].after {
|
||||
return
|
||||
}
|
||||
if err := steps[step].do(); err != nil && hookErr == nil {
|
||||
hookErr = err
|
||||
}
|
||||
step++
|
||||
})
|
||||
_, err = s.RenameData(ctx, minioMetaTmpBucket, "staged", fi, "bucket", object, RenameOptions{})
|
||||
mu.Lock()
|
||||
done := step
|
||||
mu.Unlock()
|
||||
if hookErr != nil {
|
||||
t.Fatal(hookErr)
|
||||
}
|
||||
if done != len(steps) {
|
||||
t.Fatalf("the walk made %d of %d steps", done, len(steps))
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("write failed while a cleanup walked up its path: %v", err)
|
||||
}
|
||||
got, err := s.ReadVersion(ctx, "", "bucket", object, fi.VersionID, ReadOptions{ReadData: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Size != fi.Size {
|
||||
t.Fatalf("committed size %d, want %d", got.Size, fi.Size)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+51
-4
@@ -2511,8 +2511,11 @@ func (s *xlStorage) deleteFile(basePath, deletePath string, recursive, immediate
|
||||
}
|
||||
|
||||
// Delete parent directory obviously not recursively. Errors for
|
||||
// parent directories shouldn't trickle down.
|
||||
s.deleteFile(basePath, pathutil.Dir(pathutil.Clean(deletePath)), false, false)
|
||||
// parent directories shouldn't trickle down. The trailing slash
|
||||
// removes only an empty directory: a concurrent writer can have
|
||||
// replaced the parent with a file, for example the xl.meta of an
|
||||
// object whose name is that parent path, and it must stay.
|
||||
s.deleteFile(basePath, pathutil.Dir(pathutil.Clean(deletePath))+SlashSeparator, false, false)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -2580,6 +2583,40 @@ func skipAccessChecks(volume string) (ok bool) {
|
||||
return strings.HasPrefix(volume, minioMetaBucket)
|
||||
}
|
||||
|
||||
// renameDataAll is renameAll for the two renames of RenameData. A failed
|
||||
// RenameData, typically one that runs after its caller gave up, removes its
|
||||
// object directory and the parents that are empty, so it can remove the
|
||||
// directories that renameAll is creating for this rename. The directory
|
||||
// creation or the rename then reports "not found" although the source is
|
||||
// still in place: retry, which creates the directories again. A cleanup
|
||||
// removes each directory of the path at most once and each failed attempt
|
||||
// needs at least one such removal, so one retry per path level is enough
|
||||
// against any single cleanup. The decision uses the error of the attempt
|
||||
// itself, since another write can recreate the directories in between.
|
||||
func renameDataAll(srcPath, dstPath, skipParent string) error {
|
||||
if srcPath == "" || dstPath == "" {
|
||||
return errInvalidArgument
|
||||
}
|
||||
if err := checkPathLength(srcPath); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := checkPathLength(dstPath); err != nil {
|
||||
return err
|
||||
}
|
||||
retries := strings.Count(strings.TrimPrefix(pathutil.Clean(dstPath), pathutil.Clean(skipParent)), SlashSeparator)
|
||||
for {
|
||||
err := reliableRename(srcPath, dstPath, skipParent)
|
||||
if err == nil || retries == 0 || !osIsNotExist(err) {
|
||||
return renameAllErr(err, srcPath, dstPath)
|
||||
}
|
||||
if _, serr := Lstat(pathutil.Clean(srcPath)); serr != nil {
|
||||
// The source is gone: a late call, which the caller cleans up.
|
||||
return renameAllErr(err, srcPath, dstPath)
|
||||
}
|
||||
retries--
|
||||
}
|
||||
}
|
||||
|
||||
// RenameData - rename source path to destination path atomically, metadata and data directory.
|
||||
func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string, opts RenameOptions) (res RenameDataResp, err error) {
|
||||
defer func() {
|
||||
@@ -2886,13 +2923,18 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
|
||||
if contextCanceled(ctx) {
|
||||
return res, ctx.Err()
|
||||
}
|
||||
if err = renameAll(srcDataPath, dstDataPath, skipParent); err != nil {
|
||||
if err = renameDataAll(srcDataPath, dstDataPath, skipParent); err != nil {
|
||||
if legacyPreserved {
|
||||
// Any failed rename calls un-roll previous transaction.
|
||||
s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
|
||||
}
|
||||
// if its a partial rename() do not attempt to delete recursively.
|
||||
s.deleteFile(dstVolumeDir, dstDataPath, false, false)
|
||||
// renameAll() creates the object directory before the rename; if the
|
||||
// staged data was already gone, drop it and its parents while empty,
|
||||
// otherwise they outlive the other objects there and are listed as an
|
||||
// empty prefix. The trailing slash removes only an empty directory.
|
||||
s.deleteFile(dstVolumeDir, pathutil.Dir(dstFilePath)+SlashSeparator, false, false)
|
||||
return res, osErrToFileErr(err)
|
||||
}
|
||||
diskHealthCheckOK(ctx, err)
|
||||
@@ -2920,7 +2962,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
|
||||
}
|
||||
|
||||
// Commit meta-file
|
||||
if err = renameAll(srcFilePath, dstFilePath, skipParent); err != nil {
|
||||
if err = renameDataAll(srcFilePath, dstFilePath, skipParent); err != nil {
|
||||
if legacyPreserved {
|
||||
// Any failed rename calls un-roll previous transaction.
|
||||
s.deleteFile(dstVolumeDir, legacyDataPath, true, false)
|
||||
@@ -2928,6 +2970,11 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
|
||||
// if its a partial rename() do not attempt to delete recursively.
|
||||
// this can be healed since all parts are available.
|
||||
s.deleteFile(dstVolumeDir, dstDataPath, false, false)
|
||||
// As with the data rename above, renameAll() may have created the
|
||||
// object directory; drop it and its parents while empty. A moved
|
||||
// data directory keeps it, since the trailing slash removes only an
|
||||
// empty directory.
|
||||
s.deleteFile(dstVolumeDir, pathutil.Dir(dstFilePath)+SlashSeparator, false, false)
|
||||
err = osErrToFileErr(err)
|
||||
if errors.Is(err, errFileNotFound) {
|
||||
// The staged xl.meta disappeared between staging and commit.
|
||||
|
||||
Reference in New Issue
Block a user