diff --git a/frontend/src/hooks/useClients.ts b/frontend/src/hooks/useClients.ts index b2f3bd2ae..a1991e28b 100644 --- a/frontend/src/hooks/useClients.ts +++ b/frontend/src/hooks/useClients.ts @@ -32,7 +32,7 @@ import { type BulkDetachResult, } from '@/schemas/client'; import { DefaultsPayloadSchema } from '@/schemas/defaults'; -import { TRAFFIC_POLL_INTERVAL_S } from '@/lib/traffic/poll-interval'; +import { TRAFFIC_POLL_INTERVAL_S, SIDECAR_TRAFFIC_POLL_INTERVAL_S } from '@/lib/traffic/poll-interval'; // One row sent to POST /clients/:email/externalLinks. export type ExternalLinkInput = { kind: 'link' | 'subscription'; value: string; remark: string }; @@ -271,6 +271,18 @@ export function useClients() { // the server's authoritative total for the headline count. const [allClientStats, setAllClientStats] = useState([]); const [clientSpeed, setClientSpeed] = useState>({}); + // AmneziaWG/MTProto run entirely outside xray-core, so their live speed + // never arrives via the xray-native clientTraffics broadcast below -- each + // sidecar job broadcasts its own ~10s snapshot under protocol-named keys + // instead (see internal/web/job/sidecar_traffic.go). Kept as two + // independent, protocol-only maps rather than folding into clientSpeed: a + // client's email can't move between protocols, but this hook (unlike + // useInbounds.ts) has no cheap per-email protocol lookup, so a shared map's + // full-replace could only be made safe by tagging ownership per entry -- + // two plain maps are simpler and just as correct, since each is written by + // exactly one job. + const [amneziawgClientSpeed, setAmneziawgClientSpeed] = useState>({}); + const [mtprotoClientSpeed, setMtprotoClientSpeed] = useState>({}); const summary = useMemo(() => { const serverSummary = listQuery.data?.summary ?? DEFAULT_SUMMARY; if (allClientStats.length === 0) return serverSummary; @@ -553,6 +565,8 @@ export function useClients() { const p = payload as { onlineClients?: string[]; clientTraffics?: { email: string; up: number; down: number }[]; + amneziawgClientTraffics?: { email: string; up: number; down: number }[]; + mtprotoClientTraffics?: { email: string; up: number; down: number }[]; }; if (Array.isArray(p.onlineClients)) { queryClient.setQueryData(keys.clients.onlines(), p.onlineClients); @@ -568,6 +582,28 @@ export function useClients() { } setClientSpeed(next); } + // Mirrors the block above exactly, but as two independent, protocol-only + // maps (see the amneziawgClientSpeed/mtprotoClientSpeed declaration). + const applySidecarClientTraffics = ( + traffics: { email: string; up: number; down: number }[], + setSpeed: (next: Record) => void, + ) => { + const next: Record = {}; + for (const ct of traffics) { + if (!ct || !ct.email) continue; + next[ct.email] = { + up: (ct.up || 0) / SIDECAR_TRAFFIC_POLL_INTERVAL_S, + down: (ct.down || 0) / SIDECAR_TRAFFIC_POLL_INTERVAL_S, + }; + } + setSpeed(next); + }; + if (Array.isArray(p.amneziawgClientTraffics)) { + applySidecarClientTraffics(p.amneziawgClientTraffics, setAmneziawgClientSpeed); + } + if (Array.isArray(p.mtprotoClientTraffics)) { + applySidecarClientTraffics(p.mtprotoClientTraffics, setMtprotoClientSpeed); + } }, [queryClient]); const applyClientStatsEvent = useCallback((payload: unknown) => { @@ -606,6 +642,16 @@ export function useClients() { queryRef.current = query; }, [query]); + // AmneziaWG/MTProto speed lives in its own state (see above) and is merged + // in here only for consumers -- a client's email is always exactly one + // protocol, so this can never overwrite a real xray-native entry. + const clientSpeedOut = useMemo(() => { + if (Object.keys(amneziawgClientSpeed).length === 0 && Object.keys(mtprotoClientSpeed).length === 0) { + return clientSpeed; + } + return { ...clientSpeed, ...amneziawgClientSpeed, ...mtprotoClientSpeed }; + }, [clientSpeed, amneziawgClientSpeed, mtprotoClientSpeed]); + return { clients, total, @@ -650,7 +696,7 @@ export function useClients() { exportClients, importClients, setEnable, - clientSpeed, + clientSpeed: clientSpeedOut, applyTrafficEvent, applyClientStatsEvent, }; diff --git a/frontend/src/lib/traffic/poll-interval.ts b/frontend/src/lib/traffic/poll-interval.ts index e216c9351..7806a6d28 100644 --- a/frontend/src/lib/traffic/poll-interval.ts +++ b/frontend/src/lib/traffic/poll-interval.ts @@ -1 +1,5 @@ export const TRAFFIC_POLL_INTERVAL_S = 5; + +// Mirrors cadenceAmneziaWG / cadenceMtproto in internal/web/web.go (both +// "@every 10s"). If either cadence constant changes, update this to match. +export const SIDECAR_TRAFFIC_POLL_INTERVAL_S = 10; diff --git a/frontend/src/pages/inbounds/useInbounds.ts b/frontend/src/pages/inbounds/useInbounds.ts index 84e2ef0d1..9b747c59a 100644 --- a/frontend/src/pages/inbounds/useInbounds.ts +++ b/frontend/src/pages/inbounds/useInbounds.ts @@ -13,7 +13,7 @@ import { OnlinesSchema, OnlineByNodeSchema, ActiveInboundsByNodeSchema } from '@ import { DefaultsPayloadSchema, type DefaultsPayload } from '@/schemas/defaults'; import type { InboundSpeedEntry } from './list/types'; -import { TRAFFIC_POLL_INTERVAL_S } from '@/lib/traffic/poll-interval'; +import { TRAFFIC_POLL_INTERVAL_S, SIDECAR_TRAFFIC_POLL_INTERVAL_S } from '@/lib/traffic/poll-interval'; export interface SubSettings { enable: boolean; @@ -206,6 +206,16 @@ export function useInbounds() { inboundSpeedCache = { at: Date.now(), data: inboundSpeed }; }, [inboundSpeed]); + // AmneziaWG/MTProto run entirely outside xray-core, so their live speed + // never arrives via the xray-native traffics/nodeTraffics broadcast above + // -- each sidecar job broadcasts its own ~10s snapshot under protocol-named + // keys instead (see internal/web/job/sidecar_traffic.go). Tracked in their + // own state, independent from inboundSpeed, and merged in only at read + // time (inboundSpeedOut below) -- a given inbound is exactly one protocol, + // so the maps never need to agree on the same id. + const [amneziawgInboundSpeed, setAmneziawgInboundSpeed] = useState>({}); + const [mtprotoInboundSpeed, setMtprotoInboundSpeed] = useState>({}); + const [onlineClients, setOnlineClients] = useState([]); const onlineClientsRef = useRef([]); onlineClientsRef.current = onlineClients; @@ -413,6 +423,8 @@ export function useInbounds() { const p = payload as { traffics?: TrafficDelta[]; nodeTraffics?: TrafficDelta[]; + amneziawgTraffics?: TrafficDelta[]; + mtprotoTraffics?: TrafficDelta[]; onlineClients?: string[]; onlineByGuid?: Record; activeInbounds?: Record; @@ -465,6 +477,48 @@ export function useInbounds() { }; if (Array.isArray(p.traffics)) applyTraffics(p.traffics, (ib) => ib.nodeId == null); if (Array.isArray(p.nodeTraffics)) applyTraffics(p.nodeTraffics, (ib) => ib.nodeId != null); + + // AmneziaWG/MTProto never appear in traffics/nodeTraffics above (they + // don't run inside xray-core), so each broadcasts its own ~10s + // snapshot under its own protocol-named keys instead (see + // internal/web/job/sidecar_traffic.go). Sidecar inbounds are always + // local-only (isNodeEligibleProtocol excludes both server-side), so + // no local/node scope split is needed here. + const applySidecarInboundTraffics = ( + traffics: TrafficDelta[], + protocol: string, + setSpeed: (updater: (prev: Record) => Record) => void, + ) => { + const byTag = new Map(); + for (const tr of traffics) { + if (!tr || typeof tr.Tag !== 'string') continue; + if (tr.IsInbound === false) continue; + byTag.set(tr.Tag, tr); + } + setSpeed((prev) => { + const next = { ...prev }; + for (const ib of dbInboundsRef.current) { + if (ib.protocol !== protocol) continue; + const delta = byTag.get(ib.tag); + if (delta) { + next[ib.id] = { + up: (delta.Up || 0) / SIDECAR_TRAFFIC_POLL_INTERVAL_S, + down: (delta.Down || 0) / SIDECAR_TRAFFIC_POLL_INTERVAL_S, + }; + } else { + delete next[ib.id]; + } + } + return next; + }); + }; + if (Array.isArray(p.amneziawgTraffics)) { + applySidecarInboundTraffics(p.amneziawgTraffics, Protocols.AMNEZIAWG, setAmneziawgInboundSpeed); + } + if (Array.isArray(p.mtprotoTraffics)) { + applySidecarInboundTraffics(p.mtprotoTraffics, Protocols.MTPROTO, setMtprotoInboundSpeed); + } + rebuildClientCount(); }, [rebuildClientCount], @@ -542,6 +596,16 @@ export function useInbounds() { return { up, down }; }, [dbInbounds]); + // AmneziaWG/MTProto speed lives in its own state (see above) and is merged + // in here only for consumers -- a given inbound is exactly one protocol, + // so this can never overwrite a real xray-native entry. + const inboundSpeedOut = useMemo(() => { + if (Object.keys(amneziawgInboundSpeed).length === 0 && Object.keys(mtprotoInboundSpeed).length === 0) { + return inboundSpeed; + } + return { ...inboundSpeed, ...amneziawgInboundSpeed, ...mtprotoInboundSpeed }; + }, [inboundSpeed, amneziawgInboundSpeed, mtprotoInboundSpeed]); + return { fetched, fetchError, @@ -549,7 +613,7 @@ export function useInbounds() { clientCount, onlineClients, lastOnlineMap, - inboundSpeed, + inboundSpeed: inboundSpeedOut, statsVersion, totals, expireDiff, diff --git a/internal/web/job/amneziawg_job.go b/internal/web/job/amneziawg_job.go index 1f8f82153..b9ad10772 100644 --- a/internal/web/job/amneziawg_job.go +++ b/internal/web/job/amneziawg_job.go @@ -2,6 +2,7 @@ package job import ( "github.com/mhsanaei/3x-ui/v3/internal/amneziawg" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" "github.com/mhsanaei/3x-ui/v3/internal/logger" "github.com/mhsanaei/3x-ui/v3/internal/web/service" "github.com/mhsanaei/3x-ui/v3/internal/xray" @@ -84,5 +85,9 @@ func (j *AmneziaWGJob) Run() { } } + // Live speed: AmneziaWG never runs inside xray-core, so XrayTrafficJob's + // own 5s broadcast never mentions these tags. See sidecar_traffic.go. + broadcastSidecarTraffic(string(model.AmneziaWG), traffics, clientTraffics) + j.inboundService.RefreshLocalOnlineClients(onlineEmails, activeTags) } diff --git a/internal/web/job/mtproto_job.go b/internal/web/job/mtproto_job.go index 21ebfaec4..1a71bfc76 100644 --- a/internal/web/job/mtproto_job.go +++ b/internal/web/job/mtproto_job.go @@ -1,6 +1,7 @@ package job import ( + "github.com/mhsanaei/3x-ui/v3/internal/database/model" "github.com/mhsanaei/3x-ui/v3/internal/logger" "github.com/mhsanaei/3x-ui/v3/internal/mtproto" "github.com/mhsanaei/3x-ui/v3/internal/web/service" @@ -77,5 +78,12 @@ func (j *MtprotoJob) Run() { } } + // Live speed: mtproto's mtg sidecar never runs inside xray-core, so + // XrayTrafficJob's own 5s broadcast never mentions these tags. traffics + // here already excludes routed-through-xray inbound tags (existing + // logic above, for cumulative-totals reasons) -- reused as-is, not + // recomputed. See sidecar_traffic.go. + broadcastSidecarTraffic(string(model.MTProto), traffics, clientTraffics) + j.inboundService.RefreshLocalOnlineClients(onlineEmails, activeTags) } diff --git a/internal/web/job/sidecar_traffic.go b/internal/web/job/sidecar_traffic.go new file mode 100644 index 000000000..f0a5086d2 --- /dev/null +++ b/internal/web/job/sidecar_traffic.go @@ -0,0 +1,43 @@ +package job + +import ( + "github.com/mhsanaei/3x-ui/v3/internal/web/websocket" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +// sidecarTrafficPayload builds the websocket broadcast map for one poll of a +// sidecar protocol's traffic delta. "Sidecar" means a protocol that never +// runs inside xray-core's own inbounds -- AmneziaWG's kernel tunnel and +// MTProto's mtg process are the only two today (both explicitly skipped from +// claiming an xray-core tag in GenXrayInboundConfig). XrayTrafficJob's own 5s +// broadcast (xray_traffic_job.go) carries "traffics"/"clientTraffics" for +// xray-native inbounds/clients and never mentions sidecar tags, so reusing +// those keys would let each side's broadcast clobber the other's speed on +// its very next, unrelated tick. protocol namespaces the keys instead: +// "amneziawg" produces "amneziawgTraffics"/"amneziawgClientTraffics", +// "mtproto" produces "mtprotoTraffics"/"mtprotoClientTraffics" -- distinct +// pairs per protocol, so the frontend tracks each independently and never +// has to reconcile two protocols within one map (see useInbounds.ts / +// useClients.ts). +func sidecarTrafficPayload(protocol string, traffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) map[string]any { + return map[string]any{ + protocol + "Traffics": traffics, + protocol + "ClientTraffics": clientTraffics, + } +} + +// broadcastSidecarTraffic pushes this poll's live-speed snapshot for a +// sidecar protocol ("amneziawg" or "mtproto") over the websocket, so its +// inbound/client rows show a live Speed value the same way xray-native rows +// already do. Call this every tick -- even with empty slices -- so a peer +// that just went idle clears to no-speed on the frontend instead of sticking +// at its last nonzero reading; only the AddTraffic/RefreshLocalOnlineClients +// calls around it are conditioned on non-empty data, since cumulative-totals +// accounting doesn't need this signal. Uses BroadcastSidecarTraffic (not +// BroadcastTraffic) -- see its doc comment for why. +func broadcastSidecarTraffic(protocol string, traffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) { + if !websocket.HasClients() { + return + } + websocket.BroadcastSidecarTraffic(sidecarTrafficPayload(protocol, traffics, clientTraffics)) +} diff --git a/internal/web/job/sidecar_traffic_test.go b/internal/web/job/sidecar_traffic_test.go new file mode 100644 index 000000000..9d3a887ce --- /dev/null +++ b/internal/web/job/sidecar_traffic_test.go @@ -0,0 +1,59 @@ +package job + +import ( + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +func TestSidecarTrafficPayload_UsesProtocolNamespacedKeys(t *testing.T) { + traffics := []*xray.Traffic{{IsInbound: true, Tag: "amneziawg-1", Up: 10, Down: 20}} + clientTraffics := []*xray.ClientTraffic{{Email: "a@b.c", Up: 10, Down: 20}} + + payload := sidecarTrafficPayload("amneziawg", traffics, clientTraffics) + + if _, ok := payload["amneziawgTraffics"]; !ok { + t.Fatal("expected amneziawgTraffics key") + } + if _, ok := payload["amneziawgClientTraffics"]; !ok { + t.Fatal("expected amneziawgClientTraffics key") + } + for _, wrongKey := range []string{"traffics", "clientTraffics", "mtprotoTraffics", "mtprotoClientTraffics"} { + if _, ok := payload[wrongKey]; ok { + t.Fatalf("payload must not contain %q -- it would collide with a different broadcast source", wrongKey) + } + } +} + +func TestSidecarTrafficPayload_DistinctProtocolsNamespacedIndependently(t *testing.T) { + amneziawgPayload := sidecarTrafficPayload("amneziawg", nil, nil) + mtprotoPayload := sidecarTrafficPayload("mtproto", nil, nil) + + if _, ok := amneziawgPayload["mtprotoTraffics"]; ok { + t.Fatal("amneziawg payload must not contain mtproto keys") + } + if _, ok := mtprotoPayload["amneziawgTraffics"]; ok { + t.Fatal("mtproto payload must not contain amneziawg keys") + } +} + +func TestSidecarTrafficPayload_EmptyInputsStillProduceBothKeys(t *testing.T) { + // The frontend clears a peer's speed when its tag/email is absent from + // the payload's arrays -- but only if the key itself is present. If the + // key vanished entirely for an idle poll, idle-clearing would never + // trigger and the last nonzero speed would stick forever. + payload := sidecarTrafficPayload("amneziawg", nil, nil) + + if _, ok := payload["amneziawgTraffics"]; !ok { + t.Fatal("expected amneziawgTraffics key present even for nil input") + } + if _, ok := payload["amneziawgClientTraffics"]; !ok { + t.Fatal("expected amneziawgClientTraffics key present even for nil input") + } +} + +func TestBroadcastSidecarTraffic_NoOpWithoutHub(t *testing.T) { + // No global web server/hub is configured in this test binary, so + // websocket.HasClients() is false -- this must return without panicking. + broadcastSidecarTraffic("amneziawg", nil, nil) +} diff --git a/internal/web/websocket/hub.go b/internal/web/websocket/hub.go index 63b336c42..c991a4237 100644 --- a/internal/web/websocket/hub.go +++ b/internal/web/websocket/hub.go @@ -288,6 +288,37 @@ func (h *Hub) Broadcast(messageType MessageType, payload any) { h.enqueue(data) } +// BroadcastUnthrottled behaves like Broadcast but skips the per-type rate +// limit in shouldThrottle. Use for message types whose callers are already +// self-limited by their own poll cadence (e.g. the AmneziaWG/MTProto sidecar +// traffic jobs, both "@every 10s" in internal/web/web.go), so a same-tick +// collision with another caller of the same MessageType (see +// throttledMessageTypes) never silently drops one side. robfig/cron's +// "@every" schedules are lastRun+interval grids anchored at AddJob time, not +// wall-clock-aligned, so two same-cadence jobs registered milliseconds apart +// stay in lockstep indefinitely -- with the throttled path, one of them +// would lose almost every tick, forever, not just occasionally. +func (h *Hub) BroadcastUnthrottled(messageType MessageType, payload any) { + if h == nil || payload == nil || h.GetClientCount() == 0 { + return + } + data, err := json.Marshal(Message{ + Type: messageType, + Payload: payload, + Time: time.Now().UnixMilli(), + }) + if err != nil { + logger.Error("WebSocket marshal failed:", err) + return + } + if len(data) > maxMessageSize { + logger.Debugf("WebSocket payload %d bytes exceeds limit, sending invalidate for %s", len(data), messageType) + h.broadcastInvalidate(messageType) + return + } + h.enqueue(data) +} + // broadcastInvalidate queues a lightweight signal telling clients to re-fetch // the named data type via REST. func (h *Hub) broadcastInvalidate(originalType MessageType) { diff --git a/internal/web/websocket/hub_test.go b/internal/web/websocket/hub_test.go index 231480f4f..f918d44dd 100644 --- a/internal/web/websocket/hub_test.go +++ b/internal/web/websocket/hub_test.go @@ -176,6 +176,77 @@ func TestHub_ShouldThrottle_DistinctTypesIndependent(t *testing.T) { } } +func TestHub_BroadcastUnthrottled_DeliversDespiteHotThrottleWindow(t *testing.T) { + h := NewHub() + defer h.Stop() + go h.Run() + + c := NewClient("c1") + h.Register(c) + waitClientCount(t, h, 1) + + // Consume the throttle window for MessageTypeTraffic, as + // TestHub_ShouldThrottle already does directly. + if h.shouldThrottle(MessageTypeTraffic) { + t.Fatal("first call should not throttle") + } + if !h.shouldThrottle(MessageTypeTraffic) { + t.Fatal("second immediate call should throttle -- window not hot as expected") + } + + h.BroadcastUnthrottled(MessageTypeTraffic, map[string]string{"k": "v"}) + + select { + case raw := <-c.Send: + var m Message + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("payload is not valid JSON: %v\n%s", err, raw) + } + if m.Type != MessageTypeTraffic { + t.Fatalf("Type = %q, want %q", m.Type, MessageTypeTraffic) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("BroadcastUnthrottled should deliver even with a hot throttle window") + } +} + +func TestHub_BroadcastUnthrottled_DropsWhenNoClients(t *testing.T) { + h := NewHub() + defer h.Stop() + go h.Run() + + h.BroadcastUnthrottled(MessageTypeTraffic, "payload") + + select { + case <-h.broadcast: + t.Fatal("BroadcastUnthrottled should drop when client count is zero") + case <-time.After(50 * time.Millisecond): + } +} + +func TestHub_BroadcastUnthrottled_DropsNilPayload(t *testing.T) { + h := NewHub() + defer h.Stop() + go h.Run() + + c := NewClient("c1") + h.Register(c) + waitClientCount(t, h, 1) + + h.BroadcastUnthrottled(MessageTypeTraffic, nil) + + select { + case <-c.Send: + t.Fatal("nil payload should be dropped, not delivered") + case <-time.After(50 * time.Millisecond): + } +} + +func TestHub_BroadcastUnthrottled_NilReceiverDoesNotPanic(t *testing.T) { + var h *Hub + h.BroadcastUnthrottled(MessageTypeTraffic, "anything") +} + func TestTrySend_SucceedsWithRoom(t *testing.T) { c := &Client{ID: "c", Send: make(chan []byte, 1)} if !trySend(c, []byte("hi")) { diff --git a/internal/web/websocket/notifier.go b/internal/web/websocket/notifier.go index 897047e5d..3ceaea0f8 100644 --- a/internal/web/websocket/notifier.go +++ b/internal/web/websocket/notifier.go @@ -45,6 +45,19 @@ func BroadcastTraffic(traffic any) { } } +// BroadcastSidecarTraffic broadcasts an AmneziaWG/MTProto traffic delta under +// the same "traffic" message type BroadcastTraffic uses, but bypasses the +// hub's per-type throttle (see Hub.BroadcastUnthrottled) so the two sidecar +// jobs' independent ~10s broadcasts can never starve each other. The payload +// carries protocol-namespaced keys (see internal/web/job/sidecar_traffic.go), +// so no new frontend message-type wiring is needed -- applyTrafficEvent +// already receives every "traffic" message. +func BroadcastSidecarTraffic(traffic any) { + if hub := GetHub(); hub != nil { + hub.BroadcastUnthrottled(MessageTypeTraffic, traffic) + } +} + // BroadcastClientStats broadcasts absolute per-client traffic counters. Small // installs send the complete row set each cycle (payload key snapshot=true); // above the traffic job's snapshot threshold only the rows active in the