From 14329ab5650a380a11eb846b8ef677abcaa8e8b6 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Fri, 24 Sep 2021 18:18:04 +0300
Subject: [PATCH] [#840] metabase: Distinguish objects with tombstones and GC
 marks

Each object from graveyard has tombstone or GC mark. If object has
tombstone, metabase should return `ErrAlreadyRemoved` on object requests.
This is the case when user clearly removed the object from container. GC
marks are used for physical removal which can appear even if object is still
presented in container (Control service, Policer job, etc.). In this case
metabase should return 404 error on object requests.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 pkg/local_object_storage/metabase/exists.go   | 26 ++++++++++++++-----
 pkg/local_object_storage/metabase/get.go      |  9 +++++--
 pkg/local_object_storage/metabase/get_test.go | 14 ++++++++++
 pkg/local_object_storage/metabase/select.go   |  2 +-
 4 files changed, 42 insertions(+), 9 deletions(-)

diff --git a/pkg/local_object_storage/metabase/exists.go b/pkg/local_object_storage/metabase/exists.go
index a7b37662e9..d975b18b47 100644
--- a/pkg/local_object_storage/metabase/exists.go
+++ b/pkg/local_object_storage/metabase/exists.go
@@ -1,6 +1,7 @@
 package meta
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 
@@ -62,7 +63,10 @@ func (db *DB) Exists(prm *ExistsPrm) (res *ExistsRes, err error) {
 
 func (db *DB) exists(tx *bbolt.Tx, addr *objectSDK.Address) (exists bool, err error) {
 	// check graveyard first
-	if inGraveyard(tx, addr) {
+	switch inGraveyard(tx, addr) {
+	case 1:
+		return false, object.ErrNotFound
+	case 2:
 		return false, object.ErrAlreadyRemoved
 	}
 
@@ -92,16 +96,26 @@ func (db *DB) exists(tx *bbolt.Tx, addr *objectSDK.Address) (exists bool, err er
 	return inBucket(tx, storageGroupBucketName(addr.ContainerID()), objKey), nil
 }
 
-// inGraveyard returns true if object was marked as removed.
-func inGraveyard(tx *bbolt.Tx, addr *objectSDK.Address) bool {
+// inGraveyard returns:
+//  * 0 if object is not in graveyard;
+//  * 1 if object is in graveyard with GC mark;
+//  * 2 if object is in graveyard with tombstone.
+func inGraveyard(tx *bbolt.Tx, addr *objectSDK.Address) uint8 {
 	graveyard := tx.Bucket(graveyardBucketName)
 	if graveyard == nil {
-		return false
+		return 0
 	}
 
-	tombstone := graveyard.Get(addressKey(addr))
+	val := graveyard.Get(addressKey(addr))
+	if val == nil {
+		return 0
+	}
 
-	return len(tombstone) != 0
+	if bytes.Equal(val, []byte(inhumeGCMarkValue)) {
+		return 1
+	}
+
+	return 2
 }
 
 // inBucket checks if key <key> is present in bucket <name>.
diff --git a/pkg/local_object_storage/metabase/get.go b/pkg/local_object_storage/metabase/get.go
index c28ba2d406..cd98712fe4 100644
--- a/pkg/local_object_storage/metabase/get.go
+++ b/pkg/local_object_storage/metabase/get.go
@@ -85,8 +85,13 @@ func (db *DB) get(tx *bbolt.Tx, addr *objectSDK.Address, checkGraveyard, raw boo
 	key := objectKey(addr.ObjectID())
 	cid := addr.ContainerID()
 
-	if checkGraveyard && inGraveyard(tx, addr) {
-		return nil, object.ErrAlreadyRemoved
+	if checkGraveyard {
+		switch inGraveyard(tx, addr) {
+		case 1:
+			return nil, object.ErrNotFound
+		case 2:
+			return nil, object.ErrAlreadyRemoved
+		}
 	}
 
 	// check in primary index
diff --git a/pkg/local_object_storage/metabase/get_test.go b/pkg/local_object_storage/metabase/get_test.go
index 6db4b973cf..16929f53c3 100644
--- a/pkg/local_object_storage/metabase/get_test.go
+++ b/pkg/local_object_storage/metabase/get_test.go
@@ -94,6 +94,20 @@ func TestDB_Get(t *testing.T) {
 		require.NoError(t, err)
 		require.True(t, binaryEqual(child.CutPayload().Object(), newChild))
 	})
+
+	t.Run("get removed object", func(t *testing.T) {
+		obj := generateAddress()
+		ts := generateAddress()
+
+		require.NoError(t, meta.Inhume(db, obj, ts))
+		_, err := meta.Get(db, obj)
+		require.ErrorIs(t, err, object.ErrAlreadyRemoved)
+
+		obj = generateAddress()
+		require.NoError(t, meta.Inhume(db, obj, nil))
+		_, err = meta.Get(db, obj)
+		require.ErrorIs(t, err, object.ErrNotFound)
+	})
 }
 
 // binary equal is used when object contains empty lists in the structure and
diff --git a/pkg/local_object_storage/metabase/select.go b/pkg/local_object_storage/metabase/select.go
index 6325caed8a..b11e0d8e35 100644
--- a/pkg/local_object_storage/metabase/select.go
+++ b/pkg/local_object_storage/metabase/select.go
@@ -134,7 +134,7 @@ func (db *DB) selectObjects(tx *bbolt.Tx, cid *cid.ID, fs object.SearchFilters)
 			return nil, err
 		}
 
-		if inGraveyard(tx, addr) {
+		if inGraveyard(tx, addr) > 0 {
 			continue // ignore removed objects
 		}