feat(mcp): подготовить сервер и доступ к приёмке

backend/src/services/server-poll-durable-session.sql.test.ts

@@ -0,0 +1,134 @@
+import { MIGRATION_MANIFEST } from "../db/migration-manifest.js"
+import assert from "node:assert/strict"
+import { randomUUID } from "node:crypto"
+import { createServer } from "node:http"
+import { basename, dirname, resolve } from "node:path"
+import { readFile } from "node:fs/promises"
+import { gzipSync, gunzipSync } from "node:zlib"
+import type { DurableServerPollSessionInput } from "./server-poll-durable-session.js"
+
+// Never load the configured pool before admitting a private owned DB.
+assert.equal(process.env.MMAPP_OWN_SERVER_POLL_DURABLE,"true")
+const endpoint=new URL(process.env.DATABASE_URL??"")
+assert.equal(endpoint.hostname,"127.0.0.1");assert.equal(endpoint.pathname,"/postgres")
+assert(["postgres:","postgresql:"].includes(endpoint.protocol))
+assert(process.env.MCP_STATE_DIR&&basename(dirname(resolve(process.env.MCP_STATE_DIR))).startsWith("mm-mcp-test-"))
+const {pool,closePool}=await import("../db/index.js")
+const {applySqlMigrations}=await import("../db/migrate.js")
+const {createStorageDurableServerPollSession:createDurableServerPollSession}=await import("./server-poll-durable-session.js")
+const {SERVER_PARTITION_WRITER_LOCK}=await import("./server-partition-maintenance-lease.js")
+const {withPartitionWrite,PartitionWriteError}=await import("./partition-write-transaction.js")
+const {ensurePartitionFor,PARTITION_SPECS}=await import("../db/partitions.js")
+const security=await import("../modules/mcp/security.js")
+let serverId:number|undefined,requests=0,releasedNative=false,writerPid=0
+let releaseNative!:()=>void,firstNative!:()=>void
+const nativeGate=new Promise<void>(resolve=>{releaseNative=()=>{releasedNative=true;resolve()}})
+const started=new Promise<void>(resolve=>{firstNative=resolve})
+const paths:string[]=[]
+const listener=createServer(async(req,res)=>{
+  requests++;paths.push(req.url??"");firstNative()
+  await nativeGate
+  res.setHeader("content-type","application/json")
+  const value=req.url==="/rest/system/identity"?{name:"poll-owned-renamed"}:req.url==="/rest/system/resource"?{version:"7.synthetic","board-name":"owned","uptime":"1h","cpu-load":"3","free-memory":"1024","total-memory":"2048"}:req.url==="/rest/interface"?[{name:"owned-interface",type:"ether"}]:req.url==="/rest/ip/address"?[{address:"192.0.2.1/24",interface:"owned-interface"}]:null
+  res.statusCode=value===null?404:200;res.end(JSON.stringify(value))
+})
+const waitFor=async(label:string,inspect:()=>Promise<boolean>)=>{const deadline=Date.now()+10000;while(Date.now()<deadline){if(await inspect())return;await new Promise(resolve=>setTimeout(resolve,10))};assert.fail("Timed out: "+label)}
+try{
+  await applySqlMigrations(pool)
+  await applySqlMigrations(pool)
+  assert.equal((await pool.query("SELECT count(*)::int AS n FROM schema_migrations")).rows[0].n,MIGRATION_MANIFEST.length)
+  assert.equal((await pool.query("SELECT count(*)::int AS n FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='public' AND c.relname=ANY($1) AND c.relkind='r'",[["server_poll_executions","server_poll_intents","server_poll_acks","server_poll_receipts","server_poll_quarantines"]])).rows[0].n,5)
+  assert.equal((await pool.query("SELECT count(*)::int AS n FROM pg_trigger t JOIN pg_class c ON c.oid=t.tgrelid WHERE t.tgname='server_partition_writer_guard' AND t.tgenabled='A' AND c.relkind='p'")).rows[0].n,PARTITION_SPECS.length)
+  assert.equal((await pool.query("SELECT count(*)::int AS n FROM servers")).rows[0].n,0,"Fresh isolated database required; never remove foreign fixtures")
+  await new Promise<void>((resolve,reject)=>{listener.once("error",reject);listener.listen(0,"127.0.0.1",()=>{listener.off("error",reject);resolve()})})
+  const port=(listener.address()as{port:number}).port
+  serverId=Number((await pool.query("INSERT INTO servers(name,host,port,enabled,username,password,use_ssl,verify_ssl) VALUES('poll-owned-before','127.0.0.1',$1,true,'SYNTHETIC','SYNTHETIC',false,true) RETURNING id",[port])).rows[0].id)
+  const oldAt=new Date(Date.now()-86400000)
+  await ensurePartitionFor(pool,"server_snapshots","week",oldAt)
+  await ensurePartitionFor(pool,"server_snapshots","week",new Date())
+  const expired=await ensurePartitionFor(pool,"traffic_samples","week",new Date("2000-01-03T00:00:00Z"))
+  const oldId=Number((await pool.query("INSERT INTO server_snapshots(server_id,polled_at,status,raw_interfaces,raw_ip_addresses) VALUES($1,$2,'offline','[]'::jsonb,'[]'::jsonb) RETURNING id",[serverId,oldAt])).rows[0].id)
+  const issuedAt=Date.now()-1000,expiresAt=Date.now()+600000,ownerId="owned-local-poll-"+randomUUID(),receiptId=randomUUID(),executionId=randomUUID()
+  const identity={kind:"rest"as const,authentication:"local"as const,ownerId,executionId,requestId:randomUUID(),issuedAt,expiresAt}
+  const generation=await security.accessGeneration(),initialEpoch=(await pool.query("SELECT routing_endpoint_epoch FROM servers WHERE id=$1",[serverId])).rows[0].routing_endpoint_epoch
+  const authorityProof=await security.privateSecretFingerprint(security.fingerprint({ownerId,issuedAt,expiresAt,generation,serverId,initialEpoch,port,permissions:["mm:servers:write"],global:true}),"owned-full-poll-authority-v1")
+  let revoked=false,authorityReads=0
+  const sessionRequest={serverId,receiptId}
+  const sessionInput:DurableServerPollSessionInput={request:sessionRequest,identity,authorize:async(request,required,signal)=>{
+    signal?.throwIfAborted();authorityReads++
+    assert(!revoked&&Date.now()>=issuedAt&&Date.now()<expiresAt,"Owned accepted local actor expired/revoked")
+    assert.equal(request.serverId,serverId);assert.equal(request.receiptId,receiptId)
+    // The already-provisioned generation is freshly read on every boundary.
+    // Avoid repeatedly running the create-if-absent provisioning path in this
+    // hot fixture callback; no generation/revocation decision is cached.
+    assert.equal((await readFile(resolve(security.mcpStateDirectory(),"access-generation"))).toString("hex"),generation)
+    const rows=(await pool.query("SELECT id,host,port,enabled,routing_endpoint_epoch,ARRAY(SELECT id FROM servers ORDER BY id LIMIT 1001) AS catalogue,(SELECT count(*)::int FROM mcp_protections) AS protections FROM servers ORDER BY id LIMIT 1001")).rows
+    assert.equal(rows.length,1);assert.deepEqual(rows.map(row=>Number(row.id)),[serverId])
+    assert.equal(rows[0].host,"127.0.0.1");assert.equal(rows[0].port,port);assert.equal(rows[0].enabled,true);assert.equal(rows[0].routing_endpoint_epoch,initialEpoch)
+    assert.deepEqual(rows[0].catalogue.map(Number),[serverId]);assert(required.every(id=>id===serverId));assert.equal(rows[0].protections,0,"Fixture refuses any protection rather than bypassing it")
+    // Every immutable grant component is freshly checked above; reuse only
+    // its keyed identifier, never a cached admission/revocation decision.
+    return {global:true,serverIds:[serverId!],authorityProof}
+  }}
+  const session=createDurableServerPollSession(sessionInput)
+  const plan=await session.prepare()
+  console.log("Full durable poll prepared:",JSON.stringify({maintenanceEffects:plan.maintenance.length,parents:PARTITION_SPECS.length}))
+  assert.equal(requests,0)
+  assert.equal(new Set(plan.maintenance.map(effect=>"parent"in effect?effect.parent:"")).size,PARTITION_SPECS.length)
+  assert(plan.maintenance.some(effect=>effect.kind==="partition.drop"&&effect.name===expired))
+  const pending=session.execute(plan);void pending.catch(()=>{})
+  await Promise.race([started,pending.then(()=>assert.fail("Native request must occur before completion"))])
+  // Actual ordinary writer arrives while FULL poll holds the exclusive lease.
+  // Inspect lock evidence before releasing HTTP; no target relation is touched.
+  const controller=new AbortController()
+  const blocked=withPartitionWrite(async owner=>{await owner.client.query("INSERT INTO traffic_samples(server_id,interface_name,sampled_at,rx_bytes) VALUES($1,'must-not-dispatch',now(),1)",[serverId])},{signal:controller.signal,connect:async()=>{const client=await pool.connect();writerPid=Number((await client.query("SELECT pg_backend_pid() AS pid")).rows[0].pid);return client}})
+  void blocked.catch(()=>{})
+  try{
+    await waitFor("ordinary writer waiting before target SQL",async()=>writerPid>0&&(await pool.query("SELECT count(*)::int AS n FROM pg_locks WHERE pid=$1 AND locktype='advisory' AND objid=$2 AND mode='ShareLock' AND NOT granted",[writerPid,SERVER_PARTITION_WRITER_LOCK])).rows[0].n===1)
+    assert.equal((await pool.query("SELECT count(*)::int AS n FROM pg_locks l JOIN pg_class c ON c.oid=l.relation WHERE l.pid=$1 AND l.locktype='relation' AND (c.relname='traffic_samples' OR c.relname LIKE 'traffic_samples_w_%')",[writerPid])).rows[0].n,0)
+  }finally{controller.abort();await assert.rejects(blocked,(error:unknown)=>error instanceof PartitionWriteError&&!error.outcomeUnknown);releaseNative()}
+  const receipt=await pending
+  assert.equal(receipt.complete,true);assert.equal(receipt.maintenanceComplete,true);assert.equal(receipt.nativeReads,4)
+  assert.equal(receipt.completedPublications,plan.maintenance.length+4)
+  assert.deepEqual([...paths].sort(),["/rest/interface","/rest/ip/address","/rest/system/identity","/rest/system/resource"].sort())
+  assert.equal((await pool.query("SELECT name FROM servers WHERE id=$1",[serverId])).rows[0].name,"poll-owned-renamed")
+  const old=(await pool.query("SELECT raw_interfaces,raw_ip_addresses FROM server_snapshots WHERE id=$1",[oldId])).rows[0]
+  assert.equal(old.raw_interfaces,null);assert.equal(old.raw_ip_addresses,null)
+  assert.equal((await pool.query("SELECT to_regclass($1) AS name",["public."+expired])).rows[0].name,null)
+  assert.equal((await pool.query("SELECT count(*)::int AS n FROM server_poll_intents WHERE receipt_id=$1",[receiptId])).rows[0].n,receipt.completedPublications+4)
+  assert.equal((await pool.query("SELECT count(*)::int AS n FROM server_poll_acks WHERE receipt_id=$1",[receiptId])).rows[0].n,receipt.completedPublications+4)
+  assert.equal((await pool.query("SELECT count(*)::int AS n FROM server_poll_receipts WHERE receipt_id=$1",[receiptId])).rows[0].n,1)
+  // Measure a COMPLETE real journal, including ciphertext, proofs, timestamps,
+  // admission and every stage. This is sizing evidence, not an archive format
+  // or permission to delete/compact rows. Never print the captured payload.
+  const stored=(await pool.query(`SELECT jsonb_build_object(
+    'admission',(SELECT to_jsonb(a) FROM server_poll_storage_admissions a WHERE receipt_id=$1),
+    'execution',(SELECT to_jsonb(e) FROM server_poll_executions e WHERE receipt_id=$1),
+    'intents',(SELECT jsonb_agg(to_jsonb(i) ORDER BY created_at,stage) FROM server_poll_intents i WHERE receipt_id=$1),
+    'acks',(SELECT jsonb_agg(to_jsonb(a) ORDER BY created_at,stage) FROM server_poll_acks a WHERE receipt_id=$1),
+    'receipt',(SELECT to_jsonb(r) FROM server_poll_receipts r WHERE receipt_id=$1),
+    'quarantine',(SELECT to_jsonb(q) FROM server_poll_quarantines q WHERE receipt_id=$1)) AS value`,[receiptId])).rows[0].value
+  assert(stored.admission&&stored.execution&&stored.receipt);assert.equal(stored.quarantine,null)
+  assert.equal(stored.intents.length,receipt.completedPublications+4);assert.equal(stored.acks.length,stored.intents.length)
+  const journalRows=stored.intents.length+stored.acks.length+2
+  assert.equal(stored.admission.used_rows,journalRows)
+  const fullBytes=Buffer.from(JSON.stringify(stored)),compressed=gzipSync(fullBytes,{level:6})
+  assert.deepEqual(gunzipSync(compressed),fullBytes,"Sizing must retain every original field and byte")
+  console.log("Full durable poll journal sizing:",JSON.stringify({maintenanceEffects:plan.maintenance.length,journalRows,admissionRows:1,
+    chargedBytes:Number(stored.admission.used_bytes),fullJsonBytes:fullBytes.length,gzipBytes:compressed.length,gzipRoundTripExact:true,
+    sampleFits128KiB:compressed.length<=131072,identityKind:stored.admission.identity_kind,compactionPerformed:false,
+    archiveAuthenticationAccepted:false,historicalVerifierAccepted:false,productionCapacityAccepted:false}))
+  sessionRequest.serverId++
+  identity.ownerId="mutated-outer-input";identity.expiresAt=0
+  sessionInput.authorize=async()=>{throw Error("Outer caller replaced authority provider")}
+  const repeated=await session.execute(plan);assert.equal(repeated.complete,true);assert.equal(requests,4)
+  await assert.rejects(session.execute({...plan,bindingProof:"0".repeat(64)}));assert.equal(requests,4)
+  await withPartitionWrite(async owner=>{await owner.client.query("INSERT INTO traffic_samples(server_id,interface_name,sampled_at,rx_bytes) VALUES($1,'after-full-poll',now(),1)",[serverId])})
+  revoked=true;await assert.rejects(session.execute(plan));assert.equal(requests,4)
+  console.log("Full durable poll actual SQL/HTTP acceptance:",JSON.stringify({fourNativeGets:requests,maintenanceEffects:plan.maintenance.length,parents:PARTITION_SPECS.length,completedPublications:receipt.completedPublications,sameBackendLeaseAndTransactions:true,ordinaryWriterBlockedBeforeRelationSql:true,realJournal:true,repeatedApplyNoNewNative:true,outerInputMutationCannotReplaceCapturedIdentityOrAuthority:true,changedPlanBindingRefused:true,lateRevocationRefused:true,actualHttpAuthenticationMiddleware:false,productionRegistration:false,externalNetwork:false,authorityReads}))
+}finally{
+  if(!releasedNative)releaseNative()
+  listener.closeAllConnections();if(listener.listening)await new Promise<void>(resolve=>listener.close(()=>resolve()))
+  if(serverId!==undefined)await pool.query("UPDATE servers SET enabled=false WHERE id=$1",[serverId])
+  await closePool()
+}