achuprov/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#1175] metabase: Return status error on Inhume of locked object

Browse source 
Make `DB.Inhume` to return `apistatus.ObjectLocked` if at least one of
the inhumed objects is locked.

Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
Leonard Lyubich 2022-02-16 01:16:07 +03:00 committed by LeL
parent 303eb2e078
commit 23fcacd3f2
2 changed files with 30 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
pkg/local_object_storage/metabase/inhume.go
View file

@ -5,6 +5,7 @@ import (
"errors" "errors"
"fmt" "fmt"
apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
"github.com/nspcc-dev/neofs-sdk-go/object" "github.com/nspcc-dev/neofs-sdk-go/object"
addressSDK "github.com/nspcc-dev/neofs-sdk-go/object/address" addressSDK "github.com/nspcc-dev/neofs-sdk-go/object/address"
"go.etcd.io/bbolt" "go.etcd.io/bbolt"
@ -69,6 +70,9 @@ const inhumeGCMarkValue = "GCMARK"
var errBreakBucketForEach = errors.New("bucket ForEach break") var errBreakBucketForEach = errors.New("bucket ForEach break")
// 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
// if at least one object is locked.
func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) { func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) {
err = db.boltDB.Update(func(tx *bbolt.Tx) error { err = db.boltDB.Update(func(tx *bbolt.Tx) error {
graveyard, err := tx.CreateBucketIfNotExists(graveyardBucketName) graveyard, err := tx.CreateBucketIfNotExists(graveyardBucketName)
@ -96,6 +100,11 @@ func (db *DB) Inhume(prm *InhumePrm) (res *InhumeRes, err error) {
} }
for i := range prm.target { for i := range prm.target {
// prevent locked objects to be inhumed
if objectLocked(tx, *prm.target[i].ContainerID(), *prm.target[i].ObjectID()) {
return apistatus.ObjectLocked{}
}
obj, err := db.get(tx, prm.target[i], false, true) obj, err := db.get(tx, prm.target[i], false, true)
// if object is stored and it is regular object then update bucket // if object is stored and it is regular object then update bucket

21
pkg/local_object_storage/metabase/inhume_test.go
View file

@ -6,6 +6,10 @@ import (
"github.com/nspcc-dev/neofs-node/pkg/core/object" "github.com/nspcc-dev/neofs-node/pkg/core/object"
meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase" meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase"
apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
objecttest "github.com/nspcc-dev/neofs-sdk-go/object/address/test"
oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@ -84,3 +88,20 @@ func TestInhumeTombOnTomb(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
require.False(t, res.Exists()) require.False(t, res.Exists())
} }
func TestInhumeLocked(t *testing.T) {
db := newDB(t)
locked := *objecttest.Address()
err := db.Lock(*locked.ContainerID(), *oidtest.ID(), []oid.ID{*locked.ObjectID()})
require.NoError(t, err)
var prm meta.InhumePrm
prm.WithAddresses(&locked)
_, err = db.Inhume(&prm)
var e apistatus.ObjectLocked
require.ErrorAs(t, err, &e)
}