quic: add Stream.StopSending, and rework stream/application errors Rename StreamErrorCode to StreamError, to maintain the convention that error type names end with "Error". Rename ApplicationError to ConnectionCloseError, to distinguish it from StreamError. Add a Stream.StopSending method, which is Stream.CloseRead but permits setting the application error code to send. Add an error return value to Stream.CloseRead and Stream.CloseWrite, to match the signature of net package methods with these names. Change-Id: Ie84b4b0b874636deae8adc9de26e90886a6a6964 Reviewed-on: https://go-review.googlesource.com/c/net/+/825004 Reviewed-by: Nicholas Husin <husin@google.com> LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com <golang-scoped@luci-project-accounts.iam.gserviceaccount.com> Reviewed-by: Nicholas Husin <nsh@golang.org> Auto-Submit: Damien Neil <dneil@google.com>
diff --git a/internal/http3/roundtrip.go b/internal/http3/roundtrip.go index d724ee9..3fd80dc 100644 --- a/internal/http3/roundtrip.go +++ b/internal/http3/roundtrip.go
@@ -284,7 +284,7 @@ // the unsent request. When this happens, the server will use the H3_NO_ERROR // code, and the client MUST NOT discard the response. func reqBodyIgnored(err error) bool { - if streamErr, ok := errors.AsType[quic.StreamErrorCode](err); ok { + if streamErr, ok := errors.AsType[quic.StreamError](err); ok { // TODO: the H3_NO_ERROR should arrive in a QUIC STOP_SENDING frame. // However, the quic package currently only sends code 0 in the // STOP_SENDING frame due to its API limitation.
diff --git a/internal/http3/roundtrip_test.go b/internal/http3/roundtrip_test.go index 3037fb5..54dd10f 100644 --- a/internal/http3/roundtrip_test.go +++ b/internal/http3/roundtrip_test.go
@@ -337,7 +337,7 @@ for { _, err := st.readFrameHeader() if err != nil { - var code quic.StreamErrorCode + var code quic.StreamError if !errors.As(err, &code) { t.Fatalf("request stream closed with error %v: want QUIC stream error", err) } @@ -377,7 +377,7 @@ bodyw.Write(make([]byte, req.ContentLength+1)) //io.Copy(io.Discard, st) - st.wantError(quic.StreamErrorCode(errH3InternalError)) + st.wantError(quic.StreamError(errH3InternalError)) if err := rt.response().Body.Close(); err == nil { t.Fatalf("Response.Body.Close() = %v, want error", err)
diff --git a/internal/http3/server.go b/internal/http3/server.go index a1044a1..b1ca199 100644 --- a/internal/http3/server.go +++ b/internal/http3/server.go
@@ -572,7 +572,7 @@ // abort closes the connection with an error. func (sc *serverConn) abort(err error) { if e, ok := err.(*connectionError); ok { - sc.qconn.Abort(&quic.ApplicationError{ + sc.qconn.Abort(&quic.ConnectionCloseError{ Code: uint64(e.code), Reason: e.message, })
diff --git a/internal/http3/server_test.go b/internal/http3/server_test.go index 5d89858..e074cb0 100644 --- a/internal/http3/server_test.go +++ b/internal/http3/server_test.go
@@ -202,7 +202,7 @@ reqStream.writeHeadersRaw(requestHeader(tt.header)) if tt.wantError { - reqStream.wantError(quic.StreamErrorCode(errH3MessageError)) + reqStream.wantError(quic.StreamError(errH3MessageError)) } else { reqStream.wantHeaders(nil) reqStream.wantData(body) @@ -255,7 +255,7 @@ reqStream = tc.newStream(streamTypeRequest) reqStream.writeHeaders(http.Header{}) // Missing pseudo-header. - reqStream.wantError(quic.StreamErrorCode(errH3MessageError)) + reqStream.wantError(quic.StreamError(errH3MessageError)) }) } @@ -338,7 +338,7 @@ reqStream.writeHeaders(tt.header) if tt.wantError { - reqStream.wantError(quic.StreamErrorCode(errH3MessageError)) + reqStream.wantError(quic.StreamError(errH3MessageError)) } else { reqStream.wantHeaders(nil) reqStream.wantData(body) @@ -495,7 +495,7 @@ t.Errorf(`handler got Header["Host"] = %q, want unset`, h) } } else { - reqStream.wantError(quic.StreamErrorCode(errH3MessageError)) + reqStream.wantError(quic.StreamError(errH3MessageError)) } }) } @@ -1456,7 +1456,7 @@ reqStream.writeHeaders(requestHeader(http.Header{ ":path": []string{test.path}, })) - reqStream.wantError(quic.StreamErrorCode(errH3MessageError)) + reqStream.wantError(quic.StreamError(errH3MessageError)) }) } } @@ -1521,7 +1521,7 @@ reqStream.wantData([]byte("one")) time.Sleep(2 * time.Second) // T+2. synctest.Wait() - reqStream.wantError(quic.StreamErrorCode(errH3RequestCancelled)) + reqStream.wantError(quic.StreamError(errH3RequestCancelled)) }) } @@ -1588,7 +1588,7 @@ time.Sleep(3 * time.Second) // T+3. After "three" is written. reqStream.wantData([]byte("three")) time.Sleep(3 * time.Second) // T+6. After server exceeds deadline. - reqStream.wantError(quic.StreamErrorCode(errH3RequestCancelled)) + reqStream.wantError(quic.StreamError(errH3RequestCancelled)) }) } @@ -1712,7 +1712,7 @@ time.Sleep(timeout - 1) reqStream.wantIdle("timeout has not been reached") time.Sleep(1) - reqStream.wantError(quic.StreamErrorCode(errH3RequestRejected)) + reqStream.wantError(quic.StreamError(errH3RequestRejected)) if tc.nextHandlerCall() != nil { t.Error("server handler should not be called") } @@ -1815,7 +1815,7 @@ t.Errorf("Write() after timeout = %v, want os.ErrDeadlineExceeded", err) } call.exit() - reqStream.wantError(quic.StreamErrorCode(errH3RequestCancelled)) + reqStream.wantError(quic.StreamError(errH3RequestCancelled)) }) }
diff --git a/internal/http3/transport.go b/internal/http3/transport.go index c9b3a8a..3802295 100644 --- a/internal/http3/transport.go +++ b/internal/http3/transport.go
@@ -403,7 +403,7 @@ // abort closes the connection with an error. func (cc *clientConn) abort(err error) { if e, ok := err.(*connectionError); ok { - cc.qconn.Abort(&quic.ApplicationError{ + cc.qconn.Abort(&quic.ConnectionCloseError{ Code: uint64(e.code), Reason: e.message, })
diff --git a/internal/http3/transport_test.go b/internal/http3/transport_test.go index 3f1d8fe..1c3c224 100644 --- a/internal/http3/transport_test.go +++ b/internal/http3/transport_test.go
@@ -275,7 +275,7 @@ synctest.Wait() if e, ok := want.(http3Error); ok { - want = &quic.ApplicationError{Code: uint64(e)} + want = &quic.ConnectionCloseError{Code: uint64(e)} } got := tq.qconn.Wait(canceledCtx) if errors.Is(got, context.Canceled) { @@ -508,14 +508,14 @@ } } -func (ts *testQUICStream) wantError(want quic.StreamErrorCode) { +func (ts *testQUICStream) wantError(want quic.StreamError) { ts.t.Helper() synctest.Wait() _, err := ts.ReadByte() if err == nil { ts.t.Fatalf("successfully read from stream; want stream error code %v", want) } - var got quic.StreamErrorCode + var got quic.StreamError if !errors.As(err, &got) { ts.t.Fatalf("stream error = %v; want %v", err, want) }
diff --git a/quic/conn_close.go b/quic/conn_close.go index 850416a..5720dd8 100644 --- a/quic/conn_close.go +++ b/quic/conn_close.go
@@ -276,7 +276,7 @@ // Wait will return with a non-nil error after the drain period expires. // // If the peer closes the connection with a NO_ERROR transport error, Wait returns nil. -// If the peer closes the connection with an application error, Wait returns an ApplicationError +// If the peer closes the connection with an application error, Wait returns a ConnectionCloseError // containing the peer's error code and reason. // If the peer closes the connection with any other status, Wait returns a non-nil error. func (c *Conn) Wait(ctx context.Context) error { @@ -289,7 +289,7 @@ // Abort closes the connection and returns immediately. // // If err is nil, Abort sends a transport error of NO_ERROR to the peer. -// If err is an ApplicationError, Abort sends its error code and text. +// If err is a ConnectionCloseError, Abort sends its error code and text. // Otherwise, Abort sends a transport error of APPLICATION_ERROR with the error's text. func (c *Conn) Abort(err error) { if err == nil {
diff --git a/quic/conn_close_test.go b/quic/conn_close_test.go index bfb3fbf..7501153 100644 --- a/quic/conn_close_test.go +++ b/quic/conn_close_test.go
@@ -96,9 +96,7 @@ code: 20, }) - wantErr := &ApplicationError{ - Code: 20, - } + wantErr := &ConnectionCloseError{Code: 20} if _, err := waiting.result(); !errors.Is(err, wantErr) { t.Errorf("blocked conn.Wait() = %v, want %v", err, wantErr) } @@ -123,7 +121,7 @@ tc := newTestConn(t, clientSide, qr.config) tc.handshake() - wantErr := &ApplicationError{ + wantErr := &ConnectionCloseError{ Code: 42, Reason: "why?", } @@ -133,11 +131,12 @@ }) tc.wantIdle("CONN_CLOSE response not sent until user closes this side") - if err := tc.conn.Wait(canceledContext()); !errors.Is(err, wantErr) { + err := tc.conn.Wait(canceledContext()) + if e, ok := errors.AsType[*ConnectionCloseError](err); !ok || *e != *wantErr { t.Errorf("conn.Wait() = %v, want %v", err, wantErr) } - tc.conn.Abort(&ApplicationError{ + tc.conn.Abort(&ConnectionCloseError{ Code: 9, Reason: "because", }) @@ -175,7 +174,7 @@ t.Errorf("conn.Wait() = %v, want %v", err, wantErr) } - tc.conn.Abort(&ApplicationError{Code: 1}) + tc.conn.Abort(&ConnectionCloseError{Code: 1}) tc.wantFrame("CONN_CLOSE in Initial frame is APPLICATION_ERROR", packetTypeInitial, debugFrameConnectionCloseTransport{ code: errApplicationError, @@ -207,7 +206,7 @@ } // The conn has Initial and Handshake keys, so it will send CONN_CLOSE in both spaces. - tc.conn.Abort(&ApplicationError{Code: 1}) + tc.conn.Abort(&ConnectionCloseError{Code: 1}) tc.wantFrame("CONN_CLOSE in Initial frame is APPLICATION_ERROR", packetTypeInitial, debugFrameConnectionCloseTransport{ code: errApplicationError,
diff --git a/quic/conn_recv.go b/quic/conn_recv.go index 2bf127a..b8be323 100644 --- a/quic/conn_recv.go +++ b/quic/conn_recv.go
@@ -599,7 +599,7 @@ if n < 0 { return -1 } - c.handlePeerConnectionClose(now, &ApplicationError{Code: code, Reason: reason}) + c.handlePeerConnectionClose(now, &ConnectionCloseError{Code: code, Reason: reason}) return n }
diff --git a/quic/conn_send.go b/quic/conn_send.go index 25bcd3d..4033923 100644 --- a/quic/conn_send.go +++ b/quic/conn_send.go
@@ -397,7 +397,7 @@ switch e := err.(type) { case localTransportError: c.w.appendConnectionCloseTransportFrame(e.code, 0, e.reason) - case *ApplicationError: + case *ConnectionCloseError: if space != appDataSpace { // "CONNECTION_CLOSE frames signaling application errors (type 0x1d) // MUST only appear in the application data packet number space."
diff --git a/quic/errors.go b/quic/errors.go index 1226370..9b3bc58 100644 --- a/quic/errors.go +++ b/quic/errors.go
@@ -103,27 +103,27 @@ return fmt.Sprintf("peer closed connection: %v: %q", e.code, e.reason) } -// A StreamErrorCode is an application protocol error code (RFC 9000, Section 20.2) +// A StreamError is an application protocol error code (RFC 9000, Section 20.2) // indicating why a stream is being closed. -type StreamErrorCode uint64 +type StreamError uint64 -func (e StreamErrorCode) Error() string { +func (e StreamError) Error() string { return fmt.Sprintf("stream error code %v", uint64(e)) } -// An ApplicationError is an application protocol error code (RFC 9000, Section 20.2). -// Application protocol errors may be sent when terminating a stream or connection. -type ApplicationError struct { - Code uint64 - Reason string +// A ConnectionCloseError is a notice that a connection is being closed. +// It contains the information from a CONNECTION_CLOSE frame (RFC 9000, Section 19.19). +type ConnectionCloseError struct { + Code uint64 // application protocol error code (RFC 9000, Section 20.2). + Reason string // diagnostic information for the closure } -func (e *ApplicationError) Error() string { +func (e *ConnectionCloseError) Error() string { return fmt.Sprintf("peer closed connection: %v: %q", e.Code, e.Reason) } -// Is reports a match if err is an *ApplicationError with a matching Code. -func (e *ApplicationError) Is(err error) bool { - e2, ok := err.(*ApplicationError) +// Is reports a match if err is a *ConnectionCloseError with a matching Code. +func (e *ConnectionCloseError) Is(err error) bool { + e2, ok := err.(*ConnectionCloseError) return ok && e2.Code == e.Code }
diff --git a/quic/qlog.go b/quic/qlog.go index 5d2fd0f..8fa787d 100644 --- a/quic/qlog.go +++ b/quic/qlog.go
@@ -118,7 +118,7 @@ err := c.lifetime.finalErr trigger := "error" switch e := err.(type) { - case *ApplicationError: + case *ConnectionCloseError: // TODO: Distinguish between peer and locally-initiated close. trigger = "application" case localTransportError:
diff --git a/quic/stream.go b/quic/stream.go index be1da12..094de14 100644 --- a/quic/stream.go +++ b/quic/stream.go
@@ -43,15 +43,16 @@ // // The gate condition is set if a read from the stream will not block, // either because the stream has available data or because the read will fail. - ingate gate - in pipe // received data - inwin int64 // last MAX_STREAM_DATA sent to the peer - insendmax sentVal // set when we should send MAX_STREAM_DATA to the peer - inmaxbuf int64 // maximum amount of data we will buffer - insize int64 // stream final size; -1 before this is known - inset rangeset[int64] // received ranges - inclosed sentVal // set by CloseRead - inresetcode int64 // RESET_STREAM code received from the peer; -1 if not reset + ingate gate + in pipe // received data + inwin int64 // last MAX_STREAM_DATA sent to the peer + insendmax sentVal // set when we should send MAX_STREAM_DATA to the peer + inmaxbuf int64 // maximum amount of data we will buffer + insize int64 // stream final size; -1 before this is known + inset rangeset[int64] // received ranges + inclosed sentVal // set by CloseRead + inclosedcode uint64 // code to send in STOP_SENDING + inresetcode int64 // RESET_STREAM code received from the peer; -1 if not reset // outgate's lock guards send-related state. // @@ -235,7 +236,7 @@ // If the peer closes the stream cleanly, Read returns io.EOF after // returning all data sent by the peer. // If the peer aborts reads on the stream, Read returns -// an error wrapping StreamResetCode. +// an error wrapping StreamError. // // It is not safe to call Read concurrently. func (s *Stream) Read(b []byte) (n int, err error) { @@ -284,7 +285,7 @@ return 0, err } } - return 0, fmt.Errorf("stream reset by peer: %w", StreamErrorCode(s.inresetcode)) + return 0, fmt.Errorf("stream reset by peer: %w", StreamError(s.inresetcode)) } if s.inclosed.isSet() { return 0, errors.New("read from closed stream") @@ -526,7 +527,7 @@ return err } } - return fmt.Errorf("write to reset stream: %w", StreamErrorCode(s.outresetcode)) + return fmt.Errorf("write to reset stream: %w", StreamError(s.outresetcode)) } if s.outclosed.isSet() { return errors.New("write to closed stream") @@ -570,15 +571,29 @@ } // CloseRead aborts reads on the stream. +// +// It is identical to s.StopSending(0). +func (s *Stream) CloseRead() error { + return s.StopSending(0) +} + +// StopSending aborts reads on the stream. // Any blocked reads will be unblocked and return errors. // -// CloseRead notifies the peer that the stream has been closed for reading. +// StopSending sends an application protocol error code, which must be +// less than 2^62, to the peer. // It does not wait for the peer to acknowledge the closure. // Use Close to wait for the peer's acknowledgement. -func (s *Stream) CloseRead() { +// +// StopSending only returns an error if called on a write-only stream. +func (s *Stream) StopSending(code uint64) error { if s.IsWriteOnly() { - return + return errors.New("read-close of write-only stream") } + if code > quicwire.MaxVarint { + return errors.New("stream error code out of range") + } + code = min(code, quicwire.MaxVarint) s.ingate.lock() if s.inset.isrange(0, s.insize) || s.inresetcode != -1 { // We've already received all data from the peer, @@ -586,12 +601,14 @@ // This is the same as saying we sent one and they got it. s.inclosed.setReceived() } else { + s.inclosedcode = code s.inclosed.set() } discarded := s.in.end - s.in.start s.in.discardBefore(s.in.end) s.inUnlock() s.conn.handleStreamBytesReadOffLoop(discarded) // must be done with ingate unlocked + return nil } // CloseWrite aborts writes on the stream. @@ -600,51 +617,54 @@ // CloseWrite sends any data in the stream write buffer to the peer. // It does not wait for the peer to acknowledge receipt of the data. // Use Close to wait for the peer's acknowledgement. -func (s *Stream) CloseWrite() { +// +// CloseWrite only returns an error if used on a read-only stream. +func (s *Stream) CloseWrite() error { if s.IsReadOnly() { - return + return errors.New("write-close of read-only stream") } s.outgate.lock() defer s.outUnlock() s.outclosed.set() s.flushLocked() + return nil } // Reset aborts writes on the stream and notifies the peer // that the stream was terminated abruptly. // Any blocked writes will be unblocked and return errors. // -// Reset sends the application protocol error code, which must be +// Reset sends an application protocol error code, which must be // less than 2^62, to the peer. // It does not wait for the peer to acknowledge receipt of the error. // Use Close to wait for the peer's acknowledgement. // // Reset does not affect reads. // Use CloseRead to abort reads on the stream. -func (s *Stream) Reset(code uint64) { +func (s *Stream) Reset(code uint64) error { const userClosed = true - s.resetInternal(code, userClosed) + return s.resetInternal(code, userClosed) } // resetInternal resets the send side of the stream. // // If userClosed is true, this is s.Reset. // If userClosed is false, this is a reaction to a STOP_SENDING frame. -func (s *Stream) resetInternal(code uint64, userClosed bool) { +func (s *Stream) resetInternal(code uint64, userClosed bool) error { s.outgate.lock() defer s.outUnlock() if s.IsReadOnly() { - return + return errors.New("reset of read-only stream") + } + if code > quicwire.MaxVarint { + return errors.New("stream error code out of range") } if userClosed { // Mark that the user closed the stream. s.outclosed.set() } if s.outreset.isSet() { - return - } - if code > quicwire.MaxVarint { - code = quicwire.MaxVarint + return nil } // We could check here to see if the stream is closed and the // peer has acked all the data and the FIN, but sending an @@ -658,6 +678,7 @@ s.out.discardBefore(s.out.end) s.outunsent = rangeset[int64]{} s.outblocked.clear() + return nil } // connHasClosed indicates the stream's conn has closed. @@ -985,15 +1006,11 @@ // false if not everything fit in the current packet. func (s *Stream) appendInFramesLocked(w *packetWriter, pnum packetNumber, pto bool) bool { if s.inclosed.shouldSendPTO(pto) { - // We don't currently have an API for setting the error code. - // Just send zero. - code := uint64(0) - if !w.appendStopSendingFrame(s.id, code) { + if !w.appendStopSendingFrame(s.id, s.inclosedcode) { return false } s.inclosed.setSent(pnum) } - // TODO: STOP_SENDING if s.insendmax.shouldSendPTO(pto) { // MAX_STREAM_DATA maxStreamData := s.in.start + s.inmaxbuf
diff --git a/quic/stream_test.go b/quic/stream_test.go index e7f90f6..202d8eb 100644 --- a/quic/stream_test.go +++ b/quic/stream_test.go
@@ -1008,6 +1008,20 @@ } } +func TestStreamStopSendingSendErrorCode(t *testing.T) { + synctest.Test(t, testStreamStopSendingSendErrorCode) +} +func testStreamStopSendingSendErrorCode(t *testing.T) { + tc, s := newTestConnAndRemoteStream(t, serverSide, bidiStream, permissiveTransportParameters) + wantCode := uint64(42) + s.StopSending(wantCode) + tc.wantFrame("StopSending sends a STOP_SENDING frame", + packetType1RTT, debugFrameStopSending{ + id: s.id, + code: wantCode, + }) +} + func TestStreamResetBlockedStream(t *testing.T) { synctest.Test(t, testStreamResetBlockedStream) } @@ -1284,7 +1298,7 @@ finalSize: 20, code: sentCode, }) - wantErr := StreamErrorCode(sentCode) + wantErr := StreamError(sentCode) if _, err := io.ReadAll(s); !errors.Is(err, wantErr) { t.Fatalf("Read reset stream: ReadAll got error %v; want %v", err, wantErr) } @@ -1305,7 +1319,7 @@ finalSize: 20, code: sentCode, }) - wantErr := StreamErrorCode(sentCode) + wantErr := StreamError(sentCode) if n, err := reader.result(); n != 0 || !errors.Is(err, wantErr) { t.Fatalf("Read reset stream: got %v, %v; want 0, %v", n, err, wantErr) } @@ -1330,7 +1344,7 @@ finalSize: 4, code: 2, }) - wantErr := StreamErrorCode(1) + wantErr := StreamError(1) if n, err := s.Read(make([]byte, 16)); n != 0 || !errors.Is(err, wantErr) { t.Fatalf("Read from reset stream: got %v, %v; want 0, %v", n, err, wantErr) } @@ -1342,14 +1356,10 @@ } func testStreamResetInvalidCode(t *testing.T) { tc, s := newTestConnAndLocalStream(t, serverSide, uniStream, permissiveTransportParameters) - s.Reset(1 << 62) - tc.wantFrame("reset with invalid code sends a RESET_STREAM anyway", - packetType1RTT, debugFrameResetStream{ - id: s.id, - // The code we send here isn't specified, - // so this could really be any value. - code: (1 << 62) - 1, - }) + if err := s.Reset(1 << 62); err == nil { + t.Errorf("s.Reset with an invalid code succeeded; want error") + } + tc.wantIdle("reset with an invalid code has no effect") } func TestStreamResetReceiveOnly(t *testing.T) { @@ -1387,8 +1397,9 @@ code: 42, finalSize: 4, }) - if n, err := s.Write([]byte{0}); err == nil { - t.Errorf("s.Write() after STOP_SENDING = %v, %v; want error", n, err) + wantErr := StreamError(42) + if n, err := s.Write([]byte{0}); n != 0 || !errors.Is(err, wantErr) { + t.Errorf("s.Write() after STOP_SENDING = %v, %v; want error %v", n, err, wantErr) } // This ack will result in some of the previous frames being marked as lost. tc.writeAckForLatest() @@ -1486,7 +1497,7 @@ func testStreamErrorsAfterConnectionClosed(t *testing.T) { tc, s := newTestConnAndLocalStream(t, clientSide, bidiStream, permissiveTransportParameters) - wantErr := &ApplicationError{Code: 42} + wantErr := &ConnectionCloseError{Code: 42} tc.writeFrames(packetType1RTT, debugFrameConnectionCloseApplication{ code: wantErr.Code, })