swarm: high level swarm network simulation

This commit includes TestSwarmNetwork test and changes to the swarm that are required for that test to pass.

TestSwarmNetwork contains tests for static network simulation with 10 and 100 nodes, and also tests where nodes are added or removed while the availability of files is checked between changes. On every test step, a number of nodes is created and the same number of files are uploaded on each of nodes (one file per node). The check consists of trying to retrieve all files from all nodes.

cmd/swarm/db.go

@@ -115,5 +115,6 @@ func openLDBStore(path string, basekey []byte) (*storage.LDBStore, error) {
 
 	storeparams := storage.NewDefaultStoreParams()
 	ldbparams := storage.NewLDBStoreParams(storeparams, path)
+	ldbparams.BaseKey = basekey
 	return storage.NewLDBStore(ldbparams)
 }

swarm/api/config.go

@@ -28,6 +28,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/services/swap"
 	"github.com/ethereum/go-ethereum/swarm/storage"
@@ -55,6 +56,7 @@ type Config struct {
 	Port            string
 	PublicKey       string
 	BzzKey          string
+	NodeID          string
 	NetworkId       uint64
 	SwapEnabled     bool
 	SyncEnabled     bool
@@ -113,6 +115,7 @@ func (self *Config) Init(prvKey *ecdsa.PrivateKey) {
 
 	self.PublicKey = pubkeyhex
 	self.BzzKey = keyhex
+	self.NodeID = discover.PubkeyID(&prvKey.PublicKey).String()
 
 	if self.SwapEnabled {
 		self.Swap.Init(self.Contract, prvKey)
@@ -120,6 +123,7 @@ func (self *Config) Init(prvKey *ecdsa.PrivateKey) {
 
 	self.privateKey = prvKey
 	self.LocalStoreParams.Init(self.Path)
+	self.LocalStoreParams.BaseKey = common.FromHex(keyhex)
 }
 
 func (self *Config) ShiftPrivateKey() (privKey *ecdsa.PrivateKey) {

swarm/network/kademlia.go

@@ -204,6 +204,8 @@ func (k *Kademlia) Register(peers []OverlayAddr) error {
 		k.addrCountC <- k.addrs.Size()
 	}
 	// log.Trace(fmt.Sprintf("%x registered %v peers, %v known, total: %v", k.BaseAddr()[:4], size, known, k.addrs.Size()))
+
+	k.sendNeighbourhoodDepthChange()
 	return nil
 }
 
@@ -313,13 +315,7 @@ func (k *Kademlia) On(p OverlayConn) (uint8, bool) {
 		changed = true
 		k.depth = depth
 	}
-	if k.nDepthC != nil {
+	k.sendNeighbourhoodDepthChange()
-		nDepth := k.neighbourhoodDepth()
-		if nDepth != k.nDepth {
-			k.nDepth = nDepth
-			k.nDepthC <- nDepth
-		}
-	}
 	return k.depth, changed
 }
 
@@ -334,6 +330,21 @@ func (k *Kademlia) NeighbourhoodDepthC() <-chan int {
 	return k.nDepthC
 }
 
+// sendNeighbourhoodDepthChange sends new neighbourhood depth to k.nDepth channel
+// if it is initialized.
+func (k *Kademlia) sendNeighbourhoodDepthChange() {
+	// nDepthC is initialized when NeighbourhoodDepthC is called and returned by it.
+	// It provides signaling of neighbourhood depth change.
+	// This part of the code is sending new neighbourhood depth to nDepthC if that condition is met.
+	if k.nDepthC != nil {
+		nDepth := k.neighbourhoodDepth()
+		if nDepth != k.nDepth {
+			k.nDepth = nDepth
+			k.nDepthC <- nDepth
+		}
+	}
+}
+
 // AddrCountC returns the channel that sends a new
 // address count value on each change.
 // Not receiving from the returned channel will block Register function
@@ -367,6 +378,7 @@ func (k *Kademlia) Off(p OverlayConn) {
 		if k.addrCountC != nil {
 			k.addrCountC <- k.addrs.Size()
 		}
+		k.sendNeighbourhoodDepthChange()
 	}
 }
 

swarm/network/simulations/discovery/discovery_test.go

@@ -416,7 +416,7 @@ func discoveryPersistenceSimulation(nodes, conns int, adapter adapters.NodeAdapt
 				return fmt.Errorf("error stopping node %s: %s", node.ID().TerminalString(), err)
 			}
 		}
-		log.Info(fmt.Sprintf("shutting down nodes took: %s", time.Now().Sub(shutdownStarted)))
+		log.Info(fmt.Sprintf("shutting down nodes took: %s", time.Since(shutdownStarted)))
 		persistenceEnabled = true
 		discoveryEnabled = false
 		restartTime = time.Now()
@@ -430,7 +430,7 @@ func discoveryPersistenceSimulation(nodes, conns int, adapter adapters.NodeAdapt
 			}
 		}
 
-		log.Info(fmt.Sprintf("restarting nodes took: %s", time.Now().Sub(restartTime)))
+		log.Info(fmt.Sprintf("restarting nodes took: %s", time.Since(restartTime)))
 
 		return nil
 	}

swarm/network/stream/delivery.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/swarm/network"
@@ -136,19 +137,23 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e
 	chunk, created := d.db.GetOrCreateRequest(req.Key)
 	if chunk.ReqC != nil {
 		if created {
-			if err := d.RequestFromPeers(chunk.Key[:], false, sp.ID()); err != nil {
+			if err := d.RequestFromPeers(chunk.Key[:], true, sp.ID()); err != nil {
 				log.Warn("unable to forward chunk request", "peer", sp.ID(), "key", chunk.Key, "err", err)
 				chunk.SetErrored(storage.ErrChunkForward)
 				return nil
 			}
 		}
 		go func() {
-			t := time.NewTimer(3 * time.Minute)
+			t := time.NewTimer(10 * time.Minute)
 			defer t.Stop()
 
+			log.Debug("waiting delivery", "peer", sp.ID(), "hash", req.Key, "node", common.Bytes2Hex(d.overlay.BaseAddr()), "created", created)
+			start := time.Now()
 			select {
 			case <-chunk.ReqC:
+				log.Debug("retrieve request ReqC closed", "peer", sp.ID(), "hash", req.Key, "time", time.Since(start))
 			case <-t.C:
+				log.Debug("retrieve request timeout", "peer", sp.ID(), "hash", req.Key)
 				chunk.SetErrored(storage.ErrChunkTimeout)
 				return
 			}
@@ -208,14 +213,13 @@ R:
 		}
 		chunk.SData = req.SData
 		d.db.Put(chunk)
-		chunk.WaitToStore()
+
-		err = chunk.GetErrored()
+		go func(req *ChunkDeliveryMsg) {
-		if err != nil {
+			err := chunk.WaitToStore()
 			if err == storage.ErrChunkInvalid {
 				req.peer.Drop(err)
 			}
-		}
+		}(req)
-		close(chunk.ReqC)
 	}
 }
 
@@ -241,11 +245,14 @@ func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ...
 			Key:       hash,
 			SkipCheck: skipCheck,
 		}, Top)
+		if err != nil {
+			return true
+		}
 		success = true
 		return false
 	})
 	if success {
-		return err
+		return nil
 	}
 	return errors.New("no peer found")
 }

swarm/network/stream/delivery_test.go

@@ -82,7 +82,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
 		Stream:   NewStream(swarmChunkServerStreamName, "", false),
-		History:  NewRange(0, 0),
+		History:  nil,
 		Priority: Top,
 	})
 
@@ -129,9 +129,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 	peerID := tester.IDs[0]
 	peer := streamer.getPeer(peerID)
 
+	stream := NewStream(swarmChunkServerStreamName, "", false)
+
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream:   NewStream(swarmChunkServerStreamName, "", false),
+		Stream:   stream,
-		History:  NewRange(0, 0),
+		History:  nil,
 		Priority: Top,
 	})
 
@@ -163,7 +165,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 					From:   0,
 					// TODO: why is this 32???
 					To:     32,
-					Stream: NewStream(swarmChunkServerStreamName, "", false),
+					Stream: stream,
 				},
 				Peer: peerID,
 			},

swarm/network/stream/messages.go

@@ -17,6 +17,7 @@
 package stream
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -85,7 +86,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 		}
 	}()
 
-	log.Debug("%s received subscription", "from", p.streamer.addr.ID(), "peer", p.ID(), "stream", req.Stream, "history", req.History)
+	log.Debug("received subscription", "from", p.streamer.addr.ID(), "peer", p.ID(), "stream", req.Stream, "history", req.History)
 
 	f, err := p.streamer.GetServerFunc(req.Stream.Name)
 	if err != nil {
@@ -216,7 +217,10 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	// }()
 	go func() {
 		wg.Wait()
-		c.next <- c.batchDone(p, req, hashes)
+		select {
+		case c.next <- c.batchDone(p, req, hashes):
+		case <-c.quit:
+		}
 	}()
 	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
 	// except
@@ -239,7 +243,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 		select {
 		case <-time.After(120 * time.Second):
 			log.Warn("ERROR in handleOfferedHashesMsg, DROPPING peer!", "err", "TIMEOUT")
-			p.Drop(err)
+			p.Drop(errors.New("handle offered hashes timeout"))
 			return
 		case err := <-c.next:
 			if err != nil {
@@ -247,6 +251,8 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 				p.Drop(err)
 				return
 			}
+		case <-c.quit:
+			return
 		}
 		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
 		err := p.SendPriority(msg, c.priority)

swarm/network/stream/peer.go

@@ -265,6 +265,7 @@ func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created boo
 		priority:       cp.priority,
 		to:             cp.to,
 		next:           next,
+		quit:           make(chan struct{}),
 		intervalsStore: p.streamer.intervalsStore,
 		intervalsKey:   intervalsKey,
 	}

swarm/network/stream/stream.go

@@ -92,8 +92,8 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
 	streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) {
 		return NewSwarmChunkServer(delivery.db), nil
 	})
-	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ string, _ bool) (Client, error) {
+	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t string, live bool) (Client, error) {
-		return NewSwarmSyncerClient(p, delivery.db, false)
+		return NewSwarmSyncerClient(p, delivery.db, false, NewStream(swarmChunkServerStreamName, t, live))
 	})
 	RegisterSwarmSyncerServer(streamer, db)
 	RegisterSwarmSyncerClient(streamer, db)
@@ -127,13 +127,13 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
 			// wait for kademlia table to be healthy
 			time.Sleep(options.SyncUpdateDelay)
 
-			// initial requests for syncing subscription to peers
-			streamer.updateSyncing()
-
 			kad := streamer.delivery.overlay.(*network.Kademlia)
 			depthC := latestIntC(kad.NeighbourhoodDepthC())
 			addressBookSizeC := latestIntC(kad.AddrCountC())
 
+			// initial requests for syncing subscription to peers
+			streamer.updateSyncing()
+
 			for depth := range depthC {
 				log.Debug("Kademlia neighbourhood depth change", "depth", depth)
 
@@ -516,6 +516,7 @@ type client struct {
 	sessionAt uint64
 	to        uint64
 	next      chan error
+	quit      chan struct{}
 
 	intervalsKey   string
 	intervalsStore state.Store
@@ -600,7 +601,11 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error
 }
 
 func (c *client) close() {
-	close(c.next)
+	select {
+	case <-c.quit:
+	default:
+		close(c.quit)
+	}
 	c.Close()
 }
 

swarm/network/stream/syncer.go

@@ -99,13 +99,18 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
 	if to <= from || from >= s.sessionAt {
 		to = math.MaxUint64
 	}
-	ticker := time.NewTicker(1 * time.Second)
+	var ticker *time.Ticker
-	defer ticker.Stop()
+	var wait bool
 	for {
-		select {
+		if wait {
-		case <-ticker.C:
+			if ticker == nil {
-		case <-s.quit:
+				ticker = time.NewTicker(1000 * time.Millisecond)
-			return nil, 0, 0, nil, nil
+			}
+			select {
+			case <-ticker.C:
+			case <-s.quit:
+				return nil, 0, 0, nil, nil
+			}
 		}
 		err := s.db.Iterator(from, to, s.po, func(key storage.Key, idx uint64) bool {
 			batch = append(batch, key[:]...)
@@ -119,6 +124,10 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
 		if len(batch) > 0 {
 			break
 		}
+		wait = true
+	}
+	if wait {
+		ticker.Stop()
 	}
 
 	log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po))
@@ -134,18 +143,22 @@ type SwarmSyncerClient struct {
 	retrieveC     chan *storage.Chunk
 	storeC        chan *storage.Chunk
 	db            *storage.DBAPI
-	// chunker       storage.Chunker
+	// chunker               storage.Chunker
 	currentRoot           storage.Key
 	requestFunc           func(chunk *storage.Chunk)
 	end, start            uint64
+	peer                  *Peer
 	ignoreExistingRequest bool
+	stream                Stream
 }
 
 // NewSwarmSyncerClient is a contructor for provable data exchange syncer
-func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, ignoreExistingRequest bool) (*SwarmSyncerClient, error) {
+func NewSwarmSyncerClient(p *Peer, db *storage.DBAPI, ignoreExistingRequest bool, stream Stream) (*SwarmSyncerClient, error) {
 	return &SwarmSyncerClient{
-		db: db,
+		db:   db,
+		peer: p,
 		ignoreExistingRequest: ignoreExistingRequest,
+		stream:                stream,
 	}, nil
 }
 
@@ -188,16 +201,19 @@ func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, ignoreExistingRequest bool
 // RegisterSwarmSyncerClient registers the client constructor function for
 // to handle incoming sync streams
 func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
-	streamer.RegisterClientFunc("SYNC", func(p *Peer, _ string, love bool) (Client, error) {
+	streamer.RegisterClientFunc("SYNC", func(p *Peer, t string, live bool) (Client, error) {
-		return NewSwarmSyncerClient(p, db, true)
+		return NewSwarmSyncerClient(p, db, true, NewStream("SYNC", t, live))
 	})
 }
 
 // NeedData
 func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
-	chunk, created := s.db.GetOrCreateRequest(key)
+	chunk, _ := s.db.GetOrCreateRequest(key)
 	// TODO: we may want to request from this peer anyway even if the request exists
-	if chunk.ReqC == nil || (s.ignoreExistingRequest && !created) {
+
+	// ignoreExistingRequest is temporary commented out until its functionality is verified.
+	// For now, this optimization can be disabled.
+	if chunk.ReqC == nil { //|| (s.ignoreExistingRequest && !created) {
 		return nil
 	}
 	// create request and wait until the chunk data arrives and is stored

swarm/network_test.go

@@ -0,0 +1,582 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package swarm
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"os"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"github.com/ethereum/go-ethereum/p2p/simulations"
+	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	"github.com/ethereum/go-ethereum/swarm/api"
+	"github.com/ethereum/go-ethereum/swarm/network"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+	colorable "github.com/mattn/go-colorable"
+)
+
+var (
+	loglevel     = flag.Int("loglevel", 4, "verbosity of logs")
+	longrunning  = flag.Bool("longrunning", false, "do run long-running tests")
+	waitKademlia = flag.Bool("waitkademlia", false, "wait for healthy kademlia before checking files availability")
+)
+
+func init() {
+	rand.Seed(time.Now().UnixNano())
+
+	flag.Parse()
+
+	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true))))
+}
+
+// TestSwarmNetwork runs a series of test simulations with
+// static and dynamic Swarm nodes in network simulation, by
+// uploading files to every node and retrieving them.
+func TestSwarmNetwork(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		steps    []testSwarmNetworkStep
+		timeout  time.Duration
+		disabled bool
+	}{
+		{
+			name: "10_nodes",
+			steps: []testSwarmNetworkStep{
+				{
+					nodeCount: 10,
+				},
+			},
+			timeout: 45 * time.Second,
+		},
+		{
+			name: "100_nodes",
+			steps: []testSwarmNetworkStep{
+				{
+					nodeCount: 100,
+				},
+			},
+			timeout:  3 * time.Minute,
+			disabled: !*longrunning,
+		},
+		{
+			name: "inc_node_count",
+			steps: []testSwarmNetworkStep{
+				{
+					nodeCount: 2,
+				},
+				{
+					nodeCount: 5,
+				},
+				{
+					nodeCount: 10,
+				},
+			},
+			timeout:  90 * time.Second,
+			disabled: !*longrunning,
+		},
+		{
+			name: "dec_node_count",
+			steps: []testSwarmNetworkStep{
+				{
+					nodeCount: 10,
+				},
+				{
+					nodeCount: 6,
+				},
+				{
+					nodeCount: 3,
+				},
+			},
+			timeout:  90 * time.Second,
+			disabled: !*longrunning,
+		},
+		{
+			name: "dec_inc_node_count",
+			steps: []testSwarmNetworkStep{
+				{
+					nodeCount: 5,
+				},
+				{
+					nodeCount: 3,
+				},
+				{
+					nodeCount: 10,
+				},
+			},
+			timeout: 90 * time.Second,
+		},
+		{
+			name: "inc_dec_node_count",
+			steps: []testSwarmNetworkStep{
+				{
+					nodeCount: 3,
+				},
+				{
+					nodeCount: 5,
+				},
+				{
+					nodeCount: 25,
+				},
+				{
+					nodeCount: 10,
+				},
+				{
+					nodeCount: 4,
+				},
+			},
+			timeout:  5 * time.Minute,
+			disabled: !*longrunning,
+		},
+	} {
+		if tc.disabled {
+			continue
+		}
+		t.Run(tc.name, func(t *testing.T) {
+			testSwarmNetwork(t, tc.timeout, tc.steps...)
+		})
+	}
+}
+
+// testSwarmNetworkStep is the configuration
+// for the state of the simulation network.
+type testSwarmNetworkStep struct {
+	// number of swarm nodes that must be in the Up state
+	nodeCount int
+}
+
+// file represents the file uploaded on a particular node.
+type file struct {
+	key    storage.Key
+	data   string
+	nodeID discover.NodeID
+}
+
+// check represents a reference to a file that is retrieved
+// from a particular node.
+type check struct {
+	key    string
+	nodeID discover.NodeID
+}
+
+// testSwarmNetwork is a helper function used for testing different
+// static and dynamic Swarm network simulations.
+// It is responsible for:
+//  - Setting up a Swarm network simulation, and updates the number of nodes within the network on every step according to steps.
+//  - Uploading a unique file to every node on every step.
+//  - May wait for Kademlia on every node to be healthy.
+//  - Checking if a file is retrievable from all nodes.
+func testSwarmNetwork(t *testing.T, timeout time.Duration, steps ...testSwarmNetworkStep) {
+	dir, err := ioutil.TempDir("", "swarm-network-test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(dir)
+
+	ctx := context.Background()
+	if timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, timeout)
+		defer cancel()
+	}
+
+	swarms := make(map[discover.NodeID]*Swarm)
+	files := make([]file, 0)
+
+	services := map[string]adapters.ServiceFunc{
+		"swarm": func(ctx *adapters.ServiceContext) (node.Service, error) {
+			config := api.NewConfig()
+			config.PssEnabled = false
+
+			dir, err := ioutil.TempDir(dir, "node")
+			if err != nil {
+				return nil, err
+			}
+
+			config.Path = dir
+
+			privkey, err := crypto.GenerateKey()
+			if err != nil {
+				return nil, err
+			}
+
+			config.Init(privkey)
+			s, err := NewSwarm(nil, nil, config, nil)
+			if err != nil {
+				return nil, err
+			}
+			log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", s.bzz.BaseAddr()))
+			swarms[ctx.Config.ID] = s
+			return s, nil
+		},
+	}
+
+	a := adapters.NewSimAdapter(services)
+	net := simulations.NewNetwork(a, &simulations.NetworkConfig{
+		ID:             "0",
+		DefaultService: "swarm",
+	})
+	defer net.Shutdown()
+
+	trigger := make(chan discover.NodeID)
+
+	sim := simulations.NewSimulation(net)
+
+	for i, step := range steps {
+		log.Debug("test sync step", "n", i+1, "nodes", step.nodeCount)
+
+		change := step.nodeCount - len(allNodeIDs(net))
+
+		if change > 0 {
+			_, err := addNodes(change, net)
+			if err != nil {
+				t.Fatal(err)
+			}
+		} else if change < 0 {
+			err := removeNodes(-change, net)
+			if err != nil {
+				t.Fatal(err)
+			}
+		} else {
+			t.Logf("step %v: no change in nodes", i)
+			continue
+		}
+
+		nodeIDs := allNodeIDs(net)
+		shuffle(len(nodeIDs), func(i, j int) {
+			nodeIDs[i], nodeIDs[j] = nodeIDs[j], nodeIDs[i]
+		})
+		for _, id := range nodeIDs {
+			key, data, err := uploadFile(swarms[id])
+			if err != nil {
+				t.Fatal(err)
+			}
+			log.Trace("file uploaded", "node", id, "key", key.String())
+			files = append(files, file{
+				key:    key,
+				data:   data,
+				nodeID: id,
+			})
+		}
+
+		// Prepare PeerPot map for checking Kademlia health
+		var ppmap map[string]*network.PeerPot
+		nIDs := allNodeIDs(net)
+		addrs := make([][]byte, len(nIDs))
+		if *waitKademlia {
+			for i, id := range nIDs {
+				addrs[i] = swarms[id].bzz.BaseAddr()
+			}
+			ppmap = network.NewPeerPotMap(2, addrs)
+		}
+
+		var checkStatusM sync.Map
+		var nodeStatusM sync.Map
+		var totalFoundCount uint64
+
+		result := sim.Run(ctx, &simulations.Step{
+			Action: func(ctx context.Context) error {
+				if *waitKademlia {
+					// Wait for healthy Kademlia on every node before checking files
+					ticker := time.NewTicker(200 * time.Millisecond)
+					defer ticker.Stop()
+
+					for range ticker.C {
+						healthy := true
+						log.Debug("kademlia health check", "node count", len(nIDs), "addr count", len(addrs))
+						for i, id := range nIDs {
+							swarm := swarms[id]
+							//PeerPot for this node
+							addr := common.Bytes2Hex(swarm.bzz.BaseAddr())
+							pp := ppmap[addr]
+							//call Healthy RPC
+							h := swarm.bzz.Healthy(pp)
+							//print info
+							log.Debug(swarm.bzz.String())
+							log.Debug("kademlia", "empty bins", pp.EmptyBins, "gotNN", h.GotNN, "knowNN", h.KnowNN, "full", h.Full)
+							log.Debug("kademlia", "health", h.GotNN && h.KnowNN && h.Full, "addr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()), "id", id, "i", i)
+							log.Debug("kademlia", "ill condition", !h.GotNN || !h.Full, "addr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()), "id", id, "i", i)
+							if !h.GotNN || !h.Full {
+								healthy = false
+								break
+							}
+						}
+						if healthy {
+							break
+						}
+					}
+				}
+
+				go func() {
+					// File retrieval check is repeated until all uploaded files are retrieved from all nodes
+					// or until the timeout is reached.
+					for {
+						if retrieve(net, files, swarms, trigger, &checkStatusM, &nodeStatusM, &totalFoundCount) == 0 {
+							return
+						}
+					}
+				}()
+				return nil
+			},
+			Trigger: trigger,
+			Expect: &simulations.Expectation{
+				Nodes: allNodeIDs(net),
+				Check: func(ctx context.Context, id discover.NodeID) (bool, error) {
+					// The check is done by a goroutine in the action function.
+					return true, nil
+				},
+			},
+		})
+		if result.Error != nil {
+			t.Fatal(result.Error)
+		}
+		log.Debug("done: test sync step", "n", i+1, "nodes", step.nodeCount)
+	}
+}
+
+// allNodeIDs is returning NodeID for every node that is Up.
+func allNodeIDs(net *simulations.Network) (nodes []discover.NodeID) {
+	for _, n := range net.GetNodes() {
+		if n.Up {
+			nodes = append(nodes, n.ID())
+		}
+	}
+	return
+}
+
+// addNodes adds a number of nodes to the network.
+func addNodes(count int, net *simulations.Network) (ids []discover.NodeID, err error) {
+	for i := 0; i < count; i++ {
+		nodeIDs := allNodeIDs(net)
+		l := len(nodeIDs)
+		nodeconf := adapters.RandomNodeConfig()
+		node, err := net.NewNodeWithConfig(nodeconf)
+		if err != nil {
+			return nil, fmt.Errorf("create node: %v", err)
+		}
+		err = net.Start(node.ID())
+		if err != nil {
+			return nil, fmt.Errorf("start node: %v", err)
+		}
+
+		log.Debug("created node", "id", node.ID())
+
+		// connect nodes in a chain
+		if l > 0 {
+			var otherNodeID discover.NodeID
+			for i := l - 1; i >= 0; i-- {
+				n := net.GetNode(nodeIDs[i])
+				if n.Up {
+					otherNodeID = n.ID()
+					break
+				}
+			}
+			log.Debug("connect nodes", "one", node.ID(), "other", otherNodeID)
+			if err := net.Connect(node.ID(), otherNodeID); err != nil {
+				return nil, err
+			}
+		}
+		ids = append(ids, node.ID())
+	}
+	return ids, nil
+}
+
+// removeNodes stops a random nodes in the network.
+func removeNodes(count int, net *simulations.Network) error {
+	for i := 0; i < count; i++ {
+		// allNodeIDs are returning only the Up nodes.
+		nodeIDs := allNodeIDs(net)
+		if len(nodeIDs) == 0 {
+			break
+		}
+		node := net.GetNode(nodeIDs[rand.Intn(len(nodeIDs))])
+		if err := node.Stop(); err != nil {
+			return err
+		}
+		log.Debug("removed node", "id", node.ID())
+	}
+	return nil
+}
+
+// uploadFile, uploads a short file to the swarm instance
+// using the api.Put method.
+func uploadFile(swarm *Swarm) (storage.Key, string, error) {
+	b := make([]byte, 8)
+	_, err := rand.Read(b)
+	if err != nil {
+		return nil, "", err
+	}
+	// File data is very short, but it is ensured that its
+	// uniqueness is very certain.
+	data := fmt.Sprintf("test content %s %x", time.Now().Round(0), b)
+	k, wait, err := swarm.api.Put(data, "text/plain", false)
+	if err != nil {
+		return nil, "", err
+	}
+	if wait != nil {
+		wait()
+	}
+	return k, data, nil
+}
+
+// retrieve is the function that is used for checking the availability of
+// uploaded files in testSwarmNetwork test helper function.
+func retrieve(
+	net *simulations.Network,
+	files []file,
+	swarms map[discover.NodeID]*Swarm,
+	trigger chan discover.NodeID,
+	checkStatusM *sync.Map,
+	nodeStatusM *sync.Map,
+	totalFoundCount *uint64,
+) (missing uint64) {
+	shuffle(len(files), func(i, j int) {
+		files[i], files[j] = files[j], files[i]
+	})
+
+	var totalWg sync.WaitGroup
+	errc := make(chan error)
+
+	nodeIDs := allNodeIDs(net)
+
+	totalCheckCount := len(nodeIDs) * len(files)
+
+	for _, id := range nodeIDs {
+		if _, ok := nodeStatusM.Load(id); ok {
+			continue
+		}
+		start := time.Now()
+		var checkCount uint64
+		var foundCount uint64
+
+		totalWg.Add(1)
+
+		var wg sync.WaitGroup
+
+		for _, f := range files {
+			swarm := swarms[id]
+
+			checkKey := check{
+				key:    f.key.String(),
+				nodeID: id,
+			}
+			if n, ok := checkStatusM.Load(checkKey); ok && n.(int) == 0 {
+				continue
+			}
+
+			checkCount++
+			wg.Add(1)
+			go func(f file, id discover.NodeID) {
+				defer wg.Done()
+
+				log.Debug("api get: check file", "node", id.String(), "key", f.key.String(), "total files found", atomic.LoadUint64(totalFoundCount))
+
+				r, _, _, err := swarm.api.Get(f.key, "/")
+				if err != nil {
+					errc <- fmt.Errorf("api get: node %s, key %s, kademlia %s: %v", id, f.key, swarm.bzz.Hive, err)
+					return
+				}
+				d, err := ioutil.ReadAll(r)
+				if err != nil {
+					errc <- fmt.Errorf("api get: read response: node %s, key %s: kademlia %s: %v", id, f.key, swarm.bzz.Hive, err)
+					return
+				}
+				data := string(d)
+				if data != f.data {
+					errc <- fmt.Errorf("file contend missmatch: node %s, key %s, expected %q, got %q", id, f.key, f.data, data)
+					return
+				}
+				checkStatusM.Store(checkKey, 0)
+				atomic.AddUint64(&foundCount, 1)
+				log.Info("api get: file found", "node", id.String(), "key", f.key.String(), "content", data, "files found", atomic.LoadUint64(&foundCount))
+			}(f, id)
+		}
+
+		go func(id discover.NodeID) {
+			defer totalWg.Done()
+			wg.Wait()
+
+			atomic.AddUint64(totalFoundCount, foundCount)
+
+			if foundCount == checkCount {
+				log.Info("all files are found for node", "id", id.String(), "duration", time.Since(start))
+				nodeStatusM.Store(id, 0)
+				trigger <- id
+				return
+			}
+			log.Debug("files missing for node", "id", id.String(), "check", checkCount, "found", foundCount)
+		}(id)
+
+	}
+
+	go func() {
+		totalWg.Wait()
+		close(errc)
+	}()
+
+	var errCount int
+	for err := range errc {
+		if err != nil {
+			errCount++
+		}
+		log.Warn(err.Error())
+	}
+
+	log.Info("check stats", "total check count", totalCheckCount, "total files found", atomic.LoadUint64(totalFoundCount), "total errors", errCount)
+
+	return uint64(totalCheckCount) - atomic.LoadUint64(totalFoundCount)
+}
+
+// Backported from stdlib https://golang.org/src/math/rand/rand.go?s=11175:11215#L333
+//
+// Replace with rand.Shuffle from go 1.10 when go 1.9 support is dropped.
+//
+// shuffle pseudo-randomizes the order of elements.
+// n is the number of elements. Shuffle panics if n < 0.
+// swap swaps the elements with indexes i and j.
+func shuffle(n int, swap func(i, j int)) {
+	if n < 0 {
+		panic("invalid argument to Shuffle")
+	}
+
+	// Fisher-Yates shuffle: https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle
+	// Shuffle really ought not be called with n that doesn't fit in 32 bits.
+	// Not only will it take a very long time, but with 2³¹! possible permutations,
+	// there's no way that any PRNG can have a big enough internal state to
+	// generate even a minuscule percentage of the possible permutations.
+	// Nevertheless, the right API signature accepts an int n, so handle it as best we can.
+	i := n - 1
+	for ; i > 1<<31-1-1; i-- {
+		j := int(rand.Int63n(int64(i + 1)))
+		swap(i, j)
+	}
+	for ; i > 0; i-- {
+		j := int(rand.Int31n(int32(i + 1)))
+		swap(i, j)
+	}
+}

swarm/storage/ldbstore.go

@@ -227,7 +227,11 @@ func encodeIndex(index *dpaDBIndex) []byte {
 }
 
 func encodeData(chunk *Chunk) []byte {
-	return append(chunk.Key[:], chunk.SData...)
+	// Always create a new underlying array for the returned byte slice.
+	// The chunk.Key array may be used in the returned slice which
+	// may be changed later in the code or by the LevelDB, resulting
+	// that the Key is changed as well.
+	return append(append([]byte{}, chunk.Key[:]...), chunk.SData...)
 }
 
 func decodeIndex(data []byte, index *dpaDBIndex) error {

swarm/storage/localstore.go

@@ -88,6 +88,18 @@ func NewTestLocalStoreForAddr(params *LocalStoreParams) (*LocalStore, error) {
 	return localStore, nil
 }
 
+// Put is responsible for doing validation and storage of the chunk
+// by using configured ChunkValidators, MemStore and LDBStore.
+// If the chunk is not valid, its GetErrored function will
+// return ErrChunkInvalid.
+// This method will check if the chunk is already in the MemStore
+// and it will return it if it is. If there is an error from
+// the MemStore.Get, it will be returned by calling GetErrored
+// on the chunk.
+// This method is responsible for closing Chunk.ReqC channel
+// when the chunk is stored in memstore.
+// After the LDBStore.Put, it is ensured that the MemStore
+// contains the chunk with the same data, but nil ReqC channel.
 func (self *LocalStore) Put(chunk *Chunk) {
 	valid := true
 	for _, v := range self.Validators {
@@ -97,26 +109,52 @@ func (self *LocalStore) Put(chunk *Chunk) {
 	}
 	if !valid {
 		chunk.SetErrored(ErrChunkInvalid)
-		chunk.dbStoredC <- false
+		chunk.markAsStored()
 		return
 	}
+
 	log.Trace("localstore.put", "key", chunk.Key)
 	self.mu.Lock()
 	defer self.mu.Unlock()
 
 	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
-	c := &Chunk{
+
-		Key:        Key(append([]byte{}, chunk.Key...)),
+	memChunk, err := self.memStore.Get(chunk.Key)
-		SData:      append([]byte{}, chunk.SData...),
+	switch err {
-		Size:       chunk.Size,
+	case nil:
-		dbStored:   chunk.dbStored,
+		if memChunk.ReqC == nil {
-		dbStoredC:  chunk.dbStoredC,
+			chunk.markAsStored()
-		dbStoredMu: chunk.dbStoredMu,
+			return
+		}
+	case ErrChunkNotFound:
+	default:
+		chunk.SetErrored(err)
+		return
+	}
+
+	self.memStore.Put(chunk)
+
+	if memChunk != nil && memChunk.ReqC != nil {
+		close(memChunk.ReqC)
 	}
 
 	dbStorePutCounter.Inc(1)
-	self.memStore.Put(c)
+	self.DbStore.Put(chunk)
-	self.DbStore.Put(c)
+
+	newc := NewChunk(chunk.Key, nil)
+	newc.SData = chunk.SData
+	newc.Size = chunk.Size
+	//newc.dbStored = chunk.dbStored
+	newc.dbStoredC = chunk.dbStoredC
+	//newc.dbStoredMu = chunk.dbStoredMu
+	go func() {
+		<-chunk.dbStoredC
+
+		self.mu.Lock()
+		defer self.mu.Unlock()
+
+		self.memStore.Put(newc)
+	}()
 }
 
 // Get(chunk *Chunk) looks up a chunk in the local stores

swarm/storage/resource.go

@@ -214,10 +214,7 @@ func (self *ResourceHandler) Validate(key Key, data []byte) bool {
 		}
 		return false
 	} else if signature == nil {
-		if !bytes.Equal(self.resourceHash(period, version, ens.EnsNode(name)), key) {
+		return bytes.Equal(self.resourceHash(period, version, ens.EnsNode(name)), key)
-			return false
-		}
-		return true
 	}
 	digest := self.keyDataHash(key, parseddata)
 	addr, err := getAddressFromDataSig(digest, *signature)

swarm/storage/types.go

@@ -32,7 +32,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
-const MaxPO = 7
+const MaxPO = 16
 const KeyLength = 32
 
 type Hasher func() hash.Hash
@@ -218,8 +218,9 @@ func (c *Chunk) markAsStored() {
 	}
 }
 
-func (c *Chunk) WaitToStore() {
+func (c *Chunk) WaitToStore() error {
 	<-c.dbStoredC
+	return c.GetErrored()
 }
 
 func GenerateRandomChunk(dataSize int64) *Chunk {

swarm/swarm.go

@@ -32,7 +32,6 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/contracts/chequebook"
 	"github.com/ethereum/go-ethereum/contracts/ens"
-	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
@@ -129,10 +128,18 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 	//self.cloud = &storage.Forwarder{}
 	//self.storage = storage.NewNetStore(hash, self.lstore, self.cloud, config.StoreParams)
 	log.Debug(fmt.Sprintf("-> swarm net store shared access layer to Swarm Chunk Store"))
-	nodeid := discover.PubkeyID(crypto.ToECDSAPub(common.FromHex(config.PublicKey)))
+
-	addr := network.NewAddrFromNodeID(nodeid)
+	nodeID, err := discover.HexID(config.NodeID)
+	if err != nil {
+		return nil, err
+	}
+	addr := &network.BzzAddr{
+		OAddr: common.FromHex(config.BzzKey),
+		UAddr: []byte(discover.NewNode(nodeID, net.IP{127, 0, 0, 1}, 30303, 30303).String()),
+	}
+
 	bzzconfig := &network.BzzConfig{
-		OverlayAddr:  common.FromHex(config.BzzKey),
+		OverlayAddr:  addr.OAddr,
 		UnderlayAddr: addr.UAddr,
 		HiveParams:   config.HiveParams,
 	}
@@ -220,10 +227,12 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 	self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run)
 
 	// Pss = postal service over swarm (devp2p over bzz)
-	pssparams := pss.NewPssParams(self.privateKey)
+	if config.PssEnabled {
-	self.ps = pss.NewPss(to, pssparams)
+		pssparams := pss.NewPssParams(self.privateKey)
-	if pss.IsActiveHandshake {
+		self.ps = pss.NewPss(to, pssparams)
-		pss.SetHandshakeController(self.ps, pss.NewHandshakeParams())
+		if pss.IsActiveHandshake {
+			pss.SetHandshakeController(self.ps, pss.NewHandshakeParams())
+		}
 	}
 
 	self.api = api.NewApi(self.dpa, self.dns, resourceHandler)