From a2c2241356418d877c041eb97c05dd872955dfc3 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Thu, 11 Feb 2021 18:43:50 +0300
Subject: [PATCH] [#379] storage engine: Inhume object in one shard

In previous implementation StorageEngine.Inhume operation forced Shard
.Inhume call on all internal shards. There is a need to inhume object in a
single shard. To achieve this, Inhume operation is performed in next steps:

 1. iterate over sorted shards, check object presence through Exists call;
 2. if object exists at any shard in step 1 => inhume it and return on
    success;
 3. if no shards contain the object => iterate over sorted shards again and
    try to inhume the object at first possible shard;
 4. if all Inhume calls are failed => return an error.

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

diff --git a/pkg/local_object_storage/engine/inhume.go b/pkg/local_object_storage/engine/inhume.go
index 94bf1537b..24fca3b39 100644
--- a/pkg/local_object_storage/engine/inhume.go
+++ b/pkg/local_object_storage/engine/inhume.go
@@ -1,6 +1,8 @@
 package engine
 
 import (
+	"errors"
+
 	objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
 	"go.uber.org/zap"
@@ -25,23 +27,56 @@ func (p *InhumePrm) WithTarget(addr, tombstone *objectSDK.Address) *InhumePrm {
 	return p
 }
 
+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.addr, prm.tombstone)
 
-	e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) {
-		_, err := sh.Inhume(shPrm)
+	res := e.inhume(prm.addr, shPrm, true)
+	if res == nil {
+		res = e.inhume(prm.addr, shPrm, false)
+		if res == nil {
+			return nil, errInhumeFailure
+		}
+	}
+
+	return res, nil
+}
+
+func (e *StorageEngine) inhume(addr *objectSDK.Address, prm *shard.InhumePrm, checkExists bool) (res *InhumeRes) {
+	e.iterateOverSortedShards(addr, func(_ int, sh *shard.Shard) (stop bool) {
+		if checkExists {
+			exRes, err := sh.Exists(new(shard.ExistsPrm).
+				WithAddress(addr),
+			)
+			if err != nil {
+				// TODO: smth wrong with shard, need to be processed
+				e.log.Warn("could not check for presents in shard",
+					zap.Stringer("shard", sh.ID()),
+					zap.String("error", err.Error()),
+				)
+
+				return
+			} else if !exRes.Exists() {
+				return
+			}
+		}
+
+		_, err := sh.Inhume(prm)
 		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()),
 			)
+		} else {
+			res = new(InhumeRes)
 		}
 
-		return false
+		return err == nil
 	})
 
-	return nil, nil
+	return
 }