[#222] Update Put method in storage engine

Signed-off-by: Alex Vanin <alexey@nspcc.ru>

pkg/local_object_storage/engine/delete.go

@@ -33,7 +33,7 @@ func (e *StorageEngine) Delete(prm *DeletePrm) (*DeleteRes, error) {
 	shPrm := new(shard.DeletePrm).
 		WithAddress(prm.addr)
 
-	e.iterateOverSortedShards(prm.addr, func(sh *shard.Shard) (stop bool) {
+	e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) {
 		_, err := sh.Delete(shPrm)
 		if err != nil {
 			// TODO: smth wrong with shard, need to be processed

pkg/local_object_storage/engine/get.go

@@ -69,7 +69,7 @@ func (e *StorageEngine) Get(prm *GetPrm) (*GetRes, error) {
 		shPrm = shPrm.WithRange(prm.off, int64(prm.ln))
 	}
 
-	e.iterateOverSortedShards(prm.addr, func(sh *shard.Shard) (stop bool) {
+	e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) {
 		res, err := sh.Get(shPrm)
 		if err != nil {
 			if !errors.Is(err, object.ErrNotFound) {

pkg/local_object_storage/engine/head.go

@@ -48,7 +48,7 @@ func (e *StorageEngine) Head(prm *HeadPrm) (*HeadRes, error) {
 	shPrm := new(shard.GetPrm).
 		WithAddress(prm.addr)
 
-	e.iterateOverSortedShards(prm.addr, func(sh *shard.Shard) (stop bool) {
+	e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) {
 		res, err := sh.Get(shPrm)
 		if err != nil {
 			if !errors.Is(err, object.ErrNotFound) {

pkg/local_object_storage/engine/inhume.go

@@ -30,7 +30,7 @@ func (p *InhumePrm) WithTarget(addr, tombstone *objectSDK.Address) *InhumePrm {
 func (e *StorageEngine) Inhume(prm *InhumePrm) (*InhumeRes, error) {
 	shPrm := new(shard.InhumePrm).WithTarget(prm.addr, prm.tombstone)
 
-	e.iterateOverSortedShards(prm.addr, func(sh *shard.Shard) (stop bool) {
+	e.iterateOverSortedShards(prm.addr, func(_ int, sh *shard.Shard) (stop bool) {
 		_, err := sh.Inhume(shPrm)
 		if err != nil {
 			// TODO: smth wrong with shard, need to be processed

pkg/local_object_storage/engine/put.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 
 	"github.com/nspcc-dev/neofs-node/pkg/core/object"
+	meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase/v2"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
 	"go.uber.org/zap"
 )
@@ -34,60 +35,78 @@ func (p *PutPrm) WithObject(obj *object.Object) *PutPrm {
 // Returns any error encountered that
 // did not allow to completely save the object.
 func (e *StorageEngine) Put(prm *PutPrm) (*PutRes, error) {
-	// choose shards through sorting by weight
+	alreadyRemoved := false // first check if object has not been marked as removed
-	sortedShards := e.sortShardsByWeight(prm.obj.Address())
 
-	// check object existence
+	existPrm := new(shard.ExistsPrm)
-	if e.objectExists(prm.obj, sortedShards) {
+	existPrm.WithAddress(prm.obj.Address())
-		return nil, nil
+
+	// todo: make this check parallel
+	e.iterateOverUnsortedShards(func(s *shard.Shard) (stop bool) {
+		_, err := s.Exists(existPrm)
+		if err != nil && errors.Is(err, meta.ErrAlreadyRemoved) {
+			alreadyRemoved = true
+
+			return true
 		}
 
-	shPrm := new(shard.PutPrm)
+		return false
+	})
 
-	// save the object into the "largest" possible shard
+	if alreadyRemoved {
-	for _, sh := range sortedShards {
+		return nil, meta.ErrAlreadyRemoved
-		_, err := sh.sh.Put(
+	}
-			shPrm.WithObject(prm.obj),
-		)
 
+	finished := false
+
+	e.iterateOverSortedShards(prm.obj.Address(), func(ind int, s *shard.Shard) (stop bool) {
+		exists, err := s.Exists(existPrm)
 		if err != nil {
-			// TODO: smth wrong with shard, need to be processed
+			return false // this is not ErrAlreadyRemoved error so we can go to the next shard
-			e.log.Warn("could not save object in shard",
+		}
-				zap.Stringer("shard", sh.sh.ID()),
+
+		if exists.Exists() {
+			if ind != 0 {
+				toMoveItPrm := new(shard.ToMoveItPrm)
+				toMoveItPrm.WithAddress(prm.obj.Address())
+
+				_, err = s.ToMoveIt(toMoveItPrm)
+				if err != nil {
+					e.log.Warn("could not mark object for shard relocation",
+						zap.Stringer("shard", s.ID()),
 						zap.String("error", err.Error()),
 					)
-		} else {
-			return nil, nil
 				}
 			}
 
-	return nil, errPutShard
+			finished = true
-}
 
-func (e *StorageEngine) objectExists(obj *object.Object, shards []hashedShard) bool {
+			return true
-	exists := false
+		}
 
-	for _, sh := range shards {
+		putPrm := new(shard.PutPrm)
-		res, err := sh.sh.Exists(
+		putPrm.WithObject(prm.obj)
-			new(shard.ExistsPrm).
-				WithAddress(obj.Address()),
-		)
 
+		_, err = s.Put(putPrm)
 		if err != nil {
-			// TODO: smth wrong with shard, need to be processed
+			e.log.Warn("could not put object in shard",
-			e.log.Warn("could not check object existence",
+				zap.Stringer("shard", s.ID()),
 				zap.String("error", err.Error()),
 			)
 
-			continue
+			return false
 		}
 
-		if exists = res.Exists(); exists {
+		finished = true
-			break
+		return true
-		}
+	})
+
+	var err error = nil
+
+	if !finished {
+		err = errPutShard
 	}
 
-	return exists
+	return nil, err
 }
 
 // Put writes provided object to local storage.

pkg/local_object_storage/engine/select.go

@@ -39,7 +39,7 @@ func (e *StorageEngine) Select(prm *SelectPrm) (*SelectRes, error) {
 	shPrm := new(shard.SelectPrm).
 		WithFilters(prm.filters)
 
-	e.iterateOverSortedShards(nil, func(sh *shard.Shard) (stop bool) {
+	e.iterateOverSortedShards(nil, func(_ int, sh *shard.Shard) (stop bool) {
 		res, err := sh.Select(shPrm)
 		if err != nil {
 			// TODO: smth wrong with shard, need to be processed

pkg/local_object_storage/engine/shards.go

@@ -71,8 +71,29 @@ func (e *StorageEngine) sortShardsByWeight(objAddr fmt.Stringer) []hashedShard {
 	return shards
 }
 
-func (e *StorageEngine) iterateOverSortedShards(addr *object.Address, handler func(*shard.Shard) (stop bool)) {
+func (e *StorageEngine) unsortedShards() []hashedShard {
-	for _, sh := range e.sortShardsByWeight(addr) {
+	e.mtx.RLock()
+	defer e.mtx.RUnlock()
+
+	shards := make([]hashedShard, 0, len(e.shards))
+
+	for _, sh := range e.shards {
+		shards = append(shards, hashedShard{sh})
+	}
+
+	return shards
+}
+
+func (e *StorageEngine) iterateOverSortedShards(addr *object.Address, handler func(int, *shard.Shard) (stop bool)) {
+	for i, sh := range e.sortShardsByWeight(addr) {
+		if handler(i, sh.sh) {
+			break
+		}
+	}
+}
+
+func (e *StorageEngine) iterateOverUnsortedShards(handler func(*shard.Shard) (stop bool)) {
+	for _, sh := range e.unsortedShards() {
 		if handler(sh.sh) {
 			break
 		}