quic: implement error codes

core/network/conn.go

@@ -2,6 +2,7 @@ package network
 
 import (
 	"context"
+	"fmt"
 	"io"
 
 	ic "github.com/libp2p/go-libp2p/core/crypto"
@@ -13,6 +14,14 @@ import (
 
 type ConnErrorCode uint32
 
+type ConnError struct {
+	ErrorCode uint32
+}
+
+func (c *ConnError) Error() string {
+	return fmt.Sprintf("connection closed: code: %s", c.ErrorCode)
+}
+
 // Conn is a connection to a remote peer. It multiplexes streams.
 // Usually there is no need to use a Conn directly, but it may
 // be useful to get information about the peer on the other side:

core/network/mux.go

@@ -3,6 +3,7 @@ package network
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"time"
@@ -14,7 +15,15 @@ var ErrReset = errors.New("stream reset")
 type StreamErrorCode uint32
 
 type StreamError struct {
-	ErrorCode int
+	ErrorCode StreamErrorCode
+}
+
+func (s *StreamError) Error() string {
+	return fmt.Sprintf("stream reset: code: %s", s.ErrorCode)
+}
+
+func (s *StreamError) Is(target error) bool {
+	return target == ErrReset
 }
 
 // MuxedStream is a bidirectional io pipe within a connection.

p2p/transport/quic/stream.go

@@ -2,6 +2,7 @@ package libp2pquic
 
 import (
 	"errors"
+	"math"
 
 	"github.com/libp2p/go-libp2p/core/network"
 
@@ -18,20 +19,28 @@ type stream struct {
 
 var _ network.MuxedStream = &stream{}
 
+func parseStreamError(err error) error {
+	se := &quic.StreamError{}
+	if err != nil && errors.As(err, &se) {
+		code := se.ErrorCode
+		if code > math.MaxUint32 {
+			code = 0
+		}
+		err = &network.StreamError{
+			ErrorCode: network.StreamErrorCode(code),
+		}
+	}
+	return err
+}
+
 func (s *stream) Read(b []byte) (n int, err error) {
 	n, err = s.Stream.Read(b)
-	if err != nil && errors.Is(err, &quic.StreamError{}) {
-		err = network.ErrReset
-	}
-	return n, err
+	return n, parseStreamError(err)
 }
 
 func (s *stream) Write(b []byte) (n int, err error) {
 	n, err = s.Stream.Write(b)
-	if err != nil && errors.Is(err, &quic.StreamError{}) {
-		err = network.ErrReset
-	}
-	return n, err
+	return n, parseStreamError(err)
 }
 
 func (s *stream) Reset() error {
@@ -41,7 +50,9 @@ func (s *stream) Reset() error {
 }
 
 func (s *stream) ResetWithError(errCode network.StreamErrorCode) error {
-	panic("not implemented")
+	s.Stream.CancelRead(quic.StreamErrorCode(errCode))
+	s.Stream.CancelWrite(quic.StreamErrorCode(errCode))
+	return nil
 }
 
 func (s *stream) Close() error {