go/p2p/protocol: Replace global registry with parametric

Avoid global state to enable writting unit tests with
multiple peers in the same process.

Author: martintomazic

go/p2p/p2p.go

@@ -83,6 +83,8 @@ type p2p struct {
 	registerAddresses []multiaddr.Multiaddr
 	topics            map[string]*topicHandler
 
+	protocolRegistry *protocol.Registry
+
 	logger *logging.Logger
 }
 
@@ -281,7 +283,7 @@ func (p *p2p) Publish(_ context.Context, topic string, msg any) {
 
 // Implements api.Service.
 func (p *p2p) RegisterHandler(topic string, handler api.Handler) {
-	protocol.ValidateTopicID(topic)
+	p.protocolRegistry.ValidateTopicID(topic)
 
 	p.Lock()
 	defer p.Unlock()
@@ -336,7 +338,7 @@ func (p *p2p) PeerManager() api.PeerManager {
 
 // Implements api.Service.
 func (p *p2p) RegisterProtocolServer(srv rpc.Server) {
-	protocol.ValidateProtocolID(srv.Protocol())
+	p.protocolRegistry.ValidateProtocolID(srv.Protocol())
 
 	p.host.SetStreamHandler(srv.Protocol(), srv.HandleStream)
 
@@ -436,6 +438,7 @@ func New(identity *identity.Identity, chainContext string, store *persistent.Com
 		pubsub:            pubsub,
 		registerAddresses: cfg.Addresses,
 		topics:            make(map[string]*topicHandler),
+		protocolRegistry:  protocol.NewRegistry(),
 		logger:            logger,
 	}, nil
 }

go/p2p/protocol/protocol.go

@@ -12,33 +12,32 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/p2p/api"
 )
 
-type protocolRegistry struct {
+// Registry is responsible for ensuring unique protocol ids.
+type Registry struct {
 	mu        sync.Mutex
 	protocols map[core.ProtocolID]struct{}
 }
 
-func newProtocolRegistry() *protocolRegistry {
-	return &protocolRegistry{
+func NewRegistry() *Registry {
+	return &Registry{
 		protocols: make(map[core.ProtocolID]struct{}),
 	}
 }
 
-var registry = newProtocolRegistry()
-
 // ValidateProtocolID panics if the protocol id is not unique.
-func ValidateProtocolID(p core.ProtocolID) {
-	registry.mu.Lock()
-	defer registry.mu.Unlock()
+func (r *Registry) ValidateProtocolID(p core.ProtocolID) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 
-	if _, ok := registry.protocols[p]; ok {
+	if _, ok := r.protocols[p]; ok {
 		panic(fmt.Sprintf("p2p/protocol: protocol or topic with name '%s' already exists", p))
 	}
-	registry.protocols[p] = struct{}{}
+	r.protocols[p] = struct{}{}
 }
 
 // ValidateTopicID panics if the topic id is not unique.
-func ValidateTopicID(topic string) {
-	ValidateProtocolID(core.ProtocolID(topic))
+func (r *Registry) ValidateTopicID(topic string) {
+	r.ValidateProtocolID(core.ProtocolID(topic))
 }
 
 // NewProtocolID generates a protocol identifier for a consensus P2P protocol.

go/p2p/protocol/protocol_test.go

@@ -56,11 +56,10 @@ func TestProtocolID(t *testing.T) {
 		require.Equal(expected, NewTopicIDForRuntime(chainContext, runtimeID, kind, version))
 	})
 
-	registry = newProtocolRegistry()
-
 	t.Run("ValidateProtocolID", func(_ *testing.T) {
-		ValidateProtocolID("protocol-1")
-		ValidateProtocolID("protocol-2")
+		r := NewRegistry()
+		r.ValidateProtocolID("protocol-1")
+		r.ValidateProtocolID("protocol-2")
 	})
 
 	t.Run("ValidateProtocolID panics", func(t *testing.T) {
@@ -69,9 +68,8 @@ func TestProtocolID(t *testing.T) {
 				t.Errorf("validate protocol id should fail")
 			}
 		}()
-		ValidateProtocolID("protocol")
-		ValidateProtocolID("protocol")
+		r := NewRegistry()
+		r.ValidateProtocolID("protocol")
+		r.ValidateProtocolID("protocol")
 	})
-
-	registry = newProtocolRegistry()
 }