internal/http3: implement server read and write deadlines

The HTTP/3 server now enforces the net/http server's ReadHeaderTimeout,
ReadTimeout, and WriteTimeout that is registered via RegisterServer.

Also clean up logic related to stream closure and reset in the server
implementation. Instead, just return the appropriate error and let
genericConn.handleStreamError deal with the stream closure and reset.

IdleTimeout is not yet implemented and will be added separately.

For golang/go#70914

Change-Id: Ie7aaebb673ad6a6386c6cbce273446466a6a6964
Reviewed-on: https://go-review.googlesource.com/c/net/+/798060
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: Damien Neil <dneil@google.com>
Auto-Submit: Nicholas Husin <husin@google.com>

internal/http3/server.go

diff --git a/internal/http3/server.go b/internal/http3/server.go
index 999b3f1..bfa31b0 100644
--- a/internal/http3/server.go
+++ b/internal/http3/server.go
@@ -29,10 +29,9 @@
 // The zero value for server is a valid server.
 type server struct {
 	// handler to invoke for requests, http.DefaultServeMux if nil.
-	handler http.Handler
-
-	config *quic.Config
-
+	handler    http.Handler
+	config     *quic.Config
+	srv1       *http.Server
 	listenQUIC func(addr string, config *quic.Config) (*quic.Endpoint, error)
 
 	initOnce sync.Once
@@ -101,6 +100,7 @@
 		s3 := &server{
 			config:     opts.QUICConfig,
 			listenQUIC: opts.ListenQUIC,
+			srv1:       s,
 			handler:    stdHandler,
 			serveCtx:   stdHandler.BaseContext(),
 		}
@@ -154,7 +154,7 @@
 		if err != nil {
 			return err
 		}
-		go s.newServerConn(qconn, s.handler)
+		go s.newServerConn(qconn)
 	}
 }
 
@@ -225,13 +225,42 @@
 	}
 }
 
+func (s *server) readHeaderTimeout() time.Duration {
+	if s.srv1 == nil || s.srv1.ReadHeaderTimeout == 0 {
+		return s.readTimeout()
+	}
+	return s.srv1.ReadHeaderTimeout
+}
+
+func (s *server) readTimeout() time.Duration {
+	if s.srv1 == nil {
+		return 0
+	}
+	return s.srv1.ReadTimeout
+}
+
+func (s *server) writeTimeout() time.Duration {
+	if s.srv1 == nil {
+		return 0
+	}
+	return s.srv1.WriteTimeout
+}
+
+// TODO: this is currently unused, enforce it.
+func (s *server) idleTimeout() time.Duration {
+	if s.srv1 == nil || s.srv1.IdleTimeout == 0 {
+		return s.readTimeout()
+	}
+	return s.srv1.IdleTimeout
+}
+
 type serverConn struct {
 	qconn *quic.Conn
+	srv   *server
 
 	genericConn // for handleUnidirectionalStream
 	enc         qpackEncoder
 	dec         qpackDecoder
-	handler     http.Handler
 
 	// For handling shutdown.
 	controlStream      *stream
@@ -240,10 +269,10 @@
 	goawaySent         bool
 }
 
-func (s *server) newServerConn(qconn *quic.Conn, handler http.Handler) {
+func (s *server) newServerConn(qconn *quic.Conn) {
 	sc := &serverConn{
-		qconn:   qconn,
-		handler: handler,
+		qconn: qconn,
+		srv:   s,
 	}
 	s.registerConn(sc)
 	defer s.unregisterConn(sc)
@@ -480,11 +509,27 @@
 			message: "GOAWAY request with equal or lower ID than the stream has been sent",
 		}
 	}
+
+	readStartTime := time.Now()
+	if t := sc.srv.readHeaderTimeout(); t > 0 {
+		st.readDeadline.set(readStartTime.Add(t))
+	}
 	header, pHeader, err := sc.parseHeader(st)
 	if err != nil {
+		if errors.Is(err, os.ErrDeadlineExceeded) {
+			return &streamError{
+				code:    errH3RequestRejected,
+				message: "exceeded deadline while parsing header",
+			}
+		}
 		return err
 	}
 
+	if t := sc.srv.readTimeout(); t > 0 {
+		st.readDeadline.set(readStartTime.Add(t))
+	} else {
+		st.readDeadline.set(time.Time{})
+	}
 	reqInfo := httpcommon.NewServerRequest(httpcommon.ServerRequestParam{
 		Method:    pHeader.method,
 		Scheme:    pHeader.scheme,
@@ -544,7 +589,6 @@
 			enc:    &sc.enc,
 		},
 	}
-	defer rw.close()
 	if reqInfo.NeedsContinue {
 		req.Body.(*bodyReader).send100Continue = func() {
 			rw.WriteHeader(100)
@@ -552,8 +596,11 @@
 	}
 
 	// TODO: handle panic coming from the HTTP handler.
-	sc.handler.ServeHTTP(rw, req)
-	return nil
+	if t := sc.srv.writeTimeout(); t > 0 {
+		st.writeDeadline.set(time.Now().Add(t))
+	}
+	sc.srv.handler.ServeHTTP(rw, req)
+	return rw.close()
 }
 
 // abort closes the connection with an error.
@@ -847,19 +894,25 @@
 }
 
 func (rw *responseWriter) close() error {
+	if errors.Is(rw.st.writeDeadline.err(), os.ErrDeadlineExceeded) {
+		return &streamError{
+			code:    errH3RequestCancelled,
+			message: "exceeded deadline while writing response",
+		}
+	}
+
 	retErr := rw.FlushError()
 	rw.mu.Lock()
 	defer rw.mu.Unlock()
-
 	rw.prepareTrailerForWriteLocked()
 	if err := rw.bw.Close(); retErr == nil {
 		retErr = err
 	}
-
-	if errors.Is(rw.st.writeDeadline.err(), os.ErrDeadlineExceeded) {
-		rw.st.Reset(uint64(errH3RequestCancelled))
-	} else if err := rw.st.Close(); retErr == nil {
-		retErr = err
+	if errors.Is(retErr, os.ErrDeadlineExceeded) {
+		return &streamError{
+			code:    errH3RequestCancelled,
+			message: retErr.Error(),
+		}
 	}
 	return retErr
 }

internal/http3/server_test.go

diff --git a/internal/http3/server_test.go b/internal/http3/server_test.go
index 4908471..138a878 100644
--- a/internal/http3/server_test.go
+++ b/internal/http3/server_test.go
@@ -16,6 +16,7 @@
 	"reflect"
 	"slices"
 	"strconv"
+	"sync"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -1449,6 +1450,163 @@
 	})
 }
 
+func TestServerReadHeaderTimeout(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		timeout := 10 * time.Second
+		ts := newTestServer(t, nil)
+		ts.s.srv1.ReadHeaderTimeout = timeout
+		tc := ts.connect()
+		tc.greet()
+
+		// Write some part of the header, but never finish sending it.
+		reqStream := tc.newStream(streamTypeRequest)
+		reqStream.writeVarint(int64(frameTypeHeaders))
+		if err := reqStream.Flush(); err != nil {
+			t.Fatalf("Flush() = %v, want nil", err)
+		}
+
+		// A stream error should be sent to the client as soon as the timeout
+		// is reached. Server handler should not be called.
+		time.Sleep(timeout - 1)
+		reqStream.wantIdle("timeout has not been reached")
+		time.Sleep(1)
+		reqStream.wantError(quic.StreamErrorCode(errH3RequestRejected))
+		if tc.nextHandlerCall() != nil {
+			t.Error("server handler should not be called")
+		}
+	})
+}
+
+func TestServerReadTimeout(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		timeout := 10 * time.Second
+		ts := newTestServer(t, nil)
+		ts.s.srv1.ReadTimeout = timeout
+		tc := ts.connect()
+		tc.greet()
+
+		reqStream := tc.newStream(streamTypeRequest)
+		reqStream.writeHeaders(requestHeader(nil))
+		reqStream.writeData([]byte("some body"))
+		call := tc.nextHandlerCall()
+
+		// Read within the server handler should succeed prior to timeout.
+		time.Sleep(timeout - 1)
+		synctest.Wait()
+		if _, err := call.req.Body.Read(make([]byte, 1)); err != nil {
+			t.Errorf("Read() before timeout = %v, want nil", err)
+		}
+
+		// Read within the server handler should fail once timeout is reached.
+		// Stream error should not be sent to the client, as it is up to the
+		// server handler to decide how it wants to deal with its inability to
+		// read the request body.
+		time.Sleep(1)
+		synctest.Wait()
+		if _, err := call.req.Body.Read(make([]byte, 1)); !errors.Is(err, os.ErrDeadlineExceeded) {
+			t.Errorf("Read() after timeout = %v, want os.ErrDeadlineExceeded", err)
+		}
+		call.w.Write([]byte("some body"))
+		call.exit()
+		reqStream.wantSomeHeaders(http.Header{":status": {"200"}})
+		reqStream.wantData([]byte("some body"))
+		reqStream.wantClosed("clean close expected")
+	})
+}
+
+func TestServerReadTimeoutInProgress(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		timeout := 10 * time.Second
+		ts := newTestServer(t, nil)
+		ts.s.srv1.ReadTimeout = timeout
+		tc := ts.connect()
+		tc.greet()
+
+		reqStream := tc.newStream(streamTypeRequest)
+		reqStream.writeHeaders(requestHeader(nil))
+		reqStream.Flush()
+		call := tc.nextHandlerCall()
+
+		// Read will block due to the client having sent no body, thus
+		// advancing synctest's time.
+		start := time.Now()
+		_, err := call.req.Body.Read(make([]byte, 1))
+		if !errors.Is(err, os.ErrDeadlineExceeded) {
+			t.Errorf("Read error = %v, want os.ErrDeadlineExceeded", err)
+		}
+		if got, want := time.Since(start), timeout; got != want {
+			t.Errorf("Read blocked for %v, want %v", got, want)
+		}
+		call.exit()
+	})
+}
+
+func TestServerWriteTimeout(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		timeout := 10 * time.Second
+		ts := newTestServer(t, nil)
+		ts.s.srv1.WriteTimeout = timeout
+		tc := ts.connect()
+		tc.greet()
+
+		reqStream := tc.newStream(streamTypeRequest)
+		reqStream.writeHeaders(requestHeader(nil))
+		call := tc.nextHandlerCall()
+		body := make([]byte, defaultBodyBufferCap+1)
+
+		// Write within the server handler should succeed prior to timeout.
+		time.Sleep(timeout - 1)
+		synctest.Wait()
+		if _, err := call.w.Write(body); err != nil {
+			t.Errorf("Write() before timeout = %v, want nil", err)
+		}
+		call.w.(http.Flusher).Flush()
+		reqStream.wantSomeHeaders(http.Header{":status": {"200"}})
+		reqStream.wantData(body)
+		reqStream.wantIdle("timeout has not been reached")
+
+		// Write within the server handler should fail once timeout is reached.
+		// A stream error should also be sent to the client.
+		time.Sleep(1)
+		synctest.Wait()
+		if _, err := call.w.Write(body); !errors.Is(err, os.ErrDeadlineExceeded) {
+			t.Errorf("Write() after timeout = %v, want os.ErrDeadlineExceeded", err)
+		}
+		call.exit()
+		reqStream.wantError(quic.StreamErrorCode(errH3RequestCancelled))
+	})
+}
+
+func TestServerWriteTimeoutInProgress(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		timeout := 10 * time.Second
+		ts := newTestServer(t, nil)
+		ts.s.srv1.WriteTimeout = timeout
+		tc := ts.connect()
+		tc.greet()
+
+		reqStream := tc.newStream(streamTypeRequest)
+		reqStream.writeHeaders(requestHeader(nil))
+		reqStream.Flush()
+		call := tc.nextHandlerCall()
+
+		// Keep writing body endlessly. Eventually, it will get blocked due to
+		// flow control, and start advancing synctest's time.
+		start := time.Now()
+		var err error
+		for err == nil {
+			_, err = call.w.Write([]byte("endless body"))
+		}
+		if !errors.Is(err, os.ErrDeadlineExceeded) {
+			t.Errorf("Write error = %v, want os.ErrDeadlineExceeded", err)
+		}
+		if got, want := time.Since(start), timeout; got != want {
+			t.Errorf("Write blocked for %v, want %v", got, want)
+		}
+		call.exit()
+	})
+}
+
 type testServer struct {
 	t  testing.TB
 	s  *server
@@ -1467,19 +1625,41 @@
 	ts *testServer
 
 	*testQUICConn
-	control *testQUICStream
+	control   *testQUICStream
+	localAddr netip.AddrPort
+}
+
+type testServerHandler struct {
+	ts      *testServer
+	callsMu sync.Mutex
+	calls   []*serverHandlerCall
+}
+
+// serverHandlerCall is a call to testServerHandler's ServeHTTP method.
+type serverHandlerCall struct {
+	w         http.ResponseWriter
+	req       *http.Request
+	closeOnce sync.Once
+	ch        chan func()
 }
 
 func newTestServer(t testing.TB, handler http.Handler) *testServer {
 	t.Helper()
 	ts := &testServer{
 		t: t,
-		s: &server{
-			config: &quic.Config{
-				TLSConfig: testTLSConfig,
-			},
-			handler: handler,
+	}
+	if handler == nil {
+		handler = &testServerHandler{
+			ts:    ts,
+			calls: []*serverHandlerCall{},
+		}
+	}
+	ts.s = &server{
+		config: &quic.Config{
+			TLSConfig: testTLSConfig,
 		},
+		srv1:    &http.Server{},
+		handler: handler,
 	}
 	e := ts.tn.newQUICEndpoint(t, ts.s.config)
 	ts.addr = e.LocalAddr()
@@ -1498,6 +1678,7 @@
 	tc := &testServerConn{
 		ts:           ts,
 		testQUICConn: newTestQUICConn(ts.t, qconn),
+		localAddr:    e.LocalAddr(),
 	}
 	synctest.Wait()
 	return tc
@@ -1512,3 +1693,55 @@
 	tc.control.Flush()
 	synctest.Wait()
 }
+
+// nextHandlerCall returns the next handler call that has been initiated by tc.
+// If there is no handler call, nil is returned.
+func (tc *testServerConn) nextHandlerCall() *serverHandlerCall {
+	h, ok := tc.ts.s.handler.(*testServerHandler)
+	if !ok {
+		tc.t.Fatal("nextHandlerCall is called for a testServer with non-nil handler")
+	}
+	tc.t.Helper()
+	synctest.Wait()
+	h.callsMu.Lock()
+	defer h.callsMu.Unlock()
+	for i, call := range h.calls {
+		if call.req.RemoteAddr == tc.localAddr.String() {
+			h.calls = append(h.calls[:i], h.calls[i+1:]...)
+			return call
+		}
+	}
+	return nil
+}
+
+func (h *testServerHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	call := &serverHandlerCall{
+		w:   w,
+		req: req,
+		ch:  make(chan func()),
+	}
+	h.ts.t.Cleanup(call.exit)
+	h.callsMu.Lock()
+	h.calls = append(h.calls, call)
+	h.callsMu.Unlock()
+	for f := range call.ch {
+		f()
+	}
+}
+
+// do executes f in the handler's goroutine.
+func (call *serverHandlerCall) do(f func(http.ResponseWriter, *http.Request)) {
+	donec := make(chan struct{})
+	call.ch <- func() {
+		defer close(donec)
+		f(call.w, call.req)
+	}
+	<-donec
+}
+
+// exit causes the handler to return.
+func (call *serverHandlerCall) exit() {
+	call.closeOnce.Do(func() {
+		close(call.ch)
+	})
+}