feat(urltest): add network quality speed selection

This commit is contained in:
2026-10-03 23:45:14 +03:00
parent de0da6889d
commit c4210e2fe2
27 changed files with 2868 additions and 68 deletions
+1 -1
View File
@@ -1 +1 @@
1.14.1-rr-edition-2.7.2-r1
1.14.1-rr-edition-2.7.2-r3
+349
View File
@@ -0,0 +1,349 @@
# Карта проекта sing-box-rr-edition
## Назначение
Репозиторий упаковывает RR Edition — урезанный для OpenWrt форк
`sing-box-extended`. Основной сценарий — network-only/tproxy-ориентированный
`sing-box` для роутеров с пакетами IPK (OpenWrt 24.10) и APK (OpenWrt 25.12).
Это не маленькая самостоятельная реализация: каталог `sing-box-rr/` содержит
почти полный upstream-код `github.com/sagernet/sing-box`, а профиль RR Edition
задаётся прежде всего registry, build tags и упаковкой. Наличие исходника в
дереве не означает, что функция доступна в пакетном бинарнике.
Текущий снимок при составлении карты:
- ветка: `extended`;
- версия пакета: `1.14.1-rr-edition-2.7.2-r3`;
- Go-модуль: `github.com/sagernet/sing-box`;
- требование `go.mod`: Go 1.26.4;
- пакетные build tags: `with_quic,with_utls,with_clash_api`.
## Верхний уровень
- `VERSION` — единственный источник версии и OpenWrt `r`-релиза.
- `SOURCE_DATE_EPOCH` — timestamp для воспроизводимой упаковки.
- `build-ipk.sh` — локальная сборка arm64 IPK для OpenWrt 24.10.
- `build-apk.sh` — локальная сборка arm64/x86_64 APK для OpenWrt 25.12.
- `packages/` — закоммиченные готовые пакеты по версии OpenWrt и архитектуре.
- `sing-box-rr/` — Go-код, OpenWrt feed `Makefile`, release-конфигурация и
унаследованные upstream workflows.
- `README.md` — руководство по урезанию и упаковке; отдельные детали в нём
устарели, см. раздел «Оговорки» ниже.
## Главные точки изменения
| Задача | Источник истины |
|---|---|
| Изменить версию/`rN` | `VERSION` |
| Изменить возможности OpenWrt-пакета | `sing-box-rr/release/DEFAULT_BUILD_TAGS_OTHERS` |
| Добавить/убрать runtime-тип | `sing-box-rr/include/registry.go` и соответствующий `include/*_stub.go` |
| Изменить JSON surface и валидацию | `sing-box-rr/option/` |
| Изменить сборку runtime-графа/порядок запуска | `sing-box-rr/box.go` |
| Изменить CLI | `sing-box-rr/cmd/sing-box/` |
| Изменить маршрутизацию | `sing-box-rr/route/` |
| Изменить DNS | `sing-box-rr/dns/` |
| Изменить протокол | `sing-box-rr/protocol/` и регистрация в `include/` |
| Изменить OpenWrt service/config | `sing-box-rr/release/config/` |
| Изменить feed package metadata | `sing-box-rr/Makefile` |
| Изменить локальную упаковку | `build-ipk.sh` и `build-apk.sh` |
Держать поведение в одном месте: сначала менять build tag/registry, затем
config surface и только после стабилизации удалять физические каталоги.
## Карта Go-кода
- `cmd/sing-box/` — Cobra CLI: `run`, `check`, `format`, `merge`, `generate`,
`rule-set`, `tools`, `api`, `version` и вспомогательные команды.
- `box.go` — composition root. Создаёт и связывает все менеджеры, а также
владеет фазами `PreStart`, `Start` и `Close`.
- `include/` — связывает интерфейсы с конкретными реализациями. Здесь registry
и build-tag пары «реализация/stub».
- `adapter/` — интерфейсы и менеджеры endpoint/inbound/outbound/provider/
service/certificate; граница между ядром и реализациями.
- `option/` — строгий JSON-конфиг, типы опций и первичная валидация. Неизвестные
JSON-поля запрещены.
- `route/` — router, connection/network managers, route rules и rule sets.
- `dns/` — DNS router, правила, кэш и transports.
- `protocol/` — inbound/outbound реализации протоколов.
- `transport/` — сетевые транспорты, включая V2Ray transports.
- `common/` — общие сетевые, TLS, HTTP, netns, database и platform helpers.
- `experimental/` — Clash/V2Ray API, cache file и прочие расширения.
- `service/` — отдельные runtime services.
- `test/` — интеграционные протокольные тесты, поднимающие настоящий `Box`.
## Production-путь запуска
```text
OpenWrt procd
-> /usr/bin/sing-box run -c <config> -D <workdir>
-> cmd/sing-box/main.go: mainCommand.Execute
-> cmd/sing-box/cmd.go: preRun
рабочий каталог + include.Context(registries)
-> cmd/sing-box/cmd_run.go: run
readConfig -> сортировка путей -> mergeOptionsList
-> create
box.New -> Box.Start
-> box.go
managers -> Initialize -> Start -> PostStart -> Started
-> ожидание сигнала
SIGHUP: check -> Close -> перечитать конфиг -> создать новый Box
SIGINT/SIGTERM: Close -> выход
```
`box.New` строит граф в следующем центре: endpoint, inbound, outbound,
provider, DNS transport/router, network, connection, HTTP client, router,
service и certificate-provider managers. `Box.Start` запускает их фазами;
переставлять порядок без трассировки зависимостей нельзя. `Box.Close` закрывает
компоненты явно заданным порядком и агрегирует ошибки.
Конфиг может приходить из нескольких `-c` и каталогов `-C`. Все найденные
JSON-файлы сортируются по полному пути, затем сливаются. Без флагов CLI ищет
`config.json` в рабочем каталоге.
## Фактический пакетный профиль
OpenWrt package builds читают только
`sing-box-rr/release/DEFAULT_BUILD_TAGS_OTHERS`:
```text
with_quic,with_utls,with_clash_api
```
Из `include/registry.go` пакетный runtime получает:
- inbounds: TUN, redirect, TProxy, direct, SOCKS, HTTP, mixed, Shadowsocks,
VMess, Trojan, ShadowTLS, VLESS, AnyTLS и QUIC-семейство
Hysteria/TUIC/Hysteria2;
- outbounds: direct, block, selector/urltest/fallback, SOCKS, HTTP,
Shadowsocks, VMess, Trojan, ShadowTLS, VLESS, Mieru, AnyTLS и то же
QUIC-семейство;
- DNS transports: TCP, UDP, TLS, HTTPS, DNSCrypt/SDNS, hosts, local, fake-IP,
fallback, QUIC и HTTP/3;
- services: OOM killer; profiler остаётся распознаваемым stub, потому что в
package profile нет `with_profiler`;
- Clash API и uTLS включены build tags.
Некоторые config types намеренно остаются распознаваемыми, но ведут в stub с
явной ошибкой: Naive outbound без `with_naive_outbound`, удалённые
failover/SSR/старый WireGuard, VPN endpoints, resolved/SSMAPI и функции без
QUIC при выключенном `with_quic`. Provider manager и поле `providers` ещё есть
в общем графе, но package registry провайдеров пуст.
Полные `DEFAULT_BUILD_TAGS`, `DEFAULT_BUILD_TAGS_WINDOWS` и Docker profile —
другие, значительно более широкие профили. Не использовать их как описание
возможностей опубликованного OpenWrt-пакета.
## URLTest и NetworkQuality
URLTest реализован в `sing-box-rr/protocol/group/urltest.go`, а измерение и
политика выбора по пропускной способности — в соседнем
`protocol/group/networkquality.go`. Сам измеритель не дублируется: группа
использует существующий `common/networkquality.Run` через dialer конкретного
outbound.
Конфигурация URLTest поддерживает:
- `strategy: "latency"` — поведение по умолчанию и прежний выбор по задержке;
- `strategy: "networkquality"` — среди доступных по URLTest участников
выбирается максимальный комбинированный score
`download capacity (bit/s) / URLTest delay (ms)`; при равном score
побеждает меньшая задержка, а нулевая считается равной `1ms`;
- `networkquality_ttl` — минимальный интервал до повторного сетевого замера,
период автоматического теста и TTL кэша; по умолчанию и минимум `30m`;
- `networkquality_target` — `apple` по умолчанию, встроенный `cloudflare` или
`custom`;
- `networkquality_config_url` — Config v1 URL для `custom`; если target не
указан, наличие URL автоматически выбирает `custom`;
- `networkquality_download_url` и `networkquality_upload_url` — прямые endpoint-ы
для `custom` вместо Config v1;
- `networkquality_probe_url` — необязательный маленький download
endpoint для latency/responsiveness probes; по умолчанию берётся
`networkquality_download_url`.
Apple использует прежний публичный Mensura Config v1. Cloudflare задан
встроенным Config v1 с официальными endpoint-ами
`h3.speed.cloudflare.com/__down` и `h3.speed.cloudflare.com/__up`, поэтому
отдельный внешний config-файл ему не нужен. Собственный
`network-quality/goserver` подключается как `custom`: голый host или корневой
URL автоматически дополняется `/.well-known/nq`, полный URL с путём остаётся
без изменений. Config URL и прямые URL взаимоисключаются; в прямом режиме
download, upload и probe должны использовать один host.
Успешные NetworkQuality-замеры дедуплицируются между всеми URLTest-группами
одного `Box`. Идентичность участника намеренно упрощена до нормализованного
`server:port`; ключ также включает target, max runtime и режим HTTP/3. Поэтому
разные credentials/transports на одном `server:port` используют общий замер.
Общий `common/urltest.HistoryStorage` хранит последний успешный результат и
объединяет одновременно запущенные проверки, а локальная карта группы по-прежнему
публикует результат под tag каждого своего участника. Если владелец общей
проверки отменён, независимый ожидающий запрос с живым контекстом повторно
захватывает ключ и запускает измерение сам, а не наследует чужой `context.Canceled`.
Один общий слот в `HistoryStorage` разрешает не более одного реального
NetworkQuality-теста на весь `Box`; разные ключи из разных групп ждут в общей
очереди, а ожидание прерывается контекстом запроса. Standalone daemon RPC
`StartNetworkQualityTest` использует тот же слот, поэтому тоже не пересекается с
групповыми замерами.
При `strategy: "networkquality"` первый автоматический замер запускается после
начального latency URLTest, затем повторяется через TTL после завершения
предыдущего автоматического прохода. Переключение с `latency` на
`networkquality` также запускает замер сразу и заново отсчитывает TTL после его
завершения; в режиме `latency` автоматический таймер остановлен. До истечения TTL
ручной API-вызов переиспользует кэш. Сам TTL не исключает старый download из
сортировки. Обычная ошибка повторного замера публикуется в API, но не удаляет
последние успешные capacity и исходный timestamp: они остаются в выборе до нового
успешного результата.
HTTP 429 распознаётся как отдельный rate-limit: API возвращает участнику
`rateLimited: true` и текст ошибки, но сохраняет capacity и исходный timestamp
последнего успешного замера. Такой download продолжает участвовать в выборе,
независимо от его возраста; без сохранённого результата
алгоритм возвращается к latency. Сам ответ 429 запускает новый TTL-интервал:
повторные API-вызовы внутри него возвращают сохранённый rate-limit без нового
сетевого теста.
Upload capacity измеряется, хранится и отдаётся API, но в сортировке не
участвует. Если успешных download-результатов нет, выбранный режим
`networkquality` автоматически возвращается к latency-алгоритму. Участники
с известной URLTest-задержкой тестируются строго последовательно; участник без
delay не занимает общий слот и получает в API ошибку
`URLTest delay unavailable`. Предыдущие успешные capacity и timestamp такого
участника сохраняются, но без delay он не участвует в выборе; после появления
нового delay сохранённый download снова становится доступен сортировке.
Результат каждого участника, включая пропущенного, публикуется API сразу после
обработки. Неполная первая карта в выборе не участвует: до полного
опроса всей текущей группы URLTest выбирает по задержке. NetworkQuality только публикует полный
snapshot и сам не меняет outbound; комбинированный download/latency-выбор применяет следующий
полный URLTest-проход. Отдельные delay-ответы Clash API обновляют history, но не переключают
NetworkQuality-группу по частично обновлённой карте задержек. При повторных замерах предыдущий полный snapshot
действует до готовности нового; отмена
оставляет частичные API-результаты видимыми, но не заменяет завершённый snapshot.
Живые delay-замеры всех участников сохраняются, но в history самой NetworkQuality-группы
публикуется committed delay выбранного `now`, обновляемый только после полного URLTest-прохода.
Публикуемая API-история живёт в памяти конкретной URLTest-группы, общий кэш —
в памяти `Box`; ни то ни другое не сохраняется между перезапусками.
Runtime-управление находится в Clash meta API
`experimental/clashapi/api_meta_group.go`:
- `GET|PUT /group/{name}/strategy` — прочитать или изменить стратегию;
- `GET /group/{name}/networkquality` — получить последний замер;
- `POST /group/{name}/networkquality` — последовательно обработать всех
участников и измерить только имеющих URLTest delay; тело принимает
`maxRuntimeSeconds` и `http3`.
Ответ NetworkQuality содержит `progress.current` и `progress.total`: во время
прохода это 1-based номер обрабатываемого участника и фиксированный размер
снимка группы на старте; вне активного прохода оба значения равны нулю.
Контракт между API и группой находится в `adapter/experimental.go`: прежний
`URLTestGroup` сохранён для source compatibility, а новые методы собраны в
расширяющем `NetworkQualityURLTestGroup`. JSON-поля конфига находятся в
`option/group.go`, а регрессионные тесты выбора и конкурентности — в
`protocol/group/networkquality_test.go`. Встроенный Cloudflare Config и
передача inline-конфига в общий измеритель проверяются в
`common/networkquality/networkquality_test.go`.
## OpenWrt-интеграция и упаковка
UCI-конфиг `release/config/openwrt.conf` по умолчанию включает сервис и задаёт:
- конфиг: `/etc/sing-box/config.json`;
- workdir: `/usr/share/sing-box`;
- stderr logging: включён.
Init script `release/config/openwrt.init` использует `procd`, следит за файлом
конфига, включает respawn и reload trigger, выставляет высокий `nofile`.
Оба локальных скрипта:
1. читают версию и `SOURCE_DATE_EPOCH` из корня;
2. читают package tags из `DEFAULT_BUILD_TAGS_OTHERS`;
3. собирают `./cmd/sing-box` внешним Go с `-trimpath`, version ldflag и
опциональным CGO/musl toolchain;
4. укладывают бинарник, JSON/UCI/init/keep-файлы;
5. формируют package metadata и сохраняют артефакт в `packages/`.
Пакет называется `sing-box-rr`, предоставляет `sing-box` и `sing-box-tiny`,
конфликтует с `sing-box`, зависит от `ca-bundle`, `kmod-inet-diag`, `kmod-tun`
и создаёт пользователя `sing-box` с UID/GID 5566. Совместимость с
`sing-box-tiny` нужна downstream-пакетам вроде ZeroBlock.
Локальные пути SDK и `/usr/local/go/bin/go` жёстко заданы в скриптах. Их можно
переопределять только предусмотренными переменными; переносимость не
предполагать. IPK helper поддерживает arm64, APK helper — arm64 и x86_64.
Feed `sing-box-rr/Makefile` поддерживает OpenWrt `aarch64` и `x86_64` и также
использует внешний Go вместо SDK `golang/host`.
Базовые команды из корня:
```bash
./build-ipk.sh arm64
./build-apk.sh arm64
./build-apk.sh x86_64
```
`--force` удаляет выделенные Go build/module caches в `/tmp` перед сборкой.
## Тесты и проверки
В дереве 110 файлов `*_test.go`. Основные группы:
- `common/` — низкоуровневые helpers;
- `test/` — интеграционные тесты протоколов, создающие настоящий `box.Box`;
- `route/` — запуск соединений, rule-set lifecycle и семантика правил;
- `dns/` — race/fallback/order и transport multiplexer;
- `option/` — строгий JSON и валидация;
- `transport/`, `service/`, `experimental/`, `adapter/` — точечные тесты.
Интеграционный harness в `test/box_test.go` вызывает production-путь
`include.Context -> box.New -> Box.Start -> Box.Close`. Часть тестов требует
сетевых возможностей/root; upstream workflow на Unix запускает весь suite
через `sudo`.
Рекомендуемая локальная проверка из `sing-box-rr/`:
```bash
BUILD_TAGS="$(cat release/DEFAULT_BUILD_TAGS_OTHERS)"
/usr/local/go/bin/go test -tags "$BUILD_TAGS" -ldflags "$(cat release/LDFLAGS)" ./...
```
Для полного соответствия CI на Unix может понадобиться `-exec sudo`. Go
toolchain должен удовлетворять `go.mod`; build scripts выставляют
`GOTOOLCHAIN=go<version>+auto`, если переменная не задана явно.
Workflow-файлы находятся в `sing-box-rr/.github/workflows/`, а не в корневом
`.github/workflows/` этого wrapper-репозитория. Поэтому не считать их
автоматически действующим CI корневого репозитория без отдельной проверки
хостинга; использовать их как upstream-рецепты test/lint/build.
## Оговорки и известные несоответствия
- `go.mod` требует Go 1.26.4, хотя `README.md` ещё упоминает 1.26.1. Источник
истины — `go.mod`.
- `README.md` показывает `./build-ipk.sh x86_64`, но текущий IPK helper
принимает только `arm64|all`; x86_64 реализован только в APK helper/feed.
- `README.md` говорит, что Naive outbound исключён из default builds. Это верно
для package profile `DEFAULT_BUILD_TAGS_OTHERS`, но полный
`DEFAULT_BUILD_TAGS` всё ещё содержит `with_naive_outbound`.
- Дефолтный `release/config/config.json` слушает `::`:8080 и содержит
опубликованный статический Shadowsocks-пароль. Это пример, а не безопасная
production-конфигурация; пароль и bind policy нужно менять до установки на
доступный извне роутер.
- В репозитории закоммичены бинарные IPK/APK. После изменения версии, tags,
registry или release config нужно пересобрать нужные архитектуры и проверить,
что metadata и содержимое пакетов синхронны с исходниками.
## Правило работы с проектом
Перед изменением сначала проследить существующий production call chain и
найти владельца поведения. Для урезания использовать один путь:
1. package build tag;
2. registry или build-tag stub;
3. проверка валидных конфигов и production-тестов;
4. только затем чистка option/constant/source surface.
Не дублировать feature policy одновременно в новых скриптах и Go-коде.
Версия живёт в `VERSION`, package tags — в `DEFAULT_BUILD_TAGS_OTHERS`,
runtime registration — в `include/registry.go`.
+26
View File
@@ -25,6 +25,22 @@ type URLTestHistory struct {
Delay uint16 `json:"delay"`
}
type NetworkQualityHistory struct {
Time time.Time `json:"time"`
Outbound string `json:"outbound"`
DownloadCapacity int64 `json:"downloadCapacity,omitempty"`
UploadCapacity int64 `json:"uploadCapacity,omitempty"`
DownloadRPM int32 `json:"downloadRPM,omitempty"`
UploadRPM int32 `json:"uploadRPM,omitempty"`
IdleLatencyMs int32 `json:"idleLatencyMs,omitempty"`
DownloadCapacityAccuracy int32 `json:"downloadCapacityAccuracy,omitempty"`
UploadCapacityAccuracy int32 `json:"uploadCapacityAccuracy,omitempty"`
DownloadRPMAccuracy int32 `json:"downloadRPMAccuracy,omitempty"`
UploadRPMAccuracy int32 `json:"uploadRPMAccuracy,omitempty"`
RateLimited bool `json:"rateLimited,omitempty"`
Error string `json:"error,omitempty"`
}
type V2RayServer interface {
LifecycleService
StatsService() ConnectionTracker
@@ -178,3 +194,13 @@ type URLTestGroup interface {
URLTest(ctx context.Context) (map[string]uint16, error)
PerformUpdateCheck()
}
type NetworkQualityURLTestGroup interface {
URLTestGroup
NetworkQuality(ctx context.Context, maxRuntime time.Duration, http3 bool) (map[string]*NetworkQualityHistory, error)
NetworkQualityHistory() map[string]*NetworkQualityHistory
NetworkQualityProgress() (current int, total int)
NetworkQualitySelectedHistory() *URLTestHistory
SelectionStrategy() string
SetSelectionStrategy(strategy string) error
}
+32
View File
@@ -0,0 +1,32 @@
package adapter
import (
"context"
"encoding/json"
"strings"
"testing"
)
type legacyURLTestGroup struct {
Outbound
}
func (*legacyURLTestGroup) Now() string { return "" }
func (*legacyURLTestGroup) All() []string { return nil }
func (*legacyURLTestGroup) URLTest(context.Context) (map[string]uint16, error) { return nil, nil }
func (*legacyURLTestGroup) PerformUpdateCheck() {}
var _ URLTestGroup = (*legacyURLTestGroup)(nil)
func TestNetworkQualityHistoryRateLimitedJSON(t *testing.T) {
content, err := json.Marshal(NetworkQualityHistory{RateLimited: true})
if err != nil {
t.Fatal(err)
}
if !strings.Contains(string(content), `"rateLimited":true`) {
t.Fatalf("rate limit marker is missing: %s", content)
}
}
+5
View File
@@ -51,3 +51,8 @@ type OutboundManager interface {
Remove(tag string) error
Create(ctx context.Context, router Router, logger log.ContextLogger, tag string, outboundType string, options any) error
}
// OutboundServerProvider exposes the normalized proxy server used for shared measurements.
type OutboundServerProvider interface {
OutboundServer(tag string) (string, bool)
}
+29
View File
@@ -11,6 +11,7 @@ import (
"github.com/sagernet/sing-box/common/taskmonitor"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing-box/log"
"github.com/sagernet/sing-box/option"
"github.com/sagernet/sing/common"
E "github.com/sagernet/sing/common/exceptions"
"github.com/sagernet/sing/common/logger"
@@ -28,6 +29,7 @@ type Manager struct {
stage adapter.StartStage
outbounds []adapter.Outbound
outboundByTag map[string]adapter.Outbound
serverByTag map[string]string
dependByTag map[string][]string
defaultOutbound adapter.Outbound
defaultOutboundFallback func() (adapter.Outbound, error)
@@ -40,6 +42,7 @@ func NewManager(logger logger.ContextLogger, registry adapter.OutboundRegistry,
endpoint: endpoint,
defaultTag: defaultTag,
outboundByTag: make(map[string]adapter.Outbound),
serverByTag: make(map[string]string),
dependByTag: make(map[string][]string),
}
}
@@ -208,6 +211,13 @@ func (m *Manager) Outbound(tag string) (adapter.Outbound, bool) {
return m.endpoint.Get(tag)
}
func (m *Manager) OutboundServer(tag string) (string, bool) {
m.access.RLock()
server, found := m.serverByTag[tag]
m.access.RUnlock()
return server, found
}
func (m *Manager) Default() adapter.Outbound {
m.access.RLock()
defer m.access.RUnlock()
@@ -222,6 +232,7 @@ func (m *Manager) Remove(tag string) error {
return os.ErrInvalid
}
delete(m.outboundByTag, tag)
delete(m.serverByTag, tag)
index := common.Index(m.outbounds, func(it adapter.Outbound) bool {
return it == outbound
})
@@ -296,6 +307,11 @@ func (m *Manager) Create(ctx context.Context, router adapter.Router, logger log.
}
m.outbounds = append(m.outbounds, outbound)
m.outboundByTag[tag] = outbound
if server, loaded := outboundServer(options); loaded {
m.serverByTag[tag] = server
} else {
delete(m.serverByTag, tag)
}
dependencies := outbound.Dependencies()
for _, dependency := range dependencies {
m.dependByTag[dependency] = append(m.dependByTag[dependency], tag)
@@ -308,3 +324,16 @@ func (m *Manager) Create(ctx context.Context, router adapter.Router, logger log.
}
return nil
}
func outboundServer(options any) (string, bool) {
serverOptions, loaded := options.(option.ServerOptionsWrapper)
if !loaded {
return "", false
}
server := serverOptions.TakeServerOptions()
server.Server = strings.ToLower(strings.TrimSpace(server.Server))
if server.Server == "" || server.ServerPort == 0 {
return "", false
}
return server.Build().String(), true
}
@@ -0,0 +1,22 @@
package outbound
import (
"testing"
"github.com/sagernet/sing-box/option"
)
func TestOutboundServer(t *testing.T) {
server, loaded := outboundServer(&option.SOCKSOutboundOptions{
ServerOptions: option.ServerOptions{
Server: "EXAMPLE.COM",
ServerPort: 1080,
},
})
if !loaded || server != "example.com:1080" {
t.Fatalf("outboundServer() = %q, %v", server, loaded)
}
if _, loaded = outboundServer(&option.DirectOutboundOptions{}); loaded {
t.Fatal("outbound without server was accepted")
}
}
@@ -4,6 +4,8 @@ import (
"context"
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"io"
"math"
"math/rand"
@@ -24,6 +26,21 @@ import (
const DefaultConfigURL = "https://mensura.cdn-apple.com/api/v1/gm/config"
// ErrRateLimited indicates that a network quality endpoint returned HTTP 429.
var ErrRateLimited = errors.New("network quality service rate limited")
// CloudflareConfig returns the Config v1 equivalent of Cloudflare's public RPM endpoints.
func CloudflareConfig() *Config {
return &Config{
Version: 1,
URLs: URLs{
SmallDownloadURL: "https://h3.speed.cloudflare.com/__down?bytes=10",
LargeDownloadURL: "https://h3.speed.cloudflare.com/__down?bytes=10000000000",
UploadURL: "https://h3.speed.cloudflare.com/__up",
},
}
}
type Config struct {
Version int `json:"version"`
TestEndpoint string `json:"test_endpoint"`
@@ -116,6 +133,7 @@ const (
type Options struct {
ConfigURL string
Config *Config
HTTPClient *http.Client
NewMeasurementClient MeasurementClientFactory
Serial bool
@@ -588,10 +606,13 @@ func Run(options Options) (*Result, error) {
if err != nil {
return nil, err
}
configURL := resolveConfigURL(options.ConfigURL)
config, err := fetchConfig(ctx, options.HTTPClient, configURL)
if err != nil {
return nil, E.Cause(err, "fetch config")
config := options.Config
if config == nil {
configURL := ResolveConfigURL(options.ConfigURL)
config, err = fetchConfig(ctx, options.HTTPClient, configURL)
if err != nil {
return nil, E.Cause(err, "fetch config")
}
}
resolved, err := validateConfig(config)
if err != nil {
@@ -836,7 +857,8 @@ func measureParallel(
return download.measurement, upload.measurement, nil
}
func resolveConfigURL(rawURL string) string {
// ResolveConfigURL expands a host or root URL to its Config v1 well-known URL.
func ResolveConfigURL(rawURL string) string {
rawURL = strings.TrimSpace(rawURL)
if rawURL == "" {
return DefaultConfigURL
@@ -1278,6 +1300,9 @@ func closeMeasurementClient(client *http.Client) {
}
func validateResponse(resp *http.Response) error {
if resp.StatusCode == http.StatusTooManyRequests {
return fmt.Errorf("%w: %s", ErrRateLimited, resp.Status)
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return E.New("unexpected status: ", resp.Status)
}
@@ -0,0 +1,65 @@
package networkquality
import (
"errors"
"net/http"
"strings"
"testing"
)
type rejectingRoundTripper struct {
called bool
}
func (t *rejectingRoundTripper) RoundTrip(*http.Request) (*http.Response, error) {
t.called = true
return nil, errors.New("unexpected request")
}
func TestRunUsesInlineConfig(t *testing.T) {
transport := &rejectingRoundTripper{}
_, err := Run(Options{
Config: &Config{Version: 2},
HTTPClient: &http.Client{Transport: transport},
})
if err == nil || !strings.Contains(err.Error(), "unsupported config version") {
t.Fatalf("unexpected error: %v", err)
}
if transport.called {
t.Fatal("inline config caused a config request")
}
}
func TestCloudflareConfig(t *testing.T) {
resolved, err := validateConfig(CloudflareConfig())
if err != nil {
t.Fatal(err)
}
if resolved.smallURL.Host != "h3.speed.cloudflare.com" || resolved.largeURL.Query().Get("bytes") != "10000000000" || resolved.uploadURL.Path != "/__up" {
t.Fatalf("unexpected Cloudflare config: %+v", resolved)
}
}
func TestResolveConfigURL(t *testing.T) {
for rawURL, expected := range map[string]string{
"": DefaultConfigURL,
"nq.example.com": "https://nq.example.com/.well-known/nq",
"https://nq.example.com": "https://nq.example.com/.well-known/nq",
"https://nq.example.com/": "https://nq.example.com/.well-known/nq",
"https://nq.example.com/nqv1": "https://nq.example.com/nqv1",
} {
if actual := ResolveConfigURL(rawURL); actual != expected {
t.Fatalf("ResolveConfigURL(%q) = %q, want %q", rawURL, actual, expected)
}
}
}
func TestValidateResponseRateLimited(t *testing.T) {
err := validateResponse(&http.Response{
StatusCode: http.StatusTooManyRequests,
Status: "429 Too Many Requests",
})
if !errors.Is(err, ErrRateLimited) {
t.Fatalf("unexpected error: %v", err)
}
}
@@ -0,0 +1,122 @@
package urltest
import (
"context"
"errors"
"time"
"github.com/sagernet/sing-box/common/networkquality"
)
// NetworkQualityCacheKey identifies equivalent speed measurements within one Box.
type NetworkQualityCacheKey struct {
Server string
Target string
MaxRuntime time.Duration
HTTP3 bool
}
type networkQualityCacheEntry struct {
measuredAt time.Time
result *networkquality.Result
lastTestAt time.Time
lastError error
}
type networkQualityCall struct {
done chan struct{}
entry *networkQualityCacheEntry
err error
}
// LoadOrRunNetworkQuality reuses a recent result or rate limit, or joins an identical in-flight test.
func (s *HistoryStorage) LoadOrRunNetworkQuality(
ctx context.Context,
key NetworkQualityCacheKey,
ttl time.Duration,
run func(context.Context) (*networkquality.Result, error),
) (*networkquality.Result, time.Time, error) {
for {
now := time.Now()
s.access.Lock()
previous := s.networkQualityHistory[key]
if previous != nil && now.Sub(previous.lastTestAt) <= ttl {
s.access.Unlock()
return copyNetworkQualityEntry(previous, previous.lastError)
}
if call := s.networkQualityTestsInProgress[key]; call != nil {
s.access.Unlock()
select {
case <-ctx.Done():
return nil, time.Time{}, ctx.Err()
case <-call.done:
if err := ctx.Err(); err != nil {
return nil, time.Time{}, err
}
if errors.Is(call.err, context.Canceled) || errors.Is(call.err, context.DeadlineExceeded) {
continue
}
return copyNetworkQualityEntry(call.entry, call.err)
}
}
call := &networkQualityCall{
done: make(chan struct{}),
entry: previous,
}
s.networkQualityTestsInProgress[key] = call
s.access.Unlock()
result, err := s.RunNetworkQuality(ctx, run)
measuredAt := time.Now()
if err == nil && result == nil {
err = errors.New("network quality test returned no result")
}
s.access.Lock()
if err == nil {
resultCopy := *result
call.entry = &networkQualityCacheEntry{
measuredAt: measuredAt,
result: &resultCopy,
lastTestAt: measuredAt,
}
s.networkQualityHistory[key] = call.entry
} else if errors.Is(err, networkquality.ErrRateLimited) {
call.entry = &networkQualityCacheEntry{
lastTestAt: measuredAt,
lastError: err,
}
if previous != nil && previous.result != nil {
resultCopy := *previous.result
call.entry.measuredAt = previous.measuredAt
call.entry.result = &resultCopy
}
s.networkQualityHistory[key] = call.entry
}
call.err = err
delete(s.networkQualityTestsInProgress, key)
close(call.done)
s.access.Unlock()
return copyNetworkQualityEntry(call.entry, call.err)
}
}
// RunNetworkQuality executes a measurement through the instance-wide serial slot.
func (s *HistoryStorage) RunNetworkQuality(ctx context.Context, run func(context.Context) (*networkquality.Result, error)) (*networkquality.Result, error) {
select {
case s.networkQualityTestSlot <- struct{}{}:
defer func() { <-s.networkQualityTestSlot }()
return run(ctx)
case <-ctx.Done():
return nil, ctx.Err()
}
}
func copyNetworkQualityEntry(entry *networkQualityCacheEntry, err error) (*networkquality.Result, time.Time, error) {
if entry == nil || entry.result == nil {
return nil, time.Time{}, err
}
result := *entry.result
return &result, entry.measuredAt, err
}
@@ -0,0 +1,247 @@
package urltest
import (
"context"
"errors"
"sync/atomic"
"testing"
"time"
"github.com/sagernet/sing-box/common/networkquality"
)
type observedDoneContext struct {
context.Context
observed chan<- struct{}
}
func (c observedDoneContext) Done() <-chan struct{} {
select {
case c.observed <- struct{}{}:
default:
}
return c.Context.Done()
}
func TestNetworkQualityCacheDeduplicatesRuns(t *testing.T) {
storage := NewHistoryStorage()
key := NetworkQualityCacheKey{
Server: "example.com:443",
Target: "apple",
MaxRuntime: networkquality.DefaultMaxRuntime,
}
var calls atomic.Int32
runner := func(context.Context) (*networkquality.Result, error) {
calls.Add(1)
return &networkquality.Result{DownloadCapacity: 100}, nil
}
first, firstTime, err := storage.LoadOrRunNetworkQuality(context.Background(), key, time.Hour, runner)
if err != nil {
t.Fatal(err)
}
second, secondTime, err := storage.LoadOrRunNetworkQuality(context.Background(), key, time.Hour, runner)
if err != nil {
t.Fatal(err)
}
if calls.Load() != 1 || first.DownloadCapacity != 100 || second.DownloadCapacity != 100 || !firstTime.Equal(secondTime) {
t.Fatalf("unexpected cached results: calls=%d first=%+v second=%+v", calls.Load(), first, second)
}
}
func TestNetworkQualityCacheSharesInFlightRun(t *testing.T) {
storage := NewHistoryStorage()
key := NetworkQualityCacheKey{Server: "example.com:443", Target: "apple"}
started := make(chan struct{})
release := make(chan struct{})
var calls atomic.Int32
runner := func(context.Context) (*networkquality.Result, error) {
if calls.Add(1) == 1 {
close(started)
}
<-release
return &networkquality.Result{DownloadCapacity: 100}, nil
}
type response struct {
result *networkquality.Result
err error
}
responses := make(chan response, 2)
run := func() {
result, _, err := storage.LoadOrRunNetworkQuality(context.Background(), key, time.Hour, runner)
responses <- response{result: result, err: err}
}
go run()
<-started
go run()
close(release)
for range 2 {
response := <-responses
if response.err != nil || response.result.DownloadCapacity != 100 {
t.Fatalf("unexpected response: %+v", response)
}
}
if calls.Load() != 1 {
t.Fatalf("runner called %d times", calls.Load())
}
}
func TestNetworkQualityCacheRetriesAfterOwnerCancellation(t *testing.T) {
storage := NewHistoryStorage()
key := NetworkQualityCacheKey{Server: "example.com:443", Target: "apple"}
ownerStarted := make(chan struct{})
var calls atomic.Int32
runner := func(ctx context.Context) (*networkquality.Result, error) {
if calls.Add(1) == 1 {
close(ownerStarted)
<-ctx.Done()
return nil, ctx.Err()
}
return &networkquality.Result{DownloadCapacity: 100}, nil
}
ownerCtx, cancelOwner := context.WithCancel(context.Background())
ownerDone := make(chan error, 1)
go func() {
_, _, err := storage.LoadOrRunNetworkQuality(ownerCtx, key, time.Hour, runner)
ownerDone <- err
}()
<-ownerStarted
type response struct {
result *networkquality.Result
err error
}
waiterObserved := make(chan struct{}, 1)
waiterDone := make(chan response, 1)
go func() {
waiterCtx := observedDoneContext{Context: context.Background(), observed: waiterObserved}
result, _, err := storage.LoadOrRunNetworkQuality(waiterCtx, key, time.Hour, runner)
waiterDone <- response{result: result, err: err}
}()
select {
case <-waiterObserved:
case <-time.After(time.Second):
t.Fatal("waiter did not join the in-flight test")
}
cancelOwner()
if err := <-ownerDone; !errors.Is(err, context.Canceled) {
t.Fatalf("unexpected owner error: %v", err)
}
waiterResponse := <-waiterDone
if waiterResponse.err != nil || waiterResponse.result == nil || waiterResponse.result.DownloadCapacity != 100 {
t.Fatalf("waiter inherited owner cancellation: %+v", waiterResponse)
}
if calls.Load() != 2 {
t.Fatalf("runner called %d times", calls.Load())
}
}
func TestNetworkQualityCacheSerializesDifferentKeys(t *testing.T) {
storage := NewHistoryStorage()
firstStarted := make(chan struct{})
secondStarted := make(chan struct{})
releaseFirst := make(chan struct{})
firstDone := make(chan error, 1)
secondDone := make(chan error, 1)
go func() {
_, _, err := storage.LoadOrRunNetworkQuality(context.Background(), NetworkQualityCacheKey{Server: "first.example.com:443"}, time.Hour, func(context.Context) (*networkquality.Result, error) {
close(firstStarted)
<-releaseFirst
return &networkquality.Result{DownloadCapacity: 100}, nil
})
firstDone <- err
}()
<-firstStarted
go func() {
_, _, err := storage.LoadOrRunNetworkQuality(context.Background(), NetworkQualityCacheKey{Server: "second.example.com:443"}, time.Hour, func(context.Context) (*networkquality.Result, error) {
close(secondStarted)
return &networkquality.Result{DownloadCapacity: 200}, nil
})
secondDone <- err
}()
select {
case <-secondStarted:
close(releaseFirst)
<-firstDone
<-secondDone
t.Fatal("different network quality keys ran concurrently")
case <-time.After(20 * time.Millisecond):
}
close(releaseFirst)
if err := <-firstDone; err != nil {
t.Fatal(err)
}
select {
case <-secondStarted:
case <-time.After(time.Second):
t.Fatal("queued network quality test did not start")
}
if err := <-secondDone; err != nil {
t.Fatal(err)
}
}
func TestNetworkQualityCacheCancelsQueuedTest(t *testing.T) {
storage := NewHistoryStorage()
firstStarted := make(chan struct{})
releaseFirst := make(chan struct{})
firstDone := make(chan error, 1)
go func() {
_, _, err := storage.LoadOrRunNetworkQuality(context.Background(), NetworkQualityCacheKey{Server: "first.example.com:443"}, time.Hour, func(context.Context) (*networkquality.Result, error) {
close(firstStarted)
<-releaseFirst
return &networkquality.Result{DownloadCapacity: 100}, nil
})
firstDone <- err
}()
<-firstStarted
ctx, cancel := context.WithCancel(context.Background())
cancel()
runnerCalled := false
_, _, err := storage.LoadOrRunNetworkQuality(ctx, NetworkQualityCacheKey{Server: "second.example.com:443"}, time.Hour, func(context.Context) (*networkquality.Result, error) {
runnerCalled = true
return &networkquality.Result{DownloadCapacity: 200}, nil
})
if !errors.Is(err, context.Canceled) || runnerCalled {
t.Fatalf("queued test was not canceled: called=%v err=%v", runnerCalled, err)
}
close(releaseFirst)
if err = <-firstDone; err != nil {
t.Fatal(err)
}
}
func TestNetworkQualityCacheReturnsPreviousResultOnRateLimit(t *testing.T) {
storage := NewHistoryStorage()
key := NetworkQualityCacheKey{Server: "example.com:443", Target: "apple"}
var calls atomic.Int32
result, measuredAt, err := storage.LoadOrRunNetworkQuality(context.Background(), key, time.Hour, func(context.Context) (*networkquality.Result, error) {
calls.Add(1)
return &networkquality.Result{DownloadCapacity: 100}, nil
})
if err != nil || result.DownloadCapacity != 100 {
t.Fatalf("unexpected initial result: %+v, %v", result, err)
}
result, preservedAt, err := storage.LoadOrRunNetworkQuality(context.Background(), key, -time.Nanosecond, func(context.Context) (*networkquality.Result, error) {
calls.Add(1)
return nil, networkquality.ErrRateLimited
})
if !errors.Is(err, networkquality.ErrRateLimited) || result == nil || result.DownloadCapacity != 100 || !preservedAt.Equal(measuredAt) {
t.Fatalf("previous result was not preserved: result=%+v time=%v err=%v", result, preservedAt, err)
}
result, preservedAt, err = storage.LoadOrRunNetworkQuality(context.Background(), key, time.Hour, func(context.Context) (*networkquality.Result, error) {
calls.Add(1)
return &networkquality.Result{DownloadCapacity: 200}, nil
})
if !errors.Is(err, networkquality.ErrRateLimited) || result == nil || result.DownloadCapacity != 100 || !preservedAt.Equal(measuredAt) || calls.Load() != 2 {
t.Fatalf("rate limit was not cached: calls=%d result=%+v time=%v err=%v", calls.Load(), result, preservedAt, err)
}
}
+10 -4
View File
@@ -19,14 +19,20 @@ import (
)
type HistoryStorage struct {
access sync.RWMutex
delayHistory map[string]*adapter.URLTestHistory
updateHooks []*observable.Subscriber[struct{}]
access sync.RWMutex
delayHistory map[string]*adapter.URLTestHistory
networkQualityHistory map[NetworkQualityCacheKey]*networkQualityCacheEntry
networkQualityTestsInProgress map[NetworkQualityCacheKey]*networkQualityCall
networkQualityTestSlot chan struct{}
updateHooks []*observable.Subscriber[struct{}]
}
func NewHistoryStorage() *HistoryStorage {
return &HistoryStorage{
delayHistory: make(map[string]*adapter.URLTestHistory),
delayHistory: make(map[string]*adapter.URLTestHistory),
networkQualityHistory: make(map[NetworkQualityCacheKey]*networkQualityCacheEntry),
networkQualityTestsInProgress: make(map[NetworkQualityCacheKey]*networkQualityCall),
networkQualityTestSlot: make(chan struct{}, 1),
}
}
+5
View File
@@ -73,6 +73,11 @@ const (
TypeURLTest = "urltest"
)
const (
URLTestStrategyLatency = "latency"
URLTestStrategyNetworkQuality = "networkquality"
)
func ProxyDisplayName(proxyType string) string {
switch proxyType {
case TypeTun:
+2
View File
@@ -13,6 +13,8 @@ const (
ICMPTimeout = 10 * time.Second
DefaultURLTestInterval = 3 * time.Minute
DefaultURLTestIdleTimeout = 30 * time.Minute
MinNetworkQualityTTL = 30 * time.Minute
DefaultNetworkQualityTTL = 30 * time.Minute
StartTimeout = 10 * time.Second
StopTimeout = 5 * time.Second
FatalStopTimeout = 10 * time.Second
+12 -10
View File
@@ -1500,16 +1500,18 @@ func (s *StartedService) StartNetworkQualityTest(
return err
}
result, nqErr := networkquality.Run(networkquality.Options{
ConfigURL: request.ConfigURL,
HTTPClient: httpClient,
NewMeasurementClient: measurementClientFactory,
Serial: request.Serial,
MaxRuntime: time.Duration(request.MaxRuntimeSeconds) * time.Second,
Context: server.Context(),
OnProgress: func(p networkquality.Progress) {
_ = server.Send(NewNetworkQualityTestProgress(p))
},
result, nqErr := boxService.urlTestHistoryStorage.RunNetworkQuality(server.Context(), func(ctx context.Context) (*networkquality.Result, error) {
return networkquality.Run(networkquality.Options{
ConfigURL: request.ConfigURL,
HTTPClient: httpClient,
NewMeasurementClient: measurementClientFactory,
Serial: request.Serial,
MaxRuntime: time.Duration(request.MaxRuntimeSeconds) * time.Second,
Context: ctx,
OnProgress: func(p networkquality.Progress) {
_ = server.Send(NewNetworkQualityTestProgress(p))
},
})
})
if nqErr != nil {
return server.Send(&NetworkQualityTestProgress{
@@ -14,6 +14,13 @@
"interval": "",
"tolerance": 0,
"idle_timeout": "",
"strategy": "latency",
"networkquality_ttl": "",
"networkquality_target": "apple",
"networkquality_config_url": "",
"networkquality_download_url": "",
"networkquality_upload_url": "",
"networkquality_probe_url": "",
"interrupt_exist_connections": false
}
```
@@ -42,6 +49,84 @@ The test tolerance in milliseconds. `50` will be used if empty.
The idle timeout. `30m` will be used if empty.
#### strategy
Selection strategy. Available values are `latency` and `networkquality`.
The default `latency` strategy preserves the regular URLTest behavior. The `networkquality` strategy selects the healthy outbound with the highest combined score: `download capacity in bits per second / URLTest delay in milliseconds`. A zero delay is treated as one millisecond. If scores are equal, the lower delay wins. Upload capacity is measured and reported, but does not affect selection. If no usable network quality result exists, selection falls back to latency.
#### networkquality_ttl
The cache lifetime and automatic network quality test interval. `30m` will be used if empty. Values below `30m` are rejected.
With the `networkquality` strategy, the first automatic test starts after the initial latency check. Further automatic tests start after this interval has elapsed since the previous automatic pass completed. Switching from `latency` to `networkquality` at runtime also starts a test immediately and restarts the interval from that pass. The automatic timer remains stopped in the `latency` strategy. Automatic tests use the default maximum runtime and HTTP/3 disabled.
Successful measurements are shared by all URLTest groups in the same sing-box instance. Participants with the same normalized proxy `server:port`, target, maximum runtime, and HTTP/3 mode reuse one measurement during this interval. Only one actual network quality test runs at a time across the entire sing-box instance; different participants and groups wait in the same queue. The last successful download capacity remains eligible for selection regardless of its age. If a later refresh fails, its error is reported while the previous capacities and original timestamp remain visible and eligible until a successful refresh replaces them. Only `server:port` is used as the proxy identity; credentials and transport parameters are intentionally ignored.
A participant without a URLTest delay result is treated as unavailable and is not sent to the network quality service. It still receives a result entry with `"error": "URLTest delay unavailable"`, so the completed map covers every group member. If that participant has an earlier successful capacity, the capacity and its original timestamp are preserved; it cannot be selected while its delay is missing, but the preserved capacity becomes eligible again when a later URLTest restores the delay.
Network quality tests can also be started through the Clash API:
- `POST /group/{name}/networkquality` with `{"maxRuntimeSeconds":20,"http3":false}` processes all group members sequentially, measures those with a URLTest delay, and reports both download and upload results.
- `GET /group/{name}/networkquality` returns the latest results and `progress.current` / `progress.total` for the active group test. `current` is the one-based ordinal of the participant currently being processed; `total` is fixed from the group snapshot at the start of the pass. Both are zero while no test is running.
- `PUT /group/{name}/strategy` with `{"strategy":"networkquality"}` changes the strategy at runtime. Use `latency` to switch back.
During a group test, each participant result becomes visible through `GET /group/{name}/networkquality` as soon as that participant finishes. A partial first-pass map never participates in routing selection, so regular URLTest latency selection remains active until every current group member has a result. A completed map becomes eligible for the combined score, but the network quality test itself never changes the selected outbound; the next complete URLTest group update applies it. Individual Clash API delay responses keep updating each participant's live latency history, but do not switch a `networkquality` group or change the committed delay shown for its selected `now`. The group delay is updated once by the next complete URLTest group update, preventing a dashboard refresh from exposing a partially refreshed latency map. During later speed-test refreshes, the last completed map remains eligible until its replacement is complete. If a pass is canceled, already published participant results remain visible but do not replace the completed selection map.
If the test service returns HTTP `429`, the participant result contains `"rateLimited": true` and an error message. The previous successful capacity values and their original timestamp are preserved and remain eligible for selection regardless of their age. The rate-limit response also starts a new TTL interval, so repeated API calls do not immediately retry the service.
#### networkquality_target
The service used for network quality tests. Available values are:
- `apple` (default): fetch the public Apple Mensura Config v1.
- `cloudflare`: use the built-in Cloudflare speed test endpoints.
- `custom`: fetch Config v1 from [networkquality_config_url](#networkquality_config_url), or use the direct download and upload URLs below.
This setting does not change the regular latency test [url](#url).
#### networkquality_config_url
The Config v1 URL for the `custom` network quality target. Setting this field without `networkquality_target` automatically selects `custom`.
A bare host or a URL with an empty root path is resolved to `https://host/.well-known/nq`. A URL with a non-root path is used unchanged.
This field is mutually exclusive with [networkquality_download_url](#networkquality_download_url), [networkquality_upload_url](#networkquality_upload_url), and [networkquality_probe_url](#networkquality_probe_url).
For stable, controlled measurements, deploy [network-quality/goserver](https://github.com/network-quality/goserver) and set its address here. For example:
```json
{
"networkquality_target": "custom",
"networkquality_config_url": "https://nq.example.com"
}
```
#### networkquality_download_url
The direct download endpoint for the `custom` target. It must be set together with [networkquality_upload_url](#networkquality_upload_url). Setting direct URLs without `networkquality_target` automatically selects `custom`.
The endpoint should return a sufficiently large response body and support concurrent requests for the duration of the measurement.
#### networkquality_upload_url
The direct upload endpoint for the `custom` target. It must accept HTTP `POST` requests with a chunked `application/octet-stream` body.
#### networkquality_probe_url
An optional small download endpoint used for latency and responsiveness probes in direct `custom` mode. If empty, [networkquality_download_url](#networkquality_download_url) is used for probes as well. A dedicated small response is recommended to avoid transferring the full download payload for each probe.
The direct download, upload, and probe URLs must use the same host.
```json
{
"networkquality_target": "custom",
"networkquality_download_url": "https://speed.example.com/download",
"networkquality_upload_url": "https://speed.example.com/upload",
"networkquality_probe_url": "https://speed.example.com/probe"
}
```
#### interrupt_exist_connections
Interrupt existing connections when the selected outbound has changed.
@@ -14,6 +14,13 @@
"interval": "",
"tolerance": 50,
"idle_timeout": "",
"strategy": "latency",
"networkquality_ttl": "",
"networkquality_target": "apple",
"networkquality_config_url": "",
"networkquality_download_url": "",
"networkquality_upload_url": "",
"networkquality_probe_url": "",
"interrupt_exist_connections": false
}
```
@@ -42,8 +49,86 @@
空闲超时。默认使用 `30m`。
#### strategy
选择策略。可用值为 `latency` 和 `networkquality`。
默认的 `latency` 策略保持原有 URLTest 行为。`networkquality` 策略会在健康的出站中选择组合分数最高的成员:`下载容量(比特/秒) / URLTest 延迟(毫秒)`。零延迟按一毫秒处理;分数相同时选择延迟更低的成员。上传容量也会测量并返回,但不参与选择。当没有可用的网络质量结果时,将回退到延迟选择。
#### networkquality_ttl
网络质量测量的缓存有效期和自动测试间隔。默认使用 `30m`,小于 `30m` 的值会被拒绝。
使用 `networkquality` 策略时,首次自动测试会在初始延迟检查完成后启动。之后,每次自动测试完成并经过此间隔后,会启动下一次自动测试。在运行时从 `latency` 切换到 `networkquality` 也会立即启动测试,并从该次测试完成后重新计算间隔。`latency` 策略下自动计时器保持停止。自动测试使用默认最大运行时间,并禁用 HTTP/3。
同一 sing-box 实例中的所有 URLTest 组共享成功的测量结果。代理的规范化 `server:port`、测试目标、最大运行时间和 HTTP/3 模式相同时,在此间隔内复用同一次测量。整个 sing-box 实例同一时刻只执行一个实际网络质量测试;不同参与者和组在同一队列中等待。上一次成功的下载容量无论已经过去多久,都仍可参与选择。如果后续刷新失败,会返回本次错误,同时保留先前的容量及其原始时间戳,直到新的成功刷新替换它们。代理身份仅使用 `server:port`;凭据和传输参数会被有意忽略。
没有 URLTest 延迟结果的成员会被视为不可用,并且不会向网络质量服务发起测速。该成员仍会获得包含 `"error": "URLTest delay unavailable"` 的结果记录,因此完整结果图仍覆盖组内所有成员。如果该成员已有先前成功测得的容量,则会保留容量及其原始时间戳;缺少延迟时它不能被选中,后续 URLTest 恢复延迟后,保留的容量会重新参与选择。
也可以通过 Clash API 运行网络质量测试:
- 使用 `{"maxRuntimeSeconds":20,"http3":false}` 请求 `POST /group/{name}/networkquality`,依次处理组内所有成员,只对具有 URLTest 延迟的成员测速,并返回下载和上传结果。
- `GET /group/{name}/networkquality` 返回最近的结果,以及当前组测试的 `progress.current` / `progress.total`。`current` 是当前正在处理的成员序号(从一开始),`total` 在测试开始时根据组快照固定。没有测试运行时两者均为零。
- 使用 `{"strategy":"networkquality"}` 请求 `PUT /group/{name}/strategy` 可在运行时切换策略。使用 `latency` 可切回延迟策略。
组测试期间,每个成员完成后,其结果会立即通过 `GET /group/{name}/networkquality` 可见。首次测试的部分结果绝不参与路由选择;在当前组的所有成员都有结果之前,仍由常规 URLTest 按延迟选择。完整结果可用于组合评分,但网络质量测试本身不会更改选中的出站;下一次完整的 URLTest 组更新才会应用它。单个 Clash API 延迟响应会继续更新每个成员的实时延迟历史,但不会切换 `networkquality` 组,也不会更改为其选中 `now` 显示的已提交延迟。组延迟仅在下一次完整 URLTest 组更新时统一更新,从而避免仪表板显示部分刷新的延迟结果。后续测速刷新期间,上一次完整结果会继续参与选择,直到新的完整结果替换它。如果测试被取消,已经发布的成员结果仍然可见,但不会替换用于选择的完整结果。
如果测试服务返回 HTTP `429`,参与者结果会包含 `"rateLimited": true` 和错误信息。上一次成功测量的容量值及其原始时间戳会被保留,并且无论结果的年龄如何都仍可参与选择。限流响应也会开始新的 TTL 间隔,因此重复的 API 请求不会立即再次请求测试服务。
#### networkquality_target
用于网络质量测试的服务。可用值:
- `apple`(默认):获取公开的 Apple Mensura Config v1。
- `cloudflare`:使用内置的 Cloudflare 测速端点。
- `custom`:从 [networkquality_config_url](#networkquality_config_url) 获取 Config v1,或使用下面的直接下载和上传 URL。
此设置不会改变普通延迟测试使用的 [url](#url)。
#### networkquality_config_url
`custom` 网络质量目标使用的 Config v1 URL。设置此字段而不设置 `networkquality_target` 时,会自动选择 `custom`。
仅填写主机名或使用根路径 URL 时,会解析为 `https://host/.well-known/nq`。包含非根路径的 URL 会保持不变。
此字段不能与 [networkquality_download_url](#networkquality_download_url)、[networkquality_upload_url](#networkquality_upload_url) 或 [networkquality_probe_url](#networkquality_probe_url) 同时使用。
如需稳定且可控的测量,可部署 [network-quality/goserver](https://github.com/network-quality/goserver),然后在此填写其地址。例如:
```json
{
"networkquality_target": "custom",
"networkquality_config_url": "https://nq.example.com"
}
```
#### networkquality_download_url
`custom` 目标的直接下载端点。必须与 [networkquality_upload_url](#networkquality_upload_url) 一起设置。未设置 `networkquality_target` 时,直接 URL 会自动选择 `custom`。
该端点应返回足够大的响应体,并在测量期间支持并发请求。
#### networkquality_upload_url
`custom` 目标的直接上传端点。必须接受使用分块 `application/octet-stream` 请求体的 HTTP `POST` 请求。
#### networkquality_probe_url
直接 `custom` 模式中用于延迟和响应性探测的可选小型下载端点。如果为空,探测也会使用 [networkquality_download_url](#networkquality_download_url)。建议使用独立的小响应,以避免每次探测都传输完整下载负载。
直接下载、上传和探测 URL 必须使用同一主机。
```json
{
"networkquality_target": "custom",
"networkquality_download_url": "https://speed.example.com/download",
"networkquality_upload_url": "https://speed.example.com/upload",
"networkquality_probe_url": "https://speed.example.com/probe"
}
```
#### interrupt_exist_connections
当选定的出站发生更改时,中断现有连接。
仅入站连接受此设置影响,内部连接将始终被中断。
仅入站连接受此设置影响,内部连接将始终被中断。
+29
View File
@@ -13496,6 +13496,35 @@
"idle_timeout": {
"$ref": "#/$defs/Duration"
},
"strategy": {
"enum": [
"latency",
"networkquality"
]
},
"networkquality_ttl": {
"$ref": "#/$defs/Duration",
"description": "Network quality cache lifetime and automatic test interval. Default and minimum: 30m."
},
"networkquality_target": {
"enum": [
"apple",
"cloudflare",
"custom"
]
},
"networkquality_config_url": {
"type": "string"
},
"networkquality_download_url": {
"type": "string"
},
"networkquality_upload_url": {
"type": "string"
},
"networkquality_probe_url": {
"type": "string"
},
"interrupt_exist_connections": {
"type": "boolean"
}
@@ -2,6 +2,7 @@ package clashapi
import (
"context"
"errors"
"net/http"
"strconv"
"strings"
@@ -23,10 +24,114 @@ func groupRouter(server *Server) http.Handler {
r.Use(parseProxyName, findProxyByName(server))
r.Get("/", getGroup(server))
r.Get("/delay", getGroupDelay(server))
r.Get("/strategy", getGroupStrategy())
r.Put("/strategy", updateGroupStrategy())
r.Get("/networkquality", getGroupNetworkQuality())
r.Post("/networkquality", testGroupNetworkQuality())
})
return r
}
type updateGroupStrategyRequest struct {
Strategy string `json:"strategy"`
}
func getGroupStrategy() func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
urlTestGroup, ok := r.Context().Value(CtxKeyProxy).(adapter.NetworkQualityURLTestGroup)
if !ok {
render.Status(r, http.StatusBadRequest)
render.JSON(w, r, newError("Must be a URLTest group"))
return
}
render.JSON(w, r, render.M{"strategy": urlTestGroup.SelectionStrategy()})
}
}
func updateGroupStrategy() func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
urlTestGroup, ok := r.Context().Value(CtxKeyProxy).(adapter.NetworkQualityURLTestGroup)
if !ok {
render.Status(r, http.StatusBadRequest)
render.JSON(w, r, newError("Must be a URLTest group"))
return
}
var request updateGroupStrategyRequest
if err := render.DecodeJSON(r.Body, &request); err != nil {
render.Status(r, http.StatusBadRequest)
render.JSON(w, r, ErrBadRequest)
return
}
if err := urlTestGroup.SetSelectionStrategy(request.Strategy); err != nil {
render.Status(r, http.StatusBadRequest)
render.JSON(w, r, newError(err.Error()))
return
}
render.JSON(w, r, render.M{"strategy": urlTestGroup.SelectionStrategy()})
}
}
type networkQualityTestRequest struct {
MaxRuntimeSeconds int32 `json:"maxRuntimeSeconds,omitempty"`
HTTP3 bool `json:"http3,omitempty"`
}
func getGroupNetworkQuality() func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
urlTestGroup, ok := r.Context().Value(CtxKeyProxy).(adapter.NetworkQualityURLTestGroup)
if !ok {
render.Status(r, http.StatusBadRequest)
render.JSON(w, r, newError("Must be a URLTest group"))
return
}
render.JSON(w, r, newGroupNetworkQualityResponse(urlTestGroup, urlTestGroup.NetworkQualityHistory()))
}
}
func testGroupNetworkQuality() func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
urlTestGroup, ok := r.Context().Value(CtxKeyProxy).(adapter.NetworkQualityURLTestGroup)
if !ok {
render.Status(r, http.StatusBadRequest)
render.JSON(w, r, newError("Must be a URLTest group"))
return
}
var request networkQualityTestRequest
if err := render.DecodeJSON(r.Body, &request); err != nil || request.MaxRuntimeSeconds < 0 {
render.Status(r, http.StatusBadRequest)
render.JSON(w, r, ErrBadRequest)
return
}
results, err := urlTestGroup.NetworkQuality(r.Context(), time.Duration(request.MaxRuntimeSeconds)*time.Second, request.HTTP3)
if err != nil {
switch {
case errors.Is(err, group.ErrNetworkQualityTestInProgress):
render.Status(r, http.StatusConflict)
case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
render.Status(r, http.StatusGatewayTimeout)
default:
render.Status(r, http.StatusInternalServerError)
}
render.JSON(w, r, newError(err.Error()))
return
}
render.JSON(w, r, newGroupNetworkQualityResponse(urlTestGroup, results))
}
}
func newGroupNetworkQualityResponse(urlTestGroup adapter.NetworkQualityURLTestGroup, results map[string]*adapter.NetworkQualityHistory) render.M {
current, total := urlTestGroup.NetworkQualityProgress()
return render.M{
"strategy": urlTestGroup.SelectionStrategy(),
"now": urlTestGroup.Now(),
"progress": render.M{
"current": current,
"total": total,
},
"results": results,
}
}
func getGroups(server *Server) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
groups := common.Map(common.Filter(server.outbound.Outbounds(), func(it adapter.Outbound) bool {
@@ -71,6 +71,9 @@ func proxyInfo(server *Server, detour adapter.Outbound) *badjson.JSONObject {
info.Put("name", detour.Tag())
info.Put("udp", common.Contains(detour.Network(), N.NetworkUDP))
delayHistory := server.urlTestHistory.LoadURLTestHistory(group.RealTag(server.outbound, detour))
if networkQualityGroup, isNetworkQualityGroup := detour.(adapter.NetworkQualityURLTestGroup); isNetworkQualityGroup && networkQualityGroup.SelectionStrategy() == C.URLTestStrategyNetworkQuality {
delayHistory = networkQualityGroup.NetworkQualitySelectedHistory()
}
if delayHistory != nil {
info.Put("history", []*adapter.URLTestHistory{delayHistory})
} else {
+7
View File
@@ -14,6 +14,13 @@ type URLTestOutboundOptions struct {
Interval badoption.Duration `json:"interval,omitempty"`
Tolerance uint16 `json:"tolerance,omitempty"`
IdleTimeout badoption.Duration `json:"idle_timeout,omitempty"`
Strategy string `json:"strategy,omitempty" enum:"latency,networkquality"`
NetworkQualityTTL badoption.Duration `json:"networkquality_ttl,omitempty"`
NetworkQualityTarget string `json:"networkquality_target,omitempty" enum:"apple,cloudflare,custom"`
NetworkQualityConfigURL string `json:"networkquality_config_url,omitempty"`
NetworkQualityDownloadURL string `json:"networkquality_download_url,omitempty"`
NetworkQualityUploadURL string `json:"networkquality_upload_url,omitempty"`
NetworkQualityProbeURL string `json:"networkquality_probe_url,omitempty"`
InterruptExistConnections bool `json:"interrupt_exist_connections,omitempty"`
}
@@ -0,0 +1,491 @@
package group
import (
"context"
"errors"
"strings"
"time"
"github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing-box/common/dialer"
"github.com/sagernet/sing-box/common/networkquality"
"github.com/sagernet/sing-box/common/urltest"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing/common"
E "github.com/sagernet/sing/common/exceptions"
)
var ErrNetworkQualityTestInProgress = errors.New("network quality test already in progress")
var errNetworkQualityDelayUnavailable = errors.New("URLTest delay unavailable")
const (
networkQualityTargetApple = "apple"
networkQualityTargetCloudflare = "cloudflare"
networkQualityTargetCustom = "custom"
)
type networkQualityTarget struct {
configURL string
config *networkquality.Config
cacheKey string
}
type networkQualityTargetOptions struct {
name string
configURL string
downloadURL string
uploadURL string
probeURL string
}
func newNetworkQualityTarget(options networkQualityTargetOptions) (networkQualityTarget, error) {
name := strings.TrimSpace(options.name)
configURL := strings.TrimSpace(options.configURL)
downloadURL := strings.TrimSpace(options.downloadURL)
uploadURL := strings.TrimSpace(options.uploadURL)
probeURL := strings.TrimSpace(options.probeURL)
hasDirectURLs := downloadURL != "" || uploadURL != "" || probeURL != ""
if name == "" {
if configURL == "" && !hasDirectURLs {
name = networkQualityTargetApple
} else {
name = networkQualityTargetCustom
}
}
switch name {
case networkQualityTargetApple, networkQualityTargetCloudflare:
if configURL != "" || hasDirectURLs {
return networkQualityTarget{}, E.New("custom network quality URLs are only supported by the custom target")
}
if name == networkQualityTargetCloudflare {
return networkQualityTarget{
config: networkquality.CloudflareConfig(),
cacheKey: networkQualityTargetCloudflare,
}, nil
}
return networkQualityTarget{cacheKey: networkQualityTargetApple}, nil
case networkQualityTargetCustom:
if configURL != "" && hasDirectURLs {
return networkQualityTarget{}, E.New("networkquality_config_url and direct network quality URLs are mutually exclusive")
}
if configURL != "" {
return networkQualityTarget{
configURL: configURL,
cacheKey: networkQualityTargetCustom + ":config:" + networkquality.ResolveConfigURL(configURL),
}, nil
}
if downloadURL == "" || uploadURL == "" {
return networkQualityTarget{}, E.New("networkquality_download_url and networkquality_upload_url are required for the custom direct target")
}
if probeURL == "" {
probeURL = downloadURL
}
return networkQualityTarget{
config: &networkquality.Config{
Version: 1,
URLs: networkquality.URLs{
SmallDownloadURL: probeURL,
LargeDownloadURL: downloadURL,
UploadURL: uploadURL,
},
},
cacheKey: strings.Join([]string{networkQualityTargetCustom, "direct", probeURL, downloadURL, uploadURL}, "\x00"),
}, nil
default:
return networkQualityTarget{}, E.New("unknown network quality target: ", name)
}
}
type urlTestSelectionStrategy uint32
const (
urlTestSelectionStrategyLatency urlTestSelectionStrategy = iota
urlTestSelectionStrategyNetworkQuality
)
type networkQualityRunner func(ctx context.Context, outbound adapter.Outbound, maxRuntime time.Duration, http3 bool) (*networkquality.Result, error)
type networkQualityMeasurement struct {
result *networkquality.Result
measuredAt time.Time
err error
}
func parseURLTestSelectionStrategy(strategy string) (urlTestSelectionStrategy, error) {
switch strategy {
case "", C.URLTestStrategyLatency:
return urlTestSelectionStrategyLatency, nil
case C.URLTestStrategyNetworkQuality:
return urlTestSelectionStrategyNetworkQuality, nil
default:
return 0, E.New("unknown URLTest selection strategy: ", strategy)
}
}
func (g *URLTestGroup) SelectionStrategy() string {
if urlTestSelectionStrategy(g.selectionStrategy.Load()) == urlTestSelectionStrategyNetworkQuality {
return C.URLTestStrategyNetworkQuality
}
return C.URLTestStrategyLatency
}
func (g *URLTestGroup) SetSelectionStrategy(strategyName string) error {
return g.setSelectionStrategy(strategyName)
}
func (g *URLTestGroup) setSelectionStrategy(strategyName string) error {
if strategyName == "" {
return E.New("URLTest selection strategy is required")
}
strategy, err := parseURLTestSelectionStrategy(strategyName)
if err != nil {
return err
}
previousStrategy := urlTestSelectionStrategy(g.selectionStrategy.Swap(uint32(strategy)))
g.performUpdateCheck()
if previousStrategy != strategy {
g.access.Lock()
started := g.started
g.access.Unlock()
if started {
select {
case g.automaticNetworkQualityUpdate <- struct{}{}:
default:
}
}
}
return nil
}
func (g *URLTestGroup) loopAutomaticNetworkQuality(ctx context.Context, interval time.Duration, runner networkQualityRunner) {
timer := time.NewTimer(interval)
if !timer.Stop() {
select {
case <-timer.C:
default:
}
}
defer timer.Stop()
var timerC <-chan time.Time
updateSchedule := func() bool {
if !timer.Stop() {
select {
case <-timer.C:
default:
}
}
timerC = nil
if urlTestSelectionStrategy(g.selectionStrategy.Load()) != urlTestSelectionStrategyNetworkQuality {
return true
}
g.runAutomaticNetworkQuality(ctx, runner)
if ctx.Err() != nil {
return false
}
timer.Reset(interval)
timerC = timer.C
return true
}
if !updateSchedule() {
return
}
for {
select {
case <-ctx.Done():
return
case <-g.automaticNetworkQualityUpdate:
case <-timerC:
}
if !updateSchedule() {
return
}
}
}
func (g *URLTestGroup) runAutomaticNetworkQuality(ctx context.Context, runner networkQualityRunner) {
if urlTestSelectionStrategy(g.selectionStrategy.Load()) != urlTestSelectionStrategyNetworkQuality {
return
}
_, err := g.networkQuality(ctx, 0, false, runner)
if err != nil && !errors.Is(err, ErrNetworkQualityTestInProgress) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
g.logger.Debug("automatic network quality test failed: ", err)
}
}
func (g *URLTestGroup) NetworkQuality(ctx context.Context, maxRuntime time.Duration, http3 bool) (map[string]*adapter.NetworkQualityHistory, error) {
return g.networkQuality(ctx, maxRuntime, http3, g.runNetworkQuality)
}
func (g *URLTestGroup) networkQuality(ctx context.Context, maxRuntime time.Duration, http3 bool, runner networkQualityRunner) (map[string]*adapter.NetworkQualityHistory, error) {
if g.networkQualityChecking.Swap(true) {
return nil, ErrNetworkQualityTestInProgress
}
defer g.networkQualityChecking.Store(false)
previousResults := g.NetworkQualityHistory()
results := make(map[string]*adapter.NetworkQualityHistory)
measurements := make(map[urltest.NetworkQualityCacheKey]networkQualityMeasurement)
outbounds := g.outboundsSnapshot()
g.setNetworkQualityProgress(0, len(outbounds))
defer g.setNetworkQualityProgress(0, 0)
for index, detour := range outbounds {
g.setNetworkQualityProgress(index+1, len(outbounds))
if err := ctx.Err(); err != nil {
return nil, err
}
participantTag := detour.Tag()
realTag := RealTag(g.outbound, detour)
if g.history.LoadURLTestHistory(realTag) == nil {
g.logger.Debug("outbound ", participantTag, " network quality skipped: ", errNetworkQualityDelayUnavailable)
results[participantTag] = unavailableDelayNetworkQualityHistory(previousResults[participantTag], realTag)
g.publishNetworkQualityResult(participantTag, results[participantTag])
continue
}
measuredOutbound := detour
if realTag != participantTag && g.outbound != nil {
if realOutbound, loaded := g.outbound.Outbound(realTag); loaded {
measuredOutbound = realOutbound
}
}
cacheKey := g.networkQualityCacheKey(realTag, maxRuntime, http3)
measurement, loaded := measurements[cacheKey]
if !loaded {
measurement.result, measurement.measuredAt, measurement.err = g.history.LoadOrRunNetworkQuality(
ctx,
cacheKey,
g.networkQualityTTL,
func(runCtx context.Context) (*networkquality.Result, error) {
return runner(runCtx, measuredOutbound, maxRuntime, http3)
},
)
measurements[cacheKey] = measurement
}
if ctxErr := ctx.Err(); ctxErr != nil {
return nil, ctxErr
}
if measurement.err != nil {
g.logger.Debug("outbound ", participantTag, " network quality unavailable: ", measurement.err)
results[participantTag] = failedNetworkQualityHistory(
previousResults[participantTag],
measurement,
realTag,
)
} else {
g.logger.Debug("outbound ", participantTag, " network quality download: ", measurement.result.DownloadCapacity, "bps, upload: ", measurement.result.UploadCapacity, "bps")
results[participantTag] = newNetworkQualityHistory(measurement.measuredAt, realTag, measurement.result)
}
g.publishNetworkQualityResult(participantTag, results[participantTag])
}
g.networkQualityAccess.Lock()
g.networkQualityResults = results
g.networkQualitySelectionResults = cloneNetworkQualityHistory(results)
g.networkQualityAccess.Unlock()
return cloneNetworkQualityHistory(results), nil
}
func (g *URLTestGroup) networkQualityCacheKey(realTag string, maxRuntime time.Duration, http3 bool) urltest.NetworkQualityCacheKey {
server := realTag
if serverProvider, loaded := g.outbound.(adapter.OutboundServerProvider); loaded {
if outboundServer, found := serverProvider.OutboundServer(realTag); found {
server = outboundServer
}
}
if maxRuntime == 0 {
maxRuntime = networkquality.DefaultMaxRuntime
}
return urltest.NetworkQualityCacheKey{
Server: server,
Target: g.networkQualityTarget.cacheKey,
MaxRuntime: maxRuntime,
HTTP3: http3,
}
}
func failedNetworkQualityHistory(previous *adapter.NetworkQualityHistory, measurement networkQualityMeasurement, outbound string) *adapter.NetworkQualityHistory {
var history *adapter.NetworkQualityHistory
if measurement.result != nil {
history = newNetworkQualityHistory(measurement.measuredAt, outbound, measurement.result)
} else {
history = preservedNetworkQualityHistory(previous, outbound)
}
history.RateLimited = errors.Is(measurement.err, networkquality.ErrRateLimited)
history.Error = measurement.err.Error()
return history
}
func unavailableDelayNetworkQualityHistory(previous *adapter.NetworkQualityHistory, outbound string) *adapter.NetworkQualityHistory {
history := preservedNetworkQualityHistory(previous, outbound)
history.RateLimited = false
history.Error = errNetworkQualityDelayUnavailable.Error()
return history
}
func preservedNetworkQualityHistory(previous *adapter.NetworkQualityHistory, outbound string) *adapter.NetworkQualityHistory {
if previous != nil && previous.Outbound == outbound && previous.DownloadCapacity > 0 {
cloned := *previous
return &cloned
}
return &adapter.NetworkQualityHistory{
Time: time.Now(),
Outbound: outbound,
}
}
func (g *URLTestGroup) runNetworkQuality(ctx context.Context, outbound adapter.Outbound, maxRuntime time.Duration, http3 bool) (*networkquality.Result, error) {
resolvedDialer := dialer.NewResolveDialer(g.ctx, outbound, true, "", adapter.DNSQueryOptions{}, 0)
httpClient := networkquality.NewHTTPClient(resolvedDialer)
defer httpClient.CloseIdleConnections()
measurementClientFactory, err := networkquality.NewOptionalHTTP3Factory(resolvedDialer, http3)
if err != nil {
return nil, err
}
return networkquality.Run(networkquality.Options{
ConfigURL: g.networkQualityTarget.configURL,
Config: g.networkQualityTarget.config,
HTTPClient: httpClient,
NewMeasurementClient: measurementClientFactory,
Serial: true,
MaxRuntime: maxRuntime,
Context: ctx,
})
}
func (g *URLTestGroup) NetworkQualityHistory() map[string]*adapter.NetworkQualityHistory {
g.networkQualityAccess.RLock()
defer g.networkQualityAccess.RUnlock()
return cloneNetworkQualityHistory(g.networkQualityResults)
}
func (g *URLTestGroup) NetworkQualityProgress() (current int, total int) {
g.networkQualityAccess.RLock()
defer g.networkQualityAccess.RUnlock()
return g.networkQualityCurrent, g.networkQualityTotal
}
func (g *URLTestGroup) NetworkQualitySelectedHistory() *adapter.URLTestHistory {
g.networkQualityAccess.RLock()
defer g.networkQualityAccess.RUnlock()
if g.networkQualitySelectedHistory == nil {
return nil
}
history := *g.networkQualitySelectedHistory
return &history
}
func (g *URLTestGroup) updateNetworkQualitySelectedHistory() {
if urlTestSelectionStrategy(g.selectionStrategy.Load()) != urlTestSelectionStrategyNetworkQuality {
return
}
selected := g.selectedOutboundTCP
if selected == nil {
selected = g.selectedOutboundUDP
}
var history *adapter.URLTestHistory
if selected != nil {
if current := g.history.LoadURLTestHistory(RealTag(g.outbound, selected)); current != nil {
cloned := *current
history = &cloned
}
}
g.networkQualityAccess.Lock()
g.networkQualitySelectedHistory = history
g.networkQualityAccess.Unlock()
}
func (g *URLTestGroup) setNetworkQualityProgress(current int, total int) {
g.networkQualityAccess.Lock()
g.networkQualityCurrent = current
g.networkQualityTotal = total
g.networkQualityAccess.Unlock()
}
func (g *URLTestGroup) publishNetworkQualityResult(tag string, history *adapter.NetworkQualityHistory) {
cloned := *history
g.networkQualityAccess.Lock()
g.networkQualityResults[tag] = &cloned
g.networkQualityAccess.Unlock()
}
func (g *URLTestGroup) selectByNetworkQuality(network string) adapter.Outbound {
outbounds := g.outboundsSnapshot()
g.networkQualityAccess.RLock()
defer g.networkQualityAccess.RUnlock()
for _, detour := range outbounds {
history := g.networkQualitySelectionResults[detour.Tag()]
if history == nil || history.Outbound != RealTag(g.outbound, detour) {
return nil
}
}
var (
selected adapter.Outbound
selectedScore float64
selectedDelay uint16
)
for _, detour := range outbounds {
if !common.Contains(detour.Network(), network) {
continue
}
realTag := RealTag(g.outbound, detour)
delayHistory := g.history.LoadURLTestHistory(realTag)
if delayHistory == nil {
continue
}
history := g.networkQualitySelectionResults[detour.Tag()]
if history == nil || history.Outbound != realTag || history.DownloadCapacity <= 0 {
continue
}
score := networkQualityScore(history.DownloadCapacity, delayHistory.Delay)
if selected == nil || score > selectedScore || score == selectedScore && delayHistory.Delay < selectedDelay {
selected = detour
selectedScore = score
selectedDelay = delayHistory.Delay
}
}
return selected
}
func networkQualityScore(downloadCapacity int64, delay uint16) float64 {
if delay == 0 {
delay = 1
}
return float64(downloadCapacity) / float64(delay)
}
func (g *URLTestGroup) outboundsSnapshot() []adapter.Outbound {
g.access.Lock()
defer g.access.Unlock()
return append([]adapter.Outbound(nil), g.outbounds...)
}
func newNetworkQualityHistory(measuredAt time.Time, outbound string, result *networkquality.Result) *adapter.NetworkQualityHistory {
return &adapter.NetworkQualityHistory{
Time: measuredAt,
Outbound: outbound,
DownloadCapacity: result.DownloadCapacity,
UploadCapacity: result.UploadCapacity,
DownloadRPM: result.DownloadRPM,
UploadRPM: result.UploadRPM,
IdleLatencyMs: result.IdleLatencyMs,
DownloadCapacityAccuracy: int32(result.DownloadCapacityAccuracy),
UploadCapacityAccuracy: int32(result.UploadCapacityAccuracy),
DownloadRPMAccuracy: int32(result.DownloadRPMAccuracy),
UploadRPMAccuracy: int32(result.UploadRPMAccuracy),
}
}
func cloneNetworkQualityHistory(source map[string]*adapter.NetworkQualityHistory) map[string]*adapter.NetworkQualityHistory {
result := make(map[string]*adapter.NetworkQualityHistory, len(source))
for tag, history := range source {
if history == nil {
continue
}
cloned := *history
result[tag] = &cloned
}
return result
}
@@ -0,0 +1,917 @@
package group
import (
"context"
"errors"
"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/interrupt"
"github.com/sagernet/sing-box/common/networkquality"
"github.com/sagernet/sing-box/common/urltest"
C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing-box/log"
M "github.com/sagernet/sing/common/metadata"
N "github.com/sagernet/sing/common/network"
"github.com/sagernet/sing/service"
"github.com/sagernet/sing/service/pause"
)
var _ func(context.Context, adapter.OutboundManager, log.Logger, []adapter.Outbound, string, time.Duration, uint16, time.Duration, bool) (*URLTestGroup, error) = NewURLTestGroup
type networkQualityTestOutbound struct {
outbound.Adapter
}
type networkQualityTestManager struct {
outbounds map[string]adapter.Outbound
servers map[string]string
}
func (m *networkQualityTestManager) Start(adapter.StartStage) error { return nil }
func (m *networkQualityTestManager) Close() error { return nil }
func (m *networkQualityTestManager) Outbounds() []adapter.Outbound {
result := make([]adapter.Outbound, 0, len(m.outbounds))
for _, item := range m.outbounds {
result = append(result, item)
}
return result
}
func (m *networkQualityTestManager) Outbound(tag string) (adapter.Outbound, bool) {
item, loaded := m.outbounds[tag]
return item, loaded
}
func (m *networkQualityTestManager) Default() adapter.Outbound { return nil }
func (m *networkQualityTestManager) Remove(string) error { return nil }
func (m *networkQualityTestManager) Create(context.Context, adapter.Router, log.ContextLogger, string, string, any) error {
return nil
}
func (m *networkQualityTestManager) OutboundServer(tag string) (string, bool) {
server, loaded := m.servers[tag]
return server, loaded
}
func newNetworkQualityTestOutbound(tag string) *networkQualityTestOutbound {
return &networkQualityTestOutbound{
Adapter: outbound.NewAdapter("test", tag, []string{N.NetworkTCP}, nil),
}
}
func (o *networkQualityTestOutbound) DialContext(context.Context, string, M.Socksaddr) (net.Conn, error) {
return nil, errors.New("not implemented")
}
func (o *networkQualityTestOutbound) ListenPacket(context.Context, M.Socksaddr) (net.PacketConn, error) {
return nil, errors.New("not implemented")
}
func newNetworkQualityTestGroup(outbounds ...adapter.Outbound) *URLTestGroup {
ctx, cancel := context.WithCancel(context.Background())
ctx = pause.WithDefaultManager(ctx)
return &URLTestGroup{
ctx: ctx,
logger: log.NewNOPFactory().Logger(),
outbounds: outbounds,
history: urltest.NewHistoryStorage(),
interval: time.Hour,
idleTimeout: time.Hour,
networkQualityTTL: time.Hour,
automaticNetworkQualityCtx: ctx,
automaticNetworkQualityCancel: cancel,
automaticNetworkQualityUpdate: make(chan struct{}, 1),
networkQualityTarget: networkQualityTarget{cacheKey: networkQualityTargetApple},
networkQualityResults: make(map[string]*adapter.NetworkQualityHistory),
networkQualitySelectionResults: make(map[string]*adapter.NetworkQualityHistory),
interruptGroup: interrupt.NewGroup(),
pause: service.FromContext[pause.Manager](ctx),
close: make(chan struct{}),
}
}
func setNetworkQualityTestResults(group *URLTestGroup, results map[string]*adapter.NetworkQualityHistory) {
group.networkQualityResults = cloneNetworkQualityHistory(results)
group.networkQualitySelectionResults = cloneNetworkQualityHistory(results)
}
func storeTestDelay(group *URLTestGroup, tag string, delay uint16) {
group.history.StoreURLTestHistory(tag, &adapter.URLTestHistory{
Time: time.Now(),
Delay: delay,
})
}
func TestURLTestNetworkQualitySelection(t *testing.T) {
lowLatency := newNetworkQualityTestOutbound("low-latency")
balanced := newNetworkQualityTestOutbound("balanced")
highDownload := newNetworkQualityTestOutbound("high-download")
group := newNetworkQualityTestGroup(lowLatency, balanced, highDownload)
storeTestDelay(group, lowLatency.Tag(), 10)
storeTestDelay(group, balanced.Tag(), 30)
storeTestDelay(group, highDownload.Tag(), 100)
setNetworkQualityTestResults(group, map[string]*adapter.NetworkQualityHistory{
lowLatency.Tag(): {
Time: time.Now(),
Outbound: lowLatency.Tag(),
DownloadCapacity: 10,
UploadCapacity: 10_000,
},
balanced.Tag(): {
Time: time.Now(),
Outbound: balanced.Tag(),
DownloadCapacity: 120,
},
highDownload.Tag(): {
Time: time.Now(),
Outbound: highDownload.Tag(),
DownloadCapacity: 200,
UploadCapacity: 1,
},
})
selected, exists := group.Select(N.NetworkTCP)
if !exists || selected != lowLatency {
t.Fatalf("default latency strategy selected %v, want %s", selected, lowLatency.Tag())
}
if err := group.SetSelectionStrategy(C.URLTestStrategyNetworkQuality); err != nil {
t.Fatal(err)
}
selected, exists = group.Select(N.NetworkTCP)
if !exists || selected != balanced {
t.Fatalf("networkquality strategy selected %v, want %s", selected, balanced.Tag())
}
if err := group.SetSelectionStrategy(C.URLTestStrategyLatency); err != nil {
t.Fatal(err)
}
selected, exists = group.Select(N.NetworkTCP)
if !exists || selected != lowLatency {
t.Fatalf("restored latency strategy selected %v, want %s", selected, lowLatency.Tag())
}
}
func TestURLTestNetworkQualityFallbackAndScoreTieBreak(t *testing.T) {
first := newNetworkQualityTestOutbound("first")
second := newNetworkQualityTestOutbound("second")
group := newNetworkQualityTestGroup(first, second)
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
storeTestDelay(group, first.Tag(), 30)
storeTestDelay(group, second.Tag(), 10)
setNetworkQualityTestResults(group, map[string]*adapter.NetworkQualityHistory{
first.Tag(): {
Time: time.Now().Add(-2 * time.Hour),
Outbound: first.Tag(),
DownloadCapacity: 150,
},
second.Tag(): {
Time: time.Now(),
Outbound: second.Tag(),
Error: "test failed",
},
})
selected, exists := group.Select(N.NetworkTCP)
if !exists || selected != first {
t.Fatalf("expired download result selected %v, want %s", selected, first.Tag())
}
setNetworkQualityTestResults(group, map[string]*adapter.NetworkQualityHistory{
first.Tag(): {
Time: time.Now(),
Outbound: first.Tag(),
DownloadCapacity: 150,
},
second.Tag(): {
Time: time.Now(),
Outbound: second.Tag(),
DownloadCapacity: 50,
},
})
selected, exists = group.Select(N.NetworkTCP)
if !exists || selected != second {
t.Fatalf("score tie selected %v, want lower latency %s", selected, second.Tag())
}
}
func TestURLTestNetworkQualityRunsSequentiallyAndStoresBothDirections(t *testing.T) {
first := newNetworkQualityTestOutbound("first")
second := newNetworkQualityTestOutbound("second")
failed := newNetworkQualityTestOutbound("failed")
group := newNetworkQualityTestGroup(first, second, failed)
storeTestDelay(group, first.Tag(), 10)
storeTestDelay(group, second.Tag(), 20)
storeTestDelay(group, failed.Tag(), 30)
var active atomic.Int32
var maximum atomic.Int32
runner := func(_ context.Context, tested adapter.Outbound, _ time.Duration, _ bool) (*networkquality.Result, error) {
current := active.Add(1)
for {
previous := maximum.Load()
if current <= previous || maximum.CompareAndSwap(previous, current) {
break
}
}
defer active.Add(-1)
time.Sleep(time.Millisecond)
if tested.Tag() == failed.Tag() {
return nil, errors.New("measurement failed")
}
return &networkquality.Result{
DownloadCapacity: 100 + int64(len(tested.Tag())),
UploadCapacity: 200 + int64(len(tested.Tag())),
}, nil
}
results, err := group.networkQuality(context.Background(), time.Second, false, runner)
if err != nil {
t.Fatal(err)
}
if maximum.Load() != 1 {
t.Fatalf("maximum concurrent measurements = %d, want 1", maximum.Load())
}
if results[first.Tag()].DownloadCapacity == 0 || results[first.Tag()].UploadCapacity == 0 {
t.Fatalf("both directions were not stored: %+v", results[first.Tag()])
}
if results[failed.Tag()].Error == "" {
t.Fatalf("failed participant has no error: %+v", results[failed.Tag()])
}
stored := group.NetworkQualityHistory()
if len(stored) != 3 || stored[second.Tag()].UploadCapacity != results[second.Tag()].UploadCapacity {
t.Fatalf("published results differ: %+v", stored)
}
}
func TestURLTestNetworkQualitySkipsParticipantWithoutDelay(t *testing.T) {
alive := newNetworkQualityTestOutbound("alive")
missingDelay := newNetworkQualityTestOutbound("missing-delay")
unmeasured := newNetworkQualityTestOutbound("unmeasured")
group := newNetworkQualityTestGroup(alive, missingDelay, unmeasured)
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
storeTestDelay(group, alive.Tag(), 20)
previousTime := time.Now().Add(-time.Hour)
setNetworkQualityTestResults(group, map[string]*adapter.NetworkQualityHistory{
alive.Tag(): {
Time: previousTime,
Outbound: alive.Tag(),
DownloadCapacity: 100,
},
missingDelay.Tag(): {
Time: previousTime,
Outbound: missingDelay.Tag(),
DownloadCapacity: 1_000,
UploadCapacity: 500,
RateLimited: true,
Error: networkquality.ErrRateLimited.Error(),
},
})
var calls atomic.Int32
runner := func(_ context.Context, tested adapter.Outbound, _ time.Duration, _ bool) (*networkquality.Result, error) {
calls.Add(1)
if tested != alive {
t.Fatalf("runner called for participant without delay: %s", tested.Tag())
}
return &networkquality.Result{DownloadCapacity: 200}, nil
}
results, err := group.networkQuality(context.Background(), time.Second, false, runner)
if err != nil {
t.Fatal(err)
}
if calls.Load() != 1 {
t.Fatalf("runner called %d times, want 1", calls.Load())
}
if len(results) != 3 {
t.Fatalf("result count = %d, want 3", len(results))
}
skipped := results[missingDelay.Tag()]
if skipped == nil || skipped.Error != errNetworkQualityDelayUnavailable.Error() || skipped.RateLimited {
t.Fatalf("unexpected skipped result: %+v", skipped)
}
if skipped.DownloadCapacity != 1_000 || skipped.UploadCapacity != 500 || !skipped.Time.Equal(previousTime) {
t.Fatalf("previous successful measurement was not preserved: %+v", skipped)
}
withoutPrevious := results[unmeasured.Tag()]
if withoutPrevious == nil || withoutPrevious.Error != errNetworkQualityDelayUnavailable.Error() || withoutPrevious.Outbound != unmeasured.Tag() || withoutPrevious.DownloadCapacity != 0 {
t.Fatalf("unexpected unmeasured result: %+v", withoutPrevious)
}
selected, exists := group.Select(N.NetworkTCP)
if !exists || selected != alive {
t.Fatalf("participant without delay was selected: %v", selected)
}
storeTestDelay(group, missingDelay.Tag(), 50)
selected, exists = group.Select(N.NetworkTCP)
if !exists || selected != missingDelay {
t.Fatalf("preserved measurement was not reused after delay recovery: %v", selected)
}
}
func TestURLTestNetworkQualityPublishesIncrementallyAndDefersSelectionToURLTest(t *testing.T) {
first := newNetworkQualityTestOutbound("first")
second := newNetworkQualityTestOutbound("second")
group := newNetworkQualityTestGroup(first, second)
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
storeTestDelay(group, first.Tag(), 10)
storeTestDelay(group, second.Tag(), 10)
setNetworkQualityTestResults(group, map[string]*adapter.NetworkQualityHistory{
first.Tag(): {
Time: time.Now(),
Outbound: first.Tag(),
DownloadCapacity: 100,
},
second.Tag(): {
Time: time.Now(),
Outbound: second.Tag(),
DownloadCapacity: 10,
},
})
group.performUpdateCheck()
if group.selectedOutboundTCP != first {
t.Fatalf("initial selected outbound = %v, want %s", group.selectedOutboundTCP, first.Tag())
}
secondStarted := make(chan struct{})
releaseSecond := make(chan struct{})
defer func() {
select {
case <-releaseSecond:
default:
close(releaseSecond)
}
}()
runner := func(_ context.Context, tested adapter.Outbound, _ time.Duration, _ bool) (*networkquality.Result, error) {
if tested == second {
close(secondStarted)
<-releaseSecond
return &networkquality.Result{DownloadCapacity: 200}, nil
}
return &networkquality.Result{DownloadCapacity: 5}, nil
}
testDone := make(chan error, 1)
go func() {
_, err := group.networkQuality(context.Background(), time.Second, false, runner)
testDone <- err
}()
select {
case <-secondStarted:
case <-time.After(time.Second):
t.Fatal("second participant did not start")
}
current, total := group.NetworkQualityProgress()
if current != 2 || total != 2 {
t.Fatalf("progress = %d/%d, want 2/2", current, total)
}
published := group.NetworkQualityHistory()
if published[first.Tag()].DownloadCapacity != 5 || published[second.Tag()].DownloadCapacity != 10 {
t.Fatalf("participant result was not published incrementally: %+v", published)
}
selected, exists := group.Select(N.NetworkTCP)
if !exists || selected != first {
t.Fatalf("selection changed before pass completion: %v", selected)
}
group.performUpdateCheck()
if group.selectedOutboundTCP != first {
t.Fatalf("URLTest selection used a partial pass: %v", group.selectedOutboundTCP)
}
close(releaseSecond)
if err := <-testDone; err != nil {
t.Fatal(err)
}
current, total = group.NetworkQualityProgress()
if current != 0 || total != 0 {
t.Fatalf("completed progress = %d/%d, want 0/0", current, total)
}
if group.selectedOutboundTCP != first {
t.Fatalf("network quality test changed selected outbound: %v", group.selectedOutboundTCP)
}
selected, exists = group.Select(N.NetworkTCP)
if !exists || selected != second {
t.Fatalf("completed results selected %v, want %s", selected, second.Tag())
}
group.performUpdateCheck()
if group.selectedOutboundTCP != second {
t.Fatalf("URLTest selection did not apply completed results: %v", group.selectedOutboundTCP)
}
}
func TestURLTestNetworkQualityRequiresCompleteSnapshot(t *testing.T) {
lowLatency := newNetworkQualityTestOutbound("low-latency")
highDownload := newNetworkQualityTestOutbound("high-download")
group := newNetworkQualityTestGroup(lowLatency, highDownload)
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
storeTestDelay(group, lowLatency.Tag(), 10)
storeTestDelay(group, highDownload.Tag(), 100)
group.networkQualitySelectionResults[highDownload.Tag()] = &adapter.NetworkQualityHistory{
Time: time.Now(),
Outbound: highDownload.Tag(),
DownloadCapacity: 1_000_000,
}
selected, exists := group.Select(N.NetworkTCP)
if !exists || selected != lowLatency {
t.Fatalf("partial network quality snapshot selected %v, want latency winner %s", selected, lowLatency.Tag())
}
}
func TestURLTestNetworkQualityIgnoresIndividualDelayUpdates(t *testing.T) {
first := newNetworkQualityTestOutbound("first")
second := newNetworkQualityTestOutbound("second")
group := newNetworkQualityTestGroup(first, second)
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
setNetworkQualityTestResults(group, map[string]*adapter.NetworkQualityHistory{
first.Tag(): {
Time: time.Now(),
Outbound: first.Tag(),
DownloadCapacity: 100,
},
second.Tag(): {
Time: time.Now(),
Outbound: second.Tag(),
DownloadCapacity: 100,
},
})
storeTestDelay(group, first.Tag(), 10)
storeTestDelay(group, second.Tag(), 100)
group.performUpdateCheck()
if group.selectedOutboundTCP != first {
t.Fatalf("initial selected outbound = %v, want %s", group.selectedOutboundTCP, first.Tag())
}
selectedHistory := group.NetworkQualitySelectedHistory()
if selectedHistory == nil || selectedHistory.Delay != 10 {
t.Fatalf("initial selected history = %+v, want 10ms", selectedHistory)
}
storeTestDelay(group, first.Tag(), 20)
storeTestDelay(group, second.Tag(), 1)
urlTest := &URLTest{group: group}
urlTest.PerformUpdateCheck()
if group.selectedOutboundTCP != first {
t.Fatalf("individual delay update changed selected outbound: %v", group.selectedOutboundTCP)
}
selectedHistory = group.NetworkQualitySelectedHistory()
if selectedHistory == nil || selectedHistory.Delay != 10 {
t.Fatalf("individual delay update changed selected history: %+v", selectedHistory)
}
if liveHistory := group.history.LoadURLTestHistory(first.Tag()); liveHistory == nil || liveHistory.Delay != 20 {
t.Fatalf("individual participant history was not preserved: %+v", liveHistory)
}
group.performUpdateCheck()
if group.selectedOutboundTCP != second {
t.Fatalf("complete URLTest update selected %v, want %s", group.selectedOutboundTCP, second.Tag())
}
selectedHistory = group.NetworkQualitySelectedHistory()
if selectedHistory == nil || selectedHistory.Delay != 1 {
t.Fatalf("complete URLTest update selected history = %+v, want 1ms", selectedHistory)
}
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyLatency))
storeTestDelay(group, first.Tag(), 1)
storeTestDelay(group, second.Tag(), 100)
urlTest.PerformUpdateCheck()
if group.selectedOutboundTCP != first {
t.Fatalf("individual delay update did not update latency group: %v", group.selectedOutboundTCP)
}
}
func TestURLTestNetworkQualityCancellationKeepsSelectionSnapshot(t *testing.T) {
first := newNetworkQualityTestOutbound("first")
second := newNetworkQualityTestOutbound("second")
group := newNetworkQualityTestGroup(first, second)
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
storeTestDelay(group, first.Tag(), 10)
storeTestDelay(group, second.Tag(), 10)
setNetworkQualityTestResults(group, map[string]*adapter.NetworkQualityHistory{
first.Tag(): {
Time: time.Now(),
Outbound: first.Tag(),
DownloadCapacity: 42,
},
second.Tag(): {
Time: time.Now(),
Outbound: second.Tag(),
DownloadCapacity: 100,
},
})
ctx, cancel := context.WithCancel(context.Background())
runner := func(_ context.Context, tested adapter.Outbound, _ time.Duration, _ bool) (*networkquality.Result, error) {
if tested == first {
return &networkquality.Result{DownloadCapacity: 200}, nil
}
cancel()
return &networkquality.Result{DownloadCapacity: 1}, nil
}
if _, err := group.networkQuality(ctx, time.Second, false, runner); !errors.Is(err, context.Canceled) {
t.Fatalf("error = %v, want context.Canceled", err)
}
current, total := group.NetworkQualityProgress()
if current != 0 || total != 0 {
t.Fatalf("canceled progress = %d/%d, want 0/0", current, total)
}
stored := group.NetworkQualityHistory()
if stored[first.Tag()].DownloadCapacity != 200 || stored[second.Tag()].DownloadCapacity != 100 {
t.Fatalf("completed participant was not published: %+v", stored)
}
selected, exists := group.Select(N.NetworkTCP)
if !exists || selected != second {
t.Fatalf("canceled pass changed selection snapshot: %v", selected)
}
}
func TestURLTestNetworkQualityDeduplicatesAcrossGroupsByServer(t *testing.T) {
first := newNetworkQualityTestOutbound("first")
second := newNetworkQualityTestOutbound("second")
manager := &networkQualityTestManager{
outbounds: map[string]adapter.Outbound{
first.Tag(): first,
second.Tag(): second,
},
servers: map[string]string{
first.Tag(): "example.com:443",
second.Tag(): "example.com:443",
},
}
sharedHistory := urltest.NewHistoryStorage()
firstGroup := newNetworkQualityTestGroup(first)
firstGroup.outbound = manager
firstGroup.history = sharedHistory
secondGroup := newNetworkQualityTestGroup(second)
secondGroup.outbound = manager
secondGroup.history = sharedHistory
storeTestDelay(firstGroup, first.Tag(), 10)
storeTestDelay(secondGroup, second.Tag(), 10)
var calls atomic.Int32
runner := func(context.Context, adapter.Outbound, time.Duration, bool) (*networkquality.Result, error) {
calls.Add(1)
return &networkquality.Result{DownloadCapacity: 100}, nil
}
if _, err := firstGroup.networkQuality(context.Background(), time.Second, false, runner); err != nil {
t.Fatal(err)
}
results, err := secondGroup.networkQuality(context.Background(), time.Second, false, runner)
if err != nil {
t.Fatal(err)
}
if calls.Load() != 1 {
t.Fatalf("runner called %d times", calls.Load())
}
if results[second.Tag()].DownloadCapacity != 100 || results[second.Tag()].Outbound != second.Tag() {
t.Fatalf("unexpected reused result: %+v", results[second.Tag()])
}
}
func TestURLTestNetworkQualityRateLimitPreservesPreviousResult(t *testing.T) {
detour := newNetworkQualityTestOutbound("proxy")
group := newNetworkQualityTestGroup(detour)
storeTestDelay(group, detour.Tag(), 20)
var calls atomic.Int32
runner := func(context.Context, adapter.Outbound, time.Duration, bool) (*networkquality.Result, error) {
if calls.Add(1) == 1 {
return &networkquality.Result{DownloadCapacity: 100, UploadCapacity: 50}, nil
}
return nil, networkquality.ErrRateLimited
}
first, err := group.networkQuality(context.Background(), time.Second, false, runner)
if err != nil {
t.Fatal(err)
}
group.networkQualityTTL = -time.Nanosecond
limited, err := group.networkQuality(context.Background(), time.Second, false, runner)
if err != nil {
t.Fatal(err)
}
history := limited[detour.Tag()]
if !history.RateLimited || history.Error == "" || history.DownloadCapacity != 100 || history.UploadCapacity != 50 || !history.Time.Equal(first[detour.Tag()].Time) {
t.Fatalf("previous result was not preserved: %+v", history)
}
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
selected, exists := group.Select(N.NetworkTCP)
if !exists || selected != detour {
t.Fatalf("rate-limited cached result was not selected: %v", selected)
}
group.networkQualityTTL = time.Hour
limited, err = group.networkQuality(context.Background(), time.Second, false, runner)
if err != nil {
t.Fatal(err)
}
history = limited[detour.Tag()]
if calls.Load() != 2 || !history.RateLimited || history.DownloadCapacity != 100 || !history.Time.Equal(first[detour.Tag()].Time) {
t.Fatalf("rate limit cooldown was not reused: calls=%d history=%+v", calls.Load(), history)
}
}
func TestURLTestNetworkQualityFailurePreservesPreviousResult(t *testing.T) {
detour := newNetworkQualityTestOutbound("proxy")
group := newNetworkQualityTestGroup(detour)
storeTestDelay(group, detour.Tag(), 20)
var calls atomic.Int32
runner := func(context.Context, adapter.Outbound, time.Duration, bool) (*networkquality.Result, error) {
if calls.Add(1) == 1 {
return &networkquality.Result{DownloadCapacity: 100, UploadCapacity: 50}, nil
}
return nil, errors.New("temporary measurement failure")
}
first, err := group.networkQuality(context.Background(), time.Second, false, runner)
if err != nil {
t.Fatal(err)
}
group.networkQualityTTL = -time.Nanosecond
failed, err := group.networkQuality(context.Background(), time.Second, false, runner)
if err != nil {
t.Fatal(err)
}
history := failed[detour.Tag()]
if history.RateLimited || history.Error == "" || history.DownloadCapacity != 100 || history.UploadCapacity != 50 || !history.Time.Equal(first[detour.Tag()].Time) {
t.Fatalf("previous result was not preserved after refresh failure: %+v", history)
}
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
selected, exists := group.Select(N.NetworkTCP)
if !exists || selected != detour {
t.Fatalf("preserved result was not selected after refresh failure: %v", selected)
}
}
func TestURLTestTouchCloseConcurrent(t *testing.T) {
for range 100 {
group := newNetworkQualityTestGroup()
group.started = true
start := make(chan struct{})
var waitGroup sync.WaitGroup
waitGroup.Add(2)
go func() {
defer waitGroup.Done()
<-start
group.Touch()
}()
go func() {
defer waitGroup.Done()
<-start
_ = group.Close()
}()
close(start)
waitGroup.Wait()
group.access.Lock()
started := group.started
ticker := group.ticker
group.access.Unlock()
if started || ticker != nil {
t.Fatalf("group resumed after close: started=%v ticker=%v", started, ticker)
}
}
}
func TestURLTestSelectionStrategyValidation(t *testing.T) {
group := newNetworkQualityTestGroup()
if err := group.SetSelectionStrategy(""); err == nil {
t.Fatal("empty strategy accepted")
}
if err := group.SetSelectionStrategy("unknown"); err == nil {
t.Fatal("unknown strategy accepted")
}
}
func TestURLTestNetworkQualityTTLLimits(t *testing.T) {
ctx := service.ContextWithPtr(context.Background(), urltest.NewHistoryStorage())
logger := log.NewNOPFactory().Logger()
for _, test := range []struct {
name string
ttl time.Duration
wantTTL time.Duration
wantErr bool
}{
{name: "default", wantTTL: C.DefaultNetworkQualityTTL},
{name: "below minimum", ttl: C.MinNetworkQualityTTL - time.Second, wantErr: true},
{name: "minimum", ttl: C.MinNetworkQualityTTL, wantTTL: C.MinNetworkQualityTTL},
{name: "custom", ttl: time.Hour, wantTTL: time.Hour},
} {
t.Run(test.name, func(t *testing.T) {
group, err := newURLTestGroup(ctx, nil, logger, nil, "", 0, 0, 0, C.URLTestStrategyNetworkQuality, test.ttl, networkQualityTargetOptions{}, false)
if test.wantErr {
if err == nil {
t.Fatal("networkquality_ttl below minimum accepted")
}
return
}
if err != nil {
t.Fatal(err)
}
defer group.Close()
if group.networkQualityTTL != test.wantTTL {
t.Fatalf("networkquality_ttl = %v, want %v", group.networkQualityTTL, test.wantTTL)
}
})
}
}
func TestURLTestAutomaticNetworkQualityLoop(t *testing.T) {
detour := newNetworkQualityTestOutbound("proxy")
group := newNetworkQualityTestGroup(detour)
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
group.networkQualityTTL = -time.Nanosecond
storeTestDelay(group, detour.Tag(), 10)
called := make(chan struct{}, 3)
runner := func(context.Context, adapter.Outbound, time.Duration, bool) (*networkquality.Result, error) {
called <- struct{}{}
return &networkquality.Result{DownloadCapacity: 100}, nil
}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() {
group.loopAutomaticNetworkQuality(ctx, 5*time.Millisecond, runner)
close(done)
}()
for range 2 {
select {
case <-called:
case <-time.After(time.Second):
t.Fatal("automatic network quality test was not run")
}
}
cancel()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("automatic network quality loop did not stop")
}
}
func TestURLTestAutomaticNetworkQualityStrategySwitch(t *testing.T) {
detour := newNetworkQualityTestOutbound("proxy")
group := newNetworkQualityTestGroup(detour)
group.started = true
storeTestDelay(group, detour.Tag(), 10)
called := make(chan struct{}, 1)
runner := func(context.Context, adapter.Outbound, time.Duration, bool) (*networkquality.Result, error) {
called <- struct{}{}
return &networkquality.Result{DownloadCapacity: 100}, nil
}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() {
group.loopAutomaticNetworkQuality(ctx, time.Hour, runner)
close(done)
}()
if err := group.setSelectionStrategy(C.URLTestStrategyNetworkQuality); err != nil {
t.Fatal(err)
}
select {
case <-called:
case <-time.After(time.Second):
t.Fatal("switching to networkquality did not start an automatic test")
}
cancel()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("automatic network quality loop did not stop")
}
}
func TestURLTestAutomaticNetworkQualityStrategySwitchResetsInterval(t *testing.T) {
detour := newNetworkQualityTestOutbound("proxy")
group := newNetworkQualityTestGroup(detour)
group.started = true
group.selectionStrategy.Store(uint32(urlTestSelectionStrategyNetworkQuality))
group.networkQualityTTL = -time.Nanosecond
storeTestDelay(group, detour.Tag(), 10)
const interval = 120 * time.Millisecond
called := make(chan time.Time, 4)
runner := func(context.Context, adapter.Outbound, time.Duration, bool) (*networkquality.Result, error) {
called <- time.Now()
return &networkquality.Result{DownloadCapacity: 100}, nil
}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() {
group.loopAutomaticNetworkQuality(ctx, interval, runner)
close(done)
}()
select {
case <-called:
case <-time.After(time.Second):
t.Fatal("initial automatic network quality test was not run")
}
if err := group.setSelectionStrategy(C.URLTestStrategyLatency); err != nil {
t.Fatal(err)
}
time.Sleep(90 * time.Millisecond)
if err := group.setSelectionStrategy(C.URLTestStrategyNetworkQuality); err != nil {
t.Fatal(err)
}
select {
case <-called:
case <-time.After(time.Second):
t.Fatal("runtime strategy switch did not run an immediate test")
}
select {
case <-called:
t.Fatal("automatic interval was not reset after runtime strategy switch")
case <-time.After(interval / 2):
}
select {
case <-called:
case <-time.After(interval):
t.Fatal("automatic network quality test did not resume after reset interval")
}
cancel()
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("automatic network quality loop did not stop")
}
}
func TestURLTestAutomaticNetworkQualitySkipsLatencyStrategy(t *testing.T) {
detour := newNetworkQualityTestOutbound("proxy")
group := newNetworkQualityTestGroup(detour)
var calls atomic.Int32
group.runAutomaticNetworkQuality(context.Background(), func(context.Context, adapter.Outbound, time.Duration, bool) (*networkquality.Result, error) {
calls.Add(1)
return &networkquality.Result{DownloadCapacity: 100}, nil
})
if calls.Load() != 0 {
t.Fatalf("latency strategy started %d automatic tests", calls.Load())
}
}
func TestNetworkQualityTarget(t *testing.T) {
apple, err := newNetworkQualityTarget(networkQualityTargetOptions{})
if err != nil {
t.Fatal(err)
}
if apple.config != nil || apple.configURL != "" || apple.cacheKey != networkQualityTargetApple {
t.Fatalf("unexpected Apple target: %+v", apple)
}
cloudflare, err := newNetworkQualityTarget(networkQualityTargetOptions{name: networkQualityTargetCloudflare})
if err != nil {
t.Fatal(err)
}
if cloudflare.config == nil || cloudflare.config.URLs.LargeDownloadURL != "https://h3.speed.cloudflare.com/__down?bytes=10000000000" || cloudflare.cacheKey != networkQualityTargetCloudflare {
t.Fatalf("unexpected Cloudflare target: %+v", cloudflare)
}
custom, err := newNetworkQualityTarget(networkQualityTargetOptions{configURL: "nq.example.com"})
if err != nil {
t.Fatal(err)
}
if custom.configURL != "nq.example.com" || custom.config != nil || custom.cacheKey != "custom:config:https://nq.example.com/.well-known/nq" {
t.Fatalf("unexpected custom target: %+v", custom)
}
direct, err := newNetworkQualityTarget(networkQualityTargetOptions{
downloadURL: "https://nq.example.com/download",
uploadURL: "https://nq.example.com/upload",
})
if err != nil {
t.Fatal(err)
}
if direct.configURL != "" || direct.config == nil || direct.config.URLs.SmallDownloadURL != "https://nq.example.com/download" || direct.config.URLs.LargeDownloadURL != "https://nq.example.com/download" || direct.config.URLs.UploadURL != "https://nq.example.com/upload" {
t.Fatalf("unexpected direct custom target: %+v", direct)
}
directWithProbe, err := newNetworkQualityTarget(networkQualityTargetOptions{
downloadURL: "https://nq.example.com/download",
uploadURL: "https://nq.example.com/upload",
probeURL: "https://nq.example.com/probe",
})
if err != nil {
t.Fatal(err)
}
if directWithProbe.config.URLs.SmallDownloadURL != "https://nq.example.com/probe" {
t.Fatalf("unexpected direct probe target: %+v", directWithProbe)
}
}
func TestNetworkQualityTargetValidation(t *testing.T) {
for _, options := range []networkQualityTargetOptions{
{name: networkQualityTargetApple, configURL: "https://example.com/config"},
{name: networkQualityTargetCloudflare, downloadURL: "https://example.com/download", uploadURL: "https://example.com/upload"},
{name: networkQualityTargetCustom},
{name: networkQualityTargetCustom, configURL: "https://example.com/config", downloadURL: "https://example.com/download", uploadURL: "https://example.com/upload"},
{name: networkQualityTargetCustom, downloadURL: "https://example.com/download"},
{name: "unknown"},
} {
if _, err := newNetworkQualityTarget(options); err == nil {
t.Fatalf("target options accepted: %+v", options)
}
}
}
+188 -47
View File
@@ -31,8 +31,8 @@ func RegisterURLTest(registry *outbound.Registry) {
}
var (
_ adapter.OutboundGroup = (*URLTest)(nil)
_ adapter.InterfaceUpdateListener = (*URLTest)(nil)
_ adapter.NetworkQualityURLTestGroup = (*URLTest)(nil)
_ adapter.InterfaceUpdateListener = (*URLTest)(nil)
)
type URLTest struct {
@@ -46,6 +46,13 @@ type URLTest struct {
interval time.Duration
tolerance uint16
idleTimeout time.Duration
strategy string
networkQualityTTL time.Duration
networkQualityTarget string
networkQualityConfigURL string
networkQualityDownloadURL string
networkQualityUploadURL string
networkQualityProbeURL string
group *URLTestGroup
checkAccess sync.Mutex
interruptExternalConnections bool
@@ -62,6 +69,13 @@ type URLTest struct {
}
func NewURLTest(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.URLTestOutboundOptions) (adapter.Outbound, error) {
strategy := options.Strategy
if strategy == "" {
strategy = C.URLTestStrategyLatency
}
if _, err := parseURLTestSelectionStrategy(strategy); err != nil {
return nil, err
}
outbound := &URLTest{
Adapter: outbound.NewAdapter(C.TypeURLTest, tag, []string{N.NetworkTCP, N.NetworkUDP}, options.Outbounds),
ctx: ctx,
@@ -73,6 +87,13 @@ func NewURLTest(ctx context.Context, router adapter.Router, logger log.ContextLo
interval: time.Duration(options.Interval),
tolerance: options.Tolerance,
idleTimeout: time.Duration(options.IdleTimeout),
strategy: strategy,
networkQualityTTL: time.Duration(options.NetworkQualityTTL),
networkQualityTarget: options.NetworkQualityTarget,
networkQualityConfigURL: options.NetworkQualityConfigURL,
networkQualityDownloadURL: options.NetworkQualityDownloadURL,
networkQualityUploadURL: options.NetworkQualityUploadURL,
networkQualityProbeURL: options.NetworkQualityProbeURL,
interruptExternalConnections: options.InterruptExistConnections,
provider: service.FromContext[adapter.ProviderManager](ctx),
@@ -123,7 +144,26 @@ func (s *URLTest) Start() error {
s.tags = append(s.tags, detour.Tag())
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.strategy,
s.networkQualityTTL,
networkQualityTargetOptions{
name: s.networkQualityTarget,
configURL: s.networkQualityConfigURL,
downloadURL: s.networkQualityDownloadURL,
uploadURL: s.networkQualityUploadURL,
probeURL: s.networkQualityProbeURL,
},
s.interruptExternalConnections,
)
if err != nil {
return err
}
@@ -159,11 +199,38 @@ func (s *URLTest) URLTest(ctx context.Context) (map[string]uint16, error) {
return s.group.URLTest(ctx)
}
func (s *URLTest) NetworkQuality(ctx context.Context, maxRuntime time.Duration, http3 bool) (map[string]*adapter.NetworkQualityHistory, error) {
return s.group.NetworkQuality(ctx, maxRuntime, http3)
}
func (s *URLTest) NetworkQualityHistory() map[string]*adapter.NetworkQualityHistory {
return s.group.NetworkQualityHistory()
}
func (s *URLTest) NetworkQualityProgress() (current int, total int) {
return s.group.NetworkQualityProgress()
}
func (s *URLTest) NetworkQualitySelectedHistory() *adapter.URLTestHistory {
return s.group.NetworkQualitySelectedHistory()
}
func (s *URLTest) SelectionStrategy() string {
return s.group.SelectionStrategy()
}
func (s *URLTest) SetSelectionStrategy(strategy string) error {
return s.group.SetSelectionStrategy(strategy)
}
func (s *URLTest) CheckOutbounds() {
s.group.CheckOutbounds(s.ctx, true)
}
func (s *URLTest) PerformUpdateCheck() {
if s.group.SelectionStrategy() == C.URLTestStrategyNetworkQuality {
return
}
s.group.performUpdateCheck()
}
@@ -283,8 +350,9 @@ func (s *URLTest) onProviderUpdated(tag string) error {
tags = append(tags, detour.Tag())
outbounds = append(outbounds, detour)
}
s.tags, s.group.outbounds = tags, outbounds
s.tags = tags
s.group.access.Lock()
s.group.outbounds = outbounds
if s.group.ticker != nil {
s.group.ticker.Reset(s.group.interval)
}
@@ -299,31 +367,61 @@ func (s *URLTest) onProviderUpdated(tag string) error {
}
type URLTestGroup struct {
ctx context.Context
outbound adapter.OutboundManager
pause pause.Manager
pauseCallback *list.Element[pause.Callback]
logger log.Logger
outbounds []adapter.Outbound
link string
interval time.Duration
tolerance uint16
idleTimeout time.Duration
history *urltest.HistoryStorage
checking atomic.Bool
selectedOutboundTCP adapter.Outbound
selectedOutboundUDP adapter.Outbound
interruptGroup *interrupt.Group
interruptExternalConnections bool
access sync.Mutex
updateAccess sync.Mutex
ticker *time.Ticker
close chan struct{}
started bool
lastActive common.TypedValue[time.Time]
ctx context.Context
outbound adapter.OutboundManager
pause pause.Manager
pauseCallback *list.Element[pause.Callback]
logger log.Logger
outbounds []adapter.Outbound
link string
interval time.Duration
tolerance uint16
idleTimeout time.Duration
history *urltest.HistoryStorage
checking atomic.Bool
networkQualityChecking atomic.Bool
selectionStrategy atomic.Uint32
networkQualityTTL time.Duration
automaticNetworkQualityCtx context.Context
automaticNetworkQualityCancel context.CancelFunc
automaticNetworkQualityUpdate chan struct{}
networkQualityTarget networkQualityTarget
networkQualityAccess sync.RWMutex
networkQualityResults map[string]*adapter.NetworkQualityHistory
networkQualitySelectionResults map[string]*adapter.NetworkQualityHistory
networkQualityCurrent int
networkQualityTotal int
networkQualitySelectedHistory *adapter.URLTestHistory
selectedOutboundTCP adapter.Outbound
selectedOutboundUDP adapter.Outbound
interruptGroup *interrupt.Group
interruptExternalConnections bool
access sync.Mutex
updateAccess sync.Mutex
ticker *time.Ticker
close chan struct{}
started bool
lastActive common.TypedValue[time.Time]
}
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) {
return newURLTestGroup(
ctx,
outboundManager,
logger,
outbounds,
link,
interval,
tolerance,
idleTimeout,
C.URLTestStrategyLatency,
C.DefaultNetworkQualityTTL,
networkQualityTargetOptions{},
interruptExternalConnections,
)
}
func newURLTestGroup(ctx context.Context, outboundManager adapter.OutboundManager, logger log.Logger, outbounds []adapter.Outbound, link string, interval time.Duration, tolerance uint16, idleTimeout time.Duration, strategyName string, networkQualityTTL time.Duration, targetOptions networkQualityTargetOptions, interruptExternalConnections bool) (*URLTestGroup, error) {
if interval == 0 {
interval = C.DefaultURLTestInterval
}
@@ -333,28 +431,52 @@ func NewURLTestGroup(ctx context.Context, outboundManager adapter.OutboundManage
if idleTimeout == 0 {
idleTimeout = C.DefaultURLTestIdleTimeout
}
if networkQualityTTL == 0 {
networkQualityTTL = C.DefaultNetworkQualityTTL
}
if networkQualityTTL < C.MinNetworkQualityTTL {
return nil, E.New("networkquality_ttl must be at least ", C.MinNetworkQualityTTL)
}
if interval > idleTimeout {
return nil, E.New("interval must be less or equal than idle_timeout")
}
strategy, err := parseURLTestSelectionStrategy(strategyName)
if err != nil {
return nil, err
}
networkQualityTarget, err := newNetworkQualityTarget(targetOptions)
if err != nil {
return nil, err
}
history := service.PtrFromContext[urltest.HistoryStorage](ctx)
if history == nil {
return nil, E.New("missing URL test history storage")
}
return &URLTestGroup{
ctx: ctx,
outbound: outboundManager,
logger: logger,
outbounds: outbounds,
link: link,
interval: interval,
tolerance: tolerance,
idleTimeout: idleTimeout,
history: history,
close: make(chan struct{}),
pause: service.FromContext[pause.Manager](ctx),
interruptGroup: interrupt.NewGroup(),
interruptExternalConnections: interruptExternalConnections,
}, nil
automaticNetworkQualityCtx, automaticNetworkQualityCancel := context.WithCancel(ctx)
group := &URLTestGroup{
ctx: ctx,
outbound: outboundManager,
logger: logger,
outbounds: outbounds,
link: link,
interval: interval,
tolerance: tolerance,
idleTimeout: idleTimeout,
networkQualityTTL: networkQualityTTL,
automaticNetworkQualityCtx: automaticNetworkQualityCtx,
automaticNetworkQualityCancel: automaticNetworkQualityCancel,
automaticNetworkQualityUpdate: make(chan struct{}, 1),
networkQualityTarget: networkQualityTarget,
networkQualityResults: make(map[string]*adapter.NetworkQualityHistory),
networkQualitySelectionResults: make(map[string]*adapter.NetworkQualityHistory),
history: history,
close: make(chan struct{}),
pause: service.FromContext[pause.Manager](ctx),
interruptGroup: interrupt.NewGroup(),
interruptExternalConnections: interruptExternalConnections,
}
group.selectionStrategy.Store(uint32(strategy))
return group, nil
}
func (g *URLTestGroup) PostStart() {
@@ -362,15 +484,18 @@ func (g *URLTestGroup) PostStart() {
defer g.access.Unlock()
g.started = true
g.lastActive.Store(time.Now())
go g.CheckOutbounds(g.ctx, false)
go func() {
g.CheckOutbounds(g.ctx, false)
g.loopAutomaticNetworkQuality(g.automaticNetworkQualityCtx, g.networkQualityTTL, g.runNetworkQuality)
}()
}
func (g *URLTestGroup) Touch() {
g.access.Lock()
defer g.access.Unlock()
if !g.started {
return
}
g.access.Lock()
defer g.access.Unlock()
if g.ticker != nil {
g.lastActive.Store(time.Now())
return
@@ -384,6 +509,11 @@ func (g *URLTestGroup) Touch() {
func (g *URLTestGroup) Close() error {
g.access.Lock()
defer g.access.Unlock()
g.started = false
if g.automaticNetworkQualityCancel != nil {
g.automaticNetworkQualityCancel()
g.automaticNetworkQualityCancel = nil
}
if g.ticker == nil {
return nil
}
@@ -396,8 +526,18 @@ func (g *URLTestGroup) Close() error {
}
func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) {
if urlTestSelectionStrategy(g.selectionStrategy.Load()) == urlTestSelectionStrategyNetworkQuality {
if selected := g.selectByNetworkQuality(network); selected != nil {
return selected, true
}
}
return g.selectByLatency(network)
}
func (g *URLTestGroup) selectByLatency(network string) (adapter.Outbound, bool) {
var minDelay uint16
var minOutbound adapter.Outbound
outbounds := g.outboundsSnapshot()
switch network {
case N.NetworkTCP:
if g.selectedOutboundTCP != nil {
@@ -414,7 +554,7 @@ func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) {
}
}
}
for _, detour := range g.outbounds {
for _, detour := range outbounds {
if !common.Contains(detour.Network(), network) {
continue
}
@@ -428,7 +568,7 @@ func (g *URLTestGroup) Select(network string) (adapter.Outbound, bool) {
}
}
if minOutbound == nil {
for _, detour := range g.outbounds {
for _, detour := range outbounds {
if !common.Contains(detour.Network(), network) {
continue
}
@@ -478,7 +618,7 @@ func (g *URLTestGroup) urlTest(ctx context.Context, force bool) (map[string]uint
return make(map[string]uint16), nil
}
defer g.checking.Store(false)
result := URLTestOutbounds(ctx, g.outbound, g.history, g.logger, g.outbounds, g.link, g.interval, force)
result := URLTestOutbounds(ctx, g.outbound, g.history, g.logger, g.outboundsSnapshot(), g.link, g.interval, force)
select {
case <-ctx.Done():
default:
@@ -605,6 +745,7 @@ func (g *URLTestGroup) performUpdateCheck() {
}
g.selectedOutboundUDP = outbound
}
g.updateNetworkQualitySelectedHistory()
if updated {
g.interruptGroup.Interrupt(g.interruptExternalConnections)
}