swarm: test network syncing after upload

swarm/network/stream/common_test.go

@@ -71,9 +71,14 @@ func init() {
 
 // NewStreamerService
 func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
+	var err error
 	id := ctx.Config.ID
 	addr := toAddr(id)
 	kad := network.NewKademlia(addr.Over(), network.NewKadParams())
+	stores[id], err = createTestLocalStorageForId(id, addr)
+	if err != nil {
+		return nil, err
+	}
 	store := stores[id].(*storage.LocalStore)
 	db := storage.NewDBAPI(store)
 	delivery := NewDelivery(kad, db)
@@ -90,6 +95,32 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	return &TestRegistry{Registry: r, dpa: dpa}, nil
 }
 
+//create a local store for the given node
+func createTestLocalStorageForId(id discover.NodeID, addr *network.BzzAddr) (storage.ChunkStore, error) {
+	var datadir string
+	var err error
+	datadir, err = ioutil.TempDir("", fmt.Sprintf("syncer-test-%s", id.TerminalString()))
+	if err != nil {
+		return nil, err
+	}
+	datadirs[id] = datadir
+	var store storage.ChunkStore
+	store, err = storage.NewTestLocalStoreForAddr(datadir, addr.Over())
+	if err != nil {
+		return nil, err
+	}
+	return store, nil
+}
+
+//local stores need to be cleaned up after the sim is done
+func localStoreCleanup() {
+	fmt.Println("Local store cleanup")
+	for i := 0; i < len(ids); i++ {
+		stores[ids[i]].Close()
+		os.RemoveAll(datadirs[ids[i]])
+	}
+}
+
 func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
 	// setup
 	addr := network.RandomAddr() // tested peers peer address

swarm/network/stream/snapshot_retrieval_test.go

@@ -0,0 +1,306 @@
+// 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 stream
+
+/*
+
+import (
+	"context"
+	crand "crypto/rand"
+	"flag"
+	"fmt"
+	"io"
+	"math/rand"
+//	"os"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+//	"github.com/ethereum/go-ethereum/node"
+//	"github.com/ethereum/go-ethereum/pot"
+	"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/network"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+	//streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
+)
+
+var rootHash storage.Key
+
+
+func init() {
+	flag.Parse()
+	rand.Seed(time.Now().Unix())
+
+	initRetrievalTest()
+}
+
+
+func initRetrievalTest() {
+  //assign the toAddr func so NewStreamerService can build the addr
+  toAddr = func(id discover.NodeID) *network.BzzAddr {
+    addr := network.NewAddrFromNodeID(id)
+    addr.OAddr[0] = byte(0)
+    return addr
+  }
+
+  //nodeCount is needed to load a specific json snapshot file,
+  //e.g. "snapshot_16.json"
+  nodeCount = 16
+  //is used to continuosly provide the current discoverID to NewStreamerService
+  //while loading the snapshot
+  currentId = 0
+  //local stores
+	stores        = make(map[discover.NodeID]storage.ChunkStore)
+  //data directories for each node and store
+	datadirs      = make(map[discover.NodeID]string)
+  //deliveries for each node
+	deliveries    = make(map[discover.NodeID]*Delivery)
+  //the list of the ids loaded from the snapshot
+  ids           = make([]discover.NodeID, nodeCount)
+  //mapping of nearest node addresses for chunk hashes
+  //chunksForAddressesMap = make(map[discover.NodeID][]storage.Key)
+
+
+	waitPeerErrC  = make(chan error)
+	// peerCount function gives the number of peer connections for a nodeID
+	// this is needed for the service run function to wait until
+	// each protocol  instance runs and the streamer peers are available
+	peerCount = func(id discover.NodeID) int {
+		if ids[0] == id || ids[len(ids)-1] == id {
+			return 1
+		}
+		return 2
+	}
+}
+
+
+func TestRetrieval_4(t *testing.T)   { retrievalTest(t, 4) }
+/*
+func TestRetrieval_1(t *testing.T)   { retrievalTest(t, 1) }
+func TestSyncing_4(t *testing.T) { testSyncing(t, 4) }
+func TestSyncing_8(t *testing.T) { testSyncing(t, 8) }
+func TestSyncing_32(t *testing.T) { testSyncing(t, 32) }
+func TestSyncing_128(t *testing.T) { testSyncing(t, 128) }
+func TestSyncing_256(t *testing.T) { testSyncing(t, 256) }
+func TestSyncing_1024(t *testing.T) { testSyncing(t,1024) }
+
+// Benchmarks to test the average time it takes for an N-node ring
+// to full a healthy kademlia topology
+func BenchmarkSyncing_1(b *testing.B)   { benchmarkSyncing(b, 1) }
+func BenchmarkSyncing_4(b *testing.B)  { benchmarkSyncing(b, 4) }
+func BenchmarkSyncing_8(b *testing.B)  { benchmarkSyncing(b, 8) }
+func BenchmarkSyncing_32(b *testing.B)  { benchmarkSyncing(b, 32) }
+func BenchmarkSyncing_128(b *testing.B) { benchmarkSyncing(b, 128) }
+func BenchmarkSyncing_256(b *testing.B) { benchmarkSyncing(b, 256) }
+func BenchmarkSyncing_1024(b *testing.B) { benchmarkSyncing(b, 1024) }
+
+func benchmarkSyncing(b *testing.B, chunkCount int) {
+	for i := 0; i < b.N; i++ {
+		result, err := testSyncing(b.T, chunkCount)
+		if err != nil {
+			b.Fatalf("setting up simulation failed", result)
+		}
+		if result.Error != nil {
+			b.Logf("simulation failed: %s", result.Error)
+		}
+	}
+}
+
+*/
+/*
+func retrievalTest(t *testing.T, chunkCount int) {
+  err := runRetrievalTest(chunkCount)
+  if err != nil {
+    t.Fatal(err)
+  }
+}
+
+/*
+The test generates the given number of chunks,
+then uploads these to a random node.
+Afterwards for every chunk generated, the nearest node addresses
+are identified, syncing is started, and finally we verify
+that the nodes closer to the chunk addresses actually do have 
+the chunks in their local stores.
+
+The test loads a snapshot file to construct the swarm network, 
+assuming that the snapshot file identifies a healthy
+kademlia network. The snapshot should have 'streamer' in its service list.
+*/
+/*
+func runRetrievalTest(chunkCount int) error {
+
+  //reset global vars
+  resetVars()
+  //First load the snapshot from the file
+  net,err := initNetWithSnapshot()
+  if err != nil {
+    return err
+  }
+	defer net.Shutdown()
+
+  //get the nodes of the network
+	nodes := net.GetNodes()
+  //select one index at random...
+	idx   := rand.Intn(len(nodes))
+  //...and get the the node at that index
+  //this is the node selected for upload
+  node  := nodes[idx]
+  //iterate over all nodes...
+  for c:=0; c<len(nodes); c++ {
+    //create an array of discovery nodeIDS
+    ids[c] = nodes[c].ID()
+    //and a correspondent array of overlay addresses, 
+    //later used for chunk proximity calculation
+    addrs = append(addrs, network.ToOverlayAddr(ids[c].Bytes()))
+  }
+
+	// channel to signal simulation initialisation with action call complete
+	// or node disconnections
+	//disconnectC := make(chan error)
+	//quitC := make(chan struct{})
+
+  //after the test, clean up local stores initialized with createLocalStoreForId
+  defer localStoreCleanup()
+
+	trigger := make(chan discover.NodeID)
+  //triggerCheck defines what will be checked during the test
+	triggerCheck := func(ctx context.Context, id discover.NodeID) (bool, error) {
+		select {
+		case <-ctx.Done():
+			return false, ctx.Err()
+	  //case <-disconnectC:
+    //  log.Error("Disconnect event detected")
+    //  return false, ctx.Err()
+		default:
+		}
+
+    log.Warn(fmt.Sprintf("Checking node: %s", id))
+    //select the !!!!NETstore!!! for the given node
+    /*
+    lstore := stores[id]
+    if _,err := lstore.Get(rootHash); err !=nil {
+      log.Warn("File Not Found")
+      return false, nil
+    }
+    log.Warn("File Found")
+    */
+/*
+    return true, nil
+	}
+
+  //for each tick, select a new node to be checked
+  ticker := time.NewTicker(time.Second * 1)
+  go func() {
+        for i:=0;i<len(ids);i++ {
+          <-ticker.C
+          trigger <- ids[i]
+          log.Debug(fmt.Sprintf("triggering step %d, id %s", i, ids[i]))
+        }
+  }()
+
+	timeout := 300 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+  //define the action to be performed before the test checks: start syncing
+	action := func(ctx context.Context) error {
+		// need to wait till an aynchronous process registers the peers in streamer.peers
+		// that is used by Subscribe
+		// the global peerCount function tells how many connections each node has
+		// TODO: this is to be reimplemented with peerEvent watcher without global var
+		i := 0
+		for err := range waitPeerErrC {
+			if err != nil {
+				return fmt.Errorf("error waiting for peers: %s", err)
+			}
+			i++
+			if i == len(ids)-1 {
+				break
+			}
+		}
+
+		// each node Subscribes to each other's swarmChunkServerStreamName
+		for j := 0; j < len(ids); j++ {
+      log.Debug(fmt.Sprintf("subscribe: %d",j))
+      ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+      defer cancel()
+      client,err := net.GetNode(ids[j]).Client()
+      if err != nil {
+        return err
+      }
+      //RPC call to subscribe, select bin 0
+      //client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false)
+      // report disconnect events to the error channel cos peers should not disconnect
+      //err = streamTesting.WatchDisconnections(ids[j], client, disconnectC, quitC)
+      //if err != nil {
+      //  return err
+      //}
+      // start syncing, i.e., subscribe to upstream peers po 1 bin
+      //each node subscribes to the next index, last subscribes to 0
+      idx:= j+1
+      if j==len(ids)-1 {
+        idx = 0
+      }
+      sid := ids[idx]
+      client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false)
+    }
+    //now upload the chunks to the selected random single node
+    rootHash,err = uploadFileToRandomNodeStore(node.ID(), chunkCount)
+    if err != nil {
+      return err
+    }
+    //finally map chunks to the closest addresses
+    //chunksForAddressesMap = mapIdsToKeys(chunks, ids)
+    log.Debug(fmt.Sprintf("%v",chunksForAddressesMap))
+
+		return nil
+  }
+  //run the simulation
+	result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
+		Action:  action,
+		Trigger: trigger,
+		Expect: &simulations.Expectation{
+			Nodes: ids,
+			Check: triggerCheck,
+		},
+	})
+  //close(quitC)
+	if result.Error != nil {
+    return result.Error
+	}
+  return nil
+}
+
+//upload a file(chunks) to a single local node store
+func uploadFileToRandomNodeStore(id discover.NodeID, chunkCount int) (storage.Key, error) {
+  log.Debug(fmt.Sprintf("Uploading to node id: %s", id))
+  lstore := stores[id]
+	size := chunkCount * chunkSize
+  dpa := storage.NewDPA(lstore, storage.NewChunkerParams())
+  dpa.Start()
+  rootHash, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+  wait()
+  if err != nil {
+    return nil, err
+  }
+
+  defer dpa.Stop()
+
+  return rootHash, nil
+}
+*/

swarm/network/stream/snapshot_sync_test.go

@@ -0,0 +1,513 @@
+// 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 stream
+
+import (
+	"context"
+	crand "crypto/rand"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"math/rand"
+	"os"
+	"testing"
+	"time"
+
+	"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/pot"
+	"github.com/ethereum/go-ethereum/swarm/network"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+	//streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
+)
+
+const testMinProxBinSize = 2
+
+var (
+	pof       = pot.DefaultPof(256)
+	ids       []discover.NodeID
+	datadirs  map[discover.NodeID]string
+	startTime time.Time
+)
+
+type synctestConfig struct {
+	addrs            [][]byte
+	chunks           []storage.Key
+	retrievalMap     map[string]map[string]time.Duration
+	nodesToChunksMap map[string][]int
+	chunksToNodesMap map[string][]int
+	idToAddrMap      map[discover.NodeID][]byte
+	addrToIdMap      map[string]discover.NodeID
+}
+
+func init() {
+	rand.Seed(time.Now().Unix())
+
+	initSyncTest()
+}
+
+//common_test needs to initialize the test in a init() func
+//in order for adapters to register the NewStreamerService;
+//this service is dependent on some global variables
+//we thus need to initialize first as init() as well.
+func initSyncTest() {
+	//assign the toAddr func so NewStreamerService can build the addr
+	toAddr = func(id discover.NodeID) *network.BzzAddr {
+		addr := network.NewAddrFromNodeID(id)
+		addr.OAddr[0] = byte(0)
+		return addr
+	}
+
+	//local stores
+	stores = make(map[discover.NodeID]storage.ChunkStore)
+	//data directories for each node and store
+	datadirs = make(map[discover.NodeID]string)
+	//deliveries for each node
+	deliveries = make(map[discover.NodeID]*Delivery)
+
+	//channel to wait for peers connected
+	waitPeerErrC = make(chan error)
+	// peerCount function gives the number of peer connections for a nodeID
+	// this is needed for the service run function to wait until
+	// each protocol  instance runs and the streamer peers are available
+	peerCount = func(id discover.NodeID) int {
+		if ids[0] == id || ids[len(ids)-1] == id {
+			return 1
+		}
+		return 2
+	}
+}
+
+func TestSyncing_1_16(t *testing.T)     { testSyncing(t, 1, 16) }
+func TestSyncing_1_32(t *testing.T)     { testSyncing(t, 1, 32) }
+func TestSyncing_1_64(t *testing.T)     { testSyncing(t, 1, 64) }
+func TestSyncing_1_128(t *testing.T)    { testSyncing(t, 1, 128) }
+func TestSyncing_1_256(t *testing.T)    { testSyncing(t, 1, 256) }
+func TestSyncing_4_16(t *testing.T)     { testSyncing(t, 4, 16) }
+func TestSyncing_4_32(t *testing.T)     { testSyncing(t, 4, 32) }
+func TestSyncing_4_64(t *testing.T)     { testSyncing(t, 4, 64) }
+func TestSyncing_4_128(t *testing.T)    { testSyncing(t, 4, 128) }
+func TestSyncing_4_256(t *testing.T)    { testSyncing(t, 4, 256) }
+func TestSyncing_8_16(t *testing.T)     { testSyncing(t, 8, 16) }
+func TestSyncing_8_32(t *testing.T)     { testSyncing(t, 8, 32) }
+func TestSyncing_8_64(t *testing.T)     { testSyncing(t, 8, 64) }
+func TestSyncing_8_128(t *testing.T)    { testSyncing(t, 8, 128) }
+func TestSyncing_8_256(t *testing.T)    { testSyncing(t, 8, 256) }
+func TestSyncing_32_16(t *testing.T)    { testSyncing(t, 32, 16) }
+func TestSyncing_32_32(t *testing.T)    { testSyncing(t, 32, 32) }
+func TestSyncing_32_64(t *testing.T)    { testSyncing(t, 32, 64) }
+func TestSyncing_32_128(t *testing.T)   { testSyncing(t, 32, 128) }
+func TestSyncing_32_256(t *testing.T)   { testSyncing(t, 32, 256) }
+func TestSyncing_128_16(t *testing.T)   { testSyncing(t, 128, 16) }
+func TestSyncing_128_32(t *testing.T)   { testSyncing(t, 128, 32) }
+func TestSyncing_128_64(t *testing.T)   { testSyncing(t, 128, 64) }
+func TestSyncing_128_128(t *testing.T)  { testSyncing(t, 128, 128) }
+func TestSyncing_128_256(t *testing.T)  { testSyncing(t, 128, 256) }
+func TestSyncing_256_16(t *testing.T)   { testSyncing(t, 256, 16) }
+func TestSyncing_256_32(t *testing.T)   { testSyncing(t, 256, 32) }
+func TestSyncing_256_64(t *testing.T)   { testSyncing(t, 256, 64) }
+func TestSyncing_256_128(t *testing.T)  { testSyncing(t, 256, 128) }
+func TestSyncing_256_256(t *testing.T)  { testSyncing(t, 256, 256) }
+func TestSyncing_1024_16(t *testing.T)  { testSyncing(t, 1024, 16) }
+func TestSyncing_1024_32(t *testing.T)  { testSyncing(t, 1024, 32) }
+func TestSyncing_1024_64(t *testing.T)  { testSyncing(t, 1024, 64) }
+func TestSyncing_1024_128(t *testing.T) { testSyncing(t, 1024, 128) }
+func TestSyncing_1024_256(t *testing.T) { testSyncing(t, 1024, 256) }
+
+// Benchmarks to test the average time it takes for an N-node ring
+// to full a healthy kademlia topology
+/*
+func BenchmarkSyncing_1(b *testing.B)   { benchmarkSyncing(b, 1) }
+func BenchmarkSyncing_4(b *testing.B)  { benchmarkSyncing(b, 4) }
+func BenchmarkSyncing_8(b *testing.B)  { benchmarkSyncing(b, 8) }
+func BenchmarkSyncing_32(b *testing.B)  { benchmarkSyncing(b, 32) }
+func BenchmarkSyncing_128(b *testing.B) { benchmarkSyncing(b, 128) }
+func BenchmarkSyncing_256(b *testing.B) { benchmarkSyncing(b, 256) }
+func BenchmarkSyncing_1024(b *testing.B) { benchmarkSyncing(b, 1024) }
+
+func benchmarkSyncing(b *testing.B, chunkCount int) {
+	for i := 0; i < b.N; i++ {
+		result, err := testSyncing(b.T, chunkCount)
+		if err != nil {
+			b.Fatalf("setting up simulation failed", result)
+		}
+		if result.Error != nil {
+			b.Logf("simulation failed: %s", result.Error)
+		}
+	}
+}
+*/
+
+func testSyncing(t *testing.T, chunkCount int, nodeCount int) {
+	ids = make([]discover.NodeID, nodeCount)
+	err := runSyncTest(chunkCount, nodeCount)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+/*
+The test generates the given number of chunks,
+then uploads these to a random node.
+Afterwards for every chunk generated, the nearest node addresses
+are identified, syncing is started, and finally we verify
+that the nodes closer to the chunk addresses actually do have
+the chunks in their local stores.
+
+The test loads a snapshot file to construct the swarm network,
+assuming that the snapshot file identifies a healthy
+kademlia network. The snapshot should have 'streamer' in its service list.
+*/
+func runSyncTest(chunkCount int, nodeCount int) error {
+
+	conf := &synctestConfig{}
+	//mapping of nearest node addresses for chunk hashes
+	//nodesToChunksMap = make(map[discover.NodeID][]storage.Key)
+	conf.retrievalMap = make(map[string]map[string]time.Duration)
+	conf.idToAddrMap = make(map[discover.NodeID][]byte)
+	conf.addrToIdMap = make(map[string]discover.NodeID)
+	//First load the snapshot from the file
+	net, err := initNetWithSnapshot(nodeCount)
+	if err != nil {
+		return err
+	}
+	defer net.Shutdown()
+
+	//get the nodes of the network
+	nodes := net.GetNodes()
+	//select one index at random...
+	idx := rand.Intn(len(nodes))
+	//...and get the the node at that index
+	//this is the node selected for upload
+	node := nodes[idx]
+	//iterate over all nodes...
+	for c := 0; c < len(nodes); c++ {
+		//create an array of discovery nodeIDS
+		ids[c] = nodes[c].ID()
+		//and a correspondent array of overlay addresses,
+		//later used for chunk proximity calculation
+		a := network.ToOverlayAddr(ids[c].Bytes())
+		conf.addrs = append(conf.addrs, a)
+		//the proximity calculation is on overlay addr,
+		//the p2p/simulations check func triggers on discover.NodeID,
+		//so we need to know which overlay addr maps to which nodeID
+		conf.idToAddrMap[ids[c]] = a
+		conf.addrToIdMap[string(a)] = ids[c]
+	}
+
+	// channel to signal simulation initialisation with action call complete
+	// or node disconnections
+	//disconnectC := make(chan error)
+	//quitC := make(chan struct{})
+
+	//after the test, clean up local stores initialized with createLocalStoreForId
+	defer localStoreCleanup()
+
+	trigger := make(chan discover.NodeID)
+	//triggerCheck defines what will be checked during the test
+	triggerCheck := func(ctx context.Context, id discover.NodeID) (bool, error) {
+		select {
+		case <-ctx.Done():
+			return false, ctx.Err()
+			//case <-disconnectC:
+			//  log.Error("Disconnect event detected")
+			//  return false, ctx.Err()
+		default:
+		}
+
+		log.Debug(fmt.Sprintf("Checking node: %s", id))
+		//select the local store for the given node
+		lstore := stores[id]
+		//if there are more than one chunk, test only succeeds if all expected chunks are found
+		allSuccess := true
+		//this selects which chunks are expected to be found with the given node
+		//localChunks := nodesToChunksMap[id]
+		localChunks := conf.nodesToChunksMap[string(conf.idToAddrMap[id])]
+
+		//for each expected chunk, check if it is in the local store
+		for i := 0; i < len(localChunks); i++ {
+			//ignore zero chunks
+			chunk := conf.chunks[localChunks[i]]
+			if storage.IsZeroKey(chunk) {
+				continue
+			}
+			log.Debug(fmt.Sprintf("node has chunk: %s:", chunk))
+			if _, err := lstore.Get(chunk); err != nil {
+				log.Error(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
+				allSuccess = false
+			} else {
+				log.Info(fmt.Sprintf("Chunk %s FOUND for id %s", chunk, id))
+			}
+		}
+
+		return allSuccess, nil
+	}
+
+	timeout := 300 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	//define the action to be performed before the test checks: start syncing
+	action := func(ctx context.Context) error {
+		// need to wait till an aynchronous process registers the peers in streamer.peers
+		// that is used by Subscribe
+		// the global peerCount function tells how many connections each node has
+		// TODO: this is to be reimplemented with peerEvent watcher without global var
+		i := 0
+		for err := range waitPeerErrC {
+			if err != nil {
+				return fmt.Errorf("error waiting for peers: %s", err)
+			}
+			i++
+			if i == len(ids)-1 {
+				break
+			}
+		}
+
+		peerPot := network.NewPeerPot(testMinProxBinSize, ids, conf.addrs)
+		// each node Subscribes to each other's swarmChunkServerStreamName
+		for j := 0; j < len(ids); j++ {
+			log.Debug(fmt.Sprintf("subscribe: %d", j))
+			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+			defer cancel()
+			client, err := net.GetNode(ids[j]).Client()
+			if err != nil {
+				return err
+			}
+			//RPC call to subscribe, select bin 0
+			//client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false)
+			// report disconnect events to the error channel cos peers should not disconnect
+			//err = streamTesting.WatchDisconnections(ids[j], client, disconnectC, quitC)
+			//if err != nil {
+			//  return err
+			//}
+			// start syncing, i.e., subscribe to upstream peers po 1 bin
+			//each node subscribes to the next index, last subscribes to 0
+			idx := j + 1
+			if j == len(ids)-1 {
+				idx = 0
+			}
+			sid := ids[idx]
+			client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false)
+		}
+		//now upload the chunks to the selected random single node
+		conf.chunks, err = uploadFileToSingleNodeStore(node.ID(), chunkCount)
+		if err != nil {
+			return err
+		}
+		//finally map chunks to the closest addresses
+		conf = mapKeysToNodes(conf)
+		log.Debug(fmt.Sprintf("%v", conf.nodesToChunksMap))
+
+		return nil
+	}
+
+	//for each tick, run the checks on all nodes
+	go func() {
+		ticker := time.NewTicker(time.Second * 1)
+		for range ticker.C {
+			for i := 0; i < len(ids); i++ {
+				log.Debug(fmt.Sprintf("triggering step %d, id %s", i, ids[i]))
+				trigger <- ids[i]
+			}
+		}
+	}()
+	go func() {
+		startTime = time.Now()
+		ticker := time.NewTicker(time.Second / 10)
+		for range ticker.C {
+			checkChunkIsAtNode(conf)
+		}
+	}()
+
+	//run the simulation
+	result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
+		Action:  action,
+		Trigger: trigger,
+		Expect: &simulations.Expectation{
+			Nodes: ids,
+			Check: triggerCheck,
+		},
+	})
+	//close(quitC)
+	if result.Error != nil {
+		return result.Error
+	}
+
+	return nil
+}
+
+func StartSyncing(s *Streamer, peerId discover.NodeID, po uint8, nn bool) {
+	lastPO := po
+	if nn {
+		lastPO = maxPO
+	}
+
+	for i := po; i <= lastPO; i++ {
+		s.Subscribe(peerId, "SYNC", newSyncLabel("LIVE", po), 0, 0, High, true)
+		s.Subscribe(peerId, "SYNC", newSyncLabel("HISTORY", po), 0, 0, Mid, false)
+	}
+}
+
+func checkChunkIsAtNode(conf *synctestConfig) {
+	allOk := true
+	for chunk, nodes := range conf.chunksToNodesMap {
+		for _, node := range nodes {
+			if ok, _ := stores[conf.addrToIdMap[string(conf.addrs[node])]].Get([]byte(chunk)); ok != nil {
+				if len(conf.retrievalMap[chunk]) == 0 {
+					conf.retrievalMap[chunk] = make(map[string]time.Duration)
+				}
+				conf.retrievalMap[chunk][string(conf.addrs[node])] = time.Since(startTime)
+			}
+			if conf.retrievalMap[chunk][string(conf.addrs[node])] == 0 {
+				allOk = false
+			}
+		}
+	}
+	if allOk {
+		log.Info("All chunks arrived at destination")
+		for ch, n := range conf.retrievalMap {
+			for a, t := range n {
+				log.Info(fmt.Sprintf("Chunk %s at node %s took %d ms", string(ch), string(a), t.Seconds()*1e3))
+			}
+		}
+	}
+}
+
+//map chunk keys to addresses which are responsible
+func mapKeysToNodes(conf *synctestConfig) *synctestConfig {
+	kmap := make(map[string][]int)
+	nodemap := make(map[string][]int)
+	//build a pot for chunk hashes
+	np := pot.NewPot(nil, 0)
+	mm := make(map[string]int)
+	for i, a := range conf.addrs {
+		mm[string(a)] = i
+		np, _, _ = pot.Add(np, a, pof)
+	}
+	//for each address, run EachNeighbour on the chunk hashes pot to identify closest nodes
+	fmt.Println(conf.chunks)
+	for i := 0; i < len(conf.chunks); i++ {
+		pl := 256 //highest proximity
+		var nns []int
+		np.EachNeighbour([]byte(conf.chunks[i]), pof, func(val pot.Val, po int) bool {
+			a := val.([]byte)
+			if pl == 256 || pl == po {
+				nns = append(nns, mm[string(a)])
+				nodemap[string(a)] = append(nodemap[string(a)], i)
+			}
+			if pl == 256 && len(nns) >= testMinProxBinSize {
+				pl = po
+			}
+			return true
+		})
+		//kmap[conf.chunks[i].String()] = nns
+		kmap[string(conf.chunks[i])] = nns
+		//log.Debug(fmt.Sprintf("Length for id %s: %d",ids[i],len(kmap[ids[i]])))
+	}
+	for k, v := range nodemap {
+		fmt.Print(fmt.Sprintf("Node %s: ", conf.addrToIdMap[k]))
+		for _, vv := range v {
+			fmt.Println(conf.chunks[vv])
+		}
+		fmt.Println(conf.addrToIdMap[k])
+		fmt.Println("-------------------------------")
+	}
+	for k, v := range kmap {
+		fmt.Print(fmt.Sprintf("Chunk %s: ", k))
+		for _, vv := range v {
+			fmt.Println(conf.addrToIdMap[string(conf.addrs[vv])])
+		}
+		fmt.Println("###############################")
+	}
+	conf.nodesToChunksMap = nodemap
+	conf.chunksToNodesMap = kmap
+	return conf
+}
+
+//upload a file(chunks) to a single local node store
+func uploadFileToSingleNodeStore(id discover.NodeID, chunkCount int) ([]storage.Key, error) {
+	log.Debug(fmt.Sprintf("Uploading to node id: %s", id))
+	lstore := stores[id]
+	size := chunkSize
+	dpa := storage.NewDPA(lstore, storage.NewChunkerParams())
+	dpa.Start()
+	var rootkeys []storage.Key
+	for i := 0; i < chunkCount; i++ {
+		rk, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+		wait()
+		if err != nil {
+			return nil, err
+		}
+		rootkeys = append(rootkeys, (rk))
+	}
+
+	defer dpa.Stop()
+
+	return rootkeys, nil
+}
+
+//initialize a network from a snapshot
+func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) {
+
+	adapter := "sim"
+
+	var a adapters.NodeAdapter
+	//add the streamer service to the node adapter
+	//discovery["streamer"] = NewStreamerService
+
+	if adapter == "exec" {
+		dirname, err := ioutil.TempDir(".", "")
+		if err != nil {
+			return nil, err
+		}
+		a = adapters.NewExecAdapter(dirname)
+	} else if adapter == "sock" {
+		a = adapters.NewSocketAdapter(services)
+	} else if adapter == "tcp" {
+		a = adapters.NewTCPAdapter(services)
+	} else if adapter == "sim" {
+		a = adapters.NewSimAdapter(services)
+	}
+
+	net := simulations.NewNetwork(a, &simulations.NetworkConfig{
+		ID:             "0",
+		DefaultService: "streamer",
+	})
+
+	f, err := os.Open(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+	jsonbyte, err := ioutil.ReadAll(f)
+	if err != nil {
+		return nil, err
+	}
+	var snap simulations.Snapshot
+	err = json.Unmarshal(jsonbyte, &snap)
+	if err != nil {
+		return nil, err
+	}
+	err = net.Load(&snap)
+	if err != nil {
+		return nil, err
+	}
+	return net, nil
+}