From ffa7af3c36dd177f393a87407d304390d8697e0a Mon Sep 17 00:00:00 2001 From: Sam Kleinman Date: Mon, 16 May 2022 09:40:29 -0400 Subject: [PATCH] p2p: collect errors from channel (#8544) --- go.mod | 1 + go.sum | 2 ++ internal/p2p/channel.go | 28 +++++++++++++++++++++------- internal/p2p/router.go | 9 +++++++++ 4 files changed, 33 insertions(+), 7 deletions(-) diff --git a/go.mod b/go.mod index 730cb8a95..18704c19d 100644 --- a/go.mod +++ b/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 ) diff --git a/go.sum b/go.sum index ed257b9ac..0036dd716 100644 --- a/go.sum +++ b/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= diff --git a/internal/p2p/channel.go b/internal/p2p/channel.go index b81b261f2..47664e36e 100644 --- a/internal/p2p/channel.go +++ b/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 } } diff --git a/internal/p2p/router.go b/internal/p2p/router.go index 8a5001ecb..a55701257 100644 --- a/internal/p2p/router.go +++ b/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 }