From 7a4e3efa956d21d31fab4c628a1fa1ce6aed11c8 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Fri, 19 Feb 2021 17:01:01 +0300
Subject: [PATCH] [#217] storage engine: Transform Inhume to batch operation

Make `StorageEngine.Inhume` to process list of objects per single operation.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 cmd/neofs-node/object.go                  |  2 +-
 pkg/local_object_storage/engine/inhume.go | 33 ++++++++++++++---------
 2 files changed, 21 insertions(+), 14 deletions(-)

diff --git a/cmd/neofs-node/object.go b/cmd/neofs-node/object.go
index 003db70f..ea4f578a 100644
--- a/cmd/neofs-node/object.go
+++ b/cmd/neofs-node/object.go
@@ -108,7 +108,7 @@ func (r *localObjectInhumer) DeleteObjects(ts *objectSDK.Address, addr ...*objec
 	prm := new(engine.InhumePrm)
 
 	for _, a := range addr {
-		prm.WithTarget(a, ts)
+		prm.WithTarget(ts, a)
 
 		if _, err := r.storage.Inhume(prm); err != nil {
 			r.log.Error("could not delete object",
diff --git a/pkg/local_object_storage/engine/inhume.go b/pkg/local_object_storage/engine/inhume.go
index b40154c3..a2dc130c 100644
--- a/pkg/local_object_storage/engine/inhume.go
+++ b/pkg/local_object_storage/engine/inhume.go
@@ -11,17 +11,20 @@ import (
 
 // InhumePrm encapsulates parameters for inhume operation.
 type InhumePrm struct {
-	addr, tombstone *objectSDK.Address
+	tombstone *objectSDK.Address
+	addrs     []*objectSDK.Address
 }
 
 // InhumeRes encapsulates results of inhume operation.
 type InhumeRes struct{}
 
-// WithTarget sets object address that should be inhumed and tombstone address
+// WithTarget sets list of objects that should be inhumed and tombstone address
 // as the reason for inhume operation.
-func (p *InhumePrm) WithTarget(addr, tombstone *objectSDK.Address) *InhumePrm {
+//
+// tombstone should not be nil, addr should not be empty.
+func (p *InhumePrm) WithTarget(tombstone *objectSDK.Address, addrs ...*objectSDK.Address) *InhumePrm {
 	if p != nil {
-		p.addr = addr
+		p.addrs = addrs
 		p.tombstone = tombstone
 	}
 
@@ -33,20 +36,24 @@ var errInhumeFailure = errors.New("inhume operation failed")
 // Inhume calls metabase. Inhume method to mark object as removed. It won't be
 // removed physically from shard until `Delete` operation.
 func (e *StorageEngine) Inhume(prm *InhumePrm) (*InhumeRes, error) {
-	shPrm := new(shard.InhumePrm).WithTarget(prm.tombstone, prm.addr)
+	shPrm := new(shard.InhumePrm)
 
-	res := e.inhume(prm.addr, shPrm, true)
-	if res == nil {
-		res = e.inhume(prm.addr, shPrm, false)
-		if res == nil {
-			return nil, errInhumeFailure
+	for i := range prm.addrs {
+		shPrm.WithTarget(prm.tombstone, prm.addrs[i])
+
+		ok := e.inhume(prm.addrs[i], shPrm, true)
+		if ok {
+			ok = e.inhume(prm.addrs[i], shPrm, false)
+			if ok {
+				return nil, errInhumeFailure
+			}
 		}
 	}
 
-	return res, nil
+	return new(InhumeRes), nil
 }
 
-func (e *StorageEngine) inhume(addr *objectSDK.Address, prm *shard.InhumePrm, checkExists bool) (res *InhumeRes) {
+func (e *StorageEngine) inhume(addr *objectSDK.Address, prm *shard.InhumePrm, checkExists bool) (ok bool) {
 	e.iterateOverSortedShards(addr, func(_ int, sh *shard.Shard) (stop bool) {
 		if checkExists {
 			exRes, err := sh.Exists(new(shard.ExistsPrm).
@@ -73,7 +80,7 @@ func (e *StorageEngine) inhume(addr *objectSDK.Address, prm *shard.InhumePrm, ch
 				zap.String("error", err.Error()),
 			)
 		} else {
-			res = new(InhumeRes)
+			ok = true
 		}
 
 		return err == nil