fix(uptime): сбросить снимок при смене пары и сохранить историю
This commit is contained in:
1 parent
8627459bfa
commit
e4f951175c
12 files changed
+142
-20
No files matched your search
@@ -11,8 +11,8 @@ let rows = new Map<string, SpeedProbePatch & { lastTxAvgMbps: number | null }>([
|
||||
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 })
|
||||
await run({ listProbes: async () => [...draft.values()], remove: async id => { draft.delete(id) }, upsert: async (patch, _createdAt, resetObservations) => {
|
||||
draft.set(patch.id, { ...patch, lastTxAvgMbps: resetObservations ? null : draft.get(patch.id)?.lastTxAvgMbps ?? null })
|
||||
if (failWrite) throw new Error("database write failed")
|
||||
} })
|
||||
rows = draft
|
||||
@@ -37,6 +37,13 @@ assert.equal(rows.has("old"), false); assert.equal(rows.get("s1")?.enabled, fals
|
||||
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)
|
||||
const beforePairChange = rows.get("s1")!.srcServerId
|
||||
await put({ probes: [{ ...probe, srcServerId: 2, dstServerId: 1 }] })
|
||||
assert.notEqual(rows.get("s1")?.srcServerId, beforePairChange)
|
||||
assert.equal(rows.get("s1")?.lastTxAvgMbps, null)
|
||||
rows.get("s1")!.lastTxAvgMbps = 0
|
||||
await put({ probes: [{ ...probe, srcServerId: 2, dstServerId: 1, enabled: true }] })
|
||||
assert.equal(rows.get("s1")?.lastTxAvgMbps, 0)
|
||||
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)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { emptySpeedProbeObservations } from "@mmapp/contracts"
|
||||
import { SpeedProbeEndpointsChanged } from "../services/speed-run-persistence.js"
|
||||
import { createPingProbesWriteRoute } from "./ping-probes-write.js"
|
||||
import { createSpeedProbesWriteRoute } from "./speed-probes-write.js"
|
||||
@@ -207,12 +208,12 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
|
||||
// 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),
|
||||
listProbes: async () => tx.select({ id: uptimeSpeedProbes.id, srcServerId: uptimeSpeedProbes.srcServerId, dstServerId: uptimeSpeedProbes.dstServerId }).from(uptimeSpeedProbes).for("update"),
|
||||
remove: async id => { await tx.delete(uptimeSpeedProbes).where(eq(uptimeSpeedProbes.id, id)) },
|
||||
upsert: async (row, createdAt) => {
|
||||
upsert: async (row, createdAt, resetObservations) => {
|
||||
const { id, ...patch } = row
|
||||
await tx.insert(uptimeSpeedProbes).values({ id, ...patch, createdAt })
|
||||
.onConflictDoUpdate({ target: uptimeSpeedProbes.id, set: patch })
|
||||
.onConflictDoUpdate({ target: uptimeSpeedProbes.id, set: { ...patch, ...(resetObservations ? emptySpeedProbeObservations : {}) } })
|
||||
},
|
||||
})
|
||||
}) }, probes),
|
||||
|
||||
@@ -1,16 +1,21 @@
|
||||
import { matchesSpeedProbeEndpoints, type SpeedProbeEndpoints } from "./speed-run-persistence.js"
|
||||
import type { SpeedProbeWrite } from "@mmapp/contracts"
|
||||
export type SpeedProbePatch = SpeedProbeWrite & { sortOrder: number; updatedAt: string }
|
||||
export type SpeedProbeTransaction = {
|
||||
listIds: () => Promise<string[]>
|
||||
listProbes: () => Promise<SpeedProbeEndpoints[]>
|
||||
remove: (id: string) => Promise<void>
|
||||
upsert: (row: SpeedProbePatch, createdAt: string) => Promise<void>
|
||||
upsert: (row: SpeedProbePatch, createdAt: string, resetObservations: boolean) => 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. */
|
||||
/** The adapter must lock the catalog and commit all configuration changes atomically. Observations are preserved for the same pair and reset when its endpoints change. */
|
||||
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)
|
||||
const previous = new Map((await tx.listProbes()).map(row => [row.id, row]))
|
||||
for (const id of previous.keys()) if (!nextIds.has(id)) await tx.remove(id)
|
||||
for (const [sortOrder, row] of probes.entries()) {
|
||||
const reset = !matchesSpeedProbeEndpoints(previous.get(row.id), row)
|
||||
await tx.upsert({ ...row, sortOrder, updatedAt: now }, now, reset)
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in new issue
Block a user