[#964] metabase: Drop GC marks if object not found

GC inhumes expired locks and tombstones on all the shards.
So it could be GC mark without object.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2024-02-06 19:25:24 +03:00
parent a6c9a337cd
commit bf1011131a
2 changed files with 42 additions and 2 deletions

View file

@ -254,9 +254,18 @@ func (db *DB) delete(tx *bbolt.Tx, addr oid.Address, refCounter referenceCounter
// unmarshal object, work only with physically stored (raw == true) objects
obj, err := db.get(tx, addr, key, false, true, currEpoch)
if err != nil {
if client.IsErrObjectNotFound(err) {
addrKey = addressKey(addr, key)
if garbageBKT != nil {
err := garbageBKT.Delete(addrKey)
if err != nil {
return deleteSingleResult{}, fmt.Errorf("could not remove from garbage bucket: %w", err)
}
}
return deleteSingleResult{}, nil
}
var siErr *objectSDK.SplitInfoError
if client.IsErrObjectNotFound(err) || errors.As(err, &siErr) {
if errors.As(err, &siErr) {
// if object is virtual (parent) then do nothing, it will be deleted with last child
return deleteSingleResult{}, nil
}

View file

@ -182,6 +182,37 @@ func TestDelete(t *testing.T) {
require.Equal(t, 0, len(addrs))
}
func TestDeleteDropsGCMarkIfObjectNotFound(t *testing.T) {
db := newDB(t, meta.WithEpochState(epochState{currEpoch}))
defer func() { require.NoError(t, db.Close()) }()
addr := oidtest.Address()
var prm meta.InhumePrm
prm.SetAddresses(addr)
prm.SetGCMark()
_, err := db.Inhume(context.Background(), prm)
require.NoError(t, err)
var garbageCount int
var itPrm meta.GarbageIterationPrm
itPrm.SetHandler(func(g meta.GarbageObject) error {
garbageCount++
return nil
})
require.NoError(t, db.IterateOverGarbage(context.Background(), itPrm))
require.Equal(t, 1, garbageCount)
var delPrm meta.DeletePrm
delPrm.SetAddresses(addr)
_, err = db.Delete(context.Background(), delPrm)
require.NoError(t, err)
garbageCount = 0
require.NoError(t, db.IterateOverGarbage(context.Background(), itPrm))
require.Equal(t, 0, garbageCount)
}
func metaDelete(db *meta.DB, addrs ...oid.Address) error {
var deletePrm meta.DeletePrm
deletePrm.SetAddresses(addrs...)