import {randomUUID} from 'node:crypto' import {DatabaseSync} from 'node:sqlite' import type {ComputerProvider,ComputerSpecification} from './container.js' import {ComputerError} from './container.js' export interface ComputerLease {id:string;environmentId:string;ownerKey:string;holderId:string;generation:number} type Status='free'|'preparing'|'active'|'stopping'|'uncertain' interface Entry {spec:ComputerSpecification;generation:number;status:Status;lease?:ComputerLease;expiresAt:number;updatedAt:number} interface Active {lease:ComputerLease;authorize():void;controller:AbortController;operations:Set>;detach():void} /** Durable exclusive use. A generation is invalidated before cancellation; no * successor is admitted until the old private environment is confirmed stopped. */ export class ComputerPool { private active=new Map() private changing=new Map>() private ready=false private closing=false private maintenance=false holdMaintenance(){this.maintenance=true;let released=false;return ()=>{if(!released){released=true;this.maintenance=false}}} beginMaintenance(){ if(this.maintenance||this.changing.size||this.rows().some(row=>row.status!=='free'))throw new ComputerError('computer_busy','请先停止电脑任务并交还控制权,再管理镜像') return this.holdMaintenance() } constructor(readonly db:DatabaseSync,readonly provider:ComputerProvider,readonly limits={concurrent:2,environments:32,ttlMs:30000},readonly now=Date.now) { if(![limits.concurrent,limits.environments,limits.ttlMs].every(Number.isSafeInteger)||limits.concurrent<1||limits.concurrent>8||limits.environments128||limits.ttlMs<1000||limits.ttlMs>120000)throw new Error('电脑资源配额无效') db.exec('CREATE TABLE IF NOT EXISTS computer_environments(id TEXT PRIMARY KEY,value TEXT NOT NULL)') } private rows():Entry[]{return (this.db.prepare('SELECT value FROM computer_environments').all() as {value:string}[]).map(row=>JSON.parse(row.value))} private get(id:string):Entry|undefined {const row=this.db.prepare('SELECT value FROM computer_environments WHERE id=?').get(id) as {value:string}|undefined;return row?JSON.parse(row.value):undefined} private save(entry:Entry){entry.updatedAt=this.now();this.db.prepare('INSERT INTO computer_environments VALUES(?,?) ON CONFLICT(id) DO UPDATE SET value=excluded.value').run(entry.spec.id,JSON.stringify(entry))} private serial(id:string,action:()=>Promise):Promise{ const pending=(this.changing.get(id)??Promise.resolve()).catch(()=>{}).then(action) this.changing.set(id,pending);void pending.finally(()=>{if(this.changing.get(id)===pending)this.changing.delete(id)}).catch(()=>{}) return pending } /** Must run under the Runner's OS instance lock before accepting commands. */ async recover():Promise{ if(this.ready)throw new ComputerError('computer_pool_active','运行中的资源池不能重新执行启动恢复') for(const entry of this.rows())if(entry.status!=='free')await this.serial(entry.spec.id,()=>this.stopEntry(entry)).catch(()=>{}) this.ready=true } status(ownerKey:string){return this.rows().filter(row=>row.spec.ownerKey===ownerKey).map(row=>({environmentId:row.spec.id,status:row.status,generation:row.generation,holderId:row.lease?.holderId,expiresAt:row.expiresAt}))} private require(lease:ComputerLease,activeOnly=true):Entry { const entry=this.get(lease.environmentId) if(!entry||entry.spec.ownerKey!==lease.ownerKey||entry.lease?.id!==lease.id||entry.lease.holderId!==lease.holderId||entry.generation!==lease.generation||(activeOnly&&entry.status!=='active'))throw new ComputerError('computer_lease_stale','电脑控制权已改变,旧操作不能继续') return entry } private assert(lease:ComputerLease){ const entry=this.require(lease),active=this.active.get(lease.id) if(this.closing||!active||active.controller.signal.aborted||entry.expiresAt<=this.now())throw new ComputerError('computer_lease_expired','电脑控制授权已到期') active.authorize() } async acquire(spec:ComputerSpecification,holderId:string,authorize:()=>void,signal?:AbortSignal):Promise{ spec=this.provider.validateSpecification(spec) if(!holderId||holderId.length>256)throw new ComputerError('computer_holder_invalid','电脑任务身份无效') return this.serial(spec.id,async()=>{ authorize() if(this.maintenance)throw new ComputerError('computer_busy','执行节点正在管理镜像,请稍后重试') if(!this.ready||this.closing)throw new ComputerError('computer_pool_unavailable','电脑资源正在恢复或关闭') let entry=this.get(spec.id) if(entry&&(entry.spec.ownerKey!==spec.ownerKey||JSON.stringify(entry.spec)!==JSON.stringify(spec)))throw new ComputerError('computer_owner_mismatch','电脑环境归属或配置不匹配') if(entry?.status==='active'&&entry.lease?.holderId===holderId&&this.active.has(entry.lease.id)){this.assert(entry.lease);return entry.lease} if(entry&&entry.status!=='free')throw new ComputerError('computer_busy','电脑正在使用或需要核对停止状态') const rows=this.rows() if((!entry&&rows.length>=this.limits.environments)||rows.filter(row=>row.status!=='free').length>=this.limits.concurrent)throw new ComputerError('computer_quota','电脑资源已达上限,请等待其他任务释放') const lease:ComputerLease={id:randomUUID(),environmentId:spec.id,ownerKey:spec.ownerKey,holderId,generation:(entry?.generation??0)+1} entry={spec:structuredClone(spec),lease,generation:lease.generation,status:'preparing',expiresAt:this.now()+this.limits.ttlMs,updatedAt:this.now()} // This reservation is synchronous, before any provider await or next acquire. this.save(entry) const abort=()=>{const current=this.get(spec.id);if(current?.lease?.id===lease.id)this.invalidate(current)} const active:Active={lease,authorize,controller:new AbortController(),operations:new Set(),detach:()=>signal?.removeEventListener('abort',abort)};this.active.set(lease.id,active) signal?.addEventListener('abort',abort,{once:true});if(signal?.aborted)abort() const guard=()=>{const current=this.require(lease,false);if(this.closing||active.controller.signal.aborted||current.status!=='preparing'||current.expiresAt<=this.now())throw new ComputerError('computer_lease_expired','电脑启动授权已失效');authorize()} try{ await this.provider.ensure(spec,guard);guard() entry.status='active';this.save(entry);return lease }catch(error){await this.stopEntry(entry).catch(()=>{});throw error} }) } async configure(spec:ComputerSpecification,authorize:()=>void):Promise{ spec=this.provider.validateSpecification(spec) return this.serial(spec.id,async()=>{ authorize();if(this.maintenance)throw new ComputerError('computer_busy','执行节点正在管理镜像,请稍后重试') if(!this.ready||this.closing)throw new ComputerError('computer_pool_unavailable','电脑资源正在恢复或关闭') const entry=this.get(spec.id);if(!entry)return if(entry.spec.ownerKey!==spec.ownerKey)throw new ComputerError('computer_owner_mismatch','电脑环境归属不匹配') if(JSON.stringify(entry.spec)===JSON.stringify(spec))return if(entry.status!=='free')throw new ComputerError('computer_busy','请先释放当前电脑,再应用新的配置') entry.status='stopping';entry.generation++;this.save(entry) try{await this.provider.remove(entry.spec);authorize();entry.spec=spec;entry.status='free';this.save(entry)} catch(error){entry.status='uncertain';this.save(entry);throw error} }) } async transfer(lease:ComputerLease,holderId:string,authorize:()=>void):Promise{ return this.serial(lease.environmentId,async()=>{ this.assert(lease);authorize() const active=this.active.get(lease.id)! if(active.operations.size)throw new ComputerError('computer_busy','电脑仍有在途操作,不能移交控制') const entry=this.require(lease),state=await this.provider.inspect(entry.spec) this.assert(lease);authorize() if(active.operations.size||!state?.running)throw new ComputerError('computer_busy','电脑尚未就绪,不能移交控制') const next:ComputerLease={...lease,id:randomUUID(),holderId,generation:entry.generation+1} active.detach();active.controller.abort();this.active.delete(lease.id) this.active.set(next.id,{lease:next,authorize,controller:new AbortController(),operations:new Set(),detach:()=>{}}) entry.lease=next;entry.generation=next.generation;entry.expiresAt=this.now()+this.limits.ttlMs;this.save(entry) return next }) } authorize(lease:ComputerLease){this.assert(lease)} renew(lease:ComputerLease):void{this.assert(lease);const entry=this.require(lease);entry.expiresAt=this.now()+this.limits.ttlMs;this.save(entry)} use(lease:ComputerLease,action:(context:{signal:AbortSignal;authorize():void})=>Promise):Promise{ this.assert(lease) const active=this.active.get(lease.id)! const pending=Promise.resolve().then(()=>{this.assert(lease);return action({signal:active.controller.signal,authorize:()=>this.assert(lease)})}).then(value=>{this.assert(lease);return value}) active.operations.add(pending);void pending.finally(()=>active.operations.delete(pending)).catch(()=>{}) return pending } release(lease:ComputerLease):Promise{ const existing=this.get(lease.environmentId) const entry=existing&&existing.lease?.id===lease.id&&existing.spec.ownerKey===lease.ownerKey&&existing.lease.holderId===lease.holderId&&['stopping','uncertain'].includes(existing.status)?existing:this.require(lease,false) this.invalidate(entry) return this.serial(entry.spec.id,async()=>{const current=this.get(entry.spec.id)!;if(current.status==='free')return;await this.stopEntry(current)}) } definition(ownerKey:string,id:string){const entry=this.get(id);if(entry&&entry.spec.ownerKey!==ownerKey)throw new ComputerError('computer_owner_mismatch','电脑环境归属不匹配');return entry?.spec} async retire(ownerKey:string,id:string):Promise{ return this.serial(id,async()=>{ const entry=this.get(id);if(!entry)return if(entry.spec.ownerKey!==ownerKey)throw new ComputerError('computer_owner_mismatch','电脑环境归属不匹配') if(entry.status!=='free')throw new ComputerError('computer_busy','电脑停止状态尚未确认') await this.provider.remove(entry.spec) this.db.prepare('DELETE FROM computer_environments WHERE id=?').run(id) return entry.spec }) } async stopHolder(ownerKey:string,id:string,holderId:string):Promise{ const entry=this.get(id) if(!entry||entry.spec.ownerKey!==ownerKey)throw new ComputerError('computer_owner_mismatch','电脑环境归属不匹配') if(entry.lease?.holderId!==holderId)return this.invalidate(entry);await this.serial(id,()=>this.stopEntry(entry)) } private invalidate(expected:Entry){ const entry=this.get(expected.spec.id) if(!entry||entry.lease?.id!==expected.lease?.id||entry.status==='free')return if(entry.status!=='stopping'){entry.generation++;entry.status='stopping';entry.expiresAt=0;this.save(entry)} const active=entry.lease?this.active.get(entry.lease.id):undefined active?.controller.abort() return entry } private async stopEntry(expected:Entry):Promise{ const entry=this.invalidate(expected) if(!entry)return const active=entry.lease?this.active.get(entry.lease.id):undefined try{ await this.provider.stop(entry.spec) // Retire the runtime identity as well as the lease generation. A delayed // exec targeting the old container ID can never enter its successor. await this.provider.remove(entry.spec) active?.detach();if(entry.lease)this.active.delete(entry.lease.id) entry.status='free';entry.lease=undefined;entry.expiresAt=0;this.save(entry) }catch(error){entry.status='uncertain';this.save(entry);throw error} } async expire():Promise{ for(const id of this.rows().map(entry=>entry.spec.id)){ const entry=this.get(id)! if((entry.status==='active'||entry.status==='preparing')&&entry.expiresAt<=this.now()){ this.invalidate(entry);await this.serial(id,()=>this.stopEntry(entry)).catch(()=>{}) } } } async close():Promise{ this.closing=true for(const entry of this.rows())if(entry.status!=='free')this.invalidate(entry) await Promise.allSettled(this.rows().filter(entry=>entry.status!=='free').map(entry=>this.serial(entry.spec.id,()=>this.stopEntry(this.get(entry.spec.id)!)))) if(this.rows().some(entry=>entry.status!=='free'))throw new ComputerError('computer_stop_uncertain','尚未确认所有电脑环境已停止') } }