From 1b3637aad746ee59e844cfc35ae250c10a008b25 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Thu, 14 Oct 2021 16:23:21 +0300
Subject: [PATCH] [#884] shard: Mark expired tombstones as garbage

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

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 pkg/local_object_storage/shard/gc.go | 61 ++++++++++++++++++++++++----
 1 file changed, 52 insertions(+), 9 deletions(-)

diff --git a/pkg/local_object_storage/shard/gc.go b/pkg/local_object_storage/shard/gc.go
index 395d0d46..79a3e151 100644
--- a/pkg/local_object_storage/shard/gc.go
+++ b/pkg/local_object_storage/shard/gc.go
@@ -294,13 +294,16 @@ func (s *Shard) collectExpiredTombstones(ctx context.Context, e Event) {
 	s.expiredTombstonesCallback(ctx, expired)
 }
 
-// HandleExpiredTombstones mark to be removed all objects that are expired in epoch
-// and protected by tombstone with string address from tss.
+// HandleExpiredTombstones marks to be removed all objects that are
+// protected by tombstones with string addresses from tss.
+// If successful, marks tombstones themselves as garbage.
 //
 // Does not modify tss.
 func (s *Shard) HandleExpiredTombstones(tss map[string]struct{}) {
 	inhume := make([]*object.Address, 0, len(tss))
 
+	// Collect all objects covered by the tombstones.
+
 	err := s.metaBase.IterateCoveredByTombstones(tss, func(addr *object.Address) error {
 		inhume = append(inhume, addr)
 		return nil
@@ -310,17 +313,57 @@ func (s *Shard) HandleExpiredTombstones(tss map[string]struct{}) {
 			zap.String("error", err.Error()),
 		)
 
-		return
-	} else if len(inhume) == 0 {
 		return
 	}
 
-	_, err = s.metaBase.Inhume(new(meta.InhumePrm).
-		WithAddresses(inhume...).
-		WithGCMark(),
-	)
+	// Mark collected objects as garbage.
+
+	var pInhume meta.InhumePrm
+
+	pInhume.WithGCMark()
+
+	if len(inhume) > 0 {
+		// inhume objects
+		pInhume.WithAddresses(inhume...)
+
+		_, err = s.metaBase.Inhume(&pInhume)
+		if err != nil {
+			s.log.Warn("could not inhume objects under the expired tombstone",
+				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 inhume objects under the expired tombstone",
+		s.log.Warn("could not mark tombstones as garbage",
 			zap.String("error", err.Error()),
 		)