fix(pools): preserve delete markers and recheck migration cleanup

Move solitary and null delete markers instead of dropping them, including
idempotent UUID marker retries. Check that a reader's source version still
exists before moving it, and preserve the established read winner for
conflicting timestamp ties.

Retire only the versions that were successfully moved while holding the
pooled object lock. Recheck the source and retained copies before cleanup,
so concurrent deletes and uncertain destination reads cannot revive data.
Count remaining markers when verifying decommission completion.

Add two/three-pool walker, history, purge, concurrent-delete, read-fault,
and timestamp-tie regressions.

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/erasure-server-pool-consistency.go

@@ -65,10 +65,11 @@ func (z *erasureServerPools) objectPoolInfos(ctx context.Context, bucket, object
 // 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.
+// Equal timestamps normally identify an interrupted move. Conflicting ties
+// retain the same pool-order winner as reads. 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.
@@ -91,11 +92,29 @@ func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket
 	if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
 		return nil, err
 	}
+	// The snapshot may have been purged or replaced after the reader opened.
+	// In that case the walker must leave the current source state alone.
+	sourcePresent := false
+	for _, copy := range copies {
+		if copy.Index == opts.SrcPoolIdx && copy.ObjInfo.ModTime.Equal(opts.MTime) && copy.ObjInfo.DeleteMarker == opts.DeleteMarker {
+			sourcePresent = true
+		}
+	}
+	if !sourcePresent {
+		return nil, VersionNotFound{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
+	}
 	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}
 		}
 	}
+	if winner := copies[0]; winner.Index != opts.SrcPoolIdx && winner.ObjInfo.ModTime.Equal(opts.MTime) {
+		for _, copy := range copies {
+			if copy.Index == opts.SrcPoolIdx && (copy.ObjInfo.ETag != winner.ObjInfo.ETag || copy.ObjInfo.DeleteMarker != winner.ObjInfo.DeleteMarker || copy.ObjInfo.Size != winner.ObjInfo.Size) {
+				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
@@ -107,6 +126,61 @@ func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket
 	}, nil
 }
 
+// removeMovedVersions retires only the source versions in the walker's
+// snapshot. A concurrent delete can create a new marker in the source pool;
+// deleting the whole prefix would remove that marker and expose moved data.
+func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo) error {
+	object = encodeDirObject(object)
+	lk := z.NewNSLock(bucket, object)
+	lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+	if err != nil {
+		return err
+	}
+	ctx = lkctx.Context()
+	defer lk.Unlock(lkctx)
+	for _, version := range versions {
+		vid := version.VersionID
+		if vid == "" {
+			vid = nullVersionID
+		}
+		copies, err := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{VersionID: vid, NoAuditLog: true})
+		if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
+			continue
+		}
+		if err != nil {
+			return err
+		}
+		var src *ObjectInfo
+		var retained []PoolObjInfo
+		for _, copy := range copies {
+			if copy.Index == source {
+				src = &copy.ObjInfo
+			} else {
+				retained = append(retained, copy)
+			}
+		}
+		if src == nil {
+			continue
+		}
+		if !src.ModTime.Equal(version.ModTime) || src.DeleteMarker != version.Deleted {
+			return PreConditionFailed{}
+		}
+		// Successful movement is not enough if the destination changed again.
+		// Keep the source unless another copy still preserves its read result.
+		if len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime) {
+			return PreConditionFailed{}
+		}
+		_, err = z.serverPools[source].DeleteObject(ctx, bucket, object, ObjectOptions{
+			VersionID: vid, NoLock: true, NoAuditLog: true,
+			SkipFreeVersion: true, // preserve the old prefix cleanup's tier-GC behavior
+		})
+		if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			return err
+		}
+	}
+	return 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.