From c5a6fcd84609f2cc8e7661591082925ca1237fdf Mon Sep 17 00:00:00 2001 From: balaji Date: Thu, 20 Aug 2026 08:39:25 -0700 Subject: [PATCH 1/2] fix(worker): stop paying dial timeouts for proxy pods that are gone A stateful work request names the proxy pod that issued it, so when that pod goes away every request naming it is doomed. The worker found out the slow way, and then forgot, so the next request naming the same dead pod paid the same cost again. Measured against a blackhole socket through the real dial path: a single attempt took 5.002s and a full work request 30.15s. Each of those seconds is a worker concurrency slot held while achieving nothing, so a function's drain rate collapses to concurrency/30s. With clients still retrying, demand outruns the drain and the function stays pinned, which is why removing demand has been the only reliable remedy. Two causes, fixed together. HandshakeIdleTimeout was never set, so quic-go's 5s default applied. MaxIdleTimeout is 8s and the cancel timer is 8.5s, so neither was what fired. Set it explicitly: the dial either completes across the cluster network in well under a second or it never will. Nothing remembered a dead host. Added a per-host breaker in the QUIC connection cache: three consecutive failed dials refuse the host for 30s, then a single probe decides whether to resume. It records only dial outcomes, which is what makes it safe, because a 403 arrives on a connection that dialled successfully and so proves the pod is alive. By construction it cannot blackhole a pod that is answering. Dials are already coalesced per hostname, so the threshold counts dial rounds rather than callers. The refusal is permanent for backoff so the caller stops instead of spending its budget failing fast. Measured after: single dial 2.001s, full request 6.06s, and a subsequent request for a known-dead host 0.024ms. Adds a benchmark recording all three, plus a test asserting the handshake bound is in force, so neither can silently regress the way the unset timeout did. Co-Authored-By: Balaji Ganesan --- src/libraries/go/worker/proxy/BUILD.bazel | 5 + .../go/worker/proxy/dead_host_dial_test.go | 193 +++++++++++++++++ src/libraries/go/worker/proxy/h3.go | 35 ++- src/libraries/go/worker/proxy/host_breaker.go | 204 ++++++++++++++++++ .../go/worker/proxy/host_breaker_test.go | 201 +++++++++++++++++ src/libraries/go/worker/proxy/proxy.go | 4 + 6 files changed, 639 insertions(+), 3 deletions(-) create mode 100644 src/libraries/go/worker/proxy/dead_host_dial_test.go create mode 100644 src/libraries/go/worker/proxy/host_breaker.go create mode 100644 src/libraries/go/worker/proxy/host_breaker_test.go diff --git a/src/libraries/go/worker/proxy/BUILD.bazel b/src/libraries/go/worker/proxy/BUILD.bazel index cca7ca1ab..bd996d66c 100644 --- a/src/libraries/go/worker/proxy/BUILD.bazel +++ b/src/libraries/go/worker/proxy/BUILD.bazel @@ -19,6 +19,7 @@ go_library( name = "proxy", srcs = [ "h3.go", + "host_breaker.go", "proxy.go", "push_listener.go", "routing.go", @@ -59,7 +60,9 @@ go_library( go_test( name = "proxy_test", srcs = [ + "dead_host_dial_test.go", "h3_addrlist_test.go", + "host_breaker_test.go", "proxy_e2e_test.go", "proxy_extra_test.go", "proxy_test.go", @@ -77,11 +80,13 @@ go_test( "//src/libraries/go/worker/utils", "//src/libraries/go/worker/utils/middleware", "//src/libraries/go/worker/utils/pool", + "@com_github_cenkalti_backoff_v4//:backoff", "@com_github_google_uuid//:uuid", "@com_github_nats_io_nats_go//:nats_go", "@com_github_nats_io_nats_go//jetstream", "@com_github_quic_go_quic_go//http3", "@com_github_quic_go_quic_go//integrationtests/tools", + "@com_github_stretchr_testify//assert", "@com_github_stretchr_testify//require", "@org_golang_google_protobuf//proto", "@org_golang_x_net//http2", diff --git a/src/libraries/go/worker/proxy/dead_host_dial_test.go b/src/libraries/go/worker/proxy/dead_host_dial_test.go new file mode 100644 index 000000000..144ae558f --- /dev/null +++ b/src/libraries/go/worker/proxy/dead_host_dial_test.go @@ -0,0 +1,193 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "context" + "net" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + pb "github.com/NVIDIA/nvcf/src/libraries/go/worker/proto/nvcf" +) + +// blackholeHost returns the address of a UDP socket that reads packets and +// never answers. That is the condition behind "timeout: no recent network +// activity": the proxy pod is gone, so nothing responds and the handshake has +// to time out rather than being refused. A closed port would be refused +// immediately and would not exercise this path at all. +func blackholeHost(t *testing.T) string { + t.Helper() + pc, err := net.ListenPacket("udp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = pc.Close() }) + go func() { + buf := make([]byte, 1500) + for { + if _, _, err := pc.ReadFrom(buf); err != nil { + return + } + } + }() + return pc.LocalAddr().String() +} + +func deadHostWork(addr string) *pb.WorkerInvokeFunctionRequest { + return &pb.WorkerInvokeFunctionRequest{ + RequestId: uuid.New().String(), + StatefulConfig: &pb.WorkerInvokeFunctionRequest_StatefulConfig{ + ConnectionConfigs: []*pb.WorkerInvokeFunctionRequest_StatefulConfig_ConnectionConfig{ + {Config: &pb.WorkerInvokeFunctionRequest_StatefulConfig_ConnectionConfig_Http3Config{ + Http3Config: &pb.WorkerInvokeFunctionRequest_StatefulConfig_ConnectionConfig_HTTP3ConnectionConfig{ + ProxyURI: "https://" + addr + "/v1/proxy", + ProxyAuthorizationToken: "dummy-token", + }}}, + }, + }, + } +} + +// HandshakeIdleTimeout was previously left unset, so quic-go's 5s default +// applied and a work request naming a dead pod held a worker concurrency slot +// for 30s. Nothing failed, which is why it went unnoticed; this asserts the +// bound is actually in force so it cannot silently regress. +func TestDialToDeadHostGivesUpWithinHandshakeTimeout(t *testing.T) { + setupLogger() + allowInsecure(t) + addr := blackholeHost(t) + + h3 := createH3RoundTripper() + require.Equal(t, handshakeIdleTimeout, h3.wrappedTransport.QUICConfig.HandshakeIdleTimeout, + "handshake timeout must be set explicitly, not left to the library default") + + cfg := deadHostWork(addr).StatefulConfig.ConnectionConfigs[0].GetHttp3Config() + + start := time.Now() + _, err := quicConnect(context.Background(), uuid.New().String(), cfg, h3) + elapsed := time.Since(start) + + require.Error(t, err) + assert.Less(t, elapsed, 2*handshakeIdleTimeout, + "a dial to a dead host took %v, expected to give up near %v", elapsed, handshakeIdleTimeout) +} + +// The breaker's whole purpose: the first work request naming a dead pod pays +// the dial timeouts, and everything after it is refused immediately instead of +// paying them again. Without this a deep backlog of requests naming a pod that +// no longer exists drains at one slot-blocking timeout each. +func TestDeadHostIsRefusedAfterRepeatedFailures(t *testing.T) { + setupLogger() + allowInsecure(t) + addr := blackholeHost(t) + + h3 := createH3RoundTripper() + work := deadHostWork(addr) + + // First request: pays the dial budget and trips the breaker on the way. + firstStart := time.Now() + _, err := getClientConnFromProxy(context.Background(), work, h3) + firstElapsed := time.Since(firstStart) + require.Error(t, err) + + host := "127.0.0.1:" + portOf(t, addr) + require.True(t, h3.breaker.isOpen(host), "breaker should be open after the first request failed repeatedly") + + // Second request naming the same dead host must not dial at all. + secondStart := time.Now() + _, err = getClientConnFromProxy(context.Background(), deadHostWork(addr), h3) + secondElapsed := time.Since(secondStart) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrHostUnreachable) + assert.Less(t, secondElapsed, 250*time.Millisecond, + "a request for a known-dead host should be refused immediately, took %v", secondElapsed) + assert.Less(t, secondElapsed, firstElapsed/4, + "refusal (%v) should be far cheaper than dialling (%v)", secondElapsed, firstElapsed) + + t.Logf("first request %v, subsequent request %v", firstElapsed, secondElapsed) +} + +func portOf(t *testing.T, addr string) string { + t.Helper() + _, port, err := net.SplitHostPort(addr) + require.NoError(t, err) + return port +} + +// BenchmarkDialDeadHost records what a work request naming a vanished proxy pod +// costs, which is the number that decides how long a backlog takes to drain +// after a proxy restart. Run with: +// +// go test ./proxy/ -run '^$' -bench BenchmarkDialDeadHost -benchtime 1x +func BenchmarkDialDeadHost(b *testing.B) { + quicInsecure = true + defer func() { quicInsecure = false }() + + b.Run("single dial", func(b *testing.B) { + addr := blackholeHostB(b) + h3 := createH3RoundTripper() + cfg := deadHostWork(addr).StatefulConfig.ConnectionConfigs[0].GetHttp3Config() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = quicConnect(context.Background(), uuid.New().String(), cfg, h3) + } + }) + + b.Run("full retry sequence", func(b *testing.B) { + addr := blackholeHostB(b) + h3 := createH3RoundTripper() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = getClientConnFromProxy(context.Background(), deadHostWork(addr), h3) + } + }) + + // The same sequence once the breaker has tripped, which is what every + // request after the first one in a backlog actually costs. + b.Run("full retry sequence with breaker open", func(b *testing.B) { + addr := blackholeHostB(b) + h3 := createH3RoundTripper() + _, _ = getClientConnFromProxy(context.Background(), deadHostWork(addr), h3) + b.ResetTimer() + for i := 0; i < b.N; i++ { + _, _ = getClientConnFromProxy(context.Background(), deadHostWork(addr), h3) + } + }) +} + +func blackholeHostB(b *testing.B) string { + b.Helper() + pc, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + b.Fatal(err) + } + b.Cleanup(func() { _ = pc.Close() }) + go func() { + buf := make([]byte, 1500) + for { + if _, _, err := pc.ReadFrom(buf); err != nil { + return + } + } + }() + return pc.LocalAddr().String() +} diff --git a/src/libraries/go/worker/proxy/h3.go b/src/libraries/go/worker/proxy/h3.go index 8a41058b2..6e77b3b83 100644 --- a/src/libraries/go/worker/proxy/h3.go +++ b/src/libraries/go/worker/proxy/h3.go @@ -53,8 +53,14 @@ func createH3RoundTripper() *h3ConnectionCache { QUICConfig: &quic.Config{ // TODO we are fully relying on client side timeouts for connection issue detection // TODO https://github.com/quic-go/quic-go/issues/153 is not implemented - KeepAlivePeriod: 3 * time.Second, - MaxIdleTimeout: 8 * time.Second, + KeepAlivePeriod: 3 * time.Second, + MaxIdleTimeout: 8 * time.Second, + // Set explicitly. Left unset this defaults to 5s inside quic-go, + // which is what a dial to a proxy pod that no longer exists costs + // before it gives up, six times over per work request. The dial + // either completes across the cluster network in well under a + // second or it is never going to. + HandshakeIdleTimeout: handshakeIdleTimeout, MaxIncomingStreams: math.MaxInt, MaxIncomingUniStreams: math.MaxInt, }, @@ -63,7 +69,11 @@ func createH3RoundTripper() *h3ConnectionCache { if utils.LevelFromEnv().Level() == zap.DebugLevel { h3.QUICConfig.Tracer = qlog.DefaultConnectionTracer } - return &h3ConnectionCache{wrappedTransport: h3, clients: make(map[string]*roundTripperWithCount)} + return &h3ConnectionCache{ + wrappedTransport: h3, + clients: make(map[string]*roundTripperWithCount), + breaker: newHostBreaker(), + } } // mostly copied from http3.Transport because we need to hijack the http3 client stream @@ -74,6 +84,8 @@ type h3ConnectionCache struct { quicTransport *quic.Transport mutex sync.Mutex clients map[string]*roundTripperWithCount + // breaker carries its own lock and is never held together with mutex. + breaker *hostBreaker } func (t *h3ConnectionCache) getDialedClient(ctx context.Context, hostname string) (rtc *roundTripperWithCount, isReused bool, err error) { @@ -96,6 +108,13 @@ func (t *h3ConnectionCache) getDialedClient(ctx context.Context, hostname string } func (t *h3ConnectionCache) getClient(ctx context.Context, hostname string) (rtc *roundTripperWithCount, isReused bool, err error) { + // Refuse hosts that have just failed repeatedly, before spending a dial + // timeout finding out again. Checked outside t.mutex: the breaker has its + // own lock and the two are never held together. + if err := t.breaker.allow(hostname); err != nil { + return nil, false, err + } + t.mutex.Lock() defer t.mutex.Unlock() @@ -119,8 +138,18 @@ func (t *h3ConnectionCache) getClient(ctx context.Context, hostname string) (rtc conn, rt, err := t.dial(ctx, hostname) if err != nil { cl.dialErr = err + if t.breaker.recordFailure(hostname) { + zap.L().Warn("no longer dialling proxy host after repeated failures", + zap.String("hostname", hostname), + zap.Duration("for", hostOpenDuration), + zap.Error(err)) + } return } + if t.breaker.recordSuccess(hostname) { + zap.L().Info("proxy host is accepting connections again, resuming dials", + zap.String("hostname", hostname)) + } cl.conn = conn cl.clientConn = rt context.AfterFunc(conn.Context(), func() { diff --git a/src/libraries/go/worker/proxy/host_breaker.go b/src/libraries/go/worker/proxy/host_breaker.go new file mode 100644 index 000000000..eb728f753 --- /dev/null +++ b/src/libraries/go/worker/proxy/host_breaker.go @@ -0,0 +1,204 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "errors" + "sync" + "time" + + "github.com/cenkalti/backoff/v4" +) + +// ErrHostUnreachable reports that a dial was refused because the host has just +// failed repeatedly. Callers match on this. +var ErrHostUnreachable = errors.New("proxy host is not accepting connections") + +// errHostRefused is what allow actually returns. The permanent wrapper stops +// the caller's retry loop, which would otherwise spend its whole budget failing +// fast against the same dead host. backoff unwraps it before returning, so +// callers still see ErrHostUnreachable. +var errHostRefused = backoff.Permanent(ErrHostUnreachable) + +const ( + // hostFailureThreshold is how many consecutive failed dials to one host + // open the breaker. Dials to a host are coalesced, so this counts dial + // rounds rather than callers: three means three genuinely failed dials, no + // matter how many sessions were waiting on them. Low enough to stop a dead + // pod quickly, high enough to ride out a single blip. + hostFailureThreshold = 3 + // hostOpenDuration is how long a host is refused before a probe is allowed + // through. Deliberately short: pod IPs get reused, so a stale entry must + // not be able to hold down an address that now belongs to a healthy pod. + hostOpenDuration = 30 * time.Second + // hostIdleRetention drops hosts nothing has talked to recently, so the + // table does not accumulate an entry per pod IP ever seen. + hostIdleRetention = 5 * time.Minute + // hostBreakerCapacity is a hard bound on the table. + hostBreakerCapacity = 4096 +) + +// hostBreaker refuses to dial proxy hosts that have just failed repeatedly. +// +// A work request carries the address of the proxy pod that issued it, so when +// that pod goes away every request naming it is doomed. Without any memory of +// that, each one pays the full dial-and-retry budget on its own while holding a +// worker concurrency slot, which is what turns a proxy restart into a backlog +// that takes hours to drain rather than seconds. +// +// It records only dial outcomes, which is what makes it safe. A 403 arrives on +// a connection that dialled successfully, so an authentication failure can +// never open the breaker; by construction this cannot blackhole a pod that is +// alive and answering. +type hostBreaker struct { + mu sync.Mutex + hosts map[string]*hostState + + // now is injectable so the state machine can be tested without sleeping. + now func() time.Time +} + +type hostState struct { + consecutiveFailures int + // openedAt is the zero time while the breaker is closed. + openedAt time.Time + // probing is set while a single half-open dial is in flight, so one caller + // probes and the rest are still refused. + probing bool + lastSeen time.Time +} + +func newHostBreaker() *hostBreaker { + return &hostBreaker{hosts: map[string]*hostState{}, now: time.Now} +} + +// allow reports whether a dial to host may proceed, returning +// ErrHostUnreachable when the host is being refused. +func (b *hostBreaker) allow(host string) error { + b.mu.Lock() + defer b.mu.Unlock() + + now := b.now() + b.evictLocked(now) + + st, ok := b.hosts[host] + if !ok { + b.hosts[host] = &hostState{lastSeen: now} + return nil + } + st.lastSeen = now + + if st.openedAt.IsZero() { + return nil + } + if now.Sub(st.openedAt) < hostOpenDuration { + return errHostRefused + } + // Half-open: exactly one dial is let through to find out whether the host + // is back. Everything else keeps being refused until it reports. + if st.probing { + return errHostRefused + } + st.probing = true + return nil +} + +// recordFailure reports a failed dial. Called once per dial round. +func (b *hostBreaker) recordFailure(host string) (opened bool) { + b.mu.Lock() + defer b.mu.Unlock() + + now := b.now() + st, ok := b.hosts[host] + if !ok { + st = &hostState{} + b.hosts[host] = st + } + st.lastSeen = now + + if st.probing { + // The probe failed, so the host is still gone. Restart the clock + // without letting the failure count run away. + st.probing = false + st.openedAt = now + return false + } + + st.consecutiveFailures++ + if st.consecutiveFailures >= hostFailureThreshold && st.openedAt.IsZero() { + st.openedAt = now + return true + } + return false +} + +// recordSuccess reports a dial that connected, which clears the host outright. +func (b *hostBreaker) recordSuccess(host string) (closed bool) { + b.mu.Lock() + defer b.mu.Unlock() + + st, ok := b.hosts[host] + if !ok { + b.hosts[host] = &hostState{lastSeen: b.now()} + return false + } + wasOpen := !st.openedAt.IsZero() + st.consecutiveFailures = 0 + st.openedAt = time.Time{} + st.probing = false + st.lastSeen = b.now() + return wasOpen +} + +// isOpen is for tests and logging; it does not change the state machine. +func (b *hostBreaker) isOpen(host string) bool { + b.mu.Lock() + defer b.mu.Unlock() + st, ok := b.hosts[host] + return ok && !st.openedAt.IsZero() +} + +// evictLocked keeps the table bounded. Hosts are pod addresses, so the set +// turns over continuously and would otherwise grow for the life of the worker. +func (b *hostBreaker) evictLocked(now time.Time) { + if len(b.hosts) < hostBreakerCapacity { + // Cheap path: only drop entries nothing has touched in a long time. + for host, st := range b.hosts { + if now.Sub(st.lastSeen) > hostIdleRetention { + delete(b.hosts, host) + } + } + return + } + // At capacity, drop the least recently used entry as well so an insert can + // always make room. + var oldestHost string + var oldest time.Time + for host, st := range b.hosts { + if now.Sub(st.lastSeen) > hostIdleRetention { + delete(b.hosts, host) + continue + } + if oldest.IsZero() || st.lastSeen.Before(oldest) { + oldest, oldestHost = st.lastSeen, host + } + } + if len(b.hosts) >= hostBreakerCapacity && oldestHost != "" { + delete(b.hosts, oldestHost) + } +} diff --git a/src/libraries/go/worker/proxy/host_breaker_test.go b/src/libraries/go/worker/proxy/host_breaker_test.go new file mode 100644 index 000000000..1074fa92d --- /dev/null +++ b/src/libraries/go/worker/proxy/host_breaker_test.go @@ -0,0 +1,201 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package proxy + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeClock drives the state machine without sleeping. +type fakeClock struct{ t atomic.Int64 } + +func (c *fakeClock) now() time.Time { return time.Unix(0, c.t.Load()) } +func (c *fakeClock) add(d time.Duration) { c.t.Add(int64(d)) } + +func newTestBreaker() (*hostBreaker, *fakeClock) { + clock := &fakeClock{} + clock.t.Store(int64(time.Hour)) // away from the zero time + b := newHostBreaker() + b.now = clock.now + return b, clock +} + +func TestHostBreakerOpensOnlyAfterThresholdFailures(t *testing.T) { + b, _ := newTestBreaker() + const host = "10-0-0-1.example:443" + + for i := 0; i < hostFailureThreshold-1; i++ { + require.NoError(t, b.allow(host)) + b.recordFailure(host) + assert.False(t, b.isOpen(host), "must not open before the threshold") + } + + require.NoError(t, b.allow(host)) + b.recordFailure(host) + assert.True(t, b.isOpen(host)) + assert.ErrorIs(t, b.allow(host), ErrHostUnreachable) +} + +// One success clears the count, so an occasional failure never accumulates into +// a trip against a host that is working. +func TestHostBreakerSuccessResetsFailureCount(t *testing.T) { + b, _ := newTestBreaker() + const host = "10-0-0-2.example:443" + + for i := 0; i < hostFailureThreshold-1; i++ { + require.NoError(t, b.allow(host)) + b.recordFailure(host) + } + require.NoError(t, b.allow(host)) + b.recordSuccess(host) + + for i := 0; i < hostFailureThreshold-1; i++ { + require.NoError(t, b.allow(host), "count should have restarted after the success") + b.recordFailure(host) + } + assert.False(t, b.isOpen(host)) +} + +// The retry loop must abort rather than burn its budget failing fast, so the +// refusal has to be permanent as far as backoff is concerned. +func TestHostBreakerRefusalIsPermanent(t *testing.T) { + b, _ := newTestBreaker() + const host = "10-0-0-3.example:443" + for i := 0; i < hostFailureThreshold; i++ { + require.NoError(t, b.allow(host)) + b.recordFailure(host) + } + + err := b.allow(host) + require.Error(t, err) + + var permanent *backoff.PermanentError + assert.ErrorAs(t, err, &permanent, "refusal must stop the retry loop") +} + +func TestHostBreakerAllowsOneProbeAfterOpenDuration(t *testing.T) { + b, clock := newTestBreaker() + const host = "10-0-0-4.example:443" + for i := 0; i < hostFailureThreshold; i++ { + require.NoError(t, b.allow(host)) + b.recordFailure(host) + } + require.ErrorIs(t, b.allow(host), ErrHostUnreachable) + + clock.add(hostOpenDuration + time.Second) + + assert.NoError(t, b.allow(host), "first caller after the open window should probe") + assert.ErrorIs(t, b.allow(host), ErrHostUnreachable, "only one probe may go out at a time") +} + +// A host that comes back must be usable immediately, not after another window. +func TestHostBreakerClosesWhenProbeSucceeds(t *testing.T) { + b, clock := newTestBreaker() + const host = "10-0-0-5.example:443" + for i := 0; i < hostFailureThreshold; i++ { + require.NoError(t, b.allow(host)) + b.recordFailure(host) + } + clock.add(hostOpenDuration + time.Second) + require.NoError(t, b.allow(host)) + + b.recordSuccess(host) + + assert.False(t, b.isOpen(host)) + assert.NoError(t, b.allow(host)) +} + +// A failed probe restarts the window rather than reopening immediately, so a +// host that stays dead is retried once per window and no more. +func TestHostBreakerFailedProbeReopensForAnotherWindow(t *testing.T) { + b, clock := newTestBreaker() + const host = "10-0-0-6.example:443" + for i := 0; i < hostFailureThreshold; i++ { + require.NoError(t, b.allow(host)) + b.recordFailure(host) + } + clock.add(hostOpenDuration + time.Second) + require.NoError(t, b.allow(host)) + + b.recordFailure(host) + + assert.True(t, b.isOpen(host)) + assert.ErrorIs(t, b.allow(host), ErrHostUnreachable) + + clock.add(hostOpenDuration + time.Second) + assert.NoError(t, b.allow(host), "a new window should permit another probe") +} + +// Hosts are pod addresses and turn over constantly, so the table must not grow +// for the life of the worker. +func TestHostBreakerEvictsIdleHosts(t *testing.T) { + b, clock := newTestBreaker() + + require.NoError(t, b.allow("10-0-0-7.example:443")) + clock.add(hostIdleRetention + time.Minute) + require.NoError(t, b.allow("10-0-0-8.example:443")) + + b.mu.Lock() + _, stale := b.hosts["10-0-0-7.example:443"] + b.mu.Unlock() + assert.False(t, stale, "idle host should have been evicted") +} + +func TestHostBreakerStaysBounded(t *testing.T) { + b, _ := newTestBreaker() + for i := 0; i < hostBreakerCapacity*2; i++ { + _ = b.allow(hostName(i)) + } + b.mu.Lock() + size := len(b.hosts) + b.mu.Unlock() + assert.LessOrEqual(t, size, hostBreakerCapacity) +} + +func hostName(i int) string { + return "host-" + time.Duration(i).String() + ".example:443" +} + +func TestHostBreakerConcurrentUse(t *testing.T) { + b, _ := newTestBreaker() + const host = "10-0-0-9.example:443" + + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + if err := b.allow(host); err != nil { + return + } + if i%2 == 0 { + b.recordFailure(host) + return + } + b.recordSuccess(host) + }(i) + } + wg.Wait() +} diff --git a/src/libraries/go/worker/proxy/proxy.go b/src/libraries/go/worker/proxy/proxy.go index 699fe7558..275caf2ea 100644 --- a/src/libraries/go/worker/proxy/proxy.go +++ b/src/libraries/go/worker/proxy/proxy.go @@ -64,6 +64,10 @@ import ( const ( retryConnectDelay = 10 * time.Millisecond reconnectOnErrWaitDuration = 30 * time.Second + // handshakeIdleTimeout bounds a single QUIC dial. See the note where the + // QUIC config is built: this used to be unset, so a dead host cost the + // quic-go default of 5s per attempt and 30s per work request. + handshakeIdleTimeout = 2 * time.Second ) var ErrAuth = backoff.Permanent(errors.New("permanent auth error")) From 12baa2d72fcad260d3f45364ce778769edcd05cc Mon Sep 17 00:00:00 2001 From: balaji Date: Thu, 20 Aug 2026 09:02:07 -0700 Subject: [PATCH 2/2] fix(worker): stop an unreported probe refusing a host forever Review found that a granted half-open probe was only ever cleared by a dial reporting back, so a probe that never reported refused the host for good. Idle eviction could not recover it either: a refused call still refreshes the entry, so under continuous traffic it never ages out. That is worse than the problem the breaker solves, because it blackholes a pod that may be perfectly healthy. Reproduced first: with traffic arriving every second, the host stayed refused indefinitely after one unreported probe. The probe token and the dial are genuinely not paired. getClient asks the breaker on every request but only dials when the connection cache has no entry for the host, so the token can be dropped on the floor. That is reachable rather than theoretical: a failed dial left its entry in the cache, and while the breaker was open nobody reached the cache to clear it, so the first probe after the window found the stale entry and returned it without ever dialling. Fixed at both ends. The probe is now a deadline rather than a flag, so an unreported probe expires and the host is offered another one, and a failed dial drops its cache entry immediately instead of waiting for a caller that will not arrive while the host is refused. Probes stay rate limited to one per timeout window, so a host that is genuinely gone is still dialled once per window and no more. Co-Authored-By: Balaji Ganesan --- src/libraries/go/worker/proxy/h3.go | 6 ++++ src/libraries/go/worker/proxy/host_breaker.go | 30 +++++++++++------ .../go/worker/proxy/host_breaker_test.go | 32 +++++++++++++++++++ 3 files changed, 58 insertions(+), 10 deletions(-) diff --git a/src/libraries/go/worker/proxy/h3.go b/src/libraries/go/worker/proxy/h3.go index 6e77b3b83..a1705c302 100644 --- a/src/libraries/go/worker/proxy/h3.go +++ b/src/libraries/go/worker/proxy/h3.go @@ -138,6 +138,12 @@ func (t *h3ConnectionCache) getClient(ctx context.Context, hostname string) (rtc conn, rt, err := t.dial(ctx, hostname) if err != nil { cl.dialErr = err + // Drop the failed entry now rather than leaving it for whichever + // caller happens to look next. While the breaker is open nobody + // reaches the cache, so a stale entry would survive the whole + // window and then be handed to the probe, which would return it + // without ever dialling. + cl.removeFromCache() if t.breaker.recordFailure(hostname) { zap.L().Warn("no longer dialling proxy host after repeated failures", zap.String("hostname", hostname), diff --git a/src/libraries/go/worker/proxy/host_breaker.go b/src/libraries/go/worker/proxy/host_breaker.go index eb728f753..fd6e74f97 100644 --- a/src/libraries/go/worker/proxy/host_breaker.go +++ b/src/libraries/go/worker/proxy/host_breaker.go @@ -51,6 +51,14 @@ const ( hostIdleRetention = 5 * time.Minute // hostBreakerCapacity is a hard bound on the table. hostBreakerCapacity = 4096 + // hostProbeTimeout bounds a granted half-open probe. The probe token and + // the dial are not strictly paired: the caller asks permission on every + // request but only dials when the connection cache has no entry for the + // host, so a granted token can be dropped without ever being reported. + // Without a deadline that would refuse the host forever, and idle eviction + // cannot recover it because a refused call still counts as activity. + // Comfortably longer than a dial, which is bounded by handshakeIdleTimeout. + hostProbeTimeout = 10 * time.Second ) // hostBreaker refuses to dial proxy hosts that have just failed repeatedly. @@ -77,10 +85,11 @@ type hostState struct { consecutiveFailures int // openedAt is the zero time while the breaker is closed. openedAt time.Time - // probing is set while a single half-open dial is in flight, so one caller - // probes and the rest are still refused. - probing bool - lastSeen time.Time + // probeStartedAt is set while a single half-open dial is in flight, so one + // caller probes and the rest are still refused. It is a time rather than a + // flag so an unreported probe expires instead of wedging the host. + probeStartedAt time.Time + lastSeen time.Time } func newHostBreaker() *hostBreaker { @@ -110,11 +119,12 @@ func (b *hostBreaker) allow(host string) error { return errHostRefused } // Half-open: exactly one dial is let through to find out whether the host - // is back. Everything else keeps being refused until it reports. - if st.probing { + // is back. Everything else keeps being refused until it reports or the + // probe times out. + if !st.probeStartedAt.IsZero() && now.Sub(st.probeStartedAt) < hostProbeTimeout { return errHostRefused } - st.probing = true + st.probeStartedAt = now return nil } @@ -131,10 +141,10 @@ func (b *hostBreaker) recordFailure(host string) (opened bool) { } st.lastSeen = now - if st.probing { + if !st.probeStartedAt.IsZero() { // The probe failed, so the host is still gone. Restart the clock // without letting the failure count run away. - st.probing = false + st.probeStartedAt = time.Time{} st.openedAt = now return false } @@ -160,7 +170,7 @@ func (b *hostBreaker) recordSuccess(host string) (closed bool) { wasOpen := !st.openedAt.IsZero() st.consecutiveFailures = 0 st.openedAt = time.Time{} - st.probing = false + st.probeStartedAt = time.Time{} st.lastSeen = b.now() return wasOpen } diff --git a/src/libraries/go/worker/proxy/host_breaker_test.go b/src/libraries/go/worker/proxy/host_breaker_test.go index 1074fa92d..603239c1d 100644 --- a/src/libraries/go/worker/proxy/host_breaker_test.go +++ b/src/libraries/go/worker/proxy/host_breaker_test.go @@ -199,3 +199,35 @@ func TestHostBreakerConcurrentUse(t *testing.T) { } wg.Wait() } + +// A probe that is granted but never reports must not refuse the host forever. +// The dial and the probe token are not strictly paired: getClient calls allow +// on every request but only dials when the host has no cache entry, so a +// granted token can be dropped on the floor. +func TestHostBreakerProbeThatNeverReportsDoesNotWedgeTheHost(t *testing.T) { + b, clock := newTestBreaker() + const host = "10-0-0-10.example:443" + for i := 0; i < hostFailureThreshold; i++ { + require.NoError(t, b.allow(host)) + b.recordFailure(host) + } + clock.add(hostOpenDuration + time.Second) + require.NoError(t, b.allow(host), "probe should be granted") + + // The probe never reports. Traffic keeps arriving, which is the real + // condition: each refused call refreshes lastSeen, so idle eviction never + // fires and cannot rescue the state. The host must still be offered further + // probes rather than being refused for good. + granted := 0 + for i := 0; i < 200; i++ { + clock.add(time.Second) + if b.allow(host) == nil { + granted++ + } + } + assert.NotZero(t, granted, + "a probe that never reported has refused the host permanently under continuous traffic") + // Bounded too: one probe per timeout window, not a probe per request. + assert.LessOrEqual(t, granted, int(200*time.Second/hostProbeTimeout)+1, + "probes should be rate limited to one per timeout window") +}