mirror of
https://github.com/pgsty/minio.git
synced 2026-11-09 17:21:51 +03:00
Merge R4 version-specific read expiry rechecks after R1
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
@@ -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
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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 = `<Rule><ID>both</ID><Status>Enabled</Status><Filter><Prefix>both-</Prefix></Filter>` +
|
||||
`<Expiration><Days>1</Days></Expiration><NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays></NoncurrentVersionExpiration></Rule>`
|
||||
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 = `<Rule><ID>transition</ID><Status>Enabled</Status><Filter><Prefix>transition-</Prefix></Filter>` +
|
||||
`<Transition><Days>1</Days><StorageClass>WARM</StorageClass></Transition></Rule>`
|
||||
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())
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user