diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index e5d12967f..0e451478f 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -2195,6 +2195,13 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string } } + if opts.DataMovement && opts.DeleteMarker { + // Moving a null marker replaces any older null data/marker. Moving + // a UUID marker preserves it on retries instead of purging it. + deleteMarker = true + markDelete = opts.VersionID != nullVersionID + } + purge := opts.isVersionPurge() if purge { // A delete marker describes the version being removed; it is not an diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go index 2272a2afb..bc9985934 100644 --- a/cmd/erasure-server-pool-consistency.go +++ b/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 = ©.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. diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go index dee2fdfb4..3c21825bb 100644 --- a/cmd/erasure-server-pool-decom.go +++ b/cmd/erasure-server-pool-decom.go @@ -832,27 +832,17 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool // to create the appropriate stack. versionsSorter(fivs.Versions).reverse() - var decommissioned, expired int + var decommissioned int for _, version := range fivs.Versions { stopFn := globalDecommissionMetrics.log(decomMetricDecommissionObject, idx, bi.Name, version.Name, version.VersionID) // Apply lifecycle rules on the objects that are expired. if filterLifecycle(bi.Name, version.Name, version) { - expired++ decommissioned++ stopFn(version.Size, errors.New("ILM expired object/version will be skipped")) continue } - // any object with only single DEL marker we don't need - // to decommission, just skip it, this also includes - // any other versions that have already expired. - remainingVersions := len(fivs.Versions) - expired - if version.Deleted && remainingVersions == 1 && rcfg == nil { - decommissioned++ - stopFn(version.Size, errors.New("DELETE marked object with no other non-current versions will be skipped")) - continue - } - + // Preserve every marker: older versions may live in another pool. versionID := version.VersionID if versionID == "" { versionID = nullVersionID @@ -991,15 +981,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool // if all versions were decommissioned, then we can delete the object versions. if decommissioned == len(fivs.Versions) { stopFn := globalDecommissionMetrics.log(decomMetricDecommissionRemoveObject, idx, bi.Name, entry.name) - _, err := set.DeleteObject(ctx, - bi.Name, - encodeDirObject(entry.name), - ObjectOptions{ - DeletePrefix: true, // use prefix delete to delete all versions at once. - DeletePrefixObject: true, // use prefix delete on exact object (this is an optimization to avoid fan-out calls) - NoAuditLog: true, - }, - ) + err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions) stopFn(0, err) auditLogDecom(ctx, "DecomDeleteObject", bi.Name, entry.name, "", err) if err != nil { @@ -1208,10 +1190,6 @@ func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error ignored++ continue } - if version.Deleted { - ignored++ - continue - } } versionsFound += len(fivs.Versions) - ignored diff --git a/cmd/erasure-server-pool-migration-delete_test.go b/cmd/erasure-server-pool-migration-delete_test.go new file mode 100644 index 000000000..904c87c0f --- /dev/null +++ b/cmd/erasure-server-pool-migration-delete_test.go @@ -0,0 +1,169 @@ +// Copyright (c) 2026 Feng Ruohang +// +// This file is part of Silo Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package cmd + +import ( + "fmt" + "testing" + "time" +) + +// Both walkers must preserve markers even when all older versions reside in +// another pool. Local version counts cannot prove that a marker is redundant. +func TestPoolsMigrationDeleteMarkers(t *testing.T) { + for _, kind := range []string{"rebalance", "decommission"} { + for _, pools := range []int{2, 3} { + for _, null := range []bool{false, true} { + for _, history := range []string{"none", "source", "destination"} { + t.Run(fmt.Sprintf("%s/pools=%d/null=%t/history=%s", kind, pools, null, history), func(t *testing.T) { + var z *erasureServerPools + var bucket string + if pools == 2 { + z, bucket = migrationTestPools(t) + } else { + z, bucket = migrationTestThreePools(t) + } + const object = "deleted" + old, stamp := UTCNow().Add(-time.Hour), UTCNow().Add(-time.Minute) + var prior ObjectInfo + if history != "none" { + pool := 0 + if history == "destination" { + pool = 1 + } + prior = putConsistencyObject(t, z, bucket, object, pool, "history", ObjectOptions{Versioned: true, MTime: old}) + } + // A stale null copy is a normal consequence of cross-pool overwrites. + putConsistencyObject(t, z, bucket, object, 1, "stale", ObjectOptions{MTime: old}) + vid := mustGetUUID() + if null { + vid = nullVersionID + } + opts := ObjectOptions{Versioned: true, VersionID: vid, DeleteMarker: true, MTime: stamp} + marker, err := z.serverPools[0].DeleteObject(t.Context(), bucket, object, opts) + mustIAM(t, err) + if !marker.DeleteMarker { + t.Fatal("fixture did not create marker") + } + if oi, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); !isErrObjectNotFound(err) || !oi.DeleteMarker { + t.Fatalf("fixture visible: %+v %v", oi, err) + } + if failed := migrationTestWalk(t, z, bucket, 0, kind); failed != 0 { + t.Fatalf("failed moves: %d", failed) + } + for attempt := 0; attempt < 2; attempt++ { + if oi, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); !isErrObjectNotFound(err) || !oi.DeleteMarker || !oi.ModTime.Equal(stamp) { + t.Fatalf("marker lost: %+v %v", oi, err) + } + if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: migrationTestVersion(marker)}); !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { + t.Fatalf("source marker retained: %v", err) + } + if prior.VersionID != "" && migrationTestBody(t, z.GetObjectNInfo, bucket, object, prior.VersionID) != "history" { + t.Fatal("lost historical version") + } + // Repeating a completed walk must preserve the same state. + migrationTestWalk(t, z, bucket, 0, kind) + } + }) + } + } + } + } +} + +// A reader opened before a public purge must not recreate the purged version. +func TestPoolsMigrationDoesNotResurrectPurgedSource(t *testing.T) { + for _, kind := range []string{"rebalance", "decommission"} { + for _, multipart := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/multipart=%t", kind, multipart), func(t *testing.T) { + z, bucket := migrationTestPools(t) + move := migrationTestMover(t, z, 0, kind) + oi := migrationTestObject(t, z, bucket, "purged", 0, multipart, []string{"payload"}, ObjectOptions{Versioned: true}) + gr := migrationTestReader(t, z, 0, oi) + _, err := z.DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: oi.VersionID, Versioned: true}) + mustIAM(t, err) + err = move(t.Context(), 0, bucket, gr) + if err == nil { + t.Error("move accepted a purged source") + } + if oi, err := z.GetObjectInfo(t.Context(), bucket, oi.Name, ObjectOptions{}); !isErrObjectNotFound(err) { + t.Fatalf("purged version resurrected: %+v %v", oi, err) + } + }) + } + } +} + +// Cleanup must not erase a marker created after the data was copied. It must +// also fail closed if the destination becomes unreadable before source removal. +func TestPoolsMigrationCleanupRechecksSource(t *testing.T) { + for _, change := range []string{"delete-null", "delete-uuid", "unreadable-destination", "missing-destination"} { + t.Run(change, func(t *testing.T) { + z, bucket := migrationTestPools(t) + move := migrationTestMover(t, z, 0, "decommission") + oi := migrationTestObject(t, z, bucket, "cleanup", 0, false, []string{"payload"}, ObjectOptions{}) + snapshot := []FileInfo{{VersionID: oi.VersionID, ModTime: oi.ModTime}} + mustIAM(t, move(t.Context(), 0, bucket, migrationTestReader(t, z, 0, oi))) + switch change { + case "delete-null", "delete-uuid": + _, err := z.DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionSuspended: change == "delete-null", Versioned: change == "delete-uuid"}) + mustIAM(t, err) + case "unreadable-destination": + restore := conditionalPutSwapDisks(z.serverPools[1], oi.Name, func(d StorageAPI) StorageAPI { + return consistencyReadFaultDisk{StorageAPI: d, bucket: bucket, object: oi.Name} + }) + defer restore() + case "missing-destination": + _, err := z.serverPools[1].DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: nullVersionID}) + mustIAM(t, err) + } + err := z.removeMovedVersions(t.Context(), bucket, oi.Name, 0, snapshot) + if change != "delete-uuid" && err == nil { + t.Fatal("cleanup accepted changed or uncertain state") + } + if change == "delete-null" || change == "delete-uuid" { + if got, err := z.GetObjectInfo(t.Context(), bucket, oi.Name, ObjectOptions{}); !isErrObjectNotFound(err) || !got.DeleteMarker { + t.Fatalf("cleanup resurrected data: %v", err) + } + } else if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, oi.Name, ObjectOptions{}); err != nil { + t.Fatalf("cleanup lost source: %v", err) + } + }) + } +} + +func TestPoolsMigrationConflictingTimestampTie(t *testing.T) { + for _, kind := range []string{"rebalance", "decommission"} { + for source := range 2 { + t.Run(fmt.Sprintf("%s/source=%d", kind, source), func(t *testing.T) { + z, bucket := migrationTestPools(t) + stamp := UTCNow().Add(-time.Minute) + for i := range 2 { + putConsistencyObject(t, z, bucket, "tie", i, fmt.Sprintf("pool-%d", i), ObjectOptions{MTime: stamp}) + } + if got := migrationTestBody(t, z.GetObjectNInfo, bucket, "tie", ""); got != "pool-0" { + t.Fatalf("fixture winner: %s", got) + } + migrationTestWalk(t, z, bucket, source, kind) + if got := migrationTestBody(t, z.GetObjectNInfo, bucket, "tie", ""); got != "pool-0" { + t.Fatalf("changed tie winner: %s", got) + } + }) + } + } +} diff --git a/cmd/erasure-server-pool-migration_test.go b/cmd/erasure-server-pool-migration_test.go index 7f934ad3b..cf3c7834c 100644 --- a/cmd/erasure-server-pool-migration_test.go +++ b/cmd/erasure-server-pool-migration_test.go @@ -116,15 +116,18 @@ func migrationTestPools(t *testing.T) (*erasureServerPools, string) { t.Helper() z, bucket := consistencyPools(t) for _, pool := range z.serverPools { - for _, set := range pool.sets { - original := set.getDisks - disks := append([]StorageAPI(nil), original()...) - for i := range disks { - disks[i] = tagTestCapacityDisk{StorageAPI: disks[i]} + pool.erasureDisksMu.Lock() + previous := pool.erasureDisks + wrapped := make([][]StorageAPI, len(previous)) + for i, disks := range previous { + wrapped[i] = append([]StorageAPI(nil), disks...) + for j := range wrapped[i] { + wrapped[i][j] = tagTestCapacityDisk{StorageAPI: wrapped[i][j]} } - set.getDisks = func() []StorageAPI { return disks } - t.Cleanup(func() { set.getDisks = original }) } + pool.erasureDisks = wrapped + pool.erasureDisksMu.Unlock() + t.Cleanup(func() { pool.erasureDisksMu.Lock(); pool.erasureDisks = previous; pool.erasureDisksMu.Unlock() }) } return z, bucket } @@ -705,6 +708,7 @@ func TestPoolsMigrationKeepsCurrentCopy(t *testing.T) { want, other := "source", "other" if offset == 0 { other = "source" + opts.PreserveETag = oi.ETag // interrupted copies preserve the original multipart ETag } else if offset < 0 { want = "other" } diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go index f3dc6967a..6c64f9f28 100644 --- a/cmd/erasure-server-pool-rebalance.go +++ b/cmd/erasure-server-pool-rebalance.go @@ -623,7 +623,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, // to create the appropriate stack. versionsSorter(fivs.Versions).reverse() - var rebalanced, expired int + var rebalanced int for _, version := range fivs.Versions { stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name, version.VersionID) @@ -635,21 +635,11 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, // Apply lifecycle rules on the objects that are expired. if filterLifecycle(bucket, version.Name, version) { - expired++ stopFn(version.Size, errors.New("ILM expired object/version will be skipped")) continue } - // any object with only single DEL marker we don't need - // to rebalance, just skip it, this also includes - // any other versions that have already expired. - remainingVersions := len(fivs.Versions) - expired - if version.Deleted && remainingVersions == 1 { - rebalanced++ - stopFn(version.Size, errors.New("DELETE marked object with no other non-current versions will be skipped")) - continue - } - + // Preserve every marker: older versions may live in another pool. versionID := version.VersionID if versionID == "" { versionID = nullVersionID @@ -753,15 +743,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string, // if all versions were rebalanced, we can delete the object versions. if rebalanced == len(fivs.Versions) { stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceRemoveObject, poolIdx, bucket, entry.name) - _, err := set.DeleteObject(ctx, - bucket, - encodeDirObject(entry.name), - ObjectOptions{ - DeletePrefix: true, // use prefix delete to delete all versions at once. - DeletePrefixObject: true, // use prefix delete on exact object (this is an optimization to avoid fan-out calls) - NoAuditLog: true, - }, - ) + err := z.removeMovedVersions(ctx, bucket, entry.name, poolIdx, fivs.Versions) stopFn(0, err) auditLogRebalance(ctx, "Rebalance:DeleteObject", bucket, entry.name, "", err) if err != nil { diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 461b52c96..79b70d51d 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1251,6 +1251,25 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob return ObjectInfo{}, z.deletePrefix(ctx, bucket, object) } + if opts.DataMovement { + opts.NoLock = true + if _, err := z.checkDataMovementCopies(ctx, bucket, object, opts); err != nil { + return objInfo, err + } + // A marker still needs a destination when it is the only version of + // this key outside the draining pool. + idx, err := z.getPoolIdxNoLock(ctx, bucket, object, 0) + if err != nil { + return objInfo, err + } + if idx == opts.SrcPoolIdx { + return objInfo, DataMovementOverwriteErr{Bucket: bucket, Object: object, VersionID: opts.VersionID, Err: errDataMovementSrcDstPoolSame} + } + objInfo, err = z.serverPools[idx].DeleteObject(ctx, bucket, object, opts) + objInfo.Name = decodeDirObject(object) + return objInfo, err + } + // Resolve a physical purge by its addressed version even on a replica // receiver: latest-key routing can select a different pool or miss copies. // Marker creation and specialized movement/scanner operations retain their @@ -1336,25 +1355,6 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob return pinfo.ObjInfo, nil } - // Datamovement must never be allowed on the same pool. - if opts.DataMovement && opts.SrcPoolIdx == pinfo.Index { - return pinfo.ObjInfo, DataMovementOverwriteErr{ - Bucket: bucket, - Object: object, - VersionID: opts.VersionID, - Err: errDataMovementSrcDstPoolSame, - } - } - - 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 - } - // Delete concurrently in all server pools with read quorum error for unversioned objects. if len(noReadQuorumPools) > 0 && !opts.Versioned && !opts.VersionSuspended { return z.deleteObjectFromAllPools(ctx, bucket, object, opts, noReadQuorumPools)