swarm: Lightnode mode: disable sync, retrieve, subscription

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)
 

swarm/network/stream/lightnode_test.go

@@ -0,0 +1,187 @@
+package stream
+
+import (
+	"testing"
+
+	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
+)
+
+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)
+	}
+}
+
+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)
+	}
+}
+
+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)
+	}
+}
+
+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)
+	}
+}

swarm/network/stream/messages.go

@@ -76,7 +76,17 @@ 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)
+	err = p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority)
+	if 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) {

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())

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())

swarm/network/stream/stream.go

@@ -68,6 +68,7 @@ type RegistryOptions struct {
 	SkipCheck       bool
 	DoSync          bool
 	DoRetrieve      bool
+	DoServeRetrieve bool
 	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

swarm/swarm.go

@@ -177,8 +177,9 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e
 	}
 	self.streamer = stream.NewRegistry(nodeID, delivery, self.netStore, stateStore, &stream.RegistryOptions{
 		SkipCheck:       config.DeliverySkipCheck,
-		DoSync:          config.SyncEnabled,
+		DoSync:          config.SyncEnabled && !config.LightNodeEnabled,
 		DoRetrieve:      true,
+		DoServeRetrieve: !config.LightNodeEnabled,
 		SyncUpdateDelay: config.SyncUpdateDelay,
 		MaxPeerServers:  config.MaxStreamPeerServers,
 	})