swarm: sync test cleanup,potential stall fix

swarm/network/stream/snapshot_retrieval_test.go

@@ -15,33 +15,29 @@
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
 package stream
 
-/*
-
 import (
-	"context"
+	//"context"
 	crand "crypto/rand"
 	"flag"
 	"fmt"
 	"io"
 	"math/rand"
-//	"os"
+	//	"os"
+	"github.com/ethereum/go-ethereum/log"
 	"testing"
 	"time"
-
+	//	"github.com/ethereum/go-ethereum/node"
-	"github.com/ethereum/go-ethereum/log"
+	//	"github.com/ethereum/go-ethereum/pot"
-//	"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"
-//	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	//	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
-	"github.com/ethereum/go-ethereum/swarm/network"
+	//"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())
@@ -49,47 +45,11 @@ func init() {
 	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_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) }
@@ -122,12 +82,11 @@ func benchmarkSyncing(b *testing.B, chunkCount int) {
 }
 
 */
-/*
 func retrievalTest(t *testing.T, chunkCount int) {
-  err := runRetrievalTest(chunkCount)
+	err := runRetrievalTest(chunkCount)
-  if err != nil {
+	if err != nil {
-    t.Fatal(err)
+		t.Fatal(err)
-  }
+	}
 }
 
 /*
@@ -142,165 +101,170 @@ 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 {
+	/*
+		//First load the snapshot from the file
+		net, err := initNetWithSnapshot()
+		if err != nil {
+			return err
+		}
+		defer net.Shutdown()
 
-  //reset global vars
+		//get the nodes of the network
-  resetVars()
+		nodes := net.GetNodes()
-  //First load the snapshot from the file
+		//select one index at random...
-  net,err := initNetWithSnapshot()
+		idx := rand.Intn(len(nodes))
-  if err != nil {
+		//...and get the the node at that index
-    return err
+		//this is the node selected for upload
-  }
+		uploadNode := nodes[idx]
-	defer net.Shutdown()
+		//now select a node at random which will be used to retrieve
-
+		ridx := rand.Intn(len(nodes))
-  //get the nodes of the network
+		//make sure uploadNode nad retrieveNode are not the same
-	nodes := net.GetNodes()
+		if ridx == idx {
-  //select one index at random...
+			if ridx == len(nodes)-1 {
-	idx   := rand.Intn(len(nodes))
+				ridx = 0
-  //...and get the the node at that index
+			} else {
-  //this is the node selected for upload
+				ridx += 1
-  node  := nodes[idx]
+			}
-  //iterate over all nodes...
+		}
-  for c:=0; c<len(nodes); c++ {
+		retrieveNode := nodes[ridx]
-    //create an array of discovery nodeIDS
+		//iterate over all nodes...
-    ids[c] = nodes[c].ID()
+		for c := 0; c < len(nodes); c++ {
-    //and a correspondent array of overlay addresses, 
+			//create an array of discovery nodeIDS
-    //later used for chunk proximity calculation
+			ids[c] = nodes[c].ID()
-    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))
+		// channel to signal simulation initialisation with action call complete
-    //select the !!!!NETstore!!! for the given node
+		// or node disconnections
-    /*
+		//disconnectC := make(chan error)
-    lstore := stores[id]
+		//quitC := make(chan struct{})
-    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
+		//after the test, clean up local stores initialized with createLocalStoreForId
-  ticker := time.NewTicker(time.Second * 1)
+		defer localStoreCleanup()
-  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
+		trigger := make(chan discover.NodeID)
-	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+		//triggerCheck defines what will be checked during the test
-	defer cancel()
+		triggerCheck := func(ctx context.Context, id discover.NodeID) (bool, error) {
-  //define the action to be performed before the test checks: start syncing
+			select {
-	action := func(ctx context.Context) error {
+			case <-ctx.Done():
-		// need to wait till an aynchronous process registers the peers in streamer.peers
+				return false, ctx.Err()
-		// that is used by Subscribe
+				//case <-disconnectC:
-		// the global peerCount function tells how many connections each node has
+				//  log.Error("Disconnect event detected")
-		// TODO: this is to be reimplemented with peerEvent watcher without global var
+				//  return false, ctx.Err()
-		i := 0
+			default:
-		for err := range waitPeerErrC {
+			}
+
+			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 fmt.Errorf("error waiting for peers: %s", err)
+				return err
-			}
-			i++
-			if i == len(ids)-1 {
-				break
 			}
+			//finally map chunks to the closest addresses
+			//chunksForAddressesMap = mapIdsToKeys(chunks, ids)
+			log.Debug(fmt.Sprintf("%v", chunksForAddressesMap))
+
+			return nil
 		}
-
+		//run the simulation
-		// each node Subscribes to each other's swarmChunkServerStreamName
+		result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
-		for j := 0; j < len(ids); j++ {
+			Action:  action,
-      log.Debug(fmt.Sprintf("subscribe: %d",j))
+			Trigger: trigger,
-      ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+			Expect: &simulations.Expectation{
-      defer cancel()
+				Nodes: ids,
-      client,err := net.GetNode(ids[j]).Client()
+				Check: triggerCheck,
-      if err != nil {
+			},
-        return err
+		})
-      }
+		//close(quitC)
-      //RPC call to subscribe, select bin 0
+		if result.Error != nil {
-      //client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{0}, 0, 0, Top, false)
+			return result.Error
-      // report disconnect events to the error channel cos peers should not disconnect
+		}
-      //err = streamTesting.WatchDisconnections(ids[j], client, disconnectC, quitC)
+	*/
-      //if err != nil {
+	return 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))
+	log.Debug(fmt.Sprintf("Uploading to node id: %s", id))
-  lstore := stores[id]
+	lstore := stores[id]
 	size := chunkCount * chunkSize
-  dpa := storage.NewDPA(lstore, storage.NewChunkerParams())
+	dpa := storage.NewDPA(lstore, storage.NewChunkerParams())
-  dpa.Start()
+	dpa.Start()
-  rootHash, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+	rootHash, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
-  wait()
+	wait()
-  if err != nil {
+	if err != nil {
-    return nil, err
+		return nil, err
-  }
+	}
 
-  defer dpa.Stop()
+	defer dpa.Stop()
 
-  return rootHash, nil
+	return rootHash, nil
 }
-*/

swarm/network/stream/snapshot_sync_test.go

@@ -28,12 +28,14 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p"
 	"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/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
-	//streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
+	streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -42,13 +44,15 @@ const testMinProxBinSize = 2
 var (
 	pof = pot.DefaultPof(256)
 
+	conf      *synctestConfig
 	startTime time.Time
 	ids       []discover.NodeID
 	datadirs  map[discover.NodeID]string
-	conf      *synctestConfig
 	ppmap     map[discover.NodeID]*network.PeerPot
 
-	printed bool
+	requestedSubscriptions int
+	receivedSubscriptions  int
+	printed                bool
 )
 
 type synctestConfig struct {
@@ -213,15 +217,14 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 	cleanup := func() {
 		timingTicker.Stop()
 		actionTicker.Stop()
-		//close(trigger)
 		close(quitC)
 		close(disconnectC)
-		close(waitPeerErrC)
 		//after the test, clean up local stores initialized with createLocalStoreForId
 		localStoreCleanup()
 		//shutdown the snapshot network
 		net.Shutdown()
-		//datadirsCleanup()
+		//finally clear all data directories
+		datadirsCleanup()
 	}
 	defer cleanup()
 	//get the nodes of the network
@@ -251,35 +254,11 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 
 	ppmap = network.NewPeerPot(testMinProxBinSize, ids, conf.addrs)
 
+	subscriptionsDone := make(chan struct{})
+	errc := make(chan error)
 	//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
-
-		//TODO: VALIDATE THE ASSUMPTION THAT THE FOLLOWING CODE IS NOT NEEDED,
-		//AS THE SNAPSHOT CONSTRUCTS ALL CONNECTIONS DURING LOAD, SO WE DON'T NEED TO WAIT HERE?
-
-		/*
-			i := 0
-			for err := range waitPeerErrC {
-				fmt.Println("aaaa")
-				if err != nil {
-					return fmt.Errorf("error waiting for peers: %s", err)
-				}
-				i++
-				if i == len(ids)-1 {
-					break
-				}
-			}
-
-			// wait for connections
-			time.Sleep(5 * time.Second)
-		*/
-
 		log.Info("Setting up stream subscription")
-
 		// each node Subscribes to each other's swarmChunkServerStreamName
 		for j, id := range ids {
 			log.Trace(fmt.Sprintf("subscribe: %d", j))
@@ -290,6 +269,12 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 				return err
 			}
 
+			/*
+				watchCtx, watchCancel := context.WithTimeout(ctx, 15*time.Second)
+				defer watchCancel()
+				watchSubscriptionEvents(watchCtx, id, client, subscriptionsDone, errc)
+			*/
+
 			if log.Lvl(*loglevel) == log.LvlDebug {
 				//print uploading node kademlia
 				if j == idx {
@@ -303,20 +288,26 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 					log.Debug(kt)
 				}
 			}
-			//err = streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
+			err = streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
-			//if err != nil {
+			if err != nil {
-			//	return err
+				return err
-			//}
+			}
 			err = client.CallContext(ctx, nil, "stream_startSyncing")
 			if err != nil {
 				return err
 			}
 		}
 
+		select {
+		case <-subscriptionsDone:
+			close(subscriptionsDone)
+		case err := <-errc:
+			return err
+		}
 		log.Info("Stream subscriptions successfully requested")
 		// wait for subscritpions
 		//TODO: Implement a proper sync mechanism so that we don't need to Sleep()
-		time.Sleep(10 * time.Second)
+		//time.Sleep(10 * time.Second)
 		//now upload the chunks to the selected random single node
 		conf.chunks, err = uploadFileToSingleNodeStore(node.ID(), chunkCount)
 		if err != nil {
@@ -425,62 +416,26 @@ func (r *TestRegistry) StartSyncing(ctx context.Context) error {
 		log.Debug(fmt.Sprintf("IS HEALTHY: %t", h.GotNN && h.KnowNN && h.Full))
 	}
 
-	var kadDepth int
-
-	r.delivery.overlay.EachConn(nil, 256, func(addr network.OverlayConn, po int, nn bool) bool {
-		// TODO: stop or expose by kademlia
-		if nn {
-			kadDepth = po
-		}
-		return true
-	})
-
 	kad, ok := r.delivery.overlay.(*network.Kademlia)
 	if !ok {
 		return fmt.Errorf("Not a Kademlia!")
 	}
 
-	var startPo int
-	var endPo int
-	var i int
-
 	//iterate over each bin and solicit needed subscription to bins
-	kad.EachBin(r.addr.Over(), pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
+	kad.EachBin(r.addr.Over(), pof, 0, func(conn network.OverlayConn, po int) bool {
 		//identify begin and start index of the bin(s) we want to subscribe to
-		if po < kadDepth {
+		log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), conf.addrToIdMap[string(conn.Address())], po))
-			//not nn
+
-			endPo = po
+		err = r.RequestSubscription(conf.addrToIdMap[string(conn.Address())], NewStream("SYNC", []byte{uint8(po)}, true), &Range{}, Top)
-			if i > 0 {
+		if err != nil {
-				startPo = endPo + 1
+			log.Error(fmt.Sprintf("Error in RequestSubsciption! %v", err))
-			}
+			return false
-		} else if endPo < kadDepth || endPo == 0 {
-			if po == 0 && kadDepth == 0 {
-				startPo = endPo
-			} else {
-				startPo = endPo + 1
-			}
-			endPo = maxPO
 		}
-
+		requestedSubscriptions += 1
-		// now iterate and subscribe
+		//fmt.Println(requestedSubscriptions)
-		for bin := po - startPo; bin <= endPo; bin++ {
-
-			f(func(val pot.Val, i int) bool {
-				a := val.(network.OverlayPeer)
-				log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), conf.addrToIdMap[string(a.Address())], bin))
-
-				err = r.RequestSubscription(conf.addrToIdMap[string(a.Address())], NewStream("SYNC", []byte{uint8(bin)}, true), &Range{}, Top)
-				if err != nil {
-					log.Error(fmt.Sprintf("Error in RequestSubsciption! %v", err))
-					return false
-				}
-				return true
-			})
-		}
-		i++
 		return true
-	})
 
+	})
 	return nil
 }
 
@@ -607,6 +562,38 @@ func uploadFileToSingleNodeStore(id discover.NodeID, chunkCount int) ([]storage.
 	return rootkeys, nil
 }
 
+//Here we wait until all connections from the snapshot are up
+func waitForSnapshotConnsUp(ctx context.Context, net *simulations.Network, done chan struct{}, connCount int, errc chan error) {
+	arrivedConns := 0
+	events := make(chan *simulations.Event)
+	//subscribe to all events from the network
+	sub := net.Events().Subscribe(events)
+	defer sub.Unsubscribe()
+
+	for {
+		select {
+		case <-ctx.Done():
+			errc <- fmt.Errorf("Timeout waiting for Snapshot connections")
+		case event := <-events:
+			//if the event is of type connection, is a Live event and the connection is up
+			//NOTE; this will require that all connections are UP in the snapshot!
+			if event.Type == simulations.EventTypeConn && !event.Control && event.Conn.Up {
+				arrivedConns++
+				//the amount of expected connections has been reached, so we can stop waiting
+				if arrivedConns == connCount {
+					done <- struct{}{}
+					return
+				}
+			}
+		case err := <-sub.Err():
+			if err != nil {
+				errc <- err
+			}
+		}
+	}
+	return
+}
+
 //initialize a network from a snapshot
 func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) {
 
@@ -651,10 +638,60 @@ func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) {
 	if err != nil {
 		return nil, err
 	}
+
+	log.Info("Waiting for p2p connections to be established...")
+	errc := make(chan error)
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	connCount := len(snap.Conns)
+	done := make(chan struct{})
+	go waitForSnapshotConnsUp(ctx, net, done, connCount, errc)
+
 	err = net.Load(&snap)
 	if err != nil {
 		return nil, err
 	}
-	log.Info("Snapshot loaded")
+	select {
+	case <-done:
+		close(done)
+	case err = <-errc:
+		return nil, err
+	}
+	log.Info("Snapshot loaded and connections established")
 	return net, nil
 }
+
+func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rpc.Client, done chan struct{}, errc chan error) {
+	events := make(chan *p2p.PeerEvent)
+	sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents")
+	if err != nil {
+		errc <- fmt.Errorf("error getting peer events for node %v: %s", id, err)
+		return
+	}
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case e := <-events:
+				fmt.Println(e)
+				fmt.Println(*e.MsgCode)
+				if e.Type == p2p.PeerEventTypeMsgRecv && e.Protocol == "stream" && e.MsgCode != nil && *e.MsgCode == 1 {
+					fmt.Println(receivedSubscriptions)
+					fmt.Println(requestedSubscriptions)
+					receivedSubscriptions += 1
+					if receivedSubscriptions == requestedSubscriptions {
+						done <- struct{}{}
+						return
+					}
+				}
+			case err := <-sub.Err():
+				if err != nil {
+					errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err)
+					return
+				}
+			}
+		}
+	}()
+	return
+}

swarm/network/stream/syncer.go

@@ -193,9 +193,9 @@ func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
 
 // NeedData
 func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
-	chunk, _ := s.db.GetOrCreateRequest(key)
+	chunk, need := s.db.GetOrCreateRequest(key)
 	// TODO: we may want to request from this peer anyway even if the request exists
-	if chunk.ReqC == nil {
+	if chunk.ReqC == nil || need == false {
 		return nil
 	}
 	// create request and wait until the chunk data arrives and is stored