From 39c8ec4a02eb53b7a8ad79e20d917cdcc267b842 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Tue, 8 Sep 2026 12:46:37 +0700 Subject: [PATCH] =?UTF-8?q?fix(netflow):=20=D0=BF=D0=BE=D1=87=D0=B8=D0=BD?= =?UTF-8?q?=D0=B8=D1=82=D1=8C=20live-=D0=B0=D0=BD=D0=B0=D0=BB=D0=B8=D1=82?= =?UTF-8?q?=D0=B8=D0=BA=D1=83=20=D0=B8=20=D1=81=D0=BD=D0=B8=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D0=B7=D0=B0=D0=BF=D0=B8=D1=81=D1=8C=20=D0=B2=20?= =?UTF-8?q?PostgreSQL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Коллектор снова отдаёт назначения в живом потоке. Ошибки записи видны в статусе, лишние перезаписи минутных агрегатов убраны. Co-authored-by: Cursor --- app/(main)/traffic/page.tsx | 6 +- backend/src/routes/traffic-flow.ts | 2 +- .../services/traffic-flow-analytics.test.ts | 30 ++- .../src/services/traffic-flow-analytics.ts | 11 +- .../services/traffic-flow-collector-worker.ts | 2 +- backend/src/services/traffic-flow-engine.ts | 225 +++++++++++++++--- .../src/services/traffic-flow-ingest.test.ts | 23 +- backend/src/services/traffic-flow-ingest.ts | 6 +- deploy/docker-compose.cdn-mm.yml | 6 + deploy/docker-compose.postgres.yml | 6 + deploy/docker-compose.traefik-cdn.yml | 6 + deploy/docker-compose.traefik.yml | 6 + deploy/docker-compose.yml | 6 + deploy/postgres.conf | 5 + hooks/use-flow-live.ts | 12 +- 15 files changed, 300 insertions(+), 52 deletions(-) diff --git a/app/(main)/traffic/page.tsx b/app/(main)/traffic/page.tsx index 13330d9..03127a0 100644 --- a/app/(main)/traffic/page.tsx +++ b/app/(main)/traffic/page.tsx @@ -952,10 +952,6 @@ export default function TrafficPage() { setFlowAnalytics(null) return } - if (range === "5m") { - setFlowAnalytics(null) - return - } if (range === "30d") { const month = new Date().toISOString().slice(0, 7) void getFlowMonthly(backendUrl, { @@ -1128,7 +1124,7 @@ export default function TrafficPage() { const ingestLine = flowIngestLine(flowStats) const collectorAlive = Boolean(flowStats?.listenerBound || flowStats?.packetsReceived) const flowError = liveError - || (flowLiveError && !(collectorAlive && /live HTTP 500/.test(flowLiveError)) ? flowLiveError : null) + || flowLiveError || (displayedFlow?.degraded ? "Коллектор перегружен: упрощённая аналитика" : null) const flowKpiItems = [ diff --git a/backend/src/routes/traffic-flow.ts b/backend/src/routes/traffic-flow.ts index 2836e44..afeba4b 100644 --- a/backend/src/routes/traffic-flow.ts +++ b/backend/src/routes/traffic-flow.ts @@ -275,7 +275,7 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => { try { while (!abort.signal.aborted) { - const payload = safeBuildLiveFlowSample(liveQuery) + const payload = await safeBuildLiveFlowSample(liveQuery) writeSse(reply.raw, payload.event, payload.data) await sleep(LIVE_TICK_MS, abort.signal) } diff --git a/backend/src/services/traffic-flow-analytics.test.ts b/backend/src/services/traffic-flow-analytics.test.ts index b3afb7f..c7b546e 100644 --- a/backend/src/services/traffic-flow-analytics.test.ts +++ b/backend/src/services/traffic-flow-analytics.test.ts @@ -16,6 +16,29 @@ import { seedRipeCacheForTests, } from "./traffic-flow-ripe.js" +{ + const liveErr = await formatLiveSseFromBuilder(() => { + throw new Error("SQLITE_BUSY") + }) + assert.equal(liveErr.event, "error") + assert.equal((liveErr.data as { error: string }).error, "SQLITE_BUSY") + const liveOk = await formatLiveSseFromBuilder(() => ({ ok: true })) + assert.equal(liveOk.event, "sample") + const liveAsync = await formatLiveSseFromBuilder(async () => ({ + uniqueSrc: 3, + destinations: [{ id: "8.8.8.8", label: "8.8.8.8", bytes: 1, packets: 1, bps: 1, percent: 100 }], + })) + assert.equal(liveAsync.event, "sample") + assert.notEqual(JSON.stringify(liveAsync.data), "{}") + assert.equal((liveAsync.data as { uniqueSrc: number }).uniqueSrc, 3) + assert.ok(Array.isArray((liveAsync.data as { destinations: unknown[] }).destinations)) + const liveReject = await formatLiveSseFromBuilder(async () => { + throw new Error("pg down") + }) + assert.equal(liveReject.event, "error") + assert.equal((liveReject.data as { error: string }).error, "pg down") +} + if (!(await withPgOrSkip())) { console.log("traffic-flow-analytics.test.ts: skip") process.exit(0) @@ -248,13 +271,6 @@ try { assert.equal(degraded.degraded, true) assert.equal(degraded.conversationsList.length, 0) assert.ok((degraded.bytes ?? 0) >= 12_000) - const liveErr = formatLiveSseFromBuilder(() => { - throw new Error("SQLITE_BUSY") - }) - assert.equal(liveErr.event, "error") - assert.equal((liveErr.data as { error: string }).error, "SQLITE_BUSY") - const liveOk = formatLiveSseFromBuilder(() => ({ ok: true })) - assert.equal(liveOk.event, "sample") const exporters = await listFlowExporters(5) const clients = await listFlowClients(5) assert.ok(Array.isArray(exporters.exporters)) diff --git a/backend/src/services/traffic-flow-analytics.ts b/backend/src/services/traffic-flow-analytics.ts index 6df9b84..43030c6 100644 --- a/backend/src/services/traffic-flow-analytics.ts +++ b/backend/src/services/traffic-flow-analytics.ts @@ -599,9 +599,12 @@ export async function listFlowClients(minutes: number): Promise<{ clients: { id: return { clients } } -export function formatLiveSseFromBuilder(build: () => unknown): { event: "sample" | "error"; data: unknown } { +export async function formatLiveSseFromBuilder( + build: () => unknown | Promise, +): Promise<{ event: "sample" | "error"; data: unknown }> { try { - return { event: "sample", data: build() } + const data = await build() + return { event: "sample", data } } catch (err) { const message = err instanceof Error ? err.message : String(err) return { event: "error", data: { error: message } } @@ -613,10 +616,10 @@ export function isFlowAnalyticsDegraded(): boolean { return health.pendingSize >= LIVE_DEGRADED_PENDING } -export function safeBuildLiveFlowSample(q: Omit): { +export async function safeBuildLiveFlowSample(q: Omit): Promise<{ event: "sample" | "error" data: unknown -} { +}> { return formatLiveSseFromBuilder(async () => { const skipHeavy = isFlowAnalyticsDegraded() return await buildFlowAnalytics({ ...q, minutes: LIVE_ANALYTICS_MINUTES, skipHeavy }) diff --git a/backend/src/services/traffic-flow-collector-worker.ts b/backend/src/services/traffic-flow-collector-worker.ts index 2771c91..9af2bba 100644 --- a/backend/src/services/traffic-flow-collector-worker.ts +++ b/backend/src/services/traffic-flow-collector-worker.ts @@ -60,7 +60,7 @@ function stopListener(): void { clearInterval(flushTimer) flushTimer = null } - void flushPending().catch((e) => { + void flushPending({ force: true }).catch((e) => { setEngineError(e instanceof Error ? e.message : String(e)) }) if (socket) { diff --git a/backend/src/services/traffic-flow-engine.ts b/backend/src/services/traffic-flow-engine.ts index 567e54b..060bf58 100644 --- a/backend/src/services/traffic-flow-engine.ts +++ b/backend/src/services/traffic-flow-engine.ts @@ -12,6 +12,8 @@ import { maybeRefreshIfaces } from "./traffic-flow-ifaces.js" import { pickInternetPeer } from "./traffic-flow-ip.js" export const TICK_MS = 2_000 +export const PERSIST_MS = 10_000 +export const STATS_PERSIST_MS = 15_000 export const RING_LEN = 60 export const MAX_PENDING = 50_000 export const DAILY_ASN_TOP = 500 @@ -53,6 +55,42 @@ function inetOrNull(value: string | null | undefined): string | null { return s.length > 0 ? s : null } +export function isValidFlowInet(value: string): boolean { + const s = value.trim() + if (!s) return false + const v4 = /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/.exec(s) + if (v4) { + return v4.slice(1).every((octet) => { + const n = Number(octet) + return Number.isInteger(n) && n >= 0 && n <= 255 + }) + } + if (!s.includes(":")) return false + if (!/^[0-9a-fA-F:]+$/.test(s)) return false + const parts = s.split(":") + if (parts.length < 3 || parts.length > 8) return false + return parts.every((p) => p.length <= 4) +} + +function clampProto(n: number): number { + if (!Number.isFinite(n)) return 0 + return Math.max(0, Math.min(255, Math.trunc(n))) +} + +function sanitizeFlowRow(r: PendingFlowRow): PendingFlowRow | null { + const src = (r.src || "").trim() || "0.0.0.0" + const dst = (r.dst || "").trim() || "0.0.0.0" + if (!isValidFlowInet(src) || !isValidFlowInet(dst)) return null + const next = inetOrNull(r.nextHop) + return { + ...r, + src, + dst, + nextHop: next && isValidFlowInet(next) ? next : "", + proto: clampProto(r.proto), + } +} + function flowUpsertParams(r: PendingFlowRow) { return { serverId: r.serverId, @@ -129,6 +167,9 @@ let rowsStored = 0 let lastFlushUsedTransaction = false let lastPruneAt = 0 let lastPassiveCheckpointAt = 0 +let lastPersistAt = 0 +let lastFlushedMinute = "" +let lastStatsPersistAt = 0 let dataEpoch = 0 let lastPersistedStats: { packetsReceived: number @@ -473,7 +514,7 @@ export function applyRingSnapshot(rows: Array<{ key: string; inBps: number[]; ou } } -async function persistListenerStats(): Promise { +async function persistListenerStats(force = false): Promise { if ( lastPersistedStats && lastPersistedStats.packetsReceived === packetsReceived @@ -483,6 +524,12 @@ async function persistListenerStats(): Promise { ) { return false } + const errorChanged = lastPersistedStats?.lastError !== lastError + const exporterChanged = lastPersistedStats?.lastExporterIp !== lastExporterIp + const now = Date.now() + if (!force && !errorChanged && !exporterChanged && now - lastStatsPersistAt < STATS_PERSIST_MS) { + return false + } await dbQuery(` UPDATE traffic_flow_settings SET packets_received = @packetsReceived, @@ -504,6 +551,7 @@ async function persistListenerStats(): Promise { lastExporterIp, lastError, } + lastStatsPersistAt = now invalidateTrafficFlowSettingsCache() return true } @@ -668,20 +716,70 @@ function topNPending(rows: PendingFlowRow[]): PendingFlowRow[] { return out } -export async function flushPending(): Promise { - pruneRecent() - rollFlowRings() - await persistListenerStats() - if (pending.size === 0 && minuteRollup.size === 0 && minuteDims.size === 0) { - await pruneStored() - lastFlushUsedTransaction = false - return - } - const rows = topNPending([...pending.values()].map(toPendingRow)) - pending.clear() - for (const row of rows) mergeInto(recent, row) +function persistDue(force: boolean, hasWork: boolean): boolean { + if (force) return true + if (!hasWork) return false + if (minuteBucketIso() !== lastFlushedMinute) return true + return Date.now() - lastPersistAt >= PERSIST_MS +} - const upsertSql = ` +async function upsertFlowBucketsBatch(rows: PendingFlowRow[]): Promise { + if (rows.length === 0) return + const days = new Set(rows.map((r) => r.bucketAt)) + for (const bucketAt of days) await ensureParentPartition("flow_buckets", bucketAt) + await pool.query({ + text: ` + INSERT INTO flow_buckets ( + server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms + ) + SELECT * + FROM UNNEST( + $1::bigint[], + $2::timestamptz[], + $3::inet[], + $4::inet[], + $5::smallint[], + $6::int[], + $7::int[], + $8::bigint[], + $9::bigint[], + $10::text[], + $11::text[], + $12::inet[], + $13::bigint[], + $14::bigint[] + ) AS t(server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms) + ON CONFLICT (server_id, bucket_at, src, dst, proto, src_port, dst_port, in_iface) + DO UPDATE SET + bytes = flow_buckets.bytes + excluded.bytes, + packets = flow_buckets.packets + excluded.packets, + out_iface = CASE WHEN excluded.out_iface != '' THEN excluded.out_iface ELSE flow_buckets.out_iface END, + next_hop = COALESCE(excluded.next_hop, flow_buckets.next_hop), + flow_start_ms = CASE + WHEN excluded.flow_start_ms > 0 AND (flow_buckets.flow_start_ms = 0 OR excluded.flow_start_ms < flow_buckets.flow_start_ms) + THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END, + flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms) + `, + values: [ + rows.map((r) => r.serverId), + rows.map((r) => r.bucketAt), + rows.map((r) => r.src), + rows.map((r) => r.dst), + rows.map((r) => r.proto), + rows.map((r) => r.srcPort), + rows.map((r) => r.dstPort), + rows.map((r) => r.bytes), + rows.map((r) => r.packets), + rows.map((r) => r.inIface), + rows.map((r) => r.outIface), + rows.map((r) => inetOrNull(r.nextHop)), + rows.map((r) => r.flowStartMs), + rows.map((r) => r.flowEndMs), + ], + }) +} + +const FLOW_UPSERT_SQL = ` INSERT INTO flow_buckets ( server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface, next_hop, flow_start_ms, flow_end_ms ) VALUES ( @@ -698,25 +796,87 @@ export async function flushPending(): Promise { THEN excluded.flow_start_ms ELSE flow_buckets.flow_start_ms END, flow_end_ms = GREATEST(flow_buckets.flow_end_ms, excluded.flow_end_ms) ` - lastFlushUsedTransaction = false + +async function upsertFlowBuckets(rows: PendingFlowRow[]): Promise { + if (rows.length === 0) return 0 try { - for (const r of rows) { - await ensureParentPartition("flow_buckets", r.bucketAt) - await dbQuery(upsertSql, flowUpsertParams(r)) - } - lastFlushUsedTransaction = true - rowsStored += rows.length - bumpDataEpoch() - } catch { + await upsertFlowBucketsBatch(rows) + return rows.length + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + setEngineError(`flow_buckets: ${message}`) + let stored = 0 for (const r of rows) { try { await ensureParentPartition("flow_buckets", r.bucketAt) - await dbQuery(upsertSql, flowUpsertParams(r)) - rowsStored += 1 - } catch { - /* ignore single-row failures */ + await dbQuery(FLOW_UPSERT_SQL, flowUpsertParams(r)) + stored += 1 + } catch (rowErr) { + if (stored === 0 && lastError.startsWith("flow_buckets:")) { + const rowMsg = rowErr instanceof Error ? rowErr.message : String(rowErr) + setEngineError(`flow_buckets: ${rowMsg}`) + } } } + return stored + } +} + +export async function flushPending(opts?: { force?: boolean }): Promise { + pruneRecent() + rollFlowRings() + const force = Boolean(opts?.force) + const hasWork = pending.size > 0 || minuteRollup.size > 0 || minuteDims.size > 0 + const due = persistDue(force, hasWork) + try { + await persistListenerStats(force) + } catch { + /* settings row may be absent in unit tests */ + } + + if (!hasWork) { + if (force) { + try { + await pruneStored() + } catch { + /* prune best-effort */ + } + } + lastFlushUsedTransaction = false + return + } + if (!due) { + lastFlushUsedTransaction = false + return + } + + const sanitized: PendingFlowRow[] = [] + let skippedInet = 0 + for (const row of topNPending([...pending.values()].map(toPendingRow))) { + const clean = sanitizeFlowRow(row) + if (!clean) { + skippedInet += 1 + continue + } + sanitized.push(clean) + } + pending.clear() + for (const row of sanitized) mergeInto(recent, row) + if (skippedInet > 0) { + setEngineError(`flow_buckets: пропуск ${skippedInet} строк с невалидным IP`) + } + + lastFlushUsedTransaction = false + lastPersistAt = Date.now() + lastFlushedMinute = minuteBucketIso() + try { + const stored = await upsertFlowBuckets(sanitized) + lastFlushUsedTransaction = stored === sanitized.length + rowsStored += stored + if (stored > 0) bumpDataEpoch() + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + setEngineError(`flow_buckets: ${message}`) } try { await upsertMinuteAndDaily() @@ -724,7 +884,11 @@ export async function flushPending(): Promise { } catch { /* rollup best-effort */ } - await pruneStored() + try { + await pruneStored() + } catch { + /* prune best-effort */ + } } export function lastFlushUsedTransactionForTests(): boolean { @@ -732,7 +896,7 @@ export function lastFlushUsedTransactionForTests(): boolean { } export async function flushPendingForTests(): Promise { - await flushPending() + await flushPending({ force: true }) } export function onEngineTick(): void { @@ -760,6 +924,9 @@ export function resetEngineForTests(): void { lastFlushUsedTransaction = false lastPruneAt = 0 lastPassiveCheckpointAt = Date.now() + lastPersistAt = 0 + lastFlushedMinute = "" + lastStatsPersistAt = 0 lastPersistedStats = null bumpDataEpoch() pendingCap = MAX_PENDING diff --git a/backend/src/services/traffic-flow-ingest.test.ts b/backend/src/services/traffic-flow-ingest.test.ts index e8fe6f9..ce0197b 100644 --- a/backend/src/services/traffic-flow-ingest.test.ts +++ b/backend/src/services/traffic-flow-ingest.test.ts @@ -21,7 +21,7 @@ import { setWantListenForTests, simulateWorkerExitForTests, } from "./traffic-flow-ingest.js" -import { configureEngine, droppedForTests, pendingSizeForTests } from "./traffic-flow-engine.js" +import { configureEngine, droppedForTests, getEngineStats, isValidFlowInet, pendingSizeForTests } from "./traffic-flow-engine.js" import { dbQuery } from "../db/index.js" import { withPgOrSkip } from "../test/pg.js" @@ -69,6 +69,27 @@ assert.equal(droppedForTests(), 3) assert.equal(peekPendingFlows().length, 3) setPendingCapForTests(null) +assert.equal(isValidFlowInet("10.0.0.1"), true) +assert.equal(isValidFlowInet("8.8.8.8"), true) +assert.equal(isValidFlowInet("0:0:0:0:0:0:0:1"), true) +assert.equal(isValidFlowInet("not-an-ip"), false) +assert.equal(isValidFlowInet("999.1.1.1"), false) + +resetFlowRingsForTests() +ingestParsedFlowsForServerForTests(1, [{ + src: "not-an-ip", + dst: "8.8.8.8", + proto: 6, + srcPort: 1, + dstPort: 443, + bytes: 10, + packets: 1, + inIface: "2", + outIface: "", +}]) +await flushPendingForTests() +assert.match(getEngineStats().lastError, /невалидн/) + resetFlowRingsForTests() configureEngine({ topN: 20 }) const talkers = Array.from({ length: 25 }, (_, i) => ({ diff --git a/backend/src/services/traffic-flow-ingest.ts b/backend/src/services/traffic-flow-ingest.ts index d8a3de8..ebf276c 100644 --- a/backend/src/services/traffic-flow-ingest.ts +++ b/backend/src/services/traffic-flow-ingest.ts @@ -267,7 +267,7 @@ export async function startTrafficFlowListener() { export function stopTrafficFlowListener() { wantListen = false stopWorkerProcess() - void flushPending().catch(() => { /* ignore */ }) + void flushPending({ force: true }).catch(() => { /* ignore */ }) state = { bound: false, address: null } } @@ -425,7 +425,7 @@ export async function ingestParsedFlowsForTests(exporterIp: string, flows: Parse if (serverId == null) return queueParsedFlows(serverId, flows) rollFlowRings() - await flushPending() + await flushPending({ force: true }) } export function ingestParsedFlowsForServerForTests(serverId: number, flows: ParsedFlowInput[]) { @@ -444,7 +444,7 @@ export function lastFlushUsedTransactionForTests(): boolean { } export async function flushPendingForTests(): Promise { - await flushPending() + await flushPending({ force: true }) } async function tableCount(name: string): Promise { diff --git a/deploy/docker-compose.cdn-mm.yml b/deploy/docker-compose.cdn-mm.yml index bef53c4..6f404a9 100644 --- a/deploy/docker-compose.cdn-mm.yml +++ b/deploy/docker-compose.cdn-mm.yml @@ -153,6 +153,12 @@ services: - io_workers=3 - -c - effective_io_concurrency=200 + - -c + - max_wal_size=2GB + - -c + - checkpoint_timeout=15min + - -c + - checkpoint_completion_target=0.9 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/docker-compose.postgres.yml b/deploy/docker-compose.postgres.yml index 8045a9f..f4d6bd0 100644 --- a/deploy/docker-compose.postgres.yml +++ b/deploy/docker-compose.postgres.yml @@ -39,6 +39,12 @@ services: - io_workers=3 - -c - effective_io_concurrency=200 + - -c + - max_wal_size=2GB + - -c + - checkpoint_timeout=15min + - -c + - checkpoint_completion_target=0.9 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/docker-compose.traefik-cdn.yml b/deploy/docker-compose.traefik-cdn.yml index 1718cf2..4fb5a5d 100644 --- a/deploy/docker-compose.traefik-cdn.yml +++ b/deploy/docker-compose.traefik-cdn.yml @@ -55,6 +55,12 @@ services: - io_workers=3 - -c - effective_io_concurrency=200 + - -c + - max_wal_size=2GB + - -c + - checkpoint_timeout=15min + - -c + - checkpoint_completion_target=0.9 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/docker-compose.traefik.yml b/deploy/docker-compose.traefik.yml index 61064dd..08b62ab 100644 --- a/deploy/docker-compose.traefik.yml +++ b/deploy/docker-compose.traefik.yml @@ -90,6 +90,12 @@ services: - io_workers=3 - -c - effective_io_concurrency=200 + - -c + - max_wal_size=2GB + - -c + - checkpoint_timeout=15min + - -c + - checkpoint_completion_target=0.9 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 5043890..a00834f 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -35,6 +35,12 @@ services: - io_workers=3 - -c - effective_io_concurrency=200 + - -c + - max_wal_size=2GB + - -c + - checkpoint_timeout=15min + - -c + - checkpoint_completion_target=0.9 volumes: # PostgreSQL 18+: VOLUME is /var/lib/postgresql (PGDATA = .../18/docker) - mmapp-pgdata:/var/lib/postgresql diff --git a/deploy/postgres.conf b/deploy/postgres.conf index 0a810cb..427e36b 100644 --- a/deploy/postgres.conf +++ b/deploy/postgres.conf @@ -16,3 +16,8 @@ effective_io_concurrency = 200 # TOAST: lz4 (не zstd — для колонок только pglz/lz4) default_toast_compression = lz4 wal_compression = lz4 + +# Реже checkpoint / меньше full-page writes (NetFlow upsert) +max_wal_size = 2GB +checkpoint_timeout = 15min +checkpoint_completion_target = 0.9 diff --git a/hooks/use-flow-live.ts b/hooks/use-flow-live.ts index d07ca5c..1ea4715 100644 --- a/hooks/use-flow-live.ts +++ b/hooks/use-flow-live.ts @@ -15,6 +15,12 @@ function parseSseBlock(block: string): { event: string; data: string } { return { event, data: dataLines.join("\n") } } +export function isValidFlowLiveSample(value: unknown): value is FlowAnalyticsDto { + if (!value || typeof value !== "object") return false + const v = value as Partial + return typeof v.uniqueSrc === "number" && Array.isArray(v.destinations) +} + export function useFlowLive(opts: { enabled: boolean backendUrl: string @@ -75,7 +81,11 @@ export function useFlowLive(opts: { if (!raw.trim() || raw.trim().startsWith(":")) continue const ev = parseSseBlock(raw) if (ev.event === "sample" && ev.data) { - const parsed = JSON.parse(ev.data) as FlowAnalyticsDto + const parsed = JSON.parse(ev.data) as unknown + if (!isValidFlowLiveSample(parsed)) { + setError("live sample пустой") + continue + } setSample(parsed) setError(parsed.degraded ? "Коллектор перегружен: упрощённая аналитика" : null) } else if (ev.event === "error" && ev.data) {