fix(netflow): починить live-аналитику и снизить запись в PostgreSQL

Коллектор снова отдаёт назначения в живом потоке. Ошибки записи видны в статусе, лишние перезаписи минутных агрегатов убраны.

Co-authored-by: Cursor <[email protected]>

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 = [

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)
       }

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))

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<unknown>,
+): 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<FlowAnalyticsQuery, "minutes" | "skipHeavy">): {
+export async function safeBuildLiveFlowSample(q: Omit<FlowAnalyticsQuery, "minutes" | "skipHeavy">): Promise<{
   event: "sample" | "error"
   data: unknown
-} {
+}> {
   return formatLiveSseFromBuilder(async () => {
     const skipHeavy = isFlowAnalyticsDegraded()
     return await buildFlowAnalytics({ ...q, minutes: LIVE_ANALYTICS_MINUTES, skipHeavy })

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) {

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<boolean> {
+async function persistListenerStats(force = false): Promise<boolean> {
   if (
     lastPersistedStats
     && lastPersistedStats.packetsReceived === packetsReceived
@@ -483,6 +524,12 @@ async function persistListenerStats(): Promise<boolean> {
   ) {
     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<boolean> {
     lastExporterIp,
     lastError,
   }
+  lastStatsPersistAt = now
   invalidateTrafficFlowSettingsCache()
   return true
 }
@@ -668,20 +716,70 @@ function topNPending(rows: PendingFlowRow[]): PendingFlowRow[] {
   return out
 }
 
-export async function flushPending(): Promise<void> {
-  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<void> {
+  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<void> {
         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<number> {
+  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<void> {
+  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<void> {
   } 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<void> {
-  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

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) => ({

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<void> {
-  await flushPending()
+  await flushPending({ force: true })
 }
 
 async function tableCount(name: string): Promise<number> {

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

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

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

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

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

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

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<FlowAnalyticsDto>
+  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) {