From e2b4b3005570b3b1573657afc2a8a222d0c13282 Mon Sep 17 00:00:00 2001 From: Harlan GitHub Agent Date: Thu, 12 Nov 2026 12:30:02 +1100 Subject: [PATCH] fix(artifact): stop run checks filling D1 and prune old Resolutions The skilld run sweep re-checked every runnable Skill about hourly. Each check wrote a Resolution with about 60 KB of attestations, and nothing deleted them. The 10 GB database filled on 2026-11-04 and every write since has failed. Check a ready Skill weekly and a failing one daily. Before each sweep, delete one batch of settled public Resolutions older than three days that no run check, grant, or newest Artifact attestation needs. The prune runs before the observed run records its start, so it frees space while the database is full. Co-Authored-By: Claude Opus 5.5 --- .../server/tasks/sweep-skill-runs.ts | 17 +++- .../server/utils/resolution-retention.ts | 74 +++++++++++++++ .../server/utils/run-check-state.ts | 33 +++++++ .../server/utils/run-sweep.ts | 14 ++- test/unit/resolution-retention.test.ts | 93 +++++++++++++++++++ test/unit/skill-run-sweep.test.ts | 25 ++++- 6 files changed, 247 insertions(+), 9 deletions(-) create mode 100644 layers/artifact-delivery/server/utils/resolution-retention.ts create mode 100644 test/unit/resolution-retention.test.ts 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(), } }