diff --git a/adapter/outboundgroup/fallback.go b/adapter/outboundgroup/fallback.go index 98aaf20a9d..6adccfffc4 100644 --- a/adapter/outboundgroup/fallback.go +++ b/adapter/outboundgroup/fallback.go @@ -31,6 +31,9 @@ func (f *Fallback) Now() string { // DialContext implements C.ProxyAdapter func (f *Fallback) DialContext(ctx context.Context, metadata *C.Metadata) (C.Conn, error) { proxy := f.findAliveProxy(true) + if forcedHealthCheckNeeded(proxy, f.testUrl) { + go f.healthCheck() + } c, err := proxy.DialContext(ctx, metadata) if err == nil { c.AppendToChains(f) @@ -54,6 +57,9 @@ func (f *Fallback) DialContext(ctx context.Context, metadata *C.Metadata) (C.Con // ListenPacketContext implements C.ProxyAdapter func (f *Fallback) ListenPacketContext(ctx context.Context, metadata *C.Metadata) (C.PacketConn, error) { proxy := f.findAliveProxy(true) + if forcedHealthCheckNeeded(proxy, f.testUrl) { + go f.healthCheck() + } pc, err := proxy.ListenPacketContext(ctx, metadata) if err == nil { pc.AppendToChains(f) @@ -106,12 +112,12 @@ func (f *Fallback) findAliveProxy(touch bool) C.Proxy { proxies := f.GetProxies(touch) for _, proxy := range proxies { if len(f.selected) == 0 { - if proxy.AliveForTestUrl(f.testUrl) { + if f.proxyUsable(proxy) { return proxy } } else { if proxy.Name() == f.selected { - if proxy.AliveForTestUrl(f.testUrl) { + if f.proxyUsable(proxy) { return proxy } else { f.selected = "" @@ -120,9 +126,27 @@ func (f *Fallback) findAliveProxy(touch bool) C.Proxy { } } + // No member has a live test result. A member whose alive flag is merely + // stale (e.g. it was checked before its provider finished loading) is + // still a better bet than one currently resolving to REJECT, which would + // silently blackhole traffic. + for _, proxy := range proxies { + if !resolvesToReject(proxy) { + return proxy + } + } + return proxies[0] } +// proxyUsable reports whether a member can serve traffic right now: it must +// have passed the health check and not resolve to REJECT - an empty group +// serving its empty-fallback must be skipped immediately, without waiting +// for a health check to flag it dead. +func (f *Fallback) proxyUsable(proxy C.Proxy) bool { + return proxy.AliveForTestUrl(f.testUrl) && !resolvesToReject(proxy) +} + func (f *Fallback) Set(name string) error { var p C.Proxy for _, proxy := range f.GetProxies(false) { @@ -171,6 +195,7 @@ func NewFallback(option GroupCommonOption, fallbackOption FallbackOption, emptyF ExcludeType: option.ExcludeType, TestTimeout: option.TestTimeout, MaxFailedTimes: option.MaxFailedTimes, + Interval: option.Interval, EmptyFallback: emptyFallback, Providers: providers, }), diff --git a/adapter/outboundgroup/forcedcheck_test.go b/adapter/outboundgroup/forcedcheck_test.go new file mode 100644 index 0000000000..8f4d142d07 --- /dev/null +++ b/adapter/outboundgroup/forcedcheck_test.go @@ -0,0 +1,251 @@ +package outboundgroup + +import ( + "context" + stdatomic "sync/atomic" + "testing" + "time" + + "github.com/metacubex/mihomo/adapter" + "github.com/metacubex/mihomo/adapter/outbound" + "github.com/metacubex/mihomo/common/utils" + C "github.com/metacubex/mihomo/constant" + P "github.com/metacubex/mihomo/constant/provider" +) + +// stubTestURL is never dialed: tests dial through a REJECT stub (nopConn), +// so the URL only serves as the key under which alive state is stored. +// The .invalid TLD is reserved and guaranteed to never resolve. +const stubTestURL = "https://stub.invalid" + +type testProvider struct { + name string + proxies []C.Proxy + healthChecks stdatomic.Int32 +} + +func (t *testProvider) Name() string { return t.name } +func (t *testProvider) VehicleType() P.VehicleType { return P.Compatible } +func (t *testProvider) Type() P.ProviderType { return P.Proxy } +func (t *testProvider) Initial() error { return nil } +func (t *testProvider) Update() error { return nil } +func (t *testProvider) Proxies() []C.Proxy { return t.proxies } +func (t *testProvider) Count() int { return len(t.proxies) } +func (t *testProvider) Touch() {} +func (t *testProvider) HealthCheck() { t.healthChecks.Add(1) } +func (t *testProvider) Version() uint32 { return 0 } +func (t *testProvider) HealthCheckURL() string { return "" } +func (t *testProvider) RegisterHealthCheckTask(url string, expectedStatus utils.IntRanges[uint16], filter string, interval uint) { +} + +func waitForHealthChecks(t *testing.T, tp *testProvider, want int32) { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + if tp.healthChecks.Load() >= want { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("expected at least %d health checks, got %d", want, tp.healthChecks.Load()) +} + +// A failure-triggered health check must not run again within the cooldown +// window, otherwise every burst of failed dials rescans all provider nodes. +func TestForcedHealthCheckCooldown(t *testing.T) { + tp := &testProvider{name: "stub-provider"} + gb := NewGroupBase(GroupBaseOption{ + Name: "group-under-test", + Type: C.URLTest, + Providers: []P.ProxyProvider{tp}, + }) + + gb.healthCheck() + gb.healthCheck() + + if got := tp.healthChecks.Load(); got != 1 { + t.Fatalf("expected exactly 1 health check within cooldown window, got %d", got) + } +} + +// A group with a configured interval must use that interval as its +// failure-triggered cooldown, not the fixed 30s floor - otherwise a group +// configured for e.g. interval: 900 gets fully rescanned every ~30s as long +// as dial failures keep recurring, ignoring what the user configured. +func TestForcedHealthCheckCooldownUsesConfiguredInterval(t *testing.T) { + gb := NewGroupBase(GroupBaseOption{ + Name: "group-under-test", + Type: C.URLTest, + Interval: 900, + }) + + if got := gb.forcedHealthCheckCooldown(); got != 900*time.Second { + t.Fatalf("expected cooldown to match configured interval (900s), got %s", got) + } +} + +// A group without a configured interval must fall back to the minimum +// cooldown floor, preserving prior behavior for groups that never set one. +func TestForcedHealthCheckCooldownFallsBackWithoutInterval(t *testing.T) { + gb := NewGroupBase(GroupBaseOption{ + Name: "group-under-test", + Type: C.URLTest, + }) + + if got := gb.forcedHealthCheckCooldown(); got != minForcedHealthCheckCooldown { + t.Fatalf("expected fallback cooldown of %s, got %s", minForcedHealthCheckCooldown, got) + } +} + +func newURLTestMember(t *testing.T, name string, reject C.Proxy, provider P.ProxyProvider) C.Proxy { + t.Helper() + u, err := NewURLTest( + GroupCommonOption{Name: name, URL: stubTestURL}, + URLTestOption{}, + reject, + []P.ProxyProvider{provider}, + ) + if err != nil { + t.Fatal(err) + } + return adapter.NewProxy(u) +} + +// A fallback must skip a member group that currently resolves to REJECT +// (an empty group serving its empty-fallback) immediately, without waiting +// for any health check to mark it dead. +func TestFallbackSkipsMemberResolvingToReject(t *testing.T) { + reject := adapter.NewProxy(outbound.NewReject()) + + emptyMember := newURLTestMember(t, "empty-member", reject, &testProvider{name: "no-proxies"}) + liveMember := newURLTestMember(t, "live-member", reject, &testProvider{ + name: "one-proxy", + proxies: []C.Proxy{adapter.NewProxy(outbound.NewDirect())}, + }) + + f, err := NewFallback( + GroupCommonOption{Name: "fallback-group", URL: stubTestURL}, + FallbackOption{}, + reject, + []P.ProxyProvider{&testProvider{name: "members", proxies: []C.Proxy{emptyMember, liveMember}}}, + ) + if err != nil { + t.Fatal(err) + } + + if got := f.Now(); got != "live-member" { + t.Fatalf("fallback should skip the member resolving to REJECT, now: %s", got) + } +} + +// Even when every member is flagged dead (e.g. checked before providers +// finished loading), a member with real nodes beats one that would +// blackhole traffic via REJECT. +func TestFallbackPrefersStaleDeadMemberOverReject(t *testing.T) { + reject := adapter.NewProxy(outbound.NewReject()) + + emptyMember := newURLTestMember(t, "empty-member", reject, &testProvider{name: "no-proxies"}) + liveMember := newURLTestMember(t, "live-member", reject, &testProvider{ + name: "one-proxy", + proxies: []C.Proxy{adapter.NewProxy(outbound.NewDirect())}, + }) + + f, err := NewFallback( + GroupCommonOption{Name: "fallback-group", URL: stubTestURL}, + FallbackOption{}, + reject, + []P.ProxyProvider{&testProvider{name: "members", proxies: []C.Proxy{emptyMember, liveMember}}}, + ) + if err != nil { + t.Fatal(err) + } + + // mark the live member dead without touching the network: + // URLTest with an already-cancelled context fails instantly + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, _ = liveMember.URLTest(ctx, stubTestURL, nil) + if liveMember.AliveForTestUrl(stubTestURL) { + t.Fatal("precondition failed: live member should be flagged dead") + } + + if got := f.Now(); got != "live-member" { + t.Fatalf("fallback should prefer a stale-dead member over REJECT, now: %s", got) + } +} + +// A plain dead member (not resolving to REJECT) must not force a health +// check on every dial: dead proxies already fail dials naturally and are +// handled by onDialFailed's failedTimes/maxFailedTimes/cooldown gate. +// Triggering here too bypasses that gate and reruns a full group health +// check on every single dial as long as the picked member stays dead, +// storming the network regardless of the group's configured interval. +func TestForcedHealthCheckNotNeededForPlainDeadProxy(t *testing.T) { + direct := adapter.NewProxy(outbound.NewDirect()) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, _ = direct.URLTest(ctx, stubTestURL, nil) + if direct.AliveForTestUrl(stubTestURL) { + t.Fatal("precondition failed: proxy should be flagged dead") + } + + if forcedHealthCheckNeeded(direct, stubTestURL) { + t.Fatal("a plain dead proxy (not resolving to REJECT) must not force a health check on every dial") + } +} + +// Dialing through an empty fallback group resolved to REJECT "succeeds" +// (nopConn) and never reports a dial error, so the group must proactively +// re-run its health check to notice members that came back to life. +func TestFallbackRejectDialTriggersHealthCheck(t *testing.T) { + reject := adapter.NewProxy(outbound.NewReject()) + tp := &testProvider{name: "stub-provider"} // no proxies -> group resolves to empty-fallback + + f, err := NewFallback( + GroupCommonOption{Name: "fallback-group", URL: stubTestURL}, + FallbackOption{}, + reject, + []P.ProxyProvider{tp}, + ) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + conn, err := f.DialContext(ctx, &C.Metadata{}) + if err != nil { + t.Fatalf("REJECT dial should not error, got %v", err) + } + defer conn.Close() + + waitForHealthChecks(t, tp, 1) +} + +// Same for url-test groups: an empty group serving REJECT must re-check its +// providers instead of silently swallowing traffic. +func TestURLTestRejectDialTriggersHealthCheck(t *testing.T) { + reject := adapter.NewProxy(outbound.NewReject()) + tp := &testProvider{name: "stub-provider"} // no proxies -> group resolves to empty-fallback + + u, err := NewURLTest( + GroupCommonOption{Name: "urltest-group", URL: stubTestURL}, + URLTestOption{}, + reject, + []P.ProxyProvider{tp}, + ) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + conn, err := u.DialContext(ctx, &C.Metadata{}) + if err != nil { + t.Fatalf("REJECT dial should not error, got %v", err) + } + defer conn.Close() + + waitForHealthChecks(t, tp, 1) +} diff --git a/adapter/outboundgroup/groupbase.go b/adapter/outboundgroup/groupbase.go index eed45e0d61..35e712052c 100644 --- a/adapter/outboundgroup/groupbase.go +++ b/adapter/outboundgroup/groupbase.go @@ -31,8 +31,10 @@ type GroupBase struct { failedTimes int failedTime time.Time failedTesting atomic.Bool + lastForcedCheck atomic.TypedValue[time.Time] testTimeout int maxFailedTimes int + interval time.Duration emptyFallback C.Proxy // for GetProxies @@ -51,6 +53,7 @@ type GroupBaseOption struct { ExcludeType string TestTimeout int MaxFailedTimes int + Interval int EmptyFallback C.Proxy Providers []P.ProxyProvider } @@ -88,6 +91,7 @@ func NewGroupBase(opt GroupBaseOption) *GroupBase { failedTesting: atomic.NewBool(false), testTimeout: opt.TestTimeout, maxFailedTimes: opt.MaxFailedTimes, + interval: time.Duration(opt.Interval) * time.Second, emptyFallback: opt.EmptyFallback, } @@ -294,16 +298,39 @@ func (gb *GroupBase) onDialFailed(adapterType C.AdapterType, err error, fn func( if gb.failedTimes >= gb.maxFailedTimes { log.Warnln("because %s failed multiple times, activate health check", gb.Name()) fn() + gb.failedTimes = 0 } } }() } +// minForcedHealthCheckCooldown is the floor applied when a group has no +// configured interval: rescanning every provider node on each burst of +// failed dials floods the network when a group holds hundreds of proxies. +const minForcedHealthCheckCooldown = 30 * time.Second + +// forcedHealthCheckCooldown limits how often a failure-triggered health +// check may run: it must never rescan the group's providers more often than +// the group's own configured interval, otherwise a persistently failing +// group gets rescanned in full every ~30s regardless of what interval the +// user configured. +func (gb *GroupBase) forcedHealthCheckCooldown() time.Duration { + if gb.interval > 0 { + return gb.interval + } + return minForcedHealthCheckCooldown +} + func (gb *GroupBase) healthCheck() { if gb.failedTesting.Load() { return } + if time.Since(gb.lastForcedCheck.Load()) < gb.forcedHealthCheckCooldown() { + return + } + gb.lastForcedCheck.Store(time.Now()) + gb.failedTesting.Store(true) wg := sync.WaitGroup{} for _, proxyProvider := range gb.providers { @@ -320,6 +347,30 @@ func (gb *GroupBase) healthCheck() { gb.failedTimes = 0 } +// resolvesToReject reports whether traffic sent to proxy would currently be +// blackholed: the proxy itself is a REJECT, or it is a group (e.g. an empty +// url-test serving its empty-fallback) whose current pick unwraps to one. +func resolvesToReject(proxy C.Proxy) bool { + for p := proxy; p != nil; p = p.Unwrap(nil, false) { + switch p.Type() { + case C.Reject, C.RejectDrop: + return true + } + } + return false +} + +// forcedHealthCheckNeeded reports whether traffic is being served by a proxy +// that resolves to REJECT: an empty group resolved to its empty-fallback, +// whose dials "succeed" on a nop connection and therefore never reach +// onDialFailed. A plain dead member is deliberately excluded here - it +// already fails dials naturally and is handled by onDialFailed's +// failedTimes/maxFailedTimes/cooldown gate; triggering here too would bypass +// that gate on every single dial and storm the group's health check. +func forcedHealthCheckNeeded(proxy C.Proxy, testUrl string) bool { + return resolvesToReject(proxy) +} + func (gb *GroupBase) onDialSuccess() { if !gb.failedTesting.Load() { gb.failedTimes = 0 diff --git a/adapter/outboundgroup/loadbalance.go b/adapter/outboundgroup/loadbalance.go index 88d4674a03..f2da1f853d 100644 --- a/adapter/outboundgroup/loadbalance.go +++ b/adapter/outboundgroup/loadbalance.go @@ -270,6 +270,7 @@ func NewLoadBalance(option GroupCommonOption, loadBalanceOption LoadBalanceOptio ExcludeType: option.ExcludeType, TestTimeout: option.TestTimeout, MaxFailedTimes: option.MaxFailedTimes, + Interval: option.Interval, EmptyFallback: emptyFallback, Providers: providers, }), diff --git a/adapter/outboundgroup/urltest.go b/adapter/outboundgroup/urltest.go index 64e4def0de..45ae80423c 100644 --- a/adapter/outboundgroup/urltest.go +++ b/adapter/outboundgroup/urltest.go @@ -56,6 +56,9 @@ func (u *URLTest) ForceSet(name string) { // DialContext implements C.ProxyAdapter func (u *URLTest) DialContext(ctx context.Context, metadata *C.Metadata) (c C.Conn, err error) { proxy := u.fast(true) + if forcedHealthCheckNeeded(proxy, u.testUrl) { + go u.healthCheck() + } c, err = proxy.DialContext(ctx, metadata) if err == nil { c.AppendToChains(u) @@ -79,6 +82,9 @@ func (u *URLTest) DialContext(ctx context.Context, metadata *C.Metadata) (c C.Co // ListenPacketContext implements C.ProxyAdapter func (u *URLTest) ListenPacketContext(ctx context.Context, metadata *C.Metadata) (C.PacketConn, error) { proxy := u.fast(true) + if forcedHealthCheckNeeded(proxy, u.testUrl) { + go u.healthCheck() + } pc, err := proxy.ListenPacketContext(ctx, metadata) if err == nil { pc.AppendToChains(u) @@ -207,6 +213,7 @@ func NewURLTest(option GroupCommonOption, urlTestOption URLTestOption, emptyFall ExcludeType: option.ExcludeType, TestTimeout: option.TestTimeout, MaxFailedTimes: option.MaxFailedTimes, + Interval: option.Interval, EmptyFallback: emptyFallback, Providers: providers, }),