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