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

296 lines
7.8 KiB
Go

package engine
import (
"context"
"strconv"
"testing"
"time"
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
objectcore "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"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
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/panjf2000/ants/v2"
"github.com/stretchr/testify/require"
)
type tss struct {
expEpoch uint64
}
func (t tss) IsTombstoneAvailable(ctx context.Context, _ oid.Address, epoch uint64) bool {
return t.expEpoch >= epoch
}
func TestLockUserScenario(t *testing.T) {
t.Parallel()
// Tested user actions:
// 1. stores some object
// 2. locks the object
// 3. tries to inhume the object with tombstone and expects failure
// 4. saves tombstone for LOCK-object and receives error
// 5. waits for an epoch after the lock expiration one
// 6. tries to inhume the object and expects success
const lockerExpiresAfter = 13
cnr := cidtest.ID()
tombObj := testutil.GenerateObjectWithCID(cnr)
tombForLockID := oidtest.ID()
tombObj.SetID(tombForLockID)
testEngine := testNewEngine(t).
setShardsNumAdditionalOpts(t, 2, func(id int) []shard.Option {
return []shard.Option{
shard.WithGCWorkerPoolInitializer(func(sz int) util.WorkerPool {
pool, err := ants.NewPool(sz)
require.NoError(t, err)
return pool
}),
shard.WithTombstoneSource(tss{lockerExpiresAfter}),
}
})
e := testEngine.engine
require.NoError(t, e.Open(context.Background()))
require.NoError(t, e.Init(context.Background()))
defer func() { require.NoError(t, e.Close(context.Background())) }()
lockerID := oidtest.ID()
tombID := oidtest.ID()
var err error
var objAddr oid.Address
objAddr.SetContainer(cnr)
var tombAddr oid.Address
tombAddr.SetContainer(cnr)
tombAddr.SetObject(tombID)
var lockerAddr oid.Address
lockerAddr.SetContainer(cnr)
lockerAddr.SetObject(lockerID)
var a objectSDK.Attribute
a.SetKey(objectV2.SysAttributeExpEpoch)
a.SetValue(strconv.Itoa(lockerExpiresAfter))
lockerObj := testutil.GenerateObjectWithCID(cnr)
lockerObj.SetID(lockerID)
lockerObj.SetAttributes(a)
var tombForLockAddr oid.Address
tombForLockAddr.SetContainer(cnr)
tombForLockAddr.SetObject(tombForLockID)
// 1.
obj := testutil.GenerateObjectWithCID(cnr)
id, _ := obj.ID()
objAddr.SetObject(id)
err = Put(context.Background(), e, obj)
require.NoError(t, err)
// 2.
var locker objectSDK.Lock
locker.WriteMembers([]oid.ID{id})
objectSDK.WriteLock(lockerObj, locker)
err = Put(context.Background(), e, lockerObj)
require.NoError(t, err)
err = e.Lock(context.Background(), cnr, lockerID, []oid.ID{id})
require.NoError(t, err)
// 3.
var inhumePrm InhumePrm
inhumePrm.WithTarget(tombAddr, objAddr)
var objLockedErr *apistatus.ObjectLocked
_, err = e.Inhume(context.Background(), inhumePrm)
require.ErrorAs(t, err, &objLockedErr)
// 4.
tombObj.SetType(objectSDK.TypeTombstone)
tombObj.SetID(tombForLockID)
tombObj.SetAttributes(a)
err = Put(context.Background(), e, tombObj)
require.NoError(t, err)
inhumePrm.WithTarget(tombForLockAddr, lockerAddr)
_, err = e.Inhume(context.Background(), inhumePrm)
require.ErrorIs(t, err, meta.ErrLockObjectRemoval)
// 5.
e.HandleNewEpoch(context.Background(), lockerExpiresAfter+1)
inhumePrm.WithTarget(tombAddr, objAddr)
require.Eventually(t, func() bool {
_, err = e.Inhume(context.Background(), inhumePrm)
return err == nil
}, 30*time.Second, time.Second)
}
func TestLockExpiration(t *testing.T) {
t.Parallel()
// Tested scenario:
// 1. some object is stored
// 2. lock object for it is stored, and the object is locked
// 3. lock expiration epoch is coming
// 4. after some delay the object is not locked anymore
testEngine := testNewEngine(t).
setShardsNumAdditionalOpts(t, 2, func(id int) []shard.Option {
return []shard.Option{
shard.WithGCWorkerPoolInitializer(func(sz int) util.WorkerPool {
pool, err := ants.NewPool(sz)
require.NoError(t, err)
return pool
}),
}
})
e := testEngine.engine
require.NoError(t, e.Open(context.Background()))
require.NoError(t, e.Init(context.Background()))
defer func() { require.NoError(t, e.Close(context.Background())) }()
const lockerExpiresAfter = 13
cnr := cidtest.ID()
var err error
// 1.
obj := testutil.GenerateObjectWithCID(cnr)
err = Put(context.Background(), e, obj)
require.NoError(t, err)
// 2.
var a objectSDK.Attribute
a.SetKey(objectV2.SysAttributeExpEpoch)
a.SetValue(strconv.Itoa(lockerExpiresAfter))
lock := testutil.GenerateObjectWithCID(cnr)
lock.SetType(objectSDK.TypeLock)
lock.SetAttributes(a)
err = Put(context.Background(), e, lock)
require.NoError(t, err)
id, _ := obj.ID()
idLock, _ := lock.ID()
err = e.Lock(context.Background(), cnr, idLock, []oid.ID{id})
require.NoError(t, err)
var inhumePrm InhumePrm
inhumePrm.WithTarget(oidtest.Address(), objectcore.AddressOf(obj))
var objLockedErr *apistatus.ObjectLocked
_, err = e.Inhume(context.Background(), inhumePrm)
require.ErrorAs(t, err, &objLockedErr)
// 3.
e.HandleNewEpoch(context.Background(), lockerExpiresAfter+1)
// 4.
inhumePrm.WithTarget(oidtest.Address(), objectcore.AddressOf(obj))
require.Eventually(t, func() bool {
_, err = e.Inhume(context.Background(), inhumePrm)
return err == nil
}, 30*time.Second, time.Second)
}
func TestLockForceRemoval(t *testing.T) {
t.Parallel()
// Tested scenario:
// 1. some object is stored
// 2. lock object for it is stored, and the object is locked
// 3. try to remove lock object and get error
// 4. force lock object removal
// 5. the object is not locked anymore
var e *StorageEngine
e = testNewEngine(t).
setShardsNumAdditionalOpts(t, 2, func(id int) []shard.Option {
return []shard.Option{
shard.WithGCWorkerPoolInitializer(func(sz int) util.WorkerPool {
pool, err := ants.NewPool(sz)
require.NoError(t, err)
return pool
}),
shard.WithDeletedLockCallback(e.processDeletedLocks),
}
}).engine
require.NoError(t, e.Open(context.Background()))
require.NoError(t, e.Init(context.Background()))
defer func() { require.NoError(t, e.Close(context.Background())) }()
cnr := cidtest.ID()
var err error
// 1.
obj := testutil.GenerateObjectWithCID(cnr)
err = Put(context.Background(), e, obj)
require.NoError(t, err)
// 2.
lock := testutil.GenerateObjectWithCID(cnr)
lock.SetType(objectSDK.TypeLock)
err = Put(context.Background(), e, lock)
require.NoError(t, err)
id, _ := obj.ID()
idLock, _ := lock.ID()
err = e.Lock(context.Background(), cnr, idLock, []oid.ID{id})
require.NoError(t, err)
// 3.
var inhumePrm InhumePrm
inhumePrm.MarkAsGarbage(objectcore.AddressOf(obj))
var objLockedErr *apistatus.ObjectLocked
_, err = e.Inhume(context.Background(), inhumePrm)
require.ErrorAs(t, err, &objLockedErr)
inhumePrm.WithTarget(oidtest.Address(), objectcore.AddressOf(obj))
_, err = e.Inhume(context.Background(), inhumePrm)
require.ErrorAs(t, err, &objLockedErr)
// 4.
var deletePrm DeletePrm
deletePrm.WithAddress(objectcore.AddressOf(lock))
deletePrm.WithForceRemoval()
_, err = e.Delete(context.Background(), deletePrm)
require.NoError(t, err)
// 5.
inhumePrm.MarkAsGarbage(objectcore.AddressOf(obj))
_, err = e.Inhume(context.Background(), inhumePrm)
require.NoError(t, err)
}