[Merged by Bors] - hare3: switch to sync validation for gossip messages

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