fix(api): сохранять каталог speed-проб атомарно

backend/src/routes/speed-probes-write.test.ts

@@ -0,0 +1,66 @@
+import assert from "node:assert/strict"
+import Fastify from "fastify"
+import { SignJWT } from "jose"
+import { createSpeedProbesWriteRoute } from "./speed-probes-write.js"
+import { speedProbesWriteSchema, type SpeedProbeWrite } from "@mmapp/contracts"
+import { replaceSpeedProbes, type SpeedProbePatch, type SpeedProbeStore } from "../services/speed-probe-replacement.js"
+import { permissionForRequest } from "../lib/permissions.js"
+const probe = { id: "s1", srcServerId: "1", dstServerId: 2, srcInterface: " ether1 ", dstInterface: "", protocol: "tcp", direction: "both", durationSec: "10", enabled: false }
+let transactions = 0, refreshes = 0, failWrite = false, failScheduler = false
+let rows = new Map<string, SpeedProbePatch & { lastTxAvgMbps: number | null }>([["old", { ...speedProbesWriteSchema.parse({ probes: [probe] }).probes[0], id: "old", sortOrder: 0, updatedAt: "old", lastTxAvgMbps: 12 }]])
+const store: SpeedProbeStore = { transaction: async run => {
+  transactions++
+  const draft = new Map(rows)
+  await run({ listIds: async () => [...draft.keys()], remove: async id => { draft.delete(id) }, upsert: async patch => {
+    draft.set(patch.id, { ...patch, lastTxAvgMbps: draft.get(patch.id)?.lastTxAvgMbps ?? null })
+    if (failWrite) throw new Error("database write failed")
+  } })
+  rows = draft
+} }
+const app = Fastify()
+await app.register(createSpeedProbesWriteRoute({ replace: probes => replaceSpeedProbes(store, probes, "2026-10-04T00:00:00Z"), refreshScheduler: async () => { refreshes++; if (failScheduler) throw new Error("scheduler unavailable") } }), { prefix: "/api" })
+const put = (payload: unknown) => app.inject({ method: "PUT", url: "/api/uptime/speed-probes", headers: { "content-type": "application/json" }, payload: JSON.stringify(payload) })
+for (const payload of [null, {}, { probes: null }, { probes: {} }, { probes: [probe, probe] }, ...[
+  { id: "" }, { srcServerId: "1junk" }, { srcServerId: 0 }, { srcServerId: 9007199254740992 }, { dstServerId: 1 },
+  { durationSec: "10junk" }, { durationSec: 2 }, { durationSec: 3.5 }, { durationSec: 2147483648 },
+  { protocol: "bad" }, { direction: "bad" }, { enabled: "true" }, { enabled: undefined }, { srcInterface: null },
+].map(patch => ({ probes: [probe, { ...probe, id: "s2", ...patch }] }))]) {
+  assert.equal((await put(payload)).statusCode, 400)
+}
+assert.equal(transactions, 0); assert.equal(refreshes, 0); assert.ok(rows.has("old"))
+failWrite = true
+assert.equal((await put({ probes: [probe] })).statusCode, 500)
+assert.ok(rows.has("old")); assert.equal(rows.size, 1); assert.equal(refreshes, 0)
+failWrite = false
+const saved = await put({ probes: [probe] }); assert.equal(saved.statusCode, 200); assert.deepEqual(saved.json(), { ok: true })
+assert.equal(rows.has("old"), false); assert.equal(rows.get("s1")?.enabled, false); assert.equal(rows.get("s1")?.srcInterface, "ether1")
+rows.get("s1")!.lastTxAvgMbps = 42
+await put({ probes: [{ ...probe, durationSec: 20, lastTxAvgMbps: 0 }] })
+assert.equal(rows.get("s1")?.lastTxAvgMbps, 42); assert.equal(rows.get("s1")?.durationSec, 20)
+failScheduler = true
+const warning = await put({ probes: [{ ...probe, durationSec: 30 }] }); assert.equal(warning.statusCode, 200)
+assert.equal(warning.json().ok, true); assert.match(warning.json().warning, /Пробы сохранены/); assert.equal(rows.get("s1")?.durationSec, 30)
+failScheduler = false
+assert.equal((await put({ probes: [] })).statusCode, 200); assert.equal(rows.size, 0)
+assert.equal(permissionForRequest("PUT", "/api/uptime/speed-probes"), "mm:uptime:write")
+// An ordered full replacement strips observations, preserves caller order and does not mutate the input.
+const input = speedProbesWriteSchema.parse({ probes: [probe, { ...probe, id: "s2", enabled: true }] }).probes as SpeedProbeWrite[]
+await replaceSpeedProbes(store, input)
+assert.deepEqual([...rows.values()].map(row => [row.id, row.sortOrder]), [["s1", 0], ["s2", 1]])
+assert.equal("sortOrder" in input[0], false)
+await app.close()
+console.log("Speed probes HTTP/DI: invalid catalog causes no writes; modeled rollback; order/observations; scheduler warning after commit; explicit empty replacement")
+
+process.env.AUTH_REQUIRED = "true"; process.env.AUTH_JWT_SECRET = "speed-probes-test-secret"; process.env.AUTH_ISSUER = "https://auth.test.local"; process.env.NODE_ENV = "test"
+const { default: authPlugin, requireAuth } = await import("../plugins/auth.js")
+const secured = Fastify(); await secured.register(authPlugin); secured.addHook("preHandler", requireAuth)
+let authorizedWrites = 0
+await secured.register(createSpeedProbesWriteRoute({ replace: async () => { authorizedWrites++ }, refreshScheduler: async () => undefined }), { prefix: "/api" })
+const send = (token?: string) => secured.inject({ method: "PUT", url: "/api/uptime/speed-probes", payload: { probes: [] }, headers: token ? { authorization: "Bearer " + token } : {} })
+assert.equal((await send()).statusCode, 401); assert.equal(authorizedWrites, 0)
+for (const permission of ["mm:uptime:read", "mm:servers:write", "mm:uptime:write"]) {
+  const token = await new SignJWT({ sub: "test", apps: ["mm"], permissions: [permission] }).setProtectedHeader({ alg: "HS256" }).setIssuer("https://auth.test.local").setExpirationTime("1h").sign(new TextEncoder().encode("speed-probes-test-secret"))
+  assert.equal((await send(token)).statusCode, permission === "mm:uptime:write" ? 200 : 403)
+}
+assert.equal(authorizedWrites, 1); await secured.close()
+console.log("Speed probes RBAC: signed JWT rejects missing/read/servers-write permissions before mutation")

backend/src/routes/speed-probes-write.ts

@@ -0,0 +1,21 @@
+import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
+import { speedProbesWriteSchema, type SpeedProbeWrite } from "@mmapp/contracts"
+export function createSpeedProbesWriteRoute(deps: {
+  replace: (probes: SpeedProbeWrite[]) => Promise<void>
+  refreshScheduler: () => Promise<unknown>
+}): FastifyPluginAsyncZod {
+  return async app => {
+    app.put("/uptime/speed-probes", async (req, reply) => {
+      const parsed = speedProbesWriteSchema.safeParse(req.body)
+      if (!parsed.success) return reply.code(400).send({ error: "Список speed-проб некорректен. Проверьте устройства, длительность и уникальность записей. Изменения не сохранены." })
+      await deps.replace(parsed.data.probes)
+      try {
+        await deps.refreshScheduler()
+        return reply.send({ ok: true })
+      } catch (error) {
+        req.log.error({ err: error }, "Speed probes saved; scheduler refresh failed")
+        return reply.send({ ok: true, warning: "Пробы сохранены, но расписание не обновлено. Проверьте состояние сборщика данных." })
+      }
+    })
+  }
+}

backend/src/routes/uptime.ts

@@ -1,7 +1,9 @@
+import { createSpeedProbesWriteRoute } from "./speed-probes-write.js"
+import { replaceSpeedProbes } from "../services/speed-probe-replacement.js"
 import { mapSpeedProbeRead } from "../services/speed-probe-dto.js"
 import { mapSpeedHistoryRun } from "../services/speed-history-dto.js"
 import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
-import { asc, desc, eq } from "drizzle-orm"
+import { asc, desc, eq, sql } from "drizzle-orm"
 import { db } from "../db/index.js"
 import { servers, serverSnapshots, uptimeSpeedProbes, uptimeSpeedTestRuns } from "../db/schema.js"
 import { MikrotikClient } from "../services/mikrotik.js"
@@ -198,68 +200,22 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
     }
   })
 
-  app.put("/uptime/speed-probes", async (req, reply) => {
-    const body = req.body as {
-      probes?: Array<{
-        id?: string
-        srcServerId?: string | number
-        dstServerId?: string | number
-        srcInterface?: string
-        dstInterface?: string
-        protocol?: "tcp" | "udp"
-        direction?: "transmit" | "receive" | "both"
-        durationSec?: string | number
-        enabled?: boolean
-      }>
-    }
-    const normalized = (body.probes ?? [])
-      .map((p, i) => ({
-        id: p.id || `sp-${Date.now()}-${i}`,
-        srcServerId: Number.parseInt(String(p.srcServerId ?? ""), 10),
-        dstServerId: Number.parseInt(String(p.dstServerId ?? ""), 10),
-        srcInterface: String(p.srcInterface ?? "").trim(),
-        dstInterface: String(p.dstInterface ?? "").trim(),
-        protocol: p.protocol === "udp" ? "udp" : "tcp",
-        direction: p.direction === "transmit" || p.direction === "receive" ? p.direction : "both",
-        durationSec: Math.max(3, Number.parseInt(String(p.durationSec ?? ""), 10) || 10),
-        enabled: p.enabled !== false,
-      }))
-      .filter((p) => Number.isFinite(p.srcServerId) && Number.isFinite(p.dstServerId) && p.srcServerId !== p.dstServerId)
-
-    const now = new Date().toISOString()
-    const existing = await db.select({ id: uptimeSpeedProbes.id }).from(uptimeSpeedProbes)
-    const nextIds = new Set(normalized.map((p) => p.id))
-
-    for (const row of existing) {
-      if (nextIds.has(row.id)) continue
-      await db.delete(uptimeSpeedProbes).where(eq(uptimeSpeedProbes.id, row.id))
-    }
-
-    for (let i = 0; i < normalized.length; i += 1) {
-      const p = normalized[i]
-      const patch = {
-        srcServerId: p.srcServerId,
-        dstServerId: p.dstServerId,
-        srcInterface: p.srcInterface,
-        dstInterface: p.dstInterface,
-        protocol: p.protocol as "tcp" | "udp",
-        direction: p.direction as "transmit" | "receive" | "both",
-        durationSec: p.durationSec,
-        enabled: p.enabled,
-        sortOrder: i,
-        updatedAt: now,
-      }
-      const updated = await db.update(uptimeSpeedProbes).set(patch).where(eq(uptimeSpeedProbes.id, p.id)).returning({ id: uptimeSpeedProbes.id })
-      if (updated.length > 0) continue
-      await db.insert(uptimeSpeedProbes).values({
-        id: p.id,
-        ...patch,
-        createdAt: now,
+  await app.register(createSpeedProbesWriteRoute({
+    replace: probes => replaceSpeedProbes({ transaction: run => db.transaction(async tx => {
+      // Serialize complete replacements from all backend processes, including an empty catalog.
+      await tx.execute(sql`select pg_advisory_xact_lock(hashtext('uptime-speed-probes'))`)
+      await run({
+        listIds: async () => (await tx.select({ id: uptimeSpeedProbes.id }).from(uptimeSpeedProbes)).map(row => row.id),
+        remove: async id => { await tx.delete(uptimeSpeedProbes).where(eq(uptimeSpeedProbes.id, id)) },
+        upsert: async (row, createdAt) => {
+          const { id, ...patch } = row
+          await tx.insert(uptimeSpeedProbes).values({ id, ...patch, createdAt })
+            .onConflictDoUpdate({ target: uptimeSpeedProbes.id, set: patch })
+        },
       })
-    }
-    await refreshScheduler()
-    return reply.send({ ok: true })
-  })
+    }) }, probes),
+    refreshScheduler,
+  }))
 
   app.put("/uptime/probes", async (req, reply) => {
     const body = req.body as {

backend/src/services/speed-probe-replacement.ts

@@ -0,0 +1,16 @@
+import type { SpeedProbeWrite } from "@mmapp/contracts"
+export type SpeedProbePatch = SpeedProbeWrite & { sortOrder: number; updatedAt: string }
+export type SpeedProbeTransaction = {
+  listIds: () => Promise<string[]>
+  remove: (id: string) => Promise<void>
+  upsert: (row: SpeedProbePatch, createdAt: string) => Promise<void>
+}
+export type SpeedProbeStore = { transaction: (run: (tx: SpeedProbeTransaction) => Promise<void>) => Promise<void> }
+/** The adapter must lock the catalog and commit all configuration changes atomically. Observations are never written here. */
+export async function replaceSpeedProbes(store: SpeedProbeStore, probes: SpeedProbeWrite[], now = new Date().toISOString()): Promise<void> {
+  await store.transaction(async tx => {
+    const nextIds = new Set(probes.map(row => row.id))
+    for (const id of await tx.listIds()) if (!nextIds.has(id)) await tx.remove(id)
+    for (const [sortOrder, row] of probes.entries()) await tx.upsert({ ...row, sortOrder, updatedAt: now }, now)
+  })
+}