hare3: switch to sync validation for gossip messages (#4937)

currently every validation request spawns new goroutine, the main reason for that was
to enable fetching over network without blocking other threads.

however hare3 will be spawning many goroutines and it will be much more efficient to execute protocol inline.
the same is true for beacon.

hare/consensus_test.go

@@ -498,7 +498,7 @@ func (ps *delayedPubSub) Publish(ctx context.Context, protocol string, msg []byt
 	return nil
 }
 
-func (ps *delayedPubSub) Register(protocol string, handler pubsub.GossipHandler) {
+func (ps *delayedPubSub) Register(protocol string, handler pubsub.GossipHandler, opts ...pubsub.ValidatorOpt) {
 	if ps.recvDelay != 0 {
 		handler = func(ctx context.Context, pid p2p.Peer, msg []byte) error {
 			rng := time.Duration(rand.Uint32()) * time.Second % ps.recvDelay
@@ -609,6 +609,6 @@ func (eps *equivocatePubSub) Publish(ctx context.Context, protocol string, data
 	return nil
 }
 
-func (eps *equivocatePubSub) Register(protocol string, handler pubsub.GossipHandler) {
+func (eps *equivocatePubSub) Register(protocol string, handler pubsub.GossipHandler, opts ...pubsub.ValidatorOpt) {
 	eps.ps.Register(protocol, handler)
 }

hare/flows_test.go

@@ -106,7 +106,7 @@ type p2pManipulator struct {
 	err          error
 }
 
-func (m *p2pManipulator) Register(protocol string, handler pubsub.GossipHandler) {
+func (m *p2pManipulator) Register(protocol string, handler pubsub.GossipHandler, opts ...pubsub.ValidatorOpt) {
 	m.nd.Register(protocol, handler)
 }
 

hare3/hare.go

@@ -207,7 +207,7 @@ func (h *Hare) Coins() <-chan WeakCoinOutput {
 }
 
 func (h *Hare) Start() {
-	h.pubsub.Register(h.config.ProtocolName, h.Handler)
+	h.pubsub.Register(h.config.ProtocolName, h.Handler, pubsub.WithValidatorInline(true))
 	current := h.nodeclock.CurrentLayer() + 1
 	enabled := types.MaxLayer(current, h.config.EnableLayer)
 	enabled = types.MaxLayer(enabled, types.GetEffectiveGenesis()+1)

hare3/hare_test.go

@@ -198,7 +198,7 @@ func (n *node) withOracle() *node {
 
 func (n *node) withPublisher() *node {
 	n.mpublisher = pmocks.NewMockPublishSubsciber(n.ctrl)
-	n.mpublisher.EXPECT().Register(gomock.Any(), gomock.Any()).AnyTimes()
+	n.mpublisher.EXPECT().Register(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 	return n
 }
 

p2p/pubsub/pubsub.go

@@ -119,9 +119,13 @@ type Publisher interface {
 
 // Subscriber is an interface for subcribing to messages.
 type Subscriber interface {
-	Register(string, GossipHandler)
+	Register(string, GossipHandler, ...ValidatorOpt)
 }
 
+type ValidatorOpt = pubsub.ValidatorOpt
+
+var WithValidatorInline = pubsub.WithValidatorInline
+
 // PublishSubsciber common interface for publisher and subscribing.
 type PublishSubsciber interface {
 	Publisher

p2p/pubsub/wrapper.go

@@ -26,7 +26,7 @@ type PubSub struct {
 }
 
 // Register handler for topic.
-func (ps *PubSub) Register(topic string, handler GossipHandler) {
+func (ps *PubSub) Register(topic string, handler GossipHandler, opts ...ValidatorOpt) {
 	ps.mu.Lock()
 	defer ps.mu.Unlock()
 	if _, exist := ps.topics[topic]; exist {
@@ -47,7 +47,7 @@ func (ps *PubSub) Register(topic string, handler GossipHandler) {
 		default:
 			return pubsub.ValidationAccept
 		}
-	})
+	}, opts...)
 	topich, err := ps.pubsub.Join(topic)
 	if err != nil {
 		ps.logger.With().Panic("failed to join a topic", log.String("topic", topic), log.Err(err))