network: new stream! protocol and pull syncer implementation

api/http/test_server.go

@@ -57,7 +57,7 @@ func NewTestSwarmServer(t *testing.T, serverFunc func(*api.API, *pin.API) TestSe
 	}
 
 	tags := chunk.NewTags()
-	fileStore := storage.NewFileStore(localStore, storage.NewFileStoreParams(), tags)
+	fileStore := storage.NewFileStore(localStore, localStore, storage.NewFileStoreParams(), tags)
 
 	// Swarm feeds test setup
 	feedsDir, err := ioutil.TempDir("", "swarm-feeds-test")

api/inspector.go

@@ -18,24 +18,29 @@ package api
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
 	"time"
 
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethersphere/swarm/log"
 	"github.com/ethersphere/swarm/network"
+	stream "github.com/ethersphere/swarm/network/stream/v2"
 	"github.com/ethersphere/swarm/storage"
 )
 
+const InspectorIsPullSyncingTolerance = 15 * time.Second
+
 type Inspector struct {
 	api      *API
 	hive     *network.Hive
 	netStore *storage.NetStore
+	stream   *stream.Registry
 }
 
-func NewInspector(api *API, hive *network.Hive, netStore *storage.NetStore) *Inspector {
-	return &Inspector{api, hive, netStore}
+func NewInspector(api *API, hive *network.Hive, netStore *storage.NetStore, pullSyncer *stream.Registry) *Inspector {
+	return &Inspector{api, hive, netStore, pullSyncer}
 }
 
 // Hive prints the kademlia table
@@ -49,15 +54,12 @@ func (i *Inspector) KademliaInfo() network.KademliaInfo {
 }
 
 func (i *Inspector) IsPullSyncing() bool {
-	lastReceivedChunksMsg := metrics.GetOrRegisterGauge("network.stream.received_chunks", nil)
-
-	// last received chunks msg time
-	lrct := time.Unix(0, lastReceivedChunksMsg.Value())
+	t := i.stream.LastReceivedChunkTime()
 
 	// if last received chunks msg time is after now-15sec. (i.e. within the last 15sec.) then we say that the node is still syncing
 	// technically this is not correct, because this might have been a retrieve request, but for the time being it works for our purposes
 	// because we know we are not making retrieve requests on the node while checking this
-	return lrct.After(time.Now().Add(-15 * time.Second))
+	return t.After(time.Now().Add(-InspectorIsPullSyncingTolerance))
 }
 
 // DeliveriesPerPeer returns the sum of chunks we received from a given peer
@@ -96,3 +98,15 @@ func (i *Inspector) Has(chunkAddresses []storage.Address) string {
 
 	return strings.Join(hostChunks, "")
 }
+
+func (i *Inspector) PeerStreams() (string, error) {
+	peerInfo, err := i.stream.PeerInfo()
+	if err != nil {
+		return "", err
+	}
+	v, err := json.Marshal(peerInfo)
+	if err != nil {
+		return "", err
+	}
+	return string(v), nil
+}

api/inspector_test.go

@@ -0,0 +1,66 @@
+package api
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"io/ioutil"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/ethersphere/swarm/network"
+	stream "github.com/ethersphere/swarm/network/stream/v2"
+	"github.com/ethersphere/swarm/storage"
+	"github.com/ethersphere/swarm/storage/localstore"
+
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethersphere/swarm/state"
+)
+
+// TestInspectorPeerStreams validates that response from RPC peerStream has at
+// least some data.
+func TestInspectorPeerStreams(t *testing.T) {
+	dir, err := ioutil.TempDir("", "swarm-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(dir)
+
+	baseKey := make([]byte, 32)
+	_, err = rand.Read(baseKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	localStore, err := localstore.New(dir, baseKey, &localstore.Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	netStore := storage.NewNetStore(localStore, baseKey, enode.ID{})
+
+	i := NewInspector(nil, nil, netStore, stream.New(state.NewInmemoryStore(), baseKey, stream.NewSyncProvider(netStore, network.NewKademlia(
+		baseKey,
+		network.NewKadParams(),
+	), false, false)))
+
+	server := rpc.NewServer()
+	if err := server.RegisterName("inspector", i); err != nil {
+		t.Fatal(err)
+	}
+
+	client := rpc.DialInProc(server)
+
+	var peerInfo string
+
+	err = client.Call(&peerInfo, "inspector_peerStreams")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !strings.Contains(peerInfo, `"base":"`+hex.EncodeToString(baseKey)[:16]+`"`) {
+		t.Error("missing base key in response")
+	}
+
+	t.Log(peerInfo)
+}

api/manifest_test.go

@@ -43,7 +43,7 @@ func manifest(paths ...string) (manifestReader storage.LazySectionReader) {
 
 func testGetEntry(t *testing.T, path, match string, multiple bool, paths ...string) *manifestTrie {
 	quitC := make(chan bool)
-	fileStore := storage.NewFileStore(nil, storage.NewFileStoreParams(), chunk.NewTags())
+	fileStore := storage.NewFileStore(nil, nil, storage.NewFileStoreParams(), chunk.NewTags())
 	ref := make([]byte, fileStore.HashSize())
 	trie, err := readManifest(manifest(paths...), ref, fileStore, false, quitC, NOOPDecrypt)
 	if err != nil {
@@ -100,7 +100,7 @@ func TestGetEntry(t *testing.T) {
 func TestExactMatch(t *testing.T) {
 	quitC := make(chan bool)
 	mf := manifest("shouldBeExactMatch.css", "shouldBeExactMatch.css.map")
-	fileStore := storage.NewFileStore(nil, storage.NewFileStoreParams(), chunk.NewTags())
+	fileStore := storage.NewFileStore(nil, nil, storage.NewFileStoreParams(), chunk.NewTags())
 	ref := make([]byte, fileStore.HashSize())
 	trie, err := readManifest(mf, ref, fileStore, false, quitC, nil)
 	if err != nil {
@@ -133,7 +133,7 @@ func TestAddFileWithManifestPath(t *testing.T) {
 	reader := &storage.LazyTestSectionReader{
 		SectionReader: io.NewSectionReader(bytes.NewReader(manifest), 0, int64(len(manifest))),
 	}
-	fileStore := storage.NewFileStore(nil, storage.NewFileStoreParams(), chunk.NewTags())
+	fileStore := storage.NewFileStore(nil, nil, storage.NewFileStoreParams(), chunk.NewTags())
 	ref := make([]byte, fileStore.HashSize())
 	trie, err := readManifest(reader, ref, fileStore, false, nil, NOOPDecrypt)
 	if err != nil {

cmd/swarm-smoke/upload_and_sync.go

@@ -156,6 +156,7 @@ func trackChunks(testData []byte) error {
 	wg.Wait()
 
 	checkChunksVsMostProxHosts(addrs, allHostChunks, bzzAddrs)
+	metrics.GetOrRegisterGauge("deployment.nodes", nil).Update(int64(len(hosts)))
 
 	if !hasErr {
 		// remove the chunks stored on the uploader node
@@ -207,11 +208,10 @@ func checkChunksVsMostProxHosts(addrs []storage.Address, allHostChunks map[strin
 			}
 		}
 
-		log.Debug("sync mode", "sync mode", syncMode)
-
 		if syncMode == "pullsync" || syncMode == "both" {
 			for _, maxProxHost := range maxProxHosts {
 				if allHostChunks[maxProxHost][i] == '0' {
+					metrics.GetOrRegisterCounter("upload-and-sync.pull-sync.chunk-not-max-prox", nil).Inc(1)
 					log.Error("chunk not found at max prox host", "ref", addrs[i], "host", maxProxHost, "bzzAddr", bzzAddrs[maxProxHost])
 				} else {
 					log.Trace("chunk present at max prox host", "ref", addrs[i], "host", maxProxHost, "bzzAddr", bzzAddrs[maxProxHost])
@@ -220,6 +220,7 @@ func checkChunksVsMostProxHosts(addrs []storage.Address, allHostChunks map[strin
 
 			// if chunk found at less than 2 hosts, which is actually less that the min size of a NN
 			if foundAt < 2 {
+				metrics.GetOrRegisterCounter("upload-and-sync.pull-sync.chunk-less-nn", nil).Inc(1)
 				log.Error("chunk found at less than two hosts", "foundAt", foundAt, "ref", addrs[i])
 			}
 		}
@@ -235,6 +236,7 @@ func checkChunksVsMostProxHosts(addrs []storage.Address, allHostChunks map[strin
 
 			if !found {
 				for _, maxProxHost := range maxProxHosts {
+					metrics.GetOrRegisterCounter("upload-and-sync.push-sync.chunk-not-max-prox", nil).Inc(1)
 					log.Error("chunk not found at any max prox host", "ref", addrs[i], "hosts", maxProxHost, "bzzAddr", bzzAddrs[maxProxHost])
 				}
 			}
@@ -270,6 +272,8 @@ func uploadAndSync(c *cli.Context, randomBytes []byte) error {
 	}
 	t2 := time.Since(t1)
 	metrics.GetOrRegisterResettingTimer("upload-and-sync.upload-time", nil).Update(t2)
+	uploadSpeed := float64(len(randomBytes)) / t2.Seconds() // bytes per second
+	metrics.GetOrRegisterGauge("upload-and-sync.upload-speed", nil).Update(int64(uploadSpeed))
 
 	fhash, err := digest(bytes.NewReader(randomBytes))
 	if err != nil {
@@ -315,6 +319,9 @@ func uploadAndSync(c *cli.Context, randomBytes []byte) error {
 		ended := time.Since(start)
 
 		metrics.GetOrRegisterResettingTimer("upload-and-sync.single.fetch-time", nil).Update(ended)
+		downloadSpeed := float64(len(randomBytes)) / ended.Seconds() // bytes per second
+		metrics.GetOrRegisterGauge("upload-and-sync.download-speed", nil).Update(int64(downloadSpeed))
+
 		log.Info("fetch successful", "took", ended, "endpoint", httpEndpoint(hosts[randIndex]))
 		break
 	}

cmd/swarm-snapshot/create.go

@@ -74,7 +74,7 @@ func createSnapshot(filename string, nodes int, services []string) (err error) {
 				UnderlayAddr: addr.Under(),
 				HiveParams:   hp,
 			}
-			return network.NewBzz(config, kad, nil, nil, nil), nil, nil
+			return network.NewBzz(config, kad, nil, nil, nil, nil, nil), nil, nil
 		},
 	})
 	defer sim.Close()

cmd/swarm/config_test.go

@@ -34,10 +34,11 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethersphere/swarm"
 	"github.com/ethersphere/swarm/api"
+	"github.com/ethersphere/swarm/testutil"
 )
 
 func TestConfigDump(t *testing.T) {
-	swarm := runSwarm(t, "dumpconfig")
+	swarm := runSwarm(t, "--verbosity", fmt.Sprintf("%d", *testutil.Loglevel), "dumpconfig")
 	defaultConf := api.NewConfig()
 	out, err := tomlSettings.Marshal(&defaultConf)
 	if err != nil {
@@ -53,6 +54,7 @@ func TestConfigFailsSwapEnabledNoBackendURL(t *testing.T) {
 		fmt.Sprintf("--%s", SwarmPortFlag.Name), "54545",
 		fmt.Sprintf("--%s", utils.ListenPortFlag.Name), "0",
 		fmt.Sprintf("--%s", SwarmSwapEnabledFlag.Name),
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 
 	swarm := runSwarm(t, flags...)
@@ -89,6 +91,7 @@ func TestBzzKeyFlag(t *testing.T) {
 		fmt.Sprintf("--%s", utils.DataDirFlag.Name), dir,
 		fmt.Sprintf("--%s", utils.IPCPathFlag.Name), conf.IPCPath,
 		fmt.Sprintf("--%s", SwarmBzzKeyHexFlag.Name), hexKey,
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 
 	node.Cmd = runSwarm(t, flags...)
@@ -137,6 +140,7 @@ func TestEmptyBzzAccountFlagMultipleAccounts(t *testing.T) {
 		fmt.Sprintf("--%s", SwarmPortFlag.Name), "0",
 		fmt.Sprintf("--%s", utils.ListenPortFlag.Name), "0",
 		fmt.Sprintf("--%s", utils.DataDirFlag.Name), dir,
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 
 	node.Cmd = runSwarm(t, flags...)
@@ -160,6 +164,7 @@ func TestEmptyBzzAccountFlagSingleAccount(t *testing.T) {
 		fmt.Sprintf("--%s", utils.ListenPortFlag.Name), "0",
 		fmt.Sprintf("--%s", utils.DataDirFlag.Name), dir,
 		fmt.Sprintf("--%s", utils.IPCPathFlag.Name), conf.IPCPath,
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 
 	node.Cmd = runSwarm(t, flags...)
@@ -205,6 +210,7 @@ func TestEmptyBzzAccountFlagNoAccountWrongPassword(t *testing.T) {
 		fmt.Sprintf("--%s", SwarmPortFlag.Name), "0",
 		fmt.Sprintf("--%s", utils.ListenPortFlag.Name), "0",
 		fmt.Sprintf("--%s", utils.DataDirFlag.Name), dir,
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 
 	node.Cmd = runSwarm(t, flags...)
@@ -244,6 +250,7 @@ func TestConfigCmdLineOverrides(t *testing.T) {
 		fmt.Sprintf("--%s", EnsAPIFlag.Name), "",
 		fmt.Sprintf("--%s", utils.DataDirFlag.Name), dir,
 		fmt.Sprintf("--%s", utils.IPCPathFlag.Name), conf.IPCPath,
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 	node.Cmd = runSwarm(t, flags...)
 	node.Cmd.InputLine(testPassphrase)
@@ -342,6 +349,7 @@ func TestConfigFileOverrides(t *testing.T) {
 		fmt.Sprintf("--%s", EnsAPIFlag.Name), "",
 		fmt.Sprintf("--%s", utils.DataDirFlag.Name), dir,
 		fmt.Sprintf("--%s", utils.IPCPathFlag.Name), conf.IPCPath,
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 	node.Cmd = runSwarm(t, flags...)
 	node.Cmd.InputLine(testPassphrase)
@@ -420,6 +428,7 @@ func TestConfigEnvVars(t *testing.T) {
 		"--ens-api", "",
 		"--datadir", dir,
 		"--ipcpath", conf.IPCPath,
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 
 	//node.Cmd = runSwarm(t,flags...)
@@ -551,6 +560,7 @@ func TestConfigCmdLineOverridesFile(t *testing.T) {
 		fmt.Sprintf("--%s", EnsAPIFlag.Name), "",
 		fmt.Sprintf("--%s", utils.DataDirFlag.Name), dir,
 		fmt.Sprintf("--%s", utils.IPCPathFlag.Name), conf.IPCPath,
+		"--verbosity", fmt.Sprintf("%d", *testutil.Loglevel),
 	}
 	node.Cmd = runSwarm(t, flags...)
 	node.Cmd.InputLine(testPassphrase)

cmd/swarm/explore.go

@@ -48,7 +48,7 @@ func hashes(ctx *cli.Context) {
 	}
 	defer f.Close()
 
-	fileStore := storage.NewFileStore(&storage.FakeChunkStore{}, storage.NewFileStoreParams(), chunk.NewTags())
+	fileStore := storage.NewFileStore(&storage.FakeChunkStore{}, &storage.FakeChunkStore{}, storage.NewFileStoreParams(), chunk.NewTags())
 	refs, err := fileStore.GetAllReferences(context.TODO(), f)
 	if err != nil {
 		utils.Fatalf("%v\n", err)

cmd/swarm/hash.go

@@ -78,7 +78,7 @@ func hash(ctx *cli.Context) {
 	defer f.Close()
 
 	stat, _ := f.Stat()
-	fileStore := storage.NewFileStore(&storage.FakeChunkStore{}, storage.NewFileStoreParams(), chunk.NewTags())
+	fileStore := storage.NewFileStore(&storage.FakeChunkStore{}, &storage.FakeChunkStore{}, storage.NewFileStoreParams(), chunk.NewTags())
 
 	addr, _, err := fileStore.Store(context.TODO(), f, stat.Size(), false)
 	if err != nil {

network/networkid_test.go

@@ -216,7 +216,7 @@ func newServices() adapters.Services {
 				HiveParams:   hp,
 				NetworkID:    uint64(currentNetworkID),
 			}
-			return NewBzz(config, kademlia(ctx.Config.ID), nil, nil, nil), nil
+			return NewBzz(config, kademlia(ctx.Config.ID), nil, nil, nil, nil, nil), nil
 		},
 	}
 }