Release 2022.7.0

Ensure service install directories are created before writing file

.github/workflows/check.yaml

@@ -12,8 +12,6 @@ jobs:
       uses: actions/setup-go@v2
       with:
         go-version: ${{ matrix.go-version }}
-    - name: Install go-sumtype
-      run: go get github.com/sudarshan-reddy/go-sumtype
     - name: Checkout code
       uses: actions/checkout@v2
     - name: Test

Dockerfile

@@ -6,6 +6,8 @@ ENV GO111MODULE=on \
     CGO_ENABLED=0 \
     TARGET_GOOS=${TARGET_GOOS} \
     TARGET_GOARCH=${TARGET_GOARCH}
+    
+LABEL org.opencontainers.image.source="https://github.com/cloudflare/cloudflared"
 
 WORKDIR /go/src/github.com/cloudflare/cloudflared/
 

Makefile

@@ -98,6 +98,16 @@ else
 	TARGET_PUBLIC_REPO ?= $(FLAVOR)
 endif
 
+ifneq ($(TARGET_ARM), )
+	ARM_COMMAND := GOARM=$(TARGET_ARM)
+endif
+
+ifeq ($(TARGET_ARM), 7) 
+	PACKAGE_ARCH := armhf
+else
+	PACKAGE_ARCH := $(TARGET_ARCH)
+endif
+
 .PHONY: all
 all: cloudflared test
 
@@ -111,7 +121,7 @@ ifeq ($(FIPS), true)
 	$(info Building cloudflared with go-fips)
 	cp -f fips/fips.go.linux-amd64 cmd/cloudflared/fips.go
 endif
-	GOOS=$(TARGET_OS) GOARCH=$(TARGET_ARCH) go build -v -mod=vendor $(GO_BUILD_TAGS) $(LDFLAGS) $(IMPORT_PATH)/cmd/cloudflared
+	GOOS=$(TARGET_OS) GOARCH=$(TARGET_ARCH) $(ARM_COMMAND) go build -v -mod=vendor $(GO_BUILD_TAGS) $(LDFLAGS) $(IMPORT_PATH)/cmd/cloudflared
 ifeq ($(FIPS), true)
 	rm -f cmd/cloudflared/fips.go
 	./check-fips.sh cloudflared
@@ -171,7 +181,7 @@ define build_package
 		--license 'Apache License Version 2.0' \
 		--url 'https://github.com/cloudflare/cloudflared' \
 		-m 'Cloudflare <support@cloudflare.com>' \
-		-a $(TARGET_ARCH) -v $(VERSION) -n $(DEB_PACKAGE_NAME) $(NIGHTLY_FLAGS) --after-install postinst.sh --after-remove postrm.sh \
+	    -a $(PACKAGE_ARCH) -v $(VERSION) -n $(DEB_PACKAGE_NAME) $(NIGHTLY_FLAGS) --after-install postinst.sh --after-remove postrm.sh \
 		cloudflared=$(INSTALL_BINDIR) cloudflared.1=$(INSTALL_MANDIR)
 endef
 
@@ -293,11 +303,6 @@ quic-deps:
 .PHONY: vet
 vet:
 	go vet -mod=vendor ./...
-	# go get github.com/sudarshan-reddy/go-sumtype (don't do this in build directory or this will cause vendor issues)
-	# Note: If you have github.com/BurntSushi/go-sumtype then you might have to use the repo above instead
-	# for now because it uses an older version of golang.org/x/tools.
-	which go-sumtype
-	go-sumtype $$(go list -mod=vendor ./...)
 
 .PHONY: goimports
 goimports:

README.md

@@ -43,7 +43,7 @@ Once installed, you can authenticate `cloudflared` into your Cloudflare account
 * Route traffic to that Tunnel:
   * Via public [DNS records in Cloudflare](https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/routing-to-tunnel/dns)
   * Or via a public hostname guided by a [Cloudflare Load Balancer](https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/routing-to-tunnel/lb)
-  * Or from [WARP client private traffic](https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/configuration/private-networks)
+  * Or from [WARP client private traffic](https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/private-net/)
 
 
 ## TryCloudflare

RELEASE_NOTES

@@ -1,3 +1,30 @@
+2022.7.0
+- 2022-07-05 TUN-6499: Remove log that is per datagram
+- 2022-06-24 TUN-6460: Rename metric label location to edge_location
+- 2022-06-24 TUN-6459: Add cloudflared user-agent to access calls
+- 2022-06-17 TUN-6427: Differentiate between upstream request closed/canceled and failed origin requests
+- 2022-06-17 TUN-6388: Fix first tunnel connection not retrying
+- 2022-06-13 TUN-6384: Correct duplicate connection error to fetch new IP first
+- 2022-06-13 TUN-6373: Add edge-ip-version to remotely pushed configuration
+- 2022-06-07 TUN-6010: Add component tests for --edge-ip-version
+- 2022-05-20 TUN-6007: Implement new edge discovery algorithm
+- 2022-02-18 Ensure service install directories are created before writing file
+
+2022.6.3
+- 2022-06-20 TUN-6362: Add armhf support to cloudflare packaging
+
+2022.6.2
+- 2022-06-13 TUN-6381: Write error data on QUIC stream when we fail to talk to the origin; separate logging for protocol errors vs. origin errors.
+- 2022-06-17 TUN-6414: Remove go-sumtype from cloudflared build process
+- 2022-06-01 Add Http2Origin option to force HTTP/2 origin connections
+- 2022-06-02 fix ingress rules unit test
+- 2022-06-09 Update remaining OriginRequestConfig functions for Http2Origins
+- 2022-05-31 Add image source label to docker container.
+- 2022-05-10 Warp Private Network link updated
+
+2022.6.1
+- 2022-06-14 TUN-6395: Fix writing RPM repo data
+
 2022.6.0
 - 2022-06-14 Revert "TUN-6010: Add component tests for --edge-ip-version"
 - 2022-06-14 Revert "TUN-6373: Add edge-ip-version to remotely pushed configuration"

build-packages.sh

@@ -17,10 +17,18 @@ for arch in ${windowsArchs[@]}; do
 done
 
 
-linuxArchs=("386" "amd64" "arm" "arm64")
+linuxArchs=("386" "amd64" "arm" "armhf" "arm64")
 export TARGET_OS=linux
 for arch in ${linuxArchs[@]}; do
+    unset TARGET_ARM
     export TARGET_ARCH=$arch
+    
+    ## Support for armhf builds 
+    if [[ $arch == armhf ]] ; then
+        export TARGET_ARCH=arm
+        export TARGET_ARM=7 
+    fi
+    
     make cloudflared-deb
     mv cloudflared\_$VERSION\_$arch.deb $ARTIFACT_DIR/cloudflared-linux-$arch.deb
 

cfsetup.yaml

@@ -153,7 +153,6 @@ stretch: &stretch
       - gotest-to-teamcity
     pre-cache: &test_pre_cache
       - go get golang.org/x/tools/cmd/goimports
-      - go get github.com/sudarshan-reddy/go-sumtype@v0.0.0-20210827105221-82eca7e5abb1
     post-cache:
       - export GOOS=linux
       - export GOARCH=amd64

cmd/cloudflared/access/cmd.go

@@ -56,11 +56,13 @@ const sentryDSN = "https://56a9c9fa5c364ab28f34b14f35ea0f1b@sentry.io/189878"
 
 var (
 	shutdownC chan struct{}
+	userAgent = "DEV"
 )
 
 // Init will initialize and store vars from the main program
-func Init(shutdown chan struct{}) {
+func Init(shutdown chan struct{}, version string) {
 	shutdownC = shutdown
+	userAgent = fmt.Sprintf("cloudflared/%s", version)
 }
 
 // Flags return the global flags for Access related commands (hopefully none)
@@ -505,7 +507,7 @@ func isTokenValid(options *carrier.StartOptions, log *zerolog.Logger) (bool, err
 	if err != nil {
 		return false, errors.Wrap(err, "Could not create access request")
 	}
-
+	req.Header.Set("User-Agent", userAgent)
 	// Do not follow redirects
 	client := &http.Client{
 		CheckRedirect: func(req *http.Request, via []*http.Request) error {

cmd/cloudflared/main.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/metrics"
 	"github.com/cloudflare/cloudflared/overwatch"
+	"github.com/cloudflare/cloudflared/token"
 	"github.com/cloudflare/cloudflared/tracing"
 	"github.com/cloudflare/cloudflared/watcher"
 )
@@ -85,9 +86,10 @@ func main() {
 	app.Commands = commands(cli.ShowVersion)
 
 	tunnel.Init(bInfo, graceShutdownC) // we need this to support the tunnel sub command...
-	access.Init(graceShutdownC)
+	access.Init(graceShutdownC, Version)
 	updater.Init(Version)
 	tracing.Init(Version)
+	token.Init(Version)
 	runApp(app, graceShutdownC)
 }
 

cmd/cloudflared/service_template.go

@@ -8,6 +8,7 @@ import (
 	"io/ioutil"
 	"os"
 	"os/exec"
+	"path"
 	"text/template"
 
 	homedir "github.com/mitchellh/go-homedir"
@@ -52,10 +53,17 @@ func (st *ServiceTemplate) Generate(args *ServiceTemplateArgs) error {
 	if err != nil {
 		return fmt.Errorf("error generating %s: %v", st.Path, err)
 	}
-	fileMode := os.FileMode(0644)
+	fileMode := os.FileMode(0o644)
 	if st.FileMode != 0 {
 		fileMode = st.FileMode
 	}
+
+	plistFolder := path.Dir(resolvedPath)
+	err = os.MkdirAll(plistFolder, 0o755)
+	if err != nil {
+		return fmt.Errorf("error creating %s: %v", plistFolder, err)
+	}
+
 	err = ioutil.WriteFile(resolvedPath, buffer.Bytes(), fileMode)
 	if err != nil {
 		return fmt.Errorf("error writing %s: %v", resolvedPath, err)

cmd/cloudflared/tunnel/cmd.go

@@ -837,6 +837,13 @@ func configureProxyFlags(shouldHide bool) []cli.Flag {
 			EnvVars: []string{"TUNNEL_NO_CHUNKED_ENCODING"},
 			Hidden:  shouldHide,
 		}),
+		altsrc.NewBoolFlag(&cli.BoolFlag{
+			Name:    ingress.Http2OriginFlag,
+			Usage:   "Enables HTTP/2 origin servers.",
+			EnvVars: []string{"TUNNEL_ORIGIN_ENABLE_HTTP2"},
+			Hidden:  shouldHide,
+			Value:   false,
+		}),
 	}
 	return append(flags, sshFlags(shouldHide)...)
 }

cmd/cloudflared/tunnel/configuration.go

@@ -45,7 +45,7 @@ var (
 	secretFlags     = [2]*altsrc.StringFlag{credentialsContentsFlag, tunnelTokenFlag}
 	defaultFeatures = []string{supervisor.FeatureAllowRemoteConfig, supervisor.FeatureSerializedHeaders}
 
-	configFlags = []string{"autoupdate-freq", "no-autoupdate", "retries", "protocol", "loglevel", "transport-loglevel", "origincert", "metrics", "metrics-update-freq"}
+	configFlags = []string{"autoupdate-freq", "no-autoupdate", "retries", "protocol", "loglevel", "transport-loglevel", "origincert", "metrics", "metrics-update-freq", "edge-ip-version"}
 )
 
 // returns the first path that contains a cert.pem file. If none of the DefaultConfigSearchDirectories

component-tests/cli.py

@@ -0,0 +1,95 @@
+import json
+import subprocess
+from time import sleep
+
+from setup import get_config_from_file
+
+SINGLE_CASE_TIMEOUT = 600
+
+class CloudflaredCli:
+    def __init__(self, config, config_path, logger):
+        self.basecmd = [config.cloudflared_binary, "tunnel"]
+        if config_path is not None:
+            self.basecmd += ["--config", str(config_path)]
+        origincert = get_config_from_file()["origincert"]
+        if origincert:
+            self.basecmd += ["--origincert", origincert]
+        self.logger = logger
+
+    def _run_command(self, subcmd, subcmd_name, needs_to_pass=True):
+        cmd = self.basecmd + subcmd
+        # timeout limits the time a subprocess can run. This is useful to guard against running a tunnel when
+        # command/args are in wrong order.
+        result = run_subprocess(cmd, subcmd_name, self.logger, check=needs_to_pass, capture_output=True, timeout=15)
+        return result
+
+    def list_tunnels(self):
+        cmd_args = ["list", "--output", "json"]
+        listed = self._run_command(cmd_args, "list")
+        return json.loads(listed.stdout)
+
+    def get_tunnel_info(self, tunnel_id):
+        info = self._run_command(["info", "--output", "json", tunnel_id], "info")
+        return json.loads(info.stdout)
+
+    def __enter__(self):
+        self.basecmd += ["run"]
+        self.process = subprocess.Popen(self.basecmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        self.logger.info(f"Run cmd {self.basecmd}")
+        return self.process
+
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        terminate_gracefully(self.process, self.logger, self.basecmd)
+        self.logger.debug(f"{self.basecmd} logs: {self.process.stderr.read()}")
+
+
+def terminate_gracefully(process, logger, cmd):
+    process.terminate()
+    process_terminated = wait_for_terminate(process)
+    if not process_terminated:
+        process.kill()
+        logger.warning(f"{cmd}: cloudflared did not terminate within wait period. Killing process. logs: \
+                stdout: {process.stdout.read()}, stderr: {process.stderr.read()}")
+
+
+def wait_for_terminate(opened_subprocess, attempts=10, poll_interval=1):
+    """
+        wait_for_terminate polls the opened_subprocess every x seconds for a given number of attempts.
+        It returns true if the subprocess was terminated and false if it didn't.
+    """
+    for _ in range(attempts):
+        if _is_process_stopped(opened_subprocess):
+            return True
+        sleep(poll_interval)
+    return False
+
+
+def _is_process_stopped(process):
+    return process.poll() is not None
+
+
+def cert_path():
+    return get_config_from_file()["origincert"]
+
+
+class SubprocessError(Exception):
+    def __init__(self, program, exit_code, cause):
+        self.program = program
+        self.exit_code = exit_code
+        self.cause = cause
+
+
+def run_subprocess(cmd, cmd_name, logger, timeout=SINGLE_CASE_TIMEOUT, **kargs):
+    kargs["timeout"] = timeout
+    try:
+        result = subprocess.run(cmd, **kargs)
+        logger.debug(f"{cmd} log: {result.stdout}", extra={"cmd": cmd_name})
+        return result
+    except subprocess.CalledProcessError as e:
+        err = f"{cmd} return exit code {e.returncode}, stderr" + e.stderr.decode("utf-8")
+        logger.error(err, extra={"cmd": cmd_name, "return_code": e.returncode})
+        raise SubprocessError(cmd[0], e.returncode, e)
+    except subprocess.TimeoutExpired as e:
+        err = f"{cmd} timeout after {e.timeout} seconds, stdout: {e.stdout}, stderr: {e.stderr}"
+        logger.error(err, extra={"cmd": cmd_name, "return_code": "timeout"})
+        raise e

component-tests/test_edge_discovery.py

@@ -0,0 +1,165 @@
+import ipaddress
+import socket
+
+import pytest
+
+from constants import protocols
+from cli import CloudflaredCli
+from util import get_tunnel_connector_id, LOGGER, wait_tunnel_ready, write_config
+
+
+class TestEdgeDiscovery:
+    def _extra_config(self, protocol, edge_ip_version):
+        config = {
+            "protocol": protocol,
+        }
+        if edge_ip_version:
+            config["edge-ip-version"] = edge_ip_version
+        return config
+
+    @pytest.mark.parametrize("protocol", protocols())
+    def test_default_only(self, tmp_path, component_tests_config, protocol):
+        """
+        This test runs a tunnel to connect via IPv4-only edge addresses (default is unset "--edge-ip-version 4")
+        """
+        if self.has_ipv6_only():
+            pytest.skip("Host has IPv6 only support and current default is IPv4 only")
+        self.expect_address_connections(
+            tmp_path, component_tests_config, protocol, None, self.expect_ipv4_address)
+
+    @pytest.mark.parametrize("protocol", protocols())
+    def test_ipv4_only(self, tmp_path, component_tests_config, protocol):
+        """
+        This test runs a tunnel to connect via IPv4-only edge addresses
+        """
+        if self.has_ipv6_only():
+            pytest.skip("Host has IPv6 only support")
+        self.expect_address_connections(
+            tmp_path, component_tests_config, protocol, "4", self.expect_ipv4_address)
+
+    @pytest.mark.parametrize("protocol", protocols())
+    def test_ipv6_only(self, tmp_path, component_tests_config, protocol):
+        """
+        This test runs a tunnel to connect via IPv6-only edge addresses
+        """
+        if self.has_ipv4_only():
+            pytest.skip("Host has IPv4 only support")
+        self.expect_address_connections(
+            tmp_path, component_tests_config, protocol, "6", self.expect_ipv6_address)
+
+    @pytest.mark.parametrize("protocol", protocols())
+    def test_auto_ip64(self, tmp_path, component_tests_config, protocol):
+        """
+        This test runs a tunnel to connect via auto with a preference of IPv6 then IPv4 addresses for a dual stack host
+
+        This test also assumes that the host has IPv6 preference.
+        """
+        if not self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET6):
+            pytest.skip("Host does not support dual stack with IPv6 preference")
+        self.expect_address_connections(
+            tmp_path, component_tests_config, protocol, "auto", self.expect_ipv6_address)
+
+    @pytest.mark.parametrize("protocol", protocols())
+    def test_auto_ip46(self, tmp_path, component_tests_config, protocol):
+        """
+        This test runs a tunnel to connect via auto with a preference of IPv4 then IPv6 addresses for a dual stack host
+
+        This test also assumes that the host has IPv4 preference.
+        """
+        if not self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET):
+            pytest.skip("Host does not support dual stack with IPv4 preference")
+        self.expect_address_connections(
+            tmp_path, component_tests_config, protocol, "auto", self.expect_ipv4_address)
+
+    def expect_address_connections(self, tmp_path, component_tests_config, protocol, edge_ip_version, assert_address_type):
+        config = component_tests_config(
+            self._extra_config(protocol, edge_ip_version))
+        config_path = write_config(tmp_path, config.full_config)
+        LOGGER.debug(config)
+        with CloudflaredCli(config, config_path, LOGGER):
+            wait_tunnel_ready(tunnel_url=config.get_url(),
+                              require_min_connections=4)
+            cfd_cli = CloudflaredCli(config, config_path, LOGGER)
+            tunnel_id = config.get_tunnel_id()
+            info = cfd_cli.get_tunnel_info(tunnel_id)
+            connector_id = get_tunnel_connector_id()
+            connector = next(
+                (c for c in info["conns"] if c["id"] == connector_id), None)
+            assert connector, f"Expected connection info from get tunnel info for the connected instance: {info}"
+            conns = connector["conns"]
+            assert conns == None or len(
+                conns) == 4, f"There should be 4 connections registered: {conns}"
+            for conn in conns:
+                origin_ip = conn["origin_ip"]
+                assert origin_ip, f"No available origin_ip for this connection: {conn}"
+                assert_address_type(origin_ip)
+
+    def expect_ipv4_address(self, address):
+        assert type(ipaddress.ip_address(
+            address)) is ipaddress.IPv4Address, f"Expected connection from origin to be a valid IPv4 address: {address}"
+
+    def expect_ipv6_address(self, address):
+        assert type(ipaddress.ip_address(
+            address)) is ipaddress.IPv6Address, f"Expected connection from origin to be a valid IPv6 address: {address}"
+
+    def get_addresses(self):
+        """
+        Returns a list of addresses for the host.
+        """
+        host_addresses = socket.getaddrinfo(
+            "region1.v2.argotunnel.com", 7844, socket.AF_UNSPEC, socket.SOCK_STREAM)
+        assert len(
+            host_addresses) > 0, "No addresses returned from getaddrinfo"
+        return host_addresses
+
+    def has_dual_stack(self, address_family_preference=None):
+        """
+        Returns true if the host has dual stack support and can optionally check 
+        the provided IP family preference.
+        """
+        dual_stack = not self.has_ipv6_only() and not self.has_ipv4_only()
+        if address_family_preference:
+            address = self.get_addresses()[0]
+            return dual_stack and address[0] == address_family_preference
+
+        return dual_stack
+
+    def has_ipv6_only(self):
+        """
+        Returns True if the host has only IPv6 address support.
+        """
+        return self.attempt_connection(socket.AddressFamily.AF_INET6) and not self.attempt_connection(socket.AddressFamily.AF_INET)
+
+    def has_ipv4_only(self):
+        """
+        Returns True if the host has only IPv4 address support.
+        """
+        return self.attempt_connection(socket.AddressFamily.AF_INET) and not self.attempt_connection(socket.AddressFamily.AF_INET6)
+
+    def attempt_connection(self, address_family):
+        """
+        Returns True if a successful socket connection can be made to the 
+        remote host with the provided address family to validate host support 
+        for the provided address family.
+        """
+        address = None
+        for a in self.get_addresses():
+            if a[0] == address_family:
+                address = a
+                break
+        if address is None:
+            # Couldn't even lookup the address family so we can't connect
+            return False
+        af, socktype, proto, canonname, sockaddr = address
+        s = None
+        try:
+            s = socket.socket(af, socktype, proto)
+        except OSError:
+            return False
+        try:
+            s.connect(sockaddr)
+        except OSError:
+            s.close()
+            return False
+        s.close()
+        return True

component-tests/test_service.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python
 import os
 import pathlib
-import platform
 import subprocess
 from contextlib import contextmanager
 from pathlib import Path
@@ -10,12 +9,7 @@ import pytest
 
 import test_logging
 from conftest import CfdModes
-from util import start_cloudflared, wait_tunnel_ready, write_config
-
-
-def select_platform(plat):
-    return pytest.mark.skipif(
-        platform.system() != plat, reason=f"Only runs on {plat}")
+from util import select_platform, start_cloudflared, wait_tunnel_ready, write_config
 
 
 def default_config_dir():

component-tests/util.py

@@ -1,9 +1,12 @@
 import logging
 import os
+import platform
 import subprocess
 from contextlib import contextmanager
 from time import sleep
 
+import pytest
+
 import requests
 import yaml
 from retrying import retry
@@ -12,6 +15,10 @@ from constants import METRICS_PORT, MAX_RETRIES, BACKOFF_SECS
 
 LOGGER = logging.getLogger(__name__)
 
+def select_platform(plat):
+    return pytest.mark.skipif(
+        platform.system() != plat, reason=f"Only runs on {plat}")
+
 
 def write_config(directory, config):
     config_path = directory / "config.yml"
@@ -111,6 +118,17 @@ def check_tunnel_not_connected():
         LOGGER.warning(f"Failed to connect to {url}, error: {e}")
 
 
+def get_tunnel_connector_id():
+    url = f'http://localhost:{METRICS_PORT}/ready'
+
+    try:
+        resp = requests.get(url, timeout=1)
+        return resp.json()["connectorId"]
+    # cloudflared might already terminated
+    except requests.exceptions.ConnectionError as e:
+        LOGGER.warning(f"Failed to connect to {url}, error: {e}")
+
+
 # In some cases we don't need to check response status, such as when sending batch requests to generate logs
 def send_requests(url, count, require_ok=True):
     errors = 0

config/configuration.go

@@ -227,6 +227,8 @@ type OriginRequestConfig struct {
 	ProxyType *string `yaml:"proxyType" json:"proxyType,omitempty"`
 	// IP rules for the proxy service
 	IPRules []IngressIPRule `yaml:"ipRules" json:"ipRules,omitempty"`
+	// Attempt to connect to origin with HTTP/2
+	Http2Origin *bool `yaml:"http2Origin" json:"http2Origin,omitempty"`
 }
 
 type IngressIPRule struct {

config/configuration_test.go

@@ -144,7 +144,8 @@ var rawJsonConfig = []byte(`
 			"ports": [443, 4443],
 			"allow": true
 		}
-	]
+	],
+	"http2Origin": true
 }
 `)
 
@@ -191,6 +192,7 @@ func assertConfig(
 	assert.Equal(t, true, *config.NoTLSVerify)
 	assert.Equal(t, uint(9000), *config.ProxyPort)
 	assert.Equal(t, "socks", *config.ProxyType)
+	assert.Equal(t, true, *config.Http2Origin)
 
 	privateV4 := "10.0.0.0/8"
 	privateV6 := "fc00::/7"

connection/control.go

@@ -3,6 +3,7 @@ package connection
 import (
 	"context"
 	"io"
+	"net"
 	"time"
 
 	"github.com/rs/zerolog"
@@ -19,6 +20,7 @@ type controlStream struct {
 	connectedFuse         ConnectedFuse
 	namedTunnelProperties *NamedTunnelProperties
 	connIndex             uint8
+	edgeAddress           net.IP
 
 	newRPCClientFunc RPCClientFunc
 
@@ -45,6 +47,7 @@ func NewControlStream(
 	connectedFuse ConnectedFuse,
 	namedTunnelConfig *NamedTunnelProperties,
 	connIndex uint8,
+	edgeAddress net.IP,
 	newRPCClientFunc RPCClientFunc,
 	gracefulShutdownC <-chan struct{},
 	gracePeriod time.Duration,
@@ -58,6 +61,7 @@ func NewControlStream(
 		namedTunnelProperties: namedTunnelConfig,
 		newRPCClientFunc:      newRPCClientFunc,
 		connIndex:             connIndex,
+		edgeAddress:           edgeAddress,
 		gracefulShutdownC:     gracefulShutdownC,
 		gracePeriod:           gracePeriod,
 	}
@@ -71,7 +75,7 @@ func (c *controlStream) ServeControlStream(
 ) error {
 	rpcClient := c.newRPCClientFunc(ctx, rw, c.observer.log)
 
-	registrationDetails, err := rpcClient.RegisterConnection(ctx, c.namedTunnelProperties, connOptions, c.connIndex, c.observer)
+	registrationDetails, err := rpcClient.RegisterConnection(ctx, c.namedTunnelProperties, connOptions, c.connIndex, c.edgeAddress, c.observer)
 	if err != nil {
 		rpcClient.Close()
 		return err

connection/errors.go

@@ -18,6 +18,15 @@ func (e DupConnRegisterTunnelError) Error() string {
 	return "already connected to this server, trying another address"
 }
 
+// Dial to edge server with quic failed
+type EdgeQuicDialError struct {
+	Cause error
+}
+
+func (e *EdgeQuicDialError) Error() string {
+	return "failed to dial to edge with quic: " + e.Cause.Error()
+}
+
 // RegisterTunnel error from server
 type ServerRegisterTunnelError struct {
 	Cause     error

connection/http2_test.go

@@ -41,6 +41,7 @@ func newTestHTTP2Connection() (*HTTP2Connection, net.Conn) {
 		connIndex,
 		nil,
 		nil,
+		nil,
 		1*time.Second,
 	)
 	return NewHTTP2Connection(
@@ -176,6 +177,7 @@ func (mc mockNamedTunnelRPCClient) RegisterConnection(
 	properties *NamedTunnelProperties,
 	options *tunnelpogs.ConnectionOptions,
 	connIndex uint8,
+	edgeAddress net.IP,
 	observer *Observer,
 ) (*tunnelpogs.ConnectionDetails, error) {
 	if mc.shouldFail != nil {
@@ -360,6 +362,7 @@ func TestServeControlStream(t *testing.T) {
 		mockConnectedFuse{},
 		&NamedTunnelProperties{},
 		1,
+		nil,
 		rpcClientFactory.newMockRPCClient,
 		nil,
 		1*time.Second,
@@ -410,6 +413,7 @@ func TestFailRegistration(t *testing.T) {
 		mockConnectedFuse{},
 		&NamedTunnelProperties{},
 		http2Conn.connIndex,
+		nil,
 		rpcClientFactory.newMockRPCClient,
 		nil,
 		1*time.Second,
@@ -456,6 +460,7 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 		mockConnectedFuse{},
 		&NamedTunnelProperties{},
 		http2Conn.connIndex,
+		nil,
 		rpcClientFactory.newMockRPCClient,
 		shutdownC,
 		1*time.Second,

connection/metrics.go

@@ -367,7 +367,7 @@ func initTunnelMetrics() *tunnelMetrics {
 			Name:      "server_locations",
 			Help:      "Where each tunnel is connected to. 1 means current location, 0 means previous locations.",
 		},
-		[]string{"connection_id", "location"},
+		[]string{"connection_id", "edge_location"},
 	)
 	prometheus.MustRegister(serverLocations)
 

connection/quic.go

@@ -57,7 +57,7 @@ func NewQUICConnection(
 ) (*QUICConnection, error) {
 	session, err := quic.DialAddr(edgeAddr.String(), tlsConfig, quicConfig)
 	if err != nil {
-		return nil, fmt.Errorf("failed to dial to edge: %w", err)
+		return nil, &EdgeQuicDialError{Cause: err}
 	}
 
 	datagramMuxer, err := quicpogs.NewDatagramMuxer(session, logger)
@@ -160,7 +160,7 @@ func (q *QUICConnection) handleStream(stream io.ReadWriteCloser) error {
 	case quicpogs.DataStreamProtocolSignature:
 		reqServerStream, err := quicpogs.NewRequestServerStream(stream, signature)
 		if err != nil {
-			return nil
+			return err
 		}
 		return q.handleDataStream(reqServerStream)
 	case quicpogs.RPCStreamProtocolSignature:
@@ -175,30 +175,41 @@ func (q *QUICConnection) handleStream(stream io.ReadWriteCloser) error {
 }
 
 func (q *QUICConnection) handleDataStream(stream *quicpogs.RequestServerStream) error {
-	connectRequest, err := stream.ReadConnectRequestData()
+	request, err := stream.ReadConnectRequestData()
 	if err != nil {
 		return err
 	}
 
+	if err := q.dispatchRequest(stream, err, request); err != nil {
+		_ = stream.WriteConnectResponseData(err)
+		q.logger.Err(err).Str("type", request.Type.String()).Str("dest", request.Dest).Msg("Request failed")
+	}
+
+	return nil
+}
+
+func (q *QUICConnection) dispatchRequest(stream *quicpogs.RequestServerStream, err error, request *quicpogs.ConnectRequest) error {
 	originProxy, err := q.orchestrator.GetOriginProxy()
 	if err != nil {
 		return err
 	}
 
-	switch connectRequest.Type {
+	switch request.Type {
 	case quicpogs.ConnectionTypeHTTP, quicpogs.ConnectionTypeWebsocket:
-		tracedReq, err := buildHTTPRequest(connectRequest, stream)
+		tracedReq, err := buildHTTPRequest(request, stream)
 		if err != nil {
 			return err
 		}
-
 		w := newHTTPResponseAdapter(stream)
-		return originProxy.ProxyHTTP(w, tracedReq, connectRequest.Type == quicpogs.ConnectionTypeWebsocket)
+		return originProxy.ProxyHTTP(w, tracedReq, request.Type == quicpogs.ConnectionTypeWebsocket)
+
 	case quicpogs.ConnectionTypeTCP:
 		rwa := &streamReadWriteAcker{stream}
-		metadata := connectRequest.MetadataMap()
-		return originProxy.ProxyTCP(context.Background(), rwa, &TCPRequest{Dest: connectRequest.Dest,
-			FlowID: metadata[QUICMetadataFlowID]})
+		metadata := request.MetadataMap()
+		return originProxy.ProxyTCP(context.Background(), rwa, &TCPRequest{
+			Dest:   request.Dest,
+			FlowID: metadata[QUICMetadataFlowID],
+		})
 	}
 	return nil
 }

connection/rpc.go

@@ -58,6 +58,7 @@ type NamedTunnelRPCClient interface {
 		config *NamedTunnelProperties,
 		options *tunnelpogs.ConnectionOptions,
 		connIndex uint8,
+		edgeAddress net.IP,
 		observer *Observer,
 	) (*tunnelpogs.ConnectionDetails, error)
 	SendLocalConfiguration(
@@ -95,6 +96,7 @@ func (rsc *registrationServerClient) RegisterConnection(
 	properties *NamedTunnelProperties,
 	options *tunnelpogs.ConnectionOptions,
 	connIndex uint8,
+	edgeAddress net.IP,
 	observer *Observer,
 ) (*tunnelpogs.ConnectionDetails, error) {
 	conn, err := rsc.client.RegisterConnection(
@@ -115,7 +117,7 @@ func (rsc *registrationServerClient) RegisterConnection(
 
 	observer.metrics.regSuccess.WithLabelValues("registerConnection").Inc()
 
-	observer.logServerInfo(connIndex, conn.Location, options.OriginLocalIP, fmt.Sprintf("Connection %s registered", conn.UUID))
+	observer.logServerInfo(connIndex, conn.Location, edgeAddress, fmt.Sprintf("Connection %s registered", conn.UUID))
 	observer.sendConnectedEvent(connIndex, conn.Location)
 
 	return conn, nil
@@ -291,7 +293,7 @@ func (h *h2muxConnection) registerNamedTunnel(
 	rpcClient := h.newRPCClientFunc(ctx, stream, h.observer.log)
 	defer rpcClient.Close()
 
-	if _, err = rpcClient.RegisterConnection(ctx, namedTunnel, connOptions, h.connIndex, h.observer); err != nil {
+	if _, err = rpcClient.RegisterConnection(ctx, namedTunnel, connOptions, h.connIndex, nil, h.observer); err != nil {
 		return err
 	}
 	return nil

edgediscovery/allregions/address.go

@@ -0,0 +1,64 @@
+package allregions
+
+// Region contains cloudflared edge addresses. The edge is partitioned into several regions for
+// redundancy purposes.
+type AddrSet map[*EdgeAddr]UsedBy
+
+// AddrUsedBy finds the address used by the given connection in this region.
+// Returns nil if the connection isn't using any IP.
+func (a AddrSet) AddrUsedBy(connID int) *EdgeAddr {
+	for addr, used := range a {
+		if used.Used && used.ConnID == connID {
+			return addr
+		}
+	}
+	return nil
+}
+
+// AvailableAddrs counts how many unused addresses this region contains.
+func (a AddrSet) AvailableAddrs() int {
+	n := 0
+	for _, usedby := range a {
+		if !usedby.Used {
+			n++
+		}
+	}
+	return n
+}
+
+// GetUnusedIP returns a random unused address in this region.
+// Returns nil if all addresses are in use.
+func (a AddrSet) GetUnusedIP(excluding *EdgeAddr) *EdgeAddr {
+	for addr, usedby := range a {
+		if !usedby.Used && addr != excluding {
+			return addr
+		}
+	}
+	return nil
+}
+
+// Use the address, assigning it to a proxy connection.
+func (a AddrSet) Use(addr *EdgeAddr, connID int) {
+	if addr == nil {
+		return
+	}
+	a[addr] = InUse(connID)
+}
+
+// GetAnyAddress returns an arbitrary address from the region.
+func (a AddrSet) GetAnyAddress() *EdgeAddr {
+	for addr := range a {
+		return addr
+	}
+	return nil
+}
+
+// GiveBack the address, ensuring it is no longer assigned to an IP.
+// Returns true if the address is in this region.
+func (a AddrSet) GiveBack(addr *EdgeAddr) (ok bool) {
+	if _, ok := a[addr]; !ok {
+		return false
+	}
+	a[addr] = Unused()
+	return true
+}

edgediscovery/allregions/address_test.go

@@ -0,0 +1,247 @@
+package allregions
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestAddrSet_AddrUsedBy(t *testing.T) {
+	type args struct {
+		connID int
+	}
+	tests := []struct {
+		name    string
+		addrSet AddrSet
+		args    args
+		want    *EdgeAddr
+	}{
+		{
+			name: "happy trivial test",
+			addrSet: AddrSet{
+				&addr0: InUse(0),
+			},
+			args: args{connID: 0},
+			want: &addr0,
+		},
+		{
+			name: "sad trivial test",
+			addrSet: AddrSet{
+				&addr0: InUse(0),
+			},
+			args: args{connID: 1},
+			want: nil,
+		},
+		{
+			name: "sad test",
+			addrSet: AddrSet{
+				&addr0: InUse(0),
+				&addr1: InUse(1),
+				&addr2: InUse(2),
+			},
+			args: args{connID: 3},
+			want: nil,
+		},
+		{
+			name: "happy test",
+			addrSet: AddrSet{
+				&addr0: InUse(0),
+				&addr1: InUse(1),
+				&addr2: InUse(2),
+			},
+			args: args{connID: 1},
+			want: &addr1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.addrSet.AddrUsedBy(tt.args.connID); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Region.AddrUsedBy() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestAddrSet_AvailableAddrs(t *testing.T) {
+	tests := []struct {
+		name    string
+		addrSet AddrSet
+		want    int
+	}{
+		{
+			name: "contains addresses",
+			addrSet: AddrSet{
+				&addr0: InUse(0),
+				&addr1: Unused(),
+				&addr2: InUse(2),
+			},
+			want: 1,
+		},
+		{
+			name: "all free",
+			addrSet: AddrSet{
+				&addr0: Unused(),
+				&addr1: Unused(),
+				&addr2: Unused(),
+			},
+			want: 3,
+		},
+		{
+			name: "all used",
+			addrSet: AddrSet{
+				&addr0: InUse(0),
+				&addr1: InUse(1),
+				&addr2: InUse(2),
+			},
+			want: 0,
+		},
+		{
+			name:    "empty",
+			addrSet: AddrSet{},
+			want:    0,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.addrSet.AvailableAddrs(); got != tt.want {
+				t.Errorf("Region.AvailableAddrs() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestAddrSet_GetUnusedIP(t *testing.T) {
+	type args struct {
+		excluding *EdgeAddr
+	}
+	tests := []struct {
+		name    string
+		addrSet AddrSet
+		args    args
+		want    *EdgeAddr
+	}{
+		{
+			name: "happy test with excluding set",
+			addrSet: AddrSet{
+				&addr0: Unused(),
+				&addr1: Unused(),
+				&addr2: InUse(2),
+			},
+			args: args{excluding: &addr0},
+			want: &addr1,
+		},
+		{
+			name: "happy test with no excluding",
+			addrSet: AddrSet{
+				&addr0: InUse(0),
+				&addr1: Unused(),
+				&addr2: InUse(2),
+			},
+			args: args{excluding: nil},
+			want: &addr1,
+		},
+		{
+			name: "sad test with no excluding",
+			addrSet: AddrSet{
+				&addr0: InUse(0),
+				&addr1: InUse(1),
+				&addr2: InUse(2),
+			},
+			args: args{excluding: nil},
+			want: nil,
+		},
+		{
+			name: "sad test with excluding",
+			addrSet: AddrSet{
+				&addr0: Unused(),
+				&addr1: InUse(1),
+				&addr2: InUse(2),
+			},
+			args: args{excluding: &addr0},
+			want: nil,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.addrSet.GetUnusedIP(tt.args.excluding); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Region.GetUnusedIP() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestAddrSet_GiveBack(t *testing.T) {
+	type args struct {
+		addr *EdgeAddr
+	}
+	tests := []struct {
+		name           string
+		addrSet        AddrSet
+		args           args
+		wantOk         bool
+		availableAfter int
+	}{
+		{
+			name: "sad test with excluding",
+			addrSet: AddrSet{
+				&addr1: InUse(1),
+			},
+			args:           args{addr: &addr1},
+			wantOk:         true,
+			availableAfter: 1,
+		},
+		{
+			name: "sad test with excluding",
+			addrSet: AddrSet{
+				&addr1: InUse(1),
+			},
+			args:           args{addr: &addr2},
+			wantOk:         false,
+			availableAfter: 0,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if gotOk := tt.addrSet.GiveBack(tt.args.addr); gotOk != tt.wantOk {
+				t.Errorf("Region.GiveBack() = %v, want %v", gotOk, tt.wantOk)
+			}
+			if tt.availableAfter != tt.addrSet.AvailableAddrs() {
+				t.Errorf("Region.AvailableAddrs() = %v, want %v", tt.addrSet.AvailableAddrs(), tt.availableAfter)
+			}
+		})
+	}
+}
+
+func TestAddrSet_GetAnyAddress(t *testing.T) {
+	tests := []struct {
+		name    string
+		addrSet AddrSet
+		wantNil bool
+	}{
+		{
+			name:    "Sad test -- GetAnyAddress should only fail if the region is empty",
+			addrSet: AddrSet{},
+			wantNil: true,
+		},
+		{
+			name: "Happy test (all addresses unused)",
+			addrSet: AddrSet{
+				&addr0: Unused(),
+			},
+			wantNil: false,
+		},
+		{
+			name: "Happy test (GetAnyAddress can still return addresses used by proxy conns)",
+			addrSet: AddrSet{
+				&addr0: InUse(2),
+			},
+			wantNil: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.addrSet.GetAnyAddress(); tt.wantNil != (got == nil) {
+				t.Errorf("Region.GetAnyAddress() = %v, but should it return nil? %v", got, tt.wantNil)
+			}
+		})
+	}
+}

edgediscovery/allregions/discovery.go

@@ -13,7 +13,7 @@ import (
 
 const (
 	// Used to discover HA origintunneld servers
-	srvService = "origintunneld"
+	srvService = "v2-origintunneld"
 	srvProto   = "tcp"
 	srvName    = "argotunnel.com"
 
@@ -115,6 +115,9 @@ func edgeDiscovery(log *zerolog.Logger, srvService string) ([][]*EdgeAddr, error
 		if err != nil {
 			return nil, err
 		}
+		for _, e := range edgeAddrs {
+			log.Debug().Msgf("Edge Address: %+v", *e)
+		}
 		resolvedAddrPerCNAME = append(resolvedAddrPerCNAME, edgeAddrs)
 	}
 
@@ -187,7 +190,6 @@ func ResolveAddrs(addrs []string, log *zerolog.Logger) (resolved []*EdgeAddr) {
 			UDP:       udpAddr,
 			IPVersion: version,
 		})
-
 	}
 	return
 }

edgediscovery/allregions/mocks_for_test.go

@@ -9,6 +9,115 @@ import (
 	"testing/quick"
 )
 
+var (
+	v4Addrs = []*EdgeAddr{&addr0, &addr1, &addr2, &addr3}
+	v6Addrs = []*EdgeAddr{&addr4, &addr5, &addr6, &addr7}
+	addr0   = EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("123.4.5.0"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("123.4.5.0"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V4,
+	}
+	addr1 = EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("123.4.5.1"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("123.4.5.1"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V4,
+	}
+	addr2 = EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("123.4.5.2"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("123.4.5.2"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V4,
+	}
+	addr3 = EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("123.4.5.3"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("123.4.5.3"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V4,
+	}
+	addr4 = EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("2606:4700:a0::1"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("2606:4700:a0::1"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V6,
+	}
+	addr5 = EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("2606:4700:a0::2"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("2606:4700:a0::2"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V6,
+	}
+	addr6 = EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("2606:4700:a0::3"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("2606:4700:a0::3"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V6,
+	}
+	addr7 = EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("2606:4700:a0::4"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("2606:4700:a0::4"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V6,
+	}
+)
+
 type mockAddrs struct {
 	// a set of synthetic SRV records
 	addrMap map[net.SRV][]*EdgeAddr

edgediscovery/allregions/region.go

@@ -1,79 +1,155 @@
 package allregions
 
+import "time"
+
+const (
+	timeoutDuration = 10 * time.Minute
+)
+
 // Region contains cloudflared edge addresses. The edge is partitioned into several regions for
 // redundancy purposes.
 type Region struct {
-	connFor map[*EdgeAddr]UsedBy
+	primaryIsActive bool
+	active          AddrSet
+	primary         AddrSet
+	secondary       AddrSet
+	primaryTimeout  time.Time
+	timeoutDuration time.Duration
 }
 
 // NewRegion creates a region with the given addresses, which are all unused.
-func NewRegion(addrs []*EdgeAddr) Region {
+func NewRegion(addrs []*EdgeAddr, overrideIPVersion ConfigIPVersion) Region {
 	// The zero value of UsedBy is Unused(), so we can just initialize the map's values with their
 	// zero values.
-	connFor := make(map[*EdgeAddr]UsedBy)
-	for _, addr := range addrs {
-		connFor[addr] = Unused()
+	connForv4 := make(AddrSet)
+	connForv6 := make(AddrSet)
+	systemPreference := V6
+	for i, addr := range addrs {
+		if i == 0 {
+			// First family of IPs returned is system preference of IP
+			systemPreference = addr.IPVersion
+		}
+		switch addr.IPVersion {
+		case V4:
+			connForv4[addr] = Unused()
+		case V6:
+			connForv6[addr] = Unused()
+		}
 	}
+
+	// Process as system preference
+	var primary AddrSet
+	var secondary AddrSet
+	switch systemPreference {
+	case V4:
+		primary = connForv4
+		secondary = connForv6
+	case V6:
+		primary = connForv6
+		secondary = connForv4
+	}
+
+	// Override with provided preference
+	switch overrideIPVersion {
+	case IPv4Only:
+		primary = connForv4
+		secondary = make(AddrSet) // empty
+	case IPv6Only:
+		primary = connForv6
+		secondary = make(AddrSet) // empty
+	case Auto:
+		// no change
+	default:
+		// no change
+	}
+
 	return Region{
-		connFor: connFor,
+		primaryIsActive: true,
+		active:          primary,
+		primary:         primary,
+		secondary:       secondary,
+		timeoutDuration: timeoutDuration,
 	}
 }
 
 // AddrUsedBy finds the address used by the given connection in this region.
 // Returns nil if the connection isn't using any IP.
 func (r *Region) AddrUsedBy(connID int) *EdgeAddr {
-	for addr, used := range r.connFor {
-		if used.Used && used.ConnID == connID {
-			return addr
-		}
+	edgeAddr := r.primary.AddrUsedBy(connID)
+	if edgeAddr == nil {
+		edgeAddr = r.secondary.AddrUsedBy(connID)
 	}
-	return nil
+	return edgeAddr
 }
 
 // AvailableAddrs counts how many unused addresses this region contains.
 func (r Region) AvailableAddrs() int {
-	n := 0
-	for _, usedby := range r.connFor {
-		if !usedby.Used {
-			n++
-		}
-	}
-	return n
+	return r.active.AvailableAddrs()
 }
 
-// GetUnusedIP returns a random unused address in this region.
-// Returns nil if all addresses are in use.
-func (r Region) GetUnusedIP(excluding *EdgeAddr) *EdgeAddr {
-	for addr, usedby := range r.connFor {
-		if !usedby.Used && addr != excluding {
-			return addr
-		}
-	}
-	return nil
-}
-
-// Use the address, assigning it to a proxy connection.
-func (r Region) Use(addr *EdgeAddr, connID int) {
-	if addr == nil {
-		return
-	}
-	r.connFor[addr] = InUse(connID)
-}
-
-// GetAnyAddress returns an arbitrary address from the region.
-func (r Region) GetAnyAddress() *EdgeAddr {
-	for addr := range r.connFor {
+// AssignAnyAddress returns a random unused address in this region now
+// assigned to the connID excluding the provided EdgeAddr.
+// Returns nil if all addresses are in use for the region.
+func (r Region) AssignAnyAddress(connID int, excluding *EdgeAddr) *EdgeAddr {
+	if addr := r.active.GetUnusedIP(excluding); addr != nil {
+		r.active.Use(addr, connID)
 		return addr
 	}
 	return nil
 }
 
+// GetAnyAddress returns an arbitrary address from the region.
+func (r Region) GetAnyAddress() *EdgeAddr {
+	return r.active.GetAnyAddress()
+}
+
 // GiveBack the address, ensuring it is no longer assigned to an IP.
 // Returns true if the address is in this region.
-func (r Region) GiveBack(addr *EdgeAddr) (ok bool) {
-	if _, ok := r.connFor[addr]; !ok {
-		return false
+func (r *Region) GiveBack(addr *EdgeAddr, hasConnectivityError bool) (ok bool) {
+	if ok = r.primary.GiveBack(addr); !ok {
+		// Attempt to give back the address in the secondary set
+		if ok = r.secondary.GiveBack(addr); !ok {
+			// Address is not in this region
+			return
+		}
 	}
-	r.connFor[addr] = Unused()
-	return true
+
+	// No connectivity error: no worry
+	if !hasConnectivityError {
+		return
+	}
+
+	// If using primary and returned address is IPv6 and secondary is available
+	if r.primaryIsActive && addr.IPVersion == V6 && len(r.secondary) > 0 {
+		r.active = r.secondary
+		r.primaryIsActive = false
+		r.primaryTimeout = time.Now().Add(r.timeoutDuration)
+		return
+	}
+
+	// Do nothing for IPv4 or if secondary is empty
+	if r.primaryIsActive {
+		return
+	}
+
+	// Immediately return to primary pool, regardless of current primary timeout
+	if addr.IPVersion == V4 {
+		activatePrimary(r)
+		return
+	}
+
+	// Timeout exceeded and can be reset to primary pool
+	if r.primaryTimeout.Before(time.Now()) {
+		activatePrimary(r)
+		return
+	}
+
+	return
+}
+
+// activatePrimary sets the primary set to the active set and resets the timeout.
+func activatePrimary(r *Region) {
+	r.active = r.primary
+	r.primaryIsActive = true
+	r.primaryTimeout = time.Now() // reset timeout
 }

edgediscovery/allregions/region_test.go

@@ -1,284 +1,357 @@
 package allregions
 
 import (
-	"reflect"
+	"net"
 	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
+func makeAddrSet(addrs []*EdgeAddr) AddrSet {
+	addrSet := make(AddrSet, len(addrs))
+	for _, addr := range addrs {
+		addrSet[addr] = Unused()
+	}
+	return addrSet
+}
+
 func TestRegion_New(t *testing.T) {
-	r := NewRegion([]*EdgeAddr{&addr0, &addr1, &addr2})
-	if r.AvailableAddrs() != 3 {
-		t.Errorf("r.AvailableAddrs() == %v but want 3", r.AvailableAddrs())
-	}
-}
-
-func TestRegion_AddrUsedBy(t *testing.T) {
-	type fields struct {
-		connFor map[*EdgeAddr]UsedBy
-	}
-	type args struct {
-		connID int
-	}
 	tests := []struct {
-		name   string
-		fields fields
-		args   args
-		want   *EdgeAddr
+		name          string
+		addrs         []*EdgeAddr
+		mode          ConfigIPVersion
+		expectedAddrs int
+		primary       AddrSet
+		secondary     AddrSet
 	}{
 		{
-			name: "happy trivial test",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(0),
-			}},
-			args: args{connID: 0},
-			want: &addr0,
+			name:          "IPv4 addresses with IPv4Only",
+			addrs:         v4Addrs,
+			mode:          IPv4Only,
+			expectedAddrs: len(v4Addrs),
+			primary:       makeAddrSet(v4Addrs),
+			secondary:     AddrSet{},
 		},
 		{
-			name: "sad trivial test",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(0),
-			}},
-			args: args{connID: 1},
-			want: nil,
+			name:          "IPv6 addresses with IPv4Only",
+			addrs:         v6Addrs,
+			mode:          IPv4Only,
+			expectedAddrs: 0,
+			primary:       AddrSet{},
+			secondary:     AddrSet{},
 		},
 		{
-			name: "sad test",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(0),
-				&addr1: InUse(1),
-				&addr2: InUse(2),
-			}},
-			args: args{connID: 3},
-			want: nil,
+			name:          "IPv6 addresses with IPv6Only",
+			addrs:         v6Addrs,
+			mode:          IPv6Only,
+			expectedAddrs: len(v6Addrs),
+			primary:       makeAddrSet(v6Addrs),
+			secondary:     AddrSet{},
 		},
 		{
-			name: "happy test",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(0),
-				&addr1: InUse(1),
-				&addr2: InUse(2),
-			}},
-			args: args{connID: 1},
-			want: &addr1,
+			name:          "IPv6 addresses with IPv4Only",
+			addrs:         v6Addrs,
+			mode:          IPv4Only,
+			expectedAddrs: 0,
+			primary:       AddrSet{},
+			secondary:     AddrSet{},
+		},
+		{
+			name:          "IPv4 (first) and IPv6 addresses with Auto",
+			addrs:         append(v4Addrs, v6Addrs...),
+			mode:          Auto,
+			expectedAddrs: len(v4Addrs),
+			primary:       makeAddrSet(v4Addrs),
+			secondary:     makeAddrSet(v6Addrs),
+		},
+		{
+			name:          "IPv6 (first) and IPv4 addresses with Auto",
+			addrs:         append(v6Addrs, v4Addrs...),
+			mode:          Auto,
+			expectedAddrs: len(v6Addrs),
+			primary:       makeAddrSet(v6Addrs),
+			secondary:     makeAddrSet(v4Addrs),
+		},
+		{
+			name:          "IPv4 addresses with Auto",
+			addrs:         v4Addrs,
+			mode:          Auto,
+			expectedAddrs: len(v4Addrs),
+			primary:       makeAddrSet(v4Addrs),
+			secondary:     AddrSet{},
+		},
+		{
+			name:          "IPv6 addresses with Auto",
+			addrs:         v6Addrs,
+			mode:          Auto,
+			expectedAddrs: len(v6Addrs),
+			primary:       makeAddrSet(v6Addrs),
+			secondary:     AddrSet{},
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			r := &Region{
-				connFor: tt.fields.connFor,
+			r := NewRegion(tt.addrs, tt.mode)
+			assert.Equal(t, tt.expectedAddrs, r.AvailableAddrs())
+			assert.Equal(t, tt.primary, r.primary)
+			assert.Equal(t, tt.secondary, r.secondary)
+		})
+	}
+}
+
+func TestRegion_AnyAddress_EmptyActiveSet(t *testing.T) {
+	tests := []struct {
+		name  string
+		addrs []*EdgeAddr
+		mode  ConfigIPVersion
+	}{
+		{
+			name:  "IPv6 addresses with IPv4Only",
+			addrs: v6Addrs,
+			mode:  IPv4Only,
+		},
+		{
+			name:  "IPv4 addresses with IPv6Only",
+			addrs: v4Addrs,
+			mode:  IPv6Only,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := NewRegion(tt.addrs, tt.mode)
+			addr := r.GetAnyAddress()
+			assert.Nil(t, addr)
+			addr = r.AssignAnyAddress(0, nil)
+			assert.Nil(t, addr)
+		})
+	}
+}
+
+func TestRegion_AssignAnyAddress_FullyUsedActiveSet(t *testing.T) {
+	tests := []struct {
+		name  string
+		addrs []*EdgeAddr
+		mode  ConfigIPVersion
+	}{
+		{
+			name:  "IPv6 addresses with IPv6Only",
+			addrs: v6Addrs,
+			mode:  IPv6Only,
+		},
+		{
+			name:  "IPv4 addresses with IPv4Only",
+			addrs: v4Addrs,
+			mode:  IPv4Only,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := NewRegion(tt.addrs, tt.mode)
+			total := r.active.AvailableAddrs()
+			for i := 0; i < total; i++ {
+				addr := r.AssignAnyAddress(i, nil)
+				assert.NotNil(t, addr)
 			}
-			if got := r.AddrUsedBy(tt.args.connID); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("Region.AddrUsedBy() = %v, want %v", got, tt.want)
+			addr := r.AssignAnyAddress(9, nil)
+			assert.Nil(t, addr)
+		})
+	}
+}
+
+var giveBackTests = []struct {
+	name          string
+	addrs         []*EdgeAddr
+	mode          ConfigIPVersion
+	expectedAddrs int
+	primary       AddrSet
+	secondary     AddrSet
+	primarySwap   bool
+}{
+	{
+		name:          "IPv4 addresses with IPv4Only",
+		addrs:         v4Addrs,
+		mode:          IPv4Only,
+		expectedAddrs: len(v4Addrs),
+		primary:       makeAddrSet(v4Addrs),
+		secondary:     AddrSet{},
+		primarySwap:   false,
+	},
+	{
+		name:          "IPv6 addresses with IPv6Only",
+		addrs:         v6Addrs,
+		mode:          IPv6Only,
+		expectedAddrs: len(v6Addrs),
+		primary:       makeAddrSet(v6Addrs),
+		secondary:     AddrSet{},
+		primarySwap:   false,
+	},
+	{
+		name:          "IPv4 (first) and IPv6 addresses with Auto",
+		addrs:         append(v4Addrs, v6Addrs...),
+		mode:          Auto,
+		expectedAddrs: len(v4Addrs),
+		primary:       makeAddrSet(v4Addrs),
+		secondary:     makeAddrSet(v6Addrs),
+		primarySwap:   false,
+	},
+	{
+		name:          "IPv6 (first) and IPv4 addresses with Auto",
+		addrs:         append(v6Addrs, v4Addrs...),
+		mode:          Auto,
+		expectedAddrs: len(v6Addrs),
+		primary:       makeAddrSet(v6Addrs),
+		secondary:     makeAddrSet(v4Addrs),
+		primarySwap:   true,
+	},
+	{
+		name:          "IPv4 addresses with Auto",
+		addrs:         v4Addrs,
+		mode:          Auto,
+		expectedAddrs: len(v4Addrs),
+		primary:       makeAddrSet(v4Addrs),
+		secondary:     AddrSet{},
+		primarySwap:   false,
+	},
+	{
+		name:          "IPv6 addresses with Auto",
+		addrs:         v6Addrs,
+		mode:          Auto,
+		expectedAddrs: len(v6Addrs),
+		primary:       makeAddrSet(v6Addrs),
+		secondary:     AddrSet{},
+		primarySwap:   false,
+	},
+}
+
+func TestRegion_GiveBack_NoConnectivityError(t *testing.T) {
+	for _, tt := range giveBackTests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := NewRegion(tt.addrs, tt.mode)
+			addr := r.AssignAnyAddress(0, nil)
+			assert.NotNil(t, addr)
+			assert.True(t, r.GiveBack(addr, false))
+		})
+	}
+}
+
+func TestRegion_GiveBack_ForeignAddr(t *testing.T) {
+	invalid := EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("123.4.5.0"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("123.4.5.0"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: V4,
+	}
+	for _, tt := range giveBackTests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := NewRegion(tt.addrs, tt.mode)
+			assert.False(t, r.GiveBack(&invalid, false))
+			assert.False(t, r.GiveBack(&invalid, true))
+		})
+	}
+}
+
+func TestRegion_GiveBack_SwapPrimary(t *testing.T) {
+	for _, tt := range giveBackTests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := NewRegion(tt.addrs, tt.mode)
+			addr := r.AssignAnyAddress(0, nil)
+			assert.NotNil(t, addr)
+			assert.True(t, r.GiveBack(addr, true))
+			assert.Equal(t, tt.primarySwap, !r.primaryIsActive)
+			if tt.primarySwap {
+				assert.Equal(t, r.secondary, r.active)
+				assert.False(t, r.primaryTimeout.IsZero())
+			} else {
+				assert.Equal(t, r.primary, r.active)
+				assert.True(t, r.primaryTimeout.IsZero())
 			}
 		})
 	}
 }
 
-func TestRegion_AvailableAddrs(t *testing.T) {
-	type fields struct {
-		connFor map[*EdgeAddr]UsedBy
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		want   int
-	}{
-		{
-			name: "contains addresses",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(0),
-				&addr1: Unused(),
-				&addr2: InUse(2),
-			}},
-			want: 1,
-		},
-		{
-			name: "all free",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: Unused(),
-				&addr1: Unused(),
-				&addr2: Unused(),
-			}},
-			want: 3,
-		},
-		{
-			name: "all used",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(0),
-				&addr1: InUse(1),
-				&addr2: InUse(2),
-			}},
-			want: 0,
-		},
-		{
-			name:   "empty",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{}},
-			want:   0,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			r := Region{
-				connFor: tt.fields.connFor,
-			}
-			if got := r.AvailableAddrs(); got != tt.want {
-				t.Errorf("Region.AvailableAddrs() = %v, want %v", got, tt.want)
-			}
-		})
-	}
+func TestRegion_GiveBack_IPv4_ResetPrimary(t *testing.T) {
+	r := NewRegion(append(v6Addrs, v4Addrs...), Auto)
+	// Exhaust all IPv6 addresses
+	a0 := r.AssignAnyAddress(0, nil)
+	a1 := r.AssignAnyAddress(1, nil)
+	a2 := r.AssignAnyAddress(2, nil)
+	a3 := r.AssignAnyAddress(3, nil)
+	assert.NotNil(t, a0)
+	assert.NotNil(t, a1)
+	assert.NotNil(t, a2)
+	assert.NotNil(t, a3)
+	// Give back the first IPv6 address to fallback to secondary IPv4 address set
+	assert.True(t, r.GiveBack(a0, true))
+	assert.False(t, r.primaryIsActive)
+	// Give back another IPv6 address
+	assert.True(t, r.GiveBack(a1, true))
+	// Primary shouldn't change
+	assert.False(t, r.primaryIsActive)
+	// Request an address (should be IPv4 from secondary)
+	a4_v4 := r.AssignAnyAddress(4, nil)
+	assert.NotNil(t, a4_v4)
+	assert.Equal(t, V4, a4_v4.IPVersion)
+	a5_v4 := r.AssignAnyAddress(5, nil)
+	assert.NotNil(t, a5_v4)
+	assert.Equal(t, V4, a5_v4.IPVersion)
+	a6_v4 := r.AssignAnyAddress(6, nil)
+	assert.NotNil(t, a6_v4)
+	assert.Equal(t, V4, a6_v4.IPVersion)
+	// Return IPv4 address (without failure)
+	// Primary shouldn't change because it is not a connectivity failure
+	assert.True(t, r.GiveBack(a4_v4, false))
+	assert.False(t, r.primaryIsActive)
+	// Return IPv4 address (with failure)
+	// Primary should change because it is a connectivity failure
+	assert.True(t, r.GiveBack(a5_v4, true))
+	assert.True(t, r.primaryIsActive)
+	// Return IPv4 address (with failure)
+	// Primary shouldn't change because the address is returned to the inactive
+	// secondary address set
+	assert.True(t, r.GiveBack(a6_v4, true))
+	assert.True(t, r.primaryIsActive)
+	// Return IPv6 address (without failure)
+	// Primary shoudn't change because it is not a connectivity failure
+	assert.True(t, r.GiveBack(a2, false))
+	assert.True(t, r.primaryIsActive)
 }
 
-func TestRegion_GetUnusedIP(t *testing.T) {
-	type fields struct {
-		connFor map[*EdgeAddr]UsedBy
-	}
-	type args struct {
-		excluding *EdgeAddr
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		args   args
-		want   *EdgeAddr
-	}{
-		{
-			name: "happy test with excluding set",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: Unused(),
-				&addr1: Unused(),
-				&addr2: InUse(2),
-			}},
-			args: args{excluding: &addr0},
-			want: &addr1,
-		},
-		{
-			name: "happy test with no excluding",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(0),
-				&addr1: Unused(),
-				&addr2: InUse(2),
-			}},
-			args: args{excluding: nil},
-			want: &addr1,
-		},
-		{
-			name: "sad test with no excluding",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(0),
-				&addr1: InUse(1),
-				&addr2: InUse(2),
-			}},
-			args: args{excluding: nil},
-			want: nil,
-		},
-		{
-			name: "sad test with excluding",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: Unused(),
-				&addr1: InUse(1),
-				&addr2: InUse(2),
-			}},
-			args: args{excluding: &addr0},
-			want: nil,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			r := Region{
-				connFor: tt.fields.connFor,
-			}
-			if got := r.GetUnusedIP(tt.args.excluding); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("Region.GetUnusedIP() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestRegion_GiveBack(t *testing.T) {
-	type fields struct {
-		connFor map[*EdgeAddr]UsedBy
-	}
-	type args struct {
-		addr *EdgeAddr
-	}
-	tests := []struct {
-		name           string
-		fields         fields
-		args           args
-		wantOk         bool
-		availableAfter int
-	}{
-		{
-			name: "sad test with excluding",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr1: InUse(1),
-			}},
-			args:           args{addr: &addr1},
-			wantOk:         true,
-			availableAfter: 1,
-		},
-		{
-			name: "sad test with excluding",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr1: InUse(1),
-			}},
-			args:           args{addr: &addr2},
-			wantOk:         false,
-			availableAfter: 0,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			r := Region{
-				connFor: tt.fields.connFor,
-			}
-			if gotOk := r.GiveBack(tt.args.addr); gotOk != tt.wantOk {
-				t.Errorf("Region.GiveBack() = %v, want %v", gotOk, tt.wantOk)
-			}
-			if tt.availableAfter != r.AvailableAddrs() {
-				t.Errorf("Region.AvailableAddrs() = %v, want %v", r.AvailableAddrs(), tt.availableAfter)
-			}
-		})
-	}
-}
-
-func TestRegion_GetAnyAddress(t *testing.T) {
-	type fields struct {
-		connFor map[*EdgeAddr]UsedBy
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		wantNil bool
-	}{
-		{
-			name:    "Sad test -- GetAnyAddress should only fail if the region is empty",
-			fields:  fields{connFor: map[*EdgeAddr]UsedBy{}},
-			wantNil: true,
-		},
-		{
-			name: "Happy test (all addresses unused)",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: Unused(),
-			}},
-			wantNil: false,
-		},
-		{
-			name: "Happy test (GetAnyAddress can still return addresses used by proxy conns)",
-			fields: fields{connFor: map[*EdgeAddr]UsedBy{
-				&addr0: InUse(2),
-			}},
-			wantNil: false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			r := Region{
-				connFor: tt.fields.connFor,
-			}
-			if got := r.GetAnyAddress(); tt.wantNil != (got == nil) {
-				t.Errorf("Region.GetAnyAddress() = %v, but should it return nil? %v", got, tt.wantNil)
-			}
-		})
-	}
+func TestRegion_GiveBack_Timeout(t *testing.T) {
+	r := NewRegion(append(v6Addrs, v4Addrs...), Auto)
+	a0 := r.AssignAnyAddress(0, nil)
+	a1 := r.AssignAnyAddress(1, nil)
+	a2 := r.AssignAnyAddress(2, nil)
+	assert.NotNil(t, a0)
+	assert.NotNil(t, a1)
+	assert.NotNil(t, a2)
+	// Give back IPv6 address to set timeout
+	assert.True(t, r.GiveBack(a0, true))
+	assert.False(t, r.primaryIsActive)
+	assert.False(t, r.primaryTimeout.IsZero())
+	// Request an address (should be IPv4 from secondary)
+	a3_v4 := r.AssignAnyAddress(3, nil)
+	assert.NotNil(t, a3_v4)
+	assert.Equal(t, V4, a3_v4.IPVersion)
+	assert.False(t, r.primaryIsActive)
+	// Give back IPv6 address inside timeout (no change)
+	assert.True(t, r.GiveBack(a2, true))
+	assert.False(t, r.primaryIsActive)
+	assert.False(t, r.primaryTimeout.IsZero())
+	// Accelerate timeout
+	r.primaryTimeout = time.Now().Add(-time.Minute)
+	// Return IPv6 address
+	assert.True(t, r.GiveBack(a1, true))
+	assert.True(t, r.primaryIsActive)
+	// Returning an IPv4 address after primary is active shouldn't change primary
+	// even with a connectivity error
+	assert.True(t, r.GiveBack(a3_v4, true))
+	assert.True(t, r.primaryIsActive)
 }

edgediscovery/allregions/regions.go

@@ -18,7 +18,7 @@ type Regions struct {
 // ------------------------------------
 
 // ResolveEdge resolves the Cloudflare edge, returning all regions discovered.
-func ResolveEdge(log *zerolog.Logger, region string) (*Regions, error) {
+func ResolveEdge(log *zerolog.Logger, region string, overrideIPVersion ConfigIPVersion) (*Regions, error) {
 	edgeAddrs, err := edgeDiscovery(log, getRegionalServiceName(region))
 	if err != nil {
 		return nil, err
@@ -27,8 +27,8 @@ func ResolveEdge(log *zerolog.Logger, region string) (*Regions, error) {
 		return nil, fmt.Errorf("expected at least 2 Cloudflare Regions regions, but SRV only returned %v", len(edgeAddrs))
 	}
 	return &Regions{
-		region1: NewRegion(edgeAddrs[0]),
-		region2: NewRegion(edgeAddrs[1]),
+		region1: NewRegion(edgeAddrs[0], overrideIPVersion),
+		region2: NewRegion(edgeAddrs[1], overrideIPVersion),
 	}, nil
 }
 
@@ -56,8 +56,8 @@ func NewNoResolve(addrs []*EdgeAddr) *Regions {
 	}
 
 	return &Regions{
-		region1: NewRegion(region1),
-		region2: NewRegion(region2),
+		region1: NewRegion(region1, Auto),
+		region2: NewRegion(region2, Auto),
 	}
 }
 
@@ -95,14 +95,12 @@ func (rs *Regions) GetUnusedAddr(excluding *EdgeAddr, connID int) *EdgeAddr {
 // getAddrs tries to grab address form `first` region, then `second` region
 // this is an unrolled loop over 2 element array
 func getAddrs(excluding *EdgeAddr, connID int, first *Region, second *Region) *EdgeAddr {
-	addr := first.GetUnusedIP(excluding)
+	addr := first.AssignAnyAddress(connID, excluding)
 	if addr != nil {
-		first.Use(addr, connID)
 		return addr
 	}
-	addr = second.GetUnusedIP(excluding)
+	addr = second.AssignAnyAddress(connID, excluding)
 	if addr != nil {
-		second.Use(addr, connID)
 		return addr
 	}
 
@@ -116,18 +114,18 @@ func (rs *Regions) AvailableAddrs() int {
 
 // GiveBack the address so that other connections can use it.
 // Returns true if the address is in this edge.
-func (rs *Regions) GiveBack(addr *EdgeAddr) bool {
-	if found := rs.region1.GiveBack(addr); found {
+func (rs *Regions) GiveBack(addr *EdgeAddr, hasConnectivityError bool) bool {
+	if found := rs.region1.GiveBack(addr, hasConnectivityError); found {
 		return found
 	}
-	return rs.region2.GiveBack(addr)
+	return rs.region2.GiveBack(addr, hasConnectivityError)
 }
 
 // Return regionalized service name if `region` isn't empty, otherwise return the global service name for origintunneld
 func getRegionalServiceName(region string) string {
 	if region != "" {
-		return region + "-" + srvService // Example: `us-origintunneld`
+		return region + "-" + srvService // Example: `us-v2-origintunneld`
 	}
 
-	return srvService // Global service is just `origintunneld`
+	return srvService // Global service is just `v2-origintunneld`
 }

edgediscovery/allregions/regions_test.go

@@ -1,134 +1,215 @@
 package allregions
 
 import (
-	"net"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
-var (
-	addr0 = EdgeAddr{
-		TCP: &net.TCPAddr{
-			IP:   net.ParseIP("123.4.5.0"),
-			Port: 8000,
-			Zone: "",
-		},
-		UDP: &net.UDPAddr{
-			IP:   net.ParseIP("123.4.5.0"),
-			Port: 8000,
-			Zone: "",
-		},
+func makeRegions(addrs []*EdgeAddr, mode ConfigIPVersion) Regions {
+	r1addrs := make([]*EdgeAddr, 0)
+	r2addrs := make([]*EdgeAddr, 0)
+	for i, addr := range addrs {
+		if i%2 == 0 {
+			r1addrs = append(r1addrs, addr)
+		} else {
+			r2addrs = append(r2addrs, addr)
+		}
 	}
-	addr1 = EdgeAddr{
-		TCP: &net.TCPAddr{
-			IP:   net.ParseIP("123.4.5.1"),
-			Port: 8000,
-			Zone: "",
-		},
-		UDP: &net.UDPAddr{
-			IP:   net.ParseIP("123.4.5.1"),
-			Port: 8000,
-			Zone: "",
-		},
-	}
-	addr2 = EdgeAddr{
-		TCP: &net.TCPAddr{
-			IP:   net.ParseIP("123.4.5.2"),
-			Port: 8000,
-			Zone: "",
-		},
-		UDP: &net.UDPAddr{
-			IP:   net.ParseIP("123.4.5.2"),
-			Port: 8000,
-			Zone: "",
-		},
-	}
-	addr3 = EdgeAddr{
-		TCP: &net.TCPAddr{
-			IP:   net.ParseIP("123.4.5.3"),
-			Port: 8000,
-			Zone: "",
-		},
-		UDP: &net.UDPAddr{
-			IP:   net.ParseIP("123.4.5.3"),
-			Port: 8000,
-			Zone: "",
-		},
-	}
-)
-
-func makeRegions() Regions {
-	r1 := NewRegion([]*EdgeAddr{&addr0, &addr1})
-	r2 := NewRegion([]*EdgeAddr{&addr2, &addr3})
+	r1 := NewRegion(r1addrs, mode)
+	r2 := NewRegion(r2addrs, mode)
 	return Regions{region1: r1, region2: r2}
 }
 
 func TestRegions_AddrUsedBy(t *testing.T) {
-	rs := makeRegions()
-	addr1 := rs.GetUnusedAddr(nil, 1)
-	assert.Equal(t, addr1, rs.AddrUsedBy(1))
-	addr2 := rs.GetUnusedAddr(nil, 2)
-	assert.Equal(t, addr2, rs.AddrUsedBy(2))
-	addr3 := rs.GetUnusedAddr(nil, 3)
-	assert.Equal(t, addr3, rs.AddrUsedBy(3))
+	tests := []struct {
+		name  string
+		addrs []*EdgeAddr
+		mode  ConfigIPVersion
+	}{
+		{
+			name:  "IPv4 addresses with IPv4Only",
+			addrs: v4Addrs,
+			mode:  IPv4Only,
+		},
+		{
+			name:  "IPv6 addresses with IPv6Only",
+			addrs: v6Addrs,
+			mode:  IPv6Only,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rs := makeRegions(tt.addrs, tt.mode)
+			addr1 := rs.GetUnusedAddr(nil, 1)
+			assert.Equal(t, addr1, rs.AddrUsedBy(1))
+			addr2 := rs.GetUnusedAddr(nil, 2)
+			assert.Equal(t, addr2, rs.AddrUsedBy(2))
+			addr3 := rs.GetUnusedAddr(nil, 3)
+			assert.Equal(t, addr3, rs.AddrUsedBy(3))
+		})
+	}
 }
 
 func TestRegions_Giveback_Region1(t *testing.T) {
-	rs := makeRegions()
-	rs.region1.Use(&addr0, 0)
-	rs.region1.Use(&addr1, 1)
-	rs.region2.Use(&addr2, 2)
-	rs.region2.Use(&addr3, 3)
+	tests := []struct {
+		name  string
+		addrs []*EdgeAddr
+		mode  ConfigIPVersion
+	}{
+		{
+			name:  "IPv4 addresses with IPv4Only",
+			addrs: v4Addrs,
+			mode:  IPv4Only,
+		},
+		{
+			name:  "IPv6 addresses with IPv6Only",
+			addrs: v6Addrs,
+			mode:  IPv6Only,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rs := makeRegions(tt.addrs, tt.mode)
+			addr := rs.region1.AssignAnyAddress(0, nil)
+			rs.region1.AssignAnyAddress(1, nil)
+			rs.region2.AssignAnyAddress(2, nil)
+			rs.region2.AssignAnyAddress(3, nil)
 
-	assert.Equal(t, 0, rs.AvailableAddrs())
+			assert.Equal(t, 0, rs.AvailableAddrs())
 
-	rs.GiveBack(&addr0)
-	assert.Equal(t, &addr0, rs.GetUnusedAddr(nil, 3))
+			rs.GiveBack(addr, false)
+			assert.Equal(t, addr, rs.GetUnusedAddr(nil, 0))
+		})
+	}
 }
 
 func TestRegions_Giveback_Region2(t *testing.T) {
-	rs := makeRegions()
-	rs.region1.Use(&addr0, 0)
-	rs.region1.Use(&addr1, 1)
-	rs.region2.Use(&addr2, 2)
-	rs.region2.Use(&addr3, 3)
+	tests := []struct {
+		name  string
+		addrs []*EdgeAddr
+		mode  ConfigIPVersion
+	}{
+		{
+			name:  "IPv4 addresses with IPv4Only",
+			addrs: v4Addrs,
+			mode:  IPv4Only,
+		},
+		{
+			name:  "IPv6 addresses with IPv6Only",
+			addrs: v6Addrs,
+			mode:  IPv6Only,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rs := makeRegions(tt.addrs, tt.mode)
+			rs.region1.AssignAnyAddress(0, nil)
+			rs.region1.AssignAnyAddress(1, nil)
+			addr := rs.region2.AssignAnyAddress(2, nil)
+			rs.region2.AssignAnyAddress(3, nil)
 
-	assert.Equal(t, 0, rs.AvailableAddrs())
+			assert.Equal(t, 0, rs.AvailableAddrs())
 
-	rs.GiveBack(&addr2)
-	assert.Equal(t, &addr2, rs.GetUnusedAddr(nil, 2))
+			rs.GiveBack(addr, false)
+			assert.Equal(t, addr, rs.GetUnusedAddr(nil, 2))
+		})
+	}
 }
 
 func TestRegions_GetUnusedAddr_OneAddrLeft(t *testing.T) {
-	rs := makeRegions()
+	tests := []struct {
+		name  string
+		addrs []*EdgeAddr
+		mode  ConfigIPVersion
+	}{
+		{
+			name:  "IPv4 addresses with IPv4Only",
+			addrs: v4Addrs,
+			mode:  IPv4Only,
+		},
+		{
+			name:  "IPv6 addresses with IPv6Only",
+			addrs: v6Addrs,
+			mode:  IPv6Only,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rs := makeRegions(tt.addrs, tt.mode)
+			rs.region1.AssignAnyAddress(0, nil)
+			rs.region1.AssignAnyAddress(1, nil)
+			rs.region2.AssignAnyAddress(2, nil)
+			addr := rs.region2.active.GetUnusedIP(nil)
 
-	rs.region1.Use(&addr0, 0)
-	rs.region1.Use(&addr1, 1)
-	rs.region2.Use(&addr2, 2)
-
-	assert.Equal(t, 1, rs.AvailableAddrs())
-	assert.Equal(t, &addr3, rs.GetUnusedAddr(nil, 3))
+			assert.Equal(t, 1, rs.AvailableAddrs())
+			assert.Equal(t, addr, rs.GetUnusedAddr(nil, 3))
+		})
+	}
 }
 
 func TestRegions_GetUnusedAddr_Excluding_Region1(t *testing.T) {
-	rs := makeRegions()
+	tests := []struct {
+		name  string
+		addrs []*EdgeAddr
+		mode  ConfigIPVersion
+	}{
+		{
+			name:  "IPv4 addresses with IPv4Only",
+			addrs: v4Addrs,
+			mode:  IPv4Only,
+		},
+		{
+			name:  "IPv6 addresses with IPv6Only",
+			addrs: v6Addrs,
+			mode:  IPv6Only,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rs := makeRegions(tt.addrs, tt.mode)
 
-	rs.region1.Use(&addr0, 0)
-	rs.region1.Use(&addr1, 1)
+			rs.region1.AssignAnyAddress(0, nil)
+			rs.region1.AssignAnyAddress(1, nil)
+			addr := rs.region2.active.GetUnusedIP(nil)
+			a2 := rs.region2.active.GetUnusedIP(addr)
 
-	assert.Equal(t, 2, rs.AvailableAddrs())
-	assert.Equal(t, &addr3, rs.GetUnusedAddr(&addr2, 3))
+			assert.Equal(t, 2, rs.AvailableAddrs())
+			assert.Equal(t, addr, rs.GetUnusedAddr(a2, 3))
+		})
+	}
 }
 
 func TestRegions_GetUnusedAddr_Excluding_Region2(t *testing.T) {
-	rs := makeRegions()
+	tests := []struct {
+		name  string
+		addrs []*EdgeAddr
+		mode  ConfigIPVersion
+	}{
+		{
+			name:  "IPv4 addresses with IPv4Only",
+			addrs: v4Addrs,
+			mode:  IPv4Only,
+		},
+		{
+			name:  "IPv6 addresses with IPv6Only",
+			addrs: v6Addrs,
+			mode:  IPv6Only,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rs := makeRegions(tt.addrs, tt.mode)
 
-	rs.region2.Use(&addr2, 0)
-	rs.region2.Use(&addr3, 1)
+			rs.region2.AssignAnyAddress(0, nil)
+			rs.region2.AssignAnyAddress(1, nil)
+			addr := rs.region1.active.GetUnusedIP(nil)
+			a2 := rs.region1.active.GetUnusedIP(addr)
 
-	assert.Equal(t, 2, rs.AvailableAddrs())
-	assert.Equal(t, &addr1, rs.GetUnusedAddr(&addr0, 1))
+			assert.Equal(t, 2, rs.AvailableAddrs())
+			assert.Equal(t, addr, rs.GetUnusedAddr(a2, 1))
+		})
+	}
 }
 
 func TestNewNoResolveBalancesRegions(t *testing.T) {

edgediscovery/edgediscovery.go

@@ -1,7 +1,6 @@
 package edgediscovery
 
 import (
-	"fmt"
 	"sync"
 
 	"github.com/rs/zerolog"
@@ -11,9 +10,16 @@ import (
 
 const (
 	LogFieldConnIndex = "connIndex"
+	LogFieldIPAddress = "ip"
 )
 
-var errNoAddressesLeft = fmt.Errorf("there are no free edge addresses left")
+var errNoAddressesLeft = ErrNoAddressesLeft{}
+
+type ErrNoAddressesLeft struct{}
+
+func (e ErrNoAddressesLeft) Error() string {
+	return "there are no free edge addresses left to resolve to"
+}
 
 // Edge finds addresses on the Cloudflare edge and hands them out to connections.
 type Edge struct {
@@ -28,8 +34,8 @@ type Edge struct {
 
 // ResolveEdge runs the initial discovery of the Cloudflare edge, finding Addrs that can be allocated
 // to connections.
-func ResolveEdge(log *zerolog.Logger, region string) (*Edge, error) {
-	regions, err := allregions.ResolveEdge(log, region)
+func ResolveEdge(log *zerolog.Logger, region string, edgeIpVersion allregions.ConfigIPVersion) (*Edge, error) {
+	regions, err := allregions.ResolveEdge(log, region, edgeIpVersion)
 	if err != nil {
 		return new(Edge), err
 	}
@@ -51,15 +57,6 @@ func StaticEdge(log *zerolog.Logger, hostnames []string) (*Edge, error) {
 	}, nil
 }
 
-// MockEdge creates a Cloudflare Edge from arbitrary TCP addresses. Used for testing.
-func MockEdge(log *zerolog.Logger, addrs []*allregions.EdgeAddr) *Edge {
-	regions := allregions.NewNoResolve(addrs)
-	return &Edge{
-		log:     log,
-		regions: regions,
-	}
-}
-
 // ------------------------------------
 // Methods
 // ------------------------------------
@@ -93,12 +90,15 @@ func (ed *Edge) GetAddr(connIndex int) (*allregions.EdgeAddr, error) {
 		log.Debug().Msg("edgediscovery - GetAddr: No addresses left to give proxy connection")
 		return nil, errNoAddressesLeft
 	}
-	log.Debug().Msg("edgediscovery - GetAddr: Giving connection its new address")
+	log = ed.log.With().
+		Int(LogFieldConnIndex, connIndex).
+		IPAddr(LogFieldIPAddress, addr.UDP.IP).Logger()
+	log.Debug().Msgf("edgediscovery - GetAddr: Giving connection its new address")
 	return addr, nil
 }
 
 // GetDifferentAddr gives back the proxy connection's edge Addr and uses a new one.
-func (ed *Edge) GetDifferentAddr(connIndex int) (*allregions.EdgeAddr, error) {
+func (ed *Edge) GetDifferentAddr(connIndex int, hasConnectivityError bool) (*allregions.EdgeAddr, error) {
 	log := ed.log.With().Int(LogFieldConnIndex, connIndex).Logger()
 
 	ed.Lock()
@@ -106,7 +106,7 @@ func (ed *Edge) GetDifferentAddr(connIndex int) (*allregions.EdgeAddr, error) {
 
 	oldAddr := ed.regions.AddrUsedBy(connIndex)
 	if oldAddr != nil {
-		ed.regions.GiveBack(oldAddr)
+		ed.regions.GiveBack(oldAddr, hasConnectivityError)
 	}
 	addr := ed.regions.GetUnusedAddr(oldAddr, connIndex)
 	if addr == nil {
@@ -114,8 +114,10 @@ func (ed *Edge) GetDifferentAddr(connIndex int) (*allregions.EdgeAddr, error) {
 		// note: if oldAddr were not nil, it will become available on the next iteration
 		return nil, errNoAddressesLeft
 	}
-	log.Debug().Msgf("edgediscovery - GetDifferentAddr: Giving connection its new address: %v from the address list: %v",
-		addr, ed.regions.AvailableAddrs())
+	log = ed.log.With().
+		Int(LogFieldConnIndex, connIndex).
+		IPAddr(LogFieldIPAddress, addr.UDP.IP).Logger()
+	log.Debug().Msgf("edgediscovery - GetDifferentAddr: Giving connection its new address from the address list: %v", ed.regions.AvailableAddrs())
 	return addr, nil
 }
 
@@ -128,9 +130,11 @@ func (ed *Edge) AvailableAddrs() int {
 
 // GiveBack the address so that other connections can use it.
 // Returns true if the address is in this edge.
-func (ed *Edge) GiveBack(addr *allregions.EdgeAddr) bool {
+func (ed *Edge) GiveBack(addr *allregions.EdgeAddr, hasConnectivityError bool) bool {
 	ed.Lock()
 	defer ed.Unlock()
-	ed.log.Debug().Msg("edgediscovery - GiveBack: Address now unused")
-	return ed.regions.GiveBack(addr)
+	log := ed.log.With().
+		IPAddr(LogFieldIPAddress, addr.UDP.IP).Logger()
+	log.Debug().Msgf("edgediscovery - GiveBack: Address now unused")
+	return ed.regions.GiveBack(addr, hasConnectivityError)
 }

edgediscovery/edgediscovery_test.go

@@ -11,56 +11,113 @@ import (
 )
 
 var (
-	addr0 = allregions.EdgeAddr{
+	testLogger = zerolog.Nop()
+	v4Addrs    = []*allregions.EdgeAddr{&addr0, &addr1, &addr2, &addr3}
+	v6Addrs    = []*allregions.EdgeAddr{&addr4, &addr5, &addr6, &addr7}
+	addr0      = allregions.EdgeAddr{
 		TCP: &net.TCPAddr{
-			IP:   net.ParseIP("123.0.0.0"),
+			IP:   net.ParseIP("123.4.5.0"),
 			Port: 8000,
 			Zone: "",
 		},
 		UDP: &net.UDPAddr{
-			IP:   net.ParseIP("123.0.0.0"),
+			IP:   net.ParseIP("123.4.5.0"),
 			Port: 8000,
 			Zone: "",
 		},
+		IPVersion: allregions.V4,
 	}
 	addr1 = allregions.EdgeAddr{
 		TCP: &net.TCPAddr{
-			IP:   net.ParseIP("123.0.0.1"),
+			IP:   net.ParseIP("123.4.5.1"),
 			Port: 8000,
 			Zone: "",
 		},
 		UDP: &net.UDPAddr{
-			IP:   net.ParseIP("123.0.0.1"),
+			IP:   net.ParseIP("123.4.5.1"),
 			Port: 8000,
 			Zone: "",
 		},
+		IPVersion: allregions.V4,
 	}
 	addr2 = allregions.EdgeAddr{
 		TCP: &net.TCPAddr{
-			IP:   net.ParseIP("123.0.0.2"),
+			IP:   net.ParseIP("123.4.5.2"),
 			Port: 8000,
 			Zone: "",
 		},
 		UDP: &net.UDPAddr{
-			IP:   net.ParseIP("123.0.0.2"),
+			IP:   net.ParseIP("123.4.5.2"),
 			Port: 8000,
 			Zone: "",
 		},
+		IPVersion: allregions.V4,
 	}
 	addr3 = allregions.EdgeAddr{
 		TCP: &net.TCPAddr{
-			IP:   net.ParseIP("123.0.0.3"),
+			IP:   net.ParseIP("123.4.5.3"),
 			Port: 8000,
 			Zone: "",
 		},
 		UDP: &net.UDPAddr{
-			IP:   net.ParseIP("123.0.0.3"),
+			IP:   net.ParseIP("123.4.5.3"),
 			Port: 8000,
 			Zone: "",
 		},
+		IPVersion: allregions.V4,
+	}
+	addr4 = allregions.EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("2606:4700:a0::1"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("2606:4700:a0::1"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: allregions.V6,
+	}
+	addr5 = allregions.EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("2606:4700:a0::2"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("2606:4700:a0::2"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: allregions.V6,
+	}
+	addr6 = allregions.EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("2606:4700:a0::3"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("2606:4700:a0::3"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: allregions.V6,
+	}
+	addr7 = allregions.EdgeAddr{
+		TCP: &net.TCPAddr{
+			IP:   net.ParseIP("2606:4700:a0::4"),
+			Port: 8000,
+			Zone: "",
+		},
+		UDP: &net.UDPAddr{
+			IP:   net.ParseIP("2606:4700:a0::4"),
+			Port: 8000,
+			Zone: "",
+		},
+		IPVersion: allregions.V6,
 	}
-
-	testLogger = zerolog.Nop()
 )
 
 func TestGiveBack(t *testing.T) {
@@ -75,7 +132,7 @@ func TestGiveBack(t *testing.T) {
 	assert.Equal(t, 3, edge.AvailableAddrs())
 
 	// Get it back
-	edge.GiveBack(addr)
+	edge.GiveBack(addr, false)
 	assert.Equal(t, 4, edge.AvailableAddrs())
 }
 
@@ -107,7 +164,7 @@ func TestGetAddrForRPC(t *testing.T) {
 	assert.Equal(t, 4, edge.AvailableAddrs())
 
 	// Get it back
-	edge.GiveBack(addr)
+	edge.GiveBack(addr, false)
 	assert.Equal(t, 4, edge.AvailableAddrs())
 }
 
@@ -122,13 +179,13 @@ func TestOnePerRegion(t *testing.T) {
 	assert.NotNil(t, a1)
 
 	// if the first address is bad, get the second one
-	a2, err := edge.GetDifferentAddr(connID)
+	a2, err := edge.GetDifferentAddr(connID, false)
 	assert.NoError(t, err)
 	assert.NotNil(t, a2)
 	assert.NotEqual(t, a1, a2)
 
 	// now that second one is bad, get the first one again
-	a3, err := edge.GetDifferentAddr(connID)
+	a3, err := edge.GetDifferentAddr(connID, false)
 	assert.NoError(t, err)
 	assert.Equal(t, a1, a3)
 }
@@ -144,11 +201,11 @@ func TestOnlyOneAddrLeft(t *testing.T) {
 	assert.NotNil(t, addr)
 
 	// If that edge address is "bad", there's no alternative address.
-	_, err = edge.GetDifferentAddr(connID)
+	_, err = edge.GetDifferentAddr(connID, false)
 	assert.Error(t, err)
 
 	// previously bad address should become available again on next iteration.
-	addr, err = edge.GetDifferentAddr(connID)
+	addr, err = edge.GetDifferentAddr(connID, false)
 	assert.NoError(t, err)
 	assert.NotNil(t, addr)
 }
@@ -190,8 +247,17 @@ func TestGetDifferentAddr(t *testing.T) {
 	assert.Equal(t, 3, edge.AvailableAddrs())
 
 	// If the same connection requests another address, it should get the same one.
-	addr2, err := edge.GetDifferentAddr(connID)
+	addr2, err := edge.GetDifferentAddr(connID, false)
 	assert.NoError(t, err)
 	assert.NotEqual(t, addr, addr2)
 	assert.Equal(t, 3, edge.AvailableAddrs())
 }
+
+// MockEdge creates a Cloudflare Edge from arbitrary TCP addresses. Used for testing.
+func MockEdge(log *zerolog.Logger, addrs []*allregions.EdgeAddr) *Edge {
+	regions := allregions.NewNoResolve(addrs)
+	return &Edge{
+		log:     log,
+		regions: regions,
+	}
+}

ingress/config.go

@@ -36,6 +36,7 @@ const (
 	NoChunkedEncodingFlag         = "no-chunked-encoding"
 	ProxyAddressFlag              = "proxy-address"
 	ProxyPortFlag                 = "proxy-port"
+	Http2OriginFlag               = "http2-origin"
 )
 
 const (
@@ -128,6 +129,7 @@ func originRequestFromSingeRule(c *cli.Context) OriginRequestConfig {
 	var proxyAddress = defaultProxyAddress
 	var proxyPort uint
 	var proxyType string
+	var http2Origin bool
 	if flag := ProxyConnectTimeoutFlag; c.IsSet(flag) {
 		connectTimeout = config.CustomDuration{Duration: c.Duration(flag)}
 	}
@@ -171,9 +173,13 @@ func originRequestFromSingeRule(c *cli.Context) OriginRequestConfig {
 		// Note TUN-3758 , we use Int because UInt is not supported with altsrc
 		proxyPort = uint(c.Int(flag))
 	}
+	if flag := Http2OriginFlag; c.IsSet(flag) {
+		http2Origin = c.Bool(flag)
+	}
 	if c.IsSet(Socks5Flag) {
 		proxyType = socksProxy
 	}
+
 	return OriginRequestConfig{
 		ConnectTimeout:         connectTimeout,
 		TLSTimeout:             tlsTimeout,
@@ -190,6 +196,7 @@ func originRequestFromSingeRule(c *cli.Context) OriginRequestConfig {
 		ProxyAddress:           proxyAddress,
 		ProxyPort:              proxyPort,
 		ProxyType:              proxyType,
+		Http2Origin:            http2Origin,
 	}
 }
 
@@ -255,6 +262,9 @@ func originRequestFromConfig(c config.OriginRequestConfig) OriginRequestConfig {
 			}
 		}
 	}
+	if c.Http2Origin != nil {
+		out.Http2Origin = *c.Http2Origin
+	}
 	return out
 }
 
@@ -298,6 +308,8 @@ type OriginRequestConfig struct {
 	ProxyType string `yaml:"proxyType" json:"proxyType"`
 	// IP rules for the proxy service
 	IPRules []ipaccess.Rule `yaml:"ipRules" json:"ipRules"`
+	// Attempt to connect to origin with HTTP/2
+	Http2Origin bool `yaml:"http2Origin" json:"http2Origin"`
 }
 
 func (defaults *OriginRequestConfig) setConnectTimeout(overrides config.OriginRequestConfig) {
@@ -403,6 +415,12 @@ func (defaults *OriginRequestConfig) setIPRules(overrides config.OriginRequestCo
 	}
 }
 
+func (defaults *OriginRequestConfig) setHttp2Origin(overrides config.OriginRequestConfig) {
+	if val := overrides.Http2Origin; val != nil {
+		defaults.Http2Origin = *val
+	}
+}
+
 // SetConfig gets config for the requests that cloudflared sends to origins.
 // Each field has a setter method which sets a value for the field by trying to find:
 //   1. The user config for this rule
@@ -428,6 +446,7 @@ func setConfig(defaults OriginRequestConfig, overrides config.OriginRequestConfi
 	cfg.setProxyAddress(overrides)
 	cfg.setProxyType(overrides)
 	cfg.setIPRules(overrides)
+	cfg.setHttp2Origin(overrides)
 	return cfg
 }
 
@@ -475,6 +494,7 @@ func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig
 		ProxyPort:              zeroUIntToNil(c.ProxyPort),
 		ProxyType:              emptyStringToNil(c.ProxyType),
 		IPRules:                convertToRawIPRules(c.IPRules),
+		Http2Origin:            defaultBoolToNil(c.Http2Origin),
 	}
 }
 

ingress/origin_service.go

@@ -290,6 +290,7 @@ func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerol
 		TLSHandshakeTimeout:   cfg.TLSTimeout.Duration,
 		ExpectContinueTimeout: 1 * time.Second,
 		TLSClientConfig:       &tls.Config{RootCAs: originCertPool, InsecureSkipVerify: cfg.NoTLSVerify},
+		ForceAttemptHTTP2:     cfg.Http2Origin,
 	}
 	if _, isHelloWorld := service.(*helloWorld); !isHelloWorld && cfg.OriginServerName != "" {
 		httpTransport.TLSClientConfig.ServerName = cfg.OriginServerName

ingress/rule_test.go

@@ -182,25 +182,25 @@ func TestMarshalJSON(t *testing.T) {
 		{
 			name:     "Nil",
 			path:     nil,
-			expected: `{"hostname":"example.com","path":null,"service":"https://localhost:8000","originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null}}`,
+			expected: `{"hostname":"example.com","path":null,"service":"https://localhost:8000","originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false}}`,
 			want:     true,
 		},
 		{
 			name:     "Nil regex",
 			path:     &Regexp{Regexp: nil},
-			expected: `{"hostname":"example.com","path":null,"service":"https://localhost:8000","originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null}}`,
+			expected: `{"hostname":"example.com","path":null,"service":"https://localhost:8000","originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false}}`,
 			want:     true,
 		},
 		{
 			name:     "Empty",
 			path:     &Regexp{Regexp: regexp.MustCompile("")},
-			expected: `{"hostname":"example.com","path":"","service":"https://localhost:8000","originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null}}`,
+			expected: `{"hostname":"example.com","path":"","service":"https://localhost:8000","originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false}}`,
 			want:     true,
 		},
 		{
 			name:     "Basic",
 			path:     &Regexp{Regexp: regexp.MustCompile("/echo")},
-			expected: `{"hostname":"example.com","path":"/echo","service":"https://localhost:8000","originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null}}`,
+			expected: `{"hostname":"example.com","path":"/echo","service":"https://localhost:8000","originRequest":{"connectTimeout":30,"tlsTimeout":10,"tcpKeepAlive":30,"noHappyEyeballs":false,"keepAliveTimeout":90,"keepAliveConnections":100,"httpHostHeader":"","originServerName":"","caPool":"","noTLSVerify":false,"disableChunkedEncoding":false,"bastionMode":false,"proxyAddress":"127.0.0.1","proxyPort":0,"proxyType":"","ipRules":null,"http2Origin":false}}`,
 			want:     true,
 		},
 	}

proxy/proxy.go

@@ -196,6 +196,9 @@ func (p *Proxy) proxyHTTPRequest(
 	resp, err := httpService.RoundTrip(roundTripReq)
 	if err != nil {
 		tracing.EndWithErrorStatus(ttfbSpan, err)
+		if err := roundTripReq.Context().Err(); err != nil {
+			return errors.Wrap(err, "Incoming request ended abruptly")
+		}
 		return errors.Wrap(err, "Unable to reach the origin service. The service may be down or it may not be responding to traffic from cloudflared")
 	}
 

quic/datagram.go

@@ -39,7 +39,6 @@ func (dm *DatagramMuxer) SendTo(sessionID uuid.UUID, payload []byte) error {
 	if err := dm.session.SendMessage(msgWithID); err != nil {
 		return errors.Wrap(err, "Failed to send datagram back to edge")
 	}
-	dm.logger.Debug().Str("sessionID", sessionID.String()).Int("bytes", len(payload)).Msg("Send datagram back to edge")
 	return nil
 }
 
@@ -55,7 +54,6 @@ func (dm *DatagramMuxer) ReceiveFrom() (uuid.UUID, []byte, error) {
 	if err != nil {
 		return uuid.Nil, nil, err
 	}
-	dm.logger.Debug().Str("sessionID", sessionID.String()).Int("bytes", len(payload)).Msg("Received datagram from edge")
 	return sessionID, payload, nil
 }
 

quic/pogs.go

@@ -1,6 +1,8 @@
 package quic
 
 import (
+	"fmt"
+
 	capnp "zombiezen.com/go/capnproto2"
 	"zombiezen.com/go/capnproto2/pogs"
 
@@ -16,6 +18,18 @@ const (
 	ConnectionTypeTCP
 )
 
+func (c ConnectionType) String() string {
+	switch c {
+	case ConnectionTypeHTTP:
+		return "http"
+	case ConnectionTypeWebsocket:
+		return "ws"
+	case ConnectionTypeTCP:
+		return "tcp"
+	}
+	panic(fmt.Sprintf("invalid ConnectionType: %d", c))
+}
+
 // ConnectRequest is the representation of metadata sent at the start of a QUIC application handshake.
 type ConnectRequest struct {
 	Dest     string         `capnp:"dest"`

release_pkgs.py

@@ -8,23 +8,24 @@
        them to be in an uploadable state.
     2. Upload these packages to a storage in a format that apt and yum expect.
 """
-from subprocess import Popen, PIPE
-import os
 import argparse
 import base64
-from pathlib import Path
 import logging
+import os
 import shutil
 from hashlib import sha256
+from pathlib import Path
+from subprocess import Popen, PIPE
 
-import gnupg
 import boto3
+import gnupg
 from botocore.client import Config
 from botocore.exceptions import ClientError
 
 # The front facing R2 URL to access assets from.
 R2_ASSET_URL = 'https://demo-r2-worker.cloudflare-tunnel.workers.dev/'
 
+
 class PkgUploader:
     def __init__(self, account_id, bucket_name, client_id, client_secret):
         self.account_id = account_id
@@ -35,14 +36,14 @@ class PkgUploader:
     def upload_pkg_to_r2(self, filename, upload_file_path):
         endpoint_url = f"https://{self.account_id}.r2.cloudflarestorage.com"
         token_secret_hash = sha256(self.client_secret.encode()).hexdigest()
-         
+
         config = Config(
-            region_name = 'auto',
+            region_name='auto',
             s3={
                 "addressing_style": "path",
             }
         )
-         
+
         r2 = boto3.client(
             "s3",
             endpoint_url=endpoint_url,
@@ -57,7 +58,7 @@ class PkgUploader:
         except ClientError as e:
             raise e
 
-                  
+
 class PkgCreator:
     """
         The distribution conf is what dictates to reprepro, the debian packaging building
@@ -72,15 +73,16 @@ class PkgCreator:
         description - (String)
         gpg_key_id - gpg key id of what you want to use to sign the packages.(String) 
     """
-    def create_distribution_conf(self, 
-            file_path,
-            origin,
-            label,
-            releases,
-            archs,
-            components, 
-            description,
-            gpg_key_id ):
+
+    def create_distribution_conf(self,
+                                 file_path,
+                                 origin,
+                                 label,
+                                 releases,
+                                 archs,
+                                 components,
+                                 description,
+                                 gpg_key_id):
         with open(file_path, "w+") as distributions_file:
             for release in releases:
                 distributions_file.write(f"Origin: {origin}\n")
@@ -102,6 +104,7 @@ class PkgCreator:
         db and pool contain information and metadata about builds. We can ignore these.
         dist: contains all the pkgs and signed releases that are necessary for an apt download.
     """
+
     def create_deb_pkgs(self, release, deb_file):
         print(f"creating deb pkgs: {release} : {deb_file}")
         p = Popen(["reprepro", "includedeb", release, deb_file], stdout=PIPE, stderr=PIPE)
@@ -130,8 +133,9 @@ class PkgCreator:
         gpgcheck=1
         gpgkey=https://pkg.cloudflare.com/cloudflare-main.gpg
     """
+
     def create_repo_file(self, file_path, binary_name, baseurl, gpgkey_url):
-        with open(file_path, "w+") as repo_file:
+        with open(os.path.join(file_path, binary_name + '.repo'), "w+") as repo_file:
             repo_file.write(f"[{binary_name}-stable]")
             repo_file.write(f"{binary_name}-stable")
             repo_file.write(f"baseurl={baseurl}/rpm")
@@ -141,19 +145,19 @@ class PkgCreator:
             repo_file.write(f"gpgkey={gpgkey_url}")
 
     def _sign_rpms(self, file_path):
-        p = Popen(["rpm" , "--define", f"_gpg_name {gpg_key_name}", "--addsign", file_path], stdout=PIPE, stderr=PIPE)
+        p = Popen(["rpm", "--define", f"_gpg_name {gpg_key_name}", "--addsign", file_path], stdout=PIPE, stderr=PIPE)
         out, err = p.communicate()
         if p.returncode != 0:
             print(f"rpm sign result result => {out}, {err}")
             raise
 
     def _sign_repomd(self):
-        p = Popen(["gpg", "--batch", "--detach-sign", "--armor", "./rpm/repodata/repomd.xml"], stdout=PIPE, stderr=PIPE) 
+        p = Popen(["gpg", "--batch", "--detach-sign", "--armor", "./rpm/repodata/repomd.xml"], stdout=PIPE, stderr=PIPE)
         out, err = p.communicate()
         if p.returncode != 0:
             print(f"sign repomd result => {out}, {err}")
             raise
-        
+
     """
         sets up and signs the RPM directories in the following format:
         - rpm 
@@ -163,9 +167,10 @@ class PkgCreator:
 
         this assumes the assets are in the format <prefix>-<aarch64/x86_64/386>.rpm
     """
+
     def _setup_rpm_pkg_directories(self, artifacts_path, gpg_key_name, archs=["aarch64", "x86_64", "386"]):
         for arch in archs:
-            for root, _ , files in os.walk(artifacts_path):
+            for root, _, files in os.walk(artifacts_path):
                 for file in files:
                     if file.endswith(f"{arch}.rpm"):
                         new_dir = f"./rpm/{arch}"
@@ -174,11 +179,12 @@ class PkgCreator:
                         new_path = os.path.join(new_dir, file)
                         shutil.copyfile(old_path, new_path)
                         self._sign_rpms(new_path)
-    
+
     """
         imports gpg keys into the system so reprepro and createrepo can use it to sign packages.
         it returns the GPG ID after a successful import
     """
+
     def import_gpg_keys(self, private_key, public_key):
         gpg = gnupg.GPG()
         private_key = base64.b64decode(private_key)
@@ -192,12 +198,13 @@ class PkgCreator:
         basically rpm --import <key_file>
         This enables us to sign rpms.
     """
+
     def import_rpm_key(self, public_key):
         file_name = "pb.key"
         with open(file_name, "wb") as f:
             public_key = base64.b64decode(public_key)
             f.write(public_key)
-        
+
         p = Popen(["rpm", "--import", file_name], stdout=PIPE, stderr=PIPE)
         out, err = p.communicate()
         if p.returncode != 0:
@@ -212,18 +219,21 @@ class PkgCreator:
               and the release will be uploaded to the default path. 
     binary: name of the binary to upload
 """
+
+
 def upload_from_directories(pkg_uploader, directory, release, binary):
-     for root, _ , files in os.walk(directory):
+    for root, _, files in os.walk(directory):
         for file in files:
             upload_file_name = os.path.join(binary, root, file)
             if release:
                 upload_file_name = os.path.join(release, upload_file_name)
-            filename = os.path.join(root,file)
-            try: 
+            filename = os.path.join(root, file)
+            try:
                 pkg_uploader.upload_pkg_to_r2(filename, upload_file_name)
             except ClientError as e:
                 logging.error(e)
-                return 
+                return
+
 
 """ 
     1. looks into a built_artifacts folder for cloudflared debs
@@ -237,19 +247,22 @@ def upload_from_directories(pkg_uploader, directory, release, binary):
 
     release_version: is the cloudflared release version. Only provide this if you want a permanent backup.
 """
-def create_deb_packaging(pkg_creator, pkg_uploader, releases, gpg_key_id, binary_name, archs, package_component, release_version):
+
+
+def create_deb_packaging(pkg_creator, pkg_uploader, releases, gpg_key_id, binary_name, archs, package_component,
+                         release_version):
     # set configuration for package creation.
     print(f"initialising configuration for {binary_name} , {archs}")
     Path("./conf").mkdir(parents=True, exist_ok=True)
     pkg_creator.create_distribution_conf(
-    "./conf/distributions",
-    binary_name,
-    binary_name,
-    releases,
-    archs,
-    package_component,
-    f"apt repository for {binary_name}",
-    gpg_key_id)
+        "./conf/distributions",
+        binary_name,
+        binary_name,
+        releases,
+        archs,
+        package_component,
+        f"apt repository for {binary_name}",
+        gpg_key_id)
 
     # create deb pkgs
     for release in releases:
@@ -266,6 +279,7 @@ def create_deb_packaging(pkg_creator, pkg_uploader, releases, gpg_key_id, binary
         upload_from_directories(pkg_uploader, "dists", release_version, binary_name)
         upload_from_directories(pkg_uploader, "pool", release_version, binary_name)
 
+
 def create_rpm_packaging(
         pkg_creator,
         pkg_uploader,
@@ -294,60 +308,61 @@ def parse_args():
     )
 
     parser.add_argument(
-            "--bucket", default=os.environ.get("R2_BUCKET"), help="R2 Bucket name"
+        "--bucket", default=os.environ.get("R2_BUCKET"), help="R2 Bucket name"
     )
     parser.add_argument(
-            "--id", default=os.environ.get("R2_CLIENT_ID"), help="R2 Client ID"
+        "--id", default=os.environ.get("R2_CLIENT_ID"), help="R2 Client ID"
     )
     parser.add_argument(
-            "--secret", default=os.environ.get("R2_CLIENT_SECRET"), help="R2 Client Secret"
+        "--secret", default=os.environ.get("R2_CLIENT_SECRET"), help="R2 Client Secret"
     )
     parser.add_argument(
-            "--account", default=os.environ.get("R2_ACCOUNT_ID"), help="R2 Account Tag"
+        "--account", default=os.environ.get("R2_ACCOUNT_ID"), help="R2 Account Tag"
     )
     parser.add_argument(
-            "--release-tag", default=os.environ.get("RELEASE_VERSION"), help="Release version you want your pkgs to be\
+        "--release-tag", default=os.environ.get("RELEASE_VERSION"), help="Release version you want your pkgs to be\
             prefixed with. Leave empty if you don't want tagged release versions backed up to R2."
     )
 
     parser.add_argument(
-           "--binary", default=os.environ.get("BINARY_NAME"), help="The name of the binary the packages are for"
+        "--binary", default=os.environ.get("BINARY_NAME"), help="The name of the binary the packages are for"
     )
 
     parser.add_argument(
-            "--gpg-private-key", default=os.environ.get("LINUX_SIGNING_PRIVATE_KEY"), help="GPG private key to sign the\
+        "--gpg-private-key", default=os.environ.get("LINUX_SIGNING_PRIVATE_KEY"), help="GPG private key to sign the\
             packages"
     )
 
     parser.add_argument(
-            "--gpg-public-key", default=os.environ.get("LINUX_SIGNING_PUBLIC_KEY"), help="GPG public key used for\
+        "--gpg-public-key", default=os.environ.get("LINUX_SIGNING_PUBLIC_KEY"), help="GPG public key used for\
             signing packages"
     )
 
     parser.add_argument(
-            "--gpg-public-key-url", default=os.environ.get("GPG_PUBLIC_KEY_URL"), help="GPG public key url that\
+        "--gpg-public-key-url", default=os.environ.get("GPG_PUBLIC_KEY_URL"), help="GPG public key url that\
             downloaders can use to verify signing"
     )
 
     parser.add_argument(
-            "--pkg-upload-url", default=os.environ.get("PKG_URL"), help="URL to be used by downloaders"
+        "--pkg-upload-url", default=os.environ.get("PKG_URL"), help="URL to be used by downloaders"
     )
 
     parser.add_argument(
-            "--deb-based-releases", default=["bookworm", "bullseye", "buster", "jammy", "impish", "focal", "bionic",
-            "xenial", "trusty"],
-            help="list of debian based releases that need to be packaged for"
+        "--deb-based-releases", default=["bookworm", "bullseye", "buster", "jammy", "impish", "focal", "bionic",
+                                         "xenial", "trusty"],
+        help="list of debian based releases that need to be packaged for"
     )
 
     parser.add_argument(
-            "--archs", default=["amd64", "386", "arm64", "arm"], help="list of architectures we want to package for. Note that\
+        "--archs", default=["amd64", "386", "arm64", "arm", "armhf"], help="list of architectures we want to package for. Note that\
             it is the caller's responsiblity to ensure that these debs are already present in a directory. This script\
             will not build binaries or create their debs."
-            )
+    )
     args = parser.parse_args()
 
     return args
 
+
 if __name__ == "__main__":
     try:
         args = parse_args()
@@ -362,15 +377,15 @@ if __name__ == "__main__":
     pkg_uploader = PkgUploader(args.account, args.bucket, args.id, args.secret)
     print(f"signing with gpg_key: {gpg_key_id}")
     create_deb_packaging(pkg_creator, pkg_uploader, args.deb_based_releases, gpg_key_id, args.binary, args.archs,
-            "main", args.release_tag)
-    
+                         "main", args.release_tag)
+
     create_rpm_packaging(
-            pkg_creator,
-            pkg_uploader,
-            "./built_artifacts",
-            args.release_tag,
-            args.binary,
-            gpg_key_name,
-            args.gpg_public_key_url,
-            args.pkg_upload_url,
-            )
+        pkg_creator,
+        pkg_uploader,
+        "./built_artifacts",
+        args.release_tag,
+        args.binary,
+        gpg_key_name,
+        args.gpg_public_key_url,
+        args.pkg_upload_url,
+    )

supervisor/supervisor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -38,12 +39,14 @@ const (
 // Supervisor manages non-declarative tunnels. Establishes TCP connections with the edge, and
 // reconnects them if they disconnect.
 type Supervisor struct {
-	cloudflaredUUID   uuid.UUID
-	config            *TunnelConfig
-	orchestrator      *orchestration.Orchestrator
-	edgeIPs           *edgediscovery.Edge
-	tunnelErrors      chan tunnelError
-	tunnelsConnecting map[int]chan struct{}
+	cloudflaredUUID         uuid.UUID
+	config                  *TunnelConfig
+	orchestrator            *orchestration.Orchestrator
+	edgeIPs                 *edgediscovery.Edge
+	edgeTunnelServer        EdgeTunnelServer
+	tunnelErrors            chan tunnelError
+	tunnelsConnecting       map[int]chan struct{}
+	tunnelsProtocolFallback map[int]*protocolFallback
 	// nextConnectedIndex and nextConnectedSignal are used to wait for all
 	// currently-connecting tunnels to finish connecting so we can reset backoff timer
 	nextConnectedIndex  int
@@ -72,16 +75,42 @@ func NewSupervisor(config *TunnelConfig, orchestrator *orchestration.Orchestrato
 		return nil, fmt.Errorf("failed to generate cloudflared instance ID: %w", err)
 	}
 
+	isStaticEdge := len(config.EdgeAddrs) > 0
+
 	var edgeIPs *edgediscovery.Edge
-	if len(config.EdgeAddrs) > 0 {
+	if isStaticEdge { // static edge addresses
 		edgeIPs, err = edgediscovery.StaticEdge(config.Log, config.EdgeAddrs)
 	} else {
-		edgeIPs, err = edgediscovery.ResolveEdge(config.Log, config.Region)
+		edgeIPs, err = edgediscovery.ResolveEdge(config.Log, config.Region, config.EdgeIPVersion)
 	}
 	if err != nil {
 		return nil, err
 	}
 
+	reconnectCredentialManager := newReconnectCredentialManager(connection.MetricsNamespace, connection.TunnelSubsystem, config.HAConnections)
+	log := NewConnAwareLogger(config.Log, config.Observer)
+
+	var edgeAddrHandler EdgeAddrHandler
+	if isStaticEdge { // static edge addresses
+		edgeAddrHandler = &IPAddrFallback{}
+	} else if config.EdgeIPVersion == allregions.IPv6Only || config.EdgeIPVersion == allregions.Auto {
+		edgeAddrHandler = &IPAddrFallback{}
+	} else { // IPv4Only
+		edgeAddrHandler = &DefaultAddrFallback{}
+	}
+
+	edgeTunnelServer := EdgeTunnelServer{
+		config:            config,
+		cloudflaredUUID:   cloudflaredUUID,
+		orchestrator:      orchestrator,
+		credentialManager: reconnectCredentialManager,
+		edgeAddrs:         edgeIPs,
+		edgeAddrHandler:   edgeAddrHandler,
+		reconnectCh:       reconnectCh,
+		gracefulShutdownC: gracefulShutdownC,
+		connAwareLogger:   log,
+	}
+
 	useReconnectToken := false
 	if config.ClassicTunnel != nil {
 		useReconnectToken = config.ClassicTunnel.UseReconnectToken
@@ -92,11 +121,13 @@ func NewSupervisor(config *TunnelConfig, orchestrator *orchestration.Orchestrato
 		config:                     config,
 		orchestrator:               orchestrator,
 		edgeIPs:                    edgeIPs,
+		edgeTunnelServer:           edgeTunnelServer,
 		tunnelErrors:               make(chan tunnelError),
 		tunnelsConnecting:          map[int]chan struct{}{},
-		log:                        NewConnAwareLogger(config.Log, config.Observer),
+		tunnelsProtocolFallback:    map[int]*protocolFallback{},
+		log:                        log,
 		logTransport:               config.LogTransport,
-		reconnectCredentialManager: newReconnectCredentialManager(connection.MetricsNamespace, connection.TunnelSubsystem, config.HAConnections),
+		reconnectCredentialManager: reconnectCredentialManager,
 		useReconnectToken:          useReconnectToken,
 		reconnectCh:                reconnectCh,
 		gracefulShutdownC:          gracefulShutdownC,
@@ -143,11 +174,22 @@ func (s *Supervisor) Run(
 				tunnelsActive--
 			}
 			return nil
-		// startTunnel returned with error
+		// startTunnel completed with a response
 		// (note that this may also be caused by context cancellation)
 		case tunnelError := <-s.tunnelErrors:
 			tunnelsActive--
 			if tunnelError.err != nil && !shuttingDown {
+				switch tunnelError.err.(type) {
+				case ReconnectSignal:
+					// For tunnels that closed with reconnect signal, we reconnect immediately
+					go s.startTunnel(ctx, tunnelError.index, s.newConnectedTunnelSignal(tunnelError.index))
+					tunnelsActive++
+					continue
+				}
+				// Make sure we don't continue if there is no more fallback allowed
+				if _, retry := s.tunnelsProtocolFallback[tunnelError.index].GetMaxBackoffDuration(ctx); !retry {
+					continue
+				}
 				s.log.ConnAwareLogger().Err(tunnelError.err).Int(connection.LogFieldConnIndex, tunnelError.index).Msg("Connection terminated")
 				tunnelsWaiting = append(tunnelsWaiting, tunnelError.index)
 				s.waitForNextTunnel(tunnelError.index)
@@ -155,10 +197,9 @@ func (s *Supervisor) Run(
 				if backoffTimer == nil {
 					backoffTimer = backoff.BackoffTimer()
 				}
-
-				// Previously we'd mark the edge address as bad here, but now we'll just silently use another.
 			} else if tunnelsActive == 0 {
-				// all connected tunnels exited gracefully, no more work to do
+				s.log.ConnAwareLogger().Msg("no more connections active and exiting")
+				// All connected tunnels exited gracefully, no more work to do
 				return nil
 			}
 		// Backoff was set and its timer expired
@@ -192,6 +233,8 @@ func (s *Supervisor) Run(
 }
 
 // Returns nil if initialization succeeded, else the initialization error.
+// Attempts here will be made to connect one tunnel, if successful, it will
+// connect the available tunnels up to config.HAConnections.
 func (s *Supervisor) initialize(
 	ctx context.Context,
 	connectedSignal *signal.Signal,
@@ -201,8 +244,15 @@ func (s *Supervisor) initialize(
 		s.log.Logger().Info().Msgf("You requested %d HA connections but I can give you at most %d.", s.config.HAConnections, availableAddrs)
 		s.config.HAConnections = availableAddrs
 	}
+	s.tunnelsProtocolFallback[0] = &protocolFallback{
+		retry.BackoffHandler{MaxRetries: s.config.Retries},
+		s.config.ProtocolSelector.Current(),
+		false,
+	}
 
 	go s.startFirstTunnel(ctx, connectedSignal)
+
+	// Wait for response from first tunnel before proceeding to attempt other HA edge tunnels
 	select {
 	case <-ctx.Done():
 		<-s.tunnelErrors
@@ -213,8 +263,14 @@ func (s *Supervisor) initialize(
 		return errEarlyShutdown
 	case <-connectedSignal.Wait():
 	}
+
 	// At least one successful connection, so start the rest
 	for i := 1; i < s.config.HAConnections; i++ {
+		s.tunnelsProtocolFallback[i] = &protocolFallback{
+			retry.BackoffHandler{MaxRetries: s.config.Retries},
+			s.config.ProtocolSelector.Current(),
+			false,
+		}
 		ch := signal.New(make(chan struct{}))
 		go s.startTunnel(ctx, i, ch)
 		time.Sleep(registrationInterval)
@@ -229,102 +285,65 @@ func (s *Supervisor) startFirstTunnel(
 	connectedSignal *signal.Signal,
 ) {
 	var (
-		addr *allregions.EdgeAddr
-		err  error
+		err error
 	)
 	const firstConnIndex = 0
+	isStaticEdge := len(s.config.EdgeAddrs) > 0
 	defer func() {
 		s.tunnelErrors <- tunnelError{index: firstConnIndex, err: err}
 	}()
 
-	addr, err = s.edgeIPs.GetAddr(firstConnIndex)
-	if err != nil {
-		return
-	}
-
-	err = ServeTunnelLoop(
-		ctx,
-		s.reconnectCredentialManager,
-		s.config,
-		s.orchestrator,
-		addr,
-		s.log,
-		firstConnIndex,
-		connectedSignal,
-		s.cloudflaredUUID,
-		s.reconnectCh,
-		s.gracefulShutdownC,
-	)
 	// If the first tunnel disconnects, keep restarting it.
-	edgeErrors := 0
-	for s.unusedIPs() {
+	for {
+		err = s.edgeTunnelServer.Serve(ctx, firstConnIndex, s.tunnelsProtocolFallback[firstConnIndex], connectedSignal)
 		if ctx.Err() != nil {
 			return
 		}
-		switch err.(type) {
-		case nil:
-			return
-		// try the next address if it was a quic.IdleTimeoutError, dialError(network problem) or
-		// dupConnRegisterTunnelError
-		case *quic.IdleTimeoutError, edgediscovery.DialError, connection.DupConnRegisterTunnelError:
-			edgeErrors++
-		default:
+		if err == nil {
 			return
 		}
-		if edgeErrors >= 2 {
-			addr, err = s.edgeIPs.GetDifferentAddr(firstConnIndex)
-			if err != nil {
+		// Make sure we don't continue if there is no more fallback allowed
+		if _, retry := s.tunnelsProtocolFallback[firstConnIndex].GetMaxBackoffDuration(ctx); !retry {
+			return
+		}
+		// Try again for Unauthorized errors because we hope them to be
+		// transient due to edge propagation lag on new Tunnels.
+		if strings.Contains(err.Error(), "Unauthorized") {
+			continue
+		}
+		switch err.(type) {
+		case edgediscovery.ErrNoAddressesLeft:
+			// If your provided addresses are not available, we will keep trying regardless.
+			if !isStaticEdge {
 				return
 			}
+		case connection.DupConnRegisterTunnelError,
+			*quic.IdleTimeoutError,
+			edgediscovery.DialError,
+			*connection.EdgeQuicDialError:
+			// Try again for these types of errors
+		default:
+			// Uncaught errors should bail startup
+			return
 		}
-		err = ServeTunnelLoop(
-			ctx,
-			s.reconnectCredentialManager,
-			s.config,
-			s.orchestrator,
-			addr,
-			s.log,
-			firstConnIndex,
-			connectedSignal,
-			s.cloudflaredUUID,
-			s.reconnectCh,
-			s.gracefulShutdownC,
-		)
 	}
 }
 
 // startTunnel starts a new tunnel connection. The resulting error will be sent on
-// s.tunnelErrors.
+// s.tunnelError as this is expected to run in a goroutine.
 func (s *Supervisor) startTunnel(
 	ctx context.Context,
 	index int,
 	connectedSignal *signal.Signal,
 ) {
 	var (
-		addr *allregions.EdgeAddr
-		err  error
+		err error
 	)
 	defer func() {
 		s.tunnelErrors <- tunnelError{index: index, err: err}
 	}()
 
-	addr, err = s.edgeIPs.GetDifferentAddr(index)
-	if err != nil {
-		return
-	}
-	err = ServeTunnelLoop(
-		ctx,
-		s.reconnectCredentialManager,
-		s.config,
-		s.orchestrator,
-		addr,
-		s.log,
-		uint8(index),
-		connectedSignal,
-		s.cloudflaredUUID,
-		s.reconnectCh,
-		s.gracefulShutdownC,
-	)
+	err = s.edgeTunnelServer.Serve(ctx, uint8(index), s.tunnelsProtocolFallback[index], connectedSignal)
 }
 
 func (s *Supervisor) newConnectedTunnelSignal(index int) *signal.Signal {

supervisor/tunnel.go

@@ -122,30 +122,82 @@ func StartTunnelDaemon(
 	return s.Run(ctx, connectedSignal)
 }
 
-func ServeTunnelLoop(
-	ctx context.Context,
-	credentialManager *reconnectCredentialManager,
-	config *TunnelConfig,
-	orchestrator *orchestration.Orchestrator,
-	addr *allregions.EdgeAddr,
-	connAwareLogger *ConnAwareLogger,
-	connIndex uint8,
-	connectedSignal *signal.Signal,
-	cloudflaredUUID uuid.UUID,
-	reconnectCh chan ReconnectSignal,
-	gracefulShutdownC <-chan struct{},
-) error {
+// EdgeAddrHandler provides a mechanism switch between behaviors in ServeTunnel
+// for handling the errors when attempting to make edge connections.
+type EdgeAddrHandler interface {
+	// ShouldGetNewAddress will check the edge connection error and determine if
+	// the edge address should be replaced with a new one. Also, will return if the
+	// error should be recognized as a connectivity error, or otherwise, a general
+	// application error.
+	ShouldGetNewAddress(err error) (needsNewAddress bool, isConnectivityError bool)
+}
+
+// DefaultAddrFallback will always return false for isConnectivityError since this
+// handler is a way to provide the legacy behavior in the new edge discovery algorithm.
+type DefaultAddrFallback struct {
+	edgeErrors int
+}
+
+func (f DefaultAddrFallback) ShouldGetNewAddress(err error) (needsNewAddress bool, isConnectivityError bool) {
+	switch err.(type) {
+	case nil: // maintain current IP address
+	// DupConnRegisterTunnelError should indicate to get a new address immediately
+	case connection.DupConnRegisterTunnelError:
+		return true, false
+	// Try the next address if it was a quic.IdleTimeoutError
+	case *quic.IdleTimeoutError,
+		edgediscovery.DialError,
+		*connection.EdgeQuicDialError:
+		// Wait for two failures before falling back to a new address
+		f.edgeErrors++
+		if f.edgeErrors >= 2 {
+			f.edgeErrors = 0
+			return true, false
+		}
+	default: // maintain current IP address
+	}
+	return false, false
+}
+
+// IPAddrFallback will have more conditions to fall back to a new address for certain
+// edge connection errors. This means that this handler will return true for isConnectivityError
+// for more cases like duplicate connection register and edge quic dial errors.
+type IPAddrFallback struct{}
+
+func (f IPAddrFallback) ShouldGetNewAddress(err error) (needsNewAddress bool, isConnectivityError bool) {
+	switch err.(type) {
+	case nil: // maintain current IP address
+	// Try the next address if it was a quic.IdleTimeoutError
+	// DupConnRegisterTunnelError needs to also receive a new ip address
+	case connection.DupConnRegisterTunnelError,
+		*quic.IdleTimeoutError:
+		return true, false
+	// Network problems should be retried with new address immediately and report
+	// as connectivity error
+	case edgediscovery.DialError, *connection.EdgeQuicDialError:
+		return true, true
+	default: // maintain current IP address
+	}
+	return false, false
+}
+
+type EdgeTunnelServer struct {
+	config            *TunnelConfig
+	cloudflaredUUID   uuid.UUID
+	orchestrator      *orchestration.Orchestrator
+	credentialManager *reconnectCredentialManager
+	edgeAddrHandler   EdgeAddrHandler
+	edgeAddrs         *edgediscovery.Edge
+	reconnectCh       chan ReconnectSignal
+	gracefulShutdownC <-chan struct{}
+
+	connAwareLogger *ConnAwareLogger
+}
+
+func (e EdgeTunnelServer) Serve(ctx context.Context, connIndex uint8, protocolFallback *protocolFallback, connectedSignal *signal.Signal) error {
 	haConnections.Inc()
 	defer haConnections.Dec()
 
-	logger := config.Log.With().Uint8(connection.LogFieldConnIndex, connIndex).Logger()
-	connLog := connAwareLogger.ReplaceLogger(&logger)
-
-	protocolFallback := &protocolFallback{
-		retry.BackoffHandler{MaxRetries: config.Retries},
-		config.ProtocolSelector.Current(),
-		false,
-	}
 	connectedFuse := h2mux.NewBooleanFuse()
 	go func() {
 		if connectedFuse.Await() {
@@ -154,54 +206,83 @@ func ServeTunnelLoop(
 	}()
 	// Ensure the above goroutine will terminate if we return without connecting
 	defer connectedFuse.Fuse(false)
+
+	// Fetch IP address to associated connection index
+	addr, err := e.edgeAddrs.GetAddr(int(connIndex))
+	switch err.(type) {
+	case nil: // no error
+	case edgediscovery.ErrNoAddressesLeft:
+		return err
+	default:
+		return err
+	}
+
+	logger := e.config.Log.With().
+		IPAddr(connection.LogFieldIPAddress, addr.UDP.IP).
+		Uint8(connection.LogFieldConnIndex, connIndex).
+		Logger()
+	connLog := e.connAwareLogger.ReplaceLogger(&logger)
 	// Each connection to keep its own copy of protocol, because individual connections might fallback
 	// to another protocol when a particular metal doesn't support new protocol
-	for {
-		err, recoverable := ServeTunnel(
-			ctx,
-			connLog,
-			credentialManager,
-			config,
-			orchestrator,
-			addr,
-			connIndex,
-			connectedFuse,
-			protocolFallback,
-			cloudflaredUUID,
-			reconnectCh,
-			protocolFallback.protocol,
-			gracefulShutdownC,
-		)
+	// Each connection can also have it's own IP version because individual connections might fallback
+	// to another IP version.
+	err, recoverable := ServeTunnel(
+		ctx,
+		connLog,
+		e.credentialManager,
+		e.config,
+		e.orchestrator,
+		addr,
+		connIndex,
+		connectedFuse,
+		protocolFallback,
+		e.cloudflaredUUID,
+		e.reconnectCh,
+		protocolFallback.protocol,
+		e.gracefulShutdownC,
+	)
 
-		if recoverable {
-			duration, ok := protocolFallback.GetMaxBackoffDuration(ctx)
-			if !ok {
-				return err
-			}
-			config.Observer.SendReconnect(connIndex)
-			connLog.Logger().Info().Msgf("Retrying connection in up to %s seconds", duration)
+	// If the connection is recoverable, we want to maintain the same IP
+	// but backoff a reconnect with some duration.
+	if recoverable {
+		duration, ok := protocolFallback.GetMaxBackoffDuration(ctx)
+		if !ok {
+			return err
 		}
+		e.config.Observer.SendReconnect(connIndex)
+		connLog.Logger().Info().Msgf("Retrying connection in up to %s seconds", duration)
+	}
 
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case <-gracefulShutdownC:
-			return nil
-		case <-protocolFallback.BackoffTimer():
-			if !recoverable {
-				return err
-			}
-
-			if !selectNextProtocol(
-				connLog.Logger(),
-				protocolFallback,
-				config.ProtocolSelector,
-				err,
-			) {
-				return err
-			}
+	// Check if the connection error was from an IP issue with the host or
+	// establishing a connection to the edge and if so, rotate the IP address.
+	yes, hasConnectivityError := e.edgeAddrHandler.ShouldGetNewAddress(err)
+	if yes {
+		if _, err := e.edgeAddrs.GetDifferentAddr(int(connIndex), hasConnectivityError); err != nil {
+			return err
 		}
 	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-e.gracefulShutdownC:
+		return nil
+	case <-protocolFallback.BackoffTimer():
+		if !recoverable {
+			return err
+		}
+
+		if !selectNextProtocol(
+			connLog.Logger(),
+			protocolFallback,
+			e.config.ProtocolSelector,
+			err,
+		) {
+			return err
+		}
+	}
+
+	return err
 }
 
 // protocolFallback is a wrapper around backoffHandler that will try fallback option when backoff reaches
@@ -233,6 +314,10 @@ func selectNextProtocol(
 ) bool {
 	var idleTimeoutError *quic.IdleTimeoutError
 	isNetworkActivityTimeout := errors.As(cause, &idleTimeoutError)
+	edgeQuicDialError, ok := cause.(*connection.EdgeQuicDialError)
+	if !isNetworkActivityTimeout && ok {
+		isNetworkActivityTimeout = errors.As(edgeQuicDialError.Cause, &idleTimeoutError)
+	}
 	_, hasFallback := selector.Fallback()
 
 	if protocolBackoff.ReachedMaxRetries() || (hasFallback && isNetworkActivityTimeout) {
@@ -241,7 +326,7 @@ func selectNextProtocol(
 				"Cloudflare Network with `quic` protocol, then most likely your machine/network is getting its egress " +
 				"UDP to port 7844 (or others) blocked or dropped. Make sure to allow egress connectivity as per " +
 				"https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/configuration/ports-and-ips/\n" +
-				"If you are using private routing to this Tunnel, then UDP (and Private DNS Resolution) will not work" +
+				"If you are using private routing to this Tunnel, then UDP (and Private DNS Resolution) will not work " +
 				"unless your cloudflared can connect with Cloudflare Network with `quic`.")
 		}
 
@@ -326,8 +411,12 @@ func ServeTunnel(
 				connLog.ConnAwareLogger().Msg(activeIncidentsMsg(incidents))
 			}
 			return err.Cause, !err.Permanent
+		case *connection.EdgeQuicDialError:
+			// Don't retry connection for a dial error
+			return err, false
 		case ReconnectSignal:
 			connLog.Logger().Info().
+				IPAddr(connection.LogFieldIPAddress, addr.UDP.IP).
 				Uint8(connection.LogFieldConnIndex, connIndex).
 				Msgf("Restarting connection due to reconnect signal in %s", err.Delay)
 			err.DelayBeforeReconnect()
@@ -369,6 +458,7 @@ func serveTunnel(
 		connectedFuse,
 		config.NamedTunnel,
 		connIndex,
+		addr.UDP.IP,
 		nil,
 		gracefulShutdownC,
 		config.GracePeriod,
@@ -526,6 +616,7 @@ func ServeHTTP2(
 		err := listenReconnect(serveCtx, reconnectCh, gracefulShutdownC)
 		if err != nil {
 			// forcefully break the connection (this is only used for testing)
+			connLog.Logger().Debug().Msg("Forcefully breaking http2 connection")
 			_ = tlsServerConn.Close()
 		}
 		return err
@@ -584,6 +675,7 @@ func ServeQUIC(
 		err := listenReconnect(serveCtx, reconnectCh, gracefulShutdownC)
 		if err != nil {
 			// forcefully break the connection (this is only used for testing)
+			connLogger.Logger().Debug().Msg("Forcefully breaking quic connection")
 			quicConn.Close()
 		}
 		return err

token/token.go

@@ -29,6 +29,10 @@ const (
 	AccessLoginWorkerPath = "/cdn-cgi/access/login"
 )
 
+var (
+	userAgent = "DEV"
+)
+
 type AppInfo struct {
 	AuthDomain string
 	AppAUD     string
@@ -144,6 +148,10 @@ func isTokenLocked(lockFilePath string) bool {
 	return exists && err == nil
 }
 
+func Init(version string) {
+	userAgent = fmt.Sprintf("cloudflared/%s", version)
+}
+
 // FetchTokenWithRedirect will either load a stored token or generate a new one
 // it appends the full url as the redirect URL to the access cli request if opening the browser
 func FetchTokenWithRedirect(appURL *url.URL, appInfo *AppInfo, log *zerolog.Logger) (string, error) {
@@ -261,6 +269,7 @@ func GetAppInfo(reqURL *url.URL) (*AppInfo, error) {
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create app info request")
 	}
+	appInfoReq.Header.Add("User-Agent", userAgent)
 	resp, err := client.Do(appInfoReq)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get app info")
@@ -311,6 +320,7 @@ func exchangeOrgToken(appURL *url.URL, orgToken string) (string, error) {
 	if err != nil {
 		return "", errors.Wrap(err, "failed to create app token request")
 	}
+	appTokenRequest.Header.Add("User-Agent", userAgent)
 	resp, err := client.Do(appTokenRequest)
 	if err != nil {
 		return "", errors.Wrap(err, "failed to get app token")

token/transfer.go

@@ -113,7 +113,12 @@ func transferRequest(requestURL string, log *zerolog.Logger) ([]byte, string, er
 
 // poll the endpoint for the request resource, waiting for the user interaction
 func poll(client *http.Client, requestURL string, log *zerolog.Logger) ([]byte, string, error) {
-	resp, err := client.Get(requestURL)
+	req, err := http.NewRequest(http.MethodGet, requestURL, nil)
+	if err != nil {
+		return nil, "", err
+	}
+	req.Header.Set("User-Agent", userAgent)
+	resp, err := client.Do(req)
 	if err != nil {
 		return nil, "", err
 	}