From bf9e938a3b28291991ef0aa87526d23533a4e399 Mon Sep 17 00:00:00 2001
From: Alex Vanin <alexey@nspcc.ru>
Date: Thu, 3 Dec 2020 12:31:51 +0300
Subject: [PATCH] [#230] metabase: Return SplitInfoError in Exist method

Signed-off-by: Alex Vanin <alexey@nspcc.ru>
---
 .../metabase/v2/exists.go                     | 19 ++++++++++++++++++-
 .../metabase/v2/exists_test.go                |  8 +++++---
 pkg/local_object_storage/metabase/v2/put.go   |  2 +-
 .../metabase/v2/select.go                     |  2 +-
 pkg/local_object_storage/metabase/v2/util.go  |  2 ++
 5 files changed, 27 insertions(+), 6 deletions(-)

diff --git a/pkg/local_object_storage/metabase/v2/exists.go b/pkg/local_object_storage/metabase/v2/exists.go
index 34b73e26a..94ae99ec6 100644
--- a/pkg/local_object_storage/metabase/v2/exists.go
+++ b/pkg/local_object_storage/metabase/v2/exists.go
@@ -1,11 +1,16 @@
 package meta
 
 import (
+	"errors"
+	"fmt"
+
 	objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object"
 	"github.com/nspcc-dev/neofs-node/pkg/core/object"
 	"go.etcd.io/bbolt"
 )
 
+var ErrLackSplitInfo = errors.New("no split info on parent object")
+
 // Exists returns ErrAlreadyRemoved if addr was marked as removed. Otherwise it
 // returns true if addr is in primary index or false if it is not.
 func (db *DB) Exists(addr *objectSDK.Address) (exists bool, err error) {
@@ -33,7 +38,19 @@ func (db *DB) exists(tx *bbolt.Tx, addr *objectSDK.Address) (exists bool, err er
 
 	// if primary bucket is empty, then check if object exists in parent bucket
 	if inBucket(tx, parentBucketName(addr.ContainerID()), objKey) {
-		return true, nil
+		rawSplitInfo := getFromBucket(tx, rootBucketName(addr.ContainerID()), objKey)
+		if len(rawSplitInfo) == 0 {
+			return false, ErrLackSplitInfo
+		}
+
+		splitInfo := objectSDK.NewSplitInfo()
+
+		err := splitInfo.Unmarshal(rawSplitInfo)
+		if err != nil {
+			return false, fmt.Errorf("can't unmarshal split info from root index: %w", err)
+		}
+
+		return false, objectSDK.NewSplitInfoError(splitInfo)
 	}
 
 	// if parent bucket is empty, then check if object exists in tombstone bucket
diff --git a/pkg/local_object_storage/metabase/v2/exists_test.go b/pkg/local_object_storage/metabase/v2/exists_test.go
index 011a74f89..68220e0bf 100644
--- a/pkg/local_object_storage/metabase/v2/exists_test.go
+++ b/pkg/local_object_storage/metabase/v2/exists_test.go
@@ -1,6 +1,7 @@
 package meta_test
 
 import (
+	"errors"
 	"testing"
 
 	objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object"
@@ -63,8 +64,9 @@ func TestDB_Exists(t *testing.T) {
 		err := db.Put(child.Object(), nil)
 		require.NoError(t, err)
 
-		exists, err := db.Exists(parent.Object().Address())
-		require.NoError(t, err)
-		require.True(t, exists)
+		_, err = db.Exists(parent.Object().Address())
+
+		var expectedErr *objectSDK.SplitInfoError
+		require.True(t, errors.As(err, &expectedErr))
 	})
 }
diff --git a/pkg/local_object_storage/metabase/v2/put.go b/pkg/local_object_storage/metabase/v2/put.go
index d579e3f20..5a079306a 100644
--- a/pkg/local_object_storage/metabase/v2/put.go
+++ b/pkg/local_object_storage/metabase/v2/put.go
@@ -37,7 +37,7 @@ func (db *DB) put(tx *bbolt.Tx, obj *object.Object, id *blobovnicza.ID, si *obje
 	isParent := si != nil
 
 	exists, err := db.exists(tx, obj.Address())
-	if err != nil {
+	if err != nil && !errors.As(err, &splitInfoError) {
 		return err
 	}
 
diff --git a/pkg/local_object_storage/metabase/v2/select.go b/pkg/local_object_storage/metabase/v2/select.go
index 9649c38b4..1bb2f0a12 100644
--- a/pkg/local_object_storage/metabase/v2/select.go
+++ b/pkg/local_object_storage/metabase/v2/select.go
@@ -289,7 +289,7 @@ func (db *DB) selectObjectID(
 	}
 
 	ok, err := db.exists(tx, addr)
-	if err == nil && ok {
+	if (err == nil && ok) || errors.As(err, &splitInfoError) {
 		markAddressInCache(to, fNum, addrStr)
 	}
 }
diff --git a/pkg/local_object_storage/metabase/v2/util.go b/pkg/local_object_storage/metabase/v2/util.go
index f24929e01..46cb96771 100644
--- a/pkg/local_object_storage/metabase/v2/util.go
+++ b/pkg/local_object_storage/metabase/v2/util.go
@@ -30,6 +30,8 @@ var (
 	splitPostfix        = "_splitid"
 
 	userAttributePostfix = "_attr_"
+
+	splitInfoError *object.SplitInfoError // for errors.As comparisons
 )
 
 // primaryBucketName returns <CID>.