From 0d6d195d0ddc9e6fa3e658cd5daf62760b3e21a0 Mon Sep 17 00:00:00 2001
From: Leonard Lyubich <leonard@nspcc.ru>
Date: Mon, 15 Feb 2021 14:50:21 +0300
Subject: [PATCH] [#377] metabase: Support batch Inhume operation

Replace single target address in `InhumePrm` with the list of addresses.
Rename `WithAddress` method to `WithAddresses` and change parameter to
variadic.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
---
 pkg/local_object_storage/metabase/inhume.go | 57 ++++++++++++---------
 1 file changed, 33 insertions(+), 24 deletions(-)

diff --git a/pkg/local_object_storage/metabase/inhume.go b/pkg/local_object_storage/metabase/inhume.go
index a43d6ead..367c3578 100644
--- a/pkg/local_object_storage/metabase/inhume.go
+++ b/pkg/local_object_storage/metabase/inhume.go
@@ -7,16 +7,18 @@ import (
 
 // InhumePrm encapsulates parameters for Inhume operation.
 type InhumePrm struct {
-	target, tomb *objectSDK.Address
+	tomb *objectSDK.Address
+
+	target []*objectSDK.Address
 }
 
 // InhumeRes encapsulates results of Inhume operation.
 type InhumeRes struct{}
 
-// WithAddress sets object address that should be inhumed.
-func (p *InhumePrm) WithAddress(addr *objectSDK.Address) *InhumePrm {
+// WithAddresses sets list of object addresses that should be inhumed.
+func (p *InhumePrm) WithAddresses(addrs ...*objectSDK.Address) *InhumePrm {
 	if p != nil {
-		p.target = addr
+		p.target = addrs
 	}
 
 	return p
@@ -50,7 +52,7 @@ func (p *InhumePrm) WithGCMark() *InhumePrm {
 // tomb should not be nil.
 func Inhume(db *DB, target, tomb *objectSDK.Address) error {
 	_, err := db.Inhume(new(InhumePrm).
-		WithAddress(target).
+		WithAddresses(target).
 		WithTombstoneAddress(tomb),
 	)
 
@@ -67,31 +69,38 @@ func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) {
 			return err
 		}
 
-		obj, err := db.get(tx, prm.target, false, true)
+		for i := range prm.target {
+			obj, err := db.get(tx, prm.target[i], false, true)
 
-		// if object is stored and it is regular object then update bucket
-		// with container size estimations
-		if err == nil && obj.Type() == objectSDK.TypeRegular {
-			err := changeContainerSize(
-				tx,
-				obj.ContainerID(),
-				obj.PayloadSize(),
-				false,
-			)
+			// if object is stored and it is regular object then update bucket
+			// with container size estimations
+			if err == nil && obj.Type() == objectSDK.TypeRegular {
+				err := changeContainerSize(
+					tx,
+					obj.ContainerID(),
+					obj.PayloadSize(),
+					false,
+				)
+				if err != nil {
+					return err
+				}
+			}
+
+			var val []byte
+			if prm.tomb != nil {
+				val = addressKey(prm.tomb)
+			} else {
+				val = []byte(inhumeGCMarkValue)
+			}
+
+			// consider checking if target is already in graveyard?
+			err = graveyard.Put(addressKey(prm.target[i]), val)
 			if err != nil {
 				return err
 			}
 		}
 
-		var val []byte
-		if prm.tomb != nil {
-			val = addressKey(prm.tomb)
-		} else {
-			val = []byte(inhumeGCMarkValue)
-		}
-
-		// consider checking if target is already in graveyard?
-		return graveyard.Put(addressKey(prm.target), val)
+		return nil
 	})
 
 	return