cmd/swarm, swarm/network: dynamic sync streams update

- NewRegistry creates a goroutine that updates sync streams based on changes in kademlia. It calls updateSyncing method to create new and quit unintended sync streams. To early subscriptions are prevented by waiting until no new peers are added for a configured period of time SyncUpdateDelay.

- Implements two new methods on Kademlia:
  - NeighbourhoodDepthC which returns a channel that returns kademlia neighbourhood depth on each change
  - AddressBookSizeC which returns address book size when it changes
  This methods are required for making a decision when to request sync stream subscriptions update in a dynamic network.

- To avoid the parsing of stream strings to distinguish the SYNC ones, Registry now uses Stream structure as the key for clients, servers and clientParams maps. This required that the Stream must be comparable and Key field is changed from []byte to string. The consequence is that key is changed to string in all function signatures that accepted it. Also, new functions FormatSyncBinKey and ParseSyncBinKey are introduced to explicitly handle the parsing and formating of bin numbers that are used as Key values in Stream.

- Implements RegistryOptions that holds optional values for stream Registry initialization. The number of arguments for NewRegistry is a bit high with adding another option SyncUpdateDelay, and optional and required ones are separated with RegistryOptions.

- Range now has constructor NewRange for better initialization and also has the String method for better logging.

- Registry now has the Quit method that sends QuitMsg to the peer to terminate the stream client on its side. This is required to terminate not needed sync streams when kademlia table changes by adding new peers.

- Changes the RequestSubscription to skip the request if the stream is already registered. This ensures that no request subscriptions are made if the server already exists.

- New SyncUpdateDelay cli flag is added to control the delay of syncing update. Default is set to 15s.

.travis.yml

@@ -3,17 +3,6 @@ go_import_path: github.com/ethereum/go-ethereum
 sudo: false
 matrix:
   include:
-    - os: linux
-      dist: trusty
-      sudo: required
-      go: 1.8.x
-      script:
-        - sudo modprobe fuse
-        - sudo chmod 666 /dev/fuse
-        - sudo chown root:$USER /etc/fuse.conf
-        - go run build/ci.go install
-        - go run build/ci.go test -coverage
-
     - os: linux
       dist: trusty
       sudo: required

cmd/swarm/config.go

@@ -24,6 +24,7 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
+	"time"
 	"unicode"
 
 	cli "gopkg.in/urfave/cli.v1"
@@ -66,6 +67,7 @@ const (
 	SWARM_ENV_SWAP_ENABLE          = "SWARM_SWAP_ENABLE"
 	SWARM_ENV_SWAP_API             = "SWARM_SWAP_API"
 	SWARM_ENV_SYNC_ENABLE          = "SWARM_SYNC_ENABLE"
+	SWARM_ENV_SYNC_UPDATE_DELAY    = "SWARM_ENV_SYNC_UPDATE_DELAY"
 	SWARM_ENV_ENS_API              = "SWARM_ENS_API"
 	SWARM_ENV_ENS_ADDR             = "SWARM_ENS_ADDR"
 	SWARM_ENV_CORS                 = "SWARM_CORS"
@@ -200,6 +202,10 @@ func cmdLineOverride(currentConfig *bzzapi.Config, ctx *cli.Context) *bzzapi.Con
 		currentConfig.SyncEnabled = true
 	}
 
+	if d := ctx.GlobalDuration(SwarmSyncUpdateDelay.Name); d > 0 {
+		currentConfig.SyncUpdateDelay = d
+	}
+
 	currentConfig.SwapApi = ctx.GlobalString(SwarmSwapAPIFlag.Name)
 	if currentConfig.SwapEnabled && currentConfig.SwapApi == "" {
 		utils.Fatalf(SWARM_ERR_SWAP_SET_NO_API)
@@ -293,6 +299,12 @@ func envVarsOverride(currentConfig *bzzapi.Config) (config *bzzapi.Config) {
 		}
 	}
 
+	if v := os.Getenv(SWARM_ENV_SYNC_UPDATE_DELAY); v != "" {
+		if d, err := time.ParseDuration(v); err != nil {
+			currentConfig.SyncUpdateDelay = d
+		}
+	}
+
 	if swapapi := os.Getenv(SWARM_ENV_SWAP_API); swapapi != "" {
 		currentConfig.SwapApi = swapapi
 	}

cmd/swarm/main.go

@@ -106,6 +106,11 @@ var (
 		Usage:  "Swarm Syncing enabled (default true)",
 		EnvVar: SWARM_ENV_SYNC_ENABLE,
 	}
+	SwarmSyncUpdateDelay = cli.DurationFlag{
+		Name:   "sync-update-delay",
+		Usage:  "Duration for sync subscriptions update after no new peers are added (default 15s)",
+		EnvVar: SWARM_ENV_SYNC_UPDATE_DELAY,
+	}
 	EnsAPIFlag = cli.StringSliceFlag{
 		Name:   "ens-api",
 		Usage:  "ENS API endpoint for a TLD and with contract address, can be repeated, format [tld:][contract-addr@]url",
@@ -356,6 +361,7 @@ Remove corrupt entries from a local chunk database.
 		SwarmSwapEnabledFlag,
 		SwarmSwapAPIFlag,
 		SwarmSyncEnabledFlag,
+		SwarmSyncUpdateDelay,
 		SwarmListenAddrFlag,
 		SwarmPortFlag,
 		SwarmAccountFlag,

swarm/api/config.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/contracts/ens"
@@ -57,6 +58,7 @@ type Config struct {
 	NetworkId       uint64
 	SwapEnabled     bool
 	SyncEnabled     bool
+	SyncUpdateDelay time.Duration
 	PssEnabled      bool
 	ResourceEnabled bool
 	SwapApi         string
@@ -83,6 +85,7 @@ func NewConfig() (self *Config) {
 		NetworkId:       network.NetworkID,
 		SwapEnabled:     false,
 		SyncEnabled:     true,
+		SyncUpdateDelay: 15 * time.Second,
 		PssEnabled:      true,
 		ResourceEnabled: true,
 		SwapApi:         "",

swarm/network/kademlia.go

@@ -88,6 +88,9 @@ type Kademlia struct {
 	addrs      *pot.Pot // pots container for known peer addresses
 	conns      *pot.Pot // pots container for live peer connections
 	depth      uint8    // stores the last current depth of saturation
+	nDepth     int      // stores the last neighbourhood depth
+	nDepthC    chan int // returned by DepthC function to signal neighbourhood depth change
+	addrCountC chan int // returned by AddrCountC function to signal peer count change
 }
 
 // NewKademlia creates a Kademlia table for base address addr
@@ -198,6 +201,10 @@ func (k *Kademlia) Register(peers []OverlayAddr) error {
 		}
 		size++
 	}
+	// send new address count value only if there are new addresses
+	if k.addrCountC != nil && size-known > 0 {
+		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()))
 	return nil
 }
@@ -296,6 +303,10 @@ func (k *Kademlia) On(p OverlayConn) (uint8, bool) {
 		k.addrs, _, _, _ = pot.Swap(k.addrs, p, pof, func(v pot.Val) pot.Val {
 			return e
 		})
+		// send new address count value only if the peer is inserted
+		if k.addrCountC != nil {
+			k.addrCountC <- k.addrs.Size()
+		}
 	}
 	log.Trace(k.string())
 	// calculate if depth of saturation changed
@@ -305,9 +316,38 @@ func (k *Kademlia) On(p OverlayConn) (uint8, bool) {
 		changed = true
 		k.depth = depth
 	}
+	if k.nDepthC != nil {
+		nDepth := k.neighbourhoodDepth()
+		if nDepth != k.nDepth {
+			k.nDepth = nDepth
+			k.nDepthC <- nDepth
+		}
+	}
 	return k.depth, changed
 }
 
+// NeighbourhoodDepthC returns the channel that sends a new kademlia
+// neighbourhood depth on each change.
+// Not receiving from the returned channel will block On function
+// when the neighbourhood depth is changed.
+func (k *Kademlia) NeighbourhoodDepthC() <-chan int {
+	if k.nDepthC == nil {
+		k.nDepthC = make(chan int)
+	}
+	return k.nDepthC
+}
+
+// AddrCountC returns the channel that sends a new
+// address count value on each change.
+// Not receiving from the returned channel will block Register function
+// when address count value changes.
+func (k *Kademlia) AddrCountC() <-chan int {
+	if k.addrCountC == nil {
+		k.addrCountC = make(chan int)
+	}
+	return k.addrCountC
+}
+
 // Off removes a peer from among live peers
 func (k *Kademlia) Off(p OverlayConn) {
 	k.lock.Lock()
@@ -326,6 +366,10 @@ func (k *Kademlia) Off(p OverlayConn) {
 			// v cannot be nil, but no need to check
 			return nil
 		})
+		// send new address count value only if the peer is deleted
+		if k.addrCountC != nil {
+			k.addrCountC <- k.addrs.Size()
+		}
 	}
 }
 
@@ -333,23 +377,17 @@ func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(con
 	k.lock.RLock()
 	defer k.lock.RUnlock()
 
-	var i int
 	var startPo int
 	var endPo int
-	kadDepth := int(k.depth)
+	kadDepth := k.neighbourhoodDepth()
 
 	k.conns.EachBin(base, pof, o, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
+		if startPo > 0 && endPo != k.MaxProxDisplay {
+			startPo = endPo + 1
+		}
 		if po < kadDepth {
 			endPo = po
-			if i > 0 {
+		} else {
-				startPo = endPo + 1
-			}
-		} else if endPo < kadDepth || endPo == 0 {
-			if po == 0 && kadDepth == 0 {
-				startPo = endPo
-			} else {
-				startPo = endPo + 1
-			}
 			endPo = k.MaxProxDisplay
 		}
 
@@ -358,10 +396,8 @@ func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(con
 				return eachBinFunc(val.(*entry).conn(), bin)
 			})
 		}
-		i++
 		return true
 	})
-
 }
 
 // EachConn is an iterator with args (base, po, f) applies f to each live peer

swarm/network/light/lightnode.go

@@ -1,190 +0,0 @@
-// Copyright 2018 The go-ethereum Authors
-// This file is part of the go-ethereum library.d
-//
-// 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 light
-
-import (
-	"errors"
-
-	"github.com/ethereum/go-ethereum/swarm/network/stream"
-	"github.com/ethereum/go-ethereum/swarm/storage"
-)
-
-// RemoteReader implements IncomingStreamer
-type RemoteSectionReader struct {
-	db            *storage.DBAPI
-	start         uint64
-	end           uint64
-	hashes        chan []byte
-	currentHashes []byte
-	currentData   []byte
-	quit          chan struct{}
-	root          []byte
-}
-
-// NewRemoteReader is the constructor for RemoteReader
-func NewRemoteSectionReader(root []byte, db *storage.DBAPI) *RemoteSectionReader {
-	return &RemoteSectionReader{
-		db:     db,
-		root:   root,
-		hashes: make(chan []byte),
-		quit:   make(chan struct{}),
-	}
-}
-
-func (r *RemoteSectionReader) NeedData(key []byte) func() {
-	chunk, created := r.db.GetOrCreateRequest(storage.Key(key))
-	// TODO: we may want to request from this peer anyway even if the request exists
-	if chunk.ReqC == nil || !created {
-		return nil
-	}
-	return func() {
-		select {
-		case <-chunk.ReqC:
-		case <-r.quit:
-		}
-	}
-}
-
-func (r *RemoteSectionReader) BatchDone(s stream.Stream, from uint64, hashes []byte, root []byte) func() (*stream.TakeoverProof, error) {
-	r.hashes <- hashes
-	return nil
-}
-
-func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
-	l := int64(len(b))
-	m := int64(len(r.currentData))
-	if m > l {
-		m = l
-	}
-	copy(b, r.currentData[:m])
-	if m == l {
-		r.currentData = r.currentData[m:]
-		return l, nil
-	}
-	var end bool
-	for i := 0; !end && i < len(r.currentHashes); i += stream.HashSize {
-		hash := r.currentHashes[i : i+stream.HashSize]
-		chunk, err := r.db.Get(hash)
-		if err != nil {
-			return n, err
-		}
-		m := chunk.Size
-		if n+m > l {
-			m = l - n
-			end = true
-		}
-		copy(b[n:], chunk.SData[:m])
-		n += m
-	}
-
-	for {
-		select {
-		case <-r.quit:
-			return n, errors.New("aborted")
-		case hashes := <-r.hashes:
-			var i int
-			for ; !end && i < len(hashes); i += stream.HashSize {
-				hash := hashes[i : i+stream.HashSize]
-				chunk, err := r.db.Get(hash)
-				if err != nil {
-					return n, err
-				}
-				m := chunk.Size
-				if n+m > l {
-					m = l - n
-					end = true
-
-				}
-				copy(b[n:], chunk.SData[:m])
-				n += m
-			}
-			hashes = hashes[i:]
-		}
-	}
-}
-
-func (r *RemoteSectionReader) Close() {}
-
-// RemoteSectionServer implements OutgoingStreamer
-type RemoteSectionServer struct {
-	// quit chan struct{}
-	root []byte
-	db   *storage.DBAPI
-	r    *storage.LazyChunkReader
-}
-
-// NewRemoteReader is the constructor for RemoteReader
-func NewRemoteSectionServer(db *storage.DBAPI, r *storage.LazyChunkReader) *RemoteSectionServer {
-	return &RemoteSectionServer{
-		db: db,
-		r:  r,
-	}
-}
-
-// GetData retrieves the actual chunk from localstore
-func (s *RemoteSectionServer) GetData(key []byte) ([]byte, error) {
-	chunk, err := s.db.Get(storage.Key(key))
-	if err != nil {
-		return nil, err
-	}
-	return chunk.SData, nil
-}
-
-// GetBatch retrieves the next batch of hashes from the dbstore
-func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *stream.HandoverProof, error) {
-	if to > from+stream.BatchSize {
-		to = from + stream.BatchSize
-	}
-	batch := make([]byte, (to-from)*stream.HashSize)
-	s.r.ReadAt(batch, int64(from))
-	return batch, from, to, nil, nil
-}
-
-func (s *RemoteSectionServer) Close() {}
-
-// RegisterRemoteSectionReader registers RemoteSectionReader on light downstream node
-func RegisterRemoteSectionReader(s *stream.Registry, db *storage.DBAPI) {
-	s.RegisterClientFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte, live bool) (stream.Client, error) {
-		return NewRemoteSectionReader(t, db), nil
-	})
-}
-
-// RegisterRemoteSectionServer registers RemoteSectionServer outgoing streamer on
-// upstream light server node
-func RegisterRemoteSectionServer(s *stream.Registry, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) {
-	s.RegisterServerFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte, live bool) (stream.Server, error) {
-		r := rf(t)
-		return NewRemoteSectionServer(db, r), nil
-	})
-}
-
-// RegisterRemoteDownloader registers RemoteDownloader incoming streamer
-// on downstream light  node
-// func RegisterRemoteDownloader(s *Streamer, db *storage.DBAPI) {
-// 	s.RegisterIncomingStreamer("REMOTE_DOWNLOADER", func(p *stream.Peer, t []byte) (IncomingStreamer, error) {
-// 		return NewRemoteDownloader(t, db), nil
-// 	})
-// }
-//
-// // RegisterRemoteDownloadServer registers RemoteDownloadServer outgoing streamer on
-// // upstream light server node
-// func RegisterRemoteDownloadServer(s *Streamer, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) {
-// 	s.RegisterOutgoingStreamer("REMOTE_DOWNLOADER", func(p *stream.Peer, t []byte) (OutgoingStreamer, error) {
-// 		r := rf(t)
-// 		return NewRemoteDownloadServer(db, r), nil
-// 	})
-// }

swarm/network/stream/common_test.go

@@ -78,7 +78,11 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	db := storage.NewDBAPI(store)
 	delivery := NewDelivery(kad, db)
 	deliveries[id] = delivery
-	r := NewRegistry(addr, delivery, db, state.NewMemStore(), defaultSkipCheck, false, false)
+	r := NewRegistry(addr, delivery, db, state.NewMemStore(), &RegistryOptions{
+		SkipCheck: defaultSkipCheck,
+	})
+	RegisterSwarmSyncerServer(r, db)
+	RegisterSwarmSyncerClient(r, db)
 	go func() {
 		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
 	}()
@@ -107,7 +111,9 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 
 	db := storage.NewDBAPI(localStore)
 	delivery := NewDelivery(to, db)
-	streamer := NewRegistry(addr, delivery, db, state.NewMemStore(), defaultSkipCheck, false, false)
+	streamer := NewRegistry(addr, delivery, db, state.NewMemStore(), &RegistryOptions{
+		SkipCheck: defaultSkipCheck,
+	})
 	teardown := func() {
 		streamer.Close()
 		removeDataDir()
@@ -289,19 +295,13 @@ func (r *TestExternalRegistry) EnableNotifications(peerId discover.NodeID, s Str
 // with testClient and testServer.
 
 type testExternalClient struct {
-	t []byte
-	// wait0     chan bool
-	// batchDone chan bool
 	hashes               chan []byte
 	db                   *storage.DBAPI
 	enableNotificationsC chan struct{}
 }
 
-func newTestExternalClient(t []byte, db *storage.DBAPI) *testExternalClient {
+func newTestExternalClient(db *storage.DBAPI) *testExternalClient {
 	return &testExternalClient{
-		t: t,
-		// wait0:     make(chan bool),
-		// batchDone: make(chan bool),
 		hashes:               make(chan []byte),
 		db:                   db,
 		enableNotificationsC: make(chan struct{}),
@@ -328,14 +328,14 @@ func (c *testExternalClient) Close() {}
 const testExternalServerBatchSize = 10
 
 type testExternalServer struct {
-	t         []byte
+	t         string
 	keyFunc   func(key []byte, index uint64)
 	sessionAt uint64
 	maxKeys   uint64
 	streamer  *TestExternalRegistry
 }
 
-func newTestExternalServer(t []byte, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer {
+func newTestExternalServer(t string, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer {
 	if keyFunc == nil {
 		keyFunc = binary.BigEndian.PutUint64
 	}

swarm/network/stream/delivery.go

@@ -128,7 +128,7 @@ type RetrieveRequestMsg struct {
 
 func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) error {
 	log.Debug("received request", "peer", sp.ID(), "hash", req.Key)
-	s, err := sp.getServer(NewStream(swarmChunkServerStreamName, nil, false))
+	s, err := sp.getServer(NewStream(swarmChunkServerStreamName, "", false))
 	if err != nil {
 		return err
 	}

swarm/network/stream/delivery_test.go

@@ -87,11 +87,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
 	peer := streamer.getPeer(peerID)
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream: NewStream(swarmChunkServerStreamName, nil, false),
+		Stream:   NewStream(swarmChunkServerStreamName, "", false),
-		History: &Range{
+		History:  NewRange(0, 0),
-			From: 0,
-			To:   0,
-		},
 		Priority: Top,
 	})
 
@@ -139,11 +136,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 	peer := streamer.getPeer(peerID)
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream: NewStream(swarmChunkServerStreamName, nil, false),
+		Stream:   NewStream(swarmChunkServerStreamName, "", false),
-		History: &Range{
+		History:  NewRange(0, 0),
-			From: 0,
-			To:   0,
-		},
 		Priority: Top,
 	})
 
@@ -175,7 +169,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 					From:   0,
 					// TODO: why is this 32???
 					To:     32,
-					Stream: NewStream(swarmChunkServerStreamName, nil, false),
+					Stream: NewStream(swarmChunkServerStreamName, "", false),
 				},
 				Peer: peerID,
 			},
@@ -228,7 +222,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) {
 		return &testClient{
 			t: t,
 		}, nil
@@ -236,8 +230,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 
 	peerID := tester.IDs[0]
 
-	stream := NewStream("foo", nil, true)
+	stream := NewStream("foo", "", true)
-	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
+	err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -261,11 +255,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream: stream,
+					Stream:   stream,
-					History: &Range{
+					History:  NewRange(5, 8),
-						From: 5,
-						To:   8,
-					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -392,7 +383,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
 				sid := sim.IDs[j+1]
-				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, "", false), NewRange(0, 0), Top)
 			})
 			if err != nil {
 				return err
@@ -566,7 +557,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
 				sid := sim.IDs[j+1] // the upstream peer's id
-				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, "", false), NewRange(0, 0), Top)
 			})
 			if err != nil {
 				break

swarm/network/stream/intervals_test.go

@@ -50,12 +50,14 @@ func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, er
 	db := storage.NewDBAPI(store)
 	delivery := NewDelivery(kad, db)
 	deliveries[id] = delivery
-	r := NewRegistry(addr, delivery, db, state.NewMemStore(), defaultSkipCheck, false, false)
+	r := NewRegistry(addr, delivery, db, state.NewMemStore(), &RegistryOptions{
-
+		SkipCheck: defaultSkipCheck,
-	r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Client, error) {
-		return newTestExternalClient(t, db), nil
 	})
-	r.RegisterServerFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Server, error) {
+
+	r.RegisterClientFunc(externalStreamName, func(p *Peer, t string, live bool) (Client, error) {
+		return newTestExternalClient(db), nil
+	})
+	r.RegisterServerFunc(externalStreamName, func(p *Peer, t string, live bool) (Server, error) {
 		return newTestExternalServer(t, externalStreamSessionAt, externalStreamMaxKeys, nil), nil
 	})
 
@@ -67,8 +69,8 @@ func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, er
 
 func TestIntervals(t *testing.T) {
 	testIntervals(t, true, nil)
-	testIntervals(t, false, &Range{From: 9, To: 26})
+	testIntervals(t, false, NewRange(9, 26))
-	testIntervals(t, true, &Range{From: 9, To: 26})
+	testIntervals(t, true, NewRange(9, 26))
 }
 
 func testIntervals(t *testing.T, live bool, history *Range) {
@@ -143,7 +145,7 @@ func testIntervals(t *testing.T, live bool, history *Range) {
 			ctx, cancel := context.WithTimeout(ctx, 100*time.Second)
 			defer cancel()
 
-			err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, nil, live), history, Top)
+			err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, "", live), history, Top)
 			if err != nil {
 				return err
 			}
@@ -164,7 +166,7 @@ func testIntervals(t *testing.T, live bool, history *Range) {
 
 				// live stream
 				liveHashesChan := make(chan []byte)
-				liveSubscription, err := client.Subscribe(ctx, "stream", liveHashesChan, "getHashes", sid, NewStream(externalStreamName, nil, true))
+				liveSubscription, err := client.Subscribe(ctx, "stream", liveHashesChan, "getHashes", sid, NewStream(externalStreamName, "", true))
 				if err != nil {
 					return
 				}
@@ -173,7 +175,7 @@ func testIntervals(t *testing.T, live bool, history *Range) {
 				i := externalStreamSessionAt
 
 				// we have subscribed, enable notifications
-				err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, nil, true))
+				err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, "", true))
 				if err != nil {
 					return
 				}
@@ -211,7 +213,7 @@ func testIntervals(t *testing.T, live bool, history *Range) {
 
 				// history stream
 				historyHashesChan := make(chan []byte)
-				historySubscription, err := client.Subscribe(ctx, "stream", historyHashesChan, "getHashes", sid, NewStream(externalStreamName, nil, false))
+				historySubscription, err := client.Subscribe(ctx, "stream", historyHashesChan, "getHashes", sid, NewStream(externalStreamName, "", false))
 				if err != nil {
 					return
 				}
@@ -227,7 +229,7 @@ func testIntervals(t *testing.T, live bool, history *Range) {
 				}
 
 				// we have subscribed, enable notifications
-				err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, nil, false))
+				err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, "", false))
 				if err != nil {
 					return
 				}

swarm/network/stream/messages.go

@@ -31,13 +31,13 @@ type Stream struct {
 	// Name is used for Client and Server functions identification.
 	Name string
 	// Key is the name of specific stream data.
-	Key []byte
+	Key string
 	// Live defines whether the stream delivers only new data
 	// for the specific stream.
 	Live bool
 }
 
-func NewStream(name string, key []byte, live bool) Stream {
+func NewStream(name string, key string, live bool) Stream {
 	return Stream{
 		Name: name,
 		Key:  key,
@@ -51,7 +51,7 @@ func (s Stream) String() string {
 	if s.Live {
 		t = "l"
 	}
-	return fmt.Sprintf("%s|%x|%s", s.Name, s.Key, t)
+	return fmt.Sprintf("%s|%s|%s", s.Name, s.Key, t)
 }
 
 // SubcribeMsg is the protocol msg for requesting a stream(section)
@@ -148,8 +148,15 @@ type UnsubscribeMsg struct {
 }
 
 func (p *Peer) handleUnsubscribeMsg(req *UnsubscribeMsg) error {
-	p.removeServer(req.Stream)
+	return p.removeServer(req.Stream)
-	return nil
+}
+
+type QuitMsg struct {
+	Stream Stream
+}
+
+func (p *Peer) handleQuitMsg(req *QuitMsg) error {
+	return p.removeClient(req.Stream)
 }
 
 // OfferedHashesMsg is the protocol msg for offering to hand over a

swarm/network/stream/peer.go

@@ -52,12 +52,12 @@ type Peer struct {
 	pq       *pq.PriorityQueue
 	serverMu sync.RWMutex
 	clientMu sync.RWMutex // protects both clients and clientParams
-	servers  map[string]*server
+	servers  map[Stream]*server
-	clients  map[string]*client
+	clients  map[Stream]*client
 	// clientParams map keeps required client arguments
 	// that are set on Registry.Subscribe and used
 	// on creating a new client in offered hashes handler.
-	clientParams map[string]*clientParams
+	clientParams map[Stream]*clientParams
 	quit         chan struct{}
 }
 
@@ -67,9 +67,9 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
 		Peer:         peer,
 		pq:           pq.New(int(PriorityQueue), PriorityQueueCap),
 		streamer:     streamer,
-		servers:      make(map[string]*server),
+		servers:      make(map[Stream]*server),
-		clients:      make(map[string]*client),
+		clients:      make(map[Stream]*client),
-		clientParams: make(map[string]*clientParams),
+		clientParams: make(map[Stream]*clientParams),
 		quit:         make(chan struct{}),
 	}
 	ctx, cancel := context.WithCancel(context.Background())
@@ -128,9 +128,9 @@ func (p *Peer) getServer(s Stream) (*server, error) {
 	p.serverMu.RLock()
 	defer p.serverMu.RUnlock()
 
-	server := p.servers[s.String()]
+	server := p.servers[s]
 	if server == nil {
-		return nil, fmt.Errorf("server '%v' not provided to peer %v", s, p.ID())
+		return nil, newNotFoundError("server", s)
 	}
 	return server, nil
 }
@@ -139,16 +139,15 @@ func (p *Peer) setServer(s Stream, o Server, priority uint8) (*server, error) {
 	p.serverMu.Lock()
 	defer p.serverMu.Unlock()
 
-	sk := s.String()
+	if p.servers[s] != nil {
-	if p.servers[sk] != nil {
+		return nil, fmt.Errorf("server %s already registered", s)
-		return nil, fmt.Errorf("server %v already registered", sk)
 	}
 	os := &server{
 		Server:   o,
 		stream:   s,
 		priority: priority,
 	}
-	p.servers[sk] = os
+	p.servers[s] = os
 	return os, nil
 }
 
@@ -156,28 +155,26 @@ func (p *Peer) removeServer(s Stream) error {
 	p.serverMu.Lock()
 	defer p.serverMu.Unlock()
 
-	sk := s.String()
+	server, ok := p.servers[s]
-	server, ok := p.servers[sk]
 	if !ok {
 		return newNotFoundError("server", s)
 	}
 	server.Close()
-	delete(p.servers, sk)
+	delete(p.servers, s)
 	return nil
 }
 
 func (p *Peer) getClient(ctx context.Context, s Stream) (c *client, err error) {
 	var params *clientParams
-	sk := s.String()
 	func() {
 		p.clientMu.RLock()
 		defer p.clientMu.RUnlock()
 
-		c = p.clients[sk]
+		c = p.clients[s]
 		if c != nil {
 			return
 		}
-		params = p.clientParams[sk]
+		params = p.clientParams[s]
 	}()
 	if c != nil {
 		return c, nil
@@ -193,7 +190,7 @@ func (p *Peer) getClient(ctx context.Context, s Stream) (c *client, err error) {
 	p.clientMu.RLock()
 	defer p.clientMu.RUnlock()
 
-	c = p.clients[sk]
+	c = p.clients[s]
 	if c != nil {
 		return c, nil
 	}
@@ -201,12 +198,10 @@ func (p *Peer) getClient(ctx context.Context, s Stream) (c *client, err error) {
 }
 
 func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created bool, err error) {
-	sk := s.String()
-
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
-	c = p.clients[sk]
+	c = p.clients[s]
 	if c != nil {
 		return c, false, nil
 	}
@@ -273,7 +268,7 @@ func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created boo
 		intervalsStore: p.streamer.intervalsStore,
 		intervalsKey:   intervalsKey,
 	}
-	p.clients[sk] = c
+	p.clients[s] = c
 	cp.clientCreated() // unblock all possible getClient calls that are waiting
 	next <- nil        // this is to allow wantedKeysMsg before first batch arrives
 	return c, true, nil
@@ -283,7 +278,7 @@ func (p *Peer) removeClient(s Stream) error {
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
-	client, ok := p.clients[s.String()]
+	client, ok := p.clients[s]
 	if !ok {
 		return newNotFoundError("client", s)
 	}
@@ -295,19 +290,18 @@ func (p *Peer) setClientParams(s Stream, params *clientParams) error {
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
-	sk := s.String()
+	if p.clients[s] != nil {
-	if p.clients[sk] != nil {
+		return fmt.Errorf("client %s already exists", s)
-		return fmt.Errorf("client %v already exists", sk)
 	}
-	if p.clientParams[sk] != nil {
+	if p.clientParams[s] != nil {
-		return fmt.Errorf("client params %v already set", sk)
+		return fmt.Errorf("client params %s already set", s)
 	}
-	p.clientParams[sk] = params
+	p.clientParams[s] = params
 	return nil
 }
 
 func (p *Peer) getClientParams(s Stream) (*clientParams, error) {
-	params := p.clientParams[s.String()]
+	params := p.clientParams[s]
 	if params == nil {
 		return nil, fmt.Errorf("client params '%v' not provided to peer %v", s, p.ID())
 	}
@@ -315,12 +309,11 @@ func (p *Peer) getClientParams(s Stream) (*clientParams, error) {
 }
 
 func (p *Peer) removeClientParams(s Stream) error {
-	sk := s.String()
+	_, ok := p.clientParams[s]
-	_, ok := p.clientParams[sk]
 	if !ok {
 		return newNotFoundError("client params", s)
 	}
-	delete(p.clientParams, sk)
+	delete(p.clientParams, s)
 	return nil
 }
 

swarm/network/stream/stream.go

@@ -53,43 +53,126 @@ type Registry struct {
 	clientMu       sync.RWMutex
 	serverMu       sync.RWMutex
 	peersMu        sync.RWMutex
-	serverFuncs    map[string]func(*Peer, []byte, bool) (Server, error)
+	serverFuncs    map[string]func(*Peer, string, bool) (Server, error)
-	clientFuncs    map[string]func(*Peer, []byte, bool) (Client, error)
+	clientFuncs    map[string]func(*Peer, string, bool) (Client, error)
 	peers          map[discover.NodeID]*Peer
 	delivery       *Delivery
 	intervalsStore state.Store
 	doRetrieve     bool
 }
 
+// RegistryOptions holds optional values for NewRegistry constructor.
+type RegistryOptions struct {
+	SkipCheck       bool
+	DoSync          bool
+	DoRetrieve      bool
+	SyncUpdateDelay time.Duration
+}
+
 // NewRegistry is Streamer constructor
-func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, intervalsStore state.Store, skipCheck, doSync, doRetrieve bool) *Registry {
+func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, intervalsStore state.Store, options *RegistryOptions) *Registry {
+	if options == nil {
+		options = &RegistryOptions{}
+	}
+	if options.SyncUpdateDelay <= 0 {
+		options.SyncUpdateDelay = 15 * time.Second
+	}
 	streamer := &Registry{
 		addr:           addr,
-		skipCheck:      skipCheck,
+		skipCheck:      options.SkipCheck,
-		serverFuncs:    make(map[string]func(*Peer, []byte, bool) (Server, error)),
+		serverFuncs:    make(map[string]func(*Peer, string, bool) (Server, error)),
-		clientFuncs:    make(map[string]func(*Peer, []byte, bool) (Client, error)),
+		clientFuncs:    make(map[string]func(*Peer, string, bool) (Client, error)),
 		peers:          make(map[discover.NodeID]*Peer),
 		delivery:       delivery,
 		intervalsStore: intervalsStore,
-		doRetrieve:     doRetrieve,
+		doRetrieve:     options.DoRetrieve,
 	}
 	streamer.api = NewAPI(streamer)
 	delivery.getPeer = streamer.getPeer
-	streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ []byte, _ bool) (Server, error) {
+	streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) {
 		return NewSwarmChunkServer(delivery.db), nil
 	})
-	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ []byte, _ bool) (Client, error) {
+	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ string, _ bool) (Client, error) {
 		return NewSwarmSyncerClient(p, delivery.db, nil)
 	})
 	RegisterSwarmSyncerServer(streamer, db)
 	RegisterSwarmSyncerClient(streamer, db)
 
-	if doSync {
+	if options.DoSync {
-		go func() {
+		// latestIntC function ensures that
-			// this is a temporary workaround to wait for kademlia table to be healthy
+		//   - receiving from the in chan is not blocked by processing inside the for loop
-			time.Sleep(30 * time.Second)
+		// 	 - the latest int value is delivered to the loop after the processing is done
+		// In context of NeighbourhoodDepthC:
+		// after the syncing is done updating inside the loop, we do not need to update on the intermediate
+		// depth changes, only to the latest one
+		latestIntC := func(in <-chan int) <-chan int {
+			out := make(chan int, 1)
 
-			streamer.startSyncing()
+			go func() {
+				defer close(out)
+
+				for i := range in {
+					select {
+					case <-out:
+					default:
+					}
+					out <- i
+				}
+			}()
+
+			return out
+		}
+
+		go func() {
+			// 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())
+
+			for depth := range depthC {
+				log.Debug("Kademlia neighbourhood depth change", "depth", depth)
+
+				// Prevent too early sync subscriptions by waiting until there are no
+				// new peers connecting. Sync streams updating will be done after no
+				// peers are connected for at least SyncUpdateDelay period.
+				timer := time.NewTimer(options.SyncUpdateDelay)
+				// Hard limit to sync update delay, preventing long delays
+				// on a very dynamic network
+				maxTimer := time.NewTimer(3 * time.Minute)
+			loop:
+				for {
+					select {
+					case <-maxTimer.C:
+						// force syncing update when a hard timeout is reached
+						log.Trace("Sync subscriptions update on hard timeout")
+						// request for syncing subscription to new peers
+						streamer.updateSyncing()
+						break loop
+					case <-timer.C:
+						// start syncing as no new peers has been added to kademlia
+						// for some time
+						log.Trace("Sync subscriptions update")
+						// request for syncing subscription to new peers
+						streamer.updateSyncing()
+						break loop
+					case size := <-addressBookSizeC:
+						log.Trace("Kademlia address book size changed on depth change", "size", size)
+						// new peers has been added to kademlia,
+						// reset the timer to prevent early sync subscriptions
+						if !timer.Stop() {
+							<-timer.C
+						}
+						timer.Reset(options.SyncUpdateDelay)
+					}
+				}
+				timer.Stop()
+				maxTimer.Stop()
+			}
 		}()
 	}
 
@@ -97,7 +180,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
 }
 
 // RegisterClient registers an incoming streamer constructor
-func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte, bool) (Client, error)) {
+func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, string, bool) (Client, error)) {
 	r.clientMu.Lock()
 	defer r.clientMu.Unlock()
 
@@ -105,7 +188,7 @@ func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte, bool)
 }
 
 // RegisterServer registers an outgoing streamer constructor
-func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte, bool) (Server, error)) {
+func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, string, bool) (Server, error)) {
 	r.serverMu.Lock()
 	defer r.serverMu.Unlock()
 
@@ -113,7 +196,7 @@ func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte, bool)
 }
 
 // GetClient accessor for incoming streamer constructors
-func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte, bool) (Client, error), error) {
+func (r *Registry) GetClientFunc(stream string) (func(*Peer, string, bool) (Client, error), error) {
 	r.clientMu.RLock()
 	defer r.clientMu.RUnlock()
 
@@ -125,7 +208,7 @@ func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte, bool) (Clie
 }
 
 // GetServer accessor for incoming streamer constructors
-func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Server, error), error) {
+func (r *Registry) GetServerFunc(stream string) (func(*Peer, string, bool) (Server, error), error) {
 	r.serverMu.RLock()
 	defer r.serverMu.RUnlock()
 
@@ -138,7 +221,7 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Serv
 
 func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Range, prio uint8) error {
 	// check if the stream is registered
-	if _, err := r.GetClientFunc(s.Name); err != nil {
+	if _, err := r.GetServerFunc(s.Name); err != nil {
 		return err
 	}
 
@@ -147,13 +230,20 @@ func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Rang
 		return fmt.Errorf("peer not found %v", peerId)
 	}
 
-	msg := &RequestSubscriptionMsg{
+	if _, err := peer.getServer(s); err != nil {
-		Stream:   s,
+		if e, ok := err.(*notFoundError); ok && e.t == "server" {
-		History:  h,
+			// request subscription only if the server for this stream is not created
-		Priority: prio,
+			log.Debug("RequestSubscription ", "peer", peerId, "stream", s, "history", h)
+			return peer.Send(&RequestSubscriptionMsg{
+				Stream:   s,
+				History:  h,
+				Priority: prio,
+			})
+		}
+		return err
 	}
-	log.Debug("RequestSubscription ", "peer", peerId, "stream", s, "history", h)
+	log.Trace("RequestSubscription: already subscribed", "peer", peerId, "stream", s, "history", h)
-	return peer.Send(msg)
+	return nil
 }
 
 // Subscribe initiates the streamer
@@ -214,6 +304,24 @@ func (r *Registry) Unsubscribe(peerId discover.NodeID, s Stream) error {
 	return peer.removeClient(s)
 }
 
+// Quit sends the QuitMsg to the peer to remove the
+// stream peer client and terminate the streaming.
+func (r *Registry) Quit(peerId discover.NodeID, s Stream) error {
+	peer := r.getPeer(peerId)
+	if peer == nil {
+		log.Debug("stream quit: peer not found", "peer", peerId, "stream", s)
+		// if the peer is not found, abort the request
+		return nil
+	}
+
+	msg := &QuitMsg{
+		Stream: s,
+	}
+	log.Debug("Quit ", "peer", peerId, "stream", s)
+
+	return peer.Send(msg)
+}
+
 func (r *Registry) Retrieve(chunk *storage.Chunk) error {
 	return r.delivery.RequestFromPeers(chunk.Key[:], r.skipCheck)
 }
@@ -265,7 +373,7 @@ func (r *Registry) Run(p *network.BzzPeer) error {
 	defer sp.close()
 
 	if r.doRetrieve {
-		err := r.Subscribe(p.ID(), NewStream(swarmChunkServerStreamName, nil, false), nil, Top)
+		err := r.Subscribe(p.ID(), NewStream(swarmChunkServerStreamName, "", false), nil, Top)
 		if err != nil {
 			return err
 		}
@@ -274,22 +382,70 @@ func (r *Registry) Run(p *network.BzzPeer) error {
 	return sp.Run(sp.HandleMsg)
 }
 
-func (r *Registry) startSyncing() {
+// updateSyncing subscribes to SYNC streams by iterating over the
-	// panic freely
+// kademlia connections and bins. If there are existing SYNC streams
+// and they are no longer required after iteration, request to Quit
+// them will be send to appropriate peers.
+func (r *Registry) updateSyncing() {
+	// if overlay in not Kademlia, panic
 	kad := r.delivery.overlay.(*network.Kademlia)
 
+	// map of all SYNC streams for all peers
+	// used at the and of the function to remove servers
+	// that are not needed anymore
+	subs := make(map[discover.NodeID]map[Stream]struct{})
+	r.peersMu.RLock()
+	for id, peer := range r.peers {
+		peer.serverMu.RLock()
+		for stream := range peer.servers {
+			if stream.Name == "SYNC" {
+				if _, ok := subs[id]; !ok {
+					subs[id] = make(map[Stream]struct{})
+				}
+				subs[id][stream] = struct{}{}
+			}
+		}
+		peer.serverMu.RUnlock()
+	}
+	r.peersMu.RUnlock()
+
+	// request subscriptions for all nodes and bins
 	kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(conn network.OverlayConn, bin int) bool {
 		p := conn.(network.Peer)
 		log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin))
 
-		stream := NewStream("SYNC", []byte{uint8(bin)}, true)
+		// bin is always less then 256 and it is safe to convert it to type uint8
-		err := r.RequestSubscription(p.ID(), stream, &Range{}, Top)
+		stream := NewStream("SYNC", FormatSyncBinKey(uint8(bin)), true)
+		if streams, ok := subs[p.ID()]; ok {
+			// delete live and history streams from the map, so that it won't be removed with a Quit request
+			delete(streams, stream)
+			delete(streams, getHistoryStream(stream))
+		}
+		err := r.RequestSubscription(p.ID(), stream, NewRange(0, 0), Top)
 		if err != nil {
-			log.Error("request subscription", "err", err, "peer", p.ID(), "stream", stream)
+			log.Error("Request subscription", "err", err, "peer", p.ID(), "stream", stream)
 			return false
 		}
 		return true
 	})
+
+	// remove SYNC servers that do not need to be subscribed
+	for id, streams := range subs {
+		if len(streams) == 0 {
+			continue
+		}
+		peer := r.getPeer(id)
+		if peer == nil {
+			continue
+		}
+		for stream := range streams {
+			log.Debug("Remove sync server", "peer", id, "stream", stream)
+			err := r.Quit(peer.ID(), stream)
+			if err != nil {
+				log.Error("quit", "err", err, "peer", peer.ID(), "stream", stream)
+			}
+		}
+	}
 }
 
 func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
@@ -331,6 +487,9 @@ func (p *Peer) HandleMsg(msg interface{}) error {
 	case *RequestSubscriptionMsg:
 		return p.handleRequestSubscription(msg)
 
+	case *QuitMsg:
+		return p.handleQuitMsg(msg)
+
 	default:
 		return fmt.Errorf("unknown message type: %T", msg)
 	}
@@ -490,6 +649,7 @@ var Spec = &protocols.Spec{
 		ChunkDeliveryMsg{},
 		SubscribeErrorMsg{},
 		RequestSubscriptionMsg{},
+		QuitMsg{},
 	},
 }
 
@@ -530,6 +690,17 @@ type Range struct {
 	From, To uint64
 }
 
+func NewRange(from, to uint64) *Range {
+	return &Range{
+		From: from,
+		To:   to,
+	}
+}
+
+func (r *Range) String() string {
+	return fmt.Sprintf("%v-%v", r.From, r.To)
+}
+
 func getHistoryPriority(priority uint8) uint8 {
 	if priority == 0 {
 		return 0

swarm/network/stream/streamer_test.go

@@ -32,8 +32,8 @@ func TestStreamerSubscribe(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	stream := NewStream("foo", nil, true)
+	stream := NewStream("foo", "", true)
-	err = streamer.Subscribe(tester.IDs[0], stream, &Range{From: 0, To: 0}, Top)
+	err = streamer.Subscribe(tester.IDs[0], stream, NewRange(0, 0), Top)
 	if err == nil || err.Error() != "stream foo not registered" {
 		t.Fatalf("Expected error %v, got %v", "stream foo not registered", err)
 	}
@@ -48,14 +48,14 @@ var (
 )
 
 type testClient struct {
-	t              []byte
+	t              string
 	wait0          chan bool
 	wait2          chan bool
 	batchDone      chan bool
 	receivedHashes map[string][]byte
 }
 
-func newTestClient(t []byte) *testClient {
+func newTestClient(t string) *testClient {
 	return &testClient{
 		t:              t,
 		wait0:          make(chan bool),
@@ -87,10 +87,10 @@ func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*Takeo
 func (self *testClient) Close() {}
 
 type testServer struct {
-	t []byte
+	t string
 }
 
-func newTestServer(t []byte) *testServer {
+func newTestServer(t string) *testServer {
 	return &testServer{
 		t: t,
 	}
@@ -114,14 +114,14 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) {
 		return newTestClient(t), nil
 	})
 
 	peerID := tester.IDs[0]
 
-	stream := NewStream("foo", nil, true)
+	stream := NewStream("foo", "", true)
-	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
+	err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -133,11 +133,8 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 				{
 					Code: 4,
 					Msg: &SubscribeMsg{
-						Stream: stream,
+						Stream:   stream,
-						History: &Range{
+						History:  NewRange(5, 8),
-							From: 5,
-							To:   8,
-						},
 						Priority: Top,
 					},
 					Peer: peerID,
@@ -210,9 +207,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	stream := NewStream("foo", nil, false)
+	stream := NewStream("foo", "", false)
 
-	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+	streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) {
 		return newTestServer(t), nil
 	})
 
@@ -224,11 +221,8 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream: stream,
+					Stream:   stream,
-					History: &Range{
+					History:  NewRange(5, 8),
-						From: 5,
-						To:   8,
-					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -280,9 +274,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	stream := NewStream("foo", nil, true)
+	stream := NewStream("foo", "", true)
 
-	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+	streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) {
 		return newTestServer(t), nil
 	})
 
@@ -346,11 +340,11 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+	streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) {
 		return newTestServer(t), nil
 	})
 
-	stream := NewStream("bar", nil, true)
+	stream := NewStream("bar", "", true)
 
 	peerID := tester.IDs[0]
 
@@ -360,11 +354,8 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream: stream,
+					Stream:   stream,
-					History: &Range{
+					History:  NewRange(5, 8),
-						From: 5,
-						To:   8,
-					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -393,9 +384,9 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	stream := NewStream("foo", nil, true)
+	stream := NewStream("foo", "", true)
 
-	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+	streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) {
 		return &testServer{
 			t: t,
 		}, nil
@@ -409,11 +400,8 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream: stream,
+					Stream:   stream,
-					History: &Range{
+					History:  NewRange(5, 8),
-						From: 5,
-						To:   8,
-					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -423,7 +411,7 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
 			{
 				Code: 1,
 				Msg: &OfferedHashesMsg{
-					Stream: NewStream("foo", nil, false),
+					Stream: NewStream("foo", "", false),
 					HandoverProof: &HandoverProof{
 						Handover: &Handover{},
 					},
@@ -461,18 +449,18 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	stream := NewStream("foo", nil, true)
+	stream := NewStream("foo", "", true)
 
 	var tc *testClient
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) {
 		tc = newTestClient(t)
 		return tc, nil
 	})
 
 	peerID := tester.IDs[0]
 
-	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
+	err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -483,11 +471,8 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream: stream,
+					Stream:   stream,
-					History: &Range{
+					History:  NewRange(5, 8),
-						From: 5,
-						To:   8,
-					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -555,3 +540,131 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 	}
 
 }
+
+func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
+	tester, streamer, _, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) {
+		return newTestServer(t), nil
+	})
+
+	peerID := tester.IDs[0]
+
+	stream := NewStream("foo", "", true)
+	err = streamer.RequestSubscription(peerID, stream, NewRange(5, 8), Top)
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	err = tester.TestExchanges(
+		p2ptest.Exchange{
+			Label: "RequestSubscription message",
+			Expects: []p2ptest.Expect{
+				{
+					Code: 8,
+					Msg: &RequestSubscriptionMsg{
+						Stream:   stream,
+						History:  NewRange(5, 8),
+						Priority: Top,
+					},
+					Peer: peerID,
+				},
+			},
+		},
+		p2ptest.Exchange{
+			Label: "Subscribe message",
+			Triggers: []p2ptest.Trigger{
+				{
+					Code: 4,
+					Msg: &SubscribeMsg{
+						Stream:   stream,
+						History:  NewRange(5, 8),
+						Priority: Top,
+					},
+					Peer: peerID,
+				},
+			},
+			Expects: []p2ptest.Expect{
+				{
+					Code: 1,
+					Msg: &OfferedHashesMsg{
+						Stream: NewStream("foo", "", false),
+						HandoverProof: &HandoverProof{
+							Handover: &Handover{},
+						},
+						Hashes: make([]byte, HashSize),
+						From:   6,
+						To:     9,
+					},
+					Peer: peerID,
+				},
+				{
+					Code: 1,
+					Msg: &OfferedHashesMsg{
+						Stream: stream,
+						HandoverProof: &HandoverProof{
+							Handover: &Handover{},
+						},
+						From:   1,
+						To:     1,
+						Hashes: make([]byte, HashSize),
+					},
+					Peer: peerID,
+				},
+			},
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = streamer.Quit(peerID, stream)
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "Quit message",
+		Expects: []p2ptest.Expect{
+			{
+				Code: 9,
+				Msg: &QuitMsg{
+					Stream: stream,
+				},
+				Peer: peerID,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	historyStream := getHistoryStream(stream)
+
+	err = streamer.Quit(peerID, historyStream)
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "Quit message",
+		Expects: []p2ptest.Expect{
+			{
+				Code: 9,
+				Msg: &QuitMsg{
+					Stream: historyStream,
+				},
+				Peer: peerID,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+}

swarm/network/stream/syncer.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"strconv"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -64,8 +65,11 @@ func NewSwarmSyncerServer(live bool, po uint8, db *storage.DBAPI) (*SwarmSyncerS
 const maxPO = 32
 
 func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) {
-	streamer.RegisterServerFunc("SYNC", func(p *Peer, t []byte, live bool) (Server, error) {
+	streamer.RegisterServerFunc("SYNC", func(p *Peer, t string, live bool) (Server, error) {
-		po := t[0]
+		po, err := ParseSyncBinKey(t)
+		if err != nil {
+			return nil, err
+		}
 		return NewSwarmSyncerServer(live, po, db)
 	})
 	// streamer.RegisterServerFunc(stream, func(p *Peer) (Server, error) {
@@ -99,7 +103,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
 	if to <= from || from >= s.sessionAt {
 		to = math.MaxUint64
 	}
-	ticker := time.NewTicker(10 * time.Millisecond)
+	ticker := time.NewTicker(1 * time.Second)
 	defer ticker.Stop()
 	for {
 		select {
@@ -187,7 +191,7 @@ func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, chunker storage.Chunker) (
 // 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, t []byte, love bool) (Client, error) {
+	streamer.RegisterClientFunc("SYNC", func(p *Peer, _ string, love bool) (Client, error) {
 		return NewSwarmSyncerClient(p, db, nil)
 	})
 }
@@ -253,3 +257,23 @@ func (s *SwarmSyncerClient) TakeoverProof(stream Stream, from uint64, hashes []b
 }
 
 func (s *SwarmSyncerClient) Close() {}
+
+// base for parsing and formating sync bin key
+// it must be 2 <= base <= 36
+const syncBinKeyBase = 36
+
+// FormatSyncBinKey returns a string representation of
+// Kademlia bin number to be used as key for SYNC stream.
+func FormatSyncBinKey(bin uint8) string {
+	return strconv.FormatUint(uint64(bin), syncBinKeyBase)
+}
+
+// ParseSyncBinKey parses the string representation
+// and returns the Kademlia bin number.
+func ParseSyncBinKey(s string) (uint8, error) {
+	bin, err := strconv.ParseUint(s, syncBinKeyBase, 8)
+	if err != nil {
+		return 0, err
+	}
+	return uint8(bin), nil
+}

swarm/network/stream/syncer_test.go

@@ -161,7 +161,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 				defer cancel()
 				// start syncing, i.e., subscribe to upstream peers po 1 bin
 				sid := sim.IDs[j+1]
-				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", []byte{1}, false), &Range{From: 0, To: 0}, Top)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", FormatSyncBinKey(1), false), NewRange(0, 0), Top)
 			})
 			if err != nil {
 				return err

swarm/swarm.go

@@ -153,7 +153,11 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 	if err != nil {
 		return
 	}
-	self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, false, true, true)
+	self.streamer = stream.NewRegistry(addr, delivery, db, stateStore, &stream.RegistryOptions{
+		DoSync:          true,
+		DoRetrieve:      true,
+		SyncUpdateDelay: config.SyncUpdateDelay,
+	})
 
 	self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run)