go/p2p: Ensure only server providers advertise themselves

Author: martintomazic

.changelog/6270.internal.md

@@ -0,0 +1,7 @@
+go/p2p: Ensure only server providers advertise themselves
+
+Previously, the host would advertise itself upon creation
+of p2p protocol client, even if the server was not running.
+
+Advertisement is now independent and is only triggered
+when serving the P2P protocol.

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.ProtocolID(b.chainContext, b.runtimeID))
+
 	b.storage = storage
 
 	// Wait for activation epoch.

go/oasis-node/cmd/node/node.go

@@ -549,6 +549,7 @@ func NewNode() (node *Node, err error) { // nolint: gocyclo
 
 	// Register consensus light client P2P protocol server.
 	node.P2P.RegisterProtocolServer(consensusLightP2P.NewServer(node.P2P, node.chainContext, node.Consensus.Core()))
+	node.P2P.AdvertiseProtocol(consensusLightP2P.ProtocolID(node.chainContext))
 
 	// Register the consensus service with the peer registry.
 	if mgr := node.P2P.PeerManager(); mgr != nil {

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.ProtocolID(chainContext, runtime.ID()))
 
 	return n, nil
 }

go/worker/common/p2p/txsync/protocol.go

@@ -3,6 +3,7 @@ package txsync
 import (
 	"github.com/libp2p/go-libp2p/core"
 
+	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/node"
 	"github.com/oasisprotocol/oasis-core/go/common/version"
@@ -16,6 +17,11 @@ const TxSyncProtocolID = "txsync"
 // TxSyncProtocolVersion is the supported version of the transaction sync protocol.
 var TxSyncProtocolVersion = version.Version{Major: 2, Minor: 0, Patch: 0}
 
+// ProtocolID returns the runtime transaction sync protocol ID.
+func ProtocolID(chainContext string, runtimeID common.Namespace) core.ProtocolID {
+	return protocol.NewRuntimeProtocolID(chainContext, runtimeID, TxSyncProtocolID, TxSyncProtocolVersion)
+}
+
 // Constants related to the GetTxs method.
 const (
 	MethodGetTxs   = "GetTxs"

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

@@ -5,7 +5,6 @@ import (
 
 	"github.com/oasisprotocol/oasis-core/go/common"
 	"github.com/oasisprotocol/oasis-core/go/common/cbor"
-	"github.com/oasisprotocol/oasis-core/go/p2p/protocol"
 	"github.com/oasisprotocol/oasis-core/go/p2p/rpc"
 	"github.com/oasisprotocol/oasis-core/go/runtime/txpool"
 )
@@ -52,5 +51,5 @@ func (s *service) handleGetTxs(request *GetTxsRequest) (*GetTxsResponse, error)
 
 // 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(ProtocolID(chainContext, runtimeID), &service{txPool})
 }