fix(pools): keep the latest copy when a delete fails in one pool

Signed-off-by: Denis Babchuk <babchukdenis@icloud.com>
2026-10-29 03:57:19 +03:00 · 2026-10-23 21:28:46 +03:00
parent b91e1c6f2a
commit 1841fce4c4
2 changed files with 241 additions and 37 deletions
@@ -1306,3 +1306,139 @@ 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)
+				}
+			}
+		})
+	}
+}
@@ -1387,36 +1387,67 @@ 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))
+	objInfo.Name = decodeDirObject(object)

-	// 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 {
+	// 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 {
-			derrs[idx] = InsufficientWriteQuorum{}
+			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) {
+			dobj.Name = decodeDirObject(object)
+			return dobj, derr
+		}
+	}
+	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
+}
+
+// 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.
+func (z *erasureServerPools) copyWithoutDeleteQuorum(ctx context.Context, bucket, object, versionID string) bool {
+	for _, pool := range z.serverPools {
+		if !deleteQuorumMissing(pool, object) {
 			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 {
-		if derr != nil && !isErrObjectNotFound(derr) && !isErrVersionNotFound(derr) {
-			return objInfo, derr
+		_, err := pool.GetObjectInfo(ctx, bucket, object, ObjectOptions{VersionID: versionID, NoLock: true})
+		if err == nil || isErrReadQuorum(err) || isErrMethodNotAllowed(err) {
+			return true
 		}
 	}
-	return objInfo, derrs[0]
+	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) {
@@ -1442,25 +1473,50 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
 	ctx = lkctx.Context()
 	defer multiDeleteLock.Unlock(lkctx)

+	// Pools delete the batch in parallel, so a pool that cannot take the
+	// delete would keep its copy while another pool removes a newer one.
+	// Leave an object or version untouched in every pool when a set without
+	// delete quorum holds a copy of it.
+	batch, batchIdx := objects, []int(nil)
+	if !z.SinglePool() {
+		batch = make([]ObjectToDelete, 0, len(objects))
+		batchIdx = make([]int, 0, len(objects))
+		for i, obj := range objects {
+			if derrs[i] == nil && deletesData(obj, opts) && z.copyWithoutDeleteQuorum(ctx, bucket, obj.ObjectName, obj.VersionID) {
+				dobjects[i] = DeletedObject{ObjectName: obj.ObjectName, VersionID: obj.VersionID}
+				derrs[i] = InsufficientWriteQuorum{}
+				continue
+			}
+			batch = append(batch, obj)
+			batchIdx = append(batchIdx, i)
+		}
+	}
+
 	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 +1524,20 @@ 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]
+			}
+		}
+		// A delete marker in one pool hides the object, but an object or
+		// version removed from one pool is still served from a pool that
+		// failed to remove its copy.
+		if deletesData(batch[j], opts) {
+			for pool := range z.serverPools {
+				if err := dErrsByPool[pool][j]; err != nil {
+					dobjects[i] = dObjectsByPool[pool][j]
+					derrs[i] = err
+					break
+				}
 			}
 		}
 	}