diff --git a/layers/artifact-delivery/server/utils/build.ts b/layers/artifact-delivery/server/utils/build.ts index 92ff37d0..7cfb71d5 100644 --- a/layers/artifact-delivery/server/utils/build.ts +++ b/layers/artifact-delivery/server/utils/build.ts @@ -9,7 +9,7 @@ import type { import type { ArtifactSigner } from './attestation' import type { CheckedArtifactSource } from './checks' import type { ArtifactSourceFile, PublicGithubSourceClient, SourceRejection } from './github-source' -import type { ReadyBuildLookup, ResolutionPatch, ResolutionRow } from './state' +import type { ReadyBuildLookup, ReadyPublicBuild, ResolutionPatch, ResolutionRow } from './state' import type { TrustedRoot } from './trusted-root' import { createWideEvent } from '@harlan-zw/nuxt-wide-events/standalone' import { emitOperationalEvent } from '#server/utils/operational-event' @@ -28,16 +28,19 @@ import { } from './attestation' import { checkArtifactSource, checksBlockArtifact, checksPermitSigning } from './checks' import { canonicalJson, digestHex } from './encoding' -import { isRetryableProblem } from './github-source' +import { isRetryableProblem, storedFilesPassLoadRules } from './github-source' import { ARTIFACT_POLICY_VERSION, + BYTE_COMPATIBLE_POLICY_VERSIONS, findReadyPublicBuild, getResolution, + hasLeadingBuild, + LEADING_BUILD_FRESH_SECONDS, parseCheckResults, publishArtifactRecord, transitionResolution, } from './state' -import { createDeterministicUstar } from './ustar' +import { createDeterministicUstar, readDeterministicUstar } from './ustar' export interface ArtifactBuildDependencies { db: D1Database @@ -81,10 +84,17 @@ export type ReuseMissReason export type ArtifactBuildReuseReport = { _tag: 'hit', lookup: ReadyBuildLookup['_tag'], resolutionId: string, reusedFrom: string } | { _tag: 'miss', lookup: ReadyBuildLookup['_tag'], resolutionId: string, reason: ReuseMissReason } + /** The stored bytes of a ready build were checked again, with no GitHub read. */ + | { _tag: 'recheck', lookup: ReadyBuildLookup['_tag'], resolutionId: string, reusedFrom: string } export type ArtifactBuildOutcome = { _tag: 'ready' | 'blocked' | 'failed' | 'unchanged', resolutionId: string } | { _tag: 'superseded', resolutionId: string } + /** An earlier build of the same Skill at the same commit is running. Process this one again later. */ + | { _tag: 'deferred', resolutionId: string, delaySeconds: number } + +/** How long a build waits before it looks again at the earlier build it follows. */ +export const LEADING_BUILD_WAIT_SECONDS = 2 /** * A ready public build of the same commit, verified against the current trusted @@ -122,6 +132,9 @@ type BuildLoad | ReusedBuild | { _tag: 'rejected', rejection: SourceRejection } +/** An earlier build of the same source is running, so this one waits for it. */ +interface FollowingBuild { _tag: 'following' } + type StagedStatement = { _tag: 'staged', patch: ResolutionPatch } | { _tag: 'content-changed' } @@ -160,9 +173,12 @@ export async function processArtifactBuild( // A request pinned to a commit that was built before needs no GitHub // read. An unpinned request always resolves on GitHub, so it never gets // an older commit than the branch or tag names now. - const pinned = await reusePinnedBuild(dependencies, row) + const pinned = await startPinnedBuild(dependencies, row) + if (pinned?._tag === 'following') + return { _tag: 'deferred', resolutionId, delaySeconds: LEADING_BUILD_WAIT_SECONDS } if (pinned) { - const advanced = await transitionResolution(dependencies.db, row, 'fetching', resolvedSourcePatch(pinned.build.source), now) + const pinnedSource = pinned._tag === 'reused' ? pinned.build.source : pinned.source + const advanced = await transitionResolution(dependencies.db, row, 'fetching', resolvedSourcePatch(pinnedSource), now) if (advanced._tag === 'superseded') return { _tag: 'superseded', resolutionId } row = advanced.row @@ -186,7 +202,10 @@ export async function processArtifactBuild( } if (row.state === 'fetching') { - const loaded: BuildLoad = carried ?? await reuseOrLoad(dependencies, row) + const next: BuildLoad | FollowingBuild = carried ?? await reuseOrLoad(dependencies, row) + if (next._tag === 'following') + return { _tag: 'deferred', resolutionId, delaySeconds: LEADING_BUILD_WAIT_SECONDS } + const loaded: BuildLoad = next carried = loaded._tag === 'rejected' ? null : loaded const checkResults = loaded._tag === 'loaded' ? loaded.checked.checkResults @@ -380,40 +399,140 @@ function stageReusedStatement( } } -/** A pinned public request reuses a ready build of its commit before any GitHub read. */ -async function reusePinnedBuild( +/** + * A pinned public request, before any GitHub read: reuse a ready build of its + * commit, check stored bytes again, or wait for an earlier build of it. + * Null means the request resolves on GitHub. + */ +async function startPinnedBuild( dependencies: ArtifactBuildDependencies, row: ResolutionRow, -): Promise { +): Promise { if (row.visibility !== 'public') return null const request = sourceRequestFromRow(row) if (request.ref?.type !== 'commit') return null - const decision = await decideReuse(dependencies, row, { + const lookup: ReadyBuildLookup = { _tag: 'pinned', owner: request.owner, repository: request.repository, commitSha: request.ref.value, selector: request.selector, - }) - return decision._tag === 'hit' ? { _tag: 'reused', build: decision.build } : null + } + const decision = await decideReuse(dependencies, row, lookup) + if (decision._tag === 'hit') + return { _tag: 'reused', build: decision.build } + return (bytesMayBeReused(decision.reason) ? await loadStoredBuild(dependencies, row, lookup) : null) + ?? (await followsLeadingBuild(dependencies, row, lookup) ? { _tag: 'following' } : null) } -/** After GitHub resolves a public request, reuse a ready build of that exact source, or load it. */ +/** + * After GitHub resolves a public request: reuse a ready build of that exact + * source, check its stored bytes again, wait for an earlier build of it, or + * load it. + */ async function reuseOrLoad( dependencies: ArtifactBuildDependencies, row: ResolutionRow, -): Promise { +): Promise { const source = resolvedSourceFromRow(row) if (row.visibility === 'public') { - const decision = await decideReuse(dependencies, row, { _tag: 'resolved', source }) + const lookup: ReadyBuildLookup = { _tag: 'resolved', source } + const decision = await decideReuse(dependencies, row, lookup) if (decision._tag === 'hit') return { _tag: 'reused', build: decision.build } + const stored = bytesMayBeReused(decision.reason) ? await loadStoredBuild(dependencies, row, lookup) : null + if (stored) + return stored + if (await followsLeadingBuild(dependencies, row, lookup)) + return { _tag: 'following' } } return await loadAndCheck(dependencies, row, source) } +/** + * Whether an earlier build of the same source is running, so this one should + * wait for it rather than spend GitHub quota on the same reads. A build waits + * at most {@link LEADING_BUILD_FRESH_SECONDS} from its creation. + */ +async function followsLeadingBuild( + dependencies: ArtifactBuildDependencies, + row: ResolutionRow, + lookup: ReadyBuildLookup, +): Promise { + const now = dependencies.now() + if (row.created_at < now - LEADING_BUILD_FRESH_SECONDS) + return false + return await hasLeadingBuild(dependencies.db, row, lookup, now) +} + +/** A ready build missed only because its policy or checks changed, so its bytes may still serve. */ +function bytesMayBeReused(reason: ReuseMissReason): boolean { + return reason === 'policy-changed' || reason === 'checks-changed' +} + +/** + * The files of a ready public build, read back from R2 and checked under the + * current checks, when its policy packed the same bytes as the current one. + * + * A policy bump that changes only checks then costs no GitHub read. The + * attestation must still verify, R2 must hold the exact bytes, and the files + * must pass every rule a GitHub load applies today. Null means load from + * GitHub. + */ +async function loadStoredBuild( + dependencies: ArtifactBuildDependencies, + row: ResolutionRow, + lookup: ReadyBuildLookup, +): Promise { + const ready = await findReadyPublicBuild(dependencies.db, lookup) + if (!ready) + return null + const attestation = parseStoredAttestation(ready.attestationJson) + if ( + !attestation + || (attestation.policyVersion !== ARTIFACT_POLICY_VERSION && !BYTE_COMPATIBLE_POLICY_VERSIONS.has(attestation.policyVersion)) + || !omittedNothing(attestation) + || !attestationMatchesRecord(attestation, ready) + || !await verifyArtifactAttestation(attestation, dependencies.trustedRoot, dependencies.now()) + ) { + return null + } + const object = await dependencies.bucket.get(ready.r2Key) + if (!object) + return null + const bytes = new Uint8Array(await object.arrayBuffer()) + if (bytes.byteLength !== ready.contentBytes || await digestHex('SHA-256', bytes) !== ready.contentSha256) + return null + const files = await readDeterministicUstar(bytes) + if (!files || !storedFilesPassLoadRules(files, ready.source.skillPath)) + return null + const reportReuse = dependencies.reportReuse ?? emitReuseEvent + reportReuse({ + _tag: 'recheck', + lookup: lookup._tag, + resolutionId: row.id, + reusedFrom: ready.resolutionId, + }) + return { + _tag: 'loaded', + source: ready.source, + files, + checked: await checkArtifactSource(ready.source, files, []), + } +} + +/** + * Whether a stored build left no file out. R2 holds only the files it packed, + * so a build that left some out cannot be checked again from its bytes. + * Policies before the `omitted-files` check never left a file out. + */ +function omittedNothing(attestation: ArtifactAttestation): boolean { + const omitted = attestation.checkResults.find(check => check.name === 'omitted-files') + return omitted === undefined || omitted.outcome === 'pass' +} + /** * The source for a build that resumes after `fetching` in a new invocation. * A ready build is reused only when it matches the check results and content @@ -480,16 +599,8 @@ async function findReusableBuild( return { _tag: 'miss', reason: 'policy-changed' } if (!checksPermitSigning(attestation.checkResults)) return { _tag: 'miss', reason: 'checks-changed' } - if ( - attestation.artifactId !== ready.artifactId - || ready.artifactId !== `sha256:${ready.contentSha256}` - || attestation.contentSha256 !== ready.contentSha256 - || attestation.contentBytes !== ready.contentBytes - || ready.r2Key !== artifactR2Key(ready.contentSha256) - || canonicalJson(attestation.source) !== canonicalJson(ready.source) - ) { + if (!attestationMatchesRecord(attestation, ready)) return { _tag: 'miss', reason: 'record-mismatch' } - } const stored = await hasImmutableArtifact(dependencies.bucket, { key: ready.r2Key, contentSha256: ready.contentSha256, @@ -511,6 +622,16 @@ async function findReusableBuild( } } +/** Whether a stored attestation names exactly the bytes and source of its D1 record. */ +function attestationMatchesRecord(attestation: ArtifactAttestation, ready: ReadyPublicBuild): boolean { + return attestation.artifactId === ready.artifactId + && ready.artifactId === `sha256:${ready.contentSha256}` + && attestation.contentSha256 === ready.contentSha256 + && attestation.contentBytes === ready.contentBytes + && ready.r2Key === artifactR2Key(ready.contentSha256) + && canonicalJson(attestation.source) === canonicalJson(ready.source) +} + function parseStoredAttestation(value: string): ArtifactAttestation | null { const parsed = artifactAttestationSchema.safeParse(parseJson(value)) return parsed.success ? parsed.data : null @@ -528,10 +649,10 @@ function parseJson(value: string): unknown { /** One wide event per reuse decision, so the daily check-in can count hits. */ function emitReuseEvent(report: ArtifactBuildReuseReport): void { - if (report._tag === 'hit') { + if (report._tag === 'hit' || report._tag === 'recheck') { emitOperationalEvent(createWideEvent({ 'operation': 'artifact-build-reuse', - 'outcome': 'hit', + 'outcome': report._tag, 'artifact.resolutionId': report.resolutionId, 'artifact.reuseLookup': report.lookup, 'artifact.reusedFrom': report.reusedFrom, diff --git a/layers/artifact-delivery/server/utils/github-source.ts b/layers/artifact-delivery/server/utils/github-source.ts index 861f1f85..e7f50759 100644 --- a/layers/artifact-delivery/server/utils/github-source.ts +++ b/layers/artifact-delivery/server/utils/github-source.ts @@ -184,6 +184,20 @@ interface GithubClientOptions { random?: () => number /** Unix seconds. A Retry-After delay becomes an absolute retry time against it. */ now?: () => number + /** + * Holds GitHub answers that never change: a commit and a tree, each read by + * its SHA. Pass it to a public client only, so private trees stay out. + */ + cache?: GithubObjectCache +} + +/** + * A store for immutable GitHub answers. `get` answers null for a miss. The + * caller parses every value again, so a stale or foreign value only misses. + */ +export interface GithubObjectCache { + get: (key: string) => Promise + put: (key: string, value: unknown) => Promise } export function createPublicGithubSourceClient(options: GithubClientOptions): PublicGithubSourceClient { @@ -283,9 +297,29 @@ export function createGithubSourceClient(options: GithubClientOptions): PublicGi throw new Error(`GitHub read failed at ${step} ${endpoint}: ${result.reason}`) } + /** + * A read whose answer the SHA in its key fixes forever. Only an answer that + * parsed is stored; a miss, a limit or a refusal always asks GitHub again. + */ + const requestImmutableJson = async (key: string, step: GithubReadStep, path: string, schema: z.ZodType): Promise> => { + const cache = options.cache + if (cache) { + const cached = schema.safeParse(await cache.get(key)) + if (cached.success) + return { _tag: 'ok', value: cached.data } + } + const response = await requestJson(step, path, schema) + if (cache && response._tag === 'ok') + await cache.put(key, response.value) + return response + } + + // A tree SHA names its listing in any Repository that holds it, so the key + // needs no Repository. The commit that names the tree is read per Repository. const getTree = async (owner: string, repository: string, sha: string, recursive: boolean) => { const suffix = recursive ? '?recursive=1' : '' - const response = await requestJson( + const response = await requestImmutableJson( + `tree:${sha}${recursive ? ':recursive' : ''}`, 'tree', `/repos/${encodeURIComponent(owner)}/${encodeURIComponent(repository)}/git/trees/${sha}${suffix}`, treeResponseSchema, @@ -555,7 +589,10 @@ export function createGithubSourceClient(options: GithubClientOptions): PublicGi : request.ref.value if (typeof requestedCommit !== 'string') return requestedCommit - const commit = await requestJson( + // A commit is keyed by Repository ID, so it proves the commit exists in + // this Repository, whatever its name is now. + const commit = await requestImmutableJson( + `${repository.value.id}:commit:${requestedCommit}`, 'commit', `/repos/${encodeURIComponent(repository.value.owner.login)}/${encodeURIComponent(repository.value.name)}/commits/${requestedCommit}`, commitResponseSchema, @@ -1038,6 +1075,23 @@ function githubBlobUrl(source: ResolvedSource, path: string): string { return `https://github.com/${encodeURIComponent(source.owner)}/${encodeURIComponent(source.repository)}/blob/${source.commitSha}/${repositoryPath.split('/').map(encodeURIComponent).join('/')}` } +/** + * Whether files read back from a stored Artifact pass every rule a GitHub load + * applies today, and the load would leave none of them out. A Skill checked + * again from its stored bytes then packs exactly what a load would pack. + */ +export function storedFilesPassLoadRules(files: ArtifactSourceFile[], skillPath: string): boolean { + const entries = files.map(file => ({ + path: file.path, + mode: file.mode === 493 ? '100755' : '100644', + type: 'blob' as const, + sha: file.gitBlobSha, + size: file.bytes.byteLength, + })) + const selected = selectArtifactEntries(entries, skillPath) + return selected._tag === 'selected' && selected.omitted.length === 0 +} + function skillFolderLabel(skillPath: string): string { return skillPath === '.' ? 'The Skill folder' : `The Skill folder \`${skillPath}\`` } diff --git a/layers/artifact-delivery/server/utils/queue.ts b/layers/artifact-delivery/server/utils/queue.ts index cfb0badd..aad0a5a4 100644 --- a/layers/artifact-delivery/server/utils/queue.ts +++ b/layers/artifact-delivery/server/utils/queue.ts @@ -5,7 +5,7 @@ import type { ArtifactGithubCredentialReport, ArtifactGithubCredentialRuntime, } from './github-read-credential' -import type { GithubReadFailure, PublicGithubSourceClient } from './github-source' +import type { GithubObjectCache, GithubReadFailure, PublicGithubSourceClient } from './github-source' import { createWideEvent } from '@harlan-zw/nuxt-wide-events/standalone' import { z } from 'zod' import { emitOperationalEvent } from '#server/utils/operational-event' @@ -47,8 +47,12 @@ const artifactBuildMessageSchema = z.object({ */ export const ARTIFACT_BUILD_RETRY_DELAYS_SECONDS: readonly number[] = [5, 30, 120, 480] -export async function enqueueArtifactBuild(env: Cloudflare.Env, resolutionId: string): Promise { - await env.ARTIFACT_BUILD_QUEUE.send({ version: 1, resolutionId }) +export async function enqueueArtifactBuild(env: Cloudflare.Env, resolutionId: string, delaySeconds?: number): Promise { + const message = { version: 1, resolutionId } + if (delaySeconds === undefined) + await env.ARTIFACT_BUILD_QUEUE.send(message) + else + await env.ARTIFACT_BUILD_QUEUE.send(message, { delaySeconds }) } export async function consumeArtifactBuildBatch( @@ -69,6 +73,25 @@ export async function consumeArtifactBuildBatch( const outcome = await processArtifactBuild(dependencies, parsed.data.resolutionId) .then(value => ({ _tag: 'ok' as const, value })) .catch(error => ({ _tag: 'error' as const, error })) + if (outcome._tag === 'ok' && outcome.value._tag === 'deferred') { + // A wait is no failure, so it goes back as a new message rather than a + // retry. A retry would spend the delivery budget that real failures need. + const requeued = await enqueueArtifactBuild(env, parsed.data.resolutionId, outcome.value.delaySeconds) + .then(() => true, (error: unknown) => { + console.error(JSON.stringify({ + operation: 'artifact-build', + outcome: 'requeue-failed', + resolutionId: parsed.data.resolutionId, + error: error instanceof Error ? error.message : String(error), + })) + return false + }) + if (requeued) + message.ack() + else + message.retry({ delaySeconds: outcome.value.delaySeconds }) + continue + } if (outcome._tag === 'ok') { message.ack() continue @@ -109,6 +132,33 @@ export function reportGithubReadFailure(failure: GithubReadFailure): void { })) } +const GITHUB_OBJECT_CACHE_PREFIX = 'artifact-github:v1:' +/** Commits and trees never change. The TTL only bounds what idle Repositories keep. */ +const GITHUB_OBJECT_CACHE_TTL_SECONDS = 30 * 24 * 60 * 60 + +/** + * Immutable GitHub answers in KV, shared by every build in every location. + * A KV failure is reported and reads GitHub instead, so it never fails a build. + */ +export function createKvGithubObjectCache(kv: KVNamespace): GithubObjectCache { + const report = (outcome: 'read-failed' | 'write-failed', error: unknown) => emitOperationalEvent(createWideEvent({ + operation: 'artifact-github-cache', + outcome, + reason: error instanceof Error ? error.message : String(error), + })) + return { + get: async key => await kv.get(`${GITHUB_OBJECT_CACHE_PREFIX}${key}`, 'json').catch((error: unknown) => { + report('read-failed', error) + return null + }), + put: async (key, value) => { + await kv.put(`${GITHUB_OBJECT_CACHE_PREFIX}${key}`, JSON.stringify(value), { + expirationTtl: GITHUB_OBJECT_CACHE_TTL_SECONDS, + }).catch((error: unknown) => report('write-failed', error)) + }, + } +} + /** * One installation token cache per isolate. A token lives an hour, so it * outlives the queue batch that minted it. @@ -127,11 +177,12 @@ const installationTokens = createInstallationTokenCache() export function createArtifactGithubSource( env: ArtifactGithubCredentialEnv, runtime: ArtifactGithubCredentialRuntime = defaultGithubSourceRuntime(), + cache?: GithubObjectCache, ): PublicGithubSourceClient { const credential = createArtifactGithubCredential(parseArtifactGithubCredentialConfig(env), runtime) return withGithubCredential( credential, - token => createPublicGithubSourceClient({ fetch: runtime.fetch, token, onReadFailure: reportGithubReadFailure }), + token => createPublicGithubSourceClient({ fetch: runtime.fetch, token, onReadFailure: reportGithubReadFailure, cache }), ) } @@ -160,7 +211,7 @@ export function createArtifactBuildDependencies(env: Cloudflare.Env): ArtifactBu : {} return { db: env.DB, - github: createArtifactGithubSource(env), + github: createArtifactGithubSource(env, defaultGithubSourceRuntime(), createKvGithubObjectCache(env.KV_CACHE)), bucket: env.PUBLIC_ARTIFACTS, signer: createArtifactSigner(env.ARTIFACT_SIGNER), trustedRoot: parseTrustedRoot(env.ARTIFACT_TRUSTED_ROOT_JSON, Math.floor(Date.now() / 1000)), diff --git a/layers/artifact-delivery/server/utils/state.ts b/layers/artifact-delivery/server/utils/state.ts index 3602e2be..bea13ef6 100644 --- a/layers/artifact-delivery/server/utils/state.ts +++ b/layers/artifact-delivery/server/utils/state.ts @@ -28,6 +28,20 @@ import { canonicalJson, digestHex } from './encoding' */ export const ARTIFACT_POLICY_VERSION = '2026-10-07.2' +/** + * Earlier policies under which every ready build packed the bytes this policy + * packs for the same commit and folder. A ready build under one of them is + * checked again from its stored bytes, with no GitHub read. + * + * A bump that changes only checks adds the version it replaces. A bump that + * changes the bytes of a folder an earlier policy accepted empties this set. + * + * - `2026-10-07.1`: #487 leaves large media out of a folder that policy + * refused. Every folder it accepted packs the same files. + * - `2026-08-20.1`: #481 changed checks and Skill name resolution only. + */ +export const BYTE_COMPATIBLE_POLICY_VERSIONS: ReadonlySet = new Set(['2026-10-07.1', '2026-08-20.1']) + export const ACTIVE_BUILD_STATES = [ 'requested', 'resolving', @@ -386,6 +400,56 @@ function readyPublicBuildStatement(db: D1Database, lookup: ReadyBuildLookup): D1 ).bind(lookup.commitSha, lookup.owner, lookup.repository, selector.value) } +/** + * How long a build that is ahead may go without a state change and still + * count as running. A queue consumer that died leaves a row behind, and its + * followers must not wait for it. + */ +export const LEADING_BUILD_FRESH_SECONDS = 90 + +/** + * Whether an earlier public build of the same Skill at the same commit is + * running now, so this build can wait and then reuse it. + * + * Builds order by creation time, then ID, so two builds never wait for each + * other. A `pinned` lookup matches the commit and Skill a request names, before + * any GitHub read. A `resolved` lookup matches the source GitHub resolved. + */ +export async function hasLeadingBuild( + db: D1Database, + row: ResolutionRow, + lookup: ReadyBuildLookup, + now: number, +): Promise { + const match = lookup._tag === 'resolved' + ? { + filter: 'commit_sha = ?4 AND repository_id = ?5 AND skill_path = ?6', + values: [lookup.source.commitSha, lookup.source.repositoryId, lookup.source.skillPath], + } + : { + filter: `ref_type = 'commit' AND ref_value = ?4 + AND requested_owner = ?5 COLLATE NOCASE AND requested_repository = ?6 COLLATE NOCASE + AND selector_type = ?7 AND selector_value = ?8`, + values: [ + lookup.commitSha, + lookup.owner, + lookup.repository, + lookup.selector.type, + lookup.selector.type === 'path' ? lookup.selector.path : lookup.selector.name, + ], + } + const leader = await db.prepare( + `SELECT id FROM artifact_resolutions + WHERE state IN (${ACTIVE_BUILD_STATES.map(state => `'${state}'`).join(', ')}) + AND visibility = 'public' + AND (created_at < ?1 OR (created_at = ?1 AND id < ?2)) + AND updated_at >= ?3 + AND ${match.filter} + LIMIT 1`, + ).bind(row.created_at, row.id, now - LEADING_BUILD_FRESH_SECONDS, ...match.values).first<{ id: string }>() + return leader !== null +} + export function parseCheckResults(value: string | null): CheckResult[] { if (!value) return [] diff --git a/layers/artifact-delivery/server/utils/ustar.ts b/layers/artifact-delivery/server/utils/ustar.ts index 87d4b2e9..e48d264c 100644 --- a/layers/artifact-delivery/server/utils/ustar.ts +++ b/layers/artifact-delivery/server/utils/ustar.ts @@ -1,5 +1,6 @@ import type { ArtifactSourceFile } from './github-source' import { splitUstarPath } from './checks' +import { gitBlobShaHex } from './encoding' const BLOCK_SIZE = 512 @@ -78,3 +79,51 @@ export function projectedUstarBytes(sizes: readonly number[]): number { BLOCK_SIZE * 2, ) } + +/** + * The files of an archive {@link createDeterministicUstar} wrote, or null for + * any other archive. + * + * The files must pack back to the exact input bytes. So a stored Artifact + * yields exactly the files it was packed from, and nothing a lenient parser + * could read in two ways. + */ +export async function readDeterministicUstar(archive: Uint8Array): Promise { + const entries: Array<{ path: string, mode: 420 | 493, bytes: Uint8Array }> = [] + let offset = 0 + while (offset + BLOCK_SIZE <= archive.byteLength) { + const header = archive.subarray(offset, offset + BLOCK_SIZE) + if (header.every(byte => byte === 0)) + break + const name = readText(header, 0, 100) + const prefix = readText(header, 345, 155) + const mode = readOctal(header, 100, 8) + const size = readOctal(header, 124, 12) + if (size === null || header[156] !== 0x30 || (mode !== 420 && mode !== 493)) + return null + const start = offset + BLOCK_SIZE + if (start + size > archive.byteLength) + return null + entries.push({ path: prefix ? `${prefix}/${name}` : name, mode, bytes: archive.slice(start, start + size) }) + offset = start + roundToBlock(size) + } + const files = await Promise.all(entries.map(async entry => ({ ...entry, gitBlobSha: await gitBlobShaHex(entry.bytes) }))) + const repacked = files.length > 0 ? createDeterministicUstar(files) : null + return repacked && sameBytes(repacked, archive) ? files : null +} + +/** A NUL-terminated field. Invalid UTF-8 decodes lossily and then fails the repack comparison. */ +function readText(source: Uint8Array, offset: number, size: number): string { + const field = source.subarray(offset, offset + size) + const end = field.indexOf(0) + return new TextDecoder().decode(field.subarray(0, end === -1 ? size : end)) +} + +function readOctal(source: Uint8Array, offset: number, size: number): number | null { + const text = readText(source, offset, size) + return /^[0-7]+$/.test(text) ? Number.parseInt(text, 8) : null +} + +function sameBytes(left: Uint8Array, right: Uint8Array): boolean { + return left.byteLength === right.byteLength && left.every((byte, index) => byte === right[index]) +} diff --git a/test/unit/artifact-build-reuse.test.ts b/test/unit/artifact-build-reuse.test.ts index 5dc57596..6725b405 100644 --- a/test/unit/artifact-build-reuse.test.ts +++ b/test/unit/artifact-build-reuse.test.ts @@ -181,33 +181,37 @@ describe('reusing a ready public build', () => { harness.close() }) - it.each([ - { - change: 'policy version', - reason: 'policy-changed', - edit: (statement: ArtifactAttestationStatement) => ({ ...statement, policyVersion: '2026-01-01.0' }), - }, - { - change: 'check version', - reason: 'checks-changed', - edit: (statement: ArtifactAttestationStatement) => ({ - ...statement, - checkResults: statement.checkResults.map(check => check.name === 'agent-skills-spec' - ? { ...check, version: '2026-01-01' } - : check), - }), - }, - ])('loads again when the ready build was signed under another $change', async ({ reason, edit }) => { + it('loads again when the ready build was signed under a policy that packed other bytes', async () => { const harness = await createReuseHarness() const first = await harness.build(pinned, githubServing(source)) - await harness.resignReadyBuild(first.row.id, edit) + await harness.resignReadyBuild(first.row.id, statement => ({ ...statement, policyVersion: '2026-01-01.0' })) const github = githubServing(source) const repeat = await harness.build(pinned, github) expect(repeat.row.state).toBe('ready') expect(github.load).toHaveBeenCalledOnce() - expect(harness.reports).toContainEqual({ _tag: 'miss', lookup: 'pinned', resolutionId: repeat.row.id, reason }) + expect(harness.reports).toContainEqual({ _tag: 'miss', lookup: 'pinned', resolutionId: repeat.row.id, reason: 'policy-changed' }) + harness.close() + }) + + it('checks the stored bytes again when the ready build was signed under another check version', async () => { + const harness = await createReuseHarness() + const first = await harness.build(pinned, githubServing(source)) + await harness.resignReadyBuild(first.row.id, statement => ({ + ...statement, + checkResults: statement.checkResults.map(check => check.name === 'agent-skills-spec' + ? { ...check, version: '2026-01-01' } + : check), + })) + const github = githubServing(source) + + const repeat = await harness.build(pinned, github) + + expect(repeat.row.state).toBe('ready') + expect(github.resolve).not.toHaveBeenCalled() + expect(github.load).not.toHaveBeenCalled() + expect(harness.reports).toContainEqual({ _tag: 'recheck', lookup: 'pinned', resolutionId: repeat.row.id, reusedFrom: first.row.id }) harness.close() }) diff --git a/test/unit/artifact-github-reads.test.ts b/test/unit/artifact-github-reads.test.ts new file mode 100644 index 00000000..747b2b7e --- /dev/null +++ b/test/unit/artifact-github-reads.test.ts @@ -0,0 +1,461 @@ +import type { ArtifactAttestationStatement, SourceRequest } from '../../layers/artifact-delivery/server/schemas/contracts' +import type { ArtifactBuildDependencies } from '../../layers/artifact-delivery/server/utils/build' +import type { GithubObjectCache } from '../../layers/artifact-delivery/server/utils/github-source' +import type { TrustedRoot } from '../../layers/artifact-delivery/server/utils/trusted-root' +import type { ArtifactSignerBindings } from '../../workers/artifact-signer/src/handler' +import { createHash } from 'node:crypto' +import { gzipSync } from 'node:zlib' +import { describe, expect, it, vi } from 'vitest' +import { artifactAttestationSchema } from '../../layers/artifact-delivery/server/schemas/contracts' +import { + completeAttestation, + createArtifactSigner, + createAttestationSignaturePayload, + encodeAttestationStatement, +} from '../../layers/artifact-delivery/server/utils/attestation' +import { processArtifactBuild } from '../../layers/artifact-delivery/server/utils/build' +import { bytesToBase64Url, gitBlobShaHex } from '../../layers/artifact-delivery/server/utils/encoding' +import { createPublicGithubSourceClient } from '../../layers/artifact-delivery/server/utils/github-source' +import { ARTIFACT_BUILD_QUEUE_NAME, consumeArtifactBuildBatch, createKvGithubObjectCache } from '../../layers/artifact-delivery/server/utils/queue' +import { + ARTIFACT_POLICY_VERSION, + createResolution, + getResolution, + resolutionRequestIdentity, + transitionResolution, +} from '../../layers/artifact-delivery/server/utils/state' +import { createDeterministicUstar } from '../../layers/artifact-delivery/server/utils/ustar' +import { handleArtifactSignerRequest } from '../../workers/artifact-signer/src/handler' +import { createSqliteD1 } from './helpers/d1-sqlite' + +/** + * GitHub reads per run, counted at the fetch boundary of the real source + * client. Launch traffic repeats the same Skills, and a spent GitHub quota is + * the only failure class production has shown. + */ + +const NOW = 1_787_227_200 +const MIGRATIONS = [ + 'migrations/0017_users.sql', + 'migrations/0110_artifact_delivery.sql', + 'migrations/0111_github_app_delivery.sql', + 'migrations/0112_private_artifact_keys.sql', + 'migrations/0122_artifact_resolution_retry_after.sql', + 'migrations/0131_artifact_build_reuse.sql', +] +const REPOSITORY_ID = 123456789 +const COMMIT = '0123456789abcdef0123456789abcdef01234567' +const ROOT_TREE = '1'.repeat(40) +const SKILLS_TREE = '2'.repeat(40) +const DEMO_TREE = '3'.repeat(40) +const OTHER_TREE = '4'.repeat(40) +const SKILL_MD = new TextEncoder().encode('---\nname: demo\ndescription: Use this Skill for demo work.\n---\n\n# Demo\n') +const OTHER_MD = new TextEncoder().encode('---\nname: other\ndescription: Use this Skill for other work.\n---\n\n# Other\n') + +const pinned: SourceRequest = { + provider: 'github', + owner: 'skilld-dev', + repository: 'skills', + selector: { type: 'path', path: 'skills/demo' }, + ref: { type: 'commit', value: COMMIT }, +} +const onBranch: SourceRequest = { ...pinned, ref: { type: 'branch', value: 'main' } } + +describe('gitHub reads per run', () => { + it('reads a commit and its trees from GitHub once', async () => { + const harness = await createHarness() + await harness.run(onBranch) + harness.github.reset() + + const repeat = await harness.run(onBranch) + + expect(repeat.row.state).toBe('ready') + // Before: repository, ref and commit. The commit never changes. + expect(harness.github.paths()).toEqual([ + '/repos/skilld-dev/skills', + '/repos/skilld-dev/skills/git/ref/heads/main', + ]) + harness.close() + }) + + it('reads only the new folder of another Skill at a built commit', async () => { + const harness = await createHarness() + await harness.run(pinned) + harness.github.reset() + + const other = await harness.run({ ...pinned, selector: { type: 'path', path: 'skills/other' } }) + + expect(other.row.state).toBe('ready') + expect(harness.github.paths()).toEqual([ + '/repos/skilld-dev/skills', + '/repos/skilld-dev/skills', + `/repos/skilld-dev/skills/git/trees/${OTHER_TREE}?recursive=1`, + `/repos/skilld-dev/skills/tarball/${COMMIT}`, + ]) + harness.close() + }) +}) + +// #481 changed only checks, yet its policy bump made every ready build load +// from GitHub again. R2 still held the exact bytes of each one. +describe('a policy bump that changes only checks', () => { + it('checks the stored bytes again instead of reading GitHub', async () => { + const harness = await createHarness() + const first = await harness.run(pinned) + await harness.resign(first.row.id, statement => ({ + ...statement, + policyVersion: '2026-08-20.1', + checkResults: statement.checkResults.map(check => check.name === 'agent-skills-spec' + ? { ...check, version: '2026-08-20', required: true } + : check), + })) + harness.github.reset() + + const repeat = await harness.run(pinned) + + expect(repeat.row.state).toBe('ready') + expect(harness.github.paths()).toEqual([]) + expect(repeat.row.artifact_id).toBe(first.row.artifact_id) + const attestation = artifactAttestationSchema.parse(JSON.parse(repeat.row.attestation_json!)) + expect(attestation.policyVersion).toBe(ARTIFACT_POLICY_VERSION) + expect(attestation.checkResults.find(check => check.name === 'agent-skills-spec')) + .toMatchObject({ version: '2026-10-07', required: false }) + harness.close() + }) + + it('loads from GitHub when the stored build left a file out', async () => { + const harness = await createHarness() + const first = await harness.run(pinned) + await harness.resign(first.row.id, statement => ({ + ...statement, + policyVersion: '2026-10-07.1', + checkResults: statement.checkResults.map(check => check.name === 'omitted-files' + ? { ...check, outcome: 'warn' as const, summary: '1 file over the size limits was left out of the Artifact.' } + : check), + })) + harness.github.reset() + + const repeat = await harness.run(pinned) + + expect(repeat.row.state).toBe('ready') + expect(harness.github.paths()).toContain(`/repos/skilld-dev/skills/tarball/${COMMIT}`) + harness.close() + }) + + it('loads from GitHub again after a bump that changed the bytes', async () => { + const harness = await createHarness() + const first = await harness.run(pinned) + await harness.resign(first.row.id, statement => ({ ...statement, policyVersion: '2026-01-01.0' })) + harness.github.reset() + + const repeat = await harness.run(pinned) + + expect(repeat.row.state).toBe('ready') + expect(harness.github.paths()).toContain(`/repos/skilld-dev/skills/tarball/${COMMIT}`) + harness.close() + }) +}) + +describe('two first runs of one Skill', () => { + it('lets the second wait for the first build, then reuse it', async () => { + const harness = await createHarness() + const leader = await harness.request(pinned) + const follower = await harness.request(pinned) + await harness.begin(leader) + + const waited = await harness.process(follower) + + expect(waited).toEqual({ _tag: 'deferred', resolutionId: follower, delaySeconds: 2 }) + expect(harness.github.paths()).toEqual([]) + expect(await harness.process(leader)).toEqual({ _tag: 'ready', resolutionId: leader }) + harness.github.reset() + expect(await harness.process(follower)).toEqual({ _tag: 'ready', resolutionId: follower }) + expect(harness.github.paths()).toEqual([]) + harness.close() + }) + + it('waits for a first run on a branch that resolved to the same commit', async () => { + const harness = await createHarness() + const leader = await harness.request(onBranch) + const follower = await harness.request(onBranch) + await harness.begin(leader) + await harness.beginFetching(leader) + + const waited = await harness.process(follower) + + expect(waited).toMatchObject({ _tag: 'deferred' }) + // It resolved its branch, then waited instead of loading the same commit. + expect(harness.github.paths()).toEqual([ + '/repos/skilld-dev/skills', + '/repos/skilld-dev/skills/git/ref/heads/main', + `/repos/skilld-dev/skills/commits/${COMMIT}`, + ]) + harness.close() + }) + + it('builds alone when the first build stalls', async () => { + const harness = await createHarness() + const leader = await harness.request(pinned) + const follower = await harness.request(pinned) + await harness.begin(leader) + harness.advanceClock(91) + + expect(await harness.process(follower)).toEqual({ _tag: 'ready', resolutionId: follower }) + harness.close() + }) + + it('requeues a waiting build as a new message, so the wait spends no retry', async () => { + const harness = await createHarness() + const leader = await harness.request(pinned) + const follower = await harness.request(pinned) + await harness.begin(leader) + const send = vi.fn(async () => undefined) + const message = { body: { version: 1, resolutionId: follower }, attempts: 1, ack: vi.fn(), retry: vi.fn() } + + await consumeArtifactBuildBatch( + { ARTIFACT_BUILD_QUEUE: { send } } as unknown as Cloudflare.Env, + { queue: ARTIFACT_BUILD_QUEUE_NAME, messages: [message] } as unknown as Parameters[1], + () => harness.dependencies(), + ) + + expect(send).toHaveBeenCalledWith({ version: 1, resolutionId: follower }, { delaySeconds: 2 }) + expect(message.ack).toHaveBeenCalledOnce() + expect(message.retry).not.toHaveBeenCalled() + harness.close() + }) +}) + +describe('the KV store for immutable GitHub answers', () => { + it('reads GitHub when KV fails, so a KV outage never fails a build', async () => { + const cache = createKvGithubObjectCache({ + get: vi.fn(async () => Promise.reject(new Error('KV GET failed: 503'))), + put: vi.fn(async () => Promise.reject(new Error('KV PUT failed: 503'))), + } as unknown as KVNamespace) + const github = await fakeGithub() + const client = createPublicGithubSourceClient({ fetch: github.fetch, cache }) + + const result = await client.resolve(pinned) + + expect(result).toMatchObject({ _tag: 'resolved', source: { commitSha: COMMIT, treeSha: ROOT_TREE } }) + expect(github.paths()).toContain(`/repos/skilld-dev/skills/commits/${COMMIT}`) + }) +}) + +async function createHarness() { + const sqlite = createSqliteD1(MIGRATIONS) + const storage = memoryBucket() + const github = await fakeGithub() + const cache = memoryCache() + let clock = NOW + let requests = 0 + const signing = await signingKey('skilld-production-2026-08') + const signerBindings: ArtifactSignerBindings = { + DB: sqlite.db, + PUBLIC_ARTIFACTS: storage, + PRIVATE_ARTIFACTS: storage, + ARTIFACT_SIGNING_MAX_AGE_SECONDS: '300', + ...signing.bindings, + } + const signer = createArtifactSigner({ + fetch: async (request: Request) => await handleArtifactSignerRequest(request, signerBindings, () => clock), + } as unknown as Fetcher) + + function dependencies(): ArtifactBuildDependencies { + return { + db: sqlite.db, + github: createPublicGithubSourceClient({ fetch: github.fetch, cache, now: () => clock }), + bucket: storage, + signer, + trustedRoot: signing.trustedRoot, + now: () => clock, + reportReuse: () => {}, + } + } + + async function request(sourceRequest: SourceRequest): Promise { + requests += 1 + const identity = await resolutionRequestIdentity(sourceRequest, `github-reads-request-${requests}`) + const created = await createResolution(sqlite.db, sourceRequest, identity, clock) + if (created._tag === 'idempotency-conflict') + throw new Error('Test Resolution conflicted') + clock += 1 + return created.row.id + } + + async function load(resolutionId: string) { + const row = await getResolution(sqlite.db, resolutionId) + if (!row) + throw new Error('Test Resolution was not stored') + return row + } + + return { + github, + dependencies, + request, + process: async (resolutionId: string) => await processArtifactBuild(dependencies(), resolutionId), + async run(sourceRequest: SourceRequest) { + const resolutionId = await request(sourceRequest) + const outcome = await processArtifactBuild(dependencies(), resolutionId) + return { outcome, row: await load(resolutionId) } + }, + /** A build that a queue consumer picked up and has not finished. */ + async begin(resolutionId: string) { + await transitionResolution(sqlite.db, await load(resolutionId), 'resolving', {}, clock) + }, + /** That build resolved its source on GitHub and is loading it. */ + async beginFetching(resolutionId: string) { + await transitionResolution(sqlite.db, await load(resolutionId), 'fetching', { + repositoryId: REPOSITORY_ID, + resolvedOwner: 'skilld-dev', + resolvedRepository: 'skills', + commitSha: COMMIT, + treeSha: ROOT_TREE, + skillPath: 'skills/demo', + }, clock) + }, + /** Replace a ready build's attestation with an edited statement the key signs. */ + async resign(resolutionId: string, edit: (statement: ArtifactAttestationStatement) => ArtifactAttestationStatement) { + const row = await load(resolutionId) + const { statement: _statement, signature: _signature, ...statement } = artifactAttestationSchema.parse(JSON.parse(row.attestation_json!)) + const raw = encodeAttestationStatement(edit(statement)) + const signature = await crypto.subtle.sign( + 'Ed25519', + signing.privateKey, + await createAttestationSignaturePayload(new TextEncoder().encode(raw)), + ) + const attestation = completeAttestation(raw, { + algorithm: 'Ed25519', + keyId: signing.keyId, + value: bytesToBase64Url(new Uint8Array(signature)), + }) + sqlite.raw.prepare('UPDATE artifact_resolutions SET attestation_json = ? WHERE id = ?') + .run(JSON.stringify(attestation), resolutionId) + }, + advanceClock: (seconds: number) => { + clock += seconds + }, + close: sqlite.close, + } +} + +/** One public Repository with two Skills at one commit, served the way api.github.com answers. */ +async function fakeGithub() { + const demoBlob = await gitBlobShaHex(SKILL_MD) + const otherBlob = await gitBlobShaHex(OTHER_MD) + const tarball = gzipSync(createDeterministicUstar([ + { path: `skilld-dev-skills-0123456/skills/demo/SKILL.md`, mode: 420, bytes: SKILL_MD, gitBlobSha: demoBlob }, + { path: `skilld-dev-skills-0123456/skills/other/SKILL.md`, mode: 420, bytes: OTHER_MD, gitBlobSha: otherBlob }, + ])) + const base = '/repos/skilld-dev/skills' + const routes = new Map Response>([ + [base, () => json({ id: REPOSITORY_ID, name: 'skills', owner: { login: 'skilld-dev' }, private: false, default_branch: 'main' })], + [`${base}/git/ref/heads/main`, () => json({ ref: 'refs/heads/main', object: { type: 'commit', sha: COMMIT } })], + [`${base}/commits/${COMMIT}`, () => json({ sha: COMMIT, commit: { tree: { sha: ROOT_TREE } } })], + [`${base}/git/trees/${ROOT_TREE}`, () => json({ sha: ROOT_TREE, tree: [{ path: 'skills', mode: '040000', type: 'tree', sha: SKILLS_TREE }] })], + [`${base}/git/trees/${SKILLS_TREE}`, () => json({ sha: SKILLS_TREE, tree: [ + { path: 'demo', mode: '040000', type: 'tree', sha: DEMO_TREE }, + { path: 'other', mode: '040000', type: 'tree', sha: OTHER_TREE }, + ] })], + [`${base}/git/trees/${DEMO_TREE}?recursive=1`, () => json({ sha: DEMO_TREE, truncated: false, tree: [ + { path: 'SKILL.md', mode: '100644', type: 'blob', sha: demoBlob, size: SKILL_MD.byteLength }, + ] })], + [`${base}/git/trees/${OTHER_TREE}?recursive=1`, () => json({ sha: OTHER_TREE, truncated: false, tree: [ + { path: 'SKILL.md', mode: '100644', type: 'blob', sha: otherBlob, size: OTHER_MD.byteLength }, + ] })], + [`${base}/tarball/${COMMIT}`, () => new Response(Uint8Array.from(tarball), { status: 200 })], + ]) + let requested: string[] = [] + const fetch = vi.fn(async (input: RequestInfo | URL) => { + const url = new URL(String(input)) + const path = `${url.pathname}${url.search}` + requested.push(path) + return routes.get(path)?.() ?? json({ message: 'Not Found' }, 404) + }) as unknown as typeof globalThis.fetch + return { + fetch, + paths: () => [...requested], + reset: () => { + requested = [] + }, + } +} + +function memoryCache(): GithubObjectCache { + const entries = new Map() + return { + get: async key => entries.has(key) ? JSON.parse(entries.get(key)!) as unknown : null, + put: async (key, value) => { + entries.set(key, JSON.stringify(value)) + }, + } +} + +function json(value: unknown, status = 200): Response { + return new Response(JSON.stringify(value), { status, headers: { 'content-type': 'application/json' } }) +} + +/** An R2 bucket in memory that computes checksums from the bytes, as R2 does. */ +function memoryBucket(): R2Bucket { + const objects = new Map }>() + const describe = (key: string) => { + const object = objects.get(key) + if (!object) + return null + return { + key, + size: object.bytes.byteLength, + checksums: { sha256: Uint8Array.from(createHash('sha256').update(object.bytes).digest()).buffer }, + customMetadata: object.customMetadata, + } + } + return { + put: async (key: string, value: Uint8Array, options?: R2PutOptions) => { + if (objects.has(key)) + return null + objects.set(key, { bytes: Uint8Array.from(value), customMetadata: options?.customMetadata ?? {} }) + return describe(key) + }, + head: async (key: string) => describe(key), + get: async (key: string) => { + const object = describe(key) + const stored = objects.get(key) + return object && stored ? { ...object, arrayBuffer: async () => Uint8Array.from(stored.bytes).buffer } : null + }, + } as unknown as R2Bucket +} + +async function signingKey(keyId: string) { + const keyPair = await crypto.subtle.generateKey('Ed25519', true, ['sign', 'verify']) + const publicKey = bytesToBase64Url(new Uint8Array(await crypto.subtle.exportKey('raw', keyPair.publicKey))) + const privateKey = bytesToBase64Url(new Uint8Array(await crypto.subtle.exportKey('pkcs8', keyPair.privateKey))) + const notBefore = new Date((NOW - 60) * 1000).toISOString() + const notAfter = new Date((NOW + 3600) * 1000).toISOString() + const trustedRoot: TrustedRoot = { + version: 1, + rootKeyId: 'skilld-root-2026', + rootPublicKey: publicKey, + keys: [{ + keyId, + algorithm: 'Ed25519', + publicKey, + notBefore, + notAfter, + status: 'active', + statement: bytesToBase64Url(new TextEncoder().encode(JSON.stringify({ keyId }))), + rootSignature: 'C'.repeat(86), + }], + fetchedAt: new Date(NOW * 1000).toISOString(), + } + return { + keyId, + privateKey: keyPair.privateKey, + trustedRoot, + bindings: { + ARTIFACT_SIGNING_KEY_ID: keyId, + ARTIFACT_SIGNING_KEY_NOT_BEFORE: notBefore, + ARTIFACT_SIGNING_KEY_NOT_AFTER: notAfter, + ARTIFACT_SIGNING_PRIVATE_KEY_PKCS8: privateKey, + }, + } +}