diff --git a/layers/artifact-delivery/server/tasks/sweep-skill-runs.ts b/layers/artifact-delivery/server/tasks/sweep-skill-runs.ts index 7fe8c2dd..7694be06 100644 --- a/layers/artifact-delivery/server/tasks/sweep-skill-runs.ts +++ b/layers/artifact-delivery/server/tasks/sweep-skill-runs.ts @@ -7,15 +7,17 @@ import { reportJobRun } from '~~/server/utils/sync-job-reporter' import { observedSchedulePolicy } from '#shared/schedule-policy' import { enqueueArtifactBuild } from '../utils/queue' import { fetchAdmittedSkillIdentity, requestResolution } from '../utils/request-resolution' +import { pruneSettledResolutions } from '../utils/resolution-retention' import { runSkillRunSweep } from '../utils/run-sweep' const CRON = '7,22,37,52 * * * *' /** - * 40 checks every 15 minutes is 3,840 a day. The registry indexed 5,949 Skills - * on 2026-10-07, so one cycle takes about 37 hours. A ready build of the same - * commit is reused without a GitHub read, so a repeat check of an unchanged - * Skill costs D1 rows only. A first check loads from GitHub with about 8 REST - * requests, so a cycle of first checks spends about 1,300 an hour. + * At most 40 checks every 15 minutes. A ready Skill is checked weekly and a + * failing one daily (`RECHECK_AFTER_SECONDS`), so most runs start few checks. + * Each check still writes a Resolution with about 60 KB of attestations, even + * when it reuses a ready build. Checking every Skill hourly filled the 10 GB + * database on 2026-11-04. A first check loads from GitHub with about 8 REST + * requests. */ const BATCH_SIZE = 40 @@ -42,6 +44,11 @@ export default defineScheduledTask({ return { result: { error: 'no-db' } } } + // Prune before the observed run records its start: a full database refuses + // that insert, and only a delete can free the space it needs. + const pruned = await pruneSettledResolutions(db, { now: Math.floor(Date.now() / 1000) }) + emitOperationalEvent(createWideEvent({ 'operation': 'prune-resolutions', 'outcome': 'completed', 'processed.count': pruned }), 'info') + return await runObservedScheduledTask({ db, env, diff --git a/layers/artifact-delivery/server/utils/resolution-retention.ts b/layers/artifact-delivery/server/utils/resolution-retention.ts new file mode 100644 index 00000000..64d87150 --- /dev/null +++ b/layers/artifact-delivery/server/utils/resolution-retention.ts @@ -0,0 +1,74 @@ +/// + +/** + * How long a settled public Resolution is kept. + * + * The CLI reads a Resolution only while its build runs, and gives up after 60 + * seconds. A run check settles on the next sweep run, 15 minutes later. Three + * days leaves the daily check-in its own window and a day to spare. + */ +export const RESOLUTION_RETENTION_SECONDS = 3 * 24 * 60 * 60 + +/** + * Resolutions one prune may delete. Each carries about 60 KB of attestations, + * so one batch frees about 30 MB and keeps each D1 statement small. + */ +export const RESOLUTION_PRUNE_BATCH = 500 + +/** + * Settled public Resolutions nothing needs any more. + * + * A Resolution stays while: + * - it is private, because private Artifacts and their grants hang off it; + * - it is newer than the retention window; + * - a run check still reads it; + * - a download grant names it; + * - it is ready and holds the newest attestation of its Artifact, which build + * reuse copies and the Artifact routes serve. + * + * Deleting one cascades to its attestation, check results, and requester rows. + * The first filters repeat `idx_artifact_resolutions_state`, so the scan reads + * only settled rows older than the window. + */ +const PRUNE_CANDIDATES = ` + SELECT r.id + FROM artifact_resolutions r INDEXED BY idx_artifact_resolutions_state + WHERE r.state IN ('ready', 'blocked', 'failed', 'revoked') + AND r.updated_at < ?1 + AND r.visibility = 'public' + AND r.id NOT IN ( + SELECT pending_resolution_id FROM artifact_run_checks WHERE pending_resolution_id IS NOT NULL + UNION + SELECT settled_resolution_id FROM artifact_run_checks WHERE settled_resolution_id IS NOT NULL + UNION + SELECT resolution_id FROM artifact_download_grants + ) + AND ( + r.state <> 'ready' + OR EXISTS ( + SELECT 1 + FROM artifact_attestations own + JOIN artifact_attestations newer + ON newer.artifact_id = own.artifact_id + AND (newer.created_at, newer.resolution_id) > (own.created_at, own.resolution_id) + WHERE own.resolution_id = r.id + ) + ) + LIMIT ?2 +` + +/** + * Delete one batch of settled public Resolutions nothing needs. + * + * Returns how many it deleted. A full batch means more remain. + */ +export async function pruneSettledResolutions( + db: D1Database, + input: { now: number, retentionSeconds?: number, limit?: number }, +): Promise { + const cutoff = input.now - (input.retentionSeconds ?? RESOLUTION_RETENTION_SECONDS) + const result = await db.prepare( + `DELETE FROM artifact_resolutions WHERE id IN (${PRUNE_CANDIDATES})`, + ).bind(cutoff, input.limit ?? RESOLUTION_PRUNE_BATCH).run() + return result.meta.changes ?? 0 +} diff --git a/layers/artifact-delivery/server/utils/run-check-state.ts b/layers/artifact-delivery/server/utils/run-check-state.ts index d77cb714..2de9457c 100644 --- a/layers/artifact-delivery/server/utils/run-check-state.ts +++ b/layers/artifact-delivery/server/utils/run-check-state.ts @@ -113,3 +113,36 @@ export function runCheckColumns(state: RunCheckState): RunCheckColumns { return { outcome: 'failing', tag: state.tag, detail: state.detail, retryable: 0, failing_since: state.since, failure_streak: state.streak, failed_at: state.failedAt } } } + +/** How long a settled check stands before the sweep checks that Skill again. */ +export interface RecheckAfterSeconds { + ready: number + failing: number +} + +/** + * A ready Skill is checked weekly, a failing one daily. Each check writes a + * Resolution with its attestations, about 60 KB, so checking every Skill every + * hour filled the 10 GB database in November 2026. + */ +export const RECHECK_AFTER_SECONDS: RecheckAfterSeconds = { + ready: 7 * 24 * 60 * 60, + failing: 24 * 60 * 60, +} + +/** + * Whether the sweep should check a Skill again now. + * + * An unchecked Skill and a transient failure, such as a spent quota, are due + * at once: neither says anything about the Skill yet. + */ +export function recheckDue( + state: RunCheckState, + settledAt: number | null, + now: number, + after: RecheckAfterSeconds, +): boolean { + if (state._tag === 'unchecked' || state._tag === 'transient' || settledAt === null) + return true + return now - settledAt >= (state._tag === 'ready' ? after.ready : after.failing) +} diff --git a/layers/artifact-delivery/server/utils/run-sweep.ts b/layers/artifact-delivery/server/utils/run-sweep.ts index 096cb87d..90371105 100644 --- a/layers/artifact-delivery/server/utils/run-sweep.ts +++ b/layers/artifact-delivery/server/utils/run-sweep.ts @@ -1,9 +1,9 @@ import type { SourceRequest } from '../schemas/contracts' import type { ResolutionAccess } from './request-resolution' -import type { RunCheckColumns } from './run-check-state' +import type { RecheckAfterSeconds, RunCheckColumns } from './run-check-state' import type { CreateResolutionResult, ResolutionRow } from './state' import { z } from 'zod' -import { nextRunCheckState, runCheckColumns, runCheckStateOf } from './run-check-state' +import { nextRunCheckState, RECHECK_AFTER_SECONDS, recheckDue, runCheckColumns, runCheckStateOf } from './run-check-state' import { ACTIVE_BUILD_STATES, getResolution, parseCheckResults } from './state' /** One Skill whose page shows `skilld run OWNER/REPOSITORY/NAME`. */ @@ -33,6 +33,8 @@ export interface SkillRunSweepDependencies { now: () => number /** How many new checks one run starts. */ batchSize: number + /** How long a settled check stands. Defaults to `RECHECK_AFTER_SECONDS`. */ + recheckAfter?: RecheckAfterSeconds } export interface SkillRunSweepReport { @@ -132,8 +134,14 @@ export async function runSkillRunSweep(dependencies: SkillRunSweepDependencies): return row !== undefined && runCheckStateOf(row)._tag === 'failing' }).length + const recheckAfter = dependencies.recheckAfter ?? RECHECK_AFTER_SECONDS const due = runnable - .filter(skill => !rows.get(skillKey(skill))?.pending_resolution_id) + .filter((skill) => { + const row = rows.get(skillKey(skill)) + if (!row) + return true + return !row.pending_resolution_id && recheckDue(runCheckStateOf(row), row.settled_at, now, recheckAfter) + }) .sort((left, right) => (rows.get(skillKey(left))?.settled_at ?? 0) - (rows.get(skillKey(right))?.settled_at ?? 0) || compareText(skillKey(left), skillKey(right))) diff --git a/test/unit/resolution-retention.test.ts b/test/unit/resolution-retention.test.ts new file mode 100644 index 00000000..381db9c6 --- /dev/null +++ b/test/unit/resolution-retention.test.ts @@ -0,0 +1,93 @@ +// @vitest-environment node +import type { SqliteD1 } from './helpers/d1-sqlite' +import { afterEach, describe, expect, it } from 'vitest' +import { pruneSettledResolutions, RESOLUTION_RETENTION_SECONDS } from '../../layers/artifact-delivery/server/utils/resolution-retention' +import { allMigrations, createSqliteD1 } from './helpers/d1-sqlite' + +const NOW = 1_795_000_000 +const OLD = NOW - RESOLUTION_RETENTION_SECONDS - 60 +const ARTIFACT = `sha256:${'a'.repeat(64)}` + +let sqlite: SqliteD1 + +afterEach(() => sqlite?.close()) + +interface Seed { + id: string + at: number + state?: 'ready' | 'blocked' | 'failed' | 'fetching' + visibility?: 'public' | 'private' +} + +function seed(rows: Seed[]) { + sqlite = createSqliteD1(allMigrations()) + sqlite.raw.prepare(`INSERT INTO artifacts (id, content_sha256, content_bytes, format, r2_key) + VALUES (?, ?, 1024, 'skilld-tar-v1', 'v1/public/a.bin')`).run(ARTIFACT, 'a'.repeat(64)) + for (const row of rows) { + const state = row.state ?? 'ready' + sqlite.raw.prepare(`INSERT INTO artifact_resolutions ( + id, request_fingerprint, state, visibility, requested_owner, requested_repository, + selector_type, selector_value, artifact_id, attestation_json, created_at, updated_at) + VALUES (?, ?, ?, ?, 'acme', 'skills', 'named-skill', 'demo', ?, '{}', ?, ?)`).run( + row.id, + `fp-${row.id}`, + state, + row.visibility ?? 'public', + state === 'ready' ? ARTIFACT : null, + row.at, + row.at, + ) + if (state === 'ready') { + sqlite.raw.prepare(`INSERT INTO artifact_attestations (resolution_id, artifact_id, attestation_json, created_at) + VALUES (?, ?, '{}', ?)`).run(row.id, ARTIFACT, row.at) + } + } +} + +function remaining(): string[] { + return (sqlite.raw.prepare('SELECT id FROM artifact_resolutions ORDER BY id').all() as Array<{ id: string }>).map(row => row.id) +} + +describe('pruneSettledResolutions', () => { + it('keeps the newest ready build of an Artifact and drops its older copies', async () => { + seed([ + { id: 'r1', at: OLD - 20 }, + { id: 'r2', at: OLD - 10 }, + { id: 'r3', at: OLD }, + { id: 'blocked', at: OLD, state: 'blocked' }, + ]) + + expect(await pruneSettledResolutions(sqlite.db, { now: NOW })).toBe(3) + + expect(remaining()).toEqual(['r3']) + expect(sqlite.raw.prepare('SELECT resolution_id FROM artifact_attestations').all()).toEqual([{ resolution_id: 'r3' }]) + }) + + it('keeps recent, building, private, and run-checked Resolutions', async () => { + seed([ + { id: 'checked', at: OLD - 30 }, + { id: 'private', at: OLD - 20, state: 'failed', visibility: 'private' }, + { id: 'building', at: OLD - 10, state: 'fetching' }, + { id: 'recent-failure', at: NOW - 60, state: 'failed' }, + { id: 'newest', at: NOW - 60 }, + ]) + sqlite.raw.prepare(`INSERT INTO artifact_run_checks (owner, repository, name, settled_resolution_id) + VALUES ('acme', 'skills', 'demo', 'checked')`).run() + + expect(await pruneSettledResolutions(sqlite.db, { now: NOW })).toBe(0) + + expect(remaining()).toEqual(['building', 'checked', 'newest', 'private', 'recent-failure']) + }) + + it('deletes at most one batch per call', async () => { + seed([ + { id: 'f1', at: OLD, state: 'failed' }, + { id: 'f2', at: OLD, state: 'failed' }, + { id: 'f3', at: OLD, state: 'failed' }, + ]) + + expect(await pruneSettledResolutions(sqlite.db, { now: NOW, limit: 2 })).toBe(2) + expect(await pruneSettledResolutions(sqlite.db, { now: NOW, limit: 2 })).toBe(1) + expect(remaining()).toEqual([]) + }) +}) diff --git a/test/unit/skill-run-sweep.test.ts b/test/unit/skill-run-sweep.test.ts index 4655bca8..ef250e3a 100644 --- a/test/unit/skill-run-sweep.test.ts +++ b/test/unit/skill-run-sweep.test.ts @@ -6,6 +6,7 @@ import { describe, expect, it, vi } from 'vitest' import { processArtifactBuild } from '../../layers/artifact-delivery/server/utils/build' import { requestResolution } from '../../layers/artifact-delivery/server/utils/request-resolution' import { loadRunCheckFlags } from '../../layers/artifact-delivery/server/utils/run-check-flags' +import { RECHECK_AFTER_SECONDS } from '../../layers/artifact-delivery/server/utils/run-check-state' import { runSkillRunSweep } from '../../layers/artifact-delivery/server/utils/run-sweep' import { createSqliteD1 } from './helpers/d1-sqlite' @@ -115,9 +116,29 @@ describe('skilld run sweep', () => { expect(harness.requested()).toEqual(['acme/busy/busy', 'tt-a1i/archify/archify']) harness.close() }) + + it('checks a failing Skill again only after a day, and a spent quota at once', async () => { + const harness = createHarness([archify, busy], 10, RECHECK_AFTER_SECONDS) + + await runSkillRunSweep(harness.dependencies) + await harness.buildEach({ + archify: { _tag: 'rejected', code: 'INVALID_SOURCE', summary: 'Too large.', findings: [] }, + busy: { _tag: 'rejected', code: 'RATE_LIMITED', summary: 'Spent.', findings: [] }, + }) + const settled = await runSkillRunSweep(harness.dependencies) + + expect(settled.settled).toBe(2) + expect(settled.started).toBe(1) + expect(harness.requested()).toEqual(['acme/busy/busy', 'tt-a1i/archify/archify', 'acme/busy/busy']) + harness.advance(RECHECK_AFTER_SECONDS.failing) + // Busy's pending check timed out over that day, so it is due again too. + expect((await runSkillRunSweep(harness.dependencies)).started).toBe(2) + expect(harness.requested().slice(-2)).toContain('tt-a1i/archify/archify') + harness.close() + }) }) -function createHarness(skills: RunnableSkill[], batchSize = 10) { +function createHarness(skills: RunnableSkill[], batchSize = 10, recheckAfter = { ready: 0, failing: 0 }) { const sqlite = createSqliteD1(MIGRATIONS) const enqueue = vi.fn(async (_resolutionId: string) => {}) const requestedRefs: string[] = [] @@ -138,6 +159,7 @@ function createHarness(skills: RunnableSkill[], batchSize = 10) { newIdempotencyKey: () => `sweep-test-key-${String(++keys).padStart(4, '0')}`, now: () => clock, batchSize, + recheckAfter, } /** Run the queued builds, each rejected by GitHub as the test names. */ @@ -183,6 +205,7 @@ function createHarness(skills: RunnableSkill[], batchSize = 10) { enqueue, buildEach, requested: () => requestedRefs, + advance: (seconds: number) => { clock += seconds }, close: () => sqlite.close(), } }