diff --git a/hooks/use-flow-live.ts b/hooks/use-flow-live.ts index 9471b11a..b2e881fb 100644 --- a/hooks/use-flow-live.ts +++ b/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 { + 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 { + 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, { - headers: withAuthHeaders({ Accept: "text/event-stream" }), - signal: ac.signal, - credentials: "include", - }) + const res = await openFlowLiveResponse(url, ac.signal) if (ac.signal.aborted) return if (!res.ok || !res.body) { + await res.body?.cancel() setError(`live HTTP ${res.status}`) return } diff --git a/tests/e2e/traffic-flow.spec.ts b/tests/e2e/traffic-flow.spec.ts index 12508ee6..d55ab219 100644 --- a/tests/e2e/traffic-flow.spec.ts +++ b/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) { - const control = { emptyExporters: false, failCatalog: false, hold: '', holdIface: '', fail: false, failList: false, invalidSample: undefined as unknown, mergeSample: false, metrics: {} as Record, release: () => {} } +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, release: () => {}, liveCalls: 0 } const gate = new Promise(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' diff --git a/tests/unit/flow-live-admission.test.ts b/tests/unit/flow-live-admission.test.ts new file mode 100644 index 00000000..ac1fb199 --- /dev/null +++ b/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) +})