fix(registry): bound repository purpose candidate blocking reads (#582)

Co-authored-by: Harlan GitHub Agent <harlan-github-agent[bot]@users.noreply.github.com>

Author: harlan-github-agent[bot]

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<RepositoryIdentity[]> {
+  // 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<RepositoryIdentity>()
+  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_at<?2 OR r.pushed_at>p.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_at<?2 OR r.pushed_at>p.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<RepositoryIdentity>()
+  `).bind(REPOSITORY_PURPOSE_PROMPT_VERSION, now - REPOSITORY_PURPOSE_REFRESH_SECONDS, REPOSITORY_PURPOSE_MODEL, blocked).all<RepositoryIdentity>()
   return result.results
 }
 

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)
+}