From 1bd0b1b04d252d75f02c957fdc5c752da79b62e1 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Fri, 6 Nov 2026 18:53:27 +0800 Subject: [PATCH] fix(ilm): do not replicate deletes that lifecycle makes The migration follow-up for #265 (PR #276) made lifecycle-created deletes replication-aware: the scanner, the expiry of a transitioned version and the bulk noncurrent purge evaluated the delete replication decision and queued a delete marker or a version purge for the targets. That is the content of the still-open PR #240. It changes behavior for every bucket whose replication rule replicates deletes or delete markers, the mc default: a target with longer retention loses versions on the source's schedule, and a purge keeps its space as a pending tombstone until the target confirms. SILO, like S3, does not replicate lifecycle deletes; an opt-in can come later. The release did not replicate them. Remove exactly the replication parts: the decision callbacks and the post-delete scheduling in applyExpiryOnNonTransitionedObjects and expireTransitionedObject, the scheduling in deleteObjectVersions, and ilmPurgeToDelete; the scanner builds noncurrent entries as before. The rest of #276 stays: the hot-tier delete before the remote one, the free-version retry record, the lifecycle-change fence and the migration completion checks. API deletes still replicate, and the replication heal still handles persisted pending markers and purges. As in the release, the replication heal also resyncs an empty-state marker when existing-object replication applies to it. The delete-all variant of the scheduling, which queued a task with an empty decision, goes with it. Tests: the scanner's noncurrent purge, run with replication configured, deletes the version at once and queues nothing; an ILM marker carries no replication state and the heal leaves it unless existing-object replication applies, in which case it is resynced as before; an API version delete still queues a purge. The cross-pool batch, dispatch and decommission tests now assert no replication, and a decommission whose expiry used to wait for the target completes in one run. Adding back the scanner's decision makes the first two fail. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Feng Ruohang --- cmd/bucket-lifecycle.go | 39 +- cmd/bucket-replication.go | 2 - cmd/data-scanner.go | 72 +--- ...rasure-server-pool-lifecycle-batch_test.go | 71 +--- ...asure-server-pool-lifecycle-safety_test.go | 29 +- cmd/erasure-server-pool-lifecycle_test.go | 4 +- cmd/erasure-server-pool-migration-ilm_test.go | 82 +--- cmd/object-handlers-common.go | 15 - cmd/replication-ilm-delete_test.go | 349 +++++++++--------- 9 files changed, 236 insertions(+), 427 deletions(-) diff --git a/cmd/bucket-lifecycle.go b/cmd/bucket-lifecycle.go index 54511b851..d43d02c9b 100644 --- a/cmd/bucket-lifecycle.go +++ b/cmd/bucket-lifecycle.go @@ -651,23 +651,12 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob // the remote object is still protected. keepRemote := opts.VersionID == "" && (opts.Versioned || (opts.VersionSuspended && oi.VersionID != "" && oi.VersionID != nullVersionID)) - if lcEvent.Action == lifecycle.DeleteAction || lcEvent.Action == lifecycle.DeleteVersionAction { - opts.SetEvalMetadataFn(func(goi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) { - return checkReplicateDelete(ctx, oi.Bucket, ObjectToDelete{ - ObjectV: ObjectV{ - ObjectName: goi.Name, - VersionID: opts.VersionID, - }, - }, *goi, opts, gerr), nil - }) - } // Delete the hot-tier namespace first. For a transitioned version this // leaves a durable free-version record until the remote delete succeeds; // that record is the retry point if the process stops between the two // operations. - delInfo, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts) - if err != nil { + if _, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts); err != nil { return err } @@ -677,32 +666,6 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob } } - // Queue replication of the ILM-created delete marker or version purge, - // mirroring the DELETE handler's post-delete scheduling. - if delInfo.ReplicationStatus == replication.Pending || delInfo.VersionPurgeStatus == replication.VersionPurgePending { - dmVersionID, versionID := "", "" - if delInfo.DeleteMarker && delInfo.VersionPurgeStatus.Empty() { - dmVersionID = delInfo.VersionID - } else { - versionID = delInfo.VersionID - } - if delInfo.Name == "" { - delInfo.Name = oi.Name - } - scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{ - DeletedObject: DeletedObject{ - ObjectName: delInfo.Name, - VersionID: versionID, - DeleteMarkerVersionID: dmVersionID, - DeleteMarkerMTime: DeleteMarkerMTime{delInfo.ModTime}, - DeleteMarker: delInfo.DeleteMarker, - ReplicationState: delInfo.ReplicationState(), - }, - Bucket: oi.Bucket, - EventType: ReplicateILMDelete, - }, objectAPI) - } - // Send audit for the lifecycle delete operation defer auditLogLifecycle(ctx, *oi, ILMExpiry, tags, traceFn) diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go index ee26ef355..492657e91 100644 --- a/cmd/bucket-replication.go +++ b/cmd/bucket-replication.go @@ -2062,8 +2062,6 @@ const ( ReplicateIncoming = "replicate:incoming" // ReplicateIncomingDelete - audit trail of inline replication of deletes. ReplicateIncomingDelete = "replicate:incoming:delete" - // ReplicateILMDelete - audit trail of delete replication triggered by ILM expiry. - ReplicateILMDelete = "replicate:ilm:delete" // ReplicateHeal - audit trail for healing of failed/pending replications ReplicateHeal = "replicate:heal" diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go index ca8fc5dc4..448f521b5 100644 --- a/cmd/data-scanner.go +++ b/cmd/data-scanner.go @@ -1132,7 +1132,13 @@ eventLoop: case lifecycle.DeleteVersionAction: // noncurrent versions expiration remainingVersions-- size = 0 - toDel = append(toDel, ilmPurgeToDelete(ctx, i.bucket, oi, i.replication.Config != nil)) + opts := objOpts[idx] + toDel = append(toDel, ObjectToDelete{ + ObjectV: ObjectV{ + ObjectName: opts.Name, + VersionID: opts.VersionID, + }, + }) noncurrentEvents = append(noncurrentEvents, event) case lifecycle.TransitionAction, lifecycle.TransitionVersionAction: @@ -1241,20 +1247,6 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay } } - // ILM-created delete markers and version purges must carry the same - // replication decision as API-created deletes. DeleteAll is intentionally - // excluded: it is an object-local purge and has no single version to - // replicate. - if lcEvent.Action == lifecycle.DeleteAction || lcEvent.Action == lifecycle.DeleteVersionAction { - opts.SetEvalMetadataFn(func(oi *ObjectInfo, gerr error) (dsc ReplicateDecision, err error) { - return checkReplicateDelete(ctx, obj.Bucket, ObjectToDelete{ - ObjectV: ObjectV{ - ObjectName: oi.Name, - VersionID: opts.VersionID, - }, - }, *oi, opts, gerr), nil - }) - } var ( dobj ObjectInfo err error @@ -1294,29 +1286,6 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay dobj = obj } - // Queue replication of the ILM-created delete marker or version purge, - // mirroring the DELETE handler's post-delete scheduling. - if dobj.ReplicationStatus == replication.Pending || dobj.VersionPurgeStatus == replication.VersionPurgePending { - dmVersionID, versionID := "", "" - if dobj.DeleteMarker && dobj.VersionPurgeStatus.Empty() { - dmVersionID = dobj.VersionID - } else { - versionID = dobj.VersionID - } - scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{ - DeletedObject: DeletedObject{ - ObjectName: dobj.Name, - VersionID: versionID, - DeleteMarkerVersionID: dmVersionID, - DeleteMarkerMTime: DeleteMarkerMTime{dobj.ModTime}, - DeleteMarker: dobj.DeleteMarker, - ReplicationState: dobj.ReplicationState(), - }, - Bucket: obj.Bucket, - EventType: ReplicateILMDelete, - }, objLayer) - } - tags := newLifecycleAuditEvent(src, lcEvent).Tags() tags["version-id"] = dobj.VersionID if lcEvent.Action.DeleteAll() { @@ -1404,33 +1373,6 @@ func applyExpiryRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo) { globalExpiryState.enqueueByDays(obj, event, src) } -// ilmPurgeToDelete builds the bulk delete entry for an ILM-initiated removal -// of a noncurrent version. The entry carries a pending version-purge state so -// the purge is replicated and retried like an API version purge. -func ilmPurgeToDelete(ctx context.Context, bucket string, oi ObjectInfo, replicate bool) ObjectToDelete { - otd := ObjectToDelete{ - ObjectV: ObjectV{ - ObjectName: oi.Name, - VersionID: oi.VersionID, - }, - } - if !replicate { - return otd - } - delOpts := ObjectOptions{ - Versioned: globalBucketVersioningSys.PrefixEnabled(bucket, oi.Name), - VersionSuspended: globalBucketVersioningSys.PrefixSuspended(bucket, oi.Name), - } - dsc := checkReplicateDelete(ctx, bucket, otd, oi, delOpts, nil) - if !dsc.ReplicateAny() { - return otd - } - otd.VersionPurgeStatus = replication.VersionPurgePending - otd.VersionPurgeStatuses = dsc.PendingStatus() - otd.ReplicateDecisionStr = dsc.String() - return otd -} - // objectPath returns the prefix and object name. func (i *scannerItem) objectPath() string { return pathJoin(i.prefix, i.objectName) diff --git a/cmd/erasure-server-pool-lifecycle-batch_test.go b/cmd/erasure-server-pool-lifecycle-batch_test.go index 363a8768a..f3ea7c446 100644 --- a/cmd/erasure-server-pool-lifecycle-batch_test.go +++ b/cmd/erasure-server-pool-lifecycle-batch_test.go @@ -20,12 +20,10 @@ package cmd import ( "context" "errors" - "net/http" "testing" "github.com/minio/madmin-go/v3" "github.com/minio/minio/internal/bucket/lifecycle" - "github.com/minio/minio/internal/bucket/replication" "github.com/minio/minio/internal/bucket/versioning" "github.com/minio/minio/internal/event" "github.com/minio/minio/internal/pubsub" @@ -67,7 +65,10 @@ func (o *lifecycleBatchResult) DeleteObjects(ctx context.Context, bucket string, return o.deleted, o.errs } -func TestPoolsLifecycleTieredBatchReplication(t *testing.T) { +// A batch of ILM version purges keeps the cross-pool refusal and does not +// replicate: the eligible version is gone from every pool, the refused one +// stays, and nothing reaches the replication target. +func TestPoolsLifecycleTieredBatchNotReplicated(t *testing.T) { for _, identity := range []string{"uuid", "null"} { t.Run(identity, func(t *testing.T) { s := newSplitILM(t, versioning.Enabled, splitILMRules) @@ -77,24 +78,7 @@ func TestPoolsLifecycleTieredBatchReplication(t *testing.T) { for _, v := range []splitVersion{oldest, splitTiered("v2", 4, 0), splitTiered("v3", 3, 0), splitAt("v4", 2, 0)} { s.write(key, v) } - target := newILMTestTarget(t) - registerILMTestTarget(t, s.z, s.bucket, target, func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodDelete { - t.Errorf("unexpected target method: %s", r.Method) - w.WriteHeader(500) - return - } - target.deletes.Add(1) - opts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true, ReplicationRequest: true} - opts.SetReplicaStatus(replication.Replica) - if _, err := s.z.DeleteObject(r.Context(), target.bucket, key, opts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { - t.Errorf("target purge: %v", err) - w.WriteHeader(500) - return - } - w.WriteHeader(http.StatusNoContent) - }) - setILMTestReplicationConfig(t, s.bucket, target.arn, target.bucket) + target, requests := ilmTestTarget(t, s.z, s.bucket, key, false) stats := NewReplicationStats(t.Context(), nil) previous := globalReplicationStats.Swap(stats) t.Cleanup(func() { globalReplicationStats.Store(previous) }) @@ -106,15 +90,8 @@ func TestPoolsLifecycleTieredBatchReplication(t *testing.T) { if vid == "" { vid = nullVersionID } - oi, err := s.z.GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: vid, Versioned: true}) - if err != nil { - t.Fatal(err) - } - entry := ilmPurgeToDelete(t.Context(), s.bucket, oi, true) - if entry.VersionPurgeStatus != replication.VersionPurgePending { - t.Fatalf("fixture lacks purge decision: %+v", entry) - } - entries = append(entries, entry) + // As the scanner builds a noncurrent expiry entry. + entries = append(entries, ObjectToDelete{ObjectV: ObjectV{ObjectName: key, VersionID: vid}}) putConsistencyObject(t, s.z, target.bucket, key, 0, label, ObjectOptions{Versioned: identity != "null" || label != "v1", VersionID: vid}) } result := &lifecycleBatchResult{ObjectLayer: s.z} @@ -126,36 +103,26 @@ func TestPoolsLifecycleTieredBatchReplication(t *testing.T) { t.Errorf("retained version result: %v", result.errs[1]) } if result.deleted[1].VersionPurgeStatus() != "" || result.deleted[1].ReplicationState.ReplicateDecisionStr != "" { - t.Errorf("rejected entry retained replication state: %+v", result.deleted[1]) + t.Errorf("rejected entry carries replication state: %+v", result.deleted[1]) } - source, err := s.z.GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: entries[0].VersionID, Versioned: true}) - if !isErrMethodNotAllowed(err) || source.VersionPurgeStatus != replication.VersionPurgePending { - t.Errorf("source must be a pending purge tombstone before enqueue: status=%q err=%v", source.VersionPurgeStatus, err) + for pool := range s.z.serverPools { + if _, err := s.z.serverPools[pool].GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: entries[0].VersionID, Versioned: true}); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { + t.Errorf("pool %d: purged v1 remains: %v", pool, err) + } } retained, err := s.z.GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: entries[1].VersionID, Versioned: true}) if err != nil || retained.VersionPurgeStatus != "" { t.Errorf("rejected source changed: %+v, %v", retained, err) } - queue := globalReplicationPool.Get().workers[0] - if len(queue) != 1 { - t.Fatalf("queued %d purges, want only eligible v1", len(queue)) + if n := len(globalReplicationPool.Get().workers[0]); n != 0 { + t.Fatalf("queued %d replication tasks", n) } - deletion, ok := (<-queue).(DeletedObjectReplicationInfo) - if !ok || deletion.VersionID != entries[0].VersionID || deletion.EventType != ReplicateILMDelete || deletion.VersionPurgeStatus() != replication.VersionPurgePending { - t.Fatalf("wrong purge queued: %+v", deletion) + if n := requests.Load(); n != 0 { + t.Errorf("the target saw %d requests", n) } - if result := replicateDelete(t.Context(), deletion, s.z); result.VersionPurgeStatus() != replication.VersionPurgeComplete { - t.Fatalf("replicate purge: %+v", result) - } - if target.deletes.Load() != 1 { - t.Errorf("target delete count=%d", target.deletes.Load()) - } - for _, bucket := range []string{s.bucket, target.bucket} { - if _, err := s.z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: entries[0].VersionID}); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { - t.Errorf("purged v1 remains in %s: %v", bucket, err) - } - if _, err := s.z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: entries[1].VersionID}); err != nil { - t.Errorf("retained v3 missing from %s: %v", bucket, err) + for _, e := range entries { + if _, err := s.z.GetObjectInfo(t.Context(), target.bucket, key, ObjectOptions{VersionID: e.VersionID}); err != nil { + t.Errorf("target copy %s changed: %v", e.VersionID, err) } } }) diff --git a/cmd/erasure-server-pool-lifecycle-safety_test.go b/cmd/erasure-server-pool-lifecycle-safety_test.go index 0ee319350..89ee5a948 100644 --- a/cmd/erasure-server-pool-lifecycle-safety_test.go +++ b/cmd/erasure-server-pool-lifecycle-safety_test.go @@ -257,10 +257,9 @@ func TestPoolsLifecycleBatchRejectionReplication(t *testing.T) { if err != nil { t.Fatal(err) } - entry := ilmPurgeToDelete(t.Context(), s.bucket, oi, true) - if entry.VersionPurgeStatus != replication.VersionPurgePending { - t.Fatal("fixture lacks pending purge") - } + // As the scanner builds a noncurrent expiry entry: lifecycle + // purges carry no replication decision. + entry := ObjectToDelete{ObjectV: ObjectV{ObjectName: key, VersionID: oi.VersionID}} withdraw := func() { meta, err := globalBucketMetadataSys.Get(s.bucket) if err != nil { @@ -334,27 +333,17 @@ func TestPoolsLifecycleBatchRejectionReplication(t *testing.T) { } return } - if result.errs[0] != nil || len(queue) != 1 { + if result.errs[0] != nil || len(queue) != 0 { t.Fatalf("valid/absent purge: error=%v queued=%d", result.errs[0], len(queue)) } - source, err := s.z.GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: id}) - if mode == "normal" && (!isErrMethodNotAllowed(err) || source.VersionPurgeStatus != replication.VersionPurgePending) { - t.Fatalf("purge not committed locally: %+v %v", source, err) + if _, err := s.z.GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: id}); !isErrVersionNotFound(err) { + t.Fatalf("source must be gone, not a purge tombstone: %v", err) } - if mode == "already-absent" && !isErrVersionNotFound(err) { - t.Fatalf("source must actually be absent: %v", err) - } - deletion := (<-queue).(DeletedObjectReplicationInfo) - if result := replicateDelete(t.Context(), deletion, s.z); result.VersionPurgeStatus() != replication.VersionPurgeComplete { - t.Fatalf("purge failed: %+v", result) - } - if target.deletes.Load() != 1 { + if target.deletes.Load() != 0 { t.Fatalf("remote deletes=%d", target.deletes.Load()) } - for _, bucket := range []string{s.bucket, target.bucket} { - if _, err := s.z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: id}); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { - t.Errorf("purged version remains in %s: %v", bucket, err) - } + if _, err := s.z.GetObjectInfo(t.Context(), target.bucket, key, ObjectOptions{VersionID: id}); err != nil { + t.Errorf("target copy changed: %v", err) } }) } diff --git a/cmd/erasure-server-pool-lifecycle_test.go b/cmd/erasure-server-pool-lifecycle_test.go index ea39d2819..a4e233b0e 100644 --- a/cmd/erasure-server-pool-lifecycle_test.go +++ b/cmd/erasure-server-pool-lifecycle_test.go @@ -40,6 +40,8 @@ type splitILM struct { *migrationILM ids map[string]string // object and label -> version ID bodies map[string]string // body -> label + // replication is the replication configuration the scanner runs with. + replication replicationConfig } func newSplitILM(t *testing.T, status versioning.State, rules string) *splitILM { @@ -147,7 +149,7 @@ func (s *splitILM) scan(pool int) { for i, fi := range fivs.Versions { objInfos[i] = fi.ToObjectInfo(s.bucket, object, vc != nil && vc.Versioned(object)) } - item := scannerItem{bucket: s.bucket, objectName: object, lifeCycle: lc} + item := scannerItem{bucket: s.bucket, objectName: object, lifeCycle: lc, replication: s.replication} item.applyActions(ctx, s.z, objInfos, lr, &sizeSummary{}, nil) } s.runQueue() diff --git a/cmd/erasure-server-pool-migration-ilm_test.go b/cmd/erasure-server-pool-migration-ilm_test.go index 29791c994..da14d2df9 100644 --- a/cmd/erasure-server-pool-migration-ilm_test.go +++ b/cmd/erasure-server-pool-migration-ilm_test.go @@ -19,8 +19,6 @@ package cmd import ( "context" - "errors" - "net/http" "slices" "strings" "testing" @@ -29,7 +27,6 @@ import ( "github.com/minio/minio/internal/amztime" "github.com/minio/minio/internal/bucket/lifecycle" objectlock "github.com/minio/minio/internal/bucket/object/lock" - "github.com/minio/minio/internal/bucket/replication" "github.com/minio/minio/internal/bucket/versioning" xhttp "github.com/minio/minio/internal/http" ) @@ -460,7 +457,10 @@ func TestPoolsMigrationLifecycleChangePreservesSource(t *testing.T) { // A migration must route an ILM version purge through the normal delete // replication path before it can retire the source copy. Otherwise the // source pool can disappear while the replica still retains the version. -func TestPoolsMigrationLifecycleExpiryReplication(t *testing.T) { +// A decommission whose walk leaves a noncurrent version to its ILM expiry +// completes in one run: the expiry deletes the version and does not wait for +// a replication target, which lifecycle deletes do not reach. +func TestPoolsMigrationLifecycleExpiryNotReplicated(t *testing.T) { m := newMigrationILM(t, versioning.Enabled, false, migrationILMRule("1")) m.aged(0, "obj-01") @@ -471,37 +471,8 @@ func TestPoolsMigrationLifecycleExpiryReplication(t *testing.T) { oldStats := globalReplicationStats.Swap(stats) defer globalReplicationStats.Store(oldStats) - target := newILMTestTarget(t) - registerILMTestTarget(t, m.z, m.bucket, target, func(w http.ResponseWriter, r *http.Request) { - opts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true} - switch r.Method { - case http.MethodHead: - oi, err := m.z.GetObjectInfo(r.Context(), target.bucket, "obj-01", opts) - if err != nil { - writeErrorResponseHeadersOnly(w, toAPIError(r.Context(), err)) - return - } - if oi.DeleteMarker { - w.Header().Set(xhttp.AmzDeleteMarker, "true") - w.Header().Set(xhttp.AmzVersionID, oi.VersionID) - } - w.WriteHeader(http.StatusOK) - case http.MethodDelete: - target.deletes.Add(1) - opts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true" - opts.ReplicationRequest = true - opts.SetReplicaStatus(replication.Replica) - if _, err := m.z.DeleteObject(r.Context(), target.bucket, "obj-01", opts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { - writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL) - return - } - w.WriteHeader(http.StatusNoContent) - default: - t.Errorf("unexpected remote method %s", r.Method) - } - }) - setILMTestReplicationConfig(t, m.bucket, target.arn, target.bucket) - restorePool, receive := newILMTestReplicationPool(ctx, m.z, stats) + target, requests := ilmTestTarget(t, m.z, m.bucket, "obj-01", false) + restorePool, _ := newILMTestReplicationPool(ctx, m.z, stats) defer restorePool() if err := m.z.StartDecommission(ctx, 0); err != nil { @@ -511,41 +482,14 @@ func TestPoolsMigrationLifecycleExpiryReplication(t *testing.T) { m.z.poolMetaMutex.RLock() info := *m.z.poolMeta.Pools[0].Decommission m.z.poolMetaMutex.RUnlock() - if info.Complete || !info.Failed { - t.Fatalf("decommission completed before ILM purge replication: complete=%t failed=%t", info.Complete, info.Failed) - } - - deletion := receive() - if deletion.ObjectName != "obj-01" || deletion.VersionID == "" || deletion.VersionPurgeStatus() != replication.VersionPurgePending { - t.Fatalf("unexpected migration purge: %+v", deletion) - } - if result := replicateDelete(ctx, deletion, m.z); result.VersionPurgeStatus() != replication.VersionPurgeComplete { - t.Fatalf("migration purge replication result: %+v", result) - } - if target.deletes.Load() != 1 { - t.Fatalf("replication target saw %d deletes, want 1", target.deletes.Load()) - } - - // A failed decommission may already have moved the source pool's bucket - // metadata. Recreate the metadata directory before the resumable retry, - // as the server's scanner would do in production. - for _, set := range m.z.serverPools[0].sets { - for _, disk := range set.getDisks() { - if err := disk.MakeVol(ctx, pathJoin(minioMetaBucket, bucketMetaPrefix)); err != nil && !errors.Is(err, errVolumeExists) { - t.Fatal(err) - } - } - } - - if err := m.z.StartDecommission(ctx, 0); err != nil { - t.Fatal(err) - } - m.z.doDecommissionInRoutine(ctx, 0) - m.z.poolMetaMutex.RLock() - info = *m.z.poolMeta.Pools[0].Decommission - m.z.poolMetaMutex.RUnlock() if !info.Complete || info.Failed { - t.Fatalf("decommission did not complete after ILM purge replication: complete=%t failed=%t", info.Complete, info.Failed) + t.Fatalf("decommission: complete=%t failed=%t", info.Complete, info.Failed) + } + if n := len(globalReplicationPool.Get().workers[0]); n != 0 { + t.Fatalf("the migration expiry queued %d replication tasks", n) + } + if n := requests.Load(); n != 0 || target.deletes.Load() != 0 { + t.Fatalf("the target saw %d requests", n) } } diff --git a/cmd/object-handlers-common.go b/cmd/object-handlers-common.go index 41441882a..bbce6f021 100644 --- a/cmd/object-handlers-common.go +++ b/cmd/object-handlers-common.go @@ -28,7 +28,6 @@ import ( "github.com/minio/minio/internal/amztime" "github.com/minio/minio/internal/bucket/lifecycle" - "github.com/minio/minio/internal/bucket/replication" "github.com/minio/minio/internal/crypto" "github.com/minio/minio/internal/event" "github.com/minio/minio/internal/hash" @@ -508,20 +507,6 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD } } sendEvent(evArgs) - - // ILM purges carry a delete replication decision. Queue the purge - // after the local delete, just as the multi-object DELETE handler - // does for API-originated deletes. - if errs[i] == nil || isErrObjectNotFound(errs[i]) || isErrVersionNotFound(errs[i]) { - if dobj.VersionPurgeStatus() == replication.VersionPurgePending || - dobj.DeleteMarkerReplicationStatus() == replication.Pending { - scheduleReplicationDelete(ctx, DeletedObjectReplicationInfo{ - DeletedObject: dobj, - Bucket: bucket, - EventType: ReplicateILMDelete, - }, o) - } - } } } } diff --git a/cmd/replication-ilm-delete_test.go b/cmd/replication-ilm-delete_test.go index 3ac04a871..c88279e32 100644 --- a/cmd/replication-ilm-delete_test.go +++ b/cmd/replication-ilm-delete_test.go @@ -22,6 +22,7 @@ import ( "context" "net/http" "net/http/httptest" + "slices" "strings" "sync/atomic" "testing" @@ -32,6 +33,7 @@ import ( "github.com/minio/minio/internal/auth" "github.com/minio/minio/internal/bucket/lifecycle" "github.com/minio/minio/internal/bucket/replication" + "github.com/minio/minio/internal/bucket/versioning" xhttp "github.com/minio/minio/internal/http" "github.com/minio/minio/internal/once" ) @@ -101,41 +103,20 @@ func newILMTestReplicationPool(ctx context.Context, obj ObjectLayer, stats *Repl } } -// TestReplicateILMExpiryDeleteMarker verifies that a delete marker created by -// ILM expiry carries delete replication state and is queued for replication, -// exactly like a marker created through the S3 DELETE API. -func TestReplicateILMExpiryDeleteMarker(t *testing.T) { - defer DetectTestLeak(t)() - ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ - t: t, - objAPITest: testReplicateILMExpiryDeleteMarker, - }) -} - -func testReplicateILMExpiryDeleteMarker(obj ObjectLayer, instanceType, bucket string, _ http.Handler, creds auth.Credentials, t *testing.T) { - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() - stats := NewReplicationStats(ctx, nil) - oldStats := globalReplicationStats.Swap(stats) - defer globalReplicationStats.Store(oldStats) - - const name = "ilm-object" - if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil { - t.Fatal(err) - } - // DeleteObject refuses ILM expiry (Expiration.Expire) without lifecycle - // rules; a date in the past makes every version eligible immediately. - ilmXML := []byte(`ilm-expireEnabled2020-01-01T00:00:00Z`) - if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketLifecycleConfig, ilmXML); err != nil { - t.Fatal(err) - } - oi, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("data"), 4, "", ""), ObjectOptions{Versioned: true}) - if err != nil { - t.Fatal(err) - } +// Lifecycle deletes are not replicated: a delete marker or a version purge +// that ILM makes stays on the source, as in S3. The replication heal still +// sends an empty-state marker when existing-object replication applies to +// it, as it always did; that path is pinned here too. +// ilmTestTarget registers a target that applies the deletes it receives and +// records the requests, and the replication configuration of bucket with +// existing-object replication set as given. +func ilmTestTarget(t *testing.T, obj ObjectLayer, bucket, name string, existing bool) (*ilmDeleteTarget, *atomic.Int32) { + t.Helper() target := newILMTestTarget(t) + var requests atomic.Int32 registerILMTestTarget(t, obj, bucket, target, func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) topts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true} switch r.Method { case http.MethodHead: @@ -151,9 +132,6 @@ func testReplicateILMExpiryDeleteMarker(obj ObjectLayer, instanceType, bucket st w.WriteHeader(http.StatusOK) case http.MethodDelete: target.deletes.Add(1) - if topts.VersionID == "" { - t.Error("missing remote versionId") - } topts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true" topts.ReplicationRequest = true topts.SetReplicaStatus(replication.Replica) @@ -167,67 +145,59 @@ func testReplicateILMExpiryDeleteMarker(obj ObjectLayer, instanceType, bucket st } }) setILMTestReplicationConfig(t, bucket, target.arn, target.bucket) - restorePool, receive := newILMTestReplicationPool(ctx, obj, stats) - defer restorePool() - - if ok := applyExpiryOnNonTransitionedObjects(ctx, obj, oi, lifecycle.Event{Action: lifecycle.DeleteAction}, lcEventSrc_Scanner); !ok { - t.Fatalf("%s: ILM expiry did not apply", instanceType) - } - - deletion := receive() - if deletion.ObjectName == "" { - t.Fatalf("%s: ILM delete marker was not queued for replication", instanceType) - } - if !deletion.DeleteMarker || deletion.VersionID != "" || deletion.DeleteMarkerVersionID == "" { - t.Fatalf("%s: unexpected queued delete: %+v", instanceType, deletion) - } - if deletion.EventType != ReplicateILMDelete { - t.Fatalf("%s: queued event type = %q, want %q", instanceType, deletion.EventType, ReplicateILMDelete) - } - - // The ILM-created marker must carry pending replication state. - markerVID := deletion.DeleteMarkerVersionID - moi, merr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: markerVID, Versioned: true}) - if !isErrMethodNotAllowed(merr) || !moi.DeleteMarker { - t.Fatalf("%s: ILM marker missing: %v %v", instanceType, moi, merr) - } - if moi.ReplicationStatus != replication.Pending { - t.Fatalf("%s: ILM marker replication status = %q, want PENDING", instanceType, moi.ReplicationStatus) - } - - result := replicateDelete(ctx, deletion, obj) - if result.ReplicationStatus() != replication.Completed { - t.Fatalf("%s: marker replication result: %+v", instanceType, result) - } - if target.deletes.Load() != 1 { - t.Fatalf("%s: target saw %d deletes, want 1", instanceType, target.deletes.Load()) - } - // The marker must exist on the target. - roi, rerr := obj.GetObjectInfo(ctx, target.bucket, name, ObjectOptions{VersionID: markerVID, Versioned: true}) - if !isErrMethodNotAllowed(rerr) || !roi.DeleteMarker { - t.Fatalf("%s: target marker missing: %v %v", instanceType, roi, rerr) - } - // The source marker records the completed replication per target. The - // composite ReplicationStatus field is unreliable on read-back for - // markers carrying a replication timestamp, so assert the raw state. - moi, merr = obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: markerVID}) - if !isErrMethodNotAllowed(merr) || !strings.Contains(moi.ReplicationStatusInternal, "=COMPLETED;") { - t.Fatalf("%s: source marker state = %q (err %v), want COMPLETED", instanceType, moi.ReplicationStatusInternal, merr) + if !existing { + meta, err := globalBucketMetadataSys.Get(bucket) + if err != nil { + t.Fatal(err) + } + cfg := *meta.replicationConfig + cfg.Rules = slices.Clone(cfg.Rules) + cfg.Rules[0].ExistingObjectReplication = replication.ExistingObjectReplication{Status: replication.Disabled} + meta.replicationConfig = &cfg + globalBucketMetadataSys.Set(bucket, meta) } + return target, &requests } -// TestReplicateILMExpiryNoncurrentPurge verifies that a noncurrent version -// removed by ILM carries a pending version purge state, is queued for delete -// replication and converges on the target. -func TestReplicateILMExpiryNoncurrentPurge(t *testing.T) { +// ilmTestReplicationConfig is the replication configuration the scanner and +// the replication heal use for bucket. +func ilmTestReplicationConfig(t *testing.T, bucket string) replicationConfig { + t.Helper() + cfg, err := getReplicationConfig(t.Context(), bucket) + if err != nil || cfg == nil { + t.Fatalf("replication config: %v", err) + } + remotes, err := globalBucketTargetSys.ListBucketTargets(t.Context(), bucket) + if err != nil { + t.Fatalf("bucket targets: %v", err) + } + return replicationConfig{Config: cfg, remotes: remotes} +} + +// An ILM delete marker carries no replication state and is not queued; with +// existing-object replication off, the replication heal leaves it too. +func TestILMExpiryDeleteMarkerNotReplicated(t *testing.T) { defer DetectTestLeak(t)() - ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ - t: t, - objAPITest: testReplicateILMExpiryNoncurrentPurge, - }) + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: testILMExpiryDeleteMarkerNotReplicated}) } -func testReplicateILMExpiryNoncurrentPurge(obj ObjectLayer, instanceType, bucket string, _ http.Handler, creds auth.Credentials, t *testing.T) { +func testILMExpiryDeleteMarkerNotReplicated(obj ObjectLayer, instanceType, bucket string, _ http.Handler, _ auth.Credentials, t *testing.T) { + ilmMarkerReplicationCase(obj, instanceType, bucket, t, false) +} + +// The inherited exception: with existing-object replication on, the +// replication heal sends an empty-state ILM delete marker, as it did before +// lifecycle deletes were ever replication-aware. +func TestILMExpiryDeleteMarkerExistingObjectResync(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: testILMExpiryDeleteMarkerExistingObjectResync}) +} + +func testILMExpiryDeleteMarkerExistingObjectResync(obj ObjectLayer, instanceType, bucket string, _ http.Handler, _ auth.Credentials, t *testing.T) { + ilmMarkerReplicationCase(obj, instanceType, bucket, t, true) +} + +func ilmMarkerReplicationCase(obj ObjectLayer, instanceType, bucket string, t *testing.T, existing bool) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() stats := NewReplicationStats(ctx, nil) @@ -238,96 +208,145 @@ func testReplicateILMExpiryNoncurrentPurge(obj ObjectLayer, instanceType, bucket if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil { t.Fatal(err) } - if _, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("v1"), 2, "", ""), ObjectOptions{Versioned: true}); err != nil { + ilmXML := []byte(`ilm-expireEnabled2020-01-01T00:00:00Z`) + if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketLifecycleConfig, ilmXML); err != nil { t.Fatal(err) } - v1Info, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("v2"), 2, "", ""), ObjectOptions{Versioned: true}) + oi, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("data"), 4, "", ""), ObjectOptions{Versioned: true}) if err != nil { t.Fatal(err) } - v1 := v1Info.VersionID - if v1 == "" { - t.Fatalf("%s: missing version ID on second version", instanceType) + target, requests := ilmTestTarget(t, obj, bucket, name, existing) + restorePool, _ := newILMTestReplicationPool(ctx, obj, stats) + defer restorePool() + queue := globalReplicationPool.Get().workers[0] + + if ok := applyExpiryOnNonTransitionedObjects(ctx, obj, oi, lifecycle.Event{Action: lifecycle.DeleteAction}, lcEventSrc_Scanner); !ok { + t.Fatalf("%s: ILM expiry did not apply", instanceType) + } + if n := len(queue); n != 0 { + t.Fatalf("%s: ILM expiry queued %d replication tasks", instanceType, n) + } + marker, err := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{Versioned: true}) + if !isErrMethodNotAllowed(err) && !isErrObjectNotFound(err) || !marker.DeleteMarker || marker.VersionID == "" { + t.Fatalf("%s: latest after expiry: %+v %v", instanceType, marker, err) + } + marker.Bucket, marker.Name = bucket, name + if !marker.ReplicationStatus.Empty() || marker.ReplicationStatusInternal != "" { + t.Fatalf("%s: ILM marker carries replication state %q / %q", instanceType, marker.ReplicationStatus, marker.ReplicationStatusInternal) } - target := newILMTestTarget(t) - registerILMTestTarget(t, obj, bucket, target, func(w http.ResponseWriter, r *http.Request) { - topts := ObjectOptions{VersionID: r.URL.Query().Get("versionId"), Versioned: true} - switch r.Method { - case http.MethodDelete: - target.deletes.Add(1) - if topts.VersionID == "" { - t.Error("missing remote versionId") - } - topts.DeleteMarker = r.Header.Get(xhttp.MinIOSourceDeleteMarker) == "true" - topts.ReplicationRequest = true - topts.SetReplicaStatus(replication.Replica) - if _, err := obj.DeleteObject(r.Context(), target.bucket, name, topts); err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { - writeErrorResponse(r.Context(), w, toAPIError(r.Context(), err), r.URL) - return - } - w.WriteHeader(http.StatusNoContent) - default: - t.Errorf("unexpected remote method %s", r.Method) + // The replication heal, as the scanner runs it for the marker. + queueReplicationHeal(ctx, bucket, marker, ilmTestReplicationConfig(t, bucket), 0) + if !existing { + if n := len(queue); n != 0 { + t.Fatalf("%s: replication heal queued %d tasks for an ILM marker", instanceType, n) } - }) - // Pre-create a marker version on the target with the same version ID as - // the noncurrent version being purged, so the purge can be observed. - if _, err := obj.DeleteObject(ctx, target.bucket, name, ObjectOptions{VersionID: v1, Versioned: true, DeleteMarker: true, ReplicationRequest: true}); err != nil { + if n := requests.Load(); n != 0 { + t.Fatalf("%s: the target saw %d requests", instanceType, n) + } + return + } + // The test harness reuses the bucket name for each object layer type, so + // targets registered for earlier ones are still listed: expect a task for + // at least the target of this run. Existing-object replication may also + // resync the data version; only the marker's tasks matter here. + markerTasks := 0 + for n := len(queue); n > 0; n-- { + op := <-queue + deletion, ok := op.(DeletedObjectReplicationInfo) + if !ok { + continue + } + if !deletion.DeleteMarker || deletion.DeleteMarkerVersionID != marker.VersionID { + t.Fatalf("%s: unexpected resync task %+v", instanceType, deletion) + } + markerTasks++ + replicateDelete(ctx, deletion, obj) + } + if markerTasks == 0 { + t.Fatalf("%s: existing-object resync queued no task for the marker", instanceType) + } + if target.deletes.Load() == 0 { + t.Fatalf("%s: the target saw no delete", instanceType) + } +} + +// An API delete of a version still carries a purge decision and is queued: +// only lifecycle deletes are left out of replication. +func TestAPIVersionDeleteStillReplicated(t *testing.T) { + defer DetectTestLeak(t)() + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: testAPIVersionDeleteStillReplicated}) +} + +func testAPIVersionDeleteStillReplicated(obj ObjectLayer, instanceType, bucket string, router http.Handler, creds auth.Credentials, t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + stats := NewReplicationStats(ctx, nil) + oldStats := globalReplicationStats.Swap(stats) + defer globalReplicationStats.Store(oldStats) + + const name = "api-object" + if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig); err != nil { t.Fatal(err) } - setILMTestReplicationConfig(t, bucket, target.arn, target.bucket) + if _, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("v1"), 2, "", ""), ObjectOptions{Versioned: true}); err != nil { + t.Fatal(err) + } + v1, err := obj.PutObject(ctx, bucket, name, mustGetPutObjReader(t, strings.NewReader("v2"), 2, "", ""), ObjectOptions{Versioned: true}) + if err != nil { + t.Fatal(err) + } + ilmTestTarget(t, obj, bucket, name, false) restorePool, receive := newILMTestReplicationPool(ctx, obj, stats) defer restorePool() - // The scanner evaluates noncurrent version expiry per version; build the - // delete entry exactly the way the scanner does for the oldest version. - vi, verr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1, Versioned: true}) - if verr != nil { - t.Fatalf("%s: missing noncurrent version: %v", instanceType, verr) + req, err := newTestSignedRequestV4(http.MethodDelete, getDeleteObjectURL("", bucket, name)+"?versionId="+v1.VersionID, 0, nil, creds.AccessKey, creds.SecretKey, nil) + if err != nil { + t.Fatal(err) } - otd := ilmPurgeToDelete(ctx, bucket, vi, true) - if otd.VersionPurgeStatus != replication.VersionPurgePending || otd.VersionPurgeStatuses == "" || otd.ReplicateDecisionStr == "" { - t.Fatalf("%s: ILM purge entry carries no replication state: %+v", instanceType, otd) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusNoContent { + t.Fatalf("%s: DELETE: %d %s", instanceType, rec.Code, rec.Body.String()) } - if plain := ilmPurgeToDelete(ctx, bucket, vi, false); plain.VersionPurgeStatus != "" || plain.ReplicateDecisionStr != "" { - t.Fatalf("%s: non-replicating purge entry carries state: %+v", instanceType, plain) - } - - deleteObjectVersions(ctx, obj, bucket, []ObjectToDelete{otd}, []lifecycle.Event{{Action: lifecycle.DeleteVersionAction}}) - - // The version must remain locally as a tombstone with pending purge - // state until replication completes. - pi, perr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1, Versioned: true}) - if !isErrMethodNotAllowed(perr) { - t.Fatalf("%s: purged version metadata lost: %v", instanceType, perr) - } - if pi.VersionPurgeStatus != replication.VersionPurgePending { - t.Fatalf("%s: purged version state = %q, want PENDING", instanceType, pi.VersionPurgeStatus) - } - deletion := receive() - if deletion.ObjectName == "" { - t.Fatalf("%s: ILM noncurrent purge was not queued for replication", instanceType) - } - if deletion.VersionID != v1 || deletion.DeleteMarkerVersionID != "" || deletion.DeleteMarker { - t.Fatalf("%s: unexpected queued purge: %+v", instanceType, deletion) - } - - result := replicateDelete(ctx, deletion, obj) - if result.VersionPurgeStatus() != replication.VersionPurgeComplete { - t.Fatalf("%s: purge replication result: %+v", instanceType, result) - } - if target.deletes.Load() != 1 { - t.Fatalf("%s: target saw %d deletes, want 1", instanceType, target.deletes.Load()) - } - // The purge must have converged on the target. - if _, terr := obj.GetObjectInfo(ctx, target.bucket, name, ObjectOptions{VersionID: v1, Versioned: true}); !isErrVersionNotFound(terr) && !isErrObjectNotFound(terr) { - t.Fatalf("%s: target version still present: %v", instanceType, terr) - } - // Once the purge completes the local tombstone is freed and the version - // is gone on both sides. - if _, terr := obj.GetObjectInfo(ctx, bucket, name, ObjectOptions{VersionID: v1}); !isErrVersionNotFound(terr) && !isErrObjectNotFound(terr) { - t.Fatalf("%s: purged version still present locally: %v", instanceType, terr) + if deletion.VersionID != v1.VersionID || deletion.VersionPurgeStatus() != replication.VersionPurgePending { + t.Fatalf("%s: API version delete not queued for replication: %+v", instanceType, deletion) + } +} + +// The scanner's noncurrent expiry deletes the version at once: no pending +// purge tombstone, nothing queued, the target untouched. +func TestILMNoncurrentPurgeNotReplicatedThroughScanner(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + ctx := t.Context() + stats := NewReplicationStats(ctx, nil) + oldStats := globalReplicationStats.Swap(stats) + t.Cleanup(func() { globalReplicationStats.Store(oldStats) }) + + const key = "ncv-scanner-replicated" + s.write(key, splitAt("old", 5, 0)) + s.write(key, splitAt("new", 3, 0)) + old := s.ids[key+" old"] + target, requests := ilmTestTarget(t, s.z, s.bucket, key, false) + putConsistencyObject(t, s.z, target.bucket, key, 0, "replica", ObjectOptions{Versioned: true, VersionID: old}) + restorePool, _ := newILMTestReplicationPool(ctx, s.z, stats) + t.Cleanup(restorePool) + s.replication = ilmTestReplicationConfig(t, s.bucket) + + s.scan(0) + + if _, err := s.z.GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: old}); !isErrVersionNotFound(err) { + t.Fatalf("expired version: %v, want it gone (no pending purge tombstone)", err) + } + if n := len(globalReplicationPool.Get().workers[0]); n != 0 { + t.Fatalf("noncurrent expiry queued %d replication tasks", n) + } + if n := requests.Load(); n != 0 { + t.Fatalf("the target saw %d requests", n) + } + if _, err := s.z.GetObjectInfo(ctx, target.bucket, key, ObjectOptions{VersionID: old}); err != nil { + t.Fatalf("target copy changed: %v", err) } }