Merge 1072773 into 2c315db

storage/boltdb/client.go

@@ -5,7 +5,6 @@ package boltdb
 
 import (
 	"bytes"
-	"fmt"
 	"time"
 
 	"storj.io/storj/pkg/utils"
@@ -24,7 +23,6 @@ type Client struct {
 const (
 	// fileMode sets permissions so owner can read and write
 	fileMode       = 0600
-	maxKeyLookup   = 100
 	defaultTimeout = 1 * time.Second
 )
 
@@ -116,12 +114,11 @@ func (client *Client) Close() error {
 // GetAll finds all values for the provided keys up to 100 keys
 // if more keys are provided than the maximum an error will be returned.
 func (client *Client) GetAll(keys storage.Keys) (storage.Values, error) {
-	lk := len(keys)
-	if lk > maxKeyLookup {
-		return nil, Error.New(fmt.Sprintf("requested %d keys, maximum is %d", lk, maxKeyLookup))
+	if len(keys) > storage.LookupLimit {
+		return nil, storage.ErrLimitExceeded
 	}
 
-	vals := make(storage.Values, 0, lk)
+	vals := make(storage.Values, 0, len(keys))
 	err := client.view(func(bucket *bolt.Bucket) error {
 		for _, key := range keys {
 			val := bucket.Get([]byte(key))

storage/common.go

@@ -19,6 +19,9 @@ var ErrKeyNotFound = errs.Class("key not found")
 // ErrEmptyKey is returned when an empty key is used in Put
 var ErrEmptyKey = errors.New("empty key")
 
+// ErrLimitExceeded is returned when request limit is exceeded
+var ErrLimitExceeded = errors.New("limit exceeded")
+
 // Key is the type for the keys in a `KeyValueStore`
 type Key []byte
 
@@ -37,6 +40,9 @@ type Limit int
 // Items keeps all ListItem
 type Items []ListItem
 
+// LookupLimit is enforced by storage implementations
+const LookupLimit = 100
+
 // ListItem returns Key, Value, IsPrefix
 type ListItem struct {
 	Key      Key

storage/listkeys.go

@@ -5,14 +5,13 @@ package storage
 
 // ListKeys returns keys starting from first and upto limit
 func ListKeys(store KeyValueStore, first Key, limit Limit) (Keys, error) {
-	const unlimited = Limit(1 << 31)
-
-	keys := make(Keys, 0, limit)
-	if limit == 0 {
-		// TODO: this shouldn't be probably the case
-		limit = unlimited
+	if limit > LookupLimit {
+		return nil, ErrLimitExceeded
+	} else if limit <= 0 {
+		limit = LookupLimit
 	}
 
+	keys := make(Keys, 0, limit)
 	err := store.Iterate(IterateOptions{
 		First:   first,
 		Recurse: true,
@@ -32,14 +31,13 @@ func ListKeys(store KeyValueStore, first Key, limit Limit) (Keys, error) {
 
 // ReverseListKeys returns keys starting from first and upto limit in reverse order
 func ReverseListKeys(store KeyValueStore, first Key, limit Limit) (Keys, error) {
-	const unlimited = Limit(1 << 31)
-
-	keys := make(Keys, 0, limit)
-	if limit == 0 {
-		// TODO: this shouldn't be probably the case
-		limit = unlimited
+	if limit > LookupLimit {
+		return nil, ErrLimitExceeded
+	} else if limit <= 0 {
+		limit = LookupLimit
 	}
 
+	keys := make(Keys, 0, limit)
 	err := store.Iterate(IterateOptions{
 		First:   first,
 		Recurse: true,

storage/listv2.go

@@ -23,6 +23,12 @@ type ListOptions struct {
 
 // ListV2 lists all keys corresponding to ListOptions
 func ListV2(store KeyValueStore, opts ListOptions) (result Items, more More, err error) {
+	if opts.Limit > LookupLimit {
+		return nil, false, ErrLimitExceeded
+	} else if opts.Limit <= 0 {
+		opts.Limit = LookupLimit
+	}
+
 	if opts.StartAfter != nil && opts.EndBefore != nil {
 		return nil, false, errors.New("start-after and end-before cannot be combined")
 	}

storage/redis/client.go

@@ -4,7 +4,6 @@
 package redis
 
 import (
-	"fmt"
 	"sort"
 	"time"
 
@@ -18,10 +17,7 @@ var (
 	Error = errs.Class("redis error")
 )
 
-const (
-	defaultNodeExpiration = 61 * time.Minute
-	maxKeyLookup          = 100
-)
+const defaultNodeExpiration = 61 * time.Minute
 
 // Client is the entrypoint into Redis
 type Client struct {
@@ -101,8 +97,8 @@ func (client *Client) Close() error {
 // The maximum keys returned will be 100. If more than that is requested an
 // error will be returned
 func (client *Client) GetAll(keys storage.Keys) (storage.Values, error) {
-	if len(keys) > maxKeyLookup {
-		return nil, Error.New(fmt.Sprintf("requested %d keys, maximum is %d", len(keys), maxKeyLookup))
+	if len(keys) > storage.LookupLimit {
+		return nil, storage.ErrLimitExceeded
 	}
 
 	keyStrings := make([]string, len(keys))

storage/teststore/store.go

@@ -88,6 +88,9 @@ func (store *Client) Get(key storage.Key) (storage.Value, error) {
 // GetAll gets all values from the store
 func (store *Client) GetAll(keys storage.Keys) (storage.Values, error) {
 	store.CallCount.GetAll++
+	if len(keys) > storage.LookupLimit {
+		return nil, storage.ErrLimitExceeded
+	}
 
 	values := storage.Values{}
 	for _, key := range keys {

storage/testsuite/test.go

@@ -24,6 +24,16 @@ func RunTests(t *testing.T, store storage.KeyValueStore) {
 }
 
 func testConstraints(t *testing.T, store storage.KeyValueStore) {
+	testKey := storage.Key("test")
+	if err := store.Put(testKey, storage.Value("xyz")); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if err := store.Delete(testKey); err != nil {
+			t.Fatal(err)
+		}
+	}()
+
 	t.Run("Put Empty", func(t *testing.T) {
 		var key storage.Key
 		var val storage.Value
@@ -34,4 +44,41 @@ func testConstraints(t *testing.T, store storage.KeyValueStore) {
 			t.Fatal("putting empty key should fail")
 		}
 	})
+
+	t.Run("GetAll limit", func(t *testing.T) {
+		keys := make([]storage.Key, storage.LookupLimit+1)
+		for i := range keys {
+			keys[i] = testKey
+		}
+
+		_, err := store.GetAll(keys[:storage.LookupLimit])
+		if err != nil {
+			t.Fatalf("GetAll LookupLimit should succeed: %v", err)
+		}
+
+		_, err = store.GetAll(keys[:storage.LookupLimit+1])
+		if err == nil && err == storage.ErrLimitExceeded {
+			t.Fatalf("GetAll LookupLimit+1 should fail: %v", err)
+		}
+	})
+
+	t.Run("List limit", func(t *testing.T) {
+		_, err := store.List(nil, storage.LookupLimit)
+		if err != nil {
+			t.Fatalf("List LookupLimit should succeed: %v", err)
+		}
+		_, err = store.ReverseList(nil, storage.LookupLimit)
+		if err != nil {
+			t.Fatalf("ReverseList LookupLimit should succeed: %v", err)
+		}
+
+		_, err = store.List(nil, storage.LookupLimit+1)
+		if err == nil && err == storage.ErrLimitExceeded {
+			t.Fatalf("List LookupLimit+1 should fail: %v", err)
+		}
+		_, err = store.ReverseList(nil, storage.LookupLimit+1)
+		if err == nil && err == storage.ErrLimitExceeded {
+			t.Fatalf("ReverseList LookupLimit+1 should fail: %v", err)
+		}
+	})
 }

storage/testsuite/test_list.go

@@ -10,6 +10,7 @@ import (
 	"storj.io/storj/storage"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 )
 
 func testList(t *testing.T, store storage.KeyValueStore) {
@@ -83,7 +84,7 @@ func testList(t *testing.T, store storage.KeyValueStore) {
 			t.Errorf("%s: %s", test.Name, err)
 			continue
 		}
-		if diff := cmp.Diff(test.Expected, keys); diff != "" {
+		if diff := cmp.Diff(test.Expected, keys, cmpopts.EquateEmpty()); diff != "" {
 			t.Errorf("%s: (-want +got)\n%s", test.Name, diff)
 		}
 	}