diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index 76ce5cd..9564db2 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -522,11 +522,14 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = ' .all(hostId, Number(host.readiness === 'ready' || host.mode === 'draining'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false), ...(lane === 'health' ? [new Date(Date.now() - 30_000).toISOString()] : [])) as OperationDeliveryRow[]; const preparationLimit = fleetSettings().preparationConcurrency; - const preparing = new Set((database().prepare(`SELECT DISTINCT host_id FROM ec_host_operations - WHERE kind IN ('library.download','library.import') AND last_delivered_at IS NOT NULL - AND state IN ('pending','accepted','running','unknown')`).all() as { host_id: string }[]).map(row => row.host_id)); + const libraryKinds = ['library.download','library.import']; + const preparing = preparationLimit && rows.some(operation => libraryKinds.includes(operation.kind)) + ? new Set((database().prepare(`SELECT DISTINCT host_id FROM ec_host_operations + WHERE kind IN ('library.download','library.import') AND last_delivered_at IS NOT NULL + AND state IN ('pending','accepted','running','unknown')`).all() as { host_id: string }[]).map(row => row.host_id)) + : new Set(); const admitted = rows.filter(operation => { - if (!['library.download','library.import'].includes(operation.kind) || preparing.has(hostId) || !preparationLimit) return true; + if (!libraryKinds.includes(operation.kind) || preparing.has(hostId) || !preparationLimit) return true; if (preparing.size >= preparationLimit) return false; preparing.add(hostId); return true; }); diff --git a/src/lib/server/hosts/schema.ts b/src/lib/server/hosts/schema.ts index 1e0cddf..af4ed3e 100644 --- a/src/lib/server/hosts/schema.ts +++ b/src/lib/server/hosts/schema.ts @@ -43,6 +43,7 @@ export function ensureHostSchema(handle: Database.Database): void { -- history. The queue index cannot narrow those lookups beyond host_id, -- 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 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 diff --git a/tests/unit/host-request-cost.test.ts b/tests/unit/host-request-cost.test.ts index 1d5e97f..81c0f26 100644 --- a/tests/unit/host-request-cost.test.ts +++ b/tests/unit/host-request-cost.test.ts @@ -23,6 +23,47 @@ async function enrolled() { return { ...hosts, ...context, hostId: host.id, credential }; } +describe('agent preparation admission', () => { + it('skips fleet preparation lookups for empty, health, read and unlimited library polls', async () => { + const api = await enrolled(); + const { db } = await import('../../src/lib/server/db/index.js'); + const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js'); + saveFleetSettings({...fleetSettings(),preparationConcurrency:1}); + const prepare = vi.spyOn(db(),'prepare'); + expect(api.pendingHostOperations(api.hostId)).toEqual([]); + api.queueHostOperation(api.hostId,'instance.inspect',{}); + api.pendingHostOperations(api.hostId,'health'); + api.queueHostOperation(api.hostId,'files.read',{path:'server.cfg'}); + api.pendingHostOperations(api.hostId,'read'); + saveFleetSettings({...fleetSettings(),preparationConcurrency:0}); + const download = api.queueHostOperation(api.hostId,'library.download',{}); + expect(api.pendingHostOperations(api.hostId,'write').map(op=>op.id)).toContain(download.id); + expect(prepare.mock.calls.some(([sql])=>sql.includes('SELECT DISTINCT host_id'))).toBe(false); + }); + + it('uses an indexed lookup and counts delivered preparation by host until all its jobs finish', async () => { + const api = await enrolled(); + const other = api.createHost({name:'Other host'}); + api.enrollHost(api.createEnrollment(other.id).token,inventory); + const { db } = await import('../../src/lib/server/db/index.js'); + const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js'); + saveFleetSettings({...fleetSettings(),preparationConcurrency:1}); + const download = api.queueHostOperation(api.hostId,'library.download',{}); + const imported = api.queueHostOperation(api.hostId,'library.import',{}); + const waiting = api.queueHostOperation(other.id,'library.download',{}); + const prepare = vi.spyOn(db(),'prepare'); + expect(api.pendingHostOperations(api.hostId,'write').map(op=>op.id)).toEqual([download.id,imported.id]); + const sql = prepare.mock.calls.find(([sql])=>sql.includes('SELECT DISTINCT host_id'))![0]; + const plan = db().prepare(`EXPLAIN QUERY PLAN ${sql}`).all() as {detail:string}[]; + expect(plan.map(row=>row.detail).join('\n')).toContain('USING INDEX ec_host_operations_kind_state (kind=? AND state=?)'); + expect(api.pendingHostOperations(other.id,'write')).toEqual([]); + api.reportHostOperation(api.hostId,download.id,{state:'completed'}); + expect(api.pendingHostOperations(other.id,'write')).toEqual([]); + api.reportHostOperation(api.hostId,imported.id,{state:'failed'}); + expect(api.pendingHostOperations(other.id,'write').map(op=>op.id)).toEqual([waiting.id]); + }); +}); + describe('agent credential usage writes', () => { it('writes on first use and at most once per minute, including after reopening the database', async () => { const api = await enrolled();