fix(proxy): isolate proxy health from captcha/parser errors, replace re-enable thrash with quarantine

2026-08-05 16:53:54 +08:00 · 2026-04-24 04:33:41 +03:00
parent e4a84d8914
commit 0ccb13c36e
6 changed files with 403 additions and 21 deletions
--- 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
--- 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
+}
--- 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:
--- 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()

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

--- 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
 		},