[#546] engine/inhume: Fix incorrect Inhume behavior for root objects
If object to be inhumed is root we need to continue first traverse over the shards. In case when several children are stored in different shards, inhuming object in a single shard leads to appearance of inhumed object in subsequent selections. Also, any object can be already inhumed, and this case is equivalent to successful inhume. Do not fail on `object.ErrAlreadyRemoved` error. Continue first iterating over shards if we detected root object (`SplitInfoError`). Signed-off-by: Leonard Lyubich <leonard@nspcc.ru>
This commit is contained in:
parent
c5dae76c7d
commit
1b51dcf8a4
1 changed files with 28 additions and 6 deletions
|
@ -5,6 +5,7 @@ import (
|
||||||
"errors"
|
"errors"
|
||||||
|
|
||||||
objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object"
|
objectSDK "github.com/nspcc-dev/neofs-api-go/pkg/object"
|
||||||
|
"github.com/nspcc-dev/neofs-node/pkg/core/object"
|
||||||
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
|
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
@ -75,12 +76,30 @@ func (e *StorageEngine) Inhume(prm *InhumePrm) (*InhumeRes, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (e *StorageEngine) inhume(addr *objectSDK.Address, prm *shard.InhumePrm, checkExists bool) (ok bool) {
|
func (e *StorageEngine) inhume(addr *objectSDK.Address, prm *shard.InhumePrm, checkExists bool) (ok bool) {
|
||||||
|
root := false
|
||||||
|
|
||||||
e.iterateOverSortedShards(addr, func(_ int, sh *shard.Shard) (stop bool) {
|
e.iterateOverSortedShards(addr, func(_ int, sh *shard.Shard) (stop bool) {
|
||||||
|
defer func() {
|
||||||
|
// if object is root we continue since information about it
|
||||||
|
// can be presented in other shards
|
||||||
|
if checkExists && root {
|
||||||
|
stop = false
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
if checkExists {
|
if checkExists {
|
||||||
exRes, err := sh.Exists(new(shard.ExistsPrm).
|
exRes, err := sh.Exists(new(shard.ExistsPrm).
|
||||||
WithAddress(addr),
|
WithAddress(addr),
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
if errors.Is(err, object.ErrAlreadyRemoved) {
|
||||||
|
// inhumed once - no need to be inhumed again
|
||||||
|
ok = true
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
var siErr *objectSDK.SplitInfoError
|
||||||
|
if !errors.As(err, &siErr) {
|
||||||
// TODO: smth wrong with shard, need to be processed
|
// TODO: smth wrong with shard, need to be processed
|
||||||
e.log.Warn("could not check for presents in shard",
|
e.log.Warn("could not check for presents in shard",
|
||||||
zap.Stringer("shard", sh.ID()),
|
zap.Stringer("shard", sh.ID()),
|
||||||
|
@ -88,6 +107,9 @@ func (e *StorageEngine) inhume(addr *objectSDK.Address, prm *shard.InhumePrm, ch
|
||||||
)
|
)
|
||||||
|
|
||||||
return
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
root = true
|
||||||
} else if !exRes.Exists() {
|
} else if !exRes.Exists() {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue