swarm/network/stream: implement intervals

Implement intervals for stream Client.
Change the Subscribe message handling to create both live and history
streams if required.

swarm/network/stream/common_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
 	"github.com/ethereum/go-ethereum/swarm/network"
+	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -68,7 +69,7 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	delivery := NewDelivery(kad, db)
 	deliveries[id] = delivery
 	netStore := storage.NewNetStore(store, nil)
-	r := NewRegistry(addr, delivery, netStore, defaultSkipCheck)
+	r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck)
 	RegisterSwarmSyncerServer(r, db)
 	RegisterSwarmSyncerClient(r, db)
 	go func() {
@@ -98,7 +99,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 
 	db := storage.NewDBAPI(localStore)
 	delivery := NewDelivery(to, db)
-	streamer := NewRegistry(addr, delivery, localStore, defaultSkipCheck)
+	streamer := NewRegistry(addr, delivery, localStore, intervals.NewMemStore(), defaultSkipCheck)
 	protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol)
 
 	err = waitForPeers(streamer, 1*time.Second, 1)

swarm/network/stream/delivery.go

@@ -129,7 +129,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(swarmChunkServerStreamName)
+	s, err := sp.getServer(NewStream(swarmChunkServerStreamName, nil, false))
 	if err != nil {
 		return err
 	}

swarm/network/stream/delivery_test.go

@@ -87,10 +87,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
 	peer := streamer.getPeer(peerID)
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream:   swarmChunkServerStreamName,
+		Stream: NewStream(swarmChunkServerStreamName, nil, false),
-		Key:      nil,
+		History: &Range{
-		From:     0,
+			From: 0,
-		To:       0,
+			To:   0,
+		},
 		Priority: Top,
 	})
 
@@ -138,10 +139,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 	peer := streamer.getPeer(peerID)
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream:   swarmChunkServerStreamName,
+		Stream: NewStream(swarmChunkServerStreamName, nil, false),
-		Key:      nil,
+		History: &Range{
-		From:     0,
+			From: 0,
-		To:       0,
+			To:   0,
+		},
 		Priority: Top,
 	})
 
@@ -172,9 +174,9 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 					Hashes: hash,
 					From:   0,
 					// TODO: why is this 32???
-					To:     32,
+					To:      32,
-					Key:    []byte{},
+					Stream:  NewStream(swarmChunkServerStreamName, nil, false),
-					Stream: swarmChunkServerStreamName,
+					Initial: true,
 				},
 				Peer: peerID,
 			},
@@ -227,7 +229,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
 		return &testClient{
 			t: t,
 		}, nil
@@ -235,7 +237,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 
 	peerID := tester.IDs[0]
 
-	err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true)
+	stream := NewStream("foo", nil, true)
+	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -259,10 +262,11 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 			p2ptest.Expect{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "foo",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
-					From:     5,
+						From: 5,
-					To:       8,
+						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -388,7 +392,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, swarmChunkServerStreamName, nil, 0, 0, Top, false)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top)
 			})
 			if err != nil {
 				return err
@@ -561,7 +565,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, swarmChunkServerStreamName, nil, 0, 0, Top, false)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top)
 			})
 			if err != nil {
 				break

swarm/network/stream/intervals/intervals.go

@@ -0,0 +1,154 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package intervals
+
+import (
+	"fmt"
+	"sync"
+)
+
+// Intervals store a list of intervals. Its purpose is to provide
+// methods to add new intervals and retrieve missing intervals that
+// need to be added.
+// It may be used in synchronization of streaming data to persist
+// retrieved data ranges between sessions.
+type Intervals struct {
+	start  uint64
+	ranges [][2]uint64
+	mu     sync.RWMutex
+}
+
+// New creates a new instance of Intervals.
+// Start argument limits the lower bound of intervals.
+// No range bellow start bound will be added by Add method or
+// returned by Next method. This limit may be used for
+// tracking "live" synchronization, where the sync session
+// starts from a specific value, and if "live" sync intervals
+// need to be merged with historical ones, it can be safely done.
+func NewIntervals(start uint64) *Intervals {
+	return &Intervals{
+		start: start,
+	}
+}
+
+// Add adds a new range to intervals. Range start and end are values
+// are both inclusive.
+func (i *Intervals) Add(start, end uint64) {
+	i.mu.Lock()
+	defer i.mu.Unlock()
+
+	i.add(start, end)
+}
+
+func (i *Intervals) add(start, end uint64) {
+	if start < i.start {
+		start = i.start
+	}
+	if end < i.start {
+		return
+	}
+	minStartJ := -1
+	maxEndJ := -1
+	j := 0
+	for ; j < len(i.ranges); j++ {
+		if minStartJ < 0 {
+			if (start <= i.ranges[j][0] && end+1 >= i.ranges[j][0]) || (start <= i.ranges[j][1]+1 && end+1 >= i.ranges[j][1]) {
+				if i.ranges[j][0] < start {
+					start = i.ranges[j][0]
+				}
+				minStartJ = j
+			}
+		}
+		if (start <= i.ranges[j][1] && end+1 >= i.ranges[j][1]) || (start <= i.ranges[j][0] && end+1 >= i.ranges[j][0]) {
+			if i.ranges[j][1] > end {
+				end = i.ranges[j][1]
+			}
+			maxEndJ = j
+		}
+		if end+1 <= i.ranges[j][0] {
+			break
+		}
+	}
+	if minStartJ < 0 && maxEndJ < 0 {
+		i.ranges = append(i.ranges[:j], append([][2]uint64{{start, end}}, i.ranges[j:]...)...)
+		return
+	}
+	if minStartJ >= 0 {
+		i.ranges[minStartJ][0] = start
+	}
+	if maxEndJ >= 0 {
+		i.ranges[maxEndJ][1] = end
+	}
+	if minStartJ >= 0 && maxEndJ >= 0 && minStartJ != maxEndJ {
+		i.ranges[maxEndJ][0] = start
+		i.ranges = append(i.ranges[:minStartJ], i.ranges[maxEndJ:]...)
+	}
+}
+
+// Merge adds all the intervals from the the m Interval to current one.
+func (i *Intervals) Merge(m *Intervals) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	i.mu.Lock()
+	defer i.mu.Unlock()
+
+	for _, r := range m.ranges {
+		i.add(r[0], r[1])
+	}
+}
+
+// Next returns the first range interval that is not fulfilled. Returned
+// start and end values are both inclusive, meaning that the whole range
+// including start and end need to be added in order to full the gap
+// in intervals.
+// Returned value for end is 0 if the next interval is after the whole
+// range that is stored in Intervals. Zero end value represents no limit
+// on the next interval length.
+func (i *Intervals) Next() (start, end uint64) {
+	i.mu.RLock()
+	defer i.mu.RUnlock()
+
+	l := len(i.ranges)
+	if l == 0 {
+		return i.start, 0
+	}
+	if i.ranges[0][0] != i.start {
+		return i.start, i.ranges[0][0] - 1
+	}
+	if l == 1 {
+		return i.ranges[0][1] + 1, 0
+	}
+	return i.ranges[0][1] + 1, i.ranges[1][0] - 1
+}
+
+// Last returns the value that is at the end of the last interval.
+func (i *Intervals) Last() (end uint64) {
+	i.mu.RLock()
+	defer i.mu.RUnlock()
+
+	l := len(i.ranges)
+	if l == 0 {
+		return 0
+	}
+	return i.ranges[l-1][1]
+}
+
+// String returns a descriptive representation of range intervals
+// in [] notation, as a list of two element vectors.
+func (i *Intervals) String() string {
+	return fmt.Sprint(i.ranges)
+}

swarm/network/stream/intervals/intervals_test.go

@@ -0,0 +1,395 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package intervals
+
+import "testing"
+
+// Test tests Interval methods Add, Next and Last for various
+// initial state.
+func Test(t *testing.T) {
+	for i, tc := range []struct {
+		startLimit uint64
+		initial    [][2]uint64
+		start      uint64
+		end        uint64
+		expected   string
+		nextStart  uint64
+		nextEnd    uint64
+		last       uint64
+	}{
+		{
+			initial:   nil,
+			start:     0,
+			end:       0,
+			expected:  "[[0 0]]",
+			nextStart: 1,
+			nextEnd:   0,
+			last:      0,
+		},
+		{
+			initial:   nil,
+			start:     0,
+			end:       10,
+			expected:  "[[0 10]]",
+			nextStart: 11,
+			nextEnd:   0,
+			last:      10,
+		},
+		{
+			initial:   nil,
+			start:     5,
+			end:       15,
+			expected:  "[[5 15]]",
+			nextStart: 0,
+			nextEnd:   4,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{0, 0}},
+			start:     0,
+			end:       0,
+			expected:  "[[0 0]]",
+			nextStart: 1,
+			nextEnd:   0,
+			last:      0,
+		},
+		{
+			initial:   [][2]uint64{{0, 0}},
+			start:     5,
+			end:       15,
+			expected:  "[[0 0] [5 15]]",
+			nextStart: 1,
+			nextEnd:   4,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     5,
+			end:       15,
+			expected:  "[[5 15]]",
+			nextStart: 0,
+			nextEnd:   4,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     5,
+			end:       20,
+			expected:  "[[5 20]]",
+			nextStart: 0,
+			nextEnd:   4,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     10,
+			end:       20,
+			expected:  "[[5 20]]",
+			nextStart: 0,
+			nextEnd:   4,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     0,
+			end:       20,
+			expected:  "[[0 20]]",
+			nextStart: 21,
+			nextEnd:   0,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       10,
+			expected:  "[[2 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       4,
+			expected:  "[[2 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       5,
+			expected:  "[[2 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       3,
+			expected:  "[[2 3] [5 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       4,
+			expected:  "[[2 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{0, 1}, {5, 15}},
+			start:     2,
+			end:       4,
+			expected:  "[[0 15]]",
+			nextStart: 16,
+			nextEnd:   0,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     2,
+			end:       10,
+			expected:  "[[0 10] [15 20]]",
+			nextStart: 11,
+			nextEnd:   14,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     8,
+			end:       18,
+			expected:  "[[0 5] [8 20]]",
+			nextStart: 6,
+			nextEnd:   7,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     2,
+			end:       17,
+			expected:  "[[0 20]]",
+			nextStart: 21,
+			nextEnd:   0,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     2,
+			end:       25,
+			expected:  "[[0 25]]",
+			nextStart: 26,
+			nextEnd:   0,
+			last:      25,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     5,
+			end:       14,
+			expected:  "[[0 20]]",
+			nextStart: 21,
+			nextEnd:   0,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     6,
+			end:       14,
+			expected:  "[[0 20]]",
+			nextStart: 21,
+			nextEnd:   0,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}, {30, 40}},
+			start:     6,
+			end:       29,
+			expected:  "[[0 40]]",
+			nextStart: 41,
+			nextEnd:   0,
+			last:      40,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}},
+			start:     3,
+			end:       55,
+			expected:  "[[0 60]]",
+			nextStart: 61,
+			nextEnd:   0,
+			last:      60,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}},
+			start:     21,
+			end:       49,
+			expected:  "[[0 5] [15 60]]",
+			nextStart: 6,
+			nextEnd:   14,
+			last:      60,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}},
+			start:     0,
+			end:       100,
+			expected:  "[[0 100]]",
+			nextStart: 101,
+			nextEnd:   0,
+			last:      100,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      0,
+			end:        0,
+			expected:   "[]",
+			nextStart:  100,
+			nextEnd:    0,
+			last:       0,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      20,
+			end:        30,
+			expected:   "[]",
+			nextStart:  100,
+			nextEnd:    0,
+			last:       0,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      50,
+			end:        100,
+			expected:   "[[100 100]]",
+			nextStart:  101,
+			nextEnd:    0,
+			last:       100,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      50,
+			end:        110,
+			expected:   "[[100 110]]",
+			nextStart:  111,
+			nextEnd:    0,
+			last:       110,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      120,
+			end:        130,
+			expected:   "[[120 130]]",
+			nextStart:  100,
+			nextEnd:    119,
+			last:       130,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      120,
+			end:        130,
+			expected:   "[[120 130]]",
+			nextStart:  100,
+			nextEnd:    119,
+			last:       130,
+		},
+	} {
+		intervals := NewIntervals(tc.startLimit)
+		intervals.ranges = tc.initial
+		intervals.Add(tc.start, tc.end)
+		got := intervals.String()
+		if got != tc.expected {
+			t.Errorf("interval #%d: expected %s, got %s", i, tc.expected, got)
+		}
+		nextStart, nextEnd := intervals.Next()
+		if nextStart != tc.nextStart {
+			t.Errorf("interval #%d, expected next start %d, got %d", i, tc.nextStart, nextStart)
+		}
+		if nextEnd != tc.nextEnd {
+			t.Errorf("interval #%d, expected next end %d, got %d", i, tc.nextEnd, nextEnd)
+		}
+		last := intervals.Last()
+		if last != tc.last {
+			t.Errorf("interval #%d, expected last %d, got %d", i, tc.last, last)
+		}
+	}
+}
+
+func TestMerge(t *testing.T) {
+	for i, tc := range []struct {
+		initial  [][2]uint64
+		merge    [][2]uint64
+		expected string
+	}{
+		{
+			initial:  nil,
+			merge:    nil,
+			expected: "[]",
+		},
+		{
+			initial:  [][2]uint64{{10, 20}},
+			merge:    nil,
+			expected: "[[10 20]]",
+		},
+		{
+			initial:  nil,
+			merge:    [][2]uint64{{15, 25}},
+			expected: "[[15 25]]",
+		},
+		{
+			initial:  [][2]uint64{{0, 100}},
+			merge:    [][2]uint64{{150, 250}},
+			expected: "[[0 100] [150 250]]",
+		},
+		{
+			initial:  [][2]uint64{{0, 100}},
+			merge:    [][2]uint64{{101, 250}},
+			expected: "[[0 250]]",
+		},
+		{
+			initial:  [][2]uint64{{0, 10}, {30, 40}},
+			merge:    [][2]uint64{{20, 25}, {41, 50}},
+			expected: "[[0 10] [20 25] [30 50]]",
+		},
+		{
+			initial:  [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}},
+			merge:    [][2]uint64{{6, 25}},
+			expected: "[[0 25] [30 40] [50 60]]",
+		},
+	} {
+		intervals := NewIntervals(0)
+		intervals.ranges = tc.initial
+		m := NewIntervals(0)
+		m.ranges = tc.merge
+
+		intervals.Merge(m)
+
+		got := intervals.String()
+		if got != tc.expected {
+			t.Errorf("interval #%d: expected %s, got %s", i, tc.expected, got)
+		}
+	}
+}

swarm/network/stream/intervals/store.go

@@ -0,0 +1,84 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+// Package intervals TODO: implement LevelDB based Store.
+package intervals
+
+import (
+	"errors"
+	"sync"
+)
+
+// ErrNotFound is returned by the Store implementation when the Interval
+// for a specific key does not exist.
+var ErrNotFound = errors.New("not found")
+
+// Store defines methods required to get and retrieve Intervals for different keys.
+// It is meant to be used for intervals persistance for different streams in the
+// stream package.
+type Store interface {
+	Get(key string) (i *Intervals, err error)
+	Put(key string, i *Intervals) (err error)
+	Delete(key string) (err error)
+}
+
+// MemStore is the reference implementation of Store interface that is supposed
+// to be used in tests.
+type MemStore struct {
+	db map[string]*Intervals
+	mu sync.RWMutex
+}
+
+// NewMemStore returns a new instance of MemStore.
+func NewMemStore() *MemStore {
+	return &MemStore{
+		db: make(map[string]*Intervals),
+	}
+}
+
+// Get retrieves Intervals for a specific key. If there is no Intervals
+// ErrNotFound is returned.
+func (s *MemStore) Get(key string) (i *Intervals, err error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	i, ok := s.db[key]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	return i, nil
+}
+
+// Put stores Intervals for a specific key.
+func (s *MemStore) Put(key string, i *Intervals) (err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	s.db[key] = i
+	return nil
+}
+
+// Delete removes Intervals stored under a specific key.
+func (s *MemStore) Delete(key string) (err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if _, ok := s.db[key]; !ok {
+		return ErrNotFound
+	}
+	delete(s.db, key)
+	return nil
+}

swarm/network/stream/intervals/store_test.go

@@ -0,0 +1,69 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package intervals
+
+import "testing"
+
+// TestMemStore tests basic functionality of MemStore.
+func TestMemStore(t *testing.T) {
+	s := NewMemStore()
+
+	key1 := "key1"
+	i1 := NewIntervals(0)
+	i1.Add(10, 20)
+	if err := s.Put(key1, i1); err != nil {
+		t.Fatal(err)
+	}
+	g, err := s.Get(key1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if g.String() != i1.String() {
+		t.Errorf("expected interval %s, got %s", i1, g)
+	}
+
+	key2 := "key2"
+	i2 := NewIntervals(0)
+	i2.Add(10, 20)
+	if err := s.Put(key2, i2); err != nil {
+		t.Fatal(err)
+	}
+	g, err = s.Get(key2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if g.String() != i2.String() {
+		t.Errorf("expected interval %s, got %s", i2, g)
+	}
+
+	if err := s.Delete(key1); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Get(key1); err != ErrNotFound {
+		t.Errorf("expected error %v, got %s", ErrNotFound, err)
+	}
+	if _, err := s.Get(key2); err != nil {
+		t.Errorf("expected error %v, got %s", nil, err)
+	}
+
+	if err := s.Delete(key2); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Get(key2); err != ErrNotFound {
+		t.Errorf("expected error %v, got %s", ErrNotFound, err)
+	}
+}

swarm/network/stream/messages.go

@@ -23,14 +23,42 @@ import (
 
 	"github.com/ethereum/go-ethereum/log"
 	bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
+	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
+// Stream defines a unique stream identifier.
+type Stream struct {
+	// Name is used for Client and Server functions identification.
+	Name string
+	// Key is the name of specific stream data.
+	Key []byte
+	// Live defines whether the stream delivers only new data
+	// for the specific stream.
+	Live bool
+}
+
+func NewStream(name string, key []byte, live bool) Stream {
+	return Stream{
+		Name: name,
+		Key:  key,
+		Live: live,
+	}
+}
+
+// String return a stream id based on all Stream fields.
+func (s Stream) String() string {
+	t := "h"
+	if s.Live {
+		t = "l"
+	}
+	return fmt.Sprintf("%s|%x|%s", s.Name, s.Key, t)
+}
+
 // SubcribeMsg is the protocol msg for requesting a stream(section)
 type SubscribeMsg struct {
-	Stream   string
+	Stream   Stream
-	Key      []byte
+	History  *Range
-	From, To uint64
 	Priority uint8 // delivered on priority channel
 }
 
@@ -45,24 +73,58 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 		}
 	}()
 
-	f, err := p.streamer.GetServerFunc(req.Stream)
+	log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "history", req.History)
+
+	f, err := p.streamer.GetServerFunc(req.Stream.Name)
 	if err != nil {
 		return err
 	}
-	s, err := f(p, req.Key)
+
+	s, err := f(p, req.Stream.Key, req.Stream.Live)
 	if err != nil {
 		return err
 	}
-	os, err := p.setServer(req.Stream, req.Key, s, req.Priority)
+	os, err := p.setServer(req.Stream, s, req.Priority)
 	if err != nil {
 		return err
 	}
-	log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+
+	var from uint64
+	var to uint64
+	if !req.Stream.Live && req.History != nil {
+		from = req.History.From
+		to = req.History.To
+	}
+
 	go func() {
-		if err := p.SendOfferedHashes(os, req.From, req.To); err != nil {
+		if err := p.SendOfferedHashes(os, from, to, true); err != nil {
 			p.Drop(err)
 		}
 	}()
+
+	if req.Stream.Live && req.History != nil {
+		// subscribe to the history stream as well
+		s, err := f(p, req.Stream.Key, false)
+		if err != nil {
+			return err
+		}
+		historyStream := NewStream(req.Stream.Name, req.Stream.Key, false)
+		priority := req.Priority
+		if priority > 0 {
+			// decrement history stream priority
+			priority--
+		}
+		os, err := p.setServer(historyStream, s, priority)
+		if err != nil {
+			return err
+		}
+		go func() {
+			if err := p.SendOfferedHashes(os, req.History.From, req.History.To, true); err != nil {
+				p.Drop(err)
+			}
+		}()
+	}
+
 	return nil
 }
 
@@ -75,23 +137,22 @@ func (p *Peer) handleSubscribeErrorMsg(req *SubscribeErrorMsg) (err error) {
 }
 
 type UnsubscribeMsg struct {
-	Stream string
+	Stream Stream
-	Key    []byte
 }
 
 func (p *Peer) handleUnsubscribeMsg(req *UnsubscribeMsg) error {
-	p.removeServer(req.Stream, req.Key)
+	p.removeServer(req.Stream)
 	return nil
 }
 
 // OfferedHashesMsg is the protocol msg for offering to hand over a
 // stream section
 type OfferedHashesMsg struct {
-	Stream         string // name of Stream
+	Stream         Stream // name of Stream
-	Key            []byte // subtype or key
 	From, To       uint64 // peer and db-specific entry count
 	Hashes         []byte // stream of hashes (128)
-	*HandoverProof        // HandoverProof
+	Initial        bool
+	*HandoverProof // HandoverProof
 }
 
 // String pretty prints OfferedHashesMsg
@@ -102,9 +163,7 @@ func (m OfferedHashesMsg) String() string {
 // handleOfferedHashesMsg protocol msg handler calls the incoming streamer interface
 // Filter method
 func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
-	sk := req.Stream
+	c, err := p.getClient(req.Stream)
-	sk += keyToString(req.Key)
-	s, err := p.getClient(sk)
 	if err != nil {
 		return err
 	}
@@ -117,7 +176,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	for i := 0; i < len(hashes); i += HashSize {
 		hash := hashes[i : i+HashSize]
 
-		if wait := s.NeedData(hash); wait != nil {
+		if wait := c.NeedData(hash); wait != nil {
 			want.Set(i/HashSize, true)
 			wg.Add(1)
 			// create request and wait until the chunk data arrives and is stored
@@ -142,22 +201,27 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	// }()
 	go func() {
 		wg.Wait()
-		s.next <- s.batchDone(p, req, hashes)
+		c.next <- c.batchDone(p, req, hashes)
 	}()
 	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
 	// except
-	if s.live {
+	if c.stream.Live {
-		s.sessionAt = req.From
+		c.sessionAt = req.From
+		if req.Initial {
+			// create initial intervals for live stream starting from the first From value
+			if err := c.intervalsStore.Put(peerStreamIntervalsKey(p, req.Stream), intervals.NewIntervals(req.From)); err != nil {
+				return err
+			}
+		}
 	}
-	from, to := s.nextBatch(req.To)
+	from, to := c.nextBatch(req.To)
-	log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+	log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To)
 	if from == to {
 		return nil
 	}
 
 	msg := &WantedHashesMsg{
 		Stream: req.Stream,
-		Key:    req.Key,
 		Want:   want.Bytes(),
 		From:   from,
 		To:     to,
@@ -167,14 +231,14 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 		case <-time.After(30 * time.Second):
 			p.Drop(err)
 			return
-		case err := <-s.next:
+		case err := <-c.next:
 			if err != nil {
 				p.Drop(err)
 				return
 			}
 		}
-		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To)
+		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
-		err := p.SendPriority(msg, s.priority)
+		err := p.SendPriority(msg, c.priority)
 		if err != nil {
 			p.Drop(err)
 		}
@@ -185,8 +249,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 // WantedHashesMsg is the protocol msg data for signaling which hashes
 // offered in OfferedHashesMsg downstream peer actually wants sent over
 type WantedHashesMsg struct {
-	Stream   string // name of stream
+	Stream   Stream
-	Key      []byte // subtype or key
 	Want     []byte // bitvector indicating which keys of the batch needed
 	From, To uint64 // next interval offset - empty if not to be continued
 }
@@ -200,15 +263,15 @@ func (m WantedHashesMsg) String() string {
 // * sends the next batch of unsynced keys
 // * sends the actual data chunks as per WantedHashesMsg
 func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
-	log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+	log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To)
-	s, err := p.getServer(req.Stream + keyToString(req.Key))
+	s, err := p.getServer(req.Stream)
 	if err != nil {
 		return err
 	}
 	hashes := s.currentBatch
 	// launch in go routine since GetBatch blocks until new hashes arrive
 	go func() {
-		if err := p.SendOfferedHashes(s, req.From, req.To); err != nil {
+		if err := p.SendOfferedHashes(s, req.From, req.To, false); err != nil {
 			p.Drop(err)
 		}
 	}()
@@ -237,7 +300,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
 
 // Handover represents a statement that the upstream peer hands over the stream section
 type Handover struct {
-	Stream     string // name of stream
+	Stream     Stream // name of stream
 	Start, End uint64 // index of hashes
 	Root       []byte // Root hash for indexed segment inclusion proofs
 }

swarm/network/stream/peer.go

@@ -26,6 +26,7 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue"
+	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -84,7 +85,7 @@ func (p *Peer) SendPriority(msg interface{}, priority uint8) error {
 }
 
 // SendOfferedHashes sends OfferedHashesMsg protocol msg
-func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
+func (p *Peer) SendOfferedHashes(s *server, f, t uint64, initial bool) error {
 	hashes, from, to, proof, err := s.SetNextBatch(f, t)
 	if err != nil {
 		return err
@@ -105,57 +106,56 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
 		From:          from,
 		To:            to,
 		Stream:        s.stream,
-		Key:           s.key,
+		Initial:       initial,
 	}
-	log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to)
+	log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to)
 	return p.SendPriority(msg, s.priority)
 }
 
-func (p *Peer) getServer(s string) (*server, error) {
+func (p *Peer) getServer(s Stream) (*server, error) {
 	p.serverMu.RLock()
 	defer p.serverMu.RUnlock()
 
-	server := p.servers[s]
+	server := p.servers[s.String()]
 	if server == nil {
 		return nil, fmt.Errorf("server '%v' not provided to peer %v", s, p.ID())
 	}
 	return server, nil
 }
 
-func (p *Peer) getClient(s string) (*client, error) {
+func (p *Peer) getClient(s Stream) (*client, error) {
 	p.clientMu.RLock()
 	defer p.clientMu.RUnlock()
 
-	client := p.clients[s]
+	client := p.clients[s.String()]
 	if client == nil {
 		return nil, fmt.Errorf("client '%v' not provided to peer %v", s, p.ID())
 	}
 	return client, nil
 }
 
-func (p *Peer) setServer(s string, key []byte, o Server, priority uint8) (*server, error) {
+func (p *Peer) setServer(s Stream, o Server, priority uint8) (*server, error) {
 	p.serverMu.Lock()
 	defer p.serverMu.Unlock()
 
-	sk := s + keyToString(key)
+	sk := s.String()
 	if p.servers[sk] != nil {
 		return nil, fmt.Errorf("server %v already registered", sk)
 	}
 	os := &server{
 		Server:   o,
-		priority: priority,
 		stream:   s,
-		key:      key,
+		priority: priority,
 	}
 	p.servers[sk] = os
 	return os, nil
 }
 
-func (p *Peer) removeServer(s string, key []byte) error {
+func (p *Peer) removeServer(s Stream) error {
 	p.serverMu.Lock()
 	defer p.serverMu.Unlock()
 
-	sk := s + keyToString(key)
+	sk := s.String()
 	server, ok := p.servers[sk]
 	if !ok {
 		return errServerNotFound
@@ -165,39 +165,63 @@ func (p *Peer) removeServer(s string, key []byte) error {
 	return nil
 }
 
-func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bool) error {
+func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore intervals.Store) error {
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
-	sk := s + keyToString(key)
+	sk := s.String()
 	if p.clients[sk] != nil {
 		return fmt.Errorf("client %v already registered", sk)
 	}
+
+	intervalsKey := peerStreamIntervalsKey(p, s)
+	if s.Live {
+		// try to find previous history and live intervals and merge live into history
+		historyKey := peerStreamIntervalsKey(p, NewStream(s.Name, s.Key, false))
+		historyIntervals, err := intervalsStore.Get(historyKey)
+		switch err {
+		case nil:
+			liveIntervals, err := intervalsStore.Get(intervalsKey)
+			switch err {
+			case nil:
+				historyIntervals.Merge(liveIntervals)
+				if err := intervalsStore.Put(historyKey, historyIntervals); err != nil {
+					log.Error("stream set client: put history intervals", "stream", s, "peer", p, "err", err)
+				}
+			case intervals.ErrNotFound:
+			default:
+				log.Error("stream set client: get live intervals", "stream", s, "peer", p, "err", err)
+			}
+		case intervals.ErrNotFound:
+		default:
+			log.Error("stream set client: get history intervals", "stream", s, "peer", p, "err", err)
+		}
+	} else {
+		// create intervals for history stream
+		// live stream can create intervals when the first sessionAt is known
+		if err := intervalsStore.Put(intervalsKey, intervals.NewIntervals(0)); err != nil {
+			return err
+		}
+	}
+
 	next := make(chan error, 1)
-	// var intervals *Intervals
-	// if !live {
-	// key := s + p.ID().String()
-	// intervals = NewIntervals(key, p.streamer)
-	// }
 	p.clients[sk] = &client{
-		Client: i,
+		Client:         i,
-		// intervals:        intervals,
+		stream:         s,
-		live:     live,
+		priority:       priority,
-		priority: priority,
+		next:           next,
-		next:     next,
+		intervalsStore: intervalsStore,
-		stream:   s,
+		intervalsKey:   intervalsKey,
-		key:      key,
 	}
 	next <- nil // this is to allow wantedKeysMsg before first batch arrives
 	return nil
 }
 
-func (p *Peer) removeClient(s string, key []byte) error {
+func (p *Peer) removeClient(s Stream) error {
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
-	sk := s + keyToString(key)
+	client, ok := p.clients[s.String()]
-	client, ok := p.clients[sk]
 	if !ok {
 		return errClientNotFound
 	}

swarm/network/stream/stream.go

@@ -30,6 +30,7 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	"github.com/ethereum/go-ethereum/swarm/network"
+	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -45,43 +46,45 @@ const (
 
 // Registry registry for outgoing and incoming streamer constructors
 type Registry struct {
-	api         *API
+	api            *API
-	addr        *network.BzzAddr
+	addr           *network.BzzAddr
-	skipCheck   bool
+	skipCheck      bool
-	clientMu    sync.RWMutex
+	clientMu       sync.RWMutex
-	serverMu    sync.RWMutex
+	serverMu       sync.RWMutex
-	peersMu     sync.RWMutex
+	peersMu        sync.RWMutex
-	serverFuncs map[string]func(*Peer, []byte) (Server, error)
+	serverFuncs    map[string]func(*Peer, []byte, bool) (Server, error)
-	clientFuncs map[string]func(*Peer, []byte) (Client, error)
+	clientFuncs    map[string]func(*Peer, []byte, bool) (Client, error)
-	peers       map[discover.NodeID]*Peer
+	peers          map[discover.NodeID]*Peer
-	delivery    *Delivery
+	delivery       *Delivery
-	store       storage.ChunkStore
+	store          storage.ChunkStore
+	intervalsStore intervals.Store
 }
 
 // NewRegistry is Streamer constructor
-func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, skipCheck bool) *Registry {
+func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, intervalsStore intervals.Store, skipCheck bool) *Registry {
 	streamer := &Registry{
-		addr:        addr,
+		addr:           addr,
-		skipCheck:   skipCheck,
+		skipCheck:      skipCheck,
-		store:       store,
+		store:          store,
-		serverFuncs: make(map[string]func(*Peer, []byte) (Server, error)),
+		serverFuncs:    make(map[string]func(*Peer, []byte, bool) (Server, error)),
-		clientFuncs: make(map[string]func(*Peer, []byte) (Client, error)),
+		clientFuncs:    make(map[string]func(*Peer, []byte, bool) (Client, error)),
-		peers:       make(map[discover.NodeID]*Peer),
+		peers:          make(map[discover.NodeID]*Peer),
-		delivery:    delivery,
+		delivery:       delivery,
+		intervalsStore: intervalsStore,
 	}
 	streamer.api = NewAPI(streamer, streamer.store)
 	delivery.getPeer = streamer.getPeer
-	streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, t []byte) (Server, error) {
+	streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ []byte, _ bool) (Server, error) {
 		return NewSwarmChunkServer(delivery.db), nil
 	})
-	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t []byte) (Client, error) {
+	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ []byte, _ bool) (Client, error) {
 		return NewSwarmSyncerClient(p, delivery.db, nil)
 	})
 	return streamer
 }
 
 // RegisterClient registers an incoming streamer constructor
-func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte) (Client, error)) {
+func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte, bool) (Client, error)) {
 	r.clientMu.Lock()
 	defer r.clientMu.Unlock()
 
@@ -89,7 +92,7 @@ func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte) (Clie
 }
 
 // RegisterServer registers an outgoing streamer constructor
-func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte) (Server, error)) {
+func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte, bool) (Server, error)) {
 	r.serverMu.Lock()
 	defer r.serverMu.Unlock()
 
@@ -97,7 +100,7 @@ func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte) (Serv
 }
 
 // GetClient accessor for incoming streamer constructors
-func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte) (Client, error), error) {
+func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte, bool) (Client, error), error) {
 	r.clientMu.RLock()
 	defer r.clientMu.RUnlock()
 
@@ -109,7 +112,7 @@ func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte) (Client, er
 }
 
 // GetServer accessor for incoming streamer constructors
-func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte) (Server, error), error) {
+func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Server, error), error) {
 	r.serverMu.RLock()
 	defer r.serverMu.RUnlock()
 
@@ -121,8 +124,8 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte) (Server, er
 }
 
 // Subscribe initiates the streamer
-func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
+func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error {
-	f, err := r.GetClientFunc(s)
+	f, err := r.GetClientFunc(s.Name)
 	if err != nil {
 		return err
 	}
@@ -132,29 +135,42 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, t
 		return fmt.Errorf("peer not found %v", peerId)
 	}
 
-	is, err := f(peer, t)
+	is, err := f(peer, s.Key, s.Live)
 	if err != nil {
 		return err
 	}
-	err = peer.setClient(s, t, is, priority, live)
+	err = peer.setClient(s, is, priority, r.intervalsStore)
 	if err != nil {
 		return err
 	}
 
+	if s.Live && h != nil {
+		is, err := f(peer, s.Key, false)
+		if err != nil {
+			return err
+		}
+		p := priority
+		if p > 0 {
+			p--
+		}
+		historyStream := NewStream(s.Name, s.Key, false)
+		err = peer.setClient(historyStream, is, p, r.intervalsStore)
+		if err != nil {
+			return err
+		}
+	}
+
 	msg := &SubscribeMsg{
-		Stream: s,
+		Stream:   s,
-		Key:    t,
+		History:  h,
-		// Live:     live,
-		From:     from,
-		To:       to,
 		Priority: priority,
 	}
-	log.Debug("Subscribe ", "peer", peerId, "stream", s, "key", t, "from", from, "to", to)
+	log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h)
 
 	return peer.SendPriority(msg, priority)
 }
 
-func (r *Registry) Unsubscribe(peerId discover.NodeID, s string, t []byte) error {
+func (r *Registry) Unsubscribe(peerId discover.NodeID, s Stream) error {
 	peer := r.getPeer(peerId)
 	if peer == nil {
 		return fmt.Errorf("peer not found %v", peerId)
@@ -162,14 +178,13 @@ func (r *Registry) Unsubscribe(peerId discover.NodeID, s string, t []byte) error
 
 	msg := &UnsubscribeMsg{
 		Stream: s,
-		Key:    t,
 	}
-	log.Debug("Unsubscribe ", "peer", peerId, "stream", s, "key", t)
+	log.Debug("Unsubscribe ", "peer", peerId, "stream", s)
 
 	if err := peer.Send(msg); err != nil {
 		return err
 	}
-	return peer.removeClient(s, t)
+	return peer.removeClient(s)
 }
 
 func (r *Registry) Retrieve(chunk *storage.Chunk) error {
@@ -261,20 +276,11 @@ func (p *Peer) HandleMsg(msg interface{}) error {
 	}
 }
 
-func keyToString(key []byte) string {
-	l := len(key)
-	if l == 0 {
-		return ""
-	}
-	return fmt.Sprintf("%s-%d", string(key[:l-1]), uint8(key[l-1]))
-}
-
 type server struct {
 	Server
+	stream       Stream
 	priority     uint8
 	currentBatch []byte
-	stream       string
-	key          []byte
 }
 
 // Server interface for outgoing peer Streamer
@@ -286,50 +292,59 @@ type Server interface {
 
 type client struct {
 	Client
+	stream    Stream
 	priority  uint8
 	sessionAt uint64
-	live      bool
-	stream    string
-	key       []byte
 	next      chan error
+
+	intervalsKey   string
+	intervalsStore intervals.Store
+}
+
+func peerStreamIntervalsKey(p *Peer, s Stream) string {
+	return p.ID().String() + s.String()
+}
+
+func (c client) AddInterval(start, end uint64) (err error) {
+	i, err := c.intervalsStore.Get(c.intervalsKey)
+	if err != nil {
+		return err
+	}
+	i.Add(start, end)
+	return c.intervalsStore.Put(c.intervalsKey, i)
+}
+
+func (c client) NextInterval() (start, end uint64, err error) {
+	i, err := c.intervalsStore.Get(c.intervalsKey)
+	if err != nil {
+		return 0, 0, err
+	}
+	start, end = i.Next()
+	return start, end, nil
 }
 
 // Client interface for incoming peer Streamer
 type Client interface {
 	NeedData([]byte) func()
-	BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error)
+	BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error)
 	Close()
 }
 
-// nextBatch adjusts the indexes by inspecting the intervals
 func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
-	var intervals []uint64
+	if c.stream.Live {
-	if c.live {
+		return from, 0
-		if len(intervals) == 0 {
+	} else if from >= c.sessionAt {
-			intervals = []uint64{c.sessionAt, from}
+		return from, math.MaxUint64
-		} else {
+	}
-			intervals[1] = from
+	nextFrom, nextTo, err := c.NextInterval()
-		}
+	if err != nil {
-		nextFrom = from
+		log.Error("next intervals", "stream", c.stream)
-	} else if from >= c.sessionAt { // history sync complete
+		return
-		intervals = nil
+	}
-		nextFrom = from
+	if nextTo == 0 {
-		nextTo = math.MaxUint64
-	} else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals
-		intervals = append(intervals[:1], intervals[3:]...)
-		nextFrom = intervals[1]
-		if len(intervals) > 2 {
-			nextTo = intervals[2]
-		} else {
-			nextTo = c.sessionAt
-		}
-	} else {
-		nextFrom = from
-		intervals[1] = from
 		nextTo = c.sessionAt
 	}
-	// b.intervals.set(intervals)
+	return
-	return nextFrom, nextTo
 }
 
 func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error {
@@ -338,6 +353,10 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error
 		if err != nil {
 			return err
 		}
+		// TODO: make a test case for testing if the interval is added when the batch is done
+		if err := c.AddInterval(tp.Takeover.Start, tp.Takeover.End); err != nil {
+			return err
+		}
 		return p.SendPriority(tp, c.priority)
 	}
 	return nil
@@ -399,6 +418,10 @@ func (r *Registry) Stop() error {
 	return nil
 }
 
+type Range struct {
+	From, To uint64
+}
+
 type API struct {
 	streamer *Registry
 	dpa      *storage.DPA
@@ -432,10 +455,10 @@ func (api *API) ReadAll(hash common.Hash) (int64, error) {
 	return readAll(api.dpa, hash[:])
 }
 
-func (api *API) SubscribeStream(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
+func (api *API) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error {
-	return api.streamer.Subscribe(peerId, s, t, from, to, priority, live)
+	return api.streamer.Subscribe(peerId, s, history, priority)
 }
 
-func (api *API) UnsubscribeStream(peerId discover.NodeID, s string, t []byte) error {
+func (api *API) UnsubscribeStream(peerId discover.NodeID, s Stream) error {
-	return api.streamer.Unsubscribe(peerId, s, t)
+	return api.streamer.Unsubscribe(peerId, s)
 }

swarm/network/stream/streamer_test.go

@@ -32,7 +32,8 @@ func TestStreamerSubscribe(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = streamer.Subscribe(tester.IDs[0], "foo", nil, 0, 0, Top, true)
+	stream := NewStream("foo", nil, true)
+	err = streamer.Subscribe(tester.IDs[0], stream, &Range{From: 0, To: 0}, Top)
 	if err == nil || err.Error() != "stream foo not registered" {
 		t.Fatalf("Expected error %v, got %v", "stream foo not registered", err)
 	}
@@ -72,7 +73,7 @@ func (self *testClient) NeedData(hash []byte) func() {
 	return nil
 }
 
-func (self *testClient) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) {
+func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {
 	close(batchDone)
 	return nil
 }
@@ -97,7 +98,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
 		return &testClient{
 			t: t,
 		}, nil
@@ -105,7 +106,8 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 
 	peerID := tester.IDs[0]
 
-	err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true)
+	stream := NewStream("foo", nil, true)
+	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -116,10 +118,11 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			p2ptest.Expect{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "foo",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
-					From:     5,
+						From: 5,
-					To:       8,
+						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -131,7 +134,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = streamer.Unsubscribe(peerID, "foo", []byte{})
+	err = streamer.Unsubscribe(peerID, stream)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -142,8 +145,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			p2ptest.Expect{
 				Code: 0,
 				Msg: &UnsubscribeMsg{
-					Stream: "foo",
+					Stream: stream,
-					Key:    []byte{},
 				},
 				Peer: peerID,
 			},
@@ -162,7 +164,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) {
+	stream := NewStream("foo", nil, false)
+
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
 		return &testServer{
 			t: t,
 		}, nil
@@ -176,10 +180,11 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			p2ptest.Trigger{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "foo",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
-					From:     5,
+						From: 5,
-					To:       8,
+						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -189,14 +194,14 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			p2ptest.Expect{
 				Code: 1,
 				Msg: &OfferedHashesMsg{
-					Stream: "foo",
+					Stream: stream,
-					Key:    []byte{},
 					HandoverProof: &HandoverProof{
 						Handover: &Handover{},
 					},
-					Hashes: make([]byte, HashSize),
+					Hashes:  make([]byte, HashSize),
-					From:   6,
+					From:    6,
-					To:     9,
+					To:      9,
+					Initial: true,
 				},
 				Peer: peerID,
 			},
@@ -213,8 +218,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			p2ptest.Trigger{
 				Code: 0,
 				Msg: &UnsubscribeMsg{
-					Stream: "foo",
+					Stream: stream,
-					Key:    []byte{},
 				},
 				Peer: peerID,
 			},
@@ -233,12 +237,14 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) {
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
 		return &testServer{
 			t: t,
 		}, nil
 	})
 
+	stream := NewStream("bar", nil, true)
+
 	peerID := tester.IDs[0]
 
 	err = tester.TestExchanges(p2ptest.Exchange{
@@ -247,10 +253,11 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 			p2ptest.Trigger{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "bar",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
-					From:     5,
+						From: 5,
-					To:       8,
+						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -272,6 +279,78 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 	}
 }
 
+// TODO: fix: tests with TestExchanges are inconsistent because Expects check
+// 			  ordering is not guarrantied but fails if the order is wrong.
+// func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
+// 	tester, streamer, _, teardown, err := newStreamerTester(t)
+// 	defer teardown()
+// 	if err != nil {
+// 		t.Fatal(err)
+// 	}
+
+// 	stream := NewStream("foo", nil, true)
+
+// 	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+// 		return &testServer{
+// 			t: t,
+// 		}, nil
+// 	})
+
+// 	peerID := tester.IDs[0]
+
+// 	err = tester.TestExchanges(p2ptest.Exchange{
+// 		Label: "Subscribe message",
+// 		Triggers: []p2ptest.Trigger{
+// 			{
+// 				Code: 4,
+// 				Msg: &SubscribeMsg{
+// 					Stream: stream,
+// 					History: &Range{
+// 						From: 5,
+// 						To:   8,
+// 					},
+// 					Priority: Top,
+// 				},
+// 				Peer: peerID,
+// 			},
+// 		},
+// 		Expects: []p2ptest.Expect{
+// 			{
+// 				Code: 1,
+// 				Msg: &OfferedHashesMsg{
+// 					Stream: NewStream("foo", nil, false),
+// 					HandoverProof: &HandoverProof{
+// 						Handover: &Handover{},
+// 					},
+// 					Hashes:  make([]byte, HashSize),
+// 					From:    6,
+// 					To:      9,
+// 					Initial: true,
+// 				},
+// 				Peer: peerID,
+// 			},
+// 			{
+// 				Code: 1,
+// 				Msg: &OfferedHashesMsg{
+// 					Stream: stream,
+// 					HandoverProof: &HandoverProof{
+// 						Handover: &Handover{},
+// 					},
+// 					From:    1,
+// 					To:      1,
+// 					Hashes:  make([]byte, HashSize),
+// 					Initial: true,
+// 				},
+// 				Peer: peerID,
+// 			},
+// 		},
+// 	})
+
+// 	if err != nil {
+// 		t.Fatal(err)
+// 	}
+// }
+
 func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 	tester, streamer, _, teardown, err := newStreamerTester(t)
 	defer teardown()
@@ -279,7 +358,9 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
+	stream := NewStream("foo", nil, true)
+
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
 		return &testClient{
 			t: t,
 		}, nil
@@ -287,7 +368,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 
 	peerID := tester.IDs[0]
 
-	err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true)
+	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -298,10 +379,11 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 			p2ptest.Expect{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "foo",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
-					From:     5,
+						From: 5,
-					To:       8,
+						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -320,7 +402,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 						Hashes: hashes,
 						From:   5,
 						To:     8,
-						Stream: "foo",
+						Stream: stream,
 					},
 					Peer: peerID,
 				},
@@ -329,7 +411,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 				p2ptest.Expect{
 					Code: 2,
 					Msg: &WantedHashesMsg{
-						Stream: "foo",
+						Stream: stream,
 						Want:   []byte{5},
 						From:   8,
 						To:     0,

swarm/network/stream/syncer.go

@@ -64,10 +64,9 @@ 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) (Server, error) {
+	streamer.RegisterServerFunc("SYNC", func(p *Peer, t []byte, live bool) (Server, error) {
 		po := uint8(t[0])
-		// TODO: make this work for HISTORY too
+		return NewSwarmSyncerServer(live, po, db)
-		return NewSwarmSyncerServer(false, po, db)
 	})
 	// streamer.RegisterServerFunc(stream, func(p *Peer) (Server, error) {
 	// 	return NewOutgoingProvableSwarmSyncer(po, db)
@@ -188,7 +187,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) (Client, error) {
+	streamer.RegisterClientFunc("SYNC", func(p *Peer, t []byte, love bool) (Client, error) {
 		return NewSwarmSyncerClient(p, db, nil)
 	})
 }
@@ -207,14 +206,14 @@ func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
 }
 
 // BatchDone
-func (s *SwarmSyncerClient) BatchDone(streamName string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
+func (s *SwarmSyncerClient) BatchDone(stream Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
 	if s.chunker != nil {
-		return func() (*TakeoverProof, error) { return s.TakeoverProof(streamName, from, hashes, root) }
+		return func() (*TakeoverProof, error) { return s.TakeoverProof(stream, from, hashes, root) }
 	}
 	return nil
 }
 
-func (s *SwarmSyncerClient) TakeoverProof(streamName string, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
+func (s *SwarmSyncerClient) TakeoverProof(stream Stream, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
 	// for provable syncer currentRoot is non-zero length
 	if s.chunker != nil {
 		if from > s.sessionAt { // for live syncing currentRoot is always updated
@@ -241,11 +240,10 @@ func (s *SwarmSyncerClient) TakeoverProof(streamName string, from uint64, hashes
 	}
 	s.end += uint64(len(hashes)) / HashSize
 	takeover := &Takeover{
-		Stream: streamName,
+		Stream: stream,
-		// Key:    s.Key,
+		Start:  s.start,
-		Start: s.start,
+		End:    s.end,
-		End:   s.end,
+		Root:   root,
-		Root:  root,
 	}
 	// serialise and sign
 	return &TakeoverProof{

swarm/network/stream/syncer_test.go

@@ -160,7 +160,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, "SYNC", []byte{1}, 0, 0, Top, false)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", []byte{1}, false), &Range{From: 0, To: 0}, Top)
 			})
 			if err != nil {
 				return err