From bca40a4ac9e52da1cfe10c9582afef6907cd0f8d Mon Sep 17 00:00:00 2001 From: Denis Babchuk Date: Sat, 17 Oct 2026 22:11:45 +0300 Subject: [PATCH] fix(pools): do not let data movement overwrite a newer copy Signed-off-by: Denis Babchuk --- cmd/erasure-server-pool-consistency.go | 51 ++- cmd/erasure-server-pool-decom.go | 26 +- cmd/erasure-server-pool-migration_test.go | 334 +++++++++++++++++- ...rasure-server-pool-put-conditional_test.go | 2 +- cmd/erasure-server-pool-rebalance.go | 18 +- cmd/erasure-server-pool.go | 27 +- cmd/object-api-errors.go | 11 + 7 files changed, 455 insertions(+), 14 deletions(-) diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go index acac59de5..2272a2afb 100644 --- a/cmd/erasure-server-pool-consistency.go +++ b/cmd/erasure-server-pool-consistency.go @@ -62,6 +62,51 @@ func (z *erasureServerPools) objectPoolInfos(ctx context.Context, bucket, object return copies, nil } +// checkDataMovementCopies rejects moving a version when another pool holds a +// newer copy of it, which reads return instead. A cross-pool overwrite can +// leave such a copy, and a client write can add one while the move copies. +// An equal ModTime marks the same write, e.g. after an interrupted move, and +// is moved as before. While the caller holds the pools-layer object lock, the +// returned resolver serves the set layer's reconciliation of this version from +// the same reads. Without opts.NoLock it takes the read lock itself. +func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket, object string, opts ObjectOptions) (GetObjectInfoFn, error) { + if opts.MTime.IsZero() { + // The set layer would stamp the moved copy as the newest one. + return nil, errInvalidArgument + } + if !opts.NoLock { + lk := z.NewNSLock(bucket, object) + lkctx, err := lk.GetRLock(ctx, globalOperationTimeout) + if err != nil { + return nil, err + } + ctx = lkctx.Context() + defer lk.RUnlock(lkctx) + } + versionID := opts.VersionID + if versionID == "" { + versionID = nullVersionID + } + copies, err := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{VersionID: versionID, NoAuditLog: true}) + if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { + return nil, err + } + for _, copy := range copies { + if copy.Index != opts.SrcPoolIdx && copy.ObjInfo.ModTime.After(opts.MTime) { + return nil, DataMovementOverwriteErr{Bucket: bucket, Object: object, VersionID: opts.VersionID, Err: errDataMovementSuperseded} + } + } + return func(ctx context.Context, b, o string, ropts ObjectOptions) (ObjectInfo, error) { + if ropts.VersionID == "" { + ropts.VersionID = nullVersionID + } + if b != bucket || o != object || ropts.VersionID != versionID { + return z.replicaObjectInfo(ctx, b, o, ropts) + } + return replicaCopiesInfo(copies, err, bucket, object, versionID) + }, nil +} + // Healing also commits object metadata. Both queued healing and drive healing // must use the same namespace as pooled writes, rather than racing them under // a destination set's independent namespace. @@ -137,11 +182,15 @@ func (z *erasureServerPools) replicaObjectInfo(ctx context.Context, bucket, obje opts.VersionID = nullVersionID } copies, err := z.objectPoolInfos(ctx, bucket, object, opts) + return replicaCopiesInfo(copies, err, bucket, object, opts.VersionID) +} + +func replicaCopiesInfo(copies []PoolObjInfo, err error, bucket, object, versionID string) (ObjectInfo, error) { if err != nil { return ObjectInfo{}, err } if copies[0].ObjInfo.DeleteMarker { - return copies[0].ObjInfo, MethodNotAllowed{Bucket: bucket, Object: decodeDirObject(object), VersionID: opts.VersionID} + return copies[0].ObjInfo, MethodNotAllowed{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID} } return mergedPoolObjectInfo(copies), nil } diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go index 0e422c3ae..dee2fdfb4 100644 --- a/cmd/erasure-server-pool-decom.go +++ b/cmd/erasure-server-pool-decom.go @@ -616,6 +616,12 @@ func (z *erasureServerPools) decommissionObject(ctx context.Context, idx int, bu } if objInfo.isMultipart() { + // Completion repeats this check under the object write lock. + if _, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(objInfo.Name), ObjectOptions{ + VersionID: objInfo.VersionID, MTime: objInfo.ModTime, SrcPoolIdx: idx, + }); err != nil { + return err + } res, err := z.NewMultipartUpload(ctx, bucket, objInfo.Name, ObjectOptions{ VersionID: objInfo.VersionID, UserDefined: migrationObjectMetadata(objInfo), @@ -656,6 +662,7 @@ func (z *erasureServerPools) decommissionObject(ctx context.Context, idx int, bu } } _, err = z.CompleteMultipartUpload(ctx, bucket, objInfo.Name, res.UploadID, parts, ObjectOptions{ + VersionID: objInfo.VersionID, SrcPoolIdx: idx, DataMovement: true, MTime: objInfo.ModTime, @@ -870,6 +877,9 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool SkipDecommissioned: true, // make sure we skip the decommissioned pool NoAuditLog: true, }) + if isDataMovementSuperseded(err) { + err = nil + } if err != nil { // This can happen when rebalance stop races with ongoing rebalance workers. // These rebalance failures can be ignored. @@ -898,17 +908,19 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool // gr.Close() is ensured by decommissionObject(). for range 3 { if version.IsRemote() { - if err := z.DecomTieredObject(ctx, bi.Name, version.Name, version, ObjectOptions{ + err := z.DecomTieredObject(ctx, bi.Name, version.Name, version, ObjectOptions{ VersionID: versionID, MTime: version.ModTime, UserDefined: version.Metadata, SrcPoolIdx: idx, DataMovement: true, - }); err != nil { - if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) { - ignore = true - stopFn(0, nil) - } + }) + if isDataMovementSuperseded(err) { + err = nil + } + if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) { + ignore = true + stopFn(0, nil) } if !ignore { stopFn(version.Size, err) @@ -949,7 +961,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool stopFn(version.Size, err) continue } - if err = z.decommissionObject(ctx, idx, bi.Name, gr); err != nil { + if err = z.decommissionObject(ctx, idx, bi.Name, gr); err != nil && !isDataMovementSuperseded(err) { if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) { ignore = true stopFn(0, nil) diff --git a/cmd/erasure-server-pool-migration_test.go b/cmd/erasure-server-pool-migration_test.go index f57ac8d13..7f934ad3b 100644 --- a/cmd/erasure-server-pool-migration_test.go +++ b/cmd/erasure-server-pool-migration_test.go @@ -24,11 +24,13 @@ import ( "fmt" "io" "maps" + "net/http" "strings" "sync" "testing" "time" + "github.com/minio/minio/internal/bucket/lifecycle" xhttp "github.com/minio/minio/internal/http" ) @@ -131,7 +133,10 @@ func migrationTestMover(t *testing.T, z *erasureServerPools, source int, kind st t.Helper() if kind == "rebalance" { z.rebalMu.Lock() - z.rebalMeta = &rebalanceMeta{PoolStats: []*rebalanceStats{{}, {}}} + z.rebalMeta = &rebalanceMeta{PoolStats: make([]*rebalanceStats, len(z.serverPools))} + for i := range z.rebalMeta.PoolStats { + z.rebalMeta.PoolStats[i] = &rebalanceStats{} + } z.rebalMeta.PoolStats[source] = &rebalanceStats{Participating: true, Info: rebalanceInfo{Status: rebalStarted}} z.rebalMu.Unlock() t.Cleanup(func() { z.rebalMu.Lock(); z.rebalMeta = nil; z.rebalMu.Unlock() }) @@ -590,3 +595,330 @@ func TestPoolsMigrationUnreadablePool(t *testing.T) { } } } + +// Three pools give a move two candidate destinations. +func migrationTestThreePools(t *testing.T) (*erasureServerPools, string) { + t.Helper() + ctx, cancel := context.WithCancel(t.Context()) + dirs, err := getRandomDisks(12) + if err != nil { + cancel() + t.Fatal(err) + } + var endpoints EndpointServerPools + for i := range 3 { + endpoints = append(endpoints, mustGetPoolEndpoints(i, dirs[4*i:4*i+4]...)...) + } + obj, _, err := initObjectLayer(ctx, endpoints) + if err != nil { + cancel() + removeRoots(dirs) + t.Fatal(err) + } + z := obj.(*erasureServerPools) + previous := newObjectLayerFn() + setObjectLayer(z) + t.Cleanup(func() { cancel(); z.Shutdown(context.Background()); removeRoots(dirs); setObjectLayer(previous) }) + bucket := "pool-migration" + if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + return z, bucket +} + +type migrationTestGetter func(context.Context, string, string, *HTTPRangeSpec, http.Header, ObjectOptions) (*GetObjectReader, error) + +// migrationTestBody returns the content read for a version, or the read error. +func migrationTestBody(t *testing.T, get migrationTestGetter, bucket, object, versionID string) string { + t.Helper() + gr, err := get(t.Context(), bucket, object, nil, nil, ObjectOptions{VersionID: versionID}) + if err != nil { + return fmt.Sprintf("<%v>", err) + } + defer gr.Close() + data, err := io.ReadAll(gr) + if err != nil { + t.Fatal(err) + } + return string(data) +} + +// migrationTestWalk runs one bucket of a decommission or rebalance of source, +// including the walker's accounting and source cleanup. It returns the number +// of versions decommission counted as failed. +func migrationTestWalk(t *testing.T, z *erasureServerPools, bucket string, source int, kind string) int64 { + t.Helper() + migrationTestMover(t, z, source, kind) + if kind == "rebalance" { + if err := z.rebalanceBucket(t.Context(), bucket, source); err != nil { + t.Fatal(err) + } + return 0 + } + z.poolMetaMutex.Lock() + z.poolMeta.Pools[source].LastUpdate = UTCNow() // keep pool.bin and peers out of the walk + z.poolMetaMutex.Unlock() + if err := z.decommissionPool(t.Context(), source, z.serverPools[source], decomBucketInfo{Name: bucket}); err != nil { + t.Fatal(err) + } + z.poolMetaMutex.RLock() + defer z.poolMetaMutex.RUnlock() + return z.poolMeta.Pools[source].Decommission.ItemsDecommissionFailed +} + +// migrationTestTiered stores a null version whose data is in a remote tier, in +// the form decommission moves it. +func migrationTestTiered(t *testing.T, z *erasureServerPools, bucket, object string, pool int, mtime time.Time) ObjectInfo { + t.Helper() + oi := putConsistencyObject(t, z, bucket, object, pool, "old", ObjectOptions{MTime: mtime}) + infos, errs := readAllFileInfo(t.Context(), z.serverPools[pool].getHashedSet(oi.Name).getDisks(), "", bucket, oi.Name, nullVersionID, false, false) + if errs[0] != nil { + t.Fatal(errs[0]) + } + fi := infos[0] + fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", "remote" + if err := z.serverPools[pool].DecomTieredObject(t.Context(), bucket, oi.Name, fi, ObjectOptions{}); err != nil { + t.Fatal(err) + } + return oi +} + +// A cross-pool overwrite can leave copies of one version in several pools. +// Reads return the newest copy, and moving an older one must not replace it. +// An equal ModTime marks the same write, as an interrupted move leaves it. +func TestPoolsMigrationKeepsCurrentCopy(t *testing.T) { + z, bucket := migrationTestPools(t) + base := time.Date(2026, 9, 16, 8, 0, 0, 0, time.UTC) + for _, kind := range []string{"rebalance", "decommission"} { + for _, method := range []string{"put", "multipart"} { + for _, versioned := range []bool{false, true} { + for source := range 2 { + for _, offset := range []time.Duration{-time.Minute, 0, time.Minute} { + t.Run(fmt.Sprintf("%s/%s/versioned=%t/source=%d/offset=%s", kind, method, versioned, source, offset), func(t *testing.T) { + move := migrationTestMover(t, z, source, kind) + opts := ObjectOptions{Versioned: versioned, MTime: base.Add(offset)} + if versioned { + opts.VersionID = mustGetUUID() + } + oi := migrationTestObject(t, z, bucket, t.Name(), source, method == "multipart", []string{"source"}, opts) + opts.MTime = base + want, other := "source", "other" + if offset == 0 { + other = "source" + } else if offset < 0 { + want = "other" + } + putConsistencyObject(t, z, bucket, t.Name(), 1-source, other, opts) + version := migrationTestVersion(oi) + if got := migrationTestBody(t, z.GetObjectNInfo, bucket, t.Name(), version); got != want { + t.Fatalf("fixture reads %q, want %q", got, want) + } + err := move(t.Context(), source, bucket, migrationTestReader(t, z, source, oi)) + if want == "other" && !isDataMovementSuperseded(err) || want == "source" && err != nil { + t.Fatalf("move: %v", err) + } + if err != nil && strings.Contains(err.Error(), "CompleteMultipartUpload") { + t.Error("copied the parts of a superseded version") + } + if got := migrationTestBody(t, z.serverPools[1-source].GetObjectNInfo, bucket, t.Name(), version); got != want { + t.Errorf("destination holds %q, want %q", got, want) + } + if got := migrationTestBody(t, z.GetObjectNInfo, bucket, t.Name(), version); got != want { + t.Errorf("move changed the read result to %q, want %q", got, want) + } + }) + } + } + } + } + } +} + +// A client overwrite, or a replica of the same version, committed while a move +// copies the old version is newer than the moved copy. The move must yield to +// it, also at multipart completion. +func TestPoolsMigrationYieldsToNewerWrite(t *testing.T) { + z, bucket := migrationTestPools(t) + for _, kind := range []string{"rebalance", "decommission"} { + for _, method := range []string{"put", "multipart"} { + for _, versioned := range []bool{false, true} { + for source := range 2 { + t.Run(fmt.Sprintf("%s/%s/versioned=%t/source=%d", kind, method, versioned, source), func(t *testing.T) { + migrationTestNewerWrite(t, z, bucket, kind, method, versioned, source) + }) + } + } + } + } +} + +func migrationTestNewerWrite(t *testing.T, z *erasureServerPools, bucket, kind, method string, versioned bool, source int) { + move := migrationTestMover(t, z, source, kind) + oi := migrationTestObject(t, z, bucket, t.Name(), source, method == "multipart", []string{"moved"}, ObjectOptions{Versioned: versioned, MTime: UTCNow().Add(-time.Hour)}) + gr := migrationTestReader(t, z, source, oi) + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + gate := newMigrationTestGate() + done, finished := make(chan error, 1), make(chan struct{}) + if method == "multipart" { + // Hold the move after its initial check, while it copies parts. + gr.Reader = migrationTestGateReader{Reader: gr.Reader, ctx: ctx, gate: gate} + go func() { defer close(finished); done <- move(ctx, source, bucket, gr) }() + defer func() { gate.unblock(); cancel(); <-finished }() + select { + case <-gate.entered: + case err := <-done: + t.Fatalf("migration finished before the part barrier: %v", err) + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + } + // An ordinary PUT copies under the object lock, so this write + // lands between the mover's source read and its write. + if versioned { + putConsistencyObject(t, z, bucket, t.Name(), 1-source, "newer", ObjectOptions{Versioned: true, VersionID: oi.VersionID, MTime: UTCNow()}) + } else if _, err := z.PutObject(ctx, bucket, t.Name(), mustGetPutObjReader(t, strings.NewReader("newer"), 5, "", ""), ObjectOptions{}); err != nil { + t.Fatal(err) + } + var err error + if method == "multipart" { + gate.unblock() + err = <-done + } else { + err = move(ctx, source, bucket, gr) + } + if !isDataMovementSuperseded(err) { + t.Fatalf("move over a newer write: %v", err) + } + if got := migrationTestBody(t, z.GetObjectNInfo, bucket, t.Name(), migrationTestVersion(oi)); got != "newer" { + t.Fatalf("newer write replaced by %q", got) + } +} + +// With two candidate destinations a superseded copy must neither replace the +// newer copy nor become a new duplicate in the third pool. +func TestPoolsMigrationThreePools(t *testing.T) { + z, bucket := migrationTestThreePools(t) + stale, recent := UTCNow().Add(-time.Hour), UTCNow().Add(-time.Minute) + for _, kind := range []string{"rebalance", "decommission"} { + for _, newest := range []int{0, 1} { + t.Run(fmt.Sprintf("%s/newest=%d", kind, newest), func(t *testing.T) { + move := migrationTestMover(t, z, 0, kind) + mtime := func(pool int) time.Time { + if pool == newest { + return recent + } + return stale + } + oi := migrationTestObject(t, z, bucket, t.Name(), 0, false, []string{"pool-0"}, ObjectOptions{MTime: mtime(0)}) + putConsistencyObject(t, z, bucket, t.Name(), 1, "pool-1", ObjectOptions{MTime: mtime(1)}) + if newest == 0 { + putConsistencyObject(t, z, bucket, t.Name(), 2, "pool-2", ObjectOptions{MTime: stale}) + } + want := fmt.Sprintf("pool-%d", newest) + err := move(t.Context(), 0, bucket, migrationTestReader(t, z, 0, oi)) + if newest == 1 && !isDataMovementSuperseded(err) || newest == 0 && err != nil { + t.Fatalf("move: %v", err) + } + if got := migrationTestBody(t, z.GetObjectNInfo, bucket, t.Name(), ""); got != want { + t.Errorf("reads return %q, want %q", got, want) + } + if newest == 1 { + if _, err := z.serverPools[2].GetObjectInfo(t.Context(), bucket, t.Name(), ObjectOptions{}); !isErrObjectNotFound(err) { + t.Errorf("superseded copy duplicated into pool 2: %v", err) + } + } + }) + } + } +} + +// The walkers count a superseded version as moved, so its stale source copy is +// removed with the object and the source pool drains. +func TestPoolsMigrationRetiresSupersededCopy(t *testing.T) { + for _, kind := range []string{"rebalance", "decommission"} { + t.Run(kind, func(t *testing.T) { + z, bucket := migrationTestPools(t) + ctx := t.Context() + stale, recent := UTCNow().Add(-time.Hour), UTCNow().Add(-time.Minute) + put := func(object string, pool int, body string, multipart bool, opts ObjectOptions) ObjectInfo { + return migrationTestObject(t, z, bucket, object, pool, multipart, []string{body}, opts) + } + put("stale", 0, "old", false, ObjectOptions{MTime: stale}) + put("stale", 1, "new", false, ObjectOptions{MTime: recent}) + put("stale-multipart", 0, "old", true, ObjectOptions{MTime: stale}) + put("stale-multipart", 1, "new", false, ObjectOptions{MTime: recent}) + put("newer", 0, "new", false, ObjectOptions{MTime: recent}) + put("newer", 1, "old", false, ObjectOptions{MTime: stale}) + oldest := put("history", 0, "old", false, ObjectOptions{Versioned: true, MTime: stale}) + put("history", 1, "new", false, ObjectOptions{Versioned: true, MTime: recent}) + // A suspended bucket's older null delete marker must not hide a newer + // null version. + hidden := put("stale-marker", 0, "old", false, ObjectOptions{Versioned: true, MTime: stale.Add(-time.Minute)}) + if _, err := z.serverPools[0].DeleteObject(ctx, bucket, "stale-marker", ObjectOptions{VersionSuspended: true, MTime: stale}); err != nil { + t.Fatal(err) + } + put("stale-marker", 1, "new", false, ObjectOptions{VersionSuspended: true, MTime: recent}) + objects := []string{"stale", "stale-multipart", "newer", "history", "stale-marker"} + if kind == "decommission" { + // Rebalance leaves tiered versions in place; decommission moves + // their metadata. + migrationTestTiered(t, z, bucket, "stale-tiered", 0, stale) + put("stale-tiered", 1, "new", false, ObjectOptions{MTime: recent}) + objects = append(objects, "stale-tiered") + } + want := map[string]string{} + for _, object := range objects { + want[object] = migrationTestBody(t, z.GetObjectNInfo, bucket, object, "") + } + if want["stale"] != "new" || want["stale-marker"] != "new" { + t.Fatalf("fixture reads %v", want) + } + if failed := migrationTestWalk(t, z, bucket, 0, kind); failed != 0 { + t.Fatalf("%d versions failed to move", failed) + } + for _, object := range objects { + if got := migrationTestBody(t, z.GetObjectNInfo, bucket, object, ""); got != want[object] { + t.Errorf("%s: reads %q after the move, want %q", object, got, want[object]) + } + } + var retained []string + for _, set := range z.serverPools[0].sets { + if err := set.listObjectsToDecommission(ctx, decomBucketInfo{Name: bucket}, func(entry metaCacheEntry) { + retained = append(retained, entry.name) + }); err != nil { + t.Fatal(err) + } + } + if len(retained) > 0 { + t.Errorf("source pool retained %v", retained) + } + for _, oi := range []ObjectInfo{oldest, hidden} { + if got := migrationTestBody(t, z.GetObjectNInfo, bucket, oi.Name, oi.VersionID); got != "old" { + t.Errorf("%s: lost version %s: %q", oi.Name, oi.VersionID, got) + } + } + }) + } +} + +// A tiered move that fails, here because another pool cannot be read, must keep +// its source like any other failed version. +func TestPoolsMigrationTieredFailureKeepsSource(t *testing.T) { + z, bucket := migrationTestPools(t) + tiered := migrationTestTiered(t, z, bucket, t.Name(), 0, UTCNow().Add(-time.Hour)) + set := z.serverPools[1].getHashedSet(tiered.Name) + original := set.getDisks + disks := append([]StorageAPI(nil), original()...) + for i := range disks { + disks[i] = consistencyReadFaultDisk{StorageAPI: disks[i], bucket: bucket, object: tiered.Name} + } + set.getDisks = func() []StorageAPI { return disks } + defer func() { set.getDisks = original }() + if failed := migrationTestWalk(t, z, bucket, 0, "decommission"); failed != 1 { + t.Fatalf("%d failed versions, want 1", failed) + } + if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, tiered.Name, ObjectOptions{VersionID: nullVersionID}); err != nil { + t.Fatalf("failed move removed its source: %v", err) + } +} diff --git a/cmd/erasure-server-pool-put-conditional_test.go b/cmd/erasure-server-pool-put-conditional_test.go index 9f7317c4d..cd4897921 100644 --- a/cmd/erasure-server-pool-put-conditional_test.go +++ b/cmd/erasure-server-pool-put-conditional_test.go @@ -355,7 +355,7 @@ func TestPoolsConditionalPutVersionSelection(t *testing.T) { opts.ReplicaLockReconcile, opts.ReplicationRequest = true, true want = addressed case "movement": - opts.DataMovement, opts.SrcPoolIdx = true, 1 + opts.DataMovement, opts.SrcPoolIdx, opts.MTime = true, 1, addressed.ModTime want = addressed case "tie": want = putConsistencyObject(t, z, bucket, object, 0, "tie-winner", ObjectOptions{Versioned: true, MTime: current.ModTime}) diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go index b60c9c904..f3dc6967a 100644 --- a/cmd/erasure-server-pool-rebalance.go +++ b/cmd/erasure-server-pool-rebalance.go @@ -671,6 +671,9 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, SkipRebalancing: true, // make sure we skip the decommissioned pool NoAuditLog: true, }) + if isDataMovementSuperseded(err) { + err = nil + } // This can happen when rebalance stop races with ongoing rebalance workers. // These rebalance failures can be ignored. if err != nil { @@ -719,7 +722,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, continue } - if err = z.rebalanceObject(ctx, poolIdx, bucket, gr); err != nil { + if err = z.rebalanceObject(ctx, poolIdx, bucket, gr); err != nil && !isDataMovementSuperseded(err) { // This can happen when rebalance stop races with ongoing rebalance workers. // These rebalance failures can be ignored. if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) { @@ -863,6 +866,12 @@ func (z *erasureServerPools) rebalanceObject(ctx context.Context, poolIdx int, b } if oi.isMultipart() { + // Completion repeats this check under the object write lock. + if _, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(oi.Name), ObjectOptions{ + VersionID: oi.VersionID, MTime: oi.ModTime, SrcPoolIdx: poolIdx, + }); err != nil { + return err + } res, err := z.NewMultipartUpload(ctx, bucket, oi.Name, ObjectOptions{ VersionID: oi.VersionID, UserDefined: migrationObjectMetadata(oi), @@ -900,6 +909,8 @@ func (z *erasureServerPools) rebalanceObject(ctx context.Context, poolIdx int, b } } _, err = z.CompleteMultipartUpload(ctx, bucket, oi.Name, res.UploadID, parts, ObjectOptions{ + VersionID: oi.VersionID, + SrcPoolIdx: poolIdx, DataMovement: true, MTime: oi.ModTime, NoAuditLog: true, @@ -1010,7 +1021,10 @@ const ( rebalanceMetricSaveMetadata ) -var errDataMovementSrcDstPoolSame = errors.New("source and destination pool are the same") +var ( + errDataMovementSrcDstPoolSame = errors.New("source and destination pool are the same") + errDataMovementSuperseded = errors.New("another pool holds a newer copy") +) func rebalanceTrace(r rebalanceMetric, poolIdx int, startTime time.Time, duration time.Duration, err error, path string, sz int64) madmin.TraceInfo { var errStr string diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 2075817e9..461b52c96 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1183,6 +1183,14 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec opts.CheckPrecondFn = nil } + if opts.DataMovement { + resolve, err := z.checkDataMovementCopies(ctx, bucket, object, opts) + if err != nil { + return ObjectInfo{}, err + } + opts.replicaObjectInfo = resolve + } + idx, err := z.getWritePoolIdx(ctx, bucket, object, data.Size(), true) if err != nil { return ObjectInfo{}, err @@ -1202,7 +1210,7 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec } // Public overwrites need all pools for Object Lock. Unaddressed replica // callbacks retain their set-local reads unless reconciliation is needed. - if opts.ReplicaLockReconcile || !opts.ReplicationRequest { + if opts.replicaObjectInfo == nil && (opts.ReplicaLockReconcile || !opts.ReplicationRequest) { opts.replicaObjectInfo = z.replicaObjectInfo } oi, err := z.serverPools[idx].PutObject(ctx, bucket, object, data, opts) @@ -1339,6 +1347,9 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob } if opts.DataMovement { + if _, err := z.checkDataMovementCopies(ctx, bucket, object, gopts); err != nil { + return pinfo.ObjInfo, err + } objInfo, err = z.serverPools[pinfo.Index].DeleteObject(ctx, bucket, object, opts) objInfo.Name = decodeDirObject(object) return objInfo, err @@ -2248,6 +2259,13 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket } opts.CheckPrecondFn = nil } + if opts.DataMovement { + resolve, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(object), opts) + if err != nil { + return ObjectInfo{}, err + } + opts.replicaObjectInfo = resolve + } } // Hold write locks to verify uploaded parts, also disallows any @@ -2267,7 +2285,7 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket if opts.ReplicaLockReconcile || opts.DataMovement { opts.ReplicaLockReconcile = true } - if opts.ReplicaLockReconcile || !opts.ReplicationRequest { + if opts.replicaObjectInfo == nil && (opts.ReplicaLockReconcile || !opts.ReplicationRequest) { opts.replicaObjectInfo = z.replicaObjectInfo } @@ -3258,6 +3276,11 @@ func (z *erasureServerPools) DecomTieredObject(ctx context.Context, bucket, obje defer ns.Unlock(lkctx) opts.NoLock = true } + if opts.DataMovement { + if _, err := z.checkDataMovementCopies(ctx, bucket, object, opts); err != nil { + return err + } + } idx, err := z.getPoolIdxNoLock(ctx, bucket, object, fi.Size) if err != nil { return err diff --git a/cmd/object-api-errors.go b/cmd/object-api-errors.go index 457970fb0..7b6ccfb01 100644 --- a/cmd/object-api-errors.go +++ b/cmd/object-api-errors.go @@ -834,6 +834,9 @@ func (de DataMovementOverwriteErr) Error() string { if de.VersionID != "" { objInfoStr = fmt.Sprintf("%s version-id=%s", objInfoStr, de.VersionID) } + if de.Err == errDataMovementSuperseded { + return fmt.Sprintf("data movement skipped, another pool holds a newer copy of %s", objInfoStr) + } return fmt.Sprintf("invalid data movement operation, source and destination pool are the same for %s", objInfoStr) } @@ -841,3 +844,11 @@ func isDataMovementOverWriteErr(err error) bool { var de DataMovementOverwriteErr return errors.As(err, &de) } + +// isDataMovementSuperseded reports a version that was not moved because +// another pool already holds a newer copy, which reads return. The source copy +// is a stale duplicate: the move is complete, and the source may be removed. +func isDataMovementSuperseded(err error) bool { + var de DataMovementOverwriteErr + return errors.As(err, &de) && de.Err == errDataMovementSuperseded +}