Return queued lifecycle jobs before waking provisioning

src/lib/server/backends/schema.ts

@@ -23,6 +23,7 @@ export function migrateNativeInstances(handle: Database.Database): void {
             applied_at TEXT, verified_at TEXT
         );
         CREATE INDEX IF NOT EXISTS native_provisioning_instance ON native_provisioning(instance_id);
+        CREATE INDEX IF NOT EXISTS native_provisioning_operation ON native_provisioning(operation_id, deployment_id);
         CREATE TABLE IF NOT EXISTS provisioning_safety_overrides (
             deployment_id TEXT PRIMARY KEY REFERENCES server_deployments(id) ON DELETE CASCADE,
             actor_id TEXT NOT NULL, actor_name TEXT NOT NULL, confirmed_at TEXT NOT NULL
@@ -35,6 +36,7 @@ export function migrateNativeInstances(handle: Database.Database): void {
         );
         CREATE INDEX IF NOT EXISTS ec_host_migrations_source ON ec_host_migrations(source_host_id);
         CREATE INDEX IF NOT EXISTS ec_host_migrations_target ON ec_host_migrations(target_host_id);
+        CREATE INDEX IF NOT EXISTS ec_host_migrations_operation ON ec_host_migrations(remove_operation_id, deployment_id);
         CREATE TABLE IF NOT EXISTS native_creation_cancellations (
             creation_id TEXT PRIMARY KEY REFERENCES server_deployments(id),
             cleanup_id TEXT NOT NULL UNIQUE REFERENCES server_deployments(id)

src/lib/server/db/index.ts

@@ -393,12 +393,16 @@ export function db(): Database.Database {
             ON server_deployments(server, server_deleted_at);
             CREATE INDEX IF NOT EXISTS idx_server_deployments_state
             ON server_deployments(state, action, server_deleted_at);
+            CREATE INDEX IF NOT EXISTS idx_server_deployments_server_state
+            ON server_deployments(server, server_deleted_at, state, action);
             CREATE INDEX IF NOT EXISTS idx_server_deployments_started_at
             ON server_deployments(started_at);
             CREATE INDEX IF NOT EXISTS idx_server_deployments_submission
             ON server_deployments(server, server_deleted_at, action, submitted_at);
             CREATE INDEX IF NOT EXISTS idx_server_deployments_finished_at
-            ON server_deployments(finished_at) WHERE server_deleted_at IS NULL;`);
+            ON server_deployments(finished_at) WHERE server_deleted_at IS NULL;
+            CREATE INDEX IF NOT EXISTS idx_server_deployments_unsettled
+            ON server_deployments(id) WHERE state NOT IN ('done','warning','cancelled');`);
         installReportRevisions(handle);
     } catch (err) {
         handle.close();

src/lib/server/hosts/schema.ts

@@ -44,6 +44,8 @@ export function ensureHostSchema(handle: Database.Database): void {
         -- causing a history scan for every queued command on Node's main thread.
         CREATE INDEX IF NOT EXISTS ec_host_operations_kind ON ec_host_operations(host_id, kind, created_at);
         CREATE INDEX IF NOT EXISTS ec_host_operations_kind_state ON ec_host_operations(kind, state);
+        CREATE INDEX IF NOT EXISTS ec_host_operations_active ON ec_host_operations(state, id)
+            WHERE state IN ('pending','accepted','running','unknown');
         CREATE TABLE IF NOT EXISTS ec_host_operation_payloads (
             operation_id TEXT PRIMARY KEY REFERENCES ec_host_operations(id) ON DELETE CASCADE,
             payload TEXT NOT NULL

src/lib/server/provisioning/concurrency.ts

@@ -8,18 +8,29 @@ interface Holder { id: string; action: string; host_id: string | null; target_ho
 
 export function lifecycleHolders(): Holder[] {
     ensureHostSchema(db());
+    // Find capacity owners through their active state or outstanding command.
+    // A correlated OR over d used to walk every retained job and fetch its agent
+    // receipt on every admission, even when all that history was completed.
     return db().prepare(`SELECT d.id,d.action,i.host_id,m.target_host_id,
         CASE WHEN m.remove_operation_id IS NOT NULL AND m.removed_at IS NULL THEN 1 ELSE 0 END AS removing
         FROM server_deployments d
         LEFT JOIN native_provisioning n ON n.deployment_id=d.id
         LEFT JOIN native_instances i ON i.id=n.instance_id
         LEFT JOIN ec_host_migrations m ON m.deployment_id=d.id
-        WHERE d.state IN ('running','awaiting-health')
-        OR d.id IN (SELECT value FROM json_each(?))
-        OR EXISTS (SELECT 1 FROM restart_rollout r WHERE r.operation_id=d.id AND r.healthy_at IS NULL)
-        OR (m.remove_operation_id IS NOT NULL AND d.state NOT IN ('done','warning','cancelled'))
-        OR EXISTS (SELECT 1 FROM ec_host_operations o WHERE o.id IN (n.operation_id,m.remove_operation_id)
-            AND o.state IN ('pending','accepted','running','unknown'))`)
+        WHERE d.id IN (
+            SELECT id FROM server_deployments WHERE state IN ('running','awaiting-health')
+            UNION SELECT value FROM json_each(?)
+            UNION SELECT operation_id FROM restart_rollout WHERE healthy_at IS NULL
+            UNION SELECT deployment_id FROM ec_host_migrations
+                WHERE remove_operation_id IS NOT NULL AND deployment_id IN (
+                    SELECT id FROM server_deployments WHERE state NOT IN ('done','warning','cancelled'))
+            UNION SELECT n.deployment_id FROM ec_host_operations o
+                JOIN native_provisioning n ON n.operation_id=o.id
+                WHERE o.state IN ('pending','accepted','running','unknown')
+            UNION SELECT m.deployment_id FROM ec_host_operations o
+                JOIN ec_host_migrations m ON m.remove_operation_id=o.id
+                WHERE o.state IN ('pending','accepted','running','unknown')
+        )`)
         .all(JSON.stringify([...preparingJobs])) as Holder[];
 }
 

src/routes/api/provisioning/+server.ts

@@ -2,7 +2,7 @@ import { json } from '@sveltejs/kit';
 import { deploymentListPage } from '$lib/server/provisioning/history.js';
 import { requestLifecycle } from '$lib/server/provisioning/index.js';
 import { provisioningActions, runtimeActions, type ProvisioningAction, type RuntimeAction } from '$lib/provisioning.js';
-import { kick } from '$lib/server/pipeline/worker.js';
+import { kickProvisioning } from '$lib/server/pipeline/worker.js';
 
 export function GET({ url }: { url: URL }) {
     const params = url.searchParams;
@@ -25,7 +25,9 @@ export async function POST({ request }: { request: Request }) {
     try {
         const result = requestLifecycle(body.server, body.action as RuntimeAction);
         if (!result.ok) return json({ error: result.error }, { status: result.status });
-        void kick().catch(err => console.error('[provisioning] worker kick failed:', err));
+        setImmediate(() => {
+            void kickProvisioning().catch(err => console.error('[provisioning] worker kick failed:', err));
+        });
         return json(result, { status: 202 });
     } catch (err) { return json({ error: (err as Error).message }, { status: 400 }); }
 }

src/routes/api/servers/+server.ts

@@ -1,5 +1,5 @@
 import { fleetSnapshot } from '$lib/server/fleet/snapshot.js';
-import { kick } from '$lib/server/pipeline/worker.js';
+import { kickProvisioning } from '$lib/server/pipeline/worker.js';
 import { json } from '@sveltejs/kit';
 import { queueCreateServer } from '$lib/server/provisioning/index.js';
 import type { GameMode, ZombieGame } from '$lib/server/fleet/map-aliases.js';
@@ -37,6 +37,9 @@ export async function POST({ request }: { request: Request }) {
             : 500;
         return json({ ok: false, error: result.message, kind: result.kind }, { status });
     }
-    void kick().catch((err) => console.error('[create] worker kick failed:', err));
+    // Return the durable job ID before the worker's synchronous admission work.
+    setImmediate(() => {
+        void kickProvisioning().catch((err) => console.error('[create] worker kick failed:', err));
+    });
     return json({ ok: true, name: result.name, port: result.port, deploymentId: result.deploymentId }, { status: 202 });
 }

src/routes/api/servers/[name]/+server.ts

@@ -1,4 +1,4 @@
-import { kick } from '$lib/server/pipeline/worker.js';
+import { kickProvisioning } from '$lib/server/pipeline/worker.js';
 import { json } from '@sveltejs/kit';
 import { destroyServer } from '$lib/server/provisioning/index.js';
 import { serverSnapshot } from '$lib/server/fleet/snapshot.js';
@@ -17,6 +17,8 @@ export async function DELETE({ params }: { params: { name: string } }) {
         const status = result.kind === 'not-found' ? 404 : result.kind === 'conflict' ? 409 : 500;
         return json({ ok: false, error: result.message, kind: result.kind, warnings: result.warnings }, { status });
     }
-    void kick().catch((err) => console.error('[destroy] worker kick failed:', err));
+    setImmediate(() => {
+        void kickProvisioning().catch((err) => console.error('[destroy] worker kick failed:', err));
+    });
     return json({ ok: true, deploymentId: result.deploymentId }, { status: 202 });
 }

tests/unit/concurrency.test.ts

@@ -43,3 +43,70 @@ it('enforces per-host removal and recovery limits independently of other hosts',
     expect(concurrencyBlocker('b','restart','host')).toContain('recovery');
     expect(concurrencyBlocker('b','restart','other')).toBeNull();
 });
+
+it('retains every capacity owner across failure, cancellation and migration without counting completed history', async () => {
+    const { db } = await import('../../src/lib/server/db/index.js');
+    const { createHost } = await import('../../src/lib/server/hosts/index.js');
+    const { lifecycleHolders, preparingJobs } = await import('../../src/lib/server/provisioning/concurrency.js');
+    const h = db(), host = createHost({name:'Source'});
+    const add = (id: string, state: string, command?: string, migration = false) => {
+        h.prepare("INSERT INTO server_deployments(id,server,action,state,stage,started_at,triggered_by) VALUES(?,?,'rebuild',?,'test','2026-01-01','test')").run(id,id,state);
+        h.prepare("INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at) VALUES(?,?,?,'lib','img',1,'2026-01-01')").run(id,id,host.id);
+        h.prepare('INSERT INTO native_provisioning(deployment_id,instance_id,generation,operation_id) VALUES(?,?,1,?)').run(id,id,migration ? null : command ? id : null);
+        if (command) h.prepare("INSERT INTO ec_host_operations(id,host_id,kind,state,payload,created_at,updated_at) VALUES(?,?,'instance.rebuild',?,'{}','2026-01-01','2026-01-01')").run(id,host.id,command);
+        if (migration) h.prepare('INSERT INTO ec_host_migrations(deployment_id,source_host_id,instance_id,target_host_id,remove_operation_id) VALUES(?,?,?,?,?)').run(id,host.id,id,'target',command ? id : null);
+    };
+    add('running','running','running');
+    add('health','awaiting-health','completed');
+    add('prepared','queued');
+    add('recovery','failed','failed');
+    add('cancelled-command','cancelled','unknown');
+    add('done-command','done','pending');
+    add('failed-migration','failed','completed',true);
+    add('unknown-migration','legacy-unknown','completed',true);
+    add('cancelled-migration','cancelled','accepted',true);
+    add('waiting-migration','waiting','running',true);
+    add('finished','done','completed');
+    add('cancelled-finished','cancelled','completed',true);
+    add('finished-migration','warning','completed',true);
+    add('queued','queued');
+    h.prepare("INSERT INTO servers(name) VALUES('recovery')").run();
+    h.prepare("INSERT INTO restart_rollout(server,operation_id,requested_at) VALUES('recovery','recovery','2026-01-01')").run();
+    preparingJobs.add('prepared');
+    try {
+        const holders = lifecycleHolders();
+        expect(holders.map(row=>row.id).sort()).toEqual(['running','health','prepared','recovery','cancelled-command','done-command','failed-migration','unknown-migration','cancelled-migration','waiting-migration'].sort());
+        expect(holders.find(row=>row.id==='waiting-migration')).toMatchObject({host_id:host.id,target_host_id:'target',removing:1});
+    } finally { preparingJobs.clear(); }
+});
+
+it('seeks capacity owners through active indexes instead of scanning retained jobs and receipts', async () => {
+    const { db } = await import('../../src/lib/server/db/index.js');
+    const { createHost } = await import('../../src/lib/server/hosts/index.js');
+    const { lifecycleHolders } = await import('../../src/lib/server/provisioning/concurrency.js');
+    const h = db(), host = createHost({name:'History host'});
+    h.transaction(() => {
+        const job = h.prepare("INSERT INTO server_deployments(id,server,action,state,stage,started_at,triggered_by) VALUES(?,'old','restart','done','Done','2026-01-01','test')");
+        const operation = h.prepare("INSERT INTO ec_host_operations(id,host_id,kind,state,payload,created_at,updated_at) VALUES(?,?,'instance.restart','completed','{}','2026-01-01','2026-01-01')");
+        h.prepare("INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at) VALUES('i','old',?,'lib','img',1,'2026-01-01')").run(host.id);
+        const native = h.prepare("INSERT INTO native_provisioning(deployment_id,instance_id,generation,operation_id) VALUES(?,'i',1,?)");
+        const migration = h.prepare("INSERT INTO ec_host_migrations(deployment_id,source_host_id,instance_id,remove_operation_id,removed_at) VALUES(?,?,'i',?,'2026-01-01')");
+        for (let n=0;n<4000;n++) {
+            const id=String(n); job.run(id); operation.run(id,host.id); native.run(id,id); migration.run(id,host.id,id);
+        }
+    })();
+    const prepare = vi.spyOn(h,'prepare');
+    let sql: string;
+    try {
+        expect(lifecycleHolders()).toEqual([]);
+        sql = prepare.mock.calls.find(([sql])=>sql.startsWith('SELECT d.id,d.action'))![0];
+    } finally { prepare.mockRestore(); }
+    const plan = h.prepare(`EXPLAIN QUERY PLAN ${sql}`).all('[]') as {detail:string}[];
+    const details = plan.map(row=>row.detail).join('\n');
+    for (const index of ['idx_server_deployments_state','idx_server_deployments_unsettled','ec_host_operations_active','native_provisioning_operation','ec_host_migrations_operation']) {
+        expect(details).toContain(index);
+    }
+    expect(details).not.toMatch(/SCAN (?:d|n|m|o)\b/);
+    // Scanning the small partial index of unsettled IDs is intentional.
+    expect(details).not.toMatch(/SCAN server_deployments(?! USING INDEX idx_server_deployments_unsettled)\b/);
+});

tests/unit/provisioning-history-performance.test.ts

@@ -136,7 +136,8 @@ describe('provisioning reads with retained lifecycle history', () => {
             const indexes = handle.pragma('index_list(server_deployments)') as { name: string }[];
             expect(indexes.map(index => index.name)).toEqual(expect.arrayContaining([
                 'idx_server_deployments_server', 'idx_server_deployments_state', 'idx_server_deployments_started_at',
-                'idx_server_deployments_submission', 'idx_server_deployments_finished_at'
+                'idx_server_deployments_submission', 'idx_server_deployments_finished_at',
+                'idx_server_deployments_server_state', 'idx_server_deployments_unsettled'
             ]));
             expect(handle.pragma('foreign_key_check')).toEqual([]);
             closeDb();

tests/unit/provisioning-response.test.ts

@@ -0,0 +1,54 @@
+import { afterEach, beforeEach, expect, it, vi } from 'vitest';
+import { queueCreateServer, destroyServer, requestLifecycle } from '../../src/lib/server/provisioning/index.js';
+import { kick, kickProvisioning } from '../../src/lib/server/pipeline/worker.js';
+import { POST as create } from '../../src/routes/api/servers/+server.js';
+import { DELETE as destroy } from '../../src/routes/api/servers/[name]/+server.js';
+import { POST as lifecycle } from '../../src/routes/api/provisioning/+server.js';
+
+vi.mock('../../src/lib/server/provisioning/index.js', () => ({
+    queueCreateServer: vi.fn(), destroyServer: vi.fn(), requestLifecycle: vi.fn()
+}));
+vi.mock('../../src/lib/server/pipeline/worker.js', () => ({
+    kick: vi.fn(async () => {}), kickProvisioning: vi.fn(async () => {})
+}));
+vi.mock('../../src/lib/server/fleet/snapshot.js', () => ({ fleetSnapshot: vi.fn(), serverSnapshot: vi.fn() }));
+vi.mock('../../src/lib/server/provisioning/history.js', () => ({ deploymentListPage: vi.fn() }));
+
+beforeEach(() => {
+    vi.clearAllMocks();
+    vi.useFakeTimers({ toFake: ['setImmediate'] });
+});
+afterEach(() => vi.useRealTimers());
+
+const request = (body: unknown) => new Request('http://localhost/api/servers', {
+    method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body)
+});
+const routes = [
+    { name: 'create', queue: queueCreateServer, invoke: () => create({ request: request({ game: 't5', gameMode: 'zombie', mapSlug: 'theater' }) }) },
+    { name: 'destroy', queue: destroyServer, invoke: () => destroy({ params: { name: 'server' } }) },
+    { name: 'restart', queue: requestLifecycle, invoke: () => lifecycle({ request: request({ server: 'server', action: 'restart' }) }) }
+];
+
+it.each(routes)('returns the queued $name job before starting provisioning or unrelated observers', async ({ queue, invoke }) => {
+    vi.mocked(queue).mockReturnValue({ ok: true, name: 'server', port: 28960, deploymentId: 'job', warnings: [] });
+    const response = await invoke();
+    expect(response.status).toBe(202);
+    expect(await response.json()).toMatchObject({ ok: true, deploymentId: 'job' });
+    expect(queue).toHaveBeenCalledTimes(1);
+    expect(kickProvisioning).not.toHaveBeenCalled();
+    expect(kick).not.toHaveBeenCalled();
+
+    await vi.runAllTimersAsync();
+    expect(kickProvisioning).toHaveBeenCalledTimes(1);
+    expect(kick).not.toHaveBeenCalled();
+});
+
+it.each(routes)('does not wake workers when $name admission fails', async ({ queue, invoke }) => {
+    vi.mocked(queue).mockReturnValue({ ok: false, kind: 'conflict', message: 'Busy', error: 'Busy', status: 409, warnings: [] });
+    const response = await invoke();
+    expect(response.status).toBe(409);
+    expect(await response.json()).toMatchObject({ error: 'Busy' });
+    await vi.runAllTimersAsync();
+    expect(kickProvisioning).not.toHaveBeenCalled();
+    expect(kick).not.toHaveBeenCalled();
+});