p2p: add a per-message type send and receive metric (backport #9622) (#9641)

* p2p: add a per-message type send and receive metric (#9622)

* p2p: ressurrect the p2p envelope and use to calculate message metric

Add new SendEnvelope, TrySendEnvelope, BroadcastEnvelope, and ReceiveEnvelope methods in the p2p package to work with the new envelope type.

Care was taken to ensure this was performed in a non-breaking manner.

Co-authored-by: William Banfield <4561443+williambanfield@users.noreply.github.com>
Co-authored-by: William Banfield <wbanfield@gmail.com>

p2p/base_reactor.go

@@ -44,9 +44,25 @@ type Reactor interface {
 	// copying.
 	//
 	// CONTRACT: msgBytes are not nil.
+	//
+	// Only one of Receive or ReceiveEnvelope are called per message. If ReceiveEnvelope
+	// is implemented, it will be used, otherwise the switch will fallback to
+	// using Receive.
+	// Deprecated: Reactors looking to receive data from a peer should implement ReceiveEnvelope.
+	// Receive will be deprecated in favor of ReceiveEnvelope in v0.37.
 	Receive(chID byte, peer Peer, msgBytes []byte)
 }
 
+type EnvelopeReceiver interface {
+	// ReceiveEnvelope is called by the switch when an envelope is received from any connected
+	// peer on any of the channels registered by the reactor.
+	//
+	// Only one of Receive or ReceiveEnvelope are called per message. If ReceiveEnvelope
+	// is implemented, it will be used, otherwise the switch will fallback to
+	// using Receive. Receive will be replaced by ReceiveEnvelope in a future version
+	ReceiveEnvelope(Envelope)
+}
+
 //--------------------------------------
 
 type BaseReactor struct {
@@ -67,5 +83,6 @@ func (br *BaseReactor) SetSwitch(sw *Switch) {
 func (*BaseReactor) GetChannels() []*conn.ChannelDescriptor        { return nil }
 func (*BaseReactor) AddPeer(peer Peer)                             {}
 func (*BaseReactor) RemovePeer(peer Peer, reason interface{})      {}
+func (*BaseReactor) ReceiveEnvelope(e Envelope)                    {}
 func (*BaseReactor) Receive(chID byte, peer Peer, msgBytes []byte) {}
 func (*BaseReactor) InitPeer(peer Peer) Peer                       { return peer }

p2p/conn/connection.go

@@ -724,6 +724,7 @@ type ChannelDescriptor struct {
 	SendQueueCapacity   int
 	RecvBufferCapacity  int
 	RecvMessageCapacity int
+	MessageType         proto.Message
 }
 
 func (chDesc ChannelDescriptor) FillDefaults() (filled ChannelDescriptor) {

p2p/metrics.go

@@ -1,6 +1,11 @@
 package p2p
 
 import (
+	"fmt"
+	"reflect"
+	"regexp"
+	"sync"
+
 	"github.com/go-kit/kit/metrics"
 	"github.com/go-kit/kit/metrics/discard"
 	"github.com/go-kit/kit/metrics/prometheus"
@@ -13,6 +18,13 @@ const (
 	MetricsSubsystem = "p2p"
 )
 
+var (
+	// valueToLabelRegexp is used to find the golang package name and type name
+	// so that the name can be turned into a prometheus label where the characters
+	// in the label do not include prometheus special characters such as '*' and '.'.
+	valueToLabelRegexp = regexp.MustCompile(`\*?(\w+)\.(.*)`)
+)
+
 // Metrics contains metrics exposed by this package.
 type Metrics struct {
 	// Number of peers.
@@ -25,6 +37,10 @@ type Metrics struct {
 	PeerPendingSendBytes metrics.Gauge
 	// Number of transactions submitted by each peer.
 	NumTxs metrics.Gauge
+	// Number of bytes of each message type received.
+	MessageReceiveBytesTotal metrics.Counter
+	// Number of bytes of each message type sent.
+	MessageSendBytesTotal metrics.Counter
 }
 
 // PrometheusMetrics returns Metrics build using Prometheus client library.
@@ -58,7 +74,7 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Namespace: namespace,
 			Subsystem: MetricsSubsystem,
 			Name:      "peer_pending_send_bytes",
-			Help:      "Number of pending bytes to be sent to a given peer.",
+			Help:      "Pending bytes to be sent to a given peer.",
 		}, append(labels, "peer_id")).With(labelsAndValues...),
 		NumTxs: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
 			Namespace: namespace,
@@ -66,16 +82,64 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Name:      "num_txs",
 			Help:      "Number of transactions submitted by each peer.",
 		}, append(labels, "peer_id")).With(labelsAndValues...),
+		MessageReceiveBytesTotal: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "message_receive_bytes_total",
+			Help:      "Number of bytes of each message type received.",
+		}, append(labels, "message_type")).With(labelsAndValues...),
+		MessageSendBytesTotal: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "message_send_bytes_total",
+			Help:      "Number of bytes of each message type sent.",
+		}, append(labels, "message_type")).With(labelsAndValues...),
 	}
 }
 
-// NopMetrics returns no-op Metrics.
 func NopMetrics() *Metrics {
 	return &Metrics{
-		Peers:                 discard.NewGauge(),
-		PeerReceiveBytesTotal: discard.NewCounter(),
-		PeerSendBytesTotal:    discard.NewCounter(),
-		PeerPendingSendBytes:  discard.NewGauge(),
-		NumTxs:                discard.NewGauge(),
+		Peers:                    discard.NewGauge(),
+		PeerReceiveBytesTotal:    discard.NewCounter(),
+		PeerSendBytesTotal:       discard.NewCounter(),
+		PeerPendingSendBytes:     discard.NewGauge(),
+		NumTxs:                   discard.NewGauge(),
+		MessageReceiveBytesTotal: discard.NewCounter(),
+		MessageSendBytesTotal:    discard.NewCounter(),
+	}
+}
+
+type metricsLabelCache struct {
+	mtx               *sync.RWMutex
+	messageLabelNames map[reflect.Type]string
+}
+
+// ValueToMetricLabel is a method that is used to produce a prometheus label value of the golang
+// type that is passed in.
+// This method uses a map on the Metrics struct so that each label name only needs
+// to be produced once to prevent expensive string operations.
+func (m *metricsLabelCache) ValueToMetricLabel(i interface{}) string {
+	t := reflect.TypeOf(i)
+	m.mtx.RLock()
+
+	if s, ok := m.messageLabelNames[t]; ok {
+		m.mtx.RUnlock()
+		return s
+	}
+	m.mtx.RUnlock()
+
+	s := t.String()
+	ss := valueToLabelRegexp.FindStringSubmatch(s)
+	l := fmt.Sprintf("%s_%s", ss[1], ss[2])
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
+	m.messageLabelNames[t] = l
+	return l
+}
+
+func newMetricsLabelCache() *metricsLabelCache {
+	return &metricsLabelCache{
+		mtx:               &sync.RWMutex{},
+		messageLabelNames: map[reflect.Type]string{},
 	}
 }

p2p/mock/peer.go

@@ -42,9 +42,11 @@ func NewPeer(ip net.IP) *Peer {
 	return mp
 }
 
-func (mp *Peer) FlushStop()                              { mp.Stop() } //nolint:errcheck //ignore error
-func (mp *Peer) TrySend(chID byte, msgBytes []byte) bool { return true }
-func (mp *Peer) Send(chID byte, msgBytes []byte) bool    { return true }
+func (mp *Peer) FlushStop()                          { mp.Stop() } //nolint:errcheck //ignore error
+func (mp *Peer) TrySendEnvelope(e p2p.Envelope) bool { return true }
+func (mp *Peer) SendEnvelope(e p2p.Envelope) bool    { return true }
+func (mp *Peer) TrySend(_ byte, _ []byte) bool       { return true }
+func (mp *Peer) Send(_ byte, _ []byte) bool          { return true }
 func (mp *Peer) NodeInfo() p2p.NodeInfo {
 	return p2p.DefaultNodeInfo{
 		DefaultNodeID: mp.addr.ID,

p2p/mock/reactor.go

@@ -22,4 +22,5 @@ func NewReactor() *Reactor {
 func (r *Reactor) GetChannels() []*conn.ChannelDescriptor            { return r.Channels }
 func (r *Reactor) AddPeer(peer p2p.Peer)                             {}
 func (r *Reactor) RemovePeer(peer p2p.Peer, reason interface{})      {}
+func (r *Reactor) ReceiveEnvelope(e p2p.Envelope)                    {}
 func (r *Reactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {}

p2p/mocks/peer.go

@@ -123,6 +123,34 @@ func (_m *Peer) IsRunning() bool {
 	return r0
 }
 
+// SendEnvelope provides a mock function with given fields: _a0
+func (_m *Peer) SendEnvelope(_a0 p2p.Envelope) bool {
+	ret := _m.Called(_a0)
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(p2p.Envelope) bool); ok {
+		r0 = rf(_a0)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
+// TrySendEnvelope provides a mock function with given fields: _a0
+func (_m *Peer) TrySendEnvelope(_a0 p2p.Envelope) bool {
+	ret := _m.Called(_a0)
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(p2p.Envelope) bool); ok {
+		r0 = rf(_a0)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
 // NodeInfo provides a mock function with given fields:
 func (_m *Peer) NodeInfo() p2p.NodeInfo {
 	ret := _m.Called()

p2p/peer.go

@@ -3,8 +3,11 @@ package p2p
 import (
 	"fmt"
 	"net"
+	"reflect"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
+
 	"github.com/tendermint/tendermint/libs/cmap"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/service"
@@ -34,7 +37,12 @@ type Peer interface {
 	Status() tmconn.ConnectionStatus
 	SocketAddr() *NetAddress // actual address of the socket
 
+	// Deprecated: entities looking to act as peers should implement SendEnvelope instead.
+	// Send will be removed in v0.37.
 	Send(byte, []byte) bool
+
+	// Deprecated: entities looking to act as peers should implement TrySendEnvelope instead.
+	// TrySend will be removed in v0.37.
 	TrySend(byte, []byte) bool
 
 	Set(string, interface{})
@@ -44,6 +52,55 @@ type Peer interface {
 	GetRemovalFailed() bool
 }
 
+type EnvelopeSender interface {
+	SendEnvelope(Envelope) bool
+	TrySendEnvelope(Envelope) bool
+}
+
+// EnvelopeSendShim implements a shim to allow the legacy peer type that
+// does not implement SendEnvelope to be used in places where envelopes are
+// being sent. If the peer implements the *Envelope methods, then they are used,
+// otherwise, the message is marshaled and dispatched to the legacy *Send.
+//
+// Deprecated: Will be removed in v0.37.
+func SendEnvelopeShim(p Peer, e Envelope, lg log.Logger) bool {
+	if es, ok := p.(EnvelopeSender); ok {
+		return es.SendEnvelope(e)
+	}
+	msg := e.Message
+	if w, ok := msg.(Wrapper); ok {
+		msg = w.Wrap()
+	}
+	msgBytes, err := proto.Marshal(msg)
+	if err != nil {
+		lg.Error("marshaling message to send", "error", err)
+		return false
+	}
+	return p.Send(e.ChannelID, msgBytes)
+}
+
+// EnvelopeTrySendShim implements a shim to allow the legacy peer type that
+// does not implement TrySendEnvelope to be used in places where envelopes are
+// being sent. If the peer implements the *Envelope methods, then they are used,
+// otherwise, the message is marshaled and dispatched to the legacy *Send.
+//
+// Deprecated: Will be removed in v0.37.
+func TrySendEnvelopeShim(p Peer, e Envelope, lg log.Logger) bool {
+	if es, ok := p.(EnvelopeSender); ok {
+		return es.SendEnvelope(e)
+	}
+	msg := e.Message
+	if w, ok := msg.(Wrapper); ok {
+		msg = w.Wrap()
+	}
+	msgBytes, err := proto.Marshal(msg)
+	if err != nil {
+		lg.Error("marshaling message to send", "error", err)
+		return false
+	}
+	return p.TrySend(e.ChannelID, msgBytes)
+}
+
 //----------------------------------------------------------
 
 // peerConn contains the raw connection and its config.
@@ -120,6 +177,7 @@ type peer struct {
 
 	metrics       *Metrics
 	metricsTicker *time.Ticker
+	mlc           *metricsLabelCache
 
 	// When removal of a peer fails, we set this flag
 	removalAttemptFailed bool
@@ -132,8 +190,10 @@ func newPeer(
 	mConfig tmconn.MConnConfig,
 	nodeInfo NodeInfo,
 	reactorsByCh map[byte]Reactor,
+	msgTypeByChID map[byte]proto.Message,
 	chDescs []*tmconn.ChannelDescriptor,
 	onPeerError func(Peer, interface{}),
+	mlc *metricsLabelCache,
 	options ...PeerOption,
 ) *peer {
 	p := &peer{
@@ -143,12 +203,14 @@ func newPeer(
 		Data:          cmap.NewCMap(),
 		metricsTicker: time.NewTicker(metricsTickerDuration),
 		metrics:       NopMetrics(),
+		mlc:           mlc,
 	}
 
 	p.mconn = createMConnection(
 		pc.conn,
 		p,
 		reactorsByCh,
+		msgTypeByChID,
 		chDescs,
 		onPeerError,
 		mConfig,
@@ -194,7 +256,7 @@ func (p *peer) OnStart() error {
 }
 
 // FlushStop mimics OnStop but additionally ensures that all successful
-// .Send() calls will get flushed before closing the connection.
+// SendEnvelope() calls will get flushed before closing the connection.
 // NOTE: it is not safe to call this method more than once.
 func (p *peer) FlushStop() {
 	p.metricsTicker.Stop()
@@ -247,12 +309,39 @@ func (p *peer) Status() tmconn.ConnectionStatus {
 	return p.mconn.Status()
 }
 
+// SendEnvelope sends the message in the envelope on the channel specified by the
+// envelope. Returns false if the connection times out trying to place the message
+// onto its internal queue.
+// Using SendEnvelope allows for tracking the message bytes sent and received by message type
+// as a metric which Send cannot support.
+func (p *peer) SendEnvelope(e Envelope) bool {
+	if !p.IsRunning() {
+		return false
+	} else if !p.hasChannel(e.ChannelID) {
+		return false
+	}
+	msg := e.Message
+	metricLabelValue := p.mlc.ValueToMetricLabel(msg)
+	if w, ok := msg.(Wrapper); ok {
+		msg = w.Wrap()
+	}
+	msgBytes, err := proto.Marshal(msg)
+	if err != nil {
+		p.Logger.Error("marshaling message to send", "error", err)
+		return false
+	}
+	res := p.Send(e.ChannelID, msgBytes)
+	if res {
+		p.metrics.MessageSendBytesTotal.With("message_type", metricLabelValue).Add(float64(len(msgBytes)))
+	}
+	return res
+}
+
 // Send msg bytes to the channel identified by chID byte. Returns false if the
 // send queue is full after timeout, specified by MConnection.
+// SendEnvelope replaces Send which will be deprecated in a future release.
 func (p *peer) Send(chID byte, msgBytes []byte) bool {
 	if !p.IsRunning() {
-		// see Switch#Broadcast, where we fetch the list of peers and loop over
-		// them - while we're looping, one peer may be removed and stopped.
 		return false
 	} else if !p.hasChannel(chID) {
 		return false
@@ -268,8 +357,38 @@ func (p *peer) Send(chID byte, msgBytes []byte) bool {
 	return res
 }
 
+// TrySendEnvelope attempts to sends the message in the envelope on the channel specified by the
+// envelope. Returns false immediately if the connection's internal queue is full
+// Using TrySendEnvelope allows for tracking the message bytes sent and received by message type
+// as a metric which TrySend cannot support.
+func (p *peer) TrySendEnvelope(e Envelope) bool {
+	if !p.IsRunning() {
+		// see Switch#Broadcast, where we fetch the list of peers and loop over
+		// them - while we're looping, one peer may be removed and stopped.
+		return false
+	} else if !p.hasChannel(e.ChannelID) {
+		return false
+	}
+	msg := e.Message
+	metricLabelValue := p.mlc.ValueToMetricLabel(msg)
+	if w, ok := msg.(Wrapper); ok {
+		msg = w.Wrap()
+	}
+	msgBytes, err := proto.Marshal(msg)
+	if err != nil {
+		p.Logger.Error("marshaling message to send", "error", err)
+		return false
+	}
+	res := p.TrySend(e.ChannelID, msgBytes)
+	if res {
+		p.metrics.MessageSendBytesTotal.With("message_type", metricLabelValue).Add(float64(len(msgBytes)))
+	}
+	return res
+}
+
 // TrySend msg bytes to the channel identified by chID byte. Immediately returns
 // false if the send queue is full.
+// TrySendEnvelope replaces TrySend which will be deprecated in a future release.
 func (p *peer) TrySend(chID byte, msgBytes []byte) bool {
 	if !p.IsRunning() {
 		return false
@@ -384,6 +503,7 @@ func createMConnection(
 	conn net.Conn,
 	p *peer,
 	reactorsByCh map[byte]Reactor,
+	msgTypeByChID map[byte]proto.Message,
 	chDescs []*tmconn.ChannelDescriptor,
 	onPeerError func(Peer, interface{}),
 	config tmconn.MConnConfig,
@@ -396,12 +516,33 @@ func createMConnection(
 			// which does onPeerError.
 			panic(fmt.Sprintf("Unknown channel %X", chID))
 		}
+		mt := msgTypeByChID[chID]
+		msg := proto.Clone(mt)
+		err := proto.Unmarshal(msgBytes, msg)
+		if err != nil {
+			panic(fmt.Errorf("unmarshaling message: %s into type: %s", err, reflect.TypeOf(mt)))
+		}
 		labels := []string{
 			"peer_id", string(p.ID()),
 			"chID", fmt.Sprintf("%#x", chID),
 		}
+		if w, ok := msg.(Unwrapper); ok {
+			msg, err = w.Unwrap()
+			if err != nil {
+				panic(fmt.Errorf("unwrapping message: %s", err))
+			}
+		}
 		p.metrics.PeerReceiveBytesTotal.With(labels...).Add(float64(len(msgBytes)))
-		reactor.Receive(chID, p, msgBytes)
+		p.metrics.MessageReceiveBytesTotal.With("message_type", p.mlc.ValueToMetricLabel(msg)).Add(float64(len(msgBytes)))
+		if nr, ok := reactor.(EnvelopeReceiver); ok {
+			nr.ReceiveEnvelope(Envelope{
+				ChannelID: chID,
+				Src:       p,
+				Message:   msg,
+			})
+		} else {
+			reactor.Receive(chID, p, msgBytes)
+		}
 	}
 
 	onError := func(r interface{}) {

p2p/peer_set_test.go

@@ -18,22 +18,24 @@ type mockPeer struct {
 	id ID
 }
 
-func (mp *mockPeer) FlushStop()                              { mp.Stop() } //nolint:errcheck // ignore error
-func (mp *mockPeer) TrySend(chID byte, msgBytes []byte) bool { return true }
-func (mp *mockPeer) Send(chID byte, msgBytes []byte) bool    { return true }
-func (mp *mockPeer) NodeInfo() NodeInfo                      { return DefaultNodeInfo{} }
-func (mp *mockPeer) Status() ConnectionStatus                { return ConnectionStatus{} }
-func (mp *mockPeer) ID() ID                                  { return mp.id }
-func (mp *mockPeer) IsOutbound() bool                        { return false }
-func (mp *mockPeer) IsPersistent() bool                      { return true }
-func (mp *mockPeer) Get(s string) interface{}                { return s }
-func (mp *mockPeer) Set(string, interface{})                 {}
-func (mp *mockPeer) RemoteIP() net.IP                        { return mp.ip }
-func (mp *mockPeer) SocketAddr() *NetAddress                 { return nil }
-func (mp *mockPeer) RemoteAddr() net.Addr                    { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
-func (mp *mockPeer) CloseConn() error                        { return nil }
-func (mp *mockPeer) SetRemovalFailed()                       {}
-func (mp *mockPeer) GetRemovalFailed() bool                  { return false }
+func (mp *mockPeer) FlushStop()                      { mp.Stop() } //nolint:errcheck // ignore error
+func (mp *mockPeer) TrySendEnvelope(e Envelope) bool { return true }
+func (mp *mockPeer) SendEnvelope(e Envelope) bool    { return true }
+func (mp *mockPeer) TrySend(_ byte, _ []byte) bool   { return true }
+func (mp *mockPeer) Send(_ byte, _ []byte) bool      { return true }
+func (mp *mockPeer) NodeInfo() NodeInfo              { return DefaultNodeInfo{} }
+func (mp *mockPeer) Status() ConnectionStatus        { return ConnectionStatus{} }
+func (mp *mockPeer) ID() ID                          { return mp.id }
+func (mp *mockPeer) IsOutbound() bool                { return false }
+func (mp *mockPeer) IsPersistent() bool              { return true }
+func (mp *mockPeer) Get(s string) interface{}        { return s }
+func (mp *mockPeer) Set(string, interface{})         {}
+func (mp *mockPeer) RemoteIP() net.IP                { return mp.ip }
+func (mp *mockPeer) SocketAddr() *NetAddress         { return nil }
+func (mp *mockPeer) RemoteAddr() net.Addr            { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
+func (mp *mockPeer) CloseConn() error                { return nil }
+func (mp *mockPeer) SetRemovalFailed()               {}
+func (mp *mockPeer) GetRemovalFailed() bool          { return false }
 
 // Returns a mock peer
 func newMockPeer(ip net.IP) *mockPeer {

p2p/peer_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -14,6 +15,7 @@ import (
 	"github.com/tendermint/tendermint/crypto/ed25519"
 	"github.com/tendermint/tendermint/libs/bytes"
 	"github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/proto/tendermint/p2p"
 
 	"github.com/tendermint/tendermint/config"
 	tmconn "github.com/tendermint/tendermint/p2p/conn"
@@ -70,7 +72,7 @@ func TestPeerSend(t *testing.T) {
 	})
 
 	assert.True(p.CanSend(testCh))
-	assert.True(p.Send(testCh, []byte("Asylum")))
+	assert.True(SendEnvelopeShim(p, Envelope{ChannelID: testCh, Message: &p2p.Message{}}, p.Logger))
 }
 
 func createOutboundPeerAndPerformHandshake(
@@ -82,6 +84,9 @@ func createOutboundPeerAndPerformHandshake(
 		{ID: testCh, Priority: 1},
 	}
 	reactorsByCh := map[byte]Reactor{testCh: NewTestReactor(chDescs, true)}
+	msgTypeByChID := map[byte]proto.Message{
+		testCh: &p2p.Message{},
+	}
 	pk := ed25519.GenPrivKey()
 	pc, err := testOutboundPeerConn(addr, config, false, pk)
 	if err != nil {
@@ -94,7 +99,7 @@ func createOutboundPeerAndPerformHandshake(
 		return nil, err
 	}
 
-	p := newPeer(pc, mConfig, peerNodeInfo, reactorsByCh, chDescs, func(p Peer, r interface{}) {})
+	p := newPeer(pc, mConfig, peerNodeInfo, reactorsByCh, msgTypeByChID, chDescs, func(p Peer, r interface{}) {}, newMetricsLabelCache())
 	p.SetLogger(log.TestingLogger().With("peer", addr))
 	return p, nil
 }

p2p/pex/pex_reactor.go

@@ -184,6 +184,7 @@ func (r *Reactor) GetChannels() []*conn.ChannelDescriptor {
 			Priority:            1,
 			SendQueueCapacity:   10,
 			RecvMessageCapacity: maxMsgSize,
+			MessageType:         &tmp2p.Message{},
 		},
 	}
 }
@@ -236,16 +237,10 @@ func (r *Reactor) logErrAddrBook(err error) {
 }
 
 // Receive implements Reactor by handling incoming PEX messages.
-func (r *Reactor) Receive(chID byte, src Peer, msgBytes []byte) {
-	msg, err := decodeMsg(msgBytes)
-	if err != nil {
-		r.Logger.Error("Error decoding message", "src", src, "chId", chID, "err", err)
-		r.Switch.StopPeerForError(src, err)
-		return
-	}
-	r.Logger.Debug("Received message", "src", src, "chId", chID, "msg", msg)
+func (r *Reactor) ReceiveEnvelope(e p2p.Envelope) {
+	r.Logger.Debug("Received message", "src", e.Src, "chId", e.ChannelID, "msg", e.Message)
 
-	switch msg := msg.(type) {
+	switch msg := e.Message.(type) {
 	case *tmp2p.PexRequest:
 
 		// NOTE: this is a prime candidate for amplification attacks,
@@ -255,8 +250,8 @@ func (r *Reactor) Receive(chID byte, src Peer, msgBytes []byte) {
 
 		// If we're a seed and this is an inbound peer,
 		// respond once and disconnect.
-		if r.config.SeedMode && !src.IsOutbound() {
-			id := string(src.ID())
+		if r.config.SeedMode && !e.Src.IsOutbound() {
+			id := string(e.Src.ID())
 			v := r.lastReceivedRequests.Get(id)
 			if v != nil {
 				// FlushStop/StopPeer are already
@@ -266,36 +261,36 @@ func (r *Reactor) Receive(chID byte, src Peer, msgBytes []byte) {
 			r.lastReceivedRequests.Set(id, time.Now())
 
 			// Send addrs and disconnect
-			r.SendAddrs(src, r.book.GetSelectionWithBias(biasToSelectNewPeers))
+			r.SendAddrs(e.Src, r.book.GetSelectionWithBias(biasToSelectNewPeers))
 			go func() {
 				// In a go-routine so it doesn't block .Receive.
-				src.FlushStop()
-				r.Switch.StopPeerGracefully(src)
+				e.Src.FlushStop()
+				r.Switch.StopPeerGracefully(e.Src)
 			}()
 
 		} else {
 			// Check we're not receiving requests too frequently.
-			if err := r.receiveRequest(src); err != nil {
-				r.Switch.StopPeerForError(src, err)
-				r.book.MarkBad(src.SocketAddr(), defaultBanTime)
+			if err := r.receiveRequest(e.Src); err != nil {
+				r.Switch.StopPeerForError(e.Src, err)
+				r.book.MarkBad(e.Src.SocketAddr(), defaultBanTime)
 				return
 			}
-			r.SendAddrs(src, r.book.GetSelection())
+			r.SendAddrs(e.Src, r.book.GetSelection())
 		}
 
 	case *tmp2p.PexAddrs:
 		// If we asked for addresses, add them to the book
 		addrs, err := p2p.NetAddressesFromProto(msg.Addrs)
 		if err != nil {
-			r.Switch.StopPeerForError(src, err)
-			r.book.MarkBad(src.SocketAddr(), defaultBanTime)
+			r.Switch.StopPeerForError(e.Src, err)
+			r.book.MarkBad(e.Src.SocketAddr(), defaultBanTime)
 			return
 		}
-		err = r.ReceiveAddrs(addrs, src)
+		err = r.ReceiveAddrs(addrs, e.Src)
 		if err != nil {
-			r.Switch.StopPeerForError(src, err)
+			r.Switch.StopPeerForError(e.Src, err)
 			if err == ErrUnsolicitedList {
-				r.book.MarkBad(src.SocketAddr(), defaultBanTime)
+				r.book.MarkBad(e.Src.SocketAddr(), defaultBanTime)
 			}
 			return
 		}
@@ -305,6 +300,23 @@ func (r *Reactor) Receive(chID byte, src Peer, msgBytes []byte) {
 	}
 }
 
+func (r *Reactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {
+	var msg *tmp2p.Message
+	err := proto.Unmarshal(msgBytes, msg)
+	if err != nil {
+		panic(err)
+	}
+	um, err := msg.Unwrap()
+	if err != nil {
+		panic(err)
+	}
+	r.ReceiveEnvelope(p2p.Envelope{
+		ChannelID: chID,
+		Src:       peer,
+		Message:   um,
+	})
+}
+
 // enforces a minimum amount of time between requests
 func (r *Reactor) receiveRequest(src Peer) error {
 	id := string(src.ID())
@@ -348,7 +360,10 @@ func (r *Reactor) RequestAddrs(p Peer) {
 	}
 	r.Logger.Debug("Request addrs", "from", p)
 	r.requestsSent.Set(id, struct{}{})
-	p.Send(PexChannel, mustEncode(&tmp2p.PexRequest{}))
+	p2p.SendEnvelopeShim(p, p2p.Envelope{ //nolint: staticcheck
+		ChannelID: PexChannel,
+		Message:   &tmp2p.PexRequest{},
+	}, r.Logger)
 }
 
 // ReceiveAddrs adds the given addrs to the addrbook if theres an open
@@ -406,7 +421,11 @@ func (r *Reactor) ReceiveAddrs(addrs []*p2p.NetAddress, src Peer) error {
 
 // SendAddrs sends addrs to the peer.
 func (r *Reactor) SendAddrs(p Peer, netAddrs []*p2p.NetAddress) {
-	p.Send(PexChannel, mustEncode(&tmp2p.PexAddrs{Addrs: p2p.NetAddressesToProto(netAddrs)}))
+	e := p2p.Envelope{
+		ChannelID: PexChannel,
+		Message:   &tmp2p.PexAddrs{Addrs: p2p.NetAddressesToProto(netAddrs)},
+	}
+	p2p.SendEnvelopeShim(p, e, r.Logger) //nolint: staticcheck
 }
 
 // SetEnsurePeersPeriod sets period to ensure peers connected.
@@ -763,43 +782,3 @@ func markAddrInBookBasedOnErr(addr *p2p.NetAddress, book AddrBook, err error) {
 		book.MarkAttempt(addr)
 	}
 }
-
-//-----------------------------------------------------------------------------
-// Messages
-
-// mustEncode proto encodes a tmp2p.Message
-func mustEncode(pb proto.Message) []byte {
-	msg := tmp2p.Message{}
-	switch pb := pb.(type) {
-	case *tmp2p.PexRequest:
-		msg.Sum = &tmp2p.Message_PexRequest{PexRequest: pb}
-	case *tmp2p.PexAddrs:
-		msg.Sum = &tmp2p.Message_PexAddrs{PexAddrs: pb}
-	default:
-		panic(fmt.Sprintf("Unknown message type %T", pb))
-	}
-
-	bz, err := msg.Marshal()
-	if err != nil {
-		panic(fmt.Errorf("unable to marshal %T: %w", pb, err))
-	}
-	return bz
-}
-
-func decodeMsg(bz []byte) (proto.Message, error) {
-	pb := &tmp2p.Message{}
-
-	err := pb.Unmarshal(bz)
-	if err != nil {
-		return nil, err
-	}
-
-	switch msg := pb.Sum.(type) {
-	case *tmp2p.Message_PexRequest:
-		return msg.PexRequest, nil
-	case *tmp2p.Message_PexAddrs:
-		return msg.PexAddrs, nil
-	default:
-		return nil, fmt.Errorf("unknown message: %T", msg)
-	}
-}

p2p/pex/pex_reactor_test.go

@@ -129,12 +129,11 @@ func TestPEXReactorReceive(t *testing.T) {
 	r.RequestAddrs(peer)
 
 	size := book.Size()
-	msg := mustEncode(&tmp2p.PexAddrs{Addrs: []tmp2p.NetAddress{peer.SocketAddr().ToProto()}})
-	r.Receive(PexChannel, peer, msg)
+	msg := &tmp2p.PexAddrs{Addrs: []tmp2p.NetAddress{peer.SocketAddr().ToProto()}}
+	r.ReceiveEnvelope(p2p.Envelope{ChannelID: PexChannel, Src: peer, Message: msg})
 	assert.Equal(t, size+1, book.Size())
 
-	msg = mustEncode(&tmp2p.PexRequest{})
-	r.Receive(PexChannel, peer, msg) // should not panic.
+	r.ReceiveEnvelope(p2p.Envelope{ChannelID: PexChannel, Src: peer, Message: &tmp2p.PexRequest{}})
 }
 
 func TestPEXReactorRequestMessageAbuse(t *testing.T) {
@@ -153,20 +152,19 @@ func TestPEXReactorRequestMessageAbuse(t *testing.T) {
 	require.True(t, book.HasAddress(peerAddr))
 
 	id := string(peer.ID())
-	msg := mustEncode(&tmp2p.PexRequest{})
 
 	// first time creates the entry
-	r.Receive(PexChannel, peer, msg)
+	r.ReceiveEnvelope(p2p.Envelope{ChannelID: PexChannel, Src: peer, Message: &tmp2p.PexRequest{}})
 	assert.True(t, r.lastReceivedRequests.Has(id))
 	assert.True(t, sw.Peers().Has(peer.ID()))
 
 	// next time sets the last time value
-	r.Receive(PexChannel, peer, msg)
+	r.ReceiveEnvelope(p2p.Envelope{ChannelID: PexChannel, Src: peer, Message: &tmp2p.PexRequest{}})
 	assert.True(t, r.lastReceivedRequests.Has(id))
 	assert.True(t, sw.Peers().Has(peer.ID()))
 
 	// third time is too many too soon - peer is removed
-	r.Receive(PexChannel, peer, msg)
+	r.ReceiveEnvelope(p2p.Envelope{ChannelID: PexChannel, Src: peer, Message: &tmp2p.PexRequest{}})
 	assert.False(t, r.lastReceivedRequests.Has(id))
 	assert.False(t, sw.Peers().Has(peer.ID()))
 	assert.True(t, book.IsBanned(peerAddr))
@@ -190,15 +188,15 @@ func TestPEXReactorAddrsMessageAbuse(t *testing.T) {
 	assert.True(t, r.requestsSent.Has(id))
 	assert.True(t, sw.Peers().Has(peer.ID()))
 
-	msg := mustEncode(&tmp2p.PexAddrs{Addrs: []tmp2p.NetAddress{peer.SocketAddr().ToProto()}})
+	msg := &tmp2p.PexAddrs{Addrs: []tmp2p.NetAddress{peer.SocketAddr().ToProto()}}
 
 	// receive some addrs. should clear the request
-	r.Receive(PexChannel, peer, msg)
+	r.ReceiveEnvelope(p2p.Envelope{ChannelID: PexChannel, Src: peer, Message: msg})
 	assert.False(t, r.requestsSent.Has(id))
 	assert.True(t, sw.Peers().Has(peer.ID()))
 
 	// receiving more unsolicited addrs causes a disconnect and ban
-	r.Receive(PexChannel, peer, msg)
+	r.ReceiveEnvelope(p2p.Envelope{ChannelID: PexChannel, Src: peer, Message: msg})
 	assert.False(t, sw.Peers().Has(peer.ID()))
 	assert.True(t, book.IsBanned(peer.SocketAddr()))
 }
@@ -488,8 +486,12 @@ func TestPEXReactorDoesNotAddPrivatePeersToAddrBook(t *testing.T) {
 	pexR.RequestAddrs(peer)
 
 	size := book.Size()
-	msg := mustEncode(&tmp2p.PexAddrs{Addrs: []tmp2p.NetAddress{peer.SocketAddr().ToProto()}})
-	pexR.Receive(PexChannel, peer, msg)
+	msg := &tmp2p.PexAddrs{Addrs: []tmp2p.NetAddress{peer.SocketAddr().ToProto()}}
+	pexR.ReceiveEnvelope(p2p.Envelope{
+		ChannelID: PexChannel,
+		Src:       peer,
+		Message:   msg,
+	})
 	assert.Equal(t, size, book.Size())
 
 	pexR.AddPeer(peer)
@@ -696,7 +698,9 @@ func TestPexVectors(t *testing.T) {
 	for _, tc := range testCases {
 		tc := tc
 
-		bz := mustEncode(tc.msg)
+		w := tc.msg.(p2p.Wrapper).Wrap()
+		bz, err := proto.Marshal(w)
+		require.NoError(t, err)
 
 		require.Equal(t, tc.expBytes, hex.EncodeToString(bz), tc.testName)
 	}

p2p/switch.go

@@ -6,9 +6,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/libs/cmap"
-	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/rand"
 	"github.com/tendermint/tendermint/libs/service"
 	"github.com/tendermint/tendermint/p2p/conn"
@@ -69,16 +69,17 @@ type PeerFilterFunc func(IPeerSet, Peer) error
 type Switch struct {
 	service.BaseService
 
-	config       *config.P2PConfig
-	reactors     map[string]Reactor
-	chDescs      []*conn.ChannelDescriptor
-	reactorsByCh map[byte]Reactor
-	peers        *PeerSet
-	dialing      *cmap.CMap
-	reconnecting *cmap.CMap
-	nodeInfo     NodeInfo // our node info
-	nodeKey      *NodeKey // our node privkey
-	addrBook     AddrBook
+	config        *config.P2PConfig
+	reactors      map[string]Reactor
+	chDescs       []*conn.ChannelDescriptor
+	reactorsByCh  map[byte]Reactor
+	msgTypeByChID map[byte]proto.Message
+	peers         *PeerSet
+	dialing       *cmap.CMap
+	reconnecting  *cmap.CMap
+	nodeInfo      NodeInfo // our node info
+	nodeKey       *NodeKey // our node privkey
+	addrBook      AddrBook
 	// peers addresses with whom we'll maintain constant connection
 	persistentPeersAddrs []*NetAddress
 	unconditionalPeerIDs map[ID]struct{}
@@ -91,6 +92,7 @@ type Switch struct {
 	rng *rand.Rand // seed for randomizing dial times and orders
 
 	metrics *Metrics
+	mlc     *metricsLabelCache
 }
 
 // NetAddress returns the address the switch is listening on.
@@ -108,11 +110,13 @@ func NewSwitch(
 	transport Transport,
 	options ...SwitchOption,
 ) *Switch {
+
 	sw := &Switch{
 		config:               cfg,
 		reactors:             make(map[string]Reactor),
 		chDescs:              make([]*conn.ChannelDescriptor, 0),
 		reactorsByCh:         make(map[byte]Reactor),
+		msgTypeByChID:        make(map[byte]proto.Message),
 		peers:                NewPeerSet(),
 		dialing:              cmap.NewCMap(),
 		reconnecting:         cmap.NewCMap(),
@@ -121,6 +125,7 @@ func NewSwitch(
 		filterTimeout:        defaultFilterTimeout,
 		persistentPeersAddrs: make([]*NetAddress, 0),
 		unconditionalPeerIDs: make(map[ID]struct{}),
+		mlc:                  newMetricsLabelCache(),
 	}
 
 	// Ensure we have a completely undeterministic PRNG.
@@ -164,6 +169,7 @@ func (sw *Switch) AddReactor(name string, reactor Reactor) Reactor {
 		}
 		sw.chDescs = append(sw.chDescs, chDesc)
 		sw.reactorsByCh[chID] = reactor
+		sw.msgTypeByChID[chID] = chDesc.MessageType
 	}
 	sw.reactors[name] = reactor
 	reactor.SetSwitch(sw)
@@ -182,6 +188,7 @@ func (sw *Switch) RemoveReactor(name string, reactor Reactor) {
 			}
 		}
 		delete(sw.reactorsByCh, chDesc.ID)
+		delete(sw.msgTypeByChID, chDesc.ID)
 	}
 	delete(sw.reactors, name)
 	reactor.SetSwitch(nil)
@@ -255,14 +262,49 @@ func (sw *Switch) OnStop() {
 //---------------------------------------------------------------------
 // Peers
 
+// BroadcastEnvelope runs a go routine for each attempted send, which will block trying
+// to send for defaultSendTimeoutSeconds. Returns a channel which receives
+// success values for each attempted send (false if times out). Channel will be
+// closed once msg bytes are sent to all peers (or time out).
+// BroadcastEnvelope sends to the peers using the SendEnvelope method.
+//
+// NOTE: BroadcastEnvelope uses goroutines, so order of broadcast may not be preserved.
+func (sw *Switch) BroadcastEnvelope(e Envelope) chan bool {
+	sw.Logger.Debug("Broadcast", "channel", e.ChannelID)
+
+	peers := sw.peers.List()
+	var wg sync.WaitGroup
+	wg.Add(len(peers))
+	successChan := make(chan bool, len(peers))
+
+	for _, peer := range peers {
+		go func(p Peer) {
+			defer wg.Done()
+			success := SendEnvelopeShim(p, e, sw.Logger)
+			successChan <- success
+		}(peer)
+	}
+
+	go func() {
+		wg.Wait()
+		close(successChan)
+	}()
+
+	return successChan
+}
+
 // Broadcast runs a go routine for each attempted send, which will block trying
 // to send for defaultSendTimeoutSeconds. Returns a channel which receives
 // success values for each attempted send (false if times out). Channel will be
 // closed once msg bytes are sent to all peers (or time out).
+// Broadcast sends to the peers using the Send method.
 //
 // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
+//
+// Deprecated: code looking to broadcast data to all peers should use BroadcastEnvelope.
+// Broadcast will be removed in 0.37.
 func (sw *Switch) Broadcast(chID byte, msgBytes []byte) chan bool {
-	sw.Logger.Debug("Broadcast", "channel", chID, "msgBytes", log.NewLazySprintf("%X", msgBytes))
+	sw.Logger.Debug("Broadcast", "channel", chID)
 
 	peers := sw.peers.List()
 	var wg sync.WaitGroup
@@ -623,11 +665,13 @@ func (sw *Switch) IsPeerPersistent(na *NetAddress) bool {
 func (sw *Switch) acceptRoutine() {
 	for {
 		p, err := sw.transport.Accept(peerConfig{
-			chDescs:      sw.chDescs,
-			onPeerError:  sw.StopPeerForError,
-			reactorsByCh: sw.reactorsByCh,
-			metrics:      sw.metrics,
-			isPersistent: sw.IsPeerPersistent,
+			chDescs:       sw.chDescs,
+			onPeerError:   sw.StopPeerForError,
+			reactorsByCh:  sw.reactorsByCh,
+			msgTypeByChID: sw.msgTypeByChID,
+			metrics:       sw.metrics,
+			mlc:           sw.mlc,
+			isPersistent:  sw.IsPeerPersistent,
 		})
 		if err != nil {
 			switch err := err.(type) {
@@ -726,11 +770,13 @@ func (sw *Switch) addOutboundPeerWithConfig(
 	}
 
 	p, err := sw.transport.Dial(*addr, peerConfig{
-		chDescs:      sw.chDescs,
-		onPeerError:  sw.StopPeerForError,
-		isPersistent: sw.IsPeerPersistent,
-		reactorsByCh: sw.reactorsByCh,
-		metrics:      sw.metrics,
+		chDescs:       sw.chDescs,
+		onPeerError:   sw.StopPeerForError,
+		isPersistent:  sw.IsPeerPersistent,
+		reactorsByCh:  sw.reactorsByCh,
+		msgTypeByChID: sw.msgTypeByChID,
+		metrics:       sw.metrics,
+		mlc:           sw.mlc,
 	})
 	if err != nil {
 		if e, ok := err.(ErrRejected); ok {

p2p/switch_test.go

@@ -14,6 +14,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -23,6 +24,7 @@ import (
 	"github.com/tendermint/tendermint/libs/log"
 	tmsync "github.com/tendermint/tendermint/libs/sync"
 	"github.com/tendermint/tendermint/p2p/conn"
+	p2pproto "github.com/tendermint/tendermint/proto/tendermint/p2p"
 )
 
 var cfg *config.P2PConfig
@@ -34,9 +36,8 @@ func init() {
 }
 
 type PeerMessage struct {
-	PeerID  ID
-	Bytes   []byte
-	Counter int
+	Contents proto.Message
+	Counter  int
 }
 
 type TestReactor struct {
@@ -68,16 +69,34 @@ func (tr *TestReactor) AddPeer(peer Peer) {}
 
 func (tr *TestReactor) RemovePeer(peer Peer, reason interface{}) {}
 
-func (tr *TestReactor) Receive(chID byte, peer Peer, msgBytes []byte) {
+func (tr *TestReactor) ReceiveEnvelope(e Envelope) {
 	if tr.logMessages {
 		tr.mtx.Lock()
 		defer tr.mtx.Unlock()
-		// fmt.Printf("Received: %X, %X\n", chID, msgBytes)
-		tr.msgsReceived[chID] = append(tr.msgsReceived[chID], PeerMessage{peer.ID(), msgBytes, tr.msgsCounter})
+		// fmt.Printf("Received: %X, %X\n", e.ChannelID, e.Message)
+		tr.msgsReceived[e.ChannelID] = append(tr.msgsReceived[e.ChannelID], PeerMessage{Contents: e.Message, Counter: tr.msgsCounter})
 		tr.msgsCounter++
 	}
 }
 
+func (tr *TestReactor) Receive(chID byte, peer Peer, msgBytes []byte) {
+	var msg *p2pproto.Message
+	err := proto.Unmarshal(msgBytes, msg)
+	if err != nil {
+		panic(err)
+	}
+	um, err := msg.Unwrap()
+	if err != nil {
+		panic(err)
+	}
+
+	tr.ReceiveEnvelope(Envelope{
+		ChannelID: chID,
+		Src:       peer,
+		Message:   um,
+	})
+}
+
 func (tr *TestReactor) getMsgs(chID byte) []PeerMessage {
 	tr.mtx.Lock()
 	defer tr.mtx.Unlock()
@@ -102,12 +121,12 @@ func initSwitchFunc(i int, sw *Switch) *Switch {
 
 	// Make two reactors of two channels each
 	sw.AddReactor("foo", NewTestReactor([]*conn.ChannelDescriptor{
-		{ID: byte(0x00), Priority: 10},
-		{ID: byte(0x01), Priority: 10},
+		{ID: byte(0x00), Priority: 10, MessageType: &p2pproto.Message{}},
+		{ID: byte(0x01), Priority: 10, MessageType: &p2pproto.Message{}},
 	}, true))
 	sw.AddReactor("bar", NewTestReactor([]*conn.ChannelDescriptor{
-		{ID: byte(0x02), Priority: 10},
-		{ID: byte(0x03), Priority: 10},
+		{ID: byte(0x02), Priority: 10, MessageType: &p2pproto.Message{}},
+		{ID: byte(0x03), Priority: 10, MessageType: &p2pproto.Message{}},
 	}, true))
 
 	return sw
@@ -134,31 +153,47 @@ func TestSwitches(t *testing.T) {
 	}
 
 	// Lets send some messages
-	ch0Msg := []byte("channel zero")
-	ch1Msg := []byte("channel foo")
-	ch2Msg := []byte("channel bar")
-
-	s1.Broadcast(byte(0x00), ch0Msg)
-	s1.Broadcast(byte(0x01), ch1Msg)
-	s1.Broadcast(byte(0x02), ch2Msg)
-
+	ch0Msg := &p2pproto.PexAddrs{
+		Addrs: []p2pproto.NetAddress{
+			{
+				ID: "1",
+			},
+		},
+	}
+	ch1Msg := &p2pproto.PexAddrs{
+		Addrs: []p2pproto.NetAddress{
+			{
+				ID: "1",
+			},
+		},
+	}
+	ch2Msg := &p2pproto.PexAddrs{
+		Addrs: []p2pproto.NetAddress{
+			{
+				ID: "2",
+			},
+		},
+	}
+	s1.BroadcastEnvelope(Envelope{ChannelID: byte(0x00), Message: ch0Msg})
+	s1.BroadcastEnvelope(Envelope{ChannelID: byte(0x01), Message: ch1Msg})
+	s1.BroadcastEnvelope(Envelope{ChannelID: byte(0x02), Message: ch2Msg})
 	assertMsgReceivedWithTimeout(t,
 		ch0Msg,
 		byte(0x00),
-		s2.Reactor("foo").(*TestReactor), 10*time.Millisecond, 5*time.Second)
+		s2.Reactor("foo").(*TestReactor), 200*time.Millisecond, 5*time.Second)
 	assertMsgReceivedWithTimeout(t,
 		ch1Msg,
 		byte(0x01),
-		s2.Reactor("foo").(*TestReactor), 10*time.Millisecond, 5*time.Second)
+		s2.Reactor("foo").(*TestReactor), 200*time.Millisecond, 5*time.Second)
 	assertMsgReceivedWithTimeout(t,
 		ch2Msg,
 		byte(0x02),
-		s2.Reactor("bar").(*TestReactor), 10*time.Millisecond, 5*time.Second)
+		s2.Reactor("bar").(*TestReactor), 200*time.Millisecond, 5*time.Second)
 }
 
 func assertMsgReceivedWithTimeout(
 	t *testing.T,
-	msgBytes []byte,
+	msg proto.Message,
 	channel byte,
 	reactor *TestReactor,
 	checkPeriod,
@@ -169,9 +204,13 @@ func assertMsgReceivedWithTimeout(
 		select {
 		case <-ticker.C:
 			msgs := reactor.getMsgs(channel)
+			expectedBytes, err := proto.Marshal(msgs[0].Contents)
+			require.NoError(t, err)
+			gotBytes, err := proto.Marshal(msg)
+			require.NoError(t, err)
 			if len(msgs) > 0 {
-				if !bytes.Equal(msgs[0].Bytes, msgBytes) {
-					t.Fatalf("Unexpected message bytes. Wanted: %X, Got: %X", msgBytes, msgs[0].Bytes)
+				if !bytes.Equal(expectedBytes, gotBytes) {
+					t.Fatalf("Unexpected message bytes. Wanted: %X, Got: %X", msg, msgs[0].Counter)
 				}
 				return
 			}
@@ -428,7 +467,10 @@ func TestSwitchStopPeerForError(t *testing.T) {
 
 	// send messages to the peer from sw1
 	p := sw1.Peers().List()[0]
-	p.Send(0x1, []byte("here's a message to send"))
+	SendEnvelopeShim(p, Envelope{
+		ChannelID: 0x1,
+		Message:   &p2pproto.Message{},
+	}, sw1.Logger)
 
 	// stop sw2. this should cause the p to fail,
 	// which results in calling StopPeerForError internally
@@ -825,7 +867,7 @@ func BenchmarkSwitchBroadcast(b *testing.B) {
 	// Send random message from foo channel to another
 	for i := 0; i < b.N; i++ {
 		chID := byte(i % 4)
-		successChan := s1.Broadcast(chID, []byte("test data"))
+		successChan := s1.BroadcastEnvelope(Envelope{ChannelID: chID})
 		for s := range successChan {
 			if s {
 				numSuccess++

p2p/test_util.go

@@ -149,8 +149,10 @@ func (sw *Switch) addPeerWithConnection(conn net.Conn) error {
 		MConnConfig(sw.config),
 		ni,
 		sw.reactorsByCh,
+		sw.msgTypeByChID,
 		sw.chDescs,
 		sw.StopPeerForError,
+		sw.mlc,
 	)
 
 	if err = sw.addPeer(p); err != nil {

p2p/transport.go

@@ -8,6 +8,7 @@ import (
 
 	"golang.org/x/net/netutil"
 
+	"github.com/gogo/protobuf/proto"
 	"github.com/tendermint/tendermint/crypto"
 	"github.com/tendermint/tendermint/libs/protoio"
 	"github.com/tendermint/tendermint/p2p/conn"
@@ -47,9 +48,11 @@ type peerConfig struct {
 	// isPersistent allows you to set a function, which, given socket address
 	// (for outbound peers) OR self-reported address (for inbound peers), tells
 	// if the peer is persistent or not.
-	isPersistent func(*NetAddress) bool
-	reactorsByCh map[byte]Reactor
-	metrics      *Metrics
+	isPersistent  func(*NetAddress) bool
+	reactorsByCh  map[byte]Reactor
+	msgTypeByChID map[byte]proto.Message
+	metrics       *Metrics
+	mlc           *metricsLabelCache
 }
 
 // Transport emits and connects to Peers. The implementation of Peer is left to
@@ -519,8 +522,10 @@ func (mt *MultiplexTransport) wrapPeer(
 		mt.mConfig,
 		ni,
 		cfg.reactorsByCh,
+		cfg.msgTypeByChID,
 		cfg.chDescs,
 		cfg.onPeerError,
+		cfg.mlc,
 		PeerMetrics(cfg.metrics),
 	)
 

p2p/types.go

@@ -1,8 +1,40 @@
 package p2p
 
 import (
+	"github.com/gogo/protobuf/proto"
 	"github.com/tendermint/tendermint/p2p/conn"
+	tmp2p "github.com/tendermint/tendermint/proto/tendermint/p2p"
 )
 
 type ChannelDescriptor = conn.ChannelDescriptor
 type ConnectionStatus = conn.ConnectionStatus
+
+// Envelope contains a message with sender routing info.
+type Envelope struct {
+	Src       Peer          // sender (empty if outbound)
+	Message   proto.Message // message payload
+	ChannelID byte
+}
+
+// Unwrapper is a Protobuf message that can contain a variety of inner messages
+// (e.g. via oneof fields). If a Channel's message type implements Unwrapper, the
+// p2p layer will automatically unwrap inbound messages so that reactors do not have to do this themselves.
+type Unwrapper interface {
+	proto.Message
+
+	// Unwrap will unwrap the inner message contained in this message.
+	Unwrap() (proto.Message, error)
+}
+
+// Wrapper is a companion type to Unwrapper. It is a Protobuf message that can contain a variety of inner messages. The p2p layer will automatically wrap outbound messages so that the reactors do not have to do it themselves.
+type Wrapper interface {
+	proto.Message
+
+	// Wrap will take the underlying message and wrap it in its wrapper type.
+	Wrap() proto.Message
+}
+
+var (
+	_ Wrapper = &tmp2p.PexRequest{}
+	_ Wrapper = &tmp2p.PexAddrs{}
+)