From f105132866679cb6cbbb525a6f14ada54f263b64 Mon Sep 17 00:00:00 2001 From: 25f2002654 Date: Sat, 24 Oct 2026 23:00:09 +0530 Subject: [PATCH] perf(runtime): pool the local xray gRPC connection (#6916) A client per runtime operation made a bulk delete pay a TCP and HTTP/2 handshake for every client. The local runtime now keeps one gRPC client per API port and replaces it when a core restart moves the port; the traffic poll keeps its own client, since StatsLastValues is unlocked. A pooled channel keeps grpc-go's reconnect backoff after the core restarts, so a call that fails with Unavailable drops the client and dials again, as a client per call did. --- internal/web/runtime/local.go | 21 ++++- internal/web/runtime/local_api.go | 52 +++++++++++ internal/web/runtime/local_api_test.go | 117 +++++++++++++++++++++++++ 3 files changed, 186 insertions(+), 4 deletions(-) create mode 100644 internal/web/runtime/local_api.go create mode 100644 internal/web/runtime/local_api_test.go diff --git a/internal/web/runtime/local.go b/internal/web/runtime/local.go index fc578101e..89746e58e 100644 --- a/internal/web/runtime/local.go +++ b/internal/web/runtime/local.go @@ -8,6 +8,9 @@ import ( "strings" "sync" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "github.com/mhsanaei/3x-ui/v3/internal/amneziawg" "github.com/mhsanaei/3x-ui/v3/internal/amneziawgnet" "github.com/mhsanaei/3x-ui/v3/internal/database/model" @@ -41,12 +44,22 @@ func (l *Local) withAPI(fn func(api *xray.XrayAPI) error) error { if port <= 0 { return errors.New("local xray is not running") } - var api xray.XrayAPI - if err := api.Init(port); err != nil { + api, err := localAPIForPort(port) + if err != nil { return err } - defer api.Close() - return fn(&api) + err = fn(api) + if status.Code(err) != codes.Unavailable { + return err + } + // A pooled channel keeps grpc-go's reconnect backoff after a core restart and fails fast for + // seconds; drop it and dial again, as a client per call did. + evictLocalAPIClient(api) + api, err = localAPIForPort(port) + if err != nil { + return err + } + return fn(api) } func (l *Local) AddInbound(_ context.Context, ib *model.Inbound) error { diff --git a/internal/web/runtime/local_api.go b/internal/web/runtime/local_api.go new file mode 100644 index 000000000..2a3c40d90 --- /dev/null +++ b/internal/web/runtime/local_api.go @@ -0,0 +1,52 @@ +package runtime + +import ( + "sync" + + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +// One pooled gRPC client serves local runtime operations; never use it for traffic stats, whose +// unlocked StatsLastValues a second reader would corrupt, so the traffic poll keeps its own client. +var ( + localAPIMu sync.Mutex + localAPIPort int + localAPIClient *xray.XrayAPI +) + +// localAPIForPort returns the pooled client for the API port, replacing (and closing) it when a core +// restart moved the port; callers must not close it. +func localAPIForPort(port int) (*xray.XrayAPI, error) { + localAPIMu.Lock() + defer localAPIMu.Unlock() + + if localAPIClient != nil { + if localAPIPort == port { + return localAPIClient, nil + } + localAPIClient.Close() + localAPIClient = nil + } + + api := &xray.XrayAPI{} + if err := api.Init(port); err != nil { + return nil, err + } + localAPIClient = api + localAPIPort = port + return api, nil +} + +// evictLocalAPIClient closes the pooled client so the next call dials a fresh connection, unless +// another call already replaced the one that failed. +func evictLocalAPIClient(api *xray.XrayAPI) { + localAPIMu.Lock() + defer localAPIMu.Unlock() + + if localAPIClient != api { + return + } + localAPIClient.Close() + localAPIClient = nil + localAPIPort = 0 +} diff --git a/internal/web/runtime/local_api_test.go b/internal/web/runtime/local_api_test.go new file mode 100644 index 000000000..f52b3bf77 --- /dev/null +++ b/internal/web/runtime/local_api_test.go @@ -0,0 +1,117 @@ +package runtime + +import ( + "context" + "net" + "strconv" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/xtls/xray-core/app/proxyman/command" + + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +// A client per operation made a bulk delete pay a TCP + HTTP/2 handshake per client, so the local +// runtime's client must stay open and be reused between calls. +func TestLocalWithAPIKeepsTheClientOpen(t *testing.T) { + port := 63211 + local := NewLocal(LocalDeps{APIPort: func() int { return port }}) + + var api *xray.XrayAPI + if err := local.withAPI(func(a *xray.XrayAPI) error { api = a; return nil }); err != nil { + t.Fatalf("withAPI: %v", err) + } + if api.HandlerServiceClient == nil { + t.Fatal("the client was closed when the call returned; a pooled client must stay open") + } +} + +func TestLocalWithAPIReusesOneClient(t *testing.T) { + port := 63212 + local := NewLocal(LocalDeps{APIPort: func() int { return port }}) + + clients := make([]*xray.XrayAPI, 0, 3) + for range 3 { + if err := local.withAPI(func(a *xray.XrayAPI) error { + clients = append(clients, a) + return nil + }); err != nil { + t.Fatalf("withAPI: %v", err) + } + } + for i, c := range clients { + if c != clients[0] { + t.Fatalf("call %d got a different client; the local gRPC connection must be pooled", i) + } + } +} + +// A core restart can move the API port, so the pool has to follow it and close +// the client it replaces instead of leaking it. +func TestLocalWithAPIEvictsTheClientWhenThePortChanges(t *testing.T) { + port := 63213 + local := NewLocal(LocalDeps{APIPort: func() int { return port }}) + + var first *xray.XrayAPI + if err := local.withAPI(func(a *xray.XrayAPI) error { first = a; return nil }); err != nil { + t.Fatalf("withAPI: %v", err) + } + port = 63214 + var second *xray.XrayAPI + if err := local.withAPI(func(a *xray.XrayAPI) error { second = a; return nil }); err != nil { + t.Fatalf("withAPI after the port change: %v", err) + } + if first == second { + t.Fatal("a changed API port must not reuse the old client") + } + if first.HandlerServiceClient != nil { + t.Fatal("the evicted client must be closed, not leaked") + } +} + +type fakeHandlerServer struct { + command.UnimplementedHandlerServiceServer +} + +func (fakeHandlerServer) RemoveInbound(context.Context, *command.RemoveInboundRequest) (*command.RemoveInboundResponse, error) { + return &command.RemoveInboundResponse{}, nil +} + +// startFakeCore serves the core's handler service on port (0 picks a free one). +func startFakeCore(t *testing.T, port int) (int, func()) { + t.Helper() + listener, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port))) + if err != nil { + t.Fatalf("listen on %d: %v", port, err) + } + server := grpc.NewServer() + command.RegisterHandlerServiceServer(server, fakeHandlerServer{}) + go func() { _ = server.Serve(listener) }() + return listener.Addr().(*net.TCPAddr).Port, server.Stop +} + +// A pooled channel kept grpc-go's reconnect backoff after a core restart and failed fast for +// seconds; the first call after the restart must succeed, as it did with a client per call. +func TestLocalWithAPIRecoversAfterTheCoreRestarts(t *testing.T) { + port, stop := startFakeCore(t, 0) + local := NewLocal(LocalDeps{APIPort: func() int { return port }}) + delInbound := func(api *xray.XrayAPI) error { return api.DelInbound("tag") } + + if err := local.withAPI(delInbound); err != nil { + t.Fatalf("first call: %v", err) + } + stop() + if err := local.withAPI(delInbound); status.Code(err) != codes.Unavailable { + t.Fatalf("a call while the core is down: got %v, want Unavailable", err) + } + + _, stopRestarted := startFakeCore(t, port) + t.Cleanup(stopRestarted) + if err := local.withAPI(delInbound); err != nil { + t.Fatalf("the first call after the core restarted must succeed: %v", err) + } +}