mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-09-09 11:48:18 +00:00
fix(traffic): show live Speed for AmneziaWG and MTProto inbounds/clients
The Speed column showed "--" for AmneziaWG (and MTProto, which has the identical gap) even while cumulative traffic totals were correct. XrayTrafficJob drives live speed by querying xray-core's own stats API and broadcasting the delta over websocket -- but AmneziaWG/MTProto never run inside xray-core's own runtime inbounds, so they're invisible to that API. Their own jobs already compute the same per-poll delta shape (that's what keeps cumulative totals correct) but never broadcast it. Reusing the existing "traffics"/"clientTraffics" broadcast would have two real bugs: the frontend's existing scope/replace logic would let each side clobber the other's speed on its next unrelated tick, and the websocket hub's per-message-type throttle is keyed only by message type, not caller -- since both sidecar jobs run on identical "@every 10s" grids registered milliseconds apart, one would silently lose almost every broadcast if both protocols were ever configured together. Fixed with a small unthrottled broadcast path (both sidecar jobs are already self-rate-limited by their own cron cadence) and protocol- namespaced wire keys, tracked in their own frontend state and merged into the existing inboundSpeed/clientSpeed only at read time -- so every existing consumer needs zero changes. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -32,7 +32,7 @@ import {
|
|||||||
type BulkDetachResult,
|
type BulkDetachResult,
|
||||||
} from '@/schemas/client';
|
} from '@/schemas/client';
|
||||||
import { DefaultsPayloadSchema } from '@/schemas/defaults';
|
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.
|
// One row sent to POST /clients/:email/externalLinks.
|
||||||
export type ExternalLinkInput = { kind: 'link' | 'subscription'; value: string; remark: string };
|
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.
|
// the server's authoritative total for the headline count.
|
||||||
const [allClientStats, setAllClientStats] = useState<ClientStatRow[]>([]);
|
const [allClientStats, setAllClientStats] = useState<ClientStatRow[]>([]);
|
||||||
const [clientSpeed, setClientSpeed] = useState<Record<string, ClientSpeedEntry>>({});
|
const [clientSpeed, setClientSpeed] = useState<Record<string, ClientSpeedEntry>>({});
|
||||||
|
// 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<Record<string, ClientSpeedEntry>>({});
|
||||||
|
const [mtprotoClientSpeed, setMtprotoClientSpeed] = useState<Record<string, ClientSpeedEntry>>({});
|
||||||
const summary = useMemo<ClientsSummary>(() => {
|
const summary = useMemo<ClientsSummary>(() => {
|
||||||
const serverSummary = listQuery.data?.summary ?? DEFAULT_SUMMARY;
|
const serverSummary = listQuery.data?.summary ?? DEFAULT_SUMMARY;
|
||||||
if (allClientStats.length === 0) return serverSummary;
|
if (allClientStats.length === 0) return serverSummary;
|
||||||
@@ -553,6 +565,8 @@ export function useClients() {
|
|||||||
const p = payload as {
|
const p = payload as {
|
||||||
onlineClients?: string[];
|
onlineClients?: string[];
|
||||||
clientTraffics?: { email: string; up: number; down: number }[];
|
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)) {
|
if (Array.isArray(p.onlineClients)) {
|
||||||
queryClient.setQueryData(keys.clients.onlines(), p.onlineClients);
|
queryClient.setQueryData(keys.clients.onlines(), p.onlineClients);
|
||||||
@@ -568,6 +582,28 @@ export function useClients() {
|
|||||||
}
|
}
|
||||||
setClientSpeed(next);
|
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<string, ClientSpeedEntry>) => void,
|
||||||
|
) => {
|
||||||
|
const next: Record<string, ClientSpeedEntry> = {};
|
||||||
|
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]);
|
}, [queryClient]);
|
||||||
|
|
||||||
const applyClientStatsEvent = useCallback((payload: unknown) => {
|
const applyClientStatsEvent = useCallback((payload: unknown) => {
|
||||||
@@ -606,6 +642,16 @@ export function useClients() {
|
|||||||
queryRef.current = query;
|
queryRef.current = query;
|
||||||
}, [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 {
|
return {
|
||||||
clients,
|
clients,
|
||||||
total,
|
total,
|
||||||
@@ -650,7 +696,7 @@ export function useClients() {
|
|||||||
exportClients,
|
exportClients,
|
||||||
importClients,
|
importClients,
|
||||||
setEnable,
|
setEnable,
|
||||||
clientSpeed,
|
clientSpeed: clientSpeedOut,
|
||||||
applyTrafficEvent,
|
applyTrafficEvent,
|
||||||
applyClientStatsEvent,
|
applyClientStatsEvent,
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -1 +1,5 @@
|
|||||||
export const TRAFFIC_POLL_INTERVAL_S = 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;
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ import { OnlinesSchema, OnlineByNodeSchema, ActiveInboundsByNodeSchema } from '@
|
|||||||
import { DefaultsPayloadSchema, type DefaultsPayload } from '@/schemas/defaults';
|
import { DefaultsPayloadSchema, type DefaultsPayload } from '@/schemas/defaults';
|
||||||
|
|
||||||
import type { InboundSpeedEntry } from './list/types';
|
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 {
|
export interface SubSettings {
|
||||||
enable: boolean;
|
enable: boolean;
|
||||||
@@ -206,6 +206,16 @@ export function useInbounds() {
|
|||||||
inboundSpeedCache = { at: Date.now(), data: inboundSpeed };
|
inboundSpeedCache = { at: Date.now(), data: inboundSpeed };
|
||||||
}, [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<Record<number, InboundSpeedEntry>>({});
|
||||||
|
const [mtprotoInboundSpeed, setMtprotoInboundSpeed] = useState<Record<number, InboundSpeedEntry>>({});
|
||||||
|
|
||||||
const [onlineClients, setOnlineClients] = useState<string[]>([]);
|
const [onlineClients, setOnlineClients] = useState<string[]>([]);
|
||||||
const onlineClientsRef = useRef<string[]>([]);
|
const onlineClientsRef = useRef<string[]>([]);
|
||||||
onlineClientsRef.current = onlineClients;
|
onlineClientsRef.current = onlineClients;
|
||||||
@@ -413,6 +423,8 @@ export function useInbounds() {
|
|||||||
const p = payload as {
|
const p = payload as {
|
||||||
traffics?: TrafficDelta[];
|
traffics?: TrafficDelta[];
|
||||||
nodeTraffics?: TrafficDelta[];
|
nodeTraffics?: TrafficDelta[];
|
||||||
|
amneziawgTraffics?: TrafficDelta[];
|
||||||
|
mtprotoTraffics?: TrafficDelta[];
|
||||||
onlineClients?: string[];
|
onlineClients?: string[];
|
||||||
onlineByGuid?: Record<string, string[]>;
|
onlineByGuid?: Record<string, string[]>;
|
||||||
activeInbounds?: Record<string, string[]>;
|
activeInbounds?: Record<string, string[]>;
|
||||||
@@ -465,6 +477,48 @@ export function useInbounds() {
|
|||||||
};
|
};
|
||||||
if (Array.isArray(p.traffics)) applyTraffics(p.traffics, (ib) => ib.nodeId == null);
|
if (Array.isArray(p.traffics)) applyTraffics(p.traffics, (ib) => ib.nodeId == null);
|
||||||
if (Array.isArray(p.nodeTraffics)) applyTraffics(p.nodeTraffics, (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<number, InboundSpeedEntry>) => Record<number, InboundSpeedEntry>) => void,
|
||||||
|
) => {
|
||||||
|
const byTag = new Map<string, TrafficDelta>();
|
||||||
|
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();
|
||||||
},
|
},
|
||||||
[rebuildClientCount],
|
[rebuildClientCount],
|
||||||
@@ -542,6 +596,16 @@ export function useInbounds() {
|
|||||||
return { up, down };
|
return { up, down };
|
||||||
}, [dbInbounds]);
|
}, [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 {
|
return {
|
||||||
fetched,
|
fetched,
|
||||||
fetchError,
|
fetchError,
|
||||||
@@ -549,7 +613,7 @@ export function useInbounds() {
|
|||||||
clientCount,
|
clientCount,
|
||||||
onlineClients,
|
onlineClients,
|
||||||
lastOnlineMap,
|
lastOnlineMap,
|
||||||
inboundSpeed,
|
inboundSpeed: inboundSpeedOut,
|
||||||
statsVersion,
|
statsVersion,
|
||||||
totals,
|
totals,
|
||||||
expireDiff,
|
expireDiff,
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package job
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"github.com/mhsanaei/3x-ui/v3/internal/amneziawg"
|
"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/logger"
|
||||||
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
|
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
|
||||||
"github.com/mhsanaei/3x-ui/v3/internal/xray"
|
"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)
|
j.inboundService.RefreshLocalOnlineClients(onlineEmails, activeTags)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package job
|
package job
|
||||||
|
|
||||||
import (
|
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/logger"
|
||||||
"github.com/mhsanaei/3x-ui/v3/internal/mtproto"
|
"github.com/mhsanaei/3x-ui/v3/internal/mtproto"
|
||||||
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
|
"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)
|
j.inboundService.RefreshLocalOnlineClients(onlineEmails, activeTags)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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))
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
@@ -288,6 +288,37 @@ func (h *Hub) Broadcast(messageType MessageType, payload any) {
|
|||||||
h.enqueue(data)
|
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
|
// broadcastInvalidate queues a lightweight signal telling clients to re-fetch
|
||||||
// the named data type via REST.
|
// the named data type via REST.
|
||||||
func (h *Hub) broadcastInvalidate(originalType MessageType) {
|
func (h *Hub) broadcastInvalidate(originalType MessageType) {
|
||||||
|
|||||||
@@ -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) {
|
func TestTrySend_SucceedsWithRoom(t *testing.T) {
|
||||||
c := &Client{ID: "c", Send: make(chan []byte, 1)}
|
c := &Client{ID: "c", Send: make(chan []byte, 1)}
|
||||||
if !trySend(c, []byte("hi")) {
|
if !trySend(c, []byte("hi")) {
|
||||||
|
|||||||
@@ -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
|
// BroadcastClientStats broadcasts absolute per-client traffic counters. Small
|
||||||
// installs send the complete row set each cycle (payload key snapshot=true);
|
// 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
|
// above the traffic job's snapshot threshold only the rows active in the
|
||||||
|
|||||||
Reference in New Issue
Block a user