fix(pools): do not let data movement overwrite a newer copy

Apply PR #248 from bca40a4ac9.

Signed-off-by: Denis Babchuk <babchukdenis@icloud.com>
Signed-off-by: Feng Ruohang <rh@vonng.com>

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
 }

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)

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)
+	}
+}

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})

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

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

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
+}