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

package shard

import (
	"context"
	"path/filepath"
	"testing"

	objectcore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/blobovniczatree"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/fstree"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
	cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test"
	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
	"github.com/stretchr/testify/require"
	"go.uber.org/zap"
)

func TestShard_Lock(t *testing.T) {
	t.Parallel()

	var sh *Shard

	rootPath := t.TempDir()
	opts := []Option{
		WithID(NewIDFromBytes([]byte{})),
		WithLogger(&logger.Logger{Logger: zap.NewNop()}),
		WithBlobStorOptions(
			blobstor.WithStorages([]blobstor.SubStorage{
				{
					Storage: blobovniczatree.NewBlobovniczaTree(
						blobovniczatree.WithRootPath(filepath.Join(rootPath, "blob", "blobovnicza")),
						blobovniczatree.WithBlobovniczaShallowDepth(2),
						blobovniczatree.WithBlobovniczaShallowWidth(2)),
					Policy: func(_ *objectSDK.Object, data []byte) bool {
						return len(data) <= 1<<20
					},
				},
				{
					Storage: fstree.New(
						fstree.WithPath(filepath.Join(rootPath, "blob"))),
				},
			}),
		),
		WithMetaBaseOptions(
			meta.WithPath(filepath.Join(rootPath, "meta")),
			meta.WithEpochState(epochState{}),
		),
		WithDeletedLockCallback(func(_ context.Context, addresses []oid.Address) {
			sh.HandleDeletedLocks(addresses)
		}),
	}

	sh = New(opts...)
	require.NoError(t, sh.Open(context.Background()))
	require.NoError(t, sh.Init(context.Background()))

	defer func() { require.NoError(t, sh.Close()) }()

	cnr := cidtest.ID()
	obj := testutil.GenerateObjectWithCID(cnr)
	objID, _ := obj.ID()

	lock := testutil.GenerateObjectWithCID(cnr)
	lock.SetType(objectSDK.TypeLock)
	lockID, _ := lock.ID()

	// put the object

	var putPrm PutPrm
	putPrm.SetObject(obj)

	_, err := sh.Put(context.Background(), putPrm)
	require.NoError(t, err)

	// lock the object

	err = sh.Lock(context.Background(), cnr, lockID, []oid.ID{objID})
	require.NoError(t, err)

	putPrm.SetObject(lock)
	_, err = sh.Put(context.Background(), putPrm)
	require.NoError(t, err)

	t.Run("inhuming locked objects", func(t *testing.T) {
		ts := testutil.GenerateObjectWithCID(cnr)

		var inhumePrm InhumePrm
		inhumePrm.SetTarget(objectcore.AddressOf(ts), objectcore.AddressOf(obj))

		var objLockedErr *apistatus.ObjectLocked

		_, err = sh.Inhume(context.Background(), inhumePrm)
		require.ErrorAs(t, err, &objLockedErr)

		inhumePrm.MarkAsGarbage(objectcore.AddressOf(obj))
		_, err = sh.Inhume(context.Background(), inhumePrm)
		require.ErrorAs(t, err, &objLockedErr)
	})

	t.Run("inhuming lock objects", func(t *testing.T) {
		ts := testutil.GenerateObjectWithCID(cnr)

		var inhumePrm InhumePrm
		inhumePrm.SetTarget(objectcore.AddressOf(ts), objectcore.AddressOf(lock))

		_, err = sh.Inhume(context.Background(), inhumePrm)
		require.Error(t, err)

		inhumePrm.MarkAsGarbage(objectcore.AddressOf(lock))
		_, err = sh.Inhume(context.Background(), inhumePrm)
		require.Error(t, err)
	})

	t.Run("force objects inhuming", func(t *testing.T) {
		var inhumePrm InhumePrm
		inhumePrm.MarkAsGarbage(objectcore.AddressOf(lock))
		inhumePrm.ForceRemoval()

		_, err = sh.Inhume(context.Background(), inhumePrm)
		require.NoError(t, err)

		// it should be possible to remove
		// lock object now

		inhumePrm = InhumePrm{}
		inhumePrm.MarkAsGarbage(objectcore.AddressOf(obj))

		_, err = sh.Inhume(context.Background(), inhumePrm)
		require.NoError(t, err)

		// check that object has been removed

		var getPrm GetPrm
		getPrm.SetAddress(objectcore.AddressOf(obj))

		_, err = sh.Get(context.Background(), getPrm)
		require.True(t, client.IsErrObjectNotFound(err))
	})
}

func TestShard_IsLocked(t *testing.T) {
	sh := newShard(t, false)
	defer func() { require.NoError(t, sh.Close()) }()

	cnr := cidtest.ID()
	obj := testutil.GenerateObjectWithCID(cnr)
	cnrID, _ := obj.ContainerID()
	objID, _ := obj.ID()

	lockID := oidtest.ID()

	// put the object

	var putPrm PutPrm
	putPrm.SetObject(obj)

	_, err := sh.Put(context.Background(), putPrm)
	require.NoError(t, err)

	// not locked object is not locked

	locked, err := sh.IsLocked(context.Background(), objectcore.AddressOf(obj))
	require.NoError(t, err)

	require.False(t, locked)

	// locked object is locked

	require.NoError(t, sh.Lock(context.Background(), cnrID, lockID, []oid.ID{objID}))

	locked, err = sh.IsLocked(context.Background(), objectcore.AddressOf(obj))
	require.NoError(t, err)

	require.True(t, locked)
}