[#1318] metabase: Add immediate object deletion

Mark objects with GC immediately after a Tombstone is received.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>

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)
+	// objects covered with a tombstone
-	require.True(t, equalAddresses([]*addressSDK.Address{object.AddressOf(obj1), object.AddressOf(obj2)}, buriedTS))
+	// also receive GS mark
-	require.True(t, equalAddresses([]*addressSDK.Address{object.AddressOf(obj3), object.AddressOf(obj4)}, buriedGC))
+	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 {

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)
+			bkt = garbageBKT
-			if err != nil {
-				return err
-			}
-
 			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

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
+	// as a tomb-on-tomb; metabase should return ObjectNotFound
-	res, err := db.Exists(existsPrm.WithAddress(addr1))
+	// NOT ObjectAlreadyRemoved since that record has been removed
-	require.NoError(t, err)
+	// from graveyard but addr1 is still marked with GC
-	require.False(t, res.Exists())
+	_, 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)
+	// (tomb can not be inhumed) but should be kept as object
-	res, err = db.Exists(existsPrm.WithAddress(addr1))
+	// with GC mark
-	require.NoError(t, err)
+	_, err = db.Exists(existsPrm.WithAddress(addr1))
-	require.False(t, res.Exists())
+	require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
 }
 
 func TestInhumeLocked(t *testing.T) {