mirror of
https://github.com/pgsty/minio.git
synced 2026-11-11 10:05:34 +03:00
296c9710b1
Signed-off-by: Feng Ruohang <rh@vonng.com>
126 lines
4.4 KiB
Go
126 lines
4.4 KiB
Go
// 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"
|
|
"encoding/xml"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/minio/minio/internal/auth"
|
|
)
|
|
|
|
type metadataCacheLoadBarrier struct {
|
|
ObjectLayer
|
|
key string
|
|
read, resume chan struct{}
|
|
once sync.Once
|
|
}
|
|
|
|
func (b *metadataCacheLoadBarrier) GetObjectNInfo(ctx context.Context, bucket, key string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions) (*GetObjectReader, error) {
|
|
r, err := b.ObjectLayer.GetObjectNInfo(ctx, bucket, key, rs, h, opts)
|
|
if bucket == minioMetaBucket && key == b.key {
|
|
b.once.Do(func() { close(b.read); <-b.resume })
|
|
}
|
|
return r, err
|
|
}
|
|
|
|
func TestBucketMetadataLoadPreservesConcurrentPublication(t *testing.T) {
|
|
defer DetectTestLeak(t)()
|
|
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
|
|
t: t,
|
|
objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, _ auth.Credentials, t *testing.T) {
|
|
// Disk has the old unlocked config; a concurrent publication enables
|
|
// lock and versioning after the cache miss has started loading it.
|
|
enableBucketObjectLock(t, bucket)
|
|
newer, err := globalBucketMetadataSys.Get(bucket)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
globalBucketMetadataSys.Remove(bucket)
|
|
barrier := &metadataCacheLoadBarrier{ObjectLayer: obj, key: pathJoin(bucketMetaPrefix, bucket, bucketMetadataFile), read: make(chan struct{}), resume: make(chan struct{})}
|
|
previous := newObjectLayerFn()
|
|
setObjectLayer(barrier)
|
|
defer setObjectLayer(previous)
|
|
release := sync.OnceFunc(func() { close(barrier.resume) })
|
|
defer release()
|
|
type result struct {
|
|
meta BucketMetadata
|
|
reloaded bool
|
|
err error
|
|
}
|
|
done := make(chan result, 1)
|
|
go func() {
|
|
meta, reloaded, err := globalBucketMetadataSys.GetConfig(t.Context(), bucket)
|
|
done <- result{meta, reloaded, err}
|
|
}()
|
|
select {
|
|
case <-barrier.read:
|
|
case <-time.After(10 * time.Second):
|
|
t.Fatal("metadata load did not reach disk")
|
|
}
|
|
globalBucketMetadataSys.Set(bucket, newer)
|
|
release()
|
|
select {
|
|
case got := <-done:
|
|
if got.err != nil || got.reloaded || got.meta.objectLockConfig == nil || got.meta.objectLockConfig.ObjectLockEnabled != "Enabled" {
|
|
t.Fatalf("%s: stale load replaced the publication: %+v", backend, got)
|
|
}
|
|
case <-time.After(10 * time.Second):
|
|
t.Fatal("metadata load did not finish")
|
|
}
|
|
cached, err := globalBucketMetadataSys.Get(bucket)
|
|
if err != nil || cached.objectLockConfig == nil || cached.objectLockConfig.ObjectLockEnabled != "Enabled" {
|
|
t.Fatalf("cached lock lost: %+v %v", cached, err)
|
|
}
|
|
},
|
|
})
|
|
}
|
|
|
|
func TestBulkDeleteVersionLockWithStaleBucketCache(t *testing.T) {
|
|
for _, pools := range []int{1, 2} {
|
|
z, _ := bulkDeletePools(t, pools)
|
|
bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{LockEnabled: true})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, protection := range batchLockKinds {
|
|
oi := batchPutVersion(t, z, bucket, protection, batchLockMeta(protection))
|
|
saved, err := globalBucketMetadataSys.Get(bucket)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
globalBucketMetadataSys.Set(bucket, newBucketMetadata(bucket))
|
|
rec := httptest.NewRecorder()
|
|
router.ServeHTTP(rec, bulkDeleteLockRequest(t, bucket, protection, oi.VersionID))
|
|
globalBucketMetadataSys.Set(bucket, saved)
|
|
var response DeleteObjectsResponse
|
|
if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if rec.Code != http.StatusOK || len(response.Errors) != 1 || len(response.DeletedObjects) != 0 || !batchVersionExists(t, z, bucket, protection, oi.VersionID) {
|
|
t.Fatalf("pools=%d %s: stale bucket cache bypassed version lock: %s", pools, protection, rec.Body.String())
|
|
}
|
|
}
|
|
}
|
|
}
|