From 1b51dcf8a46a3be3cfbee60ca8c053bcfd08adfa Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Thu, 20 May 2021 17:31:47 +0300
Subject: [PATCH] [#546] engine/inhume: Fix incorrect Inhume behavior for root
 objects

If object to be inhumed is root we need to continue first traverse over the
shards. In case when several children are stored in different shards,
inhuming object in a single shard leads to appearance of inhumed object in
subsequent selections. Also, any object can be already inhumed, and this
case is equivalent to successful inhume.

Do not fail on `object.ErrAlreadyRemoved` error. Continue first iterating
over shards if we detected root object (`SplitInfoError`).

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

diff --git a/pkg/local_object_storage/engine/inhume.go b/pkg/local_object_storage/engine/inhume.go
index 37172ad0..806442a7 100644
--- a/pkg/local_object_storage/engine/inhume.go
+++ b/pkg/local_object_storage/engine/inhume.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 
 	objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object"
+	"github.com/nspcc-dev/neofs-node/pkg/core/object"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
 	"go.uber.org/zap"
 )
@@ -75,19 +76,40 @@ func (e *StorageEngine) Inhume(prm *InhumePrm) (*InhumeRes, error) {
 }
 
 func (e *StorageEngine) inhume(addr *objectSDK.Address, prm *shard.InhumePrm, checkExists bool) (ok bool) {
+	root := false
+
 	e.iterateOverSortedShards(addr, func(_ int, sh *shard.Shard) (stop bool) {
+		defer func() {
+			// if object is root we continue since information about it
+			// can be presented in other shards
+			if checkExists && root {
+				stop = false
+			}
+		}()
+
 		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()),
-				)
+				if errors.Is(err, object.ErrAlreadyRemoved) {
+					// inhumed once - no need to be inhumed again
+					ok = true
+					return true
+				}
 
-				return
+				var siErr *objectSDK.SplitInfoError
+				if !errors.As(err, &siErr) {
+					// 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
+				}
+
+				root = true
 			} else if !exRes.Exists() {
 				return
 			}