fix(api): дождаться фоновых записей перед операцией с БД

backend/src/routes/backups.ts

@@ -1,3 +1,4 @@
+import { databaseBackgroundWork } from "../services/database-background-work.js"
 import { randomUUID } from "node:crypto"
 import { z } from "zod"
 import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
@@ -122,7 +123,7 @@ export function createBackupsRoutes(deps: { listServers?: () => Promise<Array<{
       return reply.status(202).send({ jobId, status: raced.status, total: raced.total, completed: raced.completed, serverIds: raced.serverIds })
     }
     backupJobs.set(jobId, job)
-    void event({
+    void databaseBackgroundWork.track(event({
       level: "info",
       eventType: "backups.job.started",
       sourceModule: "backups",
@@ -134,9 +135,9 @@ export function createBackupsRoutes(deps: { listServers?: () => Promise<Array<{
         serverIds: ids,
         notes: notes ?? null,
       },
-    })
-    queueMicrotask(async () => {
-      void processBackupJob(job, ids, notes, { create: deps.create ?? runBackupForServer, event }).catch(async (err) => {
+    })).catch(() => {})
+    void databaseBackgroundWork.track(Promise.resolve().then(async () => {
+      await processBackupJob(job, ids, notes, { create: deps.create ?? runBackupForServer, event }).catch(async (err) => {
         job.status = "failed"
         job.finishedAt = new Date().toISOString()
         const failure = {
@@ -158,7 +159,7 @@ export function createBackupsRoutes(deps: { listServers?: () => Promise<Array<{
           },
         })
       })
-    })
+    })).catch(() => {})
     return reply.status(202).send({
       jobId,
       status: job.status,
@@ -190,7 +191,7 @@ export function createBackupsRoutes(deps: { listServers?: () => Promise<Array<{
   app.post("/backups/:id/restore", { schema: { params: BackupIdParamSchema } }, async (req, reply) => {
     try {
       const result = await (deps.restore ?? restoreBackupToDevice)(req.params.id)
-      void event({
+      void databaseBackgroundWork.track(event({
         level: "warning",
         eventType: "backups.restore",
         sourceModule: "backups",
@@ -198,7 +199,7 @@ export function createBackupsRoutes(deps: { listServers?: () => Promise<Array<{
         message: `${result.filename} → ${result.serverName}`,
         entityType: "backup",
         entityId: req.params.id,
-      })
+      })).catch(() => {})
       return reply.send(result)
     } catch (err) {
       return reply.status(err instanceof BackupRestoreUnknownError ? 502 : 400).send({

backend/src/services/certificate-issue-execution.ts

@@ -1,3 +1,4 @@
+import { databaseBackgroundWork } from "./database-background-work.js"
 import { certificateIssueStepSchema } from "@mmapp/contracts/certificates"
 import type { CertificateIssueJobRow } from "../db/schema.js"
 import type { appendEvent } from "../modules/events/service/events-service.js"
@@ -70,7 +71,7 @@ export function createCertificateIssueRunner<S extends { id: number; name: strin
     } finally { runningJobs.delete(jobId) }
   }
   function queue(jobId: string): void {
-    queueMicrotask(() => { void run(jobId).catch(report) })
+    void databaseBackgroundWork.track(Promise.resolve().then(() => run(jobId))).catch(report)
   }
   return { run, queue }
 }

backend/src/services/database-background-work.test.ts

@@ -0,0 +1,62 @@
+import assert from "node:assert/strict"
+import { test } from "node:test"
+import { DatabaseBackgroundWork, databaseBackgroundWork } from "./database-background-work.js"
+import Fastify from "fastify"
+import { validatorCompiler, serializerCompiler } from "@fastify/type-provider-zod"
+import { createBackupsRoutes } from "../routes/backups.js"
+
+function deferred() {
+  let resolve!: () => void
+  const promise = new Promise<void>(r => { resolve = r })
+  return { promise, resolve }
+}
+
+test("drain includes writes scheduled by a task after its HTTP response", async () => {
+  const work = new DatabaseBackgroundWork(), parent = deferred(), child = deferred()
+  work.track(parent.promise.then(() => { work.track(child.promise) }))
+  let finished = false
+  const drain = work.drain().then(() => { finished = true })
+  parent.resolve(); await parent.promise; await Promise.resolve()
+  assert.equal(finished, false)
+  child.resolve(); await drain; assert.equal(finished, true)
+})
+
+test("timeout preserves pending tasks for the next attempt", async () => {
+  const work = new DatabaseBackgroundWork(), pending = deferred()
+  work.track(pending.promise)
+  await assert.rejects(work.drain(10), /операция с базой отменена/)
+  let finished = false
+  const retry = work.drain().then(() => { finished = true })
+  await Promise.resolve(); assert.equal(finished, false)
+  pending.resolve(); await retry; assert.equal(finished, true)
+})
+
+test("a task rejection is observed without breaking subsequent drain", async () => {
+  const work = new DatabaseBackgroundWork()
+  const task = work.track(Promise.reject(Error("audit failed")))
+  await assert.rejects(task, /audit failed/)
+  await work.drain()
+  work.track(Promise.resolve()); await work.drain()
+})
+
+test("real backup HTTP202 keeps its detached job registered through completion", async () => {
+  const pending = deferred(), started = deferred()
+  const app = Fastify(); app.setValidatorCompiler(validatorCompiler); app.setSerializerCompiler(serializerCompiler)
+  await app.register(createBackupsRoutes({
+    listServers: async () => [{ id: 1 }],
+    appendEvent: async () => undefined as never,
+    create: async () => {
+      started.resolve(); await pending.promise
+      return { id: "drain-backup", serverId: 1, serverName: "router", filename: "test.rsc", sizeBytes: 0,
+        createdAt: "2026-10-07T00:00:00Z", kind: "manual", storage: "local" }
+    },
+  }), { prefix: "/api" })
+  try {
+    assert.equal((await app.inject({ method: "POST", url: "/api/backups/create", payload: { serverIds: [1] } })).statusCode, 202)
+    await started.promise
+    let finished = false
+    const drain = databaseBackgroundWork.drain().then(() => { finished = true })
+    await Promise.resolve(); assert.equal(finished, false)
+    pending.resolve(); await drain; assert.equal(finished, true)
+  } finally { pending.resolve(); await databaseBackgroundWork.drain(); await app.close() }
+})

backend/src/services/database-background-work.ts

@@ -0,0 +1,22 @@
+/** Includes child writes registered by tasks that are already running. */
+export class DatabaseBackgroundWork {
+  private pending = new Set<Promise<unknown>>()
+
+  track<T>(task: Promise<T>): Promise<T> {
+    this.pending.add(task)
+    void task.then(() => this.pending.delete(task), () => this.pending.delete(task))
+    return task
+  }
+
+  async drain(timeoutMs = 30_000): Promise<void> {
+    let timer: ReturnType<typeof setTimeout> | undefined
+    const timeout = new Promise<never>((_, reject) => {
+      timer = setTimeout(() => reject(new Error("Не удалось дождаться фоновых записей; операция с базой отменена")), timeoutMs)
+    })
+    try {
+      while (this.pending.size) await Promise.race([Promise.allSettled([...this.pending]), timeout])
+    } finally { if (timer) clearTimeout(timer) }
+  }
+}
+
+export const databaseBackgroundWork = new DatabaseBackgroundWork()

backend/src/services/poller.ts

@@ -1,3 +1,4 @@
+import { databaseBackgroundWork } from "./database-background-work.js"
 import { eq } from "drizzle-orm"
 import { db, dbQuery, pool } from "../db/index.js"
 import { dropExpiredPartitions, ensurePartitionsAround } from "../db/partitions.js"
@@ -91,7 +92,7 @@ export async function pollServer(serverId: number): Promise<SnapshotRead> {
       [serverId, inserted.polledAt],
     )
   }
-  void dropExpiredPartitions(pool).then(() => ensurePartitionsAround(pool))
+  void databaseBackgroundWork.track(dropExpiredPartitions(pool).then(() => ensurePartitionsAround(pool))).catch(() => {})
 
   return toSnapshotRead(inserted!)
 }

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

@@ -9,6 +9,7 @@ import { pool } from "../db/index.js"
 import { withOperationLock } from "../db/operation-lock.js"
 import { executeDatabaseOperation, type DatabaseOperationWarnings } from "./system-database-operation.js"
 import { restartSchedulerAfterDatabaseOperation, stopSchedulerForDatabaseOperation } from "./scheduler.js"
+import { pauseRipeBackgroundWork, resumeRipeBackgroundWork } from "./traffic-flow-ripe.js"
 import {
   restartTrafficFlowListenerAfterDatabaseOperation,
   stopTrafficFlowListenerForDatabaseOperation,
@@ -49,9 +50,15 @@ 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: stopTrafficFlowListenerForDatabaseOperation,
-    stopScheduler: stopSchedulerForDatabaseOperation,
+    stopScheduler: async () => {
+      await stopSchedulerForDatabaseOperation()
+      await pauseRipeBackgroundWork()
+    },
     startListener: restartTrafficFlowListenerAfterDatabaseOperation,
-    startScheduler: restartSchedulerAfterDatabaseOperation,
+    startScheduler: async () => {
+      resumeRipeBackgroundWork()
+      await restartSchedulerAfterDatabaseOperation()
+    },
   }), "Операция с базой приложения уже выполняется")
 }
 

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

@@ -1,3 +1,4 @@
+import { databaseBackgroundWork } from "./database-background-work.js"
 import { OTHER_SERVICE, resolveFlowBrand } from "./traffic-flow-brands.js"
 import { db } from "../db/index.js"
 import { evobgpSettings } from "../db/schema.js"
@@ -149,5 +150,5 @@ async function fetchCatalog(): Promise<void> {
 
 /** Background refresh — analytics never awaits the HTTP. */
 export function refreshFlowCatalogInBackground(): void {
-  void fetchCatalog()
+  void databaseBackgroundWork.track(fetchCatalog())
 }

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

@@ -1,6 +1,7 @@
 import { createSocket, type Socket } from "node:dgram"
 import { parentPort } from "node:worker_threads"
 import { pool } from "../db/index.js"
+import { pauseRipeBackgroundWork } from "./traffic-flow-ripe.js"
 import type {
   CollectorStartPayload,
   ExporterMapPayload,
@@ -36,6 +37,7 @@ async function shutdown(): Promise<void> {
     try { current.close(() => resolve()) } catch (error) { reject(error) }
   })
   await flushPending({ force: true, prune: false })
+  await pauseRipeBackgroundWork()
   await pool.end()
   send({ type: "shutdown-complete" })
   parentPort?.close()

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

@@ -1,3 +1,4 @@
+import { databaseBackgroundWork } from "./database-background-work.js"
 import { dbAll, dbQuery, pool } from "../db/index.js"
 import { createSerialOperation } from "./serial-operation.js"
 import { ensurePartitionFor, specForParent } from "../db/partitions.js"
@@ -369,8 +370,8 @@ export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): vo
   const ripeMisses: string[] = []
   const topo = peekFlowTopology()
   const catalog = peekServerCatalog()
-  if (!topo) void loadFlowTopology().catch(() => {})
-  if (!catalog) void getServerCatalog().catch(() => {})
+  if (!topo) void databaseBackgroundWork.track(loadFlowTopology()).catch(() => {})
+  if (!catalog) void databaseBackgroundWork.track(getServerCatalog()).catch(() => {})
   const serverType = catalog?.byId.get(serverId)?.type
   for (const raw of flows) {
     const flow = normalizeParsedFlow(raw)

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

@@ -1,3 +1,4 @@
+import { databaseBackgroundWork } from "./database-background-work.js"
 import { eq } from "drizzle-orm"
 import { db } from "../db/index.js"
 import { servers } from "../db/schema.js"
@@ -47,7 +48,7 @@ export async function refreshServerIfaces(serverId: number, force = false): Prom
 
 export function maybeRefreshIfaces(serverId: number): boolean {
   if (!shouldRefreshIfaces(serverId)) return false
-  void refreshIfacesImpl(serverId)
+  void databaseBackgroundWork.track(refreshIfacesImpl(serverId)).catch(() => {})
   return true
 }
 

backend/src/services/traffic-flow-ripe.test.ts

@@ -10,6 +10,8 @@ import {
   ripeLastCandidateCountForTests,
   seedRipeCacheForTests,
   setRipeFetchForTests,
+  pauseRipeBackgroundWork,
+  resumeRipeBackgroundWork,
 } from "./traffic-flow-ripe.js"
 
 disableRipePersistForTests()
@@ -133,4 +135,16 @@ assert.ok(
   `index should not scan all prefixes, got ${ripeLastCandidateCountForTests()}`,
 )
 
-console.log("traffic-flow-ripe.test.ts: ok")
+resetRipeCacheForTests()
+disableRipePersistForTests()
+setRipeFetchForTests(async () => new Response(JSON.stringify({ data: {} }), { status: 200 }))
+await pauseRipeBackgroundWork()
+enqueueRipeMisses(["8.8.4.4"])
+await flushRipeQueueForTests()
+assert.equal(ripeFetchCountForTests(), 0)
+resumeRipeBackgroundWork()
+enqueueRipeMisses(["8.8.4.4"])
+await flushRipeQueueForTests()
+assert.ok(ripeFetchCountForTests() > 0)
+
+console.log("traffic-flow-ripe.test.ts: ok (including database pause/resume)")

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

@@ -1,3 +1,4 @@
+import { databaseBackgroundWork } from "./database-background-work.js"
 import { dbAll, dbQuery } from "../db/index.js"
 import { canonicalIp, ipv4ToInt, isNonPublicIp, parseCidrV4 } from "./traffic-flow-ip.js"
 import { resolveRipeCountry } from "./traffic-flow-brands.js"
@@ -26,6 +27,7 @@ const asnHolder = new Map<number, { holder: string; fetchedAt: number }>()
 const inflight = new Map<string, Promise<FlowIpMeta | null>>()
 const queue: string[] = []
 const queued = new Set<string>()
+let databasePaused = false
 const recentFetches: number[] = []
 
 interface RipeIndexed {
@@ -155,7 +157,7 @@ function loadSqlite(): void {
     return
   }
   loaded = true
-  void loadPg()
+  void databaseBackgroundWork.track(loadPg())
 }
 
 async function loadPg(): Promise<void> {
@@ -204,7 +206,7 @@ async function loadPg(): Promise<void> {
 
 function persist(entry: FlowIpMeta): void {
   if (!persistEnabled) return
-  void dbQuery(`
+  void databaseBackgroundWork.track(dbQuery(`
       INSERT INTO flow_ip_meta (prefix, asn, country, lat, lng, holder, ok, fetched_at)
       VALUES (@prefix, @asn, @country, @lat, @lng, @holder, @ok, @fetchedAt)
       ON CONFLICT(prefix) DO UPDATE SET
@@ -219,12 +221,12 @@ function persist(entry: FlowIpMeta): void {
     holder: entry.holder,
     ok: entry.ok ? 1 : 0,
     fetchedAt: new Date(entry.fetchedAt).toISOString(),
-  }).catch(() => { /* ignore persist errors */ })
+  })).catch(() => { /* ignore persist errors */ })
 }
 
 function persistAsn(asn: number, holder: string): void {
   if (!persistEnabled || !asn) return
-  void dbQuery(`
+  void databaseBackgroundWork.track(dbQuery(`
       INSERT INTO flow_asn_meta (asn, holder, fetched_at)
       VALUES (@asn, @holder, @fetchedAt)
       ON CONFLICT(asn) DO UPDATE SET holder=excluded.holder, fetched_at=excluded.fetched_at
@@ -232,7 +234,7 @@ function persistAsn(asn: number, holder: string): void {
     asn,
     holder,
     fetchedAt: new Date().toISOString(),
-  }).catch(() => { /* ignore */ })
+  })).catch(() => { /* ignore */ })
 }
 
 /** Удаляет просроченный RIPE-кэш с диска (hit 24h / negative 6h). */
@@ -434,13 +436,13 @@ async function runWorker(): Promise<void> {
     }
   } finally {
     workerRunning = false
-    if (queue.length) void runWorker()
+    if (queue.length) void databaseBackgroundWork.track(runWorker())
   }
 }
 
 /** HTTP / SSE never await this — cache miss is filled on a later tick. */
 export function enqueueRipeMisses(ips: Iterable<string>): void {
-  if (!enqueueEnabled) return
+  if (!enqueueEnabled || databasePaused) return
   loadSqlite()
   for (const raw of ips) {
     if (queue.length >= MAX_QUEUE) break
@@ -451,5 +453,16 @@ export function enqueueRipeMisses(ips: Iterable<string>): void {
     queued.add(ip)
     queue.push(ip)
   }
-  if (queue.length) void runWorker()
+  if (queue.length) void databaseBackgroundWork.track(runWorker())
+}
+
+export async function pauseRipeBackgroundWork(): Promise<void> {
+  databasePaused = true
+  queue.length = 0
+  queued.clear()
+  await databaseBackgroundWork.drain()
+}
+
+export function resumeRipeBackgroundWork(): void {
+  databasePaused = false
 }