Merge branch 'main' into jnfrati/ubi-proxy

proxy/internal/acme/locker.go

@@ -2,12 +2,14 @@ package acme
 
 import (
 	"context"
+	"fmt"
 	"path/filepath"
 
 	log "github.com/sirupsen/logrus"
 
 	"github.com/netbirdio/netbird/proxy/internal/flock"
 	"github.com/netbirdio/netbird/proxy/internal/k8s"
+	"github.com/netbirdio/netbird/shared/management/domain"
 )
 
 // certLocker provides distributed mutual exclusion for certificate operations.
@@ -74,9 +76,15 @@ func newFlockLocker(certDir string, logger *log.Logger) *flockLocker {
 	return &flockLocker{certDir: certDir, logger: logger}
 }
 
-// Lock acquires an advisory file lock for the given domain.
-func (l *flockLocker) Lock(ctx context.Context, domain string) (func(), error) {
-	lockPath := filepath.Join(l.certDir, domain+".lock")
+// Lock acquires an advisory file lock for the given domain. The domain must
+// be a valid hostname so the lock file always resolves to a direct child of
+// certDir; anything else is rejected before touching the filesystem.
+func (l *flockLocker) Lock(ctx context.Context, name string) (func(), error) {
+	if !domain.IsValidDomainNoWildcard(name) {
+		return nil, fmt.Errorf("invalid domain %q for lock file", name)
+	}
+
+	lockPath := filepath.Join(l.certDir, name+".lock")
 	lockFile, err := flock.Lock(ctx, lockPath)
 	if err != nil {
 		return nil, err
@@ -89,7 +97,7 @@ func (l *flockLocker) Lock(ctx context.Context, domain string) (func(), error) {
 
 	return func() {
 		if err := flock.Unlock(lockFile); err != nil {
-			l.logger.Debugf("release cert lock for domain %q: %v", domain, err)
+			l.logger.Debugf("release cert lock for domain %q: %v", name, err)
 		}
 	}, nil
 }

proxy/internal/acme/locker_test.go

@@ -63,3 +63,33 @@ func TestNewCertLockerK8sFallsBackToFlock(t *testing.T) {
 	_, ok := locker.(*flockLocker)
 	assert.True(t, ok, "k8s-lease without SA should fall back to flockLocker")
 }
+
+func TestFlockLockerRejectsUnsafeDomain(t *testing.T) {
+	root := t.TempDir()
+	certDir := filepath.Join(root, "certs")
+	require.NoError(t, os.Mkdir(certDir, 0o700))
+	locker := newFlockLocker(certDir, nil)
+
+	for _, d := range []string{
+		"",
+		".",
+		"..",
+		"../escape",
+		"../../etc/cron.d/attacker",
+		"sub/dir.example.com",
+		`back\slash.example.com`,
+		"*.example.com",
+	} {
+		unlock, err := locker.Lock(context.Background(), d)
+		assert.Error(t, err, "domain %q", d)
+		assert.Nil(t, unlock, "domain %q", d)
+	}
+
+	assert.NoFileExists(t, filepath.Join(root, "escape.lock"))
+	certEntries, err := os.ReadDir(certDir)
+	require.NoError(t, err)
+	assert.Empty(t, certEntries)
+	rootEntries, err := os.ReadDir(root)
+	require.NoError(t, err)
+	assert.Len(t, rootEntries, 1)
+}

proxy/internal/debug/handler.go

@@ -105,6 +105,20 @@ type Handler struct {
 	startTime  time.Time
 	templates  *template.Template
 	templateMu sync.RWMutex
+
+	// setPerformance applies a buffer cap to one client. Held as a field so
+	// tests can drive applyBufferCap without a live embedded client.
+	setPerformance func(*nbembed.Client, uint32) error
+
+	perfMu       sync.Mutex
+	perfInflight map[types.AccountID]*perfWorker
+}
+
+// perfWorker is the single in-flight retune for one account. err is valid once
+// done is closed.
+type perfWorker struct {
+	done chan struct{}
+	err  error
 }
 
 // NewHandler creates a new debug handler.
@@ -113,10 +127,11 @@ func NewHandler(provider clientProvider, healthChecker healthChecker, logger *lo
 		logger = log.StandardLogger()
 	}
 	h := &Handler{
-		provider:  provider,
-		health:    healthChecker,
-		logger:    logger,
-		startTime: time.Now(),
+		provider:       provider,
+		health:         healthChecker,
+		logger:         logger,
+		startTime:      time.Now(),
+		setPerformance: setClientPerformance,
 	}
 	if err := h.loadTemplates(); err != nil {
 		logger.Errorf("failed to load embedded templates: %v", err)
@@ -716,15 +731,7 @@ func (h *Handler) handlePerf(w http.ResponseWriter, r *http.Request) {
 	}
 
 	capN := uint32(n)
-	applied := 0
-	failed := map[string]string{}
-	for accountID, client := range h.provider.ListClientsForStartup() {
-		if err := client.SetPerformance(nbembed.Performance{PreallocatedBuffersPerPool: &capN}); err != nil {
-			failed[string(accountID)] = err.Error()
-			continue
-		}
-		applied++
-	}
+	applied, failed, inFlight := h.applyBufferCap(capN)
 
 	resp := map[string]any{
 		"success": true,
@@ -734,9 +741,165 @@ func (h *Handler) handlePerf(w http.ResponseWriter, r *http.Request) {
 	if len(failed) > 0 {
 		resp["failed"] = failed
 	}
+	if len(inFlight) > 0 {
+		resp["in_flight"] = inFlight
+	}
 	h.writeJSON(w, resp)
 }
 
+// perfApplyTimeout bounds the whole apply, however many clients are registered.
+// A var, not a const, so tests can shorten the wait.
+var perfApplyTimeout = 5 * time.Second
+
+type perfResult struct {
+	accountID types.AccountID
+	err       error
+}
+
+// setClientPerformance is the production implementation behind Handler.setPerformance.
+func setClientPerformance(client *nbembed.Client, capN uint32) error {
+	return client.SetPerformance(nbembed.Performance{PreallocatedBuffersPerPool: &capN})
+}
+
+// collectBuffered takes every result already sitting in the channel, removing
+// those accounts from pending, and returns how many of them succeeded. It is
+// called when the deadline fires: select picks at random among ready cases, so
+// a result that landed in time would otherwise be reported as a timeout.
+func collectBuffered(results <-chan perfResult, pending map[types.AccountID]*perfWorker, failed map[string]string) int {
+	applied := 0
+	for {
+		select {
+		case res := <-results:
+			delete(pending, res.accountID)
+			if res.err != nil {
+				failed[string(res.accountID)] = res.err.Error()
+				continue
+			}
+			applied++
+		default:
+			return applied
+		}
+	}
+}
+
+// resolvePending closes out the accounts still pending when the deadline fires.
+// A worker whose done channel is closed has finished, whatever the results
+// channel has managed to deliver, so its own error is the truth; the rest are
+// genuinely still running and are reported as timed out. Returns how many of
+// them had in fact succeeded.
+func resolvePending(pending map[types.AccountID]*perfWorker, failed map[string]string) int {
+	applied := 0
+	for accountID, w := range pending {
+		select {
+		case <-w.done:
+			if w.err != nil {
+				failed[string(accountID)] = w.err.Error()
+				continue
+			}
+			applied++
+		default:
+			failed[string(accountID)] = fmt.Sprintf("timed out after %s waiting for the client", perfApplyTimeout)
+		}
+	}
+	return applied
+}
+
+// startPerfWorker returns the in-flight retune for the account, starting one if
+// there is none. The bool reports whether this call started it.
+//
+// At most one retune runs per account at a time. A client wedged inside its own
+// lock never returns, so without this a caller could add one permanently blocked
+// goroutine per request just by retrying the endpoint.
+func (h *Handler) startPerfWorker(accountID types.AccountID, client *nbembed.Client, capN uint32, results chan<- perfResult) (*perfWorker, bool) {
+	h.perfMu.Lock()
+	defer h.perfMu.Unlock()
+
+	if w, ok := h.perfInflight[accountID]; ok {
+		return w, false
+	}
+
+	w := &perfWorker{done: make(chan struct{})}
+	if h.perfInflight == nil {
+		h.perfInflight = make(map[types.AccountID]*perfWorker)
+	}
+	h.perfInflight[accountID] = w
+
+	go func() {
+		err := h.setPerformance(client, capN)
+		w.err = err
+		close(w.done)
+
+		// Publish before touching the registry: perfMu is taken once per
+		// account by every caller walking the fleet, so a finishing worker
+		// can queue behind a long apply and miss its own deadline.
+		results <- perfResult{accountID: accountID, err: err}
+
+		h.perfMu.Lock()
+		delete(h.perfInflight, accountID)
+		h.perfMu.Unlock()
+	}()
+
+	return w, true
+}
+
+// applyBufferCap sets the WireGuard buffer pool cap on every registered client
+// and reports how many took it, a per-account error for those that did not, and
+// the accounts whose earlier retune has not come back yet.
+//
+// Clients are handled concurrently and the wait is bounded: SetPerformance goes
+// through the embedded client's lock, which Start and Stop hold for as long as
+// they take - and on a wedged client Stop never returns. This endpoint is the
+// recovery path for exactly that fleet, so one stuck account must neither delay
+// the others nor accumulate goroutines across retries.
+func (h *Handler) applyBufferCap(capN uint32) (int, map[string]string, []string) {
+	clients := h.provider.ListClientsForStartup()
+	results := make(chan perfResult, len(clients))
+
+	applied := 0
+	failed := map[string]string{}
+	var inFlight []string
+	pending := make(map[types.AccountID]*perfWorker, len(clients))
+
+	for accountID, client := range clients {
+		w, started := h.startPerfWorker(accountID, client, capN, results)
+		if started {
+			pending[accountID] = w
+			continue
+		}
+		// Another request owns this account's retune. Take its result if it
+		// has already landed, otherwise report it as still running instead of
+		// waiting on it again.
+		select {
+		case <-w.done:
+			if w.err != nil {
+				failed[string(accountID)] = w.err.Error()
+				continue
+			}
+			applied++
+		default:
+			inFlight = append(inFlight, string(accountID))
+		}
+	}
+
+	deadline := time.After(perfApplyTimeout)
+	for range len(pending) {
+		select {
+		case res := <-results:
+			delete(pending, res.accountID)
+			if res.err != nil {
+				failed[string(res.accountID)] = res.err.Error()
+				continue
+			}
+			applied++
+		case <-deadline:
+			applied += collectBuffered(results, pending, failed)
+			applied += resolvePending(pending, failed)
+			return applied, failed, inFlight
+		}
+	}
+	return applied, failed, inFlight
+}
+
 // handleRuntime returns cheap runtime and process stats. Safe to hit on a
 // running proxy; does not read pprof profiles.
 func (h *Handler) handleRuntime(w http.ResponseWriter, _ *http.Request) {

proxy/internal/debug/perf_test.go

@@ -0,0 +1,158 @@
+package debug
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	nbembed "github.com/netbirdio/netbird/client/embed"
+	"github.com/netbirdio/netbird/proxy/internal/health"
+	"github.com/netbirdio/netbird/proxy/internal/roundtrip"
+	"github.com/netbirdio/netbird/proxy/internal/types"
+)
+
+// perfProvider serves a fixed set of accounts. The clients are nil: the tests
+// drive Handler.setPerformance, which never dereferences them.
+type perfProvider struct {
+	accounts []types.AccountID
+}
+
+func (p *perfProvider) GetClient(types.AccountID) (*nbembed.Client, bool) { return nil, false }
+
+func (p *perfProvider) ListClientsForDebug() map[types.AccountID]roundtrip.ClientDebugInfo {
+	return nil
+}
+
+func (p *perfProvider) ListClientsForStartup() map[types.AccountID]*nbembed.Client {
+	out := make(map[types.AccountID]*nbembed.Client, len(p.accounts))
+	for _, id := range p.accounts {
+		out[id] = nil
+	}
+	return out
+}
+
+type stubHealth struct{}
+
+func (stubHealth) ReadinessProbe() bool              { return true }
+func (stubHealth) StartupProbe(context.Context) bool { return true }
+func (stubHealth) CheckClientsConnected(context.Context) (bool, map[types.AccountID]health.ClientHealth) {
+	return true, nil
+}
+
+func shortenPerfTimeout(t *testing.T, d time.Duration) {
+	t.Helper()
+	prev := perfApplyTimeout
+	perfApplyTimeout = d
+	t.Cleanup(func() { perfApplyTimeout = prev })
+}
+
+// TestCollectBufferedCountsResultsReadyAtTheDeadline covers the select-ordering
+// trap: when the deadline fires, results already buffered must be counted, not
+// reported as timeouts. Driving collectBuffered directly keeps it deterministic
+// - through applyBufferCap the two select cases race by construction.
+func TestCollectBufferedCountsResultsReadyAtTheDeadline(t *testing.T) {
+	results := make(chan perfResult, 3)
+	results <- perfResult{accountID: "ok"}
+	results <- perfResult{accountID: "broken", err: errors.New("boom")}
+
+	pending := map[types.AccountID]*perfWorker{
+		"ok":     {done: make(chan struct{})},
+		"broken": {done: make(chan struct{})},
+		"wedged": {done: make(chan struct{})},
+	}
+	failed := map[string]string{}
+
+	applied := collectBuffered(results, pending, failed)
+
+	if applied != 1 {
+		t.Fatalf("applied = %d, want 1", applied)
+	}
+	if failed["broken"] != "boom" {
+		t.Fatalf("failed = %v, want the error recorded for \"broken\"", failed)
+	}
+	if _, ok := pending["wedged"]; !ok || len(pending) != 1 {
+		t.Fatalf("pending = %v, want only the account that never answered", pending)
+	}
+}
+
+// TestApplyBufferCapSingleFlightPerAccount covers the goroutine accumulation
+// reported on PR #7452: repeated calls against a client stuck in its own lock
+// must not start a second attempt for the same account.
+func TestApplyBufferCapSingleFlightPerAccount(t *testing.T) {
+	shortenPerfTimeout(t, 50*time.Millisecond)
+
+	release := make(chan struct{})
+	t.Cleanup(func() { close(release) })
+
+	var calls atomic.Int32
+	h := &Handler{
+		provider: &perfProvider{accounts: []types.AccountID{"wedged"}},
+		health:   stubHealth{},
+		setPerformance: func(_ *nbembed.Client, _ uint32) error {
+			calls.Add(1)
+			<-release
+			return nil
+		},
+	}
+
+	for i := range 5 {
+		applied, failed, inFlight := h.applyBufferCap(4096)
+		if applied != 0 {
+			t.Fatalf("call %d: applied = %d, want 0", i, applied)
+		}
+		if i == 0 {
+			if len(failed) != 1 {
+				t.Fatalf("first call: failed = %v, want the account reported as timed out", failed)
+			}
+			continue
+		}
+		if len(inFlight) != 1 {
+			t.Fatalf("call %d: inFlight = %v, want the account reported as still running", i, inFlight)
+		}
+		if len(failed) != 0 {
+			t.Fatalf("call %d: failed = %v, want empty while the retune is in flight", i, failed)
+		}
+	}
+
+	if got := calls.Load(); got != 1 {
+		t.Fatalf("setPerformance called %d times, want 1: each retry started another blocked worker", got)
+	}
+}
+
+// TestResolvePendingTrustsFinishedWorkers covers the reporting race cubic
+// flagged on PR #7452: a retune that finished just before the deadline must be
+// reported by its outcome, not as a timeout, whatever the results channel has
+// delivered so far.
+func TestResolvePendingTrustsFinishedWorkers(t *testing.T) {
+	ok := &perfWorker{done: make(chan struct{})}
+	close(ok.done)
+
+	broken := &perfWorker{done: make(chan struct{}), err: errors.New("boom")}
+	close(broken.done)
+
+	stillRunning := &perfWorker{done: make(chan struct{})}
+
+	pending := map[types.AccountID]*perfWorker{
+		"ok":      ok,
+		"broken":  broken,
+		"running": stillRunning,
+	}
+	failed := map[string]string{}
+
+	applied := resolvePending(pending, failed)
+
+	if applied != 1 {
+		t.Fatalf("applied = %d, want 1", applied)
+	}
+	if failed["broken"] != "boom" {
+		t.Fatalf("failed[broken] = %q, want the worker's own error", failed["broken"])
+	}
+	if _, ok := failed["ok"]; ok {
+		t.Fatalf("failed = %v, want no entry for the account that succeeded", failed)
+	}
+	if got := failed["running"]; got == "" || got == "boom" {
+		t.Fatalf("failed[running] = %q, want the timeout message", got)
+	}
+}

proxy/internal/roundtrip/multi.go

@@ -26,8 +26,8 @@ import (
 // branch at all), construct the MultiTransport via NewDirectOnly.
 type MultiTransport struct {
 	embedded http.RoundTripper
-	direct   *http.Transport
-	insecure *http.Transport
+	direct   *upstreamTransport
+	insecure *upstreamTransport
 }
 
 // errNoEmbeddedTransport is returned when a request reaches the
@@ -53,7 +53,6 @@ func NewMultiTransport(embedded http.RoundTripper, logger *log.Logger) *MultiTra
 	}
 	direct := &http.Transport{
 		DialContext:           dialWithTimeout(dialer.DialContext),
-		ForceAttemptHTTP2:     true,
 		MaxIdleConns:          cfg.maxIdleConns,
 		MaxIdleConnsPerHost:   cfg.maxIdleConnsPerHost,
 		MaxConnsPerHost:       cfg.maxConnsPerHost,
@@ -70,8 +69,8 @@ func NewMultiTransport(embedded http.RoundTripper, logger *log.Logger) *MultiTra
 
 	return &MultiTransport{
 		embedded: embedded,
-		direct:   direct,
-		insecure: insecure,
+		direct:   newUpstreamTransport(direct, cfg.upstreamHTTPVersion, logger),
+		insecure: newUpstreamTransport(insecure, cfg.upstreamHTTPVersion, logger),
 	}
 }
 

proxy/internal/roundtrip/multi_test.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"strings"
 	"testing"
 	"time"
@@ -75,16 +76,71 @@ func TestMultiTransport_AppliesEnvOverridesToDirect(t *testing.T) {
 
 	mt := NewMultiTransport(&stubRoundTripper{body: "embedded"}, nil)
 
-	assert.Equal(t, 42, mt.direct.MaxIdleConns,
+	assert.Equal(t, 42, mt.direct.primary.MaxIdleConns,
 		"NB_PROXY_MAX_IDLE_CONNS must propagate to the direct transport")
-	assert.Equal(t, 11*time.Second, mt.direct.IdleConnTimeout,
+	assert.Equal(t, 11*time.Second, mt.direct.primary.IdleConnTimeout,
 		"NB_PROXY_IDLE_CONN_TIMEOUT must propagate to the direct transport")
-	assert.Equal(t, 7*time.Second, mt.direct.TLSHandshakeTimeout,
+	assert.Equal(t, 7*time.Second, mt.direct.primary.TLSHandshakeTimeout,
 		"NB_PROXY_TLS_HANDSHAKE_TIMEOUT must propagate to the direct transport")
-	assert.Equal(t, 42, mt.insecure.MaxIdleConns,
+	assert.Equal(t, 42, mt.insecure.primary.MaxIdleConns,
 		"env tuning must also apply to the insecure-skip-verify direct transport")
 }
 
+// TestMultiTransport_UpstreamHTTPVersion pins the protocol actually
+// negotiated with an HTTPS upstream that offers both h2 and http/1.1.
+// The request rides the insecure clone, so this also covers the version
+// surviving http.Transport.Clone.
+func TestMultiTransport_UpstreamHTTPVersion(t *testing.T) {
+	tests := []struct {
+		name      string
+		env       string
+		wantProto string
+	}{
+		{name: "unset negotiates h2", env: "", wantProto: "HTTP/2.0"},
+		{name: "auto negotiates h2", env: "auto", wantProto: "HTTP/2.0"},
+		{name: "1.1 pins http/1.1", env: "1.1", wantProto: "HTTP/1.1"},
+		{name: "2 negotiates h2", env: "2", wantProto: "HTTP/2.0"},
+		{name: "unsupported value keeps the default", env: "http3", wantProto: "HTTP/2.0"},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			// t.Setenv registers the restore for whatever the process
+			// inherited; unsetting afterwards lets the default row
+			// exercise a genuinely absent variable.
+			t.Setenv(EnvUpstreamHTTPVersion, tc.env)
+			if tc.env == "" {
+				require.NoError(t, os.Unsetenv(EnvUpstreamHTTPVersion))
+			}
+
+			// The test server's certificate isn't in any root pool, so the
+			// request rides the insecure branch via WithSkipTLSVerify.
+			srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				_, _ = io.WriteString(w, r.Proto)
+			}))
+			srv.EnableHTTP2 = true
+			srv.StartTLS()
+			defer srv.Close()
+
+			mt := NewDirectOnly(nil)
+			ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background()))
+			req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil)
+			require.NoError(t, err)
+
+			resp, err := mt.RoundTrip(req)
+			require.NoError(t, err)
+			body, err := io.ReadAll(resp.Body)
+			_ = resp.Body.Close()
+			require.NoError(t, err)
+
+			assert.Equal(t, tc.wantProto, resp.Proto,
+				"client-side protocol must follow %s=%q", EnvUpstreamHTTPVersion, tc.env)
+			assert.Equal(t, tc.wantProto, string(body),
+				"the upstream must see the same protocol the client negotiated")
+		})
+	}
+}
+
 // TestMultiTransport_NilEmbeddedErrorsWhenWGPathRequested guards
 // against the previous silent fallback: a MultiTransport constructed
 // without an embedded transport must reject requests that don't

proxy/internal/roundtrip/netbird.go

@@ -82,10 +82,10 @@ type serviceNotification struct {
 // clientEntry holds an embedded NetBird client and tracks which services use it.
 type clientEntry struct {
 	client    *embed.Client
-	transport *http.Transport
+	transport *upstreamTransport
 	// insecureTransport is a clone of transport with TLS verification disabled,
 	// used when per-target skip_tls_verify is set.
-	insecureTransport *http.Transport
+	insecureTransport *upstreamTransport
 	services          map[ServiceKey]serviceInfo
 	createdAt         time.Time
 	started           bool
@@ -414,7 +414,6 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account
 	// not work with reverse proxied requests.
 	transport := &http.Transport{
 		DialContext:           dialWithTimeout(client.DialContext),
-		ForceAttemptHTTP2:     true,
 		MaxIdleConns:          n.transportCfg.maxIdleConns,
 		MaxIdleConnsPerHost:   n.transportCfg.maxIdleConnsPerHost,
 		MaxConnsPerHost:       n.transportCfg.maxConnsPerHost,
@@ -426,15 +425,14 @@ func (n *NetBird) createClientEntry(ctx context.Context, accountID types.Account
 		ReadBufferSize:        n.transportCfg.readBufferSize,
 		DisableCompression:    n.transportCfg.disableCompression,
 	}
-
 	insecureTransport := transport.Clone()
 	insecureTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec
 
 	return &clientEntry{
 		client:            client,
 		services:          map[ServiceKey]serviceInfo{key: si},
-		transport:         transport,
-		insecureTransport: insecureTransport,
+		transport:         newUpstreamTransport(transport, n.transportCfg.upstreamHTTPVersion, n.logger),
+		insecureTransport: newUpstreamTransport(insecureTransport, n.transportCfg.upstreamHTTPVersion, n.logger),
 		createdAt:         time.Now(),
 		started:           false,
 		inflightMap:       make(map[backendKey]chan struct{}),

proxy/internal/roundtrip/transport.go

@@ -1,8 +1,11 @@
 package roundtrip
 
 import (
+	"crypto/tls"
+	"net/http"
 	"os"
 	"strconv"
+	"strings"
 	"time"
 
 	log "github.com/sirupsen/logrus"
@@ -21,6 +24,30 @@ const (
 	EnvReadBufferSize        = "NB_PROXY_READ_BUFFER_SIZE"
 	EnvDisableCompression    = "NB_PROXY_DISABLE_COMPRESSION"
 	EnvMaxInflight           = "NB_PROXY_MAX_INFLIGHT"
+	EnvUpstreamHTTPVersion   = "NB_PROXY_UPSTREAM_HTTP_VERSION"
+)
+
+// upstreamHTTPVersion selects the HTTP version the proxy uses towards an
+// upstream. The explicit values are absolute: they mean the same thing
+// however the transports are dialled and whatever the default becomes,
+// so operator configuration survives a change of default.
+type upstreamHTTPVersion string
+
+const (
+	// upstreamHTTPAuto leaves the choice to the upstream: h2 is offered
+	// alongside http/1.1 in the TLS handshake and the upstream picks.
+	// An upstream that picks h2 and then fails to serve it is moved to
+	// HTTP/1.1 on its own (see upstreamTransport), which is the part
+	// ALPN cannot express. This is the only value whose meaning tracks
+	// the proxy's default.
+	upstreamHTTPAuto upstreamHTTPVersion = "auto"
+	// upstreamHTTP11 never offers h2, so the upstream sees HTTP/1.1.
+	upstreamHTTP11 upstreamHTTPVersion = "1.1"
+	// upstreamHTTP2 offers h2 in the TLS handshake and keeps it there:
+	// an upstream that negotiates h2 and then breaks is never moved to
+	// HTTP/1.1. Cleartext upstreams stay on HTTP/1.1 regardless: the
+	// proxy speaks no h2c.
+	upstreamHTTP2 upstreamHTTPVersion = "2"
 )
 
 // transportConfig holds tunable parameters for the per-account HTTP transport.
@@ -37,6 +64,11 @@ type transportConfig struct {
 	disableCompression    bool
 	// maxInflight limits per-backend concurrent requests. 0 means unlimited.
 	maxInflight int
+	// upstreamHTTPVersion selects the HTTP version used towards HTTPS
+	// upstreams. The default negotiates it with each upstream; the
+	// explicit values are for backends whose advertised h2 support is
+	// unusable and whose failure mode the negotiation cannot see.
+	upstreamHTTPVersion upstreamHTTPVersion
 }
 
 func defaultTransportConfig() transportConfig {
@@ -47,6 +79,7 @@ func defaultTransportConfig() transportConfig {
 		idleConnTimeout:       90 * time.Second,
 		tlsHandshakeTimeout:   10 * time.Second,
 		expectContinueTimeout: 1 * time.Second,
+		upstreamHTTPVersion:   upstreamHTTPAuto,
 	}
 }
 
@@ -86,6 +119,9 @@ func loadTransportConfig(logger *log.Logger) transportConfig {
 	if v, ok := envInt(EnvMaxInflight, logger); ok {
 		cfg.maxInflight = v
 	}
+	if v, ok := envUpstreamHTTPVersion(EnvUpstreamHTTPVersion, logger); ok {
+		cfg.upstreamHTTPVersion = v
+	}
 
 	logger.WithFields(log.Fields{
 		"max_idle_conns":          cfg.maxIdleConns,
@@ -99,11 +135,83 @@ func loadTransportConfig(logger *log.Logger) transportConfig {
 		"read_buffer_size":        cfg.readBufferSize,
 		"disable_compression":     cfg.disableCompression,
 		"max_inflight":            cfg.maxInflight,
+		"upstream_http_version":   cfg.upstreamHTTPVersion,
 	}).Debug("backend transport configuration")
 
 	return cfg
 }
 
+// applyUpstreamHTTPVersion configures t's ALPN offer for the requested
+// HTTP version. It is the single place that decides which protocols a
+// transport offers, so changing the proxy's default only touches this
+// function and leaves every explicit operator setting intact. What
+// happens when a negotiated h2 upstream then fails belongs to
+// upstreamTransport, which owns the runtime half of "auto".
+//
+// HTTP/1.1 is pinned by clearing ForceAttemptHTTP2 and installing an
+// empty TLSNextProto, which disables h2 regardless of how the transport
+// is dialled. Relying on net/http's conservative default (h2 off
+// whenever a custom dialer is set) would silently start negotiating h2
+// again the day a transport switches to DialTLSContext.
+func applyUpstreamHTTPVersion(t *http.Transport, version upstreamHTTPVersion) {
+	if version == upstreamHTTP11 {
+		t.ForceAttemptHTTP2 = false
+		t.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{}
+		t.TLSClientConfig = withoutHTTP2ALPN(t.TLSClientConfig)
+		return
+	}
+	t.ForceAttemptHTTP2 = true
+}
+
+// withoutHTTP2ALPN drops h2 from the ALPN offer. Configuring h2 makes
+// net/http append h2 to the transport's TLSClientConfig, so a transport
+// cloned from one that already served a request carries that offer with
+// it. Left in place, the upstream would select a protocol this
+// transport then refuses to speak, and the response would come back as
+// h2 frames parsed as an HTTP/1.1 message.
+func withoutHTTP2ALPN(cfg *tls.Config) *tls.Config {
+	// A nil config offers no ALPN at all, which is already HTTP/1.1.
+	if cfg == nil || len(cfg.NextProtos) == 0 {
+		return cfg
+	}
+
+	protos := make([]string, 0, len(cfg.NextProtos))
+	for _, proto := range cfg.NextProtos {
+		if proto == "h2" {
+			continue
+		}
+		protos = append(protos, proto)
+	}
+	if len(protos) == len(cfg.NextProtos) {
+		return cfg
+	}
+
+	// Clone rather than edit in place: the caller may share this config
+	// with the transport it was cloned from.
+	stripped := cfg.Clone()
+	stripped.NextProtos = protos
+
+	return stripped
+}
+
+// envUpstreamHTTPVersion reads an upstream HTTP version from the
+// environment. An unrecognised value warns and leaves the default in
+// place rather than guessing at the operator's intent.
+func envUpstreamHTTPVersion(key string, logger *log.Logger) (upstreamHTTPVersion, bool) {
+	s := strings.TrimSpace(os.Getenv(key))
+	if s == "" {
+		return "", false
+	}
+	switch v := upstreamHTTPVersion(strings.ToLower(s)); v {
+	case upstreamHTTPAuto, upstreamHTTP11, upstreamHTTP2:
+		return v, true
+	default:
+		logger.Warnf("ignoring unsupported %s=%q, expected one of %q, %q, %q",
+			key, s, upstreamHTTPAuto, upstreamHTTP11, upstreamHTTP2)
+		return "", false
+	}
+}
+
 func envInt(key string, logger *log.Logger) (int, bool) {
 	s := os.Getenv(key)
 	if s == "" {

proxy/internal/roundtrip/upstream.go

@@ -0,0 +1,455 @@
+package roundtrip
+
+import (
+	"errors"
+	"net"
+	"net/http"
+	"net/netip"
+	"net/url"
+	"strings"
+	"sync"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// upstreamDowngradeTTL is how long an upstream stays pinned to HTTP/1.1
+// after an h2 failure that only implied it cannot serve h2. Bounded so a
+// fixed or replaced backend returns to h2 without restarting the proxy.
+// A pin the upstream asked for itself does not expire — see downgrade.
+const upstreamDowngradeTTL = 10 * time.Minute
+
+// downgrade is an upstream's HTTP/1.1 pin.
+type downgrade struct {
+	// expiry is when the pin lapses and the upstream is offered h2
+	// again. The zero time means it never does: the upstream answered
+	// HTTP_1_1_REQUIRED, which is a statement about how it is
+	// configured, not a fault that may clear on its own. Re-probing
+	// that every upstreamDowngradeTTL would buy nothing but a failed
+	// request per interval, so the pin holds until the transport goes
+	// away with the proxy or the account's client.
+	expiry time.Time
+}
+
+// permanent reports whether the upstream asked for this pin itself.
+func (d downgrade) permanent() bool {
+	return d.expiry.IsZero()
+}
+
+// active reports whether the pin still stands at now.
+func (d downgrade) active(now time.Time) bool {
+	return d.permanent() || now.Before(d.expiry)
+}
+
+// upstreamTransport carries requests to a single upstream family (one
+// TLS configuration) and implements what upstreamHTTPAuto means.
+//
+// ALPN already lets the upstream pick the protocol: primary offers both
+// h2 and http/1.1 and the server chooses. What ALPN cannot express is
+// an upstream that selects h2 and then fails to speak it — the case
+// this type handles. The first h2-level failure for a host pins that
+// host to fallback, an HTTP/1.1-only clone of primary, and the request
+// is retried there when it can be replayed.
+//
+// The downgrade is per upstream host, not per transport: one broken
+// backend must not drop every other backend to HTTP/1.1.
+type upstreamTransport struct {
+	// primary is the configured transport: h2 offered in ALPN for
+	// upstreamHTTPAuto and upstreamHTTP2, HTTP/1.1-only for
+	// upstreamHTTP11.
+	primary *http.Transport
+	// version decides whether a downgrade may happen at all. Only
+	// upstreamHTTPAuto downgrades; the explicit values are absolute.
+	version upstreamHTTPVersion
+	logger  *log.Logger
+
+	// fallbackMu guards the lazy fallback clone: most deployments never
+	// hit a broken h2 upstream and should not pay for a second
+	// connection pool.
+	fallbackMu sync.Mutex
+	fallback   *http.Transport
+
+	mu sync.RWMutex
+	// downgraded maps an upstream host to its HTTP/1.1 pin.
+	downgraded map[string]downgrade
+}
+
+// newUpstreamTransport wraps base for the requested HTTP version. base
+// must not be used directly afterwards: the wrapper owns it, including
+// its connection pool.
+func newUpstreamTransport(base *http.Transport, version upstreamHTTPVersion, logger *log.Logger) *upstreamTransport {
+	if logger == nil {
+		logger = log.StandardLogger()
+	}
+	applyUpstreamHTTPVersion(base, version)
+
+	return &upstreamTransport{
+		primary:    base,
+		version:    version,
+		logger:     logger,
+		downgraded: make(map[string]downgrade),
+	}
+}
+
+// RoundTrip implements http.RoundTripper.
+func (t *upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	if !t.mayDowngrade(req) {
+		return t.primary.RoundTrip(req)
+	}
+
+	host := upstreamKey(req.URL)
+	if t.isDowngraded(host) {
+		return t.http1().RoundTrip(req)
+	}
+
+	resp, err := t.primary.RoundTrip(req)
+	if err == nil || !isHTTP2ProtocolError(err) {
+		return resp, err
+	}
+
+	// HTTP_1_1_REQUIRED is the upstream saying it will not serve this
+	// request over h2 however often it is asked — IIS answers it for
+	// Windows Authentication and for client-certificate sites, where
+	// the cause is site configuration rather than a passing fault.
+	t.markDowngraded(host, isHTTP11Required(err))
+
+	if !safeToRetry(req, err) {
+		// The upstream may have carried out the request before failing
+		// to answer over h2, and repeating it could duplicate whatever
+		// it did. The host is pinned either way, so the next request
+		// goes out over HTTP/1.1.
+		t.logger.WithFields(log.Fields{
+			"upstream": host,
+			"method":   req.Method,
+		}).Debug("not retrying over HTTP/1.1: the upstream may already have applied this request")
+		return nil, err
+	}
+
+	retry, ok := replayable(req)
+	if !ok {
+		// The body is already consumed and cannot be regenerated, so
+		// this request fails, and the pin carries the next one.
+		return nil, err
+	}
+	return t.http1().RoundTrip(retry)
+}
+
+// CloseIdleConnections closes idle connections on both pools.
+func (t *upstreamTransport) CloseIdleConnections() {
+	t.primary.CloseIdleConnections()
+	if fallback := t.existingHTTP1(); fallback != nil {
+		fallback.CloseIdleConnections()
+	}
+}
+
+// mayDowngrade reports whether a failed request is a downgrade
+// candidate. Only upstreamHTTPAuto downgrades, and only for TLS
+// upstreams: the proxy speaks no h2c, so a cleartext upstream is
+// already on HTTP/1.1 and an error there says nothing about h2.
+func (t *upstreamTransport) mayDowngrade(req *http.Request) bool {
+	return t.version == upstreamHTTPAuto && req.URL != nil && req.URL.Scheme == "https"
+}
+
+func (t *upstreamTransport) isDowngraded(host string) bool {
+	t.mu.RLock()
+	pin, ok := t.downgraded[host]
+	t.mu.RUnlock()
+
+	if !ok {
+		return false
+	}
+	if pin.active(time.Now()) {
+		return true
+	}
+
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	// Re-read under the write lock rather than trusting the expired pin
+	// from above: a concurrent request may have re-pinned the host since,
+	// and that pin decides this request too. Reporting the stale read
+	// would send one request back to h2 against a live pin.
+	pin, ok = t.downgraded[host]
+	if !ok {
+		return false
+	}
+	if pin.active(time.Now()) {
+		return true
+	}
+	delete(t.downgraded, host)
+
+	return false
+}
+
+// markDowngraded pins host to HTTP/1.1. permanent marks a pin the
+// upstream asked for; anything else lapses after upstreamDowngradeTTL so
+// a repaired backend is offered h2 again.
+func (t *upstreamTransport) markDowngraded(host string, permanent bool) {
+	now := time.Now()
+	pin := downgrade{expiry: now.Add(upstreamDowngradeTTL)}
+	if permanent {
+		pin = downgrade{}
+	}
+
+	t.mu.Lock()
+	previous, pinned := t.downgraded[host]
+	// A permanent pin is never weakened back into an expiring one: the
+	// upstream has already said h2 is not on offer.
+	promoted := pinned && !previous.permanent() && permanent
+	if !pinned || !previous.permanent() {
+		t.downgraded[host] = pin
+	}
+	for h, existing := range t.downgraded {
+		if !existing.active(now) {
+			delete(t.downgraded, h)
+		}
+	}
+	t.mu.Unlock()
+
+	// Log a new pin, and a pin the upstream has since asked to make
+	// permanent — otherwise an operator would only ever see the "for the
+	// next 10m" line and never learn the upstream settled the question.
+	if pinned && !promoted {
+		return
+	}
+
+	entry := t.logger.WithField("upstream", host)
+	if permanent {
+		entry.Warnf("upstream answered HTTP_1_1_REQUIRED, using HTTP/1.1 for it from now on")
+		return
+	}
+	entry.Warnf("upstream negotiated HTTP/2 but failed to serve it, using HTTP/1.1 for the next %s (set %s=1.1 to pin it)",
+		upstreamDowngradeTTL, EnvUpstreamHTTPVersion)
+}
+
+// http1 returns the HTTP/1.1-only clone, creating it on first use.
+func (t *upstreamTransport) http1() *http.Transport {
+	t.fallbackMu.Lock()
+	defer t.fallbackMu.Unlock()
+
+	if t.fallback == nil {
+		fallback := t.primary.Clone()
+		applyUpstreamHTTPVersion(fallback, upstreamHTTP11)
+		t.fallback = fallback
+	}
+
+	return t.fallback
+}
+
+// existingHTTP1 returns the fallback transport only if it was already
+// created, so housekeeping never allocates a second connection pool for
+// an upstream that never needed one.
+func (t *upstreamTransport) existingHTTP1() *http.Transport {
+	t.fallbackMu.Lock()
+	defer t.fallbackMu.Unlock()
+
+	return t.fallback
+}
+
+// upstreamKey normalizes an authority for use as a pin key, so one
+// upstream cannot end up with two independent pins. DNS labels compare
+// case-insensitively, and the default HTTPS port is implied — every
+// downgrade path is TLS-only, so a bare host and the same host on :443
+// are the same upstream.
+func upstreamKey(u *url.URL) string {
+	host := normalizeUpstreamHost(u.Hostname())
+
+	port := u.Port()
+	if port == "" || port == "443" {
+		return host
+	}
+
+	// JoinHostPort rather than concatenation: an IPv6 literal needs its
+	// brackets back after Hostname stripped them.
+	return net.JoinHostPort(host, port)
+}
+
+// normalizeUpstreamHost folds the spellings of one host onto a single
+// key. An IP literal goes through netip so that the several textual
+// forms of one address (case, leading zeroes, a compressed run) collapse
+// and a v4-mapped address keys as the v4 address it is. A zone
+// identifier is left exactly as written: it names an interface, and
+// interface names are case-sensitive on the systems that have them, so
+// %eth0 and %ETH0 may be different links and must not share a pin.
+// Anything that is not an IP literal is a DNS name, which compares
+// case-insensitively.
+func normalizeUpstreamHost(host string) string {
+	if addr, err := netip.ParseAddr(host); err == nil {
+		return addr.Unmap().String()
+	}
+
+	return strings.ToLower(host)
+}
+
+// safeToRetry reports whether req may be sent a second time over
+// HTTP/1.1 after err ended its h2 attempt.
+//
+// A failure at the h2 layer does not say whether the upstream already
+// carried out the request, so replaying one that changes state could
+// duplicate it. Two cases are safe: a request whose repetition is
+// harmless by definition, and an upstream that told us it processed
+// nothing on the connection. The second is what makes the IIS case work
+// for every method — a site requiring HTTP/1.1 refuses at stream 0,
+// before the request is looked at.
+func safeToRetry(req *http.Request, err error) bool {
+	return idempotent(req) || upstreamProcessedNothing(err)
+}
+
+// idempotent reports whether repeating req is defined to be harmless.
+// It mirrors net/http's own retry rule (Request.isReplayable): a method
+// with no side effects, or a caller that promised the upstream
+// deduplicates by key.
+func idempotent(req *http.Request) bool {
+	if req.Header.Get("Idempotency-Key") != "" || req.Header.Get("X-Idempotency-Key") != "" {
+		return true
+	}
+
+	switch req.Method {
+	// An empty method means GET, as in net/http.
+	case "", http.MethodGet, http.MethodHead, http.MethodOptions, http.MethodTrace:
+		return true
+	}
+
+	return false
+}
+
+// upstreamProcessedNothing reports whether err describes a GOAWAY that
+// named this request's stream as one the upstream had not received, so
+// it cannot have acted on it. A stream error says the opposite: the
+// stream was open, so the request had been delivered.
+func upstreamProcessedNothing(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	msg := transportError(err).Error()
+
+	return strings.Contains(msg, goAwayStreamNotReceivedMarker) ||
+		strings.Contains(msg, goAwayNothingProcessedMarker)
+}
+
+// replayable returns a request that can be sent a second time, or
+// ok=false when the body is gone. A RoundTripper consumes and closes
+// the body it was given, so a retry needs either no body at all or
+// GetBody to produce a fresh one.
+func replayable(req *http.Request) (*http.Request, bool) {
+	if req.Body == nil || req.Body == http.NoBody {
+		return req, true
+	}
+	if req.GetBody == nil {
+		return nil, false
+	}
+
+	body, err := req.GetBody()
+	if err != nil {
+		return nil, false
+	}
+
+	retry := req.Clone(req.Context())
+	retry.Body = body
+
+	return retry, true
+}
+
+// http2ErrorMarkers are the substrings that identify an HTTP/2 protocol
+// failure. net/http bundles its own private copy of the http2 package,
+// so its errors cannot be matched by type from here: http2.StreamError
+// and friends in x/net are different types from the ones a
+// bundled-h2 transport returns. The strings below are the formats those
+// bundled errors print, and they are specific to h2 framing — a
+// downgrade must never be triggered by an ordinary network or TLS
+// error, which retrying on HTTP/1.1 would not fix.
+var http2ErrorMarkers = []string{
+	// Transport-level h2 failures, e.g.
+	// "http2: server sent GOAWAY and closed the connection".
+	"http2:",
+	// http2.StreamError, e.g. "stream error: stream ID 1; PROTOCOL_ERROR".
+	"stream error: stream ID",
+	// http2.ConnectionError, e.g. "connection error: PROTOCOL_ERROR".
+	"connection error: ",
+	// The code an upstream sends to say the request must be retried
+	// over HTTP/1.1, as a GOAWAY or on the stream.
+	http11RequiredMarker,
+}
+
+const (
+	// http11RequiredMarker is the error code an upstream sends to say the
+	// request belongs on HTTP/1.1. Unlike the other markers it is not a
+	// fault: the upstream is describing its own configuration.
+	http11RequiredMarker = "HTTP_1_1_REQUIRED"
+
+	// A GOAWAY carrying NO_ERROR closes a connection without complaint:
+	// a server draining before shutdown, recycling an application pool,
+	// capping requests per connection. The upstream speaks h2 perfectly
+	// well, so this must never pin it. The two spellings are the two
+	// formats the bundled transport prints the code in.
+	goAwayNoErrorEqualsMarker = "ErrCode=NO_ERROR"
+	goAwayNoErrorColonMarker  = "ErrCode:NO_ERROR"
+	// gracefulGoAwayMarker is errClientConnGotGoAway, which the bundled
+	// transport raises for a stream the server never received on a
+	// connection it is shutting down gracefully. It normally retries
+	// those itself on a new connection and this never surfaces.
+	gracefulGoAwayMarker = "Transport received Server's graceful shutdown GOAWAY"
+
+	// goAwayStreamNotReceivedMarker is the bundled transport's abort for
+	// the first stream on a connection whose GOAWAY carried a real error
+	// code — the IIS case. It sits in the same "streamID > LastStreamID"
+	// branch as the graceful abort, so the server had not received the
+	// stream (see net/http's h2_bundle.go).
+	goAwayStreamNotReceivedMarker = "Transport received GOAWAY from server ErrCode:"
+	// goAwayNothingProcessedMarker is a GoAwayError naming stream 0 as
+	// the last one received, which says the same thing. The trailing
+	// comma keeps it from matching LastStreamID=10 and the rest.
+	goAwayNothingProcessedMarker = "LastStreamID=0,"
+)
+
+// isHTTP11Required reports whether the upstream itself asked for
+// HTTP/1.1, rather than merely failing at h2.
+func isHTTP11Required(err error) bool {
+	return err != nil && strings.Contains(transportError(err).Error(), http11RequiredMarker)
+}
+
+// isHTTP2ProtocolError reports whether err says the upstream cannot
+// serve the h2 it negotiated.
+func isHTTP2ProtocolError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	msg := transportError(err).Error()
+
+	// A graceful GOAWAY is routine connection management, not an
+	// upstream that cannot serve h2.
+	if isGracefulGoAway(msg) {
+		return false
+	}
+
+	for _, marker := range http2ErrorMarkers {
+		if strings.Contains(msg, marker) {
+			return true
+		}
+	}
+
+	return false
+}
+
+// isGracefulGoAway reports whether msg describes a GOAWAY sent to close
+// a healthy connection rather than to report an inability to serve h2.
+func isGracefulGoAway(msg string) bool {
+	return strings.Contains(msg, goAwayNoErrorEqualsMarker) ||
+		strings.Contains(msg, goAwayNoErrorColonMarker) ||
+		strings.Contains(msg, gracefulGoAwayMarker)
+}
+
+// transportError strips a *url.Error wrapper, which prefixes the request
+// URL to the message. Markers are matched as substrings, so a URL left
+// in place could classify an ordinary dial or TLS failure as an h2 one
+// on the strength of the path alone.
+func transportError(err error) error {
+	var urlErr *url.Error
+	if errors.As(err, &urlErr) && urlErr.Err != nil {
+		return urlErr.Err
+	}
+
+	return err
+}

proxy/internal/roundtrip/upstream_test.go

@@ -0,0 +1,587 @@
+package roundtrip
+
+import (
+	"bufio"
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"errors"
+	"fmt"
+	"io"
+	"math/big"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/http2"
+)
+
+// TestUpstreamTransport_AutoFallsBackOnBrokenHTTP2 covers the case ALPN
+// cannot express: the upstream advertises h2, picks it, and then cannot
+// serve it. The request must still succeed, over HTTP/1.1, and the
+// upstream must stay on HTTP/1.1 for the requests that follow.
+func TestUpstreamTransport_AutoFallsBackOnBrokenHTTP2(t *testing.T) {
+	t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTPAuto))
+	srv := startBrokenHTTP2Server(t)
+
+	mt := NewDirectOnly(nil)
+	ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background()))
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil)
+	require.NoError(t, err)
+
+	resp, err := mt.RoundTrip(req)
+	require.NoError(t, err, "a replayable request must be retried on HTTP/1.1 instead of failing")
+	body, err := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	require.NoError(t, err)
+	assert.Equal(t, "HTTP/1.1", resp.Proto, "the retry must ride the HTTP/1.1 transport")
+	assert.Equal(t, "http/1.1", string(body), "the upstream must see an http/1.1 ALPN offer on the retry")
+
+	assert.True(t, mt.insecure.isDowngraded(srv.addr),
+		"the upstream must stay pinned to HTTP/1.1 after proving it cannot serve h2")
+	mt.insecure.mu.RLock()
+	pin := mt.insecure.downgraded[srv.addr]
+	mt.insecure.mu.RUnlock()
+	assert.True(t, pin.permanent(),
+		"an upstream answering HTTP_1_1_REQUIRED must not be re-probed for h2")
+
+	// The second request must not repeat the h2 attempt: the server
+	// counts h2 handshakes, so a repeat would show up here.
+	h2Attempts := srv.http2Handshakes()
+	req, err = http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil)
+	require.NoError(t, err)
+	resp, err = mt.RoundTrip(req)
+	require.NoError(t, err)
+	_ = resp.Body.Close()
+
+	assert.Equal(t, "HTTP/1.1", resp.Proto, "a pinned upstream must go straight to HTTP/1.1")
+	assert.Equal(t, h2Attempts, srv.http2Handshakes(),
+		"a pinned upstream must not be probed for h2 again until the pin expires")
+}
+
+// TestUpstreamTransport_ExplicitHTTP2NeverDowngrades pins the promise
+// that the explicit values are absolute: an operator who asked for h2
+// keeps h2, broken upstream or not.
+func TestUpstreamTransport_ExplicitHTTP2NeverDowngrades(t *testing.T) {
+	t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTP2))
+	srv := startBrokenHTTP2Server(t)
+
+	mt := NewDirectOnly(nil)
+	ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background()))
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+srv.addr, nil)
+	require.NoError(t, err)
+
+	resp, err := mt.RoundTrip(req)
+	if err == nil {
+		_ = resp.Body.Close()
+	}
+	require.Error(t, err, "NB_PROXY_UPSTREAM_HTTP_VERSION=2 must not fall back to HTTP/1.1")
+	assert.False(t, mt.insecure.isDowngraded(srv.addr), "an explicit version must never pin an upstream")
+}
+
+// TestUpstreamTransport_AutoDoesNotReplayUnsafeRequests covers the
+// other half of the fallback: an h2 failure says nothing about whether
+// the upstream already applied the request, so a state-changing one is
+// not replayed. The host is still pinned, so the next request rides
+// HTTP/1.1 without a second h2 attempt.
+func TestUpstreamTransport_AutoDoesNotReplayUnsafeRequests(t *testing.T) {
+	t.Setenv(EnvUpstreamHTTPVersion, string(upstreamHTTPAuto))
+	srv := startBrokenHTTP2Server(t)
+	// A stream error means the stream was open, so the upstream had the
+	// request in hand — unlike the GOAWAY at stream 0 the fake server
+	// sends, which states it processed nothing.
+	streamErr := http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol}
+
+	mt := NewDirectOnly(nil)
+	transport := mt.insecure
+	ctx := WithSkipTLSVerify(WithDirectUpstream(context.Background()))
+
+	assert.False(t, safeToRetry(newTestRequest(t, ctx, http.MethodPost, srv.addr), streamErr),
+		"a POST must not be replayed after a failure that may have been applied")
+	assert.True(t, safeToRetry(newTestRequest(t, ctx, http.MethodGet, srv.addr), streamErr),
+		"a GET is safe to replay whatever the failure was")
+
+	// The fake server's GOAWAY names stream 0, so even a POST is safe
+	// there and the request must succeed over HTTP/1.1.
+	resp, err := transport.RoundTrip(newTestRequest(t, ctx, http.MethodPost, srv.addr))
+	require.NoError(t, err, "a GOAWAY at stream 0 means the upstream applied nothing, so the POST may be replayed")
+	body, err := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	require.NoError(t, err)
+	assert.Equal(t, "http/1.1", string(body), "the retry must reach the upstream over http/1.1")
+
+	h2Attempts := srv.http2Handshakes()
+	resp, err = transport.RoundTrip(newTestRequest(t, ctx, http.MethodPost, srv.addr))
+	require.NoError(t, err)
+	_ = resp.Body.Close()
+	assert.Equal(t, h2Attempts, srv.http2Handshakes(),
+		"the pin must carry later requests without another h2 attempt")
+}
+
+func newTestRequest(t *testing.T, ctx context.Context, method, addr string) *http.Request {
+	t.Helper()
+
+	req, err := http.NewRequestWithContext(ctx, method, "https://"+addr, strings.NewReader("payload"))
+	require.NoError(t, err)
+
+	return req
+}
+
+func TestUpstreamKey(t *testing.T) {
+	tests := []struct {
+		name string
+		url  string
+		want string
+	}{
+		{name: "host", url: "https://backend.invalid/path", want: "backend.invalid"},
+		// DNS is case-insensitive, so these are one upstream and must
+		// share one pin.
+		{name: "mixed case", url: "https://Backend.INVALID/path", want: "backend.invalid"},
+		// The default port is implied on every path that can downgrade.
+		{name: "explicit default port", url: "https://backend.invalid:443/", want: "backend.invalid"},
+		{name: "non-default port", url: "https://backend.invalid:8443/", want: "backend.invalid:8443"},
+		// An IPv6 literal needs its brackets back after Hostname strips
+		// them, or the key is not a dialable authority.
+		{name: "ipv6 default port", url: "https://[2001:db8::1]/", want: "2001:db8::1"},
+		{name: "ipv6 with port", url: "https://[2001:db8::1]:8443/", want: "[2001:db8::1]:8443"},
+		// One address in three spellings: hex case, a leading zero and an
+		// uncompressed zero run are all the same upstream.
+		{name: "ipv6 upper case", url: "https://[2001:DB8::1]/", want: "2001:db8::1"},
+		{name: "ipv6 leading zero", url: "https://[2001:0db8::1]/", want: "2001:db8::1"},
+		{name: "ipv6 uncompressed", url: "https://[2001:db8:0:0:0:0:0:1]/", want: "2001:db8::1"},
+		// A v4-mapped address is the v4 address, not a second upstream.
+		{name: "v4-mapped", url: "https://[::ffff:192.0.2.1]/", want: "192.0.2.1"},
+		// A zone names an interface, and interface names are
+		// case-sensitive, so these two are different links.
+		{name: "ipv6 zone", url: "https://[fe80::1%25eth0]/", want: "fe80::1%eth0"},
+		{name: "ipv6 zone upper case", url: "https://[fe80::1%25ETH0]/", want: "fe80::1%ETH0"},
+		// The address before the zone still normalizes.
+		{name: "ipv6 zone with upper-case address", url: "https://[FE80::1%25eth0]/", want: "fe80::1%eth0"},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			parsed, err := url.Parse(tc.url)
+			require.NoError(t, err)
+
+			assert.Equal(t, tc.want, upstreamKey(parsed))
+		})
+	}
+}
+
+func TestSafeToRetry(t *testing.T) {
+	streamErr := http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol}
+	goAwayAtZero := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`)
+	goAwayLater := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=11, ErrCode=PROTOCOL_ERROR, debug=""`)
+
+	tests := []struct {
+		name    string
+		method  string
+		headers map[string]string
+		err     error
+		want    bool
+	}{
+		{name: "get", method: http.MethodGet, err: streamErr, want: true},
+		{name: "head", method: http.MethodHead, err: streamErr, want: true},
+		{name: "options", method: http.MethodOptions, err: streamErr, want: true},
+		{name: "trace", method: http.MethodTrace, err: streamErr, want: true},
+		{name: "post", method: http.MethodPost, err: streamErr, want: false},
+		{name: "put", method: http.MethodPut, err: streamErr, want: false},
+		{name: "patch", method: http.MethodPatch, err: streamErr, want: false},
+		{name: "delete", method: http.MethodDelete, err: streamErr, want: false},
+		// The upstream reported it handled nothing, so repeating the
+		// request cannot duplicate anything.
+		{name: "post with goaway at stream 0", method: http.MethodPost, err: goAwayAtZero, want: true},
+		// What the bundled transport actually raises for the first
+		// stream on a connection the upstream GOAWAYs with a real error
+		// code, which is the shape a real IIS site produces.
+		{
+			name:   "post with first-stream goaway abort",
+			method: http.MethodPost,
+			err:    errors.New("http2: Transport received GOAWAY from server ErrCode:HTTP_1_1_REQUIRED"),
+			want:   true,
+		},
+		// It handled earlier streams, so this one may have been applied.
+		{name: "post with goaway after other streams", method: http.MethodPost, err: goAwayLater, want: false},
+		{
+			name:    "post with idempotency key",
+			method:  http.MethodPost,
+			headers: map[string]string{"Idempotency-Key": "abc"},
+			err:     streamErr,
+			want:    true,
+		},
+		{
+			name:    "post with prefixed idempotency key",
+			method:  http.MethodPost,
+			headers: map[string]string{"X-Idempotency-Key": "abc"},
+			err:     streamErr,
+			want:    true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			req, err := http.NewRequest(tc.method, "https://backend.invalid", nil)
+			require.NoError(t, err)
+			for k, v := range tc.headers {
+				req.Header.Set(k, v)
+			}
+
+			assert.Equal(t, tc.want, safeToRetry(req, tc.err))
+		})
+	}
+}
+
+func TestUpstreamTransport_MayDowngrade(t *testing.T) {
+	tests := []struct {
+		name    string
+		version upstreamHTTPVersion
+		url     string
+		want    bool
+	}{
+		{name: "auto over TLS", version: upstreamHTTPAuto, url: "https://backend.invalid", want: true},
+		// The proxy speaks no h2c, so a cleartext upstream is already on
+		// HTTP/1.1 and its failures say nothing about h2.
+		{name: "auto cleartext", version: upstreamHTTPAuto, url: "http://backend.invalid", want: false},
+		{name: "explicit 1.1", version: upstreamHTTP11, url: "https://backend.invalid", want: false},
+		{name: "explicit 2", version: upstreamHTTP2, url: "https://backend.invalid", want: false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			transport := newUpstreamTransport(&http.Transport{}, tc.version, nil)
+			req, err := http.NewRequest(http.MethodGet, tc.url, nil)
+			require.NoError(t, err)
+
+			assert.Equal(t, tc.want, transport.mayDowngrade(req))
+		})
+	}
+}
+
+func TestUpstreamTransport_DowngradeExpires(t *testing.T) {
+	transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil)
+	transport.markDowngraded("backend.invalid:443", false)
+	require.True(t, transport.isDowngraded("backend.invalid:443"))
+
+	transport.mu.Lock()
+	transport.downgraded["backend.invalid:443"] = downgrade{expiry: time.Now().Add(-time.Second)}
+	transport.mu.Unlock()
+
+	assert.False(t, transport.isDowngraded("backend.invalid:443"),
+		"an expired pin must let the upstream be offered h2 again")
+	transport.mu.RLock()
+	_, stillTracked := transport.downgraded["backend.invalid:443"]
+	transport.mu.RUnlock()
+	assert.False(t, stillTracked, "an expired pin must not be kept around")
+}
+
+func TestUpstreamTransport_DowngradeIsPerUpstream(t *testing.T) {
+	transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil)
+	transport.markDowngraded("broken.invalid:443", false)
+
+	assert.True(t, transport.isDowngraded("broken.invalid:443"))
+	assert.False(t, transport.isDowngraded("healthy.invalid:443"),
+		"one broken upstream must not drop the others to HTTP/1.1")
+}
+
+// TestUpstreamTransport_HTTP11RequiredPinIsPermanent covers the IIS
+// case: HTTP_1_1_REQUIRED describes how the upstream is configured
+// (Windows Authentication, client certificates), so re-probing it every
+// upstreamDowngradeTTL would only buy a failed request per interval.
+func TestUpstreamTransport_HTTP11RequiredPinIsPermanent(t *testing.T) {
+	transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil)
+	transport.markDowngraded("iis.invalid:443", true)
+
+	transport.mu.RLock()
+	pin := transport.downgraded["iis.invalid:443"]
+	transport.mu.RUnlock()
+
+	assert.True(t, pin.permanent(), "an upstream that asked for HTTP/1.1 must not be re-probed")
+	assert.True(t, pin.active(time.Now().Add(100*upstreamDowngradeTTL)),
+		"a permanent pin must outlive any TTL")
+}
+
+func TestUpstreamTransport_PermanentPinSurvivesLaterFailures(t *testing.T) {
+	transport := newUpstreamTransport(&http.Transport{}, upstreamHTTPAuto, nil)
+	transport.markDowngraded("iis.invalid:443", true)
+	// A later ambiguous failure for the same upstream must not turn the
+	// permanent pin into an expiring one.
+	transport.markDowngraded("iis.invalid:443", false)
+
+	transport.mu.RLock()
+	pin := transport.downgraded["iis.invalid:443"]
+	transport.mu.RUnlock()
+
+	assert.True(t, pin.permanent(), "a permanent pin must never be weakened")
+}
+
+func TestIsHTTP11Required(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{
+			name: "goaway",
+			err:  errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`),
+			want: true,
+		},
+		{
+			name: "stream error",
+			err:  http2.StreamError{StreamID: 1, Code: http2.ErrCodeHTTP11Required},
+			want: true,
+		},
+		{
+			name: "other h2 failure",
+			err:  http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol},
+			want: false,
+		},
+		{name: "nil", err: nil, want: false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, isHTTP11Required(tc.err))
+		})
+	}
+}
+
+func TestIsHTTP2ProtocolError(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{
+			name: "goaway demanding http/1.1",
+			err:  errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=0, ErrCode=HTTP_1_1_REQUIRED, debug=""`),
+			want: true,
+		},
+		{
+			name: "stream error",
+			err:  http2.StreamError{StreamID: 1, Code: http2.ErrCodeProtocol},
+			want: true,
+		},
+		{
+			name: "connection error",
+			err:  http2.ConnectionError(http2.ErrCodeProtocol),
+			want: true,
+		},
+		{
+			name: "wrapped h2 error",
+			err:  errors.New("Get \"https://backend.invalid\": http2: client connection lost"),
+			want: true,
+		},
+		// A GOAWAY with NO_ERROR is a server draining a connection —
+		// recycling an application pool, capping requests per
+		// connection, shutting down gracefully. It speaks h2 fine.
+		{
+			name: "graceful goaway",
+			err:  errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=9, ErrCode=NO_ERROR, debug=""`),
+			want: false,
+		},
+		{
+			name: "graceful shutdown abort",
+			err:  errors.New("http2: Transport received Server's graceful shutdown GOAWAY"),
+			want: false,
+		},
+		// Markers are substrings, so a URL carried by a *url.Error must
+		// not be able to classify a plain failure as an h2 one.
+		{
+			name: "url error whose path looks like a marker",
+			err: &url.Error{
+				Op:  "Get",
+				URL: "https://backend.invalid/http2:/connection error: x",
+				Err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused"),
+			},
+			want: false,
+		},
+		// Retrying these on HTTP/1.1 fixes nothing, so they must never
+		// pin an upstream.
+		{name: "dial failure", err: errors.New("dial tcp 10.0.0.1:443: connect: connection refused"), want: false},
+		{name: "tls failure", err: errors.New("tls: failed to verify certificate: x509: certificate signed by unknown authority"), want: false},
+		{name: "context cancelled", err: context.Canceled, want: false},
+		{name: "eof", err: io.EOF, want: false},
+		{name: "nil", err: nil, want: false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, isHTTP2ProtocolError(tc.err))
+		})
+	}
+}
+
+func TestReplayable(t *testing.T) {
+	t.Run("bodyless request", func(t *testing.T) {
+		req, err := http.NewRequest(http.MethodGet, "https://backend.invalid", nil)
+		require.NoError(t, err)
+
+		retry, ok := replayable(req)
+		require.True(t, ok)
+		assert.Same(t, req, retry, "a bodyless request needs no clone")
+	})
+
+	t.Run("request with GetBody", func(t *testing.T) {
+		req, err := http.NewRequest(http.MethodPost, "https://backend.invalid", strings.NewReader("payload"))
+		require.NoError(t, err)
+		// Consume the body the way a failed RoundTrip would.
+		_, err = io.ReadAll(req.Body)
+		require.NoError(t, err)
+
+		retry, ok := replayable(req)
+		require.True(t, ok)
+		body, err := io.ReadAll(retry.Body)
+		require.NoError(t, err)
+		assert.Equal(t, "payload", string(body), "the retry must carry a fresh copy of the body")
+	})
+
+	t.Run("streamed request", func(t *testing.T) {
+		req, err := http.NewRequest(http.MethodPost, "https://backend.invalid", io.NopCloser(strings.NewReader("payload")))
+		require.NoError(t, err)
+		require.Nil(t, req.GetBody, "an opaque reader must not get a GetBody")
+
+		_, ok := replayable(req)
+		assert.False(t, ok, "a body that cannot be regenerated must not be replayed")
+	})
+}
+
+// brokenHTTP2Server advertises h2 in ALPN, accepts it, and then refuses
+// to serve it — the upstream behaviour that motivated the fallback. Over
+// http/1.1 it answers normally, so a downgraded request succeeds.
+type brokenHTTP2Server struct {
+	addr string
+
+	handshakes chan struct{}
+}
+
+func (s *brokenHTTP2Server) http2Handshakes() int {
+	return len(s.handshakes)
+}
+
+func startBrokenHTTP2Server(t *testing.T) *brokenHTTP2Server {
+	t.Helper()
+
+	ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{
+		Certificates: []tls.Certificate{selfSignedCert(t)},
+		NextProtos:   []string{"h2", "http/1.1"},
+		MinVersion:   tls.VersionTLS12,
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = ln.Close() })
+
+	srv := &brokenHTTP2Server{
+		addr: ln.Addr().String(),
+		// Buffered well past what the test drives so a stuck server
+		// never blocks the accept loop.
+		handshakes: make(chan struct{}, 64),
+	}
+
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go srv.handle(conn)
+		}
+	}()
+
+	return srv
+}
+
+func (s *brokenHTTP2Server) handle(conn net.Conn) {
+	defer func() { _ = conn.Close() }()
+
+	tlsConn, ok := conn.(*tls.Conn)
+	if !ok {
+		return
+	}
+	if err := tlsConn.Handshake(); err != nil {
+		return
+	}
+
+	proto := tlsConn.ConnectionState().NegotiatedProtocol
+	if proto == "h2" {
+		select {
+		case s.handshakes <- struct{}{}:
+		default:
+		}
+		s.refuseHTTP2(tlsConn)
+		return
+	}
+
+	s.serveHTTP1(tlsConn, proto)
+}
+
+// refuseHTTP2 completes just enough of the h2 handshake for the client
+// to accept the connection, then sends the GOAWAY an upstream uses to
+// say the request belongs on HTTP/1.1.
+//
+// The client is still writing its preface and request while the GOAWAY
+// goes out, so the connection is drained before the caller closes it.
+// Closing a socket with unread bytes still in its receive buffer makes
+// the kernel answer with RST, which reaches the client as a write error
+// rather than the GOAWAY — no h2 error, so no downgrade, and the test
+// fails on the error the client saw first.
+func (s *brokenHTTP2Server) refuseHTTP2(conn net.Conn) {
+	framer := http2.NewFramer(conn, conn)
+	if err := framer.WriteSettings(); err != nil {
+		return
+	}
+	if err := framer.WriteGoAway(0, http2.ErrCodeHTTP11Required, nil); err != nil {
+		return
+	}
+
+	// The client closes its side once it has read the GOAWAY, which ends
+	// the drain; the deadline is only a backstop against a client that
+	// never does.
+	_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
+	_, _ = io.Copy(io.Discard, conn)
+}
+
+// serveHTTP1 answers a single request with the ALPN protocol the
+// upstream actually settled on, so a test asserting on the body is
+// checking what the upstream saw rather than a constant.
+func (s *brokenHTTP2Server) serveHTTP1(conn net.Conn, alpn string) {
+	reader := bufio.NewReader(conn)
+	if _, err := http.ReadRequest(reader); err != nil {
+		return
+	}
+
+	_, _ = fmt.Fprintf(conn,
+		"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s",
+		len(alpn), alpn)
+}
+
+func selfSignedCert(t *testing.T) tls.Certificate {
+	t.Helper()
+
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	require.NoError(t, err)
+
+	template := x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "127.0.0.1"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		KeyUsage:     x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
+		IsCA:         true,
+	}
+	der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
+	require.NoError(t, err)
+
+	return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}
+}