fix(netflow): восстановить открытие потока после временного ограничения

hooks/use-flow-live.ts

@@ -20,6 +20,45 @@ export function isValidFlowLiveSample(value: unknown): value is FlowAnalyticsDto
   return parsed.success && [...parsed.data.rxSeries, ...parsed.data.txSeries].every(value => Number.isFinite(value) && value >= 0)
 }
 
+const liveAdmissionDelaysMs = [250, 750, 1500]
+
+function waitForLiveAdmission(delayMs: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve, reject) => {
+    signal.throwIfAborted()
+    const abort = () => {
+      clearTimeout(timer)
+      reject(signal.reason)
+    }
+    const timer = setTimeout(() => {
+      signal.removeEventListener("abort", abort)
+      resolve()
+    }, delayMs)
+    signal.addEventListener("abort", abort, { once: true })
+  })
+}
+
+export async function openFlowLiveResponse(
+  url: string,
+  signal: AbortSignal,
+  request: typeof fetch = fetch,
+): Promise<Response> {
+  const open = () => {
+    signal.throwIfAborted()
+    return request(url, {
+      headers: withAuthHeaders({ Accept: "text/event-stream" }),
+      signal,
+      credentials: "include",
+    })
+  }
+  for (const delayMs of liveAdmissionDelaysMs) {
+    const response = await open()
+    if (response.status !== 429) return response
+    await response.body?.cancel()
+    await waitForLiveAdmission(delayMs, signal)
+  }
+  return open()
+}
+
 export function useFlowLive(opts: {
   enabled: boolean
   backendUrl: string
@@ -64,13 +103,10 @@ export function useFlowLive(opts: {
 
     void (async () => {
       try {
-        const res = await fetch(url, {
+        const res = await openFlowLiveResponse(url, ac.signal)
-          headers: withAuthHeaders({ Accept: "text/event-stream" }),
-          signal: ac.signal,
-          credentials: "include",
-        })
         if (ac.signal.aborted) return
         if (!res.ok || !res.body) {
+          await res.body?.cancel()
           setError(`live HTTP ${res.status}`)
           return
         }

tests/e2e/traffic-flow.spec.ts

@@ -1,8 +1,8 @@
 import { installOpenLiveStream, resumeLiveStream, sendLiveHeartbeat } from "./live-stream-fixture"
 import { test, expect, type Page } from '@playwright/test'
 
-async function setup(page: Page) {
+async function setup(page: Page, liveRefusals = 0) {
-  const control = { emptyExporters: false, failCatalog: false, hold: '', holdIface: '', fail: false, failList: false, invalidSample: undefined as unknown, mergeSample: false, metrics: {} as Record<string, unknown>, release: () => {} }
+  const control = { emptyExporters: false, failCatalog: false, hold: '', holdIface: '', fail: false, failList: false, invalidSample: undefined as unknown, mergeSample: false, metrics: {} as Record<string, unknown>, release: () => {}, liveCalls: 0 }
   const gate = new Promise<void>(resolve => { control.release = resolve })
   const card = { id: '1', name: 'Экспортёр тест', subtitle: '', site: 'TEST', country: 'RU', status: 'online', rxNow: 2, txNow: 3, sessions: 1, rxSeries: [1, 2], txSeries: [2, 3], bytes: 200 }
   const data = { bpsNow: 1000000, bytes: 200, packets: 2, conversations: 1, uniqueSrc: 1, uniqueDst: 1, topProto: 'TCP', rxSeries: [1], txSeries: [2], applications: [], protocols: [], sources: [], destinations: [], interfaces: [], conversationsList: [], ifaces: [], live: false }
@@ -17,7 +17,11 @@ async function setup(page: Page) {
       if (control.fail) return route.fulfill({ status: 500, json: { error: 'Срез временно недоступен' } })
       return route.fulfill({ json: { ...data, ...control.metrics, topCategory: url.searchParams.get('serverId') === '2' ? 'Второй сервер' : 'Категория ' + range } })
     }
-    if (path === '/api/traffic/flow/live') return route.fulfill({ contentType: 'text/event-stream', body: 'event: sample\ndata: ' + JSON.stringify({ ...data, live: true }) + '\n\n' + (control.invalidSample === undefined ? '' : 'event: sample\ndata: ' + JSON.stringify(control.mergeSample ? { ...data, ...(control.invalidSample as object) } : control.invalidSample) + '\n\n') })
+    if (path === '/api/traffic/flow/live') {
+      control.liveCalls++
+      if (control.liveCalls <= liveRefusals) return route.fulfill({ status: 429, json: { error: 'Ресурс чтения NetFlow занят; повторите запрос позже' } })
+      return route.fulfill({ contentType: 'text/event-stream', body: 'event: sample\ndata: ' + JSON.stringify({ ...data, live: true }) + '\n\n' + (control.invalidSample === undefined ? '' : 'event: sample\ndata: ' + JSON.stringify(control.mergeSample ? { ...data, ...(control.invalidSample as object) } : control.invalidSample) + '\n\n') })
+    }
     if (path === '/api/traffic/flow/exporters' && control.failList) return route.fulfill({ status: 403, json: { error: 'Нет доступа к экспортёрам' } })
     const json = path === '/api/auth/config' ? { required: false }
       : path === '/api/traffic/flow/exporters' ? { exporters: control.emptyExporters ? [] : [card, { ...card, id: '2', name: 'Экспортёр второй' }] }
@@ -35,6 +39,18 @@ async function setup(page: Page) {
   return control
 }
 
+test('Flow: начальный HTTP 429 сменяется фактическими данными SSE', async ({ page }) => {
+  const statuses: number[] = []
+  page.on('response', response => {
+    if (new URL(response.url()).pathname === '/api/traffic/flow/live') statuses.push(response.status())
+  })
+  const control = await setup(page, 1)
+  await expect(page.getByLabel('Получение аналитики потоков')).toContainText('Получено (live-поток)')
+  await expect(page.getByText('live HTTP 429', { exact: true })).toHaveCount(0)
+  expect(statuses).toEqual([429, 200])
+  expect(control.liveCalls).toBe(2)
+})
+
 test('Flow: поздний ответ периода не заменяет текущую аналитику', async ({ page }) => {
   const control = await setup(page)
   control.hold = '24h'

tests/unit/flow-live-admission.test.ts

@@ -0,0 +1,54 @@
+import assert from "node:assert/strict"
+import test from "node:test"
+import { openFlowLiveResponse } from "../../hooks/use-flow-live"
+
+const url = "http://127.0.0.1/api/traffic/flow/live"
+
+test("opens the NetFlow stream after a transient 429", async () => {
+  const { signal } = new AbortController()
+  let calls = 0
+  let discarded = false
+  const response = await openFlowLiveResponse(url, signal, async (target, init) => {
+    calls++
+    assert.equal(target, url)
+    assert.equal(init?.signal, signal)
+    assert.equal(new Headers(init?.headers).get("Accept"), "text/event-stream")
+    if (calls === 1) {
+      return new Response(new ReadableStream({ cancel() { discarded = true } }), { status: 429 })
+    }
+    return new Response("event: sample\ndata: {}\n\n", { headers: { "Content-Type": "text/event-stream" } })
+  })
+  assert.equal(calls, 2)
+  assert.equal(discarded, true)
+  assert.equal(response.status, 200)
+})
+
+test("stops retrying when the NetFlow live view unmounts", async () => {
+  const controller = new AbortController()
+  let calls = 0
+  const response = openFlowLiveResponse(url, controller.signal, async () => {
+    calls++
+    return new Response(new ReadableStream({
+      cancel() { setTimeout(() => controller.abort(), 20) },
+    }), { status: 429 })
+  })
+  await assert.rejects(response, { name: "AbortError" })
+  assert.equal(calls, 1)
+})
+
+test("keeps the live admission budget bounded and leaves authorization failures terminal", async () => {
+  const { signal } = new AbortController()
+  let calls = 0
+  const limited = await openFlowLiveResponse(url, signal, async () => {
+    calls++
+    return new Response(null, { status: 429 })
+  })
+  assert.equal(limited.status, 429)
+  assert.equal(calls, 4)
+  const unauthorized = await openFlowLiveResponse(url, signal, async () => {
+    calls++
+    return new Response(null, { status: 401 })
+  })
+  assert.equal(unauthorized.status, 401)
+  assert.equal(calls, 5)
+})