swarm: extend chunk.Store

swarm/chunk/chunk.go

@@ -148,20 +148,33 @@ const (
 	ModeSetRemove
 )
 
+// Descriptor holds information required for Pull syncing. This struct
+// is provided by subscribing to pull index.
+type Descriptor struct {
+	Address        Address
+	StoreTimestamp int64
+}
+
+func (c *Descriptor) String() string {
+	if c == nil {
+		return "none"
+	}
+	return fmt.Sprintf("%s stored at %v", c.Address.Hex(), c.StoreTimestamp)
+}
+
 type Store interface {
 	Get(ctx context.Context, mode ModeGet, addr Address) (ch Chunk, err error)
 	Put(ctx context.Context, mode ModePut, ch Chunk) (err error)
-	//Set(ctx context.Context, mode ModeSet, addr Address) (err error)
 	Has(ctx context.Context, addr Address) (yes bool, err error)
+	Set(ctx context.Context, mode ModeSet, addr Address) (err error)
+	LastPullSubscriptionChunk(bin uint8) (c *Descriptor, err error)
+	SubscribePull(ctx context.Context, bin uint8, since, until *Descriptor) (c <-chan Descriptor, stop func())
 	Close() (err error)
 }
 
-// SyncStore is a Store which supports syncing
+// FetchStore is a Store which supports syncing
-type SyncStore interface {
+type FetchStore interface {
 	Store
-	// BinIndex(po uint8) uint64
-	// Iterator(from uint64, to uint64, po uint8, f func(Address, uint64) bool) error
-	// SubscribePull(ctx context.Context, bin uint8, since, until *localstore.ChunkDescriptor) (c <-chan localstore.ChunkDescriptor, stop func())
 	FetchFunc(ctx context.Context, addr Address) func(context.Context) error
 }
 

swarm/network/stream/common_test.go

@@ -231,6 +231,18 @@ func (rrs *roundRobinStore) Put(ctx context.Context, mode chunk.ModePut, ch stor
 	return rrs.stores[idx].Put(ctx, mode, ch)
 }
 
+func (rrs *roundRobinStore) Set(ctx context.Context, mode chunk.ModeSet, addr chunk.Address) (err error) {
+	return nil
+}
+
+func (rrs *roundRobinStore) LastPullSubscriptionChunk(bin uint8) (c *chunk.Descriptor, err error) {
+	return nil, nil
+}
+
+func (rrs *roundRobinStore) SubscribePull(ctx context.Context, bin uint8, since, until *chunk.Descriptor) (c <-chan chunk.Descriptor, stop func()) {
+	return nil, nil
+}
+
 func (rrs *roundRobinStore) Close() error {
 	for _, store := range rrs.stores {
 		store.Close()

swarm/network/stream/delivery.go

@@ -48,12 +48,12 @@ var (
 )
 
 type Delivery struct {
-	chunkStore storage.SyncChunkStore
+	chunkStore chunk.FetchStore
 	kad        *network.Kademlia
 	getPeer    func(enode.ID) *Peer
 }
 
-func NewDelivery(kad *network.Kademlia, chunkStore storage.SyncChunkStore) *Delivery {
+func NewDelivery(kad *network.Kademlia, chunkStore chunk.FetchStore) *Delivery {
 	return &Delivery{
 		chunkStore: chunkStore,
 		kad:        kad,

swarm/network/stream/intervals_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+	"github.com/ethereum/go-ethereum/swarm/chunk"
 	"github.com/ethereum/go-ethereum/swarm/network/simulation"
 	"github.com/ethereum/go-ethereum/swarm/state"
 	"github.com/ethereum/go-ethereum/swarm/storage"
@@ -287,11 +288,11 @@ func enableNotifications(r *Registry, peerID enode.ID, s Stream) error {
 
 type testExternalClient struct {
 	hashes               chan []byte
-	store                storage.SyncChunkStore
+	store                chunk.FetchStore
 	enableNotificationsC chan struct{}
 }
 
-func newTestExternalClient(store storage.SyncChunkStore) *testExternalClient {
+func newTestExternalClient(store chunk.FetchStore) *testExternalClient {
 	return &testExternalClient{
 		hashes:               make(chan []byte),
 		store:                store,

swarm/network/stream/stream.go

@@ -30,11 +30,11 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/swarm/chunk"
 	"github.com/ethereum/go-ethereum/swarm/log"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/state"
-	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
 const (
@@ -108,7 +108,7 @@ type RegistryOptions struct {
 }
 
 // NewRegistry is Streamer constructor
-func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.SyncChunkStore, intervalsStore state.Store, options *RegistryOptions, balance protocols.Balance) *Registry {
+func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore chunk.FetchStore, intervalsStore state.Store, options *RegistryOptions, balance protocols.Balance) *Registry {
 	if options == nil {
 		options = &RegistryOptions{}
 	}

swarm/network/stream/syncer.go

@@ -37,12 +37,12 @@ const (
 // * (live/non-live historical) chunk syncing per proximity bin
 type SwarmSyncerServer struct {
 	po    uint8
-	store storage.SyncChunkStore
+	store chunk.FetchStore
 	quit  chan struct{}
 }
 
 // NewSwarmSyncerServer is constructor for SwarmSyncerServer
-func NewSwarmSyncerServer(po uint8, syncChunkStore storage.SyncChunkStore) (*SwarmSyncerServer, error) {
+func NewSwarmSyncerServer(po uint8, syncChunkStore chunk.FetchStore) (*SwarmSyncerServer, error) {
 	return &SwarmSyncerServer{
 		po:    po,
 		store: syncChunkStore,
@@ -50,7 +50,7 @@ func NewSwarmSyncerServer(po uint8, syncChunkStore storage.SyncChunkStore) (*Swa
 	}, nil
 }
 
-func RegisterSwarmSyncerServer(streamer *Registry, syncChunkStore storage.SyncChunkStore) {
+func RegisterSwarmSyncerServer(streamer *Registry, syncChunkStore chunk.FetchStore) {
 	streamer.RegisterServerFunc("SYNC", func(_ *Peer, t string, _ bool) (Server, error) {
 		po, err := ParseSyncBinKey(t)
 		if err != nil {
@@ -134,13 +134,13 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
 
 // SwarmSyncerClient
 type SwarmSyncerClient struct {
-	store  storage.SyncChunkStore
+	store  chunk.FetchStore
 	peer   *Peer
 	stream Stream
 }
 
 // NewSwarmSyncerClient is a contructor for provable data exchange syncer
-func NewSwarmSyncerClient(p *Peer, store storage.SyncChunkStore, stream Stream) (*SwarmSyncerClient, error) {
+func NewSwarmSyncerClient(p *Peer, store chunk.FetchStore, stream Stream) (*SwarmSyncerClient, error) {
 	return &SwarmSyncerClient{
 		store:  store,
 		peer:   p,
@@ -186,7 +186,7 @@ func NewSwarmSyncerClient(p *Peer, store storage.SyncChunkStore, stream Stream)
 
 // RegisterSwarmSyncerClient registers the client constructor function for
 // to handle incoming sync streams
-func RegisterSwarmSyncerClient(streamer *Registry, store storage.SyncChunkStore) {
+func RegisterSwarmSyncerClient(streamer *Registry, store chunk.FetchStore) {
 	streamer.RegisterClientFunc("SYNC", func(p *Peer, t string, live bool) (Client, error) {
 		return NewSwarmSyncerClient(p, store, NewStream("SYNC", t, live))
 	})

swarm/storage/common_test.go

@@ -250,6 +250,18 @@ func (m *MapChunkStore) Has(ctx context.Context, ref Address) (has bool, err err
 	return has, nil
 }
 
+func (m *MapChunkStore) Set(ctx context.Context, mode chunk.ModeSet, addr chunk.Address) (err error) {
+	return nil
+}
+
+func (m *MapChunkStore) LastPullSubscriptionChunk(bin uint8) (c *chunk.Descriptor, err error) {
+	return nil, nil
+}
+
+func (m *MapChunkStore) SubscribePull(ctx context.Context, bin uint8, since, until *chunk.Descriptor) (c <-chan chunk.Descriptor, stop func()) {
+	return nil, nil
+}
+
 func (m *MapChunkStore) Close() error {
 	return nil
 }

swarm/storage/localstore/subscription_pull.go

@@ -20,7 +20,6 @@ import (
 	"bytes"
 	"context"
 	"errors"
-	"fmt"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -37,8 +36,8 @@ import (
 // function will terminate current and further iterations without errors, and also close the returned channel.
 // Make sure that you check the second returned parameter from the channel to stop iteration when its value
 // is false.
-func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkDescriptor) (c <-chan ChunkDescriptor, stop func()) {
+func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *chunk.Descriptor) (c <-chan chunk.Descriptor, stop func()) {
-	chunkDescriptors := make(chan ChunkDescriptor)
+	chunkDescriptors := make(chan chunk.Descriptor)
 	trigger := make(chan struct{}, 1)
 
 	db.pullTriggersMu.Lock()
@@ -59,7 +58,7 @@ func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkD
 	var errStopSubscription = errors.New("stop subscription")
 
 	go func() {
-		// close the returned ChunkDescriptor channel at the end to
+		// close the returned chunk.Descriptor channel at the end to
 		// signal that the subscription is done
 		defer close(chunkDescriptors)
 		// sinceItem is the Item from which the next iteration
@@ -80,7 +79,7 @@ func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkD
 				// - context is done
 				err := db.pullIndex.Iterate(func(item shed.Item) (stop bool, err error) {
 					select {
-					case chunkDescriptors <- ChunkDescriptor{
+					case chunkDescriptors <- chunk.Descriptor{
 						Address:        item.Address,
 						StoreTimestamp: item.StoreTimestamp,
 					}:
@@ -159,10 +158,10 @@ func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkD
 	return chunkDescriptors, stop
 }
 
-// LastPullSubscriptionChunk returns ChunkDescriptor of the latest Chunk
+// LastPullSubscriptionChunk returns chunk.Descriptor of the latest Chunk
 // in pull syncing index for a provided bin. If there are no chunks in
 // that bin, chunk.ErrChunkNotFound is returned.
-func (db *DB) LastPullSubscriptionChunk(bin uint8) (c *ChunkDescriptor, err error) {
+func (db *DB) LastPullSubscriptionChunk(bin uint8) (c *chunk.Descriptor, err error) {
 	item, err := db.pullIndex.Last([]byte{bin})
 	if err != nil {
 		if err == leveldb.ErrNotFound {
@@ -170,26 +169,12 @@ func (db *DB) LastPullSubscriptionChunk(bin uint8) (c *ChunkDescriptor, err erro
 		}
 		return nil, err
 	}
-	return &ChunkDescriptor{
+	return &chunk.Descriptor{
 		Address:        item.Address,
 		StoreTimestamp: item.StoreTimestamp,
 	}, nil
 }
 
-// ChunkDescriptor holds information required for Pull syncing. This struct
-// is provided by subscribing to pull index.
-type ChunkDescriptor struct {
-	Address        chunk.Address
-	StoreTimestamp int64
-}
-
-func (c *ChunkDescriptor) String() string {
-	if c == nil {
-		return "none"
-	}
-	return fmt.Sprintf("%s stored at %v", c.Address.Hex(), c.StoreTimestamp)
-}
-
 // triggerPullSubscriptions is used internally for starting iterations
 // on Pull subscriptions for a particular bin. When new item with address
 // that is in particular bin for DB's baseKey is added to pull index

swarm/storage/localstore/subscription_pull_test.go

@@ -150,11 +150,11 @@ func TestDB_SubscribePull_since(t *testing.T) {
 		return lastTimestamp
 	})()
 
-	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
+	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]chunk.Descriptor) {
 		addrsMu.Lock()
 		defer addrsMu.Unlock()
 
-		last = make(map[uint8]ChunkDescriptor)
+		last = make(map[uint8]chunk.Descriptor)
 		for i := 0; i < count; i++ {
 			ch := generateTestRandomChunk()
 
@@ -177,7 +177,7 @@ func TestDB_SubscribePull_since(t *testing.T) {
 			storeTimestamp := lastTimestamp
 			lastTimestampMu.RUnlock()
 
-			last[bin] = ChunkDescriptor{
+			last[bin] = chunk.Descriptor{
 				Address:        ch.Address(),
 				StoreTimestamp: storeTimestamp,
 			}
@@ -200,7 +200,7 @@ func TestDB_SubscribePull_since(t *testing.T) {
 	errChan := make(chan error)
 
 	for bin := uint8(0); bin <= uint8(chunk.MaxPO); bin++ {
-		var since *ChunkDescriptor
+		var since *chunk.Descriptor
 		if c, ok := last[bin]; ok {
 			since = &c
 		}
@@ -241,11 +241,11 @@ func TestDB_SubscribePull_until(t *testing.T) {
 		return lastTimestamp
 	})()
 
-	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
+	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]chunk.Descriptor) {
 		addrsMu.Lock()
 		defer addrsMu.Unlock()
 
-		last = make(map[uint8]ChunkDescriptor)
+		last = make(map[uint8]chunk.Descriptor)
 		for i := 0; i < count; i++ {
 			ch := generateTestRandomChunk()
 
@@ -268,7 +268,7 @@ func TestDB_SubscribePull_until(t *testing.T) {
 			storeTimestamp := lastTimestamp
 			lastTimestampMu.RUnlock()
 
-			last[bin] = ChunkDescriptor{
+			last[bin] = chunk.Descriptor{
 				Address:        ch.Address(),
 				StoreTimestamp: storeTimestamp,
 			}
@@ -331,11 +331,11 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
 		return lastTimestamp
 	})()
 
-	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
+	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]chunk.Descriptor) {
 		addrsMu.Lock()
 		defer addrsMu.Unlock()
 
-		last = make(map[uint8]ChunkDescriptor)
+		last = make(map[uint8]chunk.Descriptor)
 		for i := 0; i < count; i++ {
 			ch := generateTestRandomChunk()
 
@@ -358,7 +358,7 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
 			storeTimestamp := lastTimestamp
 			lastTimestampMu.RUnlock()
 
-			last[bin] = ChunkDescriptor{
+			last[bin] = chunk.Descriptor{
 				Address:        ch.Address(),
 				StoreTimestamp: storeTimestamp,
 			}
@@ -387,7 +387,7 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
 	errChan := make(chan error)
 
 	for bin := uint8(0); bin <= uint8(chunk.MaxPO); bin++ {
-		var since *ChunkDescriptor
+		var since *chunk.Descriptor
 		if c, ok := upload1[bin]; ok {
 			since = &c
 		}
@@ -434,10 +434,10 @@ func uploadRandomChunksBin(t *testing.T, db *DB, uploader *Putter, addrs map[uin
 	}
 }
 
-// readPullSubscriptionBin is a helper function that reads all ChunkDescriptors from a channel and
+// readPullSubscriptionBin is a helper function that reads all chunk.Descriptors from a channel and
-// sends error to errChan, even if it is nil, to count the number of ChunkDescriptors
+// sends error to errChan, even if it is nil, to count the number of chunk.Descriptors
 // returned by the channel.
-func readPullSubscriptionBin(ctx context.Context, bin uint8, ch <-chan ChunkDescriptor, addrs map[uint8][]chunk.Address, addrsMu *sync.Mutex, errChan chan error) {
+func readPullSubscriptionBin(ctx context.Context, bin uint8, ch <-chan chunk.Descriptor, addrs map[uint8][]chunk.Address, addrsMu *sync.Mutex, errChan chan error) {
 	var i int // address index
 	for {
 		select {
@@ -506,7 +506,7 @@ func TestDB_LastPullSubscriptionChunk(t *testing.T) {
 		return lastTimestamp
 	})()
 
-	last := make(map[uint8]ChunkDescriptor)
+	last := make(map[uint8]chunk.Descriptor)
 
 	// do a few rounds of uploads and check if
 	// last pull subscription chunk is correct
@@ -532,7 +532,7 @@ func TestDB_LastPullSubscriptionChunk(t *testing.T) {
 			storeTimestamp := lastTimestamp
 			lastTimestampMu.RUnlock()
 
-			last[bin] = ChunkDescriptor{
+			last[bin] = chunk.Descriptor{
 				Address:        ch.Address(),
 				StoreTimestamp: storeTimestamp,
 			}

swarm/storage/types.go

@@ -211,8 +211,6 @@ func (v *ContentAddressValidator) Validate(ch Chunk) bool {
 
 type ChunkStore = chunk.Store
 
-type SyncChunkStore = chunk.SyncStore
-
 // FakeChunkStore doesn't store anything, just implements the ChunkStore interface
 // It can be used to inject into a hasherStore if you don't want to actually store data just do the
 // hashing
@@ -234,6 +232,18 @@ func (f *FakeChunkStore) Get(_ context.Context, _ chunk.ModeGet, ref Address) (C
 	panic("FakeChunkStore doesn't support Get")
 }
 
+func (f *FakeChunkStore) Set(ctx context.Context, mode chunk.ModeSet, addr chunk.Address) (err error) {
+	return nil
+}
+
+func (f *FakeChunkStore) LastPullSubscriptionChunk(bin uint8) (c *chunk.Descriptor, err error) {
+	return nil, nil
+}
+
+func (f *FakeChunkStore) SubscribePull(ctx context.Context, bin uint8, since, until *chunk.Descriptor) (c <-chan chunk.Descriptor, stop func()) {
+	return nil, nil
+}
+
 // Close doesn't store anything it is just here to implement ChunkStore
 func (f *FakeChunkStore) Close() error {
 	return nil