fix(traffic): проверять live-срез и прекращать опрос выключенных устройств

backend/src/routes/traffic.ts

@@ -12,7 +12,8 @@ import {
 } from "../services/traffic-collector.js"
 import { refreshScheduler, runSchedulerJobNow } from "../services/scheduler.js"
 import { appendEvent } from "../modules/events/service/events-service.js"
-import { MikrotikClient } from "../services/mikrotik.js"
+import { readTrafficLiveSample, TrafficLiveSourceChanged } from "../services/traffic-live.js"
+import { MikrotikClient, MikrotikError } from "../services/mikrotik.js"
 import { getEnabledServerById } from "../services/wireguard-live.js"
 import { readTrafficCollectionContext } from "../services/traffic-collection-state.js"
 import { applyTrafficFreshness } from "../services/traffic-freshness.js"
@@ -20,7 +21,6 @@ import {
   buildTrafficFromSamples,
   type TrafficObservations,
   isLoopbackName,
-  parseMonitorTraffic,
 } from "../services/traffic-rate.js"
 import {
   buildBoundInterfaceTraffic,
@@ -54,7 +54,6 @@ interface TrafficInterfaceDto {
 }
 
 const LIVE_TICK_MS = 1500
-const LIVE_ROS_TIMEOUT_MS = 4000
 
 async function latestSnapshot(serverId: number) {
   return (await db
@@ -141,68 +140,6 @@ function sleep(ms: number, signal: AbortSignal): Promise<void> {
   })
 }
 
-function flattenMonitor(raw: unknown): unknown[] {
-  if (Array.isArray(raw)) return raw
-  if (raw != null) return [raw]
-  return []
-}
-
-async function listRunningIfaceNames(client: MikrotikClient): Promise<string[]> {
-  const ifaces = await client.get<Array<{ name?: string; running?: string; disabled?: string }>>(
-    "/interface",
-    LIVE_ROS_TIMEOUT_MS,
-  )
-  return ifaces
-    .filter((i) => (i.running ?? "false") === "true"
-      && (i.disabled ?? "false") !== "true"
-      && !isLoopbackName(i.name ?? ""))
-    .map((i) => i.name ?? "")
-    .filter(Boolean)
-}
-
-async function monitorTrafficOnce(
-  client: MikrotikClient,
-  onlyInterface: string | undefined,
-  cache: { names: string[]; joinedFailed: boolean },
-  signal: AbortSignal,
-): Promise<unknown> {
-  if (onlyInterface) {
-    return client.post(
-      "/interface/monitor-traffic",
-      { interface: onlyInterface, once: "" },
-      LIVE_ROS_TIMEOUT_MS,
-      signal,
-    )
-  }
-  if (cache.names.length === 0) {
-    cache.names = await listRunningIfaceNames(client)
-  }
-  if (cache.names.length === 0) return []
-  if (!cache.joinedFailed) {
-    try {
-      return await client.post(
-        "/interface/monitor-traffic",
-        { interface: cache.names.join(","), once: "" },
-        LIVE_ROS_TIMEOUT_MS,
-        signal,
-      )
-    } catch {
-      cache.joinedFailed = true
-    }
-  }
-  const chunks = await Promise.all(
-    cache.names.map((name) =>
-      client.post(
-        "/interface/monitor-traffic",
-        { interface: name, once: "" },
-        LIVE_ROS_TIMEOUT_MS,
-        signal,
-      ).then(flattenMonitor).catch(() => [] as unknown[]),
-    ),
-  )
-  return chunks.flat()
-}
-
 const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
   app.get("/traffic/settings", async (_req, reply) => {
     const settings = await getTrafficSettings()
@@ -344,13 +281,17 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
     try {
       while (!abort.signal.aborted) {
         try {
-          const raw = await monitorTrafficOnce(client, onlyInterface, cache, abort.signal)
-          const sample = parseMonitorTraffic(raw, { onlyInterface })
+          const sample = await readTrafficLiveSample(client, onlyInterface, cache, abort.signal, async () => {
+            const current = await getEnabledServerById(server.id)
+            if (!current) throw new TrafficLiveSourceChanged("Устройство выключено или удалено. Live-поток остановлен")
+            if (Date.parse(current.updatedAt) !== Date.parse(server.updatedAt)) throw new TrafficLiveSourceChanged("Конфигурация устройства изменилась. Переподключите live-поток")
+          })
           writeSse(reply.raw, "sample", sample)
         } catch (error) {
           if (abort.signal.aborted) break
           const msg = error instanceof Error ? error.message : String(error)
-          writeSse(reply.raw, "error", { error: msg })
+          writeSse(reply.raw, "error", { error: msg.slice(0,1024) })
+          if (error instanceof TrafficLiveSourceChanged || error instanceof MikrotikError && [401,403].includes(error.statusCode)) break
         }
         await sleep(LIVE_TICK_MS, abort.signal)
       }

backend/src/services/traffic-collector.ts

@@ -157,8 +157,8 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
             sampledAt: now,
             rxBytes,
             txBytes,
-            rxBps,
-            txBps,
+            rxBps: Math.round(rxBps),
+            txBps: Math.round(txBps),
             flags: encodeTrafficFlags(running, disabled, true),
           }
         })
@@ -186,8 +186,8 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
               sampledAt: now,
               rxBytes,
               txBytes,
-              rxBps,
-              txBps,
+              rxBps: Math.round(rxBps),
+              txBps: Math.round(txBps),
               flags: encodeTrafficFlags(running, disabled, true),
             })
           }
@@ -205,8 +205,8 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
           ok: true,
           interfaces: ifaces.length,
           ...(includedInterfaces > 0 && totalsMeasured ? {
-            sumRxMbps: Math.round(sumRxMbps * 1000) / 1000,
-            sumTxMbps: Math.round(sumTxMbps * 1000) / 1000,
+            sumRxMbps,
+            sumTxMbps,
           } : {}),
         })
       } catch (err) {

backend/src/services/traffic-live-sql.test.ts

@@ -0,0 +1,56 @@
+import assert from "node:assert/strict"
+import { createServer } from "node:http"
+import Fastify from "fastify"
+import { pool, closePool } from "../db/index.js"
+import { initDatabase } from "../db/bootstrap.js"
+import trafficRoutes from "../routes/traffic.js"
+
+const database=process.env.MMAPP_OWN_SQL_DB
+if(!database) console.log("traffic live SQL: skipped; explicit owned database required")
+else {
+  assert(/^mm_accept_166_\d+$/.test(database));assert.equal(new URL(process.env.DATABASE_URL!).pathname,"/"+database)
+  assert.equal((await pool.query("SELECT current_database() AS name")).rows[0].name,database)
+  const prefix="own-live179-"+Date.now()+"-"
+  let serverId:number|undefined,mode="zero",nativeRequests=0,cases=0,two=false
+  const point=(name:string)=>({name,"rx-bits-per-second":"0","tx-bits-per-second":"0"})
+  const http=createServer(async(req,res)=>{
+    nativeRequests++;res.setHeader("Content-Type","application/json")
+    if(req.method==="GET")return res.end(JSON.stringify((two?["ether1","ether2"]:["ether1"]).map(name=>({name,running:"true",disabled:"false"}))))
+    let data="";for await(const chunk of req)data+=chunk;const name=JSON.parse(data).interface
+    if(mode==="auth"||mode==="fallback-deny"&&name==="ether2"){res.statusCode=401;return res.end('{"message":"Unauthorized"}')}
+    if(mode.startsWith("fallback")&&name.includes(",")){res.statusCode=400;return res.end('{"message":"Unsupported joined interface"}')}
+    if(mode==="empty")return res.end("[]")
+    if(mode==="malformed")return res.end(JSON.stringify([{name,"rx-bits-per-second":"12bad","tx-bits-per-second":"0"}]))
+    if(mode==="partial")return res.end(JSON.stringify([point("ether1")]))
+    res.end(JSON.stringify(name.split(",").map(point)))
+  })
+  await new Promise<void>(resolve=>http.listen(0,"127.0.0.1",resolve));const address=http.address();assert(address&&typeof address==="object")
+  const app=Fastify();await app.register(trafficRoutes)
+  const streams=new Set<AbortController>()
+  async function open() {
+    const ac=new AbortController();streams.add(ac);const r=await fetch(app.listeningOrigin+`/traffic/servers/${serverId}/live`,{signal:ac.signal});assert.equal(r.status,200);assert(r.body)
+    const reader=r.body.getReader(),decoder=new TextDecoder();let buffer=""
+    return {ac, async next(){const timeout=setTimeout(()=>ac.abort(),8000);try{
+      for(;;){const end=buffer.indexOf("\n\n");if(end>=0){const block=buffer.slice(0,end);buffer=buffer.slice(end+2);const event=block.match(/^event: (.*)$/m)?.[1],data=block.match(/^data: (.*)$/m)?.[1];if(event&&data)return{event,data:JSON.parse(data)};continue}
+        const chunk=await reader.read();if(chunk.done)return null;buffer+=decoder.decode(chunk.value,{stream:true})}
+    }finally{clearTimeout(timeout)}},close(){ac.abort();streams.delete(ac)}}
+  }
+  try {
+    await initDatabase();await app.listen({host:"127.0.0.1",port:0})
+    serverId=Number((await pool.query("INSERT INTO servers(name,host,port,use_ssl,enabled,username,password,updated_at) VALUES($1,'127.0.0.1',$2,false,true,'fixture','fixture',$3) RETURNING id",[prefix+"server",address.port,new Date(Date.now()-3_600_000).toISOString()])).rows[0].id)
+    let stream=await open(),event=await stream.next();assert.equal(event?.event,"sample");assert.equal(event?.data.rxMbps,0);assert.equal(event?.data.timestampKind,"backend-receipt");stream.close();cases++
+    for(const kind of ["empty","malformed","partial"]){mode=kind;two=kind==="partial";stream=await open();event=await stream.next();assert.equal(event?.event,"error");stream.close();cases++}
+    mode="fallback";two=true;stream=await open();event=await stream.next();assert.equal(event?.event,"sample");assert.equal(event?.data.rxMbps,0);stream.close();cases++
+    mode="fallback-deny";stream=await open();event=await stream.next();assert.equal(event?.event,"error");assert.match(event?.data.error,/401/);assert.equal(await stream.next(),null);stream.close();cases++
+    mode="auth";two=false;stream=await open();event=await stream.next();assert.equal(event?.event,"error");assert.equal(await stream.next(),null);stream.close();cases++
+    mode="zero";stream=await open();await stream.next();const beforeDisable=nativeRequests
+    await pool.query("UPDATE servers SET enabled=false WHERE id=$1",[serverId]);event=await stream.next();assert.equal(event?.event,"error");assert.match(event?.data.error,/выключено/);assert.equal(await stream.next(),null);assert.equal(nativeRequests,beforeDisable);stream.close();cases++
+    const beforeInitial=nativeRequests;const disabled=await app.inject({method:"GET",url:`/traffic/servers/${serverId}/live`});assert.equal(disabled.statusCode,404);assert.equal(nativeRequests,beforeInitial);cases++
+    await pool.query("UPDATE servers SET enabled=true WHERE id=$1",[serverId]);stream=await open();await stream.next();const beforeConfig=nativeRequests
+    await pool.query("UPDATE servers SET updated_at=$1 WHERE id=$2",[new Date(Date.now()+1000).toISOString(),serverId]);event=await stream.next();assert.match(event?.data.error,/Конфигурация/);assert.equal(await stream.next(),null);assert.equal(nativeRequests,beforeConfig);stream.close();cases++
+    console.log(`traffic live owned SQL: ${cases} cases; real TCP SSE/HTTP/zero/empty/malformed/partial/fallback/auth/disabled/config passed`)
+  }finally{
+    for(const ac of streams)ac.abort();await app.close();await new Promise<void>(resolve=>http.close(()=>resolve()))
+    if(serverId!==undefined)await pool.query("DELETE FROM servers WHERE id=$1",[serverId]);await closePool()
+  }
+}

backend/src/services/traffic-live.test.ts

@@ -0,0 +1,37 @@
+import assert from "node:assert/strict"
+import { parseMonitorTraffic, rateBpsFromDelta, bpsToMbps } from "./traffic-rate.js"
+import { readTrafficLiveSample, TrafficLiveSourceChanged } from "./traffic-live.js"
+import { MikrotikError, type MikrotikClient } from "./mikrotik.js"
+const point=(name="ether1",rx:unknown="0",tx:unknown="0")=>({name,"rx-bits-per-second":rx,"tx-bits-per-second":tx})
+for(const raw of [[],{},null,[null],[{...point(),"rx-bits-per-second":undefined}], [point("ether1","12bad")],[point("ether1",-1)], [point("ether1"),point("ether1")]]) {
+  assert.throws(()=>parseMonitorTraffic(raw))
+}
+assert.throws(()=>parseMonitorTraffic([point()],{expectedInterfaces:["ether1","ether2"]}))
+assert.throws(()=>parseMonitorTraffic([point("other")],{onlyInterface:"ether1"}))
+assert.equal(parseMonitorTraffic([point()]).rxMbps,0)
+assert.equal(parseMonitorTraffic([point("ether1","1")]).rxMbps,0.000001)
+assert(rateBpsFromDelta(1000,1001,0,30_000)!>0);assert(bpsToMbps(rateBpsFromDelta(1000,1001,0,30_000)!)>0)
+assert.equal(parseMonitorTraffic([point()]).timestampKind,"backend-receipt")
+let mode="fallback",postNames:string[]=[],guards=0
+const client={get:async()=>[{name:"ether1",running:"true",disabled:"false"},{name:"ether2",running:true,disabled:false}],
+  post:async(_path:string,body:{interface:string})=>{postNames.push(body.interface);if(body.interface.includes(",")){
+    if(mode==="auth")throw new MikrotikError(401,"/interface/monitor-traffic","Unauthorized")
+    if(mode==="partial")return[point()]
+    throw new MikrotikError(400,"/interface/monitor-traffic","Unsupported joined interface")
+  }if(mode==="deny"&&body.interface==="ether2")throw new MikrotikError(401,"/interface/monitor-traffic","Unauthorized");return[point(body.interface,"1")]}}
+const cast=client as unknown as Pick<MikrotikClient,"get"|"post">
+const run=()=>readTrafficLiveSample(cast,undefined,{names:[],joinedFailed:false},new AbortController().signal,async()=>{guards++})
+const complete=await run();assert.equal(complete.rxMbps,0.000002);assert(guards>=4)
+mode="deny";await assert.rejects(run(),/401/)
+mode="auth";postNames=[];await assert.rejects(run(),/401/);assert.equal(postNames.length,1)
+mode="partial";postNames=[];await assert.rejects(run(),/все live/);assert.equal(postNames.length,1)
+await assert.rejects(readTrafficLiveSample(cast,undefined,{names:[],joinedFailed:false},new AbortController().signal,async()=>{throw new TrafficLiveSourceChanged("disabled")}),/disabled/)
+const abort=new AbortController();abort.abort();postNames=[];await assert.rejects(readTrafficLiveSample(cast,"ether1",{names:[],joinedFailed:false},abort.signal,async()=>{}));assert.equal(postNames.length,0)
+let active=0,maximum=0
+const many={get:async()=>Array.from({length:10},(_,i)=>({name:`ether${i}`,running:true,disabled:false})),post:async(_path:string,body:{interface:string})=>{
+  if(body.interface.includes(","))throw new MikrotikError(400,"/interface/monitor-traffic","Joined unsupported")
+  active++;maximum=Math.max(maximum,active);await new Promise(resolve=>setTimeout(resolve,5));active--;return[point(body.interface)]
+}} as unknown as Pick<MikrotikClient,"get"|"post">
+const bounded=await readTrafficLiveSample(many,undefined,{names:[],joinedFailed:false},new AbortController().signal,async()=>{})
+assert.equal(bounded.rxMbps,0);assert.equal(maximum,4)
+console.log("traffic live completeness/fallback/auth/source/abort/positive-small-rate tests ok")

backend/src/services/traffic-live.ts

@@ -0,0 +1,73 @@
+import { MikrotikError, type MikrotikClient } from "./mikrotik.js"
+import { isLoopbackName, parseMonitorTraffic, type MonitorLiveSample } from "./traffic-rate.js"
+
+const TIMEOUT_MS = 4000
+export class TrafficLiveSourceChanged extends Error {}
+export interface TrafficLiveCache { names: string[]; joinedFailed: boolean }
+type Client = Pick<MikrotikClient, "get" | "post">
+
+/** Never emit a partial sum, and check the source before every new native request. */
+export async function readTrafficLiveSample(client: Client, onlyInterface: string | undefined,
+  cache: TrafficLiveCache, signal: AbortSignal, assertSource: () => Promise<void>): Promise<MonitorLiveSample> {
+  const parentSignal = signal, wave = new AbortController()
+  const onAbort = () => wave.abort(parentSignal.reason)
+  parentSignal.addEventListener("abort", onAbort, { once: true })
+  if (parentSignal.aborted) onAbort()
+  const timeout = setTimeout(() => wave.abort(new Error("Не успели получить полный live-срез за 8 секунд. Выберите отдельный интерфейс")), 8000)
+  signal = wave.signal
+  try {
+  const ensure = async () => { signal.throwIfAborted(); await assertSource(); signal.throwIfAborted() }
+  const monitor = async (name: string) => {
+    await ensure()
+    return client.post<unknown>("/interface/monitor-traffic", { interface: name, once: "" }, TIMEOUT_MS, signal)
+  }
+  if (onlyInterface) {
+    const raw = await monitor(onlyInterface)
+    await ensure()
+    return parseMonitorTraffic(raw, { onlyInterface })
+  }
+  await ensure()
+  const raw = await client.get<unknown>("/interface", TIMEOUT_MS)
+  if (!Array.isArray(raw)) throw new Error("Не получен каталог интерфейсов live")
+  const names: string[] = []
+  const flag = (value: unknown) => {
+    if (value === true || value === "true" || value === "yes") return true
+    if (value === false || value === "false" || value === "no") return false
+    throw new Error("Неизвестное состояние интерфейса live")
+  }
+  for (const row of raw) {
+    if (!row || typeof row !== "object" || Array.isArray(row) || typeof row.name !== "string" || !row.name.trim()) throw new Error("Некорректный каталог интерфейсов live")
+    if (flag(row.running) && !flag(row.disabled) && !isLoopbackName(row.name)) names.push(row.name)
+  }
+  if (!names.length || new Set(names).size !== names.length) throw new Error("Нет достоверного набора активных интерфейсов live")
+  if ([...names].sort().join("\0") !== [...cache.names].sort().join("\0")) cache.joinedFailed = false
+  cache.names = names
+  if (!cache.joinedFailed) {
+    try {
+      const result = await monitor(names.join(","))
+      await ensure()
+      return parseMonitorTraffic(result, { expectedInterfaces: names })
+    } catch (error) {
+      // Unsupported joined command may fall back; auth/transport/invalid payload cannot become partial zero.
+      if (!(error instanceof MikrotikError) || error.statusCode !== 400 || names.length < 2) throw error
+      cache.joinedFailed = true
+    }
+  }
+  const results: MonitorLiveSample[] = []
+  for (let i = 0; i < names.length; i += 4) {
+    results.push(...await Promise.all(names.slice(i, i + 4).map(async name =>
+      parseMonitorTraffic(await monitor(name), { onlyInterface: name }))))
+  }
+  await ensure()
+  const rxMbps = results.reduce((sum, r) => sum + r.rxMbps, 0), txMbps = results.reduce((sum, r) => sum + r.txMbps, 0)
+  if (![rxMbps, txMbps].every(n => Number.isFinite(n) && n <= Number.MAX_SAFE_INTEGER / 1_000_000)) throw new Error("Суммарная live-скорость вне допустимого диапазона")
+  return { rxMbps, txMbps, at: new Date().toISOString(), timestampKind: "backend-receipt" }
+  } catch (error) {
+    if (signal.aborted && !parentSignal.aborted && signal.reason instanceof Error) throw signal.reason
+    throw error
+  } finally {
+    clearTimeout(timeout)
+    parentSignal.removeEventListener("abort", onAbort)
+    wave.abort()
+  }
+}

backend/src/services/traffic-rate.ts

@@ -42,6 +42,7 @@ export interface MonitorLiveSample {
   rxMbps: number
   txMbps: number
   at: string
+  timestampKind: "backend-receipt"
 }
 
 export function isLoopbackName(name: string): boolean {
@@ -61,7 +62,7 @@ export function shouldIncludeIface(
 
 export function bpsToMbps(bps: number): number {
   if (!Number.isFinite(bps) || bps <= 0) return 0
-  return Math.round((bps / 1_000_000) * 1000) / 1000
+  return bps / 1_000_000
 }
 
 /** bits/s из соседних счётчиков. null = нельзя (Δt≤0 или сброс). */
@@ -75,7 +76,7 @@ export function rateBpsFromDelta(
   const dtSec = (nextAtMs - prevAtMs) / 1000
   if (!(dtSec > 0) || !Number.isFinite(dtSec)) return null
   if (nextBytes < prevBytes) return null
-  const rate=Math.round(((nextBytes - prevBytes) * 8) / dtSec)
+  const rate=((nextBytes - prevBytes) * 8) / dtSec
   return Number.isFinite(rate)?rate:null
 }
 
@@ -274,26 +275,41 @@ export function mergeBuiltTraffic(parts: BuiltTrafficSeries[]): BuiltTrafficSeri
 
 export function parseMonitorTraffic(
   raw: unknown,
-  opts?: { onlyInterface?: string },
+  opts?: { onlyInterface?: string; expectedInterfaces?: readonly string[] },
 ): MonitorLiveSample {
   const items = Array.isArray(raw) ? raw : raw != null ? [raw] : []
   let rxBps = 0
   let txBps = 0
+  const seen = new Set<string>()
+  const expected = opts?.expectedInterfaces ? new Set(opts.expectedInterfaces) : null
+  const rate = (raw: unknown) => {
+    if (typeof raw !== "number" && (typeof raw !== "string" || !/^(?:\d+(?:\.\d+)?|\.\d+)$/.test(raw.trim()))) throw new Error("Некорректная скорость monitor-traffic")
+    const n = Number(raw)
+    if (!Number.isFinite(n) || n < 0 || n > Number.MAX_SAFE_INTEGER) throw new Error("Некорректная скорость monitor-traffic")
+    return n
+  }
   for (const item of items) {
-    if (!item || typeof item !== "object") continue
+    if (!item || typeof item !== "object" || Array.isArray(item)) throw new Error("Некорректный ответ monitor-traffic")
     const rec = item as Record<string, unknown>
-    const name = String(rec.name ?? rec.interface ?? "")
+    const name = rec.name ?? rec.interface
+    if (typeof name !== "string" || !name.trim()) throw new Error("Не указан интерфейс monitor-traffic")
     if (opts?.onlyInterface) {
-      if (name && name !== opts.onlyInterface) continue
+      if (name !== opts.onlyInterface) throw new Error("Ответ monitor-traffic относится к другому интерфейсу")
     } else if (isLoopbackName(name)) {
       continue
     }
-    rxBps += Number.parseFloat(String(rec["rx-bits-per-second"] ?? 0)) || 0
-    txBps += Number.parseFloat(String(rec["tx-bits-per-second"] ?? 0)) || 0
+    if (seen.has(name) || expected && !expected.has(name)) throw new Error("Некорректный набор интерфейсов monitor-traffic")
+    seen.add(name)
+    rxBps += rate(rec["rx-bits-per-second"])
+    txBps += rate(rec["tx-bits-per-second"])
   }
+  if (!seen.size || expected && (expected.size !== opts!.expectedInterfaces!.length ||
+    [...expected].some(name => !seen.has(name)))) throw new Error("Не получены все live-измерения интерфейсов")
+  if (![rxBps, txBps].every(n => Number.isFinite(n) && n <= Number.MAX_SAFE_INTEGER)) throw new Error("Суммарная live-скорость вне допустимого диапазона")
   return {
-    rxMbps: bpsToMbps(rxBps),
-    txMbps: bpsToMbps(txBps),
+    rxMbps: rxBps / 1_000_000,
+    txMbps: txBps / 1_000_000,
     at: new Date().toISOString(),
+    timestampKind: "backend-receipt",
   }
 }