p2p: collect errors from channel (#8544)

go.mod

@@ -42,6 +42,7 @@ require (
 	github.com/creachadair/taskgroup v0.3.2
 	github.com/golangci/golangci-lint v1.46.0
 	github.com/google/go-cmp v0.5.8
+	github.com/tychoish/emt v0.1.0
 	github.com/vektra/mockery/v2 v2.12.2
 	gotest.tools v2.2.0+incompatible
 )

go.sum

@@ -1503,6 +1503,8 @@ github.com/tommy-muehle/go-mnd/v2 v2.5.0 h1:iAj0a8e6+dXSL7Liq0aXPox36FiN1dBbjA6l
 github.com/tommy-muehle/go-mnd/v2 v2.5.0/go.mod h1:WsUAkMJMYww6l/ufffCD3m+P7LEvr8TnZn9lwVDlgzw=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
 github.com/tv42/httpunix v0.0.0-20191220191345-2ba4b9c3382c/go.mod h1:hzIxponao9Kjc7aWznkXaL4U4TWaDSs8zcsY4Ka08nM=
+github.com/tychoish/emt v0.1.0 h1:OMJ/lKO17hxijZW2UdbZY4vOfTwVKZzTb8FB0g8fh74=
+github.com/tychoish/emt v0.1.0/go.mod h1:KMhVWIsL98Q8AoorphjsOQWabcNOoYqMSpODR/rtRb4=
 github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/ultraware/funlen v0.0.3 h1:5ylVWm8wsNwH5aWo9438pwvsK0QiqVuUrt9bn7S/iLA=

internal/p2p/channel.go

@@ -14,6 +14,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/protocol"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/tychoish/emt"
 
 	"github.com/tendermint/tendermint/internal/libs/protoio"
 	"github.com/tendermint/tendermint/types"
@@ -45,6 +46,8 @@ type Wrapper interface {
 type Channel interface {
 	fmt.Stringer
 
+	Err() error
+
 	Send(context.Context, Envelope) error
 	SendError(context.Context, PeerError) error
 	Receive(context.Context) *ChannelIterator
@@ -102,6 +105,8 @@ func (ch *legacyChannel) Send(ctx context.Context, envelope Envelope) error {
 	}
 }
 
+func (ch *legacyChannel) Err() error { return nil }
+
 // SendError blocks until the given error has been sent, or ctx ends.
 // An error only occurs if the context ends before the send completes.
 func (ch *legacyChannel) SendError(ctx context.Context, pe PeerError) error {
@@ -236,6 +241,8 @@ type libp2pChannelImpl struct {
 	recvCh  chan Envelope
 	chainID string
 	wrapper Wrapper
+
+	errs emt.Catcher
 }
 
 func NewLibP2PChannel(chainID string, chDesc *ChannelDescriptor, ps *pubsub.PubSub, h host.Host) (Channel, error) {
@@ -245,6 +252,7 @@ func NewLibP2PChannel(chainID string, chDesc *ChannelDescriptor, ps *pubsub.PubS
 		host:    h,
 		chainID: chainID,
 		recvCh:  make(chan Envelope, chDesc.RecvMessageCapacity),
+		errs:    emt.NewCatcher(),
 	}
 	topic, err := ps.Join(ch.canonicalizedTopicName())
 	if err != nil {
@@ -263,6 +271,8 @@ func (ch *libp2pChannelImpl) String() string {
 	return fmt.Sprintf("Channel<%s>", ch.canonicalizedTopicName())
 }
 
+func (ch *libp2pChannelImpl) Err() error { return ch.errs.Resolve() }
+
 func (ch *libp2pChannelImpl) canonicalizedTopicName() string {
 	return fmt.Sprintf("%s.%s.%d", ch.chainID, ch.chDesc.Name, ch.chDesc.ID)
 }
@@ -287,7 +297,7 @@ func (ch *libp2pChannelImpl) Receive(ctx context.Context) *ChannelIterator {
 
 		ctx, cancel := context.WithCancel(ctx)
 		defer cancel()
-		go func() { <-ctx.Done(); _ = stream.Close() }()
+		go func() { <-ctx.Done(); ch.errs.Add(stream.Close()) }()
 
 		for {
 			payload := proto.Clone(ch.chDesc.MessageType)
@@ -296,7 +306,8 @@ func (ch *libp2pChannelImpl) Receive(ctx context.Context) *ChannelIterator {
 				if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 					return
 				}
-				// TODO: propagate or capture this error
+				ch.errs.Add(err)
+				continue
 			}
 			select {
 			case <-ctx.Done():
@@ -312,6 +323,7 @@ func (ch *libp2pChannelImpl) Receive(ctx context.Context) *ChannelIterator {
 
 	sub, err := ch.topic.Subscribe()
 	if err != nil {
+		ch.errs.Add(err)
 		return nil
 	}
 
@@ -341,20 +353,22 @@ func (ch *libp2pChannelImpl) Receive(ctx context.Context) *ChannelIterator {
 		for {
 			msg, err := sub.Next(ctx)
 			if err != nil {
-				// TODO: maybe signal to users that it
-				// was canceled, when we begin
-				// propagating errors out.
+				if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+					return
+				}
+
+				ch.errs.Add(err)
 				return
 			}
 
 			payload := proto.Clone(ch.chDesc.MessageType)
 			if err := proto.Unmarshal(msg.Data, payload); err != nil {
-				// TODO: add error reporting abilities
-				// so we don't just miss these errors.
+				ch.errs.Add(err)
 				return
 			}
 			if wrapper, ok := payload.(Wrapper); ok {
 				if payload, err = wrapper.Unwrap(); err != nil {
+					ch.errs.Add(err)
 					return
 				}
 			}

internal/p2p/router.go

@@ -1038,6 +1038,15 @@ func (r *Router) OnStart(ctx context.Context) error {
 // sender's responsibility.
 func (r *Router) OnStop() {
 	if r.options.UseLibP2P {
+		for name, ch := range r.network.channels {
+			if err := ch.Err(); err != nil {
+				r.logger.Error("shutting down channel",
+					"name", name,
+					"err", ch.Err(),
+				)
+			}
+		}
+
 		return
 	}