Welcome to mirror list, hosted at ThFree Co, Russian Federation.

github.com/MHSanaei/3x-ui.git - Unnamed repository; edit this file 'description' to name the repository.
summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorlolka1333 <xtrafcyz@gmail.com>2026-05-05 18:27:49 +0300
committerGitHub <noreply@github.com>2026-05-05 18:27:49 +0300
commit8177f6dc667f2edca66073e433baf5cff36cda41 (patch)
tree7fddbec4848c3c0ab0fc39e3dc45703c368e9340 /web/controller/inbound.go
parent77d94b25d054bd6cf7ace029571db9c58ae87fa9 (diff)
ws/inbounds: realtime fixes + perf for 10k+ client inbounds (#4123)HEADmain
* ws/inbounds: realtime fixes + perf for 10k+ client inbounds - hub: dedup, throttle, panic-restart, deadlock fix, race tests - client: backoff cap + slow-retry instead of giving up - broadcast: delta-only payload, count-based invalidate fallback - filter: fix empty online list (Inbound has no .id, use dbInbound.toInbound) - perf: O(N²)→O(N) traffic merge, bulk delete, /setEnable endpoint - traffic: monotonic all_time + UI clamp + propagate in delta handler - session: persist on update/logout (fixes logout-after-password-change) - ui: protocol tags flex, traffic bar normalize * Remove hub_test.go file * fix: ws hub, inbound service, and frontend correctness - propagate DelInbound error on disable path in SetInboundEnable - skip empty emails in updateClientTraffics to avoid constraint violations - use consistent IN ? clause, drop redundant ErrRecordNotFound guards - Hub.Unregister: direct removeClient fallback when channel is full - applyClientStatsDelta: O(1) email lookup via per-inbound Map cache - WS payload size check: Blob.size instead of .length for real byte count * fix: chunk large IN ? queries and fix IPv6 same-origin check * fix: chunk large IN ? queries and fix IPv6 same-origin check * fix: unify clientStats cache, throttle clarity, hub constants * fix(ui): align traffic/expiry cell columns across all rows * style(ui): redesign outbounds table for visual consistency * style(ui): redesign routing table for visual consistency * fix: * fix: * fix: * fix: * fix: * fix: font * refactor: simplify outbound tone functions for consistency and maintainability --------- Co-authored-by: lolka1333 <test123@gmail.com>
Diffstat (limited to 'web/controller/inbound.go')
-rw-r--r--web/controller/inbound.go76
1 files changed, 67 insertions, 9 deletions
diff --git a/web/controller/inbound.go b/web/controller/inbound.go
index ee024cc6..5b8ad38b 100644
--- a/web/controller/inbound.go
+++ b/web/controller/inbound.go
@@ -27,6 +27,34 @@ func NewInboundController(g *gin.RouterGroup) *InboundController {
return a
}
+// broadcastInboundsUpdateClientLimit is the threshold past which we skip the
+// full-list push over WebSocket and signal the frontend to re-fetch via REST.
+// Mirrors the same heuristic used by the periodic traffic job.
+const broadcastInboundsUpdateClientLimit = 5000
+
+// broadcastInboundsUpdate fetches and broadcasts the inbound list for userId.
+// At scale (10k+ clients) the marshaled JSON exceeds the WS payload ceiling,
+// so we send an invalidate signal instead — frontend re-fetches via REST.
+// Skipped entirely when no WebSocket clients are connected.
+func (a *InboundController) broadcastInboundsUpdate(userId int) {
+ if !websocket.HasClients() {
+ return
+ }
+ inbounds, err := a.inboundService.GetInbounds(userId)
+ if err != nil {
+ return
+ }
+ totalClients := 0
+ for _, ib := range inbounds {
+ totalClients += len(ib.ClientStats)
+ }
+ if totalClients > broadcastInboundsUpdateClientLimit {
+ websocket.BroadcastInvalidate(websocket.MessageTypeInbounds)
+ return
+ }
+ websocket.BroadcastInbounds(inbounds)
+}
+
// initRouter initializes the routes for inbound-related operations.
func (a *InboundController) initRouter(g *gin.RouterGroup) {
@@ -38,6 +66,7 @@ func (a *InboundController) initRouter(g *gin.RouterGroup) {
g.POST("/add", a.addInbound)
g.POST("/del/:id", a.delInbound)
g.POST("/update/:id", a.updateInbound)
+ g.POST("/setEnable/:id", a.setInboundEnable)
g.POST("/clientIps/:email", a.getClientIps)
g.POST("/clearClientIps/:email", a.clearClientIps)
g.POST("/addClient", a.addInboundClient)
@@ -134,9 +163,7 @@ func (a *InboundController) addInbound(c *gin.Context) {
if needRestart {
a.xrayService.SetToNeedRestart()
}
- // Broadcast inbounds update via WebSocket
- inbounds, _ := a.inboundService.GetInbounds(user.Id)
- websocket.BroadcastInbounds(inbounds)
+ a.broadcastInboundsUpdate(user.Id)
}
// delInbound deletes an inbound configuration by its ID.
@@ -155,10 +182,8 @@ func (a *InboundController) delInbound(c *gin.Context) {
if needRestart {
a.xrayService.SetToNeedRestart()
}
- // Broadcast inbounds update via WebSocket
user := session.GetLoginUser(c)
- inbounds, _ := a.inboundService.GetInbounds(user.Id)
- websocket.BroadcastInbounds(inbounds)
+ a.broadcastInboundsUpdate(user.Id)
}
// updateInbound updates an existing inbound configuration.
@@ -185,10 +210,43 @@ func (a *InboundController) updateInbound(c *gin.Context) {
if needRestart {
a.xrayService.SetToNeedRestart()
}
- // Broadcast inbounds update via WebSocket
user := session.GetLoginUser(c)
- inbounds, _ := a.inboundService.GetInbounds(user.Id)
- websocket.BroadcastInbounds(inbounds)
+ a.broadcastInboundsUpdate(user.Id)
+}
+
+// setInboundEnable flips only the enable flag of an inbound. This is a
+// dedicated endpoint because the regular update path serialises the entire
+// settings JSON (every client) — far too heavy for an interactive switch
+// on inbounds with thousands of clients. Frontend optimistically updates
+// the UI; we just persist + sync xray + nudge other open admin sessions.
+func (a *InboundController) setInboundEnable(c *gin.Context) {
+ id, err := strconv.Atoi(c.Param("id"))
+ if err != nil {
+ jsonMsg(c, I18nWeb(c, "pages.inbounds.toasts.inboundUpdateSuccess"), err)
+ return
+ }
+ type form struct {
+ Enable bool `json:"enable" form:"enable"`
+ }
+ var f form
+ if err := c.ShouldBind(&f); err != nil {
+ jsonMsg(c, I18nWeb(c, "somethingWentWrong"), err)
+ return
+ }
+ needRestart, err := a.inboundService.SetInboundEnable(id, f.Enable)
+ if err != nil {
+ jsonMsg(c, I18nWeb(c, "somethingWentWrong"), err)
+ return
+ }
+ jsonMsg(c, I18nWeb(c, "pages.inbounds.toasts.inboundUpdateSuccess"), nil)
+ if needRestart {
+ a.xrayService.SetToNeedRestart()
+ }
+ // Cross-admin sync: lightweight invalidate signal (a few hundred bytes)
+ // instead of fetching + serialising the whole inbound list. Other open
+ // sessions re-fetch via REST. The toggling admin's own UI already
+ // updated optimistically.
+ websocket.BroadcastInvalidate(websocket.MessageTypeInbounds)
}
// getClientIps retrieves the IP addresses associated with a client by email.