astelm/minio
1
0
Fork 0
mirror of https://github.com/pgsty/minio.git synced 2026-09-15 15:04:04 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(storage): reuse resolved copies for version DELETE callbacks

Browse Source 
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
2026-09-15 10:40:04 +08:00
parent 1cf529ce8a
commit 13bf126ebd
2 changed files with 20 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/erasure-server-pool-consistency.go
+3
View File
@@ -310,6 +310,9 @@ func (z *erasureServerPools) deleteObjectReconciled(ctx context.Context, bucket,
if opts.VersionID == "" || opts.DeleteMarker {
gerr = toObjectErr(errFileNotFound, bucket, object)
}
} else if opts.VersionID != "" {
// An addressed version already resolved every copy above.
logical = mergedPoolObjectInfo(copies)
} else {
versions, err := z.metadataPoolInfos(ctx, bucket, object, opts)
if err != nil {
cmd/erasure-server-pool-consistency_test.go
+17 -4
View File
@@ -217,8 +217,15 @@ func TestPoolsDeleteVersionUnreadablePool(t *testing.T) {
type consistencyDeleteCountDisk struct {
StorageAPI
bucket, object string
deletes *atomic.Int32
bucket, object string
metadataReads, deletes *atomic.Int32
}
func (d consistencyDeleteCountDisk) ReadVersion(ctx context.Context, origvolume, volume, path, version string, opts ReadOptions) (FileInfo, error) {
if volume == d.bucket && path == d.object {
d.metadataReads.Add(1)
}
return d.StorageAPI.ReadVersion(ctx, origvolume, volume, path, version, opts)
}
func (d consistencyDeleteCountDisk) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, force bool, opts DeleteOptions) error {
@@ -234,13 +241,13 @@ func TestPoolsDeleteVersionSingleCopy(t *testing.T) {
t.Run(fmt.Sprintf("pool=%d", primary), func(t *testing.T) {
object := fmt.Sprintf("single-copy-%d", primary)
oi := putConsistencyObject(t, z, bucket, object, primary, "payload", ObjectOptions{Versioned: true})
var deletes [2]atomic.Int32
var metadataReads, deletes [2]atomic.Int32
for i, pool := range z.serverPools {
set := pool.getHashedSet(object)
getDisks := set.getDisks
disks := append([]StorageAPI(nil), getDisks()...)
for j := range disks {
disks[j] = consistencyDeleteCountDisk{StorageAPI: disks[j], bucket: bucket, object: object, deletes: &deletes[i]}
disks[j] = consistencyDeleteCountDisk{StorageAPI: disks[j], bucket: bucket, object: object, metadataReads: &metadataReads[i], deletes: &deletes[i]}
}
set.getDisks = func() []StorageAPI { return disks }
defer func() { set.getDisks = getDisks }()
@@ -253,6 +260,12 @@ func TestPoolsDeleteVersionSingleCopy(t *testing.T) {
if err != nil || current.VersionID != oi.VersionID {
t.Errorf("metadata callback: %+v, %v", current, err)
}
// Count preflight reads before the physical delete reads its own metadata.
for i := range metadataReads {
if got := metadataReads[i].Load(); got != 16 {
t.Errorf("pool %d metadata reads before callbacks = %d; want once per disk (16)", i, got)
}
}
return ReplicateDecision{}, nil
},
EvalRetentionBypassFn: func(current ObjectInfo, err error) error {