From b52c33939de793a8ef282cb7a0f5abe1292d202c Mon Sep 17 00:00:00 2001 From: Marten Seemann Date: Fri, 28 Jun 2024 06:58:28 -0700 Subject: [PATCH] introduce a StreamLimitReachedError for Connection.Open{Uni}Stream (#4579) Using a concrete type is preferable to relying on interpreting the return values from net.Error.Timeout and net.Error.Temporary, especially since the latter has been deprecated since Go 1.18. --- interface.go | 25 +++++++++++++------------ streams_map.go | 18 ++++++++++++++---- streams_map_outgoing.go | 2 +- streams_map_outgoing_test.go | 6 ++---- streams_map_test.go | 7 +++++-- 5 files changed, 35 insertions(+), 23 deletions(-) diff --git a/interface.go b/interface.go index e7368045e64..cec92d6de6a 100644 --- a/interface.go +++ b/interface.go @@ -150,7 +150,7 @@ type SendStream interface { // * TransportError: for errors triggered by the QUIC transport (in many cases a misbehaving peer) // * IdleTimeoutError: when the peer goes away unexpectedly (this is a net.Error timeout error) // * HandshakeTimeoutError: when the cryptographic handshake takes too long (this is a net.Error timeout error) -// * StatelessResetError: when we receive a stateless reset (this is a net.Error temporary error) +// * StatelessResetError: when we receive a stateless reset // * VersionNegotiationError: returned by the client, when there's no version overlap between the peers type Connection interface { // AcceptStream returns the next stream opened by the peer, blocking until one is available. @@ -163,28 +163,29 @@ type Connection interface { AcceptUniStream(context.Context) (ReceiveStream, error) // OpenStream opens a new bidirectional QUIC stream. // There is no signaling to the peer about new streams: - // The peer can only accept the stream after data has been sent on the stream. - // If the error is non-nil, it satisfies the net.Error interface. - // When reaching the peer's stream limit, err.Temporary() will be true. - // If the connection was closed due to a timeout, Timeout() will be true. + // The peer can only accept the stream after data has been sent on the stream, + // or the stream has been reset or closed. + // When reaching the peer's stream limit, it is not possible to open a new stream until the + // peer raises the stream limit. In that case, a StreamLimitReachedError is returned. OpenStream() (Stream, error) // OpenStreamSync opens a new bidirectional QUIC stream. // It blocks until a new stream can be opened. // There is no signaling to the peer about new streams: // The peer can only accept the stream after data has been sent on the stream, // or the stream has been reset or closed. - // If the error is non-nil, it satisfies the net.Error interface. - // If the connection was closed due to a timeout, Timeout() will be true. OpenStreamSync(context.Context) (Stream, error) // OpenUniStream opens a new outgoing unidirectional QUIC stream. - // If the error is non-nil, it satisfies the net.Error interface. - // When reaching the peer's stream limit, Temporary() will be true. - // If the connection was closed due to a timeout, Timeout() will be true. + // There is no signaling to the peer about new streams: + // The peer can only accept the stream after data has been sent on the stream, + // or the stream has been reset or closed. + // When reaching the peer's stream limit, it is not possible to open a new stream until the + // peer raises the stream limit. In that case, a StreamLimitReachedError is returned. OpenUniStream() (SendStream, error) // OpenUniStreamSync opens a new outgoing unidirectional QUIC stream. // It blocks until a new stream can be opened. - // If the error is non-nil, it satisfies the net.Error interface. - // If the connection was closed due to a timeout, Timeout() will be true. + // There is no signaling to the peer about new streams: + // The peer can only accept the stream after data has been sent on the stream, + // or the stream has been reset or closed. OpenUniStreamSync(context.Context) (SendStream, error) // LocalAddr returns the local address. LocalAddr() net.Addr diff --git a/streams_map.go b/streams_map.go index 0ba23b2543f..041636c34ba 100644 --- a/streams_map.go +++ b/streams_map.go @@ -38,11 +38,21 @@ type streamOpenErr struct{ error } var _ net.Error = &streamOpenErr{} -func (e streamOpenErr) Temporary() bool { return e.error == errTooManyOpenStreams } -func (streamOpenErr) Timeout() bool { return false } +func (streamOpenErr) Timeout() bool { return false } +func (e streamOpenErr) Unwrap() error { return e.error } -// errTooManyOpenStreams is used internally by the outgoing streams maps. -var errTooManyOpenStreams = errors.New("too many open streams") +func (e streamOpenErr) Temporary() bool { + // In older versions of quic-go, the stream limit error was documented to be a net.Error.Temporary. + // This function was since deprecated, but we keep the existing behavior. + return errors.Is(e, &StreamLimitReachedError{}) +} + +// StreamLimitReachedError is returned from Connection.OpenStream and Connection.OpenUniStream +// when it is not possible to open a new stream because the number of opens streams reached +// the peer's stream limit. +type StreamLimitReachedError struct{} + +func (e StreamLimitReachedError) Error() string { return "too many open streams" } type streamsMap struct { ctx context.Context // not used for cancellations, but carries the values associated with the connection diff --git a/streams_map_outgoing.go b/streams_map_outgoing.go index fd45f4e7cf1..a8d04b04f97 100644 --- a/streams_map_outgoing.go +++ b/streams_map_outgoing.go @@ -60,7 +60,7 @@ func (m *outgoingStreamsMap[T]) OpenStream() (T, error) { // if there are OpenStreamSync calls waiting, return an error here if len(m.openQueue) > 0 || m.nextStream > m.maxStream { m.maybeSendBlockedFrame() - return *new(T), streamOpenErr{errTooManyOpenStreams} + return *new(T), streamOpenErr{&StreamLimitReachedError{}} } return m.openStream(), nil } diff --git a/streams_map_outgoing_test.go b/streams_map_outgoing_test.go index 0d84ef5279f..3ae3371776c 100644 --- a/streams_map_outgoing_test.go +++ b/streams_map_outgoing_test.go @@ -361,8 +361,7 @@ var _ = Describe("Streams Map (outgoing)", func() { Expect(bf.StreamLimit).To(BeEquivalentTo(6)) }) _, err := m.OpenStream() - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(Equal(errTooManyOpenStreams.Error())) + Expect(err).To(MatchError(&StreamLimitReachedError{})) }) It("only sends one STREAMS_BLOCKED frame for one stream ID", func() { @@ -452,8 +451,7 @@ var _ = Describe("Streams Map (outgoing)", func() { } str, err := m.OpenStream() if limit <= n { - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(Equal(errTooManyOpenStreams.Error())) + Expect(err).To(MatchError(&StreamLimitReachedError{})) } else { Expect(str.num).To(Equal(protocol.StreamNum(n + 1))) } diff --git a/streams_map_test.go b/streams_map_test.go index b4a2d91ffae..6e1f1c4e082 100644 --- a/streams_map_test.go +++ b/streams_map_test.go @@ -33,11 +33,14 @@ type streamMapping struct { } func expectTooManyStreamsError(err error) { - ExpectWithOffset(1, err).To(HaveOccurred()) - ExpectWithOffset(1, err.Error()).To(Equal(errTooManyOpenStreams.Error())) + ExpectWithOffset(1, err).To(MatchError(&StreamLimitReachedError{})) nerr, ok := err.(net.Error) ExpectWithOffset(1, ok).To(BeTrue()) ExpectWithOffset(1, nerr.Timeout()).To(BeFalse()) + //nolint:staticcheck // SA1019 + // In older versions of quic-go, the stream limit error was documented to be a net.Error.Temporary. + // This function was since deprecated, but we keep the existing behavior. + ExpectWithOffset(1, nerr.Temporary()).To(BeTrue()) } var _ = Describe("Streams Map", func() {