fix(netflow): читать активные записи на одном снимке

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

@@ -1,4 +1,4 @@
-import { flowReadDbAll } from "./traffic-flow-read-policy.js"
+import { flowReadBucketSnapshot } from "./traffic-flow-read-policy.js"
 import {flowIngestQualitySchema} from "@mmapp/contracts/traffic-flow"
 import { flowLifecycleBoundary, type FlowLifecycleAuthorization } from "./traffic-flow-lifecycle-fence.js"
 import { flowCounter } from "./traffic-flow-observed-values.js"
@@ -483,7 +483,7 @@ export async function listStoredFlowRows(sinceIso: string, serverId?: number, op
   const parameters: Array<string | number> = serverId === undefined ? [sinceIso, keep] : [sinceIso, keep, serverId]
   const untilParameter = options?.untilIso ? parameters.push(options.untilIso) : undefined
   if (options?.maxRows !== undefined) parameters.push(options.maxRows + 1)
-  const stored = await flowReadDbAll<StoredBucketRow>(`
+  const stored = await flowReadBucketSnapshot<StoredBucketRow>(`
     SELECT
       server_id, bucket_at, COALESCE(host(src), '') AS src, COALESCE(host(dst), '') AS dst, proto,
       src_port, dst_port, bytes, packets, in_iface, out_iface,

backend/src/services/traffic-flow-read-http.owned.test.ts

@@ -64,6 +64,46 @@ try{
  }
  await app.register(trafficFlowRoutes,{prefix:'/api'});await app.register(mcpTransport)
  const endpoint=await app.listen({host:'127.0.0.1',port:0})
+ const flowUrl=endpoint+'/api/traffic/flow?range=15m'
+ const before=await fetch(flowUrl)
+ assert.equal(before.status,200)
+ const originalTalkers=(await before.json() as {talkers:unknown}).talkers
+ const withBucketMeasurement=async(action:()=>Promise<void>,read:()=>Promise<void>)=>{
+  const connect=pool.connect
+  let observed=false
+  pool.connect=(function(this:typeof pool,...args:unknown[]){
+   if(args.length)return Reflect.apply(connect,this,args)
+   return (Reflect.apply(connect,this,[]) as ReturnType<typeof connect>).then(connection=>{
+    const query=connection.query,release=connection.release
+    connection.query=(async (...queryArgs:unknown[])=>{
+     const result=await Reflect.apply(query,connection,queryArgs)
+     const statement=queryArgs[0]
+     if(!observed&&typeof statement==='string'&&statement.startsWith('WITH flow_read_source AS MATERIALIZED')&&statement.includes('FROM flow_buckets fb')){
+      observed=true
+      await action()
+     }
+     return result
+    }) as typeof query
+    connection.release=(...releaseArgs)=>{
+     connection.query=query;connection.release=release;release.apply(connection,releaseArgs)
+    }
+    return connection
+   })
+  }) as typeof connect
+  try{await read();assert(observed,'The request must reach the source measurement')}finally{pool.connect=connect}
+ }
+ await withBucketMeasurement(async()=>{
+  await pool.query("INSERT INTO flow_buckets(server_id,bucket_at,src,dst,proto,src_port,dst_port,bytes,packets) VALUES(1,$1,'198.18.1.2','8.8.8.8',6,2,443,200,1)",[bucket])
+ },async()=>{
+  const overlapped=await fetch(flowUrl)
+  assert.equal(overlapped.status,200)
+  assert.deepEqual((await overlapped.json() as {talkers:unknown}).talkers,originalTalkers)
+ })
+ const after=await fetch(flowUrl)
+ assert.equal(after.status,200)
+ assert.notDeepEqual((await after.json() as {talkers:unknown}).talkers,originalTalkers)
+ await pool.query("DELETE FROM flow_buckets WHERE server_id=1 AND bucket_at=$1 AND src='198.18.1.2'::inet",[bucket])
+ console.log('PASS REST NetFlow retains one bounded bucket snapshot while the collector commits concurrently')
  client=new Client({name:'owned-netflow-read',version:'1'})
  await client.connect(new StreamableHTTPClientTransport(new URL(endpoint+'/mcp'),{requestInit:{headers:{authorization:'Bearer '+access.token}}}))
  const call=()=>client!.callTool({name:'mm_netflow_talkers_get',arguments:{serverId:1}})
@@ -81,22 +121,19 @@ try{
   assert.equal((globalResult.structuredContent as {result:{dataQuality:{bytes:string}}}).result.dataQuality.bytes,'987654')
  }finally{await globalClient.close()}
  console.log('PASS actual MCP scoped collector counters withheld; existing explicit global telemetry grant receives genuine synthetic receiver quality')
- // The installed SDK's legacy stateless path accepts multiple valid calls per HTTP response.
+ // The production admission guard rejects multi-call batches before SDK dispatch.
  const batch=await fetch(endpoint+'/mcp',{method:'POST',headers:{authorization:'Bearer '+access.token,'content-type':'application/json',accept:'application/json, text/event-stream','mcp-protocol-version':'2025-06-18'},body:JSON.stringify([1,2].map(id=>({jsonrpc:'2.0',id,method:'tools/call',params:{name:'mm_netflow_talkers_get',arguments:{serverId:1}}})))})
- assert.equal(batch.status,200);const batchText=await batch.text()
+ assert.equal(batch.status,400)
- const messages=batchText.split('\n').filter(line=>line.startsWith('data: ')).map(line=>JSON.parse(line.slice(6)))
+ const rejectedBatch=await batch.json() as {id:unknown;error:{code:number}}
- assert.equal(messages.length,2,batchText);assert.ok(messages.every(message=>!message.error&&!message.result.isError),batchText)
+ assert.equal(rejectedBatch.id,null)
- assert.ok(!batchText.includes('FORBIDDEN_'))
+ assert.equal(rejectedBatch.error.code,-32600)
  result=await client.callTool({name:'mm_netflow_talkers_get',arguments:{serverId:2}});assert.equal(result.isError,true)
  // Revocation during a real bounded source statement must prevent publication.
- const original=pool.query.bind(pool);let revoked=false
+ await withBucketMeasurement(async()=>{await setConnectionStatus(access.connection.id,'paused',actor)},async()=>{
- pool.query=((...args:unknown[])=>{
+  result=await call()
-  const text=typeof args[0]==='string'?args[0]:(args[0] as {text?:string})?.text
+  assert.equal(result.isError,true)
-  const pending=(original as (...args:unknown[])=>Promise<unknown>)(...args)
+  assert.ok(!JSON.stringify(result).includes('198.18.1.1'))
-  if(!revoked&&text?.includes('flow_read_source')&&text.includes('FROM flow_buckets')){revoked=true;return pending.then(async rows=>{await setConnectionStatus(access.connection.id,'paused',actor);return rows})}
+ })
-  return pending
- }) as typeof pool.query
- try{result=await call();assert.equal(revoked,true);assert.equal(result.isError,true);assert.ok(!JSON.stringify(result).includes('198.18.1.1'))}finally{pool.query=original as typeof pool.query}
  await setConnectionStatus(access.connection.id,'active',actor)
  // Four actual REST SSE streams retain their shared global owners. MCP and REST
  // share the same admission pool; a rejected fifth read allocates no flow source.
@@ -115,7 +152,7 @@ try{
  for(let i=0;i<100&&flowReadBudget.stats().owners;i++)await delay(20)
  assert.equal(flowReadBudget.stats().owners,0)
  result=await call();assert.ok(!result.isError)
- console.log('PASS actual production HTTP SDK: scoped source/no hidden rows, two-call legacy batch sharing one owner, source-time revocation, four REST SSE owners versus MCP/REST rejection, abort cleanup and successful readmission')
+ console.log('PASS actual production HTTP SDK: scoped source/no hidden rows, multi-call batch refusal, source-time revocation, four REST SSE owners versus MCP/REST rejection, abort cleanup and successful readmission')
 }finally{await client?.close();console.log('Owned HTTP SDK client closed');app.server.closeAllConnections();await app.close();console.log('Owned HTTP Fastify server closed');await closePool();console.log('Owned HTTP application pool closed')}
 // Imported application maintenance timers are outside this isolated read test.
 // Required transport/server/pool cleanup above must complete before terminal exit.

backend/src/services/traffic-flow-read-policy.ts

@@ -1,7 +1,7 @@
 import { getFlowReadResourceProfile } from "./traffic-flow-read-resource-profile.js"
 import { bindSql } from "../db/sql-bind.js"
 import { pool } from "../db/index.js"
-import { createFlowReadBudget } from "./traffic-flow-read-budget.js"
+import { createFlowReadBudget, trackFlowReadComputation } from "./traffic-flow-read-budget.js"
 import { readBoundedFlowSql } from "./traffic-flow-bounded-sql.js"
 import type { QueryResultRow } from "pg"
 const flowReadResourceProfile=getFlowReadResourceProfile()
@@ -11,7 +11,37 @@ export const FLOW_READ_RESERVATION_BYTES=flowReadResourceProfile.reservationByte
 export const FLOW_SSE_MAX_FRAME_BYTES=flowReadResourceProfile.responseOutputBytes
 export const flowReadBudget=createFlowReadBudget({maxOwners:flowReadResourceProfile.maxOwners,maxReservedBytes:flowReadResourceProfile.maxOwners*FLOW_READ_RESERVATION_BYTES})
 export const flowResponseReadLimits={maxOwners:1,maxReservedBytes:FLOW_READ_RESERVATION_BYTES}
-export async function flowReadDbAll<T extends QueryResultRow=QueryResultRow>(sql:string,params?:unknown[]|Record<string,unknown>):Promise<T[]> {
+const sqlSourceLimits = { maxRows: 50000, maxEncodedBytes: 16 * 1024 * 1024 }
+
+export async function flowReadDbAll<T extends QueryResultRow = QueryResultRow>(
+  sql: string, params?: unknown[] | Record<string, unknown>,
+): Promise<T[]> {
   const bound = bindSql(sql, params)
- return readBoundedFlowSql<T>(pool,bound.text,bound.values??[],{maxRows:50000,maxEncodedBytes:16*1024*1024})
+  return readBoundedFlowSql<T>(pool, bound.text, bound.values ?? [], sqlSourceLimits)
+}
+
+/** A collector commit must not change the bucket rows between budget measurement and decode. */
+export async function flowReadBucketSnapshot<T extends QueryResultRow>(sql: string, params: unknown[]): Promise<T[]> {
+  const bound = bindSql(sql, params)
+  return trackFlowReadComputation(async () => {
+    const client = await pool.connect()
+    let discard = true
+    try {
+      await client.query("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY")
+      const rows = await readBoundedFlowSql<T>(client, bound.text, bound.values ?? [], sqlSourceLimits)
+      await client.query("COMMIT")
+      discard = false
+      return rows
+    } catch (error) {
+      try {
+        await client.query("ROLLBACK")
+        discard = false
+      } catch {
+        // A connection with an uncertain transaction state must not be reused.
+      }
+      throw error
+    } finally {
+      client.release(discard)
+    }
+  })
 }