fix(api): подтвердить остановку NetFlow перед операцией с БД

backend/src/services/serial-operation.ts

@@ -0,0 +1,9 @@
+/** Each caller observes its own result; a rejection does not break the queue. */
+export function createSerialOperation() {
+  let tail: Promise<unknown> = Promise.resolve()
+  return <T>(run: () => Promise<T>): Promise<T> => {
+    const result = tail.then(run, run)
+    tail = result.catch(() => {})
+    return result
+  }
+}

backend/src/services/system-database-backup.ts

@@ -10,8 +10,8 @@ import { withOperationLock } from "../db/operation-lock.js"
 import { executeDatabaseOperation, type DatabaseOperationWarnings } from "./system-database-operation.js"
 import { restartSchedulerAfterDatabaseOperation, stopSchedulerForDatabaseOperation } from "./scheduler.js"
 import {
-  startTrafficFlowListener,
-  stopTrafficFlowListener,
+  restartTrafficFlowListenerAfterDatabaseOperation,
+  stopTrafficFlowListenerForDatabaseOperation,
 } from "./traffic-flow-ingest.js"
 
 const BACKUP_LOCK_KEY = 8_723_102
@@ -48,9 +48,9 @@ function spawnCollected(command: string, args: string[], opts?: { stdin?: Buffer
 
 async function withDatabaseOperation<T extends object>(fn: () => Promise<T>): Promise<T & DatabaseOperationWarnings> {
   return withOperationLock(() => pool.connect(), BACKUP_LOCK_KEY, () => executeDatabaseOperation(fn, {
-    stopListener: stopTrafficFlowListener,
+    stopListener: stopTrafficFlowListenerForDatabaseOperation,
     stopScheduler: stopSchedulerForDatabaseOperation,
-    startListener: startTrafficFlowListener,
+    startListener: restartTrafficFlowListenerAfterDatabaseOperation,
     startScheduler: restartSchedulerAfterDatabaseOperation,
   }), "Операция с базой приложения уже выполняется")
 }

backend/src/services/traffic-flow-collector-ipc.ts

@@ -32,9 +32,11 @@ export interface CollectorHeartbeat {
 export type MainToWorker =
   | { type: "start"; payload: CollectorStartPayload }
   | { type: "stop" }
+  | { type: "shutdown" }
   | { type: "updateExporterMap"; payload: ExporterMapPayload }
   | { type: "updateSettings"; payload: { topN: number; retentionHours: number } }
 
 export type WorkerToMain =
   | { type: "heartbeat"; payload: CollectorHeartbeat }
   | { type: "error"; payload: { message: string } }
+  | { type: "shutdown-complete" }

backend/src/services/traffic-flow-collector-worker.ts

@@ -1,5 +1,6 @@
 import { createSocket, type Socket } from "node:dgram"
 import { parentPort } from "node:worker_threads"
+import { pool } from "../db/index.js"
 import type {
   CollectorStartPayload,
   ExporterMapPayload,
@@ -21,6 +22,24 @@ let socket: Socket | null = null
 let flushTimer: ReturnType<typeof setInterval> | null = null
 let bound = false
 let address: string | null = null
+let shuttingDown = false
+
+async function shutdown(): Promise<void> {
+  shuttingDown = true
+  if (flushTimer) clearInterval(flushTimer)
+  flushTimer = null
+  const current = socket
+  socket = null
+  bound = false
+  address = null
+  if (current) await new Promise<void>((resolve, reject) => {
+    try { current.close(() => resolve()) } catch (error) { reject(error) }
+  })
+  await flushPending({ force: true, prune: false })
+  await pool.end()
+  send({ type: "shutdown-complete" })
+  parentPort?.close()
+}
 
 function send(msg: WorkerToMain): void {
   parentPort?.postMessage(msg)
@@ -85,6 +104,7 @@ function startListener(payload: CollectorStartPayload): void {
     heartbeat()
   })
   sock.on("message", (msg, rinfo) => {
+    if (shuttingDown) return
     try {
       ingestDatagram(msg, rinfo.address)
     } catch (e) {
@@ -111,6 +131,14 @@ function startListener(payload: CollectorStartPayload): void {
 }
 
 parentPort?.on("message", (msg: MainToWorker) => {
+  if (shuttingDown) return
+  if (msg.type === "shutdown") {
+    void shutdown().catch(() => {
+      send({ type: "error", payload: { message: "Не удалось завершить остановку NetFlow worker" } })
+      // No acknowledgement: the caller must not start a database command.
+    })
+    return
+  }
   try {
     if (msg.type === "start") startListener(msg.payload)
     else if (msg.type === "stop") {

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

@@ -1,4 +1,5 @@
 import { dbAll, dbQuery, pool } from "../db/index.js"
+import { createSerialOperation } from "./serial-operation.js"
 import { ensurePartitionFor, specForParent } from "../db/partitions.js"
 import { normalizeParsedFlow, parseFlowPacket, protoName, type ParsedFlow, type ParsedFlowInput } from "./traffic-flow-parse.js"
 import { classifyFlowPlaneLite } from "./traffic-flow-planes.js"
@@ -932,7 +933,13 @@ async function upsertFlowBuckets(rows: PendingFlowRow[]): Promise<number> {
   }
 }
 
-export async function flushPending(opts?: { force?: boolean; prune?: boolean }): Promise<void> {
+const serializeFlush = createSerialOperation()
+
+export function flushPending(opts?: { force?: boolean; prune?: boolean }): Promise<void> {
+  return serializeFlush(() => flushPendingInternal(opts))
+}
+
+async function flushPendingInternal(opts?: { force?: boolean; prune?: boolean }): Promise<void> {
   pruneRecent()
   rollFlowRings()
   const force = Boolean(opts?.force)

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 { waitForCollectorShutdown } from "./traffic-flow-worker-shutdown.js"
 import { db, dbAll, pool } from "../db/index.js"
 import { dropExpiredPartitions } from "../db/partitions.js"
 import { servers } from "../db/schema.js"
@@ -55,6 +56,7 @@ let restartAttempts = 0
 let lastHeartbeat: CollectorHeartbeat | null = null
 let state: FlowListenerState = { bound: false, address: null }
 let wantListen = false
+let databaseStop: Promise<void> | null = null
 
 function workerFileUrl(): URL {
   const ts = import.meta.url.includes(".ts")
@@ -115,10 +117,14 @@ function handleWorkerMessage(msg: WorkerToMain): void {
 }
 
 async function spawnWorker(): Promise<void> {
+  if (databaseStop) return
   stopWorkerProcess()
   const settings = await getTrafficFlowSettingsRow()
+  if (databaseStop) return
   configureEngine({ topN: settings.topN, retentionHours: settings.retentionHours })
   await applyExporterCtxFromDb()
+  const exporterMap = await buildExporterMapPayload()
+  if (databaseStop) return
   const w = new Worker(workerFileUrl(), { execArgv: process.execArgv })
   w.on("message", (msg: WorkerToMain) => handleWorkerMessage(msg))
   w.on("error", (err) => {
@@ -140,6 +146,7 @@ async function spawnWorker(): Promise<void> {
         }
   })
   w.on("exit", (code) => {
+    if (worker !== w) return
     worker = null
     state = { bound: false, address: null }
     if (!wantListen) return
@@ -159,7 +166,7 @@ async function spawnWorker(): Promise<void> {
       listenPort: settings.flowListenPort,
       topN: settings.topN,
       retentionHours: settings.retentionHours,
-      exporterMap: await buildExporterMapPayload(),
+      exporterMap,
     },
   })
 }
@@ -255,8 +262,10 @@ export async function getFlowRuntimeCounters() {
 }
 
 export async function startTrafficFlowListener() {
+  if (databaseStop) throw new Error("Приём NetFlow приостановлен для операции с базой")
   stopTrafficFlowListener()
   const settings = await getTrafficFlowSettingsRow()
+  if (databaseStop) throw new Error("Приём NetFlow приостановлен для операции с базой")
   if (!settings.enabled) {
     wantListen = false
     state = { bound: false, address: null }
@@ -273,6 +282,28 @@ export function stopTrafficFlowListener() {
   state = { bound: false, address: null }
 }
 
+export async function stopTrafficFlowListenerForDatabaseOperation(): Promise<void> {
+  if (databaseStop) return databaseStop
+  wantListen = false
+  if (restartTimer) clearTimeout(restartTimer)
+  restartTimer = null
+  const current = worker
+  databaseStop = (async () => {
+    if (current) await waitForCollectorShutdown(current)
+    await flushPending({ force: true, prune: false })
+    state = { bound: false, address: null }
+    lastHeartbeat = null
+  })()
+  // Keep a failed stop latched: restart must not kill a worker whose writes are unconfirmed.
+  await databaseStop
+}
+
+export async function restartTrafficFlowListenerAfterDatabaseOperation(): Promise<void> {
+  if (databaseStop) await databaseStop
+  databaseStop = null
+  await startTrafficFlowListener()
+}
+
 export async function refreshFlowExporterMap(): Promise<void> {
   await applyExporterCtxFromDb()
   postToWorker({ type: "updateExporterMap", payload: await buildExporterMapPayload() })

backend/src/services/traffic-flow-worker-shutdown.test.ts

@@ -0,0 +1,58 @@
+import assert from "node:assert/strict"
+import { EventEmitter } from "node:events"
+import { test } from "node:test"
+import { Worker } from "node:worker_threads"
+import { waitForCollectorShutdown } from "./traffic-flow-worker-shutdown.js"
+import { createSerialOperation } from "./serial-operation.js"
+
+function fixture() {
+  const events = new EventEmitter()
+  let commands = 0
+  Object.assign(events, { postMessage: (msg: { type: string }) => { assert.equal(msg.type, "shutdown"); commands++ } })
+  return { events, worker: events as unknown as Worker, commands: () => commands }
+}
+test("acknowledgement alone does not authorize a database operation", async () => {
+  const f = fixture(); let settled = false
+  const stop = waitForCollectorShutdown(f.worker).then(() => { settled = true })
+  f.events.emit("message", { type: "shutdown-complete" })
+  await Promise.resolve(); assert.equal(settled, false)
+  f.events.emit("exit", 0); await stop
+  assert.equal(settled, true); assert.equal(f.commands(), 1)
+  assert.equal(f.events.listenerCount("exit"), 0)
+})
+test("unconfirmed exit, error exit and timeout reject without terminating the worker", async () => {
+  for (const variant of ["unconfirmed", "error", "timeout"]) {
+    const f = fixture(); const stop = waitForCollectorShutdown(f.worker, 10)
+    const rejected = assert.rejects(stop, /операция с базой отменена/)
+    if (variant === "error") f.events.emit("message", { type: "shutdown-complete" })
+    if (variant !== "timeout") f.events.emit("exit", variant === "error" ? 1 : 0)
+    await rejected
+    assert.equal(f.commands(), 1)
+    assert.equal(f.events.listenerCount("message"), 0)
+  }
+})
+test("final flush waits for an in-flight flush, even when it rejects", async () => {
+  const serial = createSerialOperation(); const calls: string[] = []
+  let release!: () => void
+  const gate = new Promise<void>(resolve => { release = resolve })
+  const first = serial(async () => { calls.push("first"); await gate; throw Error("write failed") })
+  const rejected = assert.rejects(first, /write failed/)
+  const final = serial(async () => { calls.push("final") })
+  await Promise.resolve(); assert.deepEqual(calls, ["first"])
+  release(); await rejected; await final
+  assert.deepEqual(calls, ["first", "final"])
+})
+
+test("a real worker acknowledges shutdown and exits without forced termination", async () => {
+  const worker = new Worker(`
+    const { parentPort } = require('node:worker_threads');
+    parentPort.on('message', msg => {
+      if (msg.type === 'shutdown') {
+        parentPort.postMessage({ type: 'shutdown-complete' });
+        parentPort.close();
+      }
+    });
+  `, { eval: true })
+  try { await waitForCollectorShutdown(worker, 5_000) }
+  finally { await worker.terminate() }
+})

backend/src/services/traffic-flow-worker-shutdown.ts

@@ -0,0 +1,22 @@
+import type { Worker } from "node:worker_threads"
+import type { WorkerToMain } from "./traffic-flow-collector-ipc.js"
+
+/** Both an acknowledgement after pool.end() and a clean exit are required. */
+export function waitForCollectorShutdown(worker: Worker, timeoutMs = 30_000): Promise<void> {
+  return new Promise((resolve, reject) => {
+    let acknowledged = false
+    const cleanup = () => {
+      clearTimeout(timer)
+      worker.off("message", message); worker.off("exit", exit); worker.off("error", error)
+    }
+    const error = () => { cleanup(); reject(new Error("Остановка NetFlow worker не подтверждена; операция с базой отменена")) }
+    const message = (msg: WorkerToMain) => { if (msg.type === "shutdown-complete") acknowledged = true }
+    const exit = (code: number) => {
+      if (!acknowledged || code !== 0) return error()
+      cleanup(); resolve()
+    }
+    const timer = setTimeout(error, timeoutMs)
+    worker.on("message", message); worker.on("exit", exit); worker.on("error", error)
+    try { worker.postMessage({ type: "shutdown" }) } catch { error() }
+  })
+}