210 lines
20 KiB
TypeScript
210 lines
20 KiB
TypeScript
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 {compactCompletedServerPollStorage}=await import("./server-poll-terminal-compaction.js")
|
|
const {collectExpiredServerPollStorage}=await import("./server-poll-storage-retention.js")
|
|
const {tickMcpWorkers}=await import("../modules/mcp/workers.js")
|
|
const {ServerPollJournalOutcomeUnknown}=await import("./server-poll-journal.js")
|
|
const {Client}=await import("pg")
|
|
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 pool.query(`CREATE TABLE owned_poll_journal_pid(relation text,pid int,xid text,exclusive_writer boolean);
|
|
CREATE FUNCTION owned_poll_journal_observe() RETURNS trigger LANGUAGE plpgsql AS $$BEGIN
|
|
INSERT INTO public.owned_poll_journal_pid SELECT TG_TABLE_NAME,pg_backend_pid(),pg_current_xact_id()::text,
|
|
EXISTS(SELECT 1 FROM pg_catalog.pg_locks WHERE pid=pg_backend_pid() AND locktype='advisory' AND classid=0
|
|
AND objid=8723110 AND objsubid=1 AND mode='ExclusiveLock' AND granted);
|
|
RETURN NULL;
|
|
END$$;`)
|
|
for(const name of ["server_poll_storage_admissions","server_poll_storage_usage","server_poll_executions","server_poll_intents","server_poll_acks","server_poll_receipts","server_poll_quarantines"]){
|
|
await pool.query(`CREATE TRIGGER candidate_application_statement_guard BEFORE INSERT OR UPDATE OR DELETE ON public.${name}
|
|
FOR EACH STATEMENT EXECUTE FUNCTION public.server_partition_writer_guard();
|
|
ALTER TABLE public.${name} ENABLE ALWAYS TRIGGER candidate_application_statement_guard;
|
|
CREATE TRIGGER owned_poll_journal_observe AFTER INSERT OR UPDATE OR DELETE ON public.${name}
|
|
FOR EACH STATEMENT EXECUTE FUNCTION public.owned_poll_journal_observe();`)
|
|
}
|
|
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"))])
|
|
const {rows:[holder]}=await pool.query(`SELECT pid FROM pg_catalog.pg_locks WHERE locktype='advisory' AND classid=0
|
|
AND objid=$1::oid AND objsubid=1 AND mode='ExclusiveLock' AND granted`,[SERVER_PARTITION_WRITER_LOCK])
|
|
assert(Number.isInteger(holder?.pid))
|
|
const prefix=(await pool.query("SELECT relation,pid,xid,exclusive_writer FROM owned_poll_journal_pid")).rows
|
|
assert(prefix.some(row=>row.relation==="server_poll_executions"));assert(prefix.some(row=>row.relation==="server_poll_intents"))
|
|
assert(prefix.every(row=>row.pid===holder.pid&&row.exclusive_writer===true))
|
|
const anchor=prefix.find(row=>row.relation==="server_poll_executions")!
|
|
assert(prefix.some(row=>row.relation==="server_poll_storage_admissions"&&row.xid!==anchor.xid),"Reservation and consume/anchor are independently committed")
|
|
assert(prefix.some(row=>row.relation==="server_poll_storage_admissions"&&row.xid===anchor.xid),"Consume and anchor share one transaction")
|
|
const workerErrors:unknown[]=[]
|
|
await tickMcpWorkers().catch(error=>workerErrors.push(error))
|
|
assert.deepEqual(workerErrors,[],"An actual worker tick must defer maintenance without an error while native POLL owns the exclusive lock")
|
|
assert.equal((await compactCompletedServerPollStorage()).compacted,0)
|
|
assert.equal((await collectExpiredServerPollStorage()).collected,0)
|
|
assert.equal((await pool.query("SELECT count(*)::int AS n FROM pg_locks WHERE locktype='advisory' AND pid=$1 AND objid=$2::oid AND mode='ExclusiveLock' AND granted",[holder.pid,SERVER_PARTITION_WRITER_LOCK])).rows[0].n,1)
|
|
assert.equal((await pool.query("SELECT pg_try_advisory_xact_lock(hashtextextended('server-poll-storage-v1',0)) AS free")).rows[0].free,true)
|
|
// 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((await compactCompletedServerPollStorage()).compacted,0)
|
|
assert.equal((await collectExpiredServerPollStorage()).collected,0)
|
|
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)
|
|
const fullJournal=(await pool.query("SELECT relation,pid,xid,exclusive_writer FROM owned_poll_journal_pid")).rows
|
|
assert(fullJournal.every(row=>row.pid===holder.pid&&row.exclusive_writer===true))
|
|
assert(fullJournal.some(row=>row.relation==="server_poll_acks")&&fullJournal.some(row=>row.relation==="server_poll_receipts"))
|
|
assert(new Set(fullJournal.filter(row=>row.relation==="server_poll_intents").map(row=>row.xid)).size===receipt.completedPublications+4)
|
|
assert.deepEqual(await session.readExisting(plan),receipt)
|
|
// 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}))
|
|
const lostRequest={serverId,receiptId:randomUUID()},lostIdentity={...identity,executionId:randomUUID(),requestId:randomUUID()}
|
|
const originalAuthority=sessionInput.authorize
|
|
const lostSession=createDurableServerPollSession({request:lostRequest,identity:lostIdentity,authorize:async(request,ids,signal)=>{
|
|
assert.deepEqual(request,lostRequest)
|
|
return originalAuthority(sessionRequest,ids,signal)
|
|
}})
|
|
const lostPlan=await lostSession.prepare(),originalQuery=Client.prototype.query
|
|
type IdentifiedClient=InstanceType<typeof Client>&{processID?:number}
|
|
let reservedBackend:IdentifiedClient|undefined,acksLost=0
|
|
try{
|
|
Client.prototype.query=function(this:InstanceType<typeof Client>,...args:unknown[]){
|
|
const request=args[0] as {text?:string;values?:unknown[]}
|
|
if(request?.text?.startsWith("INSERT INTO server_poll_storage_admissions")&&request.values?.[0]===lostRequest.receiptId)reservedBackend=this
|
|
if(this===reservedBackend&&request?.text==="COMMIT"&&acksLost===0){
|
|
return Promise.resolve(Reflect.apply(originalQuery,this,args)).then((result:{command:string})=>{
|
|
assert.equal(result.command,"COMMIT");acksLost++;throw Error("owned real reservation COMMIT ACK lost")
|
|
})
|
|
}
|
|
return Reflect.apply(originalQuery,this,args)
|
|
} as typeof Client.prototype.query
|
|
await assert.rejects(lostSession.execute(lostPlan),error=>error instanceof ServerPollJournalOutcomeUnknown&&error.automaticReplay===false&&!error.requiresDurableQuarantine)
|
|
}finally{Client.prototype.query=originalQuery}
|
|
assert.equal(acksLost,1);assert.equal(requests,4)
|
|
const {rows:[lostState]}=await pool.query(`SELECT
|
|
(SELECT count(*)::int FROM server_poll_storage_admissions WHERE receipt_id=$1) AS charged,
|
|
(SELECT count(*)::int FROM server_poll_executions WHERE receipt_id=$1) AS anchored,
|
|
(SELECT count(*)::int FROM server_poll_quarantines WHERE receipt_id=$1) AS quarantined`,[lostRequest.receiptId])
|
|
assert.deepEqual(lostState,{charged:1,anchored:0,quarantined:1})
|
|
const {rows:[quarantine]}=await pool.query("SELECT pid,exclusive_writer FROM owned_poll_journal_pid WHERE relation='server_poll_quarantines'")
|
|
assert(Number.isInteger(reservedBackend?.processID))
|
|
assert.notEqual(quarantine.pid,reservedBackend?.processID);assert.equal(quarantine.exclusive_writer,false)
|
|
assert.equal(await lostSession.readExisting(lostPlan),undefined)
|
|
console.log("Guarded durable POLL journal acceptance:",JSON.stringify({sameHolder:true,independentCommits:true,concurrentMaintenanceNoEarlyChanges:true,workerErrors:workerErrors.length,readExisting:true,realLostCommitAckNoReplay:true,independentQuarantine:true}))
|
|
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,guardedSameHolderJournal:true,separateDurableJournalCommits:true,independentQuarantineAfterRealLostCommitAck: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()
|
|
}
|