[#884] shard: Mark expired tombstones as garbage

Mark expired tombstones as garbage in `Shard.HandleExpiredTombstones`.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
Leonard Lyubich 2021-10-14 16:23:21 +03:00 committed by Alex Vanin
parent 8efeba8010
commit 1b3637aad7

View file

@ -294,13 +294,16 @@ func (s *Shard) collectExpiredTombstones(ctx context.Context, e Event) {
s.expiredTombstonesCallback(ctx, expired) s.expiredTombstonesCallback(ctx, expired)
} }
// HandleExpiredTombstones mark to be removed all objects that are expired in epoch // HandleExpiredTombstones marks to be removed all objects that are
// and protected by tombstone with string address from tss. // protected by tombstones with string addresses from tss.
// If successful, marks tombstones themselves as garbage.
// //
// Does not modify tss. // Does not modify tss.
func (s *Shard) HandleExpiredTombstones(tss map[string]struct{}) { func (s *Shard) HandleExpiredTombstones(tss map[string]struct{}) {
inhume := make([]*object.Address, 0, len(tss)) inhume := make([]*object.Address, 0, len(tss))
// Collect all objects covered by the tombstones.
err := s.metaBase.IterateCoveredByTombstones(tss, func(addr *object.Address) error { err := s.metaBase.IterateCoveredByTombstones(tss, func(addr *object.Address) error {
inhume = append(inhume, addr) inhume = append(inhume, addr)
return nil return nil
@ -310,20 +313,60 @@ func (s *Shard) HandleExpiredTombstones(tss map[string]struct{}) {
zap.String("error", err.Error()), zap.String("error", err.Error()),
) )
return
} else if len(inhume) == 0 {
return return
} }
_, err = s.metaBase.Inhume(new(meta.InhumePrm). // Mark collected objects as garbage.
WithAddresses(inhume...).
WithGCMark(), var pInhume meta.InhumePrm
)
pInhume.WithGCMark()
if len(inhume) > 0 {
// inhume objects
pInhume.WithAddresses(inhume...)
_, err = s.metaBase.Inhume(&pInhume)
if err != nil { if err != nil {
s.log.Warn("could not inhume objects under the expired tombstone", s.log.Warn("could not inhume objects under the expired tombstone",
zap.String("error", err.Error()), zap.String("error", err.Error()),
) )
return
}
}
// Mark the tombstones as garbage.
inhume = inhume[:0]
for strAddr := range tss {
// parse address
// TODO: make type of map values *object.Address since keys are calculated from addresses
addr := object.NewAddress()
err = addr.Parse(strAddr)
if err != nil {
s.log.Error("could not parse tombstone address",
zap.String("text", strAddr),
zap.String("error", err.Error()),
)
continue // try process other tombstones
}
inhume = append(inhume, addr)
}
pInhume.WithAddresses(inhume...) // GC mark is already set above
// inhume tombstones
_, err = s.metaBase.Inhume(&pInhume)
if err != nil {
s.log.Warn("could not mark tombstones as garbage",
zap.String("error", err.Error()),
)
return return
} }
} }