From 5dcfd9fd3971f57cad39cf6831fa858ce97a9331 Mon Sep 17 00:00:00 2001 From: RuoJi6 Date: Sun, 1 Nov 2026 22:23:35 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BB=A3=E7=90=86=E6=B1=A0=E6=9C=AA?= =?UTF-8?q?=E6=8E=A2=E6=B5=8B=E4=B8=8D=E6=9A=82=E5=81=9C=E3=80=81=E5=88=87?= =?UTF-8?q?=E6=8D=A2=E6=88=90=E5=8A=9F=E7=BB=9F=E4=B8=80=E9=97=A8=E6=8E=A7?= =?UTF-8?q?=E6=81=A2=E5=A4=8D=E3=80=81=E6=B1=A0=E6=A8=A1=E5=BC=8F=E4=BF=9D?= =?UTF-8?q?=E7=95=99=E9=87=8D=E8=BF=9E=E6=AC=A1=E6=95=B0=E4=B8=8E=E8=80=97?= =?UTF-8?q?=E5=B0=BD=E3=80=81=E7=A6=81=E7=94=A8=E5=BD=93=E5=89=8D=E6=88=90?= =?UTF-8?q?=E5=91=98=E5=8D=B3=E5=88=87=E6=8D=A2=E3=80=81=E8=A1=A5=E5=85=B3?= =?UTF-8?q?=E9=94=AE=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/proxy_autoresume.go | 18 +-- server/proxy_pool.go | 301 ++++++++++++++++++++++++++----------- server/proxy_pool_test.go | 216 ++++++++++++++++++++++++++ 3 files changed, 434 insertions(+), 101 deletions(-) diff --git a/server/proxy_autoresume.go b/server/proxy_autoresume.go index a36939a..600af5a 100644 --- a/server/proxy_autoresume.go +++ b/server/proxy_autoresume.go @@ -258,16 +258,14 @@ func (s *Server) proxyReconnectTick() { return } if s.m.proxyPool != nil && len(s.m.proxyPoolEntries()) > 0 { - active := s.m.proxyPoolActiveID() - // Pool-managed reconnects are driven by a confirmed pool probe. Do not - // run the per-task reconnect loop against an entry that merely still says - // healthy: that would resume exhausted tasks without a fresh probe and - // reset their attempt budget on the next failure. The pool tick also - // handles a missing/cooling active member and can switch to another - // confirmed healthy member before calling the guarded resume path. - _ = active - s.proxyPoolTick() - return + // A pool tick may confirm the active route or switch to another available + // member. If it cannot confirm either, retain the original #106 per-task + // attempt/backoff/exhaustion loop (especially for a one-member legacy + // pool). + outcome := s.proxyPoolTick() + if outcome.confirmed || outcome.switched { + return + } } now := time.Now().Unix() max, interval := s.proxyReconnectConfig() diff --git a/server/proxy_pool.go b/server/proxy_pool.go index de06f25..6a9f27e 100644 --- a/server/proxy_pool.go +++ b/server/proxy_pool.go @@ -103,7 +103,9 @@ func ParseProxyBatch(raw string) ([]ProxyPoolEntry, []ProxyImportError) { if len(parts) > 2 { remark = strings.TrimSpace(parts[2]) } - out = append(out, ProxyPoolEntry{ID: fmt.Sprintf("proxy-%d-%d", time.Now().UnixNano(), n+1), URL: rawURL, Name: name, Remark: remark, Enabled: true, Health: "unhealthy-cooling"}) + // New entries have not failed yet. Keep them available until a probe + // establishes a consecutive-failure streak. + out = append(out, ProxyPoolEntry{ID: fmt.Sprintf("proxy-%d-%d", time.Now().UnixNano(), n+1), URL: rawURL, Name: name, Remark: remark, Enabled: true, Health: "unknown"}) } return out, errs } @@ -135,6 +137,11 @@ type proxyPoolState struct { lastHealth time.Time } +type proxyPoolTickOutcome struct { + confirmed bool + switched bool +} + func (m *Manager) initProxyPool() { m.proxyPool = &proxyPoolState{} persistedPool, poolSettingPresent, _ := m.pg.GetSetting(settingProxyPool) @@ -160,6 +167,11 @@ func (m *Manager) initProxyPool() { _ = m.pg.SetSetting(settingGlobalProxy, "") _ = m.pg.SetSetting(settingProxyPoolActive, "") } else if len(m.proxyPool.entries) > 0 { + for i := range m.proxyPool.entries { + if strings.TrimSpace(m.proxyPool.entries[i].RawURL) != "" && m.proxyPool.entries[i].Health == "" { + m.proxyPool.entries[i].Health = "unknown" + } + } chosen := "" for _, e := range m.proxyPool.entries { if e.ID == m.proxyPool.active && e.Enabled { @@ -169,7 +181,7 @@ func (m *Manager) initProxyPool() { } if chosen == "" { for _, e := range m.proxyPool.entries { - if e.Enabled && e.Health == "healthy" { + if e.Enabled && e.RawURL != "" && e.Health != "unhealthy-cooling" { chosen = e.ID break } @@ -260,7 +272,7 @@ func (m *Manager) setProxyPoolEntries(entries []proxyPoolEntry, active string) e } if active == "" { for _, e := range entries { - if e.Enabled && e.Health == "healthy" { + if e.Enabled && e.RawURL != "" && e.Health != "unhealthy-cooling" { return m.applyActiveProxyLocked(e.ID, "reconnect") } } @@ -289,7 +301,7 @@ func (m *Manager) applyActiveProxyLocked(id, reason string) error { func (m *Manager) applyActiveProxyLockedWithHealth(id, reason string, allowUnhealthy bool) error { var next string for _, e := range m.proxyPool.entries { - if e.ID == id && e.Enabled && (allowUnhealthy || e.Health == "healthy") { + if e.ID == id && e.Enabled && e.RawURL != "" && (allowUnhealthy || e.Health != "unhealthy-cooling") { next = e.RawURL break } @@ -338,13 +350,13 @@ func (m *Manager) selectHealthyLocked(exclude string) string { } pool := make([]string, 0) for _, e := range m.proxyPool.entries { - if e.Enabled && e.Health == "healthy" && e.ID != exclude { + if e.Enabled && e.Health != "unhealthy-cooling" && e.ID != exclude { pool = append(pool, e.ID) } } if len(pool) == 0 && exclude != "" { for _, e := range m.proxyPool.entries { - if e.Enabled && e.Health == "healthy" { + if e.Enabled && e.Health != "unhealthy-cooling" { pool = append(pool, e.ID) } } @@ -364,7 +376,7 @@ func (m *Manager) selectHealthyLocked(exclude string) string { } for n := 0; n < len(m.proxyPool.entries); n++ { e := m.proxyPool.entries[(start+n)%len(m.proxyPool.entries)] - if e.Enabled && e.Health == "healthy" && e.ID != exclude { + if e.Enabled && e.Health != "unhealthy-cooling" && e.ID != exclude { return e.ID } } @@ -507,22 +519,91 @@ func (s *Server) runProxyPoolSupervisor(ctx context.Context) { } } -func (s *Server) proxyPoolTick() { +func (s *Server) proxyPoolTick() proxyPoolTickOutcome { if s == nil { - return + return proxyPoolTickOutcome{} } s.proxyPoolTickMu.Lock() defer s.proxyPoolTickMu.Unlock() - s.proxyPoolTickLocked(false) + return s.proxyPoolTickLocked(false) } -func (s *Server) proxyPoolTickLocked(forceHealth bool) { +func (s *Server) proxyPoolTickLocked(forceHealth bool) proxyPoolTickOutcome { + var outcome proxyPoolTickOutcome if s.m == nil || s.m.proxyPool == nil { - return + return outcome } now := time.Now() + threshold, cooldown, healthInterval := s.m.proxyHealthConfig() + s.m.proxyPool.mu.Lock() + healthDue := s.m.proxyPool.lastHealth.IsZero() || now.Sub(s.m.proxyPool.lastHealth) >= healthInterval + if !healthDue { + // A cooling member becomes probe-eligible at its own cooldown deadline; + // do not make it wait for the next full health-check interval. + for _, e := range s.m.proxyPool.entries { + if e.Enabled && e.Health == "unhealthy-cooling" && (e.LastCheckedAt.IsZero() || now.Sub(e.LastCheckedAt) >= cooldown) { + healthDue = true + break + } + } + } + if healthDue { + s.m.proxyPool.lastHealth = now + } + s.m.proxyPool.mu.Unlock() + if healthDue || forceHealth { + entries := s.m.proxyPoolEntries() + sem := s.proxyPoolProbeSem + if sem == nil { + sem = make(chan struct{}, proxyPoolMaxProbeConcurrency) + } + activeAtProbe := s.m.proxyPoolActiveID() + probeResults := make(map[string]bool, len(entries)) + var probeMu sync.Mutex + var wg sync.WaitGroup + for _, e := range entries { + if !e.Enabled { + continue + } + if !forceHealth && e.Health == "unhealthy-cooling" && e.ConsecutiveFailures >= threshold && !e.LastCheckedAt.IsZero() && now.Sub(e.LastCheckedAt) < cooldown { + continue + } + e := e + wg.Add(1) + go func() { + defer wg.Done() + select { + case sem <- struct{}{}: + case <-serverContext(s).Done(): + return + } + defer func() { <-sem }() + res := s.runProxyProbe(serverContext(s), e.RawURL, "", defaultEgressProbes, proxyPoolProbeTimeout) + s.m.markProxyProbe(e.ID, res) + probeMu.Lock() + probeResults[e.ID] = res.OK + probeMu.Unlock() + }() + } + wg.Wait() + probeMu.Lock() + confirmed := probeResults[activeAtProbe] + probeMu.Unlock() + if confirmed { + outcome.confirmed = true + } + if s.reconcileProxyPoolAfterProbe(confirmed) { + outcome.switched = true + } + } + + // Health reconciliation runs before timed rotation. This matters when an + // import starts a fresh probe round: tasks are not paused or rerouted until + // the new members have been checked. A successful recovery switch already + // selected the best available member for this round, so avoid immediately + // rotating away from it. timed, rotateInterval, _, _ := s.m.proxyRotateConfig() - if timed { + if timed && !outcome.switched { s.m.proxyPool.mu.Lock() due := s.m.proxyPool.lastTimed.IsZero() || now.Sub(s.m.proxyPool.lastTimed) >= rotateInterval if due { @@ -530,76 +611,39 @@ func (s *Server) proxyPoolTickLocked(forceHealth bool) { } s.m.proxyPool.mu.Unlock() if due { - _ = s.m.rotateProxy("timed") - } - } - threshold, cooldown, healthInterval := s.m.proxyHealthConfig() - s.m.proxyPool.mu.Lock() - healthDue := s.m.proxyPool.lastHealth.IsZero() || now.Sub(s.m.proxyPool.lastHealth) >= healthInterval - if healthDue { - s.m.proxyPool.lastHealth = now - } - s.m.proxyPool.mu.Unlock() - if !healthDue && !forceHealth { - s.reconcileProxyPoolAfterProbe(false) - return - } - entries := s.m.proxyPoolEntries() - sem := s.proxyPoolProbeSem - if sem == nil { - sem = make(chan struct{}, proxyPoolMaxProbeConcurrency) - } - activeAtProbe := s.m.proxyPoolActiveID() - probeResults := make(map[string]bool, len(entries)) - var probeMu sync.Mutex - var wg sync.WaitGroup - for _, e := range entries { - if !e.Enabled { - continue - } - if !forceHealth && e.Health == "unhealthy-cooling" && e.ConsecutiveFailures >= threshold && !e.LastCheckedAt.IsZero() && now.Sub(e.LastCheckedAt) < cooldown { - continue - } - e := e - wg.Add(1) - go func() { - defer wg.Done() - select { - case sem <- struct{}{}: - case <-s.ctx.Done(): - return + before := s.m.proxyPoolActiveID() + if s.m.rotateProxy("timed") && before != s.m.proxyPoolActiveID() { + outcome.switched = true + s.proxyReconnectSuccess() } - defer func() { <-sem }() - res := s.runProxyProbe(s.ctx, e.RawURL, "", defaultEgressProbes, proxyPoolProbeTimeout) - s.m.markProxyProbe(e.ID, res) - probeMu.Lock() - probeResults[e.ID] = res.OK - probeMu.Unlock() - }() + } } - wg.Wait() - probeMu.Lock() - confirmed := probeResults[activeAtProbe] - probeMu.Unlock() - s.reconcileProxyPoolAfterProbe(confirmed) + return outcome } -func (s *Server) reconcileProxyPoolAfterProbe(activeConfirmed bool) { +func serverContext(s *Server) context.Context { + if s != nil && s.ctx != nil { + return s.ctx + } + return context.Background() +} + +func (s *Server) reconcileProxyPoolAfterProbe(activeConfirmed bool) bool { entries := s.m.proxyPoolEntries() activeID := s.m.proxyPoolActiveID() - activeHealthy := false + activeAvailable := false for _, e := range entries { - if e.ID == activeID && e.Enabled && e.Health == "healthy" { - activeHealthy = true + if e.ID == activeID && e.Enabled && e.RawURL != "" && e.Health != "unhealthy-cooling" { + activeAvailable = true break } } - if activeHealthy && activeConfirmed { + if activeAvailable && activeConfirmed { s.proxyReconnectSuccess() - return + return false } - if activeHealthy { - return + if activeAvailable { + return false } s.m.proxyPool.mu.Lock() id := s.m.selectHealthyLocked(activeID) @@ -610,15 +654,25 @@ func (s *Server) reconcileProxyPoolAfterProbe(activeConfirmed bool) { s.m.proxyPool.mu.Unlock() if err == nil && id != "" { s.proxyReconnectSuccess() - return + return true } - if err != nil { - s.pauseAllProxyTasks(err) - return + allCooling := false + enabled := 0 + for _, e := range entries { + if e.Enabled && e.RawURL != "" { + enabled++ + if e.Health == "unhealthy-cooling" { + allCooling = true + } else { + allCooling = false + break + } + } } - if id == "" && (activeID != "" || strings.TrimSpace(s.m.GlobalProxy()) != "") { + if id == "" && enabled > 0 && allCooling && (activeID != "" || strings.TrimSpace(s.m.GlobalProxy()) != "") { s.pauseAllProxyTasks(errors.New("代理池暂无健康成员")) } + return false } func (m *Manager) activateProxyID(id, reason string) error { @@ -641,15 +695,13 @@ func (s *Server) handleProxyPoolFailure(cause error) bool { if cur == "" { return false } - res := s.runProxyProbe(s.ctx, cur, "", defaultEgressProbes, proxyPoolProbeTimeout) + res := s.runProxyProbe(serverContext(s), cur, "", defaultEgressProbes, proxyPoolProbeTimeout) if res.OK { return false } _, _, onFailure, _ := s.m.proxyRotateConfig() s.proxyPoolTickMu.Lock() - defer s.proxyPoolTickMu.Unlock() s.m.proxyPool.mu.Lock() - defer s.m.proxyPool.mu.Unlock() active := s.m.proxyPool.active currentChanged := s.m.GlobalProxy() != cur for _, e := range s.m.proxyPool.entries { @@ -659,11 +711,16 @@ func (s *Server) handleProxyPoolFailure(cause error) bool { } } if currentChanged { + s.m.proxyPool.mu.Unlock() + s.proxyPoolTickMu.Unlock() return true } if onFailure { if id := s.m.selectHealthyLocked(active); id != "" && id != active { if err := s.m.applyActiveProxyLocked(id, "failure"); err == nil { + s.m.proxyPool.mu.Unlock() + s.proxyPoolTickMu.Unlock() + s.proxyReconnectSuccess() return true } } @@ -672,13 +729,15 @@ func (s *Server) handleProxyPoolFailure(cause error) bool { case s.proxyPoolWake <- struct{}{}: default: } + s.m.proxyPool.mu.Unlock() + s.proxyPoolTickMu.Unlock() return false } func (s *Server) pauseAllProxyTasks(cause error) { for _, t := range s.m.List() { st := t.lifecycleSnapshot() - if isTerminalStatus(st.Status) || st.Paused || s.engine.IsDeleting(t.ID) { + if isTerminalStatus(st.Status) || st.Paused || (s.engine != nil && s.engine.IsDeleting(t.ID)) { continue } s.pauseProxyTask(t.ID, cause) @@ -772,11 +831,10 @@ func (s *Server) importProxyPool(w http.ResponseWriter, r *http.Request) { writeErr(w, 400, err.Error()) return } - select { - case s.proxyPoolWake <- struct{}{}: - default: - } } + // Probe imported/unchecked entries before any pool-wide pause decision. The + // task state is untouched until this round has completed. + s.proxyPoolTick() entriesOut, activeOut, logs := s.m.ProxyPoolSnapshot() writeJSON(w, 200, map[string]any{"entries": entriesOut, "active_id": activeOut, "switch_log": logs, "errors": errs}) } @@ -839,6 +897,8 @@ func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) { return } needPause := false + switched := false + routed := false s.m.proxyPool.mu.Lock() for i := range s.m.proxyPool.entries { e := &s.m.proxyPool.entries[i] @@ -884,14 +944,53 @@ func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) { writeErr(w, 400, maskSecrets(err.Error())) return } + routed = true } else if next := s.m.selectHealthyLocked(id); next != "" { - if err := s.m.applyActiveProxyLocked(next, "manual"); err != nil { - s.m.proxyPool.mu.Unlock() - writeErr(w, 400, maskSecrets(err.Error())) - return + for _, candidate := range s.m.proxyPool.entries { + if candidate.ID == next && candidate.RawURL == "" { + next = "" + break + } + } + if next == "" { + // Fall through to the enabled-member fallback below. + } else { + if err := s.m.applyActiveProxyLocked(next, "manual"); err != nil { + s.m.proxyPool.mu.Unlock() + writeErr(w, 400, maskSecrets(err.Error())) + return + } + switched = true + routed = true + } + } + if !routed { + // A disabled or cleared active member must never remain the + // runtime route. Keep traffic on any surviving enabled member, + // including a cooling one, until the normal reconnect loop can + // establish a healthy route. + next := "" + for _, candidate := range s.m.proxyPool.entries { + if candidate.ID != id && candidate.Enabled && candidate.RawURL != "" { + next = candidate.ID + break + } + } + if next != "" { + if err := s.m.applyActiveProxyLockedWithHealth(next, "manual", true); err != nil { + s.m.proxyPool.mu.Unlock() + writeErr(w, 400, maskSecrets(err.Error())) + return + } + needPause = true + } else { + if err := s.m.applyActiveProxyLocked("", "manual"); err != nil { + s.m.proxyPool.mu.Unlock() + writeErr(w, 400, maskSecrets(err.Error())) + return + } + needPause = false } - } else { - needPause = true } } } @@ -909,6 +1008,9 @@ func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) { if needPause { s.pauseAllProxyTasks(errors.New("代理池暂无健康成员")) } + if switched { + s.proxyReconnectSuccess() + } writeJSON(w, 200, map[string]any{"entry": out}) return } @@ -923,6 +1025,7 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) { return } needPause := false + switched := false s.m.proxyPool.mu.Lock() found := false out := s.m.proxyPool.entries[:0] @@ -947,6 +1050,7 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) { writeErr(w, 500, maskSecrets(err.Error())) return } + switched = true } else { // Keep runtime traffic attached to a surviving pool member even when // every remaining member is cooling down; never retain the deleted URL. @@ -970,7 +1074,15 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) { writeErr(w, 500, maskSecrets(err.Error())) return } - needPause = true + // Clearing the last usable route leaves no proxy to recover. + // Only an enabled member that is actually cooling warrants the + // pool-wide proxy pause barrier. + for _, candidate := range s.m.proxyPool.entries { + if candidate.Enabled && candidate.RawURL != "" && candidate.Health == "unhealthy-cooling" { + needPause = true + break + } + } } } } @@ -983,15 +1095,22 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) { if needPause { s.pauseAllProxyTasks(errors.New("代理池暂无健康成员")) } + if switched { + s.proxyReconnectSuccess() + } writeJSON(w, 200, map[string]any{"ok": true}) } func (s *Server) activateProxyPool(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") + before := s.m.proxyPoolActiveID() if err := s.m.activateProxyID(id, "manual"); err != nil { writeErr(w, 400, err.Error()) return } + if before != id { + s.proxyReconnectSuccess() + } writeJSON(w, 200, map[string]any{"ok": true, "active_id": id, "active_proxy": maskSecrets(s.m.GlobalProxy())}) } diff --git a/server/proxy_pool_test.go b/server/proxy_pool_test.go index 3480a4a..e57906e 100644 --- a/server/proxy_pool_test.go +++ b/server/proxy_pool_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/base64" "encoding/json" + "errors" "fmt" "io" "net" @@ -15,6 +16,7 @@ import ( "testing" "time" + "github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/traffic" ) @@ -97,6 +99,220 @@ func TestProxyPoolHealthThresholdCooldownAndRecovery(t *testing.T) { } } +func TestProxyPoolUncheckedMemberDoesNotPauseBeforeFailureThreshold(t *testing.T) { + m := proxyTestManager(t) + task, err := m.CreateTask("unchecked proxy", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }() + m.globalProxy = "http://a" + m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{ + ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unknown"}, RawURL: "http://a", + }}} + s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)} + s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult { + return egressProbeResult{Error: "proxy down"} + } + s.proxyPoolTickLocked(true) + if st := task.lifecycleSnapshot(); st.Paused || st.PauseReason == "paused_by_proxy" { + t.Fatalf("unchecked member paused the task: %+v", st) + } + if got := m.proxyPool.entries[0].Health; got != "unknown" { + t.Fatalf("single failure changed unchecked member to %q", got) + } +} + +func TestProxyPoolCoolingWindowSkipsProbeUntilExpiry(t *testing.T) { + m := proxyTestManager(t) + m.globalProxy = "http://a" + m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{ + ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "unhealthy-cooling", ConsecutiveFailures: 3, LastCheckedAt: time.Now()}, RawURL: "http://a", + }}} + s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)} + calls := 0 + s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult { + calls++ + return egressProbeResult{OK: true, ExitIP: "203.0.113.5"} + } + s.proxyPoolTickLocked(false) + if calls != 0 { + t.Fatalf("cooling member was probed before cooldown expiry: calls=%d", calls) + } + m.proxyPool.entries[0].LastCheckedAt = time.Now().Add(-2 * time.Minute) + s.proxyPoolTickLocked(false) + if calls != 1 || m.proxyPool.entries[0].Health != "healthy" { + t.Fatalf("expired cooling member did not recover: calls=%d entry=%+v", calls, m.proxyPool.entries[0]) + } +} + +func TestProxyPoolTimedRotationUsesAvailableMembersAndRecoversTasks(t *testing.T) { + m := proxyTestManager(t) + if err := m.SetProxyPoolSettings(true, minProxyRotateInterval, true, "sequential", 3, 60, 30); err != nil { + t.Fatal(err) + } + m.globalProxy = "http://a" + m.proxyPool = &proxyPoolState{active: "a", lastTimed: time.Now().Add(-time.Minute), lastHealth: time.Now(), entries: []proxyPoolEntry{ + {ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"}, + {ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"}, + {ProxyPoolEntry: ProxyPoolEntry{ID: "c", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://c"}, + }} + s := &Server{m: m, ctx: context.Background()} + s.proxyPoolTickLocked(false) + if got := m.GlobalProxy(); got != "http://b" { + t.Fatalf("timed rotation selected %q", got) + } + if len(m.proxyPool.logs) == 0 || m.proxyPool.logs[len(m.proxyPool.logs)-1].Reason != "timed" { + t.Fatalf("timed rotation was not logged: %+v", m.proxyPool.logs) + } +} + +func TestProxyPoolFailureSwitchRecoversOnlyEligibleTasks(t *testing.T) { + m := proxyTestManager(t) + if err := m.SetProxyPoolSettings(false, minProxyRotateInterval, true, "sequential", 3, 60, 30); err != nil { + t.Fatal(err) + } + m.globalProxy = "http://a" + m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{ + {ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"}, + {ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"}, + }} + eligible, err := m.CreateTask("eligible", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + exhausted, err := m.CreateTask("exhausted", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { + _, _ = m.DeleteTask(eligible.ID, DeleteTaskOptions{}) + _, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{}) + }() + for _, task := range []*Task{eligible, exhausted} { + if err := m.ApplyProxyPause(task.ID); err != nil { + t.Fatal(err) + } + task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectGeneration = 7; v.ReconnectMax = 3 }) + } + exhausted.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectAttempt, v.ReconnectExhausted = 3, true }) + s := newAdmissionTestServer(m, nil) + m.SetTaskControlMutex(&s.concMu) + s.proxyPoolWake = make(chan struct{}, 1) + s.proxyPoolProbeSem = make(chan struct{}, proxyPoolMaxProbeConcurrency) + s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult { + return egressProbeResult{Error: "proxy down"} + } + s.engine.Pause(eligible.ID, agent.AbortPausedByProxy) + s.engine.Pause(exhausted.ID, agent.AbortPausedByProxy) + if !s.handleProxyPoolFailure(errors.New("proxyconnect tcp: connection refused")) { + t.Fatal("confirmed failure did not switch to the available member") + } + if got := m.GlobalProxy(); got != "http://b" { + t.Fatalf("failure switch selected %q", got) + } + if st := eligible.lifecycleSnapshot(); st.PauseReason != "" || st.Paused { + t.Fatalf("eligible task did not recover: %+v", st) + } + if st := exhausted.lifecycleSnapshot(); st.PauseReason != "paused_by_proxy" || !st.ReconnectExhausted { + t.Fatalf("exhausted task was automatically resumed: %+v", st) + } +} + +func TestProxyPoolReconnectKeepsAttemptsAndDoesNotAutoResumeExhausted(t *testing.T) { + m := proxyTestManager(t) + if err := m.pg.SetSetting(settingProxyReconnectMax, "2"); err != nil { + t.Fatal(err) + } + if err := m.pg.SetSetting(settingProxyReconnectInterval, "1"); err != nil { + t.Fatal(err) + } + m.globalProxy = "http://legacy" + m.proxyPool = &proxyPoolState{active: "legacy", entries: []proxyPoolEntry{{ + ProxyPoolEntry: ProxyPoolEntry{ID: "legacy", Enabled: true, Health: "healthy"}, RawURL: "http://legacy", + }}} + task, err := m.CreateTask("legacy reconnect", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }() + if err := m.ApplyProxyPause(task.ID); err != nil { + t.Fatal(err) + } + task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectGeneration = 3 }) + s := &Server{m: m, ctx: context.Background(), proxyPoolProbeSem: make(chan struct{}, proxyPoolMaxProbeConcurrency)} + s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult { + return egressProbeResult{Error: "proxy down"} + } + s.proxyReconnectTick() + first := task.lifecycleSnapshot() + if first.ReconnectAttempt != 1 || first.ReconnectExhausted { + t.Fatalf("first pool reconnect attempt=%+v", first) + } + task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectNextAt = 0 }) + s.proxyReconnectTick() + second := task.lifecycleSnapshot() + if second.ReconnectAttempt != 2 || !second.ReconnectExhausted || second.PauseReason != "paused_by_proxy" { + t.Fatalf("second pool reconnect did not exhaust: %+v", second) + } + s.proxyReconnectTick() + third := task.lifecycleSnapshot() + if third.ReconnectAttempt != second.ReconnectAttempt || !third.ReconnectExhausted || third.PauseReason != "paused_by_proxy" { + t.Fatalf("automatic tick revived exhausted task: %+v", third) + } + guards := s.snapshotProxyReconnectGuardsWithExhausted(false, true) + s.proxyReconnectSuccessWithGuards(guards, nil) + if got := task.lifecycleSnapshot(); got.PauseReason != "" { + t.Fatalf("manual guarded resume did not release exhausted task: %+v", got) + } +} + +func TestProxyPoolDisableOrClearActiveRoutesAway(t *testing.T) { + newServer := func(entries []proxyPoolEntry) (*Server, *Manager) { + m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: entries}} + return &Server{m: m}, m + } + t.Run("disable switches to healthy member", func(t *testing.T) { + s, m := newServer([]proxyPoolEntry{ + {ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"}, + {ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: true, Health: "healthy"}, RawURL: "http://b"}, + }) + r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"enabled":false}`)) + r.SetPathValue("id", "a") + w := httptest.NewRecorder() + s.updateProxyPool(w, r) + if w.Code != http.StatusOK || m.GlobalProxy() != "http://b" || m.proxyPoolActiveID() != "b" { + t.Fatalf("disable route status=%d proxy=%q active=%q body=%s", w.Code, m.GlobalProxy(), m.proxyPoolActiveID(), w.Body.String()) + } + }) + t.Run("clear last member clears runtime", func(t *testing.T) { + s, m := newServer([]proxyPoolEntry{{ + ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a", + }}) + r := httptest.NewRequest(http.MethodPatch, "/api/settings/proxy-pool/a", strings.NewReader(`{"url":""}`)) + r.SetPathValue("id", "a") + w := httptest.NewRecorder() + s.updateProxyPool(w, r) + if w.Code != http.StatusOK || m.GlobalProxy() != "" || m.proxyPoolActiveID() != "" { + t.Fatalf("clear route status=%d proxy=%q active=%q body=%s", w.Code, m.GlobalProxy(), m.proxyPoolActiveID(), w.Body.String()) + } + }) +} + +func TestProxyPoolCaptureOffUsesLocalForwarder(t *testing.T) { + fp, err := newDynamicForwardProxy(func() string { return "" }) + if err != nil { + t.Fatal(err) + } + defer fp.Close() + m := &Manager{forwardProxy: fp} + m.proxyPoolConfigured.Store(true) + got := m.ProxyAddr() + if !strings.HasPrefix(got, "http://127.0.0.1:") { + t.Fatalf("capture-off proxy address=%q, want local forwarder", got) + } +} + func TestProxyReconnectGuardsGateExhaustedAutomaticResume(t *testing.T) { task := &Task{ID: "exhausted", PauseReason: "paused_by_proxy", ReconnectGeneration: 4, ReconnectAttempt: 3, ReconnectExhausted: true} m := &Manager{tasks: map[string]*Task{task.ID: task}}