From 73f01167de523c2bd2b62f9c57fff5c149d40833 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sat, 1 Aug 2026 20:36:58 +0300 Subject: [PATCH 1/8] Add compact progressive observatory probing Drive unchanged BurstObservatory checks in bounded profile groups and publish only the affected profile result instead of repeatedly serializing complete batch snapshots. --- libv2ray_main.go | 301 ++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 300 insertions(+), 1 deletion(-) diff --git a/libv2ray_main.go b/libv2ray_main.go index 1941b71c..4cfb1912 100644 --- a/libv2ray_main.go +++ b/libv2ray_main.go @@ -2,6 +2,7 @@ package libv2ray import ( "context" + "encoding/json" "errors" "fmt" "io" @@ -16,11 +17,14 @@ import ( "time" coreapplog "github.com/xtls/xray-core/app/log" + coreobservatory "github.com/xtls/xray-core/app/observatory" corecommlog "github.com/xtls/xray-core/common/log" corenet "github.com/xtls/xray-core/common/net" corefilesystem "github.com/xtls/xray-core/common/platform/filesystem" "github.com/xtls/xray-core/common/serial" core "github.com/xtls/xray-core/core" + coreextension "github.com/xtls/xray-core/features/extension" + corerouting "github.com/xtls/xray-core/features/routing" corestats "github.com/xtls/xray-core/features/stats" coreserial "github.com/xtls/xray-core/infra/conf/serial" _ "github.com/xtls/xray-core/main/distro/all" @@ -35,9 +39,43 @@ const ( xudpBaseKey = "xray.xudp.basekey" tunFdKey = "xray.tun.fd" browserDialerAddress = "xray.browser.dialer" - libVersion = 39 // Library version, update here only + libVersion = 40 // Library version, update here only ) +// OutboundProbeHandler receives one compact update for the affected UI group. +// Calls are serialized even though the underlying checks run concurrently. +type OutboundProbeHandler interface { + OnOutboundProbeResult(groupID string, delay int64, alive, completed bool) int +} + +type outboundProbeGroup struct { + GUID string `json:"guid"` + OutboundTags []string `json:"outboundTags"` + BalancerTag string `json:"balancerTag"` +} + +// OutboundProbeController owns one finite probe batch. v2rayNG runs it in a +// disposable process so Xray's process-wide native state cannot overlap the +// long-running VPN core or a later test batch. +type OutboundProbeController struct { + access sync.Mutex + cancel context.CancelFunc + used bool +} + +func NewOutboundProbeController() *OutboundProbeController { + return &OutboundProbeController{} +} + +func (c *OutboundProbeController) Cancel() { + c.access.Lock() + cancel := c.cancel + c.access.Unlock() + if cancel != nil { + cancel() + } +} + // CoreController represents a controller for managing Xray core instance lifecycle type CoreController struct { CallbackHandler CoreCallbackHandler @@ -230,6 +268,267 @@ func MeasureOutboundDelay(ConfigureFileContent string, url string) (int64, error return measureInstDelay(context.Background(), inst, url) } +// Probe runs all UI delay-test groups through one short-lived Xray instance. +// maxConcurrency limits active UI profiles; candidates inside one policy group +// are checked together so one unresponsive candidate cannot hide faster results. +func (c *OutboundProbeController) Probe( + configContent, groupsJSON string, + maxConcurrency, samples int32, + handler OutboundProbeHandler, +) error { + groups, err := decodeOutboundProbeGroups(groupsJSON) + if err != nil { + return err + } + if maxConcurrency <= 0 { + return errors.New("outbound probe concurrency must be positive") + } + if samples <= 0 { + return errors.New("outbound probe sample count must be positive") + } + + c.access.Lock() + if c.used { + c.access.Unlock() + return errors.New("outbound probe controller is single-use") + } + ctx, cancel := context.WithCancel(context.Background()) + c.used = true + c.cancel = cancel + c.access.Unlock() + defer func() { + cancel() + c.access.Lock() + c.cancel = nil + c.access.Unlock() + }() + + config, err := coreserial.LoadJSONConfig(strings.NewReader(configContent)) + if err != nil { + return fmt.Errorf("outbound probe config load failed: %w", err) + } + config.Inbound = nil + + inst, err := core.New(config) + if err != nil { + return fmt.Errorf("outbound probe instance creation failed: %w", err) + } + defer inst.Close() + + feature := inst.GetFeature(coreextension.ObservatoryType()) + burst, ok := feature.(coreextension.BurstObservatory) + if !ok { + return errors.New("outbound probe config does not contain a burst observatory") + } + observer, ok := feature.(coreextension.Observatory) + if !ok { + return errors.New("outbound probe observatory does not expose results") + } + if err := inst.Start(); err != nil { + return fmt.Errorf("outbound probe startup failed: %w", err) + } + + return runOutboundProbeGroups( + ctx, + inst, + burst, + observer, + groups, + int(maxConcurrency), + int(samples), + handler, + ) +} + +func decodeOutboundProbeGroups(encoded string) ([]outboundProbeGroup, error) { + var groups []outboundProbeGroup + if err := json.Unmarshal([]byte(encoded), &groups); err != nil { + return nil, fmt.Errorf("outbound probe groups are invalid: %w", err) + } + if len(groups) == 0 { + return nil, errors.New("outbound probe groups are empty") + } + + seenGroups := make(map[string]struct{}) + seenTags := make(map[string]struct{}) + for groupIndex, group := range groups { + if strings.TrimSpace(group.GUID) == "" { + return nil, fmt.Errorf("outbound probe group %d has no ID", groupIndex) + } + if _, exists := seenGroups[group.GUID]; exists { + return nil, fmt.Errorf("outbound probe group ID %q is duplicated", group.GUID) + } + seenGroups[group.GUID] = struct{}{} + if len(group.OutboundTags) == 0 { + return nil, fmt.Errorf("outbound probe group %d is empty", groupIndex) + } + for _, tag := range group.OutboundTags { + if strings.TrimSpace(tag) == "" { + return nil, fmt.Errorf("outbound probe group %d contains an empty tag", groupIndex) + } + if _, exists := seenTags[tag]; exists { + return nil, fmt.Errorf("outbound probe tag %q is duplicated", tag) + } + seenTags[tag] = struct{}{} + } + } + return groups, nil +} + +type indexedOutboundProbeGroup struct { + index int + group outboundProbeGroup +} + +type outboundProbeCompletion struct { + groupIndex int + outboundTag string + acknowledge chan struct{} +} + +func runOutboundProbeGroups( + ctx context.Context, + inst *core.Instance, + burst coreextension.BurstObservatory, + observer coreextension.Observatory, + groups []outboundProbeGroup, + maxConcurrency, samples int, + handler OutboundProbeHandler, +) error { + jobs := make(chan indexedOutboundProbeGroup, len(groups)) + for index, group := range groups { + jobs <- indexedOutboundProbeGroup{index: index, group: group} + } + close(jobs) + + completed := make(chan outboundProbeCompletion) + workerCount := maxConcurrency + if workerCount > len(groups) { + workerCount = len(groups) + } + var workers sync.WaitGroup + workers.Add(workerCount) + for range workerCount { + go func() { + defer workers.Done() + for job := range jobs { + for range samples { + if ctx.Err() != nil { + return + } + var members sync.WaitGroup + members.Add(len(job.group.OutboundTags)) + for _, outboundTag := range job.group.OutboundTags { + tag := outboundTag + go func() { + defer members.Done() + if ctx.Err() != nil { + return + } + burst.Check([]string{tag}) + completion := outboundProbeCompletion{ + groupIndex: job.index, + outboundTag: tag, + acknowledge: make(chan struct{}), + } + select { + case completed <- completion: + case <-ctx.Done(): + return + } + select { + case <-completion.acknowledge: + case <-ctx.Done(): + } + }() + } + members.Wait() + } + } + }() + } + go func() { + workers.Wait() + close(completed) + }() + + counts := make([]map[string]int, len(groups)) + for index, group := range groups { + counts[index] = make(map[string]int, len(group.OutboundTags)) + } + for completion := range completed { + counts[completion.groupIndex][completion.outboundTag]++ + group := groups[completion.groupIndex] + _, delay, alive, err := currentOutboundProbeResult(inst, observer, group) + if err != nil { + log.Printf("outbound probe result unavailable for group %d: %v", completion.groupIndex, err) + } + groupCompleted := true + for _, tag := range group.OutboundTags { + if counts[completion.groupIndex][tag] < samples { + groupCompleted = false + break + } + } + if handler != nil { + handler.OnOutboundProbeResult( + group.GUID, + delay, + alive, + groupCompleted, + ) + } + close(completion.acknowledge) + } + return ctx.Err() +} + +func currentOutboundProbeResult( + inst *core.Instance, + observer coreextension.Observatory, + group outboundProbeGroup, +) (string, int64, bool, error) { + target := group.OutboundTags[0] + if group.BalancerTag != "" { + principle, ok := inst.GetFeature(corerouting.RouterType()).(corerouting.BalancerPrincipleTarget) + if !ok { + return "", -1, false, errors.New("router does not expose balancer principle targets") + } + targets, err := principle.GetPrincipleTarget(group.BalancerTag) + if err != nil { + return "", -1, false, err + } + target = "" + for _, candidate := range targets { + if candidate != "" { + target = candidate + break + } + } + } + if target == "" { + return "", -1, false, nil + } + + message, err := observer.GetObservation(context.Background()) + if err != nil { + return target, -1, false, err + } + result, ok := message.(*coreobservatory.ObservationResult) + if !ok { + return target, -1, false, errors.New("unexpected outbound probe result type") + } + for _, status := range result.GetStatus() { + if status.GetOutboundTag() == target { + if !status.GetAlive() { + return target, -1, false, nil + } + return target, status.GetDelay(), true, nil + } + } + return target, -1, false, nil +} + // CheckVersionX returns the library and Xray versions func CheckVersionX() string { return fmt.Sprintf("Lib v%d, Xray-core v%s", libVersion, core.Version()) From 2b3446564d99884514777b7c911e5443876bd4a7 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 2 Aug 2026 00:14:48 +0300 Subject: [PATCH 2/8] Limit Observatory checks with a concise probe API --- libv2ray_main.go | 160 ++++++++++++++++++++++++----------------------- 1 file changed, 81 insertions(+), 79 deletions(-) diff --git a/libv2ray_main.go b/libv2ray_main.go index 4cfb1912..70af3a4b 100644 --- a/libv2ray_main.go +++ b/libv2ray_main.go @@ -39,35 +39,35 @@ const ( xudpBaseKey = "xray.xudp.basekey" tunFdKey = "xray.tun.fd" browserDialerAddress = "xray.browser.dialer" - libVersion = 40 // Library version, update here only + libVersion = 41 // Library version, update here only ) -// OutboundProbeHandler receives one compact update for the affected UI group. +// ProbeHandler receives one compact update for the affected UI group. // Calls are serialized even though the underlying checks run concurrently. -type OutboundProbeHandler interface { - OnOutboundProbeResult(groupID string, delay int64, alive, completed bool) int +type ProbeHandler interface { + OnProbeResult(groupID string, delay int64, alive, completed bool) int } -type outboundProbeGroup struct { +type probeGroup struct { GUID string `json:"guid"` OutboundTags []string `json:"outboundTags"` BalancerTag string `json:"balancerTag"` } -// OutboundProbeController owns one finite probe batch. v2rayNG runs it in a +// ProbeController owns one finite probe batch. v2rayNG runs it in a // disposable process so Xray's process-wide native state cannot overlap the // long-running VPN core or a later test batch. -type OutboundProbeController struct { +type ProbeController struct { access sync.Mutex cancel context.CancelFunc used bool } -func NewOutboundProbeController() *OutboundProbeController { - return &OutboundProbeController{} +func NewProbeController() *ProbeController { + return &ProbeController{} } -func (c *OutboundProbeController) Cancel() { +func (c *ProbeController) Cancel() { c.access.Lock() cancel := c.cancel c.access.Unlock() @@ -269,28 +269,27 @@ func MeasureOutboundDelay(ConfigureFileContent string, url string) (int64, error } // Probe runs all UI delay-test groups through one short-lived Xray instance. -// maxConcurrency limits active UI profiles; candidates inside one policy group -// are checked together so one unresponsive candidate cannot hide faster results. -func (c *OutboundProbeController) Probe( +// maxConcurrency limits active Observatory checks across every group member. +func (c *ProbeController) Probe( configContent, groupsJSON string, maxConcurrency, samples int32, - handler OutboundProbeHandler, + handler ProbeHandler, ) error { - groups, err := decodeOutboundProbeGroups(groupsJSON) + groups, err := decodeProbeGroups(groupsJSON) if err != nil { return err } if maxConcurrency <= 0 { - return errors.New("outbound probe concurrency must be positive") + return errors.New("probe concurrency must be positive") } if samples <= 0 { - return errors.New("outbound probe sample count must be positive") + return errors.New("probe sample count must be positive") } c.access.Lock() if c.used { c.access.Unlock() - return errors.New("outbound probe controller is single-use") + return errors.New("probe controller is single-use") } ctx, cancel := context.WithCancel(context.Background()) c.used = true @@ -305,30 +304,30 @@ func (c *OutboundProbeController) Probe( config, err := coreserial.LoadJSONConfig(strings.NewReader(configContent)) if err != nil { - return fmt.Errorf("outbound probe config load failed: %w", err) + return fmt.Errorf("probe config load failed: %w", err) } config.Inbound = nil inst, err := core.New(config) if err != nil { - return fmt.Errorf("outbound probe instance creation failed: %w", err) + return fmt.Errorf("probe instance creation failed: %w", err) } defer inst.Close() feature := inst.GetFeature(coreextension.ObservatoryType()) burst, ok := feature.(coreextension.BurstObservatory) if !ok { - return errors.New("outbound probe config does not contain a burst observatory") + return errors.New("probe config does not contain a burst observatory") } observer, ok := feature.(coreextension.Observatory) if !ok { - return errors.New("outbound probe observatory does not expose results") + return errors.New("probe observatory does not expose results") } if err := inst.Start(); err != nil { - return fmt.Errorf("outbound probe startup failed: %w", err) + return fmt.Errorf("probe startup failed: %w", err) } - return runOutboundProbeGroups( + return runProbeGroups( ctx, inst, burst, @@ -340,34 +339,34 @@ func (c *OutboundProbeController) Probe( ) } -func decodeOutboundProbeGroups(encoded string) ([]outboundProbeGroup, error) { - var groups []outboundProbeGroup +func decodeProbeGroups(encoded string) ([]probeGroup, error) { + var groups []probeGroup if err := json.Unmarshal([]byte(encoded), &groups); err != nil { - return nil, fmt.Errorf("outbound probe groups are invalid: %w", err) + return nil, fmt.Errorf("probe groups are invalid: %w", err) } if len(groups) == 0 { - return nil, errors.New("outbound probe groups are empty") + return nil, errors.New("probe groups are empty") } seenGroups := make(map[string]struct{}) seenTags := make(map[string]struct{}) for groupIndex, group := range groups { if strings.TrimSpace(group.GUID) == "" { - return nil, fmt.Errorf("outbound probe group %d has no ID", groupIndex) + return nil, fmt.Errorf("probe group %d has no ID", groupIndex) } if _, exists := seenGroups[group.GUID]; exists { - return nil, fmt.Errorf("outbound probe group ID %q is duplicated", group.GUID) + return nil, fmt.Errorf("probe group ID %q is duplicated", group.GUID) } seenGroups[group.GUID] = struct{}{} if len(group.OutboundTags) == 0 { - return nil, fmt.Errorf("outbound probe group %d is empty", groupIndex) + return nil, fmt.Errorf("probe group %d is empty", groupIndex) } for _, tag := range group.OutboundTags { if strings.TrimSpace(tag) == "" { - return nil, fmt.Errorf("outbound probe group %d contains an empty tag", groupIndex) + return nil, fmt.Errorf("probe group %d contains an empty tag", groupIndex) } if _, exists := seenTags[tag]; exists { - return nil, fmt.Errorf("outbound probe tag %q is duplicated", tag) + return nil, fmt.Errorf("probe tag %q is duplicated", tag) } seenTags[tag] = struct{}{} } @@ -375,74 +374,77 @@ func decodeOutboundProbeGroups(encoded string) ([]outboundProbeGroup, error) { return groups, nil } -type indexedOutboundProbeGroup struct { - index int - group outboundProbeGroup +type probeTarget struct { + groupIndex int + outboundTag string } -type outboundProbeCompletion struct { +type probeCompletion struct { groupIndex int outboundTag string acknowledge chan struct{} } -func runOutboundProbeGroups( +func runProbeGroups( ctx context.Context, inst *core.Instance, burst coreextension.BurstObservatory, observer coreextension.Observatory, - groups []outboundProbeGroup, + groups []probeGroup, maxConcurrency, samples int, - handler OutboundProbeHandler, + handler ProbeHandler, ) error { - jobs := make(chan indexedOutboundProbeGroup, len(groups)) - for index, group := range groups { - jobs <- indexedOutboundProbeGroup{index: index, group: group} + targetCount := 0 + maxGroupSize := 0 + for _, group := range groups { + targetCount += len(group.OutboundTags) + if len(group.OutboundTags) > maxGroupSize { + maxGroupSize = len(group.OutboundTags) + } + } + jobs := make(chan probeTarget, targetCount) + // Interleave groups so a large policy group cannot put every other profile + // behind all of its candidates when concurrency is limited. + for memberIndex := 0; memberIndex < maxGroupSize; memberIndex++ { + for groupIndex, group := range groups { + if memberIndex < len(group.OutboundTags) { + jobs <- probeTarget{groupIndex, group.OutboundTags[memberIndex]} + } + } } close(jobs) - completed := make(chan outboundProbeCompletion) + completed := make(chan probeCompletion) workerCount := maxConcurrency - if workerCount > len(groups) { - workerCount = len(groups) + if workerCount > targetCount { + workerCount = targetCount } var workers sync.WaitGroup workers.Add(workerCount) for range workerCount { go func() { defer workers.Done() - for job := range jobs { + for target := range jobs { for range samples { if ctx.Err() != nil { return } - var members sync.WaitGroup - members.Add(len(job.group.OutboundTags)) - for _, outboundTag := range job.group.OutboundTags { - tag := outboundTag - go func() { - defer members.Done() - if ctx.Err() != nil { - return - } - burst.Check([]string{tag}) - completion := outboundProbeCompletion{ - groupIndex: job.index, - outboundTag: tag, - acknowledge: make(chan struct{}), - } - select { - case completed <- completion: - case <-ctx.Done(): - return - } - select { - case <-completion.acknowledge: - case <-ctx.Done(): - } - }() + burst.Check([]string{target.outboundTag}) + completion := probeCompletion{ + groupIndex: target.groupIndex, + outboundTag: target.outboundTag, + acknowledge: make(chan struct{}), + } + select { + case completed <- completion: + case <-ctx.Done(): + return + } + select { + case <-completion.acknowledge: + case <-ctx.Done(): + return } - members.Wait() } } }() @@ -459,9 +461,9 @@ func runOutboundProbeGroups( for completion := range completed { counts[completion.groupIndex][completion.outboundTag]++ group := groups[completion.groupIndex] - _, delay, alive, err := currentOutboundProbeResult(inst, observer, group) + _, delay, alive, err := currentProbeResult(inst, observer, group) if err != nil { - log.Printf("outbound probe result unavailable for group %d: %v", completion.groupIndex, err) + log.Printf("probe result unavailable for group %d: %v", completion.groupIndex, err) } groupCompleted := true for _, tag := range group.OutboundTags { @@ -471,7 +473,7 @@ func runOutboundProbeGroups( } } if handler != nil { - handler.OnOutboundProbeResult( + handler.OnProbeResult( group.GUID, delay, alive, @@ -483,10 +485,10 @@ func runOutboundProbeGroups( return ctx.Err() } -func currentOutboundProbeResult( +func currentProbeResult( inst *core.Instance, observer coreextension.Observatory, - group outboundProbeGroup, + group probeGroup, ) (string, int64, bool, error) { target := group.OutboundTags[0] if group.BalancerTag != "" { @@ -516,7 +518,7 @@ func currentOutboundProbeResult( } result, ok := message.(*coreobservatory.ObservationResult) if !ok { - return target, -1, false, errors.New("unexpected outbound probe result type") + return target, -1, false, errors.New("unexpected probe result type") } for _, status := range result.GetStatus() { if status.GetOutboundTag() == target { From 2c1ef09bfab2053acae8286a4db26ed42bb73fda Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 2 Aug 2026 00:34:54 +0300 Subject: [PATCH 3/8] Probe each Observatory target once --- libv2ray_main.go | 66 +++++++++++++++++++----------------------------- 1 file changed, 26 insertions(+), 40 deletions(-) diff --git a/libv2ray_main.go b/libv2ray_main.go index 70af3a4b..301c84a7 100644 --- a/libv2ray_main.go +++ b/libv2ray_main.go @@ -39,7 +39,7 @@ const ( xudpBaseKey = "xray.xudp.basekey" tunFdKey = "xray.tun.fd" browserDialerAddress = "xray.browser.dialer" - libVersion = 41 // Library version, update here only + libVersion = 42 // Library version, update here only ) // ProbeHandler receives one compact update for the affected UI group. @@ -269,10 +269,11 @@ func MeasureOutboundDelay(ConfigureFileContent string, url string) (int64, error } // Probe runs all UI delay-test groups through one short-lived Xray instance. -// maxConcurrency limits active Observatory checks across every group member. +// Every target is checked once. maxConcurrency limits active Observatory +// checks across every group member. func (c *ProbeController) Probe( configContent, groupsJSON string, - maxConcurrency, samples int32, + maxConcurrency int32, handler ProbeHandler, ) error { groups, err := decodeProbeGroups(groupsJSON) @@ -282,9 +283,6 @@ func (c *ProbeController) Probe( if maxConcurrency <= 0 { return errors.New("probe concurrency must be positive") } - if samples <= 0 { - return errors.New("probe sample count must be positive") - } c.access.Lock() if c.used { @@ -334,7 +332,6 @@ func (c *ProbeController) Probe( observer, groups, int(maxConcurrency), - int(samples), handler, ) } @@ -381,7 +378,6 @@ type probeTarget struct { type probeCompletion struct { groupIndex int - outboundTag string acknowledge chan struct{} } @@ -391,7 +387,7 @@ func runProbeGroups( burst coreextension.BurstObservatory, observer coreextension.Observatory, groups []probeGroup, - maxConcurrency, samples int, + maxConcurrency int, handler ProbeHandler, ) error { targetCount := 0 @@ -425,26 +421,23 @@ func runProbeGroups( go func() { defer workers.Done() for target := range jobs { - for range samples { - if ctx.Err() != nil { - return - } - burst.Check([]string{target.outboundTag}) - completion := probeCompletion{ - groupIndex: target.groupIndex, - outboundTag: target.outboundTag, - acknowledge: make(chan struct{}), - } - select { - case completed <- completion: - case <-ctx.Done(): - return - } - select { - case <-completion.acknowledge: - case <-ctx.Done(): - return - } + if ctx.Err() != nil { + return + } + burst.Check([]string{target.outboundTag}) + completion := probeCompletion{ + groupIndex: target.groupIndex, + acknowledge: make(chan struct{}), + } + select { + case completed <- completion: + case <-ctx.Done(): + return + } + select { + case <-completion.acknowledge: + case <-ctx.Done(): + return } } }() @@ -454,30 +447,23 @@ func runProbeGroups( close(completed) }() - counts := make([]map[string]int, len(groups)) + remaining := make([]int, len(groups)) for index, group := range groups { - counts[index] = make(map[string]int, len(group.OutboundTags)) + remaining[index] = len(group.OutboundTags) } for completion := range completed { - counts[completion.groupIndex][completion.outboundTag]++ + remaining[completion.groupIndex]-- group := groups[completion.groupIndex] _, delay, alive, err := currentProbeResult(inst, observer, group) if err != nil { log.Printf("probe result unavailable for group %d: %v", completion.groupIndex, err) } - groupCompleted := true - for _, tag := range group.OutboundTags { - if counts[completion.groupIndex][tag] < samples { - groupCompleted = false - break - } - } if handler != nil { handler.OnProbeResult( group.GUID, delay, alive, - groupCompleted, + remaining[completion.groupIndex] == 0, ) } close(completion.acknowledge) From 82213e3e61b492ae17bf26b42b862a63659d6393 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 2 Aug 2026 15:17:10 +0300 Subject: [PATCH 4/8] Simplify Observatory probe coordination Trust the typed v2rayNG probe plan and the fixed upstream BurstObservatory implementation instead of defending against duplicate plans, controller reuse, nil handlers, and impossible interface/result types. Remove the per-result acknowledgement barrier because the single completion consumer already serializes callbacks, while retaining bounded workers and real cancellation support. --- libv2ray_main.go | 168 ++++++++++------------------------------------- 1 file changed, 33 insertions(+), 135 deletions(-) diff --git a/libv2ray_main.go b/libv2ray_main.go index 301c84a7..0760d078 100644 --- a/libv2ray_main.go +++ b/libv2ray_main.go @@ -39,10 +39,10 @@ const ( xudpBaseKey = "xray.xudp.basekey" tunFdKey = "xray.tun.fd" browserDialerAddress = "xray.browser.dialer" - libVersion = 42 // Library version, update here only + libVersion = 43 // Library version, update here only ) -// ProbeHandler receives one compact update for the affected UI group. +// ProbeHandler receives one compact update for the affected profile group. // Calls are serialized even though the underlying checks run concurrently. type ProbeHandler interface { OnProbeResult(groupID string, delay int64, alive, completed bool) int @@ -54,13 +54,10 @@ type probeGroup struct { BalancerTag string `json:"balancerTag"` } -// ProbeController owns one finite probe batch. v2rayNG runs it in a -// disposable process so Xray's process-wide native state cannot overlap the -// long-running VPN core or a later test batch. +// ProbeController owns one cancellable probe batch. type ProbeController struct { access sync.Mutex cancel context.CancelFunc - used bool } func NewProbeController() *ProbeController { @@ -268,7 +265,7 @@ func MeasureOutboundDelay(ConfigureFileContent string, url string) (int64, error return measureInstDelay(context.Background(), inst, url) } -// Probe runs all UI delay-test groups through one short-lived Xray instance. +// Probe runs all delay-test groups through one short-lived Xray instance. // Every target is checked once. maxConcurrency limits active Observatory // checks across every group member. func (c *ProbeController) Probe( @@ -276,29 +273,16 @@ func (c *ProbeController) Probe( maxConcurrency int32, handler ProbeHandler, ) error { - groups, err := decodeProbeGroups(groupsJSON) - if err != nil { - return err - } - if maxConcurrency <= 0 { - return errors.New("probe concurrency must be positive") + var groups []probeGroup + if err := json.Unmarshal([]byte(groupsJSON), &groups); err != nil { + return fmt.Errorf("probe groups are invalid: %w", err) } c.access.Lock() - if c.used { - c.access.Unlock() - return errors.New("probe controller is single-use") - } ctx, cancel := context.WithCancel(context.Background()) - c.used = true c.cancel = cancel c.access.Unlock() - defer func() { - cancel() - c.access.Lock() - c.cancel = nil - c.access.Unlock() - }() + defer cancel() config, err := coreserial.LoadJSONConfig(strings.NewReader(configContent)) if err != nil { @@ -312,15 +296,7 @@ func (c *ProbeController) Probe( } defer inst.Close() - feature := inst.GetFeature(coreextension.ObservatoryType()) - burst, ok := feature.(coreextension.BurstObservatory) - if !ok { - return errors.New("probe config does not contain a burst observatory") - } - observer, ok := feature.(coreextension.Observatory) - if !ok { - return errors.New("probe observatory does not expose results") - } + burst := inst.GetFeature(coreextension.ObservatoryType()).(coreextension.BurstObservatory) if err := inst.Start(); err != nil { return fmt.Errorf("probe startup failed: %w", err) } @@ -329,63 +305,21 @@ func (c *ProbeController) Probe( ctx, inst, burst, - observer, groups, int(maxConcurrency), handler, ) } -func decodeProbeGroups(encoded string) ([]probeGroup, error) { - var groups []probeGroup - if err := json.Unmarshal([]byte(encoded), &groups); err != nil { - return nil, fmt.Errorf("probe groups are invalid: %w", err) - } - if len(groups) == 0 { - return nil, errors.New("probe groups are empty") - } - - seenGroups := make(map[string]struct{}) - seenTags := make(map[string]struct{}) - for groupIndex, group := range groups { - if strings.TrimSpace(group.GUID) == "" { - return nil, fmt.Errorf("probe group %d has no ID", groupIndex) - } - if _, exists := seenGroups[group.GUID]; exists { - return nil, fmt.Errorf("probe group ID %q is duplicated", group.GUID) - } - seenGroups[group.GUID] = struct{}{} - if len(group.OutboundTags) == 0 { - return nil, fmt.Errorf("probe group %d is empty", groupIndex) - } - for _, tag := range group.OutboundTags { - if strings.TrimSpace(tag) == "" { - return nil, fmt.Errorf("probe group %d contains an empty tag", groupIndex) - } - if _, exists := seenTags[tag]; exists { - return nil, fmt.Errorf("probe tag %q is duplicated", tag) - } - seenTags[tag] = struct{}{} - } - } - return groups, nil -} - type probeTarget struct { groupIndex int outboundTag string } -type probeCompletion struct { - groupIndex int - acknowledge chan struct{} -} - func runProbeGroups( ctx context.Context, inst *core.Instance, burst coreextension.BurstObservatory, - observer coreextension.Observatory, groups []probeGroup, maxConcurrency int, handler ProbeHandler, @@ -410,7 +344,7 @@ func runProbeGroups( } close(jobs) - completed := make(chan probeCompletion) + completed := make(chan int) workerCount := maxConcurrency if workerCount > targetCount { workerCount = targetCount @@ -425,17 +359,8 @@ func runProbeGroups( return } burst.Check([]string{target.outboundTag}) - completion := probeCompletion{ - groupIndex: target.groupIndex, - acknowledge: make(chan struct{}), - } select { - case completed <- completion: - case <-ctx.Done(): - return - } - select { - case <-completion.acknowledge: + case completed <- target.groupIndex: case <-ctx.Done(): return } @@ -451,70 +376,43 @@ func runProbeGroups( for index, group := range groups { remaining[index] = len(group.OutboundTags) } - for completion := range completed { - remaining[completion.groupIndex]-- - group := groups[completion.groupIndex] - _, delay, alive, err := currentProbeResult(inst, observer, group) - if err != nil { - log.Printf("probe result unavailable for group %d: %v", completion.groupIndex, err) - } - if handler != nil { - handler.OnProbeResult( - group.GUID, - delay, - alive, - remaining[completion.groupIndex] == 0, - ) - } - close(completion.acknowledge) + for groupIndex := range completed { + remaining[groupIndex]-- + group := groups[groupIndex] + delay, alive := currentProbeResult(inst, burst, group) + handler.OnProbeResult( + group.GUID, + delay, + alive, + remaining[groupIndex] == 0, + ) } return ctx.Err() } func currentProbeResult( inst *core.Instance, - observer coreextension.Observatory, + observer coreextension.BurstObservatory, group probeGroup, -) (string, int64, bool, error) { +) (int64, bool) { target := group.OutboundTags[0] if group.BalancerTag != "" { - principle, ok := inst.GetFeature(corerouting.RouterType()).(corerouting.BalancerPrincipleTarget) - if !ok { - return "", -1, false, errors.New("router does not expose balancer principle targets") - } - targets, err := principle.GetPrincipleTarget(group.BalancerTag) - if err != nil { - return "", -1, false, err - } - target = "" - for _, candidate := range targets { - if candidate != "" { - target = candidate - break - } + principle := inst.GetFeature(corerouting.RouterType()).(corerouting.BalancerPrincipleTarget) + targets, _ := principle.GetPrincipleTarget(group.BalancerTag) + if len(targets) == 0 { + return -1, false } - } - if target == "" { - return "", -1, false, nil + target = targets[0] } - message, err := observer.GetObservation(context.Background()) - if err != nil { - return target, -1, false, err - } - result, ok := message.(*coreobservatory.ObservationResult) - if !ok { - return target, -1, false, errors.New("unexpected probe result type") - } + message, _ := observer.GetObservation(context.Background()) + result := message.(*coreobservatory.ObservationResult) for _, status := range result.GetStatus() { - if status.GetOutboundTag() == target { - if !status.GetAlive() { - return target, -1, false, nil - } - return target, status.GetDelay(), true, nil + if status.GetOutboundTag() == target && status.GetAlive() { + return status.GetDelay(), true } } - return target, -1, false, nil + return -1, false } // CheckVersionX returns the library and Xray versions From 484a8771eea0128022f25c30a0bc39db3c056888 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 9 Aug 2026 16:44:22 +0300 Subject: [PATCH 5/8] Optimize large observatory probe batches --- libv2ray_main.go | 181 ++++++++++++++++++++++++++++++++++------------- 1 file changed, 133 insertions(+), 48 deletions(-) diff --git a/libv2ray_main.go b/libv2ray_main.go index 0760d078..e50c760c 100644 --- a/libv2ray_main.go +++ b/libv2ray_main.go @@ -34,12 +34,14 @@ import ( // Constants for environment variables const ( - coreAsset = "xray.location.asset" - coreCert = "xray.location.cert" - xudpBaseKey = "xray.xudp.basekey" - tunFdKey = "xray.tun.fd" - browserDialerAddress = "xray.browser.dialer" - libVersion = 43 // Library version, update here only + coreAsset = "xray.location.asset" + coreCert = "xray.location.cert" + xudpBaseKey = "xray.xudp.basekey" + tunFdKey = "xray.tun.fd" + browserDialerAddress = "xray.browser.dialer" + libVersion = 43 // Library version, update here only + defaultRealDelayTimeout = 5 * time.Second + probeResultAggregationWindow = 50 * time.Millisecond ) // ProbeHandler receives one compact update for the affected profile group. @@ -262,7 +264,9 @@ func MeasureOutboundDelay(ConfigureFileContent string, url string) (int64, error return -1, fmt.Errorf("startup failed: %w", err) } defer inst.Close() - return measureInstDelay(context.Background(), inst, url) + ctx, cancel := context.WithTimeout(context.Background(), defaultRealDelayTimeout) + defer cancel() + return measureInstDelayWithOptions(ctx, inst, url, http.MethodHead, 1, defaultRealDelayTimeout) } // Probe runs all delay-test groups through one short-lived Xray instance. @@ -332,23 +336,18 @@ func runProbeGroups( maxGroupSize = len(group.OutboundTags) } } - jobs := make(chan probeTarget, targetCount) - // Interleave groups so a large policy group cannot put every other profile - // behind all of its candidates when concurrency is limited. - for memberIndex := 0; memberIndex < maxGroupSize; memberIndex++ { - for groupIndex, group := range groups { - if memberIndex < len(group.OutboundTags) { - jobs <- probeTarget{groupIndex, group.OutboundTags[memberIndex]} - } - } + if targetCount == 0 { + return nil } - close(jobs) - - completed := make(chan int) workerCount := maxConcurrency + if workerCount < 1 { + workerCount = 1 + } if workerCount > targetCount { workerCount = targetCount } + jobs := make(chan probeTarget, workerCount) + completed := make(chan probeTarget, workerCount) var workers sync.WaitGroup workers.Add(workerCount) for range workerCount { @@ -360,13 +359,30 @@ func runProbeGroups( } burst.Check([]string{target.outboundTag}) select { - case completed <- target.groupIndex: + case completed <- target: case <-ctx.Done(): return } } }() } + go func() { + defer close(jobs) + // Interleave groups so a large policy group cannot put every other + // profile behind all of its candidates when concurrency is limited. + for memberIndex := 0; memberIndex < maxGroupSize; memberIndex++ { + for groupIndex, group := range groups { + if memberIndex >= len(group.OutboundTags) { + continue + } + select { + case jobs <- probeTarget{groupIndex, group.OutboundTags[memberIndex]}: + case <-ctx.Done(): + return + } + } + } + }() go func() { workers.Wait() close(completed) @@ -376,43 +392,102 @@ func runProbeGroups( for index, group := range groups { remaining[index] = len(group.OutboundTags) } - for groupIndex := range completed { - remaining[groupIndex]-- - group := groups[groupIndex] - delay, alive := currentProbeResult(inst, burst, group) - handler.OnProbeResult( - group.GUID, - delay, - alive, - remaining[groupIndex] == 0, - ) + for { + first, ok := <-completed + if !ok { + break + } + batch, closed := collectProbeCompletions(first, completed, workerCount) + statuses := currentProbeStatuses(burst) + results := make(map[int]probeResult, len(batch)) + for _, target := range batch { + if _, found := results[target.groupIndex]; !found { + results[target.groupIndex] = currentProbeResult(inst, groups[target.groupIndex], statuses) + } + } + for _, target := range batch { + remaining[target.groupIndex]-- + group := groups[target.groupIndex] + result := results[target.groupIndex] + handler.OnProbeResult( + group.GUID, + result.delay, + result.alive, + remaining[target.groupIndex] == 0, + ) + } + if closed { + break + } } return ctx.Err() } +func collectProbeCompletions( + first probeTarget, + completed <-chan probeTarget, + limit int, +) ([]probeTarget, bool) { + batch := []probeTarget{first} + timer := time.NewTimer(probeResultAggregationWindow) + defer timer.Stop() + for len(batch) < limit { + select { + case target, ok := <-completed: + if !ok { + return batch, true + } + batch = append(batch, target) + case <-timer.C: + return batch, false + } + } + return batch, false +} + +type probeResult struct { + delay int64 + alive bool +} + +func currentProbeStatuses(observer coreextension.BurstObservatory) map[string]*coreobservatory.OutboundStatus { + message, err := observer.GetObservation(context.Background()) + if err != nil { + return nil + } + result, ok := message.(*coreobservatory.ObservationResult) + if !ok { + return nil + } + statuses := make(map[string]*coreobservatory.OutboundStatus, len(result.GetStatus())) + for _, status := range result.GetStatus() { + statuses[status.GetOutboundTag()] = status + } + return statuses +} + func currentProbeResult( inst *core.Instance, - observer coreextension.BurstObservatory, group probeGroup, -) (int64, bool) { + statuses map[string]*coreobservatory.OutboundStatus, +) probeResult { + if len(group.OutboundTags) == 0 { + return probeResult{delay: -1} + } target := group.OutboundTags[0] if group.BalancerTag != "" { principle := inst.GetFeature(corerouting.RouterType()).(corerouting.BalancerPrincipleTarget) targets, _ := principle.GetPrincipleTarget(group.BalancerTag) if len(targets) == 0 { - return -1, false + return probeResult{delay: -1} } target = targets[0] } - - message, _ := observer.GetObservation(context.Background()) - result := message.(*coreobservatory.ObservationResult) - for _, status := range result.GetStatus() { - if status.GetOutboundTag() == target && status.GetAlive() { - return status.GetDelay(), true - } + status := statuses[target] + if status != nil && status.GetAlive() { + return probeResult{delay: status.GetDelay(), alive: true} } - return -1, false + return probeResult{delay: -1} } // CheckVersionX returns the library and Xray versions @@ -469,6 +544,16 @@ func (x *CoreController) doStartLoop(configContent string) error { // measureInstDelay measures the delay for an instance to a given URL func measureInstDelay(ctx context.Context, inst *core.Instance, url string) (int64, error) { + return measureInstDelayWithOptions(ctx, inst, url, http.MethodGet, 2, 12*time.Second) +} + +func measureInstDelayWithOptions( + ctx context.Context, + inst *core.Instance, + url, method string, + attempts int, + timeout time.Duration, +) (int64, error) { if inst == nil { return -1, errors.New("core instance is nil") } @@ -478,7 +563,7 @@ func measureInstDelay(ctx context.Context, inst *core.Instance, url string) (int } tr := &http.Transport{ - TLSHandshakeTimeout: 6 * time.Second, + TLSHandshakeTimeout: timeout, DisableKeepAlives: false, DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { dest, err := corenet.ParseDestination(fmt.Sprintf("%s:%s", network, addr)) @@ -491,7 +576,7 @@ func measureInstDelay(ctx context.Context, inst *core.Instance, url string) (int client := &http.Client{ Transport: tr, - Timeout: 12 * time.Second, + Timeout: timeout, } var minDuration int64 = -1 @@ -501,8 +586,6 @@ func measureInstDelay(ctx context.Context, inst *core.Instance, url string) (int // Close idle connections to ensure the temporary instance can be closed safely defer tr.CloseIdleConnections() - // Add exception handling and increase retry attempts - const attempts = 2 for i := 0; i < attempts; i++ { select { case <-ctx.Done(): @@ -515,7 +598,7 @@ func measureInstDelay(ctx context.Context, inst *core.Instance, url string) (int // Continue execution } - req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + req, err := http.NewRequestWithContext(ctx, method, url, nil) if err != nil { lastErr = fmt.Errorf("failed to create HTTP request: %w", err) continue @@ -528,8 +611,10 @@ func measureInstDelay(ctx context.Context, inst *core.Instance, url string) (int continue } - // Read and close body immediately to allow connection reuse for the next attempt - _, err = io.Copy(io.Discard, resp.Body) + // Read GET bodies so a subsequent attempt may reuse the connection. + if method == http.MethodGet { + _, err = io.Copy(io.Discard, resp.Body) + } resp.Body.Close() if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent { From e1f3232b3f5369177cd4472a51b3a91426396e9f Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 9 Aug 2026 19:52:15 +0300 Subject: [PATCH 6/8] Simplify observatory probe API --- libv2ray_main.go | 62 +++++++++++++++++++++++++++++------------------- 1 file changed, 37 insertions(+), 25 deletions(-) diff --git a/libv2ray_main.go b/libv2ray_main.go index e50c760c..59fb2af4 100644 --- a/libv2ray_main.go +++ b/libv2ray_main.go @@ -47,7 +47,7 @@ const ( // ProbeHandler receives one compact update for the affected profile group. // Calls are serialized even though the underlying checks run concurrently. type ProbeHandler interface { - OnProbeResult(groupID string, delay int64, alive, completed bool) int + OnProbeResult(groupID string, delay int64, completed bool) } type probeGroup struct { @@ -56,10 +56,11 @@ type probeGroup struct { BalancerTag string `json:"balancerTag"` } -// ProbeController owns one cancellable probe batch. +// ProbeController owns one cancellable probe sequence. type ProbeController struct { - access sync.Mutex - cancel context.CancelFunc + access sync.Mutex + cancel context.CancelFunc + cancelled bool } func NewProbeController() *ProbeController { @@ -68,6 +69,7 @@ func NewProbeController() *ProbeController { func (c *ProbeController) Cancel() { c.access.Lock() + c.cancelled = true cancel := c.cancel c.access.Unlock() if cancel != nil { @@ -276,13 +278,24 @@ func (c *ProbeController) Probe( configContent, groupsJSON string, maxConcurrency int32, handler ProbeHandler, -) error { +) (err error) { + // Keep malformed core state on the ordinary error path so the app can isolate + // the responsible profile instead of losing every result in the batch. + defer func() { + if value := recover(); value != nil { + err = fmt.Errorf("probe panicked: %v", value) + } + }() var groups []probeGroup if err := json.Unmarshal([]byte(groupsJSON), &groups); err != nil { return fmt.Errorf("probe groups are invalid: %w", err) } c.access.Lock() + if c.cancelled { + c.access.Unlock() + return context.Canceled + } ctx, cancel := context.WithCancel(context.Background()) c.cancel = cancel c.access.Unlock() @@ -300,7 +313,10 @@ func (c *ProbeController) Probe( } defer inst.Close() - burst := inst.GetFeature(coreextension.ObservatoryType()).(coreextension.BurstObservatory) + burst, ok := inst.GetFeature(coreextension.ObservatoryType()).(coreextension.BurstObservatory) + if !ok || burst == nil { + return errors.New("probe burst observatory is unavailable") + } if err := inst.Start(); err != nil { return fmt.Errorf("probe startup failed: %w", err) } @@ -399,20 +415,18 @@ func runProbeGroups( } batch, closed := collectProbeCompletions(first, completed, workerCount) statuses := currentProbeStatuses(burst) - results := make(map[int]probeResult, len(batch)) + results := make(map[int]int64, len(batch)) for _, target := range batch { if _, found := results[target.groupIndex]; !found { - results[target.groupIndex] = currentProbeResult(inst, groups[target.groupIndex], statuses) + results[target.groupIndex] = currentProbeDelay(inst, groups[target.groupIndex], statuses) } } for _, target := range batch { remaining[target.groupIndex]-- group := groups[target.groupIndex] - result := results[target.groupIndex] handler.OnProbeResult( group.GUID, - result.delay, - result.alive, + results[target.groupIndex], remaining[target.groupIndex] == 0, ) } @@ -445,11 +459,6 @@ func collectProbeCompletions( return batch, false } -type probeResult struct { - delay int64 - alive bool -} - func currentProbeStatuses(observer coreextension.BurstObservatory) map[string]*coreobservatory.OutboundStatus { message, err := observer.GetObservation(context.Background()) if err != nil { @@ -466,28 +475,31 @@ func currentProbeStatuses(observer coreextension.BurstObservatory) map[string]*c return statuses } -func currentProbeResult( +func currentProbeDelay( inst *core.Instance, group probeGroup, statuses map[string]*coreobservatory.OutboundStatus, -) probeResult { +) int64 { if len(group.OutboundTags) == 0 { - return probeResult{delay: -1} + return -1 } target := group.OutboundTags[0] if group.BalancerTag != "" { - principle := inst.GetFeature(corerouting.RouterType()).(corerouting.BalancerPrincipleTarget) - targets, _ := principle.GetPrincipleTarget(group.BalancerTag) - if len(targets) == 0 { - return probeResult{delay: -1} + principle, ok := inst.GetFeature(corerouting.RouterType()).(corerouting.BalancerPrincipleTarget) + if !ok || principle == nil { + return -1 + } + targets, err := principle.GetPrincipleTarget(group.BalancerTag) + if err != nil || len(targets) == 0 || targets[0] == "" { + return -1 } target = targets[0] } status := statuses[target] if status != nil && status.GetAlive() { - return probeResult{delay: status.GetDelay(), alive: true} + return status.GetDelay() } - return probeResult{delay: -1} + return -1 } // CheckVersionX returns the library and Xray versions From 93273c799e258e02905c466ac5d6230a9d158307 Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Sun, 9 Aug 2026 22:06:33 +0300 Subject: [PATCH 7/8] Refine observatory probe lifecycle --- libv2ray_main.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/libv2ray_main.go b/libv2ray_main.go index 59fb2af4..6513bb5e 100644 --- a/libv2ray_main.go +++ b/libv2ray_main.go @@ -39,12 +39,12 @@ const ( xudpBaseKey = "xray.xudp.basekey" tunFdKey = "xray.tun.fd" browserDialerAddress = "xray.browser.dialer" - libVersion = 43 // Library version, update here only + libVersion = 40 // Library version, update here only defaultRealDelayTimeout = 5 * time.Second probeResultAggregationWindow = 50 * time.Millisecond ) -// ProbeHandler receives one compact update for the affected profile group. +// ProbeHandler receives a group update whenever one of its targets finishes. // Calls are serialized even though the underlying checks run concurrently. type ProbeHandler interface { OnProbeResult(groupID string, delay int64, completed bool) @@ -307,7 +307,7 @@ func (c *ProbeController) Probe( } config.Inbound = nil - inst, err := core.New(config) + inst, err := core.NewWithContext(ctx, config) if err != nil { return fmt.Errorf("probe instance creation failed: %w", err) } @@ -575,7 +575,7 @@ func measureInstDelayWithOptions( } tr := &http.Transport{ - TLSHandshakeTimeout: timeout, + TLSHandshakeTimeout: 6 * time.Second, DisableKeepAlives: false, DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { dest, err := corenet.ParseDestination(fmt.Sprintf("%s:%s", network, addr)) From 5045739c85523082d930e310994120d89f89001a Mon Sep 17 00:00:00 2001 From: Eliot the Cougar Date: Fri, 14 Aug 2026 22:42:57 +0300 Subject: [PATCH 8/8] Serialize and cancel short-lived probes --- libv2ray_main.go | 76 +++++++++++++++++++++++++++++++++--------------- 1 file changed, 52 insertions(+), 24 deletions(-) diff --git a/libv2ray_main.go b/libv2ray_main.go index 6513bb5e..5f06359b 100644 --- a/libv2ray_main.go +++ b/libv2ray_main.go @@ -39,7 +39,7 @@ const ( xudpBaseKey = "xray.xudp.basekey" tunFdKey = "xray.tun.fd" browserDialerAddress = "xray.browser.dialer" - libVersion = 40 // Library version, update here only + libVersion = 41 // Library version, update here only defaultRealDelayTimeout = 5 * time.Second probeResultAggregationWindow = 50 * time.Millisecond ) @@ -56,25 +56,37 @@ type probeGroup struct { BalancerTag string `json:"balancerTag"` } +// Xray stores its system DNS client and outbound manager in package globals. +// Keep every short-lived probe core exclusive within this process while still +// allowing one shared core to run its checks concurrently. +var probeCoreGate = make(chan struct{}, 1) + +func acquireProbeCore(ctx context.Context) error { + select { + case probeCoreGate <- struct{}{}: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func releaseProbeCore() { + <-probeCoreGate +} + // ProbeController owns one cancellable probe sequence. type ProbeController struct { - access sync.Mutex - cancel context.CancelFunc - cancelled bool + ctx context.Context + cancel context.CancelFunc } func NewProbeController() *ProbeController { - return &ProbeController{} + ctx, cancel := context.WithCancel(context.Background()) + return &ProbeController{ctx: ctx, cancel: cancel} } func (c *ProbeController) Cancel() { - c.access.Lock() - c.cancelled = true - cancel := c.cancel - c.access.Unlock() - if cancel != nil { - cancel() - } + c.cancel() } // CoreController represents a controller for managing Xray core instance lifecycle @@ -240,7 +252,19 @@ func (x *CoreController) MeasureDelay(url string) (int64, error) { // MeasureOutboundDelay measures the outbound delay for a given configuration and URL func MeasureOutboundDelay(ConfigureFileContent string, url string) (int64, error) { - config, err := coreserial.LoadJSONConfig(strings.NewReader(ConfigureFileContent)) + return measureOutboundDelay(context.Background(), ConfigureFileContent, url) +} + +// MeasureDelay runs one individually configured fallback through this controller. +func (c *ProbeController) MeasureDelay(configContent string, url string) (int64, error) { + return measureOutboundDelay(c.ctx, configContent, url) +} + +func measureOutboundDelay(parentCtx context.Context, configContent string, url string) (int64, error) { + if err := parentCtx.Err(); err != nil { + return -1, err + } + config, err := coreserial.LoadJSONConfig(strings.NewReader(configContent)) if err != nil { return -1, fmt.Errorf("config load error: %w", err) } @@ -257,7 +281,14 @@ func MeasureOutboundDelay(ConfigureFileContent string, url string) (int64, error } config.App = essentialApp - inst, err := core.New(config) + if err := acquireProbeCore(parentCtx); err != nil { + return -1, err + } + defer releaseProbeCore() + + ctx, cancel := context.WithTimeout(parentCtx, defaultRealDelayTimeout) + defer cancel() + inst, err := core.NewWithContext(ctx, config) if err != nil { return -1, fmt.Errorf("instance creation failed: %w", err) } @@ -266,8 +297,6 @@ func MeasureOutboundDelay(ConfigureFileContent string, url string) (int64, error return -1, fmt.Errorf("startup failed: %w", err) } defer inst.Close() - ctx, cancel := context.WithTimeout(context.Background(), defaultRealDelayTimeout) - defer cancel() return measureInstDelayWithOptions(ctx, inst, url, http.MethodHead, 1, defaultRealDelayTimeout) } @@ -291,21 +320,20 @@ func (c *ProbeController) Probe( return fmt.Errorf("probe groups are invalid: %w", err) } - c.access.Lock() - if c.cancelled { - c.access.Unlock() - return context.Canceled + ctx := c.ctx + if err := ctx.Err(); err != nil { + return err } - ctx, cancel := context.WithCancel(context.Background()) - c.cancel = cancel - c.access.Unlock() - defer cancel() config, err := coreserial.LoadJSONConfig(strings.NewReader(configContent)) if err != nil { return fmt.Errorf("probe config load failed: %w", err) } config.Inbound = nil + if err := acquireProbeCore(ctx); err != nil { + return err + } + defer releaseProbeCore() inst, err := core.NewWithContext(ctx, config) if err != nil {