internal/http3: clean up stream I/O methods usages in tests

In CL 796400, I/O methods were added to the internal/http3.stream. As a
follow-up to that CL, modify our tests to stop calling I/O methods from
internal/quic.stream.

Also add internal/http.stream.WriteByte method for symmetry.

For golang/go#70914

Change-Id: I2036589d89ceac97f80e3f34a29cea786a6a6964
Reviewed-on: https://go-review.googlesource.com/c/net/+/797280
LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>
Reviewed-by: Nicholas Husin <husin@google.com>
Reviewed-by: Damien Neil <dneil@google.com>

Author: nicholashusin

internal/http3/body_test.go

@@ -224,7 +224,7 @@ func TestReadData(t *testing.T) {
 				case receiveData:
 					t.Logf("receive DATA frame content: size=%v", step.size)
 					for range step.size {
-						st.stream.stream.WriteByte(byte(bytesSent))
+						st.WriteByte(byte(bytesSent))
 						bytesSent++
 					}
 					st.Flush()
@@ -239,7 +239,7 @@ func TestReadData(t *testing.T) {
 					st.Flush()
 				case receiveEOF:
 					t.Logf("receive EOF on request stream")
-					st.stream.stream.CloseWrite()
+					st.CloseWrite()
 				case wantBody:
 					t.Logf("read %v bytes from response body", step.size)
 					want := make([]byte, step.size)

internal/http3/conn_test.go

@@ -132,7 +132,7 @@ func TestConnPeerCreatesBadUnidirectionalStream(t *testing.T) {
 			t.Fatal(err)
 		}
 		st := newTestQUICStream(tc.t, newStream(qs))
-		st.stream.stream.Close()
+		st.Close()
 
 		tc.wantClosed("after peer creates and closes uni stream", errH3StreamCreationError)
 	})

internal/http3/roundtrip_test.go

@@ -429,7 +429,7 @@ func TestRoundTripExpect100Continue(t *testing.T) {
 			":status": []string{"200"},
 		})
 		st.writeData(serverBody)
-		st.stream.stream.CloseWrite()
+		st.CloseWrite()
 
 		// Client receives the response from server.
 		rt.wantStatus(200)
@@ -478,7 +478,7 @@ func TestRoundTripExpect100ContinueRejected(t *testing.T) {
 		st.wantIdle("client does not send its body without getting status 100")
 		serverBody := []byte("server's body")
 		st.writeData(serverBody)
-		st.stream.stream.CloseWrite()
+		st.CloseWrite()
 
 		rt.wantStatus(200)
 		rt.wantBody(serverBody)
@@ -518,7 +518,7 @@ func TestRoundTripReadRespWithNoBody(t *testing.T) {
 		st.writeHeaders(http.Header{
 			":status": {"200"},
 		})
-		st.stream.stream.CloseWrite()
+		st.CloseWrite()
 		rt.wantStatus(200)
 		st.wantClosed("request is complete")
 
@@ -760,7 +760,7 @@ func TestRoundTrip103EarlyHints(t *testing.T) {
 		})
 		body := []byte("some body")
 		st.writeData(body)
-		st.stream.stream.CloseWrite()
+		st.CloseWrite()
 
 		rt.wantStatus(200)
 		rt.wantBody(body)
@@ -810,7 +810,7 @@ func TestRoundTripGzipEnabled(t *testing.T) {
 				"content-length":   []string{strconv.Itoa(buf.Len())},
 			})
 			st.writeData(buf.Bytes())
-			st.stream.stream.CloseWrite()
+			st.CloseWrite()
 
 			rt.wantStatus(200)
 
@@ -949,7 +949,7 @@ func TestRoundTripGzipWithTrailers(t *testing.T) {
 		st.writeHeaders(http.Header{
 			"server-trailer-a": {"valuea"},
 		})
-		st.stream.stream.CloseWrite()
+		st.CloseWrite()
 
 		rt.wantStatus(200)
 		rt.wantTrailers(http.Header{

internal/http3/server_test.go

@@ -429,7 +429,7 @@ func TestServerBody(t *testing.T) {
 		reqStream.writeHeaders(requestHeader(nil))
 		bodyContent := []byte("some body content that should be echoed")
 		reqStream.writeData(bodyContent)
-		reqStream.stream.stream.CloseWrite()
+		reqStream.CloseWrite()
 		reqStream.wantSomeHeaders(http.Header{":status": {"200"}})
 		// Small multiple calls to Write will be coalesced into one DATA frame.
 		reqStream.wantData(append([]byte("/"), bodyContent...))
@@ -672,7 +672,7 @@ func TestServerExpect100Continue(t *testing.T) {
 		reqStream.wantSomeHeaders(http.Header{":status": {"100"}})
 		body := []byte("body that will be echoed back if we get status 100")
 		reqStream.writeData(body)
-		reqStream.stream.stream.CloseWrite()
+		reqStream.CloseWrite()
 
 		// Receive the server's response after sending the body.
 		reqStream.wantSomeHeaders(http.Header{":status": {"200"}})
@@ -726,7 +726,7 @@ func TestServerNoExpect100ContinueAfterNormalResponse(t *testing.T) {
 		// read the request body without hanging, which would normally cause an
 		// HTTP 100 to be sent.
 		reqStream.writeData([]byte("some body"))
-		reqStream.stream.stream.CloseWrite()
+		reqStream.CloseWrite()
 
 		// Verify that no HTTP 100 was sent.
 		reqStream.wantSomeHeaders(http.Header{":status": {"200"}})
@@ -750,7 +750,7 @@ func TestServerHandlerReadReqWithNoBody(t *testing.T) {
 		// client closes the write direction of the stream.
 		reqStream := tc.newStream(streamTypeRequest)
 		reqStream.writeHeaders(requestHeader(nil))
-		reqStream.stream.stream.CloseWrite()
+		reqStream.CloseWrite()
 		reqStream.wantSomeHeaders(http.Header{":status": {"200"}})
 		reqStream.wantData(serverBody)
 		reqStream.wantClosed("request is complete")

internal/http3/stream.go

@@ -338,6 +338,20 @@ func (st *stream) Flush() error {
 	return st.writeDeadline.errOf(st.stream.Flush())
 }
 
+// WriteByte writes one byte to the stream.
+func (st *stream) WriteByte(c byte) error {
+	// Check the deadline before doing I/O operations on the QUIC layer. We do
+	// this because the QUIC layer implements a fast path for I/O operations,
+	// allowing Read & Write to succeed depending on the state of buffer, even
+	// if its context has been canceled. By always checking the deadline here,
+	// we make it so that I/O operations fail as soon as its relevant deadline
+	// has been exceeded.
+	if err := st.writeDeadline.err(); err != nil {
+		return err
+	}
+	return st.writeDeadline.errOf(st.stream.WriteByte(c))
+}
+
 // readVarint reads a QUIC variable-length integer from the stream.
 func (st *stream) readVarint() (v int64, err error) {
 	b, err := st.ReadByte()
@@ -369,24 +383,24 @@ func readVarint[T ~int64 | ~uint64](st *stream) (T, error) {
 func (st *stream) writeVarint(v int64) {
 	switch {
 	case v <= (1<<6)-1:
-		st.stream.WriteByte(byte(v))
+		st.WriteByte(byte(v))
 	case v <= (1<<14)-1:
-		st.stream.WriteByte((1 << 6) | byte(v>>8))
-		st.stream.WriteByte(byte(v))
+		st.WriteByte((1 << 6) | byte(v>>8))
+		st.WriteByte(byte(v))
 	case v <= (1<<30)-1:
-		st.stream.WriteByte((2 << 6) | byte(v>>24))
-		st.stream.WriteByte(byte(v >> 16))
-		st.stream.WriteByte(byte(v >> 8))
-		st.stream.WriteByte(byte(v))
+		st.WriteByte((2 << 6) | byte(v>>24))
+		st.WriteByte(byte(v >> 16))
+		st.WriteByte(byte(v >> 8))
+		st.WriteByte(byte(v))
 	case v <= (1<<62)-1:
-		st.stream.WriteByte((3 << 6) | byte(v>>56))
-		st.stream.WriteByte(byte(v >> 48))
-		st.stream.WriteByte(byte(v >> 40))
-		st.stream.WriteByte(byte(v >> 32))
-		st.stream.WriteByte(byte(v >> 24))
-		st.stream.WriteByte(byte(v >> 16))
-		st.stream.WriteByte(byte(v >> 8))
-		st.stream.WriteByte(byte(v))
+		st.WriteByte((3 << 6) | byte(v>>56))
+		st.WriteByte(byte(v >> 48))
+		st.WriteByte(byte(v >> 40))
+		st.WriteByte(byte(v >> 32))
+		st.WriteByte(byte(v >> 24))
+		st.WriteByte(byte(v >> 16))
+		st.WriteByte(byte(v >> 8))
+		st.WriteByte(byte(v))
 	default:
 		panic("varint too large")
 	}

internal/http3/stream_test.go

@@ -193,7 +193,7 @@ func TestStreamReadFrameHeaderPartial(t *testing.T) {
 		if err := st1.Flush(); err != nil {
 			t.Fatal(err)
 		}
-		st1.stream.CloseWrite()
+		st1.CloseWrite()
 
 		if _, err := st2.readFrameHeader(); err == nil {
 			t.Fatalf("%v/%v bytes of frame available: st.readFrameHeader() succeeded; want error", i, len(frame))
@@ -206,7 +206,7 @@ func TestStreamReadFrameDataPartial(t *testing.T) {
 	st1.writeVarint(1)          // type
 	st1.writeVarint(100)        // size
 	st1.Write(make([]byte, 50)) // data
-	st1.stream.CloseWrite()
+	st1.CloseWrite()
 	if _, err := st2.readFrameHeader(); err != nil {
 		t.Fatalf("st.readFrameHeader() = %v", err)
 	}
@@ -219,7 +219,7 @@ func TestStreamReadByteFrameDataPartial(t *testing.T) {
 	st1, st2 := newStreamPair(t)
 	st1.writeVarint(1)   // type
 	st1.writeVarint(100) // size
-	st1.stream.CloseWrite()
+	st1.CloseWrite()
 	if _, err := st2.readFrameHeader(); err != nil {
 		t.Fatalf("st.readFrameHeader() = %v", err)
 	}
@@ -242,7 +242,7 @@ func TestStreamReadFrameDataAtEOF(t *testing.T) {
 	}
 
 	st1.Write(data)         // data
-	st1.stream.CloseWrite() // end stream
+	st1.CloseWrite() // end stream
 	got := make([]byte, len(data)+1)
 	if n, err := st2.Read(got); err != nil || n != len(data) || !bytes.Equal(got[:n], data) {
 		t.Fatalf("st.Read() = %v, %v (data=%x); want %v, nil (data=%x)", n, err, got[:n], len(data), data)
@@ -297,7 +297,7 @@ func TestStreamDiscardFrame(t *testing.T) {
 	st1.writeVarint(typ)              // type
 	st1.writeVarint(int64(len(data))) // size
 	st1.Write(data)                   // data
-	st1.stream.CloseWrite()
+	st1.CloseWrite()
 
 	if got, err := st2.readFrameHeader(); err != nil || got != typ {
 		t.Fatalf("st.readFrameHeader() = %v, %v; want %v, nil", got, err, typ)

internal/http3/transport_test.go

@@ -459,7 +459,7 @@ func (ts *testQUICStream) wantClosed(reason string) {
 func (ts *testQUICStream) wantError(want quic.StreamErrorCode) {
 	ts.t.Helper()
 	synctest.Wait()
-	_, err := ts.stream.stream.ReadByte()
+	_, err := ts.ReadByte()
 	if err == nil {
 		ts.t.Fatalf("successfully read from stream; want stream error code %v", want)
 	}