From 6ec7433e1421352a67422fa8f66c65ba44ca3cc5 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Mon, 15 Feb 2021 14:42:54 +0300
Subject: [PATCH] [#377] storage engine: Change the implementation of Delete
 operation

Make `StorageEngine.Delete` to execute `Inhume` operation with
`MarkAsGarbage` parameter on the `Shard` that holds the object. Searching of
the particular shard is performed through iterating over HRW-sorted shards.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 pkg/local_object_storage/engine/delete.go | 43 +++++++++++++++--------
 1 file changed, 29 insertions(+), 14 deletions(-)

diff --git a/pkg/local_object_storage/engine/delete.go b/pkg/local_object_storage/engine/delete.go
index 664c3242..3755baf2 100644
--- a/pkg/local_object_storage/engine/delete.go
+++ b/pkg/local_object_storage/engine/delete.go
@@ -25,23 +25,38 @@ func (p *DeletePrm) WithAddresses(addr ...*objectSDK.Address) *DeletePrm {
 	return p
 }
 
-// Delete removes objects from the shards.
+// Delete marks the objects to be removed.
 func (e *StorageEngine) Delete(prm *DeletePrm) (*DeleteRes, error) {
-	shPrm := new(shard.DeletePrm).
-		WithAddresses(prm.addr...)
+	shPrm := new(shard.InhumePrm)
+	existsPrm := new(shard.ExistsPrm)
 
-	e.iterateOverUnsortedShards(func(sh *shard.Shard) (stop bool) {
-		_, err := sh.Delete(shPrm)
-		if err != nil {
-			// TODO: smth wrong with shard, need to be processed
-			e.log.Warn("could not delete object from shard",
-				zap.Stringer("shard", sh.ID()),
-				zap.String("error", err.Error()),
-			)
-		}
+	for i := range prm.addr {
+		e.iterateOverSortedShards(prm.addr[i], func(_ int, sh *shard.Shard) (stop bool) {
+			resExists, err := sh.Exists(existsPrm.WithAddress(prm.addr[i]))
+			if err != nil {
+				// TODO: smth wrong with shard, need to be processed
+				e.log.Warn("could not check object existence",
+					zap.Stringer("shard", sh.ID()),
+					zap.String("error", err.Error()),
+				)
 
-		return false
-	})
+				return false
+			} else if !resExists.Exists() {
+				return false
+			}
+
+			_, err = sh.Inhume(shPrm.MarkAsGarbage(prm.addr[i]))
+			if err != nil {
+				// TODO: smth wrong with shard, need to be processed
+				e.log.Warn("could not inhume object in shard",
+					zap.Stringer("shard", sh.ID()),
+					zap.String("error", err.Error()),
+				)
+			}
+
+			return err == nil
+		})
+	}
 
 	return nil, nil
 }