Release 2021.2.2

Also changed the socks test code so that it binds to localhost, so that
we don't get popups saying "would you like to allow socks.test to use
the network"

.gitignore

@@ -18,3 +18,4 @@ cloudflared-x86-64*
 .DS_Store
 *-session.log
 ssh_server_tests/.env
+.cover

Makefile

@@ -4,7 +4,11 @@ MSI_VERSION   := $(shell git tag -l --sort=v:refname | grep "w" | tail -1 | cut
 #e.g. w3.0.1 or w4.2.10. It trims off the w character when creating the MSI.
 
 ifeq ($(FIPS), true)
-	GO_BUILD_TAGS := "$(GO_BUILD_TAGS) fips"
+	GO_BUILD_TAGS := $(GO_BUILD_TAGS) fips
+endif
+
+ifneq ($(GO_BUILD_TAGS),)
+	GO_BUILD_TAGS := -tags $(GO_BUILD_TAGS)
 endif
 
 DATE          := $(shell date -u '+%Y-%m-%d-%H%M UTC')
@@ -76,7 +80,7 @@ clean:
 
 .PHONY: cloudflared
 cloudflared: tunnel-deps
-	GOOS=$(TARGET_OS) GOARCH=$(TARGET_ARCH) go build -v -mod=vendor -tags $(GO_BUILD_TAGS) $(VERSION_FLAGS) $(IMPORT_PATH)/cmd/cloudflared
+	GOOS=$(TARGET_OS) GOARCH=$(TARGET_ARCH) go build -v -mod=vendor $(GO_BUILD_TAGS) $(VERSION_FLAGS) $(IMPORT_PATH)/cmd/cloudflared
 
 .PHONY: container
 container:
@@ -84,7 +88,13 @@ container:
 
 .PHONY: test
 test: vet
+ifndef CI
 	go test -v -mod=vendor -race $(VERSION_FLAGS) ./...
+else
+	@mkdir -p .cover
+	go test -v -mod=vendor -race $(VERSION_FLAGS) -coverprofile=".cover/c.out" ./...
+	go tool cover -html ".cover/c.out" -o .cover/all.html
+endif
 
 .PHONY: test-ssh-server
 test-ssh-server:

RELEASE_NOTES

@@ -1,3 +1,12 @@
+2021.2.2
+- 2021-02-04 TUN-3864: Users can choose where credentials file is written after creating a tunnel
+- 2021-02-04 TUN-3869: Improve reliability of graceful shutdown.
+- 2021-02-07 TUN-3878: Do not supply -tags when none are specified
+- 2021-02-04 TUN-3635: Send event when unregistering tunnel for gracful shutdown so /ready endpoint reports down status befoe connections finish handling pending requests.
+- 2021-02-08 TUN-3890: Code coverage for cloudflared in CI
+- 2021-02-09 AUTH-3375 exchangeOrgToken deleted cookie fix
+- 2020-11-18 Update error message to use login command
+
 2021.2.1
 - 2021-02-04 TUN-3858: Do not suffix cloudflared version with -fips
 

cmd/cloudflared/access/cmd.go

@@ -291,7 +291,7 @@ func generateToken(c *cli.Context) error {
 	}
 	tok, err := token.GetAppTokenIfExists(appURL)
 	if err != nil || tok == "" {
-		fmt.Fprintln(os.Stderr, "Unable to find token for provided application. Please run token command to generate token.")
+		fmt.Fprintln(os.Stderr, "Unable to find token for provided application. Please run login command to generate token.")
 		return err
 	}
 

cmd/cloudflared/token/token.go

@@ -301,7 +301,9 @@ func exchangeOrgToken(appURL *url.URL, orgToken string) (string, error) {
 	resp.Body.Close()
 	var appToken string
 	for _, c := range resp.Cookies() {
-		if c.Name == tokenHeader {
+		//if Org token revoked on exchange, getTokensFromEdge instead
+		validAppToken := c.Name == tokenHeader && time.Now().Before(c.Expires)
+		if validAppToken {
 			appToken = c.Value
 			break
 		}

cmd/cloudflared/tunnel/cmd.go

@@ -154,7 +154,7 @@ func TunnelCommand(c *cli.Context) error {
 		return err
 	}
 	if name := c.String("name"); name != "" { // Start a named tunnel
-		return runAdhocNamedTunnel(sc, name)
+		return runAdhocNamedTunnel(sc, name, c.String(CredFileFlag))
 	}
 	if ref := config.GetConfiguration().TunnelID; ref != "" {
 		return fmt.Errorf("Use `cloudflared tunnel run` to start tunnel %s", ref)
@@ -169,10 +169,10 @@ func Init(ver string, gracefulShutdown chan struct{}) {
 }
 
 // runAdhocNamedTunnel create, route and run a named tunnel in one command
-func runAdhocNamedTunnel(sc *subcommandContext, name string) error {
+func runAdhocNamedTunnel(sc *subcommandContext, name, credentialsOutputPath string) error {
 	tunnel, ok, err := sc.tunnelActive(name)
 	if err != nil || !ok {
-		tunnel, err = sc.create(name)
+		tunnel, err = sc.create(name, credentialsOutputPath)
 		if err != nil {
 			return errors.Wrap(err, "failed to create tunnel")
 		}
@@ -539,6 +539,7 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 	flags = append(flags, configureLoggingFlags(shouldHide)...)
 	flags = append(flags, configureProxyDNSFlags(shouldHide)...)
 	flags = append(flags, []cli.Flag{
+		credentialsFileFlag,
 		altsrc.NewBoolFlag(&cli.BoolFlag{
 			Name:   "is-autoupdated",
 			Usage:  "Signal the new process that Argo Tunnel client has been autoupdated",

cmd/cloudflared/tunnel/subcommand_context.go

@@ -147,7 +147,7 @@ func (sc *subcommandContext) readTunnelCredentials(credFinder CredFinder) (conne
 	return credentials, nil
 }
 
-func (sc *subcommandContext) create(name string) (*tunnelstore.Tunnel, error) {
+func (sc *subcommandContext) create(name string, credentialsOutputPath string) (*tunnelstore.Tunnel, error) {
 	client, err := sc.client()
 	if err != nil {
 		return nil, errors.Wrap(err, "couldn't create client to talk to Argo Tunnel backend")
@@ -173,7 +173,7 @@ func (sc *subcommandContext) create(name string) (*tunnelstore.Tunnel, error) {
 		TunnelID:     tunnel.ID,
 		TunnelName:   name,
 	}
-	filePath, writeFileErr := writeTunnelCredentials(credential.certPath, &tunnelCredentials)
+	filePath, writeFileErr := writeTunnelCredentials(credential.certPath, credentialsOutputPath, &tunnelCredentials)
 	if writeFileErr != nil {
 		var errorLines []string
 		errorLines = append(errorLines, fmt.Sprintf("Your tunnel '%v' was created with ID %v. However, cloudflared couldn't write to the tunnel credentials file at %v.json.", tunnel.Name, tunnel.ID, tunnel.ID))

cmd/cloudflared/tunnel/subcommands.go

@@ -90,7 +90,7 @@ var (
 	credentialsFileFlag = altsrc.NewStringFlag(&cli.StringFlag{
 		Name:    CredFileFlag,
 		Aliases: []string{CredFileFlagAlias},
-		Usage:   "File path of tunnel credentials",
+		Usage:   "Filepath at which to read/write the tunnel credentials",
 		EnvVars: []string{"TUNNEL_CRED_FILE"},
 	})
 	forceDeleteFlag = &cli.BoolFlag{
@@ -121,7 +121,7 @@ func buildCreateCommand() *cli.Command {
   For example, to create a tunnel named 'my-tunnel' run:
 
   $ cloudflared tunnel create my-tunnel`,
-		Flags:              []cli.Flag{outputFormatFlag},
+		Flags:              []cli.Flag{outputFormatFlag, credentialsFileFlag},
 		CustomHelpTemplate: commandHelpTemplate(),
 	}
 }
@@ -144,7 +144,7 @@ func createCommand(c *cli.Context) error {
 	}
 	name := c.Args().First()
 
-	_, err = sc.create(name)
+	_, err = sc.create(name, c.String(CredFileFlag))
 	return errors.Wrap(err, "failed to create tunnel")
 }
 
@@ -154,12 +154,18 @@ func tunnelFilePath(tunnelID uuid.UUID, directory string) (string, error) {
 	return homedir.Expand(filePath)
 }
 
+// If an `outputFile` is given, write the credentials there.
+// Otherwise, write it to the same directory as the originCert,
+// with the filename `<tunnel id>.json`.
 func writeTunnelCredentials(
-	originCertPath string,
+	originCertPath, outputFile string,
 	credentials *connection.Credentials,
 ) (filePath string, err error) {
-	originCertDir := filepath.Dir(originCertPath)
-	filePath, err = tunnelFilePath(credentials.TunnelID, originCertDir)
+	filePath = outputFile
+	if outputFile == "" {
+		originCertDir := filepath.Dir(originCertPath)
+		filePath, err = tunnelFilePath(credentials.TunnelID, originCertDir)
+	}
 	if err != nil {
 		return "", err
 	}

cmd/cloudflared/ui/launch_ui.go

@@ -15,7 +15,6 @@ import (
 
 type connState struct {
 	location string
-	state    connection.Status
 }
 
 type uiModel struct {
@@ -32,6 +31,7 @@ type palette struct {
 	defaultText  string
 	disconnected string
 	reconnecting string
+	unregistered string
 }
 
 func NewUIModel(version, hostname, metricsURL string, ing *ingress.Ingress, haConnections int) *uiModel {
@@ -67,6 +67,7 @@ func (data *uiModel) Launch(
 		defaultText:  "white",
 		disconnected: "red",
 		reconnecting: "orange",
+		unregistered: "orange",
 	}
 
 	app := tview.NewApplication()
@@ -128,7 +129,7 @@ func (data *uiModel) Launch(
 		switch event.EventType {
 		case connection.Connected:
 			data.setConnTableCell(event, connTable, palette)
-		case connection.Disconnected, connection.Reconnecting:
+		case connection.Disconnected, connection.Reconnecting, connection.Unregistering:
 			data.changeConnStatus(event, connTable, log, palette)
 		case connection.SetURL:
 			tunnelHostText.SetText(event.URL)
@@ -167,10 +168,7 @@ func (data *uiModel) changeConnStatus(event connection.Event, table *tview.Table
 
 	locationState := event.Location
 
-	if event.EventType == connection.Disconnected {
-		connState.state = connection.Disconnected
-	} else if event.EventType == connection.Reconnecting {
-		connState.state = connection.Reconnecting
+	if event.EventType == connection.Reconnecting {
 		locationState = "Reconnecting..."
 	}
 
@@ -196,7 +194,6 @@ func (data *uiModel) setConnTableCell(event connection.Event, table *tview.Table
 	connectionNum := index + 1
 
 	// Update slice to keep track of connection location and state in UI table
-	data.connections[index].state = connection.Connected
 	data.connections[index].location = event.Location
 
 	// Update text in table cell to show disconnected state
@@ -218,6 +215,8 @@ func newCellText(palette palette, connectionNum int, location string, connectedS
 		dotColor = palette.disconnected
 	case connection.Reconnecting:
 		dotColor = palette.reconnecting
+	case connection.Unregistering:
+		dotColor = palette.unregistered
 	}
 
 	return fmt.Sprintf(connFmtString, dotColor, palette.defaultText, connectionNum, location)

connection/connection_test.go

@@ -27,7 +27,6 @@ var (
 		Scheme: "https",
 		Host:   "connectiontest.argotunnel.com",
 	}
-	testObserver  = NewObserver(&log, &log, false)
 	testLargeResp = make([]byte, largeFileSize)
 )
 

connection/event.go

@@ -22,4 +22,6 @@ const (
 	SetURL
 	// RegisteringTunnel means the non-named tunnel is registering its connection.
 	RegisteringTunnel
+	// We're unregistering tunnel from the edge in preparation for a disconnect
+	Unregistering
 )

connection/h2mux.go

@@ -30,7 +30,7 @@ type h2muxConnection struct {
 	connIndex    uint8
 
 	observer          *Observer
-	gracefulShutdownC chan struct{}
+	gracefulShutdownC <-chan struct{}
 	stoppedGracefully bool
 
 	// newRPCClientFunc allows us to mock RPCs during testing
@@ -63,7 +63,7 @@ func NewH2muxConnection(
 	edgeConn net.Conn,
 	connIndex uint8,
 	observer *Observer,
-	gracefulShutdownC chan struct{},
+	gracefulShutdownC <-chan struct{},
 ) (*h2muxConnection, error, bool) {
 	h := &h2muxConnection{
 		config:            config,
@@ -168,6 +168,7 @@ func (h *h2muxConnection) serveMuxer(ctx context.Context) error {
 
 func (h *h2muxConnection) controlLoop(ctx context.Context, connectedFuse ConnectedFuse, isNamedTunnel bool) {
 	updateMetricsTickC := time.Tick(h.muxerConfig.MetricsUpdateFreq)
+	var shutdownCompleted <-chan struct{}
 	for {
 		select {
 		case <-h.gracefulShutdownC:
@@ -176,6 +177,10 @@ func (h *h2muxConnection) controlLoop(ctx context.Context, connectedFuse Connect
 			}
 			h.stoppedGracefully = true
 			h.gracefulShutdownC = nil
+			shutdownCompleted = h.muxer.Shutdown()
+
+		case <-shutdownCompleted:
+			return
 
 		case <-ctx.Done():
 			// UnregisterTunnel blocks until the RPC call returns
@@ -183,6 +188,7 @@ func (h *h2muxConnection) controlLoop(ctx context.Context, connectedFuse Connect
 				h.unregister(isNamedTunnel)
 			}
 			h.muxer.Shutdown()
+			// don't wait for shutdown to finish when context is closed, this is the hard termination path
 			return
 
 		case <-updateMetricsTickC:

connection/h2mux_test.go

@@ -33,7 +33,7 @@ func newH2MuxConnection(t require.TestingT) (*h2muxConnection, *h2mux.Muxer) {
 	edgeMuxChan := make(chan *h2mux.Muxer)
 	go func() {
 		edgeMuxConfig := h2mux.MuxerConfig{
-			Log: testObserver.log,
+			Log: &log,
 			Handler: h2mux.MuxedStreamFunc(func(stream *h2mux.MuxedStream) error {
 				// we only expect RPC traffic in client->edge direction, provide minimal support for mocking
 				require.True(t, stream.IsRPCStream())
@@ -47,6 +47,7 @@ func newH2MuxConnection(t require.TestingT) (*h2muxConnection, *h2mux.Muxer) {
 		edgeMuxChan <- edgeMux
 	}()
 	var connIndex = uint8(0)
+	testObserver := NewObserver(&log, &log, false)
 	h2muxConn, err, _ := NewH2muxConnection(testConfig, testMuxerConfig, originConn, connIndex, testObserver, nil)
 	require.NoError(t, err)
 	return h2muxConn, <-edgeMuxChan

connection/http2.go

@@ -39,7 +39,7 @@ type http2Connection struct {
 
 	activeRequestsWG  sync.WaitGroup
 	connectedFuse     ConnectedFuse
-	gracefulShutdownC chan struct{}
+	gracefulShutdownC <-chan struct{}
 	stoppedGracefully bool
 	controlStreamErr  error // result of running control stream handler
 }
@@ -52,7 +52,7 @@ func NewHTTP2Connection(
 	observer *Observer,
 	connIndex uint8,
 	connectedFuse ConnectedFuse,
-	gracefulShutdownC chan struct{},
+	gracefulShutdownC <-chan struct{},
 ) *http2Connection {
 	return &http2Connection{
 		conn: conn,
@@ -142,6 +142,7 @@ func (c *http2Connection) serveControlStream(ctx context.Context, respWriter *ht
 		c.stoppedGracefully = true
 	}
 
+	c.observer.sendUnregisteringEvent(c.connIndex)
 	rpcClient.GracefulShutdown(ctx, c.config.GracePeriod)
 	c.observer.log.Info().Uint8(LogFieldConnIndex, c.connIndex).Msg("Unregistered tunnel connection")
 	return nil

connection/http2_test.go

@@ -35,7 +35,7 @@ func newTestHTTP2Connection() (*http2Connection, net.Conn) {
 		testConfig,
 		&NamedTunnelConfig{},
 		&pogs.ConnectionOptions{},
-		testObserver,
+		NewObserver(&log, &log, false),
 		connIndex,
 		mockConnectedFuse{},
 		nil,
@@ -256,8 +256,11 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 		registered:   make(chan struct{}),
 		unregistered: make(chan struct{}),
 	}
+	events := &eventCollectorSink{}
 	http2Conn.newRPCClientFunc = rpcClientFactory.newMockRPCClient
-	http2Conn.gracefulShutdownC = make(chan struct{})
+	http2Conn.observer.RegisterSink(events)
+	shutdownC := make(chan struct{})
+	http2Conn.gracefulShutdownC = shutdownC
 
 	ctx, cancel := context.WithCancel(context.Background())
 	var wg sync.WaitGroup
@@ -288,7 +291,7 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 	}
 
 	// signal graceful shutdown
-	close(http2Conn.gracefulShutdownC)
+	close(shutdownC)
 
 	select {
 	case <-rpcClientFactory.unregistered:
@@ -300,6 +303,11 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 
 	cancel()
 	wg.Wait()
+
+	events.assertSawEvent(t, Event{
+		Index:     http2Conn.connIndex,
+		EventType: Unregistering,
+	})
 }
 
 func benchmarkServeHTTP(b *testing.B, test testRequest) {

connection/observer.go

@@ -117,6 +117,10 @@ func (o *Observer) SendReconnect(connIndex uint8) {
 	o.sendEvent(Event{Index: connIndex, EventType: Reconnecting})
 }
 
+func (o *Observer) sendUnregisteringEvent(connIndex uint8) {
+	o.sendEvent(Event{Index: connIndex, EventType: Unregistering})
+}
+
 func (o *Observer) SendDisconnect(connIndex uint8) {
 	o.sendEvent(Event{Index: connIndex, EventType: Disconnected})
 }

connection/observer_test.go

@@ -66,3 +66,21 @@ func TestObserverEventsDontBlock(t *testing.T) {
 		mu.Unlock()
 	}
 }
+
+
+type eventCollectorSink struct {
+	observedEvents []Event
+	mu sync.Mutex
+}
+
+func (s *eventCollectorSink) OnTunnelEvent(event Event) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.observedEvents = append(s.observedEvents, event)
+}
+
+func (s *eventCollectorSink) assertSawEvent(t *testing.T, event Event) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	assert.Contains(t, s.observedEvents, event)
+}

connection/rpc.go

@@ -291,6 +291,8 @@ func (h *h2muxConnection) registerNamedTunnel(
 }
 
 func (h *h2muxConnection) unregister(isNamedTunnel bool) {
+	h.observer.sendUnregisteringEvent(h.connIndex)
+
 	unregisterCtx, cancel := context.WithTimeout(context.Background(), h.config.GracePeriod)
 	defer cancel()
 

hello/hello_test.go

@@ -27,7 +27,7 @@ func TestCreateTLSListenerOnlyHostSuccess(t *testing.T) {
 }
 
 func TestCreateTLSListenerOnlyPortSuccess(t *testing.T) {
-	listener, err := CreateTLSListener(":8888")
+	listener, err := CreateTLSListener("localhost:8888")
 	if err != nil {
 		t.Fatal(err)
 	}

metrics/readiness.go

@@ -32,7 +32,7 @@ func (rs *ReadyServer) OnTunnelEvent(c conn.Event) {
 		rs.Lock()
 		rs.isConnected[int(c.Index)] = true
 		rs.Unlock()
-	case conn.Disconnected, conn.Reconnecting, conn.RegisteringTunnel:
+	case conn.Disconnected, conn.Reconnecting, conn.RegisteringTunnel, conn.Unregistering:
 		rs.Lock()
 		rs.isConnected[int(c.Index)] = false
 		rs.Unlock()

metrics/readiness_test.go

@@ -107,6 +107,22 @@ func TestReadinessEventHandling(t *testing.T) {
 	assert.NotEqualValues(t, http.StatusOK, code)
 	assert.Zero(t, ready)
 
+	// other connected then unregistered  => not ok
+	rs.OnTunnelEvent(connection.Event{
+		Index:     1,
+		EventType: connection.Connected,
+	})
+	code, ready = rs.makeResponse()
+	assert.EqualValues(t, http.StatusOK, code)
+	assert.EqualValues(t, 1, ready)
+	rs.OnTunnelEvent(connection.Event{
+		Index:     1,
+		EventType: connection.Unregistering,
+	})
+	code, ready = rs.makeResponse()
+	assert.NotEqualValues(t, http.StatusOK, code)
+	assert.Zero(t, ready)
+
 	// other disconnected => not ok
 	rs.OnTunnelEvent(connection.Event{
 		Index:     1,

origin/supervisor.go

@@ -58,16 +58,18 @@ type Supervisor struct {
 	useReconnectToken          bool
 
 	reconnectCh       chan ReconnectSignal
-	gracefulShutdownC chan struct{}
+	gracefulShutdownC <-chan struct{}
 }
 
+var errEarlyShutdown = errors.New("shutdown started")
+
 type tunnelError struct {
 	index int
 	addr  *net.TCPAddr
 	err   error
 }
 
-func NewSupervisor(config *TunnelConfig, reconnectCh chan ReconnectSignal, gracefulShutdownC chan struct{}) (*Supervisor, error) {
+func NewSupervisor(config *TunnelConfig, reconnectCh chan ReconnectSignal, gracefulShutdownC <-chan struct{}) (*Supervisor, error) {
 	cloudflaredUUID, err := uuid.NewRandom()
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate cloudflared instance ID: %w", err)
@@ -108,6 +110,9 @@ func (s *Supervisor) Run(
 	connectedSignal *signal.Signal,
 ) error {
 	if err := s.initialize(ctx, connectedSignal); err != nil {
+		if err == errEarlyShutdown {
+			return nil
+		}
 		return err
 	}
 	var tunnelsWaiting []int
@@ -130,6 +135,7 @@ func (s *Supervisor) Run(
 		}
 	}
 
+	shuttingDown := false
 	for {
 		select {
 		// Context cancelled
@@ -143,7 +149,7 @@ func (s *Supervisor) Run(
 		// (note that this may also be caused by context cancellation)
 		case tunnelError := <-s.tunnelErrors:
 			tunnelsActive--
-			if tunnelError.err != nil {
+			if tunnelError.err != nil && !shuttingDown {
 				s.log.Err(tunnelError.err).Int(connection.LogFieldConnIndex, tunnelError.index).Msg("Connection terminated")
 				tunnelsWaiting = append(tunnelsWaiting, tunnelError.index)
 				s.waitForNextTunnel(tunnelError.index)
@@ -181,6 +187,8 @@ func (s *Supervisor) Run(
 				// No more tunnels outstanding, clear backoff timer
 				backoff.SetGracePeriod()
 			}
+		case <-s.gracefulShutdownC:
+			shuttingDown = true
 		}
 	}
 }
@@ -203,6 +211,8 @@ func (s *Supervisor) initialize(
 		return ctx.Err()
 	case tunnelError := <-s.tunnelErrors:
 		return tunnelError.err
+	case <-s.gracefulShutdownC:
+		return errEarlyShutdown
 	case <-connectedSignal.Wait():
 	}
 	// At least one successful connection, so start the rest

origin/tunnel.go

@@ -113,7 +113,7 @@ func StartTunnelDaemon(
 	config *TunnelConfig,
 	connectedSignal *signal.Signal,
 	reconnectCh chan ReconnectSignal,
-	graceShutdownC chan struct{},
+	graceShutdownC <-chan struct{},
 ) error {
 	s, err := NewSupervisor(config, reconnectCh, graceShutdownC)
 	if err != nil {
@@ -131,14 +131,14 @@ func ServeTunnelLoop(
 	connectedSignal *signal.Signal,
 	cloudflaredUUID uuid.UUID,
 	reconnectCh chan ReconnectSignal,
-	gracefulShutdownC chan struct{},
+	gracefulShutdownC <-chan struct{},
 ) error {
 	haConnections.Inc()
 	defer haConnections.Dec()
 
 	connLog := config.Log.With().Uint8(connection.LogFieldConnIndex, connIndex).Logger()
 
-	protocallFallback := &protocallFallback{
+	protocolFallback := &protocolFallback{
 		BackoffHandler{MaxRetries: config.Retries},
 		config.ProtocolSelector.Current(),
 		false,
@@ -162,82 +162,82 @@ func ServeTunnelLoop(
 			addr,
 			connIndex,
 			connectedFuse,
-			protocallFallback,
+			protocolFallback,
 			cloudflaredUUID,
 			reconnectCh,
-			protocallFallback.protocol,
+			protocolFallback.protocol,
 			gracefulShutdownC,
 		)
 		if !recoverable {
 			return err
 		}
 
-		err = waitForBackoff(ctx, &connLog, protocallFallback, config, connIndex, err)
-		if err != nil {
+		config.Observer.SendReconnect(connIndex)
+
+		duration, ok := protocolFallback.GetBackoffDuration(ctx)
+		if !ok {
 			return err
 		}
+		connLog.Info().Msgf("Retrying connection in %s seconds", duration)
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-gracefulShutdownC:
+			return nil
+		case <-protocolFallback.BackoffTimer():
+			if !selectNextProtocol(&connLog, protocolFallback, config.ProtocolSelector) {
+				return err
+			}
+		}
 	}
 }
 
-// protocallFallback is a wrapper around backoffHandler that will try fallback option when backoff reaches
+// protocolFallback is a wrapper around backoffHandler that will try fallback option when backoff reaches
 // max retries
-type protocallFallback struct {
+type protocolFallback struct {
 	BackoffHandler
 	protocol   connection.Protocol
 	inFallback bool
 }
 
-func (pf *protocallFallback) reset() {
+func (pf *protocolFallback) reset() {
 	pf.resetNow()
 	pf.inFallback = false
 }
 
-func (pf *protocallFallback) fallback(fallback connection.Protocol) {
+func (pf *protocolFallback) fallback(fallback connection.Protocol) {
 	pf.resetNow()
 	pf.protocol = fallback
 	pf.inFallback = true
 }
 
-// Expect err to always be non nil
-func waitForBackoff(
-	ctx context.Context,
-	log *zerolog.Logger,
-	protobackoff *protocallFallback,
-	config *TunnelConfig,
-	connIndex uint8,
-	err error,
-) error {
-	duration, ok := protobackoff.GetBackoffDuration(ctx)
-	if !ok {
-		return err
-	}
-
-	config.Observer.SendReconnect(connIndex)
-	log.Info().
-		Err(err).
-		Uint8(connection.LogFieldConnIndex, connIndex).
-		Msgf("Retrying connection in %s seconds", duration)
-	protobackoff.Backoff(ctx)
-
-	if protobackoff.ReachedMaxRetries() {
-		fallback, hasFallback := config.ProtocolSelector.Fallback()
+// selectNextProtocol picks connection protocol for the next retry iteration,
+// returns true if it was able to pick the protocol, false if we are out of options and should stop retrying
+func selectNextProtocol(
+	connLog *zerolog.Logger,
+	protocolBackoff *protocolFallback,
+	selector connection.ProtocolSelector,
+) bool {
+	if protocolBackoff.ReachedMaxRetries() {
+		fallback, hasFallback := selector.Fallback()
 		if !hasFallback {
-			return err
+			return false
 		}
 		// Already using fallback protocol, no point to retry
-		if protobackoff.protocol == fallback {
-			return err
+		if protocolBackoff.protocol == fallback {
+			return false
 		}
-		log.Info().Msgf("Fallback to use %s", fallback)
-		protobackoff.fallback(fallback)
-	} else if !protobackoff.inFallback {
-		current := config.ProtocolSelector.Current()
-		if protobackoff.protocol != current {
-			protobackoff.protocol = current
-			config.Log.Info().Msgf("Change protocol to %s", current)
+		connLog.Info().Msgf("Switching to fallback protocol %s", fallback)
+		protocolBackoff.fallback(fallback)
+	} else if !protocolBackoff.inFallback {
+		current := selector.Current()
+		if protocolBackoff.protocol != current {
+			protocolBackoff.protocol = current
+			connLog.Info().Msgf("Changing protocol to %s", current)
 		}
 	}
-	return nil
+	return true
 }
 
 // ServeTunnel runs a single tunnel connection, returns nil on graceful shutdown,
@@ -250,11 +250,11 @@ func ServeTunnel(
 	addr *net.TCPAddr,
 	connIndex uint8,
 	fuse *h2mux.BooleanFuse,
-	backoff *protocallFallback,
+	backoff *protocolFallback,
 	cloudflaredUUID uuid.UUID,
 	reconnectCh chan ReconnectSignal,
 	protocol connection.Protocol,
-	gracefulShutdownC chan struct{},
+	gracefulShutdownC <-chan struct{},
 ) (err error, recoverable bool) {
 	// Treat panics as recoverable errors
 	defer func() {
@@ -358,7 +358,7 @@ func ServeH2mux(
 	connectedFuse *connectedFuse,
 	cloudflaredUUID uuid.UUID,
 	reconnectCh chan ReconnectSignal,
-	gracefulShutdownC chan struct{},
+	gracefulShutdownC <-chan struct{},
 ) error {
 	connLog.Debug().Msgf("Connecting via h2mux")
 	// Returns error from parsing the origin URL or handshake errors
@@ -404,7 +404,7 @@ func ServeHTTP2(
 	connIndex uint8,
 	connectedFuse connection.ConnectedFuse,
 	reconnectCh chan ReconnectSignal,
-	gracefulShutdownC chan struct{},
+	gracefulShutdownC <-chan struct{},
 ) error {
 	connLog.Debug().Msgf("Connecting via http2")
 	h2conn := connection.NewHTTP2Connection(
@@ -435,7 +435,7 @@ func ServeHTTP2(
 	return errGroup.Wait()
 }
 
-func listenReconnect(ctx context.Context, reconnectCh <-chan ReconnectSignal, gracefulShutdownCh chan struct{}) error {
+func listenReconnect(ctx context.Context, reconnectCh <-chan ReconnectSignal, gracefulShutdownCh <-chan struct{}) error {
 	select {
 	case reconnect := <-reconnectCh:
 		return reconnect
@@ -448,7 +448,7 @@ func listenReconnect(ctx context.Context, reconnectCh <-chan ReconnectSignal, gr
 
 type connectedFuse struct {
 	fuse    *h2mux.BooleanFuse
-	backoff *protocallFallback
+	backoff *protocolFallback
 }
 
 func (cf *connectedFuse) Connected() {

origin/tunnel_test.go

@@ -1,8 +1,6 @@
 package origin
 
 import (
-	"context"
-	"fmt"
 	"testing"
 	"time"
 
@@ -25,13 +23,13 @@ func (dmf *dynamicMockFetcher) fetch() connection.PercentageFetcher {
 		return dmf.percentage, nil
 	}
 }
+
 func TestWaitForBackoffFallback(t *testing.T) {
 	maxRetries := uint(3)
 	backoff := BackoffHandler{
 		MaxRetries: maxRetries,
 		BaseTime:   time.Millisecond * 10,
 	}
-	ctx := context.Background()
 	log := zerolog.Nop()
 	resolveTTL := time.Duration(0)
 	namedTunnel := &connection.NamedTunnelConfig{
@@ -50,18 +48,11 @@ func TestWaitForBackoffFallback(t *testing.T) {
 		&log,
 	)
 	assert.NoError(t, err)
-	config := &TunnelConfig{
-		Log:              &log,
-		LogTransport:     &log,
-		ProtocolSelector: protocolSelector,
-		Observer:         connection.NewObserver(&log, &log, false),
-	}
-	connIndex := uint8(1)
 
 	initProtocol := protocolSelector.Current()
 	assert.Equal(t, connection.HTTP2, initProtocol)
 
-	protocallFallback := &protocallFallback{
+	protocolFallback := &protocolFallback{
 		backoff,
 		initProtocol,
 		false,
@@ -69,29 +60,33 @@ func TestWaitForBackoffFallback(t *testing.T) {
 
 	// Retry #0 and #1. At retry #2, we switch protocol, so the fallback loop has one more retry than this
 	for i := 0; i < int(maxRetries-1); i++ {
-		err := waitForBackoff(ctx, &log, protocallFallback, config, connIndex, fmt.Errorf("some error"))
-		assert.NoError(t, err)
-		assert.Equal(t, initProtocol, protocallFallback.protocol)
+		protocolFallback.BackoffTimer() // simulate retry
+		ok := selectNextProtocol(&log, protocolFallback, protocolSelector)
+		assert.True(t, ok)
+		assert.Equal(t, initProtocol, protocolFallback.protocol)
 	}
 
 	// Retry fallback protocol
 	for i := 0; i < int(maxRetries); i++ {
-		err := waitForBackoff(ctx, &log, protocallFallback, config, connIndex, fmt.Errorf("some error"))
-		assert.NoError(t, err)
+		protocolFallback.BackoffTimer() // simulate retry
+		ok := selectNextProtocol(&log, protocolFallback, protocolSelector)
+		assert.True(t, ok)
 		fallback, ok := protocolSelector.Fallback()
 		assert.True(t, ok)
-		assert.Equal(t, fallback, protocallFallback.protocol)
+		assert.Equal(t, fallback, protocolFallback.protocol)
 	}
 
 	currentGlobalProtocol := protocolSelector.Current()
 	assert.Equal(t, initProtocol, currentGlobalProtocol)
 
 	// No protocol to fallback, return error
-	err = waitForBackoff(ctx, &log, protocallFallback, config, connIndex, fmt.Errorf("some error"))
-	assert.Error(t, err)
+	protocolFallback.BackoffTimer() // simulate retry
+	ok := selectNextProtocol(&log, protocolFallback, protocolSelector)
+	assert.False(t, ok)
 
-	protocallFallback.reset()
-	err = waitForBackoff(ctx, &log, protocallFallback, config, connIndex, fmt.Errorf("new error"))
-	assert.NoError(t, err)
-	assert.Equal(t, initProtocol, protocallFallback.protocol)
+	protocolFallback.reset()
+	protocolFallback.BackoffTimer() // simulate retry
+	ok = selectNextProtocol(&log, protocolFallback, protocolSelector)
+	assert.True(t, ok)
+	assert.Equal(t, initProtocol, protocolFallback.protocol)
 }

socks/connection_handler_test.go

@@ -42,7 +42,7 @@ func startTestServer(t *testing.T, httpHandler func(w http.ResponseWriter, r *ht
 	// create a socks server
 	requestHandler := NewRequestHandler(NewNetDialer())
 	socksServer := NewConnectionHandler(requestHandler)
-	listener, err := net.Listen("tcp", ":8086")
+	listener, err := net.Listen("tcp", "localhost:8086")
 	assert.NoError(t, err)
 
 	go func() {
@@ -58,7 +58,7 @@ func startTestServer(t *testing.T, httpHandler func(w http.ResponseWriter, r *ht
 	mux.HandleFunc("/", httpHandler)
 
 	// start the servers
-	go http.ListenAndServe(":8085", mux)
+	go http.ListenAndServe("localhost:8085", mux)
 }
 
 func respondWithJSON(w http.ResponseWriter, v interface{}, status int) {