diff --git a/pkg/local_object_storage/metabase/inhume.go b/pkg/local_object_storage/metabase/inhume.go index 367c3578..9e496a1d 100644 --- a/pkg/local_object_storage/metabase/inhume.go +++ b/pkg/local_object_storage/metabase/inhume.go @@ -1,7 +1,10 @@ package meta import ( + "bytes" + objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object" + "github.com/pkg/errors" "go.etcd.io/bbolt" ) @@ -61,6 +64,8 @@ func Inhume(db *DB, target, tomb *objectSDK.Address) error { const inhumeGCMarkValue = "GCMARK" +var errBreakBucketForEach = errors.New("bucket ForEach break") + // Inhume marks objects as removed but not removes it from metabase. func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) { err = db.boltDB.Update(func(tx *bbolt.Tx) error { @@ -69,6 +74,25 @@ func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) { return err } + var tombKey []byte + if prm.tomb != nil { + tombKey = addressKey(prm.tomb) + + // it is forbidden to have a tomb-on-tomb in NeoFS, + // so graveyard keys must not be addresses of tombstones + + // tombstones can be marked for GC in graveyard, so exclude this case + data := graveyard.Get(tombKey) + if data != nil && !bytes.Equal(data, []byte(inhumeGCMarkValue)) { + err := graveyard.Delete(tombKey) + if err != nil { + return errors.Wrap(err, "could not remove grave with tombstone key") + } + } + } else { + tombKey = []byte(inhumeGCMarkValue) + } + for i := range prm.target { obj, err := db.get(tx, prm.target[i], false, true) @@ -86,15 +110,37 @@ func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) { } } - var val []byte + targetKey := addressKey(prm.target[i]) + if prm.tomb != nil { - val = addressKey(prm.tomb) - } else { - val = []byte(inhumeGCMarkValue) + targetIsTomb := false + + // iterate over graveyard and check if target address + // is the address of tombstone in graveyard. + err = graveyard.ForEach(func(k, v []byte) error { + // check if graveyard has record with key corresponding + // to tombstone address (at least one) + targetIsTomb = bytes.Equal(v, targetKey) + + if targetIsTomb { + // break bucket iterator + return errBreakBucketForEach + } + + return nil + }) + if err != nil && !errors.Is(err, errBreakBucketForEach) { + return err + } + + // do not add grave if target is a tombstone + if targetIsTomb { + continue + } } // consider checking if target is already in graveyard? - err = graveyard.Put(addressKey(prm.target[i]), val) + err = graveyard.Put(targetKey, tombKey) if err != nil { return err } diff --git a/pkg/local_object_storage/metabase/inhume_test.go b/pkg/local_object_storage/metabase/inhume_test.go index e243e181..0e581887 100644 --- a/pkg/local_object_storage/metabase/inhume_test.go +++ b/pkg/local_object_storage/metabase/inhume_test.go @@ -5,6 +5,7 @@ import ( "github.com/nspcc-dev/neofs-node/pkg/core/object" meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase" + "github.com/pkg/errors" "github.com/stretchr/testify/require" ) @@ -29,3 +30,59 @@ func TestDB_Inhume(t *testing.T) { _, err = meta.Get(db, raw.Object().Address()) require.EqualError(t, err, object.ErrAlreadyRemoved.Error()) } + +func TestInhumeTombOnTomb(t *testing.T) { + db := newDB(t) + defer releaseDB(db) + + var ( + err error + + addr1 = generateAddress() + addr2 = generateAddress() + addr3 = generateAddress() + inhumePrm = new(meta.InhumePrm) + existsPrm = new(meta.ExistsPrm) + ) + + // inhume addr1 via addr2 + _, err = db.Inhume(inhumePrm. + WithAddresses(addr1). + WithTombstoneAddress(addr2), + ) + require.NoError(t, err) + + // addr1 should become inhumed {addr1:addr2} + _, err = db.Exists(existsPrm.WithAddress(addr1)) + require.True(t, errors.Is(err, object.ErrAlreadyRemoved)) + + // try to inhume addr3 via addr1 + _, err = db.Inhume(inhumePrm. + WithAddresses(addr3). + WithTombstoneAddress(addr1), + ) + require.NoError(t, err) + + // record with {addr1:addr2} should be removed from graveyard + // as a tomb-on-tomb + res, err := db.Exists(existsPrm.WithAddress(addr1)) + require.NoError(t, err) + require.False(t, res.Exists()) + + // addr3 should be inhumed {addr3: addr1} + _, err = db.Exists(existsPrm.WithAddress(addr3)) + require.True(t, errors.Is(err, object.ErrAlreadyRemoved)) + + // try to inhume addr1 (which is already a tombstone in graveyard) + _, err = db.Inhume(inhumePrm. + WithAddresses(addr1). + WithTombstoneAddress(generateAddress()), + ) + require.NoError(t, err) + + // record with addr1 key should not appear in graveyard + // (tomb can not be inhumed) + res, err = db.Exists(existsPrm.WithAddress(addr1)) + require.NoError(t, err) + require.False(t, res.Exists()) +}