fix(traffic): согласовать опрос между процессами backend

backend/package.json

@@ -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 && tsx src/services/traffic-gap.test.ts && tsx src/services/traffic-identity.test.ts && tsx src/services/traffic-membership.test.ts && tsx src/services/traffic-aggregate-peak.test.ts && tsx src/services/traffic-users-cache.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 && tsx src/services/traffic-identity.test.ts && tsx src/services/traffic-membership.test.ts && tsx src/services/traffic-aggregate-peak.test.ts && tsx src/services/traffic-users-cache.test.ts && tsx src/services/traffic-collection-lock.test.ts && tsx src/services/traffic-concurrency-sql.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",

backend/src/services/traffic-collection-lock.test.ts

@@ -0,0 +1,17 @@
+import assert from 'node:assert/strict'
+import {withTrafficCollectionLock} from './traffic-collection-lock.js'
+for(const state of ['busy','missing','ok','run-error','unlock-false','unlock-error','acquire-error']){
+  let runs=0,released:boolean|undefined,queries=0
+  const client={query:async()=>{
+    queries++
+    if(queries===1){if(state==='acquire-error')throw Error('connectivity');return {rows:state==='missing'?[]:[{acquired:state!=='busy'}]}}
+    if(state==='unlock-error')throw Error('unlock')
+    return {rows:[{released:state!=='unlock-false'}]}
+  },release:(destroy?:boolean)=>{released=destroy}}
+  const run=()=>withTrafficCollectionLock(async()=>client,async()=>{runs++;if(state==='run-error')throw Error('run');return 42})
+  if(['missing','run-error','acquire-error'].includes(state))await assert.rejects(run)
+  else {const result=await run();assert.equal(result.acquired,state!=='busy');if(result.acquired)assert.equal(result.value,42)}
+  assert.equal(runs,['busy','missing','acquire-error'].includes(state)?0:1)
+  assert.equal(released,['missing','unlock-false','unlock-error','acquire-error'].includes(state))
+}
+console.log('traffic lock: busy/no work/acquisition uncertainty/run failure/unlock failure/destruction passed')

backend/src/services/traffic-collection-lock.ts

@@ -0,0 +1,24 @@
+type Client = {
+  query: (sql:string,params:number[])=>Promise<{rows:Record<string,unknown>[]}>
+  release: (destroy?:boolean)=>void
+}
+export const TRAFFIC_COLLECTION_LOCK_KEY=8_723_105
+
+/** Coordinate collectors in the same database without queueing stale polling runs. */
+export async function withTrafficCollectionLock<T>(connect:()=>Promise<Client>,run:()=>Promise<T>):Promise<{acquired:false}|{acquired:true;value:T}> {
+  const client=await connect()
+  let owned=false,destroy=true
+  try {
+    const result=await client.query('SELECT pg_try_advisory_lock($1) AS acquired',[TRAFFIC_COLLECTION_LOCK_KEY])
+    if(result.rows[0]?.acquired===false){destroy=false;return {acquired:false}}
+    if(result.rows[0]?.acquired!==true)throw Error('Блокировка опроса трафика не подтверждена')
+    owned=true;destroy=false
+    return {acquired:true,value:await run()}
+  }finally{
+    if(owned){
+      try{destroy=(await client.query('SELECT pg_advisory_unlock($1) AS released',[TRAFFIC_COLLECTION_LOCK_KEY])).rows[0]?.released!==true}
+      catch{destroy=true}
+    }
+    client.release(destroy)
+  }
+}

backend/src/services/traffic-collector.ts

@@ -1,5 +1,6 @@
 import { and, asc, desc, eq, gte } from "drizzle-orm"
-import { db } from "../db/index.js"
+import { db, pool } from "../db/index.js"
+import { withTrafficCollectionLock } from "./traffic-collection-lock.js"
 import { servers, trafficSamples, trafficSettings } from "../db/schema.js"
 import { decodeTrafficSampleFlags, encodeTrafficFlags, trafficCounterContinuity } from "../db/traffic-flags.js"
 import type { TrafficRunSnapshot } from "../types/scheduler-run-snapshot.js"
@@ -115,6 +116,20 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
     }
   }
   collecting = true
+  try {
+    const result = await withTrafficCollectionLock(() => pool.connect(), () => collectTrafficRun(sampledAt))
+    return result.acquired ? result.value : {
+      v:SCHEDULER_RUN_SNAPSHOT_VERSION,job:"traffic",sampledAt,skipped:true,servers:[],
+    }
+  } catch (error) {
+    return {v:SCHEDULER_RUN_SNAPSHOT_VERSION,job:"traffic",sampledAt,servers:[],
+      fatalError:error instanceof Error ? error.message : String(error)}
+  } finally {
+    collecting = false
+  }
+}
+
+async function collectTrafficRun(sampledAt:string):Promise<TrafficRunSnapshot> {
   const startedAt = Date.now()
   const now = sampledAt
   const snapshot: TrafficRunSnapshot = {
@@ -258,8 +273,6 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
       lastError: msg,
       updatedAt: now,
     }).where(eq(trafficSettings.id, 1))
-  } finally {
-    collecting = false
   }
   return snapshot
 }

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

@@ -0,0 +1,57 @@
+import assert from 'node:assert/strict'
+import {createServer} from 'node:http'
+import {spawn} from 'node:child_process'
+import {pool,closePool} from '../db/index.js'
+import {initDatabase} from '../db/bootstrap.js'
+
+const database=process.env.MMAPP_OWN_SQL_DB
+if(!database)console.log('traffic concurrency 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 baseline=process.env.MMAPP_TRAFFIC_CONCURRENCY_EXPECT==='overlap'
+  let serverId:number|undefined,interfaceRequests=0,releaseFirst:()=>void=()=>{},firstSeen:()=>void=()=>{},secondSeen:()=>void=()=>{}
+  const firstRequest=new Promise<void>(resolve=>firstSeen=resolve),secondRequest=new Promise<void>(resolve=>secondSeen=resolve)
+  const children:ReturnType<typeof spawn>[]=[]
+  const http=createServer((req,res)=>{
+    res.setHeader('Content-Type','application/json')
+    const respond=()=>res.end(JSON.stringify(req.url?.includes('/wireguard/')?[]:[{'.id':'*1',name:'ether1',running:true,disabled:false,'rx-byte':'1000','tx-byte':'1000'}]))
+    if(req.url?.includes('/interface')&&!req.url.includes('/wireguard/')){
+      interfaceRequests++
+      if(interfaceRequests===1){releaseFirst=respond;firstSeen();return}
+      secondSeen()
+    }
+    respond()
+  })
+  await new Promise<void>(resolve=>http.listen(0,'127.0.0.1',resolve))
+  const address=http.address();assert(address&&typeof address==='object')
+  const extension=import.meta.url.endsWith('.ts')?'ts':'js'
+  const childCode=`import assert from 'node:assert/strict';import {pool,closePool} from ${JSON.stringify(new URL('../db/index.'+extension,import.meta.url).href)};import {collectTrafficOnce} from ${JSON.stringify(new URL('./traffic-collector.'+extension,import.meta.url).href)};assert.equal((await pool.query('SELECT current_database() AS name')).rows[0].name,process.env.MMAPP_OWN_SQL_DB);try{console.log(JSON.stringify(await collectTrafficOnce()))}finally{await closePool()}`
+  const run=()=>new Promise<{skipped?:boolean;servers:unknown[]}>( (resolve,reject)=>{
+    const child=spawn(process.execPath,[...process.execArgv,'--input-type=module','-e',childCode],{env:process.env,stdio:['ignore','pipe','pipe']});children.push(child)
+    let output='',error='';child.stdout.on('data',b=>output+=b);child.stderr.on('data',b=>error+=b)
+    child.on('error',reject);child.on('exit',code=>{if(code!==0)return reject(Error('Owned collector child failed: '+error));try{resolve(JSON.parse(output.trim()))}catch(e){reject(e)}})
+  })
+  let timeout:ReturnType<typeof setTimeout>|undefined
+  try {
+    await initDatabase()
+    serverId=Number((await pool.query("INSERT INTO servers(name,host,port,use_ssl,enabled,username,password) VALUES('own-concurrency194','127.0.0.1',$1,false,true,'fixture','fixture') RETURNING id",[address.port])).rows[0].id)
+    await pool.query('INSERT INTO traffic_settings(id,enabled,interval_sec) VALUES(1,true,30) ON CONFLICT(id) DO UPDATE SET enabled=true,interval_sec=30')
+    const deadline=()=>new Promise((_,reject)=>{timeout=setTimeout(()=>reject(Error('Collector deadline exceeded')),8000)})
+    const first=run();await Promise.race([firstRequest,first.then(()=>{throw Error('First collector finished before its held request')}),deadline()]);clearTimeout(timeout)
+    const second=run()
+    await Promise.race([secondRequest,second,deadline()])
+    clearTimeout(timeout);releaseFirst()
+    const [a,b]=await Promise.all([first,second]);assert.equal(a.skipped,undefined)
+    assert.equal(b.skipped,baseline?undefined:true)
+    assert.equal(interfaceRequests,baseline?2:1)
+    assert.equal(Number((await pool.query('SELECT count(*) AS n FROM traffic_samples WHERE server_id=$1',[serverId])).rows[0].n),baseline?2:1)
+    const recovery=await run();assert.equal(recovery.skipped,undefined);assert.equal(interfaceRequests,baseline?3:2)
+    console.log(`traffic concurrency owned SQL: ${baseline?'overlap baseline':'cross-process skip/release/recovery'} passed; interface requests ${interfaceRequests}`)
+  }finally{
+    clearTimeout(timeout);releaseFirst();for(const child of children)if(child.exitCode===null)child.kill()
+    await new Promise<void>(resolve=>http.close(()=>resolve()))
+    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()
+  }
+}

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

@@ -37,11 +37,12 @@ console.log("traffic observation counters/rates/source confidence tests ok")
 const collector = fs.readFileSync(new URL("./traffic-collector.ts", import.meta.url), "utf8")
 const ast = ts.createSourceFile("traffic-collector.ts", collector, ts.ScriptTarget.Latest, true)
 const functions = ast.statements.filter(n => ts.isFunctionDeclaration(n) && n.name &&
-  ["waveKey", "sampleRate", "toNum", "collectTrafficOnce"].includes(n.name.text)).map(n => n.getText(ast)).join("\n")
+  ["waveKey", "sampleRate", "toNum", "collectTrafficOnce", "collectTrafficRun"].includes(n.name.text)).map(n => n.getText(ast)).join("\n")
 assert.equal(ast.statements.filter(n => ts.isFunctionDeclaration(n) && n.name && n.name.text === "collectTrafficOnce").length, 1)
 let inserted: unknown[] = [], wave = new Map<string, typeof previous & {nativeCounterIdentity?:string|null}>(), counters: unknown = "1000"
 let included = true
 const context = vm.createContext({
+  pool:{connect:()=>{}},withTrafficCollectionLock:async(_connect:unknown,run:()=>Promise<unknown>)=>({acquired:true,value:await run()}),
   Date, nativeTrafficCounterIdentity,parseTrafficPeers, observedTrafficRate, parseTrafficCounter, parseTrafficFlag, bpsToMbps, shouldIncludeIface,
   SCHEDULER_RUN_SNAPSHOT_VERSION: 1, servers: {}, trafficSamples: {}, trafficSettings: {}, eq: () => ({}),
   getSettingsRow: async () => ({}), readPreviousWave: async () => wave, rememberServerIfaces: () => {},