From 0ccb13c36eec6d0e11276217b2fd3e950d40a200 Mon Sep 17 00:00:00 2001 From: Rustem Kamalov Date: Fri, 24 Apr 2026 04:33:41 +0300 Subject: [PATCH] fix(proxy): isolate proxy health from captcha/parser errors, replace re-enable thrash with quarantine --- core/common.go | 61 +++++++++++ core/http_client.go | 22 +++- core/proxy.go | 85 ++++++++++++++- core/proxy_test.go | 247 +++++++++++++++++++++++++++++++++++++++++--- core/resilient.go | 6 +- core/server_test.go | 3 +- 6 files changed, 403 insertions(+), 21 deletions(-) diff --git a/core/common.go b/core/common.go index 3fde123..20b6a0c 100644 --- a/core/common.go +++ b/core/common.go @@ -1,9 +1,13 @@ package core import ( + "context" "errors" + "fmt" + "net" "sort" "strconv" + "strings" "time" "github.com/gofiber/fiber/v2" @@ -25,6 +29,63 @@ var ErrParser = errors.New("parser failure") // panic and converted it into a typed error. var ErrEngineInternal = errors.New("engine internal error") +// ErrProxyConnect is returned when the proxy cannot establish a network +// connection. Proxy health is degraded on this error. +var ErrProxyConnect = errors.New("proxy_connect") + +// ErrProxyAuth is returned when proxy credentials are rejected. +// Proxy health is degraded on this error. +var ErrProxyAuth = errors.New("proxy_auth") + +// ErrTimeout is returned when a network-level timeout occurs on the proxy path. +// Proxy health is degraded on this error. +var ErrTimeout = errors.New("timeout") + +// ErrEmptyResult signals a successful fetch that returned zero organic results. +// It is not a failure; the proxy stays healthy and no credit is charged. +var ErrEmptyResult = errors.New("empty_result") + +// IsProxyNetworkError reports whether err is a network-level error that +// indicates a faulty proxy (connect failure, auth rejection, or timeout). +// Parser drift, captcha pages, and engine errors must NOT degrade proxy health. +func IsProxyNetworkError(err error) bool { + return errors.Is(err, ErrProxyConnect) || + errors.Is(err, ErrProxyAuth) || + errors.Is(err, ErrTimeout) +} + +// classifyProxyNetworkError wraps common transport errors with proxy-health +// sentinels while preserving the original error for callers. +func classifyProxyNetworkError(err error) error { + if err == nil || IsProxyNetworkError(err) || errors.Is(err, context.Canceled) { + return err + } + + msg := strings.ToLower(err.Error()) + if strings.Contains(msg, "407") || strings.Contains(msg, "proxy authentication") { + return fmt.Errorf("%w: %w", ErrProxyAuth, err) + } + + var netErr net.Error + if (errors.As(err, &netErr) && netErr.Timeout()) || + errors.Is(err, context.DeadlineExceeded) || + strings.Contains(msg, "timeout") || + strings.Contains(msg, "deadline exceeded") { + return fmt.Errorf("%w: %w", ErrTimeout, err) + } + + if strings.Contains(msg, "proxyconnect") || + strings.Contains(msg, "connection refused") || + strings.Contains(msg, "connection reset") || + strings.Contains(msg, "no such host") || + strings.Contains(msg, "network is unreachable") || + strings.Contains(msg, "socks connect") { + return fmt.Errorf("%w: %w", ErrProxyConnect, err) + } + + return err +} + // SearchResult represents one normalized result item returned by any engine. type SearchResult struct { // Rank is a 1-based position in engine output. Some engines use negative diff --git a/core/http_client.go b/core/http_client.go index 4f503db..9f25f5c 100644 --- a/core/http_client.go +++ b/core/http_client.go @@ -31,9 +31,13 @@ func NewRawHTTPClient(query Query) (*http.Client, error) { if err != nil { return nil, err } + roundTripper := http.RoundTripper(transport) + if transport.Proxy != nil { + roundTripper = proxyErrorTransport{base: transport} + } return &http.Client{ - Transport: transport, + Transport: roundTripper, Timeout: rawHTTPTimeout, }, nil } @@ -84,3 +88,19 @@ func newRawTransport(query Query) (*http.Transport, error) { return transport, nil } + +type proxyErrorTransport struct { + base http.RoundTripper +} + +func (t proxyErrorTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.base.RoundTrip(req) + if err != nil { + return nil, classifyProxyNetworkError(err) + } + if resp != nil && resp.StatusCode == http.StatusProxyAuthRequired { + DrainAndCloseResponse(resp) + return nil, classifyProxyNetworkError(ErrProxyAuth) + } + return resp, nil +} diff --git a/core/proxy.go b/core/proxy.go index c8ce179..890dd26 100644 --- a/core/proxy.go +++ b/core/proxy.go @@ -8,6 +8,7 @@ import ( "sort" "strings" "sync" + "time" "github.com/sirupsen/logrus" ) @@ -19,6 +20,9 @@ const ( ProxyModeTagPool = "tag_pool" DefaultProxyFailureThreshold = 3 ProxyOverrideDirect = "direct" + // ProxyPoolQuarantineDuration is how long an exhausted tag pool stays quarantined + // before a single probe proxy is re-enabled for recovery testing. + ProxyPoolQuarantineDuration = 5 * time.Minute ) var supportedProxySchemes = map[string]struct{}{ @@ -97,6 +101,7 @@ type ProxyRegistry struct { order []string tagIndex map[string][]string nextByTag map[string]int + tagQuarantine map[string]time.Time // tag to earliest time pool may probe again failureThreshold int } @@ -347,6 +352,7 @@ func NewProxyRegistry(entries []ProxyEntryConfig, failureThreshold int) (*ProxyR order: order, tagIndex: tagIndex, nextByTag: make(map[string]int, len(tagIndex)), + tagQuarantine: make(map[string]time.Time, len(tagIndex)), failureThreshold: failureThreshold, }, nil } @@ -370,13 +376,30 @@ func (r *ProxyRegistry) NextByTagWithContext(ctx context.Context, tag string) st } if r.allDisabledLocked(urls) { - WithRequest(ctx).WithField("proxy_tag", tag).Warn( - fmt.Sprintf("Proxy tag pool exhausted for %q, re-enabling tagged proxies", tag), - ) - for _, proxyURL := range urls { - state := r.states[proxyURL] + now := time.Now() + quarantineUntil := r.tagQuarantine[tag] + if quarantineUntil.IsZero() { + r.startTagQuarantineLocked(ctx, tag, now) + return "" + } + if now.Before(quarantineUntil) { + // Pool is still in quarantine; refuse to serve any proxy. + WithRequest(ctx).WithField("proxy_tag", tag).WithField("quarantine_until", quarantineUntil.Format(time.RFC3339)). + Warn("Proxy tag pool in quarantine, no proxy served") + return "" + } + delete(r.tagQuarantine, tag) + + // Quarantine elapsed: re-enable one proxy as a recovery probe. + probe := r.leastFailedLocked(urls) + if probe != "" { + state := r.states[probe] state.disabled = false state.failures = 0 + WithRequest(ctx).WithFields(logrus.Fields{ + "proxy_tag": tag, + "proxy": MaskProxyURL(probe), + }).Warn("Proxy tag pool quarantine elapsed, probing one proxy for recovery") } } @@ -397,9 +420,18 @@ func (r *ProxyRegistry) NextByTagWithContext(ctx context.Context, tag string) st return proxyURL } + // All proxies are disabled and no probe could be selected. + r.startTagQuarantineLocked(ctx, tag, time.Now()) return "" } +// ReportFailure increments the failure counter for proxyURL. The proxy is +// disabled once the failure threshold is reached. If the owning tag pool +// becomes fully exhausted, a quarantine timer is started so that +// NextByTagWithContext will not immediately re-enable all proxies. +// +// Only proxy-network errors (ErrProxyConnect, ErrProxyAuth, ErrTimeout) should +// degrade proxy health. Callers must not call this for captcha or parser errors. func (r *ProxyRegistry) ReportFailure(ctx context.Context, proxyURL string) { proxyURL, err := NormalizeProxyURL(proxyURL) if err != nil || proxyURL == "" { @@ -421,6 +453,17 @@ func (r *ProxyRegistry) ReportFailure(ctx context.Context, proxyURL string) { "failure_count": state.failures, "proxy": MaskProxyURL(proxyURL), }).Warnf("Disabled proxy after %d failures: %s", state.failures, MaskProxyURL(proxyURL)) + + // If all proxies in every shared tag are now disabled, start quarantine. + now := time.Now() + for _, tag := range state.tags { + if r.allDisabledLocked(r.tagIndex[tag]) { + quarantineUntil := r.tagQuarantine[tag] + if quarantineUntil.IsZero() || !now.Before(quarantineUntil) { + r.startTagQuarantineLocked(ctx, tag, now) + } + } + } } } @@ -440,6 +483,11 @@ func (r *ProxyRegistry) ReportSuccess(_ context.Context, proxyURL string) { state.failures = 0 state.disabled = false + + // Clear quarantine for any tag this proxy belongs to; at least one proxy is healthy again. + for _, tag := range state.tags { + delete(r.tagQuarantine, tag) + } } func (r *ProxyRegistry) HasHealthyProxyForTag(tag string) bool { @@ -512,6 +560,33 @@ func (r *ProxyRegistry) allDisabledLocked(urls []string) bool { return true } +func (r *ProxyRegistry) startTagQuarantineLocked(ctx context.Context, tag string, now time.Time) { + quarantineUntil := now.Add(ProxyPoolQuarantineDuration) + r.tagQuarantine[tag] = quarantineUntil + WithRequest(ctx).WithFields(logrus.Fields{ + "proxy_tag": tag, + "quarantine_until": quarantineUntil.Format(time.RFC3339), + }).Warnf("Proxy tag pool %q fully exhausted, quarantined for %s", tag, ProxyPoolQuarantineDuration) +} + +// leastFailedLocked returns the URL of the disabled proxy with the lowest +// failure count, which is the cheapest probe candidate. Must hold r.mu. +func (r *ProxyRegistry) leastFailedLocked(urls []string) string { + best := "" + bestFailures := -1 + for _, proxyURL := range urls { + state, ok := r.states[proxyURL] + if !ok { + continue + } + if bestFailures < 0 || state.failures < bestFailures { + best = proxyURL + bestFailures = state.failures + } + } + return best +} + func normalizeProxyRuntime(runtime string) string { switch strings.ToLower(strings.TrimSpace(runtime)) { case ProxyRuntimeRaw: diff --git a/core/proxy_test.go b/core/proxy_test.go index fbf9f13..0494b62 100644 --- a/core/proxy_test.go +++ b/core/proxy_test.go @@ -2,6 +2,7 @@ package core import ( "context" + "errors" "fmt" "io" "log" @@ -16,6 +17,12 @@ import ( xcontext "golang.org/x/net/context" ) +type timeoutTestError struct{} + +func (timeoutTestError) Error() string { return "i/o timeout" } +func (timeoutTestError) Timeout() bool { return true } +func (timeoutTestError) Temporary() bool { return true } + func TestNormalizeProxyURL(t *testing.T) { tests := []struct { name string @@ -135,6 +142,8 @@ func TestProxyRegistryRoundRobinAndFailureRecovery(t *testing.T) { t.Fatalf("new proxy registry: %v", err) } + ctx := context.Background() + if got := registry.NextByTag("default"); got != "http://proxy1:8080" { t.Fatalf("expected first proxy1, got %s", got) } @@ -142,26 +151,35 @@ func TestProxyRegistryRoundRobinAndFailureRecovery(t *testing.T) { t.Fatalf("expected second proxy2, got %s", got) } - registry.ReportFailure(context.Background(), "http://proxy1:8080") - registry.ReportFailure(context.Background(), "http://proxy1:8080") + registry.ReportFailure(ctx, "http://proxy1:8080") + registry.ReportFailure(ctx, "http://proxy1:8080") if got := registry.NextByTag("default"); got != "http://proxy2:8080" { t.Fatalf("expected proxy2 while proxy1 disabled, got %s", got) } - registry.ReportFailure(context.Background(), "http://proxy2:8080") - registry.ReportFailure(context.Background(), "http://proxy2:8080") - if got := registry.NextByTag("default"); got != "http://proxy1:8080" { - t.Fatalf("expected tag pool reset to proxy1 after exhaustion, got %s", got) + // Exhaust pool; quarantine kicks in, no proxy served immediately. + registry.ReportFailure(ctx, "http://proxy2:8080") + registry.ReportFailure(ctx, "http://proxy2:8080") + if got := registry.NextByTag("default"); got != "" { + t.Fatalf("expected empty while pool is quarantined, got %s", got) } - registry.ReportFailure(context.Background(), "http://proxy1:8080") - registry.ReportSuccess(context.Background(), "http://proxy1:8080") - stats := registry.BuildStats() - if stats.UnhealthyCount != 0 { - t.Fatalf("expected no unhealthy proxies after success recovery, got %d", stats.UnhealthyCount) + expireProxyTagQuarantine(t, registry, "default") + + // After quarantine expiry, one probe proxy is re-enabled. + got := registry.NextByTag("default") + if got == "" { + t.Fatal("expected a probe proxy after quarantine expiry, got empty") } - if stats.HealthyCount != 2 { - t.Fatalf("expected two healthy proxies, got %d", stats.HealthyCount) + + // A success on the probe clears the quarantine and re-enables the pool. + registry.ReportSuccess(ctx, got) + stats := registry.BuildStats() + if stats.UnhealthyCount != 1 { + t.Fatalf("expected one still-disabled proxy after single probe recovery, got unhealthy_count=%d", stats.UnhealthyCount) + } + if stats.HealthyCount != 1 { + t.Fatalf("expected one healthy proxy after probe success, got %d", stats.HealthyCount) } } @@ -307,6 +325,55 @@ func TestNewRawHTTPClientSocks5hUsesProxyDNS(t *testing.T) { } } +func TestClassifyProxyNetworkError(t *testing.T) { + tests := []struct { + name string + err error + want error + }{ + {name: "timeout", err: timeoutTestError{}, want: ErrTimeout}, + {name: "connect", err: errors.New("proxyconnect tcp: connection refused"), want: ErrProxyConnect}, + {name: "auth", err: errors.New("Proxy Authentication Required 407"), want: ErrProxyAuth}, + {name: "parser", err: ErrParser, want: nil}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := classifyProxyNetworkError(tt.err) + if tt.want == nil { + if got != tt.err { + t.Fatalf("expected unchanged error, got %v", got) + } + return + } + if !errors.Is(got, tt.want) { + t.Fatalf("expected %v, got %v", tt.want, got) + } + if !errors.Is(got, tt.err) { + t.Fatalf("expected original error to be preserved, got %v", got) + } + }) + } +} + +func TestNewRawHTTPClientClassifiesProxyAuthFailure(t *testing.T) { + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusProxyAuthRequired) + })) + defer proxy.Close() + + client, err := NewRawHTTPClient(Query{ProxyURL: proxy.URL}) + if err != nil { + t.Fatalf("new raw http client: %v", err) + } + + resp, err := client.Get("http://example.com/") + DrainAndCloseResponse(resp) + if !errors.Is(err, ErrProxyAuth) { + t.Fatalf("expected proxy auth error, got %v", err) + } +} + type staticResolver struct { host string ip net.IP @@ -319,6 +386,160 @@ func (r staticResolver) Resolve(ctx xcontext.Context, name string) (xcontext.Con return ctx, nil, net.UnknownNetworkError(name) } +func expireProxyTagQuarantine(t *testing.T, registry *ProxyRegistry, tag string) { + t.Helper() + + registry.mu.Lock() + defer registry.mu.Unlock() + registry.tagQuarantine[normalizeTag(tag)] = time.Now().Add(-time.Second) +} + +func proxyHealthyCount(stats ProxyStats) int { + count := 0 + for _, entry := range stats.Entries { + if !entry.Disabled { + count++ + } + } + return count +} + +func TestReportFailureDoesNotDegradeProxyOnNonNetworkError(t *testing.T) { + registry, err := NewProxyRegistry([]ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + }, 2) + if err != nil { + t.Fatalf("new proxy registry: %v", err) + } + + rs := &ResilientSearcher{proxyRegistry: registry} + for _, err := range []error{ErrParser, ErrCaptcha} { + for range 50 { + rs.reportProxyAttempt(context.Background(), "http://proxy1:8080", err) + } + } + + stats := registry.BuildStats() + if stats.UnhealthyCount != 0 { + t.Fatalf("non-network errors must not degrade proxy health: unhealthy_count=%d", stats.UnhealthyCount) + } +} + +func TestReportFailureDegradeProxyOnNetworkError(t *testing.T) { + registry, err := NewProxyRegistry([]ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + }, 2) + if err != nil { + t.Fatalf("new proxy registry: %v", err) + } + + rs := &ResilientSearcher{proxyRegistry: registry} + rs.reportProxyAttempt(context.Background(), "http://proxy1:8080", ErrProxyConnect) + rs.reportProxyAttempt(context.Background(), "http://proxy1:8080", ErrProxyConnect) + + stats := registry.BuildStats() + if stats.UnhealthyCount != 1 { + t.Fatalf("proxy network errors must degrade proxy: unhealthy_count=%d", stats.UnhealthyCount) + } +} + +func TestPoolQuarantineAfterExhaustion(t *testing.T) { + registry, err := NewProxyRegistry([]ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + }, 1) + if err != nil { + t.Fatalf("new proxy registry: %v", err) + } + + ctx := context.Background() + + // Exhaust the single proxy; this should set a quarantine. + registry.ReportFailure(ctx, "http://proxy1:8080") + + // Verify quarantine is set and NextByTag returns empty. + got := registry.NextByTagWithContext(ctx, "default") + if got != "" { + t.Fatalf("expected empty result during quarantine, got %q", got) + } +} + +func TestPoolQuarantineExpiresAndProbesSingleProxy(t *testing.T) { + registry, err := NewProxyRegistry([]ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + {URL: "http://proxy2:8080", Tags: []string{"default"}}, + }, 1) + if err != nil { + t.Fatalf("new proxy registry: %v", err) + } + + ctx := context.Background() + + // Exhaust both proxies. + registry.ReportFailure(ctx, "http://proxy1:8080") + registry.ReportFailure(ctx, "http://proxy2:8080") + + expireProxyTagQuarantine(t, registry, "default") + + // After expiry, exactly one proxy should be re-enabled as a probe. + got := registry.NextByTagWithContext(ctx, "default") + if got == "" { + t.Fatal("expected a proxy after quarantine expiry, got empty") + } + + stats := registry.BuildStats() + if got := proxyHealthyCount(stats); got != 1 { + t.Fatalf("expected exactly 1 probe proxy re-enabled after quarantine, got %d healthy", got) + } +} + +func TestPoolQuarantineRestartsAfterFailedProbe(t *testing.T) { + registry, err := NewProxyRegistry([]ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + }, 1) + if err != nil { + t.Fatalf("new proxy registry: %v", err) + } + + ctx := context.Background() + registry.ReportFailure(ctx, "http://proxy1:8080") + expireProxyTagQuarantine(t, registry, "default") + + probe := registry.NextByTagWithContext(ctx, "default") + if probe == "" { + t.Fatal("expected probe proxy after quarantine expiry") + } + + registry.ReportFailure(ctx, probe) + if got := registry.NextByTagWithContext(ctx, "default"); got != "" { + t.Fatalf("expected renewed quarantine after failed probe, got %q", got) + } +} + +func TestReportSuccessClearsQuarantine(t *testing.T) { + registry, err := NewProxyRegistry([]ProxyEntryConfig{ + {URL: "http://proxy1:8080", Tags: []string{"default"}}, + }, 1) + if err != nil { + t.Fatalf("new proxy registry: %v", err) + } + + ctx := context.Background() + registry.ReportFailure(ctx, "http://proxy1:8080") + + // Confirm quarantine set. + if registry.NextByTagWithContext(ctx, "default") != "" { + t.Fatal("expected pool to be quarantined after exhaustion") + } + + // Recovery: success clears quarantine. + registry.ReportSuccess(ctx, "http://proxy1:8080") + + got := registry.NextByTagWithContext(ctx, "default") + if got == "" { + t.Fatal("expected proxy available after successful recovery") + } +} + func startSOCKS5TestServer(t *testing.T, host string, ip net.IP) string { t.Helper() diff --git a/core/resilient.go b/core/resilient.go index 0d890a4..e65c702 100644 --- a/core/resilient.go +++ b/core/resilient.go @@ -442,7 +442,11 @@ func (rs *ResilientSearcher) reportProxyAttempt(ctx context.Context, proxyURL st } if err != nil { - rs.proxyRegistry.ReportFailure(ctx, proxyURL) + // Only degrade proxy health for network-level errors. Captcha pages, + // parser drift, and engine errors do not indicate a faulty proxy. + if IsProxyNetworkError(err) { + rs.proxyRegistry.ReportFailure(ctx, proxyURL) + } return } diff --git a/core/server_test.go b/core/server_test.go index 1a8c39c..847d7cb 100644 --- a/core/server_test.go +++ b/core/server_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -945,7 +946,7 @@ func TestResilientRawProxyPoolRotatesOnRetry(t *testing.T) { searchFn: func(_ context.Context, q Query) ([]SearchResult, error) { attemptedProxies = append(attemptedProxies, q.ProxyURL) if q.ProxyURL == "http://bad-proxy:8080" { - return nil, errors.New("proxy failed") + return nil, fmt.Errorf("%w: dial tcp bad-proxy:8080: connection refused", ErrProxyConnect) } return []SearchResult{{Rank: 1, URL: "https://example.com/success", Title: "ok"}}, nil },