frostfs-node/pkg/local_object_storage/metabase/lock.go

package meta

import (
	"bytes"
	"context"
	"fmt"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/metaerr"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/util/logicerr"
	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"go.etcd.io/bbolt"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
)

var bucketNameLocked = []byte{lockedPrefix}

type keyValue struct {
	Key   []byte
	Value []byte
}

// returns name of the bucket with objects of type LOCK for specified container.
func bucketNameLockers(idCnr cid.ID, key []byte) []byte {
	return bucketName(idCnr, lockersPrefix, key)
}

// Lock marks objects as locked with another object. All objects are from the
// specified container.
//
// Allows locking regular objects only (otherwise returns apistatus.LockNonRegularObject).
//
// Locked list should be unique. Panics if it is empty.
func (db *DB) Lock(ctx context.Context, cnr cid.ID, locker oid.ID, locked []oid.ID) error {
	var (
		startedAt = time.Now()
		success   = false
	)
	defer func() {
		db.metrics.AddMethodDuration("Lock", time.Since(startedAt), success)
	}()

	_, span := tracing.StartSpanFromContext(ctx, "metabase.Lock",
		trace.WithAttributes(
			attribute.String("container_id", cnr.EncodeToString()),
			attribute.String("locker", locker.EncodeToString()),
			attribute.Int("locked_count", len(locked)),
		))
	defer span.End()

	db.modeMtx.RLock()
	defer db.modeMtx.RUnlock()

	if db.mode.NoMetabase() {
		return ErrDegradedMode
	} else if db.mode.ReadOnly() {
		return ErrReadOnlyMode
	}

	if len(locked) == 0 {
		panic("empty locked list")
	}

	err := db.lockInternal(locked, cnr, locker)
	success = err == nil
	return err
}

func (db *DB) lockInternal(locked []oid.ID, cnr cid.ID, locker oid.ID) error {
	bucketKeysLocked := make([][]byte, len(locked))
	for i := range locked {
		bucketKeysLocked[i] = objectKey(locked[i], make([]byte, objectKeySize))
	}
	key := make([]byte, cidSize)

	return metaerr.Wrap(db.boltDB.Update(func(tx *bbolt.Tx) error {
		if firstIrregularObjectType(tx, cnr, bucketKeysLocked...) != objectSDK.TypeRegular {
			return logicerr.Wrap(new(apistatus.LockNonRegularObject))
		}

		bucketLocked := tx.Bucket(bucketNameLocked)

		cnr.Encode(key)
		bucketLockedContainer, err := bucketLocked.CreateBucketIfNotExists(key)
		if err != nil {
			return fmt.Errorf("create container bucket for locked objects %v: %w", cnr, err)
		}

		keyLocker := objectKey(locker, key)
		var exLockers [][]byte
		var updLockers []byte

	loop:
		for i := range bucketKeysLocked {
			exLockers, err = decodeList(bucketLockedContainer.Get(bucketKeysLocked[i]))
			if err != nil {
				return fmt.Errorf("decode list of object lockers: %w", err)
			}

			for i := range exLockers {
				if bytes.Equal(exLockers[i], keyLocker) {
					continue loop
				}
			}

			updLockers, err = encodeList(append(exLockers, keyLocker))
			if err != nil {
				return fmt.Errorf("encode list of object lockers: %w", err)
			}

			err = bucketLockedContainer.Put(bucketKeysLocked[i], updLockers)
			if err != nil {
				return fmt.Errorf("update list of object lockers: %w", err)
			}
		}

		return nil
	}))
}

// FreeLockedBy unlocks all objects in DB which are locked by lockers.
// Returns slice of unlocked object ID's or an error.
func (db *DB) FreeLockedBy(lockers []oid.Address) ([]oid.Address, error) {
	var (
		startedAt = time.Now()
		success   = false
	)
	defer func() {
		db.metrics.AddMethodDuration("FreeLockedBy", time.Since(startedAt), success)
	}()

	db.modeMtx.RLock()
	defer db.modeMtx.RUnlock()

	if db.mode.NoMetabase() {
		return nil, ErrDegradedMode
	}

	var unlockedObjects []oid.Address

	if err := db.boltDB.Update(func(tx *bbolt.Tx) error {
		for i := range lockers {
			unlocked, err := freePotentialLocks(tx, lockers[i].Container(), lockers[i].Object())
			if err != nil {
				return err
			}
			unlockedObjects = append(unlockedObjects, unlocked...)
		}

		return nil
	}); err != nil {
		return nil, metaerr.Wrap(err)
	}
	success = true
	return unlockedObjects, nil
}

// checks if specified object is locked in the specified container.
func objectLocked(tx *bbolt.Tx, idCnr cid.ID, idObj oid.ID) bool {
	bucketLocked := tx.Bucket(bucketNameLocked)
	if bucketLocked != nil {
		key := make([]byte, cidSize)
		idCnr.Encode(key)
		bucketLockedContainer := bucketLocked.Bucket(key)
		if bucketLockedContainer != nil {
			return bucketLockedContainer.Get(objectKey(idObj, key)) != nil
		}
	}

	return false
}

// releases all records about the objects locked by the locker.
// Returns slice of unlocked object ID's or an error.
//
// Operation is very resource-intensive, which is caused by the admissibility
// of multiple locks. Also, if we knew what objects are locked, it would be
// possible to speed up the execution.
func freePotentialLocks(tx *bbolt.Tx, idCnr cid.ID, locker oid.ID) ([]oid.Address, error) {
	var unlockedObjects []oid.Address
	bucketLocked := tx.Bucket(bucketNameLocked)
	if bucketLocked == nil {
		return unlockedObjects, nil
	}

	key := make([]byte, cidSize)
	idCnr.Encode(key)

	bucketLockedContainer := bucketLocked.Bucket(key)
	if bucketLockedContainer == nil {
		return unlockedObjects, nil
	}

	keyLocker := objectKey(locker, key)
	updates := make([]keyValue, 0)
	err := bucketLockedContainer.ForEach(func(k, v []byte) error {
		keyLockers, err := decodeList(v)
		if err != nil {
			return fmt.Errorf("decode list of lockers in locked bucket: %w", err)
		}

		for i := range keyLockers {
			if bytes.Equal(keyLockers[i], keyLocker) {
				if len(keyLockers) == 1 {
					updates = append(updates, keyValue{
						Key:   k,
						Value: nil,
					})

					var id oid.ID
					err = id.Decode(k)
					if err != nil {
						return fmt.Errorf("decode unlocked object id error: %w", err)
					}

					var addr oid.Address
					addr.SetContainer(idCnr)
					addr.SetObject(id)

					unlockedObjects = append(unlockedObjects, addr)
				} else {
					// exclude locker
					keyLockers = append(keyLockers[:i], keyLockers[i+1:]...)

					v, err = encodeList(keyLockers)
					if err != nil {
						return fmt.Errorf("encode updated list of lockers: %w", err)
					}

					updates = append(updates, keyValue{
						Key:   k,
						Value: v,
					})
				}

				return nil
			}
		}

		return nil
	})

	if err != nil {
		return nil, err
	}

	if err = applyBucketUpdates(bucketLockedContainer, updates); err != nil {
		return nil, err
	}

	return unlockedObjects, nil
}

func applyBucketUpdates(bucket *bbolt.Bucket, updates []keyValue) error {
	for _, update := range updates {
		if update.Value == nil {
			err := bucket.Delete(update.Key)
			if err != nil {
				return fmt.Errorf("delete locked object record from locked bucket: %w", err)
			}
		} else {
			err := bucket.Put(update.Key, update.Value)
			if err != nil {
				return fmt.Errorf("update list of lockers: %w", err)
			}
		}
	}
	return nil
}

// IsLockedPrm groups the parameters of IsLocked operation.
type IsLockedPrm struct {
	addr oid.Address
}

// SetAddress sets object address that will be checked for lock relations.
func (i *IsLockedPrm) SetAddress(addr oid.Address) {
	i.addr = addr
}

// IsLockedRes groups the resulting values of IsLocked operation.
type IsLockedRes struct {
	locked bool
}

// Locked describes the requested object status according to the metabase
// current state.
func (i IsLockedRes) Locked() bool {
	return i.locked
}

// IsLocked checks is the provided object is locked by any `LOCK`. Not found
// object is considered as non-locked.
//
// Returns only non-logical errors related to underlying database.
func (db *DB) IsLocked(ctx context.Context, prm IsLockedPrm) (res IsLockedRes, err error) {
	var (
		startedAt = time.Now()
		success   = false
	)
	defer func() {
		db.metrics.AddMethodDuration("IsLocked", time.Since(startedAt), success)
	}()

	_, span := tracing.StartSpanFromContext(ctx, "metabase.IsLocked",
		trace.WithAttributes(
			attribute.String("address", prm.addr.EncodeToString()),
		))
	defer span.End()

	db.modeMtx.RLock()
	defer db.modeMtx.RUnlock()

	if db.mode.NoMetabase() {
		return res, ErrDegradedMode
	}
	err = metaerr.Wrap(db.boltDB.View(func(tx *bbolt.Tx) error {
		res.locked = objectLocked(tx, prm.addr.Container(), prm.addr.Object())
		return nil
	}))
	success = err == nil
	return res, err
}