From 881f98e511ab04bb62efb5c283139e2542483e2e Mon Sep 17 00:00:00 2001
From: Harshavardhana <harsha@minio.io>
Date: Wed, 10 Feb 2021 14:25:43 -0800
Subject: [PATCH] fix: use getPoolIdx in DeleteObjects() (#11513)

filter out relevant objects for each pool to
avoid calling, further delete operations on
subsequent pools where some of these objects
might not exist.

This is mainly useful to avoid situations
during bi-directional bucket replication.
---
 cmd/erasure-server-pool.go | 27 +++++++++++++++++++++++----
 1 file changed, 23 insertions(+), 4 deletions(-)

diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index 0595de33b..b91514d4c 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -639,6 +639,23 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
 		objSets.Add(objects[i].ObjectName)
 	}
 
+	poolObjIdxMap := map[int][]ObjectToDelete{}
+	origIndexMap := map[int][]int{}
+	if !z.SinglePool() {
+		for j, obj := range objects {
+			idx, err := z.getPoolIdx(ctx, bucket, obj.ObjectName, 1<<30)
+			if err != nil {
+				// Unhandled errors return right here.
+				for i := range derrs {
+					derrs[i] = err
+				}
+				return dobjects, derrs
+			}
+			poolObjIdxMap[idx] = append(poolObjIdxMap[idx], obj)
+			origIndexMap[idx] = append(origIndexMap[idx], j)
+		}
+	}
+
 	// Acquire a bulk write lock across 'objects'
 	multiDeleteLock := z.NewNSLock(bucket, objSets.ToSlice()...)
 	if err := multiDeleteLock.GetLock(ctx, globalOperationTimeout); err != nil {
@@ -653,13 +670,15 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
 		return z.serverPools[0].DeleteObjects(ctx, bucket, objects, opts)
 	}
 
-	for _, pool := range z.serverPools {
-		deletedObjects, errs := pool.DeleteObjects(ctx, bucket, objects, opts)
+	for idx, pool := range z.serverPools {
+		objs := poolObjIdxMap[idx]
+		orgIndexes := origIndexMap[idx]
+		deletedObjects, errs := pool.DeleteObjects(ctx, bucket, objs, opts)
 		for i, derr := range errs {
 			if derr != nil {
-				derrs[i] = derr
+				derrs[orgIndexes[i]] = derr
 			}
-			dobjects[i] = deletedObjects[i]
+			dobjects[orgIndexes[i]] = deletedObjects[i]
 		}
 	}
 	return dobjects, derrs