import { db } from '../db/index.js'; import { fleetSettings } from '../settings.js'; import { subscribe } from '../pipeline/bus.js'; import { capacityCalibration } from './calibration.js'; import { HEARTBEAT_MAX_AGE_MS, listHosts } from './index.js'; import type { CapacityCalibration, Host, NativePlacement } from '../../hosts.js'; import type { LaunchAvailability } from '../../deployment-plan.js'; interface HostAvailability { host: Host; freeCpu: number | null; freeMemory: number | null; runningCpu: number; runningMemory: number; calibration: CapacityCalibration; ports: { port: number; bind_address: string }[]; } interface Snapshot { updatedAt: string; hosts: Map; startupLimit: number; active: number; queued: number; } const snapshots = new WeakMap, Snapshot>(); const scheduled = new WeakSet>(); function snapshotSchemaReady(handle: ReturnType): boolean { // Host enrollment and game-file maintenance can run before provisioning is // initialized. Missing queue data means unknown capacity, not an empty fleet. const { count } = handle.prepare(`SELECT COUNT(*) AS count FROM sqlite_master WHERE type='table' AND name IN ('native_instances','native_provisioning','server_deployments','settings')`).get() as { count: number }; if (count === 4) return true; snapshots.delete(handle); return false; } function queueSnapshot(handle: ReturnType, hosts: Map): Snapshot { const counts = handle.prepare(`SELECT COALESCE(SUM(state IN ('running','awaiting-health')),0) AS active, COALESCE(SUM(state='queued'),0) AS queued FROM server_deployments WHERE action IN ('create','rebuild') AND state IN ('queued','running','awaiting-health') AND server_deleted_at IS NULL`).get() as { active: number; queued: number }; return { hosts, ...counts, startupLimit: fleetSettings().createConcurrency, updatedAt: new Date().toISOString() }; } function observeHost(host: Host): HostAvailability { const handle = db(); const { running, startup } = capacityCalibration(host.id); const measured = new Map(host.inventory.instanceResources?.map(sample => [sample.instanceId, sample.generation])); const pending = handle.prepare(`SELECT DISTINCT i.id,i.generation FROM native_instances i JOIN native_provisioning n ON n.instance_id=i.id JOIN server_deployments d ON d.id=n.deployment_id WHERE i.host_id=? AND i.retired_at IS NULL AND d.action IN ('create','rebuild') AND d.state IN ('running','awaiting-health')`).all(host.id) as { id: string; generation: number }[]; const incoming = handle.prepare(`SELECT COUNT(*) AS n FROM ec_host_migrations m JOIN server_deployments d ON d.id=m.deployment_id WHERE m.target_host_id=? AND (m.removed_at IS NULL OR d.state IN ('queued','waiting','interrupted'))`).get(host.id) as { n: number }; const unmeasured = pending.filter(instance => measured.get(instance.id) !== instance.generation).length + incoming.n; const { cpuCount, cpuUsagePercent, memoryTotal, memoryAvailable } = host.inventory; const freeCpu = cpuCount && cpuUsagePercent !== undefined ? Math.max(0, cpuCount * (0.95 - cpuUsagePercent / 100) - unmeasured * startup.cpuCores) : null; const freeMemory = memoryTotal && memoryAvailable !== undefined && memoryAvailable <= memoryTotal ? Math.max(0, memoryAvailable - 500_000_000 - unmeasured * startup.memoryBytes) : null; return { host, freeCpu, freeMemory, runningCpu: running.cpuCores, runningMemory: running.memoryBytes, calibration: { running, startup }, ports: handle.prepare('SELECT port,bind_address FROM ec_host_ports WHERE host_id=?').all(host.id) as HostAvailability['ports'] }; } /** Host telemetry updates the estimate once, independently of how many lobbies are previewed. */ export function refreshHostAvailability(host: Host): void { const handle = db(); if (!snapshotSchemaReady(handle)) return; const hosts = new Map(snapshots.get(handle)?.hosts); hosts.set(host.id, observeHost(host)); snapshots.set(handle, queueSnapshot(handle, hosts)); } /** Lifecycle events refresh the shared snapshot after their transaction and request have yielded. */ export function scheduleAvailabilityRefresh(): void { const handle = db(); if (scheduled.has(handle)) return; scheduled.add(handle); setImmediate(() => { scheduled.delete(handle); if (!handle.open) return; try { if (!snapshotSchemaReady(handle)) return; const hosts = new Map(listHosts().map(host => [host.id, observeHost(host)])); snapshots.set(handle, queueSnapshot(handle, hosts)); } catch (error) { console.error('[availability] snapshot refresh failed:', error); } }); } subscribe(event => { if (event.type === 'fleet-invalidated' || event.type === 'server-status') scheduleAvailabilityRefresh(); }); export function freePorts(snapshot: HostAvailability, bindAddress?: string): number { const bind = bindAddress ?? snapshot.host.bindAddress; const used = new Set(snapshot.ports.filter(row => row.port >= snapshot.host.portMin && row.port <= snapshot.host.portMax && (row.bind_address === bind || ['0.0.0.0', '::'].includes(bind) || ['0.0.0.0', '::'].includes(row.bind_address))).map(row => row.port)); return Math.max(0, snapshot.host.portMax - snapshot.host.portMin + 1 - used.size); } /** Share the launch estimate's already observed state without rerunning placement. */ export function readAvailabilitySnapshot(): Snapshot | null { const snapshot = snapshots.get(db()); if (!snapshot) scheduleAvailabilityRefresh(); return snapshot ?? null; } /** Estimates only: no placement, historical scans, reservations, or live host requests. */ export function launchAvailability(selections: NativePlacement[]): { availability: LaunchAvailability; warnings: string[] } { const snapshot = snapshots.get(db()); if (!snapshot) { scheduleAvailabilityRefresh(); return { availability: { updatedAt: null, requested: selections.length, estimatedAvailable: null, portsAvailable: null, cpuCoresAvailable: null, memoryBytesAvailable: null, startupSlots: null, startupLimit: null, queued: null }, warnings: selections.length ? ['Capacity measurements are updating. You can queue this launch; provisioning waits for an eligible host.'] : [] }; } const candidates = [...snapshot.hosts.values()].filter(({ host }) => host.readiness === 'ready' && host.lastHeartbeat && Date.now() - Date.parse(host.lastHeartbeat) <= HEARTBEAT_MAX_AGE_MS && (host.inventory.storageAvailable ?? 0) >= host.minimumFreeBytes + 1024 ** 3); const eligible = (candidate: HostAvailability, selection: NativePlacement) => (!selection.hostId || selection.automatic || candidate.host.id === selection.hostId) && !!candidate.host.inventory.runtimeImages?.includes(selection.image ?? '') && !!candidate.host.inventory.libraries?.some(library => library.id === selection.libraryId && ['ready', 'outdated'].includes(library.state)); const relevant = candidates.filter(candidate => selections.some(selection => eligible(candidate, selection))); const planned = new Map(); let estimatedAvailable = 0; for (const selection of selections) { const candidate = relevant.find(candidate => { const used = planned.get(candidate.host.id) ?? 0; return eligible(candidate, selection) && freePorts(candidate, selection.network?.bindAddress ?? selection.bindAddress) > used && candidate.freeCpu !== null && candidate.freeMemory !== null && candidate.freeCpu >= (used + 1) * candidate.runningCpu && candidate.freeMemory >= (used + 1) * candidate.runningMemory; }); if (candidate) { estimatedAvailable++; planned.set(candidate.host.id, (planned.get(candidate.host.id) ?? 0) + 1); } } const warnings: string[] = []; const unknown = relevant.some(candidate => candidate.freeCpu === null || candidate.freeMemory === null); if (selections.length && !relevant.length) warnings.push('Waiting for an eligible host with the selected runtime and game files. The launch can be queued now.'); else if (unknown) warnings.push('Some host CPU or memory measurements are unavailable. Provisioning waits for fresh capacity measurements.'); else if (estimatedAvailable < selections.length) warnings.push(`Current capacity is estimated for ${estimatedAvailable} of ${selections.length} servers. Remaining launches wait for capacity, including any removals in this collection.`); const startupSlots = Math.max(0, snapshot.startupLimit - snapshot.active); if (selections.length && !startupSlots) warnings.push('All startup slots are in use. New launches will wait in the queue.'); if (selections.length && snapshot.queued) warnings.push(`${snapshot.queued} server${snapshot.queued === 1 ? ' is' : 's are'} already queued.`); return { availability: { updatedAt: snapshot.updatedAt, requested: selections.length, estimatedAvailable: unknown ? null : estimatedAvailable, portsAvailable: relevant.reduce((total, candidate) => total + freePorts(candidate), 0), cpuCoresAvailable: unknown ? null : relevant.reduce((total, candidate) => total + (candidate.freeCpu ?? 0), 0), memoryBytesAvailable: unknown ? null : relevant.reduce((total, candidate) => total + (candidate.freeMemory ?? 0), 0), startupSlots, startupLimit: snapshot.startupLimit, queued: snapshot.queued }, warnings }; }