From 63bc3aab82ef9346a62c44f1e2dc864c43ce9b00 Mon Sep 17 00:00:00 2001
From: Alex Vanin <alexey@nspcc.ru>
Date: Thu, 10 Dec 2020 17:12:26 +0300
Subject: [PATCH] [#242] engine: Do not iterate over shards if cid is not set

Signed-off-by: Alex Vanin <alexey@nspcc.ru>
---
 pkg/local_object_storage/engine/select.go | 26 +++++++++++++++++------
 1 file changed, 20 insertions(+), 6 deletions(-)

diff --git a/pkg/local_object_storage/engine/select.go b/pkg/local_object_storage/engine/select.go
index c76f8bdbcb..90e86e232e 100644
--- a/pkg/local_object_storage/engine/select.go
+++ b/pkg/local_object_storage/engine/select.go
@@ -1,8 +1,11 @@
 package engine
 
 import (
+	"errors"
+
 	"github.com/nspcc-dev/neofs-api-go/pkg/container"
 	"github.com/nspcc-dev/neofs-api-go/pkg/object"
+	meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
 	"go.uber.org/zap"
 )
@@ -48,6 +51,8 @@ func (e *StorageEngine) Select(prm *SelectPrm) (*SelectRes, error) {
 	addrList := make([]*object.Address, 0)
 	uniqueMap := make(map[string]struct{})
 
+	var outError error
+
 	shPrm := new(shard.SelectPrm).
 		WithContainerID(prm.cid).
 		WithFilters(prm.filters)
@@ -55,11 +60,20 @@ func (e *StorageEngine) Select(prm *SelectPrm) (*SelectRes, error) {
 	e.iterateOverUnsortedShards(func(sh *shard.Shard) (stop bool) {
 		res, err := sh.Select(shPrm)
 		if err != nil {
-			// TODO: smth wrong with shard, need to be processed
-			e.log.Warn("could not select objects from shard",
-				zap.Stringer("shard", sh.ID()),
-				zap.String("error", err.Error()),
-			)
+			switch {
+			case errors.Is(err, meta.ErrMissingContainerID): // should never happen
+				e.log.Error("missing container ID parameter")
+				outError = err
+
+				return true
+			default:
+				// TODO: smth wrong with shard, need to be processed
+				e.log.Warn("could not select objects from shard",
+					zap.Stringer("shard", sh.ID()),
+					zap.String("error", err.Error()),
+				)
+				return false
+			}
 		} else {
 			for _, addr := range res.AddressList() { // save only unique values
 				if _, ok := uniqueMap[addr.String()]; !ok {
@@ -74,7 +88,7 @@ func (e *StorageEngine) Select(prm *SelectPrm) (*SelectRes, error) {
 
 	return &SelectRes{
 		addrList: addrList,
-	}, nil
+	}, outError
 }
 
 // List returns `limit` available physically storage object addresses in engine.