swarm/network/stream: start of localstore integration

swarm/network/stream/common_test.go

@@ -30,15 +30,18 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
+	"github.com/ethereum/go-ethereum/swarm/chunk"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/network/simulation"
 	"github.com/ethereum/go-ethereum/swarm/state"
 	"github.com/ethereum/go-ethereum/swarm/storage"
-	mockmem "github.com/ethereum/go-ethereum/swarm/storage/mock/mem"
+	"github.com/ethereum/go-ethereum/swarm/storage/localstore"
+	"github.com/ethereum/go-ethereum/swarm/storage/mock"
 	"github.com/ethereum/go-ethereum/swarm/testutil"
 	colorable "github.com/mattn/go-colorable"
 )
@@ -50,7 +53,6 @@ var (
 	useMockStore = flag.Bool("mockstore", false, "disabled mock store (default: enabled)")
 	longrunning  = flag.Bool("longrunning", false, "do run long-running tests")
 
-	bucketKeyDB        = simulation.BucketKey("db")
 	bucketKeyStore     = simulation.BucketKey("store")
 	bucketKeyFileStore = simulation.BucketKey("filestore")
 	bucketKeyNetStore  = simulation.BucketKey("netstore")
@@ -112,16 +114,15 @@ func newNetStoreAndDeliveryWithRequestFunc(ctx *adapters.ServiceContext, bucket
 func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map, addr *network.BzzAddr) (*storage.NetStore, *Delivery, func(), error) {
 	n := ctx.Config.Node()
 
-	store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
+	localStore, localStoreCleanup, err := newTestLocalStore(n.ID(), addr, nil)
-	if *useMockStore {
-		store, datadir, err = createMockStore(mockmem.NewGlobalStore(), n.ID(), addr)
-	}
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	localStore := store.(*storage.LocalStore)
+
 	netStore, err := storage.NewNetStore(localStore, nil)
 	if err != nil {
+		localStoreCleanup()
+		localStore.Close()
 		return nil, nil, nil, err
 	}
 
@@ -130,8 +131,7 @@ func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map,
 	kad := network.NewKademlia(addr.Over(), network.NewKadParams())
 	delivery := NewDelivery(kad, netStore)
 
-	bucket.Store(bucketKeyStore, store)
+	bucket.Store(bucketKeyStore, localStore)
-	bucket.Store(bucketKeyDB, netStore)
 	bucket.Store(bucketKeyDelivery, delivery)
 	bucket.Store(bucketKeyFileStore, fileStore)
 	// for the kademlia object, we use the global key from the simulation package,
@@ -140,13 +140,13 @@ func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map,
 
 	cleanup := func() {
 		netStore.Close()
-		os.RemoveAll(datadir)
+		localStoreCleanup()
 	}
 
 	return netStore, delivery, cleanup, nil
 }
 
-func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
+func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTester, *Registry, *localstore.DB, func(), error) {
 	// setup
 	addr := network.RandomAddr() // tested peers peer address
 	to := network.NewKademlia(addr.OAddr, network.NewKadParams())
@@ -160,11 +160,7 @@ func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTeste
 		os.RemoveAll(datadir)
 	}
 
-	params := storage.NewDefaultLocalStoreParams()
+	localStore, err := localstore.New(datadir, addr.Over(), nil)
-	params.Init(datadir)
-	params.BaseKey = addr.Over()
-
-	localStore, err := storage.NewTestLocalStoreForAddr(params)
 	if err != nil {
 		removeDataDir()
 		return nil, nil, nil, nil, err
@@ -221,24 +217,25 @@ func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore {
 }
 
 // not used in this context, only to fulfill ChunkStore interface
-func (rrs *roundRobinStore) Has(ctx context.Context, addr storage.Address) bool {
+func (rrs *roundRobinStore) Has(_ context.Context, _ storage.Address) (bool, error) {
-	panic("RoundRobinStor doesn't support HasChunk")
+	return false, errors.New("RoundRobinStore doesn't support HasChunk")
 }
 
-func (rrs *roundRobinStore) Get(ctx context.Context, addr storage.Address) (storage.Chunk, error) {
+func (rrs *roundRobinStore) Get(_ context.Context, _ chunk.ModeGet, _ storage.Address) (storage.Chunk, error) {
 	return nil, errors.New("get not well defined on round robin store")
 }
 
-func (rrs *roundRobinStore) Put(ctx context.Context, chunk storage.Chunk) error {
+func (rrs *roundRobinStore) Put(ctx context.Context, mode chunk.ModePut, ch storage.Chunk) error {
 	i := atomic.AddUint32(&rrs.index, 1)
 	idx := int(i) % len(rrs.stores)
-	return rrs.stores[idx].Put(ctx, chunk)
+	return rrs.stores[idx].Put(ctx, mode, ch)
 }
 
-func (rrs *roundRobinStore) Close() {
+func (rrs *roundRobinStore) Close() error {
 	for _, store := range rrs.stores {
 		store.Close()
 	}
+	return nil
 }
 
 func readAll(fileStore *storage.FileStore, hash []byte) (int64, error) {
@@ -304,24 +301,28 @@ func generateRandomFile() (string, error) {
 	return string(b), nil
 }
 
-//create a local store for the given node
+func newTestLocalStore(id enode.ID, addr *network.BzzAddr, globalStore mock.GlobalStorer) (localStore *localstore.DB, cleanup func(), err error) {
-func createTestLocalStorageForID(id enode.ID, addr *network.BzzAddr) (storage.ChunkStore, string, error) {
+	dir, err := ioutil.TempDir("", "swarm-stream-")
-	var datadir string
-	var err error
-	datadir, err = ioutil.TempDir("", fmt.Sprintf("syncer-test-%s", id.TerminalString()))
 	if err != nil {
-		return nil, "", err
+		return nil, nil, err
 	}
-	var store storage.ChunkStore
+	cleanup = func() {
-	params := storage.NewDefaultLocalStoreParams()
+		os.RemoveAll(dir)
-	params.ChunkDbPath = datadir
+	}
-	params.BaseKey = addr.Over()
+
-	store, err = storage.NewTestLocalStoreForAddr(params)
+	var mockStore *mock.NodeStore
+	if globalStore != nil {
+		mockStore = globalStore.NewNodeStore(common.BytesToAddress(id.Bytes()))
+	}
+
+	localStore, err = localstore.New(dir, addr.Over(), &localstore.Options{
+		MockStore: mockStore,
+	})
 	if err != nil {
-		os.RemoveAll(datadir)
+		cleanup()
-		return nil, "", err
+		return nil, nil, err
 	}
-	return store, datadir, nil
+	return localStore, cleanup, nil
 }
 
 // watchDisconnections receives simulation peer events in a new goroutine and sets atomic value

swarm/network/stream/delivery_test.go

@@ -25,6 +25,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/swarm/chunk"
+
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
@@ -189,8 +191,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 	})
 
 	hash := storage.Address(hash0[:])
-	chunk := storage.NewChunk(hash, hash)
+	ch := storage.NewChunk(hash, hash)
-	err = localStore.Put(context.TODO(), chunk)
+	err = localStore.Put(context.TODO(), chunk.ModePutUpload, ch)
 	if err != nil {
 		t.Fatalf("Expected no err got %v", err)
 	}
@@ -241,8 +243,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 	}
 
 	hash = storage.Address(hash1[:])
-	chunk = storage.NewChunk(hash, hash1[:])
+	ch = storage.NewChunk(hash, hash1[:])
-	err = localStore.Put(context.TODO(), chunk)
+	err = localStore.Put(context.TODO(), chunk.ModePutUpload, ch)
 	if err != nil {
 		t.Fatalf("Expected no err got %v", err)
 	}
@@ -420,14 +422,14 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 	defer cancel()
 
 	// wait for the chunk to get stored
-	storedChunk, err := localStore.Get(ctx, chunkKey)
+	storedChunk, err := localStore.Get(ctx, chunk.ModeGetRequest, chunkKey)
 	for err != nil {
 		select {
 		case <-ctx.Done():
 			t.Fatalf("Chunk is not in localstore after timeout, err: %v", err)
 		default:
 		}
-		storedChunk, err = localStore.Get(ctx, chunkKey)
+		storedChunk, err = localStore.Get(ctx, chunk.ModeGetRequest, chunkKey)
 		time.Sleep(50 * time.Millisecond)
 	}
 
@@ -700,7 +702,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
 			errs := make(chan error)
 			for _, hash := range hashes {
 				go func(h storage.Address) {
-					_, err := netStore.Get(ctx, h)
+					_, err := netStore.Get(ctx, chunk.ModeGetRequest, h)
 					log.Warn("test check netstore get", "hash", h, "err", err)
 					errs <- err
 				}(hash)

swarm/network/stream/snapshot_retrieval_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/swarm/storage/localstore"
 	"github.com/ethereum/go-ethereum/swarm/testutil"
 
 	"github.com/ethereum/go-ethereum/node"
@@ -278,7 +279,7 @@ func runRetrievalTest(t *testing.T, chunkCount int, nodeCount int) error {
 		if !ok {
 			return fmt.Errorf("No localstore")
 		}
-		lstore := item.(*storage.LocalStore)
+		lstore := item.(*localstore.DB)
 		conf.hashes, err = uploadFileToSingleNodeStore(node.ID(), chunkCount, lstore)
 		if err != nil {
 			return err

swarm/network/stream/snapshot_sync_test.go

@@ -31,11 +31,13 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	"github.com/ethereum/go-ethereum/swarm/chunk"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/network/simulation"
 	"github.com/ethereum/go-ethereum/swarm/pot"
 	"github.com/ethereum/go-ethereum/swarm/state"
 	"github.com/ethereum/go-ethereum/swarm/storage"
+	"github.com/ethereum/go-ethereum/swarm/storage/localstore"
 	"github.com/ethereum/go-ethereum/swarm/storage/mock"
 	mockmem "github.com/ethereum/go-ethereum/swarm/storage/mock/mem"
 	"github.com/ethereum/go-ethereum/swarm/testutil"
@@ -196,7 +198,7 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
 		if !ok {
 			return fmt.Errorf("No localstore")
 		}
-		lstore := item.(*storage.LocalStore)
+		lstore := item.(*localstore.DB)
 		hashes, err := uploadFileToSingleNodeStore(node.ID(), chunkCount, lstore)
 		if err != nil {
 			return err
@@ -225,25 +227,25 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
 				localChunks := conf.idToChunksMap[id]
 				for _, ch := range localChunks {
 					//get the real chunk by the index in the index array
-					chunk := conf.hashes[ch]
+					ch := conf.hashes[ch]
-					log.Trace(fmt.Sprintf("node has chunk: %s:", chunk))
+					log.Trace(fmt.Sprintf("node has chunk: %s:", ch))
 					//check if the expected chunk is indeed in the localstore
 					var err error
 					if *useMockStore {
 						//use the globalStore if the mockStore should be used; in that case,
 						//the complete localStore stack is bypassed for getting the chunk
-						_, err = globalStore.Get(common.BytesToAddress(id.Bytes()), chunk)
+						_, err = globalStore.Get(common.BytesToAddress(id.Bytes()), ch)
 					} else {
 						//use the actual localstore
 						item, ok := sim.NodeItem(id, bucketKeyStore)
 						if !ok {
 							return fmt.Errorf("Error accessing localstore")
 						}
-						lstore := item.(*storage.LocalStore)
+						lstore := item.(*localstore.DB)
-						_, err = lstore.Get(ctx, chunk)
+						_, err = lstore.Get(ctx, chunk.ModeGetRequest, ch)
 					}
 					if err != nil {
-						log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
+						log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", ch, id))
 						// Do not get crazy with logging the warn message
 						time.Sleep(500 * time.Millisecond)
 						continue REPEAT
@@ -251,10 +253,10 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
 					evt := &simulations.Event{
 						Type: EventTypeChunkArrived,
 						Node: sim.Net.GetNode(id),
-						Data: chunk.String(),
+						Data: ch.String(),
 					}
 					sim.Net.Events().Send(evt)
-					log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id))
+					log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", ch, id))
 				}
 			}
 			return nil
@@ -300,7 +302,7 @@ func mapKeysToNodes(conf *synctestConfig) {
 }
 
 //upload a file(chunks) to a single local node store
-func uploadFileToSingleNodeStore(id enode.ID, chunkCount int, lstore *storage.LocalStore) ([]storage.Address, error) {
+func uploadFileToSingleNodeStore(id enode.ID, chunkCount int, lstore *localstore.DB) ([]storage.Address, error) {
 	log.Debug(fmt.Sprintf("Uploading to node id: %s", id))
 	fileStore := storage.NewFileStore(lstore, storage.NewFileStoreParams())
 	size := chunkSize

swarm/network/stream/syncer_test.go

@@ -21,13 +21,13 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
-	"math"
 	"os"
 	"sync"
 	"testing"
 	"time"
 
-	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/swarm/chunk"
+
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
@@ -36,7 +36,7 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/network/simulation"
 	"github.com/ethereum/go-ethereum/swarm/state"
 	"github.com/ethereum/go-ethereum/swarm/storage"
-	"github.com/ethereum/go-ethereum/swarm/storage/mock"
+	"github.com/ethereum/go-ethereum/swarm/storage/localstore"
 	"github.com/ethereum/go-ethereum/swarm/testutil"
 )
 
@@ -55,24 +55,6 @@ func TestSyncerSimulation(t *testing.T) {
 	}
 }
 
-func createMockStore(globalStore mock.GlobalStorer, id enode.ID, addr *network.BzzAddr) (lstore storage.ChunkStore, datadir string, err error) {
-	address := common.BytesToAddress(id.Bytes())
-	mockStore := globalStore.NewNodeStore(address)
-	params := storage.NewDefaultLocalStoreParams()
-
-	datadir, err = ioutil.TempDir("", "localMockStore-"+id.TerminalString())
-	if err != nil {
-		return nil, "", err
-	}
-	params.Init(datadir)
-	params.BaseKey = addr.Over()
-	lstore, err = storage.NewLocalStore(params, mockStore)
-	if err != nil {
-		return nil, "", err
-	}
-	return lstore, datadir, nil
-}
-
 func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, po uint8) {
 
 	sim := simulation.New(map[string]simulation.ServiceFunc{
@@ -181,17 +163,29 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
 			if i < nodes-1 {
 				hashCounts[i] = hashCounts[i+1]
 			}
-			item, ok := sim.NodeItem(nodeIDs[i], bucketKeyDB)
+			item, ok := sim.NodeItem(nodeIDs[i], bucketKeyStore)
 			if !ok {
 				return fmt.Errorf("No DB")
 			}
-			netStore := item.(*storage.NetStore)
+			localStore := item.(*localstore.DB)
-			netStore.Iterator(0, math.MaxUint64, po, func(addr storage.Address, index uint64) bool {
+			until, err := localStore.LastPullSubscriptionChunk(po)
-				hashes[i] = append(hashes[i], addr)
+			if err != nil {
+				return err
+			}
+			c, _ := localStore.SubscribePull(ctx, po, nil, until)
+			for iterate := true; iterate; {
+				select {
+				case cd, ok := <-c:
+					if !ok {
+						iterate = false
+					}
+					hashes[i] = append(hashes[i], cd.Address)
 					totalHashes++
 					hashCounts[i]++
-				return true
+				case <-ctx.Done():
-			})
+					return ctx.Err()
+				}
+			}
 		}
 		var total, found int
 		for _, node := range nodeIDs {
@@ -200,12 +194,12 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
 			for j := i; j < nodes; j++ {
 				total += len(hashes[j])
 				for _, key := range hashes[j] {
-					item, ok := sim.NodeItem(nodeIDs[j], bucketKeyDB)
+					item, ok := sim.NodeItem(nodeIDs[j], bucketKeyStore)
 					if !ok {
 						return fmt.Errorf("No DB")
 					}
 					db := item.(*storage.NetStore)
-					_, err := db.Get(ctx, key)
+					_, err := db.Get(ctx, chunk.ModeGetRequest, key)
 					if err == nil {
 						found++
 					}