fix(netflow): отклонять конкурентный сброс до удаления данных

backend/src/db/operation-lock.test.ts

@@ -0,0 +1,17 @@
+import assert from "node:assert/strict"
+import {withTrafficFlowPurgeLock} from "./operation-lock.js"
+let held=false,runs=0,unlocks=0
+const releases:boolean[]=[]
+const connect=async()=>({query:async(sql:string,params:number[])=>{assert.deepEqual(params,[123]);if(sql.includes("pg_try_advisory_lock")){const acquired=!held;if(acquired)held=true;return {rows:[{acquired}]}}unlocks++;held=false;return {rows:[{released:true}]}},release:(destroy=false)=>{releases.push(destroy)}})
+let release!:()=>void
+const gate=new Promise<void>(resolve=>release=resolve)
+const first=withTrafficFlowPurgeLock(connect,123,async()=>{runs++;await gate;return "confirmed"})
+await new Promise<void>(resolve=>setImmediate(resolve))
+await assert.rejects(withTrafficFlowPurgeLock(connect,123,async()=>{runs++;return "unexpected"}),/уже выполняется/)
+assert.equal(runs,1);assert.equal(unlocks,0);assert.equal(held,true)
+release();assert.equal(await first,"confirmed");assert.equal(unlocks,1);assert.deepEqual(releases,[false,false])
+await assert.rejects(withTrafficFlowPurgeLock(connect,123,async()=>{throw Error("operation unknown")}),/operation unknown/);assert.equal(held,false);assert.equal(unlocks,2)
+for(const unlock of [false,"throws"]){let destroyed=false;const result=await withTrafficFlowPurgeLock(async()=>({query:async(sql:string)=>{if(sql.includes("pg_try"))return {rows:[{acquired:true}]};if(unlock==="throws")throw Error("unlock unavailable");return {rows:[{released:false}]}},release:(value=false)=>{destroyed=value}}),123,async()=>"committed");assert.equal(result,"committed");assert.equal(destroyed,true)}
+let released=false,discarded=false
+await assert.rejects(withTrafficFlowPurgeLock(async()=>({query:async()=>({rows:[{}]}),release:(destroy=false)=>{released=true;discarded=destroy}}),123,async()=>{throw Error("must not run")}),/не подтверждена/);assert.equal(released,true);assert.equal(discarded,true)
+console.log("NetFlow purge lock: concurrent request rejected before handler, unknown release destroys connection, operation receipt preserved")

backend/src/db/operation-lock.ts

@@ -0,0 +1,26 @@
+type Client = {
+  query: (sql: string, params: number[]) => Promise<{rows: Record<string, unknown>[]}>
+  release: (destroy?: boolean) => void
+}
+/** Nonblocking session lock: a second destructive operation is rejected rather than queued. */
+export async function withTrafficFlowPurgeLock<T>(connect: () => Promise<Client>, key: number, run: () => Promise<T>): Promise<T> {
+  const client = await connect()
+  let owned = false, destroy = false
+  try {
+    destroy = true
+    const result = await client.query("SELECT pg_try_advisory_lock($1) AS acquired", [key])
+    if (result.rows[0]?.acquired === false) { destroy = false; throw new Error("Сброс NetFlow уже выполняется") }
+    if (result.rows[0]?.acquired !== true) throw new Error("Блокировка сброса NetFlow не подтверждена")
+    owned = true; destroy = false
+    return await run()
+  } finally {
+    if (owned) {
+      try {
+        const result = await client.query("SELECT pg_advisory_unlock($1) AS released", [key])
+        destroy = result.rows[0]?.released !== true
+      } catch { destroy = true }
+    }
+    // A connection whose unlock failed must not return a held session lock to the pool.
+    client.release(destroy)
+  }
+}

backend/src/services/traffic-flow-ingest.ts

@@ -1,6 +1,7 @@
 import { deleteTrafficFlowData, executeTrafficFlowPurge } from "./traffic-flow-purge-execution.js"
+import { withTrafficFlowPurgeLock } from "../db/operation-lock.js"
 import { Worker } from "node:worker_threads"
-import { db, dbAll, pool, withAdvisoryLock } from "../db/index.js"
+import { db, dbAll, pool } from "../db/index.js"
 import { dropExpiredPartitions } from "../db/partitions.js"
 import { servers } from "../db/schema.js"
 import type { FlowPurgeDto, FlowStatsDto, FlowTalkerDto } from "@mmapp/contracts/traffic-flow"
@@ -519,7 +520,7 @@ const PURGE_LOCK_KEY = 8_723_103
 
 /** Удаляет данные одним transaction; ключи и пиры не меняются. */
 export async function purgeTrafficFlowStore(): Promise<FlowPurgeDto> {
-  return withAdvisoryLock(PURGE_LOCK_KEY, () => executeTrafficFlowPurge({
+  return withTrafficFlowPurgeLock(() => pool.connect(), PURGE_LOCK_KEY, () => executeTrafficFlowPurge({
     stop: async () => {
       wantListen = false
       await stopWorkerProcessAsync()