From 32b7812169cff51187600dcafdbb135046ca154f Mon Sep 17 00:00:00 2001 From: Denozordec Date: Sat, 3 Oct 2026 12:54:27 +0700 Subject: [PATCH] =?UTF-8?q?fix(netflow):=20=D0=BE=D1=82=D0=BA=D0=BB=D0=BE?= =?UTF-8?q?=D0=BD=D1=8F=D1=82=D1=8C=20=D0=BA=D0=BE=D0=BD=D0=BA=D1=83=D1=80?= =?UTF-8?q?=D0=B5=D0=BD=D1=82=D0=BD=D1=8B=D0=B9=20=D1=81=D0=B1=D1=80=D0=BE?= =?UTF-8?q?=D1=81=20=D0=B4=D0=BE=20=D1=83=D0=B4=D0=B0=D0=BB=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D1=8F=20=D0=B4=D0=B0=D0=BD=D0=BD=D1=8B=D1=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/package.json | 2 +- backend/src/db/operation-lock.test.ts | 17 ++++++++++++ backend/src/db/operation-lock.ts | 26 +++++++++++++++++++ backend/src/services/traffic-flow-ingest.ts | 5 ++-- components/traffic/netflow-settings-panel.tsx | 3 ++- docs/ux-ui-plan.md | 2 ++ docs/ux-ui-progress.md | 9 ++++++- docs/ux-ui-stage-11b.md | 2 ++ tests/e2e/netflow-settings.spec.ts | 2 ++ 9 files changed, 63 insertions(+), 5 deletions(-) create mode 100644 backend/src/db/operation-lock.test.ts create mode 100644 backend/src/db/operation-lock.ts diff --git a/backend/package.json b/backend/package.json index f76ce7c8..dccd6e14 100644 --- a/backend/package.json +++ b/backend/package.json @@ -28,7 +28,7 @@ "test:backups": "tsx src/services/backup-scheduler-run.test.ts && tsx src/services/backup-restore.test.ts && tsx src/services/backup-deletion.test.ts && tsx src/services/backup-dto.test.ts && tsx src/services/backup-storage-update.test.ts && tsx src/services/backup-job-execution.test.ts && tsx src/routes/backups-operations.test.ts && tsx src/services/s3-backup-client.test.ts", "test:live-maps": "tsx src/services/ospf-route-parse.test.ts && tsx src/services/vxlan-live.test.ts && tsx src/services/containers-live.test.ts && tsx src/services/firewall-poll.test.ts && tsx src/services/gre-catalog-poll.test.ts && tsx src/routes/filters-gre-catalog.test.ts && tsx src/routes/bgp-catalog.test.ts && tsx src/services/bgp-catalog-parse.test.ts && tsx src/routes/ospf-catalog.test.ts && tsx src/services/ospf-observed-values.test.ts", "test": "npm run test:probes && npm run test:exec && npm run test:alert-engine && npm run test:auth && npm run test:wireguard && npm run test:traffic-rate && npm run test:traffic-flow && npm run test:users && npm run test:pg && npm run test:backups && npm run test:scheduler && npm run test:live-maps && npm run test:ospf-optimizer && npm run test:recursive-routes && npm run test:config-sync && npm run test:ros-file && npm run test:containers && npm run test:certificates && npm run test:netflow-settings && npm run test:netflow-purge", - "test:netflow-purge": "tsx src/services/traffic-flow-purge-execution.test.ts && tsx src/routes/traffic-flow-purge.test.ts", + "test:netflow-purge": "tsx src/db/operation-lock.test.ts && tsx src/services/traffic-flow-purge-execution.test.ts && tsx src/routes/traffic-flow-purge.test.ts", "test:netflow-settings": "tsx src/services/traffic-flow-key-execution.test.ts && tsx src/routes/traffic-flow-settings.test.ts && tsx src/plugins/auth.netflow-settings.test.ts", "test:geoip": "tsx src/services/traffic-flow-geoip.test.ts && tsx src/plugins/auth.geoip.test.ts", "test:ospf-optimizer": "tsx src/services/ospf-optimizer-identity.test.ts && tsx src/services/ospf-optimizer-metrics.test.ts && tsx src/services/ospf-optimizer-plan.test.ts && tsx src/routes/ospf-optimizer.test.ts", diff --git a/backend/src/db/operation-lock.test.ts b/backend/src/db/operation-lock.test.ts new file mode 100644 index 00000000..8067194a --- /dev/null +++ b/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(resolve=>release=resolve) +const first=withTrafficFlowPurgeLock(connect,123,async()=>{runs++;await gate;return "confirmed"}) +await new Promise(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") diff --git a/backend/src/db/operation-lock.ts b/backend/src/db/operation-lock.ts new file mode 100644 index 00000000..692f619d --- /dev/null +++ b/backend/src/db/operation-lock.ts @@ -0,0 +1,26 @@ +type Client = { + query: (sql: string, params: number[]) => Promise<{rows: Record[]}> + release: (destroy?: boolean) => void +} +/** Nonblocking session lock: a second destructive operation is rejected rather than queued. */ +export async function withTrafficFlowPurgeLock(connect: () => Promise, key: number, run: () => Promise): Promise { + 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) + } +} diff --git a/backend/src/services/traffic-flow-ingest.ts b/backend/src/services/traffic-flow-ingest.ts index 842e29a3..3b7e594f 100644 --- a/backend/src/services/traffic-flow-ingest.ts +++ b/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 { - return withAdvisoryLock(PURGE_LOCK_KEY, () => executeTrafficFlowPurge({ + return withTrafficFlowPurgeLock(() => pool.connect(), PURGE_LOCK_KEY, () => executeTrafficFlowPurge({ stop: async () => { wantListen = false await stopWorkerProcessAsync() diff --git a/components/traffic/netflow-settings-panel.tsx b/components/traffic/netflow-settings-panel.tsx index 9ef28e49..a55aa579 100644 --- a/components/traffic/netflow-settings-panel.tsx +++ b/components/traffic/netflow-settings-panel.tsx @@ -1,6 +1,7 @@ "use client" import { useCallback, useEffect, useRef, useState } from "react" +import { ApiClientError } from "@/shared/api/http-client" import { can } from "@/lib/auth" import { emptyNetflowDraft, netflowDraftFromSettings, netflowDraftPatch, type NetflowDraft } from "@/shared/api/netflow-settings-form" import { FormField, FormToggle } from "@/components/form-kit" @@ -106,7 +107,7 @@ function NetflowSettingsContent({ setPurgeOpen(false);setOutcome("Сброс данных NetFlow подтверждён. "+formatFlowPurgeResult(data)) const warnings=[data.partitionError===undefined?null:"Обслуживание таблиц: "+data.partitionError,data.listenerError===undefined?null:"Запуск приёма: "+data.listenerError,data.eventError===undefined?null:"Запись журнала: "+data.eventError].filter(Boolean) if(warnings.length)setOpError(warnings.join("; ")) - }catch(error){if(reader.current.active){setPurgeOpen(false);setNeedsReview(true);setOpError(message(error)+". Результат операции неизвестен; проверьте статус и журнал перед повтором.")}} + }catch(error){if(reader.current.active){setPurgeOpen(false);setNeedsReview(true);setOpError(error instanceof ApiClientError && error.status === 409 ? "Новый сброс не запущен: " + message(error) + ". Проверьте журнал перед новым сбросом." : message(error)+". Результат операции неизвестен; проверьте статус и журнал перед повтором.")}} finally{finish()} if(reader.current.active)await load() } diff --git a/docs/ux-ui-plan.md b/docs/ux-ui-plan.md index 67e4f4a8..a5b27a8e 100644 --- a/docs/ux-ui-plan.md +++ b/docs/ux-ui-plan.md @@ -262,3 +262,5 @@ REST запрос → ожидание → подтверждённый output / Завершён 11Б.3.4.4.5: форма NetFlow known/dirty/rights/lock, selective PUT, partial outcomes и review после unknown POST; 12 E2E/form unit/typecheck/lint/production build прошли. Далее 11Б.3.4.4.6 — source/adaptive приёмка, затем общий контракт IP/CIDR/DNS. Завершён 11Б.3.4.4.6: 10 NetFlow backend source E2E, 3 responsive, 6 снимков (320 и ПК просмотрены), регрессия GeoIP/collector прошли. Далее 11Б.3.4.4.7: общий IP/CIDR/DNS ввод/фактический порт и отказ конкурентному purge вместо очереди. Затем приёмка NetFlow и Evolution. + +Завершён 11Б.3.4.4.7.1: конкурентный purge получает 409 до удаления, неподтверждённый session lock соединения исключается из пула; DI/backend/build/lint/conflict E2E прошли. Реальный PG overlap не проверен. Далее 11Б.3.4.4.7.2 — общий IP/CIDR/DNS контракт и фактические порты. diff --git a/docs/ux-ui-progress.md b/docs/ux-ui-progress.md index 700ebd09..bc367227 100644 --- a/docs/ux-ui-progress.md +++ b/docs/ux-ui-progress.md @@ -28,7 +28,7 @@ | 9 | VPN, Firewall и Filters | Завершён в границах локальной приёмки 2026-09-30; блоки 9.1–9.64, ограничения в 9.64 | | 10 | Маршрутизация, диагностика, карта и терминал | Завершён в границах локальной приёмки 2026-10-01; блоки 10.1–10.71, внешние ограничения в 10.71 | | 11А | Пользователи, контейнеры, сертификаты и резервные копии | Завершён в границах локальной приёмки 2026-10-01; 11А.1–11А.33, внешние ограничения в 11А.33 | -| 11Б | Оповещения, сбор данных, интеграции, релизы и настройки | В работе: 11Б.1–11Б.3.4.4.6; scheduler и GeoIP приняты локально; NetFlow API/backend/форма/source/adaptive проверены, далее общий сетевой ввод и Evolution | +| 11Б | Оповещения, сбор данных, интеграции, релизы и настройки | В работе: 11Б.1–11Б.3.4.4.7.1; scheduler и GeoIP приняты локально; NetFlow API/backend/форма/source/adaptive проверены, далее общий сетевой ввод и Evolution | | 12 | Итоговая доступность, адаптивность, производительность и регрессия | Не начат | @@ -36,6 +36,13 @@ +## Блок 11Б.3.4.4.7.1 — отказ конкурентному purge NetFlow (2026-10-03) + +- Блокирующий advisory lock заменён для purge на pg_try_advisory_lock: второй запрос не становится новым удалением после завершения первого, сразу получает 409. Другие операции и общий withAdvisoryLock не менялись. +- Только владелец снимает session lock; если acquire/unlock не подтверждён или unlock падает, соединение уничтожается при release и не возвращает потенциально занятый lock в пул. Ошибка unlock не отменяет уже подтверждённый receipt операции. +- UI отличает 409: «Новый сброс не запущен», требует проверки перед новым запросом. DI overlap/release/unknown tests, общий purge transaction/HTTP профиль, backend build, ESLint и новый conflict E2E прошли. Первый browser assertion попал в анимацию закрытия (два Сбросить); добавлено ожидание удаления dialog, повтор прошёл. +- Реальный PG overlap/session recovery не проверен. Далее 11Б.3.4.4.7.2 — единый контракт IP/CIDR/DNS и инструкции с фактическим портом, затем локальная приёмка NetFlow. + ## Блок 11Б.3.4.4.6 — контекст и адаптивность NetFlow (2026-10-03) - 10 source E2E проверили поздние GET/PUT/keys/purge/export, success/error старого backend не меняют настройки/ошибки/receipt/Sheet нового backend. Конфигурация :3101/local отдельна от основного :3100/same-origin. diff --git a/docs/ux-ui-stage-11b.md b/docs/ux-ui-stage-11b.md index dcd72821..4c76c708 100644 --- a/docs/ux-ui-stage-11b.md +++ b/docs/ux-ui-stage-11b.md @@ -190,3 +190,5 @@ ReUI PRO/base-nova; существующие Frame, Data Grid v9, KpiStatGrid, S Завершён 11Б.3.4.4.5: форма NetFlow known/dirty/rights/lock, selective PUT, partial outcomes и review после unknown POST; 12 E2E/form unit/typecheck/lint/production build прошли. Далее 11Б.3.4.4.6 — source/adaptive приёмка, затем общий контракт IP/CIDR/DNS. Завершён 11Б.3.4.4.6: 10 NetFlow backend source E2E, 3 responsive, 6 снимков (320 и ПК просмотрены), регрессия GeoIP/collector прошли. Далее 11Б.3.4.4.7: общий IP/CIDR/DNS ввод/фактический порт и отказ конкурентному purge вместо очереди. Затем приёмка NetFlow и Evolution. + +Завершён 11Б.3.4.4.7.1: конкурентный purge получает 409 до удаления, неподтверждённый session lock соединения исключается из пула; DI/backend/build/lint/conflict E2E прошли. Реальный PG overlap не проверен. Далее 11Б.3.4.4.7.2 — общий IP/CIDR/DNS контракт и фактические порты. diff --git a/tests/e2e/netflow-settings.spec.ts b/tests/e2e/netflow-settings.spec.ts index 5b9e6060..ecbf56ce 100644 --- a/tests/e2e/netflow-settings.spec.ts +++ b/tests/e2e/netflow-settings.spec.ts @@ -48,3 +48,5 @@ test("NetFlow: lock удерживает purge dialog и блокирует со test("NetFlow: reader claims видят статус без edit/export/mutations",async({page})=>{ const token="header."+Buffer.from(JSON.stringify({sub:"fixture",apps:["mm"],permissions:["mm:dashboard:read","mm:traffic:read"],exp:Math.floor(Date.now()/1000)+3600})).toString("base64url")+".fixture";await page.addInitScript(value=>localStorage.setItem("mmapp_token",value),token);await setup(page);await page.route("**/api/auth/config",r=>r.fulfill({json:{required:true}}));await page.route("**/api/traffic/flow/settings",r=>r.fulfill({json:keyed}));await open(page);await expect(field(page)).toBeDisabled();await expect(page.getByRole("button",{name:"Скачать файлы хоста NetFlow",exact:true})).toBeDisabled();await expect(page.getByRole("button",{name:"Ключи хоста",exact:true})).toBeDisabled();await expect(refresh(page)).toBeEnabled() }) + +test("NetFlow: конфликт purge известен и не отправляет повтор",async({page})=>{await setup(page);let posts=0;await page.route("**/api/traffic/flow/purge",r=>{posts++;return r.fulfill({status:409,json:{error:"Сброс NetFlow уже выполняется"}})});await open(page);await page.getByRole("button",{name:"Сбросить",exact:true}).click();await page.getByRole("alertdialog").getByRole("button",{name:"Сбросить",exact:true}).click();await expect(page.getByText("Новый сброс не запущен: Сброс NetFlow уже выполняется. Проверьте журнал перед новым сбросом.",{exact:true})).toBeVisible();await expect(page.getByRole("alertdialog")).toHaveCount(0);await expect(page.getByRole("button",{name:"Сбросить",exact:true})).toBeDisabled();expect(posts).toBe(1)})