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
This commit is contained in:
parent
4e693ad5a6
commit
cdf5982cfc
@ -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)
|
||||
@ -337,6 +343,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
|
||||
|
||||
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
|
||||
SkipCheck: skipCheck,
|
||||
DoServeRetrieve: true,
|
||||
})
|
||||
bucket.Store(bucketKeyRegistry, r)
|
||||
|
||||
|
210
swarm/network/stream/lightnode_test.go
Normal file
210
swarm/network/stream/lightnode_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
@ -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) {
|
||||
|
@ -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())
|
||||
|
@ -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())
|
||||
|
@ -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
|
||||
|
||||
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))
|
||||
})
|
||||
|
||||
if options.DoSync {
|
||||
RegisterSwarmSyncerServer(streamer, syncChunkStore)
|
||||
RegisterSwarmSyncerClient(streamer, syncChunkStore)
|
||||
}
|
||||
|
||||
if options.DoSync {
|
||||
// latestIntC function ensures that
|
||||
|
@ -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)
|
||||
|
Loading…
Reference in New Issue
Block a user