From 3d5169c4c9c77dd46599c8581d327526794f381c Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Wed, 17 Feb 2021 15:22:13 +0300
Subject: [PATCH] [#378] metabase: Implement iterator over the objects under
 tombstones

Implement `DB.IterateCoveredByTombstones` method that iterates over graves
and handles all objects under one of the tombstones.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 .../metabase/iterators.go                     | 41 +++++++++++++++++++
 .../metabase/iterators_test.go                | 41 +++++++++++++++++++
 2 files changed, 82 insertions(+)

diff --git a/pkg/local_object_storage/metabase/iterators.go b/pkg/local_object_storage/metabase/iterators.go
index 7e1dc0e12c..b8085048fe 100644
--- a/pkg/local_object_storage/metabase/iterators.go
+++ b/pkg/local_object_storage/metabase/iterators.go
@@ -109,3 +109,44 @@ func objectType(tx *bbolt.Tx, cid *container.ID, oidBytes []byte) object.Type {
 		return object.TypeStorageGroup
 	}
 }
+
+// IterateCoveredByTombstones iterates over all objects in DB which are covered
+// by tombstone with string address from tss.
+//
+// If h returns ErrInterruptIterator, nil returns immediately.
+// Returns other errors of h directly.
+//
+// Does not modify tss.
+func (db *DB) IterateCoveredByTombstones(tss map[string]struct{}, h func(*object.Address) error) error {
+	return db.boltDB.View(func(tx *bbolt.Tx) error {
+		return db.iterateCoveredByTombstones(tx, tss, h)
+	})
+}
+
+func (db *DB) iterateCoveredByTombstones(tx *bbolt.Tx, tss map[string]struct{}, h func(*object.Address) error) error {
+	bktGraveyard := tx.Bucket(graveyardBucketName)
+	if bktGraveyard == nil {
+		return nil
+	}
+
+	err := bktGraveyard.ForEach(func(k, v []byte) error {
+		if _, ok := tss[string(v)]; ok {
+			addr := object.NewAddress()
+
+			err := addr.Parse(string(k))
+			if err != nil {
+				return errors.Wrap(err, "could not parse address of the object under tombstone")
+			}
+
+			return h(addr)
+		}
+
+		return nil
+	})
+
+	if errors.Is(err, ErrInterruptIterator) {
+		err = nil
+	}
+
+	return err
+}
diff --git a/pkg/local_object_storage/metabase/iterators_test.go b/pkg/local_object_storage/metabase/iterators_test.go
index 16ea5075bd..f1abe01e88 100644
--- a/pkg/local_object_storage/metabase/iterators_test.go
+++ b/pkg/local_object_storage/metabase/iterators_test.go
@@ -56,3 +56,44 @@ func putWithExpiration(t *testing.T, db *meta.DB, typ object.Type, expiresAt uin
 
 	return obj.Address()
 }
+
+func TestDB_IterateCoveredByTombstones(t *testing.T) {
+	db := newDB(t)
+	defer releaseDB(db)
+
+	ts := generateAddress()
+	protected1 := generateAddress()
+	protected2 := generateAddress()
+	garbage := generateAddress()
+
+	prm := new(meta.InhumePrm)
+
+	var err error
+
+	_, err = db.Inhume(prm.
+		WithTombstoneAddress(ts).
+		WithAddresses(protected1, protected2),
+	)
+	require.NoError(t, err)
+
+	_, err = db.Inhume(prm.
+		WithAddresses(garbage).
+		WithGCMark(),
+	)
+
+	var handled []*object.Address
+
+	tss := map[string]struct{}{
+		ts.String(): {},
+	}
+
+	err = db.IterateCoveredByTombstones(tss, func(addr *object.Address) error {
+		handled = append(handled, addr)
+		return nil
+	})
+	require.NoError(t, err)
+
+	require.Len(t, handled, 2)
+	require.Contains(t, handled, protected1)
+	require.Contains(t, handled, protected2)
+}