diff --git a/docs/arch/cron.md b/docs/arch/cron.md index da41ad4a..9f0b0980 100644 --- a/docs/arch/cron.md +++ b/docs/arch/cron.md @@ -2,7 +2,7 @@ Generated from literal `defineScheduledTask` declarations. Run `pnpm cron:docs` after schedule changes. -Tasks: 25. Unique Cloudflare triggers: 16. +Tasks: 26. Unique Cloudflare triggers: 16. | Task | Cron | Source | | --- | --- | --- | @@ -11,6 +11,7 @@ Tasks: 25. Unique Cloudflare triggers: 16. | `ai-generate-submit` | `15 * * * *` | `layers/registry/server/tasks/ai-generate-submit.ts` | | `ai-ready:cron` | `*/5 * * * *` | `server/tasks/ai-ready-cron.ts` | | `backfill-skill-assets` | `*/5 * * * *` | `layers/registry/server/tasks/backfill-skill-assets.ts` | +| `backfill-skill-sources` | `*/5 * * * *` | `layers/registry/server/tasks/backfill-skill-sources.ts` | | `classify-repository-purpose` | `0 * * * *` | `layers/registry/server/tasks/classify-repository-purpose.ts` | | `cleanup-auto-index-rate-limits` | `0 * * * *` | `layers/registry/server/tasks/cleanup-auto-index-rate-limits.ts` | | `detect-star-surges` | `30 4 * * *` | `layers/registry/server/tasks/detect-star-surges.ts` | diff --git a/layers/registry/server/tasks/backfill-skill-sources.ts b/layers/registry/server/tasks/backfill-skill-sources.ts new file mode 100644 index 00000000..bc9dec26 --- /dev/null +++ b/layers/registry/server/tasks/backfill-skill-sources.ts @@ -0,0 +1,56 @@ +/// + +import { resolveCloudflareBindings } from '@harlan-zw/nuxt-cloudflare/bindings' +import { runObservedScheduledTask } from '~~/server/utils/scheduled-run' +import { reportJobRun } from '~~/server/utils/sync-job-reporter' +import { observedSchedulePolicy } from '#shared/schedule-policy' +import { backfillSkillSources } from '../utils/skill-source-backfill' + +const CRON = '*/5 * * * *' + +/** + * Scheduled task: copy every stored SKILL.md body to R2 (ADR-0019). + * + * The sync writes new bodies to R2 as it goes. This task walks the rows that + * predate that, 200 a run by rowid, so every row's hash names an object + * before the D1 column is nulled. Once the walk reaches the end of the table + * each run is one indexed read and no puts. + */ +export default defineScheduledTask({ + name: 'backfill-skill-sources', + cron: '*/5 * * * *', + description: 'Copy stored SKILL.md bodies to R2 by content hash', + async run({ context }) { + const env = resolveCloudflareBindings(context) + const db = env?.DB as D1Database | undefined + const bucket = env?.PUBLIC_ARTIFACTS + if (!env || !db || !bucket) { + emitOperationalEvent(createWideEvent({ operation: 'backfill-skill-sources', outcome: 'binding-missing' })) + return { result: { error: 'no-db' } } + } + + return await runObservedScheduledTask({ + db, + env, + context, + policy: observedSchedulePolicy('backfill-skill-sources'), + }, async () => { + const startedAt = Date.now() + const report = await backfillSkillSources(db, bucket, { now: Math.floor(startedAt / 1000) }) + emitOperationalEvent(createWideEvent({ + 'operation': 'backfill-skill-sources', + 'outcome': report.failed > 0 ? 'partial' : 'completed', + 'scanned.count': report.scanned, + 'success.count': report.stored + report.existing, + 'failed.count': report.failed, + })) + await reportJobRun(db, 'backfill-skill-sources', { + cron: CRON, + status: report.failed > 0 ? 'partial' : 'ok', + durationMs: Date.now() - startedAt, + error: report.failed > 0 ? `${report.failed} R2 put failed; cursor held at ${report.position}` : null, + }) + return { result: report } + }) + }, +}) diff --git a/layers/registry/server/utils/skill-source-backfill.spec.ts b/layers/registry/server/utils/skill-source-backfill.spec.ts new file mode 100644 index 00000000..57bbabf5 --- /dev/null +++ b/layers/registry/server/utils/skill-source-backfill.spec.ts @@ -0,0 +1,108 @@ +import type { SkillSourceBucket } from './skill-source-store' +import Database from 'better-sqlite3' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { backfillSkillSources, SKILL_SOURCE_BACKFILL_CURSOR } from './skill-source-backfill' +import { skillSourceR2Key } from './skill-source-store' + +describe('backfillSkillSources', () => { + let sqlite: Database.Database + + beforeEach(() => { + sqlite = new Database(':memory:') + sqlite.exec(` + CREATE TABLE skills ( + owner TEXT NOT NULL, repo TEXT NOT NULL, name TEXT NOT NULL, + rendered_raw TEXT, rendered_raw_sha256 TEXT, + PRIMARY KEY (owner, repo, name) + ); + CREATE TABLE registry_cursors ( + name TEXT PRIMARY KEY, position INTEGER NOT NULL DEFAULT 0, updated_at INTEGER NOT NULL + ); + `) + const insert = sqlite.prepare(`INSERT INTO skills (owner, repo, name, rendered_raw, rendered_raw_sha256) VALUES ('acme', 'skills', ?, ?, ?)`) + insert.run('one', '# One', 'a'.repeat(64)) + insert.run('unrendered', null, null) + insert.run('two', '# Two', 'b'.repeat(64)) + insert.run('three', '# Three', 'c'.repeat(64)) + }) + + afterEach(() => sqlite.close()) + + it('walks the table by rowid across runs and stores each body once', async () => { + const bucket = memoryBucket() + + const first = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 1, limit: 2 }) + const second = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 2, limit: 2 }) + const third = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 3, limit: 2 }) + + expect(first).toEqual({ scanned: 2, stored: 2, existing: 0, failed: 0, position: 3 }) + expect(second).toEqual({ scanned: 1, stored: 1, existing: 0, failed: 0, position: 4 }) + expect(third).toEqual({ scanned: 0, stored: 0, existing: 0, failed: 0, position: 4 }) + expect([...bucket.objects.keys()].sort()).toEqual( + ['a', 'b', 'c'].map(hex => skillSourceR2Key(hex.repeat(64))).sort(), + ) + expect(cursor()).toBe(4) + }) + + it('holds the cursor before a failed put so the next run retries it', async () => { + const bucket = memoryBucket(new Set([skillSourceR2Key('b'.repeat(64))])) + + const report = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 1, limit: 10 }) + + expect(report).toEqual({ scanned: 2, stored: 1, existing: 0, failed: 1, position: 1 }) + expect(cursor()).toBe(1) + }) + + it('reports an object that already exists without storing it again', async () => { + const bucket = memoryBucket() + bucket.objects.set(skillSourceR2Key('a'.repeat(64)), new Uint8Array()) + + const report = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 1, limit: 1 }) + + expect(report).toEqual({ scanned: 1, stored: 0, existing: 1, failed: 0, position: 1 }) + }) + + function cursor(): number | undefined { + return sqlite.prepare(`SELECT position FROM registry_cursors WHERE name = ?`).pluck().get(SKILL_SOURCE_BACKFILL_CURSOR) as number | undefined + } +}) + +function memoryBucket(failKeys = new Set()): SkillSourceBucket & { objects: Map } { + const objects = new Map() + return { + objects, + put: (async (key: string, value: Uint8Array, options: { onlyIf?: { etagDoesNotMatch?: string } }) => { + if (failKeys.has(key)) + throw new Error('r2 unavailable') + if (options.onlyIf?.etagDoesNotMatch === '*' && objects.has(key)) + return null + objects.set(key, value) + return { key } + }) as unknown as SkillSourceBucket['put'], + get: (async () => null) as unknown as SkillSourceBucket['get'], + head: (async () => null) as unknown as SkillSourceBucket['head'], + } +} + +function wrapSqlite(sqlite: Database.Database): D1Database { + return { + prepare(sql: string) { + return { + bind(...params: unknown[]) { + return { + async all() { + return { results: sqlite.prepare(sql).all(...params) as T[] } + }, + async first() { + return (sqlite.prepare(sql).get(...params) as T | undefined) ?? null + }, + async run() { + const info = sqlite.prepare(sql).run(...params) + return { meta: { changes: info.changes } } + }, + } + }, + } + }, + } as unknown as D1Database +} diff --git a/layers/registry/server/utils/skill-source-backfill.ts b/layers/registry/server/utils/skill-source-backfill.ts new file mode 100644 index 00000000..f41d7723 --- /dev/null +++ b/layers/registry/server/utils/skill-source-backfill.ts @@ -0,0 +1,87 @@ +/// + +import type { SkillSourceBucket } from './skill-source-store' +import { putSkillSource } from './skill-source-store' + +export const SKILL_SOURCE_BACKFILL_CURSOR = 'skill-sources-r2-v1' + +/** + * Rows one run copies. Each is one R2 put, and a put that already exists is + * one conditional request. 200 keeps a run under the subrequest budget with + * room for its D1 reads, and copies the 25,145 rows of 2026-11-12 in about + * ten hours of five-minute runs. + */ +export const SKILL_SOURCE_BACKFILL_BATCH = 200 + +interface SourceRow { + rowid: number + rendered_raw: string + rendered_raw_sha256: string +} + +export interface SkillSourceBackfillReport { + /** Rows read past the cursor this run. Zero means the walk is complete. */ + scanned: number + stored: number + existing: number + failed: number + /** The rowid the next run starts after. */ + position: number +} + +/** + * Copy the next batch of stored SKILL.md bodies to R2, by rowid. + * + * The cursor only advances past rows whose put succeeded or found the object + * already there. A failed put holds the cursor at the row before it, so the + * next run retries from that row instead of leaving a hole the readers would + * fall back to D1 for forever. + * + * A row the sync rewrites after this walk passes it is covered by the sync's + * own dual write, so one pass is enough. + */ +export async function backfillSkillSources( + db: D1Database, + bucket: SkillSourceBucket, + input: { now: number, limit?: number }, +): Promise { + const limit = input.limit ?? SKILL_SOURCE_BACKFILL_BATCH + const cursor = await db.prepare(`SELECT position FROM registry_cursors WHERE name = ?`) + .bind(SKILL_SOURCE_BACKFILL_CURSOR) + .first<{ position: number }>() + const start = cursor?.position ?? 0 + + const rows = await db.prepare( + `SELECT rowid, rendered_raw, rendered_raw_sha256 + FROM skills + WHERE rowid > ? + AND rendered_raw IS NOT NULL + AND rendered_raw_sha256 IS NOT NULL + ORDER BY rowid + LIMIT ?`, + ).bind(start, limit).all() + + const report: SkillSourceBackfillReport = { scanned: 0, stored: 0, existing: 0, failed: 0, position: start } + for (const row of rows.results ?? []) { + report.scanned += 1 + const result = await putSkillSource(bucket, { body: row.rendered_raw, contentSha256: row.rendered_raw_sha256 }) + if (result._tag === 'failed') { + report.failed += 1 + console.warn(`[backfill-skill-sources] put failed at rowid ${row.rowid}: ${result.reason}`) + break + } + if (result._tag === 'stored') + report.stored += 1 + else + report.existing += 1 + report.position = row.rowid + } + + if (report.position !== start) { + await db.prepare( + `INSERT INTO registry_cursors (name, position, updated_at) VALUES (?, ?, ?) + ON CONFLICT(name) DO UPDATE SET position = excluded.position, updated_at = excluded.updated_at`, + ).bind(SKILL_SOURCE_BACKFILL_CURSOR, report.position, input.now).run() + } + return report +} diff --git a/layers/registry/server/utils/skill-source-store.spec.ts b/layers/registry/server/utils/skill-source-store.spec.ts new file mode 100644 index 00000000..249b44ea --- /dev/null +++ b/layers/registry/server/utils/skill-source-store.spec.ts @@ -0,0 +1,73 @@ +import type { SkillSourceBucket } from './skill-source-store' +import { describe, expect, it } from 'vitest' +import { skillContentSha256 } from './skill-content-hash' +import { hasSkillSource, putSkillSource, readSkillSource, skillSourceR2Key } from './skill-source-store' + +interface StoredObject { + bytes: Uint8Array + sha256: unknown + customMetadata: unknown +} + +function memoryBucket(objects = new Map(), failPut = false): SkillSourceBucket & { objects: Map } { + return { + objects, + put: (async (key: string, value: Uint8Array, options: { onlyIf?: { etagDoesNotMatch?: string }, sha256?: unknown, customMetadata?: unknown }) => { + if (failPut) + throw new Error('r2 unavailable') + if (options.onlyIf?.etagDoesNotMatch === '*' && objects.has(key)) + return null + objects.set(key, { bytes: value, sha256: options.sha256, customMetadata: options.customMetadata }) + return { key } + }) as unknown as SkillSourceBucket['put'], + get: (async (key: string) => { + const object = objects.get(key) + return object ? { text: async () => new TextDecoder().decode(object.bytes) } : null + }) as unknown as SkillSourceBucket['get'], + head: (async (key: string) => (objects.has(key) ? { key } : null)) as unknown as SkillSourceBucket['head'], + } +} + +describe('skill source store', () => { + it('stores a body under its hash and reports a repeat as existing', async () => { + const bucket = memoryBucket() + const body = '---\nname: one\n---\n# One' + const contentSha256 = await skillContentSha256(body) + + const first = await putSkillSource(bucket, { body, contentSha256 }) + const second = await putSkillSource(bucket, { body, contentSha256 }) + + expect(first).toEqual({ _tag: 'stored', key: skillSourceR2Key(contentSha256) }) + expect(second).toEqual({ _tag: 'existing', key: skillSourceR2Key(contentSha256) }) + expect(bucket.objects.get(skillSourceR2Key(contentSha256))?.sha256).toBe(contentSha256) + expect(await hasSkillSource(bucket, contentSha256)).toBe(true) + }) + + it('returns a failure value instead of throwing when the put fails', async () => { + const bucket = memoryBucket(new Map(), true) + + const result = await putSkillSource(bucket, { body: 'x', contentSha256: 'ab'.repeat(32) }) + + expect(result).toMatchObject({ _tag: 'failed', reason: 'r2 unavailable' }) + }) + + it('reads R2 first and falls back to the stored D1 body', async () => { + const bucket = memoryBucket() + const body = '# From R2' + const contentSha256 = await skillContentSha256(body) + await putSkillSource(bucket, { body, contentSha256 }) + + await expect(readSkillSource(bucket, { contentSha256, storedBody: '# From D1' })) + .resolves + .toEqual({ _tag: 'ok', body, from: 'r2' }) + await expect(readSkillSource(bucket, { contentSha256: 'cd'.repeat(32), storedBody: '# From D1' })) + .resolves + .toEqual({ _tag: 'ok', body: '# From D1', from: 'd1' }) + await expect(readSkillSource(null, { contentSha256, storedBody: '# From D1' })) + .resolves + .toEqual({ _tag: 'ok', body: '# From D1', from: 'd1' }) + await expect(readSkillSource(bucket, { contentSha256: null, storedBody: null })) + .resolves + .toEqual({ _tag: 'missing' }) + }) +}) diff --git a/layers/registry/server/utils/skill-source-store.ts b/layers/registry/server/utils/skill-source-store.ts new file mode 100644 index 00000000..d8083353 --- /dev/null +++ b/layers/registry/server/utils/skill-source-store.ts @@ -0,0 +1,85 @@ +/// + +/** + * SKILL.md bodies in R2, addressed by content (ADR-0019). + * + * `skills.rendered_raw` held every body in D1, about 9 KB a row and 117 KB at + * the largest, and D1 is a 10 GB store. The body's SHA-256 already sits in + * `skills.rendered_raw_sha256`, so the object key is that hash. Two Skills + * with the same body share one object, and a retried sync writes the same + * bytes to the same key. + */ + +export const SKILL_SOURCE_FORMAT = 'skilld-skill-source-v1' + +/** The subset of `R2Bucket` the store uses, so a test can pass a Map. */ +export interface SkillSourceBucket { + put: R2Bucket['put'] + get: R2Bucket['get'] + head: R2Bucket['head'] +} + +export type SkillSourceWrite + = | { _tag: 'stored', key: string } + | { _tag: 'existing', key: string } + | { _tag: 'failed', key: string, reason: string } + +export type SkillSourceRead + = | { _tag: 'ok', body: string, from: 'r2' | 'd1' } + | { _tag: 'missing' } + +export function skillSourceR2Key(contentSha256: string): string { + return `v1/skill-source/sha256/${contentSha256.slice(0, 2)}/${contentSha256}.md` +} + +/** + * Store one body. The key is its hash, so an object that already exists is + * the same bytes and the put is skipped by `etagDoesNotMatch: '*'`. R2 checks + * the bytes against the hash before it accepts them. + * + * A failure is a value, never a throw: during the dual-write phase D1 still + * holds the body, and a sync must not fail for an object store hiccup. + */ +export async function putSkillSource( + bucket: SkillSourceBucket, + input: { body: string, contentSha256: string }, +): Promise { + const key = skillSourceR2Key(input.contentSha256) + const bytes = new TextEncoder().encode(input.body) + try { + const stored = await bucket.put(key, bytes, { + onlyIf: { etagDoesNotMatch: '*' }, + sha256: input.contentSha256, + httpMetadata: { contentType: 'text/markdown; charset=utf-8' }, + customMetadata: { contentSha256: input.contentSha256, format: SKILL_SOURCE_FORMAT }, + }) + return stored ? { _tag: 'stored', key } : { _tag: 'existing', key } + } + catch (error) { + return { _tag: 'failed', key, reason: error instanceof Error ? error.message : String(error) } + } +} + +/** + * Read one body. R2 by hash first; the D1 column while it still holds bytes. + * Every reader of a SKILL.md body goes through this, so the column can be + * nulled and then dropped without a reader noticing. + */ +export async function readSkillSource( + bucket: SkillSourceBucket | null, + input: { contentSha256: string | null, storedBody: string | null }, +): Promise { + if (bucket && input.contentSha256) { + const object = await bucket.get(skillSourceR2Key(input.contentSha256)) + if (object) + return { _tag: 'ok', body: await object.text(), from: 'r2' } + } + if (input.storedBody) + return { _tag: 'ok', body: input.storedBody, from: 'd1' } + return { _tag: 'missing' } +} + +/** Whether R2 holds the body for this hash. One metadata read, no body. */ +export async function hasSkillSource(bucket: SkillSourceBucket, contentSha256: string): Promise { + return (await bucket.head(skillSourceR2Key(contentSha256))) !== null +} diff --git a/layers/registry/server/utils/sync-repo.ts b/layers/registry/server/utils/sync-repo.ts index 2dd6f87c..670f68da 100644 --- a/layers/registry/server/utils/sync-repo.ts +++ b/layers/registry/server/utils/sync-repo.ts @@ -2,6 +2,7 @@ import type { GithubBindings, RepoMeta } from './github-client' import type { RepositoryMovePlan, RepositoryName } from './repository-move' +import type { SkillSourceBucket } from './skill-source-store' import type { SkillTrustTier } from './skill-trust' import { selectSkillFiles } from '#shared/skill-files' import { canonicalSkillPaths, isRegistrySkillPath, isSkilldCacheSkill } from '#shared/skill-path' @@ -16,6 +17,7 @@ import { parseSkillFile, registrySkillName } from './skill-frontmatter' import { isOfficialSkillRepo, scoreSkillIndexability } from './skill-indexability' import { parseSkillMd } from './skill-md-render' import { findSkillRenames, skillRenameStatements } from './skill-rename' +import { putSkillSource } from './skill-source-store' import { resolveSkillTrust } from './skill-trust' export interface SyncRepoStats { @@ -31,6 +33,10 @@ export interface SyncRepoStats { skillsUnreadable: number revisionsInserted: number activityEmitted: number + /** SKILL.md bodies this sync put to R2, new objects only (ADR-0019). */ + sourcesStored?: number + /** Bodies R2 refused. D1 still holds them; the backfill retries. */ + sourceStoreFailures?: number rateLimitRemaining?: number /** The GitHub bucket `rateLimitRemaining` counts: `core` or `graphql`. */ rateLimitResource?: string @@ -810,6 +816,11 @@ export interface SyncRepoOptions { * synchronous full-repo behavior used outside durable jobs. */ maxSkillFiles?: number + /** + * Where a changed SKILL.md body goes, keyed by its SHA-256 (ADR-0019). + * Omit to write D1 only, as the scripts and older tests do. + */ + sourceBucket?: SkillSourceBucket } export async function syncRepo( @@ -1123,6 +1134,7 @@ export async function syncRepo( const writes: WeightedWrite[] = [] const revisionWriteIndexes: number[] = [] const activityWriteIndexes: number[] = [] + const sourceWrites: Array<{ body: string, contentSha256: string }> = [] const contentFiles: SkillSnapshot[] = [] for (const file of slice) { @@ -1380,6 +1392,7 @@ export async function syncRepo( filePath: '', }) const renderedRawSha256 = await skillContentSha256(raw) + sourceWrites.push({ body: raw, contentSha256: renderedRawSha256 }) let modifiedAt = prev?.modified_at ?? null if (contentChanged) { @@ -1555,6 +1568,20 @@ export async function syncRepo( // so a run that dies mid-repo is replayed safely rather than lost. That same // property lets the slice split again by payload size, which is what keeps a // content-heavy repository under the 32 MiB RPC ceiling. + // The object goes first, so a row that names a hash never points at a + // missing object. A put that fails leaves the row's own copy in place. + if (opts.sourceBucket && sourceWrites.length > 0) { + const results = await Promise.all(sourceWrites.map(write => putSkillSource(opts.sourceBucket!, write))) + for (const result of results) { + if (result._tag === 'stored') + stats.sourcesStored = (stats.sourcesStored ?? 0) + 1 + if (result._tag === 'failed') { + stats.sourceStoreFailures = (stats.sourceStoreFailures ?? 0) + 1 + console.warn(`[sync] skill source put failed for ${owner}/${repo}: ${result.reason}`) + } + } + } + if (writes.length > 0) { const results = await runBoundedBatch(db, writes) for (const index of revisionWriteIndexes) diff --git a/migrations/0159_registry_cursors.sql b/migrations/0159_registry_cursors.sql new file mode 100644 index 00000000..37360838 --- /dev/null +++ b/migrations/0159_registry_cursors.sql @@ -0,0 +1,9 @@ +-- A resumable position for a scheduled task that walks a table in rowid +-- order across many runs. The first user is `backfill-skill-sources`, which +-- copies every stored SKILL.md body to R2 (ADR-0019) a few hundred rows a +-- run, and must not reread the rows it has already copied. +CREATE TABLE IF NOT EXISTS registry_cursors ( + name TEXT PRIMARY KEY, + position INTEGER NOT NULL DEFAULT 0, + updated_at INTEGER NOT NULL +); diff --git a/server/jobs/registry/repo-maintenance.ts b/server/jobs/registry/repo-maintenance.ts index 632e0dd2..156e46ab 100644 --- a/server/jobs/registry/repo-maintenance.ts +++ b/server/jobs/registry/repo-maintenance.ts @@ -355,17 +355,20 @@ export async function handleRegistryRepoJob( forceContent: true, maxSkillFiles: SKILL_PATHS_PER_INVOCATION, ...continuation, + sourceBucket: ctx.env.PUBLIC_ARTIFACTS, } : payload.operation === 'submit' ? { submitted: true, maxSkillFiles: SKILL_PATHS_PER_INVOCATION, ...continuation, + sourceBucket: ctx.env.PUBLIC_ARTIFACTS, } : { ownerVerified, maxSkillFiles: SKILL_PATHS_PER_INVOCATION, ...continuation, + sourceBucket: ctx.env.PUBLIC_ARTIFACTS, }, ) const retryAfter = await applyRateGuard(ctx.db, { diff --git a/shared/schedule-policy.ts b/shared/schedule-policy.ts index 2e7c3ce6..7942769d 100644 --- a/shared/schedule-policy.ts +++ b/shared/schedule-policy.ts @@ -59,6 +59,7 @@ export const SCHEDULE_POLICY = [ { _tag: 'observed', taskName: 'ai-generate-poll', cron: '45 * * * *', maxSilenceSeconds: 3 * 60 * 60, maxRuntimeSeconds: 30 * 60 }, { _tag: 'observed', taskName: 'ai-generate-submit', cron: '15 * * * *', maxSilenceSeconds: 3 * 60 * 60, maxRuntimeSeconds: 50 * 60 }, { _tag: 'observed', taskName: 'ai-ready:cron', cron: '*/5 * * * *', maxSilenceSeconds: 20 * 60, maxRuntimeSeconds: 4 * 60 }, + { _tag: 'observed', taskName: 'backfill-skill-sources', cron: '*/5 * * * *', maxSilenceSeconds: 20 * 60, maxRuntimeSeconds: 4 * 60 }, { _tag: 'observed', taskName: 'backfill-skill-assets', cron: '*/5 * * * *', maxSilenceSeconds: 20 * 60, maxRuntimeSeconds: 4 * 60 }, { _tag: 'observed', taskName: 'cleanup-auto-index-rate-limits', cron: '0 * * * *', maxSilenceSeconds: 3 * 60 * 60, maxRuntimeSeconds: 4 * 60 }, { _tag: 'observed', taskName: 'classify-repository-purpose', cron: '0 * * * *', maxSilenceSeconds: 3 * 60 * 60, maxRuntimeSeconds: 4 * 60 }, diff --git a/test/unit/sync-repo-source-store.test.ts b/test/unit/sync-repo-source-store.test.ts new file mode 100644 index 00000000..a2442951 --- /dev/null +++ b/test/unit/sync-repo-source-store.test.ts @@ -0,0 +1,228 @@ +import type { SkillSourceBucket } from '../../layers/registry/server/utils/skill-source-store' +import Database from 'better-sqlite3' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { skillContentSha256 } from '../../layers/registry/server/utils/skill-content-hash' +import { skillSourceR2Key } from '../../layers/registry/server/utils/skill-source-store' +import { syncRepo } from '../../layers/registry/server/utils/sync-repo' + +/** + * ADR-0019: a changed SKILL.md body goes to R2 under its SHA-256 before the + * D1 row that names that hash is written. During the dual-write phase the + * row keeps its own copy, so an R2 failure must not fail the sync. + */ + +const github = vi.hoisted(() => ({ + getBlobsBatch: vi.fn(), + getCommitsBatch: vi.fn(), + getRepoSummary: vi.fn(), + getTree: vi.fn(), + logRateLimit: vi.fn(), +})) +const renderer = vi.hoisted(() => ({ + parseSkillMd: vi.fn(async (raw: string) => ({ frontmatter: {}, body: raw, html: `

${raw}

` })), +})) + +vi.mock('../../layers/registry/server/utils/github-client', () => github) +vi.mock('../../layers/registry/server/utils/skill-md-render', () => renderer) + +const rawSkill = (name: string) => `---\nname: ${name}\ndescription: ${name} description\n---\n# ${name}` + +describe('syncRepo with a skill source bucket', () => { + let sqlite: Database.Database + let db: D1Database + + beforeEach(() => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-11-12T00:00:00Z')) + vi.clearAllMocks() + sqlite = createDatabase() + db = wrapSqlite(sqlite) + github.getRepoSummary.mockResolvedValue({ + status: 200, + data: { + repositoryId: 1, + headTreeSha: 'new-tree', + meta: { + name: 'skills', + full_name: 'acme/skills', + html_url: 'https://github.com/acme/skills', + owner: { login: 'acme' }, + default_branch: 'main', + description: 'Skills', + stargazers_count: 1, + forks_count: 0, + pushed_at: '2026-11-11T12:00:00Z', + created_at: '2025-01-01T00:00:00Z', + }, + }, + rateLimit: null, + notModified: false, + }) + github.getTree.mockResolvedValue({ + status: 200, + data: { sha: 'new-tree', tree: [{ path: 'skills/one/SKILL.md', sha: 'one-new', type: 'blob' }] }, + rateLimit: null, + notModified: false, + }) + github.getBlobsBatch.mockResolvedValue({ + status: 200, + data: new Map([['skills/one/SKILL.md', rawSkill('One')]]), + unreadable: new Set(), + rateLimit: null, + notModified: false, + }) + github.getCommitsBatch.mockImplementation(async (_owner: string, _repo: string, paths: string[]) => ({ + status: 200, + data: new Map(paths.map(path => [path, []])), + rateLimit: null, + notModified: false, + })) + sqlite.prepare(`INSERT INTO repos (owner, repo, default_branch, pushed_at, last_tree_sha) VALUES ('acme', 'skills', 'main', 0, 'old-tree')`).run() + }) + + afterEach(() => { + vi.useRealTimers() + sqlite.close() + }) + + it('puts the body to R2 under the hash the row records', async () => { + const bucket = memoryBucket() + + const result = await syncRepo('acme', 'skills', {}, db, { ownerVerified: true, sourceBucket: bucket }) + + expect(result).toMatchObject({ status: 'indexed', sourcesStored: 1 }) + const row = sqlite.prepare(`SELECT rendered_raw, rendered_raw_sha256 FROM skills`).get() as { rendered_raw: string, rendered_raw_sha256: string } + expect(row.rendered_raw_sha256).toBe(await skillContentSha256(rawSkill('One'))) + const object = bucket.objects.get(skillSourceR2Key(row.rendered_raw_sha256)) + expect(object && new TextDecoder().decode(object)).toBe(row.rendered_raw) + }) + + it('keeps the D1 row when the put fails and counts the failure', async () => { + const bucket = memoryBucket(true) + + const result = await syncRepo('acme', 'skills', {}, db, { ownerVerified: true, sourceBucket: bucket }) + + expect(result).toMatchObject({ status: 'indexed', sourceStoreFailures: 1 }) + expect(result.sourcesStored).toBeUndefined() + expect(sqlite.prepare(`SELECT rendered_raw FROM skills`).pluck().get()).toBe(rawSkill('One')) + }) + + it('writes D1 only when no bucket is given', async () => { + const result = await syncRepo('acme', 'skills', {}, db, { ownerVerified: true }) + + expect(result).toMatchObject({ status: 'indexed' }) + expect(result.sourcesStored).toBeUndefined() + expect(sqlite.prepare(`SELECT rendered_raw FROM skills`).pluck().get()).toBe(rawSkill('One')) + }) +}) + +function memoryBucket(failPut = false): SkillSourceBucket & { objects: Map } { + const objects = new Map() + return { + objects, + put: (async (key: string, value: Uint8Array) => { + if (failPut) + throw new Error('r2 unavailable') + objects.set(key, value) + return { key } + }) as unknown as SkillSourceBucket['put'], + get: (async () => null) as unknown as SkillSourceBucket['get'], + head: (async () => null) as unknown as SkillSourceBucket['head'], + } +} + +function createDatabase(): Database.Database { + const sqlite = new Database(':memory:') + sqlite.exec(` + CREATE TABLE repos ( + owner TEXT NOT NULL, repo TEXT NOT NULL, default_branch TEXT, stars INTEGER NOT NULL DEFAULT 0, + forks INTEGER NOT NULL DEFAULT 0, pushed_at INTEGER, repo_created_at INTEGER, + repo_meta_synced_at INTEGER, description TEXT, last_tree_sha TEXT, repo_kind TEXT NOT NULL DEFAULT 'creator', + repo_kind_source TEXT NOT NULL DEFAULT 'computed', repo_skill_count INTEGER NOT NULL DEFAULT 0, + broken_since INTEGER, tree_truncated_at INTEGER, source_owner TEXT, source_repo TEXT, repository_id INTEGER, is_fork INTEGER, PRIMARY KEY (owner, repo) + ); + CREATE TABLE skills ( + name TEXT NOT NULL, owner TEXT NOT NULL, repo TEXT NOT NULL, display_name TEXT NOT NULL, + installs INTEGER NOT NULL DEFAULT 0, slug TEXT NOT NULL, description TEXT, current_sha TEXT, + modified_at INTEGER, first_seen_at INTEGER, references_count INTEGER NOT NULL DEFAULT 0, + assets TEXT NOT NULL DEFAULT '[]', last_synced_at INTEGER, sync_status TEXT, + is_official INTEGER NOT NULL DEFAULT 0, source_resolved INTEGER NOT NULL DEFAULT 0, + seo_index_score INTEGER NOT NULL DEFAULT 0, seo_indexable INTEGER NOT NULL DEFAULT 0, + seo_index_reasons TEXT NOT NULL DEFAULT '[]', seo_index_synced_at INTEGER, + trust_tier TEXT NOT NULL DEFAULT 'untrusted', trust_source TEXT NOT NULL DEFAULT 'computed', + trust_score INTEGER NOT NULL DEFAULT 0, trust_reasons TEXT NOT NULL DEFAULT '[]', trust_synced_at INTEGER, + rendered_skill_path TEXT, rendered_commit_sha TEXT, rendered_status TEXT, rendered_raw TEXT, rendered_raw_sha256 TEXT, rendered_frontmatter TEXT, + rendered_html TEXT, rendered_at INTEGER, owner_verified INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (owner, repo, name) + ); + CREATE TABLE skill_revisions ( + owner TEXT NOT NULL, repo TEXT NOT NULL, name TEXT NOT NULL, sha TEXT NOT NULL, + modified_at INTEGER NOT NULL, author_login TEXT, message TEXT, + PRIMARY KEY (owner, repo, name, sha) + ); + CREATE TABLE activity ( + type TEXT NOT NULL, owner TEXT NOT NULL, repo TEXT, name TEXT NOT NULL, + occurred_at INTEGER NOT NULL, ingested_at INTEGER NOT NULL, sha TEXT + ); + CREATE TABLE skill_dirty ( + owner TEXT NOT NULL, repo TEXT NOT NULL, name TEXT NOT NULL, reason TEXT NOT NULL, + queued_at INTEGER NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (owner, repo, name, reason) + ); + CREATE TABLE repo_trust_overrides ( + owner TEXT NOT NULL, repo TEXT NOT NULL, tier TEXT NOT NULL, reason TEXT, + PRIMARY KEY (owner, repo) + ); + CREATE TABLE repo_kind_overrides ( + owner TEXT NOT NULL, repo TEXT NOT NULL, kind TEXT NOT NULL, + PRIMARY KEY (owner, repo) + ); + CREATE TABLE skill_repo_eligibility ( + owner TEXT NOT NULL, repo TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('eligible', 'rejected')), + reason TEXT NOT NULL, reviewed_by TEXT NOT NULL, + reviewed_at INTEGER NOT NULL DEFAULT (unixepoch()), + PRIMARY KEY (owner, repo) + ); + CREATE TABLE repo_star_observations ( + owner TEXT NOT NULL, repo TEXT NOT NULL, observed_day INTEGER NOT NULL, + stars INTEGER NOT NULL, PRIMARY KEY (owner, repo, observed_day) + ); + `) + return sqlite +} + +interface BoundStatement { + executeSync: () => D1Result + run: () => Promise> + all: () => Promise> + first: () => Promise +} + +function wrapSqlite(sqlite: Database.Database): D1Database { + const prepare = (sql: string) => ({ + bind(...params: unknown[]): BoundStatement { + const executeSync = () => { + const result = sqlite.prepare(sql).run(...params) + return { meta: { changes: result.changes } } as D1Result + } + return { + executeSync, + run: async () => executeSync(), + async all() { + return { results: sqlite.prepare(sql).all(...params) as T[] } as D1Result + }, + async first() { + return (sqlite.prepare(sql).get(...params) as T | undefined) ?? null + }, + } + }, + }) + return { + prepare, + async batch(statements: D1PreparedStatement[]) { + const run = sqlite.transaction(() => statements.map(statement => (statement as unknown as BoundStatement).executeSync())) + return run() as D1Result[] + }, + } as unknown as D1Database +}