diff --git a/cmd/storage-rest-common.go b/cmd/storage-rest-common.go index 17bcc5f69..d8a28ae29 100644 --- a/cmd/storage-rest-common.go +++ b/cmd/storage-rest-common.go @@ -83,13 +83,3 @@ type nsScannerResp struct { Update *dataUsageEntry `msg:"u"` Final *dataUsageCache `msg:"f"` } - -// Recycle resets the response, so that recycled responses cannot -// leak stale scanner data into a subsequent scan. -func (r *nsScannerResp) Recycle() { - if r == nil { - return - } - r.Update = nil - r.Final = nil -} diff --git a/cmd/storage-rest-server_test.go b/cmd/storage-rest-server_test.go index 354348882..93ce35803 100644 --- a/cmd/storage-rest-server_test.go +++ b/cmd/storage-rest-server_test.go @@ -1,6 +1,6 @@ -// Copyright (c) 2015-2023 MinIO, Inc. +// Copyright (c) 2026 Julien Laurenceau and Feng Ruohang // -// This file is part of MinIO Object Storage stack +// This file is part of Silo Object Storage. // // 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 @@ -17,20 +17,78 @@ package cmd -import "testing" +import ( + "context" + "testing" -// TestNSScannerRespPooling verifies that recycled NSScanner stream -// responses cannot leak stale fields from a previous scan into new -// responses, which would otherwise produce incorrect bucket usage -// data. -func TestNSScannerRespPooling(t *testing.T) { - dirty := storageNSScannerRPC.NewResponse() - dirty.Update = &dataUsageEntry{} - dirty.Final = &dataUsageCache{} - storageNSScannerRPC.PutResponse(dirty) + "github.com/minio/madmin-go/v3" + "github.com/minio/minio/internal/grid" +) - resp := storageNSScannerRPC.NewResponse() - if resp.Update != nil || resp.Final != nil { - t.Fatalf("pooled nsScannerResp leaked stale fields: Update %v, Final %v", resp.Update, resp.Final) +type scannerResponseTestDisk struct { + StorageAPI +} + +func (scannerResponseTestDisk) NSScanner(_ context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, _ madmin.HealScanMode, _ func() bool) (dataUsageCache, error) { + defer close(updates) + updates <- dataUsageEntry{Size: int64(cache.Info.NextCycle)} + cache.Info.NextCycle++ + return cache, nil +} + +func TestNSScannerHandlerClearsStaleResponses(t *testing.T) { + endpoint, err := NewEndpoint(t.TempDir()) + if err != nil { + t.Fatal(err) + } + globalLocalDrivesMu.Lock() + oldDrives, oldSets := globalLocalDrivesMap, globalLocalSetDrives + globalLocalDrivesMap = map[string]StorageAPI{endpoint.String(): scannerResponseTestDisk{}} + globalLocalSetDrives = nil + globalLocalDrivesMu.Unlock() + t.Cleanup(func() { + globalLocalDrivesMu.Lock() + globalLocalDrivesMap, globalLocalSetDrives = oldDrives, oldSets + globalLocalDrivesMu.Unlock() + }) + + oldRPC := storageNSScannerRPC + t.Cleanup(func() { storageNSScannerRPC = oldRPC }) + // Every new allocation is dirty too: the check must not depend on + // sync.Pool returning a previously used object. + storageNSScannerRPC = grid.NewStream[*nsScannerOptions, grid.NoPayload, *nsScannerResp](grid.HandlerNSScanner, + func() *nsScannerOptions { return &nsScannerOptions{} }, nil, + func() *nsScannerResp { + return &nsScannerResp{Update: &dataUsageEntry{Size: -1}, Final: &dataUsageCache{}} + }) + server := storageRESTServer{endpoint: endpoint} + for cycle := uint32(1); cycle <= 3; cycle++ { + out := make(chan *nsScannerResp, 2) + params := &nsScannerOptions{Cache: &dataUsageCache{Info: dataUsageCacheInfo{Name: "bucket", NextCycle: cycle}}} + if err := server.NSScannerHandler(t.Context(), params, out); err != nil { + t.Fatal(err) + } + if len(out) != 2 { + t.Fatalf("cycle %d: got %d responses, want an update and a final cache", cycle, len(out)) + } + for i := 0; i < 2; i++ { + resp := <-out + wire, err := resp.MarshalMsg(nil) + storageNSScannerRPC.PutResponse(resp) + if err != nil { + t.Fatal(err) + } + var decoded nsScannerResp + if _, err := decoded.UnmarshalMsg(wire); err != nil { + t.Fatal(err) + } + if i == 0 { + if decoded.Update == nil || decoded.Final != nil || decoded.Update.Size != int64(cycle) { + t.Errorf("cycle %d: invalid update response: %+v", cycle, decoded) + } + } else if decoded.Update != nil || decoded.Final == nil || decoded.Final.Info.NextCycle != cycle+1 { + t.Errorf("cycle %d: invalid final response: %+v", cycle, decoded) + } + } } } diff --git a/cmd/xl-storage-format-v2_test.go b/cmd/xl-storage-format-v2_test.go index 0076c8046..f5f682500 100644 --- a/cmd/xl-storage-format-v2_test.go +++ b/cmd/xl-storage-format-v2_test.go @@ -1214,14 +1214,34 @@ func TestXLMetaV2HeaderLegacyUnmarshalResetsEc(t *testing.T) { legacy = msgp.AppendUint8(legacy, uint8(base.Type)) legacy = msgp.AppendUint8(legacy, uint8(base.Flags)) - header := base - header.EcN, header.EcM = 4, 8 - x2 := xlMetaV2VersionHeaderV2{xlMetaV2VersionHeader: &header} - if _, err := x2.UnmarshalMsg(legacy); err != nil { - t.Fatalf("legacy V2 header unmarshal failed: %v", err) - } - if header.EcN != 0 || header.EcM != 0 { - t.Fatalf("stale erasure metadata leaked across objects: EcN %d, EcM %d", header.EcN, header.EcM) + for _, decoder := range []string{"UnmarshalMsg", "DecodeMsg", "loadIndexed"} { + t.Run(decoder, func(t *testing.T) { + header := base + header.EcN, header.EcM = 4, 8 + x2 := xlMetaV2VersionHeaderV2{xlMetaV2VersionHeader: &header} + var err error + switch decoder { + case "UnmarshalMsg": + _, err = x2.UnmarshalMsg(legacy) + case "DecodeMsg": + err = x2.DecodeMsg(msgp.NewReader(bytes.NewReader(legacy))) + case "loadIndexed": + indexed := msgp.AppendUint(nil, 2) + indexed = msgp.AppendUint(indexed, xlMetaVersion) + indexed = msgp.AppendInt(indexed, 1) + indexed = msgp.AppendBytes(indexed, legacy) + indexed = msgp.AppendBytes(indexed, nil) + xl := xlMetaV2{versions: []xlMetaV2ShallowVersion{{header: header}}} + err = xl.loadIndexed(indexed, nil) + header = xl.versions[0].header + } + if err != nil { + t.Fatal(err) + } + if header != base { + t.Fatalf("legacy header: got %+v, want %+v", header, base) + } + }) } // V1 headers (4 fields, never released) must zero the fields too. @@ -1232,7 +1252,7 @@ func TestXLMetaV2HeaderLegacyUnmarshalResetsEc(t *testing.T) { v1 = msgp.AppendUint8(v1, uint8(base.Type)) v1 = msgp.AppendUint8(v1, uint8(base.Flags)) - header = base + header := base header.EcN, header.EcM = 4, 8 header.Signature = [4]byte{9, 9, 9, 9} if _, err := header.unmarshalV1(v1); err != nil { diff --git a/internal/grid/handlers.go b/internal/grid/handlers.go index b0ca9e324..1d20fa9b9 100644 --- a/internal/grid/handlers.go +++ b/internal/grid/handlers.go @@ -698,11 +698,6 @@ func (h *StreamTypeHandler[Payload, Req, Resp]) PutRequest(r Req) { // These should be returned by the caller. func (h *StreamTypeHandler[Payload, Req, Resp]) PutResponse(r Resp) { if r != h.nilResp { - if rc, ok := any(r).(Recycler); ok { - // Recycled stream responses must not leak fields - // from a previous use into the next response. - rc.Recycle() - } //nolint:staticcheck // SA6002 IT IS A GENERIC VALUE! (and always a pointer) h.respPool.Put(r) }