fix(traffic): ограничивать доверие скорости после пропусков

This commit is contained in:
Denozordec committed 2026-10-13 00:54:42 +07:00
1 parent 9603048bd8
commit 493c30bebc
21 files changed
+1367 -22

No files matched your search

+1 -1
View File
@@ -19,7 +19,7 @@
"test:exec": "tsx src/services/mikrotik-logs.test.ts && tsx src/routes/exec.test.ts && tsx src/plugins/auth.exec.test.ts",
"test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts && tsx src/plugins/auth.ospf-optimizer.test.ts",
"test:wireguard": "npx tsx src/services/wireguard-config.test.ts && npx tsx src/services/wireguard-ipv6.test.ts && npx tsx --test src/services/wireguard-import.test.ts src/services/wireguard-import-apply.test.ts src/services/wireguard-import-revision.test.ts src/services/wireguard-delete.test.ts && npx tsx src/routes/wireguard-import.test.ts",
"test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-freshness.test.ts && tsx src/services/traffic-live.test.ts && tsx src/services/traffic-observation-sql.test.ts && tsx src/services/traffic-history-sql.test.ts && tsx src/services/traffic-live-sql.test.ts && tsx src/services/traffic-volume-sql.test.ts && tsx src/services/traffic-binding-status.test.ts && tsx src/services/traffic-status-sql.test.ts",
"test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-freshness.test.ts && tsx src/services/traffic-live.test.ts && tsx src/services/traffic-observation-sql.test.ts && tsx src/services/traffic-history-sql.test.ts && tsx src/services/traffic-live-sql.test.ts && tsx src/services/traffic-volume-sql.test.ts && tsx src/services/traffic-binding-status.test.ts && tsx src/services/traffic-status-sql.test.ts && tsx src/services/traffic-gap.test.ts",
"test:traffic-flow": "tsx src/routes/traffic-flow-query.test.ts && tsx src/services/traffic-flow-month.test.ts && tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-ipv6-pipeline.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-ifindex.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-dest.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-purge.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/services/traffic-flow-facts.test.ts && tsx src/services/traffic-flow-facts-filter.test.ts && tsx src/services/traffic-flow-facts-rebuild.test.ts && tsx src/services/statistics-aggregate.test.ts",
"test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts && tsx src/modules/users/interface-inventory.test.ts && tsx src/routes/users-catalog.test.ts",
"test:pg": "tsx src/db/pg-pool.test.ts && tsx src/db/sql-bind.test.ts && tsx src/db/sqlite-json.test.ts && tsx src/db/traffic-flags.test.ts && tsx src/db/pg-schema.test.ts && tsx src/services/config-revisions.test.ts",
+6 -3
View File
@@ -43,8 +43,9 @@ function sampleRate(
rxBytes: number,
txBytes: number,
nowMs: number,
maxIntervalMs: number,
): { rxBps: number; txBps: number; measured: boolean } {
return observedTrafficRate(prevWave.get(key), rxBytes, txBytes, nowMs)
return observedTrafficRate(prevWave.get(key), rxBytes, txBytes, nowMs, maxIntervalMs)
}
export interface TrafficCollectorState {
@@ -122,7 +123,8 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
}
try {
await getSettingsRow()
const settings = await getSettingsRow()
const maxIntervalMs = Math.max(5, settings?.intervalSec ?? 30) * 3000
const enabledServers = await db.select().from(servers).where(eq(servers.enabled, true))
for (const srv of enabledServers) {
@@ -146,7 +148,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
const txBytes = toNum(i["tx-byte"])
const running = parseTrafficFlag(i.running)
const disabled = parseTrafficFlag(i.disabled)
const { rxBps, txBps, measured } = sampleRate(prevWave, waveKey(interfaceName), rxBytes, txBytes, nowMs)
const { rxBps, txBps, measured } = sampleRate(prevWave, waveKey(interfaceName), rxBytes, txBytes, nowMs, maxIntervalMs)
if (shouldIncludeIface(interfaceName, running, disabled)) {
includedInterfaces++
totalsMeasured &&= measured
@@ -177,6 +179,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
rxBytes,
txBytes,
nowMs,
maxIntervalMs,
)
rows.push({
serverId: srv.id,
+6 -2
View File
@@ -20,12 +20,14 @@ export function mergeTrafficObservations(parts: TrafficObservations[]): TrafficO
return times.length && times.every(Number.isFinite) ? new Date(Math.min(...times)).toISOString() : null
}
const states: TrafficCurrentState[] = ["error", "pending", "stale", "disabled", "unknown"]
const intervals = parts.map(p => p.measurementIntervalMs)
return { rxNow: !!parts.length && parts.every(p => p.rxNow), txNow: !!parts.length && parts.every(p => p.txNow),
rxSeries: Array.from({ length: parts[0]?.rxSeries.length ?? 60 }, (_, i) => !!parts.length && parts.every(p => p.rxSeries[i])),
txSeries: Array.from({ length: parts[0]?.txSeries.length ?? 60 }, (_, i) => !!parts.length && parts.every(p => p.txSeries[i])),
sampledAt: oldest("sampledAt"), expiresAt: oldest("expiresAt"), lastAttemptAt: oldest("lastAttemptAt"), lastSucceededAt: oldest("lastSucceededAt"),
currentState: parts.length ? states.find(state => parts.some(p => p.currentState === state)) ?? "current" : "unknown",
error: parts.find(p => p.error)?.error ?? null,
...(intervals.some(v => v !== undefined) ? {measurementIntervalMs: intervals.every(v => typeof v === "number" && Number.isFinite(v) && v > 0) ? Math.max(...intervals as number[]) : null} : {}),
}
}
export function applyTrafficFreshness(observations: TrafficObservations,
@@ -35,6 +37,8 @@ export function applyTrafficFreshness(observations: TrafficObservations,
const period = Math.max(5, settings?.intervalSec ?? 30) * 1000
const time = (raw: string | Date | null | undefined) => raw instanceof Date ? raw.getTime() : Date.parse(raw ?? "")
const attemptMs = time(attempt?.lastAttemptAt), successMs = time(attempt?.lastSucceededAt)
const interval = observations.measurementIntervalMs
const intervalUnknown = interval !== undefined && (typeof interval !== "number" || !Number.isFinite(interval) || interval <= 0 || interval > 3 * period)
let currentState: TrafficCurrentState = "unknown"
if (!server.enabled || settings?.enabled === false) currentState = "disabled"
else if (settings && attempt && time(attempt.serverUpdatedAt) === time(server.updatedAt)) {
@@ -42,7 +46,7 @@ export function applyTrafficFreshness(observations: TrafficObservations,
else if (attempt.status === "pending") currentState = "pending"
else if (attempt.status === "error") currentState = "error"
else if (attempt.status === "ok" && successMs === attemptMs &&
Date.parse(observations.sampledAt ?? "") === successMs && (observations.rxNow || observations.txNow)) currentState = "current"
!intervalUnknown && Date.parse(observations.sampledAt ?? "") === successMs && (observations.rxNow || observations.txNow)) currentState = "current"
}
return { ...observations,
expiresAt: Number.isFinite(attemptMs + 3 * period) && Math.abs(attemptMs + 3 * period) <= 8.64e15 ? new Date(attemptMs + 3 * period).toISOString() : null,
@@ -50,6 +54,6 @@ export function applyTrafficFreshness(observations: TrafficObservations,
txNow: currentState === "current" && observations.txNow,
currentState, lastAttemptAt: Number.isFinite(attemptMs) ? new Date(attemptMs).toISOString() : null,
lastSucceededAt: Number.isFinite(successMs) ? new Date(successMs).toISOString() : null,
error: attempt?.error ?? null,
error: attempt?.error ?? (intervalUnknown ? "Интервал между измерениями слишком велик или неизвестен. История и учтённый объём сохранены; для текущей скорости нужны соседние свежие измерения." : null),
}
}
+18
View File
@@ -0,0 +1,18 @@
import assert from 'node:assert/strict'
import {buildTrafficFromSamples} from './traffic-rate.js'
import {applyTrafficFreshness,mergeTrafficObservations} from './traffic-freshness.js'
import {observedTrafficRate} from './traffic-observation.js'
const now=Date.parse('2026-10-13T00:00:00Z'),iso=new Date(now).toISOString(),server={enabled:true,updatedAt:iso},settings={enabled:true,intervalSec:30},attempt={serverUpdatedAt:iso,lastAttemptAt:iso,lastSucceededAt:iso,status:'ok',error:null}
for(const dt of[30_000,90_000,90_001,3600_000,86400_000]){
const rows=[0,1].map(i=>({interfaceName:'ether1',sampledAt:new Date(now-(1-i)*dt).toISOString(),rxBytes:i*1024**2,txBytes:i*1024**2,rxBps:0,txBps:0,running:true,disabled:false,countersKnown:true}))
const built=buildTrafficFromSamples(rows,now-86400_000,now,undefined,undefined,true),o=applyTrafficFreshness(built.observations!,server,settings,attempt,now)
assert.equal(o.rxNow,dt<=90_000);assert.equal(o.txNow,dt<=90_000);assert.equal(built.rxTotalGiB,1/1024);assert(o.rxSeries.some(Boolean));assert.equal(o.measurementIntervalMs,dt)
const rate=observedTrafficRate(rows[0],rows[1].rxBytes,rows[1].txBytes,now,90_000);assert.equal(rate.measured,dt<=90_000)
if(dt>90_000)assert(o.error?.includes('Интервал'))
}
const base={rxNow:true,txNow:true,rxSeries:[true],txSeries:[true],sampledAt:iso}
for(const interval of[null,0,-1,NaN,Infinity])assert.equal(applyTrafficFreshness({...base,measurementIntervalMs:interval},server,settings,attempt,now).rxNow,false)
assert.equal(mergeTrafficObservations([{...base,measurementIntervalMs:30_000},{...base,measurementIntervalMs:120_000}]).measurementIntervalMs,120_000)
assert.equal(mergeTrafficObservations([{...base,measurementIntervalMs:30_000},base]).measurementIntervalMs,null)
assert.equal(applyTrafficFreshness({...base,measurementIntervalMs:120_000},server,settings,{...attempt,status:'error',error:'401'},now).error,'401')
console.log('traffic gap threshold/history/volume/merge/collector guards passed')
+2 -2
View File
@@ -17,10 +17,10 @@ export function parseTrafficCounter(raw: unknown): number {
export function observedTrafficRate(
previous: { rxBytes: number; txBytes: number; sampledAt: string } | undefined,
rxBytes: number, txBytes: number, nowMs: number,
rxBytes: number, txBytes: number, nowMs: number, maxIntervalMs = Infinity,
): { rxBps: number; txBps: number; measured: boolean } {
const prevMs = previous ? Date.parse(previous.sampledAt) : NaN
if (!previous || ![previous.rxBytes, previous.txBytes, rxBytes, txBytes].every(v => Number.isSafeInteger(v) && v >= 0)) {
if (!previous || nowMs - prevMs > maxIntervalMs || ![previous.rxBytes, previous.txBytes, rxBytes, txBytes].every(v => Number.isSafeInteger(v) && v >= 0)) {
return { rxBps: 0, txBps: 0, measured: false }
}
const rx = previous ? rateBpsFromDelta(previous.rxBytes, rxBytes, prevMs, nowMs) : null
+5 -3
View File
@@ -22,7 +22,7 @@ export interface RatePoint {
txMbps: number
}
export interface TrafficObservations {rxNow:boolean;txNow:boolean;rxSeries:boolean[];txSeries:boolean[];sampledAt:string|null;expiresAt?:string|null;
export interface TrafficObservations {rxNow:boolean;txNow:boolean;rxSeries:boolean[];txSeries:boolean[];sampledAt:string|null;expiresAt?:string|null;measurementIntervalMs?:number|null;
currentState?: import("./traffic-freshness.js").TrafficCurrentState;lastAttemptAt?:string|null;lastSucceededAt?:string|null;error?:string|null}
export interface BuiltTrafficSeries {
@@ -143,7 +143,7 @@ export function buildTrafficFromSamples(
const rxPoints: Array<{ t: number; v: number }> = []
const txPoints: Array<{ t: number; v: number }> = []
const byTs = new Map<number, { rx: number; tx: number;rxObserved:boolean;txObserved:boolean;count:number }>()
const byTs = new Map<number, { rx: number; tx: number;rxObserved:boolean;txObserved:boolean;count:number;intervalMs:number }>()
let includedSeries=0,lastAttemptAt:number|null=null
let rxBytesDelta = 0
@@ -190,7 +190,8 @@ export function buildTrafficFromSamples(
if (rxBps == null && txBps == null) continue
const rxMbps = bpsToMbps(rxBps ?? 0)
const txMbps = bpsToMbps(txBps ?? 0)
const acc = byTs.get(t1) ?? { rx: 0, tx: 0,rxObserved:true,txObserved:true,count:0 }
const acc = byTs.get(t1) ?? { rx: 0, tx: 0,rxObserved:true,txObserved:true,count:0,intervalMs:0 }
acc.intervalMs = Math.max(acc.intervalMs, t1 - t0)
acc.count++;acc.rxObserved&&=rxBps!==null;acc.txObserved&&=txBps!==null
acc.rx += rxMbps
acc.tx += txMbps
@@ -226,6 +227,7 @@ export function buildTrafficFromSamples(
rxSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.rxObserved&&v.count===includedSeries&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
txSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.txObserved&&v.count===includedSeries&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
sampledAt:lastAttemptAt===null?null:new Date(lastAttemptAt).toISOString(),
measurementIntervalMs:lastTs===lastAttemptAt?last?.intervalMs??null:null,
}}:{}),
}
}
@@ -35,6 +35,10 @@ else{
denyPeer=true;peerVariant="valid";let partial=await collectTrafficOnce();assert.equal(partial.servers[0].ok,true);assert(partial.servers[0].warning?.includes("401"));rows=await bindings();assert(rows.find((r:{interfaceName:string})=>r.interfaceName==="ether1").observations.error.includes("401"));cases++
const manualPartial=await app.inject({method:"POST",url:"/traffic/collect-now"});assert.equal(manualPartial.statusCode,200);assert.equal(manualPartial.json().ok,true);assert.equal(manualPartial.json().warnings.length,1);assert(manualPartial.json().lastError.includes("401"));
denyPeer=false;partial=await collectTrafficOnce();assert.equal(partial.servers[0].warning,undefined);assert.equal((await pool.query("SELECT error FROM traffic_collection_state WHERE server_id=$1",[serverId])).rows[0].error,null);cases++
await pool.query("DELETE FROM traffic_samples WHERE server_id=$1 AND sampled_at<>(SELECT max(sampled_at) FROM traffic_samples WHERE server_id=$1)",[serverId]);
await pool.query("UPDATE traffic_samples SET sampled_at=$1 WHERE server_id=$2",[new Date(Date.now()-120_000).toISOString(),serverId]);
let gap=await collectTrafficOnce();assert.equal(gap.servers[0].ok,true);assert.equal(gap.servers[0].sumRxMbps,undefined);rows=await bindings();assert.equal(rows.find((r:{interfaceName:string})=>r.interfaceName==="ether1").observations.rxNow,false);
gap=await collectTrafficOnce();assert.equal(gap.servers[0].sumRxMbps,0);rows=await bindings();assert.equal(rows.find((r:{interfaceName:string})=>r.interfaceName==="ether1").observations.rxNow,true);cases++;
await pool.query("UPDATE servers SET enabled=false WHERE id=$1",[serverId]);rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
console.log(`traffic status owned SQL: ${cases} cases; first-wave native flags/peer config/invalid/error/legacy/stale/disabled passed`)
}finally{await app.close();await new Promise<void>(resolve=>http.close(()=>resolve()));await pool.query("DELETE FROM app_users WHERE id=$1",[userId]);if(serverId!==undefined){await pool.query("DELETE FROM traffic_samples WHERE server_id=$1",[serverId]);await pool.query("DELETE FROM servers WHERE id=$1",[serverId])}await closePool()}
@@ -42,6 +42,10 @@ else {
s=(await get(`/traffic/servers/${serverId}?range=1h`)).server;assert.equal(s.rxTotal,2048/1024**3);assert(s.observations.rxSeries.some(Boolean));assert(s.rxPeak>0);assert.equal(s.sessions,1);
const summary=(await get("/traffic/servers?range=1h")).servers.find((r:{id:string})=>r.id===String(serverId));assert.equal(summary.rxTotal,s.rxTotal);cases++
}
await pool.query("DELETE FROM traffic_samples WHERE server_id=$1",[serverId]);await put("ether1",-120_000,0);await put("ether1",0,1024**2);
s=(await get(`/traffic/servers/${serverId}?range=1h`)).server;assert.equal(s.rxTotal,1/1024);assert.equal(s.observations.rxNow,false);assert.equal(s.observations.measurementIntervalMs,120_000);assert(s.observations.error.includes("Интервал"));assert(s.observations.rxSeries.some(Boolean));cases++;
await pool.query("UPDATE traffic_samples SET sampled_at=$1 WHERE server_id=$2 AND sampled_at=$3",[new Date(Date.parse(at)-30_000).toISOString(),serverId,at]);await put("ether1",0,2*1024**2);
s=(await get(`/traffic/servers/${serverId}?range=1h`)).server;assert.equal(s.observations.rxNow,true);assert.equal(s.observations.measurementIntervalMs,30_000);assert.equal(s.rxTotal,2/1024);cases++;
console.log(`traffic volume owned SQL: ${cases} cases; exact bytes/binary unit/server/user/binding/100-part aggregate/reset passed`)
}finally{await app.close();await pool.query("DELETE FROM app_users WHERE id=$1",[userId]);if(serverId!==undefined){await pool.query("DELETE FROM traffic_samples WHERE server_id=$1",[serverId]);await pool.query("DELETE FROM servers WHERE id=$1",[serverId])}await closePool()}
}