fix(collectors): сохранять частичный результат и актуальные GRE BGP сигналы

backend/src/services/alert-engine/bgp-signals.ts

@@ -9,7 +9,7 @@ export function isKnownBgpState(state: string): boolean {
 export function collectBgpSignalsFromRuns(runs: GreBgpSnapshotRunSnapshot[]): BgpPeerSignal[] {
   const byKey = new Map<string, BgpPeerSignal>()
   const ambiguousHistory = new Set<string>()
-  for (const run of runs) {
+  for (const [index, run] of runs.entries()) {
     const batch = new Map<string, string>()
     const duplicates = new Set<string>()
     for (const row of run.bgpPeers ?? []) {
@@ -22,7 +22,8 @@ export function collectBgpSignalsFromRuns(runs: GreBgpSnapshotRunSnapshot[]): Bg
     for (const [key, state] of batch) {
       const duplicate = duplicates.has(key)
       const current = byKey.get(key)
-      if (!current) byKey.set(key, { key, state: duplicate ? "Unknown" : state, prevState: null })
+      if (!current && index === 0) byKey.set(key, { key, state: duplicate ? "Unknown" : state, prevState: null })
+      else if (!current) continue
       else if (!duplicate && !ambiguousHistory.has(key) && current.prevState === null && isKnownBgpState(state)) current.prevState = state
       if (duplicate) ambiguousHistory.add(key)
     }

backend/src/services/alert-engine/evaluate-rule.test.ts

@@ -139,6 +139,8 @@ for (const pair of [["Established", "Idle"], ["Established", "Established"], ["U
   const runs = [{ bgpPeers: pair.map(state => ({ key: "srv: peer1", state })) }] as GreBgpSnapshotRunSnapshot[]
   assert.deepEqual(collectBgpSignalsFromRuns(runs), [{ key: "srv: peer1", state: "Unknown", prevState: null }])
 }
+assert.deepEqual(collectBgpSignalsFromRuns([{ bgpPeers: [] }, { bgpPeers: [{ key: "removed-peer", state: "Established" }] }] as GreBgpSnapshotRunSnapshot[]), [])
+assert.deepEqual(collectBgpSignalsFromRuns([{ bgpPeers: [{ key: "current-peer", state: "Established" }] }, { bgpPeers: [{ key: "removed-peer", state: "Idle" }] }] as GreBgpSnapshotRunSnapshot[]), [{ key: "current-peer", state: "Established", prevState: null }])
 {
   const runs = [
     { bgpPeers: [{ key: "srv: peer1", state: "Established" }] },

backend/src/services/alert-engine/gre-signals.ts

@@ -0,0 +1,24 @@
+import type { GreBgpSnapshotRunSnapshot } from "../../types/scheduler-run-snapshot.js"
+import type { GreTunnelSignal } from "./types.js"
+
+export function collectGreSignalsFromRuns(runs: GreBgpSnapshotRunSnapshot[]): GreTunnelSignal[] {
+  const signals = new Map<string, GreTunnelSignal>()
+  const ambiguous = new Set<string>()
+  for (const row of runs[0]?.greTunnels ?? []) {
+    const label = row.targetLabel.trim()
+    if (!label || ambiguous.has(label)) continue
+    if (signals.has(label)) { signals.delete(label); ambiguous.add(label) }
+    else signals.set(label, { targetLabel: label, status: row.status, prevStatus: null })
+  }
+  for (const run of runs.slice(1)) {
+    const rows = run.greTunnels ?? []
+    const counts = new Map<string, number>()
+    for (const row of rows) { const label = row.targetLabel.trim(); counts.set(label, (counts.get(label) ?? 0) + 1) }
+    for (const row of rows) {
+      const label = row.targetLabel.trim(), signal = signals.get(label)
+      if (counts.get(label)! > 1) ambiguous.add(label)
+      else if (signal && !ambiguous.has(label) && signal.prevStatus === null) signal.prevStatus = row.status
+    }
+  }
+  return [...signals.values()]
+}

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

@@ -1,7 +1,8 @@
 import { dbAll } from "../../db/index.js"
 import { collectBgpSignalsFromRuns } from "./bgp-signals.js"
+import { collectGreSignalsFromRuns } from "./gre-signals.js"
+import { readGreBgpRunSnapshots } from "../gre-bgp-snapshot-history.js"
 import type {
-  GreBgpSnapshotRunSnapshot,
   PingRunSnapshot,
   ResourcesRunSnapshot,
   SchedulerRunSnapshot,
@@ -9,7 +10,6 @@ import type {
   TrafficRunSnapshot,
 } from "../../types/scheduler-run-snapshot.js"
 import type {
-  GreTunnelSignal,
   ProbeSignal,
   ServerSignal,
   SignalSnapshot,
@@ -184,28 +184,6 @@ function collectTrafficSignalsFromRuns(runs: TrafficRunSnapshot[]): TrafficServe
   return [...out.values()]
 }
 
-function collectGreSignalsFromRuns(runs: GreBgpSnapshotRunSnapshot[]): GreTunnelSignal[] {
-  const byTarget = new Map<string, GreTunnelSignal>()
-  for (const run of runs) {
-    const greRows = run.greTunnels ?? []
-    for (const row of greRows) {
-      const label = String(row.targetLabel ?? "").trim()
-      if (!label) continue
-      const prev = byTarget.get(label)
-      if (!prev) {
-        byTarget.set(label, {
-          targetLabel: label,
-          status: row.status,
-          prevStatus: null,
-        })
-        continue
-      }
-      if (prev.prevStatus == null) prev.prevStatus = row.status
-    }
-  }
-  return [...byTarget.values()]
-}
-
 /** Собирает сигналы alert_engine напрямую из результатов collector jobs (`scheduler_runs.result_json`). */
 export async function buildSignalSnapshotFromCollectors(): Promise<SignalSnapshot> {
   const sampledAt = new Date().toISOString()
@@ -213,7 +191,7 @@ export async function buildSignalSnapshotFromCollectors(): Promise<SignalSnapsho
   const restRuns = await readLatestOkRunSnapshots("servers_rest_ping", 16) as ServersRestPingRunSnapshot[]
   const pingRuns = await readLatestOkRunSnapshots("uptime_ping", 24) as PingRunSnapshot[]
   const trafficRuns = await readLatestOkRunSnapshots("traffic", 8) as TrafficRunSnapshot[]
-  const greBgpRuns = await readLatestOkRunSnapshots("gre_bgp", 8) as GreBgpSnapshotRunSnapshot[]
+  const greBgpRuns = await readGreBgpRunSnapshots()
 
   const resourceByKey = collectServerSignalsFromResourceRuns(resourceRuns)
   const restByKey = collectServerSignalsFromRestRuns(restRuns)

backend/src/services/bgp-peers-live.ts

@@ -8,7 +8,7 @@ import { parseBgpSessions,parseBgpCatalogSessions } from "./bgp-parse-sessions.j
 import { MikrotikClient } from "./mikrotik.js"
 
 /** Как в UI `/alerts`: `serverName: peerName`. */
-export function bgpPeerAlertKey(s: BgpSessionRead): string {
+export function bgpPeerAlertKey(s: Pick<BgpSessionRead, "serverName" | "name">): string {
   const sn = String(s.serverName ?? "").trim()
   const nm = String(s.name ?? "").trim()
   return `${sn}: ${nm}`.trim()

backend/src/services/gre-bgp-snapshot-collector.test.ts

@@ -0,0 +1,26 @@
+import assert from "node:assert/strict"
+import { collectGreBgpSnapshotOnce, getGreBgpSnapshotCollectorState } from "./gre-bgp-snapshot-collector.js"
+const peer = { serverName: "CHR", name: "own-peer", state: "Established" }
+{
+  let release!: () => void
+  let bgpStarted = false
+  const pending = collectGreBgpSnapshotOnce({ fetchGre: () => new Promise(resolve => { release = () => resolve([{ targetLabel: "own / CHR", status: "up" }]) }), fetchBgp: async () => { bgpStarted = true; return [peer] } })
+  await Promise.resolve(); await Promise.resolve()
+  assert(bgpStarted, "BGP starts before GRE finishes")
+  assert(getGreBgpSnapshotCollectorState().running)
+  let duplicateCalls = 0
+  const duplicate = await collectGreBgpSnapshotOnce({ fetchGre: async () => { duplicateCalls++; return [] }, fetchBgp: async () => { duplicateCalls++; return [] } })
+  assert(duplicate.skipped); assert.equal(duplicateCalls, 0)
+  release(); const result = await pending
+  assert.equal(result.greWritten, 1); assert.equal(result.bgpWritten, 1)
+}
+for (const failGre of [true, false]) {
+  const result = await collectGreBgpSnapshotOnce({ fetchGre: () => { if (failGre) throw Error("GRE source refused"); return Promise.resolve([{ targetLabel: "own / CHR", status: "up" }]) }, fetchBgp: async () => { if (!failGre) throw Error("BGP source refused"); return [peer] } })
+  assert.equal(result.fatalError, undefined)
+  assert.deepEqual(result.errors, [failGre ? "GRE: GRE source refused" : "BGP: BGP source refused"])
+  assert.equal(result.greWritten, failGre ? 0 : 1); assert.equal(result.bgpWritten, failGre ? 1 : 0)
+  assert.equal(getGreBgpSnapshotCollectorState().running, false)
+}
+const empty = await collectGreBgpSnapshotOnce({ fetchGre: async () => [], fetchBgp: async () => [] })
+assert.deepEqual(empty.errors, []); assert.equal(empty.bgpWritten, 0); assert.equal(empty.greWritten, 0)
+console.log("GRE/BGP collector: concurrent independent reads, scoped sync/async failures, no duplicate fetch and recovery; no SQL/RouterOS")

backend/src/services/gre-bgp-snapshot-collector.ts

@@ -2,6 +2,12 @@ import { bgpPeerAlertKey, fetchBgpSessionsForAlerts } from "./bgp-peers-live.js"
 import { fetchGreTunnelLiveRows } from "./gre-tunnels-live.js"
 import type { GreBgpSnapshotRunSnapshot } from "../types/scheduler-run-snapshot.js"
 import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
+import type { BgpSessionRead } from "../types/server.js"
+
+interface CollectorDependencies {
+  fetchGre: () => Promise<NonNullable<GreBgpSnapshotRunSnapshot["greTunnels"]>>
+  fetchBgp: () => Promise<Array<Pick<BgpSessionRead, "serverName" | "name" | "state">>>
+}
 
 let collecting = false
 
@@ -13,7 +19,7 @@ export function getGreBgpSnapshotCollectorState(): { running: boolean } {
  * Один опрос GRE + BGP по включённым серверам.
  * Снимок для алертов живёт в `scheduler_runs.result_json` (`greTunnels` / `bgpPeers`).
  */
-export async function collectGreBgpSnapshotOnce(): Promise<GreBgpSnapshotRunSnapshot> {
+export async function collectGreBgpSnapshotOnce(deps: CollectorDependencies = { fetchGre: fetchGreTunnelLiveRows, fetchBgp: fetchBgpSessionsForAlerts }): Promise<GreBgpSnapshotRunSnapshot> {
   const sampledAt = new Date().toISOString()
   if (collecting) {
     return {
@@ -41,8 +47,8 @@ export async function collectGreBgpSnapshotOnce(): Promise<GreBgpSnapshotRunSnap
 
   try {
     const [greSettled, bgpSettled] = await Promise.allSettled([
-      await fetchGreTunnelLiveRows(),
-      await fetchBgpSessionsForAlerts(),
+      Promise.resolve().then(() => deps.fetchGre()),
+      Promise.resolve().then(() => deps.fetchBgp()),
     ])
 
     if (greSettled.status === "rejected") {

backend/src/services/gre-bgp-snapshot-history.test.ts

@@ -0,0 +1,24 @@
+import assert from "node:assert/strict"
+import { readGreBgpRunSnapshots } from "./gre-bgp-snapshot-history.js"
+import { collectBgpSignalsFromRuns } from "./alert-engine/bgp-signals.js"
+import { collectGreSignalsFromRuns } from "./alert-engine/gre-signals.js"
+import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
+const old = { v: SCHEDULER_RUN_SNAPSHOT_VERSION, job: "gre_bgp", sampledAt: "2026-10-12T00:00:00Z", bgpPeers: [{ key: "old-peer", state: "Idle" }], greTunnels: [{ targetLabel: "old-gre", status: "down" }], greWritten: 1, bgpWritten: 1 }
+const row = (resultJson: unknown) => ({ resultJson, finishedAt: "2026-10-12T00:01:00Z" })
+for (const broken of [null, "{", { job: "wrong" }, { ...old, bgpPeers: [null] }, { ...old, greTunnels: "bad" }]) {
+  const snapshots = await readGreBgpRunSnapshots(async (sql, params) => { assert.match(sql, /status IN \('ok', 'error'\)/); assert.deepEqual(params, ["gre_bgp", 8]); return [row(broken), row(old)] })
+  assert(snapshots[0].fatalError)
+  assert.deepEqual(collectBgpSignalsFromRuns(snapshots), [])
+  assert.deepEqual(collectGreSignalsFromRuns(snapshots), [])
+}
+const current = { ...old, bgpPeers: [{ key: "current-peer", state: "Established" }], greTunnels: [], errors: ["GRE: source refused"] }
+const partial = await readGreBgpRunSnapshots(async () => [row(JSON.stringify(current)), row(old)])
+assert.deepEqual(collectBgpSignalsFromRuns(partial), [{ key: "current-peer", state: "Established", prevState: null }])
+assert.deepEqual(collectGreSignalsFromRuns(partial), [])
+const empty = await readGreBgpRunSnapshots(async () => [row({ ...old, bgpPeers: [], greTunnels: [] }), row(old)])
+assert.deepEqual(collectBgpSignalsFromRuns(empty), []); assert.deepEqual(collectGreSignalsFromRuns(empty), [])
+const duplicate = { ...old, greTunnels: [{ targetLabel: "old-gre", status: "up" as const }, { targetLabel: "old-gre", status: "down" as const }] }
+assert.deepEqual(collectGreSignalsFromRuns([duplicate]), [])
+const history = await readGreBgpRunSnapshots(async () => [row({ ...old, greTunnels: [{ targetLabel: "old-gre", status: "up" }] }), row(duplicate), row(old)])
+assert.deepEqual(collectGreSignalsFromRuns(history), [{ targetLabel: "old-gre", status: "up", prevStatus: null }])
+console.log("GRE/BGP history: latest completed failures retained, malformed/null barrier, partial BGP preserved, empty/deleted peers never resurrected; no SQL query")

backend/src/services/gre-bgp-snapshot-history.ts

@@ -0,0 +1,23 @@
+import { dbAll } from "../db/index.js"
+import { SCHEDULER_RUN_SNAPSHOT_VERSION, type GreBgpSnapshotRunSnapshot } from "../types/scheduler-run-snapshot.js"
+interface RunRow { resultJson: unknown; finishedAt: Date | string }
+type ReadRows = (sql: string, params: unknown[]) => Promise<RunRow[]>
+
+/** Completed failures are evidence of missing current data, not permission to reuse an older success. */
+export async function readGreBgpRunSnapshots(read: ReadRows = (sql, params) => dbAll<RunRow>(sql, params)): Promise<GreBgpSnapshotRunSnapshot[]> {
+  const rows = await read(`SELECT result_json AS "resultJson", finished_at AS "finishedAt"
+    FROM scheduler_runs WHERE job_key = ? AND status IN ('ok', 'error')
+    ORDER BY finished_at DESC LIMIT ?`, ["gre_bgp", 8])
+  return rows.map(row => {
+    let parsed: unknown = row.resultJson
+    if (typeof parsed === "string") { try { parsed = JSON.parse(parsed) } catch { parsed = null } }
+    if (parsed && typeof parsed === "object") {
+      const s = parsed as GreBgpSnapshotRunSnapshot
+      const validBgp = s.bgpPeers === undefined || (Array.isArray(s.bgpPeers) && s.bgpPeers.every(p => p && typeof p.key === "string" && typeof p.state === "string"))
+      const validGre = s.greTunnels === undefined || (Array.isArray(s.greTunnels) && s.greTunnels.every(p => p && typeof p.targetLabel === "string" && ["up", "down", "degraded"].includes(p.status)))
+      if (s.v === SCHEDULER_RUN_SNAPSHOT_VERSION && s.job === "gre_bgp" && validBgp && validGre) return s
+    }
+    const date = new Date(row.finishedAt)
+    return { v: SCHEDULER_RUN_SNAPSHOT_VERSION, job: "gre_bgp", sampledAt: Number.isFinite(date.getTime()) ? date.toISOString() : new Date(0).toISOString(), greWritten: 0, bgpWritten: 0, greTunnels: [], bgpPeers: [], fatalError: "Недоступен результат последнего опроса GRE/BGP" }
+  })
+}