From d72d37a322778a558e74eaea592701b69ef90007 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Tue, 16 Feb 2021 10:57:43 +0300
Subject: [PATCH] [#378] metabase: Implement iterator over metabase graveyard

Implement `DB.IterateOverGraveyard` method that iterates over all graves and
passes passes their descriptors (new type `Grave`) to handler (new type
`GraveHandler`). `Grave` currently have buried object address and garbage
flag.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 .../metabase/graveyard.go                     | 70 +++++++++++++++++++
 .../metabase/graveyard_test.go                | 66 +++++++++++++++++
 pkg/local_object_storage/metabase/util.go     |  6 ++
 3 files changed, 142 insertions(+)
 create mode 100644 pkg/local_object_storage/metabase/graveyard.go
 create mode 100644 pkg/local_object_storage/metabase/graveyard_test.go

diff --git a/pkg/local_object_storage/metabase/graveyard.go b/pkg/local_object_storage/metabase/graveyard.go
new file mode 100644
index 000000000..c67826792
--- /dev/null
+++ b/pkg/local_object_storage/metabase/graveyard.go
@@ -0,0 +1,70 @@
+package meta
+
+import (
+	"bytes"
+
+	"github.com/nspcc-dev/neofs-api-go/pkg/object"
+	"github.com/pkg/errors"
+	"go.etcd.io/bbolt"
+)
+
+// Grave represents descriptor of DB's graveyard item.
+type Grave struct {
+	gcMark bool
+
+	addr *object.Address
+}
+
+// WithGCMark returns true if grave marked for GC to be removed.
+func (g *Grave) WithGCMark() bool {
+	return g.gcMark
+}
+
+// Address returns buried object address.
+func (g *Grave) Address() *object.Address {
+	return g.addr
+}
+
+// GraveHandler is a Grave handling function.
+type GraveHandler func(*Grave) error
+
+// IterateOverGraveyard iterates over all graves in DB.
+//
+// Returns errors of h directly.
+func (db *DB) IterateOverGraveyard(h GraveHandler) error {
+	return db.boltDB.View(func(tx *bbolt.Tx) error {
+		return db.iterateOverGraveyard(tx, h)
+	})
+}
+
+func (db *DB) iterateOverGraveyard(tx *bbolt.Tx, h GraveHandler) error {
+	// get graveyard bucket
+	bktGraveyard := tx.Bucket(graveyardBucketName)
+	if bktGraveyard == nil {
+		return nil
+	}
+
+	// iterate over all graves
+	return bktGraveyard.ForEach(func(k, v []byte) error {
+		// parse Grave
+		g, err := graveFromKV(k, v)
+		if err != nil {
+			return errors.Wrap(err, "could not parse Grave")
+		}
+
+		// handler Grave
+		return h(g)
+	})
+}
+
+func graveFromKV(k, v []byte) (*Grave, error) {
+	addr, err := addressFromKey(k)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not parse address")
+	}
+
+	return &Grave{
+		gcMark: bytes.Equal(v, []byte(inhumeGCMarkValue)),
+		addr:   addr,
+	}, nil
+}
diff --git a/pkg/local_object_storage/metabase/graveyard_test.go b/pkg/local_object_storage/metabase/graveyard_test.go
new file mode 100644
index 000000000..154db6599
--- /dev/null
+++ b/pkg/local_object_storage/metabase/graveyard_test.go
@@ -0,0 +1,66 @@
+package meta_test
+
+import (
+	"testing"
+
+	"github.com/nspcc-dev/neofs-api-go/pkg/object"
+	meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDB_IterateOverGraveyard(t *testing.T) {
+	db := newDB(t)
+	defer releaseDB(db)
+
+	// generate and put 2 objects
+	obj1 := generateRawObject(t)
+	obj2 := generateRawObject(t)
+
+	var err error
+
+	err = putBig(db, obj1.Object())
+	require.NoError(t, err)
+
+	err = putBig(db, obj2.Object())
+	require.NoError(t, err)
+
+	inhumePrm := new(meta.InhumePrm)
+
+	// inhume with tombstone
+	addrTombstone := generateAddress()
+
+	_, err = db.Inhume(inhumePrm.
+		WithAddresses(obj1.Object().Address()).
+		WithTombstoneAddress(addrTombstone),
+	)
+	require.NoError(t, err)
+
+	// inhume with GC mark
+	_, err = db.Inhume(inhumePrm.
+		WithAddresses(obj2.Object().Address()).
+		WithGCMark(),
+	)
+
+	var (
+		counterAll         int
+		buriedTS, buriedGC []*object.Address
+	)
+
+	err = db.IterateOverGraveyard(func(g *meta.Grave) error {
+		if g.WithGCMark() {
+			buriedGC = append(buriedGC, g.Address())
+		} else {
+			buriedTS = append(buriedTS, g.Address())
+		}
+
+		counterAll++
+
+		return nil
+	})
+
+	require.NoError(t, err)
+
+	require.Equal(t, 2, counterAll)
+	require.Equal(t, []*object.Address{obj1.Object().Address()}, buriedTS)
+	require.Equal(t, []*object.Address{obj2.Object().Address()}, buriedGC)
+}
diff --git a/pkg/local_object_storage/metabase/util.go b/pkg/local_object_storage/metabase/util.go
index 02fd31bf6..8aa1964e6 100644
--- a/pkg/local_object_storage/metabase/util.go
+++ b/pkg/local_object_storage/metabase/util.go
@@ -97,6 +97,12 @@ func addressKey(addr *object.Address) []byte {
 	return []byte(addr.String())
 }
 
+// parses object address formed by addressKey.
+func addressFromKey(k []byte) (*object.Address, error) {
+	a := object.NewAddress()
+	return a, a.Parse(string(k))
+}
+
 // objectKey returns key for K-V tables when key is an object id.
 func objectKey(oid *object.ID) []byte {
 	return []byte(oid.String())