fix(api): сохранить неизвестный исход записи speed без ложной ошибки
This commit is contained in:
1 parent
c337fddac9
commit
f408afba20
12 files changed
+220
-51
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, SpeedProbeContextChanged } from "../services/speed-run-persistence.js"
|
||||
import { SpeedProbeEndpointsChanged, SpeedProbeContextChanged, SpeedRunPersistenceUnknown } 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) => ({ createdAt: "2026-10-07T00:00:00Z", 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 } })
|
||||
@@ -28,7 +28,13 @@ missing = true; assert.equal((await send(base)).statusCode, 404); assert.equal(c
|
||||
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")
|
||||
const beforeUnknown = { runs: commands.length, failures: failures.length }
|
||||
problem = new SpeedRunPersistenceUnknown("actual-run-id")
|
||||
const unknown = await send(base)
|
||||
assert.equal(unknown.statusCode, 502); assert.equal(unknown.json().outcome, "unknown")
|
||||
assert.equal(unknown.json().runId, "actual-run-id"); assert.match(unknown.json().error, /сохранение результата не подтверждено/)
|
||||
assert.equal(failures.length, beforeUnknown.failures); assert.equal(commands.length, beforeUnknown.runs + 1)
|
||||
await app.close(); console.log("Speed-test HTTP:26 invalid inputs; explicit persistence unknown/runId skips failure write and command retry")
|
||||
process.env.AUTH_REQUIRED = "true"; process.env.AUTH_JWT_SECRET = "speed-command-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)
|
||||
|
||||
@@ -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 { SpeedProbeContextChanged } from "../services/speed-run-persistence.js"
|
||||
import { SpeedProbeContextChanged, SpeedRunPersistenceUnknown } from "../services/speed-run-persistence.js"
|
||||
|
||||
export function createSpeedTestRoute(deps: {
|
||||
serverExists: (id: number) => Promise<boolean>
|
||||
@@ -18,6 +18,9 @@ export function createSpeedTestRoute(deps: {
|
||||
try {
|
||||
return reply.send({ ok: true, result: await deps.run(command) })
|
||||
} catch (error) {
|
||||
if (error instanceof SpeedRunPersistenceUnknown) {
|
||||
return reply.code(502).send({ error: error.message, outcome: "unknown", runId: error.runId })
|
||||
}
|
||||
const raw = error instanceof Error ? error.message : "Bandwidth test failed"
|
||||
await deps.failure({ ...command, message: raw })
|
||||
const message = /timed out|socket hang up|ECONNRESET/i.test(raw)
|
||||
|
||||
@@ -69,3 +69,16 @@ test("post-collection hook failure is reported separately from the recorded outc
|
||||
const receipt = await runRecordedSchedulerJob(h.deps)
|
||||
assert.equal(receipt.status, "ok"); assert.equal(h.records.length, 1); assert.equal(h.reported.length, 1)
|
||||
})
|
||||
|
||||
test("unconfirmed speed persistence is an execution limitation, not a successful job", async () => {
|
||||
const snapshot: SchedulerRunSnapshot = { v: 1, job: "uptime_speed", sampledAt: "now", runs: [{
|
||||
probeId: "probe", srcServerId: 1, srcServerName: "source", dstServerId: 2, dstServerName: "destination",
|
||||
srcInterface: "", dstInterface: "", protocol: "tcp", direction: "transmit", durationSec: 3,
|
||||
ok: false, error: "Сохранение результата не подтверждено; runId actual-run",
|
||||
}] }
|
||||
const h = harness(snapshot), receipt = await runRecordedSchedulerJob(h.deps)
|
||||
assert.equal(receipt.status, "error"); assert.match(receipt.error!, /runId actual-run/)
|
||||
assert.equal(h.events.some(event => event.eventType === "scheduler.job.ok"), false)
|
||||
assert.equal(h.records.length, 1)
|
||||
assert.equal(h.counts().collections, 1)
|
||||
})
|
||||
@@ -20,6 +20,9 @@ export function schedulerSnapshotError(snapshot: SchedulerRunSnapshot | null): s
|
||||
if (!snapshot) return null
|
||||
if ("fatalError" in snapshot && snapshot.fatalError) return snapshot.fatalError
|
||||
const errors = "errors" in snapshot ? [...(snapshot.errors ?? [])] : []
|
||||
if (snapshot.job === "uptime_speed") {
|
||||
for (const run of snapshot.runs) if (!run.ok && run.error) errors.push(run.error)
|
||||
}
|
||||
if (snapshot.job === "backups") {
|
||||
errors.push(...(snapshot.eventErrors ?? []))
|
||||
if (snapshot.metadataError && !errors.some(error => error.includes(snapshot.metadataError!))) errors.push("Метаданные: " + snapshot.metadataError)
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
export interface SpeedProbeEndpoints { id: string; srcServerId: number; dstServerId: number }
|
||||
export class SpeedRunPersistenceUnknown extends Error {
|
||||
constructor(public readonly runId: string, cause?: unknown) {
|
||||
super(`Сетевой тест завершён, сохранение результата не подтверждено. Проверьте историю по runId ${runId} перед повтором`, { cause })
|
||||
}
|
||||
}
|
||||
|
||||
export interface SpeedProbeContext extends SpeedProbeEndpoints {
|
||||
srcInterface: string; dstInterface: string; protocol: "tcp" | "udp"; direction: "both" | "transmit" | "receive"; durationSec: number
|
||||
|
||||
@@ -8,7 +8,7 @@ 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 { assertSpeedProbeContext, persistSpeedRun, type SpeedProbeContext } from "./speed-run-persistence.js"
|
||||
import { assertSpeedProbeContext, persistSpeedRun, SpeedRunPersistenceUnknown, type SpeedProbeContext } from "./speed-run-persistence.js"
|
||||
import { readBandwidthTestRates } from "./bandwidth-test-rates.js"
|
||||
|
||||
type SpeedWriteTransaction = Parameters<Parameters<typeof db.transaction>[0]>[0]
|
||||
@@ -143,49 +143,52 @@ 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, SpeedProbeContext>(probeId ? target : 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(speedProbeContextWhere(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,
|
||||
})
|
||||
},
|
||||
})
|
||||
try {
|
||||
await persistSpeedRun<SpeedWriteTransaction, SpeedProbeContext>(probeId ? target : 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(speedProbeContextWhere(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,
|
||||
})
|
||||
},
|
||||
})
|
||||
|
||||
} catch (error) { throw new SpeedRunPersistenceUnknown(runId, error) }
|
||||
|
||||
return {
|
||||
createdAt: pingNow,
|
||||
@@ -325,7 +328,7 @@ export async function runScheduledSpeedProbesOnce(): Promise<SpeedScheduledRunSn
|
||||
})
|
||||
} catch (e) {
|
||||
const msg = e instanceof Error ? e.message : String(e)
|
||||
await recordSpeedTestFailure({
|
||||
if (!(e instanceof SpeedRunPersistenceUnknown)) await recordSpeedTestFailure({
|
||||
probeId: p.id,
|
||||
srcServerId: p.srcServerId,
|
||||
dstServerId: p.dstServerId,
|
||||
|
||||
Reference in new issue
Block a user