Index lifecycle capacity checks instead of scanning retained jobs

This commit is contained in:
2026-10-06 12:54:10 -04:00
parent 37d380fa45
commit d2108eb604
6 changed files with 95 additions and 8 deletions
+2
View File
@@ -23,6 +23,7 @@ export function migrateNativeInstances(handle: Database.Database): void {
applied_at TEXT, verified_at TEXT 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_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 ( CREATE TABLE IF NOT EXISTS provisioning_safety_overrides (
deployment_id TEXT PRIMARY KEY REFERENCES server_deployments(id) ON DELETE CASCADE, 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 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_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_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 ( CREATE TABLE IF NOT EXISTS native_creation_cancellations (
creation_id TEXT PRIMARY KEY REFERENCES server_deployments(id), creation_id TEXT PRIMARY KEY REFERENCES server_deployments(id),
cleanup_id TEXT NOT NULL UNIQUE REFERENCES server_deployments(id) cleanup_id TEXT NOT NULL UNIQUE REFERENCES server_deployments(id)
+5 -1
View File
@@ -393,12 +393,16 @@ export function db(): Database.Database {
ON server_deployments(server, server_deleted_at); ON server_deployments(server, server_deleted_at);
CREATE INDEX IF NOT EXISTS idx_server_deployments_state CREATE INDEX IF NOT EXISTS idx_server_deployments_state
ON server_deployments(state, action, server_deleted_at); 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 CREATE INDEX IF NOT EXISTS idx_server_deployments_started_at
ON server_deployments(started_at); ON server_deployments(started_at);
CREATE INDEX IF NOT EXISTS idx_server_deployments_submission CREATE INDEX IF NOT EXISTS idx_server_deployments_submission
ON server_deployments(server, server_deleted_at, action, submitted_at); ON server_deployments(server, server_deleted_at, action, submitted_at);
CREATE INDEX IF NOT EXISTS idx_server_deployments_finished_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); installReportRevisions(handle);
} catch (err) { } catch (err) {
handle.close(); handle.close();
+2
View File
@@ -44,6 +44,8 @@ export function ensureHostSchema(handle: Database.Database): void {
-- causing a history scan for every queued command on Node's main thread. -- 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 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_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 ( CREATE TABLE IF NOT EXISTS ec_host_operation_payloads (
operation_id TEXT PRIMARY KEY REFERENCES ec_host_operations(id) ON DELETE CASCADE, operation_id TEXT PRIMARY KEY REFERENCES ec_host_operations(id) ON DELETE CASCADE,
payload TEXT NOT NULL payload TEXT NOT NULL
+17 -6
View File
@@ -8,18 +8,29 @@ interface Holder { id: string; action: string; host_id: string | null; target_ho
export function lifecycleHolders(): Holder[] { export function lifecycleHolders(): Holder[] {
ensureHostSchema(db()); 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, 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 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 FROM server_deployments d
LEFT JOIN native_provisioning n ON n.deployment_id=d.id LEFT JOIN native_provisioning n ON n.deployment_id=d.id
LEFT JOIN native_instances i ON i.id=n.instance_id LEFT JOIN native_instances i ON i.id=n.instance_id
LEFT JOIN ec_host_migrations m ON m.deployment_id=d.id LEFT JOIN ec_host_migrations m ON m.deployment_id=d.id
WHERE d.state IN ('running','awaiting-health') WHERE d.id IN (
OR d.id IN (SELECT value FROM json_each(?)) SELECT id FROM server_deployments WHERE state IN ('running','awaiting-health')
OR EXISTS (SELECT 1 FROM restart_rollout r WHERE r.operation_id=d.id AND r.healthy_at IS NULL) UNION SELECT value FROM json_each(?)
OR (m.remove_operation_id IS NOT NULL AND d.state NOT IN ('done','warning','cancelled')) UNION SELECT operation_id FROM restart_rollout WHERE healthy_at IS NULL
OR EXISTS (SELECT 1 FROM ec_host_operations o WHERE o.id IN (n.operation_id,m.remove_operation_id) UNION SELECT deployment_id FROM ec_host_migrations
AND o.state IN ('pending','accepted','running','unknown'))`) 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[]; .all(JSON.stringify([...preparingJobs])) as Holder[];
} }
+67
View File
@@ -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','host')).toContain('recovery');
expect(concurrencyBlocker('b','restart','other')).toBeNull(); 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/);
});
@@ -136,7 +136,8 @@ describe('provisioning reads with retained lifecycle history', () => {
const indexes = handle.pragma('index_list(server_deployments)') as { name: string }[]; const indexes = handle.pragma('index_list(server_deployments)') as { name: string }[];
expect(indexes.map(index => index.name)).toEqual(expect.arrayContaining([ 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_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([]); expect(handle.pragma('foreign_key_check')).toEqual([]);
closeDb(); closeDb();