Avoid scanning operation history on agent polls
This commit is contained in:
@@ -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
|
||||
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));
|
||||
AND state IN ('pending','accepted','running','unknown')`).all() as { host_id: string }[]).map(row => row.host_id))
|
||||
: new Set<string>();
|
||||
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;
|
||||
});
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user