diff --git a/cmd/erasure-server-pool-lifecycle.go b/cmd/erasure-server-pool-lifecycle.go index 16079e534..1f75bacd8 100644 --- a/cmd/erasure-server-pool-lifecycle.go +++ b/cmd/erasure-server-pool-lifecycle.go @@ -251,6 +251,43 @@ func (z *erasureServerPools) expiryNeedsAllPools(opts ObjectOptions) bool { return opts.DeletePrefix || opts.VersionID != "" || opts.Versioned || opts.VersionSuspended } +// recheckReadExpiry only rechecks a local ordinary-delete candidate. The +// global latest expiring does not authorize hiding a different version that +// GET/HEAD has already read. On an incomplete or rejected recheck, keep that +// successful read and do not queue an expiry. Leave response metadata intact. +func recheckReadExpiry(ctx context.Context, objectAPI ObjectLayer, oi ObjectInfo, evt lifecycle.Event) (ObjectInfo, lifecycle.Event) { + z, ok := objectAPI.(*erasureServerPools) + if !ok || !evt.Action.Delete() || evt.Action.DeleteRestored() || !z.expiryNeedsAllPools(ObjectOptions{ + Expiration: ExpirationOptions{Expire: true}, + VersionID: oi.VersionID, + Versioned: globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name), + VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name), + DeletePrefix: evt.Action.DeleteAll(), + }) { + return oi, evt + } + versions, events, err := z.lifecycleEventsAllPools(ctx, oi.Bucket, encodeDirObject(oi.Name)) + if err == nil { + versionID := oi.VersionID + if versionID == "" { + versionID = nullVersionID + } + for i, version := range versions { + id := version.VersionID + if id == "" { + id = nullVersionID + } + if id == versionID { + if events[i].Action.Delete() && !events[i].Action.DeleteRestored() { + return version, events[i] + } + break + } + } + } + return oi, lifecycle.Event{Action: lifecycle.NoneAction} +} + // migrationExpiryApplies reports whether the ILM expiry evt, which a // decommission or rebalance decided for oi on the versions of the pool it // walks, applies with the versions of the object in all pools too. Only then diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index 26784ac16..91a21d621 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -492,9 +492,10 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj return } event := evalActionFromLifecycle(ctx, *lc, rcfg, replcfg, objInfo) + expiryInfo, event := recheckReadExpiry(ctx, objectAPI, objInfo, event) if event.Action.Delete() { // apply whatever the expiry rule is. - applyExpiryRule(event, lcEventSrc_s3GetObject, objInfo) + applyExpiryRule(event, lcEventSrc_s3GetObject, expiryInfo) if !event.Action.DeleteRestored() { // If the ILM action is not on restored object return error. writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrNoSuchKey), r.URL) @@ -972,9 +973,10 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob return } event := evalActionFromLifecycle(ctx, *lc, rcfg, replcfg, objInfo) + expiryInfo, event := recheckReadExpiry(ctx, objectAPI, objInfo, event) if event.Action.Delete() { // apply whatever the expiry rule is. - applyExpiryRule(event, lcEventSrc_s3HeadObject, objInfo) + applyExpiryRule(event, lcEventSrc_s3HeadObject, expiryInfo) if !event.Action.DeleteRestored() { // If the ILM action is not on restored object return error. writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(ErrNoSuchKey)) diff --git a/cmd/object-lifecycle-read_test.go b/cmd/object-lifecycle-read_test.go new file mode 100644 index 000000000..449e3d47a --- /dev/null +++ b/cmd/object-lifecycle-read_test.go @@ -0,0 +1,380 @@ +// 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 ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/bucket/lifecycle" + "github.com/minio/minio/internal/bucket/versioning" + xhttp "github.com/minio/minio/internal/http" +) + +// An explicit-version read uses ReadVersion. Only the added lifecycle scan +// uses ReadXL, so its errors/absence can be injected without failing the read. +type lifecycleReadDisk struct { + StorageAPI + bucket, object string + scans *atomic.Int64 + err error +} + +func (d lifecycleReadDisk) ReadXL(ctx context.Context, bucket, object string, data bool) (RawFileInfo, error) { + if bucket == d.bucket && object == d.object { + d.scans.Add(1) + if d.err != nil { + return RawFileInfo{}, d.err + } + } + return d.StorageAPI.ReadXL(ctx, bucket, object, data) +} + +func lifecycleReadDisks(t *testing.T, s *splitILM, object string, failedPool int, err error) *atomic.Int64 { + t.Helper() + var scans atomic.Int64 + for p, pool := range s.z.serverPools { + pool.erasureDisksMu.Lock() + previous := pool.erasureDisks + wrapped := make([][]StorageAPI, len(previous)) + for i, disks := range previous { + wrapped[i] = append([]StorageAPI(nil), disks...) + for j, disk := range disks { + var injected error + if p == failedPool { + injected = err + } + wrapped[i][j] = lifecycleReadDisk{StorageAPI: disk, bucket: s.bucket, object: object, scans: &scans, err: injected} + } + } + pool.erasureDisks = wrapped + pool.erasureDisksMu.Unlock() + t.Cleanup(func() { + pool.erasureDisksMu.Lock() + pool.erasureDisks = previous + pool.erasureDisksMu.Unlock() + }) + } + return &scans +} + +// Real signed GET and HEAD requests use the author's two-pool storage fixture. +// Checking the queue as well as the response catches both false hiding and +// accidentally authorizing the latest version instead of the version read. +func TestPoolsLifecycleReadHTTP(t *testing.T) { + const bothRules = `bothEnabledboth-` + + `11` + for _, state := range []versioning.State{versioning.Enabled, versioning.Suspended} { + t.Run(string(state), func(t *testing.T) { + s := newSplitILM(t, state, splitILMRules+bothRules) + router := initTestAPIEndPoints(s.z, nil) + for _, tc := range []struct { + name, prefix string + latestAge float64 + readLatest bool + implicit bool + action lifecycle.Action + local lifecycle.Action + fault error + wantScan bool + }{ + {name: "old-version-new-latest", prefix: "exp-", latestAge: 0.01, local: lifecycle.DeleteAction, wantScan: true}, + {name: "old-version-expired-latest", prefix: "exp-", latestAge: 3, local: lifecycle.DeleteAction, wantScan: true}, + {name: "old-version-delete-all-latest", prefix: "eav-", latestAge: 3, local: lifecycle.DeleteAllVersionsAction, wantScan: true}, + {name: "expired-current", prefix: "exp-", latestAge: 3, readLatest: true, local: lifecycle.DeleteAction, action: lifecycle.DeleteAction, wantScan: true}, + {name: "expired-current-implicit", prefix: "exp-", latestAge: 3, readLatest: true, implicit: true, local: lifecycle.DeleteAction, action: lifecycle.DeleteAction, wantScan: true}, + {name: "expired-noncurrent", prefix: "both-", latestAge: 3, local: lifecycle.DeleteAction, action: lifecycle.DeleteVersionAction, wantScan: true}, + {name: "extra-read-quorum-error", prefix: "exp-", latestAge: 3, local: lifecycle.DeleteAction, fault: errDiskNotFound, wantScan: true}, + {name: "read-version-missing-from-recheck", prefix: "exp-", latestAge: 3, local: lifecycle.DeleteAction, fault: errFileNotFound, wantScan: true}, + {name: "local-none-global-expired", prefix: "ncv-", latestAge: 3, local: lifecycle.NoneAction}, + } { + for _, null := range []bool{false, true} { + for oldPool := range 2 { + t.Run(fmt.Sprintf("%s/null=%t/old-pool=%d", tc.name, null, oldPool), func(t *testing.T) { + key := fmt.Sprintf("%s%s-%t-%d", tc.prefix, tc.name, null, oldPool) + old := splitAt("v1", 5, oldPool) + old.null = null + s.write(key, old) + s.write(key, splitAt("v2", tc.latestAge, 1-oldPool)) + label := "v1" + if tc.readLatest { + label = "v2" + } + vid := s.ids[key+" "+label] + if vid == "" { + vid = nullVersionID + } + oi, err := s.z.GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: vid, Versioned: state == versioning.Enabled, VersionSuspended: state == versioning.Suspended}) + if err != nil { + t.Fatal(err) + } + lc, err := globalLifecycleSys.Get(s.bucket) + if err != nil { + t.Fatal(err) + } + if local := lc.Eval(oi.ToLifecycleOpts()); local.Action != tc.local { + t.Fatalf("fixture local event=%v, want %v", local.Action, tc.local) + } + versions, events, err := s.z.lifecycleEventsAllPools(t.Context(), s.bucket, key) + if err != nil { + t.Fatal(err) + } + index := 1 + if tc.readLatest { + index = 0 + } + if tc.local == lifecycle.NoneAction && events[index].Action != lifecycle.DeleteVersionAction { + t.Fatal("fixture must be globally expired but locally NoneAction") + } + scans := lifecycleReadDisks(t, s, key, oldPool, tc.fault) + for _, method := range []string{http.MethodGet, http.MethodHead} { + t.Run(method, func(t *testing.T) { + scans.Store(0) + requestID := vid + if tc.implicit { + requestID = "" + } + rec := consistencyRequest(t, router, method, s.bucket, key, requestID) + want := http.StatusOK + if tc.action.Delete() { + want = http.StatusNotFound + } + if rec.Code != want { + t.Errorf("status=%d want=%d body=%s", rec.Code, want, rec.Body.String()) + } + if want == http.StatusOK && method == http.MethodGet && rec.Body.String() != key+" "+label { + t.Errorf("body=%q, want requested version %s", rec.Body.String(), label) + } + if !tc.implicit && (scans.Load() > 0) != tc.wantScan { + t.Errorf("ReadXL calls=%d, want scan=%t", scans.Load(), tc.wantScan) + } + if tc.action.Delete() { + select { + case op := <-s.queue: + task, ok := op.(expiryTask) + src := lcEventSrc_s3GetObject + if method == http.MethodHead { + src = lcEventSrc_s3HeadObject + } + global := versions[index] + if !ok || task.src != src || task.event != events[index] || task.event.Action != tc.action || + task.objInfo.VersionID != global.VersionID || task.objInfo.IsLatest != global.IsLatest || + task.objInfo.NumVersions != global.NumVersions || !task.objInfo.SuccessorModTime.Equal(global.SuccessorModTime) { + t.Errorf("wrong queued expiry: %+v; want %s with global event %+v", task, vid, events[index]) + } + default: + t.Error("missing expiry task") + } + } + if n := len(s.queue); n != 0 { + t.Errorf("unexpected expiry tasks: %d", n) + for len(s.queue) > 0 { + <-s.queue + } + } + }) + } + }) + } + } + } + }) + } +} + +func TestPoolsLifecycleReadAuthorizationAndClose(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + initConfigSubsystem(t.Context(), s.z) + globalIAMSys.Init(t.Context(), s.z, globalEtcdClient, 2*time.Second) + denied := newObjectAttributesAuthzUser(t, "lifecycle-read", s.bucket, `"s3:PutObject"`) + router := initTestAPIEndPoints(s.z, nil) + for _, expire := range []bool{false, true} { + key := fmt.Sprintf("exp-reader-%t", expire) + v := splitAt("v1", 5, 0) + v.size = 1 << 20 // a non-inline reader retains the object read lock + s.write(key, v) + if !expire { + s.write(key, splitAt("v2", 0.01, 1)) + } + vid := s.ids[key+" v1"] + oi, err := s.z.GetObjectInfo(t.Context(), s.bucket, key, ObjectOptions{VersionID: vid, Versioned: true}) + if err != nil || oi.Inlined { + t.Fatalf("fixture must retain a reader lock: inlined=%t err=%v", oi.Inlined, err) + } + scans := lifecycleReadDisks(t, s, key, -1, nil) + for _, method := range []string{http.MethodGet, http.MethodHead} { + req, err := newTestSignedRequestV4(method, getGetObjectURL("", s.bucket, key)+"?versionId="+vid, 0, nil, denied.AccessKey, denied.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusForbidden || scans.Load() != 0 || len(s.queue) != 0 { + t.Fatalf("denied %s: status=%d scans=%d queued=%d", method, rec.Code, scans.Load(), len(s.queue)) + } + } + rec := consistencyRequest(t, router, http.MethodGet, s.bucket, key, vid) + want := http.StatusOK + if expire { + want = http.StatusNotFound + } + if rec.Code != want { + t.Errorf("authorized GET status=%d want=%d", rec.Code, want) + } + // The reader owns the multi-pool read lock until Close, including the + // early 404 path. A writer must acquire it immediately after the handler. + lk := s.z.NewNSLock(s.bucket, key) + lkctx, err := lk.GetLock(t.Context(), newDynamicTimeout(time.Second, time.Second)) + if err != nil { + t.Fatalf("GET leaked its reader/lock: %v", err) + } + lk.Unlock(lkctx) + for len(s.queue) > 0 { + <-s.queue + } + } +} + +func TestPoolsLifecycleReadBoundaries(t *testing.T) { + const transitionRules = `transitionEnabledtransition-` + + `1WARM` + for _, tc := range []struct { + name string + state versioning.State + null bool + singlePool bool + restored bool + transition bool + action lifecycle.Action + wantScan bool + }{ + {name: "single-pool", state: versioning.Enabled, singlePool: true, action: lifecycle.DeleteAction}, + {name: "single-pool-null", state: versioning.Enabled, singlePool: true, null: true, action: lifecycle.DeleteAction}, + {name: "unversioned", null: true, action: lifecycle.DeleteAction}, + {name: "current-null-enabled", state: versioning.Enabled, null: true, action: lifecycle.DeleteAction, wantScan: true}, + {name: "current-null-suspended", state: versioning.Suspended, null: true, action: lifecycle.DeleteAction, wantScan: true}, + {name: "transition", state: versioning.Enabled, transition: true, action: lifecycle.TransitionAction}, + {name: "restored", state: versioning.Enabled, restored: true, action: lifecycle.DeleteRestoredAction}, + } { + t.Run(tc.name, func(t *testing.T) { + s := newSplitILM(t, tc.state, splitILMRules+transitionRules) + key := "exp-" + tc.name + if tc.restored || tc.transition { + key = "transition-" + tc.name + } + v := splitAt("v1", 5, 0) + v.null, v.tiered = tc.null, tc.restored + s.write(key, v) + vid := s.ids[key+" v1"] + if vid == "" { + vid = nullVersionID + } + if tc.restored { + globalTierConfigMgr.drivercache["WARM"] = &tierExpiryTestBackend{objects: map[string][]byte{"remote-v1": []byte(key + " v1")}} + opts := ObjectOptions{VersionID: vid, Versioned: true} + opts.EvalMetadataFn = func(oi *ObjectInfo, _ error) (ReplicateDecision, error) { + oi.UserDefined[xhttp.AmzRestore] = completedRestoreObj(UTCNow().Add(-time.Hour)).String() + return ReplicateDecision{}, nil + } + if _, err := s.z.PutObjectMetadata(t.Context(), s.bucket, key, opts); err != nil { + t.Fatal(err) + } + } + z := s.z + if tc.singlePool { + // Exercise the single-pool dispatch with the same storage fixture. + z = &erasureServerPools{serverPools: s.z.serverPools[:1]} + } + router := initTestAPIEndPoints(z, []string{"GetObject", "HeadObject"}) + scans := lifecycleReadDisks(t, s, key, -1, nil) + for _, method := range []string{http.MethodGet, http.MethodHead} { + scans.Store(0) + rec := consistencyRequest(t, router, method, s.bucket, key, vid) + want := http.StatusOK + if tc.action.Delete() && !tc.action.DeleteRestored() { + want = http.StatusNotFound + } + if rec.Code != want || (scans.Load() > 0) != tc.wantScan { + t.Errorf("%s status=%d want=%d scans=%d wantScan=%t body=%s", method, rec.Code, want, scans.Load(), tc.wantScan, rec.Body.String()) + } + if want == http.StatusOK && method == http.MethodGet && rec.Body.String() != key+" v1" { + t.Errorf("%s lost object contents: %q", method, rec.Body.String()) + } + if tc.action.Delete() { + select { + case op := <-s.queue: + task, ok := op.(expiryTask) + if !ok || task.event.Action != tc.action || s.key(key, task.objInfo.VersionID) != s.key(key, vid) { + t.Errorf("%s wrong expiry: %+v", method, task) + } + default: + t.Errorf("%s missing expiry", method) + } + } + if n := len(s.queue); n != 0 { + t.Fatalf("%s queued %d unexpected expiries", method, n) + } + } + }) + } +} + +func TestPoolsLifecycleReadProxy(t *testing.T) { + s := newSplitILM(t, versioning.Enabled, splitILMRules) + const key, body = "exp-proxied", "remote object remains readable" + vid := mustGetUUID() + target := newILMTestTarget(t) + var remoteRequests atomic.Int64 + registerILMTestTarget(t, s.z, s.bucket, target, func(w http.ResponseWriter, r *http.Request) { + remoteRequests.Add(1) + w.Header().Set(xhttp.ContentLength, fmt.Sprint(len(body))) + w.Header().Set(xhttp.LastModified, UTCNow().Add(-5*24*time.Hour).Format(http.TimeFormat)) + w.Header().Set(xhttp.ETag, `"da39a3ee5e6b4b0d3255bfef95601890"`) + w.Header().Set(xhttp.AmzVersionID, vid) + if r.Method == http.MethodGet { + _, _ = w.Write([]byte(body)) + } + }) + setILMTestReplicationConfig(t, s.bucket, target.arn, target.bucket) + // The delete-only target fixture omits credentials; proxy discovery clones + // the target descriptor and requires its credentials field to be present. + globalBucketTargetSys.Lock() + globalBucketTargetSys.targetsMap[s.bucket][0].Credentials = &madmin.Credentials{AccessKey: "proxy-test"} + globalBucketTargetSys.Unlock() + stats := NewReplicationStats(t.Context(), nil) + previous := globalReplicationStats.Swap(stats) + t.Cleanup(func() { globalReplicationStats.Store(previous) }) + scans := lifecycleReadDisks(t, s, key, -1, nil) + router := initTestAPIEndPoints(s.z, nil) + for _, method := range []string{http.MethodGet, http.MethodHead} { + before := remoteRequests.Load() + rec := consistencyRequest(t, router, method, s.bucket, key, vid) + if rec.Code != http.StatusOK || remoteRequests.Load() == before || scans.Load() != 0 || len(s.queue) != 0 { + t.Errorf("proxy %s: status=%d remote=%d scans=%d queued=%d body=%s", method, rec.Code, remoteRequests.Load()-before, scans.Load(), len(s.queue), rec.Body.String()) + } + if method == http.MethodGet && rec.Body.String() != body { + t.Errorf("proxy body=%q", rec.Body.String()) + } + } +}