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

package shard

import (
	"os"
	"path/filepath"
	"testing"

	"github.com/nspcc-dev/neofs-node/pkg/core/object"
	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor"
	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/fstree"
	meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase"
	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/pilorama"
	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard/mode"
	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/writecache"
	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
	cidtest "github.com/nspcc-dev/neofs-sdk-go/container/id/test"
	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
	oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test"
	objecttest "github.com/nspcc-dev/neofs-sdk-go/object/test"
	"github.com/stretchr/testify/require"
	"go.uber.org/zap/zaptest"
)

func TestShardOpen(t *testing.T) {
	dir := t.TempDir()
	metaPath := filepath.Join(dir, "meta")

	newShard := func() *Shard {
		return New(
			WithLogger(zaptest.NewLogger(t)),
			WithBlobStorOptions(
				blobstor.WithRootPath(filepath.Join(dir, "blob")),
				blobstor.WithShallowDepth(1),
				blobstor.WithSmallSizeLimit(1),
				blobstor.WithBlobovniczaShallowWidth(1),
				blobstor.WithBlobovniczaShallowDepth(1)),
			WithMetaBaseOptions(meta.WithPath(metaPath)),
			WithPiloramaOptions(
				pilorama.WithPath(filepath.Join(dir, "pilorama"))),
			WithWriteCache(true),
			WithWriteCacheOptions(
				writecache.WithPath(filepath.Join(dir, "wc"))))
	}

	sh := newShard()
	require.NoError(t, sh.Open())
	require.NoError(t, sh.Init())
	require.Equal(t, mode.ReadWrite, sh.GetMode())
	require.NoError(t, sh.Close())

	// Metabase can be opened in read-only => start in ReadOnly mode.
	require.NoError(t, os.Chmod(metaPath, 0444))
	sh = newShard()
	require.NoError(t, sh.Open())
	require.NoError(t, sh.Init())
	require.Equal(t, mode.ReadOnly, sh.GetMode())
	require.Error(t, sh.SetMode(mode.ReadWrite))
	require.Equal(t, mode.ReadOnly, sh.GetMode())
	require.NoError(t, sh.Close())

	// Metabase is corrupted => start in DegradedReadOnly mode.
	require.NoError(t, os.Chmod(metaPath, 0000))
	sh = newShard()
	require.NoError(t, sh.Open())
	require.NoError(t, sh.Init())
	require.Equal(t, mode.DegradedReadOnly, sh.GetMode())
	require.NoError(t, sh.Close())
}

func TestRefillMetabaseCorrupted(t *testing.T) {
	dir := t.TempDir()

	blobOpts := []blobstor.Option{
		blobstor.WithRootPath(filepath.Join(dir, "blob")),
		blobstor.WithShallowDepth(1),
		blobstor.WithSmallSizeLimit(1),
		blobstor.WithBlobovniczaShallowWidth(1),
		blobstor.WithBlobovniczaShallowDepth(1)}

	sh := New(
		WithBlobStorOptions(blobOpts...),
		WithPiloramaOptions(pilorama.WithPath(filepath.Join(dir, "pilorama"))),
		WithMetaBaseOptions(meta.WithPath(filepath.Join(dir, "meta"))))
	require.NoError(t, sh.Open())
	require.NoError(t, sh.Init())

	obj := objecttest.Object()
	obj.SetType(objectSDK.TypeRegular)
	obj.SetPayload([]byte{0, 1, 2, 3, 4, 5})

	var putPrm PutPrm
	putPrm.SetObject(obj)
	_, err := sh.Put(putPrm)
	require.NoError(t, err)
	require.NoError(t, sh.Close())

	addr := object.AddressOf(obj)
	fs := fstree.FSTree{
		DirNameLen: 2,
		Depth:      1,
		Info:       sh.blobStor.DumpInfo(),
	}
	require.NoError(t, fs.Put(addr, []byte("not an object")))

	sh = New(
		WithBlobStorOptions(blobOpts...),
		WithPiloramaOptions(pilorama.WithPath(filepath.Join(dir, "pilorama"))),
		WithMetaBaseOptions(meta.WithPath(filepath.Join(dir, "meta_new"))),
		WithRefillMetabase(true))
	require.NoError(t, sh.Open())
	require.NoError(t, sh.Init())

	var getPrm GetPrm
	getPrm.SetAddress(addr)
	_, err = sh.Get(getPrm)
	require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
	require.NoError(t, sh.Close())
}

func TestRefillMetabase(t *testing.T) {
	p := t.Name()

	defer os.RemoveAll(p)

	blobOpts := []blobstor.Option{
		blobstor.WithRootPath(filepath.Join(p, "blob")),
		blobstor.WithBlobovniczaShallowWidth(1),
		blobstor.WithBlobovniczaShallowDepth(1),
	}

	sh := New(
		WithBlobStorOptions(blobOpts...),
		WithMetaBaseOptions(
			meta.WithPath(filepath.Join(p, "meta")),
		),
		WithPiloramaOptions(
			pilorama.WithPath(filepath.Join(p, "pilorama"))),
	)

	// open Blobstor
	require.NoError(t, sh.Open())

	// initialize Blobstor
	require.NoError(t, sh.Init())

	const objNum = 5

	type objAddr struct {
		obj  *objectSDK.Object
		addr oid.Address
	}

	mObjs := make(map[string]objAddr)
	locked := make([]oid.ID, 1, 2)
	locked[0] = oidtest.ID()
	cnrLocked := cidtest.ID()
	for i := uint64(0); i < objNum; i++ {
		obj := objecttest.Object()
		obj.SetType(objectSDK.TypeRegular)

		if len(locked) < 2 {
			obj.SetContainerID(cnrLocked)
			id, _ := obj.ID()
			locked = append(locked, id)
		}

		addr := object.AddressOf(obj)

		mObjs[addr.EncodeToString()] = objAddr{
			obj:  obj,
			addr: addr,
		}
	}

	tombObj := objecttest.Object()
	tombObj.SetType(objectSDK.TypeTombstone)

	tombstone := objecttest.Tombstone()

	tombData, err := tombstone.Marshal()
	require.NoError(t, err)

	tombObj.SetPayload(tombData)

	tombMembers := make([]oid.Address, 0, len(tombstone.Members()))

	members := tombstone.Members()
	for i := range tombstone.Members() {
		var a oid.Address
		a.SetObject(members[i])
		cnr, _ := tombObj.ContainerID()
		a.SetContainer(cnr)

		tombMembers = append(tombMembers, a)
	}

	var putPrm PutPrm

	for _, v := range mObjs {
		putPrm.SetObject(v.obj)

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

	putPrm.SetObject(tombObj)

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

	// LOCK object handling
	var lock objectSDK.Lock
	lock.WriteMembers(locked)

	lockObj := objecttest.Object()
	lockObj.SetContainerID(cnrLocked)
	objectSDK.WriteLock(lockObj, lock)

	putPrm.SetObject(lockObj)
	_, err = sh.Put(putPrm)
	require.NoError(t, err)

	lockID, _ := lockObj.ID()
	require.NoError(t, sh.Lock(cnrLocked, lockID, locked))

	var inhumePrm InhumePrm
	inhumePrm.SetTarget(object.AddressOf(tombObj), tombMembers...)

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

	var headPrm HeadPrm

	checkObj := func(addr oid.Address, expObj *objectSDK.Object) {
		headPrm.SetAddress(addr)

		res, err := sh.Head(headPrm)

		if expObj == nil {
			require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
			return
		}

		require.NoError(t, err)
		require.Equal(t, expObj.CutPayload(), res.Object())
	}

	checkAllObjs := func(exists bool) {
		for _, v := range mObjs {
			if exists {
				checkObj(v.addr, v.obj)
			} else {
				checkObj(v.addr, nil)
			}
		}
	}

	checkTombMembers := func(exists bool) {
		for _, member := range tombMembers {
			headPrm.SetAddress(member)

			_, err := sh.Head(headPrm)

			if exists {
				require.ErrorAs(t, err, new(apistatus.ObjectAlreadyRemoved))
			} else {
				require.ErrorAs(t, err, new(apistatus.ObjectNotFound))
			}
		}
	}

	checkLocked := func(t *testing.T, cnr cid.ID, locked []oid.ID) {
		var addr oid.Address
		addr.SetContainer(cnr)

		for i := range locked {
			addr.SetObject(locked[i])

			var prm InhumePrm
			prm.MarkAsGarbage(addr)

			_, err := sh.Inhume(prm)
			require.ErrorAs(t, err, new(apistatus.ObjectLocked),
				"object %s should be locked", locked[i])
		}
	}

	checkAllObjs(true)
	checkObj(object.AddressOf(tombObj), tombObj)
	checkTombMembers(true)
	checkLocked(t, cnrLocked, locked)

	err = sh.Close()
	require.NoError(t, err)

	sh = New(
		WithBlobStorOptions(blobOpts...),
		WithMetaBaseOptions(
			meta.WithPath(filepath.Join(p, "meta_restored")),
		),
		WithPiloramaOptions(
			pilorama.WithPath(filepath.Join(p, "pilorama_another"))),
	)

	// open Blobstor
	require.NoError(t, sh.Open())

	// initialize Blobstor
	require.NoError(t, sh.Init())

	defer sh.Close()

	checkAllObjs(false)
	checkObj(object.AddressOf(tombObj), nil)
	checkTombMembers(false)

	err = sh.refillMetabase()
	require.NoError(t, err)

	checkAllObjs(true)
	checkObj(object.AddressOf(tombObj), tombObj)
	checkTombMembers(true)
	checkLocked(t, cnrLocked, locked)
}