TUN-2162: Decomplect OpenStream to allow finer-grained timeouts

h2mux/error.go

@@ -21,10 +21,11 @@ var (
 	ErrInvalidStream       = MuxerProtocolError{"2003 invalid stream", http2.ErrCodeProtocol}
 
 	ErrStreamHeadersSent               = MuxerApplicationError{"3000 headers already sent"}
-	ErrConnectionClosed       = MuxerApplicationError{"3001 connection closed"}
+	ErrStreamRequestConnectionClosed   = MuxerApplicationError{"3001 connection closed while opening stream"}
 	ErrConnectionDropped               = MuxerApplicationError{"3002 connection dropped"}
-	ErrOpenStreamTimeout      = MuxerApplicationError{"3003 open stream timeout"}
+	ErrStreamRequestTimeout            = MuxerApplicationError{"3003 open stream timeout"}
 	ErrResponseHeadersTimeout          = MuxerApplicationError{"3004 timeout waiting for initial response headers"}
+	ErrResponseHeadersConnectionClosed = MuxerApplicationError{"3005 connection closed while waiting for initial response headers"}
 
 	ErrClosedStream = MuxerStreamError{"4000 stream closed", http2.ErrCodeStreamClosed}
 )

h2mux/h2mux.go

@@ -1,7 +1,6 @@
 package h2mux
 
 import (
-	"bytes"
 	"context"
 	"io"
 	"strings"
@@ -388,73 +387,53 @@ func isConnectionClosedError(err error) bool {
 // OpenStream opens a new data stream with the given headers.
 // Called by proxy server and tunnel
 func (m *Muxer) OpenStream(ctx context.Context, headers []Header, body io.Reader) (*MuxedStream, error) {
-	stream := &MuxedStream{
-		responseHeadersReceived: make(chan struct{}),
-		readBuffer:              NewSharedBuffer(),
-		writeBuffer:             &bytes.Buffer{},
-		writeBufferMaxLen:       m.config.StreamWriteBufferMaxLen,
-		writeBufferHasSpace:     make(chan struct{}, 1),
-		receiveWindow:           m.config.DefaultWindowSize,
-		receiveWindowCurrentMax: m.config.DefaultWindowSize,
-		receiveWindowMax:        m.config.MaxWindowSize,
-		sendWindow:              m.config.DefaultWindowSize,
-		readyList:               m.readyList,
-		writeHeaders:            headers,
-		dictionaries:            m.muxReader.dictionaries,
+	stream := m.NewStream(headers)
+	if err := m.MakeMuxedStreamRequest(ctx, MuxedStreamRequest{stream, body}); err != nil {
+		return nil, err
 	}
-
-	select {
-	// Will be received by mux writer
-	case <-ctx.Done():
-		return nil, ErrOpenStreamTimeout
-	case <-m.abortChan:
-		return nil, ErrConnectionClosed
-	case m.newStreamChan <- MuxedStreamRequest{stream: stream, body: body}:
+	if err := m.AwaitResponseHeaders(ctx, stream); err != nil {
+		return nil, err
 	}
-
-	select {
-	case <-ctx.Done():
-		return nil, ErrResponseHeadersTimeout
-	case <-m.abortChan:
-		return nil, ErrConnectionClosed
-	case <-stream.responseHeadersReceived:
 	return stream, nil
 }
-}
+
 
 func (m *Muxer) OpenRPCStream(ctx context.Context) (*MuxedStream, error) {
-	stream := &MuxedStream{
-		responseHeadersReceived: make(chan struct{}),
-		readBuffer:              NewSharedBuffer(),
-		writeBuffer:             &bytes.Buffer{},
-		writeBufferMaxLen:       m.config.StreamWriteBufferMaxLen,
-		writeBufferHasSpace:     make(chan struct{}, 1),
-		receiveWindow:           m.config.DefaultWindowSize,
-		receiveWindowCurrentMax: m.config.DefaultWindowSize,
-		receiveWindowMax:        m.config.MaxWindowSize,
-		sendWindow:              m.config.DefaultWindowSize,
-		readyList:               m.readyList,
-		writeHeaders:            RPCHeaders(),
-		dictionaries:            m.muxReader.dictionaries,
+	stream := m.NewStream(RPCHeaders())
+	if err := m.MakeMuxedStreamRequest(ctx, MuxedStreamRequest{stream: stream, body: nil}); err != nil {
+		return nil, err
 	}
-
-	select {
-	// Will be received by mux writer
-	case <-ctx.Done():
-		return nil, ErrOpenStreamTimeout
-	case <-m.abortChan:
-		return nil, ErrConnectionClosed
-	case m.newStreamChan <- MuxedStreamRequest{stream: stream, body: nil}:
+	if err := m.AwaitResponseHeaders(ctx, stream); err != nil {
+		return nil, err
 	}
-
-	select {
-	case <-ctx.Done():
-		return nil, ErrResponseHeadersTimeout
-	case <-m.abortChan:
-		return nil, ErrConnectionClosed
-	case <-stream.responseHeadersReceived:
 	return stream, nil
 }
+
+func (m *Muxer) NewStream(headers []Header) *MuxedStream {
+	return NewStream(m.config, headers, m.readyList, m.muxReader.dictionaries)
+}
+
+func (m *Muxer) MakeMuxedStreamRequest(ctx context.Context, request MuxedStreamRequest) error {
+	select {
+	case <-ctx.Done():
+		return ErrStreamRequestTimeout
+	case <-m.abortChan:
+		return ErrStreamRequestConnectionClosed
+	// Will be received by mux writer
+	case m.newStreamChan <- request:
+		return nil
+	}
+}
+
+func (m *Muxer) AwaitResponseHeaders(ctx context.Context, stream *MuxedStream) error {
+	select {
+	case <-ctx.Done():
+		return ErrResponseHeadersTimeout
+	case <-m.abortChan:
+		return ErrResponseHeadersConnectionClosed
+	case <-stream.responseHeadersReceived:
+		return nil
+	}
 }
 
 func (m *Muxer) Metrics() *MuxerMetrics {

h2mux/h2mux_test.go

@@ -584,8 +584,8 @@ func TestOpenAfterDisconnect(t *testing.T) {
 			[]Header{{Name: "test-header", Value: "headerValue"}},
 			nil,
 		)
-		if err != ErrConnectionClosed {
-			t.Fatalf("unexpected error in OpenStream: %s", err)
+		if err != ErrStreamRequestConnectionClosed && err != ErrResponseHeadersConnectionClosed {
+			t.Fatalf("case %v: unexpected error in OpenStream: %v", i, err)
 		}
 	}
 }

h2mux/muxedstream.go

@@ -88,6 +88,23 @@ func (th TunnelHostname) IsSet() bool {
 	return th != ""
 }
 
+func NewStream(config MuxerConfig, writeHeaders []Header, readyList *ReadyList, dictionaries h2Dictionaries) *MuxedStream {
+	return &MuxedStream{
+		responseHeadersReceived: make(chan struct{}),
+		readBuffer:              NewSharedBuffer(),
+		writeBuffer:             &bytes.Buffer{},
+		writeBufferMaxLen:       config.StreamWriteBufferMaxLen,
+		writeBufferHasSpace:     make(chan struct{}, 1),
+		receiveWindow:           config.DefaultWindowSize,
+		receiveWindowCurrentMax: config.DefaultWindowSize,
+		receiveWindowMax:        config.MaxWindowSize,
+		sendWindow:              config.DefaultWindowSize,
+		readyList:               readyList,
+		writeHeaders:            writeHeaders,
+		dictionaries:            dictionaries,
+	}
+}
+
 func (s *MuxedStream) Read(p []byte) (n int, err error) {
 	var readBuffer ReadWriteClosedCloser
 	if s.dictionaries.read != nil {