stream.go

     1  // Copyright 2025 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  package http3
     6  
     7  import (
     8  	"context"
     9  	"io"
    10  	"os"
    11  	"sync"
    12  	"time"
    13  
    14  	"golang.org/x/net/quic"
    15  )
    16  
    17  // A stream wraps a QUIC stream, providing methods to read/write various values.
    18  type stream struct {
    19  	stream *quic.Stream
    20  
    21  	// lim is the current read limit.
    22  	// Reading a frame header sets the limit to the end of the frame.
    23  	// Reading past the limit or reading less than the limit and ending the frame
    24  	// results in an error.
    25  	// -1 indicates no limit.
    26  	lim int64
    27  
    28  	readDeadline  deadline
    29  	writeDeadline deadline
    30  }
    31  
    32  // newConnStream creates a new stream on a connection.
    33  // It writes the stream header for unidirectional streams.
    34  //
    35  // The stream returned by newStream is not flushed,
    36  // and will not be sent to the peer until the caller calls
    37  // Flush or writes enough data to the stream.
    38  func newConnStream(ctx context.Context, qconn *quic.Conn, stype streamType) (*stream, error) {
    39  	var qs *quic.Stream
    40  	var err error
    41  	if stype == streamTypeRequest {
    42  		// Request streams are bidirectional.
    43  		qs, err = qconn.NewStream(ctx)
    44  	} else {
    45  		// All other streams are unidirectional.
    46  		qs, err = qconn.NewSendOnlyStream(ctx)
    47  	}
    48  	if err != nil {
    49  		return nil, err
    50  	}
    51  	st := newStream(qs)
    52  	if stype != streamTypeRequest {
    53  		// Unidirectional stream header.
    54  		st.writeVarint(int64(stype))
    55  	}
    56  	return st, err
    57  }
    58  
    59  func newStream(qs *quic.Stream) *stream {
    60  	readCtx, readCancel := context.WithCancelCause(context.Background())
    61  	writeCtx, writeCancel := context.WithCancelCause(context.Background())
    62  	st := &stream{
    63  		stream: qs,
    64  		lim:    -1, // no limit
    65  		readDeadline: deadline{
    66  			ctx:    readCtx,
    67  			cancel: readCancel,
    68  		},
    69  		writeDeadline: deadline{
    70  			ctx:    writeCtx,
    71  			cancel: writeCancel,
    72  		},
    73  	}
    74  	qs.SetReadContext(readCtx)
    75  	qs.SetWriteContext(writeCtx)
    76  	return st
    77  }
    78  
    79  func (st *stream) Close() error {
    80  	st.readDeadline.stop()
    81  	st.writeDeadline.stop()
    82  	return st.stream.Close()
    83  }
    84  
    85  func (st *stream) CloseRead() {
    86  	st.readDeadline.stop()
    87  	st.stream.CloseRead()
    88  }
    89  
    90  func (st *stream) CloseWrite() {
    91  	st.writeDeadline.stop()
    92  	st.stream.CloseWrite()
    93  }
    94  
    95  func (st *stream) Reset(code uint64) {
    96  	st.readDeadline.stop()
    97  	st.writeDeadline.stop()
    98  	st.stream.Reset(code)
    99  }
   100  
   101  // deadline manages ctx, and cancels it when timer expires, with
   102  // [os.ErrDeadlineExceeded] as the cause. If the deadline is manually stopped
   103  // before timer expires, the context will be canceled with [context.Canceled]
   104  // as the cause. Once a deadline is exceeded, its timer can no longer be
   105  // extended.
   106  // Practically, this lets the http3 package support time-based deadlines by
   107  // utilizing the quic package's support for context-based deadlines.
   108  type deadline struct {
   109  	ctx    context.Context
   110  	cancel context.CancelCauseFunc
   111  
   112  	mu    sync.Mutex // Guards below.
   113  	timer *time.Timer
   114  }
   115  
   116  // stopTimerLocked stops the deadline timer and sets it to nil.
   117  // The caller must hold d.mu.
   118  func (d *deadline) stopTimerLocked() {
   119  	if d.timer != nil {
   120  		d.timer.Stop()
   121  		d.timer = nil
   122  	}
   123  }
   124  
   125  // stop stops the deadline timer and cancels the context with
   126  // [context.Canceled] as the cause.
   127  func (d *deadline) stop() {
   128  	d.mu.Lock()
   129  	d.stopTimerLocked()
   130  	d.mu.Unlock()
   131  	d.cancel(context.Canceled)
   132  }
   133  
   134  // err returns the deadline's context cancelation cause, if any.
   135  func (d *deadline) err() error {
   136  	return context.Cause(d.ctx)
   137  }
   138  
   139  // errOf returns the deadline's context cancelation cause if the given err is
   140  // non-nil. This can be used to check whether an error value returned by I/O
   141  // operations at the QUIC layer is non-nil because the deadline has expired.
   142  func (d *deadline) errOf(err error) error {
   143  	if dErr := d.err(); err != nil && dErr != nil {
   144  		return dErr
   145  	}
   146  	return err
   147  }
   148  
   149  // set configures a new deadline using the given deadlineTime.
   150  // Once deadline is exceeded, it remains in the expired (sticky) state, and
   151  // subsequent attempts to extend or reset the deadline are ignored.
   152  func (d *deadline) set(deadlineTime time.Time) {
   153  	d.mu.Lock()
   154  	defer d.mu.Unlock()
   155  
   156  	if d.ctx.Err() != nil { // Already expired, sticky error.
   157  		return
   158  	}
   159  	if deadlineTime.IsZero() {
   160  		d.stopTimerLocked()
   161  		return
   162  	}
   163  	dur := time.Until(deadlineTime)
   164  	if dur <= 0 {
   165  		d.stopTimerLocked()
   166  		d.cancel(os.ErrDeadlineExceeded)
   167  		return
   168  	}
   169  	if d.timer == nil {
   170  		d.timer = time.AfterFunc(dur, func() {
   171  			d.cancel(os.ErrDeadlineExceeded)
   172  		})
   173  	} else {
   174  		d.timer.Reset(dur)
   175  	}
   176  }
   177  
   178  // readFrameHeader reads the type and length fields of an HTTP/3 frame.
   179  // It sets the read limit to the end of the frame.
   180  //
   181  // https://www.rfc-editor.org/rfc/rfc9114.html#section-7.1
   182  func (st *stream) readFrameHeader() (ftype frameType, err error) {
   183  	if st.lim >= 0 {
   184  		// We shouldn't call readFrameHeader before ending the previous frame.
   185  		return 0, errH3FrameError
   186  	}
   187  	ftype, err = readVarint[frameType](st)
   188  	if err != nil {
   189  		return 0, err
   190  	}
   191  	size, err := st.readVarint()
   192  	if err != nil {
   193  		return 0, err
   194  	}
   195  	st.lim = size
   196  	return ftype, nil
   197  }
   198  
   199  // endFrame is called after reading a frame to reset the read limit.
   200  // It returns an error if the entire contents of a frame have not been read.
   201  func (st *stream) endFrame() error {
   202  	if st.lim != 0 {
   203  		return &connectionError{
   204  			code:    errH3FrameError,
   205  			message: "invalid HTTP/3 frame",
   206  		}
   207  	}
   208  	st.lim = -1
   209  	return nil
   210  }
   211  
   212  // readFrameData returns the remaining data in the current frame.
   213  func (st *stream) readFrameData() ([]byte, error) {
   214  	if st.lim < 0 {
   215  		return nil, errH3FrameError
   216  	}
   217  	// TODO: Pool buffers to avoid allocation here.
   218  	b := make([]byte, st.lim)
   219  	_, err := io.ReadFull(st, b)
   220  	if err != nil {
   221  		return nil, err
   222  	}
   223  	return b, nil
   224  }
   225  
   226  // ReadByte reads one byte from the stream.
   227  func (st *stream) ReadByte() (b byte, err error) {
   228  	// Check the deadline before doing I/O operations on the QUIC layer. We do
   229  	// this because the QUIC layer implements a fast path for I/O operations,
   230  	// allowing Read & Write to succeed depending on the state of buffer, even
   231  	// if its context has been canceled. By always checking the deadline here,
   232  	// we make it so that I/O operations fail as soon as its relevant deadline
   233  	// has been exceeded.
   234  	if err := st.readDeadline.err(); err != nil {
   235  		return 0, err
   236  	}
   237  	if err := st.recordBytesRead(1); err != nil {
   238  		return 0, err
   239  	}
   240  	b, err = st.stream.ReadByte()
   241  	if err == io.EOF && st.lim >= 0 {
   242  		return 0, errH3FrameError
   243  	}
   244  	return b, st.readDeadline.errOf(err)
   245  }
   246  
   247  // Read reads from the stream.
   248  func (st *stream) Read(b []byte) (int, error) {
   249  	// Check the deadline before doing I/O operations on the QUIC layer. We do
   250  	// this because the QUIC layer implements a fast path for I/O operations,
   251  	// allowing Read & Write to succeed depending on the state of buffer, even
   252  	// if its context has been canceled. By always checking the deadline here,
   253  	// we make it so that I/O operations fail as soon as its relevant deadline
   254  	// has been exceeded.
   255  	if err := st.readDeadline.err(); err != nil {
   256  		return 0, err
   257  	}
   258  	n, err := st.stream.Read(b)
   259  	if e2 := st.recordBytesRead(n); e2 != nil {
   260  		return 0, e2
   261  	}
   262  	if err == io.EOF {
   263  		if st.lim == 0 {
   264  			// EOF at end of frame, ignore.
   265  			return n, nil
   266  		} else if st.lim > 0 {
   267  			// EOF inside frame, error.
   268  			return 0, errH3FrameError
   269  		} else {
   270  			// EOF outside of frame, surface to caller.
   271  			return n, io.EOF
   272  		}
   273  	}
   274  	return n, st.readDeadline.errOf(err)
   275  }
   276  
   277  // discardUnknownFrame discards an unknown frame.
   278  //
   279  // HTTP/3 requires that unknown frames be ignored on all streams.
   280  // However, a known frame appearing in an unexpected place is a fatal error,
   281  // so this returns an error if the frame is one we know.
   282  func (st *stream) discardUnknownFrame(ftype frameType) error {
   283  	switch ftype {
   284  	case frameTypeData,
   285  		frameTypeHeaders,
   286  		frameTypeCancelPush,
   287  		frameTypeSettings,
   288  		frameTypePushPromise,
   289  		frameTypeGoaway,
   290  		frameTypeMaxPushID:
   291  		return &connectionError{
   292  			code:    errH3FrameUnexpected,
   293  			message: "unexpected " + ftype.String() + " frame",
   294  		}
   295  	}
   296  	return st.discardFrame()
   297  }
   298  
   299  // discardFrame discards any remaining data in the current frame and resets the read limit.
   300  func (st *stream) discardFrame() error {
   301  	// TODO: Consider adding a *quic.Stream method to discard some amount of data.
   302  	for range st.lim {
   303  		_, err := st.ReadByte()
   304  		if err != nil {
   305  			return &streamError{errH3FrameError, err.Error()}
   306  		}
   307  	}
   308  	st.lim = -1
   309  	return nil
   310  }
   311  
   312  // Write writes to the stream.
   313  func (st *stream) Write(b []byte) (int, error) {
   314  	// Check the deadline before doing I/O operations on the QUIC layer. We do
   315  	// this because the QUIC layer implements a fast path for I/O operations,
   316  	// allowing Read & Write to succeed depending on the state of buffer, even
   317  	// if its context has been canceled. By always checking the deadline here,
   318  	// we make it so that I/O operations fail as soon as its relevant deadline
   319  	// has been exceeded.
   320  	if err := st.writeDeadline.err(); err != nil {
   321  		return 0, err
   322  	}
   323  	n, err := st.stream.Write(b)
   324  	return n, st.writeDeadline.errOf(err)
   325  }
   326  
   327  // Flush commits data written to the stream.
   328  func (st *stream) Flush() error {
   329  	// Check the deadline before doing I/O operations on the QUIC layer. We do
   330  	// this because the QUIC layer implements a fast path for I/O operations,
   331  	// allowing Read & Write to succeed depending on the state of buffer, even
   332  	// if its context has been canceled. By always checking the deadline here,
   333  	// we make it so that I/O operations fail as soon as its relevant deadline
   334  	// has been exceeded.
   335  	if err := st.writeDeadline.err(); err != nil {
   336  		return err
   337  	}
   338  	return st.writeDeadline.errOf(st.stream.Flush())
   339  }
   340  
   341  // WriteByte writes one byte to the stream.
   342  func (st *stream) WriteByte(c byte) error {
   343  	// Check the deadline before doing I/O operations on the QUIC layer. We do
   344  	// this because the QUIC layer implements a fast path for I/O operations,
   345  	// allowing Read & Write to succeed depending on the state of buffer, even
   346  	// if its context has been canceled. By always checking the deadline here,
   347  	// we make it so that I/O operations fail as soon as its relevant deadline
   348  	// has been exceeded.
   349  	if err := st.writeDeadline.err(); err != nil {
   350  		return err
   351  	}
   352  	return st.writeDeadline.errOf(st.stream.WriteByte(c))
   353  }
   354  
   355  // readVarint reads a QUIC variable-length integer from the stream.
   356  func (st *stream) readVarint() (v int64, err error) {
   357  	b, err := st.ReadByte()
   358  	if err != nil {
   359  		return 0, err
   360  	}
   361  	v = int64(b & 0x3f)
   362  	n := 1 << (b >> 6)
   363  	for i := 1; i < n; i++ {
   364  		b, err := st.ReadByte()
   365  		if err != nil {
   366  			if err == io.EOF {
   367  				return 0, errH3FrameError
   368  			}
   369  			return 0, err
   370  		}
   371  		v = (v << 8) | int64(b)
   372  	}
   373  	return v, nil
   374  }
   375  
   376  // readVarint reads a varint of a particular type.
   377  func readVarint[T ~int64 | ~uint64](st *stream) (T, error) {
   378  	v, err := st.readVarint()
   379  	return T(v), err
   380  }
   381  
   382  // writeVarint writes a QUIC variable-length integer to the stream.
   383  func (st *stream) writeVarint(v int64) {
   384  	switch {
   385  	case v <= (1<<6)-1:
   386  		st.WriteByte(byte(v))
   387  	case v <= (1<<14)-1:
   388  		st.WriteByte((1 << 6) | byte(v>>8))
   389  		st.WriteByte(byte(v))
   390  	case v <= (1<<30)-1:
   391  		st.WriteByte((2 << 6) | byte(v>>24))
   392  		st.WriteByte(byte(v >> 16))
   393  		st.WriteByte(byte(v >> 8))
   394  		st.WriteByte(byte(v))
   395  	case v <= (1<<62)-1:
   396  		st.WriteByte((3 << 6) | byte(v>>56))
   397  		st.WriteByte(byte(v >> 48))
   398  		st.WriteByte(byte(v >> 40))
   399  		st.WriteByte(byte(v >> 32))
   400  		st.WriteByte(byte(v >> 24))
   401  		st.WriteByte(byte(v >> 16))
   402  		st.WriteByte(byte(v >> 8))
   403  		st.WriteByte(byte(v))
   404  	default:
   405  		panic("varint too large")
   406  	}
   407  }
   408  
   409  // recordBytesRead records that n bytes have been read.
   410  // It returns an error if the read passes the current limit.
   411  func (st *stream) recordBytesRead(n int) error {
   412  	if st.lim < 0 {
   413  		return nil
   414  	}
   415  	st.lim -= int64(n)
   416  	if st.lim < 0 {
   417  		st.stream = nil // panic if we try to read again
   418  		return &connectionError{
   419  			code:    errH3FrameError,
   420  			message: "invalid HTTP/3 frame",
   421  		}
   422  	}
   423  	return nil
   424  }
   425
View as plain text