swarm: consolidate snapshot sync and retrieval tests

swarm/network/stream/snapshot_retrieval_test.go

@@ -17,31 +17,26 @@ package stream
 
 import (
 	"context"
-	//	crand "crypto/rand"
 	"fmt"
-	//"io"
 	"math/rand"
-	//"sync"
 	"testing"
 	"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"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
 func initRetrievalTest() {
+	//global func to get overlay address from discover ID
 	toAddr = func(id discover.NodeID) *network.BzzAddr {
 		addr := network.NewAddrFromNodeID(id)
 		return addr
 	}
+	//global func to create local store
 	createStoreFunc = createTestLocalStorageForId
 	//local stores
 	stores = make(map[discover.NodeID]storage.ChunkStore)
@@ -49,16 +44,15 @@ func initRetrievalTest() {
 	datadirs = make(map[discover.NodeID]string)
 	//deliveries for each node
 	deliveries = make(map[discover.NodeID]*Delivery)
+	//global retrieve func
 	getRetrieveFunc = func(id discover.NodeID) func(chunk *storage.Chunk) error {
 		return func(chunk *storage.Chunk) error {
 			skipCheck := true
-			//fmt.Println(fmt.Sprintf("-- %s", id))
 			return deliveries[id].RequestFromPeers(chunk.Key[:], skipCheck)
 		}
 	}
 	//registries, map of discover.NodeID to its streamer
 	registries = make(map[discover.NodeID]*TestRegistry)
-	//channel to wait for peers connected
 	//not needed for this test but required from common_test for NewStreamService
 	waitPeerErrC = make(chan error)
 	//also not needed for this test but required for NewStreamService
@@ -70,17 +64,27 @@ func initRetrievalTest() {
 	}
 }
 
+//This test is a retrieval test for nodes.
+//One node is randomly selected to be the pivot node.
+//A configurable number of chunks and nodes can be
+//provided to the test, the number of chunks is uploaded
+//to the pivot node and other nodes try to retrieve the chunk(s).
+//Number of chunks and nodes can be provided via commandline too.
 func TestRetrieval(t *testing.T) {
-
+	//if nodes/chunks have been provided via commandline,
+	//run the tests with these values
 	if *nodes != 0 && *chunks != 0 {
 		retrievalTest(t, *chunks, *nodes)
 	} else {
 		var nodeCnt []int
 		var chnkCnt []int
+		//if the `longrunning` flag has been provided
+		//run more test combinations
 		if *longrunning {
 			nodeCnt = []int{16, 32, 128}
 			chnkCnt = []int{4, 32, 256}
 		} else {
+			//default test
 			nodeCnt = []int{16}
 			chnkCnt = []int{32}
 		}
@@ -92,6 +96,7 @@ func TestRetrieval(t *testing.T) {
 	}
 }
 
+//Every test runs 3 times, a live, a history, and a live AND history
 func retrievalTest(t *testing.T, chunkCount int, nodeCount int) {
 	//test live and NO history
 	log.Info("Testing live and no history", "chunkCount", chunkCount, "nodeCount", nodeCount)
@@ -119,22 +124,33 @@ func retrievalTest(t *testing.T, chunkCount int, nodeCount int) {
 }
 
 /*
-The test generates the given number of chunks,
+The test generates the given number of chunks.
-then uploads these to a random node.
+
-Afterwards for every chunk generated, the nearest node addresses
+The upload is done by dependency to the global
-are identified, syncing is started, and finally we verify
+`live` and `history` variables;
-that the nodes closer to the chunk addresses actually do have
+
-the chunks in their local stores.
+If `live` is set, first stream subscriptions are established, then
+upload to a random node.
+
+If `history` is enabled, first upload then build up subscriptions.
 
 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.
+kademlia network. Nevertheless a health check runs in the
+simulation's `action` function.
+
+The snapshot should have 'streamer' in its service list.
 */
 func runRetrievalTest(chunkCount int, nodeCount int) error {
+	//for every run (live, history), int the variables
 	initRetrievalTest()
+	//the ids of the snapshot nodes, initiate only now as we need nodeCount
 	ids = make([]discover.NodeID, nodeCount)
+	//channel to check for disconnection errors
 	disconnectC := make(chan error)
+	//channel to close disconnection watcher routine
 	quitC := make(chan struct{})
+	//the test conf (using same as in `snapshot_sync_test`
 	conf = &synctestConfig{}
 	//map of discover ID to indexes of chunks expected at that ID
 	conf.idToChunksMap = make(map[discover.NodeID][]int)
@@ -142,6 +158,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 	conf.idToAddrMap = make(map[discover.NodeID][]byte)
 	//map of overlay address to discover ID
 	conf.addrToIdMap = make(map[string]discover.NodeID)
+	//array where the generated chunk hashes will be stored
 	conf.chunks = make([]storage.Key, 0)
 	//load nodes from the snapshot file
 	net, err := initNetWithSnapshot(nodeCount)
@@ -179,13 +196,11 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 	//needed for healthy call
 	ppmap = network.NewPeerPot(testMinProxBinSize, ids, conf.addrs)
 
-	// channel to signal simulation initialisation with action call complete
-	// or node disconnections
-	//disconnectC := make(chan error)
-	//quitC := make(chan struct{})
-
 	trigger := make(chan discover.NodeID)
+	//simulation action
 	action := func(ctx context.Context) error {
+		//first run the health check on all nodes,
+		//wait until nodes are all healthy
 		ticker := time.NewTicker(200 * time.Millisecond)
 		defer ticker.Stop()
 		for range ticker.C {
@@ -226,7 +241,14 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 		defer watchCancel()
 
 		log.Info("Setting up stream subscription")
-		// each node Subscribes to each other's swarmChunkServerStreamName
+		//We need two iterations, one to subscribe to the subscription events
+		//(so we know when setup phase is finished), and one to
+		//actually run the stream subscriptions. We can't do it in the same iteration,
+		//because while the first nodes in the loop are setting up subscriptions,
+		//the latter ones have not subscribed to listen to peer events yet,
+		//and then we miss events.
+
+		//first iteration: setup disconnection watcher and subscribe to peer events
 		for j, id := range ids {
 			log.Trace(fmt.Sprintf("Subscribe to subscription events: %d", j))
 			client, err := net.GetNode(id).Client()
@@ -239,9 +261,11 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 				return err
 			}
 
+			//check for `SubscribeMsg` events to know when setup phase is complete
 			watchSubscriptionEvents(ctx, id, client, errc)
 		}
 
+		//second iteration: start syncing and setup stream subscriptions
 		for j, id := range ids {
 			log.Trace(fmt.Sprintf("Start syncing and stream subscriptions: %d", j))
 			client, err := net.GetNode(id).Client()
@@ -254,7 +278,10 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 			if err != nil {
 				return err
 			}
+			//increment the number of subscriptions we need to wait for
+			//by the count returned from startSyncing (SYNC subscriptions)
 			subscriptionCount += cnt
+			//now also add the number of RETRIEVAL_REQUEST subscriptions
 			for snid := range registries[id].peers {
 				subscriptionCount++
 				err = client.CallContext(ctx, nil, "stream_subscribeStream", snid, NewStream(swarmChunkServerStreamName, "", false), nil, Top)
@@ -265,11 +292,15 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 		}
 
 		//now wait until the number of expected subscriptions has been finished
+		//`watchSubscriptionEvents` will write with a `nil` value to errc
+		//every time a `SubscriptionMsg` has been received
 		for err := range errc {
 			if err != nil {
 				return err
 			}
+			//`nil` received, decrement count
 			subscriptionCount--
+			//all subscriptions received
 			if subscriptionCount == 0 {
 				break
 			}
@@ -294,6 +325,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 	//check defines what will be checked during the test
 	check := func(ctx context.Context, id discover.NodeID) (bool, error) {
 
+		//don't check the uploader node
 		if id == uploadNode.ID() {
 			return true, nil
 		}
@@ -310,9 +342,12 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 		//if there are more than one chunk, test only succeeds if all expected chunks are found
 		allSuccess := true
 
+		//check on the node's dpa (netstore)
 		dpa := registries[id].dpa
+		//check all chunks
 		for _, chnk := range conf.chunks {
 			reader := dpa.Retrieve(chnk)
+			//assuming that reading the Size of the chunk is enough to know we found it
 			if s, err := reader.Size(nil); err != nil || s != chunkSize {
 				allSuccess = false
 				log.Warn("Retrieve error", "err", err, "chunk", chnk, "nodeId", id)
@@ -350,29 +385,10 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 			Check: check,
 		},
 	})
-	//close(quitC)
+
 	if result.Error != nil {
 		return result.Error
 	}
+
 	return nil
 }
-
-//upload a file(chunks)
-/*
-func uploadRandomChunks(net *simulations.Network, chunkCount int) 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

@@ -82,7 +82,7 @@ func initSyncTest() {
 		addr := network.NewAddrFromNodeID(id)
 		return addr
 	}
-
+	//global func to create local store
 	createStoreFunc = createTestLocalStorageForId
 	//local stores
 	stores = make(map[discover.NodeID]storage.ChunkStore)
@@ -92,7 +92,6 @@ func initSyncTest() {
 	deliveries = make(map[discover.NodeID]*Delivery)
 	//registries, map of discover.NodeID to its streamer
 	registries = make(map[discover.NodeID]*TestRegistry)
-	//channel to wait for peers connected
 	//not needed for this test but required from common_test for NewStreamService
 	waitPeerErrC = make(chan error)
 	//also not needed for this test but required for NewStreamService
@@ -104,19 +103,29 @@ func initSyncTest() {
 	}
 }
 
-//This file executes a number of syncing tests
+//This test is a syncing test for nodes.
-//node and chunk number can be provided via flags
+//One node is randomly selected to be the pivot node.
+//A configurable number of chunks and nodes can be
+//provided to the test, the number of chunks is uploaded
+//to the pivot node, and we check that nodes get the chunks
+//they are expected to store based on the syncing protocol.
+//Number of chunks and nodes can be provided via commandline too.
 func TestSyncing(t *testing.T) {
+	//if nodes/chunks have been provided via commandline,
+	//run the tests with these values
 	if *nodes != 0 && *chunks != 0 {
 		log.Info(fmt.Sprintf("Running test with %d chunks and %d nodes...", *chunks, *nodes))
 		testSyncing(t, *chunks, *nodes)
 	} else {
 		var nodeCnt []int
 		var chnkCnt []int
+		//if the `longrunning` flag has been provided
+		//run more test combinations
 		if *longrunning {
 			chnkCnt = []int{1, 8, 32, 256, 1024}
 			nodeCnt = []int{16, 32, 64, 128, 256}
 		} else {
+			//default test
 			chnkCnt = []int{4, 32}
 			nodeCnt = []int{32, 16}
 		}
@@ -129,11 +138,9 @@ func TestSyncing(t *testing.T) {
 	}
 }
 
-//do run the tests
+//Do run the tests
+//Every test runs 3 times, a live, a history, and a live AND history
 func testSyncing(t *testing.T, chunkCount int, nodeCount int) {
-	initSyncTest()
-	ids = make([]discover.NodeID, nodeCount)
-
 	//test live and NO history
 	log.Info("Testing live and no history")
 	live = true
@@ -160,12 +167,19 @@ func testSyncing(t *testing.T, chunkCount int, nodeCount int) {
 }
 
 /*
-The test generates the given number of chunks,
+The test generates the given number of chunks
-then uploads these to a random node.
+
-Afterwards for every chunk generated, the nearest node addresses
+The upload is done by dependency to the global
-are identified, syncing is started, and finally we verify
+`live` and `history` variables;
-that the nodes closer to the chunk addresses actually do have
+
-the chunks in their local stores.
+If `live` is set, first stream subscriptions are established, then
+upload to a random node.
+
+If `history` is enabled, first upload then build up subscriptions.
+
+For every chunk generated, the nearest node addresses
+are identified, 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
@@ -180,6 +194,9 @@ For every test run, a series of three tests will be executed:
   are uploaded twice, once before and once after subscriptions
 */
 func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
+	initSyncTest()
+	//the ids of the snapshot nodes, initiate only now as we need nodeCount
+	ids = make([]discover.NodeID, nodeCount)
 	//initialize the test struct
 	conf = &synctestConfig{}
 	//map of discover ID to indexes of chunks expected at that ID
@@ -188,12 +205,13 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 	conf.idToAddrMap = make(map[discover.NodeID][]byte)
 	//map of overlay address to discover ID
 	conf.addrToIdMap = make(map[string]discover.NodeID)
+	//array where the generated chunk hashes will be stored
 	conf.chunks = make([]storage.Key, 0)
-	//First load the snapshot from the file
+	//channel to trigger node checks in the simulation
 	trigger := make(chan discover.NodeID)
-	// channel to signal simulation initialisation with action call complete
+	//channel to check for disconnection errors
-	// or node disconnections
 	disconnectC := make(chan error)
+	//channel to close disconnection watcher routine
 	quitC := make(chan struct{})
 
 	//load nodes from the snapshot file
@@ -246,6 +264,8 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 
 	//define the action to be performed before the test checks: start syncing
 	action := func(ctx context.Context) error {
+		//first run the health check on all nodes,
+		//wait until nodes are all healthy
 		ticker := time.NewTicker(200 * time.Millisecond)
 		defer ticker.Stop()
 		for range ticker.C {
@@ -290,7 +310,15 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 		defer watchCancel()
 
 		log.Info("Setting up stream subscription")
-		// each node Subscribes to each other's swarmChunkServerStreamName
+
+		//We need two iterations, one to subscribe to the subscription events
+		//(so we know when setup phase is finished), and one to
+		//actually run the stream subscriptions. We can't do it in the same iteration,
+		//because while the first nodes in the loop are setting up subscriptions,
+		//the latter ones have not subscribed to listen to peer events yet,
+		//and then we miss events.
+
+		//first iteration: setup disconnection watcher and subscribe to peer events
 		for j, id := range ids {
 			log.Trace(fmt.Sprintf("Subscribe to subscription events: %d", j))
 			client, err := net.GetNode(id).Client()
@@ -309,19 +337,6 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 				rpcSubscriptionsWg.Done()
 			}()
 
-			if log.Lvl(*loglevel) >= log.LvlTrace {
-				//this will print the kademlia tables of all nodes
-				//to only print the kademlia of the pivot node,
-				//use: if j == idx {}
-				var kt string
-				err = client.CallContext(ctx, &kt, "stream_getKad")
-				if err != nil {
-					return err
-				}
-
-				log.Debug("kad table " + node.ID().String())
-				log.Debug(kt)
-			}
 			//watch for peers disconnecting
 			wdDoneC, err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
 			if err != nil {
@@ -334,6 +349,7 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 			}()
 		}
 
+		//second iteration: start syncing
 		for j, id := range ids {
 			log.Trace(fmt.Sprintf("Start syncing subscriptions: %d", j))
 			client, err := net.GetNode(id).Client()
@@ -346,15 +362,20 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 			if err != nil {
 				return err
 			}
+			//increment the number of subscriptions we need to wait for
+			//by the count returned from startSyncing (SYNC subscriptions)
 			subscriptionCount += cnt
 		}
 
 		//now wait until the number of expected subscriptions has been finished
+		//`watchSubscriptionEvents` will write with a `nil` value to errc
 		for err := range errc {
 			if err != nil {
 				return err
 			}
+			//`nil` received, decrement count
 			subscriptionCount--
+			//all subscriptions received
 			if subscriptionCount == 0 {
 				break
 			}
@@ -449,12 +470,10 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 	return nil
 }
 
-//Show kademlia of uploading node for debugging
-func (r *TestRegistry) GetKad(ctx context.Context) string {
-	return r.delivery.overlay.String()
-}
-
 //the server func to start syncing
+//issues `RequestSubscriptionMsg` to peers, based on po, by iterating over
+//the kademlia's `EachBin` function.
+//returns the number of subscriptions requested
 func (r *TestRegistry) StartSyncing(ctx context.Context) (int, error) {
 	var err error