fix(netflow): связать отметки потока SSE с диагностикой NetFlow

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()=>{
+        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 {

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)

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<string, unknown> = {}) => invokeMcpTool(principal, "mm_netflow_analytics_report_get", { serverId: 110, minutes: 5, ...extra }) as Promise<{ analytics: FlowAnalyticsDto; pages: Record<string, { total: number; nextOffset: number | null }>; 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<unknown>) => {
+    const duringQuery = (match: string, action: () => Promise<unknown>) =>
-      let fired = false
+      duringFlowSourceQuery(pool, match, action, () => assert.rejects(read()))
-      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<unknown>)(...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 }
-    }
     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)")

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<string, { total: number; nextOffset: number | null }> }>; 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
+    // Inject after the bounded data projection. The scalar preflight checks its size.
-    pool.query = ((...args: unknown[]) => {
+    result = await duringFlowSourceQuery(pool, ["FROM flow_buckets", 'AS "__flowPresent"'],
-      const sql = typeof args[0] === "string" ? args[0] : (args[0] as {text?:string})?.text
+      () => put(114, "live-in", 50, "198.18.245.4"), () => read({ samples: 2 }))
-      const pending = (originalQuery as (...args: unknown[])=>Promise<unknown>)(...args)
+    assert.equal(result.events.length, 2)
-      // Inject after the bounded data projection, not its scalar preflight:
+    assert.equal(result.events[0].data.bytes, 300)
-      // growth between preflight and data is deliberately refused by the source guard.
+    assert.equal(result.events[1].data.bytes, 350)
-      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 }
     // A finished sample cannot be published after authorization is revoked during its SQL read.
-    fired = false
+    await duringFlowSourceQuery(pool, "FROM flow_buckets",
-    pool.query = ((...args: unknown[]) => {
+      () => setConnectionStatus(access.connection.id, "paused", owner), () => assert.rejects(read()))
-      const sql = typeof args[0] === "string" ? args[0] : (args[0] as {text?:string})?.text
-      const pending = (originalQuery as (...args: unknown[])=>Promise<unknown>)(...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 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
+    const duringReadAbort = new AbortController()
-    pool.query = ((...args: unknown[]) => {
+    await duringFlowSourceQuery(pool, "FROM flow_buckets", async () => duringReadAbort.abort(),
-      const sql = typeof args[0] === "string" ? args[0] : (args[0] as {text?:string})?.text
+      () => assert.rejects(read({}, duringReadAbort.signal)))
-      const pending = (originalQuery as (...args: unknown[])=>Promise<unknown>)(...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 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]])

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<string, { total: number; nextOffset: number | null }>; 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<unknown>)=>{
+    const duringQuery = (match: string, action: () => Promise<unknown>) =>
-      let fired=false
+      duringFlowSourceQuery(pool, match, action, () => assert.rejects(read()))
-      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<unknown>)(...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}
-    }
     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()

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<T>(
+  pool: Pool,
+  match: string | readonly string[],
+  afterSourceRead: () => Promise<unknown>,
+  exercise: () => Promise<T>,
+): Promise<T> {
+  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<unknown>) => {
+    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<PoolClient>).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
+  }
+}