fix(traffic): останавливать сбор при потере соединения блокировки

backend/scripts/acceptance/traffic-lock-lifecycle.mjs

@@ -0,0 +1,33 @@
+import assert from 'node:assert/strict'
+import {createServer} from 'node:http'
+import {spawn} from 'node:child_process'
+import fs from 'node:fs'
+const database=process.env.MMAPP_OWN_SQL_DB
+assert(/^mm_accept_166_\d+$/.test(database??''))
+const url=new URL(process.env.DATABASE_URL);assert.equal(url.pathname,'/'+database);assert.equal(url.username,database)
+const {pool,closePool}=await import('/app/backend/dist/db/index.js')
+const {initDatabase}=await import('/app/backend/dist/db/bootstrap.js')
+assert.equal((await pool.query('SELECT current_database() AS name')).rows[0].name,database)
+assert(Object.values((await pool.query('SELECT rolsuper,rolcreatedb,rolcreaterole,rolbypassrls FROM pg_roles WHERE rolname=current_user')).rows[0]).every(v=>v===false))
+const baseline=process.env.MMAPP_LOCK_LOSS_BASELINE==='yes',children=[],cases=[];let holdNext=false,heldSeen=()=>{},releaseHeld=()=>{},interfaceRequests=0,serverId
+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(holdNext){holdNext=false;releaseHeld=respond;heldSeen();return}}respond()})
+await new Promise(r=>http.listen(0,'127.0.0.1',r));const address=http.address()
+function start(){const code=`import {pool,closePool} from '/app/backend/dist/db/index.js';import {collectTrafficOnce} from '/app/backend/dist/services/traffic-collector.js';try{console.log(JSON.stringify(await collectTrafficOnce()))}finally{await closePool()}`;const child=spawn(process.execPath,['--input-type=module','-e',code],{env:process.env,stdio:['ignore','pipe','pipe']});children.push(child);let out='',err='';child.stdout.on('data',b=>out+=b);child.stderr.on('data',b=>err+=b);const done=new Promise((resolve,reject)=>{child.on('error',reject);child.on('exit',(exit,signal)=>resolve({exit,signal,out,err}))});return {child,done}}
+async function held(){holdNext=true;const seen=new Promise(r=>heldSeen=r);const job=start();await Promise.race([seen,job.done.then(()=>{throw Error('Collector ended before held native request')}),new Promise((_,reject)=>{const timer=setTimeout(()=>reject(Error('Held request deadline')),8000);timer.unref()})]);return job}
+async function count(){return Number((await pool.query('SELECT count(*) AS n FROM traffic_samples WHERE server_id=$1',[serverId])).rows[0].n)}
+const report={date:'2026-10-13',block:'12.5.200',database,baseline,cases,passed:false,scope:'Production compiled collector, independent Linux child processes, owned restricted PostgreSQL and local RouterOS HTTP fixture; no production PID/native CHR touched'}
+try{
+ await initDatabase();serverId=Number((await pool.query("INSERT INTO servers(name,host,port,use_ssl,enabled,username,password) VALUES('own-lifecycle200','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 crashed=await held();crashed.child.kill('SIGKILL');const crash=await crashed.done;assert.equal(crash.signal,'SIGKILL');releaseHeld();assert.equal(await count(),0)
+ let recovered=await start().done;assert.equal(recovered.exit,0);assert.equal(JSON.parse(recovered.out).fatalError,undefined);assert.equal(await count(),1);cases.push({name:'process crash releases advisory lock',signal:'SIGKILL',crashedWaveRows:0,recoveryWaveRows:1})
+ const beforeSettings=(await pool.query('SELECT last_collected_at FROM traffic_settings WHERE id=1')).rows[0].last_collected_at
+ const lost=await held();const locks=(await pool.query("SELECT a.pid,a.datname,a.usename FROM pg_locks l JOIN pg_stat_activity a ON a.pid=l.pid WHERE l.locktype='advisory' AND l.objid=8723105 AND l.classid=0 AND l.objsubid=1 AND l.granted AND a.datname=$1 AND a.usename=$1",[database])).rows;assert.equal(locks.length,1);assert.equal(locks[0].datname,database);assert.equal(locks[0].usename,database)
+ assert.equal((await pool.query('SELECT pg_terminate_backend($1) AS terminated',[locks[0].pid])).rows[0].terminated,true)
+ await new Promise(r=>setTimeout(r,150));releaseHeld();const ended=await lost.done
+ if(baseline){assert.notEqual(ended.exit,0);assert(ended.err.includes("Unhandled 'error' event"));cases.push({name:'lock connection loss baseline',unhandledClientError:true,nonzeroChildExit:true})}
+ else{assert.equal(ended.exit,0);assert.match(JSON.parse(ended.out).fatalError,/Соединение блокировки/);assert.equal(await count(),1);assert.equal(String((await pool.query('SELECT last_collected_at FROM traffic_settings WHERE id=1')).rows[0].last_collected_at),String(beforeSettings));cases.push({name:'lock connection loss',collectorProcessSurvived:true,fatalSnapshot:true,staleSampleRows:0,staleSettingsWrite:false})}
+ recovered=await start().done;assert.equal(recovered.exit,0);assert.equal(JSON.parse(recovered.out).fatalError,undefined);assert.equal(await count(),2);cases.push({name:'next process recovers after connection loss',waveRows:2});report.interfaceRequests=interfaceRequests;report.passed=true
+}catch(e){report.errorName=e.name;fs.writeFileSync('/app/backend/acceptance/lifecycle200-private.log',String(e.stack??e),{mode:0o600})}
+finally{releaseHeld();for(const c of children)if(c.exitCode===null&&c.signalCode===null)c.kill('SIGKILL');http.closeAllConnections();await new Promise(r=>http.close(r));await closePool();fs.writeFileSync('/app/backend/acceptance/lifecycle200-'+(baseline?'baseline':'fixed')+'-result.json',JSON.stringify(report,null,2));console.log(JSON.stringify(report))}
+if(!report.passed)process.exitCode=1

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

@@ -15,3 +15,11 @@ for(const state of ['busy','missing','ok','run-error','unlock-false','unlock-err
   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')
+import {EventEmitter} from 'node:events'
+for(const event of ['error','end']){
+ const events=new EventEmitter();let destroy=false,writes=0
+ const client={query:async()=>({rows:[{acquired:true,released:true}]}),release:(value?:boolean)=>{destroy=!!value},on:events.on.bind(events),removeListener:events.removeListener.bind(events)}
+ await assert.rejects(withTrafficCollectionLock(async()=>client,async guard=>{guard();events.emit(event,Error('own loss'));guard();writes++;return 42}),/Соединение блокировки/)
+ assert.equal(writes,0);assert.equal(destroy,true);assert.equal(events.listenerCount(event),0)
+}
+console.log('traffic lock: error/end handled, guarded work aborted, connection destroyed, listeners removed')

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

@@ -1,24 +1,35 @@
 type Client = {
   query: (sql:string,params:number[])=>Promise<{rows:Record<string,unknown>[]}>
   release: (destroy?:boolean)=>void
+  on?: (event:"error"|"end",listener:()=>void)=>unknown
+  removeListener?: (event:"error"|"end",listener:()=>void)=>unknown
 }
 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}> {
+export async function withTrafficCollectionLock<T>(connect:()=>Promise<Client>,run:(assertActive:()=>void)=>Promise<T>):Promise<{acquired:false}|{acquired:true;value:T}> {
   const client=await connect()
-  let owned=false,destroy=true
+  let owned=false,destroy=true,lost=false
+  const lose=()=>{lost=true;destroy=true}
+  const assertActive=()=>{if(lost)throw Error('Соединение блокировки опроса трафика потеряно; опрос остановлен')}
+  client.on?.('error',lose)
+  client.on?.('end',lose)
   try {
     const result=await client.query('SELECT pg_try_advisory_lock($1) AS acquired',[TRAFFIC_COLLECTION_LOCK_KEY])
+    assertActive()
     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()}
+    const value=await run(assertActive)
+    assertActive()
+    return {acquired:true,value}
   }finally{
-    if(owned){
+    if(owned&&!lost){
       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)
+    client.removeListener?.('error',lose)
+    client.removeListener?.('end',lose)
   }
 }

backend/src/services/traffic-collector.ts

@@ -117,7 +117,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
   }
   collecting = true
   try {
-    const result = await withTrafficCollectionLock(() => pool.connect(), () => collectTrafficRun(sampledAt))
+    const result = await withTrafficCollectionLock(() => pool.connect(), assertActive => collectTrafficRun(sampledAt,assertActive))
     return result.acquired ? result.value : {
       v:SCHEDULER_RUN_SNAPSHOT_VERSION,job:"traffic",sampledAt,skipped:true,servers:[],
     }
@@ -129,7 +129,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
   }
 }
 
-async function collectTrafficRun(sampledAt:string):Promise<TrafficRunSnapshot> {
+async function collectTrafficRun(sampledAt:string,assertActive:()=>void=()=>{}):Promise<TrafficRunSnapshot> {
   const startedAt = Date.now()
   const now = sampledAt
   const snapshot: TrafficRunSnapshot = {
@@ -140,14 +140,18 @@ async function collectTrafficRun(sampledAt:string):Promise<TrafficRunSnapshot> {
   }
 
   try {
+    assertActive()
     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) {
+      assertActive()
       const name = srv.name || srv.host
       const ensureSource = async () => {
+        assertActive()
         const current = (await db.select({enabled:servers.enabled,updatedAt:servers.updatedAt}).from(servers).where(eq(servers.id,srv.id)).limit(1))[0]
+        assertActive()
         if (!current?.enabled || Date.parse(current.updatedAt) !== Date.parse(srv.updatedAt))
           throw new Error("Сервер выключен, удалён или его конфигурация изменилась во время опроса")
       }
@@ -247,6 +251,7 @@ async function collectTrafficRun(sampledAt:string):Promise<TrafficRunSnapshot> {
           } : {}),
         })
       } catch (err) {
+        assertActive()
         await recordTrafficCollectionAttempt(srv, "error", now, err instanceof Error ? err.message : String(err))
         snapshot.servers.push({
           serverId: srv.id,
@@ -258,6 +263,7 @@ async function collectTrafficRun(sampledAt:string):Promise<TrafficRunSnapshot> {
       }
     }
 
+    assertActive()
     await db.update(trafficSettings).set({
       lastCollectedAt: now,
       lastDurationMs: Date.now() - startedAt,
@@ -265,6 +271,7 @@ async function collectTrafficRun(sampledAt:string):Promise<TrafficRunSnapshot> {
       updatedAt: now,
     }).where(eq(trafficSettings.id, 1))
   } catch (e) {
+    assertActive()
     const msg = e instanceof Error ? e.message : String(e)
     snapshot.fatalError = msg
     await db.update(trafficSettings).set({