Fix capacity accounting and queued host placement

docs/native-hosts.md

@@ -122,8 +122,9 @@ host-wide busy time across all cores on a 0–100% scale between inventory sampl
 counts as unavailable capacity. The first sample after startup, unavailable
 counters, and older agents show **Not reported**. Offline hosts retain their
 last reported value. The controller retains CPU and memory samples for seven days (missing measurements
-remain unknown). Scheduling uses the last five minutes as well as the latest
-heartbeat; the Inventory panel continues to show the latest reading.
+remain unknown). Scheduling checks capacity against the latest heartbeat and uses
+the last five minutes to rank eligible hosts; the Inventory panel shows the latest
+reading.
 
 New servers are placed automatically. Preview and submission perform local
 capacity checks without contacting agents: a connected, active host must have
@@ -160,13 +161,24 @@ raise the cache immediately. Launch admission reads only the cached model.
 Scheduling still leaves **10% host CPU and 512 MiB RAM** for the OS and bursts.
 These allowances are scheduling reservations, not Docker limits. Queued and
 unresolved creations consume the startup allowance; established instances keep
-the running allowance even while idle or stopped. Observed host load also counts,
+the running allowance even while idle or stopped. Current container measurements
+are credited against their reservations because host usage already includes them;
+only the unused allowance is added. Missing or mismatched-generation measurements
+cannot release an unresolved launch's reservation. Observed host load also counts,
 so other processes and previously unseen load increases reduce available capacity.
 
-When startup capacity is available, provisioning rechecks all eligible hosts and
-chooses the lowest projected CPU/memory utilization, using the busier of the
-latest reading and the five-minute average. It moves the provisional reservation
-atomically and records the final host in the launch recipe. Host network defaults
+Admission counts all queued reservations. Dispatch evaluates one launch against
+existing instances and submitted or uncertain work; other never-submitted queued
+launches wait their turn. They retain their port reservations. This lets the queue
+make progress even if a changed allowance means the whole backlog no longer fits
+at once. Capacity errors report projected usage, usable capacity and reservations.
+
+When startup capacity is available, provisioning checks each eligible host’s
+current CPU and available memory against reservations and headroom.
+Among hosts that fit, it chooses the lowest projected CPU/memory utilization,
+using the busier of the latest reading and the five-minute average for ranking.
+A historical resource dip cannot block a host whose current capacity has recovered.
+Provisioning moves the provisional reservation atomically and records the final host in the launch recipe. Host network defaults
 are resolved again for the final host; explicit overrides are preserved. If no
 host qualifies, the request waits and is checked again. Once remote staging or
 creation has been submitted, placement stays fixed to prevent duplicate instances.

src/lib/server/hosts/capacity.ts

@@ -11,6 +11,9 @@ export class CapacityError extends Error {}
 export interface PlacementOptions {
     hostId?: string; libraryId: string; image: string; bindAddress?: string;
     thorough?: boolean; excludeInstanceId?: string;
+    // Execution only: not-yet-submitted launches wait their turn instead of
+    // requiring their entire backlog to fit before any one can start.
+    dispatch?: boolean;
     planned?: { hostId: string; port: number }[];
 }
 
@@ -31,34 +34,63 @@ function projectedLoad(host: Host, options: PlacementOptions): number {
     if (!cpuCount || cpuUsagePercent === undefined || !memoryTotal || memoryAvailable === undefined || memoryAvailable > memoryTotal) {
         throw new Error('Waiting for valid CPU and memory measurements.');
     }
-    let cpu = cpuUsagePercent;
-    let available = memoryAvailable;
+    let historicalCpu = cpuUsagePercent;
+    let historicalAvailable = memoryAvailable;
     if (options.thorough) {
         const history = db().prepare(`SELECT AVG(cpu_usage_percent) AS cpu, AVG(memory_available) AS memory
             FROM ec_host_resource_samples WHERE host_id=? AND sampled_at>=? AND cpu_count=? AND memory_total=?`)
             .get(host.id, new Date(Date.now() - 5 * 60_000).toISOString(), cpuCount, memoryTotal) as { cpu: number | null; memory: number | null };
-        cpu = Math.max(cpu, history.cpu ?? cpu);
-        available = Math.min(available, history.memory ?? available);
+        historicalCpu = Math.max(cpuUsagePercent, history.cpu ?? cpuUsagePercent);
+        historicalAvailable = Math.min(memoryAvailable, history.memory ?? memoryAvailable);
     }
-    // Unfinished/uncertain launches remain charged until recovery and a later
-    // heartbeat. Observed usage accounts for established servers; the count
-    // floor also reserves their minimum footprint while they are idle/stopped.
-    const instances = db().prepare(`SELECT COUNT(*) AS total, COALESCE(SUM(NOT EXISTS (
-        SELECT 1 FROM native_provisioning n JOIN server_deployments d ON d.id=n.deployment_id
-        WHERE n.instance_id=i.id AND d.action='create' AND d.state IN ('done','warning') AND d.finished_at<?
-    )),0) AS pending FROM native_instances i WHERE i.host_id=? AND i.retired_at IS NULL AND i.id!=?`)
-        .get(host.lastHeartbeat, host.id, options.excludeInstanceId ?? '') as { total: number; pending: number };
-    const planned = options.planned?.filter(item => item.hostId === host.id).length ?? 0;
-    const count = instances.total + planned + 1;
-    const pending = instances.pending + planned + 1;
+    const instances = db().prepare(`SELECT i.id, i.generation,
+        NOT EXISTS (SELECT 1 FROM native_provisioning n JOIN server_deployments d ON d.id=n.deployment_id
+            WHERE n.instance_id=i.id AND d.action='create' AND d.state IN ('done','warning') AND d.finished_at<?) AS pending,
+        EXISTS (SELECT 1 FROM native_provisioning n JOIN server_deployments d ON d.id=n.deployment_id
+            WHERE n.instance_id=i.id AND d.action='create' AND d.state='queued' AND d.submitted_at IS NULL
+            AND n.operation_id IS NULL AND n.stage_operation_id IS NULL) AS unsubmitted
+        FROM native_instances i WHERE i.host_id=? AND i.retired_at IS NULL AND i.id!=?`)
+        .all(host.lastHeartbeat, host.id, options.excludeInstanceId ?? '') as
+        { id: string; generation: number; pending: number; unsubmitted: number }[];
     const { running, startup } = capacityCalibration(host.id);
-    const cores = Math.max(cpuCount * cpu / 100 + pending * startup.cpuCores,
-        (count - pending) * running.cpuCores + pending * startup.cpuCores);
-    const memory = Math.max(memoryTotal - available + pending * startup.memoryBytes,
-        (count - pending) * running.memoryBytes + pending * startup.memoryBytes);
-    if (cores > cpuCount * 0.9) throw new Error('Insufficient CPU capacity.');
-    if (memory > memoryTotal - MEMORY_HEADROOM) throw new Error('Insufficient memory capacity.');
-    return Math.max(cores / cpuCount, memory / memoryTotal);
+    const planned = options.planned?.filter(item => item.hostId === host.id).length ?? 0;
+    let extraCpu = (planned + 1) * startup.cpuCores;
+    let extraMemory = (planned + 1) * startup.memoryBytes;
+    let reservedCpu = extraCpu;
+    let reservedMemory = extraMemory;
+    let reservations = planned + 1;
+    const measurements = new Map(host.inventory.instanceResources?.map(sample => [sample.instanceId, sample]));
+    for (const instance of instances) {
+        if (options.dispatch && options.excludeInstanceId && instance.unsubmitted) continue;
+        reservations++;
+        const allowance = instance.pending ? startup : running;
+        reservedCpu += allowance.cpuCores;
+        reservedMemory += allowance.memoryBytes;
+        const sample = measurements.get(instance.id);
+        if (sample?.generation === instance.generation) {
+            // Host usage already includes this container. Reserve only the
+            // unused part of its allowance, including while awaiting health.
+            extraCpu += Math.max(0, allowance.cpuCores - sample.cpuCores);
+            extraMemory += Math.max(0, allowance.memoryBytes - sample.memoryBytes);
+        } else if (instance.pending) {
+            // Missing observations cannot release an uncertain remote launch.
+            extraCpu += allowance.cpuCores;
+            extraMemory += allowance.memoryBytes;
+        }
+    }
+    const cores = Math.max(cpuCount * cpuUsagePercent / 100 + extraCpu, reservedCpu);
+    const usedMemory = memoryTotal - memoryAvailable;
+    const memory = Math.max(usedMemory + extraMemory, reservedMemory);
+    const usableMemory = memoryTotal - MEMORY_HEADROOM;
+    const gib = (bytes: number) => `${(bytes / 1024 ** 3).toFixed(2)} GiB`;
+    if (cores > cpuCount * 0.9) throw new Error(`Insufficient CPU capacity: ${cores.toFixed(2)} cores projected, ${(cpuCount * 0.9).toFixed(2)} usable; ${reservations} reservations including this launch.`);
+    if (memory > usableMemory) throw new Error(`Insufficient memory capacity: ${gib(memory)} projected, ${gib(usableMemory)} usable after host headroom. `
+        + `${gib(usedMemory)} currently used + ${gib(extraMemory)} additional reservations; ${reservations} reservations including this launch (${gib(startup.memoryBytes)} startup allowance).`);
+    // Recovered resources must not remain unavailable until the historical
+    // average catches up. History ranks hosts after current capacity passes.
+    const historicalMemory = Math.max(memoryTotal - historicalAvailable + extraMemory, reservedMemory);
+    const historicalCores = Math.max(cpuCount * historicalCpu / 100 + extraCpu, reservedCpu);
+    return Math.max(historicalCores / cpuCount, historicalMemory / memoryTotal);
 }
 
 /** Admission stops at the first fit; placement evaluates and ranks every fit. No remote I/O. */

src/lib/server/provisioning/native.ts

@@ -276,7 +276,7 @@ function placeQueuedCreation(job: Job, instance: NativeInstance): NativeInstance
         const network = prior.automatic ? prior.network ?? {} : prior;
         const { host, port } = choosePlacement({ hostId: prior.automatic ? undefined : instance.host_id,
             libraryId: instance.library_id, image: instance.image, bindAddress: network.bindAddress,
-            thorough: true, excludeInstanceId: instance.id });
+            thorough: true, dispatch: true, excludeInstanceId: instance.id });
         const bind = network.bindAddress ?? host.bindAddress;
         releaseHostPorts(instance.id);
         reserveHostPort(host.id, instance.id, { port, bindAddress: bind, protocol: 'udp' });
@@ -433,7 +433,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
     if (job.action === 'destroy' && count(['destroy']) >= settings.destroyConcurrency) { hold(job, 'Waiting for destruction capacity.'); return; }
     deploymentStage(job.id, 'Checking host eligibility', { phase: 'eligibility', blocker: null });
     if (job.action === 'create') choosePlacement({ hostId: instance.host_id, libraryId: instance.library_id, image: instance.image,
-        bindAddress: instance.bind_address, excludeInstanceId: instance.id, thorough: true });
+        bindAddress: instance.bind_address, excludeInstanceId: instance.id, thorough: true, dispatch: true });
     if (job.action === 'create' || job.action === 'rebuild') requireReadyLibrary(instance.host_id, instance.library_id);
     if (job.action !== 'create' && job.action !== 'start') {
         // Fast controller check keeps occupied lobbies out of the operation slots.

tests/unit/native-provisioning.test.ts

@@ -726,6 +726,91 @@ describe('automatic capacity admission and placement', () => {
         cancelCreation(first.deploymentId);
         expect(queueNativeCreation(input)).toMatchObject({ ok: true });
     });
+    it.each(['memory', 'cpu'] as const)('dispatches immediately after %s recovers instead of waiting for five-minute history', async resource => {
+        await prepareLibrary();
+        const { heartbeat } = await import('../../src/lib/server/hosts/index.js');
+        const { queueNativeCreation } = await import('../../src/lib/server/provisioning/native.js');
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const GiB = 1024 ** 3;
+        // A recently busy 4 GiB host is now mostly empty, as in the reported
+        // ec-2 inventory. Admission must not pass only to stall in dispatch.
+        heartbeat(hostId, { ...inventory(), cpuCount: 4, memoryTotal: 4 * GiB, memoryAvailable: Math.floor(3.8 * GiB) });
+        for (let minute = 1; minute <= 4; minute++) db().prepare(`INSERT INTO ec_host_resource_samples
+            (host_id,sampled_at,cpu_count,cpu_usage_percent,memory_total,memory_available) VALUES(?,?,4,?,?,?)`)
+            .run(hostId, new Date(Date.now() - minute * 60_000).toISOString(), resource === 'cpu' ? 100 : 10,
+                4 * GiB, resource === 'memory' ? GiB / 8 : Math.floor(3.8 * GiB));
+        const jobs = Array.from({ length: 3 }, () => queueNativeCreation(input));
+        expect(jobs.every(job => job.ok)).toBe(true);
+        await advance();
+        expect(await operationFor(jobs[0].ok ? jobs[0].deploymentId : 'missing')).not.toBeNull();
+    });
+    it('dispatches one queued launch when learned startup costs outgrow the original batch reservation', async () => {
+        await prepareLibrary();
+        const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js');
+        saveFleetSettings({ ...fleetSettings(), createConcurrency: 3 });
+        const { heartbeat } = await import('../../src/lib/server/hosts/index.js');
+        const { queueNativeCreation } = await import('../../src/lib/server/provisioning/native.js');
+        const { db } = await import('../../src/lib/server/db/index.js');
+        const GiB = 1024 ** 3;
+        heartbeat(hostId, { ...inventory(), cpuCount: 4, memoryTotal: 4 * GiB, memoryAvailable: Math.floor(3.8 * GiB) });
+        const jobs = Array.from({ length: 3 }, () => queueNativeCreation(input));
+        expect(jobs.every(job => job.ok)).toBe(true);
+        // Learning increases the startup allowance after admission. All three
+        // no longer fit concurrently, but this otherwise empty host fits one.
+        db().prepare(`INSERT INTO ec_resource_allowances
+            (scope,phase,cpu_cores,memory_bytes,samples,window_start,observed_at,computed_at)
+            VALUES(?,'startup',0.5,?,3,?,?,?)`).run(hostId, 2 * GiB,
+                new Date(Date.now() - 180_000).toISOString(), new Date().toISOString(), new Date().toISOString());
+        await advance();
+        const operations = await Promise.all(jobs.map(job => operationFor(job.ok ? job.deploymentId : 'missing')));
+        expect(operations.filter(Boolean)).toHaveLength(1);
+        expect(operations[0]?.kind).toBe('instance.create');
+        // The dispatched operation now consumes capacity and blocks the others,
+        // even though the next host heartbeat has not observed it yet.
+        await advance();
+        expect(db().prepare("SELECT * FROM ec_host_operations WHERE kind='instance.create'").all()).toHaveLength(1);
+        expect(queueNativeCreation(input)).toMatchObject({ ok: false, kind: 'conflict' });
+    });
+    it('does not charge a starting container twice when its usage is already in the heartbeat', async () => {
+        await prepareLibrary();
+        const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js');
+        saveFleetSettings({ ...fleetSettings(), createConcurrency: 2 });
+        const { heartbeat } = await import('../../src/lib/server/hosts/index.js');
+        const { queueNativeCreation } = await import('../../src/lib/server/provisioning/native.js');
+        const { nativeInstance } = await import('../../src/lib/server/backends/native.js');
+        const GiB = 1024 ** 3;
+        const host = { ...inventory(), cpuCount: 4, memoryTotal: 3 * GiB, memoryAvailable: 2.8 * GiB };
+        heartbeat(hostId, { ...host, memoryAvailable: Math.floor(host.memoryAvailable) });
+        const first = queueNativeCreation(input);
+        expect(first.ok).toBe(true); if (!first.ok) return;
+        await advance();
+        const instance = nativeInstance(first.name)!;
+        // 0.2 GiB OS + 0.8 GiB container. The remaining 0.2 GiB of its 1 GiB
+        // reservation plus a new 1 GiB server fit below 2.5 GiB usable memory.
+        heartbeat(hostId, { ...host, memoryAvailable: 2 * GiB, instanceResources: [{ instanceId: instance.id,
+            generation: instance.generation, uptimeSeconds: 60, cpuCores: 0.1, memoryBytes: Math.floor(0.8 * GiB) }] });
+        const second = queueNativeCreation(input);
+        expect(second).toMatchObject({ ok: true });
+        await advance();
+        expect(await operationFor(second.ok ? second.deploymentId : 'missing')).not.toBeNull();
+    });
+    it.each(['missing', 'old-generation'] as const)('keeps full reservations for a submitted launch with %s measurements', async evidence => {
+        await prepareLibrary();
+        const { heartbeat } = await import('../../src/lib/server/hosts/index.js');
+        const { queueNativeCreation } = await import('../../src/lib/server/provisioning/native.js');
+        const { nativeInstance } = await import('../../src/lib/server/backends/native.js');
+        const GiB = 1024 ** 3;
+        heartbeat(hostId, { ...inventory(), memoryTotal: 3 * GiB, memoryAvailable: Math.floor(2.8 * GiB) });
+        const first = queueNativeCreation(input);
+        expect(first.ok).toBe(true); if (!first.ok) return;
+        await advance();
+        const instance = nativeInstance(first.name)!;
+        heartbeat(hostId, { ...inventory(), memoryTotal: 3 * GiB, memoryAvailable: 2 * GiB,
+            instanceResources: evidence === 'missing' ? [] : [{ instanceId: instance.id, generation: instance.generation + 1,
+                uptimeSeconds: 60, cpuCores: 0.1, memoryBytes: Math.floor(0.8 * GiB) }] });
+        expect(queueNativeCreation(input)).toMatchObject({ ok: false, kind: 'conflict',
+            message: expect.stringContaining('3.00 GiB projected, 2.50 GiB usable') });
+    });
     it('waits when capacity disappears after admission without sending a create command', async () => {
         await prepareLibrary();
         const { queueNativeCreation } = await import('../../src/lib/server/provisioning/native.js');