Queue native collection launches before async placement

src/lib/server/hosts/availability.ts

@@ -0,0 +1,141 @@
+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 { 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;
+    ports: { port: number; bind_address: string }[];
+}
+interface Snapshot {
+    updatedAt: string;
+    hosts: Map<string, HostAvailability>;
+    startupLimit: number;
+    active: number;
+    queued: number;
+}
+const snapshots = new WeakMap<ReturnType<typeof db>, Snapshot>();
+const scheduled = new WeakSet<ReturnType<typeof db>>();
+
+function snapshotSchemaReady(handle: ReturnType<typeof db>): 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<typeof db>, hosts: Map<string, HostAvailability>): 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 unmeasured = pending.filter(instance => measured.get(instance.id) !== instance.generation).length;
+    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,
+        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();
+});
+
+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);
+}
+
+/** 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<string, number>();
+    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 };
+}

src/lib/server/hosts/index.ts

@@ -5,6 +5,7 @@ import { isIP } from 'node:net';
 import { db } from '../db/index.js';
 import { ensureHostSchema } from './schema.js';
 import { recordInstanceResources, validateInstanceResources } from './calibration.js';
+import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js';
 import type { Host, HostDecommission, HostDeletion, HostInventory, HostMode, HostOperation, HostOperationState } from '../../hosts.js';
 
 export const AGENT_PROTOCOL_VERSION = 1;
@@ -83,6 +84,17 @@ export function hostDeletionStatus(id: string): HostDeletion {
         const names = instances.slice(0, 5).map(instance => instance.server).join(', ') + (instances.length > 5 ? ', …' : '');
         blockers.push(`Remove the ${instances.length} server instance${instances.length === 1 ? '' : 's'} assigned to this host first (${names}). Queued and failed deployments also require removal.`);
     }
+    const pinned = handle.prepare(`SELECT d.server FROM server_deployments d
+        JOIN native_creation_intents q ON q.deployment_id=d.id
+        LEFT JOIN native_provisioning n ON n.deployment_id=d.id
+        WHERE d.action='create' AND d.server_deleted_at IS NULL
+        AND d.state IN ('queued','waiting','running','awaiting-health','interrupted') AND n.deployment_id IS NULL
+        AND json_extract(q.selection,'$.hostId')=? AND COALESCE(json_extract(q.selection,'$.automatic'),0)!=1
+        ORDER BY d.server LIMIT 6`).all(id) as { server: string }[];
+    if (pinned.length) {
+        const names = pinned.slice(0, 5).map(job => job.server).join(', ') + (pinned.length > 5 ? ', …' : '');
+        blockers.push(`Cancel the queued server launches pinned to this host before deleting it (${names}).`);
+    }
     const ports = handle.prepare('SELECT COUNT(*) AS count FROM ec_host_ports WHERE host_id=?').get(id) as { count: number };
     if (ports.count && !instances.length) blockers.push(`This host still has ${ports.count} reserved game port${ports.count === 1 ? '' : 's'}. Finish or reconcile the corresponding server removals first.`);
     // A pending command may have reached the agent before its acknowledgement.
@@ -174,6 +186,7 @@ export function createHost(input: Record<string, unknown>): Host {
     const bind = input.bindAddress === undefined ? '0.0.0.0' : textField(input.bindAddress, 'Bind address');
     if (!isIP(bind)) throw new Error('Bind address must be an IPv4 or IPv6 address.');
     database().prepare('INSERT INTO ec_hosts(id,name,port_min,port_max,bind_address,created_at) VALUES(?,?,?,?,?,?)').run(id, name, min, max, bind, now());
+    scheduleAvailabilityRefresh();
     return requireHost(id);
 }
 export function updateHost(id: string, input: Record<string, unknown>): Host {
@@ -190,6 +203,7 @@ export function updateHost(id: string, input: Record<string, unknown>): Host {
     const minimum = input.minimumFreeBytes ?? host.minimumFreeBytes;
     if (typeof minimum !== 'number' || !Number.isSafeInteger(minimum) || minimum < 0) throw new Error('Minimum free storage must be a nonnegative byte count.');
     database().prepare('UPDATE ec_hosts SET name=?,mode=?,port_min=?,port_max=?,bind_address=?,public_address=?,minimum_free_bytes=? WHERE id=?').run(name, mode, min, max, bind, publicAddress, minimum, id);
+    scheduleAvailabilityRefresh();
     return requireHost(id);
 }
 
@@ -256,6 +270,7 @@ export function enrollHost(token: unknown, rawInventory: unknown): { hostId: str
         const credential = issueCredential(enrollment.host_id);
         database().prepare('UPDATE ec_hosts SET enrolled_at=?,last_heartbeat=?,hostname=?,inventory=? WHERE id=?').run(now(), now(), inventory.hostname ?? '', JSON.stringify(inventory), enrollment.host_id);
         recordResources(enrollment.host_id, inventory, now());
+        scheduleAvailabilityRefresh();
         return { hostId: enrollment.host_id, credential, heartbeatSeconds: 15 };
     })();
 }
@@ -270,6 +285,7 @@ export function revokeCredentials(hostId: string): void {
     const host = requireHost(hostId);
     if (host.decommission && !host.decommission.retryable) throw new Error('Host cleanup needs its credentials until removal is confirmed.');
     database().prepare('UPDATE ec_host_credentials SET revoked_at=? WHERE host_id=? AND revoked_at IS NULL').run(now(), hostId);
+    scheduleAvailabilityRefresh();
 }
 export function rotateCredential(hostId: string, nextCredential?: string): string {
     requireHost(hostId);
@@ -292,7 +308,9 @@ export function heartbeat(hostId: string, value: unknown): Host {
         database().prepare('UPDATE ec_hosts SET last_heartbeat=?,hostname=?,inventory=? WHERE id=?').run(timestamp, inventory.hostname ?? '', JSON.stringify(inventory), hostId);
         recordResources(hostId, inventory, timestamp);
     })();
-    return requireHost(hostId);
+    const host = requireHost(hostId);
+    refreshHostAvailability(host);
+    return host;
 }
 
 function recordResources(hostId: string, inventory: HostInventory, timestamp: string): void {
@@ -504,7 +522,11 @@ export function reserveHostPort(hostId: string, instanceId: string, options: { p
         const allocated = options.port ?? nextHostPort(hostId, [], bind, protocol);
         if (allocated === null || !hostPortAvailable(hostId, allocated, bind, protocol, instanceId)) throw new Error('No available game port on the selected host.');
         database().prepare('INSERT INTO ec_host_ports(host_id,instance_id,bind_address,protocol,port,created_at) VALUES(?,?,?,?,?,?)').run(hostId, instanceId, bind, protocol, allocated, now());
+        scheduleAvailabilityRefresh();
         return allocated;
     })();
 }
-export function releaseHostPorts(instanceId: string): void { database().prepare('DELETE FROM ec_host_ports WHERE instance_id=?').run(instanceId); }
+export function releaseHostPorts(instanceId: string): void {
+    database().prepare('DELETE FROM ec_host_ports WHERE instance_id=?').run(instanceId);
+    scheduleAvailabilityRefresh();
+}

src/lib/server/hosts/preparation.ts

@@ -2,12 +2,46 @@ import { db } from '../db/index.js';
 import { config } from '../config.js';
 import { AGENT_PROTOCOL_VERSION, getHost, queueHostOperation } from './index.js';
 import { getHostLibraryCache, latestPublishedLibraries, queueLibraryDownload } from '../game-files/index.js';
+import { GAME_FILE_GAMES } from '../../game-files.js';
 import type { Host, HostOperationState, HostPreparation, HostPreparationItem } from '../../hosts.js';
 
 const activeStates = ['pending', 'accepted', 'running', 'unknown'];
 const supersededDownload = 'Superseded by newer game files before delivery to the host.';
 type PreparationOperation = { id: string; state: HostOperationState; updated_at: string; error: string | null };
 type Target = { item: HostPreparationItem; image?: string; versionId?: string; requiredBytes: number; operation?: PreparationOperation };
+type QueuedSelection = { libraryId: string; image: string | null };
+type LibraryTarget = Pick<ReturnType<typeof latestPublishedLibraries>[number], 'id' | 'sizeBytes' | 'manifestSha256'> & { name: string; queued?: boolean };
+
+/** Accepted requests retain their prerequisites until placement owns an instance. */
+function queuedSelections(hostId: string): QueuedSelection[] {
+    // Standalone host/library readers can initialize before provisioning migrations.
+    if (!db().prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='native_creation_intents'").get()) return [];
+    return db().prepare(`SELECT DISTINCT json_extract(q.selection,'$.libraryId') AS libraryId,
+        json_extract(q.selection,'$.image') AS image FROM native_creation_intents q
+        JOIN server_deployments d ON d.id=q.deployment_id
+        LEFT JOIN native_provisioning n ON n.deployment_id=d.id
+        WHERE d.action='create' AND d.server_deleted_at IS NULL AND n.deployment_id IS NULL
+        AND d.state IN ('queued','waiting','running','awaiting-health','interrupted')
+        AND (json_extract(q.selection,'$.automatic')=1 OR json_extract(q.selection,'$.hostId') IS NULL
+            OR json_extract(q.selection,'$.hostId')=?)`).all(hostId) as QueuedSelection[];
+}
+
+function targetLibraries(selections: QueuedSelection[]): LibraryTarget[] {
+    const latest = latestPublishedLibraries();
+    const latestIds = new Set(latest.map(version => version.id));
+    const queued = new Map<string, LibraryTarget>();
+    const get = db().prepare(`SELECT v.id,l.game,v.version,v.size_bytes AS sizeBytes,v.manifest_sha256 AS manifestSha256
+        FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id
+        WHERE v.id=? AND v.state='ready' AND NOT EXISTS(SELECT 1 FROM game_file_cleanup c WHERE c.version_id=v.id)`);
+    for (const selection of selections) {
+        if (latestIds.has(selection.libraryId) || queued.has(selection.libraryId)) continue;
+        const version = get.get(selection.libraryId) as { id: string; game: string; version: string; sizeBytes: number; manifestSha256: string } | undefined;
+        if (version) queued.set(version.id, { ...version, queued: true,
+            name: `${GAME_FILE_GAMES.find(game => game.id === version.game)?.name ?? version.game} · Queued version ${version.version}` });
+    }
+    // Finish prerequisites for accepted work before downloading newer optional files.
+    return [...queued.values(), ...latest];
+}
 
 function latestOperation(hostId: string, kind: string, field: string, value: string): PreparationOperation | undefined {
     return db().prepare(`SELECT id,state,updated_at,error FROM ec_host_operations
@@ -43,19 +77,24 @@ function hostBlocker(host: Host): { state: 'waiting' | 'blocked'; detail: string
 }
 
 function targets(host: Host): Target[] {
-    const image = config.komodo.defaultImage;
-    const runtimeReady = host.inventory.runtimeImages?.includes(image);
-    const result: Target[] = [{ image, requiredBytes: host.minimumFreeBytes,
-        operation: latestOperation(host.id, 'runtime.pull', 'image', image),
-        item: { kind: 'runtime', label: image, state: runtimeReady ? 'ready' : 'waiting', detail: runtimeReady ? 'Runtime is cached.' : 'Runtime will be downloaded automatically.' } }];
-    for (const version of latestPublishedLibraries()) {
+    const selections = queuedSelections(host.id);
+    const images = new Set([config.komodo.defaultImage, ...selections.flatMap(selection => selection.image ? [selection.image] : [])]);
+    const result: Target[] = [...images].map(image => {
+        const runtimeReady = host.inventory.runtimeImages?.includes(image);
+        return { image, requiredBytes: host.minimumFreeBytes,
+            operation: latestOperation(host.id, 'runtime.pull', 'image', image),
+            item: { kind: 'runtime', label: image, state: runtimeReady ? 'ready' : 'waiting', detail: runtimeReady ? 'Runtime is cached.' : 'Runtime will be downloaded automatically.' } };
+    });
+    for (const version of targetLibraries(selections)) {
         const cache = getHostLibraryCache(host.id, version.id);
         const matches = cache?.manifestSha256 === version.manifestSha256;
         const ready = matches && ['ready', 'outdated'].includes(cache!.status) && cache!.bytesDownloaded === version.sizeBytes;
         result.push({ versionId: version.id, requiredBytes: version.sizeBytes - (matches ? cache!.bytesDownloaded : 0) + host.minimumFreeBytes,
             operation: latestOperation(host.id, 'library.download', 'libraryVersionId', version.id),
             item: { kind: 'library', label: version.name, state: ready ? 'ready' : 'waiting',
-                detail: ready ? 'The latest game files are downloaded and verified.' : 'The latest game files will be downloaded and verified automatically.',
+                detail: version.queued
+                    ? ready ? 'Game files selected by queued launches are downloaded and verified.' : 'Game files selected by queued launches will be downloaded and verified automatically.'
+                    : ready ? 'The latest game files are downloaded and verified.' : 'The latest game files will be downloaded and verified automatically.',
                 bytesDownloaded: cache?.bytesDownloaded ?? 0, bytesTotal: version.sizeBytes, ...(cache ? { cache } : {}) } });
     }
     const blocker = hostBlocker(host);
@@ -100,13 +139,13 @@ export function presentHostPreparation(host: Host): Host {
 }
 
 function supersedeUndeliveredDownloads(hostId: string): void {
-    const latest = new Set(latestPublishedLibraries().map(version => version.id));
+    const required = new Set(targetLibraries(queuedSelections(hostId)).map(version => version.id));
     const pending = db().prepare(`SELECT id,json_extract(payload,'$.libraryVersionId') AS versionId
         FROM ec_host_operations WHERE host_id=? AND kind='library.download' AND state='pending'
         AND last_delivered_at IS NULL AND json_extract(payload,'$.automatic')=1`).all(hostId) as { id: string; versionId: string }[];
     const now = new Date().toISOString();
     for (const operation of pending) {
-        if (latest.has(operation.versionId)) continue;
+        if (required.has(operation.versionId)) continue;
         // Delivered operations may already be writing files. Only unsubmitted
         // automatic work can be replaced without consulting the agent journal.
         db().prepare("UPDATE ec_host_operations SET state='failed',error=?,updated_at=? WHERE id=?").run(supersededDownload, now, operation.id);

src/lib/server/hosts/schema.ts

@@ -3,6 +3,8 @@ import type Database from 'better-sqlite3';
 const initialized = new WeakSet<Database.Database>();
 export function ensureHostSchema(handle: Database.Database): void {
     if (initialized.has(handle)) return;
+    // Background readers open only after controller migrations have completed.
+    if (handle.readonly) { initialized.add(handle); return; }
     handle.exec(`
         CREATE TABLE IF NOT EXISTS ec_hosts (
             id TEXT PRIMARY KEY, name TEXT NOT NULL, hostname TEXT NOT NULL DEFAULT '',