Merge pull request #152 from monzo/configure-server-timeouts

Add support for server timeouts

Author: milesbxf

e2e_http1_test.go

@@ -6,15 +6,18 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/monzo/terrors"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/stretchr/testify/require"
 )
 
 type http1Flavour struct {
 	T *testing.T
 }
 
-func (f http1Flavour) Serve(svc Service) *Server {
-	s, err := Listen(svc, "localhost:0")
+func (f http1Flavour) Serve(svc Service, opts ...ServerOption) *Server {
+	s, err := Listen(svc, "localhost:0", opts...)
 	require.NoError(f.T, err)
 	return s
 }
@@ -31,17 +34,22 @@ func (f http1Flavour) Context() (context.Context, func()) {
 	return context.WithCancel(context.Background())
 }
 
+func (f http1Flavour) AssertConnectionResetError(t *testing.T, terr *terrors.Error) {
+	assert.Equal(t, terrors.ErrInternalService, terr.Code)
+	assert.Equal(t, "EOF", terr.Message)
+}
+
 type http1TLSFlavour struct {
 	T    *testing.T
 	cert tls.Certificate
 }
 
-func (f http1TLSFlavour) Serve(svc Service) *Server {
+func (f http1TLSFlavour) Serve(svc Service, opts ...ServerOption) *Server {
 	l, err := tls.Listen("tcp", "localhost:0", &tls.Config{
 		Certificates: []tls.Certificate{f.cert},
 		ClientAuth:   tls.NoClientCert})
 	require.NoError(f.T, err)
-	s, err := Serve(svc, l)
+	s, err := Serve(svc, l, opts...)
 	require.NoError(f.T, err)
 	return s
 }
@@ -57,3 +65,8 @@ func (f http1TLSFlavour) Proto() string {
 func (f http1TLSFlavour) Context() (context.Context, func()) {
 	return context.WithCancel(context.Background())
 }
+
+func (f http1TLSFlavour) AssertConnectionResetError(t *testing.T, terr *terrors.Error) {
+	assert.Equal(t, terrors.ErrInternalService, terr.Code)
+	assert.Equal(t, "local error: tls: bad record MAC", terr.Message)
+}

e2e_http2_test.go

@@ -6,6 +6,9 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/monzo/terrors"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/stretchr/testify/require"
 )
 
@@ -14,9 +17,9 @@ type http2H2cFlavour struct {
 	client Service
 }
 
-func (f http2H2cFlavour) Serve(svc Service) *Server {
+func (f http2H2cFlavour) Serve(svc Service, opts ...ServerOption) *Server {
 	svc = svc.Filter(H2cFilter)
-	s, err := Listen(svc, "localhost:0")
+	s, err := Listen(svc, "localhost:0", opts...)
 	require.NoError(f.T, err)
 	return s
 }
@@ -33,14 +36,19 @@ func (f http2H2cFlavour) Context() (context.Context, func()) {
 	return context.WithCancel(context.Background())
 }
 
+func (f http2H2cFlavour) AssertConnectionResetError(t *testing.T, terr *terrors.Error) {
+	assert.Equal(t, terrors.ErrInternalService, terr.Code)
+	assert.Contains(t, terr.Message, "INTERNAL_ERROR")
+}
+
 type http2H2cPriorKnowledgeFlavour struct {
 	T      *testing.T
 	client Service
 }
 
-func (f http2H2cPriorKnowledgeFlavour) Serve(svc Service) *Server {
+func (f http2H2cPriorKnowledgeFlavour) Serve(svc Service, opts ...ServerOption) *Server {
 	svc = svc.Filter(H2cFilter)
-	s, err := Listen(svc, "localhost:0")
+	s, err := Listen(svc, "localhost:0", opts...)
 	require.NoError(f.T, err)
 	return s
 }
@@ -59,19 +67,24 @@ func (f http2H2cPriorKnowledgeFlavour) Context() (context.Context, func()) {
 	return ctx, cancel
 }
 
+func (f http2H2cPriorKnowledgeFlavour) AssertConnectionResetError(t *testing.T, terr *terrors.Error) {
+	assert.Equal(t, terrors.ErrInternalService, terr.Code)
+	assert.Equal(t, "EOF", terr.Message)
+}
+
 type http2H2Flavour struct {
 	T      *testing.T
 	client Service
 	cert   tls.Certificate
 }
 
-func (f http2H2Flavour) Serve(svc Service) *Server {
+func (f http2H2Flavour) Serve(svc Service, opts ...ServerOption) *Server {
 	l, err := tls.Listen("tcp", "localhost:0", &tls.Config{
 		Certificates: []tls.Certificate{f.cert},
 		ClientAuth:   tls.NoClientCert,
 		NextProtos:   []string{"h2"}})
 	require.NoError(f.T, err)
-	s, err := Serve(svc, l)
+	s, err := Serve(svc, l, opts...)
 	require.NoError(f.T, err)
 	return s
 }
@@ -87,3 +100,8 @@ func (f http2H2Flavour) Proto() string {
 func (f http2H2Flavour) Context() (context.Context, func()) {
 	return context.WithCancel(context.Background())
 }
+
+func (f http2H2Flavour) AssertConnectionResetError(t *testing.T, terr *terrors.Error) {
+	assert.Equal(t, terrors.ErrInternalService, terr.Code)
+	assert.Contains(t, terr.Message, "INTERNAL_ERROR")
+}

e2e_test.go

@@ -23,10 +23,12 @@ import (
 )
 
 type e2eFlavour interface {
-	Serve(Service) *Server
+	Serve(Service, ...ServerOption) *Server
 	URL(*Server) string
 	Proto() string
 	Context() (context.Context, func())
+
+	AssertConnectionResetError(t *testing.T, terr *terrors.Error)
 }
 
 // flavours runs the passed E2E test with all test flavours (HTTP/1.1, HTTP/2.0/h2c, etc.)
@@ -805,3 +807,36 @@ func TestE2EDraining(t *testing.T) {
 		<-serverClosed
 	})
 }
+
+func TestE2EServerTimeouts(t *testing.T) {
+	someFlavours(t, []string{
+		"http1.1",
+		"http1.1-tls",
+		"http2.0-h2",
+
+		// The Go h2c implementation doesn't currently support server timeouts
+		// See https://github.com/golang/go/issues/52868
+		//"http2.0-h2c",
+		//"http2.0-h2c-prior-knowledge",
+	}, func(t *testing.T, flav e2eFlavour) {
+		ctx, cancel := flav.Context()
+		defer cancel()
+
+		srv := Service(func(req Request) Response {
+			time.Sleep(1 * time.Second)
+			return NewResponse(req)
+		})
+		srv = srv.Filter(ErrorFilter)
+		s := flav.Serve(srv, WithTimeout(TimeoutOptions{Write: 10 * time.Millisecond}))
+		defer s.Stop(ctx)
+
+		req := NewRequest(ctx, "GET", flav.URL(s), nil)
+		rsp := req.Send().Response()
+		if assert.Error(t, rsp.Error) {
+			terr, ok := rsp.Error.(*terrors.Error)
+			if assert.Truef(t, ok, "expected terror, got %T", rsp.Error) {
+				flav.AssertConnectionResetError(t, terr)
+			}
+		}
+	})
+}

examples/simple.go

@@ -22,7 +22,7 @@ func main() {
 	svc := router.Serve().
 		Filter(typhon.ErrorFilter).
 		Filter(typhon.H2cFilter)
-	srv, err := typhon.Listen(svc, ":8000")
+	srv, err := typhon.Listen(svc, ":8000", typhon.WithTimeout(typhon.TimeoutOptions{Read: time.Second * 10}))
 	if err != nil {
 		panic(err)
 	}

h2c.go

@@ -9,7 +9,7 @@ import (
 	"net/textproto"
 	"sync"
 
-	"github.com/deckarep/golang-set"
+	mapset "github.com/deckarep/golang-set"
 	"github.com/monzo/terrors"
 	"golang.org/x/net/http/httpguts"
 	"golang.org/x/net/http2"

request_test.go

@@ -5,12 +5,13 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"github.com/monzo/terrors"
 	"io/ioutil"
 	"math"
 	"strings"
 	"testing"
 
+	"github.com/monzo/terrors"
+
 	legacyproto "github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"

response_test.go

@@ -5,15 +5,16 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	legacyproto "github.com/golang/protobuf/proto"
-	"github.com/monzo/typhon/legacyprototest"
 	"io"
 	"io/ioutil"
 	"math"
 	"net/http"
 	"strings"
 	"testing"
 
+	legacyproto "github.com/golang/protobuf/proto"
+	"github.com/monzo/typhon/legacyprototest"
+
 	"github.com/monzo/terrors"
 	"github.com/monzo/typhon/prototest"
 	"github.com/stretchr/testify/assert"

server.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/monzo/slog"
 )
@@ -21,6 +22,9 @@ type Server struct {
 	shutdownFuncsM sync.Mutex
 }
 
+// ServerOption allows customizing the underling http.Server
+type ServerOption func(*Server)
+
 // Listener returns the network listener that this server is active on.
 func (s *Server) Listener() net.Listener {
 	return s.l
@@ -72,8 +76,8 @@ func (s *Server) addShutdownFunc(f func(context.Context)) {
 	s.shutdownFuncs = append(s.shutdownFuncs, f)
 }
 
-// Serve starts a HTTP server, binding the passed Service to the passed listener.
-func Serve(svc Service, l net.Listener) (*Server, error) {
+// Serve starts a HTTP server, binding the passed Service to the passed listener and applying the passed ServerOptions.
+func Serve(svc Service, l net.Listener, opts ...ServerOption) (*Server, error) {
 	s := &Server{
 		l:            l,
 		shuttingDown: make(chan struct{})}
@@ -83,7 +87,14 @@ func Serve(svc Service, l net.Listener) (*Server, error) {
 	})
 	s.srv = &http.Server{
 		Handler:        HttpHandler(svc),
-		MaxHeaderBytes: http.DefaultMaxHeaderBytes}
+		MaxHeaderBytes: http.DefaultMaxHeaderBytes,
+	}
+
+	// Apply any given ServerOptions
+	for _, opt := range opts {
+		opt(s)
+	}
+
 	go func() {
 		err := s.srv.Serve(l)
 		if err != nil && err != http.ErrServerClosed {
@@ -97,7 +108,7 @@ func Serve(svc Service, l net.Listener) (*Server, error) {
 	return s, nil
 }
 
-func Listen(svc Service, addr string) (*Server, error) {
+func Listen(svc Service, addr string, opts ...ServerOption) (*Server, error) {
 	// Determine on which address to listen, choosing in order one of:
 	// 1. The passed addr
 	// 2. PORT variable (listening on all interfaces)
@@ -120,5 +131,26 @@ func Listen(svc Service, addr string) (*Server, error) {
 	if err != nil {
 		return nil, err
 	}
-	return Serve(svc, l)
+	return Serve(svc, l, opts...)
+}
+
+// TimeoutOptions specifies various server timeouts. See http.Server for details of what these do.
+// There's a nice post explaining them here: https://ieftimov.com/posts/make-resilient-golang-net-http-servers-using-timeouts-deadlines-context-cancellation/#server-timeouts---first-principles
+// WARNING: Due to a Go bug, connections using h2c do not respect these timeouts.
+// See https://github.com/golang/go/issues/52868
+type TimeoutOptions struct {
+	Read       time.Duration
+	ReadHeader time.Duration
+	Write      time.Duration
+	Idle       time.Duration
+}
+
+// WithTimeout sets the server timeouts.
+func WithTimeout(opts TimeoutOptions) ServerOption {
+	return func(s *Server) {
+		s.srv.ReadTimeout = opts.Read
+		s.srv.ReadHeaderTimeout = opts.ReadHeader
+		s.srv.WriteTimeout = opts.Write
+		s.srv.IdleTimeout = opts.Idle
+	}
 }