fix(traffic): строить непрерывную историю коротких периодов

backend/src/services/traffic-flow-analytics.test.ts

@@ -4,7 +4,7 @@ import {
   ingestParsedFlowsForServerForTests,
   resetFlowRingsForTests,
 } from "./traffic-flow-ingest.js"
-import { buildFlowAnalytics, formatLiveSseFromBuilder, getFlowMonthly, listFlowClients, listFlowExporters } from "./traffic-flow-analytics.js"
+import { buildFlowAnalytics, formatLiveSseFromBuilder, getFlowMonthly, listFlowClients, listFlowExporters, resetFlowAnalyticsCacheForTests } from "./traffic-flow-analytics.js"
 import { withPgOrSkip } from "../test/pg.js"
 import { dbQuery } from "../db/index.js"
 import { seedFlowTopologyForTests, type FlowTopology } from "./traffic-flow-topology.js"
@@ -556,6 +556,22 @@ try {
   }
 }
 
+// Historical 5m/15m charts survive an empty live ring (e.g. immediately after restart).
+resetFlowRingsForTests()
+await dbQuery("DELETE FROM flow_buckets WHERE server_id=$1", [serverId])
+for (const minutesAgo of [2, 3, 4, 10]) {
+  await dbQuery(`INSERT INTO flow_buckets (server_id,bucket_at,src,dst,in_iface,bytes,packets)
+    VALUES ($1,date_trunc('minute',now())-($2::int*interval '1 minute'),'10.1.1.8','8.8.8.8','2',60000,10)`, [serverId, minutesAgo])
+}
+for (const minutes of [5, 15]) for (const dedup of [true, false]) {
+  resetFlowAnalyticsCacheForTests()
+  const history = await buildFlowAnalytics({ minutes, serverId, dedup, excludeMesh:false, excludeOverlay:false })
+  assert.equal(history.rxSeries.length, minutes)
+  assert.equal(history.rxSeries.filter(v => v > 0).length, minutes === 5 ? 3 : 4)
+  const missing = await buildFlowAnalytics({ minutes, serverId, iface:"not-an-interface" })
+  assert(missing.rxSeries.every(v => v === 0), "Historical chart respects interface filter")
+}
+
 } finally {
   for (const table of ["flow_daily_dims", "flow_hour_facts", "flow_daily_facts", "flow_buckets", "traffic_samples", "user_interface_bindings", "server_snapshots"]) {
     await dbQuery("DELETE FROM " + table + " WHERE server_id = $1", [serverId])

backend/src/services/traffic-flow-analytics.ts

@@ -423,9 +423,10 @@ async function buildFlowAnalyticsUncached(q: FlowAnalyticsQuery): Promise<FlowAn
     ? getRingMbps(ringServer, ifaceFilter === "" ? "__all__" : (ifacesMap.get(ifaceFilter)?.index || ifaceFilter))
     : { rx: Array(60).fill(0) as number[], tx: Array(60).fill(0) as number[], rxNow: 0, txNow: 0 }
 
+  // The live ring covers about one minute; historical charts must use the requested window.
   const fromBuckets = seriesFromRows(matched, q.minutes)
-  const rxSeries = q.minutes <= 15 ? ring.rx : fromBuckets.rx
-  const txSeries = q.minutes <= 15 ? ring.tx : fromBuckets.tx
+  const rxSeries = fromBuckets.rx
+  const txSeries = fromBuckets.tx
 
   const ifaceRows = [...ifacesMap.entries()]
     .sort((a, b) => b[1].bytes - a[1].bytes)

backend/src/services/traffic-flow-ingest.ts

@@ -439,7 +439,7 @@ export async function listStoredFlowRows(sinceIso: string): Promise<PendingFlowR
 
 export async function listFlowRowsForWindow(minutes: number): Promise<PendingFlowRow[]> {
   const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString()
-  if (minutes <= 15 && !worker) return await listLiveFlowRows(sinceIso)
+  // History remains available while the collector starts or is stopped; pending rows are merged below.
   return await listStoredFlowRows(sinceIso)
 }
 

backend/src/services/traffic-rate.test.ts

@@ -132,7 +132,7 @@ const observed=(rows:ReturnType<typeof observationSample>[])=>buildTrafficFromSa
 assert.ok(observed([]).observations?.rxSeries.every(v=>!v));assert.equal(observed([]).observations?.rxNow,false)
 assert.equal(observed([observationSample(0,10,10)]).observations?.rxNow,false)
 const observedZero=observed([observationSample(0,10,10),observationSample(30_000,10,10)])
-assert.equal(observedZero.rxNow,0);assert.equal(observedZero.observations?.rxNow,true);assert.equal(observedZero.observations?.rxSeries.filter(Boolean).length,1)
+assert.equal(observedZero.rxNow,0);assert.equal(observedZero.observations?.rxNow,true);assert.equal(observedZero.observations?.rxSeries.filter(Boolean).length,30)
 const partialReset=observed([observationSample(0,10,10),observationSample(30_000,1,20)])
 assert.equal(partialReset.observations?.rxNow,false);assert.equal(partialReset.observations?.txNow,true)
 const fullReset=observed([observationSample(0,10,10),observationSample(20_000,20,20),observationSample(40_000,1,1)])
@@ -147,3 +147,28 @@ console.log("traffic observation metadata tests ok")
 const complete=observed([observationSample(0,10,10),observationSample(30_000,20,20),observationSample(0,10,10,"ether2"),observationSample(30_000,20,20,"ether2")]);assert.equal(complete.observations?.rxNow,true)
 const badTime=observed([{...observationSample(0,10,10),sampledAt:"invalid"},observationSample(30_000,20,20)]);assert.equal(badTime.observations?.rxNow,false)
 assert.equal(rateBpsFromDelta(0,Number.MAX_VALUE,0,1000),null)
+
+// A 30-second collector must produce a line across measured intervals on short ranges.
+for (const minutes of [5, 15]) {
+  const span = minutes * 60_000
+  const rows = Array.from({ length: span / 30_000 + 1 }, (_, i) => ({
+    ...observationSample(i * 30_000, i * 3_750_000, i * 1_875_000),
+    continuityKnown: true, nativeCounterIdentity: "ether1-v1",
+  }))
+  const history = buildTrafficFromSamples(rows, observationStart, observationStart + span, undefined, undefined, true)
+  assert(history.observations!.rxSeries.every(Boolean), `${minutes}m continuous measured RX`)
+  assert(history.observations!.txSeries.every(Boolean), `${minutes}m continuous measured TX`)
+  assert(history.rxSeries.every(v => v === 1)); assert(history.txSeries.every(v => v === 0.5))
+  assert.equal(history.rxTotalGiB, (span / 30_000) * 3_750_000 / 1024 ** 3)
+  // Explicit loss of continuity stays unknown; it must not be bridged by chart rendering.
+  rows[2].continuityKnown = false
+  const gap = buildTrafficFromSamples(rows, observationStart, observationStart + span, undefined, undefined, true)
+  assert.equal(gap.observations!.rxSeries.filter(v => !v).length, 30_000 / (span / 60))
+  assert(gap.observations!.rxSeries.slice(0, 30_000 / (span / 60)).every(Boolean))
+  const partial = buildTrafficFromSamples(rows.slice(4), observationStart, observationStart + span, undefined, undefined, true)
+  assert(partial.observations!.rxSeries.slice(0, 120_000 / (span / 60)).every(v => !v), "No history invented before first sample")
+}
+// Unequal poll intervals contribute by covered duration when sharing a chart bucket.
+const weighted = observed([observationSample(0,0,0),observationSample(500,62_500,62_500),observationSample(1500,312_500,312_500)])
+assert.equal(weighted.rxSeries[0], 1.5)
+assert.equal(weighted.rxSeries[1], 2)

backend/src/services/traffic-rate.ts

@@ -114,6 +114,34 @@ function parseIsoMs(iso: string): number {
   return Date.parse(iso)
 }
 
+/** A counter delta measures its entire interval, not just the next poll's timestamp. */
+function intervalSeries(
+  waves: Map<number, PeakWave>, start: number, end: number, side: "rx" | "tx",
+): { values: number[]; observed: boolean[] } {
+  const sums = Array(SERIES_POINTS).fill(0) as number[]
+  const durations = Array(SERIES_POINTS).fill(0) as number[]
+  const unknown = Array(SERIES_POINTS).fill(false) as boolean[]
+  const width = (end - start) / SERIES_POINTS
+  if (!(width > 0)) return { values: sums, observed: unknown }
+  for (const [at, wave] of waves) {
+    const from = Math.max(start, wave.start), to = Math.min(end, at)
+    if (!(to > from)) continue
+    const first = Math.max(0, Math.floor((from - start) / width))
+    const last = Math.min(SERIES_POINTS - 1, Math.ceil((to - start) / width) - 1)
+    for (let i = first; i <= last; i++) {
+      const overlap = Math.min(to, start + (i + 1) * width) - Math.max(from, start + i * width)
+      if (!(overlap > 0)) continue
+      sums[i] += wave[side] * overlap
+      durations[i] += overlap
+      unknown[i] ||= !wave[side === "rx" ? "rxKnown" : "txKnown"]
+    }
+  }
+  return {
+    values: sums.map((sum, i) => durations[i] > 0 ? sum / durations[i] : 0),
+    observed: durations.map((duration, i) => duration > 0 && !unknown[i]),
+  }
+}
+
 export function sampleSeriesKey(interfaceName: string, peerPublicKey = ""): string {
   return `${interfaceName}\0${peerPublicKey}`
 }
@@ -232,8 +260,6 @@ export function buildTrafficFromSamples(
   const latestWave = membership.get(latestMembershipAt)
   if (latestWave?.declared) sessions = latestWave.active
 
-  const rxSeries = bucketAvg(rxPoints, rangeStartMs, rangeEndMs)
-  const txSeries = bucketAvg(txPoints, rangeStartMs, rangeEndMs)
   const lastTs = [...byTs.keys()].sort((a, b) => a - b).at(-1)
   const last = lastTs != null ? byTs.get(lastTs) : undefined
   const peak = (points: Array<{t:number;v:number}>, side: "rxObserved"|"txObserved") => points.reduce((m,p) => {
@@ -249,6 +275,8 @@ export function buildTrafficFromSamples(
     const complete = wave.sameStart && wave.count === expectedAt(t)
     waves.set(t,{start:wave.start,rx:wave.rx,tx:wave.tx,rxKnown:complete&&wave.rxObserved,txKnown:complete&&wave.txObserved})
   }
+  const rxHistory = intervalSeries(waves, rangeStartMs, rangeEndMs, "rx")
+  const txHistory = intervalSeries(waves, rangeStartMs, rangeEndMs, "tx")
   return retainPeakWaves({
     rxNow: last?.rx ?? 0,
     txNow: last?.tx ?? 0,
@@ -257,13 +285,13 @@ export function buildTrafficFromSamples(
     rxTotalGiB: rxBytesDelta / (1024 ** 3),
     txTotalGiB: txBytesDelta / (1024 ** 3),
     sessions,
-    rxSeries,
-    txSeries,
+    rxSeries: rxHistory.values,
+    txSeries: txHistory.values,
     ...(includeObservations?{observations:{
       rxNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.rxObserved&&last.count===expectedAt(lastTs!),
       txNow:lastAttemptAt!==null&&lastTs===lastAttemptAt&&!!last?.txObserved&&last.count===expectedAt(lastTs!),
-      rxSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.rxObserved&&v.count===expectedAt(t)&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
-      txSeries:bucketAvg([...byTs].map(([t,v])=>({t,v:Number(v.txObserved&&v.count===expectedAt(t)&&t>=rangeStartMs&&t<=rangeEndMs)})),rangeStartMs,rangeEndMs).map(v=>v===1),
+      rxSeries:rxHistory.observed,
+      txSeries:txHistory.observed,
       sampledAt:lastAttemptAt===null?null:new Date(lastAttemptAt).toISOString(),
       measurementIntervalMs:lastTs===lastAttemptAt?last?.intervalMs??null:null,
       rxPeak:[...waves.values()].some(w=>w.rxKnown),

tests/e2e/traffic-short-history.spec.ts

@@ -0,0 +1,38 @@
+import {test, expect} from '@playwright/test'
+import {buildTrafficFromSamples} from '../../backend/src/services/traffic-rate'
+
+for (const mode of ['servers','flows'] as const) for (const minutes of [5,15]) {
+  test(`${mode}: ${minutes} минут рисуются линией`, async ({page}) => {
+    const end=Date.now(), name='Короткая история'
+    const history=(span:number)=>buildTrafficFromSamples(Array.from({length:span*2+1},(_,i)=>({
+      interfaceName:'ether1',sampledAt:new Date(end-span*60000+i*30000).toISOString(),
+      rxBytes:i*3750000,txBytes:i*1875000,rxBps:0,txBps:0,running:true,disabled:false,
+    })),end-span*60000,end,undefined,undefined,true)
+    await page.route('**/health',r=>r.fulfill({json:{ok:true}}))
+    await page.route('**/api/**',r=>{
+      const url=new URL(r.request().url()),path=url.pathname,span=url.searchParams.get('range')==='5m'?5:url.searchParams.get('range')==='15m'?15:60
+      const built=history(span),server={id:'1',name,site:'TEST',country:'RU',status:'online',...built,rxTotal:1,txTotal:1,volumeUnit:'GiB'}
+      const card={id:'1',name,subtitle:'',site:'TEST',country:'RU',status:'online',rxNow:1,txNow:0,sessions:1,bytes:60000,rxSeries:[1,2,1],txSeries:[0,0,0]}
+      const analytics={bpsNow:1000000,bytes:60000,packets:10,conversations:1,uniqueSrc:1,uniqueDst:1,topProto:'TCP',rxSeries:Array.from({length:span},(_,i)=>1+i%3),txSeries:Array(span).fill(0),applications:[],protocols:[],sources:[],destinations:[],interfaces:[],conversationsList:[],ifaces:[],live:false}
+      if(path.endsWith('/live'))return r.fulfill({status:503,json:{error:'Live выключен в проверке истории'}})
+      const json=path==='/api/auth/config'?{required:false}:path==='/api/traffic/servers'?{servers:[server]}:
+        path==='/api/traffic/servers/1'?{server}:path==='/api/traffic/flow/exporters'?{exporters:[card]}:
+        path==='/api/traffic/flow/analytics'?analytics:path==='/api/traffic/flow/clients'?{clients:[]}:
+        path==='/api/traffic/flows'?{exportersOnline:1,bytesPerMin:1,uniqueSrc:1,uniqueDst:1,topProto:'TCP',talkers:[],listenerBound:true}:
+        path==='/api/traffic/users'?{users:[]}:path.endsWith('/interfaces')||path.endsWith('/bound-interfaces')?{interfaces:[]}:
+        path==='/api/servers'?[]:{}
+      return r.fulfill({json})
+    })
+    await page.goto('/traffic')
+    if(mode==='flows')await page.getByRole('group',{name:'Разделы трафика'}).getByRole('button',{name:'Потоки',exact:true}).click()
+    await page.getByRole('button',{name:`${minutes}м`,exact:true}).click()
+    const chart=page.getByRole('region',{name:'История RX/TX'})
+    await expect(chart).toBeVisible()
+    const curves=chart.locator('path.recharts-line-curve')
+    await expect(curves).toHaveCount(2)
+    await expect.poll(async()=>((await curves.first().getAttribute('d'))?.match(/[CL]/g)||[]).length).toBeGreaterThan(2)
+    await expect.poll(()=>curves.first().evaluate(el=>(el as SVGGraphicsElement).getBBox().width/el.closest('[role="region"]')!.clientWidth)).toBeGreaterThan(0.7)
+    await expect(chart.locator('[data-isolated-direction]')).toHaveCount(0)
+    await page.screenshot({path:`tmp/netflow-agent-stand/traffic-short-${mode}-${minutes}.png`,fullPage:true})
+  })
+}