diff --git a/backend/xray/config.go b/backend/xray/config.go index 965c3018..2391a527 100644 --- a/backend/xray/config.go +++ b/backend/xray/config.go @@ -73,10 +73,26 @@ type inboundUpdate struct { removeEmailSet map[string]struct{} } +// canHoldUsers reports whether this inbound may hold users the node manages. +// It lists the Xray inbound protocols that never take users from the node +// (dokodemo-door and its alias tunnel, which the node's own API_INBOUND uses, +// socks and mixed, http, tun): removing a user from them is a wasted Xray API +// call. Every other protocol counts as holding users, so a protocol added to +// isActiveInbound, syncUsers, updateUsers and the ToBytes switch gets its +// removals without a change here; one the node does not manage yet only costs +// a remove call that fails. Xray matches protocol names case-insensitively. +func (i *Inbound) canHoldUsers() bool { + switch strings.ToLower(i.Protocol) { + case "dokodemo-door", "tunnel", "socks", "mixed", "http", "tun": + return false + } + return true +} + func (c *Config) buildInboundUpdates(users []*common.User) (map[string]*Inbound, map[string]*inboundUpdate) { inboundByTag := make(map[string]*Inbound) for _, inbound := range c.InboundConfigs { - if inbound.exclude { + if inbound.exclude || !inbound.canHoldUsers() { continue } inboundByTag[inbound.Tag] = inbound diff --git a/backend/xray/jobs.go b/backend/xray/jobs.go index a02d337a..c1bf31b4 100644 --- a/backend/xray/jobs.go +++ b/backend/xray/jobs.go @@ -87,16 +87,39 @@ func (x *Xray) checkXrayHealth(baseCtx context.Context) { maxFailures := 10 // Give Xray API time to recover under load before restarting. checkInterval := 2 * time.Second + probe := func() error { + ctx, cancel := context.WithTimeout(baseCtx, 3*time.Second) + defer cancel() + _, err := x.GetSysStats(ctx) + return err + } + restart := func(reason string) { + // Lock order: syncMu, then mu, then core.mu, as in the user syncs. + // syncMu keeps user syncs off the new core until its API answers. + x.syncMu.Lock() + defer x.syncMu.Unlock() + + // A full sync may have replaced the core while this restart waited + // for syncMu: leave a core that answers alone. + if x.Started() && probe() == nil { + log.Println("xray answers again, health restart skipped") + consecutiveFailures = 0 + return + } + log.Println(reason) if tail := x.core.RuntimeLogTail(10); len(tail) > 0 { log.Printf("last %d xray log lines before restart:\n%s", len(tail), strings.Join(tail, "\n")) } - if err := x.Restart(); err != nil { + // Count from zero for the new core even if it is not ready yet, so a + // core still booting after restartReadyTimeout gets maxFailures more + // probes before the next restart. + consecutiveFailures = 0 + if err := x.restartLocked(baseCtx); err != nil { log.Println(err.Error()) } else { log.Println("xray restarted") - consecutiveFailures = 0 } } @@ -111,12 +134,11 @@ func (x *Xray) checkXrayHealth(baseCtx context.Context) { continue } - ctx, cancel := context.WithTimeout(baseCtx, 3*time.Second) - _, err := x.GetSysStats(ctx) - cancel() - - if err != nil { - if errors.Is(err, context.Canceled) { + if err := probe(); err != nil { + // Shutdown cancels baseCtx. Check it directly: GetSysStats + // re-wraps the cancelled call as a gRPC status, which + // errors.Is(err, context.Canceled) does not match. + if baseCtx.Err() != nil { return } diff --git a/backend/xray/restart_unix_test.go b/backend/xray/restart_unix_test.go new file mode 100644 index 00000000..effc8bbb --- /dev/null +++ b/backend/xray/restart_unix_test.go @@ -0,0 +1,324 @@ +//go:build !windows + +package xray + +import ( + "context" + "encoding/json" + "net" + "os" + "path/filepath" + "strconv" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" + statsCommand "github.com/xtls/xray-core/app/stats/command" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/pasarguard/node/backend/xray/api" + "github.com/pasarguard/node/common" + "github.com/pasarguard/node/config" + "github.com/pasarguard/node/pkg/logstream" + "github.com/pasarguard/node/pkg/netutil" +) + +// fakeStatsService stands in for xray's StatsService. GetSysStats fails like a +// booting core until healthy is set, or blocks until the caller gives up. +type fakeStatsService struct { + statsCommand.UnimplementedStatsServiceServer + healthy atomic.Bool + block bool + probes chan struct{} +} + +func (s *fakeStatsService) GetSysStats(ctx context.Context, _ *statsCommand.SysStatsRequest) (*statsCommand.SysStatsResponse, error) { + if s.probes != nil { + select { + case s.probes <- struct{}{}: + default: + } + } + if s.block { + <-ctx.Done() + return nil, ctx.Err() + } + if !s.healthy.Load() { + return nil, status.Error(codes.Unavailable, "xray is booting") + } + return &statsCommand.SysStatsResponse{}, nil +} + +// newFakeCoreXray returns an Xray whose core runs a stand-in for the xray +// binary (a process that never opens the API) and whose API client talks to +// stats. The core is not started. +func newFakeCoreXray(t *testing.T, stats *fakeStatsService) *Xray { + t.Helper() + + dir := t.TempDir() + executable := filepath.Join(dir, "xray") + if err := os.WriteFile(executable, []byte("#!/bin/sh\nexec sleep 600\n"), 0o755); err != nil { + t.Fatal(err) + } + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer() + statsCommand.RegisterStatsServiceServer(server, stats) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + + handler, err := api.NewXrayAPI(listener.Addr().(*net.TCPAddr).Port) + if err != nil { + t.Fatal(err) + } + t.Cleanup(handler.Close) + + cfg, err := NewConfig(`{"inbounds":[]}`, nil) + if err != nil { + t.Fatal(err) + } + + core := &Core{ + executablePath: executable, + assetsPath: dir, + configPath: dir, + logs: logstream.NewBuffer(100), + logPhase: logPhaseRuntime, + startupLogSize: 200, + runtimeLogs: newStartupLogRing(10), + } + t.Cleanup(core.Stop) + + _, cancel := context.WithCancel(context.Background()) + return &Xray{config: cfg, cfg: &config.Config{}, core: core, handler: handler, cancelFunc: cancel} +} + +func corePID(x *Xray) int { + x.core.mu.Lock() + defer x.core.mu.Unlock() + return x.core.processPID +} + +func waitUntil(t *testing.T, what string, condition func() bool) { + t.Helper() + + deadline := time.Now().Add(10 * time.Second) + for !condition() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting until %s", what) + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestSyncUsersDoesNotStartXrayAfterShutdown(t *testing.T) { + x := newFakeCoreXray(t, &fakeStatsService{}) + + syncErr := make(chan error, 1) + go func() { syncErr <- x.SyncUsers(context.Background(), nil) }() + waitUntil(t, "the full sync starts its candidate core", x.core.Started) + + // Stop while the candidate is still booting: Shutdown kills it, the sync + // sees the core die and rolls back to the previous config. + x.Shutdown() + + for start := time.Now(); time.Since(start) < 3*time.Second; time.Sleep(10 * time.Millisecond) { + if x.core.Started() { + t.Fatal("the full-sync rollback started xray again after Shutdown") + } + } + select { + case err := <-syncErr: + if err == nil { + t.Fatal("SyncUsers reported success after Shutdown") + } + case <-time.After(10 * time.Second): + t.Fatal("SyncUsers still running 10s after Shutdown") + } +} + +func TestHealthLoopStopsWhenCancelledDuringAProbe(t *testing.T) { + stats := &fakeStatsService{block: true, probes: make(chan struct{}, 1)} + x := newFakeCoreXray(t, stats) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + x.checkXrayHealth(ctx) + close(done) + }() + + select { + case <-stats.probes: + case <-time.After(5 * time.Second): + t.Fatal("the health loop sent no probe") + } + // Shutdown cancels the health context while the probe is in flight. + cancel() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatalf("the health loop still runs 1s after its context was cancelled (xray restarted: %v)", x.core.Started()) + } + if x.core.Started() { + t.Fatal("the health loop restarted xray after its context was cancelled") + } +} + +func TestHealthRestartLeavesACoreThatAFullSyncReplaced(t *testing.T) { + stats := &fakeStatsService{probes: make(chan struct{}, 1)} + x := newFakeCoreXray(t, stats) + + // A full sync holds syncMu while its candidate core boots. + x.syncMu.Lock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go x.checkXrayHealth(ctx) + + select { + case <-stats.probes: + case <-time.After(5 * time.Second): + x.syncMu.Unlock() + t.Fatal("the health loop sent no probe") + } + time.Sleep(300 * time.Millisecond) + if x.core.Started() { + x.syncMu.Unlock() + t.Fatal("the health loop restarted xray while a full sync held syncMu") + } + + // The sync's core comes up and answers, then the sync releases syncMu. + if err := x.restartCoreWithConfig(x.config); err != nil { + x.syncMu.Unlock() + t.Fatal(err) + } + pid := corePID(x) + stats.healthy.Store(true) + x.syncMu.Unlock() + + time.Sleep(500 * time.Millisecond) + if got := corePID(x); got != pid { + t.Fatalf("the health loop restarted the core the full sync had just started (pid %d, now %d)", pid, got) + } +} + +func realXrayUsers(n int, inbounds []string) []*common.User { + users := make([]*common.User, 0, n) + for i := range n { + users = append(users, &common.User{ + Email: strconv.Itoa(i + 1), + Inbounds: inbounds, + Proxies: &common.Proxy{ + Vmess: &common.Vmess{Id: uuid.NewString()}, + Vless: &common.Vless{Id: uuid.NewString()}, + Trojan: &common.Trojan{Password: "trojan-" + strconv.Itoa(i)}, + }, + }) + } + return users +} + +func realXrayConfig(t *testing.T, tags []string) *Config { + t.Helper() + + inbounds := make([]map[string]any, 0, len(tags)) + for _, tag := range tags { + settings := map[string]any{"clients": []any{}} + if tag == Vless { + settings["decryption"] = "none" + } + inbounds = append(inbounds, map[string]any{ + "tag": tag, + "listen": "127.0.0.1", + "port": netutil.FindFreePort(), + "protocol": tag, + "settings": settings, + }) + } + raw, err := json.Marshal(map[string]any{ + "log": map[string]any{"loglevel": "warning"}, + "inbounds": inbounds, + "outbounds": []any{map[string]any{"protocol": "freedom", "tag": "DIRECT"}}, + }) + if err != nil { + t.Fatal(err) + } + cfg, err := NewConfig(string(raw), nil) + if err != nil { + t.Fatal(err) + } + return cfg +} + +// Removals sent while xray boots after a restart used to fail against the +// booting core and be dropped, while the new core, built from the user list +// before the removal, kept the users live. +func TestRestartHoldsUserSyncsUntilTheXrayAPIIsUp(t *testing.T) { + executable := resolveXrayBinary(t) + tags := []string{Vless, Vmess, Trojan} + users := realXrayUsers(1000, tags) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + nodeCfg := &config.Config{ + XrayExecutablePath: executable, + XrayAssetsPath: assetsPath, + GeneratedConfigPath: t.TempDir(), + LogBufferSize: 1000, + } + x, err := New(ctx, realXrayConfig(t, tags), users, netutil.FindFreePort(), netutil.FindFreePort(), nodeCfg) + if err != nil { + t.Fatal(err) + } + t.Cleanup(x.Shutdown) + + pid := corePID(x) + restartErr := make(chan error, 1) + go func() { restartErr <- x.Restart() }() + waitUntil(t, "the restart spawns a new xray", func() bool { + current := corePID(x) + return current != 0 && current != pid + }) + + // The panel disables users while the new core boots: they are sent with no inbounds. + removed := make([]*common.User, 0, 51) + for _, user := range users[:51] { + removed = append(removed, &common.User{Email: user.Email, Proxies: user.Proxies}) + } + if err := x.UpdateUsers(ctx, removed[:50]); err != nil { + t.Fatal(err) + } + if err := x.SyncUser(ctx, removed[50]); err != nil { + t.Fatal(err) + } + if err := <-restartErr; err != nil { + t.Fatal(err) + } + waitUntil(t, "the xray API answers", func() bool { + probeCtx, probeCancel := context.WithTimeout(ctx, time.Second) + defer probeCancel() + _, err := x.GetSysStats(probeCtx) + return err == nil + }) + + // Xray's RemoveUser returns nil only for a user it still has. + live := 0 + for _, user := range removed { + for _, tag := range tags { + if x.handler.RemoveInboundUser(ctx, tag, user.GetEmail()) == nil { + live++ + } + } + } + if live > 0 { + t.Fatalf("%d of %d removed user-inbound pairs are still accepted by xray", live, len(removed)*len(tags)) + } +} diff --git a/backend/xray/user.go b/backend/xray/user.go index 28f3ef54..db7092b8 100644 --- a/backend/xray/user.go +++ b/backend/xray/user.go @@ -224,6 +224,12 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { x.syncMu.Lock() defer x.syncMu.Unlock() + // The request may have been abandoned while it queued behind a restart or + // another sync; running it now would only cause pointless work. + if err := ctx.Err(); err != nil { + return err + } + proxySetting, err := setupUserAccount(user) if err != nil { return err @@ -237,7 +243,7 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { userInbounds := user.GetInbounds() for _, inbound := range inbounds { - if inbound.exclude { + if inbound.exclude || !inbound.canHoldUsers() { continue } @@ -276,6 +282,12 @@ func (x *Xray) SyncUsers(ctx context.Context, users []*common.User) error { x.syncMu.Lock() defer x.syncMu.Unlock() + // The request may have been abandoned while it queued behind a restart or + // another sync; running it now would only cause pointless work. + if err := ctx.Err(); err != nil { + return err + } + candidate, err := x.config.Clone() if err != nil { return err @@ -330,6 +342,12 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { x.syncMu.Lock() defer x.syncMu.Unlock() + // The request may have been abandoned while it queued behind a restart or + // another sync; running it now would only cause pointless work. + if err := ctx.Err(); err != nil { + return err + } + handler := x.handler inboundByTag, updates := x.config.buildInboundUpdates(users) var errMessage strings.Builder @@ -382,6 +400,12 @@ func (x *Xray) UpdateUsersAndRestart(ctx context.Context, users []*common.User) x.syncMu.Lock() defer x.syncMu.Unlock() + // The request may have been abandoned while it queued behind a restart or + // another sync; running it now would only cause pointless work. + if err := ctx.Err(); err != nil { + return err + } + candidate, err := x.config.Clone() if err != nil { return err diff --git a/backend/xray/user_abandoned_test.go b/backend/xray/user_abandoned_test.go new file mode 100644 index 00000000..ccca3250 --- /dev/null +++ b/backend/xray/user_abandoned_test.go @@ -0,0 +1,36 @@ +package xray + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/pasarguard/node/common" +) + +func TestSyncCallsAbandonedWhileQueuedReturnTheContextError(t *testing.T) { + x := &Xray{} + ctx, cancel := context.WithCancel(context.Background()) + + x.syncMu.Lock() + errs := make(chan error, 4) + go func() { errs <- x.SyncUser(ctx, &common.User{}) }() + go func() { errs <- x.SyncUsers(ctx, nil) }() + go func() { errs <- x.UpdateUsers(ctx, nil) }() + go func() { errs <- x.UpdateUsersAndRestart(ctx, nil) }() + time.Sleep(100 * time.Millisecond) + cancel() + x.syncMu.Unlock() + + for range 4 { + select { + case err := <-errs: + if !errors.Is(err, context.Canceled) { + t.Fatalf("got %v, want context.Canceled", err) + } + case <-time.After(5 * time.Second): + t.Fatal("a queued sync call did not return after its context was cancelled") + } + } +} diff --git a/backend/xray/user_sync_test.go b/backend/xray/user_sync_test.go index fcc8560d..63fbd877 100644 --- a/backend/xray/user_sync_test.go +++ b/backend/xray/user_sync_test.go @@ -296,3 +296,66 @@ func TestUpdateUsersRetriesAnAddThatFailed(t *testing.T) { } assertOperations(t, service.take(), operations(removed("vless", "vmess", "ss", "ss2022", "hysteria")...)) } + +const recordingInboundsWithoutUsers = `{"inbounds":[ + {"tag":"vless","protocol":"vless","settings":{"clients":[],"decryption":"none"}}, + {"tag":"tunnel","protocol":"dokodemo-door","settings":{"address":"127.0.0.1","port":80}}, + {"tag":"socks","protocol":"socks","settings":{"auth":"noauth"}}, + {"tag":"http","protocol":"http","settings":{}} +]}` + +func TestCanHoldUsersOnlyExcludesProtocolsWithoutUsers(t *testing.T) { + // Every inbound protocol Xray accepts, plus one it does not know yet. A + // protocol that is not known to lack users counts as holding them, so + // removals reach a protocol the node starts managing without a change here. + for protocol, want := range map[string]bool{ + Vmess: true, + Vless: true, + Trojan: true, + Shadowsocks: true, + Hysteria: true, + Wireguard: true, + "masque": true, + "some-new-protocol": true, + "dokodemo-door": false, + "tunnel": false, + "socks": false, + "mixed": false, + "http": false, + "tun": false, + "Dokodemo-Door": false, + "SOCKS": false, + } { + inbound := &Inbound{Protocol: protocol} + if got := inbound.canHoldUsers(); got != want { + t.Errorf("canHoldUsers() for protocol %q = %v, want %v", protocol, got, want) + } + } +} + +func TestUserSyncsSkipInboundsThatCannotHoldUsers(t *testing.T) { + x, service := newRecordingXray(t) + ctx := syncContext(t) + + cfg, err := NewConfig(recordingInboundsWithoutUsers, nil) + if err != nil { + t.Fatal(err) + } + // ApplyAPI adds the node's own API_INBOUND (dokodemo-door). + if err := cfg.ApplyAPI(10085, 10086); err != nil { + t.Fatal(err) + } + x.config = cfg + + alice := recordingUser("1.alice", "vless") + bob := recordingUser("2.bob") + if err := x.UpdateUsers(ctx, []*common.User{alice, bob}); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations("vless:remove", "vless:add", "vless:remove")) + + if err := x.SyncUser(ctx, bob); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations("vless:remove")) +} diff --git a/backend/xray/xray.go b/backend/xray/xray.go index 89b10ddb..182f94a5 100644 --- a/backend/xray/xray.go +++ b/backend/xray/xray.go @@ -2,6 +2,8 @@ package xray import ( "context" + "errors" + "fmt" "log" "path/filepath" "sync" @@ -12,6 +14,8 @@ import ( "github.com/pasarguard/node/config" ) +// Lock order: syncMu, then mu, then core.mu. syncMu serializes user syncs and +// restarts; Shutdown takes only mu, so it never waits for a sync. type Xray struct { config *Config cfg *config.Config @@ -21,8 +25,15 @@ type Xray struct { cancelFunc context.CancelFunc mu sync.RWMutex syncMu sync.Mutex + closed bool // set by Shutdown under mu } +var errXrayShutdown = errors.New("xray backend is shut down") + +// restartReadyTimeout bounds how long a restart holds syncMu waiting for the +// new core's API. +const restartReadyTimeout = 30 * time.Second + func New(ctx context.Context, xrayConfig *Config, users []*common.User, apiPort, metricPort int, cfg *config.Config) (*Xray, error) { executableAbsolutePath, err := filepath.Abs(cfg.XrayExecutablePath) if err != nil { @@ -130,13 +141,41 @@ func (x *Xray) Started() bool { return x.core.Started() } +// Restart restarts the core from the current config and waits until its API +// answers. func (x *Xray) Restart() error { - return x.restartCoreWithConfig(x.config) + x.syncMu.Lock() + defer x.syncMu.Unlock() + + return x.restartLocked(context.Background()) +} + +// restartLocked restarts the core from x.config and waits until its API +// answers. The caller holds syncMu for the whole restart, so a user sync sent +// meanwhile waits for the new core. Against a core that is still booting, a +// removal fails, while the new core, built from x.config before the removal, +// keeps the user. +func (x *Xray) restartLocked(ctx context.Context) error { + if err := x.restartCoreWithConfig(x.config); err != nil { + return err + } + + ctx, cancel := context.WithTimeout(ctx, restartReadyTimeout) + defer cancel() + if err := x.checkXrayStatus(ctx); err != nil { + return fmt.Errorf("xray API not ready within %s after restart: %w", restartReadyTimeout, err) + } + return nil } func (x *Xray) restartCoreWithConfig(config *Config) error { x.mu.Lock() defer x.mu.Unlock() + // Shutdown sets closed under mu, so a restart that waited for mu (a full-sync + // rollback, a health restart) cannot start a core after Shutdown returned. + if x.closed { + return errXrayShutdown + } if err := x.core.Restart(config, x.cfg.Debug); err != nil { return err } @@ -152,6 +191,7 @@ func (x *Xray) setConfig(config *Config) { func (x *Xray) Shutdown() { x.mu.Lock() defer x.mu.Unlock() + x.closed = true // Cancel context first to stop health checks and other goroutines x.cancelFunc()