fix(traffic): принимать измерения peers только полным срезом

backend/src/services/traffic-collector.ts

@@ -9,6 +9,7 @@ import { bpsToMbps, shouldIncludeIface } from "./traffic-rate.js"
 import { observedTrafficRate, parseTrafficCounter, parseTrafficFlag } from "./traffic-observation.js"
 import { rememberServerIfaces } from "./traffic-flow-ifindex.js"
 import { recordTrafficCollectionAttempt } from "./traffic-collection-state.js"
+import { parseTrafficPeers } from "./traffic-peer-observation.js"
 
 interface RosIfaceTraffic {
   ".id"?: string
@@ -166,13 +167,8 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
         })
         try {
           const peers = await client.get<RosWgPeerTraffic[]>("/interface/wireguard/peers")
-          for (const p of peers) {
-            const interfaceName = (p.interface ?? "").trim()
-            const peerPublicKey = (p["public-key"] ?? "").trim()
-            if (!interfaceName || !peerPublicKey) continue
-            const rxBytes = toNum(p.rx)
-            const txBytes = toNum(p.tx)
-            const disabled = parseTrafficFlag(p.disabled)
+          for (const p of parseTrafficPeers(peers)) {
+            const { interfaceName, peerPublicKey, rxBytes, txBytes, disabled } = p
             const running = !disabled
             const { rxBps, txBps } = sampleRate(
               prevWave,

backend/src/services/traffic-peer-observation.ts

@@ -0,0 +1,18 @@
+import { parseTrafficCounter, parseTrafficFlag } from "./traffic-observation.js"
+
+/** Validate the whole optional peer slice before accepting any of its rows. */
+export function parseTrafficPeers(raw: unknown) {
+  if (!Array.isArray(raw)) throw new Error("RouterOS вернул некорректный список WireGuard peers")
+  const seen = new Set<string>()
+  return raw.map((value: unknown) => {
+    if (!value || typeof value !== "object") throw new Error("Некорректное измерение WireGuard peer")
+    const p = value as Record<string, unknown>
+    if (typeof p.interface !== "string" || !p.interface.trim() || typeof p["public-key"] !== "string" || !p["public-key"].trim())
+      throw new Error("Неизвестный интерфейс или ключ WireGuard peer")
+    const interfaceName = p.interface.trim(), peerPublicKey = p["public-key"].trim()
+    const key = `${interfaceName}\0${peerPublicKey}`
+    if (seen.has(key)) throw new Error("Повторное измерение WireGuard peer")
+    seen.add(key)
+    return { interfaceName, peerPublicKey, rxBytes: parseTrafficCounter(p.rx), txBytes: parseTrafficCounter(p.tx), disabled: parseTrafficFlag(p.disabled) }
+  })
+}

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

@@ -9,8 +9,8 @@ const database=process.env.MMAPP_OWN_SQL_DB
 if(!database)console.log("traffic status 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-status181-"+Date.now()+"-",userId=prefix+"user";let serverId:number|undefined,cases=0,deny=false,flag:unknown=true,peerDisabled=false
- const http=createServer((req,res)=>{res.setHeader("Content-Type","application/json");if(deny){res.statusCode=401;return res.end('{"message":"Unauthorized"}')}res.end(JSON.stringify(req.url?.includes("/wireguard/")?[{interface:"wg1","public-key":"fixturekey",rx:"1000",tx:"1000",disabled:peerDisabled}]:[{name:"ether1",running:flag,disabled:false,"rx-byte":"1000","tx-byte":"1000"}]))})
+ const prefix="own-status181-"+Date.now()+"-",userId=prefix+"user";let serverId:number|undefined,cases=0,deny=false,flag:unknown=true,peerDisabled=false,peerVariant="valid"
+ const http=createServer((req,res)=>{res.setHeader("Content-Type","application/json");if(deny){res.statusCode=401;return res.end('{"message":"Unauthorized"}')}res.end(JSON.stringify(req.url?.includes("/wireguard/")?(peerVariant==="shape"?{}:peerVariant==="duplicate"?[{interface:"wg1","public-key":"fixturekey",rx:"1000",tx:"1000",disabled:false},{interface:"wg1","public-key":"fixturekey",rx:"1000",tx:"1000",disabled:false}]:[{interface:"wg1","public-key":"fixturekey",rx:"1000",tx:"1000",disabled:peerDisabled},...(peerVariant==="partial"?[{interface:"wg1","public-key":"broken",rx:"invalid",tx:"1000",disabled:false}]:[])]):[{name:"ether1",running:flag,disabled:false,"rx-byte":"1000","tx-byte":"1000"}]))})
  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 bindings=async()=>{const r=await app.inject({method:"GET",url:"/traffic/bound-interfaces?range=1h"});assert.equal(r.statusCode,200);return r.json().interfaces.filter((i:{userId:string})=>i.userId===userId)}
  try{
@@ -27,6 +27,11 @@ else{
   flag=true;deny=true;await collectTrafficOnce();rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
   deny=false;await collectTrafficOnce();await pool.query("UPDATE traffic_samples SET flags=flags & 7 WHERE server_id=$1",[serverId]);rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
   await collectTrafficOnce();await pool.query("UPDATE traffic_collection_state SET last_attempt_at=$1 WHERE server_id=$2",[new Date(Date.now()-100_000).toISOString(),serverId]);rows=await bindings();assert(rows.every((r:{status:string})=>r.status==="unknown"));cases++
+  for(const variant of["partial","duplicate","shape"]){
+   peerVariant=variant;await pool.query("DELETE FROM traffic_samples WHERE server_id=$1",[serverId]);const result=await collectTrafficOnce();assert.equal(result.servers[0].ok,true)
+   const saved=(await pool.query("SELECT peer_public_key FROM traffic_samples WHERE server_id=$1",[serverId])).rows;assert.equal(saved.length,1);assert.equal(saved[0].peer_public_key,"")
+   rows=await bindings();assert.equal(rows.find((r:{interfaceName:string})=>r.interfaceName==="wg1").status,"unknown");assert.equal(rows.find((r:{interfaceName:string})=>r.interfaceName==="ether1").status,"online");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()}