fix(backend): sync limiter on association instead of connection

Reverted the full sync on connection hook. Instead, ensureLimiterOnNode is called within syncForwardServices to push limiter configuration immediately before pushing the service configuration that references it.
This commit is contained in:
sagit
2026-02-09 09:06:59 +00:00
parent 3d7a0b697d
commit 3420dc5460
3 changed files with 23 additions and 42 deletions
@@ -234,9 +234,9 @@ func (h *Handler) getNodeRecord(nodeID int64) (*nodeRecord, error) {
return &n, nil return &n, nil
} }
func (h *Handler) resolveUserTunnelAndLimiter(userID, tunnelID int64) (int64, *int64, error) { func (h *Handler) resolveUserTunnelAndLimiter(userID, tunnelID int64) (int64, *int64, *int, error) {
row := h.repo.DB().QueryRow(` row := h.repo.DB().QueryRow(`
SELECT ut.id, sl.id SELECT ut.id, sl.id, sl.speed
FROM user_tunnel ut FROM user_tunnel ut
LEFT JOIN speed_limit sl ON sl.id = ut.speed_id LEFT JOIN speed_limit sl ON sl.id = ut.speed_id
WHERE ut.user_id = ? AND ut.tunnel_id = ? WHERE ut.user_id = ? AND ut.tunnel_id = ?
@@ -245,18 +245,20 @@ func (h *Handler) resolveUserTunnelAndLimiter(userID, tunnelID int64) (int64, *i
`, userID, tunnelID) `, userID, tunnelID)
var userTunnelID int64 var userTunnelID int64
var limiterID sql.NullInt64 var limiterID sql.NullInt64
err := row.Scan(&userTunnelID, &limiterID) var speed sql.NullInt64
err := row.Scan(&userTunnelID, &limiterID, &speed)
if err != nil { if err != nil {
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return 0, nil, nil return 0, nil, nil, nil
} }
return 0, nil, err return 0, nil, nil, err
} }
if !limiterID.Valid || limiterID.Int64 <= 0 { if !limiterID.Valid || limiterID.Int64 <= 0 {
return userTunnelID, nil, nil return userTunnelID, nil, nil, nil
} }
v := limiterID.Int64 v := limiterID.Int64
return userTunnelID, &v, nil s := int(speed.Int64)
return userTunnelID, &v, &s, nil
} }
func (h *Handler) listUserTunnelIDs(userID, tunnelID int64) ([]int64, error) { func (h *Handler) listUserTunnelIDs(userID, tunnelID int64) ([]int64, error) {
@@ -328,13 +330,17 @@ func (h *Handler) syncForwardServices(forward *forwardRecord, method string, all
return errors.New("转发入口端口不存在") return errors.New("转发入口端口不存在")
} }
userTunnelID, limiterID, err := h.resolveUserTunnelAndLimiter(forward.UserID, forward.TunnelID) userTunnelID, limiterID, speed, err := h.resolveUserTunnelAndLimiter(forward.UserID, forward.TunnelID)
if err != nil { if err != nil {
return err return err
} }
serviceBase := buildForwardServiceBase(forward.ID, forward.UserID, userTunnelID) serviceBase := buildForwardServiceBase(forward.ID, forward.UserID, userTunnelID)
for _, fp := range ports { for _, fp := range ports {
if limiterID != nil && speed != nil {
h.ensureLimiterOnNode(fp.NodeID, *limiterID, *speed)
}
node, err := h.getNodeRecord(fp.NodeID) node, err := h.getNodeRecord(fp.NodeID)
if err != nil { if err != nil {
return err return err
@@ -362,7 +368,7 @@ func (h *Handler) controlForwardServices(forward *forwardRecord, commandType str
if len(ports) == 0 { if len(ports) == 0 {
return nil return nil
} }
userTunnelID, _, err := h.resolveUserTunnelAndLimiter(forward.UserID, forward.TunnelID) userTunnelID, _, _, err := h.resolveUserTunnelAndLimiter(forward.UserID, forward.TunnelID)
if err != nil { if err != nil {
return err return err
} }
@@ -1145,29 +1151,12 @@ func (h *Handler) sendDeleteLimiterConfig(limiterID int64, tunnelID int64) error
return nil return nil
} }
func (h *Handler) onNodeConnected(nodeID int64) { func (h *Handler) ensureLimiterOnNode(nodeID int64, limiterID int64, speed int) {
// Sync limiters
rows, err := h.repo.DB().Query(`
SELECT DISTINCT sl.id, sl.speed
FROM speed_limit sl
JOIN chain_tunnel ct ON ct.tunnel_id = sl.tunnel_id
WHERE ct.node_id = ? AND ct.chain_type = 1 AND sl.status = 1
`, nodeID)
if err == nil {
defer rows.Close()
for rows.Next() {
var id int64
var speed int
if err := rows.Scan(&id, &speed); err == nil {
rate := float64(speed) / 8.0 rate := float64(speed) / 8.0
limitStr := fmt.Sprintf("$ %.1fMB %.1fMB", rate, rate) limitStr := fmt.Sprintf("$ %.1fMB %.1fMB", rate, rate)
payload := map[string]interface{}{ payload := map[string]interface{}{
"name": strconv.FormatInt(id, 10), "name": strconv.FormatInt(limiterID, 10),
"limits": []string{limitStr}, "limits": []string{limitStr},
} }
_, _ = h.sendNodeCommand(nodeID, "AddLimiters", payload, false, false) _, _ = h.sendNodeCommand(nodeID, "AddLimiters", payload, false, false)
} }
}
}
}
+1 -3
View File
@@ -62,13 +62,11 @@ type flowItem struct {
} }
func New(repo *sqlite.Repository, jwtSecret string) *Handler { func New(repo *sqlite.Repository, jwtSecret string) *Handler {
h := &Handler{ return &Handler{
repo: repo, repo: repo,
jwtSecret: jwtSecret, jwtSecret: jwtSecret,
wsServer: ws.NewServer(repo, jwtSecret), wsServer: ws.NewServer(repo, jwtSecret),
} }
h.wsServer.OnNodeConnected = h.onNodeConnected
return h
} }
func (h *Handler) WebSocketHandler() http.Handler { func (h *Handler) WebSocketHandler() http.Handler {
-6
View File
@@ -71,8 +71,6 @@ type Server struct {
nodes map[int64]*nodeSession nodes map[int64]*nodeSession
byConn map[*websocket.Conn]*nodeSession byConn map[*websocket.Conn]*nodeSession
pending map[string]pendingRequest pending map[string]pendingRequest
OnNodeConnected func(nodeID int64)
} }
func NewServer(repo *sqlite.Repository, jwtSecret string) *Server { func NewServer(repo *sqlite.Repository, jwtSecret string) *Server {
@@ -166,10 +164,6 @@ func (s *Server) handleNode(w http.ResponseWriter, r *http.Request, nodeID int64
_ = s.repo.UpdateNodeOnline(nodeID, 1, version, httpVal, tlsVal, socksVal) _ = s.repo.UpdateNodeOnline(nodeID, 1, version, httpVal, tlsVal, socksVal)
s.broadcastStatus(nodeID, 1) s.broadcastStatus(nodeID, 1)
if s.OnNodeConnected != nil {
go s.OnNodeConnected(nodeID)
}
defer func() { defer func() {
needOfflineBroadcast := false needOfflineBroadcast := false
s.mu.Lock() s.mu.Lock()