Proxy policy wiring and fail-closed execution

- replace proxy wiring with global and per-engine tag policies
- split stats endpoints and lock fail-closed proxy behavior with tests

core/resilient.go

@@ -2,7 +2,9 @@ package core
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/sirupsen/logrus"
@@ -13,8 +15,19 @@ type ResilientSearcher struct {
 	engines   []SearchEngine
 	cbManager *CircuitBreakerManager
 	retryCfg  RetryConfig
-	proxyCfg  ProxyConfig
-	proxyPool *ProxyPool
+
+	proxyRuntime  string
+	proxyCfg      ProxyConfig
+	proxyRegistry *ProxyRegistry
+	proxyDefaults ProxyPolicy
+
+	effectivePolicies map[string]ProxyPolicy
+}
+
+type ProxyExecutionMeta struct {
+	Mode string `json:"mode"`
+	Tag  string `json:"tag,omitempty"`
+	Used string `json:"used"`
 }
 
 type ResilientConfig struct {
@@ -34,103 +47,101 @@ func DefaultResilientConfig() ResilientConfig {
 func NewResilientSearcher(engines []SearchEngine, cfg ResilientConfig) *ResilientSearcher {
 	proxyCfg, err := NormalizeProxyConfig(cfg.Proxy)
 	if err != nil {
-		logrus.Errorf("Invalid proxy config, disabling proxy support: %v", err)
+		logrus.Errorf("Invalid proxy config, using defaults: %v", err)
 		proxyCfg = DefaultProxyConfig()
+		proxyCfg, _ = NormalizeProxyConfig(proxyCfg)
 	}
 
 	rs := &ResilientSearcher{
-		engines:   engines,
-		cbManager: NewCircuitBreakerManager(cfg.CircuitBreaker),
-		retryCfg:  cfg.Retry,
-		proxyCfg:  proxyCfg,
+		engines:           engines,
+		cbManager:         NewCircuitBreakerManager(cfg.CircuitBreaker),
+		retryCfg:          cfg.Retry,
+		proxyRuntime:      proxyCfg.Runtime,
+		proxyCfg:          proxyCfg,
+		proxyRegistry:     proxyCfg.Registry,
+		proxyDefaults:     ResolveEffectiveProxyPolicy(proxyCfg.Proxies.Global, ""),
+		effectivePolicies: make(map[string]ProxyPolicy, len(engines)),
 	}
 
-	if len(proxyCfg.PoolURLs) > 0 {
-		pool, err := NewProxyPool(proxyCfg.PoolURLs, proxyCfg.PoolFailureThreshold)
-		if err != nil {
-			logrus.Errorf("Invalid proxy pool config, disabling proxy rotation: %v", err)
-		} else {
-			rs.proxyPool = pool
-			if proxyCfg.Runtime == ProxyRuntimeBrowser {
-				logrus.Warn("Proxy pool configured in browser runtime; pool remains observability-only until browser proxy rotation is implemented")
-			} else {
-				logrus.Infof("Proxy rotation enabled with %d proxies", pool.Size())
-			}
-		}
+	for _, engine := range engines {
+		engineName := normalizeEngineName(engine.Name())
+		override := proxyCfg.EnginePolicies[engineName]
+		effective := ResolveEffectiveProxyPolicy(proxyCfg.Proxies.Global, override)
+		rs.effectivePolicies[engineName] = effective
 	}
 
 	return rs
 }
 
 // SearchPrimary keeps dedicated endpoints engine-pure (no fallback).
-func (rs *ResilientSearcher) SearchPrimary(primaryEngine SearchEngine, q Query) ([]SearchResult, string, error) {
-	results, err := rs.searchWithProtection(primaryEngine, q)
+func (rs *ResilientSearcher) SearchPrimary(primaryEngine SearchEngine, q Query) ([]SearchResult, string, ProxyExecutionMeta, error) {
+	results, proxyMeta, err := rs.searchWithProtection(primaryEngine, q, false)
 	if err != nil {
-		return nil, primaryEngine.Name(), err
+		return nil, primaryEngine.Name(), proxyMeta, err
 	}
-	return results, primaryEngine.Name(), nil
+	return results, primaryEngine.Name(), proxyMeta, nil
 }
 
 // SearchWithFallback retries primary and then tries other initialized engines.
-func (rs *ResilientSearcher) SearchWithFallback(primaryEngine SearchEngine, q Query) ([]SearchResult, string, error) {
-	results, err := rs.searchWithProtection(primaryEngine, q)
-	if err == nil {
-		return results, primaryEngine.Name(), nil
-	}
-
-	logrus.Warnf("[Resilient] Primary engine %s failed: %s. Trying fallback engines...", primaryEngine.Name(), err)
-	for _, fallbackEngine := range rs.engines {
-		if fallbackEngine.Name() == primaryEngine.Name() || !fallbackEngine.IsInitialized() {
-			continue
-		}
-
-		results, err := rs.searchWithProtection(fallbackEngine, q)
-		if err == nil {
-			logrus.Infof("[Resilient] Fallback to %s succeeded with %d results", fallbackEngine.Name(), len(results))
-			return results, fallbackEngine.Name(), nil
-		}
-		logrus.Warnf("[Resilient] Fallback engine %s also failed: %s", fallbackEngine.Name(), err)
-	}
-
-	return nil, primaryEngine.Name(), ErrAllEnginesFailed
+func (rs *ResilientSearcher) SearchWithFallback(primaryEngine SearchEngine, q Query) ([]SearchResult, string, ProxyExecutionMeta, error) {
+	return rs.searchWithFallback(primaryEngine, q, false)
 }
 
-func (rs *ResilientSearcher) SearchImagePrimary(primaryEngine SearchEngine, q Query) ([]SearchResult, string, error) {
-	results, err := rs.searchImageWithProtection(primaryEngine, q)
+func (rs *ResilientSearcher) SearchImagePrimary(primaryEngine SearchEngine, q Query) ([]SearchResult, string, ProxyExecutionMeta, error) {
+	results, proxyMeta, err := rs.searchWithProtection(primaryEngine, q, true)
 	if err != nil {
-		return nil, primaryEngine.Name(), err
+		return nil, primaryEngine.Name(), proxyMeta, err
 	}
-	return results, primaryEngine.Name(), nil
+	return results, primaryEngine.Name(), proxyMeta, nil
 }
 
-func (rs *ResilientSearcher) SearchImageWithFallback(primaryEngine SearchEngine, q Query) ([]SearchResult, string, error) {
-	results, err := rs.searchImageWithProtection(primaryEngine, q)
+func (rs *ResilientSearcher) SearchImageWithFallback(primaryEngine SearchEngine, q Query) ([]SearchResult, string, ProxyExecutionMeta, error) {
+	return rs.searchWithFallback(primaryEngine, q, true)
+}
+
+func (rs *ResilientSearcher) searchWithFallback(primaryEngine SearchEngine, q Query, isImage bool) ([]SearchResult, string, ProxyExecutionMeta, error) {
+	results, proxyMeta, err := rs.searchWithProtection(primaryEngine, q, isImage)
 	if err == nil {
-		return results, primaryEngine.Name(), nil
+		return results, primaryEngine.Name(), proxyMeta, nil
+	}
+	if errors.Is(err, ErrProxyUnavailable) {
+		logrus.Warnf("[Resilient] Primary engine %s proxy policy failed closed: %s", primaryEngine.Name(), err)
+		return nil, primaryEngine.Name(), proxyMeta, err
 	}
 
-	logrus.Warnf("[Resilient] Primary engine %s image search failed: %s. Trying fallback engines...", primaryEngine.Name(), err)
+	action := "failed"
+	successMessage := "Fallback to %s succeeded with %d results"
+	if isImage {
+		action = "image search failed"
+		successMessage = "Image fallback to %s succeeded with %d results"
+	}
+
+	logrus.Warnf("[Resilient] Primary engine %s %s: %s. Trying fallback engines...", primaryEngine.Name(), action, err)
 	for _, fallbackEngine := range rs.engines {
 		if fallbackEngine.Name() == primaryEngine.Name() || !fallbackEngine.IsInitialized() {
 			continue
 		}
 
-		results, err := rs.searchImageWithProtection(fallbackEngine, q)
-		if err == nil {
-			logrus.Infof("[Resilient] Image fallback to %s succeeded with %d results", fallbackEngine.Name(), len(results))
-			return results, fallbackEngine.Name(), nil
+		results, fallbackMeta, fallbackErr := rs.searchWithProtection(fallbackEngine, q, isImage)
+		if fallbackErr == nil {
+			logrus.Infof("[Resilient] "+successMessage, fallbackEngine.Name(), len(results))
+			return results, fallbackEngine.Name(), fallbackMeta, nil
 		}
+		logrus.Warnf("[Resilient] Fallback engine %s also failed: %s", fallbackEngine.Name(), fallbackErr)
 	}
 
-	return nil, primaryEngine.Name(), ErrAllEnginesFailed
+	return nil, primaryEngine.Name(), proxyMeta, ErrAllEnginesFailed
 }
 
-func (rs *ResilientSearcher) searchWithProtection(engine SearchEngine, q Query) ([]SearchResult, error) {
+func (rs *ResilientSearcher) searchWithProtection(engine SearchEngine, q Query, isImage bool) ([]SearchResult, ProxyExecutionMeta, error) {
 	cb := rs.cbManager.Get(engine.Name())
 	if !cb.AllowRequest() {
-		return nil, ErrCircuitOpen
+		return nil, ProxyExecutionMeta{}, ErrCircuitOpen
 	}
 
+	policy := rs.effectivePolicyForEngine(engine.Name())
+	attemptMeta := rs.baseProxyMeta(policy)
+
 	result := RetryableSearch(rs.retryCfg, engine.Name(), func() ([]SearchResult, error) {
 		limiter := engine.GetRateLimiter()
 		if limiter != nil {
@@ -139,48 +150,51 @@ func (rs *ResilientSearcher) searchWithProtection(engine SearchEngine, q Query)
 			}
 		}
 
-		attemptQuery, attemptProxy, reportToPool := rs.prepareAttemptQuery(q)
-		results, err := engine.Search(attemptQuery)
-		rs.reportProxyAttempt(attemptProxy, reportToPool, err)
-		return results, err
-	})
+		attemptQuery := q
+		proxyURL := ""
+		reportToRegistry := false
+		attemptMeta = rs.baseProxyMeta(policy)
 
-	if result.Err != nil {
-		cb.RecordFailure()
-		return nil, result.Err
-	}
-
-	cb.RecordSuccess()
-	return result.Results, nil
-}
-
-func (rs *ResilientSearcher) searchImageWithProtection(engine SearchEngine, q Query) ([]SearchResult, error) {
-	cb := rs.cbManager.Get(engine.Name())
-	if !cb.AllowRequest() {
-		return nil, ErrCircuitOpen
-	}
-
-	result := RetryableSearch(rs.retryCfg, engine.Name(), func() ([]SearchResult, error) {
-		limiter := engine.GetRateLimiter()
-		if limiter != nil {
-			if err := limiter.Wait(context.Background()); err != nil {
-				return nil, err
+		switch policy.Mode {
+		case ProxyModeOff:
+			attemptQuery.ProxyURL = ""
+			attemptMeta.Used = "direct"
+		case ProxyModeTagPool:
+			proxyURL = rs.selectProxyForPolicy(policy)
+			if proxyURL == "" {
+				return nil, fmt.Errorf("%w: no healthy proxy available for tag %q", ErrProxyUnavailable, policy.Tag)
 			}
+			attemptQuery.ProxyURL = proxyURL
+			reportToRegistry = policy.Tag != ""
+			attemptMeta.Used = MaskProxyURL(proxyURL)
+		}
+
+		var (
+			results []SearchResult
+			err     error
+		)
+		if isImage {
+			results, err = engine.SearchImage(attemptQuery)
+		} else {
+			results, err = engine.Search(attemptQuery)
+		}
+
+		if reportToRegistry {
+			rs.reportProxyAttempt(proxyURL, err)
 		}
 
-		attemptQuery, attemptProxy, reportToPool := rs.prepareAttemptQuery(q)
-		results, err := engine.SearchImage(attemptQuery)
-		rs.reportProxyAttempt(attemptProxy, reportToPool, err)
 		return results, err
 	})
 
 	if result.Err != nil {
-		cb.RecordFailure()
-		return nil, result.Err
+		if !errors.Is(result.Err, ErrProxyUnavailable) {
+			cb.RecordFailure()
+		}
+		return nil, attemptMeta, result.Err
 	}
 
 	cb.RecordSuccess()
-	return result.Results, nil
+	return result.Results, attemptMeta, nil
 }
 
 // SearchAllParallel applies retry/circuit protections per engine for mega search.
@@ -202,7 +216,7 @@ func (rs *ResilientSearcher) SearchAllParallel(q Query, engines []SearchEngine)
 		go func(eng SearchEngine) {
 			defer wg.Done()
 
-			results, err := rs.searchWithProtection(eng, q)
+			results, _, err := rs.searchWithProtection(eng, q, false)
 			if err != nil {
 				return
 			}
@@ -240,7 +254,7 @@ func (rs *ResilientSearcher) SearchAllImageParallel(q Query, engines []SearchEng
 		go func(eng SearchEngine) {
 			defer wg.Done()
 
-			results, err := rs.searchImageWithProtection(eng, q)
+			results, _, err := rs.searchWithProtection(eng, q, true)
 			if err != nil {
 				return
 			}
@@ -264,88 +278,144 @@ func (rs *ResilientSearcher) GetCircuitBreakerStats() []map[string]interface{} {
 	return rs.cbManager.AllStats()
 }
 
-func (rs *ResilientSearcher) GetProxyPool() *ProxyPool {
-	return rs.proxyPool
-}
-
-func (rs *ResilientSearcher) GetProxyStats() map[string]interface{} {
-	mode, source, rotationActive := rs.proxyMode()
-
-	stats := map[string]interface{}{
-		"mode":            mode,
-		"runtime":         rs.proxyCfg.Runtime,
-		"source":          source,
-		"rotation_active": rotationActive,
+func (rs *ResilientSearcher) GetProxyStats() ProxyStats {
+	stats := ProxyStats{
+		ConfiguredCount: 0,
+		HealthyCount:    0,
+		UnhealthyCount:  0,
+		Tags:            map[string]ProxyTagSummary{},
+		Entries:         []ProxyStatsEntry{},
 	}
 
-	if rs.proxyPool != nil {
-		poolStats := rs.proxyPool.Stats()
-		stats["pool"] = map[string]interface{}{
-			"failure_threshold": poolStats.FailureThreshold,
-			"total":             poolStats.Total,
-			"active":            poolStats.Active,
-			"disabled":          poolStats.Disabled,
+	if rs.proxyRegistry != nil {
+		stats = rs.proxyRegistry.BuildStats()
+	}
+
+	engines := map[string]ProxyEngineStats{}
+	for _, engine := range rs.engines {
+		engineName := normalizeEngineName(engine.Name())
+		policy := rs.effectivePolicyForEngine(engineName)
+		engineStats := ProxyEngineStats{}
+
+		switch policy.Mode {
+		case ProxyModeOff:
+			engineStats.SelectedProxy = "direct"
+		case ProxyModeTagPool:
+			engineStats.Tag = policy.Tag
+			if global := strings.TrimSpace(rs.proxyCfg.Proxies.Global); global != "" {
+				engineStats.SelectedProxy = MaskProxyURL(global)
+			} else {
+				engineStats.SelectedProxy = "pooled"
+			}
 		}
+
+		engines[engineName] = engineStats
+	}
+	if len(engines) > 0 {
+		stats.Engines = engines
 	}
 
 	return stats
 }
 
-func (rs *ResilientSearcher) prepareAttemptQuery(q Query) (Query, string, bool) {
-	attemptQuery := q
-	if rs.proxyCfg.Runtime != ProxyRuntimeRaw {
-		return attemptQuery, "", false
+func (rs *ResilientSearcher) ResolveMegaProxyMeta(engines []SearchEngine) ProxyExecutionMeta {
+	if len(engines) == 0 {
+		return ProxyExecutionMeta{Mode: ProxyModeOff, Used: "direct"}
 	}
 
-	if rs.proxyPool != nil {
-		proxyURL := rs.proxyPool.Next()
-		if proxyURL != "" {
-			attemptQuery.ProxyURL = proxyURL
-			return attemptQuery, proxyURL, true
+	allOff := true
+	proxiedTags := map[string]struct{}{}
+	hasOff := false
+
+	for _, engine := range engines {
+		policy := rs.effectivePolicyForEngine(engine.Name())
+		if policy.Mode == ProxyModeOff {
+			hasOff = true
+			continue
+		}
+
+		allOff = false
+		if policy.Tag != "" {
+			proxiedTags[policy.Tag] = struct{}{}
 		}
 	}
 
-	attemptQuery.ProxyURL = rs.proxyCfg.StaticURL
-	return attemptQuery, "", false
+	if allOff {
+		return ProxyExecutionMeta{Mode: ProxyModeOff, Used: "direct"}
+	}
+
+	meta := ProxyExecutionMeta{Mode: ProxyModeTagPool}
+	if len(proxiedTags) == 1 {
+		for tag := range proxiedTags {
+			meta.Tag = tag
+		}
+	}
+
+	if global := strings.TrimSpace(rs.proxyCfg.Proxies.Global); global != "" && !hasOff {
+		meta.Used = MaskProxyURL(global)
+		return meta
+	}
+
+	if rs.proxyRuntime == ProxyRuntimeRaw {
+		meta.Used = "multiple"
+		if hasOff {
+			meta.Used = "mixed"
+		}
+		return meta
+	}
+	meta.Used = "multiple"
+	if hasOff {
+		meta.Used = "mixed"
+	}
+	return meta
 }
 
-func (rs *ResilientSearcher) reportProxyAttempt(proxyURL string, reportToPool bool, err error) {
-	if !reportToPool || rs.proxyPool == nil || proxyURL == "" {
+func (rs *ResilientSearcher) baseProxyMeta(policy ProxyPolicy) ProxyExecutionMeta {
+	meta := ProxyExecutionMeta{Mode: policy.Mode}
+	if policy.Mode == ProxyModeTagPool {
+		meta.Tag = policy.Tag
+		return meta
+	}
+	meta.Used = "direct"
+	return meta
+}
+
+func (rs *ResilientSearcher) effectivePolicyForEngine(engineName string) ProxyPolicy {
+	engineName = normalizeEngineName(engineName)
+	if policy, ok := rs.effectivePolicies[engineName]; ok {
+		return policy
+	}
+	return rs.proxyDefaults
+}
+
+func (rs *ResilientSearcher) selectProxyForTag(tag string) string {
+	if rs.proxyRegistry == nil {
+		return ""
+	}
+	return rs.proxyRegistry.NextByTag(tag)
+}
+
+func (rs *ResilientSearcher) reportProxyAttempt(proxyURL string, err error) {
+	if rs.proxyRegistry == nil || proxyURL == "" {
 		return
 	}
 
 	if err != nil {
-		rs.proxyPool.ReportFailure(proxyURL)
+		rs.proxyRegistry.ReportFailure(proxyURL)
 		return
 	}
 
-	rs.proxyPool.ReportSuccess(proxyURL)
+	rs.proxyRegistry.ReportSuccess(proxyURL)
 }
 
-func (rs *ResilientSearcher) proxyMode() (string, string, bool) {
-	hasStatic := rs.proxyCfg.StaticURL != ""
-	hasPool := rs.proxyPool != nil
-
-	switch rs.proxyCfg.Runtime {
-	case ProxyRuntimeRaw:
-		switch {
-		case hasPool:
-			return ProxyModePool, "proxy_pool.urls", true
-		case hasStatic:
-			return ProxyModeStatic, "app.proxy", false
-		default:
-			return ProxyModeDisabled, "none", false
-		}
-	default:
-		switch {
-		case hasStatic:
-			return ProxyModeStatic, "app.proxy", false
-		case hasPool:
-			return ProxyModePool, "proxy_pool.urls", false
-		default:
-			return ProxyModeDisabled, "none", false
-		}
+func (rs *ResilientSearcher) selectProxyForPolicy(policy ProxyPolicy) string {
+	if policy.Mode != ProxyModeTagPool {
+		return ""
 	}
+	if global := strings.TrimSpace(rs.proxyCfg.Proxies.Global); global != "" {
+		return global
+	}
+	return rs.selectProxyForTag(policy.Tag)
 }
 
 var ErrAllEnginesFailed = fmt.Errorf("all search engines failed")