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

package shard

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

	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/common"
	"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/local_object_storage/pilorama"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger/test"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
	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"
	"github.com/panjf2000/ants/v2"
	"github.com/stretchr/testify/require"
)

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

	rootPath := t.TempDir()

	var sh *Shard
	l := test.NewLogger(t)
	blobOpts := []blobstor.Option{
		blobstor.WithLogger(test.NewLogger(t)),
		blobstor.WithStorages([]blobstor.SubStorage{
			{
				Storage: blobovniczatree.NewBlobovniczaTree(
					context.Background(),
					blobovniczatree.WithLogger(test.NewLogger(t)),
					blobovniczatree.WithRootPath(filepath.Join(rootPath, "blob", "blobovnicza")),
					blobovniczatree.WithBlobovniczaShallowDepth(1),
					blobovniczatree.WithBlobovniczaShallowWidth(1)),
				Policy: func(_ *objectSDK.Object, data []byte) bool {
					return len(data) <= 1<<20
				},
			},
			{
				Storage: fstree.New(
					fstree.WithPath(filepath.Join(rootPath, "blob"))),
			},
		}),
	}

	opts := []Option{
		WithID(NewIDFromBytes([]byte{})),
		WithLogger(l),
		WithBlobStorOptions(blobOpts...),
		WithMetaBaseOptions(
			meta.WithPath(filepath.Join(rootPath, "meta")),
			meta.WithEpochState(epochState{}),
		),
		WithPiloramaOptions(pilorama.WithPath(filepath.Join(rootPath, "pilorama"))),
		WithDeletedLockCallback(func(_ context.Context, addresses []oid.Address) {
			sh.HandleDeletedLocks(addresses)
		}),
		WithExpiredLocksCallback(func(ctx context.Context, epoch uint64, a []oid.Address) {
			sh.HandleExpiredLocks(ctx, epoch, a)
		}),
		WithGCWorkerPoolInitializer(func(sz int) util.WorkerPool {
			pool, err := ants.NewPool(sz)
			require.NoError(t, err)
			return pool
		}),
		WithGCRemoverSleepInterval(1 * time.Second),
	}

	sh = New(opts...)
	sh.gcCfg.testHookRemover = func(context.Context) gcRunResult { return gcRunResult{} }
	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()
	var addr oid.Address
	addr.SetContainer(cnr)
	addr.SetObject(objID)

	var putPrm PutPrm
	putPrm.SetObject(obj)

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

	var getPrm GetPrm
	getPrm.SetAddress(objectCore.AddressOf(obj))
	_, err = sh.Get(context.Background(), getPrm)
	require.NoError(t, err, "failed to get")

	// inhume
	var inhumePrm InhumePrm
	inhumePrm.MarkAsGarbage(addr)
	_, err = sh.Inhume(context.Background(), inhumePrm)
	require.NoError(t, err, "failed to inhume")
	_, err = sh.Get(context.Background(), getPrm)
	require.Error(t, err, "get returned error")
	require.True(t, client.IsErrObjectNotFound(err), "invalid error type")

	// storageID
	var metaStIDPrm meta.StorageIDPrm
	metaStIDPrm.SetAddress(addr)
	storageID, err := sh.metaBase.StorageID(context.Background(), metaStIDPrm)
	require.NoError(t, err, "failed to get storage ID")

	// check existence in blobstore
	var bsExisted common.ExistsPrm
	bsExisted.Address = addr
	bsExisted.StorageID = storageID.StorageID()
	exRes, err := sh.blobStor.Exists(context.Background(), bsExisted)
	require.NoError(t, err, "failed to check blobstore existence")
	require.True(t, exRes.Exists, "invalid blobstore existence result")

	// drop from blobstor
	var bsDeletePrm common.DeletePrm
	bsDeletePrm.Address = addr
	bsDeletePrm.StorageID = storageID.StorageID()
	_, err = sh.blobStor.Delete(context.Background(), bsDeletePrm)
	require.NoError(t, err, "failed to delete from blobstore")

	// check existence in blobstore
	exRes, err = sh.blobStor.Exists(context.Background(), bsExisted)
	require.NoError(t, err, "failed to check blobstore existence")
	require.False(t, exRes.Exists, "invalid blobstore existence result")

	// get should return object not found
	_, err = sh.Get(context.Background(), getPrm)
	require.Error(t, err, "get returned no error")
	require.True(t, client.IsErrObjectNotFound(err), "invalid error type")
}