Localstore integration

cmd/swarm/config.go

@@ -252,15 +252,15 @@ func cmdLineOverride(currentConfig *bzzapi.Config, ctx *cli.Context) *bzzapi.Con
 	}
 
 	if storePath := ctx.GlobalString(SwarmStorePath.Name); storePath != "" {
-		currentConfig.LocalStoreParams.ChunkDbPath = storePath
+		currentConfig.ChunkDbPath = storePath
 	}
 
 	if storeCapacity := ctx.GlobalUint64(SwarmStoreCapacity.Name); storeCapacity != 0 {
-		currentConfig.LocalStoreParams.DbCapacity = storeCapacity
+		currentConfig.DbCapacity = storeCapacity
 	}
 
 	if ctx.GlobalIsSet(SwarmStoreCacheCapacity.Name) {
-		currentConfig.LocalStoreParams.CacheCapacity = ctx.GlobalUint(SwarmStoreCacheCapacity.Name)
+		currentConfig.CacheCapacity = ctx.GlobalUint(SwarmStoreCacheCapacity.Name)
 	}
 
 	if ctx.GlobalIsSet(SwarmBootnodeModeFlag.Name) {

cmd/swarm/config_test.go

@@ -447,8 +447,8 @@ func TestConfigCmdLineOverridesFile(t *testing.T) {
 		t.Fatal("Expected Sync to be disabled, but is true")
 	}
 
-	if info.LocalStoreParams.DbCapacity != 9000000 {
-		t.Fatalf("Expected Capacity to be %d, got %d", 9000000, info.LocalStoreParams.DbCapacity)
+	if info.DbCapacity != 9000000 {
+		t.Fatalf("Expected Capacity to be %d, got %d", 9000000, info.DbCapacity)
 	}
 
 	if info.HiveParams.KeepAliveInterval != 6000000000 {

cmd/swarm/db.go

@@ -25,7 +25,7 @@ import (
 	"github.com/ethereum/go-ethereum/cmd/utils"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/swarm/storage"
+	"github.com/ethereum/go-ethereum/swarm/storage/localstore"
 	"gopkg.in/urfave/cli.v1"
 )
 
@@ -135,13 +135,10 @@ func dbImport(ctx *cli.Context) {
 	log.Info(fmt.Sprintf("successfully imported %d chunks", count))
 }
 
-func openLDBStore(path string, basekey []byte) (*storage.LDBStore, error) {
+func openLDBStore(path string, basekey []byte) (*localstore.DB, error) {
 	if _, err := os.Stat(filepath.Join(path, "CURRENT")); err != nil {
 		return nil, fmt.Errorf("invalid chunkdb path: %s", err)
 	}
 
-	storeparams := storage.NewDefaultStoreParams()
-	ldbparams := storage.NewLDBStoreParams(storeparams, path)
-	ldbparams.BaseKey = basekey
-	return storage.NewLDBStore(ldbparams)
+	return localstore.New(path, basekey, nil)
 }

swarm/api/config.go

@@ -45,7 +45,13 @@ const (
 type Config struct {
 	// serialised/persisted fields
 	*storage.FileStoreParams
-	*storage.LocalStoreParams
+
+	// LocalStore
+	ChunkDbPath   string
+	DbCapacity    uint64
+	CacheCapacity uint
+	BaseKey       []byte
+
 	*network.HiveParams
 	Swap                 *swap.LocalProfile
 	Pss                  *pss.PssParams
@@ -78,7 +84,6 @@ type Config struct {
 func NewConfig() (c *Config) {
 
 	c = &Config{
-		LocalStoreParams:     storage.NewDefaultLocalStoreParams(),
 		FileStoreParams:      storage.NewFileStoreParams(),
 		HiveParams:           network.NewHiveParams(),
 		Swap:                 swap.NewDefaultSwapParams(),
@@ -130,8 +135,9 @@ func (c *Config) Init(prvKey *ecdsa.PrivateKey, nodeKey *ecdsa.PrivateKey) error
 		c.Swap.Init(c.Contract, prvKey)
 	}
 
-	c.LocalStoreParams.Init(c.Path)
-	c.LocalStoreParams.BaseKey = common.FromHex(c.BzzKey)
+	c.privateKey = prvKey
+	c.ChunkDbPath = filepath.Join(c.Path, "chunks")
+	c.BaseKey = common.FromHex(c.BzzKey)
 
 	c.Pss = c.Pss.WithPrivateKey(c.privateKey)
 	return nil

swarm/api/config_test.go

@@ -41,7 +41,6 @@ func TestConfig(t *testing.T) {
 	one := NewConfig()
 	two := NewConfig()
 
-	one.LocalStoreParams = two.LocalStoreParams
 	if equal := reflect.DeepEqual(one, two); !equal {
 		t.Fatal("Two default configs are not equal")
 	}

swarm/api/http/test_server.go

@@ -26,6 +26,7 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/api"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/ethereum/go-ethereum/swarm/storage/feed"
+	"github.com/ethereum/go-ethereum/swarm/storage/localstore"
 )
 
 type TestServer interface {
@@ -37,15 +38,12 @@ func NewTestSwarmServer(t *testing.T, serverFunc func(*api.API) TestServer, reso
 	if err != nil {
 		t.Fatal(err)
 	}
-	storeparams := storage.NewDefaultLocalStoreParams()
-	storeparams.DbCapacity = 5000000
-	storeparams.CacheCapacity = 5000
-	storeparams.Init(dir)
-	localStore, err := storage.NewLocalStore(storeparams, nil)
+	localStore, err := localstore.New(dir, make([]byte, 32), nil)
 	if err != nil {
 		os.RemoveAll(dir)
 		t.Fatal(err)
 	}
+
 	fileStore := storage.NewFileStore(localStore, storage.NewFileStoreParams())
 
 	// Swarm feeds test setup

swarm/api/inspector.go

@@ -60,7 +60,11 @@ func (inspector *Inspector) Has(chunkAddresses []storage.Address) []HasInfo {
 	for _, addr := range chunkAddresses {
 		res := HasInfo{}
 		res.Addr = addr.String()
-		res.Has = inspector.netStore.Has(context.Background(), addr)
+		has, err := inspector.netStore.Has(context.Background(), addr)
+		if err != nil {
+			has = false
+		}
+		res.Has = has
 		results = append(results, res)
 	}
 	return results

swarm/chunk/chunk.go

@@ -1,6 +1,7 @@
 package chunk
 
 import (
+	"context"
 	"errors"
 	"fmt"
 
@@ -28,7 +29,7 @@ type chunk struct {
 	sdata []byte
 }
 
-func NewChunk(addr Address, data []byte) *chunk {
+func NewChunk(addr Address, data []byte) Chunk {
 	return &chunk{
 		addr:  addr,
 		sdata: data,
@@ -107,3 +108,105 @@ func Proximity(one, other []byte) (ret int) {
 	}
 	return MaxPO
 }
+
+// ModeGet enumerates different Getter modes.
+type ModeGet int
+
+// Getter modes.
+const (
+	// ModeGetRequest: when accessed for retrieval
+	ModeGetRequest ModeGet = iota
+	// ModeGetSync: when accessed for syncing or proof of custody request
+	ModeGetSync
+	// ModeGetLookup: when accessed to lookup a a chunk in feeds or other places
+	ModeGetLookup
+)
+
+// ModePut enumerates different Putter modes.
+type ModePut int
+
+// Putter modes.
+const (
+	// ModePutRequest: when a chunk is received as a result of retrieve request and delivery
+	ModePutRequest ModePut = iota
+	// ModePutSync: when a chunk is received via syncing
+	ModePutSync
+	// ModePutUpload: when a chunk is created by local upload
+	ModePutUpload
+)
+
+// ModeSet enumerates different Setter modes.
+type ModeSet int
+
+// Setter modes.
+const (
+	// ModeSetAccess: when an update request is received for a chunk or chunk is retrieved for delivery
+	ModeSetAccess ModeSet = iota
+	// ModeSetSync: when push sync receipt is received
+	ModeSetSync
+	// ModeSetRemove: when a chunk is removed
+	ModeSetRemove
+)
+
+// Descriptor holds information required for Pull syncing. This struct
+// is provided by subscribing to pull index.
+type Descriptor struct {
+	Address Address
+	BinID   uint64
+}
+
+func (d *Descriptor) String() string {
+	if d == nil {
+		return ""
+	}
+	return fmt.Sprintf("%s bin id %v", d.Address.Hex(), d.BinID)
+}
+
+type Store interface {
+	Get(ctx context.Context, mode ModeGet, addr Address) (ch Chunk, err error)
+	Put(ctx context.Context, mode ModePut, ch Chunk) (err error)
+	Has(ctx context.Context, addr Address) (yes bool, err error)
+	Set(ctx context.Context, mode ModeSet, addr Address) (err error)
+	LastPullSubscriptionBinID(bin uint8) (id uint64, err error)
+	SubscribePull(ctx context.Context, bin uint8, since, until uint64) (c <-chan Descriptor, stop func())
+	Close() (err error)
+}
+
+// FetchStore is a Store which supports syncing
+type FetchStore interface {
+	Store
+	FetchFunc(ctx context.Context, addr Address) func(context.Context) error
+}
+
+// Validator validates a chunk.
+type Validator interface {
+	Validate(ch Chunk) bool
+}
+
+// ValidatorStore encapsulates Store by decorting the Put method
+// with validators check.
+type ValidatorStore struct {
+	Store
+	validators []Validator
+}
+
+// NewValidatorStore returns a new ValidatorStore which uses
+// provided validators to validate chunks on Put.
+func NewValidatorStore(store Store, validators ...Validator) (s *ValidatorStore) {
+	return &ValidatorStore{
+		Store:      store,
+		validators: validators,
+	}
+}
+
+// Put overrides Store put method with validators check. If one of the validators
+// return true, the chunk is considered valid and Store Put method is called.
+// If all validators return false, ErrChunkInvalid is returned.
+func (s *ValidatorStore) Put(ctx context.Context, mode ModePut, ch Chunk) (err error) {
+	for _, v := range s.validators {
+		if v.Validate(ch) {
+			return s.Store.Put(ctx, mode, ch)
+		}
+	}
+	return ErrChunkInvalid
+}