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

backend/src/db/schema.ts

@@ -341,6 +341,15 @@ export const uptimeSettings = pgTable("uptime_settings", {
   updatedAt:      ts("updated_at").notNull().defaultNow(),
 })
 
+export const trafficCollectionState = pgTable("traffic_collection_state", {
+  serverId: intPkRef().primaryKey().references(() => servers.id, { onDelete: "cascade" }),
+  serverUpdatedAt: ts("server_updated_at").notNull(),
+  lastAttemptAt: ts("last_attempt_at").notNull(),
+  lastSucceededAt: ts("last_succeeded_at"),
+  status: text("status").notNull(),
+  error: text("error"),
+})
+
 export const uptimeProbes = pgTable("uptime_probes", {
   id:          text("id").primaryKey(),
   srcServerId: bigint("src_server_id", { mode: "number" }).notNull()

backend/src/db/traffic-flags.test.ts

@@ -12,8 +12,10 @@ assert.equal(encodeTrafficFlags(true, true), 3)
 assert.equal(encodeTrafficFlags(false, false), 0)
 assert.equal(trafficFlagRunning(1), true)
 assert.equal(trafficFlagDisabled(1), false)
-assert.deepEqual(decodeTrafficSampleFlags(1), { running: true, disabled: false })
-assert.deepEqual(decodeTrafficSampleFlags(0), { running: false, disabled: false })
-assert.deepEqual(decodeTrafficSampleFlags(undefined), { running: false, disabled: false })
+assert.deepEqual(decodeTrafficSampleFlags(1), { running: true, disabled: false, countersKnown: false })
+assert.deepEqual(decodeTrafficSampleFlags(0), { running: false, disabled: false, countersKnown: false })
+assert.deepEqual(decodeTrafficSampleFlags(undefined), { running: false, disabled: false, countersKnown: false })
+assert.equal(encodeTrafficFlags(true,false,true),5)
+assert.deepEqual(decodeTrafficSampleFlags(5), {running:true,disabled:false,countersKnown:true})
 
 console.log("traffic-flags.test.ts: ok")

backend/src/db/traffic-flags.ts

@@ -1,8 +1,10 @@
 export const TRAFFIC_FLAG_RUNNING = 1
 export const TRAFFIC_FLAG_DISABLED = 2
+export const TRAFFIC_FLAG_COUNTERS_KNOWN = 4
 
-export function encodeTrafficFlags(running: boolean, disabled: boolean): number {
+export function encodeTrafficFlags(running: boolean, disabled: boolean, countersKnown = false): number {
   return (running ? TRAFFIC_FLAG_RUNNING : 0) | (disabled ? TRAFFIC_FLAG_DISABLED : 0)
+    | (countersKnown ? TRAFFIC_FLAG_COUNTERS_KNOWN : 0)
 }
 
 export function trafficFlagRunning(flags: number | null | undefined): boolean {
@@ -16,9 +18,11 @@ export function trafficFlagDisabled(flags: number | null | undefined): boolean {
 export function decodeTrafficSampleFlags(flags: number | null | undefined): {
   running: boolean
   disabled: boolean
+  countersKnown: boolean
 } {
   return {
     running: trafficFlagRunning(flags),
     disabled: trafficFlagDisabled(flags),
+    countersKnown: ((Number(flags) || 0) & TRAFFIC_FLAG_COUNTERS_KNOWN) !== 0,
   }
 }

backend/src/routes/traffic.ts

@@ -5,32 +5,28 @@ import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
 import { db } from "../db/index.js"
 import { serverSnapshots, servers } from "../db/schema.js"
 import {
-  collectTrafficOnce,
   getTrafficCollectorState,
   getTrafficSettings,
   readServerSamplesInRange,
   updateTrafficSettings,
 } from "../services/traffic-collector.js"
-import { scheduleAlertEngineAfterDataCollectors } from "../services/alert-collector-hooks.js"
-import { refreshScheduler } from "../services/scheduler.js"
+import { refreshScheduler, runSchedulerJobNow } from "../services/scheduler.js"
 import { appendEvent } from "../modules/events/service/events-service.js"
 import { MikrotikClient } from "../services/mikrotik.js"
 import { getEnabledServerById } from "../services/wireguard-live.js"
+import { readTrafficCollectionContext } from "../services/traffic-collection-state.js"
+import { applyTrafficFreshness } from "../services/traffic-freshness.js"
 import {
-  bpsToMbps,
   buildTrafficFromSamples,
   type TrafficObservations,
   isLoopbackName,
   parseMonitorTraffic,
-  rateBpsFromDelta,
 } from "../services/traffic-rate.js"
 import {
   buildBoundInterfaceTraffic,
   buildUserTrafficList,
 } from "../services/traffic-users.js"
 
-type SnapshotRow = typeof serverSnapshots.$inferSelect
-
 interface TrafficServerDto {
   observations?:TrafficObservations
   id: string
@@ -53,8 +49,8 @@ interface TrafficInterfaceDto {
   name: string
   running: boolean
   disabled: boolean
-  rxNow: number
-  txNow: number
+  rxNow: number | null
+  txNow: number | null
 }
 
 const LIVE_TICK_MS = 1500
@@ -96,6 +92,7 @@ function buildServerTraffic(
   rangeStartMs: number,
   rangeEndMs: number,
   onlyInterface?: string,
+  context?: Awaited<ReturnType<typeof readTrafficCollectionContext>>,
 ): TrafficServerDto {
   const built = buildTrafficFromSamples(rows, rangeStartMs, rangeEndMs, onlyInterface,undefined,true)
   return {
@@ -104,7 +101,7 @@ function buildServerTraffic(
     site: s.site || "—",
     country: s.country || "UN",
     status,
-    observations:built.observations,
+    observations:built.observations && applyTrafficFreshness(built.observations, s, context?.settings, context?.states.get(s.id), rangeEndMs),
     rxNow: built.rxNow,
     txNow: built.txNow,
     rxPeak: built.rxPeak,
@@ -122,23 +119,6 @@ async function snapshotStatus(serverId: number): Promise<TrafficServerDto["statu
   return snap?.status === "offline" ? "offline" : (snap?.status === "online" ? "online" : "degraded")
 }
 
-function ifaceNowMbps(
-  prev: { rxBytes: number; txBytes: number; sampledAt: string } | undefined,
-  last: { rxBytes: number; txBytes: number; sampledAt: string; rxBps: number; txBps: number },
-): { rxNow: number; txNow: number } {
-  if (!prev) {
-    return { rxNow: bpsToMbps(last.rxBps), txNow: bpsToMbps(last.txBps) }
-  }
-  const t0 = Date.parse(prev.sampledAt)
-  const t1 = Date.parse(last.sampledAt)
-  const rxBps = rateBpsFromDelta(prev.rxBytes, last.rxBytes, t0, t1)
-  const txBps = rateBpsFromDelta(prev.txBytes, last.txBytes, t0, t1)
-  return {
-    rxNow: bpsToMbps(rxBps ?? last.rxBps),
-    txNow: bpsToMbps(txBps ?? last.txBps),
-  }
-}
-
 function writeSse(raw: NodeJS.WritableStream, event: string, data: unknown) {
   raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
 }
@@ -267,8 +247,17 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
 
   app.post("/traffic/collect-now", async (_req, reply) => {
     try {
-      await collectTrafficOnce()
-      scheduleAlertEngineAfterDataCollectors()
+      const receipt = await runSchedulerJobNow("traffic")
+      if (!receipt.result || receipt.result.job !== "traffic") return reply.status(502).send({ ok: false, error: receipt.error ?? "Нет подтверждённого результата сбора" })
+      const result = receipt.result
+      if (result.skipped) return reply.status(409).send({ error: "Сбор трафика уже выполняется" })
+      const failures = result.servers.filter(s => !s.ok)
+      if (result.fatalError || failures.length) {
+        const error = result.fatalError ?? failures.map(s => `${s.name}: ${s.error ?? "Ошибка опроса"}`).join("; ")
+        await appendEvent({ level: "warning", eventType: "traffic.collect.manual.failed", sourceModule: "traffic",
+          title: "Сбор трафика завершён с ошибками", message: error })
+        return reply.status(502).send({ ok: false, error, failures, succeeded: result.servers.filter(s => s.ok).map(s => s.serverId) })
+      }
       const updated = await getTrafficSettings()
       await appendEvent({
         level: "info",
@@ -287,6 +276,8 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
         lastError: updated.lastError || null,
       })
     } catch (error) {
+      if (typeof error === "object" && error && "statusCode" in error && error.statusCode === 409)
+        return reply.status(409).send({ error: error instanceof Error ? error.message : "Сбор трафика уже выполняется" })
       const msg = error instanceof Error ? error.message : String(error)
       await appendEvent({
         level: "critical",
@@ -306,9 +297,10 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
     const rangeStartMs = rangeEndMs - minutes * 60_000
     const sinceIso = new Date(rangeStartMs).toISOString()
     const allServers = await db.select().from(servers).where(eq(servers.enabled, true))
+    const context = await readTrafficCollectionContext()
     const data = await Promise.all(allServers.map(async (s): Promise<TrafficServerDto> => {
       const rows = await readServerSamplesInRange(s.id, sinceIso)
-      return buildServerTraffic(s, await snapshotStatus(s.id), rows, rangeStartMs, rangeEndMs)
+      return buildServerTraffic(s, await snapshotStatus(s.id), rows, rangeStartMs, rangeEndMs, undefined, context)
     }))
 
     return reply.send({ servers: data })
@@ -385,9 +377,10 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
 
     const sinceIso = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString()
     const rows = await readServerSamplesInRange(serverId, sinceIso)
+    const context = await readTrafficCollectionContext(), end = Date.now()
     const byIface = new Map<string, typeof rows>()
     for (const r of rows) {
-      if (isLoopbackName(r.interfaceName)) continue
+      if (isLoopbackName(r.interfaceName) || r.peerPublicKey) continue
       const arr = byIface.get(r.interfaceName) ?? []
       arr.push(r)
       byIface.set(r.interfaceName, arr)
@@ -395,19 +388,19 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
     const interfaces: TrafficInterfaceDto[] = [...byIface.entries()].map(([name, arr]) => {
       const sorted = arr.sort((a, b) => a.sampledAt.localeCompare(b.sampledAt))
       const last = sorted[sorted.length - 1]
-      const prev = sorted[sorted.length - 2]
       if (!last) {
         return { name, running: false, disabled: true, rxNow: 0, txNow: 0 }
       }
-      const now = ifaceNowMbps(prev, last)
+      const built = buildTrafficFromSamples(sorted, Date.parse(sinceIso), end, name, undefined, true)
+      const observed = applyTrafficFreshness(built.observations!, server, context.settings, context.states.get(server.id), end)
       return {
         name,
         running: Boolean(last.running),
         disabled: Boolean(last.disabled),
-        rxNow: now.rxNow,
-        txNow: now.txNow,
+        rxNow: observed.rxNow ? built.rxNow : null,
+        txNow: observed.txNow ? built.txNow : null,
       }
-    }).sort((a, b) => (b.rxNow + b.txNow) - (a.rxNow + a.txNow))
+    }).sort((a, b) => ((b.rxNow ?? -1) + (b.txNow ?? -1)) - ((a.rxNow ?? -1) + (a.txNow ?? -1)))
 
     return reply.send({ interfaces })
   })
@@ -432,6 +425,7 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
       rangeStartMs,
       rangeEndMs,
       q.iface && q.iface !== "__all__" ? q.iface : undefined,
+      await readTrafficCollectionContext(),
     )
     return reply.send({ server: data })
   })

backend/src/services/alert-engine/signals-from-collectors.ts

@@ -4,6 +4,8 @@ import { collectGreSignalsFromRuns } from "./gre-signals.js"
 import { readGreBgpRunSnapshots } from "../gre-bgp-snapshot-history.js"
 import { collectSourceSignals, probePeriodMs, type CollectorRunRow, type CollectorSourceConfig } from "./collector-source-signals.js"
 import type { SignalSnapshot } from "./types.js"
+import type { TrafficAttempt } from "../traffic-freshness.js"
+import { applyTrafficFreshness } from "../traffic-freshness.js"
 
 export type CollectorRead = <T extends object>(query: string, params?: unknown[]) => Promise<T[]>
 export type CollectorSourceSnapshot = Pick<SignalSnapshot, "sampledAt" | "servers" | "probes" | "trafficByServer">
@@ -18,7 +20,7 @@ async function readRuns(read: CollectorRead, job: string, limit: number, since?:
 /** Read-only: do not seed defaults or load credentials while building alert facts. */
 export async function readCollectorSourceSnapshot(clock: () => number = Date.now, read: CollectorRead = dbAll): Promise<CollectorSourceSnapshot> {
   const now = clock()
-  const [servers, probes, uptime, rest, traffic] = await Promise.all([
+  const [servers, probes, uptime, rest, traffic, attempts] = await Promise.all([
     read<CollectorSourceConfig["servers"][number]>("SELECT id, name, host, enabled, updated_at AS \"updatedAt\" FROM servers"),
     read<CollectorSourceConfig["probes"][number]>(`SELECT id, name, target, src_server_id AS "srcServerId", src_interface AS "srcInterface",
       enabled, interval_sec AS "intervalSec", updated_at AS "updatedAt" FROM uptime_probes`),
@@ -26,6 +28,8 @@ export async function readCollectorSourceSnapshot(clock: () => number = Date.now
       interval_sec AS "intervalSec", probe_interval_sec AS "probeIntervalSec" FROM uptime_settings WHERE id = 1`),
     read<NonNullable<CollectorSourceConfig["rest"]>>('SELECT enabled, interval_sec AS "intervalSec" FROM servers_api_ping_settings WHERE id = 1'),
     read<NonNullable<CollectorSourceConfig["traffic"]>>('SELECT enabled, interval_sec AS "intervalSec" FROM traffic_settings WHERE id = 1'),
+    read<TrafficAttempt & {serverId:number}>(`SELECT server_id AS "serverId",server_updated_at AS "serverUpdatedAt",
+      last_attempt_at AS "lastAttemptAt",last_succeeded_at AS "lastSucceededAt",status,error FROM traffic_collection_state`),
   ])
   const config: CollectorSourceConfig = { servers, probes, uptime: uptime[0], rest: rest[0], traffic: traffic[0] }
   const periods = probes.filter(p => p.enabled).map(p => probePeriodMs(config, p)).filter(Number.isFinite)
@@ -36,9 +40,17 @@ export async function readCollectorSourceSnapshot(clock: () => number = Date.now
     readRuns(read, "uptime_ping", 4096, new Date(now - probeWindowMs).toISOString()),
     readRuns(read, "traffic", 8),
   ])
+  const facts = collectSourceSignals(config, { resources: resourceRuns, rest: restRuns, ping: pingRuns, traffic: trafficRuns }, now)
+  facts.trafficByServer = facts.trafficByServer.filter(signal => {
+    const server = servers.find(s => (s.name || s.host) === signal.serverName)
+    const attempt = server && attempts.find(a => a.serverId === Number(server.id))
+    if (!attempt) return true // Existing scheduler provenance remains readable until this device's first new attempt.
+    return !!server && applyTrafficFreshness({ rxNow: true, txNow: true, rxSeries: [], txSeries: [], sampledAt: signal.sampledAt },
+      { enabled: server.enabled === true, updatedAt: server.updatedAt ?? "" }, traffic[0], attempt, now).currentState === "current"
+  })
   return {
     sampledAt: new Date(now).toISOString(),
-    ...collectSourceSignals(config, { resources: resourceRuns, rest: restRuns, ping: pingRuns, traffic: trafficRuns }, now),
+    ...facts,
   }
 }
 

backend/src/services/scheduler-recorded-run.test.ts

@@ -55,9 +55,15 @@ test("partial backups, renew and alert errors are explicit while confirmed resul
   assert.match(schedulerSnapshotError({ v: 1, job: "alert_engine", sampledAt: "now", rulesChecked: 1, standaloneFires: 0, groupFires: 0, skippedNoTelegram: false, errors: ["send failed"] })!, /send failed/)
 })
 test("valid offline/down observations are not collector errors", () => {
-  assert.equal(schedulerSnapshotError({ v: 1, job: "traffic", sampledAt: "now", servers: [{ serverId: 1, name: "router", host: "host", ok: false, error: "offline" }] }), null)
+  assert.equal(schedulerSnapshotError({ v: 1, job: "traffic", sampledAt: "now", servers: [{ serverId: 1, name: "router", host: "host", ok: true, sumRxMbps: 0, sumTxMbps: 0 }] }), null)
   assert.equal(schedulerSnapshotError({ v: 1, job: "uptime_ping", sampledAt: "now", probes: [{ probeId: "probe", name: "probe", target: "host", srcServerId: 1, srcServerName: "router", srcInterface: "ether1", ok: false, rttMs: null, lossPct: 100, status: "down" }] }), null)
 })
+test("traffic command errors are limitations; partial measurements remain in the receipt",async()=>{
+  const snapshot={v:1 as const,job:"traffic" as const,sampledAt:"now",servers:[{serverId:1,name:"router",host:"host",ok:false,error:"401"},{serverId:2,name:"second",host:"second",ok:true,sumRxMbps:0,sumTxMbps:0}]}
+  const h=harness(snapshot as never),receipt=await runRecordedSchedulerJob(h.deps)
+  assert.equal(receipt.status,"error");assert.match(receipt.error!,/router: 401/)
+  assert.equal(receipt.result,snapshot);assert.equal(h.records.length,1);assert.equal(h.events[0].level,"warning")
+})
 test("skipped/not-due snapshots preserve their reason and do not announce completed work", async () => {
   for (const snapshot of [backup({ due: false }), backup({ skipped: true })]) {
     const h = harness(snapshot); const receipt = await runRecordedSchedulerJob(h.deps)

backend/src/services/scheduler-recorded-run.ts

@@ -20,6 +20,9 @@ export function schedulerSnapshotError(snapshot: SchedulerRunSnapshot | null): s
   if (!snapshot) return null
   if ("fatalError" in snapshot && snapshot.fatalError) return snapshot.fatalError
   const errors = "errors" in snapshot ? [...(snapshot.errors ?? [])] : []
+  if (snapshot.job === "traffic") for (const server of snapshot.servers) {
+    if (!server.ok) errors.push(`${server.name}: ${server.error ?? "Ошибка опроса трафика"}`)
+  }
   if (snapshot.job === "uptime_speed") {
     for (const run of snapshot.runs) if (!run.ok && run.error) errors.push(run.error)
   }

backend/src/services/traffic-collection-state.ts

@@ -0,0 +1,27 @@
+import { and, eq, lte } from "drizzle-orm"
+import { db } from "../db/index.js"
+import { trafficCollectionState, trafficSettings, type Server } from "../db/schema.js"
+
+export async function recordTrafficCollectionAttempt(server: Pick<Server, "id" | "updatedAt">,
+  status: "pending" | "ok" | "error", sampledAt: string, error?: string) {
+  if (status === "pending") {
+    const next = { serverId: server.id, serverUpdatedAt: server.updatedAt,
+      lastAttemptAt: sampledAt, status, error: null }
+    await db.insert(trafficCollectionState).values(next).onConflictDoUpdate({
+      target: trafficCollectionState.serverId, set: next,
+      setWhere: lte(trafficCollectionState.lastAttemptAt, sampledAt),
+    })
+  } else {
+    await db.update(trafficCollectionState).set({ status, error: error?.slice(0, 1024) ?? null,
+      ...(status === "ok" ? { lastSucceededAt: sampledAt } : {}) }).where(and(
+      eq(trafficCollectionState.serverId, server.id), eq(trafficCollectionState.lastAttemptAt, sampledAt),
+    ))
+  }
+}
+
+/** Read-only: does not create default settings or contact a device. */
+export async function readTrafficCollectionContext() {
+  const [states, settings] = await Promise.all([db.select().from(trafficCollectionState),
+    db.select().from(trafficSettings).where(eq(trafficSettings.id, 1)).limit(1)])
+  return { states: new Map(states.map(s => [s.serverId, s])), settings: settings[0] }
+}

backend/src/services/traffic-collector.ts

@@ -8,6 +8,7 @@ import { MikrotikClient } from "./mikrotik.js"
 import { bpsToMbps, shouldIncludeIface } from "./traffic-rate.js"
 import { observedTrafficRate, parseTrafficCounter } from "./traffic-observation.js"
 import { rememberServerIfaces } from "./traffic-flow-ifindex.js"
+import { recordTrafficCollectionAttempt } from "./traffic-collection-state.js"
 
 interface RosIfaceTraffic {
   ".id"?: string
@@ -86,6 +87,7 @@ async function readPreviousWave(serverId: number): Promise<Map<string, { rxBytes
       peerPublicKey: trafficSamples.peerPublicKey,
       rxBytes: trafficSamples.rxBytes,
       txBytes: trafficSamples.txBytes,
+      flags: trafficSamples.flags,
       sampledAt: trafficSamples.sampledAt,
     })
     .from(trafficSamples)
@@ -93,7 +95,8 @@ async function readPreviousWave(serverId: number): Promise<Map<string, { rxBytes
       eq(trafficSamples.serverId, serverId),
       eq(trafficSamples.sampledAt, last.sampledAt),
     ))
-  return new Map(rows.map((r) => [`${r.interfaceName}\0${r.peerPublicKey ?? ""}`, r]))
+  return new Map(rows.filter(r => decodeTrafficSampleFlags(r.flags).countersKnown)
+    .map((r) => [`${r.interfaceName}\0${r.peerPublicKey ?? ""}`, r]))
 }
 
 export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
@@ -110,7 +113,6 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
   collecting = true
   const startedAt = Date.now()
   const now = sampledAt
-  const settings = await getSettingsRow()
   const snapshot: TrafficRunSnapshot = {
     v: SCHEDULER_RUN_SNAPSHOT_VERSION,
     job: "traffic",
@@ -119,11 +121,13 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
   }
 
   try {
+    await getSettingsRow()
     const enabledServers = await db.select().from(servers).where(eq(servers.enabled, true))
 
     for (const srv of enabledServers) {
       const name = srv.name || srv.host
       try {
+        await recordTrafficCollectionAttempt(srv, "pending", now)
         const client = MikrotikClient.fromServer(srv)
         const ifaces = await client.get<RosIfaceTraffic[]>("/interface")
         rememberServerIfaces(srv.id, ifaces)
@@ -155,7 +159,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
             txBytes,
             rxBps,
             txBps,
-            flags: encodeTrafficFlags(running, disabled),
+            flags: encodeTrafficFlags(running, disabled, true),
           }
         })
         try {
@@ -184,7 +188,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
               txBytes,
               rxBps,
               txBps,
-              flags: encodeTrafficFlags(running, disabled),
+              flags: encodeTrafficFlags(running, disabled, true),
             })
           }
         } catch {
@@ -193,6 +197,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
         if (rows.length > 0) {
           await db.insert(trafficSamples).values(rows)
         }
+        await recordTrafficCollectionAttempt(srv, "ok", now)
         snapshot.servers.push({
           serverId: srv.id,
           name,
@@ -205,6 +210,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
           } : {}),
         })
       } catch (err) {
+        await recordTrafficCollectionAttempt(srv, "error", now, err instanceof Error ? err.message : String(err))
         snapshot.servers.push({
           serverId: srv.id,
           name,
@@ -218,7 +224,7 @@ export async function collectTrafficOnce(): Promise<TrafficRunSnapshot> {
     await db.update(trafficSettings).set({
       lastCollectedAt: now,
       lastDurationMs: Date.now() - startedAt,
-      lastError: "",
+      lastError: snapshot.servers.filter(s => !s.ok).map(s => `${s.name}: ${s.error ?? "Ошибка опроса"}`).join("; "),
       updatedAt: now,
     }).where(eq(trafficSettings.id, 1))
   } catch (e) {

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

@@ -0,0 +1,21 @@
+import assert from "node:assert/strict"
+import { applyTrafficFreshness, mergeTrafficObservations } from "./traffic-freshness.js"
+const now = Date.parse("2026-10-12T12:00:00Z"), iso = (offset=0) => new Date(now+offset).toISOString()
+const observed = {rxNow:true,txNow:true,rxSeries:[false,true],txSeries:[false,true],sampledAt:iso()}
+const server={enabled:true,updatedAt:iso(-3600_000)},settings={enabled:true,intervalSec:30}
+const attempt={serverUpdatedAt:server.updatedAt,lastAttemptAt:iso(),lastSucceededAt:iso(),status:"ok",error:null}
+const check=(s=server,c=settings,a:typeof attempt|undefined=attempt,o=observed,t=now)=>applyTrafficFreshness(o,s,c,a,t)
+assert.equal(check().currentState,"current");assert.equal(check().rxNow,true)
+for(const status of ["error","pending"]) {const r=check(server,settings,{...attempt,status});assert.equal(r.currentState,status);assert.equal(r.rxNow,false);assert.deepEqual(r.rxSeries,observed.rxSeries)}
+assert.equal(applyTrafficFreshness(observed,server,settings,undefined,now).currentState,"unknown")
+assert.equal(check({...server,updatedAt:iso(1)}).currentState,"unknown")
+assert.equal(check({...server,enabled:false}).currentState,"disabled")
+assert.equal(check(server,{...settings,enabled:false}).currentState,"disabled")
+assert.equal(check(server,settings,attempt,observed,now+90_000).currentState,"current")
+assert.equal(check(server,settings,attempt,observed,now+90_001).currentState,"stale")
+assert.equal(check(server,settings,{...attempt,lastAttemptAt:iso(30_001)}).currentState,"stale")
+assert.equal(check(server,settings,attempt,{...observed,sampledAt:iso(-1000)}).currentState,"unknown")
+const mixed=mergeTrafficObservations([check(),check(server,settings,{...attempt,status:"error",error:"401"} as typeof attempt)])
+assert.equal(mixed.rxNow,false);assert.equal(mixed.currentState,"error");assert.deepEqual(mixed.rxSeries,[false,true])
+assert.equal(mergeTrafficObservations([]).rxNow,false)
+console.log("traffic freshness/source barrier/merge tests ok")

backend/src/services/traffic-freshness.ts

@@ -0,0 +1,45 @@
+import type { TrafficObservations } from "./traffic-rate.js"
+
+export type TrafficCurrentState = "current" | "unknown" | "pending" | "error" | "stale" | "disabled"
+export interface TrafficAttempt {
+  serverUpdatedAt: string | Date; lastAttemptAt: string | Date; lastSucceededAt: string | Date | null; status: string; error: string | null
+}
+export function mergeTrafficObservations(parts: TrafficObservations[]): TrafficObservations {
+  const oldest = (key: "sampledAt" | "lastAttemptAt" | "lastSucceededAt" | "expiresAt") => {
+    const times = parts.map(p => Date.parse(p[key] ?? ""))
+    return times.length && times.every(Number.isFinite) ? new Date(Math.min(...times)).toISOString() : null
+  }
+  const states: TrafficCurrentState[] = ["error", "pending", "stale", "disabled", "unknown"]
+  return { rxNow: !!parts.length && parts.every(p => p.rxNow), txNow: !!parts.length && parts.every(p => p.txNow),
+    rxSeries: Array.from({ length: parts[0]?.rxSeries.length ?? 60 }, (_, i) => !!parts.length && parts.every(p => p.rxSeries[i])),
+    txSeries: Array.from({ length: parts[0]?.txSeries.length ?? 60 }, (_, i) => !!parts.length && parts.every(p => p.txSeries[i])),
+    sampledAt: oldest("sampledAt"), expiresAt: oldest("expiresAt"), lastAttemptAt: oldest("lastAttemptAt"), lastSucceededAt: oldest("lastSucceededAt"),
+    currentState: parts.length ? states.find(state => parts.some(p => p.currentState === state)) ?? "current" : "unknown",
+    error: parts.find(p => p.error)?.error ?? null,
+  }
+}
+export function applyTrafficFreshness(observations: TrafficObservations,
+  server: { enabled: boolean; updatedAt: string | Date },
+  settings: { enabled: boolean; intervalSec: number } | undefined,
+  attempt: TrafficAttempt | undefined, now: number): TrafficObservations {
+  const period = Math.max(5, settings?.intervalSec ?? 30) * 1000
+  const time = (raw: string | Date | null | undefined) => raw instanceof Date ? raw.getTime() : Date.parse(raw ?? "")
+  const attemptMs = time(attempt?.lastAttemptAt), successMs = time(attempt?.lastSucceededAt)
+  let currentState: TrafficCurrentState = "unknown"
+  if (!server.enabled || settings?.enabled === false) currentState = "disabled"
+  else if (settings && attempt && time(attempt.serverUpdatedAt) === time(server.updatedAt)) {
+    if (!Number.isFinite(attemptMs) || attemptMs > now + period || now - attemptMs > 3 * period) currentState = "stale"
+    else if (attempt.status === "pending") currentState = "pending"
+    else if (attempt.status === "error") currentState = "error"
+    else if (attempt.status === "ok" && successMs === attemptMs &&
+      Date.parse(observations.sampledAt ?? "") === successMs && (observations.rxNow || observations.txNow)) currentState = "current"
+  }
+  return { ...observations,
+    expiresAt: Number.isFinite(attemptMs + 3 * period) && Math.abs(attemptMs + 3 * period) <= 8.64e15 ? new Date(attemptMs + 3 * period).toISOString() : null,
+    rxNow: currentState === "current" && observations.rxNow,
+    txNow: currentState === "current" && observations.txNow,
+    currentState, lastAttemptAt: Number.isFinite(attemptMs) ? new Date(attemptMs).toISOString() : null,
+    lastSucceededAt: Number.isFinite(successMs) ? new Date(successMs).toISOString() : null,
+    error: attempt?.error ?? null,
+  }
+}

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

@@ -0,0 +1,61 @@
+import assert from "node:assert/strict"
+import { createServer } from "node:http"
+import Fastify from "fastify"
+import { pool, closePool } from "../db/index.js"
+import { initDatabase } from "../db/bootstrap.js"
+import { collectTrafficOnce } from "./traffic-collector.js"
+import { recordTrafficCollectionAttempt } from "./traffic-collection-state.js"
+import { buildSignalSnapshotFromCollectors } from "./alert-engine/signals-from-collectors.js"
+import trafficRoutes from "../routes/traffic.js"
+import { clearAlertCollectorHooksTimer } from "./alert-collector-hooks.js"
+
+const database=process.env.MMAPP_OWN_SQL_DB
+if(!database) console.log("traffic history SQL: skipped; explicit owned database required")
+else {
+  assert(/^mm_accept_166_\d+$/.test(database));assert.equal(new URL(process.env.DATABASE_URL!).pathname,"/"+database)
+  assert.equal((await pool.query("SELECT current_database() AS name")).rows[0].name,database)
+  const prefix="own-history178-"+Date.now()+"-",userId=prefix+"user"
+  let serverId:number|undefined,deny=false,counter="1000",cases=0,requests=0
+  const http=createServer((req,res)=>{requests++;res.setHeader("Content-Type","application/json");if(deny){res.statusCode=401;res.end('{"message":"Unauthorized"}')}else res.end(JSON.stringify(req.url?.includes("/wireguard/")?[]:[{name:"ether1",running:"true",disabled:"false","rx-byte":counter,"tx-byte":counter}]))})
+  await new Promise<void>(resolve=>http.listen(0,"127.0.0.1",resolve));const address=http.address();assert(address&&typeof address==="object")
+  const app=Fastify();await app.register(trafficRoutes)
+  const get=async(path:string)=>{const r=await app.inject({method:"GET",url:path});assert.equal(r.statusCode,200);return r.json()}
+  try {
+    await initDatabase()
+    const updatedAt=new Date(Date.now()-3_600_000).toISOString()
+    serverId=Number((await pool.query("INSERT INTO servers(name,host,port,use_ssl,enabled,username,password,updated_at) VALUES($1,'127.0.0.1',$2,false,true,'fixture','fixture',$3) RETURNING id",[prefix+"server",address.port,updatedAt])).rows[0].id)
+    await pool.query("INSERT INTO traffic_settings(id,enabled,interval_sec) VALUES(1,true,30) ON CONFLICT(id) DO UPDATE SET enabled=true,interval_sec=30")
+    await pool.query("INSERT INTO app_users(id,login,name) VALUES($1,$2,$3)",[userId,prefix+"login","History fixture"])
+    await pool.query("INSERT INTO user_interface_bindings(id,user_id,server_id,interface_name,interface_type) VALUES($1,$2,$3,'ether1','ether')",[prefix+"binding",userId,serverId])
+    for(const offset of[-60_000,-30_000])await pool.query("INSERT INTO traffic_samples(server_id,interface_name,sampled_at,rx_bytes,tx_bytes,flags) VALUES($1,'ether1',$2,1000,1000,1)",[serverId,new Date(Date.now()+offset).toISOString()])
+    const detail=async()=> (await get(`/traffic/servers/${serverId}?range=1h`)).server
+    let dto=await detail();assert.equal(dto.observations.rxNow,false);assert(dto.observations.rxSeries.every((v:boolean)=>!v));cases++
+    await collectTrafficOnce();dto=await detail();assert.equal(dto.observations.rxNow,false);cases++
+    await pool.query("DELETE FROM traffic_samples WHERE server_id=$1 AND (flags & 4)=0",[serverId])
+    await pool.query("UPDATE traffic_samples SET sampled_at=$1 WHERE server_id=$2",[new Date(Date.now()-30_000).toISOString(),serverId])
+    counter="3751000";const success=await collectTrafficOnce();dto=await detail();assert.equal(dto.observations.currentState,"current");assert.equal(dto.observations.rxNow,true);assert(dto.rxNow>0.9);cases++
+    await pool.query("INSERT INTO scheduler_runs(id,job_key,started_at,finished_at,status,result_json) VALUES($1,'traffic',$2,$3,'ok',$4::jsonb)",[prefix+"run",success.sampledAt,new Date().toISOString(),JSON.stringify(success)])
+    assert.equal((await buildSignalSnapshotFromCollectors()).trafficByServer.length,1);cases++
+    const succeeded=dto.observations.lastSucceededAt
+    deny=true;await collectTrafficOnce();dto=await detail();assert.equal(dto.observations.currentState,"error");assert.equal(dto.observations.rxNow,false);assert.equal(dto.observations.lastSucceededAt,succeeded);assert(dto.observations.rxSeries.some(Boolean));cases++
+    assert.deepEqual((await buildSignalSnapshotFromCollectors()).trafficByServer,[]);cases++
+    const users=(await get("/traffic/users?range=1h")).users, bound=(await get("/traffic/bound-interfaces?range=1h")).interfaces
+    assert.equal(users.find((u:{id:string})=>u.id===userId).observations.currentState,"error");assert.equal(bound.find((b:{userId:string})=>b.userId===userId).observations.rxNow,false);cases++
+    const interfaces=(await get(`/traffic/servers/${serverId}/interfaces`)).interfaces;assert.equal(interfaces[0].rxNow,null);cases++
+    const manual=await app.inject({method:"POST",url:"/traffic/collect-now"});assert.equal(manual.statusCode,502);assert.equal(manual.json().ok,false)
+    assert((await pool.query("SELECT status FROM scheduler_runs WHERE job_key='traffic' ORDER BY finished_at DESC LIMIT 1")).rows[0].status==="error");cases++
+    deny=false;counter="3751000";const recovery=await app.inject({method:"POST",url:"/traffic/collect-now"});assert.equal(recovery.statusCode,200);dto=await detail();assert.equal(dto.observations.currentState,"current");assert.equal(dto.rxNow,0);cases++
+    await pool.query("UPDATE traffic_collection_state SET last_attempt_at=$1,last_succeeded_at=$1 WHERE server_id=$2",[new Date(Date.now()-100_000).toISOString(),serverId]);dto=await detail();assert.equal(dto.observations.currentState,"stale");assert.equal(dto.observations.rxNow,false);assert(dto.observations.rxSeries.some(Boolean));cases++
+    await recordTrafficCollectionAttempt({id:serverId,updatedAt},"pending",new Date().toISOString());dto=await detail();assert.equal(dto.observations.currentState,"pending");cases++
+    await pool.query("UPDATE servers SET updated_at=$1 WHERE id=$2",[new Date(Date.now()+1).toISOString(),serverId]);dto=await detail();assert.equal(dto.observations.currentState,"unknown");cases++
+    await pool.query("UPDATE servers SET enabled=false WHERE id=$1",[serverId]);const before=requests;await collectTrafficOnce();assert.equal(requests,before);dto=await detail();assert.equal(dto.observations.currentState,"disabled");cases++
+    console.log(`traffic history owned SQL: ${cases} cases; legacy/current/failure/history/users/interfaces/manual/source/stale/pending/config/disabled passed`)
+  } finally {
+    clearAlertCollectorHooksTimer()
+    await app.close();await new Promise<void>(resolve=>http.close(()=>resolve()))
+    await pool.query("DELETE FROM app_users WHERE id=$1",[userId])
+    await pool.query("DELETE FROM scheduler_runs WHERE id LIKE $1",[prefix+"%"])
+    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()
+  }
+}

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

@@ -45,6 +45,7 @@ const context = vm.createContext({
   SCHEDULER_RUN_SNAPSHOT_VERSION: 1, servers: {}, trafficSamples: {}, trafficSettings: {}, eq: () => ({}),
   getSettingsRow: async () => ({}), readPreviousWave: async () => wave, rememberServerIfaces: () => {},
   encodeTrafficFlags: () => 1,
+  recordTrafficCollectionAttempt: async () => {},
   MikrotikClient: { fromServer: () => ({ get: async (path: string) => path === "/interface" ?
     [{ name: "ether1", running: String(included), disabled: "false", "rx-byte": counters, "tx-byte": counters }] : [] }) },
   db: {

backend/src/services/traffic-observation.ts

@@ -15,6 +15,9 @@ export function observedTrafficRate(
   rxBytes: number, txBytes: number, nowMs: number,
 ): { rxBps: number; txBps: number; measured: boolean } {
   const prevMs = previous ? Date.parse(previous.sampledAt) : NaN
+  if (!previous || ![previous.rxBytes, previous.txBytes, rxBytes, txBytes].every(v => Number.isSafeInteger(v) && v >= 0)) {
+    return { rxBps: 0, txBps: 0, measured: false }
+  }
   const rx = previous ? rateBpsFromDelta(previous.rxBytes, rxBytes, prevMs, nowMs) : null
   const tx = previous ? rateBpsFromDelta(previous.txBytes, txBytes, prevMs, nowMs) : null
   // Stored rates retain the legacy numeric shape; observed totals require both deltas.

backend/src/services/traffic-rate.ts

@@ -1,4 +1,5 @@
 /** Чистые формулы трафика: дельты счётчиков, корзины series, monitor-traffic. */
+import { mergeTrafficObservations } from "./traffic-freshness.js"
 
 export const SERIES_POINTS = 60
 
@@ -12,6 +13,7 @@ export interface TrafficSampleLike {
   txBps: number
   running: boolean
   disabled: boolean
+  countersKnown?: boolean
 }
 
 export interface RatePoint {
@@ -20,7 +22,8 @@ export interface RatePoint {
   txMbps: number
 }
 
-export interface TrafficObservations {rxNow:boolean;txNow:boolean;rxSeries:boolean[];txSeries:boolean[];sampledAt:string|null}
+export interface TrafficObservations {rxNow:boolean;txNow:boolean;rxSeries:boolean[];txSeries:boolean[];sampledAt:string|null;expiresAt?:string|null;
+  currentState?: import("./traffic-freshness.js").TrafficCurrentState;lastAttemptAt?:string|null;lastSucceededAt?:string|null;error?:string|null}
 
 export interface BuiltTrafficSeries {
   observations?:TrafficObservations
@@ -161,7 +164,7 @@ export function buildTrafficFromSamples(
       continue
     }
 
-    const sorted = [...arr].sort((a, b) => a.sampledAt.localeCompare(b.sampledAt))
+    const sorted = [...arr].sort((a, b) => Date.parse(a.sampledAt) - Date.parse(b.sampledAt))
     const last = sorted[sorted.length - 1]
     if (!last) continue
     if (!allowList && (!last.running || last.disabled)) continue
@@ -169,14 +172,6 @@ export function buildTrafficFromSamples(
     includedSeries++
     if (last.running && !last.disabled) sessions += 1
 
-    const first = sorted[0]
-    if (first) {
-      const dRx = last.rxBytes - first.rxBytes
-      const dTx = last.txBytes - first.txBytes
-      rxBytesDelta += dRx >= 0 ? dRx : last.rxBytes
-      txBytesDelta += dTx >= 0 ? dTx : last.txBytes
-    }
-
     for (let i = 1; i < sorted.length; i++) {
       const prev = sorted[i - 1]
       const cur = sorted[i]
@@ -185,8 +180,13 @@ export function buildTrafficFromSamples(
       const t0 = parseIsoMs(prev.sampledAt)
       const t1 = parseIsoMs(cur.sampledAt)
       if(Number.isFinite(Date.parse(cur.sampledAt))&&t1>=rangeStartMs&&t1<=rangeEndMs)lastAttemptAt=Math.max(lastAttemptAt??t1,t1)
-      const rxBps = rateBpsFromDelta(prev.rxBytes, cur.rxBytes, t0, t1)
-      const txBps = rateBpsFromDelta(prev.txBytes, cur.txBytes, t0, t1)
+      const countersKnown = prev.countersKnown !== false && cur.countersKnown !== false
+      const rxBps = countersKnown ? rateBpsFromDelta(prev.rxBytes, cur.rxBytes, t0, t1) : null
+      const txBps = countersKnown ? rateBpsFromDelta(prev.txBytes, cur.txBytes, t0, t1) : null
+      if (t1 >= rangeStartMs && t1 <= rangeEndMs) {
+        if (rxBps !== null) rxBytesDelta += cur.rxBytes - prev.rxBytes
+        if (txBps !== null) txBytesDelta += cur.txBytes - prev.txBytes
+      }
       if (rxBps == null && txBps == null) continue
       const rxMbps = bpsToMbps(rxBps ?? 0)
       const txMbps = bpsToMbps(txBps ?? 0)
@@ -269,7 +269,7 @@ export function mergeBuiltTraffic(parts: BuiltTrafficSeries[]): BuiltTrafficSeri
   }
   acc.rxTotalGiB = Number(acc.rxTotalGiB.toFixed(1))
   acc.txTotalGiB = Number(acc.txTotalGiB.toFixed(1))
-  return acc
+  return parts.every(p => p.observations) ? { ...acc, observations: mergeTrafficObservations(parts.map(p => p.observations!)) } : acc
 }
 
 export function parseMonitorTraffic(

backend/src/services/traffic-users.ts

@@ -8,8 +8,12 @@ import {
   mergeBuiltTraffic,
   type BuiltTrafficSeries,
 } from "./traffic-rate.js"
+import { readTrafficCollectionContext } from "./traffic-collection-state.js"
+import { applyTrafficFreshness } from "./traffic-freshness.js"
+import type { TrafficObservations } from "./traffic-rate.js"
 
 export interface BoundIfaceTrafficDto {
+  observations?: TrafficObservations
   id: string
   bindingId: string
   userId: string
@@ -36,6 +40,7 @@ export interface BoundIfaceTrafficDto {
 }
 
 export interface UserTrafficDto {
+  observations?: TrafficObservations
   id: string
   login: string
   displayName: string
@@ -54,6 +59,7 @@ export interface UserTrafficDto {
 
 function seriesFromBuilt(built: BuiltTrafficSeries) {
   return {
+    observations: built.observations,
     rxNow: built.rxNow,
     txNow: built.txNow,
     rxPeak: built.rxPeak,
@@ -74,6 +80,8 @@ async function serverStatus(serverId: number): Promise<"online" | "offline"> {
 export async function buildUserTrafficList(rangeStartMs: number, rangeEndMs: number): Promise<UserTrafficDto[]> {
   const sinceIso = new Date(rangeStartMs).toISOString()
   const users = await listUsers()
+  const [context, allServers] = await Promise.all([readTrafficCollectionContext(), db.select().from(servers)])
+  const catalog = new Map(allServers.map(s => [s.id, s]))
   const sampleCache = new Map<number, Awaited<ReturnType<typeof readServerSamplesInRange>>>()
 
   return Promise.all(users.map(async (user) => {
@@ -87,7 +95,10 @@ export async function buildUserTrafficList(rangeStartMs: number, rangeEndMs: num
         sampleCache.set(b.serverId, rows)
       }
       const peerKey = b.peerPublicKey ?? ""
-      const built = buildTrafficFromSamples(rows, rangeStartMs, rangeEndMs, b.interfaceName, peerKey)
+      const built = buildTrafficFromSamples(rows, rangeStartMs, rangeEndMs, b.interfaceName, peerKey, true)
+      const server = catalog.get(b.serverId)
+      if (built.observations) built.observations = applyTrafficFreshness(built.observations,
+        server ?? { enabled: false, updatedAt: "" }, context.settings, context.states.get(b.serverId), rangeEndMs)
       parts.push(built)
       const last = [...rows.filter((r) =>
         r.interfaceName === b.interfaceName && (r.peerPublicKey ?? "") === peerKey,