fix(uptime): защитить пару устройств и атомарно сохранить результат

backend/src/routes/uptime.ts

@@ -1,3 +1,4 @@
+import { SpeedProbeEndpointsChanged } from "../services/speed-run-persistence.js"
 import { createPingProbesWriteRoute } from "./ping-probes-write.js"
 import { createSpeedProbesWriteRoute } from "./speed-probes-write.js"
 import { replaceSpeedProbes } from "../services/speed-probe-replacement.js"
@@ -308,6 +309,7 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
       const message = /timed out|socket hang up|ECONNRESET/i.test(raw)
         ? "BTTest timeout/socket hang up: проверь доступность destination, /tool bandwidth-server, firewall и корректность интерфейсов"
         : raw
+      if (e instanceof SpeedProbeEndpointsChanged) return reply.status(409).send({ error: message })
       if (/активного IP|Server not found/i.test(raw)) {
         return reply.status(400).send({ error: message })
       }

backend/src/services/speed-run-persistence.test.ts

@@ -0,0 +1,66 @@
+import { test } from "node:test"
+import assert from "node:assert/strict"
+import { assertSpeedProbeEndpoints, matchesSpeedProbeEndpoints, persistSpeedRun, SpeedProbeEndpointsChanged } from "./speed-run-persistence.js"
+const target = { id: "probe", srcServerId: 1, dstServerId: 2 }
+type State = { probe: (typeof target & { average: number }) | null; runs: Array<{ probeId: string | null; average: number }> }
+function harness(failure?: "update" | "insert" | "commit") {
+  let state: State = { probe: { ...target, average: 7 }, runs: [] }
+  let updates = 0
+  return {
+    state: () => state,
+    updates: () => updates,
+    deps: {
+      transaction: async (work: (tx: State) => Promise<void>) => {
+        const tx = structuredClone(state)
+        await work(tx)
+        if (failure === "commit") throw Error("commit failed")
+        state = tx
+      },
+      updateProbe: async (tx: State, expected: typeof target) => {
+        updates++
+        if (!matchesSpeedProbeEndpoints(tx.probe, expected)) return false
+        tx.probe!.average = 0
+        if (failure === "update") throw Error("update failed")
+        return true
+      },
+      insertRun: async (tx: State, probeId: string | null) => {
+        tx.runs.push({ probeId, average: 0 })
+        if (failure === "insert") throw Error("insert failed")
+      },
+    },
+  }
+}
+test("preflight rejects foreign, changed or deleted probe endpoints", () => {
+  assert.doesNotThrow(() => assertSpeedProbeEndpoints(target, target))
+  for (const row of [null, undefined, {}, { ...target, id: "other" }, { ...target, srcServerId: 3 }, { ...target, dstServerId: 3 }, { ...target, srcServerId: "1" }]) {
+    assert.throws(() => assertSpeedProbeEndpoints(row, target), SpeedProbeEndpointsChanged)
+  }
+})
+test("one transaction publishes an explicit zero snapshot and associated history", async () => {
+  const h = harness()
+  await persistSpeedRun(target, h.deps)
+  assert.equal(h.state().probe!.average, 0)
+  assert.deepEqual(h.state().runs, [{ probeId: "probe", average: 0 }])
+  assert.equal(h.updates(), 1)
+})
+for (const changed of ["source", "destination", "deleted"] as const) test(`probe ${changed} during native test retains its snapshot and unlinks old run`, async () => {
+  const h = harness()
+  if (changed === "deleted") h.state().probe = null
+  else h.state().probe![changed === "source" ? "srcServerId" : "dstServerId"] = 3
+  await persistSpeedRun(target, h.deps)
+  assert.equal(h.state().probe?.average ?? 7, 7)
+  assert.deepEqual(h.state().runs, [{ probeId: null, average: 0 }])
+})
+for (const failure of ["update", "insert", "commit"] as const) test(`${failure} error never publishes a partial snapshot/history`, async () => {
+  const h = harness(failure)
+  await assert.rejects(persistSpeedRun(target, h.deps), new RegExp(failure))
+  assert.equal(h.state().probe!.average, 7)
+  assert.deepEqual(h.state().runs, [])
+})
+test("ad-hoc run writes history without touching any probe", async () => {
+  const h = harness()
+  await persistSpeedRun(null, h.deps)
+  assert.equal(h.state().probe!.average, 7)
+  assert.equal(h.updates(), 0)
+  assert.deepEqual(h.state().runs, [{ probeId: null, average: 0 }])
+})

backend/src/services/speed-run-persistence.ts

@@ -0,0 +1,25 @@
+export interface SpeedProbeEndpoints { id: string; srcServerId: number; dstServerId: number }
+
+export class SpeedProbeEndpointsChanged extends Error {
+  constructor() { super("Пара устройств speed-пробы изменилась. Обновите список проб перед запуском") }
+}
+export function matchesSpeedProbeEndpoints(row: unknown, target: SpeedProbeEndpoints): boolean {
+  if (!row || typeof row !== "object") return false
+  const probe = row as Partial<SpeedProbeEndpoints>
+  return probe.id === target.id && probe.srcServerId === target.srcServerId && probe.dstServerId === target.dstServerId
+}
+export function assertSpeedProbeEndpoints(row: unknown, target: SpeedProbeEndpoints): void {
+  if (!matchesSpeedProbeEndpoints(row, target)) throw new SpeedProbeEndpointsChanged()
+}
+
+/** Changed/deleted probes retain their snapshot; history keeps the actual run without a stale association. */
+export async function persistSpeedRun<C>(target: SpeedProbeEndpoints | null, deps: {
+  transaction: (work: (context: C) => Promise<void>) => Promise<void>
+  updateProbe: (context: C, target: SpeedProbeEndpoints) => Promise<boolean>
+  insertRun: (context: C, associatedProbeId: string | null) => Promise<void>
+}): Promise<void> {
+  await deps.transaction(async context => {
+    const associated = target ? await deps.updateProbe(context, target) : false
+    await deps.insertRun(context, associated && target ? target.id : null)
+  })
+}

backend/src/services/uptime-speed-service.ts

@@ -1,4 +1,4 @@
-import { asc, eq } from "drizzle-orm"
+import { and, asc, eq } from "drizzle-orm"
 import { db } from "../db/index.js"
 import { servers, uptimeSpeedProbes, uptimeSpeedTestRuns } from "../db/schema.js"
 import type { SpeedRunSnapshot, SpeedScheduledRunSnapshot } from "../types/scheduler-run-snapshot.js"
@@ -8,8 +8,14 @@ import { withBtestNodeLocks } from "./btest-node-locks.js"
 import { MikrotikClient } from "./mikrotik.js"
 import { parsePing } from "./uptime-collector.js"
 import { assertBandwidthTestCompleted } from "./bandwidth-test-status.js"
+import { assertSpeedProbeEndpoints, persistSpeedRun, type SpeedProbeEndpoints } from "./speed-run-persistence.js"
 import { readBandwidthTestRates } from "./bandwidth-test-rates.js"
 
+type SpeedWriteTransaction = Parameters<Parameters<typeof db.transaction>[0]>[0]
+function speedProbeEndpointsWhere(target: SpeedProbeEndpoints) {
+  return and(eq(uptimeSpeedProbes.id, target.id), eq(uptimeSpeedProbes.srcServerId, target.srcServerId), eq(uptimeSpeedProbes.dstServerId, target.dstServerId))
+}
+
 function firstUsableIpOnInterface(rows: RosIpAddress[], iface: string): string {
   const name = iface.trim()
   const candidates = rows.filter((a) => {
@@ -72,11 +78,16 @@ function humanizeSpeedError(raw: string): string {
 
 /** Полный цикл bandwidth-test + after-ping + запись в БД. Вызывать внутри withBtestNodeLocks при необходимости — здесь lock уже снаружи. */
 export async function executeBandwidthSpeedTestInner(params: SpeedTestRunParams): Promise<SpeedTestRunResult> {
+  const probeId = String(params.probeId ?? "").trim()
   const srcId = params.srcServerId
   const dstId = params.dstServerId
   const src = (await db.select().from(servers).where(eq(servers.id, srcId)).limit(1))[0]
   const dst = (await db.select().from(servers).where(eq(servers.id, dstId)).limit(1))[0]
   if (!src || !dst) throw new Error("Server not found")
+  if (probeId) {
+    const probe = (await db.select().from(uptimeSpeedProbes).where(eq(uptimeSpeedProbes.id, probeId)).limit(1))[0]
+    assertSpeedProbeEndpoints(probe, { id: probeId, srcServerId: srcId, dstServerId: dstId })
+  }
 
   const protocol = params.protocol === "udp" ? "udp" : "tcp"
   const direction = params.direction === "transmit" || params.direction === "receive" ? params.direction : "both"
@@ -126,44 +137,49 @@ export async function executeBandwidthSpeedTestInner(params: SpeedTestRunParams)
     }
   }
 
-  const probeId = String(params.probeId ?? "").trim()
   const runId = String(params.runId ?? "").trim() || `sr-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`
-  if (probeId) {
-    await db.update(uptimeSpeedProbes).set({
-      lastRunAt: pingNow,
-      lastTxAvgMbps: txFinal,
-      lastRxAvgMbps: rxFinal,
-      lastStatus: "done",
-      lastError: "",
-      lastPingRttMs: afterBtPing.rttMs,
-      lastPingLossPct: afterBtPing.lossPct,
-      lastPingAt: pingNow,
-      lastPingError: afterBtPing.error ?? "",
-      updatedAt: pingNow,
-    }).where(eq(uptimeSpeedProbes.id, probeId))
-  }
-  await db.insert(uptimeSpeedTestRuns).values({
-    id: runId,
-    probeId: probeId || null,
-    srcServerId: srcId,
-    dstServerId: dstId,
-    srcInterface: srcInterface || "",
-    dstInterface: dstInterface || "",
-    srcAddress: srcAddress || src.host,
-    dstAddress: initialAddress,
-    srcInterfaceAddress,
-    dstInterfaceAddress,
-    protocol,
-    direction,
-    durationSec,
-    txAvgMbps: txFinal,
-    rxAvgMbps: rxFinal,
-    pingRttMs: afterBtPing.rttMs,
-    pingLossPct: afterBtPing.lossPct,
-    pingError: afterBtPing.error ?? null,
-    status: "done",
-    error: null,
-    createdAt: pingNow,
+  await persistSpeedRun<SpeedWriteTransaction>(probeId ? { id: probeId, srcServerId: srcId, dstServerId: dstId } : null, {
+    transaction: work => db.transaction(work),
+    updateProbe: async (tx, target) => {
+      const rows = await tx.update(uptimeSpeedProbes).set({
+        lastRunAt: pingNow,
+        lastTxAvgMbps: txFinal,
+        lastRxAvgMbps: rxFinal,
+        lastStatus: "done",
+        lastError: "",
+        lastPingRttMs: afterBtPing.rttMs,
+        lastPingLossPct: afterBtPing.lossPct,
+        lastPingAt: pingNow,
+        lastPingError: afterBtPing.error ?? "",
+        updatedAt: pingNow,
+      }).where(speedProbeEndpointsWhere(target)).returning({ id: uptimeSpeedProbes.id })
+      return rows.length === 1
+    },
+    insertRun: async (tx, associatedProbeId) => {
+      await tx.insert(uptimeSpeedTestRuns).values({
+        id: runId,
+        probeId: associatedProbeId,
+        srcServerId: srcId,
+        dstServerId: dstId,
+        srcInterface: srcInterface || "",
+        dstInterface: dstInterface || "",
+        srcAddress: srcAddress || src.host,
+        dstAddress: initialAddress,
+        srcInterfaceAddress,
+        dstInterfaceAddress,
+        protocol,
+        direction,
+        durationSec,
+        txAvgMbps: txFinal,
+        rxAvgMbps: rxFinal,
+        pingRttMs: afterBtPing.rttMs,
+        pingLossPct: afterBtPing.lossPct,
+        pingError: afterBtPing.error ?? null,
+        status: "done",
+        error: null,
+        createdAt: pingNow,
+      })
+    },
   })
 
   return {
@@ -208,36 +224,42 @@ export async function recordSpeedTestFailure(params: {
 }) {
   const probeId = String(params.probeId ?? "").trim()
   const message = humanizeSpeedError(params.message)
-  if (probeId) {
-    await db.update(uptimeSpeedProbes).set({
-      lastRunAt: new Date().toISOString(),
-      lastStatus: "error",
-      lastError: message,
-      updatedAt: new Date().toISOString(),
-    }).where(eq(uptimeSpeedProbes.id, probeId))
-  }
-  await db.insert(uptimeSpeedTestRuns).values({
-    id: String(params.runId ?? "").trim() || `sr-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
-    probeId: probeId || null,
-    srcServerId: params.srcServerId,
-    dstServerId: params.dstServerId,
-    srcInterface: String(params.srcInterface ?? "").trim(),
-    dstInterface: String(params.dstInterface ?? "").trim(),
-    srcAddress: null,
-    dstAddress: null,
-    srcInterfaceAddress: null,
-    dstInterfaceAddress: null,
-    protocol: params.protocol === "udp" ? "udp" : "tcp",
-    direction: params.direction === "transmit" || params.direction === "receive" ? params.direction : "both",
-    durationSec: Math.max(3, Number.parseInt(String(params.durationSec ?? ""), 10) || 10),
-    txAvgMbps: null,
-    rxAvgMbps: null,
-    pingRttMs: null,
-    pingLossPct: null,
-    pingError: null,
-    status: "error",
-    error: message,
-    createdAt: new Date().toISOString(),
+  await persistSpeedRun<SpeedWriteTransaction>(probeId ? { id: probeId, srcServerId: params.srcServerId, dstServerId: params.dstServerId } : null, {
+    transaction: work => db.transaction(work),
+    updateProbe: async (tx, target) => {
+      const rows = await tx.update(uptimeSpeedProbes).set({
+        lastRunAt: new Date().toISOString(),
+        lastStatus: "error",
+        lastError: message,
+        updatedAt: new Date().toISOString(),
+      }).where(speedProbeEndpointsWhere(target)).returning({ id: uptimeSpeedProbes.id })
+      return rows.length === 1
+    },
+    insertRun: async (tx, associatedProbeId) => {
+      await tx.insert(uptimeSpeedTestRuns).values({
+        id: String(params.runId ?? "").trim() || `sr-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
+        probeId: associatedProbeId,
+        srcServerId: params.srcServerId,
+        dstServerId: params.dstServerId,
+        srcInterface: String(params.srcInterface ?? "").trim(),
+        dstInterface: String(params.dstInterface ?? "").trim(),
+        srcAddress: null,
+        dstAddress: null,
+        srcInterfaceAddress: null,
+        dstInterfaceAddress: null,
+        protocol: params.protocol === "udp" ? "udp" : "tcp",
+        direction: params.direction === "transmit" || params.direction === "receive" ? params.direction : "both",
+        durationSec: Math.max(3, Number.parseInt(String(params.durationSec ?? ""), 10) || 10),
+        txAvgMbps: null,
+        rxAvgMbps: null,
+        pingRttMs: null,
+        pingLossPct: null,
+        pingError: null,
+        status: "error",
+        error: message,
+        createdAt: new Date().toISOString(),
+      })
+    },
   })
 }