fix: 清空运行时代理后转发器直连并门控恢复、跳过空地址成员、补测

This commit is contained in:
RuoJi6 committed 2026-11-01 22:53:11 +08:00
1 parent 5dcfd9fd39
commit e74ab49f72
4 files changed
+149 -14

No files matched your search

+3 -1
View File
@@ -588,7 +588,9 @@ func NewManager(dir, proxyAddr string) (*Manager, error) {
m := &Manager{dir: dir, pg: pg, assets: pg.Assets(), tasks: map[string]*Task{}, interceptor: intercept.New(pg)}
if fp, ferr := newDynamicForwardProxy(func() string { return m.GlobalProxy() }); ferr == nil {
m.forwardProxy = fp
fp.direct = func() bool { return !m.proxyPoolConfigured.Load() }
// An empty runtime route means direct egress, even when the persisted
// pool still contains disabled or blank entries.
fp.direct = func() bool { return strings.TrimSpace(m.GlobalProxy()) == "" }
} else {
log.Printf("[proxy] local forwarder disabled: %v", ferr)
}
+3
View File
@@ -255,6 +255,9 @@ func (s *Server) runProxyReconnect(ctx context.Context) {
func (s *Server) proxyReconnectTick() {
if strings.TrimSpace(s.m.GlobalProxy()) == "" {
// Clearing the runtime route is a successful direct transition. Reuse the
// generation gate so only eligible, non-exhausted proxy pauses resume.
s.proxyReconnectSuccess()
return
}
if s.m.proxyPool != nil && len(s.m.proxyPoolEntries()) > 0 {
+20 -6
View File
@@ -350,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 != "unhealthy-cooling" && e.ID != exclude {
if e.Enabled && e.RawURL != "" && 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 != "unhealthy-cooling" {
if e.Enabled && e.RawURL != "" && e.Health != "unhealthy-cooling" {
pool = append(pool, e.ID)
}
}
@@ -376,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 != "unhealthy-cooling" && e.ID != exclude {
if e.Enabled && e.RawURL != "" && e.Health != "unhealthy-cooling" && e.ID != exclude {
return e.ID
}
}
@@ -541,7 +541,7 @@ func (s *Server) proxyPoolTickLocked(forceHealth bool) proxyPoolTickOutcome {
// 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) {
if e.Enabled && strings.TrimSpace(e.RawURL) != "" && e.Health == "unhealthy-cooling" && (e.LastCheckedAt.IsZero() || now.Sub(e.LastCheckedAt) >= cooldown) {
healthDue = true
break
}
@@ -562,7 +562,7 @@ func (s *Server) proxyPoolTickLocked(forceHealth bool) proxyPoolTickOutcome {
var probeMu sync.Mutex
var wg sync.WaitGroup
for _, e := range entries {
if !e.Enabled {
if !e.Enabled || strings.TrimSpace(e.RawURL) == "" {
continue
}
if !forceHealth && e.Health == "unhealthy-cooling" && e.ConsecutiveFailures >= threshold && !e.LastCheckedAt.IsZero() && now.Sub(e.LastCheckedAt) < cooldown {
@@ -899,6 +899,8 @@ func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) {
needPause := false
switched := false
routed := false
recoverPaused := false
proxyGuards := s.snapshotProxyReconnectGuards(false)
s.m.proxyPool.mu.Lock()
for i := range s.m.proxyPool.entries {
e := &s.m.proxyPool.entries[i]
@@ -990,6 +992,7 @@ func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) {
return
}
needPause = false
recoverPaused = true
}
}
}
@@ -1011,6 +1014,9 @@ func (s *Server) updateProxyPool(w http.ResponseWriter, r *http.Request) {
if switched {
s.proxyReconnectSuccess()
}
if recoverPaused {
s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
}
writeJSON(w, 200, map[string]any{"entry": out})
return
}
@@ -1026,6 +1032,8 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) {
}
needPause := false
switched := false
recoverPaused := false
proxyGuards := s.snapshotProxyReconnectGuards(false)
s.m.proxyPool.mu.Lock()
found := false
out := s.m.proxyPool.entries[:0]
@@ -1083,6 +1091,9 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) {
break
}
}
if !needPause {
recoverPaused = true
}
}
}
}
@@ -1098,6 +1109,9 @@ func (s *Server) deleteProxyPool(w http.ResponseWriter, r *http.Request) {
if switched {
s.proxyReconnectSuccess()
}
if recoverPaused {
s.proxyReconnectSuccessWithGuards(proxyGuards, nil)
}
writeJSON(w, 200, map[string]any{"ok": true})
}
@@ -1119,7 +1133,7 @@ func (s *Server) testProxyPoolEntry(w http.ResponseWriter, r *http.Request) {
defer s.proxyPoolTickMu.Unlock()
id := r.PathValue("id")
for _, e := range s.m.proxyPoolEntries() {
if e.ID == id {
if e.ID == id && e.Enabled && strings.TrimSpace(e.RawURL) != "" {
res := s.runProxyProbe(r.Context(), e.RawURL, "", defaultEgressProbes, proxyPoolProbeTimeout)
s.m.markProxyProbe(id, res)
if res.OK {
+123 -7
View File
@@ -43,10 +43,10 @@ func TestProxyPoolSnapshotMasksCredentialsAndDisabledHealth(t *testing.T) {
func TestProxyPoolSequentialSkipsUnhealthyAndDisabled(t *testing.T) {
m := &Manager{proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: false, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "c", Enabled: true, Health: "unhealthy-cooling"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "d", Enabled: true, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "b", Enabled: false, Health: "healthy"}, RawURL: "http://b"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "c", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://c"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "d", Enabled: true, Health: "healthy"}, RawURL: "http://d"},
}}}
if got := m.selectHealthyLocked("a"); got != "d" {
t.Fatalf("selected %q", got)
@@ -299,6 +299,77 @@ func TestProxyPoolDisableOrClearActiveRoutesAway(t *testing.T) {
})
}
func TestProxyPoolClearOrDeleteLastMemberRecoversOnlyEligibleProxyPauses(t *testing.T) {
run := func(t *testing.T, deleteLast bool) {
m := proxyTestManager(t)
m.globalProxy = "http://a"
m.proxyPool = &proxyPoolState{active: "a", entries: []proxyPoolEntry{{
ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a",
}}}
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)
}
userPaused, err := m.CreateTask("user paused", "goal", nil, 0, 0)
if err != nil {
t.Fatal(err)
}
defer func() {
_, _ = m.DeleteTask(eligible.ID, DeleteTaskOptions{})
_, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{})
_, _ = m.DeleteTask(userPaused.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 })
}
exhausted.updateLifecycle(func(v *taskLifecycleState) {
v.ReconnectAttempt, v.ReconnectExhausted = 3, true
})
if err := m.ApplyTaskPauseOrigin(userPaused.ID, false); err != nil {
t.Fatal(err)
}
s := newAdmissionTestServer(m, nil)
for _, task := range []*Task{eligible, exhausted, userPaused} {
s.engine.Pause(task.ID, agent.AbortPausedByProxy)
}
if deleteLast {
r := httptest.NewRequest(http.MethodDelete, "/api/settings/proxy-pool/a", nil)
r.SetPathValue("id", "a")
w := httptest.NewRecorder()
s.deleteProxyPool(w, r)
if w.Code != http.StatusOK {
t.Fatalf("delete status=%d body=%s", w.Code, w.Body.String())
}
} else {
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 {
t.Fatalf("clear status=%d body=%s", w.Code, w.Body.String())
}
}
if got := eligible.lifecycleSnapshot(); got.PauseReason != "" || got.Paused {
t.Fatalf("eligible proxy pause was not recovered: %+v", got)
}
if got := exhausted.lifecycleSnapshot(); got.PauseReason != "paused_by_proxy" || !got.ReconnectExhausted {
t.Fatalf("exhausted proxy pause was recovered: %+v", got)
}
if got := userPaused.lifecycleSnapshot(); got.PauseReason != "paused_by_user" {
t.Fatalf("user pause was recovered: %+v", got)
}
}
t.Run("clear", func(t *testing.T) { run(t, false) })
t.Run("delete", func(t *testing.T) { run(t, true) })
}
func TestProxyPoolCaptureOffUsesLocalForwarder(t *testing.T) {
fp, err := newDynamicForwardProxy(func() string { return "" })
if err != nil {
@@ -313,6 +384,31 @@ func TestProxyPoolCaptureOffUsesLocalForwarder(t *testing.T) {
}
}
func TestProxyPoolClearedRuntimeForwarderFallsBackToDirect(t *testing.T) {
target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, "direct-ok")
}))
defer target.Close()
m := proxyTestManager(t)
// A configured pool can remain persisted after its last usable member is
// disabled or cleared. The runtime route is still direct in that state.
m.proxyPoolConfigured.Store(true)
proxyURL, err := url.Parse(m.ProxyAddr())
if err != nil {
t.Fatal(err)
}
client := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)}}
resp, err := client.Get(target.URL)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusOK || string(body) != "direct-ok" {
t.Fatalf("cleared runtime route status=%d body=%q", resp.StatusCode, body)
}
}
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}}
@@ -338,9 +434,10 @@ func TestProxyPoolRandomSelectionStillExcludesDisabledAndCooling(t *testing.T) {
t.Fatal(err)
}
m.proxyPool = &proxyPoolState{entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "disabled", Enabled: false, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "cooling", Enabled: true, Health: "unhealthy-cooling"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "good", Enabled: true, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "disabled", Enabled: false, Health: "healthy"}, RawURL: "http://disabled"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "cooling", Enabled: true, Health: "unhealthy-cooling"}, RawURL: "http://cooling"},
{ProxyPoolEntry: ProxyPoolEntry{ID: "empty", Enabled: true, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "good", Enabled: true, Health: "healthy"}, RawURL: "http://good"},
}}
for i := 0; i < 20; i++ {
if got := m.selectHealthyLocked(""); got != "good" {
@@ -349,6 +446,25 @@ func TestProxyPoolRandomSelectionStillExcludesDisabledAndCooling(t *testing.T) {
}
}
func TestProxyPoolProbeSkipsEnabledEmptyURL(t *testing.T) {
m := proxyTestManager(t)
m.globalProxy = "http://good"
m.proxyPool = &proxyPoolState{active: "good", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "empty", Enabled: true, Health: "healthy"}},
{ProxyPoolEntry: ProxyPoolEntry{ID: "good", Enabled: true, Health: "healthy"}, RawURL: "http://good"},
}}
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}
}
s.proxyPoolTickLocked(true)
if calls != 1 {
t.Fatalf("probe calls=%d, want only non-empty enabled member", calls)
}
}
func TestProxyPoolFailureMarksOriginalMemberAfterConcurrentSwitch(t *testing.T) {
m := &Manager{globalProxy: "http://a", proxyPool: &proxyPoolState{active: "a", entries: []proxyPoolEntry{
{ProxyPoolEntry: ProxyPoolEntry{ID: "a", Enabled: true, Health: "healthy"}, RawURL: "http://a"},