diff --git a/frontend/src/pages/inbounds/useInbounds.ts b/frontend/src/pages/inbounds/useInbounds.ts index 723e5182f..a7dc9c00b 100644 --- a/frontend/src/pages/inbounds/useInbounds.ts +++ b/frontend/src/pages/inbounds/useInbounds.ts @@ -399,6 +399,7 @@ export function useInbounds() { if (!payload || typeof payload !== 'object') return; const p = payload as { traffics?: TrafficDelta[]; + nodeTraffics?: TrafficDelta[]; onlineClients?: string[]; onlineByGuid?: Record; activeInbounds?: Record; @@ -417,26 +418,40 @@ export function useInbounds() { if (p.lastOnlineMap && typeof p.lastOnlineMap === 'object') { setLastOnlineMap((prev) => ({ ...prev, ...p.lastOnlineMap! })); } - // Full-replace each poll so idle inbounds (and an empty array after an - // Xray stat reset) clear their speed instead of showing a stale value. - if (Array.isArray(p.traffics)) { + // Speed arrives from two independent 5s polls: the local Xray poll sends + // `traffics` (local inbounds) and the node sync sends `nodeTraffics` (node + // inbounds). Each replaces speed only within its own scope so the two don't + // clobber each other; an idle in-scope inbound — absent from its payload — + // clears instead of showing a stale value. + const applyTraffics = ( + traffics: TrafficDelta[], + inScope: (ib: DBInboundInstance) => boolean, + ) => { const byTag = new Map(); - for (const tr of p.traffics) { + for (const tr of traffics) { if (!tr || typeof tr.Tag !== 'string') continue; if (tr.IsInbound === false) continue; byTag.set(tr.Tag, tr); } - const nextSpeed: Record = {}; - for (const ib of dbInboundsRef.current) { - const delta = byTag.get(ib.tag); - if (!delta) continue; - nextSpeed[ib.id] = { - up: (delta.Up || 0) / TRAFFIC_POLL_INTERVAL_S, - down: (delta.Down || 0) / TRAFFIC_POLL_INTERVAL_S, - }; - } - setInboundSpeed(nextSpeed); - } + setInboundSpeed((prev) => { + const next = { ...prev }; + for (const ib of dbInboundsRef.current) { + if (!inScope(ib)) continue; + const delta = byTag.get(ib.tag); + if (delta) { + next[ib.id] = { + up: (delta.Up || 0) / TRAFFIC_POLL_INTERVAL_S, + down: (delta.Down || 0) / TRAFFIC_POLL_INTERVAL_S, + }; + } else { + delete next[ib.id]; + } + } + return next; + }); + }; + if (Array.isArray(p.traffics)) applyTraffics(p.traffics, (ib) => ib.nodeId == null); + if (Array.isArray(p.nodeTraffics)) applyTraffics(p.nodeTraffics, (ib) => ib.nodeId != null); rebuildClientCount(); }, [rebuildClientCount], diff --git a/internal/web/job/node_traffic_sync_job.go b/internal/web/job/node_traffic_sync_job.go index e9812dbf4..a91264763 100644 --- a/internal/web/job/node_traffic_sync_job.go +++ b/internal/web/job/node_traffic_sync_job.go @@ -12,6 +12,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" "github.com/mhsanaei/3x-ui/v3/internal/web/service" "github.com/mhsanaei/3x-ui/v3/internal/web/websocket" + "github.com/mhsanaei/3x-ui/v3/internal/xray" ) const ( @@ -37,6 +38,10 @@ type NodeTrafficSyncJob struct { // noGuidIpEndpoint tracks nodes (by id) whose client-IP attribution endpoint // returned 404, so an old-build node is noted once instead of every cycle. noGuidIpEndpoint sync.Map + // prevInboundTotals holds the previous poll's cumulative up/down per node + // inbound tag, so the next poll can derive a per-inbound speed delta (node + // inbounds have no local Xray poll). Touched only from Run (serialized). + prevInboundTotals map[string][2]int64 } type atomicBool struct { @@ -140,6 +145,10 @@ func (j *NodeTrafficSyncJob) Run() { // xray's clients and inbounds still age out between traffic polls. j.inboundService.RefreshLocalOnlineClients(nil, nil) + // Derive per-node-inbound speed every tick (keeps the baseline fresh even + // with no dashboard open); only broadcast it when someone is watching. + inboundSpeed := j.nodeInboundSpeed() + if !websocket.HasClients() { return } @@ -148,12 +157,18 @@ func (j *NodeTrafficSyncJob) Run() { if online == nil { online = []string{} } - websocket.BroadcastTraffic(map[string]any{ + trafficPayload := map[string]any{ "onlineClients": online, "onlineByGuid": j.inboundService.GetOnlineClientsByGuid(), "activeInbounds": j.inboundService.GetActiveInboundsByGuid(), "lastOnlineMap": lastOnline, - }) + } + // Always send the key so the dashboard clears node inbounds that went idle + // this tick. A nil result (query error) marshals to null and is skipped + // client-side, leaving the last shown value untouched; an empty (non-nil) + // slice marshals to [] and clears stale speeds. + trafficPayload["nodeTraffics"] = inboundSpeed + websocket.BroadcastTraffic(trafficPayload) clientStats := map[string]any{} if stats, err := j.inboundService.GetAllClientTraffics(); err != nil { @@ -181,6 +196,37 @@ func (j *NodeTrafficSyncJob) Run() { // enforce its quota locally (see InboundService.AcceptGlobalTraffic). Scoped // per node to the clients that node actually hosts, and throttled — the // aggregates only need to reach nodes on a human timescale, not every poll. +// nodeInboundSpeed diffs the current node-inbound cumulative totals against the +// previous poll's and returns per-inbound byte deltas (clamped at 0 on a reset), +// keyed by the central tag the dashboard matches. It updates the baseline for +// the next poll. A node that failed to sync this tick keeps its stale total, so +// its delta is 0 — no phantom speed — and a tag seen for the first time yields +// no delta until the next poll. +func (j *NodeTrafficSyncJob) nodeInboundSpeed() []*xray.Traffic { + totals, err := j.inboundService.GetNodeInboundTrafficTotals() + if err != nil { + return nil + } + deltas := make([]*xray.Traffic, 0, len(totals)) + for tag, cur := range totals { + if prev, ok := j.prevInboundTotals[tag]; ok { + dUp := cur[0] - prev[0] + dDown := cur[1] - prev[1] + if dUp < 0 { + dUp = 0 + } + if dDown < 0 { + dDown = 0 + } + if dUp > 0 || dDown > 0 { + deltas = append(deltas, &xray.Traffic{Tag: tag, IsInbound: true, Up: dUp, Down: dDown}) + } + } + } + j.prevInboundTotals = totals + return deltas +} + func (j *NodeTrafficSyncJob) maybePushGlobals(mgr *runtime.Manager, nodes []*model.Node) { j.globalPushMu.Lock() now := time.Now().Unix() diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 184739af5..c7782f962 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -201,6 +201,29 @@ func (s *InboundService) SetRemoteTraffic(nodeID int, snap *runtime.TrafficSnaps return structuralChange, err } +// GetNodeInboundTrafficTotals returns the current cumulative up/down for every +// node-hosted inbound, keyed by tag. The node sync diffs successive snapshots of +// this to derive per-inbound speed for the dashboard — node inbounds have no +// local Xray poll to produce live deltas the way local inbounds do. +func (s *InboundService) GetNodeInboundTrafficTotals() (map[string][2]int64, error) { + var rows []struct { + Tag string + Up int64 + Down int64 + } + if err := database.GetDB().Table("inbounds"). + Select("tag, up, down"). + Where("node_id IS NOT NULL"). + Scan(&rows).Error; err != nil { + return nil, err + } + out := make(map[string][2]int64, len(rows)) + for _, r := range rows { + out[r.Tag] = [2]int64{r.Up, r.Down} + } + return out, nil +} + func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.TrafficSnapshot, dirty bool) (bool, error) { if snap == nil || nodeID <= 0 { return false, nil