fix: refactor the InboundInfo struct

Global limit can separate settings for each node

.github/workflows/codeql-analysis.yml

@@ -35,7 +35,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
 
       # Initializes the CodeQL tools for scanning.
       - name: Initialize CodeQL

.github/workflows/docker.yml

@@ -20,9 +20,9 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Check out the repo
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v1
+        uses: docker/setup-buildx-action@v2
       - name: Log in to the Container registry
         uses: docker/login-action@v2
         with:
@@ -35,7 +35,7 @@ jobs:
         with:
           images: ghcr.io/${{ github.repository }}
       - name: Build and push
-        uses: docker/build-push-action@v2
+        uses: docker/build-push-action@v3
         with:
           context: .
           platforms: linux/arm/v7,linux/arm64,linux/amd64,linux/s390x

.github/workflows/release.yml

@@ -101,17 +101,17 @@ jobs:
       CGO_ENABLED: 0
     steps:
       - name: Checkout codebase
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
       - name: Show workflow information
         id: get_filename
         run: |
           export _NAME=$(jq ".[\"$GOOS-$GOARCH$GOARM$GOMIPS\"].friendlyName" -r < .github/build/friendly-filenames.json)
           echo "GOOS: $GOOS, GOARCH: $GOARCH, GOARM: $GOARM, GOMIPS: $GOMIPS, RELEASE_NAME: $_NAME"
-          echo "::set-output name=ASSET_NAME::$_NAME"
+          echo "ASSET_NAME=$_NAME" >> $GITHUB_OUTPUT
           echo "ASSET_NAME=$_NAME" >> $GITHUB_ENV
 
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v3
         with:
           go-version: ^1.19
 
@@ -173,7 +173,7 @@ jobs:
         run: |
           mv build_assets XrayR-$ASSET_NAME
       - name: Upload files to Artifacts
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v3
         with:
           name: XrayR-${{ steps.get_filename.outputs.ASSET_NAME }}
           path: |

common/limiter/limiter.go

@@ -25,12 +25,16 @@ type InboundInfo struct {
 	UserInfo       *sync.Map // Key: Email value: UserInfo
 	BucketHub      *sync.Map // key: Email, value: *rate.Limiter
 	UserOnlineIP   *sync.Map // Key: Email Value: *sync.Map: Key: IP, Value: UID
-	GlobalOnlineIP *sync.Map // Key: Email Value: *sync.Map: Key: IP, Value: UID
+	GlobalLimit    *GlobalLimit
 }
 
 type Limiter struct {
 	InboundInfo *sync.Map // Key: Tag, Value: *InboundInfo
-	g           *GlobalDeviceLimitConfig
+}
+
+type GlobalLimit struct {
+	*GlobalDeviceLimitConfig
+	OnlineIP *sync.Map // Key: Email Value: *sync.Map: Key: IP, Value: UID
 }
 
 func New() *Limiter {
@@ -39,19 +43,18 @@ func New() *Limiter {
 	}
 }
 
-func (l *Limiter) AddInboundLimiter(tag string, nodeSpeedLimit uint64, userList *[]api.UserInfo, globalDeviceLimit *GlobalDeviceLimitConfig) error {
+func (l *Limiter) AddInboundLimiter(tag string, nodeSpeedLimit uint64, userList *[]api.UserInfo, globalLimit *GlobalDeviceLimitConfig) error {
-	// global limit
-	if globalDeviceLimit.Enable {
-		l.g = globalDeviceLimit
-	}
-
 	inboundInfo := &InboundInfo{
 		Tag:            tag,
 		NodeSpeedLimit: nodeSpeedLimit,
 		BucketHub:      new(sync.Map),
 		UserOnlineIP:   new(sync.Map),
-		GlobalOnlineIP: new(sync.Map),
+		GlobalLimit: &GlobalLimit{
+			GlobalDeviceLimitConfig: globalLimit,
+			OnlineIP:                new(sync.Map),
+		},
 	}
+
 	userMap := new(sync.Map)
 	for _, u := range *userList {
 		userMap.Store(fmt.Sprintf("%s|%s|%d", tag, u.Email, u.UID), UserInfo{
@@ -132,13 +135,14 @@ func (l *Limiter) GetOnlineDevice(tag string) (*[]api.OnlineUser, error) {
 
 func (l *Limiter) GetUserBucket(tag string, email string, ip string) (limiter *rate.Limiter, SpeedLimit bool, Reject bool) {
 	if value, ok := l.InboundInfo.Load(tag); ok {
-		inboundInfo := value.(*InboundInfo)
-		nodeLimit := inboundInfo.NodeSpeedLimit
 		var (
 			userLimit        uint64 = 0
 			deviceLimit, uid int
 		)
 
+		inboundInfo := value.(*InboundInfo)
+		nodeLimit := inboundInfo.NodeSpeedLimit
+
 		if v, ok := inboundInfo.UserInfo.Load(email); ok {
 			u := v.(UserInfo)
 			uid = u.UID
@@ -167,10 +171,10 @@ func (l *Limiter) GetUserBucket(tag string, email string, ip string) (limiter *r
 		}
 
 		// Global device limit
-		if l.g.Enable {
+		if inboundInfo.GlobalLimit.Enable {
 			email := email[strings.Index(email, "|")+1:]
 
-			if v, ok := inboundInfo.GlobalOnlineIP.Load(email); ok {
+			if v, ok := inboundInfo.GlobalLimit.OnlineIP.Load(email); ok {
 				ipMap := v.(*sync.Map)
 				// If this is a new ip
 				if _, ok := ipMap.LoadOrStore(ip, uid); !ok {
@@ -183,10 +187,10 @@ func (l *Limiter) GetUserBucket(tag string, email string, ip string) (limiter *r
 						ipMap.Delete(ip)
 						return nil, false, true
 					}
-					go l.pushIP(email, ip)
+					go pushIP(email, ip, inboundInfo.GlobalLimit)
 				}
 			} else {
-				go l.pushIP(email, ip)
+				go pushIP(email, ip, inboundInfo.GlobalLimit)
 			}
 		}
 
@@ -210,17 +214,17 @@ func (l *Limiter) GetUserBucket(tag string, email string, ip string) (limiter *r
 }
 
 // Push new IP to redis
-func (l *Limiter) pushIP(email string, ip string) {
+func pushIP(email string, ip string, g *GlobalLimit) {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*time.Duration(l.g.Timeout))
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*time.Duration(g.Timeout))
 	defer cancel()
 
-	if err := l.g.R.HSet(ctx, email, map[string]any{ip: 0}).Err(); err != nil {
+	if err := g.R.SAdd(ctx, email, ip).Err(); err != nil {
 		newError(fmt.Errorf("redis: %v", err)).AtError().WriteToLog()
 	}
 
 	// check ttl, if ttl == -1, then set expire time.
-	if l.g.R.TTL(ctx, email).Val() == -1 {
+	if g.R.TTL(ctx, email).Val() == -1 {
-		if err := l.g.R.Expire(ctx, email, time.Duration(l.g.Expiry)*time.Minute).Err(); err != nil {
+		if err := g.R.Expire(ctx, email, time.Duration(g.Expiry)*time.Minute).Err(); err != nil {
 			newError(fmt.Errorf("redis: %v", err)).AtError().WriteToLog()
 		}
 	}

main/config.yml.example

@@ -12,13 +12,6 @@ ConnectionConfig:
   UplinkOnly: 2 # Time limit when the connection downstream is closed, Second
   DownlinkOnly: 4 # Time limit when the connection is closed after the uplink is closed, Second
   BufferSize: 64 # The internal cache size of each connection, kB
-GlobalDeviceLimitConfig:
-  Enable: true # Enable the global device limit of a user
-  RedisAddr: 10.0.0.188:6379 # The redis server address
-  RedisPassword:  # Redis password
-  RedisDB: 0 # Redis DB
-  Timeout: 5 # Timeout for redis request
-  Expiry: 60 # Expiry time (second)
 Nodes:
   -
     PanelType: "SSpanel" # Panel type: SSpanel, V2board, NewV2board, PMpanel, Proxypanel, V2RaySocks
@@ -45,6 +38,13 @@ Nodes:
         WarnTimes: 0 # After (WarnTimes) consecutive warnings, the user will be limited. Set to 0 to punish overspeed user immediately.
         LimitSpeed: 0 # The speedlimit of a limited user (unit: mbps)
         LimitDuration: 0 # How many minutes will the limiting last (unit: minute)
+      GlobalDeviceLimitConfig:
+        Enable: false # Enable the global device limit of a user
+        RedisAddr: 127.0.0.1:6379 # The redis server address
+        RedisPassword: YOUR PASSWORD # Redis password
+        RedisDB: 0 # Redis DB
+        Timeout: 5 # Timeout for redis request
+        Expiry: 60 # Expiry time (second)
       EnableFallback: false # Only support for Trojan and Vless
       FallBackConfigs:  # Support multiple fallbacks
         -

panel/config.go

@@ -2,7 +2,6 @@ package panel
 
 import (
 	"github.com/XrayR-project/XrayR/api"
-	"github.com/XrayR-project/XrayR/common/limiter"
 	"github.com/XrayR-project/XrayR/service/controller"
 )
 
@@ -13,7 +12,6 @@ type Config struct {
 	OutboundConfigPath string            `mapstructure:"OutboundConfigPath"`
 	RouteConfigPath    string            `mapstructure:"RouteConfigPath"`
 	ConnectionConfig   *ConnectionConfig `mapstructure:"ConnectionConfig"`
-	GlobalDeviceLimitConfig *limiter.GlobalDeviceLimitConfig `mapstructure:"GlobalDeviceLimitConfig"`
 	NodesConfig        []*NodesConfig    `mapstructure:"Nodes"`
 }
 

panel/panel.go

@@ -164,10 +164,6 @@ func (p *Panel) Start() {
 	}
 	p.Server = server
 
-	if p.panelConfig.GlobalDeviceLimitConfig.Enable {
-		log.Println("Global limit: Enable")
-	}
-
 	// Load Nodes config
 	for _, nodeConfig := range p.panelConfig.NodesConfig {
 		var apiClient api.API
@@ -196,7 +192,7 @@ func (p *Panel) Start() {
 				log.Panicf("Read Controller Config Failed")
 			}
 		}
-		controllerService = controller.New(server, apiClient, controllerConfig, nodeConfig.PanelType, p.panelConfig.GlobalDeviceLimitConfig)
+		controllerService = controller.New(server, apiClient, controllerConfig, nodeConfig.PanelType)
 		p.Service = append(p.Service, controllerService)
 
 	}

service/controller/config.go

@@ -1,6 +1,7 @@
 package controller
 
 import (
+	"github.com/XrayR-project/XrayR/common/limiter"
 	"github.com/XrayR-project/XrayR/common/mylego"
 )
 
@@ -18,6 +19,7 @@ type Config struct {
 	DisableIVCheck          bool                             `mapstructure:"DisableIVCheck"`
 	DisableSniffing         bool                             `mapstructure:"DisableSniffing"`
 	AutoSpeedLimitConfig    *AutoSpeedLimitConfig            `mapstructure:"AutoSpeedLimitConfig"`
+	GlobalDeviceLimitConfig *limiter.GlobalDeviceLimitConfig `mapstructure:"GlobalDeviceLimitConfig"`
 	FallBackConfigs         []*FallBackConfig                `mapstructure:"FallBackConfigs"`
 }
 

service/controller/controller.go

@@ -47,7 +47,6 @@ type Controller struct {
 	stm          stats.Manager
 	dispatcher   *mydispatcher.DefaultDispatcher
 	startAt      time.Time
-	g            *limiter.GlobalDeviceLimitConfig
 }
 
 type periodicTask struct {
@@ -56,7 +55,7 @@ type periodicTask struct {
 }
 
 // New return a Controller service with default parameters.
-func New(server *core.Instance, api api.API, config *Config, panelType string, globalConfig *limiter.GlobalDeviceLimitConfig) *Controller {
+func New(server *core.Instance, api api.API, config *Config, panelType string) *Controller {
 	controller := &Controller{
 		server:     server,
 		config:     config,
@@ -69,16 +68,6 @@ func New(server *core.Instance, api api.API, config *Config, panelType string, g
 		startAt:    time.Now(),
 	}
 
-	// Init global limit redis client
-	if globalConfig.Enable {
-		globalConfig.R = redis.NewClient(&redis.Options{
-			Addr:     globalConfig.RedisAddr,
-			Password: globalConfig.RedisPassword,
-			DB:       globalConfig.RedisDB,
-		})
-		controller.g = globalConfig
-	}
-
 	return controller
 }
 
@@ -113,7 +102,7 @@ func (c *Controller) Start() error {
 	}
 
 	// Add Limiter
-	if err := c.AddInboundLimiter(c.Tag, newNodeInfo.SpeedLimit, userInfo, c.g); err != nil {
+	if err := c.AddInboundLimiter(c.Tag, newNodeInfo.SpeedLimit, userInfo, c.initGlobal()); err != nil {
 		log.Print(err)
 	}
 
@@ -164,7 +153,7 @@ func (c *Controller) Start() error {
 	}
 
 	// Check global limit in need
-	if c.g.Enable {
+	if c.config.GlobalDeviceLimitConfig.Enable {
 		c.tasks = append(c.tasks,
 			periodicTask{
 				tag: "global limit",
@@ -273,8 +262,9 @@ func (c *Controller) nodeInfoMonitor() (err error) {
 			log.Print(err)
 			return nil
 		}
+
 		// Add Limiter
-		if err := c.AddInboundLimiter(c.Tag, newNodeInfo.SpeedLimit, newUserInfo, c.g); err != nil {
+		if err := c.AddInboundLimiter(c.Tag, newNodeInfo.SpeedLimit, newUserInfo, c.initGlobal()); err != nil {
 			log.Print(err)
 			return nil
 		}
@@ -309,6 +299,20 @@ func (c *Controller) nodeInfoMonitor() (err error) {
 	return nil
 }
 
+func (c *Controller) initGlobal() *limiter.GlobalDeviceLimitConfig {
+	// Init global limit redis client
+	globalConfig := c.config.GlobalDeviceLimitConfig
+	if c.config.GlobalDeviceLimitConfig.Enable {
+		log.Printf("[%s] Global limit: enable", c.Tag)
+		globalConfig.R = redis.NewClient(&redis.Options{
+			Addr:     globalConfig.RedisAddr,
+			Password: globalConfig.RedisPassword,
+			DB:       globalConfig.RedisDB,
+		})
+	}
+	return globalConfig
+}
+
 func (c *Controller) removeOldTag(oldTag string) (err error) {
 	err = c.removeInbound(oldTag)
 	if err != nil {
@@ -653,27 +657,27 @@ func (c *Controller) globalLimitFetch() (err error) {
 
 		inboundInfo := value.(*limiter.InboundInfo)
 		for {
-			if emails, cursor, err = c.g.R.Scan(ctx, cursor, "*", 10000).Result(); err != nil {
+			if emails, cursor, err = c.config.GlobalDeviceLimitConfig.R.Scan(ctx, cursor, "*", 10000).Result(); err != nil {
 				newError(err).AtError().WriteToLog()
 			}
-			pipe := c.g.R.Pipeline()
+			pipe := c.config.GlobalDeviceLimitConfig.R.Pipeline()
 
-			cmdMap := make(map[string]*redis.StringStringMapCmd)
+			cmdMap := make(map[string]*redis.StringSliceCmd)
 			for i := range emails {
 				email := emails[i]
-				cmdMap[email] = pipe.HGetAll(ctx, email)
+				cmdMap[email] = pipe.SMembers(ctx, email)
 			}
 
 			if _, err := pipe.Exec(ctx); err != nil {
 				newError(fmt.Errorf("redis: %v", err)).AtError().WriteToLog()
 			} else {
-				inboundInfo.GlobalOnlineIP = new(sync.Map)
+				inboundInfo.GlobalLimit.OnlineIP = new(sync.Map)
 				for k := range cmdMap {
 					ips := cmdMap[k].Val()
 					ipMap := new(sync.Map)
 					for i := range ips {
-						ipMap.Store(i, 0)
+						ipMap.Store(ips[i], 0)
-						inboundInfo.GlobalOnlineIP.Store(k, ipMap)
+						inboundInfo.GlobalLimit.OnlineIP.Store(k, ipMap)
 					}
 				}
 			}