fix(api): сбросить кэши после восстановления базы

This commit is contained in:
Denozordec committed 2026-10-07 23:12:32 +07:00
1 parent 2943032462
commit f1aac49dfc
18 files changed
+150 -17

No files matched your search

+1 -1
View File
@@ -37,7 +37,7 @@
"test:ospf-optimizer": "tsx src/services/ospf-optimizer-identity.test.ts && tsx src/services/ospf-optimizer-metrics.test.ts && tsx src/services/ospf-optimizer-plan.test.ts && tsx src/routes/ospf-optimizer.test.ts",
"test:containers": "tsx src/services/containers-state.test.ts && tsx src/routes/containers-state.test.ts",
"test:certificates": "tsx src/services/certificate-issue-execution.test.ts && tsx src/services/certificate-issue-request.test.ts && tsx src/routes/certificates-issue.test.ts && tsx src/services/certificate-issue-dto.test.ts && tsx src/services/certificate-parse.test.ts && tsx src/services/certificates-state.test.ts && tsx src/routes/certificates-state.test.ts",
"test:system-database": "tsx src/services/database-background-work.test.ts && tsx src/services/database-request-gate.test.ts && tsx src/services/traffic-flow-worker-shutdown.test.ts && tsx src/services/system-database-restore-execution.test.ts && tsx src/services/system-database-operation.test.ts && tsx src/routes/system-database.test.ts && tsx src/plugins/auth.system-database.test.ts"
"test:system-database": "tsx src/services/system-database-cache.test.ts && tsx src/services/database-background-work.test.ts && tsx src/services/database-request-gate.test.ts && tsx src/services/traffic-flow-worker-shutdown.test.ts && tsx src/services/system-database-restore-execution.test.ts && tsx src/services/system-database-operation.test.ts && tsx src/routes/system-database.test.ts && tsx src/plugins/auth.system-database.test.ts"
},
"dependencies": {
"@aws-sdk/client-s3": "^3.888.0",
@@ -10,6 +10,7 @@ import { withOperationLock } from "../db/operation-lock.js"
import { executeDatabaseOperation, type DatabaseOperationWarnings } from "./system-database-operation.js"
import { restartSchedulerAfterDatabaseOperation, stopSchedulerForDatabaseOperation } from "./scheduler.js"
import { pauseRipeBackgroundWork, resumeRipeBackgroundWork } from "./traffic-flow-ripe.js"
import { invalidateSystemDatabaseCaches } from "./system-database-cache.js"
import {
restartTrafficFlowListenerAfterDatabaseOperation,
stopTrafficFlowListenerForDatabaseOperation,
@@ -83,5 +84,6 @@ export async function restoreSystemDatabaseBackup(buffer: Buffer): Promise<Syste
write: (file, data) => writeFile(file, data),
run: async args => { await spawnCollected("pg_restore", args) },
remove: file => rm(file, { force: true }),
invalidate: invalidateSystemDatabaseCaches,
}))
}
@@ -0,0 +1,55 @@
import assert from "node:assert/strict"
import { test } from "node:test"
import { invalidateSystemDatabaseCaches } from "./system-database-cache.js"
import { applyRingSnapshot, snapshotRings } from "./traffic-flow-engine.js"
import { rememberServerIfaces, ifaceCacheHas } from "./traffic-flow-ifindex.js"
import { DatabaseBackgroundWork } from "./database-background-work.js"
import { executeDatabaseOperation } from "./system-database-operation.js"
import { executeSystemDatabaseRestore } from "./system-database-restore-execution.js"
import {
disableRipePersistForTests, seedRipeCacheForTests, lookupRipeCached,
pauseRipeBackgroundWork, resumeRipeBackgroundWork, enqueueRipeMisses, ripeFetchCountForTests,
} from "./traffic-flow-ripe.js"
test("restore discards old RIPE, interface and live data without reopening RIPE admission", async () => {
disableRipePersistForTests()
seedRipeCacheForTests({ prefix: "8.8.8.0/24", asn: 15169, country: "US", lat: null, lng: null,
holder: "old database", ok: true, fetchedAt: Date.now() })
rememberServerIfaces(1, [{ ".id": "*1", name: "ether1" }])
applyRingSnapshot([{ key: "old database", inBps: [1_000_000], outBps: [1_000_000] }])
assert.equal(lookupRipeCached("8.8.8.8")?.holder, "old database")
assert.equal(ifaceCacheHas(1), true)
assert.equal(snapshotRings().length, 1)
await pauseRipeBackgroundWork()
try {
const count = ripeFetchCountForTests()
invalidateSystemDatabaseCaches()
assert.equal(lookupRipeCached("8.8.8.8"), null)
assert.equal(ifaceCacheHas(1), false)
assert.equal(snapshotRings().length, 0)
enqueueRipeMisses(["8.8.8.8"])
assert.equal(ripeFetchCountForTests(), count)
} finally { resumeRipeBackgroundWork() }
})
test("background drain precedes restore and consumers restart with invalidated caches", async () => {
const work = new DatabaseBackgroundWork(), calls: string[] = []
let release!: () => void
work.track(new Promise<void>(r => { release = r }))
rememberServerIfaces(1, [{ ".id": "*1", name: "old-interface" }])
const operation = executeDatabaseOperation(() => executeSystemDatabaseRestore(Buffer.from("PGDMPfixture"), "fixture", {
allocate: async () => "fixture.dump", write: async () => {}, remove: async () => {},
run: async () => { calls.push("restore") },
invalidate: () => { calls.push("invalidate"); invalidateSystemDatabaseCaches() },
}), {
stopListener: () => { calls.push("stop listener") },
stopScheduler: async () => { await work.drain(); calls.push("drained") },
startListener: () => { assert.equal(ifaceCacheHas(1), false); calls.push("restart listener") },
startScheduler: () => { calls.push("restart scheduler") },
})
await Promise.resolve(); await Promise.resolve()
assert.equal(calls.includes("restore"), false)
assert.equal(ifaceCacheHas(1), true)
release(); await operation
assert.deepEqual(calls, ["stop listener", "drained", "restore", "invalidate", "restart listener", "restart scheduler"])
})
@@ -0,0 +1,20 @@
import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js"
import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"
import { invalidateRipeCache } from "./traffic-flow-ripe.js"
import { invalidateIfaceCache } from "./traffic-flow-ifindex.js"
import { invalidateFlowClassificationCache } from "./traffic-flow-classify.js"
import { invalidateFlowAnalyticsCache } from "./traffic-flow-analytics.js"
import { invalidateFlowMapHopsCache } from "./traffic-flow-map-hops.js"
import { resetEngineAfterDatabaseRestore } from "./traffic-flow-engine.js"
/** Synchronous invalidation after confirmed restore and before consumers restart. */
export function invalidateSystemDatabaseCaches(): void {
invalidateTrafficFlowSettingsCache()
invalidateFlowCatalogCache()
invalidateRipeCache()
invalidateIfaceCache()
invalidateFlowClassificationCache()
invalidateFlowAnalyticsCache()
invalidateFlowMapHopsCache()
resetEngineAfterDatabaseRestore()
}
@@ -7,3 +7,12 @@ test("invalid archive never allocates or runs command",async()=>{const f=fixture
test("restore runs one transactional command and cleans file after success",async()=>{const f=fixture();assert.deepEqual(await executeSystemDatabaseRestore(dump,"postgres://fixture",f.deps),{});assert.deepEqual(f.calls,["allocate","write","run","remove"])})
test("write or restore failure always cleans temporary file and never retries",async()=>{for(const step of ["write","run"]){const f=fixture();if(step==="write")f.setWrite();else f.setRun();await assert.rejects(executeSystemDatabaseRestore(dump,"postgres://fixture",f.deps),/failed/);assert.deepEqual(f.calls,step==="write"?["allocate","write","remove"]:["allocate","write","run","remove"])}})
test("cleanup failure keeps confirmed restore result; main failure is preserved",async()=>{const f=fixture();f.setRemove();assert.deepEqual(await executeSystemDatabaseRestore(dump,"postgres://fixture",f.deps),{cleanupError:"Не удалось удалить временный файл восстановления"});const bad=fixture();bad.setRun();bad.setRemove();await assert.rejects(executeSystemDatabaseRestore(dump,"postgres://fixture",bad.deps),/restore failed; Не удалось удалить/)})
test("cache invalidation follows confirmed restore and is skipped on process failure", async () => {
const success=fixture()
await executeSystemDatabaseRestore(dump,"postgres://fixture",{...success.deps,invalidate:()=>{success.calls.push("invalidate")}})
assert.deepEqual(success.calls,["allocate","write","run","invalidate","remove"])
const failure=fixture();failure.setRun()
await assert.rejects(executeSystemDatabaseRestore(dump,"postgres://fixture",{...failure.deps,invalidate:()=>{failure.calls.push("invalidate")}}),/restore failed/)
assert.equal(failure.calls.includes("invalidate"),false)
})
@@ -4,6 +4,7 @@ interface Dependencies {
write:(path:string,buffer:Buffer)=>Promise<void>
run:(args:string[])=>Promise<void>
remove:(path:string)=>Promise<void>
invalidate?:()=>void
}
export function validateSystemDatabaseRestoreArchive(buffer:Buffer):void{
if(buffer.length>512*1024*1024)throw new Error("Файл бэкапа слишком большой")
@@ -16,6 +17,7 @@ export async function executeSystemDatabaseRestore(buffer:Buffer,databaseUrl:str
try{
await deps.write(file,buffer)
await deps.run(["--single-transaction","--clean","--if-exists","-d",databaseUrl,file])
deps.invalidate?.()
}catch(error){failed=true;failure=error}
finally{try{await deps.remove(file)}catch{cleanupError="Не удалось удалить временный файл восстановления"}}
if(failed){const message=failure instanceof Error?failure.message:"Восстановление PostgreSQL не подтверждено";throw new Error(message+(cleanupError?"; "+cleanupError:""))}
@@ -152,7 +152,9 @@ export async function buildFlowAnalytics(q: FlowAnalyticsQuery): Promise<FlowAna
return dto
}
export function resetFlowAnalyticsCacheForTests(): void {
export const resetFlowAnalyticsCacheForTests = invalidateFlowAnalyticsCache
export function invalidateFlowAnalyticsCache(): void {
analyticsCache = null
}
@@ -28,7 +28,9 @@ export function disableCatalogFetchForTests(): void {
catalogFetchEnabled = false
}
export function resetFlowCatalogForTests(): void {
export const resetFlowCatalogForTests = invalidateFlowClassificationCache
export function invalidateFlowClassificationCache(): void {
cidrs = []
asnPurpose = new Map()
fetchedAt = 0
+9 -3
View File
@@ -24,7 +24,7 @@ import {
factsPendingSize,
flushFlowFacts,
hourBucketIso,
resetFactsForTests,
resetFactsAfterDatabaseRestore,
} from "./traffic-flow-facts.js"
export const TICK_MS = 2_000
@@ -1039,13 +1039,20 @@ export function ingestParsedFlowsForServerForTests(serverId: number, flows: Pars
}
export function resetEngineForTests(): void {
resetEngineAfterDatabaseRestore()
pendingCap = MAX_PENDING
}
/** Writers are drained before replacing the database; discard data from the old snapshot. */
export function resetEngineAfterDatabaseRestore(): void {
pending.clear()
recent.clear()
tickAccum.clear()
rings.clear()
minuteRollup.clear()
minuteDims.clear()
resetFactsForTests()
resetFactsAfterDatabaseRestore()
ensuredParts.clear()
packetsReceived = 0
lastExporterIp = null
lastError = ""
@@ -1060,7 +1067,6 @@ export function resetEngineForTests(): void {
lastStatsPersistAt = 0
lastPersistedStats = null
bumpDataEpoch()
pendingCap = MAX_PENDING
}
export function pendingSizeForTests(): number {
+3 -1
View File
@@ -270,7 +270,9 @@ export function factsPendingSize(): number {
return hourFacts.size
}
export function resetFactsForTests(): void {
export const resetFactsForTests = resetFactsAfterDatabaseRestore
export function resetFactsAfterDatabaseRestore(): void {
hourFacts.clear()
ensuredParts.clear()
}
+3 -1
View File
@@ -187,7 +187,9 @@ export function markIfaceRefreshAttempt(serverId: number, at = Date.now()): void
lastAttempt.set(serverId, at)
}
export function resetIfaceCacheForTests(): void {
export const resetIfaceCacheForTests = invalidateIfaceCache
export function invalidateIfaceCache(): void {
cache.clear()
fetchedAt.clear()
lastAttempt.clear()
@@ -121,7 +121,9 @@ function bumpFrom(acc: DstAcc, exporterId: string, bytes: number, client: { user
let hopsCache: { key: string; at: number; dto: FlowMapHopsDto } | null = null
export function resetFlowMapHopsCacheForTests(): void {
export const resetFlowMapHopsCacheForTests = invalidateFlowMapHopsCache
export function invalidateFlowMapHopsCache(): void {
hopsCache = null
}
+9 -4
View File
@@ -58,6 +58,15 @@ export function disableRipeEnqueueForTests(): void {
}
export function resetRipeCacheForTests(): void {
invalidateRipeCache()
workerRunning = false
fetchCount = 0
enqueueEnabled = true
fetchImpl = globalThis.fetch.bind(globalThis)
}
/** Call only after resolver/persistence drain; preserve pause and runtime options. */
export function invalidateRipeCache(): void {
mem.clear()
asnHolder.clear()
inflight.clear()
@@ -68,10 +77,6 @@ export function resetRipeCacheForTests(): void {
wideIndex.length = 0
lastCandidateCount = 0
loaded = persistEnabled ? false : true
workerRunning = false
fetchCount = 0
enqueueEnabled = true
fetchImpl = globalThis.fetch.bind(globalThis)
}
export function seedRipeCacheForTests(entry: FlowIpMeta): void {