frostfs-node/pkg/local_object_storage/shard/inhume.go

package shard

import (
	"context"
	"errors"
	"fmt"

	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

// InhumePrm encapsulates parameters for inhume operation.
type InhumePrm struct {
	target       []oid.Address
	tombstone    *oid.Address
	forceRemoval bool
}

// InhumeRes encapsulates results of inhume operation.
type InhumeRes struct{}

// SetTarget sets a list of objects that should be inhumed and tombstone address
// as the reason for inhume operation.
//
// tombstone should not be nil, addr should not be empty.
// Should not be called along with MarkAsGarbage.
func (p *InhumePrm) SetTarget(tombstone oid.Address, addrs ...oid.Address) {
	p.target = addrs
	p.tombstone = &tombstone
}

// MarkAsGarbage marks object to be physically removed from shard.
//
// Should not be called along with SetTarget.
func (p *InhumePrm) MarkAsGarbage(addr ...oid.Address) {
	if p != nil {
		p.target = addr
		p.tombstone = nil
	}
}

// ForceRemoval forces object removing despite any restrictions imposed
// on deleting that object. Expected to be used only in control service.
func (p *InhumePrm) ForceRemoval() {
	if p != nil {
		p.tombstone = nil
		p.forceRemoval = true
	}
}

// ErrLockObjectRemoval is returned when inhume operation is being
// performed on lock object, and it is not a forced object removal.
var ErrLockObjectRemoval = meta.ErrLockObjectRemoval

// Inhume calls metabase. Inhume method to mark object as removed. It won't be
// removed physically from blobStor and metabase until `Delete` operation.
//
// Allows inhuming non-locked objects only. Returns apistatus.ObjectLocked
// if at least one object is locked.
//
// Returns ErrReadOnlyMode error if shard is in "read-only" mode.
func (s *Shard) Inhume(ctx context.Context, prm InhumePrm) (InhumeRes, error) {
	ctx, span := tracing.StartSpanFromContext(ctx, "Shard.Inhume",
		trace.WithAttributes(
			attribute.String("shard_id", s.ID().String()),
		))
	defer span.End()

	s.m.RLock()

	if s.info.Mode.ReadOnly() {
		s.m.RUnlock()
		return InhumeRes{}, ErrReadOnlyMode
	} else if s.info.Mode.NoMetabase() {
		s.m.RUnlock()
		return InhumeRes{}, ErrDegradedMode
	}

	if s.hasWriteCache() {
		for i := range prm.target {
			_ = s.writeCache.Delete(ctx, prm.target[i])
		}
	}

	var metaPrm meta.InhumePrm
	metaPrm.SetAddresses(prm.target...)
	metaPrm.SetLockObjectHandling()

	if prm.tombstone != nil {
		metaPrm.SetTombstoneAddress(*prm.tombstone)
	} else {
		metaPrm.SetGCMark()
	}

	if prm.forceRemoval {
		metaPrm.SetForceGCMark()
	}

	res, err := s.metaBase.Inhume(ctx, metaPrm)
	if err != nil {
		if errors.Is(err, meta.ErrLockObjectRemoval) {
			s.m.RUnlock()
			return InhumeRes{}, ErrLockObjectRemoval
		}

		s.log.Debug(logs.ShardCouldNotMarkObjectToDeleteInMetabase,
			zap.String("error", err.Error()),
		)

		s.m.RUnlock()

		return InhumeRes{}, fmt.Errorf("metabase inhume: %w", err)
	}

	s.m.RUnlock()

	s.decObjectCounterBy(logical, res.AvailableInhumed())

	i := 0
	for i < res.GetDeletionInfoLength() {
		delInfo := res.GetDeletionInfoByIndex(i)
		s.addToContainerSize(delInfo.CID.EncodeToString(), -int64(delInfo.Size))
		i++
	}

	if deletedLockObjs := res.DeletedLockObjects(); len(deletedLockObjs) != 0 {
		s.deletedLockCallBack(ctx, deletedLockObjs)
	}

	return InhumeRes{}, nil
}