dstepanov-yadro/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Issues Pull requests Projects Releases Packages Wiki Activity

[#1176] metabase: Ignore locked objs in IterateCoveredByTombstones

Browse source 
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 <leonard@nspcc.ru>
This commit is contained in:
Leonard Lyubich 2022-02-15 21:24:12 +03:00 committed by LeL
parent dba3e58dc5
commit 303eb2e078
2 changed files with 25 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
pkg/local_object_storage/metabase/iterators.go
View file

@ -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 // 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. // If h returns ErrInterruptIterator, nil returns immediately.
// Returns other errors of h directly. // 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) 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) return h(addr)
} }

21
pkg/local_object_storage/metabase/iterators_test.go
View file

@ -72,6 +72,7 @@ func TestDB_IterateCoveredByTombstones(t *testing.T) {
ts := generateAddress() ts := generateAddress()
protected1 := generateAddress() protected1 := generateAddress()
protected2 := generateAddress() protected2 := generateAddress()
protectedLocked := generateAddress()
garbage := generateAddress() garbage := generateAddress()
prm := new(meta.InhumePrm) prm := new(meta.InhumePrm)
@ -80,7 +81,7 @@ func TestDB_IterateCoveredByTombstones(t *testing.T) {
_, err = db.Inhume(prm. _, err = db.Inhume(prm.
WithTombstoneAddress(ts). WithTombstoneAddress(ts).
WithAddresses(protected1, protected2), WithAddresses(protected1, protected2, protectedLocked),
) )
require.NoError(t, err) require.NoError(t, err)
@ -88,6 +89,7 @@ func TestDB_IterateCoveredByTombstones(t *testing.T) {
WithAddresses(garbage). WithAddresses(garbage).
WithGCMark(), WithGCMark(),
) )
require.NoError(t, err)
var handled []*addressSDK.Address var handled []*addressSDK.Address
@ -101,7 +103,22 @@ func TestDB_IterateCoveredByTombstones(t *testing.T) {
}) })
require.NoError(t, err) require.NoError(t, err)
require.Len(t, handled, 2) require.Len(t, handled, 3)
require.Contains(t, handled, protected1) require.Contains(t, handled, protected1)
require.Contains(t, handled, protected2) 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)
} }