diff --git a/backend/package.json b/backend/package.json index 2ad8cdf..5f12a42 100644 --- a/backend/package.json +++ b/backend/package.json @@ -14,7 +14,7 @@ "test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts", "test:wireguard": "npx tsx src/services/wireguard-config.test.ts", "test:traffic-rate": "tsx src/services/traffic-rate.test.ts", - "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-hardening.test.ts && tsx src/services/traffic-flow-purge.test.ts", + "test:traffic-flow": "tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-hardening.test.ts && tsx src/services/traffic-flow-purge.test.ts && tsx src/services/sqlite-write-opt.test.ts", "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts" }, "dependencies": { diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index 1fae44e..4b5f1e2 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -8,18 +8,85 @@ import { env } from "../config.js" import * as schema from "./schema.js" export const SQLITE_BUSY_TIMEOUT_MS = 5000 +/** ~16 MiB page cache (negative = KiB). */ +export const SQLITE_CACHE_SIZE_KIB = 16_000 +export const SQLITE_WAL_AUTOCHECKPOINT_PAGES = 1000 + +export interface SqliteWriteStats { + insert: number + update: number + delete: number + walCheckpoint: number +} + +let writeTrace: SqliteWriteStats | null = null + +function classifyWriteSql(sql: string): keyof Omit | null { + const head = sql.trimStart().slice(0, 12).toUpperCase() + if (head.startsWith("INSERT")) return "insert" + if (head.startsWith("UPDATE")) return "update" + if (head.startsWith("DELETE")) return "delete" + return null +} + +function installSqliteWriteTrace(handle: SqliteHandle): SqliteHandle { + const origPrepare = handle.prepare.bind(handle) + handle.prepare = ((sql: string) => { + const stmt = origPrepare(sql) + const kind = classifyWriteSql(sql) + if (!kind) return stmt + const origRun = stmt.run.bind(stmt) + stmt.run = ((...args: unknown[]) => { + if (writeTrace) writeTrace[kind] += 1 + return origRun(...args) + }) as typeof stmt.run + return stmt + }) as typeof handle.prepare + + const origExec = handle.exec.bind(handle) + handle.exec = ((sql: string) => { + if (writeTrace) { + for (const part of sql.split(";")) { + const kind = classifyWriteSql(part) + if (kind) writeTrace[kind] += 1 + } + } + return origExec(sql) + }) as typeof handle.exec + + const origPragma = handle.pragma.bind(handle) + handle.pragma = ((source: string, options?: { simple?: boolean }) => { + if (writeTrace && /wal_checkpoint/i.test(source)) writeTrace.walCheckpoint += 1 + return origPragma(source, options as never) + }) as typeof handle.pragma + + return handle +} + +export function countSqliteWrites(fn: () => T): { result: T; stats: SqliteWriteStats } { + const stats: SqliteWriteStats = { insert: 0, update: 0, delete: 0, walCheckpoint: 0 } + writeTrace = stats + try { + return { result: fn(), stats } + } finally { + writeTrace = null + } +} export function applySqlitePragmas(handle: SqliteHandle): void { handle.pragma("journal_mode = WAL") handle.pragma("foreign_keys = ON") handle.pragma(`busy_timeout = ${SQLITE_BUSY_TIMEOUT_MS}`) handle.pragma("synchronous = NORMAL") + handle.pragma(`wal_autocheckpoint = ${SQLITE_WAL_AUTOCHECKPOINT_PAGES}`) + handle.pragma("temp_store = MEMORY") + handle.pragma(`cache_size = -${SQLITE_CACHE_SIZE_KIB}`) } function openSqlite(): SqliteHandle { const handle = new Database(env.DATABASE_PATH) applySqlitePragmas(handle) - return handle + return installSqliteWriteTrace(handle) } let sqlite = openSqlite() diff --git a/backend/src/modules/servers/repository/servers-repository.ts b/backend/src/modules/servers/repository/servers-repository.ts index 1bd9f11..2318c8d 100644 --- a/backend/src/modules/servers/repository/servers-repository.ts +++ b/backend/src/modules/servers/repository/servers-repository.ts @@ -1,6 +1,7 @@ import { desc, eq } from "drizzle-orm" import { db } from "../../../db/index.js" import { serverSnapshots, servers } from "../../../db/schema.js" +import { invalidateFlowCatalogCache } from "../../../services/traffic-flow-topology.js" export type ServerRow = typeof servers.$inferSelect export type SnapshotRow = typeof serverSnapshots.$inferSelect @@ -17,6 +18,7 @@ export function createServerRow( values: Omit, ): ServerRow { const [inserted] = db.insert(servers).values(values).returning().all() + invalidateFlowCatalogCache() return inserted } @@ -25,11 +27,13 @@ export function updateServerRowById( values: Partial, ): ServerRow { const [updated] = db.update(servers).set(values).where(eq(servers.id, id)).returning().all() + invalidateFlowCatalogCache() return updated } export function deleteServerRowById(id: number): void { db.delete(servers).where(eq(servers.id, id)).run() + invalidateFlowCatalogCache() } export function listSnapshotsByServerId(serverId: number, limit: number): SnapshotRow[] { diff --git a/backend/src/modules/users/repository/users-repository.ts b/backend/src/modules/users/repository/users-repository.ts index 0e1af16..b7da578 100644 --- a/backend/src/modules/users/repository/users-repository.ts +++ b/backend/src/modules/users/repository/users-repository.ts @@ -1,6 +1,7 @@ -import { and, eq } from "drizzle-orm" +import { and, count, eq } from "drizzle-orm" import { db } from "../../../db/index.js" import { appUsers, userInterfaceBindings } from "../../../db/schema.js" +import { invalidateFlowCatalogCache } from "../../../services/traffic-flow-topology.js" export type AppUserRow = typeof appUsers.$inferSelect export type BindingRow = typeof userInterfaceBindings.$inferSelect @@ -19,6 +20,7 @@ export function getUserRowByLogin(login: string): AppUserRow | undefined { export function createUserRow(values: typeof appUsers.$inferInsert): AppUserRow { const [inserted] = db.insert(appUsers).values(values).returning().all() + invalidateFlowCatalogCache() return inserted } @@ -27,11 +29,13 @@ export function updateUserRowById( values: Partial, ): AppUserRow { const [updated] = db.update(appUsers).set(values).where(eq(appUsers.id, id)).returning().all() + invalidateFlowCatalogCache() return updated } export function deleteUserRowById(id: string): void { db.delete(appUsers).where(eq(appUsers.id, id)).run() + invalidateFlowCatalogCache() } export function listBindingRows(): BindingRow[] { @@ -65,13 +69,15 @@ export function getBindingByServerIfacePeer( export function createBindingRow(values: typeof userInterfaceBindings.$inferInsert): BindingRow { const [inserted] = db.insert(userInterfaceBindings).values(values).returning().all() + invalidateFlowCatalogCache() return inserted } export function deleteBindingRowById(id: string): void { db.delete(userInterfaceBindings).where(eq(userInterfaceBindings.id, id)).run() + invalidateFlowCatalogCache() } export function countUserRows(): number { - return db.select().from(appUsers).all().length + return db.select({ n: count() }).from(appUsers).all()[0]?.n ?? 0 } diff --git a/backend/src/routes/sidebar-counts.ts b/backend/src/routes/sidebar-counts.ts index 4c8dd67..0816b5b 100644 --- a/backend/src/routes/sidebar-counts.ts +++ b/backend/src/routes/sidebar-counts.ts @@ -1,4 +1,5 @@ import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod" +import { count } from "drizzle-orm" import { listCertificatesFromServers } from "../services/certificates-service.js" import { countWireGuardInterfaces } from "../services/wireguard-live.js" import { db } from "../db/index.js" @@ -11,27 +12,22 @@ import { } from "../db/schema.js" import { listUsers } from "../modules/users/service/users-service.js" +function tableCount(table: typeof servers | typeof filterRules | typeof uptimeProbes | typeof uptimeSpeedProbes | typeof recursiveRoutes): number { + return db.select({ n: count() }).from(table).all()[0]?.n ?? 0 +} + const sidebarCountsRoutes: FastifyPluginAsyncZod = async (app) => { app.get("/sidebar-counts", async (_req, reply) => { - const [ - serversTotal, - filterRulesTotal, - uptimeProbesTotal, - uptimeSpeedProbesTotal, - recursiveRoutesTotal, - certificatesTotal, - wireguardTotal, - usersTotal, - ] = await Promise.all([ - Promise.resolve(db.select().from(servers).all().length), - Promise.resolve(db.select().from(filterRules).all().length), - Promise.resolve(db.select().from(uptimeProbes).all().length), - Promise.resolve(db.select().from(uptimeSpeedProbes).all().length), - Promise.resolve(db.select().from(recursiveRoutes).all().length), + const serversTotal = tableCount(servers) + const filterRulesTotal = tableCount(filterRules) + const uptimeProbesTotal = tableCount(uptimeProbes) + const uptimeSpeedProbesTotal = tableCount(uptimeSpeedProbes) + const recursiveRoutesTotal = tableCount(recursiveRoutes) + const [certificatesTotal, wireguardTotal] = await Promise.all([ listCertificatesFromServers().then((res) => res.certificates.length), countWireGuardInterfaces().catch(() => 0), - Promise.resolve(listUsers().length), ]) + const usersTotal = listUsers().length return reply.send({ servers: serversTotal, diff --git a/backend/src/services/alert-engine/prev-live-store.ts b/backend/src/services/alert-engine/prev-live-store.ts index 6f32049..a149cc3 100644 --- a/backend/src/services/alert-engine/prev-live-store.ts +++ b/backend/src/services/alert-engine/prev-live-store.ts @@ -37,11 +37,12 @@ export function savePrevLiveMap(kind: PrevLiveKind, map: PrevLiveStringMap) { .limit(1) .all()[0] if (existing) { + if (existing.payloadJson === payloadJson) return db.update(alertEnginePrevLive) .set({ payloadJson, updatedAt: new Date().toISOString() }) .where(eq(alertEnginePrevLive.kind, kind)) .run() - } else { - db.insert(alertEnginePrevLive).values({ kind, payloadJson, updatedAt: new Date().toISOString() }).run() + return } + db.insert(alertEnginePrevLive).values({ kind, payloadJson, updatedAt: new Date().toISOString() }).run() } diff --git a/backend/src/services/gre-bgp-snapshot-collector.ts b/backend/src/services/gre-bgp-snapshot-collector.ts index f7cd7fb..e82b38a 100644 --- a/backend/src/services/gre-bgp-snapshot-collector.ts +++ b/backend/src/services/gre-bgp-snapshot-collector.ts @@ -1,5 +1,3 @@ -import { db, sqliteDatabase } from "../db/index.js" -import { alertBgpPeerSamples, alertGreTunnelSamples } from "../db/schema.js" import { bgpPeerAlertKey, fetchBgpSessionsForAlerts } from "./bgp-peers-live.js" import { fetchGreTunnelLiveRows } from "./gre-tunnels-live.js" import type { GreBgpSnapshotRunSnapshot } from "../types/scheduler-run-snapshot.js" @@ -12,8 +10,8 @@ export function getGreBgpSnapshotCollectorState(): { running: boolean } { } /** - * Один опрос GRE + BGP по включённым серверам и запись строк в SQLite для `buildSignalSnapshot`. - * Движок оповещений больше не дублирует эти REST-запросы. + * Один опрос GRE + BGP по включённым серверам. + * Снимок для алертов живёт в `scheduler_runs.result_json` (`greTunnels` / `bgpPeers`). */ export async function collectGreBgpSnapshotOnce(): Promise { const sampledAt = new Date().toISOString() @@ -62,32 +60,8 @@ export async function collectGreBgpSnapshotOnce(): Promise ({ targetLabel: r.targetLabel, status: r.status })) snapshot.bgpPeers = bgpRows.map((s) => ({ key: bgpPeerAlertKey(s), state: s.state })) - - db.transaction((tx) => { - for (const r of greRows) { - tx.insert(alertGreTunnelSamples).values({ - sampledAt, - targetLabel: r.targetLabel, - status: r.status, - }).run() - snapshot.greWritten += 1 - } - for (const s of bgpRows) { - tx.insert(alertBgpPeerSamples).values({ - sampledAt, - peerKey: bgpPeerAlertKey(s), - state: s.state, - }).run() - snapshot.bgpWritten += 1 - } - }) - - sqliteDatabase - .prepare(`DELETE FROM alert_gre_tunnel_samples WHERE sampled_at < datetime('now', '-30 days')`) - .run() - sqliteDatabase - .prepare(`DELETE FROM alert_bgp_peer_samples WHERE sampled_at < datetime('now', '-30 days')`) - .run() + snapshot.greWritten = greRows.length + snapshot.bgpWritten = bgpRows.length if (errors.length) snapshot.errors = errors } catch (e) { diff --git a/backend/src/services/poller.ts b/backend/src/services/poller.ts index 9ec48ca..b3abf31 100644 --- a/backend/src/services/poller.ts +++ b/backend/src/services/poller.ts @@ -5,6 +5,7 @@ import type { SnapshotInsert } from "../db/schema.js" import type { SnapshotRead } from "../types/server.js" import { MikrotikClient } from "./mikrotik.js" import { parseRosCpuLoadPercent, parseRosDataSizeBytes } from "./ros-metric-parse.js" +import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js" // ── pollServer ───────────────────────────────────────────────────────────────── @@ -64,11 +65,13 @@ export async function pollServer(serverId: number): Promise { rawIpAddresses: JSON.stringify(addresses), } satisfies Partial) - // Keep server.name in sync with RouterOS identity - db.update(servers) - .set({ name: identity.name, updatedAt: now }) - .where(eq(servers.id, serverId)) - .run() + if ((identity.name || "") !== (server.name || "")) { + db.update(servers) + .set({ name: identity.name, updatedAt: now }) + .where(eq(servers.id, serverId)) + .run() + invalidateFlowCatalogCache() + } } catch (err) { // Log but don't throw — we still persist the offline snapshot diff --git a/backend/src/services/scheduler.ts b/backend/src/services/scheduler.ts index 3b65063..68b0d79 100644 --- a/backend/src/services/scheduler.ts +++ b/backend/src/services/scheduler.ts @@ -7,7 +7,7 @@ * 3. `uptime_speed` — ниже (BW-test, тяжёлый); локи узлов — `withBtestNodeLocks` в speed-сервисе. * * **Оповещения (`alert_engine`):** читают SQLite после джоб сбора (см. `buildSignalSnapshot`), в т.ч. - * `gre_bgp` → `alert_gre_tunnel_samples` / `alert_bgp_peer_samples`. После успешного завершения джоб + * `gre_bgp` → snapshot в `scheduler_runs.result_json` (`greTunnels` / `bgpPeers`). После успешного завершения джоб * `traffic`, `uptime_*`, `servers_rest_ping`, `gre_bgp` планируется **дополнительный** прогон движка * (debounce), см. [`alert-collector-hooks.ts`](./alert-collector-hooks.ts); любой другой писатель сэмплов * для снимка оповещений тоже должен вызывать `scheduleAlertEngineAfterDataCollectors()` после коммита. @@ -17,7 +17,7 @@ */ import { desc, eq, lt } from "drizzle-orm" import { db } from "../db/index.js" -import { schedulerRuns } from "../db/schema.js" +import { events, schedulerRuns } from "../db/schema.js" import type { SchedulerRunSnapshot } from "../types/scheduler-run-snapshot.js" import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js" import { @@ -71,6 +71,20 @@ const timers = new Map>() const RUN_LOG_RETENTION_MS = 30 * 24 * 60 * 60 * 1000 +const QUIET_SCHEDULER_OK_JOBS = new Set([ + "traffic", + "servers_rest_ping", + "uptime_resources", + "uptime_ping", + "uptime_speed", + "gre_bgp", + "alert_engine", +]) + +export function shouldAppendSchedulerOkEvent(jobKey: SchedulerJobKey): boolean { + return !QUIET_SCHEDULER_OK_JOBS.has(jobKey) +} + function newRunId(): string { return `sch-${Date.now()}-${Math.random().toString(36).slice(2, 10)}` } @@ -84,18 +98,21 @@ function appendSchedulerRun(row: { durationMs: number result?: SchedulerRunSnapshot | null }) { - db.insert(schedulerRuns).values({ - id: newRunId(), - jobKey: row.jobKey, - startedAt: row.startedAt, - finishedAt: row.finishedAt, - status: row.status, - error: row.error, - durationMs: row.durationMs, - resultJson: row.result ? JSON.stringify(row.result) : null, - }).run() const cutoff = new Date(Date.now() - RUN_LOG_RETENTION_MS).toISOString() - db.delete(schedulerRuns).where(lt(schedulerRuns.finishedAt, cutoff)).run() + db.transaction((tx) => { + tx.insert(schedulerRuns).values({ + id: newRunId(), + jobKey: row.jobKey, + startedAt: row.startedAt, + finishedAt: row.finishedAt, + status: row.status, + error: row.error, + durationMs: row.durationMs, + resultJson: row.result ? JSON.stringify(row.result) : null, + }).run() + tx.delete(schedulerRuns).where(lt(schedulerRuns.finishedAt, cutoff)).run() + tx.delete(events).where(lt(events.createdAt, cutoff)).run() + }) } async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise { @@ -159,19 +176,21 @@ async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise { durationMs: Date.now() - startedAt, result: snapshot ?? null, }) - appendEvent({ - level: "info", - eventType: "scheduler.job.ok", - sourceModule: "scheduler", - title: "Задача планировщика завершена", - message: `${jobKey}: выполнено за ${Date.now() - startedAt} мс`, - entityType: "job", - entityId: jobKey, - payload: { - startedAt: startedIso, - finishedAt: finishedIso, - }, - }) + if (shouldAppendSchedulerOkEvent(jobKey)) { + appendEvent({ + level: "info", + eventType: "scheduler.job.ok", + sourceModule: "scheduler", + title: "Задача планировщика завершена", + message: `${jobKey}: выполнено за ${Date.now() - startedAt} мс`, + entityType: "job", + entityId: jobKey, + payload: { + startedAt: startedIso, + finishedAt: finishedIso, + }, + }) + } if ( jobKey === "traffic" || jobKey === "servers_rest_ping" || diff --git a/backend/src/services/sqlite-write-opt.test.ts b/backend/src/services/sqlite-write-opt.test.ts new file mode 100644 index 0000000..b09556a --- /dev/null +++ b/backend/src/services/sqlite-write-opt.test.ts @@ -0,0 +1,144 @@ +import assert from "node:assert/strict" +import { randomUUID } from "node:crypto" +import { countSqliteWrites, sqliteDatabase } from "../db/index.js" +import { events } from "../db/schema.js" +import { db } from "../db/index.js" +import { + attachEngineSqlite, + bumpPacketMeta, + configureEngine, + flushPending, + ingestParsedFlowsForServerForTests, + resetEngineForTests, + setEngineError, +} from "./traffic-flow-engine.js" +import { collectGreBgpSnapshotOnce } from "./gre-bgp-snapshot-collector.js" +import { shouldAppendSchedulerOkEvent, executeSchedulerJob } from "./scheduler.js" +import { savePrevLiveMap, loadPrevLiveMap } from "./alert-engine/prev-live-store.js" +import { + invalidateFlowCatalogCache, + loadFlowTopology, + seedFlowTopologyForTests, +} from "./traffic-flow-topology.js" + +resetEngineForTests() +attachEngineSqlite(sqliteDatabase) +seedFlowTopologyForTests(null) +invalidateFlowCatalogCache() + +const idle1 = countSqliteWrites(() => { + flushPending() +}) +assert.equal(idle1.stats.walCheckpoint, 0) +assert.ok(idle1.stats.update >= 1, "first idle flush persists listener stats") + +const idle2 = countSqliteWrites(() => { + flushPending() +}) +assert.equal(idle2.stats.update, 0, "unchanged listener stats skip UPDATE") +assert.equal(idle2.stats.walCheckpoint, 0) + +bumpPacketMeta("203.0.113.9") +const changed = countSqliteWrites(() => { + flushPending() +}) +assert.equal(changed.stats.update, 1, "changed packets persist once") +assert.equal(changed.stats.walCheckpoint, 0) + +setEngineError("boom") +const errWrite = countSqliteWrites(() => { + flushPending() +}) +assert.equal(errWrite.stats.update, 1) +setEngineError("") +flushPending() + +resetEngineForTests() +configureEngine({ topN: 20 }) +ingestParsedFlowsForServerForTests(9, [{ + src: "10.1.1.1", + dst: "8.8.8.8", + proto: 6, + srcPort: 40000, + dstPort: 443, + bytes: 100, + packets: 1, + inIface: "2", + outIface: "", +}]) +const withData = countSqliteWrites(() => { + flushPending() +}) +assert.equal(withData.stats.walCheckpoint, 0, "flush with data must not TRUNCATE WAL") +assert.ok(withData.stats.insert >= 1, "flow upsert writes") + +sqliteDatabase.prepare(`DELETE FROM flow_buckets WHERE server_id = 9`).run() +sqliteDatabase.prepare(`DELETE FROM flow_minute_stats WHERE server_id = 9`).run() +sqliteDatabase.prepare(`DELETE FROM flow_minute_dims WHERE server_id = 9`).run() +sqliteDatabase.prepare(`DELETE FROM flow_daily_dims WHERE server_id = 9`).run() + +const plan = sqliteDatabase.prepare(` + EXPLAIN QUERY PLAN + SELECT id FROM flow_buckets WHERE bucket_at >= ? ORDER BY bytes DESC LIMIT 100 +`).all("2000-01-01T00:00:00.000Z") as Array<{ detail?: string }> +const planText = plan.map((p) => String(p.detail ?? "")).join(" | ") +assert.ok(planText.length > 0, "EXPLAIN QUERY PLAN returned rows") + +invalidateFlowCatalogCache() +seedFlowTopologyForTests(null) +const topoA = loadFlowTopology() +const topoB = loadFlowTopology() +assert.equal(topoA, topoB, "topology cache returns same object") +invalidateFlowCatalogCache() +const topoC = loadFlowTopology() +assert.notEqual(topoA, topoC, "invalidate rebuilds topology") + +assert.equal(shouldAppendSchedulerOkEvent("traffic"), false) +assert.equal(shouldAppendSchedulerOkEvent("alert_engine"), false) +assert.equal(shouldAppendSchedulerOkEvent("gre_bgp"), false) +assert.equal(shouldAppendSchedulerOkEvent("backups"), true) +assert.equal(shouldAppendSchedulerOkEvent("certificates_renew"), true) +assert.equal(shouldAppendSchedulerOkEvent("internet_path"), true) + +savePrevLiveMap("gre", { a: "up" }) +const prevSame = countSqliteWrites(() => { + savePrevLiveMap("gre", { a: "up" }) +}) +assert.equal(prevSame.stats.update, 0) +assert.equal(prevSame.stats.insert, 0) +savePrevLiveMap("gre", { a: "down" }) +assert.equal(loadPrevLiveMap("gre").a, "down") +savePrevLiveMap("gre", { a: "up" }) + +const greBefore = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_gre_tunnel_samples`).get() as { n: number } +const bgpBefore = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_bgp_peer_samples`).get() as { n: number } +await collectGreBgpSnapshotOnce() +const greAfter = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_gre_tunnel_samples`).get() as { n: number } +const bgpAfter = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM alert_bgp_peer_samples`).get() as { n: number } +assert.equal(greAfter.n, greBefore.n) +assert.equal(bgpAfter.n, bgpBefore.n) + +const oldId = `evt-old-${randomUUID()}` +db.insert(events).values({ + id: oldId, + createdAt: new Date(Date.now() - 40 * 24 * 60 * 60 * 1000).toISOString(), + level: "info", + eventType: "test.retention", + sourceModule: "system", + title: "old", + message: "old", +}).run() +const eventsBefore = sqliteDatabase.prepare( + `SELECT COUNT(*) AS n FROM events WHERE event_type = 'scheduler.job.ok' AND entity_id = 'alert_engine'`, +).get() as { n: number } +await executeSchedulerJob("alert_engine") +const oldGone = sqliteDatabase.prepare(`SELECT COUNT(*) AS n FROM events WHERE id = ?`).get(oldId) as { n: number } +assert.equal(oldGone.n, 0, "events older than 30 days are purged") +const eventsAfter = sqliteDatabase.prepare( + `SELECT COUNT(*) AS n FROM events WHERE event_type = 'scheduler.job.ok' AND entity_id = 'alert_engine'`, +).get() as { n: number } +assert.equal(eventsAfter.n, eventsBefore.n, "quiet jobs do not append scheduler.job.ok") + +resetEngineForTests() +console.log("sqlite-write-opt.test.ts: ok") +console.log("EXPLAIN listStoredFlowRows:", planText) diff --git a/backend/src/services/traffic-flow-analytics.ts b/backend/src/services/traffic-flow-analytics.ts index 3d84c00..650f922 100644 --- a/backend/src/services/traffic-flow-analytics.ts +++ b/backend/src/services/traffic-flow-analytics.ts @@ -1,6 +1,6 @@ import { eq } from "drizzle-orm" import { db, sqliteDatabase } from "../db/index.js" -import { appUsers, servers, userInterfaceBindings } from "../db/schema.js" +import { appUsers, userInterfaceBindings } from "../db/schema.js" import type { FlowAnalyticsDto, FlowBreakdownRow, @@ -21,7 +21,7 @@ import { listFlowRowsForWindow, type PendingFlowRow, } from "./traffic-flow-ingest.js" -import { MAX_PENDING, RING_OVERLAY } from "./traffic-flow-engine.js" +import { flowDataEpoch, MAX_PENDING, RING_OVERLAY } from "./traffic-flow-engine.js" import { resolveIfaceName } from "./traffic-flow-ifaces.js" import { getTrafficFlowSettingsRow, listHostPeers } from "./traffic-flow-settings.js" import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js" @@ -33,6 +33,7 @@ import { classifyFlowPlane, flowBps, shouldKeepPlane } from "./traffic-flow-plan import { pickInternetPeer } from "./traffic-flow-ip.js" import { enGreIfaceNames, + getServerCatalog, latestWireBps, loadFlowTopology, resolveClient, @@ -142,14 +143,46 @@ function topLabel(map: Map [s.id, s.name || s.host])) - const countryById = new Map(serverRows.map((s) => [s.id, (s.country || "").toUpperCase() || "UN"])) + const catalog = getServerCatalog() + const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name])) + const countryById = new Map([...catalog.byId].map(([id, s]) => [id, s.country])) const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : "" const wantDedup = q.dedup !== false && !ifaceFilter const excludeMesh = q.excludeMesh !== false @@ -478,19 +511,19 @@ export function listFlowExporters(minutes: number): FlowExportersDto { const { bytes, sessions } = summarizeByServer(rows) const ids = new Set([...bytes.keys()]) for (const p of listHostPeers()) ids.add(p.serverId) - const serverRows = db.select().from(servers).all() + const catalog = getServerCatalog() const emptySeries = Array(60).fill(0) as number[] - const exporters = serverRows + const exporters = catalog.list .filter((s) => ids.has(s.id)) .map((s) => { const ring = getRingMbps(s.id, "__all__") const total = bytes.get(s.id) ?? 0 return { id: String(s.id), - name: s.name || s.host, + name: s.name, subtitle: s.host, - site: s.site || "—", - country: s.country || "UN", + site: s.site, + country: s.country, status: snapshotStatus(s.id), rxNow: ring.rxNow || (total * 8) / Math.max(60, minutes * 60) / 1_000_000, txNow: ring.txNow, diff --git a/backend/src/services/traffic-flow-engine.ts b/backend/src/services/traffic-flow-engine.ts index 4c7e926..3fd88d5 100644 --- a/backend/src/services/traffic-flow-engine.ts +++ b/backend/src/services/traffic-flow-engine.ts @@ -4,7 +4,8 @@ import { classifyFlowPlaneLite } from "./traffic-flow-planes.js" import { pickServerIdForExporter, type OverlayPeerRef } from "./traffic-flow-map-exporter.js" import { applicationName } from "./traffic-flow-apps.js" import { classifyFlowDst } from "./traffic-flow-classify.js" -import { enqueueRipeMisses, lookupRipeCached } from "./traffic-flow-ripe.js" +import { enqueueRipeMisses, lookupRipeCached, pruneRipeSqlite } from "./traffic-flow-ripe.js" +import { invalidateTrafficFlowSettingsCache } from "./traffic-flow-settings.js" import { isIsoCountry } from "./traffic-flow-brands.js" import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js" import { pickInternetPeer } from "./traffic-flow-ip.js" @@ -94,10 +95,27 @@ let dropped = 0 let rowsStored = 0 let lastFlushUsedTransaction = false let lastPruneAt = 0 +let lastPassiveCheckpointAt = Date.now() +let dataEpoch = 0 +let lastPersistedStats: { + packetsReceived: number + lastDatagramAt: string | null + lastExporterIp: string | null + lastError: string +} | null = null let exporterCtx: ExporterResolveCtx | null = null const PRUNE_MS = 5 * 60_000 const LIVE_WINDOW_MS = 15 * 60_000 +const PASSIVE_CHECKPOINT_MS = 60_000 + +function bumpDataEpoch(): void { + dataEpoch += 1 +} + +export function flowDataEpoch(): number { + return dataEpoch +} function nowIso(): string { return new Date().toISOString() @@ -234,6 +252,7 @@ export function getEngineStats(): EngineStats { } export function queueParsedFlows(serverId: number, flows: ParsedFlowInput[]): void { + if (flows.length) bumpDataEpoch() const bucketAt = minuteBucketIso() const ripeMisses: string[] = [] for (const raw of flows) { @@ -422,7 +441,16 @@ export function applyRingSnapshot(rows: Array<{ key: string; inBps: number[]; ou } } -function persistListenerStats(handle: SqliteHandle): void { +function persistListenerStats(handle: SqliteHandle): boolean { + if ( + lastPersistedStats + && lastPersistedStats.packetsReceived === packetsReceived + && lastPersistedStats.lastDatagramAt === lastDatagramAt + && lastPersistedStats.lastExporterIp === lastExporterIp + && lastPersistedStats.lastError === lastError + ) { + return false + } handle.prepare(` UPDATE traffic_flow_settings SET packets_received = @packetsReceived, @@ -438,6 +466,25 @@ function persistListenerStats(handle: SqliteHandle): void { lastError, updatedAt: nowIso(), }) + lastPersistedStats = { + packetsReceived, + lastDatagramAt, + lastExporterIp, + lastError, + } + invalidateTrafficFlowSettingsCache() + return true +} + +function maybePassiveCheckpoint(handle: SqliteHandle): void { + const now = Date.now() + if (now - lastPassiveCheckpointAt < PASSIVE_CHECKPOINT_MS) return + lastPassiveCheckpointAt = now + try { + handle.pragma("wal_checkpoint(PASSIVE)") + } catch { + /* ignore */ + } } function upsertMinuteAndDaily(handle: SqliteHandle): void { @@ -586,6 +633,7 @@ function pruneStored(handle: SqliteHandle): void { } } } + pruneRipeSqlite(now) } function topNPending(rows: PendingFlowRow[]): PendingFlowRow[] { @@ -616,6 +664,7 @@ export function flushPending(): void { persistListenerStats(handle) if (pending.size === 0 && minuteRollup.size === 0 && minuteDims.size === 0) { pruneStored(handle) + maybePassiveCheckpoint(handle) lastFlushUsedTransaction = false return } @@ -665,6 +714,7 @@ export function flushPending(): void { tx(rows) lastFlushUsedTransaction = true rowsStored += rows.length + bumpDataEpoch() } catch { for (const r of rows) { try { @@ -697,11 +747,7 @@ export function flushPending(): void { /* rollup best-effort */ } pruneStored(handle) - try { - handle.pragma("wal_checkpoint(TRUNCATE)") - } catch { - /* ignore */ - } + maybePassiveCheckpoint(handle) } export function lastFlushUsedTransactionForTests(): boolean { @@ -736,6 +782,9 @@ export function resetEngineForTests(): void { rowsStored = 0 lastFlushUsedTransaction = false lastPruneAt = 0 + lastPassiveCheckpointAt = Date.now() + lastPersistedStats = null + bumpDataEpoch() pendingCap = MAX_PENDING } diff --git a/backend/src/services/traffic-flow-hardening.test.ts b/backend/src/services/traffic-flow-hardening.test.ts index 17ceab1..f1b144a 100644 --- a/backend/src/services/traffic-flow-hardening.test.ts +++ b/backend/src/services/traffic-flow-hardening.test.ts @@ -1,5 +1,10 @@ import assert from "node:assert/strict" -import { SQLITE_BUSY_TIMEOUT_MS, sqliteDatabase } from "../db/index.js" +import { + SQLITE_BUSY_TIMEOUT_MS, + SQLITE_CACHE_SIZE_KIB, + SQLITE_WAL_AUTOCHECKPOINT_PAGES, + sqliteDatabase, +} from "../db/index.js" import { MAX_FLOW_LIVE_SUBSCRIBERS, resetFlowLiveSlotsForTests, @@ -11,6 +16,16 @@ const busy = sqliteDatabase.pragma("busy_timeout") as Array<{ busy_timeout: numb const busyValue = Array.isArray(busy) ? Number(Object.values(busy[0] ?? {})[0]) : Number(busy) assert.equal(busyValue, SQLITE_BUSY_TIMEOUT_MS) +function pragmaNum(name: string): number { + const rows = sqliteDatabase.pragma(name) as Array> + const row = Array.isArray(rows) ? rows[0] : rows + return Number(Object.values(row ?? {})[0]) +} + +assert.equal(pragmaNum("wal_autocheckpoint"), SQLITE_WAL_AUTOCHECKPOINT_PAGES) +assert.equal(pragmaNum("cache_size"), -SQLITE_CACHE_SIZE_KIB) +assert.equal(pragmaNum("temp_store"), 2) + resetFlowLiveSlotsForTests() for (let i = 0; i < MAX_FLOW_LIVE_SUBSCRIBERS; i++) { assert.equal(tryAcquireFlowLiveSlot(), true) diff --git a/backend/src/services/traffic-flow-ingest.ts b/backend/src/services/traffic-flow-ingest.ts index 905453b..a0f999a 100644 --- a/backend/src/services/traffic-flow-ingest.ts +++ b/backend/src/services/traffic-flow-ingest.ts @@ -33,6 +33,7 @@ import { } from "./traffic-flow-settings.js" import { applicationName } from "./traffic-flow-apps.js" import { resolveIfaceName } from "./traffic-flow-ifaces.js" +import { getServerCatalog } from "./traffic-flow-topology.js" export type { PendingFlowRow } @@ -332,8 +333,8 @@ export function listFlowTalkers(minutes = 5): FlowStatsDto { const settings = getTrafficFlowSettingsRow() const runtime = getFlowRuntimeCounters() const rows = listFlowRowsForWindow(minutes) - const serverRows = db.select().from(servers).all() - const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host])) + const catalog = getServerCatalog() + const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name])) const agg = new Map() const protoBytes = new Map() const srcs = new Set() diff --git a/backend/src/services/traffic-flow-map-hops.ts b/backend/src/services/traffic-flow-map-hops.ts index 482e85b..66b05c3 100644 --- a/backend/src/services/traffic-flow-map-hops.ts +++ b/backend/src/services/traffic-flow-map-hops.ts @@ -1,7 +1,7 @@ import { eq } from "drizzle-orm" import type { FlowMapHop, FlowMapHopsDto, FlowMapService, FlowMapServiceEdge, FlowMapServicePath } from "@mmapp/contracts/traffic-flow" import { db } from "../db/index.js" -import { servers, userInterfaceBindings } from "../db/schema.js" +import { userInterfaceBindings } from "../db/schema.js" import { applicationName, flowRowMatchesFilter } from "./traffic-flow-apps.js" import { isNamedInternetService, @@ -15,7 +15,8 @@ import { classifyFlowPlane, shouldKeepPlane } from "./traffic-flow-planes.js" import { pickInternetPeer } from "./traffic-flow-ip.js" import { lookupRipeCached, type FlowIpMeta } from "./traffic-flow-ripe.js" import { getTrafficFlowSettingsRow } from "./traffic-flow-settings.js" -import { loadFlowTopology, resolveClient, resolveEn } from "./traffic-flow-topology.js" +import { loadFlowTopology, resolveClient, resolveEn, getServerCatalog } from "./traffic-flow-topology.js" +import { flowDataEpoch } from "./traffic-flow-engine.js" export const DEFAULT_MAP_SERVICE_MIN_SHARE_PCT = 5 export const MAP_SERVICE_NODE_CAP = 20 @@ -100,6 +101,7 @@ export function clampMapServiceMinSharePct(n: unknown): number { function hopsQueryKey(q: FlowMapHopsQuery, minSharePct: number): string { return JSON.stringify({ + epoch: flowDataEpoch(), minutes: q.minutes, serverId: q.serverId ?? null, userId: q.userId ?? null, @@ -194,8 +196,8 @@ function buildFlowMapHopsUncached(q: FlowMapHopsQuery, minSharePct: number): Flo const windowSec = Math.max(60, q.minutes * 60) const raw = listFlowRowsForWindow(q.minutes) const allow = q.userId ? userIfaceAllow(q.userId) : null - const serverRows = db.select().from(servers).all() - const nameById = new Map(serverRows.map((s) => [s.id, s.name || s.host])) + const catalog = getServerCatalog() + const nameById = new Map([...catalog.byId].map(([id, s]) => [id, s.name])) const ifaceFilter = q.iface && q.iface !== "__all__" ? q.iface : "" const wantDedup = q.dedup !== false && !ifaceFilter const excludeMesh = q.excludeMesh !== false diff --git a/backend/src/services/traffic-flow-overlay.ts b/backend/src/services/traffic-flow-overlay.ts index df5f844..2d94751 100644 --- a/backend/src/services/traffic-flow-overlay.ts +++ b/backend/src/services/traffic-flow-overlay.ts @@ -21,6 +21,7 @@ import { } from "./traffic-flow-settings.js" import { refreshFlowExporterMap, startTrafficFlowListener } from "./traffic-flow-ingest.js" import { listTrafficFlowHostFiles } from "./traffic-flow-host-files.js" +import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js" const IFACE_NAME = "wg-flow" const JH_LISTEN_PORT = 13232 @@ -288,6 +289,7 @@ export async function applyFlowOverlay( mgmtTunnelIp: address, updatedAt: new Date().toISOString(), }).where(eq(servers.id, server.id)).run() + invalidateFlowCatalogCache() upsertHostPeer({ serverId: server.id, diff --git a/backend/src/services/traffic-flow-purge.test.ts b/backend/src/services/traffic-flow-purge.test.ts index 0480c3d..9c862b6 100644 --- a/backend/src/services/traffic-flow-purge.test.ts +++ b/backend/src/services/traffic-flow-purge.test.ts @@ -43,8 +43,8 @@ try { `).run(serverId) sqliteDatabase.prepare(` INSERT INTO flow_ip_meta (prefix, asn, country, holder, ok, fetched_at) - VALUES ('8.8.8.0/24', 15169, 'US', 'Google', 1, '2026-01-01T00:00:00.000Z') - `).run() + VALUES ('8.8.8.0/24', 15169, 'US', 'Google', 1, ?) + `).run(new Date().toISOString()) sqliteDatabase.prepare(` UPDATE traffic_flow_settings SET packets_received = 42, last_exporter_ip = '10.255.254.3' WHERE id = 1 `).run() diff --git a/backend/src/services/traffic-flow-ripe.ts b/backend/src/services/traffic-flow-ripe.ts index 8fced91..53fe53f 100644 --- a/backend/src/services/traffic-flow-ripe.ts +++ b/backend/src/services/traffic-flow-ripe.ts @@ -239,6 +239,20 @@ function persistAsn(asn: number, holder: string): void { } } +/** Удаляет просроченный RIPE-кэш с диска (hit 24h / negative 6h). */ +export function pruneRipeSqlite(nowMs = Date.now()): void { + if (!persistEnabled) return + try { + const hitCutoff = new Date(nowMs - HIT_TTL_MS).toISOString() + const negCutoff = new Date(nowMs - NEG_TTL_MS).toISOString() + sqliteDatabase.prepare(`DELETE FROM flow_ip_meta WHERE ok != 0 AND fetched_at < ?`).run(hitCutoff) + sqliteDatabase.prepare(`DELETE FROM flow_ip_meta WHERE ok = 0 AND fetched_at < ?`).run(negCutoff) + sqliteDatabase.prepare(`DELETE FROM flow_asn_meta WHERE fetched_at < ?`).run(hitCutoff) + } catch { + /* table may not exist in isolated tests */ + } +} + function negative(prefix: string): FlowIpMeta { return { prefix, diff --git a/backend/src/services/traffic-flow-settings.ts b/backend/src/services/traffic-flow-settings.ts index 2ced23a..fc95e17 100644 --- a/backend/src/services/traffic-flow-settings.ts +++ b/backend/src/services/traffic-flow-settings.ts @@ -4,10 +4,42 @@ import { trafficFlowSettings } from "../db/schema.js" import type { FlowHostPeer, TrafficFlowSettingsDto, TrafficFlowSettingsPatch } from "@mmapp/contracts/traffic-flow" import { generateWireGuardKeyPair } from "./wg-keys.js" +let settingsRowCache: ReturnType | null = null + function nowIso() { return new Date().toISOString() } +export function invalidateTrafficFlowSettingsCache(): void { + settingsRowCache = null +} + +function readSettingsRow() { + return db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0] +} + +export function getTrafficFlowSettingsRow() { + if (settingsRowCache) return settingsRowCache + const row = readSettingsRow() + if (row) { + settingsRowCache = row + return row + } + const now = nowIso() + db.insert(trafficFlowSettings).values({ + id: 1, + enabled: false, + collectorIp: "10.255.254.1", + flowListenPort: 4739, + wgListenPort: 51821, + prefix: "10.255.254.0/24", + createdAt: now, + updatedAt: now, + }).run() + settingsRowCache = readSettingsRow() + return settingsRowCache! +} + function parsePeers(raw: string): FlowHostPeer[] { try { const parsed = JSON.parse(raw) as unknown @@ -20,23 +52,6 @@ function parsePeers(raw: string): FlowHostPeer[] { } } -export function getTrafficFlowSettingsRow() { - const row = db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0] - if (row) return row - const now = nowIso() - db.insert(trafficFlowSettings).values({ - id: 1, - enabled: false, - collectorIp: "10.255.254.1", - flowListenPort: 4739, - wgListenPort: 51821, - prefix: "10.255.254.0/24", - createdAt: now, - updatedAt: now, - }).run() - return db.select().from(trafficFlowSettings).where(eq(trafficFlowSettings.id, 1)).limit(1).all()[0] -} - export function toTrafficFlowSettingsDto( listener: { bound: boolean; address: string | null }, ): TrafficFlowSettingsDto { @@ -81,6 +96,7 @@ export function updateTrafficFlowSettings(patch: TrafficFlowSettingsPatch) { : Math.min(100, Math.max(0, patch.mapServiceMinSharePct)), updatedAt: nowIso(), }).where(eq(trafficFlowSettings.id, 1)).run() + invalidateTrafficFlowSettingsCache() return getTrafficFlowSettingsRow() } @@ -95,6 +111,7 @@ export function ensureHostKeys(): { publicKey: string; created: boolean } { hostPrivateKey: keys.privateKey, updatedAt: nowIso(), }).where(eq(trafficFlowSettings.id, 1)).run() + invalidateTrafficFlowSettingsCache() return { publicKey: keys.publicKey, created: true } } @@ -106,6 +123,7 @@ export function upsertHostPeer(peer: FlowHostPeer) { peersJson: JSON.stringify(peers), updatedAt: nowIso(), }).where(eq(trafficFlowSettings.id, 1)).run() + invalidateTrafficFlowSettingsCache() } export function recordFlowPacket(exporterIp: string) { @@ -116,6 +134,7 @@ export function recordFlowPacket(exporterIp: string) { packetsReceived: row.packetsReceived + 1, updatedAt: nowIso(), }).where(eq(trafficFlowSettings.id, 1)).run() + invalidateTrafficFlowSettingsCache() } export function recordFlowListenerError(message: string) { @@ -123,6 +142,7 @@ export function recordFlowListenerError(message: string) { lastError: message, updatedAt: nowIso(), }).where(eq(trafficFlowSettings.id, 1)).run() + invalidateTrafficFlowSettingsCache() } export function enableTrafficFlowIngest() { @@ -130,6 +150,7 @@ export function enableTrafficFlowIngest() { enabled: true, updatedAt: nowIso(), }).where(eq(trafficFlowSettings.id, 1)).run() + invalidateTrafficFlowSettingsCache() } export function listHostPeers(): FlowHostPeer[] { @@ -144,4 +165,5 @@ export function resetFlowIngestCounters(): void { lastError: "", updatedAt: nowIso(), }).where(eq(trafficFlowSettings.id, 1)).run() + invalidateTrafficFlowSettingsCache() } diff --git a/backend/src/services/traffic-flow-topology.ts b/backend/src/services/traffic-flow-topology.ts index c5709ef..541940c 100644 --- a/backend/src/services/traffic-flow-topology.ts +++ b/backend/src/services/traffic-flow-topology.ts @@ -27,7 +27,44 @@ export interface FlowTopology { plane: PlaneTopology } +export interface ServerCatalogEntry { + id: number + name: string + country: string + host: string + type: string + site: string +} + +const CATALOG_TTL_MS = 5_000 + let seeded: FlowTopology | null = null +let topologyCache: { at: number; topo: FlowTopology } | null = null +let serverCatalogCache: { at: number; list: ServerCatalogEntry[]; byId: Map } | null = null + +export function invalidateFlowCatalogCache(): void { + topologyCache = null + serverCatalogCache = null +} + +export function getServerCatalog(): { list: ServerCatalogEntry[]; byId: Map } { + const now = Date.now() + if (serverCatalogCache && now - serverCatalogCache.at < CATALOG_TTL_MS) { + return serverCatalogCache + } + const rows = db.select().from(servers).all() + const list: ServerCatalogEntry[] = rows.map((s) => ({ + id: s.id, + name: s.name || s.host, + country: (s.country || "").toUpperCase() || "UN", + host: s.host, + type: s.type, + site: s.site || "—", + })) + const byId = new Map(list.map((s) => [s.id, s])) + serverCatalogCache = { at: now, list, byId } + return serverCatalogCache +} function parseWanUplinks(raw: string): Array<{ iface?: string; ip?: string }> { try { @@ -44,6 +81,8 @@ function ifaceKey(serverId: number, name: string): string { export function loadFlowTopology(): FlowTopology { if (seeded) return seeded + const now = Date.now() + if (topologyCache && now - topologyCache.at < CATALOG_TTL_MS) return topologyCache.topo const serverRows = db.select().from(servers).all() const users = db.select().from(appUsers).all() const binds = db.select().from(userInterfaceBindings).all() @@ -82,7 +121,7 @@ export function loadFlowTopology(): FlowTopology { for (const h of hosts) jhHosts.add(h) } } - return { + const topo: FlowTopology = { clientIfaces, clientByIface, enNodes, @@ -95,10 +134,13 @@ export function loadFlowTopology(): FlowTopology { jhHosts, }, } + topologyCache = { at: Date.now(), topo } + return topo } export function seedFlowTopologyForTests(topo: FlowTopology | null): void { seeded = topo + invalidateFlowCatalogCache() } export function resolveClient(