diff --git a/layers/registry/server/utils/repository-purpose-effect.ts b/layers/registry/server/utils/repository-purpose-effect.ts index ce6fe40a..b3b3e992 100644 --- a/layers/registry/server/utils/repository-purpose-effect.ts +++ b/layers/registry/server/utils/repository-purpose-effect.ts @@ -14,20 +14,31 @@ export type RepositoryPurposeJudge = (state: RepositoryPurposeEvidence) => Promi export interface RepositoryIdentity { owner: string, repo: string } export async function listRepositoryPurposeCandidates(db: D1Database, now: number): Promise { + // Completed jobs are retained in `jobs` forever, so a correlated read over + // that table walks every finished classification for every candidate. Active + // jobs are bounded by in-flight work and reachable through the partial + // idx_jobs_active index, and terminal failures are moved to `failed_jobs`, + // where the one day retry wait is a failed_at index seek. + const blocking = await db.prepare(` + SELECT json_extract(payload,'$.owner') AS owner, json_extract(payload,'$.repo') AS repo + FROM jobs INDEXED BY idx_jobs_active + WHERE job_type='registry/repository-purpose' AND completed_at IS NULL AND failed_at IS NULL + UNION ALL + SELECT json_extract(payload,'$.owner') AS owner, json_extract(payload,'$.repo') AS repo + FROM failed_jobs INDEXED BY idx_failed_jobs_failed_at + WHERE job_type='registry/repository-purpose' AND failed_at>?1 + `).bind(now - 86400).all() + const blocked = JSON.stringify([...new Set(blocking.results.map(row => `${row.owner}/${row.repo}`))]) const result = await db.prepare(` SELECT r.owner,r.repo FROM repos r LEFT JOIN repository_purpose p ON p.owner=r.owner AND p.repo=r.repo WHERE r.broken_since IS NULL AND r.repo_skill_count>0 - AND (p.owner IS NULL OR p.prompt_version!=?1 OR p.model_id!=?4 OR p.evaluated_atp.evaluated_at) - AND NOT EXISTS ( - SELECT 1 FROM jobs j WHERE j.job_type='registry/repository-purpose' - AND json_extract(j.payload,'$.owner')=r.owner AND json_extract(j.payload,'$.repo')=r.repo - AND ((j.completed_at IS NULL AND j.failed_at IS NULL) OR j.failed_at>?3) - ) + AND (p.owner IS NULL OR p.prompt_version!=?1 OR p.model_id!=?3 OR p.evaluated_atp.evaluated_at) + AND NOT EXISTS (SELECT 1 FROM json_each(?4) AS b WHERE b.value=r.owner||'/'||r.repo) ORDER BY r.tree_truncated_at IS NOT NULL DESC, r.repo_skill_count>100 DESC, EXISTS(SELECT 1 FROM skills s WHERE s.owner=r.owner AND s.repo=r.repo) DESC, COALESCE(p.evaluated_at,0),r.owner,r.repo LIMIT 25 - `).bind(REPOSITORY_PURPOSE_PROMPT_VERSION, now - REPOSITORY_PURPOSE_REFRESH_SECONDS, now - 86400, REPOSITORY_PURPOSE_MODEL).all() + `).bind(REPOSITORY_PURPOSE_PROMPT_VERSION, now - REPOSITORY_PURPOSE_REFRESH_SECONDS, REPOSITORY_PURPOSE_MODEL, blocked).all() return result.results } diff --git a/test/unit/repository-purpose-effect.test.ts b/test/unit/repository-purpose-effect.test.ts index c335ab65..49eee2ad 100644 --- a/test/unit/repository-purpose-effect.test.ts +++ b/test/unit/repository-purpose-effect.test.ts @@ -1,3 +1,4 @@ +import type { SqliteD1 } from './helpers/d1-sqlite' import { describe, expect, it, vi } from 'vitest' import { classifyRepositoryPurpose, REPOSITORY_PURPOSE_REFRESH_SECONDS } from '../../layers/registry/server/utils/repository-purpose' import { checkRepositoryPurposeAdmission, listRepositoryPurposeCandidates, persistRepositoryPurpose, refreshRepositoryPurpose } from '../../layers/registry/server/utils/repository-purpose-effect' @@ -52,22 +53,55 @@ describe('purpose records', () => { finally { sqlite.close() } }) + /** + * The runtime holds active jobs in `jobs` and moves terminal failures to + * `failed_jobs`, deleting them from `jobs`. A dead-lettered classification + * leaves no `jobs` row at all, so the one day retry wait has to read + * `failed_jobs` or it never applies to a real failure. + */ it('checks large inventories first and avoids active or recently failed jobs', async () => { const sqlite = createSqliteD1(allMigrations()) try { sqlite.raw.exec(` - INSERT INTO repos(owner,repo,repo_skill_count) VALUES ('org','small',3),('org','large',400),('org','busy',200); - INSERT INTO jobs(id,queue,job_type,payload,available_at,failed_at) - VALUES ('busy','repo-review-sync','registry/repository-purpose','{"owner":"org","repo":"busy"}',0,NULL); + INSERT INTO repos(owner,repo,repo_skill_count) VALUES ('org','small',3),('org','large',400),('org','busy',200),('org','retry',100); + INSERT INTO jobs(id,queue,job_type,payload,available_at) + VALUES ('busy','repo-review-sync','registry/repository-purpose','{"owner":"org","repo":"busy"}',0); + INSERT INTO failed_jobs(id,queue,job_type,payload,exception,attempts,max_attempts,failed_at) + VALUES ('retry','repo-review-sync','registry/repository-purpose','{"owner":"org","repo":"retry"}','AI binding unavailable',5,5,99999); `) expect(await listRepositoryPurposeCandidates(sqlite.db, 100000)) .toEqual([{ owner: 'org', repo: 'large' }, { owner: 'org', repo: 'small' }]) - sqlite.raw.exec('UPDATE jobs SET failed_at=99999 WHERE id=\'busy\'') + sqlite.raw.exec('UPDATE failed_jobs SET failed_at=1 WHERE id=\'retry\'') expect(await listRepositoryPurposeCandidates(sqlite.db, 100000)) - .toEqual([{ owner: 'org', repo: 'large' }, { owner: 'org', repo: 'small' }]) - sqlite.raw.exec('UPDATE jobs SET failed_at=1 WHERE id=\'busy\'') - expect(await listRepositoryPurposeCandidates(sqlite.db, 100000)) - .toEqual([{ owner: 'org', repo: 'busy' }, { owner: 'org', repo: 'large' }, { owner: 'org', repo: 'small' }]) + .toEqual([{ owner: 'org', repo: 'large' }, { owner: 'org', repo: 'retry' }, { owner: 'org', repo: 'small' }]) + } + finally { sqlite.close() } + }) + + /** + * Completed classification jobs stay in `jobs` forever, so a candidate read + * that ranges over that table walks every finished job for every outer row. + * Blocking reads must reach `jobs` only through the partial active index and + * `failed_jobs` only through the failed_at window. + */ + it('never ranges over retained job history when blocking candidates', async () => { + const sqlite = createSqliteD1(allMigrations()) + try { + sqlite.raw.exec(` + INSERT INTO repos(owner,repo,repo_skill_count) VALUES ('org','small',3); + INSERT INTO jobs(id,queue,job_type,payload,available_at,completed_at) VALUES + ('done-1','repo-review-sync','registry/repository-purpose','{"owner":"other","repo":"past"}',0,5000), + ('done-2','repo-review-sync','registry/repository-purpose','{"owner":"other","repo":"older"}',0,4000); + `) + const { db, statements } = recordingDb(sqlite) + await listRepositoryPurposeCandidates(db, 100000) + expect(statements.length).toBeGreaterThan(0) + const details = statements.flatMap(({ sql, values }) => planDetails(sqlite, sql, values)) + const text = details.join('\n') + expect(text).not.toContain('idx_jobs_type') + expect(details.some(detail => detail.includes('idx_jobs_active'))).toBe(true) + expect(details.some(detail => detail.includes('idx_failed_jobs_failed_at'))).toBe(true) + expect(details.filter(detail => /^SCAN jobs(?! USING)/.test(detail))).toEqual([]) } finally { sqlite.close() } }) @@ -137,3 +171,32 @@ describe('purpose records', () => { finally { sqlite.close() } }) }) + +/** Wraps the D1 facade so every bound statement can be explained after the call. */ +function recordingDb(sqlite: SqliteD1): { db: D1Database, statements: Array<{ sql: string, values: unknown[] }> } { + const statements: Array<{ sql: string, values: unknown[] }> = [] + const prepare = sqlite.db.prepare.bind(sqlite.db) + const db = { + prepare(sql: string) { + const statement = prepare(sql) + return { + ...statement, + bind: (...values: unknown[]) => { + statements.push({ sql, values }) + return statement.bind(...values) + }, + } + }, + } as unknown as D1Database + return { db, statements } +} + +function planDetails(sqlite: SqliteD1, sql: string, values: unknown[]): string[] { + const expanded: unknown[] = [] + const normalized = sql.replace(/\?(\d+)/g, (_, raw: string) => { + expanded.push(values[Number(raw) - 1]) + return '?' + }) + return (sqlite.raw.prepare(`EXPLAIN QUERY PLAN ${normalized}`).all(...expanded) as Array<{ detail: string }>) + .map(row => row.detail) +}