feat(nodes): show live speed for node-hosted inbounds

Inbound speed is computed on the dashboard from a 'traffics' delta feed, which only the local Xray poll produced — so node-hosted inbounds showed no speed. The node sync now diffs successive per-inbound cumulative totals (it polls @5s, same as the local poll) and broadcasts the byte deltas as a separate 'nodeTraffics' field, keyed by the central tag the dashboard already matches. The frontend applies 'traffics' to local inbounds and 'nodeTraffics' to node inbounds within their own scope, so the two 5s polls don't clobber each other and idle inbounds still clear. Deltas clamp to 0 on a reset; a node that fails to sync keeps a stale total so its delta is 0 (no phantom speed).

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<string, string[]>;
         activeInbounds?: Record<string, string[]>;
@@ -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<string, TrafficDelta>();
-        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<number, InboundSpeedEntry> = {};
-        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],

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()

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