From fe8076e60af22a5eaf3fc34d520fe5eb1eb1993f Mon Sep 17 00:00:00 2001
From: Pavel Karpy <carpawell@nspcc.ru>
Date: Thu, 8 Apr 2021 20:06:13 +0300
Subject: [PATCH] [#1318] metabase: Add immediate object deletion

Mark objects with GC immediately after a Tombstone is received.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
---
 .../metabase/graveyard_test.go                  | 17 ++++++++++++++---
 pkg/local_object_storage/metabase/inhume.go     | 15 ++++++++++-----
 .../metabase/inhume_test.go                     | 17 +++++++++--------
 3 files changed, 33 insertions(+), 16 deletions(-)

diff --git a/pkg/local_object_storage/metabase/graveyard_test.go b/pkg/local_object_storage/metabase/graveyard_test.go
index bf8acb2259..963d8135e4 100644
--- a/pkg/local_object_storage/metabase/graveyard_test.go
+++ b/pkg/local_object_storage/metabase/graveyard_test.go
@@ -70,9 +70,20 @@ func TestDB_IterateDeletedObjects(t *testing.T) {
 
 	require.NoError(t, err)
 
-	require.Equal(t, 4, counterAll)
-	require.True(t, equalAddresses([]*addressSDK.Address{object.AddressOf(obj1), object.AddressOf(obj2)}, buriedTS))
-	require.True(t, equalAddresses([]*addressSDK.Address{object.AddressOf(obj3), object.AddressOf(obj4)}, buriedGC))
+	// objects covered with a tombstone
+	// also receive GS mark
+	garbageExpected := []*addressSDK.Address{
+		object.AddressOf(obj1), object.AddressOf(obj2),
+		object.AddressOf(obj3), object.AddressOf(obj4),
+	}
+
+	graveyardExpected := []*addressSDK.Address{
+		object.AddressOf(obj1), object.AddressOf(obj2),
+	}
+
+	require.Equal(t, len(garbageExpected)+len(graveyardExpected), counterAll)
+	require.True(t, equalAddresses(graveyardExpected, buriedTS))
+	require.True(t, equalAddresses(garbageExpected, buriedGC))
 }
 
 func equalAddresses(aa1 []*addressSDK.Address, aa2 []*addressSDK.Address) bool {
diff --git a/pkg/local_object_storage/metabase/inhume.go b/pkg/local_object_storage/metabase/inhume.go
index 41f90b0e49..66e1c425cb 100644
--- a/pkg/local_object_storage/metabase/inhume.go
+++ b/pkg/local_object_storage/metabase/inhume.go
@@ -73,6 +73,8 @@ var errBreakBucketForEach = errors.New("bucket ForEach break")
 // if at least one object is locked.
 func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) {
 	err = db.boltDB.Update(func(tx *bbolt.Tx) error {
+		garbageBKT := tx.Bucket(garbageBucketName)
+
 		var (
 			// target bucket of the operation, one of the:
 			//	1. Graveyard if Inhume was called with a Tombstone
@@ -101,11 +103,7 @@ func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) {
 
 			value = tombKey
 		} else {
-			bkt, err = tx.CreateBucketIfNotExists(garbageBucketName)
-			if err != nil {
-				return err
-			}
-
+			bkt = garbageBKT
 			value = zeroValue
 		}
 
@@ -158,6 +156,13 @@ func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) {
 				if targetIsTomb {
 					continue
 				}
+
+				// if tombstone appears object must be
+				// additionally marked with GC
+				err = garbageBKT.Put(targetKey, zeroValue)
+				if err != nil {
+					return err
+				}
 			} else {
 				// garbage object can probably lock some objects, so they should become
 				// unlocked after its decay
diff --git a/pkg/local_object_storage/metabase/inhume_test.go b/pkg/local_object_storage/metabase/inhume_test.go
index eb5ccaa2bb..f9fa5f113c 100644
--- a/pkg/local_object_storage/metabase/inhume_test.go
+++ b/pkg/local_object_storage/metabase/inhume_test.go
@@ -65,10 +65,11 @@ func TestInhumeTombOnTomb(t *testing.T) {
 	require.NoError(t, err)
 
 	// record with {addr1:addr2} should be removed from graveyard
-	// as a tomb-on-tomb
-	res, err := db.Exists(existsPrm.WithAddress(addr1))
-	require.NoError(t, err)
-	require.False(t, res.Exists())
+	// as a tomb-on-tomb; metabase should return ObjectNotFound
+	// NOT ObjectAlreadyRemoved since that record has been removed
+	// from graveyard but addr1 is still marked with GC
+	_, err = db.Exists(existsPrm.WithAddress(addr1))
+	require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
 
 	// addr3 should be inhumed {addr3: addr1}
 	_, err = db.Exists(existsPrm.WithAddress(addr3))
@@ -82,10 +83,10 @@ func TestInhumeTombOnTomb(t *testing.T) {
 	require.NoError(t, err)
 
 	// record with addr1 key should not appear in graveyard
-	// (tomb can not be inhumed)
-	res, err = db.Exists(existsPrm.WithAddress(addr1))
-	require.NoError(t, err)
-	require.False(t, res.Exists())
+	// (tomb can not be inhumed) but should be kept as object
+	// with GC mark
+	_, err = db.Exists(existsPrm.WithAddress(addr1))
+	require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
 }
 
 func TestInhumeLocked(t *testing.T) {