update p2p.HandlerFunc to return error

pkg/p2p/libp2p/libp2p.go

@@ -270,7 +270,9 @@ func (s *Service) AddProtocol(p p2p.ProtocolSpec) (err error) {
 			}
 
 			s.metrics.HandledStreamCount.Inc()
-			ss.Handler(p2p.Peer{Address: overlay}, stream)
+			if err := ss.Handler(p2p.Peer{Address: overlay}, stream); err != nil {
+				s.logger.Errorf("%s: %s/%s: %w", p.Name, ss.Name, ss.Version, err)
+			}
 		})
 	}
 	return nil

pkg/p2p/mock/mock.go

@@ -48,7 +48,7 @@ func (r *Recorder) NewStream(_ context.Context, overlay, protocolName, streamNam
 	streamOut := newStream(recordIn, recordOut)
 	streamIn := newStream(recordOut, recordIn)
 
-	var handler func(p2p.Peer, p2p.Stream)
+	var handler p2p.HandlerFunc
 	for _, p := range r.protocols {
 		if p.Name == protocolName {
 			for _, s := range p.StreamSpecs {
@@ -64,7 +64,11 @@ func (r *Recorder) NewStream(_ context.Context, overlay, protocolName, streamNam
 	for _, m := range r.middlewares {
 		handler = m(handler)
 	}
-	go handler(p2p.Peer{Address: overlay}, streamIn)
+	go func() {
+		if err := handler(p2p.Peer{Address: overlay}, streamIn); err != nil {
+			panic(err) // todo: store error and export error records for inspection
+		}
+	}()
 
 	id := overlay + p2p.NewSwarmStreamName(protocolName, streamName, version)
 

pkg/p2p/p2p.go

@@ -37,7 +37,7 @@ type StreamSpec struct {
 	Handler HandlerFunc
 }
 
-type HandlerFunc func(Peer, Stream)
+type HandlerFunc func(Peer, Stream) error
 
 type HandlerMiddleware func(HandlerFunc) HandlerFunc
 

pkg/pingpong/pingpong.go

@@ -34,7 +34,6 @@ type Options struct {
 
 type Logger interface {
 	Debugf(format string, args ...interface{})
-	Errorf(format string, args ...interface{})
 }
 
 func New(o Options) *Service {
@@ -73,7 +72,7 @@ func (s *Service) Ping(ctx context.Context, address string, msgs ...string) (rtt
 		if err := w.WriteMsg(&Ping{
 			Greeting: msg,
 		}); err != nil {
-			return 0, fmt.Errorf("stream write: %w", err)
+			return 0, fmt.Errorf("write message: %w", err)
 		}
 		s.metrics.PingSentCount.Inc()
 
@@ -81,7 +80,7 @@ func (s *Service) Ping(ctx context.Context, address string, msgs ...string) (rtt
 			if err == io.EOF {
 				break
 			}
-			return 0, err
+			return 0, fmt.Errorf("read message: %w", err)
 		}
 
 		s.logger.Debugf("got pong: %q", pong.Response)
@@ -90,7 +89,7 @@ func (s *Service) Ping(ctx context.Context, address string, msgs ...string) (rtt
 	return time.Since(start), nil
 }
 
-func (s *Service) Handler(peer p2p.Peer, stream p2p.Stream) {
+func (s *Service) Handler(peer p2p.Peer, stream p2p.Stream) error {
 	w, r := protobuf.NewWriterAndReader(stream)
 	defer stream.Close()
 
@@ -100,8 +99,7 @@ func (s *Service) Handler(peer p2p.Peer, stream p2p.Stream) {
 			if err == io.EOF {
 				break
 			}
-			s.logger.Errorf("pingpong handler: read message: %v\n", err)
-			return
+			return fmt.Errorf("read message: %w", err)
 		}
 		s.logger.Debugf("got ping: %q", ping.Greeting)
 		s.metrics.PingReceivedCount.Inc()
@@ -109,9 +107,9 @@ func (s *Service) Handler(peer p2p.Peer, stream p2p.Stream) {
 		if err := w.WriteMsg(&Pong{
 			Response: "{" + ping.Greeting + "}",
 		}); err != nil {
-			s.logger.Errorf("pingpong handler: write message: %v\n", err)
-			return
+			return fmt.Errorf("write message: %w", err)
 		}
 		s.metrics.PongSentCount.Inc()
 	}
+	return nil
 }