diff --git a/cmd/erasure-server-pool-consistency_test.go b/cmd/erasure-server-pool-consistency_test.go index 29dea2dde..8b21dbc9e 100644 --- a/cmd/erasure-server-pool-consistency_test.go +++ b/cmd/erasure-server-pool-consistency_test.go @@ -1442,3 +1442,46 @@ func TestPoolsDeleteObjectsKeepsLatestWithoutDeleteQuorum(t *testing.T) { }) } } + +// 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) + } + } + }) + } +} diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index f4f580785..42211d96a 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1367,6 +1367,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) { @@ -1421,6 +1436,16 @@ func deleteQuorumMissing(pool *erasureSets, object string) bool { 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 reports whether a pool whose set cannot take a // delete holds a copy of object, or of its versionID when set. Only such sets // are read.