feat: implement user and forward max connection limit (#461)

* docs: add implementation plan for max conn limit

* feat: add CLimiters support for websocket reporter

* feat: add max_conn field to user and forward models

* feat: implement max conn limiter dispatching

* feat: add maxConn to user and forward CRUD API

* feat: add max conn UI to user management and forward rules

* fix: load MaxConn in get forward record and add e2e contract test for max conn limit

go-backend/internal/http/handler/control_plane.go

@@ -266,6 +266,22 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method
 
 	serviceBase := buildForwardServiceBaseWithResolvedUserTunnel(forward.ID, forward.UserID, userTunnelID)
 
+	user, err := h.repo.GetUserByID(forward.UserID)
+	if err != nil {
+		return nil, err
+	}
+
+	var cLimiterName string
+	var maxConnToSet int
+
+	if forward.MaxConn > 0 {
+		maxConnToSet = forward.MaxConn
+		cLimiterName = fmt.Sprintf("rule_conn_limit_%d", forward.ID)
+	} else if user != nil && user.MaxConn > 0 {
+		maxConnToSet = user.MaxConn
+		cLimiterName = fmt.Sprintf("user_conn_limit_%d", user.ID)
+	}
+
 	for _, fp := range ports {
 		if limiterID != nil && speed != nil {
 			if err := h.ensureLimiterOnNode(fp.NodeID, *limiterID, *speed); err != nil {
@@ -283,11 +299,17 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method
 			}
 		}
 
+		if cLimiterName != "" {
+			if err := h.ensureConnLimiterOnNode(fp.NodeID, cLimiterName, maxConnToSet); err != nil {
+				warnings = append(warnings, fmt.Sprintf("节点 %d 连接限制器下发失败: %v", fp.NodeID, err))
+			}
+		}
+
 		node, err := h.getNodeRecord(fp.NodeID)
 		if err != nil {
 			return nil, err
 		}
-		services := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, strings.TrimSpace(fp.InIP), limiterID)
+		services := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, strings.TrimSpace(fp.InIP), limiterID, cLimiterName)
 		_, err = h.sendNodeCommand(node.ID, method, services, true, false)
 		if err != nil && allowFallbackAdd && method == "UpdateService" {
 			if isNotFoundError(err) {
@@ -302,7 +324,7 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method
 		}
 		if err != nil && strings.EqualFold(strings.TrimSpace(method), "UpdateService") && isCannotAssignRequestedAddressError(err) {
 			var warning string
-			warning, err = h.fallbackForwardPortToDefaultBind(forward, tunnel, node, fp, serviceBase, limiterID)
+			warning, err = h.fallbackForwardPortToDefaultBind(forward, tunnel, node, fp, serviceBase, limiterID, cLimiterName)
 			if err == nil && warning != "" {
 				warnings = append(warnings, warning)
 			}
@@ -328,7 +350,7 @@ func (h *Handler) syncForwardServicesWithWarnings(forward *forwardRecord, method
 	return warnings, nil
 }
 
-func (h *Handler) fallbackForwardPortToDefaultBind(forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, fp forwardPortRecord, serviceBase string, limiterID *int64) (string, error) {
+func (h *Handler) fallbackForwardPortToDefaultBind(forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, fp forwardPortRecord, serviceBase string, limiterID *int64, cLimiterName string) (string, error) {
 	if h == nil || forward == nil || tunnel == nil || node == nil {
 		return "", errors.New("invalid bind fallback context")
 	}
@@ -345,7 +367,7 @@ func (h *Handler) fallbackForwardPortToDefaultBind(forward *forwardRecord, tunne
 	}
 
 	time.Sleep(150 * time.Millisecond)
-	defaultServices := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, "", limiterID)
+	defaultServices := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, "", limiterID, cLimiterName)
 	if _, err := h.sendNodeCommand(node.ID, "AddService", defaultServices, true, false); err != nil {
 		return "", err
 	}
@@ -1637,7 +1659,7 @@ func compactErrorMessage(msg string) string {
 	return strings.Join(strings.Fields(strings.ToLower(msg)), "")
 }
 
-func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, bindIP string, limiterID *int64) []map[string]interface{} {
+func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, bindIP string, limiterID *int64, cLimiterName string) []map[string]interface{} {
 	protocols := []string{"tcp", "udp"}
 	services := make([]map[string]interface{}, 0, 2)
 	targets := splitRemoteTargets(forward.RemoteAddr)
@@ -1680,6 +1702,9 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel
 				},
 			},
 		}
+		if cLimiterName != "" {
+			service["climiter"] = cLimiterName
+		}
 		if protocol == "udp" {
 			listenerMetadata := map[string]interface{}{
 				"keepAlive": true,
@@ -1795,6 +1820,30 @@ func (h *Handler) ensureLimiterOnNode(nodeID int64, limiterID int64, speed int)
 	return nil
 }
 
+func (h *Handler) ensureConnLimiterOnNode(nodeID int64, limiterName string, maxConn int) error {
+	limitStr := fmt.Sprintf("$ %d", maxConn)
+	
+	payload := map[string]interface{}{
+		"name":   limiterName,
+		"limits": []string{limitStr},
+	}
+	
+	if _, err := h.sendNodeCommand(nodeID, "AddCLimiters", payload, false, false); err != nil {
+		if !isAlreadyExistsMessage(err.Error()) {
+			return fmt.Errorf("连接限制器下发失败: %w", err)
+		}
+		updatePayload := map[string]interface{}{
+			"limiter": limiterName,
+			"data":    payload,
+		}
+		if _, updateErr := h.sendNodeCommand(nodeID, "UpdateCLimiters", updatePayload, false, false); updateErr != nil {
+			return fmt.Errorf("连接限制器更新失败: %w", updateErr)
+		}
+	}
+	return nil
+}
+
+
 func buildLimiterAddPayload(limiterID int64, speed int) (string, map[string]interface{}) {
 	rate := float64(speed) / 8.0
 	limitStr := fmt.Sprintf("$ %.1fMB %.1fMB", rate, rate)

go-backend/internal/http/handler/control_plane_test.go

@@ -378,7 +378,7 @@ func TestRetryTunnelServiceAddWithCleanupReturnsCleanupError(t *testing.T) {
 func TestBuildForwardServiceConfigs_UsesBindIPForListen(t *testing.T) {
 	forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7}
 	node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"}
-	services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22000, "10.9.8.7", nil)
+	services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22000, "10.9.8.7", nil, "")
 	if len(services) != 2 {
 		t.Fatalf("expected 2 services, got %d", len(services))
 	}
@@ -393,7 +393,7 @@ func TestBuildForwardServiceConfigs_UsesBindIPForListen(t *testing.T) {
 func TestBuildForwardServiceConfigs_DefaultListenAddrWhenBindIPEmpty(t *testing.T) {
 	forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7}
 	node := &nodeRecord{TCPListenAddr: "0.0.0.0", UDPListenAddr: "[::]"}
-	services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22001, "", nil)
+	services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 22001, "", nil, "")
 	if len(services) != 2 {
 		t.Fatalf("expected 2 services, got %d", len(services))
 	}
@@ -409,7 +409,7 @@ func TestBuildForwardServiceConfigs_DefaultListenAddrWhenBindIPEmpty(t *testing.
 func TestBuildForwardServiceConfigs_BindIPAlreadyContainsPort(t *testing.T) {
 	forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7}
 	node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"}
-	services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 55555, "3.3.3.3:12345", nil)
+	services := buildForwardServiceConfigs("1_2_0", forward, nil, node, 55555, "3.3.3.3:12345", nil, "")
 	if len(services) != 2 {
 		t.Fatalf("expected 2 services, got %d", len(services))
 	}
@@ -464,7 +464,7 @@ func TestBuildForwardServiceConfigs_IPv6BindIP(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			forward := &forwardRecord{RemoteAddr: "1.2.3.4:80", Strategy: "fifo", TunnelID: 7}
 			node := &nodeRecord{TCPListenAddr: "[::]", UDPListenAddr: "[::]"}
-			services := buildForwardServiceConfigs("1_2_0", forward, nil, node, tt.port, tt.bindIP, nil)
+			services := buildForwardServiceConfigs("1_2_0", forward, nil, node, tt.port, tt.bindIP, nil, "")
 			if len(services) != 2 {
 				t.Fatalf("expected 2 services, got %d", len(services))
 			}

go-backend/internal/http/handler/mutations.go

@@ -68,8 +68,9 @@ func (h *Handler) userCreate(w http.ResponseWriter, r *http.Request) {
 	}
 	roleID := 1
 	now := time.Now().UnixMilli()
+	maxConn := asInt(req["maxConn"], 0)
 
-	userID, err := h.repo.CreateUser(username, security.MD5(pwd), roleID, expTime, flow, flowResetTime, num, status, now)
+	userID, err := h.repo.CreateUser(username, security.MD5(pwd), roleID, expTime, flow, flowResetTime, num, status, maxConn, now)
 	if err != nil {
 		response.WriteJSON(w, response.Err(-2, err.Error()))
 		return
@@ -156,15 +157,16 @@ func (h *Handler) userUpdate(w http.ResponseWriter, r *http.Request) {
 	_, hasDailyQuota := req["dailyQuotaGB"]
 	_, hasMonthlyQuota := req["monthlyQuotaGB"]
 	now := time.Now().UnixMilli()
+	maxConn := asInt(req["maxConn"], 0)
 
 	pwd := asString(req["pwd"])
 	if strings.TrimSpace(pwd) == "" {
-		if err := h.repo.UpdateUserWithoutPassword(id, username, flow, num, expTime, flowResetTime, status, now); err != nil {
+		if err := h.repo.UpdateUserWithoutPassword(id, username, flow, num, expTime, flowResetTime, status, maxConn, now); err != nil {
 			response.WriteJSON(w, response.Err(-2, err.Error()))
 			return
 		}
 	} else {
-		if err := h.repo.UpdateUserWithPassword(id, username, security.MD5(pwd), flow, num, expTime, flowResetTime, status, now); err != nil {
+		if err := h.repo.UpdateUserWithPassword(id, username, security.MD5(pwd), flow, num, expTime, flowResetTime, status, maxConn, now); err != nil {
 			response.WriteJSON(w, response.Err(-2, err.Error()))
 			return
 		}
@@ -1777,7 +1779,8 @@ func (h *Handler) forwardCreate(w http.ResponseWriter, r *http.Request) {
 	if userName == "" {
 		userName = "user"
 	}
-	forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID))
+	maxConn := asInt(req["maxConn"], 0)
+	forwardID, err := h.repo.CreateForwardTx(userID, userName, name, tunnelID, remoteAddr, defaultString(asString(req["strategy"]), "fifo"), now, inx, entryNodes, port, inIp, nullableInt(speedID), maxConn)
 	if err != nil {
 		response.WriteJSON(w, response.Err(-2, err.Error()))
 		return
@@ -1928,7 +1931,9 @@ func (h *Handler) forwardUpdate(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 	now := time.Now().UnixMilli()
-	if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID); err != nil {
+	maxConn := asInt(req["maxConn"], forward.MaxConn)
+
+	if err := h.repo.UpdateForward(id, name, tunnelID, remoteAddr, strategy, now, newSpeedID, maxConn); err != nil {
 		response.WriteJSON(w, response.Err(-2, err.Error()))
 		return
 	}
@@ -4080,7 +4085,7 @@ func (h *Handler) rollbackForwardMutation(oldForward *forwardRecord, oldPorts []
 	h.repo.RollbackForwardFields(
 		oldForward.ID, oldForward.UserID, oldForward.UserName, oldForward.Name,
 		oldForward.TunnelID, oldForward.RemoteAddr, oldForward.Strategy, oldForward.Status,
-		oldForward.SpeedID,
+		oldForward.SpeedID, oldForward.MaxConn,
 		time.Now().UnixMilli(),
 	)
 

go-backend/internal/store/model/model.go

@@ -23,6 +23,7 @@ type User struct {
 	CreatedTime   int64         `gorm:"column:created_time;not null"`
 	UpdatedTime   sql.NullInt64 `gorm:"column:updated_time"`
 	Status        int           `gorm:"not null"`
+	MaxConn       int           `gorm:"column:max_conn;not null;default:0"`
 }
 
 func (User) TableName() string { return "user" }
@@ -43,6 +44,7 @@ type Forward struct {
 	Status      int           `gorm:"not null"`
 	Inx         int           `gorm:"not null;default:0"`
 	SpeedID     sql.NullInt64 `gorm:"column:speed_id"`
+	MaxConn     int           `gorm:"column:max_conn;not null;default:0"`
 }
 
 func (Forward) TableName() string { return "forward" }
@@ -537,6 +539,7 @@ type ForwardRecord struct {
 	Strategy   string
 	Status     int
 	SpeedID    sql.NullInt64
+	MaxConn    int
 }
 
 // TunnelRecord is a minimal tunnel view used by control plane.

go-backend/internal/store/repo/repository_flow.go

@@ -124,15 +124,16 @@ func (r *Repository) GetForwardRecord(forwardID int64) (*model.ForwardRecord, er
 		return nil, err
 	}
 	fr := model.ForwardRecord{
-		ID:         f.ID,
-		UserID:     f.UserID,
-		UserName:   f.UserName,
-		Name:       f.Name,
-		TunnelID:   f.TunnelID,
-		RemoteAddr: f.RemoteAddr,
-		Strategy:   f.Strategy,
-		Status:     f.Status,
-		SpeedID:    f.SpeedID,
+	        ID:         f.ID,
+	        UserID:     f.UserID,
+	        UserName:   f.UserName,
+	        Name:       f.Name,
+	        TunnelID:   f.TunnelID,
+	        RemoteAddr: f.RemoteAddr,
+	        Strategy:   f.Strategy,
+	        Status:     f.Status,
+	        SpeedID:    f.SpeedID,
+	        MaxConn:    f.MaxConn,
 	}
 	if strings.TrimSpace(fr.Strategy) == "" {
 		fr.Strategy = "fifo"

go-backend/internal/store/repo/repository_mutations.go

@@ -37,7 +37,7 @@ func (r *Repository) UserExistsExcluding(username string, excludeID int64) (bool
 	return cnt > 0, err
 }
 
-func (r *Repository) CreateUser(username, pwdHash string, roleID int, expTime, flow, flowResetTime int64, num, status int, now int64) (int64, error) {
+func (r *Repository) CreateUser(username, pwdHash string, roleID int, expTime, flow, flowResetTime int64, num, status, maxConn int, now int64) (int64, error) {
 	if r == nil || r.db == nil {
 		return 0, errors.New("repository not initialized")
 	}
@@ -51,6 +51,7 @@ func (r *Repository) CreateUser(username, pwdHash string, roleID int, expTime, f
 		OutFlow:       0,
 		FlowResetTime: flowResetTime,
 		Num:           num,
+		MaxConn:       maxConn,
 		CreatedTime:   now,
 		UpdatedTime:   sql.NullInt64{Int64: now, Valid: true},
 		Status:        status,
@@ -73,7 +74,7 @@ func (r *Repository) GetUserRoleID(userID int64) (int, error) {
 	return user.RoleID, nil
 }
 
-func (r *Repository) UpdateUserWithPassword(id int64, username, pwdHash string, flow int64, num int, expTime, flowResetTime int64, status int, now int64) error {
+func (r *Repository) UpdateUserWithPassword(id int64, username, pwdHash string, flow int64, num int, expTime, flowResetTime int64, status, maxConn int, now int64) error {
 	if r == nil || r.db == nil {
 		return errors.New("repository not initialized")
 	}
@@ -87,11 +88,12 @@ func (r *Repository) UpdateUserWithPassword(id int64, username, pwdHash string,
 			"exp_time":        expTime,
 			"flow_reset_time": flowResetTime,
 			"status":          status,
+			"max_conn":        maxConn,
 			"updated_time":    sql.NullInt64{Int64: now, Valid: true},
 		}).Error
 }
 
-func (r *Repository) UpdateUserWithoutPassword(id int64, username string, flow int64, num int, expTime, flowResetTime int64, status int, now int64) error {
+func (r *Repository) UpdateUserWithoutPassword(id int64, username string, flow int64, num int, expTime, flowResetTime int64, status, maxConn int, now int64) error {
 	if r == nil || r.db == nil {
 		return errors.New("repository not initialized")
 	}
@@ -104,6 +106,7 @@ func (r *Repository) UpdateUserWithoutPassword(id int64, username string, flow i
 			"exp_time":        expTime,
 			"flow_reset_time": flowResetTime,
 			"status":          status,
+			"max_conn":        maxConn,
 			"updated_time":    sql.NullInt64{Int64: now, Valid: true},
 		}).Error
 }
@@ -692,7 +695,7 @@ func (r *Repository) GetMinForwardPort(forwardID int64) sql.NullInt64 {
 	return p
 }
 
-func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remoteAddr, strategy string, now int64, speedID interface{}) error {
+func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remoteAddr, strategy string, now int64, speedID interface{}, maxConn int) error {
 	if r == nil || r.db == nil {
 		return errors.New("repository not initialized")
 	}
@@ -704,6 +707,7 @@ func (r *Repository) UpdateForward(id int64, name string, tunnelID int64, remote
 			"remote_addr":  remoteAddr,
 			"strategy":     strategy,
 			"speed_id":     nullInt64FromInterface(speedID),
+			"max_conn":     maxConn,
 			"updated_time": now,
 		}).Error
 }
@@ -778,7 +782,7 @@ func (r *Repository) UpdateForwardPortBindIP(forwardID, nodeID int64, port int,
 		Update("in_ip", sql.NullString{String: inIP, Valid: strings.TrimSpace(inIP) != ""}).Error
 }
 
-func (r *Repository) RollbackForwardFields(id, userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, status int, speedID interface{}, now int64) {
+func (r *Repository) RollbackForwardFields(id, userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, status int, speedID interface{}, maxConn int, now int64) {
 	if r == nil || r.db == nil {
 		return
 	}
@@ -793,6 +797,7 @@ func (r *Repository) RollbackForwardFields(id, userID int64, userName, name stri
 			"strategy":     strategy,
 			"status":       status,
 			"speed_id":     nullInt64FromInterface(speedID),
+			"max_conn":     maxConn,
 			"updated_time": now,
 		}).Error
 }
@@ -1253,7 +1258,7 @@ func (r *Repository) EnsureUserTunnelGrant(userID, tunnelID int64) (int64, bool,
 	return ut.ID, true, nil
 }
 
-func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, now int64, inx int, entryNodeIDs []int64, port int, inIp string, speedID interface{}) (int64, error) {
+func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnelID int64, remoteAddr, strategy string, now int64, inx int, entryNodeIDs []int64, port int, inIp string, speedID interface{}, maxConn int) (int64, error) {
 	if r == nil || r.db == nil {
 		return 0, errors.New("repository not initialized")
 	}
@@ -1272,6 +1277,7 @@ func (r *Repository) CreateForwardTx(userID int64, userName, name string, tunnel
 			UpdatedTime: now,
 			Status:      1,
 			Inx:         inx,
+			MaxConn:     maxConn,
 			SpeedID:     nullInt64FromInterface(speedID),
 		}
 		if err := tx.Create(&fwd).Error; err != nil {

go-backend/patch.py

@@ -0,0 +1,163 @@
+import re
+
+with open('internal/http/handler/control_plane.go', 'r') as f:
+    content = f.read()
+
+# 1. Update ensureLimiterOnNode and add ensureConnLimiterOnNode
+ensure_conn_limiter = """
+func (h *Handler) ensureConnLimiterOnNode(nodeID int64, limiterName string, maxConn int) error {
+\tlimitStr := fmt.Sprintf("$ %d", maxConn)
+\t
+\tpayload := map[string]interface{}{
+\t\t"name":   limiterName,
+\t\t"limits": []string{limitStr},
+\t}
+\t
+\tif _, err := h.sendNodeCommand(nodeID, "AddCLimiters", payload, false, false); err != nil {
+\t\tif !isAlreadyExistsMessage(err.Error()) {
+\t\t\treturn fmt.Errorf("连接限制器下发失败: %w", err)
+\t\t}
+\t\tupdatePayload := map[string]interface{}{
+\t\t\t"limiter": limiterName,
+\t\t\t"data":    payload,
+\t\t}
+\t\tif _, updateErr := h.sendNodeCommand(nodeID, "UpdateCLimiters", updatePayload, false, false); updateErr != nil {
+\t\t\treturn fmt.Errorf("连接限制器更新失败: %w", updateErr)
+\t\t}
+\t}
+\treturn nil
+}
+"""
+
+content = content.replace('func (h *Handler) ensureLimiterOnNode(nodeID int64, limiterID int64, speed int) error {\n\tif err := h.upsertLimiterOnNode(nodeID, limiterID, speed); err != nil {\n\t\treturn fmt.Errorf("限速规则下发失败: %w", err)\n\t}\n\n\treturn nil\n}',
+'func (h *Handler) ensureLimiterOnNode(nodeID int64, limiterID int64, speed int) error {\n\tif err := h.upsertLimiterOnNode(nodeID, limiterID, speed); err != nil {\n\t\treturn fmt.Errorf("限速规则下发失败: %w", err)\n\t}\n\n\treturn nil\n}\n' + ensure_conn_limiter)
+
+
+# 2. Update buildForwardServiceConfigs declaration
+content = content.replace(
+    'func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, bindIP string, limiterID *int64) []map[string]interface{} {',
+    'func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, port int, bindIP string, limiterID *int64, cLimiterName string) []map[string]interface{} {'
+)
+
+
+# 3. Inject climiter into generated service
+service_map_end = """		}
+		if protocol == "udp" {"""
+service_map_end_new = """		}
+		if cLimiterName != "" {
+			service["climiter"] = cLimiterName
+		}
+		if protocol == "udp" {"""
+content = content.replace(service_map_end, service_map_end_new)
+
+
+# 4. Update syncForwardServicesWithWarnings
+# Find user tunnel resolution
+resolution = """	serviceBase := buildForwardServiceBaseWithResolvedUserTunnel(forward.ID, forward.UserID, userTunnelID)
+
+	for _, fp := range ports {"""
+
+resolution_new = """	serviceBase := buildForwardServiceBaseWithResolvedUserTunnel(forward.ID, forward.UserID, userTunnelID)
+
+	user, err := h.repo.GetUserByID(forward.UserID)
+	if err != nil {
+		return nil, err
+	}
+
+	var cLimiterName string
+	var maxConnToSet int
+
+	if forward.MaxConn > 0 {
+		maxConnToSet = forward.MaxConn
+		cLimiterName = fmt.Sprintf("rule_conn_limit_%d", forward.ID)
+	} else if user != nil && user.MaxConn > 0 {
+		maxConnToSet = user.MaxConn
+		cLimiterName = fmt.Sprintf("user_conn_limit_%d", user.ID)
+	}
+
+	for _, fp := range ports {"""
+content = content.replace(resolution, resolution_new)
+
+# Inject ensureConnLimiterOnNode inside loop
+loop_inner = """		if limiterID != nil && speed != nil {
+			if err := h.ensureLimiterOnNode(fp.NodeID, *limiterID, *speed); err != nil {
+				// If the limiter push fails because the node is offline, skip it with a warning
+				if isNodeOfflineOrTimeoutError(err) {
+					node, _ := h.getNodeRecord(fp.NodeID)
+					nodeName := fmt.Sprintf("%d", fp.NodeID)
+					if node != nil && strings.TrimSpace(node.Name) != "" {
+						nodeName = strings.TrimSpace(node.Name)
+					}
+					warnings = append(warnings, fmt.Sprintf("节点 %s 不在线，已跳过下发", nodeName))
+					continue
+				}
+				return nil, err
+			}
+		}
+
+		node, err := h.getNodeRecord(fp.NodeID)"""
+
+loop_inner_new = """		if limiterID != nil && speed != nil {
+			if err := h.ensureLimiterOnNode(fp.NodeID, *limiterID, *speed); err != nil {
+				// If the limiter push fails because the node is offline, skip it with a warning
+				if isNodeOfflineOrTimeoutError(err) {
+					node, _ := h.getNodeRecord(fp.NodeID)
+					nodeName := fmt.Sprintf("%d", fp.NodeID)
+					if node != nil && strings.TrimSpace(node.Name) != "" {
+						nodeName = strings.TrimSpace(node.Name)
+					}
+					warnings = append(warnings, fmt.Sprintf("节点 %s 不在线，已跳过下发", nodeName))
+					continue
+				}
+				return nil, err
+			}
+		}
+
+		if cLimiterName != "" {
+			if err := h.ensureConnLimiterOnNode(fp.NodeID, cLimiterName, maxConnToSet); err != nil {
+				warnings = append(warnings, fmt.Sprintf("节点 %d 连接限制器下发失败: %v", fp.NodeID, err))
+			}
+		}
+
+		node, err := h.getNodeRecord(fp.NodeID)"""
+content = content.replace(loop_inner, loop_inner_new)
+
+# Update buildForwardServiceConfigs call in syncForwardServicesWithWarnings
+content = content.replace(
+    'services := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, strings.TrimSpace(fp.InIP), limiterID)',
+    'services := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, strings.TrimSpace(fp.InIP), limiterID, cLimiterName)'
+)
+
+# Update fallbackForwardPortToDefaultBind call
+content = content.replace(
+    'warning, err = h.fallbackForwardPortToDefaultBind(forward, tunnel, node, fp, serviceBase, limiterID)',
+    'warning, err = h.fallbackForwardPortToDefaultBind(forward, tunnel, node, fp, serviceBase, limiterID, cLimiterName)'
+)
+
+# 5. Update fallbackForwardPortToDefaultBind declaration and logic
+content = content.replace(
+    'func (h *Handler) fallbackForwardPortToDefaultBind(forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, fp forwardPortRecord, serviceBase string, limiterID *int64) (string, error) {',
+    'func (h *Handler) fallbackForwardPortToDefaultBind(forward *forwardRecord, tunnel *tunnelRecord, node *nodeRecord, fp forwardPortRecord, serviceBase string, limiterID *int64, cLimiterName string) (string, error) {'
+)
+
+content = content.replace(
+    'defaultServices := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, "", limiterID)',
+    'defaultServices := buildForwardServiceConfigs(serviceBase, forward, tunnel, node, fp.Port, "", limiterID, cLimiterName)'
+)
+
+with open('internal/http/handler/control_plane.go', 'w') as f:
+    f.write(content)
+
+
+# Update control_plane_test.go
+with open('internal/http/handler/control_plane_test.go', 'r') as f:
+    test_content = f.read()
+
+test_content = re.sub(
+    r'buildForwardServiceConfigs\((.*?),(.*?),(.*?),(.*?),(.*?),(.*?),(.*?)\)',
+    r'buildForwardServiceConfigs(\1,\2,\3,\4,\5,\6,\7, "")',
+    test_content
+)
+
+with open('internal/http/handler/control_plane_test.go', 'w') as f:
+    f.write(test_content)

go-backend/tests/contract/max_conn_limit_contract_test.go

@@ -0,0 +1,284 @@
+package contract_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+
+	"go-backend/internal/auth"
+	"go-backend/internal/http/response"
+	"go-backend/internal/security"
+)
+
+func TestMaxConnLimit(t *testing.T) {
+	secret := "contract-jwt-secret"
+	router, r := setupContractRouter(t, secret)
+	server := httptest.NewServer(router)
+	defer server.Close()
+
+	adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret)
+	if err != nil {
+		t.Fatalf("generate admin token: %v", err)
+	}
+
+	now := time.Now().UnixMilli()
+	if err := r.DB().Exec(`
+		INSERT INTO tunnel(name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx)
+		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, "max-conn-tunnel", 1.0, 1, "tls", 99999, now, now, 1, nil, 0).Error; err != nil {
+		t.Fatalf("insert tunnel: %v", err)
+	}
+	var tunnelID int64
+	if err := r.DB().Raw("SELECT id FROM tunnel WHERE name = ?", "max-conn-tunnel").Scan(&tunnelID).Error; err != nil {
+		t.Fatalf("get tunnel ID: %v", err)
+	}
+
+	if err := r.DB().Exec(`
+		INSERT INTO node(name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx)
+		VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, "max-conn-node", "max-conn-secret", "10.20.0.1", "10.20.0.1", "", "32000-32010", "", "v1", 1, 1, 1, now, now, 1, "[::]", "[::]", 0).Error; err != nil {
+		t.Fatalf("insert node: %v", err)
+	}
+	var nodeID int64
+	if err := r.DB().Raw("SELECT id FROM node WHERE name = ?", "max-conn-node").Scan(&nodeID).Error; err != nil {
+		t.Fatalf("get node ID: %v", err)
+	}
+
+	if err := r.DB().Exec(`
+	        INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol)
+	        VALUES(?, 1, ?, 32001, 'round', 1, 'tls')
+	`, tunnelID, nodeID).Error; err != nil {
+	        t.Fatalf("insert chain_tunnel: %v", err)
+	}
+
+	if err := r.DB().Exec(`
+	        INSERT INTO user_tunnel(user_id, tunnel_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status)
+	        VALUES(1, ?, 10, 99999, 0, 0, 1, ?, 1)
+	`, tunnelID, now + 365*24*3600*1000).Error; err != nil {	        t.Fatalf("insert user_tunnel: %v", err)
+	}
+
+	var commandMu sync.Mutex
+	receivedCommands := make([]string, 0)
+	var addCLimitersData json.RawMessage
+	var updateCLimitersData json.RawMessage
+
+	stopNode := startMockSessionForMaxConn(t, server.URL, "max-conn-secret", func(cmdType string, data json.RawMessage) (bool, string) {
+		commandMu.Lock()
+		defer commandMu.Unlock()
+		receivedCommands = append(receivedCommands, cmdType)
+
+		if cmdType == "AddCLimiters" {
+			addCLimitersData = append([]byte(nil), data...)
+			return true, "already exists"
+		}
+		if cmdType == "UpdateCLimiters" {
+			updateCLimitersData = append([]byte(nil), data...)
+		}
+
+		return false, ""
+	})
+	defer stopNode()
+
+	waitNodeStatus(t, r, nodeID, 1)
+
+	payload := map[string]interface{}{
+		"name":       "max-conn-forward",
+		"tunnelId":   tunnelID,
+		"remoteAddr": "1.1.1.1:443",
+		"strategy":   "fifo",
+		"maxConn":    42,
+	}
+	body, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatalf("marshal payload: %v", err)
+	}
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/forward/create", bytes.NewReader(body))
+	req.Header.Set("Authorization", adminToken)
+	req.Header.Set("Content-Type", "application/json")
+	res := httptest.NewRecorder()
+	router.ServeHTTP(res, req)
+
+	var out response.R
+	if err := json.NewDecoder(res.Body).Decode(&out); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if out.Code != 0 {
+		t.Fatalf("expected create success, got code=%d msg=%s", out.Code, out.Msg)
+	}
+
+	var forwardID int64
+	if err := r.DB().Raw("SELECT id FROM forward WHERE name = ?", "max-conn-forward").Scan(&forwardID).Error; err != nil {
+		t.Fatalf("get forward ID: %v", err)
+	}
+
+	commandMu.Lock()
+	defer commandMu.Unlock()
+
+	hasAdd := false
+	hasUpdate := false
+	for _, cmd := range receivedCommands {
+		if cmd == "AddCLimiters" {
+			hasAdd = true
+		}
+		if cmd == "UpdateCLimiters" {
+			hasUpdate = true
+		}
+	}
+
+	if !hasAdd {
+		t.Fatalf("expected AddCLimiters to be sent, but it was not. Received: %v", receivedCommands)
+	}
+	if !hasUpdate {
+		t.Fatalf("expected UpdateCLimiters to be sent after AddCLimiters failed with already exists. Received: %v", receivedCommands)
+	}
+
+	expectedName := fmt.Sprintf("rule_conn_limit_%d", forwardID)
+
+	// verify payload for AddCLimiters
+	var addData map[string]interface{}
+	if err := json.Unmarshal(addCLimitersData, &addData); err != nil {
+		t.Fatalf("unmarshal AddCLimiters data: %v", err)
+	}
+	if addData["name"] != expectedName {
+		t.Fatalf("expected limiter name %s, got %v", expectedName, addData["name"])
+	}
+	if limits, ok := addData["limits"].([]interface{}); ok {
+		if len(limits) != 1 || limits[0] != "$ 42" {
+			t.Fatalf("expected limits to contain '$ 42', got %v", limits)
+		}
+	} else {
+		t.Fatalf("invalid limits type in AddCLimiters data: %v", addData)
+	}
+
+	// verify payload for UpdateCLimiters
+	var updateData map[string]interface{}
+	if err := json.Unmarshal(updateCLimitersData, &updateData); err != nil {
+		t.Fatalf("unmarshal UpdateCLimiters data: %v", err)
+	}
+	if updateData["limiter"] != expectedName {
+		t.Fatalf("expected update limiter name %s, got %v", expectedName, updateData["limiter"])
+	}
+	nestedData, ok := updateData["data"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("expected nested 'data' in UpdateCLimiters, got %v", updateData)
+	}
+	if nestedData["name"] != expectedName {
+		t.Fatalf("expected nested name %s, got %v", expectedName, nestedData["name"])
+	}
+	if nestedLimits, ok := nestedData["limits"].([]interface{}); ok {
+		if len(nestedLimits) != 1 || nestedLimits[0] != "$ 42" {
+			t.Fatalf("expected nested limits to contain '$ 42', got %v", nestedLimits)
+		}
+	} else {
+		t.Fatalf("invalid limits type in UpdateCLimiters nested data: %v", nestedData)
+	}
+}
+
+func startMockSessionForMaxConn(t *testing.T, baseURL string, nodeSecret string, onCommand func(cmdType string, data json.RawMessage) (bool, string)) func() {
+	t.Helper()
+
+	u, err := url.Parse(baseURL)
+	if err != nil {
+		t.Fatalf("parse provider url: %v", err)
+	}
+	if strings.EqualFold(u.Scheme, "https") {
+		u.Scheme = "wss"
+	} else {
+		u.Scheme = "ws"
+	}
+	u.Path = "/system-info"
+	q := u.Query()
+	q.Set("type", "1")
+	q.Set("secret", nodeSecret)
+	q.Set("version", "v1")
+	q.Set("http", "1")
+	q.Set("tls", "1")
+	q.Set("socks", "1")
+	u.RawQuery = q.Encode()
+
+	conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	if err != nil {
+		t.Fatalf("dial mock node websocket: %v", err)
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			_, raw, readErr := conn.ReadMessage()
+			if readErr != nil {
+				return
+			}
+
+			plain := raw
+			var wrap struct {
+				Encrypted bool   `json:"encrypted"`
+				Data      string `json:"data"`
+			}
+			if err := json.Unmarshal(raw, &wrap); err == nil && wrap.Encrypted && strings.TrimSpace(wrap.Data) != "" {
+				crypto, cryptoErr := security.NewAESCrypto(nodeSecret)
+				if cryptoErr == nil {
+					if dec, decErr := crypto.Decrypt(wrap.Data); decErr == nil {
+						plain = []byte(dec)
+					}
+				}
+			}
+
+			var cmd struct {
+				Type      string          `json:"type"`
+				RequestID string          `json:"requestId"`
+				Data      json.RawMessage `json:"data"`
+			}
+			if err := json.Unmarshal(plain, &cmd); err != nil {
+				continue
+			}
+			if strings.TrimSpace(cmd.RequestID) == "" {
+				continue
+			}
+
+			shouldFail := false
+			failMsg := ""
+			if onCommand != nil {
+				shouldFail, failMsg = onCommand(strings.TrimSpace(cmd.Type), cmd.Data)
+			}
+
+			respType := fmt.Sprintf("%sResponse", cmd.Type)
+			respPayload := map[string]interface{}{
+				"type":      respType,
+				"success":   !shouldFail,
+				"message":   "OK",
+				"requestId": cmd.RequestID,
+			}
+			if shouldFail {
+				if strings.TrimSpace(failMsg) == "" {
+					failMsg = "mock command failed"
+				}
+				respPayload["message"] = failMsg
+			}
+
+			respBytes, err := json.Marshal(respPayload)
+			if err != nil {
+				continue
+			}
+			_ = conn.WriteMessage(websocket.TextMessage, respBytes)
+		}
+	}()
+
+	var stopOnce sync.Once
+	return func() {
+		stopOnce.Do(func() {
+			_ = conn.Close()
+			wg.Wait()
+		})
+	}
+}