fix(api): дождаться запросов перед операцией с базой

backend/package.json

@@ -37,7 +37,7 @@
     "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",
     "test:containers": "tsx src/services/containers-state.test.ts && tsx src/routes/containers-state.test.ts",
     "test:certificates": "tsx src/services/certificate-issue-execution.test.ts && tsx src/services/certificate-issue-request.test.ts && tsx src/routes/certificates-issue.test.ts && tsx src/services/certificate-issue-dto.test.ts && tsx src/services/certificate-parse.test.ts && tsx src/services/certificates-state.test.ts && tsx src/routes/certificates-state.test.ts",
-    "test:system-database": "tsx src/services/traffic-flow-worker-shutdown.test.ts && tsx src/services/system-database-restore-execution.test.ts && tsx src/services/system-database-operation.test.ts && tsx src/routes/system-database.test.ts && tsx src/plugins/auth.system-database.test.ts"
+    "test:system-database": "tsx src/services/database-request-gate.test.ts && tsx src/services/traffic-flow-worker-shutdown.test.ts && tsx src/services/system-database-restore-execution.test.ts && tsx src/services/system-database-operation.test.ts && tsx src/routes/system-database.test.ts && tsx src/plugins/auth.system-database.test.ts"
   },
   "dependencies": {
     "@aws-sdk/client-s3": "^3.888.0",

backend/src/index.ts

@@ -6,6 +6,7 @@ import { env } from "./config.js"
 import { initDatabase } from "./db/bootstrap.js"
 import { closePool } from "./db/index.js"
 import authPlugin, { requireAuth } from "./plugins/auth.js"
+import { installDatabaseRequestGate } from "./plugins/database-request-gate.js"
 import serversRoutes from "./routes/servers.js"
 import bgpRoutes from "./routes/bgp.js"
 import ospfRoutes from "./routes/ospf.js"
@@ -116,6 +117,7 @@ export async function buildApp(opts?: {
     })
   }
 
+  installDatabaseRequestGate(app)
   await app.register(serversRoutes, { prefix: "/api/servers" })
   await app.register(bgpRoutes, { prefix: "/api" })
   await app.register(ospfRoutes, { prefix: "/api" })

backend/src/plugins/database-request-gate.ts

@@ -0,0 +1,31 @@
+import type { FastifyInstance, FastifyRequest } from "fastify"
+import { DatabaseRequestGate } from "../services/database-request-gate.js"
+const streams = new WeakMap<FastifyRequest, { cancelled: boolean; abort?: () => void }>()
+
+export function registerDatabaseStreamAbort(request: FastifyRequest, abort: () => void): void {
+  const state = streams.get(request)
+  if (!state) return
+  state.abort = abort
+  if (state.cancelled) abort()
+}
+
+export function installDatabaseRequestGate(app: FastifyInstance): void {
+  const gate = new DatabaseRequestGate()
+  app.addHook("onRoute", options => {
+    const path = options.url
+    if (!path.startsWith("/api/") || path === "/api/auth/config") return
+    const original = options.handler
+    const operation = path === "/api/system/database/backup" || path === "/api/system/database/restore"
+    const stream = path === "/api/traffic/flow/live" || /^\/api\/traffic\/servers\/:id\/live$/.test(path)
+    options.handler = async function (request, reply) {
+      const run = async () => await original.call(this, request, reply)
+      if (operation) return gate.exclusive(run)
+      if (!stream) return gate.request(run)
+      const state = { cancelled: false, abort: undefined as (() => void) | undefined }
+      streams.set(request, state)
+      try {
+        return await gate.request(run, () => { state.cancelled = true; state.abort?.() })
+      } finally { streams.delete(request) }
+    }
+  })
+}

backend/src/routes/traffic-flow.ts

@@ -1,4 +1,5 @@
 import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
+import { registerDatabaseStreamAbort } from "../plugins/database-request-gate.js"
 import type { FastifyReply, FastifyRequest } from "fastify"
 import { env } from "../config.js"
 import {
@@ -249,6 +250,7 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => {
     }
     const abort = new AbortController()
     const onClose = () => abort.abort()
+    registerDatabaseStreamAbort(req, onClose)
     req.raw.on("close", onClose)
 
     reply.hijack()

backend/src/routes/traffic.ts

@@ -1,4 +1,5 @@
 import { env } from "../config.js"
+import { registerDatabaseStreamAbort } from "../plugins/database-request-gate.js"
 import { desc, eq } from "drizzle-orm"
 import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
 import { db } from "../db/index.js"
@@ -322,6 +323,7 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
     const onlyInterface = q.iface && q.iface !== "__all__" ? q.iface : undefined
     const abort = new AbortController()
     const onClose = () => abort.abort()
+    registerDatabaseStreamAbort(req, onClose)
     req.raw.on("close", onClose)
 
     reply.hijack()

backend/src/services/database-request-gate.test.ts

@@ -0,0 +1,74 @@
+import assert from "node:assert/strict"
+import { test } from "node:test"
+import Fastify from "fastify"
+import { DatabaseRequestGate } from "./database-request-gate.js"
+import { installDatabaseRequestGate, registerDatabaseStreamAbort } from "../plugins/database-request-gate.js"
+
+function deferred() {
+  let resolve!: () => void
+  const promise = new Promise<void>(r => { resolve = r })
+  return { promise, resolve }
+}
+
+test("exclusive operation waits for handler settlement and blocks new requests", async () => {
+  const gate = new DatabaseRequestGate(), pending = deferred(), entered = deferred()
+  const request = gate.request(async () => { entered.resolve(); await pending.promise })
+  await entered.promise
+  let commands = 0
+  const operation = gate.exclusive(async () => { commands++ })
+  await assert.rejects(gate.request(async () => {}), /повторите запрос позже/)
+  await assert.rejects(gate.exclusive(async () => {}), /уже выполняется/)
+  assert.equal(commands, 0)
+  pending.resolve(); await request; await operation
+  assert.equal(commands, 1)
+  await gate.request(async () => {})
+})
+
+test("drain timeout cancels command without declaring an active handler finished", async () => {
+  const gate = new DatabaseRequestGate(), pending = deferred()
+  const request = gate.request(() => pending.promise)
+  let commands = 0
+  await assert.rejects(gate.exclusive(async () => { commands++ }, 10), /операция с базой отменена/)
+  assert.equal(commands, 0)
+  const retry = gate.exclusive(async () => { commands++ })
+  await Promise.resolve(); assert.equal(commands, 0)
+  pending.resolve(); await request; await retry
+  assert.equal(commands, 1)
+})
+
+test("stream cancellation waits for its asynchronous cleanup", async () => {
+  const gate = new DatabaseRequestGate(), aborted = deferred(), cleanup = deferred()
+  const stream = gate.request(async () => { await aborted.promise; await cleanup.promise }, aborted.resolve)
+  let commands = 0
+  const operation = gate.exclusive(async () => { commands++ })
+  await aborted.promise; assert.equal(commands, 0)
+  cleanup.resolve(); await stream; await operation; assert.equal(commands, 1)
+})
+
+test("Fastify gate covers GET, keeps health available and drains late stream registration", async () => {
+  const app = Fastify(); installDatabaseRequestGate(app)
+  const entered = deferred(), read = deferred(), cleanup = deferred(), command = deferred(), leave = deferred()
+  app.get('/health', async () => ({ ok: true }))
+  await app.register(async child => {
+    child.get('/traffic/flow/live', async request => {
+      entered.resolve(); await read.promise
+      const aborted = deferred(); registerDatabaseStreamAbort(request, aborted.resolve)
+      await aborted.promise; await cleanup.promise
+      return { closed: true }
+    })
+    child.get('/servers', async () => ({ servers: [] }))
+    child.post('/system/database/restore', async () => { command.resolve(); await leave.promise; return { ok: true } })
+  }, { prefix: '/api' })
+  try {
+    const stream = app.inject('/api/traffic/flow/live'); await entered.promise
+    const operation = app.inject({ method: 'POST', url: '/api/system/database/restore' })
+    // inject schedules asynchronously: wait for gate to reject a subsequent GET.
+    await new Promise(r => setTimeout(r, 10))
+    assert.equal((await app.inject('/api/servers')).statusCode, 409)
+    assert.equal((await app.inject('/health')).statusCode, 200)
+    read.resolve(); cleanup.resolve(); await stream; await command.promise
+    assert.equal((await app.inject({ method: 'POST', url: '/api/system/database/restore' })).statusCode, 409)
+    leave.resolve(); assert.equal((await operation).statusCode, 200)
+    assert.equal((await app.inject('/api/servers')).statusCode, 200)
+  } finally { read.resolve(); cleanup.resolve(); leave.resolve(); await app.close() }
+})

backend/src/services/database-request-gate.ts

@@ -0,0 +1,39 @@
+function conflict(message: string): Error & { statusCode: number } {
+  return Object.assign(new Error(message), { statusCode: 409 })
+}
+
+/** Handler lifetime, not socket lifetime: disconnect does not finish an active write. */
+export class DatabaseRequestGate {
+  private paused = false
+  private active = new Map<symbol, (() => void) | undefined>()
+  private idle: (() => void) | undefined
+
+  async request<T>(run: () => Promise<T>, cancelStream?: () => void): Promise<T> {
+    if (this.paused) throw conflict("Выполняется операция с базой приложения; повторите запрос позже")
+    const lease = Symbol()
+    this.active.set(lease, cancelStream)
+    try { return await run() }
+    finally {
+      this.active.delete(lease)
+      if (!this.active.size) this.idle?.()
+    }
+  }
+
+  async exclusive<T>(run: () => Promise<T>, timeoutMs = 30_000): Promise<T> {
+    if (this.paused) throw conflict("Операция с базой приложения уже выполняется")
+    this.paused = true
+    let timer: ReturnType<typeof setTimeout> | undefined
+    try {
+      for (const cancel of this.active.values()) cancel?.()
+      if (this.active.size) await new Promise<void>((resolve, reject) => {
+        this.idle = resolve
+        timer = setTimeout(() => reject(conflict("Не удалось дождаться завершения API-запросов; операция с базой отменена")), timeoutMs)
+      })
+      return await run()
+    } finally {
+      if (timer) clearTimeout(timer)
+      this.idle = undefined
+      this.paused = false
+    }
+  }
+}