Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 17 additions & 13 deletions client.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,10 @@

import (
"context"
"crypto/tls"
"net"
"net/http"
"time"

"github.com/monzo/terrors"
"golang.org/x/net/http2"
)

var (
Expand All @@ -28,18 +25,25 @@
}

// H2cRoundTripper is a prior-knowledge H2c client. It does not support ProxyFromEnvironment.
H2cRoundTripper http.RoundTripper = &http2.Transport{
AllowHTTP: true,
// This monstrosity is needed to get the http2 Transport to dial over cleartext.
// See https://github.com/thrawn01/h2c-golang-example
DialTLSContext: func(ctx context.Context, network, addr string, _ *tls.Config) (net.Conn, error) {
var d net.Dialer
return d.DialContext(ctx, network, addr)
H2cRoundTripper http.RoundTripper = newH2cRoundTripper()
)

// newH2cRoundTripper builds a transport that speaks unencrypted HTTP/2 (h2c) with prior knowledge
func newH2cRoundTripper() *http.Transport {
t := &http.Transport{
HTTP2: &http.HTTP2Config{
// Health-check idle connections: send a ping after 30s without a frame, and close the
// connection if no response arrives within 10s.
SendPingTimeout: 30 * time.Second,
PingTimeout: 10 * time.Second,
},
ReadIdleTimeout: 30 * time.Second,
PingTimeout: 10 * time.Second,
}
)
// Enabling UnencryptedHTTP2 without HTTP1 makes the transport use h2c (prior knowledge) for
// http:// URLs with no HTTP/1.1 fallback.
t.Protocols = new(http.Protocols)
t.Protocols.SetUnencryptedHTTP2(true)
return t
}

// A ResponseFuture is a container for a Response which will materialise at some point.
type ResponseFuture struct {
Expand All @@ -60,7 +64,7 @@

// HttpService returns a Service which sends requests via the given net/http RoundTripper.
// Only use this if you need to do something custom at the transport level.
func HttpService(rt http.RoundTripper) Service {

Check failure on line 67 in client.go

View workflow job for this annotation

GitHub Actions / test (stable, ubuntu-latest)

func HttpService should be HTTPService

Check failure on line 67 in client.go

View workflow job for this annotation

GitHub Actions / test (oldstable, ubuntu-latest)

func HttpService should be HTTPService
return func(req Request) Response {
ctx := req.unwrappedContext()
httpRsp, err := rt.RoundTrip(req.Request.WithContext(ctx))
Expand Down
2 changes: 1 addition & 1 deletion e2e_http2_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ func (f http2H2cPriorKnowledgeFlavour) Context() (context.Context, func()) {

func (f http2H2cPriorKnowledgeFlavour) AssertConnectionResetError(t *testing.T, terr *terrors.Error) {
assert.Equal(t, terrors.ErrInternalService, terr.Code)
assert.Equal(t, "EOF", terr.Message)
assert.Contains(t, terr.Message, "INTERNAL_ERROR")
}

type http2H2Flavour struct {
Expand Down
68 changes: 57 additions & 11 deletions e2e_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -710,9 +710,11 @@ func TestE2EStreamingServerAbort(t *testing.T) {
<-done
body, err := ioutil.ReadAll(rsp.Body)
assert.Equal(t, "derp\n", string(body))
require.IsType(t, http2.StreamError{}, err)
streamErr := err.(http2.StreamError)
assert.Equal(t, streamErr.Code, http2.ErrCodeInternal)
// The client sees an HTTP/2 stream error carrying INTERNAL_ERROR. We match on the error
// string rather than a concrete type because the error comes from x/net/http2 for some
// flavours and from net/http's (unexported) native HTTP/2 for the prior-knowledge flavour.
require.Error(t, err)
assert.Contains(t, err.Error(), http2.ErrCodeInternal.String())
})
}

Expand Down Expand Up @@ -809,16 +811,61 @@ func TestE2EDraining(t *testing.T) {
})
}

// TestE2EForcefulShutdown verifies that when graceful shutdown can't complete before the context
// expires, the server forcibly terminates in-flight connections rather than hanging. This exercises
// the srv.Close() fallback in Server.Stop, which (for h2c) used to be provided by hand-rolled
// connection tracking but is now delegated to net/http.
func TestE2EForcefulShutdown(t *testing.T) {
flavours(t, func(t *testing.T, flav e2eFlavour) {
ctx, cancel := flav.Context()
defer cancel()

returnRsp := make(chan bool)
svc := Service(func(req Request) Response {
<-returnRsp
return NewResponse(req)
})
svc = svc.Filter(ErrorFilter)
s := flav.Serve(svc)

// Send a request, which will hang in the handler until we send to returnRsp
req := NewRequest(ctx, "GET", flav.URL(s), nil)
rspF := req.Send()
time.Sleep(10 * time.Millisecond) // allow connection to be established

// Stop the server with an already-expired context. Graceful shutdown can't complete (the
// handler is still blocked), so the server must fall back to forcibly closing connections
// and Stop must return promptly rather than waiting for the handler.
expiredCtx, cancelExpired := context.WithCancel(ctx)
cancelExpired()
serverClosed := make(chan struct{})
go func() {
s.Stop(expiredCtx)
close(serverClosed)
}()
select {
case <-serverClosed:
case <-time.After(5 * time.Second):
require.FailNow(t, "Stop did not return promptly with an expired context")
}

// The in-flight request should error out because its connection was forcibly closed.
rsp := rspF.Response()
require.Error(t, rsp.Error)

// Unblock the (still-blocked) handler so its goroutine doesn't leak; its write races against
// the forcibly-closed connection, which is fine.
returnRsp <- true
})
}

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",
"http2.0-h2c",
"http2.0-h2c-prior-knowledge",
}, func(t *testing.T, flav e2eFlavour) {
ctx, cancel := flav.Context()
defer cancel()
Expand Down Expand Up @@ -854,9 +901,8 @@ func TestE2EMaxConnectionAge_ConnectionAgeSet(t *testing.T) {
"http1.1",
"http1.1-tls",
"http2.0-h2",

// The Go h2c implementation doesn't currently support max
// connection age.
"http2.0-h2c",
"http2.0-h2c-prior-knowledge",
}, func(t *testing.T, flav e2eFlavour) {
ctx, cancel := flav.Context()
defer cancel()
Expand Down
3 changes: 1 addition & 2 deletions go.mod
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
module github.com/monzo/typhon

go 1.19
go 1.24

require (
github.com/deckarep/golang-set v1.8.0
github.com/fortytw2/leaktest v1.3.0
github.com/golang/protobuf v1.5.2
github.com/monzo/slog v0.0.0-20211123154010-52a5ddb2ba55
Expand Down
2 changes: 0 additions & 2 deletions go.sum
Original file line number Diff line number Diff line change
@@ -1,8 +1,6 @@
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/deckarep/golang-set v1.8.0 h1:sk9/l/KqpunDwP7pSjUg0keiOOLEnOBHzykLrsPppp4=
github.com/deckarep/golang-set v1.8.0/go.mod h1:5nI87KwE7wgsBU1F4GKAw2Qod7p5kyS383rP6+o6qqo=
github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
Expand Down
153 changes: 8 additions & 145 deletions h2c.go
Original file line number Diff line number Diff line change
@@ -1,150 +1,13 @@
package typhon

import (
"bufio"
"context"
"math"
"net"
"net/http"
"net/textproto"
"sync"

mapset "github.com/deckarep/golang-set"
"github.com/monzo/terrors"
"golang.org/x/net/http/httpguts"
"golang.org/x/net/http2"
"golang.org/x/net/http2/h2c"
)

// H2cFilter adds HTTP/2 h2c upgrade support to the wrapped Service (as defined in RFC 7540 §3.2, §3.4).
// H2cFilter previously added HTTP/2 h2c upgrade support to the wrapped Service by hijacking the
// connection and running golang.org/x/net/http2/h2c.
//
// Deprecated: h2c (HTTP/2 cleartext, prior knowledge) is now enabled natively for every Typhon
// server via Go's net/http support (Go 1.24+); see Serve. This filter is a no-op retained only for
// backwards compatibility and will be removed in a future release. It is no longer necessary to
// apply it. Note that the native implementation only supports h2c with prior knowledge, not the
// (unused by Go clients) RFC 7540 §3.2 `Upgrade: h2c` handshake.
func H2cFilter(req Request, svc Service) Response {
h := req.Header
// h2c with prior knowledge (RFC 7540 §3.4)
isPrior := (req.Method == "PRI" && len(h) == 0 && req.URL.Path == "*" && req.Proto == "HTTP/2.0")
// h2c upgrade (RFC 7540 §3.2)
isUpgrade := httpguts.HeaderValuesContainsToken(h[textproto.CanonicalMIMEHeaderKey("Upgrade")], "h2c") &&
httpguts.HeaderValuesContainsToken(h[textproto.CanonicalMIMEHeaderKey("Connection")], "HTTP2-Settings")
if isPrior || isUpgrade {
rsp := NewResponse(req)
rw, h2s, err := setupH2cHijacker(req, rsp.Writer())
if err != nil {
return Response{Error: err}
}
h2c.NewHandler(HttpHandler(svc), h2s).ServeHTTP(rw, &req.Request)
return rsp
}
return svc(req)
}

// Dear reader: I'm sorry, the code below isn't fun. This is because Go's h2c implementation doesn't have support for
// connection draining, and all the hooks that would make this easy are unexported.
//
// If this ticket gets resolved this code can be dramatically simplified, but it is not a priority for the Go team:
// https://github.com/golang/go/issues/26682
//
// 🤢

var h2cConns sync.Map // map[*Server]*h2cInfo

// h2cInfo stores information about connections that have been upgraded by a single Typhon server
type h2cInfo struct {
conns mapset.Set
h2s *http2.Server
}

// hijackedConn represents a network connection that has been hijacked for a h2c upgrade. This is necessary because we
// need to know when the connection has been closed, to know if/when graceful shutdown completes.
type hijackedConn struct {
net.Conn
onClose func(*hijackedConn)
closed chan struct{}
closeOnce sync.Once
}

func (c *hijackedConn) Close() error {
defer c.closeOnce.Do(func() {
close(c.closed)
c.onClose(c)
})
return c.Conn.Close()
}

type h2cHijacker struct {
http.ResponseWriter
http.Hijacker
onHijack func(*hijackedConn)
}

func (h h2cHijacker) Hijack() (net.Conn, *bufio.ReadWriter, error) {
c, r, err := h.Hijacker.Hijack()
conn := &hijackedConn{
Conn: c,
closed: make(chan struct{})}
h.onHijack(conn)
return conn, r, err
}

func shutdownH2c(ctx context.Context, srv *Server, h2c *h2cInfo) {
gracefulCloseLoop:
for _, _c := range h2c.conns.ToSlice() {
c := _c.(*hijackedConn)
select {
case <-ctx.Done():
break gracefulCloseLoop
case <-c.closed:
h2c.conns.Remove(c)
}
}
// If any connections remain after gracefulCloseLoop, we need to forcefully close them
for _, _c := range h2c.conns.ToSlice() {
c := _c.(*hijackedConn)
c.Close()
h2c.conns.Remove(c)
}
h2cConns.Delete(srv)
}

func setupH2cHijacker(req Request, rw http.ResponseWriter) (http.ResponseWriter, *http2.Server, error) {
hijacker, ok := rw.(http.Hijacker)
if !ok {
err := terrors.InternalService("hijack_impossible", "Cannot hijack response; h2c upgrade impossible", nil)
return nil, nil, err
}
srv := req.server
if srv == nil {
return rw, &http2.Server{}, nil
}

h2c := &h2cInfo{
conns: mapset.NewSet(),
h2s: &http2.Server{
// We're copying envoy and grpc by setting this to the max uint32.
// The Go default is 250 which is not ideal for long lived streaming requests.
MaxConcurrentStreams: math.MaxUint32,
}}
_h2c, loaded := h2cConns.LoadOrStore(srv, h2c)
h2c = _h2c.(*h2cInfo)
if !loaded {
// http2.ConfigureServer wires up an unexported method within the http2 library so it gracefully drains h2c
// connections when the http1 server is stopped. However, this happens asynchronously: the http1 server will
// think it has shut down before the h2c connections have finished draining. To work around this, we add
// a shutdown function of our own in the Typhon server which waits for connections to be drained, or if things
// timeout before then to terminate them forcefully.
http2.ConfigureServer(srv.srv, h2c.h2s)
srv.addShutdownFunc(func(ctx context.Context) {
shutdownH2c(ctx, srv, h2c)
})
}

h := h2cHijacker{
ResponseWriter: rw,
Hijacker: hijacker,
onHijack: func(c *hijackedConn) {
h2c.conns.Add(c)
// when the connection closes, remove from h2cInfo's to prevent refs to dead connections accumulating
c.onClose = func(c *hijackedConn) {
h2c.conns.Remove(c)
}
}}
return h, h2c.h2s, nil
}
8 changes: 4 additions & 4 deletions request.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,15 @@ import (
"context"
"encoding/json"
"fmt"
legacyproto "github.com/golang/protobuf/proto"
"github.com/monzo/terrors"
"google.golang.org/protobuf/proto"
"io"
"io/ioutil"
"net/http"
"strings"
"time"

legacyproto "github.com/golang/protobuf/proto"
"github.com/monzo/terrors"
"google.golang.org/protobuf/proto"
)

// A Request is Typhon's wrapper around http.Request, used by both clients and servers.
Expand All @@ -23,7 +24,6 @@ type Request struct {
context.Context
err error // Any error from request construction; read by ErrorFilter
hijacker http.Hijacker
server *Server
}

// unwrappedContext returns the most "unwrapped" Context possible for that in the request.
Expand Down
Loading
Loading