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