diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index c6ebae3f0..ec6148a0b 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -75,7 +75,9 @@ func TestStreamerRetrieveRequest(t *testing.T) { } func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { - tester, streamer, _, teardown, err := newStreamerTester(t, nil) + tester, streamer, _, teardown, err := newStreamerTester(t, &RegistryOptions{ + DoServeRetrieve: true, + }) defer teardown() if err != nil { t.Fatal(err) @@ -127,7 +129,9 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { // upstream request server receives a retrieve Request and responds with // offered hashes or delivery if skipHash is set to true func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { - tester, streamer, localStore, teardown, err := newStreamerTester(t, nil) + tester, streamer, localStore, teardown, err := newStreamerTester(t, &RegistryOptions{ + DoServeRetrieve: true, + }) defer teardown() if err != nil { t.Fatal(err) @@ -221,7 +225,9 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { } func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { - tester, streamer, localStore, teardown, err := newStreamerTester(t, nil) + tester, streamer, localStore, teardown, err := newStreamerTester(t, &RegistryOptions{ + DoServeRetrieve: true, + }) defer teardown() if err != nil { t.Fatal(err) @@ -336,7 +342,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ - SkipCheck: skipCheck, + SkipCheck: skipCheck, + DoServeRetrieve: true, }) bucket.Store(bucketKeyRegistry, r) diff --git a/swarm/network/stream/lightnode_test.go b/swarm/network/stream/lightnode_test.go new file mode 100644 index 000000000..0d3bc7f54 --- /dev/null +++ b/swarm/network/stream/lightnode_test.go @@ -0,0 +1,210 @@ +// 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 . +package stream + +import ( + "testing" + + p2ptest "github.com/ethereum/go-ethereum/p2p/testing" +) + +// This test checks the default behavior of the server, that is +// when it is serving Retrieve requests. +func TestLigthnodeRetrieveRequestWithRetrieve(t *testing.T) { + registryOptions := &RegistryOptions{ + DoServeRetrieve: true, + } + tester, _, _, teardown, err := newStreamerTester(t, registryOptions) + defer teardown() + if err != nil { + t.Fatal(err) + } + + node := tester.Nodes[0] + + stream := NewStream(swarmChunkServerStreamName, "", false) + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "SubscribeMsg", + Triggers: []p2ptest.Trigger{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: stream, + }, + Peer: node.ID(), + }, + }, + }) + if err != nil { + t.Fatalf("Got %v", err) + } + + err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID()}) + if err == nil || err.Error() != "timed out waiting for peers to disconnect" { + t.Fatalf("Expected no disconnect, got %v", err) + } +} + +// This test checks the Lightnode behavior of server, when serving Retrieve +// requests are disabled +func TestLigthnodeRetrieveRequestWithoutRetrieve(t *testing.T) { + registryOptions := &RegistryOptions{ + DoServeRetrieve: false, + } + tester, _, _, teardown, err := newStreamerTester(t, registryOptions) + defer teardown() + if err != nil { + t.Fatal(err) + } + + node := tester.Nodes[0] + + stream := NewStream(swarmChunkServerStreamName, "", false) + + err = tester.TestExchanges( + p2ptest.Exchange{ + Label: "SubscribeMsg", + Triggers: []p2ptest.Trigger{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: stream, + }, + Peer: node.ID(), + }, + }, + Expects: []p2ptest.Expect{ + { + Code: 7, + Msg: &SubscribeErrorMsg{ + Error: "stream RETRIEVE_REQUEST not registered", + }, + Peer: node.ID(), + }, + }, + }) + if err != nil { + t.Fatalf("Got %v", err) + } +} + +// This test checks the default behavior of the server, that is +// when syncing is enabled. +func TestLigthnodeRequestSubscriptionWithSync(t *testing.T) { + registryOptions := &RegistryOptions{ + DoSync: true, + } + tester, _, _, teardown, err := newStreamerTester(t, registryOptions) + defer teardown() + if err != nil { + t.Fatal(err) + } + + node := tester.Nodes[0] + + syncStream := NewStream("SYNC", FormatSyncBinKey(1), false) + + err = tester.TestExchanges( + p2ptest.Exchange{ + Label: "RequestSubscription", + Triggers: []p2ptest.Trigger{ + { + Code: 8, + Msg: &RequestSubscriptionMsg{ + Stream: syncStream, + }, + Peer: node.ID(), + }, + }, + Expects: []p2ptest.Expect{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: syncStream, + }, + Peer: node.ID(), + }, + }, + }) + + if err != nil { + t.Fatalf("Got %v", err) + } +} + +// This test checks the Lightnode behavior of the server, that is +// when syncing is disabled. +func TestLigthnodeRequestSubscriptionWithoutSync(t *testing.T) { + registryOptions := &RegistryOptions{ + DoSync: false, + } + tester, _, _, teardown, err := newStreamerTester(t, registryOptions) + defer teardown() + if err != nil { + t.Fatal(err) + } + + node := tester.Nodes[0] + + syncStream := NewStream("SYNC", FormatSyncBinKey(1), false) + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "RequestSubscription", + Triggers: []p2ptest.Trigger{ + { + Code: 8, + Msg: &RequestSubscriptionMsg{ + Stream: syncStream, + }, + Peer: node.ID(), + }, + }, + Expects: []p2ptest.Expect{ + { + Code: 7, + Msg: &SubscribeErrorMsg{ + Error: "stream SYNC not registered", + }, + Peer: node.ID(), + }, + }, + }, p2ptest.Exchange{ + Label: "RequestSubscription", + Triggers: []p2ptest.Trigger{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: syncStream, + }, + Peer: node.ID(), + }, + }, + Expects: []p2ptest.Expect{ + { + Code: 7, + Msg: &SubscribeErrorMsg{ + Error: "stream SYNC not registered", + }, + Peer: node.ID(), + }, + }, + }) + + if err != nil { + t.Fatalf("Got %v", err) + } +} diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 74c785d58..68503fe1f 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -76,7 +76,16 @@ type RequestSubscriptionMsg struct { func (p *Peer) handleRequestSubscription(ctx context.Context, req *RequestSubscriptionMsg) (err error) { log.Debug(fmt.Sprintf("handleRequestSubscription: streamer %s to subscribe to %s with stream %s", p.streamer.addr, p.ID(), req.Stream)) - return p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority) + if err = p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority); err != nil { + // The error will be sent as a subscribe error message + // and will not be returned as it will prevent any new message + // exchange between peers over p2p. Instead, error will be returned + // only if there is one from sending subscribe error message. + err = p.Send(ctx, SubscribeErrorMsg{ + Error: err.Error(), + }) + } + return err } func (p *Peer) handleSubscribeMsg(ctx context.Context, req *SubscribeMsg) (err error) { diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index 09d915d48..7625e02d7 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -130,6 +130,7 @@ func retrievalStreamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s no DoSync: true, SyncUpdateDelay: 3 * time.Second, DoRetrieve: true, + DoServeRetrieve: true, }) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 0d5849487..96e92c5cf 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -166,6 +166,7 @@ func streamerFunc(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Servic r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ DoSync: true, + DoServeRetrieve: true, SyncUpdateDelay: 3 * time.Second, }) @@ -358,7 +359,10 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int) delivery := NewDelivery(kad, netStore) netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New - r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), nil) + r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ + DoServeRetrieve: true, + DoSync: true, + }) bucket.Store(bucketKeyRegistry, r) fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams()) diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 3861cfcf6..1dc2a8cba 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -66,8 +66,9 @@ type Registry struct { // RegistryOptions holds optional values for NewRegistry constructor. type RegistryOptions struct { SkipCheck bool - DoSync bool - DoRetrieve bool + DoSync bool // Sets if the server syncs with peers. Default is true, set to false by lightnode or nosync flags. + DoRetrieve bool // Sets if the server issues Retrieve requests. Default is true. + DoServeRetrieve bool // Sets if the server serves Retrieve requests. Default is true, set to false by lightnode flag. SyncUpdateDelay time.Duration MaxPeerServers int // The limit of servers for each peer in registry } @@ -93,14 +94,21 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy } streamer.api = NewAPI(streamer) delivery.getPeer = streamer.getPeer - streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) { - return NewSwarmChunkServer(delivery.chunkStore), nil - }) + + if options.DoServeRetrieve { + streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) { + return NewSwarmChunkServer(delivery.chunkStore), nil + }) + } + streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t string, live bool) (Client, error) { return NewSwarmSyncerClient(p, syncChunkStore, NewStream(swarmChunkServerStreamName, t, live)) }) - RegisterSwarmSyncerServer(streamer, syncChunkStore) - RegisterSwarmSyncerClient(streamer, syncChunkStore) + + if options.DoSync { + RegisterSwarmSyncerServer(streamer, syncChunkStore) + RegisterSwarmSyncerClient(streamer, syncChunkStore) + } if options.DoSync { // latestIntC function ensures that diff --git a/swarm/swarm.go b/swarm/swarm.go index aea0989a1..7214abbda 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -175,13 +175,19 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e if err := nodeID.UnmarshalText([]byte(config.NodeID)); err != nil { return nil, err } - self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, &stream.RegistryOptions{ + registryOptions := &stream.RegistryOptions{ SkipCheck: config.DeliverySkipCheck, DoSync: config.SyncEnabled, DoRetrieve: true, + DoServeRetrieve: true, SyncUpdateDelay: config.SyncUpdateDelay, MaxPeerServers: config.MaxStreamPeerServers, - }) + } + if config.LightNodeEnabled { + registryOptions.DoSync = false + registryOptions.DoRetrieve = false + } + self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, registryOptions) // Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage self.fileStore = storage.NewFileStore(self.netStore, self.config.FileStoreParams)