From a624eb4ce50214ca04f81def87574bde8ae06735 Mon Sep 17 00:00:00 2001 From: lalith0407 <58086270+lalith0407@users.noreply.github.com> Date: Sat, 17 Oct 2026 22:41:54 -0700 Subject: [PATCH] fix: bound connecting in the shared HTTP client and log setup retries GetHTTPClient built its transports without a dialer or a TLS handshake bound, so an unreachable host held every attempt for the OS SYN timeout (about two minutes on Linux). Flow setup retries three times and logged nothing per attempt, so a flow with an unreachable LLM endpoint sat silently for about seven minutes. Give both transports a 30s dial timeout (30s keep-alive) and a 10s TLS handshake timeout, as http.DefaultTransport and the reverse proxy in router.go do. Proxy and TLS settings are unchanged, and no response header timeout is added: slow answers stay bounded by HTTP_CLIENT_TIMEOUT. retryTransient now logs one warning per failed attempt. --- backend/pkg/providers/retry.go | 11 +- backend/pkg/providers/retry_test.go | 58 +++++ backend/pkg/system/utils.go | 21 ++ .../pkg/system/utils_timeout_linux_test.go | 127 +++++++++++ backend/pkg/system/utils_timeout_test.go | 201 ++++++++++++++++++ 5 files changed, 417 insertions(+), 1 deletion(-) create mode 100644 backend/pkg/system/utils_timeout_linux_test.go create mode 100644 backend/pkg/system/utils_timeout_test.go diff --git a/backend/pkg/providers/retry.go b/backend/pkg/providers/retry.go index 9ed5ea938..635f14667 100644 --- a/backend/pkg/providers/retry.go +++ b/backend/pkg/providers/retry.go @@ -9,6 +9,7 @@ import ( "strconv" "time" + "github.com/sirupsen/logrus" "github.com/vxcontrol/langchaingo/llms" "github.com/vxcontrol/langchaingo/llms/reasoning" ) @@ -80,9 +81,17 @@ func retryTransient[T any](ctx context.Context, delay time.Duration, call func() return result, nil case errors.Is(err, context.Canceled) || isDeterministicError(err): return zero, err - case attempt == maxRetriesToCallSimpleChain: + } + + logger := logrus.WithError(err).WithFields(logrus.Fields{ + "attempt": attempt, + "max_attempts": maxRetriesToCallSimpleChain, + }) + if attempt == maxRetriesToCallSimpleChain { + logger.Warn("call failed, no attempts left") return zero, fmt.Errorf("gave up after %d attempts: %w", attempt, err) } + logger.Warn("call failed, will retry") select { case <-ctx.Done(): diff --git a/backend/pkg/providers/retry_test.go b/backend/pkg/providers/retry_test.go index 97cc2834e..ccb0b2596 100644 --- a/backend/pkg/providers/retry_test.go +++ b/backend/pkg/providers/retry_test.go @@ -10,6 +10,8 @@ import ( "pentagi/pkg/providers/pconfig" "pentagi/pkg/providers/provider" + "github.com/sirupsen/logrus" + logtest "github.com/sirupsen/logrus/hooks/test" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/vxcontrol/langchaingo/llms" @@ -240,3 +242,59 @@ func TestRetry_RetryTransient_SetupCallIsNotAskedAgainAfterALocalRefusal(t *test assert.ErrorAs(t, err, new(*llms.ErrStructuredOutputUnsupported)) assert.Equal(t, 1, prv.callCount, "a refusal the library decided locally must not be asked again") } + +func TestRetry_RetryTransient_LogsEachFailedAttempt(t *testing.T) { + hook := logtest.NewGlobal() + t.Cleanup(hook.Reset) + + transient := errors.New("dial tcp 10.255.255.1:443: i/o timeout") + tests := []struct { + name string + errs []error + wantLogs []string + }{ + { + name: "every attempt fails", + errs: []error{transient}, + wantLogs: []string{"call failed, will retry", "call failed, will retry", "call failed, no attempts left"}, + }, + { + name: "the second attempt succeeds", + errs: []error{transient, nil}, + wantLogs: []string{"call failed, will retry"}, + }, + { + name: "a deterministic failure is not retried, so nothing is logged here", + errs: []error{&reasoning.ErrThinkingRequiresStream{Model: "qwen3.6-35b-a3b"}}, + }, + { + name: "a canceled call is not logged as a failure", + errs: []error{context.Canceled}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + hook.Reset() + + calls := 0 + _, _ = retryTransient(context.Background(), time.Millisecond, func() (string, error) { + calls++ + return "ok", tt.errs[min(calls, len(tt.errs))-1] + }) + + var got []string + for i, entry := range hook.AllEntries() { + if _, ok := entry.Data["max_attempts"]; !ok { + continue + } + got = append(got, entry.Message) + assert.Equal(t, logrus.WarnLevel, entry.Level) + assert.Equal(t, len(got), entry.Data["attempt"], "entry %d", i) + assert.Equal(t, maxRetriesToCallSimpleChain, entry.Data["max_attempts"]) + assert.Equal(t, transient, entry.Data[logrus.ErrorKey]) + } + assert.Equal(t, tt.wantLogs, got) + }) + } +} diff --git a/backend/pkg/system/utils.go b/backend/pkg/system/utils.go index 7a5078c49..743cb50fd 100644 --- a/backend/pkg/system/utils.go +++ b/backend/pkg/system/utils.go @@ -4,6 +4,7 @@ import ( "crypto/tls" "crypto/x509" "fmt" + "net" "net/http" "net/url" "os" @@ -17,6 +18,15 @@ const ( defaultHTTPClientTimeout = 10 * time.Minute ) +// Without them an unreachable host holds every attempt for the OS SYN timeout +// (about two minutes on Linux). They bound connecting only: a slow model answer +// is still limited by HTTPClientTimeout alone. +var ( + httpDialTimeout = 30 * time.Second + httpDialKeepAlive = 30 * time.Second + httpTLSHandshakeTimeout = 10 * time.Second +) + func getHostname() string { hn, err := os.Hostname() if err != nil { @@ -73,6 +83,8 @@ func GetHTTPClient(cfg *config.Config) (*http.Client, error) { Proxy: func(req *http.Request) (*url.URL, error) { return url.Parse(cfg.ProxyURL) }, + DialContext: newHTTPDialer().DialContext, + TLSHandshakeTimeout: httpTLSHandshakeTimeout, TLSClientConfig: &tls.Config{ RootCAs: rootCAPool, InsecureSkipVerify: cfg.ExternalSSLInsecure, @@ -83,6 +95,8 @@ func GetHTTPClient(cfg *config.Config) (*http.Client, error) { httpClient = &http.Client{ Timeout: timeout, Transport: &http.Transport{ + DialContext: newHTTPDialer().DialContext, + TLSHandshakeTimeout: httpTLSHandshakeTimeout, TLSClientConfig: &tls.Config{ RootCAs: rootCAPool, InsecureSkipVerify: cfg.ExternalSSLInsecure, @@ -93,3 +107,10 @@ func GetHTTPClient(cfg *config.Config) (*http.Client, error) { return httpClient, nil } + +func newHTTPDialer() *net.Dialer { + return &net.Dialer{ + Timeout: httpDialTimeout, + KeepAlive: httpDialKeepAlive, + } +} diff --git a/backend/pkg/system/utils_timeout_linux_test.go b/backend/pkg/system/utils_timeout_linux_test.go new file mode 100644 index 000000000..561daa011 --- /dev/null +++ b/backend/pkg/system/utils_timeout_linux_test.go @@ -0,0 +1,127 @@ +package system + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "syscall" + "testing" + "time" +) + +// blackholeAddr returns a local address that never completes a TCP handshake: a +// listener with a zero backlog that never accepts holds one connection in its +// queue and Linux silently drops every SYN after it, as an unreachable host does. +func blackholeAddr(t *testing.T) string { + t.Helper() + + fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatalf("failed to create socket: %v", err) + } + t.Cleanup(func() { syscall.Close(fd) }) + + if err := syscall.Bind(fd, &syscall.SockaddrInet4{Addr: [4]byte{127, 0, 0, 1}}); err != nil { + t.Fatalf("failed to bind: %v", err) + } + if err := syscall.Listen(fd, 0); err != nil { + t.Fatalf("failed to listen: %v", err) + } + sa, err := syscall.Getsockname(fd) + if err != nil { + t.Fatalf("failed to read the bound address: %v", err) + } + addr := fmt.Sprintf("127.0.0.1:%d", sa.(*syscall.SockaddrInet4).Port) + + filler, err := net.DialTimeout("tcp", addr, time.Second) + if err != nil { + t.Fatalf("failed to fill the accept queue: %v", err) + } + t.Cleanup(func() { filler.Close() }) + + return addr +} + +func TestGetHTTPClient_BlackholedHostFailsWithinTheDialTimeout(t *testing.T) { + shortenConnectTimeouts(t, 300*time.Millisecond, time.Second) + + tests := []struct { + name string + target func(addr string) (url, proxy string) + }{ + { + name: "unreachable endpoint", + target: func(addr string) (string, string) { return "https://" + addr + "/v1/chat/completions", "" }, + }, + { + name: "unreachable proxy", + target: func(addr string) (string, string) { + return "https://api.openai.com/v1/chat/completions", "http://" + addr + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + url, proxyURL := tt.target(blackholeAddr(t)) + client, err := GetHTTPClient(createTestConfig("", false, proxyURL)) + if err != nil { + t.Fatalf("expected no error, got: %v", err) + } + + // The guard only keeps a missing bound from hanging the test for minutes. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + if err != nil { + t.Fatalf("failed to build request: %v", err) + } + + start := time.Now() + resp, err := client.Do(req) + elapsed := time.Since(start) + if err == nil { + resp.Body.Close() + t.Fatal("expected the dial to fail") + } + + var netErr net.Error + if ctx.Err() != nil || !errors.As(err, &netErr) || !netErr.Timeout() { + t.Errorf("expected the dial itself to time out before the guard, got: %v", err) + } + if elapsed > 2*time.Second { + t.Errorf("expected to give up within the dial timeout, took %v", elapsed) + } + }) + } +} + +func TestGetHTTPClient_CancelDuringDialReturnsAtOnce(t *testing.T) { + client, err := GetHTTPClient(createTestConfig("", false, "")) + if err != nil { + t.Fatalf("expected no error, got: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(100*time.Millisecond, cancel) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+blackholeAddr(t), nil) + if err != nil { + t.Fatalf("failed to build request: %v", err) + } + + start := time.Now() + resp, err := client.Do(req) + elapsed := time.Since(start) + if err == nil { + resp.Body.Close() + t.Fatal("expected the request to be canceled") + } + if !errors.Is(err, context.Canceled) { + t.Errorf("expected context.Canceled, got: %v", err) + } + if elapsed > 2*time.Second { + t.Errorf("expected the cancel to stop the dial, took %v", elapsed) + } +} diff --git a/backend/pkg/system/utils_timeout_test.go b/backend/pkg/system/utils_timeout_test.go new file mode 100644 index 000000000..bbe2d3540 --- /dev/null +++ b/backend/pkg/system/utils_timeout_test.go @@ -0,0 +1,201 @@ +package system + +import ( + "context" + "crypto/tls" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// shortenConnectTimeouts makes the connect bounds small enough for a test to wait them out. +func shortenConnectTimeouts(t *testing.T, dial, handshake time.Duration) { + t.Helper() + + savedDial, savedHandshake := httpDialTimeout, httpTLSHandshakeTimeout + httpDialTimeout, httpTLSHandshakeTimeout = dial, handshake + t.Cleanup(func() { + httpDialTimeout, httpTLSHandshakeTimeout = savedDial, savedHandshake + }) +} + +func TestGetHTTPClient_BoundsConnectingOnly(t *testing.T) { + certs, err := generateTestCerts() + if err != nil { + t.Fatalf("failed to generate test certs: %v", err) + } + caPath := createTempFile(t, certs.rootCAPEM) + + tests := []struct { + name string + insecure bool + proxyURL string + }{ + {name: "direct", insecure: false}, + {name: "direct insecure", insecure: true}, + {name: "through a proxy", proxyURL: "http://proxy.example.com:8080"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, err := GetHTTPClient(createTestConfig(caPath, tt.insecure, tt.proxyURL)) + if err != nil { + t.Fatalf("expected no error, got: %v", err) + } + + transport, ok := client.Transport.(*http.Transport) + if !ok { + t.Fatal("expected http.Transport") + } + + if transport.DialContext == nil { + t.Error("expected a bounded dialer") + } + if transport.TLSHandshakeTimeout != httpTLSHandshakeTimeout { + t.Errorf("expected TLS handshake timeout %v, got %v", httpTLSHandshakeTimeout, transport.TLSHandshakeTimeout) + } + if transport.ResponseHeaderTimeout != 0 { + t.Errorf("a slow model answer must not be cut off, got response header timeout %v", transport.ResponseHeaderTimeout) + } + if client.Timeout != 600*time.Second { + t.Errorf("expected the configured overall timeout, got %v", client.Timeout) + } + + if transport.TLSClientConfig == nil || transport.TLSClientConfig.RootCAs == nil { + t.Fatal("expected the custom root CA pool to be kept") + } + if transport.TLSClientConfig.InsecureSkipVerify != tt.insecure { + t.Errorf("expected InsecureSkipVerify %v", tt.insecure) + } + + if tt.proxyURL == "" { + if transport.Proxy != nil { + t.Error("expected no proxy") + } + return + } + proxy, err := transport.Proxy(httptest.NewRequest(http.MethodGet, "https://api.openai.com/v1/models", nil)) + if err != nil || proxy == nil || proxy.String() != tt.proxyURL { + t.Errorf("expected proxy %s, got %v (%v)", tt.proxyURL, proxy, err) + } + }) + } +} + +func TestGetHTTPClient_SilentServerFailsTheTLSHandshakeInTime(t *testing.T) { + shortenConnectTimeouts(t, time.Second, 200*time.Millisecond) + + // Accepts the connection and never answers the ClientHello. + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("failed to listen: %v", err) + } + t.Cleanup(func() { listener.Close() }) + go func() { + var held []net.Conn + defer func() { + for _, conn := range held { + conn.Close() + } + }() + for { + conn, err := listener.Accept() + if err != nil { + return + } + held = append(held, conn) + } + }() + + client, err := GetHTTPClient(createTestConfig("", false, "")) + if err != nil { + t.Fatalf("expected no error, got: %v", err) + } + + // The guard only keeps a missing bound from hanging the test for minutes. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+listener.Addr().String(), nil) + if err != nil { + t.Fatalf("failed to build request: %v", err) + } + + start := time.Now() + resp, err := client.Do(req) + elapsed := time.Since(start) + if err == nil { + resp.Body.Close() + t.Fatal("expected the handshake to fail") + } + if !strings.Contains(err.Error(), "TLS handshake timeout") { + t.Errorf("expected a TLS handshake timeout, got: %v", err) + } + if elapsed > 2*time.Second { + t.Errorf("expected to give up within the handshake timeout, took %v", elapsed) + } +} + +func TestGetHTTPClient_SlowAnswerIsNotCutOff(t *testing.T) { + shortenConnectTimeouts(t, 100*time.Millisecond, 100*time.Millisecond) + + certs, err := generateTestCerts() + if err != nil { + t.Fatalf("failed to generate test certs: %v", err) + } + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + time.Sleep(time.Second) // ten times both connect bounds, like a model thinking before it answers + _, _ = io.WriteString(w, "OK") + })) + server.TLS = &tls.Config{Certificates: []tls.Certificate{{ + Certificate: [][]byte{certs.serverCert.Raw, certs.intermediate.Raw}, + PrivateKey: certs.serverKey, + }}} + server.StartTLS() + t.Cleanup(server.Close) + + client, err := GetHTTPClient(createTestConfig(createTempFile(t, certs.rootCAPEM), false, "")) + if err != nil { + t.Fatalf("expected no error, got: %v", err) + } + + resp, err := client.Get(server.URL) + if err != nil { + t.Fatalf("a slow answer must not be cut off: %v", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil || string(body) != "OK" { + t.Errorf("expected body OK, got %q (%v)", body, err) + } +} + +func TestGetHTTPClient_RequestStillGoesThroughTheProxy(t *testing.T) { + shortenConnectTimeouts(t, time.Second, time.Second) + + var seen string + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + seen = r.URL.String() + _, _ = io.WriteString(w, "via proxy") + })) + t.Cleanup(proxy.Close) + + client, err := GetHTTPClient(createTestConfig("", false, proxy.URL)) + if err != nil { + t.Fatalf("expected no error, got: %v", err) + } + + resp, err := client.Get("http://llm.example.invalid/v1/models") + if err != nil { + t.Fatalf("expected the proxy to answer, got: %v", err) + } + defer resp.Body.Close() + + if seen != "http://llm.example.invalid/v1/models" { + t.Errorf("expected the proxy to receive the request, got %q", seen) + } +}