From 4aa55e9f2d69da31b5fee52050a9f94659743696 Mon Sep 17 00:00:00 2001 From: zydo Date: Fri, 7 Aug 2026 02:17:58 -0700 Subject: [PATCH 1/2] Add optional bandwidth-aware mode for urltest The latency probe measures time to response headers, which carries no information about sustained throughput. On a congested or shaped path the two decouple: a small probe completes before the connection leaves slow start, so it never observes that throughput has collapsed, and urltest ranks the slower outbound first. Add an opt-in bandwidth_test block that probes each outbound with a bounded GET, reading at most max_bytes into a single reusable buffer and cancelling at the cap rather than draining the remainder. Effective throughput is recorded alongside the existing delay and exposed through the Clash API, and two new strategies rank on it: throughput, and throughput_with_latency_floor. Disabled by default. When disabled the probe path is unchanged and no response body is ever transferred. Bandwidth probing runs on its own cadence and inherits the group's idle suspension and pause handling. Also adds urltest integration coverage, which did not previously exist. Closes #4397 --- adapter/experimental.go | 12 +- common/urltest/urltest.go | 204 ++++++++++ common/urltest/urltest_test.go | 130 +++++++ constant/timeout.go | 30 +- constant/urltest.go | 20 + daemon/started_service.go | 5 +- docs/configuration/outbound/urltest.md | 132 ++++++- docs/configuration/outbound/urltest.zh.md | 116 +++++- docs/schema.json | 43 +++ experimental/clashapi/proxies.go | 5 +- option/group.go | 34 +- protocol/group/urltest.go | 394 ++++++++++++++++++- protocol/group/urltest_test.go | 155 ++++++++ test/urltest_test.go | 436 ++++++++++++++++++++++ 14 files changed, 1676 insertions(+), 40 deletions(-) create mode 100644 common/urltest/urltest_test.go create mode 100644 constant/urltest.go create mode 100644 protocol/group/urltest_test.go create mode 100644 test/urltest_test.go diff --git a/adapter/experimental.go b/adapter/experimental.go index 134053fcbc0..cdb39e139bc 100644 --- a/adapter/experimental.go +++ b/adapter/experimental.go @@ -12,8 +12,16 @@ import ( ) type URLTestHistory struct { - Time time.Time `json:"time"` - Delay uint16 `json:"delay"` + Time time.Time `json:"time"` + // Delay is the time to response headers, in milliseconds. + Delay uint16 `json:"delay"` + // Throughput is the smoothed effective transfer rate in bytes per second, or + // zero when bandwidth testing is disabled or no sample has been taken yet. + // It is a ranking signal measured over a few hundred KiB, not a speed test + // result, and understates the capacity of a fast path. + Throughput uint32 `json:"throughput,omitempty"` + // Bytes is the number of body bytes read by the most recent bandwidth probe. + Bytes uint32 `json:"bytes,omitempty"` } type V2RayServer interface { diff --git a/common/urltest/urltest.go b/common/urltest/urltest.go index d04d34b1cab..f2a78209799 100644 --- a/common/urltest/urltest.go +++ b/common/urltest/urltest.go @@ -3,6 +3,8 @@ package urltest import ( "context" "crypto/tls" + "io" + "math" "net" "net/http" "net/url" @@ -15,6 +17,8 @@ import ( "github.com/sagernet/sing-mux" "github.com/sagernet/sing-snell" "github.com/sagernet/sing/common" + "github.com/sagernet/sing/common/buf" + E "github.com/sagernet/sing/common/exceptions" M "github.com/sagernet/sing/common/metadata" N "github.com/sagernet/sing/common/network" "github.com/sagernet/sing/common/ntp" @@ -68,6 +72,47 @@ func (s *HistoryStorage) StoreURLTestHistory(tag string, history *adapter.URLTes s.access.Unlock() } +// StoreURLTestDelay records a latency sample, preserving any bandwidth sample already +// stored for the tag. Prefer it over StoreURLTestHistory when recording a latency +// probe, so that a manually triggered delay test does not discard throughput data. +func (s *HistoryStorage) StoreURLTestDelay(tag string, delay uint16) { + s.access.Lock() + history := &adapter.URLTestHistory{ + Time: time.Now(), + Delay: delay, + } + if previous := s.delayHistory[tag]; previous != nil { + history.Throughput = previous.Throughput + history.Bytes = previous.Bytes + } + s.delayHistory[tag] = history + s.notifyUpdated() + s.access.Unlock() +} + +// StoreURLTestBandwidth records a bandwidth sample, preserving the latency sample. It +// does nothing when no latency sample exists, since an outbound without one is not +// selectable and an entry with a zero delay would corrupt latency ranking. +// +// Entries are replaced rather than mutated in place, because readers hold the pointer +// returned by LoadURLTestHistory outside the lock. +func (s *HistoryStorage) StoreURLTestBandwidth(tag string, throughput uint32, bytes uint32) { + s.access.Lock() + previous := s.delayHistory[tag] + if previous == nil { + s.access.Unlock() + return + } + s.delayHistory[tag] = &adapter.URLTestHistory{ + Time: previous.Time, + Delay: previous.Delay, + Throughput: throughput, + Bytes: bytes, + } + s.notifyUpdated() + s.access.Unlock() +} + func (s *HistoryStorage) notifyUpdated() { for _, updateHook := range s.updateHooks { updateHook.Emit(struct{}{}) @@ -153,3 +198,162 @@ func urlTest(ctx context.Context, link string, detour N.Dialer) (t uint16, err e t = uint16(time.Since(start) / time.Millisecond) return } + +// bandwidthReadBufferSize bounds the memory a bandwidth probe retains. The body is +// read into this one reused buffer and discarded, so retained memory is independent +// of the byte cap — which matters on the iOS network extension, where the process +// runs under a hard jetsam limit. +const bandwidthReadBufferSize = 32 * 1024 + +// BandwidthResult is the outcome of a bounded bandwidth probe. +type BandwidthResult struct { + // Delay is the time to response headers in milliseconds, measured with the same + // semantics as URLTest. + Delay uint16 + // Bytes is the number of body bytes read before the cap or the deadline. + Bytes uint32 + // Duration is the time spent reading those bytes, excluding Delay, so that the + // rate reflects the data phase rather than connection setup. + Duration time.Duration +} + +// minTransferDuration floors the measured transfer time. A clock with coarse +// resolution — Windows in particular — reports zero for a transfer that completes +// within one tick, and treating that as zero throughput would rank an immeasurably +// fast path as the worst one and drop it out of contention entirely. Flooring instead +// reports bytes/1ms, which understates such a path but keeps its ordering right. +const minTransferDuration = time.Millisecond + +// Throughput returns the effective transfer rate in bytes per second. +func (r *BandwidthResult) Throughput() uint32 { + if r.Bytes == 0 { + return 0 + } + duration := max(r.Duration, minTransferDuration) + throughput := float64(r.Bytes) / duration.Seconds() + if throughput >= math.MaxUint32 { + return math.MaxUint32 + } + return uint32(throughput) +} + +// BandwidthTest measures both latency and effective throughput over detour by issuing +// a GET against link and reading at most maxBytes of the response body, cancelling as +// soon as the cap is reached rather than draining the remainder. +// +// It is deliberately a ranking signal rather than a benchmark: a cap in the low +// hundreds of KiB is reached while the flow is still in or near slow start, so the +// absolute rate understates a fast path's true capacity. The ratio between a shaped +// and an unshaped path is already large at that scale, which is what selection needs. +func BandwidthTest(ctx context.Context, link string, detour N.Dialer, maxBytes uint32, timeout time.Duration) (*BandwidthResult, error) { + multiplexOutbound, isMultiplexOutbound := common.Cast[adapter.OutboundWithMultiplex](detour) + if isMultiplexOutbound && multiplexOutbound.MultiplexEnabled() { + // Warm up the multiplex session so its establishment cost is excluded, as in + // URLTest. A HEAD carries no body, so this does not consume the payload. + _, err := urlTest(ctx, link, detour) + if err != nil { + return nil, err + } + } + return bandwidthTest(ctx, link, detour, maxBytes, timeout) +} + +func bandwidthTest(ctx context.Context, link string, detour N.Dialer, maxBytes uint32, timeout time.Duration) (*BandwidthResult, error) { + linkURL, err := url.Parse(link) + if err != nil { + return nil, err + } + hostname := linkURL.Hostname() + port := linkURL.Port() + if port == "" { + switch linkURL.Scheme { + case "http": + port = "80" + case "https": + port = "443" + } + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + start := time.Now() + instance, err := detour.DialContext(ctx, "tcp", M.ParseSocksaddrHostPortStr(hostname, port)) + if err != nil { + return nil, err + } + defer instance.Close() + if N.NeedHandshakeForWrite(instance) { + start = time.Now() + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, link, nil) + if err != nil { + return nil, err + } + client := http.Client{ + Transport: &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + return instance, nil + }, + TLSClientConfig: &tls.Config{ + Time: ntp.TimeFuncFromContext(ctx), + RootCAs: adapter.RootPoolFromContext(ctx), + }, + // Measure the bytes on the wire, not what the origin chose to compress. + DisableCompression: true, + }, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + return http.ErrUseLastResponse + }, + Timeout: timeout, + } + defer client.CloseIdleConnections() + resp, err := client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + // An error page is small and arrives fast, which would score as an excellent + // path. Rate limiting in particular must not be mistaken for throughput. + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return nil, E.New("unexpected status: ", resp.Status) + } + result := &BandwidthResult{ + Delay: uint16(time.Since(start) / time.Millisecond), + } + + buffer := buf.Get(bandwidthReadBufferSize) + defer buf.Put(buffer) + transferStart := time.Now() + var readBytes uint32 + for readBytes < maxBytes { + chunk := buffer + if remaining := maxBytes - readBytes; uint32(len(chunk)) > remaining { + chunk = chunk[:remaining] + } + n, readErr := resp.Body.Read(chunk) + readBytes += uint32(n) + if readErr != nil { + if readErr != io.EOF { + err = readErr + } + break + } + } + result.Duration = time.Since(transferStart) + result.Bytes = readBytes + // Cancel before closing so the remainder of the body is torn down rather than + // drained for connection reuse; draining it would defeat the cap. + cancel() + resp.Body.Close() + + if readBytes == 0 { + if err != nil { + return nil, err + } + return nil, E.New("empty response body") + } + // A probe that timed out short of the cap still transferred real bytes, and a + // timeout is itself evidence of a slow path, so the sample is kept. + return result, nil +} diff --git a/common/urltest/urltest_test.go b/common/urltest/urltest_test.go new file mode 100644 index 00000000000..a7311049f30 --- /dev/null +++ b/common/urltest/urltest_test.go @@ -0,0 +1,130 @@ +package urltest + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "os" + "testing" + "time" + + M "github.com/sagernet/sing/common/metadata" + + "github.com/stretchr/testify/require" +) + +type directDialer struct{} + +func (directDialer) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, destination.String()) +} + +func (directDialer) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) { + return nil, os.ErrInvalid +} + +func TestBandwidthResultThroughput(t *testing.T) { + t.Parallel() + require.Equal(t, uint32(1024), (&BandwidthResult{Bytes: 1024, Duration: time.Second}).Throughput()) + require.Equal(t, uint32(2048), (&BandwidthResult{Bytes: 1024, Duration: 500 * time.Millisecond}).Throughput()) + // A transfer faster than the clock can resolve reports the floored rate, not zero: + // zero would exclude the fastest path from throughput ranking altogether. + require.Equal(t, uint32(1024000), (&BandwidthResult{Bytes: 1024}).Throughput()) + require.Equal(t, uint32(1024000), (&BandwidthResult{Bytes: 1024, Duration: time.Microsecond}).Throughput()) + // No bytes means no sample, whatever the clock says. + require.Zero(t, (&BandwidthResult{Duration: time.Second}).Throughput()) +} + +// TestBandwidthTestCap is the property the whole design rests on: the probe reads the +// cap and stops, however much the server is willing to send. +func TestBandwidthTestCap(t *testing.T) { + t.Parallel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + chunk := make([]byte, 32*1024) + // Far more than the cap, and more than the test would tolerate draining. + for range 512 { + if _, err := w.Write(chunk); err != nil { + return + } + } + })) + defer server.Close() + + const maxBytes = 64 * 1024 + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + result, err := BandwidthTest(ctx, server.URL, directDialer{}, maxBytes, 10*time.Second) + require.NoError(t, err) + require.Equal(t, uint32(maxBytes), result.Bytes) + // Duration itself may legitimately be zero on a coarse clock, so assert on the + // derived rate, which is what selection actually consumes. + require.Positive(t, result.Throughput()) +} + +func TestBandwidthTestShortBody(t *testing.T) { + t.Parallel() + const bodySize = 1000 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write(make([]byte, bodySize)) + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + result, err := BandwidthTest(ctx, server.URL, directDialer{}, 64*1024, 10*time.Second) + require.NoError(t, err) + require.Equal(t, uint32(bodySize), result.Bytes) +} + +// TestBandwidthTestRejectsErrorStatus guards the failure mode where a rate-limited +// endpoint returns a small error page fast and scores as an excellent path. +func TestBandwidthTestRejectsErrorStatus(t *testing.T) { + t.Parallel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTooManyRequests) + w.Write([]byte("slow down")) + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err := BandwidthTest(ctx, server.URL, directDialer{}, 64*1024, 10*time.Second) + require.Error(t, err) +} + +func TestBandwidthTestEmptyBody(t *testing.T) { + t.Parallel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err := BandwidthTest(ctx, server.URL, directDialer{}, 64*1024, 10*time.Second) + require.Error(t, err) +} + +func TestStoreURLTestPreservesOtherMetric(t *testing.T) { + t.Parallel() + storage := NewHistoryStorage() + defer storage.Close() + + // A bandwidth sample cannot create an entry on its own: without a latency sample + // the outbound is not selectable, and a zero delay would corrupt latency ranking. + storage.StoreURLTestBandwidth("proxy", 4096, 65536) + require.Nil(t, storage.LoadURLTestHistory("proxy")) + + storage.StoreURLTestDelay("proxy", 100) + storage.StoreURLTestBandwidth("proxy", 4096, 65536) + history := storage.LoadURLTestHistory("proxy") + require.NotNil(t, history) + require.Equal(t, uint16(100), history.Delay) + require.Equal(t, uint32(4096), history.Throughput) + + // A later latency probe must not discard the throughput sample. + storage.StoreURLTestDelay("proxy", 120) + history = storage.LoadURLTestHistory("proxy") + require.Equal(t, uint16(120), history.Delay) + require.Equal(t, uint32(4096), history.Throughput) + require.Equal(t, uint32(65536), history.Bytes) +} diff --git a/constant/timeout.go b/constant/timeout.go index bedfddf5508..8daadcad6a9 100644 --- a/constant/timeout.go +++ b/constant/timeout.go @@ -3,20 +3,22 @@ package constant import "time" const ( - TCPKeepAliveInitial = 5 * time.Minute - TCPKeepAliveInterval = 75 * time.Second - TCPConnectTimeout = 5 * time.Second - TCPTimeout = 15 * time.Second - ReadPayloadTimeout = 300 * time.Millisecond - DNSTimeout = 10 * time.Second - UDPTimeout = 5 * time.Minute - ICMPTimeout = 10 * time.Second - DefaultURLTestInterval = 3 * time.Minute - DefaultURLTestIdleTimeout = 30 * time.Minute - StartTimeout = 10 * time.Second - StopTimeout = 5 * time.Second - FatalStopTimeout = 10 * time.Second - TLSFragmentFallbackDelay = 500 * time.Millisecond + TCPKeepAliveInitial = 5 * time.Minute + TCPKeepAliveInterval = 75 * time.Second + TCPConnectTimeout = 5 * time.Second + TCPTimeout = 15 * time.Second + ReadPayloadTimeout = 300 * time.Millisecond + DNSTimeout = 10 * time.Second + UDPTimeout = 5 * time.Minute + ICMPTimeout = 10 * time.Second + DefaultURLTestInterval = 3 * time.Minute + DefaultURLTestIdleTimeout = 30 * time.Minute + DefaultURLTestBandwidthTimeout = 5 * time.Second + StartTimeout = 10 * time.Second + StopTimeout = 5 * time.Second + FatalStopTimeout = 10 * time.Second + FakeIPMetadataSaveInterval = 10 * time.Second + TLSFragmentFallbackDelay = 500 * time.Millisecond ) var PortProtocols = map[uint16]string{ diff --git a/constant/urltest.go b/constant/urltest.go new file mode 100644 index 00000000000..84dce1ec13f --- /dev/null +++ b/constant/urltest.go @@ -0,0 +1,20 @@ +package constant + +// DefaultURLTestBandwidthURLPrefix is the counterpart to the latency probe's +// generate_204 endpoint. The requested byte count is appended from max_bytes, so the +// response is exactly as large as the probe will read and nothing is sent that the +// probe will cancel away. +const DefaultURLTestBandwidthURLPrefix = "https://speed.cloudflare.com/__down?bytes=" + +// Selection strategies for the urltest outbound group. +const ( + // URLTestStrategyLatency ranks outbounds by the latency probe alone. This is the + // default, and the only strategy available when bandwidth testing is disabled. + URLTestStrategyLatency = "latency" + // URLTestStrategyThroughput ranks outbounds by measured throughput, ignoring + // latency beyond liveness. + URLTestStrategyThroughput = "throughput" + // URLTestStrategyThroughputWithLatencyFloor discards outbounds whose latency + // exceeds the configured floor, then ranks the survivors by throughput. + URLTestStrategyThroughputWithLatencyFloor = "throughput_with_latency_floor" +) diff --git a/daemon/started_service.go b/daemon/started_service.go index 0557d879fe6..77ca09fc377 100644 --- a/daemon/started_service.go +++ b/daemon/started_service.go @@ -744,10 +744,7 @@ func (s *StartedService) URLTest(ctx context.Context, request *URLTestRequest) ( if err != nil { historyStorage.DeleteURLTestHistory(outboundTag) } else { - historyStorage.StoreURLTestHistory(outboundTag, &adapter.URLTestHistory{ - Time: time.Now(), - Delay: t, - }) + historyStorage.StoreURLTestDelay(outboundTag, t) } }() } diff --git a/docs/configuration/outbound/urltest.md b/docs/configuration/outbound/urltest.md index f4b3b0aa8e9..878f6df8e59 100644 --- a/docs/configuration/outbound/urltest.md +++ b/docs/configuration/outbound/urltest.md @@ -14,7 +14,19 @@ "interval": "", "tolerance": 0, "idle_timeout": "", - "interrupt_exist_connections": false + "interrupt_exist_connections": false, + "bandwidth_test": { + "enabled": false, + "url": "", + "max_bytes": 0, + "timeout": "", + "interval": "", + "concurrency": 0, + "strategy": "", + "latency_floor": "", + "throughput_tolerance": 0, + "samples": 0 + } } ``` @@ -47,3 +59,121 @@ The idle timeout. `30m` will be used if empty. Interrupt existing connections when the selected outbound has changed. Only inbound connections are affected by this setting, internal connections will always be interrupted. + +#### bandwidth_test + +!!! question "Since sing-box 1.14.0" + +Optional bandwidth-aware probing, disabled by default. + +The latency probe measures the time to response headers, which says nothing about sustained +throughput. On a congested or shaped path the two decouple: a small probe can finish quickly on a +path whose throughput has collapsed, because it completes before the connection leaves slow start. +When this is enabled, each outbound is additionally probed with a bounded `GET`, and the effective +transfer rate becomes available as a selection input. + +When disabled, the probe path is unchanged and no response body is ever transferred. + +!!! warning "This is not a speed test" + + The measurement is deliberately bounded, and reads only a few hundred KiB — while the flow is + still in or near slow start. The absolute rate understates a fast path's real capacity, and must + not be presented to users as a speed test result. It is a *ranking* signal: the ratio between a + shaped and an unshaped path is already large at this scale. Shaping that only engages after + several MiB will not be detected. + +!!! warning "Data usage" + + Unlike the latency probe, this transfers a payload over every outbound on every interval. At the + default 256 KiB with 10 outbounds every 15 minutes, that is roughly 10 MiB/hour, which matters + on a metered connection. Probing inherits the group's idle suspension and pause handling, so it + stops while the group is unused or the device is asleep. + +#### bandwidth_test.enabled + +Enable bandwidth probing. + +#### bandwidth_test.url + +The URL to download from. It must return a body of at least `max_bytes`; `generate_204` returns no +body and cannot be used. + +`https://speed.cloudflare.com/__down?bytes=` will be used if empty — the byte count +tracks `max_bytes`, so the response is exactly as large as the probe reads. + +!!! warning "Prefer your own endpoint" + + The default is shared by every client that enables this feature. If you operate a suitable + endpoint, or your provider does, point at that instead. + +The URL must return `2xx` directly. Redirects are not followed, so a redirecting endpoint fails the +probe — use the final URL. Note that the latency probe accepts any status while this one requires +`2xx`, so an endpoint can pass latency probing and still report zero throughput. + +#### bandwidth_test.max_bytes + +A hard cap on the body bytes read per probe. `262144` (256 KiB) will be used if empty; values above +`1048576` (1 MiB) are rejected. + +This caps bytes *read*, not bytes *retained* — the body is read into one small reusable buffer and +discarded, so memory use is independent of this setting. + +#### bandwidth_test.timeout + +Per-probe timeout. `5s` will be used if empty. + +A probe that times out short of the cap still yields a sample, computed from the bytes actually +transferred: a timeout is itself evidence of a slow path. + +#### bandwidth_test.interval + +The bandwidth test interval. Five times `interval` will be used if empty. + +The right cadence for a throughput probe is much lower than for a liveness probe. Setting this +higher than `idle_timeout` means probing will rarely run. + +#### bandwidth_test.concurrency + +How many bandwidth probes may run at once. `2` will be used if empty. + +Deliberately far below the latency sweep's fixed 10: these probes consume bandwidth, so running many +at once makes them contend with each other and skews every result. + +#### bandwidth_test.strategy + +How the selection is ranked: + +| Strategy | Behavior | +| ------------------------------- | ---------------------------------------------------------------------------------------------------- | +| `latency` | Default. Rank by latency, unchanged. Throughput is measured and exposed, but not used for selection. | +| `throughput` | Rank by throughput. Latency is ignored beyond liveness. | +| `throughput_with_latency_floor` | Discard outbounds whose latency exceeds `latency_floor`, then rank the survivors by throughput. | + +Selection falls back to latency ranking whenever no outbound has a throughput sample yet — during +startup, before the first bandwidth sweep, and after a network change. + +#### bandwidth_test.latency_floor + +Under `throughput_with_latency_floor`, outbounds whose latency exceeds this are excluded from +throughput ranking. Empty or `0` disables the floor, which makes that strategy equivalent to +`throughput`. + +This keeps an outbound that is pathologically slow to connect from winning on bulk transfer alone, +which matters for interactive traffic. + +#### bandwidth_test.throughput_tolerance + +Relative hysteresis, as a percentage. `25` will be used if empty. + +A challenger only replaces the current outbound when its throughput exceeds it by this margin. The +band is relative rather than absolute because throughput ratios are the meaningful comparison. This +is the throughput counterpart of `tolerance`. + +#### bandwidth_test.samples + +How many recent samples to smooth over, using the median. `3` will be used if empty. + +Throughput samples are noisier than latency samples — a probe landing during a transient burst can +swing the value severalfold. Smoothing keeps the group from oscillating and repeatedly interrupting +connections. A failed probe is recorded as a zero sample, so a sustained failure decays an outbound +out of contention while a single transient one is absorbed. diff --git a/docs/configuration/outbound/urltest.zh.md b/docs/configuration/outbound/urltest.zh.md index 4372298afc5..ba12c4dc50c 100644 --- a/docs/configuration/outbound/urltest.zh.md +++ b/docs/configuration/outbound/urltest.zh.md @@ -14,7 +14,19 @@ "interval": "", "tolerance": 50, "idle_timeout": "", - "interrupt_exist_connections": false + "interrupt_exist_connections": false, + "bandwidth_test": { + "enabled": false, + "url": "", + "max_bytes": 0, + "timeout": "", + "interval": "", + "concurrency": 0, + "strategy": "", + "latency_floor": "", + "throughput_tolerance": 0, + "samples": 0 + } } ``` @@ -46,4 +58,104 @@ 当选定的出站发生更改时，中断现有连接。 -仅入站连接受此设置影响，内部连接将始终被中断。 \ No newline at end of file +仅入站连接受此设置影响，内部连接将始终被中断。 + +#### bandwidth_test + +!!! question "自 sing-box 1.14.0 起" + +可选的带宽感知探测，默认禁用。 + +延迟探测测量的是获取响应头所用的时间，它并不能反映持续吞吐量。在拥塞或被限速的链路上，两者会明显背离： +探测在连接离开慢启动之前就已完成，因此即使链路的持续吞吐量已经崩溃，探测仍然可以很快返回。 +启用后，将额外使用有限长度的 `GET` 请求探测每个出站，其有效传输速率可作为选择依据。 + +禁用时，探测路径保持不变，且不会传输任何响应正文。 + +!!! warning "这不是测速" + + 该测量被刻意限制在几百 KiB 以内，此时连接仍处于或接近慢启动阶段。其绝对数值会低估高速链路的真实容量， + 不应作为测速结果呈现给用户。它是一个**排序**信号：在这个量级上，被限速与未被限速链路之间的比值已经足够大。 + 仅在传输数 MiB 之后才生效的限速无法被检测到。 + +!!! warning "流量消耗" + + 与延迟探测不同，此探测会在每个间隔内通过每个出站传输数据。以默认的 256 KiB、10 个出站、每 15 分钟一次计算， + 约为每小时 10 MiB，这在计量连接上不容忽视。探测沿用了组的空闲挂起与暂停机制，因此在组未被使用或设备休眠时不会运行。 + +#### bandwidth_test.enabled + +启用带宽探测。 + +#### bandwidth_test.url + +用于下载的链接。它必须返回至少 `max_bytes` 大小的正文；`generate_204` 不返回正文，因此不可使用。 + +默认使用 `https://speed.cloudflare.com/__down?bytes=` — 请求的字节数跟随 `max_bytes`， +因此响应大小恰好等于探测将要读取的字节数。 + +!!! warning "建议使用您自己的端点" + + 默认端点将被所有启用此功能的客户端共享。如果您或您的服务提供商拥有合适的端点，请改用它。 + +该链接必须直接返回 `2xx`。重定向不会被跟随，因此会导致探测失败 — 请使用最终的链接。 +请注意，延迟探测接受任何状态码，而此探测要求 `2xx`，因此某个端点可能通过延迟探测但吞吐量始终为零。 + +#### bandwidth_test.max_bytes + +每次探测读取正文字节数的硬性上限。默认使用 `262144`（256 KiB）；超过 `1048576`（1 MiB）的值将被拒绝。 + +此项限制的是**读取**的字节数，而非**保留**的字节数 — 正文被读入一个可复用的小缓冲区后即被丢弃， +因此内存占用与此设置无关。 + +#### bandwidth_test.timeout + +单次探测超时。默认使用 `5s`。 + +未达到上限即超时的探测仍会产生一个样本，按实际传输的字节数计算：超时本身就是链路缓慢的证据。 + +#### bandwidth_test.interval + +带宽测试间隔。默认使用 `interval` 的五倍。 + +吞吐量探测的合适频率远低于存活性探测。若将其设置为大于 `idle_timeout`，探测将很少运行。 + +#### bandwidth_test.concurrency + +同时进行的带宽探测数量。默认使用 `2`。 + +该值刻意远低于延迟测试固定的 10：这些探测会实际消耗带宽，同时运行过多会使它们相互争抢，从而扭曲所有结果。 + +#### bandwidth_test.strategy + +选择的排序方式： + +| 策略 | 行为 | +| ------------------------------- | ------------------------------------------------------------------------ | +| `latency` | 默认值。按延迟排序，与原有行为一致。吞吐量会被测量并公开，但不参与选择。 | +| `throughput` | 按吞吐量排序。除存活性外忽略延迟。 | +| `throughput_with_latency_floor` | 排除延迟超过 `latency_floor` 的出站，然后按吞吐量对其余出站排序。 | + +当尚无出站拥有吞吐量样本时，选择将回退到延迟排序 — 包括启动时、首次带宽测试之前，以及网络变更之后。 + +#### bandwidth_test.latency_floor + +在 `throughput_with_latency_floor` 策略下，延迟超过此值的出站将被排除在吞吐量排序之外。 +留空或 `0` 将禁用该下限，此时该策略等同于 `throughput`。 + +这可以防止连接速度极慢的出站仅凭批量传输能力胜出，这对交互式流量很重要。 + +#### bandwidth_test.throughput_tolerance + +以百分比表示的相对容差。默认使用 `25`。 + +只有当挑战者的吞吐量超出当前出站该比例时，才会将其替换。之所以使用相对而非绝对容差， +是因为吞吐量的比值才是有意义的比较。此项是 `tolerance` 在吞吐量维度上的对应项。 + +#### bandwidth_test.samples + +用于平滑的最近样本数量，取中位数。默认使用 `3`。 + +吞吐量样本比延迟样本噪声大得多 — 恰好落在流量突发期的一次探测可能使数值波动数倍。 +平滑可避免组反复振荡并中断连接。探测失败会被记录为零样本，因此持续失败会使出站逐渐退出竞争， +而单次瞬时失败则会被吸收。 \ No newline at end of file diff --git a/docs/schema.json b/docs/schema.json index 4e0f7eab782..09e342c9554 100644 --- a/docs/schema.json +++ b/docs/schema.json @@ -14401,6 +14401,9 @@ }, "interrupt_exist_connections": { "type": "boolean" + }, + "bandwidth_test": { + "$ref": "#/$defs/URLTestBandwidthTestOptions" } }, "required": [ @@ -18067,6 +18070,46 @@ } ] }, + "URLTestBandwidthTestOptions": { + "type": "object", + "properties": { + "enabled": { + "type": "boolean" + }, + "url": { + "type": "string" + }, + "max_bytes": { + "type": "integer", + "minimum": 0, + "maximum": 4294967295 + }, + "timeout": { + "$ref": "#/$defs/Duration" + }, + "interval": { + "$ref": "#/$defs/Duration" + }, + "concurrency": { + "type": "integer" + }, + "strategy": { + "type": "string" + }, + "latency_floor": { + "$ref": "#/$defs/Duration" + }, + "throughput_tolerance": { + "type": "integer", + "minimum": 0, + "maximum": 65535 + }, + "samples": { + "type": "integer" + } + }, + "additionalProperties": false + }, "USBIPDeviceMatch": { "type": "object", "properties": { diff --git a/experimental/clashapi/proxies.go b/experimental/clashapi/proxies.go index fd8a6e23d9f..259dfc2cbab 100644 --- a/experimental/clashapi/proxies.go +++ b/experimental/clashapi/proxies.go @@ -236,10 +236,7 @@ func getProxyDelay(server *Server) func(w http.ResponseWriter, r *http.Request) if err != nil { server.urlTestHistory.DeleteURLTestHistory(realTag) } else { - server.urlTestHistory.StoreURLTestHistory(realTag, &adapter.URLTestHistory{ - Time: time.Now(), - Delay: delay, - }) + server.urlTestHistory.StoreURLTestDelay(realTag, delay) } for _, detour := range server.outbound.Outbounds() { urlTestGroup, isURLTestGroup := detour.(adapter.URLTestGroup) diff --git a/option/group.go b/option/group.go index bc0c07e4790..301b3195eae 100644 --- a/option/group.go +++ b/option/group.go @@ -9,10 +9,32 @@ type SelectorOutboundOptions struct { } type URLTestOutboundOptions struct { - Outbounds []string `json:"outbounds" reference:"outbound"` - URL string `json:"url,omitempty"` - Interval badoption.Duration `json:"interval,omitempty"` - Tolerance uint16 `json:"tolerance,omitempty"` - IdleTimeout badoption.Duration `json:"idle_timeout,omitempty"` - InterruptExistConnections bool `json:"interrupt_exist_connections,omitempty"` + Outbounds []string `json:"outbounds" reference:"outbound"` + URL string `json:"url,omitempty"` + Interval badoption.Duration `json:"interval,omitempty"` + Tolerance uint16 `json:"tolerance,omitempty"` + IdleTimeout badoption.Duration `json:"idle_timeout,omitempty"` + InterruptExistConnections bool `json:"interrupt_exist_connections,omitempty"` + BandwidthTest *URLTestBandwidthTestOptions `json:"bandwidth_test,omitempty"` +} + +// URLTestBandwidthTestOptions configures an optional bandwidth probe that supplements +// the latency probe. It is disabled by default; when disabled, the probe path is +// unchanged and no response body is ever transferred. +type URLTestBandwidthTestOptions struct { + Enabled bool `json:"enabled,omitempty"` + // URL must return a body of at least MaxBytes. When empty, a Cloudflare speed + // endpoint sized to MaxBytes is used; prefer your own, since a shared default is + // fetched by every client that enables this. + URL string `json:"url,omitempty"` + // MaxBytes is a hard cap on the number of body bytes read per probe. It bounds + // the bytes read, not the memory retained; the reader uses a small fixed buffer. + MaxBytes uint32 `json:"max_bytes,omitempty"` + Timeout badoption.Duration `json:"timeout,omitempty"` + Interval badoption.Duration `json:"interval,omitempty"` + Concurrency int `json:"concurrency,omitempty"` + Strategy string `json:"strategy,omitempty"` + LatencyFloor badoption.Duration `json:"latency_floor,omitempty"` + ThroughputTolerance uint16 `json:"throughput_tolerance,omitempty"` + Samples int `json:"samples,omitempty"` } diff --git a/protocol/group/urltest.go b/protocol/group/urltest.go index f69773ef936..d7dc27db7ed 100644 --- a/protocol/group/urltest.go +++ b/protocol/group/urltest.go @@ -4,7 +4,10 @@ import ( "context" "io" "maps" + "math" "net" + "slices" + "strconv" "sync" "sync/atomic" "time" @@ -49,6 +52,7 @@ type URLTest struct { group *URLTestGroup checkAccess sync.Mutex interruptExternalConnections bool + bandwidthTest *option.URLTestBandwidthTestOptions } func NewURLTest(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.URLTestOutboundOptions) (adapter.Outbound, error) { @@ -63,6 +67,7 @@ func NewURLTest(ctx context.Context, router adapter.Router, logger log.ContextLo tolerance: options.Tolerance, idleTimeout: time.Duration(options.IdleTimeout), interruptExternalConnections: options.InterruptExistConnections, + bandwidthTest: options.BandwidthTest, } if len(outbound.tags) == 0 { return nil, E.New("missing tags") @@ -81,7 +86,7 @@ func (s *URLTest) Start(stage adapter.StartStage, scope *adapter.Scope) error { } outbounds = append(outbounds, detour) } - group, err := NewURLTestGroup(s.ctx, s.outbound, s.logger, outbounds, s.link, s.interval, s.tolerance, s.idleTimeout, s.interruptExternalConnections) + group, err := NewURLTestGroup(s.ctx, s.outbound, s.logger, outbounds, s.link, s.interval, s.tolerance, s.idleTimeout, s.interruptExternalConnections, s.bandwidthTest) if err != nil { return err } @@ -150,6 +155,11 @@ func (s *URLTest) InterfaceUpdated(ctx context.Context) { if group.pause.IsDevicePaused() || group.pause.IsNetworkPaused() { return } + // Throughput observed on the previous network says nothing about this one, and + // re-probing every outbound on every interface change would be expensive on + // exactly the devices where interface changes are frequent. Drop the samples and + // let selection fall back to latency until the next scheduled bandwidth sweep. + group.resetBandwidth() go func() { s.checkAccess.Lock() defer s.checkAccess.Unlock() @@ -227,9 +237,125 @@ type URLTestGroup struct { close chan struct{} started bool lastActive common.TypedValue[time.Time] + + // bandwidth is nil unless bandwidth testing is enabled, which keeps every code + // path below a no-op by default. + bandwidth *bandwidthTestOptions + bandwidthChecking atomic.Bool + bandwidthTicker *time.Ticker + bandwidthPauseCallback *list.Element[pause.Callback] + bandwidthAccess sync.Mutex + bandwidthHistory map[string]*bandwidthState +} + +// bandwidthState holds the smoothing window for one outbound. Throughput samples are +// far noisier than latency samples — a probe landing during a transient burst can +// swing severalfold — so selection reads the smoothed value, never the last sample. +type bandwidthState struct { + samples []uint32 + smoothed uint32 + lastTest time.Time +} + +type bandwidthTestOptions struct { + link string + maxBytes uint32 + timeout time.Duration + interval time.Duration + concurrency int + strategy string + latencyFloor uint16 // milliseconds; zero disables the floor + throughputTolerance uint16 // percent + samples int +} + +const ( + defaultBandwidthMaxBytes = 256 * 1024 + // maxBandwidthMaxBytes caps what a single probe may pull. This is a ranking + // signal, and the cost is paid on every outbound on every interval, potentially + // over metered data. + maxBandwidthMaxBytes = 1024 * 1024 + // defaultBandwidthConcurrency is deliberately far below the latency sweep's + // fixed 10: these probes consume bandwidth, so running many at once makes them + // contend with each other and skews every result. + defaultBandwidthConcurrency = 2 + // defaultBandwidthIntervalMultiplier derives the bandwidth interval from the + // latency interval, since the right cadence for a throughput probe is much lower + // than for a liveness probe. + defaultBandwidthIntervalMultiplier = 5 + defaultBandwidthTolerance = 25 + defaultBandwidthSamples = 3 +) + +func newBandwidthTestOptions(options *option.URLTestBandwidthTestOptions, interval time.Duration) (*bandwidthTestOptions, error) { + if options == nil || !options.Enabled { + return nil, nil + } + parsed := &bandwidthTestOptions{ + link: options.URL, + maxBytes: options.MaxBytes, + timeout: time.Duration(options.Timeout), + interval: time.Duration(options.Interval), + concurrency: options.Concurrency, + strategy: options.Strategy, + throughputTolerance: options.ThroughputTolerance, + samples: options.Samples, + } + if parsed.maxBytes == 0 { + parsed.maxBytes = defaultBandwidthMaxBytes + } else if parsed.maxBytes > maxBandwidthMaxBytes { + return nil, E.New("bandwidth_test.max_bytes must be less or equal than ", maxBandwidthMaxBytes) + } + if parsed.link == "" { + // Derived after max_bytes is resolved so the endpoint returns exactly the + // number of bytes the probe intends to read. + parsed.link = C.DefaultURLTestBandwidthURLPrefix + strconv.FormatUint(uint64(parsed.maxBytes), 10) + } + if parsed.timeout == 0 { + parsed.timeout = C.DefaultURLTestBandwidthTimeout + } else if parsed.timeout < 0 { + return nil, E.New("bandwidth_test.timeout must be positive") + } + if parsed.interval == 0 { + parsed.interval = interval * defaultBandwidthIntervalMultiplier + } else if parsed.interval < 0 { + return nil, E.New("bandwidth_test.interval must be positive") + } + if parsed.concurrency == 0 { + parsed.concurrency = defaultBandwidthConcurrency + } else if parsed.concurrency < 0 { + return nil, E.New("bandwidth_test.concurrency must be positive") + } + switch parsed.strategy { + case "": + parsed.strategy = C.URLTestStrategyLatency + case C.URLTestStrategyLatency, C.URLTestStrategyThroughput, C.URLTestStrategyThroughputWithLatencyFloor: + default: + return nil, E.New("unknown bandwidth_test.strategy: ", parsed.strategy) + } + latencyFloor := time.Duration(options.LatencyFloor) + if latencyFloor < 0 { + return nil, E.New("bandwidth_test.latency_floor must be positive") + } + if floorMilliseconds := latencyFloor.Milliseconds(); floorMilliseconds > math.MaxUint16 { + // Delay is measured in uint16 milliseconds, so a larger floor can never + // exclude anything anyway. + parsed.latencyFloor = math.MaxUint16 + } else { + parsed.latencyFloor = uint16(floorMilliseconds) + } + if parsed.throughputTolerance == 0 { + parsed.throughputTolerance = defaultBandwidthTolerance + } + if parsed.samples == 0 { + parsed.samples = defaultBandwidthSamples + } else if parsed.samples < 0 { + return nil, E.New("bandwidth_test.samples must be positive") + } + return parsed, nil } -func NewURLTestGroup(ctx context.Context, outboundManager adapter.OutboundManager, logger log.Logger, outbounds []adapter.Outbound, link string, interval time.Duration, tolerance uint16, idleTimeout time.Duration, interruptExternalConnections bool) (*URLTestGroup, error) { +func NewURLTestGroup(ctx context.Context, outboundManager adapter.OutboundManager, logger log.Logger, outbounds []adapter.Outbound, link string, interval time.Duration, tolerance uint16, idleTimeout time.Duration, interruptExternalConnections bool, bandwidthOptions *option.URLTestBandwidthTestOptions) (*URLTestGroup, error) { if interval == 0 { interval = C.DefaultURLTestInterval } @@ -242,6 +368,13 @@ func NewURLTestGroup(ctx context.Context, outboundManager adapter.OutboundManage if interval > idleTimeout { return nil, E.New("interval must be less or equal than idle_timeout") } + bandwidth, err := newBandwidthTestOptions(bandwidthOptions, interval) + if err != nil { + return nil, err + } + if bandwidth != nil && bandwidth.interval > idleTimeout { + logger.Warn("urltest: bandwidth_test.interval (", bandwidth.interval, ") is greater than idle_timeout (", idleTimeout, "); bandwidth probing will rarely run") + } history := service.PtrFromContext[urltest.HistoryStorage](ctx) if history == nil { return nil, E.New("missing URL test history storage") @@ -260,6 +393,8 @@ func NewURLTestGroup(ctx context.Context, outboundManager adapter.OutboundManage pause: service.FromContext[pause.Manager](ctx), interruptGroup: interrupt.NewGroup(), interruptExternalConnections: interruptExternalConnections, + bandwidth: bandwidth, + bandwidthHistory: make(map[string]*bandwidthState), }, nil } @@ -279,29 +414,71 @@ func (g *URLTestGroup) Touch() { defer g.access.Unlock() if g.ticker != nil { g.lastActive.Store(time.Now()) + // The two tickers suspend independently, and the latency one suspends first + // because it ticks more often, so the bandwidth ticker may still need + // restarting even when the latency one is already running. + g.startBandwidthTicker() return } ticker := time.NewTicker(g.interval) g.ticker = ticker g.pauseCallback = pause.RegisterTicker(g.pause, ticker, g.interval, nil) go g.loopCheck(ticker, g.close) + g.startBandwidthTicker() +} + +// startBandwidthTicker must be called with access held. +func (g *URLTestGroup) startBandwidthTicker() { + if g.bandwidth == nil || g.bandwidthTicker != nil { + return + } + // A separate ticker rather than a divisor of the latency one, so the two cadences + // stay independent. It inherits the same pause registration and the same idle + // suspension, so no probing happens while the group is unused or the device is + // asleep. + ticker := time.NewTicker(g.bandwidth.interval) + g.bandwidthTicker = ticker + g.bandwidthPauseCallback = pause.RegisterTicker(g.pause, ticker, g.bandwidth.interval, nil) + go g.loopBandwidthCheck(ticker, g.close) } func (g *URLTestGroup) Close() error { g.access.Lock() defer g.access.Unlock() - if g.ticker == nil { + // Checked independently: either ticker may already have suspended itself on idle + // timeout while the other is still running. + if g.ticker == nil && g.bandwidthTicker == nil { return nil } - g.ticker.Stop() - g.ticker = nil - g.pause.UnregisterCallback(g.pauseCallback) - g.pauseCallback = nil + if g.ticker != nil { + g.ticker.Stop() + g.ticker = nil + g.pause.UnregisterCallback(g.pauseCallback) + g.pauseCallback = nil + } + if g.bandwidthTicker != nil { + g.bandwidthTicker.Stop() + g.bandwidthTicker = nil + g.pause.UnregisterCallback(g.bandwidthPauseCallback) + g.bandwidthPauseCallback = nil + } close(g.close) return nil } func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) { + if g.bandwidth != nil && g.bandwidth.strategy != C.URLTestStrategyLatency { + // Falls through to latency ranking until throughput samples exist, which + // covers startup, the interval before the first bandwidth sweep, and the + // case where every outbound was excluded by the latency floor. + if outbound, exists, ranked := g.selectByThroughput(network); ranked { + return outbound, exists + } + } + return g.selectByLatency(network) +} + +func (g *URLTestGroup) selectByLatency(network string) (adapter.Outbound, bool) { var minDelay uint16 var minOutbound adapter.Outbound switch network { @@ -345,6 +522,206 @@ func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) { return minOutbound, true } +// selectByThroughput ranks eligible outbounds by smoothed throughput. The third +// return value reports whether any outbound was eligible at all; when it is false the +// caller falls back to latency ranking rather than leaving the group unselected. +func (g *URLTestGroup) selectByThroughput(network string) (adapter.Outbound, bool, bool) { + var maxThroughput uint32 + var maxOutbound adapter.Outbound + // Seed with the incumbent so it keeps the same advantage it has under latency + // ranking: a challenger has to beat it by the tolerance, not merely tie it. + var selected adapter.Outbound + switch network { + case N.NetworkTCP: + selected = g.selectedOutboundTCP + case N.NetworkUDP: + selected = g.selectedOutboundUDP + } + if selected != nil { + if throughput, eligible := g.candidateThroughput(selected, network); eligible { + maxOutbound = selected + maxThroughput = throughput + } + } + for _, detour := range g.outbounds { + if !common.Contains(detour.Network(), network) { + continue + } + throughput, eligible := g.candidateThroughput(detour, network) + if !eligible { + continue + } + if maxOutbound == nil || beatsIncumbent(throughput, maxThroughput, g.bandwidth.throughputTolerance) { + maxThroughput = throughput + maxOutbound = detour + } + } + if maxOutbound == nil { + return nil, false, false + } + return maxOutbound, true, true +} + +// candidateThroughput reports the smoothed throughput of detour and whether it may be +// ranked by it: the outbound needs a latency sample (liveness), a non-zero throughput +// sample, and — under throughput_with_latency_floor — a latency within the floor. +func (g *URLTestGroup) candidateThroughput(detour adapter.Outbound, network string) (uint32, bool) { + realTag := RealTag(detour, network) + history := g.history.LoadURLTestHistory(realTag) + if history == nil { + return 0, false + } + if g.bandwidth.strategy == C.URLTestStrategyThroughputWithLatencyFloor && + g.bandwidth.latencyFloor > 0 && + history.Delay > g.bandwidth.latencyFloor { + return 0, false + } + throughput := g.loadBandwidth(realTag) + if throughput == 0 { + return 0, false + } + return throughput, true +} + +func (g *URLTestGroup) loadBandwidth(tag string) uint32 { + g.bandwidthAccess.Lock() + defer g.bandwidthAccess.Unlock() + if state := g.bandwidthHistory[tag]; state != nil { + return state.smoothed + } + return 0 +} + +// recordBandwidth appends a sample to the smoothing window and returns the new +// smoothed value. A failed probe is recorded as a zero sample rather than dropped, so +// that a genuinely broken path decays out of contention while a single transient +// failure is absorbed by the median. +func (g *URLTestGroup) recordBandwidth(tag string, throughput uint32) uint32 { + g.bandwidthAccess.Lock() + defer g.bandwidthAccess.Unlock() + state := g.bandwidthHistory[tag] + if state == nil { + state = new(bandwidthState) + g.bandwidthHistory[tag] = state + } + state.samples = append(state.samples, throughput) + if len(state.samples) > g.bandwidth.samples { + state.samples = state.samples[len(state.samples)-g.bandwidth.samples:] + } + state.lastTest = time.Now() + state.smoothed = medianThroughput(state.samples) + return state.smoothed +} + +func (g *URLTestGroup) bandwidthExpired(tag string) bool { + g.bandwidthAccess.Lock() + defer g.bandwidthAccess.Unlock() + state := g.bandwidthHistory[tag] + return state == nil || time.Since(state.lastTest) >= g.bandwidth.interval +} + +// resetBandwidth discards every sample, so selection falls back to latency until the +// next bandwidth sweep re-measures the new path. +func (g *URLTestGroup) resetBandwidth() { + if g.bandwidth == nil { + return + } + g.bandwidthAccess.Lock() + clear(g.bandwidthHistory) + g.bandwidthAccess.Unlock() +} + +// beatsIncumbent reports whether a challenger's throughput clears the incumbent's by +// the configured margin. The hysteresis is relative rather than absolute because +// throughput ratios are the meaningful comparison: a fixed byte-rate band would mean +// something entirely different on a slow link than on a fast one. +func beatsIncumbent(challenger uint32, incumbent uint32, tolerancePercent uint16) bool { + return uint64(challenger) > uint64(incumbent)*uint64(100+tolerancePercent)/100 +} + +func medianThroughput(samples []uint32) uint32 { + if len(samples) == 0 { + return 0 + } + sorted := slices.Clone(samples) + slices.Sort(sorted) + middle := len(sorted) / 2 + if len(sorted)%2 == 1 { + return sorted[middle] + } + return uint32((uint64(sorted[middle-1]) + uint64(sorted[middle])) / 2) +} + +func (g *URLTestGroup) loopBandwidthCheck(ticker *time.Ticker, closeChan <-chan struct{}) { + g.bandwidthTest(g.ctx, false) + for { + select { + case <-closeChan: + return + case <-ticker.C: + } + if time.Since(g.lastActive.Load()) > g.idleTimeout { + g.access.Lock() + if g.bandwidthTicker == ticker { + g.bandwidthTicker.Stop() + g.bandwidthTicker = nil + g.pause.UnregisterCallback(g.bandwidthPauseCallback) + g.bandwidthPauseCallback = nil + } + g.access.Unlock() + return + } + g.bandwidthTest(g.ctx, false) + } +} + +func (g *URLTestGroup) bandwidthTest(ctx context.Context, force bool) { + if g.bandwidth == nil { + return + } + if g.bandwidthChecking.Swap(true) { + return + } + defer g.bandwidthChecking.Store(false) + b, _ := batch.New(ctx, batch.WithConcurrencyNum[any](g.bandwidth.concurrency)) + checked := make(map[string]bool) + for _, detour := range g.outbounds { + tag := detour.Tag() + realTag := RealTag(detour, N.NetworkTCP) + if checked[realTag] { + continue + } + if !force && !g.bandwidthExpired(realTag) { + continue + } + checked[realTag] = true + p, loaded := g.outbound.Outbound(realTag) + if !loaded { + continue + } + b.Go(realTag, func() (any, error) { + testCtx, cancel := context.WithTimeout(g.ctx, g.bandwidth.timeout) + defer cancel() + var throughput, readBytes uint32 + result, err := urltest.BandwidthTest(testCtx, g.bandwidth.link, p, g.bandwidth.maxBytes, g.bandwidth.timeout) + if err != nil { + // Only the throughput sample is affected; the latency history is left + // alone, so a failed bandwidth probe cannot deselect an outbound that + // the latency probe still considers reachable. + g.logger.Debug("outbound ", tag, " bandwidth test failed: ", err) + } else { + throughput = result.Throughput() + readBytes = result.Bytes + g.logger.Debug("outbound ", tag, " bandwidth: ", throughput, " B/s over ", readBytes, " bytes") + } + g.history.StoreURLTestBandwidth(realTag, g.recordBandwidth(realTag, throughput), readBytes) + return nil, nil + }) + } + b.Wait() + g.performUpdateCheck() +} + func (g *URLTestGroup) loopCheck(ticker *time.Ticker, closeChan <-chan struct{}) { if time.Since(g.lastActive.Load()) > g.interval { g.lastActive.Store(time.Now()) @@ -495,6 +872,9 @@ func (b *urlTestBatch) test(outbounds []adapter.Outbound, link string, interval } func (g *URLTestGroup) performUpdateCheck() { + // The latency sweep and the bandwidth sweep have independent re-entrancy guards + // and can finish at the same time, so this needs its own lock against two + // concurrent writers to the selected outbounds. g.updateAccess.Lock() defer g.updateAccess.Unlock() var ( diff --git a/protocol/group/urltest_test.go b/protocol/group/urltest_test.go new file mode 100644 index 00000000000..532ef17fdaf --- /dev/null +++ b/protocol/group/urltest_test.go @@ -0,0 +1,155 @@ +package group + +import ( + "testing" + "time" + + C "github.com/sagernet/sing-box/constant" + "github.com/sagernet/sing-box/option" + "github.com/sagernet/sing/common/json/badoption" + + "github.com/stretchr/testify/require" +) + +func TestNewBandwidthTestOptionsDisabled(t *testing.T) { + t.Parallel() + parsed, err := newBandwidthTestOptions(nil, C.DefaultURLTestInterval) + require.NoError(t, err) + require.Nil(t, parsed) + + parsed, err = newBandwidthTestOptions(&option.URLTestBandwidthTestOptions{ + URL: "https://example.com/payload", + }, C.DefaultURLTestInterval) + require.NoError(t, err) + require.Nil(t, parsed, "an unset enabled flag must leave the feature inert") +} + +func TestNewBandwidthTestOptionsDefaults(t *testing.T) { + t.Parallel() + parsed, err := newBandwidthTestOptions(&option.URLTestBandwidthTestOptions{ + Enabled: true, + URL: "https://example.com/payload", + }, C.DefaultURLTestInterval) + require.NoError(t, err) + require.NotNil(t, parsed) + require.Equal(t, uint32(defaultBandwidthMaxBytes), parsed.maxBytes) + require.Equal(t, C.DefaultURLTestBandwidthTimeout, parsed.timeout) + require.Equal(t, defaultBandwidthConcurrency, parsed.concurrency) + require.Equal(t, uint16(defaultBandwidthTolerance), parsed.throughputTolerance) + require.Equal(t, defaultBandwidthSamples, parsed.samples) + // Latency ranking stays the default even once measurement is on. + require.Equal(t, C.URLTestStrategyLatency, parsed.strategy) + require.Zero(t, parsed.latencyFloor) + // The throughput cadence derives from the latency cadence rather than matching it. + require.Equal(t, C.DefaultURLTestInterval*defaultBandwidthIntervalMultiplier, parsed.interval) +} + +func TestNewBandwidthTestOptionsDefaultURL(t *testing.T) { + t.Parallel() + parsed, err := newBandwidthTestOptions(&option.URLTestBandwidthTestOptions{ + Enabled: true, + }, C.DefaultURLTestInterval) + require.NoError(t, err) + require.Equal(t, C.DefaultURLTestBandwidthURLPrefix+"262144", parsed.link) + + // The default endpoint is sized to the cap, so raising max_bytes must raise the + // request too rather than leaving the probe short of its own limit. + parsed, err = newBandwidthTestOptions(&option.URLTestBandwidthTestOptions{ + Enabled: true, + MaxBytes: 512 * 1024, + }, C.DefaultURLTestInterval) + require.NoError(t, err) + require.Equal(t, C.DefaultURLTestBandwidthURLPrefix+"524288", parsed.link) + + // An explicit URL is never overridden. + parsed, err = newBandwidthTestOptions(&option.URLTestBandwidthTestOptions{ + Enabled: true, + URL: "https://example.com/payload", + }, C.DefaultURLTestInterval) + require.NoError(t, err) + require.Equal(t, "https://example.com/payload", parsed.link) +} + +func TestNewBandwidthTestOptionsValidation(t *testing.T) { + t.Parallel() + for name, options := range map[string]*option.URLTestBandwidthTestOptions{ + "max bytes above cap": {MaxBytes: maxBandwidthMaxBytes + 1}, + "unknown strategy": {Strategy: "fastest"}, + "negative timeout": {Timeout: badoption.Duration(-time.Second)}, + "negative interval": {Interval: badoption.Duration(-time.Second)}, + "negative concurrency": {Concurrency: -1}, + "negative samples": {Samples: -1}, + } { + t.Run(name, func(t *testing.T) { + options.Enabled = true + options.URL = "https://example.com/payload" + _, err := newBandwidthTestOptions(options, C.DefaultURLTestInterval) + require.Error(t, err) + }) + } +} + +func TestNewBandwidthTestOptionsLatencyFloor(t *testing.T) { + t.Parallel() + parsed, err := newBandwidthTestOptions(&option.URLTestBandwidthTestOptions{ + Enabled: true, + URL: "https://example.com/payload", + Strategy: C.URLTestStrategyThroughputWithLatencyFloor, + LatencyFloor: badoption.Duration(400 * time.Millisecond), + }, C.DefaultURLTestInterval) + require.NoError(t, err) + require.Equal(t, uint16(400), parsed.latencyFloor) + + // Delay is stored as uint16 milliseconds, so a larger floor is clamped rather + // than wrapping around into an absurdly small one. + parsed, err = newBandwidthTestOptions(&option.URLTestBandwidthTestOptions{ + Enabled: true, + URL: "https://example.com/payload", + Strategy: C.URLTestStrategyThroughputWithLatencyFloor, + LatencyFloor: badoption.Duration(2 * time.Minute), + }, C.DefaultURLTestInterval) + require.NoError(t, err) + require.Equal(t, uint16(65535), parsed.latencyFloor) +} + +func TestMedianThroughput(t *testing.T) { + t.Parallel() + require.Zero(t, medianThroughput(nil)) + require.Equal(t, uint32(100), medianThroughput([]uint32{100})) + require.Equal(t, uint32(150), medianThroughput([]uint32{100, 200})) + require.Equal(t, uint32(200), medianThroughput([]uint32{100, 200, 300})) + // The point of smoothing: one transient failure among good samples must not + // evict an otherwise healthy outbound. + require.Equal(t, uint32(900), medianThroughput([]uint32{1000, 0, 900})) + // Sustained failure does decay it out of contention. + require.Zero(t, medianThroughput([]uint32{1000, 0, 0})) +} + +func TestBeatsIncumbent(t *testing.T) { + t.Parallel() + // A challenger inside the tolerance band leaves the incumbent in place, so the + // group does not flap and repeatedly interrupt live connections. + require.False(t, beatsIncumbent(1200, 1000, 25)) + require.False(t, beatsIncumbent(1250, 1000, 25)) + require.True(t, beatsIncumbent(1251, 1000, 25)) + require.True(t, beatsIncumbent(1001, 1000, 0)) + require.False(t, beatsIncumbent(1000, 1000, 0)) +} + +func TestRecordBandwidthWindow(t *testing.T) { + t.Parallel() + group := &URLTestGroup{ + bandwidth: &bandwidthTestOptions{samples: 3}, + bandwidthHistory: make(map[string]*bandwidthState), + } + require.Equal(t, uint32(100), group.recordBandwidth("proxy", 100)) + require.Equal(t, uint32(150), group.recordBandwidth("proxy", 200)) + require.Equal(t, uint32(200), group.recordBandwidth("proxy", 300)) + // The window slides rather than growing, so old samples stop counting. + require.Equal(t, uint32(300), group.recordBandwidth("proxy", 400)) + require.Len(t, group.bandwidthHistory["proxy"].samples, 3) + require.Equal(t, uint32(300), group.loadBandwidth("proxy")) + + group.resetBandwidth() + require.Zero(t, group.loadBandwidth("proxy")) +} diff --git a/test/urltest_test.go b/test/urltest_test.go new file mode 100644 index 00000000000..550ddaf97d6 --- /dev/null +++ b/test/urltest_test.go @@ -0,0 +1,436 @@ +package main + +import ( + "context" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "net/netip" + "strconv" + "sync/atomic" + "testing" + "time" + + "github.com/sagernet/sing-box" + "github.com/sagernet/sing-box/adapter" + C "github.com/sagernet/sing-box/constant" + "github.com/sagernet/sing-box/option" + "github.com/sagernet/sing-box/protocol/group" + "github.com/sagernet/sing/common" + "github.com/sagernet/sing/common/json/badoption" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" + "github.com/sagernet/sing/protocol/socks" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Each test gets its own ports. Reusing a fixed set across sequentially started +// instances is flaky: a closing instance can still answer on a port the next test has +// just bound, and its probes then fail against a backend that is already gone. +var urlTestPortCursor atomic.Uint32 + +type urlTestEnv struct { + clientPort uint16 + serverPortA uint16 + serverPortB uint16 + clashPort uint16 +} + +func newURLTestEnv() urlTestEnv { + base := uint16(19000 + urlTestPortCursor.Add(10)) + return urlTestEnv{ + clientPort: base, + serverPortA: base + 1, + serverPortB: base + 2, + clashPort: base + 3, + } +} + +const ( + // The probe URLs point at a placeholder destination; each proxy server rewrites + // it to its own backend with a route action, which is what gives the two paths + // independently controllable characteristics. + urlTestLatencyURL = "http://127.0.0.1:1/latency" + urlTestBandwidthURL = "http://127.0.0.1:1/payload" + + urlTestProbeInterval = 500 * time.Millisecond + urlTestMaxProbeBytes = 64 * 1024 + urlTestPayloadChunk = 8 * 1024 + urlTestPayloadChunks = 64 // 512 KiB available, well above the probe cap + urlTestEventually = 15 * time.Second + urlTestEventuallyTick = 100 * time.Millisecond + + // Both delays are non-zero on purpose. Select treats a delay of 0 as "no + // incumbent yet" (protocol/group/urltest.go, minDelay == 0), and a loopback probe + // really can measure 0 ms, which lets the slower outbound override the faster + // one. Keep every backend above a millisecond. + urlTestFastDelay = 50 * time.Millisecond + urlTestSlowDelay = 350 * time.Millisecond +) + +// payloadServer stands in for the endpoint a urltest group probes. Its time to +// response headers and its transfer rate are controlled separately, because the whole +// premise of bandwidth-aware selection is that those two properties decouple: a shaped +// path answers quickly and then crawls. +type payloadServer struct { + server *httptest.Server + headerDelay atomic.Int64 + chunkDelay atomic.Int64 +} + +func startPayloadServer(t *testing.T, headerDelay time.Duration, chunkDelay time.Duration) *payloadServer { + s := new(payloadServer) + s.headerDelay.Store(int64(headerDelay)) + s.chunkDelay.Store(int64(chunkDelay)) + s.server = httptest.NewServer(http.HandlerFunc(s.handle)) + t.Cleanup(s.server.Close) + return s +} + +func (s *payloadServer) handle(w http.ResponseWriter, r *http.Request) { + if delay := time.Duration(s.headerDelay.Load()); delay > 0 { + time.Sleep(delay) + } + if r.URL.Path == "/latency" { + w.WriteHeader(http.StatusNoContent) + return + } + w.Header().Set("Content-Type", "application/octet-stream") + w.WriteHeader(http.StatusOK) + flusher, canFlush := w.(http.Flusher) + chunk := make([]byte, urlTestPayloadChunk) + chunkDelay := time.Duration(s.chunkDelay.Load()) + for range urlTestPayloadChunks { + if _, err := w.Write(chunk); err != nil { + return + } + if canFlush { + flusher.Flush() + } + if chunkDelay > 0 { + time.Sleep(chunkDelay) + } + } +} + +func (s *payloadServer) port(t *testing.T) uint16 { + _, portString, err := net.SplitHostPort(s.server.Listener.Addr().String()) + require.NoError(t, err) + port, err := strconv.ParseUint(portString, 10, 16) + require.NoError(t, err) + return uint16(port) +} + +func (s *payloadServer) setHeaderDelay(delay time.Duration) { + s.headerDelay.Store(int64(delay)) +} + +// urlTestOptions builds a self-contained topology: two shadowsocks servers in the same +// instance, each routing the probe to its own backend, and a urltest group over the two +// matching outbounds. +func urlTestOptions(t *testing.T, env urlTestEnv, backendA *payloadServer, backendB *payloadServer, bandwidth *option.URLTestBandwidthTestOptions, clashAPI bool) option.Options { + options := option.Options{ + Inbounds: []option.Inbound{ + { + Type: C.TypeMixed, + Tag: "mixed-in", + Options: &option.HTTPMixedInboundOptions{ + ListenOptions: option.ListenOptions{ + Listen: common.Ptr(badoption.Addr(netip.AddrFrom4([4]byte{127, 0, 0, 1}))), + ListenPort: env.clientPort, + }, + }, + }, + { + Type: C.TypeShadowsocks, + Tag: "ss-in-a", + Options: &option.ShadowsocksInboundOptions{ + ListenOptions: option.ListenOptions{ + Listen: common.Ptr(badoption.Addr(netip.AddrFrom4([4]byte{127, 0, 0, 1}))), + ListenPort: env.serverPortA, + }, + Method: "none", + }, + }, + { + Type: C.TypeShadowsocks, + Tag: "ss-in-b", + Options: &option.ShadowsocksInboundOptions{ + ListenOptions: option.ListenOptions{ + Listen: common.Ptr(badoption.Addr(netip.AddrFrom4([4]byte{127, 0, 0, 1}))), + ListenPort: env.serverPortB, + }, + Method: "none", + }, + }, + }, + Outbounds: []option.Outbound{ + { + Type: C.TypeDirect, + Tag: "direct", + }, + { + Type: C.TypeShadowsocks, + Tag: "proxy-a", + Options: &option.ShadowsocksOutboundOptions{ + ServerOptions: option.ServerOptions{ + Server: "127.0.0.1", + ServerPort: env.serverPortA, + }, + Method: "none", + }, + }, + { + Type: C.TypeShadowsocks, + Tag: "proxy-b", + Options: &option.ShadowsocksOutboundOptions{ + ServerOptions: option.ServerOptions{ + Server: "127.0.0.1", + ServerPort: env.serverPortB, + }, + Method: "none", + }, + }, + { + Type: C.TypeURLTest, + Tag: "auto", + Options: &option.URLTestOutboundOptions{ + Outbounds: []string{"proxy-a", "proxy-b"}, + URL: urlTestLatencyURL, + Interval: badoption.Duration(urlTestProbeInterval), + IdleTimeout: badoption.Duration(time.Minute), + BandwidthTest: bandwidth, + }, + }, + }, + Route: &option.RouteOptions{ + Rules: []option.Rule{ + urlTestRouteRule("mixed-in", "auto", 0), + urlTestRouteRule("ss-in-a", "direct", backendA.port(t)), + urlTestRouteRule("ss-in-b", "direct", backendB.port(t)), + }, + }, + } + if clashAPI { + options.Experimental = &option.ExperimentalOptions{ + ClashAPI: &option.ClashAPIOptions{ + ExternalController: "127.0.0.1:" + strconv.Itoa(int(env.clashPort)), + }, + } + } + return options +} + +func urlTestRouteRule(inbound string, outbound string, overridePort uint16) option.Rule { + action := option.RouteActionOptions{ + Outbound: outbound, + } + if overridePort != 0 { + action.OverrideAddress = "127.0.0.1" + action.OverridePort = overridePort + } + return option.Rule{ + Type: C.RuleTypeDefault, + DefaultOptions: option.DefaultRule{ + RawDefaultRule: option.RawDefaultRule{ + Inbound: []string{inbound}, + }, + RuleAction: option.RuleAction{ + Action: C.RuleActionTypeRoute, + RouteOptions: action, + }, + }, + } +} + +func urlTestOutbound(t *testing.T, instance *box.Box) *group.URLTest { + t.Helper() + outbound, loaded := instance.Outbound().Outbound("auto") + require.True(t, loaded, "urltest outbound not found") + urlTest, isURLTest := outbound.(*group.URLTest) + require.True(t, isURLTest, "outbound is not a urltest group") + return urlTest +} + +func selectedOutbound(t *testing.T, instance *box.Box) string { + t.Helper() + return urlTestOutbound(t, instance).Selected(N.NetworkTCP).Tag() +} + +// touchGroup sends one connection through the group, which is what starts its probe +// tickers: an untouched group stays suspended and never probes. +func touchGroup(t *testing.T, env urlTestEnv, backend *payloadServer) { + t.Helper() + dialer := socks.NewClient(N.SystemDialer, M.ParseSocksaddrHostPort("127.0.0.1", env.clientPort), socks.Version5, "", "") + conn, err := dialer.DialContext(context.Background(), N.NetworkTCP, M.ParseSocksaddrHostPort("127.0.0.1", backend.port(t))) + if err == nil { + conn.Close() + } +} + +func requireSelected(t *testing.T, instance *box.Box, expected string) { + t.Helper() + var last string + if !assert.Eventually(t, func() bool { + last = selectedOutbound(t, instance) + return last == expected + }, urlTestEventually, urlTestEventuallyTick) { + // Eventually evaluates its message eagerly, so report the last observed + // value separately rather than the one from before polling started. + require.FailNowf(t, "selection did not converge", "expected %s, last saw %q", expected, last) + } +} + +// TestURLTestSelectsLowerLatency is the baseline the group has always promised, and +// which had no coverage before this file. +func TestURLTestSelectsLowerLatency(t *testing.T) { + env := newURLTestEnv() + backendFast := startPayloadServer(t, urlTestFastDelay, 0) + backendSlow := startPayloadServer(t, urlTestSlowDelay, 0) + instance := startInstance(t, urlTestOptions(t, env, backendFast, backendSlow, nil, false)) + // Touch the group so its ticker runs. An untouched group probes once at + // PostStart and never retries, which makes a single transient failure permanent. + touchGroup(t, env, backendFast) + requireSelected(t, instance, "proxy-a") +} + +// TestURLTestDropsFailedOutbound covers the failure path: a probe error deletes the +// history entry, which must move selection to the surviving outbound. +func TestURLTestDropsFailedOutbound(t *testing.T) { + env := newURLTestEnv() + backendFast := startPayloadServer(t, urlTestFastDelay, 0) + backendSlow := startPayloadServer(t, urlTestSlowDelay, 0) + instance := startInstance(t, urlTestOptions(t, env, backendFast, backendSlow, nil, false)) + // Touch the group so its ticker runs. An untouched group probes once at + // PostStart and never retries, which makes a single transient failure permanent. + touchGroup(t, env, backendFast) + requireSelected(t, instance, "proxy-a") + + // Take the selected path's backend away; its probe now fails while the other + // still answers. + backendFast.server.Close() + urlTestOutbound(t, instance).CheckOutbounds() + requireSelected(t, instance, "proxy-b") +} + +// TestURLTestToleranceHoldsIncumbent guards against the group flapping between two +// outbounds whose latencies are close, which would churn connections. +func TestURLTestToleranceHoldsIncumbent(t *testing.T) { + env := newURLTestEnv() + backendA := startPayloadServer(t, urlTestFastDelay, 0) + backendB := startPayloadServer(t, urlTestSlowDelay, 0) + options := urlTestOptions(t, env, backendA, backendB, nil, false) + options.Outbounds[3].Options.(*option.URLTestOutboundOptions).Tolerance = 400 + instance := startInstance(t, options) + touchGroup(t, env, backendA) + requireSelected(t, instance, "proxy-a") + + // Make B faster, but by less than the tolerance. The incumbent must hold. + backendA.setHeaderDelay(250 * time.Millisecond) + backendB.setHeaderDelay(urlTestFastDelay) + for range 3 { + urlTestOutbound(t, instance).CheckOutbounds() + } + require.Equal(t, "proxy-a", selectedOutbound(t, instance)) +} + +func TestURLTestReportsAllDelays(t *testing.T) { + env := newURLTestEnv() + backendA := startPayloadServer(t, urlTestFastDelay, 0) + backendB := startPayloadServer(t, 150*time.Millisecond, 0) + instance := startInstance(t, urlTestOptions(t, env, backendA, backendB, nil, false)) + + // URLTest reports only the outbounds it actually re-probed: an entry newer than + // the interval is skipped, and PostStart has just populated both. Wait it out so + // the sweep has something to do. + time.Sleep(urlTestProbeInterval + 200*time.Millisecond) + + result, err := urlTestOutbound(t, instance).URLTest(context.Background()) + require.NoError(t, err) + require.Contains(t, result, "proxy-a") + require.Contains(t, result, "proxy-b") +} + +// TestURLTestBandwidthMeasuresThroughput checks the metric end to end, through a real +// encrypted proxy, and asserts it reaches the Clash API where clients can read it. +func TestURLTestBandwidthMeasuresThroughput(t *testing.T) { + env := newURLTestEnv() + backendA := startPayloadServer(t, urlTestFastDelay, 0) + backendB := startPayloadServer(t, urlTestFastDelay, 0) + instance := startInstance(t, urlTestOptions(t, env, backendA, backendB, &option.URLTestBandwidthTestOptions{ + Enabled: true, + URL: urlTestBandwidthURL, + MaxBytes: urlTestMaxProbeBytes, + Interval: badoption.Duration(urlTestProbeInterval), + Timeout: badoption.Duration(10 * time.Second), + }, true)) + touchGroup(t, env, backendA) + + require.Eventually(t, func() bool { + history := clashProxyHistory(t, env, "proxy-a") + return history != nil && history.Throughput > 0 && history.Bytes > 0 + }, urlTestEventually, urlTestEventuallyTick, "throughput never appeared on the Clash API") + + history := clashProxyHistory(t, env, "proxy-a") + // The cap is a hard limit on what a single probe reads. + require.LessOrEqual(t, history.Bytes, uint32(urlTestMaxProbeBytes)) + // Latency must survive alongside it rather than being overwritten. + require.NotZero(t, history.Time) + // Measuring throughput must not disturb selection under the default strategy. + require.NotEmpty(t, selectedOutbound(t, instance)) +} + +// TestURLTestBandwidthStrategyRanking is the issue's own table made executable: the two +// paths disagree about which is better depending on which property you measure, and the +// strategy decides who wins. +func TestURLTestBandwidthStrategyRanking(t *testing.T) { + for _, testCase := range []struct { + strategy string + expected string + }{ + // proxy-a answers fastest but then crawls. + {C.URLTestStrategyLatency, "proxy-a"}, + // proxy-b is slower to first byte but actually moves data. + {C.URLTestStrategyThroughput, "proxy-b"}, + } { + t.Run(testCase.strategy, func(t *testing.T) { + env := newURLTestEnv() + shaped := startPayloadServer(t, urlTestFastDelay, 40*time.Millisecond) + quick := startPayloadServer(t, 250*time.Millisecond, 0) + instance := startInstance(t, urlTestOptions(t, env, shaped, quick, &option.URLTestBandwidthTestOptions{ + Enabled: true, + URL: urlTestBandwidthURL, + MaxBytes: urlTestMaxProbeBytes, + Interval: badoption.Duration(urlTestProbeInterval), + Timeout: badoption.Duration(20 * time.Second), + Strategy: testCase.strategy, + // One sample, so the assertion does not wait for a median to fill. + Samples: 1, + }, false)) + touchGroup(t, env, shaped) + requireSelected(t, instance, testCase.expected) + }) + } +} + +func clashProxyHistory(t *testing.T, env urlTestEnv, tag string) *adapter.URLTestHistory { + t.Helper() + response, err := http.Get("http://127.0.0.1:" + strconv.Itoa(int(env.clashPort)) + "/proxies/" + tag) + if err != nil { + return nil + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return nil + } + var proxy struct { + History []*adapter.URLTestHistory `json:"history"` + } + if json.NewDecoder(response.Body).Decode(&proxy) != nil || len(proxy.History) == 0 { + return nil + } + return proxy.History[0] +} From c9e081f4243486232134c4216ed5e9e98b99ec93 Mon Sep 17 00:00:00 2001 From: zydo Date: Sat, 22 Aug 2026 17:58:11 -0700 Subject: [PATCH 2/2] Bound urltest bandwidth rounds against stuck probes A probe parked in a read that ignores its context deadline would leave bandwidthChecking set forever, silently disabling every later sweep. Wait on the batch with a round budget of the worst legitimate time plus one wave of slack, log and abandon the round when it expires, and drop samples from in-flight probes when resetBandwidth invalidates them. --- protocol/group/urltest.go | 61 +++++++++++-- protocol/group/urltest_test.go | 157 ++++++++++++++++++++++++++++++++- 2 files changed, 205 insertions(+), 13 deletions(-) diff --git a/protocol/group/urltest.go b/protocol/group/urltest.go index d7dc27db7ed..8d0d08d1e75 100644 --- a/protocol/group/urltest.go +++ b/protocol/group/urltest.go @@ -246,6 +246,10 @@ type URLTestGroup struct { bandwidthPauseCallback *list.Element[pause.Callback] bandwidthAccess sync.Mutex bandwidthHistory map[string]*bandwidthState + // bandwidthEpoch counts resetBandwidth calls. Workers capture it before probing + // so a sample measured on a previous network path is dropped instead of written + // back into the freshly cleared history (guarded by bandwidthAccess). + bandwidthEpoch uint64 } // bandwidthState holds the smoothing window for one outbound. Throughput samples are @@ -592,13 +596,26 @@ func (g *URLTestGroup) loadBandwidth(tag string) uint32 { return 0 } -// recordBandwidth appends a sample to the smoothing window and returns the new -// smoothed value. A failed probe is recorded as a zero sample rather than dropped, so -// that a genuinely broken path decays out of contention while a single transient -// failure is absorbed by the median. -func (g *URLTestGroup) recordBandwidth(tag string, throughput uint32) uint32 { +// currentBandwidthEpoch returns the sample generation counter. A worker captures it +// before probing and hands it to recordBandwidthChecked, which drops the sample if the +// generation has since been reset. +func (g *URLTestGroup) currentBandwidthEpoch() uint64 { g.bandwidthAccess.Lock() defer g.bandwidthAccess.Unlock() + return g.bandwidthEpoch +} + +// recordBandwidthChecked appends a sample to the smoothing window and returns the new +// smoothed value, unless the probe started before the last resetBandwidth — a sample +// measured on the previous network says nothing about the current one. A failed probe +// is recorded as a zero sample rather than dropped, so that a genuinely broken path +// decays out of contention while a single transient failure is absorbed by the median. +func (g *URLTestGroup) recordBandwidthChecked(epoch uint64, tag string, throughput uint32) (uint32, bool) { + g.bandwidthAccess.Lock() + defer g.bandwidthAccess.Unlock() + if epoch != g.bandwidthEpoch { + return 0, false + } state := g.bandwidthHistory[tag] if state == nil { state = new(bandwidthState) @@ -610,7 +627,7 @@ func (g *URLTestGroup) recordBandwidth(tag string, throughput uint32) uint32 { } state.lastTest = time.Now() state.smoothed = medianThroughput(state.samples) - return state.smoothed + return state.smoothed, true } func (g *URLTestGroup) bandwidthExpired(tag string) bool { @@ -621,13 +638,15 @@ func (g *URLTestGroup) bandwidthExpired(tag string) bool { } // resetBandwidth discards every sample, so selection falls back to latency until the -// next bandwidth sweep re-measures the new path. +// next bandwidth sweep re-measures the new path. Bumping the epoch also invalidates +// in-flight probes, whose samples belong to the previous path. func (g *URLTestGroup) resetBandwidth() { if g.bandwidth == nil { return } g.bandwidthAccess.Lock() clear(g.bandwidthHistory) + g.bandwidthEpoch++ g.bandwidthAccess.Unlock() } @@ -685,6 +704,7 @@ func (g *URLTestGroup) bandwidthTest(ctx context.Context, force bool) { defer g.bandwidthChecking.Store(false) b, _ := batch.New(ctx, batch.WithConcurrencyNum[any](g.bandwidth.concurrency)) checked := make(map[string]bool) + var workers int for _, detour := range g.outbounds { tag := detour.Tag() realTag := RealTag(detour, N.NetworkTCP) @@ -699,7 +719,9 @@ func (g *URLTestGroup) bandwidthTest(ctx context.Context, force bool) { if !loaded { continue } + workers++ b.Go(realTag, func() (any, error) { + epoch := g.currentBandwidthEpoch() testCtx, cancel := context.WithTimeout(g.ctx, g.bandwidth.timeout) defer cancel() var throughput, readBytes uint32 @@ -714,11 +736,32 @@ func (g *URLTestGroup) bandwidthTest(ctx context.Context, force bool) { readBytes = result.Bytes g.logger.Debug("outbound ", tag, " bandwidth: ", throughput, " B/s over ", readBytes, " bytes") } - g.history.StoreURLTestBandwidth(realTag, g.recordBandwidth(realTag, throughput), readBytes) + if smoothed, recorded := g.recordBandwidthChecked(epoch, realTag, throughput); recorded { + g.history.StoreURLTestBandwidth(realTag, smoothed, readBytes) + } return nil, nil }) } - b.Wait() + // Every worker carries its own deadline, but #4255 shows a dialer can park in a + // read that ignores both the context deadline and the client timeout. batch.Wait + // has no deadline of its own, so one such worker would leave bandwidthChecking + // set forever and silently disable every later sweep. Bound the round by the time + // the queued workers legitimately need at worst — ceil(workers/concurrency) + // timeout-sized waves, plus one wave of slack — and abandon it when even that + // expires. The stuck worker's goroutine outlives the round either way; giving up + // on it costs one leaked goroutine instead of the whole feature. + waves := (workers + g.bandwidth.concurrency - 1) / g.bandwidth.concurrency + budget := time.Duration(waves+1) * g.bandwidth.timeout + waitDone := make(chan struct{}) + go func() { + b.Wait() + close(waitDone) + }() + select { + case <-waitDone: + case <-time.After(budget): + g.logger.Warn("urltest: bandwidth round exceeded ", budget, "; a probe is ignoring its deadline, abandoning this round") + } g.performUpdateCheck() } diff --git a/protocol/group/urltest_test.go b/protocol/group/urltest_test.go index 532ef17fdaf..7f351f74558 100644 --- a/protocol/group/urltest_test.go +++ b/protocol/group/urltest_test.go @@ -1,12 +1,24 @@ package group import ( + "context" + "net" + "sync" + "sync/atomic" "testing" "time" + "github.com/sagernet/sing-box/adapter" + "github.com/sagernet/sing-box/adapter/outbound" + "github.com/sagernet/sing-box/common/urltest" C "github.com/sagernet/sing-box/constant" + "github.com/sagernet/sing-box/log" "github.com/sagernet/sing-box/option" + E "github.com/sagernet/sing/common/exceptions" "github.com/sagernet/sing/common/json/badoption" + M "github.com/sagernet/sing/common/metadata" + N "github.com/sagernet/sing/common/network" + "github.com/sagernet/sing/service" "github.com/stretchr/testify/require" ) @@ -142,14 +154,151 @@ func TestRecordBandwidthWindow(t *testing.T) { bandwidth: &bandwidthTestOptions{samples: 3}, bandwidthHistory: make(map[string]*bandwidthState), } - require.Equal(t, uint32(100), group.recordBandwidth("proxy", 100)) - require.Equal(t, uint32(150), group.recordBandwidth("proxy", 200)) - require.Equal(t, uint32(200), group.recordBandwidth("proxy", 300)) + epoch := group.currentBandwidthEpoch() + record := func(throughput uint32) uint32 { + smoothed, recorded := group.recordBandwidthChecked(epoch, "proxy", throughput) + require.True(t, recorded) + return smoothed + } + require.Equal(t, uint32(100), record(100)) + require.Equal(t, uint32(150), record(200)) + require.Equal(t, uint32(200), record(300)) // The window slides rather than growing, so old samples stop counting. - require.Equal(t, uint32(300), group.recordBandwidth("proxy", 400)) + require.Equal(t, uint32(300), record(400)) require.Len(t, group.bandwidthHistory["proxy"].samples, 3) require.Equal(t, uint32(300), group.loadBandwidth("proxy")) group.resetBandwidth() require.Zero(t, group.loadBandwidth("proxy")) + // A probe that started before the reset measured the previous network path; + // its sample must be dropped instead of written back into the cleared history. + _, recorded := group.recordBandwidthChecked(epoch, "proxy", 500) + require.False(t, recorded) + require.Zero(t, group.loadBandwidth("proxy")) + // A probe started after the reset records normally. + newEpoch := group.currentBandwidthEpoch() + require.Equal(t, epoch+1, newEpoch) + smoothed, recorded := group.recordBandwidthChecked(newEpoch, "proxy", 600) + require.True(t, recorded) + require.Equal(t, uint32(600), smoothed) +} + +// hangingDialOutbound reproduces the #4255 failure mode: a dialer that parks +// forever while ignoring the context deadline it was handed. +type hangingDialOutbound struct { + outbound.Adapter + block chan struct{} + release sync.Once + dials atomic.Int32 + returned atomic.Int32 +} + +func (o *hangingDialOutbound) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) { + o.dials.Add(1) + <-o.block + o.returned.Add(1) + return nil, E.New("dial released") +} + +func (o *hangingDialOutbound) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) { + return nil, E.New("not implemented") +} + +// unblock lets parked dials return; safe to call more than once. +func (o *hangingDialOutbound) unblock() { + o.release.Do(func() { close(o.block) }) +} + +type staticOutboundManager struct { + outbounds []adapter.Outbound +} + +func (m *staticOutboundManager) Start(stage adapter.StartStage, scope *adapter.Scope) error { + return nil +} +func (m *staticOutboundManager) Close() error { return nil } +func (m *staticOutboundManager) Outbounds() []adapter.Outbound { return m.outbounds } +func (m *staticOutboundManager) Default() adapter.Outbound { return m.outbounds[0] } +func (m *staticOutboundManager) Remove(tag string) error { return nil } +func (m *staticOutboundManager) Outbound(tag string) (adapter.Outbound, bool) { + for _, detour := range m.outbounds { + if detour.Tag() == tag { + return detour, true + } + } + return nil, false +} + +func (m *staticOutboundManager) Create(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, outboundType string, options any) error { + return nil +} + +func newHangingBandwidthGroup(t *testing.T) (*URLTestGroup, *hangingDialOutbound) { + hang := &hangingDialOutbound{ + Adapter: outbound.NewAdapter(C.TypeDirect, "hang", []string{N.NetworkTCP, N.NetworkUDP}, nil), + block: make(chan struct{}), + } + ctx := service.ContextWithPtr[urltest.HistoryStorage](context.Background(), urltest.NewHistoryStorage()) + group, err := NewURLTestGroup( + ctx, &staticOutboundManager{outbounds: []adapter.Outbound{hang}}, + log.NewNOPFactory().Logger(), []adapter.Outbound{hang}, + "https://example.com/payload", 50*time.Millisecond, 0, time.Second, false, + &option.URLTestBandwidthTestOptions{ + Enabled: true, + URL: "https://example.com/payload", + Timeout: badoption.Duration(50 * time.Millisecond), + }, + ) + require.NoError(t, err) + t.Cleanup(hang.unblock) + return group, hang +} + +func TestBandwidthRoundCeilingReleasesGuard(t *testing.T) { + t.Parallel() + group, hang := newHangingBandwidthGroup(t) + + // One worker and default concurrency give a budget of two probe timeouts; the + // dial ignores its deadline, so the round must be abandoned at the budget + // instead of wedging on batch.Wait() forever. + done := make(chan struct{}) + go func() { + group.bandwidthTest(context.Background(), true) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("bandwidth round did not return under its budget") + } + + // The re-entrancy guard went down with the round: a second forced round spawns + // a worker and dials again instead of being silently skipped. + group.bandwidthTest(context.Background(), true) + require.Eventually(t, func() bool { + return hang.dials.Load() == 2 + }, 2*time.Second, 5*time.Millisecond) +} + +func TestBandwidthAbandonedSampleDroppedAfterReset(t *testing.T) { + t.Parallel() + group, hang := newHangingBandwidthGroup(t) + + // A round is abandoned while its worker hangs, then the network changes. + group.bandwidthTest(context.Background(), true) + require.Eventually(t, func() bool { + return hang.dials.Load() == 1 + }, 2*time.Second, 5*time.Millisecond) + group.resetBandwidth() + + // When the abandoned worker finally unblocks, its sample belongs to the + // previous network path and must not surface in the cleared history. + hang.unblock() + require.Eventually(t, func() bool { + return hang.returned.Load() == 1 + }, 2*time.Second, 5*time.Millisecond) + // The dial has returned; give the worker a moment to run its record path, + // then require that no sample survived the reset. + time.Sleep(100 * time.Millisecond) + require.Empty(t, group.bandwidthHistory) }