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) + } +}