From 303eb2e078a5925c418d1c9da7ab7c529feda38a Mon Sep 17 00:00:00 2001 From: Leonard Lyubich Date: Tue, 15 Feb 2022 21:24:12 +0300 Subject: [PATCH] [#1176] metabase: Ignore locked objs in `IterateCoveredByTombstones` Make `DB.IterateCoveredByTombstones` to not pass locked objects to the handler. The method is used by GC, therefore it will not consider locked objects as candidates for deletion even if their tombstone is expired. Signed-off-by: Leonard Lyubich --- .../metabase/iterators.go | 7 ++++++- .../metabase/iterators_test.go | 21 +++++++++++++++++-- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/pkg/local_object_storage/metabase/iterators.go b/pkg/local_object_storage/metabase/iterators.go index b1df0ec5..0b03cf57 100644 --- a/pkg/local_object_storage/metabase/iterators.go +++ b/pkg/local_object_storage/metabase/iterators.go @@ -112,7 +112,8 @@ func (db *DB) iterateExpired(tx *bbolt.Tx, epoch uint64, h ExpiredObjectHandler) } // IterateCoveredByTombstones iterates over all objects in DB which are covered -// by tombstone with string address from tss. +// by tombstone with string address from tss. Locked objects are not included +// (do not confuse with objects of type LOCK). // // If h returns ErrInterruptIterator, nil returns immediately. // Returns other errors of h directly. @@ -137,6 +138,10 @@ func (db *DB) iterateCoveredByTombstones(tx *bbolt.Tx, tss map[string]*addressSD return fmt.Errorf("could not parse address of the object under tombstone: %w", err) } + if objectLocked(tx, *addr.ContainerID(), *addr.ObjectID()) { + return nil + } + return h(addr) } diff --git a/pkg/local_object_storage/metabase/iterators_test.go b/pkg/local_object_storage/metabase/iterators_test.go index 2763f7ca..ebb63b6a 100644 --- a/pkg/local_object_storage/metabase/iterators_test.go +++ b/pkg/local_object_storage/metabase/iterators_test.go @@ -72,6 +72,7 @@ func TestDB_IterateCoveredByTombstones(t *testing.T) { ts := generateAddress() protected1 := generateAddress() protected2 := generateAddress() + protectedLocked := generateAddress() garbage := generateAddress() prm := new(meta.InhumePrm) @@ -80,7 +81,7 @@ func TestDB_IterateCoveredByTombstones(t *testing.T) { _, err = db.Inhume(prm. WithTombstoneAddress(ts). - WithAddresses(protected1, protected2), + WithAddresses(protected1, protected2, protectedLocked), ) require.NoError(t, err) @@ -88,6 +89,7 @@ func TestDB_IterateCoveredByTombstones(t *testing.T) { WithAddresses(garbage). WithGCMark(), ) + require.NoError(t, err) var handled []*addressSDK.Address @@ -101,7 +103,22 @@ func TestDB_IterateCoveredByTombstones(t *testing.T) { }) require.NoError(t, err) - require.Len(t, handled, 2) + require.Len(t, handled, 3) require.Contains(t, handled, protected1) require.Contains(t, handled, protected2) + require.Contains(t, handled, protectedLocked) + + err = db.Lock(*protectedLocked.ContainerID(), *generateAddress().ObjectID(), []oid.ID{*protectedLocked.ObjectID()}) + require.NoError(t, err) + + handled = handled[:0] + + err = db.IterateCoveredByTombstones(tss, func(addr *addressSDK.Address) error { + handled = append(handled, addr) + return nil + }) + require.NoError(t, err) + + require.Len(t, handled, 2) + require.NotContains(t, handled, protectedLocked) }