[#1461] meta: Return error on lock object removal
Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
This commit is contained in:
parent
b5c56d459a
commit
18ec5d7c8e
3 changed files with 18 additions and 6 deletions
|
@ -75,6 +75,8 @@ func (e *StorageEngine) inhume(prm InhumePrm) (InhumeRes, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
switch e.inhumeAddr(prm.addrs[i], shPrm, true) {
|
switch e.inhumeAddr(prm.addrs[i], shPrm, true) {
|
||||||
|
case 2:
|
||||||
|
return InhumeRes{}, meta.ErrLockObjectRemoval
|
||||||
case 1:
|
case 1:
|
||||||
return InhumeRes{}, apistatus.ObjectLocked{}
|
return InhumeRes{}, apistatus.ObjectLocked{}
|
||||||
case 0:
|
case 0:
|
||||||
|
@ -93,7 +95,8 @@ func (e *StorageEngine) inhume(prm InhumePrm) (InhumeRes, error) {
|
||||||
// Returns:
|
// Returns:
|
||||||
// 0 - fail
|
// 0 - fail
|
||||||
// 1 - object locked
|
// 1 - object locked
|
||||||
// 2 - ok
|
// 2 - lock object removal
|
||||||
|
// 3 - ok
|
||||||
func (e *StorageEngine) inhumeAddr(addr oid.Address, prm shard.InhumePrm, checkExists bool) (status uint8) {
|
func (e *StorageEngine) inhumeAddr(addr oid.Address, prm shard.InhumePrm, checkExists bool) (status uint8) {
|
||||||
root := false
|
root := false
|
||||||
var errLocked apistatus.ObjectLocked
|
var errLocked apistatus.ObjectLocked
|
||||||
|
@ -114,7 +117,7 @@ func (e *StorageEngine) inhumeAddr(addr oid.Address, prm shard.InhumePrm, checkE
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if shard.IsErrRemoved(err) {
|
if shard.IsErrRemoved(err) {
|
||||||
// inhumed once - no need to be inhumed again
|
// inhumed once - no need to be inhumed again
|
||||||
status = 2
|
status = 3
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -134,15 +137,19 @@ func (e *StorageEngine) inhumeAddr(addr oid.Address, prm shard.InhumePrm, checkE
|
||||||
if err != nil {
|
if err != nil {
|
||||||
e.reportShardError(sh, "could not inhume object in shard", err)
|
e.reportShardError(sh, "could not inhume object in shard", err)
|
||||||
|
|
||||||
if errors.As(err, &errLocked) {
|
switch {
|
||||||
|
case errors.As(err, &errLocked):
|
||||||
status = 1
|
status = 1
|
||||||
return true
|
return true
|
||||||
|
case errors.Is(err, meta.ErrLockObjectRemoval):
|
||||||
|
status = 2
|
||||||
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
status = 2
|
status = 3
|
||||||
|
|
||||||
return true
|
return true
|
||||||
})
|
})
|
||||||
|
|
|
@ -9,6 +9,7 @@ import (
|
||||||
|
|
||||||
objectV2 "github.com/nspcc-dev/neofs-api-go/v2/object"
|
objectV2 "github.com/nspcc-dev/neofs-api-go/v2/object"
|
||||||
objectcore "github.com/nspcc-dev/neofs-node/pkg/core/object"
|
objectcore "github.com/nspcc-dev/neofs-node/pkg/core/object"
|
||||||
|
meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase"
|
||||||
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
|
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
|
||||||
"github.com/nspcc-dev/neofs-node/pkg/util"
|
"github.com/nspcc-dev/neofs-node/pkg/util"
|
||||||
apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
|
apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
|
||||||
|
@ -133,7 +134,7 @@ func TestLockUserScenario(t *testing.T) {
|
||||||
inhumePrm.WithTarget(tombForLockAddr, lockerAddr)
|
inhumePrm.WithTarget(tombForLockAddr, lockerAddr)
|
||||||
|
|
||||||
_, err = e.Inhume(inhumePrm)
|
_, err = e.Inhume(inhumePrm)
|
||||||
require.NoError(t, err, new(apistatus.ObjectLocked))
|
require.ErrorIs(t, err, meta.ErrLockObjectRemoval)
|
||||||
|
|
||||||
// 5.
|
// 5.
|
||||||
for i := range chEvents {
|
for i := range chEvents {
|
||||||
|
|
|
@ -93,6 +93,10 @@ func Inhume(db *DB, target, tomb oid.Address) error {
|
||||||
|
|
||||||
var errBreakBucketForEach = errors.New("bucket ForEach break")
|
var errBreakBucketForEach = errors.New("bucket ForEach break")
|
||||||
|
|
||||||
|
// ErrLockObjectRemoval is returned when inhume operation is being
|
||||||
|
// performed on lock object, and it is not a forced object removal.
|
||||||
|
var ErrLockObjectRemoval = errors.New("lock object removal")
|
||||||
|
|
||||||
// Inhume marks objects as removed but not removes it from metabase.
|
// Inhume marks objects as removed but not removes it from metabase.
|
||||||
//
|
//
|
||||||
// Allows inhuming non-locked objects only. Returns apistatus.ObjectLocked
|
// Allows inhuming non-locked objects only. Returns apistatus.ObjectLocked
|
||||||
|
@ -149,7 +153,7 @@ func (db *DB) Inhume(prm InhumePrm) (res InhumeRes, err error) {
|
||||||
// `WithForceGCMark` option
|
// `WithForceGCMark` option
|
||||||
if !prm.forceRemoval {
|
if !prm.forceRemoval {
|
||||||
if isLockObject(tx, cnr, id) {
|
if isLockObject(tx, cnr, id) {
|
||||||
return fmt.Errorf("lock object removal, CID: %s, OID: %s", cnr, id)
|
return ErrLockObjectRemoval
|
||||||
}
|
}
|
||||||
|
|
||||||
lockWasChecked = true
|
lockWasChecked = true
|
||||||
|
|
Loading…
Reference in a new issue