fix(api): сохранять результат только в пробу с теми же параметрами
This commit is contained in:
1 parent
b59944b573
commit
090206c893
9 files changed
+97
-23
No files matched your search
@@ -3,7 +3,7 @@ import Fastify from "fastify"
|
||||
import { SignJWT } from "jose"
|
||||
import { speedTestCommandSchema, type SpeedTestCommand } from "@mmapp/contracts"
|
||||
import { createSpeedTestRoute } from "./speed-test.js"
|
||||
import { SpeedProbeEndpointsChanged } from "../services/speed-run-persistence.js"
|
||||
import { SpeedProbeEndpointsChanged, SpeedProbeContextChanged } from "../services/speed-run-persistence.js"
|
||||
const commands: SpeedTestCommand[] = [], failures: Array<SpeedTestCommand & { message: string }> = []
|
||||
let lookups = 0, missing = false, problem: Error | null = null
|
||||
const result = (c: SpeedTestCommand) => ({ runId: c.runId ?? "ad-hoc", srcInterface: c.srcInterface || null, dstInterface: c.dstInterface || null, address: "192.0.2.2", srcAddress: "192.0.2.1", dstAddress: "192.0.2.2", srcInterfaceAddress: null, dstInterfaceAddress: null, protocol: c.protocol, direction: c.direction, durationSec: c.durationSec, txAvgMbps: null, rxAvgMbps: 0, txSeries: [], rxSeries: [0], raw: [{ status: "done testing" }], afterBtPing: { rttMs: null, lossPct: null, error: null } })
|
||||
@@ -25,7 +25,7 @@ const explicit = await send({ srcServerId: "1", dstServerId: "2", runId: " run-2
|
||||
assert.equal(explicit.statusCode, 200); assert.equal(commands[1].runId, "run-2"); assert.equal(commands[1].probeId, "probe"); assert.equal(commands[1].srcInterface, "ether1"); assert.equal(commands[1].protocol, "udp"); assert.equal(commands[1].direction, "receive"); assert.equal(commands[1].durationSec, 3); assert.equal("lastTxAvgMbps" in commands[1], false)
|
||||
assert.equal(speedTestCommandSchema.parse({ ...base, runId: " ", probeId: "" }).probeId, undefined)
|
||||
missing = true; assert.equal((await send(base)).statusCode, 404); assert.equal(commands.length, 2); assert.equal(failures.length, 0); missing = false
|
||||
for (const [error, code] of [[new SpeedProbeEndpointsChanged(), 409], [Error("На интерфейсе назначения нет активного IP"), 400], [Error("Server not found"), 400], [Error("authentication failed"), 502], [Error("socket hang up"), 502]] as const) {
|
||||
for (const [error, code] of [[new SpeedProbeEndpointsChanged(), 409], [new SpeedProbeContextChanged(), 409], [Error("На интерфейсе назначения нет активного IP"), 400], [Error("Server not found"), 400], [Error("authentication failed"), 502], [Error("socket hang up"), 502]] as const) {
|
||||
problem = error; const response = await send({ ...base, runId: "failed", direction: "transmit", durationSec: 20 }); assert.equal(response.statusCode, code); assert.equal(failures.at(-1)?.runId, "failed"); assert.equal(failures.at(-1)?.durationSec, 20); assert.equal(failures.at(-1)?.direction, "transmit")
|
||||
}
|
||||
await app.close(); console.log("Speed-test HTTP:26 invalid inputs before lookups/native/history; defaults/ad-hoc/trim/nullable/zero/404/409/400/502")
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
|
||||
import { speedTestCommandSchema, type SpeedTestCommand } from "@mmapp/contracts"
|
||||
import type { SpeedTestRunResult } from "../services/uptime-speed-service.js"
|
||||
import { SpeedProbeEndpointsChanged } from "../services/speed-run-persistence.js"
|
||||
import { SpeedProbeContextChanged } from "../services/speed-run-persistence.js"
|
||||
|
||||
export function createSpeedTestRoute(deps: {
|
||||
serverExists: (id: number) => Promise<boolean>
|
||||
@@ -23,7 +23,7 @@ export function createSpeedTestRoute(deps: {
|
||||
const message = /timed out|socket hang up|ECONNRESET/i.test(raw)
|
||||
? "BTTest timeout/socket hang up: проверь доступность destination, /tool bandwidth-server, firewall и корректность интерфейсов"
|
||||
: raw
|
||||
if (error instanceof SpeedProbeEndpointsChanged) return reply.code(409).send({ error: message })
|
||||
if (error instanceof SpeedProbeContextChanged) return reply.code(409).send({ error: message })
|
||||
if (/активного IP|Server not found/i.test(raw)) return reply.code(400).send({ error: message })
|
||||
return reply.code(502).send({ error: message })
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { test } from "node:test"
|
||||
import assert from "node:assert/strict"
|
||||
import { assertSpeedProbeEndpoints, matchesSpeedProbeEndpoints, persistSpeedRun, SpeedProbeEndpointsChanged } from "./speed-run-persistence.js"
|
||||
import { assertSpeedProbeEndpoints, matchesSpeedProbeEndpoints, persistSpeedRun, SpeedProbeEndpointsChanged, assertSpeedProbeContext, matchesSpeedProbeContext, SpeedProbeContextChanged, type SpeedProbeContext } 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") {
|
||||
@@ -64,3 +64,36 @@ test("ad-hoc run writes history without touching any probe", async () => {
|
||||
assert.equal(h.updates(), 0)
|
||||
assert.deepEqual(h.state().runs, [{ probeId: null, average: 0 }])
|
||||
})
|
||||
|
||||
const context: SpeedProbeContext = { ...target, srcInterface: "ether1", dstInterface: "ether2", protocol: "tcp", direction: "both", durationSec: 10 }
|
||||
const edits: Partial<SpeedProbeContext>[] = [{ srcInterface: "ether3" }, { dstInterface: "ether3" }, { protocol: "udp" }, { direction: "receive" }, { durationSec: 20 }]
|
||||
test("preflight validates full context and ignores enable/order/telemetry changes", () => {
|
||||
assert.doesNotThrow(() => assertSpeedProbeContext({ ...context, enabled: false, sortOrder: 9, lastTxAvgMbps: 0 }, context))
|
||||
for (const edit of edits) assert.throws(() => assertSpeedProbeContext({ ...context, ...edit }, context), SpeedProbeContextChanged)
|
||||
assert.throws(() => assertSpeedProbeContext(null, context), SpeedProbeEndpointsChanged)
|
||||
})
|
||||
for (const edit of edits) test(`changed ${Object.keys(edit)[0]} cannot publish success or failure into new context`, async () => {
|
||||
for (const status of ["done", "error"]) {
|
||||
const probe = { ...context, ...edit, average: 7, status: "done" }
|
||||
const runs: Array<{ probeId: string | null; status: string }> = []
|
||||
await persistSpeedRun(context, {
|
||||
transaction: async work => work(probe),
|
||||
updateProbe: async (_tx, expected) => {
|
||||
if (!matchesSpeedProbeContext(probe, expected)) return false
|
||||
probe.average = 0; probe.status = status; return true
|
||||
},
|
||||
insertRun: async (_tx, probeId) => { runs.push({ probeId, status }) },
|
||||
})
|
||||
assert.equal(probe.average, 7); assert.equal(probe.status, "done")
|
||||
assert.deepEqual(runs, [{ probeId: null, status }])
|
||||
}
|
||||
})
|
||||
test("matching full context publishes explicit zero with associated history", async () => {
|
||||
const probe = { ...context, average: 7 }; let associated: string | null = null
|
||||
await persistSpeedRun(context, {
|
||||
transaction: async work => work(probe),
|
||||
updateProbe: async (_tx, expected) => { assert.equal(matchesSpeedProbeContext(probe, expected), true); probe.average = 0; return true },
|
||||
insertRun: async (_tx, probeId) => { associated = probeId },
|
||||
})
|
||||
assert.equal(probe.average, 0); assert.equal(associated, context.id)
|
||||
})
|
||||
@@ -1,6 +1,12 @@
|
||||
export interface SpeedProbeEndpoints { id: string; srcServerId: number; dstServerId: number }
|
||||
|
||||
export class SpeedProbeEndpointsChanged extends Error {
|
||||
export interface SpeedProbeContext extends SpeedProbeEndpoints {
|
||||
srcInterface: string; dstInterface: string; protocol: "tcp" | "udp"; direction: "both" | "transmit" | "receive"; durationSec: number
|
||||
}
|
||||
export class SpeedProbeContextChanged extends Error {
|
||||
constructor(message = "Параметры speed-пробы изменились. Обновите список проб перед запуском") { super(message) }
|
||||
}
|
||||
export class SpeedProbeEndpointsChanged extends SpeedProbeContextChanged {
|
||||
constructor() { super("Пара устройств speed-пробы изменилась. Обновите список проб перед запуском") }
|
||||
}
|
||||
export function matchesSpeedProbeEndpoints(row: unknown, target: SpeedProbeEndpoints): boolean {
|
||||
@@ -12,10 +18,21 @@ export function assertSpeedProbeEndpoints(row: unknown, target: SpeedProbeEndpoi
|
||||
if (!matchesSpeedProbeEndpoints(row, target)) throw new SpeedProbeEndpointsChanged()
|
||||
}
|
||||
|
||||
export function matchesSpeedProbeContext(row: unknown, target: SpeedProbeContext): boolean {
|
||||
if (!matchesSpeedProbeEndpoints(row, target)) return false
|
||||
const probe = row as Partial<SpeedProbeContext>
|
||||
return probe.srcInterface === target.srcInterface && probe.dstInterface === target.dstInterface
|
||||
&& probe.protocol === target.protocol && probe.direction === target.direction && probe.durationSec === target.durationSec
|
||||
}
|
||||
export function assertSpeedProbeContext(row: unknown, target: SpeedProbeContext): void {
|
||||
assertSpeedProbeEndpoints(row, target)
|
||||
if (!matchesSpeedProbeContext(row, target)) throw new SpeedProbeContextChanged()
|
||||
}
|
||||
|
||||
/** Changed/deleted probes retain their snapshot; history keeps the actual run without a stale association. */
|
||||
export async function persistSpeedRun<C>(target: SpeedProbeEndpoints | null, deps: {
|
||||
export async function persistSpeedRun<C, T extends SpeedProbeEndpoints = SpeedProbeEndpoints>(target: T | null, deps: {
|
||||
transaction: (work: (context: C) => Promise<void>) => Promise<void>
|
||||
updateProbe: (context: C, target: SpeedProbeEndpoints) => Promise<boolean>
|
||||
updateProbe: (context: C, target: T) => Promise<boolean>
|
||||
insertRun: (context: C, associatedProbeId: string | null) => Promise<void>
|
||||
}): Promise<void> {
|
||||
await deps.transaction(async context => {
|
||||
|
||||
@@ -8,12 +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 { assertSpeedProbeContext, persistSpeedRun, type SpeedProbeContext } 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 speedProbeContextWhere(target: SpeedProbeContext) {
|
||||
return and(eq(uptimeSpeedProbes.id, target.id), eq(uptimeSpeedProbes.srcServerId, target.srcServerId), eq(uptimeSpeedProbes.dstServerId, target.dstServerId),
|
||||
eq(uptimeSpeedProbes.srcInterface, target.srcInterface), eq(uptimeSpeedProbes.dstInterface, target.dstInterface),
|
||||
eq(uptimeSpeedProbes.protocol, target.protocol), eq(uptimeSpeedProbes.direction, target.direction), eq(uptimeSpeedProbes.durationSec, target.durationSec))
|
||||
}
|
||||
|
||||
function firstUsableIpOnInterface(rows: RosIpAddress[], iface: string): string {
|
||||
@@ -84,16 +86,17 @@ export async function executeBandwidthSpeedTestInner(params: SpeedTestRunParams)
|
||||
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"
|
||||
const durationSec = Math.max(3, Number.parseInt(String(params.durationSec ?? ""), 10) || 10)
|
||||
const srcInterface = String(params.srcInterface ?? "").trim()
|
||||
const dstInterface = String(params.dstInterface ?? "").trim()
|
||||
const target: SpeedProbeContext = { id: probeId, srcServerId: srcId, dstServerId: dstId, srcInterface, dstInterface, protocol, direction, durationSec }
|
||||
if (probeId) {
|
||||
const probe = (await db.select().from(uptimeSpeedProbes).where(eq(uptimeSpeedProbes.id, probeId)).limit(1))[0]
|
||||
assertSpeedProbeContext(probe, target)
|
||||
}
|
||||
const srcClient = MikrotikClient.fromServer(src)
|
||||
const dstAddressRows = dstInterface
|
||||
? await MikrotikClient.fromServer(dst).getIpAddresses().catch(() => [])
|
||||
@@ -138,7 +141,7 @@ export async function executeBandwidthSpeedTestInner(params: SpeedTestRunParams)
|
||||
}
|
||||
|
||||
const runId = String(params.runId ?? "").trim() || `sr-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`
|
||||
await persistSpeedRun<SpeedWriteTransaction>(probeId ? { id: probeId, srcServerId: srcId, dstServerId: dstId } : null, {
|
||||
await persistSpeedRun<SpeedWriteTransaction, SpeedProbeContext>(probeId ? target : null, {
|
||||
transaction: work => db.transaction(work),
|
||||
updateProbe: async (tx, target) => {
|
||||
const rows = await tx.update(uptimeSpeedProbes).set({
|
||||
@@ -152,7 +155,7 @@ export async function executeBandwidthSpeedTestInner(params: SpeedTestRunParams)
|
||||
lastPingAt: pingNow,
|
||||
lastPingError: afterBtPing.error ?? "",
|
||||
updatedAt: pingNow,
|
||||
}).where(speedProbeEndpointsWhere(target)).returning({ id: uptimeSpeedProbes.id })
|
||||
}).where(speedProbeContextWhere(target)).returning({ id: uptimeSpeedProbes.id })
|
||||
return rows.length === 1
|
||||
},
|
||||
insertRun: async (tx, associatedProbeId) => {
|
||||
@@ -224,7 +227,13 @@ export async function recordSpeedTestFailure(params: {
|
||||
}) {
|
||||
const probeId = String(params.probeId ?? "").trim()
|
||||
const message = humanizeSpeedError(params.message)
|
||||
await persistSpeedRun<SpeedWriteTransaction>(probeId ? { id: probeId, srcServerId: params.srcServerId, dstServerId: params.dstServerId } : null, {
|
||||
await persistSpeedRun<SpeedWriteTransaction, SpeedProbeContext>(probeId ? {
|
||||
id: probeId, srcServerId: params.srcServerId, dstServerId: params.dstServerId,
|
||||
srcInterface: String(params.srcInterface ?? "").trim(), dstInterface: String(params.dstInterface ?? "").trim(),
|
||||
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),
|
||||
} : null, {
|
||||
transaction: work => db.transaction(work),
|
||||
updateProbe: async (tx, target) => {
|
||||
const rows = await tx.update(uptimeSpeedProbes).set({
|
||||
@@ -232,7 +241,7 @@ export async function recordSpeedTestFailure(params: {
|
||||
lastStatus: "error",
|
||||
lastError: message,
|
||||
updatedAt: new Date().toISOString(),
|
||||
}).where(speedProbeEndpointsWhere(target)).returning({ id: uptimeSpeedProbes.id })
|
||||
}).where(speedProbeContextWhere(target)).returning({ id: uptimeSpeedProbes.id })
|
||||
return rows.length === 1
|
||||
},
|
||||
insertRun: async (tx, associatedProbeId) => {
|
||||
|
||||
Reference in new issue
Block a user