error-codes: add interface methods

core/network/conn.go

@@ -11,6 +11,8 @@ import (
 	ma "github.com/multiformats/go-multiaddr"
 )
 
+type ConnErrorCode uint32
+
 // 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:
@@ -19,6 +21,10 @@ import (
 type Conn interface {
 	io.Closer
 
+	// CloseWithError closes the connection with errCode. The errCode is sent
+	// to the peer.
+	CloseWithError(errCode ConnErrorCode) error
+
 	ConnSecurity
 	ConnMultiaddrs
 	ConnStat

core/network/mux.go

@@ -11,6 +11,12 @@ import (
 // ErrReset is returned when reading or writing on a reset stream.
 var ErrReset = errors.New("stream reset")
 
+type StreamErrorCode uint32
+
+type StreamError struct {
+	ErrorCode int
+}
+
 // MuxedStream is a bidirectional io pipe within a connection.
 type MuxedStream interface {
 	io.Reader
@@ -56,6 +62,10 @@ type MuxedStream interface {
 	// side to hang up and go away.
 	Reset() error
 
+	// ResetWithError closes both ends of the stream with errCode. The errCode is sent
+	// to the peer.
+	ResetWithError(errCode StreamErrorCode) error
+
 	SetDeadline(time.Time) error
 	SetReadDeadline(time.Time) error
 	SetWriteDeadline(time.Time) error
@@ -75,6 +85,10 @@ type MuxedConn interface {
 	// Close closes the stream muxer and the the underlying net.Conn.
 	io.Closer
 
+	// CloseWithError closes the connection with errCode. The errCode is sent
+	// to the peer.
+	CloseWithError(errCode ConnErrorCode) error
+
 	// IsClosed returns whether a connection is fully closed, so it can
 	// be garbage collected.
 	IsClosed() bool

p2p/muxer/yamux/conn.go

@@ -23,6 +23,10 @@ func (c *conn) Close() error {
 	return c.yamux().Close()
 }
 
+func (c *conn) CloseWithError(errCode network.ConnErrorCode) error {
+	panic("not implemented")
+}
+
 // IsClosed checks if yamux.Session is in closed state.
 func (c *conn) IsClosed() bool {
 	return c.yamux().IsClosed()

p2p/muxer/yamux/stream.go

@@ -39,6 +39,10 @@ func (s *stream) Reset() error {
 	return s.yamux().Reset()
 }
 
+func (s *stream) ResetWithError(errCode network.StreamErrorCode) error {
+	panic("not implemented")
+}
+
 func (s *stream) CloseRead() error {
 	return s.yamux().CloseRead()
 }

p2p/net/connmgr/connmgr_test.go

@@ -794,6 +794,7 @@ type mockConn struct {
 }
 
 func (m mockConn) Close() error                                          { panic("implement me") }
+func (m mockConn) CloseWithError(errCode network.ConnErrorCode) error    { panic("implement me") }
 func (m mockConn) LocalPeer() peer.ID                                    { panic("implement me") }
 func (m mockConn) RemotePeer() peer.ID                                   { panic("implement me") }
 func (m mockConn) RemotePublicKey() crypto.PubKey                        { panic("implement me") }

p2p/net/mock/mock_conn.go

@@ -86,6 +86,10 @@ func (c *conn) Close() error {
 	return nil
 }
 
+func (c *conn) CloseWithError(errCode network.ConnErrorCode) error {
+	panic("not implemented")
+}
+
 func (c *conn) teardown() {
 	for _, s := range c.allStreams() {
 		s.Reset()

p2p/net/mock/mock_stream.go

@@ -144,6 +144,10 @@ func (s *stream) Reset() error {
 	return nil
 }
 
+func (s *stream) ResetWithError(errCode network.StreamErrorCode) error {
+	panic("not implemented")
+}
+
 func (s *stream) teardown() {
 	// at this point, no streams are writing.
 	s.conn.removeStream(s)

p2p/net/swarm/swarm_conn.go

@@ -62,6 +62,10 @@ func (c *Conn) Close() error {
 	return c.err
 }
 
+func (c *Conn) CloseWithError(errCode network.ConnErrorCode) error {
+	panic("not implemented")
+}
+
 func (c *Conn) doClose() {
 	c.swarm.removeConn(c)
 

p2p/net/swarm/swarm_stream.go

@@ -91,6 +91,10 @@ func (s *Stream) Reset() error {
 	return err
 }
 
+func (s *Stream) ResetWithError(errCode network.StreamErrorCode) error {
+	panic("not implemented")
+}
+
 func (s *Stream) closeAndRemoveStream() {
 	s.closeMx.Lock()
 	defer s.closeMx.Unlock()

p2p/transport/quic/conn.go

@@ -34,6 +34,13 @@ func (c *conn) Close() error {
 	return c.closeWithError(0, "")
 }
 
+// CloseWithError closes the connection
+// It must be called even if the peer closed the connection in order for
+// garbage collection to properly work in this package.
+func (c *conn) CloseWithError(errCode network.ConnErrorCode) error {
+	return c.closeWithError(quic.ApplicationErrorCode(errCode), "")
+}
+
 func (c *conn) closeWithError(errCode quic.ApplicationErrorCode, errString string) error {
 	c.transport.removeConn(c.quicConn)
 	err := c.quicConn.CloseWithError(errCode, errString)

p2p/transport/quic/stream.go

@@ -40,6 +40,10 @@ func (s *stream) Reset() error {
 	return nil
 }
 
+func (s *stream) ResetWithError(errCode network.StreamErrorCode) error {
+	panic("not implemented")
+}
+
 func (s *stream) Close() error {
 	s.Stream.CancelRead(reset)
 	return s.Stream.Close()

p2p/transport/webrtc/connection.go

@@ -132,6 +132,10 @@ func (c *connection) Close() error {
 	return nil
 }
 
+func (c *connection) CloseWithError(errCode network.ConnErrorCode) error {
+	panic("not implemented")
+}
+
 // closeWithError is used to Close the connection when the underlying DTLS connection fails
 func (c *connection) closeWithError(err error) {
 	c.closeOnce.Do(func() {

p2p/transport/webrtc/stream.go

@@ -160,6 +160,10 @@ func (s *stream) Reset() error {
 	return errors.Join(closeReadErr, cancelWriteErr)
 }
 
+func (s *stream) ResetWithError(errCode network.StreamErrorCode) error {
+	panic("not implemented")
+}
+
 func (s *stream) closeForShutdown(closeErr error) {
 	defer s.cleanup()
 

p2p/transport/webtransport/conn.go

@@ -78,6 +78,10 @@ func (c *conn) Close() error {
 	return err
 }
 
+func (c *conn) CloseWithError(errCode network.ConnErrorCode) error {
+	panic("not implemented")
+}
+
 func (c *conn) IsClosed() bool           { return c.session.Context().Err() != nil }
 func (c *conn) Scope() network.ConnScope { return c.scope }
 func (c *conn) Transport() tpt.Transport { return c.transport }

p2p/transport/webtransport/stream.go

@@ -56,6 +56,10 @@ func (s *stream) Reset() error {
 	return nil
 }
 
+func (s *stream) ResetWithError(errCode network.StreamErrorCode) error {
+	panic("not implemented")
+}
+
 func (s *stream) Close() error {
 	s.Stream.CancelRead(reset)
 	return s.Stream.Close()