statestore: add initial implementation (#68)

* statestore: add initial implementation
Co-authored-by: Petar Radovic <petar.radovic@gmail.com>

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/sirupsen/logrus v1.4.2
 	github.com/spf13/cobra v0.0.5
 	github.com/spf13/viper v1.6.2
+	github.com/syndtr/goleveldb v1.0.0
 	github.com/uber/jaeger-client-go v2.22.1+incompatible
 	github.com/uber/jaeger-lib v2.2.0+incompatible // indirect
 	golang.org/x/crypto v0.0.0-20191219195013-becbf705a915

go.sum

@@ -84,6 +84,7 @@ github.com/golang/protobuf v1.3.0/go.mod h1:Qd/q+1AKNOZr9uGQzbzCmRO6sUih6GTPZv6a
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db h1:woRePGFeVFfLKN/pOkfl+p/TAqKOfFu+7KPlMVpok/w=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -115,6 +116,7 @@ github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpO
 github.com/huin/goupnp v1.0.0 h1:wg75sLpL6DZqwHQN6E1Cfk6mtfzS45z8OV+ic+DtHRo=
 github.com/huin/goupnp v1.0.0/go.mod h1:n9v9KO1tAxYH82qOn+UTIFQDmx5n1Zxd/ClZDMX7Bnc=
 github.com/huin/goutil v0.0.0-20170803182201-1ca381bf3150/go.mod h1:PpLOETDnJ0o3iZrZfqZzyLl6l7F3c6L1oWn7OICBi6o=
+github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/ipfs/go-cid v0.0.1/go.mod h1:GHWU/WuQdMPmIosc4Yn1bcCT7dSeX4lBafM7iqUPQvM=
 github.com/ipfs/go-cid v0.0.2/go.mod h1:GHWU/WuQdMPmIosc4Yn1bcCT7dSeX4lBafM7iqUPQvM=
@@ -160,6 +162,7 @@ github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvW
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
+github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/koron/go-ssdp v0.0.0-20180514024734-4a0ed625a78b/go.mod h1:5Ky9EC2xfoUKUor0Hjgi2BJhCSXJfMOFlmyYrVKGQMk=
 github.com/koron/go-ssdp v0.0.0-20191105050749-2e1c40ed0b5d h1:68u9r4wEvL3gYg2jvAOgROwZ3H+Y3hIDk4tbbmIjcYQ=
@@ -256,6 +259,7 @@ github.com/libp2p/go-nat v0.0.4 h1:KbizNnq8YIf7+Hn7+VFL/xE0eDrkPru2zIO9NMwL8UQ=
 github.com/libp2p/go-nat v0.0.4/go.mod h1:Nmw50VAvKuk38jUBcmNh6p9lUJLoODbJRvYAa/+KSDo=
 github.com/libp2p/go-openssl v0.0.2/go.mod h1:v8Zw2ijCSWBQi8Pq5GAixw6DbFfa9u6VIYDXnvOXkc0=
 github.com/libp2p/go-openssl v0.0.3/go.mod h1:unDrJpgy3oFr+rqXsarWifmJuNnJR4chtO1HmaZjggc=
+github.com/libp2p/go-openssl v0.0.4 h1:d27YZvLoTyMhIN4njrkr8zMDOM4lfpHIp6A+TK9fovg=
 github.com/libp2p/go-openssl v0.0.4/go.mod h1:unDrJpgy3oFr+rqXsarWifmJuNnJR4chtO1HmaZjggc=
 github.com/libp2p/go-reuseport v0.0.1 h1:7PhkfH73VXfPJYKQ6JwS5I/eVcoyYi9IMNGc6FWpFLw=
 github.com/libp2p/go-reuseport v0.0.1/go.mod h1:jn6RmB1ufnQwl0Q1f+YxAj8isJgDCQzaaxIFYDhcYEA=
@@ -400,6 +404,7 @@ github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9
 github.com/smola/gocompat v0.2.0/go.mod h1:1B0MlxbmoZNo3h8guHp8HztB3BSYR5itql9qtVc0ypY=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/spacemonkeygo/openssl v0.0.0-20181017203307-c2dcc5cca94a/go.mod h1:7AyxJNCJ7SBZ1MfVQCWD6Uqo2oubI2Eq2y2eqf+A5r0=
+github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 h1:RC6RW7j+1+HkWaX/Yh71Ee5ZHaHYt7ZP4sQgUrm6cDU=
 github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572/go.mod h1:w0SWMsp6j9O/dk4/ZpIhL+3CkG8ofA2vuv7k+ltqUMc=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
@@ -425,6 +430,7 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/subosito/gotenv v1.2.0 h1:Slr1R9HxAlEKefgq5jn9U+DnETlIUa6HfgEzj0g5d7s=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
+github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE=
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
 github.com/tdewolff/minify/v2 v2.6.1/go.mod h1:l9hbQnH096st77OkscoRUvKdd23oUM6pDZpYx381sPo=
 github.com/tdewolff/parse/v2 v2.3.14/go.mod h1:+V2lSZ93xpH2Csfs/vtNY1Fjr8kcFMsZKjyLoSkZbM0=

pkg/statestore/store.go

+// Copyright 2020 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.package storage
+
+package statestore
+
+import (
+	"encoding"
+	"encoding/json"
+
+	"github.com/ethersphere/bee/pkg/storage"
+	"github.com/syndtr/goleveldb/leveldb"
+	"github.com/syndtr/goleveldb/leveldb/util"
+)
+
+var _ storage.StateStorer = (*Store)(nil)
+
+// Store uses LevelDB to store values.
+type Store struct {
+	db *leveldb.DB
+}
+
+// New creates a new persistent state storage.
+func New(path string) (storage.StateStorer, error) {
+	db, err := leveldb.OpenFile(path, nil)
+	if err != nil {
+		return nil, err
+	}
+	return &Store{
+		db: db,
+	}, nil
+}
+
+// Get retrieves a value of the requested key. If no results are found,
+// storage.ErrNotFound will be returned.
+func (s *Store) Get(key string, i interface{}) error {
+	data, err := s.db.Get([]byte(key), nil)
+	if err != nil {
+		if err == leveldb.ErrNotFound {
+			return storage.ErrNotFound
+		}
+		return err
+	}
+
+	if unmarshaler, ok := i.(encoding.BinaryUnmarshaler); ok {
+		return unmarshaler.UnmarshalBinary(data)
+	}
+
+	return json.Unmarshal(data, i)
+}
+
+// Put stores a value for an arbitrary key. BinaryMarshaler
+// interface method will be called on the provided value
+// with fallback to JSON serialization.
+func (s *Store) Put(key string, i interface{}) (err error) {
+	var bytes []byte
+	if marshaler, ok := i.(encoding.BinaryMarshaler); ok {
+		if bytes, err = marshaler.MarshalBinary(); err != nil {
+			return err
+		}
+	} else if bytes, err = json.Marshal(i); err != nil {
+		return err
+	}
+	return s.db.Put([]byte(key), bytes, nil)
+}
+
+// Delete removes entries stored under a specific key.
+func (s *Store) Delete(key string) (err error) {
+	return s.db.Delete([]byte(key), nil)
+}
+
+// Iterate entries that match the supplied prefix.
+func (s *Store) Iterate(prefix string, iterFunc storage.StateIterFunc) (err error) {
+	iter := s.db.NewIterator(util.BytesPrefix([]byte(prefix)), nil)
+	defer iter.Release()
+	for iter.Next() {
+		stop, err := iterFunc(iter.Key(), iter.Value())
+		if err != nil {
+			return err
+		}
+		if stop {
+			break
+		}
+	}
+	return iter.Error()
+}
+
+// Close releases the resources used by the store.
+func (s *Store) Close() error {
+	return s.db.Close()
+}

pkg/statestore/test/store_test.go

+// Copyright 2020 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.package storage
+
+package test
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"reflect"
+	"testing"
+
+	"github.com/ethersphere/bee/pkg/statestore"
+	"github.com/ethersphere/bee/pkg/storage"
+)
+
+type Serializing struct {
+	value           string
+	marshalCalled   bool
+	unmarshalCalled bool
+}
+
+func (st *Serializing) MarshalBinary() (data []byte, err error) {
+	d := []byte(st.value)
+	st.marshalCalled = true
+
+	return d, nil
+}
+
+func (st *Serializing) UnmarshalBinary(data []byte) (err error) {
+	st.value = string(data)
+	st.unmarshalCalled = true
+	return nil
+}
+
+func TestStore(t *testing.T) {
+	dir, err := ioutil.TempDir("", "statestore_test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(dir)
+
+	const (
+		key1 = "key1" // stores the serialized type
+		key2 = "key2" // stores a json array
+	)
+
+	var (
+		value1 = &Serializing{value: "value1"}
+		value2 = []string{"a", "b", "c"}
+	)
+
+	// create a new persisted store
+	store, err := statestore.New(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// insert some values
+	insertValues(t, store, key1, key2, value1, value2)
+
+	// close the persisted store
+	store.Close()
+
+	// bootstrap a new store with the persisted data
+	persistedStore, err := statestore.New(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer persistedStore.Close()
+
+	// check that the persisted values match
+	testPersistedValues(t, persistedStore, key1, key2, value1, value2)
+
+	// test that the iterator works
+	testStoreIterator(t, persistedStore)
+}
+
+func insertValues(t *testing.T, store storage.StateStorer, key1, key2 string, value1 *Serializing, value2 []string) {
+	err := store.Put(key1, value1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !value1.marshalCalled {
+		t.Fatal("binaryMarshaller not called on serialized type")
+	}
+
+	err = store.Put(key2, value2)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func testPersistedValues(t *testing.T, store storage.StateStorer, key1, key2 string, value1 *Serializing, value2 []string) {
+	v := &Serializing{}
+	err := store.Get(key1, v)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !v.unmarshalCalled {
+		t.Fatal("unmarshaler not called")
+	}
+
+	if v.value != value1.value {
+		t.Fatalf("expected persisted to be %s but got %s", value1.value, v.value)
+	}
+
+	s := []string{}
+	err = store.Get(key2, &s)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for i, ss := range value2 {
+		if s[i] != ss {
+			t.Fatalf("deserialized data mismatch. expected %s but got %s", ss, s[i])
+		}
+	}
+}
+
+func testStoreIterator(t *testing.T, store storage.StateStorer) {
+	storePrefix := "test_"
+	err := store.Put(storePrefix+"key1", "value1")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// do not include prefix in one of the entries
+	err = store.Put("key2", "value2")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = store.Put(storePrefix+"key3", "value3")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	entries := make(map[string]string)
+
+	entriesIterFunction := func(key []byte, value []byte) (stop bool, err error) {
+		var entry string
+		err = json.Unmarshal(value, &entry)
+		if err != nil {
+			t.Fatal(err)
+		}
+		entries[string(key)] = entry
+		return stop, err
+	}
+
+	err = store.Iterate(storePrefix, entriesIterFunction)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectedEntries := map[string]string{"test_key1": "value1", "test_key3": "value3"}
+
+	if !reflect.DeepEqual(entries, expectedEntries) {
+		t.Fatalf("expected store entries to be %v, are %v instead", expectedEntries, entries)
+	}
+}

pkg/storage/store.go

@@ -17,9 +17,22 @@ var (
 )
 
 // ChunkValidatorFunc validates Swarm chunk address and chunk data
-type ChunkValidatorFunc func(swarm.Address, []byte) bool
+type ChunkValidatorFunc func(swarm.Address, []byte) (valid bool)
 
 type Storer interface {
 	Get(ctx context.Context, addr swarm.Address) (data []byte, err error)
-	Put(ctx context.Context, addr swarm.Address, data []byte) error
+	Put(ctx context.Context, addr swarm.Address, data []byte) (err error)
 }
+
+// StateStorer defines methods required to get, set, delete values for different keys
+// and close the underlying resources.
+type StateStorer interface {
+	Get(key string, i interface{}) (err error)
+	Put(key string, i interface{}) (err error)
+	Delete(key string) (err error)
+	Iterate(prefix string, iterFunc StateIterFunc) (err error)
+	Close() (err error)
+}
+
+// StateIterFunc is used when iterating through StateStorer key/value pairs
+type StateIterFunc func(key, value []byte) (stop bool, err error)