diff --git a/swarm/chunk/chunk.go b/swarm/chunk/chunk.go index ec4d1057c0..e730ab5b41 100644 --- a/swarm/chunk/chunk.go +++ b/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 -type SyncStore interface { +// FetchStore is a Store which supports syncing +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 } diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 07a96d7f68..b75d62e010 100644 --- a/swarm/network/stream/common_test.go +++ b/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() diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 75e87b3049..47c20e26e1 100644 --- a/swarm/network/stream/delivery.go +++ b/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, diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index 009a941ef4..1f2cdcada7 100644 --- a/swarm/network/stream/intervals_test.go +++ b/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, diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index c7c489152a..6265f6aaad 100644 --- a/swarm/network/stream/stream.go +++ b/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{} } diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go index 2e46177c9c..89ae0f5c7b 100644 --- a/swarm/network/stream/syncer.go +++ b/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)) }) diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index 68dafc7788..4834aebce8 100644 --- a/swarm/storage/common_test.go +++ b/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 } diff --git a/swarm/storage/localstore/subscription_pull.go b/swarm/storage/localstore/subscription_pull.go index 0b96102e35..64e1cda9db 100644 --- a/swarm/storage/localstore/subscription_pull.go +++ b/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()) { - chunkDescriptors := make(chan ChunkDescriptor) +func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *chunk.Descriptor) (c <-chan chunk.Descriptor, stop func()) { + 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 diff --git a/swarm/storage/localstore/subscription_pull_test.go b/swarm/storage/localstore/subscription_pull_test.go index 4c68f41296..e4e2854325 100644 --- a/swarm/storage/localstore/subscription_pull_test.go +++ b/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 -// sends error to errChan, even if it is nil, to count the number of ChunkDescriptors +// 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 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, } diff --git a/swarm/storage/types.go b/swarm/storage/types.go index fe510305fa..a955191891 100644 --- a/swarm/storage/types.go +++ b/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