fix(traffic): считать общий пик по совпадающим интервалам

backend/package.json

@@ -19,7 +19,7 @@
     "test:exec": "tsx src/services/mikrotik-logs.test.ts && tsx src/routes/exec.test.ts && tsx src/plugins/auth.exec.test.ts",
     "test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts && tsx src/plugins/auth.ospf-optimizer.test.ts",
     "test:wireguard": "npx tsx src/services/wireguard-config.test.ts && npx tsx src/services/wireguard-ipv6.test.ts && npx tsx --test src/services/wireguard-import.test.ts src/services/wireguard-import-apply.test.ts src/services/wireguard-import-revision.test.ts src/services/wireguard-delete.test.ts && npx tsx src/routes/wireguard-import.test.ts",
-    "test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-freshness.test.ts && tsx src/services/traffic-live.test.ts && tsx src/services/traffic-observation-sql.test.ts && tsx src/services/traffic-history-sql.test.ts && tsx src/services/traffic-live-sql.test.ts && tsx src/services/traffic-volume-sql.test.ts && tsx src/services/traffic-binding-status.test.ts && tsx src/services/traffic-status-sql.test.ts && tsx src/services/traffic-gap.test.ts && tsx src/services/traffic-identity.test.ts && tsx src/services/traffic-membership.test.ts",
+    "test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-freshness.test.ts && tsx src/services/traffic-live.test.ts && tsx src/services/traffic-observation-sql.test.ts && tsx src/services/traffic-history-sql.test.ts && tsx src/services/traffic-live-sql.test.ts && tsx src/services/traffic-volume-sql.test.ts && tsx src/services/traffic-binding-status.test.ts && tsx src/services/traffic-status-sql.test.ts && tsx src/services/traffic-gap.test.ts && tsx src/services/traffic-identity.test.ts && tsx src/services/traffic-membership.test.ts && tsx src/services/traffic-aggregate-peak.test.ts",
     "test:traffic-flow": "tsx src/routes/traffic-flow-query.test.ts && tsx src/services/traffic-flow-month.test.ts && tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-ipv6-pipeline.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-ifindex.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-dest.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-purge.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/services/traffic-flow-facts.test.ts && tsx src/services/traffic-flow-facts-filter.test.ts && tsx src/services/traffic-flow-facts-rebuild.test.ts && tsx src/services/statistics-aggregate.test.ts",
     "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts && tsx src/modules/users/interface-inventory.test.ts && tsx src/routes/users-catalog.test.ts",
     "test:pg": "tsx src/db/pg-pool.test.ts && tsx src/db/sql-bind.test.ts && tsx src/db/sqlite-json.test.ts && tsx src/db/traffic-flags.test.ts && tsx src/db/pg-schema.test.ts && tsx src/services/config-revisions.test.ts",

backend/src/services/traffic-aggregate-peak.test.ts

@@ -0,0 +1,31 @@
+import assert from 'node:assert/strict'
+import {buildTrafficFromSamples, mergeBuiltTraffic} from './traffic-rate.js'
+
+const at = Date.parse('2026-10-13T00:00:00Z')
+const rows = (name:string, bytes:number[], times=[0,30_000,60_000], known=[false,true,true]) => bytes.map((value,i)=>({
+  interfaceName:name,sampledAt:new Date(at+times[i]).toISOString(),rxBytes:value,txBytes:value/2,
+  rxBps:0,txBps:0,running:true,disabled:false,countersKnown:true,continuityKnown:known[i],nativeCounterIdentity:name,
+}))
+const build = (r:ReturnType<typeof rows>) => buildTrafficFromSamples(r,at,at+60_000,undefined,undefined,true)
+const a=build(rows('a',[0,375_000_000,375_000_000]))
+const b=build(rows('b',[0,0,375_000_000]))
+const separate=mergeBuiltTraffic([a,b])
+assert.equal(separate.rxPeak,100);assert.equal(separate.txPeak,50)
+assert.equal(separate.observations!.rxPeak,true)
+assert.equal(separate.rxTotalGiB,a.rxTotalGiB+b.rxTotalGiB)
+assert.equal(mergeBuiltTraffic([a,a]).rxPeak,200)
+assert.equal(mergeBuiltTraffic([separate,separate]).rxPeak,200,'nested merge preserves raw intervals')
+assert.equal(mergeBuiltTraffic([build(rows('z',[0,0,0])),build(rows('z2',[0,0,0]))]).observations!.rxPeak,true,'measured zero is known')
+const missing=build(rows('missing',[0,0,375_000_000],[0,30_000,60_000],[false,false,false]))
+assert.equal(mergeBuiltTraffic([a,missing]).observations!.rxPeak,false)
+const jitter=build(rows('jitter',[0,0,375_000_000],[1,30_001,59_999]))
+assert.equal(mergeBuiltTraffic([a,jitter]).observations!.rxPeak,false,'sharing a chart bucket does not establish a matching wave')
+const differentStart=build(rows('start',[0,375_000_000],[15_000,30_000]))
+assert.equal(mergeBuiltTraffic([a,differentStart]).observations!.rxPeak,false,'same end with different interval is not simultaneous')
+const recovery=build(rows('recovery',[0,0,375_000_000],[0,30_000,60_000],[false,false,true]))
+assert.equal(mergeBuiltTraffic([a,recovery]).rxPeak,100)
+assert.equal(mergeBuiltTraffic([a,recovery]).observations!.rxPeak,true)
+assert.equal(mergeBuiltTraffic([a,build([])]).observations!.rxPeak,false)
+assert.equal(mergeBuiltTraffic([a,{...a}]).observations!.rxPeak,false,'serialized legacy parts have no raw interval proof')
+assert(!JSON.stringify(separate).includes('rxKnown'),'internal points are not DTO fields')
+console.log('aggregate peak: separate/coincident/nested/zero/missing/jitter/interval/recovery/empty/legacy/serialization passed')

backend/src/services/traffic-rate.ts

@@ -25,6 +25,7 @@ export interface RatePoint {
 }
 
 export interface TrafficObservations {rxNow:boolean;txNow:boolean;rxSeries:boolean[];txSeries:boolean[];sampledAt:string|null;expiresAt?:string|null;measurementIntervalMs?:number|null;
+  rxPeak?:boolean;txPeak?:boolean;
   currentState?: import("./traffic-freshness.js").TrafficCurrentState;lastAttemptAt?:string|null;lastSucceededAt?:string|null;error?:string|null}
 
 export interface BuiltTrafficSeries {
@@ -40,6 +41,14 @@ export interface BuiltTrafficSeries {
   txSeries: number[]
 }
 
+type PeakWave = { start: number; rx: number; tx: number; rxKnown: boolean; txKnown: boolean }
+// Raw intervals stay internal: DTO serialization must not expose or round these points.
+const peakWaves = new WeakMap<BuiltTrafficSeries, Map<number, PeakWave>>()
+function retainPeakWaves(built: BuiltTrafficSeries, waves: Map<number, PeakWave>): BuiltTrafficSeries {
+  peakWaves.set(built, waves)
+  return built
+}
+
 export interface MonitorLiveSample {
   rxMbps: number
   txMbps: number
@@ -129,7 +138,7 @@ export function buildTrafficFromSamples(
     txSeries: Array.from({ length: SERIES_POINTS }, () => 0),
   }
   const noObservations:TrafficObservations={rxNow:false,txNow:false,rxSeries:Array(SERIES_POINTS).fill(false),txSeries:Array(SERIES_POINTS).fill(false),sampledAt:null}
-  if (rows.length === 0) return includeObservations?{...empty,observations:noObservations}:empty
+  if (rows.length === 0) return retainPeakWaves(includeObservations?{...empty,observations:{...noObservations,rxPeak:false,txPeak:false}}:empty,new Map())
 
   const bySeries = new Map<string, TrafficSampleLike[]>()
   for (const r of rows) {
@@ -145,7 +154,7 @@ export function buildTrafficFromSamples(
 
   const rxPoints: Array<{ t: number; v: number }> = []
   const txPoints: Array<{ t: number; v: number }> = []
-  const byTs = new Map<number, { rx: number; tx: number;rxObserved:boolean;txObserved:boolean;count:number;intervalMs:number }>()
+  const byTs = new Map<number, { rx: number; tx: number;rxObserved:boolean;txObserved:boolean;count:number;intervalMs:number;start:number;sameStart:boolean }>()
   let includedSeries=0,lastAttemptAt:number|null=null
   const membership = new Map<number, {count:number; declared:boolean; active:number}>()
   let latestMembershipAt = -Infinity
@@ -205,7 +214,8 @@ export function buildTrafficFromSamples(
       if (rxBps == null && txBps == null) continue
       const rxMbps = bpsToMbps(rxBps ?? 0)
       const txMbps = bpsToMbps(txBps ?? 0)
-      const acc = byTs.get(t1) ?? { rx: 0, tx: 0,rxObserved:true,txObserved:true,count:0,intervalMs:0 }
+      const acc = byTs.get(t1) ?? { rx: 0, tx: 0,rxObserved:true,txObserved:true,count:0,intervalMs:0,start:t0,sameStart:true }
+      acc.sameStart &&= acc.start === t0
       acc.intervalMs = Math.max(acc.intervalMs, t1 - t0)
       acc.count++;acc.rxObserved&&=rxBps!==null;acc.txObserved&&=txBps!==null
       acc.rx += rxMbps
@@ -228,12 +238,18 @@ export function buildTrafficFromSamples(
   const last = lastTs != null ? byTs.get(lastTs) : undefined
   const peak = (points: Array<{t:number;v:number}>, side: "rxObserved"|"txObserved") => points.reduce((m,p) => {
     const wave = byTs.get(p.t)!
-    return p.t >= rangeStartMs && p.t <= rangeEndMs && wave[side] && wave.count === expectedAt(p.t) ? Math.max(m,p.v) : m
+    return p.t >= rangeStartMs && p.t <= rangeEndMs && wave.sameStart && wave[side] && wave.count === expectedAt(p.t) ? Math.max(m,p.v) : m
   },0)
   const rxPeak = peak(rxPoints,"rxObserved")
   const txPeak = peak(txPoints,"txObserved")
 
-  return {
+  const waves = new Map<number, PeakWave>()
+  for (const [t, wave] of byTs) {
+    if (t < rangeStartMs || t > rangeEndMs) continue
+    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})
+  }
+  return retainPeakWaves({
     rxNow: last?.rx ?? 0,
     txNow: last?.tx ?? 0,
     rxPeak,
@@ -250,8 +266,10 @@ export function buildTrafficFromSamples(
       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),
       sampledAt:lastAttemptAt===null?null:new Date(lastAttemptAt).toISOString(),
       measurementIntervalMs:lastTs===lastAttemptAt?last?.intervalMs??null:null,
+      rxPeak:[...waves.values()].some(w=>w.rxKnown),
+      txPeak:[...waves.values()].some(w=>w.txKnown),
     }}:{}),
-  }
+  },waves)
 }
 
 export function mergeBuiltTraffic(parts: BuiltTrafficSeries[]): BuiltTrafficSeries {
@@ -281,8 +299,6 @@ export function mergeBuiltTraffic(parts: BuiltTrafficSeries[]): BuiltTrafficSeri
   for (const p of parts) {
     acc.rxNow += p.rxNow
     acc.txNow += p.txNow
-    acc.rxPeak += p.rxPeak
-    acc.txPeak += p.txPeak
     acc.rxTotalGiB += p.rxTotalGiB
     acc.txTotalGiB += p.txTotalGiB
     acc.sessions += p.sessions
@@ -291,7 +307,22 @@ export function mergeBuiltTraffic(parts: BuiltTrafficSeries[]): BuiltTrafficSeri
       acc.txSeries[i] += p.txSeries[i] ?? 0
     }
   }
-  return parts.every(p => p.observations) ? { ...acc, observations: mergeTrafficObservations(parts.map(p => p.observations!)) } : acc
+  const waves = new Map<number, PeakWave>()
+  const first = peakWaves.get(parts[0])
+  for (const [t, wave] of first ?? []) {
+    const matching = parts.map(p => peakWaves.get(p)?.get(t))
+    if (!matching.every(w => w && w.start === wave.start)) continue
+    const rows = matching as PeakWave[]
+    const combined = {start:wave.start,rx:rows.reduce((n,w)=>n+w.rx,0),tx:rows.reduce((n,w)=>n+w.tx,0),
+      rxKnown:rows.every(w=>w.rxKnown),txKnown:rows.every(w=>w.txKnown)}
+    waves.set(t,combined)
+    if (combined.rxKnown) acc.rxPeak = Math.max(acc.rxPeak,combined.rx)
+    if (combined.txKnown) acc.txPeak = Math.max(acc.txPeak,combined.tx)
+  }
+  return retainPeakWaves(parts.every(p => p.observations) ? { ...acc, observations: {
+    ...mergeTrafficObservations(parts.map(p => p.observations!)),
+    rxPeak:[...waves.values()].some(w=>w.rxKnown),txPeak:[...waves.values()].some(w=>w.txKnown),
+  } } : acc,waves)
 }
 
 export function parseMonitorTraffic(

backend/src/services/traffic-volume-sql.test.ts

@@ -59,6 +59,19 @@ else {
     for(const[name,offset,bytes]of[["ether1",-90_000,0],["ether1",-60_000,0],["ether1",-30_000,1024**2],["new",-30_000,0],["ether1",0,1024**2],["new",0,0]] as const)await put(name,offset,bytes);
     await pool.query("UPDATE traffic_samples SET flags=61,native_counter_identity=interface_name WHERE server_id=$1",[serverId]);await pool.query("UPDATE traffic_samples SET flags=45 WHERE server_id=$1 AND interface_name='new' AND sampled_at=$2",[serverId,new Date(Date.parse(at)-30_000).toISOString()]);
     s=(await get(`/traffic/servers/${serverId}?range=1h`)).server;assert.equal(s.observations.rxNow,true);assert.equal(s.rxPeak,0);assert.equal(s.txPeak,0);assert.equal(s.rxTotal,1/1024);assert(s.observations.rxSeries.some(Boolean));cases++;
+    await pool.query("DELETE FROM user_interface_bindings WHERE user_id=$1",[userId])
+    for(const name of ['peakA','peakB'])await pool.query("INSERT INTO user_interface_bindings(id,user_id,server_id,interface_name,interface_type) VALUES($1,$2,$3,$4,'ether')",[prefix+name,userId,serverId,name])
+    for(const mode of ['separate','coincident','jitter']){
+      await pool.query("DELETE FROM traffic_samples WHERE server_id=$1",[serverId])
+      for(const [name,offset,bytes] of [['peakA',-60_000,0],['peakA',-30_000,375_000_000],['peakA',0,375_000_000],
+        ['peakB',-60_000,0],['peakB',-30_000,mode==='coincident'?375_000_000:0],['peakB',0,375_000_000]] as const)
+        await put(name,offset+(mode==='jitter'&&name==='peakB'?-1:0),bytes)
+      const aggregate=(await get('/traffic/users?range=1h')).users.find((u:{id:string})=>u.id===userId)
+      assert.equal(aggregate.rxPeak,mode==='jitter'?0:mode==='coincident'?200:100)
+      assert.equal(aggregate.observations.rxPeak,mode!=='jitter');assert.equal(aggregate.observations.txPeak,mode!=='jitter')
+      assert.equal(aggregate.rxTotal,750_000_000/1024**3);assert(aggregate.observations.rxSeries.some(Boolean))
+      assert.equal(aggregate.interfaces.length,2);assert(aggregate.interfaces.every((i:{rxPeak:number})=>i.rxPeak===100));cases++
+    }
     console.log(`traffic volume owned SQL: ${cases} cases; exact bytes/binary unit/server/user/binding/100-part aggregate/reset passed`)
   }finally{await app.close();await pool.query("DELETE FROM app_users WHERE id=$1",[userId]);if(serverId!==undefined){await pool.query("DELETE FROM traffic_samples WHERE server_id=$1",[serverId]);await pool.query("DELETE FROM servers WHERE id=$1",[serverId])}await closePool()}
 }