fix(netflow): сохранять отдельные поля и защищать созданную пару ключей

backend/src/services/traffic-flow-key-execution.test.ts

@@ -0,0 +1,19 @@
+import assert from "node:assert/strict"
+import { ensureTrafficFlowHostKeys } from "./traffic-flow-key-execution.js"
+let pair = {publicKey:"",privateKey:""},generated=0,writes=0
+let release!:()=>void
+const bothRead = new Promise<void>(resolve=>{release=resolve})
+let initialReads=0
+const deps = {
+ read:async()=>{const snapshot={...pair}; if(initialReads<2){initialReads++;if(initialReads===2)release();await bothRead} return snapshot},
+ generate:()=>{generated++;return {publicKey:"public-"+generated,privateKey:"private-"+generated}},
+ installIfMissing:async(keys:typeof pair)=>{writes++;if(pair.publicKey&&pair.privateKey)return false;pair=keys;return true},
+}
+const results=await Promise.all([ensureTrafficFlowHostKeys(deps),ensureTrafficFlowHostKeys(deps)])
+assert.equal(generated,2);assert.equal(writes,2);assert.equal(results.filter(r=>r.created).length,1)
+assert.equal(results[0].publicKey,results[1].publicKey);assert.equal(results[0].publicKey,pair.publicKey)
+const before={...pair};assert.equal((await ensureTrafficFlowHostKeys(deps)).created,false);assert.deepEqual(pair,before);assert.equal(writes,2)
+await assert.rejects(ensureTrafficFlowHostKeys({read:async()=>({publicKey:"",privateKey:""}),generate:deps.generate,installIfMissing:async()=>true}),/не подтверждено/)
+let called=0
+await assert.rejects(ensureTrafficFlowHostKeys({read:async()=>({publicKey:"",privateKey:""}),generate:deps.generate,installIfMissing:async()=>{called++;throw Error("DB failure")}}),/DB failure/)
+assert.equal(called,1);console.log("NetFlow key execution: concurrent callers share winner, existing pair unchanged, unknown/failure never retried")

backend/src/services/traffic-flow-key-execution.ts

@@ -0,0 +1,15 @@
+type KeyPair = { publicKey: string; privateKey: string }
+type Dependencies = {
+  read: () => Promise<KeyPair>
+  generate: () => KeyPair
+  installIfMissing: (keys: KeyPair) => Promise<boolean>
+}
+/** A conditional database write selects the winner across processes; no existing pair is rotated. */
+export async function ensureTrafficFlowHostKeys(deps: Dependencies): Promise<{ publicKey: string; created: boolean }> {
+  const existing = await deps.read()
+  if (existing.publicKey && existing.privateKey) return { publicKey: existing.publicKey, created: false }
+  const created = await deps.installIfMissing(deps.generate())
+  const confirmed = await deps.read()
+  if (!confirmed.publicKey || !confirmed.privateKey) throw new Error("Создание ключей NetFlow не подтверждено")
+  return { publicKey: confirmed.publicKey, created }
+}

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

@@ -1,8 +1,9 @@
-import { eq } from "drizzle-orm"
+import { and, eq, or } from "drizzle-orm"
 import { db } from "../db/index.js"
 import { parseJsonArray } from "../db/json.js"
 import { trafficFlowSettings } from "../db/schema.js"
 import type { FlowHostPeer, TrafficFlowSettingsDto, TrafficFlowSettingsPatch } from "@mmapp/contracts/traffic-flow"
+import { ensureTrafficFlowHostKeys } from "./traffic-flow-key-execution.js"
 import { generateWireGuardKeyPair } from "./wg-keys.js"
 
 type SettingsRow = NonNullable<Awaited<ReturnType<typeof readSettingsRow>>>
@@ -38,7 +39,7 @@ export async function getTrafficFlowSettingsRow(): Promise<SettingsRow> {
     prefix: "10.255.254.0/24",
     createdAt: now,
     updatedAt: now,
-  })
+  }).onConflictDoNothing()
   row = await readSettingsRow()
   settingsRowCache = row!
   return settingsRowCache
@@ -79,39 +80,31 @@ export async function toTrafficFlowSettingsDto(
 }
 
 export async function updateTrafficFlowSettings(patch: TrafficFlowSettingsPatch) {
-  const row = await getTrafficFlowSettingsRow()
-  await db.update(trafficFlowSettings).set({
-    enabled: patch.enabled ?? row.enabled,
-    collectorIp: patch.collectorIp ?? row.collectorIp,
-    flowListenPort: patch.flowListenPort ?? row.flowListenPort,
-    wgListenPort: patch.wgListenPort ?? row.wgListenPort,
-    prefix: patch.prefix ?? row.prefix,
-    publicEndpoint: patch.publicEndpoint ?? row.publicEndpoint,
-    hubServerId: patch.hubServerId === undefined ? row.hubServerId : patch.hubServerId,
-    retentionHours: patch.retentionHours ?? row.retentionHours,
-    topN: patch.topN ?? row.topN,
-    mapServiceMinSharePct: patch.mapServiceMinSharePct == null
-      ? row.mapServiceMinSharePct
-      : Math.min(100, Math.max(0, patch.mapServiceMinSharePct)),
-    updatedAt: nowIso(),
-  }).where(eq(trafficFlowSettings.id, 1))
+  await getTrafficFlowSettingsRow()
+  // Do not copy untouched fields from a potentially stale cache into a concurrent update.
+  await db.update(trafficFlowSettings).set({ ...patch, updatedAt: nowIso() }).where(eq(trafficFlowSettings.id, 1))
   invalidateTrafficFlowSettingsCache()
   return await getTrafficFlowSettingsRow()
 }
 
 export async function ensureHostKeys(): Promise<{ publicKey: string; created: boolean }> {
-  const row = await getTrafficFlowSettingsRow()
-  if (row.hostPublicKey && row.hostPrivateKey) {
-    return { publicKey: row.hostPublicKey, created: false }
-  }
-  const keys = generateWireGuardKeyPair()
-  await db.update(trafficFlowSettings).set({
-    hostPublicKey: keys.publicKey,
-    hostPrivateKey: keys.privateKey,
-    updatedAt: nowIso(),
-  }).where(eq(trafficFlowSettings.id, 1))
-  invalidateTrafficFlowSettingsCache()
-  return { publicKey: keys.publicKey, created: true }
+  await getTrafficFlowSettingsRow()
+  return ensureTrafficFlowHostKeys({
+    read: async () => {
+      invalidateTrafficFlowSettingsCache()
+      const row = await getTrafficFlowSettingsRow()
+      return { publicKey: row.hostPublicKey, privateKey: row.hostPrivateKey }
+    },
+    generate: generateWireGuardKeyPair,
+    installIfMissing: async keys => {
+      const rows = await db.update(trafficFlowSettings).set({
+        hostPublicKey: keys.publicKey, hostPrivateKey: keys.privateKey, updatedAt: nowIso(),
+      }).where(and(eq(trafficFlowSettings.id, 1), or(eq(trafficFlowSettings.hostPublicKey, ""), eq(trafficFlowSettings.hostPrivateKey, ""))))
+        .returning({ id: trafficFlowSettings.id })
+      invalidateTrafficFlowSettingsCache()
+      return rows.length === 1
+    },
+  })
 }
 
 export async function upsertHostPeer(peer: FlowHostPeer) {