From 3cdb7e2b49e153d06feaccbae00793ddacb0be33 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sat, 31 Oct 2026 08:46:13 +0000 Subject: [PATCH] fix(traffic): keep client accounting independent of outbound counters The traffic poll read every Xray counter in one QueryStats reply under gRPC's default 4 MiB receive limit; past it the whole poll failed and no client was counted, so quotas stopped being enforced. Outbound counters carry tags derived from third-party link remarks, so one long remark was enough. Poll the user, inbound and outbound families separately with a 64 MiB limit, keep counting the families that answer, and cap a subscription link's slug at 64 runes so the tag stays small to begin with. --- internal/util/link/outbound.go | 6 ++ internal/util/link/outbound_test.go | 4 ++ internal/xray/api.go | 86 ++++++++++++++++++++--------- internal/xray/api_traffic_test.go | 59 ++++++++++++++++++-- 4 files changed, 123 insertions(+), 32 deletions(-) diff --git a/internal/util/link/outbound.go b/internal/util/link/outbound.go index aa6258bff..98c924fb8 100644 --- a/internal/util/link/outbound.go +++ b/internal/util/link/outbound.go @@ -1255,6 +1255,8 @@ func base64DecodeFlexible(s string) (string, error) { // replacing every other run of characters with a single dash. var slugRe = regexp.MustCompile(`[^\p{L}\p{N}]+`) +const maxSlugRunes = 64 + func SlugRemark(remark string) string { s := strings.ToLower(strings.TrimSpace(remark)) s = slugRe.ReplaceAllString(s, "-") @@ -1266,6 +1268,10 @@ func SlugRemark(remark string) string { for strings.Contains(s, "--") { s = strings.ReplaceAll(s, "--", "-") } + // The tag names Xray counters, a traffic row and the config; a link author picks the remark. + if r := []rune(s); len(r) > maxSlugRunes { + s = strings.TrimRight(string(r[:maxSlugRunes]), "-") + } return s } diff --git a/internal/util/link/outbound_test.go b/internal/util/link/outbound_test.go index 9de81beb4..d93fa943b 100644 --- a/internal/util/link/outbound_test.go +++ b/internal/util/link/outbound_test.go @@ -533,6 +533,10 @@ func TestSlugAndSuggest(t *testing.T) { if got := SuggestTag("ru-", "Сервер 2", 0); got != "ru-сервер-2" { t.Errorf("unicode suggest tag got %q", got) } + // The tag names Xray counters and an outbound_traffics row; a provider picks the remark. + if got := SlugRemark(strings.Repeat("Ab ", 1<<20)); got != strings.Repeat("ab-", 21)+"a" { + t.Errorf("megabyte remark slug = %.80q (%d bytes), want the first 64 runes", got, len(got)) + } } // The obfs-local plugin the panel exports carries the only description of diff --git a/internal/xray/api.go b/internal/xray/api.go index 2a38b137e..a8d00abc5 100644 --- a/internal/xray/api.go +++ b/internal/xray/api.go @@ -7,6 +7,7 @@ import ( "context" "encoding/base64" "encoding/json" + "errors" "fmt" "math" "net" @@ -54,6 +55,14 @@ var ( clientTrafficRegex = regexp.MustCompile(`user>>>([^>]+)>>>traffic>>>(downlink|uplink)`) ) +// statsFamilies are polled one query each: outbound counters are named after third-party +// subscription tags, and their reply must not take the clients' counters down with it. +var statsFamilies = []string{"user>>>", "inbound>>>", "outbound>>>"} + +// maxStatsReplyBytes replaces gRPC's 4 MiB receive default, which a stats reply of about +// 45k clients reaches alone; the peer is the panel's own core on loopback. +const maxStatsReplyBytes = 64 << 20 + // XrayAPI is a gRPC client for managing Xray core configuration, inbounds, outbounds, and statistics. type XrayAPI struct { HandlerServiceClient *command.HandlerServiceClient @@ -62,6 +71,7 @@ type XrayAPI struct { grpcClient *grpc.ClientConn isConnected bool StatsLastValues map[string]int64 + statsBaselined map[string]bool } func getRequiredUserString(user map[string]any, key string) (string, error) { @@ -99,7 +109,10 @@ func (x *XrayAPI) Init(apiPort int) error { } addr := fmt.Sprintf("127.0.0.1:%d", apiPort) - conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + conn, err := grpc.NewClient(addr, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxStatsReplyBytes)), + ) if err != nil { return fmt.Errorf("failed to connect to Xray API: %w", err) } @@ -108,6 +121,7 @@ func (x *XrayAPI) Init(apiPort int) error { x.isConnected = true if x.StatsLastValues == nil { x.StatsLastValues = make(map[string]int64) + x.statsBaselined = make(map[string]bool, len(statsFamilies)) } hsClient := command.NewHandlerServiceClient(conn) @@ -975,41 +989,61 @@ func (x *XrayAPI) GetTraffic() ([]*Traffic, []*ClientTraffic, error) { return nil, nil, common.NewError("xray StatusServiceClient is not initialized") } - resp, err := (*x.StatsServiceClient).QueryStats(ctx, &statsService.QueryStatsRequest{Reset_: false}) - if err != nil { + tagTrafficMap := make(map[string]*Traffic) + emailTrafficMap := make(map[string]*ClientTraffic) + if x.statsBaselined == nil { + x.statsBaselined = make(map[string]bool, len(statsFamilies)) + } + + var live []*statsService.Stat + var failed []error + for _, family := range statsFamilies { + resp, err := (*x.StatsServiceClient).QueryStats(ctx, &statsService.QueryStatsRequest{Pattern: family, Reset_: false}) + if err != nil { + failed = append(failed, fmt.Errorf("%s: %w", family, err)) + continue + } + baselinePass := !x.statsBaselined[family] + x.statsBaselined[family] = true + for _, stat := range resp.GetStat() { + // The core matches Pattern anywhere in the name; the prefix keeps it to one family. + if !strings.HasPrefix(stat.Name, family) { + continue + } + live = append(live, stat) + lastValue, ok := x.StatsLastValues[stat.Name] + x.StatsLastValues[stat.Name] = stat.Value + if baselinePass { + continue + } + if !ok || stat.Value < lastValue { + lastValue = 0 + } + value := stat.Value - lastValue + if matches := trafficRegex.FindStringSubmatch(stat.Name); len(matches) == 4 { + processTraffic(matches, value, tagTrafficMap) + } else if matches := clientTrafficRegex.FindStringSubmatch(stat.Name); len(matches) == 3 { + processClientTraffic(matches, value, emailTrafficMap) + } + } + } + if len(failed) == len(statsFamilies) { + err := errors.Join(failed...) logger.Debug("Failed to query Xray stats:", err) return nil, nil, err } - - tagTrafficMap := make(map[string]*Traffic) - emailTrafficMap := make(map[string]*ClientTraffic) - - baselinePass := len(x.StatsLastValues) == 0 - - for _, stat := range resp.GetStat() { - lastValue, ok := x.StatsLastValues[stat.Name] - x.StatsLastValues[stat.Name] = stat.Value - if baselinePass { - continue - } - if !ok || stat.Value < lastValue { - lastValue = 0 - } - value := stat.Value - lastValue - if matches := trafficRegex.FindStringSubmatch(stat.Name); len(matches) == 4 { - processTraffic(matches, value, tagTrafficMap) - } else if matches := clientTrafficRegex.FindStringSubmatch(stat.Name); len(matches) == 3 { - processClientTraffic(matches, value, emailTrafficMap) - } + for _, err := range failed { + logger.Warning("Failed to query Xray stats, counting the other families:", err) } // Drop delta baselines for stats that no longer exist (deleted inbounds or // clients), which otherwise linger until the next Xray restart. Only rebuild // when the map has drifted past 2x the live set, so the steady-state hot path // stays allocation-free. - if n := len(resp.GetStat()); n > 0 && len(x.StatsLastValues) > 2*n { + // A family that failed is not gone, so its baselines stay. + if n := len(live); n > 0 && len(failed) == 0 && len(x.StatsLastValues) > 2*n { pruned := make(map[string]int64, n) - for _, stat := range resp.GetStat() { + for _, stat := range live { pruned[stat.Name] = x.StatsLastValues[stat.Name] } x.StatsLastValues = pruned diff --git a/internal/xray/api_traffic_test.go b/internal/xray/api_traffic_test.go index 8682356be..937ebf316 100644 --- a/internal/xray/api_traffic_test.go +++ b/internal/xray/api_traffic_test.go @@ -3,10 +3,13 @@ package xray import ( "context" "net" + "strings" "testing" statsService "github.com/xtls/xray-core/app/stats/command" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // fakeStatsServer serves a scripted sequence of QueryStats responses, one per @@ -14,16 +17,28 @@ import ( type fakeStatsServer struct { statsService.UnimplementedStatsServiceServer rounds [][]*statsService.Stat - calls int + calls map[string]int + fail map[string]error } -func (f *fakeStatsServer) QueryStats(context.Context, *statsService.QueryStatsRequest) (*statsService.QueryStatsResponse, error) { - round := f.calls - f.calls++ +// QueryStats answers each pattern from its own round counter and, like the core, matches +// the pattern anywhere in the name. +func (f *fakeStatsServer) QueryStats(_ context.Context, req *statsService.QueryStatsRequest) (*statsService.QueryStatsResponse, error) { + if err := f.fail[req.Pattern]; err != nil { + return nil, err + } + round := f.calls[req.Pattern] + f.calls[req.Pattern]++ if round >= len(f.rounds) { round = len(f.rounds) - 1 } - return &statsService.QueryStatsResponse{Stat: f.rounds[round]}, nil + var matched []*statsService.Stat + for _, s := range f.rounds[round] { + if strings.Contains(s.Name, req.Pattern) { + matched = append(matched, s) + } + } + return &statsService.QueryStatsResponse{Stat: matched}, nil } func stat(name string, value int64) *statsService.Stat { @@ -34,12 +49,18 @@ func stat(name string, value int64) *statsService.Stat { // to it. func startFakeStats(t *testing.T, rounds [][]*statsService.Stat) *XrayAPI { t.Helper() + return startFakeStatsServer(t, &fakeStatsServer{rounds: rounds}) +} + +func startFakeStatsServer(t *testing.T, fake *fakeStatsServer) *XrayAPI { + t.Helper() + fake.calls = map[string]int{} lis, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) } srv := grpc.NewServer() - statsService.RegisterStatsServiceServer(srv, &fakeStatsServer{rounds: rounds}) + statsService.RegisterStatsServiceServer(srv, fake) go func() { _ = srv.Serve(lis) }() t.Cleanup(srv.Stop) @@ -190,3 +211,29 @@ func TestGetTrafficSkipsAPIInboundAndPrunes(t *testing.T) { t.Fatal("baselines for stats that no longer exist were not pruned") } } + +// Outbound counters are named after subscription tags a third party chooses; however +// large that family's reply grows, the clients' counters must still be counted. +func TestGetTrafficCountsClientsPastAnOversizedOutboundFamily(t *testing.T) { + huge := stat("outbound>>>"+strings.Repeat("x", 5<<20)+">>>traffic>>>uplink", 1) + rounds := [][]*statsService.Stat{ + {stat("user>>>alice>>>traffic>>>uplink", 100), huge}, + {stat("user>>>alice>>>traffic>>>uplink", 400), huge}, + } + for name, fake := range map[string]*fakeStatsServer{ + "reply over gRPC's default limit": {rounds: rounds}, + "family that fails": {rounds: rounds, fail: map[string]error{"outbound>>>": status.Error(codes.ResourceExhausted, "too large")}}, + } { + api := startFakeStatsServer(t, fake) + if _, _, err := api.GetTraffic(); err != nil { + t.Fatalf("%s: GetTraffic (baseline): %v", name, err) + } + _, clients, err := api.GetTraffic() + if err != nil { + t.Fatalf("%s: GetTraffic: %v", name, err) + } + if alice := clientTrafficByEmail(t, clients)["alice"]; alice == nil || alice.Up != 300 { + t.Fatalf("%s: alice = %+v, want 300 bytes up", name, alice) + } + } +}