go/p2p: Ensure only server providers advertise themselves

Author: martintomazic

go/oasis-node/cmd/debug/byzantine/node.go

@@ -155,6 +155,8 @@ func initializeAndRegisterByzantineNode(
 		return nil, fmt.Errorf("initializing storage node failed: %w", err)
 	}
 	b.p2p.service.RegisterProtocolServer(storageP2P.NewServer(b.chainContext, b.runtimeID, storage))
+	b.p2p.service.AdvertiseProtocol(storageP2P.RuntimeProtocolID(b.chainContext, b.runtimeID))
+
 	b.storage = storage
 
 	// Wait for activation epoch.

go/p2p/api/api.go

@@ -83,8 +83,16 @@ type Service interface {
 	RegisterProtocol(p core.ProtocolID, minPeers int, totalPeers int)
 
 	// RegisterProtocolServer registers a protocol server for the given protocol.
+	//
+	// Normally, registered protocol server should be also advertised (AdvertiseProtocol),
+	// so that remote peers find the host node.
 	RegisterProtocolServer(srv rpc.Server)
 
+	// AdvertiseProtocol starts advertising the host as available for serving the specified protocol.
+	//
+	// It should be called only after the protocol server is running and ready to serve.
+	AdvertiseProtocol(p core.ProtocolID)
+
 	// GetMinRepublishInterval returns the minimum republish interval that needs to be respected by
 	// the caller when publishing the same message. If Publish is called for the same message more
 	// quickly, the message may be dropped and not published.

go/p2p/nop.go

@@ -85,6 +85,10 @@ func (p *nopP2P) RegisterProtocol(core.ProtocolID, int, int) {
 func (p *nopP2P) RegisterProtocolServer(rpc.Server) {
 }
 
+// Implements api.Service.
+func (p *nopP2P) AdvertiseProtocol(core.ProtocolID) {
+}
+
 // Implements api.Service.
 func (p *nopP2P) GetMinRepublishInterval() time.Duration {
 	return time.Hour

go/p2p/p2p.go

@@ -345,6 +345,11 @@ func (p *p2p) RegisterProtocolServer(srv rpc.Server) {
 	)
 }
 
+// Implements api.Service.
+func (p *p2p) AdvertiseProtocol(protocol core.ProtocolID) {
+	p.peerMgr.AdvertiseProtocol(protocol)
+}
+
 // Implements api.Service.
 func (p *p2p) GetMinRepublishInterval() time.Duration {
 	return seenMessagesTTL + 5*time.Second

go/p2p/peermgmt/peermgr.go

@@ -188,15 +188,21 @@ func (m *PeerManager) RegisterProtocol(p core.ProtocolID, minPeers int, totalPee
 	}
 
 	m.protocols[p] = &watermark{minPeers, totalPeers}
-	m.discovery.startAdvertising(string(p))
-
 	m.logger.Debug("protocol registered",
 		"protocol", p,
 		"min_peers", minPeers,
 		"total_peers", totalPeers,
 	)
 }
 
+// AdvertiseProtocol starts advertising readiness to serve the specified protocol.
+func (m *PeerManager) AdvertiseProtocol(p core.ProtocolID) {
+	m.discovery.startAdvertising(string(p))
+	m.logger.Debug("triggered protocol advertisement",
+		"protocol", p,
+	)
+}
+
 // RegisterTopic starts tracking and managing peers that support the given topic.
 // If the topic is already registered, its values are updated.
 func (m *PeerManager) RegisterTopic(topic string, minPeers int, totalPeers int) {

go/worker/common/committee/node.go

@@ -956,6 +956,7 @@ func NewNode(
 
 	// Register transaction sync service.
 	p2pHost.RegisterProtocolServer(txsync.NewServer(chainContext, runtime.ID(), n.TxPool))
+	p2pHost.AdvertiseProtocol(txsync.RuntimeProtocolID(chainContext, runtime.ID()))
 
 	return n, nil
 }

go/worker/common/p2p/txsync/server.go

@@ -3,6 +3,8 @@ package txsync
 import (
 	"context"
 
+	"github.com/libp2p/go-libp2p/core"
+
 	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/common/cbor"
 	"github.com/oasisprotocol/oasis-core/go/p2p/protocol"
@@ -50,7 +52,11 @@ func (s *service) handleGetTxs(request *GetTxsRequest) (*GetTxsResponse, error)
 	return &rsp, nil
 }
 
+func RuntimeProtocolID(chainContext string, runtimeID common.Namespace) core.ProtocolID {
+	return protocol.NewRuntimeProtocolID(chainContext, runtimeID, TxSyncProtocolID, TxSyncProtocolVersion)
+}
+
 // NewServer creates a new transaction sync protocol server.
 func NewServer(chainContext string, runtimeID common.Namespace, txPool txpool.TransactionPool) rpc.Server {
-	return rpc.NewServer(protocol.NewRuntimeProtocolID(chainContext, runtimeID, TxSyncProtocolID, TxSyncProtocolVersion), &service{txPool})
+	return rpc.NewServer(RuntimeProtocolID(chainContext, runtimeID), &service{txPool})
 }

go/worker/keymanager/init.go

@@ -126,6 +126,7 @@ func New(
 
 	// Register keymanager service.
 	commonWorker.P2P.RegisterProtocolServer(p2p.NewServer(commonWorker.ChainContext, w.runtimeID, w))
+	commonWorker.P2P.AdvertiseProtocol(p2p.RuntimeProtocolID(commonWorker.ChainContext, w.runtimeID))
 
 	return w, nil
 }

go/worker/keymanager/p2p/server.go

@@ -3,6 +3,7 @@ package p2p
 import (
 	"context"
 
+	"github.com/libp2p/go-libp2p/core"
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/oasisprotocol/oasis-core/go/common"
@@ -48,9 +49,13 @@ func (s *service) handleCallEnclave(ctx context.Context, request *CallEnclaveReq
 	}, nil
 }
 
+// RuntimeProtocolID returns the runtime protocol ID for the specified chain context and runtime ID.
+func RuntimeProtocolID(chainContext string, runtimeID common.Namespace) core.ProtocolID {
+	return protocol.NewRuntimeProtocolID(chainContext, runtimeID, KeyManagerProtocolID, KeyManagerProtocolVersion)
+}
+
 // NewServer creates a new keymanager protocol server.
 func NewServer(chainContext string, runtimeID common.Namespace, km KeyManager) rpc.Server {
 	initMetrics()
-
-	return rpc.NewServer(protocol.NewRuntimeProtocolID(chainContext, runtimeID, KeyManagerProtocolID, KeyManagerProtocolVersion), &service{km})
+	return rpc.NewServer(RuntimeProtocolID(chainContext, runtimeID), &service{km})
 }

go/worker/storage/committee/node.go

@@ -274,11 +274,13 @@ func NewNode(
 
 	// Register storage sync service.
 	commonNode.P2P.RegisterProtocolServer(storageSync.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
+	commonNode.P2P.AdvertiseProtocol(storageSync.RuntimeProtocolID(commonNode.ChainContext, commonNode.Runtime.ID()))
 	n.storageSync = storageSync.NewClient(commonNode.P2P, commonNode.ChainContext, commonNode.Runtime.ID())
 
 	// Register storage pub service if configured.
 	if rpcRoleProvider != nil {
 		commonNode.P2P.RegisterProtocolServer(storagePub.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
+		commonNode.P2P.AdvertiseProtocol(storagePub.RuntimeProtocolID(commonNode.ChainContext, commonNode.Runtime.ID()))
 	}
 
 	return n, nil