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

package meta_test

import (
	"context"
	"strconv"
	"testing"

	objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
	object2 "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
	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"
)

func TestDB_IterateExpired(t *testing.T) {
	db := newDB(t)
	defer func() { require.NoError(t, db.Close()) }()

	const epoch = 13

	mAlive := map[objectSDK.Type]oid.Address{}
	mExpired := map[objectSDK.Type]oid.Address{}

	for _, typ := range []objectSDK.Type{
		objectSDK.TypeRegular,
		objectSDK.TypeTombstone,
		objectSDK.TypeLock,
	} {
		mAlive[typ] = putWithExpiration(t, db, typ, epoch)
		mExpired[typ] = putWithExpiration(t, db, typ, epoch-1)
	}

	expiredLocked := putWithExpiration(t, db, objectSDK.TypeRegular, epoch-1)

	require.NoError(t, db.Lock(context.Background(), expiredLocked.Container(), oidtest.ID(), []oid.ID{expiredLocked.Object()}))

	err := db.IterateExpired(context.Background(), epoch, func(exp *meta.ExpiredObject) error {
		if addr, ok := mAlive[exp.Type()]; ok {
			require.NotEqual(t, addr, exp.Address())
		}

		require.NotEqual(t, expiredLocked, exp.Address())

		addr, ok := mExpired[exp.Type()]
		require.True(t, ok)
		require.Equal(t, addr, exp.Address())

		delete(mExpired, exp.Type())

		return nil
	})
	require.NoError(t, err)

	require.Empty(t, mExpired)
}

func putWithExpiration(t *testing.T, db *meta.DB, typ objectSDK.Type, expiresAt uint64) oid.Address {
	obj := testutil.GenerateObject()
	obj.SetType(typ)
	testutil.AddAttribute(obj, objectV2.SysAttributeExpEpoch, strconv.FormatUint(expiresAt, 10))

	require.NoError(t, putBig(db, obj))

	return object2.AddressOf(obj)
}

func TestDB_IterateCoveredByTombstones(t *testing.T) {
	db := newDB(t)
	defer func() { require.NoError(t, db.Close()) }()

	ts := oidtest.Address()
	protected1 := oidtest.Address()
	protected2 := oidtest.Address()
	protectedLocked := oidtest.Address()
	garbage := oidtest.Address()

	var prm meta.InhumePrm
	var err error

	prm.SetAddresses(protected1, protected2, protectedLocked)
	prm.SetTombstoneAddress(ts)

	_, err = db.Inhume(context.Background(), prm)
	require.NoError(t, err)

	prm.SetAddresses(garbage)
	prm.SetGCMark()

	_, err = db.Inhume(context.Background(), prm)
	require.NoError(t, err)

	var handled []oid.Address

	tss := map[string]oid.Address{
		ts.EncodeToString(): ts,
	}

	err = db.IterateCoveredByTombstones(context.Background(), tss, func(addr oid.Address) error {
		handled = append(handled, addr)
		return nil
	})
	require.NoError(t, err)

	require.Len(t, handled, 3)
	require.Contains(t, handled, protected1)
	require.Contains(t, handled, protected2)
	require.Contains(t, handled, protectedLocked)

	err = db.Lock(context.Background(), protectedLocked.Container(), oidtest.ID(), []oid.ID{protectedLocked.Object()})
	require.NoError(t, err)

	handled = handled[:0]

	err = db.IterateCoveredByTombstones(context.Background(), tss, func(addr oid.Address) error {
		handled = append(handled, addr)
		return nil
	})
	require.NoError(t, err)

	require.Len(t, handled, 2)
	require.NotContains(t, handled, protectedLocked)
}