From 9240e8be1176ef5336cf30d77ab136bee5127c4c Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 12 Nov 2026 17:46:59 +0700 Subject: [PATCH] =?UTF-8?q?fix(netflow):=20=D1=81=D0=B2=D1=8F=D0=B7=D0=B0?= =?UTF-8?q?=D1=82=D1=8C=20=D0=BE=D1=82=D0=BC=D0=B5=D1=82=D0=BA=D0=B8=20?= =?UTF-8?q?=D0=BF=D0=BE=D1=82=D0=BE=D0=BA=D0=B0=20SSE=20=D1=81=20=D0=B4?= =?UTF-8?q?=D0=B8=D0=B0=D0=B3=D0=BD=D0=BE=D1=81=D1=82=D0=B8=D0=BA=D0=BE?= =?UTF-8?q?=D0=B9=20NetFlow?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/src/routes/traffic-flow.ts | 23 ++++++- .../traffic-flow-read-http.owned.test.ts | 1 + backend/src/test/mcp-flow-analytics.ts | 15 +---- backend/src/test/mcp-flow-live.ts | 62 +++++++++---------- backend/src/test/mcp-flow-map.ts | 17 ++--- backend/src/test/mcp-flow-query-race.ts | 45 ++++++++++++++ 6 files changed, 105 insertions(+), 58 deletions(-) create mode 100644 backend/src/test/mcp-flow-query-race.ts diff --git a/backend/src/routes/traffic-flow.ts b/backend/src/routes/traffic-flow.ts index ca3d3945..a03134e7 100644 --- a/backend/src/routes/traffic-flow.ts +++ b/backend/src/routes/traffic-flow.ts @@ -1,3 +1,5 @@ +import { randomUUID } from "node:crypto" +import { channel } from "node:diagnostics_channel" import { createFlowReadResponseOwner, writeBoundedFlowSse, withFlowReadSourceWindow } from "../services/traffic-flow-read-budget.js" import { flowReadBudget, flowResponseReadLimits, FLOW_READ_RESERVATION_BYTES, FLOW_SSE_MAX_FRAME_BYTES } from "../services/traffic-flow-read-policy.js" import { updateTrafficFlowSettingsWithListener } from "../services/traffic-flow-settings-service.js" @@ -44,6 +46,9 @@ import { listTrafficFlowHostFiles } from "../services/traffic-flow-host-files.js import { FLOW_LIVE_TICK_MS as LIVE_TICK_MS, tryAcquireFlowLiveSlot, releaseFlowLiveSlot } from "../services/traffic-flow-live-slots.js" export { MAX_FLOW_LIVE_SUBSCRIBERS, tryAcquireFlowLiveSlot, releaseFlowLiveSlot, resetFlowLiveSlotsForTests } from "../services/traffic-flow-live-slots.js" +export const flowSseDiagnostics = channel("mmapp.netflow.sse") +type FlowSseStage = "open" | "build-start" | "build-ready" | "write-start" | "write-complete" | "closed" + function rangeToMinutes(range: string | undefined): number { switch ((range ?? "5m").toLowerCase()) { case "5m": return 5 @@ -248,6 +253,12 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => { Connection: "keep-alive", "X-Accel-Buffering": "no", } + // A diagnostic subscriber can correlate its timestamps with the same stream in Edge. + const diagnosticId = flowSseDiagnostics.hasSubscribers ? randomUUID() : undefined + const trace = diagnosticId === undefined ? undefined : (stage: FlowSseStage) => { + flowSseDiagnostics.publish({ streamId: diagnosticId, stage, atUnixMs: Date.now(), monotonicMs: performance.now() }) + } + if (diagnosticId !== undefined) sseHeaders["X-MM-NetFlow-Stream"] = diagnosticId if (origin && (allowed === "*" || allowed === origin)) { sseHeaders["Access-Control-Allow-Origin"] = origin sseHeaders["Access-Control-Allow-Credentials"] = "true" @@ -255,19 +266,25 @@ const trafficFlowRoutes: FastifyPluginAsyncZod = async (app) => { sseHeaders.Vary = "Origin" } reply.raw.writeHead(200, sseHeaders) + trace?.("open") reply.raw.write(":\n\n") try { while (!abort.signal.aborted) { - await withFlowReadSourceWindow(async()=>{ - const payload = await safeBuildLiveFlowSample(liveQuery) - await writeBoundedFlowSse(reply.raw, payload.event, payload.data, FLOW_SSE_MAX_FRAME_BYTES) + await withFlowReadSourceWindow(async () => { + trace?.("build-start") + const payload = await safeBuildLiveFlowSample(liveQuery) + trace?.("build-ready") + trace?.("write-start") + await writeBoundedFlowSse(reply.raw, payload.event, payload.data, FLOW_SSE_MAX_FRAME_BYTES) + trace?.("write-complete") }) await sleep(LIVE_TICK_MS, abort.signal) } } catch { /* abort / disconnect */ } finally { + trace?.("closed") releaseFlowLiveSlot() reply.raw.off("close", onClose) try { diff --git a/backend/src/services/traffic-flow-read-http.owned.test.ts b/backend/src/services/traffic-flow-read-http.owned.test.ts index 1504b775..3ccb7355 100644 --- a/backend/src/services/traffic-flow-read-http.owned.test.ts +++ b/backend/src/services/traffic-flow-read-http.owned.test.ts @@ -143,6 +143,7 @@ try{ const controller=new AbortController();streams.push(controller) const response=await fetch(endpoint+'/api/traffic/flow/live?serverId=1',{signal:controller.signal}) assert.equal(response.status,200) + assert.equal(response.headers.get('x-mm-netflow-stream'),null) const reader=response.body!.getReader();await reader.read() } assert.equal(flowReadBudget.stats().owners,4) diff --git a/backend/src/test/mcp-flow-analytics.ts b/backend/src/test/mcp-flow-analytics.ts index e4745331..d441ed36 100644 --- a/backend/src/test/mcp-flow-analytics.ts +++ b/backend/src/test/mcp-flow-analytics.ts @@ -11,6 +11,7 @@ import { buildFlowAnalytics, invalidateFlowAnalyticsCache } from "../services/tr import { loadFlowTopology, invalidateFlowCatalogCache } from "../services/traffic-flow-topology.js" import { resetFlowRingsForTests as clearPendingFlows } from "../services/traffic-flow-ingest.js" import { disableRipeEnqueueForTests, resetRipeCacheForTests } from "../services/traffic-flow-ripe.js" +import { duringFlowSourceQuery } from "./mcp-flow-query-race.js" export async function verifyMcpFlowAnalytics(actor: AuthUser) { const owner = { ...actor, id: "flow-analytics-owner", permissions: [...actor.permissions, "mm:traffic:read"] } @@ -19,7 +20,6 @@ export async function verifyMcpFlowAnalytics(actor: AuthUser) { const access = await createMcpConnection(mcpConnectionCreateSchema.parse({ name: "Scoped analytics", mode: "observer", expiresAt: new Date(Date.now()+1800000).toISOString(), permissions: ["mm:traffic:read"], scope: { serverIds: [110], operationIds: ["traffic.flow.analytics.read"], allowGlobal: false }, limits: { callsPerMinute: 120 } }), owner) const principal = await authenticateMcpToken(access.token) - const originalQuery = pool.query.bind(pool) const read = (extra: Record = {}) => invokeMcpTool(principal, "mm_netflow_analytics_report_get", { serverId: 110, minutes: 5, ...extra }) as Promise<{ analytics: FlowAnalyticsDto; pages: Record; live: boolean; forwardingVerified: boolean }> const bucket = new Date(Date.now()-60000).toISOString() const put = (sid: number, name: string, bytes: string | number, src: string, at = bucket) => pool.query("INSERT INTO flow_buckets(server_id,bucket_at,src,dst,proto,src_port,dst_port,bytes,packets,in_iface) VALUES($1,$2,$3,'8.8.8.8',6,12345,443,$4,1,$5)", [sid,at,src,bytes,name]) @@ -55,16 +55,8 @@ export async function verifyMcpFlowAnalytics(actor: AuthUser) { await assert.rejects(read({serverId:111})); await assert.rejects(read({minutes:1441})); await assert.rejects(read({interface:"__all__"})); await assert.rejects(read({skipHeavy:true})) await assert.rejects(buildFlowAnalytics({serverId:110,minutes:5},{serverIds:[110],maxRows:1,authorize:async()=>{}}),/предел/) const cancelled = new AbortController(); cancelled.abort(); await assert.rejects(invokeMcpTool(principal,"mm_netflow_analytics_report_get",{serverId:110},cancelled.signal)) - const duringQuery = async (match: string, action: () => Promise) => { - let fired = false - pool.query = ((...args: unknown[]) => { - const sql = typeof args[0] === "string" ? args[0] : (args[0] as {text?:string})?.text - const pending = (originalQuery as (...args: unknown[])=>Promise)(...args) - if (!fired && sql?.includes(match)) { fired = true; return pending.then(async value=>{ await action(); return value }) } - return pending - }) as typeof pool.query - try { await assert.rejects(read()); assert.equal(fired,true) } finally { pool.query=originalQuery as typeof pool.query } - } + const duringQuery = (match: string, action: () => Promise) => + duringFlowSourceQuery(pool, match, action, () => assert.rejects(read())) await duringQuery("FROM flow_buckets",async()=>{await pool.query("UPDATE servers SET host='192.0.2.210' WHERE id=110");await pool.query("UPDATE servers SET host='192.0.2.110' WHERE id=110")}) await duringQuery("FROM server_snapshots",async()=>{await pool.query("UPDATE user_interface_bindings SET interface_name='changed-during-read' WHERE user_id=$1",[users[0]])}) await pool.query("UPDATE user_interface_bindings SET interface_name='analytics-in' WHERE user_id=$1",[users[0]]) @@ -74,7 +66,6 @@ export async function verifyMcpFlowAnalytics(actor: AuthUser) { assert.equal((await pool.query("SELECT count(*)::int AS n FROM mcp_plans WHERE connection_id=$1",[access.connection.id])).rows[0].n,0) console.log("✓ Scoped NetFlow analytics: REST engine parity, hidden server/user/peer exclusion, cache isolation, bounds/paging/future rows, source ABA/topology drift, cancel/pause and no plans") } finally { - pool.query=originalQuery as typeof pool.query await setConnectionStatus(access.connection.id,"paused",owner) await pool.query("DELETE FROM user_interface_bindings WHERE user_id=ANY($1::text[])",[users]); await pool.query("DELETE FROM app_users WHERE id=ANY($1::text[])",[users]) await pool.query("DELETE FROM flow_buckets WHERE server_id IN (110,111)"); await pool.query("DELETE FROM servers WHERE id IN (110,111)") diff --git a/backend/src/test/mcp-flow-live.ts b/backend/src/test/mcp-flow-live.ts index 043ff9d0..8f024818 100644 --- a/backend/src/test/mcp-flow-live.ts +++ b/backend/src/test/mcp-flow-live.ts @@ -10,11 +10,12 @@ import { ensurePartitionFor } from "../db/partitions.js" import { createMcpConnection, authenticateMcpToken, setConnectionStatus } from "../modules/mcp/store.js" import { invokeMcpTool } from "../modules/mcp/runtime.js" import { invalidateFlowAnalyticsCache } from "../services/traffic-flow-analytics.js" -import trafficFlowRoutes from "../routes/traffic-flow.js" +import trafficFlowRoutes, { flowSseDiagnostics } from "../routes/traffic-flow.js" import { invalidateFlowCatalogCache } from "../services/traffic-flow-topology.js" import { resetFlowRingsForTests } from "../services/traffic-flow-ingest.js" import { tryAcquireFlowLiveSlot, releaseFlowLiveSlot } from "../services/traffic-flow-live-slots.js" import { disableRipeEnqueueForTests, resetRipeCacheForTests } from "../services/traffic-flow-ripe.js" +import { duringFlowSourceQuery } from "./mcp-flow-query-race.js" type Observation = { events: Array<{ event: string; data: FlowAnalyticsDto; pages: Record }>; observationComplete: boolean; forwardingVerified: boolean } export async function verifyMcpFlowLive(actor: AuthUser) { @@ -22,7 +23,7 @@ export async function verifyMcpFlowLive(actor: AuthUser) { const user = "flow-live-mixed", local = "flow-live-local", bucket = new Date(Date.now()-60000) await pool.query("INSERT INTO servers(id,name,host,password) VALUES(114,'live-visible','192.0.2.114','NEVER_LIVE_PASSWORD'),(115,'NEVER_LIVE_SERVER','192.0.2.115','NEVER_LIVE_PASSWORD')") const access = await createMcpConnection(mcpConnectionCreateSchema.parse({ name: "Scoped live", mode: "observer", expiresAt: new Date(Date.now()+1800000).toISOString(), permissions: ["mm:traffic:read"], scope: { serverIds: [114], operationIds: ["traffic.flow.live.read"], allowGlobal: false }, limits: { callsPerMinute: 120 } }), owner) - const principal = await authenticateMcpToken(access.token), originalQuery = pool.query.bind(pool) + const principal = await authenticateMcpToken(access.token) const http = Fastify(); http.setValidatorCompiler(validatorCompiler); http.setSerializerCompiler(serializerCompiler) await http.register(trafficFlowRoutes,{prefix:"/api"}) const address = await http.listen({host:"127.0.0.1",port:0}) @@ -39,9 +40,14 @@ export async function verifyMcpFlowLive(actor: AuthUser) { // Read the real SSE endpoint, including its event envelope and live engine defaults. const httpAbort = new AbortController(), timeout = setTimeout(()=>httpAbort.abort(),10000) let rest: { event: string; data: FlowAnalyticsDto } + const stages: unknown[] = [] + const captureStage = (event: unknown) => stages.push(event) + flowSseDiagnostics.subscribe(captureStage) try { const response = await fetch(`${address}/api/traffic/flow/live?serverId=114`,{signal:httpAbort.signal}) assert.equal(response.status,200) + const diagnosticId = response.headers.get("x-mm-netflow-stream") + assert.match(diagnosticId ?? "", /^[a-f0-9-]{36}$/) const reader = response.body!.getReader(), decoder = new TextDecoder(); let body = "" for (;;) { const chunk = await reader.read(); assert.equal(chunk.done,false) @@ -49,7 +55,18 @@ export async function verifyMcpFlowLive(actor: AuthUser) { const match = /event: (sample|error)\ndata: ([^\n]+)\n\n/.exec(body) if (match) { rest={event:match[1],data:JSON.parse(match[2])}; break } } - } finally { clearTimeout(timeout); httpAbort.abort() } + const events = stages as {streamId: string; stage: string; atUnixMs: number; monotonicMs: number}[] + assert.deepEqual(events.map(event => event.stage), ["open", "build-start", "build-ready", "write-start", "write-complete"]) + for (const event of events) { + assert.deepEqual(Object.keys(event).sort(), ["atUnixMs", "monotonicMs", "stage", "streamId"]) + assert.equal(event.streamId, diagnosticId) + assert.ok(Number.isSafeInteger(event.atUnixMs) && Number.isFinite(event.monotonicMs)) + } + } finally { + flowSseDiagnostics.unsubscribe(captureStage) + clearTimeout(timeout) + httpAbort.abort() + } assert.equal(rest!.event,"sample"); assert.ok(JSON.stringify(rest!).includes("NEVER_LIVE_MIXED_USER")) let result = await read() assert.equal(result.events[0].data.bytes,300); assert.equal(result.events[0].event,"sample") @@ -62,36 +79,20 @@ export async function verifyMcpFlowLive(actor: AuthUser) { assert.equal((await read({interface:"live-in"})).events[0].data.bytes,100) await assert.rejects(read({userId:user})); await assert.rejects(read({serverId:115})); await assert.rejects(read({samples:4})); await assert.rejects(read({minutes:15})); await assert.rejects(read({skipHeavy:true})) // Actual source changes between live ticks must appear in the next event, bypassing REST cache. - let fired = false - pool.query = ((...args: unknown[]) => { - const sql = typeof args[0] === "string" ? args[0] : (args[0] as {text?:string})?.text - const pending = (originalQuery as (...args: unknown[])=>Promise)(...args) - // Inject after the bounded data projection, not its scalar preflight: - // growth between preflight and data is deliberately refused by the source guard. - if (!fired && sql?.includes("FROM flow_buckets") && sql.includes('AS "__flowPresent"')) { fired = true; return pending.then(async rows=>{ await put(114,"live-in",50,"198.18.245.4"); return rows }) } - return pending - }) as typeof pool.query - try { result = await read({samples:2}); assert.equal(fired,true); assert.equal(result.events.length,2); assert.equal(result.events[0].data.bytes,300); assert.equal(result.events[1].data.bytes,350) } - finally { pool.query = originalQuery as typeof pool.query } + // Inject after the bounded data projection. The scalar preflight checks its size. + result = await duringFlowSourceQuery(pool, ["FROM flow_buckets", 'AS "__flowPresent"'], + () => put(114, "live-in", 50, "198.18.245.4"), () => read({ samples: 2 })) + assert.equal(result.events.length, 2) + assert.equal(result.events[0].data.bytes, 300) + assert.equal(result.events[1].data.bytes, 350) // A finished sample cannot be published after authorization is revoked during its SQL read. - fired = false - pool.query = ((...args: unknown[]) => { - const sql = typeof args[0] === "string" ? args[0] : (args[0] as {text?:string})?.text - const pending = (originalQuery as (...args: unknown[])=>Promise)(...args) - if (!fired && sql?.includes("FROM flow_buckets")) { fired = true; return pending.then(async rows=>{ await setConnectionStatus(access.connection.id,"paused",owner); return rows }) } - return pending - }) as typeof pool.query - try { await assert.rejects(read()); assert.equal(fired,true) } finally { pool.query = originalQuery as typeof pool.query } + await duringFlowSourceQuery(pool, "FROM flow_buckets", + () => setConnectionStatus(access.connection.id, "paused", owner), () => assert.rejects(read())) await setConnectionStatus(access.connection.id,"active",owner) // Abort after a real source read started: a completed SQL result is still unpublished. - const duringReadAbort = new AbortController(); fired = false - pool.query = ((...args: unknown[]) => { - const sql = typeof args[0] === "string" ? args[0] : (args[0] as {text?:string})?.text - const pending = (originalQuery as (...args: unknown[])=>Promise)(...args) - if (!fired && sql?.includes("FROM flow_buckets")) { fired = true; return pending.then(rows=>{ duringReadAbort.abort(); return rows }) } - return pending - }) as typeof pool.query - try { await assert.rejects(read({},duringReadAbort.signal)); assert.equal(fired,true) } finally { pool.query = originalQuery as typeof pool.query } + const duringReadAbort = new AbortController() + await duringFlowSourceQuery(pool, "FROM flow_buckets", async () => duringReadAbort.abort(), + () => assert.rejects(read({}, duringReadAbort.signal))) const abort = new AbortController(); abort.abort(); await assert.rejects(read({},abort.signal)) let acquired = 0 try { while (tryAcquireFlowLiveSlot()) acquired++; assert.equal(acquired,4); await assert.rejects(read()) } @@ -100,7 +101,6 @@ export async function verifyMcpFlowLive(actor: AuthUser) { assert.equal((await pool.query("SELECT count(*)::int AS n FROM mcp_plans WHERE connection_id=$1",[access.connection.id])).rows[0].n,0) console.log("✓ NetFlow live: actual shared REST engine, uncached changing ticks, complete DTO paging, SQL scope/mixed user exclusion, pause/cancel publication fences, shared slots and no plans") } finally { - pool.query = originalQuery as typeof pool.query await http.close() await setConnectionStatus(access.connection.id,"paused",owner) await pool.query("DELETE FROM user_interface_bindings WHERE user_id=ANY($1::text[])",[[user,local]]) diff --git a/backend/src/test/mcp-flow-map.ts b/backend/src/test/mcp-flow-map.ts index 9f011233..6b35f80f 100644 --- a/backend/src/test/mcp-flow-map.ts +++ b/backend/src/test/mcp-flow-map.ts @@ -14,6 +14,7 @@ import { invalidateFlowCatalogCache } from "../services/traffic-flow-topology.js import { resetFlowRingsForTests, ingestParsedFlowsForServerForTests } from "../services/traffic-flow-ingest.js" import { disableRipeEnqueueForTests, resetRipeCacheForTests } from "../services/traffic-flow-ripe.js" import trafficFlowRoutes from "../routes/traffic-flow.js" +import { duringFlowSourceQuery } from "./mcp-flow-query-race.js" export async function verifyMcpFlowMap(actor: AuthUser) { const owner = { ...actor, id: "flow-map-owner", permissions: [...actor.permissions, "mm:traffic:read"] } @@ -21,7 +22,7 @@ export async function verifyMcpFlowMap(actor: AuthUser) { await pool.query("INSERT INTO servers(id,name,host,password,type) VALUES(112,'map-visible','192.0.2.112','NEVER_MAP_PASSWORD','exit-node'),(113,'NEVER_MAP_HIDDEN_SERVER','192.0.2.113','NEVER_MAP_PASSWORD','exit-node')") const access = await createMcpConnection(mcpConnectionCreateSchema.parse({ name: "Scoped map", mode: "observer", expiresAt: new Date(Date.now()+1800000).toISOString(), permissions: ["mm:traffic:read"], scope: { serverIds: [112], operationIds: ["traffic.flow.map.read"], allowGlobal: false }, limits: { callsPerMinute: 120 } }), owner) - const principal = await authenticateMcpToken(access.token), originalQuery = pool.query.bind(pool) + const principal = await authenticateMcpToken(access.token) const rest = Fastify(); rest.setValidatorCompiler(validatorCompiler); rest.setSerializerCompiler(serializerCompiler) await rest.register(trafficFlowRoutes,{prefix:"/api"}) type Report = { map: FlowMapHopsDto; pages: Record; live: boolean; forwardingVerified: boolean; tracerouteExecuted: boolean } @@ -58,16 +59,8 @@ export async function verifyMcpFlowMap(actor: AuthUser) { for (const input of [{userId:users[2]},{userId:users[3]},{serverId:113},{minutes:1441},{interface:"__all__"},{minSharePct:0},{nestedLimit:51}]) await assert.rejects(read(input)) await assert.rejects(buildFlowMapHops({serverId:112,minutes:5},{serverIds:[112],maxRows:1,authorize:async()=>{}}),/предел/) const cancelled=new AbortController();cancelled.abort();await assert.rejects(invokeMcpTool(principal,"mm_netflow_map_hops_get",{serverId:112},cancelled.signal)) - const duringQuery=async(match:string,action:()=>Promise)=>{ - let fired=false - pool.query=((...args:unknown[])=>{ - const sql=typeof args[0]==="string"?args[0]:(args[0] as {text?:string})?.text - const pending=(originalQuery as (...args:unknown[])=>Promise)(...args) - if(!fired&&sql?.includes(match)){fired=true;return pending.then(async value=>{await action();return value})} - return pending - }) as typeof pool.query - try{await assert.rejects(read());assert.equal(fired,true)}finally{pool.query=originalQuery as typeof pool.query} - } + const duringQuery = (match: string, action: () => Promise) => + duringFlowSourceQuery(pool, match, action, () => assert.rejects(read())) await duringQuery("FROM flow_buckets",async()=>{await pool.query("UPDATE servers SET host='192.0.2.212' WHERE id=112");await pool.query("UPDATE servers SET host='192.0.2.112' WHERE id=112")}) await duringQuery("FROM server_snapshots",async()=>{await pool.query("UPDATE user_interface_bindings SET interface_name='map-drift' WHERE user_id=$1",[users[0]])}) await pool.query("UPDATE user_interface_bindings SET interface_name='map-in-a' WHERE user_id=$1",[users[0]]) @@ -79,7 +72,7 @@ export async function verifyMcpFlowMap(actor: AuthUser) { assert.equal((await pool.query("SELECT count(*)::int AS n FROM mcp_plans WHERE connection_id=$1",[access.connection.id])).rows[0].n,0) console.log("✓ Scoped NetFlow map: actual REST HTTP parity, hidden/mixed/peer/pending exclusion, cache isolation, nested pages, filters/dedup/future/limits, endpoint ABA/topology/threshold drift, cancel/pause and no plans or traceroute") } finally { - pool.query=originalQuery as typeof pool.query;await setConnectionStatus(access.connection.id,"paused",owner);await rest.close() + await setConnectionStatus(access.connection.id,"paused",owner);await rest.close() await pool.query("DELETE FROM user_interface_bindings WHERE user_id=ANY($1::text[])",[users]);await pool.query("DELETE FROM app_users WHERE id=ANY($1::text[])",[users]) await pool.query("DELETE FROM flow_buckets WHERE server_id IN (112,113)");await pool.query("DELETE FROM servers WHERE id IN (112,113)") resetFlowRingsForTests();invalidateFlowCatalogCache();invalidateFlowMapHopsCache();resetRipeCacheForTests() diff --git a/backend/src/test/mcp-flow-query-race.ts b/backend/src/test/mcp-flow-query-race.ts new file mode 100644 index 00000000..407e0a18 --- /dev/null +++ b/backend/src/test/mcp-flow-query-race.ts @@ -0,0 +1,45 @@ +import assert from "node:assert/strict" +import type { Pool, PoolClient } from "pg" + +export async function duringFlowSourceQuery( + pool: Pool, + match: string | readonly string[], + afterSourceRead: () => Promise, + exercise: () => Promise, +): Promise { + const originalQuery = pool.query, originalConnect = pool.connect + const fragments = typeof match === "string" ? [match] : match + let scheduled = false, injected = false + const afterRead = (args: unknown[], pending: Promise) => { + const sql = typeof args[0] === "string" ? args[0] : (args[0] as { text?: string } | undefined)?.text + if (scheduled || !sql || !fragments.every(fragment => sql.includes(fragment))) return pending + scheduled = true + return pending.then(async result => { + await afterSourceRead() + injected = true + return result + }) + } + pool.query = ((...args: unknown[]) => afterRead(args, Reflect.apply(originalQuery, pool, args))) as typeof originalQuery + pool.connect = (function (this: Pool, ...args: unknown[]) { + if (args.length) return Reflect.apply(originalConnect, this, args) + return (Reflect.apply(originalConnect, this, []) as Promise).then(client => { + const query = client.query, release = client.release + client.query = ((...queryArgs: unknown[]) => afterRead(queryArgs, Reflect.apply(query, client, queryArgs))) as typeof query + client.release = (...releaseArgs) => { + client.query = query + client.release = release + release.apply(client, releaseArgs) + } + return client + }) + }) as typeof originalConnect + try { + const result = await exercise() + assert.ok(injected, "Race fixture must inject the change after the real NetFlow source query") + return result + } finally { + pool.connect = originalConnect + pool.query = originalQuery + } +}