From ad5f2a28cb23e8e101d8369d088a3003eff3bd70 Mon Sep 17 00:00:00 2001 From: PathGao <42336971+PathGao@users.noreply.github.com> Date: Thu, 30 Jul 2026 03:00:29 +0800 Subject: [PATCH] fix(xray): synchronize lifecycle state (#6138) * fix(xray): synchronize lifecycle snapshots Protect process replacement and result caching with a lifecycle state object, so read paths keep one process snapshot while restarts swap state safely. Bound version probing to prevent a stalled binary from holding the restart lock. * test(xray): cover concurrent lifecycle reads Exercise status, result, and traffic reads while the managed process is replaced, so the race detector guards the lifecycle snapshot boundary. * fix(xray): guard process config snapshots Synchronize hot-applied config snapshots, keep Telegram reads on one lifecycle snapshot, and strengthen lifecycle timeout and concurrency regression coverage. --------- Co-authored-by: PathGao --- internal/web/service/inbound_disable.go | 8 +- internal/web/service/inbound_node.go | 33 +++-- internal/web/service/inbound_traffic.go | 4 +- internal/web/service/outbound_subscription.go | 4 +- internal/web/service/server.go | 8 +- internal/web/service/tgbot/tgbot_client.go | 4 +- internal/web/service/tgbot/tgbot_report.go | 10 +- internal/web/service/xray.go | 140 +++++++++++------- internal/web/service/xray_lifecycle_test.go | 68 +++++++++ internal/web/service/xray_setting.go | 4 +- internal/xray/process.go | 14 +- internal/xray/process_version_test.go | 65 ++++++++ 12 files changed, 275 insertions(+), 87 deletions(-) create mode 100644 internal/web/service/xray_lifecycle_test.go create mode 100644 internal/xray/process_version_test.go diff --git a/internal/web/service/inbound_disable.go b/internal/web/service/inbound_disable.go index 1976350d9..01b8f7df9 100644 --- a/internal/web/service/inbound_disable.go +++ b/internal/web/service/inbound_disable.go @@ -19,7 +19,7 @@ func (s *InboundService) disableInvalidInbounds(tx *gorm.DB) (bool, int64, error now := time.Now().Unix() * 1000 needRestart := false - if p != nil { + if process := currentXrayProcess(); process != nil { var tags []string err := tx.Table("inbounds"). Select("inbounds.tag"). @@ -28,7 +28,7 @@ func (s *InboundService) disableInvalidInbounds(tx *gorm.DB) (bool, int64, error if err != nil { return false, 0, err } - _ = s.xrayApi.Init(p.GetAPIPort()) + _ = s.xrayApi.Init(process.GetAPIPort()) for _, tag := range tags { err1 := s.xrayApi.DelInbound(tag) if err1 == nil { @@ -153,8 +153,8 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error) } } - if p != nil && len(localTargets) > 0 { - _ = s.xrayApi.Init(p.GetAPIPort()) + if process := currentXrayProcess(); process != nil && len(localTargets) > 0 { + _ = s.xrayApi.Init(process.GetAPIPort()) for _, t := range localTargets { err1 := s.xrayApi.RemoveUser(t.Tag, t.Email) if err1 == nil { diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 73fe5fdfd..c5e7fc4e1 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -1036,7 +1036,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } - if p != nil { + if process := currentXrayProcess(); process != nil { tree := snap.OnlineTree switch { case len(tree) == 0 && len(snap.OnlineEmails) > 0: @@ -1058,17 +1058,18 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi tree = remapped } } - p.SetNodeOnlineTree(nodeID, tree) + process.SetNodeOnlineTree(nodeID, tree) } return structuralChange, nil } func (s *InboundService) GetOnlineClients() []string { - if p == nil { + process := currentXrayProcess() + if process == nil { return []string{} } - return p.GetOnlineClients() + return process.GetOnlineClients() } // GetOnlineClientsByGuid returns online emails keyed by the panelGuid of the @@ -1077,11 +1078,12 @@ func (s *InboundService) GetOnlineClients() []string { // node-id keying so a client three hops down is attributed to its real node, // not the intermediate one it was synced through. func (s *InboundService) GetOnlineClientsByGuid() map[string][]string { - if p == nil { + process := currentXrayProcess() + if process == nil { return map[string][]string{} } - out := p.GetMergedNodeTrees() - if local := p.GetLocalOnlineClients(); len(local) > 0 { + out := process.GetMergedNodeTrees() + if local := process.GetLocalOnlineClients(); len(local) > 0 { if guid := s.panelGuid(); guid != "" { out[guid] = mergeEmails(out[guid], local) } @@ -1094,10 +1096,11 @@ func (s *InboundService) GetOnlineClientsByGuid() map[string][]string { // report per-inbound activity, so a GUID missing from the map means "don't // gate" for that node's inbounds. func (s *InboundService) GetActiveInboundsByGuid() map[string][]string { - if p == nil { + process := currentXrayProcess() + if process == nil { return map[string][]string{} } - active := p.GetLocalActiveInbounds() + active := process.GetLocalActiveInbounds() if len(active) == 0 { return map[string][]string{} } @@ -1109,14 +1112,14 @@ func (s *InboundService) GetActiveInboundsByGuid() map[string][]string { } func (s *InboundService) SetNodeOnlineTree(nodeID int, tree map[string][]string) { - if p != nil { - p.SetNodeOnlineTree(nodeID, tree) + if process := currentXrayProcess(); process != nil { + process.SetNodeOnlineTree(nodeID, tree) } } func (s *InboundService) ClearNodeOnlineClients(nodeID int) { - if p != nil { - p.ClearNodeOnlineClients(nodeID) + if process := currentXrayProcess(); process != nil { + process.ClearNodeOnlineClients(nodeID) } } @@ -1177,8 +1180,8 @@ func (s *InboundService) GetClientsLastOnline() (map[string]int64, error) { // xray.Process for why the local sets are kept separate from the shared // last_online column. func (s *InboundService) RefreshLocalOnlineClients(activeEmails, activeInboundTags []string) { - if p != nil { - p.RefreshLocalOnline(activeEmails, activeInboundTags, time.Now().UnixMilli(), onlineGracePeriodMs) + if process := currentXrayProcess(); process != nil { + process.RefreshLocalOnline(activeEmails, activeInboundTags, time.Now().UnixMilli(), onlineGracePeriodMs) } } diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index ce61bd8fa..ae47a8ad4 100644 --- a/internal/web/service/inbound_traffic.go +++ b/internal/web/service/inbound_traffic.go @@ -452,8 +452,8 @@ func (s *InboundService) autoRenewClients(tx *gorm.DB) (bool, int64, error) { if err = clearGlobalTraffic(tx, renewEmails...); err != nil { return false, 0, err } - if p != nil { - err1 = s.xrayApi.Init(p.GetAPIPort()) + if process := currentXrayProcess(); process != nil { + err1 = s.xrayApi.Init(process.GetAPIPort()) if err1 != nil { return true, int64(len(traffics)), nil } diff --git a/internal/web/service/outbound_subscription.go b/internal/web/service/outbound_subscription.go index 6f5e92d5b..3136e7bad 100644 --- a/internal/web/service/outbound_subscription.go +++ b/internal/web/service/outbound_subscription.go @@ -30,8 +30,8 @@ import ( // CheckXrayConfig's version gate. func filterOutboundsRejectedByCore(label string, outbounds []any) ([]any, []string) { coreVersion := "Unknown" - if p != nil { - coreVersion = p.GetXrayVersion() + if process := currentXrayProcess(); process != nil { + coreVersion = process.GetXrayVersion() } kept := make([]any, 0, len(outbounds)) var dropped []string diff --git a/internal/web/service/server.go b/internal/web/service/server.go index 7afa144ca..2ccc3072c 100644 --- a/internal/web/service/server.go +++ b/internal/web/service/server.go @@ -624,8 +624,8 @@ func (s *ServerService) GetStatus(lastStatus *Status) *Status { status.AppStats.Mem = rtm.Sys } status.AppStats.Threads = uint32(runtime.NumGoroutine()) - if p != nil && p.IsRunning() { - status.AppStats.Uptime = p.GetUptime() + if process := currentXrayProcess(); process != nil && process.IsRunning() { + status.AppStats.Uptime = process.GetUptime() } else { status.AppStats.Uptime = 0 } @@ -668,8 +668,8 @@ func (s *ServerService) AppendStatusSample(t time.Time, status *Status) { systemMetrics.append("tcpCount", t, float64(status.TcpCount)) systemMetrics.append("udpCount", t, float64(status.UdpCount)) online := 0 - if p != nil && p.IsRunning() { - online = len(p.GetOnlineClients()) + if process := currentXrayProcess(); process != nil && process.IsRunning() { + online = len(process.GetOnlineClients()) } systemMetrics.append("online", t, float64(online)) if len(status.Loads) >= 3 { diff --git a/internal/web/service/tgbot/tgbot_client.go b/internal/web/service/tgbot/tgbot_client.go index 50c008025..f5d3ce65d 100644 --- a/internal/web/service/tgbot/tgbot_client.go +++ b/internal/web/service/tgbot/tgbot_client.go @@ -494,8 +494,8 @@ func (t *Tgbot) clientInfoMsg( status := t.I18nBot("tgbot.offline") isOnline := false - if service.XrayProcess().IsRunning() { - if slices.Contains(service.XrayProcess().GetOnlineClients(), traffic.Email) { + if process := service.XrayProcess(); process != nil && process.IsRunning() { + if slices.Contains(process.GetOnlineClients(), traffic.Email) { status = t.I18nBot("tgbot.online") isOnline = true } diff --git a/internal/web/service/tgbot/tgbot_report.go b/internal/web/service/tgbot/tgbot_report.go index 8f2afa972..4b587733c 100644 --- a/internal/web/service/tgbot/tgbot_report.go +++ b/internal/web/service/tgbot/tgbot_report.go @@ -105,7 +105,10 @@ func (t *Tgbot) prepareServerUsageInfo() string { t.lastStatus = t.serverService.GetStatus(t.lastStatus) t.setCachedStatus(t.lastStatus) } - onlines := service.XrayProcess().GetOnlineClients() + var onlines []string + if process := service.XrayProcess(); process != nil { + onlines = process.GetOnlineClients() + } info += t.I18nBot("tgbot.messages.hostname", "Hostname=="+hostname) info += t.I18nBot("tgbot.messages.version", "Version=="+config.GetPanelVersion()) @@ -361,11 +364,12 @@ func (t *Tgbot) notifyExhausted() { // onlineClients retrieves and sends information about online clients. func (t *Tgbot) onlineClients(chatId int64, messageID ...int) { - if !service.XrayProcess().IsRunning() { + process := service.XrayProcess() + if process == nil || !process.IsRunning() { return } - onlines := service.XrayProcess().GetOnlineClients() + onlines := process.GetOnlineClients() onlinesCount := len(onlines) output := t.I18nBot("tgbot.messages.onlinesCount", "Count=="+fmt.Sprint(onlinesCount)) keyboard := tu.InlineKeyboard(tu.InlineKeyboardRow( diff --git a/internal/web/service/xray.go b/internal/web/service/xray.go index 8ec3972f1..45f3eebaf 100644 --- a/internal/web/service/xray.go +++ b/internal/web/service/xray.go @@ -20,13 +20,44 @@ import ( ) var ( - p *xray.Process lock sync.Mutex isNeedXrayRestart atomic.Bool // Indicates that restart was requested for Xray isManuallyStopped atomic.Bool // Indicates that Xray was stopped manually from the panel - result string + xrayState xrayLifecycle ) +type xrayLifecycle struct { + mu sync.RWMutex + process *xray.Process + result string +} + +func (s *xrayLifecycle) snapshot() (*xray.Process, string) { + s.mu.RLock() + defer s.mu.RUnlock() + return s.process, s.result +} + +func (s *xrayLifecycle) replace(process *xray.Process) { + s.mu.Lock() + s.process = process + s.result = "" + s.mu.Unlock() +} + +func (s *xrayLifecycle) storeResult(process *xray.Process, result string) { + s.mu.Lock() + if s.process == process && s.result == "" { + s.result = result + } + s.mu.Unlock() +} + +func currentXrayProcess() *xray.Process { + process, _ := xrayState.snapshot() + return process +} + // XrayService provides business logic for Xray process management. // It handles starting, stopping, restarting Xray, and managing its configuration. type XrayService struct { @@ -38,23 +69,25 @@ type XrayService struct { // IsXrayRunning checks if the Xray process is currently running. func (s *XrayService) IsXrayRunning() bool { - return p != nil && p.IsRunning() + process := currentXrayProcess() + return process != nil && process.IsRunning() } // XrayProcess returns the current Xray process instance (may be nil when Xray -// is not running). It exposes the package-level process to callers outside this -// package (e.g. the tgbot subpackage) without changing access semantics. +// is not running). It exposes the lifecycle snapshot to callers outside this +// package (e.g. the tgbot subpackage). func XrayProcess() *xray.Process { - return p + return currentXrayProcess() } // GetXrayErr returns the error from the Xray process, if any. func (s *XrayService) GetXrayErr() error { - if p == nil { + process := currentXrayProcess() + if process == nil { return nil } - err := p.GetErr() + err := process.GetErr() if err == nil { return nil } @@ -70,17 +103,15 @@ func (s *XrayService) GetXrayErr() error { // GetXrayResult returns the result string from the Xray process. func (s *XrayService) GetXrayResult() string { - if result != "" { - return result + process, cachedResult := xrayState.snapshot() + if cachedResult != "" { + return cachedResult } - if s.IsXrayRunning() { - return "" - } - if p == nil { + if process == nil || process.IsRunning() { return "" } - result = p.GetResult() + result := process.GetResult() if runtime.GOOS == "windows" && result == "exit status 1" { // exit status 1 on Windows means that Xray process was killed @@ -88,15 +119,17 @@ func (s *XrayService) GetXrayResult() string { return "" } + xrayState.storeResult(process, result) return result } // GetXrayVersion returns the version of the running Xray process. func (s *XrayService) GetXrayVersion() string { - if p == nil { + process := currentXrayProcess() + if process == nil { return "Unknown" } - return p.GetXrayVersion() + return process.GetXrayVersion() } // RemoveIndex removes an element at the specified index from a slice. @@ -838,12 +871,13 @@ func stripDisabledRules(routerCfg json_util.RawMessage) json_util.RawMessage { // GetXrayTraffic fetches the current traffic statistics from the running Xray process. func (s *XrayService) GetXrayTraffic() ([]*xray.Traffic, []*xray.ClientTraffic, error) { - if !s.IsXrayRunning() { + process := currentXrayProcess() + if process == nil || !process.IsRunning() { err := errors.New("xray is not running") logger.Debug("Attempted to fetch Xray traffic, but Xray is not running:", err) return nil, nil, err } - apiPort := p.GetAPIPort() + apiPort := process.GetAPIPort() if err := s.xrayAPI.Init(apiPort); err != nil { logger.Debug("Failed to initialize Xray API:", err) return nil, nil, err @@ -866,13 +900,14 @@ func (s *XrayService) GetXrayTraffic() ([]*xray.Traffic, []*xray.ClientTraffic, // core as unsupported until the next restart, while transient errors leave the // capability undecided so a flaky poll can't lock in legacy mode. func (s *XrayService) GetOnlineUsers() ([]xray.OnlineUser, bool, error) { - if !s.IsXrayRunning() { + process := currentXrayProcess() + if process == nil || !process.IsRunning() { return nil, false, nil } - if p.OnlineAPISupport() == xray.OnlineAPIUnsupported { + if process.OnlineAPISupport() == xray.OnlineAPIUnsupported { return nil, false, nil } - if err := s.xrayAPI.Init(p.GetAPIPort()); err != nil { + if err := s.xrayAPI.Init(process.GetAPIPort()); err != nil { logger.Debug("Failed to initialize Xray API:", err) return nil, false, err } @@ -881,15 +916,15 @@ func (s *XrayService) GetOnlineUsers() ([]xray.OnlineUser, bool, error) { users, err := s.xrayAPI.GetOnlineUsers() if err != nil { if xray.IsUnimplementedErr(err) { - p.SetOnlineAPISupport(xray.OnlineAPIUnsupported) + process.SetOnlineAPISupport(xray.OnlineAPIUnsupported) logger.Info("xray core does not support the online-stats API; falling back to traffic-delta onlines and access-log IP limit") return nil, false, nil } logger.Debug("Failed to fetch Xray online users:", err) return nil, false, err } - if p.OnlineAPISupport() == xray.OnlineAPIUnknown { - p.SetOnlineAPISupport(xray.OnlineAPISupported) + if process.OnlineAPISupport() == xray.OnlineAPIUnknown { + process.SetOnlineAPISupport(xray.OnlineAPISupported) logger.Info("xray core supports the online-stats API; using connection-based onlines and access-log-free IP limit") } return users, true, nil @@ -911,13 +946,14 @@ type BalancerStatus struct { // balancers alongside live ones. func (s *XrayService) GetBalancersStatus(tags []string) ([]BalancerStatus, error) { statuses := make([]BalancerStatus, 0, len(tags)) - if !s.IsXrayRunning() { + process := currentXrayProcess() + if process == nil || !process.IsRunning() { for _, tag := range tags { statuses = append(statuses, BalancerStatus{Tag: tag}) } return statuses, nil } - if err := s.xrayAPI.Init(p.GetAPIPort()); err != nil { + if err := s.xrayAPI.Init(process.GetAPIPort()); err != nil { return nil, err } defer s.xrayAPI.Close() @@ -944,7 +980,8 @@ func (s *XrayService) GetBalancersStatus(tags []string) ([]BalancerStatus, error // another balancer, the override resolves to the loopback outbound that // routes traffic through the target balancer via the routing rules. func (s *XrayService) OverrideBalancer(tag, target string) error { - if !s.IsXrayRunning() { + process := currentXrayProcess() + if process == nil || !process.IsRunning() { return errors.New("xray is not running") } if target != "" { @@ -956,7 +993,7 @@ func (s *XrayService) OverrideBalancer(tag, target string) error { target = resolved } } - if err := s.xrayAPI.Init(p.GetAPIPort()); err != nil { + if err := s.xrayAPI.Init(process.GetAPIPort()); err != nil { return err } defer s.xrayAPI.Close() @@ -1002,10 +1039,11 @@ func (s *XrayService) resolveOverrideTarget(target string) (string, error) { // TestRoute asks the running core which outbound its router picks for the // described connection. func (s *XrayService) TestRoute(req xray.RouteTestRequest) (*xray.RouteTestResult, error) { - if !s.IsXrayRunning() { + process := currentXrayProcess() + if process == nil || !process.IsRunning() { return nil, errors.New("xray is not running") } - if err := s.xrayAPI.Init(p.GetAPIPort()); err != nil { + if err := s.xrayAPI.Init(process.GetAPIPort()); err != nil { return nil, err } defer s.xrayAPI.Close() @@ -1031,23 +1069,24 @@ func (s *XrayService) RestartXray(isForce bool) error { return err } - if s.IsXrayRunning() { - configUnchanged := p.GetConfig().Equals(xrayConfig) + process := currentXrayProcess() + if process != nil && process.IsRunning() { + configUnchanged := process.GetConfig().Equals(xrayConfig) if !isForce && configUnchanged && !isNeedXrayRestart.Load() { logger.Debug("It does not need to restart Xray") return nil } - if !isForce && !configUnchanged && s.tryHotApply(xrayConfig) { + if !isForce && !configUnchanged && s.tryHotApply(process, xrayConfig) { logger.Info("Xray config changes applied through the core API, no restart needed") return nil } - _ = p.Stop() + _ = process.Stop() } - p = xray.NewProcess(xrayConfig) - result = "" + process = xray.NewProcess(xrayConfig) + xrayState.replace(process) s.xrayAPI.StatsLastValues = nil - err = p.Start() + err = process.Start() if err != nil { return err } @@ -1061,19 +1100,19 @@ func (s *XrayService) RestartXray(isForce bool) error { // instance now matches newCfg; on any failure it returns false and the // caller falls back to a full process restart, which cleans up whatever was // partially applied. Callers must hold the package-level lock. -func (s *XrayService) tryHotApply(newCfg *xray.Config) bool { - oldCfg := p.GetConfig() +func (s *XrayService) tryHotApply(process *xray.Process, newCfg *xray.Config) bool { + oldCfg := process.GetConfig() diff, ok := xray.ComputeHotDiff(oldCfg, newCfg) if !ok { logger.Debug("hot apply: config change is not API-applicable, falling back to restart") return false } if diff.Empty() { - p.SetConfig(newCfg) + process.SetConfig(newCfg) return true } - apiPort := p.GetAPIPort() + apiPort := process.GetAPIPort() if apiPort <= 0 { return false } @@ -1131,7 +1170,7 @@ func (s *XrayService) tryHotApply(newCfg *xray.Config) bool { } } - p.SetConfig(newCfg) + process.SetConfig(newCfg) return true } @@ -1192,8 +1231,9 @@ func (s *XrayService) StopXray() error { defer lock.Unlock() isManuallyStopped.Store(true) logger.Debug("Attempting to stop Xray...") - if s.IsXrayRunning() { - return p.Stop() + process := currentXrayProcess() + if process != nil && process.IsRunning() { + return process.Stop() } return errors.New("xray is not running") } @@ -1205,14 +1245,14 @@ func (s *XrayService) SetToNeedRestart() { // GetXrayAPIPort returns the port the local xray process is listening on // for its gRPC HandlerService, or 0 when xray isn't currently running. -// Exposed for the runtime package's LocalRuntime adapter — runtime can't -// reach into the package-level `p` directly without a service-package -// import cycle. +// Exposed for the runtime package's LocalRuntime adapter without a +// service-package import cycle. func (s *XrayService) GetXrayAPIPort() int { - if p == nil || !p.IsRunning() { + process := currentXrayProcess() + if process == nil || !process.IsRunning() { return 0 } - return p.GetAPIPort() + return process.GetAPIPort() } // IsNeedRestartAndSetFalse checks if restart is needed and resets the flag to false. diff --git a/internal/web/service/xray_lifecycle_test.go b/internal/web/service/xray_lifecycle_test.go new file mode 100644 index 000000000..f0bf27f35 --- /dev/null +++ b/internal/web/service/xray_lifecycle_test.go @@ -0,0 +1,68 @@ +package service + +import ( + "sync" + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +func TestXrayLifecycleSnapshotDoesNotOverwriteNewerResult(t *testing.T) { + state := xrayLifecycle{} + first := xray.NewProcess(&xray.Config{}) + second := xray.NewProcess(&xray.Config{}) + + state.replace(first) + state.storeResult(first, "first result") + process, result := state.snapshot() + if process != first || result != "first result" { + t.Fatalf("snapshot = (%p, %q), want (%p, %q)", process, result, first, "first result") + } + state.replace(second) + state.storeResult(first, "old result") + + process, result = state.snapshot() + if process != second { + t.Fatal("snapshot returned the replaced process") + } + if result != "" { + t.Fatalf("snapshot result = %q, want empty", result) + } +} + +func TestXrayLifecycleConcurrentStatusResultAndTrafficReads(t *testing.T) { + previousProcess, previousResult := xrayState.snapshot() + t.Cleanup(func() { + xrayState.mu.Lock() + xrayState.process = previousProcess + xrayState.result = previousResult + xrayState.mu.Unlock() + }) + + first := xray.NewProcess(&xray.Config{}) + second := xray.NewProcess(&xray.Config{}) + service := XrayService{} + var wg sync.WaitGroup + + wg.Go(func() { + for range 200 { + xrayState.replace(first) + xrayState.replace(second) + } + }) + + for range 4 { + wg.Go(func() { + for range 200 { + _ = service.IsXrayRunning() + _ = service.GetXrayResult() + _, _, err := service.GetXrayTraffic() + if err == nil || err.Error() != "xray is not running" { + t.Errorf("GetXrayTraffic error = %v, want xray is not running", err) + } + } + }) + } + + wg.Wait() +} diff --git a/internal/web/service/xray_setting.go b/internal/web/service/xray_setting.go index 7a1a4ef12..ffe881121 100644 --- a/internal/web/service/xray_setting.go +++ b/internal/web/service/xray_setting.go @@ -54,8 +54,8 @@ func (s *XraySettingService) CheckXrayConfig(XrayTemplateConfig string) error { return common.NewError("xray template config invalid: outbounds is not an array:", err) } coreVersion := "Unknown" - if p != nil { - coreVersion = p.GetXrayVersion() + if process := currentXrayProcess(); process != nil { + coreVersion = process.GetXrayVersion() } for _, outbound := range outbounds { if err := xray.ValidateOutboundConfig(outbound); err != nil { diff --git a/internal/xray/process.go b/internal/xray/process.go index 7c1a93fe5..9e56d7d7b 100644 --- a/internal/xray/process.go +++ b/internal/xray/process.go @@ -126,8 +126,9 @@ func NewTestProcess(xrayConfig *Config, configPath string) *Process { } type process struct { - // mu guards the process lifecycle fields (cmd, done, exitErr) plus version and - // apiPort, which are written by Start/startCommand/refreshVersion/refreshAPIPort + // mu guards the process lifecycle fields (cmd, done, exitErr) plus version, + // apiPort, and config, which are written by Start/startCommand/refreshVersion/ + // refreshAPIPort/SetConfig // while being read concurrently by IsRunning/GetErr/GetResult/GetXrayVersion/ // GetAPIPort/Stop from other goroutines (status endpoint, check-xray-running // and traffic jobs). Snapshot under the lock, then do any blocking syscall @@ -219,6 +220,7 @@ func (p *process) SetOnlineAPISupport(v OnlineAPISupport) { var ( xrayGracefulStopTimeout = 5 * time.Second xrayForceStopTimeout = 2 * time.Second + xrayVersionTimeout = 5 * time.Second // OnCrash is called when xray crashes unexpectedly. Set from web layer. OnCrash func(err error) ) @@ -296,6 +298,8 @@ func (p *Process) GetAPIPort() int { // GetConfig returns the configuration used by the Xray process. func (p *Process) GetConfig() *Config { + p.mu.RLock() + defer p.mu.RUnlock() return p.config } @@ -303,6 +307,8 @@ func (p *Process) GetConfig() *Config { // process has been reconciled with it through the gRPC API (hot apply), so // later change detection compares against what is actually running. func (p *Process) SetConfig(config *Config) { + p.mu.Lock() + defer p.mu.Unlock() p.config = config } @@ -494,7 +500,9 @@ func (p *process) refreshAPIPort() { // refreshVersion updates the version string by running the Xray binary with -version. func (p *process) refreshVersion() { version := "Unknown" - cmd := exec.CommandContext(context.Background(), GetBinaryPath(), "-version") + ctx, cancel := context.WithTimeout(context.Background(), xrayVersionTimeout) + defer cancel() + cmd := exec.CommandContext(ctx, GetBinaryPath(), "-version") if data, err := cmd.Output(); err == nil { if datas := bytes.Split(data, []byte(" ")); len(datas) > 1 { version = string(datas[1]) diff --git a/internal/xray/process_version_test.go b/internal/xray/process_version_test.go new file mode 100644 index 000000000..82979e5eb --- /dev/null +++ b/internal/xray/process_version_test.go @@ -0,0 +1,65 @@ +package xray + +import ( + "os" + "path/filepath" + "runtime" + "sync" + "testing" + "time" +) + +func TestRefreshVersionTimesOut(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell fixture is Unix-only") + } + + dir := t.TempDir() + t.Setenv("XUI_BIN_FOLDER", dir) + binaryPath := filepath.Join(dir, GetBinaryName()) + if err := os.WriteFile(binaryPath, []byte("#!/bin/sh\nexec sleep 1\n"), 0o700); err != nil { + t.Fatalf("write xray fixture: %v", err) + } + + previousTimeout := xrayVersionTimeout + xrayVersionTimeout = 20 * time.Millisecond + t.Cleanup(func() { xrayVersionTimeout = previousTimeout }) + + p := newProcess(&Config{}) + started := time.Now() + p.refreshVersion() + elapsed := time.Since(started) + if elapsed < xrayVersionTimeout { + t.Fatalf("refreshVersion duration = %s, want at least %s", elapsed, xrayVersionTimeout) + } + if elapsed > 500*time.Millisecond { + t.Fatalf("refreshVersion duration = %s, want under 500ms", elapsed) + } + if got := p.GetXrayVersion(); got != "Unknown" { + t.Fatalf("version = %q, want Unknown", got) + } +} + +func TestProcessConfigSnapshotsAreRaceSafe(t *testing.T) { + p := NewProcess(&Config{}) + first := &Config{} + second := &Config{} + var wg sync.WaitGroup + + wg.Go(func() { + for range 1000 { + p.SetConfig(first) + p.SetConfig(second) + } + }) + for range 4 { + wg.Go(func() { + for range 1000 { + if p.GetConfig() == nil { + t.Error("config = nil") + } + } + }) + } + wg.Wait() +}