aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorAttila Gazso <attila.gazso@prezi.com>2018-10-18 01:22:37 +0800
committerViktor TrĂ³n <viktor.tron@gmail.com>2018-10-18 01:22:37 +0800
commitcdf5982cfca2cd7d5fea85c226af5e48fde837df (patch)
treeadcd4a5b24d285ca72174c085bd17a898b628721
parent4e693ad5a6f118dcf4f3586938c17fc8ef5822b5 (diff)
downloaddexon-cdf5982cfca2cd7d5fea85c226af5e48fde837df.tar.gz
dexon-cdf5982cfca2cd7d5fea85c226af5e48fde837df.tar.zst
dexon-cdf5982cfca2cd7d5fea85c226af5e48fde837df.zip
swarm: Lightnode mode: disable sync, retrieve, subscription (#17899)
* swarm: Lightnode mode: disable sync, retrieve, subscription * swarm/network/stream: assign error and check in one line * swarm: restructured RegistryOption initializing * swarm: empty commit to retrigger CI build * swarm/network/stream: Added comments explaining RegistryOptions
-rw-r--r--swarm/network/stream/delivery_test.go15
-rw-r--r--swarm/network/stream/lightnode_test.go210
-rw-r--r--swarm/network/stream/messages.go11
-rw-r--r--swarm/network/stream/snapshot_retrieval_test.go1
-rw-r--r--swarm/network/stream/snapshot_sync_test.go6
-rw-r--r--swarm/network/stream/stream.go22
-rw-r--r--swarm/swarm.go10
7 files changed, 260 insertions, 15 deletions
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 <http://www.gnu.org/licenses/>.
+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)