mirror of
https://github.com/pgsty/minio.git
synced 2026-10-30 04:20:36 +03:00
fix(pools): preserve authoritative copies across failed deletes (#257)
Preserve the latest logical object when cleanup of another pool fails, report each bulk error with its original replication identity, and support retries after partial physical deletion. Reuse existing tier-aware reconciliation to avoid premature remote-object collection. Includes the original contribution and signed-off maintainer fixes. Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
@@ -93,6 +93,9 @@ jobs:
|
||||
- name: Run conditional PUT tests under race detector
|
||||
run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m
|
||||
|
||||
- name: Run multi-pool delete consistency tests under race detector
|
||||
run: go test -race ./cmd -run '^Test(Pools(Delete|BulkDelete|RetiringCopyPreservesSharedTierObject)|BulkDeleteQuorumEvidence)' -count=1 -timeout=5m
|
||||
|
||||
- name: Run multipart listing, cancellation and cleanup tests under race detector
|
||||
env:
|
||||
MINIO_API_REQUESTS_MAX: "10000"
|
||||
|
||||
@@ -1891,6 +1891,23 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
|
||||
if err == nil {
|
||||
continue
|
||||
}
|
||||
if err == errFileNotFound || err == errFileVersionNotFound {
|
||||
// A retry after partial physical deletion may mix removals
|
||||
// with already-absent replies. Both prove the desired outcome.
|
||||
// Marker creation and replication-state updates still require
|
||||
// successful writes; absence cannot acknowledge those operations.
|
||||
physical := true
|
||||
for _, v := range dedupVersions[i].Versions {
|
||||
if v.Deleted || v.MarkDeleted || v.ReplicationState.ReplicationStatusInternal != "" ||
|
||||
v.ReplicationState.VersionPurgeStatusInternal != "" || v.ReplicationState.ReplicaStatus != "" {
|
||||
physical = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if physical {
|
||||
continue
|
||||
}
|
||||
}
|
||||
for _, v := range dedupVersions[i].Versions {
|
||||
delObjErrs[index][v.Idx] = err
|
||||
}
|
||||
|
||||
@@ -27,6 +27,7 @@ import (
|
||||
madmin "github.com/minio/madmin-go/v3"
|
||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||
xhttp "github.com/minio/minio/internal/http"
|
||||
"github.com/pgsty/silo-pkg/v3/sync/errgroup"
|
||||
)
|
||||
|
||||
// objectPoolInfos reads the addressed version in every pool, including pools
|
||||
@@ -533,3 +534,110 @@ func (z *erasureServerPools) deleteObjectReconciled(ctx context.Context, bucket,
|
||||
oi.replicationDecision = opts.DeleteReplication.ReplicateDecisionStr
|
||||
return oi, err
|
||||
}
|
||||
|
||||
// deleteObjectsReconciled removes data while retaining each item's authoritative
|
||||
// copy until every other pool has completed its cleanup. The caller holds the
|
||||
// pools-layer write locks and separates marker creation from this batch.
|
||||
func (z *erasureServerPools) deleteObjectsReconciled(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) {
|
||||
deleted := make([]DeletedObject, len(objects))
|
||||
errs := make([]error, len(objects))
|
||||
primary := make([]int, len(objects))
|
||||
tiered := make([]bool, len(objects))
|
||||
lookup := errgroup.WithNErrs(len(objects)).WithConcurrency(32)
|
||||
for i, obj := range objects {
|
||||
// Preserve the full request identity even if lookup or cleanup fails.
|
||||
// The HTTP handler includes replication state in its response-index key.
|
||||
deleted[i] = DeletedObject{
|
||||
ObjectName: obj.ObjectName,
|
||||
VersionID: obj.VersionID,
|
||||
ReplicationState: obj.ReplicationState(),
|
||||
}
|
||||
primary[i] = -1
|
||||
lookup.Go(func() error {
|
||||
copies, err := z.objectPoolInfos(ctx, bucket, obj.ObjectName, ObjectOptions{VersionID: obj.VersionID, NoAuditLog: true})
|
||||
if err == nil {
|
||||
primary[i] = copies[0].Index
|
||||
for _, copy := range copies {
|
||||
tiered[i] = tiered[i] || copy.ObjInfo.TransitionedObject.Status == lifecycle.TransitionComplete
|
||||
}
|
||||
} else if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||
if isErrReadQuorum(err) {
|
||||
err = InsufficientWriteQuorum{Bucket: bucket, Object: decodeDirObject(obj.ObjectName)}
|
||||
}
|
||||
errs[i] = err
|
||||
}
|
||||
return nil
|
||||
}, i)
|
||||
}
|
||||
lookup.Wait()
|
||||
|
||||
type poolBatch struct {
|
||||
objects []ObjectToDelete
|
||||
indices []int
|
||||
deleted []DeletedObject
|
||||
errs []error
|
||||
}
|
||||
for phase := 0; phase < 2; phase++ {
|
||||
batches := make([]poolBatch, len(z.serverPools))
|
||||
for i, obj := range objects {
|
||||
if errs[i] != nil || tiered[i] {
|
||||
continue
|
||||
}
|
||||
for p := range z.serverPools {
|
||||
if (p == primary[i]) != (phase == 1) {
|
||||
continue
|
||||
}
|
||||
batches[p].objects = append(batches[p].objects, obj)
|
||||
batches[p].indices = append(batches[p].indices, i)
|
||||
}
|
||||
}
|
||||
// Each phase still uses the set layer's bulk disk I/O and runs pools
|
||||
// concurrently. Include apparently absent pools in cleanup: read-quorum
|
||||
// absence is not proof that an offline drive holds no remaining copy.
|
||||
group := errgroup.WithNErrs(len(z.serverPools))
|
||||
for p, pool := range z.serverPools {
|
||||
if len(batches[p].objects) == 0 {
|
||||
continue
|
||||
}
|
||||
group.Go(func() error {
|
||||
batches[p].deleted, batches[p].errs = pool.DeleteObjects(ctx, bucket, batches[p].objects, opts)
|
||||
return nil
|
||||
}, p)
|
||||
}
|
||||
group.Wait()
|
||||
for _, batch := range batches {
|
||||
for j, i := range batch.indices {
|
||||
if errs[i] != nil {
|
||||
continue
|
||||
}
|
||||
if batch.errs[j] != nil {
|
||||
errs[i] = batch.errs[j]
|
||||
} else if phase == 1 || primary[i] == -1 {
|
||||
deleted[i] = batch.deleted[j]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Tiered copies can share one remote object, including restored copies.
|
||||
// Reuse the existing reconciliation which suppresses premature tier GC.
|
||||
// Serializing this uncommon path also keeps duplicate key/version entries
|
||||
// in one request from racing their metadata updates under the shared lock.
|
||||
for i, obj := range objects {
|
||||
if !tiered[i] || errs[i] != nil {
|
||||
continue
|
||||
}
|
||||
deleteOpts := opts
|
||||
deleteOpts.VersionID, deleteOpts.NoLock = obj.VersionID, true
|
||||
deleteOpts.DeleteReplication = obj.ReplicationState()
|
||||
if opts.PrefixEnabledFn != nil {
|
||||
deleteOpts.Versioned = opts.PrefixEnabledFn(obj.ObjectName)
|
||||
}
|
||||
_, err := z.deleteObjectReconciled(ctx, bucket, obj.ObjectName, deleteOpts)
|
||||
if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||
errs[i] = err
|
||||
} else {
|
||||
deleted[i].found = true
|
||||
}
|
||||
}
|
||||
return deleted, errs
|
||||
}
|
||||
|
||||
@@ -1071,6 +1071,8 @@ func TestPoolsRetiringCopyPreservesSharedTierObject(t *testing.T) {
|
||||
restored bool
|
||||
unconditional bool
|
||||
skipFreeVersion bool
|
||||
bulk bool
|
||||
unversioned bool
|
||||
}{
|
||||
{name: "metadata-copy"},
|
||||
{name: "restored-metadata-copy", restored: true},
|
||||
@@ -1081,6 +1083,14 @@ func TestPoolsRetiringCopyPreservesSharedTierObject(t *testing.T) {
|
||||
{name: "ordinary-failed-primary-delete", deleting: true, unconditional: true, failPrimaryDelete: true},
|
||||
{name: "ordinary-successful-delete", deleting: true, unconditional: true},
|
||||
{name: "ordinary-skip-free-version", deleting: true, unconditional: true, skipFreeVersion: true},
|
||||
{name: "bulk-failed-primary-delete", deleting: true, bulk: true, failPrimaryDelete: true},
|
||||
{name: "bulk-restored-failed-primary-delete", deleting: true, bulk: true, restored: true, failPrimaryDelete: true},
|
||||
{name: "bulk-successful-delete", deleting: true, bulk: true},
|
||||
{name: "bulk-distinct-reference", deleting: true, bulk: true, differentRemote: true},
|
||||
{name: "bulk-skip-free-version", deleting: true, bulk: true, skipFreeVersion: true},
|
||||
{name: "unversioned-failed-primary-delete", deleting: true, unconditional: true, unversioned: true, failPrimaryDelete: true},
|
||||
{name: "unversioned-successful-delete", deleting: true, unconditional: true, unversioned: true},
|
||||
{name: "bulk-unversioned-failed-primary-delete", deleting: true, bulk: true, unversioned: true, failPrimaryDelete: true},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
@@ -1094,13 +1104,13 @@ func TestPoolsRetiringCopyPreservesSharedTierObject(t *testing.T) {
|
||||
if test.restored {
|
||||
metadata[xhttp.AmzRestore] = completedRestoreObj(time.Now().Add(time.Hour)).String()
|
||||
}
|
||||
oi := putConsistencyObject(t, z, bucket, object, 0, "data", ObjectOptions{Versioned: true, UserDefined: metadata})
|
||||
oi := putConsistencyObject(t, z, bucket, object, 0, "data", ObjectOptions{Versioned: !test.unversioned, UserDefined: metadata})
|
||||
secondaryMetadata := maps.Clone(metadata)
|
||||
if test.differentRemote {
|
||||
secondaryMetadata[ReservedMetadataPrefixLower+TransitionedObjectName] = "other-remote-object"
|
||||
}
|
||||
putConsistencyObject(t, z, bucket, object, 1, "data", ObjectOptions{
|
||||
Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime, UserDefined: secondaryMetadata,
|
||||
Versioned: !test.unversioned, VersionID: oi.VersionID, MTime: oi.ModTime, UserDefined: secondaryMetadata,
|
||||
})
|
||||
current, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID})
|
||||
if err != nil || current.TransitionedObject.Status != "complete" || current.IsRemote() == test.restored {
|
||||
@@ -1113,7 +1123,11 @@ func TestPoolsRetiringCopyPreservesSharedTierObject(t *testing.T) {
|
||||
set := z.serverPools[0].getHashedSet(object)
|
||||
getDisks := set.getDisks
|
||||
faulty := append([]StorageAPI(nil), getDisks()...)
|
||||
var failed, attempted atomic.Int32
|
||||
for i := range faulty {
|
||||
if test.bulk {
|
||||
faulty[i] = bulkDeleteFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: object, mode: "eio", fail: true, failed: &failed, attempted: &attempted}
|
||||
}
|
||||
faulty[i] = consistencyDeleteFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: object, version: oi.VersionID}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return faulty }
|
||||
@@ -1121,13 +1135,20 @@ func TestPoolsRetiringCopyPreservesSharedTierObject(t *testing.T) {
|
||||
}
|
||||
if test.deleting {
|
||||
opts := ObjectOptions{
|
||||
Versioned: true, VersionID: oi.VersionID, SkipFreeVersion: test.skipFreeVersion,
|
||||
Versioned: !test.unversioned, VersionID: oi.VersionID, SkipFreeVersion: test.skipFreeVersion,
|
||||
CheckPrecondFn: func(info ObjectInfo) bool { return info.ETag != current.ETag },
|
||||
}
|
||||
if test.unconditional {
|
||||
opts.CheckPrecondFn = nil
|
||||
}
|
||||
_, err := z.DeleteObject(t.Context(), bucket, object, opts)
|
||||
var err error
|
||||
if test.bulk {
|
||||
opts.CheckPrecondFn = nil
|
||||
_, errs := z.DeleteObjects(t.Context(), bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: object, VersionID: oi.VersionID}}}, opts)
|
||||
err = errs[0]
|
||||
} else {
|
||||
_, err = z.DeleteObject(t.Context(), bucket, object, opts)
|
||||
}
|
||||
if (err != nil) != test.failPrimaryDelete {
|
||||
t.Fatalf("unexpected authoritative delete result: %v", err)
|
||||
}
|
||||
@@ -1145,7 +1166,7 @@ func TestPoolsRetiringCopyPreservesSharedTierObject(t *testing.T) {
|
||||
_, err = z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID})
|
||||
primaryDeleted := test.deleting && !test.failPrimaryDelete
|
||||
if primaryDeleted {
|
||||
if !isErrVersionNotFound(err) {
|
||||
if !isErrVersionNotFound(err) && (!test.unversioned || !isErrObjectNotFound(err)) {
|
||||
t.Fatalf("authoritative copy survived successful delete: %v", err)
|
||||
}
|
||||
} else if err != nil {
|
||||
@@ -1306,3 +1327,182 @@ func TestPoolsDeleteDirectoryMarker(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// An unconditional DELETE of an object without versions removes its copies in
|
||||
// every pool. When the delete fails in one pool, the latest copy must stay in
|
||||
// place: removing it anyway serves an older write of the object.
|
||||
func TestPoolsDeleteUnversionedReportsOtherPoolFailure(t *testing.T) {
|
||||
for _, offline := range []bool{false, true} {
|
||||
for latest := range 2 {
|
||||
t.Run(fmt.Sprintf("offline=%v/latest=%d", offline, latest), func(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
const object = "unversioned-failure"
|
||||
older := 1 - latest
|
||||
putConsistencyObject(t, z, bucket, object, older, "older", ObjectOptions{MaxParity: true, MTime: time.Now().Add(-time.Hour)})
|
||||
oi := putConsistencyObject(t, z, bucket, object, latest, "latest", ObjectOptions{MaxParity: true})
|
||||
set := z.serverPools[older].getHashedSet(object)
|
||||
getDisks := set.getDisks
|
||||
disks := append([]StorageAPI(nil), getDisks()...)
|
||||
for i := range disks {
|
||||
switch {
|
||||
case !offline:
|
||||
disks[i] = consistencyDeleteFaultDisk{StorageAPI: disks[i], bucket: bucket, object: object}
|
||||
case i < len(disks)/2:
|
||||
// An EC 8+8 copy stays readable, but a delete needs 9 drives.
|
||||
disks[i] = nil
|
||||
}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
defer func() { set.getDisks = getDisks }()
|
||||
if _, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{}); err == nil {
|
||||
t.Error("delete reported success despite the other pool's write failure")
|
||||
}
|
||||
set.getDisks = getDisks
|
||||
if got, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); err != nil || got.ETag != oi.ETag {
|
||||
t.Errorf("after the failed delete the object reads ETag %q, %v; want the latest write %q", got.ETag, err, oi.ETag)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// consistencyBulkDeleteFaultDisk fails the multi-object deletes of one object.
|
||||
type consistencyBulkDeleteFaultDisk struct {
|
||||
StorageAPI
|
||||
bucket, object string
|
||||
}
|
||||
|
||||
func (d consistencyBulkDeleteFaultDisk) DeleteVersions(ctx context.Context, volume string, versions []FileInfoVersions, opts DeleteOptions) []error {
|
||||
errs := make([]error, len(versions))
|
||||
var rest []FileInfoVersions
|
||||
var restIdx []int
|
||||
for i, v := range versions {
|
||||
if volume == d.bucket && v.Name == d.object {
|
||||
errs[i] = errDiskFull
|
||||
continue
|
||||
}
|
||||
rest = append(rest, v)
|
||||
restIdx = append(restIdx, i)
|
||||
}
|
||||
if len(rest) > 0 {
|
||||
for i, err := range d.StorageAPI.DeleteVersions(ctx, volume, rest, opts) {
|
||||
errs[restIdx[i]] = err
|
||||
}
|
||||
}
|
||||
return errs
|
||||
}
|
||||
|
||||
// A multi-object delete must not report an object or version as deleted while
|
||||
// a pool that failed to delete it still holds a copy.
|
||||
func TestPoolsDeleteObjectsReportsOtherPoolFailure(t *testing.T) {
|
||||
for _, versioned := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("versioned=%v", versioned), func(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
const object = "bulk-delete-failure"
|
||||
opts := ObjectOptions{Versioned: versioned}
|
||||
oi := putConsistencyObject(t, z, bucket, object, 0, "latest", opts)
|
||||
if versioned {
|
||||
// The copy a data movement leaves until it removes the source.
|
||||
putConsistencyObject(t, z, bucket, object, 1, "latest", ObjectOptions{Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime})
|
||||
} else {
|
||||
putConsistencyObject(t, z, bucket, object, 1, "older", ObjectOptions{MTime: time.Now().Add(-time.Hour)})
|
||||
}
|
||||
set := z.serverPools[1].getHashedSet(object)
|
||||
getDisks := set.getDisks
|
||||
faulty := append([]StorageAPI(nil), getDisks()...)
|
||||
for i := range faulty {
|
||||
faulty[i] = consistencyBulkDeleteFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: object}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return faulty }
|
||||
defer func() { set.getDisks = getDisks }()
|
||||
_, errs := z.DeleteObjects(t.Context(), bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: object, VersionID: oi.VersionID}}}, opts)
|
||||
if errs[0] == nil {
|
||||
t.Error("multi-object delete reported success despite the other pool's write failure")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Pools delete a multi-object batch in parallel. With half the drives of the
|
||||
// older copy's set offline, an EC 8+8 copy stays readable but cannot be
|
||||
// deleted: the batch must fail the object and keep its latest copy, instead of
|
||||
// deleting it and serving the older write once the drives are back.
|
||||
func TestPoolsDeleteObjectsKeepsLatestWithoutDeleteQuorum(t *testing.T) {
|
||||
for latest := range 2 {
|
||||
t.Run(fmt.Sprintf("latest=%d", latest), func(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
const object = "bulk-delete-quorum"
|
||||
older := 1 - latest
|
||||
putConsistencyObject(t, z, bucket, object, older, "older", ObjectOptions{MaxParity: true, MTime: time.Now().Add(-time.Hour)})
|
||||
oi := putConsistencyObject(t, z, bucket, object, latest, "latest", ObjectOptions{MaxParity: true})
|
||||
set := z.serverPools[older].getHashedSet(object)
|
||||
getDisks := set.getDisks
|
||||
disks := append([]StorageAPI(nil), getDisks()...)
|
||||
for i := range len(disks) / 2 {
|
||||
disks[i] = nil
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
defer func() { set.getDisks = getDisks }()
|
||||
_, errs := z.DeleteObjects(t.Context(), bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: object}}}, ObjectOptions{})
|
||||
if !errors.As(errs[0], &InsufficientWriteQuorum{}) {
|
||||
t.Errorf("multi-object delete: got %v, want InsufficientWriteQuorum", errs[0])
|
||||
}
|
||||
set.getDisks = getDisks
|
||||
if got, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); err != nil || got.ETag != oi.ETag {
|
||||
t.Errorf("after the failed delete the object reads ETag %q, %v; want the latest write %q", got.ETag, err, oi.ETag)
|
||||
}
|
||||
// Nothing was deleted, so the retry removes every copy.
|
||||
if _, errs = z.DeleteObjects(t.Context(), bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: object}}}, ObjectOptions{}); errs[0] != nil {
|
||||
t.Fatal(errs[0])
|
||||
}
|
||||
for i, pool := range z.serverPools {
|
||||
if _, err := pool.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); !isErrObjectNotFound(err) {
|
||||
t.Errorf("pool %d retained the object after the retry: %v", i, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A delete marker hides the object in every pool. When the versions of an
|
||||
// object are in two pools and the set of one of them cannot take a write, the
|
||||
// marker goes to the other pool instead of failing the delete. The versions
|
||||
// stay below the marker.
|
||||
func TestPoolsDeleteMarkerSkipsSetWithoutQuorum(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
degraded int // pool whose set loses drives
|
||||
offline int
|
||||
}{
|
||||
{"older version readable", 0, 8},
|
||||
{"older version unreadable", 0, 9},
|
||||
{"latest version readable", 1, 8},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
z, bucket := consistencyPools(t)
|
||||
const object = "marker-quorum"
|
||||
v1 := putConsistencyObject(t, z, bucket, object, 0, "v1", ObjectOptions{Versioned: true, MaxParity: true, MTime: time.Now().Add(-time.Hour)})
|
||||
v2 := putConsistencyObject(t, z, bucket, object, 1, "v2", ObjectOptions{Versioned: true, MaxParity: true})
|
||||
set := z.serverPools[tc.degraded].getHashedSet(object)
|
||||
getDisks := set.getDisks
|
||||
disks := append([]StorageAPI(nil), getDisks()...)
|
||||
for i := range tc.offline {
|
||||
disks[i] = nil
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
defer func() { set.getDisks = getDisks }()
|
||||
if _, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{Versioned: true}); err != nil {
|
||||
t.Fatalf("delete with the set of pool %d without quorum: %v", tc.degraded, err)
|
||||
}
|
||||
set.getDisks = getDisks
|
||||
if _, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); !isErrObjectNotFound(err) && !isErrMethodNotAllowed(err) {
|
||||
t.Errorf("object still readable after the delete: %v", err)
|
||||
}
|
||||
for pool, v := range []ObjectInfo{v1, v2} {
|
||||
if _, err := z.serverPools[pool].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: v.VersionID}); err != nil {
|
||||
t.Errorf("pool %d lost version %s: %v", pool, v.VersionID, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,582 @@
|
||||
// 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 (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/xml"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
minio "github.com/minio/minio-go/v7"
|
||||
)
|
||||
|
||||
func bulkDeletePools(t *testing.T, n int) (*erasureServerPools, string) {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
dirs, err := getRandomDisks(4 * n)
|
||||
if err != nil {
|
||||
cancel()
|
||||
t.Fatal(err)
|
||||
}
|
||||
var endpoints EndpointServerPools
|
||||
for i := range n {
|
||||
endpoints = append(endpoints, mustGetPoolEndpoints(i, dirs[4*i:4*i+4]...)...)
|
||||
}
|
||||
obj, _, err := initObjectLayer(ctx, endpoints)
|
||||
if err != nil {
|
||||
cancel()
|
||||
removeRoots(dirs)
|
||||
t.Fatal(err)
|
||||
}
|
||||
z := obj.(*erasureServerPools)
|
||||
previous := newObjectLayerFn()
|
||||
setObjectLayer(z)
|
||||
t.Cleanup(func() { cancel(); z.Shutdown(context.Background()); removeRoots(dirs); setObjectLayer(previous) })
|
||||
bucket := "bulk-delete-consistency"
|
||||
if err := z.MakeBucket(ctx, bucket, MakeBucketOptions{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return z, bucket
|
||||
}
|
||||
|
||||
func bulkDeleteBody(t *testing.T, z *erasureServerPools, bucket, object, version string) string {
|
||||
t.Helper()
|
||||
r, err := z.GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{VersionID: version})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer r.Close()
|
||||
body, err := io.ReadAll(r)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return string(body)
|
||||
}
|
||||
|
||||
// All drives are online and all metadata reads succeed. Failure starts inside
|
||||
// DeleteVersions, after any preflight. Count attempts to delete the latest copy
|
||||
// instead of waiting for them: a safe implementation must never make that call.
|
||||
type bulkDeleteFaultDisk struct {
|
||||
StorageAPI
|
||||
bucket, object, mode string
|
||||
fail, latest bool
|
||||
failed, attempted *atomic.Int32
|
||||
}
|
||||
|
||||
func (d bulkDeleteFaultDisk) DeleteVersions(ctx context.Context, bucket string, versions []FileInfoVersions, opts DeleteOptions) []error {
|
||||
if bucket != d.bucket {
|
||||
return d.StorageAPI.DeleteVersions(ctx, bucket, versions, opts)
|
||||
}
|
||||
errs := make([]error, len(versions))
|
||||
for i, v := range versions {
|
||||
if v.Name != d.object {
|
||||
errs[i] = d.StorageAPI.DeleteVersions(ctx, bucket, []FileInfoVersions{v}, opts)[0]
|
||||
continue
|
||||
}
|
||||
if d.latest {
|
||||
d.attempted.Add(1)
|
||||
}
|
||||
if !d.fail {
|
||||
errs[i] = d.StorageAPI.DeleteVersions(ctx, bucket, []FileInfoVersions{v}, opts)[0]
|
||||
continue
|
||||
}
|
||||
d.failed.Add(1)
|
||||
switch d.mode {
|
||||
case "eio":
|
||||
errs[i] = syscall.EIO
|
||||
case "disconnect":
|
||||
errs[i] = errDiskNotFound
|
||||
case "permission":
|
||||
meta := filepath.Join(d.String(), bucket, v.Name, xlStorageFormatFile)
|
||||
info, err := os.Stat(meta)
|
||||
if err == nil {
|
||||
err = os.Chmod(meta, 0)
|
||||
}
|
||||
if err != nil {
|
||||
errs[i] = err
|
||||
continue
|
||||
}
|
||||
errs[i] = d.StorageAPI.DeleteVersions(ctx, bucket, []FileInfoVersions{v}, opts)[0]
|
||||
if err := os.Chmod(meta, info.Mode()); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return errs
|
||||
}
|
||||
|
||||
func requireBulkDeletePermissionFailure(t *testing.T) {
|
||||
t.Helper()
|
||||
probe := filepath.Join(t.TempDir(), "permission-probe")
|
||||
if err := os.WriteFile(probe, []byte("metadata"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Chmod(probe, 0); err != nil {
|
||||
t.Skipf("cannot inject filesystem permissions: %v", err)
|
||||
}
|
||||
_, err := os.ReadFile(probe)
|
||||
if restoreErr := os.Chmod(probe, 0o600); restoreErr != nil {
|
||||
t.Fatal(restoreErr)
|
||||
}
|
||||
if !os.IsPermission(err) {
|
||||
t.Skip("this user/filesystem can read mode-000 files")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolsBulkDeleteKeepsLatestOnIOFailure(t *testing.T) {
|
||||
for _, mode := range []string{"eio", "disconnect", "permission"} {
|
||||
for _, n := range []int{2, 3} {
|
||||
for primary := range n {
|
||||
for _, kind := range []string{"unversioned", "null", "uuid"} {
|
||||
t.Run(fmt.Sprintf("%s/pools=%d/primary=%d/%s", mode, n, primary, kind), func(t *testing.T) {
|
||||
if mode == "permission" {
|
||||
requireBulkDeletePermissionFailure(t)
|
||||
}
|
||||
z, bucket := bulkDeletePools(t, n)
|
||||
object, old := "target", (primary+1)%n
|
||||
opts := ObjectOptions{MTime: UTCNow().Add(-time.Hour), MaxParity: true}
|
||||
oldBody := "old"
|
||||
if kind == "uuid" {
|
||||
opts.Versioned, opts.VersionID, oldBody = true, mustGetUUID(), "latest"
|
||||
}
|
||||
putConsistencyObject(t, z, bucket, object, old, oldBody, opts)
|
||||
opts.MTime = opts.MTime.Add(time.Minute)
|
||||
latest := putConsistencyObject(t, z, bucket, object, primary, "latest", opts)
|
||||
version := opts.VersionID
|
||||
if kind == "null" {
|
||||
version = nullVersionID
|
||||
}
|
||||
for _, key := range []string{"before", "after"} {
|
||||
putConsistencyObject(t, z, bucket, key, primary, "unrelated", ObjectOptions{})
|
||||
}
|
||||
var failed, attempted atomic.Int32
|
||||
var restore []func()
|
||||
for _, p := range []int{old, primary} {
|
||||
set := z.serverPools[p].getHashedSet(object)
|
||||
original := set.getDisks
|
||||
disks := append([]StorageAPI(nil), original()...)
|
||||
for i, disk := range disks {
|
||||
disks[i] = bulkDeleteFaultDisk{StorageAPI: disk, bucket: bucket, object: object, mode: mode, fail: p == old, latest: p == primary, failed: &failed, attempted: &attempted}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
restore = append(restore, func() { set.getDisks = original })
|
||||
}
|
||||
defer func() {
|
||||
for _, f := range restore {
|
||||
f()
|
||||
}
|
||||
}()
|
||||
objects := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: "before"}}, {ObjectV: ObjectV{ObjectName: object, VersionID: version}}, {ObjectV: ObjectV{ObjectName: "after"}}}
|
||||
deleted, errs := z.DeleteObjects(t.Context(), bucket, objects, ObjectOptions{})
|
||||
if failed.Load() != 4 || attempted.Load() != 0 {
|
||||
t.Errorf("cleanup failures=%d; latest deletion attempts=%d, want 4/0", failed.Load(), attempted.Load())
|
||||
}
|
||||
if errs[1] == nil {
|
||||
t.Error("cleanup failure was reported as success")
|
||||
}
|
||||
for _, i := range []int{0, 2} {
|
||||
if errs[i] != nil || deleted[i].ObjectName != objects[i].ObjectName {
|
||||
t.Errorf("unrelated item %d: %+v, %v", i, deleted[i], errs[i])
|
||||
}
|
||||
}
|
||||
if body := bulkDeleteBody(t, z, bucket, object, version); body != "latest" {
|
||||
t.Errorf("GET during failure=%q", body)
|
||||
}
|
||||
for _, f := range restore {
|
||||
f()
|
||||
}
|
||||
if body := bulkDeleteBody(t, z, bucket, object, version); body != "latest" {
|
||||
t.Errorf("GET after recovery=%q", body)
|
||||
}
|
||||
listed, err := z.ListObjectVersions(t.Context(), bucket, object, "", "", "", 100)
|
||||
if err != nil || len(listed.Objects) != 1 || listed.Objects[0].ETag != latest.ETag {
|
||||
t.Errorf("recovered versions=%+v, %v", listed.Objects, err)
|
||||
}
|
||||
_, retry := z.DeleteObjects(t.Context(), bucket, objects[1:2], ObjectOptions{})
|
||||
if retry[0] != nil {
|
||||
t.Fatal(retry[0])
|
||||
}
|
||||
_, err = z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: version})
|
||||
if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||
t.Errorf("retry retained the object: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolsBulkDeleteReplicationIdentity(t *testing.T) {
|
||||
z, _ := bulkDeletePools(t, 2)
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
defer cancel()
|
||||
bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{VersioningEnabled: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Error("unexpected outbound replication for a failed deletion")
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}))
|
||||
defer remote.Close()
|
||||
client, err := minio.New(strings.TrimPrefix(remote.URL, "http://"), &minio.Options{Region: "us-east-1", MaxRetries: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, restoreReplication := r5ReplicationFixture(t, z, bucket, client)
|
||||
defer restoreReplication()
|
||||
versions := make(map[string]string)
|
||||
for _, key := range []string{"first", "second"} {
|
||||
v := putConsistencyObject(t, z, bucket, key, 0, "payload", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)})
|
||||
putConsistencyObject(t, z, bucket, key, 1, "payload", ObjectOptions{Versioned: true, VersionID: v.VersionID, MTime: v.ModTime})
|
||||
versions[key] = v.VersionID
|
||||
}
|
||||
set := z.serverPools[0].getHashedSet("first")
|
||||
original := set.getDisks
|
||||
disks := append([]StorageAPI(nil), original()...)
|
||||
disks[0], disks[1] = nil, nil
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
defer func() { set.getDisks = original }()
|
||||
body := []byte(fmt.Sprintf("<Delete><Object><Key>first</Key><VersionId>%s</VersionId></Object><Object><Key>second</Key><VersionId>%s</VersionId></Object></Delete>", versions["first"], versions["second"]))
|
||||
req, err := newTestSignedRequestV4(http.MethodPost, getMultiDeleteObjectURL("", bucket), int64(len(body)), bytes.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
router.ServeHTTP(rec, req)
|
||||
var result struct {
|
||||
Errors []struct {
|
||||
Key, Code string
|
||||
VersionID string `xml:"VersionId"`
|
||||
} `xml:"Error"`
|
||||
Deleted []struct{ Key string } `xml:"Deleted"`
|
||||
}
|
||||
if err := xml.Unmarshal(rec.Body.Bytes(), &result); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if rec.Code != http.StatusOK || len(result.Errors) != 2 || len(result.Deleted) != 0 {
|
||||
t.Fatalf("lost response identity: HTTP %d %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
for _, e := range result.Errors {
|
||||
if versions[e.Key] == "" || e.VersionID != versions[e.Key] || e.Code != "SlowDownWrite" {
|
||||
t.Errorf("incorrect error identity: %+v", e)
|
||||
}
|
||||
delete(versions, e.Key)
|
||||
}
|
||||
if len(versions) != 0 {
|
||||
t.Errorf("missing errors: %v", versions)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolsBulkDeleteNullBeforeMarker(t *testing.T) {
|
||||
for _, n := range []int{1, 2, 3} {
|
||||
t.Run(fmt.Sprintf("pools=%d", n), func(t *testing.T) {
|
||||
z, bucket := bulkDeletePools(t, n)
|
||||
object := "suspended"
|
||||
history := putConsistencyObject(t, z, bucket, object, 0, "history", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)})
|
||||
for p := range n {
|
||||
putConsistencyObject(t, z, bucket, object, p, "null", ObjectOptions{MTime: UTCNow().Add(-time.Duration(p+1) * time.Minute)})
|
||||
}
|
||||
objects := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: object}}, {ObjectV: ObjectV{ObjectName: object, VersionID: nullVersionID}}}
|
||||
deleted, errs := z.DeleteObjects(t.Context(), bucket, objects, ObjectOptions{VersionSuspended: true})
|
||||
if errs[0] != nil || errs[1] != nil || !deleted[0].DeleteMarker || deleted[1].VersionID != nullVersionID {
|
||||
t.Fatalf("results=%+v errors=%v", deleted, errs)
|
||||
}
|
||||
info, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{})
|
||||
if !info.DeleteMarker || !isErrObjectNotFound(err) {
|
||||
t.Errorf("replacement null marker lost: %+v, %v", info, err)
|
||||
}
|
||||
if body := bulkDeleteBody(t, z, bucket, object, history.VersionID); body != "history" {
|
||||
t.Errorf("UUID history changed: %q", body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolsBulkDeletePartialCleanupRetry(t *testing.T) {
|
||||
for _, kind := range []string{"unversioned", "null", "uuid"} {
|
||||
t.Run(kind, func(t *testing.T) {
|
||||
z, bucket := bulkDeletePools(t, 2)
|
||||
object := "partial-cleanup"
|
||||
opts := ObjectOptions{MTime: UTCNow().Add(-time.Hour)}
|
||||
oldBody := "old"
|
||||
if kind == "uuid" {
|
||||
opts.Versioned, opts.VersionID, oldBody = true, mustGetUUID(), "latest"
|
||||
}
|
||||
putConsistencyObject(t, z, bucket, object, 0, oldBody, opts)
|
||||
opts.MTime = UTCNow()
|
||||
putConsistencyObject(t, z, bucket, object, 1, "latest", opts)
|
||||
version := opts.VersionID
|
||||
if kind == "null" {
|
||||
version = nullVersionID
|
||||
}
|
||||
set := z.serverPools[0].getHashedSet(object)
|
||||
original := set.getDisks
|
||||
disks := append([]StorageAPI(nil), original()...)
|
||||
var failed, attempted atomic.Int32
|
||||
for i := range 2 {
|
||||
disks[i] = bulkDeleteFaultDisk{StorageAPI: disks[i], bucket: bucket, object: object, mode: "eio", fail: true, failed: &failed, attempted: &attempted}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
defer func() { set.getDisks = original }()
|
||||
objects := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: object, VersionID: version}}}
|
||||
_, errs := z.DeleteObjects(t.Context(), bucket, objects, ObjectOptions{})
|
||||
if errs[0] == nil || failed.Load() != 2 {
|
||||
t.Fatalf("did not partially fail cleanup: %v, failures=%d", errs, failed.Load())
|
||||
}
|
||||
set.getDisks = original
|
||||
if got := bulkDeleteBody(t, z, bucket, object, version); got != "latest" {
|
||||
t.Fatalf("latest lost after partial cleanup: %q", got)
|
||||
}
|
||||
_, errs = z.DeleteObjects(t.Context(), bucket, objects, ObjectOptions{})
|
||||
if errs[0] != nil {
|
||||
t.Fatalf("recovered retry must combine successful removals and already-absent replies: %v", errs[0])
|
||||
}
|
||||
_, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: version})
|
||||
if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||
t.Errorf("retry retained object: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolsBulkDeleteDifferentPrimaries(t *testing.T) {
|
||||
for _, n := range []int{2, 3} {
|
||||
t.Run(fmt.Sprintf("pools=%d", n), func(t *testing.T) {
|
||||
z, bucket := bulkDeletePools(t, n)
|
||||
objects := make([]ObjectToDelete, n)
|
||||
for primary := range n {
|
||||
object := fmt.Sprintf("key-%d", primary)
|
||||
objects[primary].ObjectName = object
|
||||
putConsistencyObject(t, z, bucket, object, (primary+1)%n, "old", ObjectOptions{MTime: UTCNow().Add(-time.Hour)})
|
||||
putConsistencyObject(t, z, bucket, object, primary, "latest", ObjectOptions{})
|
||||
}
|
||||
set := z.serverPools[0].getHashedSet(objects[n-1].ObjectName)
|
||||
original := set.getDisks
|
||||
disks := append([]StorageAPI(nil), original()...)
|
||||
var failed, attempted atomic.Int32
|
||||
for i, disk := range disks {
|
||||
disks[i] = bulkDeleteFaultDisk{StorageAPI: disk, bucket: bucket, object: objects[n-1].ObjectName, mode: "eio", fail: true, failed: &failed, attempted: &attempted}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
defer func() { set.getDisks = original }()
|
||||
deleted, errs := z.DeleteObjects(t.Context(), bucket, objects, ObjectOptions{})
|
||||
for i := range objects {
|
||||
if deleted[i].ObjectName != objects[i].ObjectName {
|
||||
t.Errorf("wrong result index %d: %+v", i, deleted[i])
|
||||
}
|
||||
if i == n-1 {
|
||||
if errs[i] == nil || bulkDeleteBody(t, z, bucket, objects[i].ObjectName, "") != "latest" {
|
||||
t.Errorf("failed item lost latest: %v", errs[i])
|
||||
}
|
||||
continue
|
||||
}
|
||||
_, err := z.GetObjectInfo(t.Context(), bucket, objects[i].ObjectName, ObjectOptions{})
|
||||
if errs[i] != nil || !isErrObjectNotFound(err) {
|
||||
t.Errorf("healthy item %d: delete=%v GET=%v", i, errs[i], err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolsBulkDeleteTieAndUnreadable(t *testing.T) {
|
||||
for _, unreadable := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("unreadable=%t", unreadable), func(t *testing.T) {
|
||||
z, bucket := bulkDeletePools(t, 3)
|
||||
object := "same-time"
|
||||
stamp := UTCNow().Add(-time.Hour)
|
||||
for p := range 3 {
|
||||
putConsistencyObject(t, z, bucket, object, p, fmt.Sprintf("pool-%d", p), ObjectOptions{MTime: stamp})
|
||||
}
|
||||
if got := bulkDeleteBody(t, z, bucket, object, ""); got != "pool-0" {
|
||||
t.Fatalf("unexpected read winner: %q", got)
|
||||
}
|
||||
set := z.serverPools[2].getHashedSet(object)
|
||||
original := set.getDisks
|
||||
disks := append([]StorageAPI(nil), original()...)
|
||||
var failed, attempted atomic.Int32
|
||||
for i, disk := range disks {
|
||||
if unreadable {
|
||||
disks[i] = consistencyReadFaultDisk{StorageAPI: disk, bucket: bucket, object: object}
|
||||
} else {
|
||||
disks[i] = bulkDeleteFaultDisk{StorageAPI: disk, bucket: bucket, object: object, mode: "eio", fail: true, failed: &failed, attempted: &attempted}
|
||||
}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
defer func() { set.getDisks = original }()
|
||||
_, errs := z.DeleteObjects(t.Context(), bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: object}}}, ObjectOptions{})
|
||||
set.getDisks = original
|
||||
if errs[0] == nil || bulkDeleteBody(t, z, bucket, object, "") != "pool-0" {
|
||||
t.Errorf("lost tie winner or unreadable-pool error: %v", errs[0])
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type bulkDeletePauseDisk struct {
|
||||
StorageAPI
|
||||
bucket, object string
|
||||
entered chan<- struct{}
|
||||
release <-chan struct{}
|
||||
}
|
||||
|
||||
func (d bulkDeletePauseDisk) DeleteVersions(ctx context.Context, bucket string, versions []FileInfoVersions, opts DeleteOptions) []error {
|
||||
for _, v := range versions {
|
||||
if bucket != d.bucket || v.Name != d.object {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case d.entered <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case <-d.release:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
errs := make([]error, len(versions))
|
||||
for i := range errs {
|
||||
errs[i] = syscall.EIO
|
||||
}
|
||||
return errs
|
||||
}
|
||||
return d.StorageAPI.DeleteVersions(ctx, bucket, versions, opts)
|
||||
}
|
||||
|
||||
func TestPoolsBulkDeleteLockAndCancellation(t *testing.T) {
|
||||
for _, canceled := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("cancel=%t", canceled), func(t *testing.T) {
|
||||
z, bucket := bulkDeletePools(t, 2)
|
||||
object := "locked"
|
||||
putConsistencyObject(t, z, bucket, object, 0, "old", ObjectOptions{MTime: UTCNow().Add(-time.Hour)})
|
||||
putConsistencyObject(t, z, bucket, object, 1, "latest", ObjectOptions{})
|
||||
set := z.serverPools[0].getHashedSet(object)
|
||||
original := set.getDisks
|
||||
disks := append([]StorageAPI(nil), original()...)
|
||||
entered, release := make(chan struct{}, 4), make(chan struct{})
|
||||
for i, disk := range disks {
|
||||
disks[i] = bulkDeletePauseDisk{StorageAPI: disk, bucket: bucket, object: object, entered: entered, release: release}
|
||||
}
|
||||
set.getDisks = func() []StorageAPI { return disks }
|
||||
defer func() { set.getDisks = original }()
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, errs := z.DeleteObjects(ctx, bucket, []ObjectToDelete{{ObjectV: ObjectV{ObjectName: object}}}, ObjectOptions{})
|
||||
done <- errs[0]
|
||||
}()
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("old-copy cleanup did not start")
|
||||
}
|
||||
if canceled {
|
||||
cancel()
|
||||
if err := <-done; err == nil {
|
||||
t.Fatal("canceled cleanup succeeded")
|
||||
}
|
||||
set.getDisks = original
|
||||
if got := bulkDeleteBody(t, z, bucket, object, ""); got != "latest" {
|
||||
t.Errorf("cancellation lost latest: %q", got)
|
||||
}
|
||||
return
|
||||
}
|
||||
putDone := make(chan error, 1)
|
||||
reader := mustGetPutObjReader(t, bytes.NewBufferString("after-put"), 9, "", "")
|
||||
go func() {
|
||||
_, err := z.PutObject(t.Context(), bucket, object, reader, ObjectOptions{})
|
||||
putDone <- err
|
||||
}()
|
||||
select {
|
||||
case err := <-putDone:
|
||||
t.Fatalf("PUT passed the held DELETE lock: %v", err)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
close(release)
|
||||
if err := <-done; err == nil {
|
||||
t.Error("cleanup failure was lost")
|
||||
}
|
||||
if err := <-putDone; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := bulkDeleteBody(t, z, bucket, object, ""); got != "after-put" {
|
||||
t.Errorf("concurrent PUT lost: %q", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type bulkDeleteReplyDisk struct {
|
||||
StorageAPI
|
||||
err error
|
||||
}
|
||||
|
||||
func (d bulkDeleteReplyDisk) IsOnline() bool { return true }
|
||||
|
||||
func (d bulkDeleteReplyDisk) DeleteVersions(_ context.Context, _ string, versions []FileInfoVersions, _ DeleteOptions) []error {
|
||||
errs := make([]error, len(versions))
|
||||
for i := range errs {
|
||||
errs[i] = d.err
|
||||
}
|
||||
return errs
|
||||
}
|
||||
|
||||
func TestBulkDeleteQuorumEvidence(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
obj ObjectToDelete
|
||||
opts ObjectOptions
|
||||
ok bool
|
||||
}{
|
||||
{name: "unversioned", ok: true},
|
||||
{name: "null", obj: ObjectToDelete{ObjectV: ObjectV{VersionID: nullVersionID}}, ok: true},
|
||||
{name: "uuid", obj: ObjectToDelete{ObjectV: ObjectV{VersionID: mustGetUUID()}}, ok: true},
|
||||
{name: "marker", opts: ObjectOptions{Versioned: true}},
|
||||
{name: "suspended-marker", opts: ObjectOptions{VersionSuspended: true}},
|
||||
{name: "pending-purge", obj: ObjectToDelete{ObjectV: ObjectV{VersionID: mustGetUUID()}, VersionPurgeStatuses: "target=PENDING;"}},
|
||||
{name: "replicated-marker", obj: ObjectToDelete{DeleteMarkerReplicationStatus: "target=PENDING;"}, opts: ObjectOptions{Versioned: true}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
disks := []StorageAPI{
|
||||
bulkDeleteReplyDisk{},
|
||||
bulkDeleteReplyDisk{},
|
||||
bulkDeleteReplyDisk{err: errFileNotFound},
|
||||
bulkDeleteReplyDisk{err: errFileVersionNotFound},
|
||||
}
|
||||
er := erasureObjects{getDisks: func() []StorageAPI { return disks }}
|
||||
tc.obj.ObjectName, tc.opts.NoAuditLog = "object", true
|
||||
_, errs := er.DeleteObjects(t.Context(), "bucket", []ObjectToDelete{tc.obj}, tc.opts)
|
||||
if (errs[0] == nil) != tc.ok {
|
||||
t.Errorf("two writes + two absent replies: error=%v, want success=%t", errs[0], tc.ok)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+145
-40
@@ -39,6 +39,7 @@ import (
|
||||
"github.com/minio/minio-go/v7/pkg/set"
|
||||
"github.com/minio/minio-go/v7/pkg/tags"
|
||||
"github.com/minio/minio/internal/bpool"
|
||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||
"github.com/minio/minio/internal/cachevalue"
|
||||
"github.com/minio/minio/internal/config/storageclass"
|
||||
xioutil "github.com/minio/minio/internal/ioutil"
|
||||
@@ -1355,8 +1356,13 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
|
||||
return pinfo.ObjInfo, nil
|
||||
}
|
||||
|
||||
// Delete concurrently in all server pools with read quorum error for unversioned objects.
|
||||
// Retire older copies before the newest copy for unversioned objects.
|
||||
if len(noReadQuorumPools) > 0 && !opts.Versioned && !opts.VersionSuspended {
|
||||
if pinfo.ObjInfo.TransitionedObject.Status == lifecycle.TransitionComplete {
|
||||
// Restored and remote copies can share a tier object. Reuse the
|
||||
// reconciliation which keeps that reference out of premature GC.
|
||||
return z.deleteObjectReconciled(ctx, bucket, object, opts)
|
||||
}
|
||||
return z.deleteObjectFromAllPools(ctx, bucket, object, opts, noReadQuorumPools)
|
||||
}
|
||||
|
||||
@@ -1367,6 +1373,21 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
|
||||
return objInfo, err
|
||||
}
|
||||
|
||||
// A delete marker hides the object in every pool. When more than one pool
|
||||
// holds the object, put the marker into the first one with a readable
|
||||
// version whose set can take the write, so that a set without quorum does
|
||||
// not fail the delete while another pool could take the marker.
|
||||
if opts.VersionID == "" && len(noReadQuorumPools) > 1 {
|
||||
for idx, pool := range z.serverPools {
|
||||
if !poolReadable(noReadQuorumPools, idx) || deleteQuorumMissing(pool, object) {
|
||||
continue
|
||||
}
|
||||
objInfo, err = pool.DeleteObject(ctx, bucket, object, opts)
|
||||
objInfo.Name = decodeDirObject(object)
|
||||
return objInfo, err
|
||||
}
|
||||
}
|
||||
|
||||
for _, pool := range z.serverPools {
|
||||
objInfo, err := pool.DeleteObject(ctx, bucket, object, opts)
|
||||
if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
|
||||
@@ -1387,36 +1408,76 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
|
||||
}
|
||||
|
||||
func (z *erasureServerPools) deleteObjectFromAllPools(ctx context.Context, bucket string, object string, opts ObjectOptions, poolIndices []poolErrs) (objInfo ObjectInfo, err error) {
|
||||
derrs := make([]error, len(poolIndices))
|
||||
dobjects := make([]ObjectInfo, len(poolIndices))
|
||||
|
||||
// Delete concurrently in all server pools that reported no error or read quorum error
|
||||
// where the read quorum issue is from metadata inconsistency.
|
||||
var wg sync.WaitGroup
|
||||
for idx, pe := range poolIndices {
|
||||
if v, ok := pe.Err.(InsufficientReadQuorum); ok && v.Type != RQInconsistentMeta {
|
||||
derrs[idx] = InsufficientWriteQuorum{}
|
||||
continue
|
||||
}
|
||||
wg.Add(1)
|
||||
pool := z.serverPools[pe.Index]
|
||||
go func(idx int, pool *erasureSets) {
|
||||
defer wg.Done()
|
||||
dobjects[idx], derrs[idx] = pool.DeleteObject(ctx, bucket, object, opts)
|
||||
}(idx, pool)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
// the poolIndices array is pre-sorted in order of latest ModTime, we care only about pool with latest object though
|
||||
// the delete call tries to clean up other pools during DeleteObject call.
|
||||
objInfo = dobjects[0]
|
||||
objInfo.Name = decodeDirObject(object)
|
||||
for _, derr := range derrs {
|
||||
|
||||
// Delete in all server pools that reported no error or read quorum error
|
||||
// where the read quorum issue is from metadata inconsistency. The
|
||||
// poolIndices array is pre-sorted in order of latest ModTime. Retire the
|
||||
// older copies first and stop at the first failure, as
|
||||
// deleteObjectReconciled does: a failed delete then keeps the latest copy
|
||||
// in place, instead of serving an older write of the object.
|
||||
for idx := len(poolIndices) - 1; idx >= 0; idx-- {
|
||||
pe := poolIndices[idx]
|
||||
if v, ok := pe.Err.(InsufficientReadQuorum); ok && v.Type != RQInconsistentMeta {
|
||||
return objInfo, InsufficientWriteQuorum{}
|
||||
}
|
||||
dobj, derr := z.serverPools[pe.Index].DeleteObject(ctx, bucket, object, opts)
|
||||
if idx == 0 {
|
||||
objInfo, err = dobj, derr
|
||||
break
|
||||
}
|
||||
if derr != nil && !isErrObjectNotFound(derr) && !isErrVersionNotFound(derr) {
|
||||
return objInfo, derr
|
||||
dobj.Name = decodeDirObject(object)
|
||||
return dobj, derr
|
||||
}
|
||||
}
|
||||
return objInfo, derrs[0]
|
||||
objInfo.Name = decodeDirObject(object)
|
||||
return objInfo, err
|
||||
}
|
||||
|
||||
// deleteQuorumMissing reports whether the set of object in pool has fewer
|
||||
// online drives than a delete needs (N/2+1, as in erasureObjects.DeleteObjects).
|
||||
func deleteQuorumMissing(pool *erasureSets, object string) bool {
|
||||
disks := pool.getHashedSet(object).getDisks()
|
||||
return countOnlineDisks(disks) < len(disks)/2+1
|
||||
}
|
||||
|
||||
// poolReadable reports whether pools lists pool idx with a readable version.
|
||||
func poolReadable(pools []poolErrs, idx int) bool {
|
||||
for _, pe := range pools {
|
||||
if pe.Index == idx {
|
||||
return pe.Err == nil
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// copyWithoutDeleteQuorum avoids starting a known unwritable cleanup. This
|
||||
// reduces partial deletion, but successful cleanup must still precede removing
|
||||
// the authoritative copy: drives can fail after this preflight.
|
||||
func (z *erasureServerPools) copyWithoutDeleteQuorum(ctx context.Context, bucket, object, versionID string) bool {
|
||||
for _, pool := range z.serverPools {
|
||||
if deleteQuorumMissing(pool, object) {
|
||||
_, err := pool.GetObjectInfo(ctx, bucket, object, ObjectOptions{VersionID: versionID, NoLock: true})
|
||||
if err == nil || isErrReadQuorum(err) || isErrMethodNotAllowed(err) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// deletesData reports whether deleting obj removes an object or a version, as
|
||||
// opposed to adding a delete marker (see erasureObjects.DeleteObjects).
|
||||
func deletesData(obj ObjectToDelete, opts ObjectOptions) bool {
|
||||
if obj.VersionID != "" {
|
||||
return true
|
||||
}
|
||||
versioned := opts.Versioned
|
||||
if opts.PrefixEnabledFn != nil {
|
||||
versioned = opts.PrefixEnabledFn(obj.ObjectName)
|
||||
}
|
||||
return !versioned && !opts.VersionSuspended
|
||||
}
|
||||
|
||||
func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) {
|
||||
@@ -1425,6 +1486,11 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
|
||||
objSets := set.NewStringSet()
|
||||
for i := range derrs {
|
||||
objects[i].ObjectName = encodeDirObject(objects[i].ObjectName)
|
||||
dobjects[i] = DeletedObject{
|
||||
ObjectName: decodeDirObject(objects[i].ObjectName),
|
||||
VersionID: objects[i].VersionID,
|
||||
ReplicationState: objects[i].ReplicationState(),
|
||||
}
|
||||
|
||||
derrs[i] = checkDelObjArgs(ctx, bucket, objects[i].ObjectName)
|
||||
objSets.Add(objects[i].ObjectName)
|
||||
@@ -1442,25 +1508,64 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
|
||||
ctx = lkctx.Context()
|
||||
defer multiDeleteLock.Unlock(lkctx)
|
||||
|
||||
// Removing data needs the same authoritative-copy protection as a single
|
||||
// DELETE. An online-drive preflight cannot protect against later I/O errors.
|
||||
// Keep marker creation after data removal, including when the same request
|
||||
// explicitly removes a null version and creates its replacement marker.
|
||||
batch, batchIdx := objects, []int(nil)
|
||||
if !z.SinglePool() {
|
||||
batch = make([]ObjectToDelete, 0, len(objects))
|
||||
batchIdx = make([]int, 0, len(objects))
|
||||
data, dataIdx := make([]ObjectToDelete, 0, len(objects)), make([]int, 0, len(objects))
|
||||
for i, obj := range objects {
|
||||
if derrs[i] != nil {
|
||||
continue
|
||||
}
|
||||
if deletesData(obj, opts) {
|
||||
if z.copyWithoutDeleteQuorum(ctx, bucket, obj.ObjectName, obj.VersionID) {
|
||||
derrs[i] = InsufficientWriteQuorum{}
|
||||
continue
|
||||
}
|
||||
data = append(data, obj)
|
||||
dataIdx = append(dataIdx, i)
|
||||
continue
|
||||
}
|
||||
batch = append(batch, obj)
|
||||
batchIdx = append(batchIdx, i)
|
||||
}
|
||||
if len(data) > 0 {
|
||||
deleted, errs := z.deleteObjectsReconciled(ctx, bucket, data, opts)
|
||||
for j, i := range dataIdx {
|
||||
dobjects[i], derrs[i] = deleted[j], errs[j]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
dObjectsByPool := make([][]DeletedObject, len(z.serverPools))
|
||||
dErrsByPool := make([][]error, len(z.serverPools))
|
||||
|
||||
eg := errgroup.WithNErrs(len(z.serverPools)).WithConcurrency(len(z.serverPools))
|
||||
for i, pool := range z.serverPools {
|
||||
pool := pool
|
||||
eg.Go(func() error {
|
||||
dObjectsByPool[i], dErrsByPool[i] = pool.DeleteObjects(ctx, bucket, objects, opts)
|
||||
return nil
|
||||
}, i)
|
||||
if len(batch) > 0 {
|
||||
eg := errgroup.WithNErrs(len(z.serverPools)).WithConcurrency(len(z.serverPools))
|
||||
for i, pool := range z.serverPools {
|
||||
pool := pool
|
||||
eg.Go(func() error {
|
||||
dObjectsByPool[i], dErrsByPool[i] = pool.DeleteObjects(ctx, bucket, batch, opts)
|
||||
return nil
|
||||
}, i)
|
||||
}
|
||||
eg.Wait() // wait to check all the pools.
|
||||
}
|
||||
eg.Wait() // wait to check all the pools.
|
||||
|
||||
for i := range dobjects {
|
||||
for j := range batch {
|
||||
i := j
|
||||
if batchIdx != nil {
|
||||
i = batchIdx[j]
|
||||
}
|
||||
// Iterate over pools
|
||||
for pool := range z.serverPools {
|
||||
if dErrsByPool[pool][i] == nil && dObjectsByPool[pool][i].found {
|
||||
if dErrsByPool[pool][j] == nil && dObjectsByPool[pool][j].found {
|
||||
// A fast exit when the object is found and removed
|
||||
dobjects[i] = dObjectsByPool[pool][i]
|
||||
dobjects[i] = dObjectsByPool[pool][j]
|
||||
derrs[i] = nil
|
||||
break
|
||||
}
|
||||
@@ -1468,8 +1573,8 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
|
||||
// No error related to this object is found, assign this pool result
|
||||
// whether it is nil because there is no object found or because of
|
||||
// some other errors such erasure quorum errors.
|
||||
dobjects[i] = dObjectsByPool[pool][i]
|
||||
derrs[i] = dErrsByPool[pool][i]
|
||||
dobjects[i] = dObjectsByPool[pool][j]
|
||||
derrs[i] = dErrsByPool[pool][j]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user